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