Repository: apigee/registry
Pull request #1016 (2023-02-10 02:00:32)

move core/tasks.go to tasks/tasks.go
Also for #1007, this moves tasks.go into a dedicated package. 

(We need #917 so that we can get rid of those logging statements)

codecov[bot]: # [Codecov](https://codecov.io/gh/apigee/registry/pull/1016?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apigee) Report
> Merging [#1016](https://codecov.io/gh/apigee/registry/pull/1016?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apigee) (ad58220) into [main](https://codecov.io/gh/apigee/registry/commit/ef112b6e1ddd2128a2fe06780a95dcae5231a89f?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apigee) (ef112b6) will **not change** coverage.
> The diff coverage is `95.00%`.

```diff
@@           Coverage Diff           @@
##             main    #1016   +/-   ##
=======================================
  Coverage   67.88%   67.88%           
=======================================
  Files         143      143           
  Lines       11769    11769           
=======================================
  Hits         7989     7989           
  Misses       3091     3091           
  Partials      689      689           
```


| [Impacted Files](https://codecov.io/gh/apigee/registry/pull/1016?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apigee) | Coverage Δ | |
|---|---|---|
| [cmd/registry/cmd/compute/lint/lint.go](https://codecov.io/gh/apigee/registry/pull/1016?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apigee#diff-Y21kL3JlZ2lzdHJ5L2NtZC9jb21wdXRlL2xpbnQvbGludC5nbw==) | `7.40% <0.00%> (ø)` | |
| [cmd/registry/cmd/compute/vocabulary/vocabulary.go](https://codecov.io/gh/apigee/registry/pull/1016?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apigee#diff-Y21kL3JlZ2lzdHJ5L2NtZC9jb21wdXRlL3ZvY2FidWxhcnkvdm9jYWJ1bGFyeS5nbw==) | `4.46% <0.00%> (ø)` | |
| [cmd/registry/cmd/annotate/annotate.go](https://codecov.io/gh/apigee/registry/pull/1016?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apigee#diff-Y21kL3JlZ2lzdHJ5L2NtZC9hbm5vdGF0ZS9hbm5vdGF0ZS5nbw==) | `80.11% <100.00%> (ø)` | |
| [cmd/registry/cmd/check/lint/checker.go](https://codecov.io/gh/apigee/registry/pull/1016?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apigee#diff-Y21kL3JlZ2lzdHJ5L2NtZC9jaGVjay9saW50L2NoZWNrZXIuZ28=) | `76.92% <100.00%> (ø)` | |
| [cmd/registry/cmd/compute/complexity/complexity.go](https://codecov.io/gh/apigee/registry/pull/1016?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apigee#diff-Y21kL3JlZ2lzdHJ5L2NtZC9jb21wdXRlL2NvbXBsZXhpdHkvY29tcGxleGl0eS5nbw==) | `42.40% <100.00%> (ø)` | |
| [...md/registry/cmd/compute/conformance/conformance.go](https://codecov.io/gh/apigee/registry/pull/1016?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apigee#diff-Y21kL3JlZ2lzdHJ5L2NtZC9jb21wdXRlL2NvbmZvcm1hbmNlL2NvbmZvcm1hbmNlLmdv) | `55.17% <100.00%> (ø)` | |
| [cmd/registry/cmd/compute/score/score.go](https://codecov.io/gh/apigee/registry/pull/1016?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apigee#diff-Y21kL3JlZ2lzdHJ5L2NtZC9jb21wdXRlL3Njb3JlL3Njb3JlLmdv) | `56.71% <100.00%> (ø)` | |
| [cmd/registry/cmd/compute/scorecard/scorecard.go](https://codecov.io/gh/apigee/registry/pull/1016?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apigee#diff-Y21kL3JlZ2lzdHJ5L2NtZC9jb21wdXRlL3Njb3JlY2FyZC9zY29yZWNhcmQuZ28=) | `56.06% <100.00%> (ø)` | |
| [cmd/registry/cmd/delete/delete.go](https://codecov.io/gh/apigee/registry/pull/1016?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apigee#diff-Y21kL3JlZ2lzdHJ5L2NtZC9kZWxldGUvZGVsZXRlLmdv) | `94.19% <100.00%> (ø)` | |
| [cmd/registry/cmd/export/export.go](https://codecov.io/gh/apigee/registry/pull/1016?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apigee#diff-Y21kL3JlZ2lzdHJ5L2NtZC9leHBvcnQvZXhwb3J0Lmdv) | `75.22% <100.00%> (ø)` | |
| ... and [9 more](https://codecov.io/gh/apigee/registry/pull/1016?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apigee) | |

:mega: We’re building smart automated test selection to slash your CI/CD build times. [Learn more](https://about.codecov.io/iterative-testing/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apigee)


Patch:
diff --git a/cmd/registry/cmd/annotate/annotate.go b/cmd/registry/cmd/annotate/annotate.go
index abd64fd..2ec491d 100644
--- a/cmd/registry/cmd/annotate/annotate.go
+++ b/cmd/registry/cmd/annotate/annotate.go
@@ -20,7 +20,7 @@ import (
 	"strings"
 
 	"github.com/apigee/registry/cmd/registry/cmd/label"
-	"github.com/apigee/registry/cmd/registry/core"
+	"github.com/apigee/registry/cmd/registry/tasks"
 	"github.com/apigee/registry/gapic"
 	"github.com/apigee/registry/log"
 	"github.com/apigee/registry/pkg/connection"
@@ -55,7 +55,7 @@ func Command() *cobra.Command {
 				log.FromContext(ctx).WithError(err).Fatal("Failed to get client")
 			}
 
-			taskQueue, wait := core.WorkerPool(ctx, jobs)
+			taskQueue, wait := tasks.WorkerPool(ctx, jobs)
 			defer wait()
 
 			valuesToClear := make([]string, 0)
@@ -92,7 +92,7 @@ func Command() *cobra.Command {
 func matchAndHandleAnnotateCmd(
 	ctx context.Context,
 	client connection.RegistryClient,
-	taskQueue chan<- core.Task,
+	taskQueue chan<- tasks.Task,
 	name string,
 	filter string,
 	labeling *label.Labeling,
@@ -127,7 +127,7 @@ func annotateAPIs(ctx context.Context,
 	api names.Api,
 	filterFlag string,
 	labeling *label.Labeling,
-	taskQueue chan<- core.Task) error {
+	taskQueue chan<- tasks.Task) error {
 	return visitor.ListAPIs(ctx, client, api, filterFlag, func(ctx context.Context, api *rpc.Api) error {
 		taskQueue <- &annotateApiTask{
 			client:   client,
@@ -144,7 +144,7 @@ func annotateVersions(
 	version names.Version,
 	filterFlag string,
 	labeling *label.Labeling,
-	taskQueue chan<- core.Task) error {
+	taskQueue chan<- tasks.Task) error {
 	return visitor.ListVersions(ctx, client, version, filterFlag, func(ctx context.Context, version *rpc.ApiVersion) error {
 		taskQueue <- &annotateVersionTask{
 			client:   client,
@@ -161,7 +161,7 @@ func annotateSpecs(
 	spec names.Spec,
 	filterFlag string,
 	labeling *label.Labeling,
-	taskQueue chan<- core.Task) error {
+	taskQueue chan<- tasks.Task) error {
 	return visitor.ListSpecs(ctx, client, spec, filterFlag, false, func(ctx context.Context, spec *rpc.ApiSpec) error {
 		taskQueue <- &annotateSpecTask{
 			client:   client,
@@ -178,7 +178,7 @@ func annotateDeployments(
 	deployment names.Deployment,
 	filterFlag string,
 	labeling *label.Labeling,
-	taskQueue chan<- core.Task) error {
+	taskQueue chan<- tasks.Task) error {
 	return visitor.ListDeployments(ctx, client, deployment, filterFlag, func(ctx context.Context, deployment *rpc.ApiDeployment) error {
 		taskQueue <- &annotateDeploymentTask{
 			client:     client,
diff --git a/cmd/registry/cmd/check/lint/checker.go b/cmd/registry/cmd/check/lint/checker.go
index 86fa306..ae76500 100644
--- a/cmd/registry/cmd/check/lint/checker.go
+++ b/cmd/registry/cmd/check/lint/checker.go
@@ -21,7 +21,7 @@ import (
 	"strings"
 	"sync"
 
-	"github.com/apigee/registry/cmd/registry/core"
+	"github.com/apigee/registry/cmd/registry/tasks"
 	"github.com/apigee/registry/pkg/connection"
 	"github.com/apigee/registry/pkg/names"
 	"github.com/apigee/registry/pkg/visitor"
@@ -72,7 +72,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 := core.WorkerPool(ctx, jobs)
+	taskQueue, wait := tasks.WorkerPool(ctx, jobs)
 	defer func() {
 		wait()
 		if response.Error != "" { // from a panic
@@ -163,7 +163,7 @@ func (c *checkTask) runAndRecoverFromPanics(ctx context.Context, rule Rule, reso
 }
 
 type listHandler struct {
-	taskQueue chan<- core.Task
+	taskQueue chan<- tasks.Task
 	newTask   func(r Resource) *checkTask
 }
 
diff --git a/cmd/registry/cmd/compute/complexity/complexity.go b/cmd/registry/cmd/compute/complexity/complexity.go
index 6e7b398..a89b7d0 100644
--- a/cmd/registry/cmd/compute/complexity/complexity.go
+++ b/cmd/registry/cmd/compute/complexity/complexity.go
@@ -21,6 +21,7 @@ import (
 
 	"github.com/apigee/registry/cmd/registry/compress"
 	"github.com/apigee/registry/cmd/registry/core"
+	"github.com/apigee/registry/cmd/registry/tasks"
 	"github.com/apigee/registry/cmd/registry/types"
 	"github.com/apigee/registry/log"
 	"github.com/apigee/registry/pkg/connection"
@@ -68,7 +69,7 @@ func Command() *cobra.Command {
 			if err != nil {
 				log.FromContext(ctx).WithError(err).Fatal("Failed to get jobs from flags")
 			}
-			taskQueue, wait := core.WorkerPool(ctx, jobs)
+			taskQueue, wait := tasks.WorkerPool(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 a8e754f..12eb026 100644
--- a/cmd/registry/cmd/compute/conformance/conformance.go
+++ b/cmd/registry/cmd/compute/conformance/conformance.go
@@ -19,7 +19,7 @@ import (
 	"fmt"
 
 	"github.com/apigee/registry/cmd/registry/conformance"
-	"github.com/apigee/registry/cmd/registry/core"
+	"github.com/apigee/registry/cmd/registry/tasks"
 	"github.com/apigee/registry/cmd/registry/types"
 	"github.com/apigee/registry/log"
 	"github.com/apigee/registry/pkg/connection"
@@ -118,7 +118,7 @@ func processStyleGuide(ctx context.Context, client connection.RegistryClient, st
 		return
 	}
 
-	taskQueue, wait := core.WorkerPool(ctx, jobs)
+	taskQueue, wait := tasks.WorkerPool(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 6320831..ffc8944 100644
--- a/cmd/registry/cmd/compute/lint/lint.go
+++ b/cmd/registry/cmd/compute/lint/lint.go
@@ -19,6 +19,7 @@ import (
 	"fmt"
 
 	"github.com/apigee/registry/cmd/registry/core"
+	"github.com/apigee/registry/cmd/registry/tasks"
 	"github.com/apigee/registry/cmd/registry/types"
 	"github.com/apigee/registry/log"
 	"github.com/apigee/registry/pkg/connection"
@@ -62,7 +63,7 @@ func Command() *cobra.Command {
 			if err != nil {
 				log.FromContext(ctx).WithError(err).Fatal("Failed to get jobs from flags")
 			}
-			taskQueue, wait := core.WorkerPool(ctx, jobs)
+			taskQueue, wait := tasks.WorkerPool(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 5f35fc4..acd2be6 100644
--- a/cmd/registry/cmd/compute/score/score.go
+++ b/cmd/registry/cmd/compute/score/score.go
@@ -17,9 +17,9 @@ package score
 import (
 	"context"
 
-	"github.com/apigee/registry/cmd/registry/core"
 	"github.com/apigee/registry/cmd/registry/patterns"
 	"github.com/apigee/registry/cmd/registry/scoring"
+	"github.com/apigee/registry/cmd/registry/tasks"
 	"github.com/apigee/registry/log"
 	"github.com/apigee/registry/pkg/connection"
 	"github.com/apigee/registry/rpc"
@@ -55,7 +55,7 @@ func Command() *cobra.Command {
 			if err != nil {
 				log.FromContext(ctx).WithError(err).Fatal("Failed to get jobs from flags")
 			}
-			taskQueue, wait := core.WorkerPoolWithWarnings(ctx, jobs)
+			taskQueue, wait := tasks.WorkerPoolWithWarnings(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 b3358be..1dd8e8e 100644
--- a/cmd/registry/cmd/compute/scorecard/scorecard.go
+++ b/cmd/registry/cmd/compute/scorecard/scorecard.go
@@ -17,9 +17,9 @@ package scorecard
 import (
 	"context"
 
-	"github.com/apigee/registry/cmd/registry/core"
 	"github.com/apigee/registry/cmd/registry/patterns"
 	"github.com/apigee/registry/cmd/registry/scoring"
+	"github.com/apigee/registry/cmd/registry/tasks"
 	"github.com/apigee/registry/log"
 	"github.com/apigee/registry/pkg/connection"
 	"github.com/apigee/registry/rpc"
@@ -54,7 +54,7 @@ func Command() *cobra.Command {
 			if err != nil {
 				log.FromContext(ctx).WithError(err).Fatal("Failed to get jobs from flags")
 			}
-			taskQueue, wait := core.WorkerPoolWithWarnings(ctx, jobs)
+			taskQueue, wait := tasks.WorkerPoolWithWarnings(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 41fdfe5..c464814 100644
--- a/cmd/registry/cmd/compute/vocabulary/vocabulary.go
+++ b/cmd/registry/cmd/compute/vocabulary/vocabulary.go
@@ -19,6 +19,7 @@ import (
 	"fmt"
 
 	"github.com/apigee/registry/cmd/registry/core"
+	"github.com/apigee/registry/cmd/registry/tasks"
 	"github.com/apigee/registry/cmd/registry/types"
 	"github.com/apigee/registry/log"
 	"github.com/apigee/registry/pkg/connection"
@@ -67,7 +68,7 @@ func Command() *cobra.Command {
 			if err != nil {
 				log.FromContext(ctx).WithError(err).Fatal("Failed to get jobs from flags")
 			}
-			taskQueue, wait := core.WorkerPoolWithWarnings(ctx, jobs)
+			taskQueue, wait := tasks.WorkerPoolWithWarnings(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 0e43d11..74b4ed9 100644
--- a/cmd/registry/cmd/delete/delete.go
+++ b/cmd/registry/cmd/delete/delete.go
@@ -18,7 +18,7 @@ import (
 	"context"
 	"errors"
 
-	"github.com/apigee/registry/cmd/registry/core"
+	"github.com/apigee/registry/cmd/registry/tasks"
 	"github.com/apigee/registry/log"
 	"github.com/apigee/registry/pkg/connection"
 	"github.com/apigee/registry/pkg/visitor"
@@ -69,7 +69,7 @@ func Command() *cobra.Command {
 				return errors.New("no resources found")
 			}
 			// Initialize task queue.
-			taskQueue, wait := core.WorkerPool(ctx, jobs)
+			taskQueue, wait := tasks.WorkerPool(ctx, jobs)
 			defer wait()
 			// Delete all of the resources that were found.
 			for _, task := range v.tasks {
@@ -89,10 +89,10 @@ type deletionVisitor struct {
 	registryClient connection.RegistryClient
 	adminClient    connection.AdminClient
 	force          bool
-	tasks          []core.Task
+	tasks          []tasks.Task
 }
 
-func (v *deletionVisitor) enqueue(task core.Task) {
+func (v *deletionVisitor) enqueue(task tasks.Task) {
 	v.tasks = append(v.tasks, task)
 }
 
diff --git a/cmd/registry/cmd/export/export.go b/cmd/registry/cmd/export/export.go
index 8d0125a..7fb0e91 100644
--- a/cmd/registry/cmd/export/export.go
+++ b/cmd/registry/cmd/export/export.go
@@ -18,8 +18,8 @@ import (
 	"context"
 	"errors"
 
-	"github.com/apigee/registry/cmd/registry/core"
 	"github.com/apigee/registry/cmd/registry/patch"
+	"github.com/apigee/registry/cmd/registry/tasks"
 	"github.com/apigee/registry/pkg/connection"
 	"github.com/apigee/registry/pkg/names"
 	"github.com/apigee/registry/pkg/visitor"
@@ -52,7 +52,7 @@ func Command() *cobra.Command {
 				return err
 			}
 			// Initialize task queue.
-			taskQueue, wait := core.WorkerPool(ctx, jobs)
+			taskQueue, wait := tasks.WorkerPool(ctx, jobs)
 			defer wait()
 			// Create the visitor that will perform exports.
 			v := &exportVisitor{
@@ -91,7 +91,7 @@ type exportVisitor struct {
 	recursive      bool
 	root           string
 	count          int
-	taskQueue      chan<- core.Task
+	taskQueue      chan<- tasks.Task
 }
 
 func (h *exportVisitor) ProjectHandler() visitor.ProjectHandler {
diff --git a/cmd/registry/cmd/label/label.go b/cmd/registry/cmd/label/label.go
index 3d41634..24fa847 100644
--- a/cmd/registry/cmd/label/label.go
+++ b/cmd/registry/cmd/label/label.go
@@ -19,7 +19,7 @@ import (
 	"fmt"
 	"strings"
 
-	"github.com/apigee/registry/cmd/registry/core"
+	"github.com/apigee/registry/cmd/registry/tasks"
 	"github.com/apigee/registry/gapic"
 	"github.com/apigee/registry/log"
 	"github.com/apigee/registry/pkg/connection"
@@ -54,7 +54,7 @@ func Command() *cobra.Command {
 				log.FromContext(ctx).WithError(err).Fatal("Failed to get client")
 			}
 
-			taskQueue, wait := core.WorkerPool(ctx, jobs)
+			taskQueue, wait := tasks.WorkerPool(ctx, jobs)
 			defer wait()
 
 			valuesToClear := make([]string, 0)
@@ -91,7 +91,7 @@ func Command() *cobra.Command {
 func matchAndHandleLabelCmd(
 	ctx context.Context,
 	client connection.RegistryClient,
-	taskQueue chan<- core.Task,
+	taskQueue chan<- tasks.Task,
 	name string,
 	filter string,
 	labeling *Labeling,
@@ -126,7 +126,7 @@ func labelAPIs(ctx context.Context,
 	api names.Api,
 	filterFlag string,
 	labeling *Labeling,
-	taskQueue chan<- core.Task) error {
+	taskQueue chan<- tasks.Task) error {
 	return visitor.ListAPIs(ctx, client, api, filterFlag, func(ctx context.Context, api *rpc.Api) error {
 		taskQueue <- &labelApiTask{
 			client:   client,
@@ -143,7 +143,7 @@ func labelVersions(
 	version names.Version,
 	filterFlag string,
 	labeling *Labeling,
-	taskQueue chan<- core.Task) error {
+	taskQueue chan<- tasks.Task) error {
 	return visitor.ListVersions(ctx, client, version, filterFlag, func(ctx context.Context, version *rpc.ApiVersion) error {
 		taskQueue <- &labelVersionTask{
 			client:   client,
@@ -160,7 +160,7 @@ func labelSpecs(
 	spec names.Spec,
 	filterFlag string,
 	labeling *Labeling,
-	taskQueue chan<- core.Task) error {
+	taskQueue chan<- tasks.Task) error {
 	return visitor.ListSpecs(ctx, client, spec, filterFlag, false, func(ctx context.Context, spec *rpc.ApiSpec) error {
 		taskQueue <- &labelSpecTask{
 			client:   client,
@@ -177,7 +177,7 @@ func labelDeployments(
 	deployment names.Deployment,
 	filterFlag string,
 	labeling *Labeling,
-	taskQueue chan<- core.Task) error {
+	taskQueue chan<- tasks.Task) error {
 	return visitor.ListDeployments(ctx, client, deployment, filterFlag, func(ctx context.Context, deployment *rpc.ApiDeployment) error {
 		taskQueue <- &labelDeploymentTask{
 			client:     client,
diff --git a/cmd/registry/cmd/resolve/resolve.go b/cmd/registry/cmd/resolve/resolve.go
index d593754..4466f09 100644
--- a/cmd/registry/cmd/resolve/resolve.go
+++ b/cmd/registry/cmd/resolve/resolve.go
@@ -19,7 +19,7 @@ import (
 	"fmt"
 
 	"github.com/apigee/registry/cmd/registry/controller"
-	"github.com/apigee/registry/cmd/registry/core"
+	"github.com/apigee/registry/cmd/registry/tasks"
 	"github.com/apigee/registry/log"
 	"github.com/apigee/registry/pkg/connection"
 	"github.com/apigee/registry/pkg/names"
@@ -107,7 +107,7 @@ func Command() *cobra.Command {
 			}
 
 			log.Debug(ctx, "Starting execution...")
-			taskQueue, wait := core.WorkerPoolWithWarnings(ctx, jobs)
+			taskQueue, wait := tasks.WorkerPoolWithWarnings(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 3991d82..9147dc8 100644
--- a/cmd/registry/cmd/upload/csv.go
+++ b/cmd/registry/cmd/upload/csv.go
@@ -24,6 +24,7 @@ import (
 
 	"github.com/apigee/registry/cmd/registry/compress"
 	"github.com/apigee/registry/cmd/registry/core"
+	"github.com/apigee/registry/cmd/registry/tasks"
 	"github.com/apigee/registry/cmd/registry/types"
 	"github.com/apigee/registry/log"
 	"github.com/apigee/registry/pkg/connection"
@@ -65,7 +66,7 @@ func csvCommand() *cobra.Command {
 				return fmt.Errorf("parent does not exist (%s)", err)
 			}
 
-			taskQueue, wait := core.WorkerPool(ctx, jobs)
+			taskQueue, wait := tasks.WorkerPool(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 f2cba0f..4ee370f 100644
--- a/cmd/registry/cmd/upload/discovery.go
+++ b/cmd/registry/cmd/upload/discovery.go
@@ -21,6 +21,7 @@ import (
 
 	"github.com/apigee/registry/cmd/registry/compress"
 	"github.com/apigee/registry/cmd/registry/core"
+	"github.com/apigee/registry/cmd/registry/tasks"
 	"github.com/apigee/registry/cmd/registry/types"
 	"github.com/apigee/registry/log"
 	"github.com/apigee/registry/pkg/connection"
@@ -70,7 +71,7 @@ func discoveryCommand() *cobra.Command {
 			if err != nil {
 				log.FromContext(ctx).WithError(err).Fatal("Failed to get jobs from flags")
 			}
-			taskQueue, wait := core.WorkerPool(ctx, jobs)
+			taskQueue, wait := tasks.WorkerPool(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 48ae695..266385a 100644
--- a/cmd/registry/cmd/upload/openapi.go
+++ b/cmd/registry/cmd/upload/openapi.go
@@ -25,6 +25,7 @@ import (
 
 	"github.com/apigee/registry/cmd/registry/compress"
 	"github.com/apigee/registry/cmd/registry/core"
+	"github.com/apigee/registry/cmd/registry/tasks"
 	"github.com/apigee/registry/cmd/registry/types"
 	"github.com/apigee/registry/log"
 	"github.com/apigee/registry/pkg/connection"
@@ -61,7 +62,7 @@ func openAPICommand() *cobra.Command {
 			if err != nil {
 				log.FromContext(ctx).WithError(err).Fatal("Failed to get jobs from flags")
 			}
-			taskQueue, wait := core.WorkerPool(ctx, jobs)
+			taskQueue, wait := tasks.WorkerPool(ctx, jobs)
 			defer wait()
 
 			for _, arg := range args {
@@ -79,7 +80,7 @@ func openAPICommand() *cobra.Command {
 	return cmd
 }
 
-func scanDirectoryForOpenAPI(ctx context.Context, client connection.RegistryClient, parent, baseURI, directory string, taskQueue chan<- core.Task) {
+func scanDirectoryForOpenAPI(ctx context.Context, client connection.RegistryClient, parent, baseURI, directory string, taskQueue chan<- tasks.Task) {
 	// walk a directory hierarchy, uploading every API spec that matches a set of expected file names.
 	if err := filepath.Walk(directory, func(path string, info os.FileInfo, err error) error {
 		if err != nil {
diff --git a/cmd/registry/cmd/upload/protos.go b/cmd/registry/cmd/upload/protos.go
index c77acdd..7369c6e 100644
--- a/cmd/registry/cmd/upload/protos.go
+++ b/cmd/registry/cmd/upload/protos.go
@@ -28,6 +28,7 @@ import (
 
 	"github.com/apigee/registry/cmd/registry/compress"
 	"github.com/apigee/registry/cmd/registry/core"
+	"github.com/apigee/registry/cmd/registry/tasks"
 	"github.com/apigee/registry/cmd/registry/types"
 	"github.com/apigee/registry/log"
 	"github.com/apigee/registry/pkg/connection"
@@ -78,7 +79,7 @@ func protosCommand() *cobra.Command {
 			if err != nil {
 				log.FromContext(ctx).WithError(err).Fatal("Failed to get jobs from flags")
 			}
-			taskQueue, wait := core.WorkerPool(ctx, jobs)
+			taskQueue, wait := tasks.WorkerPool(ctx, jobs)
 			defer wait()
 
 			for _, arg := range args {
@@ -108,7 +109,7 @@ func protosCommand() *cobra.Command {
 	return cmd
 }
 
-func scanDirectoryForProtos(client connection.RegistryClient, parent, baseURI, start, root string, taskQueue chan<- core.Task) error {
+func scanDirectoryForProtos(client connection.RegistryClient, parent, baseURI, start, root string, taskQueue chan<- tasks.Task) error {
 	return filepath.Walk(start, func(filepath string, info os.FileInfo, err error) error {
 		if err != nil {
 			return err
diff --git a/cmd/registry/patch/apply.go b/cmd/registry/patch/apply.go
index ba3793e..39918e9 100644
--- a/cmd/registry/patch/apply.go
+++ b/cmd/registry/patch/apply.go
@@ -24,7 +24,7 @@ import (
 	"path/filepath"
 	"strings"
 
-	"github.com/apigee/registry/cmd/registry/core"
+	"github.com/apigee/registry/cmd/registry/tasks"
 	"github.com/apigee/registry/log"
 	"github.com/apigee/registry/pkg/connection"
 	"gopkg.in/yaml.v3"
@@ -67,11 +67,11 @@ func Apply(ctx context.Context, client connection.RegistryClient, path, project
 }
 
 type patchGroup struct {
-	apiTasks        []core.Task
-	versionTasks    []core.Task
-	specTasks       []core.Task
-	deploymentTasks []core.Task
-	artifactTasks   []core.Task
+	apiTasks        []tasks.Task
+	versionTasks    []tasks.Task
+	specTasks       []tasks.Task
+	deploymentTasks []tasks.Task
+	artifactTasks   []tasks.Task
 }
 
 func (p *patchGroup) add(task *applyBytesTask) {
@@ -138,15 +138,15 @@ func (p *patchGroup) parse(client connection.RegistryClient, bytes []byte, fileN
 
 func (p *patchGroup) run(ctx context.Context, jobs int) error {
 	// Apply each resource type independently in order of ownership (parents first).
-	for _, tasks := range [][]core.Task{
+	for _, taskLists := range [][]tasks.Task{
 		p.apiTasks,
 		p.versionTasks,
 		p.specTasks,
 		p.deploymentTasks,
 		p.artifactTasks,
 	} {
-		taskQueue, wait := core.WorkerPool(ctx, jobs)
-		for _, task := range tasks {
+		taskQueue, wait := tasks.WorkerPool(ctx, jobs)
+		for _, task := range taskLists {
 			taskQueue <- task
 		}
 		wait()
diff --git a/cmd/registry/patch/project.go b/cmd/registry/patch/project.go
index 5f156da..80c56c3 100644
--- a/cmd/registry/patch/project.go
+++ b/cmd/registry/patch/project.go
@@ -23,7 +23,7 @@ import (
 	"strings"
 
 	"github.com/apigee/registry/cmd/registry/compress"
-	"github.com/apigee/registry/cmd/registry/core"
+	"github.com/apigee/registry/cmd/registry/tasks"
 	"github.com/apigee/registry/gapic"
 	"github.com/apigee/registry/log"
 	"github.com/apigee/registry/pkg/connection"
@@ -75,7 +75,7 @@ func applyProjectPatchBytes(ctx context.Context, client connection.AdminClient,
 }
 
 // ExportProject writes a project into a directory of YAML files.
-func ExportProject(ctx context.Context, client *gapic.RegistryClient, projectName names.Project, root string, taskQueue chan<- core.Task) error {
+func ExportProject(ctx context.Context, client *gapic.RegistryClient, projectName names.Project, root string, taskQueue chan<- tasks.Task) error {
 	root = filepath.Join(root, projectName.ProjectID)
 	if err := visitor.ListAPIs(ctx, client, projectName.Api(""), "", func(ctx context.Context, message *rpc.Api) error {
 		taskQueue <- &exportAPITask{
@@ -168,7 +168,7 @@ func ExportProject(ctx context.Context, client *gapic.RegistryClient, projectNam
 }
 
 // ExportAPI writes an API into a directory of YAML files.
-func ExportAPI(ctx context.Context, client *gapic.RegistryClient, apiName names.Api, recursive bool, root string, taskQueue chan<- core.Task) error {
+func ExportAPI(ctx context.Context, client *gapic.RegistryClient, apiName names.Api, recursive bool, root string, taskQueue chan<- tasks.Task) error {
 	root = filepath.Join(root, apiName.ProjectID)
 	if err := visitor.ListAPIs(ctx, client, apiName, "", func(ctx context.Context, message *rpc.Api) error {
 		taskQueue <- &exportAPITask{
@@ -254,7 +254,7 @@ func ExportAPI(ctx context.Context, client *gapic.RegistryClient, apiName names.
 }
 
 // ExportAPIVersion writes an API version into a directory of YAML files.
-func ExportAPIVersion(ctx context.Context, client *gapic.RegistryClient, versionName names.Version, recursive bool, root string, taskQueue chan<- core.Task) error {
+func ExportAPIVersion(ctx context.Context, client *gapic.RegistryClient, versionName names.Version, recursive bool, root string, taskQueue chan<- tasks.Task) error {
 	root = filepath.Join(root, versionName.ProjectID)
 	if err := visitor.ListVersions(ctx, client, versionName, "", func(ctx context.Context, message *rpc.ApiVersion) error {
 		taskQueue <- &exportVersionTask{
@@ -300,7 +300,7 @@ func ExportAPIVersion(ctx context.Context, client *gapic.RegistryClient, version
 }
 
 // ExportAPISpec writes an API spec into a directory of YAML files.
-func ExportAPISpec(ctx context.Context, client *gapic.RegistryClient, specName names.Spec, recursive bool, root string, taskQueue chan<- core.Task) error {
+func ExportAPISpec(ctx context.Context, client *gapic.RegistryClient, specName names.Spec, recursive bool, root string, taskQueue chan<- tasks.Task) error {
 	root = filepath.Join(root, specName.ProjectID)
 	if err := visitor.ListSpecs(ctx, client, specName, "", false, func(ctx context.Context, message *rpc.ApiSpec) error {
 		taskQueue <- &exportSpecTask{
@@ -326,7 +326,7 @@ func ExportAPISpec(ctx context.Context, client *gapic.RegistryClient, specName n
 }
 
 // ExportAPIDeployment writes an API deployment into a directory of YAML files.
-func ExportAPIDeployment(ctx context.Context, client *gapic.RegistryClient, deploymentName names.Deployment, recursive bool, root string, taskQueue chan<- core.Task) error {
+func ExportAPIDeployment(ctx context.Context, client *gapic.RegistryClient, deploymentName names.Deployment, recursive bool, root string, taskQueue chan<- tasks.Task) error {
 	root = filepath.Join(root, deploymentName.ProjectID)
 	if err := visitor.ListDeployments(ctx, client, deploymentName, "", func(ctx context.Context, message *rpc.ApiDeployment) error {
 		taskQueue <- &exportDeploymentTask{
@@ -352,7 +352,7 @@ func ExportAPIDeployment(ctx context.Context, client *gapic.RegistryClient, depl
 }
 
 // ExportArtifact writes an artifact into a directory of YAML files.
-func ExportArtifact(ctx context.Context, client *gapic.RegistryClient, artifactName names.Artifact, root string, taskQueue chan<- core.Task) error {
+func ExportArtifact(ctx context.Context, client *gapic.RegistryClient, artifactName names.Artifact, root string, taskQueue chan<- tasks.Task) error {
 	root = filepath.Join(root, artifactName.ProjectID())
 	return visitor.GetArtifact(ctx, client, artifactName, false, func(ctx context.Context, message *rpc.Artifact) error {
 		taskQueue <- &exportArtifactTask{
diff --git a/cmd/registry/core/tasks.go b/cmd/registry/tasks/tasks.go
similarity index 99%
rename from cmd/registry/core/tasks.go
rename to cmd/registry/tasks/tasks.go
index 33c97a3..c2faf7d 100644
--- a/cmd/registry/core/tasks.go
+++ b/cmd/registry/tasks/tasks.go
@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-package core
+package tasks
 
 import (
 	"context"
diff --git a/pkg/wipeout/wipeout.go b/pkg/wipeout/wipeout.go
index 55ca6bc..c818053 100644
--- a/pkg/wipeout/wipeout.go
+++ b/pkg/wipeout/wipeout.go
@@ -17,7 +17,7 @@ package wipeout
 import (
 	"context"
 
-	"github.com/apigee/registry/cmd/registry/core"
+	"github.com/apigee/registry/cmd/registry/tasks"
 	"github.com/apigee/registry/log"
 	"github.com/apigee/registry/pkg/connection"
 	"github.com/apigee/registry/rpc"
@@ -30,20 +30,20 @@ func Wipeout(ctx context.Context, client connection.RegistryClient, projectID st
 	project := "projects/" + projectID + "/locations/global"
 	{
 		log.Debugf(ctx, "Deleting apis")
-		taskQueue, wait := core.WorkerPool(ctx, jobs)
+		taskQueue, wait := tasks.WorkerPool(ctx, jobs)
 		wipeoutApis(ctx, client, taskQueue, project)
 		wait()
 	}
 	{
 		log.Debugf(ctx, "Deleting artifacts")
-		taskQueue, wait := core.WorkerPool(ctx, jobs)
+		taskQueue, wait := tasks.WorkerPool(ctx, jobs)
 		wipeoutArtifacts(ctx, client, taskQueue, project)
 		wait()
 	}
 	log.Debugf(ctx, "Wipeout complete")
 }
 
-func wipeoutArtifacts(ctx context.Context, client connection.RegistryClient, taskQueue chan<- core.Task, parent string) {
+func wipeoutArtifacts(ctx context.Context, client connection.RegistryClient, taskQueue chan<- tasks.Task, parent string) {
 	it := client.ListArtifacts(ctx, &rpc.ListArtifactsRequest{Parent: parent})
 	names := make([]string, 0)
 	for artifact, err := it.Next(); err == nil; artifact, err = it.Next() {
@@ -54,7 +54,7 @@ func wipeoutArtifacts(ctx context.Context, client connection.RegistryClient, tas
 	}
 }
 
-func wipeoutApis(ctx context.Context, client connection.RegistryClient, taskQueue chan<- core.Task, parent string) {
+func wipeoutApis(ctx context.Context, client connection.RegistryClient, taskQueue chan<- tasks.Task, parent string) {
 	it := client.ListApis(ctx, &rpc.ListApisRequest{Parent: parent})
 	names := make([]string, 0)
 	for api, err := it.Next(); err == nil; api, err = it.Next() {


Test patch:
diff --git a/cmd/registry/patch/export_test.go b/cmd/registry/patch/export_test.go
index 4eb7f60..0288c09 100644
--- a/cmd/registry/patch/export_test.go
+++ b/cmd/registry/patch/export_test.go
@@ -22,7 +22,7 @@ import (
 	"strings"
 	"testing"
 
-	"github.com/apigee/registry/cmd/registry/core"
+	"github.com/apigee/registry/cmd/registry/tasks"
 	"github.com/apigee/registry/pkg/connection"
 	"github.com/apigee/registry/pkg/names"
 	"github.com/apigee/registry/rpc"
@@ -90,7 +90,7 @@ func TestExport(t *testing.T) {
 
 		t.Run(test.desc+"-project", func(t *testing.T) {
 			tempDir := t.TempDir()
-			taskQueue, wait := core.WorkerPool(ctx, 1)
+			taskQueue, wait := tasks.WorkerPool(ctx, 1)
 			err = ExportProject(ctx, registryClient, project, tempDir, taskQueue)
 			if err != nil {
 				t.Errorf("Failed to export project: %s", err)
@@ -101,7 +101,7 @@ func TestExport(t *testing.T) {
 
 		t.Run(test.desc+"-api", func(t *testing.T) {
 			tempDir := t.TempDir()
-			taskQueue, wait := core.WorkerPool(ctx, 1)
+			taskQueue, wait := tasks.WorkerPool(ctx, 1)
 			err = ExportAPI(ctx, registryClient, project.Api("registry"), false, tempDir, taskQueue)
 			if err != nil {
 				t.Errorf("Failed to export api: %s", err)
@@ -112,7 +112,7 @@ func TestExport(t *testing.T) {
 
 		t.Run(test.desc+"-api-recursive", func(t *testing.T) {
 			tempDir := t.TempDir()
-			taskQueue, wait := core.WorkerPool(ctx, 1)
+			taskQueue, wait := tasks.WorkerPool(ctx, 1)
 			err = ExportAPI(ctx, registryClient, project.Api("registry"), true, tempDir, taskQueue)
 			if err != nil {
 				t.Errorf("Failed to export api: %s", err)
@@ -123,7 +123,7 @@ func TestExport(t *testing.T) {
 
 		t.Run(test.desc+"-version", func(t *testing.T) {
 			tempDir := t.TempDir()
-			taskQueue, wait := core.WorkerPool(ctx, 1)
+			taskQueue, wait := tasks.WorkerPool(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)
@@ -134,7 +134,7 @@ func TestExport(t *testing.T) {
 
 		t.Run(test.desc+"-version-recursive", func(t *testing.T) {
 			tempDir := t.TempDir()
-			taskQueue, wait := core.WorkerPool(ctx, 1)
+			taskQueue, wait := tasks.WorkerPool(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)
@@ -145,7 +145,7 @@ func TestExport(t *testing.T) {
 
 		t.Run(test.desc+"-spec", func(t *testing.T) {
 			tempDir := t.TempDir()
-			taskQueue, wait := core.WorkerPool(ctx, 1)
+			taskQueue, wait := tasks.WorkerPool(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)
@@ -157,7 +157,7 @@ func TestExport(t *testing.T) {
 
 		t.Run(test.desc+"-spec-recursive", func(t *testing.T) {
 			tempDir := t.TempDir()
-			taskQueue, wait := core.WorkerPool(ctx, 1)
+			taskQueue, wait := tasks.WorkerPool(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)
@@ -168,7 +168,7 @@ func TestExport(t *testing.T) {
 
 		t.Run(test.desc+"-deployment", func(t *testing.T) {
 			tempDir := t.TempDir()
-			taskQueue, wait := core.WorkerPool(ctx, 1)
+			taskQueue, wait := tasks.WorkerPool(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)
@@ -179,7 +179,7 @@ func TestExport(t *testing.T) {
 
 		t.Run(test.desc+"-deployment-recursive", func(t *testing.T) {
 			tempDir := t.TempDir()
-			taskQueue, wait := core.WorkerPool(ctx, 1)
+			taskQueue, wait := tasks.WorkerPool(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)
@@ -190,7 +190,7 @@ func TestExport(t *testing.T) {
 
 		t.Run(test.desc+"-artifact", func(t *testing.T) {
 			tempDir := t.TempDir()
-			taskQueue, wait := core.WorkerPool(ctx, 1)
+			taskQueue, wait := tasks.WorkerPool(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
new file mode 100644
index 0000000..d032593
--- /dev/null
+++ b/cmd/registry/tasks/tasks_test.go
@@ -0,0 +1,70 @@
+// 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.
+// You may obtain a copy of the License at
+//
+//    http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package tasks
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"testing"
+)
+
+func TestWorkerPool(t *testing.T) {
+	ctx := context.Background()
+	jobs := 100
+
+	taskQueue, wait := WorkerPool(ctx, jobs)
+	defer wait()
+
+	for i := 0; i < 1000; i++ {
+		taskQueue <- &doNothingTask{i: i}
+	}
+}
+
+type doNothingTask struct {
+	i int
+}
+
+func (task *doNothingTask) String() string {
+	return fmt.Sprintf("do nothing %d", task.i)
+}
+
+func (task *doNothingTask) Run(ctx context.Context) error {
+	return nil
+}
+
+func TestWorkerPoolWithWarnings(t *testing.T) {
+	ctx := context.Background()
+	jobs := 1
+
+	taskQueue, wait := WorkerPoolWithWarnings(ctx, jobs)
+	defer wait()
+
+	for i := 0; i < 10; i++ {
+		taskQueue <- &failTask{i: i}
+	}
+}
+
+type failTask struct {
+	i int
+}
+
+func (task *failTask) String() string {
+	return fmt.Sprintf("do nothing %d", task.i)
+}
+
+func (task *failTask) Run(ctx context.Context) error {
+	return errors.New("fail")
+}