fix

Author: wxiaoguang

modules/git/catfile_batch_reader.go

@@ -16,40 +16,52 @@ import (
 
 	"code.gitea.io/gitea/modules/git/gitcmd"
 	"code.gitea.io/gitea/modules/log"
+	"code.gitea.io/gitea/modules/util"
 )
 
-var catFileBatchDebugPipeClose atomic.Pointer[func(stdPipeClose func())]
-
 type catFileBatchCommunicator struct {
-	closeFunc   func(err error)
+	closeFunc   atomic.Pointer[func(err error)]
 	reqWriter   io.Writer
 	respReader  *bufio.Reader
 	debugGitCmd *gitcmd.Command
 }
 
-func (b *catFileBatchCommunicator) Close() {
-	if b.closeFunc != nil {
-		b.closeFunc(nil)
-		b.closeFunc = nil
+func (b *catFileBatchCommunicator) Close(err ...error) {
+	if fn := b.closeFunc.Load(); fn != nil {
+		(*fn)(util.OptionalArg(err))
+		b.closeFunc.Store(nil)
 	}
 }
 
+func (b *catFileBatchCommunicator) debugKill() (ret struct {
+	beforeClose chan struct{}
+	blockClose  chan struct{}
+	afterClose  chan struct{}
+}) {
+	ret.beforeClose = make(chan struct{})
+	ret.blockClose = make(chan struct{})
+	ret.afterClose = make(chan struct{})
+	oldCloseFunc := b.closeFunc.Load()
+	b.closeFunc.Store(new(func(err error) {
+		b.closeFunc.Store(oldCloseFunc)
+		close(ret.beforeClose)
+		<-ret.blockClose
+		(*oldCloseFunc)(err)
+		close(ret.afterClose)
+	}))
+	b.debugGitCmd.DebugKill()
+	return ret
+}
+
 // newCatFileBatch opens git cat-file --batch in the provided repo and returns a stdin pipe, a stdout reader and cancel function
 func newCatFileBatch(ctx context.Context, repoPath string, cmdCatFile *gitcmd.Command) (ret *catFileBatchCommunicator) {
 	ctx, ctxCancel := context.WithCancelCause(ctx)
 
 	// We often want to feed the commits in order into cat-file --batch, followed by their trees and subtrees as necessary.
 	stdinWriter, stdoutReader, stdPipeClose := cmdCatFile.MakeStdinStdoutPipe()
-	pipeClose := func() {
-		if fn := catFileBatchDebugPipeClose.Load(); fn != nil {
-			(*fn)(stdPipeClose)
-		} else {
-			stdPipeClose()
-		}
-	}
 	closeFunc := func(err error) {
 		ctxCancel(err)
-		pipeClose()
+		stdPipeClose()
 	}
 	return newCatFileBatchWithCloseFunc(ctx, repoPath, cmdCatFile, stdinWriter, stdoutReader, closeFunc)
 }
@@ -59,17 +71,17 @@ func newCatFileBatchWithCloseFunc(ctx context.Context, repoPath string, cmdCatFi
 ) *catFileBatchCommunicator {
 	ret := &catFileBatchCommunicator{
 		debugGitCmd: cmdCatFile,
-		closeFunc:   closeFunc,
 		reqWriter:   stdinWriter,
 		respReader:  bufio.NewReaderSize(stdoutReader, 32*1024), // use a buffered reader for rich operations
 	}
+	ret.closeFunc.Store(&closeFunc)
 
 	err := cmdCatFile.WithDir(repoPath).StartWithStderr(ctx)
 	if err != nil {
 		log.Error("Unable to start git command %v: %v", cmdCatFile.LogString(), err)
 		// ideally here it should return the error, but it would require refactoring all callers
 		// so just return a dummy communicator that does nothing, almost the same behavior as before, not bad
-		closeFunc(err)
+		ret.Close(err)
 		return ret
 	}
 
@@ -78,7 +90,7 @@ func newCatFileBatchWithCloseFunc(ctx context.Context, repoPath string, cmdCatFi
 		if err != nil && !errors.Is(err, context.Canceled) {
 			log.Error("cat-file --batch command failed in repo %s, error: %v", repoPath, err)
 		}
-		closeFunc(err)
+		ret.Close(err)
 	}()
 
 	return ret

modules/git/catfile_batch_test.go

@@ -37,53 +37,49 @@ func testCatFileBatch(t *testing.T) {
 		require.Error(t, err)
 	})
 
-	setupTerminatedBatch := func(t *testing.T) (*catFileBatchCommunicator, func()) {
-		t.Helper()
+	simulateQueryTerminated := func(t *testing.T, errBeforePipeClose, errAfterPipeClose error) {
+		readError := func(t *testing.T, r io.Reader, expectedErr error) {
+			if expectedErr == nil {
+				return // expectedErr == nil means this read should be skipped
+			}
+			n, err := r.Read(make([]byte, 100))
+			assert.Zero(t, n)
+			assert.ErrorIs(t, err, expectedErr)
+		}
+
 		batch, err := NewBatch(t.Context(), filepath.Join(testReposDir, "repo1_bare"))
 		require.NoError(t, err)
-		_, _ = batch.QueryInfo("e2129701f1a4d54dc44f03c93bca0a2aec7c5449")
-		var c *catFileBatchCommunicator
+		_, err = batch.QueryInfo("e2129701f1a4d54dc44f03c93bca0a2aec7c5449")
+		require.NoError(t, err)
+
+		var batchComm *catFileBatchCommunicator
 		switch b := batch.(type) {
 		case *catFileBatchLegacy:
-			c = b.batchCheck
-			_, _ = c.reqWriter.Write([]byte("in-complete-line-"))
+			batchComm = b.batchCheck
+			_, _ = batchComm.reqWriter.Write([]byte("in-complete-line-"))
 		case *catFileBatchCommand:
-			c = b.batch
-			_, _ = c.reqWriter.Write([]byte("info"))
+			batchComm = b.batch
+			_, _ = batchComm.reqWriter.Write([]byte("info"))
 		default:
 			t.FailNow()
 		}
-		return c, func() { batch.Close() }
-	}
+		defer batchComm.Close()
+
+		require.True(t, (errBeforePipeClose != nil) != (errAfterPipeClose != nil), "must set exactly one of the expected errors")
 
+		inceptor := batchComm.debugKill()
+		<-inceptor.beforeClose                                 // wait for the command's Close to be called, the pipe is not closed yet
+		readError(t, batchComm.respReader, errBeforePipeClose) // then caller will read on an open pipe which will be closed soon
+		close(inceptor.blockClose)                             // continue to close the pipe
+		<-inceptor.afterClose                                  // wait for the pipe to be closed
+		readError(t, batchComm.respReader, errAfterPipeClose)  // then caller will read on a closed pipe
+	}
 	t.Run("QueryTerminated", func(t *testing.T) {
 		t.Run("PipeClosedBeforeRead", func(t *testing.T) {
-			closed := make(chan struct{})
-			hook := func(doClose func()) { doClose(); close(closed) }
-			c, cleanup := setupTerminatedBatch(t)
-			defer cleanup()
-			catFileBatchDebugPipeClose.Store(&hook)
-			defer catFileBatchDebugPipeClose.Store(nil)
-			c.debugGitCmd.DebugKill()
-			<-closed
-			n, err := c.respReader.Read(make([]byte, 100))
-			assert.Zero(t, n)
-			assert.ErrorIs(t, err, os.ErrClosed)
+			simulateQueryTerminated(t, io.EOF, nil)
 		})
 		t.Run("ReadBeforePipeClose", func(t *testing.T) {
-			ready := make(chan struct{})
-			proceed := make(chan struct{})
-			hook := func(doClose func()) { close(ready); <-proceed; doClose() }
-			c, cleanup := setupTerminatedBatch(t)
-			defer cleanup()
-			catFileBatchDebugPipeClose.Store(&hook)
-			defer catFileBatchDebugPipeClose.Store(nil)
-			c.debugGitCmd.DebugKill()
-			<-ready
-			n, err := c.respReader.Read(make([]byte, 100))
-			assert.Zero(t, n)
-			assert.ErrorIs(t, err, io.EOF)
-			close(proceed)
+			simulateQueryTerminated(t, nil, os.ErrClosed)
 		})
 	})