diff --git a/packages/gateway/__tests__/repo/dump-store_test.ts b/packages/gateway/__tests__/repo/dump-store_test.ts index de336b25a..2211f99b8 100644 --- a/packages/gateway/__tests__/repo/dump-store_test.ts +++ b/packages/gateway/__tests__/repo/dump-store_test.ts @@ -4,7 +4,7 @@ import { dirname, join, resolve } from 'node:path'; import { expect, test } from 'vitest'; -import { createSqliteTestDb } from './test-sqlite.ts'; +import { createSqliteTestDb, mapRunChangeCount } from './test-sqlite.ts'; import { decodeDumpBodyDescriptor } from '../../src/dump/storage-codec.ts'; import type { DumpWriteRecord } from '../../src/dump/types.ts'; import { FileDumpStore } from '../../src/repo/dump-store.ts'; @@ -320,6 +320,35 @@ test('FileDumpStore retires every dump record when retention is disabled and col assertEquals((await db.prepare('SELECT COUNT(*) AS count FROM spilled_files').first<{ count: number }>())?.count, 0); }); +test('FileDumpStore counts returned dump rows instead of trigger-amplified changes', async () => { + const db = await openDb(); + const repo = new SqlRepo(db); + const files = new MemoryFileStore(); + const store = new FileDumpStore(db, files); + await store.put('key_x', baseRecord('01HZZ0000000000000000000C1', Date.UTC(2026, 5, 1, 9))); + await store.put('key_x', baseRecord('01HZZ0000000000000000000C2', Date.UTC(2026, 5, 1, 10))); + await repo.apiKeys.update('key_x', { dumpRetentionSeconds: null }); + + const d1LikeStore = new FileDumpStore(mapRunChangeCount(db, changes => changes * 3), files); + expect(await d1LikeStore.deleteExpiredBatch('key_x', Date.now(), 1)).toBe(1); + expect((await db.prepare('SELECT COUNT(*) AS count FROM dump_records').first<{ count: number }>())?.count).toBe(1); +}); + +test('FileDumpStore counts returned active dump rows instead of trigger-amplified changes', async () => { + const db = await openDb(); + const repo = new SqlRepo(db); + const files = new MemoryFileStore(); + const store = new FileDumpStore(db, files); + const now = Date.UTC(2026, 5, 1, 12); + await store.put('key_x', baseRecord('01HZZ0000000000000000000C3', Date.UTC(2026, 5, 1, 9))); + await store.put('key_x', baseRecord('01HZZ0000000000000000000C4', Date.UTC(2026, 5, 1, 10))); + await repo.apiKeys.update('key_x', { dumpRetentionSeconds: 3600 }); + + const d1LikeStore = new FileDumpStore(mapRunChangeCount(db, changes => changes * 3), files); + expect(await d1LikeStore.deleteExpiredBatch('key_x', now, 1)).toBe(1); + expect((await db.prepare('SELECT COUNT(*) AS count FROM dump_records').first<{ count: number }>())?.count).toBe(1); +}); + test('a record-ID race leaves only the losing write\'s uniquely keyed files collectible', async () => { const db = await openDb(); const repo = new SqlRepo(db); diff --git a/packages/gateway/__tests__/repo/expiration-sweep-integrity_test.ts b/packages/gateway/__tests__/repo/expiration-sweep-integrity_test.ts new file mode 100644 index 000000000..43eccaf10 --- /dev/null +++ b/packages/gateway/__tests__/repo/expiration-sweep-integrity_test.ts @@ -0,0 +1,235 @@ +import { expect, test } from 'vitest'; + +import { createSqliteTestDb, createSqlJsDatabase, migrationSqlByFilename, wrapSqlJsDatabase } from './test-sqlite.ts'; +import { initDumpStore } from '../../src/dump/registry.ts'; +import { FileDumpStore } from '../../src/repo/dump-store.ts'; +import { initRepo } from '../../src/repo/index.ts'; +import { SqlRepo } from '../../src/repo/sql.ts'; +import { sweepExpirations } from '../../src/scheduled/expiration-sweeps.ts'; +import { MemoryFileStore } from '@floway-dev/platform'; + +const integrityMigration = (): string => { + const migration = migrationSqlByFilename.find(([filename]) => filename === '0082_expiration_sweep_integrity.sql'); + if (migration === undefined) throw new Error('missing migration 0082_expiration_sweep_integrity.sql'); + return migration[1]; +}; + +const createPreIntegrityDatabase = async () => { + const raw = await createSqlJsDatabase(); + try { + for (const [filename, sql] of migrationSqlByFilename) { + if (filename === '0082_expiration_sweep_integrity.sql') break; + raw.run(sql); + } + return { raw, db: wrapSqlJsDatabase(raw) }; + } catch (error) { + raw.close(); + throw error; + } +}; + +const insertApiKey = (raw: Awaited>, id: string): void => { + const serverSecret = [...id] + .map(character => character.charCodeAt(0).toString(16)) + .join('') + .padEnd(64, '0') + .slice(0, 64); + raw.run( + `INSERT INTO api_keys + (id, user_id, name, key, created_at, upstream_ids, deleted_at, + dump_retention_seconds, server_secret, responses_retention_seconds) + VALUES (?, 1, ?, ?, '2026-01-01T00:00:00Z', NULL, NULL, 3600, ?, 86400)`, + [id, id, `raw-${id}`, serverSecret], + ); +}; + +const insertDump = (raw: Awaited>, keyId: string, id: string): void => { + raw.run( + `INSERT INTO dump_records + (key_id, id, created_at, upstream_id, meta_json, request_headers_json, + response_headers_json, request_body_descriptor, response_body_descriptor) + VALUES (?, ?, 1, NULL, '{}', '[]', NULL, NULL, NULL)`, + [keyId, id], + ); +}; + +test('migration repairs missing coverage without disturbing existing sweeps', async () => { + const { raw } = await createPreIntegrityDatabase(); + try { + for (const id of [ + 'dump-missing', + 'item-missing', + 'snapshot-missing', + 'existing-dump', + 'existing-responses', + 'empty', + ]) insertApiKey(raw, id); + insertDump(raw, 'dump-missing', 'dump-a'); + raw.run( + `INSERT INTO responses_items + (id, api_key_id, payload_json, item_hash, payload_hash, payload_file_key, refreshed_at) + VALUES + ('item-a', 'item-missing', '{}', 'item-hash', 'payload-hash', NULL, 0), + ('item-existing', 'existing-responses', '{}', 'existing-item-hash', 'existing-payload-hash', NULL, 0)`, + ); + raw.run( + `INSERT INTO responses_snapshots (id, api_key_id, item_ids_json, refreshed_at) + VALUES ('snapshot-a', 'snapshot-missing', '[]', 0)`, + ); + insertDump(raw, 'existing-dump', 'dump-existing'); + raw.run( + `UPDATE expiration_sweeps + SET due_at = 999, revision = 7, claim_token = 'held', claimed_at = 123 + WHERE domain = 'dumps' AND key_id = 'existing-dump'`, + ); + raw.run( + `UPDATE expiration_sweeps + SET due_at = 777, revision = 5, claim_token = 'responses-held', claimed_at = 456 + WHERE domain = 'responses' AND key_id = 'existing-responses'`, + ); + raw.run("DELETE FROM expiration_sweeps WHERE key_id IN ('dump-missing', 'item-missing', 'snapshot-missing')"); + + raw.run(integrityMigration()); + + expect(raw.exec( + `SELECT domain, key_id, due_at, revision, claim_token, claimed_at + FROM expiration_sweeps ORDER BY domain, key_id`, + )[0].values).toEqual([ + ['dumps', 'dump-missing', 0, 0, null, null], + ['dumps', 'existing-dump', 999, 7, 'held', 123], + ['responses', 'existing-responses', 777, 5, 'responses-held', 456], + ['responses', 'item-missing', 0, 0, null, null], + ['responses', 'snapshot-missing', 0, 0, null, null], + ]); + } finally { + raw.close(); + } +}); + +test('queue rows cannot be removed before their domain rows', async () => { + const { raw } = await createPreIntegrityDatabase(); + try { + insertApiKey(raw, 'both'); + insertApiKey(raw, 'empty'); + insertApiKey(raw, 'identity'); + insertDump(raw, 'both', 'dump-a'); + insertDump(raw, 'identity', 'dump-identity'); + raw.run( + `INSERT INTO responses_items + (id, api_key_id, payload_json, item_hash, payload_hash, payload_file_key, refreshed_at) + VALUES ('item-a', 'both', '{}', 'item-hash', 'payload-hash', NULL, 0)`, + ); + raw.run( + `INSERT INTO responses_snapshots (id, api_key_id, item_ids_json, refreshed_at) + VALUES ('snapshot-a', 'both', '[]', 0)`, + ); + raw.run(integrityMigration()); + raw.run("INSERT INTO expiration_sweeps (domain, key_id, due_at) VALUES ('dumps', 'empty', 0)"); + + expect(() => raw.run("DELETE FROM expiration_sweeps WHERE domain = 'dumps' AND key_id = 'both'")) + .toThrow('expiration sweep cannot be removed while stored rows remain'); + expect(() => raw.run("DELETE FROM expiration_sweeps WHERE domain = 'responses' AND key_id = 'both'")) + .toThrow('expiration sweep cannot be removed while stored rows remain'); + expect(() => raw.run("UPDATE expiration_sweeps SET key_id = 'moved' WHERE domain = 'dumps' AND key_id = 'identity'")) + .toThrow('expiration sweep identity is immutable'); + expect(() => raw.run("UPDATE expiration_sweeps SET domain = 'responses' WHERE domain = 'dumps' AND key_id = 'identity'")) + .toThrow('expiration sweep identity is immutable'); + expect(() => raw.run("UPDATE expiration_sweeps SET key_id = 'moved-empty' WHERE domain = 'dumps' AND key_id = 'empty'")) + .toThrow('expiration sweep identity is immutable'); + expect(() => raw.run("UPDATE expiration_sweeps SET domain = 'responses' WHERE domain = 'dumps' AND key_id = 'empty'")) + .toThrow('expiration sweep identity is immutable'); + + raw.run("DELETE FROM dump_records WHERE key_id = 'both'"); + raw.run("DELETE FROM expiration_sweeps WHERE domain = 'dumps' AND key_id = 'both'"); + expect(raw.exec("SELECT domain FROM expiration_sweeps WHERE key_id = 'both'")[0].values) + .toEqual([['responses']]); + + raw.run("DELETE FROM responses_items WHERE api_key_id = 'both'"); + expect(() => raw.run("DELETE FROM expiration_sweeps WHERE domain = 'responses' AND key_id = 'both'")) + .toThrow('expiration sweep cannot be removed while stored rows remain'); + raw.run("DELETE FROM responses_snapshots WHERE api_key_id = 'both'"); + raw.run("DELETE FROM expiration_sweeps WHERE domain = 'responses' AND key_id = 'both'"); + expect(raw.exec("SELECT domain FROM expiration_sweeps WHERE key_id = 'both'")).toEqual([]); + } finally { + raw.close(); + } +}); + +test('drained completion cannot remove a queue whose domain still has rows', async () => { + const db = await createSqliteTestDb(); + const repo = new SqlRepo(db); + await repo.apiKeys.save({ + id: 'key-a', + userId: 1, + name: 'Key A', + key: 'raw-key-a', + serverSecret: 'aa'.repeat(32), + createdAt: '2026-01-01T00:00:00Z', + upstreamIds: null, + deletedAt: null, + dumpRetentionSeconds: 3600, + responsesRetentionSeconds: 0, + }); + await db.prepare( + `INSERT INTO dump_records + (key_id, id, created_at, upstream_id, meta_json, request_headers_json, + response_headers_json, request_body_descriptor, response_body_descriptor) + VALUES ('key-a', 'dump-a', 1, NULL, '{}', '[]', NULL, NULL, NULL)`, + ).run(); + await repo.expirationSweeps.schedule('dumps', 'key-a', 0); + const first = await repo.expirationSweeps.claim('claim-a', 10, 0); + if (first === null) throw new Error('expected first expiration claim'); + + await expect(repo.expirationSweeps.complete('claim-a', first.revision, { kind: 'drained', nextDueAt: null })) + .rejects.toThrow('expiration sweep cannot be removed while stored rows remain'); + await repo.expirationSweeps.complete('claim-a', first.revision, { kind: 'partial', retryAt: 12 }); + expect(await db.prepare( + "SELECT due_at, claim_token FROM expiration_sweeps WHERE domain = 'dumps' AND key_id = 'key-a'", + ).first()).toEqual({ due_at: 12, claim_token: null }); + + await db.prepare("DELETE FROM dump_records WHERE key_id = 'key-a'").run(); + const second = await repo.expirationSweeps.claim('claim-b', 12, 0); + if (second === null) throw new Error('expected second expiration claim'); + await repo.expirationSweeps.complete('claim-b', second.revision, { kind: 'drained', nextDueAt: null }); + expect(await db.prepare("SELECT domain FROM expiration_sweeps WHERE key_id = 'key-a'").first()).toBeNull(); +}); + +test('repaired inactive dump queues drain in bounded ticks', async () => { + const { raw, db } = await createPreIntegrityDatabase(); + try { + insertApiKey(raw, 'disabled'); + insertApiKey(raw, 'deleted'); + for (let index = 0; index < 75; index += 1) { + insertDump(raw, 'disabled', `disabled-${index}`); + insertDump(raw, 'deleted', `deleted-${index}`); + } + raw.run("UPDATE api_keys SET dump_retention_seconds = NULL WHERE id = 'disabled'"); + raw.run("UPDATE api_keys SET deleted_at = '2026-01-02T00:00:00Z' WHERE id = 'deleted'"); + raw.run("DELETE FROM expiration_sweeps WHERE domain = 'dumps'"); + raw.run(integrityMigration()); + + const repo = new SqlRepo(db); + initRepo(repo); + const files = new MemoryFileStore(); + initDumpStore(new FileDumpStore(db, files)); + + await sweepExpirations(Date.UTC(2026, 0, 3)); + expect((await db.prepare( + 'SELECT key_id, COUNT(*) AS count FROM dump_records GROUP BY key_id ORDER BY key_id', + ).all<{ key_id: string; count: number }>()).results).toEqual([ + { key_id: 'deleted', count: 25 }, + { key_id: 'disabled', count: 25 }, + ]); + expect((await db.prepare( + "SELECT key_id FROM expiration_sweeps WHERE domain = 'dumps' ORDER BY key_id", + ).all<{ key_id: string }>()).results).toEqual([ + { key_id: 'deleted' }, + { key_id: 'disabled' }, + ]); + await sweepExpirations(Date.UTC(2026, 0, 3, 0, 1)); + expect((await db.prepare('SELECT COUNT(*) AS count FROM dump_records').first<{ count: number }>())?.count).toBe(0); + expect(await db.prepare("SELECT domain FROM expiration_sweeps WHERE domain = 'dumps'").first()).toBeNull(); + } finally { + raw.close(); + } +}); diff --git a/packages/gateway/__tests__/repo/responses-items_test.ts b/packages/gateway/__tests__/repo/responses-items_test.ts index 8fd933ee3..32c292885 100644 --- a/packages/gateway/__tests__/repo/responses-items_test.ts +++ b/packages/gateway/__tests__/repo/responses-items_test.ts @@ -1,7 +1,7 @@ import { afterEach, describe, expect, test, vi } from 'vitest'; import { InMemoryRepo } from './memory.ts'; -import { createSqliteTestDb, createSqlJsDatabase, migrationSqlByFilename } from './test-sqlite.ts'; +import { createSqliteTestDb, createSqlJsDatabase, mapRunChangeCount, migrationSqlByFilename } from './test-sqlite.ts'; import { initRepo } from '../../src/repo/index.ts'; import { hashResponsesJson } from '../../src/repo/responses-hash.ts'; import { prepareStoredResponsesPayload } from '../../src/repo/responses-payload.ts'; @@ -316,6 +316,39 @@ test('SQL spill ownership is first-class and the shared collector reclaims retir expect(await db.prepare('SELECT file_key FROM spilled_files WHERE file_key = ?').bind(owned.file_key).first()).toBeNull(); }); +test('SQL counts returned Responses rows instead of trigger-amplified changes', async () => { + const db = await createSqliteTestDb(); + const repo = new SqlRepo(db); + initFileStore(new MemoryFileStore()); + const now = atDay(10, DAY_MS / 2); + vi.useFakeTimers(); + vi.setSystemTime(now); + await repo.apiKeys.save(apiKey()); + await repo.responsesItems.insertMany([storedItem('msg-counted', now, largeContent())], 0); + await repo.apiKeys.update('key-a', { responsesRetentionSeconds: 0 }); + + const d1LikeRepo = new SqlRepo(mapRunChangeCount(db, changes => changes * 2)); + expect(await d1LikeRepo.responsesItems.deleteExpiredBatch('key-a', now, 1)).toBe(1); + expect(await db.prepare('SELECT id FROM responses_items').first()).toBeNull(); +}); + +test('SQL counts returned active Responses rows instead of trigger-amplified changes', async () => { + const db = await createSqliteTestDb(); + const repo = new SqlRepo(db); + initFileStore(new MemoryFileStore()); + const now = atDay(10, DAY_MS / 2); + vi.useFakeTimers(); + vi.setSystemTime(now); + await repo.apiKeys.save(apiKey()); + await repo.responsesItems.insertMany([ + storedItem('msg-counted-active', responsesStateCutoff(now, RETENTION_SECONDS) - 1, largeContent()), + ], 0); + + const d1LikeRepo = new SqlRepo(mapRunChangeCount(db, changes => changes * 2)); + expect(await d1LikeRepo.responsesItems.deleteExpiredBatch('key-a', now, 1)).toBe(1); + expect(await db.prepare('SELECT id FROM responses_items').first()).toBeNull(); +}); + test('SQL performs no item or snapshot mutation after an earlier refresh in the same UTC day', async () => { vi.useFakeTimers(); vi.setSystemTime(atDay(10, DAY_MS / 4)); diff --git a/packages/gateway/__tests__/repo/test-sqlite.ts b/packages/gateway/__tests__/repo/test-sqlite.ts index d418ca923..be27a2234 100644 --- a/packages/gateway/__tests__/repo/test-sqlite.ts +++ b/packages/gateway/__tests__/repo/test-sqlite.ts @@ -37,6 +37,24 @@ export const createSqliteTestDb = async (): Promise => { return wrapSqlJsDatabase(db); }; +export const mapRunChangeCount = (db: SqlDatabase, mapper: (changes: number) => number): SqlDatabase => ({ + prepare(query) { + const wrap = (statement: SqlPreparedStatement): SqlPreparedStatement => ({ + bind: (...values) => wrap(statement.bind(...values)), + first: async () => await statement.first(), + all: async () => await statement.all(), + async run() { + const result = await statement.run(); + const changes = result.meta.changes; + if (changes === undefined) throw new Error('SQL run result omitted its change count'); + return { ...result, meta: { ...result.meta, changes: mapper(changes) } }; + }, + }); + return wrap(db.prepare(query)); + }, + exec: async sql => await db.exec(sql), +}); + // sql.js binds through JavaScript and happily takes values neither deployment // target accepts, so it would pass a statement that fails in production. Reject // anything outside the contract's own union here instead. diff --git a/packages/gateway/__tests__/repo/test-sqlite_test.ts b/packages/gateway/__tests__/repo/test-sqlite_test.ts index 01b957bc8..8962aef2e 100644 --- a/packages/gateway/__tests__/repo/test-sqlite_test.ts +++ b/packages/gateway/__tests__/repo/test-sqlite_test.ts @@ -1,6 +1,7 @@ -import { test } from 'vitest'; +import { expect, test, vi } from 'vitest'; -import { assertD1CompoundSelectLimit } from './test-sqlite.ts'; +import { assertD1CompoundSelectLimit, createSqliteTestDb, mapRunChangeCount } from './test-sqlite.ts'; +import type { SqlDatabase, SqlPreparedStatement } from '@floway-dev/platform'; import { assertThrows } from '@floway-dev/test-utils'; test('D1 compound SELECT verifier accepts five terms and rejects six', () => { @@ -12,3 +13,32 @@ test('D1 compound SELECT verifier accepts five terms and rejects six', () => { ); assertD1CompoundSelectLimit("SELECT 'UNION UNION UNION UNION UNION' /* UNION */"); }); + +test('mapRunChangeCount maps run metadata changes', async () => { + const db = await createSqliteTestDb(); + const mapper = vi.fn((changes: number) => changes * 3); + const wrapped = mapRunChangeCount(db, mapper); + + const result = await wrapped + .prepare('UPDATE users SET username = username WHERE id = 1') + .run(); + + expect(mapper).toHaveBeenCalledWith(1); + expect(result.meta.changes).toBe(3); +}); + +test('mapRunChangeCount rejects missing run metadata changes', async () => { + const statement: SqlPreparedStatement = { + bind: () => statement, + first: async () => null, + all: async () => ({ results: [], success: true, meta: {} }), + run: async () => ({ results: [], success: true, meta: {} }), + }; + const db: SqlDatabase = { + prepare: () => statement, + exec: async () => undefined, + }; + + await expect(mapRunChangeCount(db, changes => changes).prepare('SELECT 1').run()) + .rejects.toThrow('SQL run result omitted its change count'); +}); diff --git a/packages/gateway/__tests__/scheduled/expiration-sweeps_test.ts b/packages/gateway/__tests__/scheduled/expiration-sweeps_test.ts index 037909ba0..04a894fd2 100644 --- a/packages/gateway/__tests__/scheduled/expiration-sweeps_test.ts +++ b/packages/gateway/__tests__/scheduled/expiration-sweeps_test.ts @@ -357,7 +357,7 @@ test('expiration claims and expired-row deletions use their bounded range indexe AND stored.api_key_id = api_keys.id AND stored.refreshed_at < ? - api_keys.responses_retention_seconds * 1000 ORDER BY stored.refreshed_at, stored.rowid LIMIT ? - )`, + ) RETURNING rowid`, 'key-a', 1, 100, ); expect(responsesPlan).toContain('idx_responses_items_key_refresh'); @@ -370,7 +370,7 @@ test('expiration claims and expired-row deletions use their bounded range indexe AND records.key_id = api_keys.id AND records.created_at < ? - api_keys.dump_retention_seconds * 1000 ORDER BY records.created_at, records.rowid LIMIT ? - )`, + ) RETURNING rowid`, 'key-a', 1, 100, ); expect(dumpsPlan).toContain('idx_dump_records_key_created'); @@ -378,31 +378,40 @@ test('expiration claims and expired-row deletions use their bounded range indexe test('bounded cleanup backfill tracks rows whose API key was hard-deleted', async () => { const now = Date.UTC(2026, 6, 23, 12); - const db = await createSqliteTestDb(); - const repo = new SqlRepo(db); - await repo.apiKeys.save(key(now)); - const recordId = '01K00000000000000000ORPH'; - const fileKey = `dumps/v1/key-a/1970010100/${recordId}.req.gz`; - await db.prepare( - `INSERT INTO dump_records - (key_id, id, created_at, upstream_id, meta_json, request_headers_json, response_headers_json, request_body_descriptor, response_body_descriptor) - VALUES ('key-a', ?, 1, NULL, '{}', '[]', NULL, ?, NULL)`, - ).bind(recordId, JSON.stringify({ key: fileKey, type: 'bytes' })).run(); - await db.prepare("DELETE FROM api_keys WHERE id = 'key-a'").run(); - await db.prepare("DELETE FROM expiration_sweeps WHERE key_id = 'key-a'").run(); - await db.prepare('DELETE FROM spilled_files WHERE file_key = ?').bind(fileKey).run(); - - await repo.expirationSweeps.backfillCleanupTracking(500); - expect(await db.prepare( - "SELECT due_at FROM expiration_sweeps WHERE domain = 'dumps' AND key_id = 'key-a'", - ).first<{ due_at: number }>()).toEqual({ due_at: 0 }); - expect(await db.prepare( - 'SELECT owner_kind, owner_key, state FROM spilled_files WHERE file_key = ?', - ).bind(fileKey).first()).toEqual({ - owner_kind: 'dump-request', - owner_key: JSON.stringify(['key-a', recordId]), - state: 'owned', - }); + const raw = await createSqlJsDatabase(); + try { + for (const [filename, sql] of migrationSqlByFilename) { + if (filename === '0082_expiration_sweep_integrity.sql') break; + raw.run(sql); + } + const db = wrapSqlJsDatabase(raw); + const repo = new SqlRepo(db); + await repo.apiKeys.save(key(now)); + const recordId = '01K00000000000000000ORPH'; + const fileKey = `dumps/v1/key-a/1970010100/${recordId}.req.gz`; + await db.prepare( + `INSERT INTO dump_records + (key_id, id, created_at, upstream_id, meta_json, request_headers_json, response_headers_json, request_body_descriptor, response_body_descriptor) + VALUES ('key-a', ?, 1, NULL, '{}', '[]', NULL, ?, NULL)`, + ).bind(recordId, JSON.stringify({ key: fileKey, type: 'bytes' })).run(); + await db.prepare("DELETE FROM api_keys WHERE id = 'key-a'").run(); + await db.prepare("DELETE FROM expiration_sweeps WHERE key_id = 'key-a'").run(); + await db.prepare('DELETE FROM spilled_files WHERE file_key = ?').bind(fileKey).run(); + + await repo.expirationSweeps.backfillCleanupTracking(500); + expect(await db.prepare( + "SELECT due_at FROM expiration_sweeps WHERE domain = 'dumps' AND key_id = 'key-a'", + ).first<{ due_at: number }>()).toEqual({ due_at: 0 }); + expect(await db.prepare( + 'SELECT owner_kind, owner_key, state FROM spilled_files WHERE file_key = ?', + ).bind(fileKey).first()).toEqual({ + owner_kind: 'dump-request', + owner_key: JSON.stringify(['key-a', recordId]), + state: 'owned', + }); + } finally { + raw.close(); + } }); test('bounded cleanup backfill skips API keys without stored state', async () => { diff --git a/packages/gateway/migrations/0082_expiration_sweep_integrity.sql b/packages/gateway/migrations/0082_expiration_sweep_integrity.sql new file mode 100644 index 000000000..63ce12ec7 --- /dev/null +++ b/packages/gateway/migrations/0082_expiration_sweep_integrity.sql @@ -0,0 +1,51 @@ +CREATE TRIGGER expiration_sweeps_reject_nonempty_delete +BEFORE DELETE ON expiration_sweeps +WHEN ( + OLD.domain = 'dumps' + AND EXISTS ( + SELECT 1 FROM dump_records + WHERE key_id = OLD.key_id + ) +) OR ( + OLD.domain = 'responses' + AND ( + EXISTS ( + SELECT 1 FROM responses_items + WHERE api_key_id = OLD.key_id + ) + OR EXISTS ( + SELECT 1 FROM responses_snapshots + WHERE api_key_id = OLD.key_id + ) + ) +) +BEGIN + SELECT RAISE(ABORT, 'expiration sweep cannot be removed while stored rows remain'); +END; + +CREATE TRIGGER expiration_sweeps_reject_nonempty_identity_update +BEFORE UPDATE OF domain, key_id ON expiration_sweeps +WHEN ( + OLD.domain != NEW.domain + OR OLD.key_id != NEW.key_id +) +BEGIN + SELECT RAISE(ABORT, 'expiration sweep identity is immutable'); +END; + +INSERT INTO expiration_sweeps (domain, key_id, due_at) +SELECT 'dumps', key_id, 0 +FROM dump_records +WHERE true +GROUP BY key_id +ON CONFLICT (domain, key_id) DO NOTHING; + +INSERT INTO expiration_sweeps (domain, key_id, due_at) +SELECT 'responses', api_key_id, 0 +FROM ( + SELECT api_key_id FROM responses_items + UNION + SELECT api_key_id FROM responses_snapshots +) +WHERE true +ON CONFLICT (domain, key_id) DO NOTHING; diff --git a/packages/gateway/src/repo/dump-store.ts b/packages/gateway/src/repo/dump-store.ts index b11a3015b..ab29082fa 100644 --- a/packages/gateway/src/repo/dump-store.ts +++ b/packages/gateway/src/repo/dump-store.ts @@ -276,6 +276,11 @@ export class FileDumpStore implements DumpStore { } async deleteExpiredBatch(keyId: string, now: number, limit: number): Promise { + // D1 derives meta.changes from total_changes(), so the dump retirement trigger + // can add spilled_files writes. RETURNING counts only dump rows. + // https://github.com/cloudflare/workerd/blob/0c0f9656d3f78c75a7dc011e0c17dd85e438b44c/src/cloudflare/internal/test/d1/d1-mock.js#L83-L131 + // https://www.sqlite.org/c3ref/total_changes.html + // https://www.sqlite.org/lang_returning.html const active = await this.db .prepare( `DELETE FROM dump_records WHERE rowid IN ( @@ -289,11 +294,12 @@ export class FileDumpStore implements DumpStore { AND records.created_at < ? - api_keys.dump_retention_seconds * 1000 ORDER BY records.created_at, records.rowid LIMIT ? - )`, + ) + RETURNING rowid`, ) .bind(keyId, now, limit) - .run(); - const activeDeleted = active.meta.changes ?? 0; + .all<{ rowid: number }>(); + const activeDeleted = active.results.length; if (activeDeleted >= limit) return activeDeleted; const inactive = await this.db .prepare( @@ -308,11 +314,12 @@ export class FileDumpStore implements DumpStore { ) ORDER BY records.created_at, records.rowid LIMIT ? - )`, + ) + RETURNING rowid`, ) .bind(keyId, limit - activeDeleted) - .run(); - return activeDeleted + (inactive.meta.changes ?? 0); + .all<{ rowid: number }>(); + return activeDeleted + inactive.results.length; } async findOldestCreatedAt(keyId: string): Promise { diff --git a/packages/gateway/src/repo/responses-state-sql.ts b/packages/gateway/src/repo/responses-state-sql.ts index ddf469589..15369cb84 100644 --- a/packages/gateway/src/repo/responses-state-sql.ts +++ b/packages/gateway/src/repo/responses-state-sql.ts @@ -275,6 +275,11 @@ export class SqlResponsesItemsRepo implements ResponsesItemsRepo { } async deleteExpiredBatch(apiKeyId: string, now: number, limit: number): Promise { + // D1 derives meta.changes from total_changes(), so payload retirement + // triggers can inflate it. RETURNING counts only Responses item rows. + // https://github.com/cloudflare/workerd/blob/0c0f9656d3f78c75a7dc011e0c17dd85e438b44c/src/cloudflare/internal/test/d1/d1-mock.js#L83-L131 + // https://www.sqlite.org/c3ref/total_changes.html + // https://www.sqlite.org/lang_returning.html const active = await this.db .prepare( `DELETE FROM responses_items WHERE rowid IN ( @@ -288,11 +293,12 @@ export class SqlResponsesItemsRepo implements ResponsesItemsRepo { AND stored.refreshed_at < ? - api_keys.responses_retention_seconds * 1000 ORDER BY stored.refreshed_at, stored.rowid LIMIT ? - )`, + ) + RETURNING rowid`, ) .bind(apiKeyId, now - RESPONSES_REFRESH_GRANULARITY_MS, limit) - .run(); - const activeDeleted = active.meta.changes ?? 0; + .all<{ rowid: number }>(); + const activeDeleted = active.results.length; if (activeDeleted >= limit) return activeDeleted; const inactive = await this.db .prepare( @@ -307,11 +313,12 @@ export class SqlResponsesItemsRepo implements ResponsesItemsRepo { ) ORDER BY stored.refreshed_at, stored.rowid LIMIT ? - )`, + ) + RETURNING rowid`, ) .bind(apiKeyId, limit - activeDeleted) - .run(); - return activeDeleted + (inactive.meta.changes ?? 0); + .all<{ rowid: number }>(); + return activeDeleted + inactive.results.length; } async findOldestRefreshedAt(apiKeyId: string): Promise { @@ -394,11 +401,12 @@ export class SqlResponsesSnapshotsRepo implements ResponsesSnapshotsRepo { AND stored.refreshed_at < ? - api_keys.responses_retention_seconds * 1000 ORDER BY stored.refreshed_at, stored.rowid LIMIT ? - )`, + ) + RETURNING rowid`, ) .bind(apiKeyId, now - RESPONSES_REFRESH_GRANULARITY_MS, limit) - .run(); - const activeDeleted = active.meta.changes ?? 0; + .all<{ rowid: number }>(); + const activeDeleted = active.results.length; if (activeDeleted >= limit) return activeDeleted; const inactive = await this.db .prepare( @@ -413,11 +421,12 @@ export class SqlResponsesSnapshotsRepo implements ResponsesSnapshotsRepo { ) ORDER BY stored.refreshed_at, stored.rowid LIMIT ? - )`, + ) + RETURNING rowid`, ) .bind(apiKeyId, limit - activeDeleted) - .run(); - return activeDeleted + (inactive.meta.changes ?? 0); + .all<{ rowid: number }>(); + return activeDeleted + inactive.results.length; } async findOldestRefreshedAt(apiKeyId: string): Promise {