fix(genvm): serve latest-finalized storage reads from finalized state 🐛

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)

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,

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