feat: rework inmem resource state implementation

[smira]

This bring inmem closer to other state implementations:

• inmem.NewState() is now namespaced inside, not an instance per namespace
• one inmem state instance has a shared event buffer (stream) for all resource types and namespaces

This should provide better memory usage for any inmem user, and also it builds the foundation to build a shared, consistent watch API for the controller runtime (not in this PR).

[Copilot]

Pull request overview

This PR reworks the in-memory (inmem) state implementation to be a single shared state across namespaces/resource types, backed by a shared circular event buffer, bringing it closer to the behavior/shape of other state implementations and enabling more consistent watch semantics going forward.

Changes:

• Switch inmem.NewState() to return a single state instance (no longer per-namespace builders) and introduce a shared event buffer across all collections.
• Introduce new internal collection, eventbuffer, and errs packages to modularize storage/watch/error behavior.
• Update BoltDB backing store and tests to support the new “store is multi-namespace” BackingStore interface and simplify state setup across the repo.

Reviewed changes

Copilot reviewed 41 out of 43 changed files in this pull request and generated 1 comment.

Show a summary per file

File	Description
pkg/state/wrap_test.go	Updates conformance test setup to use the new inmem.NewState() API.
pkg/state/teardowner_test.go	Updates teardown wrapper tests to use inmem.NewState() directly.
pkg/state/teardown_and_destroyer_test.go	Updates teardown+destroy wrapper tests to use inmem.NewState() directly.
pkg/state/registry/resource_test.go	Updates registry test setup to use inmem.NewState().
pkg/state/registry/namespace_test.go	Updates registry test setup to use inmem.NewState().
pkg/state/protobuf/protobuf_test.go	Updates protobuf state server/client tests to use inmem.NewState().
pkg/state/protobuf/client/client_test.go	Updates protobuf client test to use inmem.NewState().
pkg/state/owned/owned_test.go	Updates owned-state tests to use inmem.NewState().
pkg/state/impl/store/bolt/namespaced.go	Removes per-namespace Bolt backing store wrapper (no longer needed with multi-namespace store API).
pkg/state/impl/store/bolt/example_test.go	Updates example to show mixing persistent/ephemeral via namespaced.NewState, using a shared persistent inmem.State.
pkg/state/impl/store/bolt/conformance_test.go	Simplifies conformance setup by using a single inmem state with a multi-namespace Bolt backing store.
pkg/state/impl/store/bolt/bbolt.go	Refactors Bolt backing store to implement the updated multi-namespace inmem.BackingStore interface.
pkg/state/impl/store/bolt/bbolt_test.go	Updates Bolt backing store tests for multi-namespace load/put/destroy semantics.
pkg/state/impl/namespaced/namespaced_test.go	Updates namespaced state conformance to construct per-namespace core state via a factory that returns inmem.NewState().
pkg/state/impl/inmem/options.go	Updates history option docs and increases default shared buffer capacities for the new shared-buffer model.
pkg/state/impl/inmem/local_test.go	Updates tests for new constructors and adds coverage for shared history buffer behavior.
pkg/state/impl/inmem/internal/eventbuffer/export_test.go	Exposes watcher notify channel for testing feed isolation (test-only).
pkg/state/impl/inmem/internal/eventbuffer/eventbuffer.go	Introduces shared circular buffer + per-feed subscription model for watches.
pkg/state/impl/inmem/internal/eventbuffer/eventbuffer_test.go	Adds tests for feed isolation, overruns, bookmarks, tailing, and growth.
pkg/state/impl/inmem/internal/eventbuffer/bookmark.go	Introduces bookmark encoding/validation scoped to inmem runs.
pkg/state/impl/inmem/internal/errs/errs.go	Centralizes error types compatible with state error predicates.
pkg/state/impl/inmem/internal/collection/watch.go	Implements per-collection watch/watch-kind semantics using the shared event buffer.
pkg/state/impl/inmem/internal/collection/collection.go	Implements per-(namespace,type) collection storage and persistence hooks.
pkg/state/impl/inmem/internal/collection/collection_test.go	Adds comprehensive unit tests for collection CRUD, watch semantics, and persistence interactions.
pkg/state/impl/inmem/inmem.go	Reworks inmem state to be shared across namespaces/types and to use the new internal collection/eventbuffer packages.
pkg/state/impl/inmem/errors.go	Switches public inmem error helpers to the internal errs implementations.
pkg/state/impl/inmem/collection.go	Removes old monolithic per-type collection implementation (replaced by internal packages + shared buffer).
pkg/state/impl/inmem/build.go	Removes old builder helper tied to per-namespace construction.
pkg/state/impl/inmem/backing_store.go	Updates BackingStore interface to include namespace in Put/Destroy and Load callbacks.
pkg/state/impl/inmem/backing_store_test.go	Updates backing store mock/tests for new multi-namespace backing store API.
pkg/state/filter_test.go	Updates filter tests to use inmem.NewState() directly.
pkg/safe/state_test.go	Updates safe state tests to use inmem.NewState() directly.
pkg/resource/handle/handle_test.go	Updates handle tests to use inmem.NewState() directly.
pkg/controller/runtime/runtime_test.go	Updates controller runtime tests for new inmem API and shared-buffer option usage.
pkg/controller/generic/transform/transform_test.go	Updates transform controller tests to use inmem.NewState().
pkg/controller/generic/qtransform/qtransform_test.go	Updates qtransform controller tests to use inmem.NewState().
pkg/controller/generic/destroy/destroy_test.go	Updates destroy controller tests to use inmem.NewState().
pkg/controller/generic/cleanup/cleanup_test.go	Updates cleanup controller tests to use inmem.NewState().
Makefile	Regenerates Makefile and adds an image signing target/variable.
go.sum	Updates dependency checksums to match module version changes.
go.mod	Bumps a few module versions (e.g., grpc, compress, vtprotobuf).
cmd/runtime/main.go	Updates runtime main to use inmem.NewState() directly.
.gitignore	Regenerates .gitignore and adds additional ignored patterns.

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[Copilot]

