registry_membership_durable_test.go RawHistoryFolder
1package broker
2
3import (
4 "context"
5 "errors"
6 "path/filepath"
7 "sync/atomic"
8 "testing"
9 "time"
10
11 "github.com/bnema/vev/internal/adapters/brokerstore"
12 "github.com/bnema/vev/internal/domain"
13 "github.com/bnema/vev/internal/ports"
14 "github.com/stretchr/testify/require"
15)
16
17// Durable mutable membership: the same public API exercised against the real
18// offline store, which owns the compare-and-swap and the snapshot validator.
19// Every stage reopens both owners, so membership and its revision token must
20// come from disk rather than from the previous process's memory.
21
22// newDurableMutableRegistry opens a mutable registry over a real store. The
23// registry is observation-disabled: no probe, timer, or reconciliation runs, so
24// the test observes only the durable membership path.
25func newDurableMutableRegistry(t *testing.T, epoch ports.BrokerEpoch, store ports.BrokerHostStore) *Registry {
26 t.Helper()
27 registry, err := NewRegistryWithConfig(epoch, store, nil, newManualClock(time.Unix(100, 0)), nil, RegistryConfig{
28 MembershipMode: MembershipMutable,
29 ObservationDisabled: true,
30 })
31 require.NoError(t, err)
32 registry.jitter = identityJitter
33 return registry
34}
35
36func openDurableStore(t *testing.T, dir string, fault func(string) error) *brokerstore.Store {
37 t.Helper()
38 store, err := brokerstore.Open(brokerstore.Options{Dir: dir, Fault: fault})
39 require.NoError(t, err)
40 t.Cleanup(func() { require.NoError(t, store.Close()) })
41 return store
42}
43
44// TestRegistryMutableMembershipDurableRestart proves a committed add, update,
45// and remove all survive a restart through the real store, and that the
46// remove/re-add fence is minted fresh by the mutation path rather than restored.
47func TestRegistryMutableMembershipDurableRestart(t *testing.T) {
48 dir := filepath.Join(t.TempDir(), "broker")
49 const endpoint = "user@arch:22"
50 first := poolPolicy()
51 changed := policyWithTrust("changed-trust")
52
53 var (
54 added domain.RemoteRegistration
55 updated domain.RemoteRegistration
56 readded domain.RemoteRegistration
57 revision uint64
58 )
59
60 t.Run("add commits durably", func(t *testing.T) {
61 store := openDurableStore(t, dir, nil)
62 registry := newDurableMutableRegistry(t, 1, store)
63 defer registry.settle()
64 require.Empty(t, registry.authority.Hosts)
65 revision = registry.authority.Revision
66
67 reg, err := registry.AddHost(context.Background(), endpoint, first)
68 require.NoError(t, err)
69 require.Equal(t, domain.RemoteGeneration(1), reg.Generation)
70 require.False(t, reg.IsZero())
71 added = reg
72
73 revision++
74 require.Equal(t, revision, registry.authority.Revision)
75 authority, err := store.LoadHosts()
76 require.NoError(t, err)
77 require.Equal(t, registry.authority, authority)
78 require.Equal(t, []ports.BrokerHostRecord{{Registration: reg, Pinned: true, Policy: first, Route: canonicalRoute(first, reg.Endpoint)}}, authority.Hosts)
79 registry.settle()
80 })
81
82 t.Run("restart restores the add and admits an update", func(t *testing.T) {
83 store := openDurableStore(t, dir, nil)
84 registry := newDurableMutableRegistry(t, 2, store)
85 defer registry.settle()
86 require.Equal(t, revision, registry.authority.Revision)
87 require.Equal(t, []ports.BrokerHostRecord{{Registration: added, Pinned: true, Policy: first, Route: canonicalRoute(first, added.Endpoint)}}, registry.authority.Hosts)
88 host, ok := registry.Snapshot().Find(endpoint)
89 require.True(t, ok)
90 require.Equal(t, added, host.Registration)
91 require.Equal(t, first, host.Policy, "policy is re-stamped from restored membership")
92 require.Equal(t, domain.RemoteAvailabilityUnknown, host.Availability)
93
94 reg, err := registry.UpdateHostPolicy(context.Background(), added, changed)
95 require.NoError(t, err)
96 require.Equal(t, added.Generation+1, reg.Generation)
97 require.Equal(t, added.Incarnation, reg.Incarnation)
98 updated = reg
99
100 revision++
101 require.Equal(t, revision, registry.authority.Revision)
102 authority, err := store.LoadHosts()
103 require.NoError(t, err)
104 require.Equal(t, registry.authority, authority)
105 require.Equal(t, []ports.BrokerHostRecord{{Registration: updated, Pinned: true, Policy: changed, Route: canonicalRoute(changed, updated.Endpoint)}}, authority.Hosts)
106 registry.settle()
107 })
108
109 t.Run("restart restores the update and admits a remove", func(t *testing.T) {
110 store := openDurableStore(t, dir, nil)
111 registry := newDurableMutableRegistry(t, 3, store)
112 defer registry.settle()
113 require.Equal(t, revision, registry.authority.Revision)
114 require.Equal(t, []ports.BrokerHostRecord{{Registration: updated, Pinned: true, Policy: changed, Route: canonicalRoute(changed, updated.Endpoint)}}, registry.authority.Hosts)
115
116 removed, err := registry.RemoveHost(context.Background(), updated)
117 require.NoError(t, err)
118 require.True(t, removed)
119
120 revision++
121 require.Equal(t, revision, registry.authority.Revision)
122 require.Empty(t, registry.authority.Hosts)
123 authority, err := store.LoadHosts()
124 require.NoError(t, err)
125 require.Equal(t, registry.authority, authority)
126 require.Empty(t, authority.Hosts)
127 require.Empty(t, registry.Snapshot().Daemons)
128 registry.settle()
129 })
130
131 t.Run("restart restores the removal and re-adds with a fresh incarnation", func(t *testing.T) {
132 store := openDurableStore(t, dir, nil)
133 registry := newDurableMutableRegistry(t, 4, store)
134 defer registry.settle()
135 require.Equal(t, revision, registry.authority.Revision)
136 require.Empty(t, registry.authority.Hosts)
137
138 reg, err := registry.AddHost(context.Background(), endpoint, changed)
139 require.NoError(t, err)
140 require.False(t, updated.Equal(reg), "a re-add mints a fresh identity instead of reviving the retired one")
141 require.NotEqual(t, updated.Incarnation, reg.Incarnation)
142 require.Equal(t, domain.RemoteGeneration(1), reg.Generation)
143 readded = reg
144
145 revision++
146 require.Equal(t, revision, registry.authority.Revision)
147 authority, err := store.LoadHosts()
148 require.NoError(t, err)
149 require.Equal(t, registry.authority, authority)
150 registry.settle()
151 })
152
153 t.Run("restart restores the re-add", func(t *testing.T) {
154 store := openDurableStore(t, dir, nil)
155 registry := newDurableMutableRegistry(t, 5, store)
156 defer registry.settle()
157 require.Equal(t, revision, registry.authority.Revision)
158 require.Equal(t, []ports.BrokerHostRecord{{Registration: readded, Pinned: true, Policy: changed, Route: canonicalRoute(changed, readded.Endpoint)}}, registry.authority.Hosts)
159 host, ok := registry.Snapshot().Find(endpoint)
160 require.True(t, ok)
161 require.Equal(t, readded, host.Registration)
162 require.False(t, updated.Equal(host.Registration), "the retired incarnation is never restored")
163 })
164}
165
166// TestRegistryReplaceHostsRefusesUnpublishableMembershipBeforeCAS proves the
167// public replacement path shares the runtime mutations' validated commit rule: a
168// caller-supplied target whose derived display hint is empty after the display
169// rule drops its bidi/control runes is refused by projection validation, before
170// the store CAS. Nothing is persisted, nothing is published, and a restart still
171// loads valid membership. The routing validator accepts the target, so this is
172// exactly the membership the store alone would have committed.
173func TestRegistryReplaceHostsRefusesUnpublishableMembershipBeforeCAS(t *testing.T) {
174 dir := filepath.Join(t.TempDir(), "broker")
175 valid := ports.BrokerHostRecord{Registration: registration(t, "user@arch:22", 1), Pinned: true, Policy: poolPolicy()}
176 valid.Route = canonicalRoute(valid.Policy, valid.Registration.Endpoint)
177 store := openDurableStore(t, dir, nil)
178 require.NoError(t, store.ReplaceHosts(1, []ports.BrokerHostRecord{valid}))
179
180 registry, err := NewRegistryWithConfig(1, store, nil, newManualClock(time.Unix(100, 0)), nil, RegistryConfig{ObservationDisabled: true})
181 require.NoError(t, err)
182 defer registry.settle()
183 authority := registry.authority
184 before := registry.Snapshot()
185 revision := registry.revision
186 persistedBefore, err := store.Load()
187 require.NoError(t, err)
188 require.NotZero(t, authority.Revision)
189 require.Len(t, before.Daemons, 1)
190
191 poisoned := ports.BrokerHostRecord{Registration: registration(t, "user@\u202e", 2), Pinned: true, Policy: poolPolicy()}
192 poisoned.Route = canonicalRoute(poisoned.Policy, poisoned.Registration.Endpoint)
193 require.NoError(t, domain.ValidateRemoteHostTarget(poisoned.Registration.Endpoint), "the routing validator accepts the target")
194 require.Empty(t, hostDisplayOrigin(poisoned.Registration.Endpoint), "only the derived hint is empty after sanitizing")
195
196 require.ErrorContains(t, registry.ReplaceHosts([]ports.BrokerHostRecord{poisoned}), "not publishable")
197 require.Equal(t, authority, registry.authority, "a refused replacement never moves authority")
198 require.Equal(t, before, registry.Snapshot(), "a refused replacement never publishes")
199 require.Equal(t, revision, registry.revision, "a refused replacement never advances the publication series")
200
201 // Nothing reached the durable store, even after the writer flushes: the CAS
202 // was never crossed, and the advisory snapshot is still the published one.
203 registry.settle()
204 durableHosts, err := store.LoadHosts()
205 require.NoError(t, err)
206 require.Equal(t, authority, durableHosts)
207 require.Equal(t, authority.Revision, durableHosts.Revision, "the refused replacement never crossed the store CAS")
208 persisted, err := store.Load()
209 require.NoError(t, err)
210 require.Equal(t, persistedBefore, persisted, "the refused replacement never wrote a snapshot")
211
212 // A restart remains valid: the refused membership was never stored, so
213 // reopening restores and publishes the original host.
214 require.NoError(t, store.Close())
215 reopened := openDurableStore(t, dir, nil)
216 restarted, err := NewRegistryWithConfig(2, reopened, nil, newManualClock(time.Unix(200, 0)), nil, RegistryConfig{ObservationDisabled: true})
217 require.NoError(t, err)
218 defer restarted.settle()
219 require.Equal(t, authority, restarted.authority)
220 restored := restarted.Snapshot()
221 require.NoError(t, restored.Validate())
222 require.Len(t, restored.Daemons, 1)
223 require.Equal(t, valid.Registration, restored.Daemons[0].Registration)
224}
225
226// TestRegistryMutableOutcomeUnknownRequiresReopen pins the indeterminate-outcome
227// contract against the real store: a failure after the durable commit point is
228// classified through the registry, is never acknowledged as membership, and
229// leaves the store poisoned until it is reopened and authority is reloaded.
230// Reopening reveals that the mutation did commit, which is exactly why the
231// registry may not report it either way.
232func TestRegistryMutableOutcomeUnknownRequiresReopen(t *testing.T) {
233 dir := filepath.Join(t.TempDir(), "broker")
234 fault := errors.New("power loss")
235 // The fault is armed only after Open so the store can be created and
236 // seeded normally; state.json:rename fires after the rename, so the commit
237 // point was already crossed when the failure is reported.
238 var armed atomic.Bool
239 store := openDurableStore(t, dir, func(point string) error {
240 if armed.Load() && point == "state.json:rename" {
241 return fault
242 }
243 return nil
244 })
245 registry := newDurableMutableRegistry(t, 1, store)
246 defer registry.settle()
247 before := registry.Snapshot()
248 authority := registry.authority
249 armed.Store(true)
250
251 _, err := registry.AddHost(context.Background(), "user@arch:22", poolPolicy())
252 require.Error(t, err)
253 var unknown ports.BrokerStoreOutcomeUnknownError
254 require.ErrorAs(t, err, &unknown, "a post-commit failure is classified as an unknown outcome")
255 require.ErrorIs(t, err, fault, "the underlying store failure stays inspectable")
256 require.Equal(t, authority, registry.authority, "an unknown outcome never becomes acknowledged membership")
257 require.Equal(t, before, registry.Snapshot())
258
259 // The store handle is poisoned: every later operation refuses instead of
260 // reading or writing a state it cannot vouch for.
261 _, err = store.LoadHosts()
262 require.ErrorContains(t, err, "reopen required")
263 require.ErrorAs(t, err, &unknown)
264 require.Empty(t, registry.Snapshot().Daemons)
265
266 // Reopening reloads authority from disk: the rename already published the
267 // new membership, so the only safe recovery is to reopen rather than retry.
268 require.NoError(t, store.Close())
269 armed.Store(false)
270 reopened := openDurableStore(t, dir, nil)
271 hosts, err := reopened.LoadHosts()
272 require.NoError(t, err)
273 require.Equal(t, authority.Revision+1, hosts.Revision)
274 require.Len(t, hosts.Hosts, 1)
275 require.Equal(t, "user@arch:22", hosts.Hosts[0].Registration.Endpoint)
276 require.Equal(t, poolPolicy(), hosts.Hosts[0].Policy)
277
278 recovered := newDurableMutableRegistry(t, 2, reopened)
279 defer recovered.settle()
280 require.Len(t, recovered.authority.Hosts, 1)
281 require.Equal(t, hosts.Hosts[0].Registration, recovered.authority.Hosts[0].Registration)
282}