fix: settle pending async sleeps on close

openai-java-core/src/main/kotlin/com/openai/core/DefaultSleeper.kt

@@ -8,21 +8,57 @@ import java.util.concurrent.CompletableFuture
 class DefaultSleeper : Sleeper {
 
     private val timer = Timer("DefaultSleeper", true)
+    private val lock = Any()
+    private val pending = mutableSetOf<CompletableFuture<Void>>()
+    private var closed = false
 
     override fun sleep(duration: Duration) = Thread.sleep(duration.toMillis())
 
     override fun sleepAsync(duration: Duration): CompletableFuture<Void> {
         val future = CompletableFuture<Void>()
-        timer.schedule(
+        val task =
             object : TimerTask() {
                 override fun run() {
-                    future.complete(null)
+                    if (synchronized(lock) { pending.remove(future) }) {
+                        future.complete(null)
+                    }
                 }
-            },
-            duration.toMillis(),
-        )
+            }
+
+        synchronized(lock) {
+            if (closed) {
+                future.cancel(false)
+                return future
+            }
+
+            pending.add(future)
+            future.whenComplete { _, _ ->
+                synchronized(lock) { pending.remove(future) }
+                task.cancel()
+            }
+
+            try {
+                timer.schedule(task, duration.toMillis())
+            } catch (throwable: Throwable) {
+                pending.remove(future)
+                throw throwable
+            }
+        }
+
         return future
     }
 
-    override fun close() = timer.cancel()
+    override fun close() {
+        val pendingFutures =
+            synchronized(lock) {
+                if (closed) {
+                    return
+                }
+                closed = true
+                timer.cancel()
+                pending.toList().also { pending.clear() }
+            }
+
+        pendingFutures.forEach { it.cancel(false) }
+    }
 }

openai-java-core/src/test/kotlin/com/openai/core/DefaultSleeperTest.kt

@@ -0,0 +1,55 @@
+package com.openai.core
+
+import java.time.Duration
+import java.util.concurrent.TimeUnit
+import org.assertj.core.api.Assertions.assertThat
+import org.junit.jupiter.api.Test
+
+internal class DefaultSleeperTest {
+
+    @Test
+    fun closeCancelsPendingAsyncSleep() {
+        val sleeper = DefaultSleeper()
+        val future = sleeper.sleepAsync(Duration.ofHours(1))
+
+        sleeper.close()
+
+        assertThat(future.isDone).isTrue()
+        assertThat(future.isCancelled).isTrue()
+    }
+
+    @Test
+    fun completedAsyncSleepRemainsCompletedAfterClose() {
+        val sleeper = DefaultSleeper()
+        val future = sleeper.sleepAsync(Duration.ZERO)
+
+        future.get(1, TimeUnit.SECONDS)
+        sleeper.close()
+
+        assertThat(future.isDone).isTrue()
+        assertThat(future.isCancelled).isFalse()
+        assertThat(future.isCompletedExceptionally).isFalse()
+    }
+
+    @Test
+    fun closeIsIdempotent() {
+        val sleeper = DefaultSleeper()
+        val future = sleeper.sleepAsync(Duration.ofHours(1))
+
+        sleeper.close()
+        sleeper.close()
+
+        assertThat(future.isCancelled).isTrue()
+    }
+
+    @Test
+    fun sleepAsyncAfterCloseReturnsCancelledFuture() {
+        val sleeper = DefaultSleeper()
+        sleeper.close()
+
+        val call = runCatching { sleeper.sleepAsync(Duration.ofSeconds(1)) }
+
+        assertThat(call.isSuccess).isTrue()
+        assertThat(call.getOrThrow().isCancelled).isTrue()
+    }
+}