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// StartPreview starts or replaces one connection-scoped live preview.
169// Generation is the connection-scoped preview authority; Route names the
170// observed daemon (the broker allocates the stream) and Preview is the bounded
171// viewport request sent to it.
172type StartPreview struct {
173 Epoch ports.BrokerEpoch
174 Connection ports.BrokerConnectionID
175 Generation ports.BrokerPreviewGeneration
176 Route ports.BrokerPreviewRoute
177 Preview protocol.RemotePreviewRequest
178}
179
180func (StartPreview) brokerClientMessage() {}
181
182// CancelPreview cancels one connection-scoped live preview.
183type CancelPreview struct {
184 Epoch ports.BrokerEpoch
185 Connection ports.BrokerConnectionID
186 Generation ports.BrokerPreviewGeneration
187}
188
189func (CancelPreview) brokerClientMessage() {}
190
191// Registered confirms one accepted client connection and its assigned scope.
192type Registered struct {
193 Epoch ports.BrokerEpoch
194 Connection ports.BrokerConnectionID
195 Build string
196 Retiring bool
197}
198
199func (Registered) brokerServerMessage() {}
200
201// SnapshotPart is one fragment of a snapshot publication. Parts of one
202// epoch generation and broker revision arrive in index order and carry
203// exactly one nested payload.
204type SnapshotPart struct {
205 Epoch ports.BrokerEpoch
206 Connection ports.BrokerConnectionID
207 // Generation is the client's subscription series, echoed back so the
208 // client can drop parts bound to a superseded subscription.
209 Generation SubscriptionGeneration
210 // Revision is the broker publication revision within the epoch.
211 Revision ports.BrokerRevision
212 Index uint32
213 Part SnapshotPartPayload
214}
215
216func (SnapshotPart) brokerServerMessage() {}
217
218// SnapshotPartPayload is the closed set of snapshot fragment payloads.
219type SnapshotPartPayload interface{ snapshotPartPayload() }
220
221// SnapshotBegin opens a snapshot publication with its element counts.
222// HostCount counts every daemon part in the transfer (the local daemon plus
223// every remote host); LocalPresent reports whether daemon index 0 is local.
224type SnapshotBegin struct {
225 HostCount uint32
226 SessionCount uint32
227 TombstoneCount uint32
228 LocalPresent bool
229}
230
231func (SnapshotBegin) snapshotPartPayload() {}
232
233// SnapshotDaemonPart is one daemon projection. Sessions travel as separate
234// SnapshotSessionPart parts bound by daemon index; SessionCount advertises
235// how many follow. The embedded Daemon must carry no inline sessions.
236type SnapshotDaemonPart struct {
237 HostIndex uint32
238 Daemon ports.BrokerDaemonObservation
239 SessionCount uint32
240}
241
242func (SnapshotDaemonPart) snapshotPartPayload() {}
243
244// SnapshotSessionPart is one catalogue session bound to its daemon by index.
245// Local marks a session of the local daemon: HostIndex is then exactly 0.
246type SnapshotSessionPart struct {
247 HostIndex uint32
248 SessionIndex uint32
249 Local bool
250 Session catalogue.RemoteCatalogSession
251}
252
253func (SnapshotSessionPart) snapshotPartPayload() {}
254
255// SnapshotTombstonePart fences one retired host registration so a stale
256// probe cannot revive authority for that exact incarnation.
257type SnapshotTombstonePart struct {
258 TombstoneIndex uint32
259 Registration domain.RemoteRegistration
260 RetiredRevision ports.BrokerRevision
261}
262
263func (SnapshotTombstonePart) snapshotPartPayload() {}
264
265// SnapshotEnd closes a snapshot publication.
266type SnapshotEnd struct{}
267
268func (SnapshotEnd) snapshotPartPayload() {}
269
270// OperationResult is exactly one completion for one admitted mutating
271// operation. Removed and Registration are presence-carrying authority: a
272// successful addition or policy update carries the resulting registration,
273// a successful removal reports Removed and carries no registration, and a
274// failed or outcome-unknown result carries neither authority, only its
275// typed error. Error carries the typed failure when Outcome is failed or
276// outcome-unknown; HasError=false means no detail at all, and both a
277// nonzero Error under HasError=false and an error on a successful outcome
278// are refused rather than dropped.
279type OperationResult struct {
280 Epoch ports.BrokerEpoch
281 Connection ports.BrokerConnectionID
282 Operation ports.BrokerOperationID
283 Outcome ports.BrokerMutationOutcome
284 Removed bool
285 Registration domain.RemoteRegistration
286 Error ErrorDetail
287 HasError bool
288}
289
290func (OperationResult) brokerServerMessage() {}
291
292// RegisterMutationKind is the closed taxonomy of membership mutations. A
293// result's required authority depends on which request produced it, so the
294// taxonomy is explicit instead of being re-derived from a message type at
295// each result site.
296type RegisterMutationKind uint8
297
298const (
299 // MutationKindAddHost is an AddHost request: its success carries the
300 // authoritative registration.
301 MutationKindAddHost RegisterMutationKind = iota + 1
302 // MutationKindRemoveHost is a RemoveHost request: its success reports
303 // removal and carries no registration.
304 MutationKindRemoveHost
305 // MutationKindUpdateHostPolicy is an UpdateHostPolicy request: its
306 // success carries the advanced registration.
307 MutationKindUpdateHostPolicy
308)
309
310func (k RegisterMutationKind) String() string {
311 switch k {
312 case MutationKindAddHost:
313 return "add_host"
314 case MutationKindRemoveHost:
315 return "remove_host"
316 case MutationKindUpdateHostPolicy:
317 return "update_host_policy"
318 default:
319 return "unknown"
320 }
321}
322
323// Validate rejects a kind outside the closed membership-mutation taxonomy.
324func (k RegisterMutationKind) Validate() error {
325 switch k {
326 case MutationKindAddHost, MutationKindRemoveHost, MutationKindUpdateHostPolicy:
327 return nil
328 default:
329 return errConvertRange
330 }
331}
332
333// MutationKindOf classifies one client message that mutates host membership.
334// It is the request-side authority a per-kind result rule needs; pending
335// operation tracking is deliberately not wired here.
336func MutationKindOf(message ClientMessage) (RegisterMutationKind, bool) {
337 switch m := message.(type) {
338 case AddHost:
339 return MutationKindAddHost, true
340 case *AddHost:
341 if m == nil {
342 return 0, false
343 }
344 return MutationKindAddHost, true
345 case RemoveHost:
346 return MutationKindRemoveHost, true
347 case *RemoveHost:
348 if m == nil {
349 return 0, false
350 }
351 return MutationKindRemoveHost, true
352 case UpdateHostPolicy:
353 return MutationKindUpdateHostPolicy, true
354 case *UpdateHostPolicy:
355 if m == nil {
356 return 0, false
357 }
358 return MutationKindUpdateHostPolicy, true
359 default:
360 return 0, false
361 }
362}
363
364// validateResultAuthority enforces the presence rules of one mutating result
365// that hold without knowing which request produced it: removal and
366// registration never travel together, a successful result never carries an
367// error, and a failed or outcome-unknown result carries neither authority,
368// only its typed error. The per-kind rule (an AddHost or UpdateHostPolicy
369// success must carry a registration, a RemoveHost success must not) is
370// enforced by ValidateForMutation at the owning site, because the stateless
371// codec has no pending-operation state to consult. A successful removal with
372// no authority is legal: an absent host is an idempotent no-op, not an error.
373func validateResultAuthority(outcome ports.BrokerMutationOutcome, removed, hasRegistration, hasError bool) error {
374 if err := outcome.Validate(); err != nil {
375 return ErrInvalidMessage
376 }
377 if removed && hasRegistration {
378 return ErrInvalidMessage
379 }
380 if outcome == ports.BrokerOutcomeOK {
381 if hasError {
382 return ErrInvalidMessage
383 }
384 return nil
385 }
386 if removed || hasRegistration || !hasError {
387 return ErrInvalidMessage
388 }
389 return nil
390}
391
392// ValidateForMutation enforces the full authority rule of one mutating result
393// against the kind of request it completes. AddHost and UpdateHostPolicy
394// succeed with the authoritative registration, RemoveHost succeeds with
395// removal and no registration, and every failed or outcome-unknown result
396// carries neither authority, only its validated typed error.
397func (r OperationResult) ValidateForMutation(kind RegisterMutationKind) error {
398 if err := kind.Validate(); err != nil {
399 return err
400 }
401 hasRegistration := r.Registration != (domain.RemoteRegistration{})
402 if err := validateResultAuthority(r.Outcome, r.Removed, hasRegistration, r.HasError); err != nil {
403 return err
404 }
405 if r.HasError {
406 if err := r.Error.validate(); err != nil {
407 return err
408 }
409 }
410 switch r.Outcome {
411 case ports.BrokerOutcomeOK:
412 switch kind {
413 case MutationKindRemoveHost:
414 if hasRegistration {
415 return ErrInvalidMessage
416 }
417 default:
418 if !hasRegistration {
419 return ErrInvalidMessage
420 }
421 }
422 default:
423 if r.Removed || hasRegistration || !r.HasError {
424 return ErrInvalidMessage
425 }
426 }
427 return nil
428}
429
430// StreamOpened confirms one logical stream is established.
431type StreamOpened struct {
432 Epoch ports.BrokerEpoch
433 Connection ports.BrokerConnectionID
434 Stream ports.BrokerStreamID
435}
436
437func (StreamOpened) brokerServerMessage() {}
438
439// ServerStreamData carries one opaque daemon-to-client stream frame.
440type ServerStreamData struct {
441 Epoch ports.BrokerEpoch
442 Connection ports.BrokerConnectionID
443 Stream ports.BrokerStreamID
444 Data []byte
445}
446
447func (ServerStreamData) brokerServerMessage() {}
448
449// StreamClosed ends one logical stream. An absent error is an orderly
450// close; a present error reports the typed failure that ended it.
451// HasError=false means no detail at all: encode refuses a nonzero Error rather
452// than dropping it.
453type StreamClosed struct {
454 Epoch ports.BrokerEpoch
455 Connection ports.BrokerConnectionID
456 Stream ports.BrokerStreamID
457 Error ErrorDetail
458 HasError bool
459}
460
461func (StreamClosed) brokerServerMessage() {}
462
463// Progress reports one phase of a long-running broker operation.
464type Progress struct {
465 Epoch ports.BrokerEpoch
466 Connection ports.BrokerConnectionID
467 Stream ports.BrokerStreamID
468 Phase BrokerProgressPhase
469 Text string
470}
471
472func (Progress) brokerServerMessage() {}
473
474// BrokerErrorMessage reports a broker-wide typed failure for one scope.
475type BrokerErrorMessage struct {
476 Epoch ports.BrokerEpoch
477 Connection ports.BrokerConnectionID
478 Error ErrorDetail
479}
480
481func (BrokerErrorMessage) brokerServerMessage() {}
482
483// PreviewPublication is the newest result for one preview subscription.
484// Every authority field is repeated so a client can reject a late
485// completion. Exactly one of preview/error travels via the wire result
486// oneof: a failed publication carries HasError with a typed ErrorDetail
487// and no Preview; a successful one carries Preview and no error.
488type PreviewPublication struct {
489 Epoch ports.BrokerEpoch
490 Connection ports.BrokerConnectionID
491 Generation ports.BrokerPreviewGeneration
492 Preview protocol.RemotePreview
493 Error ErrorDetail
494 HasError bool
495}
496
497func (PreviewPublication) brokerServerMessage() {}
498
499// MaxBrokerAdmissionCode is the highest admission-refusal code carried on
500// ErrorDetail. The taxonomy is closed: 0 is none, 1 limit, 2 closed, 3 stale
501// connection or stream, 4 invalid request (mirroring
502// ports.BrokerAdmissionError).
503const MaxBrokerAdmissionCode uint32 = 4
504
505// MaxBrokerFailureKind bounds ErrorDetail.FailureKind to the domain
506// transport-failure taxonomy. The zero value (RemoteFailureNone) is valid.
507const MaxBrokerFailureKind = domain.RemoteFailureInvalidResponse
508
509// ErrorDetail is the typed broker failure carried to clients: a closed
510// error code plus bounded display text, the admission refusal (if any),
511// and the sanitized transport cause. The underlying diagnostic cause is
512// local-only and never travels on this wire.
513type ErrorDetail struct {
514 Code ports.BrokerErrorCode
515 Text string
516 // AdmissionCode is the closed admission-refusal taxonomy: 0 is none,
517 // 1 limit, 2 closed, 3 stale connection or stream, 4 invalid request.
518 AdmissionCode uint32
519 FailureKind domain.RemoteFailureKind
520}
521
522// validate enforces the transport contract of one ErrorDetail: a closed
523// error code, bounded presentation-safe text, an admission code within the
524// closed refusal taxonomy, and a failure kind within the domain taxonomy.
525func (d ErrorDetail) validate() error {
526 if err := d.Code.Validate(); err != nil {
527 return ErrInvalidMessage
528 }
529 if d.AdmissionCode > MaxBrokerAdmissionCode {
530 return ErrInvalidMessage
531 }
532 if d.FailureKind > MaxBrokerFailureKind {
533 return ErrInvalidMessage
534 }
535 return validateBrokerDisplayText(d.Text, ports.BrokerMaxErrorBytes)
536}
537
538type ShutdownReason uint8
539
540const (
541 // ShutdownUnspecified is the zero value: never sent on the wire.
542 ShutdownUnspecified ShutdownReason = iota
543 // ShutdownIdleExit reports the broker stopped for lack of work.
544 ShutdownIdleExit
545 // ShutdownTerminating reports the broker process is stopping.
546 ShutdownTerminating
547)
548
549func (r ShutdownReason) validate() error {
550 switch r {
551 case ShutdownIdleExit, ShutdownTerminating:
552 return nil
553 default:
554 return errConvertRange
555 }
556}
557
558// Shutdown reports the broker process is stopping.
559type Shutdown struct {
560 Epoch ports.BrokerEpoch
561 Connection ports.BrokerConnectionID
562 Reason ShutdownReason
563 Text string
564}
565
566func (Shutdown) brokerServerMessage() {}
567
568// BrokerProgressPhase is the closed taxonomy for long-running operation
569// progress reports carried on Progress.
570type BrokerProgressPhase uint32
571
572const (
573 // BrokerProgressStarted opens a long-running operation.
574 BrokerProgressStarted BrokerProgressPhase = 1
575 // BrokerProgressProbing reports host observation is in flight.
576 BrokerProgressProbing BrokerProgressPhase = 2
577 // BrokerProgressConnecting reports physical transport dialing.
578 BrokerProgressConnecting BrokerProgressPhase = 3
579 // BrokerProgressDone closes a long-running operation.
580 BrokerProgressDone BrokerProgressPhase = 4
581)
582
583func (p BrokerProgressPhase) validate() error {
584 switch p {
585 case BrokerProgressStarted, BrokerProgressProbing, BrokerProgressConnecting, BrokerProgressDone:
586 return nil
587 default:
588 return errConvertRange
589 }
590}
591
592// SubscriptionGeneration identifies one client subscription series within a
593// scope. The wire carries it as a bare uint64 on Subscribe/Resync/
594// Unsubscribe and echoes it on every SnapshotPart; zero is valid.
595type SubscriptionGeneration = uint64
596
597// brokerTimeToWire splits one instant into seconds and nanoseconds so
598// time.Time survives the wire exactly. The zero time maps to the zero
599// timestamp and back.
600func brokerTimeToWire(t time.Time) (int64, int32) {
601 if t.IsZero() {
602 return 0, 0
603 }
604 return t.Unix(), int32(t.Nanosecond())
605}
606
607// brokerTimeFromWire reassembles one timestamp. Out-of-range values fail:
608// times must round-trip through time.Unix without overflow.
609func brokerTimeFromWire(seconds int64, nanos int32) (time.Time, error) {
610 if seconds == 0 && nanos == 0 {
611 return time.Time{}, nil
612 }
613 if nanos < 0 || nanos > 999999999 {
614 return time.Time{}, errConvertRange
615 }
616 // time.Unix panics only outside int64-derived ranges; seconds is
617 // already int64, and the nanosecond field is bounded above, so this
618 // cannot overflow on supported platforms beyond what time.Unix
619 // itself rejects. Guard the documented extremes anyway.
620 t := time.Unix(seconds, int64(nanos)).UTC()
621 if t.Unix() != seconds {
622 return time.Time{}, errConvertRange
623 }
624 return t, nil
625}