Pull request overview

Copilot reviewed 47 out of 49 changed files in this pull request and generated no new comments.

[shanduur]

I found two correctness/resource-lifecycle issues that should be fixed before merge. Full locally executed reproducers for the first two findings are included inline.

[shanduur]

Bookmarks encode only the process cookie and a feed-local sequence number. This range check therefore accepts a bookmark from another namespace/type whenever that sequence happens to fall inside this feed's valid range, and the resumed watcher silently skips that many target-feed events.

Reproducer: save this beside eventbuffer_test.go and run:

go test ./pkg/state/impl/inmem/internal/eventbuffer -run TestReviewCrossFeedBookmarkMustBeRejected -count=1

func TestReviewCrossFeedBookmarkMustBeRejected(t *testing.T) {
    buf := eventbuffer.New(16, 16, 1)
    source := buf.NewFeed("source-ns", "source-type")
    target := buf.NewFeed("target-ns", "target-type")

    source.Publish(testEvent("source-1"))
    sourceWatcher, err := source.Subscribe(eventbuffer.SubscribeOptions{TailEvents: 1})
    require.NoError(t, err)
    sourceEvents, err := sourceWatcher.Next(t.Context())
    require.NoError(t, err)
    sourceWatcher.Close()
    require.Len(t, sourceEvents, 1)

    target.Publish(testEvent("target-1"))
    target.Publish(testEvent("target-2"))

    _, err = target.Subscribe(eventbuffer.SubscribeOptions{Bookmark: sourceEvents[0].Bookmark})
    require.Error(t, err, "a bookmark from another feed must not be accepted")
}

At this head it fails with An error is expected but got nil. Please bind bookmarks to a feed identity, including buffer identity and namespace/type, validate it here, and retain this as a regression test.

[smira]

I'm not sure if it's fully valid though - this is internal package, so it might not be reachable in general as this implies someone used bookmark from one Watch call in the other unrelated Watch. Same way one could in theory supply any other bookmark.

[shanduur]

collection.New is evaluated before LoadOrStore, and it immediately registers a feed in buf.feeds. Concurrent first access to the same (namespace,type) can therefore register several feeds while only one collection wins the map insertion. The losing feeds remain reachable forever and every cleanup sweep scans them under the global buffer mutex.

