From 3f7f3594b156c497565705a9f3694d5e15151cd7 Mon Sep 17 00:00:00 2001 From: cppla Date: Sat, 3 Oct 2026 13:00:50 +0800 Subject: [PATCH] test(tunnel): witness shared backpressure at a fixed feedback checkpoint --- docs/ARCHITECTURE.md | 8 + .../pacing_feedback_integration_test.go | 106 ++++- internal/tunnel/pacing_shared_witness_test.go | 422 ++++++++++++++++++ 3 files changed, 519 insertions(+), 17 deletions(-) create mode 100644 internal/tunnel/pacing_shared_witness_test.go diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index 3cedaf5..a308468 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -262,6 +262,14 @@ can lower the estimate. This prevents an application-imposed rate reduction from repeatedly becoming the next capacity estimate; it neither bypasses QUIC congestion control nor infers wire capacity by subtracting sleep time. +Capacity history is a recent maximum of eight eligible delivery samples, not +an average or a promise of a monotonically decreasing target. Higher delivery +evidence can raise it even in a pacing-dominated interval. The shared-stream +regression test therefore witnesses actual peer-received flow-control blocking +and checks learning at a fixed, still-backpressured feedback checkpoint. It +also drains the remaining payload and verifies same-connection recovery; +an arbitrary final target alone is not evidence that capacity never adapted. + Web H2/H3 streams do not use the native pacing negotiation and report client and relay pacing as `not-applicable`. Fixed-rate is rejected with `web-auto`, `h3`, and `h2`. diff --git a/internal/tunnel/pacing_feedback_integration_test.go b/internal/tunnel/pacing_feedback_integration_test.go index c961cd8..ea5638b 100644 --- a/internal/tunnel/pacing_feedback_integration_test.go +++ b/internal/tunnel/pacing_feedback_integration_test.go @@ -43,7 +43,11 @@ func TestQUICPacingSustainedDownloadDoesNotSelfLimitToFloor(t *testing.T) { func TestQUICPacingSharedStreamsAcceptTransportBackpressure(t *testing.T) { // Each stream's fixed 16 KiB receive window is smaller than one 32 KiB // write, and cannot auto-grow to absorb the entire slow-reader payload. - f := newFeedbackQUICFixture(t, 2, 16<<10) + trace := newFeedbackSharedTrace() + f := newFeedbackQUICFixtureWithTrace(t, 2, 16<<10, trace) + for index := range f.clientStreams { + trace.streams[index].selectStream(f.clientStreams[index].StreamID()) + } warmup := f.run(func() error { return writeFeedbackPayload(f.serverStreams[0], 1<<20) }) if err := readFeedbackPayload(f.clientStreams[0], 1<<20); err != nil { t.Fatal(err) @@ -54,16 +58,24 @@ func TestQUICPacingSharedStreamsAcceptTransportBackpressure(t *testing.T) { } const rounds = 16 + // The learning checkpoint is declared before either slow writer starts. + // It is not a minimum selected from a target series, and ten reader rounds + // do not by themselves claim eight eligible controller history samples. + const checkpoint = 10 const slowBytes = rounds * feedbackChunkSize - primaryEntered := make(chan struct{}, 1) - f.rawStreams[0].entered = primaryEntered + receiptQueue := make(chan feedbackSharedReceipt, 2) + primaryWitness := newFeedbackSharedWitness(f.rawStreams[0], trace.streams[0], 1<<20, checkpoint, receiptQueue) + f.rawStreams[0].sharedWitness = primaryWitness + f.streamPacers[0].sharedWitness = primaryWitness primaryTransportBefore := f.rawStreams[0].writeTime.Load() primaryPacingBefore := f.streamPacers[0].waitTime.Load() - primary := f.run(func() error { return writeFeedbackPayload(f.serverStreams[0], slowBytes) }) - select { - case <-primaryEntered: - case <-f.ctx.Done(): - t.Fatal(f.ctx.Err()) + primaryFinished := make(chan struct{}) + primary := f.run(func() error { + defer close(primaryFinished) + return writeFeedbackPayload(f.serverStreams[0], slowBytes) + }) + if _, err := primaryWitness.awaitRaw(f.ctx, primaryFinished, 1, 1<<20); err != nil { + t.Fatalf("primary fresh backpressure before sibling: %v", err) } // Start a fast sibling while the primary cannot drain its finite receive // buffer. The real clock observes actual token sleeps, not merely calls to @@ -80,7 +92,22 @@ func TestQUICPacingSharedStreamsAcceptTransportBackpressure(t *testing.T) { before := f.pacer.controller.TargetBytesPerSecond() // Both streams now have slow readers. Unlike one blocked stream with an // unrestricted sibling, this reduces delivery for the shared connection. - siblingSlow := f.run(func() error { return writeFeedbackPayload(f.serverStreams[1], slowBytes) }) + // Arm this immutable phase only after the preceding fast writer joined. + siblingWitness := newFeedbackSharedWitness(f.rawStreams[1], trace.streams[1], 128<<10, checkpoint, receiptQueue) + f.rawStreams[1].sharedWitness = siblingWitness + f.streamPacers[1].sharedWitness = siblingWitness + siblingFinished := make(chan struct{}) + siblingSlow := f.run(func() error { + defer close(siblingFinished) + return writeFeedbackPayload(f.serverStreams[1], slowBytes) + }) + if _, err := siblingWitness.awaitRaw(f.ctx, siblingFinished, 1, 128<<10); err != nil { + t.Fatalf("sibling fresh backpressure before shared reads: %v", err) + } + if _, err := primaryWitness.awaitRaw(f.ctx, primaryFinished, 1, 1<<20); err != nil { + t.Fatalf("primary first Write did not remain blocked: %v", err) + } + var after int64 for round := 0; round < rounds; round++ { timer := time.NewTimer(120 * time.Millisecond) select { @@ -94,10 +121,31 @@ func TestQUICPacingSharedStreamsAcceptTransportBackpressure(t *testing.T) { t.Fatal(err) } } + if round+1 == checkpoint { + receipts, err := awaitFeedbackSharedReceipts(f.ctx, + [2]*feedbackSharedWitness{primaryWitness, siblingWitness}, + [2]<-chan struct{}{primaryFinished, siblingFinished}) + if err != nil { + t.Fatalf("fixed round-%d feedback checkpoint: %v", checkpoint, err) + } + // This is the last of the two prescribed receipts to arrive, not + // the lower target. Each is a live target read after production + // Observe and Wait returned, not an atomic Observe result. + after = receipts[1].target + for index, witness := range []*feedbackSharedWitness{primaryWitness, siblingWitness} { + finished := [2]<-chan struct{}{primaryFinished, siblingFinished}[index] + offset, err := witness.awaitRaw(f.ctx, finished, checkpoint+1, + witness.warmupOffset+checkpoint*feedbackChunkSize) + if err != nil { + t.Fatalf("stream %d checkpoint Write: %v", index, err) + } + t.Logf("checkpoint stream=%d raw_completed=%d new_blocked_offset=%d receipt_target=%d", + f.clientStreams[index].StreamID(), checkpoint, offset, feedbackSharedReceiptFor(receipts, witness).target) + } + } } f.await(t, primary) f.await(t, siblingSlow) - after := f.pacer.controller.TargetBytesPerSecond() transport := time.Duration(f.rawStreams[0].writeTime.Load() - primaryTransportBefore) pacing := time.Duration(f.streamPacers[0].waitTime.Load() - primaryPacingBefore) if transport <= pacing { @@ -108,7 +156,7 @@ func TestQUICPacingSharedStreamsAcceptTransportBackpressure(t *testing.T) { // cannot reduce the target below 36% of its previous value. Crossing that // bound proves capacity evidence changed, not merely a congestion penalty. if after >= before*36/100 { - t.Fatalf("shared capacity did not adapt to transport backpressure: target %d -> %d, want below %d", before, after, before*36/100) + t.Fatalf("shared capacity checkpoint did not adapt to transport backpressure: target %d -> %d, want below %d", before, after, before*36/100) } // Removing the imposed reader delays must leave the same streams usable. for index := range f.serverStreams { @@ -119,7 +167,7 @@ func TestQUICPacingSharedStreamsAcceptTransportBackpressure(t *testing.T) { f.await(t, writer) } f.checkReverse(t) - t.Logf("shared target %d -> %d; primary QUIC Write=%s, pacing Wait=%s; overlapping token sleeps=%d", + t.Logf("shared round-%d checkpoint target %d -> %d; primary QUIC Write=%s, pacing Wait=%s; overlapping token sleeps=%d", checkpoint, before, after, transport, pacing, f.clock.sleepsWithPrimaryWrite.Load()) } @@ -304,15 +352,20 @@ func readFeedbackPayload(reader io.Reader, size int) error { type feedbackRawStream struct { quicStream - entered chan struct{} - pending atomic.Bool - writeTime atomic.Int64 - writeWitness *feedbackWriteWitness + entered chan struct{} + pending atomic.Bool + writeTime atomic.Int64 + writeWitness *feedbackWriteWitness + sharedWitness *feedbackSharedWitness } func (s *feedbackRawStream) Write(p []byte) (int, error) { started := time.Now() s.pending.Store(true) + var sharedCall *feedbackSharedRawCall + if witness := s.sharedWitness; witness != nil { + sharedCall = witness.rawStarted() + } first := false if witness := s.writeWitness; witness != nil { witness.once.Do(func() { @@ -325,6 +378,9 @@ func (s *feedbackRawStream) Write(p []byte) (int, error) { default: } n, err := s.quicStream.Write(p) + if sharedCall != nil { + s.sharedWitness.rawFinished(sharedCall, n, err) + } if first { close(s.writeWitness.done) } @@ -335,13 +391,29 @@ func (s *feedbackRawStream) Write(p []byte) (int, error) { type feedbackObservedPacer struct { *connectionPacer - waitTime atomic.Int64 + waitTime atomic.Int64 + sharedWitness *feedbackSharedWitness } func (p *feedbackObservedPacer) wait(ctx context.Context, size int, conn *quic.Conn) error { started := time.Now() + var token feedbackSharedWaitToken + if witness := p.sharedWitness; witness != nil { + token = witness.waitStarted() + } err := p.connectionPacer.wait(ctx, size, conn) p.waitTime.Add(int64(time.Since(started))) + if witness := p.sharedWitness; witness != nil { + // Only the prescribed successful return needs a target read. The + // shared controller may have consumed another stream's observation. + if token.ordinal == witness.checkpoint+1 { + target := int64(0) + if err == nil { + target = p.controller.TargetBytesPerSecond() + } + witness.waitReturned(token, err, target) + } + } return err } diff --git a/internal/tunnel/pacing_shared_witness_test.go b/internal/tunnel/pacing_shared_witness_test.go new file mode 100644 index 0000000..666d50a --- /dev/null +++ b/internal/tunnel/pacing_shared_witness_test.go @@ -0,0 +1,422 @@ +package tunnel + +import ( + "context" + "errors" + "fmt" + "sync" + "testing" + "time" + + "github.com/quic-go/quic-go/qlogwriter" +) + +// Each stream keeps its own copied blocked offset. Forwarding is synchronous, +// so neither observer retains library packet or frame pointers. +type feedbackSharedTrace struct{ streams [2]*feedbackBlockedTrace } + +var _ qlogwriter.Trace = (*feedbackSharedTrace)(nil) +var _ qlogwriter.Recorder = (*feedbackSharedRecorder)(nil) + +func newFeedbackSharedTrace() *feedbackSharedTrace { + return &feedbackSharedTrace{streams: [2]*feedbackBlockedTrace{newFeedbackBlockedTrace(), newFeedbackBlockedTrace()}} +} + +func (t *feedbackSharedTrace) SupportsSchemas(schema string) bool { + return t.streams[0].SupportsSchemas(schema) && t.streams[1].SupportsSchemas(schema) +} + +func (t *feedbackSharedTrace) AddProducer() qlogwriter.Recorder { + return &feedbackSharedRecorder{trace: t} +} + +func (t *feedbackSharedTrace) RecordEvent(event qlogwriter.Event) { + for _, stream := range t.streams { + stream.RecordEvent(event) + } +} + +func (*feedbackSharedTrace) Close() error { return nil } + +type feedbackSharedRecorder struct{ trace *feedbackSharedTrace } + +func (r *feedbackSharedRecorder) RecordEvent(event qlogwriter.Event) { r.trace.RecordEvent(event) } +func (*feedbackSharedRecorder) Close() error { return nil } + +type feedbackSharedRawCall struct { + owner *feedbackSharedWitness + ordinal int + done chan struct{} + completed bool // protected by owner.mu +} + +type feedbackSharedWaitToken struct { + owner *feedbackSharedWitness + ordinal int +} + +type feedbackSharedReceipt struct { + owner *feedbackSharedWitness + waitOrdinal int + rawCompleted int + completedBytes int + target int64 +} + +// The pointer and phase configuration are installed only after this stream's +// preceding writer joined, and never replaced while subsequent writers run. +// It observes calls; it never sleeps, gates I/O, or adds controller samples. +type feedbackSharedWitness struct { + raw *feedbackRawStream + trace *feedbackBlockedTrace + warmupOffset int64 + checkpoint int + receipts chan feedbackSharedReceipt // shared capacity-two FIFO, one value per stream + changed chan struct{} // coalesced hints; never closed + + mu sync.Mutex + waits int + started int + completed int + completedBytes int + current *feedbackSharedRawCall + receiptSent bool +} + +func newFeedbackSharedWitness(raw *feedbackRawStream, trace *feedbackBlockedTrace, warmup int64, checkpoint int, receipts chan feedbackSharedReceipt) *feedbackSharedWitness { + return &feedbackSharedWitness{raw: raw, trace: trace, warmupOffset: warmup, checkpoint: checkpoint, + receipts: receipts, changed: make(chan struct{}, 1)} +} + +func (w *feedbackSharedWitness) notify() { + select { + case w.changed <- struct{}{}: + default: + } +} + +func (w *feedbackSharedWitness) rawStarted() *feedbackSharedRawCall { + w.mu.Lock() + w.started++ + call := &feedbackSharedRawCall{owner: w, ordinal: w.started, done: make(chan struct{})} + w.current = call + w.mu.Unlock() + w.notify() + return call +} + +func (w *feedbackSharedWitness) rawFinished(call *feedbackSharedRawCall, n int, _ error) { + w.mu.Lock() + if call.owner == w && !call.completed { + call.completed = true + w.completed++ + w.completedBytes += n + close(call.done) + } + w.mu.Unlock() + w.notify() +} + +func (w *feedbackSharedWitness) waitStarted() feedbackSharedWaitToken { + w.mu.Lock() + defer w.mu.Unlock() + w.waits++ + return feedbackSharedWaitToken{owner: w, ordinal: w.waits} +} + +// Called only after the underlying production wait returned. Target is a live +// read after Observe/Wait, not an atomic result of that individual Observe. +func (w *feedbackSharedWitness) waitReturned(token feedbackSharedWaitToken, err error, target int64) { + w.mu.Lock() + if token.owner != w || token.ordinal != w.checkpoint+1 || err != nil || w.receiptSent { + w.mu.Unlock() + return + } + w.receiptSent = true + receipt := feedbackSharedReceipt{owner: w, waitOrdinal: token.ordinal, rawCompleted: w.completed, + completedBytes: w.completedBytes, target: target} + w.mu.Unlock() + // Capacity two holds exactly the two selected successful returns. FIFO + // arrival order, not a lower target or select between buffered queues, + // determines the final prescribed receipt. + select { + case w.receipts <- receipt: + default: + } +} + +func (w *feedbackSharedWitness) pendingCall(ordinal int) (*feedbackSharedRawCall, error) { + w.mu.Lock() + defer w.mu.Unlock() + if w.started < ordinal { + return nil, nil + } + if w.started != ordinal || w.completed != ordinal-1 || w.completedBytes != (ordinal-1)*feedbackChunkSize { + return nil, fmt.Errorf("raw call %d replaced or completed: started=%d completed=%d bytes=%d", ordinal, w.started, w.completed, w.completedBytes) + } + call := w.current + if call == nil || call.ordinal != ordinal || call.completed || !w.raw.pending.Load() { + return nil, fmt.Errorf("raw call %d is not pending", ordinal) + } + return call, nil +} + +func (w *feedbackSharedWitness) awaitRaw(ctx context.Context, finished <-chan struct{}, ordinal int, offsetAfter int64) (int64, error) { + for { + if err := ctx.Err(); err != nil { + return 0, err + } + select { + case <-finished: + return 0, errors.New("phase writer finished before prescribed pending Write") + default: + } + call, err := w.pendingCall(ordinal) + if err != nil { + return 0, err + } + var done <-chan struct{} + if call != nil { + done = call.done + if offset := w.trace.highest(); offset > offsetAfter { + // Recheck after the actual received-frame evidence. A later raw + // call cannot stand in for the prescribed call's completion. + if current, err := w.pendingCall(ordinal); err != nil || current != call { + return 0, fmt.Errorf("raw call changed while checking blocked frame: %v", err) + } + select { + case <-done: + return 0, errors.New("prescribed raw Write completed at blocked-frame witness") + case <-finished: + return 0, errors.New("phase writer completed at blocked-frame witness") + default: + return offset, nil + } + } + } + select { + case <-ctx.Done(): + return 0, ctx.Err() + case <-finished: + return 0, errors.New("phase writer finished before prescribed pending Write") + case <-done: + return 0, errors.New("prescribed raw Write completed before fresh blocked frame") + case <-w.changed: + case <-w.trace.changed: + } + } +} + +func awaitFeedbackSharedReceipts(ctx context.Context, witnesses [2]*feedbackSharedWitness, finished [2]<-chan struct{}) ([2]feedbackSharedReceipt, error) { + var result [2]feedbackSharedReceipt + if witnesses[0].receipts != witnesses[1].receipts { + return result, errors.New("stream receipts do not share an arrival-order queue") + } + seen := make(map[*feedbackSharedWitness]bool, 2) + for index := range result { + if err := ctx.Err(); err != nil { + return result, err + } + select { + case <-ctx.Done(): + return result, ctx.Err() + case <-finished[0]: + return result, errors.New("primary phase finished before feedback checkpoint") + case <-finished[1]: + return result, errors.New("sibling phase finished before feedback checkpoint") + case receipt := <-witnesses[0].receipts: + owner := receipt.owner + if (owner != witnesses[0] && owner != witnesses[1]) || seen[owner] { + return result, errors.New("foreign or duplicate phase receipt") + } + if receipt.waitOrdinal != owner.checkpoint+1 || receipt.rawCompleted != owner.checkpoint || receipt.completedBytes != owner.checkpoint*feedbackChunkSize { + return result, fmt.Errorf("invalid feedback checkpoint receipt: wait=%d raw_completed=%d bytes=%d", receipt.waitOrdinal, receipt.rawCompleted, receipt.completedBytes) + } + seen[owner] = true + result[index] = receipt + } + } + return result, nil +} + +func feedbackSharedReceiptFor(receipts [2]feedbackSharedReceipt, witness *feedbackSharedWitness) feedbackSharedReceipt { + for _, receipt := range receipts { + if receipt.owner == witness { + return receipt + } + } + return feedbackSharedReceipt{} +} + +func feedbackSharedUnitWitness(checkpoint int, receipts chan feedbackSharedReceipt) *feedbackSharedWitness { + trace := newFeedbackBlockedTrace() + trace.selectStream(0) + return newFeedbackSharedWitness(&feedbackRawStream{}, trace, 1<<20, checkpoint, receipts) +} + +func feedbackSharedCompleteRaw(w *feedbackSharedWitness, count int) { + for range count { + call := w.rawStarted() + w.rawFinished(call, feedbackChunkSize, nil) + } +} + +func TestFeedbackSharedReceiptRequiresSuccessfulPrescribedReturn(t *testing.T) { + for _, mode := range []string{"entry-only", "earlier-ordinal", "old-phase", "failed-return", "selected-return", "duplicate-return"} { + t.Run(mode, func(t *testing.T) { + w := feedbackSharedUnitWitness(10, make(chan feedbackSharedReceipt, 2)) + feedbackSharedCompleteRaw(w, 10) + var token feedbackSharedWaitToken + for range 11 { + token = w.waitStarted() + } + switch mode { + case "entry-only": + case "earlier-ordinal": + token.ordinal = 10 + w.waitReturned(token, nil, 101) + case "old-phase": + token.owner = feedbackSharedUnitWitness(10, make(chan feedbackSharedReceipt, 2)) + w.waitReturned(token, nil, 101) + case "failed-return": + w.waitReturned(token, context.Canceled, 101) + default: + w.waitReturned(token, nil, 101) + if mode == "duplicate-return" { + w.waitReturned(token, nil, 1) + } + } + select { + case receipt := <-w.receipts: + if mode != "selected-return" && mode != "duplicate-return" { + t.Fatalf("%s produced a checkpoint receipt", mode) + } + if receipt.target != 101 || receipt.rawCompleted != 10 || receipt.completedBytes != 10*feedbackChunkSize { + t.Fatalf("wrong selected receipt: %+v", receipt) + } + default: + if mode == "selected-return" || mode == "duplicate-return" { + t.Fatal("successful prescribed return did not publish") + } + } + select { + case <-w.receipts: + t.Fatal("observer published more than one receipt") + default: + } + }) + } +} + +func TestFeedbackSharedCheckpointValidatesReceiptAndArrivalOrder(t *testing.T) { + for _, mode := range []string{"valid", "old-phase", "duplicate-phase", "wrong-ordinal", "prior-incomplete", "wrong-bytes", "missing"} { + t.Run(mode, func(t *testing.T) { + queue := make(chan feedbackSharedReceipt, 2) + first, second := feedbackSharedUnitWitness(10, queue), feedbackSharedUnitWitness(10, queue) + one := feedbackSharedReceipt{owner: first, waitOrdinal: 11, rawCompleted: 10, completedBytes: 10 * feedbackChunkSize, target: 1} + two := one + two.owner, two.target = second, 100 // Last arrival must win, not the minimum. + switch mode { + case "old-phase": + one.owner = feedbackSharedUnitWitness(10, queue) + case "duplicate-phase": + two.owner = first + case "wrong-ordinal": + one.waitOrdinal = 10 + case "prior-incomplete": + one.rawCompleted = 9 + case "wrong-bytes": + one.completedBytes-- + } + queue <- one + if mode != "missing" { + queue <- two + } + ctx, cancel := context.WithTimeout(context.Background(), 20*time.Millisecond) + defer cancel() + receipts, err := awaitFeedbackSharedReceipts(ctx, [2]*feedbackSharedWitness{first, second}, [2]<-chan struct{}{make(chan struct{}), make(chan struct{})}) + if mode == "valid" { + if err != nil || receipts[1].target != 100 || receipts[1].owner != second { + t.Fatalf("fixed arrival-order checkpoint: receipts=%+v err=%v", receipts, err) + } + } else if err == nil { + t.Fatalf("%s receipt was accepted", mode) + } else if mode == "missing" && !errors.Is(err, context.DeadlineExceeded) { + t.Fatalf("missing receipt did not wait to its bounded deadline: %v", err) + } + }) + } +} + +func TestFeedbackSharedRawGateRequiresExactPendingCall(t *testing.T) { + for _, mode := range []string{"first", "checkpoint", "later-call", "earlier-call", "equal-offset", "not-pending", "short-prior-write", "missing-call", "phase-finished"} { + t.Run(mode, func(t *testing.T) { + w := feedbackSharedUnitWitness(10, make(chan feedbackSharedReceipt, 2)) + ordinal := 11 + if mode == "first" { + ordinal = 1 + } + if mode != "missing-call" { + feedbackSharedCompleteRaw(w, ordinal-1) + if mode == "short-prior-write" { + w.completedBytes-- + } + call := w.rawStarted() + if mode == "later-call" { + w.rawFinished(call, feedbackChunkSize, nil) + w.rawStarted() // This pending later Write must not substitute. + } else if mode == "earlier-call" { + ordinal++ + } + w.raw.pending.Store(mode != "not-pending") + } + threshold := w.warmupOffset + int64((ordinal-1)*feedbackChunkSize) + maximum := threshold + (16 << 10) + if mode == "equal-offset" { + maximum = threshold + } + w.trace.RecordEvent(feedbackBlockedPacket(0, maximum)) + ctx, cancel := context.WithTimeout(context.Background(), 20*time.Millisecond) + defer cancel() + finished := make(chan struct{}) + if mode == "phase-finished" { + close(finished) + } + offset, err := w.awaitRaw(ctx, finished, ordinal, threshold) + if mode == "first" || mode == "checkpoint" { + if err != nil || offset != maximum { + t.Fatalf("fresh prescribed pending call: offset=%d err=%v", offset, err) + } + } else if err == nil { + t.Fatalf("%s established a prescribed raw-call witness", mode) + } else if (mode == "equal-offset" || mode == "missing-call" || mode == "earlier-call") && !errors.Is(err, context.DeadlineExceeded) { + t.Fatalf("%s missing evidence did not remain bounded: %v", mode, err) + } + }) + } +} + +func TestFeedbackSharedTraceSeparatesStreamsAndProducerLifetimes(t *testing.T) { + trace := newFeedbackSharedTrace() + trace.streams[0].selectStream(0) + trace.streams[1].selectStream(4) + first, second := trace.AddProducer(), trace.AddProducer() + first.RecordEvent(feedbackBlockedPacket(0, 1024)) + if err := first.Close(); err != nil { + t.Fatal(err) + } + second.RecordEvent(feedbackBlockedPacket(4, 2048)) + second.RecordEvent(feedbackBlockedPacket(8, 1<<20)) + if got := trace.streams[0].highest(); got != 1024 { + t.Fatalf("primary stream maximum = %d", got) + } + if got := trace.streams[1].highest(); got != 2048 { + t.Fatalf("sibling stream maximum = %d", got) + } + if err := second.Close(); err != nil { + t.Fatal(err) + } + if err := trace.Close(); err != nil { + t.Fatal(err) + } +}