Record merge progress durably and finish pushed merges at startup
This commit is contained in:
+129
-32
@@ -136,26 +136,82 @@ func (m *Manager) recoverMerges(ctx context.Context) error {
|
||||
}
|
||||
|
||||
func (m *Manager) recoverFinalizing(ctx context.Context) error {
|
||||
finalizing, err := m.store.FinalizingTasks(ctx)
|
||||
finalizing, err := m.store.FinalizingMerges(ctx)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
for _, t := range finalizing {
|
||||
err = m.store.ClearFinalizing(ctx, t.Id)
|
||||
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
|
||||
}
|
||||
sessions, err := m.store.ListSessions(ctx, t.Id)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
slog.Error("merge was finishing at startup", "task", t.Id)
|
||||
m.fileTaskError(ctx, t, sessions, "byre restarted while finishing this task's merge, so it may not have been pushed, or the task's worktree may be gone. Check the default branch before merging again.")
|
||||
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(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(project api.Project, commit string) (bool, error) {
|
||||
target := git.Remote + "/" + project.DefaultBranch
|
||||
err := git.Fetch(project.RepoPath)
|
||||
if err != nil {
|
||||
return false, &unfinishedMergeError{fmt.Errorf("checking whether %s reached %s failed: %w", commit, target, err)}
|
||||
}
|
||||
landed, err := git.IsAncestor(project.RepoPath, commit, target)
|
||||
if err != nil {
|
||||
return false, &unfinishedMergeError{fmt.Errorf("checking whether %s reached %s failed: %w", commit, target, err)}
|
||||
}
|
||||
return landed, nil
|
||||
}
|
||||
|
||||
type unfinishedMergeError struct {
|
||||
err error
|
||||
}
|
||||
|
||||
func (e *unfinishedMergeError) Error() string {
|
||||
return e.err.Error() + "; byre finishes the merge when it next starts"
|
||||
}
|
||||
|
||||
func (e *unfinishedMergeError) Unwrap() error {
|
||||
return e.err
|
||||
}
|
||||
|
||||
func (m *Manager) startHeldMerges(ctx context.Context) {
|
||||
projects, err := m.store.ProjectsWithQueuedMerges(ctx)
|
||||
if err != nil {
|
||||
@@ -274,6 +330,11 @@ func (m *Manager) finishMerge(l *live, login string) {
|
||||
}
|
||||
|
||||
slog.Error("finish merge", "task", l.taskID, "err", err)
|
||||
unfinished := &unfinishedMergeError{}
|
||||
if errors.As(err, &unfinished) {
|
||||
m.fileError(l, "Merging the task failed: "+err.Error()+".")
|
||||
return
|
||||
}
|
||||
cerr := m.store.ClearFinalizing(ctx, l.taskID)
|
||||
if cerr != nil {
|
||||
slog.Error("clear merge finalizing", "task", l.taskID, "err", cerr)
|
||||
@@ -315,38 +376,74 @@ func (m *Manager) squashAndClose(ctx context.Context, taskID, login string) erro
|
||||
return err
|
||||
}
|
||||
|
||||
pushed := ""
|
||||
if commit != "" {
|
||||
note, err := m.mergeNote(ctx, task, login)
|
||||
if commit == "" {
|
||||
err = m.store.SetFinalizingUnchanged(ctx, taskID)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
notes, err := git.NoteCommit(project.RepoPath, commit, note, env)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
err = git.Push(project.RepoPath, commit, project.DefaultBranch, notes)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
pushed = git.Remote + "/" + project.DefaultBranch
|
||||
err = git.SetNotes(project.RepoPath, notes)
|
||||
if err != nil {
|
||||
return fmt.Errorf("pushed %s to %s, but updating %s in the project checkout failed: %w", commit, pushed, git.NotesRef, err)
|
||||
}
|
||||
err = git.FastForward(project.RepoPath, commit)
|
||||
if err != nil {
|
||||
return fmt.Errorf("pushed %s to %s, but fast-forwarding the project checkout failed: %w", commit, pushed, err)
|
||||
}
|
||||
return m.closeUnchanged(ctx, task, project, login)
|
||||
}
|
||||
|
||||
err = m.RemoveTask(ctx, task, project, login)
|
||||
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
|
||||
}
|
||||
err = git.Push(project.RepoPath, commit, project.DefaultBranch, notes)
|
||||
if errors.Is(err, git.ErrRemoteMoved) {
|
||||
return err
|
||||
}
|
||||
if err != nil {
|
||||
landed, lerr := pushLanded(project, commit)
|
||||
if lerr != nil {
|
||||
return &unfinishedMergeError{fmt.Errorf("%w, and %w", err, errors.Unwrap(lerr))}
|
||||
}
|
||||
if !landed {
|
||||
return err
|
||||
}
|
||||
slog.Warn("push reported failure but landed", "task", taskID, "commit", commit, "err", err)
|
||||
}
|
||||
return m.closePushed(ctx, task, project, commit, notes, login)
|
||||
}
|
||||
|
||||
_, err = m.store.MergeTask(ctx, taskID, commit, pushed, login)
|
||||
return err
|
||||
func (m *Manager) closeUnchanged(ctx context.Context, task api.Task, project api.Project, login string) error {
|
||||
err := m.RemoveTask(ctx, task, project, login)
|
||||
if err != nil {
|
||||
return &unfinishedMergeError{fmt.Errorf("the merge changed nothing, but removing the task's worktree failed: %w", err)}
|
||||
}
|
||||
_, err = m.store.MergeTask(ctx, task.Id, "", "", login)
|
||||
if err != nil {
|
||||
return &unfinishedMergeError{fmt.Errorf("the merge changed nothing, but closing the task failed: %w", err)}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (m *Manager) closePushed(ctx context.Context, task api.Task, project api.Project, commit, notes, login string) error {
|
||||
pushed := git.Remote + "/" + project.DefaultBranch
|
||||
err := git.AdvanceNotes(project.RepoPath, notes)
|
||||
if err != nil {
|
||||
return &unfinishedMergeError{fmt.Errorf("pushed %s to %s, but updating %s in the project checkout failed: %w", commit, pushed, git.NotesRef, err)}
|
||||
}
|
||||
err = git.FastForward(project.RepoPath, commit)
|
||||
if err != nil {
|
||||
return &unfinishedMergeError{fmt.Errorf("pushed %s to %s, but fast-forwarding the project checkout failed: %w", commit, pushed, err)}
|
||||
}
|
||||
err = m.RemoveTask(ctx, task, project, login)
|
||||
if err != nil {
|
||||
return &unfinishedMergeError{fmt.Errorf("pushed %s to %s, but removing the task's worktree failed: %w", commit, pushed, err)}
|
||||
}
|
||||
_, err = m.store.MergeTask(ctx, task.Id, commit, pushed, login)
|
||||
if err != nil {
|
||||
return &unfinishedMergeError{fmt.Errorf("pushed %s to %s, but closing the task failed: %w", commit, pushed, err)}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (m *Manager) CoAuthors(ctx context.Context, taskID, login string) ([]api.Participant, error) {
|
||||
|
||||
+6
-1
@@ -251,8 +251,13 @@ func FastForward(repo, rev string) error {
|
||||
return err
|
||||
}
|
||||
|
||||
func FetchFastForward(repo, branch string) error {
|
||||
func Fetch(repo string) error {
|
||||
_, err := run(repo, "fetch", Remote)
|
||||
return err
|
||||
}
|
||||
|
||||
func FetchFastForward(repo, branch string) error {
|
||||
err := Fetch(repo)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
+18
-2
@@ -69,7 +69,23 @@ func NoteCommit(repo, commit, note string, env []string) (string, error) {
|
||||
return runEnv(repo, env, append(commitArgs, tree)...)
|
||||
}
|
||||
|
||||
func SetNotes(repo, notes string) error {
|
||||
_, err := run(repo, "update-ref", NotesRef, notes)
|
||||
func AdvanceNotes(repo, notes string) error {
|
||||
current, stderr, code, err := runCode(repo, nil, "rev-parse", "--verify", "--quiet", NotesRef)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if code != 0 && code != 1 {
|
||||
return fmt.Errorf("git rev-parse %s: exit status %d: %s", NotesRef, code, stderr)
|
||||
}
|
||||
if code == 0 {
|
||||
contained, err := IsAncestor(repo, notes, current)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if contained {
|
||||
return nil
|
||||
}
|
||||
}
|
||||
_, err = run(repo, "update-ref", NotesRef, notes)
|
||||
return err
|
||||
}
|
||||
|
||||
+57
-4
@@ -452,6 +452,10 @@ ALTER TABLE tasks ADD COLUMN finalizing TEXT NOT NULL DEFAULT '';
|
||||
ALTER TABLE tasks ADD COLUMN state TEXT NOT NULL DEFAULT '';
|
||||
ALTER TABLE tasks ADD COLUMN review_requests INTEGER NOT NULL DEFAULT 0;
|
||||
UPDATE tasks SET state = ` + taskStateExpr + `;
|
||||
`}, {"add-task-finalizing-commit", `
|
||||
ALTER TABLE tasks ADD COLUMN finalizing_commit TEXT NOT NULL DEFAULT '';
|
||||
ALTER TABLE tasks ADD COLUMN finalizing_notes TEXT NOT NULL DEFAULT '';
|
||||
ALTER TABLE tasks ADD COLUMN finalizing_unchanged INTEGER NOT NULL DEFAULT 0;
|
||||
`}}
|
||||
|
||||
const (
|
||||
@@ -874,7 +878,7 @@ func (s *Store) TakeMergingBy(ctx context.Context, id string) (string, error) {
|
||||
func (s *Store) ClearFinalizing(ctx context.Context, id string) error {
|
||||
return s.write(ctx, func(tx *sql.Tx, emit emitFunc) error {
|
||||
projectID := ""
|
||||
err := tx.QueryRowContext(ctx, `UPDATE tasks SET finalizing = '' WHERE id = ? AND finalizing != '' RETURNING project_id`, id).Scan(&projectID)
|
||||
err := tx.QueryRowContext(ctx, `UPDATE tasks SET finalizing = '', finalizing_commit = '', finalizing_notes = '', finalizing_unchanged = 0 WHERE id = ? AND finalizing != '' RETURNING project_id`, id).Scan(&projectID)
|
||||
if errors.Is(err, sql.ErrNoRows) {
|
||||
return nil
|
||||
}
|
||||
@@ -885,8 +889,57 @@ func (s *Store) ClearFinalizing(ctx context.Context, id string) error {
|
||||
})
|
||||
}
|
||||
|
||||
func (s *Store) FinalizingTasks(ctx context.Context) ([]api.Task, error) {
|
||||
return s.queryTasks(ctx, `SELECT `+taskColumns+` FROM tasks WHERE status = ? AND finalizing != ''`, string(api.TaskStatusActive))
|
||||
func (s *Store) SetFinalizingCommit(ctx context.Context, id, commit, notes string) error {
|
||||
return s.setFinalizing(ctx, id, commit, notes, false)
|
||||
}
|
||||
|
||||
func (s *Store) SetFinalizingUnchanged(ctx context.Context, id string) error {
|
||||
return s.setFinalizing(ctx, id, "", "", true)
|
||||
}
|
||||
|
||||
func (s *Store) setFinalizing(ctx context.Context, id, commit, notes string, unchanged bool) error {
|
||||
return s.write(ctx, func(tx *sql.Tx, emit emitFunc) error {
|
||||
res, err := tx.ExecContext(ctx, `UPDATE tasks SET finalizing_commit = ?, finalizing_notes = ?, finalizing_unchanged = ? WHERE id = ? AND status = ? AND finalizing != ''`,
|
||||
commit, notes, unchanged, id, string(api.TaskStatusActive))
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
n, err := res.RowsAffected()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if n == 0 {
|
||||
return ErrNotFound
|
||||
}
|
||||
return nil
|
||||
})
|
||||
}
|
||||
|
||||
type FinalizingMerge struct {
|
||||
Task api.Task
|
||||
By string
|
||||
Commit string
|
||||
Notes string
|
||||
Unchanged bool
|
||||
}
|
||||
|
||||
func (s *Store) FinalizingMerges(ctx context.Context) ([]FinalizingMerge, error) {
|
||||
rows, err := s.db.QueryContext(ctx, `SELECT `+taskColumns+`, finalizing, finalizing_commit, finalizing_notes, finalizing_unchanged FROM tasks WHERE status = ? AND finalizing != ''`, string(api.TaskStatusActive))
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer rows.Close()
|
||||
|
||||
merges := []FinalizingMerge{}
|
||||
for rows.Next() {
|
||||
f := FinalizingMerge{}
|
||||
f.Task, err = scanTask(trailingScanner{rows, []any{&f.By, &f.Commit, &f.Notes, &f.Unchanged}})
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
merges = append(merges, f)
|
||||
}
|
||||
return merges, rows.Err()
|
||||
}
|
||||
|
||||
func (s *Store) MergingTasks(ctx context.Context) ([]api.Task, error) {
|
||||
@@ -1107,7 +1160,7 @@ func (s *Store) DiscardTask(ctx context.Context, id, actor string) (api.Task, er
|
||||
|
||||
func (s *Store) closeTask(ctx context.Context, id string, status api.TaskStatus, commit, actor, kind string, payload map[string]any) (api.Task, error) {
|
||||
err := s.write(ctx, func(tx *sql.Tx, emit emitFunc) error {
|
||||
res, err := tx.ExecContext(ctx, `UPDATE tasks SET status = ?, closed_at = ?, merge_commit = ?, finalizing = '' WHERE id = ? AND status = ?`,
|
||||
res, err := tx.ExecContext(ctx, `UPDATE tasks SET status = ?, closed_at = ?, merge_commit = ?, finalizing = '', finalizing_commit = '', finalizing_notes = '', finalizing_unchanged = 0 WHERE id = ? AND status = ?`,
|
||||
string(status), now(), commit, id, string(api.TaskStatusActive))
|
||||
if err != nil {
|
||||
return err
|
||||
|
||||
Reference in New Issue
Block a user