feat(worker): add/set/delete repo worker tags (#2236)

implements repo changes for add/set/delete worker tag operations

internal/proto/controller/storage/servers/store/v1/worker.proto

@@ -72,6 +72,6 @@ message WorkerTag {
   string value = 30;
 
   // source is the source of the tag. Either 'configuration' or 'api'.
-  // @inject_tag: `gorm:"default:not_null"`
+  // @inject_tag: `gorm:"primary_key"`
   string source = 40;
 }

internal/server/repository_worker.go

@@ -364,7 +364,7 @@ func setWorkerTags(ctx context.Context, w db.Writer, id string, ts TagSource, ta
 	}
 	_, err := w.Exec(ctx, deleteTagsByWorkerIdSql, []interface{}{ts.String(), id})
 	if err != nil {
-		return errors.Wrap(ctx, err, op, errors.WithMsg(fmt.Sprintf("couldn't delete exist tags for worker %q", id)))
+		return errors.Wrap(ctx, err, op, errors.WithMsg(fmt.Sprintf("couldn't delete existing tags for worker %q", id)))
 	}
 
 	// If tags were cleared out entirely, then we'll have nothing
@@ -558,3 +558,156 @@ func (r *Repository) CreateWorker(ctx context.Context, worker *Worker, opt ...Op
 	}
 	return returnedWorker, nil
 }
