1package brokeripc
2
3import (
4 "context"
5 "errors"
6 "fmt"
7 "strings"
8
9 "github.com/bnema/vev/internal/adapters/brokerwire"
10 "github.com/bnema/vev/internal/ports"
11)
12
13// Adapter sentinels. They are typed so a caller can classify a listener,
14// session, framing, or backpressure outcome without matching on message text.
15// Semantic broker refusals stay ports.BrokerError, and admission refusals stay
16// ports.BrokerAdmissionError; this adapter wraps them rather than redefining
17// them.
18var (
19 // ErrConfig reports an invalid listener or dialer configuration.
20 ErrConfig = errors.New("brokeripc: invalid configuration")
21 ErrBrokerRetired = errors.New("brokeripc: stale broker is retiring for a newer build")
22 // ErrListenerClosed reports Accept on a closed listener.
23 ErrListenerClosed = errors.New("brokeripc: listener is closed")
24 // ErrSessionClosed reports session work presented after the session reached
25 // its terminal outcome.
26 ErrSessionClosed = errors.New("brokeripc: session is closed")
27 // ErrMalformedFrame reports an outer envelope this side refuses to decode:
28 // an oversize length prefix, a truncated or trailing payload, an unknown or
29 // repeated field, or a wrong-direction payload. A malformed frame is a
30 // protocol violation and settles exactly the connection that sent it.
31 ErrMalformedFrame = errors.New("brokeripc: malformed broker frame")
32 // ErrProtocol reports a well-formed frame that violates connection
33 // ordering: a duplicate Register, a frame for an unallocated stream, or
34 // stream data before the stream was established.
35 ErrProtocol = errors.New("brokeripc: broker protocol violation")
36 // ErrStreamBackpressure reports a logical stream whose local consumer is
37 // not draining its bounded inbound queue. The affected stream is settled
38 // alone; the connection and every sibling stream stay up.
39 ErrStreamBackpressure = errors.New("brokeripc: stream consumer is not keeping up")
40 // ErrStreamGone reports a stream the peer retired while this side was still
41 // starting or using it.
42 ErrStreamGone = errors.New("brokeripc: logical stream is gone")
43 // ErrConnectionClosed reports a client adapter whose connection reached its
44 // terminal outcome.
45 ErrConnectionClosed = errors.New("brokeripc: broker connection is closed")
46 // ErrScopeMismatch reports a frame carrying an epoch or connection identity
47 // that is not the one assigned at accept. It is refused as stale and never
48 // applied.
49 ErrScopeMismatch = errors.New("brokeripc: frame scope does not match the accepted connection")
50 // ErrRegistrationTimeout reports a client that completed the broker preamble
51 // and admission but did not send its Register within the accept-time
52 // handshake budget. The session is settled, its admitted core service is
53 // closed, and its listener slot is released; context.DeadlineExceeded also
54 // matches it. It is an orderly peer disconnect, so a listener draining its
55 // sessions does not report a clean shutdown as this timeout.
56 ErrRegistrationTimeout = errors.New("brokeripc: broker registration timed out")
57)
58
59// transportFailure classifies one carriage error that ended a connection. A
60// closed carriage is an orderly disconnect; anything else is a transport
61// failure.
62func transportFailure(err error) error {
63 if err == nil {
64 return ErrConnectionClosed
65 }
66 return err
67}
68
69// cancelOrCause prefers the context outcome when a wait was cancelled, so a
70// caller sees cancellation rather than a synthetic transport error.
71func cancelOrCause(ctx context.Context, cause error) error {
72 if ctxErr := ctx.Err(); ctxErr != nil {
73 return ctxErr
74 }
75 return cause
76}
77
78// admissionError maps one brokerwire tracker refusal onto the closed ports
79// admission taxonomy. An unrecognized refusal is invalid-request, never a
80// retryable limit.
81func admissionError(err error) error {
82 switch {
83 case err == nil:
84 return nil
85 case errors.Is(err, brokerwire.ErrTooManyStreams), errors.Is(err, brokerwire.ErrTooManyPendingOperations):
86 return ports.BrokerAdmissionLimit
87 case errors.Is(err, brokerwire.ErrConnectionClosed):
88 return ports.BrokerAdmissionClosed
89 case errors.Is(err, brokerwire.ErrStreamIDReused), errors.Is(err, brokerwire.ErrConnectionState),
90 errors.Is(err, brokerwire.ErrNotSubscribed), errors.Is(err, brokerwire.ErrDuplicateRegister),
91 errors.Is(err, brokerwire.ErrOperationCompleted), errors.Is(err, brokerwire.ErrOperationPending):
92 return ports.BrokerAdmissionStale
93 default:
94 return ports.BrokerAdmissionInvalid
95 }
96}
97
98// errorDetail converts one broker failure into the bounded, presentation-safe
99// wire detail. Causes stay local, except for a physical stream loss: its
100// failure kind and a sanitized, bounded description of the transport cause
101// travel as display text so the client can report why the attachment ended.
102// Otherwise only the typed code, bounded display text, and closed admission
103// code travel.
104func errorDetail(err error) brokerwire.ErrorDetail {
105 if err == nil {
106 return brokerwire.ErrorDetail{}
107 }
108 // A physical loss keeps its failure kind and a bounded description of the
109 // transport cause, so the client can report why the attachment ended.
110 var lost ports.BrokerStreamLost
111 if errors.As(err, &lost) && lost.Validate() == nil {
112 detail := brokerwire.ErrorDetail{Code: ports.BrokerErrorAttachmentLost, Text: lost.Cause.String()}
113 if lost.Err != nil {
114 detail.Text = boundedBrokerErrorText(fmt.Errorf("%s: %w", lost.Cause, lost.Err))
115 }
116 return detail
117 }
118 var failure ports.BrokerError
119 if errors.As(err, &failure) {
120 return brokerwire.ErrorDetail{Code: failure.Code, Text: failure.Text}
121 }
122 var admission ports.BrokerAdmissionError
123 if errors.As(err, &admission) {
124 detail := brokerwire.ErrorDetail{Code: ports.BrokerErrorUnavailable}
125 switch {
126 case errors.Is(err, ports.BrokerAdmissionLimit):
127 detail.AdmissionCode = 1
128 case errors.Is(err, ports.BrokerAdmissionClosed):
129 detail.AdmissionCode = 2
130 case errors.Is(err, ports.BrokerAdmissionStale):
131 detail.AdmissionCode = 3
132 default:
133 detail.AdmissionCode = 4
134 }
135 return detail
136 }
137 // The ports type carries value-receiver methods, so both the value and the
138 // pointer form are legal errors; classify either rather than letting the
139 // pointer form fall through to a retryable-looking code.
140 var outcomeUnknown ports.BrokerStoreOutcomeUnknownError
141 var outcomeUnknownPtr *ports.BrokerStoreOutcomeUnknownError
142 switch {
143 case errors.As(err, &outcomeUnknown), errors.As(err, &outcomeUnknownPtr):
144 return brokerwire.ErrorDetail{Code: ports.BrokerErrorOutcomeUnknown}
145 case errors.Is(err, context.DeadlineExceeded):
146 return brokerwire.ErrorDetail{Code: ports.BrokerErrorTimeout}
147 case errors.Is(err, context.Canceled):
148 return brokerwire.ErrorDetail{Code: ports.BrokerErrorCancelled}
149 default:
150 return brokerwire.ErrorDetail{Code: ports.BrokerErrorUnavailable, Text: boundedBrokerErrorText(err)}
151 }
152}
153
154func boundedBrokerErrorText(err error) string {
155 const max = 256
156 text := ports.SanitizeBrokerDisplayText(err.Error())
157 if len(text) > max {
158 text = strings.ToValidUTF8(text[:max], "")
159 }
160 return text
161}
162
163// failureFromDetail converts one received detail back into the typed local
164// failure. A detail with no code is the absent-detail case and yields nil.
165func failureFromDetail(detail brokerwire.ErrorDetail) error {
166 switch detail.AdmissionCode {
167 case 1:
168 return ports.BrokerAdmissionLimit
169 case 2:
170 return ports.BrokerAdmissionClosed
171 case 3:
172 return ports.BrokerAdmissionStale
173 case 4:
174 return ports.BrokerAdmissionInvalid
175 }
176 if detail.Code == 0 {
177 return nil
178 }
179 return ports.BrokerError{Code: detail.Code, Text: detail.Text}
180}