Skip to content

#265 indexer fixes #266

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
s-medvedev merged 2 commits into from
Nov 9, 2021
Merged

#265 indexer fixes #266

Changes from 1 commit
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
c36c1cb
#265 fixes
otselnik Nov 4, 2021
dec3483
#265 config defined timeout
otselnik Nov 4, 2021
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Next Next commit
#265 fixes
  • Loading branch information
@otselnik
otselnik committed Nov 4, 2021
commit c36c1cbb9957bff4bb595130b977c80e4b48fd4b
121 changes: 43 additions & 78 deletions proxy/indexer/solana_receipts_update.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -45,7 +45,7 @@ def __init__(self, signature, results, accounts = None):


class TransactionStruct:
def __init__(self, eth_trx, eth_signature, from_address, got_result, signatures, storage, blocked_accounts):
def __init__(self, eth_trx, eth_signature, from_address, got_result, signatures, storage, blocked_accounts, slot):
# logger.debug(eth_signature)
self.eth_trx = eth_trx
self.eth_signature = eth_signature
Expand All @@ -54,6 +54,7 @@ def __init__(self, eth_trx, eth_signature, from_address, got_result, signatures,
self.signatures = signatures
self.storage = storage
self.blocked_accounts = blocked_accounts
self.slot = slot


class Indexer:
Expand Down Expand Up @@ -271,6 +272,7 @@ def process_receipts(self):
continue_result.signatures,
storage_account,
continue_result.accounts,
slot
)

del continue_table[storage_account]
Expand Down Expand Up @@ -327,7 +329,16 @@ def process_receipts(self):
got_result = get_trx_results(trx)
if got_result is not None:
# self.submit_transaction(eth_trx, eth_signature, from_address, got_result, [signature])
trx_table[eth_signature] = TransactionStruct(eth_trx, eth_signature, from_address, got_result, [signature], None, None)
trx_table[eth_signature] = TransactionStruct(
eth_trx,
eth_signature,
from_address,
got_result,
[signature],
None,
None,
slot
)
else:
logger.error("RESULT NOT FOUND IN 05\n{}".format(json.dumps(trx, indent=4, sort_keys=True)))

Expand Down Expand Up @@ -357,7 +368,8 @@ def process_receipts(self):
None,
[signature],
storage_account,
blocked_accounts
blocked_accounts,
slot
)

if storage_account in continue_table:
Expand All @@ -369,11 +381,15 @@ def process_receipts(self):
trx_table[eth_signature].signatures = continue_result.signatures
del continue_table[storage_account]

elif instruction_data[0] == 0x0a: # Continue
# logger.debug("{:>10} {:>6} Continue 0x{}".format(slot, counter, instruction_data.hex()))
elif instruction_data[0] == 0x0a or instruction_data[0] == 0x14: # Continue or ContinueV02

storage_account = trx['transaction']['message']['accountKeys'][instruction['accounts'][0]]
blocked_accounts = [trx['transaction']['message']['accountKeys'][acc_idx] for acc_idx in instruction['accounts'][5:]]
if instruction_data[0] == 0x0a:
# logger.debug("{:>10} {:>6} Continue 0x{}".format(slot, counter, instruction_data.hex()))
blocked_accounts = [trx['transaction']['message']['accountKeys'][acc_idx] for acc_idx in instruction['accounts'][5:]]
if instruction_data[0] == 0x14:
# logger.debug("{:>10} {:>6} ContinueV02 0x{}".format(slot, counter, instruction_data.hex()))
blocked_accounts = [trx['transaction']['message']['accountKeys'][acc_idx] for acc_idx in instruction['accounts'][5:]]
got_result = get_trx_results(trx)

if storage_account in continue_table:
Expand Down Expand Up @@ -406,7 +422,9 @@ def process_receipts(self):
continue_table[storage_account].signatures.append(signature)

if holder_account in holder_table:
holder_table[holder_account] = HolderStruct(storage_account)
if holder_table[holder_account].storage_account != storage_account:
logger.error("Strange behavior. Pay attention. STORAGE_ACCOUNT != STORAGE_ACCOUNT")
holder_table[holder_account] = HolderStruct(storage_account)
else:
holder_table[holder_account] = HolderStruct(storage_account)
else:
Expand Down Expand Up @@ -440,7 +458,16 @@ def process_receipts(self):
if eth_signature in trx_table:
trx_table[eth_signature].signatures.append(signature)
else:
trx_table[eth_signature] = TransactionStruct(eth_trx, eth_signature, from_address, got_result, [signature], storage_account, blocked_accounts)
trx_table[eth_signature] = TransactionStruct(
eth_trx,
eth_signature,
from_address,
got_result,
[signature],
storage_account,
blocked_accounts,
slot
)

elif instruction_data[0] == 0x0e:
# logger.debug("{:>10} {:>6} ExecuteTrxFromAccountDataIterativeOrContinue 0x{}".format(slot, counter, instruction_data.hex()))
Expand All @@ -454,9 +481,11 @@ def process_receipts(self):
continue_table[storage_account].signatures.append(signature)

