diff --git a/doc/thinking/2026-04-26-namespace-manager-step26.md b/doc/thinking/2026-04-26-namespace-manager-step26.md new file mode 100644 index 00000000..0553b9a3 --- /dev/null +++ b/doc/thinking/2026-04-26-namespace-manager-step26.md @@ -0,0 +1,17 @@ +# 2026-04-26 — NamespaceManager rewrite, step 26 + +## Цель шага + +Добавить единый startup bridge для manager: bootstrap model + dispatch в subscriber-ы. + +## Что меняем + +1. В `NamespaceManager` добавляем `BootstrapAndDispatch()`. +2. Helper сначала делает `Bootstrap()`, потом вызывает `DispatchAdd()` по каждому namespace. +3. Ошибки агрегируются и не останавливают остальные namespace. +4. Добавляем unit tests на success и partial-failure. + +## Что НЕ меняем + +- не подключаем helper к production startup path; +- не меняем watcher behavior. \ No newline at end of file diff --git a/pkg/utils/namespace_manager.go b/pkg/utils/namespace_manager.go index dff6a9d2..2d74503d 100644 --- a/pkg/utils/namespace_manager.go +++ b/pkg/utils/namespace_manager.go @@ -2,6 +2,7 @@ package utils import ( "context" + "errors" "sort" "sync" "time" @@ -51,6 +52,7 @@ type NamespaceManager interface { SnapshotRecords() []NamespaceRecord Get(name string) (NamespaceRecord, bool) Bootstrap(namespaces []string, source NamespaceSource, observedAt time.Time) []NamespaceRecord + BootstrapAndDispatch(ctx context.Context, namespaces []string, source NamespaceSource, observedAt time.Time) ([]NamespaceRecord, error) Subscribe(subscriber NamespaceSubscriber) SnapshotSubscribers() []string Upsert(event NamespaceEvent) NamespaceRecord @@ -107,6 +109,27 @@ func (m *inMemoryNamespaceManager) Bootstrap(namespaces []string, source Namespa return records } +func (m *inMemoryNamespaceManager) BootstrapAndDispatch(ctx context.Context, namespaces []string, source NamespaceSource, observedAt time.Time) ([]NamespaceRecord, error) { + records := m.Bootstrap(namespaces, source, observedAt) + var joinErr error + for _, record := range records { + updatedRecord, ok, err := m.DispatchAdd(ctx, record.Name) + if ok { + record = updatedRecord + } + if err != nil { + joinErr = errors.Join(joinErr, err) + } + for index := range records { + if records[index].Name == record.Name { + records[index] = record + break + } + } + } + return records, joinErr +} + func (m *inMemoryNamespaceManager) SnapshotSubscribers() []string { m.mu.RLock() defer m.mu.RUnlock() diff --git a/pkg/utils/namespace_manager_test.go b/pkg/utils/namespace_manager_test.go index 193ac0ce..dd470b4e 100644 --- a/pkg/utils/namespace_manager_test.go +++ b/pkg/utils/namespace_manager_test.go @@ -179,6 +179,50 @@ func TestNamespaceManagerBootstrap(t *testing.T) { } } +func TestNamespaceManagerBootstrapAndDispatch(t *testing.T) { + manager := NewNamespaceManager() + router := &testNamespaceSubscriber{name: "router"} + builder := &testNamespaceSubscriber{name: "buildermgr"} + manager.Subscribe(router) + manager.Subscribe(builder) + + records, err := manager.BootstrapAndDispatch(context.Background(), []string{"tenant-b", "tenant-a"}, NamespaceSourceBackfill, time.Now().UTC()) + if err != nil { + t.Fatalf("expected bootstrap and dispatch success: %v", err) + } + if len(records) != 2 { + t.Fatalf("expected 2 records, got %d", len(records)) + } + if router.addCalls != 2 || builder.addCalls != 2 { + t.Fatalf("expected both subscribers to be called for each namespace") + } + if records[0].Phase != NamespacePhaseActive || records[1].Phase != NamespacePhaseActive { + t.Fatalf("expected active records after bootstrap dispatch") + } +} + +func TestNamespaceManagerBootstrapAndDispatchFailure(t *testing.T) { + manager := NewNamespaceManager() + router := &testNamespaceSubscriber{name: "router", addErr: errors.New("router failed")} + builder := &testNamespaceSubscriber{name: "buildermgr"} + manager.Subscribe(router) + manager.Subscribe(builder) + + records, err := manager.BootstrapAndDispatch(context.Background(), []string{"tenant-a"}, NamespaceSourceBackfill, time.Now().UTC()) + if err == nil { + t.Fatalf("expected bootstrap dispatch failure") + } + if len(records) != 1 { + t.Fatalf("expected single record result") + } + if records[0].Phase != NamespacePhaseFailed { + t.Fatalf("expected failed phase after subscriber error, got %s", records[0].Phase) + } + if records[0].RegisteredParts["router"].LastError != "router failed" { + t.Fatalf("expected router failure to be stored") + } +} + func TestNamespaceManagerPartStateHelpers(t *testing.T) { manager := NewNamespaceManager() manager.Upsert(NamespaceEvent{Type: NamespaceEventAdd, Name: "tenant-a", Source: NamespaceSourceWatcher})