test: flakes get their root causes fixed

docs/egress.md

@@ -592,10 +592,11 @@ map update per tunnel and an atomic load per request read. A changed rate
 starts with a full bucket, a direction whose rate did not change keeps its
 bucket, and a repeated cap changes nothing. A change reaches every open
 connection at once: the proxy sets a past read deadline on every open tunnel,
-whose copy resumes in the matching mode, and a request reads the cap on every
-read. A claim in [advanced network mode](advanced-network.md#host-side) keeps
-separate buckets on its tun path, so its proxy and tun traffic are capped
-apart.
+whose copy resumes in the matching mode; an uncapped direction can pass one more
+splice, up to 1 MiB at the old, uncapped rate, when its data arrives together
+with the deadline, and a request reads the cap on every read. A claim in
+[advanced network mode](advanced-network.md#host-side) keeps separate buckets on
+its tun path, so its proxy and tun traffic are capped apart.
 
 A limited download paces what the proxy reads, not what the far end sends: the
 upstream can run ahead by up to the node's TCP receive buffer per connection

docs/sandboxd-api.md

@@ -1274,7 +1274,8 @@ cap. It covers the claim's [proxy doors](egress.md#bandwidth-limit) and, in
 [advanced network mode](advanced-network.md#host-side), replaces
 `tun.limits.bytes_per_second` for the claim in each direction it sets. Preview,
 port relay and exec traffic is not capped. On a running claim the change
-applies at once to open connections, without a reconnect. The cap is persisted
+applies at once to open connections, without a reconnect; an uncapped direction
+can pass one more splice, up to 1 MiB at the old rate. The cap is persisted
 with the claim, follows it through hibernate, archive, a live move, an archive
 row and an adoption, and is accepted on a hibernated or archived claim without
 waking it; its next wake applies it. A running claim's replica row under

sandboxd/egress/limit.go

@@ -26,7 +26,7 @@ const (
 
 var limitBufs = sync.Pool{New: func() any { b := make([]byte, limitBurst); return &b }}
 
-// SetLimit caps the claim's bytes per second each way and moves every open tunnel onto the new cap at once.
+// SetLimit caps the claim's bytes per second each way and moves every open tunnel onto the new cap after at most one more splice of an uncapped direction.
 func (p *Proxy) SetLimit(l types.EgressLimit) {
 	cur := p.limits.Load()
 	if cur == nil && l == (types.EgressLimit{}) || cur != nil && cur.set == l {

sandboxd/egress/limit_test.go

@@ -61,12 +61,19 @@ func TestChangingTheLimitMovesAnOpenTunnel(t *testing.T) {
 	}
 	waitBytes(t, s, 1<<20)
 
-	p.SetLimit(types.EgressLimit{TxBytesPerSecond: testRate / 2})
+	const perSecond, spliceChunk = 2 << 20, 1 << 20
+	base := s.received.Load()
+	p.SetLimit(types.EgressLimit{TxBytesPerSecond: perSecond})
 	start := time.Now()
 	go writeUntilClosed(conn)
-	waitBytes(t, s, 1<<20+limitBurst+testRate/2)
-	if took := time.Since(start); took < 700*time.Millisecond {
-		t.Errorf("a burst and a second's worth past it took %v after the limit dropped to %d/s", took, testRate/2)
+	waitBytes(t, s, base+perSecond+spliceChunk+perSecond*3/4)
+	got := s.received.Load()
+	took := time.Since(start)
+	if took < 700*time.Millisecond {
+		t.Errorf("a burst, a splice chunk and 0.75 s at the new cap took %v once the limit dropped to %d/s", took, perSecond)
+	}
+	if over := got - base - int64(took.Seconds()*perSecond); over > perSecond+spliceChunk+perSecond/limitSlices {
+		t.Errorf("%d bytes past the cap's pace once the limit dropped, want at most its burst and one splice chunk", over)
 	}
 
 	p.SetLimit(types.EgressLimit{})

sandboxd/pool/egress.go

@@ -14,7 +14,7 @@ import (
 	"github.com/cocoonstack/sandbox/sandboxd/usage"
 )
 
-// SetEgressLimit replaces a claim's bandwidth cap without waking a hibernated or archived claim; open connections move onto it at once.
+// SetEgressLimit replaces a claim's bandwidth cap without waking a hibernated or archived claim; open connections move onto it at once, after at most one more splice of an uncapped direction.
 func (m *Manager) SetEgressLimit(ctx context.Context, id string, l types.EgressLimit) error {
 	m.mu.Lock()
 	sb := m.claimed[id]

sandboxd/pool/tun_test.go

@@ -154,6 +154,7 @@ func TestDisablingARunningClaimRevokesItsPacketPath(t *testing.T) {
 	m := tunManager(t, eng, addrs, config.PoolSpec{PoolKey: testKey, Tun: true, Egress: anyPolicy})
 	sb := mustClaim(t, m, testKey)
 	stream := openStream(t, sb)
+	waitFor(t, func() bool { return slices.Equal(tunVerbsOf(eng), []string{"up " + sb.VsockSocket}) })
 	if err := m.SetTun(t.Context(), sb.ID, new(false), nil); err != nil {
 		t.Fatalf("disable: %v", err)
 	}
@@ -165,7 +166,8 @@ func TestDisablingARunningClaimRevokesItsPacketPath(t *testing.T) {
 		t.Error("the packet door still accepts after the disable")
 	}
 	waitFor(t, func() bool {
-		return slices.Equal(tunVerbsOf(eng), []string{"up " + sb.VsockSocket, "down " + sb.VsockSocket})
+		verbs := tunVerbsOf(eng)
+		return len(verbs) >= 2 && verbs[0] == "up "+sb.VsockSocket && !slices.ContainsFunc(verbs[1:], func(v string) bool { return v != "down "+sb.VsockSocket })
 	})
 	if err := m.SetTun(t.Context(), sb.ID, new(true), nil); err != nil {
 		t.Fatalf("enable: %v", err)
@@ -175,6 +177,25 @@ func TestDisablingARunningClaimRevokesItsPacketPath(t *testing.T) {
 	}
 }
 
+func TestADisableBeforeTheFirstTunUpLeavesTheGuestDown(t *testing.T) {
+	eng, addrs := newFakeEngine(), &fakeAddrs{}
+	eng.tunUpGate = make(chan struct{})
+	m := tunManager(t, eng, addrs, config.PoolSpec{PoolKey: testKey, Tun: true, Egress: anyPolicy})
+	sb := mustClaim(t, m, testKey)
+	if err := m.SetTun(t.Context(), sb.ID, new(false), nil); err != nil {
+		t.Fatalf("disable: %v", err)
+	}
+	waitFor(t, func() bool { return slices.Contains(tunVerbsOf(eng), "down "+sb.VsockSocket) })
+	close(eng.tunUpGate)
+	waitFor(t, func() bool {
+		verbs := tunVerbsOf(eng)
+		return len(verbs) == 3 && verbs[len(verbs)-1] == "down "+sb.VsockSocket
+	})
+	if doorOpen(sb) {
+		t.Error("the packet door accepts after a disable that beat the first tun_up")
+	}
+}
+
 func TestARearmWhileTunUpIsInFlightLeavesTheGuestUp(t *testing.T) {
 	eng, addrs := newFakeEngine(), &fakeAddrs{}
 	eng.tunUpGate = make(chan struct{})