1package brokerstore
2
3import (
4 "bytes"
5 "encoding/json"
6 "errors"
7 "os"
8 "path/filepath"
9 "testing"
10
11 "github.com/bnema/vev/internal/domain"
12 "github.com/bnema/vev/internal/ports"
13 "github.com/stretchr/testify/require"
14)
15
16// canonicalRoutes are the exact routes production derives for the closed remote
17// policy transport vocabulary the fixtures use.
18func canonicalQUICRoute(endpoint string) ports.BrokerRouteSpec {
19 return ports.BrokerRouteSpec{Kind: ports.BrokerRouteSSHQUIC, Target: endpoint, Argv: []string{"vev", "_broker-mux-quic-bootstrap", "--production"}}
20}
21
22// TestInitialImportEmptyCompletesAndNeverReimports pins the explicit initial
23// import seam: when the caller supplies an empty initial membership, the import
24// is recorded as completed with no records, and a later reopen never falls back
25// to re-importing the legacy source that is still named in Options.
26func TestInitialImportEmptyCompletesAndNeverReimports(t *testing.T) {
27 dir := filepath.Join(t.TempDir(), "broker")
28 o := options(t)
29 o.Dir = dir
30 o.InitialHosts = nil
31 o.InitialImportProvided = true
32
33 store, err := Open(o)
34 require.NoError(t, err)
35 hosts, err := store.LoadHosts()
36 require.NoError(t, err)
37 require.Empty(t, hosts.Hosts, "an empty initial import seeds no membership")
38
39 raw, err := readBounded(filepath.Join(dir, "recovery.json"))
40 require.NoError(t, err)
41 var record recovery
42 require.NoError(t, strict(raw, &record))
43 require.Equal(t, 1, record.State.Manifest.ImportVersion)
44 require.True(t, record.State.Manifest.ImportCompleted, "the initial import is completed even when empty")
45
46 // Remove-all CAS-verifies the empty authority and still commits.
47 require.NoError(t, store.ReplaceHosts(hosts.Revision, nil))
48 removed, err := store.LoadHosts()
49 require.NoError(t, err)
50 require.Empty(t, removed.Hosts)
51 require.Equal(t, hosts.Revision+1, removed.Revision)
52 require.NoError(t, store.Close())
53
54 // Reopening names the legacy source again, but state.json exists, so no
55 // migration runs: the empty, completed import stays authoritative.
56 store, err = Open(o)
57 require.NoError(t, err)
58 defer store.Close()
59 again, err := store.LoadHosts()
60 require.NoError(t, err)
61 require.Empty(t, again.Hosts, "a supplied legacy source is never re-imported after the import completed")
62 require.Equal(t, removed.Revision, again.Revision)
63}
64
65// TestRecoveryRetainsRoutes proves the retained recovery record carries the
66// canonical routes as durable authority: reopening from recovery alone (state.json
67// removed) restores the same routed membership.
68func TestRecoveryRetainsRoutes(t *testing.T) {
69 o := options(t)
70 store, err := Open(o)
71 require.NoError(t, err)
72 require.NoError(t, store.Close())
73
74 raw, err := readBounded(filepath.Join(o.Dir, "recovery.json"))
75 require.NoError(t, err)
76 var record recovery
77 require.NoError(t, strict(raw, &record))
78 require.Len(t, record.State.Hosts.Hosts, 1)
79 require.Equal(t, canonicalQUICRoute("user@arch"), record.State.Hosts.Hosts[0].Route)
80
81 // Drop the committed state so the recovery record is the only source.
82 require.NoError(t, os.Remove(filepath.Join(o.Dir, "state.json")))
83 o.InitialHosts = nil
84 reopened, err := Open(o)
85 require.NoError(t, err)
86 defer reopened.Close()
87 hosts, err := reopened.LoadHosts()
88 require.NoError(t, err)
89 require.Len(t, hosts.Hosts, 1)
90 require.Equal(t, canonicalQUICRoute("user@arch"), hosts.Hosts[0].Route)
91}
92
93// TestCorruptRouteStateFailsClosed proves a committed state whose route is
94// invalid is never silently replaced or repaired: Open reports invalid durable
95// state and leaves the exact bytes it read in place for offline recovery.
96func TestCorruptRouteStateFailsClosed(t *testing.T) {
97 o := options(t)
98 store, err := Open(o)
99 require.NoError(t, err)
100 require.NoError(t, store.Close())
101
102 statePath := filepath.Join(o.Dir, "state.json")
103 raw, err := os.ReadFile(statePath)
104 require.NoError(t, err)
105 poisoned := bytes.Replace(raw, []byte(`"ssh-quic"`), []byte(`"bogus"`), 1)
106 require.NotEqual(t, raw, poisoned, "the fixture must carry a route kind to corrupt")
107 require.NoError(t, os.WriteFile(statePath, poisoned, 0600))
108
109 reopened, err := Open(o)
110 require.Error(t, err)
111 require.Nil(t, reopened)
112 require.ErrorIs(t, err, ErrInvalidState)
113 require.ErrorIs(t, err, ports.ErrBrokerStoreInvalidState)
114
115 after, err := os.ReadFile(statePath)
116 require.NoError(t, err)
117 require.Equal(t, poisoned, after, "invalid durable state is never silently replaced")
118}
119
120// TestLegacyRemoteUnixAndUnknownTransportRefused proves the migration route
121// upgrade is closed over the approved remote vocabulary: a legacy policy naming
122// a unix or unknown transport is refused before any state is committed, so a
123// legacy local route can never become broker membership.
124// TestRouteUpgradeIsAtomicAcrossReopen proves the pre-route upgrade is a durable
125// migration: a state written without routes is upgraded exactly once, a fault at
126// any upgrade boundary never leaves an unreadable state behind, and reopening
127// completes the upgrade so the restored membership carries canonical routes.
128func TestRouteUpgradeIsAtomicAcrossReopen(t *testing.T) {
129 // seedStripped opens and closes a valid routed store, then rewrites its
130 // committed state without routes, exactly as a pre-route process would have
131 // written it. Each case gets its own directory so a completed upgrade in one
132 // case never hides the migration in the next.
133 seedStripped := func(t *testing.T) string {
134 t.Helper()
135 dir := filepath.Join(t.TempDir(), "broker")
136 o := options(t)
137 o.Dir = dir
138 store, err := Open(o)
139 require.NoError(t, err)
140 require.NoError(t, store.Close())
141 statePath := filepath.Join(dir, "state.json")
142 raw, err := os.ReadFile(statePath)
143 require.NoError(t, err)
144 var legacyState state
145 require.NoError(t, strict(raw, &legacyState))
146 require.Len(t, legacyState.Hosts.Hosts, 1)
147 legacyState.Hosts.Hosts[0].Route = ports.BrokerRouteSpec{}
148 stripped, err := json.Marshal(legacyState)
149 require.NoError(t, err)
150 require.NoError(t, os.WriteFile(statePath, stripped, 0600))
151 return dir
152 }
153
154 for _, point := range []string{"recovery.json:write", "state.json:write", "state.json:rename", "state.json:dirsync"} {
155 t.Run(point, func(t *testing.T) {
156 dir := seedStripped(t)
157 fault := errors.New("power loss")
158 failed := Options{Dir: dir, Fault: func(p string) error {
159 if p == point {
160 return fault
161 }
162 return nil
163 }}
164 failedStore, err := Open(failed)
165 require.ErrorIs(t, err, fault)
166 require.Nil(t, failedStore)
167
168 // The upgrade is atomic: reopening without the fault always yields a
169 // readable, routed store, whether or not the failed attempt renamed.
170 store, err := Open(Options{Dir: dir})
171 require.NoError(t, err)
172 hosts, err := store.LoadHosts()
173 require.NoError(t, err)
174 require.Len(t, hosts.Hosts, 1)
175 require.Equal(t, canonicalQUICRoute("user@arch"), hosts.Hosts[0].Route)
176 require.NoError(t, hosts.Validate())
177 require.NoError(t, store.Close())
178
179 // The upgrade is idempotent: a further reopen keeps the same authority.
180 store, err = Open(Options{Dir: dir})
181 require.NoError(t, err)
182 defer store.Close()
183 again, err := store.LoadHosts()
184 require.NoError(t, err)
185 require.Equal(t, hosts, again)
186 })
187 }
188}
189
190// TestRouteChangeRequiresFreshGenerationAndRefusesOldFence proves a policy
191// change that re-routes an endpoint is admitted only with a fresh registration
192// generation, is persisted with the canonical route, and fences the retired
193// registration: a publication bound to the old registration is refused.
194func TestRouteChangeRequiresFreshGenerationAndRefusesOldFence(t *testing.T) {
195 o := options(t)
196 store, err := Open(o)
197 require.NoError(t, err)
198 defer store.Close()
199 h, err := store.LoadHosts()
200 require.NoError(t, err)
201 require.Len(t, h.Hosts, 1)
202 base := h.Hosts[0]
203 require.Equal(t, canonicalQUICRoute(base.Registration.Endpoint), base.Route)
204
205 stdio := base.Policy
206 stdio.Transport = "stdio"
207
208 // A re-route without a fresh generation is refused as a stale authority
209 // change: a queued observation must never be accepted under the new route.
210 changed := append([]ports.BrokerHostRecord(nil), h.Hosts...)
211 changed[0].Policy = stdio
212 require.ErrorIs(t, store.ReplaceHosts(h.Revision, changed), ErrStale)
213
214 // The same change with a fresh generation is accepted and durable, with the
215 // route the mutation path derives from the new transport.
216 changed[0].Registration.Generation++
217 changed[0].Route = ports.BrokerRouteSpec{Kind: ports.BrokerRouteSSHStdio, Target: base.Registration.Endpoint, Argv: []string{"vev", "_broker-mux-stdio", "--production"}}
218 require.NoError(t, store.ReplaceHosts(h.Revision, changed))
219 after, err := store.LoadHosts()
220 require.NoError(t, err)
221 require.Equal(t, base.Registration.Generation+1, after.Hosts[0].Registration.Generation)
222 require.Equal(t, stdio, after.Hosts[0].Policy)
223 require.Equal(t, changed[0].Route, after.Hosts[0].Route)
224
225 // The retired registration is fenced: a publication naming it is refused even
226 // though the endpoint is still live under a newer generation.
227 retired := ports.BrokerDaemonObservation{
228 Endpoint: base.Registration.Endpoint,
229 DisplayOrigin: domain.RemoteDisplayOrigin(base.Registration.Endpoint),
230 Registration: base.Registration,
231 Availability: domain.RemoteAvailabilityReachable,
232 }
233 require.ErrorIs(t, store.Store(ports.BrokerSnapshot{Epoch: 3, Revision: 1, Daemons: []ports.BrokerDaemonObservation{retired}}), ErrStale)
234}