Skip to content

fix(sync): bound recursive incentive retries with exponential backoff… #452

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
Piyush0000 wants to merge 1 commit into
base: main
Choose a base branch
from
Open

fix(sync): bound recursive incentive retries with exponential backoff… #452

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
93 changes: 93 additions & 0 deletions app/src/main/java/org/piramalswasthya/sakhi/helpers/RetryWithBackoff.kt
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,93 @@
package org.piramalswasthya.sakhi.helpers

import kotlinx.coroutines.delay
import timber.log.Timber
import java.net.SocketTimeoutException

/**
* Default number of attempts (including the initial call) used when a caller
* does not supply an explicit value.
*/
const val DEFAULT_MAX_RETRY_ATTEMPTS: Int = 3

/**
* Default delay before the second attempt, in milliseconds.
*/
const val DEFAULT_INITIAL_RETRY_DELAY_MS: Long = 1_000L

/**
* Upper bound for the per-attempt delay so a long backoff chain cannot stall
* a sync for an unbounded period.
*/
const val DEFAULT_MAX_RETRY_DELAY_MS: Long = 30_000L

/**
* Multiplicative growth factor applied between attempts.
*/
const val DEFAULT_BACKOFF_FACTOR: Double = 2.0

/**
* Executes [block] with bounded retries and exponential backoff.
*
* This replaces the legacy pattern where a repository function caught a
* transient network exception and recursively re-invoked itself. With sustained
* connectivity failures the unbounded recursion produced StackOverflowErrors
* (AMRIT#156); switching to an iterative backoff also keeps the coroutine
* cancellable while it waits because [delay] is suspending rather than
* blocking.
*
* The default [retryOn] predicate matches only [SocketTimeoutException] to
* preserve the historical scope of which failures are considered transient.
* Callers needing a broader policy can pass their own predicate.
*
* @param maxAttempts total attempts including the initial call; must be >= 1
* @param initialDelayMs delay before the second attempt; must be >= 0
* @param maxDelayMs ceiling applied after each multiplicative step;
* must be >= [initialDelayMs]
* @param backoffFactor multiplicative growth applied between attempts;
* must be > 0
* @param retryOn predicate deciding whether a thrown [Throwable] is retryable
* @param block the suspending body to execute
* @return the result of the first successful invocation of [block]
* @throws Throwable the last exception observed after attempts are exhausted,
* or the first non-retryable exception encountered
*/
suspend fun <T> retryWithBackoff(
maxAttempts: Int = DEFAULT_MAX_RETRY_ATTEMPTS,
initialDelayMs: Long = DEFAULT_INITIAL_RETRY_DELAY_MS,
maxDelayMs: Long = DEFAULT_MAX_RETRY_DELAY_MS,
backoffFactor: Double = DEFAULT_BACKOFF_FACTOR,
retryOn: (Throwable) -> Boolean = ::isTransientNetworkFailure,
block: suspend () -> T
): T {
require(maxAttempts >= 1) { "maxAttempts must be at least 1, was $maxAttempts" }
require(initialDelayMs >= 0) { "initialDelayMs must be non-negative, was $initialDelayMs" }
require(maxDelayMs >= initialDelayMs) { "maxDelayMs must be >= initialDelayMs" }
require(backoffFactor > 0.0) { "backoffFactor must be positive, was $backoffFactor" }

var currentDelay = initialDelayMs
repeat(maxAttempts - 1) { attempt ->
try {
return block()
} catch (t: Throwable) {
if (!retryOn(t)) throw t
Timber.w(
t,
"retryWithBackoff: attempt %d/%d failed; sleeping %dms before retry",
attempt + 1, maxAttempts, currentDelay
)
delay(currentDelay)
currentDelay = (currentDelay.toDouble() * backoffFactor)
.toLong()
.coerceAtMost(maxDelayMs)
}
}
return block()
}

