diff --git a/docs/guides/client-pool.md b/docs/guides/client-pool.md index f9b36a431..17a4b8523 100644 --- a/docs/guides/client-pool.md +++ b/docs/guides/client-pool.md @@ -54,6 +54,12 @@ 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 unavailable, while idle maintenance remains +active. This local throttle does not increment the degraded failure count and resets when the +pool instance is restarted. + ![Client pool lifecycle state machine](/images/client-pool-lifecycle.svg) ### There is no `release()` diff --git a/sdks/sandbox/kotlin/sandbox/src/main/kotlin/com/alibaba/opensandbox/sandbox/infrastructure/pool/PoolRateLimitState.kt b/sdks/sandbox/kotlin/sandbox/src/main/kotlin/com/alibaba/opensandbox/sandbox/infrastructure/pool/PoolRateLimitState.kt new file mode 100644 index 000000000..628a1af20 --- /dev/null +++ b/sdks/sandbox/kotlin/sandbox/src/main/kotlin/com/alibaba/opensandbox/sandbox/infrastructure/pool/PoolRateLimitState.kt @@ -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 } ?: 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) + } +} diff --git a/sdks/sandbox/kotlin/sandbox/src/main/kotlin/com/alibaba/opensandbox/sandbox/infrastructure/pool/PoolReconciler.kt b/sdks/sandbox/kotlin/sandbox/src/main/kotlin/com/alibaba/opensandbox/sandbox/infrastructure/pool/PoolReconciler.kt index f8bf0aa0c..e2975d4ac 100644 --- a/sdks/sandbox/kotlin/sandbox/src/main/kotlin/com/alibaba/opensandbox/sandbox/infrastructure/pool/PoolReconciler.kt +++ b/sdks/sandbox/kotlin/sandbox/src/main/kotlin/com/alibaba/opensandbox/sandbox/infrastructure/pool/PoolReconciler.kt @@ -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 @@ -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. @@ -72,6 +74,7 @@ internal object PoolReconciler { onDiscardSandbox: (String) -> Unit, reconcileState: ReconcileState, warmingCount: Int, + rateLimitState: PoolRateLimitState?, submitWarmups: (Int) -> Unit, ) { val poolName = config.poolName @@ -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 } diff --git a/sdks/sandbox/kotlin/sandbox/src/main/kotlin/com/alibaba/opensandbox/sandbox/pool/SandboxPool.kt b/sdks/sandbox/kotlin/sandbox/src/main/kotlin/com/alibaba/opensandbox/sandbox/pool/SandboxPool.kt index dcbd7a38c..a14a89c55 100644 --- a/sdks/sandbox/kotlin/sandbox/src/main/kotlin/com/alibaba/opensandbox/sandbox/pool/SandboxPool.kt +++ b/sdks/sandbox/kotlin/sandbox/src/main/kotlin/com/alibaba/opensandbox/sandbox/pool/SandboxPool.kt @@ -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 @@ -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.isCausedByInterruption @@ -598,7 +600,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, inFlightOperations = currentRun?.inFlightOperations?.get() ?: 0, ) @@ -914,6 +916,7 @@ class SandboxPool internal constructor( onDiscardSandbox = { sandboxId -> killSandboxBestEffort(sandboxId) }, reconcileState = reconcileState, warmingCount = run.warmingCount.get(), + rateLimitState = run.rateLimitState, submitWarmups = { count -> submitWarmups(run, count) }, ), ) @@ -995,6 +998,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, @@ -1129,7 +1183,19 @@ class SandboxPool internal constructor( is WarmupOutcome.Success -> commitWarmupSandbox(run, outcome.sandboxId) 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) + 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 @@ -1700,6 +1766,7 @@ class SandboxPool internal constructor( } finally { run.commitLock.unlock() } + cancelRateLimitReconcile(run) } /** @@ -1719,6 +1786,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 diff --git a/sdks/sandbox/kotlin/sandbox/src/test/kotlin/com/alibaba/opensandbox/sandbox/infrastructure/pool/PoolRateLimitStateTest.kt b/sdks/sandbox/kotlin/sandbox/src/test/kotlin/com/alibaba/opensandbox/sandbox/infrastructure/pool/PoolRateLimitStateTest.kt new file mode 100644 index 000000000..99afee577 --- /dev/null +++ b/sdks/sandbox/kotlin/sandbox/src/test/kotlin/com/alibaba/opensandbox/sandbox/infrastructure/pool/PoolRateLimitStateTest.kt @@ -0,0 +1,101 @@ +/* + * 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 `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() + 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) + } +} diff --git a/sdks/sandbox/kotlin/sandbox/src/test/kotlin/com/alibaba/opensandbox/sandbox/pool/SandboxPoolRateLimitTest.kt b/sdks/sandbox/kotlin/sandbox/src/test/kotlin/com/alibaba/opensandbox/sandbox/pool/SandboxPoolRateLimitTest.kt new file mode 100644 index 000000000..b1c01c663 --- /dev/null +++ b/sdks/sandbox/kotlin/sandbox/src/test/kotlin/com/alibaba/opensandbox/sandbox/pool/SandboxPoolRateLimitTest.kt @@ -0,0 +1,151 @@ +/* + * Copyright 2025 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.pool + +import com.alibaba.opensandbox.sandbox.Sandbox +import com.alibaba.opensandbox.sandbox.config.ConnectionConfig +import com.alibaba.opensandbox.sandbox.domain.exceptions.SandboxRateLimitException +import com.alibaba.opensandbox.sandbox.domain.pool.PoolCreationSpec +import com.alibaba.opensandbox.sandbox.domain.pool.PoolState +import com.alibaba.opensandbox.sandbox.domain.pool.PooledSandboxCreator +import com.alibaba.opensandbox.sandbox.infrastructure.pool.InMemoryPoolStateStore +import io.mockk.every +import io.mockk.mockk +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.util.concurrent.TimeUnit +import java.util.concurrent.atomic.AtomicInteger +import java.util.concurrent.atomic.AtomicLong + +class SandboxPoolRateLimitTest { + @Test + fun `rate limited warmup honors retry after without degrading pool`() { + val attempts = AtomicInteger(0) + val firstAttemptAt = AtomicLong(0) + val secondAttemptAt = AtomicLong(0) + val sandbox = mockk(relaxed = true) + every { sandbox.id } returns "rate-limit-recovery" + + val pool = + SandboxPool.builder() + .poolName("rate-limited-pool") + .ownerId("rate-limited-owner") + .maxIdle(1) + .warmupConcurrency(1) + .degradedThreshold(1) + .stateStore(InMemoryPoolStateStore()) + .connectionConfig(ConnectionConfig.builder().build()) + .creationSpec(PoolCreationSpec.builder().image("ubuntu:22.04").build()) + .sandboxCreator( + PooledSandboxCreator { + when (attempts.incrementAndGet()) { + 1 -> { + firstAttemptAt.set(System.nanoTime()) + throw SandboxRateLimitException( + message = "rate limited", + retryAfter = Duration.ofSeconds(1), + ) + } + else -> { + secondAttemptAt.compareAndSet(0, System.nanoTime()) + sandbox + } + } + }, + ).warmupSkipHealthCheck() + .reconcileInterval(Duration.ofSeconds(30)) + .build() + + pool.start() + try { + awaitCondition { pool.snapshot().backoffActive } + + val throttled = pool.snapshot() + assertEquals(PoolState.HEALTHY, throttled.state) + assertEquals(0, throttled.failureCount) + assertFalse( + awaitCondition(timeout = Duration.ofMillis(250)) { attempts.get() > 1 }, + "warmup must not retry before Retry-After expires", + ) + + assertTrue(awaitCondition { pool.snapshot().idleCount == 1 }) + val retryDelay = Duration.ofNanos(secondAttemptAt.get() - firstAttemptAt.get()) + assertTrue(retryDelay >= Duration.ofMillis(800), "warmup retried too early: $retryDelay") + assertEquals(2, attempts.get()) + assertFalse(pool.snapshot().backoffActive) + } finally { + pool.shutdown(graceful = false) + } + } + + @Test + fun `pool restart clears rate limit throttle from previous run`() { + val attempts = AtomicInteger(0) + val sandbox = mockk(relaxed = true) + every { sandbox.id } returns "restart-rate-limit-recovery" + val pool = + SandboxPool.builder() + .poolName("restart-rate-limited-pool") + .ownerId("restart-rate-limited-owner") + .maxIdle(1) + .warmupConcurrency(1) + .stateStore(InMemoryPoolStateStore()) + .connectionConfig(ConnectionConfig.builder().build()) + .creationSpec(PoolCreationSpec.builder().image("ubuntu:22.04").build()) + .sandboxCreator( + PooledSandboxCreator { + if (attempts.incrementAndGet() == 1) { + throw SandboxRateLimitException( + message = "rate limited", + retryAfter = Duration.ofSeconds(30), + ) + } + sandbox + }, + ).warmupSkipHealthCheck() + .reconcileInterval(Duration.ofSeconds(30)) + .build() + + pool.start() + assertTrue(awaitCondition { pool.snapshot().backoffActive }) + pool.shutdown(graceful = false) + + pool.start() + try { + assertFalse(pool.snapshot().backoffActive) + assertTrue(awaitCondition { pool.snapshot().idleCount == 1 }) + assertEquals(2, attempts.get()) + } finally { + pool.shutdown(graceful = false) + } + } + + private fun awaitCondition( + timeout: Duration = Duration.ofSeconds(5), + condition: () -> Boolean, + ): Boolean { + val deadline = System.nanoTime() + timeout.toNanos() + while (System.nanoTime() < deadline) { + if (condition()) return true + TimeUnit.MILLISECONDS.sleep(10) + } + return condition() + } +}