1package daemon
2
3import (
4 "context"
5 "errors"
6 "sync"
7 "time"
8
9 "github.com/bnema/vev/internal/ports"
10 "github.com/bnema/vev/internal/protocol"
11)
12
13const CommandRequestTimeout = 10 * time.Second
14
15var (
16 ErrCommandRequestTimeout = errors.New("command request timed out")
17 ErrCommandRequestUnavailable = errors.New("command request connection is unavailable")
18)
19
20// CommandRequestOutcome is the result delivered for one tracked request.
21type CommandRequestOutcome struct {
22 Result protocol.CommandResult
23 Err error
24}
25
26// CommandRequestTracker correlates request results for one connection. The
27// generation is part of every pending entry so a reused request ID from an
28// older connection cannot complete a newer request.
29type CommandRequestTracker struct {
30 mu sync.Mutex
31 next uint64
32 pending map[uint64]commandRequestPending
33}
34
35type commandRequestPending struct {
36 generation uint64
37 outcome chan CommandRequestOutcome
38}
39
40// NewCommandRequestTracker constructs an empty per-connection tracker.
41func NewCommandRequestTracker() *CommandRequestTracker {
42 return &CommandRequestTracker{}
43}
44
45// Publish allocates and records a request before its bytes are sent.
46func (t *CommandRequestTracker) Publish(generation uint64) (uint64, <-chan CommandRequestOutcome) {
47 t.mu.Lock()
48 defer t.mu.Unlock()
49 if t.pending == nil {
50 t.pending = make(map[uint64]commandRequestPending)
51 }
52 t.next++
53 if t.next == 0 {
54 t.next++
55 }
56 outcome := t.trackLocked(t.next, generation)
57 return t.next, outcome
58}
59
60// Track records an existing wire request ID, as used by an inbound one-shot
61// command connection. It rejects duplicate IDs on that connection.
62func (t *CommandRequestTracker) Track(requestID, generation uint64) (<-chan CommandRequestOutcome, bool) {
63 t.mu.Lock()
64 defer t.mu.Unlock()
65 if t.pending == nil {
66 t.pending = make(map[uint64]commandRequestPending)
67 }
68 if _, exists := t.pending[requestID]; exists {
69 return nil, false
70 }
71 if requestID > t.next {
72 t.next = requestID
73 }
74 outcome := t.trackLocked(requestID, generation)
75 return outcome, true
76}
77
78func (t *CommandRequestTracker) trackLocked(requestID, generation uint64) <-chan CommandRequestOutcome {
79 outcome := make(chan CommandRequestOutcome, 1)
80 t.pending[requestID] = commandRequestPending{generation: generation, outcome: outcome}
81 return outcome
82}
83
84// Wait waits without holding the tracker or connection sender lock. Timeout
85// and cancellation remove only the matching request.
86func (t *CommandRequestTracker) Wait(ctx context.Context, clock ports.Clock, requestID, generation uint64, outcome <-chan CommandRequestOutcome) (protocol.CommandResult, error) {
87 if t == nil || clock == nil {
88 return protocol.CommandResult{}, ErrCommandRequestUnavailable
89 }
90 if ctx == nil {
91 ctx = context.Background()
92 }
93 if err := ctx.Err(); err != nil {
94 t.Remove(requestID, generation)
95 return protocol.CommandResult{}, err
96 }
97 timer := clock.NewTimer(CommandRequestTimeout)
98 if timer == nil {
99 t.Remove(requestID, generation)
100 return protocol.CommandResult{}, ErrCommandRequestTimeout
101 }
102 defer timer.Stop()
103 select {
104 case completed := <-outcome:
105 return completed.Result, completed.Err
106 case <-ctx.Done():
107 t.Remove(requestID, generation)
108 return protocol.CommandResult{}, ctx.Err()
109 case <-timer.C():
110 t.Remove(requestID, generation)
111 return protocol.CommandResult{}, ErrCommandRequestTimeout
112 }
113}
114
115// Remove abandons only the exact request generation.
116func (t *CommandRequestTracker) Remove(requestID, generation uint64) {
117 if t == nil {
118 return
119 }
120 t.mu.Lock()
121 pending, ok := t.pending[requestID]
122 if ok && pending.generation == generation {
123 delete(t.pending, requestID)
124 }
125 t.mu.Unlock()
126}
127
128// Complete accepts only the exact request generation. Unknown, late, and old
129// generation results are safe no-ops.
130func (t *CommandRequestTracker) Complete(generation uint64, result protocol.CommandResult) {
131 t.finish(result.RequestID, generation, CommandRequestOutcome{Result: result})
132}
133
134// Fail completes the exact request with a transport or decode failure.
135func (t *CommandRequestTracker) Fail(requestID, generation uint64, err error) {
136 t.finish(requestID, generation, CommandRequestOutcome{Err: err})
137}
138
139func (t *CommandRequestTracker) finish(requestID, generation uint64, outcome CommandRequestOutcome) {
140 if t == nil {
141 return
142 }
143 t.mu.Lock()
144 pending, ok := t.pending[requestID]
145 if !ok || pending.generation != generation {
146 t.mu.Unlock()
147 return
148 }
149 delete(t.pending, requestID)
150 t.mu.Unlock()
151 pending.outcome <- outcome
152}
153
154// PendingCount is intended for deterministic lifecycle tests.
155func (t *CommandRequestTracker) PendingCount() int {
156 if t == nil {
157 return 0
158 }
159 t.mu.Lock()
160 defer t.mu.Unlock()
161 return len(t.pending)
162}