1// Package quicnettest_test holds the end-to-end integration tests for the
2// adverse-network proxy. It is deliberately an external test package: the
3// concrete production adapters (internal/adapters/quic, sessionwire, protocol,
4// ports) are imported only from these _test.go files, so the simulator package
5// itself stays free of production dependencies and the hexagonal boundary
6// policy still holds.
7package quicnettest_test
8
9import (
10 "bytes"
11 "context"
12 "encoding/binary"
13 "errors"
14 "net"
15 "reflect"
16 "runtime"
17 "testing"
18 "time"
19
20 quicadapter "github.com/bnema/vev/internal/adapters/quic"
21 "github.com/bnema/vev/internal/adapters/quicnettest"
22 "github.com/bnema/vev/internal/adapters/sessionwire"
23 "github.com/bnema/vev/internal/domain"
24 "github.com/bnema/vev/internal/ports"
25 "github.com/bnema/vev/internal/protocol"
26 "github.com/bnema/vev/internal/protocol/wire"
27)
28
29// These tests run the real internal/adapters/quic transport through the proxy
30// and drive a real typed sessionwire conversation over it. The proxy is the
31// only impairment source: no netem, no root, no hooks in production code.
32
33const typedConversationTimeout = 25 * time.Second
34
35type acceptResult struct {
36 transport wire.Transport
37 err error
38}
39
40// trainProfile is the deterministic "4G/train" adverse link: tens of
41// milliseconds of latency with jitter, a few percent loss, duplication, and
42// delay-based reordering. The temporary blackout is toggled while the typed
43// transcript is in flight.
44func trainProfile() quicnettest.LinkConfig {
45 return quicnettest.LinkConfig{
46 BaseLatency: 25 * time.Millisecond,
47 Jitter: 15 * time.Millisecond,
48 LossPercent: 4,
49 DuplicatePercent: 6,
50 DuplicateDelay: 2 * time.Millisecond,
51 ReorderPercent: 10,
52 ReorderDelay: 12 * time.Millisecond,
53 }
54}
55
56// startQUICFixture starts a real QUIC listener, puts a proxy in front of it,
57// and returns the proxy plus the pinned fingerprint the client must dial with.
58func startQUICFixture(t *testing.T, cfg quicnettest.Config) (*quicnettest.Proxy, *quicadapter.Listener, []byte) {
59 t.Helper()
60 cert, fingerprint, err := quicadapter.GenerateEphemeralCert()
61 if err != nil {
62 t.Fatalf("generate cert: %v", err)
63 }
64 listener, err := quicadapter.ListenConfig("127.0.0.1:0", cert, quicadapter.Config{}, 4)
65 if err != nil {
66 t.Fatalf("listen quic: %v", err)
67 }
68 t.Cleanup(func() { _ = listener.Close() })
69
70 serverAddr, err := net.ResolveUDPAddr("udp", listener.Addr())
71 if err != nil {
72 t.Fatalf("resolve listener address %q: %v", listener.Addr(), err)
73 }
74 // The event sink must never block a reader goroutine; the deterministic
75 // counters in Stats are what the assertions use.
76 events := make(chan quicnettest.Event, 1<<16)
77 cfg.ServerAddr = serverAddr
78 cfg.OnPacket = func(event quicnettest.Event) {
79 select {
80 case events <- event:
81 default:
82 }
83 }
84 proxy, err := quicnettest.New(cfg)
85 if err != nil {
86 t.Fatalf("new proxy: %v", err)
87 }
88 t.Cleanup(func() { _ = proxy.Close() })
89 return proxy, listener, fingerprint
90}
91
92func goAccept(listener *quicadapter.Listener) <-chan acceptResult {
93 accepted := make(chan acceptResult, 1)
94 go func() {
95 transport, err := listener.Accept()
96 accepted <- acceptResult{transport: transport, err: err}
97 }()
98 return accepted
99}
100
101// dialRealQUIC accepts on the listener and dials the proxy with the pinned
102// fingerprint, returning both ends of the real QUIC carriage.
103func dialRealQUIC(t *testing.T, proxy *quicnettest.Proxy, listener *quicadapter.Listener, fingerprint []byte) (wire.Transport, wire.Transport) {
104 t.Helper()
105 accepted := goAccept(listener)
106 dialer := quicadapter.DialConfig(proxy.Addr().String(), "vev-bootstrap", fingerprint, quicadapter.Config{}, 15*time.Second)
107 ctx, cancel := context.WithTimeout(context.Background(), typedConversationTimeout)
108 defer cancel()
109 client, err := dialer.Dial(ctx)
110 if err != nil {
111 t.Fatalf("dial through proxy: %v", err)
112 }
113 t.Cleanup(func() { _ = client.Close() })
114
115 select {
116 case result := <-accepted:
117 if result.err != nil {
118 t.Fatalf("accept: %v", result.err)
119 }
120 t.Cleanup(func() { _ = result.transport.Close() })
121 return client, result.transport
122 case <-time.After(typedConversationTimeout):
123 t.Fatal("timed out waiting for the QUIC listener to accept")
124 return nil, nil
125 }
126}
127
128type serverReceive struct {
129 message protocol.ClientMessage
130 err error
131}
132
133// exchangeWithReply sends one typed client message, asserts the server received
134// exactly that message, and returns the typed reply the server sends back.
135func exchangeWithReply(t *testing.T, client ports.ClientConnection, server ports.ServerConnection, message protocol.ClientMessage, reply protocol.ServerMessage) protocol.ServerMessage {
136 t.Helper()
137 received := make(chan serverReceive, 1)
138 go func() {
139 got, err := server.ReceiveClient()
140 received <- serverReceive{message: got, err: err}
141 }()
142 if err := client.SendClient(message); err != nil {
143 t.Fatalf("send %T: %v", message, err)
144 }
145 result := awaitServerReceive(t, received)
146 if result.err != nil {
147 t.Fatalf("server receive: %v", result.err)
148 }
149 if !reflect.DeepEqual(result.message, message) {
150 t.Fatalf("server received %#v, want %#v", result.message, message)
151 }
152
153 sent := make(chan error, 1)
154 go func() { sent <- server.SendServer(reply) }()
155 got, err := client.ReceiveServer()
156 if err != nil {
157 t.Fatalf("client receive reply: %v", err)
158 }
159 if err := <-sent; err != nil {
160 t.Fatalf("server send %T: %v", reply, err)
161 }
162 return got
163}
164
165func awaitServerReceive(t *testing.T, received <-chan serverReceive) serverReceive {
166 t.Helper()
167 select {
168 case result := <-received:
169 return result
170 case <-time.After(typedConversationTimeout):
171 t.Fatal("timed out waiting for the typed client message")
172 return serverReceive{}
173 }
174}
175
176func serverReceiveFrom(server ports.ServerConnection) <-chan serverReceive {
177 received := make(chan serverReceive, 1)
178 go func() {
179 message, err := server.ReceiveClient()
180 received <- serverReceive{message: message, err: err}
181 }()
182 return received
183}
184
185// startTypedConversation performs the real QUIC handshake through the proxy and
186// completes the typed session preamble with a Hello/Welcome exchange.
187func startTypedConversation(t *testing.T, proxy *quicnettest.Proxy, listener *quicadapter.Listener, fingerprint []byte) (ports.ClientConnection, ports.ServerConnection) {
188 t.Helper()
189 clientRaw, serverRaw := dialRealQUIC(t, proxy, listener, fingerprint)
190 client := sessionwire.NewClientConnection(clientRaw)
191 server := sessionwire.NewServerConnection(serverRaw)
192 t.Cleanup(func() { _ = client.Close() })
193 t.Cleanup(func() { _ = server.Close() })
194
195 hello := protocol.Hello{
196 Version: protocol.Version,
197 ClientID: [16]byte{0x4d, 0x79, 0x4e, 0x41, 0x54, 0x2d, 0x34, 0x47},
198 Name: "train-4g",
199 Size: domain.Size{Cols: 80, Rows: 24},
200 TermEnv: "xterm-256color",
201 }
202 reply := exchangeWithReply(t, client, server, hello, protocol.Welcome{SessionID: "train-4g", SessionName: "train", Capabilities: 1})
203 welcome, ok := reply.(protocol.Welcome)
204 if !ok {
205 t.Fatalf("welcome reply is %T, want protocol.Welcome", reply)
206 }
207 if welcome.SessionID != "train-4g" {
208 t.Fatalf("welcome session id = %q, want %q", welcome.SessionID, "train-4g")
209 }
210 return client, server
211}
212
213func inputPayload(seq uint64) []byte {
214 payload := make([]byte, 512)
215 binary.BigEndian.PutUint64(payload, seq)
216 for i := 8; i < len(payload); i++ {
217 payload[i] = byte(seq)
218 }
219 return payload
220}
221
222// TestRealQUICTypedConversationOverTrainProfile runs the real vev QUIC
223// transport through the proxy under a deterministic 4G/train profile and drives
224// a real typed sessionwire conversation over it: Hello/Welcome, then 200
225// sequenced Input records (each 512 bytes) in one pipelined transcript, then
226// 200 Pong replies. It requires that
227//
228// - every semantic payload arrives exactly once, in order, byte-for-byte,
229// with no gaps and no duplicates,
230// - a 500ms blackout that starts mid-transcript drops packets in both
231// directions without losing or duplicating an application message,
232// - loss, duplication, and reordering actually occurred,
233// - the bounded queue never overflowed and every accepted packet is
234// accounted for (received + duplicates == sent + drops),
235// - cleanup and goroutine exit are bounded.
236func TestRealQUICTypedConversationOverTrainProfile(t *testing.T) {
237 const total = 200
238
239 before := runtime.NumGoroutine()
240 profile := trainProfile()
241 proxy, listener, fingerprint := startQUICFixture(t, quicnettest.Config{
242 Seed: 20260913,
243 QueueCapacity: 1024,
244 ToServer: profile,
245 ToClient: profile,
246 })
247 client, server := startTypedConversation(t, proxy, listener, fingerprint)
248
249 // Forward direction: the client pipelines the whole sequenced transcript
250 // while the server drains it. The blackout starts once the transcript is
251 // under way and clears on a bounded real-time timer, because the real QUIC
252 // stack, not a test clock, owns retransmission here.
253 sendErr := make(chan error, 1)
254 go func() {
255 for seq := uint64(1); seq <= total; seq++ {
256 err := client.SendClient(protocol.Input{InputSeq: seq, ActionID: seq, Data: inputPayload(seq)})
257 if err != nil {
258 sendErr <- err
259 return
260 }
261 }
262 sendErr <- nil
263 }()
264
265 blackoutStarted := false
266 for index := 0; index < total; index++ {
267 if index == total/5 && !blackoutStarted {
268 blackoutStarted = true
269 proxy.SetBlackout(true)
270 time.AfterFunc(500*time.Millisecond, func() { proxy.SetBlackout(false) })
271 }
272 result := awaitServerReceive(t, serverReceiveFrom(server))
273 if result.err != nil {
274 t.Fatalf("server receive %d: %v", index, result.err)
275 }
276 input, ok := result.message.(protocol.Input)
277 if !ok {
278 t.Fatalf("server message %d is %T, want protocol.Input", index, result.message)
279 }
280 wantSeq := uint64(index + 1)
281 if input.InputSeq != wantSeq || input.ActionID != wantSeq {
282 t.Fatalf("server message %d has seq/action %d/%d, want %d/%d", index, input.InputSeq, input.ActionID, wantSeq, wantSeq)
283 }
284 if !bytes.Equal(input.Data, inputPayload(wantSeq)) {
285 t.Fatalf("server message %d payload differs from the sent bytes", index)
286 }
287 }
288 if err := <-sendErr; err != nil {
289 t.Fatalf("client transcript send: %v", err)
290 }
291
292 // Reverse direction: the server answers every Input with exactly one Pong.
293 replyErr := make(chan error, 1)
294 go func() {
295 for index := 0; index < total; index++ {
296 if err := server.SendServer(protocol.Pong{}); err != nil {
297 replyErr <- err
298 return
299 }
300 }
301 replyErr <- nil
302 }()
303 for index := 0; index < total; index++ {
304 message, err := client.ReceiveServer()
305 if err != nil {
306 t.Fatalf("client receive reply %d: %v", index, err)
307 }
308 if _, ok := message.(protocol.Pong); !ok {
309 t.Fatalf("client reply %d is %T, want protocol.Pong", index, message)
310 }
311 }
312 if err := <-replyErr; err != nil {
313 t.Fatalf("server reply transcript: %v", err)
314 }
315
316 stats := waitForAccounting(t, proxy, 10*time.Second)
317 if stats.ToServer.LossDrops == 0 || stats.ToServer.Duplicates == 0 || stats.ToServer.Reordered == 0 {
318 t.Fatalf("train profile was not exercised: %s", stats)
319 }
320 if stats.ToServer.BlackoutDrops == 0 || stats.ToClient.BlackoutDrops == 0 {
321 t.Fatalf("blackout did not drop packets in both directions: %s", stats)
322 }
323 if stats.ToServer.OverflowDrops != 0 || stats.ToClient.OverflowDrops != 0 {
324 t.Fatalf("bounded queue overflowed: %s", stats)
325 }
326 if stats.ToServer.DeliveryErrors != 0 || stats.ToClient.DeliveryErrors != 0 {
327 t.Fatalf("proxy delivery failed: %s", stats)
328 }
329 if stats.ToServer.Received == 0 || stats.ToServer.Sent == 0 || stats.ToClient.Sent == 0 {
330 t.Fatalf("proxy carried no traffic: %s", stats)
331 }
332
333 assertBoundedCleanup(t, proxy, listener, client, server, before)
334}
335
336// TestRealQUICNATRebindKeepsTypedConversation proves the address-change path:
337// after the proxy rebinds its upstream socket, the real quic-go server must
338// follow the client's new address (RFC 9000 section 9 passive migration) so the
339// typed conversation keeps flowing. If quic-go ever stops honouring passive
340// migration, this test fails instead of silently dropping the scenario.
341func TestRealQUICNATRebindKeepsTypedConversation(t *testing.T) {
342 before := runtime.NumGoroutine()
343 proxy, listener, fingerprint := startQUICFixture(t, quicnettest.Config{
344 Seed: 20260914,
345 QueueCapacity: 1024,
346 ToServer: quicnettest.LinkConfig{BaseLatency: 5 * time.Millisecond},
347 ToClient: quicnettest.LinkConfig{BaseLatency: 5 * time.Millisecond},
348 })
349 client, server := startTypedConversation(t, proxy, listener, fingerprint)
350
351 // Baseline typed round trip before the address change.
352 if reply := exchangeWithReply(t, client, server, protocol.Ping{}, protocol.Pong{}); reply == nil {
353 t.Fatal("baseline exchange returned no reply")
354 }
355 beforeAddr := proxy.UpstreamAddr()
356
357 if err := proxy.Rebind(); err != nil {
358 t.Fatalf("rebind: %v", err)
359 }
360 afterAddr := proxy.UpstreamAddr()
361 if beforeAddr == nil || afterAddr == nil || beforeAddr.Port == afterAddr.Port {
362 t.Fatalf("upstream address did not change: %v -> %v", beforeAddr, afterAddr)
363 }
364
365 // Every further typed round trip must survive the server-observed address
366 // change; the old upstream socket is closed, so a server that ignored the
367 // rebind would stall these exchanges.
368 for sequence := uint64(1); sequence <= 5; sequence++ {
369 message := protocol.Input{InputSeq: sequence, ActionID: sequence, Data: inputPayload(sequence)}
370 reply := exchangeWithReply(t, client, server, message, protocol.Pong{})
371 if _, ok := reply.(protocol.Pong); !ok {
372 t.Fatalf("round trip %d after rebind returned %T, want protocol.Pong", sequence, reply)
373 }
374 }
375
376 stats := waitForAccounting(t, proxy, 10*time.Second)
377 if stats.Rebinds != 1 {
378 t.Fatalf("rebinds = %d, want 1", stats.Rebinds)
379 }
380 if stats.StrayPackets != 0 {
381 t.Fatalf("stray packets after rebind = %d, want 0", stats.StrayPackets)
382 }
383 if stats.ToServer.OverflowDrops != 0 || stats.ToClient.OverflowDrops != 0 {
384 t.Fatalf("bounded queue overflowed: %s", stats)
385 }
386 if stats.ToServer.DeliveryErrors != 0 || stats.ToClient.DeliveryErrors != 0 {
387 t.Fatalf("proxy delivery failed after rebind: %s", stats)
388 }
389
390 assertBoundedCleanup(t, proxy, listener, client, server, before)
391}
392
393// waitForAccounting polls the proxy counters until every accepted packet is
394// either sent or dropped in both directions, proving no packet is stuck in the
395// bounded queue, then returns the settled snapshot.
396func waitForAccounting(t *testing.T, proxy *quicnettest.Proxy, timeout time.Duration) quicnettest.Stats {
397 t.Helper()
398 deadline := time.Now().Add(timeout)
399 stats := proxy.Stats()
400 for {
401 stats = proxy.Stats()
402 if accounted(stats.ToServer) && accounted(stats.ToClient) {
403 return stats
404 }
405 if time.Now().After(deadline) {
406 t.Fatalf("proxy accounting did not settle within %s: %s", timeout, stats)
407 }
408 time.Sleep(10 * time.Millisecond)
409 }
410}
411
412// accounted reports that a direction has no packet in flight inside the proxy.
413func accounted(link quicnettest.LinkStats) bool {
414 // Oversize datagrams are rejected before Received is incremented, so
415 // they are intentionally outside the accepted-packet identity.
416 drops := link.LossDrops + link.BlackoutDrops + link.OverflowDrops
417 return link.Received+link.Duplicates == link.Sent+drops
418}
419
420// assertBoundedCleanup closes every resource with a deadline and requires the
421// goroutine count to return near its pre-test baseline.
422func assertBoundedCleanup(t *testing.T, proxy *quicnettest.Proxy, listener *quicadapter.Listener, client, server interface{ Close() error }, baseline int) {
423 t.Helper()
424 closed := make(chan error, 1)
425 go func() {
426 closed <- errors.Join(client.Close(), server.Close(), listener.Close(), proxy.Close())
427 }()
428 select {
429 case err := <-closed:
430 if err != nil {
431 t.Fatalf("cleanup: %v", err)
432 }
433 case <-time.After(15 * time.Second):
434 t.Fatal("cleanup did not complete within 15s")
435 }
436
437 deadline := time.Now().Add(5 * time.Second)
438 const slack = 8
439 for {
440 current := runtime.NumGoroutine()
441 if current <= baseline+slack {
442 return
443 }
444 if time.Now().After(deadline) {
445 t.Fatalf("goroutines did not drain: baseline %d, still %d", baseline, current)
446 }
447 time.Sleep(25 * time.Millisecond)
448 }
449}