fix race condition in promotion

I believe there is a race condition here, and we are closing the channel
during promotion before we finish filling it.  This is because the WaitGroup
is incremented _after_ we add to the channel.

Refactor to make the logic (hopefully) clearer.

Author: justinsb

internal/legacy/dockerregistry/inventory.go

@@ -1364,6 +1364,59 @@ func (sc *SyncContext) ExecRequests(
 	return err
 }
 
+type ForkJoinFunc[K any, V any] func(k K) (V, error)
+
+type ForkJoinResult[K any, V any] struct {
+	Input  K
+	Output V
+	Error  error
+}
+
+// ExecRequests uses the Worker Pool pattern, where maxConcurrentRequests
+// determines the number of workers to spawn.
+func ForkJoin[K any, V any](
+	maxConcurrentRequests int,
+	requests []K,
+	processRequest ForkJoinFunc[K, V],
+) []ForkJoinResult[K, V] {
+
+	reqChan := make(chan K, maxConcurrentRequests*2)
+
+	var wg sync.WaitGroup
+	var resultsMutex sync.Mutex
+	var results []ForkJoinResult[K, V]
+
+	// Log any errors encountered.
+	for range maxConcurrentRequests {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+
+			for req := range reqChan {
+				result, err := processRequest(req)
+
+				resultsMutex.Lock()
+				results = append(results, ForkJoinResult[K, V]{
+					Input:  req,
+					Output: result,
+					Error:  err,
+				})
+				resultsMutex.Unlock()
+			}
+		}()
+	}
+
+	for _, req := range requests {
+		reqChan <- req
+	}
+	close(reqChan)
+
+	// Wait for all workers to finish draining the jobs.
+	wg.Wait()
+
+	return results
+}
+
 func extractRegistryTags(reader io.Reader) (*ggcrV1Google.Tags, error) {
 	tags := ggcrV1Google.Tags{}
 	decoder := json.NewDecoder(reader)
@@ -1479,64 +1532,65 @@ func (sc *SyncContext) ValidateEdge(edge *PromotionEdge) error {
 	return nil
 }
 
-// MKPopulateRequestsForPromotionEdges takes in a map of PromotionEdges to promote
-// and a PromotionContext and returns a PopulateRequests which can generate
-// requests to be processed.
-func MKPopulateRequestsForPromotionEdges(
+// BuildPopulateRequestsForPromotionEdges takes in a map of PromotionEdges to promote
+// and a PromotionContext and returns the promotion requests.
+func (sc *SyncContext) BuildPopulateRequestsForPromotionEdges(
 	toPromote map[PromotionEdge]interface{},
-) PopulateRequests {
-	return func(sc *SyncContext, reqs chan<- stream.ExternalRequest, wg *sync.WaitGroup) {
-		if len(toPromote) == 0 {
-			logrus.Info("Nothing to promote.")
-			return
-		}
+) []stream.ExternalRequest {
 
-		if sc.Confirm {
-			logrus.Info("---------- BEGIN PROMOTION ----------")
-		} else {
-			logrus.Info("---------- BEGIN PROMOTION (DRY RUN) ----------")
-		}
+	var requests []stream.ExternalRequest
 
-		for promoteMe := range toPromote {
-			var req stream.ExternalRequest
-			oldDigest := image.Digest("")
-
-			// Technically speaking none of the edges at this point should be
-			// invalid (such as trying to do tag moves), because we run
-			// ValidateEdges() in Promote() in the early stages, before passing
-			// on this closure to ExecRequests(). However the check here is so
-			// cheap that we do it anyway, just in case.
-			if err := sc.ValidateEdge(&promoteMe); err != nil {
-				logrus.Error(err)
-				continue
-			}
+	if len(toPromote) == 0 {
+		logrus.Info("Nothing to promote.")
+		return requests
+	}
 
-			// Save some information about this request. It's a bit like
-			// HTTP "headers".
-			req.RequestParams = PromotionRequest{
-				// Only support adding new tags during a promotion run. Tag
-				// moves and deletions are not supported.
-				//
-				// Although disallowing tag moves sounds a bit draconian, it
-				// does make protect production from a malformed set of promoter
-				// manifests with incorrect tag information.
-				Add,
-				// TODO: Clean up types to avoid having to split up promoteMe
-				// prematurely like this.
-				promoteMe.SrcRegistry.Name,
-				promoteMe.DstRegistry.Name,
-				promoteMe.DstRegistry.ServiceAccount,
-				promoteMe.SrcImageTag.Name,
-				promoteMe.DstImageTag.Name,
-				promoteMe.Digest,
-				oldDigest,
-				promoteMe.DstImageTag.Tag,
-			}
+	if sc.Confirm {
+		logrus.Info("---------- BEGIN PROMOTION ----------")
+	} else {
+		logrus.Info("---------- BEGIN PROMOTION (DRY RUN) ----------")
+	}
 
-			wg.Add(1)
-			reqs <- req
+	for promoteMe := range toPromote {
+		var req stream.ExternalRequest
+		oldDigest := image.Digest("")
+
+		// Technically speaking none of the edges at this point should be
+		// invalid (such as trying to do tag moves), because we run
+		// ValidateEdges() in Promote() in the early stages, before passing
+		// on this closure to ExecRequests(). However the check here is so
+		// cheap that we do it anyway, just in case.
+		if err := sc.ValidateEdge(&promoteMe); err != nil {
+			logrus.Error(err)
+			continue
+		}
+
+		// Save some information about this request. It's a bit like
+		// HTTP "headers".
+		req.RequestParams = PromotionRequest{
+			// Only support adding new tags during a promotion run. Tag
+			// moves and deletions are not supported.
+			//
+			// Although disallowing tag moves sounds a bit draconian, it
+			// does make protect production from a malformed set of promoter
+			// manifests with incorrect tag information.
+			Add,
+			// TODO: Clean up types to avoid having to split up promoteMe
+			// prematurely like this.
+			promoteMe.SrcRegistry.Name,
+			promoteMe.DstRegistry.Name,
+			promoteMe.DstRegistry.ServiceAccount,
+			promoteMe.SrcImageTag.Name,
+			promoteMe.DstImageTag.Name,
+			promoteMe.Digest,
+			oldDigest,
+			promoteMe.DstImageTag.Tag,
 		}
+
+		requests = append(requests, req)
 	}
+
+	return requests
 }
 
 // RunChecks runs defined PreChecks in order to check the promotion.
@@ -1665,7 +1719,7 @@ func getRegistriesToRead(edges map[PromotionEdge]interface{}) []registry.Context
 // Manifest.
 func (sc *SyncContext) Promote(
 	edges map[PromotionEdge]interface{},
-	customProcessRequest *ProcessRequest,
+	customProcessRequest ProcessRequestFunc,
 ) error {
 	if len(edges) == 0 {
 		logrus.Info("Nothing to promote.")
@@ -1693,17 +1747,11 @@ func (sc *SyncContext) Promote(
 	}
 
 	var (
-		populateRequests = MKPopulateRequestsForPromotionEdges(edges)
-
-		processRequest     ProcessRequest
-		processRequestReal ProcessRequest = func(
-			_ *SyncContext,
-			reqs chan stream.ExternalRequest,
-			requestResults chan<- RequestResult,
-			_ *sync.WaitGroup,
-			_ *sync.Mutex,
-		) {
-			for req := range reqs {
+		promotionRequests = sc.BuildPopulateRequestsForPromotionEdges(edges)
+
+		processRequest     func(req stream.ExternalRequest) (RequestResult, error)
+		processRequestReal ProcessRequestFunc = func(req stream.ExternalRequest) (RequestResult, error) {
+			{
 				reqRes := RequestResult{Context: req}
 				errors := make(Errors, 0)
 				// If we're adding or moving (i.e., creating a new image or
@@ -1756,8 +1804,20 @@ func (sc *SyncContext) Promote(
 					logrus.Infof("deletions are no longer supported")
 				}
 
+				if len(errors) > 0 {
+					logrus.Errorf(
+						// TODO(log): Consider logging with fields
+						"request %v: error(s) encountered: %v",
+						reqRes.Context,
+						reqRes.Errors)
+				} else {
+					// TODO(log): Consider logging with fields
+					logrus.Infof("request %v: OK", reqRes.Context.RequestParams)
+				}
+				// Log the HTTP request to GCR.
+				reqcounter.Increment()
 				reqRes.Errors = errors
-				requestResults <- reqRes
+				return reqRes, nil
 			}
 		}
 	)
@@ -1767,16 +1827,34 @@ func (sc *SyncContext) Promote(
 	if sc.Confirm {
 		processRequest = processRequestReal
 	} else {
-		processRequestDryRun := MkRequestCapturer(&captured)
+		processRequestDryRun := MkRequestCapturerFunc(&captured)
 		processRequest = processRequestDryRun
 	}
 
 	if customProcessRequest != nil {
-		processRequest = *customProcessRequest
+		processRequest = customProcessRequest
 	}
 
 	sc.PrintCapturedRequests(&captured)
-	return sc.ExecRequests(populateRequests, processRequest)
+
+	// Run requests.
+	maxConcurrentRequests := 10
+
+	if sc.Threads > 0 {
+		maxConcurrentRequests = sc.Threads
+	}
+	results := ForkJoin(maxConcurrentRequests, promotionRequests, processRequest)
+
+	var errs []error
+	for _, result := range results {
+		if len(result.Output.Errors) > 0 {
+			sc.Logs.Errors = append(sc.Logs.Errors, result.Output.Errors...)
+			errs = append(errs, errors.New("encountered an error while executing requests"))
+		}
+		errs = append(errs, result.Error)
+	}
+
+	return errors.Join(errs...)
 }
 
 // PrintCapturedRequests pretty-prints all given PromotionRequests.
@@ -1845,6 +1923,7 @@ func (pr *PromotionRequest) PrettyValue() string {
 
 // MkRequestCapturer returns a function that simply records requests as they are
 // captured (slurped out from the reqs channel).
+// Deprecated: Prefer MkRequestCapturerFunc
 func MkRequestCapturer(captured *CapturedRequests) ProcessRequest {
 	return func(
 		_ *SyncContext,
@@ -1870,6 +1949,27 @@ func MkRequestCapturer(captured *CapturedRequests) ProcessRequest {
 	}
 }
 
+// MkRequestCapturer returns a function that simply records requests as they are
+// captured (slurped out from the reqs channel).
+func MkRequestCapturerFunc(captured *CapturedRequests) ProcessRequestFunc {
+	var mutex sync.Mutex
+
+	return func(req stream.ExternalRequest) (RequestResult, error) {
+		{
+			pr := req.RequestParams.(PromotionRequest)
+
+			mutex.Lock()
+			(*captured)[pr]++
+			mutex.Unlock()
+
+			// Add a request result to signal the processing of this "request".
+			// This is necessary because ExecRequests() is the sole function in
+			// the codebase that decrements the WaitGroup semaphore.
+			return RequestResult{}, nil
+		}
+	}
+}
+
 func supportedMediaType(v string) (ggcrV1Types.MediaType, error) {
 	switch ggcrV1Types.MediaType(v) {
 	case ggcrV1Types.DockerManifestList: