Add state finalizers

chainqueue/cache/base.py

@@ -29,7 +29,7 @@ class CacheTx:
         self.nonce = None
         self.value = None
 
-        self.tx_hash = None
+        self.hash = None
         self.block_number = None
         self.tx_index = None
         self.timestamp = None
@@ -42,7 +42,7 @@ class CacheTx:
 
 
     def init(self, tx_hash, nonce, sender, recipient, value):
-        self.tx_hash = self.normalizer.hash(tx_hash)
+        self.hash = self.normalizer.hash(tx_hash)
         self.sender = self.normalizer.address(sender)
         self.recipient = self.normalizer.address(recipient)
         self.nonce = nonce
@@ -59,7 +59,7 @@ class CacheTx:
 
 
     def __str__(self):
-        return '{}: {} ({}) -> {} = {}'.format(self.tx_hash, self.sender, self.nonce, self.recipient, self.value)
+        return '{}: {} ({}) -> {} = {}'.format(self.hash, self.sender, self.nonce, self.recipient, self.value)
 
 
 
@@ -133,3 +133,7 @@ class Cache:
 
     def count(self, cache_filter=None):
         raise NotImplementedError()
+
+
+    def set_block(self, block, tx):
+        raise NotImplementedError()

chainqueue/entry.py

@@ -105,12 +105,12 @@ class QueueEntry:
         self.store.change(self.k, self.store.RESERVED, self.store.QUEUED) 
 
 
-    def fail(self, block):
+    def fail(self, block, tx):
         if self.__match_state(self.store.NETWORK_ERROR):
             return
         self.store.set(self.k, self.store.NETWORK_ERROR)
-        if self.cache:
-            self.cache.set_block(self.tx_hash, block)
+        if self.store.cache:
+            self.store.cache.set_block(self.tx_hash, block, tx)
 
 
     def cancel(self, confirmed=False):
@@ -120,8 +120,10 @@ class QueueEntry:
             self.store.change(self.k, self.store.OBSOLETE, self.store.RESERVED | self.store.QUEUED)
 
 
-    def succeed(self, block):
+    def succeed(self, block, tx):
         self.store.set(self.k, self.store.FINAL)
+        if self.store.cache:
+            self.store.cache.set_block(self.tx_hash, block, tx)
 
 
     def __str__(self):

chainqueue/store/base.py

@@ -121,6 +121,15 @@ class Store:
         entry.sendfail()
 
 
+    def final(self, k, block, tx, error=False):
+        entry = QueueEntry(self, k)
+        entry.load()
+        if error:
+            entry.fail(block, tx)
+        else:
+            entry.succeed(block, tx)
+
+
     def send_start(self, k):
         entry = QueueEntry(self, k)
         entry.load()

tests/common.py

@@ -27,7 +27,7 @@ class MockTokenCache(Cache):
         self.last_filter = None
 
     def put(self, chain_spec, cache_tx):
-        self.db[cache_tx.tx_hash] = cache_tx
+        self.db[cache_tx.hash] = cache_tx
 
 
     def get(self, chain_spec, tx_hash):

tests/test_cache.py

@@ -57,7 +57,7 @@ class TestCache(TestShepBase):
         self.assertTrue(isinstance(tx.value, float))
         self.assertEqual(tx.sender[:4], 'addr')
         self.assertEqual(tx.recipient[:4], 'addr')
-        self.assertEqual(tx.tx_hash[:11], 'ashbashhash')
+        self.assertEqual(tx.hash[:11], 'ashbashhash')
 
 
     def test_cache_putget(self):
@@ -65,7 +65,7 @@ class TestCache(TestShepBase):
         tx = MockCacheTokenTx()
         tx.deserialize(a)
         self.cache.put(self.chain_spec, tx)
-        tx_retrieved = self.cache.get(self.chain_spec, tx.tx_hash)
+        tx_retrieved = self.cache.get(self.chain_spec, tx.hash)
         self.assertEqual(tx, tx_retrieved)
 
 

tests/test_entry.py

@@ -44,7 +44,7 @@ class TestEntry(TestShepBase):
         txs = self.store.by_state(state=self.store.IN_NETWORK)
         self.assertEqual(len(txs), 1)
 
-        entry.succeed(0)
+        entry.succeed(None, None)
         txs = self.store.by_state()
         self.assertEqual(len(txs), 1)
       

tests/test_integrate.py

@@ -67,6 +67,7 @@ class TestIntegrateBase(TestShepBase):
     def test_state_defer(self):
         hx = os.urandom(4).hex()
         self.store.put(hx, os.urandom(8).hex(), cache_adapter=MockCacheTokenTx)
+        self.store.reserve(hx)
         self.store.fail(hx)
         v = self.store.deferred()
         self.assertEqual(len(v), 1)
@@ -76,9 +77,11 @@ class TestIntegrateBase(TestShepBase):
     def test_state_multiple(self):
         hx = os.urandom(4).hex()
         self.store.put(hx, os.urandom(8).hex(), cache_adapter=MockCacheTokenTx)
+        self.store.reserve(hx)
         self.store.fail(hx)
         hx = os.urandom(8).hex()
         self.store.put(hx, os.urandom(8).hex(), cache_adapter=MockCacheTokenTx)
+        self.store.reserve(hx)
         self.store.fail(hx)
         v = self.store.deferred()
         self.assertEqual(len(v), 2)
@@ -87,12 +90,14 @@ class TestIntegrateBase(TestShepBase):
     def test_state_multiple_sort(self):
         hx = os.urandom(4).hex()
         self.store.put(hx, os.urandom(8).hex(), cache_adapter=MockCacheTokenTx)
+        self.store.reserve(hx)
         self.store.fail(hx)
         hx = os.urandom(4).hex()
         self.store.put(hx, os.urandom(8).hex(), cache_adapter=MockCacheTokenTx)
         self.store.enqueue(hx)
         hx = os.urandom(4).hex()
         self.store.put(hx, os.urandom(8).hex(), cache_adapter=MockCacheTokenTx)
+        self.store.reserve(hx)
         self.store.fail(hx)
         hx = os.urandom(4).hex()
         self.store.put(hx, os.urandom(8).hex(), cache_adapter=MockCacheTokenTx)
@@ -103,12 +108,14 @@ class TestIntegrateBase(TestShepBase):
     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.reserve(hx)
         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.reserve(hx)
         self.store.fail(hx)
 
         v = self.store.deferred(threshold=then)