1package brokerwire
2
3import (
4 "errors"
5 "time"
6
7 "github.com/bnema/vev/internal/domain"
8 "github.com/bnema/vev/internal/ports"
9 "github.com/bnema/vev/internal/protocol"
10 "github.com/bnema/vev/internal/protocol/catalogue"
11)
12
13var (
14 // ErrWrongDirection reports a broker message presented to the wrong
15 // directional codec. Direction is a closed oneof property: a client
16 // envelope never decodes as server output and vice versa.
17 ErrWrongDirection = errors.New("brokerwire: wrong message direction")
18 // ErrInvalidMessage reports a broker message that fails semantic
19 // validation: bad identity, text, taxonomy, bound, time, environment,
20 // policy, chunk, or snapshot part.
21 ErrInvalidMessage = errors.New("brokerwire: invalid message")
22 // ErrTooLarge reports a stateless bound refusal established by limits.go:
23 // an envelope, chunk, environment, policy token, error text, snapshot
24 // part, or catalogue value above its ceiling.
25 ErrTooLarge = errors.New("brokerwire: message exceeds bound")
26 errConvertRange = errors.New("brokerwire: wire value out of semantic range")
27)
28
29// ClientMessage is the closed set of client-to-broker messages. Every
30// variant converts to exactly one BrokerClientEnvelope payload.
31type ClientMessage interface{ brokerClientMessage() }
32
33// ServerMessage is the closed set of broker-to-client messages. Every
34// variant converts to exactly one BrokerServerEnvelope payload.
35type ServerMessage interface{ brokerServerMessage() }
36
37// Register opens one client connection to the broker.
38type Register struct{ Build string }
39
40func (Register) brokerClientMessage() {}
41
42// Subscribe asks for snapshot and lifecycle publications for one scope.
43type Subscribe struct {
44 Epoch ports.BrokerEpoch
45 Connection ports.BrokerConnectionID
46 Generation SubscriptionGeneration
47 // Passive marks a read-only snapshot reader that must never count as
48 // observation demand.
49 Passive bool
50}
51
52func (Subscribe) brokerClientMessage() {}
53
54// Resync asks for a full snapshot publication after an epoch change or a
55// gap in the client's subscription series.
56type Resync struct {
57 Epoch ports.BrokerEpoch
58 Connection ports.BrokerConnectionID
59 Generation SubscriptionGeneration
60}
61
62func (Resync) brokerClientMessage() {}
63
64// Unsubscribe stops publications for one scope.
65type Unsubscribe struct {
66 Epoch ports.BrokerEpoch
67 Connection ports.BrokerConnectionID
68 Generation SubscriptionGeneration
69}
70
71func (Unsubscribe) brokerClientMessage() {}
72
73// AddHost adds one configured host registration under an exact policy.
74// Operation is the idempotency identity of this mutating operation. Policy
75// is required authority: an absent or invalid policy is refused rather than
76// inherited, so a membership addition can never be authorized by a zero
77// policy.
78type AddHost struct {
79 Epoch ports.BrokerEpoch
80 Connection ports.BrokerConnectionID
81 Operation ports.BrokerOperationID
82 Endpoint string
83 Policy ports.BrokerPolicy
84}
85
86func (AddHost) brokerClientMessage() {}
87
88// RemoveHost removes exactly one configured host registration: the request
89// carries the full expected authority (endpoint, incarnation, and
90// generation), never a bare endpoint name, so a stale removal can never
91// retire a re-added host.
92type RemoveHost struct {
93 Epoch ports.BrokerEpoch
94 Connection ports.BrokerConnectionID
95 Operation ports.BrokerOperationID
96 Registration domain.RemoteRegistration
97}
98
99func (RemoveHost) brokerClientMessage() {}
100
101// UpdateHostPolicy replaces the policy of exactly one configured
102// registration. Registration is the exact expected authority and Policy is
103// the required replacement, so an update can neither target an endpoint
104// name nor be authorized by a zero policy.
105type UpdateHostPolicy struct {
106 Epoch ports.BrokerEpoch
107 Connection ports.BrokerConnectionID
108 Operation ports.BrokerOperationID
109 Registration domain.RemoteRegistration
110 Policy ports.BrokerPolicy
111}
112
113func (UpdateHostPolicy) brokerClientMessage() {}
114
115// Reconcile hands the broker one exact registration to probe.
116type Reconcile struct {
117 Epoch ports.BrokerEpoch
118 Connection ports.BrokerConnectionID
119 Registration domain.RemoteRegistration
120}
121
122func (Reconcile) brokerClientMessage() {}
123
124// OpenStream asks the broker to open one independently cancellable logical
125// stream to the owning daemon. Env is the per-request session environment
126// and is never inherited from the broker process environment. Admission
127// selects the attachment admission variant; Name is the validated session
128// name for create-named and is empty otherwise. StartMode is the explicit
129// daemon-start authorization carried to the transport; it is never zero on an
130// admitted request.
131type OpenStream struct {
132 Epoch ports.BrokerEpoch
133 Connection ports.BrokerConnectionID
134 Stream ports.BrokerStreamID
135 Purpose ports.BrokerStreamPurpose
136 Admission ports.BrokerStreamAdmission
137 Name string
138 Local bool
139 Endpoint string
140 Registration domain.RemoteRegistration
141 Target protocol.ExactSessionTarget
142 Env []string
143 Policy ports.BrokerPolicy
144 StartMode ports.BrokerDaemonStartMode
145}
146
147func (OpenStream) brokerClientMessage() {}
148
149// ClientStreamData carries one opaque client-to-daemon stream frame.
150type ClientStreamData struct {
151 Epoch ports.BrokerEpoch
152 Connection ports.BrokerConnectionID
153 Stream ports.BrokerStreamID
154 Data []byte
155}
156
157func (ClientStreamData) brokerClientMessage() {}
158
159// CloseStream closes one logical stream.
160type CloseStream struct {
161 Epoch ports.BrokerEpoch
162 Connection ports.BrokerConnectionID
163 Stream ports.BrokerStreamID
164}
165
166func (CloseStream) brokerClientMessage() {}
167
168// StreamWindowUpdate grants the peer Credit more stream-data bytes on one
169// logical stream after the receiver consumed them. It travels in both
170// directions: each side grants credit for the data it receives.
171type StreamWindowUpdate struct {
172 Epoch ports.BrokerEpoch
173 Connection ports.BrokerConnectionID
174 Stream ports.BrokerStreamID
175 Credit uint64
176}
177
178func (StreamWindowUpdate) brokerClientMessage() {}
179func (StreamWindowUpdate) brokerServerMessage() {}
180
181// StartPreview starts or replaces one connection-scoped live preview.
182// Generation is the connection-scoped preview authority; Route names the
183// observed daemon (the broker allocates the stream) and Preview is the bounded
184// viewport request sent to it.
185type StartPreview struct {
186 Epoch ports.BrokerEpoch
187 Connection ports.BrokerConnectionID
188 Generation ports.BrokerPreviewGeneration
189 Route ports.BrokerPreviewRoute
190 Preview protocol.RemotePreviewRequest
191}
192
193func (StartPreview) brokerClientMessage() {}
194
195// CancelPreview cancels one connection-scoped live preview.
196type CancelPreview struct {
197 Epoch ports.BrokerEpoch
198 Connection ports.BrokerConnectionID
199 Generation ports.BrokerPreviewGeneration
200}
201
202func (CancelPreview) brokerClientMessage() {}
203
204// Registered confirms one accepted client connection and its assigned scope.
205type Registered struct {
206 Epoch ports.BrokerEpoch
207 Connection ports.BrokerConnectionID
208 Build string
209 Retiring bool
210}
211
212func (Registered) brokerServerMessage() {}
213
214// SnapshotPart is one fragment of a snapshot publication. Parts of one
215// epoch generation and broker revision arrive in index order and carry
216// exactly one nested payload.
217type SnapshotPart struct {
218 Epoch ports.BrokerEpoch
219 Connection ports.BrokerConnectionID
220 // Generation is the client's subscription series, echoed back so the
221 // client can drop parts bound to a superseded subscription.
222 Generation SubscriptionGeneration
223 // Revision is the broker publication revision within the epoch.
224 Revision ports.BrokerRevision
225 Index uint32
226 Part SnapshotPartPayload
227}
228
229func (SnapshotPart) brokerServerMessage() {}
230
231// SnapshotPartPayload is the closed set of snapshot fragment payloads.
232type SnapshotPartPayload interface{ snapshotPartPayload() }
233
234// SnapshotBegin opens a snapshot publication with its element counts.
235// HostCount counts every daemon part in the transfer (the local daemon plus
236// every remote host); LocalPresent reports whether daemon index 0 is local.
237type SnapshotBegin struct {
238 HostCount uint32
239 SessionCount uint32
240 TombstoneCount uint32
241 LocalPresent bool
242}
243
244func (SnapshotBegin) snapshotPartPayload() {}
245
246// SnapshotDaemonPart is one daemon projection. Sessions travel as separate
247// SnapshotSessionPart parts bound by daemon index; SessionCount advertises
248// how many follow. The embedded Daemon must carry no inline sessions.
249type SnapshotDaemonPart struct {
250 HostIndex uint32
251 Daemon ports.BrokerDaemonObservation
252 SessionCount uint32
253}
254
255func (SnapshotDaemonPart) snapshotPartPayload() {}
256
257// SnapshotSessionPart is one catalogue session bound to its daemon by index.
258// Local marks a session of the local daemon: HostIndex is then exactly 0.
259type SnapshotSessionPart struct {
260 HostIndex uint32
261 SessionIndex uint32
262 Local bool
263 Session catalogue.RemoteCatalogSession
264}
265
266func (SnapshotSessionPart) snapshotPartPayload() {}
267
268// SnapshotTombstonePart fences one retired host registration so a stale
269// probe cannot revive authority for that exact incarnation.
270type SnapshotTombstonePart struct {
271 TombstoneIndex uint32
272 Registration domain.RemoteRegistration
273 RetiredRevision ports.BrokerRevision
274}
275
276func (SnapshotTombstonePart) snapshotPartPayload() {}
277
278// SnapshotEnd closes a snapshot publication.
279type SnapshotEnd struct{}
280
281func (SnapshotEnd) snapshotPartPayload() {}
282
283// OperationResult is exactly one completion for one admitted mutating
284// operation. Removed and Registration are presence-carrying authority: a
285// successful addition or policy update carries the resulting registration,
286// a successful removal reports Removed and carries no registration, and a
287// failed or outcome-unknown result carries neither authority, only its
288// typed error. Error carries the typed failure when Outcome is failed or
289// outcome-unknown; HasError=false means no detail at all, and both a
290// nonzero Error under HasError=false and an error on a successful outcome
291// are refused rather than dropped.
292type OperationResult struct {
293 Epoch ports.BrokerEpoch
294 Connection ports.BrokerConnectionID
295 Operation ports.BrokerOperationID
296 Outcome ports.BrokerMutationOutcome
297 Removed bool
298 Registration domain.RemoteRegistration
299 Error ErrorDetail
300 HasError bool
301}
302
303func (OperationResult) brokerServerMessage() {}
304
305// RegisterMutationKind is the closed taxonomy of membership mutations. A
306// result's required authority depends on which request produced it, so the
307// taxonomy is explicit instead of being re-derived from a message type at
308// each result site.
309type RegisterMutationKind uint8
310
311const (
312 // MutationKindAddHost is an AddHost request: its success carries the
313 // authoritative registration.
314 MutationKindAddHost RegisterMutationKind = iota + 1
315 // MutationKindRemoveHost is a RemoveHost request: its success reports
316 // removal and carries no registration.
317 MutationKindRemoveHost
318 // MutationKindUpdateHostPolicy is an UpdateHostPolicy request: its
319 // success carries the advanced registration.
320 MutationKindUpdateHostPolicy
321)
322
323func (k RegisterMutationKind) String() string {
324 switch k {
325 case MutationKindAddHost:
326 return "add_host"
327 case MutationKindRemoveHost:
328 return "remove_host"
329 case MutationKindUpdateHostPolicy:
330 return "update_host_policy"
331 default:
332 return "unknown"
333 }
334}
335
336// Validate rejects a kind outside the closed membership-mutation taxonomy.
337func (k RegisterMutationKind) Validate() error {
338 switch k {
339 case MutationKindAddHost, MutationKindRemoveHost, MutationKindUpdateHostPolicy:
340 return nil
341 default:
342 return errConvertRange
343 }
344}
345
346// MutationKindOf classifies one client message that mutates host membership.
347// It is the request-side authority a per-kind result rule needs; pending
348// operation tracking is deliberately not wired here.
349func MutationKindOf(message ClientMessage) (RegisterMutationKind, bool) {
350 switch m := message.(type) {
351 case AddHost:
352 return MutationKindAddHost, true
353 case *AddHost:
354 if m == nil {
355 return 0, false
356 }
357 return MutationKindAddHost, true
358 case RemoveHost:
359 return MutationKindRemoveHost, true
360 case *RemoveHost:
361 if m == nil {
362 return 0, false
363 }
364 return MutationKindRemoveHost, true
365 case UpdateHostPolicy:
366 return MutationKindUpdateHostPolicy, true
367 case *UpdateHostPolicy:
368 if m == nil {
369 return 0, false
370 }
371 return MutationKindUpdateHostPolicy, true
372 default:
373 return 0, false
374 }
375}
376
377// validateResultAuthority enforces the presence rules of one mutating result
378// that hold without knowing which request produced it: removal and
379// registration never travel together, a successful result never carries an
380// error, and a failed or outcome-unknown result carries neither authority,
381// only its typed error. The per-kind rule (an AddHost or UpdateHostPolicy
382// success must carry a registration, a RemoveHost success must not) is
383// enforced by ValidateForMutation at the owning site, because the stateless
384// codec has no pending-operation state to consult. A successful removal with
385// no authority is legal: an absent host is an idempotent no-op, not an error.
386func validateResultAuthority(outcome ports.BrokerMutationOutcome, removed, hasRegistration, hasError bool) error {
387 if err := outcome.Validate(); err != nil {
388 return ErrInvalidMessage
389 }
390 if removed && hasRegistration {
391 return ErrInvalidMessage
392 }
393 if outcome == ports.BrokerOutcomeOK {
394 if hasError {
395 return ErrInvalidMessage
396 }
397 return nil
398 }
399 if removed || hasRegistration || !hasError {
400 return ErrInvalidMessage
401 }
402 return nil
403}
404
405// ValidateForMutation enforces the full authority rule of one mutating result
406// against the kind of request it completes. AddHost and UpdateHostPolicy
407// succeed with the authoritative registration, RemoveHost succeeds with
408// removal and no registration, and every failed or outcome-unknown result
409// carries neither authority, only its validated typed error.
410func (r OperationResult) ValidateForMutation(kind RegisterMutationKind) error {
411 if err := kind.Validate(); err != nil {
412 return err
413 }
414 hasRegistration := r.Registration != (domain.RemoteRegistration{})
415 if err := validateResultAuthority(r.Outcome, r.Removed, hasRegistration, r.HasError); err != nil {
416 return err
417 }
418 if r.HasError {
419 if err := r.Error.validate(); err != nil {
420 return err
421 }
422 }
423 switch r.Outcome {
424 case ports.BrokerOutcomeOK:
425 switch kind {
426 case MutationKindRemoveHost:
427 if hasRegistration {
428 return ErrInvalidMessage
429 }
430 default:
431 if !hasRegistration {
432 return ErrInvalidMessage
433 }
434 }
435 default:
436 if r.Removed || hasRegistration || !r.HasError {
437 return ErrInvalidMessage
438 }
439 }
440 return nil
441}
442
443// StreamOpened confirms one logical stream is established.
444type StreamOpened struct {
445 Epoch ports.BrokerEpoch
446 Connection ports.BrokerConnectionID
447 Stream ports.BrokerStreamID
448}
449
450func (StreamOpened) brokerServerMessage() {}
451
452// ServerStreamData carries one opaque daemon-to-client stream frame.
453type ServerStreamData struct {
454 Epoch ports.BrokerEpoch
455 Connection ports.BrokerConnectionID
456 Stream ports.BrokerStreamID
457 Data []byte
458}
459
460func (ServerStreamData) brokerServerMessage() {}
461
462// StreamClosed ends one logical stream. An absent error is an orderly
463// close; a present error reports the typed failure that ended it.
464// HasError=false means no detail at all: encode refuses a nonzero Error rather
465// than dropping it.
466type StreamClosed struct {
467 Epoch ports.BrokerEpoch
468 Connection ports.BrokerConnectionID
469 Stream ports.BrokerStreamID
470 Error ErrorDetail
471 HasError bool
472}
473
474func (StreamClosed) brokerServerMessage() {}
475
476// Progress reports one phase of a long-running broker operation.
477type Progress struct {
478 Epoch ports.BrokerEpoch
479 Connection ports.BrokerConnectionID
480 Stream ports.BrokerStreamID
481 Phase BrokerProgressPhase
482 Text string
483}
484
485func (Progress) brokerServerMessage() {}
486
487// BrokerErrorMessage reports a broker-wide typed failure for one scope.
488type BrokerErrorMessage struct {
489 Epoch ports.BrokerEpoch
490 Connection ports.BrokerConnectionID
491 Error ErrorDetail
492}
493
494func (BrokerErrorMessage) brokerServerMessage() {}
495
496// PreviewPublication is the newest result for one preview subscription.
497// Every authority field is repeated so a client can reject a late
498// completion. Exactly one of preview/error travels via the wire result
499// oneof: a failed publication carries HasError with a typed ErrorDetail
500// and no Preview; a successful one carries Preview and no error.
501type PreviewPublication struct {
502 Epoch ports.BrokerEpoch
503 Connection ports.BrokerConnectionID
504 Generation ports.BrokerPreviewGeneration
505 Preview protocol.RemotePreview
506 Error ErrorDetail
507 HasError bool
508}
509
510func (PreviewPublication) brokerServerMessage() {}
511
512// MaxBrokerAdmissionCode is the highest admission-refusal code carried on
513// ErrorDetail. The taxonomy is closed: 0 is none, 1 limit, 2 closed, 3 stale
514// connection or stream, 4 invalid request (mirroring
515// ports.BrokerAdmissionError).
516const MaxBrokerAdmissionCode uint32 = 4
517
518// MaxBrokerFailureKind bounds ErrorDetail.FailureKind to the domain
519// transport-failure taxonomy. The zero value (RemoteFailureNone) is valid.
520const MaxBrokerFailureKind = domain.RemoteFailureInvalidResponse
521
522// ErrorDetail is the typed broker failure carried to clients: a closed
523// error code plus bounded display text, the admission refusal (if any),
524// and the sanitized transport cause. The underlying diagnostic cause is
525// local-only and never travels on this wire.
526type ErrorDetail struct {
527 Code ports.BrokerErrorCode
528 Text string
529 // AdmissionCode is the closed admission-refusal taxonomy: 0 is none,
530 // 1 limit, 2 closed, 3 stale connection or stream, 4 invalid request.
531 AdmissionCode uint32
532 FailureKind domain.RemoteFailureKind
533}
534
535// validate enforces the transport contract of one ErrorDetail: a closed
536// error code, bounded presentation-safe text, an admission code within the
537// closed refusal taxonomy, and a failure kind within the domain taxonomy.
538func (d ErrorDetail) validate() error {
539 if err := d.Code.Validate(); err != nil {
540 return ErrInvalidMessage
541 }
542 if d.AdmissionCode > MaxBrokerAdmissionCode {
543 return ErrInvalidMessage
544 }
545 if d.FailureKind > MaxBrokerFailureKind {
546 return ErrInvalidMessage
547 }
548 return validateBrokerDisplayText(d.Text, ports.BrokerMaxErrorBytes)
549}
550
551type ShutdownReason uint8
552
553const (
554 // ShutdownUnspecified is the zero value: never sent on the wire.
555 ShutdownUnspecified ShutdownReason = iota
556 // ShutdownIdleExit reports the broker stopped for lack of work.
557 ShutdownIdleExit
558 // ShutdownTerminating reports the broker process is stopping.
559 ShutdownTerminating
560)
561
562func (r ShutdownReason) validate() error {
563 switch r {
564 case ShutdownIdleExit, ShutdownTerminating:
565 return nil
566 default:
567 return errConvertRange
568 }
569}
570
571// Shutdown reports the broker process is stopping.
572type Shutdown struct {
573 Epoch ports.BrokerEpoch
574 Connection ports.BrokerConnectionID
575 Reason ShutdownReason
576 Text string
577}
578
579func (Shutdown) brokerServerMessage() {}
580
581// BrokerProgressPhase is the closed taxonomy for long-running operation
582// progress reports carried on Progress.
583type BrokerProgressPhase uint32
584
585const (
586 // BrokerProgressStarted opens a long-running operation.
587 BrokerProgressStarted BrokerProgressPhase = 1
588 // BrokerProgressProbing reports host observation is in flight.
589 BrokerProgressProbing BrokerProgressPhase = 2
590 // BrokerProgressConnecting reports physical transport dialing.
591 BrokerProgressConnecting BrokerProgressPhase = 3
592 // BrokerProgressDone closes a long-running operation.
593 BrokerProgressDone BrokerProgressPhase = 4
594)
595
596func (p BrokerProgressPhase) validate() error {
597 switch p {
598 case BrokerProgressStarted, BrokerProgressProbing, BrokerProgressConnecting, BrokerProgressDone:
599 return nil
600 default:
601 return errConvertRange
602 }
603}
604
605// SubscriptionGeneration identifies one client subscription series within a
606// scope. The wire carries it as a bare uint64 on Subscribe/Resync/
607// Unsubscribe and echoes it on every SnapshotPart; zero is valid.
608type SubscriptionGeneration = uint64
609
610// brokerTimeToWire splits one instant into seconds and nanoseconds so
611// time.Time survives the wire exactly. The zero time maps to the zero
612// timestamp and back.
613func brokerTimeToWire(t time.Time) (int64, int32) {
614 if t.IsZero() {
615 return 0, 0
616 }
617 return t.Unix(), int32(t.Nanosecond())
618}
619
620// brokerTimeFromWire reassembles one timestamp. Out-of-range values fail:
621// times must round-trip through time.Unix without overflow.
622func brokerTimeFromWire(seconds int64, nanos int32) (time.Time, error) {
623 if seconds == 0 && nanos == 0 {
624 return time.Time{}, nil
625 }
626 if nanos < 0 || nanos > 999999999 {
627 return time.Time{}, errConvertRange
628 }
629 // time.Unix panics only outside int64-derived ranges; seconds is
630 // already int64, and the nanosecond field is bounded above, so this
631 // cannot overflow on supported platforms beyond what time.Unix
632 // itself rejects. Guard the documented extremes anyway.
633 t := time.Unix(seconds, int64(nanos)).UTC()
634 if t.Unix() != seconds {
635 return time.Time{}, errConvertRange
636 }
637 return t, nil
638}