chainqueue

commit d19fbf005e74d2391e5b516be44905de7a37b7e4
parent 485b33866b11d5d93e39eed1d91249a5c5b3d088
Author: lash <dev@holbrook.no>
Date:   Sun, 13 Mar 2022 17:22:39 +0000

Add date modified to state dirs

Diffstat:
M
chainqueue/store/base.py
 | 
25
+++++++++++++++++++++----
M
tests/test_integrate.py
 | 
16
++++++++++++++++

2 files changed, 37 insertions(+), 4 deletions(-)
diff --git a/chainqueue/store/base.py b/chainqueue/store/base.py
@@ -34,7 +34,14 @@ class Store:
                 continue
             v = self.state_store.from_name(s)
             setattr(self, s, v)
-        for v in ['state', 'change', 'set', 'unset', 'name']:
+        for v in [
+                'state',
+                'change',
+                'set',
+                'unset',
+                'name',
+                'modified',
+                ]:
             setattr(self, v, getattr(self.state_store, v))
 
 
@@ -48,6 +55,7 @@ class Store:
             tx = cache_adapter()
             tx.deserialize(v)
             self.cache.put(self.chain_spec, tx) 
+        return s
 
 
     def get(self, k):
@@ -56,7 +64,7 @@ class Store:
         return (s, v,)
 
 
-    def by_state(self, state=0, limit=4096, strict=False):
+    def by_state(self, state=0, limit=4096, strict=False, threshold=None):
         hashes = []
         i = 0
 
@@ -70,8 +78,17 @@ class Store:
                 item_state = self.state_store.state(ref)
                 if item_state & state != item_state:
                     continue
+
+            if threshold != None:
+                v = self.state_store.modified(ref)
+                logg.debug('compare {} {}'.format(v, threshold))
+                if v > threshold:
+                    continue
+
             hashes.append(hsh)
 
+
+
         hashes.sort()
         return hashes
 
@@ -80,8 +97,8 @@ class Store:
         return self.by_state(state=self.QUEUED, limit=limit)
 
 
-    def deferred(self, limit=4096):
-        return self.by_state(state=self.DEFERRED, limit=limit)
+    def deferred(self, limit=4096, threshold=None):
+        return self.by_state(state=self.DEFERRED, limit=limit, threshold=threshold)
 
 
     def pending(self, limit=4096):
diff --git a/tests/test_integrate.py b/tests/test_integrate.py
@@ -3,6 +3,7 @@ import os
 import tempfile
 import unittest
 import logging
+import time
 
 # external imports
 from shep.store.file import SimpleFileStoreFactory
@@ -99,5 +100,20 @@ class TestIntegrateBase(TestShepBase):
         self.assertEqual(len(v), 2)
 
 
+    def test_state_date_threshold(self):
+        hx = os.urandom(4).hex()
+        s = self.store.put(hx, os.urandom(8).hex(), cache_adapter=MockCacheTokenTx)
+        self.store.fail(hx)
+        then = self.store.modified(s)
+        time.sleep(0.1)
+
+        hx = os.urandom(4).hex()
+        s = self.store.put(hx, os.urandom(8).hex(), cache_adapter=MockCacheTokenTx)
+        self.store.fail(hx)
+
+        v = self.store.deferred(threshold=then)
+        self.assertEqual(len(v), 1)
+
+
 if __name__ == '__main__':
     unittest.main()