Add deadlines to git network ops and lock merges per project

2026-10-16 14:20:16 -07:00
parent 8c7c3bee2e
commit 6ee2421d97
5 changed files with 98 additions and 47 deletions
@@ -4,8 +4,6 @@
  `CheckResponse` reads open state outside the write transaction and `Respond` updates by id alone (`internal/agents/respond.go:26–34`, `:93`; `internal/store/inbox.go:140`), so two concurrent answers both land, or a withdrawal is overwritten. Condition the update on `state='open'`, check RowsAffected, roll back the batch on conflict and return 409.
 - A Codex RPC can deadlock the reader it waits on
  `drive` calls `Steer` synchronously while `readLoop` pushes notifications into a 256-entry channel (`internal/codex/session.go:186`, `:440–476`, `:598`); if it fills before the response is read, each waits on the other. Route RPC responses separately from notification consumption and give RPCs bounded deadlines, without replacing backpressure with an unbounded queue.
- Git network operations have no deadline and run under a cross-project lock
-  `exec.Command` has no context (`internal/git/git.go:348–364`) and Merge holds the shared `mergeMu` through fetch (`internal/agents/mergein.go:68`, `:101`), so one stalled remote blocks merges in every project. Add contextual deadlines and non-interactive git/SSH, and serialize the network phase per project while keeping same-project ordering.
 - A malformed file image URL breaks transcript rendering
  `decodeURI` throws on e.g. `file:///tmp/100%.png` (`web/src/markdown.ts:347`), and `renderAll` appends events before mounting the composer (`web/src/task.ts:608–609`), so the rest of the transcript and the composer don't render. Catch the decode error and render the plain fallback link.
 - Stale reload and focus responses overwrite newer state