/**
* Default retry predicate. Matches only [SocketTimeoutException]; aligned with
* the pre-existing catch sites in repository classes so behaviour for
* non-timeout failures (auth, parse, business errors) remains identical.
*/
fun isTransientNetworkFailure(t: Throwable): Boolean = t is SocketTimeoutException
185 changes: 93 additions & 92 deletions app/src/main/java/org/piramalswasthya/sakhi/repositories/IncentiveRepo.kt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -12,6 +12,7 @@ import okhttp3.RequestBody.Companion.asRequestBody
import org.json.JSONObject
import org.piramalswasthya.sakhi.database.room.dao.IncentiveDao
import org.piramalswasthya.sakhi.database.shared_preferences.PreferenceDao
import org.piramalswasthya.sakhi.helpers.retryWithBackoff
import org.piramalswasthya.sakhi.helpers.setToEndOfTheDay
import org.piramalswasthya.sakhi.model.IncentiveActivityListRequest
import org.piramalswasthya.sakhi.model.IncentiveActivityNetwork
Expand Down Expand Up @@ -46,61 +47,61 @@ class IncentiveRepo @Inject constructor(
suspend fun pullAndSaveAllIncentiveActivities(user: User): Boolean {
return withContext(Dispatchers.IO) {
try {
val currentLang = preferenceDao.getCurrentLanguage().symbol
val stateId = user.state.id
val districtId = user.district.id
val requestBody = IncentiveActivityListRequest(stateId, districtId, currentLang)
val response = amritApiService.getAllIncentiveActivities(requestBody = requestBody)
val statusCode = response.code()
if (statusCode == 200) {
val responseString = response.body()?.string()
if (responseString != null) {
val jsonObj = JSONObject(responseString)

val errorMessage = jsonObj.getString("errorMessage")
val responseStatusCode = jsonObj.getInt("statusCode")
Timber.d("Pull from amrit incentives data : $responseStatusCode")
when (responseStatusCode) {
200 -> {
try {
val dataObj = jsonObj.getString("data")
saveIncentiveMasterData(dataObj)
} catch (e: Exception) {
Timber.d("Incentive master data not synced $e")
return@withContext false
retryWithBackoff {
val currentLang = preferenceDao.getCurrentLanguage().symbol
val stateId = user.state.id
val districtId = user.district.id
val requestBody = IncentiveActivityListRequest(stateId, districtId, currentLang)
val response = amritApiService.getAllIncentiveActivities(requestBody = requestBody)
val statusCode = response.code()
if (statusCode == 200) {
val responseString = response.body()?.string()
if (responseString != null) {
val jsonObj = JSONObject(responseString)

val errorMessage = jsonObj.getString("errorMessage")
val responseStatusCode = jsonObj.getInt("statusCode")
Timber.d("Pull from amrit incentives data : $responseStatusCode")
when (responseStatusCode) {
200 -> {
try {
val dataObj = jsonObj.getString("data")
saveIncentiveMasterData(dataObj)
} catch (e: Exception) {
Timber.d("Incentive master data not synced $e")
return@retryWithBackoff false
}

return@retryWithBackoff true
}

return@withContext true
}

401, 5002 -> {
if (userRepo.refreshTokenTmc(
user.userName, user.password
)
) throw SocketTimeoutException("Refreshed Token!")
else throw IllegalStateException("User Logged out!!")
}
401, 5002 -> {
if (userRepo.refreshTokenTmc(
user.userName, user.password
)
) throw SocketTimeoutException("Refreshed Token!")
else throw IllegalStateException("User Logged out!!")
}

5000 -> {
if (errorMessage == "No record found") return@withContext true
}
5000 -> {
if (errorMessage == "No record found") return@retryWithBackoff true
}

else -> {
throw IllegalStateException("$responseStatusCode received, don't know what todo!?")
else -> {
throw IllegalStateException("$responseStatusCode received, don't know what todo!?")
}
}
}
}
true
}

} catch (e: SocketTimeoutException) {
Timber.e("incentives error : $e")
pullAndSaveAllIncentiveActivities(user)
return@withContext true
Timber.e(e, "incentives error after exhausting retries")
false
} catch (e: Exception) {
Timber.d("Caught $e at incentives!")
return@withContext false
false
}
true
}
}

Expand All @@ -117,67 +118,67 @@ class IncentiveRepo @Inject constructor(
suspend fun pullAndSaveAllIncentiveRecords(user: User): Boolean {
return withContext(Dispatchers.IO) {
try {

val requestBody = IncentiveRecordListRequest(
user.userId,
getDateTimeStringFromLong(
preferenceDao.lastIncentivePullTimestamp
)!!,
getDateTimeStringFromLong(
Calendar.getInstance().setToEndOfTheDay().timeInMillis
)!!,
villageID = user.state.id
)
val response = amritApiService.getAllIncentiveRecords(requestBody = requestBody)
val statusCode = response.code()
if (statusCode == 200) {
val responseString = response.body()?.string()
if (responseString != null) {
val jsonObj = JSONObject(responseString)

val errorMessage = jsonObj.getString("errorMessage")
val responseStatusCode = jsonObj.getInt("statusCode")
Timber.d("Pull from amrit incentives data : $responseStatusCode")
when (responseStatusCode) {
200 -> {
try {
val dataObj = jsonObj.getString("data")
saveIncentiveRecordsData(dataObj)
} catch (e: Exception) {
Timber.d("Incentive master data not synced $e")
return@withContext false
retryWithBackoff {
val requestBody = IncentiveRecordListRequest(
user.userId,
getDateTimeStringFromLong(
preferenceDao.lastIncentivePullTimestamp
)!!,
getDateTimeStringFromLong(
Calendar.getInstance().setToEndOfTheDay().timeInMillis
)!!,
villageID = user.state.id
)
val response = amritApiService.getAllIncentiveRecords(requestBody = requestBody)
val statusCode = response.code()
if (statusCode == 200) {
val responseString = response.body()?.string()
if (responseString != null) {
val jsonObj = JSONObject(responseString)

val errorMessage = jsonObj.getString("errorMessage")
val responseStatusCode = jsonObj.getInt("statusCode")
Timber.d("Pull from amrit incentives data : $responseStatusCode")
when (responseStatusCode) {
200 -> {
try {
val dataObj = jsonObj.getString("data")
saveIncentiveRecordsData(dataObj)
} catch (e: Exception) {
Timber.d("Incentive master data not synced $e")
return@retryWithBackoff false
}

return@retryWithBackoff true
}

return@withContext true
}

401, 5002 -> {
if (userRepo.refreshTokenTmc(
user.userName, user.password
)
) throw SocketTimeoutException("Refreshed Token!")
else throw IllegalStateException("User Logged out!!")
}
401, 5002 -> {
if (userRepo.refreshTokenTmc(
user.userName, user.password
)
) throw SocketTimeoutException("Refreshed Token!")
else throw IllegalStateException("User Logged out!!")
}

5000 -> {
if (errorMessage == "No record found") return@withContext true
}
5000 -> {
if (errorMessage == "No record found") return@retryWithBackoff true
}

else -> {
throw IllegalStateException("$responseStatusCode received, don't know what todo!?")
else -> {
throw IllegalStateException("$responseStatusCode received, don't know what todo!?")
}
}
}
}
true
}
} catch (e: SocketTimeoutException) {
Timber.e("incentives error : $e")
pullAndSaveAllIncentiveRecords(user)
return@withContext true
Timber.e(e, "incentives error after exhausting retries")
false
} catch (e: Exception) {
Timber.d("Caught $e at incentives!")
return@withContext false
false
}
true
}
}

Expand Down
Loading