fix(controller): git-push step: pull --rebase before push

docs/docs/35-references/10-promotion-steps.md

@@ -924,14 +924,32 @@ steps:
 ### `git-push`
 
 `git-push` pushes the committed changes in a specified working tree to a
-specified branch in the remote repository. This step typically follows a `git-commit` step and is often followed by a `git-open-pr` step.
+specified branch in the remote repository. This step typically follows a
+`git-commit` step and is often followed by a `git-open-pr` step.
+
+This step also implements its own, internal retry logic. If a push fails, with
+the cause determined to be the presence of new commits in the remote branch that
+are not present in the local branch, the step will attempt to rebase before
+retrying the push. Any merge conflict requiring manual resolution will
+immediately halt further attempts.
+
+:::info
+This step's internal retry logic is helpful in scenarios when concurrent
+Promotions to multiple Stages may all write to the same branch of the same
+repository.
+
+Because conflicts requiring manual resolution will halt further attempts, it is
+recommended to design your Promotion processes such that Promotions to multiple
+Stages that write to the same branch do not write to the same files.
+:::
 
 #### `git-push` Configuration
 
 | Name | Type | Required | Description |
 |------|------|----------|-------------|
 | `path` | `string` | Y | Path to a Git working tree containing committed changes. |
 | `targetBranch` | `string` | N | The branch to push to in the remote repository. Mutually exclusive with `generateTargetBranch=true`. If neither of these is provided, the target branch will be the same as the branch currently checked out in the working tree. |
+| `maxAttempts` | `int32` | N | The maximum number of attempts to make when pushing to the remote repository. Default is 50. |
 | `generateTargetBranch` | `boolean` | N | Whether to push to a remote branch named like `kargo/<project>/<stage>/promotion`. If such a branch does not already exist, it will be created. A value of 'true' is mutually exclusive with `targetBranch`. If neither of these is provided, the target branch will be the currently checked out branch. This option is useful when a subsequent promotion step will open a pull request against a Stage-specific branch. In such a case, the generated target branch pushed to by the `git-push` step can later be utilized as the source branch of the pull request. |
 
 #### `git-push` Examples

internal/controller/git/errors.go

@@ -0,0 +1,24 @@
+package git
+
+import (
+	"errors"
+)
+
+// ErrMergeConflict is returned when a merge conflict occurs.
+var ErrMergeConflict = errors.New("merge conflict")
+
+// IsMergeConflict returns true if the error is a merge conflict or wraps one
+// and false otherwise.
+func IsMergeConflict(err error) bool {
+	return errors.Is(err, ErrMergeConflict)
+}
+
+// ErrNonFastForward is returned when a push is rejected because it is not a
+// fast-forward or needs to be fetched first.
+var ErrNonFastForward = errors.New("non-fast-forward")
+
+// IsNonFastForward returns true if the error is a non-fast-forward or wraps one
+// and false otherwise.
+func IsNonFastForward(err error) bool {
+	return errors.Is(err, ErrNonFastForward)
+}

internal/controller/git/errors_test.go

