From a2102e2becf9d5ca7bbdaf9159656d68b172c6ce Mon Sep 17 00:00:00 2001 From: Claude Date: Sun, 3 May 2026 15:17:06 +0000 Subject: [PATCH] feat(okhttp): concurrent caching DNS resolver MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Avoids paying the getaddrinfo tax on every HTTP call to the same host. Adds a single-flight, LRU+TTL DNS resolver wired into both the media and relay OkHttp clients via a process-wide shared instance, so resolutions cross-cut images, relays, and NIP-05 lookups. - Per-host coalescing: N concurrent lookups for the same host share one upstream call. - Different hosts proceed in parallel — no global lock around the upstream resolver. - Negative cache (10s) prevents hammering on typos / dead hosts. - Positive cache (5m) survives a feed scroll. --- .../amethyst/service/okhttp/AmethystDns.kt | 155 +++++++++++++ .../service/okhttp/OkHttpClientFactory.kt | 1 + .../okhttp/OkHttpClientFactoryForRelays.kt | 1 + .../service/okhttp/AmethystDnsTest.kt | 204 ++++++++++++++++++ 4 files changed, 361 insertions(+) create mode 100644 amethyst/src/main/java/com/vitorpamplona/amethyst/service/okhttp/AmethystDns.kt create mode 100644 amethyst/src/test/java/com/vitorpamplona/amethyst/service/okhttp/AmethystDnsTest.kt 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 new file mode 100644 index 0000000000..a453864a5e --- /dev/null +++ b/amethyst/src/main/java/com/vitorpamplona/amethyst/service/okhttp/AmethystDns.kt @@ -0,0 +1,155 @@ +/* + * Copyright (c) 2025 Vitor Pamplona + * + * Permission is hereby granted, free of charge, to any person obtaining a copy of + * this software and associated documentation files (the "Software"), to deal in + * the Software without restriction, including without limitation the rights to use, + * copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the + * Software, and to permit persons to whom the Software is furnished to do so, + * subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in all + * copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS + * FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR + * COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN + * AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION + * WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. + */ +package com.vitorpamplona.amethyst.service.okhttp + +import okhttp3.Dns +import java.net.InetAddress +import java.net.UnknownHostException +import java.util.Collections +import java.util.concurrent.CompletableFuture +import java.util.concurrent.ConcurrentHashMap +import java.util.concurrent.ExecutionException +import java.util.concurrent.TimeUnit + +/** + * Concurrent, caching DNS resolver for OkHttp. + * + * The system resolver call ([InetAddress.getAllByName] used by [Dns.SYSTEM]) is a blocking JNI + * hop into `getaddrinfo`. On a busy feed we may issue dozens of HTTP calls to a handful of hosts + * in the same second; the default behaviour pays the resolver tax once per call and serializes + * the OkHttp dispatcher worker that asked for it. + * + * This resolver adds three things on top of the system resolver: + * + * 1. An LRU + TTL cache, so repeated lookups of the same host short-circuit before touching the + * network. Negative results get a short TTL so a typo doesn't keep hammering DNS. + * 2. Single-flight coalescing: when N OkHttp threads ask for the same host concurrently, only + * one of them performs the upstream lookup. The others block on the same future and pick up + * the result. Without this, ten parallel image requests to the same CDN make ten DNS calls. + * 3. No global lock on the slow path: lookups for *different* hosts proceed in parallel because + * the upstream resolver is invoked outside any monitor. + */ +class AmethystDns( + private val delegate: Dns = Dns.SYSTEM, + private val maxEntries: Int = 256, + positiveTtlMs: Long = TimeUnit.MINUTES.toMillis(5), + negativeTtlMs: Long = TimeUnit.SECONDS.toMillis(10), +) : Dns { + private val positiveTtlNanos = TimeUnit.MILLISECONDS.toNanos(positiveTtlMs) + private val negativeTtlNanos = TimeUnit.MILLISECONDS.toNanos(negativeTtlMs) + + private val cache: MutableMap = + Collections.synchronizedMap( + object : LinkedHashMap(64, 0.75f, true) { + override fun removeEldestEntry(eldest: Map.Entry): Boolean = size > maxEntries + }, + ) + private val inflight = ConcurrentHashMap>>() + + override fun lookup(hostname: String): List { + cache[hostname]?.let { entry -> + if (entry.expiresAtNanos > System.nanoTime()) { + return entry.unwrap(hostname) + } + } + + val newFuture = CompletableFuture>() + val existing = inflight.putIfAbsent(hostname, newFuture) + return if (existing == null) { + resolveAsLeader(hostname, newFuture) + } else { + awaitFollower(hostname, existing) + } + } + + private fun resolveAsLeader( + hostname: String, + future: CompletableFuture>, + ): List { + try { + val addresses = delegate.lookup(hostname) + put(hostname, addresses, positiveTtlNanos) + future.complete(addresses) + return addresses + } catch (e: UnknownHostException) { + put(hostname, emptyList(), negativeTtlNanos) + future.completeExceptionally(e) + throw e + } catch (e: Throwable) { + future.completeExceptionally(e) + throw e + } finally { + inflight.remove(hostname, future) + } + } + + private fun awaitFollower( + hostname: String, + future: CompletableFuture>, + ): List { + try { + val addresses = future.get() + return addresses.ifEmpty { throw UnknownHostException(hostname) } + } catch (e: ExecutionException) { + when (val cause = e.cause) { + is UnknownHostException -> throw cause + null -> throw UnknownHostException(hostname) + else -> throw UnknownHostException(hostname).apply { initCause(cause) } + } + } catch (e: InterruptedException) { + Thread.currentThread().interrupt() + throw UnknownHostException(hostname).apply { initCause(e) } + } + } + + private fun put( + hostname: String, + addresses: List, + ttlNanos: Long, + ) { + cache[hostname] = Entry(addresses, System.nanoTime() + ttlNanos) + } + + /** Drop all cached entries. Call when the network changes (e.g. WiFi <-> mobile). */ + fun invalidate() { + cache.clear() + } + + /** Drop a single host's cached entry. */ + fun invalidate(hostname: String) { + cache.remove(hostname) + } + + private class Entry( + val addresses: List, + val expiresAtNanos: Long, + ) { + fun unwrap(hostname: String): List = addresses.ifEmpty { throw UnknownHostException(hostname) } + } + + companion object { + /** + * 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. + */ + val shared: AmethystDns by lazy { AmethystDns() } + } +} diff --git a/amethyst/src/main/java/com/vitorpamplona/amethyst/service/okhttp/OkHttpClientFactory.kt b/amethyst/src/main/java/com/vitorpamplona/amethyst/service/okhttp/OkHttpClientFactory.kt index 61f3255f62..6150ad33ff 100644 --- a/amethyst/src/main/java/com/vitorpamplona/amethyst/service/okhttp/OkHttpClientFactory.kt +++ b/amethyst/src/main/java/com/vitorpamplona/amethyst/service/okhttp/OkHttpClientFactory.kt @@ -63,6 +63,7 @@ class OkHttpClientFactory( .Builder() .dispatcher(dispatcher) .connectionPool(connectionPool) + .dns(AmethystDns.shared) .eventListenerFactory(MediaCallEventListenerFactory(dispatcher, connectionPool)) .followRedirects(true) .followSslRedirects(true) diff --git a/amethyst/src/main/java/com/vitorpamplona/amethyst/service/okhttp/OkHttpClientFactoryForRelays.kt b/amethyst/src/main/java/com/vitorpamplona/amethyst/service/okhttp/OkHttpClientFactoryForRelays.kt index e4b81b31fb..fb63b34165 100644 --- a/amethyst/src/main/java/com/vitorpamplona/amethyst/service/okhttp/OkHttpClientFactoryForRelays.kt +++ b/amethyst/src/main/java/com/vitorpamplona/amethyst/service/okhttp/OkHttpClientFactoryForRelays.kt @@ -55,6 +55,7 @@ class OkHttpClientFactoryForRelays( OkHttpClient .Builder() .dispatcher(myDispatcher) + .dns(AmethystDns.shared) .followRedirects(true) .followSslRedirects(true) .addInterceptor(DefaultContentTypeInterceptor(userAgent)) 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 new file mode 100644 index 0000000000..0c0b88554a --- /dev/null +++ b/amethyst/src/test/java/com/vitorpamplona/amethyst/service/okhttp/AmethystDnsTest.kt @@ -0,0 +1,204 @@ +/* + * Copyright (c) 2025 Vitor Pamplona + * + * Permission is hereby granted, free of charge, to any person obtaining a copy of + * this software and associated documentation files (the "Software"), to deal in + * the Software without restriction, including without limitation the rights to use, + * copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the + * Software, and to permit persons to whom the Software is furnished to do so, + * subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in all + * copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS + * FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR + * COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN + * AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION + * WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. + */ +package com.vitorpamplona.amethyst.service.okhttp + +import okhttp3.Dns +import org.junit.Assert.assertEquals +import org.junit.Assert.assertSame +import org.junit.Assert.assertThrows +import org.junit.Assert.assertTrue +import org.junit.Test +import java.net.InetAddress +import java.net.UnknownHostException +import java.util.concurrent.CountDownLatch +import java.util.concurrent.Executors +import java.util.concurrent.TimeUnit +import java.util.concurrent.atomic.AtomicInteger + +class AmethystDnsTest { + private fun ip(value: String) = InetAddress.getByName(value) + + private class CountingDns( + private val responses: Map>, + ) : Dns { + val callsByHost = mutableMapOf() + + override fun lookup(hostname: String): List { + callsByHost.getOrPut(hostname) { AtomicInteger() }.incrementAndGet() + return responses[hostname] ?: throw UnknownHostException(hostname) + } + + fun calls(hostname: String): Int = callsByHost[hostname]?.get() ?: 0 + } + + private class GatedDns( + private val responses: Map>, + ) : Dns { + val started = CountDownLatch(1) + val release = CountDownLatch(1) + val calls = AtomicInteger() + + override fun lookup(hostname: String): List { + calls.incrementAndGet() + started.countDown() + release.await() + return responses[hostname] ?: throw UnknownHostException(hostname) + } + } + + @Test + fun `cache hit avoids second upstream call`() { + val upstream = CountingDns(mapOf("a.example" to listOf(ip("1.2.3.4")))) + val dns = AmethystDns(delegate = upstream) + + val first = dns.lookup("a.example") + val second = dns.lookup("a.example") + + assertEquals(listOf(ip("1.2.3.4")), first) + assertSame(first, second) + assertEquals(1, upstream.calls("a.example")) + } + + @Test + fun `negative cache short-circuits subsequent lookups`() { + val upstream = CountingDns(emptyMap()) + val dns = AmethystDns(delegate = upstream) + + assertThrows(UnknownHostException::class.java) { dns.lookup("missing.example") } + assertThrows(UnknownHostException::class.java) { dns.lookup("missing.example") } + + assertEquals(1, upstream.calls("missing.example")) + } + + @Test + fun `positive entries expire`() { + val upstream = CountingDns(mapOf("a.example" to listOf(ip("1.2.3.4")))) + val dns = + AmethystDns( + delegate = upstream, + positiveTtlMs = 1, + negativeTtlMs = 1, + ) + + dns.lookup("a.example") + Thread.sleep(20) + dns.lookup("a.example") + + assertEquals(2, upstream.calls("a.example")) + } + + @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")))) + val dns = AmethystDns(delegate = gated) + val pool = Executors.newFixedThreadPool(8) + + try { + val results = (1..8).map { pool.submit> { dns.lookup("hot.example") } } + assertTrue( + "Leader should have started the upstream lookup", + gated.started.await(2, TimeUnit.SECONDS), + ) + gated.release.countDown() + + results.forEach { + assertEquals(listOf(ip("9.9.9.9")), it.get(2, TimeUnit.SECONDS)) + } + assertEquals(1, gated.calls.get()) + } finally { + pool.shutdownNow() + } + } + + @Test + fun `lookups for different hosts run in parallel`() { + val responses = mapOf("a" to listOf(ip("1.1.1.1")), "b" to listOf(ip("2.2.2.2"))) + val parallelism = AtomicInteger() + val peak = AtomicInteger() + val release = CountDownLatch(1) + + val instrumented = + Dns { hostname -> + val now = parallelism.incrementAndGet() + peak.updateAndGet { maxOf(it, now) } + try { + release.await(2, TimeUnit.SECONDS) + responses[hostname] ?: throw UnknownHostException(hostname) + } finally { + parallelism.decrementAndGet() + } + } + val dns = AmethystDns(delegate = instrumented) + val pool = Executors.newFixedThreadPool(2) + + try { + val futureA = pool.submit> { dns.lookup("a") } + val futureB = pool.submit> { dns.lookup("b") } + + // Wait briefly for both threads to enter the resolver, then let them out. + val deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(2) + while (peak.get() < 2 && System.nanoTime() < deadline) { + Thread.sleep(5) + } + release.countDown() + + futureA.get(2, TimeUnit.SECONDS) + futureB.get(2, TimeUnit.SECONDS) + + assertEquals("Both hosts should resolve concurrently", 2, peak.get()) + } finally { + pool.shutdownNow() + } + } + + @Test + fun `invalidate clears cache so next lookup hits upstream`() { + val upstream = CountingDns(mapOf("a.example" to listOf(ip("1.2.3.4")))) + val dns = AmethystDns(delegate = upstream) + + dns.lookup("a.example") + dns.invalidate() + dns.lookup("a.example") + + assertEquals(2, upstream.calls("a.example")) + } + + @Test + fun `invalidate by host removes only that entry`() { + val upstream = + CountingDns( + mapOf( + "a.example" to listOf(ip("1.2.3.4")), + "b.example" to listOf(ip("5.6.7.8")), + ), + ) + val dns = AmethystDns(delegate = upstream) + + dns.lookup("a.example") + dns.lookup("b.example") + dns.invalidate("a.example") + dns.lookup("a.example") + dns.lookup("b.example") + + assertEquals(2, upstream.calls("a.example")) + assertEquals(1, upstream.calls("b.example")) + } +}