diff --git a/connection_state.go b/connection_state.go
index b949fc1d..f4385530 100644
--- a/connection_state.go
+++ b/connection_state.go
@@ -14,6 +14,9 @@ import (
 )
 
 const (
+	// ReplayWindow is the build-time receive window and the assumed peer window
+	// for TX reservation credit. Use a power of two; peers must match or have a
+	// larger receive window. The bitmap costs ReplayWindow/8 bytes per tunnel.
 	ReplayWindow = 8192
 
 	// RehandshakeAfterMessages rolls keys inside the AES-GCM data-volume margin (~2^-36 advantage at 64KB frames).
@@ -41,6 +44,7 @@ type ConnectionState struct {
 	peerCert       *cert.CachedCertificate
 	initiator      bool
 	messageCounter atomic.Uint64
+	txBudget       txBudget
 	window         *Bits
 	decryptLock    sync.Mutex
 	writeLock      sync.Mutex
diff --git a/examples/config.yml b/examples/config.yml
index d0f2ec49..52c8cb57 100644
--- a/examples/config.yml
+++ b/examples/config.yml
@@ -281,6 +281,9 @@ tun:
   # Receive needs a listener that can hand off its batches, which Linux has.
   # IMPORTANT: staged senders require peers with at least the 8192-entry replay window of this release. Released
   # v1.11.x peers use 1024 and are incompatible. The window is not negotiated.
+  # Staged reservations may use half the window, the other half absorbs inline and control traffic, relay forwarding
+  # and network reordering; unlimited forwarding or delay can still exceed it. Waiting for credit on a busy tunnel
+  # also delays the other tunnels of the same read, and readers are not guaranteed fairness.
   #staged:
   #  workers: 0
 
diff --git a/flow.go b/flow.go
index 5b38abb2..dfca2109 100644
--- a/flow.go
+++ b/flow.go
@@ -19,11 +19,11 @@ import (
 // the crypto and then drops packets whose destination has closed. TX uses eight slots and RX six; at listen.batch=64
 // with GRO their buffers take about 9 MiB and 24 MiB per routine. Small-MSS TX arenas can grow.
 //
-// Metrics are registered only for active pipelines. flow.{tx,rx}.batches counts reads handed to the ordered stage.
-// flow.stage.<stage>.q<queue>.busy_ns locates the limiting stage: tx_read, tx_encrypt, tx_send, rx_read, rx_decrypt
-// or rx_write. Take rates; 1e9 ns/s fills a serial stage. These are elapsed times that exclude read syscalls and
-// downstream waits; descheduling inflates them and worker times add up across the pool. Small RX batches sent to the
-// writer include decryption in writer time.
+// Metrics are registered only for active pipelines. flow.{tx,rx}.batches counts reads handed to the ordered stage and
+// flow.tx.inline oversized reads the reader sent itself. flow.stage.<stage>.q<queue>.busy_ns locates the limiting
+// stage: tx_read, tx_encrypt, tx_send, rx_read, rx_decrypt or rx_write. Take rates; 1e9 ns/s fills a serial stage.
+// These are elapsed times that exclude read syscalls and downstream waits; descheduling inflates them and worker
+// times add up across the pool. Small RX batches sent to the writer include decryption in writer time.
 
 // stagedConfig is tun.staged. workers is per routine and direction; zero keeps processing inline.
 type stagedConfig struct {
diff --git a/flow_budget.go b/flow_budget.go
new file mode 100644
index 00000000..0f8d8279
--- /dev/null
+++ b/flow_budget.go
@@ -0,0 +1,183 @@
+package nebula
+
+import (
+	"sync/atomic"
+	"time"
+
+	"github.com/slackhq/nebula/overlay/tio"
+)
+
+// txBudget is the reservation credit of one tunnel, shared by every staged reader sending on it. The low 24 bits of
+// state are outstanding counters, the next 24 spent credit, and the top 16 the sole ordered reader (q+1); owner zero
+// means shared. Readers beyond the owner field's range always take the conservative shared path.
+//
+// A cap on outstanding counters alone is not enough: a fast reader could repeatedly send and reacquire credit,
+// overtaking a paused reader by arbitrarily far. So once readers overlap, or another reader is blocked, spent credit
+// is only recycled when ALL reservations have drained. A sole reader instead gets each ordered completion refunded,
+// its later batches cannot overtake its earlier ones. All transitions are one CAS, without locks.
+//
+// The credit is half the peer's replay window. The other half is left for uncredited sends and transport reordering:
+// inline and control traffic, and relay forwarding through SendVia, which takes counters on the relay tunnel without
+// credit. Unlimited bypass traffic or network delay can still exceed it. The window is not negotiated, so
+// ci.window.length, from the build-time ReplayWindow, is also the assumed peer window.
+type txBudget struct{ state atomic.Uint64 }
+
+const txBudgetMask = uint64(1<<24 - 1)
+
+// Fail compilation for unsupported build-time windows: a power of two, at least 8192, fitting the credit fields (up
+// to 16777216).
+const (
+	_ = uint64(txBudgetMask) - ReplayWindow/2
+	_ = uint64(ReplayWindow) - 8192
+	_ = uint64(0) - (ReplayWindow & (ReplayWindow - 1))
+)
+
+func txBudgetOwner(q int) uint64 {
+	if q < 0 || q >= 65535 {
+		return 0
+	}
+	return uint64(q + 1)
+}
+
+func (b *txBudget) acquire(n, limit, owner uint64) bool {
+	for {
+		old := b.state.Load()
+		pending, spent, who := old&txBudgetMask, (old>>24)&txBudgetMask, old>>48
+		if n > limit || spent > limit-n {
+			// A blocked reader ends sole ownership so ordered completions stop refunding credit. Otherwise a busy
+			// owner could keep a larger waiter blocked forever without ever sharing a reservation.
+			if who != 0 && who != owner && !b.state.CompareAndSwap(old, old&^(uint64(0xffff)<<48)) {
+				continue
+			}
+			return false
+		}
+		if pending == 0 {
+			who = owner
+		} else if who != owner {
+			who = 0
+		}
+		next := who<<48 | (spent+n)<<24 | (pending + n)
+		if b.state.CompareAndSwap(old, next) {
+			return true
+		}
+	}
+}
+
+// release runs only after the socket flush or an explicit drop. A rollback returns credit before any message counter
+// was allocated, as when a later tunnel of a relay or multi-tunnel batch could not acquire its own.
+func (b *txBudget) release(n uint64, rollback bool) {
+	for {
+		old := b.state.Load()
+		pending, spent, owner := old&txBudgetMask, (old>>24)&txBudgetMask, old>>48
+		pending -= n
+		if rollback || owner != 0 {
+			spent -= n
+		}
+		next := owner<<48 | spent<<24 | pending
+		if pending == 0 {
+			next = 0
+		}
+		if b.state.CompareAndSwap(old, next) {
+			return
+		}
+	}
+}
+
+func (ci *ConnectionState) txBudgetLimit() uint64 { return ci.window.length / 2 }
+
+func (r *txReserver) oversized() bool {
+	for _, t := range r.tunnels {
+		if t.n > t.ci.txBudgetLimit() {
+			return true
+		}
+	}
+	return false
+}
+
+func (r *txReserver) tryBudget() bool {
+	for i, t := range r.tunnels {
+		if !t.ci.txBudget.acquire(t.n, t.ci.txBudgetLimit(), r.owner) {
+			for _, acquired := range r.tunnels[:i] {
+				acquired.ci.txBudget.release(acquired.n, true)
+			}
+			return false
+		}
+	}
+	return true
+}
+
+// waitBudget acquires credit for every tunnel of the batch, never holding a partial acquisition while it waits. Each
+// reader first flushes its own oldest batch, flushes need no credit; with none, another reader holds all outstanding
+// work and can flush it independently. So there is no circular wait, even with relay tunnels acquired in opposite
+// orders. The sleep is only for readers with no local work, not the uncontended hot path. Acquisition is
+// all-or-nothing per read, so pressure on one tunnel also delays packets for idle tunnels in the read; splitting
+// would need separate slot ownership and coupled inner/relay reservations. Reader fairness is not guaranteed.
+func (p *stagedTx) waitBudget() (time.Duration, bool) {
+	if p.tryBudget() {
+		return 0, true
+	}
+	t := time.Now()
+	for {
+		if p.f.closed.Load() {
+			return time.Since(t), false
+		}
+		if !p.flushOldest() {
+			time.Sleep(50 * time.Microsecond)
+		}
+		if p.tryBudget() {
+			return time.Since(t), true
+		}
+	}
+}
+
+func (s *txSlot) releaseBudget() {
+	for _, t := range s.reservations {
+		t.ci.txBudget.release(t.n, false)
+	}
+	clear(s.reservations)
+	s.reservations = s.reservations[:0]
+}
+
+// sendLarge handles the unusual reads that exceed a whole tunnel's credit, such as tiny-MSS superpackets. It drains
+// the pipeline, then segments and sends them synchronously with credit for each segment, never exceeding the cap and
+// never filtering or running connection manager side effects twice. This deliberately slow path avoids a lone-batch
+// exception to the shared invariant.
+func (p *stagedTx) sendLarge(items []txItem) time.Duration {
+	t := time.Now()
+	for p.flushOldest() {
+	}
+	stall := time.Since(t)
+	for _, it := range items {
+		err := tio.SegmentSuperpacket(it.pkt, func(seg []byte) error {
+			if p.f.closed.Load() {
+				return nil
+			}
+			part := it
+			part.pkt, part.segs = tio.Packet{Bytes: seg}, 1
+			p.tunnels = p.tunnels[:0]
+			part.counter = p.need(part.hostinfo, part.ci, 1)
+			if part.relayHI != nil {
+				part.relayCounter = p.need(part.relayHI, part.relayHI.ConnectionState, 1)
+			}
+			d, ok := p.waitBudget()
+			stall += d
+			if !ok {
+				return nil
+			}
+			one := [1]txItem{part}
+			if reserved := p.reserve(one[:]); len(reserved) != 0 {
+				p.f.encryptTxItem(&reserved[0], p.nb, p.sb)
+				p.f.flushSendBatch(p.sb, p.q)
+			}
+			for _, t := range p.tunnels {
+				t.ci.txBudget.release(t.n, false)
+			}
+			return nil
+		})
+		if err != nil {
+			it.hostinfo.logger(p.f.l).Error("Failed to segment superpacket for send", "error", err)
+		}
+	}
+	clear(items)
+	return stall
+}
diff --git a/flow_budget_test.go b/flow_budget_test.go
new file mode 100644
index 00000000..26107864
--- /dev/null
+++ b/flow_budget_test.go
@@ -0,0 +1,104 @@
+package nebula
+
+import (
+	"fmt"
+	"testing"
+
+	"github.com/slackhq/nebula/test"
+	"github.com/stretchr/testify/require"
+)
+
+var budgetWindows = []uint64{8192, 65536}
+
+// TestTXBudgetOvertaking: a fast sender must not recycle credit around a stalled older reservation. A pure in-flight
+// count passes the cap assertion but fails this replay test.
+func TestTXBudgetOvertaking(t *testing.T) {
+	for _, window := range budgetWindows {
+		t.Run(fmt.Sprint(window), func(t *testing.T) {
+			var b txBudget
+			limit := window / 2
+			w := NewBits(window)
+			l := test.NewLogger()
+			require.True(t, b.acquire(1, limit, 1)) // counter 1 stalls
+			for counter := uint64(2); counter <= limit; counter++ {
+				require.True(t, b.acquire(1, limit, 2))
+				require.True(t, w.Update(l, counter))
+				b.release(1, false)
+			}
+			require.Equal(t, uint64(1), b.state.Load()&txBudgetMask)
+			require.False(t, b.acquire(1, limit, 2), "spent credit cannot be recycled while the oldest sender stalls")
+			// The other half of the window allows this many unreserved sends.
+			for counter := limit + 1; counter <= window; counter++ {
+				require.True(t, w.Update(l, counter))
+			}
+			require.True(t, w.Update(l, 1), "oldest reservation must still fit after using the whole margin")
+			b.release(1, false)
+			require.Zero(t, b.state.Load())
+			require.True(t, b.acquire(limit, limit, 2))
+			require.False(t, b.acquire(1, limit, 2))
+			b.release(limit, false)
+		})
+	}
+}
+
+// TestTXBudgetBlockedReader models eight 512-segment reads of a busy owner and a 568-segment waiter, without socket
+// timing. Alone, the owner gets each completion refunded, so it keeps eight batches in flight forever, even across its
+// own failed acquisitions. But even when the owner always retries first, its completions must stop refunding credit
+// once the waiter is blocked, and let the budget drain. Waiter 0 is the conservative ID of readers beyond the owner
+// field's range.
+func TestTXBudgetBlockedReader(t *testing.T) {
+	for _, waiter := range []uint64{1, 0} {
+		t.Run(fmt.Sprintf("waiter=%d", waiter), func(t *testing.T) {
+			var b txBudget
+			const limit, n, want = uint64(4096), uint64(512), uint64(568)
+			for range 8 {
+				require.True(t, b.acquire(n, limit, 2))
+			}
+			for range 100 {
+				require.False(t, b.acquire(n, limit, 2))
+				b.release(n, false) // the sole sender flushes its oldest batch
+				require.True(t, b.acquire(n, limit, 2))
+				require.Equal(t, 8*n, b.state.Load()&txBudgetMask)
+			}
+			require.False(t, b.acquire(want, limit, waiter))
+			for range 7 {
+				b.release(n, false)
+				require.False(t, b.acquire(n, limit, 2), "blocked reader must stop the owner's credit recycling")
+				require.False(t, b.acquire(want, limit, waiter))
+			}
+			b.release(n, false)
+			require.Zero(t, b.state.Load())
+			// Even if the former owner wins the next acquisition, both now fit.
+			require.True(t, b.acquire(n, limit, 2))
+			require.True(t, b.acquire(want, limit, waiter))
+			b.release(n, false)
+			b.release(want, false)
+			require.Zero(t, b.state.Load())
+		})
+	}
+}
+
+// TestTXBudgetRelayRollback: when a later tunnel of a batch cannot acquire, the earlier ones give their credit back,
+// spent credit included.
+func TestTXBudgetRelayRollback(t *testing.T) {
+	for _, window := range budgetWindows {
+		inner := &ConnectionState{window: NewBits(window)}
+		outer := &ConnectionState{window: NewBits(window)}
+		limit := inner.txBudgetLimit()
+		require.True(t, inner.txBudget.acquire(10, limit, 1))
+		require.True(t, outer.txBudget.acquire(limit, limit, 1))
+		r := txReserver{owner: 2, tunnels: []txTunnel{{ci: inner, n: 20}, {ci: outer, n: 20}}}
+		for range 100 {
+			require.False(t, r.tryBudget())
+			require.Equal(t, uint64(10), inner.txBudget.state.Load()&txBudgetMask)
+			require.Equal(t, uint64(10), (inner.txBudget.state.Load()>>24)&txBudgetMask, "rollback must refund spent credit too")
+		}
+		outer.txBudget.release(limit, false)
+		require.True(t, r.tryBudget())
+		inner.txBudget.release(20, false)
+		outer.txBudget.release(20, false)
+		inner.txBudget.release(10, false)
+		require.Zero(t, inner.txBudget.state.Load())
+		require.Zero(t, outer.txBudget.state.Load())
+	}
+}
diff --git a/flow_test.go b/flow_test.go
index b0da417f..8e059d59 100644
--- a/flow_test.go
+++ b/flow_test.go
@@ -367,6 +367,18 @@ func stagedName(sc stagedConfig) string {
 	}
 }
 
+// send takes one read through the pipeline, as run does.
+func (p *stagedTx) send(pkts []tio.Packet) {
+	p.handle(pkts, &firewall.ParsedPacket{}, make([]byte, mtu), nil)
+}
+
+// drain waits until every batch p has in flight is on the wire.
+func (p *stagedTx) drain() {
+	for p.inFlight > 0 {
+		p.release(<-p.returned)
+	}
+}
+
 // TestFlowStagedFallback: a tun queue that cannot hand off its buffer, or a conn that cannot hand off its batches,
 // keeps that direction on the inline path.
 func TestFlowStagedFallback(t *testing.T) {
diff --git a/flow_tx.go b/flow_tx.go
index de6f0177..80ccbb54 100644
--- a/flow_tx.go
+++ b/flow_tx.go
@@ -11,6 +11,7 @@ import (
 	"github.com/slackhq/nebula/noiseutil"
 	"github.com/slackhq/nebula/overlay/batch"
 	"github.com/slackhq/nebula/overlay/tio"
+	"github.com/slackhq/nebula/udp"
 )
 
 var errFlowSegmentOverrun = errors.New("superpacket has more segments than counters reserved for it")
@@ -142,6 +143,7 @@ type txTunnel struct {
 type txReserver struct {
 	f       *Interface
 	tunnels []txTunnel
+	owner   uint64
 }
 
 // collect runs the filter over pkts and appends the sendable ones to items, tallying each tunnel's counter needs.
@@ -252,8 +254,8 @@ type txSlot struct {
 	buf  []byte
 	sb   *batch.SendBatch
 	done chan struct{}
-	// reserved is how many counters the batch reserved, for the reader's maxReserved budget
-	reserved uint64
+	// reservations hold shared tunnel credit until the flush, including for dropped counters
+	reservations []txTunnel
 }
 
 func (s *txSlot) run(nb []byte) {
@@ -272,6 +274,9 @@ type stagedTx struct {
 	senderCPU int // -1 leaves the sender within its inherited affinity
 	queue     handoffQueue
 	pool      *flowPool
+	// sb and nb are the reader's own, for the sends it does itself
+	sb *batch.SendBatch
+	nb []byte
 
 	// free holds the reader's idle slots. A dispatched slot goes to the workers and, in the same order, to the
 	// sender on ordered; the sender gives it back on returned once it is on the wire.
@@ -280,30 +285,30 @@ type stagedTx struct {
 	returned   chan *txSlot
 	inFlight   int
 	senderDone chan struct{}
-	// reserved is the counters held by in-flight slots, maxReserved this reader's half-window share
-	reserved    uint64
-	maxReserved uint64
 
 	readBusy, encryptBusy, sendBusy metrics.Counter
 	metricBatches                   metrics.Counter
+	metricInline                    metrics.Counter
 }
 
 // newStagedTx needs a queue that hands over whole read batches, the offload (TSO/USO) queue on linux. On any other
 // queue a read is one packet, and staging it would cost a handoff per packet.
 func newStagedTx(f *Interface, queue handoffQueue, q int) *stagedTx {
 	p := &stagedTx{
-		txReserver:    txReserver{f: f, tunnels: make([]txTunnel, 0, 4)},
+		txReserver:    txReserver{f: f, tunnels: make([]txTunnel, 0, 4), owner: txBudgetOwner(q)},
 		q:             q,
 		senderCPU:     -1,
 		queue:         queue,
+		sb:            batch.NewSendBatch(f.writers[q], batch.SendBatchCap, batch.SendBatchCap*(udp.MTU+32)),
+		nb:            make([]byte, 12),
 		ordered:       make(chan *txSlot, stagedTxSlots),
 		returned:      make(chan *txSlot, stagedTxSlots),
 		senderDone:    make(chan struct{}),
-		maxReserved:   (ReplayWindow / 2) / uint64(max(f.routines, 1)),
 		readBusy:      stageBusy("tx_read", q),
 		encryptBusy:   stageBusy("tx_encrypt", q),
 		sendBusy:      stageBusy("tx_send", q),
 		metricBatches: metrics.GetOrRegisterCounter("flow.tx.batches", nil),
+		metricInline:  metrics.GetOrRegisterCounter("flow.tx.inline", nil),
 	}
 	if f.pinThreads {
 		p.cpu = f.threadCPU(q)
@@ -363,21 +368,26 @@ func (p *stagedTx) handle(pkts []tio.Packet, fwPacket *firewall.ParsedPacket, re
 		return stall
 	}
 
-	var total uint64
-	for _, t := range p.tunnels {
-		total += t.n
+	if p.oversized() {
+		p.metricInline.Inc(1)
+		stall += p.sendLarge(items)
+		slot.items = items[:0]
+		p.free = append(p.free, slot)
+		return stall
 	}
-	// Over this reader's share of the window, wait for its oldest batches to go out first. A lone batch always goes.
-	if p.inFlight > 0 && p.reserved+total > p.maxReserved {
-		t := time.Now()
-		for p.inFlight > 0 && p.reserved+total > p.maxReserved {
-			p.release(<-p.returned)
-		}
-		stall += time.Since(t)
+	d, ok := p.waitBudget()
+	stall += d
+	if !ok {
+		clear(items)
+		slot.items = items[:0]
+		p.free = append(p.free, slot)
+		return stall
 	}
+	slot.reservations = append(slot.reservations[:0], p.tunnels...)
 	items = p.reserve(items)
 	slot.items = items
 	if len(items) == 0 {
+		slot.releaseBudget()
 		p.free = append(p.free, slot)
 		return stall
 	}
@@ -385,8 +395,6 @@ func (p *stagedTx) handle(pkts []tio.Packet, fwPacket *firewall.ParsedPacket, re
 	// The packets now live in the slot, and the queue reads into the slot's old buffer
 	slot.buf = p.queue.SwapRxBuf(slot.buf)
 
-	slot.reserved = total
-	p.reserved += total
 	p.inFlight++
 	p.metricBatches.Inc(1)
 	// Counters were reserved above on this goroutine and the slot takes its place in the sender's order right here,
@@ -422,9 +430,17 @@ func (p *stagedTx) reap() {
 	}
 }
 
+// flushOldest waits for the oldest batch in flight to reach the wire and takes its slot back. It reports false when
+// nothing is in flight.
+func (p *stagedTx) flushOldest() bool {
+	if p.inFlight == 0 {
+		return false
+	}
+	p.release(<-p.returned)
+	return true
+}
+
 func (p *stagedTx) release(s *txSlot) {
-	p.reserved -= s.reserved
-	s.reserved = 0
 	p.inFlight--
 	p.free = append(p.free, s)
 }
@@ -444,6 +460,7 @@ func (p *stagedTx) sender() {
 		} else {
 			p.f.flushSendBatch(s.sb, p.q)
 		}
+		s.releaseBudget()
 		p.sendBusy.Inc(int64(time.Since(t)))
 		clear(s.items)
 		s.items = s.items[:0]
diff --git a/flow_tx_test.go b/flow_tx_test.go
index 631b7b7b..7dabad48 100644
--- a/flow_tx_test.go
+++ b/flow_tx_test.go
@@ -7,6 +7,8 @@ import (
 	"net/netip"
 	"os"
 	"os/exec"
+	"slices"
+	"sync"
 	"testing"
 	"time"
 
@@ -115,20 +117,27 @@ func TestFlowStagedTXExhausted(t *testing.T) {
 	require.Empty(t, sender.conn.captured())
 	require.Equal(t, RejectAfterMessages, ci.messageCounter.Load())
 	require.Equal(t, int64(12), sender.f.messageMetrics.txExhausted.Count())
+	require.Zero(t, ci.txBudget.state.Load())
 }
 
 // TestFlowStagedTXRelay sends staged batches, and an empty read, through a relay: outer and inner layers both verify,
 // in order, and go to the relay.
+// With a tiny MSS, the middle superpacket alone exceeds the tunnels' whole credit: it must be split, not admitted as an
+// unbounded lone batch, with either replay window.
 func TestFlowStagedTXRelay(t *testing.T) {
 	defer goleak.VerifyNone(t, goleak.IgnoreCurrent())
-	for _, window := range []uint64{ReplayWindow} {
+	for _, window := range budgetWindows {
 		t.Run(fmt.Sprintf("window=%d", window), func(t *testing.T) {
 			sender, receiver := newFlowTestPair(t, noise.CipherChaChaPoly, stagedOn(2))
 			sender.hostinfo.ConnectionState.window = NewBits(window)
 			receiver.hostinfo.ConnectionState.window = NewBits(window)
-			relayRemote, _, relayPeer := addTestRelay(t, sender, window)
+			relayRemote, relayHI, relayPeer := addTestRelay(t, sender, window)
 			batches, want := flowTestBatches(2, 3, 1, 6)
 			var payload []byte
+			large := flowTCPSuperpacket(0, 9, 45)
+			large.GSO.Size = 1
+			payload = append([]byte(nil), large.Bytes[40:]...)
+			batches = [][]tio.Packet{batches[0], {large}, batches[1]}
 			sender.f.listenIn(newFlowBatchQueue(append(batches, nil)), 0)
 			wire := sender.conn.captured()
 			require.Len(t, wire, len(want)+len(payload))
@@ -156,6 +165,8 @@ func TestFlowStagedTXRelay(t *testing.T) {
 			}
 			require.Equal(t, want, got)
 			require.Equal(t, payload, tiny)
+			require.Zero(t, sender.hostinfo.ConnectionState.txBudget.state.Load())
+			require.Zero(t, relayHI.ConnectionState.txBudget.state.Load())
 		})
 	}
 }
@@ -329,3 +340,68 @@ func TestFlowStagedTXAffinity(t *testing.T) {
 func sendPaths() []stagedConfig {
 	return []stagedConfig{{}, stagedOn(2)}
 }
+
+// requireAccepted checks the receiver accepts every packet of wire in the order given. With several readers the wire
+// order differs from counter order, but every packet must still authenticate and fit within the replay window.
+func requireAccepted(t *testing.T, receiver *flowTestPeer, wire [][]byte) {
+	t.Helper()
+	for _, pkt := range wire {
+		h := &header.H{}
+		require.NoError(t, h.Parse(pkt))
+		_, err := receiver.hostinfo.ConnectionState.Decrypt(receiver.f.l, h.MessageCounter, pkt, make([]byte, 12))
+		require.NoError(t, err, "counter %d", h.MessageCounter)
+	}
+}
+
+// slowConn delays every WriteBatch, then writes to the shared capture.
+type slowConn struct {
+	*flowCaptureConn
+	delay time.Duration
+}
+
+func (c slowConn) WriteBatch(bufs [][]byte, addrs []netip.AddrPort) (int, error) {
+	time.Sleep(c.delay)
+	return c.flowCaptureConn.WriteBatch(bufs, addrs)
+}
+
+// TestFlowTXBudgetManyReaders saturates one tunnel from eight readers, one of them on a slow socket that lets the
+// others overtake it, alongside a burst of unreserved control messages: in the order they reach the wire, every packet
+// is accepted by the receiver's replay window, so none was sent twice or displaced past the window.
+func TestFlowTXBudgetManyReaders(t *testing.T) {
+	defer goleak.VerifyNone(t, goleak.IgnoreCurrent())
+	for _, window := range budgetWindows {
+		t.Run(fmt.Sprintf("window=%d", window), func(t *testing.T) {
+			sender, receiver := newFlowTestPair(t, noise.CipherChaChaPoly, stagedOn(3))
+			ci := sender.hostinfo.ConnectionState
+			ci.window = NewBits(window)
+			receiver.hostinfo.ConnectionState.window = NewBits(window)
+			sender.f.routines = 8
+			sender.f.writers = append([]udp.Conn{slowConn{sender.conn, 4 * time.Millisecond}}, slices.Repeat(sender.f.writers, 8)...)
+			start := ci.messageCounter.Load()
+			var wg sync.WaitGroup
+			for q := range 8 {
+				wg.Go(func() {
+					batches, _ := flowTestBatches(28, 8, 2, 45)
+					p := newStagedTx(sender.f, newFlowBatchQueue(nil), q)
+					defer p.close()
+					for _, packets := range batches {
+						p.send(packets)
+					}
+				})
+			}
+			// A bounded burst of unreserved control messages uses the documented margin, on writer 8
+			wg.Go(func() {
+				for range 128 {
+					sender.f.sendNoMetrics(header.Test, header.TestRequest, ci, sender.hostinfo, flowTestUnder, []byte("x"), make([]byte, 12), make([]byte, mtu), 8)
+					time.Sleep(100 * time.Microsecond)
+				}
+			})
+			wg.Wait()
+			require.Zero(t, ci.txBudget.state.Load())
+			wire := sender.conn.captured()
+			require.Len(t, wire, 8*28*360+128)
+			require.Equal(t, start+uint64(len(wire)), ci.messageCounter.Load())
+			requireAccepted(t, receiver, wire)
+		})
+	}
+}