1package brokerwire
2
3import (
4 "encoding/binary"
5 "fmt"
6 "sync"
7 "testing"
8 "time"
9
10 "github.com/stretchr/testify/assert"
11 "github.com/stretchr/testify/require"
12
13 "github.com/bnema/vev/internal/domain"
14 "github.com/bnema/vev/internal/ports"
15 "github.com/bnema/vev/internal/protocol/catalogue"
16)
17
18func testSnapshotScope() (ports.BrokerEpoch, ports.BrokerConnectionID) {
19 return 7, testConnectionID(0x21)
20}
21
22func testSnapshotAssembler(t *testing.T, opts ...SnapshotOption) *SnapshotAssembler {
23 t.Helper()
24 epoch, connection := testSnapshotScope()
25 return NewSnapshotAssembler(epoch, connection, opts...)
26}
27
28// testDaemonAt builds one catalogue-valid remote daemon projection for daemon
29// index i. Sessions are attached by the sample helpers; the emitter strips
30// them before a daemon part travels.
31func testDaemonAt(i int) ports.BrokerDaemonObservation {
32 endpoint := fmt.Sprintf("user%d@host%d:22", i, i)
33 return ports.BrokerDaemonObservation{
34 Endpoint: endpoint,
35 DisplayOrigin: endpoint,
36 Registration: domain.RemoteRegistration{Endpoint: endpoint, Incarnation: [16]byte{0x10, byte(i + 1), 0x33, 0x44, 0x55, 0x66, 0x77, 0x88, 0x99, 0xaa, 0xbb, 0xcc, 0xdd, 0xee, 0xf0, 0x11}, Generation: domain.RemoteGeneration(i + 1)},
37 Policy: testPolicy(),
38 Availability: domain.RemoteAvailabilityReachable,
39 LastSuccess: time.Unix(1700000000+int64(i), 0).UTC(),
40 InventoryKnown: true,
41 Sessions: []catalogue.RemoteCatalogSession{},
42 }
43}
44
45// testLocalDaemonAt builds the local daemon projection.
46func testLocalDaemonAt() ports.BrokerDaemonObservation {
47 return ports.BrokerDaemonObservation{
48 Local: true,
49 DisplayOrigin: "local",
50 Policy: testPolicy(),
51 Availability: domain.RemoteAvailabilityReachable,
52 InventoryKnown: true,
53 Sessions: []catalogue.RemoteCatalogSession{},
54 }
55}
56
57// testSessionAt builds one catalogue-valid session for a host and index.
58func testSessionAt(host, index int) catalogue.RemoteCatalogSession {
59 return catalogue.RemoteCatalogSession{
60 LifecycleID: domain.SessionLifecycleID{byte(host + 1), byte(index >> 8), byte(index), 0x77},
61 Name: fmt.Sprintf("s%02d-%03d", host, index),
62 State: catalogue.RemoteCatalogSessionDown,
63 Tabs: []catalogue.RemoteCatalogTab{},
64 }
65}
66
67// testTombstoneAt builds one valid retired registration for index i.
68func testTombstoneAt(i int) ports.BrokerHostTombstone {
69 endpoint := fmt.Sprintf("retired%d@old%d:22", i, i)
70 return ports.BrokerHostTombstone{
71 Endpoint: endpoint,
72 Registration: domain.RemoteRegistration{Endpoint: endpoint, Incarnation: [16]byte{0x90, byte(i + 1), 0x01, 0x02, 0x03, 0x04, 0x05, 0x06, 0x07, 0x08, 0x09, 0x0a, 0x0b, 0x0c, 0x0d, 0x0e}, Generation: domain.RemoteGeneration(i + 1)},
73 RetiredRevision: ports.BrokerRevision(i + 1),
74 }
75}
76
77func sampleSnapshotDaemons() []ports.BrokerDaemonObservation {
78 daemon0 := testDaemonAt(0)
79 daemon0.Sessions = []catalogue.RemoteCatalogSession{testSessionAt(0, 0), testSessionAt(0, 1)}
80 daemon1 := testDaemonAt(1)
81 daemon1.Sessions = []catalogue.RemoteCatalogSession{testSessionAt(1, 0)}
82 return []ports.BrokerDaemonObservation{daemon0, daemon1}
83}
84
85// sampleLocalSnapshotDaemons builds a local-first publication: the local daemon
86// at index 0 with two sessions, then one remote daemon with one session.
87func sampleLocalSnapshotDaemons() []ports.BrokerDaemonObservation {
88 local := testLocalDaemonAt()
89 local.Sessions = []catalogue.RemoteCatalogSession{testSessionAt(0, 0), testSessionAt(0, 1)}
90 remote := testDaemonAt(0)
91 remote.Sessions = []catalogue.RemoteCatalogSession{testSessionAt(1, 0)}
92 return []ports.BrokerDaemonObservation{local, remote}
93}
94
95func sampleSnapshotTombstones() []ports.BrokerHostTombstone {
96 return []ports.BrokerHostTombstone{testTombstoneAt(0)}
97}
98
99func snapshotSessionTotal(daemons []ports.BrokerDaemonObservation) int {
100 total := 0
101 for _, daemon := range daemons {
102 total += len(daemon.Sessions)
103 }
104 return total
105}
106
107// snapshotPartsFor lays out one exact transfer: Begin(0), daemons with their
108// sessions, tombstones, End. Indexes are exact sequential positions.
109func snapshotPartsFor(gen uint64, rev ports.BrokerRevision, daemons []ports.BrokerDaemonObservation, tombstones []ports.BrokerHostTombstone) []SnapshotPart {
110 epoch, connection := testSnapshotScope()
111 next := uint32(0)
112 part := func(payload SnapshotPartPayload) SnapshotPart {
113 out := SnapshotPart{Epoch: epoch, Connection: connection, Generation: gen, Revision: rev, Index: next, Part: payload}
114 next++
115 return out
116 }
117 localPresent := len(daemons) > 0 && daemons[0].Local
118 parts := []SnapshotPart{part(SnapshotBegin{
119 HostCount: uint32(len(daemons)),
120 SessionCount: uint32(snapshotSessionTotal(daemons)),
121 TombstoneCount: uint32(len(tombstones)),
122 LocalPresent: localPresent,
123 })}
124 for daemonIndex, daemon := range daemons {
125 // Daemon parts carry no inline sessions: the advertised count is
126 // followed by that many session parts.
127 projection := daemon.Clone()
128 projection.Sessions = []catalogue.RemoteCatalogSession{}
129 parts = append(parts, part(SnapshotDaemonPart{HostIndex: uint32(daemonIndex), Daemon: projection, SessionCount: uint32(len(daemon.Sessions))}))
130 for sessionIndex, session := range daemon.Sessions {
131 parts = append(parts, part(SnapshotSessionPart{HostIndex: uint32(daemonIndex), SessionIndex: uint32(sessionIndex), Local: daemon.Local, Session: session}))
132 }
133 }
134 for tombstoneIndex, tombstone := range tombstones {
135 parts = append(parts, part(SnapshotTombstonePart{
136 TombstoneIndex: uint32(tombstoneIndex),
137 Registration: tombstone.Registration,
138 RetiredRevision: tombstone.RetiredRevision,
139 }))
140 }
141 return append(parts, part(SnapshotEnd{}))
142}
143
144func cloneSnapshotParts(parts []SnapshotPart) []SnapshotPart {
145 return append([]SnapshotPart(nil), parts...)
146}
147
148// stagePart stages one part and requires acceptance.
149func stagePart(t *testing.T, a *SnapshotAssembler, part SnapshotPart) {
150 t.Helper()
151 _, _, err := a.Add(part)
152 require.NoError(t, err)
153}
154
155// runSnapshotParts stages parts in order and returns the first staging error.
156func runSnapshotParts(a *SnapshotAssembler, parts []SnapshotPart) error {
157 for _, part := range parts {
158 if _, _, err := a.Add(part); err != nil {
159 return err
160 }
161 }
162 return nil
163}
164
165func TestSnapshotAssemblerBoundsConstants(t *testing.T) {
166 require.Equal(t, 16771, MaxSnapshotParts)
167 require.Equal(t, 1+ports.BrokerMaxDaemonsPerSnapshot+ports.BrokerMaxDaemonsPerSnapshot*ports.BrokerMaxSessionsPerHost+ports.BrokerMaxTombstones+1, MaxSnapshotParts)
168 require.Equal(t, uint64(80<<20), MaxSnapshotStagedBytes)
169}
170
171// TestSnapshotAssemblerExactTransfer proves exact order, count, and index
172// acceptance, and that committed order matches the staged order.
173func TestSnapshotAssemblerExactTransfer(t *testing.T) {
174 hosts := sampleSnapshotDaemons()
175 tombstones := sampleSnapshotTombstones()
176 parts := snapshotPartsFor(5, 3, hosts, tombstones)
177 require.Len(t, parts, 1+2+3+1+1)
178
179 a := testSnapshotAssembler(t)
180 for i, part := range parts {
181 completed, snapshot, err := a.Add(part)
182 require.NoError(t, err, "part %d", i)
183 if i == len(parts)-1 {
184 require.True(t, completed)
185 require.Equal(t, ports.BrokerRevision(3), snapshot.Revision)
186 } else {
187 require.False(t, completed, "part %d", i)
188 _, committed := a.Snapshot()
189 require.False(t, committed, "part %d must not publish", i)
190 require.True(t, a.StagingActive())
191 }
192 }
193 committed, ok := a.Snapshot()
194 require.True(t, ok)
195 require.False(t, a.StagingActive())
196 epoch, connection := testSnapshotScope()
197 require.Equal(t, epoch, committed.Epoch)
198 require.Equal(t, ports.BrokerRevision(3), committed.Revision)
199 require.Len(t, committed.Daemons, 2)
200 require.Equal(t, "user0@host0:22", committed.Daemons[0].Endpoint)
201 require.Equal(t, "user1@host1:22", committed.Daemons[1].Endpoint)
202 require.Equal(t, 2, len(committed.Daemons[0].Sessions))
203 require.Equal(t, "s00-000", committed.Daemons[0].Sessions[0].Name)
204 require.Equal(t, "s00-001", committed.Daemons[0].Sessions[1].Name)
205 require.Equal(t, "s01-000", committed.Daemons[1].Sessions[0].Name)
206 require.Len(t, committed.Removed, 1)
207 require.Equal(t, "retired0@old0:22", committed.Removed[0].Endpoint)
208 _ = connection
209}
210
211func TestSnapshotAssemblerEmptyTransfer(t *testing.T) {
212 parts := snapshotPartsFor(1, 1, nil, nil)
213 require.Len(t, parts, 2)
214 a := testSnapshotAssembler(t)
215 completed, _, err := a.Add(parts[0])
216 require.NoError(t, err)
217 require.False(t, completed)
218 require.True(t, a.StagingActive())
219 completed, snapshot, err := a.Add(parts[1])
220 require.NoError(t, err)
221 require.True(t, completed)
222 require.Empty(t, snapshot.Daemons)
223 require.Empty(t, snapshot.Removed)
224}
225
226// TestSnapshotAssemblerLocalDaemonTransfer proves a local-first transfer with a
227// local daemon and its sessions commits with the local entry first, and that
228// local sessions bind to index 0 while a mismatched local flag is refused.
229func TestSnapshotAssemblerLocalDaemonTransfer(t *testing.T) {
230 daemons := sampleLocalSnapshotDaemons()
231 parts := snapshotPartsFor(5, 1, daemons, nil)
232 require.Len(t, parts, 1+2+3+1)
233 begin, ok := parts[0].Part.(SnapshotBegin)
234 require.True(t, ok)
235 require.True(t, begin.LocalPresent)
236 require.Equal(t, uint32(2), begin.HostCount)
237 require.Equal(t, uint32(3), begin.SessionCount)
238
239 a := testSnapshotAssembler(t)
240 completed, snapshot, err := addAll(a, parts)
241 require.NoError(t, err)
242 require.True(t, completed)
243 require.Len(t, snapshot.Daemons, 2)
244 require.True(t, snapshot.Daemons[0].Local)
245 require.Empty(t, snapshot.Daemons[0].Endpoint)
246 require.Equal(t, "user0@host0:22", snapshot.Daemons[1].Endpoint)
247 require.Len(t, snapshot.Daemons[0].Sessions, 2)
248 require.Equal(t, "s00-000", snapshot.Daemons[0].Sessions[0].Name)
249 require.Len(t, snapshot.Daemons[1].Sessions, 1)
250 requireValidBrokerSnapshot(t, snapshot)
251
252 t.Run("local session marked remote refused", func(t *testing.T) {
253 a := testSnapshotAssembler(t)
254 require.NoError(t, runSnapshotParts(a, parts[:2]))
255 require.True(t, a.StagingActive())
256 mismatched := parts[2]
257 mismatched.Part = SnapshotSessionPart{HostIndex: 0, SessionIndex: 0, Local: false, Session: testSessionAt(0, 0)}
258 _, _, err := a.Add(mismatched)
259 require.ErrorIs(t, err, ErrSnapshotInvalid)
260 require.False(t, a.StagingActive())
261 })
262
263 t.Run("non-local daemon under local_present refused", func(t *testing.T) {
264 a := testSnapshotAssembler(t)
265 begin := parts[0]
266 begin.Part = SnapshotBegin{HostCount: 1, LocalPresent: true}
267 require.NoError(t, runSnapshotParts(a, []SnapshotPart{begin}))
268 daemonPart := parts[1]
269 daemonPart.Part = SnapshotDaemonPart{HostIndex: 0, Daemon: testDaemonAt(0)}
270 _, _, err := a.Add(daemonPart)
271 require.ErrorIs(t, err, ErrSnapshotInvalid)
272 require.False(t, a.StagingActive())
273 })
274
275 t.Run("local daemon without local_present refused", func(t *testing.T) {
276 a := testSnapshotAssembler(t)
277 begin := parts[0]
278 begin.Part = SnapshotBegin{HostCount: 1}
279 require.NoError(t, runSnapshotParts(a, []SnapshotPart{begin}))
280 daemonPart := parts[1]
281 daemonPart.Part = SnapshotDaemonPart{HostIndex: 0, Daemon: testLocalDaemonAt()}
282 _, _, err := a.Add(daemonPart)
283 require.ErrorIs(t, err, ErrSnapshotInvalid)
284 require.False(t, a.StagingActive())
285 })
286}
287
288// addAll stages every part in order and returns the final commit result.
289func addAll(a *SnapshotAssembler, parts []SnapshotPart) (bool, ports.BrokerSnapshot, error) {
290 completed := false
291 var snapshot ports.BrokerSnapshot
292 for _, part := range parts {
293 published, committed, err := a.Add(part)
294 if err != nil {
295 return false, ports.BrokerSnapshot{}, err
296 }
297 if published {
298 completed = true
299 snapshot = committed
300 }
301 }
302 return completed, snapshot, nil
303}
304
305// TestSnapshotAssemblerTransferRejections proves malformed indexes, counts,
306// order, payloads, and assembled validation failures abort staging and
307// retain the committed snapshot.
308func TestSnapshotAssemblerTransferRejections(t *testing.T) {
309 base := func() []SnapshotPart {
310 return snapshotPartsFor(5, 3, sampleSnapshotDaemons(), sampleSnapshotTombstones())
311 }
312 mutate := func(parts []SnapshotPart, index int, apply func(*SnapshotPart)) []SnapshotPart {
313 out := cloneSnapshotParts(parts)
314 apply(&out[index])
315 return out
316 }
317 cases := []struct {
318 name string
319 parts []SnapshotPart
320 wantErr error
321 }{
322 {"begin index not zero", mutate(base(), 0, func(p *SnapshotPart) { p.Index = 1 }), ErrSnapshotInvalid},
323 {"non-begin first", base()[1:], ErrSnapshotInvalid},
324 {"missing session part", append(cloneSnapshotParts(base()[:3]), base()[4:]...), ErrSnapshotInvalid},
325 {"duplicate host index", mutate(base(), 4, func(p *SnapshotPart) {
326 p.Part = SnapshotDaemonPart{HostIndex: 0, Daemon: testDaemonAt(1), SessionCount: 1}
327 }), ErrSnapshotInvalid},
328 {"host index skips ahead", mutate(base(), 4, func(p *SnapshotPart) {
329 p.Part = SnapshotDaemonPart{HostIndex: 5, Daemon: testDaemonAt(1), SessionCount: 1}
330 }), ErrSnapshotInvalid},
331 {"session host index mismatch", mutate(base(), 3, func(p *SnapshotPart) {
332 p.Part = SnapshotSessionPart{HostIndex: 1, SessionIndex: 1, Session: testSessionAt(0, 1)}
333 }), ErrSnapshotInvalid},
334 {"session index gap", mutate(base(), 3, func(p *SnapshotPart) {
335 p.Part = SnapshotSessionPart{HostIndex: 0, SessionIndex: 2, Session: testSessionAt(0, 1)}
336 }), ErrSnapshotInvalid},
337 {"session tabs absent", mutate(base(), 2, func(p *SnapshotPart) {
338 session := testSessionAt(0, 0)
339 session.Tabs = nil
340 p.Part = SnapshotSessionPart{HostIndex: 0, SessionIndex: 0, Session: session}
341 }), ErrSnapshotInvalid},
342 {"host carries inline sessions", mutate(base(), 1, func(p *SnapshotPart) {
343 host := testDaemonAt(0)
344 host.Sessions = []catalogue.RemoteCatalogSession{testSessionAt(0, 0)}
345 p.Part = SnapshotDaemonPart{HostIndex: 0, Daemon: host, SessionCount: 1}
346 }), ErrSnapshotInvalid},
347 {"host session count over maximum", mutate(base(), 1, func(p *SnapshotPart) {
348 p.Part = SnapshotDaemonPart{HostIndex: 0, Daemon: testDaemonAt(0), SessionCount: uint32(ports.BrokerMaxSessionsPerHost) + 1}
349 }), ErrTooLarge},
350 {"host session count exceeds advertised total", mutate(base(), 1, func(p *SnapshotPart) {
351 p.Part = SnapshotDaemonPart{HostIndex: 0, Daemon: testDaemonAt(0), SessionCount: 3}
352 }), ErrSnapshotInvalid},
353 {"tombstone before all sessions", func() []SnapshotPart {
354 parts := base()
355 reordered := []SnapshotPart{parts[0], parts[1], parts[2], parts[3], parts[6], parts[4], parts[5], parts[7]}
356 for i := range reordered {
357 reordered[i].Index = uint32(i)
358 }
359 return reordered
360 }(), ErrSnapshotInvalid},
361 {"tombstone index out of order", mutate(base(), 6, func(p *SnapshotPart) {
362 p.Part = SnapshotTombstonePart{TombstoneIndex: 1, Registration: testRegistration(), RetiredRevision: 1}
363 }), ErrSnapshotInvalid},
364 {"tombstone missing retired revision", mutate(base(), 6, func(p *SnapshotPart) {
365 p.Part = SnapshotTombstonePart{TombstoneIndex: 0, Registration: testRegistration(), RetiredRevision: 0}
366 }), ErrSnapshotInvalid},
367 {"end before completion", append(cloneSnapshotParts(base()[:5]), SnapshotPart{
368 Epoch: 7, Connection: testConnectionID(0x21), Generation: 5, Revision: 3, Index: 5, Part: SnapshotEnd{},
369 }), ErrSnapshotInvalid},
370 {"begin session count mismatch", mutate(base(), 0, func(p *SnapshotPart) {
371 p.Part = SnapshotBegin{HostCount: 2, SessionCount: 4, TombstoneCount: 1}
372 }), ErrSnapshotInvalid},
373 {"begin host count over maximum", mutate(base(), 0, func(p *SnapshotPart) {
374 p.Part = SnapshotBegin{HostCount: uint32(ports.BrokerMaxDaemonsPerSnapshot) + 1}
375 }), ErrTooLarge},
376 {"begin session count over maximum", mutate(base(), 0, func(p *SnapshotPart) {
377 p.Part = SnapshotBegin{HostCount: 1, SessionCount: uint32(ports.BrokerMaxDaemonsPerSnapshot*ports.BrokerMaxSessionsPerHost) + 1}
378 }), ErrTooLarge},
379 {"begin tombstone count over maximum", mutate(base(), 0, func(p *SnapshotPart) {
380 p.Part = SnapshotBegin{HostCount: 1, TombstoneCount: uint32(ports.BrokerMaxTombstones) + 1}
381 }), ErrTooLarge},
382 {"commit duplicate host endpoint", func() []SnapshotPart {
383 hosts := sampleSnapshotDaemons()
384 hosts[1].Endpoint = hosts[0].Endpoint
385 hosts[1].Registration.Endpoint = hosts[0].Endpoint
386 return snapshotPartsFor(5, 3, hosts, nil)
387 }(), ErrSnapshotInvalid},
388 {"commit host endpoint registration mismatch", func() []SnapshotPart {
389 hosts := sampleSnapshotDaemons()
390 hosts[0].Endpoint = "other@elsewhere:22"
391 return snapshotPartsFor(5, 3, hosts, nil)
392 }(), ErrSnapshotInvalid},
393 {"commit invalid projection", func() []SnapshotPart {
394 hosts := sampleSnapshotDaemons()
395 hosts[0].Availability = 0
396 return snapshotPartsFor(5, 3, hosts, nil)
397 }(), ErrSnapshotInvalid},
398 {"commit duplicate session name", func() []SnapshotPart {
399 hosts := sampleSnapshotDaemons()
400 hosts[0].Sessions[1].Name = hosts[0].Sessions[0].Name
401 return snapshotPartsFor(5, 3, hosts, nil)
402 }(), ErrSnapshotInvalid},
403 {"commit tombstone for live host", func() []SnapshotPart {
404 hosts := sampleSnapshotDaemons()
405 tombstone := testTombstoneAt(0)
406 tombstone.Endpoint = hosts[0].Endpoint
407 tombstone.Registration.Endpoint = hosts[0].Endpoint
408 return snapshotPartsFor(5, 3, hosts, []ports.BrokerHostTombstone{tombstone})
409 }(), ErrSnapshotInvalid},
410 }
411 for _, tc := range cases {
412 t.Run(tc.name, func(t *testing.T) {
413 a := testSnapshotAssembler(t)
414 err := runSnapshotParts(a, tc.parts)
415 require.ErrorIs(t, err, tc.wantErr)
416 require.False(t, a.StagingActive())
417 _, committed := a.Snapshot()
418 require.False(t, committed)
419 })
420 }
421}
422
423// TestSnapshotAssemblerLatePartAfterCommit proves a part that follows a
424// committed transfer is refused as malformed while the committed snapshot
425// survives.
426func TestSnapshotAssemblerLatePartAfterCommit(t *testing.T) {
427 a := testSnapshotAssembler(t)
428 parts := snapshotPartsFor(5, 3, sampleSnapshotDaemons(), sampleSnapshotTombstones())
429 require.NoError(t, runSnapshotParts(a, parts))
430 late := parts[len(parts)-1]
431 late.Index = uint32(len(parts))
432 _, _, err := a.Add(late)
433 require.ErrorIs(t, err, ErrSnapshotInvalid)
434 require.False(t, a.StagingActive())
435 committed, ok := a.Snapshot()
436 require.True(t, ok)
437 require.Equal(t, ports.BrokerRevision(3), committed.Revision)
438}
439
440// TestSnapshotAssemblerBounds proves the 16,771-part and 80 MiB staged
441// ceilings: maximum counts commit, one over is refused.
442func TestSnapshotAssemblerBounds(t *testing.T) {
443 t.Run("maximum counts commit", func(t *testing.T) {
444 hosts := make([]ports.BrokerDaemonObservation, ports.BrokerMaxDaemonsPerSnapshot)
445 for i := range hosts {
446 host := testDaemonAt(i)
447 sessions := make([]catalogue.RemoteCatalogSession, ports.BrokerMaxSessionsPerHost)
448 for j := range sessions {
449 sessions[j] = testSessionAt(i, j)
450 }
451 host.Sessions = sessions
452 hosts[i] = host
453 }
454 tombstones := make([]ports.BrokerHostTombstone, ports.BrokerMaxTombstones)
455 for i := range tombstones {
456 tombstones[i] = testTombstoneAt(i)
457 }
458 parts := snapshotPartsFor(9, 12, hosts, tombstones)
459 require.Len(t, parts, MaxSnapshotParts)
460 a := testSnapshotAssembler(t)
461 completed := false
462 var snapshot ports.BrokerSnapshot
463 for _, part := range parts {
464 published, committed, err := a.Add(part)
465 require.NoError(t, err)
466 if published {
467 completed = true
468 snapshot = committed
469 }
470 }
471 require.True(t, completed)
472 require.Len(t, snapshot.Daemons, ports.BrokerMaxDaemonsPerSnapshot)
473 require.Len(t, snapshot.Removed, ports.BrokerMaxTombstones)
474 for _, host := range snapshot.Daemons {
475 require.Len(t, host.Sessions, ports.BrokerMaxSessionsPerHost)
476 }
477 require.Equal(t, ports.BrokerRevision(12), snapshot.Revision)
478 })
479
480 t.Run("one part over maximum refused", func(t *testing.T) {
481 a := testSnapshotAssembler(t)
482 _, _, err := a.Add(SnapshotPart{
483 Epoch: 7, Connection: testConnectionID(0x21), Generation: 1, Revision: 1, Index: 0,
484 Part: SnapshotBegin{HostCount: 1, SessionCount: uint32(ports.BrokerMaxDaemonsPerSnapshot*ports.BrokerMaxSessionsPerHost) + 1},
485 })
486 require.ErrorIs(t, err, ErrTooLarge)
487 require.False(t, a.StagingActive())
488 })
489
490 t.Run("staged byte ceiling aborts and retains", func(t *testing.T) {
491 a := testSnapshotAssembler(t, WithMaxStagedBytes(200))
492 parts := snapshotPartsFor(5, 3, sampleSnapshotDaemons(), nil)
493 // Begin (32) + host part stay within the cap; the first session part
494 // crosses it.
495 require.NoError(t, runSnapshotParts(a, parts[:2]))
496 require.True(t, a.StagingActive())
497 _, _, err := a.Add(parts[2])
498 require.ErrorIs(t, err, ErrTooLarge)
499 require.False(t, a.StagingActive())
500 _, committed := a.Snapshot()
501 require.False(t, committed)
502 })
503
504 t.Run("zero staged cap falls back to default", func(t *testing.T) {
505 a := testSnapshotAssembler(t, WithMaxStagedBytes(0))
506 require.Equal(t, MaxSnapshotStagedBytes, a.maxStagedBytes)
507 })
508}
509
510// TestSnapshotStagedEstimateTabCharge proves the staged-memory estimate
511// charges at least 64 bytes of overhead per tab plus the tab-list slice
512// header, so a minimal tab cannot hide behind tiny strings.
513func TestSnapshotStagedEstimateTabCharge(t *testing.T) {
514 base := testSessionAt(0, 0)
515 base.Tabs = []catalogue.RemoteCatalogTab{}
516 oneTab := base
517 oneTab.Tabs = []catalogue.RemoteCatalogTab{{Name: "t"}}
518 delta := estimateCatalogSessionBytes(oneTab) - estimateCatalogSessionBytes(base)
519 require.GreaterOrEqual(t, delta, uint64(64), "each tab must charge at least 64 bytes")
520
521 maxTabs := make([]catalogue.RemoteCatalogTab, catalogue.RemoteCatalogMaxTabsPerSess)
522 for i := range maxTabs {
523 maxTabs[i] = catalogue.RemoteCatalogTab{Index: uint16(i), Name: "t"}
524 }
525 hostile := base
526 hostile.Tabs = maxTabs
527 require.GreaterOrEqual(t, estimateCatalogSessionBytes(hostile), uint64(catalogue.RemoteCatalogMaxTabsPerSess*64))
528 // The slice header is charged when the tab list is present.
529 present := base
530 present.Tabs = []catalogue.RemoteCatalogTab{}
531 absent := base
532 absent.Tabs = nil
533 require.Greater(t, estimateCatalogSessionBytes(present), estimateCatalogSessionBytes(absent))
534}
535
536// TestSnapshotHostileMaxTabsBound proves the 80 MiB staged ceiling against
537// hostile max tabs: the worst-case arithmetic crosses the ceiling, a max-tab
538// session crosses a ceiling that fits exactly one session, and a ceiling
539// that admits it still commits.
540func TestSnapshotHostileMaxTabsBound(t *testing.T) {
541 require.Equal(t, uint64(80<<20), MaxSnapshotStagedBytes)
542
543 maxTabs := make([]catalogue.RemoteCatalogTab, catalogue.RemoteCatalogMaxTabsPerSess)
544 for i := range maxTabs {
545 maxTabs[i] = catalogue.RemoteCatalogTab{Index: uint16(i), Name: "t"}
546 }
547 session := testSessionAt(0, 0)
548 session.Tabs = maxTabs
549 perSession := estimateCatalogSessionBytes(session)
550 worst := perSession * uint64(ports.BrokerMaxDaemonsPerSnapshot*ports.BrokerMaxSessionsPerHost)
551 require.Greater(t, worst, MaxSnapshotStagedBytes, "hostile max tabs must cross the 80 MiB ceiling")
552
553 host := testDaemonAt(0)
554 host.Sessions = []catalogue.RemoteCatalogSession{session}
555 parts := snapshotPartsFor(9, 1, []ports.BrokerDaemonObservation{host}, nil)
556
557 tight := testSnapshotAssembler(t, WithMaxStagedBytes(perSession))
558 require.NoError(t, runSnapshotParts(tight, parts[:2]))
559 require.True(t, tight.StagingActive())
560 _, _, err := tight.Add(parts[2])
561 require.ErrorIs(t, err, ErrTooLarge)
562 require.False(t, tight.StagingActive())
563
564 generous := testSnapshotAssembler(t, WithMaxStagedBytes(perSession*4))
565 require.NoError(t, runSnapshotParts(generous, parts))
566 committed, ok := generous.Snapshot()
567 require.True(t, ok)
568 require.Len(t, committed.Daemons, 1)
569 require.Len(t, committed.Daemons[0].Sessions, 1)
570 require.Len(t, committed.Daemons[0].Sessions[0].Tabs, catalogue.RemoteCatalogMaxTabsPerSess)
571}
572
573// TestSnapshotAssemblerAtomicCommit proves publication happens only at End
574// and that the returned snapshot is a defensive clone.
575func TestSnapshotAssemblerAtomicCommit(t *testing.T) {
576 a := testSnapshotAssembler(t)
577 first := snapshotPartsFor(5, 1, sampleSnapshotDaemons(), sampleSnapshotTombstones())
578 require.NoError(t, runSnapshotParts(a, first))
579 committed, ok := a.Snapshot()
580 require.True(t, ok)
581 require.Equal(t, ports.BrokerRevision(1), committed.Revision)
582
583 second := snapshotPartsFor(5, 2, sampleSnapshotDaemons(), nil)
584 for _, part := range second[:len(second)-1] {
585 completed, _, err := a.Add(part)
586 require.NoError(t, err)
587 require.False(t, completed)
588 }
589 // Staging a newer revision does not disturb the committed snapshot.
590 committed, ok = a.Snapshot()
591 require.True(t, ok)
592 require.Equal(t, ports.BrokerRevision(1), committed.Revision)
593 require.Len(t, committed.Removed, 1)
594
595 completed, published, err := a.Add(second[len(second)-1])
596 require.NoError(t, err)
597 require.True(t, completed)
598 require.Equal(t, ports.BrokerRevision(2), published.Revision)
599 require.Empty(t, published.Removed)
600
601 // The committed snapshot is immutable to the caller: mutating a returned
602 // value never reaches the assembler.
603 published.Daemons[0].Endpoint = "mutated@nowhere:22"
604 published.Daemons[0].Sessions[0].Tabs = []catalogue.RemoteCatalogTab{{ID: "injected", Index: 0, Name: "injected"}}
605 published.Daemons[0].Sessions[0].Name = "mutated"
606 again, ok := a.Snapshot()
607 require.True(t, ok)
608 require.Equal(t, "user0@host0:22", again.Daemons[0].Endpoint)
609 require.Equal(t, "s00-000", again.Daemons[0].Sessions[0].Name)
610 require.Empty(t, again.Daemons[0].Sessions[0].Tabs)
611
612 // An aborted transfer retains the committed snapshot.
613 require.NoError(t, runSnapshotParts(a, snapshotPartsFor(5, 3, sampleSnapshotDaemons(), nil)[:2]))
614 _, _, err = a.Add(snapshotPartsFor(5, 3, sampleSnapshotDaemons(), nil)[4])
615 require.ErrorIs(t, err, ErrSnapshotInvalid)
616 again, ok = a.Snapshot()
617 require.True(t, ok)
618 require.Equal(t, ports.BrokerRevision(2), again.Revision)
619}
620
621// TestSnapshotAssemblerInterruptedRetention proves an interrupted transfer is
622// discarded while the committed snapshot survives, and that a restarted
623// transfer starts from index 0.
624func TestSnapshotAssemblerInterruptedRetention(t *testing.T) {
625 a := testSnapshotAssembler(t)
626 parts := snapshotPartsFor(5, 1, sampleSnapshotDaemons(), sampleSnapshotTombstones())
627 require.NoError(t, runSnapshotParts(a, parts))
628 committed, ok := a.Snapshot()
629 require.True(t, ok)
630 require.Equal(t, ports.BrokerRevision(1), committed.Revision)
631
632 restart := snapshotPartsFor(5, 2, sampleSnapshotDaemons(), sampleSnapshotTombstones())
633 require.NoError(t, runSnapshotParts(a, restart[:3]))
634 require.True(t, a.StagingActive())
635 a.DiscardStaging()
636 require.False(t, a.StagingActive())
637 committed, ok = a.Snapshot()
638 require.True(t, ok)
639 require.Equal(t, ports.BrokerRevision(1), committed.Revision)
640
641 // A restarted transfer must begin again at index 0, and re-sent parts are
642 // accepted from the start.
643 stagePart(t, a, restart[0])
644 require.True(t, a.StagingActive())
645 _, _, err := a.Add(restart[2])
646 require.ErrorIs(t, err, ErrSnapshotInvalid)
647 require.False(t, a.StagingActive())
648
649 stagePart(t, a, restart[0])
650 for _, part := range restart[1:] {
651 _, _, err := a.Add(part)
652 require.NoError(t, err)
653 }
654 committed, ok = a.Snapshot()
655 require.True(t, ok)
656 require.Equal(t, ports.BrokerRevision(2), committed.Revision)
657}
658
659// TestSnapshotAssemblerRevisionFencing proves older revisions are refused,
660// equal revisions are allowed, and a newer revision coalesces by replacing
661// staging.
662func TestSnapshotAssemblerRevisionFencing(t *testing.T) {
663 a := testSnapshotAssembler(t)
664 hosts := sampleSnapshotDaemons()
665 require.NoError(t, runSnapshotParts(a, snapshotPartsFor(5, 5, hosts, nil)))
666
667 t.Run("older revision refused", func(t *testing.T) {
668 _, _, err := a.Add(snapshotPartsFor(5, 4, hosts, nil)[0])
669 require.ErrorIs(t, err, ErrSnapshotStale)
670 require.False(t, a.StagingActive())
671 committed, ok := a.Snapshot()
672 require.True(t, ok)
673 require.Equal(t, ports.BrokerRevision(5), committed.Revision)
674 })
675
676 t.Run("equal revision republishes", func(t *testing.T) {
677 require.NoError(t, runSnapshotParts(a, snapshotPartsFor(5, 5, hosts, sampleSnapshotTombstones())))
678 committed, ok := a.Snapshot()
679 require.True(t, ok)
680 require.Equal(t, ports.BrokerRevision(5), committed.Revision)
681 require.Len(t, committed.Removed, 1)
682 })
683
684 t.Run("newer revision coalesces", func(t *testing.T) {
685 newer := snapshotPartsFor(5, 6, hosts, nil)
686 stagePart(t, a, newer[0])
687 require.True(t, a.StagingActive())
688 // A part of the superseded staging revision is refused and staging is
689 // retained intact.
690 _, _, err := a.Add(snapshotPartsFor(5, 5, hosts, nil)[1])
691 require.ErrorIs(t, err, ErrSnapshotStale)
692 require.True(t, a.StagingActive())
693 // The fresh transfer restarts at index 1.
694 for _, part := range newer[1:] {
695 _, _, err := a.Add(part)
696 require.NoError(t, err)
697 }
698 committed, ok := a.Snapshot()
699 require.True(t, ok)
700 require.Equal(t, ports.BrokerRevision(6), committed.Revision)
701 })
702
703 t.Run("revision below committed refused while staging", func(t *testing.T) {
704 stagePart(t, a, snapshotPartsFor(5, 6, hosts, nil)[0])
705 require.True(t, a.StagingActive())
706 _, _, err := a.Add(snapshotPartsFor(5, 5, hosts, nil)[0])
707 require.ErrorIs(t, err, ErrSnapshotStale)
708 require.True(t, a.StagingActive())
709 })
710}
711
712// TestSnapshotAssemblerCoalescing proves a Begin of the same generation
713// replaces staging: equal revisions restart progress, newer revisions
714// coalesce.
715func TestSnapshotAssemblerCoalescing(t *testing.T) {
716 hosts := sampleSnapshotDaemons()
717 base := snapshotPartsFor(5, 3, hosts, nil)
718
719 a := testSnapshotAssembler(t)
720 stagePart(t, a, base[0])
721 stagePart(t, a, base[1])
722 require.True(t, a.StagingActive())
723 // Same revision restarts the transfer: progress is discarded.
724 stagePart(t, a, base[0])
725 require.True(t, a.StagingActive())
726 for _, part := range base[1:] {
727 _, _, err := a.Add(part)
728 require.NoError(t, err)
729 }
730 committed, ok := a.Snapshot()
731 require.True(t, ok)
732 require.Equal(t, ports.BrokerRevision(3), committed.Revision)
733
734 // A newer revision coalesces and also resets progress.
735 newer := snapshotPartsFor(5, 4, hosts, nil)
736 require.NoError(t, runSnapshotParts(a, newer[:3]))
737 require.True(t, a.StagingActive())
738 stagePart(t, a, newer[0])
739 require.True(t, a.StagingActive())
740 for _, part := range newer[1:] {
741 _, _, err := a.Add(part)
742 require.NoError(t, err)
743 }
744 committed, ok = a.Snapshot()
745 require.True(t, ok)
746 require.Equal(t, ports.BrokerRevision(4), committed.Revision)
747}
748
749// TestSnapshotAssemblerGenerationFencing proves the assembler adopts exactly
750// one generation, ignores stale parts, rejects future parts, and refuses a
751// zero generation.
752func TestSnapshotAssemblerGenerationFencing(t *testing.T) {
753 hosts := sampleSnapshotDaemons()
754 base := snapshotPartsFor(5, 3, hosts, nil)
755
756 t.Run("zero generation refused", func(t *testing.T) {
757 a := testSnapshotAssembler(t)
758 part := base[0]
759 part.Generation = 0
760 _, _, err := a.Add(part)
761 require.ErrorIs(t, err, ErrInvalidGeneration)
762 require.False(t, a.StagingActive())
763 })
764
765 t.Run("stale ignored and future rejected while staging", func(t *testing.T) {
766 a := testSnapshotAssembler(t)
767 stagePart(t, a, base[0])
768 stale := base[1]
769 stale.Generation = 4
770 _, _, err := a.Add(stale)
771 require.ErrorIs(t, err, ErrStaleGeneration)
772 require.True(t, a.StagingActive())
773 future := base[1]
774 future.Generation = 6
775 _, _, err = a.Add(future)
776 require.ErrorIs(t, err, ErrFutureGeneration)
777 require.True(t, a.StagingActive())
778 for _, part := range base[1:] {
779 _, _, err := a.Add(part)
780 require.NoError(t, err)
781 }
782 committed, ok := a.Snapshot()
783 require.True(t, ok)
784 require.Equal(t, ports.BrokerRevision(3), committed.Revision)
785 })
786
787 t.Run("generation is adopted for the assembler lifetime", func(t *testing.T) {
788 a := testSnapshotAssembler(t)
789 require.NoError(t, runSnapshotParts(a, base))
790 stale := base[1]
791 stale.Generation = 4
792 _, _, err := a.Add(stale)
793 require.ErrorIs(t, err, ErrStaleGeneration)
794 future := base[0]
795 future.Generation = 6
796 _, _, err = a.Add(future)
797 require.ErrorIs(t, err, ErrFutureGeneration)
798 committed, ok := a.Snapshot()
799 require.True(t, ok)
800 require.Equal(t, ports.BrokerRevision(3), committed.Revision)
801 // The same generation may republish a newer revision.
802 require.NoError(t, runSnapshotParts(a, snapshotPartsFor(5, 4, hosts, nil)))
803 committed, ok = a.Snapshot()
804 require.True(t, ok)
805 require.Equal(t, ports.BrokerRevision(4), committed.Revision)
806 })
807}
808
809// TestSnapshotAssemblerScopeFencing proves every part must carry the exact
810// assigned epoch and connection, whether or not a transfer is in flight.
811func TestSnapshotAssemblerScopeFencing(t *testing.T) {
812 base := snapshotPartsFor(5, 3, sampleSnapshotDaemons(), nil)
813 cases := []struct {
814 name string
815 apply func(*SnapshotPart)
816 }{
817 {"epoch", func(p *SnapshotPart) { p.Epoch = 8 }},
818 {"epoch zero", func(p *SnapshotPart) { p.Epoch = 0 }},
819 {"connection", func(p *SnapshotPart) { p.Connection = testConnectionID(0x99) }},
820 {"connection zero", func(p *SnapshotPart) { p.Connection = ports.BrokerConnectionID{} }},
821 }
822 for _, tc := range cases {
823 t.Run(tc.name, func(t *testing.T) {
824 a := testSnapshotAssembler(t)
825 part := base[0]
826 tc.apply(&part)
827 _, _, err := a.Add(part)
828 require.ErrorIs(t, err, ErrScopeMismatch)
829 require.False(t, a.StagingActive())
830
831 // With staging in flight the mismatched part is still refused and
832 // staging is retained.
833 stagePart(t, a, base[0])
834 part = base[1]
835 tc.apply(&part)
836 _, _, err = a.Add(part)
837 require.ErrorIs(t, err, ErrScopeMismatch)
838 require.True(t, a.StagingActive())
839 })
840 }
841}
842
843// TestSnapshotAssemblerZeroRevision proves a zero revision is refused.
844func TestSnapshotAssemblerZeroRevision(t *testing.T) {
845 a := testSnapshotAssembler(t)
846 part := snapshotPartsFor(5, 3, sampleSnapshotDaemons(), nil)[0]
847 part.Revision = 0
848 _, _, err := a.Add(part)
849 require.ErrorIs(t, err, ErrSnapshotInvalid)
850 require.False(t, a.StagingActive())
851}
852
853// TestSnapshotAssemblerNilAndEmptySafety proves the nil-receiver and
854// no-transfer paths.
855func TestSnapshotAssemblerNilAndEmptySafety(t *testing.T) {
856 var a *SnapshotAssembler
857 _, ok := a.Snapshot()
858 require.False(t, ok)
859 require.False(t, a.StagingActive())
860 a.DiscardStaging()
861 _, _, err := a.Add(snapshotPartsFor(5, 3, nil, nil)[0])
862 require.ErrorIs(t, err, ErrSnapshotInvalid)
863
864 fresh := testSnapshotAssembler(t)
865 _, ok = fresh.Snapshot()
866 require.False(t, ok)
867 extra := SnapshotPart{Epoch: 7, Connection: testConnectionID(0x21), Generation: 5, Revision: 3, Index: 1, Part: SnapshotEnd{}}
868 _, _, err = fresh.Add(extra)
869 require.ErrorIs(t, err, ErrSnapshotInvalid)
870 _, _, err = fresh.Add(SnapshotPart{Epoch: 7, Connection: testConnectionID(0x21), Generation: 5, Revision: 3, Index: 0, Part: nil})
871 require.ErrorIs(t, err, ErrSnapshotInvalid)
872}
873
874// TestSnapshotAssemblerPinnedGeneration proves WithGeneration fences every
875// part from the first one.
876func TestSnapshotAssemblerPinnedGeneration(t *testing.T) {
877 a := testSnapshotAssembler(t, WithGeneration(5))
878 stale := snapshotPartsFor(4, 1, sampleSnapshotDaemons(), nil)[0]
879 _, _, err := a.Add(stale)
880 require.ErrorIs(t, err, ErrStaleGeneration)
881 future := snapshotPartsFor(6, 1, sampleSnapshotDaemons(), nil)[0]
882 _, _, err = a.Add(future)
883 require.ErrorIs(t, err, ErrFutureGeneration)
884 require.NoError(t, runSnapshotParts(a, snapshotPartsFor(5, 1, sampleSnapshotDaemons(), nil)))
885 committed, ok := a.Snapshot()
886 require.True(t, ok)
887 require.Equal(t, ports.BrokerRevision(1), committed.Revision)
888}
889
890// TestSnapshotAssemblerAdoptCommitted proves a committed snapshot can be
891// handed to a fresh assembler of the same scope, so a generation advance
892// keeps publishing it until a newer transfer commits.
893func TestSnapshotAssemblerAdoptCommitted(t *testing.T) {
894 source := testSnapshotAssembler(t)
895 require.NoError(t, runSnapshotParts(source, snapshotPartsFor(5, 3, sampleSnapshotDaemons(), sampleSnapshotTombstones())))
896 committed, ok := source.Snapshot()
897 require.True(t, ok)
898
899 target := testSnapshotAssembler(t, WithGeneration(6))
900 require.False(t, target.StagingActive())
901 _, ok = target.Snapshot()
902 require.False(t, ok)
903 target.adoptCommitted(committed)
904 adopted, ok := target.Snapshot()
905 require.True(t, ok)
906 require.Equal(t, ports.BrokerRevision(3), adopted.Revision)
907 require.Len(t, adopted.Removed, 1)
908
909 // An older revision and a foreign epoch are ignored.
910 target.adoptCommitted(ports.BrokerSnapshot{Epoch: committed.Epoch, Revision: 2, Daemons: committed.Daemons})
911 target.adoptCommitted(ports.BrokerSnapshot{Epoch: committed.Epoch + 1, Revision: 9, Daemons: committed.Daemons})
912 adopted, ok = target.Snapshot()
913 require.True(t, ok)
914 require.Equal(t, ports.BrokerRevision(3), adopted.Revision)
915
916 // A newer transfer commits over the adopted snapshot.
917 require.NoError(t, runSnapshotParts(target, snapshotPartsFor(6, 4, sampleSnapshotDaemons(), nil)))
918 adopted, ok = target.Snapshot()
919 require.True(t, ok)
920 require.Equal(t, ports.BrokerRevision(4), adopted.Revision)
921 require.Empty(t, adopted.Removed)
922
923 // The adopted value is independent of the source.
924 adopted.Daemons[0].Endpoint = "mutated@nowhere:22"
925 again, ok := source.Snapshot()
926 require.True(t, ok)
927 require.Equal(t, "user0@host0:22", again.Daemons[0].Endpoint)
928}
929
930// TestSnapshotAssemblerConcurrentReaders is a deterministic race test: one
931// writer publishes sequential revisions while readers observe the committed
932// snapshot and staging state.
933func TestSnapshotAssemblerConcurrentReaders(t *testing.T) {
934 a := testSnapshotAssembler(t)
935 const revisions = 40
936 const readers = 4
937 var wg sync.WaitGroup
938 done := make(chan struct{})
939 for i := 0; i < readers; i++ {
940 wg.Add(1)
941 go func() {
942 defer wg.Done()
943 for {
944 select {
945 case <-done:
946 return
947 default:
948 }
949 if snapshot, ok := a.Snapshot(); ok {
950 assert.NotZero(t, snapshot.Revision)
951 assert.NotZero(t, snapshot.Epoch)
952 }
953 _ = a.StagingActive()
954 }
955 }()
956 }
957 for revision := 1; revision <= revisions; revision++ {
958 parts := snapshotPartsFor(5, ports.BrokerRevision(revision), sampleSnapshotDaemons(), nil)
959 for _, part := range parts {
960 _, _, err := a.Add(part)
961 require.NoError(t, err)
962 }
963 }
964 close(done)
965 wg.Wait()
966 snapshot, ok := a.Snapshot()
967 require.True(t, ok)
968 require.Equal(t, ports.BrokerRevision(revisions), snapshot.Revision)
969 require.False(t, a.StagingActive())
970}
971
972// assemblerFrames concatenates 4-byte big-endian length-prefixed frames, so one
973// fuzz input can drive a whole part sequence.
974func assemblerFrames(frames ...[]byte) []byte {
975 out := make([]byte, 0)
976 for _, frame := range frames {
977 var header [4]byte
978 binary.BigEndian.PutUint32(header[:], uint32(len(frame)))
979 out = append(out, header[:]...)
980 out = append(out, frame...)
981 }
982 return out
983}
984
985// fuzzSnapshotAssemblerSeeds is the starting corpus: one complete valid
986// transfer, a valid staged prefix, empty input, and corrupt variants (a flipped
987// byte in one frame and a frame whose length prefix overruns the payload).
988func fuzzSnapshotAssemblerSeeds(tb testing.TB) [][]byte {
989 tb.Helper()
990 encode := func(part SnapshotPart) []byte {
991 raw, err := EncodeServer(part, testEnvelopeCeiling, testChunkCeiling)
992 require.NoError(tb, err)
993 return raw
994 }
995 parts := snapshotPartsFor(4, 2, sampleSnapshotDaemons(), sampleSnapshotTombstones())
996 frames := make([][]byte, 0, len(parts))
997 for _, part := range parts {
998 frames = append(frames, encode(part))
999 }
1000 corrupt := append([]byte(nil), frames[1]...)
1001 corrupt[len(corrupt)-1] ^= 0xFF
1002 truncated := append([]byte(nil), frames[2][:len(frames[2])-1]...)
1003 return [][]byte{
1004 assemblerFrames(frames...),
1005 assemblerFrames(frames[:len(frames)-1]...),
1006 assemblerFrames(frames[0], corrupt),
1007 assemblerFrames(frames[0], truncated),
1008 nil,
1009 }
1010}
1011
1012// FuzzSnapshotAssembler proves arbitrary part sequences never panic the
1013// assembler, never publish an invalid snapshot, and never retain an invalid
1014// committed snapshot. Input is a sequence of 4-byte big-endian length-prefixed
1015// frames, each a serialized SnapshotPart envelope; frames that fail strict
1016// scanning or semantic conversion are dropped, and every accepted part is
1017// staged in order.
1018func FuzzSnapshotAssembler(f *testing.F) {
1019 for _, seed := range fuzzSnapshotAssemblerSeeds(f) {
1020 f.Add(seed)
1021 }
1022 f.Fuzz(func(t *testing.T, payload []byte) {
1023 assembler := testSnapshotAssembler(t, WithMaxStagedBytes(1<<20))
1024 for len(payload) >= 4 {
1025 size := int(binary.BigEndian.Uint32(payload[:4]))
1026 payload = payload[4:]
1027 if size <= 0 || size > len(payload) {
1028 return
1029 }
1030 frame := payload[:size]
1031 payload = payload[size:]
1032 message, err := DecodeServer(frame, testEnvelopeCeiling, testChunkCeiling)
1033 if err != nil {
1034 continue
1035 }
1036 part, ok := message.(SnapshotPart)
1037 if !ok {
1038 continue
1039 }
1040 completed, snapshot, err := assembler.Add(part)
1041 if err != nil {
1042 continue
1043 }
1044 if completed {
1045 requireValidBrokerSnapshot(t, snapshot)
1046 }
1047 }
1048 if snapshot, ok := assembler.Snapshot(); ok {
1049 requireValidBrokerSnapshot(t, snapshot)
1050 }
1051 })
1052}
1053
1054// requireValidBrokerSnapshot asserts one published snapshot is exactly what the
1055// assembler promises: a validated snapshot with durable remote daemon
1056// projections (the local daemon is process-local and never durable).
1057func requireValidBrokerSnapshot(t *testing.T, snapshot ports.BrokerSnapshot) {
1058 t.Helper()
1059 require.NoError(t, snapshot.Validate())
1060 for _, daemon := range snapshot.Daemons {
1061 if daemon.Local {
1062 continue
1063 }
1064 require.NoError(t, ports.ValidateDurableHostProjection(daemon))
1065 }
1066}