diff --git a/packages/contrail/src/core/db/records.ts b/packages/contrail/src/core/db/records.ts index 2b34a39..5f8decb 100644 --- a/packages/contrail/src/core/db/records.ts +++ b/packages/contrail/src/core/db/records.ts @@ -566,10 +566,20 @@ export async function assertServingSourceCompatibility( ): Promise { if (!orderedSource) return; const existing = await getServingSourcePosition(db); + if (!existing) { + const legacyCursor = await getLastCursor(db); + if (legacyCursor !== null) { + await saveOrderedSourcePositionStatement( + db, + orderedSource, + legacyCursor, + ).run(); + } + return; + } if ( - existing && - (existing.position.source !== orderedSource.source || - existing.position.epoch !== orderedSource.epoch) + existing.position.source !== orderedSource.source || + existing.position.epoch !== orderedSource.epoch ) { throw new Error( `configured ordered source ${orderedSource.source}/${orderedSource.epoch} ` + diff --git a/packages/contrail/tests/serving-source-position.test.ts b/packages/contrail/tests/serving-source-position.test.ts index 9266732..f6a00da 100644 --- a/packages/contrail/tests/serving-source-position.test.ts +++ b/packages/contrail/tests/serving-source-position.test.ts @@ -42,6 +42,22 @@ describe("serving source positions", () => { }); }); + it("adopts an existing legacy cursor when an ordered source is configured", async () => { + const db = createSqliteDatabase(":memory:"); + await initSchema(db, config); + await saveCursor(db, 777); + + await new Contrail(config).init(db); + + expect(await getServingSourcePosition(db)).toMatchObject({ + position: { + source: "jetstream", + epoch: "primary-2026", + cursor: "777", + }, + }); + }); + it("rejects a configured continuity epoch that differs from durable state", async () => { const db = createSqliteDatabase(":memory:"); await initSchema(db, config);