1package brokeripc
2
3import (
4 "context"
5 "errors"
6 "io"
7 "sync"
8
9 "github.com/bnema/vev/internal/adapters/brokerwire"
10 "github.com/bnema/vev/internal/adapters/sessionwire"
11 "github.com/bnema/vev/internal/ports"
12 "github.com/bnema/vev/internal/protocol/wire"
13)
14
15// Logical stream bridging.
16//
17// One brokerwire logical stream carries the session protocol between client
18// and daemon. The broker relays complete opaque envelope payloads through
19// bounded framed carriages; neither preamble nor session message is decoded.
20//
21// Identity and cancellation stay per stream: a stream the core loses, a stream
22// whose local consumer stalled, and a stream the peer closed each settle alone,
23// sending at most one typed StreamClosed and releasing only that stream's
24// resources. The connection and every sibling stream stay up.
25
26// serverStream is the broker-side half of one logical stream.
27type serverStream struct {
28 session *serverSession
29 id ports.BrokerStreamID
30 core ports.BrokerEnvelopeStream
31 pipe *streamPipe
32 carriage *carriage
33
34 mu sync.Mutex
35 closed bool
36 done chan struct{}
37}
38
39// newServerStream builds one bridged stream over the admitted core connection.
40func newServerStream(s *serverSession, id ports.BrokerStreamID, core ports.BrokerEnvelopeStream) (*serverStream, error) {
41 if s == nil || core == nil || id == 0 {
42 return nil, ErrConfig
43 }
44 st := &serverStream{session: s, id: id, core: core, done: make(chan struct{})}
45 st.pipe = newStreamPipe(s.cfg.StreamInboundChunks, s.cfg.StreamInboundBytes, st.sendChunk)
46 st.carriage = newCarriage(st.pipe)
47 return st, nil
48}
49
50// sendChunk splits one outbound byte slice under the negotiated stream chunk
51// ceiling and writes each frame in order.
52func (st *serverStream) sendChunk(data []byte) error {
53 ceiling := int(st.session.ceilings.StreamChunkLimit)
54 if ceiling <= 0 {
55 return ErrConfig
56 }
57 for len(data) > 0 {
58 n := min(len(data), ceiling)
59 if err := st.session.send(brokerwire.ServerStreamData{
60 Epoch: st.session.epoch, Connection: st.session.scope.Connection, Stream: st.id, Data: data[:n],
61 }); err != nil {
62 return err
63 }
64 data = data[n:]
65 }
66 return nil
67}
68
69// deliver accepts one inbound chunk from the connection reader.
70func (st *serverStream) deliver(chunk []byte) error { return st.pipe.deliver(chunk) }
71
72// start launches the three per-stream workers. Every one is joined by the
73// session's WaitGroup before shutdown returns.
74func (st *serverStream) start() {
75 st.session.wg.Add(3)
76 go st.relayClient()
77 go st.relayServer()
78 go st.watchCore()
79}
80
81// relayClient carries complete client envelopes without decoding.
82func (st *serverStream) relayClient() {
83 defer st.session.wg.Done()
84 for {
85 envelope, err := st.carriage.RecvBounded(wire.AbsoluteEnvelopeLimit)
86 if err != nil {
87 st.fail(cancelOrCause(st.session.ctx, streamFailure(err)))
88 return
89 }
90 if err := st.core.SendEnvelope(envelope.Payload); err != nil {
91 st.fail(coreFailure(st.core, err))
92 return
93 }
94 }
95}
96
97// relayServer carries complete daemon envelopes without decoding.
98func (st *serverStream) relayServer() {
99 defer st.session.wg.Done()
100 for {
101 payload, err := st.core.RecvEnvelope()
102 if err != nil {
103 st.fail(coreFailure(st.core, err))
104 return
105 }
106 if len(payload) == 0 || len(payload) > wire.AbsoluteEnvelopeLimit {
107 st.fail(ErrProtocol)
108 return
109 }
110 if err := st.carriage.Send(wire.Envelope{Payload: payload}); err != nil {
111 st.fail(streamFailure(err))
112 return
113 }
114 }
115}
116
117// watchCore publishes the core connection's terminal outcome even when no read
118// is in flight. It settles the stream itself only for a failure: an orderly
119// core end (Err()==nil) is left to relayServer, whose own core.RecvEnvelope()
120// call observes the same terminal outcome right after forwarding any reply
121// that was already in flight, and settles the stream from there. Settling here
122// on every Done would race relayServer's in-flight carriage.Send and could
123// close the carriage while a received reply is still being written out to the
124// client, discarding it.
125func (st *serverStream) watchCore() {
126 defer st.session.wg.Done()
127 select {
128 case <-st.core.Done():
129 if err := st.core.Err(); err != nil {
130 st.fail(coreFailure(st.core, nil))
131 }
132 case <-st.done:
133 }
134}
135
136// fail settles the stream once and reports it to the session, which retires its
137// wire identity and sends at most one typed StreamClosed.
138func (st *serverStream) fail(err error) {
139 if st.close(err) {
140 st.session.settleStream(st, err)
141 }
142}
143
144// close releases the stream exactly once: the pipe unblocks a stalled reader,
145// the carriage joins its writer, and the core connection closes.
146func (st *serverStream) close(cause error) bool {
147 if st == nil {
148 return false
149 }
150 st.mu.Lock()
151 if st.closed {
152 st.mu.Unlock()
153 return false
154 }
155 st.closed = true
156 st.mu.Unlock()
157 st.pipe.closeWith(cause)
158 _ = st.carriage.Close()
159 _ = st.core.Close()
160 close(st.done)
161 return true
162}
163
164// streamFailure classifies one wire-side stream error: an orderly carriage end
165// is a peer close, not a transport failure.
166func streamFailure(err error) error {
167 if err == nil || errors.Is(err, io.EOF) {
168 return ErrStreamGone
169 }
170 return err
171}
172
173// coreFailure classifies one core connection outcome. The core's typed terminal
174// error wins; an orderly end is a peer close.
175func coreFailure(core ports.BrokerEnvelopeStream, err error) error {
176 if err == nil && core != nil {
177 err = core.Err()
178 }
179 if err == nil {
180 return ErrStreamGone
181 }
182 var failure ports.BrokerError
183 if errors.As(err, &failure) {
184 return err
185 }
186 if errors.Is(err, io.EOF) || errors.Is(err, context.Canceled) {
187 return ErrStreamGone
188 }
189 return err
190}
191
192// orderlyStream reports whether one stream outcome closes the stream without a
193// failure detail.
194func orderlyStream(err error) bool {
195 return err == nil || errors.Is(err, ErrStreamGone)
196}
197
198// startStream admits one inbound open on the wire tracker and starts its bridge
199// once the broker core established the stream.
200func (s *serverSession) startStream(m brokerwire.OpenStream) {
201 if err := s.conn.OpenStream(m.Stream); err != nil {
202 if sendErr := s.refuseStream(m.Stream, admissionError(err)); sendErr != nil {
203 s.abort(sendErr)
204 }
205 return
206 }
207 request := ports.BrokerOpenStreamRequest{
208 Epoch: m.Epoch, Purpose: m.Purpose, Admission: m.Admission, Name: m.Name,
209 Local: m.Local,
210 Connection: m.Connection, Stream: m.Stream,
211 Endpoint: m.Endpoint, Registration: m.Registration,
212 Target: m.Target, Env: m.Env, Policy: m.Policy,
213 StartMode: m.StartMode,
214 }
215 s.wg.Add(1)
216 go func() {
217 defer s.wg.Done()
218 core, err := s.core.OpenEnvelopeStream(s.ctx, request)
219 if err != nil {
220 _, _ = s.conn.CloseStream(m.Stream)
221 if sendErr := s.refuseStream(m.Stream, err); sendErr != nil {
222 s.abort(sendErr)
223 }
224 return
225 }
226 if core == nil {
227 _, _ = s.conn.CloseStream(m.Stream)
228 if sendErr := s.refuseStream(m.Stream, errors.Join(ports.BrokerAdmissionInvalid, errors.New("brokeripc: core returned a nil logical connection"))); sendErr != nil {
229 s.abort(sendErr)
230 }
231 return
232 }
233 stream, err := newServerStream(s, m.Stream, core)
234 if err != nil {
235 _ = core.Close()
236 _, _ = s.conn.CloseStream(m.Stream)
237 if sendErr := s.refuseStream(m.Stream, err); sendErr != nil {
238 s.abort(sendErr)
239 }
240 return
241 }
242 if !s.registerStream(stream) {
243 stream.close(ErrSessionClosed)
244 return
245 }
246 if err := s.conn.StreamOpened(m.Stream); err != nil {
247 s.settleStream(stream, errors.Join(ErrProtocol, err))
248 return
249 }
250 if err := s.send(brokerwire.StreamOpened{Epoch: s.epoch, Connection: s.scope.Connection, Stream: m.Stream}); err != nil {
251 s.settleStream(stream, transportFailure(err))
252 return
253 }
254 stream.start()
255 }()
256}
257
258// registerStream installs one bridge, or reports false when the session is
259// closing.
260func (s *serverSession) registerStream(st *serverStream) bool {
261 s.streamsMu.Lock()
262 defer s.streamsMu.Unlock()
263 if s.ctx.Err() != nil {
264 return false
265 }
266 s.streams[st.id] = st
267 return true
268}
269
270// unregisterStream removes one bridge, if present.
271func (s *serverSession) unregisterStream(id ports.BrokerStreamID) {
272 s.streamsMu.Lock()
273 delete(s.streams, id)
274 s.streamsMu.Unlock()
275}
276
277// lookupStream returns the bridge for one stream identity.
278func (s *serverSession) lookupStream(id ports.BrokerStreamID) *serverStream {
279 s.streamsMu.Lock()
280 defer s.streamsMu.Unlock()
281 return s.streams[id]
282}
283
284// settleStream retires one stream's wire identity and reports its outcome to the
285// peer. A session that is already shutting down sends nothing: the carriage is
286// closing and the peer is gone.
287func (s *serverSession) settleStream(st *serverStream, err error) {
288 if st == nil {
289 return
290 }
291 st.close(err)
292 s.unregisterStream(st.id)
293 _, _ = s.conn.CloseStream(st.id)
294 // Retire the broker-side stream reservation as well, so a stream that
295 // settled for any reason is released at the core and not only in its
296 // carriage.
297 _ = s.core.CloseStream(s.scope.Connection, st.id)
298 if s.ctx.Err() != nil {
299 return
300 }
301 detail := brokerwire.ErrorDetail{}
302 if !orderlyStream(err) {
303 detail = errorDetail(err)
304 }
305 if sendErr := s.send(brokerwire.StreamClosed{
306 Epoch: s.epoch, Connection: s.scope.Connection, Stream: st.id,
307 Error: detail, HasError: detail != (brokerwire.ErrorDetail{}),
308 }); sendErr != nil {
309 s.abort(sendErr)
310 }
311}
312
313// closeStreams settles every bridged stream during shutdown, releasing each
314// core connection and session carriage exactly once.
315func (s *serverSession) closeStreams() {
316 s.streamsMu.Lock()
317 streams := make([]*serverStream, 0, len(s.streams))
318 for _, st := range s.streams {
319 streams = append(streams, st)
320 }
321 s.streams = make(map[ports.BrokerStreamID]*serverStream)
322 s.streamsMu.Unlock()
323 for _, st := range streams {
324 st.close(ErrSessionClosed)
325 }
326}
327
328// closeStreamByPeer retires one stream the peer closed. No StreamClosed is sent
329// back: the peer already knows.
330func (s *serverSession) closeStreamByPeer(id ports.BrokerStreamID) {
331 st := s.lookupStream(id)
332 if _, err := s.conn.CloseStream(id); err != nil && !errors.Is(err, brokerwire.ErrFutureStream) {
333 // A close for a stream that was never allocated is a peer protocol
334 // violation; a late close for a retired stream is simply discarded.
335 if errors.Is(err, brokerwire.ErrConnectionState) || errors.Is(err, brokerwire.ErrScopeMismatch) {
336 s.abort(errors.Join(ErrProtocol, err))
337 return
338 }
339 }
340 if st == nil {
341 return
342 }
343 s.unregisterStream(id)
344 st.close(ErrStreamGone)
345 // Retire the broker-side stream reservation as well: the peer's close ends
346 // the client-scoped stream, not merely its carriage.
347 _ = s.core.CloseStream(s.scope.Connection, id)
348}
349
350// deliverStreamData routes one inbound client chunk to its bridge. Backpressure
351// settles only that stream; a chunk for an identity the peer never opened is a
352// protocol violation.
353func (s *serverSession) deliverStreamData(m brokerwire.ClientStreamData) error {
354 disposition, err := s.conn.StreamData(m.Stream)
355 if err != nil {
356 // Stream data for a stream that was never allocated, or before it was
357 // confirmed open, is a peer protocol violation.
358 return errors.Join(ErrProtocol, err)
359 }
360 if disposition == brokerwire.StreamDiscarded {
361 return nil
362 }
363 st := s.lookupStream(m.Stream)
364 if st == nil {
365 // The bridge was settled between the tracker check and the lookup (a
366 // stream-local failure won the race); the chunk is discarded with it.
367 return nil
368 }
369 switch err := st.deliver(m.Data); {
370 case err == nil:
371 return nil
372 case errors.Is(err, ErrStreamBackpressure):
373 s.settleStream(st, ErrStreamBackpressure)
374 return nil
375 default:
376 // The stream already reached its terminal outcome; its data is
377 // discarded.
378 return nil
379 }
380}
381
382// clientStream is the client-side half of one logical stream: the sessionwire
383// client connection over the stream's bounded pipe, plus the terminal outcome a
384// ports.BrokerLogicalConnection exposes independently of reads.
385type clientStream struct {
386 ports.ClientConnection
387 client *client
388 id ports.BrokerStreamID
389 pipe *streamPipe
390 carriage *carriage
391
392 opened chan error
393 mu sync.Mutex
394 closed bool
395 err error
396 done chan struct{}
397}
398
399// newClientStream builds one client-side stream over the broker connection.
400func newClientStream(c *client, id ports.BrokerStreamID) (*clientStream, error) {
401 if c == nil || id == 0 {
402 return nil, ErrConfig
403 }
404 st := &clientStream{
405 client: c,
406 id: id,
407 opened: make(chan error, 1),
408 done: make(chan struct{}),
409 }
410 st.pipe = newStreamPipe(c.cfg.StreamInboundChunks, c.cfg.StreamInboundBytes, st.sendChunk)
411 st.carriage = newCarriage(st.pipe)
412 st.ClientConnection = sessionwire.NewClientConnection(st.carriage)
413 if st.ClientConnection == nil {
414 return nil, ErrConfig
415 }
416 return st, nil
417}
418
419// sendChunk splits one outbound client byte slice under the negotiated stream
420// chunk ceiling.
421func (st *clientStream) sendChunk(data []byte) error {
422 ceiling := int(st.client.ceilings.StreamChunkLimit)
423 if ceiling <= 0 {
424 return ErrConfig
425 }
426 for len(data) > 0 {
427 n := min(len(data), ceiling)
428 if err := st.client.send(brokerwire.ClientStreamData{
429 Epoch: st.client.scope.Epoch, Connection: st.client.scope.Connection, Stream: st.id, Data: data[:n],
430 }); err != nil {
431 return err
432 }
433 data = data[n:]
434 }
435 return nil
436}
437
438// deliver accepts one inbound server chunk.
439func (st *clientStream) deliver(chunk []byte) error { return st.pipe.deliver(chunk) }
440
441// Done closes when the stream reached its terminal outcome.
442func (st *clientStream) Done() <-chan struct{} { return st.done }
443
444// Err reports the terminal cause; nil is an orderly close.
445func (st *clientStream) Err() error {
446 st.mu.Lock()
447 defer st.mu.Unlock()
448 return st.err
449}
450
451// Close ends the stream: it sends one CloseStream frame best-effort, releases
452// the carriage, and resolves any pending open. It is idempotent and
453// concurrent-safe.
454func (st *clientStream) Close() error {
455 if st == nil {
456 return nil
457 }
458 st.client.requestStreamClose(st.id)
459 st.terminate(nil)
460 return nil
461}
462
463// terminate settles the stream exactly once, publishes its terminal cause (nil
464// for an orderly close), and resolves a pending open.
465func (st *clientStream) terminate(cause error) bool {
466 if st == nil {
467 return false
468 }
469 st.mu.Lock()
470 if st.closed {
471 st.mu.Unlock()
472 return false
473 }
474 st.closed = true
475 st.err = cause
476 st.mu.Unlock()
477 st.pipe.closeWith(cause)
478 _ = st.carriage.Close()
479 st.client.unregisterStream(st.id)
480 // Retire the identity on the connection's tracker as well, so a frame that
481 // arrives after local retirement is classified as discarded.
482 _, _ = st.client.conn.CloseStream(st.id)
483 select {
484 case st.opened <- cause:
485 default:
486 }
487 close(st.done)
488 return true
489}
490
491// fail settles the stream and tells the broker to retire it, so a stream-local
492// failure never leaves the peer half-open.
493func (st *clientStream) fail(err error) {
494 if st.terminate(err) {
495 st.client.requestStreamClose(st.id)
496 }
497}