1package dgram
2
3import (
4 "context"
5 "errors"
6 "io"
7 "log/slog"
8 "time"
9
10 "github.com/bnema/vev/internal/ports"
11)
12
13const (
14 DefaultProxyIdleTTL = 30 * time.Second
15 DefaultProxyRetryBackoff = 100 * time.Millisecond
16)
17
18type ProxyRuntime struct {
19 Client ports.Transport
20 Daemon ports.Transport
21 Log *slog.Logger
22 IdleTTL time.Duration
23 RetryBackoff time.Duration
24 Clock ports.Clock
25}
26
27type proxyCopyErrKind uint8
28
29const (
30 proxyClientRecv proxyCopyErrKind = iota
31 proxyDaemonRecv
32 proxyClientSend
33 proxyDaemonSend
34)
35
36type proxyCopyErr struct {
37 kind proxyCopyErrKind
38 err error
39}
40
41type proxyCopier struct {
42 ctx context.Context
43 errCh chan<- proxyCopyErr
44 retryBackoff time.Duration
45 clk ports.Clock
46}
47
48type proxyCopyDirection struct {
49 src ports.Transport
50 dst ports.Transport
51 recvKind proxyCopyErrKind
52 sendKind proxyCopyErrKind
53 transform func(ports.Frame) ports.Frame
54 retryRecoverable bool
55}
56
57func (p ProxyRuntime) Run(ctx context.Context) error {
58 defer func() {
59 _ = p.Client.Close()
60 _ = p.Daemon.Close()
61 }()
62 copyCtx, cancelCopies := context.WithCancel(ctx)
63 defer cancelCopies()
64
65 idleTTL := p.IdleTTL
66 if idleTTL <= 0 {
67 idleTTL = DefaultProxyIdleTTL
68 }
69 retryBackoff := p.RetryBackoff
70 if retryBackoff <= 0 {
71 retryBackoff = DefaultProxyRetryBackoff
72 }
73 clk := p.Clock
74 if clk == nil {
75 clk = realClock{}
76 }
77
78 errCh := make(chan proxyCopyErr, 2)
79 copier := proxyCopier{ctx: copyCtx, errCh: errCh, retryBackoff: retryBackoff, clk: clk}
80 clientToDaemon := proxyCopyDirection{
81 src: p.Client, dst: p.Daemon,
82 recvKind: proxyClientRecv, sendKind: proxyDaemonSend,
83 transform: clampDatagramHelloOutputWindow,
84 }
85 daemonToClient := proxyCopyDirection{src: p.Daemon, dst: p.Client, recvKind: proxyDaemonRecv, sendKind: proxyClientSend, retryRecoverable: true}
86 go copier.copyFrames(clientToDaemon)
87 go copier.copyFrames(daemonToClient)
88
89 var linkEvents <-chan ports.LinkEvent
90 if reporter, ok := p.Client.(ports.LinkStateReporter); ok {
91 linkEvents = reporter.LinkEvents()
92 }
93 var idle <-chan time.Time
94 var timer ports.Timer
95 defer func() {
96 if timer != nil {
97 timer.Stop()
98 }
99 }()
100 startIdle := func() {
101 if idle != nil {
102 return
103 }
104 if timer == nil {
105 timer = clk.NewTimer(idleTTL)
106 } else {
107 timer.Reset(idleTTL)
108 }
109 idle = timer.C()
110 }
111 stopIdle := func() {
112 if timer != nil {
113 timer.Stop()
114 }
115 idle = nil
116 }
117 for {
118 select {
119 case <-ctx.Done():
120 return ctx.Err()
121 case ev, ok := <-linkEvents:
122 if !ok {
123 linkEvents = nil
124 continue
125 }
126 switch ev.State {
127 case ports.LinkStateDegraded, ports.LinkStateProbing, ports.LinkStateOffline, ports.LinkStateDead:
128 startIdle()
129 case ports.LinkStateConnected:
130 stopIdle()
131 }
132 case <-idle:
133 return ErrLinkDead
134 case ce := <-errCh:
135 switch ce.kind {
136 case proxyDaemonRecv:
137 if errors.Is(ce.err, io.EOF) {
138 return nil
139 }
140 return ce.err
141 case proxyClientRecv, proxyClientSend:
142 if errors.Is(ce.err, ErrLinkDead) {
143 startIdle()
144 continue
145 }
146 return ce.err
147 default:
148 // Recoverable client-side errors are retried inside the copier
149 // without dropping the frame; anything that surfaces here (a
150 // daemon send failure, or a non-recoverable client send failure)
151 // is fatal.
152 return ce.err
153 }
154 }
155 }
156}
157
158func (c proxyCopier) copyFrames(dir proxyCopyDirection) {
159 for {
160 f, err := dir.src.Recv()
161 if err != nil {
162 c.errCh <- proxyCopyErr{kind: dir.recvKind, err: err}
163 return
164 }
165 if dir.transform != nil {
166 f = dir.transform(f)
167 }
168 // Retry the same frame in place while the destination link error stays
169 // recoverable (congestion, transient path loss). The frame is never
170 // dropped — on a reliable stream a gap desynchronizes the client screen.
171 for {
172 err := dir.dst.Send(f)
173 if err == nil {
174 break
175 }
176 if dir.retryRecoverable && recoverableClientLink(dir.dst, err) {
177 t := c.clk.NewTimer(c.retryBackoff)
178 select {
179 case <-t.C():
180 t.Stop()
181 continue
182 case <-c.ctx.Done():
183 t.Stop()
184 return
185 }
186 }
187 c.errCh <- proxyCopyErr{kind: dir.sendKind, err: err}
188 return
189 }
190 }
191}
192
193func recoverableClientLink(t ports.Transport, err error) bool {
194 if errors.Is(err, ErrLinkDead) {
195 return false
196 }
197 reporter, ok := t.(ports.LinkStateReporter)
198 if !ok {
199 return false
200 }
201 switch reporter.LinkState() {
202 case ports.LinkStateConnected:
203 return errors.Is(err, ErrPendingFull)
204 case ports.LinkStateDegraded, ports.LinkStateProbing, ports.LinkStateOffline:
205 return true
206 default:
207 return false
208 }
209}
210
211// clampDatagramHelloOutputWindow enforces the datagram-safe state window at
212// the adapter boundary. Invalid Hello payloads pass through unchanged so the
213// daemon remains the single authority for strict decoding and version checks.
214func clampDatagramHelloOutputWindow(f ports.Frame) ports.Frame {
215 if f.Type != ports.MsgHello {
216 return f
217 }
218 hello, err := ports.UnmarshalHello(f.Payload)
219 if err != nil {
220 return f
221 }
222 hello.MaxOutputInFlight = 1
223 f.Payload = ports.MarshalHello(hello)
224 return f
225}