React to UDP loss and report missing data

README.md

@@ -2,7 +2,7 @@
 
 DebTik is a Debian/Proxmox router appliance with an embedded web interface, REST API and RouterOS-style CLI. It manages network interfaces, routing, firewall/NAT, DHCP/DNS, WireGuard, IPsec, PPP, VRRP and AeroTUN. AeroTUN v3 uses pinned peer identities and encrypted tunnels; it works with native TUN or a veth fallback when `/dev/net/tun` is unavailable.
 
-Current release: **1.3.7**. See [functionality corrections and verification](docs/functionality-fixes.md) for network persistence, firewall isolation/IPv6, service error handling and certificate renewal changes.
+Current release: **1.3.8**. See [functionality corrections and verification](docs/functionality-fixes.md) for network persistence, firewall isolation/IPv6, service error handling and certificate renewal changes.
 
 ## Install
 
@@ -21,10 +21,10 @@ The internal Git server uses a private CA. Trust that CA before cloning, or repl
 ```sh
 git -c http.sslVerify=false \
   -c 'http.pinnedpubkey=sha256//u4xyt0njDzn4BAQnS6OTGrXBxVE8vMg5BSNIuryUsMo=' \
-  clone --depth 1 --branch v1.3.7 https://gogs.eparmak.local/eparmak/DebTik.git
+  clone --depth 1 --branch v1.3.8 https://gogs.eparmak.local/eparmak/DebTik.git
 ```
 
-This verifies the server's pinned key for this command; it does not change global Git settings. Reverify the pin if the server's TLS key changes. For a reproducible fresh installation, `v1.3.7` contains the complete dependency installer and native Linux firewall corrections.
+This verifies the server's pinned key for this command; it does not change global Git settings. Reverify the pin if the server's TLS key changes. For a reproducible fresh installation, `v1.3.8` contains the complete dependency installer and native Linux firewall corrections.
 
 `install.sh` installs and verifies the dependencies listed in [Dependencies](docs/dependencies.md), including Go, WireGuard, firewall, BIRD, DHCP/DNS, PPP clients and servers, strongSwan plugins, diagnostics and the RSTP daemon. It builds `proxmox-router` and `debtik-cli`, installs the updater helpers, and starts `proxmox-router.service`. It can be run again; existing `/etc/proxmox-router` files and tunnel keys are preserved. Previous binaries are saved as `/usr/local/bin/*.previous`. For native TUN in an LXC container, enable `/dev/net/tun` on the Proxmox host; otherwise AeroTUN uses veth.
 

VERSION

@@ -1 +1 @@
-1.3.7
+1.3.8

docs/bwtest-rx-fix.md

@@ -90,3 +90,15 @@ automatic/zero limit, one stream and requested 1472-byte packet size. RX no long
 starved. Both services are active, installed hashes match the built artifacts,
 rollback timers were cancelled after verification, and the persistent AeroTUN
 configuration/PSK remained unchanged.
+
+## Loss-aware automatic UDP pacing (1.3.8)
+
+Automatic UDP senders use receiver packet and loss counters as well as delivered
+bytes. On observed loss the sender backs off and delays its next rate increase;
+explicit bandwidth limits are unchanged. An update is required on the sending
+side of each direction: for a local RX test, update the remote bandwidth-test
+server as well. Existing receiver-v1 peers remain compatible.
+
+A UDP test with no data received or confirmed now reports an error rather than
+a successful zero-loss result, and failed tests are excluded from successful
+history. TCP control connectivity alone does not establish UDP reachability.

internal/aerotun/integration_test.go

