diff --git a/runway/extension/merger/git/git_merger.go b/runway/extension/merger/git/git_merger.go index f0d6575f6..7f08d2a95 100644 --- a/runway/extension/merger/git/git_merger.go +++ b/runway/extension/merger/git/git_merger.go @@ -420,18 +420,21 @@ func (o *providerCheck) check(ref changeRef, stepID string) error { // strategies (REBASE, SQUASH_REBASE, MERGE), retrying on remote contention when // committing. For a dry run it applies the steps locally then discards them. func (m *gitMerger) applyTransforming(ctx context.Context, req *runwaymq.MergeRequest, steps []resolvedStep, commit bool) (*runwaymq.MergeResult, error) { - // Fetch and vet every change the request names before the first attempt, so - // an unusable request fails without having touched the checkout. This sits - // outside the retry loop deliberately: the refs are fixed for the whole - // request, and re-checking them per attempt would re-query the remote to - // learn what it already told us. + // Fetch every change the request names before the first attempt, so a + // request naming an unreachable commit fails without touching the checkout. + // + // Freshness is not checked here. It cannot be: a redelivery of work that + // already landed looks stale — its head branch has moved to the commit the + // merger made of it — and rejecting it would report an error for a change + // that is sitting in the target. The check moves to after local application, + // where "did this actually change anything" is answerable. The cost is that + // a genuinely superseded request now runs a full fetch, reset --hard, + // clean -fdx and every cherry-pick before it is rejected; only the local + // checkout is touched, and no ref is written. refs := stepChangeRefs(steps) if err := m.ensureObjects(ctx, refs); err != nil { return nil, err } - if err := m.checkStale(ctx, refs); err != nil { - return nil, err - } var lastErr error // Head branches an attempt moved before failing to push the target stay @@ -439,34 +442,20 @@ func (m *gitMerger) applyTransforming(ctx context.Context, req *runwaymq.MergeRe // the branch is moved on rather than stranded on a commit that never landed. tracked := make(headBranchTracker) for attempt := 1; attempt <= m.maxPushAttempts; attempt++ { - baseSHA, stepResults, err := m.tryApply(ctx, steps, commit, tracked) + result, baseSHA, err := m.attemptTransforming(ctx, req, steps, refs, commit, tracked) if err == nil { - if !commit { - // Discard the local commits the dry run created so the checkout - // is clean for the next operation, and report empty Outputs. - if derr := m.resetToRemote(ctx); derr != nil { - return nil, fmt.Errorf("discard after dry-run: %w", derr) - } - stripOutputs(stepResults) - } m.logger.Debugw("merge complete", "id", req.GetId(), "target", m.target, "commit", commit) - return successResult(req, stepResults), nil + return result, nil } - - // A conflict is terminal — no retry. Discard any partial dry-run state. - if errors.Is(err, merger.ErrConflict) { - if !commit { - _ = m.resetToRemote(ctx) - } + // A conflict or a superseded request is the answer, not something to + // retry. baseSHA is empty when the attempt failed before reset captured + // a base, which leaves nothing to compare the tip against. + if merger.IsTerminal(err) || !commit || baseSHA == "" { return nil, err } - // Only a push failure caused by the remote tip moving under us (between - // reset and push) is worth retrying; everything else is fatal. baseSHA - // is empty when the failure happened before reset captured a base. - if !commit || baseSHA == "" { - return nil, err - } + // Only a failure caused by the remote tip moving under us (between reset + // and push) is worth retrying; everything else is fatal. currentSHA, refetchErr := m.refetchTipSHA(ctx) if refetchErr != nil { return nil, fmt.Errorf("refetch after push failure failed: %v (original push error: %w)", refetchErr, err) @@ -490,43 +479,77 @@ func (m *gitMerger) applyTransforming(ctx context.Context, req *runwaymq.MergeRe return nil, fmt.Errorf("exceeded %d merge attempts due to remote contention: %w", m.maxPushAttempts, lastErr) } -// tryApply runs one full reset+apply(+push) cycle. The returned baseSHA is the -// SHA the cycle was based on (set as soon as resetToRemote completes) so the -// caller can distinguish concurrent-push contention from other failures. The -// tracker carries head-branch state across attempts; see headBranchTracker. -func (m *gitMerger) tryApply(ctx context.Context, steps []resolvedStep, commit bool, tracked headBranchTracker) (string, []*runwaymq.StepResult, error) { +// attemptTransforming runs one reset/apply/push cycle and reports what it +// produced. The returned baseSHA is the SHA the cycle was based on, so the +// caller can tell remote contention from a fatal failure; it is empty when the +// attempt failed before reset captured one. +func (m *gitMerger) attemptTransforming(ctx context.Context, req *runwaymq.MergeRequest, steps []resolvedStep, refs []changeRef, commit bool, tracked headBranchTracker) (*runwaymq.MergeResult, string, error) { + baseSHA, stepResults, heads, err := m.tryApply(ctx, steps) + + // Applying produced nothing, so the target already satisfies every step — + // this is a redelivery of work an earlier attempt landed. Freshness does not + // apply: there is nothing left for it to protect the target from. + if err == nil && !stepResultsHaveOutputs(stepResults) { + return successResult(req, stepResults), baseSHA, nil + } + + // Everything past here would change the target, so the request has to still + // be the current one. + if staleErr := m.checkStale(ctx, refs); staleErr != nil { + return nil, "", staleErr + } + if err != nil { + // The failing apply function aborts its own in-progress git operation; + // the next attempt starts with resetToRemote regardless. + return nil, baseSHA, err + } + + if !commit { + // Discard the local commits the dry run created so the checkout is + // clean for the next operation, and report empty Outputs. + if derr := m.resetToRemote(ctx); derr != nil { + return nil, "", fmt.Errorf("discard after dry-run: %w", derr) + } + stripOutputs(stepResults) + return successResult(req, stepResults), baseSHA, nil + } + + // The head branches move first, as their own push. A provider decides + // merged-versus-closed while processing the push to the target, against the + // head it has recorded at that moment, so a head that moves later — or in + // the same atomic push — is recorded too late. See headbranch.go. + if m.updateHeadBranch { + if err := m.updateHeadBranches(ctx, heads, tracked); err != nil { + return nil, baseSHA, err + } + } + if err := m.push(ctx); err != nil { + coremetrics.NamedCounter(m.metricsScope, "merge", "git_push_errors", 1) + return nil, baseSHA, err + } + return successResult(req, stepResults), baseSHA, nil +} + +// tryApply resets to the current target and applies every step locally. Remote +// writes remain with the caller so freshness can be checked after application +// but before any ref is updated. +func (m *gitMerger) tryApply(ctx context.Context, steps []resolvedStep) (string, []*runwaymq.StepResult, []headUpdate, error) { if err := m.resetToRemote(ctx); err != nil { coremetrics.NamedCounter(m.metricsScope, "merge", "reset_errors", 1) - return "", nil, err + return "", nil, nil, err } baseSHA, err := m.headSHA(ctx) if err != nil { - return "", nil, err + return "", nil, nil, err } stepResults, heads, err := m.applySteps(ctx, steps) if err != nil { // The failing apply function aborts its own in-progress git operation; // the next attempt starts with resetToRemote regardless. - return baseSHA, nil, err + return baseSHA, nil, nil, err } - - if commit { - // The head branches move first, as their own push. A provider decides - // merged-versus-closed while processing the push to the target, against - // the head it has recorded at that moment, so a head that moves later — - // or in the same atomic push — is recorded too late. See headbranch.go. - if m.updateHeadBranch { - if err := m.updateHeadBranches(ctx, heads, tracked); err != nil { - return baseSHA, nil, err - } - } - if err := m.push(ctx); err != nil { - coremetrics.NamedCounter(m.metricsScope, "merge", "git_push_errors", 1) - return baseSHA, nil, err - } - } - return baseSHA, stepResults, nil + return baseSHA, stepResults, heads, nil } // applied is what one step produced: the commits created on the target, and the @@ -714,17 +737,19 @@ func (m *gitMerger) promote(ctx context.Context, req *runwaymq.MergeRequest, rs sha := ref.SHA // PROMOTE does not go through tryApply, so it performs the same availability - // and freshness checks itself. Without them a commit the remote cannot - // supply turns every containment query into a plain error, which the - // consumer retries forever instead of reporting. They sit outside the retry - // loop because the commit under promotion is fixed for the whole request — - // only the target tip moves between attempts. + // check itself. Without it a commit the remote cannot supply turns every + // containment query into a plain error, which the consumer retries forever + // instead of reporting. if err := m.ensureObjects(ctx, []changeRef{ref}); err != nil { return nil, err } - if err := m.checkStale(ctx, []changeRef{ref}); err != nil { - return nil, err - } + + // Freshness is checked at most once, the first time the target is found not + // to already contain the commit. It cannot run before the containment check, + // which is what recognises an already-promoted redelivery; it need not run + // again per attempt, because the commit under promotion is fixed for the + // whole request and only the target tip moves between attempts. + freshnessChecked := false var lastErr error for attempt := 1; attempt <= m.maxPushAttempts; attempt++ { @@ -748,6 +773,13 @@ func (m *gitMerger) promote(ctx context.Context, req *runwaymq.MergeRequest, rs return promoteResult(req, rs, sha, commit), nil } + if !freshnessChecked { + if err := m.checkStale(ctx, []changeRef{ref}); err != nil { + return nil, err + } + freshnessChecked = true + } + // Only a true fast-forward is allowed; divergence is a terminal conflict. fastForward, err := m.isAncestor(ctx, tip, sha) if err != nil { @@ -1167,6 +1199,15 @@ func stripOutputs(steps []*runwaymq.StepResult) { } } +func stepResultsHaveOutputs(steps []*runwaymq.StepResult) bool { + for _, step := range steps { + if len(step.GetOutputs()) > 0 { + return true + } + } + return false +} + // successResult builds a SUCCEEDED MergeResult echoing the request id. func successResult(req *runwaymq.MergeRequest, steps []*runwaymq.StepResult) *runwaymq.MergeResult { return &runwaymq.MergeResult{ diff --git a/runway/extension/merger/git/git_merger_test.go b/runway/extension/merger/git/git_merger_test.go index b1e6e2224..87f76e41a 100644 --- a/runway/extension/merger/git/git_merger_test.go +++ b/runway/extension/merger/git/git_merger_test.go @@ -19,6 +19,7 @@ import ( "context" "errors" "fmt" + "net/url" "os" "os/exec" "path/filepath" @@ -374,8 +375,13 @@ func TestMerge_Rebase_RetriesWhenRemoteMovesUnderUs(t *testing.T) { featureSHA := f.pushPRCommit(t, "feature/a", "hello.txt", "hello\nearth\n", "tweak hello") f.installRaceHook(t, []string{raceSHA}) - m := f.newMerger(t, mergestrategypb.Strategy_REBASE) - res, err := m.Merge(context.Background(), req("b", stepOf(mergestrategypb.Strategy_REBASE, "s1", uri(featureSHA)))) + m := f.newMergerWith(t, func(p *Params) { + p.CheckStaleness = true + p.UpdateHeadBranch = true + }) + res, err := m.Merge(context.Background(), req("b", + stepOf(mergestrategypb.Strategy_REBASE, "s1", gitURI("feature/a", featureSHA)), + )) require.NoError(t, err) require.Len(t, res.GetSteps(), 1) require.Len(t, res.GetSteps()[0].GetOutputs(), 1) @@ -388,6 +394,7 @@ func TestMerge_Rebase_RetriesWhenRemoteMovesUnderUs(t *testing.T) { assert.Equal(t, raceSHA, commits[0], "race commit landed first via the hook") assert.Equal(t, res.GetSteps()[0].GetOutputs()[0].GetId(), commits[1], "our cherry-pick landed on top after the retry") + assert.Equal(t, res.GetSteps()[0].GetOutputs()[0].GetId(), f.remoteSHA(t, "feature/a")) assert.Equal(t, "hello\nearth\n", f.remoteFile(t, "hello.txt")) } @@ -1189,6 +1196,129 @@ func TestMerge_StaleChangeRejected(t *testing.T) { require.NoError(t, err) } +func TestMerge_StaleAlreadySatisfiedSucceeds(t *testing.T) { + tests := []struct { + name string + strategy mergestrategypb.Strategy + }{ + {name: "rebase", strategy: mergestrategypb.Strategy_REBASE}, + {name: "squash rebase", strategy: mergestrategypb.Strategy_SQUASH_REBASE}, + {name: "merge", strategy: mergestrategypb.Strategy_MERGE}, + {name: "promote", strategy: mergestrategypb.Strategy_PROMOTE}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + f := setupGitFixture(t) + stale := f.pushPRCommit(t, "feature/satisfied", "satisfied.txt", "satisfied\n", "satisfied") + switch tt.strategy { + case mergestrategypb.Strategy_REBASE, mergestrategypb.Strategy_SQUASH_REBASE: + f.landOnMain(t, stale) + case mergestrategypb.Strategy_MERGE, mergestrategypb.Strategy_PROMOTE: + f.advanceMain(t, stale) + } + mainBefore := f.remoteHEAD(t) + current := f.pushPRCommitFrom(t, stale, "feature/satisfied", "current.txt", "current\n", "current") + require.NotEqual(t, stale, current) + + m := f.newMergerWith(t, func(p *Params) { p.CheckStaleness = true }) + res, err := m.Merge(context.Background(), req("b", + stepOf(tt.strategy, "s1", gitURI("feature/satisfied", stale)), + )) + require.NoError(t, err) + assert.Equal(t, runwaypb.Outcome_SUCCEEDED, res.GetOutcome()) + assert.Equal(t, mainBefore, f.remoteHEAD(t)) + }) + } +} + +func TestMerge_StalePendingChangeRejected(t *testing.T) { + tests := []struct { + name string + strategy mergestrategypb.Strategy + }{ + {name: "rebase", strategy: mergestrategypb.Strategy_REBASE}, + {name: "squash rebase", strategy: mergestrategypb.Strategy_SQUASH_REBASE}, + {name: "merge", strategy: mergestrategypb.Strategy_MERGE}, + {name: "promote", strategy: mergestrategypb.Strategy_PROMOTE}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + f := setupGitFixture(t) + base := f.remoteHEAD(t) + stale := f.pushPRCommitFrom(t, base, "feature/pending", "stale.txt", "stale\n", "stale") + mustGit(t, f.authorDir, "push", "origin", stale+":refs/heads/archive/stale") + current := f.pushPRCommitFrom(t, base, "feature/pending", "current.txt", "current\n", "current") + require.NotEqual(t, stale, current) + + m := f.newMergerWith(t, func(p *Params) { p.CheckStaleness = true }) + _, err := m.Merge(context.Background(), req("b", + stepOf(tt.strategy, "s1", gitURI("feature/pending", stale)), + )) + require.Error(t, err) + assert.True(t, errors.Is(err, merger.ErrInvalidRequest)) + assert.Equal(t, base, f.remoteHEAD(t)) + }) + } +} + +// A delivery that moved a change's head branch and then failed to push the +// target leaves the branch on a commit the merger made and the target where it +// was. The redelivery still pins the original commit, so the branch no longer +// answers to it — but the change is not superseded, and rejecting it as stale +// would report an error for work that was always meant to land. +func TestMerge_HeadBranchMovedByEarlierDelivery_IsNotStale(t *testing.T) { + f := setupGitFixture(t) + head := f.pushPRCommit(t, "feature/redeliver", "r.txt", "r\n", "add r") + mainBefore := f.remoteHEAD(t) + + m := f.newMergerWith(t, func(p *Params) { + p.CheckStaleness = true + p.UpdateHeadBranch = true + }) + request := req("b", stepOf(mergestrategypb.Strategy_REBASE, "s1", gitURI("feature/redeliver", head))) + + f.installRefRejectHook(t, "refs/heads/main") + _, err := m.Merge(context.Background(), request) + require.Error(t, err) + assert.False(t, errors.Is(err, merger.ErrInvalidRequest), "a rejected target push is not a bad request") + assert.Equal(t, mainBefore, f.remoteHEAD(t), "the target never moved") + movedTo := f.remoteSHA(t, "feature/redeliver") + require.NotEqual(t, head, movedTo, "but the head branch did") + + f.removeHooks(t) + res, err := m.Merge(context.Background(), request) + require.NoError(t, err) + assert.Equal(t, runwaypb.Outcome_SUCCEEDED, res.GetOutcome()) + landed := f.remoteHEAD(t) + assert.NotEqual(t, mainBefore, landed, "the redelivery landed the change") + assert.Equal(t, "r\n", f.remoteFile(t, "r.txt")) + assert.Equal(t, landed, f.remoteSHA(t, "feature/redeliver"), + "and moved the head branch on from where the first attempt stranded it") +} + +// The same shape, but the branch moved because its author pushed. That is a +// real supersession and must still be rejected. +func TestMerge_HeadBranchMovedByAuthor_IsStale(t *testing.T) { + f := setupGitFixture(t) + base := f.remoteHEAD(t) + stale := f.pushPRCommitFrom(t, base, "feature/authored", "a.txt", "a\n", "a") + current := f.pushPRCommitFrom(t, base, "feature/authored", "b.txt", "b\n", "b") + require.NotEqual(t, stale, current) + + m := f.newMergerWith(t, func(p *Params) { + p.CheckStaleness = true + p.UpdateHeadBranch = true + }) + _, err := m.Merge(context.Background(), req("b", + stepOf(mergestrategypb.Strategy_REBASE, "s1", gitURI("feature/authored", stale)), + )) + require.Error(t, err) + assert.True(t, errors.Is(err, merger.ErrInvalidRequest)) + assert.Equal(t, base, f.remoteHEAD(t)) +} + func TestMerge_StalenessCheckOffByDefault(t *testing.T) { f := setupGitFixture(t) stale := f.pushPRCommit(t, "feature/s", "s.txt", "v1\n", "v1") @@ -1603,6 +1733,10 @@ func uri(sha string) string { return fmt.Sprintf("github://github.example.com/uber/submitqueue/pull/1/%s", sha) } +func gitURI(branch, sha string) string { + return fmt.Sprintf("git://git.example.com/uber/submitqueue/%s/%s", url.PathEscape("refs/heads/"+branch), sha) +} + // installRaceHook writes a pre-receive hook on the bare remote that simulates // concurrent pushes. On its Nth invocation it reads the Nth line of race-shas, // points refs/heads/main at that SHA via update-ref, and exits 1 (rejecting the @@ -1682,6 +1816,13 @@ exit 0 require.NoError(t, os.WriteFile(hookPath, []byte(script), 0o755)) } +// removeHooks clears the pre-receive hook installed on the bare remote, so a +// later push in the same test is allowed through. +func (f gitFixture) removeHooks(t *testing.T) { + t.Helper() + require.NoError(t, os.Remove(filepath.Join(f.remoteDir, "hooks", "pre-receive"))) +} + // hookInvocations returns the number of times the pre-receive race hook has // fired. Used by retry tests to verify the loop ran the expected number of // attempts. diff --git a/runway/extension/merger/git/headbranch.go b/runway/extension/merger/git/headbranch.go index 49c54c456..2ad18de46 100644 --- a/runway/extension/merger/git/headbranch.go +++ b/runway/extension/merger/git/headbranch.go @@ -139,6 +139,14 @@ func (m *gitMerger) updateHeadBranchFor(ctx context.Context, u headUpdate, tips case 1: branch, lease = candidates[0], u.ref.SHA case 0: + // An earlier delivery that moved the branch and then failed to push + // the target leaves nothing at the pinned SHA, so the tips cannot + // name it and the tracker is empty in a fresh process. The URI names + // the branch directly, which is enough to move it on. + if moved, tip, ok := m.headBranchMovedByMerger(ctx, u.ref); ok { + branch, lease = moved, tip + break + } // Nothing on this remote is at the change's head. The ordinary cause is // a change proposed from a fork, whose head branch lives in another // repository entirely and is not ours to move; a deleted or already @@ -206,3 +214,31 @@ func (m *gitMerger) remoteBranchTips(ctx context.Context) (map[string][]string, } return tips, nil } + +// headBranchMovedByMerger resolves a change's own head branch and the commit it +// currently sits on, for the case where it has already been moved off the SHA +// the URI pinned. +// +// It answers only when the branch is on a commit this merger made. A branch its +// author advanced is a change that moved on, not one a previous attempt got +// half-way through, and moving it would discard their push — the lease would +// refuse it anyway, but declining here keeps the reason legible. +func (m *gitMerger) headBranchMovedByMerger(ctx context.Context, ref changeRef) (branch, tip string, ok bool) { + if !strings.HasPrefix(ref.Ref, headBranchPrefix) { + return "", "", false + } + out, err := m.run(ctx, nil, "ls-remote", m.remote, ref.Ref) + if err != nil { + return "", "", false + } + fields := strings.Fields(string(out)) + if len(fields) == 0 { + return "", "", false + } + current := fields[0] + if !m.isOwnCommit(ctx, current) { + return "", "", false + } + coremetrics.NamedCounter(m.metricsScope, "head_branch", "resumed", 1) + return ref.Ref, current, true +} diff --git a/runway/extension/merger/git/objects.go b/runway/extension/merger/git/objects.go index 3a919e5c5..434db0b36 100644 --- a/runway/extension/merger/git/objects.go +++ b/runway/extension/merger/git/objects.go @@ -110,6 +110,16 @@ func (m *gitMerger) checkStale(ctx context.Context, refs []changeRef) error { continue } if current := fields[0]; current != ref.SHA { + // The merger moves a change's head branch itself, to the commit it + // made of that change. Finding the branch on one of those commits + // means an earlier attempt got that far, not that the author + // replaced the change, so it is not evidence of staleness. The + // attempt may have been an earlier delivery of this same message, + // which is why the committer is read from the commit rather than + // remembered in the in-flight head-branch tracker. + if m.isOwnCommit(ctx, current) { + continue + } coremetrics.NamedCounter(m.metricsScope, "merge", "stale_changes", 1) return fmt.Errorf("%w: change is stale: %s names commit %s but %s now points at %s", merger.ErrInvalidRequest, ref.Label, ref.SHA, ref.Ref, current) @@ -117,3 +127,27 @@ func (m *gitMerger) checkStale(ctx context.Context, refs []changeRef) error { } return nil } + +// isOwnCommit reports whether this merger created the commit at sha. Every git +// command the merger runs injects its committer identity, so that identity is +// the durable record of authorship — it survives the process that wrote it, +// unlike anything held in memory for the length of one request. +// +// A commit that cannot be resolved is not ours as far as this reports: the +// caller's fallback is to treat the change as superseded, which is the safe +// direction. +func (m *gitMerger) isOwnCommit(ctx context.Context, sha string) bool { + if !m.hasCommit(ctx, sha) { + if _, err := m.run(ctx, nil, "fetch", m.remote, sha); err != nil { + return false + } + if !m.hasCommit(ctx, sha) { + return false + } + } + out, err := m.run(ctx, nil, "show", "-s", "--format=%ce", sha) + if err != nil { + return false + } + return strings.TrimSpace(string(out)) == m.committerEmail +} diff --git a/test/e2e/submitqueue/ISS-001.md b/test/e2e/submitqueue/ISS-001.md new file mode 100644 index 000000000..583c686c8 --- /dev/null +++ b/test/e2e/submitqueue/ISS-001.md @@ -0,0 +1,73 @@ +# ISS-001: stale retry after a successful Git push + +## Reproduction + +The regression test uses the real Git-backed SubmitQueue E2E stack with `checkStaleness: true` and `updateHeadBranch: true`. + +Initial Git refs: + +```text +refs/heads/main = M0 +refs/heads/feature/retry-result = A +``` + +SubmitQueue eventually publishes a `runway-merge` message whose `id` is the batch ID and whose change URI pins the feature branch to `A`: + +```json +{ + "id": "e2e-git-queue/batch/1", + "queue_name": "e2e-git-queue", + "steps": [{ + "change": { + "uris": ["git://git.example.com/sandbox/refs%2Fheads%2Ffeature%2Fretry-result/A"] + }, + "strategy": "REBASE" + }] +} +``` + +The test closes the `runway-merge` consumer gate until it can read the exact batch ID from the application database's `request_batch` table. It then seeds a batch-scoped MyISAM guard row and installs a `BEFORE INSERT` trigger on the queue database's `queue_messages` table. + +The trigger counts every attempted `SUCCEEDED` publication for the batch and rejects only the first one. It decrements the MyISAM guard before raising MySQL error 1213, so the counter changes survive the failed InnoDB insert and the next successful result is allowed. + +On the first delivery, Runway rebases `A` into `A'` and updates both refs before result publication: + +```text +refs/heads/main = A' +refs/heads/feature/retry-result = A' +``` + +The injected publication failure causes the original `runway-merge` delivery to be retried. Its unchanged URI still pins `A`, while the feature ref now points to `A'`. + +Before the fix, Runway rejects this retry as stale and publishes `FAILED`, leaving the contradictory state: + +```text +Git main: A' (the code landed) +SubmitQueue request: error +``` + +After the fix, Runway applies the request locally against current `main` before enforcing staleness. The application is a no-op because `A` is already represented by `A'`, so Runway publishes `SUCCEEDED` without another remote write. The final state converges: + +```text +Git main: A' +feature branch: A' +SubmitQueue request: landed +successful-result attempts: 2 +``` + +The two result-publication attempts are the durable witness that `runway-merge` was delivered twice. The queue's acknowledged message and delivery-state rows may be garbage-collected before the terminal request status is observable. + +## Second shape: the head branch moved, the target did not + +The rebase pushes a change's head branch before it pushes the target, so a delivery can end after the first push and before the second — the process is killed, or the target push is rejected. What the redelivery finds is: + +```text +refs/heads/main = M0 (unchanged) +refs/heads/feature/… = A' (moved by the lost delivery) +``` + +The request still pins `A`, which the branch has left, so the freshness check sees a change that looks superseded. Applying it locally is not a no-op either — the target really does not have it — so the early-success path above does not cover this. + +What separates the two cases is who moved the branch. Runway records its own committer identity (`SubmitQueue Runway `) on every commit it makes, so a head branch sitting on a commit Runway committed was moved by an earlier attempt, not by the change's author. That identity is read back off the commit rather than remembered in memory, which is what makes it survive the delivery that wrote it. A branch the author moved carries the author's identity and is still rejected as stale. + +`TestLand_HeadBranchMovedByLostDelivery_StillLands` sets up that ref state directly — the lost delivery is reproduced by its effect, since what the redelivery sees is the refs — and asserts the change lands instead of erroring. diff --git a/test/e2e/submitqueue/git_suite_test.go b/test/e2e/submitqueue/git_suite_test.go index 321051ec0..492cf02c3 100644 --- a/test/e2e/submitqueue/git_suite_test.go +++ b/test/e2e/submitqueue/git_suite_test.go @@ -38,12 +38,16 @@ import ( "path/filepath" "strings" "testing" + "time" "github.com/stretchr/testify/require" "github.com/stretchr/testify/suite" changepb "github.com/uber/submitqueue/api/base/change/protopb" mergestrategypb "github.com/uber/submitqueue/api/base/mergestrategy/protopb" + runwaymq "github.com/uber/submitqueue/api/runway/messagequeue" gatewaypb "github.com/uber/submitqueue/api/submitqueue/gateway/protopb" + "github.com/uber/submitqueue/platform/extension/consumergate" + consumergatefile "github.com/uber/submitqueue/platform/extension/consumergate/file" gitexectest "github.com/uber/submitqueue/platform/git/exectest" "github.com/uber/submitqueue/submitqueue/entity" "github.com/uber/submitqueue/test/testutil" @@ -54,6 +58,14 @@ import ( // service/submitqueue/demo/provider/git/merge.yaml. const gitQueue = "e2e-git-queue" +// runwayCommitter is the identity Runway records as committer on every commit +// it makes. The merger reads it back off a change's head branch to tell a +// branch it moved itself from one the change's author updated. +const ( + runwayCommitterName = "SubmitQueue Runway" + runwayCommitterEmail = "runway@submitqueue.invalid" +) + // sandboxRemote is the host name used in git:// change URIs. The merger reads // the commit and ref out of the URI and reaches the repository through its own // configured remote, so this identifies the change rather than routing to it. @@ -67,6 +79,7 @@ type GitMergeSuite struct { gatewayClient gatewaypb.SubmitQueueGatewayClient db *sql.DB queueDB *sql.DB + gate *consumergatefile.Store // git is the pinned git the test drives the bare repository with — the same // build the merger uses inside the container. @@ -89,7 +102,9 @@ func (s *GitMergeSuite) SetupSuite() { s.git = gitexectest.Git(t) containerUser := dockerContainerUser(t) t.Setenv("SQ_CONTAINER_USER", containerUser) - t.Setenv("SQ_CONSUMER_GATE_DIR", t.TempDir()) + gateDir := t.TempDir() + t.Setenv("SQ_CONSUMER_GATE_DIR", gateDir) + s.gate = consumergatefile.New(gateDir) // The bare repository lives in a host directory bind-mounted into Runway, so // the test can seed it and read back exactly what the merger pushed. @@ -218,6 +233,63 @@ func (s *GitMergeSuite) TestLand_MovesEachChangeHeadBranchToItsLandedCommit() { s.True(s.isAncestorOfMain(s.branchSHA("feature/head-2"))) } +func (s *GitMergeSuite) TestLand_RetryAfterResultPublishFailure_ReconcilesLandedState() { + const consumerGroup = "runway-merge" + gateKey := consumergate.Key{ConsumerGroup: consumerGroup, PartitionKey: gitQueue} + require.NoError(s.T(), s.gate.Close(s.ctx, gateKey, consumergate.Metadata{ + Reason: "install a batch-scoped fail-once result publisher before merge", + CreatedBy: "GitMergeSuite", + CreatedAtMs: time.Now().UnixMilli(), + })) + defer func() { + require.NoError(s.T(), s.gate.Open(s.ctx, gateKey)) + }() + + before := s.mainSHA() + head := s.pushChange("feature/retry-result", map[string]string{"retry-result.txt": "retry result\n"}, "add retry result") + sqid := s.land(gitQueue, s.uri("feature/retry-result", head)) + batchID := s.awaitBatchID(sqid) + s.awaitParkedMerge(batchID) + s.installMergeSignalFailOnce(batchID) + + require.NoError(s.T(), s.gate.Open(s.ctx, gateKey)) + s.requireStatus(sqid, entity.RequestStatusLanded) + + landed := s.mainSHA() + s.NotEqual(before, landed) + s.NotEqual(head, landed) + s.Equal("retry result\n", s.fileOnMain("retry-result.txt")) + s.Equal(landed, s.branchSHA("feature/retry-result")) + s.Equal(2, s.mergeSignalPublishAttemptCount(batchID), + "two result publications prove the runway-merge delivery was retried") +} + +// A delivery that moved a change's head branch and then died before pushing the +// target leaves the branch on a commit Runway made, with the target where it +// was. The redelivery still pins the commit the branch has since left, so the +// change looks superseded while in fact nothing but Runway touched it. +// +// The lost delivery is reproduced by its effect rather than by killing Runway +// mid-push: what the redelivery sees is the ref state, and that is what this +// sets up. +func (s *GitMergeSuite) TestLand_HeadBranchMovedByLostDelivery_StillLands() { + before := s.mainSHA() + head := s.pushChange("feature/lost-delivery", map[string]string{"lost.txt": "lost\n"}, "add lost") + updates := s.mainRefUpdateCount() + + movedTo := s.replayAsRunway("feature/lost-delivery", head) + s.NotEqual(head, movedTo, "the head branch moved to the commit Runway made of the change") + s.Equal(before, s.mainSHA(), "and the target never moved") + s.Equal(updates, s.mainRefUpdateCount()) + + s.requireStatus(s.land(gitQueue, s.uri("feature/lost-delivery", head)), entity.RequestStatusLanded) + + landed := s.mainSHA() + s.NotEqual(before, landed, "the change landed rather than being rejected as stale") + s.Equal("lost\n", s.fileOnMain("lost.txt")) + s.Equal(landed, s.branchSHA("feature/lost-delivery")) +} + func (s *GitMergeSuite) TestLand_Conflict_FailsAndLeavesTheTargetUntouched() { // Two changes editing the same line from the same base: the first lands, // the second cannot be replayed onto it. @@ -236,11 +308,7 @@ func (s *GitMergeSuite) TestLand_Conflict_FailsAndLeavesTheTargetUntouched() { s.Equal(loser, s.branchSHA("feature/conflict-b")) } -func (s *GitMergeSuite) TestLand_ResubmittedAfterLanding_IsRejectedAsStale() { - // Landing a change moves its head branch to the commit it became, so the - // URI that was submitted no longer describes where that branch points. The - // staleness check catches exactly that, which is what stops a change from - // being replayed onto the target a second time. +func (s *GitMergeSuite) TestLand_ResubmittedAfterLanding_IsSuccessfulNoOp() { head := s.pushChange("feature/already", map[string]string{"already.txt": "already\n"}, "add already") s.requireStatus(s.land(gitQueue, s.uri("feature/already", head)), entity.RequestStatusLanded) @@ -249,8 +317,8 @@ func (s *GitMergeSuite) TestLand_ResubmittedAfterLanding_IsRejectedAsStale() { landedAs := s.branchSHA("feature/already") s.NotEqual(head, landedAs, "the head branch moved to the landed commit") - s.requireStatus(s.land(gitQueue, s.uri("feature/already", head)), entity.RequestStatusError) - s.Equal(settled, s.mainSHA(), "a stale resubmission must not move the target") + s.requireStatus(s.land(gitQueue, s.uri("feature/already", head)), entity.RequestStatusLanded) + s.Equal(settled, s.mainSHA(), "an already-satisfied resubmission must not move the target") s.Equal(updates, s.mainRefUpdateCount(), "and must not push at all") s.Equal(landedAs, s.branchSHA("feature/already"), "nor disturb the change's branch") } @@ -286,6 +354,101 @@ func (s *GitMergeSuite) requireStatus(sqid string, want entity.RequestStatus) { s.Require().Equal(want, got, "request %s reached the wrong terminal status", sqid) } +func (s *GitMergeSuite) awaitBatchID(sqid string) string { + var batchID string + pollUntil(persistPollInterval, func() bool { + err := s.db.QueryRowContext(s.ctx, + "SELECT batch_id FROM request_batch WHERE queue = ? AND request_id = ? ORDER BY batch_id LIMIT 1", + gitQueue, sqid, + ).Scan(&batchID) + return err == nil + }) + return batchID +} + +func (s *GitMergeSuite) awaitParkedMerge(batchID string) { + pollUntil(persistPollInterval, func() bool { + parked, err := s.gate.ListParked(s.ctx, "runway-merge") + if err != nil { + return false + } + for _, delivery := range parked { + if delivery.Topic == runwaymq.TopicKeyMerge.String() && delivery.MessageID == batchID { + return true + } + } + return false + }) +} + +// installMergeSignalFailOnce makes the first SUCCEEDED merge-signal publication +// for a batch fail, so the merge delivery is redelivered with Git already +// updated. That is the only way, from outside the process, to stop a controller +// between two of its own steps. +// +// Doing it with a trigger ties the helper to the MySQL queue backend: it needs a +// server that supports triggers, a non-transactional table so the counters +// survive the rolled-back insert, and SIGNAL to raise an error the publisher +// treats as retryable. A queue on any other backend would need its own +// injection mechanism, and this test would not run against it. +// +// TODO(e2e): give the harness a backend-independent way to fail a controller +// partway — a fault-injection seam the consumer honours — so a test like this +// stops depending on the queue's storage engine. +func (s *GitMergeSuite) installMergeSignalFailOnce(messageID string) { + t := s.T() + const ( + triggerName = "e2e_fail_merge_signal_once" + tableName = "e2e_merge_signal_fail_once" + ) + _, err := s.queueDB.ExecContext(s.ctx, "DROP TRIGGER IF EXISTS "+triggerName) + require.NoError(t, err) + _, err = s.queueDB.ExecContext(s.ctx, "DROP TABLE IF EXISTS "+tableName) + require.NoError(t, err) + _, err = s.queueDB.ExecContext(s.ctx, "CREATE TABLE "+tableName+" (message_id VARCHAR(191) CHARACTER SET ascii PRIMARY KEY, remaining INT NOT NULL, attempts INT NOT NULL) ENGINE=MyISAM") + require.NoError(t, err) + _, err = s.queueDB.ExecContext(s.ctx, "INSERT INTO "+tableName+" (message_id, remaining, attempts) VALUES (?, 1, 0)", messageID) + require.NoError(t, err) + _, err = s.queueDB.ExecContext(s.ctx, ` +CREATE TRIGGER `+triggerName+` +BEFORE INSERT ON queue_messages +FOR EACH ROW +BEGIN + IF NEW.topic = 'merge-signal' + AND JSON_UNQUOTE(JSON_EXTRACT(CONVERT(NEW.payload USING utf8mb4), '$.outcome')) = 'SUCCEEDED' + THEN + UPDATE `+tableName+` + SET attempts = attempts + 1 + WHERE message_id = NEW.id; + UPDATE `+tableName+` + SET remaining = remaining - 1 + WHERE message_id = NEW.id AND remaining > 0; + IF ROW_COUNT() > 0 THEN + SIGNAL SQLSTATE '40001' + SET MYSQL_ERRNO = 1213, + MESSAGE_TEXT = 'injected fail-once merge-signal publication failure'; + END IF; + END IF; +END`) + require.NoError(t, err) + t.Cleanup(func() { + _, dropTriggerErr := s.queueDB.ExecContext(s.ctx, "DROP TRIGGER IF EXISTS "+triggerName) + require.NoError(t, dropTriggerErr) + _, dropTableErr := s.queueDB.ExecContext(s.ctx, "DROP TABLE IF EXISTS "+tableName) + require.NoError(t, dropTableErr) + }) +} + +func (s *GitMergeSuite) mergeSignalPublishAttemptCount(batchID string) int { + var attempts int + err := s.queueDB.QueryRowContext(s.ctx, + "SELECT attempts FROM e2e_merge_signal_fail_once WHERE message_id = ?", + batchID, + ).Scan(&attempts) + s.Require().NoError(err) + return attempts +} + // uri builds the git:// change URI for a branch pinned at a commit. The ref is // percent-encoded so a branch name containing slashes stays one path segment. func (s *GitMergeSuite) uri(branch, sha string) string { @@ -361,6 +524,24 @@ func (s *GitMergeSuite) pushChangeOnto(base, branch string, files map[string]str } // mainSHA is the current tip of the target branch on the bare repository. +// replayAsRunway replays a change onto the current target under Runway's +// committer identity and moves the change's branch there, without touching the +// target. That is exactly the state a delivery leaves behind when it updates +// head branches and then fails before pushing the target. +func (s *GitMergeSuite) replayAsRunway(branch, sha string) string { + s.runGit(s.work, "fetch", "origin") + s.runGit(s.work, "checkout", "-B", "runway-replay", "origin/main") + s.runGit(s.work, + "-c", "user.name="+runwayCommitterName, + "-c", "user.email="+runwayCommitterEmail, + "cherry-pick", sha, + ) + replayed := strings.TrimSpace(s.runGit(s.work, "rev-parse", "HEAD")) + s.runGit(s.work, "push", "--force", "origin", replayed+":refs/heads/"+branch) + s.runGit(s.work, "checkout", "main") + return replayed +} + func (s *GitMergeSuite) mainSHA() string { return s.runGit(s.bare, "rev-parse", "refs/heads/main") } @@ -424,6 +605,7 @@ func (s *GitMergeSuite) runGit(dir string, args ...string) string { s.T().Helper() cmd := exec.Command(s.git, args...) cmd.Dir = dir + cmd.Env = append(os.Environ(), "GIT_EXEC_PATH="+filepath.Dir(s.git)) var stderr strings.Builder cmd.Stderr = &stderr out, err := cmd.Output()