From e3554eba540d4631dc6d21011087e03891ba3e54 Mon Sep 17 00:00:00 2001 From: Daniel Salazar Date: Sun, 1 Nov 2026 22:04:22 -0700 Subject: [PATCH] fix(events): pre-release fixes for jobs, schedule seams and delayed sends - The delayed-send write is a field on the notifications driver, not a method, so /drivers/call can't reach it. - A delayed send checks for its own earlier write by uid, so it takes no gap lock that every other notification insert waits on. - The notification send_key index builds in place without locking writes. - Reaps re-check expiry and suspension on the primary, so a row resumed since the replica read is kept. - Revocation settles skip schedule rows they can't settle. - An empty handler set read from a replica is re-read on the primary. - One ended notice per settle, not one per page. - Sweeps that run once per fleet take more batches per run. - The global usage rate check keeps its 14-minute minimum window. - shutdown() waits for prepare hooks a drain already started. --- .../migrations/mysql/mysql_mig_54.sql | 2 +- .../NotificationDriver.sendAt.test.ts | 4 +- .../notification/NotificationDriver.ts | 13 +- src/backend/server.test.ts | 27 +++ src/backend/server.ts | 15 +- src/backend/services/events/EventsService.ts | 196 +++++++++++------- .../events/kvHandleRevoke.integration.test.ts | 2 +- .../events/scheduleSeam.integration.test.ts | 65 ++++++ .../MeteringService.rateCheck.test.ts | 8 +- .../services/metering/MeteringService.ts | 4 + .../notification/NotificationService.test.ts | 22 +- .../notification/NotificationService.ts | 7 +- ...rableSubscriptionStore.integration.test.ts | 25 +++ .../stores/events/DurableSubscriptionStore.ts | 50 ++++- .../EventHandlerStore.integration.test.ts | 32 +++ .../stores/events/EventHandlerStore.ts | 8 +- .../stores/notification/NotificationStore.js | 19 +- .../NotificationStore.sendKey.test.ts | 44 ++++ 18 files changed, 424 insertions(+), 119 deletions(-) diff --git a/src/backend/clients/database/migrations/mysql/mysql_mig_54.sql b/src/backend/clients/database/migrations/mysql/mysql_mig_54.sql index 92e127be6..f0a182c86 100644 --- a/src/backend/clients/database/migrations/mysql/mysql_mig_54.sql +++ b/src/backend/clients/database/migrations/mysql/mysql_mig_54.sql @@ -33,7 +33,7 @@ BEGIN AND INDEX_NAME = 'idx_notification_send_key' ) THEN ALTER TABLE `notification` ADD UNIQUE KEY `idx_notification_send_key` - (`send_key`); + (`send_key`), ALGORITHM=INPLACE, LOCK=NONE; END IF; END// DELIMITER ; diff --git a/src/backend/drivers/notification/NotificationDriver.sendAt.test.ts b/src/backend/drivers/notification/NotificationDriver.sendAt.test.ts index 797b3f26c..ec13ae36f 100644 --- a/src/backend/drivers/notification/NotificationDriver.sendAt.test.ts +++ b/src/backend/drivers/notification/NotificationDriver.sendAt.test.ts @@ -145,7 +145,7 @@ const write = ( signal?: AbortSignal; } = {}, ) => - driver().writeDeferredAppMessage( + driver().deferredAppMessages.write( { holderUserId: holder.user!.id, appUid: over.appUid ?? app.uid, @@ -312,7 +312,7 @@ describe('notify({ sendAt }) with services.schedules', () => { }); }); -describe('writeDeferredAppMessage', () => { +describe('deferredAppMessages.write', () => { it('writes one row keyed by the send, in the caller`s transaction', async () => { const token = uuidv4(); const key = uuidv4(); diff --git a/src/backend/drivers/notification/NotificationDriver.ts b/src/backend/drivers/notification/NotificationDriver.ts index 791e07f04..78f24557b 100644 --- a/src/backend/drivers/notification/NotificationDriver.ts +++ b/src/backend/drivers/notification/NotificationDriver.ts @@ -539,13 +539,20 @@ export class NotificationDriver extends PuterDriver { }; } + /** + * For extensions. A field, not a method, so `/drivers/call` can't reach it. + */ + readonly deferredAppMessages = { + write: (input: DeferredWriteInput, tx: DeferredWriteTx) => + this.#writeDeferredAppMessage(input, tx), + }; + /** * Check the sender and recipient as they stand now, spend the recipient's * limits, and write the mailbox row through `tx`, fenced on its guard. - * Called by extensions. Refusals are the same `HttpError`s an immediate - * send throws. + * Refusals are the same `HttpError`s an immediate send throws. */ - async writeDeferredAppMessage( + async #writeDeferredAppMessage( input: DeferredWriteInput, tx: DeferredWriteTx, ): Promise { diff --git a/src/backend/server.test.ts b/src/backend/server.test.ts index 6a411a6d1..47d310b8c 100644 --- a/src/backend/server.test.ts +++ b/src/backend/server.test.ts @@ -1543,6 +1543,33 @@ describe('PuterServer lifecycle hooks', () => { }); }); +describe('PuterServer shutdown during prepare', () => { + it('waits for prepare hooks the drain already started before tearing down', async () => { + const port = await allocateEphemeralPort(); + const server = await setupTestServer({ port } as unknown as IConfig, { + listen: true, + }); + const order: string[] = []; + (server.services as unknown as Record).slowPrepare = { + onServerPrepareShutdown: async () => { + order.push('prepare:start'); + await new Promise((resolve) => setTimeout(resolve, 200)); + order.push('prepare:done'); + }, + onServerShutdown: () => { + order.push('shutdown'); + }, + }; + + await server.prepareShutdown({ drainMs: 0 }); + // The listener's close callback starts the prepare hooks. + await new Promise((resolve) => setTimeout(resolve, 50)); + await server.shutdown(); + + expect(order).toEqual(['prepare:start', 'prepare:done', 'shutdown']); + }); +}); + /** * An extension may take over `services.jobs` (a fleet-wide scheduler). Core * services register their jobs on it at start, so it must keep the built-in's diff --git a/src/backend/server.ts b/src/backend/server.ts index 398578b26..951f3166f 100644 --- a/src/backend/server.ts +++ b/src/backend/server.ts @@ -1765,16 +1765,19 @@ export class PuterServer { } } - #prepareShutdownHooksRan = false; + #prepareShutdownHooks: Promise | null = null; /** * Run every layer's `onServerPrepareShutdown` exactly once, whichever of - * `prepareShutdown()` / `shutdown()` gets there first. + * `prepareShutdown()` / `shutdown()` gets there first; the other waits for + * that run to finish. */ - async #runPrepareShutdownHooks() { - if (this.#prepareShutdownHooksRan) return; - this.#prepareShutdownHooksRan = true; - await this.#runHook('bottomUp', 'onServerPrepareShutdown'); + #runPrepareShutdownHooks(): Promise { + this.#prepareShutdownHooks ??= this.#runHook( + 'bottomUp', + 'onServerPrepareShutdown', + ); + return this.#prepareShutdownHooks; } async prepareShutdown(info: ServerDrainInfo = {}) { diff --git a/src/backend/services/events/EventsService.ts b/src/backend/services/events/EventsService.ts index d9af652e2..36e7674c2 100644 --- a/src/backend/services/events/EventsService.ts +++ b/src/backend/services/events/EventsService.ts @@ -706,14 +706,45 @@ const EXPIRY_SWEEP_EVERY_MS = 60 * 60 * 1000; const EXPIRY_SWEEP_LEASE_MS = 15 * 60 * 1000; /** Rows one delete takes. Small enough not to hold a lock anyone waits on. */ const EXPIRY_BATCH_SIZE = 500; -/** Batches one sweep takes, so a large backlog drains over several passes. */ -const EXPIRY_MAX_BATCHES = 50; +/** + * Batches one sweep takes at most. It runs once for the fleet, so this is the + * fleet's hourly drain; the job's signal stops it sooner. + */ +const EXPIRY_MAX_BATCHES = 1000; /** * Subjects one "subscriptions ended" notification names before it stops * listing. */ const ENDED_SUBJECTS_LISTED = 20; +/** Ended rows by kind, holder and app: a count and the subjects listed. */ +type EndedGroups = Map< + string, + { first: DurableSubscription; subjects: string[]; count: number } +>; + +/** + * Add ended rows to `groups`. A schedule is news of its own: its holder knows + * it as a schedule, not as the subscription row behind it. + */ +const collectEnded = ( + groups: EndedGroups, + rows: readonly DurableSubscription[], +): void => { + for (const row of rows) { + const kind = isScheduleRow(row) ? 'schedule' : 'subscription'; + const key = `${kind}|${row.holderUserId}|${row.appUid ?? ''}`; + let group = groups.get(key); + if (!group) { + group = { first: row, subjects: [], count: 0 }; + groups.set(key, group); + } + group.count++; + if (group.subjects.length < ENDED_SUBJECTS_LISTED) + group.subjects.push(row.subject); + } +}; + /** Re-reads an unsubscribe takes to catch a row racing a reanchor. */ const UNSUBSCRIBE_READS = 3; @@ -809,7 +840,7 @@ const unknownSubscription = (): HttpError => const scheduleSubscription = (): HttpError => new HttpError( 404, - 'This subscription belongs to a schedule; remove it through /events/schedules/remove', + 'This subscription belongs to a schedule; remove it with puter.events.removeSchedule()', { legacyCode: 'events_schedule_not_found' }, ); @@ -4610,39 +4641,49 @@ export class EventsService extends PuterService { // Withdrawing an app's access wholesale is how it is uninstalled. const scheduleEndReason = revocation.permission === null ? 'uninstalled' : undefined; - return this.#eachHolderPage( - (afterId) => - this.stores.durableSubscription.listActiveForHolderPage( - revocation.holderUserId, - revocation.appUid, - afterId, - undefined, - { includeSuspended: true }, - ), - async (held) => { - const settling = await this.#leftSettling( - held, - revocation.permission, - ); - if (settling.length === 0) return 0; + // A schedule stands only on the app's access and on consent, so a + // specific grant can't settle one. + const excludeSchedules = + revocation.permission !== null && + revocation.permission !== EVENTS_BACKGROUND_PERMISSION; + const ended: EndedGroups = new Map(); + try { + return await this.#eachHolderPage( + (afterId) => + this.stores.durableSubscription.listActiveForHolderPage( + revocation.holderUserId, + revocation.appUid, + afterId, + undefined, + { includeSuspended: true, excludeSchedules }, + ), + async (held) => { + const settling = await this.#leftSettling( + held, + revocation.permission, + ); + if (settling.length === 0) return 0; - // The `permission_revoked` arm of the shared policy purges the - // backlog at once: it holds the paths of a resource its holder - // has just lost the right to see, and keeping it for a resume - // that by design never comes turns a revocation into a delayed - // disclosure. - const suspended = await this.#suspend( - settling, - 'permission_revoked', - { - override: RESUMABLE_REASONS, - ...(scheduleEndReason ? { scheduleEndReason } : {}), - }, - ); - await this.#notifyEnded(suspended, 'permission_revoked'); - return suspended.length; - }, - ); + // The `permission_revoked` arm of the shared policy purges + // the backlog at once: it holds the paths of a resource its + // holder has just lost the right to see, and keeping it for + // a resume that by design never comes turns a revocation + // into a delayed disclosure. + const suspended = await this.#suspend( + settling, + 'permission_revoked', + { + override: RESUMABLE_REASONS, + ...(scheduleEndReason ? { scheduleEndReason } : {}), + }, + ); + collectEnded(ended, suspended); + return suspended.length; + }, + ); + } finally { + await this.#reportEnded(ended, 'permission_revoked'); + } } /** @@ -4676,30 +4717,36 @@ export class EventsService extends PuterService { } const handles = new Set(retired.handles); - const durable = await this.#eachHolderPage( - (afterId) => - this.stores.durableSubscription.listActiveForHolderPage( - retired.userId, - null, - afterId, - undefined, - { includeSuspended: true }, - ), - async (held) => { - const settling = held.filter((row) => { - const handle = kvHandleFromSubject(row.subject); - return handle !== null && handles.has(handle); - }); - if (settling.length === 0) return 0; - const suspended = await this.#suspend( - settling, - 'permission_revoked', - { override: RESUMABLE_REASONS }, - ); - await this.#notifyEnded(suspended, 'permission_revoked'); - return suspended.length; - }, - ); + const ended: EndedGroups = new Map(); + let durable = 0; + try { + durable = await this.#eachHolderPage( + (afterId) => + this.stores.durableSubscription.listActiveForHolderPage( + retired.userId, + null, + afterId, + undefined, + { includeSuspended: true, excludeSchedules: true }, + ), + async (held) => { + const settling = held.filter((row) => { + const handle = kvHandleFromSubject(row.subject); + return handle !== null && handles.has(handle); + }); + if (settling.length === 0) return 0; + const suspended = await this.#suspend( + settling, + 'permission_revoked', + { override: RESUMABLE_REASONS }, + ); + collectEnded(ended, suspended); + return suspended.length; + }, + ); + } finally { + await this.#reportEnded(ended, 'permission_revoked'); + } return sessions + durable; } @@ -5210,30 +5257,29 @@ export class EventsService extends PuterService { rows: readonly DurableSubscription[], reason: SubscriptionEndReason, ): Promise { - // A schedule is news of its own: its holder knows it as a schedule, - // not as the subscription row behind it. - const groups = new Map(); - for (const row of rows) { - const kind = isScheduleRow(row) ? 'schedule' : 'subscription'; - const key = `${kind}|${row.holderUserId}|${row.appUid ?? ''}`; - groups.set(key, [...(groups.get(key) ?? []), row]); - } - for (const group of groups.values()) { - const [first] = group; + const groups: EndedGroups = new Map(); + collectEnded(groups, rows); + await this.#reportEnded(groups, reason); + } + + /** Send what {@link collectEnded} gathered, one notification per group. */ + async #reportEnded( + groups: EndedGroups, + reason: SubscriptionEndReason, + ): Promise { + for (const { first, subjects, count } of groups.values()) { const noun = isScheduleRow(first) ? 'schedule' : 'subscription'; try { await this.services.notification.notify( [first.holderUserId], { title: - group.length === 1 + count === 1 ? `A ${noun} ended` - : `${group.length} ${noun}s ended`, + : `${count} ${noun}s ended`, subject: first.subject, - subjects: group - .slice(0, ENDED_SUBJECTS_LISTED) - .map((row) => row.subject), - count: group.length, + subjects, + count, reason, }, { type: 'app.events.ended', appUid: first.appUid }, diff --git a/src/backend/services/events/kvHandleRevoke.integration.test.ts b/src/backend/services/events/kvHandleRevoke.integration.test.ts index 52bab3b38..0450ed3d1 100644 --- a/src/backend/services/events/kvHandleRevoke.integration.test.ts +++ b/src/backend/services/events/kvHandleRevoke.integration.test.ts @@ -402,7 +402,7 @@ describe('revoking a handle', () => { null, 0, undefined, - { includeSuspended: true }, + { includeSuspended: true, excludeSchedules: true }, ]); } finally { byHolder.mockRestore(); diff --git a/src/backend/services/events/scheduleSeam.integration.test.ts b/src/backend/services/events/scheduleSeam.integration.test.ts index 0013a9de8..13ef39413 100644 --- a/src/backend/services/events/scheduleSeam.integration.test.ts +++ b/src/backend/services/events/scheduleSeam.integration.test.ts @@ -47,6 +47,7 @@ import { } from '../../stores/events/types.js'; import { setupTestServer } from '../../testUtil.js'; import type { IConfig } from '../../types.js'; +import { EVENTS_BACKGROUND_PERMISSION } from './authorization.js'; const BOOT_TIMEOUT_MS = 120_000; const APP_UID = 'app-seam'; @@ -226,6 +227,70 @@ describe('the schedules lifecycle hook', () => { ]); }); + it('reads no schedule rows to settle a specific grant, and still settles them when consent goes', async () => { + const calls = recording(); + const schedule = await scheduleRow(); + const plain = await plainRow(); + const page = vi.spyOn(durable(), 'listActiveForHolderPage'); + const readSubIds = async () => + (await Promise.all(page.mock.results.map((r) => r.value))).flatMap( + (p) => p.rows.map((row: DurableSubscription) => row.subId), + ); + + await server.services.events.settleRevokedGrant({ + holderUserId: userId, + appUid: APP_UID, + permission: `fs:${anchorUid}:list`, + }); + const read = await readSubIds(); + expect(read).toContain(plain.subId); + expect(read).not.toContain(schedule.subId); + expect(calls).toEqual([]); + + page.mockClear(); + await server.services.events.settleRevokedGrant({ + holderUserId: userId, + appUid: APP_UID, + permission: EVENTS_BACKGROUND_PERMISSION, + }); + expect(await readSubIds()).toContain(schedule.subId); + expect(calls).toEqual([ + { + subIds: [schedule.subId], + change: { type: 'suspended', reason: 'permission_revoked' }, + }, + ]); + }); + + it('tells the holder once per kind when an uninstall settles several pages', async () => { + recording(); + const plains = [await plainRow(), await plainRow(), await plainRow()]; + const schedules = [await scheduleRow(), await scheduleRow()]; + const store = durable(); + const listPage = store.listActiveForHolderPage.bind(store); + vi.spyOn(store, 'listActiveForHolderPage').mockImplementation( + (holder, app, afterId, _batchSize, opts) => + listPage(holder, app, afterId, 1, opts), + ); + const notify = vi.spyOn(server.services.notification, 'notify'); + + await expect( + server.services.events.settleRevokedGrant({ + holderUserId: userId, + appUid: APP_UID, + permission: null, + }), + ).resolves.toBe(5); + const ended = notify.mock.calls + .filter(([, , meta]) => meta?.type === 'app.events.ended') + .map(([, body]) => body as { count: number; subjects: string[] }); + expect(ended).toHaveLength(2); + expect(ended.map((body) => body.count).sort()).toEqual([2, 3]); + expect(ended.flatMap((body) => body.subjects).sort()).toEqual( + [...plains, ...schedules].map((row) => row.subject).sort(), + ); + }); + it('ends schedules app_deleted when their app is deleted', async () => { const calls = recording(); const schedule = await scheduleRow(); diff --git a/src/backend/services/metering/MeteringService.rateCheck.test.ts b/src/backend/services/metering/MeteringService.rateCheck.test.ts index b9328daf8..c8a1276dc 100644 --- a/src/backend/services/metering/MeteringService.rateCheck.test.ts +++ b/src/backend/services/metering/MeteringService.rateCheck.test.ts @@ -125,16 +125,16 @@ describe('MeteringService.checkRateOfChange', () => { }); }); - it('measures against a baseline of any age, since the job runs it once per cadence', async () => { + it('skips the shard read when the baseline is only 5 minutes old', async () => { const { svc, kv, alarmCreate } = makeService( - { maxGlobalUsagePerMinute: 1e9 }, + { maxGlobalUsagePerMinute: 100 }, { total: 0, timestamp: Date.now() - 5 * 60_000 }, ); await svc.checkRateOfChange(); - expect(kv.get).toHaveBeenCalledTimes(2); - expect(kv.set).toHaveBeenCalledTimes(1); + expect(kv.get).toHaveBeenCalledTimes(1); + expect(kv.set).not.toHaveBeenCalled(); expect(alarmCreate).not.toHaveBeenCalled(); }); }); diff --git a/src/backend/services/metering/MeteringService.ts b/src/backend/services/metering/MeteringService.ts index c86c6fe37..31e2ca0fb 100644 --- a/src/backend/services/metering/MeteringService.ts +++ b/src/backend/services/metering/MeteringService.ts @@ -2948,6 +2948,10 @@ export class MeteringService extends PuterService { timestamp: number; } | null; + // Several schedulers can share the baseline (one per region without a + // fleet scheduler, or old nodes mid-rollout); keep windows wide. + if (lastChange && now - lastChange.timestamp <= 14 * 60 * 1000) return; + const globalUsage = await this.getGlobalUsage(); const currTotal = globalUsage.total; diff --git a/src/backend/services/notification/NotificationService.test.ts b/src/backend/services/notification/NotificationService.test.ts index 5c5851c32..e92972279 100644 --- a/src/backend/services/notification/NotificationService.test.ts +++ b/src/backend/services/notification/NotificationService.test.ts @@ -912,17 +912,17 @@ describe('NotificationService.sweepExpired', () => { }); it('stops at the pass cap and leaves the rest for the next sweep', async () => { - await notifications.sweepExpired(); - const user = await makeUser(); - // 50 passes * 500/batch = 25,000 — a backlog past that so the cap, - // not a short batch, is what ends the first call. - await seedExpired(user.id, 25_050); - - expect(await notifications.sweepExpired()).toBe(25_000); - expect( - await server.stores.notification.listByUserId(user.id), - ).toHaveLength(50); - expect(await notifications.sweepExpired()).toBe(50); + // A backlog that always fills the batch: the cap, not a short batch, + // is what ends the call. Seeding 500k rows would take minutes. + const deletes = vi + .spyOn(server.stores.notification, 'deleteCreatedBefore') + .mockResolvedValue(500); + try { + expect(await notifications.sweepExpired()).toBe(500_000); + expect(deletes).toHaveBeenCalledTimes(1000); + } finally { + deletes.mockRestore(); + } }); it('stops between batches once its signal aborts', async () => { diff --git a/src/backend/services/notification/NotificationService.ts b/src/backend/services/notification/NotificationService.ts index 453d3b11f..32c55b8ec 100644 --- a/src/backend/services/notification/NotificationService.ts +++ b/src/backend/services/notification/NotificationService.ts @@ -53,8 +53,11 @@ const RETENTION_SWEEP_EVERY_MS = 60 * 60 * 1000; const RETENTION_SWEEP_LEASE_MS = 15 * 60 * 1000; /** Rows one delete takes. Small enough not to hold a lock anyone waits on. */ const RETENTION_BATCH_SIZE = 500; -/** Batches one sweep takes, so a large backlog drains over several passes. */ -const RETENTION_MAX_BATCHES = 50; +/** + * Batches one sweep takes at most. It runs once for the fleet, so this is the + * fleet's hourly drain; the job's signal stops it sooner. + */ +const RETENTION_MAX_BATCHES = 1000; /** Unread notifications one reconnect replays. */ const UNREAD_REPLAY_LIMIT = 200; diff --git a/src/backend/stores/events/DurableSubscriptionStore.integration.test.ts b/src/backend/stores/events/DurableSubscriptionStore.integration.test.ts index 112e5f2fc..adbf560de 100644 --- a/src/backend/stores/events/DurableSubscriptionStore.integration.test.ts +++ b/src/backend/stores/events/DurableSubscriptionStore.integration.test.ts @@ -772,6 +772,31 @@ describe('the expiry sweep', () => { }); }); +describe('a sweep reading a lagging replica', () => { + it('keeps a row the primary resumed since, and does not report it', async () => { + const { row } = await durable().create(input()); + await durable().suspend([row], 'permission_revoked'); + const [stale] = await env.server.clients.db.pread( + 'SELECT * FROM `event_subscriptions` WHERE `sub_id` = ?', + [row.subId], + ); + await durable().resume([row]); + // The replica still has it suspended. + vi.spyOn(env.server.clients.db, 'read').mockResolvedValueOnce([stale]); + + const reaped = await durable().sweepSuspended( + Math.floor(Date.now() / 1000) + 3600, + 500, + ); + + expect(reaped.rows).toEqual([]); + await expect(durable().getBySubId(row.subId)).resolves.toMatchObject({ + subId: row.subId, + suspendedAt: null, + }); + }); +}); + describe('listHolderIdsForApp', () => { it('names every distinct holder bound to the app`s handlers', async () => { const appUid = `app-${uuidv4()}`; diff --git a/src/backend/stores/events/DurableSubscriptionStore.ts b/src/backend/stores/events/DurableSubscriptionStore.ts index 4af50c705..ac3e4d00c 100644 --- a/src/backend/stores/events/DurableSubscriptionStore.ts +++ b/src/backend/stores/events/DurableSubscriptionStore.ts @@ -626,7 +626,11 @@ export class DurableSubscriptionStore extends PuterStore { * stops delivering against it without waiting for a rebuild. */ async sweepExpired(batchSize: number): Promise { - return this.#reap(await this.#listExpired(nowSeconds(), batchSize)); + const cutoff = nowSeconds(); + return this.#reap(await this.#listExpired(cutoff, batchSize), { + where: '`expires_at` IS NOT NULL AND `expires_at` <= ?', + values: [cutoff], + }); } /** @@ -638,7 +642,10 @@ export class DurableSubscriptionStore extends PuterStore { cutoff: number, batchSize: number, ): Promise { - return this.#reap(await this.#listSuspendedBefore(cutoff, batchSize)); + return this.#reap(await this.#listSuspendedBefore(cutoff, batchSize), { + where: '`suspended_at` IS NOT NULL AND `suspended_at` <= ?', + values: [cutoff], + }); } // -- Reads ------------------------------------------------------- @@ -768,13 +775,15 @@ export class DurableSubscriptionStore extends PuterStore { * lifts. A revocation needs them: those reasons hold a backlog, and a row * skipped here comes back in service on the next resume and hands it over. * Rows suspended for good are left out. + * + * `excludeSchedules` is for a settle no schedule row can match. */ async listActiveForHolderPage( holderUserId: number, appUid: string | null, afterId: number, batchSize: number = HOLDER_SETTLE_BATCH, - opts: { includeSuspended?: boolean } = {}, + opts: { includeSuspended?: boolean; excludeSchedules?: boolean } = {}, ): Promise { const where = [ '`holder_user_id` = ?', @@ -788,6 +797,10 @@ export class DurableSubscriptionStore extends PuterStore { ...(opts.includeSuspended ? TERMINAL_SUSPENDED_REASONS : []), nowSeconds(), ]; + if (opts.excludeSchedules) { + where.push(NOT_SCHEDULE_CLAUSE); + params.push(SCHEDULE_TOKEN_PATTERN); + } if (appUid !== null) { where.push('`app_uid` = ?'); params.push(appUid); @@ -1096,15 +1109,36 @@ export class DurableSubscriptionStore extends PuterStore { * Delete a batch and stop this region delivering against any of it. Other * regions stop once the caller publishes the returned bumps; the deleted * rows come back so the caller can end the schedules among them. + * + * `stillDue` re-checks candidates read from a replica on the primary, so a + * row resumed or extended since is kept and not reported. */ - async #reap(rows: readonly DurableSubscription[]): Promise { - if (rows.length === 0) return { rows: [], bumps: [] }; + async #reap( + candidates: readonly DurableSubscription[], + stillDue?: { where: string; values: unknown[] }, + ): Promise { + if (candidates.length === 0) return { rows: [], bumps: [] }; + const ids = candidates.map((row) => row.subId); + const inList = `(${ids.map(() => '?').join(', ')})`; await this.clients.db.write( - `DELETE FROM \`${TABLE}\` WHERE \`sub_id\` IN ` + - `(${rows.map(() => '?').join(', ')})`, - rows.map((row) => row.subId), + `DELETE FROM \`${TABLE}\` WHERE \`sub_id\` IN ${inList}` + + (stillDue ? ` AND ${stillDue.where}` : ''), + [...ids, ...(stillDue?.values ?? [])], ); + let rows = candidates; + if (stillDue) { + const kept = new Set( + ( + await this.clients.db.pread( + `SELECT \`sub_id\` FROM \`${TABLE}\` WHERE \`sub_id\` IN ${inList}`, + ids, + ) + ).map((row) => String(row.sub_id)), + ); + rows = candidates.filter((row) => !kept.has(row.subId)); + if (rows.length === 0) return { rows: [], bumps: [] }; + } const cached = rows.filter((row) => !isScheduleRow(row)); await this.stores.eventSubscription.dropDurables(cached); diff --git a/src/backend/stores/events/EventHandlerStore.integration.test.ts b/src/backend/stores/events/EventHandlerStore.integration.test.ts index 5afaabd6c..f81e457fe 100644 --- a/src/backend/stores/events/EventHandlerStore.integration.test.ts +++ b/src/backend/stores/events/EventHandlerStore.integration.test.ts @@ -506,6 +506,38 @@ describe('the set a delivery addresses', () => { expect(again[0].sourceHash).toBe(hashContent(SOURCE)); }); + it('checks an empty replica read on the primary', async () => { + await handlers().publish({ + appUid: cachedApp, + name: 'a', + source: SOURCE, + }); + // The replica hasn't seen the publish yet. + const read = vi + .spyOn(env.server.clients.db, 'read') + .mockResolvedValueOnce([]); + const pread = vi.spyOn(env.server.clients.db, 'pread'); + try { + const set = await handlers().cachedSetForApp(cachedApp, { + missRead: 'replica', + }); + expect(set.map((h) => h.name)).toEqual(['a']); + expect(pread).toHaveBeenCalledTimes(1); + + // The primary's answer is cached for the next run. (SQLite's + // `pread` goes through `read`, so count from here.) + const reads = read.mock.calls.length; + await handlers().cachedSetForApp(cachedApp, { + missRead: 'replica', + }); + expect(read).toHaveBeenCalledTimes(reads); + expect(pread).toHaveBeenCalledTimes(1); + } finally { + read.mockRestore(); + pread.mockRestore(); + } + }); + it('follows every publish and removal', async () => { await handlers().publish({ appUid: cachedApp, diff --git a/src/backend/stores/events/EventHandlerStore.ts b/src/backend/stores/events/EventHandlerStore.ts index f1b0c8c3c..2e9ebca5a 100644 --- a/src/backend/stores/events/EventHandlerStore.ts +++ b/src/backend/stores/events/EventHandlerStore.ts @@ -366,8 +366,9 @@ export class EventHandlerStore extends PuterStore { * every write here. An empty set is not cached: it fails the delivery, and * must not outlast the publish that fills it. Schedule runs read a miss * from the replica (`missRead: 'replica'`), since they come by the clock - * rather than after a write, and cache it under a key of their own. Called - * by extensions. + * rather than after a write, and cache it under a key of their own; an + * empty replica read is checked on the primary, since a lagging replica + * right after a publish would fail the run. Called by extensions. */ async cachedSetForApp( appUid: string, @@ -387,7 +388,8 @@ export class EventHandlerStore extends PuterStore { // Read the table instead. } - const set = await this.setForApp(appUid, { replica }); + let set = await this.setForApp(appUid, { replica }); + if (replica && set.length === 0) set = await this.setForApp(appUid); if (set.length > 0) { try { await this.clients.redis.set( diff --git a/src/backend/stores/notification/NotificationStore.js b/src/backend/stores/notification/NotificationStore.js index da8b2bf3e..d8cb86d2a 100644 --- a/src/backend/stores/notification/NotificationStore.js +++ b/src/backend/stores/notification/NotificationStore.js @@ -428,6 +428,20 @@ export class NotificationStore extends PuterStore { ], }); } + // Already written: the new row's uid is the key, or a replaced row + // carries it. Point lookups on `uid`, since a miss on the mostly-NULL + // `send_key` index gap-locks where every plain insert goes. + const written = send.replaceUid + ? { + sql: + 'NOT EXISTS (SELECT 1 FROM `notification` WHERE `uid` = ?) ' + + 'AND NOT EXISTS (SELECT 1 FROM `notification` WHERE `uid` = ? AND `send_key` = ?)', + values: [send.sendKey, send.replaceUid, send.sendKey], + } + : { + sql: 'NOT EXISTS (SELECT 1 FROM `notification` WHERE `uid` = ?)', + values: [send.sendKey], + }; // Postgres can't type a bare parameter in a SELECT list. const param = (type) => db.case({ postgres: `CAST(? AS ${type})`, otherwise: '?' }); @@ -438,8 +452,7 @@ export class NotificationStore extends PuterStore { `SELECT ${param('TEXT')}, ${param('INTEGER')}, ${param('JSONB')}, ` + `${param('TEXT')}, ${param('TEXT')}, ${param('TEXT')}, ` + `${param('TEXT')}, ${db.timestampFromUnixSql()}, ${param('TEXT')} ` + - `FROM \`${guard.table}\` WHERE ${guard.where} AND NOT EXISTS ` + - '(SELECT 1 FROM `notification` WHERE `send_key` = ?)', + `FROM \`${guard.table}\` WHERE ${guard.where} AND ${written.sql}`, values: [ send.sendKey, send.userId, @@ -451,7 +464,7 @@ export class NotificationStore extends PuterStore { seconds, send.sendKey, ...guard.values, - send.sendKey, + ...written.values, ], }); return entries; diff --git a/src/backend/stores/notification/NotificationStore.sendKey.test.ts b/src/backend/stores/notification/NotificationStore.sendKey.test.ts index cc8bffcf7..7efbe4464 100644 --- a/src/backend/stores/notification/NotificationStore.sendKey.test.ts +++ b/src/backend/stores/notification/NotificationStore.sendKey.test.ts @@ -182,3 +182,47 @@ describe('clearing send_key on a rewrite', () => { for (const uid of keys) expect(await sendKeyOf(uid)).toBeNull(); }); }); + +describe('writing a delayed send twice', () => { + const entriesFor = (sendKey: string, replaceUid: string | null = null) => + node({}).sendEntries( + { + sendKey, + userId, + value: { title: 'later' }, + type: 'app-message', + audience: 'account', + appUid: null, + replaceUid, + at: Date.now(), + }, + { table: 'user', where: '`id` = ?', values: [userId] }, + ); + const rowsKeyed = async (uid: string): Promise => + ( + await server.clients.db.pread( + 'SELECT 1 FROM `notification` WHERE `uid` = ?', + [uid], + ) + ).length; + + it('writes nothing again once the row was rewritten and lost its key', async () => { + const sendKey = await writeSent(node({})); + await node({}).updateValue(sendKey, userId, { title: 'replaced' }); + expect(await sendKeyOf(sendKey)).toBeNull(); + + await server.clients.db.batchWrite(entriesFor(sendKey)); + expect(await rowsKeyed(sendKey)).toBe(1); + }); + + it('adds no row once the replaced row carries the key', async () => { + const replaced = await writeSent(node({})); + const sendKey = uuidv4(); + await server.clients.db.batchWrite(entriesFor(sendKey, replaced)); + expect(await sendKeyOf(replaced)).toBe(sendKey); + expect(await rowsKeyed(sendKey)).toBe(0); + + await server.clients.db.batchWrite(entriesFor(sendKey, replaced)); + expect(await rowsKeyed(sendKey)).toBe(0); + }); +});