v2: ensure the controller caches are fully populated before first use (#20421)

The new controller caches are initialized before the DependencyMappers or the 
Reconciler run, but importantly they are not populated. The expectation is that 
when the WatchList call is made to the resource service it will send an initial 
snapshot of all resources matching a single type, and then perpetually send 
UPSERT/DELETE events afterward. This initial snapshot will cycle through the 
caching layer and will catch it up to reflect the stored data.

Critically the dependency mappers and reconcilers will race against the restoration 
of the caches on server startup or leader election. During this time it is possible a
 mapper or reconciler will use the cache to lookup a specific relationship and 
not find it. That very same reconciler may choose to then recompute some 
persisted resource and in effect rewind it to a prior computed state.

Change

- Since we are updating the behavior of the WatchList RPC, it was aligned to 
  match that of pbsubscribe and pbpeerstream using a protobuf oneof instead of the enum+fields option.

- The WatchList rpc now has 3 alternating response events: Upsert, Delete, 
  EndOfSnapshot. When set the initial batch of "snapshot" Upserts sent on a new 
  watch, those operations will be followed by an EndOfSnapshot event before beginning 
  the never-ending sequence of Upsert/Delete events.

- Within the Controller startup code we will launch N+1 goroutines to execute WatchList 
  queries for the watched types. The UPSERTs will be applied to the nascent cache
   only (no mappers will execute).

- Upon witnessing the END operation, those goroutines will terminate.

- When all cache priming routines complete, then the normal set of N+1 long lived 
watch routines will launch to officially witness all events in the system using the 
primed cached.

agent/grpc-external/services/resource/watch.go

@@ -62,22 +62,41 @@ func (s *Server) WatchList(req *pbresource.WatchListRequest, stream pbresource.R
 			return status.Errorf(codes.Internal, "failed next: %v", err)
 		}
 
+		var resource *pbresource.Resource
+		switch {
+		case event.GetUpsert() != nil:
+			resource = event.GetUpsert().GetResource()
+		case event.GetDelete() != nil:
+			resource = event.GetDelete().GetResource()
+		case event.GetEndOfSnapshot() != nil:
+			// skip the rest and send the event.
+			if err = stream.Send(event); err != nil {
+				return err
+			}
+			continue
+		default:
+			// skip unknown type of operation
+			continue
+		}
+
+		// From here on out we assume the event is operating on a non-nil resource.
+
 		// drop group versions that don't match
-		if event.Resource.Id.Type.GroupVersion != req.Type.GroupVersion {
+		if resource.Id.Type.GroupVersion != req.Type.GroupVersion {
 			continue
 		}
 
 		// Need to rebuild authorizer per resource since wildcard inputs may
 		// result in different tenancies. Consider caching per tenancy if this
 		// is deemed expensive.
-		entMeta = v2TenancyToV1EntMeta(event.Resource.Id.Tenancy)
+		entMeta = v2TenancyToV1EntMeta(resource.Id.Tenancy)
 		authz, authzContext, err = s.getAuthorizer(token, entMeta)
 		if err != nil {
 			return err
 		}
 
 		// filter out items that don't pass read ACLs
-		err = reg.ACLs.Read(authz, authzContext, event.Resource.Id, event.Resource)
+		err = reg.ACLs.Read(authz, authzContext, resource.Id, resource)
 		switch {
 		case acl.IsErrPermissionDenied(err):
 			continue

agent/grpc-external/services/resource/watch_test.go

@@ -17,6 +17,7 @@ import (
 	"google.golang.org/grpc/status"
 
 	"github.com/hashicorp/consul/acl"
+	"github.com/hashicorp/consul/acl/resolver"
 	svc "github.com/hashicorp/consul/agent/grpc-external/services/resource"
 	svctest "github.com/hashicorp/consul/agent/grpc-external/services/resource/testing"
 	"github.com/hashicorp/consul/agent/grpc-external/testutils"
@@ -129,9 +130,10 @@ func TestWatchList_TypeNotFound(t *testing.T) {
 func TestWatchList_GroupVersionMatches(t *testing.T) {
 	t.Parallel()
 
-	server := testServer(t)
+	b := svctest.NewResourceServiceBuilder().
-	client := testClient(t, server)
+		WithRegisterFns(demo.RegisterTypes)
-	demo.RegisterTypes(server.Registry)
+	client := b.Run(t)
+
 	ctx := context.Background()
 
 	// create a watch
@@ -143,29 +145,34 @@ func TestWatchList_GroupVersionMatches(t *testing.T) {
 	require.NoError(t, err)
 	rspCh := handleResourceStream(t, stream)
 
+	mustGetEndOfSnapshot(t, rspCh)
+
 	artist, err := demo.GenerateV2Artist()
 	require.NoError(t, err)
 
 	// insert and verify upsert event received
-	r1, err := server.Backend.WriteCAS(ctx, artist)
+	r1Resp, err := client.Write(ctx, &pbresource.WriteRequest{Resource: artist})
+	r1 := r1Resp.Resource
 	require.NoError(t, err)
 	rsp := mustGetResource(t, rspCh)
-	require.Equal(t, pbresource.WatchEvent_OPERATION_UPSERT, rsp.Operation)
+	require.NotNil(t, rsp.GetUpsert())
-	prototest.AssertDeepEqual(t, r1, rsp.Resource)
+	prototest.AssertDeepEqual(t, r1, rsp.GetUpsert().Resource)
 
 	// update and verify upsert event received
 	r2 := modifyArtist(t, r1)
-	r2, err = server.Backend.WriteCAS(ctx, r2)
+	r2Resp, err := client.Write(ctx, &pbresource.WriteRequest{Resource: r2})
 	require.NoError(t, err)
+	r2 = r2Resp.Resource
 	rsp = mustGetResource(t, rspCh)
-	require.Equal(t, pbresource.WatchEvent_OPERATION_UPSERT, rsp.Operation)
+	require.NotNil(t, rsp.GetUpsert())
-	prototest.AssertDeepEqual(t, r2, rsp.Resource)
+	prototest.AssertDeepEqual(t, r2, rsp.GetUpsert().Resource)
 
 	// delete and verify delete event received
-	err = server.Backend.DeleteCAS(ctx, r2.Id, r2.Version)
+	_, err = client.Delete(ctx, &pbresource.DeleteRequest{Id: r2.Id, Version: r2.Version})
 	require.NoError(t, err)
 	rsp = mustGetResource(t, rspCh)
-	require.Equal(t, pbresource.WatchEvent_OPERATION_DELETE, rsp.Operation)
+	require.NotNil(t, rsp.GetDelete())
+	prototest.AssertDeepEqual(t, r2.Id, rsp.GetDelete().Resource.Id)
 }
 
 func TestWatchList_Tenancy_Defaults_And_Normalization(t *testing.T) {
@@ -186,6 +193,8 @@ func TestWatchList_Tenancy_Defaults_And_Normalization(t *testing.T) {
 			require.NoError(t, err)
 			rspCh := handleResourceStream(t, stream)
 
+			mustGetEndOfSnapshot(t, rspCh)
+
 			// Testcase will pick one of executive, recordLabel or artist based on scope of type.
 			recordLabel, err := demo.GenerateV1RecordLabel("looney-tunes")
 			require.NoError(t, err)
@@ -215,8 +224,8 @@ func TestWatchList_Tenancy_Defaults_And_Normalization(t *testing.T) {
 			}
 
 			rsp := mustGetResource(t, rspCh)
-			require.Equal(t, pbresource.WatchEvent_OPERATION_UPSERT, rsp.Operation)
+			require.NotNil(t, rsp.GetUpsert())
-			prototest.AssertDeepEqual(t, expected, rsp.Resource)
+			prototest.AssertDeepEqual(t, expected, rsp.GetUpsert().Resource)
 		})
 	}
 }
@@ -227,9 +236,10 @@ func TestWatchList_GroupVersionMismatch(t *testing.T) {
 	// Then no watch events should be emitted
 	t.Parallel()
 
-	server := testServer(t)
+	b := svctest.NewResourceServiceBuilder().
-	demo.RegisterTypes(server.Registry)
+		WithRegisterFns(demo.RegisterTypes)
-	client := testClient(t, server)
+	client := b.Run(t)
+
 	ctx := context.Background()
 
 	// create a watch for TypeArtistV1
@@ -241,20 +251,24 @@ func TestWatchList_GroupVersionMismatch(t *testing.T) {
 	require.NoError(t, err)
 	rspCh := handleResourceStream(t, stream)
 
+	mustGetEndOfSnapshot(t, rspCh)
+
 	artist, err := demo.GenerateV2Artist()
 	require.NoError(t, err)
 
 	// insert
-	r1, err := server.Backend.WriteCAS(ctx, artist)
+	r1Resp, err := client.Write(ctx, &pbresource.WriteRequest{Resource: artist})
 	require.NoError(t, err)
+	r1 := r1Resp.Resource
 
 	// update
 	r2 := clone(r1)
-	r2, err = server.Backend.WriteCAS(ctx, r2)
+	r2Resp, err := client.Write(ctx, &pbresource.WriteRequest{Resource: r2})
 	require.NoError(t, err)
+	r2 = r2Resp.Resource
 
 	// delete
-	err = server.Backend.DeleteCAS(ctx, r2.Id, r2.Version)
+	_, err = client.Delete(ctx, &pbresource.DeleteRequest{Id: r2.Id, Version: r2.Version})
 	require.NoError(t, err)
 
 	// verify no events received
@@ -266,7 +280,7 @@ func TestWatchList_ACL_ListDenied(t *testing.T) {
 	t.Parallel()
 
 	// deny all
-	rspCh, _ := roundTripACL(t, testutils.ACLNoPermissions(t))
+	rspCh, _ := roundTripACL(t, testutils.ACLNoPermissions(t), true)
 
 	// verify key:list denied
 	err := mustGetError(t, rspCh)
@@ -284,7 +298,7 @@ func TestWatchList_ACL_ListAllowed_ReadDenied(t *testing.T) {
 		key_prefix "resource/" { policy = "list" }
 		key_prefix "resource/demo.v2.Artist/" { policy = "deny" }
 		`)
-	rspCh, _ := roundTripACL(t, authz)
+	rspCh, _ := roundTripACL(t, authz, false)
 
 	// verify resource filtered out by key:read denied, hence no events
 	mustGetNoResource(t, rspCh)
@@ -299,23 +313,26 @@ func TestWatchList_ACL_ListAllowed_ReadAllowed(t *testing.T) {
 		key_prefix "resource/" { policy = "list" }
 		key_prefix "resource/demo.v2.Artist/" { policy = "read" }
 	`)
-	rspCh, artist := roundTripACL(t, authz)
+	rspCh, artist := roundTripACL(t, authz, false)
 
 	// verify resource not filtered out by acl
 	event := mustGetResource(t, rspCh)
-	prototest.AssertDeepEqual(t, artist, event.Resource)
+
+	require.NotNil(t, event.GetUpsert())
+	prototest.AssertDeepEqual(t, artist, event.GetUpsert().Resource)
 }
 
 // roundtrip a WatchList which attempts to stream back a single write event
-func roundTripACL(t *testing.T, authz acl.Authorizer) (<-chan resourceOrError, *pbresource.Resource) {
+func roundTripACL(t *testing.T, authz acl.Authorizer, expectErr bool) (<-chan resourceOrError, *pbresource.Resource) {
-	server := testServer(t)
-	client := testClient(t, server)
-
 	mockACLResolver := &svc.MockACLResolver{}
 	mockACLResolver.On("ResolveTokenAndDefaultMeta", mock.Anything, mock.Anything, mock.Anything).
-		Return(authz, nil)
+		Return(resolver.Result{Authorizer: authz}, nil)
-	server.ACLResolver = mockACLResolver
+
-	demo.RegisterTypes(server.Registry)
+	b := svctest.NewResourceServiceBuilder().
+		WithRegisterFns(demo.RegisterTypes).
+		WithACLResolver(mockACLResolver)
+	client := b.Run(t)
+	server := b.ServiceImpl()
 
 	artist, err := demo.GenerateV2Artist()
 	require.NoError(t, err)
@@ -328,6 +345,10 @@ func roundTripACL(t *testing.T, authz acl.Authorizer) (<-chan resourceOrError, *
 	require.NoError(t, err)
 	rspCh := handleResourceStream(t, stream)
 
+	if !expectErr {
+		mustGetEndOfSnapshot(t, rspCh)
+	}
+
 	// induce single watch event
 	artist, err = server.Backend.WriteCAS(context.Background(), artist)
 	require.NoError(t, err)
@@ -348,6 +369,11 @@ func mustGetNoResource(t *testing.T, ch <-chan resourceOrError) {
 	}
 }
 
+func mustGetEndOfSnapshot(t *testing.T, ch <-chan resourceOrError) {
+	event := mustGetResource(t, ch)
+	require.NotNil(t, event.GetEndOfSnapshot(), "expected EndOfSnapshot but got got event %T", event.GetEvent())
+}
+
 func mustGetResource(t *testing.T, ch <-chan resourceOrError) *pbresource.WatchEvent {
 	t.Helper()
 
@@ -415,6 +441,8 @@ func TestWatchList_NoTenancy(t *testing.T) {
 	require.NoError(t, err)
 	rspCh := handleResourceStream(t, stream)
 
+	mustGetEndOfSnapshot(t, rspCh)
+
 	recordLabel, err := demo.GenerateV1RecordLabel("looney-tunes")
 	require.NoError(t, err)
 
@@ -424,6 +452,6 @@ func TestWatchList_NoTenancy(t *testing.T) {
 
 	rsp2 := mustGetResource(t, rspCh)
 
-	require.Equal(t, pbresource.WatchEvent_OPERATION_UPSERT, rsp2.Operation)
+	require.NotNil(t, rsp2.GetUpsert())
-	prototest.AssertDeepEqual(t, rsp1.Resource, rsp2.Resource)
+	prototest.AssertDeepEqual(t, rsp1.Resource, rsp2.GetUpsert().Resource)
 }

grpcmocks/proto-public/pbresource/mock_isWatchEvent_Event.go

@@ -0,0 +1,64 @@
+// Code generated by mockery v2.37.1. DO NOT EDIT.
+
+package mockpbresource
+
+import mock "github.com/stretchr/testify/mock"
+
+// isWatchEvent_Event is an autogenerated mock type for the isWatchEvent_Event type
+type isWatchEvent_Event struct {
+	mock.Mock
+}
+
+type isWatchEvent_Event_Expecter struct {
+	mock *mock.Mock
+}
+
+func (_m *isWatchEvent_Event) EXPECT() *isWatchEvent_Event_Expecter {
+	return &isWatchEvent_Event_Expecter{mock: &_m.Mock}
+}
+
+// isWatchEvent_Event provides a mock function with given fields:
+func (_m *isWatchEvent_Event) isWatchEvent_Event() {
+	_m.Called()
+}
+
+// isWatchEvent_Event_isWatchEvent_Event_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'isWatchEvent_Event'
+type isWatchEvent_Event_isWatchEvent_Event_Call struct {
+	*mock.Call
+}
+
+// isWatchEvent_Event is a helper method to define mock.On call
+func (_e *isWatchEvent_Event_Expecter) isWatchEvent_Event() *isWatchEvent_Event_isWatchEvent_Event_Call {
+	return &isWatchEvent_Event_isWatchEvent_Event_Call{Call: _e.mock.On("isWatchEvent_Event")}
+}
+
+func (_c *isWatchEvent_Event_isWatchEvent_Event_Call) Run(run func()) *isWatchEvent_Event_isWatchEvent_Event_Call {
+	_c.Call.Run(func(args mock.Arguments) {
+		run()
+	})
+	return _c
+}
+
+func (_c *isWatchEvent_Event_isWatchEvent_Event_Call) Return() *isWatchEvent_Event_isWatchEvent_Event_Call {
+	_c.Call.Return()
+	return _c
+}
+
+func (_c *isWatchEvent_Event_isWatchEvent_Event_Call) RunAndReturn(run func()) *isWatchEvent_Event_isWatchEvent_Event_Call {
+	_c.Call.Return(run)
+	return _c
+}
+
+// newIsWatchEvent_Event creates a new instance of isWatchEvent_Event. It also registers a testing interface on the mock and a cleanup function to assert the mocks expectations.
+// The first argument is typically a *testing.T value.
+func newIsWatchEvent_Event(t interface {
+	mock.TestingT
+	Cleanup(func())
+}) *isWatchEvent_Event {
+	mock := &isWatchEvent_Event{}
+	mock.Mock.Test(t)
+
+	t.Cleanup(func() { mock.AssertExpectations(t) })
+
+	return mock
+}

internal/controller/mem_consistency_test.go

@@ -9,6 +9,10 @@ import (
 	"testing"
 	"time"
 
+	"github.com/stretchr/testify/mock"
+	"github.com/stretchr/testify/require"
+	"google.golang.org/grpc"
+
 	mockpbresource "github.com/hashicorp/consul/grpcmocks/proto-public/pbresource"
 	"github.com/hashicorp/consul/internal/resource"
 	"github.com/hashicorp/consul/internal/resource/resourcetest"
@@ -16,9 +20,6 @@ import (
 	"github.com/hashicorp/consul/proto-public/pbresource"
 	"github.com/hashicorp/consul/proto/private/prototest"
 	"github.com/hashicorp/consul/sdk/testutil"
-	"github.com/stretchr/testify/mock"
-	"github.com/stretchr/testify/require"
-	"google.golang.org/grpc"
 )
 
 var (
@@ -151,14 +152,26 @@ func watchListEvents(t testutil.TestingTB, events ...*pbresource.WatchEvent) pbr
 
 	// Return the events in the specified order as soon as they are requested
 	for _, event := range events {
+		evt := event
 		watchListClient.EXPECT().
 			Recv().
 			RunAndReturn(func() (*pbresource.WatchEvent, error) {
-				return event, nil
+				return evt, nil
 			}).
 			Once()
 	}
 
+	watchListClient.EXPECT().
+		Recv().
+		RunAndReturn(func() (*pbresource.WatchEvent, error) {
+			return &pbresource.WatchEvent{
+				Event: &pbresource.WatchEvent_EndOfSnapshot_{
+					EndOfSnapshot: &pbresource.WatchEvent_EndOfSnapshot{},
+				},
+			}, nil
+		}).
+		Once()
+
 	// Now that all specified events have been exhausted we loop until the test finishes
 	// and the context bound to the tests lifecycle has been cancelled. This prevents getting
 	// any weird errors from the controller manager/runner.
@@ -194,17 +207,27 @@ func TestControllerRuntimeMemoryCloning(t *testing.T) {
 
 	// Create the v1 watch list client to be returned when the controller runner
 	// calls WatchList on the v1 testing type.
-	v1WatchListClient := watchListEvents(t, &pbresource.WatchEvent{
+	v1WatchListClientCreate := func() pbresource.ResourceService_WatchListClient {
-		Operation: pbresource.WatchEvent_OPERATION_UPSERT,
+		return watchListEvents(t, &pbresource.WatchEvent{
-		Resource:  res1,
+			Event: &pbresource.WatchEvent_Upsert_{
-	})
+				Upsert: &pbresource.WatchEvent_Upsert{
+					Resource: res1,
+				},
+			},
+		})
+	}
 
 	// Create the v2 watch list client to be returned when the controller runner
 	// calls WatchList on the v2 testing type.
-	v2WatchListClient := watchListEvents(t, nil, &pbresource.WatchEvent{
+	v2WatchListClientCreate := func() pbresource.ResourceService_WatchListClient {
-		Operation: pbresource.WatchEvent_OPERATION_UPSERT,
+		return watchListEvents(t, nil, &pbresource.WatchEvent{
-		Resource:  res2,
+			Event: &pbresource.WatchEvent_Upsert_{
-	})
+				Upsert: &pbresource.WatchEvent_Upsert{
+					Resource: res2,
+				},
+			},
+		})
+	}
 
 	// Create the mock resource service client
 	mres := mockpbresource.NewResourceServiceClient(t)
@@ -218,8 +241,10 @@ func TestControllerRuntimeMemoryCloning(t *testing.T) {
 				Namespace: storage.Wildcard,
 			},
 		}).
-		Return(v2WatchListClient, nil).
+		RunAndReturn(func(_ context.Context, _ *pbresource.WatchListRequest, _ ...grpc.CallOption) (pbresource.ResourceService_WatchListClient, error) {
-		Once()
+			return v2WatchListClientCreate(), nil
+		}).
+		Twice() // once for cache prime, once for the rest
 
 	// Setup the expectation for the controller runner to issue a WatchList
 	// request for the secondary Watch type (fake v1 type)
@@ -231,8 +256,10 @@ func TestControllerRuntimeMemoryCloning(t *testing.T) {
 				Namespace: storage.Wildcard,
 			},
 		}).
-		Return(v1WatchListClient, nil).
+		RunAndReturn(func(_ context.Context, _ *pbresource.WatchListRequest, _ ...grpc.CallOption) (pbresource.ResourceService_WatchListClient, error) {
-		Once()
+			return v1WatchListClientCreate(), nil
+		}).
+		Twice() // once for cache prime, once for the rest
 
 	// The cloning resource clients will forward actual calls onto the main resource service client.
 	// Here we are configuring the service mock to return either of the resources depending on the
@@ -292,10 +319,15 @@ func TestControllerRunnerSharedMemoryCache(t *testing.T) {
 
 	// Create the v2 watch list client to be returned when the controller runner
 	// calls WatchList on the v2 testing type.
-	v2WatchListClient := watchListEvents(t, nil, &pbresource.WatchEvent{
+	v2WatchListClientCreate := func() pbresource.ResourceService_WatchListClient {
-		Operation: pbresource.WatchEvent_OPERATION_UPSERT,
+		return watchListEvents(t, nil, &pbresource.WatchEvent{
-		Resource:  res,
+			Event: &pbresource.WatchEvent_Upsert_{
-	})
+				Upsert: &pbresource.WatchEvent_Upsert{
+					Resource: res,
+				},
+			},
+		})
+	}
 
 	// Create the mock resource service client
 	mres := mockpbresource.NewResourceServiceClient(t)
@@ -309,8 +341,10 @@ func TestControllerRunnerSharedMemoryCache(t *testing.T) {
 				Namespace: storage.Wildcard,
 			},
 		}).
-		Return(v2WatchListClient, nil).
+		RunAndReturn(func(_ context.Context, _ *pbresource.WatchListRequest, _ ...grpc.CallOption) (pbresource.ResourceService_WatchListClient, error) {
-		Once()
+			return v2WatchListClientCreate(), nil
+		}).
+		Twice() // once for cache prime, once for the rest
 
 	// The cloning resource clients will forward actual calls onto the main resource service client.
 	// Here we are configuring the service mock to return our singular resource always.

internal/controller/runner.go

@@ -9,9 +9,10 @@ import (
 	"fmt"
 	"time"
 
-	"github.com/hashicorp/go-hclog"
 	"golang.org/x/sync/errgroup"
 
+	"github.com/hashicorp/go-hclog"
+
 	"github.com/hashicorp/consul/agent/consul/controller/queue"
 	"github.com/hashicorp/consul/internal/controller/cache"
 	"github.com/hashicorp/consul/internal/protoutil"
@@ -93,6 +94,32 @@ func (c *controllerRunner) run(ctx context.Context) error {
 		defer c.ctrl.stopCb(ctx, c.runtime(c.logger))
 	}
 
+	// Before we launch the reconciler or the dependency mappers, ensure the
+	// cache is properly primed to avoid errant reconciles.
+	//
+	// Without doing this the cache is unsafe for general use without causing
+	// reconcile regressions in certain cases.
+	{
+		c.logger.Debug("priming caches")
+		primeGroup, primeGroupCtx := errgroup.WithContext(ctx)
+		// Managed Type Events
+		primeGroup.Go(func() error {
+			return c.primeCache(primeGroupCtx, c.ctrl.managedTypeWatch.watchedType)
+		})
+		for _, w := range c.ctrl.watches {
+			watcher := w
+			// Watched Type Events
+			primeGroup.Go(func() error {
+				return c.primeCache(primeGroupCtx, watcher.watchedType)
+			})
+		}
+
+		if err := primeGroup.Wait(); err != nil {
+			return err
+		}
+		c.logger.Debug("priming caches complete")
+	}
+
 	group, groupCtx := errgroup.WithContext(ctx)
 	recQueue := runQueue[Request](groupCtx, c.ctrl)
 
@@ -153,6 +180,41 @@ func runQueue[T queue.ItemType](ctx context.Context, ctrl *Controller) queue.Wor
 	return queue.RunWorkQueue[T](ctx, base, max)
 }
 
+func (c *controllerRunner) primeCache(ctx context.Context, typ *pbresource.Type) error {
+	wl, err := c.watchClient.WatchList(ctx, &pbresource.WatchListRequest{
+		Type: typ,
+		Tenancy: &pbresource.Tenancy{
+			Partition: storage.Wildcard,
+			Namespace: storage.Wildcard,
+		},
+	})
+	if err != nil {
+		c.logger.Error("failed to create cache priming watch", "error", err)
+		return err
+	}
+
+	for {
+		event, err := wl.Recv()
+		if err != nil {
+			c.logger.Warn("error received from cache priming watch", "error", err)
+			return err
+		}
+
+		switch {
+		case event.GetUpsert() != nil:
+			c.cache.Insert(event.GetUpsert().Resource)
+		case event.GetDelete() != nil:
+			c.cache.Delete(event.GetDelete().Resource)
+		case event.GetEndOfSnapshot() != nil:
+			// This concludes the initial snapshot. The cache is primed.
+			return nil
+		default:
+			c.logger.Warn("skipping unexpected event type", "type", hclog.Fmt("%T", event.GetEvent()))
+			continue
+		}
+	}
+}
+
 func (c *controllerRunner) watch(ctx context.Context, typ *pbresource.Type, add func(*pbresource.Resource)) error {
 	wl, err := c.watchClient.WatchList(ctx, &pbresource.WatchListRequest{
 		Type: typ,
@@ -177,11 +239,19 @@ func (c *controllerRunner) watch(ctx context.Context, typ *pbresource.Type, add
 		// to ensure that any mapper/reconciliation queue deduping wont
 		// hide events from being observed and updating the cache state.
 		// Therefore we should do this before any queueing.
-		switch event.Operation {
+		var resource *pbresource.Resource
-		case pbresource.WatchEvent_OPERATION_UPSERT:
+		switch {
-			c.cache.Insert(event.Resource)
+		case event.GetUpsert() != nil:
-		case pbresource.WatchEvent_OPERATION_DELETE:
+			resource = event.GetUpsert().GetResource()
-			c.cache.Delete(event.Resource)
+			c.cache.Insert(resource)
+		case event.GetDelete() != nil:
+			resource = event.GetDelete().GetResource()
+			c.cache.Delete(resource)
+		case event.GetEndOfSnapshot() != nil:
+			continue // ignore
+		default:
+			c.logger.Warn("skipping unexpected event type", "type", hclog.Fmt("%T", event.GetEvent()))
+			continue
 		}
 
 		// Before adding the resource into the queue we must clone it.
@@ -192,7 +262,7 @@ func (c *controllerRunner) watch(ctx context.Context, typ *pbresource.Type, add
 		// mutation of data held by the cache (and presumably by the resource
 		// service assuming that the regular client we were given is the inmem
 		// variant)
-		add(protoutil.Clone(event.Resource))
+		add(protoutil.Clone(resource))
 	}
 }
 

internal/storage/conformance/conformance.go

@@ -29,6 +29,15 @@ type TestOptions struct {
 	// SupportsStronglyConsistentList indicates whether the given storage backend
 	// supports strongly consistent list operations.
 	SupportsStronglyConsistentList bool
+
+	// IgnoreWatchListSnapshotOperations indicates whether a given storage
+	// backend is expected to be consistent enough with reads to emit
+	// WatchEvent_Upsert after the initial sync.
+	//
+	// For instance, a replicated copy of the state store will have stale data
+	// and may return an initial snapshot of nothing, and follow it up by an
+	// upsert.
+	IgnoreWatchListSnapshotOperations bool
 }
 
 // Test runs a suite of tests against a storage.Backend implementation to check
@@ -416,15 +425,22 @@ func testListWatch(t *testing.T, opts TestOptions) {
 				require.NoError(t, err)
 				t.Cleanup(watch.Close)
 
-				for i := 0; i < len(tc.results); i++ {
+				expectNum := len(tc.results) + 1
+				for i := 0; i < expectNum; i++ {
 					ctx, cancel := context.WithTimeout(ctx, 5*time.Second)
 					t.Cleanup(cancel)
 
 					event, err := watch.Next(ctx)
 					require.NoError(t, err)
 
-					require.Equal(t, pbresource.WatchEvent_OPERATION_UPSERT, event.Operation)
+					if opts.IgnoreWatchListSnapshotOperations && event.GetEndOfSnapshot() != nil {
-					prototest.AssertContainsElement(t, tc.results, event.Resource, ignoreVersion)
+						continue // ignore
+					} else if !opts.IgnoreWatchListSnapshotOperations && i == expectNum-1 {
+						require.NotNil(t, event.GetEndOfSnapshot(), "expected EndOfSnapshot got %T", event.GetEvent())
+						continue
+					}
+					require.NotNil(t, event.GetUpsert(), "index=%d", i)
+					prototest.AssertContainsElement(t, tc.results, event.GetUpsert().Resource, ignoreVersion)
 				}
 			})
 
@@ -436,6 +452,15 @@ func testListWatch(t *testing.T, opts TestOptions) {
 				require.NoError(t, err)
 				t.Cleanup(watch.Close)
 
+				{ // read snapshot end
+					ctx, cancel := context.WithTimeout(ctx, 5*time.Second)
+					t.Cleanup(cancel)
+					event, err := watch.Next(ctx)
+					require.NoError(t, err)
+
+					require.NotNil(t, event.GetEndOfSnapshot())
+				}
+
 				// Write the seed data after the watch has been established.
 				for _, r := range seedData {
 					_, err := backend.WriteCAS(ctx, r)
@@ -449,13 +474,13 @@ func testListWatch(t *testing.T, opts TestOptions) {
 					event, err := watch.Next(ctx)
 					require.NoError(t, err)
 
-					require.Equal(t, pbresource.WatchEvent_OPERATION_UPSERT, event.Operation)
+					require.NotNil(t, event.GetUpsert())
-					prototest.AssertContainsElement(t, tc.results, event.Resource, ignoreVersion)
+					prototest.AssertContainsElement(t, tc.results, event.GetUpsert().Resource, ignoreVersion)
 
 					// Check that Read implements "monotonic reads" with Watch.
-					readRes, err := backend.Read(ctx, storage.EventualConsistency, event.Resource.Id)
+					readRes, err := backend.Read(ctx, storage.EventualConsistency, event.GetUpsert().Resource.Id)
 					require.NoError(t, err)
-					prototest.AssertDeepEqual(t, event.Resource, readRes)
+					prototest.AssertDeepEqual(t, event.GetUpsert().Resource, readRes)
 				}
 
 				// Delete a random resource to check we get an event.
@@ -469,8 +494,8 @@ func testListWatch(t *testing.T, opts TestOptions) {
 				event, err := watch.Next(ctx)
 				require.NoError(t, err)
 
-				require.Equal(t, pbresource.WatchEvent_OPERATION_DELETE, event.Operation)
+				require.NotNil(t, event.GetDelete())
-				prototest.AssertDeepEqual(t, del, event.Resource)
+				prototest.AssertDeepEqual(t, del, event.GetDelete().Resource)
 
 				// Check that Read implements "monotonic reads" with Watch.
 				_, err = backend.Read(ctx, storage.EventualConsistency, del.Id)

internal/storage/inmem/snapshot_test.go

@@ -70,8 +70,14 @@ func TestSnapshotRestore(t *testing.T) {
 	t.Cleanup(cancel)
 
 	// Expect the initial state on the watch.
-	_, err = watch.Next(ctx)
+	got, err := watch.Next(ctx)
 	require.NoError(t, err)
+	require.NotNil(t, got.GetUpsert())
+
+	// expect to get snapshot end op
+	got, err = watch.Next(ctx)
+	require.NoError(t, err)
+	require.NotNil(t, got.GetEndOfSnapshot())
 
 	restore, err := newStore.Restore()
 	require.NoError(t, err)

internal/storage/inmem/store.go

@@ -141,7 +141,13 @@ func (s *Store) WriteCAS(res *pbresource.Resource, vsn string) error {
 	}
 	tx.Commit()
 
-	s.publishEvent(idx, pbresource.WatchEvent_OPERATION_UPSERT, res)
+	s.publishEvent(idx, &pbresource.WatchEvent{
+		Event: &pbresource.WatchEvent_Upsert_{
+			Upsert: &pbresource.WatchEvent_Upsert{
+				Resource: res,
+			},
+		},
+	})
 
 	return nil
 }
@@ -188,7 +194,13 @@ func (s *Store) DeleteCAS(id *pbresource.ID, vsn string) error {
 	}
 	tx.Commit()
 
-	s.publishEvent(idx, pbresource.WatchEvent_OPERATION_DELETE, res)
+	s.publishEvent(idx, &pbresource.WatchEvent{
+		Event: &pbresource.WatchEvent_Delete_{
+			Delete: &pbresource.WatchEvent_Delete{
+				Resource: res,
+			},
+		},
+	})
 
 	return nil
 }

internal/storage/inmem/watch.go

@@ -36,7 +36,20 @@ func (w *Watch) Next(ctx context.Context) (*pbresource.WatchEvent, error) {
 		}
 
 		event := e.Payload.(eventPayload).event
-		if w.query.matches(event.Resource) {
+
+		var resource *pbresource.Resource
+		switch {
+		case event.GetUpsert() != nil:
+			resource = event.GetUpsert().GetResource()
+		case event.GetDelete() != nil:
+			resource = event.GetDelete().GetResource()
+		case event.GetEndOfSnapshot() != nil:
+			return event, nil
+		default:
+			return nil, fmt.Errorf("unexpected resource event type: %T", event.GetEvent())
+		}
+
+		if w.query.matches(resource) {
 			return event, nil
 		}
 	}
@@ -108,7 +121,8 @@ type eventPayload struct {
 func (p eventPayload) Subject() stream.Subject { return p.subject }
 
 // These methods are required by the stream.Payload interface, but we don't use them.
-func (eventPayload) HasReadPermission(acl.Authorizer) bool         { return false }
+func (eventPayload) HasReadPermission(acl.Authorizer) bool { return false }
+
 func (eventPayload) ToSubscriptionEvent(uint64) *pbsubscribe.Event { return nil }
 
 type wildcardSubject struct {
@@ -136,10 +150,17 @@ func (s tenancySubject) String() string {
 }
 
 // publishEvent sends the event to the relevant Watches.
-func (s *Store) publishEvent(idx uint64, op pbresource.WatchEvent_Operation, res *pbresource.Resource) {
+func (s *Store) publishEvent(idx uint64, event *pbresource.WatchEvent) {
-	id := res.Id
+	var id *pbresource.ID
+	switch {
+	case event.GetUpsert() != nil:
+		id = event.GetUpsert().GetResource().GetId()
+	case event.GetDelete() != nil:
+		id = event.GetDelete().GetResource().GetId()
+	default:
+		panic(fmt.Sprintf("(*Store).publishEvent cannot handle events of type %T", event.GetEvent()))
+	}
 	resourceType := storage.UnversionedTypeFrom(id.Type)
-	event := &pbresource.WatchEvent{Operation: op, Resource: res}
 
 	// We publish two copies of the event: one to the tenancy-specific subject and
 	// another to a wildcard subject. Ideally, we'd be able to put the type in the
@@ -200,20 +221,32 @@ func (s *Store) watchSnapshot(req stream.SubscribeRequest, snap stream.SnapshotA
 		return 0, nil
 	}
 
-	events := make([]stream.Event, len(results))
+	events := make([]stream.Event, 0, len(results)+1)
-	for i, r := range results {
+	addEvent := func(event *pbresource.WatchEvent) {
-		events[i] = stream.Event{
+		events = append(events, stream.Event{
 			Topic: eventTopic,
 			Index: idx,
 			Payload: eventPayload{
 				subject: req.Subject,
-				event: &pbresource.WatchEvent{
+				event:   event,
-					Operation: pbresource.WatchEvent_OPERATION_UPSERT,
+			},
-					Resource:  r,
+		})
+	}
+
+	for _, r := range results {
+		addEvent(&pbresource.WatchEvent{
+			Event: &pbresource.WatchEvent_Upsert_{
+				Upsert: &pbresource.WatchEvent_Upsert{
+					Resource: r,
 				},
 			},
-		}
+		})
 	}
+	addEvent(&pbresource.WatchEvent{
+		Event: &pbresource.WatchEvent_EndOfSnapshot_{
+			EndOfSnapshot: &pbresource.WatchEvent_EndOfSnapshot{},
+		},
+	})
 	snap.Append(events)
 
 	return idx, nil

internal/storage/raft/conformance_test.go

@@ -42,7 +42,8 @@ func TestBackend_Conformance(t *testing.T) {
 				_, follower := newRaftCluster(t)
 				return follower
 			},
-			SupportsStronglyConsistentList: true,
+			SupportsStronglyConsistentList:    true,
+			IgnoreWatchListSnapshotOperations: true,
 		})
 	})
 }

proto-public/pbresource/resource.pb.binary.go

@@ -227,6 +227,36 @@ func (msg *WatchEvent) UnmarshalBinary(b []byte) error {
 	return proto.Unmarshal(b, msg)
 }
 
+// MarshalBinary implements encoding.BinaryMarshaler
+func (msg *WatchEvent_Upsert) MarshalBinary() ([]byte, error) {
+	return proto.Marshal(msg)
+}
+
+// UnmarshalBinary implements encoding.BinaryUnmarshaler
+func (msg *WatchEvent_Upsert) UnmarshalBinary(b []byte) error {
+	return proto.Unmarshal(b, msg)
+}
+
+// MarshalBinary implements encoding.BinaryMarshaler
+func (msg *WatchEvent_Delete) MarshalBinary() ([]byte, error) {
+	return proto.Marshal(msg)
+}
+
+// UnmarshalBinary implements encoding.BinaryUnmarshaler
+func (msg *WatchEvent_Delete) UnmarshalBinary(b []byte) error {
+	return proto.Unmarshal(b, msg)
+}
+
+// MarshalBinary implements encoding.BinaryMarshaler
+func (msg *WatchEvent_EndOfSnapshot) MarshalBinary() ([]byte, error) {
+	return proto.Marshal(msg)
+}
+
+// UnmarshalBinary implements encoding.BinaryUnmarshaler
+func (msg *WatchEvent_EndOfSnapshot) UnmarshalBinary(b []byte) error {
+	return proto.Unmarshal(b, msg)
+}
+
 // MarshalBinary implements encoding.BinaryMarshaler
 func (msg *MutateAndValidateRequest) MarshalBinary() ([]byte, error) {
 	return proto.Marshal(msg)

proto-public/pbresource/resource.pb.go

@@ -83,62 +83,6 @@ func (Condition_State) EnumDescriptor() ([]byte, []int) {
 	return file_pbresource_resource_proto_rawDescGZIP(), []int{5, 0}
 }
 
-// Operation describes the type of event.
-type WatchEvent_Operation int32
-
-const (
-	// OPERATION_UNSPECIFIED is the default/zero value. You should not see it
-	// in practice.
-	WatchEvent_OPERATION_UNSPECIFIED WatchEvent_Operation = 0
-	// OPERATION_UPSERT indicates that the resource was written (i.e. created or
-	// updated). All events from the initial state-of-the-world will be upsert
-	// events.
-	WatchEvent_OPERATION_UPSERT WatchEvent_Operation = 1
-	// OPERATION_DELETED indicates that the resource was deleted.
-	WatchEvent_OPERATION_DELETE WatchEvent_Operation = 2
-)
-
-// Enum value maps for WatchEvent_Operation.
-var (
-	WatchEvent_Operation_name = map[int32]string{
-		0: "OPERATION_UNSPECIFIED",
-		1: "OPERATION_UPSERT",
-		2: "OPERATION_DELETE",
-	}
-	WatchEvent_Operation_value = map[string]int32{
-		"OPERATION_UNSPECIFIED": 0,
-		"OPERATION_UPSERT":      1,
-		"OPERATION_DELETE":      2,
-	}
-)
-
-func (x WatchEvent_Operation) Enum() *WatchEvent_Operation {
-	p := new(WatchEvent_Operation)
-	*p = x
-	return p
-}
-
-func (x WatchEvent_Operation) String() string {
-	return protoimpl.X.EnumStringOf(x.Descriptor(), protoreflect.EnumNumber(x))
-}
-
-func (WatchEvent_Operation) Descriptor() protoreflect.EnumDescriptor {
-	return file_pbresource_resource_proto_enumTypes[1].Descriptor()
-}
-
-func (WatchEvent_Operation) Type() protoreflect.EnumType {
-	return &file_pbresource_resource_proto_enumTypes[1]
-}
-
-func (x WatchEvent_Operation) Number() protoreflect.EnumNumber {
-	return protoreflect.EnumNumber(x)
-}
-
-// Deprecated: Use WatchEvent_Operation.Descriptor instead.
-func (WatchEvent_Operation) EnumDescriptor() ([]byte, []int) {
-	return file_pbresource_resource_proto_rawDescGZIP(), []int{21, 0}
-}
-
 // Type describes a resource's type. It follows the GVK (Group Version Kind)
 // [pattern](https://book.kubebuilder.io/cronjob-tutorial/gvks.html) established
 // by Kubernetes.
@@ -1490,10 +1434,12 @@ type WatchEvent struct {
 	sizeCache     protoimpl.SizeCache
 	unknownFields protoimpl.UnknownFields
 
-	// Operation describes the type of event.
+	// Types that are assignable to Event:
-	Operation WatchEvent_Operation `protobuf:"varint,1,opt,name=operation,proto3,enum=hashicorp.consul.resource.WatchEvent_Operation" json:"operation,omitempty"`
+	//
-	// Resource the event relates to.
+	//	*WatchEvent_Upsert_
-	Resource *Resource `protobuf:"bytes,2,opt,name=resource,proto3" json:"resource,omitempty"`
+	//	*WatchEvent_Delete_
+	//	*WatchEvent_EndOfSnapshot_
+	Event isWatchEvent_Event `protobuf_oneof:"event"`
 }
 
 func (x *WatchEvent) Reset() {
@@ -1528,20 +1474,56 @@ func (*WatchEvent) Descriptor() ([]byte, []int) {
 	return file_pbresource_resource_proto_rawDescGZIP(), []int{21}
 }
 
-func (x *WatchEvent) GetOperation() WatchEvent_Operation {
+func (m *WatchEvent) GetEvent() isWatchEvent_Event {
-	if x != nil {
+	if m != nil {
-		return x.Operation
+		return m.Event
-	}
-	return WatchEvent_OPERATION_UNSPECIFIED
-}
-
-func (x *WatchEvent) GetResource() *Resource {
-	if x != nil {
-		return x.Resource
 	}
 	return nil
 }
 
+func (x *WatchEvent) GetUpsert() *WatchEvent_Upsert {
+	if x, ok := x.GetEvent().(*WatchEvent_Upsert_); ok {
+		return x.Upsert
+	}
+	return nil
+}
+
+func (x *WatchEvent) GetDelete() *WatchEvent_Delete {
+	if x, ok := x.GetEvent().(*WatchEvent_Delete_); ok {
+		return x.Delete
+	}
+	return nil
+}
+
+func (x *WatchEvent) GetEndOfSnapshot() *WatchEvent_EndOfSnapshot {
+	if x, ok := x.GetEvent().(*WatchEvent_EndOfSnapshot_); ok {
+		return x.EndOfSnapshot
+	}
+	return nil
+}
+
+type isWatchEvent_Event interface {
+	isWatchEvent_Event()
+}
+
+type WatchEvent_Upsert_ struct {
+	Upsert *WatchEvent_Upsert `protobuf:"bytes,1,opt,name=upsert,proto3,oneof"`
+}
+
+type WatchEvent_Delete_ struct {
+	Delete *WatchEvent_Delete `protobuf:"bytes,2,opt,name=delete,proto3,oneof"`
+}
+
+type WatchEvent_EndOfSnapshot_ struct {
+	EndOfSnapshot *WatchEvent_EndOfSnapshot `protobuf:"bytes,3,opt,name=end_of_snapshot,json=endOfSnapshot,proto3,oneof"`
+}
+
+func (*WatchEvent_Upsert_) isWatchEvent_Event() {}
+
+func (*WatchEvent_Delete_) isWatchEvent_Event() {}
+
+func (*WatchEvent_EndOfSnapshot_) isWatchEvent_Event() {}
+
 // MutateAndValidateRequest contains the parameters to the MutateAndValidate endpoint.
 type MutateAndValidateRequest struct {
 	state         protoimpl.MessageState
@@ -1638,6 +1620,144 @@ func (x *MutateAndValidateResponse) GetResource() *Resource {
 	return nil
 }
 
+// Upsert indicates that the resource was written (i.e. created or
+// updated). All events from the initial state-of-the-world will be upsert
+// events.
+type WatchEvent_Upsert struct {
+	state         protoimpl.MessageState
+	sizeCache     protoimpl.SizeCache
+	unknownFields protoimpl.UnknownFields
+
+	Resource *Resource `protobuf:"bytes,1,opt,name=resource,proto3" json:"resource,omitempty"`
+}
+
+func (x *WatchEvent_Upsert) Reset() {
+	*x = WatchEvent_Upsert{}
+	if protoimpl.UnsafeEnabled {
+		mi := &file_pbresource_resource_proto_msgTypes[26]
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		ms.StoreMessageInfo(mi)
+	}
+}
+
+func (x *WatchEvent_Upsert) String() string {
+	return protoimpl.X.MessageStringOf(x)
+}
+
+func (*WatchEvent_Upsert) ProtoMessage() {}
+
+func (x *WatchEvent_Upsert) ProtoReflect() protoreflect.Message {
+	mi := &file_pbresource_resource_proto_msgTypes[26]
+	if protoimpl.UnsafeEnabled && x != nil {
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		if ms.LoadMessageInfo() == nil {
+			ms.StoreMessageInfo(mi)
+		}
+		return ms
+	}
+	return mi.MessageOf(x)
+}
+
+// Deprecated: Use WatchEvent_Upsert.ProtoReflect.Descriptor instead.
+func (*WatchEvent_Upsert) Descriptor() ([]byte, []int) {
+	return file_pbresource_resource_proto_rawDescGZIP(), []int{21, 0}
+}
+
+func (x *WatchEvent_Upsert) GetResource() *Resource {
+	if x != nil {
+		return x.Resource
+	}
+	return nil
+}
+
+// Delete indicates that the resource was deleted.
+type WatchEvent_Delete struct {
+	state         protoimpl.MessageState
+	sizeCache     protoimpl.SizeCache
+	unknownFields protoimpl.UnknownFields
+
+	Resource *Resource `protobuf:"bytes,1,opt,name=resource,proto3" json:"resource,omitempty"`
+}
+
+func (x *WatchEvent_Delete) Reset() {
+	*x = WatchEvent_Delete{}
+	if protoimpl.UnsafeEnabled {
+		mi := &file_pbresource_resource_proto_msgTypes[27]
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		ms.StoreMessageInfo(mi)
+	}
+}
+
+func (x *WatchEvent_Delete) String() string {
+	return protoimpl.X.MessageStringOf(x)
+}
+
+func (*WatchEvent_Delete) ProtoMessage() {}
+
+func (x *WatchEvent_Delete) ProtoReflect() protoreflect.Message {
+	mi := &file_pbresource_resource_proto_msgTypes[27]
+	if protoimpl.UnsafeEnabled && x != nil {
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		if ms.LoadMessageInfo() == nil {
+			ms.StoreMessageInfo(mi)
+		}
+		return ms
+	}
+	return mi.MessageOf(x)
+}
+
+// Deprecated: Use WatchEvent_Delete.ProtoReflect.Descriptor instead.
+func (*WatchEvent_Delete) Descriptor() ([]byte, []int) {
+	return file_pbresource_resource_proto_rawDescGZIP(), []int{21, 1}
+}
+
+func (x *WatchEvent_Delete) GetResource() *Resource {
+	if x != nil {
+		return x.Resource
+	}
+	return nil
+}
+
+// EndOfSnapshot is sent to indicate that the initial snapshot events have
+// been sent and future events will modify that set.
+type WatchEvent_EndOfSnapshot struct {
+	state         protoimpl.MessageState
+	sizeCache     protoimpl.SizeCache
+	unknownFields protoimpl.UnknownFields
+}
+
+func (x *WatchEvent_EndOfSnapshot) Reset() {
+	*x = WatchEvent_EndOfSnapshot{}
+	if protoimpl.UnsafeEnabled {
+		mi := &file_pbresource_resource_proto_msgTypes[28]
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		ms.StoreMessageInfo(mi)
+	}
+}
+
+func (x *WatchEvent_EndOfSnapshot) String() string {
+	return protoimpl.X.MessageStringOf(x)
+}
+
+func (*WatchEvent_EndOfSnapshot) ProtoMessage() {}
+
+func (x *WatchEvent_EndOfSnapshot) ProtoReflect() protoreflect.Message {
+	mi := &file_pbresource_resource_proto_msgTypes[28]
+	if protoimpl.UnsafeEnabled && x != nil {
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		if ms.LoadMessageInfo() == nil {
+			ms.StoreMessageInfo(mi)
+		}
+		return ms
+	}
+	return mi.MessageOf(x)
+}
+
+// Deprecated: Use WatchEvent_EndOfSnapshot.ProtoReflect.Descriptor instead.
+func (*WatchEvent_EndOfSnapshot) Descriptor() ([]byte, []int) {
+	return file_pbresource_resource_proto_rawDescGZIP(), []int{21, 2}
+}
+
 var File_pbresource_resource_proto protoreflect.FileDescriptor
 
 var file_pbresource_resource_proto_rawDesc = []byte{
@@ -1825,106 +1945,118 @@ var file_pbresource_resource_proto_rawDesc = []byte{
 	0x72, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x2e, 0x54, 0x65, 0x6e, 0x61, 0x6e, 0x63, 0x79,
 	0x52, 0x07, 0x74, 0x65, 0x6e, 0x61, 0x6e, 0x63, 0x79, 0x12, 0x1f, 0x0a, 0x0b, 0x6e, 0x61, 0x6d,
 	0x65, 0x5f, 0x70, 0x72, 0x65, 0x66, 0x69, 0x78, 0x18, 0x03, 0x20, 0x01, 0x28, 0x09, 0x52, 0x0a,
-	0x6e, 0x61, 0x6d, 0x65, 0x50, 0x72, 0x65, 0x66, 0x69, 0x78, 0x22, 0xf0, 0x01, 0x0a, 0x0a, 0x57,
+	0x6e, 0x61, 0x6d, 0x65, 0x50, 0x72, 0x65, 0x66, 0x69, 0x78, 0x22, 0xab, 0x03, 0x0a, 0x0a, 0x57,
-	0x61, 0x74, 0x63, 0x68, 0x45, 0x76, 0x65, 0x6e, 0x74, 0x12, 0x4d, 0x0a, 0x09, 0x6f, 0x70, 0x65,
+	0x61, 0x74, 0x63, 0x68, 0x45, 0x76, 0x65, 0x6e, 0x74, 0x12, 0x46, 0x0a, 0x06, 0x75, 0x70, 0x73,
-	0x72, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0e, 0x32, 0x2f, 0x2e, 0x68,
+	0x65, 0x72, 0x74, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x2c, 0x2e, 0x68, 0x61, 0x73, 0x68,
-	0x61, 0x73, 0x68, 0x69, 0x63, 0x6f, 0x72, 0x70, 0x2e, 0x63, 0x6f, 0x6e, 0x73, 0x75, 0x6c, 0x2e,
-	0x72, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x2e, 0x57, 0x61, 0x74, 0x63, 0x68, 0x45, 0x76,
-	0x65, 0x6e, 0x74, 0x2e, 0x4f, 0x70, 0x65, 0x72, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x52, 0x09, 0x6f,
-	0x70, 0x65, 0x72, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x12, 0x3f, 0x0a, 0x08, 0x72, 0x65, 0x73, 0x6f,
-	0x75, 0x72, 0x63, 0x65, 0x18, 0x02, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x23, 0x2e, 0x68, 0x61, 0x73,
-	0x68, 0x69, 0x63, 0x6f, 0x72, 0x70, 0x2e, 0x63, 0x6f, 0x6e, 0x73, 0x75, 0x6c, 0x2e, 0x72, 0x65,
-	0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x2e, 0x52, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x52,
-	0x08, 0x72, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x22, 0x52, 0x0a, 0x09, 0x4f, 0x70, 0x65,
-	0x72, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x12, 0x19, 0x0a, 0x15, 0x4f, 0x50, 0x45, 0x52, 0x41, 0x54,
-	0x49, 0x4f, 0x4e, 0x5f, 0x55, 0x4e, 0x53, 0x50, 0x45, 0x43, 0x49, 0x46, 0x49, 0x45, 0x44, 0x10,
-	0x00, 0x12, 0x14, 0x0a, 0x10, 0x4f, 0x50, 0x45, 0x52, 0x41, 0x54, 0x49, 0x4f, 0x4e, 0x5f, 0x55,
-	0x50, 0x53, 0x45, 0x52, 0x54, 0x10, 0x01, 0x12, 0x14, 0x0a, 0x10, 0x4f, 0x50, 0x45, 0x52, 0x41,
-	0x54, 0x49, 0x4f, 0x4e, 0x5f, 0x44, 0x45, 0x4c, 0x45, 0x54, 0x45, 0x10, 0x02, 0x22, 0x5b, 0x0a,
-	0x18, 0x4d, 0x75, 0x74, 0x61, 0x74, 0x65, 0x41, 0x6e, 0x64, 0x56, 0x61, 0x6c, 0x69, 0x64, 0x61,
-	0x74, 0x65, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x12, 0x3f, 0x0a, 0x08, 0x72, 0x65, 0x73,
-	0x6f, 0x75, 0x72, 0x63, 0x65, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x23, 0x2e, 0x68, 0x61,
-	0x73, 0x68, 0x69, 0x63, 0x6f, 0x72, 0x70, 0x2e, 0x63, 0x6f, 0x6e, 0x73, 0x75, 0x6c, 0x2e, 0x72,
-	0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x2e, 0x52, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65,
-	0x52, 0x08, 0x72, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x22, 0x5c, 0x0a, 0x19, 0x4d, 0x75,
-	0x74, 0x61, 0x74, 0x65, 0x41, 0x6e, 0x64, 0x56, 0x61, 0x6c, 0x69, 0x64, 0x61, 0x74, 0x65, 0x52,
-	0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x12, 0x3f, 0x0a, 0x08, 0x72, 0x65, 0x73, 0x6f, 0x75,
-	0x72, 0x63, 0x65, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x23, 0x2e, 0x68, 0x61, 0x73, 0x68,
 	0x69, 0x63, 0x6f, 0x72, 0x70, 0x2e, 0x63, 0x6f, 0x6e, 0x73, 0x75, 0x6c, 0x2e, 0x72, 0x65, 0x73,
-	0x6f, 0x75, 0x72, 0x63, 0x65, 0x2e, 0x52, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x52, 0x08,
+	0x6f, 0x75, 0x72, 0x63, 0x65, 0x2e, 0x57, 0x61, 0x74, 0x63, 0x68, 0x45, 0x76, 0x65, 0x6e, 0x74,
-	0x72, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x32, 0x8e, 0x07, 0x0a, 0x0f, 0x52, 0x65, 0x73,
+	0x2e, 0x55, 0x70, 0x73, 0x65, 0x72, 0x74, 0x48, 0x00, 0x52, 0x06, 0x75, 0x70, 0x73, 0x65, 0x72,
-	0x6f, 0x75, 0x72, 0x63, 0x65, 0x53, 0x65, 0x72, 0x76, 0x69, 0x63, 0x65, 0x12, 0x61, 0x0a, 0x04,
+	0x74, 0x12, 0x46, 0x0a, 0x06, 0x64, 0x65, 0x6c, 0x65, 0x74, 0x65, 0x18, 0x02, 0x20, 0x01, 0x28,
-	0x52, 0x65, 0x61, 0x64, 0x12, 0x26, 0x2e, 0x68, 0x61, 0x73, 0x68, 0x69, 0x63, 0x6f, 0x72, 0x70,
+	0x0b, 0x32, 0x2c, 0x2e, 0x68, 0x61, 0x73, 0x68, 0x69, 0x63, 0x6f, 0x72, 0x70, 0x2e, 0x63, 0x6f,
+	0x6e, 0x73, 0x75, 0x6c, 0x2e, 0x72, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x2e, 0x57, 0x61,
+	0x74, 0x63, 0x68, 0x45, 0x76, 0x65, 0x6e, 0x74, 0x2e, 0x44, 0x65, 0x6c, 0x65, 0x74, 0x65, 0x48,
+	0x00, 0x52, 0x06, 0x64, 0x65, 0x6c, 0x65, 0x74, 0x65, 0x12, 0x5d, 0x0a, 0x0f, 0x65, 0x6e, 0x64,
+	0x5f, 0x6f, 0x66, 0x5f, 0x73, 0x6e, 0x61, 0x70, 0x73, 0x68, 0x6f, 0x74, 0x18, 0x03, 0x20, 0x01,
+	0x28, 0x0b, 0x32, 0x33, 0x2e, 0x68, 0x61, 0x73, 0x68, 0x69, 0x63, 0x6f, 0x72, 0x70, 0x2e, 0x63,
+	0x6f, 0x6e, 0x73, 0x75, 0x6c, 0x2e, 0x72, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x2e, 0x57,
+	0x61, 0x74, 0x63, 0x68, 0x45, 0x76, 0x65, 0x6e, 0x74, 0x2e, 0x45, 0x6e, 0x64, 0x4f, 0x66, 0x53,
+	0x6e, 0x61, 0x70, 0x73, 0x68, 0x6f, 0x74, 0x48, 0x00, 0x52, 0x0d, 0x65, 0x6e, 0x64, 0x4f, 0x66,
+	0x53, 0x6e, 0x61, 0x70, 0x73, 0x68, 0x6f, 0x74, 0x1a, 0x49, 0x0a, 0x06, 0x55, 0x70, 0x73, 0x65,
+	0x72, 0x74, 0x12, 0x3f, 0x0a, 0x08, 0x72, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x18, 0x01,
+	0x20, 0x01, 0x28, 0x0b, 0x32, 0x23, 0x2e, 0x68, 0x61, 0x73, 0x68, 0x69, 0x63, 0x6f, 0x72, 0x70,
 	0x2e, 0x63, 0x6f, 0x6e, 0x73, 0x75, 0x6c, 0x2e, 0x72, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65,
-	0x2e, 0x52, 0x65, 0x61, 0x64, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x27, 0x2e, 0x68,
+	0x2e, 0x52, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x52, 0x08, 0x72, 0x65, 0x73, 0x6f, 0x75,
-	0x61, 0x73, 0x68, 0x69, 0x63, 0x6f, 0x72, 0x70, 0x2e, 0x63, 0x6f, 0x6e, 0x73, 0x75, 0x6c, 0x2e,
+	0x72, 0x63, 0x65, 0x1a, 0x49, 0x0a, 0x06, 0x44, 0x65, 0x6c, 0x65, 0x74, 0x65, 0x12, 0x3f, 0x0a,
-	0x72, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x2e, 0x52, 0x65, 0x61, 0x64, 0x52, 0x65, 0x73,
+	0x08, 0x72, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0b, 0x32,
-	0x70, 0x6f, 0x6e, 0x73, 0x65, 0x22, 0x08, 0xe2, 0x86, 0x04, 0x04, 0x08, 0x02, 0x10, 0x0b, 0x12,
+	0x23, 0x2e, 0x68, 0x61, 0x73, 0x68, 0x69, 0x63, 0x6f, 0x72, 0x70, 0x2e, 0x63, 0x6f, 0x6e, 0x73,
-	0x64, 0x0a, 0x05, 0x57, 0x72, 0x69, 0x74, 0x65, 0x12, 0x27, 0x2e, 0x68, 0x61, 0x73, 0x68, 0x69,
+	0x75, 0x6c, 0x2e, 0x72, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x2e, 0x52, 0x65, 0x73, 0x6f,
-	0x63, 0x6f, 0x72, 0x70, 0x2e, 0x63, 0x6f, 0x6e, 0x73, 0x75, 0x6c, 0x2e, 0x72, 0x65, 0x73, 0x6f,
+	0x75, 0x72, 0x63, 0x65, 0x52, 0x08, 0x72, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x1a, 0x0f,
-	0x75, 0x72, 0x63, 0x65, 0x2e, 0x57, 0x72, 0x69, 0x74, 0x65, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73,
+	0x0a, 0x0d, 0x45, 0x6e, 0x64, 0x4f, 0x66, 0x53, 0x6e, 0x61, 0x70, 0x73, 0x68, 0x6f, 0x74, 0x42,
-	0x74, 0x1a, 0x28, 0x2e, 0x68, 0x61, 0x73, 0x68, 0x69, 0x63, 0x6f, 0x72, 0x70, 0x2e, 0x63, 0x6f,
+	0x07, 0x0a, 0x05, 0x65, 0x76, 0x65, 0x6e, 0x74, 0x22, 0x5b, 0x0a, 0x18, 0x4d, 0x75, 0x74, 0x61,
-	0x6e, 0x73, 0x75, 0x6c, 0x2e, 0x72, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x2e, 0x57, 0x72,
+	0x74, 0x65, 0x41, 0x6e, 0x64, 0x56, 0x61, 0x6c, 0x69, 0x64, 0x61, 0x74, 0x65, 0x52, 0x65, 0x71,
-	0x69, 0x74, 0x65, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x22, 0x08, 0xe2, 0x86, 0x04,
+	0x75, 0x65, 0x73, 0x74, 0x12, 0x3f, 0x0a, 0x08, 0x72, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65,
-	0x04, 0x08, 0x03, 0x10, 0x0b, 0x12, 0x76, 0x0a, 0x0b, 0x57, 0x72, 0x69, 0x74, 0x65, 0x53, 0x74,
+	0x18, 0x01, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x23, 0x2e, 0x68, 0x61, 0x73, 0x68, 0x69, 0x63, 0x6f,
-	0x61, 0x74, 0x75, 0x73, 0x12, 0x2d, 0x2e, 0x68, 0x61, 0x73, 0x68, 0x69, 0x63, 0x6f, 0x72, 0x70,
+	0x72, 0x70, 0x2e, 0x63, 0x6f, 0x6e, 0x73, 0x75, 0x6c, 0x2e, 0x72, 0x65, 0x73, 0x6f, 0x75, 0x72,
+	0x63, 0x65, 0x2e, 0x52, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x52, 0x08, 0x72, 0x65, 0x73,
+	0x6f, 0x75, 0x72, 0x63, 0x65, 0x22, 0x5c, 0x0a, 0x19, 0x4d, 0x75, 0x74, 0x61, 0x74, 0x65, 0x41,
+	0x6e, 0x64, 0x56, 0x61, 0x6c, 0x69, 0x64, 0x61, 0x74, 0x65, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e,
+	0x73, 0x65, 0x12, 0x3f, 0x0a, 0x08, 0x72, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x18, 0x01,
+	0x20, 0x01, 0x28, 0x0b, 0x32, 0x23, 0x2e, 0x68, 0x61, 0x73, 0x68, 0x69, 0x63, 0x6f, 0x72, 0x70,
 	0x2e, 0x63, 0x6f, 0x6e, 0x73, 0x75, 0x6c, 0x2e, 0x72, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65,
-	0x2e, 0x57, 0x72, 0x69, 0x74, 0x65, 0x53, 0x74, 0x61, 0x74, 0x75, 0x73, 0x52, 0x65, 0x71, 0x75,
+	0x2e, 0x52, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x52, 0x08, 0x72, 0x65, 0x73, 0x6f, 0x75,
-	0x65, 0x73, 0x74, 0x1a, 0x2e, 0x2e, 0x68, 0x61, 0x73, 0x68, 0x69, 0x63, 0x6f, 0x72, 0x70, 0x2e,
+	0x72, 0x63, 0x65, 0x32, 0x8e, 0x07, 0x0a, 0x0f, 0x52, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65,
+	0x53, 0x65, 0x72, 0x76, 0x69, 0x63, 0x65, 0x12, 0x61, 0x0a, 0x04, 0x52, 0x65, 0x61, 0x64, 0x12,
+	0x26, 0x2e, 0x68, 0x61, 0x73, 0x68, 0x69, 0x63, 0x6f, 0x72, 0x70, 0x2e, 0x63, 0x6f, 0x6e, 0x73,
+	0x75, 0x6c, 0x2e, 0x72, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x2e, 0x52, 0x65, 0x61, 0x64,
+	0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x27, 0x2e, 0x68, 0x61, 0x73, 0x68, 0x69, 0x63,
+	0x6f, 0x72, 0x70, 0x2e, 0x63, 0x6f, 0x6e, 0x73, 0x75, 0x6c, 0x2e, 0x72, 0x65, 0x73, 0x6f, 0x75,
+	0x72, 0x63, 0x65, 0x2e, 0x52, 0x65, 0x61, 0x64, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65,
+	0x22, 0x08, 0xe2, 0x86, 0x04, 0x04, 0x08, 0x02, 0x10, 0x0b, 0x12, 0x64, 0x0a, 0x05, 0x57, 0x72,
+	0x69, 0x74, 0x65, 0x12, 0x27, 0x2e, 0x68, 0x61, 0x73, 0x68, 0x69, 0x63, 0x6f, 0x72, 0x70, 0x2e,
 	0x63, 0x6f, 0x6e, 0x73, 0x75, 0x6c, 0x2e, 0x72, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x2e,
-	0x57, 0x72, 0x69, 0x74, 0x65, 0x53, 0x74, 0x61, 0x74, 0x75, 0x73, 0x52, 0x65, 0x73, 0x70, 0x6f,
+	0x57, 0x72, 0x69, 0x74, 0x65, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x28, 0x2e, 0x68,
-	0x6e, 0x73, 0x65, 0x22, 0x08, 0xe2, 0x86, 0x04, 0x04, 0x08, 0x03, 0x10, 0x0b, 0x12, 0x61, 0x0a,
-	0x04, 0x4c, 0x69, 0x73, 0x74, 0x12, 0x26, 0x2e, 0x68, 0x61, 0x73, 0x68, 0x69, 0x63, 0x6f, 0x72,
-	0x70, 0x2e, 0x63, 0x6f, 0x6e, 0x73, 0x75, 0x6c, 0x2e, 0x72, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63,
-	0x65, 0x2e, 0x4c, 0x69, 0x73, 0x74, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x27, 0x2e,
-	0x68, 0x61, 0x73, 0x68, 0x69, 0x63, 0x6f, 0x72, 0x70, 0x2e, 0x63, 0x6f, 0x6e, 0x73, 0x75, 0x6c,
-	0x2e, 0x72, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x2e, 0x4c, 0x69, 0x73, 0x74, 0x52, 0x65,
-	0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x22, 0x08, 0xe2, 0x86, 0x04, 0x04, 0x08, 0x02, 0x10, 0x0b,
-	0x12, 0x76, 0x0a, 0x0b, 0x4c, 0x69, 0x73, 0x74, 0x42, 0x79, 0x4f, 0x77, 0x6e, 0x65, 0x72, 0x12,
-	0x2d, 0x2e, 0x68, 0x61, 0x73, 0x68, 0x69, 0x63, 0x6f, 0x72, 0x70, 0x2e, 0x63, 0x6f, 0x6e, 0x73,
-	0x75, 0x6c, 0x2e, 0x72, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x2e, 0x4c, 0x69, 0x73, 0x74,
-	0x42, 0x79, 0x4f, 0x77, 0x6e, 0x65, 0x72, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x2e,
-	0x2e, 0x68, 0x61, 0x73, 0x68, 0x69, 0x63, 0x6f, 0x72, 0x70, 0x2e, 0x63, 0x6f, 0x6e, 0x73, 0x75,
-	0x6c, 0x2e, 0x72, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x2e, 0x4c, 0x69, 0x73, 0x74, 0x42,
-	0x79, 0x4f, 0x77, 0x6e, 0x65, 0x72, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x22, 0x08,
-	0xe2, 0x86, 0x04, 0x04, 0x08, 0x02, 0x10, 0x0b, 0x12, 0x67, 0x0a, 0x06, 0x44, 0x65, 0x6c, 0x65,
-	0x74, 0x65, 0x12, 0x28, 0x2e, 0x68, 0x61, 0x73, 0x68, 0x69, 0x63, 0x6f, 0x72, 0x70, 0x2e, 0x63,
-	0x6f, 0x6e, 0x73, 0x75, 0x6c, 0x2e, 0x72, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x2e, 0x44,
-	0x65, 0x6c, 0x65, 0x74, 0x65, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x29, 0x2e, 0x68,
 	0x61, 0x73, 0x68, 0x69, 0x63, 0x6f, 0x72, 0x70, 0x2e, 0x63, 0x6f, 0x6e, 0x73, 0x75, 0x6c, 0x2e,
-	0x72, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x2e, 0x44, 0x65, 0x6c, 0x65, 0x74, 0x65, 0x52,
+	0x72, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x2e, 0x57, 0x72, 0x69, 0x74, 0x65, 0x52, 0x65,
-	0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x22, 0x08, 0xe2, 0x86, 0x04, 0x04, 0x08, 0x03, 0x10,
+	0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x22, 0x08, 0xe2, 0x86, 0x04, 0x04, 0x08, 0x03, 0x10, 0x0b,
-	0x0b, 0x12, 0x6b, 0x0a, 0x09, 0x57, 0x61, 0x74, 0x63, 0x68, 0x4c, 0x69, 0x73, 0x74, 0x12, 0x2b,
+	0x12, 0x76, 0x0a, 0x0b, 0x57, 0x72, 0x69, 0x74, 0x65, 0x53, 0x74, 0x61, 0x74, 0x75, 0x73, 0x12,
+	0x2d, 0x2e, 0x68, 0x61, 0x73, 0x68, 0x69, 0x63, 0x6f, 0x72, 0x70, 0x2e, 0x63, 0x6f, 0x6e, 0x73,
+	0x75, 0x6c, 0x2e, 0x72, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x2e, 0x57, 0x72, 0x69, 0x74,
+	0x65, 0x53, 0x74, 0x61, 0x74, 0x75, 0x73, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x2e,
 	0x2e, 0x68, 0x61, 0x73, 0x68, 0x69, 0x63, 0x6f, 0x72, 0x70, 0x2e, 0x63, 0x6f, 0x6e, 0x73, 0x75,
-	0x6c, 0x2e, 0x72, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x2e, 0x57, 0x61, 0x74, 0x63, 0x68,
+	0x6c, 0x2e, 0x72, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x2e, 0x57, 0x72, 0x69, 0x74, 0x65,
-	0x4c, 0x69, 0x73, 0x74, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x25, 0x2e, 0x68, 0x61,
+	0x53, 0x74, 0x61, 0x74, 0x75, 0x73, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x22, 0x08,
-	0x73, 0x68, 0x69, 0x63, 0x6f, 0x72, 0x70, 0x2e, 0x63, 0x6f, 0x6e, 0x73, 0x75, 0x6c, 0x2e, 0x72,
+	0xe2, 0x86, 0x04, 0x04, 0x08, 0x03, 0x10, 0x0b, 0x12, 0x61, 0x0a, 0x04, 0x4c, 0x69, 0x73, 0x74,
-	0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x2e, 0x57, 0x61, 0x74, 0x63, 0x68, 0x45, 0x76, 0x65,
+	0x12, 0x26, 0x2e, 0x68, 0x61, 0x73, 0x68, 0x69, 0x63, 0x6f, 0x72, 0x70, 0x2e, 0x63, 0x6f, 0x6e,
-	0x6e, 0x74, 0x22, 0x08, 0xe2, 0x86, 0x04, 0x04, 0x08, 0x02, 0x10, 0x0b, 0x30, 0x01, 0x12, 0x88,
+	0x73, 0x75, 0x6c, 0x2e, 0x72, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x2e, 0x4c, 0x69, 0x73,
-	0x01, 0x0a, 0x11, 0x4d, 0x75, 0x74, 0x61, 0x74, 0x65, 0x41, 0x6e, 0x64, 0x56, 0x61, 0x6c, 0x69,
+	0x74, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x27, 0x2e, 0x68, 0x61, 0x73, 0x68, 0x69,
-	0x64, 0x61, 0x74, 0x65, 0x12, 0x33, 0x2e, 0x68, 0x61, 0x73, 0x68, 0x69, 0x63, 0x6f, 0x72, 0x70,
+	0x63, 0x6f, 0x72, 0x70, 0x2e, 0x63, 0x6f, 0x6e, 0x73, 0x75, 0x6c, 0x2e, 0x72, 0x65, 0x73, 0x6f,
+	0x75, 0x72, 0x63, 0x65, 0x2e, 0x4c, 0x69, 0x73, 0x74, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73,
+	0x65, 0x22, 0x08, 0xe2, 0x86, 0x04, 0x04, 0x08, 0x02, 0x10, 0x0b, 0x12, 0x76, 0x0a, 0x0b, 0x4c,
+	0x69, 0x73, 0x74, 0x42, 0x79, 0x4f, 0x77, 0x6e, 0x65, 0x72, 0x12, 0x2d, 0x2e, 0x68, 0x61, 0x73,
+	0x68, 0x69, 0x63, 0x6f, 0x72, 0x70, 0x2e, 0x63, 0x6f, 0x6e, 0x73, 0x75, 0x6c, 0x2e, 0x72, 0x65,
+	0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x2e, 0x4c, 0x69, 0x73, 0x74, 0x42, 0x79, 0x4f, 0x77, 0x6e,
+	0x65, 0x72, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x2e, 0x2e, 0x68, 0x61, 0x73, 0x68,
+	0x69, 0x63, 0x6f, 0x72, 0x70, 0x2e, 0x63, 0x6f, 0x6e, 0x73, 0x75, 0x6c, 0x2e, 0x72, 0x65, 0x73,
+	0x6f, 0x75, 0x72, 0x63, 0x65, 0x2e, 0x4c, 0x69, 0x73, 0x74, 0x42, 0x79, 0x4f, 0x77, 0x6e, 0x65,
+	0x72, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x22, 0x08, 0xe2, 0x86, 0x04, 0x04, 0x08,
+	0x02, 0x10, 0x0b, 0x12, 0x67, 0x0a, 0x06, 0x44, 0x65, 0x6c, 0x65, 0x74, 0x65, 0x12, 0x28, 0x2e,
+	0x68, 0x61, 0x73, 0x68, 0x69, 0x63, 0x6f, 0x72, 0x70, 0x2e, 0x63, 0x6f, 0x6e, 0x73, 0x75, 0x6c,
+	0x2e, 0x72, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x2e, 0x44, 0x65, 0x6c, 0x65, 0x74, 0x65,
+	0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x29, 0x2e, 0x68, 0x61, 0x73, 0x68, 0x69, 0x63,
+	0x6f, 0x72, 0x70, 0x2e, 0x63, 0x6f, 0x6e, 0x73, 0x75, 0x6c, 0x2e, 0x72, 0x65, 0x73, 0x6f, 0x75,
+	0x72, 0x63, 0x65, 0x2e, 0x44, 0x65, 0x6c, 0x65, 0x74, 0x65, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e,
+	0x73, 0x65, 0x22, 0x08, 0xe2, 0x86, 0x04, 0x04, 0x08, 0x03, 0x10, 0x0b, 0x12, 0x6b, 0x0a, 0x09,
+	0x57, 0x61, 0x74, 0x63, 0x68, 0x4c, 0x69, 0x73, 0x74, 0x12, 0x2b, 0x2e, 0x68, 0x61, 0x73, 0x68,
+	0x69, 0x63, 0x6f, 0x72, 0x70, 0x2e, 0x63, 0x6f, 0x6e, 0x73, 0x75, 0x6c, 0x2e, 0x72, 0x65, 0x73,
+	0x6f, 0x75, 0x72, 0x63, 0x65, 0x2e, 0x57, 0x61, 0x74, 0x63, 0x68, 0x4c, 0x69, 0x73, 0x74, 0x52,
+	0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x25, 0x2e, 0x68, 0x61, 0x73, 0x68, 0x69, 0x63, 0x6f,
+	0x72, 0x70, 0x2e, 0x63, 0x6f, 0x6e, 0x73, 0x75, 0x6c, 0x2e, 0x72, 0x65, 0x73, 0x6f, 0x75, 0x72,
+	0x63, 0x65, 0x2e, 0x57, 0x61, 0x74, 0x63, 0x68, 0x45, 0x76, 0x65, 0x6e, 0x74, 0x22, 0x08, 0xe2,
+	0x86, 0x04, 0x04, 0x08, 0x02, 0x10, 0x0b, 0x30, 0x01, 0x12, 0x88, 0x01, 0x0a, 0x11, 0x4d, 0x75,
+	0x74, 0x61, 0x74, 0x65, 0x41, 0x6e, 0x64, 0x56, 0x61, 0x6c, 0x69, 0x64, 0x61, 0x74, 0x65, 0x12,
+	0x33, 0x2e, 0x68, 0x61, 0x73, 0x68, 0x69, 0x63, 0x6f, 0x72, 0x70, 0x2e, 0x63, 0x6f, 0x6e, 0x73,
+	0x75, 0x6c, 0x2e, 0x72, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x2e, 0x4d, 0x75, 0x74, 0x61,
+	0x74, 0x65, 0x41, 0x6e, 0x64, 0x56, 0x61, 0x6c, 0x69, 0x64, 0x61, 0x74, 0x65, 0x52, 0x65, 0x71,
+	0x75, 0x65, 0x73, 0x74, 0x1a, 0x34, 0x2e, 0x68, 0x61, 0x73, 0x68, 0x69, 0x63, 0x6f, 0x72, 0x70,
 	0x2e, 0x63, 0x6f, 0x6e, 0x73, 0x75, 0x6c, 0x2e, 0x72, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65,
 	0x2e, 0x4d, 0x75, 0x74, 0x61, 0x74, 0x65, 0x41, 0x6e, 0x64, 0x56, 0x61, 0x6c, 0x69, 0x64, 0x61,
-	0x74, 0x65, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x34, 0x2e, 0x68, 0x61, 0x73, 0x68,
+	0x74, 0x65, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x22, 0x08, 0xe2, 0x86, 0x04, 0x04,
-	0x69, 0x63, 0x6f, 0x72, 0x70, 0x2e, 0x63, 0x6f, 0x6e, 0x73, 0x75, 0x6c, 0x2e, 0x72, 0x65, 0x73,
+	0x08, 0x02, 0x10, 0x0b, 0x42, 0xe9, 0x01, 0x0a, 0x1d, 0x63, 0x6f, 0x6d, 0x2e, 0x68, 0x61, 0x73,
-	0x6f, 0x75, 0x72, 0x63, 0x65, 0x2e, 0x4d, 0x75, 0x74, 0x61, 0x74, 0x65, 0x41, 0x6e, 0x64, 0x56,
+	0x68, 0x69, 0x63, 0x6f, 0x72, 0x70, 0x2e, 0x63, 0x6f, 0x6e, 0x73, 0x75, 0x6c, 0x2e, 0x72, 0x65,
-	0x61, 0x6c, 0x69, 0x64, 0x61, 0x74, 0x65, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x22,
+	0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x42, 0x0d, 0x52, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65,
-	0x08, 0xe2, 0x86, 0x04, 0x04, 0x08, 0x02, 0x10, 0x0b, 0x42, 0xe9, 0x01, 0x0a, 0x1d, 0x63, 0x6f,
+	0x50, 0x72, 0x6f, 0x74, 0x6f, 0x50, 0x01, 0x5a, 0x33, 0x67, 0x69, 0x74, 0x68, 0x75, 0x62, 0x2e,
-	0x6d, 0x2e, 0x68, 0x61, 0x73, 0x68, 0x69, 0x63, 0x6f, 0x72, 0x70, 0x2e, 0x63, 0x6f, 0x6e, 0x73,
+	0x63, 0x6f, 0x6d, 0x2f, 0x68, 0x61, 0x73, 0x68, 0x69, 0x63, 0x6f, 0x72, 0x70, 0x2f, 0x63, 0x6f,
-	0x75, 0x6c, 0x2e, 0x72, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x42, 0x0d, 0x52, 0x65, 0x73,
+	0x6e, 0x73, 0x75, 0x6c, 0x2f, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x2d, 0x70, 0x75, 0x62, 0x6c, 0x69,
-	0x6f, 0x75, 0x72, 0x63, 0x65, 0x50, 0x72, 0x6f, 0x74, 0x6f, 0x50, 0x01, 0x5a, 0x33, 0x67, 0x69,
+	0x63, 0x2f, 0x70, 0x62, 0x72, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0xa2, 0x02, 0x03, 0x48,
-	0x74, 0x68, 0x75, 0x62, 0x2e, 0x63, 0x6f, 0x6d, 0x2f, 0x68, 0x61, 0x73, 0x68, 0x69, 0x63, 0x6f,
+	0x43, 0x52, 0xaa, 0x02, 0x19, 0x48, 0x61, 0x73, 0x68, 0x69, 0x63, 0x6f, 0x72, 0x70, 0x2e, 0x43,
-	0x72, 0x70, 0x2f, 0x63, 0x6f, 0x6e, 0x73, 0x75, 0x6c, 0x2f, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x2d,
+	0x6f, 0x6e, 0x73, 0x75, 0x6c, 0x2e, 0x52, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0xca, 0x02,
-	0x70, 0x75, 0x62, 0x6c, 0x69, 0x63, 0x2f, 0x70, 0x62, 0x72, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63,
+	0x19, 0x48, 0x61, 0x73, 0x68, 0x69, 0x63, 0x6f, 0x72, 0x70, 0x5c, 0x43, 0x6f, 0x6e, 0x73, 0x75,
-	0x65, 0xa2, 0x02, 0x03, 0x48, 0x43, 0x52, 0xaa, 0x02, 0x19, 0x48, 0x61, 0x73, 0x68, 0x69, 0x63,
+	0x6c, 0x5c, 0x52, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0xe2, 0x02, 0x25, 0x48, 0x61, 0x73,
-	0x6f, 0x72, 0x70, 0x2e, 0x43, 0x6f, 0x6e, 0x73, 0x75, 0x6c, 0x2e, 0x52, 0x65, 0x73, 0x6f, 0x75,
+	0x68, 0x69, 0x63, 0x6f, 0x72, 0x70, 0x5c, 0x43, 0x6f, 0x6e, 0x73, 0x75, 0x6c, 0x5c, 0x52, 0x65,
-	0x72, 0x63, 0x65, 0xca, 0x02, 0x19, 0x48, 0x61, 0x73, 0x68, 0x69, 0x63, 0x6f, 0x72, 0x70, 0x5c,
+	0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x5c, 0x47, 0x50, 0x42, 0x4d, 0x65, 0x74, 0x61, 0x64, 0x61,
-	0x43, 0x6f, 0x6e, 0x73, 0x75, 0x6c, 0x5c, 0x52, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0xe2,
+	0x74, 0x61, 0xea, 0x02, 0x1b, 0x48, 0x61, 0x73, 0x68, 0x69, 0x63, 0x6f, 0x72, 0x70, 0x3a, 0x3a,
-	0x02, 0x25, 0x48, 0x61, 0x73, 0x68, 0x69, 0x63, 0x6f, 0x72, 0x70, 0x5c, 0x43, 0x6f, 0x6e, 0x73,
+	0x43, 0x6f, 0x6e, 0x73, 0x75, 0x6c, 0x3a, 0x3a, 0x52, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65,
-	0x75, 0x6c, 0x5c, 0x52, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x5c, 0x47, 0x50, 0x42, 0x4d,
+	0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33,
-	0x65, 0x74, 0x61, 0x64, 0x61, 0x74, 0x61, 0xea, 0x02, 0x1b, 0x48, 0x61, 0x73, 0x68, 0x69, 0x63,
-	0x6f, 0x72, 0x70, 0x3a, 0x3a, 0x43, 0x6f, 0x6e, 0x73, 0x75, 0x6c, 0x3a, 0x3a, 0x52, 0x65, 0x73,
-	0x6f, 0x75, 0x72, 0x63, 0x65, 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33,
 }
 
 var (
@@ -1939,96 +2071,101 @@ func file_pbresource_resource_proto_rawDescGZIP() []byte {
 	return file_pbresource_resource_proto_rawDescData
 }
 
-var file_pbresource_resource_proto_enumTypes = make([]protoimpl.EnumInfo, 2)
+var file_pbresource_resource_proto_enumTypes = make([]protoimpl.EnumInfo, 1)
-var file_pbresource_resource_proto_msgTypes = make([]protoimpl.MessageInfo, 26)
+var file_pbresource_resource_proto_msgTypes = make([]protoimpl.MessageInfo, 29)
 var file_pbresource_resource_proto_goTypes = []interface{}{
 	(Condition_State)(0),              // 0: hashicorp.consul.resource.Condition.State
-	(WatchEvent_Operation)(0),         // 1: hashicorp.consul.resource.WatchEvent.Operation
+	(*Type)(nil),                      // 1: hashicorp.consul.resource.Type
-	(*Type)(nil),                      // 2: hashicorp.consul.resource.Type
+	(*Tenancy)(nil),                   // 2: hashicorp.consul.resource.Tenancy
-	(*Tenancy)(nil),                   // 3: hashicorp.consul.resource.Tenancy
+	(*ID)(nil),                        // 3: hashicorp.consul.resource.ID
-	(*ID)(nil),                        // 4: hashicorp.consul.resource.ID
+	(*Resource)(nil),                  // 4: hashicorp.consul.resource.Resource
-	(*Resource)(nil),                  // 5: hashicorp.consul.resource.Resource
+	(*Status)(nil),                    // 5: hashicorp.consul.resource.Status
-	(*Status)(nil),                    // 6: hashicorp.consul.resource.Status
+	(*Condition)(nil),                 // 6: hashicorp.consul.resource.Condition
-	(*Condition)(nil),                 // 7: hashicorp.consul.resource.Condition
+	(*Reference)(nil),                 // 7: hashicorp.consul.resource.Reference
-	(*Reference)(nil),                 // 8: hashicorp.consul.resource.Reference
+	(*Tombstone)(nil),                 // 8: hashicorp.consul.resource.Tombstone
-	(*Tombstone)(nil),                 // 9: hashicorp.consul.resource.Tombstone
+	(*ReadRequest)(nil),               // 9: hashicorp.consul.resource.ReadRequest
-	(*ReadRequest)(nil),               // 10: hashicorp.consul.resource.ReadRequest
+	(*ReadResponse)(nil),              // 10: hashicorp.consul.resource.ReadResponse
-	(*ReadResponse)(nil),              // 11: hashicorp.consul.resource.ReadResponse
+	(*ListRequest)(nil),               // 11: hashicorp.consul.resource.ListRequest
-	(*ListRequest)(nil),               // 12: hashicorp.consul.resource.ListRequest
+	(*ListResponse)(nil),              // 12: hashicorp.consul.resource.ListResponse
-	(*ListResponse)(nil),              // 13: hashicorp.consul.resource.ListResponse
+	(*ListByOwnerRequest)(nil),        // 13: hashicorp.consul.resource.ListByOwnerRequest
-	(*ListByOwnerRequest)(nil),        // 14: hashicorp.consul.resource.ListByOwnerRequest
+	(*ListByOwnerResponse)(nil),       // 14: hashicorp.consul.resource.ListByOwnerResponse
-	(*ListByOwnerResponse)(nil),       // 15: hashicorp.consul.resource.ListByOwnerResponse
+	(*WriteRequest)(nil),              // 15: hashicorp.consul.resource.WriteRequest
-	(*WriteRequest)(nil),              // 16: hashicorp.consul.resource.WriteRequest
+	(*WriteResponse)(nil),             // 16: hashicorp.consul.resource.WriteResponse
-	(*WriteResponse)(nil),             // 17: hashicorp.consul.resource.WriteResponse
+	(*WriteStatusRequest)(nil),        // 17: hashicorp.consul.resource.WriteStatusRequest
-	(*WriteStatusRequest)(nil),        // 18: hashicorp.consul.resource.WriteStatusRequest
+	(*WriteStatusResponse)(nil),       // 18: hashicorp.consul.resource.WriteStatusResponse
-	(*WriteStatusResponse)(nil),       // 19: hashicorp.consul.resource.WriteStatusResponse
+	(*DeleteRequest)(nil),             // 19: hashicorp.consul.resource.DeleteRequest
-	(*DeleteRequest)(nil),             // 20: hashicorp.consul.resource.DeleteRequest
+	(*DeleteResponse)(nil),            // 20: hashicorp.consul.resource.DeleteResponse
-	(*DeleteResponse)(nil),            // 21: hashicorp.consul.resource.DeleteResponse
+	(*WatchListRequest)(nil),          // 21: hashicorp.consul.resource.WatchListRequest
-	(*WatchListRequest)(nil),          // 22: hashicorp.consul.resource.WatchListRequest
+	(*WatchEvent)(nil),                // 22: hashicorp.consul.resource.WatchEvent
-	(*WatchEvent)(nil),                // 23: hashicorp.consul.resource.WatchEvent
+	(*MutateAndValidateRequest)(nil),  // 23: hashicorp.consul.resource.MutateAndValidateRequest
-	(*MutateAndValidateRequest)(nil),  // 24: hashicorp.consul.resource.MutateAndValidateRequest
+	(*MutateAndValidateResponse)(nil), // 24: hashicorp.consul.resource.MutateAndValidateResponse
-	(*MutateAndValidateResponse)(nil), // 25: hashicorp.consul.resource.MutateAndValidateResponse
+	nil,                               // 25: hashicorp.consul.resource.Resource.MetadataEntry
-	nil,                               // 26: hashicorp.consul.resource.Resource.MetadataEntry
+	nil,                               // 26: hashicorp.consul.resource.Resource.StatusEntry
-	nil,                               // 27: hashicorp.consul.resource.Resource.StatusEntry
+	(*WatchEvent_Upsert)(nil),         // 27: hashicorp.consul.resource.WatchEvent.Upsert
-	(*anypb.Any)(nil),                 // 28: google.protobuf.Any
+	(*WatchEvent_Delete)(nil),         // 28: hashicorp.consul.resource.WatchEvent.Delete
-	(*timestamppb.Timestamp)(nil),     // 29: google.protobuf.Timestamp
+	(*WatchEvent_EndOfSnapshot)(nil),  // 29: hashicorp.consul.resource.WatchEvent.EndOfSnapshot
+	(*anypb.Any)(nil),                 // 30: google.protobuf.Any
+	(*timestamppb.Timestamp)(nil),     // 31: google.protobuf.Timestamp
 }
 var file_pbresource_resource_proto_depIdxs = []int32{
-	2,  // 0: hashicorp.consul.resource.ID.type:type_name -> hashicorp.consul.resource.Type
+	1,  // 0: hashicorp.consul.resource.ID.type:type_name -> hashicorp.consul.resource.Type
-	3,  // 1: hashicorp.consul.resource.ID.tenancy:type_name -> hashicorp.consul.resource.Tenancy
+	2,  // 1: hashicorp.consul.resource.ID.tenancy:type_name -> hashicorp.consul.resource.Tenancy
-	4,  // 2: hashicorp.consul.resource.Resource.id:type_name -> hashicorp.consul.resource.ID
+	3,  // 2: hashicorp.consul.resource.Resource.id:type_name -> hashicorp.consul.resource.ID
-	4,  // 3: hashicorp.consul.resource.Resource.owner:type_name -> hashicorp.consul.resource.ID
+	3,  // 3: hashicorp.consul.resource.Resource.owner:type_name -> hashicorp.consul.resource.ID
-	26, // 4: hashicorp.consul.resource.Resource.metadata:type_name -> hashicorp.consul.resource.Resource.MetadataEntry
+	25, // 4: hashicorp.consul.resource.Resource.metadata:type_name -> hashicorp.consul.resource.Resource.MetadataEntry
-	27, // 5: hashicorp.consul.resource.Resource.status:type_name -> hashicorp.consul.resource.Resource.StatusEntry
+	26, // 5: hashicorp.consul.resource.Resource.status:type_name -> hashicorp.consul.resource.Resource.StatusEntry
-	28, // 6: hashicorp.consul.resource.Resource.data:type_name -> google.protobuf.Any
+	30, // 6: hashicorp.consul.resource.Resource.data:type_name -> google.protobuf.Any
-	7,  // 7: hashicorp.consul.resource.Status.conditions:type_name -> hashicorp.consul.resource.Condition
+	6,  // 7: hashicorp.consul.resource.Status.conditions:type_name -> hashicorp.consul.resource.Condition
-	29, // 8: hashicorp.consul.resource.Status.updated_at:type_name -> google.protobuf.Timestamp
+	31, // 8: hashicorp.consul.resource.Status.updated_at:type_name -> google.protobuf.Timestamp
 	0,  // 9: hashicorp.consul.resource.Condition.state:type_name -> hashicorp.consul.resource.Condition.State
-	8,  // 10: hashicorp.consul.resource.Condition.resource:type_name -> hashicorp.consul.resource.Reference
+	7,  // 10: hashicorp.consul.resource.Condition.resource:type_name -> hashicorp.consul.resource.Reference
-	2,  // 11: hashicorp.consul.resource.Reference.type:type_name -> hashicorp.consul.resource.Type
+	1,  // 11: hashicorp.consul.resource.Reference.type:type_name -> hashicorp.consul.resource.Type
-	3,  // 12: hashicorp.consul.resource.Reference.tenancy:type_name -> hashicorp.consul.resource.Tenancy
+	2,  // 12: hashicorp.consul.resource.Reference.tenancy:type_name -> hashicorp.consul.resource.Tenancy
-	4,  // 13: hashicorp.consul.resource.Tombstone.owner:type_name -> hashicorp.consul.resource.ID
+	3,  // 13: hashicorp.consul.resource.Tombstone.owner:type_name -> hashicorp.consul.resource.ID
-	4,  // 14: hashicorp.consul.resource.ReadRequest.id:type_name -> hashicorp.consul.resource.ID
+	3,  // 14: hashicorp.consul.resource.ReadRequest.id:type_name -> hashicorp.consul.resource.ID
-	5,  // 15: hashicorp.consul.resource.ReadResponse.resource:type_name -> hashicorp.consul.resource.Resource
+	4,  // 15: hashicorp.consul.resource.ReadResponse.resource:type_name -> hashicorp.consul.resource.Resource
-	2,  // 16: hashicorp.consul.resource.ListRequest.type:type_name -> hashicorp.consul.resource.Type
+	1,  // 16: hashicorp.consul.resource.ListRequest.type:type_name -> hashicorp.consul.resource.Type
-	3,  // 17: hashicorp.consul.resource.ListRequest.tenancy:type_name -> hashicorp.consul.resource.Tenancy
+	2,  // 17: hashicorp.consul.resource.ListRequest.tenancy:type_name -> hashicorp.consul.resource.Tenancy
-	5,  // 18: hashicorp.consul.resource.ListResponse.resources:type_name -> hashicorp.consul.resource.Resource
+	4,  // 18: hashicorp.consul.resource.ListResponse.resources:type_name -> hashicorp.consul.resource.Resource
-	4,  // 19: hashicorp.consul.resource.ListByOwnerRequest.owner:type_name -> hashicorp.consul.resource.ID
+	3,  // 19: hashicorp.consul.resource.ListByOwnerRequest.owner:type_name -> hashicorp.consul.resource.ID
-	5,  // 20: hashicorp.consul.resource.ListByOwnerResponse.resources:type_name -> hashicorp.consul.resource.Resource
+	4,  // 20: hashicorp.consul.resource.ListByOwnerResponse.resources:type_name -> hashicorp.consul.resource.Resource
-	5,  // 21: hashicorp.consul.resource.WriteRequest.resource:type_name -> hashicorp.consul.resource.Resource
+	4,  // 21: hashicorp.consul.resource.WriteRequest.resource:type_name -> hashicorp.consul.resource.Resource
-	5,  // 22: hashicorp.consul.resource.WriteResponse.resource:type_name -> hashicorp.consul.resource.Resource
+	4,  // 22: hashicorp.consul.resource.WriteResponse.resource:type_name -> hashicorp.consul.resource.Resource
-	4,  // 23: hashicorp.consul.resource.WriteStatusRequest.id:type_name -> hashicorp.consul.resource.ID
+	3,  // 23: hashicorp.consul.resource.WriteStatusRequest.id:type_name -> hashicorp.consul.resource.ID
-	6,  // 24: hashicorp.consul.resource.WriteStatusRequest.status:type_name -> hashicorp.consul.resource.Status
+	5,  // 24: hashicorp.consul.resource.WriteStatusRequest.status:type_name -> hashicorp.consul.resource.Status
-	5,  // 25: hashicorp.consul.resource.WriteStatusResponse.resource:type_name -> hashicorp.consul.resource.Resource
+	4,  // 25: hashicorp.consul.resource.WriteStatusResponse.resource:type_name -> hashicorp.consul.resource.Resource
-	4,  // 26: hashicorp.consul.resource.DeleteRequest.id:type_name -> hashicorp.consul.resource.ID
+	3,  // 26: hashicorp.consul.resource.DeleteRequest.id:type_name -> hashicorp.consul.resource.ID
-	2,  // 27: hashicorp.consul.resource.WatchListRequest.type:type_name -> hashicorp.consul.resource.Type
+	1,  // 27: hashicorp.consul.resource.WatchListRequest.type:type_name -> hashicorp.consul.resource.Type
-	3,  // 28: hashicorp.consul.resource.WatchListRequest.tenancy:type_name -> hashicorp.consul.resource.Tenancy
+	2,  // 28: hashicorp.consul.resource.WatchListRequest.tenancy:type_name -> hashicorp.consul.resource.Tenancy
-	1,  // 29: hashicorp.consul.resource.WatchEvent.operation:type_name -> hashicorp.consul.resource.WatchEvent.Operation
+	27, // 29: hashicorp.consul.resource.WatchEvent.upsert:type_name -> hashicorp.consul.resource.WatchEvent.Upsert
-	5,  // 30: hashicorp.consul.resource.WatchEvent.resource:type_name -> hashicorp.consul.resource.Resource
+	28, // 30: hashicorp.consul.resource.WatchEvent.delete:type_name -> hashicorp.consul.resource.WatchEvent.Delete
-	5,  // 31: hashicorp.consul.resource.MutateAndValidateRequest.resource:type_name -> hashicorp.consul.resource.Resource
+	29, // 31: hashicorp.consul.resource.WatchEvent.end_of_snapshot:type_name -> hashicorp.consul.resource.WatchEvent.EndOfSnapshot
-	5,  // 32: hashicorp.consul.resource.MutateAndValidateResponse.resource:type_name -> hashicorp.consul.resource.Resource
+	4,  // 32: hashicorp.consul.resource.MutateAndValidateRequest.resource:type_name -> hashicorp.consul.resource.Resource
-	6,  // 33: hashicorp.consul.resource.Resource.StatusEntry.value:type_name -> hashicorp.consul.resource.Status
+	4,  // 33: hashicorp.consul.resource.MutateAndValidateResponse.resource:type_name -> hashicorp.consul.resource.Resource
-	10, // 34: hashicorp.consul.resource.ResourceService.Read:input_type -> hashicorp.consul.resource.ReadRequest
+	5,  // 34: hashicorp.consul.resource.Resource.StatusEntry.value:type_name -> hashicorp.consul.resource.Status
-	16, // 35: hashicorp.consul.resource.ResourceService.Write:input_type -> hashicorp.consul.resource.WriteRequest
+	4,  // 35: hashicorp.consul.resource.WatchEvent.Upsert.resource:type_name -> hashicorp.consul.resource.Resource
-	18, // 36: hashicorp.consul.resource.ResourceService.WriteStatus:input_type -> hashicorp.consul.resource.WriteStatusRequest
+	4,  // 36: hashicorp.consul.resource.WatchEvent.Delete.resource:type_name -> hashicorp.consul.resource.Resource
-	12, // 37: hashicorp.consul.resource.ResourceService.List:input_type -> hashicorp.consul.resource.ListRequest
+	9,  // 37: hashicorp.consul.resource.ResourceService.Read:input_type -> hashicorp.consul.resource.ReadRequest
-	14, // 38: hashicorp.consul.resource.ResourceService.ListByOwner:input_type -> hashicorp.consul.resource.ListByOwnerRequest
+	15, // 38: hashicorp.consul.resource.ResourceService.Write:input_type -> hashicorp.consul.resource.WriteRequest
-	20, // 39: hashicorp.consul.resource.ResourceService.Delete:input_type -> hashicorp.consul.resource.DeleteRequest
+	17, // 39: hashicorp.consul.resource.ResourceService.WriteStatus:input_type -> hashicorp.consul.resource.WriteStatusRequest
-	22, // 40: hashicorp.consul.resource.ResourceService.WatchList:input_type -> hashicorp.consul.resource.WatchListRequest
+	11, // 40: hashicorp.consul.resource.ResourceService.List:input_type -> hashicorp.consul.resource.ListRequest
-	24, // 41: hashicorp.consul.resource.ResourceService.MutateAndValidate:input_type -> hashicorp.consul.resource.MutateAndValidateRequest
+	13, // 41: hashicorp.consul.resource.ResourceService.ListByOwner:input_type -> hashicorp.consul.resource.ListByOwnerRequest
-	11, // 42: hashicorp.consul.resource.ResourceService.Read:output_type -> hashicorp.consul.resource.ReadResponse
+	19, // 42: hashicorp.consul.resource.ResourceService.Delete:input_type -> hashicorp.consul.resource.DeleteRequest
-	17, // 43: hashicorp.consul.resource.ResourceService.Write:output_type -> hashicorp.consul.resource.WriteResponse
+	21, // 43: hashicorp.consul.resource.ResourceService.WatchList:input_type -> hashicorp.consul.resource.WatchListRequest
-	19, // 44: hashicorp.consul.resource.ResourceService.WriteStatus:output_type -> hashicorp.consul.resource.WriteStatusResponse
+	23, // 44: hashicorp.consul.resource.ResourceService.MutateAndValidate:input_type -> hashicorp.consul.resource.MutateAndValidateRequest
-	13, // 45: hashicorp.consul.resource.ResourceService.List:output_type -> hashicorp.consul.resource.ListResponse
+	10, // 45: hashicorp.consul.resource.ResourceService.Read:output_type -> hashicorp.consul.resource.ReadResponse
-	15, // 46: hashicorp.consul.resource.ResourceService.ListByOwner:output_type -> hashicorp.consul.resource.ListByOwnerResponse
+	16, // 46: hashicorp.consul.resource.ResourceService.Write:output_type -> hashicorp.consul.resource.WriteResponse
-	21, // 47: hashicorp.consul.resource.ResourceService.Delete:output_type -> hashicorp.consul.resource.DeleteResponse
+	18, // 47: hashicorp.consul.resource.ResourceService.WriteStatus:output_type -> hashicorp.consul.resource.WriteStatusResponse
-	23, // 48: hashicorp.consul.resource.ResourceService.WatchList:output_type -> hashicorp.consul.resource.WatchEvent
+	12, // 48: hashicorp.consul.resource.ResourceService.List:output_type -> hashicorp.consul.resource.ListResponse
-	25, // 49: hashicorp.consul.resource.ResourceService.MutateAndValidate:output_type -> hashicorp.consul.resource.MutateAndValidateResponse
+	14, // 49: hashicorp.consul.resource.ResourceService.ListByOwner:output_type -> hashicorp.consul.resource.ListByOwnerResponse
-	42, // [42:50] is the sub-list for method output_type
+	20, // 50: hashicorp.consul.resource.ResourceService.Delete:output_type -> hashicorp.consul.resource.DeleteResponse
-	34, // [34:42] is the sub-list for method input_type
+	22, // 51: hashicorp.consul.resource.ResourceService.WatchList:output_type -> hashicorp.consul.resource.WatchEvent
-	34, // [34:34] is the sub-list for extension type_name
+	24, // 52: hashicorp.consul.resource.ResourceService.MutateAndValidate:output_type -> hashicorp.consul.resource.MutateAndValidateResponse
-	34, // [34:34] is the sub-list for extension extendee
+	45, // [45:53] is the sub-list for method output_type
-	0,  // [0:34] is the sub-list for field type_name
+	37, // [37:45] is the sub-list for method input_type
+	37, // [37:37] is the sub-list for extension type_name
+	37, // [37:37] is the sub-list for extension extendee
+	0,  // [0:37] is the sub-list for field type_name
 }
 
 func init() { file_pbresource_resource_proto_init() }
@@ -2325,14 +2462,55 @@ func file_pbresource_resource_proto_init() {
 				return nil
 			}
 		}
+		file_pbresource_resource_proto_msgTypes[26].Exporter = func(v interface{}, i int) interface{} {
+			switch v := v.(*WatchEvent_Upsert); i {
+			case 0:
+				return &v.state
+			case 1:
+				return &v.sizeCache
+			case 2:
+				return &v.unknownFields
+			default:
+				return nil
+			}
+		}
+		file_pbresource_resource_proto_msgTypes[27].Exporter = func(v interface{}, i int) interface{} {
+			switch v := v.(*WatchEvent_Delete); i {
+			case 0:
+				return &v.state
+			case 1:
+				return &v.sizeCache
+			case 2:
+				return &v.unknownFields
+			default:
+				return nil
+			}
+		}
+		file_pbresource_resource_proto_msgTypes[28].Exporter = func(v interface{}, i int) interface{} {
+			switch v := v.(*WatchEvent_EndOfSnapshot); i {
+			case 0:
+				return &v.state
+			case 1:
+				return &v.sizeCache
+			case 2:
+				return &v.unknownFields
+			default:
+				return nil
+			}
+		}
+	}
+	file_pbresource_resource_proto_msgTypes[21].OneofWrappers = []interface{}{
+		(*WatchEvent_Upsert_)(nil),
+		(*WatchEvent_Delete_)(nil),
+		(*WatchEvent_EndOfSnapshot_)(nil),
 	}
 	type x struct{}
 	out := protoimpl.TypeBuilder{
 		File: protoimpl.DescBuilder{
 			GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
 			RawDescriptor: file_pbresource_resource_proto_rawDesc,
-			NumEnums:      2,
+			NumEnums:      1,
-			NumMessages:   26,
+			NumMessages:   29,
 			NumExtensions: 0,
 			NumServices:   1,
 		},

proto-public/pbresource/resource.proto

@@ -473,26 +473,27 @@ message WatchListRequest {
 
 // WatchEvent is emitted on the WatchList stream when a resource changes.
 message WatchEvent {
-  // Operation describes the type of event.
+  // Upsert indicates that the resource was written (i.e. created or
-  enum Operation {
+  // updated). All events from the initial state-of-the-world will be upsert
-    // OPERATION_UNSPECIFIED is the default/zero value. You should not see it
+  // events.
-    // in practice.
+  message Upsert {
-    OPERATION_UNSPECIFIED = 0;
+    Resource resource = 1;
-
-    // OPERATION_UPSERT indicates that the resource was written (i.e. created or
-    // updated). All events from the initial state-of-the-world will be upsert
-    // events.
-    OPERATION_UPSERT = 1;
-
-    // OPERATION_DELETED indicates that the resource was deleted.
-    OPERATION_DELETE = 2;
   }
 
-  // Operation describes the type of event.
+  // Delete indicates that the resource was deleted.
-  Operation operation = 1;
+  message Delete {
+    Resource resource = 1;
+  }
 
-  // Resource the event relates to.
+  // EndOfSnapshot is sent to indicate that the initial snapshot events have
-  Resource resource = 2;
+  // been sent and future events will modify that set.
+  message EndOfSnapshot {}
+
+  oneof event {
+    Upsert upsert = 1;
+    Delete delete = 2;
+    EndOfSnapshot end_of_snapshot = 3;
+  }
 }
 
 // MutateAndValidateRequest contains the parameters to the MutateAndValidate endpoint.

proto-public/pbresource/resource_deepcopy.gen.go

@@ -467,6 +467,69 @@ func (in *WatchEvent) DeepCopyInterface() interface{} {
 	return in.DeepCopy()
 }
 
+// DeepCopyInto supports using WatchEvent_Upsert within kubernetes types, where deepcopy-gen is used.
+func (in *WatchEvent_Upsert) DeepCopyInto(out *WatchEvent_Upsert) {
+	proto.Reset(out)
+	proto.Merge(out, proto.Clone(in))
+}
+
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new WatchEvent_Upsert. Required by controller-gen.
+func (in *WatchEvent_Upsert) DeepCopy() *WatchEvent_Upsert {
+	if in == nil {
+		return nil
+	}
+	out := new(WatchEvent_Upsert)
+	in.DeepCopyInto(out)
+	return out
+}
+
+// DeepCopyInterface is an autogenerated deepcopy function, copying the receiver, creating a new WatchEvent_Upsert. Required by controller-gen.
+func (in *WatchEvent_Upsert) DeepCopyInterface() interface{} {
+	return in.DeepCopy()
+}
+
+// DeepCopyInto supports using WatchEvent_Delete within kubernetes types, where deepcopy-gen is used.
+func (in *WatchEvent_Delete) DeepCopyInto(out *WatchEvent_Delete) {
+	proto.Reset(out)
+	proto.Merge(out, proto.Clone(in))
+}
+
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new WatchEvent_Delete. Required by controller-gen.
+func (in *WatchEvent_Delete) DeepCopy() *WatchEvent_Delete {
+	if in == nil {
+		return nil
+	}
+	out := new(WatchEvent_Delete)
+	in.DeepCopyInto(out)
+	return out
+}
+
+// DeepCopyInterface is an autogenerated deepcopy function, copying the receiver, creating a new WatchEvent_Delete. Required by controller-gen.
+func (in *WatchEvent_Delete) DeepCopyInterface() interface{} {
+	return in.DeepCopy()
+}
+
+// DeepCopyInto supports using WatchEvent_EndOfSnapshot within kubernetes types, where deepcopy-gen is used.
+func (in *WatchEvent_EndOfSnapshot) DeepCopyInto(out *WatchEvent_EndOfSnapshot) {
+	proto.Reset(out)
+	proto.Merge(out, proto.Clone(in))
+}
+
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new WatchEvent_EndOfSnapshot. Required by controller-gen.
+func (in *WatchEvent_EndOfSnapshot) DeepCopy() *WatchEvent_EndOfSnapshot {
+	if in == nil {
+		return nil
+	}
+	out := new(WatchEvent_EndOfSnapshot)
+	in.DeepCopyInto(out)
+	return out
+}
+
+// DeepCopyInterface is an autogenerated deepcopy function, copying the receiver, creating a new WatchEvent_EndOfSnapshot. Required by controller-gen.
+func (in *WatchEvent_EndOfSnapshot) DeepCopyInterface() interface{} {
+	return in.DeepCopy()
+}
+
 // DeepCopyInto supports using MutateAndValidateRequest within kubernetes types, where deepcopy-gen is used.
 func (in *MutateAndValidateRequest) DeepCopyInto(out *MutateAndValidateRequest) {
 	proto.Reset(out)

proto-public/pbresource/resource_json.gen.go

@@ -247,6 +247,39 @@ func (this *WatchEvent) UnmarshalJSON(b []byte) error {
 	return ResourceUnmarshaler.Unmarshal(b, this)
 }
 
+// MarshalJSON is a custom marshaler for WatchEvent_Upsert
+func (this *WatchEvent_Upsert) MarshalJSON() ([]byte, error) {
+	str, err := ResourceMarshaler.Marshal(this)
+	return []byte(str), err
+}
+
+// UnmarshalJSON is a custom unmarshaler for WatchEvent_Upsert
+func (this *WatchEvent_Upsert) UnmarshalJSON(b []byte) error {
+	return ResourceUnmarshaler.Unmarshal(b, this)
+}
+
+// MarshalJSON is a custom marshaler for WatchEvent_Delete
+func (this *WatchEvent_Delete) MarshalJSON() ([]byte, error) {
+	str, err := ResourceMarshaler.Marshal(this)
+	return []byte(str), err
+}
+
+// UnmarshalJSON is a custom unmarshaler for WatchEvent_Delete
+func (this *WatchEvent_Delete) UnmarshalJSON(b []byte) error {
+	return ResourceUnmarshaler.Unmarshal(b, this)
+}
+
+// MarshalJSON is a custom marshaler for WatchEvent_EndOfSnapshot
+func (this *WatchEvent_EndOfSnapshot) MarshalJSON() ([]byte, error) {
+	str, err := ResourceMarshaler.Marshal(this)
+	return []byte(str), err
+}
+
+// UnmarshalJSON is a custom unmarshaler for WatchEvent_EndOfSnapshot
+func (this *WatchEvent_EndOfSnapshot) UnmarshalJSON(b []byte) error {
+	return ResourceUnmarshaler.Unmarshal(b, this)
+}
+
 // MarshalJSON is a custom marshaler for MutateAndValidateRequest
 func (this *MutateAndValidateRequest) MarshalJSON() ([]byte, error) {
 	str, err := ResourceMarshaler.Marshal(this)