diff --git a/purchases/src/androidTestDefaults/kotlin/com/revenuecat/purchases/integration/BaseIntegrationPurchasesTest.kt b/purchases/src/androidTestDefaults/kotlin/com/revenuecat/purchases/integration/BaseIntegrationPurchasesTest.kt index c2aa3e4ec9..4802c497e9 100644 --- a/purchases/src/androidTestDefaults/kotlin/com/revenuecat/purchases/integration/BaseIntegrationPurchasesTest.kt +++ b/purchases/src/androidTestDefaults/kotlin/com/revenuecat/purchases/integration/BaseIntegrationPurchasesTest.kt @@ -515,7 +515,7 @@ abstract class BaseIntegrationPurchasesTest : BasePurchasesIntegrationTest() { return "test-storefront" } }, - apiSourceProvider = null, + apiSourceFailover = null, localeProvider = DefaultLocaleProvider(), ) } diff --git a/purchases/src/main/kotlin/com/revenuecat/purchases/PurchasesFactory.kt b/purchases/src/main/kotlin/com/revenuecat/purchases/PurchasesFactory.kt index 3ed0a4894a..227fb9b898 100644 --- a/purchases/src/main/kotlin/com/revenuecat/purchases/PurchasesFactory.kt +++ b/purchases/src/main/kotlin/com/revenuecat/purchases/PurchasesFactory.kt @@ -29,7 +29,9 @@ import com.revenuecat.purchases.common.events.BackendStoredEvent 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.APISourceFailover import com.revenuecat.purchases.common.networking.ETagManager +import com.revenuecat.purchases.common.networking.SourceHealthChecker import com.revenuecat.purchases.common.offerings.OfferingsCache import com.revenuecat.purchases.common.offerings.OfferingsFactory import com.revenuecat.purchases.common.offerings.OfferingsManager @@ -211,6 +213,7 @@ internal class PurchasesFactory( remoteConfigDiskCache?.read()?.topics?.get(it.wireName) } val apiSourceProvider = DefaultRemoteConfigSourceProvider(remoteConfigTopicStore) + val apiSourceFailover = APISourceFailover(appConfig, apiSourceProvider, SourceHealthChecker()) val httpClient = HTTPClient( appConfig, @@ -218,7 +221,7 @@ internal class PurchasesFactory( diagnosticsTracker, signingManager, cache, - apiSourceProvider, + apiSourceFailover, localeProvider = localeProvider, forceServerErrorStrategy = forceServerErrorStrategy, ) diff --git a/purchases/src/main/kotlin/com/revenuecat/purchases/common/HTTPClient.kt b/purchases/src/main/kotlin/com/revenuecat/purchases/common/HTTPClient.kt index 6534b99fab..088cf903be 100644 --- a/purchases/src/main/kotlin/com/revenuecat/purchases/common/HTTPClient.kt +++ b/purchases/src/main/kotlin/com/revenuecat/purchases/common/HTTPClient.kt @@ -13,6 +13,7 @@ import com.revenuecat.purchases.Store import com.revenuecat.purchases.VerificationResult import com.revenuecat.purchases.api.BuildConfig import com.revenuecat.purchases.common.diagnostics.DiagnosticsTracker +import com.revenuecat.purchases.common.networking.APISourceFailover import com.revenuecat.purchases.common.networking.ConnectionErrorReason import com.revenuecat.purchases.common.networking.ETagManager import com.revenuecat.purchases.common.networking.Endpoint @@ -24,7 +25,6 @@ import com.revenuecat.purchases.common.networking.NullPointerReadingErrorStreamE import com.revenuecat.purchases.common.networking.RCContainer import com.revenuecat.purchases.common.networking.RCContainerFormatException import com.revenuecat.purchases.common.networking.RCHTTPStatusCodes -import com.revenuecat.purchases.common.remoteconfig.RemoteConfigSourceProvider import com.revenuecat.purchases.common.verification.SignatureVerificationException import com.revenuecat.purchases.common.verification.SignatureVerificationMode import com.revenuecat.purchases.common.verification.SigningManager @@ -71,7 +71,7 @@ internal class HTTPClient( private val diagnosticsTrackerIfEnabled: DiagnosticsTracker?, val signingManager: SigningManager, private val storefrontProvider: StorefrontProvider, - private val apiSourceProvider: RemoteConfigSourceProvider?, + private val apiSourceFailover: APISourceFailover?, private val dateProvider: DateProvider = DefaultDateProvider(), private val mapConverter: MapConverter = MapConverter(), private val localeProvider: LocaleProvider, @@ -94,6 +94,10 @@ internal class HTTPClient( // Per-element codecs the SDK can decode, advertised so the server never picks one we can't // (e.g. brotli/zstd). "gzip" implies "identity" is also acceptable. const val RC_FORMAT_ACCEPT_ENCODING = "gzip" + + // Defensive cap on API source attempts within one request, in case the source list is re-armed + // (topic rebuild or interval restart) while a request is walking it. + const val MAX_API_SOURCE_ATTEMPTS = 5 } private val enableExtraRequestLogging = BuildConfig.ENABLE_EXTRA_REQUEST_LOGGING && appConfig.isDebugBuild @@ -156,7 +160,7 @@ internal class HTTPClient( * @throws JSONException Thrown for any JSON errors, not thrown for returned HTTP error codes * @throws IOException Thrown for any unexpected errors, not thrown for returned HTTP error codes */ - @Suppress("LongParameterList", "LongMethod", "CyclomaticComplexMethod", "InstanceOfCheckForException") + @Suppress("LongParameterList", "LongMethod", "CyclomaticComplexMethod") @Throws(JSONException::class, IOException::class) fun performRequest( baseURL: URL, @@ -193,8 +197,106 @@ internal class HTTPClient( val isMainBackend = fallbackURLIndex == 0 - val requestBaseURL = apiSourceURL(endpoint, baseURL, isFallbackAttempt = !isMainBackend) ?: baseURL + var source = apiSourceFailover?.currentSource(endpoint, baseURL, isFallbackAttempt = !isMainBackend) + var sourceAttempts = 0 + + while (true) { + sourceAttempts++ + val outcome = performAttempt( + requestBaseURL = source?.url ?: baseURL, + isFallbackURL = fallbackURLIndex > 0, + isMainBackend = isMainBackend, + fallbackAvailable = canUseFallback(), + endpoint = endpoint, + body = body, + postFieldsToSign = postFieldsToSign, + requestHeaders = requestHeaders, + refreshETag = refreshETag, + ) + if (outcome.canFailOverToNextSource) { + val nextSource = sourceToRetryOn(source, sourceAttempts, endpoint) + if (nextSource != null) { + source = nextSource + continue + } + } + return when (outcome) { + is AttemptOutcome.Failed -> { + if (!canUseFallback()) { + throw outcome.exception + } + // Unlike iOS, we keep failing over on every connection-level IOException here, including + // ones that may be caused by the device being offline. iOS suppresses the host switch on + // device connectivity errors, but Android has no equivalent signal at this layer: a device + // with no connectivity and a host whose DNS fails both surface as UnknownHostException, and + // telling them apart requires a ConnectivityManager check (ACCESS_NETWORK_STATE), which the + // SDK does not currently have. + var fallbackResult = performRequestToFallbackURL() + if (RCHTTPStatusCodes.isServerError(fallbackResult.responseCode) && canUseFallback()) { + fallbackResult = performRequestToFallbackURL() + } + fallbackResult + } + + is AttemptOutcome.Completed -> { + val result = outcome.result + when { + result == null -> { + log(LogIntent.WARNING) { NetworkStrings.ETAG_RETRYING_CALL } + performRequest( + baseURL, + endpoint, + body, + postFieldsToSign, + requestHeaders, + refreshETag = true, + fallbackBaseURLs, + fallbackURLIndex, + ) + } + + RCHTTPStatusCodes.isServerError(result.responseCode) && canUseFallback() -> + // Handle server errors with fallback URLs + performRequestToFallbackURL() + + else -> result + } + } + } + } + } + + /** The result of one request attempt against a single host. */ + private sealed interface AttemptOutcome { + /** [result] is null when an ETag cache miss requires retrying with a refreshed ETag. */ + data class Completed(val result: HTTPResult?) : AttemptOutcome + + data class Failed(val exception: IOException) : AttemptOutcome + /** Whether this outcome is a failure that API sources may recover from by switching hosts. */ + val canFailOverToNextSource: Boolean + get() = when (this) { + is Failed -> true + is Completed -> result?.let { RCHTTPStatusCodes.isServerError(it.responseCode) } == true + } + } + + /** + * Performs one request attempt against [requestBaseURL], recording its timeout bookkeeping and + * diagnostics. Connection-level failures come back as [AttemptOutcome.Failed] instead of throwing. + */ + @Suppress("LongParameterList", "InstanceOfCheckForException") + private fun performAttempt( + requestBaseURL: URL, + isFallbackURL: Boolean, + isMainBackend: Boolean, + fallbackAvailable: Boolean, + endpoint: Endpoint, + body: Map?, + postFieldsToSign: List>?, + requestHeaders: Map, + refreshETag: Boolean, + ): AttemptOutcome { var callSuccessful = false val requestStartTime = dateProvider.now var callResult: HTTPResult? = null @@ -204,7 +306,7 @@ internal class HTTPClient( try { callResult = performCall( requestBaseURL, - fallbackURLIndex > 0, + isFallbackURL, endpoint, body, postFieldsToSign, @@ -218,19 +320,9 @@ internal class HTTPClient( } } catch (e: IOException) { exceptionHit = e - if (e is SocketTimeoutException && isMainBackend && canUseFallback()) { - requestResult = HTTPTimeoutManager.RequestResult.TIMEOUT_ON_MAIN_BACKEND_FOR_FALLBACK_SUPPORTED_ENDPOINT - callResult = performRequestToFallbackURL() - } else if (canUseFallback()) { - // Unlike iOS, we keep failing over on every connection-level IOException here, including ones - // that may be caused by the device being offline. iOS suppresses the host switch on device - // connectivity errors, but Android has no equivalent signal at this layer: a device with no - // connectivity and a host whose DNS fails both surface as UnknownHostException, and telling - // them apart requires a ConnectivityManager check (ACCESS_NETWORK_STATE), which the SDK does - // not currently have. - callResult = performRequestToFallbackURL() - } else { - throw e + if (e is SocketTimeoutException && isMainBackend && fallbackAvailable) { + requestResult = + HTTPTimeoutManager.RequestResult.TIMEOUT_ON_MAIN_BACKEND_FOR_FALLBACK_SUPPORTED_ENDPOINT } } finally { timeoutManager.recordRequestResult(requestResult) @@ -245,42 +337,26 @@ internal class HTTPClient( connectionException = exceptionHit, ) } - if (callResult == null) { - log(LogIntent.WARNING) { NetworkStrings.ETAG_RETRYING_CALL } - callResult = performRequest( - baseURL, - endpoint, - body, - postFieldsToSign, - requestHeaders, - refreshETag = true, - fallbackBaseURLs, - fallbackURLIndex, - ) - } else if (RCHTTPStatusCodes.isServerError(callResult.responseCode) && canUseFallback()) { - // Handle server errors with fallback URLs - callResult = performRequestToFallbackURL() - } - return callResult + return exceptionHit?.let { AttemptOutcome.Failed(it) } ?: AttemptOutcome.Completed(callResult) } /** - * The API base URL to use for [endpoint], or null to keep using [baseURL]. - * - * API sources apply only when the `usesRemoteConfigAPISources` dangerous setting is enabled, this is not - * an endpoint fallback-host attempt, the endpoint opts in via [Endpoint.usesAPISources], and [baseURL] is - * still the default host (a proxy or an overridden base URL pins the host and bypasses API sources). + * The next source to retry [endpoint]'s failed request on, or null when the request didn't target an + * API [source], the attempt cap is reached, or [APISourceFailover] decides against failing over (the + * source's health check passed, or every source is already unhealthy). */ - private fun apiSourceURL(endpoint: Endpoint, baseURL: URL, isFallbackAttempt: Boolean): URL? { - val provider = apiSourceProvider ?: return null - val eligible = appConfig.usesRemoteConfigAPISources && - !isFallbackAttempt && - endpoint.usesAPISources && - baseURL.toString() == AppConfig.baseUrlString - return if (eligible) { - provider.currentAPISource()?.url?.let { runCatching { URL(it) }.getOrNull() } - } else { - null + private fun sourceToRetryOn( + source: APISourceFailover.ResolvedSource?, + sourceAttempts: Int, + endpoint: Endpoint, + ): APISourceFailover.ResolvedSource? { + val decision = source + ?.takeIf { sourceAttempts < MAX_API_SOURCE_ATTEMPTS } + ?.let { apiSourceFailover?.onRequestFailure(it) } + return (decision as? APISourceFailover.FailureDecision.RetryNextSource)?.next?.also { + log(LogIntent.DEBUG) { + NetworkStrings.RETRYING_CALL_WITH_NEXT_API_SOURCE.format(endpoint.name, it.url) + } } } diff --git a/purchases/src/main/kotlin/com/revenuecat/purchases/common/networking/APISourceFailover.kt b/purchases/src/main/kotlin/com/revenuecat/purchases/common/networking/APISourceFailover.kt new file mode 100644 index 0000000000..fefd39f9b4 --- /dev/null +++ b/purchases/src/main/kotlin/com/revenuecat/purchases/common/networking/APISourceFailover.kt @@ -0,0 +1,89 @@ +package com.revenuecat.purchases.common.networking + +import com.revenuecat.purchases.common.AppConfig +import com.revenuecat.purchases.common.LogIntent +import com.revenuecat.purchases.common.log +import com.revenuecat.purchases.common.remoteconfig.RemoteConfigSourceHandle +import com.revenuecat.purchases.common.remoteconfig.RemoteConfigSourceProvider +import com.revenuecat.purchases.common.warnLog +import java.net.MalformedURLException +import java.net.URL + +/** + * Decides which API source a request should target and whether a failed request should fail over to + * the next source. Only connection-level failures and 5xx responses can trigger a failover, and only + * after a health check against the current source fails: a 2xx health response means the source is + * healthy (the request failure was something else, e.g. endpoint-specific), so the original error + * surfaces without switching hosts. A source whose health check returns non-2xx or cannot complete + * is reported unhealthy and the next one takes over. 4xx responses never fail over: they are request + * errors, not source problems. + */ +internal class APISourceFailover( + private val appConfig: AppConfig, + private val sourceProvider: RemoteConfigSourceProvider, + private val healthChecker: SourceHealthChecker, +) { + + /** A source handle plus its parsed base URL; only parseable sources are ever handed out. */ + data class ResolvedSource(val handle: RemoteConfigSourceHandle, val url: URL) + + sealed interface FailureDecision { + /** The current source failed its health check; retry the request on [next]. */ + data class RetryNextSource(val next: ResolvedSource) : FailureDecision + + /** The current source is healthy, so failing over is pointless: surface the original error. */ + object SourceHealthy : FailureDecision + + /** Every source has been reported unhealthy; surface the original error. */ + object SourcesExhausted : FailureDecision + } + + /** + * The source a request to [endpoint] should target, or null to keep using [baseURL]. Sources apply + * only when the `usesRemoteConfigAPISources` dangerous setting is enabled, this is not an endpoint + * fallback-host attempt, the endpoint opts in via [Endpoint.usesAPISources], and [baseURL] is still + * the default host (a proxy or an overridden base URL pins the host and bypasses API sources). + */ + fun currentSource(endpoint: Endpoint, baseURL: URL, isFallbackAttempt: Boolean): ResolvedSource? { + val eligible = appConfig.usesRemoteConfigAPISources && + !isFallbackAttempt && + endpoint.usesAPISources && + baseURL.toString() == AppConfig.baseUrlString + return if (eligible) currentResolvedSource() else null + } + + /** + * Decides what to do after a request to [source] failed at the connection level or returned a + * 5xx, health-checking the source to tell a source outage apart from anything else. The health + * check performs blocking network I/O on the calling thread, so this must always be called from + * a background thread (as [com.revenuecat.purchases.common.HTTPClient] requests already are). + */ + fun onRequestFailure(source: ResolvedSource): FailureDecision { + if (healthChecker.isHealthy(source.handle.url)) { + log(LogIntent.DEBUG) { + "API source ${source.handle.url} is healthy despite the request failing; not failing over." + } + return FailureDecision.SourceHealthy + } + sourceProvider.reportUnhealthy(source.handle) + return currentResolvedSource()?.let { FailureDecision.RetryNextSource(it) } + ?: FailureDecision.SourcesExhausted + } + + /** + * The provider's current source with its URL parsed. Sources with malformed URLs are reported + * unhealthy and skipped, so they participate in failover instead of silently pinning requests to + * the default host. The walk ends when the provider runs out of sources. + */ + private fun currentResolvedSource(): ResolvedSource? { + while (true) { + val handle = sourceProvider.currentAPISource() ?: return null + try { + return ResolvedSource(handle, URL(handle.url)) + } catch (e: MalformedURLException) { + warnLog { "Skipping API source with malformed url ${handle.url}: $e" } + sourceProvider.reportUnhealthy(handle) + } + } + } +} diff --git a/purchases/src/main/kotlin/com/revenuecat/purchases/common/networking/SourceHealthChecker.kt b/purchases/src/main/kotlin/com/revenuecat/purchases/common/networking/SourceHealthChecker.kt new file mode 100644 index 0000000000..78753df702 --- /dev/null +++ b/purchases/src/main/kotlin/com/revenuecat/purchases/common/networking/SourceHealthChecker.kt @@ -0,0 +1,100 @@ +package com.revenuecat.purchases.common.networking + +import com.revenuecat.purchases.InternalRevenueCatAPI +import com.revenuecat.purchases.common.DateProvider +import com.revenuecat.purchases.common.DefaultDateProvider +import com.revenuecat.purchases.common.verboseLog +import com.revenuecat.purchases.utils.DefaultUrlConnectionFactory +import com.revenuecat.purchases.utils.UrlConnection +import com.revenuecat.purchases.utils.UrlConnectionFactory +import java.io.IOException +import java.util.concurrent.CountDownLatch + +/** + * Checks whether a source is healthy by hitting its `/v1/health/connectivity` endpoint: only a 2xx + * response counts as healthy; any other response or a connection failure does not. Checks are + * on-demand only (callers probe after a request to the source has already failed, never proactively). + * + * [isHealthy] performs blocking network I/O on the calling thread, so it must always be called from + * a background thread. Thread-safe: concurrent checks for the same source share a single request + * (joiners block until the owner's result arrives), and results are cached briefly so a burst of + * failing requests fires one probe. + */ +@OptIn(InternalRevenueCatAPI::class) +internal class SourceHealthChecker( + private val urlConnectionFactory: UrlConnectionFactory = DefaultUrlConnectionFactory(), + private val dateProvider: DateProvider = DefaultDateProvider(), +) { + + private class PendingCheck { + val latch = CountDownLatch(1) + + @Volatile + var isHealthy = false + } + + private data class CachedResult(val timestampMs: Long, val isHealthy: Boolean) + + private val lock = Any() + private val pendingChecks = mutableMapOf() + private val cachedResults = mutableMapOf() + + fun isHealthy(sourceBaseUrl: String): Boolean { + val healthUrl = healthUrl(sourceBaseUrl) + var isOwner = false + val check = synchronized(lock) { + validCachedResult(healthUrl)?.let { return it } + pendingChecks.getOrPut(healthUrl) { + isOwner = true + PendingCheck() + } + } + return if (isOwner) { + performOwnedCheck(healthUrl, check) + } else { + check.latch.await() + check.isHealthy + } + } + + private fun performOwnedCheck(healthUrl: String, check: PendingCheck): Boolean { + try { + check.isHealthy = performCheck(healthUrl) + } finally { + synchronized(lock) { + cachedResults[healthUrl] = CachedResult(dateProvider.now.time, check.isHealthy) + pendingChecks -= healthUrl + } + check.latch.countDown() + } + return check.isHealthy + } + + private fun validCachedResult(healthUrl: String): Boolean? = cachedResults[healthUrl]?.let { cached -> + cached.isHealthy.takeIf { dateProvider.now.time - cached.timestampMs < RESULT_VALIDITY_MS } + } + + private fun performCheck(healthUrl: String): Boolean { + var connection: UrlConnection? = null + return try { + connection = urlConnectionFactory.createConnection(healthUrl) + val responseCode = connection.responseCode + val isHealthy = responseCode in HEALTHY_RESPONSE_CODES + verboseLog { "Health check for $healthUrl returned $responseCode (healthy=$isHealthy)" } + isHealthy + } catch (e: IOException) { + verboseLog { "Health check for $healthUrl failed to connect: $e" } + false + } finally { + connection?.disconnect() + } + } + + private fun healthUrl(sourceBaseUrl: String): String = "${sourceBaseUrl.trimEnd('/')}/$HEALTH_PATH" + + private companion object { + const val HEALTH_PATH = "v1/health/connectivity" + const val RESULT_VALIDITY_MS = 10_000L + val HEALTHY_RESPONSE_CODES = 200..299 + } +} diff --git a/purchases/src/main/kotlin/com/revenuecat/purchases/common/remoteconfig/RemoteConfigSourceProvider.kt b/purchases/src/main/kotlin/com/revenuecat/purchases/common/remoteconfig/RemoteConfigSourceProvider.kt index 25131da4a6..bcc6e1e059 100644 --- a/purchases/src/main/kotlin/com/revenuecat/purchases/common/remoteconfig/RemoteConfigSourceProvider.kt +++ b/purchases/src/main/kotlin/com/revenuecat/purchases/common/remoteconfig/RemoteConfigSourceProvider.kt @@ -1,5 +1,8 @@ package com.revenuecat.purchases.common.remoteconfig +import com.revenuecat.purchases.InternalRevenueCatAPI +import com.revenuecat.purchases.common.DateProvider +import com.revenuecat.purchases.common.DefaultDateProvider import com.revenuecat.purchases.common.warnLog import kotlinx.serialization.json.JsonArray import kotlinx.serialization.json.JsonObject @@ -78,11 +81,19 @@ internal interface RemoteConfigSourceProvider { * sources have no embedded default: they are only useful alongside a fetched config, which carries * its own. Sources are deduped by url and ordered via [WeightedSourceSelector]. * + * The api list additionally restarts on its own once [API_FAILOVER_RESTART_INTERVAL_MS] has elapsed + * since it first failed over, so the SDK periodically returns to the primary source (or re-arms an + * exhausted list) instead of sticking on a backup forever. During a persistent outage this retries + * the primary at most once per interval. Blob failover has no such timer: it is restarted per fetch + * cycle by its callers. + * * Thread-safe. */ +@OptIn(InternalRevenueCatAPI::class) internal class DefaultRemoteConfigSourceProvider( private val topicStore: RemoteConfigTopicStore, private val random: Random = Random.Default, + private val dateProvider: DateProvider = DefaultDateProvider(), ) : RemoteConfigSourceProvider { private val lock = Any() @@ -90,6 +101,11 @@ internal class DefaultRemoteConfigSourceProvider( // Content hash of the `sources` topic the current failovers were built from. Null means there is no // sources topic (absent, or none seen yet), in which case the failovers hold the embedded defaults. private var builtHash: String? = null + + // When the api failover first advanced past the top of its list, or null while it sits at the top. + // Stamped on the first advance (not the latest) so a flapping source can't postpone the periodic + // return to the primary indefinitely. + private var apiFailoverStartedAtMs: Long? = null private var api = SourceFailover( RemoteConfigSourceHandle.Purpose.API, dedupe(sourcesFor(null, RemoteConfigSourceHandle.Purpose.API)), @@ -104,14 +120,21 @@ internal class DefaultRemoteConfigSourceProvider( override fun getCurrent(purpose: RemoteConfigSourceHandle.Purpose): RemoteConfigSourceHandle? = synchronized(lock) { rebuildIfChanged() + restartAPIIfExpired() failoverFor(purpose).current } override fun reportUnhealthy(handle: RemoteConfigSourceHandle) { synchronized(lock) { // Rebuild happened, no need to report unhealthy - if (!rebuildIfChanged()) { - failoverFor(handle.purpose).reportUnhealthy(handle) + if (rebuildIfChanged()) return + restartAPIIfExpired() + val advanced = failoverFor(handle.purpose).reportUnhealthy(handle) + if (advanced && + handle.purpose == RemoteConfigSourceHandle.Purpose.API && + apiFailoverStartedAtMs == null + ) { + apiFailoverStartedAtMs = dateProvider.now.time } } } @@ -121,6 +144,9 @@ internal class DefaultRemoteConfigSourceProvider( // Rebuild happened, no need to restart if (!rebuildIfChanged()) { failoverFor(purpose).restart() + if (purpose == RemoteConfigSourceHandle.Purpose.API) { + apiFailoverStartedAtMs = null + } } } } @@ -129,9 +155,13 @@ internal class DefaultRemoteConfigSourceProvider( synchronized(lock) { // A config change already re-armed both lists from the top; nothing more to do here. if (rebuildIfChanged()) return true + restartAPIIfExpired() val failover = failoverFor(purpose) if (failover.current == null) { failover.restart() + if (purpose == RemoteConfigSourceHandle.Purpose.API) { + apiFailoverStartedAtMs = null + } true } else { false @@ -150,6 +180,20 @@ internal class DefaultRemoteConfigSourceProvider( RemoteConfigSourceHandle.Purpose.BLOB -> blob } + /** + * Restarts the api failover once [API_FAILOVER_RESTART_INTERVAL_MS] has elapsed since it first + * advanced, so a backup (or an exhausted list) periodically retries the primary source. The + * restart bumps the token, so handles from before it can no longer advance the list. Callers + * must hold [lock]. + */ + private fun restartAPIIfExpired() { + val startedAtMs = apiFailoverStartedAtMs ?: return + if (dateProvider.now.time - startedAtMs >= API_FAILOVER_RESTART_INTERVAL_MS) { + api.restart() + apiFailoverStartedAtMs = null + } + } + /** * Rebuilds both failovers from the latest `sources` topic when its hash changed, returning whether * a rebuild happened. Callers must hold [lock]. @@ -179,9 +223,11 @@ internal class DefaultRemoteConfigSourceProvider( nextToken, ) builtHash = topic?.contentHash + apiFailoverStartedAtMs = null } private companion object { + private const val API_FAILOVER_RESTART_INTERVAL_MS = 600_000L private const val API_ITEM = "api" private const val BLOB_ITEM = "blob" private const val SOURCES_KEY = "sources" @@ -288,10 +334,12 @@ private class SourceFailover( val current: RemoteConfigSourceHandle? get() = selector.current?.let { RemoteConfigSourceHandle(purpose, it, token) } - fun reportUnhealthy(handle: RemoteConfigSourceHandle) { - if (handle.token != token) return + /** Advances past the handle's source, returning whether it did (stale reports are ignored). */ + fun reportUnhealthy(handle: RemoteConfigSourceHandle): Boolean { + if (handle.token != token) return false selector.advance() token++ + return true } fun restart() { diff --git a/purchases/src/main/kotlin/com/revenuecat/purchases/strings/NetworkStrings.kt b/purchases/src/main/kotlin/com/revenuecat/purchases/strings/NetworkStrings.kt index 337188b7df..8350bb3aef 100644 --- a/purchases/src/main/kotlin/com/revenuecat/purchases/strings/NetworkStrings.kt +++ b/purchases/src/main/kotlin/com/revenuecat/purchases/strings/NetworkStrings.kt @@ -7,6 +7,7 @@ internal object NetworkStrings { const val ETAG_RETRYING_CALL = "We were expecting to be able to return a cached response, but we can't find it. " + "Retrying call with a new ETag" const val RETRYING_CALL_WITH_FALLBACK_URL = "Retrying request %s using fallback URL %s" + const val RETRYING_CALL_WITH_NEXT_API_SOURCE = "Retrying request %s using next API source host %s" const val ETAG_CALL_ALREADY_RETRIED = "We can't find the cached response, but call has already been retried. " + "Returning result from backend: %s" const val SAME_CALL_SCHEDULED_WITHOUT_JITTER = "Request already scheduled without jitter delay, adding " + diff --git a/purchases/src/test/java/com/revenuecat/purchases/backend_integration_tests/BaseBackendIntegrationTest.kt b/purchases/src/test/java/com/revenuecat/purchases/backend_integration_tests/BaseBackendIntegrationTest.kt index a8e7f75db6..4a3bcb372d 100644 --- a/purchases/src/test/java/com/revenuecat/purchases/backend_integration_tests/BaseBackendIntegrationTest.kt +++ b/purchases/src/test/java/com/revenuecat/purchases/backend_integration_tests/BaseBackendIntegrationTest.kt @@ -136,7 +136,7 @@ internal abstract class BaseBackendIntegrationTest { diagnosticsTrackerIfEnabled = null, signingManager, deviceCache, - apiSourceProvider = null, + apiSourceFailover = null, localeProvider = DefaultLocaleProvider(), forceServerErrorStrategy = forceServerErrorStrategy, requestResponseListener = goldenFileRecorder diff --git a/purchases/src/test/java/com/revenuecat/purchases/common/BaseHTTPClientTest.kt b/purchases/src/test/java/com/revenuecat/purchases/common/BaseHTTPClientTest.kt index 74bf6778b6..a2119ebd4d 100644 --- a/purchases/src/test/java/com/revenuecat/purchases/common/BaseHTTPClientTest.kt +++ b/purchases/src/test/java/com/revenuecat/purchases/common/BaseHTTPClientTest.kt @@ -9,10 +9,12 @@ import com.revenuecat.purchases.PurchasesAreCompletedBy.REVENUECAT import com.revenuecat.purchases.Store import com.revenuecat.purchases.VerificationResult import com.revenuecat.purchases.common.diagnostics.DiagnosticsTracker +import com.revenuecat.purchases.common.networking.APISourceFailover import com.revenuecat.purchases.common.networking.ETagManager import com.revenuecat.purchases.common.networking.HTTPRequest import com.revenuecat.purchases.common.networking.HTTPResult import com.revenuecat.purchases.common.networking.HTTPTimeoutManager +import com.revenuecat.purchases.common.networking.SourceHealthChecker import com.revenuecat.purchases.common.remoteconfig.RemoteConfigSourceProvider import com.revenuecat.purchases.common.verification.SigningManager import com.revenuecat.purchases.interfaces.StorefrontProvider @@ -72,6 +74,7 @@ internal abstract class BaseHTTPClientTest { signingManager: SigningManager? = null, storefrontProvider: StorefrontProvider = mockStorefrontProvider, apiSourceProvider: RemoteConfigSourceProvider? = null, + sourceHealthChecker: SourceHealthChecker = SourceHealthChecker(), localeProvider: LocaleProvider = DefaultLocaleProvider(), forceServerErrorStrategy: ForceServerErrorStrategy? = null, timeoutManager: HTTPTimeoutManager? = null, @@ -81,7 +84,7 @@ internal abstract class BaseHTTPClientTest { diagnosticsTracker, signingManager ?: mockSigningManager, storefrontProvider, - apiSourceProvider, + apiSourceProvider?.let { APISourceFailover(appConfig, it, sourceHealthChecker) }, dateProvider, localeProvider = localeProvider, forceServerErrorStrategy = forceServerErrorStrategy, diff --git a/purchases/src/test/java/com/revenuecat/purchases/common/HTTPClientTest.kt b/purchases/src/test/java/com/revenuecat/purchases/common/HTTPClientTest.kt index ce19177130..4945a6cfcf 100644 --- a/purchases/src/test/java/com/revenuecat/purchases/common/HTTPClientTest.kt +++ b/purchases/src/test/java/com/revenuecat/purchases/common/HTTPClientTest.kt @@ -16,10 +16,13 @@ import com.revenuecat.purchases.common.networking.HTTPRequest import com.revenuecat.purchases.common.networking.HTTPResult import com.revenuecat.purchases.common.networking.HTTPTimeoutManager import com.revenuecat.purchases.common.networking.RCHTTPStatusCodes +import com.revenuecat.purchases.common.networking.SourceHealthChecker import com.revenuecat.purchases.common.remoteconfig.RemoteConfigSource import com.revenuecat.purchases.common.remoteconfig.RemoteConfigSourceHandle import com.revenuecat.purchases.common.remoteconfig.RemoteConfigSourceProvider import com.revenuecat.purchases.utils.Responses +import com.revenuecat.purchases.utils.TestUrlConnection +import com.revenuecat.purchases.utils.TestUrlConnectionFactory import io.mockk.Runs import io.mockk.every import io.mockk.just @@ -36,6 +39,7 @@ import org.junit.Before import org.junit.Test import org.junit.runner.RunWith import org.robolectric.ParameterizedRobolectricTestRunner +import java.io.ByteArrayInputStream import java.io.IOException import java.net.SocketTimeoutException import java.net.URL @@ -238,26 +242,361 @@ internal class HTTPClientTest: BaseHTTPClientTest() { assertThat(result.responseCode).isEqualTo(RCHTTPStatusCodes.SUCCESS) } - /** A minimal [RemoteConfigSourceProvider] whose current API source is the first of [urls]. */ + /** A minimal [RemoteConfigSourceProvider] that walks [urls] in order as they are reported unhealthy. */ private class FakeAPISourceProvider(urls: List) : RemoteConfigSourceProvider { - private val handles = urls.mapIndexed { index, url -> - RemoteConfigSourceHandle( - purpose = RemoteConfigSourceHandle.Purpose.API, - source = RemoteConfigSource(url = url, priority = index, weight = 1), - token = index, - ) + private val sources = urls.mapIndexed { index, url -> + RemoteConfigSource(url = url, priority = index, weight = 1) } + private var index = 0 + private var token = 0 + val unhealthyReports = mutableListOf() override fun getCurrent(purpose: RemoteConfigSourceHandle.Purpose): RemoteConfigSourceHandle? = - handles.firstOrNull() + sources.getOrNull(index)?.let { RemoteConfigSourceHandle(purpose, it, token) } + + override fun reportUnhealthy(handle: RemoteConfigSourceHandle) { + unhealthyReports.add(handle.url) + if (handle.token == token) { + index++ + token++ + } + } + + override fun restart(purpose: RemoteConfigSourceHandle.Purpose) { + index = 0 + token++ + } + + override fun restartIfExhausted(purpose: RemoteConfigSourceHandle.Purpose): Boolean = + if (getCurrent(purpose) == null) { + restart(purpose) + true + } else { + false + } + } + + // endregion + + // region API source failover + + private fun healthCheckerReturning(responseCode: Int, factory: TestUrlConnectionFactory? = null) = + SourceHealthChecker( + factory ?: TestUrlConnectionFactory( + connectionProvider = { TestUrlConnection(responseCode, ByteArrayInputStream(ByteArray(0))) }, + ), + ) + + private fun unreachableHealthChecker() = SourceHealthChecker( + TestUrlConnectionFactory(connectionProvider = { throw IOException("health endpoint unreachable") }), + ) + + /** A server whose port refuses connections, to simulate an unreachable source. */ + private fun unreachableSourceUrl(): String { + val downServer = MockWebServer() + val url = downServer.url("/").toString() + downServer.shutdown() + return url + } + + @Test + fun `performRequest does not fail over on 5xx when the source's health check passes`() { + val endpoint = Endpoint.GetCustomerInfo("test_user_id") + val secondSourceServer = MockWebServer() + val provider = FakeAPISourceProvider( + listOf(server.url("/").toString(), secondSourceServer.url("/").toString()), + ) + val healthFactory = TestUrlConnectionFactory( + connectionProvider = { TestUrlConnection(200, ByteArrayInputStream(ByteArray(0))) }, + ) + val client = createClient( + appConfig = createAppConfig(proxyURL = null, usesRemoteConfigAPISources = true), + apiSourceProvider = provider, + sourceHealthChecker = healthCheckerReturning(200, healthFactory), + ) + enqueue(endpoint.getPath(), expectedResult = HTTPResult.createResult(RCHTTPStatusCodes.ERROR)) + + val result = client.performRequest( + URL(AppConfig.baseUrlString), + endpoint, + body = null, + postFieldsToSign = null, + mapOf("" to ""), + ) + + assertThat(result.responseCode).isEqualTo(RCHTTPStatusCodes.ERROR) + assertThat(server.requestCount).isEqualTo(1) + assertThat(secondSourceServer.requestCount).isEqualTo(0) + assertThat(healthFactory.createdConnections) + .containsExactly(server.url("/v1/health/connectivity").toString()) + assertThat(provider.unhealthyReports).isEmpty() + secondSourceServer.shutdown() + } + + @Test + fun `performRequest fails over on 5xx when the source's health check fails`() { + val endpoint = Endpoint.GetCustomerInfo("test_user_id") + val secondSourceServer = MockWebServer() + val provider = FakeAPISourceProvider( + listOf(server.url("/").toString(), secondSourceServer.url("/").toString()), + ) + val client = createClient( + appConfig = createAppConfig(proxyURL = null, usesRemoteConfigAPISources = true), + apiSourceProvider = provider, + sourceHealthChecker = unreachableHealthChecker(), + ) + enqueue(endpoint.getPath(), expectedResult = HTTPResult.createResult(RCHTTPStatusCodes.ERROR)) + enqueue(endpoint.getPath(), expectedResult = HTTPResult.createResult(), server = secondSourceServer) + + val result = client.performRequest( + URL(AppConfig.baseUrlString), + endpoint, + body = null, + postFieldsToSign = null, + mapOf("" to ""), + ) + + assertThat(result.responseCode).isEqualTo(RCHTTPStatusCodes.SUCCESS) + assertThat(server.requestCount).isEqualTo(1) + assertThat(secondSourceServer.requestCount).isEqualTo(1) + assertThat(secondSourceServer.takeRequest().path).isEqualTo("/v1/subscribers/test_user_id") + assertThat(provider.unhealthyReports).containsExactly(server.url("/").toString()) + secondSourceServer.shutdown() + } + + @Test + fun `performRequest does not fail over on 4xx and never health checks`() { + val endpoint = Endpoint.GetCustomerInfo("test_user_id") + val secondSourceServer = MockWebServer() + val provider = FakeAPISourceProvider( + listOf(server.url("/").toString(), secondSourceServer.url("/").toString()), + ) + val healthFactory = TestUrlConnectionFactory( + connectionProvider = { TestUrlConnection(200, ByteArrayInputStream(ByteArray(0))) }, + ) + val client = createClient( + appConfig = createAppConfig(proxyURL = null, usesRemoteConfigAPISources = true), + apiSourceProvider = provider, + sourceHealthChecker = healthCheckerReturning(200, healthFactory), + ) + enqueue(endpoint.getPath(), expectedResult = HTTPResult.createResult(RCHTTPStatusCodes.NOT_FOUND)) + + val result = client.performRequest( + URL(AppConfig.baseUrlString), + endpoint, + body = null, + postFieldsToSign = null, + mapOf("" to ""), + ) + + assertThat(result.responseCode).isEqualTo(RCHTTPStatusCodes.NOT_FOUND) + assertThat(server.requestCount).isEqualTo(1) + assertThat(secondSourceServer.requestCount).isEqualTo(0) + assertThat(healthFactory.createdConnections).isEmpty() + assertThat(provider.unhealthyReports).isEmpty() + secondSourceServer.shutdown() + } + + @Test + fun `performRequest does not fail over on connection failure when the source's health check passes`() { + val endpoint = Endpoint.GetCustomerInfo("test_user_id") + val unreachableUrl = unreachableSourceUrl() + val provider = FakeAPISourceProvider(listOf(unreachableUrl, server.url("/").toString())) + val client = createClient( + appConfig = createAppConfig(proxyURL = null, usesRemoteConfigAPISources = true), + apiSourceProvider = provider, + sourceHealthChecker = healthCheckerReturning(200), + ) + + assertThatThrownBy { + client.performRequest( + URL(AppConfig.baseUrlString), + endpoint, + body = null, + postFieldsToSign = null, + mapOf("" to ""), + ) + }.isInstanceOf(IOException::class.java) + + assertThat(server.requestCount).isEqualTo(0) + assertThat(provider.unhealthyReports).isEmpty() + } + + @Test + fun `performRequest fails over on connection failure when the source's health check fails`() { + val endpoint = Endpoint.GetCustomerInfo("test_user_id") + val unreachableUrl = unreachableSourceUrl() + val provider = FakeAPISourceProvider(listOf(unreachableUrl, server.url("/").toString())) + val client = createClient( + appConfig = createAppConfig(proxyURL = null, usesRemoteConfigAPISources = true), + apiSourceProvider = provider, + sourceHealthChecker = unreachableHealthChecker(), + ) + enqueue(endpoint.getPath(), expectedResult = HTTPResult.createResult()) + enqueue(endpoint.getPath(), expectedResult = HTTPResult.createResult()) - override fun reportUnhealthy(handle: RemoteConfigSourceHandle) = Unit + val result = client.performRequest( + URL(AppConfig.baseUrlString), + endpoint, + body = null, + postFieldsToSign = null, + mapOf("" to ""), + ) + + assertThat(result.responseCode).isEqualTo(RCHTTPStatusCodes.SUCCESS) + assertThat(server.requestCount).isEqualTo(1) + assertThat(provider.unhealthyReports).containsExactly(unreachableUrl) + + // The provider advanced, so a subsequent request starts directly on the healthy source. + val secondResult = client.performRequest( + URL(AppConfig.baseUrlString), + endpoint, + body = null, + postFieldsToSign = null, + mapOf("" to ""), + ) + assertThat(secondResult.responseCode).isEqualTo(RCHTTPStatusCodes.SUCCESS) + assertThat(server.requestCount).isEqualTo(2) + assertThat(provider.unhealthyReports).containsExactly(unreachableUrl) + } + + @Test + fun `performRequest surfaces the original error once connection failures exhaust the sources`() { + val endpoint = Endpoint.GetCustomerInfo("test_user_id") + val unreachableUrl = unreachableSourceUrl() + val provider = FakeAPISourceProvider(listOf(unreachableUrl)) + val client = createClient( + appConfig = createAppConfig(proxyURL = null, usesRemoteConfigAPISources = true), + apiSourceProvider = provider, + sourceHealthChecker = unreachableHealthChecker(), + ) + + assertThatThrownBy { + client.performRequest( + URL(AppConfig.baseUrlString), + endpoint, + body = null, + postFieldsToSign = null, + mapOf("" to ""), + ) + }.isInstanceOf(IOException::class.java) + + assertThat(provider.unhealthyReports).containsExactly(unreachableUrl) + } + + @Test + fun `performRequest still uses the endpoint fallback URL once the sources are exhausted`() { + val endpoint = Endpoint.GetOfferings("test_user_id") + assert(endpoint.supportsFallbackBaseURLs) + val fallbackServer = MockWebServer() + val provider = FakeAPISourceProvider(listOf(server.url("/").toString())) + val client = createClient( + appConfig = createAppConfig(proxyURL = null, usesRemoteConfigAPISources = true), + apiSourceProvider = provider, + sourceHealthChecker = unreachableHealthChecker(), + ) + enqueue(endpoint.getPath(), expectedResult = HTTPResult.createResult(RCHTTPStatusCodes.ERROR)) + enqueue( + endpoint.getPath(useFallback = true), + expectedResult = HTTPResult.createResult(), + server = fallbackServer, + isFallbackURL = true, + ) - override fun restart(purpose: RemoteConfigSourceHandle.Purpose) = Unit + val result = client.performRequest( + URL(AppConfig.baseUrlString), + endpoint, + body = null, + postFieldsToSign = null, + mapOf("" to ""), + fallbackBaseURLs = listOf(fallbackServer.url("/v1").toUrl()), + ) + + assertThat(result.responseCode).isEqualTo(RCHTTPStatusCodes.SUCCESS) + assertThat(server.requestCount).isEqualTo(1) + assertThat(fallbackServer.requestCount).isEqualTo(1) + assertThat(fallbackServer.takeRequest().path).isEqualTo("/v1/offerings") + fallbackServer.shutdown() + } + + @Test + fun `performRequest retries a POST with its body on the next source`() { + val endpoint = Endpoint.PostReceipt + val unreachableUrl = unreachableSourceUrl() + val provider = FakeAPISourceProvider(listOf(unreachableUrl, server.url("/").toString())) + val client = createClient( + appConfig = createAppConfig(proxyURL = null, usesRemoteConfigAPISources = true), + apiSourceProvider = provider, + sourceHealthChecker = unreachableHealthChecker(), + ) + enqueue(endpoint.getPath(), expectedResult = HTTPResult.createResult()) + + val result = client.performRequest( + URL(AppConfig.baseUrlString), + endpoint, + body = mapOf("fetch_token" to "token"), + postFieldsToSign = null, + mapOf("" to ""), + ) + + assertThat(result.responseCode).isEqualTo(RCHTTPStatusCodes.SUCCESS) + val request = server.takeRequest() + assertThat(request.method).isEqualTo("POST") + assertThat(request.path).isEqualTo("/v1/receipts") + assertThat(request.body.readUtf8()).contains("\"fetch_token\":\"token\"") + } + + /** + * A provider whose source list re-arms on every unhealthy report (same url, fresh token), like a + * topic rebuild or interval restart landing mid-walk, so a walk would never exhaust on its own. + */ + private class ReArmingAPISourceProvider(private val url: String) : RemoteConfigSourceProvider { + private var token = 0 + var unhealthyReportCount = 0 + private set + + override fun getCurrent(purpose: RemoteConfigSourceHandle.Purpose): RemoteConfigSourceHandle = + RemoteConfigSourceHandle(purpose, RemoteConfigSource(url = url, priority = 0, weight = 1), token) + + override fun reportUnhealthy(handle: RemoteConfigSourceHandle) { + unhealthyReportCount++ + token++ + } + + override fun restart(purpose: RemoteConfigSourceHandle.Purpose) { + token++ + } override fun restartIfExhausted(purpose: RemoteConfigSourceHandle.Purpose): Boolean = false } + @Test + fun `performRequest stops after MAX_API_SOURCE_ATTEMPTS when the source list re-arms mid-walk`() { + val endpoint = Endpoint.GetCustomerInfo("test_user_id") + val provider = ReArmingAPISourceProvider(server.url("/").toString()) + val client = createClient( + appConfig = createAppConfig(proxyURL = null, usesRemoteConfigAPISources = true), + apiSourceProvider = provider, + sourceHealthChecker = unreachableHealthChecker(), + ) + repeat(HTTPClient.MAX_API_SOURCE_ATTEMPTS) { + enqueue(endpoint.getPath(), expectedResult = HTTPResult.createResult(RCHTTPStatusCodes.ERROR)) + } + + val result = client.performRequest( + URL(AppConfig.baseUrlString), + endpoint, + body = null, + postFieldsToSign = null, + mapOf("" to ""), + ) + + assertThat(result.responseCode).isEqualTo(RCHTTPStatusCodes.ERROR) + assertThat(server.requestCount).isEqualTo(HTTPClient.MAX_API_SOURCE_ATTEMPTS) + // The last attempt hits the cap without consulting the failover, so it is never reported. + assertThat(provider.unhealthyReportCount).isEqualTo(HTTPClient.MAX_API_SOURCE_ATTEMPTS - 1) + } + // endregion @Test diff --git a/purchases/src/test/java/com/revenuecat/purchases/common/backend/BackendAPISourceFailoverIntegrationTest.kt b/purchases/src/test/java/com/revenuecat/purchases/common/backend/BackendAPISourceFailoverIntegrationTest.kt new file mode 100644 index 0000000000..d898e482e2 --- /dev/null +++ b/purchases/src/test/java/com/revenuecat/purchases/common/backend/BackendAPISourceFailoverIntegrationTest.kt @@ -0,0 +1,317 @@ +package com.revenuecat.purchases.common.backend + +import android.content.Context +import android.content.SharedPreferences +import androidx.test.ext.junit.runners.AndroidJUnit4 +import com.revenuecat.purchases.APIKeyValidator +import com.revenuecat.purchases.CustomerInfo +import com.revenuecat.purchases.DangerousSettings +import com.revenuecat.purchases.PurchasesAreCompletedBy +import com.revenuecat.purchases.PurchasesError +import com.revenuecat.purchases.PurchasesErrorCode +import com.revenuecat.purchases.Store +import com.revenuecat.purchases.common.AppConfig +import com.revenuecat.purchases.common.Backend +import com.revenuecat.purchases.common.BackendHelper +import com.revenuecat.purchases.common.DefaultLocaleProvider +import com.revenuecat.purchases.common.HTTPClient +import com.revenuecat.purchases.common.PlatformInfo +import com.revenuecat.purchases.common.SyncDispatcher +import com.revenuecat.purchases.common.getLocale +import com.revenuecat.purchases.common.networking.APISourceFailover +import com.revenuecat.purchases.common.networking.ETagManager +import com.revenuecat.purchases.common.networking.ETagPayloadStore +import com.revenuecat.purchases.common.networking.SourceHealthChecker +import com.revenuecat.purchases.common.remoteconfig.ConfigTopic +import com.revenuecat.purchases.common.remoteconfig.DefaultRemoteConfigSourceProvider +import com.revenuecat.purchases.common.remoteconfig.RemoteConfigSource +import com.revenuecat.purchases.common.remoteconfig.RemoteConfigTopic +import com.revenuecat.purchases.common.remoteconfig.RemoteConfiguration +import com.revenuecat.purchases.common.verification.SignatureVerificationMode +import com.revenuecat.purchases.common.verification.SigningManager +import com.revenuecat.purchases.interfaces.StorefrontProvider +import com.revenuecat.purchases.utils.Responses +import io.mockk.Runs +import io.mockk.every +import io.mockk.just +import io.mockk.mockk +import kotlinx.serialization.json.addJsonObject +import kotlinx.serialization.json.buildJsonObject +import kotlinx.serialization.json.put +import kotlinx.serialization.json.putJsonArray +import okhttp3.mockwebserver.Dispatcher +import okhttp3.mockwebserver.MockResponse +import okhttp3.mockwebserver.MockWebServer +import okhttp3.mockwebserver.RecordedRequest +import okhttp3.mockwebserver.SocketPolicy +import org.assertj.core.api.Assertions.assertThat +import org.junit.After +import org.junit.Before +import org.junit.Rule +import org.junit.Test +import org.junit.rules.TemporaryFolder +import org.junit.runner.RunWith +import org.robolectric.annotation.Config +import java.util.Locale +import java.util.concurrent.TimeUnit + +/** + * End-to-end coverage of API source failover through a real [Backend.getCustomerInfo] request: real + * [HTTPClient], [APISourceFailover], [DefaultRemoteConfigSourceProvider], [ETagManager] and + * [SourceHealthChecker] (real HTTP health checks), with [MockWebServer]s standing in for the sources. + */ +@RunWith(AndroidJUnit4::class) +@Config(manifest = Config.NONE) +internal class BackendAPISourceFailoverIntegrationTest { + + private sealed interface FetchResult { + data class Success(val customerInfo: CustomerInfo) : FetchResult + data class Error(val error: PurchasesError, val isServerError: Boolean) : FetchResult + } + + @get:Rule + val temporaryFolder = TemporaryFolder() + + private val apiKey = "test-api-key" + private val appUserID = "integration-test-user" + private val customerInfoPath = "/v1/subscribers/$appUserID" + private val healthPath = "/v1/health/connectivity" + + private lateinit var sourceAServer: MockWebServer + private lateinit var sourceBServer: MockWebServer + + @Before + fun setUp() { + sourceAServer = MockWebServer() + sourceBServer = MockWebServer() + } + + @After + fun tearDown() { + sourceAServer.shutdown() + sourceBServer.shutdown() + } + + @Test + fun `getCustomerInfo succeeds on the primary source without health checks`() { + sourceAServer.routeResponses(endpoint = customerInfoResponse(), health = response(200)) + val backend = createBackend(sourceAServer.url("/").toString(), sourceBServer.url("/").toString()) + + val result = fetchCustomerInfo(backend) + + assertThat(result).isInstanceOf(FetchResult.Success::class.java) + assertThat(sourceAServer.recordedPaths()).containsExactly(customerInfoPath) + assertThat(sourceBServer.requestCount).isEqualTo(0) + } + + @Test + fun `getCustomerInfo surfaces a 5xx without failing over when the source is healthy`() { + sourceAServer.routeResponses(endpoint = response(500), health = response(200)) + val backend = createBackend(sourceAServer.url("/").toString(), sourceBServer.url("/").toString()) + + val result = fetchCustomerInfo(backend) + + assertThat(result).isInstanceOf(FetchResult.Error::class.java) + assertThat((result as FetchResult.Error).isServerError).isTrue() + assertThat(sourceAServer.recordedPaths()).containsExactly(customerInfoPath, healthPath) + assertThat(sourceBServer.requestCount).isEqualTo(0) + } + + @Test + fun `getCustomerInfo fails over on a 5xx when the source is unhealthy, and stays failed over`() { + sourceAServer.routeResponses(endpoint = response(500), health = response(503)) + sourceBServer.routeResponses(endpoint = customerInfoResponse(), health = response(200)) + val backend = createBackend(sourceAServer.url("/").toString(), sourceBServer.url("/").toString()) + + val result = fetchCustomerInfo(backend) + + assertThat(result).isInstanceOf(FetchResult.Success::class.java) + assertThat(sourceAServer.recordedPaths()).containsExactly(customerInfoPath, healthPath) + assertThat(sourceBServer.recordedPaths()).containsExactly(customerInfoPath) + + // The provider advanced, so a subsequent request goes straight to the second source. + val secondResult = fetchCustomerInfo(backend) + + assertThat(secondResult).isInstanceOf(FetchResult.Success::class.java) + assertThat(sourceAServer.requestCount).isEqualTo(2) + assertThat(sourceBServer.requestCount).isEqualTo(2) + assertThat(sourceBServer.takeRequest(1, TimeUnit.SECONDS)?.path).isEqualTo(customerInfoPath) + } + + @Test + fun `getCustomerInfo fails over when the source is unreachable`() { + val unreachableUrl = unreachableSourceUrl() + sourceBServer.routeResponses(endpoint = customerInfoResponse(), health = response(200)) + val backend = createBackend(unreachableUrl, sourceBServer.url("/").toString()) + + val result = fetchCustomerInfo(backend) + + assertThat(result).isInstanceOf(FetchResult.Success::class.java) + assertThat(sourceBServer.recordedPaths()).containsExactly(customerInfoPath) + } + + @Test + fun `getCustomerInfo surfaces a connection error without failing over when the source is healthy`() { + // DISCONNECT_AFTER_REQUEST (not AT_START): the server must read the request for the path-routing + // dispatcher to run, and it still closes the socket without a response, so the client sees an + // IOException on the customer info request while the health endpoint keeps answering 200. + sourceAServer.routeResponses( + endpoint = response(200).apply { socketPolicy = SocketPolicy.DISCONNECT_AFTER_REQUEST }, + health = response(200), + ) + val backend = createBackend(sourceAServer.url("/").toString(), sourceBServer.url("/").toString()) + + val result = fetchCustomerInfo(backend) + + assertThat(result).isInstanceOf(FetchResult.Error::class.java) + assertThat((result as FetchResult.Error).error.code).isEqualTo(PurchasesErrorCode.NetworkError) + // The health check ran and passed, and we still never failed over to the second source. + assertThat(sourceAServer.recordedPaths()).contains(healthPath) + assertThat(sourceBServer.requestCount).isEqualTo(0) + } + + @Test + fun `getCustomerInfo surfaces the error once every source is unhealthy`() { + sourceAServer.routeResponses(endpoint = response(500), health = response(503)) + sourceBServer.routeResponses(endpoint = response(500), health = response(503)) + val backend = createBackend(sourceAServer.url("/").toString(), sourceBServer.url("/").toString()) + + val result = fetchCustomerInfo(backend) + + assertThat(result).isInstanceOf(FetchResult.Error::class.java) + assertThat((result as FetchResult.Error).isServerError).isTrue() + assertThat(sourceAServer.recordedPaths()).containsExactly(customerInfoPath, healthPath) + assertThat(sourceBServer.recordedPaths()).containsExactly(customerInfoPath, healthPath) + } + + @Test + fun `getCustomerInfo does not fail over or health check on a 4xx`() { + sourceAServer.routeResponses(endpoint = response(404), health = response(200)) + val backend = createBackend(sourceAServer.url("/").toString(), sourceBServer.url("/").toString()) + + val result = fetchCustomerInfo(backend) + + assertThat(result).isInstanceOf(FetchResult.Error::class.java) + assertThat((result as FetchResult.Error).isServerError).isFalse() + assertThat(sourceAServer.recordedPaths()).containsExactly(customerInfoPath) + assertThat(sourceBServer.requestCount).isEqualTo(0) + } + + // region Helpers + + private fun fetchCustomerInfo(backend: Backend): FetchResult { + var result: FetchResult? = null + backend.getCustomerInfo( + appUserID = appUserID, + appInBackground = false, + onSuccess = { result = FetchResult.Success(it) }, + onError = { error, isServerError -> result = FetchResult.Error(error, isServerError) }, + ) + return result ?: error("Expected the request to complete synchronously") + } + + /** Wires the full real stack: sources come from a `sources` topic pointing at [sourceUrls] in order. */ + private fun createBackend(vararg sourceUrls: String): Backend { + val appConfig = createAppConfig() + val topic = sourcesTopic(sourceUrls.toList()) + val sourceProvider = DefaultRemoteConfigSourceProvider( + { requestedTopic -> topic.takeIf { requestedTopic == RemoteConfigTopic.Sources } }, + ) + val apiSourceFailover = APISourceFailover(appConfig, sourceProvider, SourceHealthChecker()) + val sharedPreferencesEditor = mockk().apply { + every { putString(any(), any()) } returns this + every { remove(any()) } returns this + every { apply() } just Runs + } + val sharedPreferences = mockk().apply { + every { getString(any(), any()) } answers { secondArg() as String? } + every { edit() } returns sharedPreferencesEditor + } + val eTagManager = ETagManager( + mockk(), + lazy { sharedPreferences }, + payloadStore = ETagPayloadStore(temporaryFolder.newFolder()), + ) + val signingManager = SigningManager(SignatureVerificationMode.Disabled, appConfig, apiKey) + val storefrontProvider = mockk().apply { + every { getStorefront() } returns "JP" + } + val httpClient = HTTPClient( + appConfig, + eTagManager, + diagnosticsTrackerIfEnabled = null, + signingManager, + storefrontProvider, + apiSourceFailover, + localeProvider = DefaultLocaleProvider(), + ) + val dispatcher = SyncDispatcher() + val backendHelper = BackendHelper(apiKey, dispatcher, appConfig, httpClient) + return Backend(appConfig, dispatcher, dispatcher, httpClient, backendHelper) + } + + private fun createAppConfig(): AppConfig = AppConfig( + context = createContext(), + purchasesAreCompletedBy = PurchasesAreCompletedBy.REVENUECAT, + showInAppMessagesAutomatically = false, + platformInfo = PlatformInfo("native", "1.0"), + proxyURL = null, + store = Store.PLAY_STORE, + isDebugBuild = false, + apiKeyValidationResult = APIKeyValidator.ValidationResult.VALID, + dangerousSettings = DangerousSettings(usesRemoteConfigAPISources = true), + runningTests = true, + ) + + private fun createContext(): Context = mockk(relaxed = true).apply { + every { packageName } answers { "mock-package-name" } + every { getLocale() } returns Locale.US + } + + /** Builds a `sources` ConfigTopic whose api entries point at [urls], highest priority first. */ + private fun sourcesTopic(urls: List): ConfigTopic { + val item = RemoteConfiguration.ConfigItem( + metadata = buildJsonObject { + putJsonArray("sources") { + urls.forEachIndexed { index, url -> + addJsonObject { + put("url", url) + put("priority", index * 10) + put("weight", 1) + } + } + } + }, + ) + return ConfigTopic(mapOf("api" to item)) + } + + /** Routes requests by path so a server answers both its API endpoint and its health endpoint. */ + private fun MockWebServer.routeResponses(endpoint: MockResponse, health: MockResponse) { + dispatcher = object : Dispatcher() { + override fun dispatch(request: RecordedRequest): MockResponse = + if (request.path == healthPath) health else endpoint + } + } + + private fun MockWebServer.recordedPaths(): List = + (0 until requestCount).mapNotNull { takeRequest(1, TimeUnit.SECONDS)?.path } + + private fun response(responseCode: Int): MockResponse = MockResponse() + .setResponseCode(responseCode) + .setBody("{}") + + private fun customerInfoResponse(): MockResponse = MockResponse() + .setResponseCode(200) + .setBody(Responses.validFullPurchaserResponse) + + /** A url whose port refuses connections, to simulate an unreachable source. */ + private fun unreachableSourceUrl(): String { + val downServer = MockWebServer() + val url = downServer.url("/").toString() + downServer.shutdown() + return url + } + + // endregion +} diff --git a/purchases/src/test/java/com/revenuecat/purchases/common/networking/APISourceFailoverTest.kt b/purchases/src/test/java/com/revenuecat/purchases/common/networking/APISourceFailoverTest.kt new file mode 100644 index 0000000000..387b0d232f --- /dev/null +++ b/purchases/src/test/java/com/revenuecat/purchases/common/networking/APISourceFailoverTest.kt @@ -0,0 +1,210 @@ +package com.revenuecat.purchases.common.networking + +import androidx.test.ext.junit.runners.AndroidJUnit4 +import com.revenuecat.purchases.common.AppConfig +import com.revenuecat.purchases.common.networking.APISourceFailover.FailureDecision +import com.revenuecat.purchases.common.remoteconfig.RemoteConfigSource +import com.revenuecat.purchases.common.remoteconfig.RemoteConfigSourceHandle +import com.revenuecat.purchases.common.remoteconfig.RemoteConfigSourceProvider +import com.revenuecat.purchases.utils.TestUrlConnection +import com.revenuecat.purchases.utils.TestUrlConnectionFactory +import io.mockk.every +import io.mockk.mockk +import org.assertj.core.api.Assertions.assertThat +import org.junit.Before +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.URL + +@RunWith(AndroidJUnit4::class) +@Config(manifest = Config.NONE) +internal class APISourceFailoverTest { + + private class FakeSourceProvider(urls: List) : RemoteConfigSourceProvider { + private val sources = urls.map { RemoteConfigSource(url = it, priority = 0, weight = 0) } + private var index = 0 + private var token = 0 + val unhealthyReports = mutableListOf() + + override fun getCurrent(purpose: RemoteConfigSourceHandle.Purpose): RemoteConfigSourceHandle? = + sources.getOrNull(index)?.let { RemoteConfigSourceHandle(purpose, it, token) } + + override fun reportUnhealthy(handle: RemoteConfigSourceHandle) { + unhealthyReports.add(handle) + if (handle.token == token) { + index++ + token++ + } + } + + override fun restart(purpose: RemoteConfigSourceHandle.Purpose) { + index = 0 + token++ + } + + override fun restartIfExhausted(purpose: RemoteConfigSourceHandle.Purpose): Boolean { + if (index < sources.size) return false + restart(purpose) + return true + } + } + + private val defaultBaseURL = URL(AppConfig.baseUrlString) + private val eligibleEndpoint = Endpoint.GetCustomerInfo("test-user-id") + + private lateinit var appConfig: AppConfig + + @Before + fun setUp() { + appConfig = mockk() + every { appConfig.usesRemoteConfigAPISources } returns true + } + + private fun healthConnection(responseCode: Int) = + TestUrlConnection(responseCode, ByteArrayInputStream(ByteArray(0))) + + private fun failover( + provider: RemoteConfigSourceProvider, + connectionFactory: TestUrlConnectionFactory = TestUrlConnectionFactory( + connectionProvider = { healthConnection(200) }, + ), + ) = APISourceFailover(appConfig, provider, SourceHealthChecker(connectionFactory)) + + // region currentSource eligibility + + @Test + fun `currentSource resolves the provider's current source when eligible`() { + val provider = FakeSourceProvider(listOf("https://a.revenuecat.com/")) + val source = failover(provider).currentSource(eligibleEndpoint, defaultBaseURL, isFallbackAttempt = false) + assertThat(source?.url).isEqualTo(URL("https://a.revenuecat.com/")) + assertThat(source?.handle?.url).isEqualTo("https://a.revenuecat.com/") + } + + @Test + fun `currentSource is null when the dangerous setting is disabled`() { + every { appConfig.usesRemoteConfigAPISources } returns false + val provider = FakeSourceProvider(listOf("https://a.revenuecat.com/")) + assertThat(failover(provider).currentSource(eligibleEndpoint, defaultBaseURL, isFallbackAttempt = false)) + .isNull() + } + + @Test + fun `currentSource is null for endpoint fallback attempts`() { + val provider = FakeSourceProvider(listOf("https://a.revenuecat.com/")) + assertThat(failover(provider).currentSource(eligibleEndpoint, defaultBaseURL, isFallbackAttempt = true)) + .isNull() + } + + @Test + fun `currentSource is null for endpoints that do not use API sources`() { + val provider = FakeSourceProvider(listOf("https://a.revenuecat.com/")) + assertThat(failover(provider).currentSource(Endpoint.PostDiagnostics, defaultBaseURL, isFallbackAttempt = false)) + .isNull() + } + + @Test + fun `currentSource is null when the base URL is not the default host`() { + val provider = FakeSourceProvider(listOf("https://a.revenuecat.com/")) + val proxyURL = URL("https://proxy.example.com/") + assertThat(failover(provider).currentSource(eligibleEndpoint, proxyURL, isFallbackAttempt = false)) + .isNull() + } + + @Test + fun `currentSource is null when the provider is exhausted`() { + val provider = FakeSourceProvider(emptyList()) + assertThat(failover(provider).currentSource(eligibleEndpoint, defaultBaseURL, isFallbackAttempt = false)) + .isNull() + } + + @Test + fun `currentSource skips and reports sources with malformed urls`() { + val provider = FakeSourceProvider(listOf("not a url", "https://b.revenuecat.com/")) + val source = failover(provider).currentSource(eligibleEndpoint, defaultBaseURL, isFallbackAttempt = false) + assertThat(source?.url).isEqualTo(URL("https://b.revenuecat.com/")) + assertThat(provider.unhealthyReports.map { it.url }).containsExactly("not a url") + } + + @Test + fun `currentSource is null when every source url is malformed`() { + val provider = FakeSourceProvider(listOf("not a url", "also not a url")) + assertThat(failover(provider).currentSource(eligibleEndpoint, defaultBaseURL, isFallbackAttempt = false)) + .isNull() + assertThat(provider.unhealthyReports).hasSize(2) + } + + // endregion + + // region onRequestFailure + + @Test + fun `onRequestFailure does not fail over when the source's health check passes`() { + val provider = FakeSourceProvider(listOf("https://a.revenuecat.com/", "https://b.revenuecat.com/")) + val factory = TestUrlConnectionFactory(connectionProvider = { healthConnection(200) }) + val failover = failover(provider, factory) + + val source = failover.currentSource(eligibleEndpoint, defaultBaseURL, isFallbackAttempt = false)!! + val decision = failover.onRequestFailure(source) + + assertThat(decision).isEqualTo(FailureDecision.SourceHealthy) + assertThat(provider.unhealthyReports).isEmpty() + } + + @Test + fun `onRequestFailure fails over when the health check returns a non-2xx response`() { + val provider = FakeSourceProvider(listOf("https://a.revenuecat.com/", "https://b.revenuecat.com/")) + val factory = TestUrlConnectionFactory(connectionProvider = { healthConnection(503) }) + val failover = failover(provider, factory) + + val source = failover.currentSource(eligibleEndpoint, defaultBaseURL, isFallbackAttempt = false)!! + val decision = failover.onRequestFailure(source) + + assertThat(decision).isInstanceOf(FailureDecision.RetryNextSource::class.java) + assertThat((decision as FailureDecision.RetryNextSource).next.url) + .isEqualTo(URL("https://b.revenuecat.com/")) + assertThat(provider.unhealthyReports.map { it.url }).containsExactly("https://a.revenuecat.com/") + } + + @Test + fun `onRequestFailure fails over when the health check cannot connect`() { + val provider = FakeSourceProvider(listOf("https://a.revenuecat.com/", "https://b.revenuecat.com/")) + val factory = TestUrlConnectionFactory(connectionProvider = { throw IOException("unreachable") }) + val failover = failover(provider, factory) + + val source = failover.currentSource(eligibleEndpoint, defaultBaseURL, isFallbackAttempt = false)!! + val decision = failover.onRequestFailure(source) + + assertThat(decision).isInstanceOf(FailureDecision.RetryNextSource::class.java) + assertThat((decision as FailureDecision.RetryNextSource).next.url) + .isEqualTo(URL("https://b.revenuecat.com/")) + } + + @Test + fun `onRequestFailure reports exhaustion once the last source fails its health check`() { + val provider = FakeSourceProvider(listOf("https://a.revenuecat.com/")) + val factory = TestUrlConnectionFactory(connectionProvider = { throw IOException("unreachable") }) + val failover = failover(provider, factory) + + val source = failover.currentSource(eligibleEndpoint, defaultBaseURL, isFallbackAttempt = false)!! + assertThat(failover.onRequestFailure(source)).isEqualTo(FailureDecision.SourcesExhausted) + assertThat(provider.unhealthyReports.map { it.url }).containsExactly("https://a.revenuecat.com/") + } + + @Test + fun `onRequestFailure health-checks the current source's health path`() { + val provider = FakeSourceProvider(listOf("https://a.revenuecat.com/", "https://b.revenuecat.com/")) + val factory = TestUrlConnectionFactory(connectionProvider = { healthConnection(200) }) + val failover = failover(provider, factory) + + val source = failover.currentSource(eligibleEndpoint, defaultBaseURL, isFallbackAttempt = false)!! + failover.onRequestFailure(source) + + assertThat(factory.createdConnections) + .containsExactly("https://a.revenuecat.com/v1/health/connectivity") + } + + // endregion +} diff --git a/purchases/src/test/java/com/revenuecat/purchases/common/networking/SourceHealthCheckerTest.kt b/purchases/src/test/java/com/revenuecat/purchases/common/networking/SourceHealthCheckerTest.kt new file mode 100644 index 0000000000..b33118f179 --- /dev/null +++ b/purchases/src/test/java/com/revenuecat/purchases/common/networking/SourceHealthCheckerTest.kt @@ -0,0 +1,178 @@ +package com.revenuecat.purchases.common.networking + +import androidx.test.ext.junit.runners.AndroidJUnit4 +import com.revenuecat.purchases.common.DateProvider +import com.revenuecat.purchases.utils.TestUrlConnection +import com.revenuecat.purchases.utils.TestUrlConnectionFactory +import com.revenuecat.purchases.utils.UrlConnection +import com.revenuecat.purchases.utils.UrlConnectionFactory +import org.assertj.core.api.Assertions.assertThat +import org.junit.Test +import org.junit.runner.RunWith +import org.robolectric.annotation.Config +import java.io.ByteArrayInputStream +import java.io.IOException +import java.util.Date +import java.util.concurrent.CountDownLatch +import java.util.concurrent.TimeUnit +import java.util.concurrent.atomic.AtomicLong +import kotlin.concurrent.thread + +@RunWith(AndroidJUnit4::class) +@Config(manifest = Config.NONE) +internal class SourceHealthCheckerTest { + + private class FakeDateProvider(private val currentTime: AtomicLong = AtomicLong(1_000_000L)) : DateProvider { + override val now: Date + get() = Date(currentTime.get()) + + fun advanceTime(millis: Long) { + currentTime.addAndGet(millis) + } + } + + private val sourceBaseUrl = "https://api.revenuecat.com/" + private val healthUrl = "https://api.revenuecat.com/v1/health/connectivity" + + private fun connection(responseCode: Int) = TestUrlConnection(responseCode, ByteArrayInputStream(ByteArray(0))) + + private fun checker( + factory: UrlConnectionFactory, + dateProvider: DateProvider = FakeDateProvider(), + ) = SourceHealthChecker(factory, dateProvider) + + @Test + fun `2xx responses are healthy`() { + listOf(200, 204, 299).forEach { responseCode -> + val factory = TestUrlConnectionFactory(mapOf(healthUrl to connection(responseCode))) + assertThat(checker(factory).isHealthy(sourceBaseUrl)) + .withFailMessage("Expected $responseCode to be healthy") + .isTrue() + } + } + + @Test + fun `non-2xx responses are not healthy`() { + listOf(301, 404, 500, 503).forEach { responseCode -> + val factory = TestUrlConnectionFactory(mapOf(healthUrl to connection(responseCode))) + assertThat(checker(factory).isHealthy(sourceBaseUrl)) + .withFailMessage("Expected $responseCode to not be healthy") + .isFalse() + } + } + + @Test + fun `connection failure is not healthy`() { + val factory = TestUrlConnectionFactory(connectionProvider = { throw IOException("no route to host") }) + assertThat(checker(factory).isHealthy(sourceBaseUrl)).isFalse() + } + + @Test + fun `checks the health connectivity path of the source`() { + val factory = TestUrlConnectionFactory(mapOf(healthUrl to connection(200))) + checker(factory).isHealthy(sourceBaseUrl) + assertThat(factory.createdConnections).containsExactly(healthUrl) + } + + @Test + fun `builds the health url for a base url without a trailing slash`() { + val factory = TestUrlConnectionFactory(mapOf(healthUrl to connection(200))) + checker(factory).isHealthy("https://api.revenuecat.com") + assertThat(factory.createdConnections).containsExactly(healthUrl) + } + + @Test + fun `disconnects the connection`() { + val connection = connection(200) + val factory = TestUrlConnectionFactory(mapOf(healthUrl to connection)) + checker(factory).isHealthy(sourceBaseUrl) + assertThat(connection.isDisconnected).isTrue() + } + + @Test + fun `disconnects the connection when reading the response fails`() { + val connection = object : UrlConnection { + var isDisconnected = false + override val responseCode: Int + get() = throw IOException("connection reset") + override val inputStream = ByteArrayInputStream(ByteArray(0)) + override fun disconnect() { + isDisconnected = true + } + } + val factory = object : UrlConnectionFactory { + override fun createConnection(url: String, requestMethod: String): UrlConnection = connection + } + assertThat(SourceHealthChecker(factory, FakeDateProvider()).isHealthy(sourceBaseUrl)).isFalse() + assertThat(connection.isDisconnected).isTrue() + } + + @Test + fun `caches the result within its validity window`() { + val factory = TestUrlConnectionFactory(connectionProvider = { connection(200) }) + val checker = checker(factory) + assertThat(checker.isHealthy(sourceBaseUrl)).isTrue() + assertThat(checker.isHealthy(sourceBaseUrl)).isTrue() + assertThat(factory.createdConnections).hasSize(1) + } + + @Test + fun `caches unhealthy results too`() { + val factory = TestUrlConnectionFactory(connectionProvider = { connection(500) }) + val checker = checker(factory) + assertThat(checker.isHealthy(sourceBaseUrl)).isFalse() + assertThat(checker.isHealthy(sourceBaseUrl)).isFalse() + assertThat(factory.createdConnections).hasSize(1) + } + + @Test + fun `checks again once the cached result expires`() { + val dateProvider = FakeDateProvider() + val factory = TestUrlConnectionFactory(connectionProvider = { connection(200) }) + val checker = checker(factory, dateProvider) + checker.isHealthy(sourceBaseUrl) + dateProvider.advanceTime(10_000L) + checker.isHealthy(sourceBaseUrl) + assertThat(factory.createdConnections).hasSize(2) + } + + @Test + fun `caches results per source`() { + val otherHealthUrl = "https://api.rc-backup.com/v1/health/connectivity" + val factory = TestUrlConnectionFactory( + mapOf( + healthUrl to connection(200), + otherHealthUrl to connection(503), + ), + ) + val checker = checker(factory) + assertThat(checker.isHealthy(sourceBaseUrl)).isTrue() + assertThat(checker.isHealthy("https://api.rc-backup.com/")).isFalse() + assertThat(factory.createdConnections).containsExactly(healthUrl, otherHealthUrl) + } + + @Test + fun `concurrent checks for the same source share one request`() { + val checkStarted = CountDownLatch(1) + val releaseCheck = CountDownLatch(1) + val factory = TestUrlConnectionFactory( + connectionProvider = { + checkStarted.countDown() + assertThat(releaseCheck.await(5, TimeUnit.SECONDS)).isTrue() + connection(200) + }, + ) + val checker = checker(factory) + var firstResult = false + var secondResult = false + val firstCheck = thread { firstResult = checker.isHealthy(sourceBaseUrl) } + assertThat(checkStarted.await(5, TimeUnit.SECONDS)).isTrue() + val secondCheck = thread { secondResult = checker.isHealthy(sourceBaseUrl) } + releaseCheck.countDown() + firstCheck.join(5_000) + secondCheck.join(5_000) + assertThat(firstResult).isTrue() + assertThat(secondResult).isTrue() + assertThat(factory.createdConnections).hasSize(1) + } +} diff --git a/purchases/src/test/java/com/revenuecat/purchases/common/remoteconfig/RemoteConfigSourceProviderTest.kt b/purchases/src/test/java/com/revenuecat/purchases/common/remoteconfig/RemoteConfigSourceProviderTest.kt index 29fb3ea4d6..3f340f2c53 100644 --- a/purchases/src/test/java/com/revenuecat/purchases/common/remoteconfig/RemoteConfigSourceProviderTest.kt +++ b/purchases/src/test/java/com/revenuecat/purchases/common/remoteconfig/RemoteConfigSourceProviderTest.kt @@ -1,6 +1,7 @@ package com.revenuecat.purchases.common.remoteconfig import androidx.test.ext.junit.runners.AndroidJUnit4 +import com.revenuecat.purchases.common.DateProvider import com.revenuecat.purchases.common.remoteconfig.RemoteConfigSourceHandle.Purpose import kotlinx.serialization.json.addJsonObject import kotlinx.serialization.json.buildJsonObject @@ -10,10 +11,12 @@ import org.assertj.core.api.Assertions.assertThat import org.junit.Test import org.junit.runner.RunWith import org.robolectric.annotation.Config +import java.util.Date import java.util.concurrent.ConcurrentHashMap import java.util.concurrent.CountDownLatch import java.util.concurrent.Executors import java.util.concurrent.TimeUnit +import java.util.concurrent.atomic.AtomicLong import kotlin.random.Random @RunWith(AndroidJUnit4::class) @@ -366,6 +369,145 @@ class RemoteConfigSourceProviderTest { // endregion + // region API restart interval (TTL re-arm) + + @Test + fun `advanced api list restarts from the top after the restart interval`() { + val dateProvider = FakeDateProvider() + val provider = apiProvider(listOf(source("a"), source("b")), dateProvider) + + provider.reportUnhealthy(provider.getCurrent(Purpose.API)!!) // a -> b + assertThat(provider.getCurrent(Purpose.API)?.url).isEqualTo(url("b")) + + dateProvider.advanceTime(600_000L) + assertThat(provider.getCurrent(Purpose.API)?.url).isEqualTo(url("a")) + } + + @Test + fun `advanced api list keeps its position just under the restart interval`() { + val dateProvider = FakeDateProvider() + val provider = apiProvider(listOf(source("a"), source("b")), dateProvider) + + provider.reportUnhealthy(provider.getCurrent(Purpose.API)!!) // a -> b + + dateProvider.advanceTime(599_999L) + assertThat(provider.getCurrent(Purpose.API)?.url).isEqualTo(url("b")) + } + + @Test + fun `exhausted api list re-arms after the restart interval`() { + val dateProvider = FakeDateProvider() + val provider = apiProvider(listOf(source("a"), source("b")), dateProvider) + + provider.reportUnhealthy(provider.getCurrent(Purpose.API)!!) // a -> b + provider.reportUnhealthy(provider.getCurrent(Purpose.API)!!) // b -> exhausted + assertThat(provider.getCurrent(Purpose.API)).isNull() + + dateProvider.advanceTime(599_999L) + assertThat(provider.getCurrent(Purpose.API)).isNull() + + dateProvider.advanceTime(1L) + assertThat(provider.getCurrent(Purpose.API)?.url).isEqualTo(url("a")) + } + + @Test + fun `the restart interval counts from the first advance, not the latest`() { + // A flapping list must not postpone the periodic return to the primary: advancing again + // mid-interval does not reset the timer. + val dateProvider = FakeDateProvider() + val provider = apiProvider(listOf(source("a"), source("b"), source("c")), dateProvider) + + provider.reportUnhealthy(provider.getCurrent(Purpose.API)!!) // a -> b, timer starts + dateProvider.advanceTime(500_000L) + provider.reportUnhealthy(provider.getCurrent(Purpose.API)!!) // b -> c, timer unchanged + + dateProvider.advanceTime(100_000L) // 600s since the first advance + assertThat(provider.getCurrent(Purpose.API)?.url).isEqualTo(url("a")) + } + + @Test + fun `handle from before the interval restart is stale`() { + val dateProvider = FakeDateProvider() + val provider = apiProvider(listOf(source("a"), source("b")), dateProvider) + + provider.reportUnhealthy(provider.getCurrent(Purpose.API)!!) // a -> b + val stale = provider.getCurrent(Purpose.API) + assertThat(stale?.url).isEqualTo(url("b")) + + dateProvider.advanceTime(600_000L) + assertThat(provider.getCurrent(Purpose.API)?.url).isEqualTo(url("a")) + + // The pre-restart handle belongs to the previous cycle, so it must not advance the list. + provider.reportUnhealthy(stale!!) + assertThat(provider.getCurrent(Purpose.API)?.url).isEqualTo(url("a")) + } + + @Test + fun `a new failover cycle after an interval restart needs a full interval again`() { + val dateProvider = FakeDateProvider() + val provider = apiProvider(listOf(source("a"), source("b")), dateProvider) + + provider.reportUnhealthy(provider.getCurrent(Purpose.API)!!) // a -> b + dateProvider.advanceTime(600_000L) + assertThat(provider.getCurrent(Purpose.API)?.url).isEqualTo(url("a")) + + provider.reportUnhealthy(provider.getCurrent(Purpose.API)!!) // a -> b, new timer + dateProvider.advanceTime(599_999L) + assertThat(provider.getCurrent(Purpose.API)?.url).isEqualTo(url("b")) + + dateProvider.advanceTime(1L) + assertThat(provider.getCurrent(Purpose.API)?.url).isEqualTo(url("a")) + } + + @Test + fun `explicit restart resets the interval timer`() { + val dateProvider = FakeDateProvider() + val provider = apiProvider(listOf(source("a"), source("b")), dateProvider) + + provider.reportUnhealthy(provider.getCurrent(Purpose.API)!!) // a -> b, timer starts + dateProvider.advanceTime(500_000L) + provider.restart(Purpose.API) // timer cleared + + provider.reportUnhealthy(provider.getCurrent(Purpose.API)!!) // a -> b, fresh timer + dateProvider.advanceTime(599_999L) // over 600s since the ORIGINAL advance, under since the fresh one + assertThat(provider.getCurrent(Purpose.API)?.url).isEqualTo(url("b")) + } + + @Test + fun `a changed sources topic resets the interval timer`() { + val dateProvider = FakeDateProvider() + val store = FakeTopicStore(sourcesTopic(api = listOf(source("a"), source("b")), blob = emptyList())) + val provider = DefaultRemoteConfigSourceProvider(store, FakeRandom(0), dateProvider) + + provider.reportUnhealthy(provider.getCurrent(Purpose.API)!!) // a -> b, timer starts + dateProvider.advanceTime(500_000L) + + store.sources = sourcesTopic(api = listOf(source("x"), source("y")), blob = emptyList()) + provider.reportUnhealthy(provider.getCurrent(Purpose.API)!!) // x -> y, fresh timer + + dateProvider.advanceTime(599_999L) + assertThat(provider.getCurrent(Purpose.API)?.url).isEqualTo(url("y")) + + dateProvider.advanceTime(1L) + assertThat(provider.getCurrent(Purpose.API)?.url).isEqualTo(url("x")) + } + + @Test + fun `blob progress is unaffected by elapsed time`() { + val dateProvider = FakeDateProvider() + val provider = provider( + api = listOf(source("api1")), + blob = listOf(source("blob1", priority = 0), source("blob2", priority = 10)), + dateProvider = dateProvider, + ) + + provider.reportUnhealthy(provider.getCurrent(Purpose.BLOB)!!) // blob1 -> blob2 + dateProvider.advanceTime(600_000L) + assertThat(provider.getCurrent(Purpose.BLOB)?.url).isEqualTo(url("blob2")) + } + + // endregion + // region Sources topic changes @Test @@ -488,14 +630,18 @@ class RemoteConfigSourceProviderTest { private fun source(host: String, priority: Int = 0, weight: Int = 0): RemoteConfigSource = RemoteConfigSource(url = url(host), priority = priority, weight = weight) - private fun apiProvider(sources: List): RemoteConfigSourceProvider = - provider(api = sources, blob = emptyList()) + private fun apiProvider( + sources: List, + dateProvider: DateProvider = FakeDateProvider(), + ): RemoteConfigSourceProvider = + provider(api = sources, blob = emptyList(), dateProvider = dateProvider) private fun provider( api: List, blob: List, + dateProvider: DateProvider = FakeDateProvider(), ): RemoteConfigSourceProvider = - DefaultRemoteConfigSourceProvider(FakeTopicStore(sourcesTopic(api, blob)), FakeRandom(0)) + DefaultRemoteConfigSourceProvider(FakeTopicStore(sourcesTopic(api, blob)), FakeRandom(0), dateProvider) /** Builds a `sources` ConfigTopic matching the backend shape: api entries use `url`, blob use `url_format`. */ private fun sourcesTopic(api: List, blob: List): ConfigTopic { @@ -520,6 +666,15 @@ class RemoteConfigSourceProviderTest { if (topic == RemoteConfigTopic.Sources) sources else null } + private class FakeDateProvider(private val currentTime: AtomicLong = AtomicLong(1_000_000L)) : DateProvider { + override val now: Date + get() = Date(currentTime.get()) + + fun advanceTime(millis: Long) { + currentTime.addAndGet(millis) + } + } + private fun runConcurrently(iterations: Int, block: () -> Unit) { val pool = Executors.newFixedThreadPool(16) val start = CountDownLatch(1)