fix(cassettes): order state events by created_at, not by one remembered id

The gate on the ATM-state consumer compared the incoming event id against
the id stored on a single arbitrary row (SELECT ... LIMIT 1, no ORDER BY).
That is a one-event memory, not a watermark: a re-delivered A, B, A applied
three times. Worse, created_at was parsed, written to state_at and then
never compared, so an event arriving late overwrote newer state — nothing
in the path ever looked at the clock.

Events are now applied only when strictly newer than the OLDEST state stamp
on file. Strict '>' subsumes replay dedup, since a replay carries the same
stamp. Oldest rather than newest is deliberate: every execute in this data
layer commits on its own, so a multi-row apply cannot be made atomic here,
and gating on the oldest means a crash mid-apply is re-applied on the next
event instead of being mistaken for a complete one. The ATM republishes on
a heartbeat, so it converges.

Stamps are compared as unix floats because SQLite returns integers,
Postgres returns timestamps and the incoming value is tz-aware; comparing
raw would either raise or quietly mislead. An unparseable incoming stamp
fails closed.

Also renames apply_bootstrap_state to apply_reported_state and corrects the
module comments. There has never been a once-per-machine guard, so calling
it a one-shot bootstrap consumer described something the code did not do.

Refs #43

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
This commit is contained in:
Padreug 2026-09-22 22:23:19 +02:00
commit 27449e1d11
4 changed files with 144 additions and 85 deletions

View file

@ -244,20 +244,20 @@ async def _record_rejected(payment: Payment, machine: Machine, exc: Exception) -
# =============================================================================
# Cassette bootstrap consumer (#29 v1)
# Cassette state consumer (#29)
# =============================================================================
# Subscribes to kind-30078 bitspire-cassettes-state:<atm_pubkey_hex> events
# published by each active machine's ATM on first boot (lamassu-next#56's
# bootstrap publish path). Decrypts the NIP-44 v2 content with the operator's
# privkey + ATM sender pubkey, validates as PublishCassettesPayload, and
# upserts cassette_configs via apply_bootstrap_state.
# published by each active machine's ATM. Decrypts the NIP-44 v2 content with
# the operator's privkey + ATM sender pubkey, validates as
# PublishCassettesPayload, and reconciles cassette_configs via
# apply_reported_state.
#
# v1 = one-shot per machine (ATM's meta.bootstrapPublishedAt makes the
# publish idempotent on ATM-side restart; spirekeeper's apply_bootstrap_
# state dedups on state_event_id for relay re-delivery).
#
# v2 (separate issue) = continuous reverse-channel consumer with a
# last_state_created_at watermark for reconciliation UI.
# This is continuous, not one-shot: the ATM publishes on startup, after every
# change to its bays, and on a heartbeat, and each event replaces the last.
# The comments here used to call it a "v1 bootstrap" consumer, which was
# misleading — there has never been a once-per-machine guard, so every event
# an ATM published was already being applied. Ordering is enforced in
# apply_reported_state by created_at; it is not inferred from arrival order.
#
# Implementation: polls nostrclient.router.NostrRouter.received_subscription_
# events keyed by our subscription_id. nostrclient's NostrRouter design is
@ -285,7 +285,7 @@ async def wait_for_cassette_state_events() -> None:
between retries (operator may install it later)
- inbound event fails sig-verify / decrypt / parse → log + skip
the event, continue the loop
- apply_bootstrap_state errors → log + skip
- apply_reported_state errors → log + skip
"""
logger.info(
"spirekeeper v2: cassette bootstrap consumer starting "
@ -335,7 +335,7 @@ async def _cassette_consumer_tick(current_filter_key: str | None) -> str:
from .cassette_transport import build_state_d_tags_for_machines
from .crud import (
apply_bootstrap_state,
apply_reported_state,
get_machine_by_atm_pubkey_hex,
list_all_active_machines,
)
@ -372,7 +372,7 @@ async def _cassette_consumer_tick(current_filter_key: str | None) -> str:
await _handle_cassette_state_event(
event_message,
get_machine_by_atm_pubkey_hex,
apply_bootstrap_state,
apply_reported_state,
)
except Exception as exc:
logger.warning(
@ -386,7 +386,7 @@ async def _cassette_consumer_tick(current_filter_key: str | None) -> str:
async def _handle_cassette_state_event(
event_message,
get_machine_by_atm_pubkey_hex,
apply_bootstrap_state,
apply_reported_state,
) -> None:
"""Verify signature, resolve the operator's signer, decrypt via the
signer abstraction (bunker round-trip for RemoteBunkerSigner; direct
@ -482,7 +482,7 @@ async def _handle_cassette_state_event(
created_at_unix = event_obj.get("created_at", 0)
event_created_at = _datetime.fromtimestamp(int(created_at_unix), tz=_timezone.utc)
applied = await apply_bootstrap_state(
applied = await apply_reported_state(
machine.id, event_id, event_created_at, payload
)
if applied: