Airdropper: Store latest processed slot, use ascending slot order

proxy/__main__.py

@@ -29,16 +29,7 @@
         log_level = os.environ['LOG_LEVEL']
         neon_decimals = int(os.environ.get('NEON_DECIMALS', '9'))
 
-        start_slot = os.environ.get('START_SLOT', None)
-        finalized = os.environ.get('FINALIZED', 'finalized')
-        if start_slot == 'LATEST':
-            client = Client(solana_url)
-            start_slot = client.get_slot(commitment=finalized)["result"]
-        if start_slot is None: # by default
-            start_slot = 0
-        else: # try to convert into integer
-            start_slot = int(start_slot)
-
+        start_slot = os.environ.get('START_SLOT', 0)
         pp_solana_url = os.environ.get('PP_SOLANA_URL', None)
         max_conf = float(os.environ.get('MAX_CONFIDENCE_INTERVAL', 0.02))
 

proxy/indexer/airdropper.py

@@ -1,4 +1,3 @@
-from time import time
 from solana.publickey import PublicKey
 from proxy.indexer.indexer_base import IndexerBase, logger
 from proxy.indexer.pythnetwork import PythNetworkClient
@@ -8,6 +7,7 @@
 import logging
 from datetime import datetime
 from decimal import Decimal
+import os
 
 try:
     from utils import check_error
@@ -20,7 +20,7 @@
 AIRDROP_AMOUNT_SOL = ACCOUNT_CREATION_PRICE_SOL / 2
 NEON_PRICE_USD = Decimal('0.25')
 
