1package brokeripc
2
3import (
4 "errors"
5
6 "github.com/bnema/vev/internal/adapters/brokerwire"
7 "github.com/bnema/vev/internal/ports"
8 "github.com/bnema/vev/internal/protocol/catalogue"
9)
10
11// Snapshot publication.
12//
13// A broker publication travels as the exact multipart transfer the brokerwire
14// assembler accepts: Begin(index 0, element counts), then every daemon part in
15// daemon-index order immediately followed by that daemon's session parts in
16// session-index order, then the tombstones, then End at the final index. The
17// local daemon, when present, is daemon index 0 and its sessions carry
18// Local=true. Daemon parts carry no inline sessions; sessions travel as their
19// own parts. Both sides agree on one deterministic layout, so the receiving
20// assembler commits only a transfer whose indexes, counts, and order are exact.
21//
22// The emitter refuses a transfer the peer's assembler would refuse: more parts
23// than MaxSnapshotParts, or a total encoded payload above the staged ceiling.
24// The staged ceiling check is deliberately conservative - the encoded total is
25// an upper bound of the deterministic staged estimate - so a publication this
26// side accepts is never one the peer silently aborts.
27
28// snapshotParts splits one published snapshot into its ordered multipart
29// transfer for one connection scope and subscription generation.
30func snapshotParts(snapshot ports.BrokerSnapshot, epoch ports.BrokerEpoch, connection ports.BrokerConnectionID, generation brokerwire.SubscriptionGeneration) ([]brokerwire.SnapshotPart, error) {
31 if epoch == 0 || connection.IsZero() {
32 return nil, ErrScopeMismatch
33 }
34 if snapshot.Revision == 0 {
35 return nil, errors.New("brokeripc: snapshot publication has no revision")
36 }
37 if len(snapshot.Daemons) > ports.BrokerMaxDaemonsPerSnapshot || len(snapshot.Removed) > ports.BrokerMaxTombstones {
38 return nil, errors.New("brokeripc: snapshot publication exceeds daemon bound")
39 }
40 sessions := 0
41 for _, daemon := range snapshot.Daemons {
42 if len(daemon.Sessions) > ports.BrokerMaxSessionsPerHost {
43 return nil, errors.New("brokeripc: snapshot publication exceeds session bound")
44 }
45 sessions += len(daemon.Sessions)
46 }
47 total := 1 + len(snapshot.Daemons) + sessions + len(snapshot.Removed) + 1
48 if total > brokerwire.MaxSnapshotParts {
49 return nil, errors.New("brokeripc: snapshot publication exceeds part bound")
50 }
51 localPresent := len(snapshot.Daemons) > 0 && snapshot.Daemons[0].Local
52
53 parts := make([]brokerwire.SnapshotPart, 0, total)
54 base := func(part brokerwire.SnapshotPartPayload) brokerwire.SnapshotPart {
55 return brokerwire.SnapshotPart{
56 Epoch: epoch, Connection: connection, Generation: generation,
57 Revision: snapshot.Revision, Index: uint32(len(parts)), Part: part,
58 }
59 }
60 parts = append(parts, base(brokerwire.SnapshotBegin{
61 HostCount: uint32(len(snapshot.Daemons)),
62 SessionCount: uint32(sessions),
63 TombstoneCount: uint32(len(snapshot.Removed)),
64 LocalPresent: localPresent,
65 }))
66 for daemonIndex, daemon := range snapshot.Daemons {
67 projection := daemon.Clone()
68 sessionCount := len(projection.Sessions)
69 projection.Sessions = nil
70 parts = append(parts, base(brokerwire.SnapshotDaemonPart{
71 HostIndex: uint32(daemonIndex),
72 Daemon: projection,
73 SessionCount: uint32(sessionCount),
74 }))
75 for sessionIndex, session := range daemon.Sessions {
76 parts = append(parts, base(brokerwire.SnapshotSessionPart{
77 HostIndex: uint32(daemonIndex),
78 SessionIndex: uint32(sessionIndex),
79 Local: daemon.Local,
80 Session: cloneCatalogSession(session),
81 }))
82 }
83 }
84 for tombstoneIndex, tombstone := range snapshot.Removed {
85 parts = append(parts, base(brokerwire.SnapshotTombstonePart{
86 TombstoneIndex: uint32(tombstoneIndex),
87 Registration: tombstone.Registration,
88 RetiredRevision: tombstone.RetiredRevision,
89 }))
90 }
91 parts = append(parts, base(brokerwire.SnapshotEnd{}))
92 return parts, nil
93}
94
95// cloneCatalogSession copies one session and guarantees its tabs slice is
96// non-nil: the assembler refuses a session part whose tabs are absent, so an
97// empty inventory publishes an empty list rather than a missing one.
98func cloneCatalogSession(session catalogue.RemoteCatalogSession) catalogue.RemoteCatalogSession {
99 out := session
100 if session.Tabs == nil {
101 out.Tabs = []catalogue.RemoteCatalogTab{}
102 return out
103 }
104 tabs := make([]catalogue.RemoteCatalogTab, len(session.Tabs))
105 copy(tabs, session.Tabs)
106 out.Tabs = tabs
107 return out
108}
109
110// encodeSnapshotTransfer encodes every part of one prepared transfer once and
111// sums the encoded size, so the publisher can refuse a publication above the
112// staged ceiling before the peer aborts it and then send the same bytes.
113func encodeSnapshotTransfer(parts []brokerwire.SnapshotPart, maxEnvelopeBytes, maxChunkBytes uint64) ([][]byte, uint64, error) {
114 payloads := make([][]byte, 0, len(parts))
115 var total uint64
116 for _, part := range parts {
117 payload, err := brokerwire.EncodeServer(part, maxEnvelopeBytes, maxChunkBytes)
118 if err != nil {
119 return nil, 0, err
120 }
121 payloads = append(payloads, payload)
122 total += uint64(len(payload))
123 }
124 return payloads, total, nil
125}