Skip to content

feat(core): Brought a worker's stale worktree forward before it drains. #14

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
moetemp wants to merge 1 commit into from
Closed

feat(core): Brought a worker's stale worktree forward before it drains. #14

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
168 changes: 128 additions & 40 deletions packages/core/src/session/execution/worktree.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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<void>
readonly ensure: (directory: string, options?: { readonly pauseBeforeLock?: number }) => Effect.Effect<void>
}

export class Service extends Context.Service<Service, Interface>()("@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<WorktreeMaterializeError>()(
"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<string>()
Expand All @@ -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) {
Expand All @@ -75,57 +115,105 @@ 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,
tree: tip.tree,
})
})

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}`,
}),
)
}),
),
)
Expand All @@ -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],
})
48 changes: 36 additions & 12 deletions packages/core/src/snapshot-sync.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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"
Expand All @@ -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<void>
}

Expand Down Expand Up @@ -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 =
Expand Down Expand Up @@ -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),
Expand Down
66 changes: 66 additions & 0 deletions packages/core/src/snapshot/chain.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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 = <T extends ChainRow>(rows: readonly T[]): Map<string, number> => {
const byID = new Map(rows.map((row) => [row.id, row]))
const depth = new Map<string, number>()
// 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<string>()
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 = <T extends ChainRow>(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 = <T extends ChainRow>(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 = <T extends ChainRow & { readonly tree: string }>(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)
}
Loading
Loading