1package app
2
3import (
4 "context"
5 "encoding/hex"
6 "encoding/json"
7 "errors"
8 "fmt"
9 "io"
10 "io/fs"
11 "os"
12 "strconv"
13 "time"
14
15 "github.com/bnema/vev/internal/adapters/brokeripc"
16 "github.com/bnema/vev/internal/adapters/ipc"
17 "github.com/bnema/vev/internal/domain"
18 "github.com/bnema/vev/internal/ports"
19 "github.com/bnema/vev/internal/protocol"
20)
21
22const brokerReadyCommand = "_broker-ready"
23
24type brokerReadyOptions struct {
25 require string
26 timeout time.Duration
27 maxAge time.Duration
28}
29
30type brokerReadyResult struct {
31 Schema string `json:"schema"`
32 Status string `json:"status"`
33 Require string `json:"require"`
34 Reason string `json:"reason"`
35 Pending []string `json:"pending"`
36 Epoch string `json:"epoch"`
37 Revision string `json:"revision"`
38 Local *brokerReadyLocal `json:"local"`
39}
40type brokerReadyLocal struct {
41 Identity string `json:"identity"`
42 Incarnation string `json:"incarnation"`
43 Version string `json:"version"`
44 Availability string `json:"availability"`
45 InventoryKnown bool `json:"inventory_known"`
46 LastSuccess string `json:"last_success"`
47}
48
49func parseBrokerReadyArgs(args []string) (brokerReadyOptions, error) {
50 o := brokerReadyOptions{timeout: 10 * time.Second, maxAge: 5 * time.Second}
51 seen := map[string]bool{}
52 for i := 0; i < len(args); i += 2 {
53 if i+1 >= len(args) {
54 return o, usagef("`%s` requires a value", args[i])
55 }
56 name, value := args[i], args[i+1]
57 if seen[name] {
58 return o, usagef("duplicate `%s`", name)
59 }
60 seen[name] = true
61 switch name {
62 case "--require":
63 o.require = value
64 case "--timeout":
65 d, err := time.ParseDuration(value)
66 if err != nil || d <= 0 {
67 return o, usagef("`--timeout` must be positive")
68 }
69 o.timeout = d
70 case "--max-observation-age":
71 d, err := time.ParseDuration(value)
72 if err != nil || d <= 0 {
73 return o, usagef("`--max-observation-age` must be positive")
74 }
75 o.maxAge = d
76 default:
77 return o, usagef("unknown `%s` option", name)
78 }
79 }
80 if o.require != "local-authority" && o.require != "local-catalogue" {
81 return o, usagef("`--require` must be local-authority or local-catalogue")
82 }
83 if o.require != "local-catalogue" && seen["--max-observation-age"] {
84 return o, usagef("`--max-observation-age` is only valid for local-catalogue")
85 }
86 return o, nil
87}
88
89var brokerReadyDial = func(ctx context.Context, path string) (ports.BrokerService, error) {
90 return brokeripc.Dial(ctx, path, brokeripc.Config{})
91}
92
93func runBrokerReady(ctx context.Context, o brokerReadyOptions, out io.Writer) error {
94 deadline, cancel := context.WithTimeout(ctx, o.timeout)
95 defer cancel()
96 path := brokeripc.SocketPath(productionBrokerLayout().Runtime)
97 backoff := 5 * time.Millisecond
98 for {
99 if err := deadline.Err(); err != nil {
100 if errors.Is(ctx.Err(), context.Canceled) {
101 return writeReady(out, o, "canceled", "canceled", nil, 5)
102 }
103 return writeReady(out, o, "timeout", "deadline", nil, 3)
104 }
105 // A path that exists but is not a Unix socket, and a path this process
106 // may not inspect, are terminal security failures: retrying them into
107 // the deadline would report a foreign path as a slow broker. Only
108 // proven absence is retried, which is what lets an on-demand broker
109 // still come up under this probe.
110 if reason, terminal := brokerReadyEndpointSecurity(path); terminal {
111 return writeReady(out, o, "terminal", reason, nil, 4)
112 }
113 service, err := brokerReadyDial(deadline, path)
114 if err != nil {
115 if reason, terminal := terminalReadyReason(err); terminal {
116 return writeReady(out, o, "terminal", reason, nil, 4)
117 }
118 select {
119 case <-deadline.Done():
120 continue
121 case <-time.After(backoff):
122 }
123 if backoff < 50*time.Millisecond {
124 backoff *= 2
125 if backoff > 50*time.Millisecond {
126 backoff = 50 * time.Millisecond
127 }
128 }
129 continue
130 }
131 result := observeBrokerReady(deadline, o, service)
132 _ = service.Close()
133 if result != nil {
134 return emitReady(out, *result)
135 }
136 }
137}
138
139func observeBrokerReady(ctx context.Context, o brokerReadyOptions, service ports.BrokerService) *brokerReadyResult {
140 sub, err := service.Subscribe()
141 if err != nil {
142 r := readyResult(o, "terminal", "protocol", nil)
143 return &r
144 }
145 defer sub.Close()
146 for {
147 snapshot := service.Snapshot()
148 if snapshot.Epoch != 0 {
149 if snapshot.Validate() != nil {
150 r := readyResult(o, "terminal", "protocol", nil)
151 return &r
152 }
153 locals := make([]ports.BrokerDaemonObservation, 0, 1)
154 for _, d := range snapshot.Daemons {
155 if d.Local {
156 locals = append(locals, d)
157 }
158 }
159 if len(locals) > 1 {
160 r := readyResult(o, "terminal", "protocol", nil)
161 return &r
162 }
163 if len(locals) == 1 {
164 r := evaluateReady(o, snapshot, locals[0])
165 if r.Status == "ready" || r.Status == "terminal" {
166 return &r
167 }
168 }
169 }
170 select {
171 case <-ctx.Done():
172 return nil
173 case <-service.Done():
174 return nil
175 case <-sub.Changed():
176 }
177 }
178}
179
180func evaluateReady(o brokerReadyOptions, s ports.BrokerSnapshot, local ports.BrokerDaemonObservation) brokerReadyResult {
181 r := readyResult(o, "pending", "local-not-ready", &local)
182 r.Epoch = strconv.FormatUint(uint64(s.Epoch), 10)
183 r.Revision = strconv.FormatUint(uint64(s.Revision), 10)
184 if o.require == "local-authority" {
185 r.Status = "ready"
186 r.Reason = "authority"
187 r.Pending = []string{}
188 return r
189 }
190 valid := local.Availability == domain.RemoteAvailabilityReachable && local.InventoryKnown && local.Identity.Validate() == nil && local.Incarnation.Validate() == nil && local.ProtocolVersion == protocol.Version && !local.LastSuccess.IsZero()
191 age := time.Since(local.LastSuccess)
192 valid = valid && age >= 0 && age <= o.maxAge
193 if valid {
194 r.Status = "ready"
195 r.Reason = "catalogue"
196 r.Pending = []string{}
197 } else {
198 r.Pending = []string{"local-catalogue"}
199 }
200 return r
201}
202
203func readyResult(o brokerReadyOptions, status, reason string, local *ports.BrokerDaemonObservation) brokerReadyResult {
204 r := brokerReadyResult{Schema: "vev.broker-ready/v1", Status: status, Require: o.require, Reason: reason, Pending: []string{o.require}, Epoch: "0", Revision: "0"}
205 if local != nil {
206 r.Local = &brokerReadyLocal{Identity: string(local.Identity), Incarnation: hex.EncodeToString(local.Incarnation[:]), Version: strconv.FormatUint(uint64(local.ProtocolVersion), 10), Availability: local.Availability.String(), InventoryKnown: local.InventoryKnown, LastSuccess: local.LastSuccess.UTC().Format(time.RFC3339Nano)}
207 }
208 return r
209}
210
211// brokerReadyEndpointSecurity classifies the selected endpoint path before any
212// dial. Absence stays retryable; a path that exists and is not a Unix socket,
213// and a path this process may not inspect, are terminal security failures. Only
214// the specific not-exist outcome is absence, so an unreadable parent, a foreign
215// path, or any other inspection failure fails closed.
216func brokerReadyEndpointSecurity(path string) (string, bool) {
217 info, err := os.Lstat(path)
218 switch {
219 case errors.Is(err, fs.ErrNotExist):
220 return "", false
221 case err != nil:
222 return "security", true
223 case info.Mode()&os.ModeSocket == 0:
224 return "security", true
225 default:
226 return "", false
227 }
228}
229
230// terminalReadyReason classifies one failed dial attempt that must never be
231// retried into the readiness deadline. A refusal that proves the path, the peer,
232// the credentials, the configuration, or the broker conversation itself is not
233// the expected broker is terminal; every other outcome, including an absent or
234// refused socket, stays retryable so a broker can still come up.
235func terminalReadyReason(err error) (string, bool) {
236 switch {
237 case err == nil:
238 return "", false
239 case errors.Is(err, context.Canceled), errors.Is(err, context.DeadlineExceeded):
240 return "", false
241 case errors.Is(err, ipc.ErrMuxPath), errors.Is(err, ipc.ErrMuxForeignPath),
242 errors.Is(err, ipc.ErrMuxPeerRejected), errors.Is(err, ipc.ErrMuxUnsupported),
243 errors.Is(err, os.ErrPermission):
244 return "security", true
245 case errors.Is(err, brokeripc.ErrConfig):
246 return "config", true
247 case errors.Is(err, brokeripc.ErrProtocol), errors.Is(err, brokeripc.ErrMalformedFrame),
248 errors.Is(err, ipc.ErrFrameTooLarge), errors.Is(err, ipc.ErrZeroLengthFrame):
249 return "protocol", true
250 default:
251 return "", false
252 }
253}
254
255func emitReady(w io.Writer, r brokerReadyResult) error {
256 b, err := json.Marshal(r)
257 if err != nil || len(b)+1 > 4096 {
258 return &exitCoded{code: 1, err: errors.New("broker readiness output failure")}
259 }
260 if _, err = w.Write(append(b, '\n')); err != nil {
261 return &exitCoded{code: 1, err: errors.New("broker readiness output failure")}
262 }
263 switch r.Status {
264 case "ready":
265 return nil
266 case "timeout":
267 return &exitCoded{code: 3, err: errors.New("broker readiness timeout")}
268 case "terminal":
269 return &exitCoded{code: 4, err: errors.New("broker readiness terminal failure")}
270 case "canceled":
271 return &exitCoded{code: 5, err: context.Canceled}
272 default:
273 return &exitCoded{code: 3, err: errors.New("broker readiness timeout")}
274 }
275}
276func writeReady(w io.Writer, o brokerReadyOptions, status, reason string, local *ports.BrokerDaemonObservation, code int) error {
277 r := readyResult(o, status, reason, local)
278 err := emitReady(w, r)
279 if code != 0 {
280 return &exitCoded{code: code, err: fmt.Errorf("broker readiness %s", reason)}
281 }
282 return err
283}