in mempool, use input addresses from parser instead of reading them from database
[electrum-server.git] / backends / bitcoind / blockchain_processor.py
index e93439e..d95764e 100644 (file)
@@ -68,6 +68,7 @@ class BlockchainProcessor(Processor):
             print_log('initializing database')
             self.height = 0
             self.last_hash = '000000000019d6689c085ae165831e934ff763ae46a2a6c172b3f1b60a8ce26f'
+            db_version = self.db_version
 
         # check version
         if self.db_version != db_version:
@@ -87,7 +88,9 @@ class BlockchainProcessor(Processor):
                 shared.stop()
                 sys.exit(0)
 
-        print_log("blockchain is up to date.")
+        print_log("Blockchain is up to date.")
+        self.memorypool_update()
+        print_log("Memory pool initialized.")
 
         threading.Timer(10, self.main_iteration).start()
 
@@ -266,14 +269,14 @@ class BlockchainProcessor(Processor):
         # sort history, because redeeming transactions are next to the corresponding txout
         hist.sort(key=lambda tup: tup[2])
 
-        # uniqueness
-        hist = set(map(lambda x: (x[0], x[2]), hist))
-
         # add memory pool
         with self.mempool_lock:
             for txid in self.mempool_hist.get(addr, []):
                 hist.append((txid, 0, 0))
 
+        # uniqueness
+        hist = set(map(lambda x: (x[0], x[2]), hist))
+
         # convert to dict
         hist = map(lambda x: {'tx_hash': x[0], 'height': x[1]}, hist)
 
@@ -339,7 +342,7 @@ class BlockchainProcessor(Processor):
         for i in range(l-1, -1, -1):
             item = serialized_hist[80*i:80*(i+1)]
             item_height = int(rev_hex(item[36:39].encode('hex')), 16)
-            if item_height < tx_height:
+            if item_height <= tx_height:
                 serialized_hist = serialized_hist[0:80*(i+1)] + s + serialized_hist[80*(i+1):]
                 break
         else:
@@ -426,6 +429,7 @@ class BlockchainProcessor(Processor):
                 serialized_hist = serialized_hist[0:80*i] + new_item + serialized_hist[80*(i+1):]
                 break
         else:
+            self.shared.stop()
             hist = self.deserialize(serialized_hist)
             raise BaseException("prevout not found", addr, hist, txi.encode('hex'))
 
@@ -493,10 +497,14 @@ class BlockchainProcessor(Processor):
             for txi in block_inputs:
                 try:
                     addr = self.db.Get(txi)
-                except:
-                    # print "addr not in db", txi.encode('hex')
+                except KeyError:
                     # the input could come from the same block
                     continue
+                except:
+                    traceback.print_exc(file=sys.stdout)
+                    self.shared.stop()
+                    raise
+
                 self.batch_txio[txi] = addr
                 addr_to_read.append(addr)
 
@@ -525,8 +533,12 @@ class BlockchainProcessor(Processor):
         for addr in addr_to_read:
             try:
                 self.batch_list[addr] = self.db.Get(addr)
-            except:
+            except KeyError:
                 self.batch_list[addr] = ''
+            except:
+                traceback.print_exc(file=sys.stdout)
+                self.shared.stop()
+                raise
 
 
         # process
@@ -806,6 +818,7 @@ class BlockchainProcessor(Processor):
     def memorypool_update(self):
         mempool_hashes = self.bitcoind('getrawmempool')
 
+        touched_addresses = []
         for tx_hash in mempool_hashes:
             if tx_hash in self.mempool_hashes:
                 continue
@@ -814,24 +827,21 @@ class BlockchainProcessor(Processor):
             if not tx:
                 continue
 
+            mpa = self.mempool_addresses.get(tx_hash, [])
             for x in tx.get('inputs'):
-                txi = (x.get('prevout_hash') + int_to_hex(x.get('prevout_n'), 4)).decode('hex')
-                try:
-                    addr = self.db.Get(txi)
-                except:
-                    continue
-                l = self.mempool_addresses.get(tx_hash, [])
-                if addr not in l:
-                    l.append(addr)
-                    self.mempool_addresses[tx_hash] = l
+                # we assume that the input address can be parsed by deserialize(); this is true for Electrum transactions
+                addr = x.get('address')
+                if addr and addr not in mpa:
+                    mpa.append(addr)
+                    touched_addresses.append(addr)
 
             for x in tx.get('outputs'):
                 addr = x.get('address')
-                l = self.mempool_addresses.get(tx_hash, [])
-                if addr not in l:
-                    l.append(addr)
-                    self.mempool_addresses[tx_hash] = l
+                if addr and addr not in mpa:
+                    mpa.append(addr)
+                    touched_addresses.append(addr)
 
+            self.mempool_addresses[tx_hash] = mpa
             self.mempool_hashes.append(tx_hash)
 
         # remove older entries from mempool_hashes
@@ -841,6 +851,8 @@ class BlockchainProcessor(Processor):
         for tx_hash, addresses in self.mempool_addresses.items():
             if tx_hash not in self.mempool_hashes:
                 self.mempool_addresses.pop(tx_hash)
+                for addr in addresses:
+                    touched_addresses.append(addr)
 
         # rebuild mempool histories
         new_mempool_hist = {}
@@ -851,27 +863,14 @@ class BlockchainProcessor(Processor):
                     h.append(tx_hash)
                 new_mempool_hist[addr] = h
 
-        # invalidate cache for mempool addresses whose mempool history has changed
-        new_mempool_hist_keys = new_mempool_hist.keys()
-        self_mempool_hist_keys = self.mempool_hist.keys()
-        
-        for addr in new_mempool_hist_keys:
-            if addr in self_mempool_hist_keys:
-                if self.mempool_hist[addr] != new_mempool_hist[addr]:
-                    self.invalidate_cache(addr)
-            else:
-                self.invalidate_cache(addr)
-
-        # invalidate cache for addresses that are removed from mempool ?
-        # this should not be necessary if they go into a block, but they might not
-        for addr in self_mempool_hist_keys:
-            if addr not in new_mempool_hist_keys:
-                self.invalidate_cache(addr)
-        
-
         with self.mempool_lock:
             self.mempool_hist = new_mempool_hist
 
+        # invalidate cache for touched addresses
+        for addr in touched_addresses:
+            self.invalidate_cache(addr)
+
+
     def invalidate_cache(self, address):
         with self.cache_lock:
             if address in self.history_cache:
@@ -879,6 +878,7 @@ class BlockchainProcessor(Processor):
                 self.history_cache.pop(address)
 
         if address in self.watched_addresses:
+            # TODO: update cache here. if new value equals cached value, do not send notification
             self.address_queue.put(address)
 
     def main_iteration(self):
@@ -892,8 +892,6 @@ class BlockchainProcessor(Processor):
             t2 = time.time()
 
         self.memorypool_update()
-        t3 = time.time()
-        # print "mempool:", len(self.mempool_addresses), len(self.mempool_hist), "%.3fs"%(t3 - t2)
 
         if self.sent_height != self.height:
             self.sent_height = self.height