diff --git a/.changes/unreleased/Added-20260805-051311.yaml b/.changes/unreleased/Added-20260805-051311.yaml new file mode 100644 index 000000000..16114baed --- /dev/null +++ b/.changes/unreleased/Added-20260805-051311.yaml @@ -0,0 +1,3 @@ +kind: Added +body: 'merge: Eligible linear GitHub pull request stacks are merged atomically, with bottom-up fallback when native range merging is unavailable.' +time: 2026-08-05T05:13:11.409252-07:00 diff --git a/internal/forge/forge.go b/internal/forge/forge.go index 2a9b70a5e..d1e929578 100644 --- a/internal/forge/forge.go +++ b/internal/forge/forge.go @@ -206,6 +206,8 @@ type Repository interface { FindChangesByBranch(ctx context.Context, branch string, opts FindChangesOptions) ([]*FindChangeItem, error) FindChangeByID(ctx context.Context, id ChangeID) (*FindChangeItem, error) + + // ChangeStatuses returns one status for each ID in the same order. ChangeStatuses(ctx context.Context, ids []ChangeID) ([]ChangeStatus, error) // ChangeChecks reports CI/checks for the given change. diff --git a/internal/handler/merge/handler.go b/internal/handler/merge/handler.go index 01bafde3a..8de943a88 100644 --- a/internal/handler/merge/handler.go +++ b/internal/handler/merge/handler.go @@ -635,6 +635,33 @@ func (h *Handler) executePlan( } } + // The forge plans only the selected changes it can merge atomically. The + // scheduler will preserve every omitted change as an ordinary merge item. + var nativePlans []forge.MergeRangePlan + if stackRepository, ok := h.RemoteRepository.(forge.StackRepository); opts.Command == "" && ok { + changeByBranch := make(map[string]forge.ChangeID, len(plan)) + for _, item := range plan { + changeByBranch[item.branch] = item.changeID + } + + changes := make([]forge.StackChange, len(plan)) + for i, item := range plan { + changes[i] = forge.StackChange{ + Change: item.changeID, + BaseChange: changeByBranch[item.base], + BaseBranch: item.base, + } + } + + var err error + nativePlans, err = stackRepository.PlanMergeRanges(ctx, changes) + if errors.Is(err, forge.ErrUnsupported) { + nativePlans = nil + } else if err != nil { + return fmt.Errorf("plan native merge ranges: %w", err) + } + } + var progress mergeProgress if runner, ok := h.View.(ui.ModelView); ok { widgetProgress := newWidgetMergeProgress( @@ -673,6 +700,8 @@ func (h *Handler) executePlan( Method: opts.Method, }) if opts.Command != "" { + // A custom command owns merge transport and may implement semantics + // that one provider-native range request cannot preserve. mergeRequester = &commandMergeRequester{ Runner: getCommandRunner(), Script: opts.Command, @@ -701,6 +730,7 @@ func (h *Handler) executePlan( Progress: progress, MergeRequester: mergeRequester, ReadinessChecker: readinessChecker, + MergeRangePlans: nativePlans, Trunk: h.Store.Trunk(), ReadyTimeout: opts.ReadyTimeout, @@ -933,48 +963,6 @@ func (e *mergePlanExecutor) awaitMergeabilityWithDelay( } } -// awaitMerged polls until the given change shows as merged. -// Uses exponential backoff starting at 500ms, capped at 8s. -func (e *mergePlanExecutor) awaitMerged( - ctx context.Context, item *mergeItem, -) error { - const ( - _initialDelay = 500 * time.Millisecond - _maxDelay = 8 * time.Second - ) - - ctx, cancel := context.WithTimeout(ctx, e.MergeTimeout) - defer cancel() - - // TODO: This only waits for the immediate change to reach - // the merged state. - // Server-side merge queues and richer merge workflows - // need a more expressive wait state. - delay := _initialDelay - for { - statuses, err := e.RemoteRepository.ChangeStatuses( - ctx, []forge.ChangeID{item.changeID}, - ) - if err != nil { - return fmt.Errorf("poll state: %w", err) - } - - if statuses[0].State == forge.ChangeMerged { - return nil - } - - e.Progress.Event(mergeProgressEvent{ - Kind: mergeProgressWaitingForMerge, - Item: item, - }) - if err := sleep(ctx, delay); err != nil { - return errors.New("timed out waiting for merge") - } - - delay = min(delay*2, _maxDelay) - } -} - func sleep(ctx context.Context, d time.Duration) error { select { case <-time.After(d): diff --git a/internal/handler/merge/handler_test.go b/internal/handler/merge/handler_test.go index 3155e82fe..29b05c642 100644 --- a/internal/handler/merge/handler_test.go +++ b/internal/handler/merge/handler_test.go @@ -40,6 +40,19 @@ type fakeChangeID string func (f fakeChangeID) String() string { return string(f) } +type scriptedMergeOperation struct { + statuses []forge.MergeOperationStatus + calls int +} + +func (o *scriptedMergeOperation) Status( + context.Context, +) (forge.MergeOperationStatus, error) { + status := o.statuses[o.calls] + o.calls++ + return status, nil +} + func TestOptions_mergeTimeoutDefault(t *testing.T) { var got Options parser, err := kong.New(&got) @@ -92,7 +105,12 @@ func TestAwaitMerged_immediate(t *testing.T) { Method: forge.MergeMethodDefault, } - err := executor.awaitMerged(t.Context(), item) + items := []*mergeItem{item} + err := executor.awaitMerged( + t.Context(), + items, + newChangeCompletionChecker(h.RemoteRepository, items), + ) require.NoError(t, err) } @@ -144,8 +162,53 @@ func TestAwaitMerged_afterPolling(t *testing.T) { Method: forge.MergeMethodDefault, } - err := executor.awaitMerged(t.Context(), item) + items := []*mergeItem{item} + err := executor.awaitMerged( + t.Context(), + items, + newChangeCompletionChecker(h.RemoteRepository, items), + ) + require.NoError(t, err) + }) +} + +func TestAwaitMerged_operationAcceptedThenChangesMerge(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + ctrl := gomock.NewController(t) + items := []*mergeItem{ + {branch: "feat1", changeID: fakeChangeID("pr-1")}, + {branch: "feat2", changeID: fakeChangeID("pr-2")}, + } + ids := []forge.ChangeID{ + fakeChangeID("pr-1"), + fakeChangeID("pr-2"), + } + + mockRepo := forgetest.NewMockRepository(ctrl) + mockRepo.EXPECT(). + ChangeStatuses(gomock.Any(), ids). + Return([]forge.ChangeStatus{ + {State: forge.ChangeMerged}, + {State: forge.ChangeMerged}, + }, nil) + h := newTestHandler(t, ctrl, testHandlerOpts{forgeRepo: mockRepo}) + executor := new(mergePlanExecutor) + executor.Progress = newLogMergeProgress(silog.Nop()) + executor.MergeTimeout = 2 * time.Minute + operation := &scriptedMergeOperation{ + statuses: []forge.MergeOperationStatus{ + forge.MergeOperationPending, + forge.MergeOperationAccepted, + }, + } + + changes := newChangeCompletionChecker(h.RemoteRepository, items) + err := executor.awaitMerged(t.Context(), items, &operationCompletionChecker{ + operation: operation, + finalState: changes, + }) require.NoError(t, err) + assert.Equal(t, 2, operation.calls) }) } @@ -190,7 +253,12 @@ func TestAwaitMerged_respectsMergeTimeout(t *testing.T) { Method: forge.MergeMethodDefault, } - err := executor.awaitMerged(t.Context(), item) + items := []*mergeItem{item} + err := executor.awaitMerged( + t.Context(), + items, + newChangeCompletionChecker(h.RemoteRepository, items), + ) require.Error(t, err) assert.EqualError(t, err, "timed out waiting for merge") }) @@ -2359,7 +2427,16 @@ func TestExecutePlan_mergeCommandRequestsThenAwaitsMerge(t *testing.T) { Return(nil) h := newTestHandler(t, ctrl, testHandlerOpts{ - forgeRepo: mockForge, + forgeRepo: &testStackRepository{ + Repository: mockForge, + planMergeRanges: func( + context.Context, + []forge.StackChange, + ) ([]forge.MergeRangePlan, error) { + t.Fatal("custom merge command must bypass native range planning") + return nil, nil + }, + }, sync: mockSync, logBuffer: &logBuffer, }) @@ -2702,7 +2779,7 @@ func TestValidateSynced_errorSkipped(t *testing.T) { // newTestHandler supplies inert collaborators and an in-memory store // for fields left unset. type testHandlerOpts struct { - forgeRepo *forgetest.MockRepository + forgeRepo forge.Repository store Store service *MockService restack *MockRestackHandler diff --git a/internal/handler/merge/scheduler.go b/internal/handler/merge/scheduler.go index 0edf0b50e..3a9469839 100644 --- a/internal/handler/merge/scheduler.go +++ b/internal/handler/merge/scheduler.go @@ -1,11 +1,14 @@ package merge import ( + "cmp" "context" + "errors" "fmt" "time" "go.abhg.dev/gs/internal/forge" + "go.abhg.dev/gs/internal/graph" "go.abhg.dev/gs/internal/handler/sync" "go.abhg.dev/gs/internal/mergequeue" ) @@ -28,6 +31,11 @@ type mergePlanExecutor struct { MergeRequester mergeRequester // required ReadinessChecker readinessChecker // required + // MergeRangePlans identifies the disjoint selected paths the forge can + // merge atomically. Selected changes omitted from the plans remain ordinary + // queue items. + MergeRangePlans []forge.MergeRangePlan + Trunk string // required ReadyTimeout time.Duration // required MergeTimeout time.Duration // required @@ -47,22 +55,9 @@ func (e *mergePlanExecutor) Execute( ctx context.Context, plan []*mergeItem, ) error { - inQueue := make(map[string]struct{}, len(plan)) - for _, item := range plan { - inQueue[item.branch] = struct{}{} - } - - items := make([]mergequeue.Item, 0, len(plan)) - for _, item := range plan { - var parent string - if _, ok := inQueue[item.base]; item.base != e.Trunk && ok { - parent = item.base - } - items = append(items, &mergeQueueItem{ - mergeItem: item, - executor: e, - parent: parent, - }) + items, err := e.mergeQueueItems(plan) + if err != nil { + return fmt.Errorf("build merge queue items: %w", err) } barrier := func(ctx context.Context) error { @@ -90,32 +85,232 @@ func (e *mergePlanExecutor) Execute( return scheduler.Run(ctx) } -var _ mergequeue.Item = (*mergeQueueItem)(nil) +// mergeQueueItems overlays provider-selected atomic ranges onto the complete +// user-selected merge forest. Changes omitted from the provider plans remain +// ordinary queue items, so native-stack limitations never remove work from the +// user's merge request. +func (e *mergePlanExecutor) mergeQueueItems( + plan []*mergeItem, +) ([]mergequeue.Item, error) { + byBranch := make(map[string]*mergeItem, len(plan)) + byChange := make(map[string]*mergeItem, len(plan)) + for _, item := range plan { + if _, ok := byBranch[item.branch]; ok { + return nil, fmt.Errorf("duplicate branch %q", item.branch) + } + byBranch[item.branch] = item + + changeKey := item.changeID.String() + if previous, ok := byChange[changeKey]; ok { + return nil, fmt.Errorf( + "branches %q and %q track the same change %v", + previous.branch, + item.branch, + item.changeID, + ) + } + byChange[changeKey] = item + } + + ordered, err := graph.Toposort(plan, + func(item *mergeItem) (*mergeItem, bool) { + base, ok := byBranch[item.base] + return base, ok + }) + if err != nil { + return nil, err + } + + // Contract each provider plan into one scheduler item. ownerByBranch lets + // later ordinary changes depend on the whole atomic range containing their + // immediate base rather than on a branch ID that no longer exists in the + // queue. + ownerByBranch := make(map[string]mergeQueueProgressItem, len(plan)) + for planIndex, nativePlan := range e.MergeRangePlans { + if nativePlan == nil { + return nil, fmt.Errorf("native merge plan %d is nil", planIndex) + } + + changeIDs := nativePlan.Changes() + if len(changeIDs) == 0 { + return nil, fmt.Errorf("native merge plan %d is empty", planIndex) + } + + items := make([]*mergeItem, len(changeIDs)) + for i, changeID := range changeIDs { + item, ok := byChange[changeID.String()] + if !ok { + return nil, fmt.Errorf( + "native merge plan %d contains unselected change %v", + planIndex, + changeID, + ) + } + if _, claimed := ownerByBranch[item.branch]; claimed { + return nil, fmt.Errorf( + "native merge plan %d overlaps at change %v", + planIndex, + changeID, + ) + } + if i > 0 && item.base != items[i-1].branch { + return nil, fmt.Errorf( + "native merge plan %d change %v has base branch %q, want %q", + planIndex, + changeID, + item.base, + items[i-1].branch, + ) + } + items[i] = item + } + + queueItem := &rangeMergeQueueItem{ + items: items, + plan: nativePlan, + executor: e, + } + for _, item := range items { + ownerByBranch[item.branch] = queueItem + } + } + + for _, item := range ordered { + if _, planned := ownerByBranch[item.branch]; planned { + continue + } + ownerByBranch[item.branch] = &changeMergeQueueItem{ + item: item, + executor: e, + } + } + + queueItems := make([]mergequeue.Item, 0, len(plan)) + added := make(map[mergeQueueProgressItem]struct{}, len(plan)) + for _, item := range ordered { + queueItem := ownerByBranch[item.branch] + if _, ok := added[queueItem]; ok { + continue + } + added[queueItem] = struct{}{} + + bottom := queueItem.changes()[0] + if baseOwner, ok := ownerByBranch[bottom.base]; ok { + queueItem.setParent(baseOwner.ID()) + } + queueItems = append(queueItems, queueItem) + } + return queueItems, nil +} + +// mergeQueueProgressItem exposes the change-level outcome represented by one +// scheduler item. The observer uses it to translate a range failure or skip +// back into progress for each affected change. +type mergeQueueProgressItem interface { + mergequeue.Item + + setParent(string) + changes() []*mergeItem + unmergedChanges() []*mergeItem +} + +var ( + _ mergeQueueProgressItem = (*changeMergeQueueItem)(nil) + _ mergeQueueProgressItem = (*rangeMergeQueueItem)(nil) +) + +// changeMergeQueueItem preserves the ordinary per-change merge path for a +// selected change that the forge omitted from its native plans. +type changeMergeQueueItem struct { + item *mergeItem + + executor *mergePlanExecutor + parent string +} + +func (i *changeMergeQueueItem) ID() string { + return i.item.branch +} + +func (i *changeMergeQueueItem) Parent() string { + return i.parent +} + +func (i *changeMergeQueueItem) setParent(parent string) { + i.parent = parent +} -type mergeQueueItem struct { - *mergeItem +func (i *changeMergeQueueItem) Prepare(ctx context.Context) error { + return i.executor.prepareItem(ctx, i.item) +} + +func (i *changeMergeQueueItem) Run(ctx context.Context) error { + return i.executor.mergeItem(ctx, i.item) +} + +func (i *changeMergeQueueItem) changes() []*mergeItem { + return []*mergeItem{i.item} +} + +func (i *changeMergeQueueItem) unmergedChanges() []*mergeItem { + return []*mergeItem{i.item} +} + +// rangeMergeQueueItem owns one linear range merge, +// including the prepared range request +// and partial completion if the repository requires ordinary fallback. +type rangeMergeQueueItem struct { + items []*mergeItem // bottom-to-top + plan forge.MergeRangePlan // required executor *mergePlanExecutor + parent string + + // Prepare captures the request after aligning every item; Run consumes it. + request forge.MergeRangeRequest - // parent is the queue-local branch dependency. - // Empty means the dependency is already satisfied outside this queue. - parent string + // completed counts bottom-most items merged before Run returned. + completed int } -func (i *mergeQueueItem) ID() string { - return i.branch +func (i *rangeMergeQueueItem) ID() string { + return i.items[len(i.items)-1].branch } -func (i *mergeQueueItem) Parent() string { +func (i *rangeMergeQueueItem) Parent() string { return i.parent } -func (i *mergeQueueItem) Prepare(ctx context.Context) error { - return i.executor.prepareItem(ctx, i.mergeItem) +func (i *rangeMergeQueueItem) setParent(parent string) { + i.parent = parent } -func (i *mergeQueueItem) Run(ctx context.Context) error { - return i.executor.mergeItem(ctx, i.mergeItem) +func (i *rangeMergeQueueItem) Prepare(ctx context.Context) error { + request, err := i.executor.prepareMergeRange(ctx, i.items) + if err != nil { + return err + } + i.request = request + return nil +} + +func (i *rangeMergeQueueItem) Run(ctx context.Context) error { + completed, err := i.executor.mergePreparedRange( + ctx, + i.plan, + i.items, + i.request, + ) + i.completed = completed + return err +} + +func (i *rangeMergeQueueItem) changes() []*mergeItem { + return i.items +} + +func (i *rangeMergeQueueItem) unmergedChanges() []*mergeItem { + return i.items[i.completed:] } func (e *mergePlanExecutor) prepareItem( @@ -142,6 +337,92 @@ func (e *mergePlanExecutor) prepareItem( return nil } +// prepareMergeRange aligns every member with its current base, then snapshots +// the provider-facing branch path after any restack and submit operations. +func (e *mergePlanExecutor) prepareMergeRange( + ctx context.Context, + items []*mergeItem, +) (forge.MergeRangeRequest, error) { + for _, item := range items { + e.Progress.Event(mergeProgressEvent{ + Kind: mergeProgressPreparing, + Item: item, + }) + if err := e.prepareForMerge(ctx, item); err != nil { + e.Progress.Event(mergeProgressEvent{ + Kind: mergeProgressPrepareFailed, + Item: item, + }) + return forge.MergeRangeRequest{}, fmt.Errorf( + "prepare %q: %w", + item.branch, + err, + ) + } + } + + // Preparation may restack branches or change their published metadata. + // Reload the graph before capturing the provider-facing path. + branchGraph, err := e.Service.BranchGraph(ctx, nil) + if err != nil { + return forge.MergeRangeRequest{}, fmt.Errorf( + "refresh branch graph: %w", + err, + ) + } + + changes := make([]forge.MergeRangeChange, 0, len(items)) + for idx, item := range items { + branch, ok := branchGraph.Lookup(item.branch) + if !ok { + return forge.MergeRangeRequest{}, fmt.Errorf( + "branch %q is no longer tracked", + item.branch, + ) + } + if branch.Change == nil || + branch.Change.ChangeID().String() != item.changeID.String() { + return forge.MergeRangeRequest{}, fmt.Errorf( + "branch %q no longer tracks change %v", + item.branch, + item.changeID, + ) + } + if idx > 0 && branch.Base != items[idx-1].branch { + return forge.MergeRangeRequest{}, fmt.Errorf( + "branch %q now has base %q, want %q", + item.branch, + branch.Base, + items[idx-1].branch, + ) + } + if branch.Head != item.headHash { + return forge.MergeRangeRequest{}, fmt.Errorf( + "branch %q head changed to %s after preparation, expected %s", + item.branch, + branch.Head, + item.headHash, + ) + } + + base := branch.Base + if baseBranch, ok := branchGraph.Lookup(base); ok { + base = cmp.Or(baseBranch.UpstreamBranch, baseBranch.Name) + } + changes = append(changes, forge.MergeRangeChange{ + Change: item.changeID, + Base: base, + Head: cmp.Or(branch.UpstreamBranch, branch.Name), + HeadHash: item.headHash, + }) + } + + return forge.MergeRangeRequest{ + Changes: changes, + Method: e.Method, + }, nil +} + func (e *mergePlanExecutor) mergeItem( ctx context.Context, item *mergeItem, @@ -165,7 +446,13 @@ func (e *mergePlanExecutor) mergeItem( }) return fmt.Errorf("wait for merge readiness: %w", err) } + return e.requestMergeItem(ctx, item) +} +func (e *mergePlanExecutor) requestMergeItem( + ctx context.Context, + item *mergeItem, +) error { e.Progress.Event(mergeProgressEvent{ Kind: mergeProgressMerging, Item: item, @@ -184,7 +471,12 @@ func (e *mergePlanExecutor) mergeItem( Kind: mergeProgressWaitingForMerge, Item: item, }) - if err := e.awaitMerged(ctx, item); err != nil { + items := []*mergeItem{item} + if err := e.awaitMerged( + ctx, + items, + newChangeCompletionChecker(e.RemoteRepository, items), + ); err != nil { e.Progress.Event(mergeProgressEvent{ Kind: mergeProgressMergeIncomplete, Item: item, @@ -198,6 +490,272 @@ func (e *mergePlanExecutor) mergeItem( return nil } +// mergePreparedRange waits for every prepared member and requests one range +// merge. ErrUnsupported resumes the ordinary bottom-up sequence. +func (e *mergePlanExecutor) mergePreparedRange( + ctx context.Context, + plan forge.MergeRangePlan, + items []*mergeItem, + request forge.MergeRangeRequest, +) (int, error) { + // A range can merge only when every member has reached the same remote + // head that was captured during preparation and is independently ready. + for _, item := range items { + if err := e.awaitChangeHead(ctx, item); err != nil { + e.Progress.Event(mergeProgressEvent{ + Kind: mergeProgressForgeHeadFailed, + Item: item, + }) + return 0, fmt.Errorf( + "%s: wait for pushed head: %w", + item.branch, + err, + ) + } + if err := e.awaitMergeability(ctx, item); err != nil { + e.Progress.Event(mergeProgressEvent{ + Kind: mergeProgressMergeabilityFailed, + Item: item, + }) + return 0, fmt.Errorf( + "%s: wait for merge readiness: %w", + item.branch, + err, + ) + } + } + + operation, err := plan.Merge(ctx, request) + if errors.Is(err, forge.ErrUnsupported) { + // [forge.ErrUnsupported] guarantees that no range merge started. + return e.mergeRangeIndividually(ctx, items) + } + if err != nil { + for _, item := range items { + e.Progress.Event(mergeProgressEvent{ + Kind: mergeProgressMergeFailed, + Item: item, + }) + } + return 0, fmt.Errorf("merge range: %w", err) + } + + for _, item := range items { + e.Progress.Event(mergeProgressEvent{ + Kind: mergeProgressMerging, + Item: item, + URL: item.mergeURL, + }) + e.Progress.Event(mergeProgressEvent{ + Kind: mergeProgressWaitingForMerge, + Item: item, + }) + } + completion := mergeCompletionChecker( + newChangeCompletionChecker(e.RemoteRepository, items), + ) + if operation != nil { + completion = &operationCompletionChecker{ + operation: operation, + finalState: completion, + } + } + if err := e.awaitMerged(ctx, items, completion); err != nil { + for _, item := range items { + e.Progress.Event(mergeProgressEvent{ + Kind: mergeProgressMergeIncomplete, + Item: item, + }) + } + return 0, fmt.Errorf("await merge range: %w", err) + } + for _, item := range items { + e.Progress.Event(mergeProgressEvent{ + Kind: mergeProgressMerged, + Item: item, + }) + } + return len(items), nil +} + +// mergeRangeIndividually resumes the ordinary bottom-up workflow after the +// forge declines a native range without starting it. Range preflight already +// established readiness for the first item. Each later item is synchronized, +// prepared against its newly merged base, and checked again before merging. +func (e *mergePlanExecutor) mergeRangeIndividually( + ctx context.Context, + items []*mergeItem, +) (int, error) { + completed := 0 + for idx, item := range items { + if idx > 0 { + if err := e.Sync.SyncTrunk(ctx, &sync.TrunkOptions{ + ClosedChanges: sync.ClosedChangesIgnore, + }); err != nil { + return completed, fmt.Errorf("sync trunk: %w", err) + } + if err := e.prepareItem(ctx, item); err != nil { + return completed, err + } + } + + var err error + if idx == 0 { + err = e.requestMergeItem(ctx, item) + } else { + err = e.mergeItem(ctx, item) + } + if err != nil { + return completed, fmt.Errorf( + "fallback merge %q: %w", + item.branch, + err, + ) + } + completed++ + } + return completed, nil +} + +// mergeCompletionChecker reports whether a requested merge has completed. +// Implementations hide the provider-specific probes from the polling loop. +type mergeCompletionChecker interface { + CheckMergeComplete(context.Context) (bool, error) +} + +// changeCompletionChecker observes final change state for one merge request. +// ChangeStatuses returns results in request order, so each status maps back to +// the same merge item without a second positional representation. +type changeCompletionChecker struct { + repository forge.Repository // required + items []*mergeItem // changes whose merged state is observed +} + +func newChangeCompletionChecker( + repository forge.Repository, + items []*mergeItem, +) *changeCompletionChecker { + return &changeCompletionChecker{ + repository: repository, + items: items, + } +} + +func (c *changeCompletionChecker) CheckMergeComplete( + ctx context.Context, +) (bool, error) { + changeIDs := make([]forge.ChangeID, len(c.items)) + for i, item := range c.items { + changeIDs[i] = item.changeID + } + statuses, err := c.repository.ChangeStatuses(ctx, changeIDs) + if err != nil { + return false, fmt.Errorf("poll state: %w", err) + } + if len(statuses) != len(c.items) { + return false, fmt.Errorf( + "poll state: forge returned %d change statuses, want %d", + len(statuses), + len(c.items), + ) + } + + allMerged := true + for i, status := range statuses { + switch status.State { + case forge.ChangeMerged: + case forge.ChangeOpen: + allMerged = false + case forge.ChangeClosed: + return false, fmt.Errorf( + "%s: change closed without merging", + c.items[i].branch, + ) + default: + return false, fmt.Errorf( + "%s: unknown change state %v", + c.items[i].branch, + status.State, + ) + } + } + return allMerged, nil +} + +// operationCompletionChecker waits for provider acceptance before observing +// final change state. Clearing operation records the phase transition so later +// polls never repeat a completed provider operation. +type operationCompletionChecker struct { + operation forge.MergeOperation // required until accepted + finalState mergeCompletionChecker // required +} + +func (c *operationCompletionChecker) CheckMergeComplete( + ctx context.Context, +) (bool, error) { + if c.operation != nil { + status, err := c.operation.Status(ctx) + if err != nil { + return false, fmt.Errorf("poll merge operation: %w", err) + } + switch status { + case forge.MergeOperationPending: + return false, nil + case forge.MergeOperationAccepted: + c.operation = nil + default: + return false, fmt.Errorf( + "poll merge operation: unknown status %v", + status, + ) + } + } + return c.finalState.CheckMergeComplete(ctx) +} + +// awaitMerged gives an ordinary change and an atomic range the same timeout, +// progress, and polling policy while their checkers own provider-specific +// completion phases. +func (e *mergePlanExecutor) awaitMerged( + ctx context.Context, + items []*mergeItem, + completion mergeCompletionChecker, +) error { + const ( + _initialDelay = 500 * time.Millisecond + _maxDelay = 8 * time.Second + ) + + ctx, cancel := context.WithTimeout(ctx, e.MergeTimeout) + defer cancel() + + delay := _initialDelay + for { + merged, err := completion.CheckMergeComplete(ctx) + if err != nil { + if errors.Is(ctx.Err(), context.DeadlineExceeded) { + return errors.New("timed out waiting for merge") + } + return err + } + if merged { + return nil + } + + for _, item := range items { + e.Progress.Event(mergeProgressEvent{ + Kind: mergeProgressWaitingForMerge, + Item: item, + }) + } + if err := sleep(ctx, delay); err != nil { + return errors.New("timed out waiting for merge") + } + + delay = min(delay*2, _maxDelay) + } +} + // mergeQueueObserver adapts scheduler decisions // back into merge progress events. type mergeQueueObserver struct { @@ -209,20 +767,23 @@ func (o *mergeQueueObserver) Prepared(mergequeue.Item) {} func (o *mergeQueueObserver) Done(mergequeue.Item) {} func (o *mergeQueueObserver) Failed(queueItem mergequeue.Item, _ error) { - item := queueItem.(*mergeQueueItem).mergeItem - o.progress.Event(mergeProgressEvent{ - Kind: mergeProgressFailed, - Item: item, - }) + item := queueItem.(mergeQueueProgressItem) + for _, change := range item.unmergedChanges() { + o.progress.Event(mergeProgressEvent{ + Kind: mergeProgressFailed, + Item: change, + }) + } } func (o *mergeQueueObserver) Skipped( queueItem mergequeue.Item, _ mergequeue.SkipReason, ) { - item := queueItem.(*mergeQueueItem).mergeItem - o.progress.Event(mergeProgressEvent{ - Kind: mergeProgressSkipped, - Item: item, - }) + for _, item := range queueItem.(mergeQueueProgressItem).changes() { + o.progress.Event(mergeProgressEvent{ + Kind: mergeProgressSkipped, + Item: item, + }) + } } diff --git a/internal/handler/merge/scheduler_test.go b/internal/handler/merge/scheduler_test.go index e06d9acba..937ab73c3 100644 --- a/internal/handler/merge/scheduler_test.go +++ b/internal/handler/merge/scheduler_test.go @@ -17,8 +17,463 @@ import ( "go.abhg.dev/gs/internal/forge/forgetest" "go.abhg.dev/gs/internal/git" "go.abhg.dev/gs/internal/handler/sync" + "go.abhg.dev/gs/internal/spice" + "go.abhg.dev/gs/internal/spice/spicetest" ) +func TestMergePlanExecutor_mergeQueueItems_overlaysNativePlans(t *testing.T) { + pr1 := fakeChangeID("pr-1") + pr2 := fakeChangeID("pr-2") + pr3 := fakeChangeID("pr-3") + pr4 := fakeChangeID("pr-4") + plan := []*mergeItem{ + testPlanEntry("feat1", "main", pr1), + testPlanEntry("feat2", "feat1", pr2), + testPlanEntry("feat3", "feat2", pr3), + testPlanEntry("feat4", "feat2", pr4), + } + + executor := new(mergePlanExecutor) + executor.MergeRangePlans = []forge.MergeRangePlan{&testMergeRangePlan{ + changes: []forge.ChangeID{pr1, pr2, pr4}, + }} + queueItems, err := executor.mergeQueueItems(plan) + require.NoError(t, err) + require.Len(t, queueItems, 2) + + assert.Equal(t, "feat4", queueItems[0].ID()) + assert.Empty(t, queueItems[0].Parent()) + assert.Equal(t, "feat3", queueItems[1].ID()) + assert.Equal(t, "feat4", queueItems[1].Parent()) + + nativeRange := queueItems[0].(*rangeMergeQueueItem) + assert.Equal(t, []string{"feat1", "feat2", "feat4"}, []string{ + nativeRange.items[0].branch, + nativeRange.items[1].branch, + nativeRange.items[2].branch, + }) + ordinary := queueItems[1].(*changeMergeQueueItem) + assert.Equal(t, "feat3", ordinary.item.branch) +} + +func TestMergePlanExecutor_mergeQueueItems_rejectsInvalidNativePlans(t *testing.T) { + pr1 := fakeChangeID("pr-1") + pr2 := fakeChangeID("pr-2") + pr3 := fakeChangeID("pr-3") + plan := []*mergeItem{ + testPlanEntry("feat1", "main", pr1), + testPlanEntry("feat2", "feat1", pr2), + testPlanEntry("feat3", "feat1", pr3), + } + + t.Run("Empty", func(t *testing.T) { + executor := new(mergePlanExecutor) + executor.MergeRangePlans = []forge.MergeRangePlan{&testMergeRangePlan{}} + _, err := executor.mergeQueueItems(plan) + require.ErrorContains(t, err, "native merge plan 0 is empty") + }) + + t.Run("UnselectedChange", func(t *testing.T) { + executor := new(mergePlanExecutor) + executor.MergeRangePlans = []forge.MergeRangePlan{&testMergeRangePlan{ + changes: []forge.ChangeID{fakeChangeID("pr-4")}, + }} + _, err := executor.mergeQueueItems(plan) + require.ErrorContains(t, err, "contains unselected change pr-4") + }) + + t.Run("Overlap", func(t *testing.T) { + executor := new(mergePlanExecutor) + executor.MergeRangePlans = []forge.MergeRangePlan{ + &testMergeRangePlan{changes: []forge.ChangeID{pr1, pr2}}, + &testMergeRangePlan{changes: []forge.ChangeID{pr2}}, + } + _, err := executor.mergeQueueItems(plan) + require.ErrorContains(t, err, "native merge plan 1 overlaps") + }) + + t.Run("Nonlinear", func(t *testing.T) { + executor := new(mergePlanExecutor) + executor.MergeRangePlans = []forge.MergeRangePlan{&testMergeRangePlan{ + changes: []forge.ChangeID{pr2, pr3}, + }} + _, err := executor.mergeQueueItems(plan) + require.ErrorContains(t, err, `base branch "feat1", want "feat2"`) + }) +} + +func TestExecutePlan_nativePlanningFailure(t *testing.T) { + ctrl := gomock.NewController(t) + stackRepo := &testStackRepository{ + Repository: forgetest.NewMockRepository(ctrl), + planErr: errors.New("boom"), + } + h := newTestHandler(t, ctrl, testHandlerOpts{forgeRepo: stackRepo}) + + err := h.executePlan(t.Context(), []*mergeItem{ + testPlanEntry("feat1", "main", "pr-1"), + }, mergeExecutionOptions{}) + require.ErrorContains(t, err, "plan native merge ranges: boom") +} + +func TestExecutePlan_nativePlanningUnsupportedFallsBack(t *testing.T) { + ctrl := gomock.NewController(t) + changeID := fakeChangeID("pr-1") + mockForge := forgetest.NewMockRepository(ctrl) + mockForge.EXPECT(). + ChangeStatuses(gomock.Any(), []forge.ChangeID{changeID}). + Return([]forge.ChangeStatus{{ + State: forge.ChangeOpen, + HeadHash: "head-1", + }}, nil) + mockForge.EXPECT(). + ChangeMergeability(gomock.Any(), changeID). + Return(forge.ChangeMergeability{ + State: forge.ChangeMergeabilityReady, + }, nil) + mockForge.EXPECT(). + MergeChange(gomock.Any(), changeID, gomock.Any()). + Return(nil) + mockForge.EXPECT(). + ChangeStatuses(gomock.Any(), []forge.ChangeID{changeID}). + Return([]forge.ChangeStatus{{State: forge.ChangeMerged}}, nil) + + stackRepo := &testStackRepository{ + Repository: mockForge, + planErr: forge.ErrUnsupported, + } + h := newTestHandler(t, ctrl, testHandlerOpts{forgeRepo: stackRepo}) + item := testPlanEntry("feat1", "main", changeID) + item.headHash = "head-1" + + require.NoError(t, h.executePlan( + t.Context(), + []*mergeItem{item}, + mergeExecutionOptions{}, + )) +} + +func TestMergeScheduler_nativeLinearPathWithUnselectedDivergence(t *testing.T) { + ctrl := gomock.NewController(t) + pr1 := fakeChangeID("pr-1") + pr2 := fakeChangeID("pr-2") + pr3 := fakeChangeID("pr-3") + + mockForge := forgetest.NewMockRepository(ctrl) + for _, change := range []struct { + id fakeChangeID + head git.Hash + }{ + {id: pr1, head: "head-1"}, + {id: pr2, head: "head-2"}, + {id: pr3, head: "head-3"}, + } { + mockForge.EXPECT(). + ChangeStatuses(gomock.Any(), []forge.ChangeID{change.id}). + Return([]forge.ChangeStatus{{ + State: forge.ChangeOpen, + HeadHash: change.head, + }}, nil) + mockForge.EXPECT(). + ChangeMergeability(gomock.Any(), change.id). + Return(forge.ChangeMergeability{ + State: forge.ChangeMergeabilityReady, + }, nil) + } + mockForge.EXPECT(). + ChangeStatuses(gomock.Any(), []forge.ChangeID{pr1, pr2, pr3}). + Return([]forge.ChangeStatus{ + {State: forge.ChangeMerged}, + {State: forge.ChangeMerged}, + {State: forge.ChangeMerged}, + }, nil) + + var gotRequest forge.MergeRangeRequest + rangePlan := &testMergeRangePlan{ + changes: []forge.ChangeID{pr1, pr2, pr3}, + merge: func( + _ context.Context, + req forge.MergeRangeRequest, + ) (forge.MergeOperation, error) { + gotRequest = req + return nil, nil + }, + } + rangeRepo := &testStackRepository{ + Repository: mockForge, + planMergeRanges: func( + _ context.Context, + changes []forge.StackChange, + ) ([]forge.MergeRangePlan, error) { + assert.Equal(t, []forge.StackChange{ + {Change: pr1, BaseBranch: "main"}, + {Change: pr2, BaseChange: pr1, BaseBranch: "feat1"}, + {Change: pr3, BaseChange: pr2, BaseBranch: "feat2"}, + }, changes) + return []forge.MergeRangePlan{rangePlan}, nil + }, + } + + mockService := NewMockService(ctrl) + mockGit := NewMockGitRepository(ctrl) + for idx, branch := range []string{"feat1", "feat2", "feat3"} { + mockService.EXPECT().VerifyRestacked(gomock.Any(), branch).Return(nil) + mockGit.EXPECT().PeelToCommit(gomock.Any(), branch). + Return(git.Hash(fmt.Sprintf("head-%d", idx+1)), nil) + } + mockService.EXPECT(). + BranchGraph(gomock.Any(), nil). + Return(spicetest.NewBranchGraph(t, spicetest.BranchGraphConfig{ + Trunk: "main", + Branches: []spice.LoadBranchItem{ + { + Name: "feat1", + Head: "head-1", + Base: "main", + Change: testChangeMetadata(pr1), + UpstreamBranch: "remote-feat1", + }, + { + Name: "feat2", + Head: "head-2", + Base: "feat1", + Change: testChangeMetadata(pr2), + UpstreamBranch: "remote-feat2", + }, + { + Name: "feat3", + Head: "head-3", + Base: "feat2", + Change: testChangeMetadata(pr3), + UpstreamBranch: "remote-feat3", + }, + { + Name: "feat4", + Head: "head-4", + Base: "feat2", + Change: testChangeMetadata("pr-4"), + UpstreamBranch: "remote-feat4", + }, + }, + }), nil) + + syncHandler := &recordingSyncHandler{} + h := newTestHandler(t, ctrl, testHandlerOpts{ + forgeRepo: rangeRepo, + service: mockService, + gitRepo: mockGit, + sync: syncHandler, + }) + err := h.executePlan(t.Context(), testMergePlanWithBases( + testPlanEntry("feat1", "main", pr1), + testPlanEntry("feat2", "feat1", pr2), + testPlanEntry("feat3", "feat2", pr3), + ), mergeExecutionOptions{ + Method: forge.MergeMethodSquash, + MergeTimeout: time.Second, + }) + require.NoError(t, err) + assert.Equal(t, 1, syncHandler.calls) + assert.Equal(t, forge.MergeRangeRequest{ + Method: forge.MergeMethodSquash, + Changes: []forge.MergeRangeChange{ + { + Change: pr1, + Base: "main", + Head: "remote-feat1", + HeadHash: "head-1", + }, + { + Change: pr2, + Base: "remote-feat1", + Head: "remote-feat2", + HeadHash: "head-2", + }, + { + Change: pr3, + Base: "remote-feat2", + Head: "remote-feat3", + HeadHash: "head-3", + }, + }, + }, gotRequest) +} + +func TestMergeScheduler_nativeRangeUnsupportedFallsBack(t *testing.T) { + ctrl := gomock.NewController(t) + pr1 := fakeChangeID("pr-1") + pr2 := fakeChangeID("pr-2") + operations := &operationRecorder{} + + mockForge := forgetest.NewMockRepository(ctrl) + mockForge.EXPECT(). + ChangeStatuses(gomock.Any(), []forge.ChangeID{pr1}). + Return([]forge.ChangeStatus{{ + State: forge.ChangeOpen, + HeadHash: "head-1", + }}, nil) + mockForge.EXPECT(). + ChangeMergeability(gomock.Any(), pr1). + Return(forge.ChangeMergeability{ + State: forge.ChangeMergeabilityReady, + }, nil) + mockForge.EXPECT(). + ChangeStatuses(gomock.Any(), []forge.ChangeID{pr2}). + Return([]forge.ChangeStatus{{ + State: forge.ChangeOpen, + HeadHash: "head-2", + }}, nil). + Times(2) + mockForge.EXPECT(). + ChangeMergeability(gomock.Any(), pr2). + Return(forge.ChangeMergeability{ + State: forge.ChangeMergeabilityReady, + }, nil). + Times(2) + mockForge.EXPECT(). + MergeChange(gomock.Any(), pr1, gomock.Any()). + DoAndReturn(func( + context.Context, + forge.ChangeID, + forge.MergeChangeOptions, + ) error { + operations.append("merge pr-1") + return nil + }) + mockForge.EXPECT(). + ChangeStatuses(gomock.Any(), []forge.ChangeID{pr1}). + Return([]forge.ChangeStatus{{State: forge.ChangeMerged}}, nil) + mockForge.EXPECT(). + MergeChange(gomock.Any(), pr2, gomock.Any()). + DoAndReturn(func( + context.Context, + forge.ChangeID, + forge.MergeChangeOptions, + ) error { + operations.append("merge pr-2") + return nil + }) + mockForge.EXPECT(). + ChangeStatuses(gomock.Any(), []forge.ChangeID{pr2}). + Return([]forge.ChangeStatus{{State: forge.ChangeMerged}}, nil) + + rangePlan := &testMergeRangePlan{ + changes: []forge.ChangeID{pr1, pr2}, + merge: func( + context.Context, + forge.MergeRangeRequest, + ) (forge.MergeOperation, error) { + operations.append("range") + return nil, forge.ErrUnsupported + }, + } + rangeRepo := &testStackRepository{ + Repository: mockForge, + plans: []forge.MergeRangePlan{rangePlan}, + } + + mockService := NewMockService(ctrl) + mockService.EXPECT().VerifyRestacked(gomock.Any(), "feat1").Return(nil) + mockService.EXPECT().VerifyRestacked(gomock.Any(), "feat2").Return(nil).Times(2) + mockGit := NewMockGitRepository(ctrl) + mockGit.EXPECT().PeelToCommit(gomock.Any(), "feat1").Return(git.Hash("head-1"), nil) + mockGit.EXPECT().PeelToCommit(gomock.Any(), "feat2").Return(git.Hash("head-2"), nil).Times(2) + mockService.EXPECT(). + BranchGraph(gomock.Any(), nil). + Return(spicetest.NewBranchGraph(t, spicetest.BranchGraphConfig{ + Trunk: "main", + Branches: []spice.LoadBranchItem{ + { + Name: "feat1", + Head: "head-1", + Base: "main", + Change: testChangeMetadata(pr1), + UpstreamBranch: "feat1", + }, + { + Name: "feat2", + Head: "head-2", + Base: "feat1", + Change: testChangeMetadata(pr2), + UpstreamBranch: "feat2", + }, + }, + }), nil) + + syncHandler := &recordingSyncHandler{operations: operations} + h := newTestHandler(t, ctrl, testHandlerOpts{ + forgeRepo: rangeRepo, + service: mockService, + gitRepo: mockGit, + sync: syncHandler, + }) + err := h.executePlan(t.Context(), testMergePlanWithBases( + testPlanEntry("feat1", "main", pr1), + testPlanEntry("feat2", "feat1", pr2), + ), mergeExecutionOptions{MergeTimeout: time.Second}) + require.NoError(t, err) + assert.Equal(t, []string{ + "range", + "merge pr-1", + "sync", + "merge pr-2", + "sync", + }, operations.snapshot()) +} + +func TestMergePreparedRange_genuineFailure(t *testing.T) { + ctrl := gomock.NewController(t) + changeID := fakeChangeID("pr-1") + item := testPlanEntry("feat1", "main", changeID) + item.headHash = "head-1" + + mockForge := forgetest.NewMockRepository(ctrl) + mockForge.EXPECT(). + ChangeStatuses(gomock.Any(), []forge.ChangeID{changeID}). + Return([]forge.ChangeStatus{{ + State: forge.ChangeOpen, + HeadHash: "head-1", + }}, nil) + mockForge.EXPECT(). + ChangeMergeability(gomock.Any(), changeID). + Return(forge.ChangeMergeability{ + State: forge.ChangeMergeabilityReady, + }, nil) + rangePlan := &testMergeRangePlan{ + changes: []forge.ChangeID{changeID}, + merge: func( + context.Context, + forge.MergeRangeRequest, + ) (forge.MergeOperation, error) { + return nil, errors.New("boom") + }, + } + executor := new(mergePlanExecutor) + executor.RemoteRepository = mockForge + executor.Progress = new(recordingMergeProgress) + executor.ReadinessChecker = &forgeReadinessChecker{ + Repository: mockForge, + } + executor.ReadyTimeout = time.Second + + completed, err := executor.mergePreparedRange( + t.Context(), + rangePlan, + []*mergeItem{item}, + forge.MergeRangeRequest{ + Changes: []forge.MergeRangeChange{{ + Change: changeID, + Base: "main", + Head: "feat1", + HeadHash: "head-1", + }}, + }, + ) + assert.Zero(t, completed) + require.Error(t, err) + assert.ErrorContains(t, err, "merge range: boom") +} + func TestMergeScheduler_parentMergeUnlocksIndependentChildren(t *testing.T) { ctrl := gomock.NewController(t) @@ -792,3 +1247,50 @@ func expectMergeWithRecord( ChangeStatuses(gomock.Any(), []forge.ChangeID{id}). Return([]forge.ChangeStatus{{State: forge.ChangeMerged}}, nil) } + +type testStackRepository struct { + forge.Repository + + plans []forge.MergeRangePlan + planErr error + planMergeRanges func( + context.Context, + []forge.StackChange, + ) ([]forge.MergeRangePlan, error) +} + +func (*testStackRepository) PlanStackUpdate( + context.Context, + []forge.StackChange, +) (forge.StackUpdatePlan, error) { + return nil, forge.ErrUnsupported +} + +func (r *testStackRepository) PlanMergeRanges( + ctx context.Context, + changes []forge.StackChange, +) ([]forge.MergeRangePlan, error) { + if r.planMergeRanges != nil { + return r.planMergeRanges(ctx, changes) + } + return r.plans, r.planErr +} + +type testMergeRangePlan struct { + changes []forge.ChangeID + merge func( + context.Context, + forge.MergeRangeRequest, + ) (forge.MergeOperation, error) +} + +func (p *testMergeRangePlan) Changes() []forge.ChangeID { + return p.changes +} + +func (p *testMergeRangePlan) Merge( + ctx context.Context, + req forge.MergeRangeRequest, +) (forge.MergeOperation, error) { + return p.merge(ctx, req) +} diff --git a/testdata/script/downstack_merge_native_range.txt b/testdata/script/downstack_merge_native_range.txt new file mode 100644 index 000000000..f2e71045b --- /dev/null +++ b/testdata/script/downstack_merge_native_range.txt @@ -0,0 +1,51 @@ +# ShamHub merges an enabled native stack through one atomic range operation. + +as 'Test ' +at '2026-08-05T13:00:00Z' + +cd repo +git init +git config spice.experiment.merge true +git config spice.forge.shamhub.stacks on +git commit --allow-empty -m 'Initial commit' +git tag before + +shamhub-setup +shamhub new origin alice/example.git +shamhub register alice +git push origin main +env SHAMHUB_USERNAME=alice +gs auth login + +git add bottom.txt +gs bc bottom -m 'Add bottom' +git add middle.txt +gs bc middle -m 'Add middle' +git add top.txt +gs bc top -m 'Add top' + +gs downstack submit --fill +stderr 'Created #1' +stderr 'Created #2' +stderr 'Created #3' + +gs downstack merge --no-prompt +stderr 'bottom: #1 was merged' +stderr 'middle: #2 was merged' +stderr 'top: #3 was merged' +stderr 'All 3 change.s. merged.' +! stderr 'restacked on main' +! stderr 'Updated #2' +! stderr 'Updated #3' + +git rev-list --first-parent --count before..main +stdout '^1$' +git log -1 --format=%s main +stdout '^Merge changes #1 through #3$' + +-- repo/bottom.txt -- +bottom +-- repo/middle.txt -- +middle +-- repo/top.txt -- +top