diff --git a/backend/node/base.py b/backend/node/base.py index 106d26c33..3f4db62bb 100644 --- a/backend/node/base.py +++ b/backend/node/base.py @@ -173,11 +173,12 @@ def __init__( self._collect_metrics = collect_metrics # Primary contract fast path: # decode the full primary contract state once and then read by bytes slot key. - self._primary_decoded: dict[bytes, bytes] | None = None + # {state_status: {slot: decoded_value_bytes}} + self._primary_decoded: dict[str, dict[bytes, bytes]] = {} # Per-contract decoded slot cache: - # {contract_address: {base64_slot_key: decoded_value_bytes}} + # {(state_status, contract_address): {base64_slot_key: decoded_value_bytes}} # This is only used for non-primary contract lazy reads. - self._decoded_slots: dict[str, dict[str, bytes]] = {} + self._decoded_slots: dict[tuple[str, str], dict[str, bytes]] = {} self._slot_keys: dict[bytes, str] = {} # Execution-scoped metrics to quantify cross-contract read cold/warm behavior. if collect_metrics: @@ -239,16 +240,18 @@ def _slot_key(self, slot: bytes) -> str: self._slot_keys[slot] = slot_key return slot_key - def _get_contract_slot_cache(self, snap: ContractSnapshot) -> dict[str, bytes]: - return self._decoded_slots.setdefault(snap.contract_address, {}) + def _get_contract_slot_cache( + self, snap: ContractSnapshot, status: str + ) -> dict[str, bytes]: + return self._decoded_slots.setdefault((status, snap.contract_address), {}) - def _get_primary_decoded(self) -> dict[bytes, bytes]: - decoded = self._primary_decoded + def _get_primary_decoded(self, status: str) -> dict[bytes, bytes]: + decoded = self._primary_decoded.get(status) if decoded is not None: return decoded decoded = {} - state = self.snapshot.states.get(self.state_status, {}) + state = self.snapshot.states.get(status, {}) shared_cache = self._shared_decoded_value_cache for slot_key, raw in state.items(): slot = base64.b64decode(slot_key) @@ -273,14 +276,13 @@ def _get_primary_decoded(self) -> dict[bytes, bytes]: metrics["shared_decoded_cache_hits"] += 1 decoded[slot] = value - self._primary_decoded = decoded + self._primary_decoded[status] = decoded return decoded - def _read_primary_slot_value(self, slot: bytes) -> bytes: - return self._get_primary_decoded().get(slot, b"") - - def _read_cross_slot_value(self, snap: ContractSnapshot, slot: bytes) -> bytes: - slot_cache = self._get_contract_slot_cache(snap) + def _read_cross_slot_value( + self, snap: ContractSnapshot, slot: bytes, status: str + ) -> bytes: + slot_cache = self._get_contract_slot_cache(snap, status) slot_key = self._slot_key(slot) data = slot_cache.get(slot_key) if data is not None: @@ -292,7 +294,7 @@ def _read_cross_slot_value(self, snap: ContractSnapshot, slot: bytes) -> bytes: metrics = self._metrics if metrics is not None: metrics["decoded_cache_misses"] += 1 - raw = snap.states.get(self.state_status, {}).get(slot_key) + raw = snap.states.get(status, {}).get(slot_key) shared_cache = self._shared_decoded_value_cache if raw is not None and shared_cache is not None: data = shared_cache.get(raw) @@ -328,15 +330,25 @@ def get_metrics(self) -> dict[str, int]: def storage_read( self, account: Address, slot: bytes, index: int, le: int, / + ) -> bytes: + return self._storage_read(self.state_status, account, slot, index, le) + + def storage_read_finalized( + self, account: Address, slot: bytes, index: int, le: int, / + ) -> bytes: + return self._storage_read("finalized", account, slot, index, le) + + def _storage_read( + self, status: str, account: Address, slot: bytes, index: int, le: int ) -> bytes: if account == self.contract_address: metrics = self._metrics if metrics is not None: metrics["primary_contract_lookups"] += 1 - data = self._read_primary_slot_value(slot) + data = self._get_primary_decoded(status).get(slot, b"") else: snap = self._get_snapshot(account) - data = self._read_cross_slot_value(snap, slot) + data = self._read_cross_slot_value(snap, slot, status) end = index + le if end <= len(data): return data[index:end] @@ -356,7 +368,7 @@ def storage_write( assert not self.readonly slot_key = self._slot_key(slot) state_bucket = self.snapshot.states.setdefault(self.state_status, {}) - primary_decoded = self._get_primary_decoded() + primary_decoded = self._get_primary_decoded(self.state_status) existing = primary_decoded.get(slot, b"") data = bytearray(existing) mem = memoryview(got) diff --git a/backend/node/genvm/base.py b/backend/node/genvm/base.py index 8e3e372a9..d89da495f 100644 --- a/backend/node/genvm/base.py +++ b/backend/node/genvm/base.py @@ -140,6 +140,12 @@ def storage_read( self, account: Address, slot: bytes, index: int, le: int, / ) -> bytes: ... + def storage_read_finalized( + self, account: Address, slot: bytes, index: int, le: int, / + ) -> bytes: + """Read from the latest finalized state; a slot never finalized reads as zeroes.""" + return bytes(le) + @abc.abstractmethod def get_balance(self, addr: Address) -> int: ... @@ -822,10 +828,12 @@ async def close_connections(self) -> None: async def storage_read( self, type: StorageView, address: bytes, slot: bytes, offset: int, le: int, / ) -> bytes: - assert type != StorageView.LATEST_FINALIZED - return await asyncio.to_thread( - self._state_proxy.storage_read, Address(address), slot, offset, le + read = ( + self._state_proxy.storage_read_finalized + if type == StorageView.LATEST_FINALIZED + else self._state_proxy.storage_read ) + return await asyncio.to_thread(read, Address(address), slot, offset, le) async def resolve_call_contract_executor( self, diff --git a/tests/unit/test_storage_read_latest_finalized.py b/tests/unit/test_storage_read_latest_finalized.py new file mode 100644 index 000000000..323677f77 --- /dev/null +++ b/tests/unit/test_storage_read_latest_finalized.py @@ -0,0 +1,136 @@ +"""`StorageView.LATEST_FINALIZED` reads, e.g. `view(state=LATEST_FINALIZED)`. + +Served from the finalized state; slots with no finalized value read as zeroes. +""" + +import base64 +from unittest.mock import MagicMock + +import pytest + +from backend.database_handler.contract_snapshot import ContractSnapshot +from backend.node.base import _SnapshotView, _StateProxyNone +from backend.node.genvm.base import Host +from backend.node.genvm.origin.public_abi import StorageView +from backend.node.types import Address + +PRIMARY = "0x" + "aa" * 20 +OTHER = "0x" + "bb" * 20 +SLOT = b"\x07" * 32 +SLOT_KEY = base64.b64encode(SLOT).decode("ascii") + + +def _raw(value: bytes) -> str: + return base64.b64encode(value).decode("ascii") + + +def _snapshot(address: str, states: dict) -> ContractSnapshot: + snapshot = MagicMock(spec=ContractSnapshot) + snapshot.contract_address = address + snapshot.states = states + snapshot.balance = 0 + return snapshot + + +def _host(primary_states: dict, other_states: dict | None = None) -> Host: + other = _snapshot(OTHER, other_states if other_states is not None else {}) + host = Host.__new__(Host) + host._state_proxy = _SnapshotView( + _snapshot(PRIMARY, primary_states), + lambda _addr: other, + readonly=True, + shared_decoded_value_cache={}, + ) + return host + + +async def _read(host: Host, view: StorageView, addr: str, le: int, offset: int = 0): + return await host.storage_read(view, Address(addr).as_bytes, SLOT, offset, le) + + +@pytest.mark.asyncio +async def test_cross_contract_reads_finalized_state(): + host = _host( + {}, + { + "accepted": {SLOT_KEY: _raw(b"accepted")}, + "finalized": {SLOT_KEY: _raw(b"final")}, + }, + ) + assert await _read(host, StorageView.LATEST_FINALIZED, OTHER, 5) == b"final" + + +@pytest.mark.asyncio +async def test_primary_contract_reads_finalized_state(): + host = _host( + { + "accepted": {SLOT_KEY: _raw(b"accepted")}, + "finalized": {SLOT_KEY: _raw(b"final")}, + } + ) + assert await _read(host, StorageView.LATEST_FINALIZED, PRIMARY, 5) == b"final" + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "other_states", + [ + pytest.param( + {"accepted": {SLOT_KEY: _raw(b"accepted")}}, id="no-finalized-state" + ), + pytest.param( + {"accepted": {SLOT_KEY: _raw(b"accepted")}, "finalized": {}}, + id="slot-not-finalized", + ), + ], +) +@pytest.mark.parametrize("addr", [PRIMARY, OTHER]) +async def test_missing_finalized_value_reads_zeroes(other_states, addr): + host = _host(other_states, other_states) + assert await _read(host, StorageView.LATEST_FINALIZED, addr, 8) == b"\x00" * 8 + + +@pytest.mark.asyncio +@pytest.mark.parametrize("read_before_write", [False, True]) +async def test_write_does_not_leak_into_finalized_view(read_before_write): + host = Host.__new__(Host) + view = _SnapshotView( + _snapshot(PRIMARY, {"accepted": {}, "finalized": {SLOT_KEY: _raw(b"old")}}), + lambda _addr: None, + readonly=False, + shared_decoded_value_cache={}, + ) + host._state_proxy = view + if read_before_write: + assert await _read(host, StorageView.LATEST_FINALIZED, PRIMARY, 3) == b"old" + view.storage_write(SLOT, 0, b"new") + assert await _read(host, StorageView.DEFAULT, PRIMARY, 3) == b"new" + assert await _read(host, StorageView.LATEST_FINALIZED, PRIMARY, 3) == b"old" + + +@pytest.mark.asyncio +async def test_finalized_read_pads_past_value_end(): + host = _host({}, {"finalized": {SLOT_KEY: _raw(b"abc")}}) + assert await _read(host, StorageView.LATEST_FINALIZED, OTHER, 4, offset=1) == ( + b"bc\x00\x00" + ) + + +@pytest.mark.asyncio +@pytest.mark.parametrize("addr", [PRIMARY, OTHER]) +async def test_finalized_and_default_reads_do_not_share_cache(addr): + states = { + "accepted": {SLOT_KEY: _raw(b"accepted")}, + "finalized": {SLOT_KEY: _raw(b"final")}, + } + host = _host(states, states) + assert await _read(host, StorageView.DEFAULT, addr, 8) == b"accepted" + assert await _read(host, StorageView.LATEST_FINALIZED, addr, 5) == b"final" + assert await _read(host, StorageView.DEFAULT, addr, 8) == b"accepted" + + +@pytest.mark.asyncio +async def test_proxy_without_finalized_state_reads_zeroes(): + host = Host.__new__(Host) + host._state_proxy = _StateProxyNone(Address(PRIMARY)) + assert await _read(host, StorageView.LATEST_FINALIZED, PRIMARY, 4) == b"\x00" * 4