1package client
2
3import (
4 "context"
5 "errors"
6 "fmt"
7 "log/slog"
8 "time"
9
10 "github.com/bnema/vev/internal/domain"
11 "github.com/bnema/vev/internal/ports"
12 "github.com/bnema/vev/internal/protocol"
13)
14
15// Client control operations over independent broker streams.
16//
17// List, Command, Kill, KillAll, and StopDaemon are client use-case operations
18// that borrow a BrokerService and open one typed, independently cancellable
19// logical stream per attempt. The broker routes and admits; the daemon executes
20// and answers. No CLI code touches a ClientConnection or a daemon dialer after
21// cutover, and this object owns neither the service nor any attachment state:
22// each call owns and closes exactly its own stream.
23//
24// The correlation contract is deliberately strict:
25//
26// - One operation is one logical stream carrying one request and one reply;
27// streams are never pipelined, reused, or replayed.
28// - The exact epoch/connection/stream triplet names the attempt. On a
29// single-use stream the request ID is therefore the constant 1: the
30// correlation is per connection to a typed daemon, never global, so no RNG
31// and no fictitious deduplication key is introduced.
32// - Before the request reaches the wire, every refusal (invalid input, a
33// stale route, a failed open, an elapsed bound, an already-cancelled
34// caller) is a definite not-sent outcome wrapping
35// ErrBrokerOperationNotSent. Nothing was mutated.
36// - From the moment SendClient is entered, no failure proves non-delivery: a
37// send error, EOF, a lost broker, a timeout, a cancellation, or a reply
38// that is wrong-typed, wrongly correlated, or malformed all become a
39// synthetic, correctly correlated OutcomeUnknown with a nil error. The
40// caller inspects the outcome and never replays.
41// - A terminal, correctly correlated, well-formed reply is returned verbatim
42// with a nil error, including Failed and Unknown. A failure is never hidden
43// behind a plain context error and never converted into a nil result.
44//
45// Every operation shares one absolute bound of brokerOperationTimeout, driven
46// by the injected ports.Clock/ports.Timer and never restarted, which covers
47// acquisition (stream open) and the reply. Cancellation closes only the stream
48// this call owns, so it unblocks Send/Receive without disturbing the borrowed
49// service or any sibling stream. Timers and callbacks are disarmed and the
50// receive goroutine is joined before return, so a late reply can never change
51// the verdict.
52
53// ErrBrokerOperationNotSent reports a control operation that definitely never
54// reached the wire: validation, route authority, stream admission, or a
55// pre-dispatch cancellation refused it before any message was sent. The cause
56// stays available through errors.Is/As, and no mutation may have happened.
57var ErrBrokerOperationNotSent = errors.New("client: broker operation not sent")
58
59// errBrokerOperationLost reports a dispatched control request whose reply can
60// no longer be trusted: a send error, a physical loss, EOF, a timeout, a
61// cancellation, or a wrong/malformed reply. It is local-only and is folded into
62// a synthetic OutcomeUnknown for mutating operations; List reports it directly.
63var errBrokerOperationLost = errors.New("client: broker operation reply lost")
64
65// brokerOperationRequestID is the RequestID carried on a single-use control
66// stream. Correlation is by connection to a typed daemon, not globally, so a
67// constant is exactly sufficient; it is never derived from an RNG.
68const brokerOperationRequestID uint64 = 1
69
70// brokerOperationTimeout bounds every control operation end to end, from stream
71// acquisition through the terminal reply. It is the minimum with the caller's
72// own context and is never restarted.
73const brokerOperationTimeout = 10 * time.Second
74
75// BrokerOperationRoute identifies the exact broker epoch and destination one
76// control operation may address. A local destination is exactly Local=true with
77// no registration; a remote destination carries the complete registration
78// observed in a broker snapshot, never a bare hostname and never an invented
79// identity.
80type BrokerOperationRoute struct {
81 Epoch ports.BrokerEpoch
82 Destination ports.BrokerEndpointFence
83}
84
85// Validate checks the closed route shape: a non-zero epoch and a valid
86// destination fence. It performs no I/O and consults no snapshot.
87func (r BrokerOperationRoute) Validate() error {
88 if r.Epoch == 0 {
89 return errors.New("broker operation route carries no epoch")
90 }
91 if err := r.Destination.Validate(); err != nil {
92 return fmt.Errorf("broker operation route destination: %w", err)
93 }
94 return nil
95}
96
97// BrokerOperations executes client control operations over independent broker
98// streams. It is concurrent-safe, holds no attachment or picker state, never
99// closes the borrowed service, and exposes no way to open or retarget a
100// session.
101type BrokerOperations struct {
102 service ports.BrokerNavigator
103 clock ports.Clock
104}
105
106// NewBrokerOperations builds the executor over a borrowed broker service. It
107// refuses a nil or typed-nil service or clock so no operation can panic on a
108// missing dependency; the caller keeps ownership of the service lifetime.
109func NewBrokerOperations(service ports.BrokerNavigator, clock ports.Clock) (*BrokerOperations, error) {
110 if supervisorNil(service) {
111 return nil, errors.New("broker operations require a broker service")
112 }
113 if supervisorNil(clock) {
114 return nil, errors.New("broker operations require a clock")
115 }
116 return &BrokerOperations{service: service, clock: clock}, nil
117}
118
119// List opens one control stream, sends one protocol.List, and returns exactly
120// the daemon's protocol.Sessions. It never substitutes the broker cache for a
121// fresh reading and never fabricates an empty result when the daemon is
122// unreachable: a refused or lost listing is an ordinary error. A new explicit
123// list is always safe, so the implementation never retries on its own.
124func (o *BrokerOperations) List(ctx context.Context, route BrokerOperationRoute) ([]protocol.SessionInfo, error) {
125 return o.list(ctx, route, ports.BrokerDaemonExistingOnly)
126}
127
128// ListStarting is List that may start the destination daemon first, so the
129// reply includes sessions it restores from disk. Use it when the answer
130// decides an attach, where a stopped daemon must not read as "no sessions".
131func (o *BrokerOperations) ListStarting(ctx context.Context, route BrokerOperationRoute) ([]protocol.SessionInfo, error) {
132 return o.list(ctx, route, ports.BrokerDaemonStartIfNeeded)
133}
134
135func (o *BrokerOperations) list(ctx context.Context, route BrokerOperationRoute, mode ports.BrokerDaemonStartMode) ([]protocol.SessionInfo, error) {
136 reply, err := o.exchange(ctx, route, mode, protocol.List{})
137 if err != nil {
138 if errors.Is(err, errBrokerOperationLost) {
139 return nil, fmt.Errorf("reading session list: %w", err)
140 }
141 return nil, err
142 }
143 sessions, ok := reply.(protocol.Sessions)
144 if !ok {
145 logBrokerOperationCause("list reply", fmt.Errorf("unexpected reply %T", reply))
146 return nil, fmt.Errorf("unexpected reply %T to list", reply)
147 }
148 return sessions.Sessions, nil
149}
150
151// Command dispatches one control command on its own un-attached control
152// stream. Version must match the negotiated protocol, RequestID must be zero at
153// entry because this operation allocates it, and Attached and Self must be
154// false: the composition normalizes --self into an explicit target before
155// admission, so a Self request here has no usable context and is never sent.
156// Args is copied at admission, so a caller can never mutate a dispatched
157// request.
158func (o *BrokerOperations) Command(ctx context.Context, route BrokerOperationRoute, request protocol.CommandRequest) (protocol.CommandResult, error) {
159 if request.Version != protocol.Version {
160 return protocol.CommandResult{}, brokerOperationNotSent(fmt.Errorf("command protocol version %d is not supported", request.Version))
161 }
162 if request.RequestID != 0 {
163 return protocol.CommandResult{}, brokerOperationNotSent(errors.New("command request carries a request ID"))
164 }
165 if request.Attached {
166 return protocol.CommandResult{}, brokerOperationNotSent(errors.New("command request is attached"))
167 }
168 if request.Self {
169 return protocol.CommandResult{}, brokerOperationNotSent(errors.New("command request is not normalized to an explicit target"))
170 }
171 request.RequestID = brokerOperationRequestID
172 request.Args = append([]string(nil), request.Args...)
173
174 reply, err := o.exchange(ctx, route, ports.BrokerDaemonStartIfNeeded, request)
175 if err != nil {
176 if errors.Is(err, errBrokerOperationLost) {
177 o.reconcile(route)
178 return unknownBrokerCommandResult(brokerOperationRequestID, "command outcome unknown"), nil
179 }
180 return protocol.CommandResult{}, err
181 }
182 result, ok := reply.(protocol.CommandResult)
183 if !ok || result.RequestID != brokerOperationRequestID || !result.Valid() {
184 logBrokerOperationCause("command reply", fmt.Errorf("unexpected reply %T", reply))
185 o.reconcile(route)
186 return unknownBrokerCommandResult(brokerOperationRequestID, "command outcome unknown"), nil
187 }
188 if result.Outcome == protocol.CommandOutcomeUnknown {
189 o.reconcile(route)
190 }
191 return result, nil
192}
193
194// Kill asks the daemon to terminate one named session on its own control
195// stream. The name is validated session-name authority and addresses the
196// session's current incarnation at execution time, exactly as the v58 Kill
197// message carries; no lifecycle-exact guarantee is announced that the message
198// cannot carry.
199func (o *BrokerOperations) Kill(ctx context.Context, route BrokerOperationRoute, name string) (protocol.KillResult, error) {
200 if err := domain.ValidateSessionName(name); err != nil {
201 return protocol.KillResult{}, brokerOperationNotSent(fmt.Errorf("kill session name: %w", err))
202 }
203 return o.kill(ctx, route, ports.BrokerDaemonStartIfNeeded, protocol.Kill{
204 RequestID: brokerOperationRequestID,
205 Name: name,
206 Scope: protocol.KillSession,
207 })
208}
209
210// KillAll asks the daemon to purge every session on its own control stream
211// with an empty name and the KillAll scope. It never stops the daemon: final
212// session removal is not a shutdown.
213func (o *BrokerOperations) KillAll(ctx context.Context, route BrokerOperationRoute) (protocol.KillResult, error) {
214 return o.kill(ctx, route, ports.BrokerDaemonStartIfNeeded, protocol.Kill{
215 RequestID: brokerOperationRequestID,
216 Scope: protocol.KillAll,
217 })
218}
219
220// StopDaemon asks the daemon to shut down on its own control stream with an
221// empty name and the KillDaemon scope, under the ExistingOnly start mode so the
222// stop can never start the target it is stopping. A success is exactly one
223// correlated KillSucceeded: a close or EOF is never a success, and a lost
224// acknowledgement is OutcomeUnknown.
225func (o *BrokerOperations) StopDaemon(ctx context.Context, route BrokerOperationRoute) (protocol.KillResult, error) {
226 return o.kill(ctx, route, ports.BrokerDaemonExistingOnly, protocol.Kill{
227 RequestID: brokerOperationRequestID,
228 Scope: protocol.KillDaemon,
229 })
230}
231
232// kill runs one Kill-family operation and maps its reply through the shared
233// correlation table: a lost or wrong/malformed reply becomes a synthetic
234// OutcomeUnknown with a nil error, while a terminal reply is returned verbatim.
235func (o *BrokerOperations) kill(ctx context.Context, route BrokerOperationRoute, mode ports.BrokerDaemonStartMode, request protocol.Kill) (protocol.KillResult, error) {
236 reply, err := o.exchange(ctx, route, mode, request)
237 if err != nil {
238 if errors.Is(err, errBrokerOperationLost) {
239 o.reconcile(route)
240 return unknownBrokerKillResult(brokerOperationRequestID, "kill outcome unknown"), nil
241 }
242 return protocol.KillResult{}, err
243 }
244 result, ok := reply.(protocol.KillResult)
245 if !ok || !validBrokerKillResult(result, brokerOperationRequestID) {
246 logBrokerOperationCause("kill reply", fmt.Errorf("unexpected reply %T", reply))
247 o.reconcile(route)
248 return unknownBrokerKillResult(brokerOperationRequestID, "kill outcome unknown"), nil
249 }
250 if result.Outcome == protocol.KillOutcomeUnknown {
251 o.reconcile(route)
252 }
253 return result, nil
254}
255
256// exchange owns the whole lifecycle of one attempt: route revalidation against
257// the current snapshot, one allocated stream identity, one bounded open, one
258// dispatch, and exactly one correlated reply.
259//
260// It reports a definite not-sent outcome (wrapping ErrBrokerOperationNotSent)
261// for every refusal that happens before SendClient is entered, and
262// errBrokerOperationLost once the request may have reached the wire. The
263// caller decides how each class maps onto a result.
264func (o *BrokerOperations) exchange(ctx context.Context, route BrokerOperationRoute, mode ports.BrokerDaemonStartMode, message protocol.ClientMessage) (protocol.ServerMessage, error) {
265 if err := route.Validate(); err != nil {
266 return nil, brokerOperationNotSent(err)
267 }
268 opCtx, _, finish := newBoundedContext(ctx, o.clock, brokerOperationTimeout)
269 defer finish()
270 if err := opCtx.Err(); err != nil {
271 return nil, brokerOperationNotSent(err)
272 }
273 stream, err := o.openControlStream(opCtx, route, mode)
274 if err != nil {
275 return nil, brokerOperationNotSent(err)
276 }
277 // The stream is owned by exactly this attempt: every exit path closes it.
278 // Closing only this stream never disturbs the borrowed service or a sibling
279 // stream, and the BrokerLogicalConnection contract requires Close to unblock
280 // Send/Receive, so cancellation needs no service shutdown.
281 defer func() { _ = stream.Close() }()
282 if err := opCtx.Err(); err != nil {
283 return nil, brokerOperationNotSent(err)
284 }
285 stop := context.AfterFunc(opCtx, func() { _ = stream.Close() })
286 defer stop()
287
288 if err := stream.SendClient(message); err != nil {
289 // The request may have been partially delivered, so a send error never
290 // proves the daemon did not act.
291 logBrokerOperationCause("dispatch", err)
292 return nil, errBrokerOperationLost
293 }
294 replies := make(chan brokerOperationReply, 1)
295 go func() {
296 reply, recvErr := stream.ReceiveServer()
297 replies <- brokerOperationReply{message: reply, err: recvErr}
298 }()
299 var reply brokerOperationReply
300 select {
301 case reply = <-replies:
302 // The bound is authoritative: if it lapsed while the reply was being
303 // read, the reply is late and can never upgrade the verdict back to a
304 // terminal result. Only a reply observed strictly inside the bound is
305 // trusted.
306 if opCtx.Err() != nil {
307 return nil, errBrokerOperationLost
308 }
309 case <-opCtx.Done():
310 // The bound lapsed or the caller cancelled before a reply was observed:
311 // close the stream to unblock the read, join the reader, and keep the
312 // verdict indeterminate. The channel is buffered so the reader always
313 // publishes, and a reply that arrives concurrently with the bound can
314 // never upgrade this attempt back to a terminal result.
315 _ = stream.Close()
316 <-replies
317 return nil, errBrokerOperationLost
318 }
319 if reply.err != nil {
320 logBrokerOperationCause("reply", reply.err)
321 return nil, errBrokerOperationLost
322 }
323 return reply.message, nil
324}
325
326// brokerOperationReply is one joined read result from a control stream.
327type brokerOperationReply struct {
328 message protocol.ServerMessage
329 err error
330}
331
332// openControlStream revalidates the route against the current snapshot, then
333// allocates one stream identity and opens the exact control stream it names.
334// Policy is copied from the snapshot's selected observation and the process
335// environment is never read.
336func (o *BrokerOperations) openControlStream(ctx context.Context, route BrokerOperationRoute, mode ports.BrokerDaemonStartMode) (ports.BrokerLogicalConnection, error) {
337 snapshot := o.service.Snapshot()
338 if err := snapshot.Validate(); err != nil {
339 return nil, fmt.Errorf("broker catalogue is unavailable: %w", err)
340 }
341 if route.Epoch != snapshot.Epoch {
342 return nil, fmt.Errorf("broker operation route epoch %d does not match the current epoch %d", route.Epoch, snapshot.Epoch)
343 }
344 observation, ok := brokerOperationAuthority(snapshot, route.Destination)
345 if !ok {
346 return nil, errors.New("broker operation destination is not in the current snapshot")
347 }
348 stream, err := o.service.NextStreamID()
349 if err != nil {
350 return nil, err
351 }
352 request := ports.BrokerOpenStreamRequest{
353 Epoch: route.Epoch,
354 Purpose: ports.BrokerStreamControl,
355 Local: route.Destination.Local,
356 Connection: o.service.ConnectionID(),
357 Stream: stream,
358 Policy: observation.Policy,
359 StartMode: mode,
360 }
361 if !route.Destination.Local {
362 request.Endpoint = observation.Endpoint
363 request.Registration = observation.Registration
364 }
365 if err := request.Validate(); err != nil {
366 return nil, err
367 }
368 return o.service.OpenStream(ctx, request)
369}
370
371// brokerOperationAuthority selects the exact observation one destination
372// names: the local daemon for a local destination, or the observation whose
373// complete registration (endpoint, incarnation, and generation) is exactly
374// equal for a remote one. A missing or replaced registration is never resolved
375// by name or by fallback.
376func brokerOperationAuthority(snapshot ports.BrokerSnapshot, destination ports.BrokerEndpointFence) (ports.BrokerDaemonObservation, bool) {
377 for i := range snapshot.Daemons {
378 observation := snapshot.Daemons[i]
379 if destination.Local {
380 if observation.Local {
381 return observation, true
382 }
383 continue
384 }
385 if observation.Local {
386 continue
387 }
388 if observation.Registration.Equal(destination.Registration) {
389 return observation, true
390 }
391 }
392 return ports.BrokerDaemonObservation{}, false
393}
394
395// reconcile asks for one coalesced re-observation of the operation's
396// destination after an indeterminate outcome. The hint is non-blocking and is
397// never awaited: it cannot prove the mutation succeeded, and a broker loss may
398// simply drop it, after which the client resynchronizes normally. The local
399// destination is normalized to the empty endpoint in both implementations.
400func (o *BrokerOperations) reconcile(route BrokerOperationRoute) {
401 if route.Destination.Local {
402 o.service.RequestReconcile("")
403 return
404 }
405 o.service.RequestReconcile(route.Destination.Registration.Endpoint)
406}
407
408// brokerOperationNotSent wraps a definite pre-dispatch refusal with the shared
409// sentinel while keeping the exact cause reachable.
410func brokerOperationNotSent(cause error) error {
411 if cause == nil {
412 return ErrBrokerOperationNotSent
413 }
414 return fmt.Errorf("%w: %w", ErrBrokerOperationNotSent, cause)
415}
416
417// unknownBrokerCommandResult builds a correctly correlated synthetic Unknown
418// command outcome: the expected request ID, code zero, bounded presentation
419// text, and no output. The diagnostic cause is logged, never placed in the
420// result text.
421func unknownBrokerCommandResult(requestID uint64, text string) protocol.CommandResult {
422 return protocol.CommandResult{
423 RequestID: requestID,
424 Outcome: protocol.CommandOutcomeUnknown,
425 Text: text,
426 }
427}
428
429// unknownBrokerKillResult builds a correctly correlated synthetic Unknown kill
430// outcome: the expected request ID, code zero, bounded presentation text, and
431// no failures.
432func unknownBrokerKillResult(requestID uint64, text string) protocol.KillResult {
433 return protocol.KillResult{
434 RequestID: requestID,
435 Outcome: protocol.KillOutcomeUnknown,
436 Text: text,
437 }
438}
439
440// validBrokerKillResult reports whether a decoded kill reply is well formed and
441// correlated to the expected request. The daemon's success and failure shapes
442// require a consistent code, while a daemon-reported Unknown may carry a
443// shutdown code.
444func validBrokerKillResult(result protocol.KillResult, requestID uint64) bool {
445 if result.RequestID != requestID {
446 return false
447 }
448 switch result.Outcome {
449 case protocol.KillSucceeded:
450 return result.Code == 0
451 case protocol.KillFailed:
452 return result.Code != 0
453 case protocol.KillOutcomeUnknown:
454 return true
455 default:
456 return false
457 }
458}
459
460// logBrokerOperationCause records a control-operation diagnostic without ever
461// copying it into presentation text.
462func logBrokerOperationCause(stage string, err error) {
463 if err == nil {
464 return
465 }
466 slog.Default().Debug("broker control operation indeterminate", "stage", stage, "err", err)
467}