diff --git a/packages/core/src/session/execution/worktree.ts b/packages/core/src/session/execution/worktree.ts index 8fbc5259b330..6d31f200a138 100644 --- a/packages/core/src/session/execution/worktree.ts +++ b/packages/core/src/session/execution/worktree.ts @@ -1,42 +1,80 @@ export * as WorktreeMaterializer from "./worktree" -// Rebuilds a missing project worktree from the shared store before a drain runs. This closes the -// host-local gap in cross-host resume: file tools need the tree, and a fresh worker does not have -// it. The capture side (snapshot-sync.ts) ships each snapshot as an incremental git pack; this -// side indexes every pack for the worktree into a fresh repo and checks out the newest tree. -// Ignored files and dependencies are not captured, so a bootstrap step (install, build) stays the -// project's own concern. +// Brings the project worktree to the newest state in the shared store before a drain runs. This +// closes the host-local gap in cross-host resume: file tools need the tree, and a worker that never +// ran an earlier step either has no tree at all or has one from the step it last ran. The capture +// side (snapshot-sync.ts) ships each snapshot as an incremental git pack; this side indexes every +// pack for the worktree and checks out the newest tree. Ignored files and dependencies are not +// captured, so a bootstrap step (install, build) stays the project's own concern. +// +// A tree is only ever moved forward, and only when this host's own note (snapshot/tip.ts) says it +// is behind the store. What that leaves open: two workers running tools of the SAME step on two +// hosts still cannot see each other's writes, because those are not captured until the step is +// sealed. One worker per worktree is what makes a step's tools share a tree. -import { rm, writeFile } from "node:fs/promises" +import { readdir, rm, writeFile } from "node:fs/promises" import path from "path" -import { Cause, Context, Effect, Layer } from "effect" +import { Cause, Context, Effect, Layer, Schema } from "effect" import { ChildProcess } from "effect/unstable/process" -import { asc, desc, eq } from "drizzle-orm" +import { eq } from "drizzle-orm" import { Database } from "../../database/database" import { makeGlobalNode } from "../../effect/app-node" import { KeyedMutex } from "../../effect/keyed-mutex" import { FSUtil } from "../../fs-util" import { Git } from "../../git" +import { Global } from "../../global" import { AppProcess } from "../../process" import { AbsolutePath } from "../../schema" import { SnapshotPackTable } from "../../snapshot/sql" +import { chainHead, isBehind, orderChain } from "../../snapshot/chain" +import { readWorktreeTip, writeWorktreeTip } from "../../snapshot/tip" export interface Interface { /** - * Make sure the session's directory exists, rebuilding its worktree from stored snapshot packs - * when it does not. A directory with no stored packs, or one whose worktree root already - * exists, is left alone. Never fails the caller. + * Make sure the session's directory holds the newest state the shared store has for it, + * rebuilding its worktree from stored snapshot packs when it is missing or behind. A directory + * with no stored packs, and a tree this host has neither built nor captured from, are left + * alone. + * + * A rebuild that fails dies with `WorktreeMaterializeError` rather than returning: running the + * step against whatever is in the directory tells the model those files are the project, and for + * a fresh host that is nothing at all. Tagged so the activity boundary retries it elsewhere. + * + * `pauseBeforeLock` waits between reading the directory and taking the lock. Zero everywhere but + * the check that reproduces what a concurrent drain does in that gap: nothing outside this module + * can hold a caller there, and what the check asserts is the real outcome, whether a failed + * rebuild removes a directory this call did not create. */ - readonly ensure: (directory: string) => Effect.Effect + readonly ensure: (directory: string, options?: { readonly pauseBeforeLock?: number }) => Effect.Effect } export class Service extends Context.Service()("@opencode/v2/WorktreeMaterializer") {} +// HEAD of a rebuilt tree, so the rebuilt repo reads as a clean checkout rather than an unborn +// branch over a full untracked tree. +const RESTORED = "refs/heads/opencode-restore" + +/** A rebuild that did not finish. Tagged so the boundary can tell it from a refusal and retry it. */ +export class WorktreeMaterializeError extends Schema.TaggedErrorClass()( + "WorktreeMaterializer.MaterializeError", + { message: Schema.String }, +) {} + +// Nothing in it at all, so there is no work to protect and nothing to lose by checking a tree out +// over it. Unreadable counts as not empty: a directory we cannot look into is not one to overwrite. +const isEmptyDir = (dir: string) => + Effect.promise(() => + readdir(dir) + .then((entries) => entries.length === 0) + .catch(() => false), + ) + const layer = Layer.effect( Service, Effect.gen(function* () { const fs = yield* FSUtil.Service const git = yield* Git.Service + const global = yield* Global.Service const proc = yield* AppProcess.Service const { db } = yield* Database.Service const locks = KeyedMutex.makeUnsafe() @@ -47,13 +85,15 @@ const layer = Layer.effect( .create({ worktree, gitDirectory: AbsolutePath.make(path.join(worktree, ".git")) }) .pipe(Effect.orDie) // Index every pack shipped for this worktree; objects accumulate, the newest tree wins. - const rows = yield* db + const stored = yield* db .select() .from(SnapshotPackTable) .where(eq(SnapshotPackTable.worktree, tip.worktree)) - .orderBy(asc(SnapshotPackTable.time_created)) .all() .pipe(Effect.orDie) + // A pack cannot be indexed before the one it was built on, and the write clock does not order + // them: two hosts disagree about the time, and one behind puts its pack first. + const rows = orderChain(stored) const packDirectory = path.join(repository.gitDirectory, "objects", "pack") yield* fs.ensureDir(packDirectory).pipe(Effect.orDie) for (const row of rows) { @@ -75,22 +115,21 @@ const layer = Layer.effect( // unborn branch over a full untracked tree. Cosmetic; the files above are what matter. yield* proc .run( - ChildProcess.make( - "git", - ["--git-dir", repository.gitDirectory, "update-ref", "refs/heads/opencode-restore", tip.id], - { cwd: worktree, extendEnv: true }, - ), + ChildProcess.make("git", ["--git-dir", repository.gitDirectory, "update-ref", RESTORED, tip.id], { + cwd: worktree, + extendEnv: true, + }), ) .pipe(Effect.ignore) yield* proc .run( - ChildProcess.make( - "git", - ["--git-dir", repository.gitDirectory, "symbolic-ref", "HEAD", "refs/heads/opencode-restore"], - { cwd: worktree, extendEnv: true }, - ), + ChildProcess.make("git", ["--git-dir", repository.gitDirectory, "symbolic-ref", "HEAD", RESTORED], { + cwd: worktree, + extendEnv: true, + }), ) .pipe(Effect.ignore) + yield* writeWorktreeTip(global.data, tip.worktree, tip.tree) yield* Effect.logInfo("materialized worktree from snapshot packs", { worktree: tip.worktree, packs: rows.length, @@ -98,34 +137,83 @@ const layer = Layer.effect( }) }) - const ensure = Effect.fn("WorktreeMaterializer.ensure")(function* (directory: string) { - if (yield* fs.existsSafe(directory)) return - // The newest capture whose session ran in this directory decides which worktree to rebuild. - const tip = yield* db + // Whether the tree on this host is older than what the store holds. No note means the tree was + // neither built from packs nor captured from here, so it belongs to whoever put it there. A + // note the store has never seen means a capture that never shipped, so this host holds work + // nothing else has and must not be moved back to an older tree. + const behind = Effect.fnUntraced(function* (tip: typeof SnapshotPackTable.$inferSelect) { + const held = yield* readWorktreeTip(global.data, tip.worktree) + if (!held || held === tip.tree) return false + const rows = yield* db .select() .from(SnapshotPackTable) - .where(eq(SnapshotPackTable.directory, directory)) - .orderBy(desc(SnapshotPackTable.time_created)) - .limit(1) - .get() + .where(eq(SnapshotPackTable.worktree, tip.worktree)) + .all() .pipe(Effect.orDie) + return isBehind(rows, held) + }) + + const ensure = Effect.fn("WorktreeMaterializer.ensure")(function* ( + directory: string, + options?: { readonly pauseBeforeLock?: number }, + ) { + // The newest capture whose session ran in this directory decides which worktree to rebuild, + // and which state a tree that is already here has to be brought to. + const tip = chainHead( + yield* db + .select() + .from(SnapshotPackTable) + .where(eq(SnapshotPackTable.directory, directory)) + .all() + .pipe(Effect.orDie), + ) if (!tip) return + // An empty directory is not somebody's working copy, so the rule that protects one does not + // apply to it. Treating it as present is what stops a fresh host from ever building the tree: + // it has no tip note, so `behind` says no, and the tools then run against nothing. A mounted + // path that exists but holds nothing is the ordinary shape of a host that has never seen this + // project, which is exactly the case the packs are for. + const present = (yield* fs.existsSafe(tip.worktree)) && !(yield* isEmptyDir(tip.worktree)) + // `behind` is already the whole rule. It is false unless this host has a note of its own, and + // a note means this host agreed to that state: either it built the tree from packs or it + // captured the tree from here. Moving it forward from a state it agreed to loses nothing. The + // rebuild marker is not part of the rule: a host that seeded the session from its own checkout + // has a note but no marker, and gating on the marker would refuse every activity it draws once + // any other host ships. A developer's checkout is protected by having no note at all. + if (present && !(yield* behind(tip))) return + if (options?.pauseBeforeLock) yield* Effect.sleep(options.pauseBeforeLock) yield* locks.withLock(tip.worktree)( Effect.gen(function* () { - // Re-check inside the lock (a concurrent drain may have rebuilt it), and never touch a - // worktree root that already exists: something else owns that tree. - if (yield* fs.existsSafe(tip.worktree)) return + // Re-check inside the lock: a concurrent drain may have done this already. Same notion of + // present as above, or an empty directory bails out here instead and the tree that the + // outer check just decided to build never gets built. + const here = (yield* fs.existsSafe(tip.worktree)) && !(yield* isEmptyDir(tip.worktree)) + if (here && !(yield* behind(tip))) return yield* materialize(tip).pipe( Effect.catchCauseIf( (cause) => !Cause.hasInterrupts(cause), (cause) => Effect.gen(function* () { - // A half-built tree would pass the exists check forever; remove what we created. - yield* Effect.promise(() => rm(tip.worktree, { recursive: true, force: true })) - yield* Effect.logWarning("failed to materialize worktree", { + // A half-built tree would pass the exists check forever, so what we created is + // removed. A tree that was already here is not ours to remove: a failed refresh + // leaves it as stale as it was. Asked of the reading taken inside the lock, which + // is the only one that describes the directory this attempt started from: the + // outer one is why the re-check exists, and a drain that materialized while this + // one waited makes it name a directory that no longer exists. + if (!here) yield* Effect.promise(() => rm(tip.worktree, { recursive: true, force: true })) + yield* Effect.logError("failed to materialize worktree", { worktree: tip.worktree, cause, }) + // Swallowing this ran the step against whatever was in the directory, which for + // a fresh host is nothing at all. Tagged, so the activity boundary can retry it: + // git and the filesystem fail for reasons that pass, and the alternative is a + // turn that fails for good because one worker had a bad minute. + return yield* Effect.die( + new WorktreeMaterializeError({ + message: `could not materialize ${tip.worktree} at ${tip.tree}`, + }), + ) }), ), ) @@ -140,5 +228,5 @@ const layer = Layer.effect( export const node = makeGlobalNode({ service: Service, layer, - deps: [Database.node, FSUtil.node, Git.node, AppProcess.node], + deps: [Database.node, FSUtil.node, Git.node, Global.node, AppProcess.node], }) diff --git a/packages/core/src/snapshot-sync.ts b/packages/core/src/snapshot-sync.ts index 99541265e763..be29569430c7 100644 --- a/packages/core/src/snapshot-sync.ts +++ b/packages/core/src/snapshot-sync.ts @@ -10,7 +10,7 @@ import os from "node:os" import path from "path" import { Cause, Context, Effect, Layer } from "effect" import { ChildProcess } from "effect/unstable/process" -import { desc, eq } from "drizzle-orm" +import { eq } from "drizzle-orm" import { Database } from "./database/database" import { makeLocationNode } from "./effect/app-node" import { FSUtil } from "./fs-util" @@ -21,10 +21,12 @@ import { AppProcess } from "./process" import { AbsolutePath } from "./schema" import type { Snapshot } from "./snapshot" import { SnapshotPackTable } from "./snapshot/sql" +import { chainHead } from "./snapshot/chain" +import { readWorktreeTip, writeWorktreeTip } from "./snapshot/tip" import { Hash } from "./util/hash" export interface Interface { - /** Ship a captured tree to the shared store as an incremental pack. Never fails the caller. */ + /** A stale tip fails the caller; packing and insertion errors are logged. */ readonly push: (tree: Snapshot.ID) => Effect.Effect } @@ -61,19 +63,39 @@ const layer = Layer.effect( { stdin }, ) + // The newest state the store holds for this worktree, read off the chain the packs form rather + // than off `time_created`, which is whichever host wrote the row. + const newest = () => + db + .select() + .from(SnapshotPackTable) + .where(eq(SnapshotPackTable.worktree, worktree)) + .all() + .pipe(Effect.orDie, Effect.map(chainHead)) + const push = Effect.fn("SnapshotSync.push")(function* (tree: Snapshot.ID) { + // A host that has not caught up must not publish its older files as the next tree. + // This reading is not an atomic head claim and does not fence concurrent publishers. + if (source) { + const stoodOn = yield* readWorktreeTip(global.data, worktree) + const ahead = yield* newest() + if (ahead && ahead.tree !== tree && stoodOn !== ahead.tree) { + yield* Effect.die( + new Error( + `refusing to ship ${worktree}: this host stood on ${stoodOn ?? "nothing"}, ` + + `and the store is at ${ahead.tree}`, + ), + ) + } + } yield* Effect.gen(function* () { if (!source) return - const latest = yield* db - .select() - .from(SnapshotPackTable) - .where(eq(SnapshotPackTable.worktree, worktree)) - .orderBy(desc(SnapshotPackTable.time_created)) - .limit(1) - .get() - .pipe(Effect.orDie) - // The newest shipped state already is this tree: nothing to pack. - if (latest?.tree === tree) return + const latest = yield* newest() + // The newest shipped state already is this tree: nothing to pack, and the note is true. + if (latest?.tree === tree) { + yield* writeWorktreeTip(global.data, worktree, tree) + return + } // Chain onto the previous sync commit only when this host has it; a base absent locally // would produce a delta pack the pack builder cannot compute. const base = @@ -107,6 +129,8 @@ const layer = Layer.effect( .onConflictDoNothing() .run() .pipe(Effect.orDie) + // A note must not name a state whose insertion failed. + yield* writeWorktreeTip(global.data, worktree, tree) }).pipe( Effect.catchCauseIf( (cause) => !Cause.hasInterrupts(cause), diff --git a/packages/core/src/snapshot/chain.ts b/packages/core/src/snapshot/chain.ts new file mode 100644 index 000000000000..efe7b01740ea --- /dev/null +++ b/packages/core/src/snapshot/chain.ts @@ -0,0 +1,66 @@ +// Parent depth avoids ordering an intact chain by clocks from different hosts. +// Forks still use the timestamp tiebreaker. Ordering does not reject a competing publication. + +export interface ChainRow { + readonly id: string + readonly base: string | null + readonly time_created: number +} + +const depths = (rows: readonly T[]): Map => { + const byID = new Map(rows.map((row) => [row.id, row])) + const depth = new Map() + // Iterative, because the chain is one link per capture and nothing prunes it: a long session + // would put a stack frame per tool call that changed a file. + for (const start of rows) { + if (depth.has(start.id)) continue + const pending: T[] = [] + const seen = new Set() + let at: T | undefined = start + while (at && !depth.has(at.id) && !seen.has(at.id)) { + seen.add(at.id) + pending.push(at) + at = at.base ? (byID.get(at.base) as T | undefined) : undefined + } + // A root, a row whose base is not in the store, or a cycle: all start the count at zero. + let below = at && depth.has(at.id) ? depth.get(at.id)! : -1 + for (const row of pending.reverse()) depth.set(row.id, ++below) + } + return depth +} + +/** Packs in an order where a pack's base always comes before it, which is what indexing them needs. */ +export const orderChain = (rows: readonly T[]): T[] => { + const depth = depths(rows) + return [...rows].sort((a, b) => (depth.get(a.id) ?? 0) - (depth.get(b.id) ?? 0) || a.time_created - b.time_created) +} + +/** The newest state the store holds, which is the deepest link in the chain. */ +export const chainHead = (rows: readonly T[], depth = depths(rows)): T | undefined => { + let head: T | undefined + for (const row of rows) { + if (!head) { + head = row + continue + } + const here = depth.get(row.id) ?? 0 + const best = depth.get(head.id) ?? 0 + if (here > best || (here === best && row.time_created > head.time_created)) head = row + } + return head +} + +/** + * Whether `tree` is an earlier state than the head, as opposed to one the store has never seen. + * A tree the store does not hold is this host's own uncaptured work, and moving off it would drop + * work nothing else has. + */ +export const isBehind = (rows: readonly T[], tree: string): boolean => { + const depth = depths(rows) + const head = chainHead(rows, depth) + if (!head || head.tree === tree) return false + const mine = rows.filter((row) => row.tree === tree) + if (mine.length === 0) return false + const deepest = Math.max(...mine.map((row) => depth.get(row.id) ?? 0)) + return deepest < (depth.get(head.id) ?? 0) +} diff --git a/packages/core/src/snapshot/tip.ts b/packages/core/src/snapshot/tip.ts new file mode 100644 index 000000000000..b3495ea74dfd --- /dev/null +++ b/packages/core/src/snapshot/tip.ts @@ -0,0 +1,34 @@ +// The state a host's worktree is known to hold, kept in that host's own data directory. +// +// The shared store says what the newest captured tree is; it cannot say whether THIS host has it. +// The git objects cannot say either: a capture writes its tree into the side snapshot repository +// while a materialization indexes packs into the project repository, so "the tree is present" has +// two different answers depending on which host produced it. A host-local note is the one place +// both paths can agree on. +// +// It is written whenever this host captures (snapshot-sync) or rebuilds (execution/worktree), and +// read to answer one question: is this tree behind the store, or ahead of it? Being wrong towards +// "ahead" only leaves a stale tree; being wrong towards "behind" checks out over work nothing else +// holds, so an unreadable or missing note always means "leave the tree alone". + +import { mkdir, readFile, rm, writeFile } from "node:fs/promises" +import path from "path" +import { Effect } from "effect" +import { Hash } from "../util/hash" + +const noteFile = (data: string, worktree: string) => path.join(data, "worktree-tip", `${Hash.fast(worktree)}.tree`) + +export const readWorktreeTip = (data: string, worktree: string) => + Effect.promise(() => + readFile(noteFile(data, worktree), "utf8").then( + (text) => text.trim() || undefined, + () => undefined, + ), + ) + +export const writeWorktreeTip = (data: string, worktree: string, tree: string) => + Effect.promise(async () => { + const file = noteFile(data, worktree) + await mkdir(path.dirname(file), { recursive: true }).catch(() => {}) + await writeFile(file, tree).catch(() => {}) + }) diff --git a/packages/core/test/snapshot-chain.test.ts b/packages/core/test/snapshot-chain.test.ts new file mode 100644 index 000000000000..012e2bd3d0bf --- /dev/null +++ b/packages/core/test/snapshot-chain.test.ts @@ -0,0 +1,58 @@ +// The packs form a chain, and the chain is what orders them. `time_created` is whichever host +// wrote the row, and hosts do not agree on the time, so ordering by it would let a worker whose +// clock is behind make its older tree the newest one that every other host then checks out. +import { describe, expect, test } from "bun:test" +import { chainHead, isBehind, orderChain } from "@opencode-ai/core/snapshot/chain" + +const row = (id: string, base: string | null, time: number, tree = `tree-${id}`) => ({ + id, + base, + time_created: time, + tree, +}) + +describe("snapshot chain", () => { + test("orders a chain by its links, not by the write clock", () => { + // Written by a host five minutes behind, so `b` claims an earlier time than its own parent. + const rows = [row("b", "a", 1_000), row("a", null, 300_000), row("c", "b", 2_000)] + expect(orderChain(rows).map((r) => r.id)).toEqual(["a", "b", "c"]) + }) + + test("the head is the deepest link, whatever the clock says", () => { + const rows = [row("a", null, 300_000), row("b", "a", 1_000), row("c", "b", 2_000)] + expect(chainHead(rows)?.id).toBe("c") + }) + + test("an empty store has no head", () => { + expect(chainHead([])).toBeUndefined() + }) + + test("a fork is decided by depth, and the clock only breaks a tie", () => { + // `x` and `y` both build on `a`. `y` is deeper, so it wins even though `x` was written later. + const rows = [row("a", null, 1), row("x", "a", 99_000), row("y", "a", 2), row("z", "y", 3)] + expect(chainHead(rows)?.id).toBe("z") + }) + + test("a row whose base is not in the store is treated as a root", () => { + const rows = [row("b", "missing", 5), row("c", "b", 6)] + expect(orderChain(rows).map((r) => r.id)).toEqual(["b", "c"]) + expect(chainHead(rows)?.id).toBe("c") + }) + + test("behind means earlier in the chain, not earlier on a clock", () => { + const rows = [row("a", null, 300_000), row("b", "a", 1_000)] + expect(isBehind(rows, "tree-a")).toBe(true) + expect(isBehind(rows, "tree-b")).toBe(false) + }) + + test("a tree the store has never seen is this host's own work, not a state behind", () => { + const rows = [row("a", null, 1), row("b", "a", 2)] + expect(isBehind(rows, "tree-never-shipped")).toBe(false) + }) + + test("a row that names itself as its base does not run the stack out", () => { + const rows = [row("a", "a", 1), row("b", "a", 2)] + expect(() => chainHead(rows)).not.toThrow() + expect(chainHead(rows)?.id).toBe("b") + }) +}) diff --git a/packages/core/test/worktree-materialize.test.ts b/packages/core/test/worktree-materialize.test.ts index 70d18a155344..3e004606a86f 100644 --- a/packages/core/test/worktree-materialize.test.ts +++ b/packages/core/test/worktree-materialize.test.ts @@ -5,10 +5,10 @@ import { describe, expect } from "bun:test" import { $ } from "bun" import { realpathSync } from "node:fs" -import { mkdir, readFile, rm, writeFile } from "node:fs/promises" +import { mkdir, readdir, readFile, rm, writeFile } from "node:fs/promises" import path from "path" import { asc } from "drizzle-orm" -import { Effect, Layer } from "effect" +import { Effect, Fiber, Layer } from "effect" import { AppNodeBuilder } from "@opencode-ai/core/effect/app-node-builder" import { LayerNode } from "@opencode-ai/core/effect/layer-node" import { Database } from "@opencode-ai/core/database/database" @@ -18,6 +18,7 @@ import { AbsolutePath } from "@opencode-ai/core/schema" import { Snapshot } from "@opencode-ai/core/snapshot" import { SnapshotSync } from "@opencode-ai/core/snapshot-sync" import { SnapshotPackTable } from "@opencode-ai/core/snapshot/sql" +import { writeWorktreeTip } from "@opencode-ai/core/snapshot/tip" import { WorktreeMaterializer } from "@opencode-ai/core/session/execution/worktree" import { testEffect } from "./lib/effect" import { tmpdir } from "./fixture/tmpdir" @@ -33,9 +34,14 @@ const captureStack = (file: string, worktree: string, data: string) => [Global.node, Layer.succeed(Global.Service, Global.make({ data }))], ]) -// The resuming host: only the shared store, no location, no snapshot repo, no worktree. -const materializeStack = (file: string) => - AppNodeBuilder.build(WorktreeMaterializer.node, [[Database.node, Database.layerFromPath(file)]]) +// The resuming host: the shared store and its own data directory, no location, no snapshot repo, +// no worktree. The data directory is what makes two of these independent hosts: it holds the note +// saying which state that host's tree is at. +const materializeStack = (file: string, data: string) => + AppNodeBuilder.build(WorktreeMaterializer.node, [ + [Database.node, Database.layerFromPath(file)], + [Global.node, Layer.succeed(Global.Service, Global.make({ data }))], + ]) describe("WorktreeMaterializer", () => { it.live("rebuilds a deleted worktree from shared-store packs, incremental chain included", () => @@ -80,7 +86,7 @@ describe("WorktreeMaterializer", () => { // The fresh host: the tree is gone, only the shared store remains. yield* Effect.promise(() => rm(worktree, { recursive: true, force: true })) - const B = yield* Layer.build(materializeStack(file)) + const B = yield* Layer.build(materializeStack(file, path.join(root, "host-b-data"))) yield* WorktreeMaterializer.Service.use((w) => w.ensure(worktree)).pipe(Effect.provide(B)) const [tracked, untracked, extra] = yield* Effect.promise(() => @@ -102,6 +108,321 @@ describe("WorktreeMaterializer", () => { }), ) + // A rebuild that fails removes what it created, and only that. The reading it asks is the one + // taken inside the lock: another drain can fill the directory while this one waits for it, and + // the reading from before the wait then names a directory that no longer exists. What that costs + // is not the rebuild, which retries, but the files git ignores in what it removed: an install, a + // build, a `.env`. `pauseBeforeLock` is the wait, and it is the only thing invented here. + it.live("keeps a directory another drain filled while a failed rebuild waited for the lock", () => + Effect.gen(function* () { + const tmp = yield* Effect.promise(() => tmpdir()) + const root = realpathSync(tmp.path) + const worktree = path.join(root, "project") + const file = path.join(root, "shared.db") + const data = path.join(root, "host-b-data") + yield* Effect.promise(async () => { + await mkdir(worktree, { recursive: true }) + await $`git init -q ${worktree}`.quiet() + await $`git -C ${worktree} config user.email t@t`.quiet() + await $`git -C ${worktree} config user.name t`.quiet() + await writeFile(path.join(worktree, "tracked.txt"), "v1\n") + await $`git -C ${worktree} add .`.quiet() + await $`git -C ${worktree} commit -qm seed`.quiet() + }) + + const A = yield* Layer.build(captureStack(file, worktree, path.join(root, "host-a-data"))) + const first = yield* Snapshot.Service.use((s) => s.capture()).pipe(Effect.provide(A)) + if (!first) throw new Error("expected a capture") + yield* SnapshotSync.Service.use((s) => s.push(first)).pipe(Effect.provide(A)) + const stored = yield* Database.Service.use(({ db }) => db.select().from(SnapshotPackTable).all()).pipe( + Effect.orDie, + Effect.provide(Database.layerFromPath(file)), + Effect.scoped, + ) + + // The newest state in the store, and a pack that is not a pack: indexing it is how a rebuild + // fails for reasons the store cannot rule out. + yield* Effect.sleep(10) + yield* Database.Service.use(({ db }) => + db + .insert(SnapshotPackTable) + .values([ + { + id: "f".repeat(40), + directory: worktree, + worktree, + tree: "e".repeat(40), + base: stored[0]!.id, + pack: Buffer.from([0x50, 0x41, 0x43, 0x4b]), + }, + ]) + .run(), + ).pipe(Effect.orDie, Effect.provide(Database.layerFromPath(file)), Effect.scoped) + + // This host is empty and behind, which is the state that decides to rebuild. + yield* Effect.promise(() => rm(worktree, { recursive: true, force: true })) + const B = yield* Layer.build(materializeStack(file, data)) + const rebuilding = yield* WorktreeMaterializer.Service.use((w) => + w.ensure(worktree, { pauseBeforeLock: 400 }), + ).pipe(Effect.provide(B), Effect.exit, Effect.forkChild) + + // What another drain leaves behind while this one waits: a checkout, the files git ignores, + // and the note saying this host agreed to that state. + yield* Effect.sleep(150) + yield* Effect.promise(async () => { + await mkdir(worktree, { recursive: true }) + await writeFile(path.join(worktree, "tracked.txt"), "v1\n") + await writeFile(path.join(worktree, ".env"), "SECRET=1\n") + }) + yield* writeWorktreeTip(data, worktree, stored[0]!.tree) + + const outcome = yield* Fiber.join(rebuilding) + // The rebuild really did fail, which is the premise: a check where it succeeded would say + // nothing about what a failure removes. + expect(outcome._tag).toBe("Failure") + + // The rebuild failed on the bad pack. The packs would restore `tracked.txt` on a retry; the + // ignored file is in no pack and nothing else has a copy. + const left = yield* Effect.promise(() => readdir(worktree).catch(() => [] as string[])) + // A `.git` the failed rebuild made on its way is fine; what must survive is the other drain's + // work, and above all the file no pack carries. + expect(left).toContain("tracked.txt") + expect(left).toContain(".env") + + yield* Effect.promise(() => tmp[Symbol.asyncDispose]()) + }), + ) + + it.live("rebuilds into a directory that exists but is empty", () => + Effect.gen(function* () { + const tmp = yield* Effect.promise(() => tmpdir()) + const root = realpathSync(tmp.path) + const worktree = path.join(root, "project") + const file = path.join(root, "shared.db") + yield* Effect.promise(async () => { + await mkdir(worktree, { recursive: true }) + await $`git init -q ${worktree}`.quiet() + await $`git -C ${worktree} config user.email t@t`.quiet() + await $`git -C ${worktree} config user.name t`.quiet() + await writeFile(path.join(worktree, "note.txt"), "travelled\n") + await $`git -C ${worktree} add .`.quiet() + await $`git -C ${worktree} commit -qm seed`.quiet() + }) + + const A = yield* Layer.build(captureStack(file, worktree, path.join(root, "host-a-data"))) + const captured = yield* Snapshot.Service.use((s) => s.capture()).pipe(Effect.provide(A)) + if (!captured) throw new Error("expected a capture") + yield* SnapshotSync.Service.use((s) => s.push(captured)).pipe(Effect.provide(A)) + + // The shape a container gives a fresh host: the path is there because something mounted it, + // and there is nothing in it. Deleting the directory instead is the case already covered, + // and it is the easy one: an absent tree is obviously safe to build. + yield* Effect.promise(async () => { + await rm(worktree, { recursive: true, force: true }) + await mkdir(worktree, { recursive: true }) + }) + + const B = yield* Layer.build(materializeStack(file, path.join(root, "host-b-data"))) + yield* WorktreeMaterializer.Service.use((w) => w.ensure(worktree)).pipe(Effect.provide(B)) + + expect(yield* Effect.promise(() => readFile(path.join(worktree, "note.txt"), "utf8"))).toBe("travelled\n") + + yield* Effect.promise(() => tmp[Symbol.asyncDispose]()) + }), + ) + + it.live("moves a tree that is behind forward, and leaves one already at the tip alone", () => + Effect.gen(function* () { + const tmp = yield* Effect.promise(() => tmpdir()) + const root = realpathSync(tmp.path) + const worktree = path.join(root, "project") + const file = path.join(root, "shared.db") + const tracked = path.join(worktree, "tracked.txt") + const content = () => Effect.promise(() => readFile(tracked, "utf8")) + const put = (text: string) => Effect.promise(() => writeFile(tracked, text)) + yield* Effect.promise(async () => { + await mkdir(worktree, { recursive: true }) + await $`git init -q ${worktree}`.quiet() + await $`git -C ${worktree} config user.email t@t`.quiet() + await $`git -C ${worktree} config user.name t`.quiet() + await writeFile(tracked, "v1\n") + await $`git -C ${worktree} add .`.quiet() + await $`git -C ${worktree} commit -qm seed`.quiet() + }) + + const A = yield* Layer.build(captureStack(file, worktree, path.join(root, "host-a-data"))) + const ship = Effect.gen(function* () { + const tree = yield* Snapshot.Service.use((s) => s.capture()) + if (!tree) throw new Error("expected a capture to produce a tree") + yield* SnapshotSync.Service.use((s) => s.push(tree)) + }).pipe(Effect.provide(A)) + yield* ship + + // A checkout this host already had is somebody's working copy: however far behind the store + // it is, checking a stored tree out over it would rewrite files under whoever owns them. + const C = yield* Layer.build(materializeStack(file, path.join(root, "host-c-data"))) + yield* put("mine\n") + yield* WorktreeMaterializer.Service.use((w) => w.ensure(worktree)).pipe(Effect.provide(C)) + expect(yield* content()).toBe("mine\n") + + // Host B builds the tree from the store, which is what makes it B's to move. + yield* Effect.promise(() => rm(worktree, { recursive: true, force: true })) + const B = yield* Layer.build(materializeStack(file, path.join(root, "host-b-data"))) + const ensureB = WorktreeMaterializer.Service.use((w) => w.ensure(worktree)).pipe(Effect.provide(B)) + yield* ensureB + expect(yield* content()).toBe("v1\n") + + // The store moves on while B's tree does not: another host captured a newer state, which is + // what a worker picking up a later step of the same session arrives to. + yield* Effect.sleep(10) + yield* put("v2\n") + yield* ship + yield* put("v1\n") + + yield* ensureB + + // Rebuilding only a missing tree is not enough: a present one is served as it was left, and + // the tools of the step read files from whichever step this worker last ran. + expect(yield* content()).toBe("v2\n") + + // A tree already at the newest state is left alone whatever is in it. This is what keeps the + // later tools of a step from checking out over what its earlier tools wrote, since nothing + // captures those until the step is sealed. + yield* put("uncaptured\n") + yield* ensureB + expect(yield* content()).toBe("uncaptured\n") + + yield* Effect.promise(() => tmp[Symbol.asyncDispose]()) + }), + ) + + // The write direction. A host the store has moved past must not pack its older files: by time it + // would become the newest, and every other host would check that out over the work they were + // shipped to carry. The same rule as the read direction, applied where it was missing. + it.live("refuses to ship from a host the store has moved past", () => + Effect.gen(function* () { + const tmp = yield* Effect.promise(() => tmpdir()) + const root = realpathSync(tmp.path) + const worktree = path.join(root, "project") + const file = path.join(root, "shared.db") + yield* Effect.promise(async () => { + await mkdir(worktree, { recursive: true }) + await $`git init -q ${worktree}`.quiet() + await $`git -C ${worktree} config user.email t@t`.quiet() + await $`git -C ${worktree} config user.name t`.quiet() + await writeFile(path.join(worktree, "f.txt"), "v1\n") + await $`git -C ${worktree} add .`.quiet() + await $`git -C ${worktree} commit -qm seed`.quiet() + }) + + const A = yield* Layer.build(captureStack(file, worktree, path.join(root, "a-data"))) + const first = yield* Snapshot.Service.use((s) => s.capture()).pipe(Effect.provide(A)) + yield* SnapshotSync.Service.use((s) => s.push(first!)).pipe(Effect.provide(A)) + + // Another host ships while this one is not looking. Written straight into the store, because + // two capture stacks for one worktree resolve to the same host: the node builder keys them by + // location, so the second host has to be the row rather than a second stack. + const elsewhere = "e".repeat(40) + yield* Effect.sleep(10) + yield* Database.Service.use(({ db }) => + db + .insert(SnapshotPackTable) + .values([ + { + id: "d".repeat(40), + directory: worktree, + worktree, + tree: elsewhere, + pack: Buffer.from([0x50, 0x41, 0x43, 0x4b]), + }, + ]) + .run(), + ).pipe(Effect.orDie, Effect.provide(Database.layerFromPath(file)), Effect.scoped) + + // This host is still standing on `first`, so what it holds is not built on what the store now + // says the project is. Shipping it would revert the other host. + yield* Effect.promise(() => writeFile(path.join(worktree, "f.txt"), "stale\n")) + const stale = yield* Snapshot.Service.use((s) => s.capture()).pipe(Effect.provide(A)) + const exit = yield* SnapshotSync.Service.use((s) => s.push(stale!)).pipe(Effect.provide(A), Effect.exit) + expect(exit._tag).toBe("Failure") + + // Nothing was added, and the note was not moved either: a refused ship must leave this host + // saying what it actually holds. + const rows = yield* Database.Service.use(({ db }) => + db.select().from(SnapshotPackTable).orderBy(asc(SnapshotPackTable.time_created)).all(), + ).pipe(Effect.orDie, Effect.provide(Database.layerFromPath(file)), Effect.scoped) + expect(rows).toHaveLength(2) + expect(rows[1]?.tree).toBe(elsewhere) + + yield* Effect.promise(() => tmp[Symbol.asyncDispose]()) + }), + ) + + // A host that seeded the session from its own checkout has a note but no rebuild marker, because + // only a rebuild writes one. Gating the move on that marker would fail every activity such a host + // draws as soon as any other host ships. The note is the rule: a host that agreed to a state may + // be moved off it. + it.live("moves a tree the host captured rather than rebuilt", () => + Effect.gen(function* () { + const tmp = yield* Effect.promise(() => tmpdir()) + const root = realpathSync(tmp.path) + const worktree = path.join(root, "project") + const file = path.join(root, "shared.db") + const data = path.join(root, "seed-host-data") + yield* Effect.promise(async () => { + await mkdir(worktree, { recursive: true }) + await $`git init -q ${worktree}`.quiet() + await $`git -C ${worktree} config user.email t@t`.quiet() + await $`git -C ${worktree} config user.name t`.quiet() + await writeFile(path.join(worktree, "f.txt"), "seeded\n") + await $`git -C ${worktree} add .`.quiet() + await $`git -C ${worktree} commit -qm seed`.quiet() + }) + + // This host captures from its own checkout, so it gets a note and no rebuild marker. + const A = yield* Layer.build(captureStack(file, worktree, data)) + const first = yield* Snapshot.Service.use((s) => s.capture()).pipe(Effect.provide(A)) + yield* SnapshotSync.Service.use((s) => s.push(first!)).pipe(Effect.provide(A)) + const packs = yield* Database.Service.use(({ db }) => db.select().from(SnapshotPackTable).all()).pipe( + Effect.orDie, + Effect.provide(Database.layerFromPath(file)), + Effect.scoped, + ) + + // Another host ships on top, so this one is behind. + yield* Effect.sleep(10) + yield* Database.Service.use(({ db }) => + db + .insert(SnapshotPackTable) + .values([ + { + id: "d".repeat(40), + directory: worktree, + worktree, + tree: "e".repeat(40), + base: packs[0]!.id, + pack: Buffer.from([0x50, 0x41, 0x43, 0x4b]), + }, + ]) + .run(), + ).pipe(Effect.orDie, Effect.provide(Database.layerFromPath(file)), Effect.scoped) + + const B = yield* Layer.build(materializeStack(file, data)) + const exit = yield* WorktreeMaterializer.Service.use((w) => w.ensure(worktree)).pipe( + Effect.provide(B), + Effect.exit, + ) + + // The pack above is not a real one, so the rebuild itself cannot succeed here. What this pins + // is which failure: a rebuild that was attempted and failed, not a refusal to try. + const why = String(exit) + expect(why).not.toContain("was not built") + expect(why).toContain("could not materialize") + + yield* Effect.promise(() => tmp[Symbol.asyncDispose]()) + }), + ) + // The shared-store deployment uses the libsql backend, so the pack blob has to survive that // driver's parameter path too, not only bun's. it.live("round-trips a pack blob through the libsql backend", () => @@ -126,3 +447,4 @@ describe("WorktreeMaterializer", () => { }), ) }) + diff --git a/packages/temporal/README.md b/packages/temporal/README.md index ccf24e2063ee..707f47649a4c 100644 --- a/packages/temporal/README.md +++ b/packages/temporal/README.md @@ -181,3 +181,8 @@ Host-local state that does not ride the store: the snapshot object store and the tool-output files under the data directory. The runner never reads these to rebuild context; they only affect the diff, restore and full-output features. Point the data directory at shared storage to make them portable. + +One rule bounds what the rebuild may touch, because checking a stored tree out over the wrong one +destroys work: a tree is moved only when a host-local note (`snapshot/tip.ts`) says this host is +behind the store, so a host holding a capture that never shipped is left alone. Packs are ordered +by their chain, not by `time_created`, because hosts do not agree on the time.