+
+// AddWorkerTags adds specified api tags to the repo worker and returns its new tags.
+// No options are currently supported.
+func (r *Repository) AddWorkerTags(ctx context.Context, workerId string, workerVersion uint32, tags []*Tag, _ ...Option) ([]*Tag, error) {
+	const op = "server.(Repository).AddWorkerTags"
+	switch {
+	case workerId == "":
+		return nil, errors.New(ctx, errors.InvalidParameter, op, "worker public id is empty")
+	case workerVersion == 0:
+		return nil, errors.New(ctx, errors.InvalidParameter, op, "missing version")
+	case len(tags) == 0:
+		return nil, errors.New(ctx, errors.InvalidParameter, op, "no tags provided")
+	}
+
+	worker, err := lookupWorker(ctx, r.reader, workerId)
+	if err != nil {
+		return nil, errors.Wrap(ctx, err, op)
+	}
+	if worker == nil {
+		return nil, errors.New(ctx, errors.InvalidParameter, op, fmt.Sprintf("no worker found with public id %s", workerId))
+	}
+
+	newTags := worker.apiTags
+	for _, t := range tags {
+		newTags = append(newTags, t)
+	}
+	_, err = r.writer.DoTx(ctx, db.StdRetryCnt, db.ExpBackoff{}, func(reader db.Reader, w db.Writer) error {
+		worker := worker.clone()
+		worker.PublicId = workerId
+		worker.Version = workerVersion + 1
+		rowsUpdated, err := w.Update(ctx, worker, []string{"Version"}, nil, db.WithVersion(&workerVersion))
+		if err != nil {
+			return errors.Wrap(ctx, err, op, errors.WithMsg("unable to update worker version"))
+		}
+		if rowsUpdated != 1 {
+			return errors.New(ctx, errors.MultipleRecords, op, fmt.Sprintf("updated worker version and %d rows updated", rowsUpdated))
+		}
+		err = setWorkerTags(ctx, w, workerId, ApiTagSource, newTags)
+		if err != nil {
+			return errors.Wrap(ctx, err, op)
+		}
+		return nil
+	})
+	if err != nil {
+		return nil, errors.Wrap(ctx, err, op)
+	}
+	return newTags, nil
+}
+
+// SetWorkerTags clears the current repo worker's api tags and sets them from the input parameters.
+// Returns the current repo worker tags. No options are currently supported.
+func (r *Repository) SetWorkerTags(ctx context.Context, workerId string, workerVersion uint32, tags []*Tag, _ ...Option) ([]*Tag, error) {
+	const op = "server.(Repository).SetWorkerTags"
+	switch {
+	case workerId == "":
+		return nil, errors.New(ctx, errors.InvalidParameter, op, "worker public id is empty")
+	case workerVersion == 0:
+		return nil, errors.New(ctx, errors.InvalidParameter, op, "missing version")
+	}
+
+	worker, err := lookupWorker(ctx, r.reader, workerId)
+	if err != nil {
+		return nil, errors.Wrap(ctx, err, op)
+	}
+	if worker == nil {
+		return nil, errors.New(ctx, errors.InvalidParameter, op, fmt.Sprintf("no worker found with public id %s", workerId))
+	}
+
+	_, err = r.writer.DoTx(ctx, db.StdRetryCnt, db.ExpBackoff{}, func(reader db.Reader, w db.Writer) error {
+		worker := worker.clone()
+		worker.PublicId = workerId
+		worker.Version = workerVersion + 1
+		rowsUpdated, err := w.Update(ctx, worker, []string{"Version"}, nil, db.WithVersion(&workerVersion))
+		if err != nil {
+			return errors.Wrap(ctx, err, op, errors.WithMsg("unable to update worker version"))
+		}
+		if rowsUpdated != 1 {
+			return errors.New(ctx, errors.MultipleRecords, op, fmt.Sprintf("updated worker version and %d rows updated", rowsUpdated))
+		}
+		err = setWorkerTags(ctx, w, workerId, ApiTagSource, tags)
+		if err != nil {
+			return errors.Wrap(ctx, err, op)
+		}
+		return nil
+	})
+	if err != nil {
+		return nil, errors.Wrap(ctx, err, op)
+	}
+	return tags, nil
+}
+
+// DeleteWorkerTags deletes specified api worker tags from the repo. Returns the number of rows deleted.
+// No options are currently supported.
+func (r *Repository) DeleteWorkerTags(ctx context.Context, workerId string, workerVersion uint32, tags []*Tag, _ ...Option) (int, error) {
+	const op = "server.(Repository).DeleteWorkerTags"
+	switch {
+	case workerId == "":
+		return db.NoRowsAffected, errors.New(ctx, errors.InvalidParameter, op, "worker public id is empty")
+	case workerVersion == 0:
+		return db.NoRowsAffected, errors.New(ctx, errors.InvalidParameter, op, "missing version")
+	case len(tags) == 0:
+		return db.NoRowsAffected, errors.New(ctx, errors.InvalidParameter, op, "no tags provided")
+	}
+
+	worker, err := lookupWorker(ctx, r.reader, workerId)
+	if err != nil {
+		return db.NoRowsAffected, errors.Wrap(ctx, err, op)
+	}
+	if worker == nil {
+		return db.NoRowsAffected, errors.New(ctx, errors.InvalidParameter, op, fmt.Sprintf("no worker found with public id %s", workerId))
+	}
+
+	rowsDeleted := 0
+	deleteTags := make([]interface{}, 0, len(tags))
+	for _, t := range tags {
+		if t == nil {
+			return db.NoRowsAffected, errors.New(ctx, errors.InvalidParameter, op, "found nil tag value in input")
+		}
+		deleteTags = append(deleteTags, &store.WorkerTag{
+			WorkerId: workerId,
+			Key:      t.Key,
+			Value:    t.Value,
+			Source:   ApiTagSource.String(),
+		})
+	}
+
+	_, err = r.writer.DoTx(ctx, db.StdRetryCnt, db.ExpBackoff{}, func(reader db.Reader, w db.Writer) error {
+		worker := worker.clone()
+		worker.PublicId = workerId
+		worker.Version = workerVersion + 1
+		rowsUpdated, err := w.Update(ctx, worker, []string{"Version"}, nil, db.WithVersion(&workerVersion))
+		if err != nil {
+			return errors.Wrap(ctx, err, op, errors.WithMsg("unable to update worker version"))
+		}
+		if rowsUpdated != 1 {
+			return errors.New(ctx, errors.MultipleRecords, op, fmt.Sprintf("updated worker version and %d rows updated", rowsUpdated))
+		}
+
+		rowsDeleted, err = w.DeleteItems(ctx, deleteTags)
+		if err != nil {
+			return errors.Wrap(ctx, err, op, errors.WithMsg("unable to delete worker tags"))
+		}
+		if rowsDeleted != len(deleteTags) {
+			return errors.New(ctx, errors.MultipleRecords, op, fmt.Sprintf("tags deleted %d did not match request for %d", rowsDeleted, len(tags)))
+		}
+		return nil
+	})
+
+	if err != nil {
+		return db.NoRowsAffected, errors.Wrap(ctx, err, op)
+	}
+	return rowsDeleted, nil
+}

