Cache DNS answers and paint saved people before launch revalidation

conductarr-client/src/main/kotlin/org/justinvisser/wolfskin/conductarr/ConductarrClient.kt

@@ -67,6 +67,7 @@ class ConductarrClient(
     timeouts: ClientTimeouts = ClientTimeouts(),
     private val nowMillis: () -> Long = System::currentTimeMillis,
     private val onEvent: (ClientEvent) -> Unit = {},
+    dns: PersistentDns? = null,
 ) : Closeable {
     private val base = baseUrl.toHttpUrl()
     private val scope: String
@@ -74,6 +75,7 @@ class ConductarrClient(
     private val cacheIdentityLock = Any()
     private val catalogueReads = Semaphore(2)
     private val http = OkHttpClient.Builder()
+        .apply { if (dns != null) DnsConnectionRetry.configure(this, dns) }
         .connectTimeout(timeouts.connectSeconds, TimeUnit.SECONDS)
         .readTimeout(timeouts.readSeconds, TimeUnit.SECONDS)
         .callTimeout(timeouts.callSeconds, TimeUnit.SECONDS)

conductarr-client/src/main/kotlin/org/justinvisser/wolfskin/conductarr/DnsConnectionRetry.kt

@@ -0,0 +1,59 @@
+// SPDX-License-Identifier: GPL-2.0-only
+package org.justinvisser.wolfskin.conductarr
+
+import java.io.IOException
+import java.net.InetSocketAddress
+import java.net.Proxy
+import java.util.concurrent.ConcurrentHashMap
+import okhttp3.Call
+import okhttp3.EventListener
+import okhttp3.Interceptor
+import okhttp3.OkHttpClient
+import okhttp3.Protocol
+import okhttp3.Response
+
+/** Retry once with fresh DNS only if connection establishment failed before any HTTP headers were sent. */
+class DnsConnectionRetry(private val dns: PersistentDns) : Interceptor, EventListener.Factory {
+    private class Attempt {
+        @Volatile var connectionFailed = false
+        @Volatile var headersSent = false
+    }
+    private val attempts = ConcurrentHashMap<Call, Attempt>()
+
+    override fun create(call: Call): EventListener {
+        val attempt = Attempt()
+        attempts[call] = attempt
+        return object : EventListener() {
+            override fun connectFailed(call: Call, inetSocketAddress: InetSocketAddress, proxy: Proxy, protocol: Protocol?, ioe: IOException) {
+                attempt.connectionFailed = true
+            }
+            override fun requestHeadersStart(call: Call) { attempt.headersSent = true }
+            override fun callEnd(call: Call) { attempts.remove(call) }
+            override fun callFailed(call: Call, ioe: IOException) { attempts.remove(call) }
+        }
+    }
+
+    override fun intercept(chain: Interceptor.Chain): Response {
+        val attempt = attempts.getValue(chain.call())
+        return try {
+            chain.proceed(chain.request())
+        } catch (failure: IOException) {
+            if (!attempt.connectionFailed || attempt.headersSent || chain.call().isCanceled()) throw failure
+            try {
+                dns.lookupFresh(chain.request().url.host)
+            } catch (lookupFailure: IOException) {
+                lookupFailure.addSuppressed(failure)
+                throw lookupFailure
+            }
+            chain.proceed(chain.request())
+        }
+    }
+
+    companion object {
+        /** Install the same resolver and its pre-request recovery on a separately owned client. */
+        fun configure(builder: OkHttpClient.Builder, dns: PersistentDns): OkHttpClient.Builder {
+            val retry = DnsConnectionRetry(dns)
+            return builder.dns(dns).eventListenerFactory(retry).addInterceptor(retry)
+        }
+    }
+}

conductarr-client/src/main/kotlin/org/justinvisser/wolfskin/conductarr/PersistentDns.kt

@@ -0,0 +1,142 @@
+// SPDX-License-Identifier: GPL-2.0-only
+package org.justinvisser.wolfskin.conductarr
+
+import java.io.File
+import java.io.IOException
+import java.net.InetAddress
+import java.net.UnknownHostException
+import java.util.concurrent.ConcurrentHashMap
+import java.util.concurrent.Executor
+import java.util.concurrent.Executors
+import kotlinx.serialization.Serializable
+import kotlinx.serialization.Transient
+import kotlinx.serialization.encodeToString
+import okhttp3.Dns
+
+/** Last-known system answers stay usable for one day while a single background lookup refreshes them. */
+class PersistentDns(
+    private val file: () -> File,
+    private val resolver: Dns = Dns.SYSTEM,
+    private val executor: Executor = Executors.newFixedThreadPool(2) { task -> Thread(task, "wolfskin-dns").apply { isDaemon = true } },
+    private val nowMillis: () -> Long = System::currentTimeMillis,
+) : Dns {
+    private class Host {
+        @Volatile var answer: Answer? = null
+        @Volatile var refreshAfter = 0L
+        var refreshing = false
+        val resolutionLock = Any()
+    }
+
+    @Serializable
+    private data class Answer(
+        val addresses: List<String>,
+        val savedAtMillis: Long,
+        @Transient val resolvedAddresses: List<InetAddress> = decodeAddresses(addresses),
+    )
+    private val hosts = ConcurrentHashMap<String, Host>()
+    private val diskLock = Any()
+    @Volatile private var loaded = false
+
+    override fun lookup(hostname: String): List<InetAddress> {
+        load()
+        val host = hosts.computeIfAbsent(hostname) { Host() }
+        val saved = host.answer
+        if (saved != null && nowMillis() - saved.savedAtMillis in 0..MAX_STALENESS_MILLIS) {
+            refresh(hostname, host)
+            return saved.resolvedAddresses
+        }
+        return synchronized(host.resolutionLock) {
+            val current = host.answer
+            if (current != null && current !== saved && nowMillis() - current.savedAtMillis in 0..MAX_STALENESS_MILLIS) {
+                current.resolvedAddresses
+            } else resolve(hostname, host)
+        }
+    }
+
+    /** A failed connection bypasses the cached answer and any completed background refresh. */
+    fun lookupFresh(hostname: String): List<InetAddress> {
+        load()
+        val host = hosts.computeIfAbsent(hostname) { Host() }
+        return synchronized(host.resolutionLock) {
+            host.answer = null
+            resolve(hostname, host)
+        }
+    }
+
+    private fun refresh(hostname: String, host: Host) {
+        synchronized(host) {
+            if (host.refreshing || nowMillis() < host.refreshAfter) return
+            host.refreshing = true
+        }
+        executor.execute {
+            try {
+                synchronized(host.resolutionLock) { resolve(hostname, host) }
+            } catch (_: UnknownHostException) {
+                // Failed refreshes do not extend the lifetime of the last successful answer.
+                host.refreshAfter = nowMillis() + FAILURE_BACKOFF_MILLIS
+            } finally {
+                synchronized(host) { host.refreshing = false }
+            }
+        }
+    }
+
+    private fun resolve(hostname: String, host: Host): List<InetAddress> {
+        val addresses = resolver.lookup(hostname)
+        if (addresses.isEmpty()) throw UnknownHostException(hostname)
+        host.answer = Answer(addresses.map { it.hostAddress!! }, nowMillis(), addresses)
+        host.refreshAfter = nowMillis() + REFRESH_INTERVAL_MILLIS
+        persist()
+        return addresses
+    }
+
+    private fun load() {
+        if (loaded) return
+        synchronized(diskLock) {
+            if (loaded) return
+            try {
+                val saved = file()
+                if (saved.isFile && saved.length() <= MAX_FILE_BYTES) {
+                    val entries = WireJson.decodeFromString<Map<String, Answer>>(saved.readText())
+                    require(entries.size <= MAX_HOSTS)
+                    entries.forEach { (name, answer) -> hosts[name] = Host().apply { this.answer = answer } }
+                }
+            } catch (_: IOException) {
+                // A missing or damaged hint file cannot prevent normal system resolution.
+            } catch (_: IllegalArgumentException) {
+                // Serialization and literal-validation errors invalidate the entire hint file.
+            }
+            loaded = true
+        }
+    }
+
+    private fun persist() = synchronized(diskLock) {
+        try {
+            val entries = hosts.entries.mapNotNull { (name, host) -> host.answer?.let { name to it } }
+                .sortedByDescending { it.second.savedAtMillis }.take(MAX_HOSTS).toMap()
+            val target = file()
+            target.parentFile?.mkdirs()
+            val temporary = File(target.path + ".tmp")
+            temporary.writeText(WireJson.encodeToString(entries))
+            if (!temporary.renameTo(target)) temporary.delete()
+        } catch (_: IOException) {
+            // The in-memory answer remains usable when private storage is unavailable.
+        }
+    }
+
+    companion object {
+        const val MAX_STALENESS_MILLIS = 24 * 60 * 60 * 1000L
+        private const val REFRESH_INTERVAL_MILLIS = 60_000L
+        private const val FAILURE_BACKOFF_MILLIS = 2_000L
+        private const val MAX_HOSTS = 32
+        private const val MAX_ADDRESSES = 32
+        private const val MAX_FILE_BYTES = 64 * 1024L
+        private fun decodeAddresses(addresses: List<String>): List<InetAddress> {
+            require(addresses.isNotEmpty() && addresses.size <= MAX_ADDRESSES)
+            return addresses.map { address ->
+                // Numeric literals only: decoding the hint file must never perform DNS.
+                require(address.contains(':') || address.matches(Regex("[0-9]+\\.[0-9]+\\.[0-9]+\\.[0-9]+")))
+                InetAddress.getByName(address)
+            }
+        }
+    }
+}

conductarr-client/src/test/kotlin/org/justinvisser/wolfskin/conductarr/ClientSmoke.kt

@@ -2,6 +2,10 @@
 package org.justinvisser.wolfskin.conductarr
 
 import java.nio.file.Files
+import java.io.File
+import java.net.InetAddress
+import java.util.concurrent.Executor
+import okhttp3.Dns
 import kotlinx.coroutines.flow.first
 import kotlinx.coroutines.flow.toList
 import kotlinx.coroutines.runBlocking
@@ -14,13 +18,24 @@ fun main() = runBlocking {
     try {
         MockWebServer().use { server ->
             server.start()
-            val baseUrl = server.url("/").toString()
+            val baseUrl = server.url("/").newBuilder().host("localhost").build().toString()
+            var resolutions = 0
+            val dead = InetAddress.getByName("127.0.0.2")
+            val live = InetAddress.getByName("127.0.0.1")
+            val dns = PersistentDns({ File(directory, "dns.json") }, Dns {
+                resolutions++
+                if (resolutions == 1) listOf(dead) else listOf(live)
+            }, Executor { it.run() })
+            check(dns.lookup("localhost") == listOf(dead))
             server.enqueue(MockResponse.Builder().body(PROFILES_JSON).addHeader("ETag", "\"smoke\"").build())
-            ConductarrClient(baseUrl, TEST_TOKEN, FileCacheStore(directory)).use { client ->
+            ConductarrClient(baseUrl, TEST_TOKEN, FileCacheStore(directory), dns = dns).use { client ->
                 val live = client.read(TvRoutes.profiles()).toList().single() as ReadResult.Value
                 check(live.saved.value!!.profiles!!.single().name == "Justin")
+                check(resolutions == 2)
+                println("DNS SMOKE: cached 127.0.0.2 refused -> fresh 127.0.0.1 -> profile Justin; requests=${server.requestCount}, resolutions=$resolutions")
             }
-            ConductarrClient(baseUrl, TEST_TOKEN, FileCacheStore(directory)).use { client ->
+            val reopened = PersistentDns({ File(directory, "dns.json") }, Dns { listOf(live) }, Executor { it.run() })
+            ConductarrClient(baseUrl, TEST_TOKEN, FileCacheStore(directory), dns = reopened).use { client ->
                 val saved = client.read(TvRoutes.profiles(), ReadTrigger.CACHE_ONLY).first() as ReadResult.Value
                 check(saved.saved.fromDisk)
                 server.enqueue(MockResponse.Builder().code(304).build())

conductarr-client/src/test/kotlin/org/justinvisser/wolfskin/conductarr/PersistentDnsTest.kt

@@ -0,0 +1,144 @@
+// SPDX-License-Identifier: GPL-2.0-only
+package org.justinvisser.wolfskin.conductarr
+
+import java.io.File
+import java.net.InetAddress
+import java.net.Proxy
+import java.net.UnknownHostException
+import java.util.concurrent.Executor
+import java.util.concurrent.TimeUnit
+import mockwebserver3.MockResponse
+import mockwebserver3.MockWebServer
+import okhttp3.ConnectionPool
+import okhttp3.Dns
+import okhttp3.OkHttpClient
+import okhttp3.Request
+import okhttp3.RequestBody.Companion.toRequestBody
+import org.junit.Assert.*
+import org.junit.Rule
+import org.junit.Test
+import org.junit.rules.TemporaryFolder
+
+/** Fake time and queued refreshes prove stale answers, disk hints and dead-address recovery. */
+class PersistentDnsTest {
+    @get:Rule val directory = TemporaryFolder()
+    private var now = 1_000_000L
+    private val queued = ArrayDeque<Runnable>()
+    private val background = Executor { queued.add(it) }
+    private val old = InetAddress.getByName("127.0.0.2")
+    private val fresh = InetAddress.getByName("127.0.0.1")
+    private var answer = listOf(old)
+    private var lookups = 0
+    private val resolver = Dns { lookups++; answer }
+    private val file get() = File(directory.root, "dns.json")
+    private fun cache(resolve: Dns = resolver) = PersistentDns({ file }, resolve, background, { now })
+
+    @Test fun staleAnswerReturnsBeforeBackgroundRefreshAndNextLookupUsesChangedIp() {
+        val dns = cache()
+        assertEquals(listOf(old), dns.lookup("jellyfin.test"))
+        now += 120_000
+        answer = listOf(fresh)
+        assertEquals(listOf(old), dns.lookup("jellyfin.test"))
+        assertEquals(1, lookups)
+        assertEquals(listOf(old), dns.lookup("jellyfin.test"))
+        assertEquals(1, queued.size)
+        queued.removeFirst().run()
+        assertEquals(listOf(fresh), dns.lookup("jellyfin.test"))
+        assertEquals(2, lookups)
+    }
+
+    @Test fun answersOlderThanOneDayCannotBeServedEvenWhenRefreshFails() {
+        val dns = cache()
+        dns.lookup("jellyfin.test")
+        now += PersistentDns.MAX_STALENESS_MILLIS + 1
+        val expired = cache(Dns { throw UnknownHostException() })
+        assertThrows(UnknownHostException::class.java) { expired.lookup("jellyfin.test") }
+    }
+
+    @Test fun persistedAnswersRetainTimestampAndSystemAddressOrdering() {
+        answer = listOf(InetAddress.getByName("::1"), fresh)
+        cache().lookup("jellyfin.test")
+        now += 120_000
+        val reopened = cache(Dns { error("Disk answer must return before resolving") })
+        assertEquals(answer, reopened.lookup("jellyfin.test"))
+        assertEquals(1, queued.size)
+        now += PersistentDns.MAX_STALENESS_MILLIS
+        assertThrows(UnknownHostException::class.java) {
+            cache(Dns { throw UnknownHostException() }).lookup("jellyfin.test")
+        }
+    }
+
+    @Test fun corruptOrNonnumericDiskAnswersAreIgnored() {
+        for (contents in listOf("not json", "{\"jellyfin.test\":{\"addresses\":[\"not-an-ip.test\"],\"savedAtMillis\":1000000}}")) {
+            file.writeText(contents)
+            answer = listOf(fresh)
+            assertEquals(listOf(fresh), cache().lookup("jellyfin.test"))
+        }
+        assertEquals(2, lookups)
+    }
+
+    @Test fun negativeAnswersDoNotDelayRecoveryAndFailedRefreshDoesNotExtendStaleness() {
+        var offline = true
+        val dns = cache(Dns { if (offline) throw UnknownHostException(); listOf(fresh) })
+        assertThrows(UnknownHostException::class.java) { dns.lookup("jellyfin.test") }
+        offline = false
+        assertEquals(listOf(fresh), dns.lookup("jellyfin.test"))
+        offline = true
+        now += 120_000
+        assertEquals(listOf(fresh), dns.lookup("jellyfin.test"))
+        queued.removeFirst().run()
+        assertEquals(listOf(fresh), dns.lookup("jellyfin.test"))
+        assertTrue(queued.isEmpty())
+        now += 2_001
+        dns.lookup("jellyfin.test")
+        assertEquals(1, queued.size)
+    }
+
+    @Test fun failedCachedAddressForcesFreshResolutionAndRetriesBeforeSendingPost() {
+        MockWebServer().use { server ->
+            server.start()
+            val dns = cache()
+            dns.lookup("jellyfin.test")
+            answer = listOf(fresh)
+            server.enqueue(MockResponse.Builder().body("new address").build())
+            val client = DnsConnectionRetry.configure(OkHttpClient.Builder(), dns)
+                .proxy(Proxy.NO_PROXY).retryOnConnectionFailure(false).connectTimeout(1, TimeUnit.SECONDS)
+                .connectionPool(ConnectionPool(0, 1, TimeUnit.MILLISECONDS)).build()
+            try {
+                val url = server.url("/").newBuilder().host("jellyfin.test").build()
+                client.newCall(Request.Builder().url(url).post("body".toRequestBody()).build()).execute().use {
+                    assertEquals("new address", it.body.string())
+                }
+                assertEquals(2, lookups)
+                assertEquals(1, server.requestCount)
+                assertEquals(listOf(fresh), dns.lookup("jellyfin.test"))
+            } finally {
+                client.connectionPool.evictAll()
+                client.dispatcher.executorService.shutdown()
+            }
+        }
+    }
+
+    @Test fun backgroundIpChangeIsUsedByTheNextHttpCall() {
+        MockWebServer().use { server ->
+            server.start()
+            val dns = cache()
+            dns.lookup("jellyfin.test")
+            now += 120_000
+            answer = listOf(fresh)
+            assertEquals(listOf(old), dns.lookup("jellyfin.test"))
+            queued.removeFirst().run()
+            server.enqueue(MockResponse.Builder().body("refreshed").build())
+            val client = DnsConnectionRetry.configure(OkHttpClient.Builder(), dns).proxy(Proxy.NO_PROXY).build()
+            try {
+                val url = server.url("/").newBuilder().host("jellyfin.test").build()
+                client.newCall(Request.Builder().url(url).build()).execute().use { assertEquals("refreshed", it.body.string()) }
+                assertEquals(2, lookups)
+                assertEquals(1, server.requestCount)
+            } finally {
+                client.connectionPool.evictAll()
+                client.dispatcher.executorService.shutdown()
+            }
+        }
+    }
+}