fix(collab): cancel shutdown storage waits

CHANGELOG.md

@@ -8,6 +8,7 @@ minor versions may ship breaking changes (see [SemVer §4](https://semver.org/#s
 - **Local watcher recovery releases resources.** Recreated directories are re-registered, and macOS watcher shutdown no longer retains file descriptors.
 - **Local watcher wakes scan only affected paths.** Materialization echoes no longer trigger extra full-tree scans, and transient macOS watcher `ENOENT` races recover without stopping the watcher. Full scans remain the safety path for startup, policy changes, uncertain batches, and the one-minute deadline. (#328)
 - **Malformed collab Redis settings no longer expose credentials.** BearDrive still refuses to start with an invalid setting.
+- **Shutdowns no longer hang on a busy local file remote.** A canceled file CAS lock wait returns without disturbing the other writer, while normal graceful shutdowns still save live co-editing changes.
 - **Share links no longer run out because your team synced.** The hub reported every byte a device pulled through `/store/*` to `QuotaProvider.RecordEgress`, the counter a managed hub enforces its monthly share-link transfer cap against, so ordinary syncing used up the allowance and every `/s/*` link in the org answered "This link has exceeded its transfer limit". `RecordEgress` now reports share-link transfers only, and sync traffic goes to `RecordSyncEgress` on a provider that implements the new optional `SyncEgressRecorder` — measured, never enforced. Existing providers keep compiling and simply aren't told.
 
 

architecture/webapp-server.md

@@ -40,6 +40,13 @@ classDiagram
         +headBlob(ctx, volume, path) (sha, ok)
         -pathLocks sync.Map (one mutex per project+path)
         -comments *commentCache (comments.go)
+        -collabMu sync.Mutex
+        -collabBG context.Context
+        -collabCancel context.CancelFunc
+        -collabClosing context.Context (only during Close)
+        +Close(ctx) error
+        -collabContext() process-lifetime context
+        -collabHookContext() closing caller or process context
     }
     class recordingMux {
         <<http.ServeMux wrapper>>
@@ -56,6 +63,7 @@ classDiagram
     note for Server "writeJSONCached is writeJSON for a response worth revalidating rather than re-sending: tree and heat carry an ETag over the encoded bytes plus Cache-Control: no-cache, and answer If-None-Match with a bodiless 304. The body is still built and hashed to answer one — the saving is the transfer, not the work, and the work was already being done."
     note for recordingMux "Exists for one caller: cmd/bdrive/desktop.go must classify every per-project route this hub serves, because a route it does not know falls through to local state and answers plausibly and WRONGLY. That has shipped twice. Recording beats parsing server.go: the per-project block builds eight patterns at runtime by concatenating a prefix, so a source scanner would silently miss exactly the routes it is meant to police."
     note for Server "upload/content is CONDITIONAL when the caller asks. A browser sends If-Match carrying the sha its buffer was read at (the ETag the read already returned), and a mismatch is a 409 carrying the sha that replaced it — because the alternative, taking the body wholesale, is what two browsers holding different documents did to each other every few seconds when one of them lost the co-editing relay: six characters typed by one editor gone, no conflict copy, no warning. The relay is gone and those two browsers now share one document, so that particular divergence cannot recur — but this door still guards the file against every writer that never touches a CRDT at all: an agent, the CLI, a device syncing. A stale base carrying the SAME content is not a conflict, it is a late arrival, and answering it 409 made the editor park copies of text nobody disputed. Optional by design: a caller that sends no If-Match gets the old behaviour. MCP write (given a sha) and edit (always) send one, so the agent door and the editor share this one precondition and this one lock. The check and the write are serialized per path (lockPath), or both halves of a simultaneous save pass a check that the other invalidates and the second still erases the first; a lock holds one process, so on a journal-backed source the base also rides into appendOpsIf (commitBase). The sha reported back is READ BACK, never the content sha just uploaded: a DirSource identifies a file by mtime and size, so handing the caller what it sent would hand it a value the next comparison could never match, and every save after the first on `bdrive serve --dir` would be parked as a conflict copy."
+    note for Server "Close publishes its caller context and cancels the process-lifetime collab context BEFORE snapshotting or taking leaseMu. Renewers, snapshot timers, writer waits and park retries capture the process context when created; OnLoad combines it with the request context. Explicit shutdown snapshots and lease release use the independent caller context, so a healthy tail is still written. Storage helpers carry the selected context through file reads, history/head, seed claims and upload CAS; an expired lease release leaves storage to expire by TTL."
     note for Server "ReportRead is the desktop's read seam, dead on a hub. The sidecar keeps no ReadLedger and answers the viewer routes locally, so a person reading in the Mac app reached no ledger at all while the same file in the web app counted. The hook hands each viewer read to the sidecar, which forwards it to the project's own hub as HUMAN traffic — routing it through the agent report route instead would have filed a person's browsing as a device's."
     note for Server "Desktop marks the loopback sidecar posture (`bdrive desktop`), NOT a hub: the server fronts this machine's own volume stores. It is set in exactly one production place — cmd/bdrive/desktop.go — so on a deployed hub it stays false and both branches below are dead code, leaving /api/config byte-identical to a hub without it. DesktopMe supplies `me` from the device's saved sign-in (settings.json), a func because the tray can change it at runtime; the desktop has no AuthProvider"
     note for Server "clientIP is a METHOD now, not a package func: X-Forwarded-For is honored when the PEER is loopback/private (the operator's own proxy) or TrustProxy is set, and then only its LAST hop. Every caller that gates on an IP — the auth rate limiter, /s/*, device rows, share telemetry — goes through it, so a client-supplied header cannot forge the identity a limiter counts"
@@ -249,6 +257,7 @@ classDiagram
         +PutIf(ctx, key, r, size, expect) Version
     }
     note for ConditionalPutter "internal/remote — optional, same mold again: gcsBackend implements it over object generations, and localBackend under an exclusive flock on the directory with the object's content hash as its version (local_cas.go — local filesystems only; it is what lets two bdrive serve processes, or two Servers in a test, share one directory). It exists because a journal append is a READ-MODIFY-WRITE and Put is last-writer-wins: two hub processes both read one journal, both append their own ops, and whichever writes second silently erases the other's with no error anywhere, because both writes succeeded. That is the lost update the &quot;each device writes only its own journal&quot; invariant prevents BETWEEN devices, reappearing inside ONE device id as soon as the hub runs twice. The precondition has to be evaluated by the STORE because that is the only party the two processes share — nothing either of them holds can serialise the other. Version is OPAQUE (a generation, an ETag, a fingerprint share nothing but equality) and VersionAbsent means &quot;create, and fail if somebody beat me&quot;. appendOps retries a bounded five times, and every attempt reads the journals again AFTER reading the version it is conditioned on, re-checks the write's commitBase and re-stamps seq and lamport — so a retry can neither reuse a seq the other process has just taken nor land on a version it was not built on. remote.Prefixed forwards it (as it forwards PutSigner): it used to drop it, so every hub project appended last-writer-wins even on GCS. upmu is kept — it stops N goroutines in one process taking the slow path, which is a different job. S3 deliberately does NOT implement it: conditional PutObject is unevenly supported"
+    note for ConditionalPutter "file:// CAS opens one descriptor, tries LOCK_EX|LOCK_NB immediately and creates a reusable timer only under contention. Cancellation closes an unacquired descriptor without unlocking the external holder. A successful acquisition is checked for cancellation and released once by its owner."
     note for Watcher "internal/remote — optional, in the PutSigner mold: only httpBackend implements it (GET /api/p/id/events). The channel carries NO payload — the daemon's only question is &quot;talk to the remote now?&quot;, and the answer to a burst is one cycle, so the impl coalesces on a buffer of one and can never make the hub's writer block. A closed channel means the path is gone (hub down, proxy timeout, an older hub with no route) and the caller falls back to its interval. Accelerator only: nothing may depend on a signal arriving, because for every object-store backend none ever will. Its stream gets a client WITHOUT httpBackend's 5-minute whole-request timeout (doWith), which would otherwise sever a healthy idle stream on the dot"
     note for Backend "internal/remote — impls: localBackend (file://), s3Backend, gcsBackend, httpBackend (https:// hub), Prefixed wrapper"
     note for Backend "Key handling is fallible now: Prefixed.key and localBackend.path RETURN AN ERROR (safeKey / store.UnderRoot) rather than concatenating, so a `..` key cannot walk out of a project's prefix or out of a file:// root — and Prefixed.List re-checks the STRIPPED key on the way out, since the prefix it removes is the only thing that was ever validated. The httpBackend client is origin-bound: the device token is keyed to settings.Server, SameOrigin is the one rule, refuseOffOriginRedirect is its CheckRedirect, a presign target must be https on a trusted origin (directTargetOK), and List drops keys failing journal.SafePath and clamps a negative Size. gcs SignPut now signs Content-Length too. Object carries Modified (S3 LastModified, GCS Updated, file mtime; zero where the backend has none) — RemoteSource.verify reads it to decide when a blob can no longer be rewritten by a presigned URL"

internal/remote/local_cas.go

@@ -10,6 +10,7 @@ import (
 	"os"
 	"path/filepath"
 	"syscall"
+	"time"
 )
 
 /* Compare-and-swap for a directory.
@@ -30,15 +31,22 @@ import (
 
 var _ ConditionalPutter = (*localBackend)(nil)
 
-// Under the temp-file prefix, so the scanner and List already skip it.
-const localLockName = ".bdrive-tmp-cas.lock"
+// Keep the retry interval private: it is an implementation detail, not a
+// configuration knob or a timing contract.
+const (
+	localLockName         = ".bdrive-tmp-cas.lock"
+	localCASRetryInterval = 5 * time.Millisecond
+)
 
-func (b *localBackend) GetVersioned(_ context.Context, key string) (io.ReadCloser, Version, error) {
+func (b *localBackend) GetVersioned(ctx context.Context, key string) (io.ReadCloser, Version, error) {
+	if err := ctx.Err(); err != nil {
+		return nil, VersionAbsent, err
+	}
 	p, err := b.path(key)
 	if err != nil {
 		return nil, VersionAbsent, err
 	}
-	unlock, err := b.lock()
+	unlock, err := b.lock(ctx)
 	if err != nil {
 		return nil, VersionAbsent, err
 	}
@@ -51,6 +59,9 @@ func (b *localBackend) GetVersioned(_ context.Context, key string) (io.ReadClose
 }
 
 func (b *localBackend) PutIf(ctx context.Context, key string, r io.Reader, _ int64, expect Version) (Version, error) {
+	if err := ctx.Err(); err != nil {
+		return VersionAbsent, err
+	}
 	p, err := b.path(key)
 	if err != nil {
 		return VersionAbsent, err
@@ -59,7 +70,7 @@ func (b *localBackend) PutIf(ctx context.Context, key string, r io.Reader, _ int
 	if err != nil {
 		return VersionAbsent, err
 	}
-	unlock, err := b.lock()
+	unlock, err := b.lock(ctx)
 	if err != nil {
 		return VersionAbsent, err
 	}
@@ -85,17 +96,51 @@ func versionOf(data []byte) Version {
 	return Version(hex.EncodeToString(sum[:]))
 }
 
-func (b *localBackend) lock() (func(), error) {
+func (b *localBackend) lock(ctx context.Context) (func(), error) {
+	if err := ctx.Err(); err != nil {
+		return nil, err
+	}
 	f, err := os.OpenFile(filepath.Join(b.root, localLockName), os.O_CREATE|os.O_RDWR, 0o644)
 	if err != nil {
 		return nil, err
 	}
-	if err := syscall.Flock(int(f.Fd()), syscall.LOCK_EX); err != nil {
-		f.Close()
+	unlock := func() {
+		_ = syscall.Flock(int(f.Fd()), syscall.LOCK_UN)
+		_ = f.Close()
+	}
+	try := func() error {
+		return syscall.Flock(int(f.Fd()), syscall.LOCK_EX|syscall.LOCK_NB)
+	}
+	if err := try(); err == nil {
+		if err := ctx.Err(); err != nil {
+			unlock()
+			return nil, err
+		}
+		return unlock, nil
+	} else if !errors.Is(err, syscall.EAGAIN) && !errors.Is(err, syscall.EWOULDBLOCK) {
+		_ = f.Close()
 		return nil, err
 	}
-	return func() {
-		_ = syscall.Flock(int(f.Fd()), syscall.LOCK_UN)
-		f.Close()
-	}, nil
+
+	timer := time.NewTimer(localCASRetryInterval)
+	defer timer.Stop()
+	for {
+		select {
+		case <-ctx.Done():
+			_ = f.Close()
+			return nil, ctx.Err()
+		case <-timer.C:
+			if err := try(); err == nil {
+				if err := ctx.Err(); err != nil {
+					unlock()
+					return nil, err
+				}
+				return unlock, nil
+			} else if !errors.Is(err, syscall.EAGAIN) && !errors.Is(err, syscall.EWOULDBLOCK) {
+				_ = f.Close()
+				return nil, err
+			}
+			timer.Reset(localCASRetryInterval)
+		}
+	}
 }

internal/remote/local_cas_test.go

@@ -4,11 +4,227 @@ import (
 	"context"
 	"errors"
 	"io"
+	"os"
+	"path/filepath"
 	"strings"
 	"sync"
+	"syscall"
 	"testing"
+	"time"
 )
 
+const localCASTestGuard = 3 * time.Second
+
+func holdLocalCASLock(t *testing.T, root string) (*os.File, func()) {
+	t.Helper()
+	f, err := os.OpenFile(filepath.Join(root, localLockName), os.O_CREATE|os.O_RDWR, 0o644)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if err := syscall.Flock(int(f.Fd()), syscall.LOCK_EX); err != nil {
+		_ = f.Close()
+		t.Fatal(err)
+	}
+	var once sync.Once
+	release := func() {
+		once.Do(func() {
+			_ = syscall.Flock(int(f.Fd()), syscall.LOCK_UN)
+			_ = f.Close()
+		})
+	}
+	t.Cleanup(release)
+	return f, release
+}
+
+func assertLocalCASLockBlocked(t *testing.T, f *os.File) {
+	t.Helper()
+	err := syscall.Flock(int(f.Fd()), syscall.LOCK_EX|syscall.LOCK_NB)
+	if !errors.Is(err, syscall.EAGAIN) && !errors.Is(err, syscall.EWOULDBLOCK) {
+		t.Fatalf("third descriptor acquired held CAS lock: %v", err)
+	}
+}
+
+func assertLocalCASLockFree(t *testing.T, f *os.File) {
+	t.Helper()
+	if err := syscall.Flock(int(f.Fd()), syscall.LOCK_EX|syscall.LOCK_NB); err != nil {
+		t.Fatalf("third descriptor could not acquire CAS lock after release: %v", err)
+	}
+	if err := syscall.Flock(int(f.Fd()), syscall.LOCK_UN); err != nil {
+		t.Fatal(err)
+	}
+}
+
+func waitLocalCASCall(t *testing.T, done <-chan error, release func()) error {
+	t.Helper()
+	select {
+	case err := <-done:
+		return err
+	case <-time.After(localCASTestGuard):
+		release()
+		select {
+		case <-done:
+		case <-time.After(localCASTestGuard):
+		}
+		t.Fatal("CAS operation did not observe context cancellation while the external lock was held")
+		return nil
+	}
+}
+
+func TestLocalCASLockCancellationPreservesHolderAndBytes(t *testing.T) {
+	for _, operation := range []string{"GetVersioned", "PutIf"} {
+		t.Run(operation, func(t *testing.T) {
+			root := t.TempDir()
+			be, err := newLocal(root)
+			if err != nil {
+				t.Fatal(err)
+			}
+			const key = "journal/dev.jsonl"
+			const original = "original\n"
+			if err := be.Put(context.Background(), key, strings.NewReader(original), int64(len(original))); err != nil {
+				t.Fatal(err)
+			}
+
+			var expect Version
+			if operation == "PutIf" {
+				rc, version, err := be.GetVersioned(context.Background(), key)
+				if err != nil {
+					t.Fatal(err)
+				}
+				if _, err := io.ReadAll(rc); err != nil {
+					t.Fatal(err)
+				}
+				if err := rc.Close(); err != nil {
+					t.Fatal(err)
+				}
+				expect = version
+			}
+
+			_, release := holdLocalCASLock(t, root)
+			probe, err := os.OpenFile(filepath.Join(root, localLockName), os.O_CREATE|os.O_RDWR, 0o644)
+			if err != nil {
+				t.Fatal(err)
+			}
+			defer probe.Close()
+
+			ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond)
+			defer cancel()
+			done := make(chan error, 1)
+			go func() {
+				if operation == "GetVersioned" {
+					rc, _, err := be.GetVersioned(ctx, key)
+					if rc != nil {
+						_ = rc.Close()
+					}
+					done <- err
+					return
+				}
+				_, err := be.PutIf(ctx, key, strings.NewReader("replacement\n"), int64(len("replacement\n")), expect)
+				done <- err
+			}()
+
+			err = waitLocalCASCall(t, done, release)
+			if ctx.Err() != context.DeadlineExceeded || !errors.Is(err, ctx.Err()) {
+				t.Fatalf("%s error = %v, context error = %v, want context deadline exceeded", operation, err, ctx.Err())
+			}
+			assertLocalCASLockBlocked(t, probe)
+			gotWhileHeld, err := os.ReadFile(filepath.Join(root, filepath.FromSlash(key)))
+			if err != nil {
+				t.Fatal(err)
+			}
+			if string(gotWhileHeld) != original {
+				t.Fatalf("cancelled %s changed bytes while the external lock was held: %q", operation, gotWhileHeld)
+			}
+
+			release()
+			assertLocalCASLockFree(t, probe)
+
+			rc, _, err := be.GetVersioned(context.Background(), key)
+			if err != nil {
+				t.Fatal(err)
+			}
+			got, err := io.ReadAll(rc)
+			if closeErr := rc.Close(); err != nil {
+				t.Fatal(err)
+			} else if closeErr != nil {
+				t.Fatal(closeErr)
+			}
+			if string(got) != original {
+				t.Fatalf("cancelled %s changed bytes to %q", operation, got)
+			}
+		})
+	}
+}
+
+func TestLocalCASAlreadyCanceledContextDoesNotWriteOrHoldLock(t *testing.T) {
+	for _, operation := range []string{"GetVersioned", "PutIf"} {
+		t.Run(operation, func(t *testing.T) {
+			root := t.TempDir()
+			be, err := newLocal(root)
+			if err != nil {
+				t.Fatal(err)
+			}
+			const key = "journal/dev.jsonl"
+			const original = "original\n"
+			if err := be.Put(context.Background(), key, strings.NewReader(original), int64(len(original))); err != nil {
+				t.Fatal(err)
+			}
+
+			var expect Version
+			if operation == "PutIf" {
+				rc, version, err := be.GetVersioned(context.Background(), key)
+				if err != nil {
+					t.Fatal(err)
+				}
+				if _, err := io.ReadAll(rc); err != nil {
+					t.Fatal(err)
+				}
+				if err := rc.Close(); err != nil {
+					t.Fatal(err)
+				}
+				expect = version
+			}
+
+			ctx, cancel := context.WithCancel(context.Background())
+			cancel()
+			if operation == "GetVersioned" {
+				rc, _, err := be.GetVersioned(ctx, key)
+				if rc != nil {
+					_ = rc.Close()
+				}
+				if !errors.Is(err, context.Canceled) {
+					t.Fatalf("%s error = %v, want context canceled", operation, err)
+				}
+			} else {
+				_, err := be.PutIf(ctx, key, strings.NewReader("replacement\n"), int64(len("replacement\n")), expect)
+				if !errors.Is(err, context.Canceled) {
+					t.Fatalf("%s error = %v, want context canceled", operation, err)
+				}
+			}
+
+			probe, err := os.OpenFile(filepath.Join(root, localLockName), os.O_CREATE|os.O_RDWR, 0o644)
+			if err != nil {
+				t.Fatal(err)
+			}
+			defer probe.Close()
+			assertLocalCASLockFree(t, probe)
+
+			rc, _, err := be.GetVersioned(context.Background(), key)
+			if err != nil {
+				t.Fatal(err)
+			}
+			got, err := io.ReadAll(rc)
+			if closeErr := rc.Close(); err != nil {
+				t.Fatal(err)
+			} else if closeErr != nil {
+				t.Fatal(closeErr)
+			}
+			if string(got) != original {
+				t.Fatalf("cancelled %s changed bytes to %q", operation, got)
+			}
+		})
+	}
+}
+
 func TestLocalPutIfIsACompareAndSwapAcrossProcesses(t *testing.T) {
 	root := t.TempDir()
 	a, err := newLocal(root)