1// Stateful multipart snapshot assembly.
2//
3// SnapshotAssembler stages one multipart SnapshotPart publication and
4// publishes it atomically only after full validation. It builds on the
5// stateless codec: parts are already strict-scanned and semantically
6// converted, so this file owns only transfer state: exact indexes, counts,
7// order, staged bounds, revision fencing, generation fencing, and atomic
8// publication.
9//
10// Transfer layout (indexes are exact positions):
11//
12// Begin(0) Daemon/Session... Tombstone... End(N-1)
13//
14// Begin carries the total daemon, session, and tombstone counts plus whether
15// the local daemon is present. Daemon parts arrive in host_index order, with
16// the local daemon (when present) first at index 0 and remotes following in
17// registration order; each Daemon advertises how many Session parts follow it
18// immediately, with per-daemon session_index order. A session of the local
19// daemon carries Local=true and host_index 0. Tombstones arrive in
20// tombstone_index order after all daemons and sessions. End closes the
21// transfer. Total parts are 1 + daemons + sessions + tombstones + 1, bounded
22// by MaxSnapshotParts (65 daemons, 65*256 sessions, 64 tombstones).
23//
24// Bounds: at most MaxSnapshotParts parts and MaxSnapshotStagedBytes staged
25// estimate per transfer. The staged estimate is a deterministic function of
26// endpoint, origin, policy, identity, session, tab, and tombstone text lengths
27// plus fixed per-part overhead; it bounds hostile memory without measuring
28// wire bytes.
29//
30// Publication is atomic: the committed snapshot is replaced only when End
31// arrives with exact indexes and the assembled BrokerSnapshot passes
32// ports.BrokerSnapshot.Validate plus, for every remote daemon,
33// ports.ValidateDurableHostProjection. Any malformed index, count, order, or
34// validation failure aborts staging and retains the old snapshot. A new Begin
35// for the same generation with a newer revision coalesces by replacing
36// staging; an older revision is rejected and retains both the committed
37// snapshot and any staging in flight. Equal revisions are allowed so a resync
38// may republish the committed revision.
39//
40// Generation fencing: the assembler belongs to exactly one nonzero
41// subscription generation, adopted from the first Begin or pinned by
42// WithGeneration. Parts with a stale generation are ignored without touching
43// staging; parts with a future generation are rejected, whether or not a
44// transfer is in flight. Scope fencing is exact: every part must carry the
45// assembler's epoch and connection.
46//
47// There is no I/O, socket, pump, or transport here. The owning
48// Connection discards staging on generation advance, unsubscribe, or close.
49package brokerwire
50
51import (
52 "errors"
53 "sync"
54
55 "github.com/bnema/vev/internal/ports"
56 "github.com/bnema/vev/internal/protocol/catalogue"
57)
58
59const (
60 // MaxSnapshotParts bounds one multipart publication: 1 Begin + 65 daemons
61 // + 65*256 sessions + 64 tombstones + 1 End.
62 MaxSnapshotParts = 1 + ports.BrokerMaxDaemonsPerSnapshot + ports.BrokerMaxDaemonsPerSnapshot*ports.BrokerMaxSessionsPerHost + ports.BrokerMaxTombstones + 1
63 // MaxSnapshotStagedBytes bounds the deterministic staged estimate of
64 // one transfer (80 MiB).
65 MaxSnapshotStagedBytes = uint64(80 << 20)
66)
67
68var (
69 // ErrSnapshotInvalid reports a malformed transfer: wrong index, count,
70 // order, payload type, or assembled validation failure. Staging is
71 // aborted and the committed snapshot is retained.
72 ErrSnapshotInvalid = errors.New("brokerwire: invalid snapshot transfer")
73 // ErrSnapshotStale reports an older revision: below the committed
74 // revision or below staging. Nothing changes.
75 ErrSnapshotStale = errors.New("brokerwire: stale snapshot revision")
76 // ErrStaleGeneration reports a part bound to a superseded
77 // subscription. It is ignored without touching staging.
78 ErrStaleGeneration = errors.New("brokerwire: stale subscription generation")
79 // ErrFutureGeneration reports a part bound to a generation past the
80 // active one. It is rejected.
81 ErrFutureGeneration = errors.New("brokerwire: future subscription generation")
82 // ErrInvalidGeneration reports a zero subscription generation.
83 ErrInvalidGeneration = errors.New("brokerwire: invalid subscription generation")
84 // ErrScopeMismatch reports a part whose epoch or connection does not
85 // exactly match the assigned scope. It is rejected.
86 ErrScopeMismatch = errors.New("brokerwire: scope mismatch")
87)
88
89// SnapshotOption tunes a SnapshotAssembler. Tests use a small staged cap
90// to prove the bound without allocating 80 MiB; production uses defaults.
91type SnapshotOption func(*SnapshotAssembler)
92
93// WithGeneration pins the subscription generation this assembler belongs to,
94// so a part of any other generation is fenced from the first one. The zero
95// default adopts the generation from the first Begin.
96func WithGeneration(generation uint64) SnapshotOption {
97 return func(a *SnapshotAssembler) {
98 a.generation = generation
99 }
100}
101
102// SnapshotAssembler stages one multipart publication per generation and
103// holds the last committed snapshot. It is safe for concurrent use.
104type SnapshotAssembler struct {
105 mu sync.Mutex
106 epoch ports.BrokerEpoch
107 connection ports.BrokerConnectionID
108 generation SubscriptionGeneration
109 committed ports.BrokerSnapshot
110 hasCommitted bool
111 staging *snapshotStaging
112 maxStagedBytes uint64
113}
114
115type snapshotStaging struct {
116 generation SubscriptionGeneration
117 revision ports.BrokerRevision
118 hostCount uint32
119 sessionTotal uint32
120 tombstoneCount uint32
121 localPresent bool
122 totalParts uint32
123 nextIndex uint32
124 daemons []ports.BrokerDaemonObservation
125 daemonExpected []uint32
126 daemonSeen []uint32
127 daemonsSeen uint32
128 sessionsSeen uint32
129 tombstones []ports.BrokerHostTombstone
130 tombstonesSeen uint32
131 stagedBytes uint64
132}
133
134// newSnapshotStaging allocates one transfer's staging from a validated Begin.
135func newSnapshotStaging(generation SubscriptionGeneration, revision ports.BrokerRevision, begin SnapshotBegin) *snapshotStaging {
136 total := 1 + uint64(begin.HostCount) + uint64(begin.SessionCount) + uint64(begin.TombstoneCount) + 1
137 return &snapshotStaging{
138 generation: generation,
139 revision: revision,
140 hostCount: begin.HostCount,
141 sessionTotal: begin.SessionCount,
142 tombstoneCount: begin.TombstoneCount,
143 localPresent: begin.LocalPresent,
144 totalParts: uint32(total),
145 nextIndex: 1,
146 daemons: make([]ports.BrokerDaemonObservation, begin.HostCount),
147 daemonExpected: make([]uint32, begin.HostCount),
148 daemonSeen: make([]uint32, begin.HostCount),
149 tombstones: make([]ports.BrokerHostTombstone, begin.TombstoneCount),
150 }
151}
152
153// NewSnapshotAssembler binds an assembler to one exact scope assigned by
154// Registered. Every part must carry that scope.
155func NewSnapshotAssembler(epoch ports.BrokerEpoch, connection ports.BrokerConnectionID, opts ...SnapshotOption) *SnapshotAssembler {
156 a := &SnapshotAssembler{epoch: epoch, connection: connection, maxStagedBytes: MaxSnapshotStagedBytes}
157 for _, opt := range opts {
158 if opt != nil {
159 opt(a)
160 }
161 }
162 if a.maxStagedBytes == 0 {
163 a.maxStagedBytes = MaxSnapshotStagedBytes
164 }
165 return a
166}
167
168// Snapshot returns the last committed snapshot and whether one exists.
169// The returned value is a defensive clone.
170func (a *SnapshotAssembler) Snapshot() (ports.BrokerSnapshot, bool) {
171 if a == nil {
172 return ports.BrokerSnapshot{}, false
173 }
174 a.mu.Lock()
175 defer a.mu.Unlock()
176 if !a.hasCommitted {
177 return ports.BrokerSnapshot{}, false
178 }
179 return a.committed.Clone(), true
180}
181
182// AdoptCommitted seeds the assembler with a snapshot already committed for
183// the same scope, so a subscription generation advance keeps publishing the
184// last committed revision until a newer transfer commits. An older revision
185// or a foreign epoch is ignored, and no transfer in flight is disturbed.
186func (a *SnapshotAssembler) adoptCommitted(snapshot ports.BrokerSnapshot) {
187 if a == nil || snapshot.Epoch != a.epoch || snapshot.Revision == 0 {
188 return
189 }
190 a.mu.Lock()
191 defer a.mu.Unlock()
192 if a.hasCommitted && snapshot.Revision < a.committed.Revision {
193 return
194 }
195 a.committed = snapshot.Clone()
196 a.hasCommitted = true
197}
198
199// StagingActive reports whether a transfer is in flight.
200func (a *SnapshotAssembler) StagingActive() bool {
201 if a == nil {
202 return false
203 }
204 a.mu.Lock()
205 defer a.mu.Unlock()
206 return a.staging != nil
207}
208
209// DiscardStaging abandons any transfer in flight and retains the committed
210// snapshot.
211func (a *SnapshotAssembler) DiscardStaging() {
212 if a == nil {
213 return
214 }
215 a.mu.Lock()
216 defer a.mu.Unlock()
217 a.staging = nil
218}
219
220// Add stages one SnapshotPart. It returns completed=true with the newly
221// committed clone only when End validates and publishes. Stale generations
222// are ignored, future generations rejected, older revisions rejected, and
223// malformed transfers aborted; all retain the committed snapshot.
224func (a *SnapshotAssembler) Add(part SnapshotPart) (bool, ports.BrokerSnapshot, error) {
225 if a == nil {
226 return false, ports.BrokerSnapshot{}, ErrSnapshotInvalid
227 }
228 a.mu.Lock()
229 defer a.mu.Unlock()
230 if part.Epoch != a.epoch || part.Connection != a.connection {
231 return false, ports.BrokerSnapshot{}, ErrScopeMismatch
232 }
233 if part.Generation == 0 {
234 return false, ports.BrokerSnapshot{}, ErrInvalidGeneration
235 }
236 // Generation fencing holds for the assembler's whole lifetime: the
237 // first Begin adopts the generation, and every later part must carry
238 // it even when no transfer is staged.
239 if a.generation != 0 && part.Generation != a.generation {
240 if part.Generation < a.generation {
241 return false, ports.BrokerSnapshot{}, ErrStaleGeneration
242 }
243 return false, ports.BrokerSnapshot{}, ErrFutureGeneration
244 }
245 if part.Revision == 0 {
246 return false, ports.BrokerSnapshot{}, ErrSnapshotInvalid
247 }
248 if a.staging == nil {
249 begin, ok := normalizeBegin(part.Part)
250 if !ok {
251 return false, ports.BrokerSnapshot{}, ErrSnapshotInvalid
252 }
253 if part.Index != 0 {
254 return false, ports.BrokerSnapshot{}, ErrSnapshotInvalid
255 }
256 if err := checkBeginCounts(begin); err != nil {
257 return false, ports.BrokerSnapshot{}, err
258 }
259 if a.hasCommitted && part.Revision < a.committed.Revision {
260 return false, ports.BrokerSnapshot{}, ErrSnapshotStale
261 }
262 staging := newSnapshotStaging(part.Generation, part.Revision, begin)
263 staged := estimatePartBytes(part)
264 if staged > a.maxStagedBytes {
265 return false, ports.BrokerSnapshot{}, ErrTooLarge
266 }
267 staging.stagedBytes = staged
268 a.generation = part.Generation
269 a.staging = staging
270 return false, ports.BrokerSnapshot{}, nil
271 }
272 staging := a.staging
273 if part.Generation != staging.generation {
274 if part.Generation < staging.generation {
275 return false, ports.BrokerSnapshot{}, ErrStaleGeneration
276 }
277 return false, ports.BrokerSnapshot{}, ErrFutureGeneration
278 }
279 if begin, ok := normalizeBegin(part.Part); ok {
280 if part.Index != 0 {
281 a.staging = nil
282 return false, ports.BrokerSnapshot{}, ErrSnapshotInvalid
283 }
284 if err := checkBeginCounts(begin); err != nil {
285 a.staging = nil
286 return false, ports.BrokerSnapshot{}, err
287 }
288 switch {
289 case part.Revision == staging.revision:
290 // Restart of the same transfer: discard progress.
291 case part.Revision > staging.revision:
292 if a.hasCommitted && part.Revision < a.committed.Revision {
293 return false, ports.BrokerSnapshot{}, ErrSnapshotStale
294 }
295 default:
296 return false, ports.BrokerSnapshot{}, ErrSnapshotStale
297 }
298 fresh := newSnapshotStaging(part.Generation, part.Revision, begin)
299 staged := estimatePartBytes(part)
300 if staged > a.maxStagedBytes {
301 a.staging = nil
302 return false, ports.BrokerSnapshot{}, ErrTooLarge
303 }
304 fresh.stagedBytes = staged
305 a.staging = fresh
306 return false, ports.BrokerSnapshot{}, nil
307 }
308 if part.Revision != staging.revision {
309 if part.Revision < staging.revision {
310 return false, ports.BrokerSnapshot{}, ErrSnapshotStale
311 }
312 a.staging = nil
313 return false, ports.BrokerSnapshot{}, ErrSnapshotInvalid
314 }
315 if part.Index != staging.nextIndex {
316 a.staging = nil
317 return false, ports.BrokerSnapshot{}, ErrSnapshotInvalid
318 }
319 if uint64(part.Index) >= uint64(staging.totalParts) {
320 a.staging = nil
321 return false, ports.BrokerSnapshot{}, ErrSnapshotInvalid
322 }
323 staged := staging.stagedBytes + estimatePartBytes(part)
324 if staged > a.maxStagedBytes {
325 a.staging = nil
326 return false, ports.BrokerSnapshot{}, ErrTooLarge
327 }
328 switch payload := part.Part.(type) {
329 case SnapshotDaemonPart, *SnapshotDaemonPart:
330 daemonPart := normalizeDaemonPart(payload)
331 if err := a.stageDaemon(staging, daemonPart); err != nil {
332 a.staging = nil
333 return false, ports.BrokerSnapshot{}, err
334 }
335 case SnapshotSessionPart, *SnapshotSessionPart:
336 sessionPart := normalizeSessionPart(payload)
337 if err := a.stageSession(staging, sessionPart); err != nil {
338 a.staging = nil
339 return false, ports.BrokerSnapshot{}, err
340 }
341 case SnapshotTombstonePart, *SnapshotTombstonePart:
342 tombPart := normalizeTombstonePart(payload)
343 if err := a.stageTombstone(staging, tombPart); err != nil {
344 a.staging = nil
345 return false, ports.BrokerSnapshot{}, err
346 }
347 case SnapshotEnd, *SnapshotEnd:
348 snapshot, err := a.commitStaging(staging)
349 if err != nil {
350 a.staging = nil
351 return false, ports.BrokerSnapshot{}, err
352 }
353 staging.stagedBytes = staged
354 a.committed = snapshot
355 a.hasCommitted = true
356 a.staging = nil
357 return true, snapshot.Clone(), nil
358 default:
359 a.staging = nil
360 return false, ports.BrokerSnapshot{}, ErrSnapshotInvalid
361 }
362 staging.stagedBytes = staged
363 staging.nextIndex++
364 return false, ports.BrokerSnapshot{}, nil
365}
366
367func normalizeBegin(payload SnapshotPartPayload) (SnapshotBegin, bool) {
368 switch p := payload.(type) {
369 case SnapshotBegin:
370 return p, true
371 case *SnapshotBegin:
372 if p == nil {
373 return SnapshotBegin{}, false
374 }
375 return *p, true
376 default:
377 return SnapshotBegin{}, false
378 }
379}
380
381func normalizeDaemonPart(payload SnapshotPartPayload) SnapshotDaemonPart {
382 switch p := payload.(type) {
383 case SnapshotDaemonPart:
384 return p
385 case *SnapshotDaemonPart:
386 if p == nil {
387 return SnapshotDaemonPart{HostIndex: ^uint32(0)}
388 }
389 return *p
390 default:
391 return SnapshotDaemonPart{HostIndex: ^uint32(0)}
392 }
393}
394
395func normalizeSessionPart(payload SnapshotPartPayload) SnapshotSessionPart {
396 switch p := payload.(type) {
397 case SnapshotSessionPart:
398 return p
399 case *SnapshotSessionPart:
400 if p == nil {
401 return SnapshotSessionPart{HostIndex: ^uint32(0)}
402 }
403 return *p
404 default:
405 return SnapshotSessionPart{HostIndex: ^uint32(0)}
406 }
407}
408
409func normalizeTombstonePart(payload SnapshotPartPayload) SnapshotTombstonePart {
410 switch p := payload.(type) {
411 case SnapshotTombstonePart:
412 return p
413 case *SnapshotTombstonePart:
414 if p == nil {
415 return SnapshotTombstonePart{TombstoneIndex: ^uint32(0)}
416 }
417 return *p
418 default:
419 return SnapshotTombstonePart{TombstoneIndex: ^uint32(0)}
420 }
421}
422
423func checkBeginCounts(begin SnapshotBegin) error {
424 if begin.HostCount > uint32(ports.BrokerMaxDaemonsPerSnapshot) ||
425 begin.SessionCount > uint32(ports.BrokerMaxDaemonsPerSnapshot*ports.BrokerMaxSessionsPerHost) ||
426 begin.TombstoneCount > uint32(ports.BrokerMaxTombstones) {
427 return ErrTooLarge
428 }
429 // A local daemon occupies daemon index 0, so it can only be present when
430 // at least one daemon part follows.
431 if begin.LocalPresent && begin.HostCount == 0 {
432 return ErrSnapshotInvalid
433 }
434 return nil
435}
436
437func (a *SnapshotAssembler) stageDaemon(staging *snapshotStaging, part SnapshotDaemonPart) error {
438 if staging.daemonsSeen >= staging.hostCount {
439 return ErrSnapshotInvalid
440 }
441 if part.HostIndex != staging.daemonsSeen {
442 return ErrSnapshotInvalid
443 }
444 if part.HostIndex >= staging.hostCount {
445 return ErrSnapshotInvalid
446 }
447 if part.SessionCount > uint32(ports.BrokerMaxSessionsPerHost) {
448 return ErrTooLarge
449 }
450 if len(part.Daemon.Sessions) != 0 {
451 return ErrSnapshotInvalid
452 }
453 // The local daemon is always first: daemon index 0 is local exactly when
454 // Begin advertised it, and no later daemon may be local.
455 if part.HostIndex == 0 {
456 if part.Daemon.Local != staging.localPresent {
457 return ErrSnapshotInvalid
458 }
459 } else if part.Daemon.Local {
460 return ErrSnapshotInvalid
461 }
462 var seen uint64
463 for _, n := range staging.daemonExpected[:staging.daemonsSeen] {
464 seen += uint64(n)
465 }
466 if seen+uint64(part.SessionCount) > uint64(staging.sessionTotal) {
467 return ErrSnapshotInvalid
468 }
469 // Remaining daemons must still be able to fill the advertised total:
470 // even if every remaining daemon carried the per-daemon maximum, the
471 // total must stay reachable. This is checked exactly at End; here we
472 // only refuse an already-exceeded sum.
473 staging.daemons[part.HostIndex] = part.Daemon.Clone()
474 staging.daemonExpected[part.HostIndex] = part.SessionCount
475 staging.daemonSeen[part.HostIndex] = 0
476 staging.daemonsSeen++
477 return nil
478}
479
480func (a *SnapshotAssembler) stageSession(staging *snapshotStaging, part SnapshotSessionPart) error {
481 if staging.daemonsSeen == 0 || staging.daemonsSeen > staging.hostCount {
482 return ErrSnapshotInvalid
483 }
484 current := staging.daemonsSeen - 1
485 if part.HostIndex != current {
486 return ErrSnapshotInvalid
487 }
488 if current >= staging.hostCount {
489 return ErrSnapshotInvalid
490 }
491 expected := staging.daemonExpected[current]
492 if staging.daemonSeen[current] >= expected {
493 return ErrSnapshotInvalid
494 }
495 if part.SessionIndex != staging.daemonSeen[current] {
496 return ErrSnapshotInvalid
497 }
498 // A session belongs to the daemon it follows: a local session binds to
499 // the local daemon at index 0, and a remote session to a remote daemon.
500 if part.Local != staging.daemons[current].Local {
501 return ErrSnapshotInvalid
502 }
503 if part.HostIndex >= uint32(ports.BrokerMaxDaemonsPerSnapshot) || part.SessionIndex >= uint32(ports.BrokerMaxSessionsPerHost) {
504 return ErrTooLarge
505 }
506 if staging.sessionsSeen >= staging.sessionTotal {
507 return ErrSnapshotInvalid
508 }
509 session := part.Session
510 if session.Tabs == nil {
511 return ErrSnapshotInvalid
512 }
513 tabs := make([]catalogue.RemoteCatalogTab, len(session.Tabs))
514 copy(tabs, session.Tabs)
515 session.Tabs = tabs
516 daemon := staging.daemons[current]
517 daemon.Sessions = append(daemon.Sessions, session)
518 staging.daemons[current] = daemon
519 staging.daemonSeen[current]++
520 staging.sessionsSeen++
521 return nil
522}
523
524func (a *SnapshotAssembler) stageTombstone(staging *snapshotStaging, part SnapshotTombstonePart) error {
525 if staging.daemonsSeen != staging.hostCount || staging.sessionsSeen != staging.sessionTotal {
526 return ErrSnapshotInvalid
527 }
528 if staging.tombstonesSeen >= staging.tombstoneCount {
529 return ErrSnapshotInvalid
530 }
531 if part.TombstoneIndex != staging.tombstonesSeen {
532 return ErrSnapshotInvalid
533 }
534 if part.RetiredRevision == 0 {
535 return ErrSnapshotInvalid
536 }
537 staging.tombstones[part.TombstoneIndex] = ports.BrokerHostTombstone{
538 Endpoint: part.Registration.Endpoint,
539 Registration: part.Registration,
540 RetiredRevision: part.RetiredRevision,
541 }
542 staging.tombstonesSeen++
543 return nil
544}
545
546func (a *SnapshotAssembler) commitStaging(staging *snapshotStaging) (ports.BrokerSnapshot, error) {
547 if staging.daemonsSeen != staging.hostCount || staging.sessionsSeen != staging.sessionTotal || staging.tombstonesSeen != staging.tombstoneCount {
548 return ports.BrokerSnapshot{}, ErrSnapshotInvalid
549 }
550 if staging.nextIndex != staging.totalParts-1 {
551 return ports.BrokerSnapshot{}, ErrSnapshotInvalid
552 }
553 var sum uint64
554 for _, n := range staging.daemonExpected {
555 sum += uint64(n)
556 }
557 if sum != uint64(staging.sessionTotal) {
558 return ports.BrokerSnapshot{}, ErrSnapshotInvalid
559 }
560 daemons := make([]ports.BrokerDaemonObservation, len(staging.daemons))
561 for i, daemon := range staging.daemons {
562 daemons[i] = daemon.Clone()
563 if daemons[i].Sessions == nil {
564 daemons[i].Sessions = []catalogue.RemoteCatalogSession{}
565 }
566 }
567 removed := append([]ports.BrokerHostTombstone(nil), staging.tombstones...)
568 snapshot := ports.BrokerSnapshot{Epoch: a.epoch, Revision: staging.revision, Daemons: daemons, Removed: removed}
569 if err := snapshot.Validate(); err != nil {
570 return ports.BrokerSnapshot{}, ErrSnapshotInvalid
571 }
572 for _, daemon := range daemons {
573 // The local daemon is process-local state and is never durable.
574 if daemon.Local {
575 continue
576 }
577 if err := ports.ValidateDurableHostProjection(daemon); err != nil {
578 return ports.BrokerSnapshot{}, ErrSnapshotInvalid
579 }
580 }
581 return snapshot, nil
582}
583
584func estimatePartBytes(part SnapshotPart) uint64 {
585 const overhead = 32
586 switch payload := part.Part.(type) {
587 case SnapshotBegin, *SnapshotBegin:
588 return overhead
589 case SnapshotDaemonPart:
590 return estimateDaemonBytes(payload.Daemon) + overhead
591 case *SnapshotDaemonPart:
592 if payload == nil {
593 return overhead
594 }
595 return estimateDaemonBytes(payload.Daemon) + overhead
596 case SnapshotSessionPart:
597 return estimateCatalogSessionBytes(payload.Session) + overhead
598 case *SnapshotSessionPart:
599 if payload == nil {
600 return overhead
601 }
602 return estimateCatalogSessionBytes(payload.Session) + overhead
603 case SnapshotTombstonePart:
604 return uint64(len(payload.Registration.Endpoint)+32) + overhead
605 case *SnapshotTombstonePart:
606 if payload == nil {
607 return overhead
608 }
609 return uint64(len(payload.Registration.Endpoint)+32) + overhead
610 case SnapshotEnd, *SnapshotEnd:
611 return overhead
612 default:
613 return overhead
614 }
615}
616
617// estimateDaemonBytes bounds one daemon projection's staged memory: every
618// configured-authority and observed-identity string plus a fixed per-daemon
619// charge.
620func estimateDaemonBytes(daemon ports.BrokerDaemonObservation) uint64 {
621 text := len(daemon.Endpoint) + len(daemon.DisplayOrigin) + len(daemon.Registration.Endpoint) +
622 len(daemon.Policy.Transport) + len(daemon.Policy.Trust) + len(daemon.Policy.Launch) + len(daemon.Policy.Isolation) +
623 len(daemon.Identity)
624 return uint64(text + 64)
625}
626
627// estimateCatalogSessionBytes bounds one session's staged memory. It charges
628// every tab at least estimateTabBytes (covering the tab struct, its string
629// headers, and its slice slot) plus the variable-length text, and adds one
630// slice-header charge when the tab list is present. This deliberately
631// over-charges minimal tabs so a hostile max-tab publication cannot hide
632// behind tiny strings and slip under MaxSnapshotStagedBytes.
633const (
634 estimateTabBytes = 64
635 estimateTabSliceBytes = 24
636 estimateSessionBase = 32
637)
638
639func estimateCatalogSessionBytes(session catalogue.RemoteCatalogSession) uint64 {
640 n := len(session.Name) + len(session.Reason) + len(session.ActiveTabID) + estimateSessionBase
641 if session.Tabs != nil {
642 n += estimateTabSliceBytes
643 }
644 for _, tab := range session.Tabs {
645 n += len(tab.ID) + len(tab.Name) + len(tab.Detail) + estimateTabBytes
646 }
647 if n < 0 {
648 return 0
649 }
650 return uint64(n)
651}