diff --git a/src/backend/server.test.ts b/src/backend/server.test.ts index 4e12bf10f..6a411a6d1 100644 --- a/src/backend/server.test.ts +++ b/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).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, diff --git a/src/backend/services/auth/AuthService.test.ts b/src/backend/services/auth/AuthService.test.ts index 87aba46dc..da50b1418 100644 --- a/src/backend/services/auth/AuthService.test.ts +++ b/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(); } }); }); diff --git a/src/backend/services/auth/AuthService.ts b/src/backend/services/auth/AuthService.ts index f747005e4..6a7af36db 100644 --- a/src/backend/services/auth/AuthService.ts +++ b/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; - #sessionPurge: ReturnType | 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::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 { - 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 { 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 { - 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 -------------------------------------------------- diff --git a/src/backend/services/events/EventsService.test.ts b/src/backend/services/events/EventsService.test.ts index 18f9f5b5c..ae1aef507 100644 --- a/src/backend/services/events/EventsService.test.ts +++ b/src/backend/services/events/EventsService.test.ts @@ -293,12 +293,14 @@ const buildService = ( metered: MeteredLine[]; forwarded: ForwardedCall[]; eventBus: { on: ReturnType; emit: ReturnType }; + jobs: { register: ReturnType }; } => { 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 }; + 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 }, diff --git a/src/backend/services/events/EventsService.ts b/src/backend/services/events/EventsService.ts index 270ad9e30..c33a944ea 100644 --- a/src/backend/services/events/EventsService.ts +++ b/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 = { // -- 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(); /** Paces settles started by deliveries that found consent withdrawn. */ readonly #consentSettles = new ConsentSettles(); - #expirySweep: ReturnType | null = null; - #expiryKick: ReturnType | null = null; #pendingSweep: ReturnType | null = null; - #creditSweep: ReturnType | 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 { + /** Drop rows past their expiry, in batches, and report how many went. */ + async sweepExpired(signal?: AbortSignal): Promise { 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 { + async sweepSuspended(signal?: AbortSignal): Promise { 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 { + async sweepStrandedWorkerSessions({ + signal, + state, + saveState, + }: JobRunContext): Promise { 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 { + async sweepNoCredit({ + signal, + state, + saveState, + }: JobRunContext): Promise { 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, + pass: (batchSize: number) => Promise, + signal?: AbortSignal, ): Promise { 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 { diff --git a/src/backend/services/events/appDeletion.integration.test.ts b/src/backend/services/events/appDeletion.integration.test.ts index 9df4b484f..1d4117ca9 100644 --- a/src/backend/services/events/appDeletion.integration.test.ts +++ b/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(); + }); }); diff --git a/src/backend/services/events/durable.integration.test.ts b/src/backend/services/events/durable.integration.test.ts index f99f10e79..d3cfedd71 100644 --- a/src/backend/services/events/durable.integration.test.ts +++ b/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> = []; + const listener = (_key: string, data: unknown) => { + heard.push(data as Record); + }; + 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', () => { diff --git a/src/backend/services/events/forwarding.test.ts b/src/backend/services/events/forwarding.test.ts index bdc0ac649..9a3bba751 100644 --- a/src/backend/services/events/forwarding.test.ts +++ b/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: [], }; }, }, diff --git a/src/backend/services/events/metering.test.ts b/src/backend/services/events/metering.test.ts index 56d61c239..41f491192 100644 --- a/src/backend/services/events/metering.test.ts +++ b/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; /** 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([]); }); }); diff --git a/src/backend/services/jobs/systemJobs.test.ts b/src/backend/services/jobs/systemJobs.test.ts new file mode 100644 index 000000000..43c1381b1 --- /dev/null +++ b/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 . + */ + +/** + * 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; + +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(); +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 + ) => { + 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 + ) => { + 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); + }); +}); diff --git a/src/backend/services/kvShare/KvShareService.ts b/src/backend/services/kvShare/KvShareService.ts index 0d4ac8cac..1daaec39b 100644 --- a/src/backend/services/kvShare/KvShareService.ts +++ b/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 { - 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 { 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') { diff --git a/src/backend/services/metering/MeteringService.rateCheck.test.ts b/src/backend/services/metering/MeteringService.rateCheck.test.ts index c8a1276dc..b9328daf8 100644 --- a/src/backend/services/metering/MeteringService.rateCheck.test.ts +++ b/src/backend/services/metering/MeteringService.rateCheck.test.ts @@ -125,16 +125,16 @@ describe('MeteringService.checkRateOfChange', () => { }); }); - it('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(); }); }); diff --git a/src/backend/services/metering/MeteringService.ts b/src/backend/services/metering/MeteringService.ts index 130f3b9b8..c86c6fe37 100644 --- a/src/backend/services/metering/MeteringService.ts +++ b/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 | null = null; private usageBufferTimer: ReturnType | 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 { - 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; diff --git a/src/backend/services/notification/NotificationService.test.ts b/src/backend/services/notification/NotificationService.test.ts index dd3f48e71..8c753dd91 100644 --- a/src/backend/services/notification/NotificationService.test.ts +++ b/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, diff --git a/src/backend/services/notification/NotificationService.ts b/src/backend/services/notification/NotificationService.ts index b7ff22f27..4174debb9 100644 --- a/src/backend/services/notification/NotificationService.ts +++ b/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>(); - #retentionSweep: ReturnType | 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 { + async sweepExpired(signal?: AbortSignal): Promise { 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 diff --git a/src/backend/stores/events/DurableSubscriptionStore.integration.test.ts b/src/backend/stores/events/DurableSubscriptionStore.integration.test.ts index 62cbe7f84..b53d406d6 100644 --- a/src/backend/stores/events/DurableSubscriptionStore.integration.test.ts +++ b/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: [] }); }); }); diff --git a/src/backend/stores/events/DurableSubscriptionStore.ts b/src/backend/stores/events/DurableSubscriptionStore.ts index a5c52df19..0d974a81d 100644 --- a/src/backend/stores/events/DurableSubscriptionStore.ts +++ b/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 { 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 { + async sweepExpired(batchSize: number): Promise { 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 { + async sweepSuspended( + cutoff: number, + batchSize: number, + ): Promise { 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 { + async reapForApp(appUid: string, batchSize: number): Promise { 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 { + ): Promise { 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 { - 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 { + 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 { diff --git a/src/backend/stores/events/KvShareHandleStore.integration.test.ts b/src/backend/stores/events/KvShareHandleStore.integration.test.ts index a67e66962..0e2f5fbf3 100644 --- a/src/backend/stores/events/KvShareHandleStore.integration.test.ts +++ b/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); }); });