From 66151ed3edb7ef9183d3a70ab2b9181e441b8c84 Mon Sep 17 00:00:00 2001 From: raphastacks Date: Sat, 29 Aug 2026 06:06:45 +0200 Subject: [PATCH] fix(subscriptions): bound chain evaluator tip by per-trigger decoder progress Chain print subscriptions silently dropped events. The evaluator's block-source tip tracks block INGESTION, which runs ahead of decode; each event type is decoded by an independent, differently-paced decoder (print is heavier and trails ft_transfer). When the evaluator processed a height before the print decoder committed it, the print wasn't in decoded_events yet, the match was missed, and the forward-only cursor never revisited it. Bound the evaluator tip by the MIN decoder_checkpoints height over ONLY the decoders feeding the event types its active subscriptions read (derived from referencedEventTypes: decode..v1). A stalled decoder no subscription reads (e.g. a defunct pox4) is excluded and can never gate deliveries. Null floor -> fall back to the raw tip rather than stall. Co-Authored-By: Claude Opus 4.8 (1M context) Claude-Session: https://claude.ai/code/session_01YMvYzMv1qpiUMUeGA81bXB --- .../trigger-evaluator-decoder-floor.test.ts | 86 +++++++++++++++++++ .../src/runtime/trigger-evaluator-loop.ts | 17 +++- .../src/runtime/trigger-evaluator.ts | 79 +++++++++++++++++ 3 files changed, 181 insertions(+), 1 deletion(-) create mode 100644 packages/subgraphs/src/runtime/trigger-evaluator-decoder-floor.test.ts diff --git a/packages/subgraphs/src/runtime/trigger-evaluator-decoder-floor.test.ts b/packages/subgraphs/src/runtime/trigger-evaluator-decoder-floor.test.ts new file mode 100644 index 000000000..b6cd5d665 --- /dev/null +++ b/packages/subgraphs/src/runtime/trigger-evaluator-decoder-floor.test.ts @@ -0,0 +1,86 @@ +import { describe, expect, test } from "bun:test"; +import type { Subscription } from "@secondlayer/shared/db"; +import { + decoderFloorHeight, + lowestDecoderHeight, + referencedDecoderNames, +} from "./trigger-evaluator.ts"; + +/** Minimal chain subscription with the given triggers. */ +function chainSub(triggers: Array>): Subscription { + return { kind: "chain", triggers } as unknown as Subscription; +} + +/** + * Fake source DB whose `decoder_checkpoints` query returns the canned cursor for + * each requested decoder name — enough to exercise `decoderFloorHeight` without + * a Postgres. Records nothing; just filters the `WHERE decoder_name IN (…)` set. + */ +function fakeSourceDb(checkpoints: Record) { + let requested: string[] = []; + const qb = { + selectFrom() { + return qb; + }, + select() { + return qb; + }, + where(_col: string, _op: string, names: string[]) { + requested = names; + return qb; + }, + async execute() { + return requested + .filter((n) => n in checkpoints) + .map((n) => ({ last_cursor: checkpoints[n] })); + }, + }; + // biome-ignore lint/suspicious/noExplicitAny: minimal Kysely stub for the test + return qb as any; +} + +describe("chain evaluator decoder-progress floor (per-trigger)", () => { + test("referencedDecoderNames maps triggers to only their decoders", () => { + expect( + referencedDecoderNames([chainSub([{ type: "print_event", contractId: "*" }])]), + ).toEqual(["decode.print.v1"]); + expect( + referencedDecoderNames([ + chainSub([{ type: "ft_transfer", assetIdentifier: "*" }]), + ]), + ).toEqual(["decode.ft_transfer.v1"]); + }); + + test("lowestDecoderHeight picks the min and skips absent/unparseable cursors", () => { + // The race: print (8_864_633) trails ft_transfer (8_864_861) → floor = print. + expect( + lowestDecoderHeight(["8864861:2147483647", "8864633:620"]), + ).toBe(8_864_633); + expect(lowestDecoderHeight(["8864861:0", null, ""])).toBe(8_864_861); + expect(lowestDecoderHeight([null, undefined])).toBeNull(); + expect(lowestDecoderHeight([])).toBeNull(); + }); + + test("floor for a print sub tracks print, ignoring a faster ft and a stalled pox4", async () => { + // print behind ft (the race) AND a defunct pox4 stalled far back. The floor + // must follow print (the only decoder this sub reads), NOT be dragged down + // by pox4 (unsubscribed) nor float up to the ingestion-fast ft_transfer. + const db = fakeSourceDb({ + "decode.print.v1": "8864633:620", + "decode.ft_transfer.v1": "8864861:2147483647", + "decode.pox4.v1": "8000000:2147483647", + }); + const names = referencedDecoderNames([ + chainSub([{ type: "print_event", contractId: "*fakfun-market-registry*" }]), + ]); + expect(names).toEqual(["decode.print.v1"]); + await expect(decoderFloorHeight(names, { sourceDb: db })).resolves.toBe( + 8_864_633, + ); + }); + + test("no referenced decoders → null floor (caller falls back to raw tip)", async () => { + const db = fakeSourceDb({ "decode.print.v1": "8864633:0" }); + await expect(decoderFloorHeight([], { sourceDb: db })).resolves.toBeNull(); + }); +}); diff --git a/packages/subgraphs/src/runtime/trigger-evaluator-loop.ts b/packages/subgraphs/src/runtime/trigger-evaluator-loop.ts index 51b3ff431..cb6f819d7 100644 --- a/packages/subgraphs/src/runtime/trigger-evaluator-loop.ts +++ b/packages/subgraphs/src/runtime/trigger-evaluator-loop.ts @@ -8,10 +8,12 @@ import { buildChainBlockSource } from "./block-source.ts"; import { buildSourcesMap, buildTraitContracts, + decoderFloorHeight, emitChainOutbox, emitSbtcOutbox, emitSbtcSettlementOutbox, evaluateBlock, + referencedDecoderNames, referencedEventTypes, } from "./trigger-evaluator.ts"; @@ -123,7 +125,20 @@ export async function runEvaluatorOnce( let emitted = await emitSbtcSettlementOutbox(db, chainSubs); const source = buildChainBlockSource(referencedEventTypes(chainSubs)); - const tip = await source.getTip(); + const rawTip = await source.getTip(); + if (rawTip <= 0) return emitted; + + // The block-source tip tracks block INGESTION, which runs ahead of decode. + // Each event type is decoded by an independent, differently-paced decoder, so + // processing a height before the decoder feeding a referenced event type + // (e.g. print, which trails ft_transfer) has committed it would drop the match + // against the forward-only cursor. Bound the tip by the MIN checkpoint over + // ONLY the decoders these subscriptions read — a stalled decoder nobody + // subscribes to (e.g. a defunct pox4) is excluded and can't block deliveries. + // Null floor (no referenced decoders / none checkpointed) → fall back to the + // raw tip rather than stalling. + const floor = await decoderFloorHeight(referencedDecoderNames(chainSubs)); + const tip = floor === null ? rawTip : Math.min(rawTip, floor); if (tip <= 0) return emitted; const cursor = await readCursor(db); diff --git a/packages/subgraphs/src/runtime/trigger-evaluator.ts b/packages/subgraphs/src/runtime/trigger-evaluator.ts index db105ec00..5edeacff0 100644 --- a/packages/subgraphs/src/runtime/trigger-evaluator.ts +++ b/packages/subgraphs/src/runtime/trigger-evaluator.ts @@ -149,6 +149,85 @@ export function referencedEventTypes(chainSubs: Subscription[]): string[] { return indexEventTypesForFilterTypes([...filterTypes]); } +/** + * Decoder-progress floor for the chain evaluator. + * + * Each decoded event type is produced by an INDEPENDENT decoder with its own + * `decoder_checkpoints` cursor, and they advance at different rates — `print` + * is heavier to decode (arbitrary Clarity payloads) and routinely trails + * `ft_transfer` by a beat. The evaluator's block source tip tracks block + * INGESTION, which runs ahead of decode. If the evaluator processes a height + * before the decoder feeding an event type its subscriptions read has committed + * that height, the events aren't in `decoded_events` yet, the match is missed, + * and the forward-only cursor never revisits it (the print-subscription miss + * this module was hardened against). + * + * So the evaluator bounds its tip by the MIN checkpoint over ONLY the decoders + * feeding the event types its active subscriptions actually read — NOT over all + * decoders. A decoder no subscription reads (e.g. a stalled/defunct `pox4`) + * must never gate deliveries. + * + * The decoded-event decoders are named `decode..v1` uniformly (see + * `DECODER_EVENT_TYPES` in the indexer's decode/storage). Deriving the name + * keeps the runtime free of a dependency on the indexer package. + */ +function decoderNameForEventType(indexEventType: string): string { + return `decode.${indexEventType}.v1`; +} + +/** Decoder names feeding the event types these chain subs read. */ +export function referencedDecoderNames(chainSubs: Subscription[]): string[] { + return referencedEventTypes(chainSubs).map(decoderNameForEventType); +} + +/** Parse the block height out of a `height:event_index` decoder cursor. */ +function decoderCursorHeight(cursor: string | null | undefined): number | null { + if (!cursor) return null; + const [height] = cursor.split(":"); + if (!height || !/^(0|[1-9]\d*)$/.test(height)) return null; + return Number(height); +} + +/** + * Lowest committed height across a set of decoder cursors — pure, so the min + * logic is unit-testable without a DB. Cursors that are absent/unparseable (a + * decoder that has not checkpointed yet) are skipped; returns null when none are + * usable, so the caller falls back to the unbounded tip rather than stalling. + */ +export function lowestDecoderHeight( + cursors: ReadonlyArray, +): number | null { + let min: number | null = null; + for (const c of cursors) { + const h = decoderCursorHeight(c); + if (h === null) continue; + if (min === null || h < min) min = h; + } + return min; +} + +/** + * The height at/below which every decoder in `decoderNames` has committed — + * the data-availability floor for exactly the event types the evaluator reads. + * Reads the SOURCE-plane `decoder_checkpoints` (same rationale as + * `buildTraitContracts`: the evaluator runs on the TARGET handle where those + * rows are empty). Returns null when no name has a usable cursor, so the caller + * falls back to the raw tip. + */ +export async function decoderFloorHeight( + decoderNames: string[], + opts?: { sourceDb?: Kysely }, +): Promise { + if (decoderNames.length === 0) return null; + const sourceDb = opts?.sourceDb ?? getSourceDb(); + const rows = await sourceDb + .selectFrom("decoder_checkpoints") + .select("last_cursor") + .where("decoder_name", "in", decoderNames) + .execute(); + return lowestDecoderHeight(rows.map((r) => r.last_cursor)); +} + /** Distinct traits referenced across all chain triggers. */ export function referencedTraits(chainSubs: Subscription[]): string[] { const traits = new Set();