fix(core): coordinate snapshot writers

Co-authored-by: Hona <10430890+Hona@users.noreply.github.com>

packages/core/src/snapshot.ts

@@ -2,7 +2,7 @@ export * as Snapshot from "./snapshot.js"
 
 import { makeLocationNode } from "@opencode-ai/util/effect/app-node"
 import path from "path"
-import { Context, Effect, Fiber, Layer, Schema, Scope } from "effect"
+import { Context, Duration, Effect, Fiber, Layer, Schema, Scope } from "effect"
 import { File } from "./file.js"
 import { FSUtil } from "@opencode-ai/util/fs-util"
 import { Git } from "./git.js"
@@ -12,6 +12,7 @@ import { AbsolutePath, RelativePath } from "./schema.js"
 import { ID } from "@opencode-ai/schema/snapshot"
 import { Hash } from "@opencode-ai/util/hash"
 import { State } from "./state.js"
+import { EffectFlock } from "@opencode-ai/util/effect-flock"
 
 export { ID }
 
@@ -69,6 +70,11 @@ export interface Interface extends State.Transformable<Draft> {
 
 export class Service extends Context.Service<Service, Interface>()("@opencode/Snapshot") {}
 
+const LOCK_TIMEOUT_MS = 30_000
+const CAPTURE_LOCK_TIMEOUT_MS = 100
+const INITIAL_LOCK_RETRY_MS = 5_000
+const MAX_LOCK_RETRY_MS = 60_000
+
 const layer = Layer.effect(
   Service,
   Effect.gen(function* () {
@@ -76,6 +82,7 @@ const layer = Layer.effect(
     const git = yield* Git.Service
     const global = yield* Global.Service
     const location = yield* Location.Service
+    const flock = yield* EffectFlock.Service
     const lifetime = yield* Scope.Scope
     const state = State.create<{ enabled: boolean }, Draft>({
       name: "snapshot",
@@ -87,7 +94,7 @@ const layer = Layer.effect(
       }),
     })
     // Cache a scope-owned fiber so caller cancellation stops waiting without poisoning shared initialization.
-    const repositoryFiber = yield* Effect.cached(
+    const [repositoryFiber, invalidateRepository] = yield* Effect.cachedInvalidateWithTTL(
       Effect.gen(function* () {
         const source = yield* git.repo.discover(location.project.directory)
         if (!source) return yield* new Error({ operation: "capture", message: "Project is not a Git repository" })
@@ -95,15 +102,27 @@ const layer = Layer.effect(
         const gitDirectory = AbsolutePath.make(
           path.join(global.data, "snapshot", location.project.id, Hash.fast(worktree)),
         )
-        const snapshotRepository = (yield* fs.existsSafe(path.join(gitDirectory, "HEAD")))
-          ? new Git.Repository({ worktree, gitDirectory, commonDirectory: gitDirectory })
-          : yield* git.repo
+        const snapshotRepository = yield* flock.withLock(
+          Effect.gen(function* () {
+            if (yield* fs.existsSafe(path.join(gitDirectory, "HEAD")))
+              return new Git.Repository({ worktree, gitDirectory, commonDirectory: gitDirectory })
+            return yield* git.repo
               .create({ worktree, gitDirectory, seed: source })
               .pipe(Effect.mapError((cause) => failure("capture", cause)))
+          }),
+          gitDirectory,
+          undefined,
+          { timeoutMs: LOCK_TIMEOUT_MS },
+        )
         return { source, worktree, snapshotRepository }
       }).pipe(Effect.forkIn(lifetime)),
+      Duration.infinity,
+    )
+    const repository = repositoryFiber.pipe(
+      Effect.uninterruptible,
+      Effect.flatMap(Fiber.join),
+      Effect.tapError((cause) => (lockFailure(cause) ? invalidateRepository : Effect.void)),
     )
-    const repository = repositoryFiber.pipe(Effect.uninterruptible, Effect.flatMap(Fiber.join))
 
     const scope = Effect.fnUntraced(function* (worktree: AbsolutePath) {
       const relative = path.relative(worktree, location.directory)
@@ -113,21 +132,48 @@ const layer = Layer.effect(
     })
 
     const enabled = () => location.vcs?.type === "git" && state.get().enabled
+    let retryAt = Number.NEGATIVE_INFINITY
+    let retryMs = INITIAL_LOCK_RETRY_MS
 
     const capture = Effect.fn("Snapshot.capture")(function* () {
       if (!enabled()) return undefined
+      if ((yield* Effect.clockWith((clock) => clock.currentTimeMillis)) < retryAt) return undefined
       return yield* Effect.gen(function* () {
         const repo = yield* repository
-        return ID.make(
-          yield* git.tree.capture({
-            repository: repo.snapshotRepository,
-            scopes: [yield* scope(repo.worktree)],
-            ignores: repo.source,
-            maximumUntrackedFileBytes: 2 * 1024 * 1024,
+        const snapshot = yield* flock.withLock(
+          Effect.gen(function* () {
+            if ((yield* Effect.clockWith((clock) => clock.currentTimeMillis)) < retryAt) return undefined
+            return ID.make(
+              yield* git.tree.capture({
+                repository: repo.snapshotRepository,
+                scopes: [yield* scope(repo.worktree)],
+                ignores: repo.source,
+                maximumUntrackedFileBytes: 2 * 1024 * 1024,
+              }),
+            )
           }),
+          repo.snapshotRepository.gitDirectory,
+          undefined,
+          { timeoutMs: CAPTURE_LOCK_TIMEOUT_MS },
         )
+        if (snapshot !== undefined) {
+          retryAt = Number.NEGATIVE_INFINITY
+          retryMs = INITIAL_LOCK_RETRY_MS
+        }
+        return snapshot
       }).pipe(
-        Effect.catch((cause) => Effect.logWarning("failed to capture snapshot", { cause }).pipe(Effect.as(undefined))),
+        Effect.catch((cause) =>
+          Effect.gen(function* () {
+            const lock = lockFailure(cause)
+            const delay = lock ? retryMs : undefined
+            if (lock) {
+              retryAt = (yield* Effect.clockWith((clock) => clock.currentTimeMillis)) + retryMs
+              retryMs = Math.min(retryMs * 2, MAX_LOCK_RETRY_MS)
+            }
+            yield* Effect.logWarning("failed to capture snapshot", { cause, retryMs: delay })
+            return undefined
+          }),
+        ),
       )
     })
 
@@ -181,9 +227,12 @@ const layer = Layer.effect(
     const restore = Effect.fn("Snapshot.restore")(function* (input: RestoreInput) {
       if (!enabled()) return yield* new Error({ operation: "restore", message: "Snapshots are disabled" })
       const repo = yield* repository.pipe(Effect.mapError((cause) => failure("restore", cause)))
-      yield* git.tree
-        .restore({ repository: repo.snapshotRepository, files: yield* plan(repo.worktree, input) })
-        .pipe(Effect.mapError((cause) => failure("restore", cause)))
+      yield* flock.withLock(
+        git.tree.restore({ repository: repo.snapshotRepository, files: yield* plan(repo.worktree, input) }),
+        repo.snapshotRepository.gitDirectory,
+        undefined,
+        { timeoutMs: LOCK_TIMEOUT_MS },
+      ).pipe(Effect.mapError((cause) => failure("restore", cause)))
     })
 
     return Service.of({ transform: state.transform, reload: state.reload, capture, files, diff, restore })
@@ -193,7 +242,7 @@ const layer = Layer.effect(
 export const node = makeLocationNode({
   service: Service,
   layer,
-  deps: [FSUtil.node, Git.node, Global.node, Location.node],
+  deps: [FSUtil.node, Git.node, Global.node, Location.node, EffectFlock.node],
 })
 
 export const noopLayer = Layer.succeed(
@@ -216,3 +265,8 @@ function failure(operation: Error["operation"], cause: unknown) {
     cause,
   })
 }
+
+function lockFailure(cause: unknown) {
+  if (cause instanceof EffectFlock.LockTimeoutError || cause instanceof EffectFlock.LockCompromisedError) return true
+  return cause instanceof Git.OperationError && /index\.lock|another git process|unable to create.*lock/i.test(cause.message)
+}

packages/core/test/snapshot.test.ts

@@ -2,7 +2,8 @@ import { $ } from "bun"
 import { describe, expect } from "bun:test"
 import fs from "fs/promises"
 import path from "path"
-import { Deferred, Effect, Fiber, Layer } from "effect"
+import { Deferred, Effect, Fiber, Function, Layer } from "effect"
+import { TestClock } from "effect/testing"
 import { AppNodeBuilder } from "@opencode-ai/core/effect/app-node-builder"
 import { Git } from "@opencode-ai/core/git"
 import { Global } from "@opencode-ai/util/global"
@@ -10,6 +11,7 @@ import { Location } from "@opencode-ai/core/location"
 import { AbsolutePath, RelativePath } from "@opencode-ai/core/schema"
 import { Snapshot } from "@opencode-ai/core/snapshot"
 import { Hash } from "@opencode-ai/util/hash"
+import { EffectFlock } from "@opencode-ai/util/effect-flock"
 import { tmpdir } from "./fixture/tmpdir"
 import { testEffect } from "./lib/effect"
 
@@ -127,6 +129,98 @@ describe("Snapshot", () => {
     ),
   )
 
+  testEffect(Layer.empty).live("retries lazy repository initialization after lock timeout", () =>
+    Effect.acquireUseRelease(
+      Effect.promise(() => tmpdir()),
+      (tmp) =>
+        Effect.gen(function* () {
+          const project = path.join(tmp.path, "project")
+          yield* Effect.promise(async () => {
+            await fs.mkdir(project)
+            await fs.writeFile(path.join(project, "tracked.txt"), "one\n")
+            await initGit(project)
+          })
+          const flock = yield* EffectFlock.Service.pipe(Effect.provide(AppNodeBuilder.build(EffectFlock.node)))
+          let acquisitions = 0
+          const withLock: EffectFlock.Interface["withLock"] = Function.dual(
+            (args) => Effect.isEffect(args[0]),
+            <A, E, R>(body: Effect.Effect<A, E, R>, key: string, directory?: string, options?: EffectFlock.Options) => {
+              acquisitions++
+              if (acquisitions === 1) return Effect.fail(new EffectFlock.LockTimeoutError({ key }))
+              return flock.withLock(body, key, directory, options)
+            },
+          )
+          const instrumented = EffectFlock.Service.of({
+            ...flock,
+            withLock,
+          })
+          const layer = AppNodeBuilder.build(Snapshot.node, [
+            [
+              Location.node,
+              Location.boundNode(Location.Ref.make({ directory: AbsolutePath.make(project) })),
+            ],
+            [Global.node, Global.layerWith({ data: tmp.path, config: path.join(tmp.path, "config") })],
+            [EffectFlock.node, Layer.succeed(EffectFlock.Service, instrumented)],
+          ])
+
+          yield* Effect.gen(function* () {
+            const snapshot = yield* Snapshot.Service
+            expect(yield* snapshot.capture()).toBeUndefined()
+            expect(acquisitions).toBe(1)
+            yield* TestClock.adjust("5 seconds")
+            expect(yield* snapshot.capture()).toBeDefined()
+            expect(acquisitions).toBe(3)
+          }).pipe(Effect.provide(layer), Effect.provide(TestClock.layer()))
+        }),
+      (tmp) => Effect.promise(() => tmp[Symbol.asyncDispose]()),
+    ),
+  )
+
+  testEffect(Layer.empty).live("backs off a stale index lock and recovers without removing it", () =>
+    Effect.acquireUseRelease(
+      Effect.promise(() => tmpdir()),
+      (tmp) =>
+        Effect.gen(function* () {
+          const project = path.join(tmp.path, "project")
+          yield* Effect.promise(async () => {
+            await fs.mkdir(project)
+            await fs.writeFile(path.join(project, "tracked.txt"), "one\n")
+            await initGit(project)
+          })
+          const projectID = yield* Effect.gen(function* () {
+            return (yield* Location.Service).project.id
+          }).pipe(
+            Effect.provide(
+              AppNodeBuilder.build(Location.boundNode(Location.Ref.make({ directory: AbsolutePath.make(project) }))),
+            ),
+          )
+          const lock = path.join(
+            tmp.path,
+            "snapshot",
+            projectID,
+            Hash.fast(yield* Effect.promise(() => fs.realpath(project))),
+            "index.lock",
+          )
+
+          yield* Effect.gen(function* () {
+            const snapshot = yield* Snapshot.Service
+            expect(yield* snapshot.capture()).toBeDefined()
+            yield* Effect.promise(() => fs.writeFile(lock, ""))
+            yield* Effect.promise(() => fs.writeFile(path.join(project, "tracked.txt"), "two\n"))
+
+            expect(yield* snapshot.capture()).toBeUndefined()
+            expect(yield* Effect.promise(() => fs.stat(lock))).toBeDefined()
+            yield* Effect.promise(() => fs.rm(lock))
+            expect(yield* snapshot.capture()).toBeUndefined()
+
+            yield* TestClock.adjust("5 seconds")
+            expect(yield* snapshot.capture()).toBeDefined()
+          }).pipe(Effect.provide(snapshotLayer(tmp.path, project)), Effect.provide(TestClock.layer()))
+        }),
+      (tmp) => Effect.promise(() => tmp[Symbol.asyncDispose]()),
+    ),
+  )
+
   testEffect(Layer.empty).live("applies availability transforms", () =>
     Effect.acquireUseRelease(
       Effect.promise(() => tmpdir()),

packages/util/src/effect-flock.ts

@@ -82,8 +82,17 @@ export namespace EffectFlock {
   export interface Interface {
     readonly acquire: (key: string, dir?: string, options?: Options) => Effect.Effect<void, LockError, Scope.Scope>
     readonly withLock: {
-      (key: string, dir?: string): <A, E, R>(body: Effect.Effect<A, E, R>) => Effect.Effect<A, E | LockError, R>
-      <A, E, R>(body: Effect.Effect<A, E, R>, key: string, dir?: string): Effect.Effect<A, E | LockError, R>
+      (
+        key: string,
+        dir?: string,
+        options?: Options,
+      ): <A, E, R>(body: Effect.Effect<A, E, R>) => Effect.Effect<A, E | LockError, R>
+      <A, E, R>(
+        body: Effect.Effect<A, E, R>,
+        key: string,
+        dir?: string,
+        options?: Options,
+      ): Effect.Effect<A, E | LockError, R>
     }
   }
 
@@ -295,10 +304,15 @@ export namespace EffectFlock {
 
       const withLock: Interface["withLock"] = Function.dual(
         (args) => Effect.isEffect(args[0]),
-        <A, E, R>(body: Effect.Effect<A, E, R>, key: string, dir?: string): Effect.Effect<A, E | LockError, R> =>
+        <A, E, R>(
+          body: Effect.Effect<A, E, R>,
+          key: string,
+          dir?: string,
+          options?: Options,
+        ): Effect.Effect<A, E | LockError, R> =>
           Effect.scoped(
             Effect.gen(function* () {
-              yield* acquire(key, dir)
+              yield* acquire(key, dir, options)
               return yield* body
             }),
           ),