feat: rework inmem resource state implementation

.gitignore

@@ -1,5 +1,7 @@
-# THIS FILE WAS AUTOMATICALLY GENERATED, PLEASE DO NOT EDIT.
+# THIS FILE WAS AUTOMATICALLY GENERATED BY KRES, PLEASE DO NOT EDIT.
 #
-# Generated on 2020-08-28T20:43:11Z by kres 292ed36-dirty.
+# Generated on 2026-08-04T10:58:40Z by kres 93d28ca.
 
 _out
+*.local.md
+*.override.md

Makefile

@@ -1,6 +1,6 @@
 # THIS FILE WAS AUTOMATICALLY GENERATED BY KRES, PLEASE DO NOT EDIT.
 #
-# Generated on 2026-07-15T06:26:20Z by kres d66531d.
+# Generated on 2026-08-04T10:58:40Z by kres 93d28ca.
 
 # common variables
 
@@ -39,6 +39,7 @@ GOTOOLCHAIN ?= local
 GOEXPERIMENT ?=
 GO_BUILDFLAGS += -tags $(GO_BUILDTAGS)
 TESTPKGS ?= ./...
+IMAGE_SIGNER_IMAGE ?= ghcr.io/siderolabs/image-signer:v0.3.2
 KRES_IMAGE ?= ghcr.io/siderolabs/kres:latest
 CONFORMANCE_IMAGE ?= ghcr.io/siderolabs/conform:latest
 
@@ -250,6 +251,19 @@ lint-fmt: lint-golangci-lint-fmt  ## Run all linter formatters and fix up the so
 image-runtime:  ## Builds image for runtime.
 	@$(MAKE) registry-$@ IMAGE_NAME="runtime"
 
+.PHONY: sign-image-runtime
+sign-image-runtime:  ## Signs the image for runtime. Requires interactive Google authentication.
+	@test -n "$$GITHUB_TOKEN" || { \
+	  echo 'GITHUB_TOKEN with write:packages scope must be set: gh auth refresh -s write:packages, then GITHUB_TOKEN=$$(gh auth token) make' $@; \
+	  exit 1; \
+	}
+	@TMP=$$(mktemp -d) && trap 'rm -rf "$$TMP"' EXIT && \
+	  printf '{"auths":{"$(REGISTRY)":{"username":"x","password":"%s"}}}' "$$GITHUB_TOKEN" > "$$TMP/config.json" && \
+	  docker run --rm -p 127.0.0.1:8585:8585 \
+	    -v "$$TMP:/dc:ro" -e DOCKER_CONFIG=/dc \
+	    $(IMAGE_SIGNER_IMAGE) sign --timeout=15m \
+	    $(REGISTRY)/$(USERNAME)/runtime:$(IMAGE_TAG)
+
 .PHONY: rekres
 rekres:
 	@docker pull $(KRES_IMAGE)

cmd/runtime/main.go

@@ -30,7 +30,6 @@ import (
 	"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"
 	"github.com/cosi-project/runtime/pkg/state/protobuf/server"
 )
 
@@ -77,7 +76,7 @@ func run() error {
 		return fmt.Errorf("failed to listen on network address: %w", err)
 	}
 
-	inmemState := state.WrapCore(namespaced.NewState(inmem.Build))
+	inmemState := state.WrapCore(inmem.NewState())
 
 	logger := logging.DefaultLogger()
 

go.mod