-
+FINALIZED = os.environ.get('FINALIZED', 'finalized')
 
 class Airdropper(IndexerBase):
     def __init__(self,
@@ -34,8 +34,26 @@ def __init__(self,
                  start_slot = 0,
                  pp_solana_url = None,
                  max_conf = 0.1): # maximum confidence interval deviation related to price
+        self._constants = SQLDict(tablename="constants")
+        if start_slot == 'CONTINUE':
+            logger.info('Trying to use latest processed slot from previous run')
+            start_slot = self._constants.get('latest_processed_slot', 0)
+        elif start_slot == 'LATEST':
+            logger.info('Airdropper will start at latest blockchain slot')
+            client = SolanaClient(solana_url)
+            start_slot = client.get_slot(commitment=FINALIZED)["result"]
+        else:
+            try:
+                start_slot = int(start_slot)
+            except Exception as err:
+                logger.warning(f'''Unsupported value for start_slot: {start_slot}. 
+                Must be either integer value or one of [CONTINUE,LATEST]''')
+                raise
+        logger.info(f'Start slot is {start_slot}')
+
+
         IndexerBase.__init__(self, solana_url, evm_loader_id, log_level, start_slot)
-        self.latest_processed_slot = 0
+        self.latest_processed_slot = start_slot
 
         # collection of eth-address-to-create-accout-trx mappings
         # for every addresses that was already funded with airdrop
@@ -161,7 +179,7 @@ def find_instructions(trx, predicate):
     def get_sol_usd_price(self):
         should_reload = self.always_reload_price
         if not should_reload:
-            if self.recent_price == None or self.recent_price['valid_slot'] < self.latest_processed_slot:
+            if self.recent_price == None or self.recent_price['valid_slot'] < self.current_slot:
                 should_reload = True
 
         if should_reload:
@@ -238,11 +256,12 @@ def process_functions(self):
 
     def process_receipts(self):
         max_slot = 0
-        for slot, _, trx in self.transaction_receipts.get_trxs(self.latest_processed_slot, reverse=True):
+        for slot, _, trx in self.transaction_receipts.get_trxs(self.latest_processed_slot):
             max_slot = max(max_slot, slot)
             if trx['transaction']['message']['instructions'] is not None:
                 self.process_trx_airdropper_mode(trx)
         self.latest_processed_slot = max(self.latest_processed_slot, max_slot)
+        self._constants['latest_processed_slot'] = self.latest_processed_slot
 
 
 def run_airdropper(solana_url,

proxy/testing/test_airdropper.py

@@ -1,7 +1,8 @@
+from multiprocessing.connection import Client
 import unittest
 
+from solana.rpc.api import Client as SolanaClient
 from solana.publickey import PublicKey
-from proxy.indexer.pythnetwork import PythNetworkClient
 from proxy.testing.mock_server import MockServer
 from proxy.indexer.airdropper import Airdropper, AIRDROP_AMOUNT_SOL, NEON_PRICE_USD
 from proxy.indexer.sql_dict import SQLDict
@@ -51,22 +52,30 @@ def create_price_info(valid_slot: int, price: Decimal, conf: Decimal):
 
 
 class Test_Airdropper(unittest.TestCase):
+    def create_airdropper(self, start_slot):
+        return Airdropper(solana_url         =f'http://{self.address}:8899',
+                          evm_loader_id       =self.evm_loader_id,
+                          pyth_mapping_account=self.pyth_mapping_account,
+                          faucet_url          =f'http://{self.address}:{self.faucet_port}',
+                          wrapper_whitelist   =self.wrapper_whitelist,
+                          log_level           ='INFO',
+                          neon_decimals       =self.neon_decimals,
+                          start_slot          =start_slot)
+
     @classmethod
-    def setUpClass(cls) -> None:
+    @patch.object(SQLDict, 'get')
+    @patch.object(SolanaClient, 'get_slot')
+    def setUpClass(cls, mock_get_slot, mock_dict_get) -> None:
         print("testing indexer in airdropper mode")
         cls.address = 'localhost'
         cls.faucet_port = 3333
         cls.evm_loader_id = evm_loader_addr
         cls.pyth_mapping_account = PublicKey(b'TestMappingAccount')
         cls.wrapper_whitelist = wrapper_whitelist
         cls.neon_decimals = 9
-        cls.airdropper = Airdropper(solana_url          =f'http://{cls.address}:8899',
-                                    evm_loader_id       =cls.evm_loader_id,
-                                    pyth_mapping_account=cls.pyth_mapping_account,
-                                    faucet_url          =f'http://{cls.address}:{cls.faucet_port}',
-                                    wrapper_whitelist   =cls.wrapper_whitelist,
-                                    log_level           ='INFO',
-                                    neon_decimals       =cls.neon_decimals)
+        cls.airdropper = cls.create_airdropper(cls, 0)
+        mock_get_slot.assert_not_called()
+        mock_dict_get.assert_not_called()
 
         cls.airdropper.always_reload_price = True
         cls.mock_airdrop_ready = Mock()
@@ -224,4 +233,25 @@ def test_get_price_error(self):
             self.mock_pyth_client.update_mapping.assert_called_once()
             self.mock_pyth_client.get_price.assert_called_once_with('SOL/USD')
         except Exception as err:
-            self.fail(f'Excpected not throws exception but it does: {err}')
+            self.fail(f'Excpected not throws exception but it does: {err}')
+
+    @patch.object(SQLDict, 'get')
+    @patch.object(SolanaClient, 'get_slot')
+    def test_init_airdropper_slot_continue(self, mock_get_slot, mock_dict_get):
+        start_slot = 1234
+        mock_dict_get.side_effect = [start_slot]
+        new_airdropper = self.create_airdropper('CONTINUE')
+        self.assertEqual(new_airdropper.latest_processed_slot, start_slot)
+        mock_dict_get.assert_called_once_with('latest_processed_slot', ANY)
+        mock_get_slot.assert_not_called()
+
+
+    @patch.object(SQLDict, 'get')
+    @patch.object(SolanaClient, 'get_slot')
+    def test_init_airdropper_slot_latest(self, mock_get_slot, mock_dict_get):
+        start_slot = 1234
+        mock_get_slot.side_effect = [{'result': start_slot}]
+        new_airdropper = self.create_airdropper('LATEST')
+        self.assertEqual(new_airdropper.latest_processed_slot, start_slot)
+        mock_get_slot.assert_called_once_with(commitment='finalized')
+        mock_dict_get.assert_not_called()