refactor: simplify the init-concurrency limiter to a single shared budget

config/config.go

@@ -161,14 +161,11 @@ type ConcurrencyConfig struct {
 	// of queueing them.
 	MaxQueued ct.OptInt `conf:"INIT_MAX_QUEUED"`
 
-	// PerEnvMaxPercent limits the share of the budget that any single environment may
-	// use, as a percentage of MaxConcurrent plus MaxQueued. This keeps one busy
-	// environment from starving the others. A value of 0 applies no per-environment
-	// limit.
-	PerEnvMaxPercent ct.OptInt `conf:"INIT_PER_ENV_MAX_PERCENT"`
-
-	// SendTimeout releases a delivery slot if a streaming initialization payload cannot
-	// make progress to its client within this duration. It defaults to 30s.
+	// SendTimeout is the absolute cap on how long a single gated delivery may hold a slot. A
+	// throughput floor (64 KB/s) closes a client that stalls or is slower than the floor well
+	// before this; the cap only backstops a client stuck right at the floor on a very large
+	// payload. If a delivery exceeds it, the connection is closed to reclaim the slot (and the
+	// SDK reconnects). It defaults to 2m.
 	SendTimeout ct.OptDuration `conf:"INIT_SEND_TIMEOUT"`
 }
 

config/config_concurrency_test.go

