Increment version of host during updates (#1707)

Currently host does not use a version, which means it's not protected
against race conditions of either itself or its value objects. This adds
and plumbs through version to ensure that if a host or its value objects
change that they are not going to be affected by a race condition.

internal/db/schema/migrations/oss/postgres/20/06_plugin_host_views.up.sql

@@ -53,10 +53,7 @@ from
 comment on view host_plugin_catalog_with_secret is
   'host plugin catalog with its associated persisted data';
 
--- host_plugin_host_with_value_obj_and_set_memberships is useful for reading a
--- plugin host with its associated value objects (ip addresses, dns names) and
--- set membership as columns with delimited values. The delimiter depends on
--- the value objects (e.g. if they need ordering).
+-- REPLACED in 20/08_plugin_host_views.up.sql
 create view host_plugin_host_with_value_obj_and_set_memberships as
 select
   h.public_id,

internal/db/schema/migrations/oss/postgres/20/08_plugin_host_views.up.sql

@@ -0,0 +1,41 @@
+begin;
+
+alter table host_plugin_host
+  add column version wt_version;
+
+create trigger update_version_column after update on host_plugin_host
+  for each row execute procedure update_version_column();
+
+drop view host_plugin_host_with_value_obj_and_set_memberships;
+
+-- host_plugin_host_with_value_obj_and_set_memberships is useful for reading a
+-- plugin host with its associated value objects (ip addresses, dns names) and
+-- set membership as columns with delimited values. The delimiter depends on
+-- the value objects (e.g. if they need ordering).
+create view host_plugin_host_with_value_obj_and_set_memberships as
+select
+  h.public_id,
+  h.catalog_id,
+  h.external_id,
+  hc.scope_id,
+  hc.plugin_id,
+  h.name,
+  h.description,
+  h.create_time,
+  h.update_time,
+  h.version,
+  -- the string_agg(..) column will be null if there are no associated value objects
+  string_agg(distinct host(hip.address), '|') as ip_addresses,
+  string_agg(distinct hdns.name, '|') as dns_names,
+  string_agg(distinct hpsm.set_id, '|') as set_ids
+from
+  host_plugin_host h
+    join host_plugin_catalog hc                  on h.catalog_id = hc.public_id
+    left outer join host_ip_address hip          on h.public_id = hip.host_id
+    left outer join host_dns_name hdns           on h.public_id = hdns.host_id
+    left outer join host_plugin_set_member hpsm  on h.public_id = hpsm.host_id
+group by h.public_id, hc.plugin_id, hc.scope_id;
+comment on view host_plugin_host_with_value_obj_and_set_memberships is
+  'host plugin host with its associated value objects';
+
+commit;

internal/host/plugin/host.go

@@ -116,6 +116,7 @@ type hostAgg struct {
 	Description string
 	CreateTime  *timestamp.Timestamp
 	UpdateTime  *timestamp.Timestamp
+	Version     uint32
 	IpAddresses string
 	DnsNames    string
 	SetIds      string
@@ -132,6 +133,7 @@ func (agg *hostAgg) toHost() *Host {
 	h.Description = agg.Description
 	h.CreateTime = agg.CreateTime
 	h.UpdateTime = agg.UpdateTime
+	h.Version = agg.Version
 
 	if agg.IpAddresses != "" {
 		h.IpAddresses = strings.Split(agg.IpAddresses, aggregateDelimiter)

internal/host/plugin/job_set_sync.go

@@ -414,22 +414,36 @@ func (r *SetSyncJob) upsertHosts(
 			db.ExpBackoff{},
 			func(r db.Reader, w db.Writer) error {
 				msgs := make([]*oplog.Message, 0)
-				ticket, err := w.GetTicket(hi.h)
+				ticket, err := w.GetTicket(ret)
 				if err != nil {
 					return errors.Wrap(ctx, err, op, errors.WithMsg("unable to get ticket"))
 				}
 
-				if hi.dirtyHost {
-					var hOplogMsg oplog.Message
-					onConflict := &db.OnConflict{
-						Target: db.Constraint("host_plugin_host_pkey"),
-						Action: db.SetColumns([]string{"name", "description"}),
-					}
-					if err := w.Create(ctx, ret, db.NewOplogMsg(&hOplogMsg), db.WithOnConflict(onConflict)); err != nil {
-						return errors.Wrap(ctx, err, op)
-					}
-					msgs = append(msgs, &hOplogMsg)
+				// We always write the host itself as we need to update version
+				// for optimistic locking for any value object update.
+				var hOplogMsg oplog.Message
+				onConflict := &db.OnConflict{
+					Target: db.Constraint("host_plugin_host_pkey"),
+					Action: db.SetColumns([]string{"name", "description", "version"}),
 				}
+				var rowsAffected int64
+				dbOpts := []db.Option{
+					db.NewOplogMsg(&hOplogMsg),
+					db.WithOnConflict(onConflict),
+					db.WithReturnRowsAffected(&rowsAffected),
+				}
+				version := ret.Version
+				if version > 0 {
+					dbOpts = append(dbOpts, db.WithVersion(&version))
+					ret.Version += 1
+				}
+				if err := w.Create(ctx, ret, dbOpts...); err != nil {
+					return errors.Wrap(ctx, err, op)
+				}
+				if rowsAffected != 1 {
+					return errors.New(ctx, errors.UnexpectedRowsAffected, op, "no rows affected during upsert")
+				}
+				msgs = append(msgs, &hOplogMsg)
 
 				// IP handling
 				{
@@ -440,7 +454,7 @@ func (r *SetSyncJob) upsertHosts(
 							return err
 						}
 						if count != len(hi.ipsToRemove) {
-							return errors.New(ctx, errors.UnexpectedRowsAffected, op, fmt.Sprintf("expected to remove %d ips from host %s, removed %d", len(hi.ipsToRemove), hi.h.PublicId, count))
+							return errors.New(ctx, errors.UnexpectedRowsAffected, op, fmt.Sprintf("expected to remove %d ips from host %s, removed %d", len(hi.ipsToRemove), ret.PublicId, count))
 						}
 						msgs = append(msgs, oplogMsgs...)
 					}
@@ -451,7 +465,7 @@ func (r *SetSyncJob) upsertHosts(
 							Action: db.DoNothing(true),
 						}
 						if err := w.CreateItems(ctx, hi.ipsToAdd.toArray(), db.NewOplogMsgs(&oplogMsgs), db.WithOnConflict(onConflict)); err != nil {
-							return errors.Wrap(ctx, err, op, errors.WithMsg(fmt.Sprintf("adding ips %v for host %q", hi.ipsToAdd.toArray(), hi.h.GetPublicId())))
+							return errors.Wrap(ctx, err, op, errors.WithMsg(fmt.Sprintf("adding ips %v for host %q", hi.ipsToAdd.toArray(), ret.GetPublicId())))
 						}
 						msgs = append(msgs, oplogMsgs...)
 					}
@@ -466,7 +480,7 @@ func (r *SetSyncJob) upsertHosts(
 							return err
 						}
 						if count != len(hi.dnsNamesToRemove) {
-							return errors.New(ctx, errors.UnexpectedRowsAffected, op, fmt.Sprintf("expected to remove %d dns names from host %s, removed %d", len(hi.dnsNamesToRemove), hi.h.PublicId, count))
+							return errors.New(ctx, errors.UnexpectedRowsAffected, op, fmt.Sprintf("expected to remove %d dns names from host %s, removed %d", len(hi.dnsNamesToRemove), ret.PublicId, count))
 						}
 						msgs = append(msgs, oplogMsgs...)
 					}
@@ -483,7 +497,7 @@ func (r *SetSyncJob) upsertHosts(
 					}
 				}
 
-				metadata := hi.h.oplog(oplog.OpType_OP_TYPE_UPDATE)
+				metadata := ret.oplog(oplog.OpType_OP_TYPE_UPDATE)
 				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/job_set_sync_test.go

@@ -2,6 +2,7 @@ package plugin
 
 import (
 	"context"
+	"fmt"
 	"testing"
 	"time"
 
@@ -159,17 +160,19 @@ func TestSetSyncJob_Run(t *testing.T) {
 
 	cat := TestCatalog(t, conn, prj.GetPublicId(), plg.GetPublicId())
 	set1 := TestSet(t, conn, kmsCache, sched, cat, plgm)
+	counter := new(uint32)
 	plgServer.ListHostsFn = func(ctx context.Context, req *plgpb.ListHostsRequest) (*plgpb.ListHostsResponse, error) {
 		assert.GreaterOrEqual(1, len(req.GetSets()))
 		var setIds []string
 		for _, s := range req.GetSets() {
 			setIds = append(setIds, s.GetId())
 		}
+		*counter += 1
 		return &plgpb.ListHostsResponse{
 			Hosts: []*plgpb.ListHostsResponseHost{
 				{
 					ExternalId:  "first",
-					IpAddresses: []string{"10.0.0.1"},
+					IpAddresses: []string{fmt.Sprintf("10.0.0.%d", *counter)},
 					DnsNames:    []string{"foo.com"},
 					SetIds:      setIds,
 				},
@@ -177,6 +180,9 @@ func TestSetSyncJob_Run(t *testing.T) {
 		}, nil
 	}
 
+	hostRepo, err := NewRepository(rw, rw, kmsCache, sche, plgm)
+	require.NoError(err)
+
 	hsa := &hostSetAgg{PublicId: set1.GetPublicId()}
 	require.NoError(rw.LookupByPublicId(ctx, hsa))
 	assert.Less(hsa.LastSyncTime.AsTime().UnixNano(), hsa.CreateTime.AsTime().UnixNano())
@@ -187,6 +193,13 @@ func TestSetSyncJob_Run(t *testing.T) {
 	// The single existing set should have been processed
 	assert.Equal(1, r.numSets)
 	assert.Equal(1, r.numProcessed)
+	// Check the version number of the host(s)
+	hosts, _, err := hostRepo.ListHostsByCatalogId(ctx, hsa.CatalogId)
+	require.NoError(err)
+	assert.Len(hosts, 1)
+	for _, host := range hosts {
+		assert.Equal(uint32(1), host.Version)
+	}
 
 	require.NoError(rw.LookupByPublicId(ctx, hsa))
 	assert.Greater(hsa.LastSyncTime.AsTime().UnixNano(), hsa.CreateTime.AsTime().UnixNano())
@@ -215,6 +228,13 @@ func TestSetSyncJob_Run(t *testing.T) {
 	// The single existing set should have been processed
 	assert.Equal(1, r.numSets)
 	assert.Equal(1, r.numProcessed)
+	// Check the version number of the host(s) again
+	hosts, _, err = hostRepo.ListHostsByCatalogId(ctx, hsa.CatalogId)
+	require.NoError(err)
+	assert.Len(hosts, 1)
+	for _, host := range hosts {
+		assert.Equal(uint32(2), host.Version)
+	}
 
 	// Run sync with a new second set
 	_ = TestSet(t, conn, kmsCache, sched, cat, plgm)

internal/host/plugin/repository_host_test.go

@@ -175,6 +175,7 @@ func TestJob_UpsertHosts(t *testing.T) {
 
 				ph.SetIds = ph.SetIds[0 : len(ph.SetIds)-1]
 				e.SetIds = e.SetIds[0 : len(e.SetIds)-1]
+				e.Version++
 				return &input{
 					catalog: catalog,
 					sets:    setIds,

internal/host/plugin/repository_host_util.go

@@ -75,6 +75,10 @@ func createNewHostMap(ctx context.Context,
 		// assume catalog ID and external ID don't change because if they do
 		// the public ID will be different as well.
 		currHost := currentHostMap[newHost.PublicId]
+		if currHost != nil {
+			hi.h.Version = currHost.Version
+		}
+
 		switch {
 		case currHost == nil,
 			currHost.Name != newHost.Name,

internal/host/plugin/store/host.pb.go

@@ -447,8 +447,9 @@ type Host struct {
 	// plugin_host_catalog and must be set.
 	// @inject_tag: `gorm:"not_null"`
 	CatalogId string `protobuf:"bytes,7,opt,name=catalog_id,json=catalogId,proto3" json:"catalog_id,omitempty" gorm:"not_null"`
-	// @inject_tag: `gorm:"-"`
-	Version uint32 `protobuf:"varint,8,opt,name=version,proto3" json:"version,omitempty" gorm:"-"`
+	// version allows optimistic locking of the resource
+	// @inject_tag: `gorm:"default:null"`
+	Version uint32 `protobuf:"varint,8,opt,name=version,proto3" json:"version,omitempty" gorm:"default:null"`
 	// ip_addresses are the ip addresses associated with this host and will
 	// be persisted in the db through the HostAddress message.
 	// @inject_tag: `gorm:"-"`

internal/host/plugin/testing.go

@@ -189,6 +189,7 @@ func TestExternalHosts(t *testing.T, catalog *HostCatalog, setIds []string, coun
 				ExternalId:  externalId,
 				IpAddresses: []string{ipStr},
 				DnsNames:    []string{dnsName},
+				Version:     1,
 			},
 		})
 	}

internal/proto/local/controller/storage/host/plugin/store/v1/host.proto

@@ -160,7 +160,8 @@ message Host {
   // @inject_tag: `gorm:"not_null"`
   string catalog_id = 7;
 
-  // @inject_tag: `gorm:"-"`
+  // version allows optimistic locking of the resource
+  // @inject_tag: `gorm:"default:null"`
   uint32 version = 8;
 
   // ip_addresses are the ip addresses associated with this host and will

internal/servers/controller/handlers/hosts/host_service.go

@@ -598,10 +598,6 @@ func toProto(ctx context.Context, in host.Host, opt ...handlers.Option) (*pb.Hos
 		if outputFields.Has(globals.ExternalIdField) {
 			out.ExternalId = h.ExternalId
 		}
-		// FIXME? Seems we should be _using_ a version...
-		if outputFields.Has(globals.VersionField) {
-			out.Version = 1
-		}
 	}
 	return &out, nil
 }

internal/servers/controller/handlers/hosts/host_service_test.go

@@ -758,7 +758,7 @@ func TestCreate(t *testing.T) {
 	}
 }
 
-func TestUpdate(t *testing.T) {
+func TestUpdate_Static(t *testing.T) {
 	t.Parallel()
 	conn, _ := db.TestSetup(t, "postgres")
 	wrapper := db.TestWrapper(t)