diff --git a/amethyst/src/main/java/com/vitorpamplona/amethyst/service/okhttp/AmethystDns.kt b/amethyst/src/main/java/com/vitorpamplona/amethyst/service/okhttp/AmethystDns.kt index 7648ba37d..22797293e 100644 --- a/amethyst/src/main/java/com/vitorpamplona/amethyst/service/okhttp/AmethystDns.kt +++ b/amethyst/src/main/java/com/vitorpamplona/amethyst/service/okhttp/AmethystDns.kt @@ -26,44 +26,55 @@ import java.net.UnknownHostException import java.util.concurrent.CompletableFuture import java.util.concurrent.ConcurrentHashMap import java.util.concurrent.ExecutionException +import java.util.concurrent.Executor +import java.util.concurrent.Executors +import java.util.concurrent.RejectedExecutionException +import java.util.concurrent.ThreadLocalRandom import java.util.concurrent.TimeUnit /** - * Concurrent, caching DNS resolver for OkHttp. + * Concurrent, caching, stale-while-revalidate DNS resolver for OkHttp. * * Tuned for an Amethyst-shaped workload: ~700 relays plus a small set of media/profile/NIP-05 - * hosts that reappear constantly. Steady state is well under [maxEntries] distinct hosts whose - * IPs change on the order of days, so we want to resolve each one once per session and never - * touch DNS again. + * hosts that reappear constantly, whose IPs change on the order of days. * * Properties: * - * 1. **Lock-free reads.** Cache is a [ConcurrentHashMap], so the hot path (cache hit) does no - * locking. The previous incarnation used `synchronizedMap(LinkedHashMap(access-order=true))`, - * which turned every `get` into a monitor-protected write because access-order LRU rewrites - * the linked list on read — at 700 concurrent relay reconnects, the lock dominated. - * 2. **Single-flight coalescing.** N concurrent OkHttp threads asking for the same host share - * one upstream `getaddrinfo`. The leader resolves; followers block on the same future. If a - * peer leader refreshed the entry while we were claiming leadership, the leader re-checks - * the cache and skips the system call entirely. - * 3. **Generous positive TTL.** Defaults to 24h. Relay and CDN IPs almost never move, and we - * are not a recursive resolver — there is no correctness reason to honor authoritative TTLs. - * Pair with [invalidate] on connection failures or network changes to recover from real - * IP moves. - * 4. **Short negative TTL.** Failed lookups are remembered for 10s so a typo or a transiently - * down host doesn't keep paying for `getaddrinfo`, but a real outage recovers quickly. + * 1. **Lock-free reads.** Cache is a [ConcurrentHashMap]. The hot path (cache hit) takes no + * locks, so 700 concurrent relay reconnects fan out across OkHttp dispatcher threads + * instead of serializing through one monitor. + * 2. **Single-flight coalescing.** N concurrent threads asking for the same host share one + * upstream `getaddrinfo`. The leader resolves; followers block on the same future. + * 3. **Stale-while-revalidate.** Once a host has been resolved, recurring lookups *never* + * block on `getaddrinfo` again. After the soft TTL expires, we return the previous answer + * immediately and kick a background refresh. The refresh is coalesced through the same + * `inflight` map, so a burst of stale reads triggers one refresh per host. If the refresh + * fails (`UnknownHostException`), the cache entry is demoted to negative, so the next + * caller gets a fresh failure rather than forever-wrong stale IPs. + * 4. **Negative entries do *not* serve stale.** When an `UnknownHostException` cache entry + * expires, the next call goes through the synchronous path. We want transient failures to + * recover quickly, not keep returning stale failures. + * 5. **Generous positive TTL with jitter.** Defaults to 24h plus up to 24h of random jitter, + * so each entry expires somewhere in [base, base + jitter]. This breaks the synchronized + * herd that would otherwise form when ~700 relay reconnects all populate the cache in the + * same second: their expiries spread across a 24h window instead of landing at the same + * instant 24h later. A user who opens the app once a day catches only a small fraction of + * entries stale per session, and refreshes drip through the executor pool naturally. * - * Remaining blocking points are unavoidable: the leader's [Dns.lookup] call is a synchronous JNI - * hop into the system resolver, and followers must wait on the leader's future. Both are - * bypassed entirely on cache hit, which is the steady state. + * Remaining blocking points: the very first lookup of a host blocks on `getaddrinfo` + * (unavoidable — there's nothing to serve stale yet), and followers waiting on that first + * lookup block on `future.get()`. Background refreshes never block any caller. */ class AmethystDns( private val delegate: Dns = Dns.SYSTEM, private val maxEntries: Int = 2000, positiveTtlMs: Long = TimeUnit.HOURS.toMillis(24), + positiveTtlJitterMs: Long = TimeUnit.HOURS.toMillis(24), negativeTtlMs: Long = TimeUnit.SECONDS.toMillis(10), + private val refreshExecutor: Executor = DEFAULT_REFRESH_EXECUTOR, ) : Dns { private val positiveTtlNanos = TimeUnit.MILLISECONDS.toNanos(positiveTtlMs) + private val positiveTtlJitterNanos = TimeUnit.MILLISECONDS.toNanos(positiveTtlJitterMs) private val negativeTtlNanos = TimeUnit.MILLISECONDS.toNanos(negativeTtlMs) private val cache = ConcurrentHashMap() @@ -71,9 +82,16 @@ class AmethystDns( override fun lookup(hostname: String): List { cache[hostname]?.let { entry -> - if (entry.expiresAtNanos > System.nanoTime()) { + val now = System.nanoTime() + if (entry.expiresAtNanos > now) { return entry.unwrap(hostname) } + // Soft-expired positive entry: serve stale, refresh in background. Negative + // entries fall through to the sync path so transient failures recover quickly. + if (entry.addresses.isNotEmpty()) { + triggerBackgroundRefresh(hostname) + return entry.addresses + } } val newFuture = CompletableFuture>() @@ -85,6 +103,24 @@ class AmethystDns( } } + private fun triggerBackgroundRefresh(hostname: String) { + val refreshFuture = CompletableFuture>() + // Coalesce: if a sync lookup or a prior refresh is already in flight, skip. + if (inflight.putIfAbsent(hostname, refreshFuture) != null) return + try { + refreshExecutor.execute { + try { + resolveAsLeader(hostname, refreshFuture) + } catch (_: Throwable) { + // Caller already got the stale answer; the failure is recorded on the + // future and (for UnknownHostException) demoted in the cache. + } + } + } catch (_: RejectedExecutionException) { + inflight.remove(hostname, refreshFuture) + } + } + private fun resolveAsLeader( hostname: String, future: CompletableFuture>, @@ -98,9 +134,9 @@ class AmethystDns( cached.unwrap(hostname) } else { try { - delegate.lookup(hostname).also { put(hostname, it, positiveTtlNanos) } + delegate.lookup(hostname).also { putPositive(hostname, it) } } catch (e: UnknownHostException) { - put(hostname, emptyList(), negativeTtlNanos) + putNegative(hostname) throw e } } @@ -133,12 +169,24 @@ class AmethystDns( } } - private fun put( + private fun putPositive( hostname: String, addresses: List, - ttlNanos: Long, ) { - cache[hostname] = Entry(addresses, System.nanoTime() + ttlNanos) + // Jitter the expiry so a burst of co-written entries (e.g. ~700 relay reconnects at app + // start) doesn't all expire at the same instant 24h later. + val jitter = + if (positiveTtlJitterNanos > 0L) { + ThreadLocalRandom.current().nextLong(positiveTtlJitterNanos) + } else { + 0L + } + cache[hostname] = Entry(addresses, System.nanoTime() + positiveTtlNanos + jitter) + if (cache.size > maxEntries) evictExpired() + } + + private fun putNegative(hostname: String) { + cache[hostname] = Entry(emptyList(), System.nanoTime() + negativeTtlNanos) if (cache.size > maxEntries) evictExpired() } @@ -168,6 +216,15 @@ class AmethystDns( } companion object { + // Small fixed pool of daemon threads. Refreshes block on getaddrinfo (~tens of ms), + // so a handful of threads is plenty even when many hosts go stale at once — extra + // refreshes queue up without blocking any caller, since callers always get the + // stale answer instantly. + private val DEFAULT_REFRESH_EXECUTOR: Executor = + Executors.newFixedThreadPool(8) { r -> + Thread(r, "amethyst-dns-refresh").apply { isDaemon = true } + } + /** * Process-wide instance shared by every OkHttp client built in the app, so a host resolved * for an image fetch is reused when a relay handshake or NIP-05 lookup hits the same host. diff --git a/amethyst/src/test/java/com/vitorpamplona/amethyst/service/okhttp/AmethystDnsTest.kt b/amethyst/src/test/java/com/vitorpamplona/amethyst/service/okhttp/AmethystDnsTest.kt index 0c0b88554..91d25e257 100644 --- a/amethyst/src/test/java/com/vitorpamplona/amethyst/service/okhttp/AmethystDnsTest.kt +++ b/amethyst/src/test/java/com/vitorpamplona/amethyst/service/okhttp/AmethystDnsTest.kt @@ -29,9 +29,11 @@ import org.junit.Test import java.net.InetAddress import java.net.UnknownHostException import java.util.concurrent.CountDownLatch +import java.util.concurrent.Executor import java.util.concurrent.Executors import java.util.concurrent.TimeUnit import java.util.concurrent.atomic.AtomicInteger +import java.util.concurrent.atomic.AtomicReference class AmethystDnsTest { private fun ip(value: String) = InetAddress.getByName(value) @@ -89,22 +91,135 @@ class AmethystDnsTest { } @Test - fun `positive entries expire`() { + fun `expired positive entry serves stale and refreshes in background`() { val upstream = CountingDns(mapOf("a.example" to listOf(ip("1.2.3.4")))) + val syncRefresh = Executor { it.run() } val dns = AmethystDns( delegate = upstream, positiveTtlMs = 1, + positiveTtlJitterMs = 0, negativeTtlMs = 1, + refreshExecutor = syncRefresh, ) dns.lookup("a.example") Thread.sleep(20) - dns.lookup("a.example") + // Returns the stale cached value AND triggers a refresh on the synchronous executor. + assertEquals(listOf(ip("1.2.3.4")), dns.lookup("a.example")) assertEquals(2, upstream.calls("a.example")) } + @Test + fun `expired negative entry does not stale-while-revalidate`() { + val upstream = CountingDns(emptyMap()) + val dns = + AmethystDns( + delegate = upstream, + positiveTtlJitterMs = 0, + negativeTtlMs = 1, + ) + + assertThrows(UnknownHostException::class.java) { dns.lookup("missing.example") } + Thread.sleep(20) + assertThrows(UnknownHostException::class.java) { dns.lookup("missing.example") } + + // Two synchronous calls — failed lookups must retry quickly, not be served stale. + assertEquals(2, upstream.calls("missing.example")) + } + + @Test + fun `stale read returns previous IP while refresh updates the cache`() { + val responses = AtomicReference>(listOf(ip("1.2.3.4"))) + val calls = AtomicInteger() + val upstream = + Dns { _ -> + calls.incrementAndGet() + responses.get() + } + val syncRefresh = Executor { it.run() } + val dns = + AmethystDns( + delegate = upstream, + positiveTtlMs = 1, + positiveTtlJitterMs = 0, + refreshExecutor = syncRefresh, + ) + + assertEquals(listOf(ip("1.2.3.4")), dns.lookup("a.example")) + Thread.sleep(20) + + // Upstream now returns a new IP. The next lookup should still serve the OLD IP + // immediately, while the synchronous executor performs the refresh inline. + responses.set(listOf(ip("5.6.7.8"))) + val stale = dns.lookup("a.example") + assertEquals(listOf(ip("1.2.3.4")), stale) + assertEquals(2, calls.get()) + + // Cache now holds the refreshed IP — no further upstream calls. + assertEquals(listOf(ip("5.6.7.8")), dns.lookup("a.example")) + assertEquals(2, calls.get()) + } + + @Test + fun `stale burst on the same host triggers a single refresh`() { + val gated = GatedDns(mapOf("hot.example" to listOf(ip("9.9.9.9")))) + // Pre-populate via a separate, non-gated upstream then swap in the gated one for + // the refresh — easiest way: bootstrap by writing through a delegate that completes + // immediately, then attach the gate to count parallel refresh calls. + val bootstrapCalls = AtomicInteger() + val dynamic = + object : Dns { + @Volatile var useGated = false + + override fun lookup(hostname: String): List = + if (useGated) { + gated.lookup(hostname) + } else { + bootstrapCalls.incrementAndGet() + listOf(ip("9.9.9.9")) + } + } + val pool = Executors.newFixedThreadPool(8) + val dns = + AmethystDns( + delegate = dynamic, + positiveTtlMs = 1, + positiveTtlJitterMs = 0, + refreshExecutor = pool, + ) + try { + dns.lookup("hot.example") + assertEquals(1, bootstrapCalls.get()) + Thread.sleep(20) + dynamic.useGated = true + + // Fan out 16 stale reads concurrently. They all return the stale answer + // immediately; only one refresh should be queued/executing. + val callerPool = Executors.newFixedThreadPool(16) + try { + val results = + (1..16).map { + callerPool.submit> { dns.lookup("hot.example") } + } + results.forEach { assertEquals(listOf(ip("9.9.9.9")), it.get(2, TimeUnit.SECONDS)) } + assertTrue( + "Refresh should have started", + gated.started.await(2, TimeUnit.SECONDS), + ) + gated.release.countDown() + // Allow refresh to complete. + Thread.sleep(100) + assertEquals("Only one refresh upstream call", 1, gated.calls.get()) + } finally { + callerPool.shutdownNow() + } + } finally { + pool.shutdownNow() + } + } + @Test fun `concurrent lookups for the same host coalesce to one upstream call`() { val gated = GatedDns(mapOf("hot.example" to listOf(ip("9.9.9.9"))))