1package broker
2
3import (
4 "context"
5 "path/filepath"
6 "testing"
7 "time"
8
9 "github.com/bnema/vev/internal/adapters/brokerstore"
10 "github.com/bnema/vev/internal/domain"
11 "github.com/bnema/vev/internal/ports"
12 portsmocks "github.com/bnema/vev/internal/ports/mocks"
13 "github.com/bnema/vev/internal/protocol"
14 "github.com/stretchr/testify/require"
15)
16
17// Exercise the real CAS and snapshot validator, not the observation-only fake.
18// Each stage closes and reopens both owners, so membership and observations
19// must come from disk, including the empty set and a fresh re-added identity.
20func TestRegistryDurableReplaceHostsRestart(t *testing.T) {
21 dir := filepath.Join(t.TempDir(), "broker")
22 first := ports.BrokerHostRecord{Registration: registration(t, "host", 1), Pinned: true, Policy: poolPolicy(), Route: canonicalRoute(poolPolicy(), "host")}
23 readded := first
24 readded.Registration = registration(t, "host", 2)
25 readded.Pinned, readded.Learned = false, true
26 readded.Policy.Trust = "new-trust"
27 var previous []ports.BrokerHostRecord
28 for i, stage := range []struct {
29 name string
30 records []ports.BrokerHostRecord
31 }{
32 {"add", []ports.BrokerHostRecord{first}}, {"remove", nil}, {"re-add", []ports.BrokerHostRecord{readded}}, {"restart", []ports.BrokerHostRecord{readded}},
33 } {
34 t.Run(stage.name, func(t *testing.T) {
35 store, err := brokerstore.Open(brokerstore.Options{Dir: dir})
36 require.NoError(t, err)
37 defer store.Close()
38 r, err := NewRegistry(ports.BrokerEpoch(i+1), store, newTestProbe(1), newManualClock(time.Unix(100, 0)), nil)
39 require.NoError(t, err)
40 defer r.settle()
41 require.Equal(t, previous, r.authority.Hosts)
42 require.Len(t, r.Snapshot().Daemons, len(previous))
43 if len(previous) > 0 {
44 require.True(t, r.Snapshot().Daemons[0].InventoryKnown, "advisory observation must also survive restart")
45 }
46 require.NoError(t, r.ReplaceHosts(stage.records))
47 authority, err := store.LoadHosts()
48 require.NoError(t, err)
49 require.Equal(t, r.authority, authority)
50 require.Equal(t, stage.records, authority.Hosts)
51 require.Len(t, r.Snapshot().Daemons, len(stage.records))
52 if len(stage.records) > 0 {
53 reg := stage.records[0].Registration
54 ctx, cancel := context.WithCancel(context.Background())
55 defer cancel()
56 r.inflight[reg.Endpoint] = &probeAttempt{token: 1, cancel: cancel}
57 r.apply(probeResult{endpoint: reg.Endpoint, registration: reg, attempt: 1, at: r.clock.Now(), snapshot: ports.BrokerDaemonObservation{Availability: domain.RemoteAvailabilityReachable, InventoryKnown: true}})
58 require.ErrorIs(t, ctx.Err(), context.Canceled)
59 }
60 r.settle() // flush before inspecting persisted projection
61 snapshot, err := store.Load()
62 require.NoError(t, err)
63 require.Equal(t, durable(r.Snapshot()), snapshot)
64 require.ErrorIs(t, r.ReplaceHosts(stage.records), ports.ErrBrokerRegistryClosed, "shutdown must not acknowledge mutations")
65 previous = stage.records
66 })
67 }
68}
69
70func TestRegistryDurableConflictDoesNotMutateProjection(t *testing.T) {
71 for _, mode := range []string{"stale revision", "policy without new generation"} {
72 t.Run(mode, func(t *testing.T) {
73 store, err := brokerstore.Open(brokerstore.Options{Dir: filepath.Join(t.TempDir(), "broker")})
74 require.NoError(t, err)
75 defer store.Close()
76 record := ports.BrokerHostRecord{Registration: registration(t, "host", 1), Pinned: true, Policy: poolPolicy(), Route: canonicalRoute(poolPolicy(), "host")}
77 require.NoError(t, store.ReplaceHosts(1, []ports.BrokerHostRecord{record}))
78 r, err := NewRegistry(1, store, newTestProbe(1), newManualClock(time.Unix(100, 0)), nil)
79 require.NoError(t, err)
80 defer r.settle()
81 before := r.Snapshot()
82 authority := r.authority
83 ctx, cancel := context.WithCancel(context.Background())
84 defer cancel()
85 r.inflight[record.Registration.Endpoint] = &probeAttempt{token: 1, cancel: cancel}
86 if mode == "stale revision" {
87 // Simulate an illicit second mutator. Even identical local records must
88 // CAS rather than returning success from a projection-only no-op.
89 require.NoError(t, store.ReplaceHosts(authority.Revision, nil))
90 } else {
91 record.Policy.Trust = "changed-trust"
92 }
93 require.ErrorIs(t, r.ReplaceHosts([]ports.BrokerHostRecord{record}), ports.ErrBrokerHostConflict)
94 require.Equal(t, before, r.Snapshot())
95 require.Equal(t, authority, r.authority)
96 require.NoError(t, ctx.Err(), "failed CAS must not cancel the current observation")
97 })
98 }
99}
100
101func TestRegistryReplaceHostsStoreFailureIsSynchronous(t *testing.T) {
102 store := portsmocks.NewMockBrokerHostStore(t)
103 store.EXPECT().Load().Return(ports.BrokerSnapshot{}, nil).Once()
104 store.EXPECT().LoadHosts().Return(ports.BrokerHosts{Revision: 7}, nil).Once()
105 record := ports.BrokerHostRecord{Registration: registration(t, "host", 1), Pinned: true, Policy: poolPolicy(), Route: canonicalRoute(poolPolicy(), "host")}
106 records := []ports.BrokerHostRecord{record}
107 store.EXPECT().ReplaceHosts(uint64(7), records).Return(context.DeadlineExceeded).Once()
108 r, err := NewRegistry(1, store, newTestProbe(1), newManualClock(time.Unix(100, 0)), nil)
109 require.NoError(t, err)
110 before := r.Snapshot()
111 require.ErrorIs(t, r.ReplaceHosts(records), context.DeadlineExceeded)
112 require.Equal(t, before, r.Snapshot())
113 require.Equal(t, uint64(7), r.authority.Revision)
114 require.Error(t, r.ReplaceHosts([]ports.BrokerHostRecord{{Registration: record.Registration}}))
115 require.Equal(t, before, r.Snapshot())
116}
117
118// TestRegistryReplaceHostsAfterShutdownIsTyped pins the closed sentinel: once
119// the run has settled, a membership mutation is refused with a typed error, so
120// a caller can tell shutdown apart from a store failure without matching text.
121func TestRegistryReplaceHostsAfterShutdownIsTyped(t *testing.T) {
122 store := newTestStore()
123 r := newTestRegistry(t, 1, store, newTestProbe(1), newManualClock(time.Unix(100, 0)))
124 record := ports.BrokerHostRecord{Registration: registration(t, "host", 1), Pinned: true, Policy: poolPolicy(), Route: canonicalRoute(poolPolicy(), "host")}
125 require.NoError(t, r.setHosts(hostRecords(record.Registration)))
126 r.settle()
127
128 before := r.Snapshot()
129 authority := r.authority
130 err := r.ReplaceHosts([]ports.BrokerHostRecord{record})
131 require.ErrorIs(t, err, ports.ErrBrokerRegistryClosed)
132 require.NotErrorIs(t, err, ports.ErrBrokerRevisionExhausted)
133 require.Equal(t, before, r.Snapshot())
134 require.Equal(t, authority, r.authority)
135}
136
137// Queued pre-CAS observations may be rejected as stale by the real store, but
138// the newest post-CAS publication must always flush; churn cannot leave the
139// registry permanently publishing membership that the store will reject.
140func TestRegistryDurableChurnFlushesLatestMembership(t *testing.T) {
141 store, err := brokerstore.Open(brokerstore.Options{Dir: filepath.Join(t.TempDir(), "broker")})
142 require.NoError(t, err)
143 defer store.Close()
144 r, err := NewRegistry(1, store, newTestProbe(1), newManualClock(time.Unix(100, 0)), nil)
145 require.NoError(t, err)
146 defer r.settle()
147 for i := byte(1); i <= 20; i++ {
148 record := ports.BrokerHostRecord{Registration: registration(t, "host", i), Pinned: true, Policy: poolPolicy(), Route: canonicalRoute(poolPolicy(), "host")}
149 require.NoError(t, r.ReplaceHosts([]ports.BrokerHostRecord{record}))
150 // A policy change is admitted only with a new registration generation.
151 record.Registration.Generation++
152 record.Policy.Trust = "replacement-trust"
153 require.NoError(t, r.ReplaceHosts([]ports.BrokerHostRecord{record}))
154 require.Equal(t, record.Registration, r.Snapshot().Daemons[0].Registration)
155 if i < 20 {
156 require.NoError(t, r.ReplaceHosts(nil))
157 }
158 }
159 r.settle()
160 snapshot, err := store.Load()
161 require.NoError(t, err)
162 require.Equal(t, durable(r.Snapshot()), snapshot)
163 authority, err := store.LoadHosts()
164 require.NoError(t, err)
165 require.Equal(t, r.authority, authority)
166}
167
168// TestRegistryDurableRoundTripObservedIdentity proves the durable format carries
169// observed daemon identity, incarnation, protocol version, and capabilities
170// across a restart. The durable copy deliberately omits policy; the loader
171// re-stamps it from the matching membership record, which remains the single
172// policy authority.
173func TestRegistryDurableRoundTripObservedIdentity(t *testing.T) {
174 dir := filepath.Join(t.TempDir(), "broker")
175 record := ports.BrokerHostRecord{Registration: registration(t, "host", 1), Pinned: true, Policy: poolPolicy(), Route: canonicalRoute(poolPolicy(), "host")}
176
177 store, err := brokerstore.Open(brokerstore.Options{Dir: dir})
178 require.NoError(t, err)
179 r, err := NewRegistry(1, store, newTestProbe(1), newManualClock(time.Unix(100, 0)), nil)
180 require.NoError(t, err)
181 require.NoError(t, r.ReplaceHosts([]ports.BrokerHostRecord{record}))
182
183 reg := record.Registration
184 ctx, cancel := context.WithCancel(context.Background())
185 defer cancel()
186 r.inflight[reg.Endpoint] = &probeAttempt{token: 1, cancel: cancel}
187 r.apply(probeResult{endpoint: reg.Endpoint, registration: reg, attempt: 1, at: r.clock.Now(), snapshot: ports.BrokerDaemonObservation{
188 Identity: "authed-daemon",
189 Incarnation: ports.BrokerDaemonIncarnation{4, 5},
190 ProtocolVersion: protocol.Version,
191 Capabilities: protocol.CapabilityResume,
192 Availability: domain.RemoteAvailabilityReachable,
193 }})
194 observed, ok := r.Snapshot().Find(reg.Endpoint)
195 require.True(t, ok)
196 require.Equal(t, ports.BrokerDaemonIdentity("authed-daemon"), observed.Identity)
197 require.ErrorIs(t, ctx.Err(), context.Canceled, "applying the result retires the attempt context")
198 // Policy is membership authority, never durable observation state.
199 require.Equal(t, ports.BrokerPolicy{}, durable(r.Snapshot()).Daemons[0].Policy)
200
201 r.settle() // flush before reopening
202 require.NoError(t, store.Close())
203
204 reopened, err := brokerstore.Open(brokerstore.Options{Dir: dir})
205 require.NoError(t, err)
206 defer reopened.Close()
207 persisted, err := reopened.Load()
208 require.NoError(t, err)
209 require.Len(t, persisted.Daemons, 1)
210 require.Equal(t, ports.BrokerPolicy{}, persisted.Daemons[0].Policy, "the durable format omits policy")
211 require.Equal(t, ports.BrokerDaemonIdentity("authed-daemon"), persisted.Daemons[0].Identity)
212 require.Equal(t, ports.BrokerDaemonIncarnation{4, 5}, persisted.Daemons[0].Incarnation)
213 require.Equal(t, protocol.Version, persisted.Daemons[0].ProtocolVersion)
214 require.Equal(t, protocol.CapabilityResume, persisted.Daemons[0].Capabilities)
215 r2, err := NewRegistry(2, reopened, newTestProbe(1), newManualClock(time.Unix(200, 0)), nil)
216 require.NoError(t, err)
217 defer r2.settle()
218
219 restored, ok := r2.Snapshot().Find(reg.Endpoint)
220 require.True(t, ok)
221 require.Equal(t, observed.Identity, restored.Identity)
222 require.Equal(t, observed.Incarnation, restored.Incarnation)
223 require.Equal(t, observed.ProtocolVersion, restored.ProtocolVersion)
224 require.Equal(t, observed.Capabilities, restored.Capabilities)
225 require.Equal(t, domain.RemoteAvailabilityReachable, restored.Availability)
226 require.Equal(t, poolPolicy(), restored.Policy, "the loader re-stamps policy from membership")
227}