refactor(jobs): run core's periodic sweeps as system jobs, and publish the generation bumps of reaped subscriptions

src/backend/server.test.ts

@@ -36,6 +36,10 @@ import { Controller, Get } from './core/http/decorators.ts';
 import { HttpError } from './core/http/HttpError.ts';
 import { extension, extensionStore } from './extensions.ts';
 import { PuterServer } from './server.ts';
+import {
+    JobScheduler,
+    type JobDefinition,
+} from './services/jobs/JobScheduler.ts';
 import {
     allocateEphemeralPort,
     createTestUser,
@@ -1539,6 +1543,42 @@ describe('PuterServer lifecycle hooks', () => {
     });
 });
 
+/**
+ * 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
+ * first place in the layer.
+ */
+describe('PuterServer jobs replacement', () => {
+    it('lets an extension service registered as `jobs` replace the built-in and keep its first position', async () => {
+        const registered: string[] = [];
+        class ReplacementJobs extends JobScheduler {
+            override register(name: string, definition: JobDefinition) {
+                registered.push(name);
+                super.register(name, definition);
+            }
+        }
+        const started = vi.spyOn(JobScheduler.prototype, 'onServerStart');
+        extension.registerService('jobs', ReplacementJobs as never);
+        let server: PuterServer | undefined;
+        try {
+            server = await setupTestServer();
+            const { jobs } = server.services;
+            expect(jobs).toBeInstanceOf(ReplacementJobs);
+            expect(Object.keys(server.services)[0]).toBe('jobs');
+            expect(extension.import('service').jobs).toBeInstanceOf(
+                ReplacementJobs,
+            );
+            // Only the replacement gets lifecycle hooks.
+            expect(started.mock.contexts).toEqual([jobs]);
+            expect(registered).toContain('auth.sessions.purge');
+        } finally {
+            delete (extensionStore.services as Record<string, unknown>).jobs;
+            started.mockRestore();
+            await server?.shutdown();
+        }
+    });
+});
+
 /**
  * Each layer's `onServerShutdown` has to run while the layers beneath it (the
  * ones it writes through) are still up, so teardown goes top-down: drivers,

src/backend/services/auth/AuthService.test.ts

@@ -24,6 +24,7 @@ import { makeActor, type Actor } from '../../core/actor.js';
 import { PuterServer } from '../../server.js';
 import { WORKER_SUBDOMAIN_PREFIX } from '../../stores/subdomain/SubdomainStore.js';
 import { setupTestServer } from '../../testUtil.js';
+import { testJobContext } from '../jobs/jobsTestUtil.js';
 import { generateDefaultFsentries } from '../../util/userProvisioning.js';
 import { EVENTS_WORKER_SESSION_NAME } from '../events/workerRuntime.js';
 import { FULL_API_ACCESS } from '../permission/consts.js';
@@ -4048,7 +4049,6 @@ describe('AuthService (integration)', () => {
     });
 
     describe('purgeEndedSessions', () => {
-        const GATE_KEY = 'sessions:purge:gate';
         const DAY = 24 * 60 * 60;
 
         const makeSession = async (
@@ -4077,8 +4077,15 @@ describe('AuthService (integration)', () => {
             return rows.length > 0;
         };
 
+        const idOf = async (uuid: string) => {
+            const [row] = await server.clients.db.read(
+                'SELECT `id` FROM `sessions` WHERE `uuid` = ?',
+                [uuid],
+            );
+            return Number(row.id);
+        };
+
         it('deletes rows ended past their retention and keeps the rest', async () => {
-            await server.clients.redis.del(GATE_KEY);
             const user = await makeUser();
             const id = user.id;
             const oldRevoked = await makeSession(id, 'asset', 'revoked_at', 31);
@@ -4100,7 +4107,7 @@ describe('AuthService (integration)', () => {
             const live = await makeSession(id, 'web');
 
             expect(
-                await authService.purgeEndedSessions(),
+                await authService.purgeEndedSessions(testJobContext()),
             ).toBeGreaterThanOrEqual(3);
 
             expect(await exists(oldRevoked)).toBe(false);
@@ -4112,34 +4119,84 @@ describe('AuthService (integration)', () => {
             expect(await exists(live)).toBe(true);
         });
 
-        it('skips while another run holds the gate', async () => {
-            await server.clients.redis.del(GATE_KEY);
-            await server.clients.redis.set(GATE_KEY, 'other-node');
+        it('starts from the saved cursor and clears it once the walk is done', async () => {
             const user = await makeUser();
-            const oldRevoked = await makeSession(
-                user.id,
-                'asset',
-                'revoked_at',
-                31,
-            );
+            const below = await makeSession(user.id, 'asset', 'revoked_at', 31);
+            const above = await makeSession(user.id, 'asset', 'revoked_at', 31);
+            const ctx = testJobContext({ afterId: await idOf(below) });
 
-            expect(await authService.purgeEndedSessions()).toBeNull();
-            expect(await exists(oldRevoked)).toBe(true);
-            await server.clients.redis.del(GATE_KEY);
+            expect(
+                await authService.purgeEndedSessions(ctx),
+            ).toBeGreaterThanOrEqual(1);
+            expect(await exists(below)).toBe(true);
+            expect(await exists(above)).toBe(false);
+            expect(ctx.saved).toEqual([null]);
         });
 
-        it('frees the gate when the walk fails', async () => {
-            await server.clients.redis.del(GATE_KEY);
-            const spy = vi
-                .spyOn(server.clients.db, 'read')
-                .mockRejectedValueOnce(new Error('db down'));
+        it('clears its own checkpoints once a walk completes, so the next run starts at zero', async () => {
+            const SPAN = 5000;
+            vi.spyOn(server.stores.session, 'maxId').mockResolvedValue(
+                SPAN * 11,
+            );
+            const ranges: number[] = [];
+            const purge = vi
+                .spyOn(server.stores.session, 'purgeEndedInRange')
+                .mockImplementation(async (afterId: number) => {
+                    ranges.push(afterId);
+                    return 0;
+                });
             try {
-                await expect(authService.purgeEndedSessions()).rejects.toThrow(
-                    'db down',
+                const first = testJobContext();
+                await authService.purgeEndedSessions(first);
+                expect(ranges).toHaveLength(11);
+                // Checkpointed at ten ranges, then cleared at the end.
+                expect(first.saved).toEqual([{ afterId: SPAN * 10 }, null]);
+
+                ranges.length = 0;
+                await authService.purgeEndedSessions(
+                    testJobContext(first.saved.at(-1)),
                 );
-                expect(await server.clients.redis.get(GATE_KEY)).toBeNull();
+                expect(ranges[0]).toBe(0);
+                expect(ranges).toHaveLength(11);
             } finally {
-                spy.mockRestore();
+                purge.mockRestore();
+                vi.restoreAllMocks();
+            }
+        });
+
+        it('stops between ranges once aborted, and the next run resumes at the saved cursor', async () => {
+            const SPAN = 5000;
+            vi.spyOn(server.stores.session, 'maxId').mockResolvedValue(
+                SPAN * 30,
+            );
+            const ranges: number[] = [];
+            const first = testJobContext();
+            const purge = vi
+                .spyOn(server.stores.session, 'purgeEndedInRange')
+                .mockImplementation(async (afterId: number) => {
+                    ranges.push(afterId);
+                    // Aborted mid-walk, as a lost lease or a shutdown would.
+                    if (ranges.length === 12) first.abort();
+                    return 0;
+                });
+            try {
+                await authService.purgeEndedSessions(first);
+                expect(ranges).toHaveLength(12);
+                // Saved every 10 ranges, and where it stopped.
+                expect(first.saved).toEqual([
+                    { afterId: SPAN * 10 },
+                    { afterId: SPAN * 12 },
+                ]);
+
+                ranges.length = 0;
+                const next = testJobContext(first.saved.at(-1));
+                await authService.purgeEndedSessions(next);
+                expect(ranges[0]).toBe(SPAN * 12);
+                expect(ranges.at(-1)).toBe(SPAN * 29);
+                expect(next.saved.at(-1)).toBeNull();
+            } finally {
+                purge.mockRestore();
+                vi.restoreAllMocks();
             }
         });
     });

src/backend/services/auth/AuthService.ts

@@ -46,6 +46,7 @@ import type { puterServices } from '../index';
 import { EVENTS_WORKER_SESSION_NAME } from '../events/workerRuntime.js';
 import { FULL_API_ACCESS, PERMISSION_MAX_LEN } from '../permission/consts';
 import type { GrantMeta } from '../permission/PermissionService.js';
+import { savedAfterId, type JobRunContext } from '../jobs/JobScheduler.js';
 import { PuterService } from '../types';
 import type {
     AccessTokenPayload,
@@ -64,13 +65,14 @@ const MAX_ORIGIN_UID_GENERATIONS = 8;
 // rows double as sign-in history, so they stay longer.
 const SESSION_RETENTION_SECONDS = 30 * 24 * 60 * 60;
 const WEB_SESSION_RETENTION_SECONDS = 365 * 24 * 60 * 60;
-// Every node checks hourly; the gate lets one node through per week.
-const SESSION_PURGE_CHECK_MS = 60 * 60 * 1000;
-const SESSION_PURGE_GATE_KEY = 'sessions:purge:gate';
-const SESSION_PURGE_GATE_SECONDS = 7 * 24 * 60 * 60;
+const SESSION_PURGE_JOB = 'auth.sessions.purge';
+const SESSION_PURGE_EVERY_MS = 7 * 24 * 60 * 60 * 1000;
+const SESSION_PURGE_LEASE_MS = 30 * 60 * 1000;
 // Ids per DELETE, with a pause between, so replicas keep up.
 const SESSION_PURGE_SPAN = 5000;
 const SESSION_PURGE_PAUSE_MS = 50;
+/** Ranges walked between saves of the cursor. */
+const SESSION_PURGE_SAVE_EVERY = 10;
 
 const nowSeconds = (): number => Math.floor(Date.now() / 1000);
 