if holder_account in holder_table:
logger.error("Strange behavior. Pay attention. HOLDER ACCOUNT FOUND")
holder_table[holder_account] = HolderStruct(storage_account)
if holder_table[holder_account].storage_account != storage_account:
logger.error("Strange behavior. Pay attention. STORAGE_ACCOUNT != STORAGE_ACCOUNT")
holder_table[holder_account] = HolderStruct(storage_account)
else:
logger.error("Strange behavior. Pay attention. HOLDER ACCOUNT NOT FOUND")
holder_table[holder_account] = HolderStruct(storage_account)

if got_result:
Expand All @@ -467,67 +496,8 @@ def process_receipts(self):

continue_table[storage_account].results = got_result
else:
got_result = get_trx_results(trx)
if got_result is not None:
continue_table[storage_account] = ContinueStruct(signature, got_result, blocked_accounts)
holder_table[holder_account] = HolderStruct(storage_account)
else:
self.add_hunged_storage(trx, storage_account)

elif instruction_data[0] == 0x13: # PartialCallFromRawEthereumTXv02
# logger.debug("{:>10} {:>6} PartialCallFromRawEthereumTXv02 0x{}".format(slot, counter, instruction_data.hex()))

storage_account = trx['transaction']['message']['accountKeys'][instruction['accounts'][0]]

if storage_account in continue_table:
# collateral_pool_buf = instruction_data[1:5]
# step_count = instruction_data[5:13]
# from_addr = instruction_data[13:33]

sign = instruction_data[33:98]
unsigned_msg = instruction_data[98:]

(eth_trx, eth_signature, from_address) = get_trx_receipts(unsigned_msg, sign)

continue_result = continue_table[storage_account]

self.submit_transaction(eth_trx, eth_signature, from_address, continue_result.results, continue_result.signatures)

del continue_table[storage_account]
else:
self.add_hunged_storage(trx, storage_account)

elif instruction_data[0] == 0x14: # ContinueV02
# logger.debug("{:>10} {:>6} ContinueV02 0x{}".format(slot, counter, instruction_data.hex()))

storage_account = trx['transaction']['message']['accountKeys'][instruction['accounts'][0]]

if storage_account in continue_table:
continue_table[storage_account].signatures.append(signature)
else:
got_result = get_trx_results(trx)
if got_result is not None:
continue_table[storage_account] = ContinueStruct(signature, got_result)
else:
self.add_hunged_storage(trx, storage_account)

elif instruction_data[0] == 0x16: # ExecuteTrxFromAccountDataIterativeV02
# logger.debug("{:>10} {:>6} ExecuteTrxFromAccountDataIterativeV02 0x{}".format(slot, counter, instruction_data.hex()))

holder_account = trx['transaction']['message']['accountKeys'][instruction['accounts'][0]]
storage_account = trx['transaction']['message']['accountKeys'][instruction['accounts'][1]]

if storage_account in continue_table:
continue_table[storage_account].signatures.append(signature)

if holder_account in holder_table:
# logger.debug("holder_account found")
# logger.debug("Strange behavior. Pay attention.")
holder_table[holder_account] = HolderStruct(storage_account)
else:
holder_table[holder_account] = HolderStruct(storage_account)
else:
self.add_hunged_storage(trx, storage_account)
continue_table[storage_account] = ContinueStruct(signature, got_result, blocked_accounts)
holder_table[holder_account] = HolderStruct(storage_account)

if instruction_data[0] > 0x16:
logger.debug("{:>10} {:>6} Unknown 0x{}".format(slot, counter, instruction_data.hex()))
Expand All @@ -538,12 +508,12 @@ def process_receipts(self):
if trx_struct.got_result:
self.submit_transaction(trx_struct)
elif trx_struct.storage:
self.blocked_storages[trx_struct.storage] = (trx_struct.eth_trx, trx_struct.blocked_accounts)
if abs(trx_struct.slot - self.current_slot) > 16:
self.blocked_storages[trx_struct.storage] = (trx_struct.eth_trx, trx_struct.blocked_accounts)
else:
logger.error(trx_struct)



def submit_transaction(self, trx_struct):
(logs, status, gas_used, return_value, slot) = trx_struct.got_result
(_slot, block_hash) = self.get_block(slot)
Expand Down Expand Up @@ -605,11 +575,6 @@ def get_block(self, slot):
return (slot, block_hash)


def add_hunged_storage(self, trx, storage):
if abs(trx['slot'] - self.current_slot) > 16:
self.blocked_storages.add(storage)


def run_indexer():
logging.basicConfig(format='%(asctime)s - pid:%(process)d [%(levelname)-.1s] %(funcName)s:%(lineno)d - %(message)s')
logger.setLevel(logging.DEBUG)
Expand Down