feat: support new Bootstrapped and Errored watch events

In the regular COSI, new event types are enabled immediately.

With protobuf client, new types are enabled when using new COSI client,
old COSI client against new server won't receive new events, as old
implementation panics on nil Resource in the event structure.

Fixes #175

Signed-off-by: Andrey Smirnov <andrey.smirnov@talos-systems.com>

Author: smira

.kres.yaml

@@ -5,19 +5,19 @@ spec:
     - --experimental_allow_proto3_optional
   vtProtobufEnabled: true
   specs:
-    - source: https://raw.githubusercontent.com/cosi-project/specification/c0f4e7ba6f9a4c7b857b7d110bff3dc5d523cacf/proto/v1alpha1/resource.proto
+    - source: https://raw.githubusercontent.com/cosi-project/specification/aec70d02f588efc9463b85e551057313c9754dda/proto/v1alpha1/resource.proto
       subdirectory: v1alpha1/
       genGateway: true
       external: false
-    - source: https://raw.githubusercontent.com/cosi-project/specification/c0f4e7ba6f9a4c7b857b7d110bff3dc5d523cacf/proto/v1alpha1/state.proto
+    - source: https://raw.githubusercontent.com/cosi-project/specification/aec70d02f588efc9463b85e551057313c9754dda/proto/v1alpha1/state.proto
       subdirectory: v1alpha1/
       genGateway: true
       external: false
-    - source: https://raw.githubusercontent.com/cosi-project/specification/c0f4e7ba6f9a4c7b857b7d110bff3dc5d523cacf/proto/v1alpha1/runtime.proto
+    - source: https://raw.githubusercontent.com/cosi-project/specification/aec70d02f588efc9463b85e551057313c9754dda/proto/v1alpha1/runtime.proto
       subdirectory: v1alpha1/
       genGateway: true
       external: false
-    - source: https://raw.githubusercontent.com/cosi-project/specification/c0f4e7ba6f9a4c7b857b7d110bff3dc5d523cacf/proto/v1alpha1/meta.proto
+    - source: https://raw.githubusercontent.com/cosi-project/specification/aec70d02f588efc9463b85e551057313c9754dda/proto/v1alpha1/meta.proto
       subdirectory: v1alpha1/
       genGateway: true
       external: false

Dockerfile

@@ -2,7 +2,7 @@
 
 # THIS FILE WAS AUTOMATICALLY GENERATED, PLEASE DO NOT EDIT.
 #
-# Generated on 2022-11-21T12:24:35Z by kres latest.
+# Generated on 2022-11-25T18:59:30Z by kres 3ac53a8.
 
 ARG TOOLCHAIN
 
@@ -21,10 +21,10 @@ RUN markdownlint --ignore "CHANGELOG.md" --ignore "**/node_modules/**" --ignore
 
 # collects proto specs
 FROM scratch AS proto-specs
-ADD https://raw.githubusercontent.com/cosi-project/specification/c0f4e7ba6f9a4c7b857b7d110bff3dc5d523cacf/proto/v1alpha1/resource.proto /api/v1alpha1/
-ADD https://raw.githubusercontent.com/cosi-project/specification/c0f4e7ba6f9a4c7b857b7d110bff3dc5d523cacf/proto/v1alpha1/state.proto /api/v1alpha1/
-ADD https://raw.githubusercontent.com/cosi-project/specification/c0f4e7ba6f9a4c7b857b7d110bff3dc5d523cacf/proto/v1alpha1/runtime.proto /api/v1alpha1/
-ADD https://raw.githubusercontent.com/cosi-project/specification/c0f4e7ba6f9a4c7b857b7d110bff3dc5d523cacf/proto/v1alpha1/meta.proto /api/v1alpha1/
+ADD https://raw.githubusercontent.com/cosi-project/specification/aec70d02f588efc9463b85e551057313c9754dda/proto/v1alpha1/resource.proto /api/v1alpha1/
+ADD https://raw.githubusercontent.com/cosi-project/specification/aec70d02f588efc9463b85e551057313c9754dda/proto/v1alpha1/state.proto /api/v1alpha1/
+ADD https://raw.githubusercontent.com/cosi-project/specification/aec70d02f588efc9463b85e551057313c9754dda/proto/v1alpha1/runtime.proto /api/v1alpha1/
+ADD https://raw.githubusercontent.com/cosi-project/specification/aec70d02f588efc9463b85e551057313c9754dda/proto/v1alpha1/meta.proto /api/v1alpha1/
 ADD api/key_storage/key_storage.proto /api/key_storage/
 
 # base toolchain image

api/v1alpha1/state.pb.go

