diff --git a/purchases/src/main/kotlin/com/revenuecat/purchases/PurchasesFactory.kt b/purchases/src/main/kotlin/com/revenuecat/purchases/PurchasesFactory.kt index 89f6804a9a..0191e5aa03 100644 --- a/purchases/src/main/kotlin/com/revenuecat/purchases/PurchasesFactory.kt +++ b/purchases/src/main/kotlin/com/revenuecat/purchases/PurchasesFactory.kt @@ -30,6 +30,7 @@ import com.revenuecat.purchases.common.events.EventsManager import com.revenuecat.purchases.common.isDeviceProtectedStorageCompat import com.revenuecat.purchases.common.log import com.revenuecat.purchases.common.networking.ETagManager +import com.revenuecat.purchases.common.networking.HTTPTimeoutManager import com.revenuecat.purchases.common.offerings.OfferingsCache import com.revenuecat.purchases.common.offerings.OfferingsFactory import com.revenuecat.purchases.common.offerings.OfferingsManager @@ -37,6 +38,7 @@ import com.revenuecat.purchases.common.offlineentitlements.OfflineCustomerInfoCa import com.revenuecat.purchases.common.offlineentitlements.OfflineEntitlementsManager import com.revenuecat.purchases.common.offlineentitlements.PurchasedProductsFetcher import com.revenuecat.purchases.common.remoteconfig.DefaultRemoteConfigSourceProvider +import com.revenuecat.purchases.common.remoteconfig.RemoteConfigBlobFetcher import com.revenuecat.purchases.common.remoteconfig.RemoteConfigBlobStore import com.revenuecat.purchases.common.remoteconfig.RemoteConfigDiskCache import com.revenuecat.purchases.common.remoteconfig.RemoteConfigManager @@ -212,6 +214,7 @@ internal class PurchasesFactory( } val apiSourceProvider = DefaultRemoteConfigSourceProvider(remoteConfigTopicStore) + val timeoutManager = HTTPTimeoutManager(appConfig) val httpClient = HTTPClient( appConfig, eTagManager, @@ -221,6 +224,7 @@ internal class PurchasesFactory( apiSourceProvider, localeProvider = localeProvider, forceServerErrorStrategy = forceServerErrorStrategy, + timeoutManager = timeoutManager, ) val backendHelper = BackendHelper(apiKey, backendDispatcher, appConfig, httpClient) val backend = Backend( @@ -287,12 +291,14 @@ internal class PurchasesFactory( ) val remoteConfigManager = if (remoteConfigDiskCache != null) { + val remoteConfigBlobStore = RemoteConfigBlobStore(contextForStorage) RemoteConfigManager( backend = backend, diskCache = remoteConfigDiskCache, - blobStore = RemoteConfigBlobStore(contextForStorage), + blobStore = remoteConfigBlobStore, topicStore = remoteConfigTopicStore, sourceProvider = apiSourceProvider, + blobFetcher = RemoteConfigBlobFetcher(remoteConfigBlobStore, apiSourceProvider, timeoutManager), // Bootstrap source for a cold on-demand read's self-triggered sync (see blobData()); after // the first identity change the manager syncs for the user clearCache() binds instead. appUserIDProvider = { cache.getCachedAppUserID() }, diff --git a/purchases/src/main/kotlin/com/revenuecat/purchases/common/remoteconfig/RemoteConfigBlobFetcher.kt b/purchases/src/main/kotlin/com/revenuecat/purchases/common/remoteconfig/RemoteConfigBlobFetcher.kt index cb52eb2ff6..44f83d736b 100644 --- a/purchases/src/main/kotlin/com/revenuecat/purchases/common/remoteconfig/RemoteConfigBlobFetcher.kt +++ b/purchases/src/main/kotlin/com/revenuecat/purchases/common/remoteconfig/RemoteConfigBlobFetcher.kt @@ -1,6 +1,7 @@ package com.revenuecat.purchases.common.remoteconfig import com.revenuecat.purchases.common.errorLog +import com.revenuecat.purchases.common.networking.HTTPTimeoutManager import com.revenuecat.purchases.common.remoteconfig.RemoteConfigSourceHandle.Purpose import com.revenuecat.purchases.common.verboseLog import com.revenuecat.purchases.utils.DefaultUrlConnectionFactory @@ -17,6 +18,9 @@ import kotlinx.coroutines.coroutineScope import kotlinx.coroutines.launch import java.io.IOException import java.net.HttpURLConnection +import java.net.MalformedURLException +import java.net.SocketTimeoutException +import java.net.URL import java.nio.ByteBuffer import java.util.PriorityQueue @@ -45,6 +49,7 @@ private const val BLOB_REF_PLACEHOLDER = "{blob_ref}" internal class RemoteConfigBlobFetcher( private val blobStore: RemoteConfigBlobStore, private val sourceProvider: RemoteConfigSourceProvider, + private val timeoutManager: HTTPTimeoutManager, private val urlConnectionFactory: UrlConnectionFactory = DefaultUrlConnectionFactory(), private val scope: CoroutineScope = CoroutineScope(SupervisorJob() + Dispatchers.IO), ) { @@ -208,11 +213,20 @@ internal class RemoteConfigBlobFetcher( } private fun tryDownloadVerifyStore(url: String, ref: String): DownloadOutcome { + val host = hostOf(url) + val timeout = timeoutManager.getTimeoutForRequest( + host = host, + isFallback = false, + endpointSupportsFallbackURLs = false, + isProxied = false, + ).toInt() + var connection: UrlConnection? = null return try { - connection = urlConnectionFactory.createConnection(url) + connection = urlConnectionFactory.createConnection(url, timeout, timeout) when (val code = connection.responseCode) { HttpURLConnection.HTTP_OK -> { + timeoutManager.recordRequestResult(host, HTTPTimeoutManager.RequestResult.SUCCESS_ON_MAIN_BACKEND) val bytes = connection.inputStream.use { it.readBytes() } verifyAndStore(bytes, ref, url) } @@ -225,6 +239,10 @@ internal class RemoteConfigBlobFetcher( DownloadOutcome.SOURCE_UNHEALTHY } } + } catch (e: SocketTimeoutException) { + errorLog(e) { "Timed out downloading remote config blob '$ref' from $url." } + timeoutManager.recordRequestResult(host, HTTPTimeoutManager.RequestResult.MAIN_SOURCE_TIMED_OUT) + DownloadOutcome.SOURCE_UNHEALTHY } catch (e: IOException) { errorLog(e) { "Failed to download remote config blob '$ref' from $url." } DownloadOutcome.SOURCE_UNHEALTHY @@ -233,6 +251,13 @@ internal class RemoteConfigBlobFetcher( } } + private fun hostOf(url: String): String? = try { + URL(url).host + } catch (e: MalformedURLException) { + verboseLog { "Could not resolve host for remote config blob URL $url: ${e.message}" } + null + } + private fun verifyAndStore(bytes: ByteArray, ref: String, url: String): DownloadOutcome { if (RemoteConfigUtils.contentAddressRef(bytes) != ref) { errorLog { "Remote config blob '$ref' from $url failed content-address verification." } diff --git a/purchases/src/main/kotlin/com/revenuecat/purchases/common/remoteconfig/RemoteConfigManager.kt b/purchases/src/main/kotlin/com/revenuecat/purchases/common/remoteconfig/RemoteConfigManager.kt index d92d621bb2..595f5a8a19 100644 --- a/purchases/src/main/kotlin/com/revenuecat/purchases/common/remoteconfig/RemoteConfigManager.kt +++ b/purchases/src/main/kotlin/com/revenuecat/purchases/common/remoteconfig/RemoteConfigManager.kt @@ -82,7 +82,7 @@ internal class RemoteConfigManager( private val ioDispatcher: CoroutineDispatcher = Dispatchers.IO, private val topicStore: RemoteConfigTopicStore, private val sourceProvider: RemoteConfigSourceProvider, - private val blobFetcher: RemoteConfigBlobFetcher = RemoteConfigBlobFetcher(blobStore, sourceProvider), + private val blobFetcher: RemoteConfigBlobFetcher, private val appUserIDProvider: () -> String? = { null }, ) { private val isRefreshing = AtomicBoolean(false) diff --git a/purchases/src/main/kotlin/com/revenuecat/purchases/storage/DefaultFileRepository.kt b/purchases/src/main/kotlin/com/revenuecat/purchases/storage/DefaultFileRepository.kt index 1587aec37b..84f0a4e7be 100644 --- a/purchases/src/main/kotlin/com/revenuecat/purchases/storage/DefaultFileRepository.kt +++ b/purchases/src/main/kotlin/com/revenuecat/purchases/storage/DefaultFileRepository.kt @@ -8,6 +8,7 @@ import com.revenuecat.purchases.common.currentLogHandler import com.revenuecat.purchases.common.verboseLog import com.revenuecat.purchases.models.Checksum import com.revenuecat.purchases.models.toHexString +import com.revenuecat.purchases.utils.DEFAULT_CONNECTION_TIMEOUT_MS import com.revenuecat.purchases.utils.DefaultUrlConnectionFactory import com.revenuecat.purchases.utils.UrlConnection import com.revenuecat.purchases.utils.UrlConnectionFactory @@ -130,7 +131,11 @@ internal class DefaultFileRepository( private fun downloadFile(url: URL): UrlConnection = try { verboseLog { "Downloading remote file from $url" } - val connection = urlConnectionFactory.createConnection(url.toString()) + val connection = urlConnectionFactory.createConnection( + url.toString(), + DEFAULT_CONNECTION_TIMEOUT_MS, + DEFAULT_CONNECTION_TIMEOUT_MS, + ) if (connection.responseCode != HttpURLConnection.HTTP_OK) { connection.disconnect() diff --git a/purchases/src/main/kotlin/com/revenuecat/purchases/utils/UrlConnectionFactory.kt b/purchases/src/main/kotlin/com/revenuecat/purchases/utils/UrlConnectionFactory.kt index f001be6d82..e7b85ade36 100644 --- a/purchases/src/main/kotlin/com/revenuecat/purchases/utils/UrlConnectionFactory.kt +++ b/purchases/src/main/kotlin/com/revenuecat/purchases/utils/UrlConnectionFactory.kt @@ -15,7 +15,12 @@ import java.net.URL import java.security.MessageDigest internal interface UrlConnectionFactory { - fun createConnection(url: String, requestMethod: String = "GET"): UrlConnection + fun createConnection( + url: String, + connectTimeoutMillis: Int, + readTimeoutMillis: Int, + requestMethod: String = "GET", + ): UrlConnection } internal interface UrlConnection { @@ -24,13 +29,19 @@ internal interface UrlConnection { fun disconnect() } -private const val TIMEOUT = 5000 +/** Default connect/read timeout for downloads that don't tune it per attempt. */ +internal const val DEFAULT_CONNECTION_TIMEOUT_MS = 5000 internal class DefaultUrlConnectionFactory : UrlConnectionFactory { - override fun createConnection(url: String, requestMethod: String): UrlConnection { + override fun createConnection( + url: String, + connectTimeoutMillis: Int, + readTimeoutMillis: Int, + requestMethod: String, + ): UrlConnection { val connection = URL(url).openConnection() as HttpURLConnection - connection.connectTimeout = TIMEOUT - connection.readTimeout = TIMEOUT + connection.connectTimeout = connectTimeoutMillis + connection.readTimeout = readTimeoutMillis connection.requestMethod = requestMethod connection.doInput = true return DefaultUrlConnection(connection) @@ -60,7 +71,7 @@ internal fun UrlConnectionFactory.downloadToFile( verboseLog { "Downloading $description from $url" } var connection: UrlConnection? = null try { - connection = createConnection(url) + connection = createConnection(url, DEFAULT_CONNECTION_TIMEOUT_MS, DEFAULT_CONNECTION_TIMEOUT_MS) if (connection.responseCode != HttpURLConnection.HTTP_OK) { throw IOException( "HTTP ${connection.responseCode} when downloading $description from $url", @@ -88,7 +99,7 @@ internal fun UrlConnectionFactory.downloadToFileAndVerifyChecksum( val digest = MessageDigest.getInstance(expectedChecksum.algorithm.algorithmName) var connection: UrlConnection? = null try { - connection = createConnection(url) + connection = createConnection(url, DEFAULT_CONNECTION_TIMEOUT_MS, DEFAULT_CONNECTION_TIMEOUT_MS) if (connection.responseCode != HttpURLConnection.HTTP_OK) { throw IOException( "HTTP ${connection.responseCode} when downloading $description from $url", diff --git a/purchases/src/test/java/com/revenuecat/purchases/common/remoteconfig/RemoteConfigBlobFetcherTest.kt b/purchases/src/test/java/com/revenuecat/purchases/common/remoteconfig/RemoteConfigBlobFetcherTest.kt index 24d16a7933..e4b3b62877 100644 --- a/purchases/src/test/java/com/revenuecat/purchases/common/remoteconfig/RemoteConfigBlobFetcherTest.kt +++ b/purchases/src/test/java/com/revenuecat/purchases/common/remoteconfig/RemoteConfigBlobFetcherTest.kt @@ -2,6 +2,9 @@ package com.revenuecat.purchases.common.remoteconfig import android.util.Base64 import androidx.test.ext.junit.runners.AndroidJUnit4 +import com.revenuecat.purchases.common.AppConfig +import com.revenuecat.purchases.common.DateProvider +import com.revenuecat.purchases.common.networking.HTTPTimeoutManager import com.revenuecat.purchases.common.remoteconfig.RemoteConfigSourceHandle.Purpose import com.revenuecat.purchases.utils.UrlConnection import com.revenuecat.purchases.utils.UrlConnectionFactory @@ -33,13 +36,17 @@ import org.junit.Test import org.junit.runner.RunWith import org.robolectric.annotation.Config import java.io.ByteArrayInputStream +import java.io.IOException +import java.net.SocketTimeoutException import java.nio.ByteBuffer import java.security.MessageDigest +import java.util.Date import java.util.concurrent.CopyOnWriteArrayList import java.util.concurrent.CountDownLatch import java.util.concurrent.TimeUnit import java.util.concurrent.atomic.AtomicBoolean import java.util.concurrent.atomic.AtomicInteger +import java.util.concurrent.atomic.AtomicLong import kotlin.random.Random @OptIn(ExperimentalCoroutinesApi::class) @@ -49,6 +56,8 @@ class RemoteConfigBlobFetcherTest { private lateinit var blobStore: RemoteConfigBlobStore private lateinit var urlConnectionFactory: UrlConnectionFactory + private lateinit var dateProvider: FakeDateProvider + private lateinit var timeoutManager: HTTPTimeoutManager // Real multi-threaded scope for the behavioural / real-concurrency tests; cancelled in tearDown. private val scope = CoroutineScope(SupervisorJob() + Dispatchers.IO) @@ -59,6 +68,9 @@ class RemoteConfigBlobFetcherTest { every { blobStore.contains(any()) } returns false every { blobStore.write(any(), any()) } returns true urlConnectionFactory = mockk() + val appConfig = mockk().apply { every { runningTests } returns false } + dateProvider = FakeDateProvider() + timeoutManager = HTTPTimeoutManager(appConfig, dateProvider) } @After @@ -73,7 +85,7 @@ class RemoteConfigBlobFetcherTest { assertThat(download(fetcher, REF_A)).isTrue() - verify(exactly = 0) { urlConnectionFactory.createConnection(any(), any()) } + verify(exactly = 0) { urlConnectionFactory.createConnection(any(), any(), any(), any()) } verify(exactly = 0) { blobStore.write(any(), any()) } } @@ -87,7 +99,7 @@ class RemoteConfigBlobFetcherTest { assertThat(download(fetcher, ref)).isTrue() // The placeholder is substituted with the ref before the request is made. - verify { urlConnectionFactory.createConnection(urlFor(ref), any()) } + verify { urlConnectionFactory.createConnection(urlFor(ref), any(), any(), any()) } val written = slot() verify { blobStore.write(ref, capture(written)) } assertThat(written.captured.toBytes()).isEqualTo(bytes) @@ -128,7 +140,7 @@ class RemoteConfigBlobFetcherTest { val cachedConcurrently = AtomicBoolean(false) every { blobStore.contains(ref) } answers { cachedConcurrently.get() } // The download fails, but a concurrent writer (e.g. inline extraction) lands the blob mid-fetch. - every { urlConnectionFactory.createConnection(urlFor(ref), any()) } answers { + every { urlConnectionFactory.createConnection(urlFor(ref), any(), any(), any()) } answers { cachedConcurrently.set(true) connection(code = 404) } @@ -202,7 +214,7 @@ class RemoteConfigBlobFetcherTest { verify(exactly = 0) { blobStore.write(any(), any()) } // The provider was re-armed once exhausted, so the second download reached the source again. verify { provider.restartIfExhausted(Purpose.BLOB) } - verify(exactly = 2) { urlConnectionFactory.createConnection(urlFor(ref), any()) } + verify(exactly = 2) { urlConnectionFactory.createConnection(urlFor(ref), any(), any(), any()) } } @Test @@ -217,7 +229,7 @@ class RemoteConfigBlobFetcherTest { // The source was healthy, so restartIfExhausted was a no-op and never rewound the failover. verify(exactly = 0) { provider.restart(any()) } - verify(exactly = 1) { urlConnectionFactory.createConnection(urlFor(ref), any()) } + verify(exactly = 1) { urlConnectionFactory.createConnection(urlFor(ref), any(), any(), any()) } } @Test @@ -227,7 +239,7 @@ class RemoteConfigBlobFetcherTest { val maxLive = AtomicInteger(0) val allBlocked = CountDownLatch(concurrency) val release = CountDownLatch(1) - every { urlConnectionFactory.createConnection(any(), any()) } answers { + every { urlConnectionFactory.createConnection(any(), any(), any(), any()) } answers { maxLive.accumulateAndGet(live.incrementAndGet()) { a, b -> maxOf(a, b) } allBlocked.countDown() release.await(WAIT_SECONDS, TimeUnit.SECONDS) @@ -252,7 +264,7 @@ class RemoteConfigBlobFetcherTest { val ref = refOf(bytes) stubConnection(urlFor(ref), code = 200, body = bytes) val fetcherScope = CoroutineScope(StandardTestDispatcher(testScheduler)) - val fetcher = RemoteConfigBlobFetcher(blobStore, provider(blobSource(TEMPLATE)), urlConnectionFactory, fetcherScope) + val fetcher = RemoteConfigBlobFetcher(blobStore, provider(blobSource(TEMPLATE)), timeoutManager, urlConnectionFactory, fetcherScope) // Both requests are enqueued before any worker runs (StandardTestDispatcher), so the second joins the first. val first = async { fetcher.ensureDownloaded(ref) } @@ -261,7 +273,7 @@ class RemoteConfigBlobFetcherTest { assertThat(first.await()).isTrue() assertThat(second.await()).isTrue() - verify(exactly = 1) { urlConnectionFactory.createConnection(urlFor(ref), any()) } + verify(exactly = 1) { urlConnectionFactory.createConnection(urlFor(ref), any(), any(), any()) } fetcherScope.cancel() } @@ -270,7 +282,7 @@ class RemoteConfigBlobFetcherTest { val started = CopyOnWriteArrayList() recordStartedConnections(started) val fetcherScope = CoroutineScope(StandardTestDispatcher(testScheduler)) - val fetcher = RemoteConfigBlobFetcher(blobStore, provider(blobSource(TEMPLATE)), urlConnectionFactory, fetcherScope) + val fetcher = RemoteConfigBlobFetcher(blobStore, provider(blobSource(TEMPLATE)), timeoutManager, urlConnectionFactory, fetcherScope) val backlog = (1..3).map { refOf("low-$it".toByteArray()) } val onDemand = refOf("on-demand".toByteArray()) @@ -305,7 +317,7 @@ class RemoteConfigBlobFetcherTest { val started = CopyOnWriteArrayList() recordStartedConnections(started) val fetcherScope = CoroutineScope(StandardTestDispatcher(testScheduler)) - val fetcher = RemoteConfigBlobFetcher(blobStore, provider(blobSource(TEMPLATE)), urlConnectionFactory, fetcherScope) + val fetcher = RemoteConfigBlobFetcher(blobStore, provider(blobSource(TEMPLATE)), timeoutManager, urlConnectionFactory, fetcherScope) val others = (1..3).map { refOf("low-$it".toByteArray()) } val boosted = refOf("boosted".toByteArray()) @@ -327,7 +339,7 @@ class RemoteConfigBlobFetcherTest { assertThat(download(fetcher, "not-a-valid-ref")).isFalse() - verify(exactly = 0) { urlConnectionFactory.createConnection(any(), any()) } + verify(exactly = 0) { urlConnectionFactory.createConnection(any(), any(), any(), any()) } verify(exactly = 0) { blobStore.write(any(), any()) } } @@ -336,7 +348,7 @@ class RemoteConfigBlobFetcherTest { val started = CopyOnWriteArrayList() recordStartedConnections(started) val fetcherScope = CoroutineScope(StandardTestDispatcher(testScheduler)) - val fetcher = RemoteConfigBlobFetcher(blobStore, provider(blobSource(TEMPLATE)), urlConnectionFactory, fetcherScope) + val fetcher = RemoteConfigBlobFetcher(blobStore, provider(blobSource(TEMPLATE)), timeoutManager, urlConnectionFactory, fetcherScope) val valid = refOf("valid".toByteArray()) fetcher.prefetch(listOf("malformed", valid)) @@ -374,20 +386,236 @@ class RemoteConfigBlobFetcherTest { val ref = refOf(bytes) stubConnection(urlFor(ref), code = 200, body = bytes) val fetcherScope = CoroutineScope(StandardTestDispatcher(testScheduler)) - val fetcher = RemoteConfigBlobFetcher(blobStore, provider(blobSource(TEMPLATE)), urlConnectionFactory, fetcherScope) + val fetcher = RemoteConfigBlobFetcher(blobStore, provider(blobSource(TEMPLATE)), timeoutManager, urlConnectionFactory, fetcherScope) val result = async { fetcher.ensureDownloaded(listOf(ref, ref)) } advanceUntilIdle() assertThat(result.await()).isTrue() - verify(exactly = 1) { urlConnectionFactory.createConnection(urlFor(ref), any()) } + verify(exactly = 1) { urlConnectionFactory.createConnection(urlFor(ref), any(), any(), any()) } fetcherScope.cancel() } + // region per-source timeouts + + @Test + fun `a fresh blob source uses the no-fallback base timeout`() { + val bytes = "body".toByteArray() + val ref = refOf(bytes) + val connectTimeout = slot() + val readTimeout = slot() + every { + urlConnectionFactory.createConnection(urlFor(ref), capture(connectTimeout), capture(readTimeout), any()) + } returns connection(code = 200, body = bytes) + val fetcher = realFetcher(provider(blobSource(TEMPLATE))) + + assertThat(download(fetcher, ref)).isTrue() + + assertThat(connectTimeout.captured).isEqualTo(HTTPTimeoutManager.MAIN_SOURCE_NO_FALLBACK_TIMEOUT_MS.toInt()) + assertThat(readTimeout.captured).isEqualTo(HTTPTimeoutManager.MAIN_SOURCE_NO_FALLBACK_TIMEOUT_MS.toInt()) + } + + @Test + fun `a blob source that just timed out uses the reduced timeout on the next attempt`() { + val bytes = "body".toByteArray() + val ref = refOf(bytes) + val timeouts = CopyOnWriteArrayList() + val calls = AtomicInteger(0) + every { urlConnectionFactory.createConnection(urlFor(ref), any(), any(), any()) } answers { + timeouts.add(secondArg()) + if (calls.getAndIncrement() == 0) throw SocketTimeoutException("timed out") + connection(code = 200, body = bytes) + } + val fetcher = realFetcher(provider(blobSource(TEMPLATE))) + + // The first attempt times out and exhausts the single source; the on-demand retry re-arms and hits it again. + assertThat(download(fetcher, ref)).isFalse() + assertThat(download(fetcher, ref)).isTrue() + + assertThat(timeouts).containsExactly( + HTTPTimeoutManager.MAIN_SOURCE_NO_FALLBACK_TIMEOUT_MS.toInt(), + HTTPTimeoutManager.MAIN_SOURCE_NO_FALLBACK_REDUCED_TIMEOUT_MS.toInt(), + ) + } + + @Test + fun `a successful blob download clears the source's fail-fast memory`() { + val bytes = "body".toByteArray() + val ref = refOf(bytes) + val timeouts = CopyOnWriteArrayList() + val calls = AtomicInteger(0) + every { urlConnectionFactory.createConnection(urlFor(ref), any(), any(), any()) } answers { + timeouts.add(secondArg()) + if (calls.getAndIncrement() == 0) throw SocketTimeoutException("timed out") + connection(code = 200, body = bytes) + } + val fetcher = realFetcher(provider(blobSource(TEMPLATE))) + + assertThat(download(fetcher, ref)).isFalse() // times out, arming the reduced timeout + assertThat(download(fetcher, ref)).isTrue() // succeeds at the reduced timeout, clearing the memory + assertThat(download(fetcher, ref)).isTrue() // back to the base timeout + + assertThat(timeouts).containsExactly( + HTTPTimeoutManager.MAIN_SOURCE_NO_FALLBACK_TIMEOUT_MS.toInt(), + HTTPTimeoutManager.MAIN_SOURCE_NO_FALLBACK_REDUCED_TIMEOUT_MS.toInt(), + HTTPTimeoutManager.MAIN_SOURCE_NO_FALLBACK_TIMEOUT_MS.toInt(), + ) + } + + @Test + fun `blob source timeout memory is isolated per host`() { + val primary = "https://primary.example/$PLACEHOLDER" + val secondary = "https://secondary.example/$PLACEHOLDER" + val refA = refOf("a".toByteArray()) + val bytesB = "b".toByteArray() + val refB = refOf(bytesB) + every { + urlConnectionFactory.createConnection(primary.replace(PLACEHOLDER, refA), any(), any(), any()) + } throws SocketTimeoutException("timed out") + val timeoutB = slot() + every { + urlConnectionFactory.createConnection(secondary.replace(PLACEHOLDER, refB), capture(timeoutB), any(), any()) + } returns connection(code = 200, body = bytesB) + + // Both fetchers share the same timeout manager (the field), so any cross-host leak would show up here. + assertThat(download(realFetcher(provider(blobSource(primary))), refA)).isFalse() + assertThat(download(realFetcher(provider(blobSource(secondary))), refB)).isTrue() + + // The second host never timed out, so it still uses the base timeout. + assertThat(timeoutB.captured).isEqualTo(HTTPTimeoutManager.MAIN_SOURCE_NO_FALLBACK_TIMEOUT_MS.toInt()) + } + + @Test + fun `a blob source's reduced timeout expires after the reset interval`() { + val bytes = "body".toByteArray() + val ref = refOf(bytes) + val timeouts = CopyOnWriteArrayList() + val calls = AtomicInteger(0) + every { urlConnectionFactory.createConnection(urlFor(ref), any(), any(), any()) } answers { + timeouts.add(secondArg()) + if (calls.getAndIncrement() == 0) throw SocketTimeoutException("timed out") + connection(code = 200, body = bytes) + } + val fetcher = realFetcher(provider(blobSource(TEMPLATE))) + + assertThat(download(fetcher, ref)).isFalse() + // Let the 10-minute per-host memory expire before the next attempt. + dateProvider.advanceTime(HTTPTimeoutManager.TIMEOUT_RESET_INTERVAL_MS + 1) + assertThat(download(fetcher, ref)).isTrue() + + assertThat(timeouts).containsExactly( + HTTPTimeoutManager.MAIN_SOURCE_NO_FALLBACK_TIMEOUT_MS.toInt(), + HTTPTimeoutManager.MAIN_SOURCE_NO_FALLBACK_TIMEOUT_MS.toInt(), + ) + } + + @Test + fun `a timed-out blob source is reported unhealthy and fails over to the next source`() { + val bytes = "body".toByteArray() + val ref = refOf(bytes) + val primary = "https://primary.example/$PLACEHOLDER" + val secondary = "https://secondary.example/$PLACEHOLDER" + val provider = provider(blobSource(primary, priority = 1), blobSource(secondary, priority = 2)) + every { + urlConnectionFactory.createConnection(primary.replace(PLACEHOLDER, ref), any(), any(), any()) + } throws SocketTimeoutException("timed out") + stubConnection(secondary.replace(PLACEHOLDER, ref), code = 200, body = bytes) + val fetcher = realFetcher(provider) + + assertThat(download(fetcher, ref)).isTrue() + + verify { provider.reportUnhealthy(match { it.url == primary }) } + verify { blobStore.write(ref, any()) } + } + + @Test + fun `a non-timeout network failure does not arm the source's fail-fast memory`() { + val bytes = "body".toByteArray() + val ref = refOf(bytes) + val timeouts = CopyOnWriteArrayList() + val calls = AtomicInteger(0) + every { urlConnectionFactory.createConnection(urlFor(ref), any(), any(), any()) } answers { + timeouts.add(secondArg()) + // A plain IOException (e.g. connection refused) is not a timeout: SocketTimeoutException is an + // IOException subclass, so this guards the catch order and that only timeouts arm the reduced tier. + if (calls.getAndIncrement() == 0) throw IOException("connection refused") + connection(code = 200, body = bytes) + } + val fetcher = realFetcher(provider(blobSource(TEMPLATE))) + + assertThat(download(fetcher, ref)).isFalse() + assertThat(download(fetcher, ref)).isTrue() + + // The failure never armed the source, so both attempts used the base timeout. + assertThat(timeouts).containsExactly( + HTTPTimeoutManager.MAIN_SOURCE_NO_FALLBACK_TIMEOUT_MS.toInt(), + HTTPTimeoutManager.MAIN_SOURCE_NO_FALLBACK_TIMEOUT_MS.toInt(), + ) + } + + @Test + fun `a non-timeout failure does not clear an already-armed source`() { + val bytes = "body".toByteArray() + val ref = refOf(bytes) + val timeouts = CopyOnWriteArrayList() + val calls = AtomicInteger(0) + every { urlConnectionFactory.createConnection(urlFor(ref), any(), any(), any()) } answers { + timeouts.add(secondArg()) + when (calls.getAndIncrement()) { + 0 -> throw SocketTimeoutException("timed out") // arms the reduced tier + 1 -> connection(code = 500) // a non-timeout failure must not clear the armed memory + else -> connection(code = 200, body = bytes) + } + } + val fetcher = realFetcher(provider(blobSource(TEMPLATE))) + + assertThat(download(fetcher, ref)).isFalse() + assertThat(download(fetcher, ref)).isFalse() + assertThat(download(fetcher, ref)).isTrue() + + // The 500 left the fail-fast memory intact, so the source stayed on the reduced timeout. + assertThat(timeouts).containsExactly( + HTTPTimeoutManager.MAIN_SOURCE_NO_FALLBACK_TIMEOUT_MS.toInt(), + HTTPTimeoutManager.MAIN_SOURCE_NO_FALLBACK_REDUCED_TIMEOUT_MS.toInt(), + HTTPTimeoutManager.MAIN_SOURCE_NO_FALLBACK_REDUCED_TIMEOUT_MS.toInt(), + ) + } + + @Test + fun `a 200 with a checksum mismatch still clears the source's fail-fast memory`() { + val expected = "expected".toByteArray() + val ref = refOf(expected) + val timeouts = CopyOnWriteArrayList() + val calls = AtomicInteger(0) + every { urlConnectionFactory.createConnection(urlFor(ref), any(), any(), any()) } answers { + timeouts.add(secondArg()) + when (calls.getAndIncrement()) { + 0 -> throw SocketTimeoutException("timed out") // arms the reduced tier + 1 -> connection(code = 200, body = "tampered".toByteArray()) // 200: source is network-healthy + else -> connection(code = 200, body = expected) + } + } + val fetcher = realFetcher(provider(blobSource(TEMPLATE))) + + assertThat(download(fetcher, ref)).isFalse() // timeout arms the reduced tier + assertThat(download(fetcher, ref)).isFalse() // 200 answered but the payload fails verification + assertThat(download(fetcher, ref)).isTrue() + + // The 200 (even with a bad payload) proved the source healthy and cleared its memory, so the last + // attempt reverted to the base timeout. + assertThat(timeouts).containsExactly( + HTTPTimeoutManager.MAIN_SOURCE_NO_FALLBACK_TIMEOUT_MS.toInt(), + HTTPTimeoutManager.MAIN_SOURCE_NO_FALLBACK_REDUCED_TIMEOUT_MS.toInt(), + HTTPTimeoutManager.MAIN_SOURCE_NO_FALLBACK_TIMEOUT_MS.toInt(), + ) + } + + // endregion + // region helpers private fun realFetcher(provider: RemoteConfigSourceProvider) = - RemoteConfigBlobFetcher(blobStore, provider, urlConnectionFactory, scope) + RemoteConfigBlobFetcher(blobStore, provider, timeoutManager, urlConnectionFactory, scope) /** * A spy over a real provider (fed a `sources` topic with only blob entries) so we get real failover @@ -425,7 +653,7 @@ class RemoteConfigBlobFetcherTest { runBlocking { withTimeout(WAIT_MS) { fetcher.ensureDownloaded(refs) } } private fun stubConnection(url: String, code: Int, body: ByteArray = ByteArray(0)) { - every { urlConnectionFactory.createConnection(url, any()) } returns connection(code, body) + every { urlConnectionFactory.createConnection(url, any(), any(), any()) } returns connection(code, body) } private fun connection(code: Int, body: ByteArray = ByteArray(0)): UrlConnection { @@ -437,7 +665,7 @@ class RemoteConfigBlobFetcherTest { /** Records the requested ref of every connection (in start order) so tests can assert scheduling order. */ private fun recordStartedConnections(started: MutableList) { - every { urlConnectionFactory.createConnection(any(), any()) } answers { + every { urlConnectionFactory.createConnection(any(), any(), any(), any()) } answers { started.add(refFromUrl(firstArg())) connection(code = 200) } @@ -462,6 +690,17 @@ class RemoteConfigBlobFetcherTest { if (topic == RemoteConfigTopic.Sources) sources else null } + private class FakeDateProvider( + private val currentTime: AtomicLong = AtomicLong(System.currentTimeMillis()), + ) : DateProvider { + override val now: Date + get() = Date(currentTime.get()) + + fun advanceTime(millis: Long) { + currentTime.addAndGet(millis) + } + } + /** Deterministic randomizer for weighted source selection: always returns the first candidate. */ private class FakeRandom(vararg values: Int) : Random() { private val values: IntArray = if (values.isEmpty()) intArrayOf(0) else values diff --git a/purchases/src/test/java/com/revenuecat/purchases/common/workflows/WorkflowsConfigIntegrationTest.kt b/purchases/src/test/java/com/revenuecat/purchases/common/workflows/WorkflowsConfigIntegrationTest.kt index cadbd2a747..0410b89146 100644 --- a/purchases/src/test/java/com/revenuecat/purchases/common/workflows/WorkflowsConfigIntegrationTest.kt +++ b/purchases/src/test/java/com/revenuecat/purchases/common/workflows/WorkflowsConfigIntegrationTest.kt @@ -86,6 +86,7 @@ class WorkflowsConfigIntegrationTest { val fetcher = RemoteConfigBlobFetcher( blobStore = blobStore, sourceProvider = FakeBlobSourceProvider, + timeoutManager = mockk(relaxed = true), urlConnectionFactory = fakeUrlConnectionFactory(), scope = testScope, ) @@ -403,7 +404,12 @@ class WorkflowsConfigIntegrationTest { } private fun fakeUrlConnectionFactory() = object : UrlConnectionFactory { - override fun createConnection(url: String, requestMethod: String): UrlConnection { + override fun createConnection( + url: String, + connectTimeoutMillis: Int, + readTimeoutMillis: Int, + requestMethod: String, + ): UrlConnection { val ref = url.substringAfterLast("/") val bytes = downloads[ref] if (bytes != null) { diff --git a/purchases/src/test/java/com/revenuecat/purchases/utils/TestUrlConnectionFactory.kt b/purchases/src/test/java/com/revenuecat/purchases/utils/TestUrlConnectionFactory.kt index b5b2a07243..675f6f323d 100644 --- a/purchases/src/test/java/com/revenuecat/purchases/utils/TestUrlConnectionFactory.kt +++ b/purchases/src/test/java/com/revenuecat/purchases/utils/TestUrlConnectionFactory.kt @@ -26,7 +26,12 @@ internal class TestUrlConnectionFactory( val createdConnections: List get() = _createdConnections.toList() - override fun createConnection(url: String, requestMethod: String): UrlConnection { + override fun createConnection( + url: String, + connectTimeoutMillis: Int, + readTimeoutMillis: Int, + requestMethod: String, + ): UrlConnection { _createdConnections.add(url) return connectionProvider?.invoke(url) ?: mockedConnections[url]