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, 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.

[utkuozdemir]

This question still stands

[utkuozdemir]

The stale boundary is computed from writePos before this publish, but a watcher subscribing right after it sees the buffer with one more event. So once the buffer is full, a tail watch can start gap - 1 slots away from the eviction edge instead of gap, and with gap 1 it gets overrun by the very next publish. On main the same scenario kept the gap.

Reproducer, save beside eventbuffer_test.go:

func TestReviewTailKeepsTheGap(t *testing.T) {
	buf := eventbuffer.New(4, 4, 1)
	f := buf.NewFeed("ns", "type")

	for i := range 4 {
		f.Publish(testEvent(strconv.Itoa(i)))
	}

	w, err := f.Subscribe(eventbuffer.SubscribeOptions{TailEvents: 4})
	require.NoError(t, err)
	t.Cleanup(w.Close)

	f.Publish(testEvent("4"))

	events, err := w.Next(t.Context())
	require.NoError(t, err)
	require.Equal(t, []resource.ID{"1", "2", "3", "4"}, eventIDs(events))
}

At this head it fails with buffer overrun: ... published 5, evicted 1, consumed 0.

[utkuozdemir]

The sweep releases only the events above max(staleSeq, droppedSeq), so the events which are stale because of the gap but still sitting in the buffer are never released by the cleanup. If the publishing stops, they keep their resources pinned for good, up to gap events (50 with the defaults).

Reproducer, save beside eventbuffer_test.go:

func TestReviewCleanupReleasesStaleEvents(t *testing.T) {
	buf := eventbuffer.New(4, 4, 2)
	f := buf.NewFeed("ns", "type")

	for i := range 4 {
		f.Publish(testEvent(strconv.Itoa(i)))
	}

	require.Zero(t, buf.Cleanup())
	require.Equal(t, 4, buf.Cleanup())
}

At this head the second sweep returns 3.

[utkuozdemir]

A bit hypothetical, but calling the builder under the lock can deadlock if a builder ever reaches back into the same namespaced.State (e.g. reads another namespace), and a slow builder blocks the first access to all the other namespaces. Maybe use a sync.OnceValue per namespace instead and drop the lock? The builder still runs once per namespace, only the callers of that namespace wait, and nothing runs under the router's lock:

build, _ := st.namespaces.LoadOrStore(ns, sync.OnceValue(func() state.CoreState {
	return st.builder(ns)
}))

return build()