fix(subscriptions): bound chain evaluator tip by per-trigger decoder …

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<Record<string, unknown>>): 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<string, string | null>) {
+	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();
+	});
+});

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);

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.<event_type>.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<string | null | undefined>,
+): 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<Database> },
+): Promise<number | null> {
+	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<string>();