diff --git a/internal/federation/federation_test.go b/internal/federation/federation_test.go new file mode 100644 index 0000000..230090e --- /dev/null +++ b/internal/federation/federation_test.go @@ -0,0 +1,518 @@ +package federation + +import ( + "context" + "encoding/json" + "errors" + "net/http" + "net/http/httptest" + "sync/atomic" + "testing" + "time" +) + +// ---- MemoryStore tests ---- + +func TestMemoryStoreGetMissing(t *testing.T) { + s := NewMemoryStore() + _, err := s.Get(context.Background(), "nope") + if !errors.Is(err, ErrPolicyNotFound) { + t.Fatalf("Get missing: got err=%v, want ErrPolicyNotFound", err) + } +} + +func TestMemoryStorePutGetDeleteList(t *testing.T) { + s := NewMemoryStore() + ctx := context.Background() + + if err := s.Put(ctx, &Policy{ID: "b", Version: 1, Body: []byte("B")}); err != nil { + t.Fatalf("Put b: %v", err) + } + if err := s.Put(ctx, &Policy{ID: "a", Version: 1, Body: []byte("A")}); err != nil { + t.Fatalf("Put a: %v", err) + } + + got, err := s.Get(ctx, "a") + if err != nil { + t.Fatalf("Get a: %v", err) + } + if string(got.Body) != "A" { + t.Fatalf("Get a body = %q, want %q", got.Body, "A") + } + + // List is ordered by ID. + list, err := s.List(ctx) + if err != nil { + t.Fatalf("List: %v", err) + } + if len(list) != 2 || list[0].ID != "a" || list[1].ID != "b" { + t.Fatalf("List = %+v, want [a,b]", list) + } + + // Mutating a returned copy must not corrupt stored state. + got.Body[0] = 'X' + again, _ := s.Get(ctx, "a") + if string(again.Body) != "A" { + t.Fatalf("stored body mutated via returned copy: %q", again.Body) + } + + if err := s.Delete(ctx, "a"); err != nil { + t.Fatalf("Delete a: %v", err) + } + if _, err := s.Get(ctx, "a"); !errors.Is(err, ErrPolicyNotFound) { + t.Fatalf("after Delete, got err=%v, want ErrPolicyNotFound", err) + } + // Delete of an absent id is a no-op. + if err := s.Delete(ctx, "absent"); err != nil { + t.Fatalf("Delete absent: unexpected err %v", err) + } +} + +func TestMemoryStorePutNil(t *testing.T) { + if err := (NewMemoryStore()).Put(context.Background(), nil); err == nil { + t.Fatal("Put(nil) returned nil, want error") + } +} + +// ---- RegionRegistry + Router tests ---- + +func TestRegionRegistrySelfProtected(t *testing.T) { + r := NewRegionRegistry("us") + r.Register(Region{ID: "eu", Healthy: true}) + r.Unregister("us") // no-op for self + r.Unregister("eu") + + if _, ok := r.Get("us"); !ok { + t.Fatal("self region was removed by Unregister(self)") + } + if _, ok := r.Get("eu"); ok { + t.Fatal("eu should have been removed") + } +} + +func TestRegionRegistrySetHealthLatencyUnknown(t *testing.T) { + r := NewRegionRegistry("us") + if err := r.SetHealth("nope", true); !errors.Is(err, ErrRegionNotFound) { + t.Fatalf("SetHealth unknown: got err=%v, want ErrRegionNotFound", err) + } + if err := r.SetLatency("nope", 10*time.Millisecond); !errors.Is(err, ErrRegionNotFound) { + t.Fatalf("SetLatency unknown: got err=%v, want ErrRegionNotFound", err) + } +} + +func TestRouterRoutePicksLowestLatency(t *testing.T) { + r := NewRegionRegistry("") + r.Register(Region{ID: "near", Healthy: true, Latency: 10 * time.Millisecond}) + r.Register(Region{ID: "far", Healthy: true, Latency: 200 * time.Millisecond}) + // unknown-latency region ranks last even if "closer" by id/priority. + r.Register(Region{ID: "zzz", Healthy: true, Latency: 0, Priority: -1}) + + rt := NewRouter(r) + got, err := rt.Route() + if err != nil { + t.Fatalf("Route: %v", err) + } + if got.ID != "near" { + t.Fatalf("Route = %q, want near (lowest known latency)", got.ID) + } +} + +func TestRouterRouteNoHealthy(t *testing.T) { + r := NewRegionRegistry("") + r.Register(Region{ID: "a", Healthy: false, Latency: 1 * time.Millisecond}) + if _, err := (NewRouter(r)).Route(); !errors.Is(err, ErrRegionNotFound) { + t.Fatalf("Route with no healthy: got err=%v, want ErrRegionNotFound", err) + } +} + +func TestRouterRouteFailover(t *testing.T) { + r := NewRegionRegistry("") + r.Register(Region{ID: "primary", Healthy: false, Latency: 5 * time.Millisecond}) + r.Register(Region{ID: "backup", Healthy: true, Latency: 50 * time.Millisecond}) + + rt := NewRouter(r) + // primary down -> failover to backup. + got, err := rt.RouteFailover("primary") + if err != nil { + t.Fatalf("RouteFailover: %v", err) + } + if got.ID != "backup" { + t.Fatalf("RouteFailover = %q, want backup", got.ID) + } + // primary back up -> returns preferred. + if err := r.SetHealth("primary", true); err != nil { + t.Fatal(err) + } + got, _ = rt.RouteFailover("primary") + if got.ID != "primary" { + t.Fatalf("RouteFailover healthy preferred = %q, want primary", got.ID) + } +} + +func TestRouterRoutePriorityTiebreak(t *testing.T) { + r := NewRegionRegistry("") + // Equal latency: lower Priority wins. + r.Register(Region{ID: "a", Healthy: true, Latency: 10 * time.Millisecond, Priority: 5}) + r.Register(Region{ID: "b", Healthy: true, Latency: 10 * time.Millisecond, Priority: 1}) + got, _ := NewRouter(r).Route() + if got.ID != "b" { + t.Fatalf("Route tie = %q, want b (lower priority)", got.ID) + } +} + +// ---- LocalTransport tests ---- + +func TestLocalTransportUnknownRegion(t *testing.T) { + tt := NewLocalTransport() + ctx := context.Background() + if _, err := tt.GetPolicy(ctx, "x", "p"); !errors.Is(err, ErrRegionNotFound) { + t.Fatalf("GetPolicy unknown region: got err=%v, want ErrRegionNotFound", err) + } + if err := tt.PutPolicy(ctx, "x", &Policy{ID: "p"}); !errors.Is(err, ErrRegionNotFound) { + t.Fatalf("PutPolicy unknown region: got err=%v, want ErrRegionNotFound", err) + } + if _, err := tt.ListPolicies(ctx, "x"); !errors.Is(err, ErrRegionNotFound) { + t.Fatalf("ListPolicies unknown region: got err=%v, want ErrRegionNotFound", err) + } + if err := tt.Ping(ctx, "x"); !errors.Is(err, ErrRegionNotFound) { + t.Fatalf("Ping unknown region: got err=%v, want ErrRegionNotFound", err) + } +} + +func TestLocalTransportMountRoundTrip(t *testing.T) { + tt := NewLocalTransport() + store := NewMemoryStore() + tt.Mount("us", store) + ctx := context.Background() + + if err := tt.PutPolicy(ctx, "us", &Policy{ID: "p", Version: 1, Body: []byte("hi")}); err != nil { + t.Fatalf("PutPolicy: %v", err) + } + got, err := tt.GetPolicy(ctx, "us", "p") + if err != nil { + t.Fatalf("GetPolicy: %v", err) + } + if got.Version != 1 { + t.Fatalf("Version = %d, want 1", got.Version) + } + list, err := tt.ListPolicies(ctx, "us") + if err != nil || len(list) != 1 { + t.Fatalf("ListPolicies = %v (len %d), want 1", err, len(list)) + } +} + +// ---- Replicator anti-entropy tests ---- + +// newFleet builds three region federations (a, b, c) sharing one transport +// and registry, each backed by its own MemoryStore mounted on the transport. +// It returns the federations keyed by id and the shared transport. +func newFleet(t *testing.T, cfg ReplicationConfig) (map[string]*Federation, *LocalTransport) { + t.Helper() + transport := NewLocalTransport() + registry := NewRegionRegistry("") + + stores := map[string]*MemoryStore{} + feds := map[string]*Federation{} + for _, id := range []string{"a", "b", "c"} { + s := NewMemoryStore() + stores[id] = s + transport.Mount(id, s) + registry.Register(Region{ID: id, Healthy: true, Latency: 10 * time.Millisecond}) + feds[id] = New(id, WithStore(stores[id]), WithRegistry(registry), WithTransport(transport), WithConfig(cfg)) + } + return feds, transport +} + +func TestReplicatorPushAndPull(t *testing.T) { + feds, _ := newFleet(t, ReplicationConfig{Interval: time.Second}) + ctx := context.Background() + + // a has a policy b does not: reconcile a => push to b and c. + if err := feds["a"].store.Put(ctx, &Policy{ID: "p1", Version: 1, Body: []byte("a1"), OriginRegion: "a", UpdatedAt: time.Now()}); err != nil { + t.Fatal(err) + } + if err := feds["a"].Reconcile(ctx); err != nil { + t.Fatalf("a.Reconcile: %v", err) + } + if got, err := feds["b"].store.Get(ctx, "p1"); err != nil || string(got.Body) != "a1" { + t.Fatalf("b did not receive p1 via push: err=%v got=%v", err, got) + } + if got := atomic.LoadUint64(&feds["a"].replicator.pushed); got != 2 { + t.Fatalf("pushed = %d, want 2 (b and c)", got) + } + + // c now authors a policy a does not: reconcile c => pull into a. + if err := feds["c"].store.Put(ctx, &Policy{ID: "p2", Version: 1, Body: []byte("c1"), OriginRegion: "c", UpdatedAt: time.Now()}); err != nil { + t.Fatal(err) + } + if err := feds["c"].Reconcile(ctx); err != nil { + t.Fatalf("c.Reconcile: %v", err) + } + if got, err := feds["a"].store.Get(ctx, "p2"); err != nil || string(got.Body) != "c1" { + t.Fatalf("a did not receive p2 via pull: err=%v got=%v", err, got) + } +} + +func TestReplicatorNewerRevisionWins(t *testing.T) { + feds, _ := newFleet(t, ReplicationConfig{Interval: time.Second}) + ctx := context.Background() + + older := &Policy{ID: "p", Version: 1, Body: []byte("old"), OriginRegion: "a", UpdatedAt: time.UnixMilli(1000)} + newer := &Policy{ID: "p", Version: 2, Body: []byte("new"), OriginRegion: "b", UpdatedAt: time.UnixMilli(2000)} + _ = feds["a"].store.Put(ctx, older) + _ = feds["b"].store.Put(ctx, newer) + + // a reconciles: local v1 < remote v2 => pull newer from b. + if err := feds["a"].Reconcile(ctx); err != nil { + t.Fatal(err) + } + got, _ := feds["a"].store.Get(ctx, "p") + if got.Version != 2 || string(got.Body) != "new" { + t.Fatalf("a did not pull newer revision: v=%d body=%s", got.Version, got.Body) + } + // b reconciles: its v2 > a's (now v2) => equal => skip, no pull of stale. + before := atomic.LoadUint64(&feds["b"].replicator.pulled) + _ = feds["b"].Reconcile(ctx) + if got := atomic.LoadUint64(&feds["b"].replicator.pulled); got != before { + t.Fatalf("b pulled when revisions were equal: delta=%d", got-before) + } +} + +func TestReplicatorLastWriteWinsOnVersionTie(t *testing.T) { + feds, _ := newFleet(t, ReplicationConfig{Interval: time.Second}) + ctx := context.Background() + + // Two regions concurrently author the same Version with different + // timestamps: the later UpdatedAt must win. + earlier := &Policy{ID: "p", Version: 5, Body: []byte("early"), OriginRegion: "a", UpdatedAt: time.UnixMilli(1000)} + later := &Policy{ID: "p", Version: 5, Body: []byte("late"), OriginRegion: "b", UpdatedAt: time.UnixMilli(9000)} + _ = feds["a"].store.Put(ctx, earlier) + _ = feds["b"].store.Put(ctx, later) + + _ = feds["a"].Reconcile(ctx) // a sees b's later write => pull. + got, _ := feds["a"].store.Get(ctx, "p") + if string(got.Body) != "late" { + t.Fatalf("LWW tie did not pick later write: %s", got.Body) + } +} + +func TestReplicatorEventualConsistency(t *testing.T) { + feds, _ := newFleet(t, ReplicationConfig{Interval: time.Second}) + ctx := context.Background() + + // Each region authors a distinct policy. + _ = feds["a"].store.Put(ctx, &Policy{ID: "pa", Version: 1, Body: []byte("A"), OriginRegion: "a", UpdatedAt: time.Now()}) + _ = feds["b"].store.Put(ctx, &Policy{ID: "pb", Version: 1, Body: []byte("B"), OriginRegion: "b", UpdatedAt: time.Now()}) + _ = feds["c"].store.Put(ctx, &Policy{ID: "pc", Version: 1, Body: []byte("C"), OriginRegion: "c", UpdatedAt: time.Now()}) + + // Each region reconciles once. + for _, id := range []string{"a", "b", "c"} { + if err := feds[id].Reconcile(ctx); err != nil { + t.Fatalf("%s.Reconcile: %v", id, err) + } + } + // a pulled pb/pc on its pass, but to fully converge c needs a's pull of + // pb propagated: run a second round. + for _, id := range []string{"a", "b", "c"} { + _ = feds[id].Reconcile(ctx) + } + + for _, id := range []string{"a", "b", "c"} { + list, err := feds[id].store.List(ctx) + if err != nil { + t.Fatal(err) + } + if len(list) != 3 { + ids := make([]string, len(list)) + for i, p := range list { + ids[i] = p.ID + } + t.Fatalf("region %s has %d policies (%v), want 3 converged", id, len(list), ids) + } + } +} + +func TestReplicatorAllowlistOverridesDiscovery(t *testing.T) { + // cfg pins replication to "b" only, even though c is a healthy peer. + feds, _ := newFleet(t, ReplicationConfig{Interval: time.Second, Regions: []string{"b"}}) + ctx := context.Background() + + _ = feds["a"].store.Put(ctx, &Policy{ID: "p", Version: 1, Body: []byte("x"), OriginRegion: "a", UpdatedAt: time.Now()}) + _ = feds["a"].Reconcile(ctx) + + if _, err := feds["b"].store.Get(ctx, "p"); err != nil { + t.Fatalf("b (allowlisted) did not receive p: %v", err) + } + if _, err := feds["c"].store.Get(ctx, "p"); !errors.Is(err, ErrPolicyNotFound) { + t.Fatalf("c (not allowlisted) received p: %v", err) + } +} + +// ---- Federation façade tests ---- + +func TestFederationPutIncrementsVersion(t *testing.T) { + f := New("us") + ctx := context.Background() + + p1, err := f.Put(ctx, "policy-x", []byte("v1")) + if err != nil { + t.Fatalf("Put v1: %v", err) + } + if p1.Version != 1 || p1.OriginRegion != "us" { + t.Fatalf("p1 = %+v, want version 1 origin us", p1) + } + p2, err := f.Put(ctx, "policy-x", []byte("v2")) + if err != nil { + t.Fatalf("Put v2: %v", err) + } + if p2.Version != 2 { + t.Fatalf("p2.Version = %d, want 2", p2.Version) + } + got, _ := f.Get(ctx, "policy-x") + if string(got.Body) != "v2" { + t.Fatalf("Get body = %s, want v2", got.Body) + } + if f.Stats().Authored != 2 { + t.Fatalf("Authored = %d, want 2", f.Stats().Authored) + } +} + +func TestFederationPutEmptyID(t *testing.T) { + if _, err := (New("us")).Put(context.Background(), "", []byte("x")); err == nil { + t.Fatal("Put(empty id) returned nil, want error") + } +} + +func TestFederationSyncModePushesImmediately(t *testing.T) { + // Sync mode: Put pushes to targets before returning, no Reconcile needed. + cfg := ReplicationConfig{Mode: ReplicationSync, Interval: time.Second} + feds, _ := newFleet(t, cfg) + ctx := context.Background() + + if _, err := feds["a"].Put(ctx, "hot", []byte("now")); err != nil { + t.Fatalf("Put sync: %v", err) + } + for _, id := range []string{"b", "c"} { + got, err := feds[id].Get(ctx, "hot") + if err != nil || string(got.Body) != "now" { + t.Fatalf("sync push did not reach %s: err=%v got=%v", id, err, got) + } + } +} + +func TestFederationAsyncModeRequiresReconcile(t *testing.T) { + // Async mode (default): Put updates local only until Reconcile. + feds, _ := newFleet(t, ReplicationConfig{Mode: ReplicationAsync, Interval: time.Second}) + ctx := context.Background() + + if _, err := feds["a"].Put(ctx, "lazy", []byte("later")); err != nil { + t.Fatal(err) + } + if _, err := feds["b"].Get(ctx, "lazy"); !errors.Is(err, ErrPolicyNotFound) { + t.Fatalf("async pushed before Reconcile: %v", err) + } + _ = feds["a"].Reconcile(ctx) + if got, err := feds["b"].Get(ctx, "lazy"); err != nil || string(got.Body) != "later" { + t.Fatalf("async did not propagate after Reconcile: err=%v got=%v", err, got) + } +} + +func TestFederationBackgroundLoopReplicates(t *testing.T) { + // A short-interval background loop must propagate without manual Reconcile. + cfg := ReplicationConfig{Mode: ReplicationAsync, Interval: 5 * time.Millisecond} + feds, _ := newFleet(t, cfg) + ctx := context.Background() + + for _, id := range []string{"a", "b", "c"} { + feds[id].Start() + defer feds[id].Stop() + } + + if _, err := feds["a"].Put(ctx, "bg", []byte("z")); err != nil { + t.Fatal(err) + } + + // Poll for convergence (bounded so the test stays fast and non-flaky). + deadline := time.Now().Add(2 * time.Second) + var converged bool + for time.Now().Before(deadline) && !converged { + converged = true + for _, id := range []string{"a", "b", "c"} { + if got, err := feds[id].Get(ctx, "bg"); err != nil || string(got.Body) != "z" { + converged = false + break + } + } + if !converged { + time.Sleep(5 * time.Millisecond) + } + } + if !converged { + t.Fatal("background loop did not converge within deadline") + } +} + +func TestFederationStats(t *testing.T) { + cfg := ReplicationConfig{Mode: ReplicationSync, Interval: 250 * time.Millisecond} + feds, _ := newFleet(t, cfg) + ctx := context.Background() + + _, _ = feds["a"].Put(ctx, "p", []byte("x")) + st := feds["a"].Stats() + if st.Self != "a" { + t.Fatalf("Self = %q, want a", st.Self) + } + if st.Mode != "sync" { + t.Fatalf("Mode = %q, want sync", st.Mode) + } + if st.Regions != 3 || st.HealthyRegions != 3 { + t.Fatalf("Regions=%d Healthy=%d, want 3/3", st.Regions, st.HealthyRegions) + } + if st.Authored != 1 { + t.Fatalf("Authored = %d, want 1", st.Authored) + } + if st.SyncPushes < 2 { + t.Fatalf("SyncPushes = %d, want >=2 (sync push to b and c)", st.SyncPushes) + } +} + +func TestFederationRegisterRoutes(t *testing.T) { + f := New("us") + f.Registry().Register(Region{ID: "eu", Healthy: true, Latency: 20 * time.Millisecond}) + ctx := context.Background() + _, _ = f.Put(ctx, "p", []byte("x")) + + mux := http.NewServeMux() + f.RegisterRoutes(mux) + + ts := httptest.NewServer(mux) + defer ts.Close() + + // /stats returns the expected telemetry shape. + resp, err := http.Get(ts.URL + "/v1/admin/federation/stats") + if err != nil { + t.Fatalf("GET stats: %v", err) + } + defer resp.Body.Close() + var stats FederationStats + if err := json.NewDecoder(resp.Body).Decode(&stats); err != nil { + t.Fatalf("decode stats: %v", err) + } + if stats.Self != "us" || stats.Regions != 2 || stats.Policies != 1 { + t.Fatalf("stats = %+v, want self=us regions=2 policies=1", stats) + } + + // /regions returns the region list. + resp2, err := http.Get(ts.URL + "/v1/admin/federation/regions") + if err != nil { + t.Fatalf("GET regions: %v", err) + } + defer resp2.Body.Close() + var regions []Region + if err := json.NewDecoder(resp2.Body).Decode(®ions); err != nil { + t.Fatalf("decode regions: %v", err) + } + if len(regions) != 2 { + t.Fatalf("regions = %d, want 2", len(regions)) + } +} diff --git a/internal/federation/policy.go b/internal/federation/policy.go new file mode 100644 index 0000000..580b7f6 --- /dev/null +++ b/internal/federation/policy.go @@ -0,0 +1,215 @@ +// Package federation implements multi-region policy replication for the +// symkernel verification service, the production-orchestration tier called +// out by Milestone 11. +// +// A symkernel deployment may run in several regions for latency and +// availability. This package lets every region hold a local copy of the +// policy set and converge on the same view through an anti-entropy +// replication protocol (eventual consistency), and it routes work to the +// best healthy region for latency optimization and disaster recovery. +// +// The package is built from three cooperating pieces, each transport- and +// storage-agnostic so a production deployment can swap the network seam: +// +// - RegionRegistry / Router: tracks the participating regions, their +// observed latency and health, and selects a region for a request +// preferring low latency with automatic failover to a healthy peer when +// the preferred region is unavailable (disaster recovery). +// - PolicyStore: the local, per-region store of versioned policies. A +// MemoryStore ships for tests and single-process simulation. +// - Replicator + Transport: a background anti-entropy loop reconciles the +// local store against each peer region through the Transport seam, +// pushing newer local revisions and pulling newer remote ones and +// resolving concurrent writes by last-write-wins so the fleet converges. +// +// The Federation type wires the three pieces into a single façade and +// exposes telemetry through Stats / RegisterRoutes, mirroring the +// distributed-cache admin surface. As with internal/distributed, no real +// network client is required: the LocalTransport simulates a fleet of +// regions in one process so the replication protocol is exercised by tests. +package federation + +import ( + "context" + "errors" + "sort" + "sync" + "time" +) + +// ErrPolicyNotFound is returned by PolicyStore.Get and Transport.GetPolicy +// when a policy is absent. Callers should treat it as a miss, not a hard +// error (errors.Is is the supported check). +var ErrPolicyNotFound = errors.New("federation: policy not found") + +// ErrRegionNotFound is returned when a region id is unknown to the registry +// or transport. +var ErrRegionNotFound = errors.New("federation: region not found") + +// Policy is a versioned policy artifact replicated across regions. Body is +// opaque to the federation layer (typically a serialized CEL expression, a +// composed-policy document, or a Z3 constraint) so the package can carry any +// policy shape symkernel supports. +type Policy struct { + // ID is the stable policy identifier shared across all regions. + ID string `json:"id"` + + // Version is the monotonic revision number assigned by the region that + // authored this revision (OriginRegion). Replication compares versions to + // decide direction; ties are broken by UpdatedAt (last-write-wins). + Version uint64 `json:"version"` + + // Body is the opaque policy payload. + Body []byte `json:"body"` + + // OriginRegion is the region that authored this Version. It is carried + // for diagnostics and audit; it does not direct replication. + OriginRegion string `json:"origin_region"` + + // UpdatedAt is the wall-clock time of the last write, used to resolve + // concurrent writes from different regions (last-write-wins). + UpdatedAt time.Time `json:"updated_at"` +} + +// PolicyStore is the local, per-region policy storage contract. It holds the +// authoritative copy for policies authored in this region and a best-effort +// copy of policies replicated from peers. Implementations must be safe for +// concurrent use. +type PolicyStore interface { + // Get returns the policy for id, or an error wrapping ErrPolicyNotFound + // if it is absent. + Get(ctx context.Context, id string) (*Policy, error) + + // Put upserts the policy keyed by ID. It does not bump Version; callers + // assign a monotonic version on authored writes. + Put(ctx context.Context, p *Policy) error + + // Delete removes the policy for id. Deleting an absent policy is a no-op. + Delete(ctx context.Context, id string) error + + // List returns every policy currently held, ordered by ID. + List(ctx context.Context) ([]*Policy, error) +} + +// MemoryStore is an in-process PolicyStore backed by a map. It is the +// default store and the one used by tests; it is also a reasonable choice +// for single-region deployments that want the federation API without an +// external database. It is safe for concurrent use. +type MemoryStore struct { + mu sync.Mutex + policies map[string]*Policy +} + +// NewMemoryStore returns an empty MemoryStore. +func NewMemoryStore() *MemoryStore { + return &MemoryStore{policies: make(map[string]*Policy)} +} + +// Get returns a copy of the policy for id, or ErrPolicyNotFound if absent. +func (s *MemoryStore) Get(_ context.Context, id string) (*Policy, error) { + s.mu.Lock() + defer s.mu.Unlock() + p, ok := s.policies[id] + if !ok { + return nil, ErrPolicyNotFound + } + return clonePolicy(p), nil +} + +// Put stores a copy of p keyed by p.ID. +func (s *MemoryStore) Put(_ context.Context, p *Policy) error { + if p == nil { + return errors.New("federation: Put(nil policy)") + } + s.mu.Lock() + defer s.mu.Unlock() + s.policies[p.ID] = clonePolicy(p) + return nil +} + +// Delete removes the policy for id. It is a no-op for an absent id. +func (s *MemoryStore) Delete(_ context.Context, id string) error { + s.mu.Lock() + defer s.mu.Unlock() + delete(s.policies, id) + return nil +} + +// List returns copies of every held policy, ordered by ID. +func (s *MemoryStore) List(_ context.Context) ([]*Policy, error) { + s.mu.Lock() + defer s.mu.Unlock() + out := make([]*Policy, 0, len(s.policies)) + for _, p := range s.policies { + out = append(out, clonePolicy(p)) + } + sort.Slice(out, func(i, j int) bool { return out[i].ID < out[j].ID }) + return out, nil +} + +// clonePolicy returns a deep copy of p so callers cannot mutate stored state +// through the returned pointer or its Body slice. +func clonePolicy(p *Policy) *Policy { + if p == nil { + return nil + } + body := make([]byte, len(p.Body)) + copy(body, p.Body) + return &Policy{ + ID: p.ID, + Version: p.Version, + Body: body, + OriginRegion: p.OriginRegion, + UpdatedAt: p.UpdatedAt, + } +} + +// ReplicationMode controls how aggressively authored policies propagate to +// peer regions. +type ReplicationMode int + +const ( + // ReplicationAsync propagates policies via the background anti-entropy + // loop only. Writes return as soon as the local store is updated; peers + // observe the change within one scan interval (eventual consistency). + // This is the default and maximizes write availability. + ReplicationAsync ReplicationMode = iota + + // ReplicationSync additionally attempts an immediate one-shot push of a + // newly authored policy to every target region before the write returns. + // The local write is committed regardless (high availability); the + // returned error, if any, reports which peers could not be reached so the + // caller knows replication was partial. + ReplicationSync +) + +// ReplicationConfig governs which policies replicate to which regions and +// how often the anti-entropy loop runs. +type ReplicationConfig struct { + // Mode selects async (background) or sync (immediate-push) propagation. + // The zero value is ReplicationAsync. + Mode ReplicationMode + + // Interval is the anti-entropy scan period. Zero defaults to 30 seconds + // (see NewReplicator). Shorter intervals converge faster at the cost of + // more cross-region traffic. + Interval time.Duration + + // Regions is the allowlist of region IDs that receive replicated + // policies. When non-empty it overrides automatic peer discovery and the + // Replicator replicates to exactly these regions (minus the local one), + // letting an operator pin replication to a subset for cost or + // data-residency reasons. When empty the Replicator targets every healthy + // peer region in the registry. + Regions []string +} + +// modeString returns the telemetry name for a replication mode. +func modeString(m ReplicationMode) string { + switch m { + case ReplicationSync: + return "sync" + default: + return "async" + } +} diff --git a/internal/federation/region.go b/internal/federation/region.go new file mode 100644 index 0000000..d79bb0d --- /dev/null +++ b/internal/federation/region.go @@ -0,0 +1,209 @@ +package federation + +import ( + "sort" + "sync" + "time" +) + +// Region is a deployment region participating in federation. +type Region struct { + // ID is the stable region identifier (e.g. "us-east-1"). + ID string `json:"id"` + + // Endpoint is the address of the region's symkernel cluster, used by a + // real Transport to reach peer stores. Opaque to the registry. + Endpoint string `json:"endpoint,omitempty"` + + // Priority is a static routing preference; lower is preferred. It breaks + // ties when two regions report similar latency and lets operators pin a + // primary region for cost or data-residency reasons. + Priority int `json:"priority"` + + // Latency is the most recently observed round-trip time to the region, + // the signal region-aware routing optimizes for. Zero means "unknown", + // which ranks after all known latencies. + Latency time.Duration `json:"latency"` + + // Healthy gates whether the region receives routed traffic and automatic + // replication. A region marked unhealthy is skipped by the Router + // (disaster-recovery failover) and by the auto-discovery replication + // path, so a down region neither serves requests nor blocks convergence. + Healthy bool `json:"healthy"` +} + +// RegionRegistry tracks the regions known to this federation member, their +// observed latency, and health. It is safe for concurrent use. +type RegionRegistry struct { + mu sync.RWMutex + self string + regions map[string]*Region +} + +// NewRegionRegistry returns a registry seeded with the local region. self +// identifies the region this process runs in; if non-empty it is registered +// as healthy with zero (unknown) latency. +func NewRegionRegistry(self string) *RegionRegistry { + r := &RegionRegistry{self: self, regions: make(map[string]*Region)} + if self != "" { + r.regions[self] = &Region{ID: self, Healthy: true} + } + return r +} + +// Self returns the local region id. +func (r *RegionRegistry) Self() string { return r.self } + +// Register upserts a region. Use this to announce a peer region (with its +// endpoint and an initial latency/health estimate) or to refresh the local +// region's metadata. +func (r *RegionRegistry) Register(reg Region) { + r.mu.Lock() + defer r.mu.Unlock() + r.regions[reg.ID] = ® +} + +// Unregister removes a region. The local region is never removed; calling +// Unregister with Self is a no-op so a member cannot de-list itself. +func (r *RegionRegistry) Unregister(id string) { + r.mu.Lock() + defer r.mu.Unlock() + if id == r.self { + return + } + delete(r.regions, id) +} + +// SetHealth updates the healthy flag for id. It returns ErrRegionNotFound if +// id is not registered. +func (r *RegionRegistry) SetHealth(id string, healthy bool) error { + r.mu.Lock() + defer r.mu.Unlock() + reg, ok := r.regions[id] + if !ok { + return ErrRegionNotFound + } + reg.Healthy = healthy + return nil +} + +// SetLatency records the observed round-trip time to id. It returns +// ErrRegionNotFound if id is not registered. +func (r *RegionRegistry) SetLatency(id string, d time.Duration) error { + r.mu.Lock() + defer r.mu.Unlock() + reg, ok := r.regions[id] + if !ok { + return ErrRegionNotFound + } + reg.Latency = d + return nil +} + +// Get returns a snapshot of the region with id and whether it exists. +func (r *RegionRegistry) Get(id string) (Region, bool) { + r.mu.RLock() + defer r.mu.RUnlock() + reg, ok := r.regions[id] + if !ok { + return Region{}, false + } + return *reg, true +} + +// List returns snapshots of every region ordered by ID. +func (r *RegionRegistry) List() []Region { + r.mu.RLock() + defer r.mu.RUnlock() + out := make([]Region, 0, len(r.regions)) + for _, reg := range r.regions { + out = append(out, *reg) + } + sort.Slice(out, func(i, j int) bool { return out[i].ID < out[j].ID }) + return out +} + +// Healthy returns the healthy regions ranked best-first by the routing key: +// known latency before unknown, then lower latency, then lower priority, +// then ID. It is the order the Router selects from. +func (r *RegionRegistry) Healthy() []Region { + r.mu.RLock() + defer r.mu.RUnlock() + out := make([]Region, 0, len(r.regions)) + for _, reg := range r.regions { + if reg.Healthy { + out = append(out, *reg) + } + } + return rankRegions(out) +} + +// rankRegions returns a copy of in sorted by the routing key: regions with a +// known latency rank before unknown-latency ones, then by ascending latency, +// then ascending priority, then ID. The ordering is the heart of +// region-aware routing for latency optimization. +func rankRegions(in []Region) []Region { + out := append([]Region(nil), in...) + sort.Slice(out, func(i, j int) bool { + a, b := out[i], out[j] + ak, bk := a.Latency > 0, b.Latency > 0 + if ak != bk { + return ak // known latency ranks first + } + if a.Latency != b.Latency { + return a.Latency < b.Latency + } + if a.Priority != b.Priority { + return a.Priority < b.Priority + } + return a.ID < b.ID + }) + return out +} + +// Router performs region-aware routing over a RegionRegistry. Among the +// healthy regions it selects the one with the lowest observed latency +// (latency optimization), breaking ties by Priority then ID, and it fails +// over to the next-best healthy region when the preferred region is +// unavailable (disaster recovery). +type Router struct { + registry *RegionRegistry +} + +// NewRouter returns a Router backed by reg. +func NewRouter(reg *RegionRegistry) *Router { + return &Router{registry: reg} +} + +// Route returns the best healthy region, or ErrRegionNotFound if no region +// is healthy. "Best" follows RegionRegistry.Healthy's ranking: lowest known +// latency, then priority, then ID. +func (rt *Router) Route() (Region, error) { + healthy := rt.registry.Healthy() + if len(healthy) == 0 { + return Region{}, ErrRegionNotFound + } + return healthy[0], nil +} + +// RouteFailover returns preferred when it is healthy; otherwise it returns +// the best healthy peer. This is the disaster-recovery routing path: a +// request normally bound for preferred is rerouted to a healthy peer the +// moment preferred is down. Returns ErrRegionNotFound if no region at all is +// healthy. +func (rt *Router) RouteFailover(preferred string) (Region, error) { + healthy := rt.registry.Healthy() + if len(healthy) == 0 { + return Region{}, ErrRegionNotFound + } + if preferred != "" { + for _, reg := range healthy { + if reg.ID == preferred { + return reg, nil + } + } + } + // preferred is absent or unhealthy; healthy is already best-first and + // cannot contain an unhealthy preferred region. + return healthy[0], nil +} diff --git a/internal/federation/replicator.go b/internal/federation/replicator.go new file mode 100644 index 0000000..032875b --- /dev/null +++ b/internal/federation/replicator.go @@ -0,0 +1,573 @@ +package federation + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "net/http" + "sync" + "sync/atomic" + "time" +) + +// defaultReplicationInterval is the anti-entropy scan period applied when +// ReplicationConfig.Interval is unset. +const defaultReplicationInterval = 30 * time.Second + +// Transport is the cross-region delivery seam: it lets the Replicator read +// and write policies on a peer region without coupling to a concrete network +// client. LocalTransport is the in-process implementation; a production +// deployment plugs in an HTTP client speaking the symkernel federation API. +type Transport interface { + // GetPolicy fetches the current revision of id from region, returning an + // error wrapping ErrPolicyNotFound when the region does not hold it. + GetPolicy(ctx context.Context, region, id string) (*Policy, error) + + // PutPolicy upserts p on region. + PutPolicy(ctx context.Context, region string, p *Policy) error + + // ListPolicies returns every policy held by region. + ListPolicies(ctx context.Context, region string) ([]*Policy, error) + + // Ping verifies region reachability. + Ping(ctx context.Context, region string) error +} + +// LocalTransport is an in-process Transport that routes operations to a set +// of named PolicyStores, one per simulated region. It is the transport used +// by tests and by single-process multi-region simulation; a real deployment +// replaces it with an HTTP client. Operations on a region that has no +// mounted store return ErrRegionNotFound. +type LocalTransport struct { + mu sync.RWMutex + stores map[string]PolicyStore +} + +// NewLocalTransport returns a Transport with no regions mounted. +func NewLocalTransport() *LocalTransport { + return &LocalTransport{stores: make(map[string]PolicyStore)} +} + +// Mount binds store as the destination for region, replacing any previous +// binding. Mounting a region into a shared LocalTransport is how a +// single-process test simulates a fleet of peer regions. +func (t *LocalTransport) Mount(region string, store PolicyStore) { + t.mu.Lock() + defer t.mu.Unlock() + t.stores[region] = store +} + +// Unmount removes the binding for region. +func (t *LocalTransport) Unmount(region string) { + t.mu.Lock() + defer t.mu.Unlock() + delete(t.stores, region) +} + +func (t *LocalTransport) storeFor(region string) (PolicyStore, error) { + t.mu.RLock() + defer t.mu.RUnlock() + s, ok := t.stores[region] + if !ok { + return nil, ErrRegionNotFound + } + return s, nil +} + +// GetPolicy fetches id from the store mounted at region. +func (t *LocalTransport) GetPolicy(ctx context.Context, region, id string) (*Policy, error) { + s, err := t.storeFor(region) + if err != nil { + return nil, err + } + return s.Get(ctx, id) +} + +// PutPolicy upserts p on the store mounted at region. +func (t *LocalTransport) PutPolicy(ctx context.Context, region string, p *Policy) error { + s, err := t.storeFor(region) + if err != nil { + return err + } + return s.Put(ctx, p) +} + +// ListPolicies returns every policy held by the store mounted at region. +func (t *LocalTransport) ListPolicies(ctx context.Context, region string) ([]*Policy, error) { + s, err := t.storeFor(region) + if err != nil { + return nil, err + } + return s.List(ctx) +} + +// Ping reports whether a store is mounted at region. +func (t *LocalTransport) Ping(_ context.Context, region string) error { + t.mu.RLock() + defer t.mu.RUnlock() + if _, ok := t.stores[region]; !ok { + return ErrRegionNotFound + } + return nil +} + +// Replicator drives the anti-entropy replication loop. On each scan it +// reconciles the local store against every target region reachable through +// the Transport: for each policy present on either side, the newer revision +// (by Version, ties by UpdatedAt) wins and is copied to the losing side. +// Repeated scans converge the fleet to the same view — eventual consistency. +// +// Target regions come from ReplicationConfig.Regions when that allowlist is +// non-empty, otherwise from the registry's healthy peers; the local (self) +// region is never a target. +type Replicator struct { + self string + store PolicyStore + registry *RegionRegistry + transport Transport + cfg ReplicationConfig + + stopCh chan struct{} + wg sync.WaitGroup + + // Telemetry counters (atomic for lock-free reads on the hot path). + scans uint64 + pushed uint64 + pulled uint64 + skipped uint64 + errors uint64 +} + +// NewReplicator returns a Replicator that reconciles store against the +// configured targets. A zero Interval defaults to 30 seconds. +func NewReplicator(self string, store PolicyStore, registry *RegionRegistry, transport Transport, cfg ReplicationConfig) *Replicator { + if cfg.Interval <= 0 { + cfg.Interval = defaultReplicationInterval + } + return &Replicator{ + self: self, + store: store, + registry: registry, + transport: transport, + cfg: cfg, + stopCh: make(chan struct{}), + } +} + +// Targets returns the region IDs this replicator pushes to and pulls from: +// the configured allowlist (minus self) when non-empty, otherwise the +// registry's healthy peers (minus self). +func (rep *Replicator) Targets() []string { + if len(rep.cfg.Regions) > 0 { + out := make([]string, 0, len(rep.cfg.Regions)) + seen := make(map[string]bool, len(rep.cfg.Regions)) + for _, id := range rep.cfg.Regions { + if id == rep.self || seen[id] { + continue + } + seen[id] = true + out = append(out, id) + } + return out + } + healthy := rep.registry.Healthy() + out := make([]string, 0, len(healthy)) + for _, reg := range healthy { + if reg.ID == rep.self { + continue + } + out = append(out, reg.ID) + } + return out +} + +// Reconcile performs one anti-entropy pass over every target region and +// returns a joined error of any per-region failures. Partial failures do not +// abort the pass: a region that cannot be reached is skipped and the +// remaining regions still reconcile. +func (rep *Replicator) Reconcile(ctx context.Context) error { + atomic.AddUint64(&rep.scans, 1) + var errs []error + for _, region := range rep.Targets() { + if err := rep.reconcileRegion(ctx, region); err != nil { + errs = append(errs, fmt.Errorf("region %q: %w", region, err)) + } + } + return errors.Join(errs...) +} + +// reconcileRegion reconciles the local store against a single peer. +func (rep *Replicator) reconcileRegion(ctx context.Context, region string) error { + local, err := rep.store.List(ctx) + if err != nil { + atomic.AddUint64(&rep.errors, 1) + return fmt.Errorf("list local: %w", err) + } + remote, err := rep.transport.ListPolicies(ctx, region) + if err != nil { + atomic.AddUint64(&rep.errors, 1) + return fmt.Errorf("list remote: %w", err) + } + + remoteByID := make(map[string]*Policy, len(remote)) + for _, p := range remote { + remoteByID[p.ID] = p + } + + var errs []error + for _, lp := range local { + rp := remoteByID[lp.ID] + delete(remoteByID, lp.ID) // remainder becomes remote-only pulls + if err := rep.reconcilePolicy(ctx, region, lp, rp); err != nil { + errs = append(errs, err) + } + } + for _, rp := range remoteByID { + if err := rep.reconcilePolicy(ctx, region, nil, rp); err != nil { + errs = append(errs, err) + } + } + return errors.Join(errs...) +} + +// reconcilePolicy reconciles a single policy across the local store and a +// peer region, copying the newer revision to the losing side. local/remote +// may be nil to signal "absent on this side". +func (rep *Replicator) reconcilePolicy(ctx context.Context, region string, local, remote *Policy) error { + switch { + case local == nil && remote == nil: + return nil + case local == nil: + // remote-only: pull. + if err := rep.store.Put(ctx, remote); err != nil { + atomic.AddUint64(&rep.errors, 1) + return fmt.Errorf("pull %q: %w", remote.ID, err) + } + atomic.AddUint64(&rep.pulled, 1) + return nil + case remote == nil: + // local-only: push. + if err := rep.transport.PutPolicy(ctx, region, local); err != nil { + atomic.AddUint64(&rep.errors, 1) + return fmt.Errorf("push %q: %w", local.ID, err) + } + atomic.AddUint64(&rep.pushed, 1) + return nil + } + + switch compareRevision(local, remote) { + case 1: // local newer: push. + if err := rep.transport.PutPolicy(ctx, region, local); err != nil { + atomic.AddUint64(&rep.errors, 1) + return fmt.Errorf("push %q: %w", local.ID, err) + } + atomic.AddUint64(&rep.pushed, 1) + case -1: // remote newer: pull. + if err := rep.store.Put(ctx, remote); err != nil { + atomic.AddUint64(&rep.errors, 1) + return fmt.Errorf("pull %q: %w", remote.ID, err) + } + atomic.AddUint64(&rep.pulled, 1) + default: + atomic.AddUint64(&rep.skipped, 1) + } + return nil +} + +// compareRevision returns 1 if a is newer than b, -1 if older, 0 if equal. +// Newer means a higher Version; when versions tie — because two regions +// authored different revisions of the same policy concurrently — the later +// UpdatedAt wins (last-write-wins). Equal Version and UpdatedAt is a no-op. +func compareRevision(a, b *Policy) int { + if a.Version != b.Version { + if a.Version > b.Version { + return 1 + } + return -1 + } + switch { + case a.UpdatedAt.Before(b.UpdatedAt): + return -1 + case a.UpdatedAt.After(b.UpdatedAt): + return 1 + default: + return 0 + } +} + +// Start launches the background anti-entropy loop. Call Stop to terminate it. +// The loop is optional: callers may drive replication deterministically via +// Reconcile instead. +func (rep *Replicator) Start() { + rep.wg.Add(1) + go rep.loop() +} + +func (rep *Replicator) loop() { + defer rep.wg.Done() + ticker := time.NewTicker(rep.cfg.Interval) + defer ticker.Stop() + for { + select { + case <-rep.stopCh: + return + case <-ticker.C: + _ = rep.Reconcile(context.Background()) + } + } +} + +// Stop terminates the background loop and waits for it to exit. It is safe +// to call Stop only after Start and only once. +func (rep *Replicator) Stop() { + close(rep.stopCh) + rep.wg.Wait() +} + +// Federation wires a local PolicyStore, RegionRegistry, Transport, Router, +// and Replicator into a single multi-region policy-replication façade. It is +// the primary entry point for the federation tier; the zero value is +// unusable, use New. +type Federation struct { + self string + store PolicyStore + registry *RegionRegistry + transport Transport + router *Router + replicator *Replicator + cfg ReplicationConfig + + // authored counts policies accepted via Put (the local-author path). + authored uint64 + // syncPushes counts policies pushed immediately by sync-mode Put, which + // bypasses the anti-entropy loop; it is separate from the Replicator's + // push counter so operators can tell the two paths apart. + syncPushes uint64 +} + +// Option configures a Federation during construction. +type Option func(*Federation) + +// WithStore installs a specific local PolicyStore. Defaults to a MemoryStore. +func WithStore(s PolicyStore) Option { return func(f *Federation) { f.store = s } } + +// WithRegistry installs a specific RegionRegistry. Defaults to one seeded +// with the local region. +func WithRegistry(r *RegionRegistry) Option { return func(f *Federation) { f.registry = r } } + +// WithTransport installs a specific Transport. Defaults to a LocalTransport. +func WithTransport(t Transport) Option { return func(f *Federation) { f.transport = t } } + +// WithConfig sets the replication configuration (mode, interval, allowlist). +func WithConfig(cfg ReplicationConfig) Option { return func(f *Federation) { f.cfg = cfg } } + +// New creates a Federation for the local region self with the given options. +// Missing dependencies are defaulted so New("us-east-1") yields a usable, +// single-region federation; mount peer regions and stores on the Transport +// (or supply your own) to go multi-region. +func New(self string, opts ...Option) *Federation { + f := &Federation{self: self} + for _, o := range opts { + o(f) + } + if f.store == nil { + f.store = NewMemoryStore() + } + if f.registry == nil { + f.registry = NewRegionRegistry(self) + } + if f.transport == nil { + f.transport = NewLocalTransport() + } + if f.cfg.Interval <= 0 { + f.cfg.Interval = defaultReplicationInterval + } + f.router = NewRouter(f.registry) + f.replicator = NewReplicator(self, f.store, f.registry, f.transport, f.cfg) + return f +} + +// Store returns the local policy store. +func (f *Federation) Store() PolicyStore { return f.store } + +// Registry returns the region registry. +func (f *Federation) Registry() *RegionRegistry { return f.registry } + +// Router returns the region-aware router. +func (f *Federation) Router() *Router { return f.router } + +// Transport returns the cross-region transport. +func (f *Federation) Transport() Transport { return f.transport } + +// Replicator returns the anti-entropy replicator. +func (f *Federation) Replicator() *Replicator { return f.replicator } + +// Regions returns a snapshot of every known region. +func (f *Federation) Regions() []Region { return f.registry.List() } + +// SetRegionHealth updates a region's health gate, used by routing and +// auto-discovery replication. Returns ErrRegionNotFound for an unknown id. +func (f *Federation) SetRegionHealth(id string, healthy bool) error { + return f.registry.SetHealth(id, healthy) +} + +// SetRegionLatency records an observed round-trip time to a region, the +// signal region-aware routing optimizes for. Returns ErrRegionNotFound for an +// unknown id. +func (f *Federation) SetRegionLatency(id string, d time.Duration) error { + return f.registry.SetLatency(id, d) +} + +// Put authors a new revision of the policy id in the local store and +// propagates it according to the configured replication mode. In async mode +// (the default) the write returns once the local store is updated and the +// background loop (if started) carries it to peers. In sync mode the write +// additionally attempts an immediate push to every target region; the local +// revision is committed regardless, and a non-nil returned error reports +// which peers could not be reached so the caller knows replication was +// partial. +func (f *Federation) Put(ctx context.Context, id string, body []byte) (*Policy, error) { + if id == "" { + return nil, errors.New("federation: empty policy id") + } + cur, err := f.store.Get(ctx, id) + if err != nil && !errors.Is(err, ErrPolicyNotFound) { + return nil, err + } + p := &Policy{ + ID: id, + Version: 1, + Body: body, + OriginRegion: f.self, + UpdatedAt: time.Now(), + } + if cur != nil { + p.Version = cur.Version + 1 + } + if err := f.store.Put(ctx, p); err != nil { + return nil, err + } + atomic.AddUint64(&f.authored, 1) + + if f.cfg.Mode == ReplicationSync { + // Local revision is committed; report partial replication only. + if perr := f.pushNow(ctx, p); perr != nil { + return p, fmt.Errorf("federation: replicated locally, partial push: %w", perr) + } + } + return p, nil +} + +// pushNow synchronously pushes p to every target region, joining per-region +// failures into one error. +func (f *Federation) pushNow(ctx context.Context, p *Policy) error { + var errs []error + for _, region := range f.replicator.Targets() { + if err := f.transport.PutPolicy(ctx, region, p); err != nil { + errs = append(errs, fmt.Errorf("region %q: %w", region, err)) + continue + } + atomic.AddUint64(&f.syncPushes, 1) + } + return errors.Join(errs...) +} + +// Get returns the locally held revision of id, or ErrPolicyNotFound. +func (f *Federation) Get(ctx context.Context, id string) (*Policy, error) { + return f.store.Get(ctx, id) +} + +// Delete removes id from the local store. Deletion is local only: without a +// tombstone a peer that still holds the policy will re-replicate it on the +// next anti-entropy scan. Tombstone-based deletion is a follow-up; for now a +// delete intended to take fleet-wide effect should be expressed as a new +// "disabled" revision via Put. +func (f *Federation) Delete(ctx context.Context, id string) error { + return f.store.Delete(ctx, id) +} + +// Reconcile performs one anti-entropy pass against every target region. It +// is the explicit equivalent of one tick of the background loop and is the +// way tests and batch jobs drive deterministic convergence. +func (f *Federation) Reconcile(ctx context.Context) error { + return f.replicator.Reconcile(ctx) +} + +// Start launches the background anti-entropy loop. Call Stop to terminate. +func (f *Federation) Start() { f.replicator.Start() } + +// Stop terminates the background loop and waits for it to exit. +func (f *Federation) Stop() { f.replicator.Stop() } + +// FederationStats is a point-in-time snapshot of the federation state and +// replication telemetry. +type FederationStats struct { + Self string `json:"self"` + Mode string `json:"mode"` + Interval string `json:"interval"` + Regions int `json:"regions"` + HealthyRegions int `json:"healthy_regions"` + Targets int `json:"targets"` + Policies int `json:"policies"` + Authored uint64 `json:"authored"` + SyncPushes uint64 `json:"sync_pushes"` + Scans uint64 `json:"scans"` + Pushed uint64 `json:"pushed"` + Pulled uint64 `json:"pulled"` + Skipped uint64 `json:"skipped"` + ReplicationErrors uint64 `json:"replication_errors"` +} + +// Stats returns a point-in-time snapshot of federation state and the +// replication counters. +func (f *Federation) Stats() FederationStats { + st := FederationStats{ + Self: f.self, + Mode: modeString(f.cfg.Mode), + Interval: f.cfg.Interval.String(), + Authored: atomic.LoadUint64(&f.authored), + SyncPushes: atomic.LoadUint64(&f.syncPushes), + Scans: atomic.LoadUint64(&f.replicator.scans), + Pushed: atomic.LoadUint64(&f.replicator.pushed), + Pulled: atomic.LoadUint64(&f.replicator.pulled), + Skipped: atomic.LoadUint64(&f.replicator.skipped), + ReplicationErrors: atomic.LoadUint64(&f.replicator.errors), + Targets: len(f.replicator.Targets()), + } + for _, reg := range f.registry.List() { + st.Regions++ + if reg.Healthy { + st.HealthyRegions++ + } + } + if list, err := f.store.List(context.Background()); err == nil { + st.Policies = len(list) + } + return st +} + +// RegisterRoutes mounts the federation admin endpoints on mux: +// +// GET /v1/admin/federation/stats — federation telemetry +// GET /v1/admin/federation/regions — known regions and health +func (f *Federation) RegisterRoutes(mux *http.ServeMux) { + mux.Handle("GET /v1/admin/federation/stats", f.statsHandler()) + mux.Handle("GET /v1/admin/federation/regions", f.regionsHandler()) +} + +// statsHandler handles GET /v1/admin/federation/stats. +func (f *Federation) statsHandler() http.HandlerFunc { + return func(w http.ResponseWriter, _ *http.Request) { + w.Header().Set("Content-Type", "application/json") + json.NewEncoder(w).Encode(f.Stats()) //nolint:errcheck // partial write to ResponseWriter is unrecoverable + } +} + +// regionsHandler handles GET /v1/admin/federation/regions. +func (f *Federation) regionsHandler() http.HandlerFunc { + return func(w http.ResponseWriter, _ *http.Request) { + w.Header().Set("Content-Type", "application/json") + json.NewEncoder(w).Encode(f.Regions()) //nolint:errcheck // partial write to ResponseWriter is unrecoverable + } +}