1package sessionwire
2
3import (
4 "context"
5 "sync"
6 "time"
7
8 "github.com/bnema/vev/internal/protocol/wire"
9)
10
11// preambleRunner runs one directional preamble and returns the ceilings it
12// negotiated.
13type preambleRunner func(context.Context, wire.Transport, protoCeilings) (protoCeilings, error)
14
15// handshake is the lazy, once-only preamble lifecycle shared by client and
16// server connections: the fixed absolute deadline, the negotiated ceilings,
17// the completion signal, and the completion hooks.
18type handshake struct {
19 raw wire.Transport
20 ceilings protoCeilings
21 // ceilingsMu guards ceilings. The lazy preamble publishes the negotiated
22 // values from whichever goroutine first uses the connection, while another
23 // goroutine may already be asking for capabilities or starting a send, so
24 // the once-guarded write is not by itself a happens-before edge for them.
25 ceilingsMu sync.Mutex
26
27 preambleOnce sync.Once
28 preambleErr error
29 deadline time.Time
30
31 // preambleDone closes exactly once when the handshake has run, whether it
32 // succeeded or failed. It backs the handshake completion seam a logical mux
33 // connection exposes; the absolute deadline is never restarted.
34 preambleDone chan struct{}
35
36 hooksMu sync.Mutex
37 handshakeDone bool
38 hooksRun bool
39 hooks []func()
40}
41
42func newHandshake(raw wire.Transport, deadline time.Time) handshake {
43 return handshake{raw: raw, ceilings: defaultProtoCeilings(), deadline: deadline, preambleDone: make(chan struct{})}
44}
45
46// HandshakeDeadline returns the absolute local deadline of the session
47// handshake this connection started with. It is the one accepted deadline the
48// preamble, Hello/Welcome, and first committed publication share; the value is
49// fixed for the connection's lifetime and is never restarted.
50func (h *handshake) HandshakeDeadline() time.Time { return h.deadline }
51
52// HandshakeDone returns a channel that is closed exactly once when the
53// handshake has run, whether it succeeded or failed.
54func (h *handshake) HandshakeDone() <-chan struct{} { return h.preambleDone }
55
56// OnHandshakeComplete registers fn to run exactly once when the handshake has
57// run. Registering after completion runs fn immediately; a nil fn is ignored.
58func (h *handshake) OnHandshakeComplete(fn func()) {
59 if fn == nil {
60 return
61 }
62 h.hooksMu.Lock()
63 if h.handshakeDone {
64 h.hooksMu.Unlock()
65 fn()
66 return
67 }
68 h.hooks = append(h.hooks, fn)
69 h.hooksMu.Unlock()
70}
71
72// ensurePreamble runs the preamble once, bounded by the handshake deadline.
73// A failed preamble closes the raw transport.
74func (h *handshake) ensurePreamble(run preambleRunner) error {
75 h.preambleOnce.Do(func() {
76 ctx, cancel := context.WithDeadline(context.Background(), h.deadline)
77 defer cancel()
78 next, err := run(ctx, h.raw, h.limits())
79 h.publishLimits(next)
80 h.preambleErr = err
81 if err != nil {
82 _ = h.raw.Close()
83 }
84 h.finishPreamble()
85 })
86 h.runHandshakeHooks()
87 return h.preambleErr
88}
89
90// finishPreamble publishes handshake completion from inside the preamble's
91// sync.Once: it closes the done channel and marks the handshake complete so a
92// later registration runs immediately. It deliberately runs no registered hook,
93// because a hook that calls back into the connection would re-enter sync.Once
94// and deadlock; runHandshakeHooks runs the hooks after the Once body returns.
95// It tolerates a zero-value handshake that never allocated the done channel
96// (test literals and seam-only construction), so closing is nil-safe.
97func (h *handshake) finishPreamble() {
98 if h.preambleDone != nil {
99 close(h.preambleDone)
100 }
101 h.hooksMu.Lock()
102 h.handshakeDone = true
103 h.hooksMu.Unlock()
104}
105
106// runHandshakeHooks runs the completion hooks registered before the handshake
107// finished exactly once, after the preamble's sync.Once body has returned.
108// Running them outside the Once is what lets a hook call back into the
109// connection - for example to send or receive a message - without re-entering
110// sync.Once and deadlocking. Every ensurePreamble caller invokes it; only the
111// first drains the hooks, and a registrant that arrives after completion
112// observes the completed flag and runs itself.
113func (h *handshake) runHandshakeHooks() {
114 h.hooksMu.Lock()
115 if h.hooksRun {
116 h.hooksMu.Unlock()
117 return
118 }
119 h.hooksRun = true
120 hooks := h.hooks
121 h.hooks = nil
122 h.hooksMu.Unlock()
123 for _, fn := range hooks {
124 fn()
125 }
126}
127
128// limits returns the negotiated ceilings under the lock the preamble publishes
129// them through. Send paths may read them without the lock once ensurePreamble
130// has returned, because that call orders the publish before the read.
131func (h *handshake) limits() protoCeilings {
132 h.ceilingsMu.Lock()
133 defer h.ceilingsMu.Unlock()
134 return h.ceilings
135}
136
137// publishLimits records the ceilings one preamble negotiation produced.
138func (h *handshake) publishLimits(next protoCeilings) {
139 h.ceilingsMu.Lock()
140 h.ceilings = next
141 h.ceilingsMu.Unlock()
142}