PyBook 04: The Ledger

Sync is defined in exactly Plaid's envelope,

{ "added": [...], "modified": [...], "removed": [...],
  "next_cursor": "...", "has_more": true }

, with both of Plaid's hard rules kept: "order every array by ascending last-modified time" and "require the client to page until has_more == false before persisting the cursor." And the payoff: "a chain reorg is removed + re-added … a first-class event rather than a magic boolean."

LedgerPort is a runtime_checkable Protocol (rule #12, storage is a port), tenant-scoped throughout (rule #6, multi-tenancy is designed in, never retrofitted). MemoryLedger is the reference backend.

from auradefi.ledger.backends.memory import MemoryLedger
from auradefi.ledger.models import (
    Direction,
    Entry,
    LedgerTransaction,
    SyncEventKind,
    transaction_id,
)
from auradefi.ledger.port import LedgerPort
from auradefi.money.quantity import Quantity

ledger = MemoryLedger()
assert isinstance(ledger, LedgerPort)  # satisfies the port structurally

ASSET = "ast_a31ab4a449ad2399"  # USDC, from PyBook 03


def make_txn(tx_hash: str, block: int, raw: int = 10**18) -> LedgerTransaction:
    return LedgerTransaction(
        id=transaction_id("eip155:1", tx_hash, "acct_a"),
        chain_id="eip155:1",
        tx_hash=tx_hash,
        account_id="acct_a",
        block_number=block,
        initiated_at=1_754_000_000_000,      # ms epoch, always
        confirmed_at=1_754_000_012_000,
        entries=(Entry(asset_id=ASSET, quantity=Quantity(raw, 18), direction=Direction.IN),),
    )


# Identity is deterministic over (chain_id, tx_hash, account_id).
assert transaction_id("eip155:1", "0xabc", "acct_a") == transaction_id("eip155:1", "0xabc", "acct_a")
assert transaction_id("eip155:1", "0xabc", "acct_a") != transaction_id("eip155:1", "0xabc", "acct_b")
print("deterministic id:", transaction_id("eip155:1", "0xabc", "acct_a"))

deterministic id: txn_a6da120868d16444

Upsert

Five confirmed transactions land in tenant A's ledger. Each write gets the tenant's next monotonic last_modified_seq (starting at 1) and emits an ADDED event.

txns = [make_txn(f"0x{i:064x}", block=100 + i) for i in range(5)]

events = ledger.upsert("tenant_a", txns)
assert [e.kind for e in events] == [SyncEventKind.ADDED] * 5
assert [e.transaction.last_modified_seq for e in events] == [1, 2, 3, 4, 5]
print("upserted", len(events), "transactions, seqs 1..5")

upserted 5 transactions, seqs 1..5

Cursor sync: page until `has_more` is `False`

Events come back ordered by ascending last-modified seq: last-modified order, not transaction date, which "is what lets a two-year-old row reappear" when it changes. The cursor is an opaque token; a malformed one raises CursorError, never a silent restart from zero.

from auradefi.errors import CursorError

cursor, collected, pages = None, [], 0
while True:
    page = ledger.sync("tenant_a", cursor=cursor, limit=2)
    collected.extend(page.events)
    cursor = page.next_cursor
    pages += 1
    if not page.has_more:
        break

assert (pages, len(collected)) == (3, 5)            # 2 + 2 + 1
seqs = [e.transaction.last_modified_seq for e in collected]
assert seqs == sorted(seqs) == [1, 2, 3, 4, 5]      # ascending last-modified
print(f"drained in {pages} pages; cursor now {cursor!r}")

try:
    ledger.sync("tenant_a", cursor="not-a-cursor")
except CursorError as exc:
    print("malformed cursor:", exc)
else:
    raise AssertionError("expected CursorError")

drained in 3 pages; cursor now '00000000000000000005'
malformed cursor: cursor token must be exactly 20 ASCII digits: 'not-a-cursor'

Idempotent re-upsert

Redelivery is the normal case for any source that pages. A payload-identical re-upsert emits no event and bumps no seq: backend bookkeeping (last_modified_seq, removed) is excluded from payload equality, so redelivery can never masquerade as change.

assert ledger.upsert("tenant_a", txns) == []            # nothing to say

page = ledger.sync("tenant_a", cursor=cursor)
assert page.events == () and page.has_more is False
assert page.next_cursor == cursor                        # cursor unmoved
print("re-upsert of identical payloads: no events, cursor unmoved")

re-upsert of identical payloads: no events, cursor unmoved

`mark_removed`: removal is a first-class event

Removal is a REMOVED event with its own bumped seq. The row stays, flagged, so sync tells every client it is gone. Sync pages are state-based: one event per transaction, reflecting its latest state, so a client that wants to observe an intermediate state must drain its cursor before that state is overwritten. Exactly the "page until has_more is False" discipline the cursor contract demands.

dropped = txns[4]  # the block-104 transaction

removed_events = ledger.mark_removed("tenant_a", [dropped.id])
assert [e.kind for e in removed_events] == [SyncEventKind.REMOVED]
assert ledger.get("tenant_a", dropped.id).removed is True
assert ledger.mark_removed("tenant_a", [dropped.id]) == []   # already removed: no-op

page = ledger.sync("tenant_a", cursor=cursor)
cursor = page.next_cursor
assert [e.kind for e in page.events] == [SyncEventKind.REMOVED]
assert page.events[0].transaction.id == dropped.id
print("client observed the removal:", [e.kind.value for e in page.events])

client observed the removal: ['removed']

Resurrection

The inverse is first-class too: re-upserting the identical payload of a removed row resurrects it: stored removed=False, a bumped seq, and a fresh ADDED event (re-add semantics). Idempotence keys on the stored removed flag, so "the source says this transaction is canonical again" is never mistaken for a redundant redelivery.

resurrected = ledger.upsert("tenant_a", [dropped])   # same payload as before
assert [e.kind for e in resurrected] == [SyncEventKind.ADDED]
assert ledger.get("tenant_a", dropped.id).removed is False

page = ledger.sync("tenant_a", cursor=cursor)
cursor = page.next_cursor
assert [e.kind for e in page.events] == [SyncEventKind.ADDED]
assert page.events[0].transaction.id == dropped.id
print("…then the re-add:", [e.kind.value for e in page.events])

…then the re-add: ['added']

Reorg: `removed` + re-`added`

A fork at block 103 orphans the stored block-103 transaction and lands a different one in its place. plan_reorg is a pure diff of the stored view against the canonical view; apply_reorg applies it as mark_removed + upsert. "Zerion re-delivers reorged transactions with deleted: true; we make it a first-class event rather than a magic boolean."

from auradefi.ledger.reorg import plan_reorg

fork_block = 103
orphaned = txns[3]                                        # stored at block 103
replacement = make_txn("0x" + "f" * 64, block=103, raw=2 * 10**18)

stored_view = [ledger.get("tenant_a", txn.id) for txn in txns]
canonical_view = [txn for txn in txns if txn.block_number < fork_block] + [
    make_txn(f"0x{4:064x}", block=104),                   # block-104 survivor, unchanged
    replacement,
]

plan = plan_reorg(stored_view, canonical_view, from_block=fork_block)
assert plan.remove_ids == (orphaned.id,)                  # orphaned branch
assert [txn.id for txn in plan.add] == [replacement.id]   # the new canonical txn
# Untouched pre-fork rows and payload-identical survivors are in NEITHER bucket.

reorg_events = ledger.apply_reorg("tenant_a", plan)
assert [e.kind for e in reorg_events] == [SyncEventKind.REMOVED, SyncEventKind.ADDED]

page = ledger.sync("tenant_a", cursor=cursor)
cursor = page.next_cursor
assert [e.kind for e in page.events] == [SyncEventKind.REMOVED, SyncEventKind.ADDED]
seqs = [e.transaction.last_modified_seq for e in page.events]
assert seqs == sorted(seqs)                               # cursor stays monotonic
print("reorg observed through sync:", [e.kind.value for e in page.events])

reorg observed through sync: ['removed', 'added']

Tenant isolation (rule #6)

Every method is tenant-scoped. Another tenant's transaction is indistinguishable from a missing one. NotFoundError, not a permission error that would leak existence. And a tenant id that isn't a real string never reaches storage at all (TenantIsolationError).

from auradefi.errors import NotFoundError, TenantIsolationError

# Tenant B asks for tenant A's transaction by its exact id:
try:
    ledger.get("tenant_b", txns[0].id)
except NotFoundError as exc:
    print("tenant_b sees:", exc)          # not found: existence not leaked
else:
    raise AssertionError("tenant isolation is broken")

# Tenant B's view of the world is simply empty.
assert ledger.sync("tenant_b").events == ()

# Junk tenant ids are rejected before any read or write.
try:
    ledger.get("   ", txns[0].id)
except TenantIsolationError as exc:
    print("guarded:", exc)
else:
    raise AssertionError("expected TenantIsolationError")

print("\nledger walkthrough complete: every assertion held")

tenant_b sees: transaction not found: 'txn_3f9735dd702f3103'
guarded: tenant_id must be a non-empty, non-whitespace string

ledger walkthrough complete: every assertion held