Worker frames are now FrameType + payload; there is no stream id. Each player gets its own worker conn. maxTunnels (default 256) caps concurrent tunnels. The old maxConn pool size is ignored so existing configs do not silently admit only a handful of players. Resume, per-direction windows, the control session, and the DATA-only shaper stay. A dropped worker still hangs that one player and reattaches over a fresh conn. Add a hub-side per-IP limiter for player intents only (default 8/s, burst 16, 64 concurrent). Unmatched hostnames consume a token; Intent 17 is never counted. 0 disables each knob.
148 lines
4.7 KiB
Go
148 lines
4.7 KiB
Go
package e2e
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
"io"
|
|
"net"
|
|
"testing"
|
|
"time"
|
|
|
|
"github.com/iceBear67/redapricot/client"
|
|
)
|
|
|
|
// startClientWithBandwidth is startClient with an egress cap, for the shaping
|
|
// tests. Only the client→hub direction is shaped, which in this harness is the
|
|
// leg carrying the mock destination's echo back to the player.
|
|
func startClientWithBandwidth(t *testing.T, hubAddr, psk string, maxTunnels int, bandwidth string, mappings []client.Mapping) *client.Client {
|
|
t.Helper()
|
|
cfg := &client.Config{
|
|
Server: hubAddr,
|
|
PSK: psk,
|
|
MaxTunnels: maxTunnels,
|
|
PingIntervalMs: 20000,
|
|
MaxBandwidth: bandwidth,
|
|
Mappings: mappings,
|
|
}
|
|
c := client.New(cfg)
|
|
ctx, cancel := context.WithCancel(context.Background())
|
|
t.Cleanup(func() {
|
|
cancel()
|
|
c.Close()
|
|
})
|
|
if err := c.Start(ctx); err != nil {
|
|
t.Fatalf("client start: %v", err)
|
|
}
|
|
time.Sleep(200 * time.Millisecond)
|
|
return c
|
|
}
|
|
|
|
// drain reads a connection until it fails, so a greedy player keeps pulling
|
|
// bytes instead of stalling on its own receive window.
|
|
func drain(conn net.Conn) {
|
|
go func() { _, _ = io.Copy(io.Discard, conn) }()
|
|
}
|
|
|
|
// TestBandwidthCapIsEnforced pushes a payload that cannot fit in the burst and
|
|
// asserts the transfer takes at least as long as the configured rate implies.
|
|
// This is the first timing assertion in the suite, so the bounds are wide: the
|
|
// unshaped path completes this in well under a second, making the lower bound
|
|
// an unambiguous signal rather than a tight measurement.
|
|
func TestBandwidthCapIsEnforced(t *testing.T) {
|
|
const psk = "e2e-bwcap"
|
|
port := freePort(t)
|
|
hubAddr := fmt.Sprintf("127.0.0.1:%d", port)
|
|
startHub(t, port, psk)
|
|
dest := newMockDest(t, modeEcho)
|
|
startClientWithBandwidth(t, hubAddr, psk, 1, "1MB/s", []client.Mapping{
|
|
{Pattern: "mc.local", Destination: dest.addr},
|
|
})
|
|
|
|
pc := dialPlayer(t, hubAddr, "mc.local")
|
|
defer pc.Close()
|
|
|
|
payload := make([]byte, 2*1024*1024)
|
|
for i := range payload {
|
|
payload[i] = byte(i*31 + 7)
|
|
}
|
|
|
|
start := time.Now()
|
|
writeErr := make(chan error, 1)
|
|
go func() {
|
|
_, err := pc.Write(payload)
|
|
writeErr <- err
|
|
}()
|
|
got := make([]byte, len(payload))
|
|
_ = pc.SetReadDeadline(time.Now().Add(60 * time.Second))
|
|
if _, err := io.ReadFull(pc, got); err != nil {
|
|
t.Fatalf("read echo: %v", err)
|
|
}
|
|
elapsed := time.Since(start)
|
|
if err := <-writeErr; err != nil {
|
|
t.Fatalf("write payload: %v", err)
|
|
}
|
|
|
|
// 2 MiB at 1 MiB/s, minus the 200 KiB the bucket has banked, is ~1.8 s.
|
|
const floor = 1200 * time.Millisecond
|
|
if elapsed < floor {
|
|
t.Errorf("2 MiB echoed back in %v under a 1MB/s cap; expected at least %v, so the cap is not taking effect", elapsed, floor)
|
|
}
|
|
if elapsed > 30*time.Second {
|
|
t.Errorf("transfer took %v, far beyond the ~1.8s the rate implies", elapsed)
|
|
}
|
|
}
|
|
|
|
// TestCappedBandwidthDoesNotStarveLightStreams is the point of the fair queue:
|
|
// with the link deliberately capped and three players saturating it, a fourth
|
|
// player exchanging small messages must keep round-tripping promptly. A plain
|
|
// FIFO token bucket would leave it queued behind the heavy players' backlog,
|
|
// which for a real Minecraft client means a keepalive timeout — the exact
|
|
// failure this feature exists to prevent.
|
|
func TestCappedBandwidthDoesNotStarveLightStreams(t *testing.T) {
|
|
const psk = "e2e-bwfair"
|
|
port := freePort(t)
|
|
hubAddr := fmt.Sprintf("127.0.0.1:%d", port)
|
|
startHub(t, port, psk)
|
|
dest := newMockDest(t, modeEcho)
|
|
// Each player has its own worker conn; the shaper is still the only thing
|
|
// splitting the shared uplink budget.
|
|
startClientWithBandwidth(t, hubAddr, psk, 4, "1MB/s", []client.Mapping{
|
|
{Pattern: "mc.local", Destination: dest.addr},
|
|
})
|
|
|
|
// Three greedy players: each writes megabytes and keeps draining the echo,
|
|
// so they compete for the cap for the whole test rather than parking on
|
|
// their own flow-control windows.
|
|
for i := 0; i < 3; i++ {
|
|
heavy := dialPlayer(t, hubAddr, "mc.local")
|
|
defer heavy.Close()
|
|
drain(heavy)
|
|
go func() { _, _ = heavy.Write(make([]byte, 8*1024*1024)) }()
|
|
}
|
|
time.Sleep(500 * time.Millisecond) // let them saturate the shaper
|
|
|
|
light := dialPlayer(t, hubAddr, "mc.local")
|
|
defer light.Close()
|
|
payload := make([]byte, 4*1024)
|
|
for i := range payload {
|
|
payload[i] = byte(i*13 + 5)
|
|
}
|
|
|
|
var worst time.Duration
|
|
for i := 0; i < 10; i++ {
|
|
start := time.Now()
|
|
playerEcho(t, light, payload)
|
|
if d := time.Since(start); d > worst {
|
|
worst = d
|
|
}
|
|
}
|
|
|
|
// Fair sharing puts a 4 KiB round at roughly 4 KiB / (1 MiB/s ÷ 4) ≈ 16 ms of
|
|
// link time. The bound is two orders of magnitude looser so only genuine
|
|
// starvation trips it.
|
|
const limit = 3 * time.Second
|
|
if worst > limit {
|
|
t.Errorf("slowest small round-trip took %v (limit %v); heavy streams are crowding out the light one", worst, limit)
|
|
}
|
|
}
|