diff --git a/apps/cli/src/index.test.ts b/apps/cli/src/index.test.ts index 4202e1d..9758a4b 100644 --- a/apps/cli/src/index.test.ts +++ b/apps/cli/src/index.test.ts @@ -35,6 +35,7 @@ function environmentWithDaemon(daemon: Partial): Environment { serviceUrl: "https://www.getcrosscode.dev", createService: unused, signIn: unused, + refreshSession: unused, createDaemon: () => ({ start: async () => ({ alreadyRunning: false, pid: 1 }), stop: async () => ({ wasRunning: false }), diff --git a/apps/cli/src/setup.test.ts b/apps/cli/src/setup.test.ts index d099733..ef9e348 100644 --- a/apps/cli/src/setup.test.ts +++ b/apps/cli/src/setup.test.ts @@ -9,7 +9,7 @@ import { syncDaemonConfigSchema } from "../../../packages/protocol/src/sync.js"; // two sides agree, so asserting it with the CLI's helper alone would prove nothing. import { readSyncConfig, syncConfigPath } from "../../daemon/src/sync-config.js"; import { installAgentSurface } from "./agent-surface.js"; -import { configPath, readConfig } from "./config.js"; +import { configPath, readConfig, writeConfig } from "./config.js"; import type { DaemonControl } from "./daemon.js"; import { httpSyncService, type SyncService } from "./service.js"; import { setup, type Environment } from "./setup.js"; @@ -43,8 +43,8 @@ const LEGAL_DOCUMENTS = [ { document: "privacy", version: "2026-08-01", url: "/docs/privacy-policy.html" } ]; -function stubEnvironment(overrides: { service?: Partial } = {}) { - const calls = { createProject: 0, redeemInvite: 0, signIn: 0, daemonStart: 0 }; +function stubEnvironment(overrides: { service?: Partial; refreshSession?: Environment["refreshSession"] } = {}) { + const calls = { createProject: 0, redeemInvite: 0, signIn: 0, daemonStart: 0, refreshSession: 0 }; /** * The terms half, counted separately so the assertions above keep saying exactly what they * said: `calls` is the "did start do this twice" ledger. @@ -92,6 +92,10 @@ function stubEnvironment(overrides: { service?: Partial } = {}) { calls.signIn += 1; return { session: SESSION, githubToken: "gho_test" }; }, + refreshSession: async (session) => { + calls.refreshSession += 1; + return overrides.refreshSession ? overrides.refreshSession(session) : session; + }, createDaemon: () => daemon, // The real installer, not a stub: whether installing twice duplicates anything is the // question, so it has to be the code that writes the files. @@ -116,7 +120,7 @@ describe("crosscode start", () => { const result = await setup(root, environment); expect(result).toMatchObject({ repoRoot: root, repo: "acme/app", projectId: "project-1", signedIn: "just-now", project: "created" }); - expect(calls).toEqual({ createProject: 1, redeemInvite: 0, signIn: 1, daemonStart: 1 }); + expect(calls).toEqual({ createProject: 1, redeemInvite: 0, signIn: 1, daemonStart: 1, refreshSession: 0 }); // What it wrote is the contract's own shape, not a CLI-private one. expect(syncDaemonConfigSchema.parse(JSON.parse(await readFile(await configPath(root), "utf8")))).toMatchObject({ projectId: "project-1", @@ -143,6 +147,41 @@ describe("crosscode start", () => { expect(await readSyncConfig(root)).toMatchObject({ projectId: "project-1", repo: "acme/app" }); }); + /** + * The dead end this closes. Supabase access tokens last an hour and refresh tokens are + * spent on first use, so a checkout comes back to a session that is on disk and useless. + * The daemon died on it saying "run `crosscode start`"; `start` saw a session, said + * "Already signed in for this checkout", and started the same doomed daemon again. There + * was no way out of that loop from inside the CLI. + */ + it("signs in again when the stored session cannot be refreshed", async () => { + const root = await checkout(); + const { environment, calls } = stubEnvironment({ refreshSession: async () => undefined }); + await setup(root, environment); + await writeConfig(root, { ...(await readSyncConfig(root)), service: { url: environment.serviceUrl, session: { ...SESSION, expiresAt: "2020-01-01T00:00:00.000Z" } } }); + + const second = await setup(root, environment); + + expect(calls.refreshSession).toBe(1); + expect(second.signedIn).toBe("just-now"); + expect(calls.signIn).toBe(2); + }); + + it("reuses a session it can still refresh, without a second sign-in", async () => { + const root = await checkout(); + const refreshed = { ...SESSION, accessToken: "refreshed", expiresAt: "2031-01-01T00:00:00.000Z" }; + const { environment, calls } = stubEnvironment({ refreshSession: async () => refreshed }); + await setup(root, environment); + await writeConfig(root, { ...(await readSyncConfig(root)), service: { url: environment.serviceUrl, session: { ...SESSION, expiresAt: "2020-01-01T00:00:00.000Z" } } }); + + const second = await setup(root, environment); + + expect(second.signedIn).toBe("already"); + expect(calls.signIn).toBe(1); + // The refreshed token is what the daemon will be started with, so it has to be stored. + expect((await readSyncConfig(root)).service.session?.accessToken).toBe("refreshed"); + }); + // The property the invite flow depends on: the person sending the link has already run // start, and the person receiving it is told to run the same command. it("is idempotent -- one project, one daemon, no duplicated MCP, skill, or hook install", async () => { @@ -152,7 +191,7 @@ describe("crosscode start", () => { const first = await setup(root, environment); const second = await setup(root, environment); - expect(calls).toEqual({ createProject: 1, redeemInvite: 0, signIn: 1, daemonStart: 1 }); + expect(calls).toEqual({ createProject: 1, redeemInvite: 0, signIn: 1, daemonStart: 1, refreshSession: 0 }); expect(second.projectId).toBe(first.projectId); expect(second).toMatchObject({ signedIn: "already", project: "existing", daemon: { alreadyRunning: true } }); expect(second.agent).toMatchObject({ mcp: { changed: false }, skill: { changed: false }, hooks: { changed: false } }); @@ -261,7 +300,7 @@ describe("crosscode join ", () => { const result = await setup(root, environment, { code: "CC-7F3A-9C2E" }); - expect(calls).toEqual({ createProject: 0, redeemInvite: 1, signIn: 1, daemonStart: 1 }); + expect(calls).toEqual({ createProject: 0, redeemInvite: 1, signIn: 1, daemonStart: 1, refreshSession: 0 }); expect(result).toMatchObject({ projectId: "project-from-invite", repo: "acme/app", project: "joined", daemon: { alreadyRunning: false } }); expect(await readConfig(root)).toMatchObject({ projectId: "project-from-invite", repo: "acme/app", service: { session: SESSION } }); expect(result.agent).toMatchObject({ mcp: { changed: true }, skill: { changed: true }, hooks: { changed: true } }); diff --git a/apps/cli/src/setup.ts b/apps/cli/src/setup.ts index 9f6c0a1..124e9b2 100644 --- a/apps/cli/src/setup.ts +++ b/apps/cli/src/setup.ts @@ -8,6 +8,7 @@ import { localDaemon, type DaemonControl } from "./daemon.js"; import { CliError } from "./errors.js"; import { confirmTerms, type ConfirmKind } from "./notice.js"; import { httpSyncService, type SyncService } from "./service.js"; +import { refreshStoredSession, type StoredSession } from "../../daemon/src/supabase-client.js"; import { SERVICE_URL } from "./version.js"; /** @@ -27,6 +28,8 @@ export type Environment = { serviceUrl: string; createService(accessToken: string): SyncService; signIn: SignIn; + /** Trades a refresh token for a live one, or resolves undefined if it cannot. */ + refreshSession(session: StoredSession): Promise; createDaemon(repoRoot: string): DaemonControl; installAgentSurface(repoRoot: string): Promise; /** @@ -41,6 +44,7 @@ export function defaultEnvironment(): Environment { serviceUrl: SERVICE_URL, createService: (accessToken) => httpSyncService(SERVICE_URL, accessToken), signIn: githubSignIn(), + refreshSession: (session) => refreshStoredSession(session).catch(() => undefined), createDaemon: (repoRoot) => localDaemon(repoRoot), installAgentSurface: (repoRoot) => installAgentSurface(repoRoot), confirmTerms: (options) => confirmTerms(options) @@ -92,12 +96,20 @@ export async function setup(directory: string, environment: Environment, options // so it exists for exactly as long as the handshake that produced it. `join` therefore // signs in even in a checkout that is already signed in, rather than failing at the // redeem call with nothing to offer. - const reuseSession = session && !options.code; - if (reuseSession) report("Already signed in for this checkout."); - const signedIn = reuseSession + // + // A stored session is only worth reusing if it still works. Supabase access tokens last + // an hour and refresh tokens are spent on first use, so "there is a session on disk" and + // "this checkout can talk to the service" are different questions. Answering the first + // one left the user with no way out: the daemon died on the dead session telling them to + // run `crosscode start`, and `start` said "Already signed in for this checkout" and + // started the same doomed daemon again. + const usable = session && !options.code ? await usableSession(session, environment) : undefined; + if (usable) report("Already signed in for this checkout."); + else if (session && !options.code) report("This checkout's sign-in has expired; signing in again."); + const signedIn = usable ? undefined : await environment.signIn({ serviceUrl: environment.serviceUrl, openBrowser: options.openBrowser !== false, report }); - const activeSession = signedIn?.session ?? session!; + const activeSession = signedIn?.session ?? usable!; const service = environment.createService(activeSession.accessToken); await acceptTerms(service, environment, { assumeYes, noticeShown, report }); @@ -119,13 +131,26 @@ export async function setup(directory: string, environment: Environment, options repo, projectId, branch: repository.branch, - signedIn: reuseSession ? "already" : "just-now", + signedIn: usable ? "already" : "just-now", project: origin, daemon, agent }; } +/** + * The stored session if it can still be used, refreshed first if it is close to expiring, + * and undefined if the refresh token is spent or revoked -- in which case the only way + * forward is signing in again, which is what the caller does. + */ +async function usableSession(session: StoredSession, environment: Environment): Promise { + if (Date.parse(session.expiresAt) - Date.now() > EXPIRY_MARGIN_MS) return session; + return environment.refreshSession(session); +} + +/** Refresh this far ahead of expiry, so setup never hands the daemon a token about to die. */ +const EXPIRY_MARGIN_MS = 60_000; + /** * Records this person's acceptance, but only of something they were actually shown. * diff --git a/apps/daemon/src/main.ts b/apps/daemon/src/main.ts index 0a28f5e..8f4ab1f 100644 --- a/apps/daemon/src/main.ts +++ b/apps/daemon/src/main.ts @@ -1,6 +1,7 @@ #!/usr/bin/env node import { spawn } from "node:child_process"; import { createDaemonTelemetry } from "./observability.js"; +import { SessionExpiredError } from "./supabase-client.js"; import { supervise } from "./supervisor.js"; import { SyncDaemon } from "./sync-daemon.js"; import { VERSION } from "./version.js"; @@ -33,6 +34,13 @@ if (!directory) { const daemon = await SyncDaemon.start(directory).catch(async (error: unknown) => { telemetry.capture(error, "startup"); await telemetry.flush(); + // An expired sign-in is the one start-up failure with a one-line fix, and a stack trace + // buries it. Everything else still throws, because it is a bug and the trace is the + // useful part. + if (error instanceof SessionExpiredError) { + process.stderr.write(`${error.message}\n`); + process.exit(1); + } throw error; }); process.stdout.write(`${JSON.stringify({ ready: true, pid: process.pid, port: daemon.connection.port })}\n`); diff --git a/apps/daemon/src/supabase-client.ts b/apps/daemon/src/supabase-client.ts index 3e78125..9d81071 100644 --- a/apps/daemon/src/supabase-client.ts +++ b/apps/daemon/src/supabase-client.ts @@ -70,6 +70,27 @@ export function getSupabaseClient(): SupabaseClient { return createClient(url, anonKey, { auth: { persistSession: false, autoRefreshToken: false } }); } +/** Re-signing-in is the only fix, so this is worth telling apart from a network failure. */ +export class SessionExpiredError extends Error {} + +/** + * Trades a refresh token for a fresh access token. Supabase access tokens last an hour, so + * without this a daemon that has been running since before lunch holds a token the service + * refuses -- which is exactly what happened: `register()` threw "Access token is invalid or + * expired" and took the process down, and the only way back was signing in again. + */ +export async function refreshStoredSession(session: StoredSession): Promise { + const supabase = getSupabaseClient(); + const { data, error } = await supabase.auth.refreshSession({ refresh_token: session.refreshToken }); + if (error || !data.session) { + // A refresh token is spent on first use and rotated, so "already used" and "expired" + // both mean the same thing to the person reading this: sign in again. Say that, rather + // than leaving them with Supabase's wording and a stack trace. + throw new SessionExpiredError(`This checkout's Crosscode sign-in is no longer valid (${error?.message ?? "no session returned"}); run \`crosscode start\` to sign in again`); + } + return toStoredSession(data.session); +} + export function toStoredSession(session: Session): StoredSession { const expiresAtMs = session.expires_at ? session.expires_at * 1_000 : Date.now() + (session.expires_in ?? 3_600) * 1_000; return { accessToken: session.access_token, refreshToken: session.refresh_token, expiresAt: new Date(expiresAtMs).toISOString() }; diff --git a/apps/daemon/src/sync-daemon.test.ts b/apps/daemon/src/sync-daemon.test.ts index 360272b..7f33792 100644 --- a/apps/daemon/src/sync-daemon.test.ts +++ b/apps/daemon/src/sync-daemon.test.ts @@ -1,4 +1,5 @@ import { execFile } from "node:child_process"; +import { createServer } from "node:http"; import { mkdtemp, mkdir, readFile, rm, writeFile } from "node:fs/promises"; import { tmpdir } from "node:os"; import { dirname, join } from "node:path"; @@ -36,6 +37,7 @@ const FAST: SyncDaemonOptions = { flushMs: 100, deferralMs: 60, gitPollMs: 100, + streamPollMs: 100, // The 10s hot window is real behaviour, not a constant worth waiting out in a test. engine: { hotWindowMs: 150, maxDeferrals: 3 } }; @@ -219,6 +221,102 @@ describe("the daemon around the engine", () => { await waitFor("bob's cursor to advance", async () => bob.status().cursor > 0); }, 60_000); + /** + * The production outage this covers: the service runs as a serverless function, which + * cannot answer a WebSocket upgrade, so `/v1/stream` 404s forever. Publishing worked and + * receiving did not -- an uploader, not a sync -- and `crosscode status` said + * `connected: false` with no way for it to ever become true. + */ + it("still receives a peer's edits when the host cannot serve a stream at all", async () => { + const service = await startSyncServiceStub({ streams: false }); + services.push(service); + const origin = await seedOrigin({ "a.txt": "0\n", "b.txt": "0\n" }); + const roots = await Promise.all(["alice", "bob"].map((name) => checkout(origin, name, service))); + const [alice, bob] = [await start(roots[0]!), await start(roots[1]!)]; + + // Reachable by the only transport there is, and honest about saying so. + await waitFor("both daemons to report themselves connected", async () => alice.status().connected && bob.status().connected); + + // Each direction gets its own file. Bouncing one file back immediately would land the + // reply inside the hot window the first write opened, which is a deferral test, not a + // transport one. + await type(roots[0]!, "a.txt", "arrived without a socket\n"); + await waitFor("bob to receive it", () => allEqual(roots, "a.txt", "arrived without a socket\n")); + + // And the other way, so it is a two-way sync rather than one lucky direction. + await type(roots[1]!, "b.txt", "and back again\n"); + await waitFor("alice to receive it", () => allEqual(roots, "b.txt", "and back again\n")); + }, 60_000); + + /** + * `crosscode start` polls for the descriptor every 100ms and reports success the moment + * it appears. Written when the loopback server bound, it appeared before the stream was + * up -- so a daemon that then died on an expired session still had `start` announcing + * "Daemon started; this checkout is syncing" about a process that was already gone. + */ + it("does not advertise itself as ready while start-up is still in flight", async () => { + // Accepts the connection and never answers, so the daemon is unambiguously still + // starting up while the assertions below run. + const hanging = createServer(() => {}); + await new Promise((ready) => hanging.listen(0, "127.0.0.1", ready)); + const service = await startSyncServiceStub(); + services.push(service); + const origin = await seedOrigin({ "a.txt": "0\n" }); + const root = await checkout(origin, "alice", service); + await writeSyncConfig(root, { + projectId: "project-1", repo: "acme/app", + service: { url: `http://127.0.0.1:${(hanging.address() as { port: number }).port}` } + }); + + const starting = start(root).catch(() => undefined); + // Comfortably past the point the loopback server has bound and the old code would have + // published the descriptor, and well inside the request that never returns. + await new Promise((settle) => setTimeout(settle, 2_000)); + const descriptor = await read(root, ".git/crosscode/daemon.json"); + + // Dropping the connection fails the pending request now rather than at its 10s timeout, + // which is 8s this suite does not need to spend. + hanging.closeAllConnections(); + await new Promise((closed) => hanging.close(() => closed())); + await starting; + + expect(descriptor).toBeNull(); + }, 60_000); + + /** + * A change built on a blob this checkout has never held makes apply() call catchUp() to + * fill the gap. Called from inside a drain, that re-walked the same page from the same + * cursor -- and the cursor cannot move past a change that has not applied, so it never + * moved. One such change from one peer span the real service at several requests a second + * indefinitely, while `crosscode status` showed `cursor` frozen and no error anywhere. + */ + it("does not spin the service on a change it cannot rebase", async () => { + const service = await startSyncServiceStub(); + services.push(service); + const origin = await seedOrigin({ "a.txt": "0\n" }); + const root = await checkout(origin, "alice", service); + // A modify whose baseHash names a blob that exists nowhere: the sender's history is one + // this checkout has no way to reach. + service.changes.push({ + sequence: 1, projectId: "project-1", branch: "main", + replicaId: "00000000-1111-4222-8333-444455556666", + createdAt: "2026-01-01T00:00:00.000Z", + version: { + path: "a.txt", op: "modify", baseHash: "0".repeat(40), + contentHash: "1".repeat(40), content: "from a history we never had\n", encoding: "utf8" + } + }); + + const daemon = await start(root); + await waitFor("the cursor to get past it", async () => daemon.status().cursor >= 1); + const afterSettling = service.listCalls; + await new Promise((settle) => setTimeout(settle, 1_000)); + + // Ten polls a second would be a spin; the configured interval is 100ms in these tests, + // so a healthy daemon asks a handful of times in a second and no more. + expect(service.listCalls - afterSettling).toBeLessThan(20); + }, 60_000); + it("resyncs from full content when the service says the cursor is too old", async () => { const service = await startSyncServiceStub(); services.push(service); diff --git a/apps/daemon/src/sync-daemon.ts b/apps/daemon/src/sync-daemon.ts index 29d0d95..67086c4 100644 --- a/apps/daemon/src/sync-daemon.ts +++ b/apps/daemon/src/sync-daemon.ts @@ -21,9 +21,11 @@ import { readSyncState, removeOwnedDescriptor, writeDaemonDescriptor, + writeSyncConfig, writeSyncState, type DaemonDescriptor } from "./sync-config.js"; +import type { StoredSession } from "./supabase-client.js"; import { SyncServiceClient } from "./sync-service-client.js"; /** @@ -44,6 +46,8 @@ export type SyncDaemonOptions = { deferralMs?: number; /** How often the daemon looks for a branch switch or an in-progress git operation. */ gitPollMs?: number; + /** How often changes are pulled when the stream is unavailable. Real default is 3s. */ + streamPollMs?: number; port?: number; actor?: string; engine?: EngineOptions; @@ -66,7 +70,10 @@ class HttpError extends Error { export class SyncDaemon { private watcher?: FSWatcher; private server?: Server; + private draining = false; private descriptor?: DaemonDescriptor; + /** Built when the loopback server binds, published only once start-up has fully succeeded. */ + private pendingDescriptor?: DaemonDescriptor; private timers: NodeJS.Timeout[] = []; private readonly debounce = new Map(); private readonly dirty = new Set(); @@ -103,7 +110,7 @@ export class SyncDaemon { const state = await readSyncState(root); const branch = await currentBranch(root); const engine = await SyncEngine.open(root, options.engine ?? {}); - const client = new SyncServiceClient(config, branch, state.branch === branch ? state.replicaId : undefined); + const client = new SyncServiceClient(config, branch, state.branch === branch ? state.replicaId : undefined, persistSession(root, config), options.streamPollMs); const daemon = new SyncDaemon(root, config, engine, client, branch, await headCommit(root), { debounceMs: options.debounceMs ?? 300, flushMs: options.flushMs ?? 2_000, @@ -127,6 +134,7 @@ export class SyncDaemon { this.timers.push(interval(() => void this.enqueue(() => this.observeGit()), this.options.gitPollMs)); // Anything edited while the daemon was down is still a change the peers need. await this.enqueue(() => this.publishAll()); + await this.announceReady(); } // ---- serialization ---------------------------------------------------- @@ -231,7 +239,11 @@ export class SyncDaemon { // Announce on every (re)subscribe: a peer that joins after we committed would // otherwise never learn our HEAD, since presence is only sent when something happens. onConnected: () => { void this.enqueue(async () => { await this.catchUp(); await this.announce(); }); }, - onDisconnected: () => {} + onDisconnected: () => {}, + // The stream's fallback. Same catch-up, on a timer instead of on a subscribe, so a + // checkout that can never open a socket still receives its peers' edits. Presence is + // not sent here: it lives only on the socket, so the room goes quiet but not blind. + onPoll: () => this.enqueue(() => this.catchUp()) }, () => this.cursor); } @@ -248,6 +260,13 @@ export class SyncDaemon { // The sender built on a blob we have never held, so we are genuinely behind. Fill // the gap from the cursor and retry once; a second miss has to surface rather than // loop, which is what allowCatchup: false forces. + // + // That guard covers the retry but not the fill: called from inside a drain, catchUp + // re-walks the same page from the same cursor -- which the cursor cannot leave, because + // it only advances past a change that applied. One unrebasable change from a peer + // therefore span the service at several requests a second, forever, while `crosscode + // status` showed a cursor that never moved. The guard inside catchUp makes the nested + // call a no-op, so the retry below decides, and an honest conflict reaches the agent. await this.catchUp(); result = await this.engine.receive(change.version, { peer: change.replicaId, allowCatchup: false }); } @@ -268,6 +287,18 @@ export class SyncDaemon { * is what turns a dropped socket into a gap that closes itself. */ private async catchUp(): Promise { + // Not re-entrant: apply() calls this to fill a gap, and the drain that is already + // running is the fill. See the note there for what the recursion cost. + if (this.draining) return; + this.draining = true; + try { + await this.drainChanges(); + } finally { + this.draining = false; + } + } + + private async drainChanges(): Promise { for (let page = 0; page < 100; page += 1) { const response = await this.client.changes(this.cursor); if ("status" in response) { @@ -331,7 +362,7 @@ export class SyncDaemon { this.headNotice = undefined; this.reviewedHeads = ""; this.client.stop(); - this.client = new SyncServiceClient(this.config, branch); + this.client = new SyncServiceClient(this.config, branch, undefined, persistSession(this.root, this.config), this.options.streamPollMs); if (this.options.stream ?? true) await this.startStream(); await this.publishAll(); return; @@ -371,7 +402,7 @@ export class SyncDaemon { /** Tells the room where this checkout is. Silent when there is no stream to say it on. */ private async announce(paths: string[] = []): Promise { - if (!this.client.id || !this.client.connected) return; + if (!this.client.id || !this.client.streaming) return; await this.client.presence(paths, this.options.actor ?? "you", this.head ?? undefined).catch(() => {}); } @@ -466,12 +497,24 @@ export class SyncDaemon { server.listen(this.options.port ?? 0, "127.0.0.1", () => { server.off("error", failed); ready(); }); }); this.server = server; - this.descriptor = daemonDescriptorSchema.parse({ + this.pendingDescriptor = daemonDescriptorSchema.parse({ pid: process.pid, port: (server.address() as { port: number }).port, secret, startedAt: new Date().toISOString() }); + } + + /** + * The descriptor is the daemon saying "I am up", so it is written last -- after the + * stream, the watcher, and the first publish have all succeeded. Written from startHttp() + * as it used to be, a daemon that then died on an expired session still left `crosscode + * start` reporting "Daemon started; this checkout is syncing" about a process that was + * gone a second later, because `start` polls for exactly this file. + */ + private async announceReady(): Promise { + if (!this.pendingDescriptor) throw new Error("The daemon's loopback server was never started"); + this.descriptor = this.pendingDescriptor; await writeDaemonDescriptor(this.root, this.descriptor); } @@ -557,6 +600,17 @@ function interval(task: () => void, ms: number): NodeJS.Timeout { const MAX_BODY_BYTES = 4 * 1024 * 1024; +/** + * Writes a refreshed session back to the checkout's config. Supabase rotates the refresh + * token on every use, so this is not an optimization: without it a restart would present + * the token the last refresh already spent, and sign-in would be the only way back. + */ +function persistSession(root: string, config: SyncDaemonConfig): (session: StoredSession) => Promise { + return async (session) => { + await writeSyncConfig(root, { ...config, service: { ...config.service, session } }); + }; +} + async function readJsonBody(request: import("node:http").IncomingMessage): Promise { if (request.headers["content-type"]?.split(";")[0]?.trim() !== "application/json") throw new HttpError(415, "Unsupported content type"); const chunks: Buffer[] = []; diff --git a/apps/daemon/src/sync-service-client.test.ts b/apps/daemon/src/sync-service-client.test.ts new file mode 100644 index 0000000..7aec77c --- /dev/null +++ b/apps/daemon/src/sync-service-client.test.ts @@ -0,0 +1,87 @@ +import { createServer, type Server } from "node:http"; +import type { AddressInfo } from "node:net"; +import { afterEach, describe, expect, it, vi } from "vitest"; +import type { StoredSession } from "./supabase-client.js"; + +/** + * Supabase access tokens last an hour and a daemon is expected to outlive that. Before + * this, one did not: `register()` threw "Access token is invalid or expired", the daemon + * died with a stack trace, and the refresh token sitting in the config next to the dead + * access token was never spent. + */ + +const refreshStoredSession = vi.hoisted(() => vi.fn()); +vi.mock("./supabase-client.js", () => ({ refreshStoredSession })); + +const { SyncServiceClient } = await import("./sync-service-client.js"); + +const servers: Server[] = []; +afterEach(async () => { + refreshStoredSession.mockReset(); + await Promise.all(servers.splice(0).map((server) => new Promise((resolve) => server.close(() => resolve())))); +}); + +/** Records the bearer on every request, and answers whatever the client asked for. */ +async function recordingService(): Promise<{ url: string; bearers: string[] }> { + const bearers: string[] = []; + const server = createServer((request, response) => { + bearers.push(String(request.headers.authorization)); + response.writeHead(200, { "content-type": "application/json" }); + response.end(JSON.stringify({ ok: true, data: { replicaId: "11111111-2222-4333-8444-555566667777", cursor: 0 } })); + }); + servers.push(server); + await new Promise((ready) => server.listen(0, "127.0.0.1", ready)); + return { url: `http://127.0.0.1:${(server.address() as AddressInfo).port}`, bearers }; +} + +const session = (expiresAt: string, accessToken = "old-token"): StoredSession => + ({ accessToken, refreshToken: "refresh-1", expiresAt }); + +const configFor = (url: string, expiresAt: string) => + ({ projectId: "project-1", repo: "acme/app", service: { url, session: session(expiresAt) } }); + +const inAnHour = () => new Date(Date.now() + 3_600_000).toISOString(); +const anHourAgo = () => new Date(Date.now() - 3_600_000).toISOString(); + +describe("the sync client's session", () => { + it("refreshes an expired token, persists it, and sends the new one", async () => { + const service = await recordingService(); + refreshStoredSession.mockResolvedValue(session(inAnHour(), "new-token")); + const persisted: StoredSession[] = []; + const client = new SyncServiceClient(configFor(service.url, anHourAgo()), "main", undefined, async (next) => { persisted.push(next); }); + + await client.register(); + + expect(refreshStoredSession).toHaveBeenCalledTimes(1); + expect(service.bearers).toEqual(["Bearer new-token"]); + // Persisted, because Supabase rotates the refresh token: the next start-up needs this + // one, not the one that was already spent. + expect(persisted.map((entry) => entry.accessToken)).toEqual(["new-token"]); + }); + + it("leaves a token that is still good alone", async () => { + const service = await recordingService(); + const client = new SyncServiceClient(configFor(service.url, inAnHour()), "main"); + + await client.register(); + + expect(refreshStoredSession).not.toHaveBeenCalled(); + expect(service.bearers).toEqual(["Bearer old-token"]); + }); + + it("spends the refresh token once when several calls find it expired together", async () => { + const service = await recordingService(); + // Supabase rotates the refresh token, so a second concurrent refresh would be spending + // one that no longer exists. + refreshStoredSession.mockImplementation(async () => { + await new Promise((resolve) => setTimeout(resolve, 20)); + return session(inAnHour(), "new-token"); + }); + const client = new SyncServiceClient(configFor(service.url, anHourAgo()), "main"); + + await Promise.all([client.register(), client.register(), client.register()]); + + expect(refreshStoredSession).toHaveBeenCalledTimes(1); + expect(service.bearers).toEqual(["Bearer new-token", "Bearer new-token", "Bearer new-token"]); + }); +}); diff --git a/apps/daemon/src/sync-service-client.ts b/apps/daemon/src/sync-service-client.ts index a7ec79b..d8e0c51 100644 --- a/apps/daemon/src/sync-service-client.ts +++ b/apps/daemon/src/sync-service-client.ts @@ -1,4 +1,5 @@ import { WebSocket } from "ws"; +import { refreshStoredSession, type StoredSession } from "./supabase-client.js"; import { changesResponseSchema, presenceSchema, @@ -15,6 +16,14 @@ import { /** * The daemon's half of the sync wire contract: four routes and one socket. Nothing here * knows what a shadow ref is -- it moves `FileVersion`s and hands back `Change`s. + * + * The socket is the fast path, not the only path. Where the service runs as a serverless + * function it cannot answer a WebSocket upgrade at all, so `/v1/stream` 404s on every + * attempt and no amount of reconnecting will ever help. Without a fallback that leaves the + * daemon publishing happily and receiving nothing -- an uploader, not a sync. So a failed + * stream falls back to polling `/v1/changes?since=`, the same route the daemon already + * uses to catch up after a restart. Slower, and presence goes quiet because it only exists + * on the socket, but edits still arrive. */ export type SyncStreamHandlers = { @@ -23,26 +32,52 @@ export type SyncStreamHandlers = { /** Fired on every (re)subscribe: the daemon catches up from its cursor here. */ onConnected?: () => void; onDisconnected?: () => void; + /** + * Fired on each poll while the stream is unavailable. Resolve to report the service was + * reachable, reject to report it was not; the client reports that as `connected`. + */ + onPoll?: () => Promise; }; const INITIAL_BACKOFF_MS = 500; const MAX_BACKOFF_MS = 15_000; +const POLL_INTERVAL_MS = 3_000; +/** Refresh this far ahead of expiry, so a request in flight never crosses the boundary. */ +const EXPIRY_MARGIN_MS = 60_000; export class SyncServiceClient { private socket?: WebSocket; private reconnectTimer?: NodeJS.Timeout; + private pollTimer?: NodeJS.Timeout; + private pollActive = false; + private polling = false; + private reportedPollFailure = false; private backoffMs = INITIAL_BACKOFF_MS; private stopped = false; private handlers?: SyncStreamHandlers; private cursorProvider: () => number = () => 0; + private session?: StoredSession; + private refreshing?: Promise; + constructor( private readonly config: SyncDaemonConfig, readonly branch: string, - private replicaId?: string - ) {} + private replicaId?: string, + /** Called with a refreshed session so the caller can persist it for the next start-up. */ + private readonly onSession?: (session: StoredSession) => Promise, + private readonly pollIntervalMs = POLL_INTERVAL_MS + ) { + this.session = config.service.session; + } + /** In touch with the service, by either transport -- what `crosscode status` reports. */ get connected(): boolean { + return this.streaming || this.polling; + } + + /** The socket specifically. Presence exists only here, so it asks this and not `connected`. */ + get streaming(): boolean { return this.socket?.readyState === WebSocket.OPEN; } @@ -78,7 +113,7 @@ export class SyncServiceClient { } async presence(paths: string[], actor: string, head?: string): Promise { - if (!this.connected) return; + if (!this.streaming) return; const message = { type: "presence" as const, peers: [presenceSchema.parse({ replicaId: this.requireReplicaId(), actor, branch: this.branch, paths: paths.slice(0, 50), ...(head ? { head } : {}) })] }; this.socket!.send(JSON.stringify(message)); } @@ -88,20 +123,27 @@ export class SyncServiceClient { this.handlers = handlers; this.cursorProvider = cursor; this.stopped = false; - this.connect(); + void this.connect(); } stop(): void { this.stopped = true; if (this.reconnectTimer) clearTimeout(this.reconnectTimer); + if (this.pollTimer) clearTimeout(this.pollTimer); + this.polling = false; const socket = this.socket; this.socket = undefined; socket?.close(); } - private connect(): void { + private async connect(): Promise { if (this.stopped || !this.replicaId) return; - const socket = new WebSocket(`${streamUrl(this.config.service.url)}/v1/stream`, this.authorization() ? { headers: { authorization: this.authorization()! } } : {}); + // A refresh can fail -- a revoked or rotated-away refresh token -- and that must not + // take the daemon down from a reconnect timer. Falling through to polling is the + // honest outcome: it fails the same way, and reports `connected: false`. + const authorization = await this.authorization().catch(() => undefined); + if (this.stopped) return; + const socket = new WebSocket(`${streamUrl(this.config.service.url)}/v1/stream`, authorization ? { headers: { authorization } } : {}); this.socket = socket; socket.on("open", () => { socket.send(JSON.stringify({ @@ -112,6 +154,7 @@ export class SyncServiceClient { since: this.cursorProvider() })); this.backoffMs = INITIAL_BACKOFF_MS; + this.stopPolling(); this.handlers?.onConnected?.(); }); socket.on("message", (data) => { @@ -127,6 +170,9 @@ export class SyncServiceClient { if (this.socket === socket) this.socket = undefined; this.handlers?.onDisconnected?.(); this.scheduleReconnect(); + // The socket is down for an unknown length of time -- one dropped frame, or a host + // that will never serve one. Polling covers both, and stops as soon as one opens. + this.startPolling(); }); socket.on("error", () => socket.close()); } @@ -135,22 +181,81 @@ export class SyncServiceClient { if (this.stopped) return; const delay = this.backoffMs; this.backoffMs = Math.min(this.backoffMs * 2, MAX_BACKOFF_MS); - this.reconnectTimer = setTimeout(() => this.connect(), delay); + this.reconnectTimer = setTimeout(() => void this.connect(), delay); this.reconnectTimer.unref(); } + /** + * Self-scheduling rather than an interval, so a slow catch-up never overlaps itself. + * `pollActive` is separate from `pollTimer` because the timer is unset while a tick is in + * flight, and the socket fails often enough -- once per backoff -- that starting a second + * loop in that window would be the normal case, not the rare one. + */ + private startPolling(): void { + if (this.pollActive || this.stopped || !this.handlers?.onPoll) return; + this.pollActive = true; + const tick = async () => { + this.pollTimer = undefined; + if (this.stopped || this.streaming) { this.stopPolling(); return; } + try { + await this.handlers?.onPoll?.(); + this.polling = true; + this.reportedPollFailure = false; + } catch (error) { + // Unreachable by either transport, which is what `connected: false` should mean. + // Reported once per run of failures rather than every 3s: a cursor that stops + // advancing is otherwise completely silent, and `connected: false` on its own does + // not say whether the service is down or a change cannot be applied. + if (!this.reportedPollFailure) { + this.reportedPollFailure = true; + process.stderr.write(`Crosscode: sync poll failed: ${error instanceof Error ? error.message : String(error)}\n`); + } + this.polling = false; + } + if (this.stopped || this.streaming) { this.stopPolling(); return; } + this.pollTimer = setTimeout(() => void tick(), this.pollIntervalMs); + this.pollTimer.unref(); + }; + void tick(); + } + + private stopPolling(): void { + if (this.pollTimer) clearTimeout(this.pollTimer); + this.pollTimer = undefined; + this.pollActive = false; + this.polling = false; + } + private requireReplicaId(): string { if (!this.replicaId) throw new Error("This replica is not registered yet"); return this.replicaId; } - private authorization(): string | undefined { - const token = this.config.service.session?.accessToken; - return token ? `Bearer ${token}` : undefined; + /** + * The bearer header, refreshed if it is about to expire. Supabase access tokens last an + * hour; a daemon is expected to outlive that, so every authenticated call goes through + * here rather than reading the token that was written at sign-in. + * + * Concurrent callers share one refresh via `refreshing`: the socket reconnecting and a + * publish landing at the same moment would otherwise each spend the refresh token, and + * Supabase rotates it, so the second would be spending one that no longer exists. + */ + private async authorization(): Promise { + const session = this.session; + if (!session) return undefined; + if (Date.parse(session.expiresAt) - Date.now() > EXPIRY_MARGIN_MS) return `Bearer ${session.accessToken}`; + this.refreshing ??= refreshStoredSession(session) + .then(async (refreshed) => { + this.session = refreshed; + await this.onSession?.(refreshed); + return refreshed; + }) + .finally(() => { this.refreshing = undefined; }); + return `Bearer ${(await this.refreshing).accessToken}`; } private async request(method: "GET" | "POST", path: string, body?: unknown): Promise { - const authorization = this.authorization(); + const authorization = await this.authorization(); const response = await fetch(new URL(path, this.config.service.url), { method, headers: { diff --git a/apps/daemon/src/sync-service-stub.ts b/apps/daemon/src/sync-service-stub.ts index 9a5f5a7..44e7850 100644 --- a/apps/daemon/src/sync-service-stub.ts +++ b/apps/daemon/src/sync-service-stub.ts @@ -28,6 +28,8 @@ type Subscriber = { socket: WebSocket; projectId: string; branch: string; replic export type SyncServiceStub = { url: string; changes: Change[]; + /** How many times `GET /v1/changes` has been asked, so a test can catch a spin. */ + readonly listCalls: number; /** Sequences at or below this are "no longer retained", to exercise cursor-too-old. */ retentionFloor: number; /** Drops every open stream, as a network partition would. */ @@ -35,11 +37,20 @@ export type SyncServiceStub = { close(): Promise; }; -export async function startSyncServiceStub(): Promise { +export type SyncServiceStubOptions = { + /** + * Whether the stub answers a WebSocket upgrade at all. False stands in for a serverless + * host, which cannot: the routes work, `/v1/stream` never will, and the daemon has to + * fall back to polling rather than reconnect forever. + */ + streams?: boolean; +}; + +export async function startSyncServiceStub(options: SyncServiceStubOptions = {}): Promise { const changes: Change[] = []; const subscribers = new Set(); let sequence = 0; - const state = { retentionFloor: 0 }; + const state = { retentionFloor: 0, listCalls: 0 }; const send = (response: import("node:http").ServerResponse, status: number, body: unknown) => { response.writeHead(status, { "content-type": "application/json" }); @@ -78,6 +89,7 @@ export async function startSyncServiceStub(): Promise { return; } if (request.method === "GET" && url.pathname === "/v1/changes") { + state.listCalls += 1; const query = listChangesQuerySchema.parse({ projectId: url.searchParams.get("projectId"), branch: url.searchParams.get("branch"), @@ -113,8 +125,10 @@ export async function startSyncServiceStub(): Promise { } }; - const sockets = new WebSocketServer({ server, path: "/v1/stream" }); - sockets.on("connection", (socket) => { + // No upgrade listener at all when streams are off, which is what makes node destroy the + // socket -- the client's view of a host that does not serve WebSockets. + const sockets = options.streams === false ? undefined : new WebSocketServer({ server, path: "/v1/stream" }); + sockets?.on("connection", (socket) => { let subscriber: Subscriber | undefined; socket.on("message", (data) => { const message: unknown = JSON.parse(data.toString()); @@ -145,6 +159,7 @@ export async function startSyncServiceStub(): Promise { return { url: `http://127.0.0.1:${port}`, changes, + get listCalls() { return state.listCalls; }, get retentionFloor() { return state.retentionFloor; }, set retentionFloor(value: number) { state.retentionFloor = value; }, disconnectAll() { @@ -153,7 +168,7 @@ export async function startSyncServiceStub(): Promise { }, async close() { for (const subscriber of [...subscribers]) subscriber.socket.terminate(); - sockets.close(); + sockets?.close(); // `server.close` only stops new connections; it waits for the open ones, and the // daemon's fetch keeps its sockets alive between requests. Without this the close // never resolves and the test's teardown hook dies on its own timeout instead. diff --git a/docs/protocol.md b/docs/protocol.md index 6ce6ec5..63b0aca 100644 --- a/docs/protocol.md +++ b/docs/protocol.md @@ -92,6 +92,12 @@ The client sends `subscribe` (`projectId`, `branch`, `replicaId`, `since`) and t `replicaId`, `actor`, `branch`, and up to 50 recently touched `paths`, which is how an agent can answer "who is working on what" without any ambient UI. +The stream is the fast path, not the only one. A host that cannot serve a WebSocket upgrade +— a serverless function, for one — leaves a daemon publishing happily and receiving +nothing, so a daemon that cannot open the stream falls back to polling `GET /v1/changes?since=` +every few seconds. Edits still arrive, a few seconds later than they would have. Presence +does not: it exists only on the socket, so a polling room goes quiet about who is where. + ## Daemon config and local API `syncDaemonConfig` is what the daemon stores per checkout: `projectId`, `repo`, and the diff --git a/packages/sync/src/paths.test.ts b/packages/sync/src/paths.test.ts new file mode 100644 index 0000000..a140a27 --- /dev/null +++ b/packages/sync/src/paths.test.ts @@ -0,0 +1,31 @@ +import { describe, expect, it } from "vitest"; +import { isDenied, isSafeRelativePath } from "./paths.js"; + +describe("what Crosscode refuses to sync", () => { + it("refuses the files `crosscode start` installs into the checkout", () => { + // Observed the first time two real checkouts joined one project: the teammate arrived + // holding three conflicts before typing anything, because each checkout had installed + // its own copy of these and no two shared an ancestor. + expect(isDenied(".mcp.json")).toBe(true); + expect(isDenied(".claude/settings.json")).toBe(true); + expect(isDenied(".claude/skills/crosscode/SKILL.md")).toBe(true); + }); + + it("still syncs the rest of the agent's directory, which is the team's to share", () => { + expect(isDenied(".claude/skills/deploy/SKILL.md")).toBe(false); + expect(isDenied(".claude/commands/review.md")).toBe(false); + expect(isDenied("src/index.ts")).toBe(false); + }); + + it("still refuses secrets, which is what the denylist was for first", () => { + expect(isDenied(".env")).toBe(true); + expect(isDenied(".env.local")).toBe(true); + }); + + it("refuses anything that would escape the checkout or touch git's own state", () => { + expect(isSafeRelativePath("../outside")).toBe(false); + expect(isSafeRelativePath("/etc/passwd")).toBe(false); + expect(isSafeRelativePath(".git/config")).toBe(false); + expect(isSafeRelativePath("src/a.ts")).toBe(true); + }); +}); diff --git a/packages/sync/src/paths.ts b/packages/sync/src/paths.ts index 1ceb3a8..f8d4626 100644 --- a/packages/sync/src/paths.ts +++ b/packages/sync/src/paths.ts @@ -18,5 +18,24 @@ export function isSafeRelativePath(path: string): boolean { * those would be a serious incident, so it is checked on the way out *and* on the way in. */ export function isDenied(path: string): boolean { - return redactPath(path); + return redactPath(path) || isCrosscodeOwned(path); +} + +/** + * The files `crosscode start` writes into the checkout itself: the MCP registration, the + * skill, and the hook in the agent's settings. + * + * Every checkout installs its own copy, so no two peers share an ancestor for them, and + * syncing them means the first teammate to join arrives holding a conflict on all three + * before they have typed anything. Worse, they can never converge: `start` rewrites them + * locally on every run, so a resolution is undone by the next start. + * + * They are also not content in the sense the rest of the sync means. They are per-machine + * setup, regenerated on demand by the command the invite already tells a teammate to run. + * Nothing is lost by leaving them out, and nothing is fixable by leaving them in. + */ +function isCrosscodeOwned(path: string): boolean { + return path === ".mcp.json" + || path === ".claude/settings.json" + || path.startsWith(".claude/skills/crosscode/"); }