Repository: apigee/registry (go)

Issue:
Registry tool: errors encountered by concurrent workers are logged and not returned back to the calling command
As noted in the discussion of https://github.com/apigee/registry/pull/910#issuecomment-1377552891, when concurrent workers encounter errors, they [log them](https://github.com/apigee/registry/blob/98c9c9ae95a0874448f0ebc2e738a3c9ca68468d/cmd/registry/core/tasks.go#L81) instead of returning them to the calling command.

This conflicts with the goal in #827 of making error conditions testable.

Pull request description:
return task error from WorkerPool
Fixes #917

Patch:
diff --git a/cmd/registry/cmd/annotate/annotate.go b/cmd/registry/cmd/annotate/annotate.go
index bcf6da4..2310731 100644
--- a/cmd/registry/cmd/annotate/annotate.go
+++ b/cmd/registry/cmd/annotate/annotate.go
@@ -55,7 +55,7 @@ func Command() *cobra.Command {
 				log.FromContext(ctx).WithError(err).Fatal("Failed to get client")
 			}
 
-			taskQueue, wait := tasks.WorkerPool(ctx, jobs)
+			taskQueue, wait := tasks.WorkerPoolIgnoreError(ctx, jobs)
 			defer wait()
 
 			valuesToClear := make([]string, 0)
