From a9e47c6e592c6bc6e4b2f73cfc87c6422c526b43 Mon Sep 17 00:00:00 2001 From: Abdullah Sultan Date: Tue, 4 Aug 2026 03:09:30 -0500 Subject: [PATCH] feat: store operation content once and enforce per-plan history retention `operations` and `operation_files` were the only unbounded tables, and every file body was written three times: once in `operations.event` (the envelope, whose payload is the transaction), again in `operations.transaction`, and again in `operation_files.payload`. Migration 012 drops the latter two, making the envelope the single home for content and leaving `operation_files` as the per-path index into it. GET /v1/operations is byte-identical either way, since jsonb canonicalizes a value the same wherever it is stored. Retention is now enforced against PLAN_LIMITS[plan].historyRetentionDays. Pruning `operations` used to be unsafe because cursor reconnect cannot tell a truncated list from "caught up", so the sweep deletes strictly a prefix of each workspace's sequence and records how far it reached in workspaces.operations_pruned_through. A cursor below that watermark is answered with an explicit cursor-too-old resync status instead of a short page, and with 410 Gone for daemons that predate the status, so an older client cannot read it as success. The daemon adopts the watermark, logs the gap, and surfaces it in `status().service`. The sweep runs on a service-side interval with its own privileged connection, because the request-serving role deliberately cannot delete operations; `pnpm service:prune` remains the manual tool. Co-Authored-By: Claude Opus 5 (1M context) --- BUILD_INSTRUCTIONS.md | 24 +-- README.md | 4 +- apps/daemon/src/client.ts | 2 +- apps/daemon/src/index.test.ts | 81 +++++++++- apps/daemon/src/index.ts | 69 ++++++-- apps/daemon/src/local-event.ts | 8 + apps/daemon/src/reconnect.integration.test.ts | 65 +++++++- apps/daemon/src/service-client.ts | 18 ++- .../012_single_content_home_and_retention.sql | 26 +++ apps/service/src/billing.ts | 4 +- apps/service/src/http.test.ts | 37 ++++- apps/service/src/http.ts | 24 ++- apps/service/src/main.ts | 39 +++++ apps/service/src/prune.test.ts | 145 ++++++++++++++++- apps/service/src/prune.ts | 25 ++- apps/service/src/retention.test.ts | 62 +++++++ apps/service/src/retention.ts | 60 +++++++ apps/service/src/store.integration.test.ts | 84 +++++++++- apps/service/src/store.ts | 152 +++++++++++++++--- docs/protocol.md | 26 +++ docs/security.md | 5 +- packages/protocol/src/index.test.ts | 15 ++ packages/protocol/src/index.ts | 32 ++++ 23 files changed, 947 insertions(+), 60 deletions(-) create mode 100644 apps/service/migrations/012_single_content_home_and_retention.sql create mode 100644 apps/service/src/retention.test.ts create mode 100644 apps/service/src/retention.ts diff --git a/BUILD_INSTRUCTIONS.md b/BUILD_INSTRUCTIONS.md index 013df0d..e1bbe6b 100644 --- a/BUILD_INSTRUCTIONS.md +++ b/BUILD_INSTRUCTIONS.md @@ -168,8 +168,8 @@ running at a loss; it is not the optimization target. The axes actually used: -- **History retention** — bounds `operation_files.payload`, the only table that grows - without limit. This is the real cost governor. +- **History retention** — bounds `operations`, the only table that grows without limit. + This is the real cost governor. - **Autonomy tier availability** — auto-always (Phase 9) is the "I trust it now" moment, which is the honest point to ask for money. - **Org controls** — SSO, audit export, SLA. What organizations actually buy. @@ -206,17 +206,21 @@ ever stop working. **Not yet done:** - The Stripe account itself, and therefore any real checkout. -- **Retention is declared but not enforced.** `PLAN_LIMITS[plan].historyRetentionDays` is - surfaced to users and is the intended cost governor, but nothing prunes against it yet. - This is not a small wiring job: `apps/service/src/prune.ts` deliberately refuses to touch - `operations` because cursor-based reconnect lets a long-offline replica download - everything after its last-known cursor, and age-pruning would silently break that - guarantee. A safe implementation needs either (a) pruning only below the minimum cursor - across live replicas, or (b) a defined "cursor too old, resync from Git" protocol - response. Until one of those exists, storage is unbounded on every plan. - Per-seat pricing mechanics for Team (the plan enforces Unlimited's caps today; the per-seat *charge* has no implementation because there is no billing provider). +**Retention is enforced** (option (b) of the two designs sketched here previously). +`PgStore.pruneOperationsByRetention()` deletes each workspace's operations outside +`PLAN_LIMITS[plan].historyRetentionDays` and records how far it reached in +`workspaces.operations_pruned_through`. `GET /v1/operations` answers a cursor below that +watermark with an explicit `cursor-too-old` resync status (`410 Gone` for daemons that +predate it) instead of a truncated page, and the daemon adopts the watermark and reports +the gap — see docs/protocol.md. The sweep runs on a service-side interval configured with +`CROSSCODE_RETENTION_DATABASE_URL` (the request-serving role deliberately cannot delete +operations); `pnpm service:prune` runs it manually. Content is also no longer stored +twice: `operations.event` is the single home of a transaction's file bodies, and +`operation_files` is a per-path index into it. + `assertSemanticReviewCallAvailable` is deliberately left unwired rather than pending: semantic review is delegated to the member's own MCP agent and never reaches the service, so there is no per-call cost to meter, and wiring it would mean adding a network round-trip before every local review purely to bill for it (see the comment above `incrementSemanticReviewUsage` in `apps/service/src/billing.ts`). Every plan now carries an unlimited cap for it, so it can never fire. ### Phase 11 — Pairing a checkout to an account (backend v1 shipped) diff --git a/README.md b/README.md index 554a7c1..fc30c07 100644 --- a/README.md +++ b/README.md @@ -223,7 +223,7 @@ Set on the host: `DATABASE_URL`, `SUPABASE_URL`, and — when a proxy in front t product, which means no browser origin may call the API cross-origin; it exists for anyone building their own browser client against the service. -Run `pnpm service:migrate` with a migration-owner connection before starting a new service version. `CROSSCODE_RUNTIME_DB_ROLE` applies the required least-privilege grants, and service startup refuses a role that can update/delete immutable operations or audit rows. The runtime never executes DDL. Non-loopback PostgreSQL URLs must specify exactly one `sslmode=verify-full` and cannot use host/SSL query overrides. For local-only testing against a plain (non-Supabase) Postgres instance, `infra/docker-compose.yml` still starts one on `127.0.0.1:5432`; it is not used in production, where `DATABASE_URL` points at Supabase. +Run `pnpm service:migrate` with a migration-owner connection before starting a new service version. Set `CROSSCODE_RETENTION_DATABASE_URL` (optionally `CROSSCODE_RETENTION_SWEEP_MINUTES`, default 60) to that same privileged connection to enable the scheduled history-retention sweep; the least-privilege runtime role cannot delete operations, so without it retention only runs when an admin invokes `pnpm service:prune`. `CROSSCODE_RUNTIME_DB_ROLE` applies the required least-privilege grants, and service startup refuses a role that can update/delete immutable operations or audit rows. The runtime never executes DDL. Non-loopback PostgreSQL URLs must specify exactly one `sslmode=verify-full` and cannot use host/SSL query overrides. For local-only testing against a plain (non-Supabase) Postgres instance, `infra/docker-compose.yml` still starts one on `127.0.0.1:5432`; it is not used in production, where `DATABASE_URL` points at Supabase. ## Workspaces, members, and invites (CLI and API only) @@ -433,7 +433,7 @@ For the implementation plan and current milestone ledger, see [BUILD_INSTRUCTION ## Current limitations -- Production PostgreSQL role grants still need environment-specific deployment hardening. Retention is opt-in and admin-only: `pnpm service:prune -- --older-than-days ` deletes old audit events and ended sessions; cursor-reconnect-dependent tables are deliberately never pruned. +- Production PostgreSQL role grants still need environment-specific deployment hardening. Operation history is pruned to the workspace plan's `historyRetentionDays` — on a service-side schedule when `CROSSCODE_RETENTION_DATABASE_URL` names a role that may delete, and on demand via `pnpm service:prune`, which also deletes audit events and ended sessions older than `--older-than-days `. A replica whose cursor falls outside the retained window is told to resynchronize explicitly; the other cursor-reconnect tables (tasks, claims, handoffs, intents, validations) are still never pruned. - Supabase refresh tokens are stored in the OS keychain when available (macOS `security`, Linux `secret-tool`); otherwise, including on Windows, the mode-`0600` Git-directory configuration fallback is used. - Binary files are shared base64-encoded with byte-exact materialization; any conflict involving a binary file requires human approval, since deterministic hunk/merge analysis is text-only. - Renames are tracked as first-class rename changes (old path, new path, content); a rename conflicting with pending work on either path, moving into or out of a critical path, or whose source has diverged locally always requires approval. diff --git a/apps/daemon/src/client.ts b/apps/daemon/src/client.ts index f33a5c2..a6d43e5 100644 --- a/apps/daemon/src/client.ts +++ b/apps/daemon/src/client.ts @@ -22,7 +22,7 @@ type Status = { eventSequence: number; remoteCursor: number; pendingOutbound: number; - service: { configured: boolean; online: boolean; lastSyncAt?: string; lastSyncError?: string }; + service: { configured: boolean; online: boolean; lastSyncAt?: string; lastSyncError?: string; lastResyncAt?: string; lastResyncMessage?: string }; }; /** diff --git a/apps/daemon/src/index.test.ts b/apps/daemon/src/index.test.ts index 94c3bc9..6ea980d 100644 --- a/apps/daemon/src/index.test.ts +++ b/apps/daemon/src/index.test.ts @@ -8,12 +8,30 @@ import { afterEach, describe, expect, it } from "vitest"; import { CoordinationService } from "../../service/src/index.js"; import { contentHash } from "@crosscode/core"; import { discoverRepository, unifiedDiff } from "@crosscode/git"; -import { LocalDaemon } from "./index.js"; +import { LocalDaemon, type RemoteSyncTransport } from "./index.js"; const exec = promisify(execFile); const directories: string[] = []; async function repo(): Promise { const path = await mkdtemp(join(tmpdir(), "crosscode-daemon-")); directories.push(path); await exec("git", ["init", "-q", "-b", "main", path]); await exec("git", ["-C", path, "config", "user.email", "test@example.com"]); await exec("git", ["-C", path, "config", "user.name", "Test"]); await writeFile(join(path, "a.txt"), "one\n"); await exec("git", ["-C", path, "add", "."]); await exec("git", ["-C", path, "commit", "-qm", "initial"]); return path; } afterEach(async () => { await Promise.all(directories.splice(0).map((path) => rm(path, { recursive: true, force: true }))); }); +/** A transport with nothing to sync, for tests that care about exactly one of its methods. */ +function emptyTransport(): RemoteSyncTransport { + return { + upload: async (record) => ({ id: record.transaction.id, workspaceId: "w", senderReplicaId: "replica", transaction: record.transaction, sequence: 1, createdAt: new Date().toISOString() }), + list: async (after) => ({ operations: [], nextCursor: after }), + uploadTask: async (record) => ({ eventId: record.event.id, workspaceId: "w", senderReplicaId: "replica", task: record.event.payload, updatedAt: new Date().toISOString() }), + listTasks: async (after) => ({ tasks: [], nextCursor: after }), + uploadClaim: async (record) => ({ eventId: record.event.id, workspaceId: "w", senderReplicaId: "replica", claim: record.event.payload, released: record.event.type === "claim.released", updatedAt: new Date().toISOString() }), + listClaims: async (after) => ({ claims: [], nextCursor: after }), + uploadHandoff: async (record) => ({ eventId: record.event.id, workspaceId: "w", senderReplicaId: "replica", handoff: record.event.payload, updatedAt: new Date().toISOString() }), + listHandoffs: async (after) => ({ handoffs: [], nextCursor: after }), + uploadIntent: async (record) => ({ eventId: record.event.id, workspaceId: "w", senderReplicaId: "replica", intent: record.event.payload, updatedAt: new Date().toISOString() }), + listIntents: async (after) => ({ intents: [], nextCursor: after }), + uploadValidation: async (record) => ({ eventId: record.event.id, workspaceId: "w", senderReplicaId: "replica", validation: record.event.payload, createdAt: new Date().toISOString() }), + listValidations: async (after) => ({ validations: [], nextCursor: after }) + }; +} + describe("local daemon coordination", () => { it("shares a proposal only after explicit acceptance", async () => { const senderRoot = await repo(); const receiverRoot = await repo(); const service = new CoordinationService(); @@ -579,6 +597,67 @@ describe("local daemon coordination", () => { expect(result.uploaded).toBe(1); }); + // The failure this guards against is silent: with retention deleting operations, a + // replica whose cursor sits below the deleted range would be handed a short (often empty) + // list, conclude it was caught up, and never learn the proposals existed. The service now + // refuses that cursor outright, and the daemon's job is to adopt the watermark, say so, + // and keep syncing rather than stall. + it("resynchronizes from the retention watermark when its cursor is too old to serve", async () => { + const root = await repo(); + let daemon = await LocalDaemon.open(root, { workspaceId: "w", replicaId: "replica", actorId: "actor" }); + const survivor = { + id: "operation-after-retention", + workspaceId: "w", + senderReplicaId: "other", + transaction: { + id: "operation-after-retention", + base: { files: [] }, + changes: [{ path: "b.txt", kind: "add" as const, afterContent: "kept\n", afterHash: contentHash("kept\n") }], + provenance: { source: "filesystem" as const, confidence: "known" as const }, + safety: { risk: "low" as const, requiresApproval: false } + }, + sequence: 6, + createdAt: new Date().toISOString() + }; + const requested: number[] = []; + const transport = { + ...emptyTransport(), + list: async (after: number) => { + requested.push(after); + // Everything at or below sequence 5 aged out of the plan's window. + if (after < 5) return { status: "cursor-too-old" as const, resyncFrom: 5, retentionDays: 7 }; + return { operations: [survivor], nextCursor: 6 }; + } + }; + + const result = await daemon.syncRemote(transport); + + expect(requested).toEqual([0, 5]); + expect(result).toEqual({ uploaded: 0, downloaded: 1, cursor: 6 }); + expect(daemon.operations.get(survivor.id)?.status).toBe("proposed"); + const service = (await daemon.status()).service as { lastResyncAt?: string; lastResyncMessage?: string }; + expect(service.lastResyncAt).toEqual(expect.any(String)); + expect(service.lastResyncMessage).toContain("7 days"); + expect(service.lastResyncMessage).toContain("resynchronized from sequence 0 to 5"); + + // The jump has to be durable, or the next start walks back into the unservable cursor. + daemon.close(); + daemon = await LocalDaemon.open(root, { workspaceId: "w", replicaId: "replica", actorId: "actor" }); + expect((await daemon.status()).remoteCursor).toBe(6); + }); + + it("refuses a resync order that would rewind its cursor", async () => { + const root = await repo(); + const daemon = await LocalDaemon.open(root, { workspaceId: "w", replicaId: "replica", actorId: "actor" }); + // resyncFrom at or below the cursor contradicts the refusal: the service can serve this + // cursor. Obeying it would re-download and re-propose operations already resolved here. + await expect(daemon.syncRemote({ + ...emptyTransport(), + list: async () => ({ status: "cursor-too-old" as const, resyncFrom: 0, retentionDays: 7 }) + })).rejects.toThrow("resync to a cursor it can already serve"); + expect((await daemon.status()).remoteCursor).toBe(0); + }); + it("recognizes a same-HEAD hard reset as a Git transition", async () => { const root = await repo(); const daemon = await LocalDaemon.open(root, { workspaceId: "w", replicaId: "replica", actorId: "actor" }); diff --git a/apps/daemon/src/index.ts b/apps/daemon/src/index.ts index b93799a..5302c91 100644 --- a/apps/daemon/src/index.ts +++ b/apps/daemon/src/index.ts @@ -109,9 +109,16 @@ export type LocalCoordinationSink = { getAutonomyTier(workspaceId: string): AutonomyTier; }; +/** + * The service's answer when history retention has deleted everything after our cursor: + * there is no page that would be honest, so it names the oldest cursor it can still serve + * completely and we continue from there. See packages/protocol cursorTooOldResponseSchema. + */ +export type RemoteCursorTooOld = { status: "cursor-too-old"; resyncFrom: number; retentionDays: number }; + export type RemoteSyncTransport = { upload(record: OutboundRecord): Promise; - list(after: number): Promise<{ operations: LocalOperation[]; nextCursor: number }>; + list(after: number): Promise<{ operations: LocalOperation[]; nextCursor: number } | RemoteCursorTooOld>; uploadTask(record: TaskOutboundRecord): Promise; listTasks(after: string): Promise<{ tasks: RemoteTask[]; nextCursor: string }>; uploadClaim(record: ClaimOutboundRecord): Promise; @@ -148,6 +155,9 @@ export class LocalDaemon { private gitState: GitState; private materializationPaused = false; private serviceStatus: { configured: boolean; online: boolean; lastSyncAt?: string; lastSyncError?: string } = { configured: false, online: false }; + // Kept apart from serviceStatus, which every successful sync replaces wholesale: a + // retention resync is a gap in this replica's history and must stay visible afterwards. + private lastResync: { at: string; message: string } | undefined; private autonomyTier: AutonomyTier = 0; private mutationTail: Promise = Promise.resolve(); private readonly transactionListeners = new Set<(operation: StoredOperation) => void>(); @@ -173,7 +183,7 @@ export class LocalDaemon { * fields makes this an intentional contract, so the next field added to RepositoryState * is private until someone decides otherwise rather than published by accident. */ - async status() { const repository = await discoverRepository(this.root); return { root: repository.root, head: repository.head, branch: repository.branch, worktree: repository.worktree, remotes: repository.remotes, dirty: repository.dirty, indexTree: repository.indexTree, operation: repository.operation, workspaceId: this.options.workspaceId, replicaId: this.options.replicaId, tasks: this.tasks.size, claims: this.claims.size, proposals: [...this.operations.values()].filter((operation) => operation.status === "proposed").length, materializationPaused: this.materializationPaused, eventSequence: this.eventSequence, remoteCursor: this.remoteCursor, pendingOutbound: [...this.outbound.values()].filter((record) => record.acknowledgedServerSequence === undefined).length, remoteValidations: [...this.remoteValidations.values()], service: { ...this.serviceStatus } }; } + async status() { const repository = await discoverRepository(this.root); return { root: repository.root, head: repository.head, branch: repository.branch, worktree: repository.worktree, remotes: repository.remotes, dirty: repository.dirty, indexTree: repository.indexTree, operation: repository.operation, workspaceId: this.options.workspaceId, replicaId: this.options.replicaId, tasks: this.tasks.size, claims: this.claims.size, proposals: [...this.operations.values()].filter((operation) => operation.status === "proposed").length, materializationPaused: this.materializationPaused, eventSequence: this.eventSequence, remoteCursor: this.remoteCursor, pendingOutbound: [...this.outbound.values()].filter((record) => record.acknowledgedServerSequence === undefined).length, remoteValidations: [...this.remoteValidations.values()], service: { ...this.serviceStatus, ...(this.lastResync ? { lastResyncAt: this.lastResync.at, lastResyncMessage: this.lastResync.message } : {}) } }; } configureRemoteSync(): void { this.serviceStatus = { ...this.serviceStatus, configured: true }; } currentAutonomyTier(): AutonomyTier { return this.autonomyTier; } /** @@ -456,14 +466,24 @@ export class LocalDaemon { } uploaded += 1; } - const page = await transport.list(this.remoteCursor); - const ordered = page.operations.every((operation, index) => operation.sequence === (index === 0 ? this.remoteCursor + 1 : page.operations[index - 1]!.sequence + 1)); - const expectedCursor = page.operations.at(-1)?.sequence ?? this.remoteCursor; - if (!ordered || page.nextCursor !== expectedCursor || page.operations.some((operation) => operation.workspaceId !== this.options.workspaceId || operation.sequence <= this.remoteCursor)) throw new Error("Service cursor response was invalid"); + let page = await transport.list(this.remoteCursor); + if ("status" in page) { + // Our cursor points into history the service no longer keeps. Re-listing from the + // watermark is the whole recovery; a second refusal means the watermark moved again + // mid-resync (or the service is answering nonsense), and that is worth failing on + // rather than looping. + await this.resyncFromRetentionWatermark(page); + page = await transport.list(this.remoteCursor); + if ("status" in page) throw new Error("Coordination service refused the cursor it just told us to resynchronize from"); + } + const listed = page; + const ordered = listed.operations.every((operation, index) => operation.sequence === (index === 0 ? this.remoteCursor + 1 : listed.operations[index - 1]!.sequence + 1)); + const expectedCursor = listed.operations.at(-1)?.sequence ?? this.remoteCursor; + if (!ordered || listed.nextCursor !== expectedCursor || listed.operations.some((operation) => operation.workspaceId !== this.options.workspaceId || operation.sequence <= this.remoteCursor)) throw new Error("Service cursor response was invalid"); const previousCursor = this.remoteCursor; const insertedIds: string[] = []; let downloaded = 0; - for (const remote of page.operations) { + for (const remote of listed.operations) { const transaction = changeTransactionSchema.parse(remote.transaction); transaction.changes.forEach(assertChangeIntegrity); if (remote.senderReplicaId === this.options.replicaId || this.operations.has(remote.id)) continue; @@ -472,8 +492,8 @@ export class LocalDaemon { insertedIds.push(remote.id); downloaded += 1; } - if (page.nextCursor !== this.remoteCursor || downloaded) { - this.remoteCursor = page.nextCursor; + if (listed.nextCursor !== this.remoteCursor || downloaded) { + this.remoteCursor = listed.nextCursor; try { await this.persist("remote.synchronized", { cursor: this.remoteCursor, downloaded }); } catch (error) { this.remoteCursor = previousCursor; @@ -487,6 +507,37 @@ export class LocalDaemon { return { uploaded, downloaded, cursor: this.remoteCursor }; } + /** + * Adopts the oldest cursor the service can still serve, after it reported that ours has + * fallen out of the workspace's history retention window. Nothing can bring the deleted + * operations back, so the only alternative to jumping forward is polling an unservable + * cursor forever. + * + * What a resync costs is proposals this replica never downloaded and now never will: + * other replicas' unreviewed edits. It costs nothing that Git holds -- commits, the + * working tree, and our own outbound queue are untouched -- which is exactly why moving + * the cursor is the right answer rather than a data-loss bug. The event log and + * `status().service` both record it so the gap is visible after the fact. + */ + private async resyncFromRetentionWatermark(status: RemoteCursorTooOld): Promise { + const previousCursor = this.remoteCursor; + // A watermark at or below our cursor contradicts the refusal itself: the service can + // serve this cursor. Rewinding on it would re-propose operations we already resolved. + if (status.resyncFrom <= previousCursor) throw new Error("Service asked for a resync to a cursor it can already serve"); + this.remoteCursor = status.resyncFrom; + this.lastResync = { + at: now(), + message: `Coordination service no longer retains operations at or below sequence ${status.resyncFrom} (plan history retention is ${status.retentionDays} days); resynchronized from sequence ${previousCursor} to ${status.resyncFrom}. Proposals inside that window were not downloaded and are gone; Git history and the working tree are unaffected.` + }; + try { await this.persist("remote.resync_required", { cursor: this.remoteCursor, previousCursor, retentionDays: status.retentionDays }); } + catch (error) { + this.remoteCursor = previousCursor; + this.lastResync = undefined; + throw error; + } + process.stderr.write(`${this.lastResync.message}\n`); + } + private async syncTasks(transport: RemoteSyncTransport): Promise { for (const record of [...this.taskOutbound.values()].filter((item) => item.acknowledgedAt === undefined).sort((left, right) => left.event.clientSequence - right.event.clientSequence)) { const remote = await transport.uploadTask(record); diff --git a/apps/daemon/src/local-event.ts b/apps/daemon/src/local-event.ts index 4961c7b..5617937 100644 --- a/apps/daemon/src/local-event.ts +++ b/apps/daemon/src/local-event.ts @@ -82,6 +82,14 @@ export const localEventSchema = z.discriminatedUnion("type", [ event("transaction.created", storedOperationSchema), event("transaction.published", z.union([storedOperationSchema, z.object({ eventId: z.string().min(1), operationId: z.string().min(1), serverSequence: z.number().int().positive() }).strict()])), event("remote.synchronized", cursorDownloadSchema), + // The service dropped history below `cursor` under the workspace's plan retention, so + // this replica jumped its cursor forward from `previousCursor`. Recorded because it is + // the only trace that operations in that range existed and were never seen here. + event("remote.resync_required", z.object({ + cursor: z.number().int().nonnegative(), + previousCursor: z.number().int().nonnegative(), + retentionDays: z.number().int().positive() + }).strict()), event("transaction.proposed", z.object({ proposals: z.array(storedOperationSchema), remoteCursor: z.number().int().nonnegative() }).strict()), event("transaction.applying", storedOperationSchema), event("transaction.apply_rolled_back", z.object({ id: z.string(), checkpoint: z.string(), recovery: z.string() }).strict()), diff --git a/apps/daemon/src/reconnect.integration.test.ts b/apps/daemon/src/reconnect.integration.test.ts index 6708a9e..8fa61e9 100644 --- a/apps/daemon/src/reconnect.integration.test.ts +++ b/apps/daemon/src/reconnect.integration.test.ts @@ -52,7 +52,8 @@ describe.skipIf(!databaseUrl)("PostgreSQL daemon reconnect", () => { await receiver.syncRemote(receiverClient); await receiver.syncRemote(receiverClient); - expect((await store.listOperations(sender_.principal.workspaceId, 0, 100)).items).toHaveLength(1); + const page = await store.listOperations(sender_.principal.workspaceId, 0, 100); + expect(page.status === "ok" && page.items).toHaveLength(1); expect([...receiver.operations.values()].filter((operation) => operation.status === "proposed")).toHaveLength(1); expect(await readFile(join(receiverRoot, "a.txt"), "utf8")).toBe("one\n"); } finally { @@ -62,4 +63,66 @@ describe.skipIf(!databaseUrl)("PostgreSQL daemon reconnect", () => { await store.close(); } }); + + // End to end over a real service and a real daemon: the replica has been offline long + // enough that retention deleted everything it had not downloaded. A plain age-based + // DELETE would answer its cursor with an empty list -- identical to "you are caught up" -- + // and it would carry on, permanently missing two proposals with nothing logged anywhere. + it("tells a replica whose cursor fell out of the retention window to resync instead of serving it a short list", async () => { + const store = new PgStore(databaseUrl!); + await store.migrate(); + const sender_ = await provisionTestPrincipal(store, { workspaceName: "retention-resync-test", actorId: "sender" }); + const receiver_ = await provisionTestPrincipal(store, { workspaceId: sender_.principal.workspaceId, actorId: "receiver" }); + const workspaceId = sender_.principal.workspaceId; + const server = createServiceServer({ store, jwks: await testSupabaseJwks(), supabaseUrl: TEST_SUPABASE_URL }); + await new Promise((resolveListen) => server.listen(0, "127.0.0.1", resolveListen)); + const url = `http://127.0.0.1:${(server.address() as AddressInfo).port}`; + try { + const senderRoot = await repository(); + const receiverRoot = await repository(); + const sender = await LocalDaemon.open(senderRoot, sender_.principal); + const receiver = await LocalDaemon.open(receiverRoot, receiver_.principal); + const senderClient = new CoordinationServiceClient(sender_.principal, { url, session: { accessToken: sender_.accessToken, refreshToken: "test-unused", expiresAt: sender_.expiresAt } }); + const receiverClient = new CoordinationServiceClient(receiver_.principal, { url, session: { accessToken: receiver_.accessToken, refreshToken: "test-unused", expiresAt: receiver_.expiresAt } }); + + await writeFile(join(senderRoot, "a.txt"), "seen by the receiver\n"); + await sender.capture("first"); + await sender.syncRemote(senderClient); + await receiver.syncRemote(receiverClient); + expect((await receiver.status()).remoteCursor).toBe(1); + + // Two more proposals the receiver never downloads, because it is offline. + for (const content of ["second\n", "third\n"]) { + await writeFile(join(senderRoot, "a.txt"), content); + await sender.capture(`capture ${content.trim()}`); + await sender.syncRemote(senderClient); + } + await store.pool.query("UPDATE operations SET created_at = now() - interval '40 days' WHERE workspace_id = $1", [workspaceId]); + const swept = await store.pruneOperationsByRetention(); + expect(swept.find((result) => result.workspaceId === workspaceId)).toMatchObject({ deleted: 3, prunedThrough: 3 }); + + await receiver.syncRemote(receiverClient); + + const status = await receiver.status(); + expect(status.remoteCursor).toBe(3); + expect(status.service.lastResyncMessage).toContain("resynchronized from sequence 1 to 3"); + // The loss is real and now explicit: the receiver never saw operations 2 and 3, and + // says so, rather than reporting a clean sync it did not have. + expect([...receiver.operations.values()]).toHaveLength(1); + // Steady state afterwards: the adopted cursor is servable, so the next sync is quiet. + await expect(receiver.syncRemote(receiverClient)).resolves.toMatchObject({ downloaded: 0, cursor: 3 }); + + // A daemon released before this status omits protocolVersion. It gets a hard 410 -- + // never a 200 it could read as a successful, empty page. + const legacy = await fetch(`${url}/v1/operations?afterSequence=1`, { + headers: { authorization: `Bearer ${receiver_.accessToken}`, "x-crosscode-workspace-id": workspaceId } + }); + expect(legacy.status).toBe(410); + } finally { + await new Promise((resolveClose) => server.close(() => resolveClose())); + await store.pool.query("DELETE FROM audit_events WHERE workspace_id = $1", [workspaceId]); + await store.pool.query("DELETE FROM workspaces WHERE id = $1", [workspaceId]); + await store.close(); + } + }); }); diff --git a/apps/daemon/src/service-client.ts b/apps/daemon/src/service-client.ts index 1721f07..98c3032 100644 --- a/apps/daemon/src/service-client.ts +++ b/apps/daemon/src/service-client.ts @@ -3,11 +3,12 @@ import { hostname } from "node:os"; import { claimIngestReceiptSchema, claimCursorResponseSchema, - cursorResponseSchema, handoffCursorResponseSchema, handoffIngestReceiptSchema, intentCursorResponseSchema, intentIngestReceiptSchema, + operationsResponseSchema, + OPERATIONS_PROTOCOL_VERSION, registerReplicaRequestSchema, registerReplicaResponseSchema, remoteOperationSchema, @@ -28,7 +29,7 @@ import { } from "@crosscode/protocol"; import type { LocalOperation } from "./types.js"; import type { ClaimOutboundRecord, HandoffOutboundRecord, IntentOutboundRecord, OutboundRecord, TaskOutboundRecord, ValidationOutboundRecord } from "./state.js"; -import type { RemoteSyncTransport } from "./index.js"; +import type { RemoteCursorTooOld, RemoteSyncTransport } from "./index.js"; import { getSupabaseClient, toStoredSession, type StoredSession } from "./supabase-client.js"; type Envelope = { ok: true; data: T } | { ok: false; error: string }; @@ -125,8 +126,17 @@ export class CoordinationServiceClient implements RemoteSyncTransport { }; } - async list(after: number): Promise<{ operations: LocalOperation[]; nextCursor: number }> { - const data = cursorResponseSchema.parse(await this.authorizedRequest(`/v1/operations?afterSequence=${after}`, "GET")); + /** + * `protocolVersion` tells the service this client understands the cursor-too-old status. + * Without it the service answers an unservable cursor with 410 rather than a body an + * older daemon could misread, so sending it is what opts this daemon into resynchronizing + * instead of failing. + */ + async list(after: number): Promise<{ operations: LocalOperation[]; nextCursor: number } | RemoteCursorTooOld> { + const data = operationsResponseSchema.parse( + await this.authorizedRequest(`/v1/operations?afterSequence=${after}&protocolVersion=${OPERATIONS_PROTOCOL_VERSION}`, "GET") + ); + if ("status" in data) return { status: data.status, resyncFrom: data.resyncFrom, retentionDays: data.retentionDays }; return { nextCursor: data.nextCursor, operations: data.operations.map((operation) => { diff --git a/apps/service/migrations/012_single_content_home_and_retention.sql b/apps/service/migrations/012_single_content_home_and_retention.sql new file mode 100644 index 0000000..6f046bf --- /dev/null +++ b/apps/service/migrations/012_single_content_home_and_retention.sql @@ -0,0 +1,26 @@ +-- Stores file content exactly once, and gives per-plan retention the state it needs. +-- +-- (1) operations.event already holds the entire transaction.created envelope, and that +-- envelope's payload IS the ChangeTransaction -- afterContent, unifiedPatch and all. +-- operations.transaction and operation_files.payload were verbatim second and third +-- copies of the same bytes, and file bodies are the bulk of what this database stores. +-- Both are dropped. operation_files keeps its metadata columns and stays what it always +-- was in practice: a per-path index into an operation, which now *references* the one +-- copy of the content in operations.event via (workspace_id, operation_id, path). +-- +-- (2) workspaces.operations_pruned_through is the highest server_sequence retention has +-- deleted for a workspace. Pruning always removes a prefix of the sequence, so every +-- sequence above this watermark is still present: a replica whose cursor is at or above +-- it can be served a complete list, and one below it must be told to resync rather than +-- handed a short list it would read as "caught up" (see store.ts listOperations). +ALTER TABLE operations DROP COLUMN IF EXISTS transaction; +ALTER TABLE operation_files DROP COLUMN IF EXISTS payload; + +ALTER TABLE workspaces + ADD COLUMN IF NOT EXISTS operations_pruned_through bigint NOT NULL DEFAULT 0 + CHECK (operations_pruned_through >= 0); + +-- The retention sweep asks for "the newest sequence older than the plan's window" per +-- workspace; without this it is a full scan of the largest table in the database. +CREATE INDEX IF NOT EXISTS operations_workspace_created_idx + ON operations (workspace_id, created_at); diff --git a/apps/service/src/billing.ts b/apps/service/src/billing.ts index bc455f2..de8ef19 100644 --- a/apps/service/src/billing.ts +++ b/apps/service/src/billing.ts @@ -18,7 +18,9 @@ const ALL_AUTONOMY_TIERS: readonly AutonomyTier[] = ["always-ask", "auto-if-clea // // What is left as a wall: auto-always autonomy (the "I trust it now" moment) and // historyRetentionDays, which bounds the only table that grows without limit -// (operation_files.payload). Team is differentiated by org controls -- SSO, audit export +// (operations, whose event column holds every proposed file body). It is enforced by +// PgStore.pruneOperationsByRetention(), swept on a schedule by retention.ts. +// Team is differentiated by org controls -- SSO, audit export // -- not by seat count, which is why it shares Unlimited's caps. Student mirrors Pro's // limits at Essential's price (verification is enforced elsewhere, not by these caps). export const PLAN_LIMITS: Record { registerReplica: async () => ({ replicaId: "replica-1", createdAt: "2026-01-01T00:00:00.000Z", projectId: null }), assertReplicaOwnership: async () => {}, appendOperation: async () => operation, - listOperations: async () => ({ items: [operation], nextCursor: 1, hasMore: false }) + listOperations: async () => ({ status: "ok", items: [operation], nextCursor: 1, hasMore: false }) } as unknown as PgStore; const base = await listen(store); const accessToken = await signToken(membership.userId); @@ -207,7 +207,7 @@ describe("service HTTP boundary", () => { unattributed.serverSequence = 2; const store = { resolveMembership: async () => membership, - listOperations: async () => ({ items: [attributed, unattributed], nextCursor: 2, hasMore: false }), + listOperations: async () => ({ status: "ok", items: [attributed, unattributed], nextCursor: 2, hasMore: false }), listPresence: async () => [ { replicaId: "replica-1", actorId: membership.actorId, status: "online", lastSeenAt: "2026-01-01T00:00:00.000Z", cursor: 0, projectId }, { replicaId: "replica-2", actorId: membership.actorId, status: "offline", lastSeenAt: null, cursor: null, projectId: null } @@ -234,6 +234,39 @@ describe("service HTTP boundary", () => { expect(Object.keys(presence.data.sessions[1])).toContain("projectId"); }); + // A cursor pointing below the retention watermark has exactly one honest answer, and it + // is not a page: serving the surviving rows (or an empty list, once everything the + // replica had not seen is deleted) is indistinguishable from "you are caught up", which + // is how a replica silently loses proposals forever. + it("answers a cursor below the retention watermark with a resync order, never a short page", async () => { + const store = { + resolveMembership: async () => membership, + listOperations: async () => ({ status: "cursor-too-old", resyncFrom: 42, retentionDays: 7 }) + } as unknown as PgStore; + const base = await listen(store); + const accessToken = await signToken(membership.userId); + const headers = { authorization: `Bearer ${accessToken}`, [WORKSPACE_HEADER]: membership.workspaceId }; + + const current = await fetch(`${base}/v1/operations?afterSequence=3&protocolVersion=2`, { headers }); + expect(current.status).toBe(200); + expect(await current.json()).toEqual({ + ok: true, + data: { status: "cursor-too-old", protocolVersion: 2, resyncFrom: 42, retentionDays: 7 } + }); + + // A daemon built before this status sends no protocolVersion. It must not receive a + // 200 at all: it would parse the body with cursorResponseSchema and, whatever that + // does, "the request succeeded" is the one conclusion it must never reach. A 410 lands + // in the sync-error path it already has. + const legacy = await fetch(`${base}/v1/operations?afterSequence=3`, { headers }); + expect(legacy.status).toBe(410); + const legacyBody = await legacy.json() as { ok: boolean; error: string }; + expect(legacyBody.ok).toBe(false); + expect(legacyBody.error).toContain("upgrade the daemon"); + + expect((await fetch(`${base}/v1/operations?afterSequence=3&protocolVersion=nope`, { headers })).status).toBe(400); + }); + it("registers a replica with its repository so the replica is attributed to a project", async () => { const seen: Array = []; const store = { diff --git a/apps/service/src/http.ts b/apps/service/src/http.ts index b50a2c2..5897820 100644 --- a/apps/service/src/http.ts +++ b/apps/service/src/http.ts @@ -11,6 +11,8 @@ import { createWorkspaceRequestSchema, createWorkspaceResponseSchema, cursorQuerySchema, + cursorTooOldResponseSchema, + OPERATIONS_PROTOCOL_VERSION, handoffIngestRequestSchema, handoffIngestReceiptSchema, intentIngestRequestSchema, @@ -370,8 +372,27 @@ async function handleRequest( if (!/^\d+$/.test(rawCursor)) throw new HttpError(400, "afterSequence must be a non-negative integer"); const afterSequence = Number(rawCursor); if (!Number.isSafeInteger(afterSequence)) throw new HttpError(400, "afterSequence is outside the supported range"); + // Absent means version 1: a daemon built before the cursor-too-old status existed. + const rawVersion = url.searchParams.get("protocolVersion"); + if (rawVersion !== null && !/^\d+$/.test(rawVersion)) throw new HttpError(400, "protocolVersion must be a positive integer"); + const clientProtocolVersion = rawVersion === null ? 1 : Number(rawVersion); const query = cursorQuerySchema.parse({ afterSequence }); const page = await options.store.listOperations(identity.workspaceId, query.afterSequence, 200); + if (page.status === "cursor-too-old") { + // Never answer this with a 200 page. Serving what survives would be indistinguishable + // from "caught up" and would silently drop every proposal retention deleted, which is + // the whole failure this status exists to prevent. + if (clientProtocolVersion < OPERATIONS_PROTOCOL_VERSION) { + throw new HttpError(410, `Operations before ${page.resyncFrom} are outside this workspace's ${page.retentionDays}-day history retention and have been deleted; upgrade the daemon to resynchronize automatically`); + } + send(response, 200, cursorTooOldResponseSchema.parse({ + status: "cursor-too-old", + protocolVersion: OPERATIONS_PROTOCOL_VERSION, + resyncFrom: page.resyncFrom, + retentionDays: page.retentionDays + })); + return; + } send(response, 200, { operations: page.items.map(toRemoteOperation), nextCursor: page.nextCursor @@ -621,7 +642,8 @@ async function readJson(request: IncomingMessage, maximumBytes: number): Promise } } -function toRemoteOperation(operation: StoredOperation): RemoteOperation { +/** Exported so tests can assert on the exact bytes `GET /v1/operations` serializes. */ +export function toRemoteOperation(operation: StoredOperation): RemoteOperation { return { id: operation.id, eventId: operation.eventId, diff --git a/apps/service/src/main.ts b/apps/service/src/main.ts index b73dbdd..dcc9323 100644 --- a/apps/service/src/main.ts +++ b/apps/service/src/main.ts @@ -1,6 +1,7 @@ import { readFile } from "node:fs/promises"; import { createSupabaseJwks } from "./auth.js"; import { assertSafeServiceBinding, createServiceServer } from "./http.js"; +import { DEFAULT_SWEEP_INTERVAL_MS, startRetentionSweep } from "./retention.js"; import { PgStore } from "./store.js"; export async function main(environment: NodeJS.ProcessEnv = process.env): Promise { @@ -43,11 +44,13 @@ export async function main(environment: NodeJS.ProcessEnv = process.env): Promis throw error; } process.stdout.write(`Crosscode service listening on ${tls ? "https" : "http"}://${host}:${port}\n`); + const retention = startConfiguredRetentionSweep(environment); let stopping = false; const stop = async () => { if (stopping) return; stopping = true; await new Promise((resolve) => server.close(() => resolve())); + await retention?.stop(); await store.close(); }; const onSignal = () => void stop() @@ -60,6 +63,42 @@ export async function main(environment: NodeJS.ProcessEnv = process.env): Promis process.once("SIGTERM", onSignal); } +/** + * Enforces PLAN_LIMITS[plan].historyRetentionDays on a schedule, if this deployment gave it + * a role that can. DATABASE_URL deliberately cannot delete operations, so the sweep needs + * CROSSCODE_RETENTION_DATABASE_URL (or the MIGRATION_DATABASE_URL that already exists for + * `pnpm service:migrate`). Without one, retention only happens when an admin runs + * `pnpm service:prune`, which is worth saying out loud at startup rather than leaving the + * operator to discover from a growing bill. + */ +function startConfiguredRetentionSweep(environment: NodeJS.ProcessEnv) { + const databaseUrl = environment.CROSSCODE_RETENTION_DATABASE_URL ?? environment.MIGRATION_DATABASE_URL; + if (!databaseUrl) { + process.stdout.write("Crosscode retention sweep is disabled: set CROSSCODE_RETENTION_DATABASE_URL to a role with DELETE on operations, or run 'pnpm service:prune' manually\n"); + return undefined; + } + const intervalMs = parseSweepIntervalMs(environment.CROSSCODE_RETENTION_SWEEP_MINUTES); + process.stdout.write(`Crosscode retention sweep running every ${Math.round(intervalMs / 60_000)} minute(s)\n`); + return startRetentionSweep({ + databaseUrl, + intervalMs, + onSwept: (results) => { + for (const result of results) { + process.stdout.write(`Crosscode retention: workspace ${result.workspaceId} (${result.plan}, ${result.retentionDays}d) deleted ${result.deleted} operation(s) through sequence ${result.prunedThrough}\n`); + } + }, + onError: (error: unknown) => { + process.stderr.write(`Crosscode retention sweep failed: ${error instanceof Error ? error.message : String(error)}\n`); + } + }); +} + +function parseSweepIntervalMs(value: string | undefined): number { + if (value === undefined) return DEFAULT_SWEEP_INTERVAL_MS; + if (!/^\d+$/.test(value) || Number(value) < 1) throw new Error("CROSSCODE_RETENTION_SWEEP_MINUTES must be a positive integer"); + return Number(value) * 60_000; +} + async function loadTls(environment: NodeJS.ProcessEnv) { const keyPath = environment.CROSSCODE_TLS_KEY; const certPath = environment.CROSSCODE_TLS_CERT; diff --git a/apps/service/src/prune.test.ts b/apps/service/src/prune.test.ts index 2065fe9..5daf232 100644 --- a/apps/service/src/prune.test.ts +++ b/apps/service/src/prune.test.ts @@ -1,9 +1,51 @@ import { randomUUID } from "node:crypto"; +import type { TransactionCreatedEvent } from "@crosscode/protocol"; +import { contentHash } from "@crosscode/core"; import { describe, expect, it } from "vitest"; -import { PgStore } from "./store.js"; +import { PgStore, type Membership } from "./store.js"; const databaseUrl = process.env.CROSSCODE_TEST_DATABASE_URL; +type TestWorkspace = { workspaceId: string; membership: Membership; replicaId: string }; + +async function freshWorkspace(store: PgStore): Promise { + await store.migrate(); + const userId = randomUUID(); + const { workspaceId } = await store.provisionAdmin({ + workspaceName: `test-${randomUUID()}`, userId, actorId: `owner-${randomUUID()}@example.com` + }); + const membership = await store.resolveMembership(userId, workspaceId); + const replica = await store.registerReplica(userId, workspaceId, `replica-${randomUUID()}`); + return { workspaceId, membership, replicaId: replica.replicaId }; +} + +function makeEvent(workspace: TestWorkspace, id: string, clientSequence: number): TransactionCreatedEvent { + return { + id, + schemaVersion: 1, + workspaceId: workspace.workspaceId, + replicaId: workspace.replicaId, + actorId: workspace.membership.actorId, + type: "transaction.created", + clientSequence, + createdAt: new Date().toISOString(), + payload: { + id, + base: { files: [] }, + changes: [{ path: `src/${id}.ts`, kind: "add", afterContent: "content", afterHash: contentHash("content") }], + provenance: { source: "filesystem", confidence: "known" }, + safety: { risk: "low", requiresApproval: false } + } + }; +} + +async function cleanup(store: PgStore, workspaceIds: readonly string[]): Promise { + for (const workspaceId of workspaceIds) { + await store.pool.query("DELETE FROM audit_events WHERE workspace_id = $1", [workspaceId]); + await store.pool.query("DELETE FROM workspaces WHERE id = $1", [workspaceId]); + } +} + describe.skipIf(!databaseUrl)("PostgreSQL retention pruning", () => { it("prunes only audit_events and ended sessions past the retention window", async () => { const store = new PgStore(databaseUrl!); @@ -81,6 +123,107 @@ describe.skipIf(!databaseUrl)("PostgreSQL retention pruning", () => { } }); + // The two halves of retention: it has to actually delete, and it has to leave behind + // enough state that a replica reading the history can tell "deleted" from "nothing new". + it("deletes operations outside the plan's window, per plan, and records the watermark", async () => { + const store = new PgStore(databaseUrl!); + const free = await freshWorkspace(store); + const paid = await freshWorkspace(store); + try { + await store.pool.query("UPDATE workspaces SET plan = 'unlimited' WHERE id = $1", [paid.workspaceId]); + for (const workspace of [free, paid]) { + for (let clientSequence = 1; clientSequence <= 3; clientSequence += 1) { + await store.appendOperation(workspace.membership, makeEvent(workspace, randomUUID(), clientSequence)); + } + // The first two operations are 40 days old: outside free's 7-day window, inside + // unlimited's 365-day one. + await store.pool.query( + "UPDATE operations SET created_at = now() - interval '40 days' WHERE workspace_id = $1 AND server_sequence <= 2", + [workspace.workspaceId] + ); + } + + const swept = await store.pruneOperationsByRetention(); + const freeResult = swept.find((result) => result.workspaceId === free.workspaceId)!; + const paidResult = swept.find((result) => result.workspaceId === paid.workspaceId)!; + + expect(freeResult).toMatchObject({ plan: "free", retentionDays: 7, deleted: 2, prunedThrough: 2 }); + // Same rows, same age, different plan: retention is the plan's window, not a constant. + expect(paidResult).toMatchObject({ plan: "unlimited", retentionDays: 365, deleted: 0, prunedThrough: 0 }); + + const remaining = await store.pool.query<{ server_sequence: string }>( + "SELECT server_sequence FROM operations WHERE workspace_id = $1 ORDER BY server_sequence", + [free.workspaceId] + ); + expect(remaining.rows.map((row) => Number(row.server_sequence))).toEqual([3]); + // operation_files is the per-path index into an operation, so it must not outlive it. + const orphanedFiles = await store.pool.query<{ count: string }>( + "SELECT count(*) FROM operation_files WHERE workspace_id = $1", [free.workspaceId] + ); + expect(Number(orphanedFiles.rows[0]!.count)).toBe(1); + const watermark = await store.pool.query<{ operations_pruned_through: string }>( + "SELECT operations_pruned_through FROM workspaces WHERE id = $1", [free.workspaceId] + ); + expect(Number(watermark.rows[0]!.operations_pruned_through)).toBe(2); + + // A cursor at or above the watermark is still served completely -- pruning takes a + // prefix, so nothing above it is missing. + const servable = await store.listOperations(free.workspaceId, 2, 100); + expect(servable.status === "ok" && servable.items.map((item) => item.serverSequence)).toEqual([3]); + + // Sweeping again is a no-op, and never walks the watermark backwards. + const second = await store.pruneOperationsByRetention(); + expect(second.find((result) => result.workspaceId === free.workspaceId)).toMatchObject({ deleted: 0, prunedThrough: 2 }); + } finally { + await cleanup(store, [free.workspaceId, paid.workspaceId]); + await store.close(); + } + }); + + // The regression this exists for: a replica reconnects by asking for everything after its + // last-known server_sequence, so an empty answer means "you are caught up". Once retention + // deletes rows below that cursor, a plain age-based DELETE makes those two situations + // produce byte-identical responses -- the replica silently loses every proposal in the + // deleted range and never errors. The watermark is what keeps them distinguishable. + it("answers a cursor that fell off the retention window with a resync, not a truncated list", async () => { + const store = new PgStore(databaseUrl!); + const workspace = await freshWorkspace(store); + try { + for (let clientSequence = 1; clientSequence <= 3; clientSequence += 1) { + await store.appendOperation(workspace.membership, makeEvent(workspace, randomUUID(), clientSequence)); + } + // A replica that got as far as sequence 1 and then went offline for a month. + const replicaCursor = 1; + await store.pool.query( + "UPDATE operations SET created_at = now() - interval '40 days' WHERE workspace_id = $1", [workspace.workspaceId] + ); + const swept = await store.pruneOperationsByRetention(); + expect(swept.find((result) => result.workspaceId === workspace.workspaceId)).toMatchObject({ deleted: 3, prunedThrough: 3 }); + + // What the naive implementation would have served this replica: an empty list, which + // is exactly what it also receives when it is genuinely up to date. + const naive = await store.pool.query( + "SELECT id FROM operations WHERE workspace_id = $1 AND server_sequence > $2", + [workspace.workspaceId, replicaCursor] + ); + expect(naive.rows).toHaveLength(0); + + // What it is served instead. + const page = await store.listOperations(workspace.workspaceId, replicaCursor, 100); + expect(page).toEqual({ status: "cursor-too-old", resyncFrom: 3, retentionDays: 7 }); + // Including the case where the cursor is 0 -- a brand-new replica reading a history + // whose beginning is gone is the same problem. + expect(await store.listOperations(workspace.workspaceId, 0, 100)).toMatchObject({ status: "cursor-too-old" }); + // And the resync cursor it is handed is one the service can actually answer. + expect(await store.listOperations(workspace.workspaceId, 3, 100)).toEqual({ + status: "ok", items: [], nextCursor: 3, hasMore: false + }); + } finally { + await cleanup(store, [workspace.workspaceId]); + await store.close(); + } + }); + it("rejects non-positive or non-integer olderThanDays", async () => { const store = new PgStore(databaseUrl!); try { diff --git a/apps/service/src/prune.ts b/apps/service/src/prune.ts index bbd7759..cc7a79c 100644 --- a/apps/service/src/prune.ts +++ b/apps/service/src/prune.ts @@ -1,8 +1,17 @@ -// Admin-invoked-only retention tool. Most tables (operations, tasks, claims, handoffs, -// intents, validations) participate in cursor-based reconnect, where a long-offline replica -// downloads everything after its last-known cursor — pruning them by age would silently break -// that guarantee. Only audit_events (a pure audit trail) and ended sessions (presence -// bookkeeping) are safe to prune here; do not extend this script to any other table. +// Admin-invoked retention tool; the service also runs the operations sweep on a schedule +// (retention.ts). +// +// Tables that participate in cursor-based reconnect -- tasks, claims, handoffs, intents, +// validations -- are still off limits here: a long-offline replica downloads everything +// after its last-known cursor, and age-pruning them would hand it a short list it would +// read as "caught up". Do not extend this script to any of them. +// +// `operations` used to be on that list and no longer is, because it is the one table that +// now has a protocol answer for the problem: pruneOperationsByRetention() deletes strictly +// a prefix of each workspace's sequence and records how far it reached, and a replica +// asking for a cursor below that watermark is told to resync (cursor-too-old) instead of +// being served a truncated page. The window is the plan's, not this script's +// --older-than-days, which governs only audit_events and sessions. import { PgStore } from "./store.js"; function parseOlderThanDays(argv: readonly string[]): number { @@ -23,6 +32,12 @@ async function main(): Promise { const sessionsDeleted = await store.pruneEndedSessions(olderThanDays); process.stdout.write(`audit_events deleted: ${auditEventsDeleted}\n`); process.stdout.write(`sessions deleted: ${sessionsDeleted}\n`); + const swept = await store.pruneOperationsByRetention(); + const operationsDeleted = swept.reduce((total, result) => total + result.deleted, 0); + process.stdout.write(`operations deleted: ${operationsDeleted}\n`); + for (const result of swept.filter((entry) => entry.deleted > 0)) { + process.stdout.write(` workspace ${result.workspaceId} (${result.plan}, ${result.retentionDays}d): ${result.deleted} through sequence ${result.prunedThrough}\n`); + } } finally { await store.close(); } } diff --git a/apps/service/src/retention.test.ts b/apps/service/src/retention.test.ts new file mode 100644 index 0000000..373a473 --- /dev/null +++ b/apps/service/src/retention.test.ts @@ -0,0 +1,62 @@ +import { randomUUID } from "node:crypto"; +import type { TransactionCreatedEvent } from "@crosscode/protocol"; +import { contentHash } from "@crosscode/core"; +import { describe, expect, it } from "vitest"; +import { startRetentionSweep } from "./retention.js"; +import { PgStore, type RetentionSweepResult } from "./store.js"; + +const databaseUrl = process.env.CROSSCODE_TEST_DATABASE_URL; + +describe.skipIf(!databaseUrl)("scheduled retention sweep", () => { + // Retention that only runs when an admin remembers to run it is retention that does not + // run. This is the wiring that makes it happen on its own. + it("prunes on its own as soon as it starts, and stops cleanly", async () => { + const store = new PgStore(databaseUrl!); + await store.migrate(); + const userId = randomUUID(); + const { workspaceId } = await store.provisionAdmin({ + workspaceName: `test-${randomUUID()}`, userId, actorId: `owner-${randomUUID()}@example.com` + }); + const membership = await store.resolveMembership(userId, workspaceId); + const replica = await store.registerReplica(userId, workspaceId, `replica-${randomUUID()}`); + const operationId = randomUUID(); + const event: TransactionCreatedEvent = { + id: operationId, + schemaVersion: 1, + workspaceId, + replicaId: replica.replicaId, + actorId: membership.actorId, + type: "transaction.created", + clientSequence: 1, + createdAt: new Date().toISOString(), + payload: { + id: operationId, + base: { files: [] }, + changes: [{ path: "src/a.ts", kind: "add", afterContent: "content", afterHash: contentHash("content") }], + provenance: { source: "filesystem", confidence: "known" }, + safety: { risk: "low", requiresApproval: false } + } + }; + await store.appendOperation(membership, event); + await store.pool.query("UPDATE operations SET created_at = now() - interval '40 days' WHERE workspace_id = $1", [workspaceId]); + + let report: (results: readonly RetentionSweepResult[]) => void = () => {}; + const swept = new Promise((resolve) => { report = resolve; }); + const errors: unknown[] = []; + // An hour-long interval: what is under test is that the sweep runs at startup rather + // than only on its first tick. + const sweep = startRetentionSweep({ databaseUrl: databaseUrl!, intervalMs: 3_600_000, onSwept: report, onError: (error) => errors.push(error) }); + try { + const results = await swept; + expect(errors).toEqual([]); + expect(results.find((result) => result.workspaceId === workspaceId)).toMatchObject({ plan: "free", deleted: 1, prunedThrough: 1 }); + const remaining = await store.pool.query("SELECT id FROM operations WHERE workspace_id = $1", [workspaceId]); + expect(remaining.rows).toHaveLength(0); + } finally { + await sweep.stop(); + await store.pool.query("DELETE FROM audit_events WHERE workspace_id = $1", [workspaceId]); + await store.pool.query("DELETE FROM workspaces WHERE id = $1", [workspaceId]); + await store.close(); + } + }); +}); diff --git a/apps/service/src/retention.ts b/apps/service/src/retention.ts new file mode 100644 index 0000000..870dd79 --- /dev/null +++ b/apps/service/src/retention.ts @@ -0,0 +1,60 @@ +import { PgStore, type RetentionSweepResult } from "./store.js"; + +/** + * The scheduled half of per-plan history retention. `pnpm service:prune` stays the manual + * admin tool; this is what makes retention actually happen on a running deployment. + * + * It opens its own connection because it must: the request-serving role is deliberately + * denied DELETE on operations (assertRuntimePrivileges), so the sweep is configured with a + * privileged URL and is simply off when one is not supplied. + */ +export const DEFAULT_SWEEP_INTERVAL_MS = 60 * 60 * 1_000; + +export type RetentionSweep = { stop: () => Promise }; + +export type RetentionSweepOptions = { + /** A role with DELETE on operations -- not the service's least-privilege runtime role. */ + databaseUrl: string; + intervalMs?: number; + onSwept?: (results: readonly RetentionSweepResult[]) => void; + onError?: (error: unknown) => void; +}; + +export function startRetentionSweep(options: RetentionSweepOptions): RetentionSweep { + const store = new PgStore(options.databaseUrl); + const intervalMs = options.intervalMs ?? DEFAULT_SWEEP_INTERVAL_MS; + // One sweep at a time. A sweep that outruns its interval (a large backlog on its first + // run, say) must not have a second one pile up behind it competing for the same row locks. + let inFlight: Promise = Promise.resolve(); + let running = false; + let stopped = false; + + const sweep = (): void => { + if (running || stopped) return; + running = true; + inFlight = store.pruneOperationsByRetention() + .then((results) => { + options.onSwept?.(results.filter((result) => result.deleted > 0)); + }) + .catch((error: unknown) => { + // A failed sweep is a cost problem, never a correctness one: nothing was deleted, + // so no cursor was invalidated. Report it and let the next tick try again. + options.onError?.(error); + }) + .finally(() => { running = false; }); + }; + + const timer = setInterval(sweep, intervalMs); + // Retention must never be the reason the process stays alive. + timer.unref(); + sweep(); + + return { + async stop() { + stopped = true; + clearInterval(timer); + await inFlight; + await store.close(); + } + }; +} diff --git a/apps/service/src/store.integration.test.ts b/apps/service/src/store.integration.test.ts index 145e8d3..09a6235 100644 --- a/apps/service/src/store.integration.test.ts +++ b/apps/service/src/store.integration.test.ts @@ -2,10 +2,17 @@ import { randomUUID } from "node:crypto"; import { EPOCH_CURSOR, type HandoffRequestedEvent, type IntentPublishedEvent, type TransactionCreatedEvent } from "@crosscode/protocol"; import { contentHash } from "@crosscode/core"; import { describe, expect, it } from "vitest"; -import { StoreConflictError, StoreUnauthorizedError, PgStore, type Membership } from "./store.js"; +import { toRemoteOperation } from "./http.js"; +import { StoreConflictError, StoreUnauthorizedError, PgStore, type Membership, type OperationPage, type StoredOperation } from "./store.js"; const databaseUrl = process.env.CROSSCODE_TEST_DATABASE_URL; +/** Unwraps a page, failing loudly if retention refused the cursor instead of answering it. */ +function items(page: OperationPage): StoredOperation[] { + if (page.status !== "ok") throw new Error(`Expected an operation page, got '${page.status}'`); + return page.items; +} + describe.skipIf(!databaseUrl)("PostgreSQL service store", () => { it("provisions a member, registers a replica, and sequences exact event retries idempotently", async () => { const store = new PgStore(databaseUrl!); @@ -36,7 +43,7 @@ describe.skipIf(!databaseUrl)("PostgreSQL service store", () => { const first = await store.appendOperation(membership, event); const retry = await store.appendOperation(membership, event); expect(retry.serverSequence).toBe(first.serverSequence); - expect((await store.listOperations(provisioned.workspaceId, 0, 100)).items).toHaveLength(1); + expect(items(await store.listOperations(provisioned.workspaceId, 0, 100))).toHaveLength(1); await expect(store.appendOperation(membership, makeEvent(membership, replica.replicaId, randomUUID(), 1))) .rejects.toBeInstanceOf(StoreConflictError); @@ -155,7 +162,7 @@ describe.skipIf(!databaseUrl)("PostgreSQL service store", () => { // The column being right is not enough -- the read paths a consumer uses have to // return it. This asserts on what listOperations/listPresence hand back, which is // what GET /v1/operations and GET /v1/presence serialize verbatim. - const listed = (await store.listOperations(owner.workspaceId, 0, 100)).items.find((item) => item.id === operationId); + const listed = items(await store.listOperations(owner.workspaceId, 0, 100)).find((item) => item.id === operationId); expect(listed?.projectId).toBe(first!.id); // The same value must survive the write path's own return, since that object is // what gets broadcast over the WebSocket immediately after ingest. @@ -426,4 +433,75 @@ describe.skipIf(!databaseUrl)("PostgreSQL project keys", () => { await store.close(); } }); + +}); + +describe.skipIf(!databaseUrl)("PostgreSQL operation content storage", () => { + // File bodies are the bulk of this database, and they used to be written three times: + // operations.event (the envelope, whose payload is the transaction), operations.transaction + // (a verbatim copy of that payload), and operation_files.payload (a verbatim copy of each + // change inside it). Only the envelope stores them now; the other two are references. + it("stores a change's content exactly once, and reads back byte-identical operations", async () => { + const store = new PgStore(databaseUrl!); + let workspaceId: string | undefined; + try { + await store.migrate(); + const userId = randomUUID(); + const provisioned = await store.provisionAdmin({ workspaceName: `test-${randomUUID()}`, userId, actorId: `owner-${randomUUID()}@example.com` }); + workspaceId = provisioned.workspaceId; + const membership = await store.resolveMembership(userId, workspaceId); + const replica = await store.registerReplica(userId, workspaceId, `replica-${randomUUID()}`); + + // A sentinel long and unique enough that counting its occurrences across whole rows + // is an exact census of where this file's body is stored. + const body = `sentinel-${randomUUID()}-${"x".repeat(64)}`; + const operationId = randomUUID(); + const event = makeEvent(membership, replica.replicaId, operationId, 1); + event.payload = { + ...event.payload, + changes: [{ path: "src/big.ts", kind: "add", afterContent: body, afterHash: contentHash(body) }] + }; + const stored = await store.appendOperation(membership, event); + + const copies = await store.pool.query<{ operations: string; operation_files: string }>( + `SELECT + (SELECT coalesce(sum((length(o::text) - length(replace(o::text, $2, ''))) / length($2)), 0) + FROM operations o WHERE o.workspace_id = $1 AND o.id = $3) AS operations, + (SELECT coalesce(sum((length(f::text) - length(replace(f::text, $2, ''))) / length($2)), 0) + FROM operation_files f WHERE f.workspace_id = $1 AND f.operation_id = $3) AS operation_files`, + [workspaceId, body, operationId] + ); + expect(Number(copies.rows[0]!.operations)).toBe(1); + expect(Number(copies.rows[0]!.operation_files)).toBe(0); + // operation_files is still the per-path index into the operation it always was. + const indexed = await store.pool.query<{ path: string; kind: string; after_hash: string | null }>( + "SELECT path, kind, after_hash FROM operation_files WHERE workspace_id = $1 AND operation_id = $2", + [workspaceId, operationId] + ); + expect(indexed.rows).toEqual([{ path: "src/big.ts", kind: "add", after_hash: contentHash(body) }]); + + // Byte-identity, not just deep equality. GET /v1/operations serializes whatever + // listOperations hands back, and the transaction now comes out of the envelope rather + // than the dropped operations.transaction column. `$1::jsonb` reproduces exactly what + // that column stored and returned -- same input, same jsonb canonicalization -- so + // comparing the serialized forms is comparing the response before and after the change. + const legacyColumn = await store.pool.query<{ transaction: unknown }>( + "SELECT $1::jsonb AS transaction", [JSON.stringify(event.payload)] + ); + const listed = items(await store.listOperations(workspaceId, 0, 100)).find((item) => item.id === operationId)!; + expect(JSON.stringify(toRemoteOperation(listed))).toBe(JSON.stringify( + toRemoteOperation({ ...listed, transaction: legacyColumn.rows[0]!.transaction as typeof listed.transaction }) + )); + // And the same for the object appendOperation returns, which is what the WebSocket + // fan-out broadcasts immediately after ingest. + expect(JSON.stringify(toRemoteOperation(stored))).toBe(JSON.stringify(toRemoteOperation(listed))); + expect(listed.transaction.changes[0]?.afterContent).toBe(body); + } finally { + if (workspaceId) { + await store.pool.query("DELETE FROM audit_events WHERE workspace_id = $1", [workspaceId]); + await store.pool.query("DELETE FROM workspaces WHERE id = $1", [workspaceId]); + } + await store.close(); + } + }); }); diff --git a/apps/service/src/store.ts b/apps/service/src/store.ts index 45286a4..f7d0a46 100644 --- a/apps/service/src/store.ts +++ b/apps/service/src/store.ts @@ -1,13 +1,13 @@ import { createHash, randomBytes, randomUUID } from "node:crypto"; import { readFile } from "node:fs/promises"; import type { - ChangeTransaction, Claim, ClaimCreatedEvent, ClaimReleasedEvent, EventEnvelope, Handoff, HandoffRequestedEvent, + Claim, ClaimCreatedEvent, ClaimReleasedEvent, EventEnvelope, Handoff, HandoffRequestedEvent, HandoffRespondedEvent, Intent, IntentPublishedEvent, PairingStatus, Project, RemoteClaim, RemoteHandoff, RemoteIntent, RemoteOperation, RemoteTask, RemoteValidation, Task, TaskCreatedEvent, TaskUpdatedEvent, TransactionCreatedEvent, Validation, ValidationCompletedEvent } from "@crosscode/protocol"; import { PAIRING_CODE_ALPHABET, PAIRING_CODE_TTL_MS, WORKSPACE_TOKEN_PREFIX } from "@crosscode/protocol"; import { Pool, type PoolClient, type PoolConfig } from "pg"; -import { assertPlanAllowsAutonomyTier, assertSeatCapAvailable, type AutonomyTier, type Plan } from "./billing.js"; +import { assertPlanAllowsAutonomyTier, assertSeatCapAvailable, PLAN_LIMITS, type AutonomyTier, type Plan } from "./billing.js"; import { hashCanonicalPayload } from "./crypto.js"; import { normalizeRepoRemote, normalizeRepoRoot, projectNameFrom } from "./projects.js"; @@ -24,6 +24,26 @@ export type StoredOperation = RemoteOperation & { event: EventEnvelope; }; +/** + * One page of the operation history, or a refusal to answer this cursor at all because + * retention has deleted the rows it asks for. `resyncFrom` is the oldest cursor that can + * still be served completely; `retentionDays` is the plan window that caused the deletion, + * so the message a client shows can name it. + */ +export type OperationPage = + | { status: "ok"; items: StoredOperation[]; nextCursor: number; hasMore: boolean } + | { status: "cursor-too-old"; resyncFrom: number; retentionDays: number }; + +/** What one workspace's retention sweep did; `deleted: 0` means it was already inside its window. */ +export type RetentionSweepResult = { + workspaceId: string; + plan: Plan; + retentionDays: number; + deleted: number; + /** The watermark after the sweep: the highest server_sequence no longer present. */ + prunedThrough: number; +}; + export type PresenceSummary = { replicaId: string; actorId: string; @@ -173,6 +193,8 @@ export class PgStore { await client.query(projectsSql); const teamPlanSql = await readFile(new URL("../migrations/011_team_plan.sql", import.meta.url), "utf8"); await client.query(teamPlanSql); + const contentHomeSql = await readFile(new URL("../migrations/012_single_content_home_and_retention.sql", import.meta.url), "utf8"); + await client.query(contentHomeSql); } finally { await client.query("SELECT pg_advisory_unlock(hashtext('crosscode_migrate'))"); client.release(); @@ -819,7 +841,7 @@ export class PgStore { if (!workspace.rows[0]) throw new StoreUnauthorizedError("Workspace is not available"); const duplicate = await client.query( - `SELECT id, workspace_id, replica_id, project_id, event, transaction, server_sequence, created_at, payload_hash + `SELECT id, workspace_id, replica_id, project_id, event, server_sequence, created_at, payload_hash FROM operations WHERE workspace_id = $1 AND (id = $2 OR event_id = $3 OR (replica_id = $4 AND client_sequence = $5))`, @@ -843,23 +865,31 @@ export class PgStore { // the replica already declared its repository at registration, and a client must // not be able to attribute its edits to an arbitrary project. NULL when the // replica registered before projects existed. + // + // `event` is the single home of this operation's content: its payload is the + // ChangeTransaction, whose changes[].afterContent are the file bodies. Nothing + // else stores those bytes -- mapOperation() reads the transaction back out of + // this column, and operation_files below indexes into it by path. `INSERT INTO operations (id, workspace_id, event_id, client_sequence, server_sequence, replica_id, member_id, - actor_id, payload_hash, event, transaction, project_id) - VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10::jsonb, $11::jsonb, + actor_id, payload_hash, event, project_id) + VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10::jsonb, (SELECT project_id FROM replicas WHERE id = $6 AND workspace_id = $2)) - RETURNING id, workspace_id, replica_id, project_id, event, transaction, server_sequence, created_at, payload_hash`, + RETURNING id, workspace_id, replica_id, project_id, event, server_sequence, created_at, payload_hash`, [ transaction.id, identity.workspaceId, event.id, event.clientSequence, sequence, event.replicaId, - identity.memberId, identity.actorId, payloadHash, JSON.stringify(storedEvent), JSON.stringify(transaction) + identity.memberId, identity.actorId, payloadHash, JSON.stringify(storedEvent) ] ); + // A per-path index into the operation above, not a second copy of it: path, kind and + // the two hashes are what a "who else touched this file" query needs, and the change + // itself (content included) is reachable from (workspace_id, operation_id, path). for (const file of transaction.changes) { await client.query( `INSERT INTO operation_files - (workspace_id, operation_id, path, kind, before_hash, after_hash, payload) - VALUES ($1, $2, $3, $4, $5, $6, $7::jsonb)`, - [identity.workspaceId, transaction.id, file.path, file.kind, file.beforeHash ?? null, file.afterHash ?? null, JSON.stringify(file)] + (workspace_id, operation_id, path, kind, before_hash, after_hash) + VALUES ($1, $2, $3, $4, $5, $6)`, + [identity.workspaceId, transaction.id, file.path, file.kind, file.beforeHash ?? null, file.afterHash ?? null] ); } await client.query("UPDATE workspaces SET next_sequence = $2 WHERE id = $1", [identity.workspaceId, sequence]); @@ -910,11 +940,36 @@ export class PgStore { }); } - async listOperations(workspaceId: string, cursor: number, limit: number): Promise<{ - items: StoredOperation[]; nextCursor: number; hasMore: boolean; - }> { + /** + * Cursor-based reconnect, with retention made explicit. + * + * A replica resumes by asking for everything after its last-known server_sequence, so a + * short answer and "you are caught up" are the same message on the wire. Once retention + * deletes rows, that ambiguity becomes silent proposal loss: a replica whose cursor sits + * below the deleted range would be handed whatever survives -- possibly nothing -- and + * conclude it had seen everything. + * + * operations_pruned_through is what removes the ambiguity. Retention only ever deletes a + * prefix of the sequence, so every sequence above the watermark is still present and any + * cursor at or above it can be answered completely. A cursor below it is answered with + * "cursor-too-old" instead, which callers must surface as a resync rather than a page. + */ + async listOperations(workspaceId: string, cursor: number, limit: number): Promise { + const workspace = await this.pool.query<{ plan: Plan; operations_pruned_through: string }>( + "SELECT plan, operations_pruned_through FROM workspaces WHERE id = $1", + [workspaceId] + ); + if (!workspace.rows[0]) throw new StoreUnauthorizedError("Workspace is not available"); + const prunedThrough = Number(workspace.rows[0].operations_pruned_through); + if (cursor < prunedThrough) { + return { + status: "cursor-too-old", + resyncFrom: prunedThrough, + retentionDays: PLAN_LIMITS[workspace.rows[0].plan].historyRetentionDays + }; + } const result = await this.pool.query( - `SELECT id, workspace_id, replica_id, project_id, event, transaction, server_sequence, created_at + `SELECT id, workspace_id, replica_id, project_id, event, server_sequence, created_at FROM operations WHERE workspace_id = $1 AND server_sequence > $2 ORDER BY server_sequence ASC @@ -922,7 +977,7 @@ export class PgStore { [workspaceId, cursor, limit + 1] ); const items = result.rows.slice(0, limit).map(mapOperation); - return { items, nextCursor: items.at(-1)?.serverSequence ?? cursor, hasMore: result.rows.length > limit }; + return { status: "ok", items, nextCursor: items.at(-1)?.serverSequence ?? cursor, hasMore: result.rows.length > limit }; } async upsertTask(identity: Membership, event: TaskCreatedEvent | TaskUpdatedEvent): Promise { @@ -1152,6 +1207,63 @@ export class PgStore { return result.rowCount ?? result.rows.length; } + /** + * Enforces PLAN_LIMITS[plan].historyRetentionDays across every workspace. Safe to run + * concurrently with ingest and with itself: each workspace is swept under its own row + * lock, and the watermark only ever moves forward. + * + * Requires a role with DELETE on operations, which the request-serving role deliberately + * does not have (assertRuntimePrivileges). Callers pass a privileged connection: the + * scheduled sweep in retention.ts, or `pnpm service:prune`. + */ + async pruneOperationsByRetention(): Promise { + const workspaces = await this.pool.query<{ id: string; plan: Plan }>("SELECT id, plan FROM workspaces ORDER BY id"); + const results: RetentionSweepResult[] = []; + for (const workspace of workspaces.rows) { + results.push(await this.pruneWorkspaceOperations(workspace.id, workspace.plan)); + } + return results; + } + + private async pruneWorkspaceOperations(workspaceId: string, plan: Plan): Promise { + const retentionDays = PLAN_LIMITS[plan].historyRetentionDays; + assertPositiveInteger(retentionDays, "historyRetentionDays"); + return this.transaction(async (client) => { + const locked = await client.query<{ operations_pruned_through: string }>( + "SELECT operations_pruned_through FROM workspaces WHERE id = $1 FOR UPDATE", + [workspaceId] + ); + const unchanged = { workspaceId, plan, retentionDays, deleted: 0 }; + if (!locked.rows[0]) return { ...unchanged, prunedThrough: 0 }; + const prunedThrough = Number(locked.rows[0].operations_pruned_through); + // Deleted by sequence, never directly by age. server_sequence is assigned under the + // workspace row lock while created_at is the inserting transaction's clock, so two + // concurrent ingests can commit with their timestamps inverted relative to their + // sequences. Deleting everything at or below the newest expired sequence keeps what + // remains a contiguous suffix -- which is precisely what the watermark promises + // readers -- at the cost of occasionally taking one barely-inside-the-window row + // with it. + const cutoff = await client.query<{ cutoff: string | null }>( + `SELECT max(server_sequence) AS cutoff + FROM operations + WHERE workspace_id = $1 AND created_at < now() - ($2 || ' days')::interval`, + [workspaceId, retentionDays] + ); + const cutoffSequence = Number(cutoff.rows[0]?.cutoff ?? 0); + if (cutoffSequence <= prunedThrough) return { ...unchanged, prunedThrough }; + // operation_files rows follow via ON DELETE CASCADE. + const deleted = await client.query( + "DELETE FROM operations WHERE workspace_id = $1 AND server_sequence <= $2", + [workspaceId, cutoffSequence] + ); + await client.query( + "UPDATE workspaces SET operations_pruned_through = $2 WHERE id = $1", + [workspaceId, cutoffSequence] + ); + return { workspaceId, plan, retentionDays, deleted: deleted.rowCount ?? 0, prunedThrough: cutoffSequence }; + }); + } + async pruneEndedSessions(olderThanDays: number): Promise { assertPositiveInteger(olderThanDays, "olderThanDays"); const result = await this.pool.query( @@ -1198,8 +1310,8 @@ type OperationRow = { workspace_id: string; replica_id: string; project_id: string | null; - event: EventEnvelope; - transaction: ChangeTransaction; + /** The stored transaction.created envelope; its payload is this operation's transaction. */ + event: TransactionCreatedEvent; server_sequence: string; created_at: Date; payload_hash?: string; @@ -1213,7 +1325,11 @@ function mapOperation(row: OperationRow): StoredOperation { senderReplicaId: row.replica_id, projectId: row.project_id, event: row.event, - transaction: row.transaction, + // Read out of the envelope rather than from a column of its own. jsonb canonicalizes + // a value the same way wherever it is stored, so this is byte-for-byte what the + // dropped operations.transaction column returned -- see the byte-identity assertion + // in store.integration.test.ts. + transaction: row.event.payload, serverSequence: Number(row.server_sequence), createdAt: new Date(row.created_at).toISOString() }; diff --git a/docs/protocol.md b/docs/protocol.md index 60145a3..05ab88d 100644 --- a/docs/protocol.md +++ b/docs/protocol.md @@ -76,6 +76,32 @@ own cursor. `POST /v1/validations` / `GET /v1/validations` follow the same request/receipt/cursor/fan-out pattern as `task`/`claim`/`handoff`/`intent` above, letting replicas see each other's local validation results. +## Reading operations, and history retention + +`GET /v1/operations?afterSequence=` is how a replica resumes: it asks for everything +after its last-known `server_sequence` and gets a `cursorResponseSchema` page back. + +That works only while the history is complete. Per-plan retention +(`PLAN_LIMITS[plan].historyRetentionDays`) deletes operations once they age out, and a +short page and "you are caught up" are the same message on this endpoint — so a replica +whose cursor points into the deleted range would silently lose those proposals. The +service therefore records, per workspace, the highest `server_sequence` retention has +deleted, and refuses any cursor below it rather than answering with what survives: + +```ts +{ status: "cursor-too-old"; protocolVersion: 2; resyncFrom: number; retentionDays: number } +``` + +`resyncFrom` is the oldest cursor that can still be served in full. A replica adopts it +and continues from there. What it loses is proposals it never downloaded; Git remains the +source of truth, so no committed or working-tree work is at stake. + +`protocolVersion` (a query parameter on the request, echoed in this response) versions this +endpoint's answers, separately from the envelope's `schemaVersion`. A client that does not +send `protocolVersion=2` predates the status above, so an unservable cursor is answered +with `410 Gone` instead — a hard failure it already surfaces, rather than a body it might +misread as success. `operationsResponseSchema` is the union a version-2 client parses. + ## Relationship to the daemon's local event log The schemas above govern only what crosses the wire between a daemon and the diff --git a/docs/security.md b/docs/security.md index b1f2f7e..59be931 100644 --- a/docs/security.md +++ b/docs/security.md @@ -217,7 +217,10 @@ Trust boundaries: - **Service ↔ PostgreSQL:** the runtime connects with a least-privilege role (`CROSSCODE_RUNTIME_DB_ROLE`) that cannot update/delete immutable `operations` or `audit_events` rows, and the service refuses to start with a role that can. - The runtime never executes DDL. Row Level Security policies + The runtime never executes DDL. History retention is the one thing that deletes + `operations`, and it is deliberately kept outside that role: the scheduled sweep + opens a second connection with `CROSSCODE_RETENTION_DATABASE_URL`, so no + request-handling code path can reach a connection able to erase history. Row Level Security policies (`004_supabase_auth.sql`) are defense-in-depth on top of this — the service itself still connects with a privileged role rather than through PostgREST, so application-level authorization in `resolveMembership` remains the primary diff --git a/packages/protocol/src/index.test.ts b/packages/protocol/src/index.test.ts index 5933a85..951378b 100644 --- a/packages/protocol/src/index.test.ts +++ b/packages/protocol/src/index.test.ts @@ -10,6 +10,9 @@ import { pairingStatusResponseSchema, workspaceTokenSchema, cursorResponseSchema, + cursorTooOldResponseSchema, + operationsResponseSchema, + OPERATIONS_PROTOCOL_VERSION, daemonConfigSchema, daemonConnectionSchema, eventEnvelopeSchema, @@ -157,6 +160,18 @@ describe("protocol schemas", () => { expect(cursorResponseSchema.parse({ operations: [operation], nextCursor: 1 }).nextCursor).toBe(1); expect(() => cursorQuerySchema.parse({ afterSequence: -1 })).toThrow(); expect(() => serviceIngestReceiptSchema.parse({ eventId: transaction.id, operationId: transaction.id, serverSequence: 0 })).toThrow(); + + // The resync status and a page are two shapes with no overlap. cursorResponseSchema is + // what every daemon built before this status parses `GET /v1/operations` with, so the + // rejection below is the guarantee that such a daemon cannot read "your cursor fell out + // of retention" as "here is your (empty) page, you are caught up". + const tooOld = { status: "cursor-too-old" as const, protocolVersion: OPERATIONS_PROTOCOL_VERSION, resyncFrom: 12, retentionDays: 7 }; + expect(cursorTooOldResponseSchema.parse(tooOld)).toEqual(tooOld); + expect(() => cursorResponseSchema.parse(tooOld)).toThrow(); + expect(() => cursorTooOldResponseSchema.parse({ ...tooOld, protocolVersion: 1 })).toThrow(); + expect(() => cursorTooOldResponseSchema.parse({ ...tooOld, retentionDays: 0 })).toThrow(); + expect(operationsResponseSchema.parse(tooOld)).toEqual(tooOld); + expect(operationsResponseSchema.parse({ operations: [operation], nextCursor: 1 })).toEqual({ operations: [operation], nextCursor: 1 }); }); it("accepts secure or loopback daemon service configuration only", () => { diff --git a/packages/protocol/src/index.ts b/packages/protocol/src/index.ts index 4ccdf0e..d40bcc9 100644 --- a/packages/protocol/src/index.ts +++ b/packages/protocol/src/index.ts @@ -428,6 +428,38 @@ export const cursorResponseSchema = z.object({ }).strict(); export type CursorResponse = z.infer; +/** + * Version of the `GET /v1/operations` read surface the client understands, sent as a + * `protocolVersion` query parameter. Distinct from the envelope's `schemaVersion`, which + * versions event shapes rather than this endpoint's answers. + * + * 1 -> 2 added the cursor-too-old status below. A daemon that predates it sends nothing + * (so the service reads version 1) and is refused with `410 Gone` instead: it would parse + * the status body with cursorResponseSchema, which is `.strict()` and has no `status` key, + * so it cannot mistake it for a page -- but a hard HTTP failure it already surfaces as a + * sync error is a much clearer answer than a validation crash. + */ +export const OPERATIONS_PROTOCOL_VERSION = 2; + +/** + * Retention has deleted the operations this cursor asks for, so no page can answer it + * honestly. `resyncFrom` is the oldest cursor the service can still serve completely; a + * replica adopts it and continues from there, accepting that proposals inside the deleted + * range are gone. That is safe because Git, not this history, is the source of truth -- + * what is lost is unreviewed proposals, never committed or working-tree work. + */ +export const cursorTooOldResponseSchema = z.object({ + status: z.literal("cursor-too-old"), + protocolVersion: z.literal(OPERATIONS_PROTOCOL_VERSION), + resyncFrom: z.number().int().nonnegative(), + retentionDays: z.number().int().positive() +}).strict(); +export type CursorTooOldResponse = z.infer; + +/** What `GET /v1/operations` answers a `protocolVersion=2` client: a page, or a resync order. */ +export const operationsResponseSchema = z.union([cursorResponseSchema, cursorTooOldResponseSchema]); +export type OperationsResponse = z.infer; + export const validationSchema = z.object({ id: z.string(), profile: z.string(), command: z.string(), exitCode: z.number().int(), durationMs: z.number().nonnegative(), tree: z.string().optional(), output: z.string(), runnerId: z.string(), createdAt: z.string().datetime() }); export type Validation = z.infer;