1package broker
2
3import (
4 "errors"
5 "testing"
6 "time"
7
8 "github.com/bnema/vev/internal/domain"
9 "github.com/bnema/vev/internal/ports"
10 portsmocks "github.com/bnema/vev/internal/ports/mocks"
11 "github.com/stretchr/testify/require"
12)
13
14func TestRegistryRestoresAuthoritativeMembership(t *testing.T) {
15 for _, mode := range []string{"removed", "replaced", "never observed", "matching"} {
16 t.Run(mode, func(t *testing.T) {
17 old := registration(t, "host", 1)
18 current := old
19 store := portsmocks.NewMockBrokerHostStore(t)
20 snapshot := ports.BrokerSnapshot{Epoch: 1, Revision: 9, Daemons: []ports.BrokerDaemonObservation{{Endpoint: "host", Registration: old, Availability: domain.RemoteAvailabilityReachable, InventoryKnown: true}}}
21 hosts := ports.BrokerHosts{Revision: 4}
22 if mode == "replaced" {
23 current.Generation++
24 }
25 if mode != "removed" {
26 hosts.Hosts = []ports.BrokerHostRecord{{Registration: current, Pinned: true, Policy: poolPolicy(), Route: canonicalRoute(poolPolicy(), current.Endpoint)}}
27 }
28 if mode == "never observed" {
29 snapshot = ports.BrokerSnapshot{}
30 }
31 store.EXPECT().Load().Return(snapshot, nil).Once()
32 store.EXPECT().LoadHosts().Return(hosts, nil).Once()
33 r, err := NewRegistry(2, store, newTestProbe(1), newManualClock(time.Unix(100, 0)), nil)
34 require.NoError(t, err)
35 got := r.Snapshot()
36 require.Equal(t, ports.BrokerEpoch(2), got.Epoch)
37 require.Equal(t, ports.BrokerRevision(1), got.Revision)
38 if mode == "removed" {
39 require.Empty(t, got.Daemons)
40 return
41 }
42 require.Len(t, got.Daemons, 1)
43 require.Equal(t, current, got.Daemons[0].Registration)
44 require.Equal(t, mode == "matching", got.Daemons[0].InventoryKnown)
45 })
46 }
47}
48
49// TestRegistryRejectsAuthorityAndLoadFailures covers the negative store seam:
50// NewRegistry requires BrokerHostStore explicitly, so a load error, an
51// authority error, or authority that fails validation is fatal instead of
52// being skipped as an optional capability.
53func TestRegistryRejectsAuthorityAndLoadFailures(t *testing.T) {
54 clock := newManualClock(time.Unix(100, 0))
55 reg := registration(t, "host", 1)
56 loadErr := errors.New("snapshot load failed")
57 authorityErr := errors.New("authority load failed")
58 invalidAuthority := map[string]ports.BrokerHosts{
59 "missing revision": {Hosts: []ports.BrokerHostRecord{{Registration: reg, Pinned: true, Policy: poolPolicy(), Route: canonicalRoute(poolPolicy(), reg.Endpoint)}}},
60 "duplicate endpoint": {Revision: 1, Hosts: []ports.BrokerHostRecord{
61 {Registration: reg, Pinned: true, Policy: poolPolicy(), Route: canonicalRoute(poolPolicy(), reg.Endpoint)},
62 {Registration: reg, Learned: true, Policy: poolPolicy(), Route: canonicalRoute(poolPolicy(), reg.Endpoint)},
63 }},
64 "unanchored record": {Revision: 1, Hosts: []ports.BrokerHostRecord{{Registration: reg, Policy: poolPolicy(), Route: canonicalRoute(poolPolicy(), reg.Endpoint)}}},
65 "zero policy": {Revision: 1, Hosts: []ports.BrokerHostRecord{{Registration: reg, Pinned: true, Route: canonicalRoute(poolPolicy(), reg.Endpoint)}}},
66 }
67 for name, hosts := range invalidAuthority {
68 t.Run(name, func(t *testing.T) {
69 store := portsmocks.NewMockBrokerHostStore(t)
70 store.EXPECT().Load().Return(ports.BrokerSnapshot{}, nil).Once()
71 store.EXPECT().LoadHosts().Return(hosts, nil).Once()
72 registry, err := NewRegistry(1, store, newTestProbe(1), clock, nil)
73 require.Error(t, err)
74 require.Nil(t, registry)
75 })
76 }
77 t.Run("snapshot load error", func(t *testing.T) {
78 store := portsmocks.NewMockBrokerHostStore(t)
79 store.EXPECT().Load().Return(ports.BrokerSnapshot{}, loadErr).Once()
80 registry, err := NewRegistry(1, store, newTestProbe(1), clock, nil)
81 require.ErrorIs(t, err, loadErr)
82 require.Nil(t, registry)
83 })
84 t.Run("authority load error", func(t *testing.T) {
85 store := portsmocks.NewMockBrokerHostStore(t)
86 store.EXPECT().Load().Return(ports.BrokerSnapshot{}, nil).Once()
87 store.EXPECT().LoadHosts().Return(ports.BrokerHosts{}, authorityErr).Once()
88 registry, err := NewRegistry(1, store, newTestProbe(1), clock, nil)
89 require.ErrorIs(t, err, authorityErr)
90 require.Nil(t, registry)
91 })
92}
93
94// TestRegistryRestoreIsReadOnlyAndPublicationLocal pins the revision series: the
95// durable authority revision is store-local and never seeds the publication
96// series, and construction writes nothing back.
97func TestRegistryRestoreIsReadOnlyAndPublicationLocal(t *testing.T) {
98 clock := newManualClock(time.Unix(100, 0))
99 reg := registration(t, "host", 1)
100 store := portsmocks.NewMockBrokerHostStore(t)
101 store.EXPECT().Load().Return(ports.BrokerSnapshot{Epoch: 5, Revision: 4096, Daemons: []ports.BrokerDaemonObservation{{
102 Endpoint: reg.Endpoint, Registration: reg, Availability: domain.RemoteAvailabilityReachable, InventoryKnown: true,
103 }}}, nil).Once()
104 store.EXPECT().LoadHosts().Return(ports.BrokerHosts{Revision: 99, Hosts: []ports.BrokerHostRecord{{Registration: reg, Pinned: true, Policy: poolPolicy(), Route: canonicalRoute(poolPolicy(), reg.Endpoint)}}}, nil).Once()
105
106 r, err := NewRegistry(6, store, newTestProbe(1), clock, nil)
107 require.NoError(t, err)
108 snapshot := r.Snapshot()
109 require.Equal(t, ports.BrokerRevision(1), snapshot.Revision)
110 require.Equal(t, ports.BrokerEpoch(6), snapshot.Epoch)
111 // Restoring reuses the matching observation but never republishes it: a
112 // fresh process adopts authority and rewrites only when it observes.
113 require.Len(t, snapshot.Daemons, 1)
114 require.True(t, snapshot.Daemons[0].InventoryKnown)
115 require.Equal(t, domain.RemoteAvailabilityReachable, snapshot.Daemons[0].Availability)
116}