@@ -0,0 +1,79 @@
+package git
+
+import (
+	"errors"
+	"fmt"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestIsMergeConflict(t *testing.T) {
+	testCases := []struct {
+		name     string
+		err      error
+		expected bool
+	}{
+		{
+			name:     "nil error",
+			err:      nil,
+			expected: false,
+		},
+		{
+			name:     "not a merge conflict",
+			err:      errors.New("something went wrong"),
+			expected: false,
+		},
+		{
+			name:     "a merge conflict",
+			err:      ErrMergeConflict,
+			expected: true,
+		},
+		{
+			name:     "a wrapped merge conflict",
+			err:      fmt.Errorf("an error occurred: %w", ErrMergeConflict),
+			expected: true,
+		},
+	}
+	for _, testCase := range testCases {
+		t.Run(testCase.name, func(t *testing.T) {
+			actual := IsMergeConflict(testCase.err)
+			require.Equal(t, testCase.expected, actual)
+		})
+	}
+}
+
+func TestIsNonFastForward(t *testing.T) {
+	testCases := []struct {
+		name     string
+		err      error
+		expected bool
+	}{
+		{
+			name:     "nil error",
+			err:      nil,
+			expected: false,
+		},
+		{
+			name:     "not a non-fast-forward error",
+			err:      errors.New("something went wrong"),
+			expected: false,
+		},
+		{
+			name:     "a non-fast-forward error",
+			err:      ErrNonFastForward,
+			expected: true,
+		},
+		{
+			name:     "a wrapped fast forward error",
+			err:      fmt.Errorf("an error occurred: %w", ErrNonFastForward),
+			expected: true,
+		},
+	}
+	for _, testCase := range testCases {
+		t.Run(testCase.name, func(t *testing.T) {
+			actual := IsNonFastForward(testCase.err)
+			require.Equal(t, testCase.expected, actual)
+		})
+	}
+}

internal/controller/git/work_tree.go

@@ -6,6 +6,8 @@
 	"errors"
 	"fmt"
 	"os"
+	"path/filepath"
+	"regexp"
 	"strings"
 	"time"
 
@@ -57,6 +59,9 @@
 	GetDiffPathsForCommitID(commitID string) ([]string, error)
 	// IsAncestor returns true if parent branch is an ancestor of child
 	IsAncestor(parent string, child string) (bool, error)
+	// IsRebasing returns a bool indicating whether the working tree is currently
+	// in the middle of a rebase operation.
+	IsRebasing() (bool, error)
 	// LastCommitID returns the ID (sha) of the most recent commit to the current
 	// branch.
 	LastCommitID() (string, error)
@@ -316,6 +321,31 @@
 	return false, fmt.Errorf("error testing ancestry of branches %q, %q: %w", parent, child, err)
 }
 
