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.

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 ;

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();

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<DeferredWriteOutcome> {

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<string, unknown>).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

src/backend/server.ts

@@ -1765,16 +1765,19 @@ export class PuterServer {
         }
     }
 
-    #prepareShutdownHooksRan = false;
+    #prepareShutdownHooks: Promise<void> | 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<void> {
+        this.#prepareShutdownHooks ??= this.#runHook(
+            'bottomUp',
+            'onServerPrepareShutdown',
+        );
+        return this.#prepareShutdownHooks;
     }
 
     async prepareShutdown(info: ServerDrainInfo = {}) {

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<void> {
-        // 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<string, DurableSubscription[]>();
-        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<void> {
+        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 },

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();

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();

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();
     });
 });

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;
 

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 () => {

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;
 

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()}`;

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<DurableReap> {
-        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<DurableReap> {
-        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<DurablePage> {
         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<DurableReap> {
-        if (rows.length === 0) return { rows: [], bumps: [] };
+    async #reap(
+        candidates: readonly DurableSubscription[],
+        stillDue?: { where: string; values: unknown[] },
+    ): Promise<DurableReap> {
+        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);

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,

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(

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;

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<number> =>
+        (
+            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);
+    });
+});