1package brokeripc
2
3import (
4 "context"
5 "errors"
6 "io"
7
8 "google.golang.org/protobuf/proto"
9
10 "github.com/bnema/vev/internal/adapters/brokerwire"
11 "github.com/bnema/vev/internal/protocol/wire"
12)
13
14// Broker preamble exchange.
15//
16// The first client frame on a broker connection is always the broker
17// PreambleRequest (role 3) and the first server frame is always the broker
18// PreambleResponse (role 4). Both are built, validated, and refusal-coded by
19// brokerwire: this file only moves one bounded preamble envelope in each
20// direction over the carriage and returns the negotiated effective ceilings.
21//
22// The preamble envelope is bounded to brokerwire.BrokerPreambleLimit (4 KiB)
23// before allocation, strict-scanned with wire.ScanEnvelope before generated
24// unmarshal, and is never restarted. The caller's context (the configured
25// setup/handshake budget) bounds the whole exchange: a context alone cannot
26// interrupt the blocking carriage I/O, so the client half closes the carriage
27// on expiry to release the worker and joins it before returning.
28
29// runClientPreamble sends the local broker PreambleRequest offer and accepts
30// exactly one PreambleResponse. It returns the effective negotiated ceilings.
31// On ctx expiry it closes the carriage, which interrupts the blocking send or
32// receive, and joins its worker before returning, so a peer that stalls
33// mid-preamble never leaves a setup goroutine parked on the carriage.
34func runClientPreamble(ctx context.Context, transport wire.Transport, offer brokerwire.Ceilings) (brokerwire.Ceilings, error) {
35 if err := offer.Validate(); err != nil {
36 return brokerwire.Ceilings{}, errors.Join(ErrConfig, err)
37 }
38 type result struct {
39 ceilings brokerwire.Ceilings
40 err error
41 }
42 done := make(chan result, 1)
43 go func() {
44 request := brokerwire.EncodePreambleRequest(offer)
45 raw, err := proto.Marshal(request)
46 if err != nil {
47 done <- result{err: err}
48 return
49 }
50 if err := transport.Send(wire.Envelope{Payload: raw}); err != nil {
51 done <- result{err: transportFailure(err)}
52 return
53 }
54 payload, err := recvPreamble(transport)
55 if err != nil {
56 done <- result{err: err}
57 return
58 }
59 var response wire.PreambleResponse
60 if err := wire.ScanEnvelope(&response, payload); err != nil {
61 done <- result{err: errors.Join(ErrMalformedFrame, err)}
62 return
63 }
64 if err := (proto.UnmarshalOptions{DiscardUnknown: false}).Unmarshal(payload, &response); err != nil {
65 done <- result{err: errors.Join(ErrMalformedFrame, err)}
66 return
67 }
68 accepted, err := brokerwire.DecodePreambleResponse(&response)
69 if err != nil {
70 done <- result{err: err}
71 return
72 }
73 done <- result{ceilings: brokerwire.EffectiveCeilings(offer, accepted.Ceilings)}
74 }()
75 select {
76 case <-ctx.Done():
77 // A context alone cannot reach a Send or RecvBounded parked on a silent
78 // peer; closing the carriage releases the worker, and draining done
79 // joins it before returning. No setup goroutine outlives the exchange.
80 _ = transport.Close()
81 <-done
82 return brokerwire.Ceilings{}, ctx.Err()
83 case outcome := <-done:
84 return outcome.ceilings, outcome.err
85 }
86}
87
88// runServerPreamble accepts exactly one broker PreambleRequest, answers it with
89// acceptance or the precise brokerwire refusal code, and returns the effective
90// negotiated ceilings. A refused request is answered before the carriage is
91// closed by the caller, so the client sees the typed refusal rather than a
92// bare disconnect.
93func runServerPreamble(ctx context.Context, transport wire.Transport, offer brokerwire.Ceilings) (brokerwire.Ceilings, error) {
94 if err := offer.Validate(); err != nil {
95 return brokerwire.Ceilings{}, errors.Join(ErrConfig, err)
96 }
97 type result struct {
98 ceilings brokerwire.Ceilings
99 err error
100 }
101 done := make(chan result, 1)
102 go func() {
103 payload, err := recvPreamble(transport)
104 if err != nil {
105 done <- result{err: err}
106 return
107 }
108 var request wire.PreambleRequest
109 if err := wire.ScanEnvelope(&request, payload); err != nil {
110 _ = sendPreambleResponse(transport, brokerwire.EncodePreambleResponse(false, offer, brokerwire.RejectionCodeFor(nil, err)))
111 done <- result{err: errors.Join(ErrMalformedFrame, err)}
112 return
113 }
114 if err := (proto.UnmarshalOptions{DiscardUnknown: false}).Unmarshal(payload, &request); err != nil {
115 _ = sendPreambleResponse(transport, brokerwire.EncodePreambleResponse(false, offer, brokerwire.RejectionCodeFor(nil, err)))
116 done <- result{err: errors.Join(ErrMalformedFrame, err)}
117 return
118 }
119 accepted, err := brokerwire.DecodePreambleRequest(&request)
120 if err != nil {
121 _ = sendPreambleResponse(transport, brokerwire.EncodePreambleResponse(false, offer, brokerwire.RejectionCodeFor(&request, err)))
122 done <- result{err: err}
123 return
124 }
125 effective := brokerwire.EffectiveCeilings(offer, accepted.Ceilings)
126 if err := sendPreambleResponse(transport, brokerwire.EncodePreambleResponse(true, effective, 0)); err != nil {
127 done <- result{err: transportFailure(err)}
128 return
129 }
130 done <- result{ceilings: effective}
131 }()
132 select {
133 case <-ctx.Done():
134 return brokerwire.Ceilings{}, ctx.Err()
135 case outcome := <-done:
136 return outcome.ceilings, outcome.err
137 }
138}
139
140// recvPreamble reads exactly one length-delimited preamble envelope, enforcing
141// the 4 KiB broker preamble bound against the length prefix before any body
142// allocation and before any scan or unmarshal.
143func recvPreamble(transport wire.Transport) ([]byte, error) {
144 bounded, ok := transport.(wire.BoundedTransport)
145 if !ok {
146 return nil, errors.Join(ErrConfig, errors.New("brokeripc: carriage is not bounded"))
147 }
148 envelope, err := bounded.RecvBounded(brokerwire.BrokerPreambleLimit)
149 if err != nil {
150 if errors.Is(err, io.EOF) {
151 return nil, transportFailure(err)
152 }
153 if errors.Is(err, wire.ErrScanLength) {
154 return nil, errors.Join(ErrMalformedFrame, err)
155 }
156 return nil, transportFailure(err)
157 }
158 if err := brokerwire.CheckPreambleSize(envelope.Payload); err != nil {
159 return nil, errors.Join(ErrMalformedFrame, err)
160 }
161 return envelope.Payload, nil
162}
163
164// sendPreambleResponse writes one already-validated broker preamble response.
165func sendPreambleResponse(transport wire.Transport, response *wire.PreambleResponse) error {
166 raw, err := proto.Marshal(response)
167 if err != nil {
168 return err
169 }
170 if err := brokerwire.CheckPreambleSize(raw); err != nil {
171 return err
172 }
173 return transport.Send(wire.Envelope{Payload: raw})
174}