1package ports
2
3import (
4 "context"
5 "errors"
6 "fmt"
7 "strings"
8 "unicode"
9 "unicode/utf8"
10
11 "github.com/bnema/vev/internal/domain"
12 "github.com/bnema/vev/internal/protocol"
13 "github.com/bnema/vev/internal/protocol/catalogue"
14)
15
16// Broker semantic contracts.
17//
18// The broker is the machine-wide owner of connectivity: configured hosts,
19// daemon reachability and catalogue snapshots, pooled physical transports,
20// and multiplexed logical streams. It never owns picker UI, navigation
21// decisions, sessions, PTYs, terminal geometry, or attachment semantics.
22//
23// Identity levels stay distinct:
24// - configured host registration/incarnation (domain.RemoteRegistration)
25// - authenticated daemon identity used for pooling (BrokerDaemonIdentity)
26// - daemon process incarnation used to detect restart
27// (BrokerDaemonIncarnation)
28// - exact session lifecycle identity (domain.SessionLifecycleID via
29// protocol.ExactSessionTarget)
30// - broker epoch/revision (BrokerEpoch/BrokerRevision)
31// - client connection identity (BrokerConnectionID)
32// - logical stream identity (BrokerStreamID)
33//
34// An epoch change forces a full resynchronization: stream IDs and events
35// are scoped to one client connection and broker epoch, so stale
36// completions cannot revive an attachment. Tombstones and registration
37// incarnation fence remove/re-add races independently of the epoch.
38//
39// Raw frames, codecs, environment policy enforcement, and worker
40// implementations live outside this package. Per-request session
41// environment (BrokerOpenStreamRequest.Env) stays separate from the broker
42// process environment.
43
44const (
45 // BrokerMaxHosts bounds one snapshot publication. It mirrors the
46 // catalogue host budget so a broker snapshot never exceeds what
47 // discovery can produce.
48 BrokerMaxHosts = catalogue.RemoteCatalogMaxHosts
49 // BrokerMaxSessionsPerHost bounds sessions carried per host in one
50 // snapshot publication.
51 BrokerMaxSessionsPerHost = catalogue.RemoteCatalogMaxSessions
52 // BrokerMaxTombstones bounds retired registrations carried alongside
53 // one snapshot for remove/re-add fencing.
54 BrokerMaxTombstones = catalogue.RemoteCatalogMaxHosts
55 // BrokerMaxErrorBytes bounds human-readable error text in broker
56 // results. The underlying cause stays behind errors.Is/As where the
57 // contract carries an error value.
58 BrokerMaxErrorBytes = 512
59 // BrokerMaxEnvEntries bounds per-request environment entries. The
60 // broker never merges these into its own process environment.
61 BrokerMaxEnvEntries = 128
62 // BrokerMaxEnvEntryBytes bounds one per-request environment entry.
63 BrokerMaxEnvEntryBytes = 4 << 10
64 // BrokerMaxTransportBytes bounds the opaque transport policy token.
65 BrokerMaxTransportBytes = 64
66 // BrokerMaxIdentityBytes bounds the authenticated daemon identity token.
67 BrokerMaxIdentityBytes = 256
68 // BrokerMaxPolicyTokenBytes bounds each authoritative policy identity.
69 BrokerMaxPolicyTokenBytes = 256
70)
71
72// Broker registry lifecycle sentinels. They are typed so a caller can classify
73// why the registry refused a membership mutation without importing the use
74// case or matching on message text.
75var (
76 // ErrBrokerRegistryClosed reports that the registry run has settled: its
77 // durable writer is flushed, in-flight attempts are retired, and no further
78 // membership mutation is acknowledged.
79 ErrBrokerRegistryClosed = errors.New("ports: broker registry is closed")
80 // ErrBrokerRevisionExhausted reports that the broker epoch has consumed its
81 // entire revision series. The registry fails closed: it keeps the last valid
82 // snapshot published and refuses further mutations instead of wrapping to
83 // zero or publishing a revision below the newest one, either of which would
84 // silently freeze durable persistence.
85 ErrBrokerRevisionExhausted = errors.New("ports: broker revision series exhausted")
86)
87
88// BrokerEpoch identifies one broker process incarnation. Every broker
89// process samples a fresh non-zero epoch at startup; every publication
90// carries a revision within that epoch. Clients discard all broker state
91// and resubscribe when the epoch changes.
92type BrokerEpoch uint64
93
94// BrokerRevision is monotonic within one broker epoch. Revisions never
95// compare across epochs.
96type BrokerRevision uint64
97
98// BrokerConnectionID identifies one client connection to the broker,
99// scoped to a broker epoch. It never identifies a session or a stream.
100type BrokerConnectionID [16]byte
101
102// BrokerStreamID identifies one logical stream within a client connection
103// and broker epoch. It never crosses connections or epochs.
104type BrokerStreamID uint64
105
106// BrokerOperationID is the idempotency identity for one mutating broker
107// operation. Clients refresh authoritative state on outcome-unknown and
108// never blindly replay a non-idempotent operation.
109type BrokerOperationID [16]byte
110
111// BrokerDaemonIncarnation identifies one daemon process incarnation and
112// detects restarts behind a stable authenticated identity.
113type BrokerDaemonIncarnation [16]byte
114
115func (id BrokerConnectionID) IsZero() bool { return id == (BrokerConnectionID{}) }
116
117func (id BrokerConnectionID) Validate() error {
118 if id.IsZero() {
119 return errors.New("ports: broker connection ID is zero")
120 }
121 return nil
122}
123
124func (id BrokerStreamID) Validate() error {
125 if id == 0 {
126 return errors.New("ports: broker stream ID is zero")
127 }
128 return nil
129}
130
131func (id BrokerOperationID) IsZero() bool { return id == (BrokerOperationID{}) }
132
133func (id BrokerOperationID) Validate() error {
134 if id.IsZero() {
135 return errors.New("ports: broker operation ID is zero")
136 }
137 return nil
138}
139
140func (id BrokerDaemonIncarnation) IsZero() bool { return id == (BrokerDaemonIncarnation{}) }
141
142func (id BrokerDaemonIncarnation) Validate() error {
143 if id.IsZero() {
144 return errors.New("ports: broker daemon incarnation is zero")
145 }
146 return nil
147}
148
149// BrokerDaemonIdentity is the authenticated daemon/service identity used
150// for physical transport pooling. Display aliases are never pooled: two
151// endpoints share a transport only after authenticated identity and every
152// trust, launch, isolation, and transport policy compare compatible.
153type BrokerDaemonIdentity string
154
155func (id BrokerDaemonIdentity) Validate() error {
156 return validateBrokerToken(string(id), BrokerMaxIdentityBytes, "daemon identity")
157}
158
159// BrokerPolicy is the exact connection policy required before two
160// endpoints may share a pooled physical transport. Pooling requires exact
161// equality: conflicting policy is rejected, never merged or inherited
162// from the first launching client.
163type BrokerPolicy struct {
164 ProtocolVersion uint16
165 CatalogSchemaVersion uint16
166 EnvironmentPolicy protocol.EnvironmentPolicy
167 Transport string
168 // Opaque, authoritative policy identities, not display labels or secrets.
169 Trust string
170 Launch string
171 Isolation string
172}
173
174func (p BrokerPolicy) Validate() error {
175 for _, token := range []struct{ value, label string }{{p.Trust, "policy trust"}, {p.Launch, "policy launch"}, {p.Isolation, "policy isolation"}} {
176 if err := validateBrokerToken(token.value, BrokerMaxPolicyTokenBytes, token.label); err != nil {
177 return err
178 }
179 }
180 if p.ProtocolVersion == 0 {
181 return errors.New("ports: broker policy has no protocol version")
182 }
183 if p.CatalogSchemaVersion == 0 {
184 return errors.New("ports: broker policy has no catalog schema version")
185 }
186 if err := protocol.ValidateEnvironmentPolicy(p.EnvironmentPolicy); err != nil {
187 return fmt.Errorf("ports: broker policy: %w", err)
188 }
189 if err := validateBrokerToken(p.Transport, BrokerMaxTransportBytes, "policy transport"); err != nil {
190 return err
191 }
192 return nil
193}
194
195// Compatible reports whether two validated policies may share a pooled
196// physical transport. It requires exact equality across every field.
197func (p BrokerPolicy) Compatible(other BrokerPolicy) bool {
198 return p == other
199}
200
201// BrokerHostTombstone fences a removed host registration so a stale probe,
202// snapshot, or completion bound to the retired incarnation cannot overwrite
203// or revive current authority. A re-added endpoint carries a new
204// incarnation and is never fenced by an older tombstone.
205type BrokerHostTombstone struct {
206 Endpoint string
207 Registration domain.RemoteRegistration
208 RetiredRevision BrokerRevision
209}
210
211func (t BrokerHostTombstone) Validate() error {
212 if err := domain.ValidateRemoteHostTarget(t.Endpoint); err != nil {
213 return fmt.Errorf("ports: broker tombstone: %w", err)
214 }
215 if err := t.Registration.Validate(); err != nil {
216 return fmt.Errorf("ports: broker tombstone: %w", err)
217 }
218 if t.Endpoint != t.Registration.Endpoint {
219 return errors.New("ports: broker tombstone endpoint does not match registration")
220 }
221 if t.RetiredRevision == 0 {
222 return errors.New("ports: broker tombstone has no retired revision")
223 }
224 return nil
225}
226
227// Fences reports whether this tombstone retires the supplied registration:
228// same endpoint and incarnation, with a retired generation at or past the
229// candidate. A re-added endpoint with a fresh incarnation is never fenced.
230func (t BrokerHostTombstone) Fences(candidate domain.RemoteRegistration) bool {
231 return t.Endpoint == candidate.Endpoint &&
232 t.Registration.Incarnation == candidate.Incarnation &&
233 t.Registration.Generation >= candidate.Generation
234}
235
236// BrokerSnapshot is an immutable, fully defensive broker publication.
237// Daemons carries the broker-native daemon observations:
238// the local daemon entry first when present, then remote hosts in
239// registration order. Removed carries the bounded tombstone set that fences
240// stale authority. Nested slices are never mutated after publication.
241// Tombstones and the local daemon observation are process-local state and
242// are never durable: a BrokerSnapshotStore persists and reloads remote
243// daemon observations only.
244type BrokerSnapshot struct {
245 Epoch BrokerEpoch
246 Revision BrokerRevision
247 Daemons []BrokerDaemonObservation
248 Removed []BrokerHostTombstone
249}
250
251func (s BrokerSnapshot) Validate() error {
252 if s.Epoch == 0 {
253 return errors.New("ports: broker snapshot has no epoch")
254 }
255 if s.Revision == 0 {
256 return errors.New("ports: broker snapshot has no revision")
257 }
258 if len(s.Daemons) > BrokerMaxDaemonsPerSnapshot {
259 return errors.New("ports: broker snapshot has too many daemons")
260 }
261 local := false
262 seen := make(map[string]struct{}, len(s.Daemons))
263 for i, daemon := range s.Daemons {
264 if err := daemon.Validate(); err != nil {
265 return fmt.Errorf("ports: broker snapshot daemon %d: %w", i, err)
266 }
267 if daemon.Local {
268 if local {
269 return errors.New("ports: broker snapshot has more than one local daemon")
270 }
271 // The wire layout carries the local daemon at index zero only, so a
272 // local observation anywhere else is a publication the peer's
273 // assembler refuses. Refuse it at the contract gate instead.
274 if i != 0 {
275 return fmt.Errorf("ports: broker snapshot carries its local daemon at index %d", i)
276 }
277 local = true
278 continue
279 }
280 if _, dup := seen[daemon.Endpoint]; dup {
281 return fmt.Errorf("ports: broker snapshot has duplicate host %q", daemon.Endpoint)
282 }
283 seen[daemon.Endpoint] = struct{}{}
284 }
285 if len(s.Removed) > BrokerMaxTombstones {
286 return errors.New("ports: broker snapshot has too many tombstones")
287 }
288 for _, tombstone := range s.Removed {
289 if err := tombstone.Validate(); err != nil {
290 return err
291 }
292 if _, live := seen[tombstone.Endpoint]; live {
293 return fmt.Errorf("ports: broker snapshot carries live host %q as tombstone", tombstone.Endpoint)
294 }
295 }
296 return nil
297}
298
299// Clone returns a defensive copy with independent daemon, session, and
300// tombstone slices.
301func (s BrokerSnapshot) Clone() BrokerSnapshot {
302 out := s
303 out.Daemons = make([]BrokerDaemonObservation, len(s.Daemons))
304 for i, daemon := range s.Daemons {
305 out.Daemons[i] = daemon.Clone()
306 }
307 out.Removed = append([]BrokerHostTombstone(nil), s.Removed...)
308 return out
309}
310
311// Find returns the observation for an exact remote endpoint identity. The
312// local daemon has no endpoint and is located by its Local flag.
313func (s BrokerSnapshot) Find(endpoint string) (BrokerDaemonObservation, bool) {
314 for _, daemon := range s.Daemons {
315 if !daemon.Local && daemon.Endpoint == endpoint {
316 return daemon, true
317 }
318 }
319 return BrokerDaemonObservation{}, false
320}
321
322// Supersedes reports whether s is newer than other. Any epoch change
323// forces a full resynchronization: a snapshot with a valid epoch
324// supersedes every snapshot from another epoch (or none), regardless of
325// numeric order, because epochs are fresh per broker process and never
326// ordered across processes. Within one epoch the higher revision wins.
327// Revisions never compare across epochs. A zero-epoch snapshot is
328// invalid and never supersedes.
329func (s BrokerSnapshot) Supersedes(other BrokerSnapshot) bool {
330 if s.Epoch == 0 {
331 return false
332 }
333 if s.Epoch != other.Epoch {
334 return true
335 }
336 return s.Revision > other.Revision
337}
338
339// BrokerStreamAdmission is the closed attachment-admission taxonomy:
340// an attachment stream names exactly how the daemon must admit it.
341// Control and observation streams carry no admission.
342type BrokerStreamAdmission uint8
343
344const (
345 // BrokerAdmissionExact attaches to one exact session lifecycle identity;
346 // the daemon revalidates it before attachment.
347 BrokerAdmissionExact BrokerStreamAdmission = iota + 1
348 // BrokerAdmissionCreateNamed creates one named session when the stream
349 // attaches; the name is validated session-name authority.
350 BrokerAdmissionCreateNamed
351 // BrokerAdmissionCreateEphemeral creates one ephemeral session when the
352 // stream attaches.
353 BrokerAdmissionCreateEphemeral
354 // BrokerAdmissionAttachNamed attaches to the session the daemon knows by
355 // this validated name, restoring it if needed. The daemon, not a broker
356 // catalogue, decides existence and refuses an unknown name.
357 BrokerAdmissionAttachNamed
358)
359
360func (a BrokerStreamAdmission) String() string {
361 switch a {
362 case BrokerAdmissionExact:
363 return "exact"
364 case BrokerAdmissionCreateNamed:
365 return "create_named"
366 case BrokerAdmissionCreateEphemeral:
367 return "create_ephemeral"
368 case BrokerAdmissionAttachNamed:
369 return "attach_named"
370 default:
371 return fmt.Sprintf("invalid(%d)", uint8(a))
372 }
373}
374
375func (a BrokerStreamAdmission) Validate() error {
376 switch a {
377 case BrokerAdmissionExact, BrokerAdmissionCreateNamed, BrokerAdmissionCreateEphemeral, BrokerAdmissionAttachNamed:
378 return nil
379 default:
380 return errors.New("ports: invalid broker stream admission")
381 }
382}
383
384// BrokerOpenStreamRequest asks the broker to open one independently
385// cancellable logical stream to the owning daemon. The daemon revalidates
386// the exact session identity before attachment. Admission selects the
387// attachment admission variant; Name is the validated session name for
388// BrokerAdmissionCreateNamed and BrokerAdmissionAttachNamed and is empty for
389// every other variant. StartMode
390// is the explicit daemon-start authorization this request carries to the
391// transport; Pending acquisition keys on it so an ExistingOnly request is
392// never coalesced with a dial that may start the target. Env is the
393// per-request session environment; it is never inherited from the broker
394// process environment.
395type BrokerOpenStreamRequest struct {
396 Epoch BrokerEpoch
397 Purpose BrokerStreamPurpose
398 Admission BrokerStreamAdmission
399 Name string
400 Local bool
401 Connection BrokerConnectionID
402 Stream BrokerStreamID
403 Endpoint string
404 Registration domain.RemoteRegistration
405 Target protocol.ExactSessionTarget
406 Env []string
407 Policy BrokerPolicy
408 StartMode BrokerDaemonStartMode
409}
410
411func (r BrokerOpenStreamRequest) Validate() error {
412 if r.Epoch == 0 {
413 return errors.New("ports: missing broker epoch")
414 }
415 if err := r.Connection.Validate(); err != nil {
416 return err
417 }
418 if err := r.Stream.Validate(); err != nil {
419 return err
420 }
421 if r.Local {
422 if r.Endpoint != "" || r.Registration != (domain.RemoteRegistration{}) {
423 return errors.New("ports: local endpoint carries remote registration")
424 }
425 } else {
426 if err := r.Registration.Validate(); err != nil {
427 return err
428 }
429 if r.Endpoint != r.Registration.Endpoint {
430 return errors.New("ports: endpoint registration mismatch")
431 }
432 }
433 switch r.Purpose {
434 case BrokerStreamAttachment:
435 // An attachment or creation may need to start the target daemon, so an
436 // ExistingOnly authorization is incompatible with it: the mode is
437 // refused rather than silently widened into a spawn.
438 if r.StartMode != BrokerDaemonStartIfNeeded {
439 return errors.New("ports: attachment requires start-if-needed authorization")
440 }
441 if err := r.Admission.Validate(); err != nil {
442 return err
443 }
444 switch r.Admission {
445 case BrokerAdmissionExact:
446 if err := r.Target.Validate(); err != nil {
447 return err
448 }
449 if r.Name != "" {
450 return errors.New("ports: exact admission carries a creation name")
451 }
452 case BrokerAdmissionCreateNamed, BrokerAdmissionAttachNamed:
453 if r.Target != (protocol.ExactSessionTarget{}) {
454 return errors.New("ports: named admission carries an exact target")
455 }
456 if err := domain.ValidateSessionName(r.Name); err != nil {
457 return fmt.Errorf("ports: invalid admission session name: %w", err)
458 }
459 case BrokerAdmissionCreateEphemeral:
460 if r.Target != (protocol.ExactSessionTarget{}) {
461 return errors.New("ports: ephemeral creation carries an exact target")
462 }
463 if r.Name != "" {
464 return errors.New("ports: ephemeral creation carries a session name")
465 }
466 }
467 case BrokerStreamControl, BrokerStreamObservation:
468 if r.Admission != 0 || r.Name != "" || r.Target != (protocol.ExactSessionTarget{}) || len(r.Env) != 0 {
469 return errors.New("ports: control/observation carries attachment state")
470 }
471 if r.Purpose == BrokerStreamObservation && r.StartMode != BrokerDaemonExistingOnly {
472 // Observation never starts a target: an observation stream that
473 // authorized a spawn is refused rather than silently narrowed, so a
474 // caller can always trust that observing cannot start a daemon.
475 return errors.New("ports: observation requires existing-only start mode")
476 }
477 default:
478 return errors.New("ports: invalid stream purpose")
479 }
480 if err := r.StartMode.Validate(); err != nil {
481 return err
482 }
483 if uint64(len(r.Env)) > BrokerMaxEnvEntries {
484 return errors.New("ports: broker open stream has too many environment entries")
485 }
486 for _, entry := range r.Env {
487 if len(entry) > BrokerMaxEnvEntryBytes {
488 return errors.New("ports: broker open stream environment entry too long")
489 }
490 if !utf8.ValidString(entry) {
491 return errors.New("ports: broker open stream environment is not valid UTF-8")
492 }
493 }
494 if err := r.Policy.Validate(); err != nil {
495 return err
496 }
497 return nil
498}
499
500// BrokerMutationOutcome is the closed outcome taxonomy for a mutating
501// broker operation. OutcomeUnknown means connectivity was lost after the
502// mutation may have committed but before its result arrived: the client
503// refreshes authoritative state and never blindly replays a
504// non-idempotent operation.
505type BrokerMutationOutcome uint8
506
507const (
508 BrokerOutcomeOK BrokerMutationOutcome = iota + 1
509 BrokerOutcomeFailed
510 BrokerOutcomeUnknown
511)
512
513func (o BrokerMutationOutcome) String() string {
514 switch o {
515 case BrokerOutcomeOK:
516 return "ok"
517 case BrokerOutcomeFailed:
518 return "failed"
519 case BrokerOutcomeUnknown:
520 return "outcome_unknown"
521 default:
522 return "unknown"
523 }
524}
525
526func (o BrokerMutationOutcome) Validate() error {
527 switch o {
528 case BrokerOutcomeOK, BrokerOutcomeFailed, BrokerOutcomeUnknown:
529 return nil
530 default:
531 return errors.New("ports: invalid broker mutation outcome")
532 }
533}
534
535// BrokerOperationResult is exactly one completion for one admitted mutating
536// operation. Text is bounded and presentation-safe; the outcome decides
537// whether the client may retry, must refresh, or must not replay.
538type BrokerOperationResult struct {
539 Operation BrokerOperationID
540 Outcome BrokerMutationOutcome
541 Text string
542}
543
544func (r BrokerOperationResult) Validate() error {
545 if err := r.Operation.Validate(); err != nil {
546 return err
547 }
548 if err := r.Outcome.Validate(); err != nil {
549 return err
550 }
551 if err := validateBrokerDisplayText(r.Text, BrokerMaxErrorBytes, "operation result text"); err != nil {
552 return err
553 }
554 if r.Outcome == BrokerOutcomeOK && r.Text != "" {
555 return errors.New("ports: successful broker operation carries no text")
556 }
557 return nil
558}
559
560// BrokerErrorCode is the closed taxonomy for broker and stream lifecycle
561// failures surfaced to clients. Every failure returns the client to the
562// picker in the same process with a visible typed error; none terminates
563// the client except an explicit exit or a fatal terminal failure.
564type BrokerErrorCode uint8
565
566const (
567 BrokerErrorUnavailable BrokerErrorCode = iota + 1
568 BrokerErrorIncompatible
569 BrokerErrorTimeout
570 BrokerErrorCancelled
571 BrokerErrorOutcomeUnknown
572 BrokerErrorAttachmentLost
573 BrokerErrorStaleEpoch
574 BrokerErrorConflictingPolicy
575 BrokerErrorExplicitExit
576 BrokerErrorFatalTerminal
577 BrokerErrorHostConflict
578 BrokerErrorMembershipImmutable
579 BrokerErrorNoDaemon
580)
581
582func (c BrokerErrorCode) String() string {
583 switch c {
584 case BrokerErrorUnavailable:
585 return "unavailable"
586 case BrokerErrorIncompatible:
587 return "incompatible"
588 case BrokerErrorTimeout:
589 return "timeout"
590 case BrokerErrorCancelled:
591 return "cancelled"
592 case BrokerErrorOutcomeUnknown:
593 return "outcome_unknown"
594 case BrokerErrorAttachmentLost:
595 return "attachment_lost"
596 case BrokerErrorStaleEpoch:
597 return "stale_epoch"
598 case BrokerErrorConflictingPolicy:
599 return "conflicting_policy"
600 case BrokerErrorExplicitExit:
601 return "explicit_exit"
602 case BrokerErrorFatalTerminal:
603 return "fatal_terminal"
604 case BrokerErrorHostConflict:
605 return "host_conflict"
606 case BrokerErrorMembershipImmutable:
607 return "membership_immutable"
608 case BrokerErrorNoDaemon:
609 return "no_daemon"
610 default:
611 return "unknown"
612 }
613}
614
615func (c BrokerErrorCode) Validate() error {
616 switch c {
617 case BrokerErrorUnavailable, BrokerErrorIncompatible, BrokerErrorTimeout,
618 BrokerErrorCancelled, BrokerErrorOutcomeUnknown, BrokerErrorAttachmentLost,
619 BrokerErrorStaleEpoch, BrokerErrorConflictingPolicy,
620 BrokerErrorExplicitExit, BrokerErrorFatalTerminal,
621 BrokerErrorHostConflict, BrokerErrorMembershipImmutable, BrokerErrorNoDaemon:
622 return nil
623 default:
624 return errors.New("ports: invalid broker error code")
625 }
626}
627
628// BrokerError is the typed lifecycle failure carried to clients. Terminal
629// reports whether the client process itself must exit: only an explicit
630// exit or a fatal terminal failure is terminal; every other code returns
631// to the picker with a visible notification.
632type BrokerError struct {
633 Code BrokerErrorCode
634 Text string
635 // Cause is local-only diagnostic authority, never serialized as display text.
636 Cause error
637}
638
639func (e BrokerError) Error() string {
640 if e.Text == "" {
641 return "broker " + e.Code.String()
642 }
643 return "broker " + e.Code.String() + ": " + e.Text
644}
645
646func (e BrokerError) Unwrap() error { return e.Cause }
647
648func (e BrokerError) Validate() error {
649 if err := e.Code.Validate(); err != nil {
650 return err
651 }
652 return validateBrokerDisplayText(e.Text, BrokerMaxErrorBytes, "error text")
653}
654
655// Terminal reports whether this failure ends the client process.
656func (e BrokerError) Terminal() bool {
657 return e.Code == BrokerErrorExplicitExit || e.Code == BrokerErrorFatalTerminal
658}
659
660// BrokerStreamLost is the independent typed loss event delivered per
661// affected logical stream when its physical connection fails. One stream
662// failure never closes unrelated streams; a physical loss fans out one
663// event per stream.
664type BrokerStreamLost struct {
665 Connection BrokerConnectionID
666 Stream BrokerStreamID
667 Epoch BrokerEpoch
668 Cause domain.RemoteFailureKind
669 Err error
670}
671
672// Error and Unwrap retain exact stream scope while allowing generic lifecycle handling.
673func (e BrokerStreamLost) Error() string { return "broker attachment_lost" }
674func (e BrokerStreamLost) Unwrap() error {
675 return BrokerError{Code: BrokerErrorAttachmentLost, Cause: e.Err}
676}
677
678func (e BrokerStreamLost) Validate() error {
679 if e.Cause < domain.RemoteFailureTransport || e.Cause > domain.RemoteFailureInvalidResponse {
680 return errors.New("ports: broker stream loss has invalid cause")
681 }
682 if err := e.Connection.Validate(); err != nil {
683 return err
684 }
685 if err := e.Stream.Validate(); err != nil {
686 return err
687 }
688 if e.Epoch == 0 {
689 return errors.New("ports: broker stream loss has no epoch")
690 }
691 return nil
692}
693
694// BrokerSubscription wakes one subscriber when a newer snapshot is
695// available. Each subscriber owns a capacity-one channel and re-reads the
696// newest snapshot on wake; channels are never shared between subscribers.
697type BrokerSubscription interface {
698 Changed() <-chan struct{}
699 Close()
700}
701
702// BrokerPreviewGeneration identifies one selected-row preview authority within
703// a client connection. It is strictly monotone for that connection and never
704// crosses broker epochs.
705type BrokerPreviewGeneration uint64
706
707func (g BrokerPreviewGeneration) Validate() error {
708 if g == 0 {
709 return errors.New("ports: broker preview generation is zero")
710 }
711 return nil
712}
713
714// BrokerPreviewRoute names the daemon one preview observes. The broker
715// allocates the observation stream itself, so the route carries no stream
716// identity: the local flag or the exact remote registration, plus the policy
717// the observed daemon connection must be compatible with.
718type BrokerPreviewRoute struct {
719 Local bool
720 Endpoint string
721 Registration domain.RemoteRegistration
722 Policy BrokerPolicy
723}
724
725func (r BrokerPreviewRoute) Validate() error {
726 if r.Local {
727 if r.Endpoint != "" || r.Registration != (domain.RemoteRegistration{}) {
728 return errors.New("ports: local preview route carries remote authority")
729 }
730 return nil
731 }
732 if err := r.Registration.Validate(); err != nil {
733 return err
734 }
735 if r.Endpoint != r.Registration.Endpoint {
736 return errors.New("ports: preview route endpoint registration mismatch")
737 }
738 return nil
739}
740
741// BrokerPreviewLocalEndpoint is the canonical preview target endpoint of the
742// local daemon.
743const BrokerPreviewLocalEndpoint = "local"
744
745// BrokerPreviewRequest starts or replaces one connection-scoped live preview.
746// Route names the observed daemon; Preview carries the exact session/tab
747// target and bounded viewport dimensions sent to it.
748type BrokerPreviewRequest struct {
749 Epoch BrokerEpoch
750 Connection BrokerConnectionID
751 Generation BrokerPreviewGeneration
752 Route BrokerPreviewRoute
753 Preview protocol.RemotePreviewRequest
754}
755
756func (r BrokerPreviewRequest) Validate() error {
757 if r.Epoch == 0 {
758 return errors.New("ports: broker preview has no epoch")
759 }
760 if err := r.Connection.Validate(); err != nil {
761 return err
762 }
763 if err := r.Generation.Validate(); err != nil {
764 return err
765 }
766 if err := r.Route.Validate(); err != nil {
767 return fmt.Errorf("ports: broker preview route: %w", err)
768 }
769 if err := protocol.ValidateRemotePreviewRequest(r.Preview); err != nil {
770 return fmt.Errorf("ports: broker preview request: %w", err)
771 }
772 if r.Route.Local != (r.Preview.Target.Endpoint == BrokerPreviewLocalEndpoint) ||
773 (!r.Route.Local && r.Preview.Target.Endpoint != r.Route.Endpoint) {
774 return errors.New("ports: broker preview route does not own the target")
775 }
776 return nil
777}
778
779// BrokerPreviewPublication is the newest result for one preview subscription.
780// Every authority field is repeated so a client can reject a late completion.
781type BrokerPreviewPublication struct {
782 Epoch BrokerEpoch
783 Connection BrokerConnectionID
784 Generation BrokerPreviewGeneration
785 Target domain.RemoteSessionTarget
786 Width uint16
787 Height uint16
788 Preview protocol.RemotePreview
789 Err error
790}
791
792func (p BrokerPreviewPublication) Validate() error {
793 if p.Epoch == 0 {
794 return errors.New("ports: broker preview publication has no epoch")
795 }
796 if err := p.Connection.Validate(); err != nil {
797 return err
798 }
799 if err := p.Generation.Validate(); err != nil {
800 return err
801 }
802 if p.Target.Validate() != nil || p.Target.Stopped {
803 return errors.New("ports: broker preview publication has invalid target")
804 }
805 if p.Width == 0 || p.Height == 0 || int(p.Width) > protocol.RemotePreviewMaxWidth || int(p.Height) > protocol.RemotePreviewMaxHeight {
806 return errors.New("ports: broker preview publication has invalid dimensions")
807 }
808 if p.Err != nil {
809 if p.Preview.Version != 0 || p.Preview.Status != 0 || p.Preview.LifecycleID != (domain.SessionLifecycleID{}) || p.Preview.TabID != "" || p.Preview.Revision != 0 || p.Preview.Width != 0 || p.Preview.Height != 0 || len(p.Preview.Cells) != 0 {
810 return errors.New("ports: failed broker preview publication carries a preview")
811 }
812 return nil
813 }
814 if err := protocol.ValidateRemotePreview(p.Preview); err != nil {
815 return fmt.Errorf("ports: broker preview publication: %w", err)
816 }
817 return nil
818}
819
820// BrokerPreviewSubscription owns one client-scoped selected-row preview. Changed
821// is capacity one; callers re-read Latest after every wake. Close cancels any
822// in-flight logical observation and is idempotent.
823type BrokerPreviewSubscription interface {
824 Changed() <-chan struct{}
825 Latest() BrokerPreviewPublication
826 Close()
827}
828
829// BrokerConnection is one client connection's identity and lifetime.
830type BrokerConnection interface {
831 // ConnectionID returns the ID assigned when this client connection was
832 // accepted. Clients carry it on stream operations to fence stale requests.
833 ConnectionID() BrokerConnectionID
834 // NextStreamID allocates this connection's next logical stream identity. It
835 // is thread-safe, strictly monotone, and never zero, performs no I/O, and
836 // refuses a closed connection or an exhausted counter rather than wrapping.
837 // The calling service is the only allocator on its own connection, and an
838 // allocated identity is consumed even when the open it names is refused.
839 NextStreamID() (BrokerStreamID, error)
840 // Done closes exactly once when this connection is terminal: the broker
841 // retired it, the carriage failed, or the client closed it locally. Err is
842 // stable afterwards. A client observes Done/Err to attribute a broker loss
843 // to the connection even while it holds no logical stream.
844 Done() <-chan struct{}
845 // Err returns the terminal cause, stable once Done is closed. It is nil for
846 // an orderly local Close and non-nil for a broker-side loss or failure.
847 Err() error
848 Close() error
849}
850
851// BrokerObservation reads committed broker state. Snapshot access performs no
852// I/O; RequestReconcile is a non-blocking coalesced hint.
853type BrokerObservation interface {
854 Snapshot() BrokerSnapshot
855 Subscribe() (BrokerSubscription, error)
856 SubscribePreview(request BrokerPreviewRequest) (BrokerPreviewSubscription, error)
857 RequestReconcile(endpoint string)
858}
859
860// BrokerHostMembership mutates durable configured-host membership. Every
861// mutation is bounded by its context.
862type BrokerHostMembership interface {
863 // AddHost durably adds or pins endpoint under policy and returns its authority.
864 AddHost(ctx context.Context, endpoint string, policy BrokerPolicy) (domain.RemoteRegistration, error)
865 // RemoveHost removes only the exact expected registration.
866 RemoveHost(ctx context.Context, expected domain.RemoteRegistration) (bool, error)
867 // UpdateHostPolicy changes policy only for exact expected authority and returns the advanced registration.
868 UpdateHostPolicy(ctx context.Context, expected domain.RemoteRegistration, policy BrokerPolicy) (domain.RemoteRegistration, error)
869}
870
871// BrokerSession is the complete connection-scoped broker surface shared by
872// the client IPC connection and the broker-side admitted connection.
873type BrokerSession interface {
874 BrokerConnection
875 BrokerObservation
876 BrokerHostMembership
877 CloseStream(connection BrokerConnectionID, stream BrokerStreamID) error
878}
879
880// BrokerService is the client-facing broker façade. All clients reach local
881// and remote daemons through it, including the local daemon.
882type BrokerService interface {
883 BrokerSession
884 OpenStream(ctx context.Context, request BrokerOpenStreamRequest) (BrokerLogicalConnection, error)
885}
886
887// BrokerNavigator is what a session frontend needs from the broker: its
888// connection, committed observations, and logical streams. It never mutates
889// host membership. Every BrokerService is a BrokerNavigator.
890type BrokerNavigator interface {
891 BrokerConnection
892 BrokerObservation
893 OpenStream(ctx context.Context, request BrokerOpenStreamRequest) (BrokerLogicalConnection, error)
894}
895
896var _ BrokerNavigator = (BrokerService)(nil)
897
898// BrokerEnvelopeStream carries opaque complete session envelopes across the core.
899type BrokerEnvelopeStream interface {
900 SendEnvelope([]byte) error
901 RecvEnvelope() ([]byte, error)
902 Done() <-chan struct{}
903 Err() error
904 Close() error
905}
906
907// BrokerCoreService is one admitted connection on the broker side.
908type BrokerCoreService interface {
909 BrokerSession
910 OpenEnvelopeStream(ctx context.Context, request BrokerOpenStreamRequest) (BrokerEnvelopeStream, error)
911}
912
913// SessionCodec adapts a raw core stream for broker-owned typed operations.
914type SessionCodec interface {
915 Client(BrokerEnvelopeStream) ClientConnection
916}
917
918// BrokerSnapshotStore persists the durable broker snapshot independently
919// from session state. The broker is its sole writer; session persistence
920// stays daemon-owned and is never moved here. Load is called once while the
921// broker is constructed; Store runs on the broker's serialized writer
922// goroutine. Store must return promptly and must never block indefinitely:
923// this seam deliberately carries no cancellation context, and broker
924// shutdown waits for the in-flight write to finish, so a Store that blocks
925// without bound blocks shutdown. Implementations must bound their own write
926// time and return an error rather than wait forever.
927type BrokerSnapshotStore interface {
928 Load() (BrokerSnapshot, error)
929 Store(BrokerSnapshot) error
930}
931
932// BrokerHostProbe observes one exact host registration without creating an
933// attachment. The probe reports only observed state in a
934// BrokerDaemonObservation: the observed daemon identity, incarnation,
935// protocol version, capabilities, availability, failure, freshness, and
936// session inventory. Configured authority (endpoint, registration, policy,
937// display origin, rank) is stamped by the registry from its own records and
938// never trusted from the probe; unknown identity, version, or inventory is
939// zero and is never invented. The returned observation must satisfy the
940// durable projection rules (see ValidateDurableHostProjection): the registry
941// re-checks exactly those rules, so an over-bound or catalogue-invalid session
942// inventory is classified as an invalid response and never published.
943// Implementations should therefore validate the observation with
944// catalogue.ValidateRemoteCatalog against the RemoteCatalogMax* bounds
945// before returning it rather than relying on that rejection. Probe must
946// return promptly once ctx is cancelled: the registry cancels the attempt
947// context when a registration is replaced or removed, and when the run
948// settles. A probe that ignores cancellation delays that retirement and can
949// overlap a replacement attempt for the same endpoint. Stale completions
950// that arrive anyway are fenced by epoch and registration identity before
951// publication.
952type BrokerHostProbe interface {
953 Probe(ctx context.Context, registration domain.RemoteRegistration) (BrokerDaemonObservation, error)
954}
955
956// BrokerPhysicalConnection is one pooled physical transport to an
957// authenticated daemon identity. It opens one independently cancellable
958// typed session connection per logical stream and never shares stream
959// state, identity, geometry, input, output, cancellation, or application
960// errors between streams.
961type BrokerPhysicalConnection interface {
962 // Done is the physical terminal authority, independent of reads. On
963 // physical failure, publish Err and FailureKind and close Done BEFORE
964 // terminating affected logical streams. Both results are stable after Done.
965 // FailureKind is nonzero for failure; None is allowed only for local Close.
966 Done() <-chan struct{}
967 Err() error
968 FailureKind() domain.RemoteFailureKind
969 Identity() BrokerDaemonIdentity
970 Incarnation() BrokerDaemonIncarnation
971 Policy() BrokerPolicy
972 OpenStream(ctx context.Context, request BrokerOpenStreamRequest) (BrokerEnvelopeStream, error)
973 Close() error
974}
975
976// BrokerEndpointConnector establishes the pooled physical transport for
977// one endpoint under an exact compatible policy. Conflicting policy is
978// rejected; the first launching client never becomes policy authority.
979type BrokerEndpointConnector interface {
980 Connect(ctx context.Context, target BrokerDialTarget) (BrokerPhysicalConnection, error)
981}
982
983// BrokerConnector establishes the client-facing broker connection: it connects
984// to the per-user broker endpoint and returns the connection-scoped
985// BrokerService for exactly that accepted connection. It is the client
986// process's admission seam, distinct from BrokerEndpointConnector (which opens
987// a pooled physical transport for one endpoint inside the broker) and from
988// BrokerAuthority (which admits a listener-side accepted connection).
989//
990// Connect must honor cancellation: a cancelled connect returns promptly and
991// leaves no service behind, so a client that abandons an attempt never leaks a
992// connection. Each successful Connect is independent and the caller owns Close
993// on the returned service; a connector retains no connection state of its own.
994// Semantic rejection is reported as a typed BrokerError and cancellation as a
995// context error.
996//
997// A successful service is independent of the setup context: Connect must be
998// given a context that bounds only the setup, and once Connect returns a
999// service the caller may cancel or let that context expire without closing,
1000// interrupting, or otherwise disturbing the established connection. The
1001// returned service's lifetime is bounded solely by its own Close and by the
1002// broker, so a caller that abandons the setup context (a bounded attempt) still
1003// owns a live connection it must Close.
1004type BrokerConnector interface {
1005 Connect(ctx context.Context) (BrokerNavigator, error)
1006}
1007
1008// BrokerAuthority admits one accepted client connection to the broker core
1009// and returns the BrokerService scoped to exactly that connection. It is the
1010// admission seam between the broker listener adapter and the broker use case:
1011// the listener owns the carriage and calls AdmitClient once per accepted
1012// connection, the implementation assigns the connection identity, and the
1013// returned service owns that connection until Close. The adapter never
1014// constructs a broker itself.
1015//
1016// ctx bounds admission only. It carries the listener's handshake deadline and
1017// is canceled as soon as admission returns or when the listener closes, so an
1018// implementation must not retain it or start work that outlives admission
1019// from it; honoring that cancellation is what lets a listener drop a client
1020// parked in admission immediately. The admitted service gets its own
1021// connection-lived context and must not derive from this one.
1022type BrokerAuthority interface {
1023 AdmitClient(ctx context.Context) (BrokerCoreService, error)
1024}
1025
1026// BrokerListener accepts client connections to the per-user broker
1027// endpoint. Each Accept returns a BrokerService bound to exactly one
1028// accepted client connection: the implementation assigns that
1029// connection's BrokerConnectionID at accept, ends the connection on
1030// Close, and rejects requests carrying a mismatched Connection as stale.
1031// Implementations own the owner-only directory/socket, peer credential
1032// checks, bounded clients and queues, and cancellation-safe accept loops.
1033type BrokerListener interface {
1034 Accept() (BrokerCoreService, error)
1035 Close() error
1036 Addr() string
1037}
1038
1039func validateBrokerToken(value string, maxBytes int, label string) error {
1040 if value == "" {
1041 return fmt.Errorf("ports: broker %s is empty", label)
1042 }
1043 if len(value) > maxBytes {
1044 return fmt.Errorf("ports: broker %s too long", label)
1045 }
1046 if !utf8.ValidString(value) {
1047 return fmt.Errorf("ports: broker %s is not valid UTF-8", label)
1048 }
1049 if strings.TrimSpace(value) != value {
1050 return fmt.Errorf("ports: broker %s has surrounding whitespace", label)
1051 }
1052 for _, r := range value {
1053 if unicode.IsControl(r) || unicode.IsSpace(r) || unicode.Is(unicode.Bidi_Control, r) || r == '\u2028' || r == '\u2029' {
1054 return fmt.Errorf("ports: broker %s contains disallowed characters", label)
1055 }
1056 }
1057 return nil
1058}
1059
1060func validateBrokerDisplayText(value string, maxBytes int, label string) error {
1061 if len(value) > maxBytes {
1062 return fmt.Errorf("ports: broker %s too long", label)
1063 }
1064 if !utf8.ValidString(value) {
1065 return fmt.Errorf("ports: broker %s is not valid UTF-8", label)
1066 }
1067 for _, r := range value {
1068 if !brokerDisplayRuneAllowed(r) {
1069 return fmt.Errorf("ports: broker %s contains disallowed characters", label)
1070 }
1071 }
1072 return nil
1073}
1074
1075// brokerDisplayRuneAllowed reports whether one rune may appear in broker
1076// display text. Controls, bidi controls, and the Unicode line separators are
1077// refused: none of them carries presentation value, and each lets a value
1078// reorder or split the text rendered around it.
1079func brokerDisplayRuneAllowed(r rune) bool {
1080 return !unicode.IsControl(r) && !unicode.Is(unicode.Bidi_Control, r) && r != '\u2028' && r != '\u2029'
1081}
1082
1083// SanitizeBrokerDisplayText returns value with every rune broker display text
1084// refuses removed, and always as valid UTF-8. Producers apply it to the
1085// presentation hints they derive rather than receive: a hint derived from
1086// routing authority is not itself authority, so a configured target the
1087// routing validator accepts still yields text the display rule accepts instead
1088// of a publication the wire refuses. The result is not length-checked; the
1089// caller applies the bound for its own field.
1090func SanitizeBrokerDisplayText(value string) string {
1091 if utf8.ValidString(value) && strings.IndexFunc(value, func(r rune) bool { return !brokerDisplayRuneAllowed(r) }) < 0 {
1092 return value
1093 }
1094 var safe strings.Builder
1095 safe.Grow(len(value))
1096 for _, r := range value {
1097 if brokerDisplayRuneAllowed(r) {
1098 safe.WriteRune(r)
1099 }
1100 }
1101 return safe.String()
1102}