@@ -9,7 +9,7 @@ require (
 	github.com/grpc-ecosystem/grpc-gateway/v2 v2.30.0
 	github.com/hashicorp/go-multierror v1.1.1
 	github.com/klauspost/compress v1.19.2
-	github.com/planetscale/vtprotobuf v0.6.1-0.20240319094008-0393e58bdf10
+	github.com/planetscale/vtprotobuf v0.6.1-0.20260702190614-8ae5a48058df
 	github.com/siderolabs/gen v0.8.7
 	github.com/siderolabs/go-pointer v1.0.1
 	github.com/siderolabs/go-retry v0.3.3

go.sum

@@ -37,8 +37,8 @@ github.com/kr/pretty v0.2.1 h1:Fmg33tUaq4/8ym9TJN1x7sLJnHVwhP33CNkpYV/7rwI=
 github.com/kr/pretty v0.2.1/go.mod h1:ipq/a2n7PKx3OHsz4KJII5eveXtPO4qwEXGdVfWzfnI=
 github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY=
 github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE=
-github.com/planetscale/vtprotobuf v0.6.1-0.20240319094008-0393e58bdf10 h1:GFCKgmp0tecUJ0sJuv4pzYCqS9+RGSn52M3FUwPs+uo=
-github.com/planetscale/vtprotobuf v0.6.1-0.20240319094008-0393e58bdf10/go.mod h1:t/avpk3KcrXxUnYOhZhMXJlSEyie6gQbtLq5NM3loB8=
+github.com/planetscale/vtprotobuf v0.6.1-0.20260702190614-8ae5a48058df h1:x2ymdov8jnZLDPfI+VVcf/ZvzuZ2u36ieXTuQASMkWI=
+github.com/planetscale/vtprotobuf v0.6.1-0.20260702190614-8ae5a48058df/go.mod h1:araspv2uYKozbi5lrKaqpv1/Uei7eQSml8JCw2A3IRg=
 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.8.7 h1:Nu31kL0ln/facRHBfNX7zcB7w9VZ9tifXKsP4lUtWHw=

pkg/controller/generic/cleanup/cleanup_test.go

@@ -22,13 +22,12 @@ import (
 	"github.com/cosi-project/runtime/pkg/resource/rtestutils"
 	"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 runTest(t *testing.T, f func(ctx context.Context, t *testing.T, st state.State, rt *runtime.Runtime)) {
 	defer goleak.VerifyNone(t, goleak.IgnoreCurrent())
 
-	st := state.WrapCore(namespaced.NewState(inmem.Build))
+	st := state.WrapCore(inmem.NewState())
 
 	logger := logging.DefaultLogger()
 

pkg/controller/generic/destroy/destroy_test.go

@@ -26,7 +26,6 @@ import (
 	"github.com/cosi-project/runtime/pkg/resource/typed"
 	"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"
 )
 
 // ANamespaceName is the namespace of A resource.
@@ -68,7 +67,7 @@ func (a ASpec) DeepCopy() ASpec {
 func runTest(t *testing.T, f func(ctx context.Context, t *testing.T, st state.State, rt *runtime.Runtime)) {
 	defer goleak.VerifyNone(t, goleak.IgnoreCurrent())
 
-	st := state.WrapCore(namespaced.NewState(inmem.Build))
+	st := state.WrapCore(inmem.NewState())
 
 	logger := logging.DefaultLogger()
 

pkg/controller/generic/qtransform/qtransform_test.go

@@ -34,7 +34,6 @@ import (
 	"github.com/cosi-project/runtime/pkg/safe"
 	"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"
 )
 
 type ABController = qtransform.QController[*A, *B]
@@ -891,7 +890,7 @@ func TestMappedByLabelInput(t *testing.T) {
 func setup(t *testing.T, f func(ctx context.Context, st state.State, rt *runtime.Runtime), opts ...options.Option) {
 	defer goleak.VerifyNone(t, goleak.IgnoreCurrent())
 
-	st := state.WrapCore(namespaced.NewState(inmem.Build))
+	st := state.WrapCore(inmem.NewState())
 
 	logger := logging.DefaultLogger()
 

pkg/controller/generic/transform/transform_test.go

@@ -30,7 +30,6 @@ import (
 	"github.com/cosi-project/runtime/pkg/safe"
 	"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"
 )
 
 type ABController = transform.Controller[*A, *B]
@@ -639,7 +638,7 @@ func TestHooks(t *testing.T) {
 func setup(t *testing.T, f func(ctx context.Context, st state.State, rt *runtime.Runtime)) {
 	defer goleak.VerifyNone(t, goleak.IgnoreCurrent())
 
-	st := state.WrapCore(namespaced.NewState(inmem.Build))
+	st := state.WrapCore(inmem.NewState())
 
 	logger := logging.DefaultLogger()
 

pkg/controller/runtime/runtime_test.go

@@ -32,7 +32,6 @@ import (
 	"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"
 )
@@ -97,7 +96,7 @@ func TestRuntimeConformance(t *testing.T) {
 			suiterunner.Run(t, &conformance.RuntimeSuite{
 				MetricsReadCacheEnabled: tt.metricsReadCacheEnabled,
 				SetupRuntime: func(rs *conformance.RuntimeSuite) {
-					rs.State = state.WrapCore(namespaced.NewState(inmem.Build))
+					rs.State = state.WrapCore(inmem.NewState())
 					logger := zaptest.NewLogger(rs.T())
 					rs.Runtime = must.Value(runtime.NewRuntime(rs.State, logger, tt.opts...))(rs.T())
 				},
@@ -119,7 +118,7 @@ func TestRuntimeConformance(t *testing.T) {
 					l := must.Value((&net.ListenConfig{}).Listen(t.Context(), "tcp", listenOn))(rs.T())
 
 					grpcServer := grpc.NewServer()
-					inmemState := state.WrapCore(namespaced.NewState(inmem.Build))
+					inmemState := state.WrapCore(inmem.NewState())
 					v1alpha1.RegisterStateServer(grpcServer, server.NewState(inmemState))
 
 					go func() { assert.NoError(rs.T(), grpcServer.Serve(l)) }()
@@ -153,12 +152,11 @@ 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.WithHistoryMaxCapacity(10),
-			inmem.WithHistoryGap(5),
-		)(ns)
-	}))
+	st := state.WrapCore(inmem.NewStateWithOptions(
+		inmem.WithHistoryInitialCapacity(10),
+		inmem.WithHistoryMaxCapacity(10),
+		inmem.WithHistoryGap(5),
+	))
 
 	logger := zaptest.NewLogger(t)
 	rt, err := runtime.NewRuntime(st, logger)
@@ -192,7 +190,7 @@ func TestRuntimeWatchOverrun(t *testing.T) {
 
 	defer goleak.VerifyNone(t, goleak.IgnoreCurrent())
 
-	st := state.WrapCore(namespaced.NewState(inmem.Build))
+	st := state.WrapCore(inmem.NewState())
 
 	logger := zaptest.NewLogger(t)
 	rt, err := runtime.NewRuntime(st, logger)
@@ -245,7 +243,7 @@ func TestRuntimeWatchOverrun(t *testing.T) {
 func TestRuntimeCachedState(t *testing.T) {
 	defer goleak.VerifyNone(t, goleak.IgnoreCurrent())
 
-	st := state.WrapCore(namespaced.NewState(inmem.Build))
+	st := state.WrapCore(inmem.NewState())
 
 	logger := zaptest.NewLogger(t)
 	rt, err := runtime.NewRuntime(st, logger, options.WithCachedResource("cached", conformance.IntResourceType))

pkg/resource/handle/handle_test.go

@@ -22,7 +22,6 @@ import (
 	"github.com/cosi-project/runtime/pkg/safe"
 	"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"
 )
 
 //nolint:govet
@@ -93,7 +92,7 @@ func TestResource(t *testing.T) {
 		defer cancel()
 	}
 
-	st := state.WrapCore(namespaced.NewState(inmem.Build))
+	st := state.WrapCore(inmem.NewState())
 	r1 := typed.NewResource[testSpec, testExtension](
 		resource.NewMetadata("default", "testResource", "aaa", resource.VersionUndefined),
 		testSpec{Value: &Handle{someString: "my string"}},

pkg/safe/state_test.go

@@ -19,7 +19,6 @@ import (
 	"github.com/cosi-project/runtime/pkg/safe"
 	"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 setup(t *testing.T) (context.Context, string, string, *conformance.IntResource, state.State, chan safe.WrappedStateEvent[*conformance.IntResource], chan state.Event) { //nolint:ireturn
@@ -32,7 +31,7 @@ func setup(t *testing.T) (context.Context, string, string, *conformance.IntResou
 	r := conformance.NewIntResource(testNamespace, testID, 2)
 	s := state.WrapCore(
 		state.Filter(
-			namespaced.NewState(inmem.Build),
+			inmem.NewState(),
 			func(context.Context, state.Access) error {
 				return nil
 			},

pkg/state/filter_test.go

@@ -17,7 +17,6 @@ import (
 	"github.com/cosi-project/runtime/pkg/state"
 	"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"
 )
 
 func TestFilterPasshtroughConformance(t *testing.T) {
@@ -26,7 +25,7 @@ func TestFilterPasshtroughConformance(t *testing.T) {
 	suite.Run(t, &conformance.StateSuite{
 		State: state.WrapCore(
 			state.Filter(
-				namespaced.NewState(inmem.Build),
+				inmem.NewState(),
 				func(context.Context, state.Access) error {
 					return nil
 				},
@@ -47,7 +46,7 @@ func TestFilterSingleResource(t *testing.T) {
 
 	resources := state.WrapCore(
 		state.Filter(
-			namespaced.NewState(inmem.Build),
+			inmem.NewState(),
 			func(_ context.Context, access state.Access) error {
 				if access.ResourceNamespace != namespace || access.ResourceType != resourceType || access.ResourceID != resourceID {
 					return fmt.Errorf("access denied")

pkg/state/impl/inmem/backing_store.go

@@ -11,7 +11,7 @@ import (
 )
 
 // LoadHandler is called for each resource loaded from the backing store.
-type LoadHandler func(resourceType resource.Type, resource resource.Resource) error
+type LoadHandler func(ns resource.Namespace, resourceType resource.Type, resource resource.Resource) error
 
 // BackingStore provides a way to persist contents of in-memory resource collection.
 //
@@ -20,14 +20,21 @@ type LoadHandler func(resourceType resource.Type, resource resource.Resource) er
 //
 // BackingStore is responsible for marshaling/unmarshaling of resources.
 //
-// BackingStore is optional for in-memory resource collection.
+// BackingStore is optional for in-memory resource collection: a State is either fully backed by
+// the store, or fully ephemeral. Use namespaced.NewState to combine persistent and ephemeral
+// namespaces in a single state.
+//
+// A BackingStore should back at most one State: Load pulls in every namespace the store holds, so
+// two States sharing a store would each load the full contents and then diverge, as neither sees
+// the writes of the other. To split the persistent namespaces across several States, give each one
+// its own store.
 type BackingStore interface {
 	// Load contents of the backing store into the in-memory resource collection.
 	//
-	// Handler should be called for each resource in the backing store.
+	// Handler should be called for each resource in the backing store, across all the namespaces.
 	Load(ctx context.Context, handler LoadHandler) error
 	// Put the resource to the backing store.
-	Put(ctx context.Context, resourceType resource.Type, resource resource.Resource) error
+	Put(ctx context.Context, ns resource.Namespace, resourceType resource.Type, resource resource.Resource) error
 	// Destroy the resource from the backing store.
-	Destroy(ctx context.Context, resourceType resource.Type, resourcePointer resource.Pointer) error
+	Destroy(ctx context.Context, ns resource.Namespace, resourceType resource.Type, resourcePointer resource.Pointer) error
 }

pkg/state/impl/inmem/backing_store_test.go

@@ -26,24 +26,24 @@ type backingStoreMock struct {
 
 func (mock *backingStoreMock) Load(_ context.Context, handler inmem.LoadHandler) error {
 	for _, r := range mock.store {
-		if err := handler(r.Metadata().Type(), r); err != nil {
+		if err := handler(r.Metadata().Namespace(), r.Metadata().Type(), r); err != nil {
 			return err
 		}
 	}
 
 	return nil
 }
 
-func (mock *backingStoreMock) Put(_ context.Context, resourceType resource.Type, resource resource.Resource) error {
-	key := fmt.Sprintf("%s/%s", resourceType, resource.Metadata().ID())
+func (mock *backingStoreMock) Put(_ context.Context, ns resource.Namespace, resourceType resource.Type, resource resource.Resource) error {
+	key := fmt.Sprintf("%s/%s/%s", ns, resourceType, resource.Metadata().ID())
 
 	mock.store[key] = resource.DeepCopy()
 
 	return nil
 }
 
-func (mock *backingStoreMock) Destroy(_ context.Context, resourceType resource.Type, ptr resource.Pointer) error {
-	key := fmt.Sprintf("%s/%s", resourceType, ptr.ID())
+func (mock *backingStoreMock) Destroy(_ context.Context, ns resource.Namespace, resourceType resource.Type, ptr resource.Pointer) error {
+	key := fmt.Sprintf("%s/%s/%s", ns, resourceType, ptr.ID())
 
 	delete(mock.store, key)
 
@@ -57,7 +57,7 @@ func TestLocalConformanceWithBackingStore(t *testing.T) {
 		State: state.WrapCore(
 			inmem.NewStateWithOptions(
 				inmem.WithBackingStore(&backingStoreMock{store: map[string]resource.Resource{}}),
-			)("default"),
+			),
 		),
 		Namespaces: []resource.Namespace{"default"},
 	})
@@ -75,7 +75,7 @@ func TestBackingStore(t *testing.T) {
 	// create st with backing store and put some resources
 	st := state.WrapCore(inmem.NewStateWithOptions(
 		inmem.WithBackingStore(backingStore),
-	)(namespace))
+	))
 
 	path1 := conformance.NewPathResource(namespace, "var/run")
 	path2 := conformance.NewPathResource(namespace, "var/lib")
@@ -86,7 +86,7 @@ func TestBackingStore(t *testing.T) {
 	// re-create the state with backing store, resources should be still available
 	st = state.WrapCore(inmem.NewStateWithOptions(
 		inmem.WithBackingStore(backingStore),
-	)(namespace))
+	))
 
 	r, err := st.Get(ctx, path1.Metadata())
 	require.NoError(t, err)
@@ -101,7 +101,7 @@ func TestBackingStore(t *testing.T) {
 	// re-create the state with backing store, deleted resources should not be available
 	st = state.WrapCore(inmem.NewStateWithOptions(
 		inmem.WithBackingStore(backingStore),
-	)(namespace))
+	))
 
 	_, err = st.Get(ctx, path1.Metadata())
 	require.Error(t, err)