From 5613ba1da3731dcc295116e19290e5d8714bc41b Mon Sep 17 00:00:00 2001 From: cppla Date: Fri, 2 Oct 2026 23:23:18 +0800 Subject: [PATCH] test: witness real QUIC receiver backpressure --- .../pacing_backpressure_witness_test.go | 299 ++++++++++++++++++ .../pacing_feedback_integration_test.go | 26 +- .../pacing_recovery_integration_test.go | 34 +- .../tunnel/pacing_recovery_witness_test.go | 179 +++++++++++ 4 files changed, 526 insertions(+), 12 deletions(-) create mode 100644 internal/tunnel/pacing_backpressure_witness_test.go create mode 100644 internal/tunnel/pacing_recovery_witness_test.go diff --git a/internal/tunnel/pacing_backpressure_witness_test.go b/internal/tunnel/pacing_backpressure_witness_test.go new file mode 100644 index 0000000..9a7ea7a --- /dev/null +++ b/internal/tunnel/pacing_backpressure_witness_test.go @@ -0,0 +1,299 @@ +package tunnel + +import ( + "sync" + "testing" + + quic "github.com/quic-go/quic-go" + "github.com/quic-go/quic-go/qlog" + "github.com/quic-go/quic-go/qlogwriter" +) + +// feedbackBlockedTrace observes peer-received STREAM_DATA_BLOCKED frames for +// one selected stream. It keeps only a copied flow-control high-water mark; +// the library can reuse packet/frame pointers as soon as RecordEvent returns. +// Notifications are coalesced hints: waiters must inspect highest themselves. +type feedbackBlockedTrace struct { + mu sync.Mutex + selected bool + streamID quic.StreamID + maximum int64 + changed <-chan struct{} + notify chan<- struct{} +} + +var _ qlogwriter.Trace = (*feedbackBlockedTrace)(nil) +var _ qlogwriter.Recorder = (*feedbackBlockedTrace)(nil) + +func newFeedbackBlockedTrace() *feedbackBlockedTrace { + changed := make(chan struct{}, 1) + return &feedbackBlockedTrace{changed: changed, notify: changed} +} + +// selectStream is called before warmup so warmup frames contribute to the +// baseline later sampled by the recovery test. Changing streams starts a new +// high-water mark; choosing the same stream does not discard its history. +func (t *feedbackBlockedTrace) selectStream(streamID quic.StreamID) { + t.mu.Lock() + defer t.mu.Unlock() + if !t.selected || t.streamID != streamID { + t.maximum = 0 + } + t.streamID = streamID + t.selected = true +} + +func (t *feedbackBlockedTrace) highest() int64 { + t.mu.Lock() + defer t.mu.Unlock() + return t.maximum +} + +func (t *feedbackBlockedTrace) SupportsSchemas(schema string) bool { + return schema == qlog.EventSchema +} + +// Producers share the same in-memory observation. There is no output resource +// to close, and one producer finishing must not stop another's notifications. +func (t *feedbackBlockedTrace) AddProducer() qlogwriter.Recorder { + return &feedbackBlockedRecorder{trace: t} +} + +type feedbackBlockedRecorder struct{ trace *feedbackBlockedTrace } + +func (r *feedbackBlockedRecorder) RecordEvent(event qlogwriter.Event) { + r.trace.RecordEvent(event) +} + +func (r *feedbackBlockedRecorder) Close() error { return nil } + +func (t *feedbackBlockedTrace) Close() error { return nil } + +func (t *feedbackBlockedTrace) RecordEvent(event qlogwriter.Event) { + var packet qlog.PacketReceived + switch event := event.(type) { + case qlog.PacketReceived: + packet = event + case *qlog.PacketReceived: + if event == nil { + return + } + packet = *event + default: + return + } + + t.mu.Lock() + defer t.mu.Unlock() + if !t.selected { + return + } + advanced := false + for _, frame := range packet.Frames { + blocked, ok := frame.Frame.(*qlog.StreamDataBlockedFrame) + if !ok || blocked == nil { + continue + } + streamID, maximum := quic.StreamID(blocked.StreamID), int64(blocked.MaximumStreamData) + if streamID != t.streamID || maximum <= t.maximum { + continue + } + t.maximum = maximum + advanced = true + } + if advanced { + select { + case t.notify <- struct{}{}: + default: + } + } +} + +func feedbackBlockedPacket(streamID quic.StreamID, maximum int64) qlog.PacketReceived { + frame := &qlog.StreamDataBlockedFrame{StreamID: streamID} + setFeedbackBlockedMaximum(&frame.MaximumStreamData, maximum) + return qlog.PacketReceived{Frames: []qlog.Frame{{Frame: frame}}} +} + +// qlog publicly aliases a frame whose ByteCount type lives in a library +// internal package. Infer that int64-derived type instead of importing it. +func setFeedbackBlockedMaximum[T ~int64](field *T, maximum int64) { + *field = T(maximum) +} + +func feedbackBlockedNotice(t *testing.T, trace *feedbackBlockedTrace, want bool) { + t.Helper() + got := false + select { + case _, open := <-trace.changed: + if !open { + t.Fatal("observer closed its shared notification channel") + } + got = true + default: + } + if got != want { + t.Fatalf("high-water notification = %t, want %t", got, want) + } +} + +func TestFeedbackBlockedTraceValueAndHighWatermark(t *testing.T) { + trace := newFeedbackBlockedTrace() + trace.RecordEvent(feedbackBlockedPacket(0, 4096)) + if got := trace.highest(); got != 0 { + t.Fatalf("unselected stream high-water mark = %d", got) + } + feedbackBlockedNotice(t, trace, false) + trace.selectStream(0) + for _, packet := range []qlog.PacketReceived{ + feedbackBlockedPacket(4, 1<<20), + feedbackBlockedPacket(0, 0), + feedbackBlockedPacket(0, -1), + } { + trace.RecordEvent(packet) + } + if got := trace.highest(); got != 0 { + t.Fatalf("irrelevant frames set high-water mark = %d", got) + } + feedbackBlockedNotice(t, trace, false) + + // The pinned library calls RecordEvent with a PacketReceived value, not + // a pointer. Warmup establishes a baseline that stale frames cannot beat. + trace.RecordEvent(feedbackBlockedPacket(0, 4096)) + if got := trace.highest(); got != 4096 { + t.Fatalf("warmup high-water mark = %d", got) + } + feedbackBlockedNotice(t, trace, true) + for _, maximum := range []int64{2048, 4096, 4096, 0, -1} { + trace.RecordEvent(feedbackBlockedPacket(0, maximum)) + } + if got := trace.highest(); got != 4096 { + t.Fatalf("old or duplicate frame changed baseline to %d", got) + } + feedbackBlockedNotice(t, trace, false) + trace.RecordEvent(feedbackBlockedPacket(0, 8192)) + trace.RecordEvent(feedbackBlockedPacket(0, 12288)) + if got := trace.highest(); got != 12288 { + t.Fatalf("new blocked evidence high-water mark = %d", got) + } + feedbackBlockedNotice(t, trace, true) + feedbackBlockedNotice(t, trace, false) +} + +func TestFeedbackBlockedTraceCopiesScalarsAndHandlesPointers(t *testing.T) { + trace := newFeedbackBlockedTrace() + trace.selectStream(4) + packet := feedbackBlockedPacket(4, 16384) + frame := packet.Frames[0].Frame.(*qlog.StreamDataBlockedFrame) + trace.RecordEvent(&packet) + feedbackBlockedNotice(t, trace, true) + frame.StreamID = 8 + frame.MaximumStreamData = 1 << 30 + packet.Frames[0].Frame = &qlog.StreamDataBlockedFrame{StreamID: 4, MaximumStreamData: 1 << 31} + packet.Frames = nil + if got := trace.highest(); got != 16384 { + t.Fatalf("reusing recorded packet/frame changed copied high-water mark to %d", got) + } + feedbackBlockedNotice(t, trace, false) + trace.RecordEvent((*qlog.PacketReceived)(nil)) + trace.RecordEvent(qlog.PacketReceived{Frames: []qlog.Frame{ + {}, + {Frame: (*qlog.StreamDataBlockedFrame)(nil)}, + {Frame: &qlog.PingFrame{}}, + }}) + trace.RecordEvent(qlog.PacketSent{Frames: []qlog.Frame{ + {Frame: &qlog.StreamDataBlockedFrame{StreamID: 4, MaximumStreamData: 1 << 32}}, + }}) + trace.RecordEvent(nil) + if got := trace.highest(); got != 16384 { + t.Fatalf("ignored event changed high-water mark to %d", got) + } + feedbackBlockedNotice(t, trace, false) +} + +func TestFeedbackBlockedTraceSelectionAndSchemas(t *testing.T) { + trace := newFeedbackBlockedTrace() + for _, schema := range []string{"", "urn:ietf:params:qlog:events:quic-11", "urn:ietf:params:qlog:events:http3-12"} { + if trace.SupportsSchemas(schema) { + t.Fatalf("observer advertised unsupported schema %q", schema) + } + } + if !trace.SupportsSchemas(qlog.EventSchema) { + t.Fatal("observer does not support the pinned QUIC event schema") + } + trace.selectStream(0) + trace.RecordEvent(feedbackBlockedPacket(0, 1024)) + feedbackBlockedNotice(t, trace, true) + trace.selectStream(0) + if got := trace.highest(); got != 1024 { + t.Fatalf("reselecting the same stream discarded warmup: %d", got) + } + trace.selectStream(4) + if got := trace.highest(); got != 0 { + t.Fatalf("another stream inherited previous high-water mark: %d", got) + } + trace.RecordEvent(feedbackBlockedPacket(0, 1<<20)) + feedbackBlockedNotice(t, trace, false) + trace.RecordEvent(feedbackBlockedPacket(4, 2048)) + if got := trace.highest(); got != 2048 { + t.Fatalf("newly selected stream high-water mark = %d", got) + } + feedbackBlockedNotice(t, trace, true) +} + +func TestFeedbackBlockedTraceConcurrentProducersAndClose(t *testing.T) { + trace := newFeedbackBlockedTrace() + trace.selectStream(0) + const producers, eventsPerProducer = 8, 64 + var workers sync.WaitGroup + start := make(chan struct{}) + workers.Add(producers) + for index := range producers { + go func() { + defer workers.Done() + <-start + recorder := trace.AddProducer() + for sequence := range eventsPerProducer { + recorder.RecordEvent(feedbackBlockedPacket(0, int64(index*eventsPerProducer+sequence+1))) + _ = trace.highest() + } + // Each recorder closes after its own calls finish; other producers + // may still record events, as the Trace contract explicitly allows. + _ = recorder.Close() + }() + } + close(start) + workers.Wait() + if got := trace.highest(); got != producers*eventsPerProducer { + t.Fatalf("concurrent high-water mark = %d", got) + } + feedbackBlockedNotice(t, trace, true) + feedbackBlockedNotice(t, trace, false) + if err := trace.Close(); err != nil { + t.Fatal(err) + } + feedbackBlockedNotice(t, trace, false) +} + +func TestFeedbackBlockedTraceProducerCloseKeepsSiblingActive(t *testing.T) { + trace := newFeedbackBlockedTrace() + trace.selectStream(0) + first, sibling := trace.AddProducer(), trace.AddProducer() + first.RecordEvent(feedbackBlockedPacket(0, 1024)) + if err := first.Close(); err != nil { + t.Fatal(err) + } + feedbackBlockedNotice(t, trace, true) + feedbackBlockedNotice(t, trace, false) + // The sibling is a distinct recorder and remains active after another + // producer's Close. Do not call RecordEvent on the closed first recorder. + sibling.RecordEvent(feedbackBlockedPacket(0, 2048)) + if got := trace.highest(); got != 2048 { + t.Fatalf("closing another producer stopped the sibling: high-water mark = %d", got) + } + feedbackBlockedNotice(t, trace, true) + if err := sibling.Close(); err != nil { + t.Fatal(err) + } + feedbackBlockedNotice(t, trace, false) +} diff --git a/internal/tunnel/pacing_feedback_integration_test.go b/internal/tunnel/pacing_feedback_integration_test.go index 31dde5f..c961cd8 100644 --- a/internal/tunnel/pacing_feedback_integration_test.go +++ b/internal/tunnel/pacing_feedback_integration_test.go @@ -12,6 +12,7 @@ import ( "github.com/cppla/autocar/internal/accel" quic "github.com/quic-go/quic-go" + "github.com/quic-go/quic-go/qlogwriter" ) const feedbackChunkSize = 32 << 10 @@ -134,6 +135,11 @@ type feedbackQUICFixture struct { } func newFeedbackQUICFixture(t *testing.T, streamCount int, receiveWindow uint64) *feedbackQUICFixture { + t.Helper() + return newFeedbackQUICFixtureWithTrace(t, streamCount, receiveWindow, nil) +} + +func newFeedbackQUICFixtureWithTrace(t *testing.T, streamCount int, receiveWindow uint64, trace qlogwriter.Trace) *feedbackQUICFixture { t.Helper() ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second) f := &feedbackQUICFixture{ctx: ctx, clock: &feedbackRealClock{}} @@ -169,6 +175,9 @@ func newFeedbackQUICFixture(t *testing.T, streamCount int, receiveWindow uint64) t.Fatal(err) } quicConfig := hardenedQUICClientConfig(nil) + if trace != nil { + quicConfig.Tracer = func(context.Context, bool, quic.ConnectionID) qlogwriter.Trace { return trace } + } if receiveWindow != 0 { quicConfig.InitialStreamReceiveWindow = receiveWindow quicConfig.MaxStreamReceiveWindow = receiveWindow @@ -295,19 +304,30 @@ func readFeedbackPayload(reader io.Reader, size int) error { type feedbackRawStream struct { quicStream - entered chan struct{} - pending atomic.Bool - writeTime atomic.Int64 + entered chan struct{} + pending atomic.Bool + writeTime atomic.Int64 + writeWitness *feedbackWriteWitness } func (s *feedbackRawStream) Write(p []byte) (int, error) { started := time.Now() s.pending.Store(true) + first := false + if witness := s.writeWitness; witness != nil { + witness.once.Do(func() { + first = true + close(witness.started) + }) + } select { case s.entered <- struct{}{}: default: } n, err := s.quicStream.Write(p) + if first { + close(s.writeWitness.done) + } s.pending.Store(false) s.writeTime.Add(int64(time.Since(started))) return n, err diff --git a/internal/tunnel/pacing_recovery_integration_test.go b/internal/tunnel/pacing_recovery_integration_test.go index de4d896..6215d89 100644 --- a/internal/tunnel/pacing_recovery_integration_test.go +++ b/internal/tunnel/pacing_recovery_integration_test.go @@ -9,12 +9,16 @@ import ( // changes from fast to slow and back. The small, fixed receive window makes // its backpressure observable instead of buffering the whole slow phase. func TestQUICPacingRecoversAfterReceiverBackpressure(t *testing.T) { - f := newFeedbackQUICFixture(t, 1, 16<<10) - fast := runPacingRecoveryPhase(t, f, "fast before", 1<<20, 0) - slow := runPacingRecoveryPhase(t, f, "receiver limited", 16*feedbackChunkSize, 120*time.Millisecond) - if slow.write <= slow.wait { - t.Fatalf("fixture did not establish transport backpressure: Write=%s, Wait=%s", slow.write, slow.wait) - } + trace := newFeedbackBlockedTrace() + f := newFeedbackQUICFixtureWithTrace(t, 1, 16<<10, trace) + trace.selectStream(f.clientStreams[0].StreamID()) + const warmupBytes = 1 << 20 + fast := runPacingRecoveryPhase(t, f, "fast before", warmupBytes, 0, nil) + // Arm only after the warmup writer has joined. A blocked offset beyond + // warmup excludes delayed/retransmitted old frames. Withhold the first + // read until the same first raw Write is still pending on real flow control. + gate := newFeedbackBackpressureGate(f.rawStreams[0], trace, warmupBytes) + slow := runPacingRecoveryPhase(t, f, "receiver limited", 16*feedbackChunkSize, 120*time.Millisecond, gate) // Even both worst-case balanced-profile penalties cannot reduce an // unchanged capacity target below 36%. Require an actual capacity decline, // not merely a transient RTT/loss penalty, before checking rediscovery. @@ -24,7 +28,7 @@ func TestQUICPacingRecoversAfterReceiverBackpressure(t *testing.T) { var recovered [4]pacingRecoveryPhase for index := range recovered { - recovered[index] = runPacingRecoveryPhase(t, f, "fast after", 512<<10, 0) + recovered[index] = runPacingRecoveryPhase(t, f, "fast after", 512<<10, 0, nil) } f.checkReverse(t) // The first two windows allow discovery. Require sustained useful progress @@ -53,12 +57,24 @@ type pacingRecoveryPhase struct { wait, write time.Duration } -func runPacingRecoveryPhase(t *testing.T, f *feedbackQUICFixture, name string, size int, readDelay time.Duration) pacingRecoveryPhase { +func runPacingRecoveryPhase(t *testing.T, f *feedbackQUICFixture, name string, size int, readDelay time.Duration, gate *feedbackBackpressureGate) pacingRecoveryPhase { t.Helper() started := time.Now() waitBefore := f.streamPacers[0].waitTime.Load() writeBefore := f.rawStreams[0].writeTime.Load() - writer := f.run(func() error { return writeFeedbackPayload(f.serverStreams[0], size) }) + finished := make(chan struct{}) + writer := f.run(func() error { + defer close(finished) + return writeFeedbackPayload(f.serverStreams[0], size) + }) + if gate != nil { + offset, err := gate.await(f.ctx, finished) + if err != nil { + t.Fatalf("%s: real receiver backpressure not established: %v", name, err) + } + t.Logf("%s: received STREAM_DATA_BLOCKED stream=%d offset=%d > warmup=%d; first raw Write still pending before first read", + name, f.clientStreams[0].StreamID(), offset, gate.warmupOffset) + } for read := 0; read < size; read += feedbackChunkSize { if readDelay > 0 { timer := time.NewTimer(readDelay) diff --git a/internal/tunnel/pacing_recovery_witness_test.go b/internal/tunnel/pacing_recovery_witness_test.go new file mode 100644 index 0000000..49f3a66 --- /dev/null +++ b/internal/tunnel/pacing_recovery_witness_test.go @@ -0,0 +1,179 @@ +package tunnel + +import ( + "context" + "errors" + "sync" + "testing" + "time" + + "github.com/quic-go/quic-go/qlog" +) + +// Installed only after preceding writers have joined. It witnesses exactly +// the first following raw Write, not a later call or the whole phase worker. +type feedbackWriteWitness struct { + started chan struct{} + done chan struct{} + once sync.Once +} + +type feedbackBackpressureGate struct { + raw *feedbackRawStream + trace *feedbackBlockedTrace + first *feedbackWriteWitness + warmupOffset int64 +} + +func newFeedbackBackpressureGate(raw *feedbackRawStream, trace *feedbackBlockedTrace, warmupOffset int64) *feedbackBackpressureGate { + first := &feedbackWriteWitness{started: make(chan struct{}), done: make(chan struct{})} + raw.writeWitness = first + return &feedbackBackpressureGate{raw: raw, trace: trace, first: first, warmupOffset: warmupOffset} +} + +func (g *feedbackBackpressureGate) completed(finished <-chan struct{}) error { + select { + case <-g.first.done: + return errors.New("first raw Write completed before receiver backpressure was witnessed") + default: + } + select { + case <-finished: + return errors.New("phase writer completed before receiver backpressure was witnessed") + default: + } + return nil +} + +func (g *feedbackBackpressureGate) await(ctx context.Context, finished <-chan struct{}) (int64, error) { + select { + case <-ctx.Done(): + return 0, ctx.Err() + case <-g.first.started: + case <-finished: + return 0, errors.New("phase writer never established a pending raw Write") + } + for { + if err := ctx.Err(); err != nil { + return 0, err + } + if err := g.completed(finished); err != nil { + return 0, err + } + if offset := g.trace.highest(); offset > g.warmupOffset { + // Check completion after reading the received-frame evidence too. + // A pending later call must not mask completion of the first call. + if err := g.completed(finished); err != nil { + return 0, err + } + if !g.raw.pending.Load() { + return 0, errors.New("first raw Write is not pending at the received blocked-frame witness") + } + return offset, nil + } + select { + case <-ctx.Done(): + return 0, ctx.Err() + case <-g.first.done: + return 0, errors.New("first raw Write completed before receiver backpressure was witnessed") + case <-finished: + return 0, errors.New("phase writer completed before receiver backpressure was witnessed") + case <-g.trace.changed: + } + } +} + +func TestFeedbackBackpressureGateRequiresPendingFirstWrite(t *testing.T) { + for _, mode := range []string{"pending", "first-completed", "phase-completed", "raw-not-pending", "canceled"} { + t.Run(mode, func(t *testing.T) { + trace := newFeedbackBlockedTrace() + trace.selectStream(0) + raw := &feedbackRawStream{} + gate := newFeedbackBackpressureGate(raw, trace, 1<<20) + finished := make(chan struct{}) + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() + raw.pending.Store(mode != "raw-not-pending") + close(gate.first.started) + trace.RecordEvent(qlog.PacketReceived{Frames: []qlog.Frame{{Frame: &qlog.StreamDataBlockedFrame{ + StreamID: 0, MaximumStreamData: (1 << 20) + (16 << 10), + }}}}) + switch mode { + case "first-completed": + // A later raw call may be pending, but cannot replace the first. + close(gate.first.done) + case "phase-completed": + close(finished) + case "canceled": + cancel() + } + offset, err := gate.await(ctx, finished) + if mode == "pending" { + if err != nil || offset <= gate.warmupOffset { + t.Fatalf("pending fresh blocked write: offset=%d err=%v", offset, err) + } + } else if err == nil { + t.Fatalf("%s unexpectedly established backpressure at offset %d", mode, offset) + } + }) + } +} + +func TestFeedbackBackpressureGateBoundsMissingEvidence(t *testing.T) { + trace := newFeedbackBlockedTrace() + trace.selectStream(0) + gate := newFeedbackBackpressureGate(&feedbackRawStream{}, trace, 1<<20) + gate.raw.pending.Store(true) + close(gate.first.started) + // Boundary and stale warmup evidence cannot release the slow-phase gate. + trace.RecordEvent(qlog.PacketReceived{Frames: []qlog.Frame{{Frame: &qlog.StreamDataBlockedFrame{ + StreamID: 0, MaximumStreamData: 1 << 20, + }}}}) + ctx, cancel := context.WithTimeout(context.Background(), 20*time.Millisecond) + defer cancel() + offset, err := gate.await(ctx, make(chan struct{})) + if !errors.Is(err, context.DeadlineExceeded) || offset != 0 { + t.Fatalf("stale/equal-boundary evidence must wait until its deadline: offset=%d err=%v", offset, err) + } +} + +func TestQUICPacingBackpressureWitnessRequiresReceiveLimit(t *testing.T) { + trace := newFeedbackBlockedTrace() + // More than the entire warmup and slow phase can fit without a read. + // Retain the same controller, payload, deadlines and actual QUIC path. + f := newFeedbackQUICFixtureWithTrace(t, 1, 2<<20, trace) + trace.selectStream(f.clientStreams[0].StreamID()) + const warmupBytes = 1 << 20 + runPacingRecoveryPhase(t, f, "wide-window warmup", warmupBytes, 0, nil) + gate := newFeedbackBackpressureGate(f.rawStreams[0], trace, warmupBytes) + finished := make(chan struct{}) + writer := f.run(func() error { + defer close(finished) + return writeFeedbackPayload(f.serverStreams[0], 16*feedbackChunkSize) + }) + ctx, cancel := context.WithTimeout(f.ctx, 2*time.Second) + defer cancel() + offset, err := gate.await(ctx, finished) + if err == nil { + t.Fatalf("unrestricted receive window established backpressure: stream=%d offset=%d", f.clientStreams[0].StreamID(), offset) + } + if ctx.Err() != nil { + t.Fatalf("wide-window control did not reject a completed first Write before its deadline: %v", err) + } + select { + case <-gate.first.done: + default: + t.Fatalf("wide-window control did not complete its first raw Write: %v", err) + } + if got := trace.highest(); got > warmupBytes { + t.Fatalf("wide-window control received new blocking offset %d beyond warmup %d", got, warmupBytes) + } + // Drain every original slow byte and join the actual writer independently + // even though the setup gate was rejected. Fixture cleanup joins on failure. + if err := readFeedbackPayload(f.clientStreams[0], 16*feedbackChunkSize); err != nil { + t.Fatal(err) + } + f.await(t, writer) + f.checkReverse(t) + t.Logf("wide-window negative control rejected without timeout: %v; verified %d slow bytes", err, 16*feedbackChunkSize) +}