@@ -13,16 +13,14 @@ import (
 // the INIT_* environment variables.
 func TestConcurrencyConfigFromEnvironment(t *testing.T) {
 	withEnvironment(map[string]string{
-		"INIT_MAX_CONCURRENT":      "200",
-		"INIT_MAX_QUEUED":          "1000",
-		"INIT_PER_ENV_MAX_PERCENT": "40",
-		"INIT_SEND_TIMEOUT":        "15s",
+		"INIT_MAX_CONCURRENT": "200",
+		"INIT_MAX_QUEUED":     "1000",
+		"INIT_SEND_TIMEOUT":   "15s",
 	}, func() {
 		var c Config
 		require.NoError(t, LoadConfigFromEnvironment(&c, slog.Default()))
 		assert.Equal(t, 200, c.Concurrency.MaxConcurrent.GetOrElse(-1))
 		assert.Equal(t, 1000, c.Concurrency.MaxQueued.GetOrElse(-1))
-		assert.Equal(t, 40, c.Concurrency.PerEnvMaxPercent.GetOrElse(-1))
 		assert.Equal(t, 15*time.Second, c.Concurrency.SendTimeout.GetOrElse(0))
 	})
 }

internal/concurrency/limiter.go

@@ -1,8 +1,7 @@
 // Package concurrency provides an admission limiter that bounds how much concurrent
 // work a burst of requests or connections can impose on Relay. It uses two limits: a
 // maximum number of slots held at once, and a bounded FIFO queue of callers waiting for
-// a slot. An optional per-environment gate keeps one environment from using the whole
-// budget.
+// a slot.
 package concurrency
 
 import (
@@ -19,13 +18,9 @@ type Params struct {
 	// MaxQueued is the number of callers that may wait in FIFO order for a slot once all
 	// slots are held. A value of 0 rejects callers immediately instead of waiting.
 	MaxQueued int
-	// PerEnvMax limits how many of a single environment's callers may participate,
-	// counting both held and waiting, at once. A value of 0 applies no per-environment
-	// limit. The per-environment gate never blocks; it only rejects.
-	PerEnvMax int
 }
 
-// Stats is a point-in-time snapshot of a Limiter's counters, for logging/metrics.
+// Stats is a point-in-time snapshot of a Limiter's counters for logging and metrics.
 type Stats struct {
 	Enabled       bool
 	MaxConcurrent int
@@ -36,30 +31,23 @@ type Stats struct {
 	Rejected      int64
 }
 
-// Limiter bounds concurrency with two limits plus an optional per-environment gate that
-// never blocks and only rejects. The zero value is not usable; construct one with New.
+// Limiter bounds concurrency with two limits: a maximum number of slots held at once and
+// a bounded FIFO queue of waiters. The zero value is not usable; construct one with New.
 type Limiter struct {
 	name    string
 	enabled bool
 
-	tokens    chan struct{} // holds MaxConcurrent slots; receive to acquire, send to release
+	tokens    chan struct{} // holds the free slots; receive to acquire one, send to release one
 	maxQueued int64
 	waiting   int64
 	shutdown  chan struct{}
 	closeOnce sync.Once
 
-	perEnvMax int
-	perEnv    sync.Map // envKey -> *envGate
-
 	admitted atomic.Int64
 	rejected atomic.Int64
 }
 
-type envGate struct {
-	slots chan struct{} // cap = perEnvMax; try-receive to enter, send to leave
-}
-
-// New builds a Limiter. name is used only for logging/metrics identification.
+// New builds a Limiter. name identifies the limiter in logs and metrics.
 func New(name string, p Params) *Limiter {
 	l := &Limiter{name: name}
 	if p.MaxConcurrent <= 0 {
@@ -72,9 +60,6 @@ func New(name string, p Params) *Limiter {
 	}
 	l.maxQueued = int64(p.MaxQueued)
 	l.shutdown = make(chan struct{})
-	if p.PerEnvMax > 0 {
-		l.perEnvMax = p.PerEnvMax
-	}
 	return l
 }
 
@@ -84,84 +69,51 @@ func (l *Limiter) Name() string { return l.name }
 // Enabled reports whether the limiter is enforcing a limit.
 func (l *Limiter) Enabled() bool { return l != nil && l.enabled }
 
-// Acquire attempts to admit one unit of work for the given environment key.
-// On success it returns a release func (call exactly once) and ok=true. If the
-// per-env gate or the global backlog is full, or ctx is cancelled, or the
-// limiter is shut down, it returns a no-op release and ok=false. A disabled or
-// nil limiter always admits immediately.
-func (l *Limiter) Acquire(ctx context.Context, envKey string) (release func(), ok bool) {
+// Acquire attempts to admit one unit of work. On success it returns a release function,
+// which the caller must call exactly once, and ok is true. It returns a no-op release and
+// ok=false if the queue is full, ctx is cancelled, or the limiter is shut down. A disabled
+// or nil limiter always admits immediately.
+func (l *Limiter) Acquire(ctx context.Context) (release func(), ok bool) {
 	if !l.Enabled() {
 		return func() {}, true
 	}
 
-	// Per-environment gate. It never blocks; it rejects when the environment is over its share.
-	var releaseEnv func()
-	if l.perEnvMax > 0 {
-		g := l.gateFor(envKey)
-		select {
-		case g.slots <- struct{}{}:
-			releaseEnv = func() { <-g.slots }
-		default:
-			l.rejected.Add(1)
-			return func() {}, false
-		}
-	}
-
-	reject := func() (func(), bool) {
-		if releaseEnv != nil {
-			releaseEnv()
-		}
-		l.rejected.Add(1)
-		return func() {}, false
-	}
-
-	// Fast path: a token is immediately available.
+	// Take a free slot if one is available.
 	select {
 	case <-l.tokens:
-		return l.admit(releaseEnv), true
+		return l.admit(), true
 	default:
 	}
 
-	// No token free: enter the bounded FIFO backlog, or reject.
+	// No free slot. Join the bounded queue, or reject if it is full.
 	if atomic.AddInt64(&l.waiting, 1) > l.maxQueued {
 		atomic.AddInt64(&l.waiting, -1)
-		return reject()
+		l.rejected.Add(1)
+		return func() {}, false
 	}
 	defer atomic.AddInt64(&l.waiting, -1)
 
 	select {
 	case <-l.tokens:
-		return l.admit(releaseEnv), true
+		return l.admit(), true
 	case <-ctx.Done():
-		return reject()
+		l.rejected.Add(1)
+		return func() {}, false
 	case <-l.shutdown:
-		return reject()
+		l.rejected.Add(1)
+		return func() {}, false
 	}
 }
 
-func (l *Limiter) admit(releaseEnv func()) func() {
+func (l *Limiter) admit() func() {
 	l.admitted.Add(1)
 	var once sync.Once
 	return func() {
-		once.Do(func() {
-			l.tokens <- struct{}{}
-			if releaseEnv != nil {
-				releaseEnv()
-			}
-		})
-	}
-}
-
-func (l *Limiter) gateFor(envKey string) *envGate {
-	if g, ok := l.perEnv.Load(envKey); ok {
-		return g.(*envGate)
+		once.Do(func() { l.tokens <- struct{}{} })
 	}
-	g := &envGate{slots: make(chan struct{}, l.perEnvMax)}
-	actual, _ := l.perEnv.LoadOrStore(envKey, g)
-	return actual.(*envGate)
 }
 
-// Close unblocks all waiters (they receive ok=false). Idempotent.
+// Close unblocks all waiters, which then receive ok=false. It may be called more than once.
 func (l *Limiter) Close() {
 	if !l.Enabled() {
 		return

internal/concurrency/limiter_test.go

@@ -13,7 +13,7 @@ func TestDisabledLimiterAlwaysAdmits(t *testing.T) {
 		t.Fatal("expected disabled")
 	}
 	for i := 0; i < 100; i++ {
-		release, ok := l.Acquire(context.Background(), "env")
+		release, ok := l.Acquire(context.Background())
 		if !ok {
 			t.Fatal("disabled limiter must always admit")
 		}
@@ -23,7 +23,7 @@ func TestDisabledLimiterAlwaysAdmits(t *testing.T) {
 
 func TestNilLimiterAdmits(t *testing.T) {
 	var l *Limiter
-	release, ok := l.Acquire(context.Background(), "env")
+	release, ok := l.Acquire(context.Background())
 	if !ok {
 		t.Fatal("nil limiter must admit")
 	}
@@ -32,9 +32,9 @@ func TestNilLimiterAdmits(t *testing.T) {
 
 func TestRejectWhenNoBacklog(t *testing.T) {
 	l := New("t", Params{MaxConcurrent: 2, MaxQueued: 0})
-	r1, ok1 := l.Acquire(context.Background(), "e")
-	r2, ok2 := l.Acquire(context.Background(), "e")
-	_, ok3 := l.Acquire(context.Background(), "e")
+	r1, ok1 := l.Acquire(context.Background())
+	r2, ok2 := l.Acquire(context.Background())
+	_, ok3 := l.Acquire(context.Background())
 	if !ok1 || !ok2 {
 		t.Fatal("first two should be admitted")
 	}
@@ -50,13 +50,13 @@ func TestRejectWhenNoBacklog(t *testing.T) {
 
 func TestQueueThenAdmitOnRelease(t *testing.T) {
 	l := New("t", Params{MaxConcurrent: 1, MaxQueued: 1})
-	r1, ok1 := l.Acquire(context.Background(), "e")
+	r1, ok1 := l.Acquire(context.Background())
 	if !ok1 {
 		t.Fatal("first should be admitted")
 	}
 	admitted := make(chan struct{})
 	go func() {
-		r2, ok2 := l.Acquire(context.Background(), "e") // must queue, then admit when r1 releases
+		r2, ok2 := l.Acquire(context.Background()) // must queue, then admit when r1 releases
 		if ok2 {
 			close(admitted)
 			r2()
@@ -79,36 +79,23 @@ func TestQueueThenAdmitOnRelease(t *testing.T) {
 
 func TestBacklogFullRejects(t *testing.T) {
 	l := New("t", Params{MaxConcurrent: 1, MaxQueued: 1})
-	r1, _ := l.Acquire(context.Background(), "e") // holds the only token
+	r1, _ := l.Acquire(context.Background()) // holds the only token
 	defer r1()
-	go l.Acquire(context.Background(), "e") // fills the single backlog slot
+	go l.Acquire(context.Background()) // fills the single backlog slot
 	time.Sleep(50 * time.Millisecond)
-	if _, ok := l.Acquire(context.Background(), "e"); ok {
+	if _, ok := l.Acquire(context.Background()); ok {
 		t.Fatal("expected rejection when backlog is full")
 	}
 }
 
-func TestPerEnvGateIsolatesEnvironments(t *testing.T) {
-	// Global room for 10, but each env may hold at most 1 (participant cap).
-	l := New("t", Params{MaxConcurrent: 10, MaxQueued: 10, PerEnvMax: 1})
-	rA, okA := l.Acquire(context.Background(), "A")
-	_, okA2 := l.Acquire(context.Background(), "A") // second A rejected by per-env gate
-	rB, okB := l.Acquire(context.Background(), "B") // different env still admitted
-	if !okA || okA2 || !okB {
-		t.Fatalf("per-env gate failed: okA=%v okA2=%v okB=%v", okA, okA2, okB)
-	}
-	rA()
-	rB()
-}
-
 func TestContextCancelUnblocksWaiter(t *testing.T) {
 	l := New("t", Params{MaxConcurrent: 1, MaxQueued: 5})
-	r1, _ := l.Acquire(context.Background(), "e")
+	r1, _ := l.Acquire(context.Background())
 	defer r1()
 	ctx, cancel := context.WithCancel(context.Background())
 	done := make(chan bool)
 	go func() {
-		_, ok := l.Acquire(ctx, "e")
+		_, ok := l.Acquire(ctx)
 		done <- ok
 	}()
 	time.Sleep(50 * time.Millisecond)
@@ -125,15 +112,15 @@ func TestContextCancelUnblocksWaiter(t *testing.T) {
 
 func TestReleaseIsIdempotent(t *testing.T) {
 	l := New("t", Params{MaxConcurrent: 1, MaxQueued: 0})
-	r, _ := l.Acquire(context.Background(), "e")
+	r, _ := l.Acquire(context.Background())
 	r()
 	r() // must not release a second token
 	// Two acquires should now succeed sequentially, proving only one token exists.
-	r1, ok1 := l.Acquire(context.Background(), "e")
+	r1, ok1 := l.Acquire(context.Background())
 	if !ok1 {
 		t.Fatal("expected admit after release")
 	}
-	if _, ok2 := l.Acquire(context.Background(), "e"); ok2 {
+	if _, ok2 := l.Acquire(context.Background()); ok2 {
 		t.Fatal("double release leaked a token")
 	}
 	r1()
@@ -149,7 +136,7 @@ func TestConcurrentAcquireBoundsHeld(t *testing.T) {
 		wg.Add(1)
 		go func() {
 			defer wg.Done()
-			release, ok := l.Acquire(context.Background(), "e")
+			release, ok := l.Acquire(context.Background())
 			if !ok {
 				return
 			}