Changes in base algorithm

proxy/indexer/airdropper.py

@@ -1,10 +1,13 @@
+from web3 import eth
 from proxy.indexer.indexer_base import IndexerBase, logger
 from proxy.indexer.price_provider import PriceProvider, mainnet_solana, mainnet_price_accounts
+from typing import List, Dict
 import os
 import requests
 import base58
 import json
 import logging
+from datetime import date, datetime
 
 try:
     from utils import check_error
@@ -17,6 +20,7 @@
 AIRDROP_AMOUNT_SOL = ACCOUNT_CREATION_PRICE_SOL / 2
 NEON_PRICE_USD = 0.25
 
+
 class Airdropper(IndexerBase):
     def __init__(self,
                  solana_url,
@@ -32,6 +36,7 @@ def __init__(self,
         # collection of eth-address-to-create-accout-trx mappings
         # for every addresses that was already funded with airdrop
         self.airdrop_ready = SQLDict(tablename="airdrop_ready")
+        self.airdrop_scheduled = SQLDict(tablename="airdrop_scheduled")
         self.wrapper_whitelist = wrapper_whitelist
         self.faucet_url = faucet_url
 
@@ -72,15 +77,18 @@ def _check_transfer(self, account_keys, create_token_acc, token_transfer) -> boo
         return account_keys[create_token_acc['accounts'][1]] == account_keys[token_transfer['accounts'][1]]
 
 
-    def _airdrop_to(self, create_acc):
+    def _schedule_airdrop(self, create_acc):
         eth_address = "0x" + bytearray(base58.b58decode(create_acc['data'])[20:][:20]).hex()
         if eth_address in self.airdrop_ready:  # transaction already processed
             return
+        self.airdrop_scheduled[eth_address] = { 'scheduled': datetime.now().timestamp() }
+
 
+    def _airdrop_to(self, eth_address, schedule_time):
         sol_price_usd = self.price_provider.get_price('SOL/USD')
         if sol_price_usd is None:
             logger.warning("Failed to get SOL/USD price")
-            return
+            return False
 
         logger.info(f'SOL/USD = ${sol_price_usd}')
         airdrop_amount_usd = AIRDROP_AMOUNT_SOL * sol_price_usd
@@ -94,12 +102,15 @@ def _airdrop_to(self, create_acc):
         resp = requests.post(self.faucet_url + '/request_neon_in_galans', json = json_data)
         if not resp.ok:
             logger.warning(f'Failed to airdrop: {resp.status_code}')
-            return
+            return False
 
-        self.airdrop_ready[eth_address] = create_acc
+        self.airdrop_ready[eth_address] = { 'amount': airdrop_galans, 
+                                            'scheduled': schedule_time,
+                                            'finished': datetime.now().timestamp() }
+        return True
 
 
-    def process_trx_airdropper_mode(self, trx):
+    def _process_trx_airdropper_mode(self, trx):
         if check_error(trx):
             return
 
@@ -133,30 +144,29 @@ def find_instructions(trx, predicate):
                 for token_transfer in token_transfer_list:
                     if not self._check_transfer(account_keys, create_token_acc, token_transfer):
                         continue
-                    self._airdrop_to(create_acc)
+                    self._schedule_airdrop(create_acc)
+
+
+    def _process_scheduled_trxs(self):
+        success_addresses = set()
+        for eth_address, sched_info in self.airdrop_scheduled.items():
+            if not self._airdrop_to(eth_address, sched_info['scheduled']):
+                continue
+            success_addresses.add(eth_address)
+
+        for eth_address in success_addresses:
+            del self.airdrop_scheduled[eth_address]
 
 
     def process_functions(self):
         IndexerBase.process_functions(self)
-        logger.debug("Process receipts")
-        self.process_receipts()
-
-
-    def process_receipts(self):
-        counter = 0
-        for signature in self.transaction_order:
-            counter += 1
-            if signature in self.transaction_receipts:
-                trx = self.transaction_receipts[signature]
-                if trx is None:
-                    logger.error("trx is None")
-                    del self.transaction_receipts[signature]
-                    continue
-                if 'slot' not in trx:
-                    logger.debug("\n{}".format(json.dumps(trx, indent=4, sort_keys=True)))
-                    exit()
-                if trx['transaction']['message']['instructions'] is not None:
-                    self.process_trx_airdropper_mode(trx)
+        logger.debug("Airdropping...")
+        self._process_scheduled_trxs()
+
+
+    def handle_new_transactions(self, ordered_signs, receipts):
+        for signature in ordered_signs:
+            self._process_trx_airdropper_mode(receipts[signature]) 
 
 
 def run_airdropper(solana_url,

proxy/indexer/indexer.py

@@ -7,6 +7,7 @@
 import time
 import logging
 from multiprocessing.dummy import Pool as ThreadPool
+from typing import List, Dict
 
 
 try:
@@ -65,6 +66,8 @@ def __init__(self,
         if 'last_block' not in self.constants:
             self.constants['last_block'] = 0
         self.blocked_storages = {}
+        self.transaction_order = []
+        self.transaction_receipts = SQLDict(tablename="known_transactions")
 
 
     def process_functions(self):
@@ -78,6 +81,29 @@ def process_functions(self):
         self.blocked_storages = {}
 
 
+    def handle_new_transactions(self, ordered_signs, receipts):
+        # Combine transactions received just now with previous ones
+        for signature in ordered_signs:
+            if signature not in self.transaction_receipts:
+                trx = receipts.get(signature, None)
+                if not trx: #Normally, this will not happen (means error in upper level logic)
+                    logger.error(f'Signature {signature} not found in receipts dictionary')
+                    continue
+                self.transaction_receipts[signature] = trx
+
+        # merge current transaction sequence with rest of the history
+        if len(self.transaction_order):
+            index = 0
+            try:
+                index = ordered_signs.index(self.transaction_order[0])
+            except ValueError:
+                self.transaction_order = ordered_signs + self.transaction_order
+            else:
+                self.transaction_order = ordered_signs[:index] + self.transaction_order
+        else:
+            self.transaction_order = ordered_signs
+
+
     def process_receipts(self):
         counter = 0
         holder_table = {}

proxy/indexer/indexer_base.py

@@ -3,7 +3,7 @@
 import logging
 from solana.rpc.api import Client
 from multiprocessing.dummy import Pool as ThreadPool
-from typing import Dict, Union
+from typing import Dict, List, Set, Union
 
 try:
     from sql_dict import SQLDict
@@ -40,10 +40,9 @@ def __init__(self,
 
         self.evm_loader_id = evm_loader_id
         self.client = Client(solana_url)
-        self.transaction_receipts = SQLDict(tablename="known_transactions")
         self.last_slot = start_slot
         self.current_slot = 0
-        self.transaction_order = []
+
         self.counter_ = 0
 
 
@@ -62,7 +61,7 @@ def process_functions(self):
 
     def gather_unknown_transactions(self):
         poll_txs = set()
-        ordered_txs = []
+        ordered_signs = []
 
         minimal_tx = None
         continue_flag = True
@@ -93,10 +92,8 @@ def gather_unknown_transactions(self):
                     continue_flag = False
                     break
 
-                ordered_txs.append(solana_signature)
-
-                if solana_signature not in self.transaction_receipts:
-                    poll_txs.add(solana_signature)
+                ordered_signs.append(solana_signature)
+                poll_txs.add(solana_signature)
 
                 if slot < minimal_slot:
                     minimal_slot = slot
@@ -111,33 +108,26 @@ def gather_unknown_transactions(self):
 
         logger.debug("start getting receipts")
         pool = ThreadPool(PARALLEL_REQUESTS)
-        pool.map(self.get_tx_receipts, poll_txs)
-
-        if len(self.transaction_order):
-            index = 0
-            try:
-                index = ordered_txs.index(self.transaction_order[0])
-            except ValueError:
-                self.transaction_order = ordered_txs + self.transaction_order
-            else:
-                self.transaction_order = ordered_txs[:index] + self.transaction_order
-        else:
-            self.transaction_order = ordered_txs
+        receipts = { entry[0]:entry[1] for entry in pool.map(self.get_tx_receipts, poll_txs) } 
 
         self.last_slot = maximum_slot
         self.current_slot = current_slot
-
         self.counter_ = 0
 
+        logger.debug("Start processing received receipts")
+        self.handle_new_transactions(ordered_signs, receipts)
+
+
+    def handle_new_transactions(self, ordered_signs, receipts): None
+
 
     def get_tx_receipts(self, solana_signature):
-        # trx = None
+        trx = None
         retry = True
 
         while retry:
             try:
                 trx = self.client.get_confirmed_transaction(solana_signature)['result']
-                self.transaction_receipts[solana_signature] = trx
                 retry = False
             except Exception as err:
                 logger.debug(err)
@@ -147,4 +137,4 @@ def get_tx_receipts(self, solana_signature):
         if self.counter_ % 100 == 0:
             logger.debug(self.counter_)
 
-        # return (solana_signature, trx)
+        return (solana_signature, trx)