fix: ensure that controller conformance tests work over net too

go.mod

@@ -47,4 +47,7 @@ require (
 	google.golang.org/genproto/googleapis/rpc v0.0.0-20241202173237-19429a94021a // indirect
 )
 
-retract v0.4.7 // Wait with locked mutex leads to the deadlock
+retract (
+	v0.7.3 // Typo in the test type result
+	v0.4.7 // Wait with locked mutex leads to the deadlock
+)

pkg/controller/conformance/resources.go

@@ -61,7 +61,7 @@ func NewStrResource(ns resource.Namespace, id resource.ID, value string) *StrRes
 type strSpec struct{ ValueGetSet[string] } //nolint:recvcheck
 
 func (s *strSpec) FromProto(bytes []byte)       { s.value = string(bytes) }
-func (s strSpec) MarshalProto() ([]byte, error) { return []byte(s.value + "stuff"), nil }
+func (s strSpec) MarshalProto() ([]byte, error) { return []byte(s.value), nil }
 
 // SentenceResourceType is the type of SentenceResource.
 const SentenceResourceType = resource.Type("test/sentence")

pkg/controller/conformance/runtime.go

@@ -9,6 +9,7 @@ import (
 	"expvar"
 	"fmt"
 	"math/rand"
+	"strings"
 	"sync"
 	"time"
 
@@ -31,8 +32,8 @@ type RuntimeSuite struct { //nolint:govet
 
 	Runtime controller.Engine
 
-	SetupRuntime    func()
-	TearDownRuntime func()
+	SetupRuntime    func(suite *RuntimeSuite)
+	TearDownRuntime func(suite *RuntimeSuite)
 
 	MetricsReadCacheEnabled bool
 
@@ -42,12 +43,15 @@ type RuntimeSuite struct { //nolint:govet
 	ctxCancel context.CancelFunc
 }
 
+// Context provides the context for the test suite.
+func (suite *RuntimeSuite) Context() context.Context { return suite.ctx }
+
 // SetupTest ...
 func (suite *RuntimeSuite) SetupTest() {
 	suite.ctx, suite.ctxCancel = context.WithTimeout(context.Background(), 3*time.Minute)
 
 	if suite.SetupRuntime != nil {
-		suite.SetupRuntime()
+		suite.SetupRuntime(suite)
 	}
 }
 
@@ -57,7 +61,12 @@ func (suite *RuntimeSuite) startRuntime(ctx context.Context) {
 	go func() {
 		defer suite.wg.Done()
 
-		suite.Assert().NoError(suite.Runtime.Run(ctx))
+		err := suite.Runtime.Run(ctx)
+		// we can safely ignore canceled error,
+		// but we can't check for it using errors.Is because it's a rpc error
+		if err != nil && !strings.Contains(err.Error(), "context canceled") {
+			suite.Assert().NoError(err)
+		}
 	}()
 }
 
@@ -161,7 +170,7 @@ func (suite *RuntimeSuite) TearDownTest() {
 	suite.Assert().NoError(suite.State.Create(context.Background(), NewSentenceResource("sentences", "xxx", "")))
 
 	if suite.TearDownRuntime != nil {
-		suite.TearDownRuntime()
+		suite.TearDownRuntime(suite)
 	}
 }
 

pkg/controller/runtime/runtime_test.go

@@ -6,31 +6,51 @@ package runtime_test
 
 import (
 	"context"
+	"net"
 	goruntime "runtime"
 	"strconv"
 	"testing"
 	"time"
 
+	"github.com/siderolabs/gen/xtesting/must"
 	"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"
+	"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"
 	"github.com/cosi-project/runtime/pkg/controller/runtime"
 	"github.com/cosi-project/runtime/pkg/controller/runtime/options"
 	"github.com/cosi-project/runtime/pkg/future"
 	"github.com/cosi-project/runtime/pkg/resource"
+	"github.com/cosi-project/runtime/pkg/resource/protobuf"
 	"github.com/cosi-project/runtime/pkg/safe"
 	"github.com/cosi-project/runtime/pkg/state"
 	stateconformance "github.com/cosi-project/runtime/pkg/state/conformance"
 	"github.com/cosi-project/runtime/pkg/state/impl/inmem"
 	"github.com/cosi-project/runtime/pkg/state/impl/namespaced"
+	"github.com/cosi-project/runtime/pkg/state/protobuf/client"
+	"github.com/cosi-project/runtime/pkg/state/protobuf/server"
 )
 
+func noError(err error) {
+	if err != nil {
+		panic(err)
+	}
+}
+
+func init() {
+	noError(protobuf.RegisterResource(conformance.IntResourceType, &conformance.IntResource{}))
+	noError(protobuf.RegisterResource(conformance.StrResourceType, &conformance.StrResource{}))
+	noError(protobuf.RegisterResource(conformance.SentenceResourceType, &conformance.SentenceResource{}))
+}
+
 func TestRuntimeConformance(t *testing.T) {
-	for _, tt := range []struct {
+	tests := []struct {
 		name                    string
 		opts                    []options.Option
 		metricsReadCacheEnabled bool
@@ -68,25 +88,63 @@ func TestRuntimeConformance(t *testing.T) {
 				options.WithWarnOnUncachedReads(true),
 			},
 		},
-	} {
+	}
+
+	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
 			t.Cleanup(func() { goleak.VerifyNone(t, goleak.IgnoreCurrent()) })
 
-			suite := &conformance.RuntimeSuite{
+			suiterunner.Run(t, &conformance.RuntimeSuite{
 				MetricsReadCacheEnabled: tt.metricsReadCacheEnabled,
-			}
-			suite.SetupRuntime = func() {
-				suite.State = state.WrapCore(namespaced.NewState(inmem.Build))
+				SetupRuntime: func(rs *conformance.RuntimeSuite) {
+					rs.State = state.WrapCore(namespaced.NewState(inmem.Build))
+					logger := zaptest.NewLogger(rs.T())
+					rs.Runtime = must.Value(runtime.NewRuntime(rs.State, logger, tt.opts...))(rs.T())
+				},
+			})
+		})
+	}
 
-				var err error
+	const listenOn = "127.0.0.1:0"
 
-				logger := zaptest.NewLogger(t)
+	t.Log("testing networked runtime")
 
-				suite.Runtime, err = runtime.NewRuntime(suite.State, logger, tt.opts...)
-				suite.Require().NoError(err)
-			}
+	for _, tt := range tests {
+		t.Run(tt.name+"_over-network", func(t *testing.T) {
+			t.Cleanup(func() { goleak.VerifyNone(t, goleak.IgnoreCurrent()) })
+
+			suiterunner.Run(t, &conformance.RuntimeSuite{
+				MetricsReadCacheEnabled: tt.metricsReadCacheEnabled,
+				SetupRuntime: func(rs *conformance.RuntimeSuite) {
+					l := must.Value(net.Listen("tcp", listenOn))(rs.T())
+
+					grpcServer := grpc.NewServer()
+					inmemState := state.WrapCore(namespaced.NewState(inmem.Build))
+					v1alpha1.RegisterStateServer(grpcServer, server.NewState(inmemState))
+
+					go func() { assert.NoError(rs.T(), grpcServer.Serve(l)) }()
+
+					rs.T().Cleanup(func() { grpcServer.Stop() })
+
+					grpcConn := must.Value(grpc.NewClient(
+						l.Addr().String(),
+						grpc.WithTransportCredentials(insecure.NewCredentials()),
+					))(rs.T())
+
+					rs.T().Cleanup(func() { assert.NoError(rs.T(), grpcConn.Close()) })
+
+					stateClient := v1alpha1.NewStateClient(grpcConn)
+					rs.State = state.WrapCore(client.NewAdapter(stateClient))
+
+					must.Value(rs.State.List(rs.Context(), conformance.NewIntResource("default", "zero", 0).Metadata()))(rs.T())
 
-			suiterunner.Run(t, suite)
+					rs.Runtime = must.Value(runtime.NewRuntime(
+						rs.State,
+						zaptest.NewLogger(rs.T()),
+						tt.opts...,
+					))(rs.T())
+				},
+			})
 		})
 	}
 }