Reproducer: save this as pkg/state/impl/inmem/concurrent_collection_review_test.go and run:

go test ./pkg/state/impl/inmem -run TestReviewConcurrentCollectionCreationDoesNotLeakFeeds -count=1

package inmem

import (
    "reflect"
    "sync"
    "testing"
)

func TestReviewConcurrentCollectionCreationDoesNotLeakFeeds(t *testing.T) {
    for attempt := 0; attempt < 100; attempt++ {
        st := NewState()
        start := make(chan struct{})
        var wg sync.WaitGroup

        for range 128 {
            wg.Add(1)
            go func() {
                defer wg.Done()
                <-start
                st.getCollection("same-ns", "same-type")
            }()
        }

        close(start)
        wg.Wait()

        feeds := reflect.ValueOf(st.buffer).Elem().FieldByName("feeds").Len()
        if feeds != 1 {
            t.Fatalf("attempt %d: one installed collection registered %d feeds", attempt, feeds)
        }
    }
}

On my run it failed immediately with attempt 0: one installed collection registered 4 feeds. Please make creation single-winner or unregister losing candidates, and retain a concurrent same-key regression test.

[smira]

this is great catch!

[shanduur]

Please clear the discarded tail before returning, for example with clear(slc[len(r):]). The compacted slice shares its backing array with the original event batch, so an aggregated-watch consumer retaining a small filtered batch also retains all rejected state.Event resource references beyond len(r). That defeats history cleanup for those resources and can retain a full backlog for the lifetime of the consumer-held batch.

[smira]

this is pre-existing (this function got moved around), but nice catch!

[utkuozdemir]

Wondering if

• we should worry about this goroutine panicking
• leave concurrency to the caller? Like a blocking RunCleanup(ctx) error or similar?

[smira]

if we do this, I think we should move the history cleanup interval as well out

[utkuozdemir]

I'm torn, so your call - I often lean towards that kind of design myself, also remember us having discussions about us not liking libraries spawning their goroutines.

[utkuozdemir]

The cleanup and the expiry account the same events into staleSeq twice. The sweep assigns it here after releasing the events, and when the write position later wraps over those same slots, expireLocked increments it once more per slot. For a quiet feed staleSeq can then exceed publishedSeq, which makes the tail rewind in Subscribe negative, so a new watcher starts ahead of the feed and the next published events are silently skipped instead of being delivered or reported as an overrun. Also, once the buffer wraps, boundary <= f.staleSeq becomes permanently true and the sweep stops releasing anything.

Reproducer: save this beside eventbuffer_test.go and run:

go test ./pkg/state/impl/inmem/internal/eventbuffer -run TestReviewCleanupMustNotSkipFutureEvents -count=1

func TestReviewCleanupMustNotSkipFutureEvents(t *testing.T) {
	buf := eventbuffer.New(8, 8, 1)
	quiet := buf.NewFeed("ns", "quiet")
	busy := buf.NewFeed("ns", "busy")

	quiet.Publish(testEvent("old"))

	// the feed has no watchers, so two sweeps release the "old" event
	require.Zero(t, buf.Cleanup())
	require.Equal(t, 1, buf.Cleanup())

	// rotate the buffer over the released slot
	for i := range 8 {
		busy.Publish(testEvent(strconv.Itoa(i)))
	}

	tail, err := quiet.Subscribe(eventbuffer.SubscribeOptions{TailEvents: 1})
	require.NoError(t, err)
	defer tail.Close()

	quiet.Publish(testEvent("new"))

	ctx, cancel := context.WithTimeout(t.Context(), 100*time.Millisecond)
	defer cancel()

	events, err := tail.Next(ctx)
	require.NoError(t, err)
	require.Equal(t, []resource.ID{"new"}, eventIDs(events))
}

At this head it fails with the watcher receiving no events at all, the "new" event published to the feed is never delivered.

[utkuozdemir]

While the buffer is still growing, no event can be overwritten yet (the capacity doubles exactly at writePos == capacity, before any slot is reused), so the gap protects nothing at that stage. But the stale boundary here is computed from the current capacity, and staleSeq only ever grows, so events get permanently marked stale while the buffer is nowhere near full. On main this window was computed at subscribe time against the already grown capacity, so growth made the early events reachable again. The same scenario through the public API replays all events on main and loses some of them at this head.

