diff --git a/CHANGELOG.md b/CHANGELOG.md index 7ca940a5..1c6761d0 100644 --- a/CHANGELOG.md +++ b/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. diff --git a/architecture/webapp-server.md b/architecture/webapp-server.md index ce327bef..90fe147d 100644 --- a/architecture/webapp-server.md +++ b/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 { <> @@ -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 "each device writes only its own journal" 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 "create, and fail if somebody beat me". 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 "talk to the remote now?", 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" diff --git a/internal/remote/local_cas.go b/internal/remote/local_cas.go index 01c28884..6001809f 100644 --- a/internal/remote/local_cas.go +++ b/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) + } + } } diff --git a/internal/remote/local_cas_test.go b/internal/remote/local_cas_test.go index 5c85a3b3..db9367b1 100644 --- a/internal/remote/local_cas_test.go +++ b/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) diff --git a/internal/webapp/data_test.go b/internal/webapp/data_test.go index bcb29700..ba901257 100644 --- a/internal/webapp/data_test.go +++ b/internal/webapp/data_test.go @@ -594,7 +594,7 @@ func TestDataQuerySQLReportsAPassThatMissedAFile(t *testing.T) { } wait("a pass that could not fetch a file still reports the index complete", func(a dataAnswer) bool { return !a.Complete }) - if err := os.WriteFile(filepath.Join(root, p.ID, "blobs", blob), []byte(body), 0o644); err != nil { + if err := srv.Root.Put(context.Background(), p.ID+"/blobs/"+blob, strings.NewReader(body), int64(len(body))); err != nil { t.Fatal(err) } wait("the index never caught up once the file could be fetched", diff --git a/internal/webapp/lease.go b/internal/webapp/lease.go index 66209f39..bd8d94d7 100644 --- a/internal/webapp/lease.go +++ b/internal/webapp/lease.go @@ -62,6 +62,9 @@ func (s *Server) nodeID() string { // holdLease reports whether this process writes the room, taking the lease // if it is free or expired and renewing it once half its life has passed. func (s *Server) holdLease(ctx context.Context, room string) bool { + if ctx.Err() != nil { + return false + } project, path, ok := strings.Cut(room, "/") if !ok { return false @@ -78,6 +81,9 @@ func (s *Server) holdLease(ctx context.Context, room string) bool { // ponytail: one lock across every room's lease round-trip; per-room locks if hundreds of live rooms contend r.leaseMu.Lock() defer r.leaseMu.Unlock() + if ctx.Err() != nil { + return false + } now := time.Now() held := r.leases[room] if held != nil && now.Before(held.until.Add(-leaseTTL/2)) { @@ -105,7 +111,9 @@ func (s *Server) holdLease(ctx context.Context, room string) bool { body, _ := json.Marshal(leaseRecord{Holder: s.nodeID(), Until: now.Add(leaseTTL)}) v, err := cond.PutIf(ctx, leaseKey(path), bytes.NewReader(body), int64(len(body)), expect) if err != nil { - delete(r.leases, room) + if ctx.Err() == nil { + delete(r.leases, room) + } if !errors.Is(err, remote.ErrVersionMismatch) { log.Printf("bdrive: taking lease for %s: %v", room, err) } @@ -115,29 +123,34 @@ func (s *Server) holdLease(ctx context.Context, room string) bool { return true } -// releaseLease gives a room's lease up as the room ends, so the next process -// to open the file does not wait out the TTL. Best effort: an expired lease -// is the same thing, later. -func (s *Server) releaseLease(room string) { +// releaseLease gives a room's lease up as the room ends. If its context is +// already cancelled, it leaves the in-memory record for a later attempt or +// TTL expiry rather than hiding a lease that is still stored remotely. +func (s *Server) releaseLease(ctx context.Context, room string) { project, path, _ := strings.Cut(room, "/") r := s.rooms() r.leaseMu.Lock() defer r.leaseMu.Unlock() held := r.leases[room] - if held == nil { + if held == nil || ctx.Err() != nil { return } - delete(r.leases, room) be, ok := s.projectBackend(project) if !ok { + delete(r.leases, room) return } cond, ok := be.(remote.ConditionalPutter) if !ok { + delete(r.leases, room) return } body, _ := json.Marshal(leaseRecord{Holder: s.nodeID()}) - _, _ = cond.PutIf(context.Background(), leaseKey(path), bytes.NewReader(body), int64(len(body)), held.version) + _, err := cond.PutIf(ctx, leaseKey(path), bytes.NewReader(body), int64(len(body)), held.version) + if err != nil && ctx.Err() != nil { + return + } + delete(r.leases, room) } // keepLease renews a room's lease for as long as the room lives — a quiet @@ -145,6 +158,10 @@ func (s *Server) releaseLease(room string) { // process that does not hold it but could write, keeps asking, which is how // the writer's job passes on when the holder's room ends. func (s *Server) keepLease(room string) { + bg := s.collabContext() + if bg.Err() != nil { + return + } r := s.rooms() r.mu.Lock() if r.renewals[room] != nil { @@ -166,12 +183,14 @@ func (s *Server) keepLease(room string) { defer t.Stop() for { select { + case <-bg.Done(): + return case <-stop: return case <-t.C: _, who, ok := r.get(room) if ok && (who.Email != "" || s.holdsLease(room)) { - s.holdLease(context.Background(), room) + s.holdLease(bg, room) } } } diff --git a/internal/webapp/server.go b/internal/webapp/server.go index 3dd26953..024a8fff 100644 --- a/internal/webapp/server.go +++ b/internal/webapp/server.go @@ -116,6 +116,12 @@ type Server struct { // collabRelay carries document updates between processes; nil means // one process holds each document (lease.go). collabRelay cluster.Relay + // collabMu protects the process-lifetime collab context and the caller + // context exposed only while Close is tearing rooms down. + collabMu sync.Mutex + collabBG context.Context + collabCancel context.CancelFunc + collabClosing context.Context // This process's name on a room lease (lease.go). nodeOnce sync.Once node string diff --git a/internal/webapp/upload.go b/internal/webapp/upload.go index 087ca304..724e0d4e 100644 --- a/internal/webapp/upload.go +++ b/internal/webapp/upload.go @@ -784,7 +784,7 @@ func (s *Server) handleUploadContent(v *volume, w http.ResponseWriter, r *http.R if !ok { head = blob } - s.adoptRoomWrite(projectID(r), p, head, tmp) + s.adoptRoomWrite(r.Context(), projectID(r), p, head, tmp) writeJSON(w, map[string]any{"ok": true, "path": p, "sha": head}) } diff --git a/internal/webapp/upload_shrink_test.go b/internal/webapp/upload_shrink_test.go index d7d3c1da..34e310cb 100644 --- a/internal/webapp/upload_shrink_test.go +++ b/internal/webapp/upload_shrink_test.go @@ -115,7 +115,7 @@ func TestFinalSnapshotParksARefusedShrink(t *testing.T) { t.Fatalf("seed write: %d", rec.Code) } doc := crdt.New() - if err := srv.seedDoc(room, doc); err != nil { + if err := srv.seedDoc(context.Background(), room, doc); err != nil { t.Fatal(err) } srv.rooms().load(room, doc) @@ -193,7 +193,7 @@ func TestRefusedShrinkCopyIsRetriedAfterTheRoomCloses(t *testing.T) { full := strings.Repeat("a standup line\n", 100) putContent(t, h, "/api/p/"+p.ID+"/upload/content?path=notes.md", full, "") doc := crdt.New() - if err := srv.seedDoc(room, doc); err != nil { + if err := srv.seedDoc(context.Background(), room, doc); err != nil { t.Fatal(err) } srv.rooms().load(room, doc) diff --git a/internal/webapp/ycollab.go b/internal/webapp/ycollab.go index 08a6a652..82b28fc6 100644 --- a/internal/webapp/ycollab.go +++ b/internal/webapp/ycollab.go @@ -51,6 +51,29 @@ import ( way a cgo sqlite would, which is the constraint that made this a relay in the first place and the one that changed. */ +// collabContext is captured when background work is created, so Close can +// cancel work already waiting in storage without rebinding its context. +func (s *Server) collabContext() context.Context { + s.collabMu.Lock() + defer s.collabMu.Unlock() + if s.collabBG == nil { + s.collabBG, s.collabCancel = context.WithCancel(context.Background()) + } + return s.collabBG +} + +func (s *Server) collabHookContext() context.Context { + s.collabMu.Lock() + defer s.collabMu.Unlock() + if s.collabClosing != nil { + return s.collabClosing + } + if s.collabBG == nil { + s.collabBG, s.collabCancel = context.WithCancel(context.Background()) + } + return s.collabBG +} + // ydocs is the document server, built once. Rooms are created on demand and // swept when idle (ygo's own idle sweep), which is the shape the memory // measurement argued for: a held document is ~100 KB while actively edited — @@ -64,32 +87,55 @@ func (s *Server) ydocs() *ygows.Server { // kept addressable while the room lives, and written back when the // last editor leaves. srv.OnLoadDocument = func(ctx context.Context, room string, doc *crdt.Doc) error { + bg := s.collabContext() + ctx, cancel := context.WithCancel(ctx) + stop := context.AfterFunc(bg, cancel) + defer func() { + stop() + cancel() + }() + if bg.Err() != nil { + cancel() + } + if err := ctx.Err(); err != nil { + return err + } s.rooms().load(room, doc) - // Without a relay the processes cannot share a document, so - // only the lease holder opens the room; ygo answers the refused - // joiner with a 500 and y-websocket's retry carries it to - // another process (lease.go). With one, the lease is taken by - // the snapshot path instead — a room the relay created on a - // process nobody is editing from cannot write, so it must not - // hold the pen either. + // Without a relay, only the lease holder can open the room. if s.collabRelay == nil && !s.holdLease(ctx, room) { s.rooms().drop(room) + if err := ctx.Err(); err != nil { + return err + } return errRoomElsewhere } - if err := s.seedDoc(room, doc); err != nil { - s.dropRoom(room) + if err := s.seedDoc(ctx, room, doc); err != nil { + s.rooms().drop(room) + s.releaseLease(s.collabHookContext(), room) + return err + } + if err := ctx.Err(); err != nil { + s.rooms().drop(room) + s.releaseLease(s.collabHookContext(), room) return err } - // AFTER the seed, so the bytes the document was built from are - // never mistaken for an edit that needs writing back. s.watchRoom(room, doc) s.keepLease(room) return nil } - srv.OnLastPeer = func(ctx context.Context, room string) { s.finalSnapshot(ctx, room) } - srv.OnUnloadDocument = func(ctx context.Context, room string) { + srv.OnLastPeer = func(_ context.Context, room string) { + s.finalSnapshot(s.collabHookContext(), room) + } + srv.OnUnloadDocument = func(_ context.Context, room string) { + ctx := s.collabHookContext() s.finalSnapshot(ctx, room) - s.dropRoom(room) + // Close owns the final caller-context snapshot after it cancels + // an unload already in flight. Keep its document addressable. + if ctx.Err() != nil { + return + } + s.rooms().drop(room) + s.releaseLease(ctx, room) } /* Authorization runs per CONNECTION, after proj() has already decided this caller may see this project. @@ -171,8 +217,11 @@ func (s *Server) handleYCollab(v *volume, w http.ResponseWriter, r *http.Request // // The hub can just do it. This runs once, when the room is created and before // any client is attached, so there is nothing to claim and nothing to race. -func (s *Server) seedDoc(room string, doc *crdt.Doc) error { - body, sha, ok := s.roomBytes(room) +func (s *Server) seedDoc(ctx context.Context, room string, doc *crdt.Doc) error { + if err := ctx.Err(); err != nil { + return err + } + body, sha, ok := s.roomBytes(ctx, room) if !ok || len(body) == 0 { return nil // a file that does not exist yet starts empty } @@ -213,11 +262,11 @@ func (s *Server) seedDoc(room string, doc *crdt.Doc) error { nothing. Only a hub that never saw this file being edited starts from the bytes. */ var newest []byte - if head := s.roomHead(project, path); head != "" && head != sha { - newest = s.roomHistory(project, head) + if head := s.roomHead(ctx, project, path); head != "" && head != sha { + newest = s.roomHistory(ctx, project, head) } if newest == nil { - if state := s.roomHistory(project, sha); state != nil && replayInto(doc, state, text) { + if state := s.roomHistory(ctx, project, sha); state != nil && replayInto(doc, state, text) { return nil } } @@ -241,7 +290,7 @@ func (s *Server) seedDoc(room string, doc *crdt.Doc) error { ttxt.Insert(txn, 0, text, nil) }) } - state := s.claimRoomHistory(project, path, sha, tmp.EncodeStateAsUpdate(), text, newest) + state := s.claimRoomHistory(ctx, project, path, sha, tmp.EncodeStateAsUpdate(), text, newest) if !replayInto(doc, state, text) { return fmt.Errorf("%s: the stored document does not spell the file", room) } @@ -253,17 +302,16 @@ func (s *Server) seedDoc(room string, doc *crdt.Doc) error { // and knows everything the newest state does, ours otherwise — written with // compare-and-swap, so two processes seeding at once end up with one of // their seeds rather than one each. -func (s *Server) claimRoomHistory(project, path, sha string, state []byte, body string, newest []byte) []byte { +func (s *Server) claimRoomHistory(ctx context.Context, project, path, sha string, state []byte, body string, newest []byte) []byte { be, ok := s.projectBackend(project) if !ok || sha == "" || len(state) > maxHistoryBytes { return state } cond, ok := be.(remote.ConditionalPutter) if !ok { - s.storeRoomHistory(project, path, sha, state) + s.storeRoomHistory(ctx, project, path, sha, state) return state } - ctx := context.Background() key := historyKey(sha) var err error for range 3 { @@ -279,7 +327,7 @@ func (s *Server) claimRoomHistory(project, path, sha string, state []byte, body } } if _, err = cond.PutIf(ctx, key, bytes.NewReader(state), int64(len(state)), v); err == nil { - s.storeRoomHead(project, path, sha) + s.storeRoomHead(ctx, project, path, sha) return state } if !errors.Is(err, remote.ErrVersionMismatch) { @@ -379,7 +427,7 @@ func (s *Server) foldIntoRoom(ctx context.Context, project, path, sha string) bo if _, _, ok := s.rooms().get(room); !ok { return false } - body, cur, ok := s.roomBytes(room) + body, cur, ok := s.roomBytes(ctx, room) if !ok || cur != sha { return false } @@ -402,7 +450,7 @@ func (s *Server) foldIntoRoom(ctx context.Context, project, path, sha string) bo } s.rooms().rebase(room, sha) if doc, _, ok := s.rooms().get(room); ok { - s.storeRoomHistory(project, path, sha, doc.EncodeStateAsUpdate()) + s.storeRoomHistory(ctx, project, path, sha, doc.EncodeStateAsUpdate()) } return true } @@ -413,7 +461,7 @@ func (s *Server) foldIntoRoom(ctx context.Context, project, path, sha string) bo // snapshot must build on this version rather than be refused for a stale // base and parked as a conflict copy. A write that differs from the // document is an outsider's and leaves the room alone. -func (s *Server) adoptRoomWrite(project, path, sha string, body io.ReadSeeker) { +func (s *Server) adoptRoomWrite(ctx context.Context, project, path, sha string, body io.ReadSeeker) { room := project + "/" + path doc, _, ok := s.rooms().get(room) if !ok || sha == "" { @@ -431,7 +479,7 @@ func (s *Server) adoptRoomWrite(project, path, sha string, body io.ReadSeeker) { return } s.rooms().rebase(room, sha) - s.storeRoomHistory(project, path, sha, state) + s.storeRoomHistory(ctx, project, path, sha, state) } // A room's history lives in the project's own storage as ydoc/: the @@ -441,12 +489,12 @@ func (s *Server) adoptRoomWrite(project, path, sha string, body io.ReadSeeker) { // room still works, it just cannot be rebuilt losslessly. func historyKey(sha string) string { return "ydoc/" + sha } -func (s *Server) roomHistory(project, sha string) []byte { +func (s *Server) roomHistory(ctx context.Context, project, sha string) []byte { be, ok := s.projectBackend(project) if !ok || sha == "" { return nil } - rc, err := be.Get(context.Background(), historyKey(sha)) + rc, err := be.Get(ctx, historyKey(sha)) if err != nil { if !remote.IsNotExist(err) { log.Printf("bdrive: reading room history %s/%s: %v", project, sha, err) @@ -461,28 +509,31 @@ func (s *Server) roomHistory(project, sha string) []byte { return state } -func (s *Server) storeRoomHistory(project, path, sha string, state []byte) { +func (s *Server) storeRoomHistory(ctx context.Context, project, path, sha string, state []byte) { + if ctx.Err() != nil { + return + } be, ok := s.projectBackend(project) if !ok || sha == "" || len(state) > maxHistoryBytes { return } - if err := be.Put(context.Background(), historyKey(sha), bytes.NewReader(state), int64(len(state))); err != nil { + if err := be.Put(ctx, historyKey(sha), bytes.NewReader(state), int64(len(state))); err != nil { log.Printf("bdrive: storing room history %s/%s: %v", project, sha, err) return } - s.storeRoomHead(project, path, sha) + s.storeRoomHead(ctx, project, path, sha) } // ydoc/head/ names the newest state stored for a path — what the next // room for it builds from, on any process (seedDoc). func headKey(path string) string { return "ydoc/head/" + path } -func (s *Server) roomHead(project, path string) string { +func (s *Server) roomHead(ctx context.Context, project, path string) string { be, ok := s.projectBackend(project) if !ok { return "" } - rc, err := be.Get(context.Background(), headKey(path)) + rc, err := be.Get(ctx, headKey(path)) if err != nil { return "" } @@ -491,12 +542,15 @@ func (s *Server) roomHead(project, path string) string { return strings.TrimSpace(string(sha)) } -func (s *Server) storeRoomHead(project, path, sha string) { +func (s *Server) storeRoomHead(ctx context.Context, project, path, sha string) { + if ctx.Err() != nil { + return + } be, ok := s.projectBackend(project) if !ok { return } - if err := be.Put(context.Background(), headKey(path), strings.NewReader(sha), int64(len(sha))); err != nil { + if err := be.Put(ctx, headKey(path), strings.NewReader(sha), int64(len(sha))); err != nil { log.Printf("bdrive: storing room head %s/%s: %v", project, path, err) } } @@ -515,7 +569,7 @@ func (s *Server) projectBackend(project string) (remote.Backend, bool) { // roomBytes is the file behind a room, bounded, with the blob it currently is // — the sha every write to it must name in If-Match. -func (s *Server) roomBytes(room string) ([]byte, string, bool) { +func (s *Server) roomBytes(ctx context.Context, room string) ([]byte, string, bool) { project, path, ok := strings.Cut(room, "/") if !ok { return nil, "", false @@ -524,7 +578,6 @@ func (s *Server) roomBytes(room string) ([]byte, string, bool) { if err != nil { return nil, "", false } - ctx := context.Background() // A room is built from what the file IS, not from a snapshot a background // rebuild has not caught up on yet. snap, err := v.current(ctx) @@ -568,6 +621,9 @@ func (s *Server) snapshotRoom(ctx context.Context, room string) { s.snapshot(ctx func (s *Server) finalSnapshot(ctx context.Context, room string) { s.snapshot(ctx, room, true) } func (s *Server) snapshot(ctx context.Context, room string, final bool) { + if ctx.Err() != nil { + return + } project, path, ok := strings.Cut(room, "/") if !ok { return @@ -585,10 +641,13 @@ func (s *Server) snapshot(ctx context.Context, room string, final bool) { if !ok { return } - cur, curSha, have := s.roomBytes(room) + cur, curSha, have := s.roomBytes(ctx, room) + if ctx.Err() != nil { + return + } if have && string(cur) == text { if s.rooms().base(room) != curSha { - s.storeRoomHistory(project, path, curSha, state) + s.storeRoomHistory(ctx, project, path, curSha, state) s.rooms().rebase(room, curSha) // Written by another process. Its change frame reached only // its own browsers; ours still show the edit as unsaved until @@ -601,14 +660,17 @@ func (s *Server) snapshot(ctx context.Context, room string, final bool) { if !s.holdLease(ctx, room) { // Another process writes this file. Look again once it has had // time to, so this process's browsers see the save land. - s.rooms().awaitWriter(room, func() { s.snapshotRoom(context.Background(), room) }) + bg := s.collabContext() + if ctx.Err() == nil && bg.Err() == nil { + s.rooms().awaitWriter(room, func() { s.snapshotRoom(bg, room) }) + } return } if s.apiMux == nil { return } base := s.rooms().base(room) - if have && curSha != base && s.contains(doc, project, curSha) { + if have && curSha != base && s.contains(ctx, doc, project, curSha) { // The head was written by this room on another process — the // previous lease holder — and the relay has already delivered every // item it wrote, so build on it instead of parking beside it. @@ -619,7 +681,7 @@ func (s *Server) snapshot(ctx context.Context, room string, final bool) { switch w.code { case http.StatusOK: if sha := reportedSha(w); sha != "" { - s.storeRoomHistory(project, path, sha, state) + s.storeRoomHistory(ctx, project, path, sha, state) s.rooms().rebase(room, sha) } case http.StatusConflict: @@ -658,7 +720,7 @@ func (s *Server) snapshot(ctx context.Context, room string, final bool) { // must not start another. s.rooms().park(room, text) copy := conflictPath(path, who, time.Now()) - s.parkCopy(who, project, path, copy, text) + s.parkCopy(ctx, who, project, path, copy, text) return } } @@ -675,9 +737,9 @@ var ( // budget: once the room is evicted this text exists nowhere else, so giving // up would be the loss the copy exists to prevent. Held in memory only, so a // hub restart before the store recovers still loses it. -func (s *Server) parkCopy(who User, project, path, copy, text string) { - try := func() bool { - cw := s.putAs(context.Background(), who, project, copy, text, "") +func (s *Server) parkCopy(ctx context.Context, who User, project, path, copy, text string) { + try := func(ctx context.Context) bool { + cw := s.putAs(ctx, who, project, copy, text, "") if cw.code == http.StatusOK { log.Printf("bdrive: %s: the room's text would empty or sharply shrink it; parked beside it as %s", path, copy) return true @@ -685,17 +747,29 @@ func (s *Server) parkCopy(who User, project, path, copy, text string) { log.Printf("bdrive: %s: could not park the room's refused text (%d bytes) as %s (%d); retrying", path, len(text), copy, cw.code) return false } - if try() { + backoff, backoffMax := parkBackoff, parkBackoffMax + if try(ctx) { + return + } + bg := s.collabContext() + if bg.Err() != nil { return } go func() { - wait := parkBackoff + wait := backoff + timer := time.NewTimer(wait) + defer timer.Stop() for { - time.Sleep(wait) - if try() { + select { + case <-bg.Done(): return + case <-timer.C: + if bg.Err() != nil || try(bg) { + return + } + wait = min(wait*2, backoffMax) + timer.Reset(wait) } - wait = min(wait*2, parkBackoffMax) } }() } @@ -863,7 +937,12 @@ var ( ) func (s *Server) watchRoom(room string, doc *crdt.Doc) { - unsub := doc.OnUpdate(func(_ []byte, _ any) { s.rooms().touched(room, func() { s.snapshotRoom(context.Background(), room) }) }) + bg := s.collabContext() + unsub := doc.OnUpdate(func(_ []byte, _ any) { + if bg.Err() == nil { + s.rooms().touched(room, func() { s.snapshotRoom(bg, room) }) + } + }) r := s.rooms() r.mu.Lock() if old := r.unsubs[room]; old != nil { @@ -1004,7 +1083,7 @@ var errRoomElsewhere = errors.New("another hub process holds this document") // dropRoom forgets a room and gives its lease back. func (s *Server) dropRoom(room string) { s.rooms().drop(room) - s.releaseLease(room) + s.releaseLease(s.collabHookContext(), room) } // contains reports whether the room's document already holds everything the @@ -1012,8 +1091,8 @@ func (s *Server) dropRoom(room string) { // document, not an outsider's write". Compared by item identity, which is the // only comparison a CRDT offers: two texts that read the same are not the // same document (seedDoc). -func (s *Server) contains(doc *crdt.Doc, project, sha string) bool { - state := s.roomHistory(project, sha) +func (s *Server) contains(ctx context.Context, doc *crdt.Doc, project, sha string) bool { + state := s.roomHistory(ctx, project, sha) if state == nil { return false } @@ -1031,9 +1110,16 @@ var emptyUpdate = crdt.EncodeStateAsUpdateV1(crdt.New(), nil) // typing that exists only in browsers. Bounded by ctx; a Close past its // deadline costs the tail, never a hang. func (s *Server) Close(ctx context.Context) error { - if s.roomReg == nil { - return nil - } + _ = s.collabContext() + s.collabMu.Lock() + s.collabClosing = ctx + s.collabCancel() + s.collabMu.Unlock() + defer func() { + s.collabMu.Lock() + s.collabClosing = nil + s.collabMu.Unlock() + }() r := s.rooms() r.mu.Lock() rooms := make([]string, 0, len(r.docs)) @@ -1049,13 +1135,23 @@ func (s *Server) Close(ctx context.Context) error { err = s.y.Shutdown(ctx) } // Shutdown closes peers, and a room with none — one the relay or Apply - // created — is never unloaded by it, so its renewer and timers would - // outlive the server. The process is going away: every room ends here. + // created — is never unloaded by it. Include rooms that finished loading + // after the first snapshot, so their leases and workers cannot outlive us. + r.mu.Lock() + rooms = rooms[:0] + for room := range r.docs { + rooms = append(rooms, room) + } + r.mu.Unlock() for _, room := range rooms { - s.dropRoom(room) + r.drop(room) + s.releaseLease(ctx, room) } if s.collabRelay != nil { s.collabRelay.Close() } + if err == nil { + err = ctx.Err() + } return err } diff --git a/internal/webapp/ycollab_cluster_test.go b/internal/webapp/ycollab_cluster_test.go index 2ee8da1d..1753c6d6 100644 --- a/internal/webapp/ycollab_cluster_test.go +++ b/internal/webapp/ycollab_cluster_test.go @@ -6,10 +6,12 @@ import ( "errors" "io" "net/http" + "os" "path/filepath" "strconv" "strings" "sync" + "syscall" "testing" "time" @@ -109,8 +111,512 @@ func openVia(s *Server, room string) error { return s.ydocs().OnLoadDocument(context.Background(), room, crdt.New()) } +// This observer delegates every operation to the real file backend. Its +// channels only expose entry/return ordering for a selected CAS key. +type shutdownCASCall struct { + key string + returned <-chan error +} + +type shutdownCASBackend struct { + remote.Backend + cond remote.ConditionalPutter + mu sync.Mutex + watch string + entered chan shutdownCASCall +} + +func (b *shutdownCASBackend) observe(key string) func(error) { + b.mu.Lock() + match := b.watch != "" && strings.Contains(key, b.watch) + b.mu.Unlock() + if !match { + return func(error) {} + } + returned := make(chan error, 1) + select { + case b.entered <- shutdownCASCall{key: key, returned: returned}: + default: + } + return func(err error) { returned <- err } +} + +func (b *shutdownCASBackend) GetVersioned(ctx context.Context, key string) (io.ReadCloser, remote.Version, error) { + finish := b.observe(key) + rc, version, err := b.cond.GetVersioned(ctx, key) + finish(err) + return rc, version, err +} + +func (b *shutdownCASBackend) PutIf(ctx context.Context, key string, r io.Reader, size int64, expect remote.Version) (remote.Version, error) { + finish := b.observe(key) + version, err := b.cond.PutIf(ctx, key, r, size, expect) + finish(err) + return version, err +} + +func (b *shutdownCASBackend) watchKey(key string) { + b.mu.Lock() + b.watch = key + b.mu.Unlock() +} + +func newShutdownHub(t *testing.T) (*Server, Project, string, *shutdownCASBackend) { + t.Helper() + var observed *shutdownCASBackend + srv, p, root := newHub(t, true, func(be remote.Backend) remote.Backend { + cond, ok := be.(remote.ConditionalPutter) + if !ok { + t.Fatal("file backend does not support compare-and-swap") + } + observed = &shutdownCASBackend{Backend: be, cond: cond, entered: make(chan shutdownCASCall, 16)} + return observed + }) + return srv, p, root, observed +} + +func holdCollabShutdownLock(t *testing.T, root string) (release func(), assertBlocked func(), assertFree func()) { + t.Helper() + path := filepath.Join(root, ".bdrive-tmp-cas.lock") + holder, err := os.OpenFile(path, os.O_CREATE|os.O_RDWR, 0o644) + if err != nil { + t.Fatal(err) + } + if err := syscall.Flock(int(holder.Fd()), syscall.LOCK_EX); err != nil { + _ = holder.Close() + t.Fatal(err) + } + probe, err := os.OpenFile(path, os.O_RDWR, 0) + if err != nil { + _ = holder.Close() + t.Fatal(err) + } + var once sync.Once + release = func() { + once.Do(func() { + _ = syscall.Flock(int(holder.Fd()), syscall.LOCK_UN) + _ = holder.Close() + }) + } + t.Cleanup(func() { + release() + _ = probe.Close() + }) + assertBlocked = func() { + t.Helper() + err := syscall.Flock(int(probe.Fd()), syscall.LOCK_EX|syscall.LOCK_NB) + if !errors.Is(err, syscall.EAGAIN) && !errors.Is(err, syscall.EWOULDBLOCK) { + if err == nil { + _ = syscall.Flock(int(probe.Fd()), syscall.LOCK_UN) + } + t.Fatalf("external holder lost its lock: %v", err) + } + } + assertFree = func() { + t.Helper() + if err := syscall.Flock(int(probe.Fd()), syscall.LOCK_EX|syscall.LOCK_NB); err != nil { + t.Fatalf("CAS lock remains held after external holder release: %v", err) + } + _ = syscall.Flock(int(probe.Fd()), syscall.LOCK_UN) + } + return release, assertBlocked, assertFree +} + +func waitShutdownCall(t *testing.T, call func() error, release func()) error { + t.Helper() + done := make(chan error, 1) + go func() { done <- call() }() + select { + case err := <-done: + return err + case <-time.After(3 * time.Second): + release() + select { + case <-done: + case <-time.After(3 * time.Second): + } + t.Fatal("operation did not return while the external CAS lock was held") + return nil + } +} + +func awaitShutdownCAS(t *testing.T, b *shutdownCASBackend) shutdownCASCall { + t.Helper() + select { + case call := <-b.entered: + return call + case <-time.After(3 * time.Second): + t.Fatal("operation never reached the real file CAS") + return shutdownCASCall{} + } +} + +func awaitShutdownCASReturn(t *testing.T, call shutdownCASCall) { + t.Helper() + select { + case err := <-call.returned: + if !errors.Is(err, context.Canceled) && !errors.Is(err, context.DeadlineExceeded) { + t.Fatalf("CAS %s returned %v, want context cancellation", call.key, err) + } + case <-time.After(3 * time.Second): + t.Fatalf("CAS %s outlived shutdown", call.key) + } +} + +// The underlying bytes and CAS remain real; the selected I/O waits just as +// a context-aware network backend would, until its caller cancels it. +type shutdownIOBackend struct { + *shutdownCASBackend + mu sync.Mutex + method string + prefix string + entered chan struct{} + abort chan struct{} + once sync.Once +} + +func (b *shutdownIOBackend) wait(ctx context.Context, method, key string) error { + b.mu.Lock() + match := b.method == method && strings.HasPrefix(key, b.prefix) + b.mu.Unlock() + if !match { + return nil + } + select { + case b.entered <- struct{}{}: + default: + } + select { + case <-ctx.Done(): + return ctx.Err() + case <-b.abort: + return errors.New("test storage guard released") + } +} + +func (b *shutdownIOBackend) Get(ctx context.Context, key string) (io.ReadCloser, error) { + if err := b.wait(ctx, "GET", key); err != nil { + return nil, err + } + return b.Backend.Get(ctx, key) +} + +func (b *shutdownIOBackend) Put(ctx context.Context, key string, r io.Reader, size int64) error { + if err := b.wait(ctx, "PUT", key); err != nil { + return err + } + return b.Backend.Put(ctx, key, r, size) +} + +func TestCollabClosePropagatesContextToRoomStorage(t *testing.T) { + for _, step := range []string{"room_bytes", "room_history", "room_head"} { + t.Run(step, func(t *testing.T) { + var blocked *shutdownIOBackend + srv, p, _ := newHub(t, true, func(be remote.Backend) remote.Backend { + cond, ok := be.(remote.ConditionalPutter) + if !ok { + t.Fatal("file backend does not support CAS") + } + observed := &shutdownCASBackend{Backend: be, cond: cond, entered: make(chan shutdownCASCall, 16)} + blocked = &shutdownIOBackend{shutdownCASBackend: observed, entered: make(chan struct{}, 1), abort: make(chan struct{})} + return blocked + }) + t.Cleanup(func() { blocked.once.Do(func() { close(blocked.abort) }) }) + h := srv.Handler() + if rec := putContent(t, h, "/api/p/"+p.ID+"/upload/content?path=notes.md", "before\n", ""); rec.Code != http.StatusOK { + t.Fatalf("seed: %d", rec.Code) + } + doc := openRoom(t, srv, p.ID+"/notes.md") + typeInto(doc, "dirty tail\n") + blocked.mu.Lock() + if step == "room_bytes" { + blocked.method, blocked.prefix = "GET", p.ID+"/blobs/" + } else if step == "room_head" { + blocked.method, blocked.prefix = "PUT", p.ID+"/ydoc/head/" + } else { + blocked.method, blocked.prefix = "PUT", p.ID+"/ydoc/" + } + blocked.mu.Unlock() + ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond) + defer cancel() + err := waitShutdownCall(t, func() error { return srv.Close(ctx) }, func() { + blocked.mu.Lock() + blocked.method = "" + blocked.mu.Unlock() + blocked.once.Do(func() { close(blocked.abort) }) + }) + if !errors.Is(err, context.DeadlineExceeded) { + t.Fatalf("Close = %v, want deadline exceeded", err) + } + select { + case <-blocked.entered: + default: + t.Fatal("Close never reached the selected room storage operation") + } + blocked.mu.Lock() + blocked.method = "" + blocked.mu.Unlock() + }) + } +} + +func TestCollabCloseCancelsOnLoadStorageWhileRequestLives(t *testing.T) { + for _, step := range []string{"room_head", "room_history"} { + t.Run(step, func(t *testing.T) { + var blocked *shutdownIOBackend + srv, p, _ := newHub(t, true, func(be remote.Backend) remote.Backend { + cond, ok := be.(remote.ConditionalPutter) + if !ok { + t.Fatal("file backend does not support CAS") + } + observed := &shutdownCASBackend{Backend: be, cond: cond, entered: make(chan shutdownCASCall, 16)} + blocked = &shutdownIOBackend{shutdownCASBackend: observed, entered: make(chan struct{}, 1), abort: make(chan struct{})} + return blocked + }) + h := srv.Handler() + if rec := putContent(t, h, "/api/p/"+p.ID+"/upload/content?path=notes.md", "before\n", ""); rec.Code != http.StatusOK { + t.Fatalf("seed: %d", rec.Code) + } + room := p.ID + "/notes.md" + abort := func() { + blocked.mu.Lock() + blocked.method = "" + blocked.mu.Unlock() + blocked.once.Do(func() { close(blocked.abort) }) + } + t.Cleanup(func() { + abort() + srv.rooms().drop(room) + srv.rooms().settle() + }) + blocked.mu.Lock() + blocked.method = "GET" + blocked.prefix = p.ID + "/ydoc/head/" + if step == "room_history" { + blocked.prefix = p.ID + "/ydoc/" + shaOf("before\n") + } + blocked.mu.Unlock() + requestCtx, cancelRequest := context.WithCancel(context.Background()) + defer cancelRequest() + loaded := make(chan error, 1) + go func() { loaded <- srv.ydocs().OnLoadDocument(requestCtx, room, crdt.New()) }() + select { + case <-blocked.entered: + case <-time.After(3 * time.Second): + t.Fatal("OnLoad never reached the selected seed storage read") + } + ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond) + defer cancel() + if err := waitShutdownCall(t, func() error { return srv.Close(ctx) }, abort); err != nil && !errors.Is(err, context.DeadlineExceeded) { + t.Fatalf("Close: %v", err) + } + if err := waitShutdownCall(t, func() error { return <-loaded }, abort); !errors.Is(err, context.Canceled) { + t.Fatalf("OnLoad = %v, want lifecycle cancellation", err) + } + if requestCtx.Err() != nil { + t.Fatalf("request context was canceled: %v", requestCtx.Err()) + } + }) + } +} + +func TestCollabCloseCancelsLeaseRenewalAndPendingSnapshot(t *testing.T) { + oldTTL := leaseTTL + leaseTTL = 120 * time.Millisecond + t.Cleanup(func() { leaseTTL = oldTTL }) + fastRoom(t, 50*time.Millisecond, 100*time.Millisecond) + srv, p, root, observed := newShutdownHub(t) + h := srv.Handler() + if rec := putContent(t, h, "/api/p/"+p.ID+"/upload/content?path=notes.md", "before\n", ""); rec.Code != http.StatusOK { + t.Fatalf("seed: %d", rec.Code) + } + room := p.ID + "/notes.md" + doc := crdt.New() + if err := srv.ydocs().OnLoadDocument(context.Background(), room, doc); err != nil { + t.Fatal(err) + } + srv.rooms().writer(room, coEditor) + t.Cleanup(func() { + srv.rooms().drop(room) + srv.rooms().settle() + }) + release, assertBlocked, assertFree := holdCollabShutdownLock(t, root) + observed.watchKey("/lease/notes.md") + typeInto(doc, "dirty tail\n") + call := awaitShutdownCAS(t, observed) + + ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond) + defer cancel() + if err := waitShutdownCall(t, func() error { return srv.Close(ctx) }, release); !errors.Is(err, context.DeadlineExceeded) { + t.Fatalf("Close = %v, want deadline exceeded", err) + } + awaitShutdownCASReturn(t, call) + assertBlocked() + leasePath := filepath.Join(root, p.ID, "lease", "notes.md") + atClose, err := os.ReadFile(leasePath) + if err != nil { + t.Fatal(err) + } + // This crosses both timers, not a shutdown latency assertion. + time.Sleep(snapshotMax + leaseTTL) + if got, err := os.ReadFile(leasePath); err != nil || string(got) != string(atClose) { + t.Fatalf("lease changed after Close while externally locked: %q, %v", got, err) + } + if got := get(t, h, "/api/p/"+p.ID+"/file?path=notes.md").Body.String(); got != "before\n" { + t.Fatalf("pending snapshot wrote after Close: %q", got) + } + assertBlocked() + release() + assertFree() + time.Sleep(snapshotMax + leaseTTL) + if got, err := os.ReadFile(leasePath); err != nil || string(got) != string(atClose) { + t.Fatalf("lease renewed after external unlock: %q, %v", got, err) + } + again, cancelAgain := context.WithTimeout(context.Background(), 3*time.Second) + defer cancelAgain() + if err := waitShutdownCall(t, func() error { return srv.Close(again) }, release); err != nil { + t.Fatalf("uncontended second Close: %v", err) + } +} + +func TestCollabCloseCancelsOnLoadWithoutCancelingRequest(t *testing.T) { + srv, p, root, observed := newShutdownHub(t) + h := srv.Handler() + if rec := putContent(t, h, "/api/p/"+p.ID+"/upload/content?path=notes.md", "before\n", ""); rec.Code != http.StatusOK { + t.Fatalf("seed: %d", rec.Code) + } + t.Cleanup(func() { + srv.rooms().drop(p.ID + "/notes.md") + srv.rooms().settle() + }) + release, assertBlocked, assertFree := holdCollabShutdownLock(t, root) + observed.watchKey("/lease/notes.md") + requestCtx, cancelRequest := context.WithCancel(context.Background()) + defer cancelRequest() + loaded := make(chan error, 1) + go func() { + loaded <- srv.ydocs().OnLoadDocument(requestCtx, p.ID+"/notes.md", crdt.New()) + }() + call := awaitShutdownCAS(t, observed) + ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond) + defer cancel() + if err := waitShutdownCall(t, func() error { return srv.Close(ctx) }, release); err != nil && !errors.Is(err, context.DeadlineExceeded) { + t.Fatalf("Close: %v", err) + } + awaitShutdownCASReturn(t, call) + if err := waitShutdownCall(t, func() error { return <-loaded }, release); !errors.Is(err, context.Canceled) { + t.Fatalf("OnLoad = %v, want server lifecycle cancellation", err) + } + if err := requestCtx.Err(); err != nil { + t.Fatalf("request was canceled by Close: %v", err) + } + assertBlocked() + release() + assertFree() + + healthy, hp, _ := newHub(t, true, nil) + hh := healthy.Handler() + if rec := putContent(t, hh, "/api/p/"+hp.ID+"/upload/content?path=notes.md", "before\n", ""); rec.Code != http.StatusOK { + t.Fatalf("healthy seed: %d", rec.Code) + } + doc := openRoom(t, healthy, hp.ID+"/notes.md") + typeInto(doc, "caller tail\n") + healthyCtx, cancelHealthy := context.WithTimeout(context.Background(), 3*time.Second) + defer cancelHealthy() + if err := waitShutdownCall(t, func() error { return healthy.Close(healthyCtx) }, release); err != nil { + t.Fatalf("healthy Close: %v", err) + } + if got := get(t, hh, "/api/p/"+hp.ID+"/file?path=notes.md").Body.String(); got != "before\ncaller tail\n" { + t.Fatalf("caller context did not save dirty tail: %q", got) + } +} + +func TestCollabParkCopyCallerCancellationPreservesBytes(t *testing.T) { + srv, p, root, observed := newShutdownHub(t) + h := srv.Handler() + release, assertBlocked, assertFree := holdCollabShutdownLock(t, root) + observed.watchKey("/journal/") + ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond) + defer cancel() + if err := waitShutdownCall(t, func() error { + srv.parkCopy(ctx, coEditor, p.ID, "notes.md", "parked.md", "parked tail\n") + return ctx.Err() + }, release); !errors.Is(err, context.DeadlineExceeded) { + t.Fatalf("park caller = %v, want deadline exceeded", err) + } + awaitShutdownCASReturn(t, awaitShutdownCAS(t, observed)) + // Cancel any delayed retry before releasing the fault injection. + closeCtx, cancelClose := context.WithTimeout(context.Background(), 100*time.Millisecond) + defer cancelClose() + if err := waitShutdownCall(t, func() error { return srv.Close(closeCtx) }, release); err != nil { + t.Fatalf("Close: %v", err) + } + assertBlocked() + if got := get(t, h, "/api/p/"+p.ID+"/file?path=parked.md"); got.Code != http.StatusNotFound { + t.Fatalf("canceled caller parked bytes while locked: %d %q", got.Code, got.Body.String()) + } + release() + assertFree() +} + +func TestCollabParkCopyCancellationUsesTheUploadCAS(t *testing.T) { + for _, closedFirst := range []bool{false, true} { + name := "inflight_retry" + if closedFirst { + name = "closed_before_failed_attempt" + } + t.Run(name, func(t *testing.T) { + oldBackoff, oldMax := parkBackoff, parkBackoffMax + parkBackoff, parkBackoffMax = 20*time.Millisecond, 20*time.Millisecond + srv, p, root, observed := newShutdownHub(t) + t.Cleanup(func() { + _ = srv.Close(context.Background()) + parkBackoff, parkBackoffMax = oldBackoff, oldMax + }) + srv.Handler() + if closedFirst { + if err := srv.Close(context.Background()); err != nil { + t.Fatal(err) + } + } + release, assertBlocked, assertFree := holdCollabShutdownLock(t, root) + observed.watchKey("/journal/") + ctx, cancel := context.WithCancel(context.Background()) + cancel() + if err := waitShutdownCall(t, func() error { + srv.parkCopy(ctx, coEditor, p.ID, "notes.md", "parked.md", "parked tail\n") + return nil + }, release); err != nil { + t.Fatal(err) + } + initial := awaitShutdownCAS(t, observed) + awaitShutdownCASReturn(t, initial) + if !closedFirst { + call := awaitShutdownCAS(t, observed) + closeCtx, cancelClose := context.WithTimeout(context.Background(), 100*time.Millisecond) + defer cancelClose() + if err := waitShutdownCall(t, func() error { return srv.Close(closeCtx) }, release); err != nil { + t.Fatalf("Close with park retry: %v", err) + } + awaitShutdownCASReturn(t, call) + } + assertBlocked() + time.Sleep(3 * parkBackoffMax) + if got := get(t, srv.Handler(), "/api/p/"+p.ID+"/file?path=parked.md"); got.Code != http.StatusNotFound { + t.Fatalf("canceled park wrote while externally locked: %d %q", got.Code, got.Body.String()) + } + release() + assertFree() + time.Sleep(3 * parkBackoffMax) + if got := get(t, srv.Handler(), "/api/p/"+p.ID+"/file?path=parked.md"); got.Code != http.StatusNotFound { + t.Fatalf("park retry wrote after shutdown and unlock: %d %q", got.Code, got.Body.String()) + } + }) + } +} + func TestCollabCloseWritesLiveRooms(t *testing.T) { - srv, p, _ := newHub(t, true, nil) + srv, p, root := newHub(t, true, nil) h := srv.Handler() url := "/api/p/" + p.ID + "/file?path=notes.md" if rec := putContent(t, h, "/api/p/"+p.ID+"/upload/content?path=notes.md", "before\n", ""); rec.Code != http.StatusOK { @@ -124,6 +630,97 @@ func TestCollabCloseWritesLiveRooms(t *testing.T) { if got := get(t, h, url).Body.String(); got != "before\ntyped just before the deploy\n" { t.Fatalf("after Close the file says %q", got) } + body, err := os.ReadFile(filepath.Join(root, p.ID, "lease", "notes.md")) + if err != nil && !errors.Is(err, os.ErrNotExist) { + t.Fatal(err) + } + if err == nil { + var lease leaseRecord + if err := json.Unmarshal(body, &lease); err != nil { + t.Fatal(err) + } + if lease.Holder != srv.nodeID() || !lease.Until.IsZero() { + t.Fatalf("healthy Close left or replaced the room lease: %s", body) + } + } +} + +func TestCollabCloseCancelsContendedFileCAS(t *testing.T) { + srv, p, root := newHub(t, true, nil) + h := srv.Handler() + if rec := putContent(t, h, "/api/p/"+p.ID+"/upload/content?path=notes.md", "before\n", ""); rec.Code != http.StatusOK { + t.Fatalf("seed write: %d", rec.Code) + } + room := p.ID + "/notes.md" + doc := crdt.New() + if err := srv.ydocs().OnLoadDocument(context.Background(), room, doc); err != nil { + t.Fatal(err) + } + srv.rooms().writer(room, coEditor) + t.Cleanup(func() { + srv.rooms().drop(room) + srv.rooms().settle() + }) + typeInto(doc, "dirty shutdown tail\n") + if got := doc.GetText("body").ToString(); got != "before\ndirty shutdown tail\n" { + t.Fatalf("dirty document = %q", got) + } + + lockPath := filepath.Join(root, ".bdrive-tmp-cas.lock") + holder, err := os.OpenFile(lockPath, os.O_CREATE|os.O_RDWR, 0o644) + if err != nil { + t.Fatal(err) + } + defer holder.Close() + if err := syscall.Flock(int(holder.Fd()), syscall.LOCK_EX); err != nil { + t.Fatal(err) + } + var unlockHolder sync.Once + releaseHolder := func() { + unlockHolder.Do(func() { _ = syscall.Flock(int(holder.Fd()), syscall.LOCK_UN) }) + } + defer releaseHolder() + + probe, err := os.OpenFile(lockPath, os.O_RDWR, 0) + 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() { done <- srv.Close(ctx) }() + + select { + case err := <-done: + if ctx.Err() != context.DeadlineExceeded || !errors.Is(err, ctx.Err()) { + t.Fatalf("Close error = %v, context error = %v, want context deadline exceeded", err, ctx.Err()) + } + case <-time.After(3 * time.Second): + // The guard bounds a broken implementation; it is not a shutdown deadline assertion. + releaseHolder() + select { + case <-done: + case <-time.After(3 * time.Second): + } + t.Fatal("Close did not return while the external CAS lock was held") + } + + if err := syscall.Flock(int(probe.Fd()), syscall.LOCK_EX|syscall.LOCK_NB); !errors.Is(err, syscall.EAGAIN) && !errors.Is(err, syscall.EWOULDBLOCK) { + t.Fatalf("third descriptor acquired CAS lock before holder release: %v", err) + } + if got := get(t, h, "/api/p/"+p.ID+"/file?path=notes.md").Body.String(); got != "before\n" { + t.Fatalf("contended Close changed file to %q", got) + } + + releaseHolder() + if err := syscall.Flock(int(probe.Fd()), syscall.LOCK_EX|syscall.LOCK_NB); err != nil { + t.Fatalf("third descriptor could not acquire CAS lock after holder release: %v", err) + } + if err := syscall.Flock(int(probe.Fd()), syscall.LOCK_UN); err != nil { + t.Fatal(err) + } } func TestCollabSecondProcessRefusesARoomWithoutARelay(t *testing.T) { @@ -185,11 +782,11 @@ func TestCollabSeedsAFileOnceAcrossProcesses(t *testing.T) { if !strings.Contains(key, "/ydoc/") { t.Fatalf("first conditional write was %s, not the seed", key) } - if err := b.seedDoc(room, docB); err != nil { + if err := b.seedDoc(context.Background(), room, docB); err != nil { t.Fatal(err) } } - if err := a.seedDoc(room, docA); err != nil { + if err := a.seedDoc(context.Background(), room, docA); err != nil { t.Fatal(err) } if err := docA.ApplyUpdate(docB.EncodeStateAsUpdate()); err != nil { @@ -226,7 +823,7 @@ func TestCollabReopenedRoomKnowsTheEditsARevertUndid(t *testing.T) { write("base\n") // reverted from outside: this blob has a stored state from before "more" re := crdt.New() - if err := srv.seedDoc(room, re); err != nil { + if err := srv.seedDoc(context.Background(), room, re); err != nil { t.Fatal(err) } if got := re.GetText("body").ToString(); got != "base\n" { diff --git a/internal/webapp/ycollab_test.go b/internal/webapp/ycollab_test.go index be7d1df1..442c278e 100644 --- a/internal/webapp/ycollab_test.go +++ b/internal/webapp/ycollab_test.go @@ -175,7 +175,7 @@ func TestYCollabSeedsTheDocumentFromTheFile(t *testing.T) { } doc := crdt.New() - if err := srv.seedDoc(p.ID+"/notes.md", doc); err != nil { + if err := srv.seedDoc(context.Background(), p.ID+"/notes.md", doc); err != nil { t.Fatal(err) } if doc.GetText("body").Len() == 0 { @@ -196,7 +196,7 @@ func TestYCollabSeedsEmptyForWhatIsNotThere(t *testing.T) { "malformed-room-name", } { doc := crdt.New() - if err := srv.seedDoc(room, doc); err != nil { + if err := srv.seedDoc(context.Background(), room, doc); err != nil { t.Errorf("%s: %v", room, err) } if n := doc.GetText("body").Len(); n != 0 { @@ -299,7 +299,7 @@ func TestYCollabSnapshotsTheDocumentAsTheHuman(t *testing.T) { // A room, as ygo would hand it to us, edited by somebody. doc := crdt.New() - if err := srv.seedDoc(room, doc); err != nil { + if err := srv.seedDoc(context.Background(), room, doc); err != nil { t.Fatal(err) } srv.rooms().load(room, doc) @@ -341,7 +341,7 @@ func TestYCollabSnapshotOfUnchangedTextIsNotAWrite(t *testing.T) { putContent(t, h, "/api/p/"+p.ID+"/upload/content?path=notes.md", body, "") doc := crdt.New() - if err := srv.seedDoc(room, doc); err != nil { + if err := srv.seedDoc(context.Background(), room, doc); err != nil { t.Fatal(err) } srv.rooms().load(room, doc) @@ -364,7 +364,7 @@ func TestYCollabSnapshotNeedsSomeoneToAttribute(t *testing.T) { putContent(t, h, "/api/p/"+p.ID+"/upload/content?path=notes.md", "original\n", "") doc := crdt.New() - _ = srv.seedDoc(room, doc) + _ = srv.seedDoc(context.Background(), room, doc) srv.rooms().load(room, doc) // loaded, but no writer recorded txt := doc.GetText("body") doc.Transact(func(txn *crdt.Transaction) { txt.Insert(txn, 0, "SHOULD NOT LAND ", nil) }) @@ -397,7 +397,7 @@ func TestYCollabRebuiltRoomMergesAReturningDocumentOnce(t *testing.T) { open := func() *crdt.Doc { t.Helper() doc := crdt.New() - if err := srv.seedDoc(room, doc); err != nil { + if err := srv.seedDoc(context.Background(), room, doc); err != nil { t.Fatal(err) } srv.rooms().load(room, doc) @@ -468,7 +468,7 @@ func TestYCollabSnapshotStoresTheHistoryOfWhatItWrote(t *testing.T) { room := p.ID + "/notes.md" putContent(t, h, "/api/p/"+p.ID+"/upload/content?path=notes.md", "x\n", "") doc := crdt.New() - if err := srv.seedDoc(room, doc); err != nil { + if err := srv.seedDoc(context.Background(), room, doc); err != nil { t.Fatal(err) } srv.rooms().load(room, doc) @@ -478,7 +478,7 @@ func TestYCollabSnapshotStoresTheHistoryOfWhatItWrote(t *testing.T) { srv.snapshotRoom(context.Background(), room) sha := srv.rooms().base(room) - state := srv.roomHistory(p.ID, sha) + state := srv.roomHistory(context.Background(), p.ID, sha) if state == nil { t.Fatalf("no history stored for %s", sha) } diff --git a/internal/webapp/ycollab_writer_test.go b/internal/webapp/ycollab_writer_test.go index faea1c07..50474ca0 100644 --- a/internal/webapp/ycollab_writer_test.go +++ b/internal/webapp/ycollab_writer_test.go @@ -36,7 +36,7 @@ func openRoom(t *testing.T, srv *Server, room string) *crdt.Doc { t.Helper() doc := crdt.New() srv.rooms().load(room, doc) - if err := srv.seedDoc(room, doc); err != nil { + if err := srv.seedDoc(context.Background(), room, doc); err != nil { t.Fatal(err) } srv.rooms().writer(room, User{Email: "editor@example.com", Name: "An Editor"})