diff --git a/agent/grpc-external/services/resource/watch.go b/agent/grpc-external/services/resource/watch.go
index f1874e6edb..c4f21860c0 100644
--- a/agent/grpc-external/services/resource/watch.go
+++ b/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
diff --git a/agent/grpc-external/services/resource/watch_test.go b/agent/grpc-external/services/resource/watch_test.go
index bd37dbef32..5ccdb609ba 100644
--- a/agent/grpc-external/services/resource/watch_test.go
+++ b/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)
-	client := testClient(t, server)
-	demo.RegisterTypes(server.Registry)
+	b := svctest.NewResourceServiceBuilder().
+		WithRegisterFns(demo.RegisterTypes)
+	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)
-	prototest.AssertDeepEqual(t, r1, rsp.Resource)
+	require.NotNil(t, rsp.GetUpsert())
+	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)
-	prototest.AssertDeepEqual(t, r2, rsp.Resource)
+	require.NotNil(t, rsp.GetUpsert())
+	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)
-			prototest.AssertDeepEqual(t, expected, rsp.Resource)
+			require.NotNil(t, rsp.GetUpsert())
+			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)
-	demo.RegisterTypes(server.Registry)
-	client := testClient(t, server)
+	b := svctest.NewResourceServiceBuilder().
+		WithRegisterFns(demo.RegisterTypes)
+	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) {
-	server := testServer(t)
-	client := testClient(t, server)
-
+func roundTripACL(t *testing.T, authz acl.Authorizer, expectErr bool) (<-chan resourceOrError, *pbresource.Resource) {
 	mockACLResolver := &svc.MockACLResolver{}
 	mockACLResolver.On("ResolveTokenAndDefaultMeta", mock.Anything, mock.Anything, mock.Anything).
-		Return(authz, nil)
-	server.ACLResolver = mockACLResolver
-	demo.RegisterTypes(server.Registry)
+		Return(resolver.Result{Authorizer: authz}, nil)
+
+	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)
-	prototest.AssertDeepEqual(t, rsp1.Resource, rsp2.Resource)
+	require.NotNil(t, rsp2.GetUpsert())
+	prototest.AssertDeepEqual(t, rsp1.Resource, rsp2.GetUpsert().Resource)
 }
