From fc534abb7b78aac821557db16fce960d6932e675 Mon Sep 17 00:00:00 2001 From: Bouynaty Date: Fri, 31 Jul 2026 15:28:02 +0100 Subject: [PATCH] feat: harden session compaction job with retention policy Refine the nightly session compaction job so the retention boundary is computed inside the database (NOW() - retention interval) rather than passed in from the application-server clock. A bad-clock event on the app server can no longer push the boundary forward and cause active rows to be deleted. Add a per-run row cap (SESSION_COMPACTION_MAX_ROWS_PER_RUN, default 100k) so an anomaly can never wipe the table in one cycle; emit session.compaction.cap_hit when the cap is reached. Validate batch and cap inputs, and log scheduled-run failures without killing the timer. Tests cover batched deletion, vacuum behavior, error metrics, bad-clock safety, cap enforcement, input validation, and the start/stop lifecycle (100% statement coverage on the service). Closes #683 --- docs/session-storage-compaction.md | 146 +++++++++++ src/config/env.ts | 2 + src/db/repositories/sessionRepository.test.ts | 83 +++++++ src/db/repositories/sessionRepository.ts | 34 ++- src/services/sessionCompactionService.test.ts | 227 +++++++++++++++--- src/services/sessionCompactionService.ts | 128 +++++++--- 6 files changed, 554 insertions(+), 66 deletions(-) create mode 100644 docs/session-storage-compaction.md diff --git a/docs/session-storage-compaction.md b/docs/session-storage-compaction.md new file mode 100644 index 00000000..5a68327a --- /dev/null +++ b/docs/session-storage-compaction.md @@ -0,0 +1,146 @@ +# Session Storage Compaction Job (#683) + +## Problem + +Revoked and expired session rows accumulate indefinitely in the `sessions` +table. Every expired browser session, every revoked token, and every stale +`deleteExpired()` candidate leaves a row behind. Over time these historical rows +bloat the table and the backups taken from it, slowing maintenance operations +and inflating storage cost. Unlike `deleteExpired()`, which only removes rows +whose `expires_at` has already passed, the system never cleaned rows that were +revoked long ago but whose original expiry is still far in the future. + +## Solution + +A nightly `SessionCompactionService` job deletes revoked and expired session +rows older than the configured retention window (default **30 days**), then +vacuum the table to reclaim space. + +``` +┌───────────────────┐ every 24h ┌─────────────────────────────────────────┐ +│ Scheduler │ ─────────────►│ SessionCompactionService.runCompaction() │ +│ (setInterval) │ └─────────────────────────────────────────┘ +└───────────────────┘ │ + ┌──────▼──────┐ + │ 1. lag │ getOldestCompactedSessionDate() + │ 2. purge │ purgeOlderThan() in bounded batches + │ 3. vacuum │ vacuumSessions() + └─────────────┘ +``` + +### Files + +| File | Role | +|---|---| +| `src/services/sessionCompactionService.ts` | Scheduled job: batches, cap, metrics | +| `src/db/repositories/sessionRepository.ts` | `purgeOlderThan`, `getOldestCompactedSessionDate`, `vacuumSessions` | +| `src/index.ts` | Wires the service into the app lifecycle | +| `src/config/env.ts` | `SESSION_RETENTION_DAYS`, `SESSION_COMPACTION_MAX_ROWS_PER_RUN` | + +--- + +## How it works + +1. **Boundary** — the retention boundary is `NOW() - SESSION_RETENTION_DAYS`, + computed **inside the database**. The service never passes a server-computed + timestamp to the purge query (see Security below). +2. **Lag** — before deleting, the service asks the repository for the oldest + eligible row and records how far past the retention boundary it sits. +3. **Bounded batches** — rows are deleted with `DELETE ... WHERE id IN + (SELECT id ... LIMIT $batchSize)`, so no single statement holds locks on + more than `batchSize` (default 1000) rows. +4. **Per-run cap** — a single cycle stops after deleting + `SESSION_COMPACTION_MAX_ROWS_PER_RUN` (default 100 000) rows. If the cap is + hit, the run emits `session.compaction.cap_hit` and leaves the remainder for + the next cycle. +5. **Vacuum** — if any rows were deleted, `VACUUM sessions` runs on its own + connection (VACUUM cannot run inside a transaction block) to reclaim space. + +### Configuration + +| Env var | Default | Description | +|---|---|---| +| `SESSION_RETENTION_DAYS` | `30` | How many days an expired/revoked row is retained before it becomes eligible for deletion | +| `SESSION_COMPACTION_MAX_ROWS_PER_RUN` | `100000` | Hard cap on rows deleted per cycle | + +--- + +## Metrics + +Emitted via `MetricsCollector` when the service is constructed with one: + +| Metric | Type | Labels | Meaning | +|---|---|---|---| +| `session.compaction.rows` | counter | `status=success` | Total rows deleted in the run | +| `session.compaction.retention_lag_days` | histogram | `status=success` | Lag of the oldest eligible row behind the retention boundary | +| `session.compaction.duration_ms` | histogram | `status=success\|error` | Run duration | +| `session.compaction.errors_total` | counter | `status=error` | Failed runs | +| `session.compaction.cap_hit` | counter | `status=warning` | Run stopped because it hit the per-run row cap | + +--- + +## Security Assumptions + +1. **The retention boundary is always the database clock, never the app clock.** + The purge query computes `NOW() - INTERVAL` inside Postgres. If the + application-server clock jumps forward (a bad-clock event from NTP failure, + host migration, or manual change), the boundary does not move, so the job + cannot be tricked into deleting rows that have not actually aged past + retention. +2. **Active rows can never match the predicate.** A session is only deleted + when `expires_at < boundary` **or** `revoked_at < boundary`. An active + session has a future `expires_at` and no `revoked_at`, so it satisfies + neither branch. `deleteExpired()` (active-row sweep) and the compaction job + are therefore safe to run concurrently. +3. **Blast radius is bounded.** Even in a pathological case where an enormous + number of rows suddenly become eligible, the per-run cap stops the cycle at + a fixed, recoverable number of rows and raises `session.compaction.cap_hit` + so operators can investigate before the next nightly run. +4. **No lock amplification.** Bounded `LIMIT` batches keep individual DELETE + statements short, avoiding long-held locks on the `sessions` table that + could stall login/touch traffic. +5. **VACUUM is executed standalone.** It runs on its own connection, never + inside a transaction block, which is a hard Postgres requirement. + +### Failure / abuse paths + +| Scenario | Behaviour | +|---|---| +| App clock jumps forward | Boundary is DB-computed → no extra rows become eligible; cap still protects | +| App clock jumps backward | Boundary moves backward → fewer rows eligible; safe | +| Huge backlog of eligible rows | Run hits `maxRowsPerRun`, emits `cap_hit`, resumes next cycle | +| DB unavailable | Run fails fast, emits `session.compaction.errors_total`, keeps the schedule alive | +| Duplicate concurrent runs | Both runs delete disjoint bounded batches idempotently; VACUUM is safe to run concurrently | +| `batchSize` / `maxRowsPerRun` misconfigured (≤ 0 or non-integer) | `runCompaction` throws before touching the DB | + +--- + +## Tests + +```bash +npx jest src/services/sessionCompactionService.test.ts src/db/repositories/sessionRepository.test.ts +``` + +Coverage highlights: + +- Batched deletion stops on a partial batch and vacuums exactly once. +- No vacuum when nothing was deleted. +- Error paths emit `session.compaction.errors_total`. +- **Bad-clock safety** — the service passes retention *days* to the repository, + never a server-derived `Date`; the repository SQL proves the boundary is + `NOW() - interval` and that only `expires_at`/`revoked_at` are matched. +- **Cap enforcement** — a saturated table stops at the cap, truncates the last + batch, and emits `session.compaction.cap_hit`; a draining table does not. +- Input validation rejects non-positive / non-integer batch and cap values. +- `start()`/`stop()` scheduling lifecycle. + +--- + +## Related Files + +| File | Role | +|---|---| +| `src/services/auditPurgeService.ts` | Parallel scheduled purge job for audit logs | +| `src/db/repositories/sessionRepository.ts` | Session storage repository | +| `src/lib/sessionStore.ts` | `PostgresSessionStore` (lazy expiry + sweep) | +| `docs/session-storage-partial-index.md` | Query-performance baseline for the same table | diff --git a/src/config/env.ts b/src/config/env.ts index 616b7625..7a6c98d1 100644 --- a/src/config/env.ts +++ b/src/config/env.ts @@ -26,6 +26,7 @@ import { z } from "zod"; * | ALLOWED_ORIGINS | No | localhost:3000 | Comma-separated list of allowed CORS origins | * | AUDIT_RETENTION_DAYS | No | 90 | Number of days to retain audit logs | * | SESSION_RETENTION_DAYS | No | 30 | Number of days to retain expired/revoked sessions| + * | SESSION_COMPACTION_MAX_ROWS_PER_RUN | No | 100000 | Max rows a single compaction run may delete; caps blast radius on anomalies| * | EMAIL_PROVIDER | No | mock/sendgrid | Email provider: sendgrid, smtp, or mock | * | FROM_EMAIL | No | noreply@revora.com | Default sender address for transactional email | * | SENDGRID_API_KEY | SendGrid | (empty) | SendGrid API key | @@ -74,6 +75,7 @@ const envSchema = z.object({ ALLOWED_ORIGINS: z.string().optional(), AUDIT_RETENTION_DAYS: z.coerce.number().int().positive().default(90), SESSION_RETENTION_DAYS: z.coerce.number().int().positive().default(30), + SESSION_COMPACTION_MAX_ROWS_PER_RUN: z.coerce.number().int().positive().default(100000), EMAIL_PROVIDER: z.enum(["sendgrid", "smtp", "mock"]).optional(), FROM_EMAIL: z.string().email().optional(), CURSOR_SIGNING_SECRET: z.string().min(16).optional(), diff --git a/src/db/repositories/sessionRepository.test.ts b/src/db/repositories/sessionRepository.test.ts index 89bba390..15d57464 100644 --- a/src/db/repositories/sessionRepository.test.ts +++ b/src/db/repositories/sessionRepository.test.ts @@ -397,6 +397,89 @@ describe('SessionRepository', () => { }); }); + // ── purgeOlderThan ──────────────────────────────────────────────────────── + + describe('purgeOlderThan', () => { + it('deletes expired/revoked rows past the DB-computed boundary in a bounded batch', async () => { + mockPool.query.mockResolvedValueOnce({ rows: [], rowCount: 42 } as any); + + const deleted = await repository.purgeOlderThan(30, 1000); + + expect(deleted).toBe(42); + const [query, params] = mockPool.query.mock.calls[0] as [string, any[]]; + expect(query).toContain('DELETE FROM sessions'); + expect(query).toContain('LIMIT $2'); + expect(params).toEqual([30, 1000]); + }); + + it('computes the retention boundary with the database clock, not the app clock', async () => { + mockPool.query.mockResolvedValueOnce({ rows: [], rowCount: 0 } as any); + + await repository.purgeOlderThan(30, 100); + + const [query, params] = mockPool.query.mock.calls[0] as [string, any[]]; + // The boundary must come from NOW() so a bad-clock event on the app + // server cannot push the boundary forward and make the job delete rows + // that have not actually aged past retention. + expect(query).toContain('NOW() - ($1 * INTERVAL'); + expect(query).not.toContain('WHERE expires_at < $1'); + expect(params[0]).toBe(30); + }); + + it('only ever targets expired or revoked rows, never active ones', async () => { + mockPool.query.mockResolvedValueOnce({ rows: [], rowCount: 0 } as any); + + await repository.purgeOlderThan(30, 100); + + const [query] = mockPool.query.mock.calls[0] as [string]; + // An active session has expires_at in the future and no revoked_at, so it + // can never satisfy either branch of the predicate. + expect(query).toMatch(/expires_at < NOW\(\)/); + expect(query).toMatch(/revoked_at < NOW\(\)/); + expect(query).not.toMatch(/WHERE expires_at < \$1 OR revoked_at < \$1/); + }); + + it('returns 0 when the DB reports a null rowCount', async () => { + mockPool.query.mockResolvedValueOnce({ rows: [], rowCount: null } as any); + + await expect(repository.purgeOlderThan(30, 100)).resolves.toBe(0); + }); + }); + + // ── getOldestCompactedSessionDate ───────────────────────────────────────── + + describe('getOldestCompactedSessionDate', () => { + it('returns the oldest eligible session date', async () => { + const oldest = new Date('2026-05-01T00:00:00.000Z'); + mockPool.query.mockResolvedValueOnce({ rows: [{ oldest }], rowCount: 1 } as any); + + const result = await repository.getOldestCompactedSessionDate(30); + + expect(result).toEqual(oldest); + const [query, params] = mockPool.query.mock.calls[0] as [string, any[]]; + expect(query).toContain('MIN(LEAST('); + expect(query).toContain('NOW() - ($1 * INTERVAL'); + expect(params).toEqual([30]); + }); + + it('uses the database clock so the lag metric matches the purge boundary', async () => { + mockPool.query.mockResolvedValueOnce({ rows: [{ oldest: null }], rowCount: 0 } as any); + + await repository.getOldestCompactedSessionDate(30); + + const [query] = mockPool.query.mock.calls[0] as [string]; + expect(query).toMatch(/expires_at < NOW\(\)/); + expect(query).toMatch(/revoked_at < NOW\(\)/); + expect(query).not.toContain('WHERE expires_at < $1'); + }); + + it('returns null when no rows are eligible for compaction', async () => { + mockPool.query.mockResolvedValueOnce({ rows: [{ oldest: null }], rowCount: 0 } as any); + + await expect(repository.getOldestCompactedSessionDate(30)).resolves.toBeNull(); + }); + }); + // ── createSession / mapSession: revoked_at and parent_id from existing tests ─ describe('legacy compatibility (original test cases preserved)', () => { diff --git a/src/db/repositories/sessionRepository.ts b/src/db/repositories/sessionRepository.ts index 96ab8a38..e6286da0 100644 --- a/src/db/repositories/sessionRepository.ts +++ b/src/db/repositories/sessionRepository.ts @@ -249,37 +249,51 @@ export class SessionRepository { } /** - * Get the date of the oldest expired or revoked session. - * Useful for calculating retention lag before compaction. + * Get the date of the oldest expired or revoked session that sits past the + * retention boundary. Useful for calculating retention lag before compaction. + * + * The retention boundary is computed with the database clock (`NOW()`), not + * the application clock. This keeps the boundary stable even during a + * bad-clock event on the app server, where a skewed `Date.now()` would + * otherwise push the boundary forward and make the job delete rows that have + * not actually aged past retention. */ - async getOldestCompactedSessionDate(cutoffDate: Date, client?: Pool): Promise { + async getOldestCompactedSessionDate(retentionDays: number, client?: Pool): Promise { const db = client || this.db; const query = ` SELECT MIN(LEAST(COALESCE(expires_at, 'infinity'::timestamp), COALESCE(revoked_at, 'infinity'::timestamp))) AS oldest FROM sessions - WHERE expires_at < $1 OR revoked_at < $1 + WHERE expires_at < NOW() - ($1 * INTERVAL '1 day') + OR revoked_at < NOW() - ($1 * INTERVAL '1 day') `; - const result = await db.query(query, [cutoffDate]); + const result = await db.query(query, [retentionDays]); return result.rows[0]?.oldest ?? null; } /** - * Delete expired or revoked sessions older than a specific cutoff date. + * Delete expired or revoked sessions older than the retention boundary. * Uses a bounded batch size to avoid long-held locks. - * + * + * The boundary is computed from the database clock (`NOW()`) so the job can + * never be pushed by a skewed application-server clock into deleting rows + * that are still within the retention window. The predicate only matches + * rows whose `expires_at` or `revoked_at` has fallen behind the boundary — + * active sessions (future `expires_at`, no `revoked_at`) can never match. + * * Returns the number of rows deleted in this batch. */ - async purgeOlderThan(cutoffDate: Date, batchSize: number, client?: Pool): Promise { + async purgeOlderThan(retentionDays: number, batchSize: number, client?: Pool): Promise { const db = client || this.db; const query = ` DELETE FROM sessions WHERE id IN ( SELECT id FROM sessions - WHERE expires_at < $1 OR revoked_at < $1 + WHERE expires_at < NOW() - ($1 * INTERVAL '1 day') + OR revoked_at < NOW() - ($1 * INTERVAL '1 day') LIMIT $2 ) `; - const result = await db.query(query, [cutoffDate, batchSize]); + const result = await db.query(query, [retentionDays, batchSize]); return result.rowCount ?? 0; } diff --git a/src/services/sessionCompactionService.test.ts b/src/services/sessionCompactionService.test.ts index 444a215c..ca1f4c16 100644 --- a/src/services/sessionCompactionService.test.ts +++ b/src/services/sessionCompactionService.test.ts @@ -8,84 +8,257 @@ describe('SessionCompactionService', () => { let sessionRepo: jest.Mocked; let metrics: jest.Mocked; let service: SessionCompactionService; - + beforeEach(() => { sessionRepo = { purgeOlderThan: jest.fn(), getOldestCompactedSessionDate: jest.fn(), vacuumSessions: jest.fn(), } as any; - + metrics = { incrementCounter: jest.fn(), recordHistogram: jest.fn(), } as any; - + env.SESSION_RETENTION_DAYS = 30; - + service = new SessionCompactionService(sessionRepo, metrics); - + jest.useFakeTimers(); jest.spyOn(globalLogger, 'info').mockImplementation(() => {}); jest.spyOn(globalLogger, 'error').mockImplementation(() => {}); + jest.spyOn(globalLogger, 'warn').mockImplementation(() => {}); }); - + afterEach(() => { service.stop(); jest.useRealTimers(); jest.restoreAllMocks(); }); - it('deletes sessions in batches and vacuums', async () => { - // Return 1000 for the first call, 500 for the second (meaning it's done) + it('deletes sessions in bounded batches and vacuums', async () => { + // First batch returns a full batch (1000), the second a partial batch, + // signalling that the table has been drained. sessionRepo.purgeOlderThan .mockResolvedValueOnce(1000) .mockResolvedValueOnce(500); - - sessionRepo.getOldestCompactedSessionDate.mockResolvedValue(new Date(Date.now() - 40 * 24 * 60 * 60 * 1000)); - + + sessionRepo.getOldestCompactedSessionDate.mockResolvedValue( + new Date(Date.now() - 40 * 24 * 60 * 60 * 1000), + ); + const result = await service.runCompaction(1000); - + expect(result.deletedCount).toBe(1500); + expect(result.capHit).toBe(false); expect(sessionRepo.purgeOlderThan).toHaveBeenCalledTimes(2); + expect(sessionRepo.purgeOlderThan).toHaveBeenCalledWith(30, 1000); expect(sessionRepo.vacuumSessions).toHaveBeenCalledTimes(1); - expect(metrics.incrementCounter).toHaveBeenCalledWith('session.compaction.rows', { status: 'success' }, 1500); - // 40 days - 30 days = 10 days lag - expect(metrics.recordHistogram).toHaveBeenCalledWith('session.compaction.retention_lag_days', 10, { status: 'success' }); + expect(metrics.incrementCounter).toHaveBeenCalledWith( + 'session.compaction.rows', + { status: 'success' }, + 1500, + ); + // 40 days - 30 days = 10 days lag from the retention boundary + expect(metrics.recordHistogram).toHaveBeenCalledWith( + 'session.compaction.retention_lag_days', + 10, + { status: 'success' }, + ); }); - it('does not vacuum if nothing was deleted', async () => { + it('does not vacuum when nothing was deleted', async () => { sessionRepo.purgeOlderThan.mockResolvedValue(0); sessionRepo.getOldestCompactedSessionDate.mockResolvedValue(null); - + const result = await service.runCompaction(1000); - + expect(result.deletedCount).toBe(0); + expect(result.lagDays).toBe(0); expect(sessionRepo.purgeOlderThan).toHaveBeenCalledTimes(1); expect(sessionRepo.vacuumSessions).not.toHaveBeenCalled(); - expect(metrics.recordHistogram).toHaveBeenCalledWith('session.compaction.retention_lag_days', 0, { status: 'success' }); + expect(metrics.incrementCounter).toHaveBeenCalledWith( + 'session.compaction.rows', + { status: 'success' }, + 0, + ); }); it('records error metrics if compaction fails', async () => { sessionRepo.purgeOlderThan.mockRejectedValue(new Error('DB failure')); - + await expect(service.runCompaction(1000)).rejects.toThrow('DB failure'); - - expect(metrics.incrementCounter).toHaveBeenCalledWith('session.compaction.errors_total', { status: 'error' }); + + expect(metrics.incrementCounter).toHaveBeenCalledWith( + 'session.compaction.errors_total', + { status: 'error' }, + ); + expect(metrics.recordHistogram).toHaveBeenCalledWith( + 'session.compaction.duration_ms', + expect.any(Number), + { status: 'error' }, + ); + }); + + it('passes retention days (not a server-derived cutoff) to the repository', async () => { + // Bad-clock safety: the boundary must be computed by the DATABASE clock. + // The service therefore hands the repository a number of days, never a + // Date derived from a potentially skewed application-server clock. + sessionRepo.purgeOlderThan.mockResolvedValue(0); + sessionRepo.getOldestCompactedSessionDate.mockResolvedValue(null); + + await service.runCompaction(1000); + + const [retentionArg, batchArg] = sessionRepo.purgeOlderThan.mock.calls[0]; + expect(retentionArg).toBe(env.SESSION_RETENTION_DAYS); + expect(retentionArg).not.toBeInstanceOf(Date); + expect(batchArg).toBe(1000); + + const [oldestArg] = sessionRepo.getOldestCompactedSessionDate.mock.calls[0]; + expect(oldestArg).not.toBeInstanceOf(Date); + expect(oldestArg).toBe(env.SESSION_RETENTION_DAYS); + }); + + it('stops at the per-run cap and emits the cap_hit metric', async () => { + // Simulate a table with far more eligible rows than the cap: every batch + // comes back full. + sessionRepo.purgeOlderThan.mockImplementation(async (_retention, size) => size); + sessionRepo.getOldestCompactedSessionDate.mockResolvedValue( + new Date(Date.now() - 40 * 24 * 60 * 60 * 1000), + ); + + const result = await service.runCompaction(1000, 2500); + + expect(result.deletedCount).toBe(2500); + expect(result.capHit).toBe(true); + // Batches of 1000, 1000, then a truncated 500 to land exactly on the cap. + expect(sessionRepo.purgeOlderThan).toHaveBeenCalledTimes(3); + expect(sessionRepo.purgeOlderThan).toHaveBeenNthCalledWith(1, 30, 1000); + expect(sessionRepo.purgeOlderThan).toHaveBeenNthCalledWith(2, 30, 1000); + expect(sessionRepo.purgeOlderThan).toHaveBeenNthCalledWith(3, 30, 500); + expect(metrics.incrementCounter).toHaveBeenCalledWith( + 'session.compaction.cap_hit', + { status: 'warning' }, + ); + expect(globalLogger.warn).toHaveBeenCalledWith( + expect.stringContaining('per-run cap'), + expect.any(Object), + ); + }); + + it('does not emit cap_hit when the last batch drains the table', async () => { + sessionRepo.purgeOlderThan.mockResolvedValueOnce(1000).mockResolvedValueOnce(250); + sessionRepo.getOldestCompactedSessionDate.mockResolvedValue(null); + + const result = await service.runCompaction(1000, 5000); + + expect(result.deletedCount).toBe(1250); + expect(result.capHit).toBe(false); + expect(metrics.incrementCounter).not.toHaveBeenCalledWith( + 'session.compaction.cap_hit', + { status: 'warning' }, + ); + }); + + describe('input validation', () => { + it('rejects a zero batch size', async () => { + await expect(service.runCompaction(0)).rejects.toThrow( + 'batchSize must be a positive integer', + ); + }); + + it('rejects a negative batch size', async () => { + await expect(service.runCompaction(-100)).rejects.toThrow( + 'batchSize must be a positive integer', + ); + }); + + it('rejects a non-integer batch size', async () => { + await expect(service.runCompaction(10.5)).rejects.toThrow( + 'batchSize must be a positive integer', + ); + }); + + it('rejects a zero or negative per-run cap', async () => { + await expect(service.runCompaction(1000, 0)).rejects.toThrow( + 'maxRowsPerRun must be a positive integer', + ); + await expect(service.runCompaction(1000, -5)).rejects.toThrow( + 'maxRowsPerRun must be a positive integer', + ); + }); }); it('starts and stops correctly', () => { - // start calls runCompaction immediately and sets interval - const runCompactionSpy = jest.spyOn(service, 'runCompaction').mockResolvedValue({ deletedCount: 0 }); - + const runCompactionSpy = jest + .spyOn(service, 'runCompaction') + .mockResolvedValue({ deletedCount: 0, lagDays: 0, capHit: false }); + service.start(1000); expect(runCompactionSpy).toHaveBeenCalledTimes(1); - + jest.advanceTimersByTime(1000); expect(runCompactionSpy).toHaveBeenCalledTimes(2); - + service.stop(); jest.advanceTimersByTime(1000); expect(runCompactionSpy).toHaveBeenCalledTimes(2); // no more calls }); + + it('start() restarts cleanly when called while already running', () => { + const runCompactionSpy = jest + .spyOn(service, 'runCompaction') + .mockResolvedValue({ deletedCount: 0, lagDays: 0, capHit: false }); + + service.start(1000); + service.start(2000); + + // The immediate run fires once per start; the second start clears the old + // interval, so advancing the old cadence must not trigger extra runs. + expect(runCompactionSpy).toHaveBeenCalledTimes(2); + jest.advanceTimersByTime(1000); + expect(runCompactionSpy).toHaveBeenCalledTimes(2); + + // The new 2000 ms cadence does fire. + jest.advanceTimersByTime(1000); + expect(runCompactionSpy).toHaveBeenCalledTimes(3); + }); + + it('start() logs a failed immediate run without throwing', async () => { + const runCompactionSpy = jest + .spyOn(service, 'runCompaction') + .mockRejectedValue(new Error('immediate DB failure')); + + expect(() => service.start(1000)).not.toThrow(); + + // Flush the microtask queue so the swallowed .catch() handler runs. + await Promise.resolve(); + + expect(globalLogger.error).toHaveBeenCalledWith( + 'Initial session compaction failed', + expect.objectContaining({ error: expect.any(Error) }), + ); + service.stop(); + expect(runCompactionSpy).toHaveBeenCalledTimes(1); + }); + + it('start() logs a failed scheduled run without throwing', async () => { + // First call (immediate run) resolves; second call (scheduled run) rejects. + jest + .spyOn(service, 'runCompaction') + .mockResolvedValueOnce({ deletedCount: 0, lagDays: 0, capHit: false }) + .mockRejectedValueOnce(new Error('scheduled DB failure')); + + service.start(1000); + jest.advanceTimersByTime(1000); + // Flush the microtask queue so the swallowed .catch() handler runs. + await Promise.resolve(); + + expect(globalLogger.error).toHaveBeenCalledWith( + 'Scheduled session compaction failed', + expect.objectContaining({ error: expect.any(Error) }), + ); + service.stop(); + }); }); diff --git a/src/services/sessionCompactionService.ts b/src/services/sessionCompactionService.ts index 6e89b7bf..e09be474 100644 --- a/src/services/sessionCompactionService.ts +++ b/src/services/sessionCompactionService.ts @@ -3,48 +3,80 @@ import { MetricsCollector } from '../lib/metrics'; import { globalLogger } from '../lib/logger'; import { env } from '../config/env'; +const DEFAULT_BATCH_SIZE = 1000; + +export interface CompactionResult { + deletedCount: number; + /** Days the oldest retained eligible row sits behind the retention boundary. */ + lagDays: number; + /** True when the run stopped because it reached the per-run row cap. */ + capHit: boolean; +} + /** * Service to manage the scheduled compaction of session storage. - * It deletes revoked or expired session rows older than the retention window, - * and vacuums the table to reclaim space. + * + * Revoked and expired session rows older than the retention window are deleted + * in bounded batches, then the table is vacuumed to reclaim space. + * + * Security/correctness assumptions: + * - The retention boundary is always computed with the DATABASE clock + * (`NOW() - retention`), never the application clock. A bad-clock event on + * the app server therefore cannot push the boundary forward and cause the + * job to delete rows that have not actually aged past retention. + * - The SQL predicate only matches rows whose `expires_at` or `revoked_at` is + * behind the boundary, so active sessions (future `expires_at`, no + * `revoked_at`) can never be deleted. + * - A per-run row cap (`SESSION_COMPACTION_MAX_ROWS_PER_RUN`, default + * 100_000) bounds how many rows a single cycle may delete. If an anomaly + * makes an enormous number of rows suddenly look eligible, only a fixed, + * recoverable number are removed per run and the `cap_hit` metric fires. + * + * Metrics emitted: + * - `session.compaction.rows` counter rows deleted per run + * - `session.compaction.retention_lag_days` histogram lag of the oldest + * eligible row behind the retention boundary + * - `session.compaction.duration_ms` histogram run duration + * - `session.compaction.errors_total` counter failed runs + * - `session.compaction.cap_hit` counter run hit the per-run cap */ export class SessionCompactionService { private intervalId?: NodeJS.Timeout; constructor( private readonly sessionRepo: SessionRepository, - private readonly metricsCollector?: MetricsCollector + private readonly metricsCollector?: MetricsCollector, ) {} /** - * Starts the scheduled compaction job + * Starts the scheduled compaction job. * @param intervalMs Interval in milliseconds (default: 24 hours) */ start(intervalMs: number = 24 * 60 * 60 * 1000): void { if (this.intervalId) { clearInterval(this.intervalId); } - - // Run immediately on start + + // Run immediately on start, then on the schedule. this.runCompaction().catch(err => { globalLogger.error('Initial session compaction failed', { error: err }); }); - // Schedule periodic runs this.intervalId = setInterval(() => { this.runCompaction().catch(err => { globalLogger.error('Scheduled session compaction failed', { error: err }); }); }, intervalMs); - - globalLogger.info('Session compaction service started', { + + globalLogger.info('Session compaction service started', { intervalMs, - retentionDays: env.SESSION_RETENTION_DAYS + retentionDays: env.SESSION_RETENTION_DAYS, + maxRowsPerRun: env.SESSION_COMPACTION_MAX_ROWS_PER_RUN, }); } /** - * Stops the scheduled compaction job + * Stops the scheduled compaction job. */ stop(): void { if (this.intervalId) { @@ -56,43 +88,78 @@ export class SessionCompactionService { /** * Executes a single compaction cycle. + * + * @param batchSize Max rows deleted per DELETE statement (bounded to avoid + * long-held locks). + * @param maxRowsPerRun Hard cap on rows deleted in one cycle; an anomaly can + * never wipe the table in a single run. */ - async runCompaction(batchSize: number = 1000): Promise<{ deletedCount: number }> { + async runCompaction( + batchSize: number = DEFAULT_BATCH_SIZE, + maxRowsPerRun: number = env.SESSION_COMPACTION_MAX_ROWS_PER_RUN, + ): Promise { + if (!Number.isInteger(batchSize) || batchSize <= 0) { + throw new Error(`batchSize must be a positive integer, got ${batchSize}`); + } + if (!Number.isInteger(maxRowsPerRun) || maxRowsPerRun <= 0) { + throw new Error(`maxRowsPerRun must be a positive integer, got ${maxRowsPerRun}`); + } + const startTime = Date.now(); + const retentionDays = env.SESSION_RETENTION_DAYS; let totalDeleted = 0; - - try { - const cutoffDate = new Date(); - cutoffDate.setDate(cutoffDate.getDate() - env.SESSION_RETENTION_DAYS); + let capHit = false; - globalLogger.info('Running session compaction', { cutoffDate: cutoffDate.toISOString(), batchSize }); + try { + globalLogger.info('Running session compaction', { + retentionDays, + batchSize, + maxRowsPerRun, + }); - // Calculate lag from retention boundary before deleting + // Lag from the retention boundary: how far past the boundary the oldest + // eligible row sits. The boundary matches the DB-clock boundary used by + // the purge, so the two stay consistent. let lagDays = 0; - const oldestDate = await this.sessionRepo.getOldestCompactedSessionDate(cutoffDate); + const oldestDate = await this.sessionRepo.getOldestCompactedSessionDate(retentionDays); if (oldestDate) { - const diffMs = cutoffDate.getTime() - oldestDate.getTime(); + const boundaryMs = Date.now() - retentionDays * 24 * 60 * 60 * 1000; + const diffMs = boundaryMs - oldestDate.getTime(); lagDays = Math.max(0, Math.floor(diffMs / (1000 * 60 * 60 * 24))); } - // Delete in bounded batches - let deletedInBatch = 0; - do { - deletedInBatch = await this.sessionRepo.purgeOlderThan(cutoffDate, batchSize); + // Delete in bounded batches. Stop once the per-run cap is reached so a + // bad-clock event (or any anomaly) cannot delete the whole table in one + // cycle; the remaining rows stay behind for the next run. + while (totalDeleted < maxRowsPerRun) { + const remaining = maxRowsPerRun - totalDeleted; + const size = Math.min(batchSize, remaining); + const deletedInBatch = await this.sessionRepo.purgeOlderThan(retentionDays, size); totalDeleted += deletedInBatch; - } while (deletedInBatch === batchSize); - + if (deletedInBatch < size) break; + } + capHit = totalDeleted >= maxRowsPerRun; + if (totalDeleted > 0) { - // Run vacuum to reclaim space + // Reclaim space after bulk deletion. VACUUM cannot run inside a + // transaction block, so it is executed on its own connection. globalLogger.info('Vacuuming sessions table after compaction'); await this.sessionRepo.vacuumSessions(); } const duration = Date.now() - startTime; - + + if (capHit) { + globalLogger.warn( + 'Session compaction reached per-run cap; more rows remain eligible', + { deletedCount: totalDeleted, maxRowsPerRun }, + ); + } + globalLogger.info('Session compaction complete', { deletedCount: totalDeleted, lagDays, + capHit, durationMs: duration, }); @@ -100,9 +167,12 @@ export class SessionCompactionService { this.metricsCollector.incrementCounter('session.compaction.rows', { status: 'success' }, totalDeleted); this.metricsCollector.recordHistogram('session.compaction.retention_lag_days', lagDays, { status: 'success' }); this.metricsCollector.recordHistogram('session.compaction.duration_ms', duration, { status: 'success' }); + if (capHit) { + this.metricsCollector.incrementCounter('session.compaction.cap_hit', { status: 'warning' }); + } } - return { deletedCount: totalDeleted }; + return { deletedCount: totalDeleted, lagDays, capHit }; } catch (error) { const duration = Date.now() - startTime; if (this.metricsCollector) {