internal/server/store/worker.pb.go

@@ -179,8 +179,8 @@ type WorkerTag struct {
 	// @inject_tag: `gorm:"primary_key"`
 	Value string `protobuf:"bytes,30,opt,name=value,proto3" json:"value,omitempty" gorm:"primary_key"`
 	// source is the source of the tag. Either 'configuration' or 'api'.
-	// @inject_tag: `gorm:"default:not_null"`
-	Source string `protobuf:"bytes,40,opt,name=source,proto3" json:"source,omitempty" gorm:"default:not_null"`
+	// @inject_tag: `gorm:"primary_key"`
+	Source string `protobuf:"bytes,40,opt,name=source,proto3" json:"source,omitempty" gorm:"primary_key"`
 }
 
 func (x *WorkerTag) Reset() {

internal/server/worker_tags_test.go

@@ -5,8 +5,11 @@ import (
 	"testing"
 
 	"github.com/hashicorp/boundary/internal/db"
+	"github.com/hashicorp/boundary/internal/errors"
+	"github.com/hashicorp/boundary/internal/kms"
 	"github.com/hashicorp/boundary/internal/server/store"
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 )
 
 func TestWorkerTags_Create(t *testing.T) {
@@ -106,3 +109,603 @@ func TestWorkerTags_Create(t *testing.T) {
 		})
 	}
 }