diff --git a/grpcmocks/proto-public/pbresource/mock_isWatchEvent_Event.go b/grpcmocks/proto-public/pbresource/mock_isWatchEvent_Event.go
new file mode 100644
index 0000000000..abefc113f0
--- /dev/null
+++ b/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
+}
diff --git a/internal/controller/mem_consistency_test.go b/internal/controller/mem_consistency_test.go
index 978b412bf7..d61b276772 100644
--- a/internal/controller/mem_consistency_test.go
+++ b/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{
-		Operation: pbresource.WatchEvent_OPERATION_UPSERT,
-		Resource:  res1,
-	})
+	v1WatchListClientCreate := func() pbresource.ResourceService_WatchListClient {
+		return watchListEvents(t, &pbresource.WatchEvent{
+			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{
-		Operation: pbresource.WatchEvent_OPERATION_UPSERT,
-		Resource:  res2,
-	})
+	v2WatchListClientCreate := func() pbresource.ResourceService_WatchListClient {
+		return watchListEvents(t, nil, &pbresource.WatchEvent{
+			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).
-		Once()
+		RunAndReturn(func(_ context.Context, _ *pbresource.WatchListRequest, _ ...grpc.CallOption) (pbresource.ResourceService_WatchListClient, error) {
+			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).
-		Once()
+		RunAndReturn(func(_ context.Context, _ *pbresource.WatchListRequest, _ ...grpc.CallOption) (pbresource.ResourceService_WatchListClient, error) {
+			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{
-		Operation: pbresource.WatchEvent_OPERATION_UPSERT,
-		Resource:  res,
-	})
+	v2WatchListClientCreate := func() pbresource.ResourceService_WatchListClient {
+		return watchListEvents(t, nil, &pbresource.WatchEvent{
+			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).
-		Once()
+		RunAndReturn(func(_ context.Context, _ *pbresource.WatchListRequest, _ ...grpc.CallOption) (pbresource.ResourceService_WatchListClient, error) {
+			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.
diff --git a/internal/controller/runner.go b/internal/controller/runner.go
index 884344da51..c4240c3cf0 100644
--- a/internal/controller/runner.go
+++ b/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 {
-		case pbresource.WatchEvent_OPERATION_UPSERT:
-			c.cache.Insert(event.Resource)
-		case pbresource.WatchEvent_OPERATION_DELETE:
-			c.cache.Delete(event.Resource)
+		var resource *pbresource.Resource
+		switch {
+		case event.GetUpsert() != nil:
+			resource = event.GetUpsert().GetResource()
+			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))
 	}
 }
 
diff --git a/internal/storage/conformance/conformance.go b/internal/storage/conformance/conformance.go
index 762159d0ff..0b815a847b 100644
--- a/internal/storage/conformance/conformance.go
+++ b/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)
-					prototest.AssertContainsElement(t, tc.results, event.Resource, ignoreVersion)
+					if opts.IgnoreWatchListSnapshotOperations && event.GetEndOfSnapshot() != nil {
+						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)
-					prototest.AssertContainsElement(t, tc.results, event.Resource, ignoreVersion)
+					require.NotNil(t, event.GetUpsert())
+					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)
-				prototest.AssertDeepEqual(t, del, event.Resource)
+				require.NotNil(t, event.GetDelete())
+				prototest.AssertDeepEqual(t, del, event.GetDelete().Resource)
 
 				// Check that Read implements "monotonic reads" with Watch.
 				_, err = backend.Read(ctx, storage.EventualConsistency, del.Id)
diff --git a/internal/storage/inmem/snapshot_test.go b/internal/storage/inmem/snapshot_test.go
index 22757d5f57..d19d5b13ee 100644
--- a/internal/storage/inmem/snapshot_test.go
+++ b/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)
diff --git a/internal/storage/inmem/store.go b/internal/storage/inmem/store.go
index 051edd4c89..99f4ddc3d5 100644
--- a/internal/storage/inmem/store.go
+++ b/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
 }
diff --git a/internal/storage/inmem/watch.go b/internal/storage/inmem/watch.go
index 7660d7b5c2..d0157abf08 100644
--- a/internal/storage/inmem/watch.go
+++ b/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) {
-	id := res.Id
+func (s *Store) publishEvent(idx uint64, event *pbresource.WatchEvent) {
+	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))
-	for i, r := range results {
-		events[i] = stream.Event{
+	events := make([]stream.Event, 0, len(results)+1)
+	addEvent := func(event *pbresource.WatchEvent) {
+		events = append(events, stream.Event{
 			Topic: eventTopic,
 			Index: idx,
 			Payload: eventPayload{
 				subject: req.Subject,
-				event: &pbresource.WatchEvent{
-					Operation: pbresource.WatchEvent_OPERATION_UPSERT,
-					Resource:  r,
+				event:   event,
+			},
+		})
+	}
+
+	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
diff --git a/internal/storage/raft/conformance_test.go b/internal/storage/raft/conformance_test.go
index d76ae1b456..21802ca671 100644
--- a/internal/storage/raft/conformance_test.go
+++ b/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,
 		})
 	})
 }
diff --git a/proto-public/pbresource/resource.pb.binary.go b/proto-public/pbresource/resource.pb.binary.go
index d7e4ff8b89..5f2aabfbcc 100644
--- a/proto-public/pbresource/resource.pb.binary.go
+++ b/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)
diff --git a/proto-public/pbresource/resource.pb.go b/proto-public/pbresource/resource.pb.go
index 0f8f53405e..2b344a24b5 100644
--- a/proto-public/pbresource/resource.pb.go
+++ b/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.
-	Operation WatchEvent_Operation `protobuf:"varint,1,opt,name=operation,proto3,enum=hashicorp.consul.resource.WatchEvent_Operation" json:"operation,omitempty"`
-	// Resource the event relates to.
-	Resource *Resource `protobuf:"bytes,2,opt,name=resource,proto3" json:"resource,omitempty"`
+	// Types that are assignable to Event:
+	//
+	//	*WatchEvent_Upsert_
+	//	*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 {
-	if x != nil {
-		return x.Operation
-	}
-	return WatchEvent_OPERATION_UNSPECIFIED
-}
-
-func (x *WatchEvent) GetResource() *Resource {
-	if x != nil {
-		return x.Resource
+func (m *WatchEvent) GetEvent() isWatchEvent_Event {
+	if m != nil {
+		return m.Event
 	}
 	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,
-	0x61, 0x74, 0x63, 0x68, 0x45, 0x76, 0x65, 0x6e, 0x74, 0x12, 0x4d, 0x0a, 0x09, 0x6f, 0x70, 0x65,
-	0x72, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0e, 0x32, 0x2f, 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, 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,
+	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, 0x46, 0x0a, 0x06, 0x75, 0x70, 0x73,
+	0x65, 0x72, 0x74, 0x18, 0x01, 0x20, 0x01, 0x28, 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, 0x52, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x52, 0x08,
-	0x72, 0x65, 0x73, 0x6f, 0x75, 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,
+	0x6f, 0x75, 0x72, 0x63, 0x65, 0x2e, 0x57, 0x61, 0x74, 0x63, 0x68, 0x45, 0x76, 0x65, 0x6e, 0x74,
+	0x2e, 0x55, 0x70, 0x73, 0x65, 0x72, 0x74, 0x48, 0x00, 0x52, 0x06, 0x75, 0x70, 0x73, 0x65, 0x72,
+	0x74, 0x12, 0x46, 0x0a, 0x06, 0x64, 0x65, 0x6c, 0x65, 0x74, 0x65, 0x18, 0x02, 0x20, 0x01, 0x28,
+	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,
-	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, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73,
-	0x74, 0x1a, 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, 0x57, 0x72,
-	0x69, 0x74, 0x65, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x22, 0x08, 0xe2, 0x86, 0x04,
-	0x04, 0x08, 0x03, 0x10, 0x0b, 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, 0x52, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x52, 0x08, 0x72, 0x65, 0x73, 0x6f, 0x75,
+	0x72, 0x63, 0x65, 0x1a, 0x49, 0x0a, 0x06, 0x44, 0x65, 0x6c, 0x65, 0x74, 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, 0x72, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x1a, 0x0f,
+	0x0a, 0x0d, 0x45, 0x6e, 0x64, 0x4f, 0x66, 0x53, 0x6e, 0x61, 0x70, 0x73, 0x68, 0x6f, 0x74, 0x42,
+	0x07, 0x0a, 0x05, 0x65, 0x76, 0x65, 0x6e, 0x74, 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, 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,
+	0x2e, 0x52, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x52, 0x08, 0x72, 0x65, 0x73, 0x6f, 0x75,
+	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,
-	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,
+	0x57, 0x72, 0x69, 0x74, 0x65, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 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, 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,
+	0x72, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x2e, 0x57, 0x72, 0x69, 0x74, 0x65, 0x52, 0x65,
+	0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x22, 0x08, 0xe2, 0x86, 0x04, 0x04, 0x08, 0x03, 0x10, 0x0b,
+	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,
-	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,
+	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, 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, 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,
-	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, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x22,
-	0x08, 0xe2, 0x86, 0x04, 0x04, 0x08, 0x02, 0x10, 0x0b, 0x42, 0xe9, 0x01, 0x0a, 0x1d, 0x63, 0x6f,
-	0x6d, 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, 0x42, 0x0d, 0x52, 0x65, 0x73,
-	0x6f, 0x75, 0x72, 0x63, 0x65, 0x50, 0x72, 0x6f, 0x74, 0x6f, 0x50, 0x01, 0x5a, 0x33, 0x67, 0x69,
-	0x74, 0x68, 0x75, 0x62, 0x2e, 0x63, 0x6f, 0x6d, 0x2f, 0x68, 0x61, 0x73, 0x68, 0x69, 0x63, 0x6f,
-	0x72, 0x70, 0x2f, 0x63, 0x6f, 0x6e, 0x73, 0x75, 0x6c, 0x2f, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x2d,
-	0x70, 0x75, 0x62, 0x6c, 0x69, 0x63, 0x2f, 0x70, 0x62, 0x72, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63,
-	0x65, 0xa2, 0x02, 0x03, 0x48, 0x43, 0x52, 0xaa, 0x02, 0x19, 0x48, 0x61, 0x73, 0x68, 0x69, 0x63,
-	0x6f, 0x72, 0x70, 0x2e, 0x43, 0x6f, 0x6e, 0x73, 0x75, 0x6c, 0x2e, 0x52, 0x65, 0x73, 0x6f, 0x75,
-	0x72, 0x63, 0x65, 0xca, 0x02, 0x19, 0x48, 0x61, 0x73, 0x68, 0x69, 0x63, 0x6f, 0x72, 0x70, 0x5c,
-	0x43, 0x6f, 0x6e, 0x73, 0x75, 0x6c, 0x5c, 0x52, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0xe2,
-	0x02, 0x25, 0x48, 0x61, 0x73, 0x68, 0x69, 0x63, 0x6f, 0x72, 0x70, 0x5c, 0x43, 0x6f, 0x6e, 0x73,
-	0x75, 0x6c, 0x5c, 0x52, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x5c, 0x47, 0x50, 0x42, 0x4d,
-	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,
+	0x74, 0x65, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x22, 0x08, 0xe2, 0x86, 0x04, 0x04,
+	0x08, 0x02, 0x10, 0x0b, 0x42, 0xe9, 0x01, 0x0a, 0x1d, 0x63, 0x6f, 0x6d, 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, 0x42, 0x0d, 0x52, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65,
+	0x50, 0x72, 0x6f, 0x74, 0x6f, 0x50, 0x01, 0x5a, 0x33, 0x67, 0x69, 0x74, 0x68, 0x75, 0x62, 0x2e,
+	0x63, 0x6f, 0x6d, 0x2f, 0x68, 0x61, 0x73, 0x68, 0x69, 0x63, 0x6f, 0x72, 0x70, 0x2f, 0x63, 0x6f,
+	0x6e, 0x73, 0x75, 0x6c, 0x2f, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x2d, 0x70, 0x75, 0x62, 0x6c, 0x69,
+	0x63, 0x2f, 0x70, 0x62, 0x72, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0xa2, 0x02, 0x03, 0x48,
+	0x43, 0x52, 0xaa, 0x02, 0x19, 0x48, 0x61, 0x73, 0x68, 0x69, 0x63, 0x6f, 0x72, 0x70, 0x2e, 0x43,
+	0x6f, 0x6e, 0x73, 0x75, 0x6c, 0x2e, 0x52, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0xca, 0x02,
+	0x19, 0x48, 0x61, 0x73, 0x68, 0x69, 0x63, 0x6f, 0x72, 0x70, 0x5c, 0x43, 0x6f, 0x6e, 0x73, 0x75,
+	0x6c, 0x5c, 0x52, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0xe2, 0x02, 0x25, 0x48, 0x61, 0x73,
+	0x68, 0x69, 0x63, 0x6f, 0x72, 0x70, 0x5c, 0x43, 0x6f, 0x6e, 0x73, 0x75, 0x6c, 0x5c, 0x52, 0x65,
+	0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x5c, 0x47, 0x50, 0x42, 0x4d, 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_msgTypes = make([]protoimpl.MessageInfo, 26)
+var file_pbresource_resource_proto_enumTypes = make([]protoimpl.EnumInfo, 1)
+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),                      // 2: hashicorp.consul.resource.Type
-	(*Tenancy)(nil),                   // 3: hashicorp.consul.resource.Tenancy
-	(*ID)(nil),                        // 4: hashicorp.consul.resource.ID
-	(*Resource)(nil),                  // 5: hashicorp.consul.resource.Resource
-	(*Status)(nil),                    // 6: hashicorp.consul.resource.Status
-	(*Condition)(nil),                 // 7: hashicorp.consul.resource.Condition
-	(*Reference)(nil),                 // 8: hashicorp.consul.resource.Reference
-	(*Tombstone)(nil),                 // 9: hashicorp.consul.resource.Tombstone
-	(*ReadRequest)(nil),               // 10: hashicorp.consul.resource.ReadRequest
-	(*ReadResponse)(nil),              // 11: hashicorp.consul.resource.ReadResponse
-	(*ListRequest)(nil),               // 12: hashicorp.consul.resource.ListRequest
-	(*ListResponse)(nil),              // 13: hashicorp.consul.resource.ListResponse
-	(*ListByOwnerRequest)(nil),        // 14: hashicorp.consul.resource.ListByOwnerRequest
-	(*ListByOwnerResponse)(nil),       // 15: hashicorp.consul.resource.ListByOwnerResponse
-	(*WriteRequest)(nil),              // 16: hashicorp.consul.resource.WriteRequest
-	(*WriteResponse)(nil),             // 17: hashicorp.consul.resource.WriteResponse
-	(*WriteStatusRequest)(nil),        // 18: hashicorp.consul.resource.WriteStatusRequest
-	(*WriteStatusResponse)(nil),       // 19: hashicorp.consul.resource.WriteStatusResponse
-	(*DeleteRequest)(nil),             // 20: hashicorp.consul.resource.DeleteRequest
-	(*DeleteResponse)(nil),            // 21: hashicorp.consul.resource.DeleteResponse
-	(*WatchListRequest)(nil),          // 22: hashicorp.consul.resource.WatchListRequest
-	(*WatchEvent)(nil),                // 23: hashicorp.consul.resource.WatchEvent
-	(*MutateAndValidateRequest)(nil),  // 24: hashicorp.consul.resource.MutateAndValidateRequest
-	(*MutateAndValidateResponse)(nil), // 25: hashicorp.consul.resource.MutateAndValidateResponse
-	nil,                               // 26: hashicorp.consul.resource.Resource.MetadataEntry
-	nil,                               // 27: hashicorp.consul.resource.Resource.StatusEntry
-	(*anypb.Any)(nil),                 // 28: google.protobuf.Any
-	(*timestamppb.Timestamp)(nil),     // 29: google.protobuf.Timestamp
+	(*Type)(nil),                      // 1: hashicorp.consul.resource.Type
+	(*Tenancy)(nil),                   // 2: hashicorp.consul.resource.Tenancy
+	(*ID)(nil),                        // 3: hashicorp.consul.resource.ID
+	(*Resource)(nil),                  // 4: hashicorp.consul.resource.Resource
+	(*Status)(nil),                    // 5: hashicorp.consul.resource.Status
+	(*Condition)(nil),                 // 6: hashicorp.consul.resource.Condition
+	(*Reference)(nil),                 // 7: hashicorp.consul.resource.Reference
+	(*Tombstone)(nil),                 // 8: hashicorp.consul.resource.Tombstone
+	(*ReadRequest)(nil),               // 9: hashicorp.consul.resource.ReadRequest
+	(*ReadResponse)(nil),              // 10: hashicorp.consul.resource.ReadResponse
+	(*ListRequest)(nil),               // 11: hashicorp.consul.resource.ListRequest
+	(*ListResponse)(nil),              // 12: hashicorp.consul.resource.ListResponse
+	(*ListByOwnerRequest)(nil),        // 13: hashicorp.consul.resource.ListByOwnerRequest
+	(*ListByOwnerResponse)(nil),       // 14: hashicorp.consul.resource.ListByOwnerResponse
+	(*WriteRequest)(nil),              // 15: hashicorp.consul.resource.WriteRequest
+	(*WriteResponse)(nil),             // 16: hashicorp.consul.resource.WriteResponse
+	(*WriteStatusRequest)(nil),        // 17: hashicorp.consul.resource.WriteStatusRequest
+	(*WriteStatusResponse)(nil),       // 18: hashicorp.consul.resource.WriteStatusResponse
+	(*DeleteRequest)(nil),             // 19: hashicorp.consul.resource.DeleteRequest
+	(*DeleteResponse)(nil),            // 20: hashicorp.consul.resource.DeleteResponse
+	(*WatchListRequest)(nil),          // 21: hashicorp.consul.resource.WatchListRequest
+	(*WatchEvent)(nil),                // 22: hashicorp.consul.resource.WatchEvent
+	(*MutateAndValidateRequest)(nil),  // 23: hashicorp.consul.resource.MutateAndValidateRequest
+	(*MutateAndValidateResponse)(nil), // 24: hashicorp.consul.resource.MutateAndValidateResponse
+	nil,                               // 25: hashicorp.consul.resource.Resource.MetadataEntry
+	nil,                               // 26: hashicorp.consul.resource.Resource.StatusEntry
+	(*WatchEvent_Upsert)(nil),         // 27: hashicorp.consul.resource.WatchEvent.Upsert
+	(*WatchEvent_Delete)(nil),         // 28: hashicorp.consul.resource.WatchEvent.Delete
+	(*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
-	3,  // 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
-	4,  // 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
-	27, // 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
-	7,  // 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
+	1,  // 0: hashicorp.consul.resource.ID.type:type_name -> hashicorp.consul.resource.Type
+	2,  // 1: hashicorp.consul.resource.ID.tenancy:type_name -> hashicorp.consul.resource.Tenancy
+	3,  // 2: hashicorp.consul.resource.Resource.id:type_name -> hashicorp.consul.resource.ID
+	3,  // 3: hashicorp.consul.resource.Resource.owner:type_name -> hashicorp.consul.resource.ID
+	25, // 4: hashicorp.consul.resource.Resource.metadata:type_name -> hashicorp.consul.resource.Resource.MetadataEntry
+	26, // 5: hashicorp.consul.resource.Resource.status:type_name -> hashicorp.consul.resource.Resource.StatusEntry
+	30, // 6: hashicorp.consul.resource.Resource.data:type_name -> google.protobuf.Any
+	6,  // 7: hashicorp.consul.resource.Status.conditions:type_name -> hashicorp.consul.resource.Condition
+	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
-	2,  // 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
-	4,  // 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
-	5,  // 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
-	3,  // 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,  // 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
-	5,  // 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,  // 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,  // 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
-	2,  // 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
-	1,  // 29: hashicorp.consul.resource.WatchEvent.operation:type_name -> hashicorp.consul.resource.WatchEvent.Operation
-	5,  // 30: hashicorp.consul.resource.WatchEvent.resource:type_name -> hashicorp.consul.resource.Resource
-	5,  // 31: hashicorp.consul.resource.MutateAndValidateRequest.resource:type_name -> hashicorp.consul.resource.Resource
-	5,  // 32: hashicorp.consul.resource.MutateAndValidateResponse.resource:type_name -> hashicorp.consul.resource.Resource
-	6,  // 33: hashicorp.consul.resource.Resource.StatusEntry.value:type_name -> hashicorp.consul.resource.Status
-	10, // 34: hashicorp.consul.resource.ResourceService.Read:input_type -> hashicorp.consul.resource.ReadRequest
-	16, // 35: hashicorp.consul.resource.ResourceService.Write:input_type -> hashicorp.consul.resource.WriteRequest
-	18, // 36: hashicorp.consul.resource.ResourceService.WriteStatus:input_type -> hashicorp.consul.resource.WriteStatusRequest
-	12, // 37: hashicorp.consul.resource.ResourceService.List:input_type -> hashicorp.consul.resource.ListRequest
-	14, // 38: hashicorp.consul.resource.ResourceService.ListByOwner:input_type -> hashicorp.consul.resource.ListByOwnerRequest
-	20, // 39: hashicorp.consul.resource.ResourceService.Delete:input_type -> hashicorp.consul.resource.DeleteRequest
-	22, // 40: hashicorp.consul.resource.ResourceService.WatchList:input_type -> hashicorp.consul.resource.WatchListRequest
-	24, // 41: hashicorp.consul.resource.ResourceService.MutateAndValidate:input_type -> hashicorp.consul.resource.MutateAndValidateRequest
-	11, // 42: hashicorp.consul.resource.ResourceService.Read:output_type -> hashicorp.consul.resource.ReadResponse
-	17, // 43: hashicorp.consul.resource.ResourceService.Write:output_type -> hashicorp.consul.resource.WriteResponse
-	19, // 44: hashicorp.consul.resource.ResourceService.WriteStatus:output_type -> hashicorp.consul.resource.WriteStatusResponse
-	13, // 45: hashicorp.consul.resource.ResourceService.List:output_type -> hashicorp.consul.resource.ListResponse
-	15, // 46: hashicorp.consul.resource.ResourceService.ListByOwner:output_type -> hashicorp.consul.resource.ListByOwnerResponse
-	21, // 47: hashicorp.consul.resource.ResourceService.Delete:output_type -> hashicorp.consul.resource.DeleteResponse
-	23, // 48: hashicorp.consul.resource.ResourceService.WatchList:output_type -> hashicorp.consul.resource.WatchEvent
-	25, // 49: hashicorp.consul.resource.ResourceService.MutateAndValidate:output_type -> hashicorp.consul.resource.MutateAndValidateResponse
-	42, // [42:50] is the sub-list for method output_type
-	34, // [34:42] is the sub-list for method input_type
-	34, // [34:34] is the sub-list for extension type_name
-	34, // [34:34] is the sub-list for extension extendee
-	0,  // [0:34] is the sub-list for field type_name
+	7,  // 10: hashicorp.consul.resource.Condition.resource:type_name -> hashicorp.consul.resource.Reference
+	1,  // 11: hashicorp.consul.resource.Reference.type:type_name -> hashicorp.consul.resource.Type
+	2,  // 12: hashicorp.consul.resource.Reference.tenancy:type_name -> hashicorp.consul.resource.Tenancy
+	3,  // 13: hashicorp.consul.resource.Tombstone.owner:type_name -> hashicorp.consul.resource.ID
+	3,  // 14: hashicorp.consul.resource.ReadRequest.id:type_name -> hashicorp.consul.resource.ID
+	4,  // 15: hashicorp.consul.resource.ReadResponse.resource:type_name -> hashicorp.consul.resource.Resource
+	1,  // 16: hashicorp.consul.resource.ListRequest.type:type_name -> hashicorp.consul.resource.Type
+	2,  // 17: hashicorp.consul.resource.ListRequest.tenancy:type_name -> hashicorp.consul.resource.Tenancy
+	4,  // 18: hashicorp.consul.resource.ListResponse.resources:type_name -> hashicorp.consul.resource.Resource
+	3,  // 19: hashicorp.consul.resource.ListByOwnerRequest.owner:type_name -> hashicorp.consul.resource.ID
+	4,  // 20: hashicorp.consul.resource.ListByOwnerResponse.resources:type_name -> hashicorp.consul.resource.Resource
+	4,  // 21: hashicorp.consul.resource.WriteRequest.resource:type_name -> hashicorp.consul.resource.Resource
+	4,  // 22: hashicorp.consul.resource.WriteResponse.resource:type_name -> hashicorp.consul.resource.Resource
+	3,  // 23: hashicorp.consul.resource.WriteStatusRequest.id:type_name -> hashicorp.consul.resource.ID
+	5,  // 24: hashicorp.consul.resource.WriteStatusRequest.status:type_name -> hashicorp.consul.resource.Status
+	4,  // 25: hashicorp.consul.resource.WriteStatusResponse.resource:type_name -> hashicorp.consul.resource.Resource
+	3,  // 26: hashicorp.consul.resource.DeleteRequest.id:type_name -> hashicorp.consul.resource.ID
+	1,  // 27: hashicorp.consul.resource.WatchListRequest.type:type_name -> hashicorp.consul.resource.Type
+	2,  // 28: hashicorp.consul.resource.WatchListRequest.tenancy:type_name -> hashicorp.consul.resource.Tenancy
+	27, // 29: hashicorp.consul.resource.WatchEvent.upsert:type_name -> hashicorp.consul.resource.WatchEvent.Upsert
+	28, // 30: hashicorp.consul.resource.WatchEvent.delete:type_name -> hashicorp.consul.resource.WatchEvent.Delete
+	29, // 31: hashicorp.consul.resource.WatchEvent.end_of_snapshot:type_name -> hashicorp.consul.resource.WatchEvent.EndOfSnapshot
+	4,  // 32: hashicorp.consul.resource.MutateAndValidateRequest.resource:type_name -> hashicorp.consul.resource.Resource
+	4,  // 33: hashicorp.consul.resource.MutateAndValidateResponse.resource:type_name -> hashicorp.consul.resource.Resource
+	5,  // 34: hashicorp.consul.resource.Resource.StatusEntry.value:type_name -> hashicorp.consul.resource.Status
+	4,  // 35: hashicorp.consul.resource.WatchEvent.Upsert.resource:type_name -> hashicorp.consul.resource.Resource
+	4,  // 36: hashicorp.consul.resource.WatchEvent.Delete.resource:type_name -> hashicorp.consul.resource.Resource
+	9,  // 37: hashicorp.consul.resource.ResourceService.Read:input_type -> hashicorp.consul.resource.ReadRequest
+	15, // 38: hashicorp.consul.resource.ResourceService.Write:input_type -> hashicorp.consul.resource.WriteRequest
+	17, // 39: hashicorp.consul.resource.ResourceService.WriteStatus:input_type -> hashicorp.consul.resource.WriteStatusRequest
+	11, // 40: hashicorp.consul.resource.ResourceService.List:input_type -> hashicorp.consul.resource.ListRequest
+	13, // 41: hashicorp.consul.resource.ResourceService.ListByOwner:input_type -> hashicorp.consul.resource.ListByOwnerRequest
+	19, // 42: hashicorp.consul.resource.ResourceService.Delete:input_type -> hashicorp.consul.resource.DeleteRequest
+	21, // 43: hashicorp.consul.resource.ResourceService.WatchList:input_type -> hashicorp.consul.resource.WatchListRequest
+	23, // 44: hashicorp.consul.resource.ResourceService.MutateAndValidate:input_type -> hashicorp.consul.resource.MutateAndValidateRequest
+	10, // 45: hashicorp.consul.resource.ResourceService.Read:output_type -> hashicorp.consul.resource.ReadResponse
+	16, // 46: hashicorp.consul.resource.ResourceService.Write:output_type -> hashicorp.consul.resource.WriteResponse
+	18, // 47: hashicorp.consul.resource.ResourceService.WriteStatus:output_type -> hashicorp.consul.resource.WriteStatusResponse
+	12, // 48: hashicorp.consul.resource.ResourceService.List:output_type -> hashicorp.consul.resource.ListResponse
+	14, // 49: hashicorp.consul.resource.ResourceService.ListByOwner:output_type -> hashicorp.consul.resource.ListByOwnerResponse
+	20, // 50: hashicorp.consul.resource.ResourceService.Delete:output_type -> hashicorp.consul.resource.DeleteResponse
+	22, // 51: hashicorp.consul.resource.ResourceService.WatchList:output_type -> hashicorp.consul.resource.WatchEvent
+	24, // 52: hashicorp.consul.resource.ResourceService.MutateAndValidate:output_type -> hashicorp.consul.resource.MutateAndValidateResponse
+	45, // [45:53] is the sub-list for method output_type
+	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,
-			NumMessages:   26,
+			NumEnums:      1,
+			NumMessages:   29,
 			NumExtensions: 0,
 			NumServices:   1,
 		},
diff --git a/proto-public/pbresource/resource.proto b/proto-public/pbresource/resource.proto
index 1a36492265..29328e587a 100644
--- a/proto-public/pbresource/resource.proto
+++ b/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.
-  enum Operation {
-    // OPERATION_UNSPECIFIED is the default/zero value. You should not see it
-    // in practice.
-    OPERATION_UNSPECIFIED = 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.
-    OPERATION_UPSERT = 1;
-
-    // OPERATION_DELETED indicates that the resource was deleted.
-    OPERATION_DELETE = 2;
+  // 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.
+  message Upsert {
+    Resource resource = 1;
   }
 
-  // Operation describes the type of event.
-  Operation operation = 1;
+  // Delete indicates that the resource was deleted.
+  message Delete {
+    Resource resource = 1;
+  }
 
-  // Resource the event relates to.
-  Resource resource = 2;
+  // EndOfSnapshot is sent to indicate that the initial snapshot events have
+  // 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.
diff --git a/proto-public/pbresource/resource_deepcopy.gen.go b/proto-public/pbresource/resource_deepcopy.gen.go
index b964d036d0..e9a0b01bc1 100644
--- a/proto-public/pbresource/resource_deepcopy.gen.go
+++ b/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)
diff --git a/proto-public/pbresource/resource_json.gen.go b/proto-public/pbresource/resource_json.gen.go
index a09fe1cf26..b30b535ff7 100644
--- a/proto-public/pbresource/resource_json.gen.go
+++ b/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)