1package brokeripc
2
3import (
4 "context"
5 "crypto/rand"
6 "errors"
7 "sync"
8
9 "github.com/bnema/vev/internal/adapters/brokerwire"
10 "github.com/bnema/vev/internal/adapters/ipc"
11 "github.com/bnema/vev/internal/adapters/streamframe"
12 "github.com/bnema/vev/internal/domain"
13 "github.com/bnema/vev/internal/ports"
14 "github.com/bnema/vev/internal/protocol/wire"
15)
16
17// Broker IPC client.
18//
19// Dial connects to one per-user broker endpoint through the private IPC
20// carriage (same-user kernel peer credentials on dial), completes the broker
21// preamble, sends exactly one Register, adopts the scope the listener assigned
22// in the Registered answer, and then runs one reader for the connection. The
23// returned value implements ports.BrokerService, so a client process reaches
24// local and remote daemons through the same façade it would use in-process.
25//
26// Identity is the connection's, not the caller's: OpenStream fills in an absent
27// epoch, connection, and stream identity from this connection and refuses a
28// caller-supplied identity that belongs elsewhere as stale. Stream identities
29// are strictly increasing and never reused. Mutating operations carry a fresh
30// operation identity, are bounded while pending, and report outcome-unknown
31// rather than a silent retry when a reply is lost after the request was sent.
32
33// Dial connects to the broker endpoint at path and returns a ports.BrokerService
34// bound to exactly one accepted connection.
35func Dial(ctx context.Context, path string, cfg Config) (ports.BrokerService, error) {
36 return dial(ctx, path, cfg, ipc.SameUserPeerVerifier())
37}
38
39// Connector dials the private broker IPC carriage. It holds only the endpoint
40// and its bounds, so every Connect is an independent attempt and the caller
41// owns Close on the returned service. Navigator adapts it to
42// ports.BrokerConnector for session frontends.
43type Connector struct {
44 path string
45 cfg Config
46}
47
48// Navigator returns c as a ports.BrokerConnector, which hands session
49// frontends only the navigation surface of each connection.
50func (c *Connector) Navigator() ports.BrokerConnector { return navigatorConnector{c} }
51
52type navigatorConnector struct{ c *Connector }
53
54func (n navigatorConnector) Connect(ctx context.Context) (ports.BrokerNavigator, error) {
55 service, err := n.c.Connect(ctx)
56 if err != nil {
57 return nil, err
58 }
59 return service, nil
60}
61
62// NewConnector returns a connector for one broker endpoint. An empty path is
63// refused by Connect, exactly as Dial refuses it, so a misconfigured connector
64// fails closed instead of dialing an unspecified socket.
65func NewConnector(path string, cfg Config) *Connector {
66 return &Connector{path: path, cfg: cfg}
67}
68
69// Connect dials the configured endpoint and returns the connection-scoped
70// service. Cancellation is honored by the underlying Dial, which bounds the
71// whole setup and closes the carriage when its context expires.
72func (c *Connector) Connect(ctx context.Context) (ports.BrokerService, error) {
73 if c == nil {
74 return nil, ErrConfig
75 }
76 return Dial(ctx, c.path, c.cfg)
77}
78
79// dial is Dial with an injected peer verifier, so a test can observe a
80// deterministic same-user refusal without root.
81func dial(ctx context.Context, path string, cfg Config, verify ipc.PeerVerifier) (ports.BrokerService, error) {
82 cfg = cfg.withDefaults()
83 if err := cfg.validate(); err != nil {
84 return nil, err
85 }
86 if path == "" {
87 return nil, ErrConfig
88 }
89 // One deadline bounds the whole setup: the Unix dial, the preamble
90 // exchange, the Register send, and the wait for Registered. Each setup step
91 // closes the carriage when that deadline expires, because a context alone
92 // cannot interrupt blocking framing I/O; that is what stops a peer that
93 // stalls after a successful preamble from parking Dial forever.
94 setup, cancelSetup := context.WithTimeout(ctx, cfg.HandshakeTimeout)
95 defer cancelSetup()
96 transport, err := ipc.DialMuxContextWithPeerVerifier(setup, path, verify)
97 if err != nil {
98 return nil, err
99 }
100 ceilings, err := runClientPreamble(setup, transport, brokerwire.DefaultCeilings())
101 if err != nil {
102 _ = transport.Close()
103 return nil, err
104 }
105 scope, staleBuild, err := register(setup, transport, ceilings, cfg.Build)
106 if err != nil {
107 _ = transport.Close()
108 return nil, err
109 }
110 // Setup is complete and the carriage is healthy, so the established
111 // connection detaches from the setup deadline: it gets its own
112 // connection-lived context below, and cancelSetup (deferred) only ends an
113 // already-finished setup.
114 conn, err := brokerwire.NewConnection(scope)
115 if err != nil {
116 _ = transport.Close()
117 return nil, err
118 }
119 if err := conn.SignalPreamble(); err != nil {
120 _ = transport.Close()
121 return nil, err
122 }
123 if err := conn.Register(); err != nil {
124 _ = transport.Close()
125 return nil, err
126 }
127 ctx, cancelSession := context.WithCancel(context.Background())
128 c := &client{
129 transport: transport,
130 ceilings: ceilings,
131 scope: scope,
132 staleBuild: staleBuild,
133 conn: conn,
134 cfg: cfg,
135 ctx: ctx,
136 cancel: cancelSession,
137 pending: make(map[ports.BrokerOperationID]chan operationResult),
138 kinds: make(map[ports.BrokerOperationID]brokerwire.RegisterMutationKind),
139 streams: make(map[ports.BrokerStreamID]*clientStream),
140 done: make(chan struct{}),
141 }
142 go c.run()
143 return c, nil
144}
145
146// register completes the broker registration exchange: one Register follows the
147// preamble, and the first server frame must be the Registered answer carrying
148// the scope assigned at accept. Any other first frame is a protocol violation.
149//
150// The wait is bounded by ctx: a peer that stalls after a successful preamble is
151// interrupted by closing the carriage, and the exchange worker is joined before
152// returning, so no setup goroutine is abandoned on the carriage. A Registered
153// observed after ctx expired is not a success.
154func register(ctx context.Context, transport wire.BoundedTransport, ceilings brokerwire.Ceilings, build string) (brokerwire.Scope, string, error) {
155 type result struct {
156 scope brokerwire.Scope
157 staleBuild string
158 err error
159 }
160 done := make(chan result, 1)
161 go func() {
162 scope, staleBuild, err := exchangeRegister(transport, ceilings, build)
163 done <- result{scope: scope, staleBuild: staleBuild, err: err}
164 }()
165 select {
166 case <-ctx.Done():
167 _ = transport.Close()
168 <-done
169 return brokerwire.Scope{}, "", ctx.Err()
170 case outcome := <-done:
171 if outcome.err == nil {
172 if err := ctx.Err(); err != nil {
173 return brokerwire.Scope{}, "", err
174 }
175 }
176 return outcome.scope, outcome.staleBuild, outcome.err
177 }
178}
179
180// exchangeRegister runs one Register/Registered round trip on the carriage. It
181// is synchronous: the caller bounds it by closing the carriage on ctx expiry.
182func exchangeRegister(transport wire.BoundedTransport, ceilings brokerwire.Ceilings, build string) (brokerwire.Scope, string, error) {
183 payload, err := brokerwire.EncodeClient(brokerwire.Register{Build: build}, ceilings.MaxReceiveEnvelopeBytes, ceilings.StreamChunkLimit)
184 if err != nil {
185 return brokerwire.Scope{}, "", err
186 }
187 if err := transport.Send(wire.Envelope{Payload: payload}); err != nil {
188 return brokerwire.Scope{}, "", transportFailure(err)
189 }
190 envelope, err := transport.RecvBounded(ceilings.MaxReceiveEnvelopeBytes)
191 if err != nil {
192 return brokerwire.Scope{}, "", transportFailure(err)
193 }
194 message, err := brokerwire.DecodeServer(envelope.Payload, ceilings.MaxReceiveEnvelopeBytes, ceilings.StreamChunkLimit)
195 if err != nil {
196 return brokerwire.Scope{}, "", errors.Join(ErrMalformedFrame, err)
197 }
198 registered, ok := message.(brokerwire.Registered)
199 if !ok {
200 return brokerwire.Scope{}, "", errors.Join(ErrProtocol, errors.New("brokeripc: broker did not answer Register with Registered"))
201 }
202 scope := brokerwire.Scope{Epoch: registered.Epoch, Connection: registered.Connection}
203 if err := scope.Validate(); err != nil {
204 return brokerwire.Scope{}, "", errors.Join(ErrProtocol, err)
205 }
206 if registered.Retiring {
207 _ = transport.Close()
208 return brokerwire.Scope{}, "", ErrBrokerRetired
209 }
210 if registered.Build != build {
211 return scope, registered.Build, nil
212 }
213 return scope, "", nil
214}
215
216// operationResult is one reply to a mutating operation.
217type operationResult struct {
218 outcome ports.BrokerMutationOutcome
219 removed bool
220 registration domain.RemoteRegistration
221 detail brokerwire.ErrorDetail
222}
223
224// client implements ports.BrokerService over one broker IPC connection.
225type client struct {
226 transport wire.BoundedTransport
227 ceilings brokerwire.Ceilings
228 scope brokerwire.Scope
229 staleBuild string
230 conn *brokerwire.Connection
231 cfg Config
232
233 ctx context.Context
234 cancel context.CancelFunc
235
236 mu sync.Mutex
237 closed bool
238 err error
239 sub *clientSubscription
240 generation brokerwire.SubscriptionGeneration
241 preview *clientPreviewSubscription
242 previewGeneration ports.BrokerPreviewGeneration
243 nextStream ports.BrokerStreamID
244 snapshot ports.BrokerSnapshot
245 pending map[ports.BrokerOperationID]chan operationResult
246 kinds map[ports.BrokerOperationID]brokerwire.RegisterMutationKind
247 // The connection's own stream tracker is the anti-replay authority for
248 // this connection: it admits every identity at most once inside the bounded
249 // window, whether the open is afterwards accepted or refused.
250 streams map[ports.BrokerStreamID]*clientStream
251 closeOnce sync.Once
252 done chan struct{}
253}
254
255var _ ports.BrokerService = (*client)(nil)
256
257// run is the connection's single reader.
258func (c *client) run() {
259 for {
260 envelope, err := c.transport.RecvBounded(c.ceilings.MaxReceiveEnvelopeBytes)
261 if err != nil {
262 c.closeWith(transportFailure(err))
263 return
264 }
265 message, err := brokerwire.DecodeServer(envelope.Payload, c.ceilings.MaxReceiveEnvelopeBytes, c.ceilings.StreamChunkLimit)
266 if err != nil {
267 c.closeWith(errors.Join(ErrMalformedFrame, err))
268 return
269 }
270 if err := c.dispatch(message); err != nil {
271 c.closeWith(err)
272 return
273 }
274 }
275}
276
277// dispatch applies one server frame. A returned error is a protocol violation
278// that settles the connection.
279func (c *client) dispatch(message brokerwire.ServerMessage) error {
280 switch m := message.(type) {
281 case brokerwire.SnapshotPart:
282 if !c.scopeMatches(m.Epoch, m.Connection) {
283 return errors.Join(ErrScopeMismatch, ErrProtocol)
284 }
285 committed, snapshot, err := c.conn.AddSnapshotPart(m)
286 if err != nil {
287 if errors.Is(err, brokerwire.ErrStaleGeneration) || errors.Is(err, brokerwire.ErrFutureGeneration) || errors.Is(err, brokerwire.ErrSnapshotStale) {
288 return nil
289 }
290 return errors.Join(ErrMalformedFrame, err)
291 }
292 if committed {
293 c.publish(snapshot)
294 }
295 return nil
296 case brokerwire.PreviewPublication:
297 if !c.scopeMatches(m.Epoch, m.Connection) {
298 return errors.Join(ErrScopeMismatch, ErrProtocol)
299 }
300 c.mu.Lock()
301 sub := c.preview
302 active := sub != nil && c.previewGeneration == m.Generation
303 c.mu.Unlock()
304 if !active {
305 return nil
306 }
307 publication := ports.BrokerPreviewPublication{Epoch: m.Epoch, Connection: m.Connection, Generation: m.Generation, Target: sub.request.Preview.Target, Width: sub.request.Preview.Width, Height: sub.request.Preview.Height, Preview: m.Preview}
308 if m.HasError {
309 publication.Err = failureFromDetail(m.Error)
310 }
311 if err := publication.Validate(); err != nil {
312 return errors.Join(ErrMalformedFrame, err)
313 }
314 sub.publish(publication)
315 return nil
316 case brokerwire.OperationResult:
317 if !c.scopeMatches(m.Epoch, m.Connection) {
318 return errors.Join(ErrScopeMismatch, ErrProtocol)
319 }
320 c.mu.Lock()
321 _, ok := c.pending[m.Operation]
322 kind := c.kinds[m.Operation]
323 c.mu.Unlock()
324 // An abandoned or replayed completion has no authority over current
325 // state and is dropped. A completion for a live operation must satisfy
326 // that request kind's payload invariant.
327 if !ok {
328 return nil
329 }
330 if err := m.ValidateForMutation(kind); err != nil {
331 c.completeOperation(m.Operation, operationResult{outcome: ports.BrokerOutcomeUnknown, detail: errorDetail(err)})
332 return errors.Join(ErrMalformedFrame, err)
333 }
334 c.completeOperation(m.Operation, operationResult{outcome: m.Outcome, removed: m.Removed, registration: m.Registration, detail: m.Error})
335 return nil
336 case brokerwire.StreamOpened:
337 if !c.scopeMatches(m.Epoch, m.Connection) {
338 return errors.Join(ErrScopeMismatch, ErrProtocol)
339 }
340 if err := c.conn.StreamOpened(m.Stream); err != nil && !errors.Is(err, brokerwire.ErrStreamState) {
341 return errors.Join(ErrProtocol, err)
342 }
343 if st := c.lookupStream(m.Stream); st != nil {
344 select {
345 case st.opened <- nil:
346 default:
347 }
348 }
349 return nil
350 case brokerwire.ServerStreamData:
351 if !c.scopeMatches(m.Epoch, m.Connection) {
352 return errors.Join(ErrScopeMismatch, ErrProtocol)
353 }
354 st, err := c.streamFor(m.Stream)
355 if st == nil || err != nil {
356 return err
357 }
358 if err := st.deliver(m.Data); errors.Is(err, ErrStreamCredit) {
359 st.fail(ErrStreamCredit)
360 }
361 return nil
362 case brokerwire.StreamWindowUpdate:
363 if !c.scopeMatches(m.Epoch, m.Connection) {
364 return errors.Join(ErrScopeMismatch, ErrProtocol)
365 }
366 st, err := c.streamFor(m.Stream)
367 if st == nil || err != nil {
368 return err
369 }
370 if err := st.pipe.granted(m.Credit); err != nil {
371 st.fail(err)
372 }
373 return nil
374 case brokerwire.StreamClosed:
375 if !c.scopeMatches(m.Epoch, m.Connection) {
376 return errors.Join(ErrScopeMismatch, ErrProtocol)
377 }
378 _, _ = c.conn.CloseStream(m.Stream)
379 if st := c.lookupStream(m.Stream); st != nil {
380 var cause error
381 if m.HasError {
382 cause = failureFromDetail(m.Error)
383 }
384 st.terminate(cause)
385 }
386 return nil
387 case brokerwire.Progress:
388 if !c.scopeMatches(m.Epoch, m.Connection) {
389 return errors.Join(ErrScopeMismatch, ErrProtocol)
390 }
391 _, _ = c.conn.StreamProgress(m.Stream)
392 return nil
393 case brokerwire.BrokerErrorMessage:
394 if !c.scopeMatches(m.Epoch, m.Connection) {
395 return errors.Join(ErrScopeMismatch, ErrProtocol)
396 }
397 cause := failureFromDetail(m.Error)
398 if cause == nil {
399 cause = ports.BrokerError{Code: ports.BrokerErrorUnavailable, Text: "empty broker error frame"}
400 }
401 return cause
402 case brokerwire.Shutdown:
403 text := m.Text
404 if text == "" {
405 text = "broker shutdown"
406 }
407 return ports.BrokerError{Code: ports.BrokerErrorUnavailable, Text: text}
408 default:
409 return errors.Join(ErrProtocol, errors.New("brokeripc: unexpected server message"))
410 }
411}
412
413// scopeMatches reports whether one frame carries this connection's accepted
414// scope.
415func (c *client) scopeMatches(epoch ports.BrokerEpoch, connection ports.BrokerConnectionID) bool {
416 return epoch == c.scope.Epoch && connection == c.scope.Connection
417}
418
419// publish installs one committed snapshot and wakes the active subscription.
420func (c *client) publish(snapshot ports.BrokerSnapshot) {
421 c.mu.Lock()
422 c.snapshot = snapshot
423 sub := c.sub
424 c.mu.Unlock()
425 if sub != nil {
426 sub.notify()
427 }
428}
429
430// releaseOperation removes all client-side state for an operation.
431func (c *client) releaseOperation(operation ports.BrokerOperationID) {
432 c.mu.Lock()
433 delete(c.pending, operation)
434 delete(c.kinds, operation)
435 c.mu.Unlock()
436}
437
438// completeOperation hands one result to the operation's waiter, if any.
439func (c *client) completeOperation(operation ports.BrokerOperationID, result operationResult) {
440 c.mu.Lock()
441 reply, ok := c.pending[operation]
442 delete(c.pending, operation)
443 delete(c.kinds, operation)
444 c.mu.Unlock()
445 if !ok {
446 return
447 }
448 select {
449 case reply <- result:
450 default:
451 }
452}
453
454// ConnectionID returns the identity assigned to this connection at accept.
455func (c *client) ConnectionID() ports.BrokerConnectionID { return c.scope.Connection }
456
457// NextStreamID allocates this connection's next logical stream identity. It is
458// strictly monotone, never zero, and consumes no wire frame: it is the local
459// allocation of the calling service, which then carries the exact
460// epoch/connection/stream triplet on OpenStream. It performs no I/O and refuses
461// a closed connection or an exhausted counter rather than wrapping.
462func (c *client) NextStreamID() (ports.BrokerStreamID, error) {
463 if c == nil {
464 return 0, ErrConnectionClosed
465 }
466 c.mu.Lock()
467 defer c.mu.Unlock()
468 if c.closed {
469 return 0, c.terminalErrorLocked()
470 }
471 if c.nextStream == ^ports.BrokerStreamID(0) {
472 return 0, ports.BrokerAdmissionLimit
473 }
474 c.nextStream++
475 return c.nextStream, nil
476}
477
478// Done closes exactly once when this connection is terminal: the carriage
479// failed, the broker retired it, or this side closed it. Err is stable
480// afterwards.
481func (c *client) Done() <-chan struct{} { return c.done }
482
483// Err returns this connection's terminal cause: nil for an orderly local Close,
484// otherwise the failure that settled it. The cause is recorded exactly once and
485// never overwritten, so it is stable once Done is closed.
486func (c *client) Err() error {
487 c.mu.Lock()
488 defer c.mu.Unlock()
489 return c.err
490}
491
492// Snapshot returns the newest fully committed publication.
493func (c *client) Snapshot() ports.BrokerSnapshot {
494 c.mu.Lock()
495 defer c.mu.Unlock()
496 return c.snapshot.Clone()
497}
498
499// Subscribe asks the broker for publications of the current scope. The call is
500// a bounded, non-blocking hint: it enqueues one Subscribe frame carrying a fresh
501// generation and returns immediately. Each notification means a newer committed
502// snapshot is available through Snapshot.
503//
504// Adopting the generation and enqueueing its frame are one step on this
505// connection: the local generation series is consumed as soon as the frame is
506// admitted, so a send that then fails has already superseded the previous
507// subscription. That previous subscription is therefore retired rather than
508// restored, and the connection is left with no current subscription: the caller
509// sees the send failure and subscribes again with a fresh generation.
510func (c *client) Subscribe() (ports.BrokerSubscription, error) {
511 c.mu.Lock()
512 if c.closed {
513 err := c.terminalErrorLocked()
514 c.mu.Unlock()
515 return nil, err
516 }
517 generation := c.generation + 1
518 if err := c.conn.Subscribe(generation); err != nil {
519 c.mu.Unlock()
520 return nil, admissionError(err)
521 }
522 c.generation = generation
523 sub := newClientSubscription()
524 previous := c.sub
525 c.sub = sub
526 c.mu.Unlock()
527 if err := c.sendAsync(brokerwire.Subscribe{
528 Epoch: c.scope.Epoch, Connection: c.scope.Connection, Generation: generation,
529 Passive: c.cfg.PassiveSubscribe,
530 }); err != nil {
531 // The generation was already consumed by the connection's subscription
532 // tracker, so the previous subscription's generation is no longer
533 // current even though this frame never left. Restoring it would present
534 // a superseded generation as the live subscription; retire both and
535 // leave the connection with no current subscription instead.
536 c.mu.Lock()
537 if c.sub == sub {
538 c.sub = nil
539 }
540 c.mu.Unlock()
541 sub.Close()
542 previous.Close()
543 return nil, err
544 }
545 if previous != nil {
546 previous.Close()
547 }
548 return sub, nil
549}
550
551// OpenStream opens one independently cancellable logical stream. An absent
552// epoch or connection identity is filled from this connection and a
553// caller-supplied identity that belongs elsewhere is refused as stale. The
554// stream identity must be the exact one the calling service allocated with
555// NextStreamID: it is admitted on this connection's bounded anti-replay window
556// before any frame travels, so a duplicate and an identity evicted past the
557// window are both refused, while a concurrent lower identity that was allocated
558// first is still admitted.
559func (c *client) SubscribePreview(request ports.BrokerPreviewRequest) (ports.BrokerPreviewSubscription, error) {
560 if err := request.Validate(); err != nil {
561 return nil, errors.Join(ports.BrokerAdmissionInvalid, err)
562 }
563 if request.Epoch != c.scope.Epoch || request.Connection != c.scope.Connection {
564 return nil, ports.BrokerAdmissionStale
565 }
566 c.mu.Lock()
567 if c.closed {
568 err := c.terminalErrorLocked()
569 c.mu.Unlock()
570 return nil, err
571 }
572 if request.Generation <= c.previewGeneration {
573 c.mu.Unlock()
574 return nil, ports.BrokerAdmissionStale
575 }
576 previous := c.preview
577 sub := newClientPreviewSubscription(c, request)
578 c.preview, c.previewGeneration = sub, request.Generation
579 c.mu.Unlock()
580 if err := c.sendAsync(brokerwire.StartPreview{Epoch: request.Epoch, Connection: request.Connection, Generation: request.Generation, Route: request.Route, Preview: request.Preview}); err != nil {
581 // The generation was already consumed as the connection's current
582 // preview generation, so the previous subscription is superseded even
583 // though this frame never left. Retire both and leave the connection
584 // with no current preview, mirroring Subscribe's send-error path.
585 c.mu.Lock()
586 if c.preview == sub {
587 c.preview = nil
588 }
589 c.mu.Unlock()
590 sub.Close()
591 previous.Close()
592 return nil, err
593 }
594 if previous != nil {
595 previous.Close()
596 }
597 return sub, nil
598}
599
600func (c *client) OpenStream(ctx context.Context, request ports.BrokerOpenStreamRequest) (ports.BrokerLogicalConnection, error) {
601 scoped, err := c.scopeRequest(request)
602 if err != nil {
603 return nil, err
604 }
605 c.mu.Lock()
606 if c.closed {
607 err := c.terminalErrorLocked()
608 c.mu.Unlock()
609 return nil, err
610 }
611 // Admit the identity on the connection's own tracker before any frame
612 // travels, so a racing StreamOpened can never precede local admission. The
613 // tracker applies the bounded anti-replay window: a duplicate and an
614 // identity evicted past the window are refused, while a concurrent lower
615 // identity that was allocated first is still admitted.
616 if err := c.conn.OpenStream(scoped.Stream); err != nil {
617 c.mu.Unlock()
618 return nil, admissionError(err)
619 }
620 stream, err := newClientStream(c, scoped.Stream)
621 if err != nil {
622 _, _ = c.conn.CloseStream(scoped.Stream)
623 c.mu.Unlock()
624 return nil, err
625 }
626 c.streams[scoped.Stream] = stream
627 c.mu.Unlock()
628
629 // OpenStream treats every send failure alike: a stream whose request
630 // definitely never left is retired exactly like one that may have been
631 // dispatched, because the local identity is already admitted either way.
632 if _, err := c.sendContext(ctx, brokerwire.OpenStream{
633 Epoch: scoped.Epoch, Connection: scoped.Connection, Stream: scoped.Stream,
634 Purpose: scoped.Purpose, Admission: scoped.Admission, Name: scoped.Name,
635 Local: scoped.Local,
636 Endpoint: scoped.Endpoint, Registration: scoped.Registration,
637 Target: scoped.Target, Env: scoped.Env, Policy: scoped.Policy,
638 StartMode: scoped.StartMode,
639 }); err != nil {
640 c.retireStream(scoped.Stream, err)
641 return nil, err
642 }
643 select {
644 case err := <-stream.opened:
645 if err != nil {
646 return nil, err
647 }
648 return stream, nil
649 case <-ctx.Done():
650 // The request was sent, so the broker may still establish the stream.
651 // Retire it locally and tell the broker; the caller sees cancellation.
652 c.retireStream(scoped.Stream, ctx.Err())
653 return nil, ctx.Err()
654 case <-c.done:
655 return nil, c.terminalError()
656 }
657}
658
659// CloseStream retires one scope-checked logical stream.
660func (c *client) CloseStream(connection ports.BrokerConnectionID, stream ports.BrokerStreamID) error {
661 if !connection.IsZero() && connection != c.scope.Connection {
662 return errors.Join(ErrScopeMismatch, ports.BrokerAdmissionStale)
663 }
664 c.mu.Lock()
665 if c.closed {
666 err := c.terminalErrorLocked()
667 c.mu.Unlock()
668 return err
669 }
670 _, err := c.conn.CloseStream(stream)
671 c.mu.Unlock()
672 if err != nil {
673 return admissionError(err)
674 }
675 c.requestStreamClose(stream)
676 if st := c.lookupStream(stream); st != nil {
677 st.terminate(nil)
678 }
679 return nil
680}
681
682// AddHost sends one exact endpoint and policy mutation.
683func (c *client) AddHost(ctx context.Context, endpoint string, policy ports.BrokerPolicy) (domain.RemoteRegistration, error) {
684 result, err := c.mutate(ctx, brokerwire.MutationKindAddHost, func(operation ports.BrokerOperationID) brokerwire.ClientMessage {
685 return brokerwire.AddHost{Epoch: c.scope.Epoch, Connection: c.scope.Connection, Operation: operation, Endpoint: endpoint, Policy: policy}
686 })
687 if err != nil {
688 return domain.RemoteRegistration{}, err
689 }
690 return result.registration, mutationFailure(result)
691}
692
693// RemoveHost sends one exact-registration mutation.
694func (c *client) RemoveHost(ctx context.Context, expected domain.RemoteRegistration) (bool, error) {
695 result, err := c.mutate(ctx, brokerwire.MutationKindRemoveHost, func(operation ports.BrokerOperationID) brokerwire.ClientMessage {
696 return brokerwire.RemoveHost{Epoch: c.scope.Epoch, Connection: c.scope.Connection, Operation: operation, Registration: expected}
697 })
698 if err != nil {
699 return false, err
700 }
701 return result.removed, mutationFailure(result)
702}
703
704// UpdateHostPolicy sends one exact-registration replacement policy mutation.
705func (c *client) UpdateHostPolicy(ctx context.Context, expected domain.RemoteRegistration, policy ports.BrokerPolicy) (domain.RemoteRegistration, error) {
706 result, err := c.mutate(ctx, brokerwire.MutationKindUpdateHostPolicy, func(operation ports.BrokerOperationID) brokerwire.ClientMessage {
707 return brokerwire.UpdateHostPolicy{Epoch: c.scope.Epoch, Connection: c.scope.Connection, Operation: operation, Registration: expected, Policy: policy}
708 })
709 if err != nil {
710 return domain.RemoteRegistration{}, err
711 }
712 return result.registration, mutationFailure(result)
713}
714
715// RequestReconcile asks the broker to re-observe one endpoint. The port carries
716// an endpoint while the wire carries an exact registration, so the hint is sent
717// only for an endpoint this connection already holds a registration for; an
718// unknown endpoint is a no-op. It never blocks and never fails the caller.
719func (c *client) RequestReconcile(endpoint string) {
720 c.mu.Lock()
721 host, ok := c.snapshot.Find(endpoint)
722 closed := c.closed
723 c.mu.Unlock()
724 if closed || !ok {
725 return
726 }
727 _ = c.sendAsync(brokerwire.Reconcile{Epoch: c.scope.Epoch, Connection: c.scope.Connection, Registration: host.Registration})
728}
729
730// Close ends the connection, releasing its streams and subscription. It is
731// idempotent and concurrent-safe.
732func (c *client) Close() error {
733 if c == nil {
734 return nil
735 }
736 c.closeWith(nil)
737 return nil
738}
739
740// scopeRequest fills an absent epoch or connection identity from this
741// connection and refuses one that belongs elsewhere, then validates the
742// resulting request. The stream identity must be present: after cutover the
743// calling service allocates it with NextStreamID and carries the exact
744// epoch/connection/stream triplet, so this adapter never invents an identity on
745// an incomplete request.
746func (c *client) scopeRequest(request ports.BrokerOpenStreamRequest) (ports.BrokerOpenStreamRequest, error) {
747 scoped := request
748 if scoped.Epoch == 0 {
749 scoped.Epoch = c.scope.Epoch
750 } else if scoped.Epoch != c.scope.Epoch {
751 return ports.BrokerOpenStreamRequest{}, errors.Join(ErrScopeMismatch, ports.BrokerError{Code: ports.BrokerErrorStaleEpoch})
752 }
753 if scoped.Connection.IsZero() {
754 scoped.Connection = c.scope.Connection
755 } else if scoped.Connection != c.scope.Connection {
756 return ports.BrokerOpenStreamRequest{}, errors.Join(ErrScopeMismatch, ports.BrokerAdmissionStale)
757 }
758 if scoped.Stream == 0 {
759 return ports.BrokerOpenStreamRequest{}, errors.Join(ports.BrokerAdmissionInvalid, errors.New("brokeripc: open stream request has no stream identity"))
760 }
761 return scoped, nil
762}
763
764// mutate sends one mutating operation and waits for its bounded completion. The
765// operation identity is admitted on the connection's tracker (bounded by the
766// same per-connection bound the wire enforces) and completed exactly once, so an
767// abandoned or timed-out operation releases its pending slot and a late or
768// replayed result is deduped rather than applied twice.
769//
770// A request that definitely never left the process fails as an ordinary definite
771// error and records a failed outcome; once the wire attempt was launched, every
772// failure is reported as outcome-unknown, so a caller refreshes authoritative
773// state instead of replaying a non-idempotent mutation blindly.
774func (c *client) mutate(ctx context.Context, kind brokerwire.RegisterMutationKind, build func(ports.BrokerOperationID) brokerwire.ClientMessage) (operationResult, error) {
775 operation, err := newOperationID()
776 if err != nil {
777 return operationResult{}, err
778 }
779 reply := make(chan operationResult, 1)
780 c.mu.Lock()
781 if c.closed {
782 err := c.terminalErrorLocked()
783 c.mu.Unlock()
784 return operationResult{}, err
785 }
786 if len(c.pending) >= c.cfg.MaxPendingOperations {
787 c.mu.Unlock()
788 return operationResult{}, ports.BrokerAdmissionLimit
789 }
790 if err := c.conn.AdmitOperation(operation); err != nil {
791 c.mu.Unlock()
792 return operationResult{}, admissionError(err)
793 }
794 c.pending[operation] = reply
795 if c.kinds == nil {
796 c.kinds = make(map[ports.BrokerOperationID]brokerwire.RegisterMutationKind)
797 }
798 c.kinds[operation] = kind
799 c.mu.Unlock()
800 // No result observed is exactly the outcome-unknown case; a request that
801 // was never launched is recorded as failed instead.
802 outcome := ports.BrokerOutcomeUnknown
803 defer func() {
804 c.releaseOperation(operation)
805 _ = c.conn.CompleteOperation(operation, outcome)
806 }()
807 attempted, err := c.sendContext(ctx, build(operation))
808 if err != nil {
809 if !attempted {
810 // The request definitely did not travel, so a definite failure is
811 // the honest report and the operation is recorded as failed.
812 outcome = ports.BrokerOutcomeFailed
813 return operationResult{}, err
814 }
815 // The wire attempt was launched: the broker may have committed the
816 // operation even though this side observed no result.
817 return operationResult{outcome: ports.BrokerOutcomeUnknown, detail: errorDetail(err)}, nil
818 }
819 select {
820 case result := <-reply:
821 outcome = result.outcome
822 return result, nil
823 case <-ctx.Done():
824 // The operation was already sent: the broker may have committed it.
825 return operationResult{outcome: ports.BrokerOutcomeUnknown, detail: errorDetail(ctx.Err())}, nil
826 case <-c.done:
827 return operationResult{outcome: ports.BrokerOutcomeUnknown, detail: errorDetail(c.terminalError())}, nil
828 }
829}
830
831// send encodes and writes one client frame synchronously.
832func (c *client) send(message brokerwire.ClientMessage) error {
833 payload, err := brokerwire.EncodeClient(message, c.ceilings.MaxReceiveEnvelopeBytes, c.ceilings.StreamChunkLimit)
834 if err != nil {
835 return err
836 }
837 if err := c.transport.Send(wire.Envelope{Payload: payload}); err != nil {
838 return transportFailure(err)
839 }
840 return nil
841}
842
843// sendAsync admits one client frame for ordered background transmission. It
844// reports backpressure as a retryable unavailability rather than blocking a
845// caller that has no context to spend.
846func (c *client) sendAsync(message brokerwire.ClientMessage) error {
847 payload, err := brokerwire.EncodeClient(message, c.ceilings.MaxReceiveEnvelopeBytes, c.ceilings.StreamChunkLimit)
848 if err != nil {
849 return err
850 }
851 if async, ok := c.transport.(wire.AsyncTransport); ok {
852 if err := async.SendAsync(wire.Envelope{Payload: payload}); err != nil {
853 if errors.Is(err, streamframe.ErrBackpressure) {
854 return ports.BrokerError{Code: ports.BrokerErrorUnavailable, Text: "broker IPC send queue is full"}
855 }
856 return transportFailure(err)
857 }
858 return nil
859 }
860 if err := c.transport.Send(wire.Envelope{Payload: payload}); err != nil {
861 return transportFailure(err)
862 }
863 return nil
864}
865
866// sendContext writes one client frame, honoring cancellation while the carriage
867// egress queue is full. It reports whether the wire attempt was launched, which
868// is exactly the dispatch uncertainty a caller needs: an encoding failure, an
869// already-cancelled call, or a connection already terminal before the attempt
870// begins means the frame definitely did not travel, while every failure
871// observed after the attempt is launched means it may have been dispatched and
872// the caller must not replay blindly. The wire attempt is raced against the
873// caller's context and the connection's terminal outcome, so a stalled send
874// never pins the caller: the racing goroutine observes the closed carriage and
875// returns.
876func (c *client) sendContext(ctx context.Context, message brokerwire.ClientMessage) (attempted bool, err error) {
877 payload, err := brokerwire.EncodeClient(message, c.ceilings.MaxReceiveEnvelopeBytes, c.ceilings.StreamChunkLimit)
878 if err != nil {
879 return false, err
880 }
881 // Synchronous preflight. Checking before the goroutine starts is what makes
882 // the not-dispatched claim true: a context that is already done or a
883 // connection that is already terminal can no longer race a launched Send.
884 if err := ctx.Err(); err != nil {
885 return false, err
886 }
887 select {
888 case <-c.done:
889 return false, c.terminalError()
890 default:
891 }
892 done := make(chan error, 1)
893 go func() { done <- c.transport.Send(wire.Envelope{Payload: payload}) }()
894 select {
895 case err := <-done:
896 if err != nil {
897 return true, transportFailure(err)
898 }
899 return true, nil
900 case <-ctx.Done():
901 return true, ctx.Err()
902 case <-c.done:
903 return true, c.terminalError()
904 }
905}
906
907// requestStreamClose tells the broker one stream is retired. It is best-effort:
908// the stream is already settled locally, so a full queue or a dead carriage
909// needs no further action.
910func (c *client) requestStreamClose(stream ports.BrokerStreamID) {
911 _ = c.sendAsync(brokerwire.CloseStream{Epoch: c.scope.Epoch, Connection: c.scope.Connection, Stream: stream})
912}
913
914// retireStream settles one local stream and tells the broker to retire it.
915func (c *client) retireStream(stream ports.BrokerStreamID, cause error) {
916 st := c.lookupStream(stream)
917 if st != nil {
918 st.terminate(cause)
919 }
920 c.requestStreamClose(stream)
921}
922
923// lookupStream returns the local stream for one identity.
924func (c *client) lookupStream(stream ports.BrokerStreamID) *clientStream {
925 c.mu.Lock()
926 defer c.mu.Unlock()
927 return c.streams[stream]
928}
929
930// streamFor classifies one stream-scoped frame (data or a credit grant) and
931// returns its live stream. A frame for a retired stream, or one retired locally
932// between the tracker check and the lookup, yields no stream and no error; a
933// frame for a stream that was never opened is a peer protocol violation.
934func (c *client) streamFor(id ports.BrokerStreamID) (*clientStream, error) {
935 disposition, err := c.conn.StreamData(id)
936 if err != nil {
937 return nil, errors.Join(ErrProtocol, err)
938 }
939 if disposition == brokerwire.StreamDiscarded {
940 return nil, nil
941 }
942 return c.lookupStream(id), nil
943}
944
945// unregisterStream drops one local stream record.
946func (c *client) unregisterStream(stream ports.BrokerStreamID) {
947 c.mu.Lock()
948 delete(c.streams, stream)
949 c.mu.Unlock()
950}
951
952// terminalErrorLocked returns the terminal cause with the connection lock held.
953func (c *client) terminalErrorLocked() error {
954 if c.err != nil {
955 return c.err
956 }
957 return ErrConnectionClosed
958}
959
960// terminalError returns the terminal cause.
961func (c *client) terminalError() error {
962 c.mu.Lock()
963 defer c.mu.Unlock()
964 return c.terminalErrorLocked()
965}
966
967// closeWith tears the connection down exactly once: cancel, close the carriage
968// (unblocking a blocked read and any frame send), settle every local stream and
969// the active subscription, and wake every waiter. A nil cause is an orderly
970// local close.
971func (c *client) closeWith(cause error) {
972 if c == nil {
973 return
974 }
975 c.closeOnce.Do(func() {
976 c.cancel()
977 _ = c.transport.Close()
978 c.mu.Lock()
979 c.closed = true
980 if cause != nil && c.err == nil {
981 c.err = cause
982 }
983 terminal := c.terminalErrorLocked()
984 sub := c.sub
985 c.sub = nil
986 preview := c.preview
987 c.preview = nil
988 streams := make([]*clientStream, 0, len(c.streams))
989 for _, st := range c.streams {
990 streams = append(streams, st)
991 }
992 c.streams = make(map[ports.BrokerStreamID]*clientStream)
993 c.mu.Unlock()
994 for _, st := range streams {
995 st.terminate(terminal)
996 }
997 if sub != nil {
998 sub.Close()
999 }
1000 if preview != nil {
1001 preview.Close()
1002 }
1003 close(c.done)
1004 })
1005}
1006
1007// mutationFailure converts one operation outcome into its typed caller error.
1008func mutationFailure(result operationResult) error {
1009 switch result.outcome {
1010 case ports.BrokerOutcomeOK:
1011 return nil
1012 case ports.BrokerOutcomeFailed:
1013 if err := failureFromDetail(result.detail); err != nil {
1014 return err
1015 }
1016 return ports.BrokerError{Code: ports.BrokerErrorUnavailable}
1017 default:
1018 return ports.BrokerError{Code: ports.BrokerErrorOutcomeUnknown}
1019 }
1020}
1021
1022// newOperationID samples one fresh non-zero operation identity.
1023func newOperationID() (ports.BrokerOperationID, error) {
1024 var id ports.BrokerOperationID
1025 if _, err := rand.Read(id[:]); err != nil {
1026 return ports.BrokerOperationID{}, err
1027 }
1028 if id.IsZero() {
1029 id[15] = 1
1030 }
1031 return id, nil
1032}
1033
1034// clientSubscription is one local capacity-one change notification. Each
1035// subscriber owns its own channel and re-reads the newest snapshot on wake.
1036type clientSubscription struct {
1037 changed chan struct{}
1038 once sync.Once
1039 mu sync.Mutex
1040 closed bool
1041}
1042
1043func newClientSubscription() *clientSubscription {
1044 return &clientSubscription{changed: make(chan struct{}, 1)}
1045}
1046
1047// Changed is signalled once per committed publication.
1048func (s *clientSubscription) Changed() <-chan struct{} { return s.changed }
1049
1050// Close ends the subscription exactly once. The channel is closed under the same
1051// lock notify sends under, so a racing publication can never send on a closed
1052// channel.
1053func (s *clientSubscription) Close() {
1054 if s == nil {
1055 return
1056 }
1057 s.once.Do(func() {
1058 s.mu.Lock()
1059 s.closed = true
1060 close(s.changed)
1061 s.mu.Unlock()
1062 })
1063}
1064
1065// notify coalesces one publication wake.
1066func (s *clientSubscription) notify() {
1067 if s == nil {
1068 return
1069 }
1070 s.mu.Lock()
1071 defer s.mu.Unlock()
1072 if s.closed {
1073 return
1074 }
1075 select {
1076 case s.changed <- struct{}{}:
1077 default:
1078 }
1079}
1080
1081// StaleBuild reports the identity of a connected broker from another build.
1082func StaleBuild(service ports.BrokerService) (string, bool) {
1083 c, ok := service.(*client)
1084 if !ok || c == nil || c.staleBuild == "" {
1085 return "", false
1086 }
1087 return c.staleBuild, true
1088}