DebTik/internal/bwtest/udp_control_test.go

package bwtest

import (
	"bufio"
	"context"
	"encoding/json"
	"net"
	"strings"
	"testing"
	"time"
)

func TestUDPPacerUsesDeliveryFeedback(t *testing.T) {
	p := newUDPPacer(0)
	initial := p.rate
	now := p.started.Add(2 * time.Second)
	p.observe(10000, 10, 0, now)
	if p.rate >= initial {
		t.Fatal("automatic rate did not fall after severe loss")
	}
	lower := p.rate
	p.observe(10000+lower/8, 110, 0, now.Add(time.Second))
	if p.rate <= lower {
		t.Fatal("automatic rate did not grow after successful delivery")
	}
	fixed := newUDPPacer(1234567)
	fixed.observe(0, 0, 0, fixed.started.Add(2*time.Second))
	if fixed.rate != 1234567 {
		t.Fatal("receiver feedback changed explicit rate limit")
	}
}

func TestAutomaticUDPPacerBacksOffForPacketLoss(t *testing.T) {
	p := newUDPPacer(0)
	now := p.started.Add(2 * time.Second)
	// Delivery remains high enough that byte-only feedback would increase rate.
	p.observe(480000, 1000, 20, now)
	if p.rate >= 2_000_000 {
		t.Fatalf("rate did not fall on loss: %d", p.rate)
	}
	lower := p.rate
	p.observe(480000+lower/8, 2000, 20, now.Add(time.Second))
	if p.rate > lower {
		t.Fatalf("rate rose during loss cooldown: %d", p.rate)
	}
	p.observe(480000+3*lower/8, 3000, 20, now.Add(3*time.Second))
	if p.rate <= lower {
		t.Fatalf("rate failed to recover after cooldown: %d", p.rate)
	}
	fixed := newUDPPacer(2_000_000)
	fixed.observe(480000, 1000, 100, fixed.started.Add(2*time.Second))
	if fixed.rate != 2_000_000 {
		t.Fatal("fixed rate changed on loss")
	}
	legacy := newUDPPacer(0)
	legacy.observe(500000, 0, 0, legacy.started.Add(2*time.Second))
	if legacy.rate <= 2_000_000 {
		t.Fatal("byte-only feedback from older peer no longer works")
	}
}

func TestUDPNoDeliveryReportsError(t *testing.T) {
	ln, err := net.Listen("tcp", "127.0.0.1:0")
	if err != nil {
		t.Fatal(err)
	}
	defer ln.Close()
	u, err := net.ListenUDP("udp", &net.UDPAddr{IP: net.ParseIP("127.0.0.1")})
	if err != nil {
		t.Fatal(err)
	}
	defer u.Close()
	go func() {
		for {
			conn, e := ln.Accept()
			if e != nil {
				return
			}
			go func() {
				defer conn.Close()
				rd := bufio.NewReader(conn)
				if _, e := rd.ReadBytes('\n'); e != nil {
					return
				}
				json.NewEncoder(conn).Encode(SessionHandshake{Type: "accept", UDPFeedback: udpFeedbackVersion, UDPPort: u.LocalAddr().(*net.UDPAddr).Port, PacketSize: 1200})
				for {
					if _, e := rd.ReadBytes('\n'); e != nil {
						return
					}
				}
			}()
		}
	}()
	for _, direction := range []string{"tx", "rx", "both"} {
		c := NewClient()
		if err := c.Run(context.Background(), SessionRequest{Target: "127.0.0.1", Port: ln.Addr().(*net.TCPAddr).Port, Protocol: "udp", Direction: direction, Duration: 2, PacketSize: 1200}); err != nil {
			t.Fatal(err)
		}
		deadline := time.Now().Add(4 * time.Second)
		for c.GetStatus().Running && time.Now().Before(deadline) {
			time.Sleep(10 * time.Millisecond)
		}
		st := c.GetStatus()
		c.Stop()
		if st.Running || !strings.Contains(st.Error, "No UDP") {
			t.Fatalf("%s no-data not reported: %+v", direction, st)
		}
	}
}
func TestUDPPacerCancellation(t *testing.T) {
	p := newUDPPacer(1)
	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Millisecond)
	defer cancel()
	if p.wait(ctx, 1472) {
		t.Fatal("rate limiter ignored cancellation")
	}
}
func TestUDPTrackerStreamsReorderingDuplicates(t *testing.T) {
	var tracker udpTracker
	for _, v := range []struct {
		stream uint8
		seq    uint64
		lost   int64
	}{{0, 1, 0}, {1, 1, 0}, {0, 3, 1}, {1, 2, 1}, {0, 2, 0}, {0, 4, 0}} {
		ok, lost := tracker.accept(v.stream, v.seq)
		if !ok || lost != v.lost {
			t.Fatalf("stream %d seq %d: accepted=%v lost=%d want=%d", v.stream, v.seq, ok, lost, v.lost)
		}
	}
	if ok, lost := tracker.accept(0, 3); ok || lost != 0 {
		t.Fatal("duplicate counted as data or loss")
	}
}
func TestBandwidthUDPDirectionsAndAutomaticPacing(t *testing.T) {
	s := testServer(t)
	for _, direction := range []string{"rx", "both", "tx"} {
		t.Run(direction, func(t *testing.T) {
			c := NewClient()
			t.Cleanup(c.Stop)
			if err := c.Run(context.Background(), SessionRequest{Target: "127.0.0.1", Port: s.config.Port, Protocol: "udp", Direction: direction, Streams: 4, PacketSize: 1472, Duration: 2}); err != nil {
				t.Fatal(err)
			}
			deadline := time.Now().Add(5 * time.Second)
			for c.GetStatus().Running && time.Now().Before(deadline) {
				time.Sleep(10 * time.Millisecond)
			}
			st := c.GetStatus()
			if st.Running || st.Error != "" || st.Summary == nil {
				t.Fatalf("UDP test failed: %+v", st)
			}
			if direction != "rx" && st.Summary.TotalTxBytes < 100000 {
				t.Fatalf("no confirmed TX: %+v", st.Summary)
			}
			if direction != "tx" && st.Summary.TotalRxBytes < 100000 {
				t.Fatalf("no reverse UDP: %+v", st.Summary)
			}
			if st.Summary.AvgLossPct != 0 {
				t.Fatalf("streams or reordered packets falsely counted as loss: %+v", st.Summary)
			}
			if st.Summary.AvgTxMbps > 20 || st.Summary.AvgRxMbps > 20 {
				t.Fatal("automatic UDP started with an unbounded flood")
			}
			if direction == "tx" && st.Summary.TotalRxBytes != 0 {
				t.Fatal("TX delivery reports mislabeled as downloaded data")
			}
		})
	}
}