1package ipc
2
3import (
4 "errors"
5 "net"
6 "sync"
7
8 "github.com/bnema/vev/internal/adapters/streamframe"
9 "github.com/bnema/vev/internal/ports"
10 "github.com/bnema/vev/internal/protocol/wire"
11)
12
13// ErrZeroLengthFrame is returned by Recv when an envelope's length field is
14// zero. Kept as an alias so existing callers keep compiling; new code
15// prefers streamframe.ErrZeroLength.
16var ErrZeroLengthFrame = streamframe.ErrZeroLength
17
18// ErrFrameTooLarge is returned by Recv when an envelope's length field
19// exceeds the ceiling, and by Send when a payload is too large to encode.
20var ErrFrameTooLarge = streamframe.ErrTooLarge
21
22// ErrBackpressure means the bounded IPC egress queue is full. Callers using
23// SendAsync must treat it as a failed render emission rather than spawning an
24// unbounded queue of their own.
25var ErrBackpressure = streamframe.ErrBackpressure
26
27var errClosed = streamframe.ErrClosed
28
29// unixTransport implements wire.Transport over a net.Conn (in practice an
30// AF_UNIX SOCK_STREAM connection, but any net.Conn works — this also lets
31// tests exercise it over net.Pipe).
32//
33// Framing, queueing, and close behavior live in streamframe; this adapter
34// retains lifecycle ownership (conn close releases blocked Recv), peer
35// checks, and runtime observation.
36type Option func(*unixTransport)
37
38// WithRuntimeObserver enables process-local transport marks. It accepts no
39// clock: timestamp ownership belongs to the configured observer.
40func WithRuntimeObserver(observer ports.SerializedRuntimeObserver) Option {
41 return func(t *unixTransport) { t.observer = observer }
42}
43
44type unixTransport struct {
45 conn net.Conn
46 observer ports.SerializedRuntimeObserver
47 framer *streamframe.Framer
48
49 operationMu sync.Mutex
50 operationCount int
51 observedClosing bool
52 operationsDone chan struct{}
53
54 closeOnce sync.Once
55 closeErr error
56}
57
58// NewTransport wraps conn as a wire.Transport speaking vev's length-framed
59// envelope protocol.
60func NewTransport(conn net.Conn, opts ...Option) wire.Transport {
61 t := &unixTransport{conn: conn}
62 for _, opt := range opts {
63 if opt != nil {
64 opt(t)
65 }
66 }
67 t.framer = streamframe.NewFramer(conn, conn, conn.Close)
68 return t
69}
70
71var _ wire.BoundedTransport = (*unixTransport)(nil)
72
73// Send queues the envelope for the sole writer and waits for its wire attempt.
74func (t *unixTransport) Send(envelope wire.Envelope) error {
75 end := t.beginOperation(ports.RuntimeAdapterSendStart, uint64(len(envelope.Payload)))
76 err := t.framer.Send(envelope.Payload)
77 end(err == nil)
78 return err
79}
80
81// SendAsync accepts the envelope for ordered background transmission without
82// waiting for a socket write. The queue is bounded and ownership includes a
83// payload copy.
84func (t *unixTransport) SendAsync(envelope wire.Envelope) error {
85 end := t.beginOperation(ports.RuntimeAdapterSendStart, uint64(len(envelope.Payload)))
86 err := t.framer.SendAsync(envelope.Payload)
87 end(err == nil)
88 return err
89}
90
91// Recv reads one complete envelope. It blocks until a full envelope
92// arrives, the connection is closed (io.EOF), or an error occurs.
93func (t *unixTransport) Recv() (wire.Envelope, error) {
94 end := t.beginOperation(ports.RuntimeAdapterReceiveStart, 0)
95 payload, err := t.framer.Recv()
96 if err != nil {
97 end(false)
98 return wire.Envelope{}, mapFramerError(err)
99 }
100 end(true)
101 return wire.Envelope{Payload: payload}, nil
102}
103
104// RecvBounded reads one complete envelope bounded to limit before allocation,
105// sharing the framer's validation with Recv: an over-limit length prefix is
106// refused without reading its body. It is the bounded raw-carriage capability
107// a preamble-negotiated multiplexer consumes.
108func (t *unixTransport) RecvBounded(limit uint64) (wire.Envelope, error) {
109 end := t.beginOperation(ports.RuntimeAdapterReceiveStart, 0)
110 payload, err := t.framer.RecvBounded(limit)
111 if err != nil {
112 end(false)
113 return wire.Envelope{}, mapFramerError(err)
114 }
115 end(true)
116 return wire.Envelope{Payload: payload}, nil
117}
118
119func mapFramerError(err error) error {
120 switch {
121 case errors.Is(err, streamframe.ErrZeroLength):
122 return ErrZeroLengthFrame
123 case errors.Is(err, streamframe.ErrTooLarge):
124 return ErrFrameTooLarge
125 case errors.Is(err, streamframe.ErrClosed):
126 return errClosed
127 default:
128 return err
129 }
130}
131
132func (t *unixTransport) beginOperation(start ports.RuntimeMarkKind, bytes uint64) func(bool) {
133 if t.observer == nil {
134 return func(bool) {}
135 }
136 if !t.beginObservedOperation() {
137 return func(bool) {}
138 }
139 correlation := ports.NewRuntimeCorrelation()
140 t.observer.ObserveRuntime(ports.NewRuntimeMarkWithCorrelation("ipc", correlation, start, bytes, true))
141 end := ports.RuntimeAdapterSendEnd
142 if start == ports.RuntimeAdapterReceiveStart {
143 end = ports.RuntimeAdapterReceiveEnd
144 }
145 return func(valid bool) {
146 defer t.finishObservedOperation()
147 t.observer.ObserveRuntime(ports.NewRuntimeMarkWithCorrelation("ipc", correlation, end, bytes, valid))
148 }
149}
150
151func (t *unixTransport) beginObservedOperation() bool {
152 t.operationMu.Lock()
153 defer t.operationMu.Unlock()
154 if t.observedClosing {
155 return false
156 }
157 if t.operationCount == 0 {
158 t.operationsDone = make(chan struct{})
159 }
160 t.operationCount++
161 return true
162}
163
164func (t *unixTransport) finishObservedOperation() {
165 t.operationMu.Lock()
166 defer t.operationMu.Unlock()
167 t.operationCount--
168 if t.operationCount == 0 {
169 close(t.operationsDone)
170 }
171}
172
173func (t *unixTransport) beginShutdown() <-chan struct{} {
174 t.operationMu.Lock()
175 defer t.operationMu.Unlock()
176 t.observedClosing = true
177 if t.operationCount == 0 {
178 return nil
179 }
180 return t.operationsDone
181}
182
183// Close interrupts Recv (via conn.Close, which releases the framer's
184// blocked read) and the writer, fails queued envelopes, and waits until
185// the sole writer and all observed operations have stopped.
186func (t *unixTransport) Close() error {
187 t.closeOnce.Do(func() {
188 operationsDone := t.beginShutdown()
189 t.closeErr = t.framer.Close()
190 if operationsDone != nil {
191 <-operationsDone
192 }
193 })
194 return t.closeErr
195}