@@ -75,7 +75,9 @@ Before starting the turn, Merge runs `git fetch origin` (every configured refspe

 Each squash commit gets a git note under `refs/notes/byre`, the durable record of the task (`mergeNote` in `internal/agents/notes.go`): a Markdown heading with the merge title; the task's id, title, creator, merging user, TODO item and fork source; a pointer to the full transcripts as the task's events in byre (`GET /api/events?task_id=<id>`), which the note doesn't copy; the task summary, when the agent wrote one; every handoff its sessions wrote on compaction (the `result` of each `turn.ended` with `handoff: true`); each review's state, error, `change` and `summary`, by snapshot; and each `answered` question with its context and the responses to it. Plans and eval results aren't recorded yet, since byre has neither. `git.NoteCommit` builds the notes commit on top of the checkout's current `refs/notes/byre` in a temporary index, authored as the merging user, without moving the ref, so a refused push leaves the checkout's notes as they were. Notes don't fetch by default, so registering a project (`POST /api/projects`), and every byre startup for each existing project, adds the refspec `refs/notes/byre*:refs/notes/byre*` to the checkout's `remote.origin.fetch` if it isn't there (`git.ConfigureNotesFetch`); it's a pattern because git refuses a fetch whose exact refspec names a ref the remote doesn't have yet, as it doesn't before the first note is pushed. Another clone adds the same refspec with `git config --add remote.origin.fetch 'refs/notes/byre*:refs/notes/byre*'`. `git log --notes=byre` shows the notes, and `git notes --ref=byre show <commit>` one commit's.

-One merge runs per project at a time. The Manager keeps the task holding each project's merge in memory (`mergeHolders` in `internal/agents/manager.go`), from just before the fetch until the merge closes the task, fails, or is abandoned as above, or the task is discarded; on startup it is rebuilt from the active tasks with `merging_by` set. Merge on another task of that project while one is held doesn't fetch or start a turn: it records the request in `tasks.merge_queued_by` and `merge_queued_at` (`merge_queued_at` on the Task), emits `task.merge_queued` (a transcript note), and returns the task with 202. A second Merge on a held or merging task is a 409. Whenever a project's merge is released, any task turn in it ends, and on startup, byre starts the oldest held merge in the project whose task session is not `running`: it clears the request (emitting `task.merge_unqueued`, not drawn) and goes through the same fetch and turn as a direct Merge by the user who asked, so it merges in what the earlier merge pushed. If that start fails (the fetch, or the user has no git identity), byre files the error in the task's inbox and tries the next held merge. The Merge button reads **Merge held**, disabled, while the task has `merge_queued_at`; discarding the task drops the request. `sqlite3 byre.db "SELECT id, merge_queued_by, merge_queued_at FROM tasks WHERE merge_queued_by != ''"` lists held merges.
+git's network commands (`git fetch` and `git push`, through `runNetwork` in `internal/git/git.go`) are bounded by `networkTimeout` and by the caller's context (a Merge request's, so a dropped request stops its fetch), and run non-interactively: `GIT_TERMINAL_PROMPT=0`, and `GIT_SSH_COMMAND` set to `ssh` with `BatchMode=yes` and a connect timeout (`networkEnv`), which replaces any `core.sshCommand` the checkout configures. A remote needing a password or an unknown host key fails at once instead of waiting for input. Every git command runs in its own process group, and cancelling one kills the whole group, so an `ssh` or remote helper it started can't outlive the deadline and finish a fetch or push after byre has moved on. A timed-out fetch fails the Merge like any other fetch error; a timed-out push is ambiguous and goes through the landed check above.
+
+One merge runs per project at a time. The Manager keeps, per project, a lock and the task holding that project's merge in memory (`projectMerge` in `internal/agents/mergein.go`, in `Manager.merges`), from just before the fetch until the merge closes the task, fails, or is abandoned as above, or the task is discarded; on startup it is rebuilt from the active tasks with `merging_by` set. Starting a merge holds only its project's lock through the fetch, so a stalled remote delays merges in that project alone, while Merge requests and held-merge starts in the same project wait behind it and then see its holder. Merge on another task of that project while one is held doesn't fetch or start a turn: it records the request in `tasks.merge_queued_by` and `merge_queued_at` (`merge_queued_at` on the Task), emits `task.merge_queued` (a transcript note), and returns the task with 202. A second Merge on a held or merging task is a 409. Whenever a project's merge is released, any task turn in it ends, and on startup, byre starts the oldest held merge in the project whose task session is not `running`: it clears the request (emitting `task.merge_unqueued`, not drawn) and goes through the same fetch and turn as a direct Merge by the user who asked, so it merges in what the earlier merge pushed. If that start fails (the fetch, or the user has no git identity), byre files the error in the task's inbox and tries the next held merge. The Merge button reads **Merge held**, disabled, while the task has `merge_queued_at`; discarding the task drops the request. `sqlite3 byre.db "SELECT id, merge_queued_by, merge_queued_at FROM tasks WHERE merge_queued_by != ''"` lists held merges.

 ## Subtasks

@@ -59,8 +59,8 @@ type Manager struct {
 	snapMu sync.Mutex
 	snaps  map[string]*snapState

-	mergeMu      sync.Mutex
-	mergeHolders map[string]string
+	mergeMu sync.Mutex
+	merges  map[string]*projectMerge

 	reviewMu sync.Mutex

@@ -130,7 +130,7 @@ func New(st *store.Store, worktreeRoot, leadRoot, codexHome string) *Manager {
 		finishing:    map[string]bool{},
 		signals:      make(chan []api.Event, 1024),
 		snaps:        map[string]*snapState{},
-		mergeHolders: map[string]string{},
+		merges:       map[string]*projectMerge{},
 		bad:          newBadLines(),
 		watchStop:    make(chan struct{}),
 	}
@@ -8,6 +8,7 @@ import (
 	"log/slog"
 	"slices"
 	"strings"
+	"sync"
 	"unicode/utf8"

 	"g.fc.run/misc/byre/internal/api"
@@ -65,10 +66,11 @@ func (m *Manager) Merge(ctx context.Context, task api.Task, project api.Project,
 		return api.Task{}, err
 	}

-	m.mergeMu.Lock()
-	defer m.mergeMu.Unlock()
+	pm := m.projectMerge(project.Id)
+	pm.mu.Lock()
+	defer pm.mu.Unlock()

-	if m.mergeHolders[project.Id] == task.Id {
+	if pm.holder == task.Id {
 		return api.Task{}, fmt.Errorf("%w: task %s is already merging", ErrConflict, task.Id)
 	}
 	if title != task.MergeTitle {
@@ -79,9 +81,9 @@ func (m *Manager) Merge(ctx context.Context, task api.Task, project api.Project,
 		task.MergeTitle = title
 	}

-	switch m.mergeHolders[project.Id] {
+	switch pm.holder {
 	case "":
-		err := m.startMerge(ctx, task, project, user)
+		err := m.startMerge(ctx, pm, task, project, user)
 		if err != nil {
 			return api.Task{}, err
 		}
@@ -97,8 +99,24 @@ func (m *Manager) Merge(ctx context.Context, task api.Task, project api.Project,
 	return m.store.GetTask(ctx, task.Id)
 }

-func (m *Manager) startMerge(ctx context.Context, task api.Task, project api.Project, user api.User) error {
-	err := git.FetchFastForward(project.RepoPath, project.DefaultBranch)
+type projectMerge struct {
+	mu     sync.Mutex
+	holder string
+}
+
+func (m *Manager) projectMerge(projectID string) *projectMerge {
+	m.mergeMu.Lock()
+	defer m.mergeMu.Unlock()
+	pm := m.merges[projectID]
+	if pm == nil {
+		pm = &projectMerge{}
+		m.merges[projectID] = pm
+	}
+	return pm
+}
+
+func (m *Manager) startMerge(ctx context.Context, pm *projectMerge, task api.Task, project api.Project, user api.User) error {
+	err := git.FetchFastForward(ctx, project.RepoPath, project.DefaultBranch)
 	if err != nil {
 		return err
 	}
@@ -110,13 +128,13 @@ func (m *Manager) startMerge(ctx context.Context, task api.Task, project api.Pro
 	if err != nil {
 		return err
 	}
-	m.mergeHolders[project.Id] = task.Id
+	pm.holder = task.Id

 	prompt := mergeInPrompt(project.DefaultBranch) + fmt.Sprintf(" When this turn ends, byre squashes this branch onto %s and pushes it; if you stop, call stop_merge first so it doesn't. ", project.DefaultBranch) + summaryRequest
 	_, err = m.sendMergeIn(ctx, task, user, prompt)
 	if err != nil {
 		m.takeMerge(task.Id)
-		delete(m.mergeHolders, project.Id)
+		pm.holder = ""
 	}
 	return err
 }
@@ -127,10 +145,11 @@ func (m *Manager) recoverMerges(ctx context.Context) error {
 		return err
 	}

-	m.mergeMu.Lock()
-	defer m.mergeMu.Unlock()
 	for _, t := range merging {
-		m.mergeHolders[t.ProjectId] = t.Id
+		pm := m.projectMerge(t.ProjectId)
+		pm.mu.Lock()
+		pm.holder = t.Id
+		pm.mu.Unlock()
 	}
 	return nil
 }
@@ -177,7 +196,7 @@ func (m *Manager) resumeFinalizing(ctx context.Context, f store.FinalizingMerge)
 		return errors.New("nothing was pushed. Merge again to retry")
 	}

-	landed, err := pushLanded(project, f.Commit)
+	landed, err := pushLanded(ctx, project, f.Commit)
 	if err != nil {
 		return err
 	}
@@ -187,9 +206,9 @@ func (m *Manager) resumeFinalizing(ctx context.Context, f store.FinalizingMerge)
 	return m.closePushed(ctx, f.Task, project, f.Commit, f.Notes, f.By)
 }

-func pushLanded(project api.Project, commit string) (bool, error) {
+func pushLanded(ctx context.Context, project api.Project, commit string) (bool, error) {
 	target := git.Remote + "/" + project.DefaultBranch
-	err := git.Fetch(project.RepoPath)
+	err := git.Fetch(ctx, project.RepoPath)
 	if err != nil {
 		return false, &unfinishedMergeError{fmt.Errorf("checking whether %s reached %s failed: %w", commit, target, err)}
 	}
@@ -231,22 +250,24 @@ func (m *Manager) abandonMerge(taskID, projectID string) {
 }

 func (m *Manager) releaseMerge(projectID, taskID string) {
-	m.mergeMu.Lock()
-	defer m.mergeMu.Unlock()
-	if m.mergeHolders[projectID] == taskID {
-		delete(m.mergeHolders, projectID)
+	pm := m.projectMerge(projectID)
+	pm.mu.Lock()
+	defer pm.mu.Unlock()
+	if pm.holder == taskID {
+		pm.holder = ""
 	}
-	m.startNextMerge(projectID)
+	m.startNextMerge(pm, projectID)
 }

 func (m *Manager) nextMerge(projectID string) {
-	m.mergeMu.Lock()
-	defer m.mergeMu.Unlock()
-	m.startNextMerge(projectID)
+	pm := m.projectMerge(projectID)
+	pm.mu.Lock()
+	defer pm.mu.Unlock()
+	m.startNextMerge(pm, projectID)
 }

-func (m *Manager) startNextMerge(projectID string) {
-	if m.mergeHolders[projectID] != "" {
+func (m *Manager) startNextMerge(pm *projectMerge, projectID string) {
+	if pm.holder != "" {
 		return
 	}

@@ -278,7 +299,7 @@ func (m *Manager) startNextMerge(projectID string) {
 			continue
 		}

-		err = m.startHeldMerge(ctx, task, login)
+		err = m.startHeldMerge(ctx, pm, task, login)
 		if err == nil {
 			slog.Info("started held merge", "task", task.Id, "by", login)
 			return
@@ -288,7 +309,7 @@ func (m *Manager) startNextMerge(projectID string) {
 	}
 }

-func (m *Manager) startHeldMerge(ctx context.Context, task api.Task, login string) error {
+func (m *Manager) startHeldMerge(ctx context.Context, pm *projectMerge, task api.Task, login string) error {
 	project, err := m.store.GetProject(ctx, task.ProjectId)
 	if err != nil {
 		return err
@@ -300,7 +321,7 @@ func (m *Manager) startHeldMerge(ctx context.Context, task api.Task, login strin
 	if user.GitName == "" || user.GitEmail == "" {
 		return ErrNoGitIdentity
 	}
-	return m.startMerge(ctx, task, project, user)
+	return m.startMerge(ctx, pm, task, project, user)
 }

 func (m *Manager) fileTaskError(ctx context.Context, task api.Task, sessions []api.AgentSession, msg string) {
@@ -396,12 +417,12 @@ func (m *Manager) squashAndClose(ctx context.Context, taskID, login string) erro
 	if err != nil {
 		return err
 	}
-	err = git.Push(project.RepoPath, commit, project.DefaultBranch, notes)
+	err = git.Push(ctx, project.RepoPath, commit, project.DefaultBranch, notes)
 	if errors.Is(err, git.ErrRemoteMoved) {
 		return err
 	}
 	if err != nil {
-		landed, lerr := pushLanded(project, commit)
+		landed, lerr := pushLanded(ctx, project, commit)
 		if lerr != nil {
 			return &unfinishedMergeError{fmt.Errorf("%w, and %w", err, errors.Unwrap(lerr))}
 		}
@@ -1,6 +1,7 @@
 package git

 import (
+	"context"
 	"errors"
 	"fmt"
 	"io"
@@ -10,6 +11,8 @@ import (
 	"slices"
 	"strconv"
 	"strings"
+	"syscall"
+	"time"
 )

 type RepoInfo struct {
@@ -228,10 +231,17 @@ func SquashCommit(repo, worktree, target, message string, env []string) (string,

 const Remote = "origin"

+const networkTimeout = 2 * time.Minute
+
+var networkEnv = []string{
+	"GIT_TERMINAL_PROMPT=0",
+	"GIT_SSH_COMMAND=ssh -o BatchMode=yes -o ConnectTimeout=30",
+}
+
 var ErrRemoteMoved = errors.New("the remote's branch has moved")

-func Push(repo, commit, branch, notes string) error {
-	out, stderr, code, err := runCode(repo, nil, "push", "--atomic", "--porcelain", Remote, commit+":refs/heads/"+branch, notes+":"+NotesRef)
+func Push(ctx context.Context, repo, commit, branch, notes string) error {
+	out, stderr, code, err := runNetwork(ctx, repo, "push", "--atomic", "--porcelain", Remote, commit+":refs/heads/"+branch, notes+":"+NotesRef)
 	if err != nil {
 		return err
 	}
@@ -251,13 +261,19 @@ func FastForward(repo, rev string) error {
 	return err
 }

-func Fetch(repo string) error {
-	_, err := run(repo, "fetch", Remote)
-	return err
+func Fetch(ctx context.Context, repo string) error {
+	_, stderr, code, err := runNetwork(ctx, repo, "fetch", Remote)
+	if err != nil {
+		return err
+	}
+	if code != 0 {
+		return fmt.Errorf("git fetch %s: exit status %d: %s", Remote, code, stderr)
+	}
+	return nil
 }

-func FetchFastForward(repo, branch string) error {
-	err := Fetch(repo)
+func FetchFastForward(ctx context.Context, repo, branch string) error {
+	err := Fetch(ctx, repo)
 	if err != nil {
 		return err
 	}
@@ -336,7 +352,7 @@ func runEnv(dir string, env []string, args ...string) (string, error) {
 }

 func runInput(dir string, env []string, input string, args ...string) (string, error) {
-	out, stderr, code, err := runCodeInput(dir, env, strings.NewReader(input), args...)
+	out, stderr, code, err := runCodeInput(context.Background(), dir, env, strings.NewReader(input), args...)
 	if err != nil {
 		return "", err
 	}
@@ -347,17 +363,31 @@ func runInput(dir string, env []string, input string, args ...string) (string, e
 }

 func runCode(dir string, env []string, args ...string) (string, string, int, error) {
-	return runCodeInput(dir, env, nil, args...)
+	return runCodeInput(context.Background(), dir, env, nil, args...)
 }

-func runCodeInput(dir string, env []string, stdin io.Reader, args ...string) (string, string, int, error) {
-	cmd := exec.Command("git", args...)
+func runNetwork(ctx context.Context, dir string, args ...string) (string, string, int, error) {
+	ctx, cancel := context.WithTimeout(ctx, networkTimeout)
+	defer cancel()
+	return runCodeInput(ctx, dir, networkEnv, nil, args...)
+}
+
+func runCodeInput(ctx context.Context, dir string, env []string, stdin io.Reader, args ...string) (string, string, int, error) {
+	cmd := exec.CommandContext(ctx, "git", args...)
 	cmd.Dir = dir
 	cmd.Env = append(os.Environ(), env...)
 	cmd.Stdin = stdin
+	cmd.SysProcAttr = &syscall.SysProcAttr{Setpgid: true}
+	cmd.Cancel = func() error {
+		return syscall.Kill(-cmd.Process.Pid, syscall.SIGKILL)
+	}
+	cmd.WaitDelay = 5 * time.Second
 	stderr := &strings.Builder{}
 	cmd.Stderr = stderr
 	out, err := cmd.Output()
+	if ctx.Err() != nil {
+		return "", "", 0, fmt.Errorf("git %s: %w: %s", strings.Join(args, " "), context.Cause(ctx), strings.TrimSpace(stderr.String()))
+	}
 	exit := &exec.ExitError{}
 	if errors.As(err, &exit) {
 		return strings.TrimSpace(string(out)), strings.TrimSpace(stderr.String()), exit.ExitCode(), nil