diff --git a/cmd/registry/cmd/check/lint/checker.go b/cmd/registry/cmd/check/lint/checker.go
index a51db6e..4fcc093 100644
--- a/cmd/registry/cmd/check/lint/checker.go
+++ b/cmd/registry/cmd/check/lint/checker.go
@@ -73,7 +73,7 @@ func (l *Checker) Check(ctx context.Context, admin connection.AdminClient, clien
 
 	// enable rules to access client
 	ctx = context.WithValue(ctx, ContextKeyRegistryClient, client)
-	taskQueue, wait := tasks.WorkerPool(ctx, jobs)
+	taskQueue, wait := tasks.WorkerPoolIgnoreError(ctx, jobs)
 	defer func() {
 		wait()
 		if response.Error != "" { // from a panic
diff --git a/cmd/registry/cmd/compute/complexity/complexity.go b/cmd/registry/cmd/compute/complexity/complexity.go
index f583f2f..16f198f 100644
--- a/cmd/registry/cmd/compute/complexity/complexity.go
+++ b/cmd/registry/cmd/compute/complexity/complexity.go
@@ -68,7 +68,7 @@ func Command() *cobra.Command {
 			if err != nil {
 				log.FromContext(ctx).WithError(err).Fatal("Failed to get jobs from flags")
 			}
-			taskQueue, wait := tasks.WorkerPool(ctx, jobs)
+			taskQueue, wait := tasks.WorkerPoolIgnoreError(ctx, jobs)
 			defer wait()
 
 			parsed, err := names.ParseSpecRevision(args[0])
diff --git a/cmd/registry/cmd/compute/conformance/conformance.go b/cmd/registry/cmd/compute/conformance/conformance.go
index 8a61db6..86d9d14 100644
--- a/cmd/registry/cmd/compute/conformance/conformance.go
+++ b/cmd/registry/cmd/compute/conformance/conformance.go
@@ -119,7 +119,7 @@ func processStyleGuide(ctx context.Context, client connection.RegistryClient, st
 		return
 	}
 
-	taskQueue, wait := tasks.WorkerPool(ctx, jobs)
+	taskQueue, wait := tasks.WorkerPoolIgnoreError(ctx, jobs)
 	defer wait()
 
 	for _, spec := range specs {
diff --git a/cmd/registry/cmd/compute/lint/lint.go b/cmd/registry/cmd/compute/lint/lint.go
index be05805..cc49959 100644
--- a/cmd/registry/cmd/compute/lint/lint.go
+++ b/cmd/registry/cmd/compute/lint/lint.go
@@ -63,7 +63,7 @@ func Command() *cobra.Command {
 			if err != nil {
 				log.FromContext(ctx).WithError(err).Fatal("Failed to get jobs from flags")
 			}
-			taskQueue, wait := tasks.WorkerPool(ctx, jobs)
+			taskQueue, wait := tasks.WorkerPoolIgnoreError(ctx, jobs)
 			defer wait()
 
 			spec, err := names.ParseSpec(args[0])
diff --git a/cmd/registry/cmd/compute/score/score.go b/cmd/registry/cmd/compute/score/score.go
index 39a14b1..313c049 100644
--- a/cmd/registry/cmd/compute/score/score.go
+++ b/cmd/registry/cmd/compute/score/score.go
@@ -61,7 +61,7 @@ func Command() *cobra.Command {
 			if err != nil {
 				log.FromContext(ctx).WithError(err).Fatal("Failed to get jobs from flags")
 			}
-			taskQueue, wait := tasks.WorkerPool(ctx, jobs)
+			taskQueue, wait := tasks.WorkerPoolIgnoreError(ctx, jobs)
 			defer wait()
 
 			inputPattern, err := patterns.ParseResourcePattern(args[0])
diff --git a/cmd/registry/cmd/compute/scorecard/scorecard.go b/cmd/registry/cmd/compute/scorecard/scorecard.go
index d3487b5..26b642b 100644
--- a/cmd/registry/cmd/compute/scorecard/scorecard.go
+++ b/cmd/registry/cmd/compute/scorecard/scorecard.go
@@ -62,7 +62,7 @@ func Command() *cobra.Command {
 			if err != nil {
 				log.FromContext(ctx).WithError(err).Fatal("Failed to get jobs from flags")
 			}
-			taskQueue, wait := tasks.WorkerPool(ctx, jobs)
+			taskQueue, wait := tasks.WorkerPoolIgnoreError(ctx, jobs)
 			defer wait()
 
 			inputPattern, err := patterns.ParseResourcePattern(args[0])
diff --git a/cmd/registry/cmd/compute/vocabulary/vocabulary.go b/cmd/registry/cmd/compute/vocabulary/vocabulary.go
index 32fce99..da6a48d 100644
--- a/cmd/registry/cmd/compute/vocabulary/vocabulary.go
+++ b/cmd/registry/cmd/compute/vocabulary/vocabulary.go
@@ -67,7 +67,7 @@ func Command() *cobra.Command {
 			if err != nil {
 				log.FromContext(ctx).WithError(err).Fatal("Failed to get jobs from flags")
 			}
-			taskQueue, wait := tasks.WorkerPool(ctx, jobs)
+			taskQueue, wait := tasks.WorkerPoolIgnoreError(ctx, jobs)
 			defer wait()
 
 			parsed, err := names.ParseSpecRevision(path)
diff --git a/cmd/registry/cmd/delete/delete.go b/cmd/registry/cmd/delete/delete.go
index b169c97..9b5c331 100644
--- a/cmd/registry/cmd/delete/delete.go
+++ b/cmd/registry/cmd/delete/delete.go
@@ -69,7 +69,7 @@ func Command() *cobra.Command {
 				return errors.New("no resources found")
 			}
 			// Initialize task queue.
-			taskQueue, wait := tasks.WorkerPool(ctx, jobs)
+			taskQueue, wait := tasks.WorkerPoolIgnoreError(ctx, jobs)
 			defer wait()
 			// Delete all of the resources that were found.
 			for _, task := range v.tasks {
diff --git a/cmd/registry/cmd/export/export.go b/cmd/registry/cmd/export/export.go
index 7fb0e91..4b7991b 100644
--- a/cmd/registry/cmd/export/export.go
+++ b/cmd/registry/cmd/export/export.go
@@ -52,7 +52,7 @@ func Command() *cobra.Command {
 				return err
 			}
 			// Initialize task queue.
-			taskQueue, wait := tasks.WorkerPool(ctx, jobs)
+			taskQueue, wait := tasks.WorkerPoolIgnoreError(ctx, jobs)
 			defer wait()
 			// Create the visitor that will perform exports.
 			v := &exportVisitor{
diff --git a/cmd/registry/cmd/label/label.go b/cmd/registry/cmd/label/label.go
index 2aafa27..c1cb200 100644
--- a/cmd/registry/cmd/label/label.go
+++ b/cmd/registry/cmd/label/label.go
@@ -54,7 +54,7 @@ func Command() *cobra.Command {
 				log.FromContext(ctx).WithError(err).Fatal("Failed to get client")
 			}
 
-			taskQueue, wait := tasks.WorkerPool(ctx, jobs)
+			taskQueue, wait := tasks.WorkerPoolIgnoreError(ctx, jobs)
 			defer wait()
 
 			valuesToClear := make([]string, 0)
diff --git a/cmd/registry/cmd/resolve/resolve.go b/cmd/registry/cmd/resolve/resolve.go
index 5972dff..aaf6a3d 100644
--- a/cmd/registry/cmd/resolve/resolve.go
+++ b/cmd/registry/cmd/resolve/resolve.go
@@ -108,7 +108,7 @@ func Command() *cobra.Command {
 			}
 
 			log.Debug(ctx, "Starting execution...")
-			taskQueue, wait := tasks.WorkerPool(ctx, jobs)
+			taskQueue, wait := tasks.WorkerPoolIgnoreError(ctx, jobs)
 			defer wait()
 			// Submit tasks to taskQueue
 			for i := 0; i < len(actions) && i < maxActions; i++ {
diff --git a/cmd/registry/cmd/upload/csv.go b/cmd/registry/cmd/upload/csv.go
index 96d286c..061bae4 100644
--- a/cmd/registry/cmd/upload/csv.go
+++ b/cmd/registry/cmd/upload/csv.go
@@ -68,7 +68,7 @@ func csvCommand() *cobra.Command {
 				return fmt.Errorf("parent does not exist (%s)", err)
 			}
 
-			taskQueue, wait := tasks.WorkerPool(ctx, jobs)
+			taskQueue, wait := tasks.WorkerPoolIgnoreError(ctx, jobs)
 			defer wait()
 
 			file, err := os.Open(args[0])
diff --git a/cmd/registry/cmd/upload/discovery.go b/cmd/registry/cmd/upload/discovery.go
index 7accb64..b539b76 100644
--- a/cmd/registry/cmd/upload/discovery.go
+++ b/cmd/registry/cmd/upload/discovery.go
@@ -71,7 +71,7 @@ func discoveryCommand() *cobra.Command {
 			if err != nil {
 				log.FromContext(ctx).WithError(err).Fatal("Failed to get jobs from flags")
 			}
-			taskQueue, wait := tasks.WorkerPool(ctx, jobs)
+			taskQueue, wait := tasks.WorkerPoolIgnoreError(ctx, jobs)
 			defer wait()
 
 			discoveryResponse, err := fetchDiscoveryList(service)
diff --git a/cmd/registry/cmd/upload/openapi.go b/cmd/registry/cmd/upload/openapi.go
index 9eeb686..da68c9b 100644
--- a/cmd/registry/cmd/upload/openapi.go
+++ b/cmd/registry/cmd/upload/openapi.go
@@ -62,7 +62,7 @@ func openAPICommand() *cobra.Command {
 			if err != nil {
 				log.FromContext(ctx).WithError(err).Fatal("Failed to get jobs from flags")
 			}
-			taskQueue, wait := tasks.WorkerPool(ctx, jobs)
+			taskQueue, wait := tasks.WorkerPoolIgnoreError(ctx, jobs)
 			defer wait()
 
 			for _, arg := range args {
diff --git a/cmd/registry/cmd/upload/protos.go b/cmd/registry/cmd/upload/protos.go
index 0c095b9..0dd8cc3 100644
--- a/cmd/registry/cmd/upload/protos.go
+++ b/cmd/registry/cmd/upload/protos.go
@@ -79,7 +79,7 @@ func protosCommand() *cobra.Command {
 			if err != nil {
 				log.FromContext(ctx).WithError(err).Fatal("Failed to get jobs from flags")
 			}
-			taskQueue, wait := tasks.WorkerPool(ctx, jobs)
+			taskQueue, wait := tasks.WorkerPoolIgnoreError(ctx, jobs)
 			defer wait()
 
 			for _, arg := range args {
diff --git a/cmd/registry/patch/apply.go b/cmd/registry/patch/apply.go
index 8d4047a..c204104 100644
--- a/cmd/registry/patch/apply.go
+++ b/cmd/registry/patch/apply.go
@@ -155,7 +155,7 @@ func (p *patchGroup) run(ctx context.Context, jobs int) error {
 		p.deploymentTasks,
 		p.artifactTasks,
 	} {
-		taskQueue, wait := tasks.WorkerPool(ctx, jobs)
+		taskQueue, wait := tasks.WorkerPoolIgnoreError(ctx, jobs)
 		for _, task := range taskLists {
 			taskQueue <- task
 		}
diff --git a/cmd/registry/tasks/tasks.go b/cmd/registry/tasks/tasks.go
index 69fb154..a2bcd44 100644
--- a/cmd/registry/tasks/tasks.go
+++ b/cmd/registry/tasks/tasks.go
@@ -1,4 +1,4 @@
-// Copyright 2020 Google LLC. All Rights Reserved.
+// Copyright 2023 Google LLC. All Rights Reserved.
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
@@ -19,6 +19,7 @@ import (
 	"sync"
 
 	"github.com/apigee/registry/pkg/log"
+	"golang.org/x/sync/errgroup"
 )
 
 // Task is a generic interface for a runnable operation
@@ -27,39 +28,64 @@ type Task interface {
 	String() string
 }
 
-// This function creates a waitgroup and a taskQueue for the workerPool.
+// WorkerPool creates a waitgroup and a taskQueue for a worker pool.
 // It will create "n" workers which will listen for Tasks on the taskQueue.
-// It returns the taskQueue and a wait func.
-// The clients should add new tasks to this taskQueue
-// and call the call the wait func when done.
-// Do not separately close the taskQueue, make use of the wait func.
-func WorkerPool(ctx context.Context, n int) (chan<- Task, func()) {
-	var wg sync.WaitGroup
+// The return value is the taskQueue and a wait function.
+// Do not directly close the taskQueue, use the wait function.
+// Clients should add new tasks to this taskQueue and call the wait function
+// when done. If a worker fails with an error, the task context will be
+// canceled, instructing the queue and workers to terminate. Tasks should
+// check ctx as appropriate. The first error encountered will be returned
+// from the wait function and all errors will be logged at Warn level.
+func WorkerPool(ctx context.Context, n int) (chan<- Task, func() error) {
+	eg, ctx := errgroup.WithContext(ctx)
+	eg.SetLimit(n) // pool size
 	taskQueue := make(chan Task, 1024)
-	for i := 0; i < n; i++ {
-		wg.Add(1)
-		go worker(ctx, &wg, taskQueue, false)
-	}
+	var wg sync.WaitGroup
+	wg.Add(1)
 
-	wait := func() {
+	go func() {
+		for task := range taskQueue {
+			select {
+			case <-ctx.Done():
+				return
+			default:
+				t := task
+				f := func() error {
+					err := t.Run(ctx)
+					if err != nil {
+						log.FromContext(ctx).WithError(err).Warnf("task failed: %s", task)
+					}
+					return err
+				}
+				eg.Go(f) // blocks at n runners
+			}
+		}
+		wg.Done()
+	}()
+
+	wait := func() error {
 		close(taskQueue)
-		wg.Wait()
+		wg.Wait()        // wait for all work to be added
+		err := eg.Wait() // wait for all work to be completed
+		if err != nil {
+			log.FromContext(ctx).WithError(err).Warnf("WorkerPool terminated")
+		}
+		return err
 	}
 
 	return taskQueue, wait
 }
 
-// A worker which pulls tasks from the taskQueue, executes them and logs errors if any.
-func worker(ctx context.Context, wg *sync.WaitGroup, taskQueue <-chan Task, warnOnError bool) {
-	defer wg.Done()
-	for task := range taskQueue {
-		select {
-		case <-ctx.Done():
-			return
-		default:
-			if err := task.Run(ctx); err != nil {
-				log.FromContext(ctx).WithError(err).Warnf("Task failed: %s", task)
-			}
+// WorkerPoolIgnoreError does the same as WorkerPool, but returns a wait
+// function that captures and logs the error at Error level. This is convenient
+// for defer.
+func WorkerPoolIgnoreError(ctx context.Context, n int) (chan<- Task, func()) {
+	wp, w := WorkerPool(ctx, n)
+	wait := func() {
+		if err := w(); err != nil {
+			log.FromContext(ctx).WithError(err).Errorf("unhandled WorkerPool error")
 		}
 	}
+	return wp, wait
 }
diff --git a/go.mod b/go.mod
index 2c4c916..466ed4b 100644
--- a/go.mod
+++ b/go.mod
@@ -82,7 +82,7 @@ require (
 	go.uber.org/zap v1.23.0 // indirect
 	golang.org/x/crypto v0.0.0-20220926161630-eccd6366d1be // indirect
 	golang.org/x/net v0.0.0-20221002022538-bcab6841153b // indirect
-	golang.org/x/sync v0.0.0-20220929204114-8fcdb60fdcc0 // indirect
+	golang.org/x/sync v0.0.0-20220929204114-8fcdb60fdcc0
 	golang.org/x/sys v0.0.0-20220928140112-f11e5e49a4ec // indirect
 	golang.org/x/text v0.3.7 // indirect
 	golang.org/x/time v0.0.0-20220922220347-f3bd1da661af // indirect
diff --git a/pkg/wipeout/wipeout.go b/pkg/wipeout/wipeout.go
index 0e080b0..abdc9d2 100644
--- a/pkg/wipeout/wipeout.go
+++ b/pkg/wipeout/wipeout.go
@@ -30,13 +30,13 @@ func Wipeout(ctx context.Context, client connection.RegistryClient, projectID st
 	project := "projects/" + projectID + "/locations/global"
 	{
 		log.Debugf(ctx, "Deleting apis")
-		taskQueue, wait := tasks.WorkerPool(ctx, jobs)
+		taskQueue, wait := tasks.WorkerPoolIgnoreError(ctx, jobs)
 		wipeoutApis(ctx, client, taskQueue, project)
 		wait()
 	}
 	{
 		log.Debugf(ctx, "Deleting artifacts")
-		taskQueue, wait := tasks.WorkerPool(ctx, jobs)
+		taskQueue, wait := tasks.WorkerPoolIgnoreError(ctx, jobs)
 		wipeoutArtifacts(ctx, client, taskQueue, project)
 		wait()
 	}


Test patch:
diff --git a/cmd/registry/patch/export_test.go b/cmd/registry/patch/export_test.go
index df2a146..2281538 100644
--- a/cmd/registry/patch/export_test.go
+++ b/cmd/registry/patch/export_test.go
@@ -58,7 +58,7 @@ func TestExport(t *testing.T) {
 
 		t.Run(test.desc+"-project", func(t *testing.T) {
 			tempDir := t.TempDir()
-			taskQueue, wait := tasks.WorkerPool(ctx, 1)
+			taskQueue, wait := tasks.WorkerPoolIgnoreError(ctx, 1)
 			err = ExportProject(ctx, registryClient, project, tempDir, taskQueue)
 			if err != nil {
 				t.Errorf("Failed to export project: %s", err)
@@ -69,7 +69,7 @@ func TestExport(t *testing.T) {
 
 		t.Run(test.desc+"-api", func(t *testing.T) {
 			tempDir := t.TempDir()
-			taskQueue, wait := tasks.WorkerPool(ctx, 1)
+			taskQueue, wait := tasks.WorkerPoolIgnoreError(ctx, 1)
 			err = ExportAPI(ctx, registryClient, project.Api("registry"), false, tempDir, taskQueue)
 			if err != nil {
 				t.Errorf("Failed to export api: %s", err)
@@ -80,7 +80,7 @@ func TestExport(t *testing.T) {
 
 		t.Run(test.desc+"-api-recursive", func(t *testing.T) {
 			tempDir := t.TempDir()
-			taskQueue, wait := tasks.WorkerPool(ctx, 1)
+			taskQueue, wait := tasks.WorkerPoolIgnoreError(ctx, 1)
 			err = ExportAPI(ctx, registryClient, project.Api("registry"), true, tempDir, taskQueue)
 			if err != nil {
 				t.Errorf("Failed to export api: %s", err)
@@ -91,7 +91,7 @@ func TestExport(t *testing.T) {
 
 		t.Run(test.desc+"-version", func(t *testing.T) {
 			tempDir := t.TempDir()
-			taskQueue, wait := tasks.WorkerPool(ctx, 1)
+			taskQueue, wait := tasks.WorkerPoolIgnoreError(ctx, 1)
 			err = ExportAPIVersion(ctx, registryClient, project.Api("registry").Version("v1"), false, tempDir, taskQueue)
 			if err != nil {
 				t.Errorf("Failed to export version: %s", err)
@@ -102,7 +102,7 @@ func TestExport(t *testing.T) {
 
 		t.Run(test.desc+"-version-recursive", func(t *testing.T) {
 			tempDir := t.TempDir()
-			taskQueue, wait := tasks.WorkerPool(ctx, 1)
+			taskQueue, wait := tasks.WorkerPoolIgnoreError(ctx, 1)
 			err = ExportAPIVersion(ctx, registryClient, project.Api("registry").Version("v1"), true, tempDir, taskQueue)
 			if err != nil {
 				t.Errorf("Failed to export version: %s", err)
@@ -113,7 +113,7 @@ func TestExport(t *testing.T) {
 
 		t.Run(test.desc+"-spec", func(t *testing.T) {
 			tempDir := t.TempDir()
-			taskQueue, wait := tasks.WorkerPool(ctx, 1)
+			taskQueue, wait := tasks.WorkerPoolIgnoreError(ctx, 1)
 			err = ExportAPISpec(ctx, registryClient, project.Api("registry").Version("v1").Spec("openapi"), false, tempDir, taskQueue)
 			if err != nil {
 				t.Errorf("Failed to export spec: %s", err)
@@ -125,7 +125,7 @@ func TestExport(t *testing.T) {
 
 		t.Run(test.desc+"-spec-recursive", func(t *testing.T) {
 			tempDir := t.TempDir()
-			taskQueue, wait := tasks.WorkerPool(ctx, 1)
+			taskQueue, wait := tasks.WorkerPoolIgnoreError(ctx, 1)
 			err = ExportAPISpec(ctx, registryClient, project.Api("registry").Version("v1").Spec("openapi"), true, tempDir, taskQueue)
 			if err != nil {
 				t.Errorf("Failed to export spec: %s", err)
@@ -136,7 +136,7 @@ func TestExport(t *testing.T) {
 
 		t.Run(test.desc+"-deployment", func(t *testing.T) {
 			tempDir := t.TempDir()
-			taskQueue, wait := tasks.WorkerPool(ctx, 1)
+			taskQueue, wait := tasks.WorkerPoolIgnoreError(ctx, 1)
 			err = ExportAPIDeployment(ctx, registryClient, project.Api("registry").Deployment("prod"), false, tempDir, taskQueue)
 			if err != nil {
 				t.Errorf("Failed to export deployment: %s", err)
@@ -147,7 +147,7 @@ func TestExport(t *testing.T) {
 
 		t.Run(test.desc+"-deployment-recursive", func(t *testing.T) {
 			tempDir := t.TempDir()
-			taskQueue, wait := tasks.WorkerPool(ctx, 1)
+			taskQueue, wait := tasks.WorkerPoolIgnoreError(ctx, 1)
 			err = ExportAPIDeployment(ctx, registryClient, project.Api("registry").Deployment("prod"), true, tempDir, taskQueue)
 			if err != nil {
 				t.Errorf("Failed to export deployment: %s", err)
@@ -158,7 +158,7 @@ func TestExport(t *testing.T) {
 
 		t.Run(test.desc+"-artifact", func(t *testing.T) {
 			tempDir := t.TempDir()
-			taskQueue, wait := tasks.WorkerPool(ctx, 1)
+			taskQueue, wait := tasks.WorkerPoolIgnoreError(ctx, 1)
 			err = ExportArtifact(ctx, registryClient, project.Api("registry").Artifact("api-references"), tempDir, taskQueue)
 			if err != nil {
 				t.Errorf("Failed to export artifact: %s", err)
diff --git a/cmd/registry/tasks/tasks_test.go b/cmd/registry/tasks/tasks_test.go
index ce2787d..fd9498f 100644
--- a/cmd/registry/tasks/tasks_test.go
+++ b/cmd/registry/tasks/tasks_test.go
@@ -17,54 +17,97 @@ package tasks
 import (
 	"context"
 	"errors"
-	"fmt"
+	"sync"
 	"testing"
 )
 
 func TestWorkerPool(t *testing.T) {
 	ctx := context.Background()
 	jobs := 100
+	counter := new(atomicInt32)
+	want := 1000
 
-	taskQueue, wait := WorkerPool(ctx, jobs)
-	defer wait()
+	func() {
+		taskQueue, wait := WorkerPool(ctx, jobs)
+		for i := 0; i < want; i++ {
+			taskQueue <- &incrTask{counter}
+		}
+		if err := wait(); err != nil {
+			t.Error(err)
+		}
+	}()
 
-	for i := 0; i < 1000; i++ {
-		taskQueue <- &doNothingTask{i: i}
+	got := counter.Load()
+	if got != int32(want) {
+		t.Errorf("want %d got: %d", want, got)
 	}
 }
 
-type doNothingTask struct {
-	i int
+type incrTask struct {
+	counter *atomicInt32
 }
 
-func (task *doNothingTask) String() string {
-	return fmt.Sprintf("do nothing %d", task.i)
+func (t *incrTask) String() string {
+	return "add 1"
 }
 
-func (task *doNothingTask) Run(ctx context.Context) error {
+func (t *incrTask) Run(ctx context.Context) error {
+	t.counter.Add(1)
 	return nil
 }
 
-func TestWorkerPoolWithWarnings(t *testing.T) {
+func TestWorkerPoolWithError(t *testing.T) {
 	ctx := context.Background()
 	jobs := 1
 
 	taskQueue, wait := WorkerPool(ctx, jobs)
-	defer wait()
+	for i := 0; i < 2; i++ {
+		taskQueue <- &failTask{}
+	}
+	if err := wait(); err == nil {
+		t.Error("want error, got nil")
+	}
+}
+
+func TestWorkerPoolIgnoreError(t *testing.T) {
+	ctx := context.Background()
+	jobs := 1
 
-	for i := 0; i < 10; i++ {
-		taskQueue <- &failTask{i: i}
+	taskQueue, wait := WorkerPoolIgnoreError(ctx, jobs)
+	defer wait()
+	for i := 0; i < 2; i++ {
+		taskQueue <- &failTask{}
 	}
 }
 
 type failTask struct {
-	i int
 }
 
 func (task *failTask) String() string {
-	return fmt.Sprintf("do nothing %d", task.i)
+	return "fail task"
 }
 
 func (task *failTask) Run(ctx context.Context) error {
 	return errors.New("fail")
 }
+
+// can be replaced by atomic.Int32 when we move to go 1.19+
+type atomicInt32 struct {
+	sync.RWMutex
+	v int32
+}
+
+// Load atomically loads and returns the value stored in x.
+func (a *atomicInt32) Load() int32 {
+	a.RLock()
+	defer a.RUnlock()
+	return a.v
+}
+
+// Add atomically adds delta to x and returns the new value.
+func (a *atomicInt32) Add(delta int32) int32 {
+	a.Lock()
+	defer a.Unlock()
+	a.v = a.v + delta
+	return a.v
+}
diff --git a/tests/benchmark/benchmark_artifacts_test.go b/tests/benchmark/benchmark_artifacts_test.go
index e24211a..93430b6 100644
--- a/tests/benchmark/benchmark_artifacts_test.go
+++ b/tests/benchmark/benchmark_artifacts_test.go
@@ -114,7 +114,7 @@ func (task *seedArtifactsTask) Run(ctx context.Context) error {
 func BenchmarkListArtifacts(b *testing.B) {
 	ctx, client := setup(b)
 	// Try to speed up creation with a worker pool (this might not help much but seemed worth a try).
-	taskQueue, wait := tasks.WorkerPool(ctx, 5)
+	taskQueue, wait := tasks.WorkerPoolIgnoreError(ctx, 5)
 	projectName := names.Project{ProjectID: projectID}
 	total := 1000
 	step := 5