From edd36b467c671ce91abab4018235c19e1db91260 Mon Sep 17 00:00:00 2001 From: Vitor Pamplona Date: Wed, 29 Jul 2026 21:25:20 -0400 Subject: [PATCH] test(commons): deflake FeedMetadataCoordinatorTest MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `loadMetadataBatched follows the same retry semantics` failed on the macOS build-desktop runner (1431 tests in the module, 1 failed). The coordinator is fine; the test was wall-clock coupled. It fired call 1 with `timeoutMs = 200`, waited a flat `delay(350)`, then asserted that call 2 re-subscribed. That leaves a 150ms budget for `scope.launch` to be scheduled, subscribe, `awaitAll(200)`, unsubscribe and roll the pubkeys out of `inFlightBatchedMetadata`. On a loaded runner the launch itself can be queued past the margin, so the roll-back lands late, call 2 short-circuits by design, and the assertion fails on healthy code. Every test in the file had the same shape. "Has call 1 finished?" is a question about the job tree, not the clock, and the test owns the scope — so it can just ask. `awaitCoordinatorIdle()` waits until no child of the test scope is active; `waitUntil {}` polls the few preconditions that aren't expressible that way (listener registered before EOSEs are fired). Both carry a generous deadline and fail with a message rather than hanging. Safe here because the coordinator launches nothing at construction and `BatchEoseGate` closes and joins its consumer inside `awaitAll`, so the scope does reach idle. Also made the test fake thread-safe. `subscriptions` and `subscribeCalls` were a plain map and list, written from the coordinator's `Dispatchers.Default` coroutines (and `Dispatchers.IO` in the concurrent-EOSE test) and read from the test thread: an unsynchronized `size` read can be stale, and `fireEose` iterating `subscriptions.values` while a coordinator coroutine calls `unsubscribe` can throw ConcurrentModificationException. Concurrency is the thing under test, so the fake shouldn't be the weak link. Now ConcurrentHashMap + synchronizedList + AtomicInteger, with a snapshot in `fireEose`. Verified deterministic rather than just green: - 3/3 idle, 4/4 under 24 busy loops on 12 cores (the old shape needed the margin; the new one is load-independent). - Still catches its regression: reintroducing the original mark-on-send bug (`eosedRelays > 0` -> `>= 0`) fails exactly this test. A deflaked test that no longer detects the fault would be worse than the flake. This is a pre-existing flake (test dates from 5217035f94, 2026-07-07) unrelated to the rest of this branch, which touches no commons code — fixed here because it blocks the branch's CI. Co-Authored-By: Claude Opus 5 (1M context) --- .../assemblers/FeedMetadataCoordinatorTest.kt | 96 ++++++++++++++----- 1 file changed, 72 insertions(+), 24 deletions(-) diff --git a/commons/src/jvmTest/kotlin/com/vitorpamplona/amethyst/commons/relayClient/assemblers/FeedMetadataCoordinatorTest.kt b/commons/src/jvmTest/kotlin/com/vitorpamplona/amethyst/commons/relayClient/assemblers/FeedMetadataCoordinatorTest.kt index c0f8a609d4..4dc8837e8d 100644 --- a/commons/src/jvmTest/kotlin/com/vitorpamplona/amethyst/commons/relayClient/assemblers/FeedMetadataCoordinatorTest.kt +++ b/commons/src/jvmTest/kotlin/com/vitorpamplona/amethyst/commons/relayClient/assemblers/FeedMetadataCoordinatorTest.kt @@ -31,6 +31,7 @@ import kotlinx.coroutines.Dispatchers import kotlinx.coroutines.SupervisorJob import kotlinx.coroutines.cancel import kotlinx.coroutines.delay +import kotlinx.coroutines.job import kotlinx.coroutines.launch import kotlinx.coroutines.runBlocking import org.junit.After @@ -38,6 +39,9 @@ import org.junit.Assert.assertEquals import org.junit.Assert.assertTrue import org.junit.Before import org.junit.Test +import java.util.Collections +import java.util.concurrent.ConcurrentHashMap +import java.util.concurrent.atomic.AtomicInteger /** * Regression tests for PR #3483 review findings on FeedMetadataCoordinator: @@ -73,34 +77,78 @@ class FeedMetadataCoordinatorTest { private fun pubkey(seed: Int): HexKey = seed.toString(16).padStart(64, '0') + /** + * Suspends until every coroutine the coordinator launched into [scope] has finished. + * + * This is what makes the assertions deterministic. The coordinator does its work in + * `scope.launch { subscribe(); gate.awaitAll(timeout); unsubscribe(); promote-or-roll-back }`, + * so "has call 1 finished?" is a question about the job tree, not about the clock. The tests + * used to answer it with `delay(timeoutMs + margin)` and assert straight after — which holds + * only while the dispatcher is free to start that coroutine promptly. On a loaded runner + * (macOS CI, 1431 tests in the same module) the launch itself can be queued past the margin, + * the roll-back lands late, the next call short-circuits, and the assertion fails on a + * perfectly healthy coordinator. Waiting on the jobs removes the margin entirely. + */ + private suspend fun awaitCoordinatorIdle(timeoutMs: Long = 30_000) { + val deadline = System.currentTimeMillis() + timeoutMs + while (scope.coroutineContext.job.children + .any { it.isActive } + ) { + check(System.currentTimeMillis() < deadline) { "coordinator work did not settle within ${timeoutMs}ms" } + delay(2) + } + } + + /** Polls [predicate] to a deadline. For preconditions we cannot express as job completion. */ + private suspend fun waitUntil( + message: String, + timeoutMs: Long = 30_000, + predicate: () -> Boolean, + ) { + val deadline = System.currentTimeMillis() + timeoutMs + while (!predicate()) { + check(System.currentTimeMillis() < deadline) { "timed out waiting for: $message" } + delay(2) + } + } + /** * Fake client that captures subscribe/unsubscribe and lets the test * drive EOSE notifications on any dispatcher we choose. + * + * Every collection here is written from the coordinator's coroutines (`Dispatchers.Default`, + * and `Dispatchers.IO` for the concurrent-EOSE test) and read from the test thread, so plain + * `mutableMapOf`/`mutableListOf` were two more races: an unsynchronized `size` read can be + * stale, and `fireEose` iterating `subscriptions.values` while a coordinator coroutine calls + * `unsubscribe` can throw ConcurrentModificationException. Concurrency is the thing under + * test here, so the fake must not be the weak link. */ private class ControllableClient( private val delegate: INostrClient = EmptyNostrClient(), ) : INostrClient by delegate { - val subscriptions = mutableMapOf() - val subscribeCalls = mutableListOf>>() - var unsubscribeCallCount = 0 - private set + val subscriptions = ConcurrentHashMap() + val subscribeCalls: MutableList>> = + Collections.synchronizedList(mutableListOf()) + private val unsubscribes = AtomicInteger(0) + val unsubscribeCallCount: Int get() = unsubscribes.get() override fun subscribe( subId: String, filters: Map>, listener: SubscriptionListener?, ) { - subscriptions[subId] = listener + listener?.let { subscriptions[subId] = it } subscribeCalls.add(filters) } override fun unsubscribe(subId: String) { subscriptions.remove(subId) - unsubscribeCallCount++ + unsubscribes.incrementAndGet() } fun fireEose(relay: NormalizedRelayUrl) { - subscriptions.values.filterNotNull().forEach { it.onEose(relay, forFilters = null) } + // Snapshot: a coordinator coroutine may unsubscribe concurrently. + subscriptions.values.toList().forEach { it.onEose(relay, forFilters = null) } } } @@ -119,13 +167,13 @@ class FeedMetadataCoordinatorTest { // Call 1 — no relay EOSEs; must time out. coordinator.loadKind3Batched(pubkeys, timeoutMs = 200) - delay(350) // exceed the timeout + awaitCoordinatorIdle() // call 1 timed out and rolled back // Call 2 — the same pubkeys must be re-subscribed since call 1 // never got a successful EOSE. The old code would silently // short-circuit here. coordinator.loadKind3Batched(pubkeys, timeoutMs = 200) - delay(50) // let the launcher run + awaitCoordinatorIdle() assertEquals( "Zero-EOSE timeout must not permanently dedup pubkeys", @@ -158,13 +206,13 @@ class FeedMetadataCoordinatorTest { val pubkeys = listOf(pubkey(1), pubkey(2)) coordinator.loadKind3Batched(pubkeys, timeoutMs = 1_000) - // Give the launcher time to register the listener before we fire. - delay(50) + // The listener must be registered before we fire, or the EOSEs go nowhere. + waitUntil("subscription registered") { client.subscriptions.isNotEmpty() } indexRelays.forEach(client::fireEose) - delay(200) // let the coordinator finish + promote to queued + awaitCoordinatorIdle() // coordinator finished + promoted to queued coordinator.loadKind3Batched(pubkeys, timeoutMs = 200) - delay(50) + awaitCoordinatorIdle() assertEquals( "Successful call must dedup subsequent identical calls", @@ -187,13 +235,13 @@ class FeedMetadataCoordinatorTest { val pubkeys = listOf(pubkey(1)) coordinator.loadKind3Batched(pubkeys, timeoutMs = 300) - delay(30) + waitUntil("subscription registered") { client.subscriptions.isNotEmpty() } // Only 1 of 3 EOSEs — timeout still fires but we made progress. client.fireEose(relay1) - delay(400) + awaitCoordinatorIdle() coordinator.loadKind3Batched(pubkeys, timeoutMs = 200) - delay(50) + awaitCoordinatorIdle() assertEquals( "≥1 EOSE = progress = promote to queued (avoid re-asking)", @@ -222,7 +270,7 @@ class FeedMetadataCoordinatorTest { ) coordinator.loadKind3Batched(listOf(pubkey(1)), timeoutMs = 2_000) - delay(50) // wait for subscription + waitUntil("subscription registered") { client.subscriptions.isNotEmpty() } // Fire EOSEs concurrently from many dispatchers. val jobs = @@ -235,9 +283,9 @@ class FeedMetadataCoordinatorTest { // The 2nd call must short-circuit — every relay EOSE'd, so // pubkey(1) is now in queuedKind3Pubkeys. - delay(100) + awaitCoordinatorIdle() coordinator.loadKind3Batched(listOf(pubkey(1)), timeoutMs = 200) - delay(50) + awaitCoordinatorIdle() assertEquals( "Under concurrent EOSE from all relays, aggregator must reach target", @@ -261,10 +309,10 @@ class FeedMetadataCoordinatorTest { // Call 1 — zero EOSE, timeout. coordinator.loadMetadataBatched(pubkeys, timeoutMs = 200) - delay(350) + awaitCoordinatorIdle() // Call 2 — must re-subscribe. coordinator.loadMetadataBatched(pubkeys, timeoutMs = 200) - delay(50) + awaitCoordinatorIdle() assertTrue( "Metadata batch also retries on zero-EOSE timeout", @@ -284,13 +332,13 @@ class FeedMetadataCoordinatorTest { ) coordinator.loadKind3Batched(listOf(pubkey(1)), timeoutMs = 200) - delay(50) + waitUntil("subscription registered") { client.subscriptions.isNotEmpty() } // clear() must drop the in-flight tracker even mid-request. coordinator.clear() - delay(300) // let call 1 finish + roll back + awaitCoordinatorIdle() // call 1 finished + rolled back coordinator.loadKind3Batched(listOf(pubkey(1)), timeoutMs = 200) - delay(50) + awaitCoordinatorIdle() assertTrue(client.subscribeCalls.size >= 2) }