diff --git a/backend/node/base.py b/backend/node/base.py index ffa967cdb..dc343d4fb 100644 --- a/backend/node/base.py +++ b/backend/node/base.py @@ -990,7 +990,7 @@ async def _run_genvm( self.timing_callback("GENVM_PREPARATION_START") leader_res: None | dict[int, bytes] - if self.leader_receipt is None or not self.leader_receipt.eq_outputs: + if self.leader_receipt is None or self.leader_receipt.eq_outputs is None: leader_res = None else: leader_res = { diff --git a/backend/node/genvm/base.py b/backend/node/genvm/base.py index c46cbbadc..3e77000cb 100644 --- a/backend/node/genvm/base.py +++ b/backend/node/genvm/base.py @@ -51,6 +51,7 @@ # subset the SDK exposes, so import it last to win over the star import. from .origin.host_fns import ResultCode from .origin import logger as genvm_logger +from .origin.leader_public_data import LeaderPublicData from .error_codes import ( extract_error_code, extract_error_code_from_timeout, @@ -656,8 +657,10 @@ def provide_result( ) ) - # Extract eq_outputs from result_nondet_results - eq_outputs = {i: data for i, data in enumerate(res.result_nondet_results)} + leader_public_data = LeaderPublicData.decode(res.result_leader_public_data) + eq_outputs = { + i: data for i, data in enumerate(leader_public_data.nondet_block_outputs) + } execution_stats = dict(ctx.stats) llm_token_metrics = _extract_llm_token_metrics(res.metrics) @@ -909,16 +912,16 @@ def _create_timeout_result( ) -def _leader_results_to_list( +def _encode_leader_public_data( leader_results: dict[int, bytes] | None, -) -> list[bytes] | None: - """Convert dict[int, bytes] keyed by call_no to ordered list[bytes].""" +) -> bytes | None: if leader_results is None: return None - if not leader_results: - return [] - max_key = max(leader_results.keys()) - return [leader_results.get(i, b"") for i in range(max_key + 1)] + outputs = [] + if leader_results: + max_key = max(leader_results.keys()) + outputs = [leader_results.get(i, b"") for i in range(max_key + 1)] + return LeaderPublicData(outputs).encode() async def run_genvm_host( @@ -1035,7 +1038,7 @@ async def run_genvm_host( leader_results = fresh_args.get( "leader_results", host_args.get("leader_results") ) - leader_nondet_results = _leader_results_to_list(leader_results) + leader_public_data = _encode_leader_public_data(leader_results) try: # Fresh manager websocket per attempt: run_genvm never owns @@ -1065,7 +1068,7 @@ async def run_genvm_host( calldata=fresh_args.get( "calldata_bytes", host_args.get("calldata_bytes", b"") ), - leader_nondet_results=leader_nondet_results, + leader_public_data=leader_public_data, unsafe_overrides=base_host.UnsafeOverrides( reroute_to=genvm_executor_selector or "" ), diff --git a/backend/node/genvm/origin/base_host.py b/backend/node/genvm/origin/base_host.py index 7701c6dda..c2ff5c505 100644 --- a/backend/node/genvm/origin/base_host.py +++ b/backend/node/genvm/origin/base_host.py @@ -506,7 +506,7 @@ class ConsumedResult: result_fingerprint: ResultFingerprint | None = None result_storage_deltas: list[tuple[bytes, bytes]] = field(default_factory=list) result_emissions: list[ResultEmission] = field(default_factory=list) - result_nondet_results: list[bytes] = field(default_factory=list) + result_leader_public_data: bytes = b"" data_fees_remaining: list[int] = field(default_factory=list) @classmethod @@ -559,7 +559,7 @@ def decode(cls, raw: typing.Any) -> "ConsumedResult": result_fingerprint=decoded.get("fingerprint"), result_storage_deltas=decoded.get("storage_deltas", []), result_emissions=decoded.get("emissions", []), - result_nondet_results=decoded.get("nondet_results", []), + result_leader_public_data=decoded.get("leader_public_data", b""), data_fees_remaining=decoded.get("data_fees_remaining", []), ) @@ -579,7 +579,7 @@ class RunHostAndProgramRes: result_fingerprint: ResultFingerprint | None result_storage_deltas: list[tuple[bytes, bytes]] result_emissions: list[ResultEmission] - result_nondet_results: list[bytes] + result_leader_public_data: bytes data_fees_remaining: list[int] metrics: dict[str, typing.Any] | None = None vm_error_description: str | None = None @@ -1177,7 +1177,7 @@ async def run_genvm( bucket_totals: list[int], code: bytes | None = None, calldata: bytes, - leader_nondet_results: list[bytes] | None = None, + leader_public_data: bytes | None = None, message_fee_allocation: collections.abc.Sequence[fees.MessageAllocationNode] = (), unsafe_overrides: UnsafeOverrides | None = None, request_extra: collections.abc.Mapping[ @@ -1228,7 +1228,7 @@ async def run_genvm( "extra_args": list(extra_args), "code": code, "calldata": calldata, - "leader_nondet_results": leader_nondet_results, + "leader_public_data": leader_public_data, "bucket_totals": bucket_totals, "gas_data": effective_gas_data, "message_fee_allocation": list(message_fee_allocation), @@ -1403,7 +1403,7 @@ async def cancel_on_shutdown(): result_fingerprint=consumed.result_fingerprint, result_storage_deltas=consumed.result_storage_deltas, result_emissions=consumed.result_emissions, - result_nondet_results=consumed.result_nondet_results, + result_leader_public_data=consumed.result_leader_public_data, data_fees_remaining=consumed.data_fees_remaining, vm_error_description=vm_error_description, execution_time=time.time() - started_at, diff --git a/backend/node/genvm/origin/leader_public_data.py b/backend/node/genvm/origin/leader_public_data.py new file mode 100644 index 000000000..85b837c07 --- /dev/null +++ b/backend/node/genvm/origin/leader_public_data.py @@ -0,0 +1,98 @@ +import collections.abc +from dataclasses import dataclass +from typing import Self + +_PADDING = b"padded" + + +@dataclass +class LeaderPublicData: + nondet_block_outputs: list[bytes] + + def encode(self) -> bytes: + payload = b"".join( + _encode_bytes(value) for value in (*self.nondet_block_outputs, _PADDING) + ) + return _encode_length(len(payload), 0xC0, 0xF7) + payload + + @classmethod + def decode(cls, encoded: collections.abc.Buffer) -> Self: + data = bytes(encoded) + if not data: + return cls([]) + + payload_start, payload_len = _decode_length(data, 0, list_=True) + payload_end = payload_start + payload_len + if payload_end != len(data): + raise ValueError("trailing leader public data") + + outputs: list[bytes] = [] + cursor = payload_start + while cursor < payload_end: + value_start, value_len = _decode_length(data, cursor, list_=False) + value_end = value_start + value_len + if value_end > payload_end: + raise ValueError("leader public data item exceeds list") + outputs.append(data[value_start:value_end]) + cursor = value_end + + if not outputs or outputs[-1] != _PADDING: + raise ValueError("leader public data padding is missing") + return cls(outputs[:-1]) + + +def encode(value: LeaderPublicData) -> bytes: + return value.encode() + + +def decode(encoded: collections.abc.Buffer) -> LeaderPublicData: + return LeaderPublicData.decode(encoded) + + +def _encode_bytes(value: bytes) -> bytes: + if len(value) == 1 and value[0] < 0x80: + return value + return _encode_length(len(value), 0x80, 0xB7) + value + + +def _encode_length(length: int, short_base: int, long_base: int) -> bytes: + if length <= 55: + return bytes([short_base + length]) + length_bytes = length.to_bytes((length.bit_length() + 7) // 8, "big") + return bytes([long_base + len(length_bytes)]) + length_bytes + + +def _decode_length( + encoded: bytes, + offset: int, + *, + list_: bool, +) -> tuple[int, int]: + if offset >= len(encoded): + raise ValueError("truncated leader public data") + + prefix = encoded[offset] + short_base = 0xC0 if list_ else 0x80 + long_base = 0xF7 if list_ else 0xB7 + if not list_ and prefix < 0x80: + return offset, 1 + if prefix < short_base or prefix > long_base + 8: + raise ValueError("invalid RLP prefix") + if prefix <= long_base: + if not list_ and prefix == 0x81 and offset + 1 < len(encoded): + if encoded[offset + 1] < 0x80: + raise ValueError("non-canonical RLP string") + return offset + 1, prefix - short_base + + length_len = prefix - long_base + length_start = offset + 1 + length_end = length_start + length_len + if length_end > len(encoded): + raise ValueError("truncated RLP length") + length_bytes = encoded[length_start:length_end] + if length_bytes[0] == 0: + raise ValueError("non-canonical RLP length") + length = int.from_bytes(length_bytes, "big") + if length <= 55: + raise ValueError("non-canonical long RLP value") + return length_end, length diff --git a/tests/unit/test_node_state_proxy_metrics.py b/tests/unit/test_node_state_proxy_metrics.py index 7dca358db..e54d9281a 100644 --- a/tests/unit/test_node_state_proxy_metrics.py +++ b/tests/unit/test_node_state_proxy_metrics.py @@ -13,6 +13,7 @@ import backend.node.genvm.origin.calldata as gvm_calldata from backend.node.genvm.origin.base_host import RunHostAndProgramRes from backend.node.genvm.origin.host_fns import ResultCode +from backend.node.genvm.origin.leader_public_data import LeaderPublicData from backend.node.types import Address, ExecutionMode, ExecutionResultStatus from backend.protocol_rpc.fees import ( GENVM_UNMETERED_DATA_FEE_BUCKET, @@ -186,7 +187,7 @@ def test_host_provide_result_preserves_fee_metadata_from_genvm_emissions(): "gasUsed": 123, }, ], - result_nondet_results=[], + result_leader_public_data=LeaderPublicData([]).encode(), data_fees_remaining=[100, 90, 80], ) diff --git a/tests/unit/test_provide_result_readonly.py b/tests/unit/test_provide_result_readonly.py index 7c9a565e6..4081e6b48 100644 --- a/tests/unit/test_provide_result_readonly.py +++ b/tests/unit/test_provide_result_readonly.py @@ -3,6 +3,7 @@ from backend.node.genvm.base import Context, Host from backend.node.genvm.origin.base_host import RunHostAndProgramRes from backend.node.genvm.origin.host_fns import ResultCode +from backend.node.genvm.origin.leader_public_data import LeaderPublicData def _result_with_storage_change(): @@ -17,7 +18,7 @@ def _result_with_storage_change(): result_fingerprint=None, result_storage_deltas=[(b"\x11" * 32 + (0).to_bytes(4, "big"), b"\xaa")], result_emissions=[], - result_nondet_results=[], + result_leader_public_data=LeaderPublicData([]).encode(), data_fees_remaining=[], ) diff --git a/tests/unit/test_usage_metrics_service.py b/tests/unit/test_usage_metrics_service.py index 2af180b07..aec2434cc 100644 --- a/tests/unit/test_usage_metrics_service.py +++ b/tests/unit/test_usage_metrics_service.py @@ -5,6 +5,7 @@ from backend.node.genvm import base as genvm_base from backend.node.genvm.origin import host_fns +from backend.node.genvm.origin.leader_public_data import LeaderPublicData from backend.services.usage_metrics_service import UsageMetricsService @@ -169,7 +170,7 @@ def test_provide_result_preserves_llm_token_metrics(): result_data={"ok": True}, result_storage_deltas=[], result_emissions=[], - result_nondet_results=[], + result_leader_public_data=LeaderPublicData([]).encode(), stdout="", stderr="", genvm_log=[], diff --git a/third_party/genvm/version b/third_party/genvm/version index 07f057afb..09d75c98f 100644 --- a/third_party/genvm/version +++ b/third_party/genvm/version @@ -1 +1 @@ -fix/vm-fatal-errors:8a4816ebea2d2c1f210c6167778e94330fb3d46a +feat/leader-data-to-genvm:201dc510c61c0fe86371cd40518f3e40fd3fb8c7