Skip to content

Refactor extrinsic receipt event parsing with shared helpers #287

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
thewhaleking merged 3 commits into from
Apr 7, 2026
Merged

Refactor extrinsic receipt event parsing with shared helpers #287

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
e3ca5e2
Add sync and async extrinsic receipt event parsing tests
it-education-md Apr 2, 2026
0803855
Refactor sync and async extrinsic receipt event parsing with shared h…
it-education-md Apr 2, 2026
b83f09f
Merge branch 'staging' into refactor/receipt-event-parsing
thewhaleking Apr 7, 2026
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
177 changes: 57 additions & 120 deletions async_substrate_interface/async_substrate.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -78,6 +78,16 @@
convert_account_ids,
decode_query_map_async,
)
from async_substrate_interface.utils.receipt import (
build_system_error_message,
extract_failure_details,
extract_fallback_deposit_fee_amount,
extract_success_weight,
extract_total_fee_amount,
is_extrinsic_failure_event,
is_extrinsic_success_event,
normalize_module_error,
)
from async_substrate_interface.utils.storage import StorageKey
from async_substrate_interface.type_registry import _TYPE_REGISTRY

Expand Down Expand Up @@ -261,142 +271,69 @@ async def process_events(self):
if await self.triggered_events:
self.__total_fee_amount = 0

# Process fees
has_transaction_fee_paid_event = False

for event in await self.triggered_events:
if (
event["event"]["module_id"] == "TransactionPayment"
and event["event"]["event_id"] == "TransactionFeePaid"
):
self.__total_fee_amount = event["event"]["attributes"]["actual_fee"]
has_transaction_fee_paid_event = True
events = await self.triggered_events
self.__total_fee_amount, has_transaction_fee_paid_event = (
extract_total_fee_amount(events)
)

# Process other events
possible_success = False
for event in await self.triggered_events:
# TODO make this more readable
# Check events
if (
event["event"]["module_id"] == "System"
and event["event"]["event_id"] == "ExtrinsicSuccess"
):
for event in events:
if is_extrinsic_success_event(event):
possible_success = True

if "dispatch_info" in event["event"]["attributes"]:
self.__weight = event["event"]["attributes"]["dispatch_info"][
"weight"
]
else:
# Backwards compatibility
self.__weight = event["event"]["attributes"]["weight"]

elif (
event["event"]["module_id"] == "System"
and event["event"]["event_id"] == "ExtrinsicFailed"
) or (
event["event"]["module_id"] == "MevShield"
and event["event"]["event_id"]
in ("DecryptedRejected", "DecryptionFailed")
):
self.__weight = extract_success_weight(event)
elif is_extrinsic_failure_event(event):
possible_success = False
self.__is_success = False
failure_details = extract_failure_details(event)
if failure_details["has_weight"]:
self.__weight = failure_details["weight"]
if failure_details["error_message"] is not None:
self.__error_message = failure_details["error_message"]
continue

if event["event"]["module_id"] == "System":
dispatch_info = event["event"]["attributes"]["dispatch_info"]
dispatch_error = event["event"]["attributes"]["dispatch_error"]
self.__weight = dispatch_info["weight"]
else:
# MEV shield extrinsics
if event["event"]["event_id"] == "DecryptedRejected":
dispatch_info = event["event"]["attributes"]["reason"][
"post_info"
]
dispatch_error = event["event"]["attributes"]["reason"][
"error"
]
self.__weight = event["event"]["attributes"]["reason"][
"post_info"
]["actual_weight"]
else:
self.__error_message = {
"type": "MevShield",
"name": "DecryptionFailed",
"docs": event["event"]["attributes"]["reason"],
}
continue

if "Module" in dispatch_error:
if isinstance(dispatch_error["Module"], tuple):
module_index = dispatch_error["Module"][0]
error_index = dispatch_error["Module"][1]
else:
module_index = dispatch_error["Module"]["index"]
error_index = dispatch_error["Module"]["error"]

if isinstance(error_index, str):
# Actual error index is first u8 in new [u8; 4] format
error_index = int(error_index[2:4], 16)
dispatch_error = failure_details["dispatch_error"]
if dispatch_error is None:
continue

