1package broker
2
3import (
4 "context"
5 "errors"
6 "fmt"
7 "sync"
8 "testing"
9 "time"
10
11 "github.com/bnema/vev/internal/domain"
12 "github.com/bnema/vev/internal/ports"
13 portsmocks "github.com/bnema/vev/internal/ports/mocks"
14 "github.com/bnema/vev/internal/protocol"
15 "github.com/stretchr/testify/require"
16)
17
18// Phase A mutable membership: the registry is the only synchronous owner of
19// durable authority, so every admitted mutation must CAS the store, install its
20// projection only after that CAS succeeds, and leave authority, projection,
21// probe attempts, and tombstones untouched on every refusal. The tests below
22// drive the public mutation API and the production fencing path directly, so a
23// regression in any of those rules is observable instead of being masked by the
24// observation scheduler.
25
26// incarnationSequencer is a deterministic entropy seam: every call yields a
27// distinct, never-zero incarnation, so a test can assert that a remove/re-add
28// really minted a fresh identity and that a refused mutation minted none.
29type incarnationSequencer struct {
30 mu sync.Mutex
31 next byte
32}
33
34func (s *incarnationSequencer) generate() ([16]byte, error) {
35 s.mu.Lock()
36 defer s.mu.Unlock()
37 s.next++
38 var id [16]byte
39 id[0] = 0xA5
40 id[15] = s.next
41 return id, nil
42}
43
44func (s *incarnationSequencer) generated() int {
45 s.mu.Lock()
46 defer s.mu.Unlock()
47 return int(s.next)
48}
49
50// membershipStore is a race-safe durable store with an observable CAS counter
51// and an injectable CAS failure. The counter proves a mutation reached the
52// store (including the no-op paths, which must still CAS), and the injected
53// failure proves a store fault is synchronous and never installs a projection.
54type membershipStore struct {
55 *testStore
56 casMu sync.Mutex
57 casErr error
58 casCall int
59}
60
61func newMembershipStore(hosts ...ports.BrokerHostRecord) *membershipStore {
62 return &membershipStore{testStore: &testStore{hosts: ports.BrokerHosts{Revision: 1, Hosts: hosts}}}
63}
64
65func (s *membershipStore) ReplaceHosts(expected uint64, hosts []ports.BrokerHostRecord) error {
66 s.casMu.Lock()
67 s.casCall++
68 err := s.casErr
69 s.casMu.Unlock()
70 if err != nil {
71 return err
72 }
73 return s.testStore.ReplaceHosts(expected, hosts)
74}
75
76func (s *membershipStore) failCAS(err error) {
77 s.casMu.Lock()
78 s.casErr = err
79 s.casMu.Unlock()
80}
81
82func (s *membershipStore) casCalls() int {
83 s.casMu.Lock()
84 defer s.casMu.Unlock()
85 return s.casCall
86}
87
88// durableAuthority reads the store's committed authority, exactly as a reopen
89// would: membership and its revision token come from the store, never from the
90// registry's in-memory copy.
91func (s *membershipStore) durableAuthority() ports.BrokerHosts {
92 s.testStore.mu.Lock()
93 defer s.testStore.mu.Unlock()
94 return ports.BrokerHosts{Revision: s.testStore.hosts.Revision, Hosts: append([]ports.BrokerHostRecord(nil), s.testStore.hosts.Hosts...)}
95}
96
97// newMutableRegistry builds a mutable registry with exact scheduling, a
98// deterministic incarnation seam, and its scripted probe.
99func newMutableRegistry(t *testing.T, store ports.BrokerHostStore, clock *manualClock, generator func() ([16]byte, error)) (*Registry, *testProbe) {
100 t.Helper()
101 probe := newTestProbe(2)
102 registry, err := NewRegistryWithConfig(1, store, probe, clock, nil, RegistryConfig{
103 MembershipMode: MembershipMutable,
104 IncarnationGenerator: generator,
105 ObservationDisabled: false,
106 })
107 require.NoError(t, err)
108 registry.jitter = identityJitter
109 return registry, probe
110}
111
112func policyWithTrust(trust string) ports.BrokerPolicy {
113 policy := poolPolicy()
114 policy.Trust = trust
115 return policy
116}
117
118func pinnedRecord(reg domain.RemoteRegistration, policy ports.BrokerPolicy) ports.BrokerHostRecord {
119 return ports.BrokerHostRecord{Registration: reg, Pinned: true, Policy: policy, Route: canonicalRoute(policy, reg.Endpoint)}
120}
121
122// admittedAttempt is one in-flight probe attempt a test installs directly, so a
123// completion can be delivered to apply without racing the scheduler.
124type admittedAttempt struct {
125 token uint64
126 ctx context.Context
127 cancel context.CancelFunc
128}
129
130func admitMembershipAttempt(t *testing.T, r *Registry, endpoint string) admittedAttempt {
131 t.Helper()
132 ctx, cancel := context.WithCancel(context.Background())
133 t.Cleanup(cancel)
134 r.mu.Lock()
135 defer r.mu.Unlock()
136 r.attempts++
137 attempt := admittedAttempt{token: r.attempts, ctx: ctx, cancel: cancel}
138 r.inflight[endpoint] = &probeAttempt{token: attempt.token, cancel: cancel}
139 return attempt
140}
141
142// membershipFences snapshots the tombstone set so a failed mutation can be
143// proven to have retired nothing.
144func membershipFences(r *Registry) map[string]ports.BrokerHostTombstone {
145 r.mu.Lock()
146 defer r.mu.Unlock()
147 out := make(map[string]ports.BrokerHostTombstone, len(r.tombstones))
148 for endpoint, tombstone := range r.tombstones {
149 out[endpoint] = tombstone
150 }
151 return out
152}
153
154func staleObservation() ports.BrokerDaemonObservation {
155 return ports.BrokerDaemonObservation{
156 Identity: "stale-daemon",
157 Incarnation: ports.BrokerDaemonIncarnation{7},
158 ProtocolVersion: protocol.Version,
159 Availability: domain.RemoteAvailabilityReachable,
160 InventoryKnown: true,
161 }
162}
163
164func currentObservation() ports.BrokerDaemonObservation {
165 return ports.BrokerDaemonObservation{
166 Identity: "current-daemon",
167 Incarnation: ports.BrokerDaemonIncarnation{9},
168 ProtocolVersion: protocol.Version,
169 Availability: domain.RemoteAvailabilityReachable,
170 InventoryKnown: true,
171 }
172}
173
174// TestRegistryMutableConcurrentAddsLoseNoUpdate proves concurrent membership
175// mutations serialize over durable authority: every added endpoint survives,
176// each addition advances the revision by exactly one, and the registry's copy
177// still equals what the store committed.
178func TestRegistryMutableConcurrentAddsLoseNoUpdate(t *testing.T) {
179 const adds = 32
180 store := newMembershipStore()
181 registry, _ := newMutableRegistry(t, store, newManualClock(time.Unix(100, 0)), (&incarnationSequencer{}).generate)
182 start := registry.authority.Revision
183
184 endpoints := make([]string, adds)
185 for i := range endpoints {
186 endpoints[i] = fmt.Sprintf("host-%d.test", i)
187 }
188
189 var wg sync.WaitGroup
190 errs := make([]error, adds)
191 regs := make([]domain.RemoteRegistration, adds)
192 startGate := make(chan struct{})
193 for i, endpoint := range endpoints {
194 wg.Add(1)
195 go func(i int, endpoint string) {
196 defer wg.Done()
197 <-startGate
198 regs[i], errs[i] = registry.AddHost(context.Background(), endpoint, poolPolicy())
199 }(i, endpoint)
200 }
201 close(startGate)
202 wg.Wait()
203
204 for i, err := range errs {
205 require.NoErrorf(t, err, "concurrent add %q", endpoints[i])
206 }
207 require.Equal(t, adds, store.casCalls(), "every admitted add must reach the store CAS")
208 require.Equal(t, start+adds, registry.authority.Revision, "each add advances the revision exactly once")
209 require.Len(t, registry.authority.Hosts, adds)
210 require.Equal(t, registry.authority, store.durableAuthority())
211
212 // Admission order is the lock order, not the launch order, so the
213 // publication order is compared against the returned identities by endpoint.
214 returned := make(map[string]domain.RemoteRegistration, adds)
215 for i, endpoint := range endpoints {
216 require.False(t, regs[i].IsZero(), "an admitted add returns its minted identity")
217 require.Equal(t, endpoint, regs[i].Endpoint)
218 require.Equal(t, domain.RemoteGeneration(1), regs[i].Generation)
219 returned[endpoint] = regs[i]
220 }
221 require.Len(t, returned, adds, "every concurrent add minted a distinct identity")
222
223 seen := make(map[string]domain.RemoteRegistration, adds)
224 for _, record := range registry.authority.Hosts {
225 require.NotContains(t, seen, record.Registration.Endpoint, "no endpoint may be added twice")
226 require.Equal(t, returned[record.Registration.Endpoint], record.Registration)
227 seen[record.Registration.Endpoint] = record.Registration
228 require.Equal(t, poolPolicy(), record.Policy)
229 require.True(t, record.Pinned)
230 }
231 require.Len(t, seen, adds)
232
233 snapshot := registry.Snapshot()
234 require.Len(t, snapshot.Daemons, adds)
235 require.NoError(t, snapshot.Validate())
236 for _, endpoint := range endpoints {
237 host, ok := snapshot.Find(endpoint)
238 require.True(t, ok, "a concurrently added endpoint must be published")
239 require.Equal(t, seen[endpoint], host.Registration)
240 require.Equal(t, domain.RemoteAvailabilityUnknown, host.Availability)
241 }
242}
243
244// TestRegistryMembershipModeRejectsUnknownMode proves the run mode is a closed
245// set: an out-of-range mode is refused at construction, before any store call,
246// so a caller can never run a registry whose mutation policy is undefined.
247func TestRegistryMembershipModeRejectsUnknownMode(t *testing.T) {
248 // The mock has no expectations: any store call during construction fails
249 // the test, so a refusal must happen before the durable store is touched.
250 store := portsmocks.NewMockBrokerHostStore(t)
251 registry, err := NewRegistryWithConfig(1, store, newTestProbe(1), newManualClock(time.Unix(100, 0)), nil,
252 RegistryConfig{MembershipMode: MembershipMode(9)})
253 require.ErrorContains(t, err, "invalid membership mode")
254 require.Nil(t, registry)
255}
256
257// TestRegistryImmutableMembershipRefusesEveryMutation pins the default: a
258// registry that was not explicitly configured mutable refuses every membership
259// mutation, including the ones that would be no-ops, and refuses them without
260// touching the durable store or its own projection.
261func TestRegistryImmutableMembershipRefusesEveryMutation(t *testing.T) {
262 reg := registration(t, "alpha.test", 1)
263 store := portsmocks.NewMockBrokerHostStore(t)
264 store.EXPECT().Load().Return(ports.BrokerSnapshot{}, nil).Once()
265 store.EXPECT().LoadHosts().Return(ports.BrokerHosts{Revision: 3, Hosts: []ports.BrokerHostRecord{pinnedRecord(reg, poolPolicy())}}, nil).Once()
266 registry, err := NewRegistry(1, store, newTestProbe(1), newManualClock(time.Unix(100, 0)), nil)
267 require.NoError(t, err)
268 authority := registry.authority
269 before := registry.Snapshot()
270
271 tests := []struct {
272 name string
273 run func(r *Registry) error
274 }{
275 {"add new endpoint", func(r *Registry) error {
276 _, err := r.AddHost(context.Background(), "beta.test", poolPolicy())
277 return err
278 }},
279 {"add duplicate endpoint with equal policy is a no-op", func(r *Registry) error {
280 _, err := r.AddHost(context.Background(), reg.Endpoint, poolPolicy())
281 return err
282 }},
283 {"remove the exact registration", func(r *Registry) error {
284 _, err := r.RemoveHost(context.Background(), reg)
285 return err
286 }},
287 {"remove an absent endpoint is a no-op", func(r *Registry) error {
288 _, err := r.RemoveHost(context.Background(), registration(t, "gamma.test", 3))
289 return err
290 }},
291 {"update to a different policy", func(r *Registry) error {
292 _, err := r.UpdateHostPolicy(context.Background(), reg, policyWithTrust("other-trust"))
293 return err
294 }},
295 {"update to the equal policy is a no-op", func(r *Registry) error {
296 _, err := r.UpdateHostPolicy(context.Background(), reg, poolPolicy())
297 return err
298 }},
299 }
300 for _, tc := range tests {
301 t.Run(tc.name, func(t *testing.T) {
302 require.ErrorIs(t, tc.run(registry), ports.ErrBrokerMembershipImmutable)
303 require.Equal(t, authority, registry.authority)
304 require.Equal(t, before, registry.Snapshot())
305 require.Empty(t, membershipFences(registry))
306 })
307 }
308
309 // Control: the same duplicate add is admitted when the mode is mutable, so
310 // the refusals above are the mode and not the validity of the operation.
311 seq := &incarnationSequencer{}
312 mutable, _ := newMutableRegistry(t, newMembershipStore(pinnedRecord(reg, poolPolicy())), newManualClock(time.Unix(100, 0)), seq.generate)
313 added, err := mutable.AddHost(context.Background(), reg.Endpoint, poolPolicy())
314 require.NoError(t, err)
315 require.Equal(t, reg, added)
316 require.Equal(t, 0, seq.generated(), "a duplicate add must not mint an identity")
317}
318
319// TestRegistryMutableAddPinsDuplicatePreservingIdentity proves a duplicate
320// same-policy addition pins the existing record without changing its identity,
321// and that even that no-op CAS-verifies durable authority.
322func TestRegistryMutableAddPinsDuplicatePreservingIdentity(t *testing.T) {
323 seq := &incarnationSequencer{}
324 store := newMembershipStore(ports.BrokerHostRecord{Registration: registration(t, "alpha.test", 1), Learned: true, Policy: poolPolicy(), Route: canonicalRoute(poolPolicy(), "alpha.test")})
325 registry, _ := newMutableRegistry(t, store, newManualClock(time.Unix(100, 0)), seq.generate)
326 reg := registration(t, "alpha.test", 1)
327
328 got, err := registry.AddHost(context.Background(), reg.Endpoint, poolPolicy())
329 require.NoError(t, err)
330 require.Equal(t, reg, got, "a duplicate add preserves the registration identity")
331 require.Equal(t, 1, store.casCalls())
332 require.Len(t, registry.authority.Hosts, 1)
333 record := registry.authority.Hosts[0]
334 require.Equal(t, reg, record.Registration)
335 require.Equal(t, poolPolicy(), record.Policy)
336 require.True(t, record.Pinned, "the duplicate add pins the learned registration")
337 require.True(t, record.Learned, "pinning never discards learned provenance")
338 require.Equal(t, uint64(2), registry.authority.Revision)
339 require.Equal(t, registry.authority, store.durableAuthority())
340 require.Equal(t, 0, seq.generated(), "pinning mints no identity")
341
342 host, ok := registry.Snapshot().Find(reg.Endpoint)
343 require.True(t, ok)
344 require.Equal(t, reg, host.Registration)
345 require.Equal(t, domain.RemoteAvailabilityUnknown, host.Availability)
346
347 // A second duplicate is a no-op that still CAS-verifies: authority moves
348 // durably, the projection does not republish, and identity is unchanged.
349 before := registry.Snapshot()
350 again, err := registry.AddHost(context.Background(), reg.Endpoint, poolPolicy())
351 require.NoError(t, err)
352 require.Equal(t, reg, again)
353 require.Equal(t, 2, store.casCalls())
354 require.Equal(t, uint64(3), registry.authority.Revision)
355 require.Equal(t, registry.authority, store.durableAuthority())
356 require.Equal(t, before, registry.Snapshot())
357}
358
359// TestRegistryMutableAddConflictingPolicyConflicts proves a conflicting policy
360// is a definite conflict: it is refused before the store CAS, and it mints no
361// identity.
362func TestRegistryMutableAddConflictingPolicyConflicts(t *testing.T) {
363 seq := &incarnationSequencer{}
364 reg := registration(t, "alpha.test", 1)
365 store := newMembershipStore(pinnedRecord(reg, poolPolicy()))
366 registry, _ := newMutableRegistry(t, store, newManualClock(time.Unix(100, 0)), seq.generate)
367 authority := registry.authority
368 before := registry.Snapshot()
369
370 _, err := registry.AddHost(context.Background(), reg.Endpoint, policyWithTrust("other-trust"))
371 require.ErrorIs(t, err, ports.ErrBrokerHostConflict)
372 require.Equal(t, 0, store.casCalls(), "a conflict is refused before the CAS")
373 require.Equal(t, authority, registry.authority)
374 require.Equal(t, before, registry.Snapshot())
375 require.Equal(t, 0, seq.generated())
376}
377
378// TestRegistryMutableAddEntropyFailureLeavesAuthorityUnchanged pins the entropy
379// seam: an unusable incarnation is refused before anything durable is staged.
380func TestRegistryMutableAddEntropyFailureLeavesAuthorityUnchanged(t *testing.T) {
381 tests := []struct {
382 name string
383 generator func() ([16]byte, error)
384 want string
385 }{
386 {"entropy source fails", func() ([16]byte, error) { return [16]byte{}, errors.New("no entropy") }, "no entropy"},
387 {"entropy source returns a zero identity", func() ([16]byte, error) { return [16]byte{}, nil }, "zero"},
388 }
389 for _, tc := range tests {
390 t.Run(tc.name, func(t *testing.T) {
391 store := newMembershipStore()
392 registry, _ := newMutableRegistry(t, store, newManualClock(time.Unix(100, 0)), tc.generator)
393 authority := registry.authority
394 before := registry.Snapshot()
395
396 _, err := registry.AddHost(context.Background(), "alpha.test", poolPolicy())
397 require.ErrorContains(t, err, tc.want)
398 require.Equal(t, 0, store.casCalls())
399 require.Equal(t, authority, registry.authority)
400 require.Equal(t, before, registry.Snapshot())
401 })
402 }
403}
404
405// TestRegistryMutableRemoveExactStaleAndAbsent pins removal admission: only the
406// exact expected registration is removed, a stale generation or incarnation is
407// a definite conflict, and an absent endpoint reports false only after the no-op
408// CAS verified the loaded revision.
409func TestRegistryMutableRemoveExactStaleAndAbsent(t *testing.T) {
410 alpha := registration(t, "alpha.test", 1)
411 beta := registration(t, "beta.test", 2)
412 store := newMembershipStore(pinnedRecord(alpha, poolPolicy()), pinnedRecord(beta, poolPolicy()))
413 registry, _ := newMutableRegistry(t, store, newManualClock(time.Unix(100, 0)), (&incarnationSequencer{}).generate)
414 authority := registry.authority
415 before := registry.Snapshot()
416
417 staleGeneration := alpha
418 staleGeneration.Generation++
419 staleIncarnation := alpha
420 staleIncarnation.Incarnation = [16]byte{0xEE}
421 for name, expected := range map[string]domain.RemoteRegistration{
422 "stale generation": staleGeneration,
423 "stale incarnation": staleIncarnation,
424 } {
425 t.Run(name, func(t *testing.T) {
426 removed, err := registry.RemoveHost(context.Background(), expected)
427 require.ErrorIs(t, err, ports.ErrBrokerHostConflict)
428 require.False(t, removed)
429 require.Equal(t, 0, store.casCalls(), "a mismatch is refused before the CAS")
430 require.Equal(t, authority, registry.authority)
431 require.Equal(t, before, registry.Snapshot())
432 require.Empty(t, membershipFences(registry))
433 })
434 }
435
436 removed, err := registry.RemoveHost(context.Background(), alpha)
437 require.NoError(t, err)
438 require.True(t, removed)
439 require.Equal(t, 1, store.casCalls())
440 require.Len(t, registry.authority.Hosts, 1)
441 require.Equal(t, beta, registry.authority.Hosts[0].Registration)
442 require.Equal(t, uint64(2), registry.authority.Revision)
443 require.Equal(t, registry.authority, store.durableAuthority())
444 afterRemove := registry.Snapshot()
445 require.Len(t, afterRemove.Daemons, 1)
446 require.Equal(t, beta, afterRemove.Daemons[0].Registration)
447 require.Len(t, afterRemove.Removed, 1)
448 require.Equal(t, alpha, afterRemove.Removed[0].Registration)
449 require.Equal(t, afterRemove.Revision, afterRemove.Removed[0].RetiredRevision,
450 "a tombstone is retired at the revision of the publication that carries it")
451
452 // An absent endpoint is a no-op that still CAS-verifies: durable authority
453 // advances, while nothing is republished and nothing is retired.
454 absent, err := registry.RemoveHost(context.Background(), alpha)
455 require.NoError(t, err)
456 require.False(t, absent)
457 require.Equal(t, 2, store.casCalls())
458 require.Equal(t, uint64(3), registry.authority.Revision)
459 require.Equal(t, registry.authority, store.durableAuthority())
460 require.Equal(t, afterRemove, registry.Snapshot())
461}
462
463// TestRegistryMutableNoOpRemoveVerifiesCAS pins the no-op CAS itself: with
464// authority already moved by another mutator, an absent-endpoint removal must
465// report a conflict instead of a hollow success, because the loaded revision was
466// never verified.
467func TestRegistryMutableNoOpRemoveVerifiesCAS(t *testing.T) {
468 alpha := registration(t, "alpha.test", 1)
469 store := newMembershipStore(pinnedRecord(alpha, poolPolicy()))
470 registry, _ := newMutableRegistry(t, store, newManualClock(time.Unix(100, 0)), (&incarnationSequencer{}).generate)
471 authority := registry.authority
472 before := registry.Snapshot()
473
474 // An illicit second mutator moves durable authority behind the registry.
475 require.NoError(t, store.testStore.ReplaceHosts(authority.Revision, []ports.BrokerHostRecord{pinnedRecord(alpha, poolPolicy())}))
476
477 removed, err := registry.RemoveHost(context.Background(), registration(t, "gamma.test", 3))
478 require.ErrorIs(t, err, ports.ErrBrokerHostConflict)
479 require.False(t, removed)
480 require.Equal(t, 1, store.casCalls())
481 require.Equal(t, authority, registry.authority)
482 require.Equal(t, before, registry.Snapshot())
483}
484
485// TestRegistryMutableRemoveReAddGetsFreshIncarnation pins the ABA fence at the
486// mutation boundary: a full removal followed by a re-add mints a fresh
487// incarnation with a fresh generation series, and the retired registration is
488// fenced rather than revived.
489func TestRegistryMutableRemoveReAddGetsFreshIncarnation(t *testing.T) {
490 seq := &incarnationSequencer{}
491 store := newMembershipStore()
492 registry, _ := newMutableRegistry(t, store, newManualClock(time.Unix(100, 0)), seq.generate)
493
494 first, err := registry.AddHost(context.Background(), "alpha.test", poolPolicy())
495 require.NoError(t, err)
496 require.Equal(t, 1, seq.generated())
497
498 removed, err := registry.RemoveHost(context.Background(), first)
499 require.NoError(t, err)
500 require.True(t, removed)
501
502 second, err := registry.AddHost(context.Background(), "alpha.test", poolPolicy())
503 require.NoError(t, err)
504 require.Equal(t, 2, seq.generated())
505 require.False(t, first.Equal(second), "a re-add must not reuse the retired registration")
506 require.NotEqual(t, first.Incarnation, second.Incarnation)
507 require.Equal(t, domain.RemoteGeneration(1), second.Generation)
508 require.Len(t, registry.authority.Hosts, 1)
509 require.Equal(t, second, registry.authority.Hosts[0].Registration)
510 require.Equal(t, registry.authority, store.durableAuthority())
511
512 snapshot := registry.Snapshot()
513 require.Len(t, snapshot.Daemons, 1)
514 require.Equal(t, second, snapshot.Daemons[0].Registration)
515 require.Empty(t, snapshot.Removed, "a live endpoint is never published as a tombstone")
516 host, ok := snapshot.Find(second.Endpoint)
517 require.True(t, ok)
518 require.Equal(t, domain.RemoteAvailabilityUnknown, host.Availability)
519}
520
521// TestRegistryMutableUpdatePolicyIncrementsGenerationOnce pins the policy
522// update: a changed policy advances the registration generation exactly once and
523// is the only thing that does; an unmatchable expectation is a definite conflict
524// refused before the CAS.
525func TestRegistryMutableUpdatePolicyIncrementsGenerationOnce(t *testing.T) {
526 alpha := registration(t, "alpha.test", 1)
527 store := newMembershipStore(pinnedRecord(alpha, poolPolicy()))
528 registry, _ := newMutableRegistry(t, store, newManualClock(time.Unix(100, 0)), (&incarnationSequencer{}).generate)
529
530 // Not found and a stale expectation are definite conflicts, refused before
531 // the store is touched.
532 authority := registry.authority
533 before := registry.Snapshot()
534 _, err := registry.UpdateHostPolicy(context.Background(), registration(t, "gamma.test", 3), policyWithTrust("other-trust"))
535 require.ErrorIs(t, err, ports.ErrBrokerHostConflict)
536 stale := alpha
537 stale.Generation++
538 _, err = registry.UpdateHostPolicy(context.Background(), stale, policyWithTrust("other-trust"))
539 require.ErrorIs(t, err, ports.ErrBrokerHostConflict)
540 require.Equal(t, 0, store.casCalls())
541 require.Equal(t, authority, registry.authority)
542 require.Equal(t, before, registry.Snapshot())
543
544 updated, err := registry.UpdateHostPolicy(context.Background(), alpha, policyWithTrust("other-trust"))
545 require.NoError(t, err)
546 require.Equal(t, alpha.Generation+1, updated.Generation, "a policy change advances the generation exactly once")
547 require.Equal(t, alpha.Incarnation, updated.Incarnation, "a policy change never touches the incarnation")
548 require.Equal(t, 1, store.casCalls())
549 require.Equal(t, uint64(2), registry.authority.Revision)
550 require.Equal(t, registry.authority, store.durableAuthority())
551 require.Equal(t, []ports.BrokerHostRecord{pinnedRecord(updated, policyWithTrust("other-trust"))}, store.durableAuthority().Hosts)
552
553 host, ok := registry.Snapshot().Find(alpha.Endpoint)
554 require.True(t, ok)
555 require.Equal(t, updated, host.Registration)
556 require.Equal(t, policyWithTrust("other-trust"), host.Policy)
557 require.Equal(t, domain.RemoteAvailabilityUnknown, host.Availability)
558}
559
560// TestRegistryMutableUpdateEqualPolicyPreservesIdentityAndVerifiesCAS proves an
561// equal-policy update is an identity-preserving no-op that still CAS-verifies
562// durable authority.
563func TestRegistryMutableUpdateEqualPolicyPreservesIdentityAndVerifiesCAS(t *testing.T) {
564 alpha := registration(t, "alpha.test", 1)
565 store := newMembershipStore(pinnedRecord(alpha, poolPolicy()))
566 registry, _ := newMutableRegistry(t, store, newManualClock(time.Unix(100, 0)), (&incarnationSequencer{}).generate)
567 before := registry.Snapshot()
568
569 got, err := registry.UpdateHostPolicy(context.Background(), alpha, poolPolicy())
570 require.NoError(t, err)
571 require.Equal(t, alpha, got, "an equal policy preserves the registration identity")
572 require.Equal(t, 1, store.casCalls())
573 require.Equal(t, uint64(2), registry.authority.Revision)
574 require.Equal(t, registry.authority, store.durableAuthority())
575 require.Equal(t, before, registry.Snapshot(), "an unchanged policy republishes nothing")
576
577 // The no-op still verified the loaded revision: a revision moved behind the
578 // registry turns the same call into a definite conflict.
579 require.NoError(t, store.testStore.ReplaceHosts(registry.authority.Revision, []ports.BrokerHostRecord{pinnedRecord(alpha, poolPolicy())}))
580 authority := registry.authority
581 _, err = registry.UpdateHostPolicy(context.Background(), alpha, poolPolicy())
582 require.ErrorIs(t, err, ports.ErrBrokerHostConflict)
583 require.Equal(t, 2, store.casCalls())
584 require.Equal(t, authority, registry.authority)
585 require.Equal(t, before, registry.Snapshot())
586}
587
588// TestRegistryMutableUpdatePolicyOverflowRefused pins the generation series: a
589// policy change at the last generation fails closed instead of wrapping to zero,
590// which would re-trust an already queued observation.
591func TestRegistryMutableUpdatePolicyOverflowRefused(t *testing.T) {
592 alpha := registration(t, "alpha.test", 1)
593 alpha.Generation = ^domain.RemoteGeneration(0)
594 store := newMembershipStore(pinnedRecord(alpha, poolPolicy()))
595 registry, _ := newMutableRegistry(t, store, newManualClock(time.Unix(100, 0)), (&incarnationSequencer{}).generate)
596 authority := registry.authority
597 before := registry.Snapshot()
598
599 _, err := registry.UpdateHostPolicy(context.Background(), alpha, policyWithTrust("other-trust"))
600 require.ErrorIs(t, err, ports.ErrBrokerRevisionExhausted)
601 require.Equal(t, 0, store.casCalls())
602 require.Equal(t, authority, registry.authority)
603 require.Equal(t, before, registry.Snapshot())
604}
605
606// TestRegistryMutableProbeCompletionFencedByChangedGeneration proves a
607// completion whose registration was replaced by a policy update can never
608// populate the new generation, whether it carries the retired registration or
609// the current one with a retired attempt token.
610func TestRegistryMutableProbeCompletionFencedByChangedGeneration(t *testing.T) {
611 clock := newManualClock(time.Unix(100, 0))
612 store := newMembershipStore()
613 registry, _ := newMutableRegistry(t, store, clock, (&incarnationSequencer{}).generate)
614 regA, err := registry.AddHost(context.Background(), "alpha.test", poolPolicy())
615 require.NoError(t, err)
616 retired := admitMembershipAttempt(t, registry, regA.Endpoint)
617 regB, err := registry.UpdateHostPolicy(context.Background(), regA, policyWithTrust("other-trust"))
618 require.NoError(t, err)
619 require.Equal(t, regA.Generation+1, regB.Generation)
620 require.ErrorIs(t, retired.ctx.Err(), context.Canceled, "replacing a registration retires its attempt")
621 require.Zero(t, attemptFor(registry, regA.Endpoint))
622
623 before := registry.Snapshot()
624 for _, tc := range []struct {
625 name string
626 result probeResult
627 }{
628 {"retired registration", probeResult{endpoint: regA.Endpoint, registration: regA, attempt: retired.token, at: clock.Now(), snapshot: staleObservation()}},
629 {"current registration with a retired token", probeResult{endpoint: regA.Endpoint, registration: regB, attempt: retired.token, at: clock.Now(), snapshot: staleObservation()}},
630 {"retired registration with the current token", probeResult{endpoint: regA.Endpoint, registration: regA, attempt: retired.token + 1, at: clock.Now(), snapshot: staleObservation()}},
631 } {
632 t.Run(tc.name, func(t *testing.T) {
633 registry.apply(tc.result)
634 require.Equal(t, before, registry.Snapshot())
635 host, ok := registry.Snapshot().Find(regA.Endpoint)
636 require.True(t, ok)
637 require.Equal(t, regB, host.Registration)
638 require.Equal(t, domain.RemoteAvailabilityUnknown, host.Availability)
639 require.Empty(t, host.Identity)
640 require.True(t, host.LastSuccess.IsZero())
641 require.Zero(t, attemptFor(registry, regA.Endpoint), "a stale completion never admits an attempt")
642 })
643 }
644}
645
646// TestRegistryMutableStaleCompletionDoesNotClearLiveAttempt proves the fence on
647// the live side: a completion for the retired generation is dropped while a
648// current attempt is in flight, so it neither clears the in-flight slot nor
649// publishes under the retired registration.
650func TestRegistryMutableStaleCompletionDoesNotClearLiveAttempt(t *testing.T) {
651 clock := newManualClock(time.Unix(100, 0))
652 registry, _ := newMutableRegistry(t, newMembershipStore(), clock, (&incarnationSequencer{}).generate)
653 regA, err := registry.AddHost(context.Background(), "alpha.test", poolPolicy())
654 require.NoError(t, err)
655 retired := admitMembershipAttempt(t, registry, regA.Endpoint)
656 regB, err := registry.UpdateHostPolicy(context.Background(), regA, policyWithTrust("other-trust"))
657 require.NoError(t, err)
658
659 live := admitMembershipAttempt(t, registry, regB.Endpoint)
660 before := registry.Snapshot()
661 for _, result := range []probeResult{
662 {endpoint: regB.Endpoint, registration: regA, attempt: retired.token, at: clock.Now(), snapshot: staleObservation()},
663 {endpoint: regB.Endpoint, registration: regB, attempt: retired.token, at: clock.Now(), snapshot: staleObservation()},
664 {endpoint: regB.Endpoint, registration: regA, attempt: live.token, at: clock.Now(), snapshot: staleObservation()},
665 } {
666 registry.apply(result)
667 }
668 require.Equal(t, live.token, attemptFor(registry, regB.Endpoint), "a stale completion must not clear the live attempt")
669 require.NoError(t, live.ctx.Err(), "a stale completion must not retire the live attempt")
670 require.Equal(t, before, registry.Snapshot())
671
672 // The live attempt still completes normally and publishes its own result.
673 registry.apply(probeResult{endpoint: regB.Endpoint, registration: regB, attempt: live.token, at: clock.Now(), snapshot: currentObservation()})
674 require.ErrorIs(t, live.ctx.Err(), context.Canceled)
675 host, ok := registry.Snapshot().Find(regB.Endpoint)
676 require.True(t, ok)
677 require.Equal(t, regB, host.Registration)
678 require.Equal(t, ports.BrokerDaemonIdentity("current-daemon"), host.Identity)
679 require.Equal(t, domain.RemoteAvailabilityReachable, host.Availability)
680}
681
682// TestRegistryMutableStaleProbeCannotReviveRemovedRegistration proves the
683// remove/re-add fence: a completion for the retired incarnation is dropped after
684// the endpoint was re-added under a fresh incarnation, and the re-add supersedes
685// rather than revives the retirement.
686func TestRegistryMutableStaleProbeCannotReviveRemovedRegistration(t *testing.T) {
687 clock := newManualClock(time.Unix(100, 0))
688 seq := &incarnationSequencer{}
689 registry, _ := newMutableRegistry(t, newMembershipStore(), clock, seq.generate)
690 regA, err := registry.AddHost(context.Background(), "alpha.test", poolPolicy())
691 require.NoError(t, err)
692 retired := admitMembershipAttempt(t, registry, regA.Endpoint)
693
694 removed, err := registry.RemoveHost(context.Background(), regA)
695 require.NoError(t, err)
696 require.True(t, removed)
697 require.ErrorIs(t, retired.ctx.Err(), context.Canceled)
698 afterRemove := registry.Snapshot()
699 require.Len(t, afterRemove.Removed, 1)
700 require.Equal(t, regA, afterRemove.Removed[0].Registration)
701
702 regB, err := registry.AddHost(context.Background(), "alpha.test", poolPolicy())
703 require.NoError(t, err)
704 require.NotEqual(t, regA.Incarnation, regB.Incarnation)
705 before := registry.Snapshot()
706 require.Len(t, before.Daemons, 1)
707 require.Empty(t, before.Removed)
708
709 registry.apply(probeResult{endpoint: regA.Endpoint, registration: regA, attempt: retired.token, at: clock.Now(), snapshot: staleObservation()})
710 require.Equal(t, before, registry.Snapshot())
711 host, ok := registry.Snapshot().Find(regA.Endpoint)
712 require.True(t, ok)
713 require.Equal(t, regB, host.Registration)
714 require.Equal(t, domain.RemoteAvailabilityUnknown, host.Availability)
715 require.Empty(t, host.Identity)
716 require.Zero(t, attemptFor(registry, regA.Endpoint))
717}
718
719// TestRegistryMutablePolicyChangeRetiresInflightProbe drives the same fence
720// through the production run loop: a policy change retires the in-flight probe
721// promptly, a fresh attempt observes the new registration, and the retired
722// answer can never surface as the new generation's observation.
723func TestRegistryMutablePolicyChangeRetiresInflightProbe(t *testing.T) {
724 registry, probe := newMutableRegistry(t, newMembershipStore(), newManualClock(time.Unix(100, 0)), (&incarnationSequencer{}).generate)
725 regA, err := registry.AddHost(context.Background(), "alpha.test", poolPolicy())
726 require.NoError(t, err)
727 startRegistry(t, registry)
728
729 first := receiveCall(t, probe)
730 require.Equal(t, regA, first.registration)
731
732 regB, err := registry.UpdateHostPolicy(context.Background(), regA, policyWithTrust("other-trust"))
733 require.NoError(t, err)
734 require.Equal(t, regA.Generation+1, regB.Generation)
735 select {
736 case <-probe.canceled:
737 case <-time.After(time.Second):
738 t.Fatal("a policy change must retire the in-flight probe")
739 }
740
741 second := receiveCall(t, probe)
742 require.Equal(t, regB, second.registration, "the replacement generation is observed")
743
744 // The retired answer arrives after its attempt was retired; only the new
745 // generation's answer may become the published observation.
746 first.result <- probeAnswer{snapshot: staleObservation()}
747 second.result <- probeAnswer{snapshot: currentObservation()}
748 host := waitHost(t, registry, regB.Endpoint, func(host ports.BrokerDaemonObservation) bool {
749 return host.Identity == ports.BrokerDaemonIdentity("current-daemon")
750 })
751 require.Equal(t, regB, host.Registration)
752 require.Equal(t, domain.RemoteAvailabilityReachable, host.Availability)
753 requireHostStableWithin(t, registry, regB.Endpoint, host, 50*time.Millisecond)
754}
755
756// membershipFixture seeds the state a failed mutation must leave untouched: two
757// pinned hosts, one live probe attempt, and one retirement tombstone.
758type membershipFixture struct {
759 registry *Registry
760 store *membershipStore
761 alpha domain.RemoteRegistration
762 attempt admittedAttempt
763 authority ports.BrokerHosts
764 before ports.BrokerSnapshot
765 fences map[string]ports.BrokerHostTombstone
766 casBefore int
767}
768
769func newMembershipFixture(t *testing.T) *membershipFixture {
770 t.Helper()
771 alpha := registration(t, "alpha.test", 1)
772 fixture := &membershipFixture{alpha: alpha}
773 fixture.store = newMembershipStore(pinnedRecord(alpha, poolPolicy()), pinnedRecord(registration(t, "beta.test", 2), poolPolicy()))
774 registry, _ := newMutableRegistry(t, fixture.store, newManualClock(time.Unix(100, 0)), (&incarnationSequencer{}).generate)
775 fixture.registry = registry
776 fixture.attempt = admitMembershipAttempt(t, registry, alpha.Endpoint)
777
778 // Retire beta so the fixture carries a live tombstone alongside the live
779 // attempt at the moment the failure is injected.
780 removed, err := registry.RemoveHost(context.Background(), registration(t, "beta.test", 2))
781 require.NoError(t, err)
782 require.True(t, removed)
783
784 fixture.authority = registry.authority
785 fixture.before = registry.Snapshot()
786 fixture.fences = membershipFences(registry)
787 fixture.casBefore = fixture.store.casCalls()
788 require.Len(t, fixture.fences, 1)
789 return fixture
790}
791
792// TestRegistryMutableStoreFailureLeavesStateUnchanged pins the commit ordering:
793// a definite or indeterminate store failure returns before any projection,
794// authority, attempt, or tombstone changes, and the store was genuinely reached.
795func TestRegistryMutableStoreFailureLeavesStateUnchanged(t *testing.T) {
796 definite := []struct {
797 name string
798 err error
799 }{
800 {"conflict", ports.ErrBrokerHostConflict},
801 {"plain failure", context.DeadlineExceeded},
802 }
803 indeterminate := errors.New("power loss")
804
805 for _, failure := range definite {
806 t.Run(failure.name, func(t *testing.T) {
807 runMembershipStoreFailures(t, failure.err)
808 })
809 }
810 t.Run("indeterminate outcome", func(t *testing.T) {
811 runMembershipStoreFailures(t, ports.BrokerStoreOutcomeUnknownError{Err: indeterminate})
812 })
813}
814
815func runMembershipStoreFailures(t *testing.T, storeErr error) {
816 t.Helper()
817 ops := []struct {
818 name string
819 run func(fixture *membershipFixture) (bool, error)
820 }{
821 {"add new", func(f *membershipFixture) (bool, error) {
822 _, err := f.registry.AddHost(context.Background(), "gamma.test", poolPolicy())
823 return false, err
824 }},
825 {"add duplicate", func(f *membershipFixture) (bool, error) {
826 _, err := f.registry.AddHost(context.Background(), f.alpha.Endpoint, poolPolicy())
827 return false, err
828 }},
829 {"remove exact", func(f *membershipFixture) (bool, error) {
830 return f.registry.RemoveHost(context.Background(), f.alpha)
831 }},
832 {"remove absent", func(f *membershipFixture) (bool, error) {
833 return f.registry.RemoveHost(context.Background(), registration(t, "delta.test", 4))
834 }},
835 {"update policy", func(f *membershipFixture) (bool, error) {
836 _, err := f.registry.UpdateHostPolicy(context.Background(), f.alpha, policyWithTrust("other-trust"))
837 return false, err
838 }},
839 {"update equal policy", func(f *membershipFixture) (bool, error) {
840 _, err := f.registry.UpdateHostPolicy(context.Background(), f.alpha, poolPolicy())
841 return false, err
842 }},
843 }
844 for _, op := range ops {
845 t.Run(op.name, func(t *testing.T) {
846 fixture := newMembershipFixture(t)
847 fixture.store.failCAS(storeErr)
848
849 removed, err := op.run(fixture)
850 require.Error(t, err)
851 // A commit that never happened is never reported as a removal, even
852 // for the one operation whose success is a boolean: a caller that
853 // trusted removed=true from a failed commit would drop live authority.
854 require.False(t, removed, "a failed commit must never report a removal")
855 var unknown ports.BrokerStoreOutcomeUnknownError
856 if errors.As(storeErr, &unknown) {
857 require.ErrorAs(t, err, &unknown, "an indeterminate outcome stays classifiable through the registry")
858 } else {
859 require.ErrorIs(t, err, storeErr)
860 require.False(t, errors.As(err, &unknown), "a definite failure is not an unknown outcome")
861 }
862
863 require.Equal(t, fixture.authority, fixture.registry.authority, "authority must not move")
864 require.Equal(t, fixture.before, fixture.registry.Snapshot(), "projection must not move")
865 require.Equal(t, fixture.attempt.token, attemptFor(fixture.registry, fixture.alpha.Endpoint), "the in-flight attempt must survive")
866 require.NoError(t, fixture.attempt.ctx.Err(), "a failed store call must not retire the attempt")
867 require.Equal(t, fixture.fences, membershipFences(fixture.registry), "tombstones must not move")
868 require.Equal(t, fixture.casBefore+1, fixture.store.casCalls(), "the mutation reached the store CAS")
869 })
870 }
871}