Preserve reliable server receive lanes

Author: bitprecipice

connect/receive_queue_policy_test.go

@@ -15,66 +15,277 @@ import (
 	"github.com/urnetwork/server"
 )
 
-func TestTrySendPooledReceiveRefusesFullQueueWithoutWaiting(t *testing.T) {
+func TestSendPooledReceiveUnreliableRefusesWithOwnership(t *testing.T) {
 	destination := make(chan []byte, 1)
-	queued := clientconnect.MessagePoolGet(1)
-	destination <- queued
+	destination <- clientconnect.MessagePoolGet(1)
 	defer func() { clientconnect.MessagePoolReturn(<-destination) }()
+	message := clientconnect.MessagePoolGet(173)
+	witness := clientconnect.MessagePoolShareReadOnly(message)
+	result := sendPooledReceive(
+		make(chan struct{}),
+		nil,
+		destination,
+		message,
+		clientconnect.CarrierReliabilityUnreliable,
+	)
+	if result != pooledMessageSendDropped {
+		t.Fatalf("full datagram queue result=%d, want dropped", result)
+	}
+	if !clientconnect.MessagePoolReturn(witness) {
+		t.Fatal("datagram refusal retained pooled bytes")
+	}
+}
 
-	result := pooledMessageSendDelivered
+func TestSendPooledReceiveReliableWaitsThenDelivers(t *testing.T) {
+	destination := make(chan []byte, 1)
+	destination <- clientconnect.MessagePoolGet(1)
+	message := clientconnect.MessagePoolGet(181)
+	witness := clientconnect.MessagePoolShareReadOnly(message)
+	result := pooledMessageSendDropped
 	done := make(chan struct{})
+	waiting := make(chan struct{})
 	go func() {
 		defer close(done)
-		result = trySendPooledReceive(
+		result = sendPooledReceive(
 			make(chan struct{}),
 			nil,
 			destination,
-			clientconnect.MessagePoolGet(173),
+			message,
+			clientconnect.CarrierReliabilityReliable,
+			func() { close(waiting) },
 		)
 	}()
+	<-waiting
+	select {
+	case <-done:
+		t.Fatal("reliable handoff returned while its queue was full")
+	default:
+	}
+	clientconnect.MessagePoolReturn(<-destination)
 	select {
 	case <-done:
 	case <-time.After(time.Second):
-		t.Fatal("receive-side queue offer waited for capacity")
+		t.Fatal("reliable handoff did not resume when capacity opened")
 	}
-	if result != pooledMessageSendDropped {
-		t.Fatalf("full receive queue result=%d, want dropped", result)
+	if result != pooledMessageSendDelivered {
+		t.Fatalf("resumed reliable handoff result=%d, want delivered", result)
+	}
+	clientconnect.MessagePoolReturn(<-destination)
+	if !clientconnect.MessagePoolReturn(witness) {
+		t.Fatal("delivered reliable message has an unexpected pooled owner")
 	}
 }
 
-func TestResidentTransportTrySendMessageRefusesWithoutWaiting(t *testing.T) {
+func TestSendPooledReceiveReliableCancellationReturnsOwnership(t *testing.T) {
+	destination := make(chan []byte, 1)
+	destination <- clientconnect.MessagePoolGet(1)
+	defer func() { clientconnect.MessagePoolReturn(<-destination) }()
 	ctxDone := make(chan struct{})
+	message := clientconnect.MessagePoolGet(191)
+	witness := clientconnect.MessagePoolShareReadOnly(message)
+	result := pooledMessageSendDelivered
+	done := make(chan struct{})
+	waiting := make(chan struct{})
+	go func() {
+		defer close(done)
+		result = sendPooledReceive(
+			ctxDone,
+			nil,
+			destination,
+			message,
+			clientconnect.CarrierReliabilityReliable,
+			func() { close(waiting) },
+		)
+	}()
+	<-waiting
+	select {
+	case <-done:
+		t.Fatal("reliable handoff returned before cancellation")
+	default:
+	}
+	close(ctxDone)
+	select {
+	case <-done:
+	case <-time.After(time.Second):
+		t.Fatal("reliable handoff ignored cancellation")
+	}
+	if result != pooledMessageSendDone {
+		t.Fatalf("canceled reliable handoff result=%d, want done", result)
+	}
+	if !clientconnect.MessagePoolReturn(witness) {
+		t.Fatal("canceled reliable handoff retained pooled bytes")
+	}
+}
+
+func benchmarkSendPooledReceiveReady(
+	benchmark *testing.B,
+	reliability clientconnect.CarrierReliability,
+) {
+	destination := make(chan []byte, 1)
+	message := []byte{1}
+	benchmark.ReportAllocs()
+	benchmark.ResetTimer()
+	for range benchmark.N {
+		result := sendPooledReceive(nil, nil, destination, message, reliability)
+		if result != pooledMessageSendDelivered {
+			benchmark.Fatalf("ready handoff result=%d, want delivered", result)
+		}
+		<-destination
+	}
+}
+
+func BenchmarkSendPooledReceiveReadyReliable(benchmark *testing.B) {
+	benchmarkSendPooledReceiveReady(benchmark, clientconnect.CarrierReliabilityReliable)
+}
+
+func BenchmarkSendPooledReceiveReadyUnreliable(benchmark *testing.B) {
+	benchmarkSendPooledReceiveReady(benchmark, clientconnect.CarrierReliabilityUnreliable)
+}
+
+func benchmarkResidentTransportSendReceivedReady(
+	benchmark *testing.B,
+	reliability clientconnect.CarrierReliability,
+) {
+	destination := make(chan []byte, 1)
+	transport := &ResidentTransport{ctx: context.Background(), send: destination}
+	message := []byte{1}
+	benchmark.ReportAllocs()
+	benchmark.ResetTimer()
+	for range benchmark.N {
+		result := transport.sendReceivedMessage(nil, message, reliability)
+		if result != pooledMessageSendDelivered {
+			benchmark.Fatalf("ready resident handoff result=%d, want delivered", result)
+		}
+		<-destination
+	}
+}
+
+func BenchmarkResidentTransportSendReceivedReadyReliable(benchmark *testing.B) {
+	benchmarkResidentTransportSendReceivedReady(
+		benchmark,
+		clientconnect.CarrierReliabilityReliable,
+	)
+}
+
+func BenchmarkResidentTransportSendReceivedReadyUnreliable(benchmark *testing.B) {
+	benchmarkResidentTransportSendReceivedReady(
+		benchmark,
+		clientconnect.CarrierReliabilityUnreliable,
+	)
+}
+
+// Reproduces the internal-hop root cause with explicit sequence markers. A
+// full reliable exchange queue must preserve 0,1 order instead of dropping 1
+// and letting a later frame pin Transfer recovery behind an artificial gap.
+func TestReliableExchangeQueueSaturationPreservesFramedOrder(t *testing.T) {
+	destination := make(chan []byte, 1)
+	first := clientconnect.MessagePoolGet(8)
+	first[0] = 0
+	second := clientconnect.MessagePoolGet(8)
+	second[0] = 1
+	destination <- first
+	result := pooledMessageSendDropped
+	done := make(chan struct{})
+	waiting := make(chan struct{})
+	go func() {
+		defer close(done)
+		result = sendPooledReceive(
+			make(chan struct{}),
+			nil,
+			destination,
+			second,
+			clientconnect.CarrierReliabilityReliable,
+			func() { close(waiting) },
+		)
+	}()
+	<-waiting
+	select {
+	case <-done:
+		t.Fatal("second reliable frame bypassed full queue")
+	default:
+	}
+	got := <-destination
+	if got[0] != 0 {
+		t.Fatalf("first framed marker=%d want=0", got[0])
+	}
+	clientconnect.MessagePoolReturn(got)
+	select {
+	case <-done:
+	case <-time.After(time.Second):
+		t.Fatal("second reliable frame did not resume")
+	}
+	if result != pooledMessageSendDelivered {
+		t.Fatalf("second framed result=%d want delivered", result)
+	}
+	got = <-destination
+	if got[0] != 1 {
+		t.Fatalf("second framed marker=%d want=1", got[0])
+	}
+	clientconnect.MessagePoolReturn(got)
+}
+
+func TestExchangeGenerationRetiresAfterAnyUndeliveredFrame(t *testing.T) {
+	if !pooledMessageSendKeepsGeneration(pooledMessageSendDelivered) {
+		t.Fatal("delivered exchange frame retired its generation")
+	}
+	for _, result := range []pooledMessageSendResult{
+		pooledMessageSendDropped,
+		pooledMessageSendDone,
+	} {
+		if pooledMessageSendKeepsGeneration(result) {
+			t.Fatalf("undelivered exchange result=%d kept a gapped generation", result)
+		}
+	}
+}
+
+func TestResidentTransportReceiveAdmissionUsesExactLane(t *testing.T) {
 	ctx, cancel := context.WithCancel(context.Background())
-	defer cancel()
 	destination := make(chan []byte, 1)
-	queued := clientconnect.MessagePoolGet(1)
-	destination <- queued
+	destination <- clientconnect.MessagePoolGet(1)
 	defer func() { clientconnect.MessagePoolReturn(<-destination) }()
+	transport := &ResidentTransport{ctx: ctx, send: destination}
+	waiting := make(chan struct{})
+	transport.beforeReliableReceiveWaitForTest = func() { close(waiting) }
 
-	transport := &ResidentTransport{
-		ctx:  ctx,
-		send: destination,
+	unreliable := clientconnect.MessagePoolGet(197)
+	if result := transport.sendReceivedMessage(
+		make(chan struct{}),
+		unreliable,
+		clientconnect.CarrierReliabilityUnreliable,
+	); result != pooledMessageSendDropped {
+		t.Fatalf("full resident DATAGRAM lane result=%d, want dropped", result)
 	}
+
+	reliable := clientconnect.MessagePoolGet(199)
+	witness := clientconnect.MessagePoolShareReadOnly(reliable)
 	result := pooledMessageSendDelivered
 	done := make(chan struct{})
 	go func() {
 		defer close(done)
-		result = transport.trySendMessage(
-			ctxDone,
-			clientconnect.MessagePoolGet(197),
+		result = transport.sendReceivedMessage(
+			make(chan struct{}),
+			reliable,
+			clientconnect.CarrierReliabilityReliable,
 		)
 	}()
+	<-waiting
+	select {
+	case <-done:
+		t.Fatal("resident reliable stream did not backpressure")
+	default:
+	}
+	cancel()
 	select {
 	case <-done:
 	case <-time.After(time.Second):
-		t.Fatal("resident transport receive handoff waited for capacity")
+		t.Fatal("resident reliable stream ignored generation cancellation")
 	}
-	if result != pooledMessageSendDropped {
-		t.Fatalf("full resident transport result=%d, want dropped", result)
+	if result != pooledMessageSendDone || !clientconnect.MessagePoolReturn(witness) {
+		t.Fatal("resident cancellation did not return reliable message ownership")
 	}
 }
 
-func TestProductionSocketReadersUseZeroWaitQueueAdmission(t *testing.T) {
+func TestProductionSocketReadersDeclareExactReceiveLanes(t *testing.T) {
 	checks := []struct {
 		path              string
 		required          map[string]int
@@ -83,21 +294,22 @@ func TestProductionSocketReadersUseZeroWaitQueueAdmission(t *testing.T) {
 		{
 			path: "transport.go",
 			required: map[string]int{
-				"residentTransport.trySendMessage(": 2,
+				"residentTransport.sendReceivedMessage(": 2,
+				"connect.CarrierReliabilityUnreliable":   1,
 			},
 			forbiddenSnippets: []string{
-				"residentTransport.sendMessage(",
+				"residentTransport.trySendMessage(",
 			},
 		},
 		{
 			path: "resident.go",
 			required: map[string]int{
-				"trySendPooledReceive(": 6, // declaration plus five receive boundaries
+				"sendPooledReceive(":                6, // declaration plus five receive boundaries
+				"pooledMessageSendKeepsGeneration(": 3, // declaration plus transport/forward writers
+				"ReceiveReliability:":               1, // resident's framed TCP receive route
 			},
 			forbiddenSnippets: []string{
-				"case receive <- message:",
-				"case forward <- message:",
-				"case self.receive <- message:",
+				"trySendPooledReceive(",
 			},
 		},
 	}
@@ -120,7 +332,7 @@ func TestProductionSocketReadersUseZeroWaitQueueAdmission(t *testing.T) {
 	}
 }
 
-func TestResidentForwardCallbackRefusesFullIngressWithoutWaiting(t *testing.T) {
+func TestResidentForwardCallbackRetiresFullIngressWithoutWaiting(t *testing.T) {
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
 	clientId := server.NewId()
@@ -156,6 +368,11 @@ func TestResidentForwardCallbackRefusesFullIngressWithoutWaiting(t *testing.T) {
 	case <-time.After(time.Second):
 		t.Fatal("resident forward callback waited for ingress capacity")
 	}
+	select {
+	case <-ctx.Done():
+	default:
+		t.Fatal("reliable forward refusal did not retire the resident generation")
+	}
 	clientconnect.MessagePoolReturn(message)
 	if !clientconnect.MessagePoolReturn(witness) {
 		t.Fatal("resident forward refusal retained callback-owned bytes")
@@ -273,6 +490,7 @@ func TestResidentClientCallbacksContainOnlyZeroWaitIngressWork(t *testing.T) {
 				"MessagePoolShareReadOnly(transferFrameBytes)",
 				"case self.forwardIngress[shardIndex] <- message:",
 				"default:",
+				"self.cancel()",
 			},
 			forbidden: []string{
 				"stateLock.",