[READ-ONLY] Mirror of https://github.com/flo-bit/contrail. atproto backend in a bottle
flo-bit.dev/contrail/
Merge pull request #56 from tompscanlan/fix/save-cursor-before-identity-refresh
fix(contrail-appview): persist ingest cursor before identity-refresh tail
authored by
Tom Scanlan
and committed by
GitHub
(Jun 15, 2026, 3:33 PM EDT)
88d6918a
79d4da05
+26
-10
+7
.changeset/persist-cursor-before-identity-refresh.md
+7
.changeset/persist-cursor-before-identity-refresh.md
···
1
+
---
2
+
"@atmo-dev/contrail-appview": patch
3
+
---
4
+
5
+
Persist the jetstream ingest cursor before the identity-refresh tail in `runIngestCycle`.
6
+
7
+
`saveCursor` previously ran after `refreshStaleIdentities`, whose per-DID network calls can run long. If the ingest isolate was aborted (e.g. a scheduled-invocation deadline) before the save, the cursor never advanced and the next cycle re-drained the same jetstream window indefinitely. Records are durably applied before this point, so the cursor is now saved first; identity refresh is idempotent and staleness-driven, so deferring it past the save is safe.
+19
-10
packages/contrail-appview/src/core/jetstream.ts
+19
-10
packages/contrail-appview/src/core/jetstream.ts
···
385
385
log.log(`[ingest] applied ${identityUpdates.size} identity event(s)`);
386
386
}
387
387
388
-
// Refresh stale/missing identities for DIDs in this batch
389
-
const uniqueDids = [...new Set(events.map((e) => e.did))];
390
-
if (uniqueDids.length > 0) {
391
-
try {
392
-
await refreshStaleIdentities(db, uniqueDids, config);
393
-
} catch (err) {
394
-
log.warn(`Identity refresh failed: ${err}`);
395
-
}
396
-
}
397
-
388
+
// Persist the cursor BEFORE the best-effort enrichment tail below. Records are
389
+
// already durably applied (applyEvents) and handle changes recorded, so the
390
+
// cursor's forward progress is real and must be committed now. The steps that
391
+
// follow — refreshStaleIdentities especially — make per-DID network calls and
392
+
// can run long; if the cron isolate is aborted (e.g. a scheduled-invocation
393
+
// deadline) while they run, an un-saved cursor makes the next cycle re-drain the
394
+
// identical window forever. Identity refresh is idempotent and staleness-driven,
395
+
// so deferring it past the save costs nothing.
398
396
if (lastCursor !== null) {
399
397
await saveCursor(db, lastCursor);
400
398
log.log(
···
404
402
);
405
403
} else {
406
404
log.log(`[ingest] no cursor returned from subscription; not saving`);
405
+
}
406
+
407
+
// Refresh stale/missing identities for DIDs in this batch (best-effort; runs
408
+
// after the cursor save so its network latency can't strand forward progress).
409
+
const uniqueDids = [...new Set(events.map((e) => e.did))];
410
+
if (uniqueDids.length > 0) {
411
+
try {
412
+
await refreshStaleIdentities(db, uniqueDids, config);
413
+
} catch (err) {
414
+
log.warn(`Identity refresh failed: ${err}`);
415
+
}
407
416
}
408
417
409
418
// Newly-discovered DIDs: ask Constellation for back-edges so they