diff --git a/packages/sync/src/domain/sync-job-dispatch.service.db.test.ts b/packages/sync/src/domain/sync-job-dispatch.service.db.test.ts index 526abdfee..fd8aa51cb 100644 --- a/packages/sync/src/domain/sync-job-dispatch.service.db.test.ts +++ b/packages/sync/src/domain/sync-job-dispatch.service.db.test.ts @@ -260,7 +260,9 @@ describe("dispatchSyncJob", () => { jobFor(resource, "incrementalPull"), now, ); - expect(outcome).toEqual({ result: "done" }); + // Subject is the invalidation feed; this seed has no channel, so it also + // carries a bootstrap followup that is not what this test is about. + expect(outcome.result).toBe("done"); const feed = await storage .db() @@ -275,9 +277,20 @@ describe("dispatchSyncJob", () => { }); }); - it("settles an applied incremental pull as done with no followup", async () => { + it("settles an applied incremental pull as done when the channel is already live", async () => { const calendar = await seedCalendar(); const resource = await seedResource(calendar, "cursor-0"); + await resources.updateSubscription( + calendar.tenantId, + calendar.principalId, + resource._id, + { + subscriptionId: "channel-1", + subscriptionResourceId: "provider-resource-1", + subscriptionToken: "token-1", + subscriptionExpiresAt: new Date("2026-08-01T00:00:00.000Z"), + }, + ); const reader = new FakeReader([page([single("new-1")], "cursor-1")]); const outcome = await dispatchSyncJob( @@ -300,6 +313,34 @@ describe("dispatchSyncJob", () => { }); }); + it("bootstraps a push channel when an applied pull finds the calendar has none", async () => { + // The initialImport followup used to be the only thing that ever opened a + // channel, and the renewal sweep only renews channels that already exist, + // so a calendar imported by any other route could never become watchable. + // Production preseeded 938 calendars straight into the store during the + // Sync cutover: cursors present, syncing fine, no channel, and nothing in + // the system able to give them one (2026-08-01). + const calendar = await seedCalendar(); + const resource = await seedResource(calendar, "cursor-0"); + expect(resource.subscriptionId).toBeNull(); + const reader = new FakeReader([page([single("new-1")], "cursor-1")]); + + const outcome = await dispatchSyncJob( + deps(reader), + jobFor(resource, "incrementalPull"), + now, + ); + + if (outcome.result !== "done" || !outcome.followup) { + throw new Error("expected a followup"); + } + expect(outcome.followup.kind).toBe("subscriptionMaintain"); + expect(outcome.followup.coalescingKey).toBe( + `subscriptionMaintain:${resource._id}`, + ); + expect(outcome.followup.resourceId).toBe(resource._id); + }); + it("hands off an expired-cursor pull to a repair followup", async () => { const calendar = await seedCalendar(); const resource = await seedResource(calendar, "stale-cursor"); diff --git a/packages/sync/src/domain/sync-job-dispatch.service.ts b/packages/sync/src/domain/sync-job-dispatch.service.ts index ba9515e8e..a59628b31 100644 --- a/packages/sync/src/domain/sync-job-dispatch.service.ts +++ b/packages/sync/src/domain/sync-job-dispatch.service.ts @@ -233,6 +233,27 @@ async function runSyncJob( const pull = await pullCalendarChanges(deps, calendar, now); if (pull.status === "applied") { await appendCalendarInvalidation(deps, calendar, now()); + // Bootstrap a channel for an imported calendar that has none. The + // initialImport followup is otherwise the ONLY thing that ever opens + // one, and the renewal sweep only renews channels that already exist + // (listExpiringSubscriptions filters on subscriptionId), so a calendar + // imported by any other route could never become watchable. Production + // preseeded 938 calendars straight into the store during the Sync + // cutover, bypassing that job: they held cursors, synced correctly, and + // had no push channel with nothing in the system able to give them one + // (2026-08-01). Pulls already run for every stale calendar, so + // piggybacking here needs no new sweep and heals the whole fleet. + // + // Wart: a calendar the provider refuses to watch reports "unsupported" + // and is not recorded as such, so it re-attempts one watch per pull + // (~1/100min at current sweep cadence). Cheap, bounded, and self- + // limiting in practice since unwatchable calendars are rare. + if (pull.resource.subscriptionId === null) { + return { + result: "done", + followup: subscriptionFollowup(pull.resource, now), + }; + } return { result: "done" }; } if (pull.status === "notImported") { diff --git a/packages/sync/src/domain/sync-job-worker.service.db.test.ts b/packages/sync/src/domain/sync-job-worker.service.db.test.ts index 10d7a5ccc..4b38bfb2b 100644 --- a/packages/sync/src/domain/sync-job-worker.service.db.test.ts +++ b/packages/sync/src/domain/sync-job-worker.service.db.test.ts @@ -614,7 +614,11 @@ describe("SyncJobWorker", () => { ]), ).drain(); - expect(processed).toBe(2); + // Four, not two: neither seeded resource has a push channel, so each + // applied pull enqueues a subscriptionMaintain followup that this same + // drain then picks up. That is the property under test — drain keeps going + // until nothing is due, including work spawned partway through it. + expect(processed).toBe(4); expect( await storage .db()