if self.block_hash:
runtime = await self.substrate.init_runtime(
block_hash=self.block_hash
)
else:
runtime = await self.substrate.init_runtime(
block_id=self.block_number
module_error = normalize_module_error(dispatch_error)
if module_error is not None:
self.__error_message = (
await self._resolve_module_error_message(
module_index=module_error["module_index"],
error_index=module_error["error_index"],
)
module_error = runtime.metadata.get_module_error(
module_index=module_index, error_index=error_index
)
self.__error_message = {
"type": "Module",
"name": module_error.name,
"docs": module_error.docs,
}
elif "BadOrigin" in dispatch_error:
self.__error_message = {
"type": "System",
"name": "BadOrigin",
"docs": "Bad origin",
}
elif "CannotLookup" in dispatch_error:
self.__error_message = {
"type": "System",
"name": "CannotLookup",
"docs": "Cannot lookup",
}
elif "Other" in dispatch_error:
self.__error_message = {
"type": "System",
"name": "Other",
"docs": "Unspecified error occurred",
}
elif "Token" in dispatch_error:
self.__error_message = {
"type": "System",
"name": "Token",
"docs": dispatch_error["Token"],
}

else:
self.__error_message = build_system_error_message(
dispatch_error
)
elif not has_transaction_fee_paid_event:
if (
event["event"]["module_id"] == "Treasury"
and event["event"]["event_id"] == "Deposit"
):
self.__total_fee_amount += event["event"]["attributes"]["value"]
elif (
event["event"]["module_id"] == "Balances"
and event["event"]["event_id"] == "Deposit"
):
self.__total_fee_amount += event["event"]["attributes"][
"amount"
]
self.__total_fee_amount += extract_fallback_deposit_fee_amount(
event
)
if possible_success is True and self.__error_message is None:
# we delay the positive setting of the __is_success flag until we have finished iteration of the
# events and have ensured nothing has set an error message
self.__is_success = True

async def _resolve_module_error_message(
self, module_index: int, error_index: int
) -> dict:
if self.block_hash:
runtime = await self.substrate.init_runtime(block_hash=self.block_hash)
else:
runtime = await self.substrate.init_runtime(block_id=self.block_number)

module_error = runtime.metadata.get_module_error(
module_index=module_index, error_index=error_index
)
return {
"type": "Module",
"name": module_error.name,
"docs": module_error.docs,
}

@property
async def is_success(self) -> bool:
"""
Expand Down
161 changes: 48 additions & 113 deletions async_substrate_interface/sync_substrate.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -50,6 +50,16 @@
legacy_scale_decode,
convert_account_ids,
)
from async_substrate_interface.utils.receipt import (
build_system_error_message,
extract_failure_details,
extract_fallback_deposit_fee_amount,
extract_success_weight,
extract_total_fee_amount,
is_extrinsic_failure_event,
is_extrinsic_success_event,
normalize_module_error,
)
from async_substrate_interface.utils.storage import StorageKey
from async_substrate_interface.type_registry import _TYPE_REGISTRY

Expand Down Expand Up @@ -223,134 +233,59 @@ def process_events(self):
if self.triggered_events:
self.__total_fee_amount = 0

# Process fees
has_transaction_fee_paid_event = False

for event in self.triggered_events:
if (
event["event"]["module_id"] == "TransactionPayment"
and event["event"]["event_id"] == "TransactionFeePaid"
):
self.__total_fee_amount = event["event"]["attributes"]["actual_fee"]
has_transaction_fee_paid_event = True
self.__total_fee_amount, has_transaction_fee_paid_event = (
extract_total_fee_amount(self.triggered_events)
)

# Process other events
possible_success = False
for event in self.triggered_events:
# TODO make this more readable
# Check events
if (
event["event"]["module_id"] == "System"
and event["event"]["event_id"] == "ExtrinsicSuccess"
):
if is_extrinsic_success_event(event):
possible_success = True

if "dispatch_info" in event["event"]["attributes"]:
self.__weight = event["event"]["attributes"]["dispatch_info"][
"weight"
]
else:
# Backwards compatibility
self.__weight = event["event"]["attributes"]["weight"]

elif (
event["event"]["module_id"] == "System"
and event["event"]["event_id"] == "ExtrinsicFailed"
) or (
event["event"]["module_id"] == "MevShield"
and event["event"]["event_id"]
in ("DecryptedRejected", "DecryptionFailed")
):
self.__weight = extract_success_weight(event)
elif is_extrinsic_failure_event(event):
possible_success = False
self.__is_success = False

if event["event"]["module_id"] == "System":
dispatch_info = event["event"]["attributes"]["dispatch_info"]
dispatch_error = event["event"]["attributes"]["dispatch_error"]
self.__weight = dispatch_info["weight"]
failure_details = extract_failure_details(event)
if failure_details["has_weight"]:
self.__weight = failure_details["weight"]
if failure_details["error_message"] is not None:
self.__error_message = failure_details["error_message"]
continue

dispatch_error = failure_details["dispatch_error"]
if dispatch_error is None:
continue

module_error = normalize_module_error(dispatch_error)
if module_error is not None:
self.__error_message = self._resolve_module_error_message(
module_index=module_error["module_index"],
error_index=module_error["error_index"],
)
else:
# MEV shield extrinsics
if event["event"]["event_id"] == "DecryptedRejected":
dispatch_info = event["event"]["attributes"]["reason"][
"post_info"
]
dispatch_error = event["event"]["attributes"]["reason"][
"error"
]
self.__weight = event["event"]["attributes"]["reason"][
"post_info"
]["actual_weight"]
else:
self.__error_message = {
"type": "MevShield",
"name": "DecryptionFailed",
"docs": event["event"]["attributes"]["reason"],
}
continue

if "Module" in dispatch_error:
if isinstance(dispatch_error["Module"], tuple):
module_index = dispatch_error["Module"][0]
error_index = dispatch_error["Module"][1]
else:
module_index = dispatch_error["Module"]["index"]
error_index = dispatch_error["Module"]["error"]

if isinstance(error_index, str):
# Actual error index is first u8 in new [u8; 4] format
error_index = int(error_index[2:4], 16)

module_error = self.substrate.metadata.get_module_error(
module_index=module_index, error_index=error_index
self.__error_message = build_system_error_message(
dispatch_error
)
self.__error_message = {
"type": "Module",
"name": module_error.name,
"docs": module_error.docs,
}
elif "BadOrigin" in dispatch_error:
self.__error_message = {
"type": "System",
"name": "BadOrigin",
"docs": "Bad origin",
}
elif "CannotLookup" in dispatch_error:
self.__error_message = {
"type": "System",
"name": "CannotLookup",
"docs": "Cannot lookup",
}
elif "Other" in dispatch_error:
self.__error_message = {
"type": "System",
"name": "Other",
"docs": "Unspecified error occurred",
}
elif "Token" in dispatch_error:
self.__error_message = {
"type": "System",
"name": "Token",
"docs": dispatch_error["Token"],
}

elif not has_transaction_fee_paid_event:
if (
event["event"]["module_id"] == "Treasury"
and event["event"]["event_id"] == "Deposit"
):
self.__total_fee_amount += event["event"]["attributes"]["value"]
elif (
event["event"]["module_id"] == "Balances"
and event["event"]["event_id"] == "Deposit"
):
self.__total_fee_amount += event["event"]["attributes"][
"amount"
]
self.__total_fee_amount += extract_fallback_deposit_fee_amount(
event
)
if possible_success is True and self.__error_message is None:
# we delay the positive setting of the __is_success flag until we have finished iteration of the
# events and have ensured nothing has set an error message
self.__is_success = True

def _resolve_module_error_message(self, module_index: int, error_index: int) -> dict:
module_error = self.substrate.metadata.get_module_error(
module_index=module_index, error_index=error_index
)
return {
"type": "Module",
"name": module_error.name,
"docs": module_error.docs,
}

@property
def is_success(self) -> bool:
"""
Expand Down
Loading
Loading