commit: Wait for the git-log(1) process to exit in FindCommitsej-5671_FindCommits-using-structured-error

In this commit, we add a new wait logic to wait for git-log(1)
command to finish;

Previously we did not Wait() for the git-log(1) process to exit
in FindCommits, causing us to fail to include its execution time
in the command details
and not log any errors it returned.

We differentiate benign errors from real errors.
Benign errors are caused by terminating the stream
since response limit is reached;
while real errors are caused by git log command failures.

Real errors are wrapped and returned as a structured error.

Changelog: fixed

2 files changed, 243 insertions, 11 deletions

diff --git a/internal/gitaly/service/commit/find_commits.go b/internal/gitaly/service/commit/find_commits.go
index 15425330c..53981f794 100644
--- a/internal/gitaly/service/commit/find_commits.go
+++ b/internal/gitaly/service/commit/find_commits.go
@@ -65,14 +65,39 @@ func (s *server) FindCommits(req *gitalypb.FindCommitsRequest, stream gitalypb.C
 	return nil
 }
 
-func (s *server) findCommits(ctx context.Context, req *gitalypb.FindCommitsRequest, stream gitalypb.CommitService_FindCommitsServer) error {
+func (s *server) findCommits(ctx context.Context, req *gitalypb.FindCommitsRequest, stream gitalypb.CommitService_FindCommitsServer) (returnedErr error) {
+	commitCounter := int32(0)
 	opts := git.ConvertGlobalOptions(req.GetGlobalOptions())
 	repo := s.localrepo(req.GetRepository())
 
-	logCmd, err := repo.Exec(ctx, getLogCommandSubCmd(req), append(opts, git.WithSetupStdout())...)
+	var stderr strings.Builder
+	gitLogCmd, limit := getLogCommandSubCmd(req)
+	logCmd, err := repo.Exec(ctx, gitLogCmd, append(opts, git.WithSetupStdout(), git.WithStderr(&stderr))...)
 	if err != nil {
 		return fmt.Errorf("error when creating git log command: %w", err)
 	}
+	defer func() {
+		if err := logCmd.Wait(); err != nil {
+			// We differentiate benign errors from real errors in this deferred function.
+			// Benign errors are the errors that are not harmful to the system, see the cases below.
+			if limit >= 0 && commitCounter == limit {
+				// Benign error caused by terminating the stream since response limit is reached;
+				// We already send the maximum number of commits, so git log command is terminated.
+				s.logger.Debug("git log command terminated because maximum number of commits is reached")
+				return
+			}
+
+			// Real errors are caused by git log command failures such as timeout or OOM kill.
+			if returnedErr == nil {
+				// When returnedErr is nil, it means that we haven't encountered any errors in the findCommits function.
+				// But git log command still failed, so we put the git log command error in returnedErr
+				// and avoid overriding the real error.
+				returnedErr = wrapGitLogCmdError(req.GetRevision(), commitCounter, err, stderr.String())
+			} else {
+				s.logger.WithError(err).WithField("stderr", stderr.String()).Error("listing commits failed")
+			}
+		}
+	}()
 
 	objectReader, cancel, err := s.catfileCache.ObjectReader(ctx, repo)
 	if err != nil {
@@ -95,9 +120,11 @@ func (s *server) findCommits(ctx context.Context, req *gitalypb.FindCommitsReque
 		}
 	}
 
-	if err := streamCommits(getCommits, stream, req.GetTrailers(), req.GetIncludeShortstat(), len(req.GetIncludeReferencedBy()) > 0); err != nil {
+	commitCounter, err = streamCommits(getCommits, stream, req.GetTrailers(), req.GetIncludeShortstat(), len(req.GetIncludeReferencedBy()) > 0)
+	if err != nil {
 		return fmt.Errorf("error streaming commits: %w", err)
 	}
+
 	return nil
 }
 
@@ -198,8 +225,9 @@ func (g *GetCommits) Commit(ctx context.Context, trailers, shortStat, refs bool)
 	return commit.GitCommit, nil
 }
 
-func streamCommits(getCommits *GetCommits, stream gitalypb.CommitService_FindCommitsServer, trailers, shortStat bool, refs bool) error {
+func streamCommits(getCommits *GetCommits, stream gitalypb.CommitService_FindCommitsServer, trailers, shortStat bool, refs bool) (int32, error) {
 	ctx := stream.Context()
+	commitCounter := int32(0)
 
 	chunker := chunk.New(&commitsSender{
 		send: func(commits []*gitalypb.GitCommit) error {
@@ -212,22 +240,23 @@ func streamCommits(getCommits *GetCommits, stream gitalypb.CommitService_FindCom
 	for getCommits.Scan() {
 		commit, err := getCommits.Commit(ctx, trailers, shortStat, refs)
 		if err != nil {
-			return err
+			return commitCounter, err
 		}
 
 		if err := chunker.Send(commit); err != nil {
-			return err
+			return commitCounter, err
 		}
+		commitCounter++
 	}
 
 	if getCommits.Err() != nil {
-		return fmt.Errorf("get commits: %w", getCommits.Err())
+		return commitCounter, fmt.Errorf("get commits: %w", getCommits.Err())
 	}
 
-	return chunker.Flush()
+	return commitCounter, chunker.Flush()
 }
 
-func getLogCommandSubCmd(req *gitalypb.FindCommitsRequest) git.Command {
+func getLogCommandSubCmd(req *gitalypb.FindCommitsRequest) (git.Command, int32) {
 	logFormatOption := "--format=%H"
 	// To split the commits by '\x01' instead of '\n'
 	if req.GetIncludeShortstat() {
@@ -298,7 +327,7 @@ func getLogCommandSubCmd(req *gitalypb.FindCommitsRequest) git.Command {
 		}
 	}
 
-	return subCmd
+	return subCmd, limit
 }
 
 func parseRefs(refsLine string) [][]byte {
@@ -391,3 +420,58 @@ func splitStat(data []byte, atEOF bool) (int, []byte, error) {
 
 	return index + 1, data[:index], nil
 }
+
+// wrapGitError wraps git log error with a structError
+func wrapGitLogCmdError(revision []byte, commitCounter int32, err error, stderr string) (structError structerr.Error) {
+	// when `git log` returns empty and also exit with an error code.
+	if commitCounter == 0 {
+		switch {
+		case strings.HasPrefix(stderr, "fatal: ambiguous argument "):
+			// for example, git log non-existing-branch will cause
+			// fatal: ambiguous argument 'non-existing-branch': unknown revision or path not in the working tree.
+			return structerr.NewNotFound("commits not found").
+				WithMetadata("error", err).
+				WithMetadata("stderr", stderr).
+				WithDetail(&gitalypb.FindCommitsError{
+					Error: &gitalypb.FindCommitsError_AmbiguousRef{
+						AmbiguousRef: &gitalypb.AmbiguousReferenceError{
+							Reference: revision,
+						},
+					},
+				})
+		case strings.HasPrefix(stderr, "fatal: bad object"):
+			// for example, git log 37811987837aacbd3b1d8ceb8de669b33f7c7c0a will cause
+			// fatal: bad object 37811987837aacbd3b1d8ceb8de669b33f7c7c0a
+			return structerr.NewNotFound("commits not found").
+				WithMetadata("error", err).
+				WithMetadata("stderr", stderr).
+				WithDetail(&gitalypb.FindCommitsError{
+					Error: &gitalypb.FindCommitsError_BadObject{
+						BadObject: &gitalypb.BadObjectError{
+							ObjectId: revision,
+						},
+					},
+				})
+		case strings.HasPrefix(stderr, "fatal: Invalid revision range"):
+			// for example git log 37811987837aacbd3b1d8ceb8de669b33f7c7c0a..37811987837aacbd3b1d8ceb8de669b33f7c7c0b will cause
+			// fatal: Invalid revision range 37811987837aacbd3b1d8ceb8de669b33f7c7c0a..37811987837aacbd3b1d8ceb8de669b33f7c7c0b
+			return structerr.NewNotFound("commits not found").
+				WithMetadata("error", err).
+				WithMetadata("stderr", stderr).
+				WithDetail(&gitalypb.FindCommitsError{
+					Error: &gitalypb.FindCommitsError_InvalidRange{
+						InvalidRange: &gitalypb.InvalidRevisionRange{
+							Range: revision,
+						},
+					},
+				})
+		default:
+			return structerr.NewNotFound("commits not found").
+				// other cases we don't know right now, can be added when necessary
+				WithMetadata("error", err).
+				WithMetadata("stderr", stderr)
+
+		}
+	}
+	return structerr.NewInternal("listing commits failed").WithMetadata("error", err).WithMetadata("stderr", stderr)
+}
diff --git a/internal/gitaly/service/commit/find_commits_test.go b/internal/gitaly/service/commit/find_commits_test.go
index 21a92e71a..ed64374b1 100644
--- a/internal/gitaly/service/commit/find_commits_test.go
+++ b/internal/gitaly/service/commit/find_commits_test.go
@@ -17,6 +17,7 @@ import (
 	"gitlab.com/gitlab-org/gitaly/v16/internal/gitaly/storage/storagemgr"
 	"gitlab.com/gitlab-org/gitaly/v16/internal/structerr"
 	"gitlab.com/gitlab-org/gitaly/v16/internal/testhelper"
+	"gitlab.com/gitlab-org/gitaly/v16/internal/testhelper/testserver"
 	"gitlab.com/gitlab-org/gitaly/v16/proto/go/gitalypb"
 	"golang.org/x/text/encoding/charmap"
 	"google.golang.org/grpc/codes"
@@ -651,16 +652,28 @@ func TestFindCommits_quarantine(t *testing.T) {
 		desc          string
 		altDirs       []string
 		expectedCount int
+		expectedErr   error
 	}{
 		{
 			desc:          "present GIT_ALTERNATE_OBJECT_DIRECTORIES",
 			altDirs:       []string{altObjectsDir},
 			expectedCount: 1,
+			expectedErr:   nil,
 		},
 		{
 			desc:          "empty GIT_ALTERNATE_OBJECT_DIRECTORIES",
 			altDirs:       []string{},
 			expectedCount: 0,
+			expectedErr: testhelper.ToInterceptedMetadata(
+				structerr.NewNotFound("commits not found").WithMetadata("error", "exit status 128").WithMetadata("stderr",
+					fmt.Sprintf("fatal: bad object %s\n", commitID.String()))).WithDetail(
+				&gitalypb.FindCommitsError{
+					Error: &gitalypb.FindCommitsError_BadObject{
+						BadObject: &gitalypb.BadObjectError{
+							ObjectId: []byte(commitID.String()),
+						},
+					},
+				}),
 		},
 	} {
 		t.Run(tc.desc, func(t *testing.T) {
@@ -681,12 +694,147 @@ func TestFindCommits_quarantine(t *testing.T) {
 				Revision:   []byte(commitID.String()),
 				Limit:      1,
 			})
-			require.NoError(t, err)
+			testhelper.RequireGrpcError(t, tc.expectedErr, err)
+			require.Len(t, commits, tc.expectedCount)
+		})
+	}
+}
+
+func TestFindCommits_simulateGitLogWaitError(t *testing.T) {
+	t.Parallel()
+
+	ctx := testhelper.Context(t)
+	logger := testhelper.NewLogger(t)
+	cfg, client := setupCommitService(t, ctx, testserver.WithLogger(logger))
+
+	repo, repoPath := gittest.CreateRepository(t, ctx, cfg)
+	// altObjectsDir is used to trigger git log wait error
+	// we will set this to empty string to trigger the error
+	altObjectsDir := "./alt-objects"
+	commitID := gittest.WriteCommit(t, cfg, repoPath,
+		gittest.WithAlternateObjectDirectory(filepath.Join(repoPath, altObjectsDir)),
+	)
+
+	for _, tc := range []struct {
+		desc          string
+		altDirs       []string
+		expectedCount int
+		expectedErr   error
+		limit         int32
+	}{
+		{
+			desc:          "git log exit with error, with limit 0",
+			altDirs:       []string{},
+			limit:         0,
+			expectedCount: 0,
+			expectedErr:   nil,
+		},
+		{
+			desc:          "limit 1, git log exit with error",
+			altDirs:       []string{},
+			limit:         1,
+			expectedCount: 0,
+			expectedErr: testhelper.ToInterceptedMetadata(
+				structerr.NewNotFound("commits not found").
+					WithMetadata("error", "exit status 128").
+					WithMetadata("stderr", fmt.Sprintf("fatal: bad object %s\n", commitID.String()))).WithDetail(
+				&gitalypb.FindCommitsError{
+					Error: &gitalypb.FindCommitsError_BadObject{
+						BadObject: &gitalypb.BadObjectError{
+							ObjectId: []byte(commitID.String()),
+						},
+					},
+				}),
+		},
+	} {
+		tc := tc
+		t.Run(tc.desc, func(t *testing.T) {
+			repo.GitAlternateObjectDirectories = tc.altDirs
+			commits, err := getCommits(t, ctx, client, &gitalypb.FindCommitsRequest{
+				Repository: repo,
+				Revision:   []byte(commitID.String()),
+				Limit:      tc.limit,
+			})
+			testhelper.RequireGrpcError(t, tc.expectedErr, err)
 			require.Len(t, commits, tc.expectedCount)
 		})
 	}
 }
 
+func TestWrapGitLogCmdError(t *testing.T) {
+	t.Parallel()
+	for _, tc := range []struct {
+		desc            string
+		commitCounter   int32
+		gitLogCmdArg    string
+		gitLogCmdStdErr string
+		expectedErr     error
+	}{
+		{
+			desc:            "no commits found due to ambiguous argument",
+			commitCounter:   0,
+			gitLogCmdArg:    "non-existing-ref",
+			gitLogCmdStdErr: "fatal: ambiguous argument 'non-existing-ref': unknown revision or path not in the working tree.",
+			expectedErr: structerr.NewNotFound("commits not found").WithDetail(
+				&gitalypb.FindCommitsError{
+					Error: &gitalypb.FindCommitsError_AmbiguousRef{
+						AmbiguousRef: &gitalypb.AmbiguousReferenceError{
+							Reference: []byte("non-existing-ref"),
+						},
+					},
+				}),
+		},
+		{
+			desc:            "no commits found due to bad object id",
+			commitCounter:   0,
+			gitLogCmdArg:    "37811987837aacbd3b1d8ceb8de669b33f7c7c0a",
+			gitLogCmdStdErr: "fatal: bad object 37811987837aacbd3b1d8ceb8de669b33f7c7c0a",
+			expectedErr: structerr.NewNotFound("commits not found").WithDetail(
+				&gitalypb.FindCommitsError{
+					Error: &gitalypb.FindCommitsError_BadObject{
+						BadObject: &gitalypb.BadObjectError{
+							ObjectId: []byte("37811987837aacbd3b1d8ceb8de669b33f7c7c0a"),
+						},
+					},
+				}),
+		},
+		{
+			desc:            "no commits found due to invalid range",
+			commitCounter:   0,
+			gitLogCmdArg:    "37811987837aacbd3b1d8ceb8de669b33f7c7c0a..37811987837aacbd3b1d8ceb8de669b33f7c7c0b",
+			gitLogCmdStdErr: "fatal: Invalid revision range 37811987837aacbd3b1d8ceb8de669b33f7c7c0a..37811987837aacbd3b1d8ceb8de669b33f7c7c0b\n",
+			expectedErr: structerr.NewNotFound("commits not found").WithDetail(
+				&gitalypb.FindCommitsError{
+					Error: &gitalypb.FindCommitsError_InvalidRange{
+						InvalidRange: &gitalypb.InvalidRevisionRange{
+							Range: []byte("37811987837aacbd3b1d8ceb8de669b33f7c7c0a..37811987837aacbd3b1d8ceb8de669b33f7c7c0b"),
+						},
+					},
+				}),
+		},
+		{
+			desc:            "uncategorized error that causes empty commits",
+			commitCounter:   0,
+			gitLogCmdArg:    "",
+			gitLogCmdStdErr: "fatal: some other cause can't be foreseen right now",
+			expectedErr:     structerr.NewNotFound("commits not found"),
+		},
+		{
+			desc:            "mock terminated by user error",
+			commitCounter:   10,
+			gitLogCmdArg:    "main",
+			gitLogCmdStdErr: "terminated by user",
+			expectedErr:     structerr.NewInternal("listing commits failed"),
+		},
+	} {
+		tc := tc
+		t.Run(tc.desc, func(t *testing.T) {
+			actualErr := wrapGitLogCmdError([]byte(tc.gitLogCmdArg), tc.commitCounter, nil, tc.gitLogCmdStdErr)
+			testhelper.RequireGrpcError(t, tc.expectedErr, actualErr)
+		})
+	}
+}
+
 func TestFindCommits_followWithOffset(t *testing.T) {
 	t.Parallel()