Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 8 additions & 0 deletions docs/guides/client-pool.md
Original file line number Diff line number Diff line change
Expand Up @@ -64,6 +64,14 @@ Health is tracked separately as `HEALTHY | DEGRADED | DRAINING | STOPPED`; after
exponential backoff before retrying warmup. Callers do not need to observe these states
directly — `snapshot()` exposes them for diagnostics.

The Kotlin pool treats HTTP 429 warmup responses as server back-pressure rather than
ordinary create failures. New warmups pause for the server's `Retry-After` duration (capped
at 60 seconds), or 10 seconds when the header is missing, zero, or otherwise non-positive,
while idle maintenance remains active. This local throttle does not increment the degraded
failure count and resets when the pool instance is restarted. While the throttle is active,
`snapshot().backoffActive` is also true so operators can see that creates are paused even
though `failureCount` / `lastError` stay unchanged.

![Client pool lifecycle state machine](/images/client-pool-lifecycle.svg)

### There is no `release()`
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -24,7 +24,7 @@ package com.alibaba.opensandbox.sandbox.domain.pool
* @property idleCount Number of idle sandboxes in the store.
* @property maxIdle Current max idle target visible to this pool.
* @property failureCount Number of consecutive reconcile failures currently tracked.
* @property backoffActive Whether reconcile create attempts are currently suppressed by backoff.
* @property backoffActive Whether reconcile create attempts are currently suppressed by degraded backoff or an active warmup rate-limit throttle.
* @property lastError Last error message if pool is DEGRADED or after failure; null otherwise.
* @property inFlightOperations Number of pool operations currently in flight on this node.
*/
Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,64 @@
/*
* Copyright 2026 Alibaba Group Holding Ltd.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/

package com.alibaba.opensandbox.sandbox.infrastructure.pool

import com.alibaba.opensandbox.sandbox.transport.RETRY_AFTER_CAP
import java.time.Duration
import java.time.Instant

/** Per-run warmup throttle established by rate-limited sandbox creates. */
internal class PoolRateLimitState(
private val defaultDelay: Duration = DEFAULT_RATE_LIMIT_DELAY,
private val maxDelay: Duration = RETRY_AFTER_CAP,
) {
init {
require(!defaultDelay.isNegative) { "defaultDelay must not be negative" }
require(!maxDelay.isNegative) { "maxDelay must not be negative" }
}

@Volatile
private var throttleUntil: Instant? = null

/** Extends, but never shortens, the current throttle deadline. */
@Synchronized
fun recordRateLimit(
retryAfter: Duration?,
now: Instant = Instant.now(),
) {
val requestedDelay = retryAfter?.takeUnless { it.isNegative || it.isZero } ?: defaultDelay
val candidate = now.plus(minOf(requestedDelay, maxDelay))
val current = throttleUntil
if (current == null || candidate.isAfter(current)) {
throttleUntil = candidate
}
}

fun isActive(now: Instant = Instant.now()): Boolean {
val until = throttleUntil ?: return false
return now.isBefore(until)
}

fun remainingDelay(now: Instant = Instant.now()): Duration {
val until = throttleUntil ?: return Duration.ZERO
val remaining = Duration.between(now, until)
return if (remaining.isNegative || remaining.isZero) Duration.ZERO else remaining
}

companion object {
internal val DEFAULT_RATE_LIMIT_DELAY: Duration = Duration.ofSeconds(10)
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -44,6 +44,7 @@ internal object PoolReconciler {
onDiscardSandbox: (String) -> Unit = {},
reconcileState: ReconcileState,
warmingCount: Int,
rateLimitState: PoolRateLimitState? = null,
submitWarmups: (Int) -> Unit,
): Boolean {
val poolName = config.poolName
Expand All @@ -60,6 +61,7 @@ internal object PoolReconciler {
onDiscardSandbox = onDiscardSandbox,
reconcileState = reconcileState,
warmingCount = warmingCount,
rateLimitState = rateLimitState,
submitWarmups = submitWarmups,
)
// Do not release primary lock here; leader holds until renew fails or TTL expires.
Expand All @@ -72,6 +74,7 @@ internal object PoolReconciler {
onDiscardSandbox: (String) -> Unit,
reconcileState: ReconcileState,
warmingCount: Int,
rateLimitState: PoolRateLimitState?,
submitWarmups: (Int) -> Unit,
) {
val poolName = config.poolName
Expand Down Expand Up @@ -101,17 +104,20 @@ internal object PoolReconciler {
warmupConcurrency = config.warmupConcurrency,
)

if (plan.toSubmit == 0 || reconcileState.isBackoffActive(now)) {
val degradedBackoffActive = reconcileState.isBackoffActive(now)
val rateLimitActive = rateLimitState?.isActive(now) == true
if (plan.toSubmit == 0 || degradedBackoffActive || rateLimitActive) {
stateStore.renewPrimaryLock(poolName, ownerId, ttl)
logger.debug(
"Reconcile tick: pool_name={} idle={} warming={} deficit={} available_slots={} " +
"to_submit=0 backoff={}",
"to_submit=0 backoff={} rate_limited={}",
poolName,
counters.idleCount,
warmingCount,
plan.deficit,
plan.availableSlots,
reconcileState.isBackoffActive(now),
degradedBackoffActive,
rateLimitActive,
)
return
}
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -24,6 +24,7 @@ import com.alibaba.opensandbox.sandbox.domain.exceptions.PoolDestroyedException
import com.alibaba.opensandbox.sandbox.domain.exceptions.PoolEmptyException
import com.alibaba.opensandbox.sandbox.domain.exceptions.PoolNotRunningException
import com.alibaba.opensandbox.sandbox.domain.exceptions.PoolStateStoreUnavailableException
import com.alibaba.opensandbox.sandbox.domain.exceptions.SandboxRateLimitException
import com.alibaba.opensandbox.sandbox.domain.pool.AcquirePolicy
import com.alibaba.opensandbox.sandbox.domain.pool.IdleEntry
import com.alibaba.opensandbox.sandbox.domain.pool.PoolConfig
Expand All @@ -36,6 +37,7 @@ import com.alibaba.opensandbox.sandbox.domain.pool.PoolStateStore
import com.alibaba.opensandbox.sandbox.domain.pool.PooledSandboxCreateContext
import com.alibaba.opensandbox.sandbox.domain.pool.PooledSandboxCreator
import com.alibaba.opensandbox.sandbox.domain.pool.SandboxPreparer
import com.alibaba.opensandbox.sandbox.infrastructure.pool.PoolRateLimitState
import com.alibaba.opensandbox.sandbox.infrastructure.pool.PoolReconciler
import com.alibaba.opensandbox.sandbox.infrastructure.pool.ReconcileState
import com.alibaba.opensandbox.sandbox.internal.PoolTracer
Expand Down Expand Up @@ -643,7 +645,7 @@ class SandboxPool internal constructor(
idleCount = counters.idleCount,
maxIdle = resolveMaxIdle(),
failureCount = reconcileState.failureCount,
backoffActive = reconcileState.isBackoffActive(),
backoffActive = reconcileState.isBackoffActive() || currentRun?.rateLimitState?.isActive() == true,
lastError = reconcileState.lastError,
Comment thread
Pangjiping marked this conversation as resolved.
inFlightOperations = currentRun?.inFlightOperations?.get() ?: 0,
)
Expand Down Expand Up @@ -962,6 +964,7 @@ class SandboxPool internal constructor(
onDiscardSandbox = { sandboxId -> killSandboxBestEffort(sandboxId) },
reconcileState = reconcileState,
warmingCount = run.warmingCount.get(),
rateLimitState = run.rateLimitState,
submitWarmups = { count -> submitWarmups(run, count) },
),
)
Expand Down Expand Up @@ -1043,6 +1046,57 @@ class SandboxPool internal constructor(
}
}

private fun scheduleRateLimitReconcile(run: RunContext) {
if (!isCurrentRun(run) || lifecycleState.get() != LifecycleState.RUNNING) return
synchronized(run.rateLimitScheduleLock) {
if (!isCurrentRun(run) || lifecycleState.get() != LifecycleState.RUNNING) return
run.rateLimitReconcileTask?.cancel(false)
val sequence = ++run.rateLimitReconcileSequence
val delayNanos = run.rateLimitState.remainingDelay().toNanos()
try {
run.rateLimitReconcileTask =
run.scheduler.schedule(
{ onRateLimitReconcileDue(run, sequence) },
delayNanos,
TimeUnit.NANOSECONDS,
)
} catch (e: Exception) {
run.rateLimitReconcileTask = null
if (lifecycleState.get() == LifecycleState.RUNNING) {
logger.debug(
"Pool rate-limit reconcile submit rejected: pool_name={} error={}",
config.poolName,
e.message,
)
}
}
}
}

private fun onRateLimitReconcileDue(
run: RunContext,
sequence: Long,
) {
synchronized(run.rateLimitScheduleLock) {
if (sequence != run.rateLimitReconcileSequence) return
run.rateLimitReconcileTask = null
}
if (!isCurrentRun(run) || lifecycleState.get() != LifecycleState.RUNNING) return
if (run.rateLimitState.isActive()) {
scheduleRateLimitReconcile(run)
} else {
requestReconcile(run)
}
}

private fun cancelRateLimitReconcile(run: RunContext) {
synchronized(run.rateLimitScheduleLock) {
run.rateLimitReconcileSequence++
run.rateLimitReconcileTask?.cancel(false)
run.rateLimitReconcileTask = null
}
}

private fun submitWarmups(
run: RunContext,
count: Int,
Expand Down Expand Up @@ -1214,7 +1268,19 @@ class SandboxPool internal constructor(
is WarmupOutcome.Success -> commitWarmupSandbox(run, outcome.sandboxId, trace)
is WarmupOutcome.Failure -> {
if (isCurrentRun(run) && lifecycleState.get() == LifecycleState.RUNNING) {
reconcileState.recordAsyncFailure(outcome.error.message)
val error = outcome.error
if (error is SandboxRateLimitException) {
run.rateLimitState.recordRateLimit(error.retryAfter)
scheduleRateLimitReconcile(run)
Comment thread
Pangjiping marked this conversation as resolved.
logger.debug(
"Pool warmup rate limited: pool_name={} retry_after_ms={} throttle_remaining_ms={}",
config.poolName,
error.retryAfter?.toMillis(),
run.rateLimitState.remainingDelay().toMillis(),
)
} else {
reconcileState.recordAsyncFailure(error.message)
}
}
}
WarmupOutcome.Cancelled -> Unit
Expand Down Expand Up @@ -1818,6 +1884,7 @@ class SandboxPool internal constructor(
} finally {
run.commitLock.unlock()
}
cancelRateLimitReconcile(run)
}

/**
Expand All @@ -1837,6 +1904,12 @@ class SandboxPool internal constructor(
val warmingCount = AtomicInteger(0)
val warmupSubmissionsOpen = AtomicBoolean(true)
val reconcileQueued = AtomicBoolean(false)
val rateLimitState = PoolRateLimitState()
val rateLimitScheduleLock = Any()

@Volatile
var rateLimitReconcileTask: ScheduledFuture<*>? = null
var rateLimitReconcileSequence: Long = 0

@Volatile
var nextCompletionReconcileAtNanos: Long = 0
Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,111 @@
/*
* Copyright 2026 Alibaba Group Holding Ltd.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/

package com.alibaba.opensandbox.sandbox.infrastructure.pool

import com.alibaba.opensandbox.sandbox.config.ConnectionConfig
import com.alibaba.opensandbox.sandbox.domain.pool.PoolConfig
import com.alibaba.opensandbox.sandbox.domain.pool.PoolCreationSpec
import org.junit.jupiter.api.Assertions.assertEquals
import org.junit.jupiter.api.Assertions.assertFalse
import org.junit.jupiter.api.Assertions.assertTrue
import org.junit.jupiter.api.Test
import java.time.Duration
import java.time.Instant
import java.util.concurrent.atomic.AtomicInteger

class PoolRateLimitStateTest {
private val now: Instant = Instant.parse("2026-08-14T00:00:00Z")

@Test
fun `zero or non-positive retry after falls back to default delay`() {
val state = PoolRateLimitState()

state.recordRateLimit(retryAfter = Duration.ZERO, now = now)

assertTrue(state.isActive(now.plusSeconds(9)))
assertFalse(state.isActive(now.plusSeconds(10)))
}

@Test
fun `missing retry after uses bounded default delay`() {
val state = PoolRateLimitState()

state.recordRateLimit(retryAfter = null, now = now)

assertTrue(state.isActive(now.plusSeconds(9)))
assertFalse(state.isActive(now.plusSeconds(10)))
}

@Test
fun `retry after is capped at transport ceiling`() {
val state = PoolRateLimitState()

state.recordRateLimit(retryAfter = Duration.ofMinutes(5), now = now)

assertTrue(state.isActive(now.plusSeconds(59)))
assertFalse(state.isActive(now.plusSeconds(60)))
}

@Test
fun `concurrent rate limits only extend throttle deadline`() {
val state = PoolRateLimitState()

state.recordRateLimit(retryAfter = Duration.ofSeconds(30), now = now)
state.recordRateLimit(retryAfter = Duration.ofSeconds(5), now = now.plusSeconds(1))

assertEquals(Duration.ofSeconds(1), state.remainingDelay(now.plusSeconds(29)))
state.recordRateLimit(retryAfter = Duration.ofSeconds(60), now = now.plusSeconds(1))
assertTrue(state.isActive(now.plusSeconds(60)))
assertFalse(state.isActive(now.plusSeconds(61)))
}

@Test
fun `rate limit suppresses warmups without blocking excess idle shrink`() {
val stateStore = InMemoryPoolStateStore()
val poolName = "rate-limited-shrink"
stateStore.putIdle(poolName, "idle-1")
stateStore.putIdle(poolName, "idle-2")
val config =
PoolConfig.builder()
.poolName(poolName)
.ownerId("owner-1")
.maxIdle(1)
.warmupConcurrency(1)
.stateStore(stateStore)
.connectionConfig(ConnectionConfig.builder().build())
.creationSpec(PoolCreationSpec.builder().image("ubuntu:22.04").build())
.build()
val rateLimitState = PoolRateLimitState()
rateLimitState.recordRateLimit(Duration.ofSeconds(30))
val discarded = mutableListOf<String>()
val submitted = AtomicInteger(0)

PoolReconciler.runReconcileTick(
config = config,
stateStore = stateStore,
onDiscardSandbox = { discarded += it },
reconcileState = ReconcileState(degradedThreshold = 3),
warmingCount = 0,
rateLimitState = rateLimitState,
submitWarmups = { submitted.addAndGet(it) },
)

assertEquals(1, discarded.size)
assertEquals(0, submitted.get())
assertEquals(1, stateStore.snapshotCounters(poolName).idleCount)
}
}
Loading
Loading