+
+func TestRepository_AddWorkerTags(t *testing.T) {
+	t.Parallel()
+	conn, _ := db.TestSetup(t, "postgres")
+	rw := db.New(conn)
+	wrapper := db.TestWrapper(t)
+	kms := kms.TestKms(t, conn, wrapper)
+
+	assert, require := assert.New(t), require.New(t)
+	repo, err := NewRepository(rw, rw, kms)
+	require.NoError(err)
+	require.NotNil(repo)
+	// WithWorkerTags sets config tags to ensure they are not affected by api tag operations
+	worker := TestKmsWorker(t, conn, wrapper, WithWorkerTags(&Tag{Key: "key_c", Value: "value_c"}))
+
+	type args struct {
+		publicId string
+		version  uint32
+		tags     []*Tag
+	}
+
+	tests := []struct {
+		name            string
+		args            args
+		want            []*Tag
+		wantIsErr       errors.Code
+		wantErrContains string
+	}{
+		{
+			name: "empty-public-id",
+			args: args{
+				publicId: "",
+				version:  worker.Version,
+				tags: []*Tag{{
+					Key:   "key",
+					Value: "value",
+				}},
+			},
+			wantIsErr:       errors.InvalidParameter,
+			wantErrContains: "worker public id is empty",
+		},
+		{
+			name: "zero-version",
+			args: args{
+				publicId: worker.PublicId,
+				version:  0,
+				tags: []*Tag{{
+					Key:   "key",
+					Value: "value",
+				}},
+			},
+			wantIsErr:       errors.InvalidParameter,
+			wantErrContains: "missing version",
+		},
+		{
+			name: "bad-version",
+			args: args{
+				publicId: worker.PublicId,
+				version:  100,
+				tags: []*Tag{{
+					Key:   "key",
+					Value: "value",
+				}},
+			},
+			wantIsErr:       errors.MultipleRecords,
+			wantErrContains: "updated worker version and 0 rows updated",
+		},
+		{
+			name: "nil-tags",
+			args: args{
+				publicId: worker.PublicId,
+				version:  worker.Version,
+				tags:     nil,
+			},
+			wantIsErr:       errors.InvalidParameter,
+			wantErrContains: "no tags provided",
+		},
+		{
+			name: "add-valid-tag",
+			args: args{
+				publicId: worker.PublicId,
+				version:  worker.Version,
+				tags: []*Tag{{
+					Key:   "key",
+					Value: "value",
+				}},
+			},
+			want: []*Tag{{
+				Key:   "key",
+				Value: "value",
+			}},
+		},
+		{
+			name: "add-many-tags",
+			args: func() args {
+				// reset test worker to avoid worker version conflicts when running tests individually vs sequentially
+				worker := TestKmsWorker(t, conn, wrapper, WithWorkerTags(&Tag{Key: "key_c", Value: "value_c"}))
+				return args{
+					publicId: worker.PublicId,
+					version:  worker.Version,
+					tags: []*Tag{
+						{
+							Key:   "key",
+							Value: "value",
+						},
+						{
+							Key:   "key2",
+							Value: "value2",
+						},
+						{
+							Key:   "key3",
+							Value: "value3",
+						},
+					},
+				}
+			}(),
+			want: []*Tag{
+				{
+					Key:   "key",
+					Value: "value",
+				},
+				{
+					Key:   "key2",
+					Value: "value2",
+				},
+				{
+					Key:   "key3",
+					Value: "value3",
+				},
+			},
+		},
+		{
+			name: "add-preexisting-config-tags",
+			args: func() args {
+				worker := TestKmsWorker(t, conn, wrapper, WithWorkerTags(&Tag{Key: "key_c", Value: "value_c"}))
+				return args{
+					publicId: worker.PublicId,
+					version:  worker.Version,
+					tags: []*Tag{{
+						Key:   "key_c",
+						Value: "value_c",
+					}},
+				}
+			}(),
+			want: []*Tag{{
+				Key:   "key_c",
+				Value: "value_c",
+			}},
+		},
+	}
+
+	for _, tt := range tests {
+		tt := tt
+		t.Run(tt.name, func(t *testing.T) {
+			got, err := repo.AddWorkerTags(context.Background(), tt.args.publicId, tt.args.version, tt.args.tags)
+			if tt.wantErrContains != "" {
+				assert.Truef(errors.Match(errors.T(tt.wantIsErr), err), "want err: %q got: %q", tt.wantIsErr, err)
+				assert.Contains(err.Error(), tt.wantErrContains)
+				if tt.args.publicId != "" {
+					repoWorker, err := repo.LookupWorker(context.Background(), tt.args.publicId)
+					require.NoError(err)
+					assert.Equal(uint32(1), repoWorker.Version)
+				}
+				return
+			}
+			assert.NoError(err)
+			assert.Equal(tt.want, got)
+			repoWorker, err := repo.LookupWorker(context.Background(), tt.args.publicId)
+			require.NoError(err)
+			assert.Equal(tt.args.version+1, repoWorker.Version)
+		})
+	}
+}
+
+func TestRepository_SetWorkerTags(t *testing.T) {
+	t.Parallel()
+	conn, _ := db.TestSetup(t, "postgres")
+	rw := db.New(conn)
+	wrapper := db.TestWrapper(t)
+	kms := kms.TestKms(t, conn, wrapper)
+	worker := TestKmsWorker(t, conn, wrapper, WithWorkerTags(&Tag{Key: "key_c", Value: "value_c"}))
+
+	assert, require := assert.New(t), require.New(t)
+	repo, err := NewRepository(rw, rw, kms)
+	require.NoError(err)
+	require.NotNil(repo)
+
+	type args struct {
+		publicId string
+		version  uint32
+		tags     []*Tag
+	}
+
+	tests := []struct {
+		name            string
+		args            args
+		want            []*Tag
+		wantIsErr       errors.Code
+		wantErrContains string
+	}{
+		{
+			name: "empty-public-id",
+			args: args{
+				publicId: "",
+				version:  worker.Version,
+				tags: []*Tag{{
+					Key:   "key",
+					Value: "value",
+				}},
+			},
+			wantIsErr:       errors.InvalidParameter,
+			wantErrContains: "worker public id is empty",
+		},
+		{
+			name: "zero-version",
+			args: args{
+				publicId: worker.PublicId,
+				version:  0,
+				tags: []*Tag{{
+					Key:   "key",
+					Value: "value",
+				}},
+			},
+			wantIsErr:       errors.InvalidParameter,
+			wantErrContains: "missing version",
+		},
+		{
+			name: "bad-version",
+			args: args{
+				publicId: worker.PublicId,
+				version:  100,
+				tags: []*Tag{{
+					Key:   "key",
+					Value: "value",
+				}},
+			},
+			wantIsErr:       errors.MultipleRecords,
+			wantErrContains: "updated worker version and 0 rows updated",
+		},
+		{
+			name: "set-nil-tags",
+			args: args{
+				publicId: worker.PublicId,
+				version:  worker.Version,
+				tags:     nil,
+			},
+			want: nil,
+		},
+		{
+			name: "set-many-tags",
+			args: func() args {
+				worker := TestKmsWorker(t, conn, wrapper, WithWorkerTags(&Tag{Key: "key_c", Value: "value_c"}))
+				return args{
+					publicId: worker.PublicId,
+					version:  worker.Version,
+					tags: []*Tag{
+						{
+							Key:   "key",
+							Value: "value",
+						},
+						{
+							Key:   "key2",
+							Value: "value2",
+						},
+						{
+							Key:   "key3",
+							Value: "value3",
+						},
+					},
+				}
+			}(),
+			want: []*Tag{
+				{
+					Key:   "key",
+					Value: "value",
+				},
+				{
+					Key:   "key2",
+					Value: "value2",
+				},
+				{
+					Key:   "key3",
+					Value: "value3",
+				},
+			},
+		},
+		{
+			name: "set-preexisting-config-tags",
+			args: func() args {
+				worker := TestKmsWorker(t, conn, wrapper, WithWorkerTags(&Tag{Key: "key_c", Value: "value_c"}))
+				return args{
+					publicId: worker.PublicId,
+					version:  worker.Version,
+					tags: []*Tag{{
+						Key:   "key_c",
+						Value: "value_c",
+					}},
+				}
+			}(),
+			want: []*Tag{{
+				Key:   "key_c",
+				Value: "value_c",
+			}},
+		},
+	}
+
+	for _, tt := range tests {
+		tt := tt
+		t.Run(tt.name, func(t *testing.T) {
+			got, err := repo.SetWorkerTags(context.Background(), tt.args.publicId, tt.args.version, tt.args.tags)
+			if tt.wantErrContains != "" {
+				assert.Truef(errors.Match(errors.T(tt.wantIsErr), err), "want err: %q got: %q", tt.wantIsErr, err)
+				assert.Contains(err.Error(), tt.wantErrContains)
+				if tt.args.publicId != "" {
+					repoWorker, err := repo.LookupWorker(context.Background(), tt.args.publicId)
+					require.NoError(err)
+					assert.Equal(uint32(1), repoWorker.Version)
+				}
+				return
+			}
+			assert.NoError(err)
+			assert.Equal(tt.want, got)
+			repoWorker, err := repo.LookupWorker(context.Background(), tt.args.publicId)
+			require.NoError(err)
+			assert.Equal(tt.args.version+1, repoWorker.Version)
+		})
+	}
+}
+
+func TestRepository_DeleteWorkerTags(t *testing.T) {
+	// Note: more delete operation testcases are found in subsequent func TestRepository_WorkerTagsConsequent
+	t.Parallel()
+	conn, _ := db.TestSetup(t, "postgres")
+	rw := db.New(conn)
+	wrapper := db.TestWrapper(t)
+	kms := kms.TestKms(t, conn, wrapper)
+	worker := TestKmsWorker(t, conn, wrapper)
+
+	assert, require := assert.New(t), require.New(t)
+	repo, err := NewRepository(rw, rw, kms)
+	require.NoError(err)
+	require.NotNil(repo)
+
+	type args struct {
+		publicId string
+		version  uint32
+		tags     []*Tag
+	}
+
+	tests := []struct {
+		name            string
+		args            args
+		want            int
+		wantIsErr       errors.Code
+		wantErrContains string
+	}{
+		{
+			name: "empty-public-id",
+			args: args{
+				publicId: "",
+				version:  worker.Version,
+				tags: []*Tag{{
+					Key:   "key",
+					Value: "value",
+				}},
+			},
+			wantIsErr:       errors.InvalidParameter,
+			wantErrContains: "worker public id is empty",
+		},
+		{
+			name: "zero-version",
+			args: args{
+				publicId: worker.PublicId,
+				version:  0,
+				tags: []*Tag{{
+					Key:   "key",
+					Value: "value",
+				}},
+			},
+			wantIsErr:       errors.InvalidParameter,
+			wantErrContains: "missing version",
+		},
+		{
+			name: "bad-version",
+			args: args{
+				publicId: worker.PublicId,
+				version:  100,
+				tags: []*Tag{{
+					Key:   "key",
+					Value: "value",
+				}},
+			},
+			wantIsErr:       errors.MultipleRecords,
+			wantErrContains: "updated worker version and 0 rows updated",
+		},
+		{
+			name: "nil-tags",
+			args: args{
+				publicId: worker.PublicId,
+				version:  worker.Version,
+				tags:     nil,
+			},
+			wantIsErr:       errors.InvalidParameter,
+			wantErrContains: "no tags provided",
+		},
+		{
+			name: "one-nil-tag",
+			args: args{
+				publicId: worker.PublicId,
+				version:  worker.Version,
+				tags: []*Tag{
+					{
+						Key:   "key",
+						Value: "value",
+					},
+					nil,
+				},
+			},
+			wantIsErr:       errors.InvalidParameter,
+			wantErrContains: "found nil tag value in input",
+		},
+		{
+			name: "nonexistent-tag",
+			args: args{
+				publicId: worker.PublicId,
+				version:  worker.Version,
+				tags: []*Tag{
+					{
+						Key:   "bad_key",
+						Value: "bad_value",
+					},
+				},
+			},
+			wantIsErr:       errors.MultipleRecords,
+			wantErrContains: "tags deleted 0 did not match request for 1",
+		},
+		{
+			name: "valid-delete",
+			args: func() args {
+				worker := TestKmsWorker(t, conn, wrapper)
+				_, err = repo.AddWorkerTags(context.Background(), worker.PublicId, worker.Version, []*Tag{
+					{Key: "key", Value: "value"},
+				})
+				require.NoError(err)
+				return args{
+					publicId: worker.PublicId,
+					version:  worker.Version + 1,
+					tags: []*Tag{
+						{
+							Key:   "key",
+							Value: "value",
+						},
+					},
+				}
+			}(),
+			want: 1,
+		},
+	}
+
+	for _, tt := range tests {
+		tt := tt
+		t.Run(tt.name, func(t *testing.T) {
+			got, err := repo.DeleteWorkerTags(context.Background(), tt.args.publicId, tt.args.version, tt.args.tags)
+			if tt.wantErrContains != "" {
+				assert.Truef(errors.Match(errors.T(tt.wantIsErr), err), "want err: %q got: %q", tt.wantIsErr, err)
+				assert.Contains(err.Error(), tt.wantErrContains)
+				if tt.args.publicId != "" {
+					repoWorker, err := repo.LookupWorker(context.Background(), tt.args.publicId)
+					require.NoError(err)
+					assert.Equal(uint32(1), repoWorker.Version)
+				}
+				return
+			}
+			assert.NoError(err)
+			assert.Equal(tt.want, got)
+			repoWorker, err := repo.LookupWorker(context.Background(), tt.args.publicId)
+			require.NoError(err)
+			assert.Equal(tt.args.version+1, repoWorker.Version)
+		})
+	}
+}
+
+func TestRepository_WorkerTagsConsequent(t *testing.T) {
+	t.Parallel()
+	assert, require := assert.New(t), require.New(t)
+	conn, _ := db.TestSetup(t, "postgres")
+	rw := db.New(conn)
+	wrapper := db.TestWrapper(t)
+	kms := kms.TestKms(t, conn, wrapper)
+	repo, err := NewRepository(rw, rw, kms)
+	require.NoError(err)
+	require.NotNil(repo)
+
+	// Create worker and set various overlapping config tags
+	worker := TestKmsWorker(t, conn, wrapper, WithWorkerTags(
+		&Tag{Key: "key", Value: "value"},
+		&Tag{Key: "keykey", Value: "valval"},
+		&Tag{Key: "key3", Value: "value3"},
+		&Tag{Key: "key?", Value: "value?"}))
+
+	// Add three valid tags to worker
+	manyTags := []*Tag{
+		{Key: "key", Value: "value"},
+		{Key: "key2", Value: "value2"},
+		{Key: "key3", Value: "value3"},
+	}
+	added, err := repo.AddWorkerTags(context.Background(), worker.PublicId, worker.Version, manyTags)
+	assert.NoError(err)
+	assert.Equal(manyTags, added)
+	worker, err = repo.LookupWorker(context.Background(), worker.PublicId)
+	require.NoError(err)
+	assert.Equal(uint32(2), worker.Version)
+
+	// Test adding a duplicate tag
+	added, err = repo.AddWorkerTags(context.Background(), worker.PublicId, worker.Version, []*Tag{
+		{Key: "key", Value: "value"},
+	})
+	assert.Error(err)
+	assert.Contains(err.Error(), "duplicate key value violates unique constraint")
+	assert.Nil(added)
+	worker, err = repo.LookupWorker(context.Background(), worker.PublicId)
+	require.NoError(err)
+	assert.Equal(uint32(2), worker.Version)
+
+	// Test adding/setting/deleting an invalid batch of tags
+	invalidTags := []*Tag{
+		{Key: "keya", Value: "valuea"},
+		{Key: "keyb", Value: "valueb"},
+		{Key: "keykey", Value: "valval"},
+		{Key: "keykey", Value: "valval"},
+	}
+	added, err = repo.AddWorkerTags(context.Background(), worker.PublicId, worker.Version, invalidTags)
+	assert.Contains(err.Error(), "duplicate key value violates unique constraint")
+	assert.Nil(added)
+	set, err := repo.SetWorkerTags(context.Background(), worker.PublicId, worker.Version, invalidTags)
+	assert.Contains(err.Error(), "duplicate key value violates unique constraint")
+	assert.Nil(set)
+	rowsDeleted, err := repo.DeleteWorkerTags(context.Background(), worker.PublicId, worker.Version, invalidTags)
+	assert.Contains(err.Error(), "tags deleted 0 did not match request for 4")
+	assert.Equal(0, rowsDeleted)
+	worker, err = repo.LookupWorker(context.Background(), worker.PublicId)
+	require.NoError(err)
+	assert.Equal(uint32(2), worker.Version)
+	assert.Equal(len(manyTags), len(worker.apiTags))
+
+	// Delete a valid tag from worker
+	rowsDeleted, err = repo.DeleteWorkerTags(context.Background(), worker.PublicId, worker.Version, []*Tag{
+		{Key: "key3", Value: "value3"},
+	})
+	assert.Equal(1, rowsDeleted)
+	assert.NoError(err)
+	worker, err = repo.LookupWorker(context.Background(), worker.PublicId)
+	require.NoError(err)
+	assert.Equal(uint32(3), worker.Version)
+	assert.Contains(worker.apiTags, &Tag{Key: "key", Value: "value"})
+	assert.Contains(worker.apiTags, &Tag{Key: "key2", Value: "value2"})
+
+	// Add another valid tag to worker
+	added, err = repo.AddWorkerTags(context.Background(), worker.PublicId, worker.Version, []*Tag{
+		{Key: "key!", Value: "value!"},
+	})
+	assert.NoError(err)
+	worker, err = repo.LookupWorker(context.Background(), worker.PublicId)
+	require.NoError(err)
+	assert.Equal(uint32(4), worker.Version)
+	assert.Contains(worker.apiTags, &Tag{Key: "key!", Value: "value!"})
+	assert.Equal(3, len(worker.apiTags))
+
+	// Set all tags to nil
+	set, err = repo.SetWorkerTags(context.Background(), worker.PublicId, worker.Version, nil)
+	assert.Equal([]*Tag(nil), set)
+	assert.NoError(err)
+	worker, err = repo.LookupWorker(context.Background(), worker.PublicId)
+	require.NoError(err)
+	assert.Equal(uint32(5), worker.Version)
+	assert.Equal(set, worker.apiTags)
+	assert.Equal(0, len(worker.apiTags))
+
+	// Ensure config tags are untouched
+	for _, ct := range []*Tag{
+		{Key: "key", Value: "value"},
+		{Key: "keykey", Value: "valval"},
+		{Key: "key3", Value: "value3"},
+		{Key: "key?", Value: "value?"},
+	} {
+		assert.Contains(worker.configTags, ct)
+	}
+	assert.Equal(4, len(worker.configTags))
+
+	// Go full circle
+	added, err = repo.AddWorkerTags(context.Background(), worker.PublicId, worker.Version, manyTags)
+	assert.NoError(err)
+	worker, err = repo.LookupWorker(context.Background(), worker.PublicId)
+	require.NoError(err)
+	assert.Equal(uint32(6), worker.Version)
+	assert.Equal(len(manyTags), len(worker.apiTags))
+	for _, t := range manyTags {
+		assert.Contains(worker.apiTags, t)
+	}
+}