mirror of
https://github.com/vitorpamplona/amethyst.git
synced 2026-08-09 16:14:40 +00:00
feat(okhttp): concurrent caching DNS resolver
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.
This commit is contained in:
Claude
@@ -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<String, Entry> =
|
||||
Collections.synchronizedMap(
|
||||
object : LinkedHashMap<String, Entry>(64, 0.75f, true) {
|
||||
override fun removeEldestEntry(eldest: Map.Entry<String, Entry>): Boolean = size > maxEntries
|
||||
},
|
||||
)
|
||||
private val inflight = ConcurrentHashMap<String, CompletableFuture<List<InetAddress>>>()
|
||||
|
||||
override fun lookup(hostname: String): List<InetAddress> {
|
||||
cache[hostname]?.let { entry ->
|
||||
if (entry.expiresAtNanos > System.nanoTime()) {
|
||||
return entry.unwrap(hostname)
|
||||
}
|
||||
}
|
||||
|
||||
val newFuture = CompletableFuture<List<InetAddress>>()
|
||||
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<InetAddress>>,
|
||||
): List<InetAddress> {
|
||||
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<InetAddress>>,
|
||||
): List<InetAddress> {
|
||||
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<InetAddress>,
|
||||
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<InetAddress>,
|
||||
val expiresAtNanos: Long,
|
||||
) {
|
||||
fun unwrap(hostname: String): List<InetAddress> = 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() }
|
||||
}
|
||||
}
|
||||
+1
@@ -63,6 +63,7 @@ class OkHttpClientFactory(
|
||||
.Builder()
|
||||
.dispatcher(dispatcher)
|
||||
.connectionPool(connectionPool)
|
||||
.dns(AmethystDns.shared)
|
||||
.eventListenerFactory(MediaCallEventListenerFactory(dispatcher, connectionPool))
|
||||
.followRedirects(true)
|
||||
.followSslRedirects(true)
|
||||
|
||||
+1
@@ -55,6 +55,7 @@ class OkHttpClientFactoryForRelays(
|
||||
OkHttpClient
|
||||
.Builder()
|
||||
.dispatcher(myDispatcher)
|
||||
.dns(AmethystDns.shared)
|
||||
.followRedirects(true)
|
||||
.followSslRedirects(true)
|
||||
.addInterceptor(DefaultContentTypeInterceptor(userAgent))
|
||||
|
||||
@@ -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<String, List<InetAddress>>,
|
||||
) : Dns {
|
||||
val callsByHost = mutableMapOf<String, AtomicInteger>()
|
||||
|
||||
override fun lookup(hostname: String): List<InetAddress> {
|
||||
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<String, List<InetAddress>>,
|
||||
) : Dns {
|
||||
val started = CountDownLatch(1)
|
||||
val release = CountDownLatch(1)
|
||||
val calls = AtomicInteger()
|
||||
|
||||
override fun lookup(hostname: String): List<InetAddress> {
|
||||
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<List<InetAddress>> { 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<List<InetAddress>> { dns.lookup("a") }
|
||||
val futureB = pool.submit<List<InetAddress>> { 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"))
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user