1// Raw-carrier bridge.
2//
3// One daemonmux physical connection rides an existing raw framed transport:
4// IPC, QUIC, or SSH stdio, each of which already owns vev's shared 4-byte
5// big-endian length framing and a caller-bounded receive (wire.BoundedTransport).
6// This file adapts that raw carriage to the private FramedCarrier the pump
7// (pump.go) and the physical preamble handshake (handshake.go) drive. It
8// introduces no socket, no framing of its own, no listener, and no sessionwire.
9//
10// The bridge is narrow and bounded. Send is synchronous: it hands one envelope
11// to the raw transport and returns the raw write result, so no error means the
12// carriage accepted the frame. Receive is bounded to wire.PreambleLimit while
13// the connection is in its preamble phase, and the explicit, one-way Negotiate
14// transition raises that bound to the daemonmux envelope ceiling the physical
15// preamble negotiated: a receive never allocates a body larger than the bound
16// that applied when it started, and an oversized application frame after
17// negotiation is refused rather than accepted. Negotiate must run before the
18// pump that reads application frames; a receive already in flight keeps the
19// bound it started with.
20//
21// Close is idempotent and concurrent-safe: it marks the bridge closed, closes
22// the raw transport through the adapter prompt-close contract so a blocked
23// Send or RecvBounded releases, and then waits for every bridge operation that
24// was already admitted. Send and Receive honour their context the same way:
25// cancelling it closes the raw transport to release the blocked call - a
26// carriage that can no longer make progress is terminal - and joins the one
27// worker it started, so no goroutine is abandoned. Because Close joins the
28// operations it admitted and each operation joins its own worker, the bridge
29// owns no long-lived goroutine.
30package daemonmux
31
32import (
33 "context"
34 "errors"
35 "sync"
36 "sync/atomic"
37
38 "github.com/bnema/vev/internal/protocol/wire"
39)
40
41var (
42 // ErrCarrierState reports an invalid construction or state transition: a
43 // nil raw transport, or a second Negotiate once the preamble phase ended.
44 ErrCarrierState = errors.New("daemonmux: invalid carrier state")
45 // ErrCarrierClosed reports Send or Receive on a bridge that was closed.
46 ErrCarrierClosed = errors.New("daemonmux: carrier closed")
47)
48
49// RawFramedTransport is the narrow raw framed carriage the bridge adapts: Send
50// writes one complete envelope synchronously, RecvBounded reads one complete
51// envelope bounded to limit before allocating or reading its body, and Close
52// releases blocked I/O. Every wire.BoundedTransport satisfies it; the bridge
53// deliberately does not require the unbounded Recv it never calls.
54type RawFramedTransport interface {
55 Send(envelope wire.Envelope) error
56 RecvBounded(limit uint64) (wire.Envelope, error)
57 Close() error
58}
59
60// Every raw wire.BoundedTransport satisfies the bridge's narrow capability.
61var _ RawFramedTransport = (wire.BoundedTransport)(nil)
62
63// FramedCarrierBridge adapts a RawFramedTransport to the private FramedCarrier
64// contract. It is safe for concurrent use: Send and Receive may run from
65// different goroutines, Close is idempotent and concurrent-safe, and the
66// receive bound is read atomically so Negotiate never races a receive.
67type FramedCarrierBridge struct {
68 raw RawFramedTransport
69
70 // limit is the receive bound the next Receive applies: wire.PreambleLimit
71 // until Negotiate records the negotiated envelope ceiling.
72 limit atomic.Uint64
73 negotiated atomic.Bool
74
75 // mu guards closed and closeErr and serializes operation admission against
76 // Close, so ops is never Add-ed concurrently with Wait.
77 mu sync.Mutex
78 closed bool
79 closeErr error
80
81 // done is closed exactly once by Close so a blocked operation wakes.
82 done chan struct{}
83
84 // ops counts admitted operations between admission and return. Close waits
85 // for it after closing the raw transport, which is what guarantees no
86 // bridge worker outlives Close.
87 ops sync.WaitGroup
88
89 rawCloseOnce sync.Once
90 closeOnce sync.Once
91}
92
93var _ FramedCarrier = (*FramedCarrierBridge)(nil)
94
95// NewPreambleCarrier wraps raw in its preamble phase: every Receive is bounded
96// to wire.PreambleLimit until Negotiate raises the bound to the negotiated
97// daemonmux envelope ceiling. A nil raw transport is refused with
98// ErrCarrierState.
99func NewPreambleCarrier(raw RawFramedTransport) (*FramedCarrierBridge, error) {
100 if raw == nil {
101 return nil, ErrCarrierState
102 }
103 c := &FramedCarrierBridge{raw: raw, done: make(chan struct{})}
104 c.limit.Store(wire.PreambleLimit)
105 return c, nil
106}
107
108// Negotiate records the connection's negotiated ceilings and raises the receive
109// bound from wire.PreambleLimit to ceilings.MaxReceiveEnvelopeBytes. It is the
110// explicit, one-way state transition out of the preamble phase: an invalid
111// advertisement, a second call, or a call after Close is refused rather than
112// silently widening the bound. The transition must precede the pump that reads
113// application frames; a receive already in flight keeps its old bound.
114func (c *FramedCarrierBridge) Negotiate(ceilings MuxCeilings) error {
115 if c == nil || c.raw == nil {
116 return ErrCarrierState
117 }
118 if err := ceilings.Validate(); err != nil {
119 return err
120 }
121 c.mu.Lock()
122 defer c.mu.Unlock()
123 if c.closed {
124 return ErrCarrierClosed
125 }
126 if c.negotiated.Load() {
127 return ErrCarrierState
128 }
129 c.limit.Store(ceilings.MaxReceiveEnvelopeBytes)
130 c.negotiated.Store(true)
131 return nil
132}
133
134// ReceiveLimit returns the bound the next Receive applies: wire.PreambleLimit
135// before Negotiate and the negotiated envelope ceiling afterwards.
136func (c *FramedCarrierBridge) ReceiveLimit() uint64 {
137 if c == nil {
138 return 0
139 }
140 return c.limit.Load()
141}
142
143// Preamble reports whether the carrier is still bounded to wire.PreambleLimit.
144func (c *FramedCarrierBridge) Preamble() bool {
145 if c == nil {
146 return false
147 }
148 return !c.negotiated.Load()
149}
150
151// Send hands one envelope to the raw transport and returns its write result, so
152// it is synchronous: a nil error means the raw write attempt completed. A
153// closed bridge reports ErrCarrierClosed. Cancelling ctx closes the raw
154// transport to release a blocked write and joins the worker before returning,
155// so a send that cannot finish never strands a goroutine.
156func (c *FramedCarrierBridge) Send(ctx context.Context, payload []byte) error {
157 if c == nil || c.raw == nil {
158 return ErrCarrierState
159 }
160 if ctx == nil {
161 ctx = context.Background()
162 }
163 if err := c.enter(); err != nil {
164 return err
165 }
166 defer c.ops.Done()
167 result := make(chan error, 1)
168 go func() { result <- c.raw.Send(wire.Envelope{Payload: payload}) }()
169 select {
170 case err := <-result:
171 return err
172 case <-c.done:
173 <-result
174 return ErrCarrierClosed
175 case <-ctx.Done():
176 c.closeRaw()
177 <-result
178 return ctx.Err()
179 }
180}
181
182// Receive blocks for the next complete envelope, validated against the
183// carrier's current bound (wire.PreambleLimit in the preamble phase, the
184// negotiated envelope ceiling afterwards) before its body is allocated. A
185// closed bridge reports ErrCarrierClosed. Cancelling ctx closes the raw
186// transport to release the blocked read and joins the worker before returning.
187func (c *FramedCarrierBridge) Receive(ctx context.Context) ([]byte, error) {
188 if c == nil || c.raw == nil {
189 return nil, ErrCarrierState
190 }
191 if ctx == nil {
192 ctx = context.Background()
193 }
194 if err := c.enter(); err != nil {
195 return nil, err
196 }
197 defer c.ops.Done()
198 limit := c.limit.Load()
199 result := make(chan carrierRecv, 1)
200 go func() {
201 envelope, err := c.raw.RecvBounded(limit)
202 result <- carrierRecv{payload: envelope.Payload, err: err}
203 }()
204 select {
205 case r := <-result:
206 return r.payload, r.err
207 case <-c.done:
208 <-result
209 return nil, ErrCarrierClosed
210 case <-ctx.Done():
211 c.closeRaw()
212 <-result
213 return nil, ctx.Err()
214 }
215}
216
217// Close interrupts a blocked Send or Receive and is idempotent and
218// concurrent-safe. It marks the bridge closed, closes the raw transport
219// through the adapter prompt-close contract, then waits for every admitted
220// operation, so it never returns with a bridge worker still running. It returns
221// the raw transport's close error; every caller observes the same result.
222func (c *FramedCarrierBridge) Close() error {
223 if c == nil || c.raw == nil {
224 return ErrCarrierState
225 }
226 c.closeOnce.Do(func() {
227 c.mu.Lock()
228 c.closed = true
229 c.mu.Unlock()
230 close(c.done)
231 c.closeRaw()
232 c.ops.Wait()
233 })
234 c.mu.Lock()
235 defer c.mu.Unlock()
236 return c.closeErr
237}
238
239// enter admits one operation unless the bridge already closed. It holds mu so
240// an admission can never race Close's Wait.
241func (c *FramedCarrierBridge) enter() error {
242 c.mu.Lock()
243 defer c.mu.Unlock()
244 if c.closed {
245 return ErrCarrierClosed
246 }
247 c.ops.Add(1)
248 return nil
249}
250
251// closeRaw closes the raw transport exactly once through the adapter
252// prompt-close contract, unblocking a blocked Send or RecvBounded. The first
253// close's error is retained for Close.
254func (c *FramedCarrierBridge) closeRaw() {
255 c.rawCloseOnce.Do(func() {
256 err := c.raw.Close()
257 c.mu.Lock()
258 c.closeErr = err
259 c.mu.Unlock()
260 })
261}
262
263// carrierRecv is one bounded receive result delivered from a bridge worker.
264type carrierRecv struct {
265 payload []byte
266 err error
267}