review: watermark rate sampling, egress door connection cap, Python bulk-frame cases

docs/egress.md

@@ -41,7 +41,10 @@ use the SOCKS5 listener silkd binds at `127.0.0.1:1080` and relays over
 second door: CONNECT only (BIND and UDP ASSOCIATE answer *command not
 supported*), no authentication (the socket path already carries the sandbox's
 identity), and a `DOMAINNAME` destination is resolved host-side, so the guest
-still needs no resolver.
+still needs no resolver. Each door serves at most 256 connections per sandbox
+at a time; later ones sit unserved in the socket backlog until one closes, and
+the proxy keeps at most 64 idle upstream connections per sandbox, so a guest
+cannot turn its proxy into a host descriptor sink.
 
 A SOCKS5 tunnel takes the decision an HTTP `CONNECT` to the same host takes,
 through the same code: a rule with a nonempty `methods` list that omits

sandboxd/egress/proxy.go

@@ -19,7 +19,11 @@ import (
 	"github.com/cocoonstack/sandbox/sandboxd/utils"
 )
 
-const idleConnTimeout = 90 * time.Second
+const (
+	idleConnTimeout = 90 * time.Second
+	// maxIdleConns bounds the upstream pool per sandbox; a guest walking many hosts cannot park a descriptor per host.
+	maxIdleConns = 64
+)
 
 // hopHeaders are hop-by-hop and proxy-scoped headers this hop owns, never passed on.
 var hopHeaders = []string{
@@ -90,7 +94,7 @@ func New(sandbox, tenant string, policy Evaluator, secrets Secrets, ca *CA, dial
 		dial:    dial,
 		holder:  holder,
 		// the stdlib default of 2 idle conns per host re-dials bursty same-host traffic.
-		tr:    &http.Transport{DialContext: dial, MaxIdleConnsPerHost: 8, IdleConnTimeout: idleConnTimeout},
+		tr:    &http.Transport{DialContext: dial, MaxIdleConns: maxIdleConns, MaxIdleConnsPerHost: 8, IdleConnTimeout: idleConnTimeout},
 		conns: map[net.Conn]struct{}{},
 	}
 	if ca != nil {

sandboxd/pool/egress.go

@@ -13,13 +13,17 @@ import (
 	"time"
 
 	"github.com/projecteru2/core/log"
+	"golang.org/x/net/netutil"
 
 	"github.com/cocoonstack/sandbox/sandboxd/egress"
 	"github.com/cocoonstack/sandbox/sandboxd/engine"
 	"github.com/cocoonstack/sandbox/sandboxd/netfilter"
 	"github.com/cocoonstack/sandbox/sandboxd/types"
 )
 
+// egressDoorConns bounds one sandbox's open connections per door; the guest's next dial waits in the backlog.
+const egressDoorConns = 256
+
 var (
 	nat64Range = netip.MustParsePrefix("64:ff9b::/96") // RFC 6052 NAT64; the embedded v4 is checked instead
 
@@ -176,6 +180,10 @@ func (m *Manager) armEgressProxy(ctx context.Context, sb *types.Sandbox) error {
 	}
 	el.srv = &http.Server{Handler: proxy, ReadHeaderTimeout: 30 * time.Second}
 	el.proxy = proxy
+	el.ln = netutil.LimitListener(el.ln, egressDoorConns)
+	if el.socks != nil {
+		el.socks = netutil.LimitListener(el.socks, egressDoorConns)
+	}
 	m.mu.Lock()
 	displaced := m.egressListeners[id]
 	m.egressListeners[id] = el

sandboxd/pool/egress_test.go

@@ -12,6 +12,7 @@ import (
 	"path/filepath"
 	"slices"
 	"strings"
+	"syscall"
 	"testing"
 	"time"
 
@@ -657,6 +658,58 @@ func TestTrimmedWarmVMClosesItsDoors(t *testing.T) {
 	}
 }
 
+func TestEgressDoorCapsConcurrentConnections(t *testing.T) {
+	origin := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
+		_, _ = io.WriteString(w, "ok")
+	}))
+	t.Cleanup(origin.Close)
+	pol := &egress.Policy{Allow: []egress.Rule{{Host: mustHostname(t, origin.URL)}}}
+	m := egressManager(t, newFakeEngine(), config.PoolSpec{PoolKey: testKey, Warm: 1, Egress: pol})
+	m.dial = (&net.Dialer{}).DialContext
+	sb := vsockSandbox(t, "sb_cap")
+	if err := m.armEgress(t.Context(), sb); err != nil {
+		t.Fatalf("arm egress: %v", err)
+	}
+	path := engine.EgressSocketPath(sb.VsockSocket)
+
+	idle := make([]net.Conn, 0, egressDoorConns)
+	for len(idle) < egressDoorConns {
+		conn, err := net.Dial("unix", path)
+		if errors.Is(err, syscall.ECONNREFUSED) {
+			time.Sleep(10 * time.Millisecond)
+			continue
+		}
+		if err != nil {
+			t.Fatalf("idle dial %d: %v", len(idle), err)
+		}
+		idle = append(idle, conn)
+	}
+	t.Cleanup(func() {
+		for _, conn := range idle {
+			_ = conn.Close()
+		}
+	})
+
+	client := egressClient(path)
+	client.Timeout = 500 * time.Millisecond
+	if resp, err := client.Get(origin.URL + "/"); err == nil {
+		resp.Body.Close()
+		t.Fatal("a request past the door cap was served")
+	}
+
+	_ = idle[0].Close()
+	client.Timeout = 5 * time.Second
+	resp, err := client.Get(origin.URL + "/")
+	if err != nil {
+		t.Fatalf("request after a slot freed: %v", err)
+	}
+	resp.Body.Close()
+	if resp.StatusCode != http.StatusOK {
+		t.Errorf("status %d after a slot freed, want 200", resp.StatusCode)
+	}
+	m.disarmEgress(sb.ID, true)
+}
+
 func dialDoors(t *testing.T, sb *types.Sandbox) {
 	t.Helper()
 	for _, path := range []string{engine.EgressSocketPath(sb.VsockSocket), engine.SocksSocketPath(sb.VsockSocket)} {

sandboxd/pool/pool.go

@@ -180,6 +180,8 @@ type pool struct {
 	rate          float64
 	lead          time.Duration
 	lastArrival   time.Time
+	binStart      time.Time
+	binCount      int
 
 	goldenDir string
 	building  bool

sandboxd/pool/watermark.go

@@ -8,19 +8,24 @@ import (
 const (
 	// ewmaAlpha weights the newest observation in the arrival-rate and provision-lead averages
 	ewmaAlpha = 0.3
+	rateBin   = time.Second
 	// rateDecayTau halves the observed arrival rate roughly every 40s of silence
 	rateDecayTau = 60 * time.Second
 	// leadSafety over-provisions against the measured lead: arrivals are bursty, not uniform
 	leadSafety = 2.0
 )
 
-// noteArrival folds one claim into the arrival-rate EWMA. Caller holds the manager mutex.
+// noteArrival counts one claim; a bin that has spanned rateBin folds its arrivals per second into the rate EWMA. Caller holds the manager mutex.
 func (p *pool) noteArrival(now time.Time) {
-	if !p.lastArrival.IsZero() {
-		if dt := now.Sub(p.lastArrival).Seconds(); dt > 0 {
-			p.rate = ewmaAlpha*(1/dt) + (1-ewmaAlpha)*p.rate
-		}
+	switch elapsed := now.Sub(p.binStart); {
+	case p.binStart.IsZero():
+		p.binStart = now
+	case elapsed >= rateBin:
+		decayed := p.rate * math.Exp(-elapsed.Seconds()/rateDecayTau.Seconds())
+		p.rate = ewmaAlpha*(float64(p.binCount)/elapsed.Seconds()) + (1-ewmaAlpha)*decayed
+		p.binStart, p.binCount = now, 0
 	}
+	p.binCount++
 	p.lastArrival = now
 }
 

sandboxd/pool/watermark_test.go

@@ -15,19 +15,53 @@ func TestEffectiveTargetTracksDemand(t *testing.T) {
 		t.Fatalf("quiet pool target %d, want the floor", got)
 	}
 
-	for i := range 20 {
+	for i := range 60 {
 		p.noteArrival(now.Add(time.Duration(i) * 100 * time.Millisecond))
 	}
-	burstEnd := now.Add(2 * time.Second)
-	if got := p.effectiveTarget(burstEnd); got != 8 {
-		t.Errorf("burst target %d, want warmMax 8", got)
+	sustainedEnd := now.Add(6 * time.Second)
+	if got := p.effectiveTarget(sustainedEnd); got != 8 {
+		t.Errorf("sustained 10/s target %d, want warmMax 8", got)
 	}
 
-	if got := p.effectiveTarget(burstEnd.Add(5 * time.Minute)); got != 2 {
+	if got := p.effectiveTarget(sustainedEnd.Add(5 * time.Minute)); got != 2 {
 		t.Errorf("post-silence target %d, want the floor", got)
 	}
 }
 
+func TestEffectiveTargetIgnoresATwoClaimBurst(t *testing.T) {
+	now := time.Now()
+	p := &pool{key: types.PoolKey{}, floor: 2, warmMax: 20, lead: 500 * time.Millisecond}
+
+	p.noteArrival(now)
+	p.noteArrival(now.Add(time.Millisecond))
+	if got := p.effectiveTarget(now.Add(time.Second)); got != 2 {
+		t.Errorf("target %d after two claims 1 ms apart, want the floor", got)
+	}
+
+	p.noteArrival(now.Add(2 * time.Second))
+	if got := p.effectiveTarget(now.Add(2 * time.Second)); got != 2 {
+		t.Errorf("target %d after the burst's bin closed at 1/s, want the floor", got)
+	}
+}
+
+func TestEffectiveTargetForgetsRateAcrossSilence(t *testing.T) {
+	now := time.Now()
+	p := &pool{key: types.PoolKey{}, floor: 2, warmMax: 8, lead: 500 * time.Millisecond}
+	for i := range 60 {
+		p.noteArrival(now.Add(time.Duration(i) * 100 * time.Millisecond))
+	}
+	if got := p.effectiveTarget(now.Add(6 * time.Second)); got != 8 {
+		t.Fatalf("sustained target %d, want warmMax 8", got)
+	}
+
+	later := now.Add(10 * time.Minute)
+	p.noteArrival(later)
+	p.noteArrival(later.Add(time.Second))
+	if got := p.effectiveTarget(later.Add(time.Second)); got != 2 {
+		t.Errorf("target %d after two claims following ten minutes of silence, want the floor", got)
+	}
+}
+
 func TestEffectiveTargetOffWithoutWarmMax(t *testing.T) {
 	now := time.Now()
 	p := &pool{floor: 1, lead: time.Second}

sdk/python/tests/test_frames.py

@@ -21,14 +21,35 @@ def test_nested_data_field_not_shadowed():
     assert frame["meta"] == {"data": "WFhY"}
 
 
+def test_tag_after_other_keys_takes_the_full_parse():
+    frame = frames.decode_response(b'{"data":"aGk=","type":"stdout"}')
+    assert frame == {"data": b"hi", "type": "stdout"}
+
+
 @pytest.mark.parametrize(
     "raw",
     [
         b'{"type":"stdout","data":"aGk="}garbage',
         b'{"type":"data","data":"QUJD\r\nREVG"}',
         b'{"type":"stdout","data":"aGk="',
+        b'{"type":"stdout"}garbage"data":"QQ=="}',
+        b'{"type":"stdout"garbage,"data":"QQ=="}',
+        b'{"type":"stdout":,"data":"QQ=="}',
+        b'{"type":"stdout",garbage"data":"QQ=="}',
+        b'{"type":"stdout","data":"QQ==" }x',
+        b'{"type":"stdout","data":"Q\nQ=="}',
+    ],
+    ids=[
+        "trailing_bytes",
+        "control_bytes_in_base64",
+        "unterminated_frame",
+        "closed_before_data",
+        "garbage_after_tag",
+        "colon_after_tag",
+        "garbage_before_data",
+        "space_before_close",
+        "newline_in_base64",
     ],
-    ids=["trailing_bytes", "control_bytes_in_base64", "unterminated_frame"],
 )
 def test_malformed_frame_rejected(raw):
     with pytest.raises(json.JSONDecodeError):