+func (w *workTree) IsRebasing() (bool, error) {
+	res, err := libExec.Exec(w.buildGitCommand("rev-parse", "--git-path", "rebase-merge"))
+	if err != nil {
+		return false, fmt.Errorf("error determining rebase status: %w", err)
+	}
+	rebaseMerge := filepath.Join(w.dir, strings.TrimSpace(string(res)))
+	if _, err = os.Stat(rebaseMerge); !os.IsNotExist(err) {
+		if err != nil {
+			return false, err
+		}
+		return true, nil
+	}
+	if res, err = libExec.Exec(w.buildGitCommand("rev-parse", "--git-path", "rebase-apply")); err != nil {
+		return false, fmt.Errorf("error determining rebase status: %w", err)
+	}
+	rebaseApply := filepath.Join(w.dir, strings.TrimSpace(string(res)))
+	if _, err = os.Stat(rebaseApply); !os.IsNotExist(err) {
+		if err != nil {
+			return false, err
+		}
+		return true, nil
+	}
+	return false, nil
+}
+
 func (w *workTree) LastCommitID() (string, error) {
 	shaBytes, err := libExec.Exec(w.buildGitCommand("rev-parse", "HEAD"))
 	if err != nil {
@@ -481,22 +511,56 @@
 	// TargetBranch specifies the branch to push to. If empty, the current branch
 	// will be pushed to a remote branch by the same name.
 	TargetBranch string
+	// PullRebase indicates whether to pull and rebase before pushing. This can
+	// be useful when pushing changes to a remote branch that has been updated
+	// in the time since the local branch was last pulled.
+	PullRebase bool
 }
 
+// https://regex101.com/r/aNYjHP/1
+//
+// nolint: lll
+var nonFastForwardRegex = regexp.MustCompile(`(?m)^\s*!\s+\[(?:remote )?rejected].+\((?:non-fast-forward|fetch first|cannot lock ref.*)\)\s*$`)
+
 func (w *workTree) Push(opts *PushOptions) error {
 	if opts == nil {
 		opts = &PushOptions{}
 	}
-	args := []string{"push", "origin"}
-	if opts.TargetBranch != "" {
-		args = append(args, fmt.Sprintf("HEAD:%s", opts.TargetBranch))
-	} else {
-		args = append(args, "HEAD")
+	targetBranch := opts.TargetBranch
+	if targetBranch == "" {
+		var err error
+		if targetBranch, err = w.CurrentBranch(); err != nil {
+			return err
+		}
+	}
+	if opts.PullRebase {
+		exists, err := w.RemoteBranchExists(targetBranch)
+		if err != nil {
+			return err
+		}
+		// We only want to pull and rebase if the remote branch exists.
+		if exists {
+			if _, err = libExec.Exec(w.buildGitCommand("pull", "--rebase", "origin", targetBranch)); err != nil {
+				// The error we're most concerned with is a merge conflict requiring
+				// manual resolution, because it's an error that no amount of retries
+				// will fix. If we find that a rebase is in progress, this is what
+				// has happened.
+				if isRebasing, isRebasingErr := w.IsRebasing(); isRebasingErr == nil && isRebasing {
+					return ErrMergeConflict
+				}
+				// If we get to here, the error isn't a merge conflict.
+				return fmt.Errorf("error pulling and rebasing branch: %w", err)
+			}
+		}
 	}
+	args := []string{"push", "origin", fmt.Sprintf("HEAD:%s", targetBranch)}
 	if opts.Force {
 		args = append(args, "--force")
 	}
-	if _, err := libExec.Exec(w.buildGitCommand(args...)); err != nil {
+	if res, err := libExec.Exec(w.buildGitCommand(args...)); err != nil {
+		if nonFastForwardRegex.MatchString(string(res)) {
+			return fmt.Errorf("error pushing branch: %w", ErrNonFastForward)
+		}
 		return fmt.Errorf("error pushing branch: %w", err)
 	}
 	return nil

internal/directives/errors.go

@@ -0,0 +1,24 @@
+package directives
+
+import "errors"
+
+// terminalError wraps another error to indicate to the step execution engine
+// that the step that produced the error should not be retried.
+type terminalError struct {
+	err error
+}
+
+// Error implements the error interface.
+func (e *terminalError) Error() string {
+	if e.err == nil {
+		return ""
+	}
+	return e.err.Error()
+}
+
+// isTerminal returns true if the error is a terminal error or wraps one and
+// false otherwise.
+func isTerminal(err error) bool {
+	te := &terminalError{}
+	return errors.As(err, &te)
+}

internal/directives/errors_test.go

@@ -0,0 +1,47 @@
+package directives
+
+import (
+	"errors"
+	"fmt"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestIsTerminal(t *testing.T) {
+	testCases := []struct {
+		name     string
+		err      error
+		expected bool
+	}{
+		{
+			name:     "nil error",
+			err:      nil,
+			expected: false,
+		},
+		{
+			name:     "not a terminal error",
+			err:      errors.New("something went wrong"),
+			expected: false,
+		},
+		{
+			name:     "a terminal error",
+			err:      &terminalError{err: errors.New("something went wrong")},
+			expected: true,
+		},
+		{
+			name: "a wrapped terminal error",
+			err: fmt.Errorf(
+				"an error occurred: %w",
+				&terminalError{err: errors.New("something went wrong")},
+			),
+			expected: true,
+		},
+	}
+	for _, testCase := range testCases {
+		t.Run(testCase.name, func(t *testing.T) {
+			actual := isTerminal(testCase.err)
+			require.Equal(t, testCase.expected, actual)
+		})
+	}
+}

internal/directives/git_pr_waiter.go

@@ -114,10 +114,8 @@ func (g *gitPRWaiter) runPromotionStep(
 		return PromotionStepResult{Status: kargoapi.PromotionPhaseRunning}, nil
 	}
 	if !pr.Merged {
-		return PromotionStepResult{
-			Status:  kargoapi.PromotionPhaseFailed,
-			Message: fmt.Sprintf("pull request %d was closed without being merged", prNumber),
-		}, err
+		return PromotionStepResult{Status: kargoapi.PromotionPhaseFailed},
+			&terminalError{err: fmt.Errorf("pull request %d was closed without being merged", prNumber)}
 	}
 	return PromotionStepResult{
 		Status: kargoapi.PromotionPhaseSucceeded,

internal/directives/git_pr_waiter_test.go

@@ -150,8 +150,8 @@ func Test_gitPRWaiter_runPromotionStep(t *testing.T) {
 				},
 			},
 			assertions: func(t *testing.T, res PromotionStepResult, err error) {
-				require.NoError(t, err)
-				require.Contains(t, res.Message, "closed without being merged")
+				require.ErrorContains(t, err, "closed without being merged")
+				require.True(t, isTerminal(err))
 				require.Equal(t, kargoapi.PromotionPhaseFailed, res.Status)
 			},
 		},