gitpipe: Fix sending of events to be race-free with context cancellation

When sending events down the gitpipe pipelines, we need to ensure that
we're paying close attention to context cancellation: if the context got
cancelled, then we shouldn't continue driving the pipeline and instead
exit early. While we already do this in all of our pipelines, we're not
treating context cancellation the same everywhere. In some pipelines we
prioritize context cancellation errors over errors that happened while
serving the request, while in others we treat them the same. Treating
these errors the same can result in racy error propagation though, where
we randomly return either the context error, or the error of the process
we just tried to read from but which got killed because of the context
cancellation.

This issue has originally been fixed in 3a65ca3bb (gitpipe: Prioritize
context cancellation, 2021-07-09), but the fix was only a partial fix
that didn't apply to all pipelines. Fix the remaining cases so that we
always prioritize context cancellation to unify the behaviour.

Note that this commit also fixes multiple cases where we just sent an
error down the pipeline, but didn't actually abort the pipeline. This
was very likely an oversight by the original author who has been doing
too much C programming, where `case` statements automatically fall
through.

Changelog: fixed

2 files changed, 53 insertions, 47 deletions

diff --git a/internal/git/gitpipe/catfile_info.go b/internal/git/gitpipe/catfile_info.go
index b756e08f3..534a12c4b 100644
--- a/internal/git/gitpipe/catfile_info.go
+++ b/internal/git/gitpipe/catfile_info.go
@@ -74,48 +74,34 @@ func CatfileInfo(
 		var i int64
 		for it.Next() {
 			if err := queue.RequestRevision(it.ObjectID().Revision()); err != nil {
-				select {
-				case requestChan <- catfileInfoRequest{err: err}:
-				case <-ctx.Done():
-					return
-				}
+				sendCatfileInfoRequest(ctx, requestChan, catfileInfoRequest{err: err})
+				return
 			}
 
-			select {
-			case requestChan <- catfileInfoRequest{
+			if isDone := sendCatfileInfoRequest(ctx, requestChan, catfileInfoRequest{
 				objectID:   it.ObjectID(),
 				objectName: it.ObjectName(),
-			}:
-			case <-ctx.Done():
+			}); isDone {
 				return
 			}
 
 			i++
 			if i%int64(cap(requestChan)) == 0 {
 				if err := queue.Flush(); err != nil {
-					select {
-					case requestChan <- catfileInfoRequest{err: err}:
-					case <-ctx.Done():
-						return
-					}
+					sendCatfileInfoRequest(ctx, requestChan, catfileInfoRequest{err: err})
+					return
 				}
 			}
 		}
 
 		if err := it.Err(); err != nil {
-			select {
-			case requestChan <- catfileInfoRequest{err: err}:
-			case <-ctx.Done():
-				return
-			}
+			sendCatfileInfoRequest(ctx, requestChan, catfileInfoRequest{err: err})
+			return
 		}
 
 		if err := queue.Flush(); err != nil {
-			select {
-			case requestChan <- catfileInfoRequest{err: err}:
-			case <-ctx.Done():
-				return
-			}
+			sendCatfileInfoRequest(ctx, requestChan, catfileInfoRequest{err: err})
+			return
 		}
 	}()
 
@@ -252,3 +238,19 @@ func sendCatfileInfoResult(ctx context.Context, ch chan<- CatfileInfoResult, res
 		return true
 	}
 }
+
+func sendCatfileInfoRequest(ctx context.Context, ch chan<- catfileInfoRequest, request catfileInfoRequest) bool {
+	// Please refer to `sendCatfileInfoResult()` for why we treat the context specially.
+	select {
+	case <-ctx.Done():
+		return true
+	default:
+	}
+
+	select {
+	case ch <- request:
+		return false
+	case <-ctx.Done():
+		return true
+	}
+}
diff --git a/internal/git/gitpipe/catfile_object.go b/internal/git/gitpipe/catfile_object.go
index 49e33b859..a393ed2ae 100644
--- a/internal/git/gitpipe/catfile_object.go
+++ b/internal/git/gitpipe/catfile_object.go
@@ -48,48 +48,52 @@ func CatfileObject(
 	go func() {
 		defer close(requestChan)
 
+		sendRequest := func(request catfileObjectRequest) bool {
+			// Please refer to `sendResult()` for why we treat the context specially.
+			select {
+			case <-ctx.Done():
+				return true
+			default:
+			}
+
+			select {
+			case requestChan <- request:
+				return false
+			case <-ctx.Done():
+				return true
+			}
+		}
+
 		var i int64
 		for it.Next() {
 			if err := queue.RequestRevision(it.ObjectID().Revision()); err != nil {
-				select {
-				case requestChan <- catfileObjectRequest{err: err}:
-				case <-ctx.Done():
-					return
-				}
+				sendRequest(catfileObjectRequest{err: err})
+				return
 			}
 
-			select {
-			case requestChan <- catfileObjectRequest{objectName: it.ObjectName()}:
-			case <-ctx.Done():
+			if isDone := sendRequest(catfileObjectRequest{
+				objectName: it.ObjectName(),
+			}); isDone {
 				return
 			}
 
 			i++
 			if i%int64(cap(requestChan)) == 0 {
 				if err := queue.Flush(); err != nil {
-					select {
-					case requestChan <- catfileObjectRequest{err: err}:
-					case <-ctx.Done():
-						return
-					}
+					sendRequest(catfileObjectRequest{err: err})
+					return
 				}
 			}
 		}
 
 		if err := it.Err(); err != nil {
-			select {
-			case requestChan <- catfileObjectRequest{err: err}:
-			case <-ctx.Done():
-				return
-			}
+			sendRequest(catfileObjectRequest{err: err})
+			return
 		}
 
 		if err := queue.Flush(); err != nil {
-			select {
-			case requestChan <- catfileObjectRequest{err: err}:
-			case <-ctx.Done():
-				return
-			}
+			sendRequest(catfileObjectRequest{err: err})
+			return
 		}
 	}()