1package daemon
2
3import (
4 "context"
5 "errors"
6 "sync"
7
8 "github.com/bnema/vev/internal/ports"
9)
10
11var errHandshakeTimeout = errors.New("handshake timed out")
12
13// newHandshakeContext owns one deadline for the complete inbound handshake.
14// The caller must stop it before entering the long-lived connection loop.
15func (d *Daemon) newHandshakeContext(parent context.Context) (context.Context, <-chan struct{}, func()) {
16 if parent == nil {
17 parent = context.Background()
18 }
19 ctx, cancel := context.WithCancel(parent)
20 timedOut := make(chan struct{})
21 clock := ports.Clock(systemClock{})
22 if d != nil && d.clock != nil {
23 clock = d.clock
24 }
25 timer := clock.NewTimer(ports.HandshakeTimeout)
26 if timer == nil {
27 timer = systemClock{}.NewTimer(ports.HandshakeTimeout)
28 }
29 stop := make(chan struct{})
30 done := make(chan struct{})
31 go func() {
32 defer close(done)
33 select {
34 case <-timer.C():
35 close(timedOut)
36 cancel()
37 case <-stop:
38 }
39 }()
40 var once sync.Once
41 finish := func() {
42 once.Do(func() {
43 close(stop)
44 <-done
45 timer.Stop()
46 cancel()
47 })
48 }
49 return ctx, timedOut, finish
50}
51
52// watchHandshakeTransport closes a transport when the handshake context ends.
53// Transport.Close is required to interrupt blocked Send and Recv operations.
54func watchHandshakeTransport(ctx context.Context, transport ports.Transport) func() {
55 stop := make(chan struct{})
56 done := make(chan struct{})
57 go func() {
58 defer close(done)
59 select {
60 case <-ctx.Done():
61 _ = transport.Close()
62 case <-stop:
63 }
64 }()
65 var once sync.Once
66 return func() {
67 once.Do(func() {
68 close(stop)
69 <-done
70 })
71 }
72}
73
74// boundedHandshakeOperation waits for one transport operation while the shared
75// handshake context remains live. Closing the transport is the interruption
76// mechanism because ports.Transport methods do not accept contexts.
77func boundedHandshakeOperation(ctx context.Context, transport ports.Transport, operation func() error) error {
78 _, err := boundedHandshakeOperationTracked(ctx, transport, operation)
79 return err
80}
81
82// boundedHandshakeOperationTracked also returns a completion signal for the
83// operation worker. Callers that own an effect gate must wait for this signal
84// before releasing the gate after cancellation; the transport operation itself
85// may still be unwinding after Close interrupts it.
86func boundedHandshakeOperationTracked(ctx context.Context, transport ports.Transport, operation func() error) (<-chan struct{}, error) {
87 done := make(chan struct{})
88 if err := ctx.Err(); err != nil {
89 close(done)
90 _ = transport.Close()
91 return done, err
92 }
93 completed := make(chan error, 1)
94 go func() {
95 defer close(done)
96 completed <- operation()
97 }()
98 select {
99 case err := <-completed:
100 if ctxErr := ctx.Err(); ctxErr != nil {
101 _ = transport.Close()
102 return done, ctxErr
103 }
104 return done, err
105 case <-ctx.Done():
106 _ = transport.Close()
107 // The result channel is buffered, so the operation can publish its
108 // completion after Close without keeping this cancellation path stuck.
109 return done, ctx.Err()
110 }
111}
112
113func handshakeContextError(parent context.Context, timedOut <-chan struct{}, fallback error) error {
114 select {
115 case <-timedOut:
116 return errors.Join(errHandshakeTimeout, context.DeadlineExceeded)
117 default:
118 }
119 if parent == nil {
120 return fallback
121 }
122 if err := parent.Err(); err != nil {
123 return err
124 }
125 return fallback
126}
127
128// failHandshakeAttachment synchronously retires the exact connection admitted
129// by route. A fresh route-owned session is purged only after its attachment is
130// removed; existing sessions and their unrelated attachments are preserved.
131func (d *Daemon) failHandshakeAttachment(sess *session, ac *attachedClient, tr ports.Transport, welcomed bool) {
132 if sess == nil || ac == nil {
133 if tr != nil {
134 _ = tr.Close()
135 }
136 return
137 }
138 if d.abortResumeClaim(ac) {
139 // abortResumeClaim closes the replacement transport and restores the
140 // original parked credential. Closing the exact handshake link again is
141 // harmless for transports and makes this boundary synchronous.
142 if tr != nil {
143 _ = tr.Close()
144 }
145 return
146 }
147 if welcomed {
148 d.clientGone(sess, ac, tr, true)
149 } else {
150 d.clientGoneWithoutNotice(sess, ac, tr, true)
151 }
152 if tr != nil {
153 _ = tr.Close()
154 }
155 if ac.routeCreatedSession {
156 sess.mu.Lock()
157 empty := len(sess.attachments) == 0
158 sess.mu.Unlock()
159 if empty {
160 // killSession revalidates the attachment set under its publication
161 // locks before removing the session.
162 _ = d.killSession(sess, ports.ReasonSessionKilled, ac.routeSessionPurge)
163 }
164 }
165}