diff --git a/packages/sync/src/app.ts b/packages/sync/src/app.ts index 99eeb19b4..6e3eefc9c 100644 --- a/packages/sync/src/app.ts +++ b/packages/sync/src/app.ts @@ -407,7 +407,15 @@ function buildSchedulers( { sweep: async (before) => { const enqueued = await reconcileStaleCalendars( - { resources, jobs }, + { + resources, + jobs, + onEnqueueError: (error, resourceId) => + logger.error( + `Sync reconcile sweep could not enqueue resource ${resourceId}; skipping it and continuing`, + error, + ), + }, before, () => new Date(), ); @@ -431,7 +439,15 @@ function buildSchedulers( { sweep: (before) => maintainExpiringSubscriptions( - { resources, jobs }, + { + resources, + jobs, + onEnqueueError: (error, resourceId) => + logger.error( + `Sync subscription sweep could not enqueue resource ${resourceId}; skipping it and continuing`, + error, + ), + }, before, () => new Date(), ), @@ -497,6 +513,7 @@ function buildSchedulers( events: repos.events, calendars: repos.calendars, occurrences: repos.eventOccurrences, + resources: repos.syncResources, markers: repos.deletionMarkers, execution: config.EXECUTION, provider: { diff --git a/packages/sync/src/domain/cloud-command.service.db.test.ts b/packages/sync/src/domain/cloud-command.service.db.test.ts index 053bf8361..5a9d924f1 100644 --- a/packages/sync/src/domain/cloud-command.service.db.test.ts +++ b/packages/sync/src/domain/cloud-command.service.db.test.ts @@ -28,6 +28,7 @@ import { DeletionMarkerRepository } from "@sync/storage/repositories/deletion-ma import { EventRepository } from "@sync/storage/repositories/event.repository"; import { EventOccurrenceRepository } from "@sync/storage/repositories/event-occurrence.repository"; import { ProviderCalendarRepository } from "@sync/storage/repositories/provider-calendar.repository"; +import { SyncResourceRepository } from "@sync/storage/repositories/sync-resource.repository"; import { type SyncMongoService } from "@sync/storage/sync-mongo.service"; import { beforeEach, describe, expect, it, spyOn } from "bun:test"; @@ -120,6 +121,7 @@ describe("submitCloudCommand provider dispatch", () => { let commands: CommandRepository; let events: EventRepository; let occurrences: EventOccurrenceRepository; + let resources: SyncResourceRepository; let calendars: ProviderCalendarRepository; let markers: DeletionMarkerRepository; @@ -181,6 +183,7 @@ describe("submitCloudCommand provider dispatch", () => { commands = new CommandRepository(mongo.db); events = new EventRepository(mongo.db); occurrences = new EventOccurrenceRepository(mongo.db, mongo.client); + resources = new SyncResourceRepository(mongo.db); calendars = new ProviderCalendarRepository(mongo.db); markers = new DeletionMarkerRepository(mongo.db); }); @@ -197,6 +200,7 @@ describe("submitCloudCommand provider dispatch", () => { events, calendars, occurrences, + resources, markers, execution: "active", provider: provider(writer), @@ -224,6 +228,7 @@ describe("submitCloudCommand provider dispatch", () => { events, calendars, occurrences, + resources, markers, execution: "passive", provider: provider(writer), @@ -247,6 +252,7 @@ describe("submitCloudCommand provider dispatch", () => { events, calendars, occurrences, + resources, markers, execution: "active", }, @@ -268,6 +274,7 @@ describe("submitCloudCommand provider dispatch", () => { events, calendars, occurrences, + resources, markers, execution: "active", provider: provider(writer), @@ -306,6 +313,7 @@ describe("submitCloudCommand provider dispatch", () => { events, calendars, occurrences, + resources, markers, execution: "active", }, @@ -335,6 +343,7 @@ describe("submitCloudCommand provider dispatch", () => { events, calendars, occurrences, + resources, markers, execution: "active", }, @@ -415,6 +424,7 @@ describe("submitCloudCommand provider dispatch", () => { events, calendars, occurrences, + resources, markers, execution: "passive" as const, }); @@ -464,6 +474,7 @@ describe("submitCloudCommand provider dispatch", () => { events, calendars, occurrences, + resources, markers, execution: "active", provider: provider(writer), @@ -522,6 +533,7 @@ describe("submitCloudCommand provider dispatch", () => { events, calendars, occurrences, + resources, markers, execution: "active", provider: provider(writer), @@ -564,6 +576,7 @@ describe("submitCloudCommand provider dispatch", () => { events, calendars, occurrences, + resources, markers, execution: "active", provider: provider(writer), @@ -598,6 +611,7 @@ describe("submitCloudCommand provider dispatch", () => { events, calendars, occurrences, + resources, markers, execution: "active", provider: provider(writer), @@ -687,6 +701,7 @@ describe("submitCloudCommand provider dispatch", () => { events, calendars, occurrences, + resources, markers, execution: "active", provider: provider(writer), @@ -744,6 +759,7 @@ describe("submitCloudCommand provider dispatch", () => { events, calendars, occurrences, + resources, markers, execution: "active", provider: provider(writer), diff --git a/packages/sync/src/domain/cloud-command.service.ts b/packages/sync/src/domain/cloud-command.service.ts index a5edcc701..6af476f19 100644 --- a/packages/sync/src/domain/cloud-command.service.ts +++ b/packages/sync/src/domain/cloud-command.service.ts @@ -44,6 +44,7 @@ import { type DeletionMarkerRepository } from "@sync/storage/repositories/deleti import { type EventRepository } from "@sync/storage/repositories/event.repository"; import { type EventOccurrenceRepository } from "@sync/storage/repositories/event-occurrence.repository"; import { type ProviderCalendarRepository } from "@sync/storage/repositories/provider-calendar.repository"; +import { type SyncResourceRepository } from "@sync/storage/repositories/sync-resource.repository"; // A provider-targeted write arrived while provider work is unavailable // (execution is passive, or no provider is configured). Nothing re-dispatches a @@ -67,6 +68,9 @@ export interface CloudCommandDeps { // The derived occurrence projection, rebuilt for an event's horizon whenever // a cloud command changes it so range queries stay current. occurrences: EventOccurrenceRepository; + // Which generation reads serve per calendar, so a provider-linked create + // projects where reads will look for it. + resources: SyncResourceRepository; // The deletion-marker store, for the tombstone a provider delete leaves. markers: DeletionMarkerRepository; execution: SyncExecutionMode; @@ -146,6 +150,7 @@ export async function submitCloudCommand( commands: deps.commands, events: deps.events, occurrences: deps.occurrences, + resources: deps.resources, writer: deps.provider.writer, custody: deps.provider.custody, }, @@ -339,6 +344,7 @@ async function applyCloudMutation( commands: deps.commands, events: deps.events, occurrences: deps.occurrences, + resources: deps.resources, writer: deps.provider.writer, custody: deps.provider.custody, }, @@ -386,6 +392,7 @@ async function dispatchProviderDelete( commands: deps.commands, events: deps.events, occurrences: deps.occurrences, + resources: deps.resources, writer: deps.provider.writer, custody: deps.provider.custody, markers: deps.markers, @@ -421,6 +428,7 @@ async function dispatchProviderSeriesUpdate( commands: deps.commands, events: deps.events, occurrences: deps.occurrences, + resources: deps.resources, writer: deps.provider.writer, custody: deps.provider.custody, }, @@ -458,6 +466,7 @@ async function dispatchProviderOccurrenceDelete( commands: deps.commands, events: deps.events, occurrences: deps.occurrences, + resources: deps.resources, writer: deps.provider.writer, custody: deps.provider.custody, }, @@ -488,6 +497,7 @@ async function dispatchProviderOccurrenceUpdate( commands: deps.commands, events: deps.events, occurrences: deps.occurrences, + resources: deps.resources, writer: deps.provider.writer, custody: deps.provider.custody, }, @@ -518,6 +528,7 @@ async function dispatchProviderSeriesFollowingDelete( commands: deps.commands, events: deps.events, occurrences: deps.occurrences, + resources: deps.resources, writer: deps.provider.writer, custody: deps.provider.custody, markers: deps.markers, @@ -549,6 +560,7 @@ async function dispatchProviderSeriesFollowingUpdate( commands: deps.commands, events: deps.events, occurrences: deps.occurrences, + resources: deps.resources, writer: deps.provider.writer, custody: deps.provider.custody, }, diff --git a/packages/sync/src/domain/failed-job-requeue.service.db.test.ts b/packages/sync/src/domain/failed-job-requeue.service.db.test.ts index 8e93143f6..25fee726d 100644 --- a/packages/sync/src/domain/failed-job-requeue.service.db.test.ts +++ b/packages/sync/src/domain/failed-job-requeue.service.db.test.ts @@ -102,4 +102,30 @@ describe("requeueFailedJobs", () => { exhaustedJobs: [], }); }); + + it("requeues a job written before requeuedCount existed", async () => { + // Mongo's {$lt: n} does not match a missing field, so the self-heal sweep + // could not see the very jobs most likely to be wedged: the ones old + // enough to predate its own bookkeeping field. Three such jobs sat failed + // in prod while this sweep reported nothing to do (2026-07-31). + const id = await seedFailed({ + runAfter: new Date("2026-07-20T10:00:00.000Z"), + }); + await storage + .db() + .collection("jobs") + .updateOne({ _id: id as never }, { $unset: { requeuedCount: "" } }); + + const result = await requeueFailedJobs(deps(), cooldownBefore, now, 3); + + expect(result.requeued).toBe(1); + expect(result.exhausted).toBe(0); + const raw = await storage + .db() + .collection("jobs") + .findOne({ _id: id as never }); + expect(raw?.state).toBe("pending"); + // Absence counted as zero, so the requeue is its first, not its last. + expect(raw?.requeuedCount).toBe(1); + }); }); diff --git a/packages/sync/src/domain/provider-command.service.db.test.ts b/packages/sync/src/domain/provider-command.service.db.test.ts index 0480f73be..495f18080 100644 --- a/packages/sync/src/domain/provider-command.service.db.test.ts +++ b/packages/sync/src/domain/provider-command.service.db.test.ts @@ -48,6 +48,7 @@ import { DeletionMarkerRepository } from "@sync/storage/repositories/deletion-ma import { EventRepository } from "@sync/storage/repositories/event.repository"; import { EventOccurrenceRepository } from "@sync/storage/repositories/event-occurrence.repository"; import { ProviderCalendarRepository } from "@sync/storage/repositories/provider-calendar.repository"; +import { SyncResourceRepository } from "@sync/storage/repositories/sync-resource.repository"; import { type SyncMongoService } from "@sync/storage/sync-mongo.service"; const storage = setupSyncStorage(import.meta.url); @@ -143,6 +144,7 @@ describe("executeProviderCreate", () => { let commands: CommandRepository; let events: EventRepository; let occurrences: EventOccurrenceRepository; + let resources: SyncResourceRepository; let calendars: ProviderCalendarRepository; const createInput = ( @@ -210,6 +212,7 @@ describe("executeProviderCreate", () => { commands = new CommandRepository(mongo.db); events = new EventRepository(mongo.db); occurrences = new EventOccurrenceRepository(mongo.db, mongo.client); + resources = new SyncResourceRepository(mongo.db); calendars = new ProviderCalendarRepository(mongo.db); }); @@ -220,7 +223,14 @@ describe("executeProviderCreate", () => { const writer = new FakeWriter(); const result = await executeProviderCreate( - { commands, events, occurrences, writer, custody: tokenSource() }, + { + commands, + events, + occurrences, + resources, + writer, + custody: tokenSource(), + }, command, calendar, now, @@ -261,7 +271,14 @@ describe("executeProviderCreate", () => { const writer = new FakeWriter(); await executeProviderCreate( - { commands, events, occurrences, writer, custody: tokenSource() }, + { + commands, + events, + occurrences, + resources, + writer, + custody: tokenSource(), + }, command, calendar, now, @@ -277,6 +294,7 @@ describe("executeProviderCreate", () => { commands, events, occurrences, + resources, writer, custody: tokenSource(), }; @@ -294,13 +312,63 @@ describe("executeProviderCreate", () => { expect(owned).toHaveLength(1); }); + it("projects a create at the calendar's active generation, not zero", async () => { + // 2026-08-01: a repaired calendar reads at generation 1, but creates + // hardcoded their occurrences to generation 0, so a new event saved + // successfully to Google and was then invisible in Compass. That was + // meant to self-heal on the next incremental pull; when the sweeps froze, + // the window stayed open for a day. + const { tenantId, principalId, calendar, command } = await seed(); + const resource = await resources.ensure({ + tenantId, + principalId, + connectionId: calendar.connectionId, + resourceKind: "events", + calendarId: calendar._id, + }); + await resources.startNewGeneration(tenantId, principalId, resource._id); + await resources.activateGeneration(tenantId, principalId, resource._id, 1); + + await executeProviderCreate( + { + commands, + events, + occurrences, + resources, + writer: new FakeWriter(), + custody: tokenSource(), + }, + command, + calendar, + now, + ); + + // Visible to a read at the generation the calendar actually serves. + const atActive = await events.listByCalendar({ + tenantId, + principalId, + calendarId: calendar._id, + generation: 1, + limit: 10, + }); + expect(atActive).toHaveLength(1); + expect(atActive[0]?._id).toBe(command.eventId); + }); + it("leaves the command pending on a transient write failure", async () => { const { tenantId, principalId, calendar, command } = await seed(); const writer = new FakeWriter(); writer.error = new ProviderWriteError("transient", "network blip"); const result = await executeProviderCreate( - { commands, events, occurrences, writer, custody: tokenSource() }, + { + commands, + events, + occurrences, + resources, + writer, + custody: tokenSource(), + }, command, calendar, now, @@ -318,7 +386,14 @@ describe("executeProviderCreate", () => { writer.error = new ProviderWriteError("readOnlyCalendar", "read only"); const result = await executeProviderCreate( - { commands, events, occurrences, writer, custody: tokenSource() }, + { + commands, + events, + occurrences, + resources, + writer, + custody: tokenSource(), + }, command, calendar, now, @@ -431,6 +506,7 @@ describe("executeProviderUpdate", () => { let commands: CommandRepository; let events: EventRepository; let occurrences: EventOccurrenceRepository; + let resources: SyncResourceRepository; let calendars: ProviderCalendarRepository; const now = () => new Date("2026-07-10T00:00:00.000Z"); @@ -531,6 +607,7 @@ describe("executeProviderUpdate", () => { commands = new CommandRepository(mongo.db); events = new EventRepository(mongo.db); occurrences = new EventOccurrenceRepository(mongo.db, mongo.client); + resources = new SyncResourceRepository(mongo.db); calendars = new ProviderCalendarRepository(mongo.db); }); @@ -541,7 +618,14 @@ describe("executeProviderUpdate", () => { writer.fetched = providerEvent("Old", "etag-1"); const result = await executeProviderUpdate( - { commands, events, occurrences, writer, custody: tokenSource() }, + { + commands, + events, + occurrences, + resources, + writer, + custody: tokenSource(), + }, command, event, calendar, @@ -573,7 +657,14 @@ describe("executeProviderUpdate", () => { writer.fetched = providerEvent("New", "etag-2"); const result = await executeProviderUpdate( - { commands, events, occurrences, writer, custody: tokenSource() }, + { + commands, + events, + occurrences, + resources, + writer, + custody: tokenSource(), + }, command, event, calendar, @@ -619,7 +710,14 @@ describe("executeProviderUpdate", () => { }; const result = await executeProviderUpdate( - { commands, events, occurrences, writer, custody: tokenSource() }, + { + commands, + events, + occurrences, + resources, + writer, + custody: tokenSource(), + }, command, event, calendar, @@ -639,7 +737,14 @@ describe("executeProviderUpdate", () => { writer.patchError = new ProviderWriteError("versionConflict", "stale"); const result = await executeProviderUpdate( - { commands, events, occurrences, writer, custody: tokenSource() }, + { + commands, + events, + occurrences, + resources, + writer, + custody: tokenSource(), + }, command, event, calendar, @@ -658,7 +763,14 @@ describe("executeProviderUpdate", () => { writer.fetched = null; const result = await executeProviderUpdate( - { commands, events, occurrences, writer, custody: tokenSource() }, + { + commands, + events, + occurrences, + resources, + writer, + custody: tokenSource(), + }, command, event, calendar, @@ -679,7 +791,14 @@ describe("executeProviderUpdate", () => { writer.patchError = new ProviderWriteError("transient", "blip"); const result = await executeProviderUpdate( - { commands, events, occurrences, writer, custody: tokenSource() }, + { + commands, + events, + occurrences, + resources, + writer, + custody: tokenSource(), + }, command, event, calendar, diff --git a/packages/sync/src/domain/provider-command.service.ts b/packages/sync/src/domain/provider-command.service.ts index fcbe356f6..a1581c0f0 100644 --- a/packages/sync/src/domain/provider-command.service.ts +++ b/packages/sync/src/domain/provider-command.service.ts @@ -47,6 +47,7 @@ import { type CommandRepository } from "@sync/storage/repositories/command.repos import { type DeletionMarkerRepository } from "@sync/storage/repositories/deletion-marker.repository"; import { type EventRepository } from "@sync/storage/repositories/event.repository"; import { type EventOccurrenceRepository } from "@sync/storage/repositories/event-occurrence.repository"; +import { type SyncResourceRepository } from "@sync/storage/repositories/sync-resource.repository"; // The slice of credential custody the executor needs — a valid access token for // a connection, plus discard of a provider-invalidated grant. Narrow so tests @@ -62,6 +63,9 @@ export interface ProviderMutationDeps { // The derived occurrence projection, rebuilt (or cleared, on delete) so a // provider-linked event appears in range queries. occurrences: EventOccurrenceRepository; + // Reads serve a calendar's active generation, so a create has to ask which + // generation that is rather than assume the calendar has never been repaired. + resources: SyncResourceRepository; writer: ProviderEventWriter; custody: AccessTokenSource; } @@ -135,7 +139,21 @@ export async function executeProviderCreate( // Commit the provider identity to the canonical event and project its // occurrences, then confirm. Both run before confirmation, so a crash leaves // the command pending and a retry re-runs them idempotently. - const record = buildLinkedEventRecord(command, calendar, result, now()); + // Ask which generation reads will serve this calendar before projecting, so + // a create onto a repaired calendar is visible immediately rather than + // waiting for a pull to reproject it. + const generations = await deps.resources.activeGenerationByCalendar( + command.tenantId, + command.principalId, + [input.calendarId], + ); + const record = buildLinkedEventRecord( + command, + calendar, + result, + now(), + generations.get(input.calendarId) ?? 0, + ); await deps.events.put(record); await reprojectOccurrences(deps.occurrences, record, now); @@ -181,6 +199,7 @@ function buildLinkedEventRecord( calendar: ProviderCalendarRecord, result: ProviderWriteResult, now: Date, + generation: number, ): EventRecord { if (command.input.kind !== "create") { throw new Error("buildLinkedEventRecord requires a create command"); @@ -207,17 +226,14 @@ function buildLinkedEventRecord( ? { kind: "seriesMaster", rules: input.recurrence.rules } : { kind: "single" }, lifecycleState: "active", - // generation 0 is correct in steady state (a calendar's active generation is - // 0 until a repair bumps it). The one gap it leaves is self-healing: if a - // repair has already advanced this calendar's active generation, a just- - // created event's occurrences land at generation 0 and reads (which serve - // the active generation) miss it — until the next incremental pull, which - // re-reads this event from the provider (it IS at the provider, linked here) - // and reprojects it at the active generation. Repairs are rare and pulls are - // frequent, so the window is small and closes on its own; resolving the - // active generation here would thread a resources dependency through the - // whole command path for a case that corrects itself. - generation: 0, + // The calendar's active generation, resolved by the caller — NOT a + // hardcoded 0. Reads serve the active generation, so on a calendar a + // repair has already advanced, generation-0 occurrences are invisible. + // That gap used to be left to "the next incremental pull will reproject + // it", which holds only while pulls are running: when the sweeps froze on + // 2026-07-31 the window stayed open for a day and users watched their new + // events save successfully to Google and then vanish from Compass. + generation, createdAt: now, updatedAt: now, confirmedAt: now, diff --git a/packages/sync/src/domain/reconcile.service.db.test.ts b/packages/sync/src/domain/reconcile.service.db.test.ts index e7cf2719a..e5efce8fb 100644 --- a/packages/sync/src/domain/reconcile.service.db.test.ts +++ b/packages/sync/src/domain/reconcile.service.db.test.ts @@ -183,4 +183,56 @@ describe("reconcileStaleCalendars", () => { expect(await jobByKey(`incrementalPull:${healthy._id}`)).not.toBeNull(); expect(await jobByKey(`incrementalPull:${deadCredential._id}`)).toBeNull(); }); + + it("skips a resource whose existing job cannot be read and sweeps the rest", async () => { + // 2026-07-31: three job docs written before `requeuedCount` existed made + // enqueue's coalescing read throw, and the sweep's loop had no per-item + // guard — so one unreadable doc abandoned the whole batch on every cycle + // and froze calendar sync fleet-wide for 23h. The finders sort + // deterministically, so the poisoned resource re-won the front of the + // ordering every time; nothing behind it ever ran again. + const poisoned = await seedResource(new Date("2026-07-01T00:00:00.000Z")); + const healthy = await seedResource(new Date("2026-07-02T00:00:00.000Z")); + // A job doc that cannot be parsed back out (kind is not a JobKind), sitting + // on the coalescing key the sweep is about to reuse. + await storage + .db() + .collection(SYNC_COLLECTIONS.jobs) + .insertOne({ + _id: objectId(), + coalescingKey: `incrementalPull:${poisoned._id}`, + kind: "notAJobKind", + } as never); + const failures: string[] = []; + + const enqueued = await reconcileStaleCalendars( + { resources, jobs, onEnqueueError: (_e, id) => failures.push(id) }, + staleBefore, + now, + ); + + // The healthy resource behind the poisoned one still got its pull, and the + // count reflects what was actually enqueued rather than what was found. + expect(enqueued).toBe(1); + expect(await jobByKey(`incrementalPull:${healthy._id}`)).not.toBeNull(); + expect(failures).toEqual([poisoned._id]); + }); + + it("enqueues onto a job written before requeuedCount existed", async () => { + // The specific doc shape that caused the freeze: valid work, just older + // than the field. It must read back as zero requeues, not as an error. + const stale = await seedResource(new Date("2026-07-01T00:00:00.000Z")); + await reconcileStaleCalendars(deps(), staleBefore, now); + await storage + .db() + .collection(SYNC_COLLECTIONS.jobs) + .updateOne( + { coalescingKey: `incrementalPull:${stale._id}` }, + { $unset: { requeuedCount: "" } }, + ); + + const enqueued = await reconcileStaleCalendars(deps(), staleBefore, now); + + expect(enqueued).toBe(1); + }); }); diff --git a/packages/sync/src/domain/reconcile.service.ts b/packages/sync/src/domain/reconcile.service.ts index b3ca430be..b13b609e8 100644 --- a/packages/sync/src/domain/reconcile.service.ts +++ b/packages/sync/src/domain/reconcile.service.ts @@ -5,6 +5,8 @@ import { type SyncResourceRepository } from "@sync/storage/repositories/sync-res export interface ReconcileDeps { resources: SyncResourceRepository; jobs: JobRepository; + // Reported per resource that fails to enqueue; the sweep continues. + onEnqueueError?: (error: unknown, resourceId: string) => void; } // The reconcile sweep — the missed-webhook fallback. Find events resources that diff --git a/packages/sync/src/domain/resource-sweep-enqueue.ts b/packages/sync/src/domain/resource-sweep-enqueue.ts index 7b7604f79..64bb866b2 100644 --- a/packages/sync/src/domain/resource-sweep-enqueue.ts +++ b/packages/sync/src/domain/resource-sweep-enqueue.ts @@ -7,11 +7,23 @@ import { type JobRepository } from "@sync/storage/repositories/job.repository"; // The shared shape behind reconcile.service.ts and subscription-sweep.service.ts: // find due resources with `finder`, enqueue one `kind` job per resource, and -// return how many. The only difference between the two sweeps is which finder -// they call and which job kind they enqueue — everything else, including the -// coalescing key template, is identical. +// return how many were enqueued. The only difference between the two sweeps is +// which finder they call and which job kind they enqueue — everything else, +// including the coalescing key template, is identical. +// +// Each resource is enqueued independently: one that throws is reported and +// skipped, never allowed to abandon the rest of the batch. The sweeps are the +// only liveness path for resources without a push channel, and the finders sort +// deterministically, so a single doomed resource at the front of the ordering +// would otherwise starve every resource behind it on every cycle, forever +// (2026-07-31: one unparseable job doc froze calendar sync fleet-wide for 23h). export async function enqueueForResources( - deps: { jobs: JobRepository }, + deps: { + jobs: JobRepository; + // Called once per resource that could not be enqueued. The sweep keeps + // going; the caller decides how loud to be. + onEnqueueError?: (error: unknown, resourceId: string) => void; + }, finder: (before: Date, limit: number) => Promise, kind: JobKind, before: Date, @@ -19,6 +31,7 @@ export async function enqueueForResources( limit = 100, ): Promise { const due = await finder(before, limit); + let enqueued = 0; for (const resource of due) { const enqueue: JobEnqueue = { tenantId: resource.tenantId, @@ -31,7 +44,12 @@ export async function enqueueForResources( runAfter: now(), coalescingKey: `${kind}:${resource._id}`, }; - await deps.jobs.enqueue(enqueue); + try { + await deps.jobs.enqueue(enqueue); + enqueued += 1; + } catch (error) { + deps.onEnqueueError?.(error, resource._id); + } } - return due.length; + return enqueued; } diff --git a/packages/sync/src/domain/subscription-sweep.service.ts b/packages/sync/src/domain/subscription-sweep.service.ts index 5adf2f1fc..37f9b05d1 100644 --- a/packages/sync/src/domain/subscription-sweep.service.ts +++ b/packages/sync/src/domain/subscription-sweep.service.ts @@ -5,6 +5,8 @@ import { type SyncResourceRepository } from "@sync/storage/repositories/sync-res export interface SubscriptionSweepDeps { resources: SyncResourceRepository; jobs: JobRepository; + // Reported per resource that fails to enqueue; the sweep continues. + onEnqueueError?: (error: unknown, resourceId: string) => void; } // The subscription-maintenance sweep. Find events resources whose push channel diff --git a/packages/sync/src/server/command.routes.ts b/packages/sync/src/server/command.routes.ts index a1d5491f5..8e6286428 100644 --- a/packages/sync/src/server/command.routes.ts +++ b/packages/sync/src/server/command.routes.ts @@ -101,6 +101,7 @@ export function registerCommandRoutes( events, calendars: repos.calendars, occurrences: repos.eventOccurrences, + resources: repos.syncResources, markers: repos.deletionMarkers, execution: deps.execution, provider, diff --git a/packages/sync/src/storage/contracts/job.contracts.ts b/packages/sync/src/storage/contracts/job.contracts.ts index b41bd7e36..84daef7c7 100644 --- a/packages/sync/src/storage/contracts/job.contracts.ts +++ b/packages/sync/src/storage/contracts/job.contracts.ts @@ -54,7 +54,13 @@ export const JobRecordSchema = z.strictObject({ // state:"failed". Distinct from `attempt` (which a requeue resets to give // the job a fresh retry ladder) so a resource that keeps failing cannot be // requeued forever — the sweep stops once this hits its cap. - requeuedCount: z.number().int().min(0), + // + // Defaulted, not required: this field was introduced after jobs already + // existed in production, and a job doc predating it is still perfectly + // valid work. Parsing one must not throw — enqueue coalesces onto whatever + // doc already holds a key and re-parses it, so a single unparseable job + // took down every sweep fleet-wide for 23h (2026-07-31). + requeuedCount: z.number().int().min(0).default(0), createdAt: z.date(), updatedAt: z.date(), }); diff --git a/packages/sync/src/storage/repositories/job.repository.ts b/packages/sync/src/storage/repositories/job.repository.ts index b061d547c..535a9efb5 100644 --- a/packages/sync/src/storage/repositories/job.repository.ts +++ b/packages/sync/src/storage/repositories/job.repository.ts @@ -22,6 +22,9 @@ export type ExhaustedFailedJob = { updatedAt: Date; }; +// The exact complement of listFailedForRequeue's eligibility filter: {$gte} +// does not match a missing requeuedCount, so a legacy job counts as eligible +// there and never as exhausted here. Keep the two in step. function exhaustedFailedFilter(maxRequeues: number) { return { state: "failed" as const, @@ -251,7 +254,15 @@ export class JobRepository { .find({ state: "failed", failureClass: { $ne: "permanent" }, - requeuedCount: { $lt: maxRequeues }, + // A job predating the requeuedCount field has been requeued zero + // times, not too many. Mongo's {$lt: n} does not match a missing + // field, so the absence has to be spelled out — otherwise the very + // jobs most likely to be wedged (the oldest ones) are the only ones + // the self-heal sweep can never see. + $or: [ + { requeuedCount: { $lt: maxRequeues } }, + { requeuedCount: { $exists: false } }, + ], runAfter: { $lte: before }, }) .sort({ runAfter: 1 })