1package brokerwire
2
3// Broker preamble.
4//
5// The first client frame on a broker connection is always PreambleRequest;
6// the first server frame is always PreambleResponse. The broker reuses the
7// session preamble shapes with broker-specific roles: 3 names the broker
8// client, 4 names the broker server. Magic must equal wire.PreambleMagic,
9// epoch must equal wire.ProtocolEpoch, and version must exactly equal
10// protocol.Version. Capability bits must be zero: the broker negotiates no
11// optional capabilities here.
12//
13// The preamble negotiates immutable per-connection ceilings once: the
14// advertised envelope ceiling (1 MiB..16 MiB) rides in
15// max_receive_envelope_bytes, and the stream chunk ceiling (1..64 KiB)
16// rides in output_data_limit. Exact version equality is mandatory.
17//
18// Refusals carry the precise PreambleRejectionCode taxonomy: 1 bad magic,
19// 2 epoch mismatch, 3 version mismatch, 4 wrong role, 5 duplicate
20// preamble, 7 limit refused. Code 6 is reserved for the connection
21// dispatcher: out-of-order application frames are a framing-order failure
22// that this stateless codec never observes. Nonzero capability bits have no
23// dedicated code and are refused as code 7 (limit refused), since
24// the broker negotiates no optional capabilities.
25
26import (
27 "errors"
28
29 "github.com/bnema/vev/internal/protocol"
30 "github.com/bnema/vev/internal/protocol/wire"
31)
32
33// ErrPreambleRejected reports a broker preamble that fails validation:
34// bad magic, epoch, version, role, capability bits, or limit
35// advertisement. Callers map it to a precise rejection code with
36// RejectionCodeFor.
37var ErrPreambleRejected = errors.New("brokerwire: preamble rejected")
38
39// Preamble rejection codes. They mirror the session preamble refusal
40// taxonomy carried on wire.PreambleRejectionCode.
41const (
42 // RejectionBadMagic reports a preamble magic mismatch.
43 RejectionBadMagic uint32 = 1
44 // RejectionEpochMismatch reports a protocol epoch mismatch.
45 RejectionEpochMismatch uint32 = 2
46 // RejectionVersionMismatch reports a session protocol version mismatch.
47 RejectionVersionMismatch uint32 = 3
48 // RejectionWrongRole reports a preamble role mismatch.
49 RejectionWrongRole uint32 = 4
50 // RejectionDuplicate reports a second preamble on one connection.
51 RejectionDuplicate uint32 = 5
52 // RejectionOutOfOrder is reserved for the connection dispatcher, which
53 // owns framing-order detection. The stateless codec never emits it.
54 RejectionOutOfOrder uint32 = 6
55 // RejectionLimitRefused reports an unnegotiable limit advertisement. It
56 // also covers nonzero capability bits: the protocol negotiates no optional
57 // capabilities, so any capability bit is an unnegotiable advertisement.
58 RejectionLimitRefused uint32 = 7
59)
60
61// BrokerPreambleRequest is the validated broker client preamble: the
62// negotiated ceilings plus the raw wire message for stateless re-encode.
63type BrokerPreambleRequest struct {
64 Ceilings Ceilings
65}
66
67// BrokerPreambleResponse is the validated broker server preamble.
68type BrokerPreambleResponse struct {
69 Ceilings Ceilings
70 Accepted bool
71 Code uint32
72}
73
74// EncodePreambleRequest builds the local broker client preamble offer.
75func EncodePreambleRequest(ceilings brokerCeilings) *wire.PreambleRequest {
76 return &wire.PreambleRequest{
77 Magic: wire.PreambleMagic,
78 Epoch: wire.ProtocolEpoch,
79 Version: uint32(protocol.Version),
80 Role: &wire.PreambleRole{Role: BrokerRoleClient},
81 MaxReceiveEnvelopeBytes: ceilings.MaxReceiveEnvelopeBytes,
82 OutputDataLimit: ceilings.StreamChunkLimit,
83 }
84}
85
86// EncodePreambleResponse builds the broker server preamble answer:
87// acceptance with the effective ceilings, or a typed refusal carrying the
88// precise rejection code.
89func EncodePreambleResponse(accepted bool, ceilings brokerCeilings, code uint32) *wire.PreambleResponse {
90 response := &wire.PreambleResponse{
91 Magic: wire.PreambleMagic,
92 Epoch: wire.ProtocolEpoch,
93 Version: uint32(protocol.Version),
94 Role: &wire.PreambleRole{Role: BrokerRoleServer},
95 MaxReceiveEnvelopeBytes: ceilings.MaxReceiveEnvelopeBytes,
96 OutputDataLimit: ceilings.StreamChunkLimit,
97 Accepted: accepted,
98 }
99 if !accepted {
100 response.Rejection = &wire.PreambleRejectionCode{Code: code}
101 }
102 return response
103}
104
105// DecodePreambleRequest validates one raw broker client preamble. Magic,
106// epoch, version, role, and zero capability bits are checked before the
107// limit advertisement; the refusal code is precise.
108func DecodePreambleRequest(message *wire.PreambleRequest) (BrokerPreambleRequest, error) {
109 if message == nil {
110 return BrokerPreambleRequest{}, ErrPreambleRejected
111 }
112 if message.GetMagic() != wire.PreambleMagic {
113 return BrokerPreambleRequest{}, ErrPreambleRejected
114 }
115 if message.GetEpoch() != wire.ProtocolEpoch {
116 return BrokerPreambleRequest{}, ErrPreambleRejected
117 }
118 if message.GetVersion() != uint32(protocol.Version) {
119 return BrokerPreambleRequest{}, ErrPreambleRejected
120 }
121 if message.GetRole().GetRole() != BrokerRoleClient {
122 return BrokerPreambleRequest{}, ErrPreambleRejected
123 }
124 if message.GetCapabilityBits() != 0 {
125 return BrokerPreambleRequest{}, ErrPreambleRejected
126 }
127 remote := brokerCeilings{
128 MaxReceiveEnvelopeBytes: message.GetMaxReceiveEnvelopeBytes(),
129 StreamChunkLimit: message.GetOutputDataLimit(),
130 }
131 if err := checkBrokerCeilings(remote); err != nil {
132 return BrokerPreambleRequest{}, err
133 }
134 return BrokerPreambleRequest{Ceilings: remote}, nil
135}
136
137// DecodePreambleResponse validates one raw broker server preamble.
138// Refusals fail with ErrPreambleRejected; acceptance additionally
139// validates the server role and its limit advertisement.
140func DecodePreambleResponse(message *wire.PreambleResponse) (BrokerPreambleResponse, error) {
141 if message == nil {
142 return BrokerPreambleResponse{}, ErrPreambleRejected
143 }
144 if message.GetMagic() != wire.PreambleMagic {
145 return BrokerPreambleResponse{}, ErrPreambleRejected
146 }
147 if message.GetEpoch() != wire.ProtocolEpoch {
148 return BrokerPreambleResponse{}, ErrPreambleRejected
149 }
150 if message.GetVersion() != uint32(protocol.Version) {
151 return BrokerPreambleResponse{}, ErrPreambleRejected
152 }
153 if message.GetRole().GetRole() != BrokerRoleServer {
154 return BrokerPreambleResponse{}, ErrPreambleRejected
155 }
156 if message.GetCapabilityBits() != 0 {
157 return BrokerPreambleResponse{}, ErrPreambleRejected
158 }
159 if !message.GetAccepted() {
160 return BrokerPreambleResponse{Code: message.GetRejection().GetCode()}, ErrPreambleRejected
161 }
162 remote := brokerCeilings{
163 MaxReceiveEnvelopeBytes: message.GetMaxReceiveEnvelopeBytes(),
164 StreamChunkLimit: message.GetOutputDataLimit(),
165 }
166 if err := checkBrokerCeilings(remote); err != nil {
167 return BrokerPreambleResponse{}, err
168 }
169 return BrokerPreambleResponse{Ceilings: remote, Accepted: true}, nil
170}
171
172// RejectionCodeFor maps a preamble validation failure to its precise
173// refusal code. Limit, magic, epoch, version, and role failures each carry
174// their own code; nonzero capability bits are refused as
175// RejectionLimitRefused because the protocol negotiates no capabilities; unknown
176// scan failures are limit refusals. Code 6 is never returned here: the
177// dispatcher owns out-of-order framing.
178func RejectionCodeFor(request *wire.PreambleRequest, err error) uint32 {
179 if err == nil {
180 return 0
181 }
182 if request != nil {
183 if request.GetMagic() != wire.PreambleMagic {
184 return RejectionBadMagic
185 }
186 if request.GetEpoch() != wire.ProtocolEpoch {
187 return RejectionEpochMismatch
188 }
189 if request.GetVersion() != uint32(protocol.Version) {
190 return RejectionVersionMismatch
191 }
192 if request.GetRole().GetRole() != BrokerRoleClient {
193 return RejectionWrongRole
194 }
195 if request.GetCapabilityBits() != 0 {
196 // No capability is negotiable here, so any bit is a refused
197 // limit advertisement (code 7), documented explicitly.
198 return RejectionLimitRefused
199 }
200 remote := brokerCeilings{
201 MaxReceiveEnvelopeBytes: request.GetMaxReceiveEnvelopeBytes(),
202 StreamChunkLimit: request.GetOutputDataLimit(),
203 }
204 if checkBrokerCeilings(remote) != nil {
205 return RejectionLimitRefused
206 }
207 }
208 if errors.Is(err, wire.ErrScanDuplicate) {
209 return RejectionDuplicate
210 }
211 return RejectionLimitRefused
212}