@@ -253,7 +253,7 @@ func runBandwidthPair(t *testing.T, compress bool) {
 		t.Logf("%s: %.1f Mbit/s, %d received bytes", mode, result.End.Received.BPS/1e6, result.End.Received.Bytes)
 	}
 	if !compress {
-		ctx, cancel := context.WithTimeout(t.Context(), 15*time.Second)
+		ctx, cancel := context.WithTimeout(t.Context(), 30*time.Second)
 		defer cancel()
 		for i, name := range []string{"atest-a", "atest-b"} {
 			out, err := nsCmd(ctx, i, "tc", "qdisc", "add", "dev", name, "root", "netem", "rate", "5mbit", "delay", "15ms", "limit", "64").CombinedOutput()
@@ -320,10 +320,10 @@ func TestUDPOverAeroTunConstrainedLink(t *testing.T) {
 		t.Run(direction, func(t *testing.T) {
 			client := bwtest.NewClient()
 			defer client.Stop()
-			if err := client.Run(t.Context(), bwtest.SessionRequest{Target: "10.99.0.1", Port: 15200, Protocol: "udp", Direction: direction, Streams: 4, PacketSize: 1472, Duration: 4}); err != nil {
+			if err := client.Run(t.Context(), bwtest.SessionRequest{Target: "10.99.0.1", Port: 15200, Protocol: "udp", Direction: direction, Streams: 4, PacketSize: 1472, Duration: 10}); err != nil {
 				t.Fatal(err)
 			}
-			deadline := time.Now().Add(6 * time.Second)
+			deadline := time.Now().Add(13 * time.Second)
 			for client.GetStatus().Running && time.Now().Before(deadline) {
 				time.Sleep(20 * time.Millisecond)
 			}

internal/bwtest/client.go

@@ -333,12 +333,26 @@ func (c *BwTestClient) executeTest(ctx context.Context, cancel context.CancelFun
 				serverRxBytes.Store(rep.RxBytes)
 				serverRxPackets.Store(rep.RxPackets)
 				serverLostPackets.Store(rep.LostPackets)
-				udpPacing.observe(rep.RxBytes, time.Now())
+				udpPacing.observe(rep.RxBytes, rep.RxPackets, rep.LostPackets, time.Now())
 			}
 		}
 	}()
 
 	var wg sync.WaitGroup
+	if req.Protocol == "udp" {
+		defer func() {
+			c.mu.Lock()
+			defer c.mu.Unlock()
+			if c.status.Error != "" || time.Since(c.status.StartTime) < time.Second {
+				return
+			}
+			if req.Direction != "tx" && localRxPackets.Load() == 0 {
+				c.status.Error = "No UDP data received; check UDP connectivity and firewall rules"
+			} else if req.Direction != "rx" && serverRxPackets.Load() == 0 && serverRxBytes.Load() == 0 {
+				c.status.Error = "No UDP delivery confirmed by the receiver; check UDP connectivity and firewall rules"
+			}
+		}()
+	}
 
 	// UDP Data path
 	if req.Protocol == "udp" {

internal/bwtest/manager.go

@@ -119,7 +119,7 @@ func (m *Manager) GetTestStatus() TestStatus {
 	st := m.client.GetStatus()
 
 	// If test completed with a summary, persist into history
-	if !st.Running && st.Summary != nil {
+	if !st.Running && st.Summary != nil && st.Error == "" {
 		m.mu.Lock()
 		alreadyExists := false
 		for _, h := range m.history {

internal/bwtest/server.go

@@ -380,7 +380,7 @@ func (s *BwTestServer) handleClientControl(serverCtx context.Context, conn net.C
 				break
 			}
 			if msg.Type == "report" && msg.SessionID == sess.id {
-				sess.udpPacing.observe(msg.RxBytes, time.Now())
+				sess.udpPacing.observe(msg.RxBytes, msg.RxPackets, msg.LostPackets, time.Now())
 			}
 		}
 	}

internal/bwtest/telemetry_test.go

@@ -29,6 +29,21 @@ func TestSummaryCoversSamplesBeyondRetentionWindow(t *testing.T) {
 	}
 }
 
+func TestFailedSessionIsNotRecordedAsSuccessfulHistory(t *testing.T) {
+	c := NewClient()
+	c.status = TestStatus{SessionID: "failed", Error: "No UDP data received", Summary: &TestSummary{}}
+	m := &Manager{client: c}
+	m.GetTestStatus()
+	if len(m.GetHistory()) != 0 {
+		t.Fatal("failed session was recorded as successful")
+	}
+	c.status.Error = ""
+	m.GetTestStatus()
+	if len(m.GetHistory()) != 1 {
+		t.Fatal("successful session not recorded")
+	}
+}
+
 func TestObserversReceiveBroadcastAndCanUnsubscribe(t *testing.T) {
 	c := NewClient()
 	a, stopA := c.Subscribe()

internal/bwtest/udp_control.go

@@ -21,7 +21,9 @@ type udpPacer struct {
 	automatic                         bool
 	rate                              int64
 	started, lastFeedback, lastChange time.Time
+	lastLoss                          time.Time
 	delivered                         int64
+	adjustPackets, adjustLost         int64
 }
 
 func newUDPPacer(limit int64) *udpPacer {
@@ -69,7 +71,7 @@ func (p *udpPacer) wait(ctx context.Context, n int) bool {
 	}
 }
 func (p *udpPacer) setRate(rate int64) { p.rate = rate; p.limiter.bps = rate }
-func (p *udpPacer) observe(bytes int64, now time.Time) {
+func (p *udpPacer) observe(bytes, packets, lost int64, now time.Time) {
 	p.mu.Lock()
 	defer p.mu.Unlock()
 	dt := now.Sub(p.lastFeedback).Seconds()
@@ -82,12 +84,21 @@ func (p *udpPacer) observe(bytes int64, now time.Time) {
 	if !p.automatic || now.Sub(p.started) < time.Second || now.Sub(p.lastChange) < 450*time.Millisecond {
 		return
 	}
+	// Loss counts can decrease when delayed packets arrive. Compare cumulative
+	// counters since the previous decision and never interpret reordering as
+	// negative loss. Older peers without packet counts still use byte feedback.
+	packetDelta := max(int64(0), packets-p.adjustPackets)
+	lossDelta := max(int64(0), lost-p.adjustLost)
+	p.adjustPackets, p.adjustLost = packets, lost
 	deliveredRate := int64(float64(delta) * 8 / dt)
 	switch {
+	case packetDelta+lossDelta >= 10 && lossDelta*200 >= packetDelta+lossDelta:
+		p.setRate(max(int64(128000), min(p.rate*8/10, deliveredRate*9/10)))
+		p.lastLoss = now
 	case deliveredRate < p.rate*7/10:
 		p.setRate(max(int64(128000), deliveredRate*11/10))
-	case deliveredRate >= p.rate*85/100:
-		p.setRate(min(int64(10_000_000_000), p.rate*5/4))
+	case deliveredRate >= p.rate*85/100 && (p.lastLoss.IsZero() || now.Sub(p.lastLoss) >= 2*time.Second):
+		p.setRate(min(int64(10_000_000_000), p.rate*115/100))
 	}
 	p.lastChange = now
 }

internal/bwtest/udp_control_test.go

@@ -1,7 +1,11 @@
 package bwtest
 
 import (
+	"bufio"
 	"context"
+	"encoding/json"
+	"net"
+	"strings"
 	"testing"
 	"time"
 )
@@ -10,21 +14,99 @@ func TestUDPPacerUsesDeliveryFeedback(t *testing.T) {
 	p := newUDPPacer(0)
 	initial := p.rate
 	now := p.started.Add(2 * time.Second)
-	p.observe(10000, now)
+	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, now.Add(time.Second))
+	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, fixed.started.Add(2*time.Second))
+	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)