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
13var errHandshakeTimeout = errors.New("handshake timed out")
14
15// acceptedHandshakeDeadline returns the absolute handshake deadline an accepted
16// connection already owns, or the zero time when the connection exposes no
17// deadline seam and the daemon must start a fresh budget. The seam is asserted
18// structurally so the daemon neither imports the mux/sessionwire adapters nor
19// widens the ServerConnection contract.
20func acceptedHandshakeDeadline(tr ports.ServerConnection) time.Time {
21 provider, ok := tr.(ports.HandshakeDeadlineProvider)
22 if !ok {
23 return time.Time{}
24 }
25 return provider.HandshakeDeadline()
26}
27
28// newHandshakeContext owns one deadline for the complete inbound handshake.
29// acceptedDeadline is the absolute deadline an accepted connection already
30// owns; the zero time starts a fresh protocol.HandshakeTimeout budget, which
31// preserves the behavior of ordinary connections lacking the optional seam.
32// A non-zero acceptedDeadline is adopted verbatim: the timer covers only the
33// time the connection has left, so queue delay is consumed instead of
34// restarting the budget, and an already-elapsed deadline expires immediately.
35// The caller must stop it before entering the long-lived connection loop.
36func (d *Daemon) newHandshakeContext(parent context.Context, acceptedDeadline time.Time) (context.Context, <-chan struct{}, func()) {
37 if parent == nil {
38 parent = context.Background()
39 }
40 ctx, cancel := context.WithCancel(parent)
41 timedOut := make(chan struct{})
42 clock := ports.Clock(systemClock{})
43 if d != nil && d.clock != nil {
44 clock = d.clock
45 }
46 budget := protocol.HandshakeTimeout
47 if !acceptedDeadline.IsZero() {
48 budget = acceptedDeadline.Sub(clock.Now())
49 }
50 if budget <= 0 {
51 // The accepted connection already spent its whole budget before the
52 // daemon reached it. Expire now rather than start a second budget; the
53 // closed timedOut channel keeps handshakeContextError reporting a
54 // deadline rather than a bare cancellation.
55 close(timedOut)
56 cancel()
57 return ctx, timedOut, cancel
58 }
59 timer := clock.NewTimer(budget)
60 if timer == nil {
61 timer = systemClock{}.NewTimer(budget)
62 }
63 stop := make(chan struct{})
64 done := make(chan struct{})
65 go func() {
66 defer close(done)
67 select {
68 case <-timer.C():
69 close(timedOut)
70 cancel()
71 case <-stop:
72 }
73 }()
74 var once sync.Once
75 finish := func() {
76 once.Do(func() {
77 close(stop)
78 <-done
79 timer.Stop()
80 cancel()
81 })
82 }
83 return ctx, timedOut, finish
84}
85
86// watchHandshakeTransport closes a transport when the handshake context ends.
87// Transport.Close is required to interrupt blocked Send and Recv operations.
88func watchHandshakeTransport(ctx context.Context, transport ports.ServerConnection) func() {
89 stop := make(chan struct{})
90 done := make(chan struct{})
91 go func() {
92 defer close(done)
93 select {
94 case <-ctx.Done():
95 _ = transport.Close()
96 case <-stop:
97 }
98 }()
99 var once sync.Once
100 return func() {
101 once.Do(func() {
102 close(stop)
103 <-done
104 })
105 }
106}
107
108// boundedHandshakeOperation waits for one transport operation while the shared
109// handshake context remains live. Closing the transport is the interruption
110// mechanism because ServerConnection methods do not accept contexts.
111func boundedHandshakeOperation(ctx context.Context, transport ports.ServerConnection, operation func() error) error {
112 _, err := boundedHandshakeOperationTracked(ctx, transport, operation)
113 return err
114}
115
116// boundedHandshakeOperationTracked also returns a completion signal for the
117// operation worker. Callers that own an effect gate must wait for this signal
118// before releasing the gate after cancellation; the transport operation itself
119// may still be unwinding after Close interrupts it.
120func boundedHandshakeOperationTracked(ctx context.Context, transport ports.ServerConnection, operation func() error) (<-chan struct{}, error) {
121 done := make(chan struct{})
122 if err := ctx.Err(); err != nil {
123 close(done)
124 _ = transport.Close()
125 return done, err
126 }
127 completed := make(chan error, 1)
128 go func() {
129 defer close(done)
130 completed <- operation()
131 }()
132 select {
133 case err := <-completed:
134 if ctxErr := ctx.Err(); ctxErr != nil {
135 _ = transport.Close()
136 return done, ctxErr
137 }
138 return done, err
139 case <-ctx.Done():
140 _ = transport.Close()
141 // The result channel is buffered, so the operation can publish its
142 // completion after Close without keeping this cancellation path stuck.
143 return done, ctx.Err()
144 }
145}
146
147func handshakeContextError(parent context.Context, timedOut <-chan struct{}, fallback error) error {
148 select {
149 case <-timedOut:
150 return errors.Join(errHandshakeTimeout, context.DeadlineExceeded)
151 default:
152 }
153 if parent == nil {
154 return fallback
155 }
156 if err := parent.Err(); err != nil {
157 return err
158 }
159 return fallback
160}
161
162// failHandshakeAttachment synchronously retires the exact connection admitted
163// by route. A fresh route-owned session is purged only after its attachment is
164// removed; existing sessions and their unrelated attachments are preserved.
165func (d *Daemon) failHandshakeAttachment(sess *session, ac *attachedClient, tr ports.ServerConnection, welcomed bool) {
166 if sess == nil || ac == nil {
167 if tr != nil {
168 _ = tr.Close()
169 }
170 return
171 }
172 if d.abortResumeClaim(ac) {
173 // abortResumeClaim closes the replacement transport and restores the
174 // original parked credential. Closing the exact handshake link again is
175 // harmless for transports and makes this boundary synchronous.
176 if tr != nil {
177 _ = tr.Close()
178 }
179 return
180 }
181 if welcomed {
182 // A failed handshake after Welcome is not an explicit user detach. The
183 // client must return to its picker and show the destination failure.
184 d.clientGoneWithNoticeReason(sess, ac, tr, true, true, protocol.ReasonServerShutdown)
185 } else {
186 d.clientGoneWithoutNotice(sess, ac, tr, true)
187 }
188 if tr != nil {
189 _ = tr.Close()
190 }
191 if ac.routeCreatedSession {
192 _ = d.killSessionIfEmpty(sess, protocol.ReasonSessionKilled, ac.routeSessionPurge)
193 }
194}