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