Two ways this shows up: with the default options (initial capacity 256, gap 50) a fresh state loses tail and bookmark access to its earliest events after about 200 publishes. With an initial capacity below the gap (e.g. WithHistoryInitialCapacity(8) with the default gap) almost every event goes stale the moment it is published, and bookmarks handed out one event ago are already rejected.

Reproducer covering both, save it beside eventbuffer_test.go and run:

go test ./pkg/state/impl/inmem/internal/eventbuffer -run TestReviewGrowthMustNotMakeEventsStale -count=1

func TestReviewGrowthMustNotMakeEventsStale(t *testing.T) {
	for _, test := range []struct {
		name            string
		initialCapacity int
		gap             int
		publish         int
	}{
		{name: "gap below initial capacity", initialCapacity: 16, gap: 8, publish: 20},
		{name: "gap above initial capacity", initialCapacity: 8, gap: 50, publish: 6},
	} {
		t.Run(test.name, func(t *testing.T) {
			buf := eventbuffer.New(test.initialCapacity, 40960, test.gap)
			feed := buf.NewFeed("ns", "type")

			for i := range test.publish {
				feed.Publish(testEvent(strconv.Itoa(i)))
			}

			w, err := feed.Subscribe(eventbuffer.SubscribeOptions{TailEvents: test.publish})
			require.NoError(t, err)

			defer w.Close()

			events, err := w.Next(t.Context())
			require.NoError(t, err)
			require.Len(t, events, test.publish, "no event was ever overwritten, all should be replayable")
		})
	}
}

At this head both subtests fail, replaying 13 of 20 and 1 of 6 events.

[utkuozdemir]

I think we should create the ephemeral inmem state here only once, with the new changes. Otherwise each ephemeral namespace gets its own state and event buffer.

Suggested change

	// an inmem.State is either fully backed by the store, or fully ephemeral, so build a separate
	// persistent state, and route the namespaces to it with namespaced.NewState
	persistentState := inmem.NewStateWithOptions(
	inmem.WithBackingStore(backingStore),
	)
	// create resource state with following namespaces
	// * backed by BoltDB: persistent, system
	// * in-memory: runtime
	resources := state.WrapCore(namespaced.NewState(
	func(ns resource.Namespace) state.CoreState {
	switch ns {
	case "persistent", "system":
	// use in-memory state backed by BoltDB
	return inmem.NewStateWithOptions(
	inmem.WithBackingStore(backingStore.WithNamespace(ns)),
	)(ns)
	case "runtime":
	return inmem.NewState(ns)
	default:
	panic("unexpected namespace")
	}
	},
	))
	// * in-memory: any other namespace, e.g. runtime
	resources := state.WrapCore(namespaced.NewState(func(ns resource.Namespace) state.CoreState {
	switch ns {
	case "persistent", "system":
	return persistentState
	default:
	return inmem.NewState()
	}
	}))
	// an inmem.State is either fully backed by the store, or fully ephemeral, so build a separate
	// persistent state, and route the namespaces to it with namespaced.NewState
	persistentState := inmem.NewStateWithOptions(
	inmem.WithBackingStore(backingStore),
	)
	// a single ephemeral state instance is shared by all the other namespaces
	ephemeralState := inmem.NewState()
	// create resource state with following namespaces
	// * backed by BoltDB: persistent, system
	// * in-memory: any other namespace, e.g. runtime
	resources := state.WrapCore(namespaced.NewState(func(ns resource.Namespace) state.CoreState {
	switch ns {
	case "persistent", "system":
	return persistentState
	default:
	return ephemeralState
	}
	}))

[utkuozdemir]

Should we update the recommendation on WithHistoryGap ("gap 5% of the capacity")? The new defaults (gap 50 for max capacity 40960) are way below 5%.

[utkuozdemir]

Wondering if, for the single-resource watches (by ID), we could do the filtering here in collectLocked, e.g. by applying SubscribeOptions.Filter, which already exists for the tail rewind. This way the whole event batch of the type would not be copied to each per-ID watcher on every wakeup, only to be dropped in the ID check afterwards.