handshake.go RawHistoryFolder

1package client
2
3import (
4	"context"
5	"errors"
6	"sync"
7	"time"
8
9	"github.com/bnema/vev/internal/ports"
10)
11
12var errHandshakeTimeout = errors.New("handshake timed out")
13
14type systemClock struct{}
15type systemTimer struct{ *time.Timer }
16
17func (systemClock) Now() time.Time { return time.Now() }
18func (systemClock) NewTimer(delay time.Duration) ports.Timer {
19	return systemTimer{Timer: time.NewTimer(delay)}
20}
21func (t systemTimer) C() <-chan time.Time        { return t.Timer.C }
22func (t systemTimer) Reset(d time.Duration) bool { return t.Timer.Reset(d) }
23func (t systemTimer) Stop() bool                 { return t.Timer.Stop() }
24
25// newHandshakeContext owns one deadline for the complete outbound handshake.
26// The caller must stop it before entering the long-lived connection loop.
27func newHandshakeContext(parent context.Context, clock ports.Clock) (context.Context, <-chan struct{}, func()) {
28	if parent == nil {
29		parent = context.Background()
30	}
31	ctx, cancel := context.WithCancel(parent)
32	timedOut := make(chan struct{})
33	if clock == nil {
34		clock = systemClock{}
35	}
36	timer := clock.NewTimer(ports.HandshakeTimeout)
37	if timer == nil {
38		timer = systemClock{}.NewTimer(ports.HandshakeTimeout)
39	}
40	stop := make(chan struct{})
41	done := make(chan struct{})
42	go func() {
43		defer close(done)
44		select {
45		case <-timer.C():
46			close(timedOut)
47			cancel()
48		case <-stop:
49		}
50	}()
51	var once sync.Once
52	finish := func() {
53		once.Do(func() {
54			close(stop)
55			<-done
56			timer.Stop()
57			cancel()
58		})
59	}
60	return ctx, timedOut, finish
61}
62
63// watchHandshakeTransport closes a transport when the handshake context ends.
64// Transport.Close is required to interrupt blocked Send and Recv operations.
65func watchHandshakeTransport(ctx context.Context, transport ports.Transport) func() {
66	stop := make(chan struct{})
67	done := make(chan struct{})
68	go func() {
69		defer close(done)
70		select {
71		case <-ctx.Done():
72			_ = transport.Close()
73		case <-stop:
74		}
75	}()
76	var once sync.Once
77	return func() {
78		once.Do(func() {
79			close(stop)
80			<-done
81		})
82	}
83}
84
85func boundedHandshakeOperation(ctx context.Context, transport ports.Transport, operation func() error) error {
86	if err := ctx.Err(); err != nil {
87		_ = transport.Close()
88		return err
89	}
90	completed := make(chan error, 1)
91	go func() { completed <- operation() }()
92	select {
93	case err := <-completed:
94		if ctxErr := ctx.Err(); ctxErr != nil {
95			_ = transport.Close()
96			return ctxErr
97		}
98		return err
99	case <-ctx.Done():
100		_ = transport.Close()
101		// The result channel is buffered, so the operation can publish its
102		// completion after Close without keeping this cancellation path stuck.
103		return ctx.Err()
104	}
105}
106
107func boundedDial(ctx context.Context, dialer ports.Dialer) (ports.Transport, error) {
108	// The dial context only bounds startup. A successful carriage outlives the
109	// handshake context; canceling that context after Welcome must not kill SSH.
110	dialCtx, cancel := context.WithCancel(context.WithoutCancel(ctx))
111	result := make(chan struct {
112		transport ports.Transport
113		err       error
114	}, 1)
115	go func() {
116		transport, err := dialer.Dial(dialCtx)
117		if err != nil {
118			cancel()
119		} else if ctx.Err() != nil && transport != nil {
120			_ = transport.Close()
121		}
122		result <- struct {
123			transport ports.Transport
124			err       error
125		}{transport: transport, err: err}
126	}()
127	select {
128	case result := <-result:
129		return result.transport, result.err
130	case <-ctx.Done():
131		cancel()
132		return nil, ctx.Err()
133	}
134}
135
136func handshakeContextError(parent context.Context, timedOut <-chan struct{}, fallback error) error {
137	select {
138	case <-timedOut:
139		return errors.Join(errHandshakeTimeout, context.DeadlineExceeded)
140	default:
141	}
142	if err := parent.Err(); err != nil {
143		return err
144	}
145	return fallback
146}