test(dgram): use state-bearing flood outputs

commit: 5f6184712a112f99dd63a644e0695759c1ab6c58
author: brice · 3mo ago · 2026-07-10 05:56 UTC
parent: be2c9fb
signature: GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email

1 file · +28 -4

M internal/adapters/dgram/network_sim_test.go +28 -4


@@ -428,7 +428,7 @@ func TestTransportFloodClassification(t *testing.T) {
428	428		}
429	429		a.mu.Unlock()
430	430		go func() {
431		-	freshReturned <- a.Send(ports.Frame{Type: ports.MsgOutput, Payload: floodPayload(128)})
	431	+	freshReturned <- a.Send(ports.Frame{Type: ports.MsgOutput, Payload: floodOutputPayload(floodRecordCount, 128)})
432	432		}()
433	433		awaitSignal(t, freshAtPace, "fresh output pacing barrier")
434	434		a.mu.Lock()

@@ -479,7 +479,31 @@ func floodState(sender, receiver Transport, link simulatedLink) floodTransport
479	479		return floodTransportState{packetAge: packetAge, recordAge: recordAge, ackAge: ackAge, queuedBytes: queuedBytes, retransmits: retransmits}
480	480		}
481	481
482		-	func floodPayload(mtu int) []byte { return make([]byte, floodRecordMTUs*mtu) }
	482	+	func floodOutputPayload(baseState, mtu int) []byte {
	483	+	return ports.MarshalOutput(ports.Output{
	484	+	BaseStateNum: uint64(baseState),
	485	+	NewStateNum: uint64(baseState + 1),
	486	+	Data: make([]byte, floodRecordMTUs*mtu),
	487	+	})
	488	+	}
	489	+
	490	+	func TestFloodOutputPayloadIsIncrementalStateBearingOutput(t *testing.T) {
	491	+	const mtu = 128
	492	+	for state := range floodRecordCount {
	493	+	payload := floodOutputPayload(state, mtu)
	494	+	output, err := ports.UnmarshalOutput(payload)
	495	+	if err != nil {
	496	+	t.Fatalf("state %d: %v", state, err)
	497	+	}
	498	+	if output.BaseStateNum != uint64(state) \|\| output.NewStateNum != uint64(state+1) {
	499	+	t.Fatalf("output state = %d -> %d, want %d -> %d", output.BaseStateNum, output.NewStateNum, state, state+1)
	500	+	}
	501	+	minimumBytes := floodRecordMTUs * mtu
	502	+	if len(output.Data) < minimumBytes \|\| len(payload) < minimumBytes {
	503	+	t.Fatalf("state %d: data bytes=%d encoded bytes=%d, want each at least %d", state, len(output.Data), len(payload), minimumBytes)
	504	+	}
	505	+	}
	506	+	}
483	507
484	508		func sendFloodOutputs(t testing.T, sender Transport, count, mtu int) {
485	509		t.Helper()

@@ -487,11 +511,11 @@ func sendFloodOutputs(t testing.T, sender Transport, count, mtu int) {
487	511		sender.dataPaceRemaining = count * floodRecordMTUs * 4
488	512		sender.dataPaceNext = sender.clock.Now().Add(time.Hour)
489	513		sender.dataPaceMu.Unlock()
490		-	for range count {
	514	+	for state := range count {
491	515		sender.mu.Lock()
492	516		sender.outputNext = time.Time{}
493	517		sender.mu.Unlock()
494		-	if err := sender.Send(ports.Frame{Type: ports.MsgOutput, Payload: floodPayload(mtu)}); err != nil {
	518	+	if err := sender.Send(ports.Frame{Type: ports.MsgOutput, Payload: floodOutputPayload(state, mtu)}); err != nil {
495	519		t.Fatal(err)
496	520		}
497	521		}

 		}
 		a.mu.Unlock()
 		go func() {
-			freshReturned <- a.Send(ports.Frame{Type: ports.MsgOutput, Payload: floodPayload(128)})
+			freshReturned <- a.Send(ports.Frame{Type: ports.MsgOutput, Payload: floodOutputPayload(floodRecordCount, 128)})
 		}()
 		awaitSignal(t, freshAtPace, "fresh output pacing barrier")
 		a.mu.Lock()
 	return floodTransportState{packetAge: packetAge, recordAge: recordAge, ackAge: ackAge, queuedBytes: queuedBytes, retransmits: retransmits}
 }
-func floodPayload(mtu int) []byte { return make([]byte, floodRecordMTUs*mtu) }
+func floodOutputPayload(baseState, mtu int) []byte {
+	return ports.MarshalOutput(ports.Output{
+		BaseStateNum: uint64(baseState),
+		NewStateNum:  uint64(baseState + 1),
+		Data:         make([]byte, floodRecordMTUs*mtu),
+	})
+}
+func TestFloodOutputPayloadIsIncrementalStateBearingOutput(t *testing.T) {
+	const mtu = 128
+	for state := range floodRecordCount {
+		payload := floodOutputPayload(state, mtu)
+		output, err := ports.UnmarshalOutput(payload)
+		if err != nil {
+			t.Fatalf("state %d: %v", state, err)
+		}
+		if output.BaseStateNum != uint64(state) || output.NewStateNum != uint64(state+1) {
+			t.Fatalf("output state = %d -> %d, want %d -> %d", output.BaseStateNum, output.NewStateNum, state, state+1)
+		}
+		minimumBytes := floodRecordMTUs * mtu
+		if len(output.Data) < minimumBytes || len(payload) < minimumBytes {
+			t.Fatalf("state %d: data bytes=%d encoded bytes=%d, want each at least %d", state, len(output.Data), len(payload), minimumBytes)
+		}
+	}
+}
 func sendFloodOutputs(t *testing.T, sender *Transport, count, mtu int) {
 	t.Helper()
 	sender.dataPaceRemaining = count * floodRecordMTUs * 4
 	sender.dataPaceNext = sender.clock.Now().Add(time.Hour)
 	sender.dataPaceMu.Unlock()
-	for range count {
+	for state := range count {
 		sender.mu.Lock()
 		sender.outputNext = time.Time{}
 		sender.mu.Unlock()
-		if err := sender.Send(ports.Frame{Type: ports.MsgOutput, Payload: floodPayload(mtu)}); err != nil {
+		if err := sender.Send(ports.Frame{Type: ports.MsgOutput, Payload: floodOutputPayload(state, mtu)}); err != nil {
 			t.Fatal(err)
 		}
 	}