From beb11d9d1d01fdbc0101894ca2e11d2aa180b9c3 Mon Sep 17 00:00:00 2001 From: Patrick Erichsen Date: Thu, 18 Jun 2026 17:19:17 -0700 Subject: [PATCH] feat: enforce convex retention policy --- .agents/skills/convex-retention/SKILL.md | 35 ++++ AGENTS.md | 1 + convex/_generated/api.d.ts | 4 + convex/crons.test.ts | 25 +++ convex/crons.ts | 17 +- convex/downloadMetrics.test.ts | 8 +- convex/downloadMetrics.ts | 3 +- convex/lib/retentionPolicy.test.ts | 42 ++++ convex/lib/retentionPolicy.ts | 245 +++++++++++++++++++++++ convex/rateLimits.ts | 3 +- convex/retention.test.ts | 173 ++++++++++++++++ convex/retention.ts | 85 ++++++++ convex/schema.ts | 19 ++ convex/telemetry.test.ts | 6 +- convex/telemetry.ts | 3 +- 15 files changed, 658 insertions(+), 11 deletions(-) create mode 100644 .agents/skills/convex-retention/SKILL.md create mode 100644 convex/lib/retentionPolicy.test.ts create mode 100644 convex/lib/retentionPolicy.ts create mode 100644 convex/retention.test.ts create mode 100644 convex/retention.ts diff --git a/.agents/skills/convex-retention/SKILL.md b/.agents/skills/convex-retention/SKILL.md new file mode 100644 index 0000000000..a3ebd261b6 --- /dev/null +++ b/.agents/skills/convex-retention/SKILL.md @@ -0,0 +1,35 @@ +--- +name: convex-retention +description: Use when adding or changing ClawHub Convex tables, TTL fields, cleanup crons, retention policy, auth/session cleanup, metric dedupe cleanup, or deprecated table removal +--- + +# Convex Retention + +## Overview + +ClawHub retention is code-owned. Every current Convex table must be classified in +`convex/lib/retentionPolicy.ts`, and ephemeral tables need an indexed, bounded cleanup path unless +their lifecycle is handled by usage-time validation or a documented component. + +## Checklist + +- Read `convex/_generated/ai/guidelines.md` and the Convex ops rules in `AGENTS.md` first. +- Add every new schema table to `RETENTION_POLICIES`; the `Record` type + is the enforcement gate. +- For ephemeral tables, prefer an explicit expiration field plus index, then prune with `.withIndex()` + and `.take(...)`. +- For new generic TTL tables, prefer `expirationTime` to match Convex Auth. Keep existing `expiresAt`, + `dayStart`, and `processedAt` fields unless that table already needs a real migration. +- Use `RETENTION_STANDARD_BATCH_SIZE` for ordinary retention jobs. Keep incident-tested special cases, + such as `skillStatEvents`, on their documented caps. +- Cron jobs should schedule bounded cleanup entrypoints only. Large one-off production migrations or + destructive backfills still start with `convex-migration-helper`. +- Do not bulk-clear active auth state. Expired `authSessions` and `authRefreshTokens` are pruned by + `convex/retention.ts`. + +## Verification + +- Add or update focused tests for policy classification and cleanup behavior. +- Run the focused Vitest slice for touched cleanup modules. +- Run `bunx convex codegen` after schema/API changes. +- Run a real Convex runtime check such as `bunx convex dev --once --typecheck=disable`. diff --git a/AGENTS.md b/AGENTS.md index 42449da7ba..480785e084 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -108,6 +108,7 @@ Specialized corpus, scanner, security-worker, UI proof, proof publishing, Crabbo ## Convex Migrations & Backfills - Any Convex production data migration, backfill, destructive cleanup, schema narrowing, or table reshaping must start with the `convex-migration-helper` skill. Default to `@convex-dev/migrations` for production data changes because it provides batching, dry runs, resume/progress tracking, and safer operator UX. Exceptions require an explicit note explaining why the component is unnecessary, plus equivalent dry-run support, cursor batching, resume/progress behavior, confirmation for destructive writes, and real Convex runtime validation. +- When adding or changing Convex tables, TTL fields, cleanup crons, retention policy, auth/session cleanup, metric dedupe cleanup, or deprecated table removal, use the repo-local `convex-retention` skill and update `convex/lib/retentionPolicy.ts`. - Use `convex/migrations.ts` for component-backed table-wide backfills; keep custom repairs, admin-gated operations, and incident-specific workflows in `convex/maintenance.ts`. - After a migration or cleanup is verified complete, remove temporary migration functions/code in a follow-up PR unless they are intentionally retained as ongoing maintenance tooling. diff --git a/convex/_generated/api.d.ts b/convex/_generated/api.d.ts index 1e7e088c31..0f05e8f9c0 100644 --- a/convex/_generated/api.d.ts +++ b/convex/_generated/api.d.ts @@ -97,6 +97,7 @@ import type * as lib_registryArtifactBackup from "../lib/registryArtifactBackup. import type * as lib_reporting from "../lib/reporting.js"; import type * as lib_reservedHandles from "../lib/reservedHandles.js"; import type * as lib_reservedSlugs from "../lib/reservedSlugs.js"; +import type * as lib_retentionPolicy from "../lib/retentionPolicy.js"; import type * as lib_searchText from "../lib/searchText.js"; import type * as lib_securityPrompt from "../lib/securityPrompt.js"; import type * as lib_skillBackfill from "../lib/skillBackfill.js"; @@ -134,6 +135,7 @@ import type * as registryArtifactBackups from "../registryArtifactBackups.js"; import type * as registryArtifactBackupsNode from "../registryArtifactBackupsNode.js"; import type * as registryArtifactRestore from "../registryArtifactRestore.js"; import type * as registryArtifactRestoreMutations from "../registryArtifactRestoreMutations.js"; +import type * as retention from "../retention.js"; import type * as search from "../search.js"; import type * as securityDataset from "../securityDataset.js"; import type * as securityDatasetNode from "../securityDatasetNode.js"; @@ -247,6 +249,7 @@ declare const fullApi: ApiFromModules<{ "lib/reporting": typeof lib_reporting; "lib/reservedHandles": typeof lib_reservedHandles; "lib/reservedSlugs": typeof lib_reservedSlugs; + "lib/retentionPolicy": typeof lib_retentionPolicy; "lib/searchText": typeof lib_searchText; "lib/securityPrompt": typeof lib_securityPrompt; "lib/skillBackfill": typeof lib_skillBackfill; @@ -284,6 +287,7 @@ declare const fullApi: ApiFromModules<{ registryArtifactBackupsNode: typeof registryArtifactBackupsNode; registryArtifactRestore: typeof registryArtifactRestore; registryArtifactRestoreMutations: typeof registryArtifactRestoreMutations; + retention: typeof retention; search: typeof search; securityDataset: typeof securityDataset; securityDatasetNode: typeof securityDatasetNode; diff --git a/convex/crons.test.ts b/convex/crons.test.ts index 88452e9f9b..17a82d5435 100644 --- a/convex/crons.test.ts +++ b/convex/crons.test.ts @@ -8,6 +8,8 @@ const mocks = vi.hoisted(() => { const installTelemetryDedupePruneRef = Symbol("install-telemetry-dedupe-prune"); const rateLimitCountersPruneRef = Symbol("rate-limit-counters-prune"); const skillStatEventPruneRef = Symbol("skill-stat-event-prune"); + const authSessionsPruneRef = Symbol("auth-sessions-prune"); + const authRefreshTokensPruneRef = Symbol("auth-refresh-tokens-prune"); return { interval, githubSkillSyncRef, @@ -15,6 +17,8 @@ const mocks = vi.hoisted(() => { installTelemetryDedupePruneRef, rateLimitCountersPruneRef, skillStatEventPruneRef, + authSessionsPruneRef, + authRefreshTokensPruneRef, }; }); @@ -63,6 +67,10 @@ vi.mock("./_generated/api", () => ({ rateLimits: { pruneRateLimitCountersInternal: mocks.rateLimitCountersPruneRef, }, + retention: { + pruneExpiredAuthSessionsInternal: mocks.authSessionsPruneRef, + pruneExpiredAuthRefreshTokensInternal: mocks.authRefreshTokensPruneRef, + }, }, })); @@ -140,6 +148,23 @@ describe("crons", () => { ); }); + it("prunes expired auth sessions and refresh tokens with the standard batch size", async () => { + await import("./crons"); + + expect(mocks.interval).toHaveBeenCalledWith( + "auth-session-retention-prune", + { hours: 1 }, + mocks.authSessionsPruneRef, + { batchSize: 500 }, + ); + expect(mocks.interval).toHaveBeenCalledWith( + "auth-refresh-token-retention-prune", + { hours: 6 }, + mocks.authRefreshTokensPruneRef, + { batchSize: 500 }, + ); + }); + it("prunes processed skill stat events daily with a seven-day retention window", async () => { await import("./crons"); diff --git a/convex/crons.ts b/convex/crons.ts index bd9da1566a..19e535e031 100644 --- a/convex/crons.ts +++ b/convex/crons.ts @@ -1,5 +1,6 @@ import { cronJobs } from "convex/server"; import { internal } from "./_generated/api"; +import { RETENTION_STANDARD_BATCH_SIZE } from "./lib/retentionPolicy"; const crons = cronJobs(); @@ -134,11 +135,25 @@ if (process.env.CLAWHUB_DISABLE_CRONS !== "1") { {}, ); + crons.interval( + "auth-session-retention-prune", + { hours: 1 }, + internal.retention.pruneExpiredAuthSessionsInternal, + { batchSize: RETENTION_STANDARD_BATCH_SIZE }, + ); + + crons.interval( + "auth-refresh-token-retention-prune", + { hours: 6 }, + internal.retention.pruneExpiredAuthRefreshTokensInternal, + { batchSize: RETENTION_STANDARD_BATCH_SIZE }, + ); + crons.interval( "rate-limit-counters-prune", { minutes: 15 }, internal.rateLimits.pruneRateLimitCountersInternal, - { batchSize: 500 }, + { batchSize: RETENTION_STANDARD_BATCH_SIZE }, ); } diff --git a/convex/downloadMetrics.test.ts b/convex/downloadMetrics.test.ts index 00f4c2ea1b..9539606ca8 100644 --- a/convex/downloadMetrics.test.ts +++ b/convex/downloadMetrics.test.ts @@ -227,7 +227,7 @@ describe("download metric helpers", () => { expect(result).toEqual({ deleted: 2, hasMore: false }); expect(indexCalls[0]?.table).toBe("downloadMetricDedupes"); expect(indexCalls[0]?.indexName).toBe("by_day"); - expect(take).toHaveBeenCalledWith(200); + expect(take).toHaveBeenCalledWith(500); expect(delete_).toHaveBeenCalledWith("downloadMetricDedupes:one"); expect(delete_).toHaveBeenCalledWith("downloadMetricDedupes:two"); }); @@ -258,7 +258,7 @@ describe("download metric helpers", () => { it("reschedules stale dedupe pruning when one bounded batch fills", async () => { vi.setSystemTime(30 * 86_400_000); - const rows = Array.from({ length: 200 }, (_, index) => ({ + const rows = Array.from({ length: 500 }, (_, index) => ({ _id: `downloadMetricDedupes:${index}`, })); const { db, delete_ } = makeDb({}, { downloadMetricDedupes: rows }); @@ -266,8 +266,8 @@ describe("download metric helpers", () => { const result = await pruneDownloadMetricDedupesHandler({ db, scheduler: { runAfter } }, {}); - expect(result).toEqual({ deleted: 200, hasMore: true }); - expect(delete_).toHaveBeenCalledTimes(200); + expect(result).toEqual({ deleted: 500, hasMore: true }); + expect(delete_).toHaveBeenCalledTimes(500); expect(runAfter).toHaveBeenCalledWith(0, expect.anything(), {}); }); }); diff --git a/convex/downloadMetrics.ts b/convex/downloadMetrics.ts index 40a2af7484..466e2f7b38 100644 --- a/convex/downloadMetrics.ts +++ b/convex/downloadMetrics.ts @@ -3,12 +3,13 @@ import { internal } from "./_generated/api"; import type { Id } from "./_generated/dataModel"; import { internalMutation } from "./functions"; import { getClientIp } from "./lib/httpRateLimit"; +import { RETENTION_STANDARD_BATCH_SIZE } from "./lib/retentionPolicy"; import { hashToken } from "./lib/tokens"; import { insertStatEvent } from "./skillStatEvents"; const DAY_MS = 86_400_000; const DEDUPE_RETENTION_MS = 14 * DAY_MS; -const PRUNE_BATCH_SIZE = 200; +const PRUNE_BATCH_SIZE = RETENTION_STANDARD_BATCH_SIZE; const identityKindValidator = v.union(v.literal("user"), v.literal("ip")); diff --git a/convex/lib/retentionPolicy.test.ts b/convex/lib/retentionPolicy.test.ts new file mode 100644 index 0000000000..2d0e367416 --- /dev/null +++ b/convex/lib/retentionPolicy.test.ts @@ -0,0 +1,42 @@ +/* @vitest-environment node */ +import { describe, expect, it } from "vitest"; +import { + RETENTION_POLICIES, + RETENTION_STANDARD_BATCH_SIZE, + getRetentionPolicy, +} from "./retentionPolicy"; + +describe("retention policies", () => { + it("classifies auth session tables as indexed expiring tables", () => { + expect(getRetentionPolicy("authSessions")).toMatchObject({ + classification: "ephemeral", + expirationField: "expirationTime", + expirationIndex: "by_expiration_time", + prune: "retention.pruneExpiredAuthSessionsInternal", + }); + expect(getRetentionPolicy("authRefreshTokens")).toMatchObject({ + classification: "ephemeral", + expirationField: "expirationTime", + expirationIndex: "by_expiration_time", + prune: "retention.pruneExpiredAuthRefreshTokensInternal", + }); + }); + + it("uses one standard batch size for retention jobs", () => { + expect(RETENTION_STANDARD_BATCH_SIZE).toBe(500); + const batchSizes = Object.values(RETENTION_POLICIES) + .filter((policy) => policy.classification === "ephemeral") + .map((policy) => policy.standardBatchSize); + + expect(batchSizes.length).toBeGreaterThan(0); + expect(new Set(batchSizes)).toEqual(new Set([RETENTION_STANDARD_BATCH_SIZE])); + }); + + it("documents active expiring operational tables", () => { + expect(getRetentionPolicy("rateLimitCounters")).toMatchObject({ + classification: "ephemeral", + expirationField: "expiresAt", + expirationIndex: "by_expires_at", + }); + }); +}); diff --git a/convex/lib/retentionPolicy.ts b/convex/lib/retentionPolicy.ts new file mode 100644 index 0000000000..fa45b67e9c --- /dev/null +++ b/convex/lib/retentionPolicy.ts @@ -0,0 +1,245 @@ +import type { TableNames } from "../_generated/dataModel"; + +export const RETENTION_STANDARD_BATCH_SIZE = 500; + +type BaseRetentionPolicy = { + reason: string; +}; + +type PermanentRetentionPolicy = BaseRetentionPolicy & { + classification: "permanent"; +}; + +type DerivedRetentionPolicy = BaseRetentionPolicy & { + classification: "derived"; + rebuildSource: string; +}; + +type DeprecatedRetentionPolicy = BaseRetentionPolicy & { + classification: "deprecated"; + replacementTable?: TableNames; + removalIssue?: string; +}; + +type EphemeralRetentionPolicy = BaseRetentionPolicy & { + classification: "ephemeral"; + standardBatchSize: typeof RETENTION_STANDARD_BATCH_SIZE; + prune: string; + expirationField?: "expiresAt" | "expirationTime" | "dayStart" | "processedAt" | "createdAt"; + expirationIndex?: string; + retention: string; +}; + +export type RetentionPolicy = + | PermanentRetentionPolicy + | DerivedRetentionPolicy + | DeprecatedRetentionPolicy + | EphemeralRetentionPolicy; + +const permanent = (reason: string): PermanentRetentionPolicy => ({ + classification: "permanent", + reason, +}); + +const derived = (reason: string, rebuildSource: string): DerivedRetentionPolicy => ({ + classification: "derived", + reason, + rebuildSource, +}); + +const ephemeral = ( + reason: string, + options: Omit, +): EphemeralRetentionPolicy => ({ + classification: "ephemeral", + reason, + standardBatchSize: RETENTION_STANDARD_BATCH_SIZE, + ...options, +}); + +export const RETENTION_POLICIES = { + users: permanent("Canonical user profiles and account state."), + authSessions: ephemeral("Convex Auth sessions expire after their total session duration.", { + expirationField: "expirationTime", + expirationIndex: "by_expiration_time", + prune: "retention.pruneExpiredAuthSessionsInternal", + retention: "Convex Auth session total duration.", + }), + authAccounts: permanent("Provider account links for active users."), + authRefreshTokens: ephemeral("Convex Auth refresh tokens expire after inactive duration.", { + expirationField: "expirationTime", + expirationIndex: "by_expiration_time", + prune: "retention.pruneExpiredAuthRefreshTokensInternal", + retention: "Convex Auth inactive session duration.", + }), + authVerificationCodes: ephemeral("One-time verification codes expire by timestamp.", { + expirationField: "expirationTime", + prune: "convex-auth internal validation", + retention: "Provider code expiration.", + }), + authVerifiers: ephemeral("OAuth PKCE verifier rows are temporary sign-in state.", { + prune: "convex-auth sign-in/session cleanup", + retention: "OAuth verifier lifecycle.", + }), + authRateLimits: ephemeral("Convex Auth OTP/password rate-limit rows are operational state.", { + prune: "convex-auth internal rate-limit lifecycle", + retention: "Auth provider rate-limit window.", + }), + publishers: permanent("Canonical publisher profiles."), + publisherMembers: permanent("Canonical publisher membership records."), + officialPublishers: permanent("Manual official publisher assignments."), + githubSkillSources: permanent("Tracked GitHub source configuration."), + githubSkillContents: derived("Cached GitHub source content snapshots.", "githubSkillSources"), + githubSkillScans: derived("Cached GitHub source scan state.", "githubSkillSources"), + skills: permanent("Canonical skill records."), + skillSlugAliases: permanent("Historical slug routing aliases."), + packages: permanent("Canonical package records."), + packageReleases: permanent("Canonical package release records."), + catalogClassificationResults: derived( + "Catalog classification output can be recomputed from package and skill metadata.", + "skills/packages", + ), + packageInspectorWarnings: permanent("Package inspector findings are user-facing review history."), + packageInspectorFindingNotifications: permanent( + "Notification sent-log prevents duplicate emails.", + ), + packageInspectorScanCursors: permanent("Package inspector scan progress cursor."), + securityScanJobs: permanent("Security scan job history and current processing state."), + skillScanRequests: ephemeral( + "Uploaded or GitHub scan requests expire and delete temporary blobs.", + { + expirationField: "expiresAt", + expirationIndex: "by_expires_at", + prune: "securityScan.pruneExpiredSkillScanRequestsInternal", + retention: "Scan request TTL.", + }, + ), + skillScanRequestFileChunks: ephemeral("Temporary chunk rows owned by expiring scan requests.", { + prune: "securityScan.pruneExpiredSkillScanRequestsInternal", + retention: "Parent skillScanRequests TTL.", + }), + skillCardGenerationJobs: permanent("Card generation job history and retry state."), + packageStatEvents: ephemeral("Package stat event log only needs to survive processing.", { + expirationField: "processedAt", + expirationIndex: "by_unprocessed", + prune: "pending packageStatEvents retention work", + retention: "After stat processing succeeds.", + }), + packageTrustedPublishers: permanent("Trusted publishing configuration."), + packagePublishTokens: ephemeral("Package publish tokens expire and can be revoked.", { + expirationField: "expiresAt", + prune: "usage-time validation plus pending retention cleanup", + retention: "Publish token expiry.", + }), + packagePublishUploadTickets: ephemeral("Upload tickets expire shortly after creation.", { + expirationField: "expiresAt", + prune: "usage-time validation plus pending retention cleanup", + retention: "Upload ticket TTL.", + }), + packageBadges: permanent("Curated package badges."), + packageSearchDigest: derived("Search projection of package state.", "packages"), + packageTopicSearchDigest: derived("Topic search projection of package state.", "packages"), + packagePluginCategorySearchDigest: derived( + "Plugin category search projection of package state.", + "packages", + ), + skillVersions: permanent("Canonical skill version records."), + skillVersionFingerprints: derived("Fingerprint projection of skill versions.", "skillVersions"), + skillBadges: permanent("Curated skill badges."), + skillEmbeddings: derived("Search embedding projection of skill versions.", "skillVersions"), + embeddingSkillMap: derived("Lookup map for embedding rows.", "skillEmbeddings"), + skillSearchDigest: derived("Search projection of skill state.", "skills"), + curatedSkillSearchDigest: derived("Curated search projection of skill state.", "skills"), + skillTopicSearchDigest: derived("Topic search projection of skill state.", "skills"), + skillDailyStats: permanent("Daily aggregate stats are product analytics."), + skillLeaderboards: derived("Leaderboard snapshots can be rebuilt from stats.", "skillDailyStats"), + skillStatBackfillState: permanent("Backfill cursor state."), + globalStats: derived("Global stats aggregate can be recalculated.", "skills/packages"), + skillStatEvents: ephemeral( + "Skill stat event log is retained only after both consumers pass it.", + { + expirationField: "processedAt", + expirationIndex: "by_unprocessed", + prune: "skillStatEvents.pruneProcessedSkillStatEventsInternal", + retention: "Processed and older than 7 days, capped by stat cursor.", + }, + ), + skillStatUpdateCursors: permanent("Stat processing cursor state."), + skillStatDocSyncLeases: ephemeral("Short-lived stat sync leases.", { + prune: "lease overwrite/expiry semantics", + retention: "Lease duration.", + }), + skillReports: permanent("Moderation reports and audit history."), + skillAppeals: permanent("Moderation appeals and audit history."), + skillModerationEventLogs: permanent("Moderation event audit log."), + packageReports: permanent("Package moderation reports and audit history."), + packageAppeals: permanent("Package moderation appeals and audit history."), + packageModerationEventLogs: permanent("Package moderation event audit log."), + officialPluginMigrations: permanent("Official plugin migration state."), + stars: permanent("User star records."), + auditLogs: permanent("Audit logs are durable compliance/security history."), + publisherAbuseScoreRuns: permanent("Abuse scoring run history."), + publisherAbuseScores: permanent("Abuse score history used for review decisions."), + publisherAbuseReviewNominations: permanent("Abuse review workflow state."), + publisherAbuseReviewEvents: permanent("Abuse review event history."), + vtScanLogs: permanent("VirusTotal scan log history."), + apiTokens: permanent("User API tokens until revoked."), + cliDeviceCodes: ephemeral("CLI device codes expire quickly.", { + expirationField: "expiresAt", + expirationIndex: "by_status_expires", + prune: "usage-time expiry plus pending retention cleanup", + retention: "Device code TTL.", + }), + rateLimitCounters: ephemeral("Active rate-limit counters expire after their rate-limit window.", { + expirationField: "expiresAt", + expirationIndex: "by_expires_at", + prune: "rateLimits.pruneRateLimitCountersInternal", + retention: "Rate-limit window plus buffer.", + }), + downloadMetricDedupes: ephemeral( + "Download dedupe rows are only needed for recent metric windows.", + { + expirationField: "dayStart", + expirationIndex: "by_day", + prune: "downloadMetrics.pruneDownloadMetricDedupesInternal", + retention: "14 days.", + }, + ), + packageInstallMetricDedupes: ephemeral( + "Package install dedupe rows are only needed for recent metric windows.", + { + expirationField: "dayStart", + expirationIndex: "by_day", + prune: "downloadMetrics.pruneDownloadMetricDedupesInternal", + retention: "14 days.", + }, + ), + installTelemetryDedupes: ephemeral( + "Install telemetry dedupe rows are only needed for recent metric windows.", + { + expirationField: "dayStart", + expirationIndex: "by_day", + prune: "telemetry.pruneInstallTelemetryDedupesInternal", + retention: "14 days.", + }, + ), + reservedSlugs: ephemeral("Deleted slug reservations release after the cooldown window.", { + expirationField: "expiresAt", + expirationIndex: "by_expiry", + prune: "usage-time release plus pending retention cleanup", + retention: "Slug reservation cooldown.", + }), + reservedHandles: permanent("Reserved handles are explicit policy records until released."), + registryArtifactBackupSyncState: permanent("Registry artifact backup cursor state."), + registryArtifactBackupJobs: permanent("Registry artifact backup job history and retry state."), + userSkillInstalls: permanent("Current user install records."), + skillOwnershipTransfers: ephemeral("Ownership transfer invitations expire.", { + expirationField: "expiresAt", + prune: "usage-time validation plus pending retention cleanup", + retention: "Transfer invitation TTL.", + }), +} satisfies Record; + +export function getRetentionPolicy(tableName: TableNames) { + return RETENTION_POLICIES[tableName]; +} diff --git a/convex/rateLimits.ts b/convex/rateLimits.ts index decd142b7d..e6086d3235 100644 --- a/convex/rateLimits.ts +++ b/convex/rateLimits.ts @@ -2,9 +2,10 @@ import { v } from "convex/values"; import { internal } from "./_generated/api"; import { internalMutation, internalQuery } from "./functions"; import { RATE_LIMIT_COUNTER_SHARDS } from "./lib/rateLimitConfig"; +import { RETENTION_STANDARD_BATCH_SIZE } from "./lib/retentionPolicy"; const RATE_LIMIT_COUNTER_RETENTION_BUFFER_MS = 5 * 60_000; -const DEFAULT_PRUNE_RATE_LIMIT_COUNTERS_BATCH_SIZE = 200; +const DEFAULT_PRUNE_RATE_LIMIT_COUNTERS_BATCH_SIZE = RETENTION_STANDARD_BATCH_SIZE; const MAX_PRUNE_RATE_LIMIT_COUNTERS_BATCH_SIZE = 1_000; /** diff --git a/convex/retention.test.ts b/convex/retention.test.ts new file mode 100644 index 0000000000..421da92f4c --- /dev/null +++ b/convex/retention.test.ts @@ -0,0 +1,173 @@ +/* @vitest-environment node */ +import { describe, expect, it, vi } from "vitest"; + +const retentionRefs = vi.hoisted(() => ({ + pruneExpiredAuthSessionsInternal: Symbol("pruneExpiredAuthSessionsInternal"), + pruneExpiredAuthRefreshTokensInternal: Symbol("pruneExpiredAuthRefreshTokensInternal"), +})); + +vi.mock("./_generated/api", () => ({ + internal: { + retention: retentionRefs, + }, +})); + +const { pruneExpiredAuthRefreshTokensInternal, pruneExpiredAuthSessionsInternal } = + await import("./retention"); + +type WrappedHandler = { + _handler: (ctx: unknown, args: TArgs) => Promise; +}; + +const pruneSessionsHandler = ( + pruneExpiredAuthSessionsInternal as unknown as WrappedHandler< + { batchSize?: number }, + { deletedSessions: number; deletedRefreshTokens: number; hasMore: boolean } + > +)._handler; +const pruneTokensHandler = ( + pruneExpiredAuthRefreshTokensInternal as unknown as WrappedHandler< + { batchSize?: number }, + { deleted: number; hasMore: boolean } + > +)._handler; + +function makeDb(base: { query: ReturnType; delete: ReturnType }) { + return { + ...base, + get: vi.fn(), + insert: vi.fn(), + patch: vi.fn(), + replace: vi.fn(), + normalizeId: vi.fn(), + system: { + get: vi.fn(), + query: vi.fn(), + }, + }; +} + +describe("auth retention", () => { + it("deletes expired sessions and their refresh tokens in bounded batches", async () => { + const now = 1_000_000; + vi.spyOn(Date, "now").mockReturnValue(now); + const runAfter = vi.fn(); + const deleteDoc = vi.fn(); + const tokenRows = [{ _id: "authRefreshTokens:one" }, { _id: "authRefreshTokens:two" }]; + const sessionRows = [{ _id: "authSessions:expired", expirationTime: now - 1 }]; + const indexCalls: Array<{ table: string; indexName: string; field: string; value: number }> = + []; + + const ctx = { + db: makeDb({ + query: vi.fn((table: string) => ({ + withIndex: vi.fn((indexName: string, build: (q: unknown) => unknown) => { + const q = { + lt: vi.fn((field: string, value: number) => { + indexCalls.push({ table, indexName, field, value }); + return q; + }), + eq: vi.fn(() => q), + }; + build(q); + return { + take: vi.fn(async (limit?: number) => + table === "authSessions" ? sessionRows : tokenRows.slice(0, limit), + ), + }; + }), + })), + delete: deleteDoc, + }), + scheduler: { runAfter }, + }; + + const result = await pruneSessionsHandler(ctx as never, {}); + + expect(result).toEqual({ + deletedSessions: 1, + deletedRefreshTokens: 2, + hasMore: false, + }); + expect(indexCalls).toContainEqual({ + table: "authSessions", + indexName: "by_expiration_time", + field: "expirationTime", + value: now, + }); + expect(deleteDoc).toHaveBeenCalledWith("authRefreshTokens:one"); + expect(deleteDoc).toHaveBeenCalledWith("authRefreshTokens:two"); + expect(deleteDoc).toHaveBeenCalledWith("authSessions:expired"); + expect(runAfter).not.toHaveBeenCalled(); + }); + + it("keeps an expired session until its refresh tokens are drained", async () => { + const now = 1_000_000; + vi.spyOn(Date, "now").mockReturnValue(now); + const runAfter = vi.fn(); + const deleteDoc = vi.fn(); + const tokenRows = [{ _id: "authRefreshTokens:one" }]; + const sessionRows = [{ _id: "authSessions:expired", expirationTime: now - 1 }]; + + const ctx = { + db: makeDb({ + query: vi.fn((table: string) => ({ + withIndex: vi.fn((_indexName: string, build: (q: unknown) => unknown) => { + const q = { + lt: vi.fn(() => q), + eq: vi.fn(() => q), + }; + build(q); + return { + take: vi.fn(async (limit?: number) => + table === "authSessions" ? sessionRows : tokenRows.slice(0, limit), + ), + }; + }), + })), + delete: deleteDoc, + }), + scheduler: { runAfter }, + }; + + const result = await pruneSessionsHandler(ctx as never, { batchSize: 1 }); + + expect(result).toEqual({ + deletedSessions: 0, + deletedRefreshTokens: 1, + hasMore: true, + }); + expect(deleteDoc).toHaveBeenCalledWith("authRefreshTokens:one"); + expect(deleteDoc).not.toHaveBeenCalledWith("authSessions:expired"); + expect(runAfter).toHaveBeenCalledWith(0, retentionRefs.pruneExpiredAuthSessionsInternal, { + batchSize: 1, + }); + }); + + it("deletes expired orphan refresh tokens by expiration time", async () => { + const now = 2_000_000; + vi.spyOn(Date, "now").mockReturnValue(now); + const deleteDoc = vi.fn(); + const rows = [{ _id: "authRefreshTokens:expired", expirationTime: now - 1 }]; + const lt = vi.fn(() => ({})); + + const ctx = { + db: makeDb({ + query: vi.fn(() => ({ + withIndex: vi.fn((_indexName: string, build: (q: unknown) => unknown) => { + build({ lt }); + return { take: vi.fn(async () => rows) }; + }), + })), + delete: deleteDoc, + }), + scheduler: { runAfter: vi.fn() }, + }; + + const result = await pruneTokensHandler(ctx as never, {}); + + expect(result).toEqual({ deleted: 1, hasMore: false }); + expect(lt).toHaveBeenCalledWith("expirationTime", now); + expect(deleteDoc).toHaveBeenCalledWith("authRefreshTokens:expired"); + }); +}); diff --git a/convex/retention.ts b/convex/retention.ts new file mode 100644 index 0000000000..dac96aff8f --- /dev/null +++ b/convex/retention.ts @@ -0,0 +1,85 @@ +import { v } from "convex/values"; +import { internal } from "./_generated/api"; +import { internalMutation } from "./functions"; +import { RETENTION_STANDARD_BATCH_SIZE } from "./lib/retentionPolicy"; + +const RETENTION_MAX_BATCH_SIZE = 1_000; + +function normalizeRetentionBatchSize(batchSize: number | undefined) { + const requested = Number.isFinite(batchSize) + ? Math.floor(batchSize ?? RETENTION_STANDARD_BATCH_SIZE) + : RETENTION_STANDARD_BATCH_SIZE; + return Math.max(1, Math.min(requested, RETENTION_MAX_BATCH_SIZE)); +} + +export const pruneExpiredAuthSessionsInternal = internalMutation({ + args: { + batchSize: v.optional(v.number()), + }, + handler: async (ctx, args) => { + const batchSize = normalizeRetentionBatchSize(args.batchSize); + const now = Date.now(); + const sessions = await ctx.db + .query("authSessions") + .withIndex("by_expiration_time", (q) => q.lt("expirationTime", now)) + .take(batchSize); + + let deletedSessions = 0; + let deletedRefreshTokens = 0; + let deletedDocuments = 0; + for (const session of sessions) { + const remainingBatchSize = batchSize - deletedDocuments; + if (remainingBatchSize <= 0) break; + const refreshTokens = await ctx.db + .query("authRefreshTokens") + .withIndex("sessionId", (q) => q.eq("sessionId", session._id)) + .take(remainingBatchSize); + for (const refreshToken of refreshTokens) { + await ctx.db.delete(refreshToken._id); + deletedRefreshTokens += 1; + deletedDocuments += 1; + } + + if (refreshTokens.length === remainingBatchSize) break; + + await ctx.db.delete(session._id); + deletedSessions += 1; + deletedDocuments += 1; + } + + const hasMore = sessions.length === batchSize || deletedDocuments >= batchSize; + if (hasMore) { + await ctx.scheduler.runAfter(0, internal.retention.pruneExpiredAuthSessionsInternal, { + batchSize, + }); + } + + return { deletedSessions, deletedRefreshTokens, hasMore }; + }, +}); + +export const pruneExpiredAuthRefreshTokensInternal = internalMutation({ + args: { + batchSize: v.optional(v.number()), + }, + handler: async (ctx, args) => { + const batchSize = normalizeRetentionBatchSize(args.batchSize); + const stale = await ctx.db + .query("authRefreshTokens") + .withIndex("by_expiration_time", (q) => q.lt("expirationTime", Date.now())) + .take(batchSize); + + for (const refreshToken of stale) { + await ctx.db.delete(refreshToken._id); + } + + const hasMore = stale.length === batchSize; + if (hasMore) { + await ctx.scheduler.runAfter(0, internal.retention.pruneExpiredAuthRefreshTokensInternal, { + batchSize, + }); + } + + return { deleted: stale.length, hasMore }; + }, +}); diff --git a/convex/schema.ts b/convex/schema.ts index e593abfac6..4c4f0a0655 100644 --- a/convex/schema.ts +++ b/convex/schema.ts @@ -194,6 +194,23 @@ const users = defineTable({ .index("by_deactivated_purged_at", ["deactivatedAt", "purgedAt"]) .index("by_active_handle", ["deletedAt", "deactivatedAt", "handle"]); +const authSessions = defineTable({ + userId: v.id("users"), + expirationTime: v.number(), +}) + .index("userId", ["userId"]) + .index("by_expiration_time", ["expirationTime"]); + +const authRefreshTokens = defineTable({ + sessionId: v.id("authSessions"), + expirationTime: v.number(), + firstUsedTime: v.optional(v.number()), + parentRefreshTokenId: v.optional(v.id("authRefreshTokens")), +}) + .index("sessionId", ["sessionId"]) + .index("sessionIdAndParentRefreshTokenId", ["sessionId", "parentRefreshTokenId"]) + .index("by_expiration_time", ["expirationTime"]); + const publishers = defineTable({ kind: v.union(v.literal("user"), v.literal("org")), handle: v.string(), @@ -2654,6 +2671,8 @@ const skillOwnershipTransfers = defineTable({ export default defineSchema({ ...authTables, + authSessions, + authRefreshTokens, users, publishers, publisherMembers, diff --git a/convex/telemetry.test.ts b/convex/telemetry.test.ts index d2d788c5b2..d774fce98c 100644 --- a/convex/telemetry.test.ts +++ b/convex/telemetry.test.ts @@ -392,7 +392,7 @@ describe("telemetry install events", () => { const result = await pruneInstallTelemetryDedupesHandler(ctx); - expect(take).toHaveBeenCalledWith(200); + expect(take).toHaveBeenCalledWith(500); expect(result).toEqual({ deleted: 2, hasMore: false }); expect(deleteDoc).toHaveBeenCalledWith("installTelemetryDedupes:one"); expect(deleteDoc).toHaveBeenCalledWith("installTelemetryDedupes:two"); @@ -400,7 +400,7 @@ describe("telemetry install events", () => { it("reschedules stale dedupe pruning when one bounded batch fills", async () => { vi.setSystemTime(20 * 86_400_000); - const stale = Array.from({ length: 200 }, (_, index) => ({ + const stale = Array.from({ length: 500 }, (_, index) => ({ _id: `installTelemetryDedupes:${index}`, })); const runAfter = vi.fn(); @@ -416,7 +416,7 @@ describe("telemetry install events", () => { const result = await pruneInstallTelemetryDedupesHandler(ctx); - expect(result).toEqual({ deleted: 200, hasMore: true }); + expect(result).toEqual({ deleted: 500, hasMore: true }); expect(runAfter).toHaveBeenCalledWith( 0, telemetryRefs.pruneInstallTelemetryDedupesInternal, diff --git a/convex/telemetry.ts b/convex/telemetry.ts index ed78004992..18f2ab091c 100644 --- a/convex/telemetry.ts +++ b/convex/telemetry.ts @@ -4,11 +4,12 @@ import type { Id } from "./_generated/dataModel"; import type { MutationCtx } from "./_generated/server"; import { internalMutation, mutation } from "./functions"; import { requireUser } from "./lib/access"; +import { RETENTION_STANDARD_BATCH_SIZE } from "./lib/retentionPolicy"; import { insertStatEvent } from "./skillStatEvents"; const DAY_MS = 86_400_000; const INSTALL_TELEMETRY_DEDUPE_RETENTION_MS = 14 * DAY_MS; -const PRUNE_BATCH_SIZE = 200; +const PRUNE_BATCH_SIZE = RETENTION_STANDARD_BATCH_SIZE; const CLEAR_INSTALLS_BATCH_SIZE = 5_000; const CLEAR_DEDUPES_BATCH_SIZE = 10_000;