@@ -14,7 +14,7 @@ require (
 	github.com/grpc-ecosystem/grpc-gateway/v2 v2.13.0
 	github.com/hashicorp/go-memdb v1.3.4
 	github.com/hashicorp/go-multierror v1.1.1
-	github.com/siderolabs/gen v0.4.0
+	github.com/siderolabs/gen v0.4.1
 	github.com/siderolabs/go-pointer v1.0.0
 	github.com/siderolabs/go-retry v0.3.2
 	github.com/siderolabs/protoenc v0.2.0
@@ -31,6 +31,7 @@ require (
 require (
 	github.com/ProtonMail/go-crypto v0.0.0-20220822140716-1678d6eb0cbe // indirect
 	github.com/ProtonMail/go-mime v0.0.0-20220302105931-303f85f7fe0f // indirect
+	github.com/benbjohnson/clock v1.1.0 // indirect
 	github.com/cloudflare/circl v1.1.0 // indirect
 	github.com/davecgh/go-spew v1.1.1 // indirect
 	github.com/hashicorp/errwrap v1.0.0 // indirect

api/v1alpha1/state_vtproto.pb.go

@@ -6,6 +6,7 @@ github.com/ProtonMail/go-mime v0.0.0-20220302105931-303f85f7fe0f/go.mod h1:NYt+V
 github.com/ProtonMail/gopenpgp/v2 v2.4.10 h1:EYgkxzwmQvsa6kxxkgP1AwzkFqKHscF2UINxaSn6rdI=
 github.com/ProtonMail/gopenpgp/v2 v2.4.10/go.mod h1:CTRA7/toc/4DxDy5Du4hPDnIZnJvXSeQ8LsRTOUJoyc=
 github.com/benbjohnson/clock v1.1.0 h1:Q92kusRqC1XV2MjkWETPvjJVqKetz1OzxZB7mHJLju8=
+github.com/benbjohnson/clock v1.1.0/go.mod h1:J11/hYXuz8f4ySSvYwY0FKfm+ezbsZBKZxNJlLklBHA=
 github.com/brianvoe/gofakeit/v6 v6.17.0 h1:obbQTJeHfktJtiZzq0Q1bEpsNUs+yHrYlPVWt7BtmJ4=
 github.com/bwesterb/go-ristretto v1.2.0/go.mod h1:fUIoIZaG73pV5biE2Blr2xEzDoMj7NFEuV9ekS419A0=
 github.com/cenkalti/backoff/v4 v4.2.0 h1:HN5dHm3WBOgndBH6E8V0q2jIYIR3s9yglV8k/+MN3u4=
@@ -48,8 +49,8 @@ github.com/pkg/errors v0.9.1 h1:FEBLx1zS214owpjy7qsBeixbURkuhQAwrK5UwLGTwt4=
 github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
 github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
 github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
-github.com/siderolabs/gen v0.4.0 h1:t+f2TimPoAGBNeXW3qLQBlbDC5GnFik+E/4PWL/uf9c=
-github.com/siderolabs/gen v0.4.0/go.mod h1:vzcXVRNpo9j2tyQJU+9ZQ1J6yoebX9KL1TkHt18HNqw=
+github.com/siderolabs/gen v0.4.1 h1:NoPmxjLFwOGRFaKypk2hoT5/XIzoM6AhF0eqy61bQLk=
+github.com/siderolabs/gen v0.4.1/go.mod h1:wS8tFq7sn5vqKAuyS30vJUig3tX5v6q79VG4KfUnILM=
 github.com/siderolabs/go-pointer v1.0.0 h1:6TshPKep2doDQJAAtHUuHWXbca8ZfyRySjSBT/4GsMU=
 github.com/siderolabs/go-pointer v1.0.0/go.mod h1:HTRFUNYa3R+k0FFKNv11zgkaCLzEkWVzoYZ433P3kHc=
 github.com/siderolabs/go-retry v0.3.2 h1:FzWslFm4y8RY1wU0gIskm0oZHOpsSibZqlR8N8/k4Eo=

go.mod

@@ -12,6 +12,7 @@ import (
 	"github.com/stretchr/testify/require"
 	suiterunner "github.com/stretchr/testify/suite"
 	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials/insecure"
 
 	"github.com/cosi-project/runtime/api/v1alpha1"
 	"github.com/cosi-project/runtime/pkg/controller/conformance"
@@ -72,7 +73,7 @@ func TestProtobufConformance(t *testing.T) {
 			suite.grpcServer.Serve(l) //nolint:errcheck
 		}()
 
-		suite.grpcConn, err = grpc.Dial("unix://"+suite.sock.Name(), grpc.WithInsecure()) //nolint:staticcheck
+		suite.grpcConn, err = grpc.Dial("unix://"+suite.sock.Name(), grpc.WithTransportCredentials(insecure.NewCredentials()))
 		require.NoError(t, err)
 
 		stateClient := v1alpha1.NewStateClient(suite.grpcConn)

go.sum

@@ -29,16 +29,18 @@ type Runtime struct { //nolint:govet
 	state  state.State
 	logger *zap.Logger
 
-	watchCh   chan state.Event
-	watchedMu sync.Mutex
-	watched   map[watchKey]struct{}
+	watchCh     chan state.Event
+	watchErrors chan error
+	watchedMu   sync.Mutex
+	watched     map[watchKey]struct{}
 
 	controllersMu      sync.RWMutex
 	controllersCond    *sync.Cond
 	controllersRunning int
 	controllers        map[string]*adapter
 
-	runCtx context.Context //nolint:containedctx
+	runCtx       context.Context //nolint:containedctx
+	runCtxCancel context.CancelFunc
 }
 
 type watchKey struct {
@@ -53,6 +55,7 @@ func NewRuntime(st state.State, logger *zap.Logger) (*Runtime, error) {
 		logger:      logger,
 		controllers: make(map[string]*adapter),
 		watchCh:     make(chan state.Event),
+		watchErrors: make(chan error, 1),
 		watched:     make(map[watchKey]struct{}),
 	}
 
@@ -131,9 +134,11 @@ func (runtime *Runtime) Run(ctx context.Context) error {
 			return fmt.Errorf("runtime has already been started")
 		}
 
-		runtime.runCtx = ctx
+		runtime.runCtx, runtime.runCtxCancel = context.WithCancel(ctx)
 
 		if err := runtime.setupWatches(); err != nil {
+			runtime.runCtxCancel()
+
 			return err
 		}
 
@@ -163,17 +168,25 @@ func (runtime *Runtime) Run(ctx context.Context) error {
 		return err
 	}
 
-	<-runtime.runCtx.Done()
+	var watchErr error
+
+	select {
+	case <-runtime.runCtx.Done():
+	case watchErr = <-runtime.watchErrors:
+		watchErr = fmt.Errorf("controller runtime watch error: %w", watchErr)
+	}
 
 	runtime.controllersMu.Lock()
 
+	runtime.runCtxCancel()
+
 	for runtime.controllersRunning > 0 {
 		runtime.controllersCond.Wait()
 	}
 
 	runtime.controllersMu.Unlock()
 
-	return nil
+	return watchErr
 }
 
 // GetDependencyGraph returns dependency graph between resources and controllers.
@@ -231,6 +244,13 @@ func (runtime *Runtime) processWatched() {
 		case e = <-runtime.watchCh:
 		}
 
+		if e.Type == state.Errored {
+			// watch failed, we need to abort
+			runtime.watchErrors <- e.Error
+
+			return
+		}
+
 		md := e.Resource.Metadata()
 
 		controllers, err := runtime.depDB.GetDependentControllers(controller.Input{

pkg/controller/protobuf/protobuf_test.go

@@ -5,22 +5,25 @@
 package runtime_test
 
 import (
+	"context"
+	"strconv"
 	"testing"
+	"time"
 
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 	suiterunner "github.com/stretchr/testify/suite"
 	"go.uber.org/goleak"
+	"go.uber.org/zap/zaptest"
 
 	"github.com/cosi-project/runtime/pkg/controller/conformance"
 	"github.com/cosi-project/runtime/pkg/controller/runtime"
-	"github.com/cosi-project/runtime/pkg/logging"
 	"github.com/cosi-project/runtime/pkg/state"
 	"github.com/cosi-project/runtime/pkg/state/impl/inmem"
 	"github.com/cosi-project/runtime/pkg/state/impl/namespaced"
 )
 
 func TestRuntimeConformance(t *testing.T) {
-	t.Parallel()
-
 	defer goleak.VerifyNone(t, goleak.IgnoreCurrent())
 
 	suite := &conformance.RuntimeSuite{}
@@ -29,11 +32,50 @@ func TestRuntimeConformance(t *testing.T) {
 
 		var err error
 
-		logger := logging.DefaultLogger()
+		logger := zaptest.NewLogger(t)
 
 		suite.Runtime, err = runtime.NewRuntime(suite.State, logger)
 		suite.Require().NoError(err)
 	}
 
 	suiterunner.Run(t, suite)
 }
+
+func TestRuntimeWatchError(t *testing.T) {
+	defer goleak.VerifyNone(t, goleak.IgnoreCurrent())
+
+	// create a state with tiny capacity
+	st := state.WrapCore(namespaced.NewState(func(ns string) state.CoreState {
+		return inmem.NewStateWithOptions(inmem.WithHistoryCapacity(10), inmem.WithHistoryGap(5))(ns)
+	}))
+
+	logger := zaptest.NewLogger(t)
+	runtime, err := runtime.NewRuntime(st, logger)
+	require.NoError(t, err)
+
+	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+	t.Cleanup(cancel)
+
+	errCh := make(chan error)
+
+	go func() {
+		errCh <- runtime.Run(ctx)
+	}()
+
+	require.NoError(t, runtime.RegisterController(&conformance.IntToStrController{
+		SourceNamespace: "default",
+		TargetNamespace: "default",
+	}))
+
+	// overfill the history buffer
+	for i := 0; i < 10000; i++ {
+		require.NoError(t, st.Create(ctx, conformance.NewIntResource("default", strconv.Itoa(i), i)))
+	}
+
+	err = <-errCh
+	require.Error(t, err)
+
+	assert.EqualError(t, err, "controller runtime watch error: buffer overrun")
+
+	cancel()
+}