Run the merge queue on a state-driven coordinator
This commit is contained in:
Ian Gulliver
+101
-280
@@ -8,7 +8,6 @@ import (
|
||||
"log/slog"
|
||||
"os"
|
||||
"strings"
|
||||
"sync"
|
||||
"unicode/utf8"
|
||||
|
||||
"g.fc.run/ai/byre/internal/api"
|
||||
@@ -87,13 +86,6 @@ func (m *Manager) Merge(ctx context.Context, task api.Task, project api.Project,
|
||||
return api.Task{}, err
|
||||
}
|
||||
|
||||
pm := m.projectMerge(project.Id)
|
||||
pm.mu.Lock()
|
||||
defer pm.mu.Unlock()
|
||||
|
||||
if pm.holder == task.Id {
|
||||
return api.Task{}, fmt.Errorf("%w: task %s is already merging", ErrConflict, task.Id)
|
||||
}
|
||||
if title != task.MergeTitle {
|
||||
err = m.storeMergeTitle(ctx, task.Id, title, user.Login)
|
||||
if err != nil {
|
||||
@@ -102,111 +94,41 @@ func (m *Manager) Merge(ctx context.Context, task api.Task, project api.Project,
|
||||
task.MergeTitle = title
|
||||
}
|
||||
|
||||
switch pm.holder {
|
||||
case "":
|
||||
err := m.startMerge(ctx, pm, task, project, user)
|
||||
if err != nil {
|
||||
return api.Task{}, err
|
||||
}
|
||||
default:
|
||||
err := m.store.QueueMerge(ctx, task.Id, user.Login)
|
||||
if errors.Is(err, store.ErrNotFound) {
|
||||
return api.Task{}, fmt.Errorf("%w: task %s is closed, being discarded or already waiting to merge", ErrConflict, task.Id)
|
||||
}
|
||||
if err != nil {
|
||||
return api.Task{}, err
|
||||
}
|
||||
held, err := m.store.RequestMerge(ctx, task.Id, user.Login)
|
||||
switch {
|
||||
case errors.Is(err, store.ErrNotFound):
|
||||
return api.Task{}, fmt.Errorf("%w: task %s is closed, being discarded or already waiting to merge", ErrConflict, task.Id)
|
||||
case errors.Is(err, store.ErrMerging):
|
||||
return api.Task{}, fmt.Errorf("%w: task %s is already merging", ErrConflict, task.Id)
|
||||
case err != nil:
|
||||
return api.Task{}, err
|
||||
}
|
||||
if held {
|
||||
m.headStartMerge(ctx, task, user, project.DefaultBranch)
|
||||
}
|
||||
m.clearMergeAfterReview(ctx, task.Id)
|
||||
return m.store.GetTask(ctx, task.Id)
|
||||
}
|
||||
|
||||
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 := m.store.SetMergingBy(ctx, task.Id, user.Login)
|
||||
if errors.Is(err, store.ErrNotFound) {
|
||||
return fmt.Errorf("%w: task %s is being discarded", ErrConflict, task.Id)
|
||||
}
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
pm.holder = task.Id
|
||||
|
||||
err = m.beginMergeTurn(ctx, task, project, user)
|
||||
if err != nil {
|
||||
m.takeMerge(task.Id)
|
||||
pm.holder = ""
|
||||
}
|
||||
return err
|
||||
}
|
||||
|
||||
func (m *Manager) beginMergeTurn(ctx context.Context, task api.Task, project api.Project, user api.User) error {
|
||||
sess, err := m.mergeSession(ctx, task, user)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
go m.launchMerge(sess, task, project, mergeTurnPrompt(project))
|
||||
return nil
|
||||
}
|
||||
|
||||
func mergeTurnPrompt(project api.Project) string {
|
||||
return 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
|
||||
}
|
||||
|
||||
func (m *Manager) advanceMergeClaims(ctx context.Context) {
|
||||
ids, err := m.store.MergeClaims(ctx)
|
||||
if err != nil {
|
||||
slog.Error("list merge claims", "err", err)
|
||||
return
|
||||
}
|
||||
for _, id := range ids {
|
||||
m.advanceMergeClaim(id)
|
||||
}
|
||||
}
|
||||
|
||||
func (m *Manager) advanceMergeClaim(taskID string) {
|
||||
ctx := context.Background()
|
||||
err := m.beginClaimedMerge(ctx, taskID)
|
||||
func (m *Manager) startClaimedMerge(ctx context.Context, c store.ClaimedMerge) {
|
||||
err := m.beginClaimedMerge(ctx, c)
|
||||
if err == nil {
|
||||
return
|
||||
}
|
||||
slog.Error("start claimed merge", "task", taskID, "err", err)
|
||||
task, terr := m.store.GetTask(ctx, taskID)
|
||||
slog.Error("start claimed merge", "task", c.TaskID, "err", err)
|
||||
_, terr := m.store.TakeMergingBy(ctx, c.TaskID)
|
||||
if terr != nil {
|
||||
slog.Error("load task for claimed merge", "task", taskID, "err", terr)
|
||||
return
|
||||
slog.Error("drop claimed merge", "task", c.TaskID, "err", terr)
|
||||
}
|
||||
m.abandonMerge(taskID, task.ProjectId)
|
||||
sessions, serr := m.store.ListSessions(ctx, taskID)
|
||||
if serr != nil {
|
||||
slog.Error("list sessions for claimed merge", "task", taskID, "err", serr)
|
||||
return
|
||||
}
|
||||
m.fileTaskError(ctx, task, sessions, "Starting the held merge failed: "+err.Error())
|
||||
m.fileError(ctx, c.TaskID, "Starting the merge failed: "+err.Error())
|
||||
}
|
||||
|
||||
func (m *Manager) beginClaimedMerge(ctx context.Context, taskID string) error {
|
||||
login, err := m.store.MergeClaim(ctx, taskID)
|
||||
if err != nil || login == "" {
|
||||
return err
|
||||
}
|
||||
task, err := m.store.GetTask(ctx, taskID)
|
||||
func (m *Manager) beginClaimedMerge(ctx context.Context, c store.ClaimedMerge) error {
|
||||
task, err := m.store.GetTask(ctx, c.TaskID)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
@@ -214,7 +136,7 @@ func (m *Manager) beginClaimedMerge(ctx context.Context, taskID string) error {
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
user, err := m.store.GetUser(ctx, login)
|
||||
user, err := m.store.GetUser(ctx, c.By)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
@@ -229,7 +151,7 @@ func (m *Manager) beginClaimedMerge(ctx context.Context, taskID string) error {
|
||||
if err != nil || !reserved {
|
||||
return err
|
||||
}
|
||||
slog.Info("started claimed merge", "task", taskID, "by", login)
|
||||
slog.Info("started claimed merge", "task", task.Id, "by", c.By)
|
||||
go m.launchMerge(sess, task, project, mergeTurnPrompt(project))
|
||||
return nil
|
||||
}
|
||||
@@ -259,7 +181,6 @@ func (m *Manager) launchMerge(sess api.AgentSession, task api.Task, project api.
|
||||
}
|
||||
slog.Error("start merge turn", "task", task.Id, "err", err)
|
||||
m.failStart(ctx, sess.Id, "byre", err)
|
||||
m.abandonMerge(task.Id, project.Id)
|
||||
}
|
||||
|
||||
func (m *Manager) deliverMerge(ctx context.Context, sess api.AgentSession, task api.Task, project api.Project, prompt string) error {
|
||||
@@ -296,73 +217,6 @@ func (m *Manager) deliverMerge(ctx context.Context, sess api.AgentSession, task
|
||||
return m.send(ctx, l, prompt, nil)
|
||||
}
|
||||
|
||||
func (m *Manager) recoverMerges(ctx context.Context) error {
|
||||
merging, err := m.store.MergingTasks(ctx)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
for _, t := range merging {
|
||||
pm := m.projectMerge(t.ProjectId)
|
||||
pm.mu.Lock()
|
||||
pm.holder = t.Id
|
||||
pm.mu.Unlock()
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (m *Manager) recoverFinalizing(ctx context.Context) error {
|
||||
finalizing, err := m.store.FinalizingMerges(ctx)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
for _, f := range finalizing {
|
||||
ferr := m.resumeFinalizing(ctx, f)
|
||||
if ferr == nil {
|
||||
slog.Info("finished merge at startup", "task", f.Task.Id, "commit", f.Commit)
|
||||
continue
|
||||
}
|
||||
|
||||
slog.Error("finish merge at startup", "task", f.Task.Id, "err", ferr)
|
||||
unfinished := &unfinishedMergeError{}
|
||||
if !errors.As(ferr, &unfinished) {
|
||||
err = m.store.ClearFinalizing(ctx, f.Task.Id)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
sessions, err := m.store.ListSessions(ctx, f.Task.Id)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
m.fileTaskError(ctx, f.Task, sessions, "byre restarted while finishing this task's merge: "+ferr.Error()+".")
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (m *Manager) resumeFinalizing(ctx context.Context, f store.FinalizingMerge) error {
|
||||
project, err := m.store.GetProject(ctx, f.Task.ProjectId)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if f.Unchanged {
|
||||
return m.closeUnchanged(ctx, f.Task, project, f.By)
|
||||
}
|
||||
if f.Commit == "" {
|
||||
return errors.New("nothing was pushed. Merge again to retry")
|
||||
}
|
||||
|
||||
landed, err := pushLanded(ctx, project, f.Commit)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if !landed {
|
||||
return fmt.Errorf("its squash commit %s is not on %s/%s, so the push didn't land. Merge again to retry", f.Commit, git.Remote, project.DefaultBranch)
|
||||
}
|
||||
return m.closePushed(ctx, f.Task, project, f.Commit, f.Notes, f.By)
|
||||
}
|
||||
|
||||
func pushLanded(ctx context.Context, project api.Project, commit string) (bool, error) {
|
||||
target := git.Remote + "/" + project.DefaultBranch
|
||||
err := git.Fetch(ctx, project.RepoPath)
|
||||
@@ -388,67 +242,18 @@ func (e *unfinishedMergeError) Unwrap() error {
|
||||
return e.err
|
||||
}
|
||||
|
||||
func (m *Manager) startHeldMerges(ctx context.Context) {
|
||||
projects, err := m.store.ProjectsWithQueuedMerges(ctx)
|
||||
func (m *Manager) fileError(ctx context.Context, taskID, msg string) {
|
||||
task, err := m.store.GetTask(ctx, taskID)
|
||||
if err != nil {
|
||||
slog.Error("list projects with held merges", "err", err)
|
||||
slog.Error("load task to file error on", "task", taskID, "error", msg, "err", err)
|
||||
return
|
||||
}
|
||||
for _, p := range projects {
|
||||
m.nextMerge(p)
|
||||
}
|
||||
}
|
||||
|
||||
func (m *Manager) abandonMerge(taskID, projectID string) {
|
||||
if m.takeMerge(taskID) == "" {
|
||||
return
|
||||
}
|
||||
go m.releaseMerge(projectID, taskID)
|
||||
}
|
||||
|
||||
func (m *Manager) releaseMerge(projectID, taskID string) {
|
||||
pm := m.projectMerge(projectID)
|
||||
pm.mu.Lock()
|
||||
defer pm.mu.Unlock()
|
||||
if pm.holder == taskID {
|
||||
pm.holder = ""
|
||||
}
|
||||
m.startNextMerge(pm, projectID)
|
||||
}
|
||||
|
||||
func (m *Manager) nextMerge(projectID string) {
|
||||
pm := m.projectMerge(projectID)
|
||||
pm.mu.Lock()
|
||||
defer pm.mu.Unlock()
|
||||
m.startNextMerge(pm, projectID)
|
||||
}
|
||||
|
||||
func (m *Manager) startNextMerge(pm *projectMerge, projectID string) {
|
||||
if pm.holder != "" {
|
||||
return
|
||||
}
|
||||
|
||||
ctx := context.Background()
|
||||
held, err := m.store.QueuedMerges(ctx, projectID)
|
||||
sessions, err := m.store.ListSessions(ctx, taskID)
|
||||
if err != nil {
|
||||
slog.Error("list held merges", "project", projectID, "err", err)
|
||||
return
|
||||
}
|
||||
|
||||
for _, task := range held {
|
||||
login, err := m.store.ClaimQueuedMerge(ctx, task.Id)
|
||||
if err != nil {
|
||||
slog.Error("take held merge", "task", task.Id, "err", err)
|
||||
continue
|
||||
}
|
||||
if login == "" {
|
||||
continue
|
||||
}
|
||||
pm.holder = task.Id
|
||||
slog.Info("claimed held merge", "task", task.Id, "by", login)
|
||||
go m.advanceMergeClaim(task.Id)
|
||||
slog.Error("list sessions to file error on", "task", taskID, "error", msg, "err", err)
|
||||
return
|
||||
}
|
||||
m.fileTaskError(ctx, task, sessions, msg)
|
||||
}
|
||||
|
||||
func (m *Manager) fileTaskError(ctx context.Context, task api.Task, sessions []api.AgentSession, msg string) {
|
||||
@@ -460,85 +265,100 @@ func (m *Manager) fileTaskError(ctx context.Context, task api.Task, sessions []a
|
||||
m.recordError(&live{id: last.Id, taskID: task.Id, projectID: task.ProjectId}, msg)
|
||||
}
|
||||
|
||||
func (m *Manager) takeMerge(taskID string) string {
|
||||
login, err := m.store.TakeMergingBy(context.Background(), taskID)
|
||||
if err != nil {
|
||||
slog.Error("take merge intent", "task", taskID, "err", err)
|
||||
}
|
||||
return login
|
||||
}
|
||||
|
||||
func (m *Manager) finishMerge(l *live, login string) {
|
||||
defer m.releaseMerge(l.projectID, l.taskID)
|
||||
|
||||
func (m *Manager) finalizeMerge(f store.FinalizingMerge) {
|
||||
ctx := context.Background()
|
||||
err := m.squashAndClose(ctx, l.taskID, login)
|
||||
m.endMergeTurn(l.id, false)
|
||||
unlock := m.lockTask(f.Task.Id)
|
||||
err := m.landMerge(ctx, f)
|
||||
unlock()
|
||||
if err == nil {
|
||||
return
|
||||
}
|
||||
|
||||
slog.Error("finish merge", "task", l.taskID, "err", err)
|
||||
slog.Error("finish merge", "task", f.Task.Id, "err", err)
|
||||
unfinished := &unfinishedMergeError{}
|
||||
if errors.As(err, &unfinished) {
|
||||
m.recordError(l, "Merging the task failed: "+err.Error()+".")
|
||||
m.fileError(ctx, f.Task.Id, "Merging the task failed: "+err.Error()+".")
|
||||
return
|
||||
}
|
||||
cerr := m.store.ClearFinalizing(ctx, l.taskID)
|
||||
cerr := m.store.ClearFinalizing(ctx, f.Task.Id)
|
||||
if cerr != nil {
|
||||
slog.Error("clear merge finalizing", "task", l.taskID, "err", cerr)
|
||||
slog.Error("clear merge finalizing", "task", f.Task.Id, "err", cerr)
|
||||
}
|
||||
m.recordError(l, "Merging the task failed: "+err.Error())
|
||||
err = m.Deliver(ctx, l.id)
|
||||
m.fileError(ctx, f.Task.Id, "Merging the task failed: "+err.Error())
|
||||
m.deliverIdle(ctx, f.Task.Id)
|
||||
}
|
||||
|
||||
func (m *Manager) deliverIdle(ctx context.Context, taskID string) {
|
||||
sessions, err := m.store.ListSessions(ctx, taskID)
|
||||
if err != nil {
|
||||
slog.Error("deliver inbox responses", "session", l.id, "err", err)
|
||||
slog.Error("list sessions to deliver to", "task", taskID, "err", err)
|
||||
return
|
||||
}
|
||||
for _, s := range sessions {
|
||||
if s.Role != api.Implementer || s.State != api.AgentSessionStateIdle || m.interrupted(s.Id) {
|
||||
continue
|
||||
}
|
||||
err = m.Deliver(ctx, s.Id)
|
||||
if err != nil {
|
||||
slog.Error("deliver inbox responses", "session", s.Id, "err", err)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (m *Manager) squashAndClose(ctx context.Context, taskID, login string) error {
|
||||
task, err := m.store.GetTask(ctx, taskID)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
func (m *Manager) landMerge(ctx context.Context, f store.FinalizingMerge) error {
|
||||
task := f.Task
|
||||
project, err := m.store.GetProject(ctx, task.ProjectId)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
user, err := m.store.GetUser(ctx, login)
|
||||
if err != nil {
|
||||
return err
|
||||
if f.Unchanged {
|
||||
return m.closeUnchanged(ctx, task, project, f.By)
|
||||
}
|
||||
|
||||
coAuthors, err := m.CoAuthors(ctx, taskID, login)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
env := git.IdentityEnv(user.GitName, user.GitEmail)
|
||||
commit, err := git.SquashCommit(project.RepoPath, task.WorktreePath, project.DefaultBranch, squashMessage(task.MergeTitle, coAuthors), env, m.signer(ctx, user))
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
if commit == "" {
|
||||
err = m.store.SetFinalizingUnchanged(ctx, taskID)
|
||||
commit, notes := f.Commit, f.Notes
|
||||
if commit != "" {
|
||||
landed, err := pushLanded(ctx, project, commit)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
return m.closeUnchanged(ctx, task, project, login)
|
||||
if landed {
|
||||
return m.closePushed(ctx, task, project, commit, notes, f.By)
|
||||
}
|
||||
}
|
||||
if commit == "" {
|
||||
user, err := m.store.GetUser(ctx, f.By)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
coAuthors, err := m.CoAuthors(ctx, task.Id, f.By)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
env := git.IdentityEnv(user.GitName, user.GitEmail)
|
||||
commit, err = git.SquashCommit(project.RepoPath, task.WorktreePath, project.DefaultBranch, squashMessage(task.MergeTitle, coAuthors), env, m.signer(ctx, user))
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if commit == "" {
|
||||
err = m.store.SetFinalizingUnchanged(ctx, task.Id)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
return m.closeUnchanged(ctx, task, project, f.By)
|
||||
}
|
||||
|
||||
note, err := m.mergeNote(ctx, task, login)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
notes, err := git.NoteCommit(project.RepoPath, commit, note, env)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
err = m.store.SetFinalizingCommit(ctx, taskID, commit, notes)
|
||||
if err != nil {
|
||||
return err
|
||||
note, err := m.mergeNote(ctx, task, f.By)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
notes, err = git.NoteCommit(project.RepoPath, commit, note, env)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
err = m.store.SetFinalizingCommit(ctx, task.Id, commit, notes)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
err = git.Push(ctx, project.RepoPath, commit, project.DefaultBranch, notes)
|
||||
if errors.Is(err, git.ErrRemoteMoved) {
|
||||
@@ -552,9 +372,9 @@ func (m *Manager) squashAndClose(ctx context.Context, taskID, login string) erro
|
||||
if !landed {
|
||||
return err
|
||||
}
|
||||
slog.Warn("push reported failure but landed", "task", taskID, "commit", commit, "err", err)
|
||||
slog.Warn("push reported failure but landed", "task", task.Id, "commit", commit, "err", err)
|
||||
}
|
||||
return m.closePushed(ctx, task, project, commit, notes, login)
|
||||
return m.closePushed(ctx, task, project, commit, notes, f.By)
|
||||
}
|
||||
|
||||
func (m *Manager) closeUnchanged(ctx context.Context, task api.Task, project api.Project, login string) error {
|
||||
@@ -749,10 +569,11 @@ func (m *Manager) stopMerge(ctx context.Context, l *live, raw json.RawMessage) (
|
||||
if !m.leadsTask(ctx, l.taskID, l.id) {
|
||||
return "", fmt.Errorf("%w: only the team's first agent runs the merge, so only it can stop it", ErrConflict)
|
||||
}
|
||||
switch {
|
||||
case m.takeMerge(l.taskID) != "":
|
||||
go m.releaseMerge(l.projectID, l.taskID)
|
||||
default:
|
||||
mergingBy, err := m.store.TakeMergingBy(ctx, l.taskID)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
if mergingBy == "" {
|
||||
queuedBy, err := m.store.TakeQueuedMerge(ctx, l.taskID)
|
||||
if err != nil {
|
||||
return "", err
|
||||
|
||||
Reference in New Issue
Block a user