Changing ping system.

2026-05-19 08:04:38 +03:00 · 2026-04-21 11:16:00 +03:30
parent 236ae711c3
commit 26e51fc031
10 changed files with 704 additions and 137 deletions
@@ -9,6 +9,27 @@ import (
 	"masterhttprelayvpn/internal/protocol"
 )

+func testClientConfig() config.Config {
+	return config.Config{
+		MaxChunkSize:               1024,
+		MaxPacketsPerBatch:         4,
+		MaxBatchBytes:              4096,
+		WorkerCount:                2,
+		MaxConcurrentBatches:       2,
+		MaxPacketsPerSOCKSPerBatch: 1,
+		MuxRotateEveryBatches:      1,
+		MuxBurstThresholdBytes:     1024,
+		WorkerPollIntervalMS:       200,
+		IdlePollIntervalMS:         1000,
+		PingWarmThresholdMS:        5000,
+		PingBackoffBaseMS:          5000,
+		PingBackoffStepMS:          5000,
+		PingMaxIntervalMS:          60000,
+		MaxQueueBytesPerSOCKS:      4096,
+		HTTPBatchRandomize:         false,
+	}
+}
+
 func TestSOCKSConnectionStoreDeleteClearsTransportState(t *testing.T) {
 	store := NewSOCKSConnectionStore()
 	chunkPolicy := ChunkPolicy{
@@ -71,7 +92,7 @@ func TestSOCKSConnectionStoreDeleteClearsTransportState(t *testing.T) {
 func TestSOCKSConnectionInboundReorderQueuesAndDrainsInOrder(t *testing.T) {
 	socksConn := &SOCKSConnection{
 		ConnectAccepted: true,
-		PendingInbound:  make(map[uint64]PendingInboundPacket),
+		PendingInbound:  make(map[uint64][]PendingInboundPacket),
 	}

 	packet2 := protocol.NewPacket("client-session", protocol.PacketTypeSOCKSData)
@@ -106,12 +127,12 @@ func TestSOCKSConnectionInboundReorderQueuesAndDrainsInOrder(t *testing.T) {

 func TestSOCKSConnectionInboundGapTimeout(t *testing.T) {
 	socksConn := &SOCKSConnection{
-		PendingInbound: make(map[uint64]PendingInboundPacket),
+		PendingInbound: make(map[uint64][]PendingInboundPacket),
 	}
-	socksConn.PendingInbound[5] = PendingInboundPacket{
+	socksConn.PendingInbound[5] = []PendingInboundPacket{{
 		Packet:   protocol.Packet{Sequence: 5},
 		QueuedAt: time.Now().Add(-2 * time.Second),
-	}
+	}}

 	if !socksConn.hasExpiredInboundGap(500 * time.Millisecond) {
 		t.Fatal("expected inbound gap timeout to trigger")
@@ -123,7 +144,7 @@ func TestSOCKSConnectionInboundGapTimeout(t *testing.T) {

 func TestSOCKSConnectionInboundDataWaitsForConnectAck(t *testing.T) {
 	socksConn := &SOCKSConnection{
-		PendingInbound: make(map[uint64]PendingInboundPacket),
+		PendingInbound: make(map[uint64][]PendingInboundPacket),
 	}

 	packet1 := protocol.NewPacket("client-session", protocol.PacketTypeSOCKSData)
@@ -150,18 +171,10 @@ func TestSOCKSConnectionInboundDataWaitsForConnectAck(t *testing.T) {
 }

 func TestBuildNextBatchRotatesAcrossConnections(t *testing.T) {
-	cfg := config.Config{
-		MaxChunkSize:               1024,
-		MaxPacketsPerBatch:         1,
-		MaxBatchBytes:              4096,
-		WorkerCount:                1,
-		MaxConcurrentBatches:       1,
-		MaxPacketsPerSOCKSPerBatch: 1,
-		MuxRotateEveryBatches:      1,
-		MuxBurstThresholdBytes:     1024,
-		MaxQueueBytesPerSOCKS:      4096,
-		HTTPBatchRandomize:         false,
-	}
+	cfg := testClientConfig()
+	cfg.MaxPacketsPerBatch = 1
+	cfg.WorkerCount = 1
+	cfg.MaxConcurrentBatches = 1

 	client := New(cfg, nil)
 	client.chunkPolicy = newChunkPolicy(cfg)
@@ -195,18 +208,7 @@ func TestBuildNextBatchRotatesAcrossConnections(t *testing.T) {
 }

 func TestBuildNextBatchHonorsPerSOCKSPacketLimit(t *testing.T) {
-	cfg := config.Config{
-		MaxChunkSize:               1024,
-		MaxPacketsPerBatch:         4,
-		MaxBatchBytes:              4096,
-		WorkerCount:                2,
-		MaxConcurrentBatches:       2,
-		MaxPacketsPerSOCKSPerBatch: 1,
-		MuxRotateEveryBatches:      1,
-		MuxBurstThresholdBytes:     1024,
-		MaxQueueBytesPerSOCKS:      4096,
-		HTTPBatchRandomize:         false,
-	}
+	cfg := testClientConfig()

 	client := New(cfg, nil)
 	client.chunkPolicy = newChunkPolicy(cfg)
@@ -242,13 +244,10 @@ func TestBuildNextBatchHonorsPerSOCKSPacketLimit(t *testing.T) {
 }

 func TestEffectiveConcurrentBatchesUsesBurstThreshold(t *testing.T) {
-	cfg := config.Config{
-		WorkerCount:                4,
-		MaxConcurrentBatches:       3,
-		MaxPacketsPerSOCKSPerBatch: 2,
-		MuxRotateEveryBatches:      1,
-		MuxBurstThresholdBytes:     4096,
-	}
+	cfg := testClientConfig()
+	cfg.WorkerCount = 4
+	cfg.MaxConcurrentBatches = 3
+	cfg.MuxBurstThresholdBytes = 4096

 	client := New(cfg, nil)
 	if got := client.effectiveConcurrentBatches(1024); got != 1 {
@@ -258,3 +257,232 @@ func TestEffectiveConcurrentBatchesUsesBurstThreshold(t *testing.T) {
 		t.Fatalf("expected burst concurrency of 3, got %d", got)
 	}
 }
+
+func TestBuildPollBatchSkipsWhenTransportBusy(t *testing.T) {
+	cfg := testClientConfig()
+	client := New(cfg, nil)
+	client.chunkPolicy = newChunkPolicy(cfg)
+
+	socksConn := client.socksConnections.New(client.clientSessionKey, "127.0.0.1:1001", client.chunkPolicy)
+	if err := socksConn.EnqueuePacket(socksConn.BuildSOCKSDataPacket([]byte("busy"), false)); err != nil {
+		t.Fatalf("enqueue packet: %v", err)
+	}
+
+	batch, ok := client.buildPollBatch(client.socksConnections.Snapshot(), queuedBytesAcross(client.socksConnections.Snapshot()))
+	if ok || len(batch.Packets) != 0 {
+		t.Fatal("expected poll batch to be suppressed while queued payload exists")
+	}
+}
+
+func TestBuildPollBatchAllowsOnlySinglePingInFlight(t *testing.T) {
+	cfg := testClientConfig()
+	client := New(cfg, nil)
+	client.chunkPolicy = newChunkPolicy(cfg)
+	client.socksConnections.New(client.clientSessionKey, "127.0.0.1:1001", client.chunkPolicy)
+	client.noteMeaningfulActivity(time.Now().Add(-10 * time.Second))
+
+	batch, ok := client.buildPollBatch(client.socksConnections.Snapshot(), 0)
+	if !ok || len(batch.Packets) != 1 || batch.Packets[0].Type != protocol.PacketTypePing {
+		t.Fatal("expected first idle batch to be a ping")
+	}
+
+	batch, ok = client.buildPollBatch(client.socksConnections.Snapshot(), 0)
+	if ok || len(batch.Packets) != 0 {
+		t.Fatal("expected second ping to be suppressed while first ping is still in flight")
+	}
+}
+
+func TestBuildPollBatchAllowsSessionPingWithoutActiveConnections(t *testing.T) {
+	cfg := testClientConfig()
+	client := New(cfg, nil)
+
+	now := time.Now()
+	client.noteMeaningfulActivity(now.Add(-10 * time.Second))
+	client.nextPingDueUnixMS.Store(now.Add(-1 * time.Second).UnixMilli())
+
+	batch, ok := client.buildPollBatch(nil, 0)
+	if !ok || len(batch.Packets) != 1 || batch.Packets[0].Type != protocol.PacketTypePing {
+		t.Fatal("expected session-level ping even without active socks connections")
+	}
+}
+
+func TestBuildPollBatchSkipsWithoutSessionActivity(t *testing.T) {
+	cfg := testClientConfig()
+	client := New(cfg, nil)
+	client.nextPingDueUnixMS.Store(time.Now().Add(-1 * time.Second).UnixMilli())
+
+	batch, ok := client.buildPollBatch(nil, 0)
+	if ok || len(batch.Packets) != 0 {
+		t.Fatal("expected ping to stay suppressed before the session has any real activity")
+	}
+}
+
+func TestShouldSendPingWhenIdleIntervalHasElapsed(t *testing.T) {
+	cfg := testClientConfig()
+	client := New(cfg, nil)
+	client.chunkPolicy = newChunkPolicy(cfg)
+	client.socksConnections.New(client.clientSessionKey, "127.0.0.1:1001", client.chunkPolicy)
+
+	now := time.Now()
+	client.nextPingDueUnixMS.Store(now.Add(-2 * time.Second).UnixMilli())
+	if !client.shouldSendPing(client.socksConnections.Snapshot(), 0, now) {
+		t.Fatal("expected ping to be due after idle interval elapsed")
+	}
+}
+
+func TestShouldNotSendPingBeforeIdleInterval(t *testing.T) {
+	cfg := testClientConfig()
+	client := New(cfg, nil)
+	client.chunkPolicy = newChunkPolicy(cfg)
+	client.socksConnections.New(client.clientSessionKey, "127.0.0.1:1001", client.chunkPolicy)
+
+	now := time.Now()
+	client.nextPingDueUnixMS.Store(now.Add(500 * time.Millisecond).UnixMilli())
+	if client.shouldSendPing(client.socksConnections.Snapshot(), 0, now) {
+		t.Fatal("expected ping to stay suppressed until idle interval elapses")
+	}
+}
+
+func TestShouldSendPingWithOnlyInFlightPackets(t *testing.T) {
+	cfg := testClientConfig()
+	client := New(cfg, nil)
+	client.chunkPolicy = newChunkPolicy(cfg)
+	socksConn := client.socksConnections.New(client.clientSessionKey, "127.0.0.1:1001", client.chunkPolicy)
+
+	packet := socksConn.BuildSOCKSDataPacket([]byte("hello"), false)
+	item := &SOCKSOutboundQueueItem{
+		IdentityKey: protocol.PacketIdentityKey(
+			packet.ClientSessionKey,
+			packet.SOCKSID,
+			packet.Type,
+			packet.Sequence,
+			packet.FragmentID,
+		),
+		Packet:      packet,
+		QueuedAt:    time.Now(),
+		SentAt:      time.Now(),
+		PayloadSize: len(packet.Payload),
+	}
+	socksConn.MarkInFlight([]*SOCKSOutboundQueueItem{item})
+
+	now := time.Now()
+	client.noteMeaningfulActivity(now.Add(-10 * time.Second))
+	client.nextPingDueUnixMS.Store(now.Add(-1 * time.Second).UnixMilli())
+
+	if !client.shouldSendPing(client.socksConnections.Snapshot(), 0, now) {
+		t.Fatal("expected ping to be allowed while only in-flight packets remain")
+	}
+}
+
+func TestIdleIntervalForStreakBacksOffWithIdlePongs(t *testing.T) {
+	cfg := testClientConfig()
+	client := New(cfg, nil)
+
+	if got := client.idleIntervalForStreak(0); got != 5*time.Second {
+		t.Fatalf("expected base backoff interval, got %v", got)
+	}
+
+	if got := client.idleIntervalForStreak(1); got != 10*time.Second {
+		t.Fatalf("expected first stepped backoff interval, got %v", got)
+	}
+
+	if got := client.idleIntervalForStreak(20); got != 60*time.Second {
+		t.Fatalf("expected capped backoff interval, got %v", got)
+	}
+}
+
+func TestCompletePingWithPongIncrementsStreakOnlyWithoutRealTraffic(t *testing.T) {
+	cfg := testClientConfig()
+	client := New(cfg, nil)
+	now := time.Now()
+
+	client.noteMeaningfulActivity(now.Add(-10 * time.Second))
+	if !client.tryBeginPing(now) {
+		t.Fatal("expected ping to start")
+	}
+	client.completePingWithPong()
+	if got := client.idlePongStreak.Load(); got != 1 {
+		t.Fatalf("expected pong streak to increment to 1, got %d", got)
+	}
+	nextDue := client.nextPingDueUnixMS.Load()
+	if nextDue <= now.UnixMilli() {
+		t.Fatal("expected next ping due to be scheduled in the future after idle pong")
+	}
+
+	client.noteMeaningfulActivity(now.Add(1 * time.Second))
+	if !client.tryBeginPing(now.Add(2 * time.Second)) {
+		t.Fatal("expected second ping to start")
+	}
+	client.noteMeaningfulActivity(now.Add(3 * time.Second))
+	client.completePingWithPong()
+	if got := client.idlePongStreak.Load(); got != 0 {
+		t.Fatalf("expected pong streak reset after real traffic, got %d", got)
+	}
+	if client.nextPingDueUnixMS.Load() <= now.UnixMilli() {
+		t.Fatal("expected next ping due to be rescheduled after meaningful traffic")
+	}
+}
+
+func TestCompletePingWithPongStaysAggressiveBeforeWarmThreshold(t *testing.T) {
+	cfg := testClientConfig()
+	client := New(cfg, nil)
+	now := time.Now()
+
+	client.noteMeaningfulActivity(now.Add(-3 * time.Second))
+	if !client.tryBeginPing(now) {
+		t.Fatal("expected ping to start")
+	}
+
+	client.completePingWithPong()
+
+	if got := client.idlePongStreak.Load(); got != 0 {
+		t.Fatalf("expected pong streak to stay at 0 before warm threshold, got %d", got)
+	}
+
+	nextDue := client.nextPingDueUnixMS.Load()
+	expectedMin := now.Add(900 * time.Millisecond).UnixMilli()
+	expectedMax := now.Add(1100 * time.Millisecond).UnixMilli()
+	if nextDue < expectedMin || nextDue > expectedMax {
+		t.Fatalf("expected aggressive next ping around idle interval, got %d", nextDue)
+	}
+}
+
+func TestInboundReorderAllowsCloseReadAndCloseWriteOnSameSequence(t *testing.T) {
+	cfg := testClientConfig()
+	client := New(cfg, nil)
+	client.chunkPolicy = newChunkPolicy(cfg)
+
+	socksConn := client.socksConnections.New(client.clientSessionKey, "127.0.0.1:1001", client.chunkPolicy)
+	socksConn.ConnectAccepted = true
+
+	closeWrite := protocol.NewPacket(client.clientSessionKey, protocol.PacketTypeSOCKSCloseWrite)
+	closeWrite.SOCKSID = socksConn.ID
+	closeWrite.Sequence = 2
+
+	closeRead := protocol.NewPacket(client.clientSessionKey, protocol.PacketTypeSOCKSCloseRead)
+	closeRead.SOCKSID = socksConn.ID
+	closeRead.Sequence = 2
+
+	ready, duplicate, overflow := socksConn.queueInboundPacket(closeWrite, 8)
+	if duplicate || overflow || len(ready) != 0 {
+		t.Fatalf("expected first close packet to buffer, duplicate=%t overflow=%t ready=%d", duplicate, overflow, len(ready))
+	}
+
+	ready, duplicate, overflow = socksConn.queueInboundPacket(closeRead, 8)
+	if duplicate || overflow || len(ready) != 0 {
+		t.Fatalf("expected second close packet on same sequence to buffer, duplicate=%t overflow=%t ready=%d", duplicate, overflow, len(ready))
+	}
+
+	data := protocol.NewPacket(client.clientSessionKey, protocol.PacketTypeSOCKSData)
+	data.SOCKSID = socksConn.ID
+	data.Sequence = 1
+	data.Payload = []byte("ok")
+
+	ready, duplicate, overflow = socksConn.queueInboundPacket(data, 8)
+	if duplicate || overflow || len(ready) != 3 {
+		t.Fatalf("expected data and both close packets to drain, duplicate=%t overflow=%t ready=%d", duplicate, overflow, len(ready))
+	}
+	if ready[0].Type != protocol.PacketTypeSOCKSData || ready[1].Type != protocol.PacketTypeSOCKSCloseRead || ready[2].Type != protocol.PacketTypeSOCKSCloseWrite {
+		t.Fatalf("unexpected drain order: %s, %s, %s", ready[0].Type, ready[1].Type, ready[2].Type)
+	}
+}