Fix cancelled delayed jobs not being disposed of in TestDispatcher

kotlinx-coroutines-test/common/src/TestDispatcher.kt

@@ -31,7 +31,14 @@ public abstract class TestDispatcher internal constructor() : CoroutineDispatche
     /** @suppress */
     override fun scheduleResumeAfterDelay(timeMillis: Long, continuation: CancellableContinuation<Unit>) {
         val timedRunnable = CancellableContinuationRunnable(continuation, this)
-        scheduler.registerEvent(this, timeMillis, timedRunnable, continuation.context, ::cancellableRunnableIsCancelled)
+        val handle = scheduler.registerEvent(
+            this,
+            timeMillis,
+            timedRunnable,
+            continuation.context,
+            ::cancellableRunnableIsCancelled
+        )
+        continuation.invokeOnCancellation { handle.dispose() }
     }
 
     /** @suppress */

kotlinx-coroutines-test/jvm/test/MemoryLeakTest.kt

@@ -0,0 +1,28 @@
+/*
+ * Copyright 2016-2022 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license.
+ */
+import kotlinx.coroutines.*
+import kotlinx.coroutines.test.*
+import java.lang.ref.*
+import kotlin.test.*
+
+class MemoryLeakTest {
+
+    @Test
+    fun testCancellationLeakInTestCoroutineScheduler() = runTest {
+        lateinit var weakRef: WeakReference<*>
+        val job = launch(start = CoroutineStart.UNDISPATCHED) {
+            val leakingObject = Any()
+            weakRef = WeakReference(leakingObject)
+            delay(3)
+            // This code is needed to hold a reference to `leakingObject` until the job itself is weakly reachable.
+            leakingObject.hashCode()
+        }
+        job.cancel()
+        System.gc()
+        assertNotNull(weakRef.get()) // the cancellation handler has not run yet
+        runCurrent()
+        System.gc()
+        assertNull(weakRef.get()) // the cancellation handler has run, disposing of the job
+    }
+}