1package brokerstore
2
3import (
4 "errors"
5 "fmt"
6 "os"
7 "os/exec"
8 "path/filepath"
9 "strings"
10 "sync"
11 "syscall"
12 "testing"
13 "time"
14
15 "github.com/bnema/vev/internal/domain"
16 "github.com/bnema/vev/internal/ports"
17 "github.com/bnema/vev/internal/protocol"
18 "github.com/bnema/vev/internal/protocol/catalogue"
19 "github.com/stretchr/testify/require"
20)
21
22func testPolicy() ports.BrokerPolicy {
23 return ports.BrokerPolicy{ProtocolVersion: 1, CatalogSchemaVersion: 1, EnvironmentPolicy: protocol.EnvironmentPolicyDaemonOwned, Transport: "quic", Trust: "known-hosts", Launch: "explicit", Isolation: "user"}
24}
25
26// observation builds a remote daemon observation that satisfies the durable
27// shape: an exact registration, a derived display origin, and a reachable
28// availability. Policy is intentionally left absent: the loader stamps it from
29// membership, never from the durable snapshot.
30func observation(endpoint string, registration domain.RemoteRegistration) ports.BrokerDaemonObservation {
31 return ports.BrokerDaemonObservation{
32 Endpoint: endpoint,
33 DisplayOrigin: domain.RemoteDisplayOrigin(endpoint),
34 Registration: registration,
35 Availability: domain.RemoteAvailabilityReachable,
36 }
37}
38
39func testHostRecord() ports.BrokerHostRecord {
40 reg := domain.RemoteRegistration{Endpoint: "user@arch", Incarnation: [16]byte{1}, Generation: 7}
41 return ports.BrokerHostRecord{Registration: reg, Learned: true, Pinned: true, Policy: testPolicy(), Route: ports.BrokerRouteSpec{Kind: ports.BrokerRouteSSHQUIC, Target: reg.Endpoint, Argv: []string{"vev", "_broker-mux-quic-bootstrap", "--production"}}}
42}
43
44func options(t *testing.T) Options {
45 t.Helper()
46 return Options{Dir: filepath.Join(t.TempDir(), "broker"), InitialHosts: []ports.BrokerHostRecord{testHostRecord()}, InitialImportProvided: true}
47}
48func TestMigrationFaultRestart(t *testing.T) {
49 for _, file := range []string{"recovery.json", "state.json"} {
50 for _, point := range []string{"write", "sync", "rename", "dirsync"} {
51 t.Run(file+point, func(t *testing.T) {
52 o := options(t)
53 fault := errors.New("power loss")
54 o.Fault = func(p string) error {
55 if p == file+":"+point {
56 return fault
57 }
58 return nil
59 }
60 s, err := Open(o)
61 require.ErrorIs(t, err, fault)
62 require.Nil(t, s)
63 var saved recovery
64 raw, e := readBounded(filepath.Join(o.Dir, "recovery.json"))
65 if e == nil {
66 require.NoError(t, strict(raw, &saved))
67 }
68 o.Fault = nil
69 s, err = Open(o)
70 require.NoError(t, err)
71 defer s.Close()
72 hosts, err := s.LoadHosts()
73 require.NoError(t, err)
74 if e == nil {
75 require.Equal(t, saved.State.Hosts, hosts)
76 }
77 })
78 }
79 }
80}
81func TestLockLifetime(t *testing.T) {
82 if dir := os.Getenv("VEV_BROKERSTORE_LOCK_TEST"); dir != "" {
83 s, err := Open(Options{Dir: dir})
84 if s != nil {
85 s.Close()
86 }
87 if !errors.Is(err, ErrLocked) {
88 os.Exit(2)
89 }
90 return
91 }
92 o := options(t)
93 s, err := Open(o)
94 require.NoError(t, err)
95 // The lifetime lock carries the legacy host writer's exact file name, so a
96 // store opened beside a live hosts.json excludes that writer too.
97 _, err = os.Stat(filepath.Join(o.Dir, "hosts.json.lock"))
98 require.NoError(t, err)
99 cmd := exec.Command(os.Args[0], "-test.run=^TestLockLifetime$")
100 cmd.Env = append(os.Environ(), "VEV_BROKERSTORE_LOCK_TEST="+o.Dir)
101 require.NoError(t, cmd.Run())
102 // Assertions never run inside the competing goroutines: t.FailNow from a
103 // non-test goroutine is invalid, so each worker records its outcome and the
104 // test body checks them after the join.
105 var wg sync.WaitGroup
106 outcomes := make([]error, 8)
107 for i := 0; i < len(outcomes); i++ {
108 wg.Add(1)
109 go func(i int) {
110 defer wg.Done()
111 other, e := Open(o)
112 if other != nil {
113 _ = other.Close()
114 outcomes[i] = fmt.Errorf("concurrent Open returned a store: %w", ports.ErrBrokerStoreLocked)
115 return
116 }
117 if !errors.Is(e, ErrLocked) || !errors.Is(e, ports.ErrBrokerStoreLocked) {
118 outcomes[i] = fmt.Errorf("concurrent Open error = %v, want ErrLocked", e)
119 }
120 }(i)
121 }
122 wg.Wait()
123 for i, outcome := range outcomes {
124 require.NoError(t, outcome, "worker %d", i)
125 }
126 require.NoError(t, s.Close())
127 _, err = s.Load()
128 require.Error(t, err)
129 s, err = Open(o)
130 require.NoError(t, err)
131 require.NoError(t, s.Close())
132}
133
134// TestLockExcludesLegacyHostWriter holds the lock the way the legacy host
135// writer does and proves the store refuses to own the same directory.
136func TestLockExcludesLegacyHostWriter(t *testing.T) {
137 o := options(t)
138 s, err := Open(o)
139 require.NoError(t, err)
140 require.NoError(t, s.Close())
141
142 legacyLock, err := os.OpenFile(filepath.Join(o.Dir, "hosts.json.lock"), os.O_CREATE|os.O_RDWR, 0600)
143 require.NoError(t, err)
144 defer legacyLock.Close()
145 require.NoError(t, syscall.Flock(int(legacyLock.Fd()), syscall.LOCK_EX|syscall.LOCK_NB))
146
147 _, err = Open(o)
148 require.ErrorIs(t, err, ErrLocked)
149 require.ErrorIs(t, err, ports.ErrBrokerStoreLocked)
150
151 require.NoError(t, syscall.Flock(int(legacyLock.Fd()), syscall.LOCK_UN))
152 s, err = Open(o)
153 require.NoError(t, err)
154 require.NoError(t, s.Close())
155}
156func TestStrictCorruption(t *testing.T) {
157 for _, raw := range []string{`{"version":1,"version":1}`, `{"unknown":1}`, `{} {`, string([]byte{255}), strings.Repeat(" ", MaxFileBytes+1)} {
158 var v state
159 require.Error(t, strict([]byte(raw), &v))
160 }
161 o := options(t)
162 s, err := Open(o)
163 require.NoError(t, err)
164 require.NoError(t, s.Close())
165 require.NoError(t, os.WriteFile(filepath.Join(o.Dir, "state.json"), []byte(`{`), 0600))
166 _, err = Open(o)
167 require.Error(t, err)
168}
169func TestDurableStateSentinelMapping(t *testing.T) {
170 o := options(t)
171 s, err := Open(o)
172 require.NoError(t, err)
173
174 // A conflicting authority token is a host conflict, not a generic failure.
175 require.ErrorIs(t, s.ReplaceHosts(0, nil), ErrStale)
176 require.ErrorIs(t, s.ReplaceHosts(0, nil), ports.ErrBrokerHostConflict)
177
178 // A locked, corrupt, or closed store is classified without importing the
179 // adapter: callers retry, resolve state offline, or reopen respectively.
180 require.ErrorIs(t, ErrLocked, ports.ErrBrokerStoreLocked)
181 require.ErrorIs(t, ErrInvalidState, ports.ErrBrokerStoreInvalidState)
182
183 require.NoError(t, s.Close())
184 // Closing releases ownership without discarding the durable files.
185 require.Error(t, s.ReplaceHosts(1, nil))
186 require.NoError(t, os.WriteFile(filepath.Join(o.Dir, "state.json"), []byte(`{`), 0600))
187 _, err = Open(o)
188 require.ErrorIs(t, err, ErrInvalidState)
189 require.ErrorIs(t, err, ports.ErrBrokerStoreInvalidState)
190}
191
192// TestCommittedStateAndRecoveryAreNeverFollowed replaces the committed state and
193// the recovery record with symlinks to valid bytes: the store reads both with
194// O_NOFOLLOW, so a symlink fails closed instead of redirecting a rename or
195// silently adopting attacker-chosen state.
196func TestCommittedStateAndRecoveryAreNeverFollowed(t *testing.T) {
197 t.Run("state", func(t *testing.T) {
198 o := options(t)
199 s, err := Open(o)
200 require.NoError(t, err)
201 require.NoError(t, s.Close())
202
203 state := filepath.Join(o.Dir, "state.json")
204 raw, err := os.ReadFile(state)
205 require.NoError(t, err)
206 target := filepath.Join(o.Dir, "elsewhere.json")
207 require.NoError(t, os.WriteFile(target, raw, 0600))
208 require.NoError(t, os.Remove(state))
209 require.NoError(t, os.Symlink(target, state))
210
211 _, err = Open(o)
212 require.ErrorIs(t, err, ErrInvalidState)
213 require.ErrorIs(t, err, ports.ErrBrokerStoreInvalidState)
214 after, err := os.ReadFile(target)
215 require.NoError(t, err)
216 require.Equal(t, raw, after)
217 })
218 t.Run("recovery", func(t *testing.T) {
219 o := options(t)
220 s, err := Open(o)
221 require.NoError(t, err)
222 require.NoError(t, s.Close())
223
224 recoveryPath := filepath.Join(o.Dir, "recovery.json")
225 raw, err := os.ReadFile(recoveryPath)
226 require.NoError(t, err)
227 target := filepath.Join(o.Dir, "elsewhere.json")
228 require.NoError(t, os.WriteFile(target, raw, 0600))
229 require.NoError(t, os.Remove(recoveryPath))
230 require.NoError(t, os.Remove(filepath.Join(o.Dir, "state.json")))
231 require.NoError(t, os.Symlink(target, recoveryPath))
232
233 _, err = Open(o)
234 require.ErrorIs(t, err, ErrInvalidState)
235 require.ErrorIs(t, err, ports.ErrBrokerStoreInvalidState)
236 _, err = os.Stat(filepath.Join(o.Dir, "state.json"))
237 require.True(t, os.IsNotExist(err))
238 })
239}
240
241func TestRevisionOverflowIsRefused(t *testing.T) {
242 o := options(t)
243 s, err := Open(o)
244 require.NoError(t, err)
245 defer s.Close()
246 h, err := s.LoadHosts()
247 require.NoError(t, err)
248 for _, hosts := range [][]ports.BrokerHostRecord{h.Hosts, nil} {
249 require.ErrorIs(t, s.ReplaceHosts(^uint64(0), hosts), ErrStale)
250 }
251 after, err := s.LoadHosts()
252 require.NoError(t, err)
253 require.Equal(t, h, after)
254}
255
256// TestAuthorityGenerationPolicyAndRemoveReAdd exercises the durable authority
257// rules: policy changes need a fresh generation, generations never regress, a
258// removed incarnation's observation never returns, and a re-add carries a new
259// incarnation.
260func TestAuthorityGenerationPolicyAndRemoveReAdd(t *testing.T) {
261 o := options(t)
262 s, err := Open(o)
263 require.NoError(t, err)
264 defer s.Close()
265 h, err := s.LoadHosts()
266 require.NoError(t, err)
267 require.Len(t, h.Hosts, 1)
268 base := h.Hosts[0]
269
270 // A policy change with an unchanged registration identity is refused: an
271 // already queued observation must never be accepted under changed trust.
272 changed := append([]ports.BrokerHostRecord(nil), h.Hosts...)
273 changed[0].Policy.Trust = "changed"
274 require.ErrorIs(t, s.ReplaceHosts(h.Revision, changed), ErrStale)
275
276 // Re-supplying identical membership is allowed and still advances the
277 // authority revision.
278 require.NoError(t, s.ReplaceHosts(h.Revision, h.Hosts))
279 h2, err := s.LoadHosts()
280 require.NoError(t, err)
281 require.Equal(t, h.Revision+1, h2.Revision)
282
283 // The same policy change with a bumped generation is accepted and durable.
284 changed[0].Registration.Generation++
285 require.NoError(t, s.ReplaceHosts(h2.Revision, changed))
286 h3, err := s.LoadHosts()
287 require.NoError(t, err)
288 require.Equal(t, "changed", h3.Hosts[0].Policy.Trust)
289
290 // A regressed generation on the same incarnation is refused.
291 regressed := append([]ports.BrokerHostRecord(nil), h3.Hosts...)
292 regressed[0].Registration.Generation = base.Registration.Generation
293 require.ErrorIs(t, s.ReplaceHosts(h3.Revision, regressed), ErrStale)
294
295 // Full removal drops the observation with the registration.
296 require.NoError(t, s.ReplaceHosts(h3.Revision, nil))
297 removed, err := s.LoadHosts()
298 require.NoError(t, err)
299 require.Empty(t, removed.Hosts)
300
301 // A queued publication from the retired incarnation is refused even though
302 // its revision is newer.
303 require.ErrorIs(t, s.Store(ports.BrokerSnapshot{Epoch: 9, Revision: 1, Daemons: []ports.BrokerDaemonObservation{observation(base.Registration.Endpoint, base.Registration)}}), ErrStale)
304
305 // Re-adding the endpoint needs a fresh incarnation; the retired
306 // observation never comes back for it.
307 var incarnation [16]byte
308 incarnation[0] = 9
309 registration, err := domain.NewRemoteRegistration(base.Registration.Endpoint, incarnation)
310 require.NoError(t, err)
311 reAdd := append([]ports.BrokerHostRecord(nil), base)
312 reAdd[0].Registration = registration
313 require.NoError(t, s.ReplaceHosts(removed.Revision, reAdd))
314 readded, err := s.LoadHosts()
315 require.NoError(t, err)
316 require.True(t, readded.Hosts[0].Registration.Equal(registration))
317 snapshot, err := s.Load()
318 require.NoError(t, err)
319 _, ok := snapshot.Find(base.Registration.Endpoint)
320 require.False(t, ok, "a removed incarnation's observation must not survive a re-add")
321}
322
323// TestReplaceHostsRejectsCallerInputBeforeCAS proves malformed caller
324// membership is rejected as a caller error before the compare-and-swap, never
325// as ErrBrokerStoreInvalidState: the durable state is a victim, not the cause.
326func TestReplaceHostsRejectsCallerInputBeforeCAS(t *testing.T) {
327 o := options(t)
328 s, err := Open(o)
329 require.NoError(t, err)
330 defer s.Close()
331 h, err := s.LoadHosts()
332 require.NoError(t, err)
333 require.Len(t, h.Hosts, 1)
334 base := h.Hosts[0]
335
336 invalid := map[string][]ports.BrokerHostRecord{
337 "zero registration": {{Pinned: true, Policy: base.Policy}},
338 "zero policy": {{Registration: base.Registration, Pinned: true}},
339 "unanchored": {{Registration: base.Registration, Policy: base.Policy}},
340 "duplicate endpoint": {
341 {Registration: base.Registration, Pinned: true, Policy: base.Policy},
342 {Registration: base.Registration, Learned: true, Policy: base.Policy},
343 },
344 }
345 for name, records := range invalid {
346 t.Run(name, func(t *testing.T) {
347 // The CAS token matches, so a CAS-first store would have accepted the
348 // malformed set and failed later as invalid durable state.
349 err := s.ReplaceHosts(h.Revision, records)
350 require.Error(t, err)
351 require.NotErrorIs(t, err, ErrStale)
352 require.NotErrorIs(t, err, ErrInvalidState)
353 require.NotErrorIs(t, err, ports.ErrBrokerStoreInvalidState)
354 })
355 }
356 t.Run("over bound", func(t *testing.T) {
357 records := make([]ports.BrokerHostRecord, 0, ports.BrokerMaxHosts+1)
358 for index := 0; index <= ports.BrokerMaxHosts; index++ {
359 registration, err := domain.NewRemoteRegistration(fmt.Sprintf("bound-%d.test", index), [16]byte{byte(index + 1)})
360 require.NoError(t, err)
361 records = append(records, ports.BrokerHostRecord{Registration: registration, Pinned: true, Policy: base.Policy})
362 }
363 err := s.ReplaceHosts(h.Revision, records)
364 require.Error(t, err)
365 require.NotErrorIs(t, err, ports.ErrBrokerStoreInvalidState)
366 })
367
368 // Rejected input never commits: the authority revision and membership are
369 // exactly as before.
370 after, err := s.LoadHosts()
371 require.NoError(t, err)
372 require.Equal(t, h, after)
373}
374
375// TestDurableSnapshotDropsProcessLocalTombstones proves a snapshot's
376// process-local tombstone set is sanitized before persistence: storing a
377// snapshot whose tombstone names a live host is accepted and reopened without
378// Removed, instead of being rejected as durable-state corruption.
379func TestDurableSnapshotDropsProcessLocalTombstones(t *testing.T) {
380 o := options(t)
381 s, err := Open(o)
382 require.NoError(t, err)
383 h, err := s.LoadHosts()
384 require.NoError(t, err)
385 require.Len(t, h.Hosts, 1)
386 reg := h.Hosts[0].Registration
387
388 snap, err := s.Load()
389 require.NoError(t, err)
390 snap.Epoch = 5
391 snap.Revision = 7
392 snap.Removed = []ports.BrokerHostTombstone{{Endpoint: reg.Endpoint, Registration: reg, RetiredRevision: 6}}
393 require.NoError(t, s.Store(snap))
394
395 stored, err := s.Load()
396 require.NoError(t, err)
397 require.Empty(t, stored.Removed)
398 require.NoError(t, s.Close())
399
400 s, err = Open(o)
401 require.NoError(t, err)
402 defer s.Close()
403 reopened, err := s.Load()
404 require.NoError(t, err)
405 require.Empty(t, reopened.Removed)
406}
407
408// TestStoreZeroesTabAttentionOnPersist pins C3: a tab's Attention (bell) is
409// transient observation state, not durable identity, so it is zeroed before
410// every durable write. Without this, a bell recorded the instant before a
411// broker restart would resurrect on reload even though nothing is still
412// ringing. This ports the intent of main's
413// internal/adapters/remote/catalog_cache_test.go:71, which pins the same rule
414// for the legacy remote monitor's cache (there, dynamic tab fields including
415// Attention never reach the stored JSON at all).
416func TestStoreZeroesTabAttentionOnPersist(t *testing.T) {
417 tests := []struct {
418 name string
419 tabs []catalogue.RemoteCatalogTab
420 }{
421 {name: "single tab with attention", tabs: []catalogue.RemoteCatalogTab{
422 {ID: "t1", Index: 0, Attention: true},
423 }},
424 {name: "mixed attention across tabs", tabs: []catalogue.RemoteCatalogTab{
425 {ID: "t1", Index: 0, Attention: true},
426 {ID: "t2", Index: 1},
427 {ID: "t3", Index: 2, Attention: true},
428 }},
429 {name: "no attention stays false", tabs: []catalogue.RemoteCatalogTab{
430 {ID: "t1", Index: 0},
431 {ID: "t2", Index: 1},
432 }},
433 }
434 for _, tt := range tests {
435 t.Run(tt.name, func(t *testing.T) {
436 o := options(t)
437 s, err := Open(o)
438 require.NoError(t, err)
439 h, err := s.LoadHosts()
440 require.NoError(t, err)
441 require.Len(t, h.Hosts, 1)
442 reg := h.Hosts[0].Registration
443
444 snap, err := s.Load()
445 require.NoError(t, err)
446 snap.Epoch = 9
447 snap.Revision = 1
448 daemon := observation(reg.Endpoint, reg)
449 daemon.InventoryKnown = true
450 daemon.LastSuccess = time.Unix(1, 0)
451 daemon.Sessions = []catalogue.RemoteCatalogSession{{
452 LifecycleID: domain.SessionLifecycleID{1},
453 Name: "work",
454 State: catalogue.RemoteCatalogSessionUp,
455 Tabs: tt.tabs,
456 }}
457 snap.Daemons = []ports.BrokerDaemonObservation{daemon}
458 require.NoError(t, s.Store(snap))
459
460 stored, err := s.Load()
461 require.NoError(t, err)
462 require.Len(t, stored.Daemons, 1)
463 require.Len(t, stored.Daemons[0].Sessions, 1)
464 require.Len(t, stored.Daemons[0].Sessions[0].Tabs, len(tt.tabs))
465 for _, tab := range stored.Daemons[0].Sessions[0].Tabs {
466 require.False(t, tab.Attention, "tab %q must never persist a live bell", tab.ID)
467 }
468 require.NoError(t, s.Close())
469
470 // The zeroing is durable, not just an in-memory view: a fresh Open
471 // reading the same file back must never resurrect the bell either.
472 s, err = Open(o)
473 require.NoError(t, err)
474 defer s.Close()
475 reopened, err := s.Load()
476 require.NoError(t, err)
477 require.Len(t, reopened.Daemons, 1)
478 require.Len(t, reopened.Daemons[0].Sessions, 1)
479 for _, tab := range reopened.Daemons[0].Sessions[0].Tabs {
480 require.False(t, tab.Attention)
481 }
482 })
483 }
484}
485
486func TestFencingAndCommitFaults(t *testing.T) {
487 for _, point := range []string{"write", "sync", "rename", "dirsync"} {
488 t.Run(point, func(t *testing.T) {
489 o := options(t)
490 s, err := Open(o)
491 require.NoError(t, err)
492 h, err := s.LoadHosts()
493 require.NoError(t, err)
494 snap, err := s.Load()
495 require.NoError(t, err)
496 snap.Epoch = 20
497 snap.Revision = 1
498 require.NoError(t, s.Store(snap))
499 require.ErrorIs(t, s.Store(snap), ErrStale)
500 other := snap
501 other.Epoch = 21
502 require.ErrorIs(t, s.Store(other), ErrStale)
503 require.ErrorIs(t, s.ReplaceHosts(0, nil), ErrStale)
504 changed := append([]ports.BrokerHostRecord(nil), h.Hosts...)
505 changed[0].Policy.Trust = "new"
506 require.ErrorIs(t, s.ReplaceHosts(h.Revision, changed), ErrStale)
507 s.fault = func(p string) error {
508 if p == "state.json:"+point {
509 return errors.New("disk fault")
510 }
511 return nil
512 }
513 require.Error(t, s.ReplaceHosts(h.Revision, nil))
514 _, err = s.LoadHosts()
515 require.Error(t, err)
516 require.NoError(t, s.Close())
517 s, err = Open(o)
518 require.NoError(t, err)
519 defer s.Close()
520 h, err = s.LoadHosts()
521 require.NoError(t, err)
522 if point == "rename" || point == "dirsync" {
523 require.Empty(t, h.Hosts)
524 snap.Revision++
525 require.NoError(t, s.Store(snap))
526 } else {
527 require.Len(t, h.Hosts, 1)
528 }
529 })
530 }
531 t.Run("generation and restart", func(t *testing.T) {
532 o := options(t)
533 s, err := Open(o)
534 require.NoError(t, err)
535 h, err := s.LoadHosts()
536 require.NoError(t, err)
537 snap, err := s.Load()
538 require.NoError(t, err)
539 h.Hosts[0].Registration.Generation++
540 require.NoError(t, s.ReplaceHosts(h.Revision, h.Hosts))
541 require.ErrorIs(t, s.Store(snap), ErrStale)
542 require.NoError(t, s.Close())
543 s, err = Open(o)
544 require.NoError(t, err)
545 defer s.Close()
546 got, err := s.Load()
547 require.NoError(t, err)
548 require.Empty(t, got.Daemons)
549 gotH, err := s.LoadHosts()
550 require.NoError(t, err)
551 require.Equal(t, h.Hosts, gotH.Hosts)
552 })
553}
554
555// TestLocalDaemonObservationIsNeverDurable pins the durable boundary: a local
556// daemon observation is process-local state and the store refuses it outright,
557// so it can never become durable authority.
558func TestLocalDaemonObservationIsNeverDurable(t *testing.T) {
559 s, err := Open(options(t))
560 require.NoError(t, err)
561 defer s.Close()
562 err = s.Store(ports.BrokerSnapshot{Epoch: 9, Revision: 1, Daemons: []ports.BrokerDaemonObservation{{
563 Local: true,
564 DisplayOrigin: "local",
565 Policy: testPolicy(),
566 Availability: domain.RemoteAvailabilityReachable,
567 }}})
568 require.Error(t, err)
569 require.Contains(t, err.Error(), "never durable")
570}