diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index 90a5354..de38051 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -146,6 +146,19 @@ owns a UDP socket and may return traffic only from destinations previously requested by that session. Closing the SOCKS control connection closes the AutoCAR control stream and unblocks both receive loops. +Completed native requests use bounded per-requested-target FIFO lanes: a slow +domain resolution no longer serializes another target's work on the same +association. Every packet still invokes the policy resolver with its original +address; this is not a DNS cache or a destination-policy bypass. The configured +`UDPReceiveQueue` bounds total active plus waiting requests. A lazy pool has at +most eight workers/target lanes, with a 256 KiB payload-plus-address charge cap; +one target gets at most half the packet/byte budget (packet half rounded up). +Metadata is separately bounded by packet and target counts. Idle lanes retire. +Saturation drops new packets, and several slow targets can still exhaust the +shared budget; this is not a delivery guarantee or cross-target ordering. +Closing cancels resolution, discards queued work, closes the socket and joins +all workers before releasing the association's admission. + UDP has no TLS/TCP fallback. Explicit `tls` mode therefore does not advertise SOCKS5 UDP ASSOCIATE. `auto` does advertise it, but an association always makes a QUIC attempt and fails if QUIC is unavailable; the TCP fallback circuit applies @@ -159,6 +172,17 @@ logical PacketConn, but the H3 implementation creates one authenticated request stream lazily for each normalized target and reuses that stream for later datagrams to the same target. +The SOCKS frontend also isolates lazy H3 target opening, but only for transports +that explicitly implement `transport.PacketConcurrentSender`. Canonical target +aliases share one FIFO lane; the WebClient and close-once wrappers preserve +this capability. Legacy/custom PacketConn implementations remain serial unless +they opt in to both concurrent Send safety and Close interrupting every Send. +The frontend uses a shared 32-packet active-plus-waiting budget, the same byte +and half-per-target caps, and at most eight workers (further capped by the +transport's reported limit). It joins sends after closing the endpoints. A +locally queued packet still does not prove remote delivery or H3 path health. +Fresh connection authentication remains serialized; no bootstrap is bypassed. + The HMAC ticket binds the H3 origin, Extended CONNECT protocol, and default MASQUE target path. After authentication, the relay parses the target, applies the same UDP resolver policy, freezes one successfully dialed numeric endpoint, diff --git a/docs/WEB_COVER.md b/docs/WEB_COVER.md index b030b34..a1af77a 100644 --- a/docs/WEB_COVER.md +++ b/docs/WEB_COVER.md @@ -470,6 +470,19 @@ both pacing fields as `not-applicable`, and `--pacing=fixed-rate` is rejected. `h2` does not advertise UDP. The H3-to-H2 fallback carries only new TCP streams: UDP never falls back to H2 and fails when H3 is unavailable. +For an established SOCKS association, opening a slow new H3 UDP target does +not serialize sends to another warm target while scheduler capacity remains. +Aliases are grouped using the same canonical target identity as CONNECT-UDP. +This keeps connection authentication and all server destination/admission +checks intact; it neither retries packets nor caches native DNS resolutions. +The frontend queue counts active and waiting packets together (32 total), with +at most eight workers and a 256 KiB payload-plus-address cap. One target can use +at most half the packet/byte budget. Multiple busy targets may still fill the +shared queue, in which case packets are dropped. Close interrupts and joins +pending sends before the SOCKS association is released. Custom packet +transports retain serial sending unless they explicitly opt in to this +concurrency/close contract; see [Architecture](ARCHITECTURE.md). + Only a newly authenticated CONNECT-UDP response is fresh evidence that the H3 path has recovered. Sending on a cached UDP target merely queues a datagram locally; it does not clear the TCP fallback cooldown or change the last diff --git a/internal/proxy/socks5.go b/internal/proxy/socks5.go index 4d80798..3f2578a 100644 --- a/internal/proxy/socks5.go +++ b/internal/proxy/socks5.go @@ -424,6 +424,20 @@ func (c *closeOncePacketConn) MaxPayloadSize() int { return 0 } +func (c *closeOncePacketConn) ConcurrentSendLimit() int { + if sender, ok := c.PacketConn.(transport.PacketConcurrentSender); ok { + return sender.ConcurrentSendLimit() + } + return 0 +} + +func (c *closeOncePacketConn) SendTargetKey(address string) (string, error) { + if sender, ok := c.PacketConn.(transport.PacketConcurrentSender); ok { + return sender.SendTargetKey(address) + } + return address, nil +} + // accept records the first valid source port when the UDP ASSOCIATE request // specified port zero. Every datagram must originate from the control TCP // connection's peer IP, preventing the relay from becoming an open UDP proxy. @@ -482,6 +496,16 @@ func runSOCKSUDPAssociation( } } finish := func() { finished <- struct{}{} } + var sendQueue *transport.PacketSendQueue + var sendErrors <-chan error + var concurrent transport.PacketConcurrentSender + if sender, ok := upstream.(transport.PacketConcurrentSender); ok && sender.ConcurrentSendLimit() > 1 { + concurrent = sender + sendQueue = transport.NewPacketSendQueue(context.Background(), 32, sender.ConcurrentSendLimit(), func(_ context.Context, payload []byte, target string) error { + return upstream.Send(payload, target) + }) + sendErrors = sendQueue.Errors() + } go func() { defer finish() @@ -497,7 +521,16 @@ func runSOCKSUDPAssociation( if err != nil || len(payload) > maxPayloadSize || !endpoint.accept(source) { continue } - if err := upstream.Send(payload, target); err != nil { + var sendErr error + if sendQueue != nil { + target, sendErr = concurrent.SendTargetKey(target) + if sendErr == nil { + sendErr = sendQueue.Enqueue(payload, target) + } + } else { + sendErr = upstream.Send(payload, target) + } + if err := sendErr; err != nil { if errors.Is(err, transport.ErrPacketQueueFull) || errors.Is(err, transport.ErrPacketTargetUnavailable) { // Datagram delivery is best effort. Queue pressure or a rejected // target drops this packet, not other targets on the association. @@ -550,6 +583,8 @@ wait: break wait case <-controlDone: break wait + case <-sendErrors: + break wait case <-activity: if timer != nil { if !timer.Stop() { @@ -568,7 +603,13 @@ wait: // Join packet workers before the caller joins its control reader and // releases the tracked connection. Closing endpoints interrupts their I/O. _ = local.Close() + if sendQueue != nil { + sendQueue.Stop() + } _ = upstream.Close() + if sendQueue != nil { + sendQueue.Wait() + } for completed < 2 { <-finished completed++ diff --git a/internal/proxy/socks5_udp_concurrency_test.go b/internal/proxy/socks5_udp_concurrency_test.go new file mode 100644 index 0000000..df2ce15 --- /dev/null +++ b/internal/proxy/socks5_udp_concurrency_test.go @@ -0,0 +1,483 @@ +package proxy + +import ( + "bytes" + "context" + "errors" + "fmt" + "net" + "net/netip" + "strconv" + "strings" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/cppla/autocar/internal/transport" +) + +// These frontend contract tests use real, owned loopback UDP ingress/egress, +// but a deliberately controlled PacketConn. They do not model relay DNS, +// authenticate an H3 peer, or prove end-to-end delivery through a tunnel. +// The local structural interface also lets this source compile before the +// optional production interface exists; missing capability is a behavioral +// assertion, not a missing-type compilation oracle. +type socksUDPConcurrencyCapability interface { + transport.PacketConn + ConcurrentSendLimit() int + SendTargetKey(string) (string, error) +} + +type socksUDPConcurrencyCall struct { + payload []byte + address string + err error + done chan struct{} +} + +type socksUDPConcurrencyPacket struct { + started chan *socksUDPConcurrencyCall + incoming chan packetRecord + closed chan struct{} + receiveDone chan struct{} + closeOnce sync.Once + receiveOnce sync.Once + closes atomic.Int32 + active atomic.Int32 + maximum atomic.Int32 + send func([]byte, string) error +} + +func newSOCKSUDPConcurrencyPacket() *socksUDPConcurrencyPacket { + return &socksUDPConcurrencyPacket{ + started: make(chan *socksUDPConcurrencyCall, 16), + incoming: make(chan packetRecord, 16), + closed: make(chan struct{}), + receiveDone: make(chan struct{}), + } +} + +func (p *socksUDPConcurrencyPacket) Send(payload []byte, address string) (err error) { + n := p.active.Add(1) + for old := p.maximum.Load(); n > old && !p.maximum.CompareAndSwap(old, n); old = p.maximum.Load() { + } + call := &socksUDPConcurrencyCall{ + payload: append([]byte(nil), payload...), address: address, done: make(chan struct{}), + } + defer func() { + p.active.Add(-1) + call.err = err + close(call.done) + }() + select { + case p.started <- call: + case <-p.closed: + return net.ErrClosed + } + if p.send != nil { + return p.send(payload, address) + } + return nil +} + +func (p *socksUDPConcurrencyPacket) Receive() ([]byte, string, error) { + select { + case record := <-p.incoming: + return record.payload, record.address, nil + case <-p.closed: + p.receiveOnce.Do(func() { close(p.receiveDone) }) + return nil, "", net.ErrClosed + } +} + +func (p *socksUDPConcurrencyPacket) Close() error { + p.closes.Add(1) + p.closeOnce.Do(func() { close(p.closed) }) + return nil +} + +type socksUDPConcurrencyOptional struct { + *socksUDPConcurrencyPacket + limit int + keys chan string + key func(string) (string, error) +} + +func (p *socksUDPConcurrencyOptional) ConcurrentSendLimit() int { return p.limit } + +func (p *socksUDPConcurrencyOptional) SendTargetKey(address string) (string, error) { + p.keys <- address + if p.key != nil { + return p.key(address) + } + return socksUDPConcurrencyCanonicalKey(address) +} + +// This is a fixture-provided key, not the production CONNECT-UDP normalizer. +// Its purpose is to prove that the frontend obeys its provider's identity. +func socksUDPConcurrencyCanonicalKey(address string) (string, error) { + host, port, err := net.SplitHostPort(address) + if err != nil { + return "", err + } + n, err := strconv.ParseUint(port, 10, 16) + if err != nil || n == 0 { + return "", errors.New("fixture invalid target port") + } + if ip, err := netip.ParseAddr(host); err == nil { + if ip.Zone() != "" { + return "", errors.New("fixture target zone") + } + host = ip.Unmap().String() + } else { + host = strings.ToLower(host) + } + return net.JoinHostPort(host, strconv.FormatUint(n, 10)), nil +} + +type socksUDPConcurrencyFixture struct { + t *testing.T + packet *socksUDPConcurrencyPacket + upstream *closeOncePacketConn + local *net.UDPConn + client *net.UDPConn + relay *net.UDPAddr + control chan struct{} + stopOnce sync.Once + done chan struct{} + started bool + releases []func() +} + +func newSOCKSUDPConcurrencyFixture(t *testing.T, packet *socksUDPConcurrencyPacket, upstream transport.PacketConn, releases ...func()) *socksUDPConcurrencyFixture { + t.Helper() + f := &socksUDPConcurrencyFixture{ + t: t, packet: packet, upstream: &closeOncePacketConn{PacketConn: upstream}, + control: make(chan struct{}), done: make(chan struct{}), releases: releases, + } + t.Cleanup(func() { + // Release every callback/return gate before shutting down owned I/O, + // including the path taken by Fatal at any earlier checkpoint. + for _, release := range f.releases { + release() + } + f.stop() + if f.local != nil { + _ = f.local.Close() + } + if f.client != nil { + _ = f.client.Close() + } + _ = f.upstream.Close() + if f.started { + socksUDPConcurrencyWait(t, f.done, "association cleanup join") + socksUDPConcurrencyWait(t, packet.receiveDone, "upstream Receive cleanup join") + } + }) + var err error + f.local, err = net.ListenUDP("udp4", &net.UDPAddr{IP: net.IPv4(127, 0, 0, 1)}) + if err != nil { + t.Fatal(err) + } + f.client, err = net.ListenUDP("udp4", &net.UDPAddr{IP: net.IPv4(127, 0, 0, 1)}) + if err != nil { + t.Fatal(err) + } + f.relay = f.local.LocalAddr().(*net.UDPAddr) + endpoint := &socksUDPClientEndpoint{ + peerIP: net.IPv4(127, 0, 0, 1), + requestedPort: f.client.LocalAddr().(*net.UDPAddr).Port, + } + f.started = true + go func() { + defer close(f.done) + runSOCKSUDPAssociation(f.control, f.local, f.upstream, endpoint, 0) + }() + return f +} + +func (f *socksUDPConcurrencyFixture) stop() { f.stopOnce.Do(func() { close(f.control) }) } + +func (f *socksUDPConcurrencyFixture) write(target, payload string) { + f.t.Helper() + packet, err := buildSOCKSUDPDatagram([]byte(payload), target) + if err != nil { + f.t.Fatal(err) + } + if err := f.client.SetWriteDeadline(time.Now().Add(time.Second)); err != nil { + f.t.Fatal(err) + } + if _, err := f.client.WriteToUDP(packet, f.relay); err != nil { + f.t.Fatal(err) + } +} + +func (f *socksUDPConcurrencyFixture) call(payload, address string) *socksUDPConcurrencyCall { + f.t.Helper() + select { + case call := <-f.packet.started: + if string(call.payload) != payload || call.address != address { + f.t.Fatalf("actual Send = %q to %q, want %q to %q", call.payload, call.address, payload, address) + } + return call + case <-time.After(2 * time.Second): + f.t.Fatal("expected actual Send did not enter") + return nil + } +} + +func (f *socksUDPConcurrencyFixture) echo(payload, address string) { + f.t.Helper() + f.packet.incoming <- packetRecord{payload: []byte(payload), address: address} + if err := f.client.SetReadDeadline(time.Now().Add(2 * time.Second)); err != nil { + f.t.Fatal(err) + } + buffer := make([]byte, 256) + n, source, err := f.client.ReadFromUDP(buffer) + if err != nil { + f.t.Fatal(err) + } + got, target, err := parseSOCKSUDPDatagram(buffer[:n]) + if err != nil || !bytes.Equal(got, []byte(payload)) || target != address || !source.IP.Equal(f.relay.IP) || source.Port != f.relay.Port { + f.t.Fatalf("actual UDP response=%q target=%q source=%v err=%v", got, target, source, err) + } +} + +func (f *socksUDPConcurrencyFixture) assertJoined() { + f.t.Helper() + socksUDPConcurrencyWait(f.t, f.done, "association join") + socksUDPConcurrencyWait(f.t, f.packet.receiveDone, "Receive join") + if got := f.packet.active.Load(); got != 0 { + f.t.Errorf("association returned with %d active Send calls", got) + } + if got := f.packet.closes.Load(); got != 1 { + f.t.Errorf("upstream actual Close calls=%d, want one", got) + } +} + +func socksUDPConcurrencyWait(t *testing.T, done <-chan struct{}, name string) bool { + t.Helper() + select { + case <-done: + return true + case <-time.After(2 * time.Second): + t.Errorf("%s did not finish", name) + return false + } +} + +func (f *socksUDPConcurrencyFixture) await(done <-chan struct{}, name string) { + f.t.Helper() + if !socksUDPConcurrencyWait(f.t, done, name) { + f.t.Fatal("required causal checkpoint failed") + } +} + +func socksUDPConcurrencyGate() (<-chan struct{}, func()) { + gate := make(chan struct{}) + return gate, sync.OnceFunc(func() { close(gate) }) +} + +func socksUDPConcurrencyKey(t *testing.T, keys <-chan string, want string) { + t.Helper() + select { + case got := <-keys: + if got != want { + t.Fatalf("provider key input=%q, want %q", got, want) + } + case <-time.After(2 * time.Second): + t.Fatal("frontend did not request provider's canonical target key") + } +} + +func TestSOCKS5UDPConcurrencyLegacyRemainsSerial(t *testing.T) { + for _, mode := range []string{"legacy", "limit_zero", "limit_one"} { + t.Run(mode, func(t *testing.T) { + packet := newSOCKSUDPConcurrencyPacket() + gate, release := socksUDPConcurrencyGate() + packet.send = func(payload []byte, _ string) error { + if string(payload) == "first" { + select { + case <-gate: + case <-packet.closed: + return net.ErrClosed + } + } + return nil + } + optional := &socksUDPConcurrencyOptional{packet, 0, make(chan string, 16), nil} + var upstream transport.PacketConn = packet + if mode != "legacy" { + upstream = optional + } + if mode == "limit_one" { + optional.limit = 1 + } + f := newSOCKSUDPConcurrencyFixture(t, packet, upstream, release) + f.write("first.invalid:5300", "first") + first := f.call("first", "first.invalid:5300") + f.write("second.invalid:5301", "second") + // This bounded absence window is a serial-mode control, not an entry + // witness or performance measurement. Actual first Send is held. + select { + case call := <-packet.started: + t.Fatalf("serial mode overlapped actual Send: %q", call.payload) + case <-time.After(75 * time.Millisecond): + } + release() + f.await(first.done, "first Send") + second := f.call("second", "second.invalid:5301") + f.await(second.done, "second Send") + f.echo("second response", "second.invalid:5301") + f.stop() + f.assertJoined() + if packet.maximum.Load() != 1 || len(optional.keys) != 0 { + t.Errorf("serial control maximum=%d key calls=%d", packet.maximum.Load(), len(optional.keys)) + } + }) + } +} + +func TestSOCKS5UDPConcurrencyOptionalTargetIsolationAndCanonicalFIFO(t *testing.T) { + packet := newSOCKSUDPConcurrencyPacket() + gate, release := socksUDPConcurrencyGate() + packet.send = func(payload []byte, _ string) error { + if string(payload) == "slow first" { + select { + case <-gate: + if string(payload) != "slow first" { + return errors.New("frontend reused ingress payload buffer") + } + case <-packet.closed: + return net.ErrClosed + } + } + return nil + } + optional := &socksUDPConcurrencyOptional{packet, 2, make(chan string, 16), nil} + f := newSOCKSUDPConcurrencyFixture(t, packet, optional, release) + capability, ok := any(f.upstream).(socksUDPConcurrencyCapability) + if !ok || capability.ConcurrentSendLimit() != 2 { + t.Fatal("close-once wrapper did not preserve optional concurrent-send capacity") + } + for _, test := range []struct{ address, want string }{ + {"UPPER.invalid:053", "upper.invalid:53"}, + {"[2001:db8:0:0:0:0:0:1]:53", "[2001:db8::1]:53"}, + {"[::ffff:127.0.0.1]:53", "127.0.0.1:53"}, + } { + got, err := capability.SendTargetKey(test.address) + if err != nil || got != test.want { + t.Fatalf("wrapper key(%q)=%q err=%v, want %q", test.address, got, err, test.want) + } + socksUDPConcurrencyKey(t, optional.keys, test.address) + } + f.write("SLOW.invalid:5300", "slow first") + socksUDPConcurrencyKey(t, optional.keys, "SLOW.invalid:5300") + first := f.call("slow first", "slow.invalid:5300") + f.write("slow.invalid:5300", "slow second") + socksUDPConcurrencyKey(t, optional.keys, "slow.invalid:5300") + f.write("fast.invalid:5301", "fast") + socksUDPConcurrencyKey(t, optional.keys, "fast.invalid:5301") + fast := f.call("fast", "fast.invalid:5301") + f.await(fast.done, "fast target Send") + f.echo("fast response", "fast.invalid:5301") + select { + case <-first.done: + t.Fatal("slow actual callback ended before its release gate") + default: + } + if packet.maximum.Load() != 2 { + t.Errorf("actual concurrent Send maximum=%d, want two", packet.maximum.Load()) + } + release() + f.await(first.done, "slow first Send") + if first.err != nil { + t.Errorf("slow first Send after release=%v", first.err) + } + second := f.call("slow second", "slow.invalid:5300") + f.await(second.done, "slow second Send") + f.echo("slow second response", "slow.invalid:5300") + f.stop() + f.assertJoined() +} + +func TestSOCKS5UDPConcurrencyErrorsPreserveAssociationPolicy(t *testing.T) { + for _, test := range []struct { + name string + err error + recoverable bool + }{ + {"queue_full", fmt.Errorf("fixture pressure: %w", transport.ErrPacketQueueFull), true}, + {"target_unavailable", fmt.Errorf("fixture target: %w", transport.ErrPacketTargetUnavailable), true}, + {"unknown", errors.New("fixture terminal failure"), false}, + {"canceled", fmt.Errorf("fixture terminal cancellation: %w", context.Canceled), false}, + {"deadline", fmt.Errorf("fixture terminal deadline: %w", context.DeadlineExceeded), false}, + {"closed", fmt.Errorf("fixture terminal close: %w", net.ErrClosed), false}, + } { + t.Run(test.name, func(t *testing.T) { + packet := newSOCKSUDPConcurrencyPacket() + packet.send = func(payload []byte, _ string) error { + if string(payload) == "error" { + return test.err + } + return nil + } + optional := &socksUDPConcurrencyOptional{packet, 2, make(chan string, 16), nil} + f := newSOCKSUDPConcurrencyFixture(t, packet, optional) + f.write("policy.invalid:5300", "error") + failed := f.call("error", "policy.invalid:5300") + f.await(failed.done, "failed Send") + if failed.err != test.err { + t.Errorf("controlled Send error identity changed: %v", failed.err) + } + if test.recoverable { + f.write("policy.invalid:5300", "healthy") + healthy := f.call("healthy", "policy.invalid:5300") + f.await(healthy.done, "healthy Send") + f.echo("healthy response", "policy.invalid:5300") + select { + case <-packet.closed: + t.Fatal("recoverable Send error closed the association") + default: + } + f.stop() + } + f.assertJoined() + if len(packet.started) != 0 { + t.Error("failed packet was transparently retried") + } + }) + } +} + +func TestSOCKS5UDPConcurrencyCloseJoinsPendingSends(t *testing.T) { + packet := newSOCKSUDPConcurrencyPacket() + returnGate, release := socksUDPConcurrencyGate() + observedClose := make(chan struct{}) + packet.send = func([]byte, string) error { + <-packet.closed + close(observedClose) + // This explicit post-cancellation return gate tests owner join, not + // network backpressure or a natural scheduling race. + <-returnGate + return net.ErrClosed + } + optional := &socksUDPConcurrencyOptional{packet, 2, make(chan string, 16), nil} + f := newSOCKSUDPConcurrencyFixture(t, packet, optional, release) + f.write("close.invalid:5300", "blocked") + call := f.call("blocked", "close.invalid:5300") + f.stop() + f.await(observedClose, "actual Send observing upstream Close") + select { + case <-f.done: + t.Fatal("association returned before its accepted Send completed") + case <-time.After(75 * time.Millisecond): + } + if packet.active.Load() != 1 { + t.Errorf("held actual Send count=%d, want one", packet.active.Load()) + } + release() + f.await(call.done, "canceled Send completion") + f.assertJoined() +} diff --git a/internal/transport/packet_send_queue.go b/internal/transport/packet_send_queue.go new file mode 100644 index 0000000..88843a8 --- /dev/null +++ b/internal/transport/packet_send_queue.go @@ -0,0 +1,212 @@ +package transport + +import ( + "context" + "errors" + "net" + "sync" +) + +const ( + packetSendQueueMaxTargets = 8 + packetSendQueueMaxBytes = 256 << 10 + packetSendTargetMaxBytes = packetSendQueueMaxBytes / 2 +) + +type packetSendItem struct { + payload []byte + address string + charge int +} + +type packetSendLane struct { + queue []packetSendItem + packets int + bytes int + active bool +} + +// PacketSendQueue isolates slow targets with bounded, best-effort FIFO lanes. +// All limits include active callbacks, not just packets waiting to run. One +// target can use at most half the packet and byte budgets (rounded up for +// packets); multiple busy targets may still exhaust the shared budget. +// There is no delivery guarantee, retry, or ordering across different targets. +type PacketSendQueue struct { + ctx context.Context + cancel context.CancelFunc + send func(context.Context, []byte, string) error + + mu sync.Mutex + stopped bool + lanes map[string]*packetSendLane + packets int + bytes int + maxPackets int + maxTargetPackets int + maxTargets int + workers int + ready chan *packetSendLane + wg sync.WaitGroup + fatalOnce sync.Once + errors chan error +} + +// NewPacketSendQueue does not start workers until Enqueue accepts a packet. +// Nonpositive maxPackets selects one packet; nonpositive maxTargets selects +// the eight-target ceiling. The target count also cannot exceed maxPackets. +// parent and send must be non-nil. A callback must return when ctx is canceled +// or its owner closes the underlying I/O endpoint. The queue cannot forcibly +// interrupt custom code. The byte limit covers payload plus address charges, +// not total resident memory; the separate packet/target limits bound metadata. +func NewPacketSendQueue(parent context.Context, maxPackets, maxTargets int, send func(context.Context, []byte, string) error) *PacketSendQueue { + if maxPackets <= 0 { + maxPackets = 1 + } + if maxTargets <= 0 { + maxTargets = packetSendQueueMaxTargets + } + ctx, cancel := context.WithCancel(parent) + return &PacketSendQueue{ + ctx: ctx, cancel: cancel, send: send, + lanes: make(map[string]*packetSendLane), + maxPackets: maxPackets, maxTargetPackets: maxPackets/2 + maxPackets%2, + maxTargets: min(maxTargets, maxPackets, packetSendQueueMaxTargets), + ready: make(chan *packetSendLane, min(maxTargets, maxPackets, packetSendQueueMaxTargets)), + errors: make(chan error, 1), + } +} + +// Enqueue consumes payload before returning and never waits for a callback. +// address is both the destination and the FIFO key. Callers with aliases must +// canonicalize it first, without resolving or bypassing destination policy. +// Payload plus address bytes are charged before copying; even empty inputs +// cost one byte, so arbitrarily large count settings cannot admit free work. +func (q *PacketSendQueue) Enqueue(payload []byte, address string) error { + q.mu.Lock() + defer q.mu.Unlock() + if q.stopped || q.ctx.Err() != nil { + return net.ErrClosed + } + if len(payload) > packetSendQueueMaxBytes || len(address) > packetSendQueueMaxBytes-len(payload) { + return ErrPacketQueueFull + } + charge := max(1, len(payload)+len(address)) + lane := q.lanes[address] + if q.packets >= q.maxPackets || charge > packetSendQueueMaxBytes-q.bytes || charge > packetSendTargetMaxBytes { + return ErrPacketQueueFull + } + if lane == nil { + if len(q.lanes) >= q.maxTargets { + return ErrPacketQueueFull + } + lane = &packetSendLane{} + } else if lane.packets >= q.maxTargetPackets || charge > packetSendTargetMaxBytes-lane.bytes { + return ErrPacketQueueFull + } + item := packetSendItem{payload: append([]byte(nil), payload...), address: address, charge: charge} + lane.queue = append(lane.queue, item) + lane.packets++ + lane.bytes += charge + q.packets++ + q.bytes += charge + if q.lanes[address] == nil { + q.lanes[address] = lane + q.ready <- lane // One ready entry per non-active lane; capacity is maxTargets. + } + if q.workers < len(q.lanes) { + // Add and admission share mu with Stop. Wait after Stop can never race + // with a new positive Add. Workers are a lazy bounded pool: idle lanes + // retire, but no per-packet goroutine or unbounded exiting tail is created. + q.workers++ + q.wg.Add(1) + go q.run() + } + return nil +} + +func (q *PacketSendQueue) run() { + defer q.wg.Done() + for { + var lane *packetSendLane + select { + case <-q.ctx.Done(): + q.Stop() + return + case lane = <-q.ready: + } + q.mu.Lock() + if q.stopped || q.ctx.Err() != nil { + q.mu.Unlock() + q.Stop() + return + } + lane.active = true + item := lane.queue[0] + lane.queue[0] = packetSendItem{} + lane.queue = lane.queue[1:] + if len(lane.queue) == 0 { + lane.queue = nil + } + q.mu.Unlock() + + err := q.send(q.ctx, item.payload, item.address) + if err != nil && q.ctx.Err() == nil && !errors.Is(err, ErrPacketQueueFull) && !errors.Is(err, ErrPacketTargetUnavailable) { + q.fatalOnce.Do(func() { + q.Stop() + q.errors <- err // Exactly once into a one-entry buffer; no blocking reader needed. + }) + } + q.mu.Lock() + lane.packets-- + lane.bytes -= item.charge + q.packets-- + q.bytes -= item.charge + lane.active = false + if q.stopped || q.ctx.Err() != nil || len(lane.queue) == 0 { + q.discardQueued(lane) + delete(q.lanes, item.address) + } else { + // Yield after each datagram instead of draining a hot target. No + // other worker can select this lane until its callback has returned. + q.ready <- lane + } + q.mu.Unlock() + } +} + +// discardQueued is called only under mu. An active item remains charged until +// its callback actually returns; Stop must not make live work disappear from +// the resource budget or free a lane that can still use the owner's endpoint. +func (q *PacketSendQueue) discardQueued(lane *packetSendLane) { + for _, item := range lane.queue { + lane.packets-- + lane.bytes -= item.charge + q.packets-- + q.bytes -= item.charge + } + lane.queue = nil +} + +// Stop forbids admission, cancels callbacks and discards waiting packets. It +// does not wait: an owner may need to close its socket to unblock active I/O. +func (q *PacketSendQueue) Stop() { + q.mu.Lock() + q.stopped = true + q.cancel() + for address, lane := range q.lanes { + q.discardQueued(lane) + if !lane.active { + delete(q.lanes, address) + } + } + q.mu.Unlock() +} + +// Wait joins every target worker. Call Stop and interrupt underlying I/O first. +// In particular, do not call Wait from inside the queue's own callback. +func (q *PacketSendQueue) Wait() { q.wg.Wait() } + +// Errors reports the first terminal callback error, preserving its identity. +// Queue-full and target-unavailable errors drop one packet, never retry it, +// and do not appear here. The channel is intentionally never closed. +func (q *PacketSendQueue) Errors() <-chan error { return q.errors } diff --git a/internal/transport/packet_send_queue_test.go b/internal/transport/packet_send_queue_test.go new file mode 100644 index 0000000..512b751 --- /dev/null +++ b/internal/transport/packet_send_queue_test.go @@ -0,0 +1,648 @@ +package transport + +import ( + "context" + "errors" + "fmt" + "net" + "strings" + "sync" + "sync/atomic" + "testing" + "time" +) + +const packetQueueTestTargetBytes = 128 << 10 + +type packetQueueTestEvent struct { + address string + payload string + ctx context.Context +} + +type packetQueueTestGate struct { + done chan struct{} + once sync.Once +} + +func (g *packetQueueTestGate) release() { g.once.Do(func() { close(g.done) }) } + +type packetQueueTestFixture struct { + t *testing.T + ctx context.Context + cancel context.CancelFunc + gates []*packetQueueTestGate + queues []*PacketSendQueue + auxJoined []<-chan struct{} + unexpected atomic.Int32 +} + +func newPacketQueueTestFixture(t *testing.T) *packetQueueTestFixture { + t.Helper() + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + t.Cleanup(cancel) + f := &packetQueueTestFixture{t: t, ctx: ctx, cancel: cancel} + t.Cleanup(func() { + f.cancel() + for _, gate := range f.gates { + gate.release() + } + for _, q := range f.queues { + joined := make(chan struct{}) + go func(q *PacketSendQueue) { + q.Stop() + q.Wait() + close(joined) + }(q) + select { + case <-joined: + case <-time.After(2 * time.Second): + t.Error("packet queue cleanup did not join its callbacks") + } + } + for _, joined := range f.auxJoined { + select { + case <-joined: + case <-time.After(2 * time.Second): + t.Error("packet queue fixture caller did not join") + } + } + if got := f.unexpected.Load(); got != 0 { + t.Errorf("unexpected callback event/error count=%d", got) + } + }) + return f +} + +func (f *packetQueueTestFixture) gate() *packetQueueTestGate { + g := &packetQueueTestGate{done: make(chan struct{})} + f.gates = append(f.gates, g) + return g +} + +func (f *packetQueueTestFixture) queue(maxPackets, maxTargets int, send func(context.Context, []byte, string) error) *PacketSendQueue { + q := NewPacketSendQueue(f.ctx, maxPackets, maxTargets, send) + f.queues = append(f.queues, q) + return q +} + +func (f *packetQueueTestFixture) publish(ch chan<- packetQueueTestEvent, event packetQueueTestEvent) { + select { + case ch <- event: + default: + f.unexpected.Add(1) + } +} + +func (f *packetQueueTestFixture) next(ch <-chan packetQueueTestEvent, label string) packetQueueTestEvent { + f.t.Helper() + select { + case event := <-ch: + return event + case <-time.After(2 * time.Second): + f.t.Fatalf("missing %s callback witness", label) + case <-f.ctx.Done(): + f.t.Fatalf("fixture ended waiting for %s: %v", label, context.Cause(f.ctx)) + } + return packetQueueTestEvent{} +} + +func (f *packetQueueTestFixture) enqueue(q *PacketSendQueue, payload []byte, address string) { + f.t.Helper() + if err := q.Enqueue(payload, address); err != nil { + f.t.Fatalf("Enqueue(payloadBytes=%d, addressBytes=%d): %v", len(payload), len(address), err) + } +} + +func (f *packetQueueTestFixture) wantFull(q *PacketSendQueue, payload []byte, address string) { + f.t.Helper() + if err := q.Enqueue(payload, address); !errors.Is(err, ErrPacketQueueFull) { + f.t.Errorf("Enqueue rejected budget with %v, want ErrPacketQueueFull", err) + } +} + +func (f *packetQueueTestFixture) stopJoin(q *PacketSendQueue) { + f.t.Helper() + joined := make(chan struct{}) + f.auxJoined = append(f.auxJoined, joined) + go func() { q.Stop(); q.Wait(); close(joined) }() + select { + case <-joined: + case <-time.After(2 * time.Second): + f.t.Fatal("Stop/Wait did not join packet callbacks") + } +} + +func (f *packetQueueTestFixture) stop(q *PacketSendQueue) { + f.t.Helper() + joined := make(chan struct{}) + f.auxJoined = append(f.auxJoined, joined) + go func() { q.Stop(); close(joined) }() + select { + case <-joined: + case <-time.After(2 * time.Second): + f.t.Fatal("Stop waited for a controlled active callback instead of returning") + } +} + +// A callback's completion notification precedes its return. Observe idle-lane +// pruning through public admission, not a private map or a scheduling sleep. +// Only rejected Enqueues are retried; an accepted packet is never retried. +func (f *packetQueueTestFixture) enqueueAfterRetirement(q *PacketSendQueue, payload []byte, address string) { + f.t.Helper() + deadline := time.NewTimer(2 * time.Second) + defer deadline.Stop() + tick := time.NewTicker(time.Millisecond) + defer tick.Stop() + for { + err := q.Enqueue(payload, address) + if err == nil { + return + } + if !errors.Is(err, ErrPacketQueueFull) { + f.t.Fatalf("admission after completed callback: %v", err) + } + select { + case <-tick.C: + case <-deadline.C: + f.t.Fatal("completed target lane retained capacity") + case <-f.ctx.Done(): + f.t.Fatalf("fixture ended before lane retirement: %v", context.Cause(f.ctx)) + } + } +} + +func packetQueueTestNoError(t *testing.T, q *PacketSendQueue) { + t.Helper() + select { + case err, ok := <-q.Errors(): + t.Errorf("Errors received value=%v open=%t, want open empty channel", err, ok) + default: + } +} + +func TestPacketSendQueueIsolatesTargetsAndPreservesFIFO(t *testing.T) { + f := newPacketQueueTestFixture(t) + first, other := f.gate(), f.gate() + started := make(chan packetQueueTestEvent, 8) + q := f.queue(8, 3, func(ctx context.Context, payload []byte, address string) error { + f.publish(started, packetQueueTestEvent{address: address, payload: string(payload), ctx: ctx}) + var gate *packetQueueTestGate + if address == "a:1" && string(payload) == "a1" { + gate = first + } else if address == "b:2" { + gate = other + } + if gate != nil { + select { + case <-gate.done: + case <-ctx.Done(): + return context.Cause(ctx) + } + } + return nil + }) + f.enqueue(q, []byte("a1"), "a:1") + if event := f.next(started, "blocked a1"); event.address != "a:1" || event.payload != "a1" { + t.Fatalf("first callback=%+v", event) + } + f.enqueue(q, []byte("a2"), "a:1") + f.enqueue(q, []byte("a3"), "a:1") + f.enqueue(q, []byte("b1"), "b:2") + if event := f.next(started, "independent b1"); event.address != "b:2" || event.payload != "b1" { + t.Fatalf("callback while a1 blocked=%+v, want b1", event) + } + select { + case event := <-started: + t.Errorf("same-target callback overlapped blocked a1: %+v", event) + case <-time.After(50 * time.Millisecond): + } + first.release() + for _, want := range []string{"a2", "a3"} { + if event := f.next(started, want); event.address != "a:1" || event.payload != want { + t.Errorf("same-target FIFO callback=%+v, want %s", event, want) + } + } + other.release() + f.stopJoin(q) + packetQueueTestNoError(t, q) + select { + case event := <-started: + t.Errorf("unexpected retry/extra callback=%+v", event) + default: + } +} + +func TestPacketSendQueueBudgetsIncludeActivePackets(t *testing.T) { + t.Run("nonpositive-packet-normalization", func(t *testing.T) { + for _, maxPackets := range []int{0, -1} { + t.Run(fmt.Sprintf("packets-%d", maxPackets), func(t *testing.T) { + f := newPacketQueueTestFixture(t) + seen := make(chan packetQueueTestEvent, 2) + q := f.queue(maxPackets, 0, func(ctx context.Context, _ []byte, address string) error { + f.publish(seen, packetQueueTestEvent{address: address}) + <-ctx.Done() + return context.Cause(ctx) + }) + f.enqueue(q, nil, "a:1") + f.next(seen, "normalized single active packet") + f.wantFull(q, nil, "a:1") + f.wantFull(q, nil, "b:2") + f.stopJoin(q) + packetQueueTestNoError(t, q) + }) + } + }) + t.Run("target-ceiling-and-default", func(t *testing.T) { + for _, maxTargets := range []int{0, -1, 9} { + t.Run(fmt.Sprintf("targets-%d", maxTargets), func(t *testing.T) { + f := newPacketQueueTestFixture(t) + seen := make(chan packetQueueTestEvent, 16) + q := f.queue(32, maxTargets, func(ctx context.Context, _ []byte, address string) error { + f.publish(seen, packetQueueTestEvent{address: address}) + <-ctx.Done() + return context.Cause(ctx) + }) + for index := 0; index < 8; index++ { + f.enqueue(q, nil, fmt.Sprintf("active-%d:1", index)) + f.next(seen, "eight-target callback ceiling") + } + f.wantFull(q, nil, "ninth:1") + f.stopJoin(q) + packetQueueTestNoError(t, q) + }) + } + }) + t.Run("packets", func(t *testing.T) { + f := newPacketQueueTestFixture(t) + gate := f.gate() + started := make(chan packetQueueTestEvent, 8) + q := f.queue(4, 3, func(ctx context.Context, payload []byte, address string) error { + f.publish(started, packetQueueTestEvent{address: address, payload: string(payload)}) + select { + case <-gate.done: + return nil + case <-ctx.Done(): + return context.Cause(ctx) + } + }) + f.enqueue(q, []byte("a1"), "a:1") + f.next(started, "active a1") + f.enqueue(q, []byte("a2"), "a:1") + f.wantFull(q, []byte("a3"), "a:1") // ceil(4/2), though global space remains. + f.enqueue(q, []byte("b1"), "b:2") + f.next(started, "active b1") + f.enqueue(q, []byte("b2"), "b:2") + f.wantFull(q, []byte("c1"), "c:3") // Four total, including both active callbacks. + f.stopJoin(q) + packetQueueTestNoError(t, q) + }) + t.Run("target-bytes", func(t *testing.T) { + f := newPacketQueueTestFixture(t) + gate := f.gate() + started := make(chan packetQueueTestEvent, 8) + q := f.queue(8, 3, func(ctx context.Context, _ []byte, address string) error { + f.publish(started, packetQueueTestEvent{address: address}) + select { + case <-gate.done: + return nil + case <-ctx.Done(): + return context.Cause(ctx) + } + }) + f.enqueue(q, make([]byte, packetQueueTestTargetBytes-len("a:1")), "a:1") + f.next(started, "active full-byte a") + f.wantFull(q, nil, "a:1") + f.enqueue(q, []byte("other target still fits"), "b:2") + f.next(started, "independent byte budget") + f.stopJoin(q) + packetQueueTestNoError(t, q) + }) + t.Run("global-bytes", func(t *testing.T) { + f := newPacketQueueTestFixture(t) + gate := f.gate() + started := make(chan packetQueueTestEvent, 8) + q := f.queue(8, 3, func(ctx context.Context, _ []byte, address string) error { + f.publish(started, packetQueueTestEvent{address: address}) + select { + case <-gate.done: + return nil + case <-ctx.Done(): + return context.Cause(ctx) + } + }) + for _, address := range []string{"a:1", "b:2"} { + f.enqueue(q, make([]byte, packetQueueTestTargetBytes-len(address)), address) + f.next(started, "active global-byte charge") + } + f.wantFull(q, nil, "c:3") // 256 KiB is charged while callbacks remain active. + f.stopJoin(q) + packetQueueTestNoError(t, q) + }) +} + +func TestPacketSendQueueOwnsPayloadAndRetiresTargets(t *testing.T) { + f := newPacketQueueTestFixture(t) + gate := f.gate() + started, observed := make(chan packetQueueTestEvent, 32), make(chan packetQueueTestEvent, 32) + q := f.queue(4, 1, func(ctx context.Context, payload []byte, address string) error { + f.publish(started, packetQueueTestEvent{address: address}) + if address == "a:1" { + select { + case <-gate.done: + case <-ctx.Done(): + return context.Cause(ctx) + } + } + f.publish(observed, packetQueueTestEvent{address: address, payload: string(payload)}) + return nil + }) + payload := []byte("owned-original") + f.enqueue(q, payload, "a:1") + f.next(started, "copy callback admission") + copy(payload, "caller-mutated") + f.wantFull(q, []byte("b"), "b:2") // A live target occupies the only target slot. + gate.release() + if event := f.next(observed, "owned payload"); event.payload != "owned-original" { + t.Errorf("callback payload=%q, want the Enqueue-owned copy", event.payload) + } + for index := 0; index < 12; index++ { + address := fmt.Sprintf("retired-%d:2", index) + f.enqueueAfterRetirement(q, []byte("next"), address) + if event := f.next(started, "replacement target"); event.address != address { + t.Errorf("replacement callback address=%q, want %q", event.address, address) + } + if event := f.next(observed, "replacement completion"); event.address != address || event.payload != "next" { + t.Errorf("replacement callback=%+v", event) + } + } + f.stopJoin(q) + packetQueueTestNoError(t, q) +} + +func TestPacketSendQueueErrorsDoNotRetry(t *testing.T) { + t.Run("recoverable", func(t *testing.T) { + f := newPacketQueueTestFixture(t) + seen := make(chan packetQueueTestEvent, 8) + q := f.queue(8, 1, func(_ context.Context, payload []byte, address string) error { + f.publish(seen, packetQueueTestEvent{address: address, payload: string(payload)}) + switch string(payload) { + case "queue-full": + return fmt.Errorf("owned send: %w", ErrPacketQueueFull) + case "target-unavailable": + return fmt.Errorf("owned send: %w", ErrPacketTargetUnavailable) + default: + return nil + } + }) + for _, payload := range []string{"queue-full", "target-unavailable", "healthy"} { + f.enqueue(q, []byte(payload), "a:1") + } + for _, want := range []string{"queue-full", "target-unavailable", "healthy"} { + if event := f.next(seen, want); event.payload != want { + t.Errorf("callback=%+v, want %s once in FIFO order", event, want) + } + } + f.stopJoin(q) + packetQueueTestNoError(t, q) + select { + case event := <-seen: + t.Errorf("recoverable send was retried: %+v", event) + default: + } + }) + t.Run("terminal", func(t *testing.T) { + f := newPacketQueueTestFixture(t) + gate := f.gate() + seen := make(chan packetQueueTestEvent, 8) + want := errors.New("owned terminal send failure") + q := f.queue(8, 1, func(_ context.Context, payload []byte, address string) error { + f.publish(seen, packetQueueTestEvent{address: address, payload: string(payload)}) + <-gate.done + return want + }) + f.enqueue(q, []byte("active"), "a:1") + f.next(seen, "terminal callback") + f.enqueue(q, []byte("must-drop"), "a:1") + gate.release() + select { + case err, ok := <-q.Errors(): + if !ok || err != want { + t.Errorf("terminal Errors value=%v open=%t, want exact original error", err, ok) + } + case <-time.After(2 * time.Second): + t.Fatal("terminal callback did not publish its error") + } + if err := q.Enqueue(nil, "b:2"); !errors.Is(err, net.ErrClosed) { + t.Errorf("Enqueue after terminal error=%v, want automatic Stop/net.ErrClosed", err) + } + f.stopJoin(q) + packetQueueTestNoError(t, q) + select { + case event := <-seen: + t.Errorf("terminal/queued packet callback repeated: %+v", event) + default: + } + }) +} + +func TestPacketSendQueueStopJoinsAndRejectsEnqueue(t *testing.T) { + t.Run("active-join", func(t *testing.T) { + f := newPacketQueueTestFixture(t) + gate := f.gate() + seen := make(chan packetQueueTestEvent, 8) + q := f.queue(4, 2, func(ctx context.Context, payload []byte, address string) error { + f.publish(seen, packetQueueTestEvent{address: address, payload: string(payload), ctx: ctx}) + <-gate.done // Deliberately return only when independently released. + return errors.New("controlled callback returned after Stop") + }) + var active []packetQueueTestEvent + for _, address := range []string{"a:1", "b:2"} { + f.enqueue(q, []byte("active"), address) + active = append(active, f.next(seen, "active callback")) + f.enqueue(q, []byte("must-drop"), address) + } + f.stop(q) + for _, event := range active { + if event.ctx.Err() == nil { + t.Error("Stop did not cancel an active callback's context") + } + } + if err := q.Enqueue(nil, "new:3"); !errors.Is(err, net.ErrClosed) { + t.Errorf("Enqueue after Stop=%v, want net.ErrClosed", err) + } + waiting, joined := make(chan struct{}, 2), make(chan struct{}, 2) + for index := 0; index < 2; index++ { + callerJoined := make(chan struct{}) + f.auxJoined = append(f.auxJoined, callerJoined) + go func() { + defer close(callerJoined) + waiting <- struct{}{} + q.Wait() + joined <- struct{}{} + }() + } + for index := 0; index < 2; index++ { + select { + case <-waiting: + case <-time.After(2 * time.Second): + t.Fatal("concurrent Wait worker did not start") + } + } + alreadyJoined := 0 + select { + case <-joined: + alreadyJoined++ + t.Error("Wait returned while active callbacks remained gated") + case <-time.After(50 * time.Millisecond): + } + gate.release() + for index := alreadyJoined; index < 2; index++ { + select { + case <-joined: + case <-time.After(2 * time.Second): + t.Fatal("concurrent Wait did not join released callbacks") + } + } + packetQueueTestNoError(t, q) // Shutdown callback errors are suppressed. + select { + case event := <-seen: + t.Errorf("Stop ran a queued packet: %+v", event) + default: + } + }) + t.Run("parent-cancellation", func(t *testing.T) { + f := newPacketQueueTestFixture(t) + parent, cancel := context.WithCancelCause(f.ctx) + defer cancel(nil) + seen := make(chan packetQueueTestEvent, 2) + q := NewPacketSendQueue(parent, 4, 2, func(ctx context.Context, _ []byte, address string) error { + f.publish(seen, packetQueueTestEvent{address: address, ctx: ctx}) + <-ctx.Done() + return context.Cause(ctx) + }) + f.queues = append(f.queues, q) + f.enqueue(q, []byte("active"), "a:1") + event := f.next(seen, "parent-owned callback") + want := errors.New("private parent cancellation") + cancel(want) + <-parent.Done() + f.stopJoin(q) + if got := context.Cause(event.ctx); got != want { + t.Errorf("callback context cause=%v, want exact parent cause", got) + } + if err := q.Enqueue(nil, "b:2"); !errors.Is(err, net.ErrClosed) { + t.Errorf("canceled queue Enqueue=%v, want net.ErrClosed", err) + } + packetQueueTestNoError(t, q) + }) + t.Run("stop-versus-enqueue", func(t *testing.T) { + f := newPacketQueueTestFixture(t) + seen := make(chan packetQueueTestEvent, 8) + q := f.queue(16, 4, func(ctx context.Context, _ []byte, address string) error { + f.publish(seen, packetQueueTestEvent{address: address}) + <-ctx.Done() + return context.Cause(ctx) + }) + f.enqueue(q, []byte("already-active"), "lane-0:1") + f.next(seen, "active callback before concurrent Stop/Enqueue") + start := make(chan struct{}) + var workers sync.WaitGroup + workers.Add(5) + for index := 0; index < 4; index++ { + go func(index int) { + defer workers.Done() + <-start + for packet := 0; packet < 64; packet++ { + err := q.Enqueue([]byte{byte(packet)}, fmt.Sprintf("lane-%d:1", index)) + if err != nil && !errors.Is(err, ErrPacketQueueFull) && !errors.Is(err, net.ErrClosed) { + f.unexpected.Add(1) + } + } + }(index) + } + go func() { defer workers.Done(); <-start; q.Stop() }() + close(start) + joined := make(chan struct{}) + f.auxJoined = append(f.auxJoined, joined) + go func() { workers.Wait(); close(joined) }() + select { + case <-joined: + case <-time.After(2 * time.Second): + t.Fatal("concurrent Stop/Enqueue callers did not join") + } + f.stopJoin(q) + if err := q.Enqueue(nil, "later:1"); !errors.Is(err, net.ErrClosed) { + t.Errorf("Enqueue after joined Stop=%v, want net.ErrClosed", err) + } + packetQueueTestNoError(t, q) + }) +} + +func TestPacketSendQueueAddressAndZeroPayloadAreCharged(t *testing.T) { + t.Run("address-bytes", func(t *testing.T) { + f := newPacketQueueTestFixture(t) + gate := f.gate() + seen := make(chan packetQueueTestEvent, 4) + q := f.queue(8, 3, func(ctx context.Context, _ []byte, address string) error { + f.publish(seen, packetQueueTestEvent{address: address}) + select { + case <-gate.done: + return nil + case <-ctx.Done(): + return context.Cause(ctx) + } + }) + first, second := strings.Repeat("a", packetQueueTestTargetBytes), strings.Repeat("b", packetQueueTestTargetBytes) + f.enqueue(q, nil, first) + f.next(seen, "empty-payload address charge") + f.wantFull(q, nil, first) + f.enqueue(q, nil, second) + f.next(seen, "second address charge") + f.wantFull(q, nil, "c") + f.stopJoin(q) + packetQueueTestNoError(t, q) + }) + t.Run("minimum-one-byte", func(t *testing.T) { + f := newPacketQueueTestFixture(t) + gate := f.gate() + var callbacks atomic.Int32 + seen := make(chan packetQueueTestEvent, 2) + // The packet quota deliberately leaves one spare slot after 128 KiB + // zero-byte packets; only minimum-one-byte accounting can reject next. + q := f.queue(2*(packetQueueTestTargetBytes+1), 1, func(ctx context.Context, _ []byte, _ string) error { + callbacks.Add(1) + f.publish(seen, packetQueueTestEvent{ctx: ctx}) + select { + case <-gate.done: + return nil + case <-ctx.Done(): + return context.Cause(ctx) + } + }) + f.enqueue(q, nil, "") + f.next(seen, "one active zero-byte packet") + for index := 1; index < packetQueueTestTargetBytes; index++ { + f.enqueue(q, nil, "") + } + f.wantFull(q, nil, "") + f.stopJoin(q) + if got := callbacks.Load(); got != 1 { + t.Errorf("Stop processed queued zero-byte packets: callbacks=%d, want 1", got) + } + packetQueueTestNoError(t, q) + }) + t.Run("oversized-and-combined", func(t *testing.T) { + f := newPacketQueueTestFixture(t) + var callbacks atomic.Int32 + q := f.queue(8, 2, func(context.Context, []byte, string) error { callbacks.Add(1); return nil }) + f.wantFull(q, make([]byte, packetQueueTestTargetBytes+1), "a") + f.wantFull(q, nil, strings.Repeat("a", packetQueueTestTargetBytes+1)) + f.wantFull(q, make([]byte, packetQueueTestTargetBytes), "a") + f.stopJoin(q) + if got := callbacks.Load(); got != 0 { + t.Errorf("oversized packet started callbacks=%d", got) + } + packetQueueTestNoError(t, q) + }) +} diff --git a/internal/transport/transport.go b/internal/transport/transport.go index cf30d1a..48bf540 100644 --- a/internal/transport/transport.go +++ b/internal/transport/transport.go @@ -46,6 +46,18 @@ type PacketConn interface { Close() error } +// PacketConcurrentSender is an optional, explicit Send concurrency contract. +// A limit greater than one permits independent-target Send calls to overlap; +// Close must interrupt every such call, as well as Receive. Owners still join +// their Send workers after Close. Unknown implementations remain serial. +// SendTargetKey must return the same canonical address used by Send, without +// DNS, I/O, or relaxing destination policy. Alias spellings then share a FIFO. +type PacketConcurrentSender interface { + PacketConn + ConcurrentSendLimit() int + SendTargetKey(address string) (string, error) +} + // PacketPayloadSizer is an optional capability for packet transports with a // logical-message limit below the UDP protocol maximum. Frontends use it to // reject an oversized payload without tearing down an otherwise healthy diff --git a/internal/tunnel/datagram.go b/internal/tunnel/datagram.go index f39dbca..1469c56 100644 --- a/internal/tunnel/datagram.go +++ b/internal/tunnel/datagram.go @@ -303,7 +303,6 @@ func (d *serverDatagramDispatcher) handleRequest(ctx context.Context, stream dea conn: socket, ctx: sessionCtx, cancel: cancel, - requests: make(chan autodatagram.Message, d.manager.config.receiveQueue), allowed: make(map[netip.AddrPort]struct{}), } sessionID, err := d.register(session) @@ -314,7 +313,6 @@ func (d *serverDatagramDispatcher) handleRequest(ctx context.Context, stream dea _ = protocol.WriteResponse(stream, protocol.Response{Status: protocol.StatusBusy, Message: "UDP dispatcher unavailable"}) return streamRequestOptions{Handled: true} } - session.start() response.SessionID = sessionID if err := protocol.WriteResponse(stream, response); err != nil { _ = session.Close() @@ -351,6 +349,9 @@ func (d *serverDatagramDispatcher) register(session *serverUDPSession) (uint32, continue } session.id = d.nextSession + // Initialize workers before publishing the session. Connection shutdown + // must never observe a zero WaitGroup before start adds its workers. + session.start() d.sessions[session.id] = session return session.id, nil } @@ -430,7 +431,7 @@ type serverUDPSession struct { conn *net.UDPConn ctx context.Context cancel context.CancelFunc - requests chan autodatagram.Message + requests *transport.PacketSendQueue allowedMu sync.RWMutex allowed map[netip.AddrPort]struct{} @@ -441,11 +442,13 @@ type serverUDPSession struct { } func (s *serverUDPSession) start() { - s.wg.Add(2) - go func() { - defer s.wg.Done() - s.forwardRequests() - }() + s.requests = transport.NewPacketSendQueue(s.ctx, s.manager.config.receiveQueue, 8, func(ctx context.Context, payload []byte, address string) error { + s.forwardRequest(ctx, payload, address) + // Resolution/write failure drops only this native packet, exactly as + // before; it must not become a terminal multi-target association error. + return nil + }) + s.wg.Add(1) go func() { defer s.wg.Done() s.forwardResponses() @@ -453,40 +456,20 @@ func (s *serverUDPSession) start() { } func (s *serverUDPSession) enqueue(message autodatagram.Message) { - select { - case <-s.ctx.Done(): - return - default: - } - select { - case s.requests <- message: - case <-s.ctx.Done(): - default: - // QUIC DATAGRAM is intentionally unreliable. Dropping one overloaded - // session must not block the connection-wide receive dispatcher. - } -} - -func (s *serverUDPSession) forwardRequests() { - for { - select { - case <-s.ctx.Done(): - return - case message := <-s.requests: - s.forwardRequest(message) - } - } + // One shared packet/byte budget covers all targets, including active + // resolutions. Pressure drops newest work without blocking the dispatcher. + _ = s.requests.Enqueue(message.Payload, message.Address) } -func (s *serverUDPSession) forwardRequest(message autodatagram.Message) { - ctx, cancel := context.WithTimeout(s.ctx, s.manager.config.resolveTimeout) - destinations, err := resolveUDPDestinations(ctx, s.manager.config.resolver, message.Address) +func (s *serverUDPSession) forwardRequest(parent context.Context, payload []byte, address string) { + ctx, cancel := context.WithTimeout(parent, s.manager.config.resolveTimeout) + destinations, err := resolveUDPDestinations(ctx, s.manager.config.resolver, address) cancel() if err != nil { return } for _, destination := range destinations { - if _, err := s.writeToDestination(message.Payload, destination); err == nil { + if _, err := s.writeToDestination(payload, destination); err == nil { return } } @@ -551,7 +534,9 @@ func (s *serverUDPSession) forwardResponses() { func (s *serverUDPSession) Close() error { s.closeOnce.Do(func() { s.cancel() + s.requests.Stop() s.closeErr = s.conn.Close() + s.requests.Wait() s.wg.Wait() s.dispatcher.unregister(s.id, s) s.manager.release(s.sourceKey) diff --git a/internal/tunnel/datagram_icmp_recovery_test.go b/internal/tunnel/datagram_icmp_recovery_test.go index 77e41ae..4cfcd66 100644 --- a/internal/tunnel/datagram_icmp_recovery_test.go +++ b/internal/tunnel/datagram_icmp_recovery_test.go @@ -61,11 +61,17 @@ func TestQUICDatagramClosedPortPreservesAssociation(t *testing.T) { resolver := UDPResolverFunc(func(ctx context.Context, address string) ([]netip.AddrPort, error) { endpoints, err := numeric.ResolveUDPContext(ctx, address) if err == nil && address == closedAddress.Load().(string) { - closedRequests.Add(1) + count := closedRequests.Add(1) select { case resolvedClosed <- struct{}{}: default: } + if count%9 == 0 { + // Same-target callbacks are FIFO. Entry into this ninth call + // witnesses the previous eight real write attempts returning. + // Reject this barrier call, so it cannot write after rebind. + return nil, errors.New("fixture completed closed-target write barrier") + } } return endpoints, err }) @@ -178,16 +184,16 @@ func TestQUICDatagramClosedPortPreservesAssociation(t *testing.T) { t.Fatal(err) } guard = nil - for index := 0; index < 8; index++ { + for index := 0; index < 9; index++ { if err := packet.Send([]byte{byte(round), byte(index)}, closedEndpoint.String()); err != nil { t.Fatal(err) } } - for closedRequests.Load() == before { + for closedRequests.Load() < before+9 { select { case <-resolvedClosed: case <-ctx.Done(): - t.Fatal("no closed-port request reached the relay") + t.Fatal("closed-target write-completion barrier did not reach the relay") } } // Give naturally generated local errors an opportunity to arrive. @@ -199,9 +205,9 @@ func TestQUICDatagramClosedPortPreservesAssociation(t *testing.T) { timer.Stop() t.Fatal(ctx.Err()) } - // Keep the destination closed until the healthy reply arrives. The - // relay processes requests serially, so this also places its observed - // closed-destination write attempt before reopening the guard socket. + // Keep the destination closed until both the same-target write barrier + // and the healthy reply. A different target's echo does not order this + // target's writes: independent target lanes are intentionally parallel. checkEcho([]byte{byte(round), 'o', 'k'}) guard, err = net.ListenUDP("udp4", closedEndpoint) if isWebTestAddressInUse(err) { diff --git a/internal/tunnel/udp_target_isolation_test.go b/internal/tunnel/udp_target_isolation_test.go new file mode 100644 index 0000000..3d65a96 --- /dev/null +++ b/internal/tunnel/udp_target_isolation_test.go @@ -0,0 +1,706 @@ +package tunnel + +import ( + "bytes" + "context" + "encoding/binary" + "errors" + "fmt" + "io" + "net" + "net/http" + "net/netip" + "strconv" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/cppla/autocar/internal/proxy" + "github.com/cppla/autocar/internal/transport" +) + +// Only resolver scheduling is synthetic. All payloads cross authenticated +// QUIC/H3, a real owned UDP socket, and (where selected) real SOCKS TCP/UDP. +// The independent 500ms wire oracle runs while the resolver's 2s budget is +// still live; a failed oracle still releases the gate and checks recovery. +func TestNativeUDPSlowTargetDoesNotBlockWarmTarget(t *testing.T) { + f := newUDPTargetIsolation(t, "native") + f.isolation() + f.sameTargetFIFO() + f.closeWhileResolving() +} + +func TestWebH3UDPConcurrentSendKeepsWarmTargetLive(t *testing.T) { + f := newUDPTargetIsolation(t, "h3") + f.isolation() + f.closeWhileResolving() +} + +func TestWebH3SOCKSUDPSlowTargetDoesNotBlockWarmTarget(t *testing.T) { + f := newUDPTargetIsolation(t, "socks") + f.isolation() + f.closeWhileResolving() +} + +func TestWebAutoSOCKSUDPSlowTargetDoesNotBlockWarmTarget(t *testing.T) { + f := newUDPTargetIsolation(t, "socks_auto") + f.isolation() + f.closeWhileResolving() +} + +type udpTargetIsolationGate struct { + entered chan context.Context + release chan struct{} + joined chan struct{} + canceled chan error + finish chan struct{} + once sync.Once + finishOnce sync.Once + closing bool + visits atomic.Int32 +} + +func newUDPTargetIsolationGate(closing bool) *udpTargetIsolationGate { + return &udpTargetIsolationGate{entered: make(chan context.Context, 1), release: make(chan struct{}), joined: make(chan struct{}), canceled: make(chan error, 1), finish: make(chan struct{}), closing: closing} +} + +func (g *udpTargetIsolationGate) Release() { g.once.Do(func() { close(g.release) }) } +func (g *udpTargetIsolationGate) Finish() { g.finishOnce.Do(func() { close(g.finish) }) } + +type udpTargetIsolationRecord struct { + payload string + address string + source netip.AddrPort + sequence int +} + +type udpTargetIsolationCall struct { + joined <-chan struct{} + result <-chan error +} + +type udpTargetIsolationFixture struct { + *webH2DialSharingFixture + mode string + echo *net.UDPConn + endpoint netip.AddrPort + warm, slow, fifo, closingTarget string + gates map[string]*udpTargetIsolationGate + packet transport.PacketConn + h3 *WebH3Client + auto *WebClient + physical *webH3ClientSession + entropy webH2AuthEntropyCounter + serverSlots func() int + control *net.TCPConn + local *net.UDPConn + relay *net.UDPAddr + closing atomic.Bool + unexpected atomic.Int32 + errors chan error + receipts, replies chan udpTargetIsolationRecord + receiptPending, replyPending map[string]udpTargetIsolationRecord + warmSource netip.AddrPort + mu sync.Mutex + handlers []<-chan struct{} +} + +func newUDPTargetIsolation(t *testing.T, mode string) *udpTargetIsolationFixture { + t.Helper() + f := &udpTargetIsolationFixture{webH2DialSharingFixture: newWebH2DialSharingFixture(t), mode: mode, + errors: make(chan error, 32), receipts: make(chan udpTargetIsolationRecord, 32), replies: make(chan udpTargetIsolationRecord, 32), + receiptPending: make(map[string]udpTargetIsolationRecord), replyPending: make(map[string]udpTargetIsolationRecord)} + // Also covers early constructor failure before the close list is complete. + t.Cleanup(func() { f.closing.Store(true) }) + echo, err := net.ListenUDP("udp4", &net.UDPAddr{IP: net.IPv4(127, 0, 0, 1)}) + if err != nil { + t.Fatal(err) + } + f.echo, f.endpoint = echo, echo.LocalAddr().(*net.UDPAddr).AddrPort() + port := strconv.Itoa(int(f.endpoint.Port())) + f.warm, f.slow = "warm.invalid:"+port, "slow.invalid:"+port + if mode == "native" { + f.warm = f.endpoint.String() + } + f.fifo, f.closingTarget = "fifo.invalid:"+port, "closing.invalid:"+port + f.gates = map[string]*udpTargetIsolationGate{f.slow: newUDPTargetIsolationGate(false), f.fifo: newUDPTargetIsolationGate(false), f.closingTarget: newUDPTargetIsolationGate(true)} + f.close = append(f.close, func() { _ = echo.Close() }) + echoJoined := make(chan struct{}) + f.register(echoJoined) + go func() { + defer close(echoJoined) + var buffer [2048]byte + sequence := 0 + for { + _ = echo.SetDeadline(time.Now().Add(7 * time.Second)) + n, source, err := echo.ReadFromUDPAddrPort(buffer[:]) + if err != nil { + f.noteIO(err) + return + } + sequence++ + f.publish(f.receipts, udpTargetIsolationRecord{payload: string(buffer[:n]), source: source, sequence: sequence}) + if _, err := echo.WriteToUDPAddrPort(buffer[:n], source); err != nil { + f.noteIO(err) + return + } + } + }() + resolver := UDPResolverFunc(func(ctx context.Context, address string) ([]netip.AddrPort, error) { + joined := make(chan struct{}) + f.register(joined) + defer close(joined) + if address != f.warm && f.gates[address] == nil { + f.unexpected.Add(1) + return nil, errors.New("isolation fixture forbids non-owned UDP targets") + } + if gate := f.gates[address]; gate != nil && gate.visits.Add(1) == 1 { + defer close(gate.joined) + gate.entered <- ctx + select { + case <-gate.release: + case <-ctx.Done(): + gate.canceled <- ctx.Err() + if gate.closing { + // This explicit completion gate distinguishes cancellation from + // callback completion when checking the real server lease. + select { + case <-gate.finish: + case <-f.ctx.Done(): + } + } + return nil, context.Cause(ctx) + case <-f.ctx.Done(): + return nil, context.Cause(f.ctx) + } + } + return []netip.AddrPort{f.endpoint}, nil + }) + serverTLS, clientTLS := testTLSConfigs(t) + refuseTCP := transport.DialFunc(func(context.Context, string, string) (net.Conn, error) { + f.unexpected.Add(1) + return nil, errors.New("isolation fixture forbids TCP destinations") + }) + var serve func(context.Context) error + if mode == "native" { + server, err := ListenQUIC(QUICServerConfig{Address: "127.0.0.1:0", Token: testToken, TLSConfig: serverTLS, + Dialer: refuseTCP, UDPResolver: resolver, DialTimeout: 2 * time.Second, HandshakeTimeout: time.Second, + MaxUDPSessions: 1, MaxClientUDPSessions: 1, MaxUDPDestinations: 4}) + if err != nil { + t.Fatal(err) + } + f.serverSlots = func() int { return len(server.udp.slots) } + f.close = append(f.close, func() { + if err := normalizeWebServerCloseError(server.Close()); err != nil { + t.Errorf("native server Close: %v", err) + } + }) + serve = server.Serve + client, err := NewClient(ClientConfig{ServerAddress: server.Addr().String(), Token: testToken, TLSConfig: clientTLS, QUICDialTimeout: time.Second, HandshakeTimeout: time.Second}) + if err != nil { + t.Fatal(err) + } + f.close = append(f.close, func() { _ = client.Close() }) + f.startServe(serve) + f.packet, err = client.DialPacket(f.ctx) + if err != nil { + t.Fatal(err) + } + verified := f.packet.(*quicPacketConn).dispatcher.conn.ConnectionState().TLS + if len(verified.VerifiedChains) == 0 { + t.Fatal("native association did not verify its real TLS peer") + } + } else { + server, err := ListenWebH3(WebH3ServerConfig{Address: "127.0.0.1:0", Token: webTestToken, TLSConfig: serverTLS, + Dialer: refuseTCP, UDPResolver: resolver, DialTimeout: 2 * time.Second, HandshakeTimeout: time.Second, + MaxUDPSessions: 4, MaxClientUDPSessions: 4, MaxUDPDestinations: 4, + Cover: http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { f.unexpected.Add(1); http.NotFound(w, r) })}) + if err != nil { + t.Fatal(err) + } + original := server.server.Handler + manager := original.(*webTunnelHandler).udp + f.serverSlots = func() int { return len(manager.slots) } + server.server.Handler = http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + joined := make(chan struct{}) + f.register(joined) + f.mu.Lock() + f.handlers = append(f.handlers, joined) + f.mu.Unlock() + defer close(joined) + original.ServeHTTP(w, r) + }) + f.close = append(f.close, func() { + if err := normalizeWebServerCloseError(server.Close()); err != nil { + t.Errorf("H3 server Close: %v", err) + } + }) + var dialer transport.Dialer + var packetDialer transport.PacketDialer + if mode == "socks_auto" { + client, err := NewWebClient(WebClientConfig{ServerAddress: server.Addr().String(), Token: webTestToken, TLSConfig: clientTLS, + H3FingerprintProfile: H3FingerprintNative, H3DialTimeout: time.Second, H2DialTimeout: time.Second, + HandshakeTimeout: time.Second, PrimaryAttemptTimeout: time.Second, MaxUDPSessions: 4, MaxUDPDestinations: 4}) + if err != nil { + t.Fatal(err) + } + f.auto, f.h3 = client, client.primary.dialer.(*WebH3Client) + dialer, packetDialer = client, client + f.close = append(f.close, func() { _ = client.Close() }) + } else { + client, err := NewWebH3Client(WebH3ClientConfig{ServerAddress: server.Addr().String(), Token: webTestToken, TLSConfig: clientTLS, + FingerprintProfile: H3FingerprintNative, DialTimeout: time.Second, HandshakeTimeout: time.Second, MaxUDPSessions: 4, MaxUDPDestinations: 4}) + if err != nil { + t.Fatal(err) + } + f.h3 = client + dialer, packetDialer = client, client + f.close = append(f.close, func() { _ = client.Close() }) + } + f.h3.signer = newWebAuthSigner(mustWebAuthKey(t, webTestToken), nil, &f.entropy) + f.startServe(server.Serve) + if f.isSOCKS() { + f.startSOCKS(dialer, packetDialer) + } else { + f.packet, err = packetDialer.DialPacket(f.ctx) + if err != nil { + t.Fatal(err) + } + } + } + f.close = append(f.close, func() { _ = f.packet.Close() }) + for _, gate := range f.gates { + gate := gate + f.close = append(f.close, func() { gate.Release(); gate.Finish() }) + } + f.close = append(f.close, func() { f.closing.Store(true) }) + f.checks = append(f.checks, func() { + for { + select { + case err := <-f.errors: + t.Errorf("owned isolation I/O: %v", err) + default: + goto drained + } + } + drained: + if n := f.unexpected.Load(); n != 0 { + t.Errorf("unexpected target or observer overflow count=%d", n) + } + }) + f.startReceiver() + return f +} + +func (f *udpTargetIsolationFixture) startServe(serve func(context.Context) error) { + joined, results := make(chan struct{}), make(chan error, 1) + f.register(joined) + f.checks = append(f.checks, func() { + if err := <-results; err != nil { + f.t.Errorf("owned Serve: %v", err) + } + }) + go func() { defer close(joined); results <- serve(f.ctx) }() +} + +// Return the exact PacketConn rather than a PacketConn-only wrapper: optional +// downstream capabilities must survive this observation-only PacketDialer. +type udpTargetIsolationDialer struct { + transport.Dialer + packetDialer transport.PacketDialer + returned chan transport.PacketConn +} + +func (d *udpTargetIsolationDialer) DialPacket(ctx context.Context) (transport.PacketConn, error) { + packet, err := d.packetDialer.DialPacket(ctx) + if err == nil { + d.returned <- packet + } + return packet, err +} + +func (f *udpTargetIsolationFixture) startSOCKS(client transport.Dialer, packetDialer transport.PacketDialer) { + f.t.Helper() + dialer := &udpTargetIsolationDialer{Dialer: client, packetDialer: packetDialer, returned: make(chan transport.PacketConn, 1)} + server, err := proxy.NewSOCKS5Server(proxy.Config{Dialer: dialer, IdleTimeout: 3 * time.Second}) + if err != nil { + f.t.Fatal(err) + } + listener, err := net.ListenTCP("tcp4", &net.TCPAddr{IP: net.IPv4(127, 0, 0, 1)}) + if err != nil { + f.t.Fatal(err) + } + f.close = append(f.close, func() { + ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second) + defer cancel() + if err := server.Shutdown(ctx); err != nil { + f.t.Errorf("SOCKS Shutdown: %v", err) + } + _ = listener.Close() + }) + joined, result := make(chan struct{}), make(chan error, 1) + f.register(joined) + f.checks = append(f.checks, func() { + if err := <-result; err != nil { + f.t.Errorf("SOCKS Serve: %v", err) + } + }) + go func() { defer close(joined); result <- server.Serve(listener) }() + control, err := net.DialTCP("tcp4", nil, listener.Addr().(*net.TCPAddr)) + if err != nil { + f.t.Fatal(err) + } + f.control = control + f.close = append(f.close, func() { _ = control.Close() }) + _ = control.SetDeadline(time.Now().Add(2 * time.Second)) + if _, err := control.Write([]byte{5, 1, 0}); err != nil { + f.t.Fatal(err) + } + var greeting [2]byte + if _, err := io.ReadFull(control, greeting[:]); err != nil || greeting != [2]byte{5, 0} { + f.t.Fatalf("SOCKS greeting=%v/%v", greeting, err) + } + if _, err := control.Write([]byte{5, 3, 0, 1, 0, 0, 0, 0, 0, 0}); err != nil { + f.t.Fatal(err) + } + var reply [10]byte + if _, err := io.ReadFull(control, reply[:]); err != nil || !bytes.Equal(reply[:4], []byte{5, 0, 0, 1}) { + f.t.Fatalf("SOCKS reply=%v/%v", reply, err) + } + _ = control.SetDeadline(time.Time{}) + f.relay = &net.UDPAddr{IP: append(net.IP(nil), reply[4:8]...), Port: int(binary.BigEndian.Uint16(reply[8:]))} + select { + case f.packet = <-dialer.returned: + case <-f.ctx.Done(): + f.t.Fatal("SOCKS packet ownership missing") + } + f.local, err = net.ListenUDP("udp4", &net.UDPAddr{IP: net.IPv4(127, 0, 0, 1)}) + if err != nil { + f.t.Fatal(err) + } + f.close = append(f.close, func() { _ = f.local.Close() }) +} + +func (f *udpTargetIsolationFixture) publish(channel chan<- udpTargetIsolationRecord, record udpTargetIsolationRecord) { + select { + case channel <- record: + default: + f.unexpected.Add(1) + } +} + +func (f *udpTargetIsolationFixture) noteIO(err error) { + if err == nil || (f.closing.Load() && errors.Is(err, net.ErrClosed)) { + return + } + select { + case f.errors <- err: + default: + f.unexpected.Add(1) + } +} + +func (f *udpTargetIsolationFixture) isSOCKS() bool { + return f.mode == "socks" || f.mode == "socks_auto" +} + +func (f *udpTargetIsolationFixture) startReceiver() { + joined := make(chan struct{}) + f.register(joined) + go func() { + defer close(joined) + for { + if !f.isSOCKS() { + payload, address, err := f.packet.Receive() + if err != nil { + f.noteIO(err) + return + } + f.publish(f.replies, udpTargetIsolationRecord{payload: string(payload), address: address}) + continue + } + var buffer [2048]byte + _ = f.local.SetReadDeadline(time.Now().Add(7 * time.Second)) + n, source, err := f.local.ReadFromUDP(buffer[:]) + if err != nil { + f.noteIO(err) + return + } + if source.String() != f.relay.String() || n < 7 || !bytes.Equal(buffer[:4], []byte{0, 0, 0, 3}) { + f.unexpected.Add(1) + continue + } + length := int(buffer[4]) + if n < 7+length { + f.unexpected.Add(1) + continue + } + address := net.JoinHostPort(string(buffer[5:5+length]), strconv.Itoa(int(binary.BigEndian.Uint16(buffer[5+length:7+length])))) + f.publish(f.replies, udpTargetIsolationRecord{payload: string(buffer[7+length : n]), address: address}) + } + }() +} + +func (f *udpTargetIsolationFixture) send(payload, target string) udpTargetIsolationCall { + joined, result := make(chan struct{}), make(chan error, 1) + f.register(joined) + go func() { + defer close(joined) + if !f.isSOCKS() { + result <- f.packet.Send([]byte(payload), target) + return + } + host, port, err := net.SplitHostPort(target) + if err != nil { + result <- err + return + } + portNumber, err := strconv.ParseUint(port, 10, 16) + if err != nil { + result <- err + return + } + _ = f.local.SetWriteDeadline(time.Now().Add(time.Second)) + _, err = f.local.WriteToUDP(webSOCKSUDPDomainPacket(host, uint16(portNumber), []byte(payload)), f.relay) + result <- err + }() + return udpTargetIsolationCall{joined: joined, result: result} +} + +func (f *udpTargetIsolationFixture) joinCall(call udpTargetIsolationCall, label string) error { + f.wait(call.joined, label) + return <-call.result +} + +func (f *udpTargetIsolationFixture) entered(gate *udpTargetIsolationGate) context.Context { + f.t.Helper() + select { + case ctx := <-gate.entered: + if err := ctx.Err(); err != nil { + f.t.Errorf("resolver was not live at actual entry: %v", err) + } + return ctx + case <-f.ctx.Done(): + f.t.Fatal("owned resolver entry witness missing") + return nil + } +} + +func (f *udpTargetIsolationFixture) record(ctx context.Context, channel <-chan udpTargetIsolationRecord, pending map[string]udpTargetIsolationRecord, payload string) (udpTargetIsolationRecord, bool) { + if record, ok := pending[payload]; ok { + delete(pending, payload) + return record, true + } + for { + select { + case record := <-channel: + if record.payload == payload { + return record, true + } + if _, exists := pending[record.payload]; exists { + f.unexpected.Add(1) + } + pending[record.payload] = record + case <-ctx.Done(): + return udpTargetIsolationRecord{}, false + } + } +} + +func (f *udpTargetIsolationFixture) wire(payload, target string, timeout time.Duration) (udpTargetIsolationRecord, bool) { + ctx, cancel := context.WithTimeout(f.ctx, timeout) + defer cancel() + receipt, ok := f.record(ctx, f.receipts, f.receiptPending, payload) + if !ok { + return receipt, false + } + reply, ok := f.record(ctx, f.replies, f.replyPending, payload) + if !ok { + f.receiptPending[payload] = receipt + return receipt, false + } + want := target + if f.mode == "native" { + want = f.endpoint.String() + } + if reply.address != want { + f.t.Errorf("actual UDP reply=%q, want %q", reply.address, want) + } + if target == f.warm && f.warmSource.IsValid() && receipt.source != f.warmSource { + f.t.Errorf("warm target relay source changed: %s -> %s", f.warmSource, receipt.source) + } + return receipt, true +} + +func (f *udpTargetIsolationFixture) requireWire(payload, target string) udpTargetIsolationRecord { + f.t.Helper() + record, ok := f.wire(payload, target, 2*time.Second) + if !ok { + f.t.Fatalf("unexpected fixture/recovery failure: actual %s echo did not complete", payload) + } + return record +} + +func (f *udpTargetIsolationFixture) isolation() { + f.t.Helper() + if err := f.joinCall(f.send("warm-A1", f.warm), "first warm Send"); err != nil { + f.t.Fatal(err) + } + f.warmSource = f.requireWire("warm-A1", f.warm).source + if f.h3 != nil { + f.physical = webH3SelectedSession(f.t, f.h3) + } + gate := f.gates[f.slow] + slow := f.send("slow-B1", f.slow) + resolverContext := f.entered(gate) + healthy := f.send("warm-A2", f.warm) + _, beforeRelease := f.wire("warm-A2", f.warm, 500*time.Millisecond) + if !beforeRelease { + f.t.Errorf("warm A2 did not actually echo while the independent slow resolver gate remained held") + } + if err := resolverContext.Err(); err != nil { + f.t.Errorf("slow resolver ended before gate release: %v", err) + } + select { + case <-gate.release: + f.t.Error("wire oracle ran after gate release") + default: + } + gate.Release() // Recovery still runs after the old implementation's failure. + if err := f.joinCall(slow, "released slow Send"); err != nil { + f.t.Errorf("unexpected slow-target recovery error: %v", err) + } + if err := f.joinCall(healthy, "warm A2 Send"); err != nil { + f.t.Errorf("unexpected warm-target recovery error: %v", err) + } + if !beforeRelease { + f.requireWire("warm-A2", f.warm) + } + f.requireWire("slow-B1", f.slow) + f.wait(gate.joined, "released resolver callback") + f.assertH3Continuity() + f.t.Logf("%s: actual warm source=%s, A2-before-release=%t; both complete recovery payloads verified", f.mode, f.warmSource, beforeRelease) +} + +func (f *udpTargetIsolationFixture) sameTargetFIFO() { + // Send completion proves the native client accepted each datagram, not + // server ingress before gate release. This checks actual loopback delivery + // order only; server queue barriers belong to the dedicated queue tests. + gate := f.gates[f.fifo] + first := f.send("fifo-1", f.fifo) + f.entered(gate) + second := f.send("fifo-2", f.fifo) + if err := f.joinCall(second, "second native FIFO enqueue"); err != nil { + f.t.Fatal(err) + } + third := f.send("fifo-3", f.fifo) + if err := f.joinCall(third, "third native FIFO enqueue"); err != nil { + f.t.Fatal(err) + } + gate.Release() + if err := f.joinCall(first, "first native FIFO enqueue"); err != nil { + f.t.Fatal(err) + } + var previous udpTargetIsolationRecord + for _, payload := range []string{"fifo-1", "fifo-2", "fifo-3"} { + record := f.requireWire(payload, f.fifo) + if previous.sequence != 0 && record.sequence != previous.sequence+1 { + f.t.Errorf("native same-target loopback FIFO changed: %+v -> %+v", previous, record) + } + if record.source != f.warmSource { + f.t.Errorf("native target switch changed the association socket: %s/%s", record.source, f.warmSource) + } + previous = record + } + f.wait(gate.joined, "native FIFO resolver callback") +} + +func (f *udpTargetIsolationFixture) assertH3Continuity() { + if f.h3 == nil { + return + } + f.h3.mu.Lock() + same := f.h3.conns[f.h3.conn] == f.physical + ready := f.physical != nil && f.physical.authState == webH3ClientAuthReady && f.physical.auth != nil && !f.physical.retired + f.h3.mu.Unlock() + if !same || !ready || f.physical.conn.Context().Err() != nil || f.entropy.nonceReads.Load() != 1 { + f.t.Errorf("actual H3 physical/Ready/bootstrap continuity=%t/%t/%d", same, ready, f.entropy.nonceReads.Load()) + } + if f.auto != nil && f.auto.SelectedTransport() != webAuthTransportH3 { + f.t.Errorf("actual automatic UDP path selected %q, want h3 with no H2 fallback", f.auto.SelectedTransport()) + } +} + +func (f *udpTargetIsolationFixture) closeWhileResolving() { + gate := f.gates[f.closingTarget] + send := f.send("close-pending", f.closingTarget) + resolverContext := f.entered(gate) + f.closing.Store(true) + closed, result := make(chan struct{}), make(chan error, 1) + f.register(closed) + go func() { + defer close(closed) + if !f.isSOCKS() { + result <- f.packet.Close() + return + } + if err := f.control.CloseWrite(); err != nil { + result <- err + return + } + _ = f.control.SetReadDeadline(time.Now().Add(2 * time.Second)) + var one [1]byte + _, err := f.control.Read(one[:]) + if errors.Is(err, io.EOF) { + err = nil + } + result <- err + }() + select { + case err := <-gate.canceled: + if err != context.Canceled || resolverContext.Err() != context.Canceled { + f.t.Errorf("Close ended live resolver with %v/%v, want cancellation", err, resolverContext.Err()) + } + case <-time.After(2 * time.Second): + f.t.Error("Close did not cancel the actual pending resolver") + } + if slots := f.serverSlots(); slots < 1 { + f.t.Errorf("server released all UDP leases while its canceled resolver callback was still held: %d", slots) + } + gate.Finish() + gate.Release() // Independently bounds negative-control cleanup too. + f.wait(gate.joined, "canceled resolver callback joined") + f.wait(closed, "association Close/control peer EOF") + if err := <-result; err != nil { + f.t.Errorf("association close/actual control EOF: %v", err) + } + if err := f.joinCall(send, "pending Send joined after Close"); f.mode == "h3" && err == nil { + f.t.Error("pending H3 Send succeeded after PacketConn Close") + } + if f.h3 != nil { + f.mu.Lock() + handlers := append([]<-chan struct{}(nil), f.handlers...) + f.mu.Unlock() + for _, handler := range handlers { + f.wait(handler, "actual H3 handler completion") + } + if slots := len(f.h3.udpSlots); slots != 0 { + f.t.Errorf("client retained UDP leases after Close and handler joins: %d", slots) + } + } + timer, tick := time.NewTimer(2*time.Second), time.NewTicker(10*time.Millisecond) + defer timer.Stop() + defer tick.Stop() + for f.serverSlots() != 0 { + select { + case <-tick.C: + case <-timer.C: + f.t.Errorf("server retained UDP leases after resolver/association shutdown: %d", f.serverSlots()) + return + } + } + f.t.Log(fmt.Sprintf("%s: Close canceled pending resolver, retained admission until callback completion, and joined shutdown with zero server leases", f.mode)) +} + +var _ transport.PacketDialer = (*udpTargetIsolationDialer)(nil) diff --git a/internal/tunnel/web_client.go b/internal/tunnel/web_client.go index 0838ce1..a7eb6ca 100644 --- a/internal/tunnel/web_client.go +++ b/internal/tunnel/web_client.go @@ -541,6 +541,20 @@ func (c *webClientPacketConn) MaxPayloadSize() int { return webConnectUDPMaxPayloadSize } +func (c *webClientPacketConn) ConcurrentSendLimit() int { + if sender, ok := c.inner.(transport.PacketConcurrentSender); ok { + return sender.ConcurrentSendLimit() + } + return 0 +} + +func (c *webClientPacketConn) SendTargetKey(address string) (string, error) { + if sender, ok := c.inner.(transport.PacketConcurrentSender); ok { + return sender.SendTargetKey(address) + } + return address, nil +} + // Close prevents new streams, interrupts in-flight dials, and closes both // underlying multiplexed transports. It is safe to call concurrently. func (c *WebClient) Close() error { @@ -560,3 +574,4 @@ var _ transport.Dialer = (*WebClient)(nil) var _ transport.PacketDialer = (*WebClient)(nil) var _ transport.PacketConn = (*webClientPacketConn)(nil) var _ transport.PacketPayloadSizer = (*webClientPacketConn)(nil) +var _ transport.PacketConcurrentSender = (*webClientPacketConn)(nil) diff --git a/internal/tunnel/web_udp.go b/internal/tunnel/web_udp.go index 20dcb93..82b2c2f 100644 --- a/internal/tunnel/web_udp.go +++ b/internal/tunnel/web_udp.go @@ -683,6 +683,13 @@ func (p *webUDPPacketConn) Close() error { func (p *webUDPPacketConn) MaxPayloadSize() int { return webConnectUDPMaxPayloadSize } +func (p *webUDPPacketConn) ConcurrentSendLimit() int { return min(8, p.maxTargets) } + +func (p *webUDPPacketConn) SendTargetKey(address string) (string, error) { + _, canonical, err := connectUDPPath(address) + return canonical, err +} + type webUDPClientSession struct { packet *webUDPPacketConn target string @@ -1075,3 +1082,4 @@ func (c *WebH3Client) DialPacket(ctx context.Context) (transport.PacketConn, err var _ transport.PacketDialer = (*WebH3Client)(nil) var _ transport.PacketConn = (*webUDPPacketConn)(nil) var _ transport.PacketPayloadSizer = (*webUDPPacketConn)(nil) +var _ transport.PacketConcurrentSender = (*webUDPPacketConn)(nil) diff --git a/internal/tunnel/web_udp_concurrency_capability_test.go b/internal/tunnel/web_udp_concurrency_capability_test.go new file mode 100644 index 0000000..fefd9e4 --- /dev/null +++ b/internal/tunnel/web_udp_concurrency_capability_test.go @@ -0,0 +1,42 @@ +package tunnel + +import ( + "testing" + + "github.com/cppla/autocar/internal/transport" +) + +// These pure capability checks use the production CONNECT-UDP canonicalizer, +// not a fixture-provided key. Real Send/Close and wrapper-chain behavior are +// covered by the separate authenticated slow-target wire tests. +func TestWebUDPConcurrentSendCapabilityPreservesTargetIdentity(t *testing.T) { + for _, maxTargets := range []int{0, 1, 7, 32} { + inner := &webUDPPacketConn{maxTargets: maxTargets} + wrapped := &webClientPacketConn{inner: inner} + for _, sender := range []transport.PacketConcurrentSender{inner, wrapped} { + if got, want := sender.ConcurrentSendLimit(), min(8, maxTargets); got != want { + t.Errorf("%T concurrent limit=%d, want %d", sender, got, want) + } + for _, test := range []struct{ address, want string }{ + {"UPPER.invalid:0053", "upper.invalid:53"}, + {"UPPER.invalid.:53", "upper.invalid.:53"}, + {"bücher.invalid:53", "xn--bcher-kva.invalid:53"}, + {"[2001:0db8:0:0:0:0:0:1]:0053", "[2001:db8::1]:53"}, + {"[::ffff:127.0.0.1]:53", "127.0.0.1:53"}, + } { + got, err := sender.SendTargetKey(test.address) + if err != nil || got != test.want { + t.Errorf("%T key(%q)=%q, %v; want %q", sender, test.address, got, err, test.want) + } + } + for _, address := range []string{"missing-port", "example.invalid:0", "[fe80::1%zone]:53"} { + if _, err := sender.SendTargetKey(address); err == nil { + t.Errorf("%T accepted invalid target key %q", sender, address) + } + } + } + } + if got := (&webClientPacketConn{}).ConcurrentSendLimit(); got != 0 { + t.Errorf("empty wrapper advertised concurrency %d", got) + } +}