Call OnCreate or OnUpdate on the plugin after db constraints are checked (#1683)

* Only call the Plugin at create/update time after the db has been partially updated. This allows more constraints to be checked prior to calling the plugin.

internal/host/plugin/repository_host_catalog.go

@@ -76,33 +76,17 @@ func (r *Repository) CreateCatalog(ctx context.Context, c *HostCatalog, _ ...Opt
 	if !ok || plgClient == nil {
 		return nil, nil, errors.New(ctx, errors.InvalidParameter, op, fmt.Sprintf("plugin %q not available", c.GetPluginId()))
 	}
-	plgResp, err := plgClient.OnCreateCatalog(ctx, &plgpb.OnCreateCatalogRequest{Catalog: plgHc})
-	if err != nil {
-		if status.Code(err) != codes.Unimplemented {
-			return nil, nil, errors.Wrap(ctx, err, op)
-		}
-	}
-
-	var hcSecret *HostCatalogSecret
-	if plgResp != nil && plgResp.GetPersisted().GetSecrets() != nil {
-		hcSecret, err = newHostCatalogSecret(ctx, id, plgResp.GetPersisted().GetSecrets())
-		if err != nil {
-			return nil, nil, errors.Wrap(ctx, err, op)
-		}
-		dbWrapper, err := r.kms.GetWrapper(ctx, c.ScopeId, kms.KeyPurposeDatabase)
-		if err != nil {
-			return nil, nil, errors.Wrap(ctx, err, op, errors.WithMsg("unable to get db wrapper"))
-		}
-		if err := hcSecret.encrypt(ctx, dbWrapper); err != nil {
-			return nil, nil, errors.Wrap(ctx, err, op)
-		}
-	}
 
 	oplogWrapper, err := r.kms.GetWrapper(ctx, c.ScopeId, kms.KeyPurposeOplog)
 	if err != nil {
 		return nil, nil, errors.Wrap(ctx, err, op, errors.WithMsg("unable to get oplog wrapper"))
 	}
 
+	// If the call to the plugin succeeded, we do not want to call it again if
+	// the transaction failed and is being retried.
+	var pluginCalledSuccessfully bool
+	var plgResp *plgpb.OnCreateCatalogResponse
+
 	var newHostCatalog *HostCatalog
 	_, err = r.writer.DoTx(
 		ctx,
@@ -122,17 +106,40 @@ func (r *Repository) CreateCatalog(ctx context.Context, c *HostCatalog, _ ...Opt
 			}
 			msgs = append(msgs, &cOplogMsg)
 
-			if hcSecret != nil {
-				newSecret := hcSecret.clone()
-				q, v := newSecret.upsertQuery()
-				rows, err := w.Exec(ctx, q, v)
+			if !pluginCalledSuccessfully {
+				plgResp, err = plgClient.OnCreateCatalog(ctx, &plgpb.OnCreateCatalogRequest{Catalog: plgHc})
+				if err != nil {
+					if status.Code(err) != codes.Unimplemented {
+						return errors.Wrap(ctx, err, op)
+					}
+				}
+				pluginCalledSuccessfully = true
+			}
+
+			if plgResp != nil && plgResp.GetPersisted().GetSecrets() != nil {
+				hcSecret, err := newHostCatalogSecret(ctx, id, plgResp.GetPersisted().GetSecrets())
 				if err != nil {
 					return errors.Wrap(ctx, err, op)
 				}
-				if rows > 1 {
-					return errors.New(ctx, errors.MultipleRecords, op, "more than 1 catalog secret would have been created")
+				dbWrapper, err := r.kms.GetWrapper(ctx, c.ScopeId, kms.KeyPurposeDatabase)
+				if err != nil {
+					return errors.Wrap(ctx, err, op, errors.WithMsg("unable to get db wrapper"))
+				}
+				if err := hcSecret.encrypt(ctx, dbWrapper); err != nil {
+					return errors.Wrap(ctx, err, op)
+				}
+				if hcSecret != nil {
+					newSecret := hcSecret.clone()
+					q, v := newSecret.upsertQuery()
+					rows, err := w.Exec(ctx, q, v)
+					if err != nil {
+						return errors.Wrap(ctx, err, op)
+					}
+					if rows > 1 {
+						return errors.New(ctx, errors.MultipleRecords, op, "more than 1 catalog secret would have been created")
+					}
+					msgs = append(msgs, newSecret.oplogMessage(db.CreateOp))
 				}
-				msgs = append(msgs, newSecret.oplogMessage(db.CreateOp))
 			}
 
 			metadata := c.oplog(oplog.OpType_OP_TYPE_CREATE)
@@ -284,51 +291,19 @@ func (r *Repository) UpdateCatalog(ctx context.Context, c *HostCatalog, version
 		return nil, nil, db.NoRowsAffected, errors.Wrap(ctx, err, op)
 	}
 
-	plgResp, err := plgClient.OnUpdateCatalog(ctx, &plgpb.OnUpdateCatalogRequest{
-		CurrentCatalog: currPlgHc,
-		NewCatalog:     newPlgHc,
-		Persisted:      currentCatalogPersisted,
-	})
+	dbWrapper, err := r.kms.GetWrapper(ctx, newCatalog.ScopeId, kms.KeyPurposeDatabase)
 	if err != nil {
-		if status.Code(err) != codes.Unimplemented {
-			return nil, nil, db.NoRowsAffected, errors.Wrap(ctx, err, op)
-		}
+		return nil, nil, db.NoRowsAffected, errors.Wrap(ctx, err, op, errors.WithMsg("unable to get db wrapper"))
 	}
-
-	// Success for OnUpdateCatalog. This means we can start updating.
-	// First, check for returned persisted data and encrypt it.
-	var hcSecret *HostCatalogSecret
-	var deleteSecrets bool
-	if plgResp != nil && plgResp.GetPersisted().GetSecrets() != nil {
-		if len(plgResp.GetPersisted().GetSecrets().GetFields()) == 0 {
-			// Flag the secret to be deleted.
-			hcSecret, err = newHostCatalogSecret(ctx, currentCatalog.GetPublicId(), plgResp.GetPersisted().GetSecrets())
-			if err != nil {
-				return nil, nil, db.NoRowsAffected, errors.Wrap(ctx, err, op)
-			}
-
-			deleteSecrets = true
-		} else {
-			hcSecret, err = newHostCatalogSecret(ctx, currentCatalog.GetPublicId(), plgResp.GetPersisted().GetSecrets())
-			if err != nil {
-				return nil, nil, db.NoRowsAffected, errors.Wrap(ctx, err, op)
-			}
-			dbWrapper, err := r.kms.GetWrapper(ctx, newCatalog.ScopeId, kms.KeyPurposeDatabase)
-			if err != nil {
-				return nil, nil, db.NoRowsAffected, errors.Wrap(ctx, err, op, errors.WithMsg("unable to get db wrapper"))
-			}
-			if err := hcSecret.encrypt(ctx, dbWrapper); err != nil {
-				return nil, nil, db.NoRowsAffected, errors.Wrap(ctx, err, op)
-			}
-		}
-	}
-
 	// Get the oplog.
 	oplogWrapper, err := r.kms.GetWrapper(ctx, newCatalog.ScopeId, kms.KeyPurposeOplog)
 	if err != nil {
 		return nil, nil, db.NoRowsAffected, errors.Wrap(ctx, err, op, errors.WithMsg("unable to get oplog wrapper"))
 	}
 
+	var pluginCalledSuccessfully bool
+	var plgResp *plgpb.OnUpdateCatalogResponse
+
 	var recordUpdated bool
 	var returnedCatalog *HostCatalog
 	_, err = r.writer.DoTx(
@@ -368,13 +343,31 @@ func (r *Repository) UpdateCatalog(ctx context.Context, c *HostCatalog, version
 				returnedCatalog = currentCatalog.clone()
 			}
 
-			if hcSecret != nil {
-				if deleteSecrets {
+			if !pluginCalledSuccessfully {
+				plgResp, err = plgClient.OnUpdateCatalog(ctx, &plgpb.OnUpdateCatalogRequest{
+					CurrentCatalog: currPlgHc,
+					NewCatalog:     newPlgHc,
+					Persisted:      currentCatalogPersisted,
+				})
+				if err != nil {
+					if status.Code(err) != codes.Unimplemented {
+						return errors.Wrap(ctx, err, op)
+					}
+				}
+				pluginCalledSuccessfully = true
+			}
+			var updatedPersisted bool
+			if plgResp != nil && plgResp.GetPersisted().GetSecrets() != nil {
+				if len(plgResp.GetPersisted().GetSecrets().GetFields()) == 0 {
+					// Flag the secret to be deleted.
+					hcSecret, err := newHostCatalogSecret(ctx, currentCatalog.GetPublicId(), plgResp.GetPersisted().GetSecrets())
+					if err != nil {
+						return errors.Wrap(ctx, err, op)
+					}
 					// We didn't set/encrypt the persisted data because there was
 					// none returned. Just delete the entry.
 					deletedSecret := hcSecret.clone()
 					q, v := deletedSecret.deleteQuery()
-					var err error
 					secretsUpdated, err := w.Exec(ctx, q, v)
 					if err != nil {
 						return errors.Wrap(ctx, err, op)
@@ -382,12 +375,20 @@ func (r *Repository) UpdateCatalog(ctx context.Context, c *HostCatalog, version
 					if secretsUpdated != 1 {
 						return errors.New(ctx, errors.MultipleRecords, op, fmt.Sprintf("expected 1 catalog secret to be deleted, got %d", secretsUpdated))
 					}
+					updatedPersisted = true
 					msgs = append(msgs, deletedSecret.oplogMessage(db.DeleteOp))
 				} else {
+					hcSecret, err := newHostCatalogSecret(ctx, currentCatalog.GetPublicId(), plgResp.GetPersisted().GetSecrets())
+					if err != nil {
+						return errors.Wrap(ctx, err, op)
+					}
+					if err := hcSecret.encrypt(ctx, dbWrapper); err != nil {
+						return errors.Wrap(ctx, err, op)
+					}
+
 					// Update the secrets.
 					updatedSecret := hcSecret.clone()
 					q, v := updatedSecret.upsertQuery()
-					var err error
 					secretsUpdated, err := w.Exec(ctx, q, v)
 					if err != nil {
 						return errors.Wrap(ctx, err, op)
@@ -395,32 +396,33 @@ func (r *Repository) UpdateCatalog(ctx context.Context, c *HostCatalog, version
 					if secretsUpdated != 1 {
 						return errors.New(ctx, errors.MultipleRecords, op, fmt.Sprintf("expected 1 catalog secret to be updated, got %d", secretsUpdated))
 					}
+					updatedPersisted = true
 					msgs = append(msgs, updatedSecret.oplogMessage(db.UpdateOp))
 				}
+			}
 
-				if !recordUpdated {
-					// we only updated secrets, so we need to increment the
-					// version of the host catalog manually.
-					returnedCatalog = newCatalog.clone()
-					returnedCatalog.Version = uint32(version) + 1
-					var cOplogMsg oplog.Message
-					catalogsUpdated, err := w.Update(
-						ctx,
-						returnedCatalog,
-						[]string{"version"},
-						[]string{},
-						db.NewOplogMsg(&cOplogMsg),
-						db.WithVersion(&version),
-					)
-					if err != nil {
-						return errors.Wrap(ctx, err, op)
-					}
-					if catalogsUpdated != 1 {
-						return errors.New(ctx, errors.MultipleRecords, op, fmt.Sprintf("expected 1 catalog to be updated, got %d", catalogsUpdated))
-					}
-					msgs = append(msgs, &cOplogMsg)
-					recordUpdated = true
+			if !recordUpdated && updatedPersisted {
+				// we only updated secrets, so we need to increment the
+				// version of the host catalog manually.
+				returnedCatalog = newCatalog.clone()
+				returnedCatalog.Version = uint32(version) + 1
+				var cOplogMsg oplog.Message
+				catalogsUpdated, err := w.Update(
+					ctx,
+					returnedCatalog,
+					[]string{"version"},
+					[]string{},
+					db.NewOplogMsg(&cOplogMsg),
+					db.WithVersion(&version),
+				)
+				if err != nil {
+					return errors.Wrap(ctx, err, op)
 				}
+				if catalogsUpdated != 1 {
+					return errors.New(ctx, errors.MultipleRecords, op, fmt.Sprintf("expected 1 catalog to be updated, got %d", catalogsUpdated))
+				}
+				msgs = append(msgs, &cOplogMsg)
+				recordUpdated = true
 			}
 
 			if len(msgs) != 0 {

internal/host/plugin/repository_host_catalog_test.go

@@ -36,26 +36,12 @@ func TestRepository_CreateCatalog(t *testing.T) {
 	plg := hostplg.TestPlugin(t, conn, "test")
 	unimplementedPlugin := hostplg.TestPlugin(t, conn, "unimplemented")
 
-	// gotPluginAttrs tracks which attributes a plugin has received through a closure and can be compared in the
-	// test against the expected values sent to the plugin.
-	var gotPluginAttrs *structpb.Struct
-	plgm := map[string]plgpb.HostPluginServiceClient{
-		plg.GetPublicId(): &WrappingPluginClient{
-			Server: &TestPluginServer{
-				OnCreateCatalogFn: func(_ context.Context, req *plgpb.OnCreateCatalogRequest) (*plgpb.OnCreateCatalogResponse, error) {
-					gotPluginAttrs = req.GetCatalog().GetAttributes()
-					return &plgpb.OnCreateCatalogResponse{Persisted: &plgpb.HostCatalogPersisted{Secrets: req.GetCatalog().GetSecrets()}}, nil
-				},
-			},
-		},
-		unimplementedPlugin.GetPublicId(): &WrappingPluginClient{Server: &plgpb.UnimplementedHostPluginServiceServer{}},
-	}
-
 	tests := []struct {
 		name       string
 		in         *HostCatalog
 		opts       []Option
 		want       *HostCatalog
+		wantPluginCalled bool
 		wantSecret *structpb.Struct
 		wantIsErr  errors.Code
 	}{
@@ -114,6 +100,7 @@ func TestRepository_CreateCatalog(t *testing.T) {
 					Attributes: []byte{},
 				},
 			},
+			wantPluginCalled: true,
 		},
 		{
 			name: "valid-unimplemented-plugin",
@@ -131,6 +118,7 @@ func TestRepository_CreateCatalog(t *testing.T) {
 					Attributes: []byte{},
 				},
 			},
+			wantPluginCalled: true,
 		},
 		{
 			name: "not-found-plugin",
@@ -161,6 +149,7 @@ func TestRepository_CreateCatalog(t *testing.T) {
 					Attributes: []byte{},
 				},
 			},
+			wantPluginCalled: true,
 		},
 		{
 			name: "valid-with-description",
@@ -180,6 +169,7 @@ func TestRepository_CreateCatalog(t *testing.T) {
 					Attributes:  []byte{},
 				},
 			},
+			wantPluginCalled: true,
 		},
 		{
 			name: "valid-with-attributes",
@@ -212,6 +202,7 @@ func TestRepository_CreateCatalog(t *testing.T) {
 					}(),
 				},
 			},
+			wantPluginCalled: true,
 		},
 		{
 			name: "valid-with-secret",
@@ -249,6 +240,7 @@ func TestRepository_CreateCatalog(t *testing.T) {
 				require.NoError(t, err)
 				return st
 			}(),
+			wantPluginCalled: true,
 		},
 	}
 
@@ -257,10 +249,34 @@ func TestRepository_CreateCatalog(t *testing.T) {
 		t.Run(tt.name, func(t *testing.T) {
 			assert := assert.New(t)
 			kmsCache := kms.TestKms(t, conn, wrapper)
+
+			// gotPluginAttrs tracks which attributes a plugin has received through a closure and can be compared in the
+			// test against the expected values sent to the plugin.
+			var gotPluginAttrs *structpb.Struct
+			var pluginCalled bool
+			plgm := map[string]plgpb.HostPluginServiceClient{
+				plg.GetPublicId(): &WrappingPluginClient{
+					Server: &TestPluginServer{
+						OnCreateCatalogFn: func(_ context.Context, req *plgpb.OnCreateCatalogRequest) (*plgpb.OnCreateCatalogResponse, error) {
+							pluginCalled = true
+							gotPluginAttrs = req.GetCatalog().GetAttributes()
+							return &plgpb.OnCreateCatalogResponse{Persisted: &plgpb.HostCatalogPersisted{Secrets: req.GetCatalog().GetSecrets()}}, nil
+						},
+					},
+				},
+				unimplementedPlugin.GetPublicId(): &WrappingPluginClient{Server: &TestPluginServer{
+					OnCreateCatalogFn: func(ctx context.Context, req *plgpb.OnCreateCatalogRequest) (*plgpb.OnCreateCatalogResponse, error) {
+						pluginCalled = true
+						gotPluginAttrs = req.GetCatalog().GetAttributes()
+						return plgpb.UnimplementedHostPluginServiceServer{}.OnCreateCatalog(ctx, req)
+					},
+				}},
+			}
 			repo, err := NewRepository(rw, rw, kmsCache, sched, plgm)
 			assert.NoError(err)
 			assert.NotNil(repo)
 			got, _, err := repo.CreateCatalog(ctx, tt.in, tt.opts...)
+			assert.Equal(tt.wantPluginCalled, pluginCalled)
 			if tt.wantIsErr != 0 {
 				assert.Truef(errors.Match(errors.T(tt.wantIsErr), err), "want err: %q got: %q", tt.wantIsErr, err)
 				assert.Nil(got)
@@ -308,6 +324,17 @@ func TestRepository_CreateCatalog(t *testing.T) {
 	t.Run("invalid-duplicate-names", func(t *testing.T) {
 		assert := assert.New(t)
 		kms := kms.TestKms(t, conn, wrapper)
+		var pluginCalled bool
+		plgm := map[string]plgpb.HostPluginServiceClient{
+			plg.GetPublicId(): &WrappingPluginClient{
+				Server: &TestPluginServer{
+					OnCreateCatalogFn: func(_ context.Context, req *plgpb.OnCreateCatalogRequest) (*plgpb.OnCreateCatalogResponse, error) {
+						pluginCalled = true
+						return &plgpb.OnCreateCatalogResponse{Persisted: &plgpb.HostCatalogPersisted{Secrets: req.GetCatalog().GetSecrets()}}, nil
+					},
+				},
+			},
+		}
 		repo, err := NewRepository(rw, rw, kms, sched, plgm)
 		assert.NoError(err)
 		assert.NotNil(repo)
@@ -324,13 +351,18 @@ func TestRepository_CreateCatalog(t *testing.T) {
 		got, _, err := repo.CreateCatalog(context.Background(), in)
 		assert.NoError(err)
 		assert.NotNil(got)
+		assert.True(pluginCalled)
 		assertPluginBasedPublicId(t, HostCatalogPrefix, got.PublicId)
 		assert.NotSame(in, got)
 		assert.Equal(in.Name, got.Name)
 		assert.Equal(in.Description, got.Description)
 		assert.Equal(got.CreateTime, got.UpdateTime)
 
+		// Reset pluginCalled so we can see that the plugin wasn't called w/
+		// the duplicate name.
+		pluginCalled = false
 		got2, _, err := repo.CreateCatalog(context.Background(), in)
+		assert.False(pluginCalled)
 		assert.Truef(errors.Match(errors.T(errors.NotUnique), err), "want err code: %v got err: %v", errors.NotUnique, err)
 		assert.Nil(got2)
 	})
@@ -338,6 +370,17 @@ func TestRepository_CreateCatalog(t *testing.T) {
 	t.Run("valid-duplicate-names-diff-scopes", func(t *testing.T) {
 		assert := assert.New(t)
 		kms := kms.TestKms(t, conn, wrapper)
+		var pluginCalled bool
+		plgm := map[string]plgpb.HostPluginServiceClient{
+			plg.GetPublicId(): &WrappingPluginClient{
+				Server: &TestPluginServer{
+					OnCreateCatalogFn: func(_ context.Context, req *plgpb.OnCreateCatalogRequest) (*plgpb.OnCreateCatalogResponse, error) {
+						pluginCalled = true
+						return &plgpb.OnCreateCatalogResponse{Persisted: &plgpb.HostCatalogPersisted{Secrets: req.GetCatalog().GetSecrets()}}, nil
+					},
+				},
+			},
+		}
 		repo, err := NewRepository(rw, rw, kms, sched, plgm)
 		assert.NoError(err)
 		assert.NotNil(repo)
@@ -355,6 +398,7 @@ func TestRepository_CreateCatalog(t *testing.T) {
 		got, _, err := repo.CreateCatalog(context.Background(), in)
 		assert.NoError(err)
 		assert.NotNil(got)
+		assert.True(pluginCalled)
 		assertPluginBasedPublicId(t, HostCatalogPrefix, got.PublicId)
 		assert.NotSame(in, got)
 		assert.Equal(in.Name, got.Name)
@@ -362,9 +406,11 @@ func TestRepository_CreateCatalog(t *testing.T) {
 		assert.Equal(got.CreateTime, got.UpdateTime)
 
 		in2.ScopeId = org.GetPublicId()
+		pluginCalled = false // reset pluginCalled for this next call to Create
 		got2, _, err := repo.CreateCatalog(context.Background(), in2)
 		assert.NoError(err)
 		assert.NotNil(got2)
+		assert.True(pluginCalled)
 		assertPluginBasedPublicId(t, HostCatalogPrefix, got2.PublicId)
 		assert.NotSame(in2, got2)
 		assert.Equal(in2.Name, got2.Name)

internal/host/plugin/repository_host_set.go

@@ -85,11 +85,6 @@ func (r *Repository) CreateSet(ctx context.Context, scopeId string, s *HostSet,
 	if err != nil {
 		return nil, nil, errors.Wrap(ctx, err, op)
 	}
-	if _, err := plgClient.OnCreateSet(ctx, &plgpb.OnCreateSetRequest{Catalog: plgHc, Set: plgHs, Persisted: per}); err != nil {
-		if status.Code(err) != codes.Unimplemented {
-			return nil, nil, errors.Wrap(ctx, err, op)
-		}
-	}
 
 	var preferredEndpoints []interface{}
 	if s.PreferredEndpoints != nil {
@@ -108,6 +103,8 @@ func (r *Repository) CreateSet(ctx context.Context, scopeId string, s *HostSet,
 		return nil, nil, errors.Wrap(ctx, err, op, errors.WithMsg("unable to get oplog wrapper"))
 	}
 
+	var calledPluginSuccessfully bool
+
 	var returnedHostSet *HostSet
 	_, err = r.writer.DoTx(
 		ctx,
@@ -136,6 +133,15 @@ func (r *Repository) CreateSet(ctx context.Context, scopeId string, s *HostSet,
 				msgs = append(msgs, peOplogMsgs...)
 			}
 
+			if !calledPluginSuccessfully {
+				if _, err := plgClient.OnCreateSet(ctx, &plgpb.OnCreateSetRequest{Catalog: plgHc, Set: plgHs, Persisted: per}); err != nil {
+					if status.Code(err) != codes.Unimplemented {
+						return errors.Wrap(ctx, err, op)
+					}
+				}
+				calledPluginSuccessfully = true
+			}
+
 			metadata := s.oplog(oplog.OpType_OP_TYPE_CREATE)
 			if err := w.WriteOplogEntryWith(ctx, oplogWrapper, ticket, metadata, msgs); err != nil {
 				return errors.Wrap(ctx, err, op, errors.WithMsg("unable to write oplog"))

internal/host/plugin/repository_host_set_test.go

@@ -38,15 +38,6 @@ func TestRepository_CreateSet(t *testing.T) {
 	plg := hostplg.TestPlugin(t, conn, "create")
 	unimplementedPlugin := hostplg.TestPlugin(t, conn, "unimplemented")
 
-	var pluginReceivedAttrs *structpb.Struct
-	plgm := map[string]plgpb.HostPluginServiceClient{
-		plg.GetPublicId(): NewWrappingPluginClient(TestPluginServer{OnCreateSetFn: func(ctx context.Context, req *plgpb.OnCreateSetRequest) (*plgpb.OnCreateSetResponse, error) {
-			pluginReceivedAttrs = req.GetSet().GetAttributes()
-			return &plgpb.OnCreateSetResponse{}, nil
-		}}),
-		unimplementedPlugin.GetPublicId(): NewWrappingPluginClient(&plgpb.UnimplementedHostPluginServiceServer{}),
-	}
-
 	catalog := TestCatalog(t, conn, prj.PublicId, plg.GetPublicId())
 	unimplementedPluginCatalog := TestCatalog(t, conn, prj.PublicId, plg.GetPublicId())
 	attrs := []byte{}
@@ -56,6 +47,7 @@ func TestRepository_CreateSet(t *testing.T) {
 		in        *HostSet
 		opts      []Option
 		want      *HostSet
+		wantPluginCalled bool
 		wantIsErr errors.Code
 	}{
 		{
@@ -110,6 +102,7 @@ func TestRepository_CreateSet(t *testing.T) {
 					Attributes: attrs,
 				},
 			},
+			wantPluginCalled: true,
 		},
 		{
 			name: "valid-preferred-endpoints",
@@ -127,6 +120,7 @@ func TestRepository_CreateSet(t *testing.T) {
 					PreferredEndpoints: []string{"cidr:1.2.3.4/32", "dns:a.b.c"},
 				},
 			},
+			wantPluginCalled: true,
 		},
 		{
 			name: "valid-with-name",
@@ -144,6 +138,7 @@ func TestRepository_CreateSet(t *testing.T) {
 					Attributes: attrs,
 				},
 			},
+			wantPluginCalled: true,
 		},
 		{
 			name: "valid-unimplemented-plugin",
@@ -161,6 +156,7 @@ func TestRepository_CreateSet(t *testing.T) {
 					Attributes: attrs,
 				},
 			},
+			wantPluginCalled: true,
 		},
 		{
 			name: "valid-with-description",
@@ -178,6 +174,7 @@ func TestRepository_CreateSet(t *testing.T) {
 					Attributes:  attrs,
 				},
 			},
+			wantPluginCalled: true,
 		},
 		{
 			name: "valid-with-custom-attributes",
@@ -208,6 +205,7 @@ func TestRepository_CreateSet(t *testing.T) {
 					}(),
 				},
 			},
+			wantPluginCalled: true,
 		},
 	}
 
@@ -215,10 +213,25 @@ func TestRepository_CreateSet(t *testing.T) {
 		tt := tt
 		t.Run(tt.name, func(t *testing.T) {
 			assert, require := assert.New(t), require.New(t)
+			var pluginReceivedAttrs *structpb.Struct
+			var pluginCalled bool
+			plgm := map[string]plgpb.HostPluginServiceClient{
+				plg.GetPublicId(): NewWrappingPluginClient(TestPluginServer{OnCreateSetFn: func(ctx context.Context, req *plgpb.OnCreateSetRequest) (*plgpb.OnCreateSetResponse, error) {
+					pluginCalled = true
+					pluginReceivedAttrs = req.GetSet().GetAttributes()
+					return &plgpb.OnCreateSetResponse{}, nil
+				}}),
+				unimplementedPlugin.GetPublicId(): NewWrappingPluginClient(TestPluginServer{OnCreateSetFn: func(ctx context.Context, req *plgpb.OnCreateSetRequest) (*plgpb.OnCreateSetResponse, error) {
+					pluginCalled = true
+					pluginReceivedAttrs = req.GetSet().GetAttributes()
+					return plgpb.UnimplementedHostPluginServiceServer{}.OnCreateSet(ctx, req)
+				}}),
+			}
 			repo, err := NewRepository(rw, rw, kms, sched, plgm)
 			require.NoError(err)
 			require.NotNil(repo)
 			got, plgInfo, err := repo.CreateSet(context.Background(), prj.GetPublicId(), tt.in, tt.opts...)
+			assert.Equal(tt.wantPluginCalled, pluginCalled)
 			if tt.wantIsErr != 0 {
 				assert.Truef(errors.Match(errors.T(tt.wantIsErr), err), "want err: %q got: %q", tt.wantIsErr, err)
 				assert.Nil(got)
@@ -243,6 +256,13 @@ func TestRepository_CreateSet(t *testing.T) {
 
 	t.Run("invalid-duplicate-names", func(t *testing.T) {
 		assert, require := assert.New(t), require.New(t)
+		var pluginCalled bool
+		plgm := map[string]plgpb.HostPluginServiceClient{
+			plg.GetPublicId(): NewWrappingPluginClient(TestPluginServer{OnCreateSetFn: func(ctx context.Context, req *plgpb.OnCreateSetRequest) (*plgpb.OnCreateSetResponse, error) {
+				pluginCalled = true
+				return &plgpb.OnCreateSetResponse{}, nil
+			}}),
+		}
 		repo, err := NewRepository(rw, rw, kms, sched, plgm)
 		require.NoError(err)
 		require.NotNil(repo)
@@ -261,19 +281,31 @@ func TestRepository_CreateSet(t *testing.T) {
 		got, _, err := repo.CreateSet(context.Background(), prj.GetPublicId(), in)
 		require.NoError(err)
 		require.NotNil(got)
+		assert.True(pluginCalled)
 		assert.True(strings.HasPrefix(got.GetPublicId(), HostSetPrefix))
 		assert.NotSame(in, got)
 		assert.Equal(in.Name, got.GetName())
 		assert.Equal(in.Description, got.GetDescription())
 		assert.Equal(got.GetCreateTime(), got.GetUpdateTime())
 
+		// reset pluginCalled so we can see the duplicate name causes the plugin
+		// to not get called.
+		pluginCalled = false
 		got2, _, err := repo.CreateSet(context.Background(), prj.GetPublicId(), in)
+		assert.False(pluginCalled)
 		assert.Truef(errors.Match(errors.T(errors.NotUnique), err), "want err code: %v got err: %v", errors.NotUnique, err)
 		assert.Nil(got2)
 	})
 
 	t.Run("valid-duplicate-names-diff-catalogs", func(t *testing.T) {
 		assert, require := assert.New(t), require.New(t)
+		var pluginCalled bool
+		plgm := map[string]plgpb.HostPluginServiceClient{
+			plg.GetPublicId(): NewWrappingPluginClient(TestPluginServer{OnCreateSetFn: func(ctx context.Context, req *plgpb.OnCreateSetRequest) (*plgpb.OnCreateSetResponse, error) {
+				pluginCalled = true
+				return &plgpb.OnCreateSetResponse{}, nil
+			}}),
+		}
 		repo, err := NewRepository(rw, rw, kms, sched, plgm)
 		require.NoError(err)
 		require.NotNil(repo)
@@ -291,7 +323,11 @@ func TestRepository_CreateSet(t *testing.T) {
 		in2 := in.clone()
 
 		in.CatalogId = catalogA.PublicId
+		// reset pluginCalled so we can see that the plugin is called during this
+		// createSet call
+		pluginCalled = false
 		got, _, err := repo.CreateSet(context.Background(), prj.GetPublicId(), in)
+		assert.True(pluginCalled)
 		require.NoError(err)
 		require.NotNil(got)
 		assert.True(strings.HasPrefix(got.GetPublicId(), HostSetPrefix))
@@ -301,9 +337,13 @@ func TestRepository_CreateSet(t *testing.T) {
 		assert.Equal(got.GetCreateTime(), got.GetUpdateTime())
 
 		in2.CatalogId = catalogB.PublicId
+		// reset pluginCalled so we can see that the plugin is called during this
+		// createSet call
+		pluginCalled = false
 		got2, _, err := repo.CreateSet(context.Background(), prj.GetPublicId(), in2)
 		require.NoError(err)
 		require.NotNil(got2)
+		assert.True(pluginCalled)
 		assert.True(strings.HasPrefix(got.GetPublicId(), HostSetPrefix))
 		assert.NotSame(in2, got2)
 		assert.Equal(in2.Name, got2.GetName())