feat(sdks/kotlin): add acquireMinRemainingTtl to pool to skip near-ex…

sdks/sandbox/kotlin/sandbox-pool-redis/src/main/kotlin/com/alibaba/opensandbox/sandbox/infrastructure/pool/RedisPoolStateStore.kt

@@ -49,13 +49,29 @@ class RedisPoolStateStore(
 ) : PoolStateStore {
     private val defaultIdleTtl: Duration = Duration.ofHours(24)
 
-    override fun tryTakeIdle(poolName: String): String? =
+    override fun tryTakeIdle(poolName: String): String? = takeIdle(poolName, "0")
+
+    override fun tryTakeIdle(
+        poolName: String,
+        minRemainingTtl: Duration,
+    ): String? {
+        if (minRemainingTtl.isNegative || minRemainingTtl.isZero) {
+            return tryTakeIdle(poolName)
+        }
+        val minRemainingTtlMs = minRemainingTtl.toMillis().coerceAtLeast(0).toString()
+        return takeIdle(poolName, minRemainingTtlMs)
+    }
+
+    private fun takeIdle(
+        poolName: String,
+        minRemainingTtlMs: String,
+    ): String? =
         execute("tryTakeIdle", poolName) {
             val result =
                 redis.eval(
                     TAKE_IDLE_SCRIPT,
                     listOf(idleListKey(poolName), idleExpiresKey(poolName)),
-                    emptyList<String>(),
+                    listOf(minRemainingTtlMs),
                 )
             result as? String
         }
@@ -131,12 +147,31 @@ class RedisPoolStateStore(
     override fun reapExpiredIdle(
         poolName: String,
         now: Instant,
+    ) {
+        reapIdle(poolName, "0")
+    }
+
+    override fun reapExpiredIdle(
+        poolName: String,
+        now: Instant,
+        minRemainingTtl: Duration,
+    ) {
+        if (minRemainingTtl.isNegative || minRemainingTtl.isZero) {
+            reapIdle(poolName, "0")
+            return
+        }
+        reapIdle(poolName, minRemainingTtl.toMillis().coerceAtLeast(0).toString())
+    }
+
+    private fun reapIdle(
+        poolName: String,
+        minRemainingTtlMs: String,
     ) {
         execute("reapExpiredIdle", poolName) {
             redis.eval(
                 REAP_EXPIRED_SCRIPT,
                 listOf(idleListKey(poolName), idleExpiresKey(poolName)),
-                emptyList<String>(),
+                listOf(minRemainingTtlMs),
             )
         }
     }
@@ -240,6 +275,8 @@ class RedisPoolStateStore(
             """
             local redis_time = redis.call('TIME')
             local now_ms = tonumber(redis_time[1]) * 1000 + math.floor(tonumber(redis_time[2]) / 1000)
+            local min_remaining_ttl_ms = tonumber(ARGV[1]) or 0
+            local cutoff_ms = now_ms + min_remaining_ttl_ms
             while true do
               local sandbox_id = redis.call('LPOP', KEYS[1])
               if not sandbox_id then
@@ -248,7 +285,7 @@ class RedisPoolStateStore(
               local expires_at = redis.call('HGET', KEYS[2], sandbox_id)
               if expires_at then
                 redis.call('HDEL', KEYS[2], sandbox_id)
-                if tonumber(expires_at) > now_ms then
+                if tonumber(expires_at) > cutoff_ms then
                   return sandbox_id
                 end
               end
@@ -292,9 +329,11 @@ class RedisPoolStateStore(
             """
             local redis_time = redis.call('TIME')
             local now_ms = tonumber(redis_time[1]) * 1000 + math.floor(tonumber(redis_time[2]) / 1000)
+            local min_remaining_ttl_ms = tonumber(ARGV[1]) or 0
+            local cutoff_ms = now_ms + min_remaining_ttl_ms
             local entries = redis.call('HGETALL', KEYS[2])
             for i = 1, #entries, 2 do
-              if tonumber(entries[i + 1]) <= now_ms then
+              if tonumber(entries[i + 1]) <= cutoff_ms then
                 redis.call('HDEL', KEYS[2], entries[i])
                 redis.call('LREM', KEYS[1], 0, entries[i])
               end

sdks/sandbox/kotlin/sandbox-pool-redis/src/test/kotlin/com/alibaba/opensandbox/sandbox/infrastructure/pool/RedisPoolStateStoreTest.kt

@@ -136,6 +136,70 @@ class RedisPoolStateStoreTest {
         assertEquals(7, stateStore.getMaxIdle(poolName))
     }
 
+    @Test
+    fun `tryTakeIdle with minRemainingTtl skips entries below the threshold`() {
+        val stateStore = requireStore()
+
+        // Entries get a 200ms TTL.
+        stateStore.setIdleEntryTtl(poolName, Duration.ofMillis(200))
+        stateStore.putIdle(poolName, "id-1")
+        stateStore.putIdle(poolName, "id-2")
+
+        // Demand more remaining TTL than the entries can possibly have.
+        // Both should be discarded; the call returns null.
+        assertNull(stateStore.tryTakeIdle(poolName, Duration.ofSeconds(60)))
+        // Discarded entries are also removed from idle membership.
+        assertEquals(0, stateStore.snapshotCounters(poolName).idleCount)
+    }
+
+    @Test
+    fun `reapExpiredIdle with minRemainingTtl evicts near-expiry entries`() {
+        val stateStore = requireStore()
+
+        stateStore.setIdleEntryTtl(poolName, Duration.ofMillis(500))
+        stateStore.putIdle(poolName, "id-1")
+        stateStore.putIdle(poolName, "id-2")
+
+        // 60s threshold against 500ms TTL → both entries are near-expiry and reaped.
+        stateStore.reapExpiredIdle(poolName, Instant.now(), Duration.ofSeconds(60))
+
+        assertEquals(0, stateStore.snapshotCounters(poolName).idleCount)
+        assertNull(stateStore.tryTakeIdle(poolName))
+    }
+
+    @Test
+    fun `reapExpiredIdle with minRemainingTtl preserves entries above the threshold`() {
+        val stateStore = requireStore()
+
+        stateStore.setIdleEntryTtl(poolName, Duration.ofMinutes(10))
+        stateStore.putIdle(poolName, "id-1")
+
+        stateStore.reapExpiredIdle(poolName, Instant.now(), Duration.ofSeconds(60))
+
+        assertEquals(1, stateStore.snapshotCounters(poolName).idleCount)
+    }
+
+    @Test
+    fun `tryTakeIdle with minRemainingTtl returns entries that satisfy the threshold`() {
+        val stateStore = requireStore()
+
+        stateStore.setIdleEntryTtl(poolName, Duration.ofMinutes(10))
+        stateStore.putIdle(poolName, "id-1")
+
+        // 10 minutes of TTL is well above a 60-second threshold.
+        assertEquals("id-1", stateStore.tryTakeIdle(poolName, Duration.ofSeconds(60)))
+    }
+
+    @Test
+    fun `tryTakeIdle with zero minRemainingTtl behaves like the base call`() {
+        val stateStore = requireStore()
+
+        stateStore.putIdle(poolName, "id-1")
+        // Duration.ZERO must not change behavior.
+        assertEquals("id-1", stateStore.tryTakeIdle(poolName, Duration.ZERO))
+        assertNull(stateStore.tryTakeIdle(poolName, Duration.ZERO))
+    }
+
     @Test
     fun `setIdleEntryTtl validates positive duration`() {
         val stateStore = requireStore()

sdks/sandbox/kotlin/sandbox/src/main/kotlin/com/alibaba/opensandbox/sandbox/domain/pool/PoolConfig.kt

@@ -41,6 +41,12 @@ import kotlin.math.ceil
  * ready (default: 200ms).
  * @property acquireHealthCheck Optional custom health check for sandboxes returned by acquire.
  * @property acquireSkipHealthCheck When true, skip readiness checks for sandboxes returned by acquire (default: false).
+ * @property acquireMinRemainingTtl Minimum remaining TTL an idle sandbox must have to be returned by acquire.
+ * Idle entries closer to expiry than this threshold are discarded so the subsequent ready-check and
+ * any user-side renew have time to run before server-side expiry. Defaults to 60s, which covers the
+ * default [acquireReadyTimeout] of 30s with comfortable headroom. Must be strictly less than
+ * [idleTimeout] — a value at or above [idleTimeout] would reject every freshly warmed entry. Set to
+ * [Duration.ZERO] to opt out and restore the pre-existing binary-expiry behavior.
  * @property warmupReadyTimeout Max time to wait for a pool-created sandbox to become ready (default: 30s).
  * @property warmupHealthCheckPollingInterval Poll interval while waiting for a pool-created sandbox to become ready
  * (default: 200ms).
@@ -65,6 +71,7 @@ class PoolConfig private constructor(
     val acquireHealthCheckPollingInterval: Duration,
     val acquireHealthCheck: ((Sandbox) -> Boolean)?,
     val acquireSkipHealthCheck: Boolean,
+    val acquireMinRemainingTtl: Duration,
     val warmupReadyTimeout: Duration,
     val warmupHealthCheckPollingInterval: Duration,
     val warmupHealthCheck: ((Sandbox) -> Boolean)?,
@@ -87,6 +94,11 @@ class PoolConfig private constructor(
         require(!acquireHealthCheckPollingInterval.isNegative && !acquireHealthCheckPollingInterval.isZero) {
             "acquireHealthCheckPollingInterval must be positive"
         }
+        require(!acquireMinRemainingTtl.isNegative) { "acquireMinRemainingTtl must be non-negative" }
+        require(acquireMinRemainingTtl < idleTimeout) {
+            "acquireMinRemainingTtl ($acquireMinRemainingTtl) must be strictly less than " +
+                "idleTimeout ($idleTimeout); otherwise every warmed idle entry would be rejected"
+        }
         require(!warmupReadyTimeout.isNegative && !warmupReadyTimeout.isZero) { "warmupReadyTimeout must be positive" }
         require(!warmupHealthCheckPollingInterval.isNegative && !warmupHealthCheckPollingInterval.isZero) {
             "warmupHealthCheckPollingInterval must be positive"
@@ -101,6 +113,7 @@ class PoolConfig private constructor(
         private const val DEFAULT_DEGRADED_THRESHOLD = 3
         private val DEFAULT_ACQUIRE_READY_TIMEOUT = Duration.ofSeconds(30)
         private val DEFAULT_ACQUIRE_HEALTH_CHECK_POLLING_INTERVAL = Duration.ofMillis(200)
+        private val DEFAULT_ACQUIRE_MIN_REMAINING_TTL: Duration = Duration.ofSeconds(60)
         private val DEFAULT_WARMUP_READY_TIMEOUT = Duration.ofSeconds(30)
         private val DEFAULT_WARMUP_HEALTH_CHECK_POLLING_INTERVAL = Duration.ofMillis(200)
         private val DEFAULT_IDLE_TIMEOUT = Duration.ofHours(24)
@@ -126,6 +139,7 @@ class PoolConfig private constructor(
             acquireHealthCheckPollingInterval = acquireHealthCheckPollingInterval,
             acquireHealthCheck = acquireHealthCheck,
             acquireSkipHealthCheck = acquireSkipHealthCheck,
+            acquireMinRemainingTtl = acquireMinRemainingTtl,
             warmupReadyTimeout = warmupReadyTimeout,
             warmupHealthCheckPollingInterval = warmupHealthCheckPollingInterval,
             warmupHealthCheck = warmupHealthCheck,
@@ -151,6 +165,7 @@ class PoolConfig private constructor(
         private var acquireHealthCheckPollingInterval: Duration = DEFAULT_ACQUIRE_HEALTH_CHECK_POLLING_INTERVAL
         private var acquireHealthCheck: ((Sandbox) -> Boolean)? = null
         private var acquireSkipHealthCheck: Boolean = false
+        private var acquireMinRemainingTtl: Duration = DEFAULT_ACQUIRE_MIN_REMAINING_TTL
         private var warmupReadyTimeout: Duration = DEFAULT_WARMUP_READY_TIMEOUT
         private var warmupHealthCheckPollingInterval: Duration = DEFAULT_WARMUP_HEALTH_CHECK_POLLING_INTERVAL
         private var warmupHealthCheck: ((Sandbox) -> Boolean)? = null
@@ -229,6 +244,19 @@ class PoolConfig private constructor(
             return this
         }
 
+        /**
+         * Sets the minimum remaining TTL an idle sandbox must have to be returned by acquire.
+         * Idle entries closer to expiry than [acquireMinRemainingTtl] are discarded so the subsequent
+         * ready-check and any user-side renew have time to run before the server-side expiry kicks in.
+         *
+         * Must be non-negative and strictly less than [idleTimeout]. Defaults to 60s. Set to
+         * [Duration.ZERO] to opt out and restore the pre-existing binary-expiry behavior.
+         */
+        fun acquireMinRemainingTtl(acquireMinRemainingTtl: Duration): Builder {
+            this.acquireMinRemainingTtl = acquireMinRemainingTtl
+            return this
+        }
+
         fun warmupReadyTimeout(warmupReadyTimeout: Duration): Builder {
             this.warmupReadyTimeout = warmupReadyTimeout
             return this
@@ -293,6 +321,7 @@ class PoolConfig private constructor(
                 acquireHealthCheckPollingInterval = acquireHealthCheckPollingInterval,
                 acquireHealthCheck = acquireHealthCheck,
                 acquireSkipHealthCheck = acquireSkipHealthCheck,
+                acquireMinRemainingTtl = acquireMinRemainingTtl,
                 warmupReadyTimeout = warmupReadyTimeout,
                 warmupHealthCheckPollingInterval = warmupHealthCheckPollingInterval,
                 warmupHealthCheck = warmupHealthCheck,

sdks/sandbox/kotlin/sandbox/src/main/kotlin/com/alibaba/opensandbox/sandbox/domain/pool/PoolStateStore.kt

@@ -38,6 +38,28 @@ interface PoolStateStore {
      */
     fun tryTakeIdle(poolName: String): String?
 
+    /**
+     * Variant of [tryTakeIdle] that skips entries whose remaining TTL is below [minRemainingTtl].
+     *
+     * Atomically removes and returns one idle sandbox ID for the pool whose expiry is at least
+     * [minRemainingTtl] in the future, or null if no such entry exists. Entries failing the check
+     * are still consumed (removed from idle membership) so the pool can replenish with fresh ones.
+     *
+     * Default implementation falls back to [tryTakeIdle] when [minRemainingTtl] is zero or negative
+     * so existing custom store implementations remain source-compatible.
+     */
+    fun tryTakeIdle(
+        poolName: String,
+        minRemainingTtl: Duration,
+    ): String? {
+        if (minRemainingTtl.isNegative || minRemainingTtl.isZero) {
+            return tryTakeIdle(poolName)
+        }
+        // Custom stores that do not override this method fall back to the binary-expiry behavior.
+        // Calling sites that pass a positive minRemainingTtl rely on overrides for correct filtering.
+        return tryTakeIdle(poolName)
+    }
+
     /**
      * Adds a sandbox ID to the idle set for the pool.
      * Idempotent: duplicate put for same sandboxId leaves membership single-copy.
@@ -91,6 +113,28 @@ interface PoolStateStore {
         now: Instant,
     )
 
+    /**
+     * Variant of [reapExpiredIdle] that also evicts entries whose remaining TTL is below
+     * [minRemainingTtl]. Reconcile calls this so near-expiry entries are reclaimed proactively
+     * (rather than waiting for them to fully expire), letting the pool replenish them with fresh
+     * sandboxes before a future acquire would discard them.
+     *
+     * Default implementation falls back to [reapExpiredIdle] when [minRemainingTtl] is zero or
+     * negative so existing custom store implementations remain source-compatible.
+     */
+    fun reapExpiredIdle(
+        poolName: String,
+        now: Instant,
+        minRemainingTtl: Duration,
+    ) {
+        if (minRemainingTtl.isNegative || minRemainingTtl.isZero) {
+            reapExpiredIdle(poolName, now)
+            return
+        }
+        // Custom stores that do not override this method fall back to the strict-expiry sweep.
+        reapExpiredIdle(poolName, now)
+    }
+
     /**
      * Returns a snapshot of counters for the pool (at least idle count).
      * Eventually consistent for distributed stores.

sdks/sandbox/kotlin/sandbox/src/main/kotlin/com/alibaba/opensandbox/sandbox/infrastructure/pool/InMemoryPoolStateStore.kt

@@ -43,14 +43,33 @@ class InMemoryPoolStateStore : PoolStateStore {
     /** Per pool: (map = sandboxId -> entry for idempotent put + expiry, queue = FIFO order for take). */
     private val pools = ConcurrentHashMap<String, PoolIdleState>()
 
-    override fun tryTakeIdle(poolName: String): String? {
+    override fun tryTakeIdle(poolName: String): String? = takeIdle(poolName, Instant.now())
+
+    override fun tryTakeIdle(
+        poolName: String,
+        minRemainingTtl: Duration,
+    ): String? {
+        if (minRemainingTtl.isNegative || minRemainingTtl.isZero) {
+            return tryTakeIdle(poolName)
+        }
+        return takeIdle(poolName, Instant.now().plus(minRemainingTtl))
+    }
+
+    /**
+     * Drains the FIFO queue, returning the first idle entry whose [IdleEntry.expiresAt] is
+     * strictly after [cutoff]. Entries failing the check (including already-expired and near-expiry)
+     * are removed from idle membership and discarded so reconcile can replenish.
+     */
+    private fun takeIdle(
+        poolName: String,
+        cutoff: Instant,
+    ): String? {
         val state = pools[poolName] ?: return null
-        val now = Instant.now()
         while (true) {
             val sandboxId = state.queue.poll() ?: return null
             val entry = state.map.remove(sandboxId) ?: continue // already removed (e.g. by removeIdle)
-            if (entry.expiresAt.isAfter(now)) return sandboxId
-            // expired, discard and poll next
+            if (entry.expiresAt.isAfter(cutoff)) return sandboxId
+            // expired or below minRemainingTtl, discard and poll next
         }
     }
 
@@ -108,6 +127,21 @@ class InMemoryPoolStateStore : PoolStateStore {
         state.queue.removeIf { sandboxId -> !state.map.containsKey(sandboxId) }
     }
 
+    override fun reapExpiredIdle(
+        poolName: String,
+        now: Instant,
+        minRemainingTtl: Duration,
+    ) {
+        if (minRemainingTtl.isNegative || minRemainingTtl.isZero) {
+            reapExpiredIdle(poolName, now)
+            return
+        }
+        val state = pools[poolName] ?: return
+        val cutoff = now.plus(minRemainingTtl)
+        state.map.entries.removeIf { !it.value.expiresAt.isAfter(cutoff) }
+        state.queue.removeIf { sandboxId -> !state.map.containsKey(sandboxId) }
+    }
+
     override fun snapshotCounters(poolName: String): StoreCounters {
         val state = pools[poolName] ?: return StoreCounters(idleCount = 0)
         val now = Instant.now()

sdks/sandbox/kotlin/sandbox/src/main/kotlin/com/alibaba/opensandbox/sandbox/infrastructure/pool/PoolReconciler.kt

@@ -72,7 +72,7 @@ internal object PoolReconciler {
         val ttl = config.primaryLockTtl
         val now = Instant.now()
 
-        stateStore.reapExpiredIdle(poolName, now)
+        stateStore.reapExpiredIdle(poolName, now, config.acquireMinRemainingTtl)
         val counters = stateStore.snapshotCounters(poolName)
         val excess = (counters.idleCount - config.maxIdle).coerceAtLeast(0)
         val toRemove = minOf(excess, config.warmupConcurrency)

sdks/sandbox/kotlin/sandbox/src/main/kotlin/com/alibaba/opensandbox/sandbox/pool/SandboxPool.kt

@@ -209,7 +209,7 @@ class SandboxPool internal constructor(
                 throw PoolNotRunningException("Cannot acquire when pool state is $state")
             }
             val poolName = config.poolName
-            val sandboxId = stateStore.tryTakeIdle(poolName)
+            val sandboxId = stateStore.tryTakeIdle(poolName, config.acquireMinRemainingTtl)
             var noIdleReason: String? = null // null = got a sandbox from idle; non-null = reason we have no usable idle
             var idleConnectFailure: Exception? = null
             if (sandboxId != null) {
@@ -724,6 +724,11 @@ class SandboxPool internal constructor(
             return this
         }
 
+        fun acquireMinRemainingTtl(acquireMinRemainingTtl: Duration): Builder {
+            configBuilder.acquireMinRemainingTtl(acquireMinRemainingTtl)
+            return this
+        }
+
         fun warmupReadyTimeout(warmupReadyTimeout: Duration): Builder {
             configBuilder.warmupReadyTimeout(warmupReadyTimeout)
             return this