commit: Wait for the git-log(1) process to exit in FindCommits

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.
Non-benign errors are logged as errors.

Changelog: fixed

2 files changed, 94 insertions, 11 deletions

diff --git a/internal/gitaly/service/commit/find_commits.go b/internal/gitaly/service/commit/find_commits.go
index e2d717b25..167909b23 100644
--- a/internal/gitaly/service/commit/find_commits.go
+++ b/internal/gitaly/service/commit/find_commits.go
@@ -65,14 +65,33 @@ 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 caused by terminating the stream since response limit is reached;
+			// real errors are caused by git log command failures such as timeout or OOM kill.
+			if limit >= 0 && commitCounter == limit {
+				// We already send the maximum number of commits, git log command is terminated.
+				s.logger.Debug("git log command terminated because maximum number of commits is reached")
+			} else if returnedErr == nil {
+				// Avoid overriding the real error.
+				returnedErr = structerr.NewInternal("listing commits failed").WithMetadata("error", err).WithMetadata("stderr", 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 +114,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 +219,9 @@ func (g *GetCommits) Commit(ctx context.Context, trailers, shortStat, refs bool)
 	return commit, 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 +234,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 +321,7 @@ func getLogCommandSubCmd(req *gitalypb.FindCommitsRequest) git.Command {
 		}
 	}
 
-	return subCmd
+	return subCmd, limit
 }
 
 func parseRefs(refsLine string) [][]byte {
diff --git a/internal/gitaly/service/commit/find_commits_test.go b/internal/gitaly/service/commit/find_commits_test.go
index 50648526c..d46cb178a 100644
--- a/internal/gitaly/service/commit/find_commits_test.go
+++ b/internal/gitaly/service/commit/find_commits_test.go
@@ -15,6 +15,7 @@ import (
 	"gitlab.com/gitlab-org/gitaly/v16/internal/gitaly/storage"
 	"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"
@@ -648,16 +649,21 @@ 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.NewInternal("listing commits failed").WithMetadata("error", "exit status 128").WithMetadata("stderr",
+					fmt.Sprintf("fatal: bad object %s\n", commitID.String()))),
 		},
 	} {
 		t.Run(tc.desc, func(t *testing.T) {
@@ -668,7 +674,61 @@ 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.NewInternal("listing commits failed").
+					WithMetadata("error", "exit status 128").WithMetadata("stderr",
+					fmt.Sprintf("fatal: bad object %s\n", 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)
 		})
 	}