diff --git a/cmd/codeaf/app_take_test.go b/cmd/codeaf/app_take_test.go index 25a712a33f..4bf81eb4a7 100644 --- a/cmd/codeaf/app_take_test.go +++ b/cmd/codeaf/app_take_test.go @@ -322,6 +322,22 @@ func appTakeRoundTrip(t *testing.T, daemon bool, missingWorkspace ...bool) { Agent appwire.Agent `json:"agent"` }](t, a.w, "agents.create", map[string]any{"workspace": workspace}).Agent e2Turn(t, a, created.ID, "first") + heldFence := func() uint64 { + rows := machineCall[appwire.ChatsResult](t, a.w, "chats.list", struct{}{}) + for _, r := range rows.Chats { + if r.Cell == created.ID && r.Here { + return r.Seen.Fence + } + } + return 0 + } + var firstFence uint64 + if !absent { + appSyncWait(t, "A's first published fence", func() bool { + firstFence = heldFence() + return firstFence > 0 + }) + } // Start continuation from a chat saved by the real app leaving path. This // exercises a saved chat independently of background first-publication timing. e2Quit(t, a) @@ -332,6 +348,11 @@ func appTakeRoundTrip(t *testing.T, daemon bool, missingWorkspace ...bool) { } else { a = e2StartApp(t, dirA, daemon) machineCall[appwire.TakeResult](t, a.w, "agents.open", appwire.IDParams{ID: created.ID}) + // Reopening owns the journal before its background drive acquires. + // Take consent must name that new hold rather than race its arrival. + appSyncWait(t, "A's reopened chat holds a new fence", func() bool { + return heldFence() > firstFence + }) } t.Logf("created %s", string(e2JSON(created))) diff --git a/cmd/codeaf/appsync_admission_test.go b/cmd/codeaf/appsync_admission_test.go index 00aaa54f0c..31d706014d 100644 --- a/cmd/codeaf/appsync_admission_test.go +++ b/cmd/codeaf/appsync_admission_test.go @@ -50,10 +50,29 @@ func TestC6AppSendDoesNotWaitForSyncStartupOrCapture(t *testing.T) { entered, release := make(chan struct{}), make(chan struct{}) var once sync.Once unblock := func() { once.Do(func() { close(release) }) } - var requests atomic.Int64 + var requestMu sync.Mutex + var requestPaths []string + recordRequest := func(r *http.Request) { + requestMu.Lock() + defer requestMu.Unlock() + requestPaths = append(requestPaths, r.Method+" "+r.URL.Path) + } + requests := func() []string { + requestMu.Lock() + defer requestMu.Unlock() + return append([]string(nil), requestPaths...) + } var first sync.Once srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { - requests.Add(1) + // The drive's startup opens with a device listing, at the same + // moment as the app's own chat listing and watch. Every request + // is held, so a startup held at its listing cannot reach the + // chat's cell; the listings themselves are not admission's work + // and arrive in either order, so they are not counted. + listing := r.Method == http.MethodGet && (r.URL.Path == "/v1/dir/list" || r.URL.Path == "/v1/dir/watch") + if boundary != "startup" || !listing { + recordRequest(r) + } if boundary == "startup" { first.Do(func() { close(entered) }) select { @@ -120,7 +139,7 @@ func TestC6AppSendDoesNotWaitForSyncStartupOrCapture(t *testing.T) { } }) } - before := requests.Load() + before := len(requests()) type sent struct { turn uint64 err error @@ -145,8 +164,8 @@ func TestC6AppSendDoesNotWaitForSyncStartupOrCapture(t *testing.T) { case <-time.After(10 * time.Second): t.Fatal("the admitted send never reached the real session model") } - if got := requests.Load(); got != before { - t.Fatalf("admission added relay work: before %d, after %d", before, got) + if got := requests(); len(got) != before { + t.Fatalf("admission added relay work: before %d, after %d; added %v", before, len(got), got[before:]) } unblock() releaseModel() diff --git a/cmd/codeaf/chatv3.go b/cmd/codeaf/chatv3.go index 465dc02a50..730e2e553c 100644 --- a/cmd/codeaf/chatv3.go +++ b/cmd/codeaf/chatv3.go @@ -1473,7 +1473,8 @@ func openV3Agent(cfg session.Config, workspace string, open func(session.Config) if model != "" { cfg.Model = model } - cfg = v3Seated(v3Migrated(cfg)) + cfg, claim := v3SeatedPending(v3Migrated(cfg)) + defer claim(false) // Restore the gate before construction, so restored work cannot start behind // the profile default. Return the launch config unchanged: a subsequent new // conversation must not inherit this one's saved override. @@ -1505,6 +1506,7 @@ func openV3Agent(cfg session.Config, workspace string, open func(session.Config) } agent, err := open(restored) if err == nil { + claim(true) if model != "" { if err := agent.ContinueWithModel(model, modelNotice); err != nil { closeV3Agent(agent) @@ -1579,11 +1581,8 @@ func v3TakeOverInstead(cfg session.Config, workspace string) (*session.Agent, se if err != nil { return nil, cfg, err } - agent, err := v3OpenSession(fresh) - if err != nil { - return nil, cfg, err - } - return agent, fresh, nil + agent, fresh, _, err := openV3Agent(fresh, workspace, v3OpenSession) + return agent, fresh, err } // sessionHeldElsewhereSentence is what a person reads when the conversation diff --git a/cmd/codeaf/chatv3_layout.go b/cmd/codeaf/chatv3_layout.go index 468220e305..7d05494349 100644 --- a/cmd/codeaf/chatv3_layout.go +++ b/cmd/codeaf/chatv3_layout.go @@ -787,16 +787,23 @@ func v3Migrated(cfg session.Config) session.Config { return cfg } -// v3Seated gives the session the executor it owns, built once here from the -// class the cell declared and the workspace its tools run in. +// v3Seated is for a caller that already holds the journal. It gives the session +// the executor it owns, built from the cell class and its tools' workspace. // // A folder that is not a readable cell gets no seat and its calls run on the // host as they always did, and so does every folder without a compatible engine. The // seat is set on every call, so a config reused for the next conversation never // carries the last one's executor. func v3Seated(cfg session.Config) session.Config { + cfg, claim := v3SeatedPending(cfg) + claim(true) + return cfg +} + +// v3SeatedPending builds the seat before committing its drive to journal ownership. +func v3SeatedPending(cfg session.Config) (session.Config, func(bool)) { cfg.Seat, cfg.Machine, cfg.Seals, cfg.Interrupted = nil, nil, nil, nil - seat, machine, seals := v3SeatWith(cfg.Place, nil, syncDrives) + seat, machine, seals, claim := v3SeatPending(cfg.Place, nil, syncDrives) cfg.Seat, cfg.Interrupted = seat, executor.InterruptedOn(seat) cfg.Renewals = syncDrives.renewalsOf(cfg.Place.ID()) if machine != nil { @@ -805,7 +812,7 @@ func v3Seated(cfg session.Config) session.Config { if seals != nil { cfg.Seals = seals } - return cfg + return cfg, claim } // v3SeatOf is the seat of the cell a place names, the machine view it was @@ -822,12 +829,19 @@ func v3SeatOf(place session.Place, say func(string)) (executor.Seat, *preflight. // the chat its tool calls are refused. A nil book, or sync off, is the seat // exactly as it was. func v3SeatWith(place session.Place, say func(string), drives *driveBook) (executor.Seat, *preflight.Machine, *cellstore.SealWatch) { + seat, machine, watch, claim := v3SeatPending(place, say, drives) + claim(true) + return seat, machine, watch +} + +func v3SeatPending(place session.Place, say func(string), drives *driveBook) (executor.Seat, *preflight.Machine, *cellstore.SealWatch, func(bool)) { + noop := func(bool) {} if !cell.Available() || place.Dir == "" || place.Workspace == "" { - return nil, nil, nil + return nil, nil, nil, noop } c, err := cell.OpenAt(place.Dir, filepath.Base(place.Dir)) if err != nil { - return nil, nil, nil + return nil, nil, nil, noop } class, ok := executor.ParseClass(string(c.Meta().Class)) if !ok { @@ -836,7 +850,7 @@ func v3SeatWith(place session.Place, say func(string), drives *driveBook) (execu machine, err := preflight.OpenMachine(c.Root, place.Workspace) watch := &cellstore.SealWatch{Say: say} failed(watch, err) - drive := drives.driveOf(c, cellstore.EngineFor(place.Workspace), watch.Report) + drive, claim := drives.pendingDriveOf(c, cellstore.EngineFor(place.Workspace), watch.Report) if drive != nil { watch.OnTurnEnd = drive.Idle // the agent waits for the person: upload what is sealed now } @@ -847,7 +861,7 @@ func v3SeatWith(place session.Place, say func(string), drives *driveBook) (execu // the first input; their journal writes need a tail seal even before chat. // Window attachment reuses the engine's seat rather than constructing one. watch.Waiting() - return executor.Gated(seat, driveGate(drive), driveWorkGate(drive)), machine, watch + return executor.Gated(seat, driveGate(drive), driveWorkGate(drive)), machine, watch, claim } // sealTail seals what the chat wrote after the turn-end seal (titles, summaries, diff --git a/cmd/codeaf/chatv3_session_test.go b/cmd/codeaf/chatv3_session_test.go index c68b4364bd..01fe0541dd 100644 --- a/cmd/codeaf/chatv3_session_test.go +++ b/cmd/codeaf/chatv3_session_test.go @@ -143,7 +143,7 @@ func TestALaunchThatMeetsALockOpensAFreshOneAndCarriesTheHeldRow(t *testing.T) { } defer func() { _ = first.Close() }() - _, _, _, err = openV3Agent(cfg, workspace, v3OpenSession) + _, seated, _, err := openV3Agent(cfg, workspace, v3OpenSession) // THE FAILURE IS TYPED AND CARRIES THE PATH, because the sentence // deliberately does not contain one and the surface needs one to point at. var held *sessionHeldElsewhere @@ -159,7 +159,7 @@ func TestALaunchThatMeetsALockOpensAFreshOneAndCarriesTheHeldRow(t *testing.T) { t.Fatalf("%v no longer reads as a locked session", err) } - second, fresh, err := v3TakeOverInstead(cfg, workspace) + second, fresh, err := v3TakeOverInstead(seated, workspace) if err != nil { t.Fatalf("the second window was left with nothing: %v", err) } diff --git a/cmd/codeaf/heldchat_drive_test.go b/cmd/codeaf/heldchat_drive_test.go new file mode 100644 index 0000000000..ef508eee0a --- /dev/null +++ b/cmd/codeaf/heldchat_drive_test.go @@ -0,0 +1,619 @@ +package main + +import ( + "bytes" + "context" + "crypto/sha256" + "errors" + "fmt" + "net/http" + "net/http/httptest" + "os" + "path/filepath" + "runtime" + "strings" + "sync/atomic" + "testing" + "time" + + "github.com/Agent-Field/codeaf/internal/cell" + "github.com/Agent-Field/codeaf/internal/cell/celltest" + "github.com/Agent-Field/codeaf/internal/cellstore" + "github.com/Agent-Field/codeaf/internal/directory" + "github.com/Agent-Field/codeaf/internal/executor" + "github.com/Agent-Field/codeaf/internal/filelock" + "github.com/Agent-Field/codeaf/internal/home" + "github.com/Agent-Field/codeaf/internal/identity" + "github.com/Agent-Field/codeaf/internal/relayserve" + "github.com/Agent-Field/codeaf/internal/session" + "github.com/Agent-Field/codeaf/internal/syncsetup" +) + +// heldChatRelay counts every cell write, including creation, rather than only +// the heartbeat that first exposed the second opener's ownership side effect. +func heldChatRelay(t *testing.T, id string) (*syncsetup.Sync, *atomic.Int64) { + t.Helper() + service := relayserve.New(relayserve.Config{Store: t.TempDir()}) + writes := &atomic.Int64{} + path := "/v1/dir/cells/" + id + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodGet && (r.URL.Path == path || strings.HasPrefix(r.URL.Path, path+"/")) { + writes.Add(1) + } + service.Handler.ServeHTTP(w, r) + })) + t.Cleanup(func() { server.Close(); _ = service.Close() }) + t.Setenv(syncsetup.URLVar, server.URL) + t.Setenv(syncsetup.IntervalVar, "3600000") + if _, err := identity.Ensure(home.Dir()); err != nil { + t.Fatalf("make this computer's identity: %v", err) + } + if syncsetup.Quiet(home.Dir()) { + t.Fatalf("the existing non-solo identity must permit relay requests") + } + s, ok, err := syncsetup.Open(home.Dir()) + if err != nil || !ok { + t.Fatalf("open relay sync = %v, %v; want true, nil", ok, err) + } + return s, writes +} + +type heldChatReceipt struct { + path string + raw []byte + info os.FileInfo +} + +func snapshotHeldChatReceipt(t *testing.T, s *syncsetup.Sync, id string) heldChatReceipt { + t.Helper() + key := sha256.Sum256([]byte(s.Ledger + "/" + s.Device.ID() + "/" + id)) + path := filepath.Join(home.Dir(), "v3", "sync", "holds", fmt.Sprintf("%x", key)) + raw, err := os.ReadFile(path) + if err != nil { + t.Fatalf("read P1's hold receipt: %v", err) + } + info, err := os.Stat(path) + if err != nil { + t.Fatalf("stat P1's hold receipt: %v", err) + } + return heldChatReceipt{path: path, raw: raw, info: info} +} + +func (before heldChatReceipt) check(t *testing.T) { + t.Helper() + raw, err := os.ReadFile(before.path) + if err != nil { + t.Fatalf("P1's hold receipt after P2 quit = %v; want the original receipt", err) + } + if !bytes.Equal(raw, before.raw) { + t.Fatalf("P1's hold receipt = %q; want unchanged %q", raw, before.raw) + } + info, err := os.Stat(before.path) + if err != nil { + t.Fatalf("stat P1's hold receipt after P2 quit: %v", err) + } else if !os.SameFile(before.info, info) { + t.Fatalf("P2 rewrote P1's hold receipt; want the same file") + } +} + +func checkHeldChatLease(t *testing.T, s *syncsetup.Sync, id string, before directory.CellView) { + t.Helper() + after, err := s.Dir.Cell(t.Context(), id) + if err != nil { + t.Fatalf("read the held chat after P2 quit: %v", err) + } + if after.Cell.Lease.Device != before.Cell.Lease.Device || after.Cell.Lease.Fence != before.Cell.Lease.Fence { + t.Fatalf("lease after P2 quit = %s fence %d; want %s fence %d", after.Cell.Lease.Device, after.Cell.Lease.Fence, before.Cell.Lease.Device, before.Cell.Lease.Fence) + } + if after.Cell.Lease.Expires == 0 || after.Cell.Lease.Expires < before.Cell.Lease.Expires { + t.Fatalf("lease expiry after P2 quit = %d; want unreleased and at least %d", after.Cell.Lease.Expires, before.Cell.Lease.Expires) + } + if after.Cell.Head != before.Cell.Head { + t.Fatalf("relay head after P2 quit = %q; want %q", after.Cell.Head, before.Cell.Head) + } +} + +func refuseHeldChatOpen(t *testing.T, cfg session.Config, writes *atomic.Int64) session.Config { + t.Helper() + _, seated, _, err := openV3Agent(cfg, cfg.Place.Workspace, session.New) + var held *sessionHeldElsewhere + if !errors.As(err, &held) { + t.Fatalf("P2 open = %v; want sessionHeldElsewhere from the real journal lock", err) + } + // On the unfixed product startup is asynchronous. Observe its first write + // before quitting; closeAll then joins startup and deterministically releases it. + deadline := time.Now().Add(2 * time.Second) + for writes.Load() == 0 && time.Now().Before(deadline) { + time.Sleep(10 * time.Millisecond) + } + syncDrives.closeAll() + return seated +} + +func TestAChatAnotherProcessHoldsIsNeitherRenewedNorReleasedFromHere(t *testing.T) { + celltest.Available(t) + t.Setenv(home.EnvVar, t.TempDir()) + t.Setenv("HOME", t.TempDir()) + // A missing explicit binary prevents even an installed engine from being + // used by the failed opener's background warm capture. P1 needs no seals. + t.Setenv("CODEAF_FURROW", filepath.Join(t.TempDir(), "absent-engine")) + c, err := cell.CreateIn(t.TempDir(), cell.Options{Class: cell.HostBound}) + if err != nil { + t.Fatalf("create the held chat: %v", err) + } + place := session.Place{Dir: c.Root, Workspace: t.TempDir()} + cfg := session.Config{Place: place, Workspace: place.Workspace, SessionFile: place.Transcript(), Model: "stub/chat", APIKey: "stub", BaseURL: "http://127.0.0.1:1"} + journal, err := os.OpenFile(place.Transcript(), os.O_CREATE|os.O_WRONLY|os.O_APPEND, 0600) + if err != nil { + t.Fatalf("open P1's journal: %v", err) + } + defer journal.Close() + if err := filelock.Lock(journal, true, true); err != nil { + t.Fatalf("hold P1's journal with the session's exclusive nonblocking flock: %v", err) + } + defer filelock.Unlock(journal) + s, writes := heldChatRelay(t, c.ID) + if _, err := s.Dir.Create(t.Context(), c.ID, directory.CellInit{}); err != nil { + t.Fatalf("create P1's relay record and lease: %v", err) + } + prior := syncDrives + firstBook := &driveBook{} + syncDrives = firstBook + defer func() { syncDrives.closeAll(); syncDrives = prior }() + // P1 resumes the relay's fence and records its receipt through the real + // Drive. Its sleeper holds all renewal ticks until Close cancels them. + first := newLiveDrive(func(func()) (*syncsetup.Drive, error) { + return s.Drive(t.Context(), cellstore.Engine{}, c, syncsetup.DriveOptions{Sleep: func(ctx context.Context, _ time.Duration) error { + <-ctx.Done() + return ctx.Err() + }}) + }) + firstBook.drive = map[string]*liveDrive{c.ID: first} + defer firstBook.closeAll() + if err := first.Gate(); err != nil { + t.Fatalf("P1's initial gate: %v", err) + } + writer := first.started() + if writer == nil || !writer.HoldsLease(t.Context()) { + t.Fatalf("P1 must start a real writer holding the relay lease") + } + before, err := s.Dir.Cell(t.Context(), c.ID) + if err != nil || before.Cell.Lease.Fence < 1 || before.Cell.Lease.Expires <= before.Now { + t.Fatalf("P1's initial relay lease = %+v, %v; want a live fence", before, err) + } + receipt := snapshotHeldChatReceipt(t, s, c.ID) + writes.Store(0) + syncDrives = &driveBook{} + seated := refuseHeldChatOpen(t, cfg, writes) + // Drain only this failed opener's warm capture before removing its files. + // Warm preparation screens locally and does not require a sealing binary. + _ = executor.Drain(t.Context(), seated.Seat) + if got := writes.Load(); got != 0 { + t.Fatalf("P2 made %d lease writes while opening and quitting a held chat; want 0", got) + } + checkHeldChatLease(t, s, c.ID, before) + receipt.check(t) + if !writer.HoldsLease(t.Context()) { + t.Fatalf("P1 lost its lease when P2 quit") + } + if err := writer.Gate(); err != nil { + t.Fatalf("P1's next work gate = %v; want nil", err) + } + confirmed, err := writer.ConfirmWork(t.Context()) + if err != nil || !confirmed { + t.Fatalf("P1's next work confirmation = %v, %v; want true, nil", confirmed, err) + } + if d := syncDrives.renewalsOf(c.ID); d != nil && d.started() != nil { + t.Fatalf("P2 retained a started drive for a chat whose journal it never held") + } +} + +func TestAQuitAtTheMoveQuestionLeavesTheOtherWindowsChatSendable(t *testing.T) { + cfg, c := spokenDoorChat(t) + s, writes := heldChatRelay(t, c.ID) + firstBook := syncDrives + defer firstBook.closeAll() + agent, firstSeat, _, err := openV3Agent(cfg, cfg.Place.Workspace, session.New) + if err != nil { + t.Fatalf("open P1's seated chat: %v", err) + } + defer agent.Close() + send := func(command string) { + t.Helper() + events, err := agent.SubmitBash(t.Context(), command) + if err != nil { + t.Fatalf("P1 send %q = %v; want admitted", command, err) + } + for event := range events { + if event.Kind == session.EventError { + t.Fatalf("P1 send %q reported %q; want no EventError", command, event.Text) + } + } + if err := executor.Drain(t.Context(), firstSeat.Seat); err != nil { + t.Fatalf("settle P1 send %q: %v", command, err) + } + } + send("! printf fixture") + deadline := time.Now().Add(15 * time.Second) + var before directory.CellView + for { + head, headErr := cellstore.Head(c) + before, err = s.Dir.Cell(t.Context(), c.ID) + _, receiptFence, receiptErr := s.SendLease(c.ID) + if headErr == nil && head != nil && err == nil && before.Cell.Head == head.Turn.ID && before.Cell.Lease.Fence > 0 && receiptErr == nil && receiptFence == before.Cell.Lease.Fence { + break + } + if time.Now().After(deadline) { + t.Fatalf("P1's spoken turn did not publish: relay = %+v, %v; local head error = %v", before, err, headErr) + } + time.Sleep(10 * time.Millisecond) + } + receipt := snapshotHeldChatReceipt(t, s, c.ID) + first := firstBook.renewalsOf(c.ID) + writes.Store(0) + syncDrives = &driveBook{} + t.Cleanup(func() { syncDrives.closeAll(); syncDrives = firstBook }) + seated := refuseHeldChatOpen(t, cfg, writes) + if err := executor.Drain(t.Context(), seated.Seat); err != nil { + t.Fatalf("settle P2's unused seat: %v", err) + } + checkHeldChatLease(t, s, c.ID, before) + receipt.check(t) + send("! printf again") + turns, err := cellstore.Turns(c) + if err != nil || len(turns) == 0 { + t.Fatalf("read P1's newest sealed turn = %d turns, %v", len(turns), err) + } + // A sealed turn names its writer by the device certificate's stamp, which + // is what the drive's store seals with, not by the relay's device id. + newest := turns[len(turns)-1] + if newest.Device != s.Device.Cert.Device || newest.Fence != before.Cell.Lease.Fence { + t.Fatalf("P1's next turn = %s fence %d; want %s fence %d", newest.Device, newest.Fence, s.Device.Cert.Device, before.Cell.Lease.Fence) + } + if first == nil || first.started() == nil || !first.started().HoldsLease(t.Context()) { + t.Fatalf("P1's drive must still hold its original lease after its next send") + } +} + +func TestTheFreshChatBesideAHeldOneWorksOnItsOwnSeat(t *testing.T) { + cfg, c := spokenDoorChat(t) + s, writes := heldChatRelay(t, c.ID) + firstBook := syncDrives + defer firstBook.closeAll() + agent, firstCfg, _, err := openV3Agent(cfg, cfg.Place.Workspace, session.New) + if err != nil { + t.Fatalf("open P1's seated chat: %v", err) + } + defer agent.Close() + events, err := agent.SubmitBash(t.Context(), "! printf fixture") + if err != nil { + t.Fatalf("P1 send = %v; want admitted", err) + } + for event := range events { + if event.Kind == session.EventError { + t.Fatalf("P1 send reported %q; want no EventError", event.Text) + } + } + if err := executor.Drain(t.Context(), firstCfg.Seat); err != nil { + t.Fatalf("settle P1's spoken turn: %v", err) + } + deadline := time.Now().Add(15 * time.Second) + var before directory.CellView + for { + head, headErr := cellstore.Head(c) + before, err = s.Dir.Cell(t.Context(), c.ID) + _, receiptFence, receiptErr := s.SendLease(c.ID) + if headErr == nil && head != nil && err == nil && before.Cell.Head == head.Turn.ID && before.Cell.Lease.Fence > 0 && receiptErr == nil && receiptFence == before.Cell.Lease.Fence { + break + } + if time.Now().After(deadline) { + t.Fatalf("P1's spoken turn did not publish: relay = %+v, %v; local head error = %v", before, err, headErr) + } + time.Sleep(10 * time.Millisecond) + } + heldTurns, err := cellstore.Turns(c) + if err != nil || len(heldTurns) == 0 { + t.Fatalf("P1's published turns = %d, %v; want a sealed turn", len(heldTurns), err) + } + receipt := snapshotHeldChatReceipt(t, s, c.ID) + first := firstBook.renewalsOf(c.ID) + writes.Store(0) + syncDrives = &driveBook{} + defer func() { syncDrives.closeAll(); syncDrives = firstBook }() + _, seated, _, err := openV3Agent(cfg, cfg.Place.Workspace, session.New) + var held *sessionHeldElsewhere + if !errors.As(err, &held) { + t.Fatalf("P2 open = %v; want sessionHeldElsewhere", err) + } + fresh, freshCfg, err := v3TakeOverInstead(seated, cfg.Place.Workspace) + if err != nil { + t.Fatalf("open the fresh chat beside P1: %v", err) + } + defer fresh.Close() + if freshCfg.Place.Dir == c.Root { + t.Fatal("the fresh chat shares P1's cell") + } + if err := executor.Drain(t.Context(), seated.Seat); err != nil { + t.Fatalf("settle P2's unused seat: %v", err) + } + if err := executor.Drain(t.Context(), freshCfg.Seat); err != nil { + t.Fatalf("settle the fresh chat's startup: %v", err) + } + freshCell, err := cell.OpenAt(freshCfg.Place.Dir, freshCfg.Place.ID()) + if err != nil { + t.Fatalf("open the fresh chat's cell: %v", err) + } + freshBefore, err := cellstore.Turns(freshCell) + if err != nil { + t.Fatalf("read the fresh chat before sending: %v", err) + } + events, err = fresh.SubmitBash(t.Context(), "! printf beside") + if err != nil { + t.Fatalf("fresh chat send = %v; want admitted", err) + } + for event := range events { + if event.Kind == session.EventError { + t.Fatalf("fresh chat send reported %q; want no EventError", event.Text) + } + } + if err := executor.Drain(t.Context(), freshCfg.Seat); err != nil { + t.Fatalf("settle the fresh chat's command: %v", err) + } + checkHeld := func() { + t.Helper() + turns, err := cellstore.Turns(c) + if err != nil || len(turns) != len(heldTurns) { + t.Fatalf("P1's sealed turns = %d, %v; want unchanged count %d", len(turns), err, len(heldTurns)) + } + if got, want := turns[len(turns)-1].ID, heldTurns[len(heldTurns)-1].ID; got != want { + t.Fatalf("P1's newest turn = %s; want unchanged %s", got, want) + } + if got := writes.Load(); got != 0 { + t.Fatalf("P2 made %d writes to P1's relay cell; want 0", got) + } + checkHeldChatLease(t, s, c.ID, before) + receipt.check(t) + if first == nil || first.started() == nil || !first.started().HoldsLease(t.Context()) { + t.Fatal("P1 lost its hold while P2 used or quit the fresh chat") + } + } + checkHeld() + turns, err := cellstore.Turns(freshCell) + if err != nil || len(turns) <= len(freshBefore) { + t.Fatalf("fresh chat's sealed turns = %d, %v; want a new turn beyond %d", len(turns), err, len(freshBefore)) + } + if d := syncDrives.renewalsOf(freshCell.ID); d == nil || d.started() == nil { + t.Fatal("P2 has no started drive for the fresh chat's own cell") + } + if d := syncDrives.renewalsOf(c.ID); d != nil && d.started() != nil { + t.Fatal("P2 started a drive for P1's held chat") + } + syncDrives.closeAll() + checkHeld() +} + +// waitParkedIn waits until some goroutine is blocked on a channel inside fn, +// which is how a waiter on a parked drive's startup boundary looks. +func waitParkedIn(t *testing.T, fn string) { + t.Helper() + deadline := time.Now().Add(10 * time.Second) + buf := make([]byte, 1<<20) + for { + n := runtime.Stack(buf, true) + for n == len(buf) { + buf = make([]byte, len(buf)*2) + n = runtime.Stack(buf, true) + } + for _, g := range strings.Split(string(buf[:n]), "\n\n") { + head, _, _ := strings.Cut(g, "\n") + if (strings.Contains(head, "[chan receive") || strings.Contains(head, "[select")) && strings.Contains(g, fn) { + return + } + } + if time.Now().After(deadline) { + t.Fatalf("no goroutine is waiting inside %s on the parked startup", fn) + } + time.Sleep(5 * time.Millisecond) + } +} + +func TestAParkedChatDriveWaitsForItsJournalDecision(t *testing.T) { + celltest.Available(t) + t.Setenv(home.EnvVar, t.TempDir()) + t.Setenv(syncsetup.URLVar, "http://relay.invalid") + + fixture := func(t *testing.T) (*driveBook, *liveDrive, *atomic.Int64, func(bool)) { + t.Helper() + calls := &atomic.Int64{} + d := newAdmissionDrive(func(func()) (*syncsetup.Drive, error) { + calls.Add(1) + return nil, nil + }) + book := &driveBook{drive: map[string]*liveDrive{"chat": d}} + t.Cleanup(book.closeAll) + got, claim := book.pendingDriveOf(cell.Cell{ID: "chat"}, cellstore.Engine{}, func(error) {}) + if got != d { + t.Fatalf("the parked drive was replaced while its journal open was in flight") + } + return book, d, calls, claim + } + claimAgain := func(t *testing.T, book *driveBook, d *liveDrive) func(bool) { + t.Helper() + got, claim := book.pendingDriveOf(cell.Cell{ID: "chat"}, cellstore.Engine{}, func(error) {}) + if got != d { + t.Fatalf("a second opener must share the existing drive") + } + return claim + } + waiters := func(t *testing.T, d *liveDrive) func() { + t.Helper() + gate, store := make(chan error, 1), make(chan error, 1) + entered := make(chan struct{}, 2) + ctx, cancel := context.WithTimeout(t.Context(), 10*time.Second) + t.Cleanup(cancel) + go func() { entered <- struct{}{}; gate <- d.Gate() }() + go func() { + entered <- struct{}{} + chosen, used, err := (lateStore{live: d, engine: cellstore.Engine{Workspace: "local"}}).readyStore(ctx) + if err == nil { + local, ok := chosen.(cellstore.Engine) + if !ok || local.Workspace != "local" || used != nil { + err = fmt.Errorf("store = %T, drive present = %v; want the local engine and no drive", chosen, used != nil) + } + } + store <- err + }() + <-entered + <-entered + // Both waiters must be parked on the startup boundary, not merely + // scheduled: the goroutine dump names each one blocked inside its wait. + waitParkedIn(t, "(*liveDrive).Gate") + waitParkedIn(t, "lateStore.readyStore") + select { + case err := <-gate: + t.Fatalf("parked gate returned %v before journal ownership was decided", err) + case err := <-store: + t.Fatalf("parked store returned %v before journal ownership was decided", err) + default: + } + return func() { + t.Helper() + for _, waiter := range []<-chan error{gate, store} { + select { + case err := <-waiter: + if err != nil { + t.Fatalf("resolved parked waiter = %v; want nil", err) + } + case <-ctx.Done(): + t.Fatalf("journal ownership was decided but its parked waiter did not return") + } + } + } + } + + t.Run("a successful journal open starts once and releases both waiters", func(t *testing.T) { + book, d, calls, claim := fixture(t) + ready := d.pending() + if ready == nil { + t.Fatalf("a parked drive must already have its startup channel") + } + finish := waiters(t, d) + d.try() + if got := calls.Load(); got != 0 || d.pending() != ready { + t.Fatalf("parked try called start %d times or replaced ready; want 0 and the original channel", got) + } + claim(true) + claim(false) + claim(true) + finish() + if got := calls.Load(); got != 1 { + t.Fatalf("successful claim called start %d times; want 1", got) + } + if book.renewalsOf("chat") != d { + t.Fatalf("the successful journal opener lost its drive") + } + }) + + t.Run("every failed journal open abandons without starting", func(t *testing.T) { + book, d, calls, first := fixture(t) + second := claimAgain(t, book, d) + finish := waiters(t, d) + first(false) + first(true) + if d.pending() == nil { + t.Fatalf("one failed opener abandoned a drive another opener still claims") + } + second(false) + second(true) + finish() + if got := calls.Load(); got != 0 { + t.Fatalf("failed journal opens called start %d times; want 0", got) + } + if book.renewalsOf("chat") != nil { + t.Fatalf("the book retained the abandoned drive") + } + if _, found := workAdmissions.Load(d); found { + t.Fatalf("the abandoned drive retained work admission state") + } + }) + + for _, operation := range []string{"Close", "closeAll", "take", "closeCell", "takeCell"} { + t.Run(operation+" abandons immediately and a later success cannot start", func(t *testing.T) { + book, d, calls, claim := fixture(t) + finish := waiters(t, d) + ctx, cancel := context.WithCancel(t.Context()) + cancel() + done := make(chan error, 1) + go func() { + switch operation { + case "Close": + done <- d.Close(ctx) + case "closeAll": + book.closeAll() + done <- nil + case "take": + book.take() + done <- nil + case "closeCell": + done <- book.closeCell("chat") + case "takeCell": + book.takeCell("chat") + done <- nil + } + }() + select { + case err := <-done: + if err != nil { + t.Fatalf("closing parked drive = %v; want nil even with a canceled context", err) + } + case <-time.After(driveCloseWithin / 3): + // A close that waited for the parked startup would take the + // whole driveCloseWithin; a third of it is generous under load. + t.Fatalf("closing a parked drive waited for startup or driveCloseWithin") + } + claim(true) + d.try() + finish() + if got := calls.Load(); got != 0 { + t.Fatalf("claim after close called start %d times; want 0", got) + } + }) + } + + for _, successFirst := range []bool{false, true} { + t.Run(fmt.Sprintf("two openers keep one drive when success comes first %v", successFirst), func(t *testing.T) { + book, d, calls, first := fixture(t) + second := claimAgain(t, book, d) + first(successFirst) + second(!successFirst) + if err := d.Gate(); err != nil { + t.Fatalf("the successful opener's gate = %v; want nil", err) + } + if got := calls.Load(); got != 1 { + t.Fatalf("two openers called start %d times; want 1", got) + } + if book.renewalsOf("chat") != d { + t.Fatalf("a failed opener removed the successful opener's drive") + } + if _, found := workAdmissions.Load(d); !found { + t.Fatalf("a failed opener removed the successful opener's work admission") + } + }) + } + + t.Run("a claim on startup already begun changes nothing", func(t *testing.T) { + book, d, calls, first := fixture(t) + first(true) + if err := d.Gate(); err != nil { + t.Fatalf("initial gate = %v; want nil", err) + } + claimAgain(t, book, d)(false) + claimAgain(t, book, d)(true) + d.mu.Lock() + closed := d.closed + d.mu.Unlock() + if got := calls.Load(); got != 1 || closed || book.renewalsOf("chat") != d { + t.Fatalf("claim on begun drive: starts = %d, closed = %v, kept = %v; want 1, false, true", got, closed, book.renewalsOf("chat") == d) + } + }) +} diff --git a/cmd/codeaf/syncdrive.go b/cmd/codeaf/syncdrive.go index c7b7abf989..f5b2d2df07 100644 --- a/cmd/codeaf/syncdrive.go +++ b/cmd/codeaf/syncdrive.go @@ -26,8 +26,9 @@ var syncDrives = &driveBook{} // same process (a resume, a switch back) goes on with the drive it had instead // of taking a second lease. type driveBook struct { - mu sync.Mutex - drive map[string]*liveDrive + mu sync.Mutex + drive map[string]*liveDrive + claims map[*liveDrive]int // notices is installed by an app assembly before any drive starts. Nil // retains the terminal's desk and its marking first-run behaviour. notices *driveNotices @@ -47,17 +48,25 @@ type driveNotices struct { // moment one is made (see liveDrive). A nil book answers nil, which is how a // door that never syncs asks. func (b *driveBook) driveOf(c cell.Cell, engine cellstore.Engine, report func(error)) *liveDrive { + d, claim := b.pendingDriveOf(c, engine, report) + claim(true) + return d +} + +// pendingDriveOf reserves a drive until one opener owns the journal. +func (b *driveBook) pendingDriveOf(c cell.Cell, engine cellstore.Engine, report func(error)) (*liveDrive, func(bool)) { + noop := func(bool) {} if b == nil { - return nil + return nil, noop } if syncsetup.Resolve(home.Dir()).Off || !cell.Available() { - return nil + return nil, noop } b.mu.Lock() defer b.mu.Unlock() if d := b.drive[c.ID]; d != nil { if !d.superseded() { - return d + return d, b.claim(d, c.ID) } // Settle the displaced writer before the replacement reads its branch map // or acquires a lease, so late uploads cannot change the new drive's target. @@ -66,7 +75,7 @@ func (b *driveBook) driveOf(c cell.Cell, engine cellstore.Engine, report func(er cancel() if err != nil { report(err) - return d + return d, b.claim(d, c.ID) } } // A drive side that no longer holds the chat's lease (another machine took it, or it was moved @@ -84,7 +93,36 @@ func (b *driveBook) driveOf(c cell.Cell, engine cellstore.Engine, report func(er } b.drive[c.ID] = d } - return d + return d, b.claim(d, c.ID) +} + +// claim is made under the book lock; no drive path takes that lock in return. +func (b *driveBook) claim(d *liveDrive, id string) func(bool) { + if !d.isParked() { + return func(bool) {} + } + if b.claims == nil { + b.claims = make(map[*liveDrive]int) + } + b.claims[d]++ + var once sync.Once + return func(opened bool) { + once.Do(func() { + b.mu.Lock() + defer b.mu.Unlock() + b.claims[d]-- + if opened { + d.begin() + } + if b.claims[d] <= 0 { + delete(b.claims, d) + if d.abandon() && b.drive[id] == d { + delete(b.drive, id) + workAdmissions.Delete(d) + } + } + }) + } } // startDrive opens the drive side of c. The error it answers is already @@ -193,6 +231,7 @@ func (b *driveBook) take() map[string]*liveDrive { defer b.mu.Unlock() drives := b.drive for _, d := range drives { + d.abandon() workAdmissions.Delete(d) } b.drive = nil @@ -254,6 +293,9 @@ func (b *driveBook) takeCell(id string) *liveDrive { b.mu.Lock() defer b.mu.Unlock() d := b.drive[id] + if d != nil { + d.abandon() + } workAdmissions.Delete(d) delete(b.drive, id) return d diff --git a/cmd/codeaf/syncdrive_admission.go b/cmd/codeaf/syncdrive_admission.go index 5f88b200b5..0d389d3b0d 100644 --- a/cmd/codeaf/syncdrive_admission.go +++ b/cmd/codeaf/syncdrive_admission.go @@ -39,7 +39,7 @@ type workAdmission struct { func newAdmissionDrive(start func(func()) (*syncsetup.Drive, error)) *liveDrive { check := &workAdmission{} - d := newLiveDrive(func(changed func()) (*syncsetup.Drive, error) { + d := newPendingLiveDrive(func(changed func()) (*syncsetup.Drive, error) { drive, err := start(changed) if err == nil && drive != nil && drive.Gate() == nil { check.mu.Lock() diff --git a/cmd/codeaf/syncdrive_live.go b/cmd/codeaf/syncdrive_live.go index 21246e594e..b5faba5e9f 100644 --- a/cmd/codeaf/syncdrive_live.go +++ b/cmd/codeaf/syncdrive_live.go @@ -27,6 +27,7 @@ type liveDrive struct { mu sync.Mutex drive *syncsetup.Drive + parked bool // Startup awaits a journal owner. waiting bool // sync is on and only an identity, or the end of solo, is missing ready chan struct{} attempted bool @@ -40,9 +41,15 @@ type liveDrive struct { // newLiveDrive starts the drive side behind the first paint. Local startup // screening stays off the surface loop; saving and new tools wait for ownership. func newLiveDrive(start func(func()) (*syncsetup.Drive, error)) *liveDrive { - l := &liveDrive{changes: make(chan struct{}, 1)} + l := newPendingLiveDrive(start) + l.begin() + return l +} + +// newPendingLiveDrive reserves the startup boundary before journal ownership is known. +func newPendingLiveDrive(start func(func()) (*syncsetup.Drive, error)) *liveDrive { + l := &liveDrive{changes: make(chan struct{}, 1), parked: true, ready: make(chan struct{})} l.start = func() (*syncsetup.Drive, error) { return start(l.renewalChanged) } - l.try() return l } @@ -52,11 +59,52 @@ func newLiveDrive(start func(func()) (*syncsetup.Drive, error)) *liveDrive { func (l *liveDrive) try() { l.mu.Lock() defer l.mu.Unlock() - if l.closed || l.drive != nil || l.ready != nil || (l.attempted && !l.waiting) || (l.waiting && !syncsetup.MayTalk(home.Dir())) { + if l.parked || l.closed || l.drive != nil || l.ready != nil || (l.attempted && !l.waiting) || (l.waiting && !syncsetup.MayTalk(home.Dir())) { return } l.ready = make(chan struct{}) - ready := l.ready + l.launch(l.ready) +} + +// begin spends the reserved startup exactly once after the journal opens. +func (l *liveDrive) begin() { + l.mu.Lock() + defer l.mu.Unlock() + if !l.parked { + return + } + l.parked = false + l.launch(l.ready) +} + +// isParked says the startup is still waiting for a journal owner, so an opener +// of the same chat in this process shares that wait rather than starting it. +func (l *liveDrive) isParked() bool { + l.mu.Lock() + defer l.mu.Unlock() + return l.parked +} + +// abandon releases parked waiters without ever opening the drive side. +func (l *liveDrive) abandon() bool { + l.mu.Lock() + defer l.mu.Unlock() + return l.abandonLocked() +} + +func (l *liveDrive) abandonLocked() bool { + if !l.parked { + return false + } + l.parked, l.closed = false, true + close(l.ready) + l.ready = nil + return true +} + +// launch shares startup and panic cleanup between immediate and reserved starts. +// The caller holds mu, so closing cannot slip between reserving and launching. +func (l *liveDrive) launch(ready chan struct{}) { guard.Go("chat sync startup", func() { var d *syncsetup.Drive var err error @@ -207,6 +255,7 @@ func (l *liveDrive) Close(ctx context.Context) error { ready := func() <-chan struct{} { l.mu.Lock() defer l.mu.Unlock() + l.abandonLocked() l.closed = true if l.retry != nil { l.retry.Stop() diff --git a/docs/changes/unreleased/2046-held-chat-drive-waits-for-its-journal.md b/docs/changes/unreleased/2046-held-chat-drive-waits-for-its-journal.md new file mode 100644 index 0000000000..ea23b855ae --- /dev/null +++ b/docs/changes/unreleased/2046-held-chat-drive-waits-for-its-journal.md @@ -0,0 +1,16 @@ +--- +kind: fixed +title: Opening a held chat leaves the other window's sync lease alone +pr: 2046 +surface: [chat, engine] +invalidates: + - "Opening a chat used to start its sync drive before checking the journal lock. A process refused the journal no longer starts, renews, releases or forgets that chat's sync hold." + - "The fresh chat offered beside a held chat used to inherit the held chat's executor and seals. It now opens on its own seat with its own sync drive." +--- + +A terminal opened beside a chat held by the desktop app or another process can +still offer to move the conversation here. Leaving that question, quitting the +terminal, or stopping its host keeps the original window's chat sendable. +The fresh chat offered beside it runs commands and saves turns in its own cell. +This does not change the overlap when a drive already started in this process +outlives its released journal and a later open is refused.