1package sessionwire
2
3import (
4 "context"
5 "errors"
6 "sync"
7 "time"
8
9 "google.golang.org/protobuf/proto"
10
11 "github.com/bnema/vev/internal/ports"
12 "github.com/bnema/vev/internal/protocol"
13 "github.com/bnema/vev/internal/protocol/wire"
14)
15
16type clientConnection struct {
17 raw wire.Transport
18 ceilings protoCeilings
19 // ceilingsMu guards ceilings. The lazy preamble publishes the negotiated
20 // values from whichever goroutine first uses the connection, while another
21 // goroutine may already be asking for capabilities or starting a send, so
22 // the once-guarded write is not by itself a happens-before edge for them.
23 ceilingsMu sync.Mutex
24
25 preambleOnce sync.Once
26 preambleErr error
27 deadline time.Time
28
29 // preambleDone closes exactly once when the handshake has run, whether it
30 // succeeded or failed. It backs the handshake completion hook that a logical
31 // mux connection exposes; the absolute deadline is never restarted.
32 preambleDone chan struct{}
33
34 hooksMu sync.Mutex
35 handshakeDone bool
36 hooksRun bool
37 hooks []func()
38}
39
40var _ ports.ClientConnection = (*clientConnection)(nil)
41
42// NewClientConnection wraps one raw client connection incarnation. The
43// preamble runs lazily on first use, bounded by the handshake deadline
44// started here.
45func NewClientConnection(raw wire.Transport) ports.ClientConnection {
46 if raw == nil {
47 return nil
48 }
49 return &clientConnection{raw: raw, ceilings: defaultProtoCeilings(), deadline: time.Now().Add(protocol.HandshakeTimeout), preambleDone: make(chan struct{})}
50}
51
52// HandshakeDeadline returns the absolute local deadline of the session
53// handshake this connection started with. It is the one accepted deadline the
54// preamble, Hello/Welcome, and first committed publication share; the value is
55// fixed for the connection's lifetime and is never restarted.
56func (c *clientConnection) HandshakeDeadline() time.Time { return c.deadline }
57
58// HandshakeDone returns a channel that is closed exactly once when the
59// handshake has run, whether it succeeded or failed.
60func (c *clientConnection) HandshakeDone() <-chan struct{} { return c.preambleDone }
61
62// OnHandshakeComplete registers fn to run exactly once when the handshake has
63// run. Registering after completion runs fn immediately; a nil fn is ignored.
64func (c *clientConnection) OnHandshakeComplete(fn func()) {
65 if fn == nil {
66 return
67 }
68 c.hooksMu.Lock()
69 if c.handshakeDone {
70 c.hooksMu.Unlock()
71 fn()
72 return
73 }
74 c.hooks = append(c.hooks, fn)
75 c.hooksMu.Unlock()
76}
77
78// finishPreamble publishes handshake completion from inside the preamble's
79// sync.Once: it closes the done channel and marks the handshake complete so a
80// later registration runs immediately. It deliberately runs no registered hook,
81// because a hook that calls back into the connection would re-enter sync.Once
82// and deadlock; runHandshakeHooks runs the hooks after the Once body returns.
83// It tolerates a zero-value connection that never allocated the done channel
84// (test literals and seam-only construction), so closing is nil-safe.
85func (c *clientConnection) finishPreamble() {
86 if c.preambleDone != nil {
87 close(c.preambleDone)
88 }
89 c.hooksMu.Lock()
90 c.handshakeDone = true
91 c.hooksMu.Unlock()
92}
93
94// runHandshakeHooks runs the completion hooks registered before the handshake
95// finished exactly once, after the preamble's sync.Once body has returned.
96// Running them outside the Once is what lets a hook call back into the
97// connection - for example to send a message - without re-entering sync.Once
98// and deadlocking. Every ensurePreamble caller invokes it; only the first
99// drains the hooks, and a registrant that arrives after completion observes the
100// completed flag and runs itself.
101func (c *clientConnection) runHandshakeHooks() {
102 c.hooksMu.Lock()
103 if c.hooksRun {
104 c.hooksMu.Unlock()
105 return
106 }
107 c.hooksRun = true
108 hooks := c.hooks
109 c.hooks = nil
110 c.hooksMu.Unlock()
111 for _, fn := range hooks {
112 fn()
113 }
114}
115
116func (c *clientConnection) ensurePreamble() error {
117 c.preambleOnce.Do(func() {
118 ctx, cancel := context.WithDeadline(context.Background(), c.deadline)
119 defer cancel()
120 next, err := runProtoClientPreamble(ctx, c.raw, c.limits())
121 c.publishLimits(next)
122 c.preambleErr = err
123 if err != nil {
124 _ = c.raw.Close()
125 }
126 c.finishPreamble()
127 })
128 c.runHandshakeHooks()
129 return c.preambleErr
130}
131
132// limits returns the negotiated ceilings under the lock the preamble publishes
133// them through.
134func (c *clientConnection) limits() protoCeilings {
135 c.ceilingsMu.Lock()
136 defer c.ceilingsMu.Unlock()
137 return c.ceilings
138}
139
140// publishLimits records the ceilings one preamble negotiation produced.
141func (c *clientConnection) publishLimits(next protoCeilings) {
142 c.ceilingsMu.Lock()
143 c.ceilings = next
144 c.ceilingsMu.Unlock()
145}
146
147func (c *clientConnection) SendClient(message protocol.ClientMessage) error {
148 if err := c.ensurePreamble(); err != nil {
149 return err
150 }
151 envelope, err := encodeProtoClient(message)
152 if err != nil {
153 return err
154 }
155 raw, err := proto.Marshal(envelope)
156 if err != nil {
157 return err
158 }
159 if err := checkCategoryCeiling(raw, clientEnvelopeCategory(raw), c.limits().maxReceiveEnvelopeBytes); err != nil {
160 return err
161 }
162 return c.raw.Send(wire.Envelope{Payload: raw})
163}
164
165func (c *clientConnection) ReceiveServer() (protocol.ServerMessage, error) {
166 if err := c.ensurePreamble(); err != nil {
167 return nil, err
168 }
169 envelope, err := c.raw.Recv()
170 if err != nil {
171 return nil, err
172 }
173 limits := c.limits()
174 message, decodeErr := decodeServerEnvelopeWithin(envelope.Payload, limits.maxReceiveEnvelopeBytes, limits.outputDataLimit)
175 if decodeErr == nil {
176 return message, nil
177 }
178 return nil, decodeErr
179}
180
181func (c *clientConnection) Capabilities() protocol.ConnectionCapabilities {
182 return rawCapabilities(c.raw, c.limits().outputDataLimit)
183}
184
185func (c *clientConnection) LinkState() ports.LinkState { return rawLinkState(c.raw) }
186func (c *clientConnection) LinkEvents() <-chan ports.LinkEvent { return rawLinkEvents(c.raw) }
187func (c *clientConnection) Close() error { return c.raw.Close() }
188
189// DecodeServerEnvelope unwraps one scanned server envelope payload into
190// its semantic message for hidden one-shot carriages.
191func DecodeServerEnvelope(payload []byte) (protocol.ServerMessage, error) {
192 return decodeServerEnvelope(payload)
193}
194
195func decodeServerEnvelope(payload []byte) (protocol.ServerMessage, error) {
196 return decodeServerEnvelopeWithin(payload, wire.AbsoluteEnvelopeLimit, uint64(protocol.MaxOutputDataLen))
197}
198
199// decodeServerEnvelopeWithin unwraps one server envelope under the
200// negotiated receive ceilings: the serialized envelope must fit
201// maxReceiveEnvelopeBytes and any Output must declare uncompressed data at
202// or below outputDataLimit. The one-shot exported decoder uses the absolute
203// ceilings because no preamble was negotiated there.
204func decodeServerEnvelopeWithin(payload []byte, envelopeLimit, outputDataLimit uint64) (protocol.ServerMessage, error) {
205 envelope := &wire.ServerEnvelope{}
206 if err := checkCategoryCeiling(payload, serverEnvelopeCategory(payload), envelopeLimit); err != nil {
207 return nil, classifyEnvelopeError(err, envelope)
208 }
209 if err := wire.ScanEnvelope(envelope, payload); err != nil {
210 return nil, classifyEnvelopeError(err, envelope)
211 }
212 if err := (proto.UnmarshalOptions{DiscardUnknown: false}.Unmarshal(payload, envelope)); err != nil {
213 return nil, &protocol.DecodeFailure{Category: protocol.DecodeMalformed, Err: err}
214 }
215 if err := checkOutputDataCeiling(envelope, outputDataLimit); err != nil {
216 return nil, &protocol.DecodeFailure{Category: protocol.DecodeMalformed, Err: err}
217 }
218 message, err := decodeProtoServer(envelope)
219 if err != nil {
220 return nil, classifyEnvelopeError(err, envelope)
221 }
222 return message, nil
223}
224
225func classifyEnvelopeError(err error, _ *wire.ServerEnvelope) *protocol.DecodeFailure {
226 failure := &protocol.DecodeFailure{Category: protocol.DecodeMalformed, Err: err}
227 if errors.Is(err, wire.ErrScanUnknown) {
228 failure.Category = protocol.DecodeUnknownType
229 } else if errors.Is(err, ErrWrongDirection) {
230 failure.Category = protocol.DecodeWrongDirection
231 }
232 return failure
233}