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 disposition, err := c.conn.StreamData(m.Stream)
355 if err != nil {
356 return errors.Join(ErrProtocol, err)
357 }
358 if disposition == brokerwire.StreamDiscarded {
359 return nil
360 }
361 st := c.lookupStream(m.Stream)
362 if st == nil {
363 // The stream was retired locally between the tracker check and the
364 // lookup; its data is discarded, exactly like a retired stream.
365 return nil
366 }
367 if err := st.deliver(m.Data); err != nil {
368 if errors.Is(err, ErrStreamBackpressure) {
369 st.fail(ErrStreamBackpressure)
370 }
371 }
372 return nil
373 case brokerwire.StreamClosed:
374 if !c.scopeMatches(m.Epoch, m.Connection) {
375 return errors.Join(ErrScopeMismatch, ErrProtocol)
376 }
377 _, _ = c.conn.CloseStream(m.Stream)
378 if st := c.lookupStream(m.Stream); st != nil {
379 var cause error
380 if m.HasError {
381 cause = failureFromDetail(m.Error)
382 }
383 st.terminate(cause)
384 }
385 return nil
386 case brokerwire.Progress:
387 if !c.scopeMatches(m.Epoch, m.Connection) {
388 return errors.Join(ErrScopeMismatch, ErrProtocol)
389 }
390 _, _ = c.conn.StreamProgress(m.Stream)
391 return nil
392 case brokerwire.BrokerErrorMessage:
393 if !c.scopeMatches(m.Epoch, m.Connection) {
394 return errors.Join(ErrScopeMismatch, ErrProtocol)
395 }
396 cause := failureFromDetail(m.Error)
397 if cause == nil {
398 cause = ports.BrokerError{Code: ports.BrokerErrorUnavailable, Text: "empty broker error frame"}
399 }
400 return cause
401 case brokerwire.Shutdown:
402 text := m.Text
403 if text == "" {
404 text = "broker shutdown"
405 }
406 return ports.BrokerError{Code: ports.BrokerErrorUnavailable, Text: text}
407 default:
408 return errors.Join(ErrProtocol, errors.New("brokeripc: unexpected server message"))
409 }
410}
411
412// scopeMatches reports whether one frame carries this connection's accepted
413// scope.
414func (c *client) scopeMatches(epoch ports.BrokerEpoch, connection ports.BrokerConnectionID) bool {
415 return epoch == c.scope.Epoch && connection == c.scope.Connection
416}
417
418// publish installs one committed snapshot and wakes the active subscription.
419func (c *client) publish(snapshot ports.BrokerSnapshot) {
420 c.mu.Lock()
421 c.snapshot = snapshot
422 sub := c.sub
423 c.mu.Unlock()
424 if sub != nil {
425 sub.notify()
426 }
427}
428
429// releaseOperation removes all client-side state for an operation.
430func (c *client) releaseOperation(operation ports.BrokerOperationID) {
431 c.mu.Lock()
432 delete(c.pending, operation)
433 delete(c.kinds, operation)
434 c.mu.Unlock()
435}
436
437// completeOperation hands one result to the operation's waiter, if any.
438func (c *client) completeOperation(operation ports.BrokerOperationID, result operationResult) {
439 c.mu.Lock()
440 reply, ok := c.pending[operation]
441 delete(c.pending, operation)
442 delete(c.kinds, operation)
443 c.mu.Unlock()
444 if !ok {
445 return
446 }
447 select {
448 case reply <- result:
449 default:
450 }
451}
452
453// ConnectionID returns the identity assigned to this connection at accept.
454func (c *client) ConnectionID() ports.BrokerConnectionID { return c.scope.Connection }
455
456// NextStreamID allocates this connection's next logical stream identity. It is
457// strictly monotone, never zero, and consumes no wire frame: it is the local
458// allocation of the calling service, which then carries the exact
459// epoch/connection/stream triplet on OpenStream. It performs no I/O and refuses
460// a closed connection or an exhausted counter rather than wrapping.
461func (c *client) NextStreamID() (ports.BrokerStreamID, error) {
462 if c == nil {
463 return 0, ErrConnectionClosed
464 }
465 c.mu.Lock()
466 defer c.mu.Unlock()
467 if c.closed {
468 return 0, c.terminalErrorLocked()
469 }
470 if c.nextStream == ^ports.BrokerStreamID(0) {
471 return 0, ports.BrokerAdmissionLimit
472 }
473 c.nextStream++
474 return c.nextStream, nil
475}
476
477// Done closes exactly once when this connection is terminal: the carriage
478// failed, the broker retired it, or this side closed it. Err is stable
479// afterwards.
480func (c *client) Done() <-chan struct{} { return c.done }
481
482// Err returns this connection's terminal cause: nil for an orderly local Close,
483// otherwise the failure that settled it. The cause is recorded exactly once and
484// never overwritten, so it is stable once Done is closed.
485func (c *client) Err() error {
486 c.mu.Lock()
487 defer c.mu.Unlock()
488 return c.err
489}
490
491// Snapshot returns the newest fully committed publication.
492func (c *client) Snapshot() ports.BrokerSnapshot {
493 c.mu.Lock()
494 defer c.mu.Unlock()
495 return c.snapshot.Clone()
496}
497
498// Subscribe asks the broker for publications of the current scope. The call is
499// a bounded, non-blocking hint: it enqueues one Subscribe frame carrying a fresh
500// generation and returns immediately. Each notification means a newer committed
501// snapshot is available through Snapshot.
502//
503// Adopting the generation and enqueueing its frame are one step on this
504// connection: the local generation series is consumed as soon as the frame is
505// admitted, so a send that then fails has already superseded the previous
506// subscription. That previous subscription is therefore retired rather than
507// restored, and the connection is left with no current subscription: the caller
508// sees the send failure and subscribes again with a fresh generation.
509func (c *client) Subscribe() (ports.BrokerSubscription, error) {
510 c.mu.Lock()
511 if c.closed {
512 err := c.terminalErrorLocked()
513 c.mu.Unlock()
514 return nil, err
515 }
516 generation := c.generation + 1
517 if err := c.conn.Subscribe(generation); err != nil {
518 c.mu.Unlock()
519 return nil, admissionError(err)
520 }
521 c.generation = generation
522 sub := newClientSubscription()
523 previous := c.sub
524 c.sub = sub
525 c.mu.Unlock()
526 if err := c.sendAsync(brokerwire.Subscribe{
527 Epoch: c.scope.Epoch, Connection: c.scope.Connection, Generation: generation,
528 Passive: c.cfg.PassiveSubscribe,
529 }); err != nil {
530 // The generation was already consumed by the connection's subscription
531 // tracker, so the previous subscription's generation is no longer
532 // current even though this frame never left. Restoring it would present
533 // a superseded generation as the live subscription; retire both and
534 // leave the connection with no current subscription instead.
535 c.mu.Lock()
536 if c.sub == sub {
537 c.sub = nil
538 }
539 c.mu.Unlock()
540 sub.Close()
541 previous.Close()
542 return nil, err
543 }
544 if previous != nil {
545 previous.Close()
546 }
547 return sub, nil
548}
549
550// OpenStream opens one independently cancellable logical stream. An absent
551// epoch or connection identity is filled from this connection and a
552// caller-supplied identity that belongs elsewhere is refused as stale. The
553// stream identity must be the exact one the calling service allocated with
554// NextStreamID: it is admitted on this connection's bounded anti-replay window
555// before any frame travels, so a duplicate and an identity evicted past the
556// window are both refused, while a concurrent lower identity that was allocated
557// first is still admitted.
558func (c *client) SubscribePreview(request ports.BrokerPreviewRequest) (ports.BrokerPreviewSubscription, error) {
559 if err := request.Validate(); err != nil {
560 return nil, errors.Join(ports.BrokerAdmissionInvalid, err)
561 }
562 if request.Epoch != c.scope.Epoch || request.Connection != c.scope.Connection {
563 return nil, ports.BrokerAdmissionStale
564 }
565 c.mu.Lock()
566 if c.closed {
567 err := c.terminalErrorLocked()
568 c.mu.Unlock()
569 return nil, err
570 }
571 if request.Generation <= c.previewGeneration {
572 c.mu.Unlock()
573 return nil, ports.BrokerAdmissionStale
574 }
575 previous := c.preview
576 sub := newClientPreviewSubscription(c, request)
577 c.preview, c.previewGeneration = sub, request.Generation
578 c.mu.Unlock()
579 if err := c.sendAsync(brokerwire.StartPreview{Epoch: request.Epoch, Connection: request.Connection, Generation: request.Generation, Route: request.Route, Preview: request.Preview}); err != nil {
580 // The generation was already consumed as the connection's current
581 // preview generation, so the previous subscription is superseded even
582 // though this frame never left. Retire both and leave the connection
583 // with no current preview, mirroring Subscribe's send-error path.
584 c.mu.Lock()
585 if c.preview == sub {
586 c.preview = nil
587 }
588 c.mu.Unlock()
589 sub.Close()
590 previous.Close()
591 return nil, err
592 }
593 if previous != nil {
594 previous.Close()
595 }
596 return sub, nil
597}
598
599func (c *client) OpenStream(ctx context.Context, request ports.BrokerOpenStreamRequest) (ports.BrokerLogicalConnection, error) {
600 scoped, err := c.scopeRequest(request)
601 if err != nil {
602 return nil, err
603 }
604 c.mu.Lock()
605 if c.closed {
606 err := c.terminalErrorLocked()
607 c.mu.Unlock()
608 return nil, err
609 }
610 // Admit the identity on the connection's own tracker before any frame
611 // travels, so a racing StreamOpened can never precede local admission. The
612 // tracker applies the bounded anti-replay window: a duplicate and an
613 // identity evicted past the window are refused, while a concurrent lower
614 // identity that was allocated first is still admitted.
615 if err := c.conn.OpenStream(scoped.Stream); err != nil {
616 c.mu.Unlock()
617 return nil, admissionError(err)
618 }
619 stream, err := newClientStream(c, scoped.Stream)
620 if err != nil {
621 _, _ = c.conn.CloseStream(scoped.Stream)
622 c.mu.Unlock()
623 return nil, err
624 }
625 c.streams[scoped.Stream] = stream
626 c.mu.Unlock()
627
628 // OpenStream treats every send failure alike: a stream whose request
629 // definitely never left is retired exactly like one that may have been
630 // dispatched, because the local identity is already admitted either way.
631 if _, err := c.sendContext(ctx, brokerwire.OpenStream{
632 Epoch: scoped.Epoch, Connection: scoped.Connection, Stream: scoped.Stream,
633 Purpose: scoped.Purpose, Admission: scoped.Admission, Name: scoped.Name,
634 Local: scoped.Local,
635 Endpoint: scoped.Endpoint, Registration: scoped.Registration,
636 Target: scoped.Target, Env: scoped.Env, Policy: scoped.Policy,
637 StartMode: scoped.StartMode,
638 }); err != nil {
639 c.retireStream(scoped.Stream, err)
640 return nil, err
641 }
642 select {
643 case err := <-stream.opened:
644 if err != nil {
645 return nil, err
646 }
647 return stream, nil
648 case <-ctx.Done():
649 // The request was sent, so the broker may still establish the stream.
650 // Retire it locally and tell the broker; the caller sees cancellation.
651 c.retireStream(scoped.Stream, ctx.Err())
652 return nil, ctx.Err()
653 case <-c.done:
654 return nil, c.terminalError()
655 }
656}
657
658// CloseStream retires one scope-checked logical stream.
659func (c *client) CloseStream(connection ports.BrokerConnectionID, stream ports.BrokerStreamID) error {
660 if !connection.IsZero() && connection != c.scope.Connection {
661 return errors.Join(ErrScopeMismatch, ports.BrokerAdmissionStale)
662 }
663 c.mu.Lock()
664 if c.closed {
665 err := c.terminalErrorLocked()
666 c.mu.Unlock()
667 return err
668 }
669 _, err := c.conn.CloseStream(stream)
670 c.mu.Unlock()
671 if err != nil {
672 return admissionError(err)
673 }
674 c.requestStreamClose(stream)
675 if st := c.lookupStream(stream); st != nil {
676 st.terminate(nil)
677 }
678 return nil
679}
680
681// AddHost sends one exact endpoint and policy mutation.
682func (c *client) AddHost(ctx context.Context, endpoint string, policy ports.BrokerPolicy) (domain.RemoteRegistration, error) {
683 result, err := c.mutate(ctx, brokerwire.MutationKindAddHost, func(operation ports.BrokerOperationID) brokerwire.ClientMessage {
684 return brokerwire.AddHost{Epoch: c.scope.Epoch, Connection: c.scope.Connection, Operation: operation, Endpoint: endpoint, Policy: policy}
685 })
686 if err != nil {
687 return domain.RemoteRegistration{}, err
688 }
689 return result.registration, mutationFailure(result)
690}
691
692// RemoveHost sends one exact-registration mutation.
693func (c *client) RemoveHost(ctx context.Context, expected domain.RemoteRegistration) (bool, error) {
694 result, err := c.mutate(ctx, brokerwire.MutationKindRemoveHost, func(operation ports.BrokerOperationID) brokerwire.ClientMessage {
695 return brokerwire.RemoveHost{Epoch: c.scope.Epoch, Connection: c.scope.Connection, Operation: operation, Registration: expected}
696 })
697 if err != nil {
698 return false, err
699 }
700 return result.removed, mutationFailure(result)
701}
702
703// UpdateHostPolicy sends one exact-registration replacement policy mutation.
704func (c *client) UpdateHostPolicy(ctx context.Context, expected domain.RemoteRegistration, policy ports.BrokerPolicy) (domain.RemoteRegistration, error) {
705 result, err := c.mutate(ctx, brokerwire.MutationKindUpdateHostPolicy, func(operation ports.BrokerOperationID) brokerwire.ClientMessage {
706 return brokerwire.UpdateHostPolicy{Epoch: c.scope.Epoch, Connection: c.scope.Connection, Operation: operation, Registration: expected, Policy: policy}
707 })
708 if err != nil {
709 return domain.RemoteRegistration{}, err
710 }
711 return result.registration, mutationFailure(result)
712}
713
714// RequestReconcile asks the broker to re-observe one endpoint. The port carries
715// an endpoint while the wire carries an exact registration, so the hint is sent
716// only for an endpoint this connection already holds a registration for; an
717// unknown endpoint is a no-op. It never blocks and never fails the caller.
718func (c *client) RequestReconcile(endpoint string) {
719 c.mu.Lock()
720 host, ok := c.snapshot.Find(endpoint)
721 closed := c.closed
722 c.mu.Unlock()
723 if closed || !ok {
724 return
725 }
726 _ = c.sendAsync(brokerwire.Reconcile{Epoch: c.scope.Epoch, Connection: c.scope.Connection, Registration: host.Registration})
727}
728
729// Close ends the connection, releasing its streams and subscription. It is
730// idempotent and concurrent-safe.
731func (c *client) Close() error {
732 if c == nil {
733 return nil
734 }
735 c.closeWith(nil)
736 return nil
737}
738
739// scopeRequest fills an absent epoch or connection identity from this
740// connection and refuses one that belongs elsewhere, then validates the
741// resulting request. The stream identity must be present: after cutover the
742// calling service allocates it with NextStreamID and carries the exact
743// epoch/connection/stream triplet, so this adapter never invents an identity on
744// an incomplete request.
745func (c *client) scopeRequest(request ports.BrokerOpenStreamRequest) (ports.BrokerOpenStreamRequest, error) {
746 scoped := request
747 if scoped.Epoch == 0 {
748 scoped.Epoch = c.scope.Epoch
749 } else if scoped.Epoch != c.scope.Epoch {
750 return ports.BrokerOpenStreamRequest{}, errors.Join(ErrScopeMismatch, ports.BrokerError{Code: ports.BrokerErrorStaleEpoch})
751 }
752 if scoped.Connection.IsZero() {
753 scoped.Connection = c.scope.Connection
754 } else if scoped.Connection != c.scope.Connection {
755 return ports.BrokerOpenStreamRequest{}, errors.Join(ErrScopeMismatch, ports.BrokerAdmissionStale)
756 }
757 if scoped.Stream == 0 {
758 return ports.BrokerOpenStreamRequest{}, errors.Join(ports.BrokerAdmissionInvalid, errors.New("brokeripc: open stream request has no stream identity"))
759 }
760 return scoped, nil
761}
762
763// mutate sends one mutating operation and waits for its bounded completion. The
764// operation identity is admitted on the connection's tracker (bounded by the
765// same per-connection bound the wire enforces) and completed exactly once, so an
766// abandoned or timed-out operation releases its pending slot and a late or
767// replayed result is deduped rather than applied twice.
768//
769// A request that definitely never left the process fails as an ordinary definite
770// error and records a failed outcome; once the wire attempt was launched, every
771// failure is reported as outcome-unknown, so a caller refreshes authoritative
772// state instead of replaying a non-idempotent mutation blindly.
773func (c *client) mutate(ctx context.Context, kind brokerwire.RegisterMutationKind, build func(ports.BrokerOperationID) brokerwire.ClientMessage) (operationResult, error) {
774 operation, err := newOperationID()
775 if err != nil {
776 return operationResult{}, err
777 }
778 reply := make(chan operationResult, 1)
779 c.mu.Lock()
780 if c.closed {
781 err := c.terminalErrorLocked()
782 c.mu.Unlock()
783 return operationResult{}, err
784 }
785 if len(c.pending) >= c.cfg.MaxPendingOperations {
786 c.mu.Unlock()
787 return operationResult{}, ports.BrokerAdmissionLimit
788 }
789 if err := c.conn.AdmitOperation(operation); err != nil {
790 c.mu.Unlock()
791 return operationResult{}, admissionError(err)
792 }
793 c.pending[operation] = reply
794 if c.kinds == nil {
795 c.kinds = make(map[ports.BrokerOperationID]brokerwire.RegisterMutationKind)
796 }
797 c.kinds[operation] = kind
798 c.mu.Unlock()
799 // No result observed is exactly the outcome-unknown case; a request that
800 // was never launched is recorded as failed instead.
801 outcome := ports.BrokerOutcomeUnknown
802 defer func() {
803 c.releaseOperation(operation)
804 _ = c.conn.CompleteOperation(operation, outcome)
805 }()
806 attempted, err := c.sendContext(ctx, build(operation))
807 if err != nil {
808 if !attempted {
809 // The request definitely did not travel, so a definite failure is
810 // the honest report and the operation is recorded as failed.
811 outcome = ports.BrokerOutcomeFailed
812 return operationResult{}, err
813 }
814 // The wire attempt was launched: the broker may have committed the
815 // operation even though this side observed no result.
816 return operationResult{outcome: ports.BrokerOutcomeUnknown, detail: errorDetail(err)}, nil
817 }
818 select {
819 case result := <-reply:
820 outcome = result.outcome
821 return result, nil
822 case <-ctx.Done():
823 // The operation was already sent: the broker may have committed it.
824 return operationResult{outcome: ports.BrokerOutcomeUnknown, detail: errorDetail(ctx.Err())}, nil
825 case <-c.done:
826 return operationResult{outcome: ports.BrokerOutcomeUnknown, detail: errorDetail(c.terminalError())}, nil
827 }
828}
829
830// send encodes and writes one client frame synchronously.
831func (c *client) send(message brokerwire.ClientMessage) error {
832 payload, err := brokerwire.EncodeClient(message, c.ceilings.MaxReceiveEnvelopeBytes, c.ceilings.StreamChunkLimit)
833 if err != nil {
834 return err
835 }
836 if err := c.transport.Send(wire.Envelope{Payload: payload}); err != nil {
837 return transportFailure(err)
838 }
839 return nil
840}
841
842// sendAsync admits one client frame for ordered background transmission. It
843// reports backpressure as a retryable unavailability rather than blocking a
844// caller that has no context to spend.
845func (c *client) sendAsync(message brokerwire.ClientMessage) error {
846 payload, err := brokerwire.EncodeClient(message, c.ceilings.MaxReceiveEnvelopeBytes, c.ceilings.StreamChunkLimit)
847 if err != nil {
848 return err
849 }
850 if async, ok := c.transport.(wire.AsyncTransport); ok {
851 if err := async.SendAsync(wire.Envelope{Payload: payload}); err != nil {
852 if errors.Is(err, streamframe.ErrBackpressure) {
853 return ports.BrokerError{Code: ports.BrokerErrorUnavailable, Text: "broker IPC send queue is full"}
854 }
855 return transportFailure(err)
856 }
857 return nil
858 }
859 if err := c.transport.Send(wire.Envelope{Payload: payload}); err != nil {
860 return transportFailure(err)
861 }
862 return nil
863}
864
865// sendContext writes one client frame, honoring cancellation while the carriage
866// egress queue is full. It reports whether the wire attempt was launched, which
867// is exactly the dispatch uncertainty a caller needs: an encoding failure, an
868// already-cancelled call, or a connection already terminal before the attempt
869// begins means the frame definitely did not travel, while every failure
870// observed after the attempt is launched means it may have been dispatched and
871// the caller must not replay blindly. The wire attempt is raced against the
872// caller's context and the connection's terminal outcome, so a stalled send
873// never pins the caller: the racing goroutine observes the closed carriage and
874// returns.
875func (c *client) sendContext(ctx context.Context, message brokerwire.ClientMessage) (attempted bool, err error) {
876 payload, err := brokerwire.EncodeClient(message, c.ceilings.MaxReceiveEnvelopeBytes, c.ceilings.StreamChunkLimit)
877 if err != nil {
878 return false, err
879 }
880 // Synchronous preflight. Checking before the goroutine starts is what makes
881 // the not-dispatched claim true: a context that is already done or a
882 // connection that is already terminal can no longer race a launched Send.
883 if err := ctx.Err(); err != nil {
884 return false, err
885 }
886 select {
887 case <-c.done:
888 return false, c.terminalError()
889 default:
890 }
891 done := make(chan error, 1)
892 go func() { done <- c.transport.Send(wire.Envelope{Payload: payload}) }()
893 select {
894 case err := <-done:
895 if err != nil {
896 return true, transportFailure(err)
897 }
898 return true, nil
899 case <-ctx.Done():
900 return true, ctx.Err()
901 case <-c.done:
902 return true, c.terminalError()
903 }
904}
905
906// requestStreamClose tells the broker one stream is retired. It is best-effort:
907// the stream is already settled locally, so a full queue or a dead carriage
908// needs no further action.
909func (c *client) requestStreamClose(stream ports.BrokerStreamID) {
910 _ = c.sendAsync(brokerwire.CloseStream{Epoch: c.scope.Epoch, Connection: c.scope.Connection, Stream: stream})
911}
912
913// retireStream settles one local stream and tells the broker to retire it.
914func (c *client) retireStream(stream ports.BrokerStreamID, cause error) {
915 st := c.lookupStream(stream)
916 if st != nil {
917 st.terminate(cause)
918 }
919 c.requestStreamClose(stream)
920}
921
922// lookupStream returns the local stream for one identity.
923func (c *client) lookupStream(stream ports.BrokerStreamID) *clientStream {
924 c.mu.Lock()
925 defer c.mu.Unlock()
926 return c.streams[stream]
927}
928
929// unregisterStream drops one local stream record.
930func (c *client) unregisterStream(stream ports.BrokerStreamID) {
931 c.mu.Lock()
932 delete(c.streams, stream)
933 c.mu.Unlock()
934}
935
936// terminalErrorLocked returns the terminal cause with the connection lock held.
937func (c *client) terminalErrorLocked() error {
938 if c.err != nil {
939 return c.err
940 }
941 return ErrConnectionClosed
942}
943
944// terminalError returns the terminal cause.
945func (c *client) terminalError() error {
946 c.mu.Lock()
947 defer c.mu.Unlock()
948 return c.terminalErrorLocked()
949}
950
951// closeWith tears the connection down exactly once: cancel, close the carriage
952// (unblocking a blocked read and any frame send), settle every local stream and
953// the active subscription, and wake every waiter. A nil cause is an orderly
954// local close.
955func (c *client) closeWith(cause error) {
956 if c == nil {
957 return
958 }
959 c.closeOnce.Do(func() {
960 c.cancel()
961 _ = c.transport.Close()
962 c.mu.Lock()
963 c.closed = true
964 if cause != nil && c.err == nil {
965 c.err = cause
966 }
967 terminal := c.terminalErrorLocked()
968 sub := c.sub
969 c.sub = nil
970 preview := c.preview
971 c.preview = nil
972 streams := make([]*clientStream, 0, len(c.streams))
973 for _, st := range c.streams {
974 streams = append(streams, st)
975 }
976 c.streams = make(map[ports.BrokerStreamID]*clientStream)
977 c.mu.Unlock()
978 for _, st := range streams {
979 st.terminate(terminal)
980 }
981 if sub != nil {
982 sub.Close()
983 }
984 if preview != nil {
985 preview.Close()
986 }
987 close(c.done)
988 })
989}
990
991// mutationFailure converts one operation outcome into its typed caller error.
992func mutationFailure(result operationResult) error {
993 switch result.outcome {
994 case ports.BrokerOutcomeOK:
995 return nil
996 case ports.BrokerOutcomeFailed:
997 if err := failureFromDetail(result.detail); err != nil {
998 return err
999 }
1000 return ports.BrokerError{Code: ports.BrokerErrorUnavailable}
1001 default:
1002 return ports.BrokerError{Code: ports.BrokerErrorOutcomeUnknown}
1003 }
1004}
1005
1006// newOperationID samples one fresh non-zero operation identity.
1007func newOperationID() (ports.BrokerOperationID, error) {
1008 var id ports.BrokerOperationID
1009 if _, err := rand.Read(id[:]); err != nil {
1010 return ports.BrokerOperationID{}, err
1011 }
1012 if id.IsZero() {
1013 id[15] = 1
1014 }
1015 return id, nil
1016}
1017
1018// clientSubscription is one local capacity-one change notification. Each
1019// subscriber owns its own channel and re-reads the newest snapshot on wake.
1020type clientSubscription struct {
1021 changed chan struct{}
1022 once sync.Once
1023 mu sync.Mutex
1024 closed bool
1025}
1026
1027func newClientSubscription() *clientSubscription {
1028 return &clientSubscription{changed: make(chan struct{}, 1)}
1029}
1030
1031// Changed is signalled once per committed publication.
1032func (s *clientSubscription) Changed() <-chan struct{} { return s.changed }
1033
1034// Close ends the subscription exactly once. The channel is closed under the same
1035// lock notify sends under, so a racing publication can never send on a closed
1036// channel.
1037func (s *clientSubscription) Close() {
1038 if s == nil {
1039 return
1040 }
1041 s.once.Do(func() {
1042 s.mu.Lock()
1043 s.closed = true
1044 close(s.changed)
1045 s.mu.Unlock()
1046 })
1047}
1048
1049// notify coalesces one publication wake.
1050func (s *clientSubscription) notify() {
1051 if s == nil {
1052 return
1053 }
1054 s.mu.Lock()
1055 defer s.mu.Unlock()
1056 if s.closed {
1057 return
1058 }
1059 select {
1060 case s.changed <- struct{}{}:
1061 default:
1062 }
1063}
1064
1065// StaleBuild reports the identity of a connected broker from another build.
1066func StaleBuild(service ports.BrokerService) (string, bool) {
1067 c, ok := service.(*client)
1068 if !ok || c == nil || c.staleBuild == "" {
1069 return "", false
1070 }
1071 return c.staleBuild, true
1072}