@@ -138,11 +140,17 @@ export interface AuthResult {
 export class AuthService extends PuterService {
     declare protected services: LayerInstances<typeof puterServices>;
 
-    #sessionPurge: ReturnType<typeof setInterval> | null = null;
-    #stopping = false;
-
     override onServerStart(): void {
-        this.#armSessionPurge();
+        this.services.jobs.register(SESSION_PURGE_JOB, {
+            every: SESSION_PURGE_EVERY_MS,
+            scope: 'fleet',
+            nearPrimary: true,
+            leaseMs: SESSION_PURGE_LEASE_MS,
+            run: async (ctx) => {
+                const purged = await this.purgeEndedSessions(ctx);
+                console.log(`[auth] purged ${purged} ended sessions`);
+            },
+        });
 
         // Users implicitly hold read access to their own email — needed for
         // any permission-gated path that asks for `user:<uuid>:email:read`
@@ -164,85 +172,54 @@ export class AuthService extends PuterService {
         });
     }
 
-    override onServerPrepareShutdown(): void {
-        this.#stopping = true;
-        if (this.#sessionPurge) clearInterval(this.#sessionPurge);
-        this.#sessionPurge = null;
-    }
-
     // -- Session purge -----------------------------------------------
 
     /**
      * Delete sessions revoked or expired longer ago than their retention,
-     * walking the table by id. Returns the rows deleted, or null when another
-     * node already holds this week's gate.
+     * walking the table by id from the cursor an interrupted run saved. Returns
+     * the rows deleted; stops early once `signal` aborts.
      */
-    async purgeEndedSessions(): Promise<number | null> {
-        const claimed = await this.clients.redis.set(
-            SESSION_PURGE_GATE_KEY,
-            String(Date.now()),
-            'EX',
-            SESSION_PURGE_GATE_SECONDS,
-            'NX',
-        );
-        if (claimed !== 'OK') return null;
-
+    async purgeEndedSessions({
+        signal,
+        state,
+        saveState,
+    }: JobRunContext): Promise<number> {
         const now = nowSeconds();
         const cutoffs = {
             cutoff: now - SESSION_RETENTION_SECONDS,
             webCutoff: now - WEB_SESSION_RETENTION_SECONDS,
         };
+        const maxId = await this.stores.session.maxId();
+        let afterId = savedAfterId(state);
         let purged = 0;
-        let finished = false;
-        try {
-            const maxId = await this.stores.session.maxId();
-            for (
-                let afterId = 0;
-                afterId < maxId;
-                afterId += SESSION_PURGE_SPAN
-            ) {
-                if (this.#stopping) return purged;
-                purged += await this.stores.session.purgeEndedInRange(
-                    afterId,
-                    SESSION_PURGE_SPAN,
-                    cutoffs,
-                );
-                await new Promise((resolve) =>
-                    setTimeout(resolve, SESSION_PURGE_PAUSE_MS),
-                );
-            }
-            finished = true;
-            return purged;
-        } finally {
-            // A walk cut short frees the gate, so another node restarts it on
-            // its next check instead of next week.
-            if (!finished) {
+        let unsaved = 0;
+        while (afterId < maxId) {
+            if (signal.aborted) {
+                // Kept for the next run if this one still holds the job.
                 try {
-                    await this.clients.redis.del(SESSION_PURGE_GATE_KEY);
+                    await saveState({ afterId });
                 } catch {
-                    // The gate expires on its own.
+                    // Lost the job; the next run redoes these ranges.
                 }
+                return purged;
             }
-        }
-    }
-
-    #armSessionPurge(): void {
-        const timer = setInterval(() => {
-            void this.#runSessionPurge();
-        }, SESSION_PURGE_CHECK_MS);
-        timer.unref?.();
-        this.#sessionPurge = timer;
-    }
-
-    async #runSessionPurge(): Promise<void> {
-        try {
-            const purged = await this.purgeEndedSessions();
-            if (purged !== null) {
-                console.log(`[auth] purged ${purged} ended sessions`);
+            purged += await this.stores.session.purgeEndedInRange(
+                afterId,
+                SESSION_PURGE_SPAN,
+                cutoffs,
+            );
+            afterId += SESSION_PURGE_SPAN;
+            if (++unsaved >= SESSION_PURGE_SAVE_EVERY) {
+                await saveState({ afterId });
+                unsaved = 0;
             }
-        } catch (err) {
-            console.warn('[auth] session purge failed', err);
+            await new Promise((resolve) =>
+                setTimeout(resolve, SESSION_PURGE_PAUSE_MS),
+            );
         }
+        // Clears any checkpoint, this run's included, so next week starts at 0.
+        await saveState(null);
+        return purged;
     }
 
     // -- Public API --------------------------------------------------

src/backend/services/events/EventsService.test.ts

@@ -293,12 +293,14 @@ const buildService = (
     metered: MeteredLine[];
     forwarded: ForwardedCall[];
     eventBus: { on: ReturnType<typeof vi.fn>; emit: ReturnType<typeof vi.fn> };
+    jobs: { register: ReturnType<typeof vi.fn> };
 } => {
     const outbox: Array<{ socket?: string; envelope: DeliveryEnvelope }> = [];
     const counted: DeliveryEnvelope[] = [];
     const lines: MeteredLine[] = [];
     const forwardedCalls: ForwardedCall[] = [];
     const bus = { on: vi.fn(), emit: vi.fn() };
+    const jobs = { register: vi.fn() };
     const built = new EventsService(
         config,
         {
@@ -315,6 +317,7 @@ const buildService = (
             kvShareHandle: kvShareHandleStore,
         } as never,
         {
+            jobs,
             eventForward: {
                 // A deployment with no peers has nowhere to forward to, which
                 // is what every test here is.
@@ -376,6 +379,7 @@ const buildService = (
         metered: lines,
         forwarded: forwardedCalls,
         eventBus: bus,
+        jobs,
     };
 };
 
@@ -564,6 +568,52 @@ describe('the feature switch', () => {
         );
     });
 
+    it('hands its table sweeps to the job scheduler and arms only the pending sweep itself', () => {
+        const intervals = vi.spyOn(globalThis, 'setInterval');
+        const timeouts = vi.spyOn(globalThis, 'setTimeout');
+        let jobs: { register: ReturnType<typeof vi.fn> };
+        try {
+            ({ jobs } = buildService({ events: { enabled: true } } as IConfig));
+            // The pending sweep is the region's queue, so every node runs it.
+            expect(intervals.mock.calls.map(([, ms]) => ms)).toEqual([10_000]);
+            expect(timeouts).not.toHaveBeenCalled();
+        } finally {
+            intervals.mockRestore();
+            timeouts.mockRestore();
+        }
+        expect(
+            jobs.register.mock.calls.map(([name, job]) => ({
+                name,
+                every: job.every,
+                scope: job.scope,
+                nearPrimary: job.nearPrimary ?? false,
+                leaseMs: job.leaseMs,
+            })),
+        ).toEqual([
+            {
+                name: 'events.subscriptions.expirySweep',
+                every: 60 * 60_000,
+                scope: 'fleet',
+                nearPrimary: true,
+                leaseMs: 15 * 60_000,
+            },
+            {
+                name: 'events.subscriptions.creditSweep',
+                every: 15 * 60_000,
+                scope: 'fleet',
+                nearPrimary: false,
+                leaseMs: 10 * 60_000,
+            },
+        ]);
+    });
+
+    it('registers no job when off', () => {
+        const { jobs } = buildService({
+            events: { enabled: false },
+        } as IConfig);
+        expect(jobs.register).not.toHaveBeenCalled();
+    });
+
     it('spends nothing at all on a dispatch when off', async () => {
         const { service: off } = buildService({
             events: { enabled: false },

src/backend/services/events/EventsService.ts

@@ -56,7 +56,9 @@ import {
 import { HttpError, isHttpError } from '../../core/http/HttpError.js';
 import { checkRateLimit } from '../../core/http/middleware/rateLimit.js';
 import type {
+    DurableCreated,
     ReanchorInput,
+    ReapResult,
     SuspendedReason,
 } from '../../stores/events/DurableSubscriptionStore.js';
 import type { LiveKvShareHandle } from '../../stores/events/KvShareHandleStore.js';
@@ -116,6 +118,7 @@ import {
     appSocketRoom,
     type SocketSpecifier,
 } from '../socket/SocketService.js';
+import { savedAfterId, type JobRunContext } from '../jobs/JobScheduler.js';
 import { PuterService } from '../types.js';
 import { DEFAULT_FREE_SUBSCRIPTION } from '../metering/consts.js';
 import type { IConfig } from '../../types.js';
@@ -677,11 +680,11 @@ const ENDED_MESSAGES: Record<SubscriptionEndReason, string> = {
 
 // -- Expiry sweep -----------------------------------------------------
 
+/** Expired and long-suspended rows, then stranded worker sessions. */
+const EXPIRY_SWEEP_JOB = 'events.subscriptions.expirySweep';
 /** An expiry is a date, not a deadline, so hourly is close enough. */
-const EXPIRY_SWEEP_INTERVAL_MS = 60 * 60 * 1000;
-// The first pass lands within this long of boot, so a fleet that redeploys more
-// often than the interval still sweeps.
-const EXPIRY_SWEEP_INITIAL_DELAY_MS = 5 * 60 * 1000;
+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. */
@@ -695,15 +698,16 @@ const ENDED_SUBJECTS_LISTED = 20;
 /** Re-reads an unsubscribe takes to catch a row racing a reanchor. */
 const UNSUBSCRIBE_READS = 3;
 
-/**
- * Worker sessions one page of the stray-session sweep reads, and pages one pass
- * takes. Bounded low: the sweep is a safety net for sessions no API path could
- * reach any more, not a bulk cleanup, and it runs every hour.
- */
 /** A deleted account's apps settled at once. */
 const DELETED_ACCOUNT_SETTLE_CONCURRENCY = 4;
 /** Session rows ended at once when their handle stops standing. */
 const SESSION_END_CONCURRENCY = 4;
+/**
+ * Worker sessions one page of the stray-session sweep reads, and pages one pass
+ * takes. Bounded low: the sweep is a safety net for sessions no API path could
+ * reach any more, not a bulk cleanup; it runs every hour and resumes where the
+ * last pass stopped.
+ */
 const WORKER_SESSION_SWEEP_BATCH = 500;
 const WORKER_SESSION_SWEEP_MAX_PAGES = 2;
 
@@ -751,12 +755,14 @@ const METER_ACTOR_LIMIT = 5_000;
 const NO_CREDIT_SWEEP_BATCH = 500;
 const NO_CREDIT_SWEEP_MAX_BATCHES = 200;
 
+const NO_CREDIT_SWEEP_JOB = 'events.subscriptions.creditSweep';
 /**
  * How often suspensions waiting on a restored balance are re-checked. Well
  * inside the hour a `no_credit` backlog is held for, so a top-up gets the
  * subscription back before what it was owed expires.
  */
-const NO_CREDIT_SWEEP_INTERVAL_MS = 15 * 60 * 1000;
+const NO_CREDIT_SWEEP_EVERY_MS = 15 * 60 * 1000;
+const NO_CREDIT_SWEEP_LEASE_MS = 10 * 60 * 1000;
 
 /** The part of a socket this service uses, so tests need not build one. */
 export interface EventSocket {
@@ -1376,10 +1382,7 @@ export class EventsService extends PuterService {
     readonly #flushing = new Map<string, number>();
     /** Paces settles started by deliveries that found consent withdrawn. */
     readonly #consentSettles = new ConsentSettles();
-    #expirySweep: ReturnType<typeof setInterval> | null = null;
-    #expiryKick: ReturnType<typeof setTimeout> | null = null;
     #pendingSweep: ReturnType<typeof setInterval> | null = null;
-    #creditSweep: ReturnType<typeof setInterval> | null = null;
 
     /**
      * What runs an app's handler. Replaced at start-up by the invoker that
@@ -1524,20 +1527,13 @@ export class EventsService extends PuterService {
             });
         });
 
-        this.#armExpirySweep();
         this.#armPendingSweep();
-        this.#armCreditSweep();
+        if (this.enabled) this.#registerJobs();
     }
 
     override onServerPrepareShutdown(): void {
-        if (this.#expiryKick) clearTimeout(this.#expiryKick);
-        this.#expiryKick = null;
-        if (this.#expirySweep) clearInterval(this.#expirySweep);
-        this.#expirySweep = null;
         if (this.#pendingSweep) clearInterval(this.#pendingSweep);
         this.#pendingSweep = null;
-        if (this.#creditSweep) clearInterval(this.#creditSweep);
-        this.#creditSweep = null;
     }
 
     override onServerShutdown(): void {
@@ -1886,7 +1882,7 @@ export class EventsService extends PuterService {
         if (includeValue) assertValueDeliverable(rawSubject);
         const anchor = await this.#resolveSubscribeAnchor(actor, rawSubject);
 
-        let created: { row: DurableSubscription; bump: GenerationBump };
+        let created: DurableCreated;
         try {
             created = await this.stores.durableSubscription.create({
                 holderUserId,
@@ -1913,8 +1909,11 @@ export class EventsService extends PuterService {
                 this.#publishGeneration(err.bump, true);
             throw err;
         }
-        const { row, bump } = created;
+        const { row, bump, reaped } = created;
         this.#publishGeneration(bump, true);
+        for (const trimmed of reaped)
+            if (trimmed.userId !== bump.userId)
+                this.#publishGeneration(trimmed, true);
 
         return { sub: toDurableView(row) };
     }
@@ -2667,15 +2666,13 @@ export class EventsService extends PuterService {
         );
     }
 
-    /**
-     * Drop rows past their expiry, in batches, and report how many went. Every
-     * node sweeps; the delete is idempotent, so two overlapping costs a few
-     * empty batches rather than correctness.
-     */
-    async sweepExpired(): Promise<number> {
+    /** Drop rows past their expiry, in batches, and report how many went. */
+    async sweepExpired(signal?: AbortSignal): Promise<number> {
         if (!this.enabled) return 0;
-        return this.#sweepInBatches((batchSize) =>
-            this.stores.durableSubscription.sweepExpired(batchSize),
+        return this.#sweepInBatches(
+            (batchSize) =>
+                this.stores.durableSubscription.sweepExpired(batchSize),
+            signal,
         );
     }
 
@@ -2684,13 +2681,18 @@ export class EventsService extends PuterService {
      * revoked subscription never resumes, so the row survives only as the
      * answer its holder gets from `list` when they ask why it stopped.
      */
-    async sweepSuspended(): Promise<number> {
+    async sweepSuspended(signal?: AbortSignal): Promise<number> {
         if (!this.enabled) return 0;
         const cutoff =
             Math.floor(Date.now() / 1000) -
             SUSPENDED_ROW_TTL_DAYS * 24 * 60 * 60;
-        return this.#sweepInBatches((batchSize) =>
-            this.stores.durableSubscription.sweepSuspended(cutoff, batchSize),
+        return this.#sweepInBatches(
+            (batchSize) =>
+                this.stores.durableSubscription.sweepSuspended(
+                    cutoff,
+                    batchSize,
+                ),
+            signal,
         );
     }
 
@@ -2701,20 +2703,26 @@ export class EventsService extends PuterService {
      * existed). Scoped to this one worker name; the same strand for
      * `WorkerDriver`-minted sessions is a separate finding, not covered here.
      */
-    async sweepStrandedWorkerSessions(): Promise<number> {
+    async sweepStrandedWorkerSessions({
+        signal,
+        state,
+        saveState,
+    }: JobRunContext): Promise<number> {
         if (!this.enabled) return 0;
 
         let revoked = 0;
-        let afterId = 0;
-        for (let page = 0; page < WORKER_SESSION_SWEEP_MAX_PAGES; page++) {
+        let afterId = savedAfterId(state);
+        let saved = afterId;
+        for (
+            let page = 0;
+            page < WORKER_SESSION_SWEEP_MAX_PAGES && !signal.aborted;
+            page++
+        ) {
             const rows = await this.stores.session.listWorkerSessions({
                 workerName: EVENTS_WORKER_SESSION_NAME,
                 afterId,
                 limit: WORKER_SESSION_SWEEP_BATCH,
             });
-            if (rows.length === 0) break;
-            afterId = rows[rows.length - 1].id;
-
             const appUids = [
                 ...new Set(rows.map((row) => row.appUid).filter(Boolean)),
             ] as string[];
@@ -2734,7 +2742,16 @@ export class EventsService extends PuterService {
                 }
             }
 
-            if (rows.length < WORKER_SESSION_SWEEP_BATCH) break;
+            // A short page is the end of the table; the next run wraps.
+            afterId =
+                rows.length < WORKER_SESSION_SWEEP_BATCH
+                    ? 0
+                    : rows[rows.length - 1].id;
+            if (afterId !== saved) {
+                await saveState(afterId === 0 ? null : { afterId });
+                saved = afterId;
+            }
+            if (afterId === 0) break;
         }
         return revoked;
     }
@@ -2744,13 +2761,25 @@ export class EventsService extends PuterService {
      * top-up is not something this service hears about, and coupling delivery
      * to the payment path would make one more thing that has to be told. The
      * cost of the delay is bounded by how long a `no_credit` backlog is held.
+     *
+     * Each run picks up where the last one stopped and wraps at the end of the
+     * table, so a backlog past one run's cap is still reached.
      */
-    async sweepNoCredit(): Promise<number> {
+    async sweepNoCredit({
+        signal,
+        state,
+        saveState,
+    }: JobRunContext): Promise<number> {
         if (!this.enabled || !this.services.metering) return 0;
 
         let resumed = 0;
-        let after = 0;
-        for (let pass = 0; pass < NO_CREDIT_SWEEP_MAX_BATCHES; pass++) {
+        let after = savedAfterId(state);
+        let saved = after;
+        for (
+            let pass = 0;
+            pass < NO_CREDIT_SWEEP_MAX_BATCHES && !signal.aborted;
+            pass++
+        ) {
             const page =
                 await this.stores.durableSubscription.listSuspendedPage(
                     'no_credit',
@@ -2759,6 +2788,8 @@ export class EventsService extends PuterService {
                 );
             const holders = new Set(page.rows.map((row) => row.holderUserId));
             for (const holderUserId of holders) {
+                // The page is redone next run.
+                if (signal.aborted) return resumed;
                 const actor = await this.#meterActor({
                     holderUserId,
                     appUid: null,
@@ -2777,20 +2808,30 @@ export class EventsService extends PuterService {
                 }
                 resumed += await this.resumeForCredit(holderUserId);
             }
-            if (page.nextId === null) break;
-            after = page.nextId;
+            after = page.nextId ?? 0;
+            if (after !== saved) {
+                await saveState(after === 0 ? null : { afterId: after });
+                saved = after;
+            }
+            if (after === 0) break;
         }
         return resumed;
     }
 
+    /**
+     * Reap in batches until a short one, publishing each batch's bumps so every
+     * region stops delivering against what went.
+     */
     async #sweepInBatches(
-        pass: (batchSize: number) => Promise<number>,
+        pass: (batchSize: number) => Promise<ReapResult>,
+        signal?: AbortSignal,
     ): Promise<number> {
         let removed = 0;
-        for (let i = 0; i < EXPIRY_MAX_BATCHES; i++) {
-            const batch = await pass(EXPIRY_BATCH_SIZE);
-            removed += batch;
-            if (batch < EXPIRY_BATCH_SIZE) break;
+        for (let i = 0; i < EXPIRY_MAX_BATCHES && !signal?.aborted; i++) {
+            const { rows, bumps } = await pass(EXPIRY_BATCH_SIZE);
+            for (const bump of bumps) this.#publishGeneration(bump, true);
+            removed += rows.length;
+            if (rows.length < EXPIRY_BATCH_SIZE) break;
         }
         return removed;
     }
@@ -6465,37 +6506,27 @@ export class EventsService extends PuterService {
         }
     }
 
-    #armExpirySweep(): void {
-        if (!this.enabled) return;
-        const run = () => {
-            void this.sweepExpired()
-                .then(() => this.sweepSuspended())
-                .then(() => this.sweepStrandedWorkerSessions())
-                .catch((err) => {
-                    console.warn('[events] expiry sweep failed', err);
-                });
-        };
-        // Jittered so a deploy does not have every node sweep at once.
-        const kick = setTimeout(
-            run,
-            EXPIRY_SWEEP_INITIAL_DELAY_MS * (0.5 + Math.random()),
-        );
-        kick.unref?.();
-        this.#expiryKick = kick;
-        const sweep = setInterval(run, EXPIRY_SWEEP_INTERVAL_MS);
-        sweep.unref?.();
-        this.#expirySweep = sweep;
-    }
-
-    #armCreditSweep(): void {
-        if (!this.enabled) return;
-        const sweep = setInterval(() => {
-            void this.sweepNoCredit().catch((err) => {
-                console.warn('[events] credit sweep failed', err);
-            });
-        }, NO_CREDIT_SWEEP_INTERVAL_MS);
-        sweep.unref?.();
-        this.#creditSweep = sweep;
+    /** The table sweeps, each run once per cadence across the fleet. */
+    #registerJobs(): void {
+        this.services.jobs.register(EXPIRY_SWEEP_JOB, {
+            every: EXPIRY_SWEEP_EVERY_MS,
+            scope: 'fleet',
+            nearPrimary: true,
+            leaseMs: EXPIRY_SWEEP_LEASE_MS,
+            run: async (ctx) => {
+                await this.sweepExpired(ctx.signal);
+                await this.sweepSuspended(ctx.signal);
+                await this.sweepStrandedWorkerSessions(ctx);
+            },
+        });
+        this.services.jobs.register(NO_CREDIT_SWEEP_JOB, {
+            every: NO_CREDIT_SWEEP_EVERY_MS,
+            scope: 'fleet',
+            leaseMs: NO_CREDIT_SWEEP_LEASE_MS,
+            run: async (ctx) => {
+                await this.sweepNoCredit(ctx);
+            },
+        });
     }
 
     #armPendingSweep(): void {

src/backend/services/events/appDeletion.integration.test.ts

@@ -29,6 +29,7 @@
 import { afterAll, beforeAll, beforeEach, describe, expect, it, vi } from 'vitest';
 import { v4 as uuidv4 } from 'uuid';
 import { setupPuterTestEnv, type PuterTestEnv } from '../../testUtil.js';
+import { testJobContext } from '../jobs/jobsTestUtil.js';
 import type { IConfig } from '../../types.js';
 import { EVENTS_BACKGROUND_PERMISSION } from './authorization.js';
 import { EVENTS_WORKER_SESSION_NAME } from './workerRuntime.js';
@@ -327,7 +328,9 @@ describe('an app deleted while it has events state', () => {
         await env.server.stores.app.delete((row as { id: number }).id);
 
         const revoked =
-            await env.server.services.events.sweepStrandedWorkerSessions();
+            await env.server.services.events.sweepStrandedWorkerSessions(
+                testJobContext(),
+            );
         expect(revoked).toBeGreaterThanOrEqual(1);
 
         // `authenticate()` can no longer resolve the token at all — the app
@@ -366,9 +369,60 @@ describe('an app deleted while it has events state', () => {
             EVENTS_WORKER_SESSION_NAME,
         );
 
-        await env.server.services.events.sweepStrandedWorkerSessions();
+        await env.server.services.events.sweepStrandedWorkerSessions(
+            testJobContext(),
+        );
 
         const reauth = await env.server.services.auth.authenticate(token);
         expect(reauth).not.toHaveProperty('reauth');
     });
+
+    it('reaches a stranded session past one run`s two pages on the next run, then wraps', async () => {
+        const db = env.server.clients.db;
+        const now = Math.floor(Date.now() / 1000);
+        const meta = JSON.stringify({
+            worker_name: EVENTS_WORKER_SESSION_NAME,
+        });
+        const workerSession = (appUid: string) => ({
+            statement:
+                'INSERT INTO `sessions` (`uuid`, `user_id`, `meta`, `last_activity`, `created_at`, `kind`, `app_uid`) ' +
+                "VALUES (?, ?, ?, ?, ?, 'worker', ?)",
+            values: [uuidv4(), userId, meta, now, now, appUid],
+        });
+        // A run reads two pages of 500; these fill them, all still standing.
+        const live = Array.from(
+            { length: 1000 },
+            (_, i) => `app-live-${uuidv4().slice(0, 8)}-${i}`,
+        );
+        await db.batchWrite(
+            live.map((uid) => ({
+                statement:
+                    'INSERT INTO `apps` (`uid`, `name`, `title`, `index_url`, `owner_user_id`) VALUES (?, ?, ?, ?, ?)',
+                values: [uid, uid, uid, `https://${uid}.example/`, userId],
+            })),
+        );
+        await db.batchWrite(live.map(workerSession));
+        const stranded = workerSession(`app-gone-${uuidv4()}`);
+        await db.batchWrite([stranded]);
+        const strandedRevoked = async () => {
+            const [row] = await db.pread(
+                'SELECT `revoked_at` FROM `sessions` WHERE `uuid` = ?',
+                [stranded.values[0]],
+            );
+            return row.revoked_at !== null;
+        };
+
+        const first = testJobContext();
+        await env.server.services.events.sweepStrandedWorkerSessions(first);
+        expect(await strandedRevoked()).toBe(false);
+        expect(first.saved.at(-1)).toMatchObject({
+            afterId: expect.any(Number),
+        });
+
+        const second = testJobContext(first.saved.at(-1));
+        await env.server.services.events.sweepStrandedWorkerSessions(second);
+        expect(await strandedRevoked()).toBe(true);
+        // The end of the table wraps the cursor back to the start.
+        expect(second.saved.at(-1)).toBeNull();
+    });
 });

src/backend/services/events/durable.integration.test.ts

@@ -872,6 +872,71 @@ describe('what a write costs once a durable row exists', () => {
 
         expect(deliveryOf(path)).toBeUndefined();
     });
+
+    it('publishes the expiry reap`s bump, so a region still caching the row drops it', async () => {
+        await clearRows();
+        const created = await subscribe(env.users.user.token, {
+            expiresAt: Math.floor(Date.now() / 1000) + 3600,
+        });
+        const subId = created.body.subId as string;
+        const stale =
+            await env.server.stores.durableSubscription.getBySubId(subId);
+        await env.server.clients.db.write(
+            `UPDATE \`${TABLE}\` SET \`expires_at\` = ? WHERE \`sub_id\` = ?`,
+            [Math.floor(Date.now() / 1000) - 1, subId],
+        );
+
+        const heard: Array<Record<string, unknown>> = [];
+        const listener = (_key: string, data: unknown) => {
+            heard.push(data as Record<string, unknown>);
+        };
+        const announced = vi.spyOn(
+            env.server.services.eventForward,
+            'announceGeneration',
+        );
+        env.server.clients.event.on(
+            'outer.pubsub.events.generationBumped',
+            listener as never,
+        );
+        try {
+            await expect(events().sweepExpired()).resolves.toBe(1);
+            // Over the addressed peer channel as well as the bus.
+            expect(announced).toHaveBeenCalledWith(
+                expect.objectContaining({ userId }),
+                true,
+                'subscription',
+            );
+        } finally {
+            env.server.clients.event.off(
+                'outer.pubsub.events.generationBumped',
+                listener as never,
+            );
+            announced.mockRestore();
+        }
+        const published = heard.find(
+            (bump) => bump.userId === userId && bump.durable === true,
+        );
+        expect(published).toBeDefined();
+
+        // Another region's Redis still holds the row; this one stands in for it.
+        await env.server.stores.eventSubscription.cacheDurable([stale!]);
+        delivered.length = 0;
+        const before = `${anchor}/stale-${uuidv4().slice(0, 8)}.txt`;
+        await fs().touch(userId, { path: before });
+        await settle(before);
+        expect(deliveryOf(before)?.subId).toBe(subId);
+
+        await env.server.clients.event.emitAndWait(
+            'outer.pubsub.events.generationBumped',
+            published!,
+            { from_outside: true },
+        );
+        delivered.length = 0;
+        const after = `${anchor}/peer-swept-${uuidv4().slice(0, 8)}.txt`;
+        await fs().touch(userId, { path: after });
+        await quiet();
+        expect(deliveryOf(after)).toBeUndefined();
+    });
 });
 
 describe('where a durable delivery is addressed', () => {

src/backend/services/events/forwarding.test.ts

@@ -441,6 +441,7 @@ const makeRegion = (
             ): Promise<{
                 row: DurableSubscription;
                 bump: { userId: number; generation: number };
+                reaped: [];
             }> => {
                 const row = durableRow({
                     holderUserId: input.holderUserId as number,
@@ -469,6 +470,7 @@ const makeRegion = (
                             row.ownerUserId,
                         ),
                     },
+                    reaped: [],
                 };
             },
         },

src/backend/services/events/metering.test.ts

@@ -43,6 +43,7 @@ import {
     DEFAULT_TEMP_SUBSCRIPTION,
 } from '../metering/consts.js';
 import type { UsageInput } from '../metering/types.js';
+import { testJobContext } from '../jobs/jobsTestUtil.js';
 import type { IConfig } from '../../types.js';
 import { EVENTS_COSTS } from './costs.js';
 import { EventsService } from './EventsService.js';
@@ -62,6 +63,8 @@ let solvent: Set<number>;
 /** Set to have the next create lose the last slot after its insert. */
 let raceLost: boolean;
 let emitted: Array<{ key: string; data: unknown }>;
+/** Rows one credit-sweep page takes; null for the service's own batch. */
+let pageSize: number | null;
 
 const anchorUid = (): string => `docs-${seq}`;
 const anchorPath = (): string => `/u${userId}/Documents`;
@@ -124,6 +127,7 @@ beforeEach(() => {
     solvent = new Set([userId]);
     raceLost = false;
     emitted = [];
+    pageSize = null;
 
     service = new EventsService(
         { events: { enabled: true } } as IConfig,
@@ -155,17 +159,28 @@ beforeEach(() => {
                     return {
                         row,
                         bump: { userId: input.ownerUserId, generation: 1 },
+                        reaped: [],
+                    };
+                },
+                // Keyset paging over `rows`, a row's id being its position + 1.
+                listSuspendedPage: async (
+                    reason: string,
+                    afterId: number,
+                    batchSize: number,
+                ) => {
+                    const size = pageSize ?? batchSize;
+                    const page = rows
+                        .map((row, i) => ({ row, id: i + 1 }))
+                        .filter(
+                            ({ row, id }) =>
+                                id > afterId && row.suspendedReason === reason,
+                        )
+                        .slice(0, size);
+                    return {
+                        rows: page.map(({ row }) => row),
+                        nextId: page.length < size ? null : page.at(-1)!.id,
                     };
                 },
-                listSuspendedPage: async (reason: string, afterId: number) =>
-                    afterId === 0
-                        ? {
-                              rows: rows.filter(
-                                  (row) => row.suspendedReason === reason,
-                              ),
-                              nextId: null,
-                          }
-                        : { rows: [], nextId: null },
                 listSuspendedForHolder: async (
                     holderUserId: number,
                     reason: string,
@@ -258,23 +273,78 @@ describe('a suspension waiting on a balance', () => {
             suspendedReason: 'no_credit',
         });
 
-    it('comes back once the account has budget again', async () => {
+    it('comes back once the account has budget again, and every region hears it', async () => {
         const row = suspended();
         rows.push(row);
 
-        await expect(service.sweepNoCredit()).resolves.toBe(1);
+        await expect(service.sweepNoCredit(testJobContext())).resolves.toBe(1);
         expect(resumed).toEqual([row.subId]);
         expect(row.suspendedReason).toBeNull();
+        expect(emitted).toContainEqual({
+            key: 'outer.pubsub.events.generationBumped',
+            data: { userId, generation: 1, durable: true },
+        });
     });
 
     it('stays out of service while the account still has none', async () => {
         rows.push(suspended());
         solvent.clear();
 
-        await expect(service.sweepNoCredit()).resolves.toBe(0);
+        await expect(service.sweepNoCredit(testJobContext())).resolves.toBe(0);
         expect(resumed).toEqual([]);
     });
 
+    it('reaches rows past one run`s page cap across runs, then wraps', async () => {
+        // One row per page, so a run's 200 pages stop at 200 rows.
+        pageSize = 1;
+        for (let i = 0; i < 450; i++) rows.push(suspended());
+        for (const [i, row] of rows.entries()) row.holderUserId = 10_000 + i;
+        // Only the last holder has a balance again.
+        solvent = new Set([10_000 + 449]);
+
+        const first = testJobContext();
+        await expect(service.sweepNoCredit(first)).resolves.toBe(0);
+        expect(first.saved.at(-1)).toEqual({ afterId: 200 });
+
+        const second = testJobContext(first.saved.at(-1));
+        await expect(service.sweepNoCredit(second)).resolves.toBe(0);
+        expect(second.saved.at(-1)).toEqual({ afterId: 400 });
+
+        const third = testJobContext(second.saved.at(-1));
+        await expect(service.sweepNoCredit(third)).resolves.toBe(1);
+        expect(resumed).toEqual([rows[449].subId]);
+        // The end of the table wraps the cursor back to the start.
+        expect(third.saved.at(-1)).toBeNull();
+    });
+
+    it('stops between pages once its signal aborts, keeping the last full page', async () => {
+        pageSize = 1;
+        for (let i = 0; i < 5; i++) rows.push(suspended());
+        for (const [i, row] of rows.entries()) row.holderUserId = 20_000 + i;
+        const ctx = testJobContext();
+        const save = ctx.saveState;
+        ctx.saveState = async (next) => {
+            await save(next);
+            if (ctx.saved.length === 2) ctx.abort();
+        };
+
+        await service.sweepNoCredit(ctx);
+        expect(ctx.saved).toEqual([{ afterId: 1 }, { afterId: 2 }]);
+    });
+
+    it('wraps a saved cursor past the end of a shrunken table back to the start', async () => {
+        rows.push(suspended());
+        const ctx = testJobContext({ afterId: 999 });
+        await expect(service.sweepNoCredit(ctx)).resolves.toBe(0);
+        expect(ctx.saved).toEqual([null]);
+    });
+
+    it('writes no cursor while there is nothing to page through', async () => {
+        const ctx = testJobContext();
+        await expect(service.sweepNoCredit(ctx)).resolves.toBe(0);
+        expect(ctx.saved).toEqual([]);
+    });
+
     it('leaves a suspension it did not cause alone', async () => {
         rows.push(
             durableRow({
@@ -283,7 +353,7 @@ describe('a suspension waiting on a balance', () => {
             }),
         );
 
-        await expect(service.sweepNoCredit()).resolves.toBe(0);
+        await expect(service.sweepNoCredit(testJobContext())).resolves.toBe(0);
         expect(resumed).toEqual([]);
     });
 });

src/backend/services/jobs/systemJobs.test.ts

@@ -0,0 +1,267 @@
+/*
+ * Copyright (C) 2024-present Puter Technologies Inc.
+ *
+ * This file is part of Puter.
+ *
+ * Puter is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as published
+ * by the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program.  If not, see <https://www.gnu.org/licenses/>.
+ */
+
+/**
+ * The backend's own periodic jobs on the default scheduler: what registers,
+ * that each runs once across nodes sharing Redis, and that a single self-hosted
+ * node (sqlite, in-process Redis) runs every one of them.
+ */
+
+import { randomUUID } from 'node:crypto';
+import { afterAll, beforeAll, describe, expect, it, vi } from 'vitest';
+import type { PuterServer } from '../../server.js';
+import { setupTestServer } from '../../testUtil.js';
+import type { IConfig } from '../../types.js';
+import {
+    JobScheduler,
+    type JobDefinition,
+    type JobRunContext,
+} from './JobScheduler.js';
+
+type SchedulerArgs = ConstructorParameters<typeof JobScheduler>;
+
+const MINUTE = 60_000;
+const HOUR = 60 * MINUTE;
+
+const EXPECTED_JOBS = {
+    'auth.sessions.purge': {
+        every: 7 * 24 * HOUR,
+        scope: 'fleet',
+        nearPrimary: true,
+        leaseMs: 30 * MINUTE,
+    },
+    'kvShare.handles.purgeRevoked': {
+        every: 24 * HOUR,
+        scope: 'fleet',
+        nearPrimary: true,
+        leaseMs: 15 * MINUTE,
+    },
+    'notifications.retention': {
+        every: HOUR,
+        scope: 'fleet',
+        nearPrimary: true,
+        leaseMs: 15 * MINUTE,
+    },
+    'metering.globalRateCheck': {
+        every: 25 * MINUTE,
+        scope: 'fleet',
+        nearPrimary: false,
+        leaseMs: 5 * MINUTE,
+    },
+    'events.subscriptions.expirySweep': {
+        every: HOUR,
+        scope: 'fleet',
+        nearPrimary: true,
+        leaseMs: 15 * MINUTE,
+    },
+    'events.subscriptions.creditSweep': {
+        every: 15 * MINUTE,
+        scope: 'fleet',
+        nearPrimary: false,
+        leaseMs: 10 * MINUTE,
+    },
+};
+const JOB_NAMES = Object.keys(EXPECTED_JOBS);
+
+/** A timer armed while the server booted, and the frame that armed it. */
+interface ArmedTimer {
+    kind: 'interval' | 'timeout';
+    ms: number | undefined;
+    caller: string;
+}
+
+let server: PuterServer;
+const registered = new Map<string, JobDefinition>();
+const armed: ArmedTimer[] = [];
+
+/** The first stack frame outside this file and node_modules. */
+const callerOf = (stack: string | undefined): string =>
+    (stack ?? '')
+        .split('\n')
+        .slice(1)
+        .find(
+            (line) =>
+                !line.includes('systemJobs.test') &&
+                !line.includes('node_modules'),
+        ) ?? '';
+
+beforeAll(async () => {
+    const register = vi.spyOn(JobScheduler.prototype, 'register');
+    const realInterval = globalThis.setInterval;
+    const realTimeout = globalThis.setTimeout;
+    const intervals = vi.spyOn(globalThis, 'setInterval').mockImplementation(((
+        ...args: Parameters<typeof setInterval>
+    ) => {
+        armed.push({
+            kind: 'interval',
+            ms: args[1],
+            caller: callerOf(new Error().stack),
+        });
+        return realInterval(...args);
+    }) as never);
+    const timeouts = vi.spyOn(globalThis, 'setTimeout').mockImplementation(((
+        ...args: Parameters<typeof setTimeout>
+    ) => {
+        armed.push({
+            kind: 'timeout',
+            ms: args[1],
+            caller: callerOf(new Error().stack),
+        });
+        return realTimeout(...args);
+    }) as never);
+    try {
+        server = await setupTestServer({
+            events: { enabled: true },
+            notificationRetentionDays: 14,
+            maxGlobalUsagePerMinute: 1e15,
+            // A self-hosted node: nothing turns its jobs off.
+            jobs: { enabled: true },
+        } as IConfig);
+    } finally {
+        intervals.mockRestore();
+        timeouts.mockRestore();
+    }
+    for (const [name, definition] of register.mock.calls)
+        registered.set(name, definition);
+    register.mockRestore();
+});
+
+afterAll(async () => {
+    await server?.shutdown();
+});
+
+const keysOf = (namespace: string, name: string) => ({
+    lock: `jobs:{${namespace}:${name}}:lock`,
+    gate: `jobs:{${namespace}:${name}}:next`,
+    run: `jobs:{${namespace}:${name}}:run`,
+});
+
+/** A node of its own in `region`, on the test server's Redis. */
+const node = (namespace: string, region: string): JobScheduler =>
+    new JobScheduler(
+        {
+            domain: 'puter.test',
+            serverId: `node-${region}`,
+            jobs: { namespace },
+        } as IConfig,
+        server.clients as SchedulerArgs[1],
+        server.stores as SchedulerArgs[2],
+        {
+            broadcast: { regionId: region },
+        } as unknown as SchedulerArgs[3],
+    );
+
+describe('system jobs', () => {
+    it('registers every moved job with its real cadence, scope and lease', () => {
+        expect(
+            Object.fromEntries(
+                [...registered].map(([name, job]) => [
+                    name,
+                    {
+                        every: job.every,
+                        scope: job.scope,
+                        nearPrimary: job.nearPrimary ?? false,
+                        leaseMs: job.leaseMs,
+                    },
+                ]),
+            ),
+        ).toEqual(EXPECTED_JOBS);
+    });
+
+    it('leaves no per-node timer behind for a moved job', () => {
+        const armedFrom = (file: string) =>
+            armed
+                .filter(({ caller }) => caller.includes(file))
+                .map(({ kind, ms }) => ({ kind, ms }));
+
+        expect(armedFrom('services/auth/AuthService')).toEqual([]);
+        expect(armedFrom('services/notification/NotificationService')).toEqual(
+            [],
+        );
+        // Node-local: the usage buffer flush.
+        expect(armedFrom('services/metering/MeteringService')).toEqual([
+            { kind: 'interval', ms: 10_000 },
+        ]);
+        // Node-local: the share-job runner (DB claims per job), and its kick.
+        expect(armedFrom('services/kvShare/KvShareService')).toEqual([
+            { kind: 'timeout', ms: expect.any(Number) },
+            { kind: 'interval', ms: MINUTE },
+        ]);
+        // Node-local: the region's pending-delivery sweep.
+        expect(armedFrom('services/events/EventsService')).toEqual([
+            { kind: 'interval', ms: 10_000 },
+        ]);
+    });
+
+    it.each(JOB_NAMES)(
+        'runs %s once across three nodes sharing Redis',
+        async (name) => {
+            const namespace = `ns-${randomUUID()}`;
+            const nodes = ['n1', 'n2', 'n3'].map((id) => node(namespace, id));
+            const job = registered.get(name)!;
+            let runs = 0;
+            for (const scheduler of nodes)
+                scheduler.register(name, {
+                    ...job,
+                    run: async (ctx: JobRunContext) => {
+                        runs += 1;
+                        await job.run(ctx);
+                    },
+                });
+
+            await Promise.all(nodes.map((scheduler) => scheduler.poll()));
+            await Promise.all(nodes.map((scheduler) => scheduler.poll()));
+
+            expect(runs).toBe(1);
+            const redis = server.clients.redis;
+            const keys = keysOf(namespace, name);
+            // A failure would have cut the gate to a 2 s backoff.
+            expect(await redis.pttl(keys.gate)).toBeGreaterThan(
+                job.every - MINUTE,
+            );
+            expect(await redis.hget(keys.run, 'failures')).toBeNull();
+        },
+    );
+
+    it('runs every job on a single self-hosted node, without the old per-job gates', async () => {
+        const jobs = server.services.jobs;
+        const redis = server.clients.redis;
+
+        await jobs.poll();
+
+        // The node's own poll timer may have started some of them first.
+        await vi.waitFor(
+            async () => {
+                for (const name of JOB_NAMES) {
+                    const keys = keysOf(jobs.namespace, name);
+                    expect(await redis.exists(keys.gate), name).toBe(1);
+                    // Released once the run is over.
+                    expect(await redis.exists(keys.lock), name).toBe(0);
+                    expect(await redis.hget(keys.run, 'failures'), name).toBe(
+                        null,
+                    );
+                }
+            },
+            { timeout: 10_000, interval: 100 },
+        );
+        await expect(
+            redis.exists('sessions:purge:gate', 'kvShare:purge:gate'),
+        ).resolves.toBe(0);
+    });
+});

src/backend/services/kvShare/KvShareService.ts

@@ -371,11 +371,9 @@ const SHARE_CHANGE_LOCK_RETRY_MS = 50;
 
 /** Retired handles are kept this long for their owner's listing, then deleted. */
 export const KV_SHARE_REVOKED_RETENTION_SECONDS = 30 * 24 * 60 * 60;
-// Every node checks hourly; the gate lets one node through per day.
-const PURGE_CHECK_MS = 60 * 60 * 1000;
-const PURGE_INITIAL_DELAY_MS = 5 * 60 * 1000;
-const PURGE_GATE_KEY = 'kvShare:purge:gate';
-const PURGE_GATE_SECONDS = 24 * 60 * 60;
+const PURGE_JOB = 'kvShare.handles.purgeRevoked';
+const PURGE_EVERY_MS = 24 * 60 * 60 * 1000;
+const PURGE_LEASE_MS = 15 * 60 * 1000;
 // Rows per delete, with a pause between, so replicas keep up.
 const PURGE_BATCH = 500;
 const PURGE_PAUSE_MS = 50;
@@ -536,7 +534,17 @@ export class KvShareService extends PuterService {
     #jobCursor: string | undefined;
 
     override onServerStart(): void {
-        this.#armPurge();
+        this.services.jobs.register(PURGE_JOB, {
+            every: PURGE_EVERY_MS,
+            scope: 'fleet',
+            nearPrimary: true,
+            leaseMs: PURGE_LEASE_MS,
+            run: async ({ signal }) => {
+                const purged = await this.purgeRevokedHandles(signal);
+                if (purged)
+                    console.log(`[kvShare] purged ${purged} retired handles`);
+            },
+        });
         this.#armJobs();
 
         // A check at one level is answered by a grant at any level above it.
@@ -607,51 +615,24 @@ export class KvShareService extends PuterService {
     // -- Retention purge ---------------------------------------------
 
     /**
-     * Delete handles retired longer ago than the retention, in batches. Returns
-     * the rows deleted, or null when another node holds today's gate. A missing
-     * row reads like a retired one everywhere a handle is resolved.
+     * Delete handles retired longer ago than the retention, in batches, until
+     * none are left or `signal` aborts. Returns the rows deleted. A missing row
+     * reads like a retired one everywhere a handle is resolved.
      */
-    async purgeRevokedHandles(): Promise<number | null> {
-        const claimed = await this.clients.redis.set(
-            PURGE_GATE_KEY,
-            String(Date.now()),
-            'EX',
-            PURGE_GATE_SECONDS,
-            'NX',
-        );
-        if (claimed !== 'OK') return null;
-
+    async purgeRevokedHandles(signal: AbortSignal): Promise<number> {
         const cutoff =
             Math.floor(Date.now() / 1000) - KV_SHARE_REVOKED_RETENTION_SECONDS;
         let purged = 0;
-        let finished = false;
-        try {
-            for (let i = 0; i < PURGE_MAX_BATCHES; i++) {
-                if (this.#stopping) return purged;
-                const batch =
-                    await this.stores.kvShareHandle.deleteRevokedBefore(
-                        cutoff,
-                        PURGE_BATCH,
-                    );
-                purged += batch;
-                if (batch < PURGE_BATCH) break;
-                await new Promise((resolve) =>
-                    setTimeout(resolve, PURGE_PAUSE_MS),
-                );
-            }
-            finished = true;
-            return purged;
-        } finally {
-            // A run cut short frees the gate, so another node picks it up on
-            // its next check instead of tomorrow.
-            if (!finished) {
-                try {
-                    await this.clients.redis.del(PURGE_GATE_KEY);
-                } catch {
-                    // The gate expires on its own.
-                }
-            }
+        for (let i = 0; i < PURGE_MAX_BATCHES && !signal.aborted; i++) {
+            const batch = await this.stores.kvShareHandle.deleteRevokedBefore(
+                cutoff,
+                PURGE_BATCH,
+            );
+            purged += batch;
+            if (batch < PURGE_BATCH) break;
+            await new Promise((resolve) => setTimeout(resolve, PURGE_PAUSE_MS));
         }
+        return purged;
     }
 
     /**
@@ -1047,27 +1028,6 @@ export class KvShareService extends PuterService {
         this.#timers.push(kick, timer);
     }
 
-    #armPurge(): void {
-        const run = async () => {
-            try {
-                const purged = await this.purgeRevokedHandles();
-                if (purged)
-                    console.log(`[kvShare] purged ${purged} retired handles`);
-            } catch (err) {
-                console.warn('[kvShare] retired-handle purge failed', err);
-            }
-        };
-        // Jittered so a deploy does not have every node check at once.
-        const kick = setTimeout(
-            () => void run(),
-            PURGE_INITIAL_DELAY_MS * (0.5 + Math.random()),
-        );
-        kick.unref?.();
-        const timer = setInterval(() => void run(), PURGE_CHECK_MS);
-        timer.unref?.();
-        this.#timers.push(kick, timer);
-    }
-
     /** Whether the level a mint asks for is available on this server. */
     #assertAccessEnabled(access: KvShareAccess): void {
         if (access === 'watch') {

src/backend/services/metering/MeteringService.rateCheck.test.ts

@@ -125,16 +125,16 @@ describe('MeteringService.checkRateOfChange', () => {
         });
     });
 
-    it('skips the shard read when the baseline is only 5 minutes old', async () => {
+    it('measures against a baseline of any age, since the job runs it once per cadence', async () => {
         const { svc, kv, alarmCreate } = makeService(
-            { maxGlobalUsagePerMinute: 100 },
+            { maxGlobalUsagePerMinute: 1e9 },
             { total: 0, timestamp: Date.now() - 5 * 60_000 },
         );
 
         await svc.checkRateOfChange();
 
-        expect(kv.get).toHaveBeenCalledTimes(1);
-        expect(kv.set).not.toHaveBeenCalled();
+        expect(kv.get).toHaveBeenCalledTimes(2);
+        expect(kv.set).toHaveBeenCalledTimes(1);
         expect(alarmCreate).not.toHaveBeenCalled();
     });
 });

src/backend/services/metering/MeteringService.ts

@@ -111,6 +111,11 @@ type UsageWriteResult = {
     detail?: FlatUsageDetail;
 };
 
+/** Compares fleet-wide spend against `maxGlobalUsagePerMinute`. */
+const GLOBAL_RATE_CHECK_JOB = 'metering.globalRateCheck';
+const GLOBAL_RATE_CHECK_EVERY_MS = 25 * 60 * 1000;
+const GLOBAL_RATE_CHECK_LEASE_MS = 5 * 60 * 1000;
+
 // -- Helpers ----------------------------------------------------------
 
 /**
@@ -284,7 +289,6 @@ export class MeteringService extends PuterService {
      */
     static APP_TOTALS_LIST_LIMIT = 4_000;
 
-    private rateCheckTimer: ReturnType<typeof setInterval> | null = null;
     private usageBufferTimer: ReturnType<typeof setInterval> | null = null;
     private extraPolicies: SubscriptionPolicy[] = [];
     private subscriptionResolvers: SubscriptionResolver[] = [];
@@ -415,15 +419,15 @@ export class MeteringService extends PuterService {
             },
         );
 
-        this.rateCheckTimer = setInterval(
-            () => {
-                this.checkRateOfChange().catch((e) => {
-                    console.error('[metering] rate-of-change check failed', e);
-                });
-            },
-            1000 * 60 * 25,
-        );
-        this.rateCheckTimer.unref?.();
+        const maxPerMinute = this.config.maxGlobalUsagePerMinute;
+        if (maxPerMinute && maxPerMinute > 0) {
+            this.services.jobs.register(GLOBAL_RATE_CHECK_JOB, {
+                every: GLOBAL_RATE_CHECK_EVERY_MS,
+                scope: 'fleet',
+                leaseMs: GLOBAL_RATE_CHECK_LEASE_MS,
+                run: () => this.checkRateOfChange(),
+            });
+        }
 
         const flushInterval =
             this.config.meteringUsageBufferFlushMs &&
@@ -454,10 +458,6 @@ export class MeteringService extends PuterService {
     }
 
     override async onServerShutdown(): Promise<void> {
-        if (this.rateCheckTimer) {
-            clearInterval(this.rateCheckTimer);
-            this.rateCheckTimer = null;
-        }
         if (this.usageBufferTimer) {
             clearInterval(this.usageBufferTimer);
             this.usageBufferTimer = null;
@@ -2948,8 +2948,6 @@ export class MeteringService extends PuterService {
             timestamp: number;
         } | null;
 
-        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

@@ -895,6 +895,31 @@ describe('NotificationService.sweepExpired', () => {
         expect(await notifications.sweepExpired()).toBe(50);
     });
 
+    it('stops between batches once its signal aborts', async () => {
+        await notifications.sweepExpired();
+        const user = await makeUser();
+        await seedExpired(user.id, 1200);
+        const controller = new AbortController();
+        const deleteBatch = server.stores.notification.deleteCreatedBefore.bind(
+            server.stores.notification,
+        );
+        const spy = vi
+            .spyOn(server.stores.notification, 'deleteCreatedBefore')
+            .mockImplementation(async (days: number, limit: number) => {
+                const removed = await deleteBatch(days, limit);
+                controller.abort();
+                return removed;
+            });
+        try {
+            expect(await notifications.sweepExpired(controller.signal)).toBe(
+                500,
+            );
+        } finally {
+            spy.mockRestore();
+        }
+        expect(await notifications.sweepExpired()).toBe(700);
+    });
+
     it('sweeps nothing when no retention is configured', async () => {
         const unbounded = await setupTestServer({
             notificationRetentionDays: 0,

src/backend/services/notification/NotificationService.ts

@@ -48,8 +48,9 @@ export type {
 } from './notificationTypes.js';
 export { canViewNotification } from './notificationAudience.js';
 
-/** How often the retention sweep runs. */
-const RETENTION_SWEEP_INTERVAL_MS = 60 * 60 * 1000;
+const RETENTION_SWEEP_JOB = 'notifications.retention';
+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. */
@@ -69,11 +70,20 @@ const UNREAD_REPLAY_LIMIT = 200;
 export class NotificationService extends PuterService {
     /** User.id → debounce timeout */
     #connectTimeouts = new Map<number, ReturnType<typeof setTimeout>>();
-    #retentionSweep: ReturnType<typeof setInterval> | null = null;
     #adapter: NotificationSocketAdapter | null = null;
 
     override onServerStart(): void {
-        this.#armRetentionSweep();
+        if (this.#retentionDays() > 0) {
+            this.services.jobs.register(RETENTION_SWEEP_JOB, {
+                every: RETENTION_SWEEP_EVERY_MS,
+                scope: 'fleet',
+                nearPrimary: true,
+                leaseMs: RETENTION_SWEEP_LEASE_MS,
+                run: async ({ signal }) => {
+                    await this.sweepExpired(signal);
+                },
+            });
+        }
         this.#socket().attach();
 
         // When a user opens the GUI, send their pending unreads.
@@ -103,11 +113,6 @@ export class NotificationService extends PuterService {
         );
     }
 
-    override onServerPrepareShutdown(): void {
-        if (this.#retentionSweep) clearInterval(this.#retentionSweep);
-        this.#retentionSweep = null;
-    }
-
     // -- Public API --------------------------------------------------
 
     /**
@@ -348,17 +353,15 @@ export class NotificationService extends PuterService {
      * Drop notifications past the retention window, in batches, and report how
      * many went. Deleting is all there is to it: nothing is pushed, because a
      * two-week-old row is not news, and a client listing again simply stops
-     * seeing it.
-     *
-     * Every node sweeps. Batches are small and the delete is idempotent, so two
-     * nodes overlapping costs a few empty batches, not correctness.
+     * seeing it. Stops between batches once `signal` aborts.
      */
-    async sweepExpired(): Promise<number> {
+    async sweepExpired(signal?: AbortSignal): Promise<number> {
         const days = this.#retentionDays();
         if (days <= 0) return 0;
 
         let removed = 0;
         for (let pass = 0; pass < RETENTION_MAX_BATCHES; pass++) {
+            if (signal?.aborted) break;
             const batch = await this.stores.notification.deleteCreatedBefore(
                 days,
                 RETENTION_BATCH_SIZE,
@@ -470,17 +473,6 @@ export class NotificationService extends PuterService {
         return Number.isFinite(configured) && configured > 0 ? configured : 0;
     }
 
-    #armRetentionSweep(): void {
-        if (this.#retentionDays() <= 0) return;
-        const sweep = setInterval(() => {
-            void this.sweepExpired().catch((err) => {
-                console.warn('[notification] retention sweep failed', err);
-            });
-        }, RETENTION_SWEEP_INTERVAL_MS);
-        sweep.unref?.();
-        this.#retentionSweep = sweep;
-    }
-
     /**
      * What a reconnecting client missed. One statement marks the batch shown —
      * a per-row update is a round trip per notification on a path that runs

src/backend/stores/events/DurableSubscriptionStore.integration.test.ts

@@ -565,7 +565,9 @@ describe('rows suspended for good', () => {
             },
         );
 
-        await durable().create(input());
+        const created = await durable().create(input());
+        // The trim's bump, for the caller to publish to the other regions.
+        expect(created.reaped.map((bump) => bump.userId)).toEqual([userId]);
 
         await expect(durable().countForHolder(userId)).resolves.toEqual({
             total: 1,
@@ -744,7 +746,7 @@ describe('the expiry sweep', () => {
         // one back to prove the sweep is what removes it.
         await cache().cacheDurable([expired.row]);
 
-        await expect(durable().sweepExpired(500)).resolves.toBe(1);
+        expect((await durable().sweepExpired(500)).rows).toHaveLength(1);
 
         await expect(durable().getBySubId(expired.row.subId)).resolves.toBeNull();
         await expect(cache().getForTokens(userId, [token()])).resolves.toEqual([
@@ -754,7 +756,7 @@ describe('the expiry sweep', () => {
 
     it('leaves a subscription with no expiry alone', async () => {
         await durable().create(input());
-        await expect(durable().sweepExpired(500)).resolves.toBe(0);
+        expect((await durable().sweepExpired(500)).rows).toHaveLength(0);
         await expect(durable().countForHolder(userId)).resolves.toMatchObject({
             total: 1,
         });
@@ -808,15 +810,17 @@ describe('reapForApp', () => {
         const untouched = await durable().create(input());
         const before = await cache().getGeneration(userId);
 
-        await expect(durable().reapForApp(appUid, 500)).resolves.toBe(1);
+        const reaped = await durable().reapForApp(appUid, 500);
+        expect(reaped.rows.map((row) => row.subId)).toEqual([mine.row.subId]);
 
         await expect(durable().getBySubId(mine.row.subId)).resolves.toBeNull();
         await expect(
             durable().getBySubId(untouched.row.subId),
         ).resolves.not.toBeNull();
-        await expect(cache().getGeneration(userId)).resolves.toBeGreaterThan(
-            before,
-        );
+        const after = await cache().getGeneration(userId);
+        expect(after).toBeGreaterThan(before);
+        // The bump the caller publishes to the other regions.
+        expect(reaped.bumps).toEqual([{ userId, generation: after }]);
     });
 
     it('never touches another app`s rows', async () => {
@@ -837,7 +841,7 @@ describe('reapForApp', () => {
     it('answers 0 for an app with nothing to reap', async () => {
         await expect(
             durable().reapForApp(`app-${uuidv4()}`, 500),
-        ).resolves.toBe(0);
+        ).resolves.toEqual({ rows: [], bumps: [] });
     });
 });
 

src/backend/stores/events/DurableSubscriptionStore.ts

@@ -207,6 +207,22 @@ const quotaRefusal = (
     return null;
 };
 
+/**
+ * What a reap deleted, and the generation it bumped per owner. The caller
+ * publishes the bumps, since the reap only invalidates this region.
+ */
+export interface ReapResult {
+    rows: DurableSubscription[];
+    bumps: GenerationBump[];
+}
+
+/** A created row; `reaped` holds the bumps of terminal rows it trimmed. */
+export interface DurableCreated {
+    row: DurableSubscription;
+    bump: GenerationBump;
+    reaped: GenerationBump[];
+}
+
 /**
  * The quota refusal for a create that concurrent creates beat to the last slot
  * after its own insert. Its row is already gone again; `bump` still has to be
@@ -292,9 +308,7 @@ export class DurableSubscriptionStore extends PuterStore {
      * returning. The generation bump is what tells every other region, and the
      * caller broadcasts it.
      */
-    async create(
-        input: DurableSubscriptionInput,
-    ): Promise<{ row: DurableSubscription; bump: GenerationBump }> {
+    async create(input: DurableSubscriptionInput): Promise<DurableCreated> {
         this.#assertContext(input.context);
         assertColumnWidths(EVENT_SUBSCRIPTION_WIDTHS, {
             token: input.token,
@@ -368,18 +382,20 @@ export class DurableSubscriptionStore extends PuterStore {
         );
         const lost = quotaRefusal(input, held, 1);
         if (lost) throw new DurableQuotaRaceLost(lost, await this.remove(row));
-        if (held.terminal > EVENTS_REVOKED_SUBSCRIPTIONS_KEPT)
-            await this.#dropOldestTerminal(
-                input.holderUserId,
-                held.terminal - EVENTS_REVOKED_SUBSCRIPTIONS_KEPT,
-            );
+        const reaped =
+            held.terminal > EVENTS_REVOKED_SUBSCRIPTIONS_KEPT
+                ? await this.#dropOldestTerminal(
+                      input.holderUserId,
+                      held.terminal - EVENTS_REVOKED_SUBSCRIPTIONS_KEPT,
+                  )
+                : [];
 
         // Write-through, and over the whole owner rather than the one row: it
         // costs the same indexed read as a warm would, on a path rate-limited
         // to a few calls a minute, and it leaves this region needing nothing
         // from the table before it can deliver.
         await this.#rebuildRegion(row.ownerUserId);
-        return { row, bump: await this.#bump(row.ownerUserId) };
+        return { row, bump: await this.#bump(row.ownerUserId), reaped };
     }
 
     /**
@@ -573,7 +589,7 @@ export class DurableSubscriptionStore extends PuterStore {
      * Reap rows past their expiry, dropping each from the cache so the region
      * stops delivering against it without waiting for a rebuild.
      */
-    async sweepExpired(batchSize: number): Promise<number> {
+    async sweepExpired(batchSize: number): Promise<ReapResult> {
         return this.#reap(await this.#listExpired(nowSeconds(), batchSize));
     }
 
@@ -582,7 +598,10 @@ export class DurableSubscriptionStore extends PuterStore {
      * never resumes is a row kept only so its holder can see why it stopped,
      * and that answer has a shelf life.
      */
-    async sweepSuspended(cutoff: number, batchSize: number): Promise<number> {
+    async sweepSuspended(
+        cutoff: number,
+        batchSize: number,
+    ): Promise<ReapResult> {
         return this.#reap(await this.#listSuspendedBefore(cutoff, batchSize));
     }
 
@@ -846,20 +865,21 @@ export class DurableSubscriptionStore extends PuterStore {
      * What app deletion reaps: `#reap` already purges the backlog, drops the
      * cache and bumps a generation per owner, so this is the whole of it.
      */
-    async reapForApp(appUid: string, batchSize: number): Promise<number> {
+    async reapForApp(appUid: string, batchSize: number): Promise<ReapResult> {
         return this.#reap(await this.#listForApp(appUid, batchSize));
     }
 
     // -- Internals ---------------------------------------------------
 
     /**
-     * Delete a holder's oldest rows suspended for good, past what `list` keeps.
-     * Best-effort: the next subscribe tries again.
+     * Delete a holder's oldest rows suspended for good, past what `list` keeps,
+     * and return the reap's bumps. Best-effort: the next subscribe tries
+     * again.
      */
     async #dropOldestTerminal(
         holderUserId: number,
         count: number,
-    ): Promise<void> {
+    ): Promise<GenerationBump[]> {
         try {
             const rows = await this.clients.db.pread(
                 `SELECT ${SELECT_COLUMNS} FROM \`${TABLE}\` ` +
@@ -867,9 +887,10 @@ export class DurableSubscriptionStore extends PuterStore {
                     'ORDER BY `suspended_at`, `id` LIMIT ?',
                 [holderUserId, ...TERMINAL_SUSPENDED_REASONS, count],
             );
-            await this.#reap(rows.map(toRow));
+            return (await this.#reap(rows.map(toRow))).bumps;
         } catch (err) {
             console.warn('[events] could not drop old revoked rows', err);
+            return [];
         }
     }
 
@@ -952,9 +973,12 @@ export class DurableSubscriptionStore extends PuterStore {
         return rows.map(toRow);
     }
 
-    /** Delete a batch and leave no region delivering against any of it. */
-    async #reap(rows: readonly DurableSubscription[]): Promise<number> {
-        if (rows.length === 0) return 0;
+    /**
+     * Delete a batch and stop this region delivering against any of it. Other
+     * regions stop once the caller publishes the returned bumps.
+     */
+    async #reap(rows: readonly DurableSubscription[]): Promise<ReapResult> {
+        if (rows.length === 0) return { rows: [], bumps: [] };
 
         await this.clients.db.write(
             `DELETE FROM \`${TABLE}\` WHERE \`sub_id\` IN ` +
@@ -965,11 +989,17 @@ export class DurableSubscriptionStore extends PuterStore {
         await this.stores.eventSubscription.dropDurables(rows);
         for (const row of rows) {
             // Whatever was still owed to a row that no longer exists.
-            await this.stores.pendingDelivery.purge(row.subId).catch(() => {});
+            try {
+                await this.stores.pendingDelivery.purge(row.subId);
+            } catch {
+                // The pending sweep purges it once it finds the row gone.
+            }
         }
+        const bumps: GenerationBump[] = [];
         const owners = new Set(rows.map((row) => row.ownerUserId));
-        for (const ownerUserId of owners) await this.#bump(ownerUserId);
-        return rows.length;
+        for (const ownerUserId of owners)
+            bumps.push(await this.#bump(ownerUserId));
+        return { rows: [...rows], bumps };
     }
 
     #assertContext(context: string | null): void {

src/backend/stores/events/KvShareHandleStore.integration.test.ts

@@ -326,7 +326,7 @@ describe('the notify flag', () => {
 });
 
 describe('live handles by owner and app', () => {
-    it('lists only the owner\'s live handles in that app, oldest first', async () => {
+    it("lists only the owner's live handles in that app, oldest first", async () => {
         const appUid = 'app-owner-listing';
         const first = await store().mint(
             input({ appUid, keyPrefix: 'owned-a:' }),
@@ -502,13 +502,11 @@ describe('the live cache across nodes', () => {
         expect(await store().getLive(row.handle)).toBeNull();
     });
 
-    it('drops an owner\'s cached rows when the owner is suspended, here and on other nodes', async () => {
+    it("drops an owner's cached rows when the owner is suspended, here and on other nodes", async () => {
         const row = await store().mint(
             input({ appUid: 'app-forget', keyPrefix: 'forget-owner:' }),
         );
-        expect((await store().getLive(row.handle))?.ownerSuspended).toBe(
-            false,
-        );
+        expect((await store().getLive(row.handle))?.ownerSuspended).toBe(false);
         const setSuspended = (value: number) =>
             env.server.clients.db.write(
                 'UPDATE `user` SET `suspended` = ? WHERE `id` = ?',
@@ -534,11 +532,7 @@ describe('the live cache across nodes', () => {
                     ([key]) => key === 'outer.pubsub.kv.share.forgetOwner',
                 ),
             ).toEqual([
-                [
-                    'outer.pubsub.kv.share.forgetOwner',
-                    { ownerUserId },
-                    {},
-                ],
+                ['outer.pubsub.kv.share.forgetOwner', { ownerUserId }, {}],
             ]);
 
             // Another node lifting the suspension reaches this one's cache.
@@ -557,13 +551,11 @@ describe('the live cache across nodes', () => {
         }
     });
 
-    it('drops an owner\'s cached rows when the user store suspends or unsuspends them', async () => {
+    it("drops an owner's cached rows when the user store suspends or unsuspends them", async () => {
         const row = await store().mint(
             input({ appUid: 'app-forget', keyPrefix: 'forget-store:' }),
         );
-        expect((await store().getLive(row.handle))?.ownerSuspended).toBe(
-            false,
-        );
+        expect((await store().getLive(row.handle))?.ownerSuspended).toBe(false);
         try {
             await env.server.stores.user.update(ownerUserId, {
                 suspended: true,
@@ -895,21 +887,27 @@ describe('retired handle retention', () => {
         expect(await exists(live)).toBe(true);
     });
 
-    it('purges once per day across nodes, keeping live and recent rows', async () => {
+    it('purges retired handles past the retention, keeping live and recent rows', async () => {
         const kvShare = env.server.services.kvShare;
-        await env.server.clients.redis.del('kvShare:purge:gate');
         const old = await retiredAt('purge-old:', now() - 31 * DAY);
         const recent = await retiredAt('purge-recent:', now() - 29 * DAY);
         const live = await retiredAt('purge-live:', null);
 
-        expect(await kvShare.purgeRevokedHandles()).toBeGreaterThanOrEqual(1);
+        expect(
+            await kvShare.purgeRevokedHandles(new AbortController().signal),
+        ).toBeGreaterThanOrEqual(1);
         expect(await exists(old)).toBe(false);
         expect(await exists(recent)).toBe(true);
         expect(await exists(live)).toBe(true);
+    });
 
-        // Another node checking the same day finds the gate held.
-        const again = await retiredAt('purge-again:', now() - 31 * DAY);
-        expect(await kvShare.purgeRevokedHandles()).toBeNull();
-        expect(await exists(again)).toBe(true);
+    it('stops purging between batches once its signal aborts', async () => {
+        const kvShare = env.server.services.kvShare;
+        const old = await retiredAt('purge-aborted:', now() - 31 * DAY);
+        const controller = new AbortController();
+        controller.abort();
+
+        expect(await kvShare.purgeRevokedHandles(controller.signal)).toBe(0);
+        expect(await exists(old)).toBe(true);
     });
 });