layer1: add namespace bootstrap dispatch step 26
This commit is contained in:
@@ -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.
|
||||||
@@ -2,6 +2,7 @@ package utils
|
|||||||
|
|
||||||
import (
|
import (
|
||||||
"context"
|
"context"
|
||||||
|
"errors"
|
||||||
"sort"
|
"sort"
|
||||||
"sync"
|
"sync"
|
||||||
"time"
|
"time"
|
||||||
@@ -51,6 +52,7 @@ type NamespaceManager interface {
|
|||||||
SnapshotRecords() []NamespaceRecord
|
SnapshotRecords() []NamespaceRecord
|
||||||
Get(name string) (NamespaceRecord, bool)
|
Get(name string) (NamespaceRecord, bool)
|
||||||
Bootstrap(namespaces []string, source NamespaceSource, observedAt time.Time) []NamespaceRecord
|
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)
|
Subscribe(subscriber NamespaceSubscriber)
|
||||||
SnapshotSubscribers() []string
|
SnapshotSubscribers() []string
|
||||||
Upsert(event NamespaceEvent) NamespaceRecord
|
Upsert(event NamespaceEvent) NamespaceRecord
|
||||||
@@ -107,6 +109,27 @@ func (m *inMemoryNamespaceManager) Bootstrap(namespaces []string, source Namespa
|
|||||||
return records
|
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 {
|
func (m *inMemoryNamespaceManager) SnapshotSubscribers() []string {
|
||||||
m.mu.RLock()
|
m.mu.RLock()
|
||||||
defer m.mu.RUnlock()
|
defer m.mu.RUnlock()
|
||||||
|
|||||||
@@ -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) {
|
func TestNamespaceManagerPartStateHelpers(t *testing.T) {
|
||||||
manager := NewNamespaceManager()
|
manager := NewNamespaceManager()
|
||||||
manager.Upsert(NamespaceEvent{Type: NamespaceEventAdd, Name: "tenant-a", Source: NamespaceSourceWatcher})
|
manager.Upsert(NamespaceEvent{Type: NamespaceEventAdd, Name: "tenant-a", Source: NamespaceSourceWatcher})
|
||||||
|
|||||||
Reference in New Issue
Block a user