Skip to content

refactor: batch paykit contact cleanup #1099

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
ben-kaufman wants to merge 2 commits into
base: codex/paykit-incoming-payment-requests-rc39
Choose a base branch
from
+327 −60
Open

refactor: batch paykit contact cleanup #1099

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
f2427b1
refactor: batch paykit contact cleanup
ben-kaufman Jul 22, 2026
0b91ae2
fix: preserve paykit cleanup retries
ben-kaufman Jul 22, 2026
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
259 changes: 199 additions & 60 deletions app/src/main/java/to/bitkit/repositories/PrivatePaykitRepo.kt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -107,6 +107,22 @@ class PrivatePaykitRepo @Inject constructor(
val firstError: Throwable?,
)

private data class PrivateEndpointCleanupPreparation(
val clearedRetryKeys: List<PrivateMessageDrainRetryKey>,
val failedPublicKeys: Set<String>,
val firstError: Throwable?,
)

private data class NormalizedPublicKeyBatch(
val publicKeys: List<String>,
val hadInvalidKey: Boolean,
)

private data class LinkedReceiverPathInspection(
val receiverPaths: Set<String>,
val error: Throwable?,
)

private data class PrivateMessageDrainRetryKey(
val publicKey: String,
val receiverPath: String,
Expand Down Expand Up @@ -703,6 +719,7 @@ class PrivatePaykitRepo @Inject constructor(
var firstError: Throwable? = null
val updates = mutableListOf<PrivatePaymentListReservationUpdateInput>()
val linkRetryKeys = mutableListOf<PrivateMessageDrainRetryKey>()
val linkedReceiverPathsSnapshot = linkedReceiverPathsSnapshotOrNull(reason)

for (publicKey in publicKeys) {
val receiverPaths = runSuspendCatching { receiverPathsForSavedContact(publicKey) }
Expand All @@ -719,27 +736,21 @@ class PrivatePaykitRepo @Inject constructor(
val publicationReceiverPaths = receiverPathSelection.publishableReceiverPaths
receiverPathSelection.error?.let {
firstError = firstError ?: it
Logger.warn(
"Failed to inspect private Paykit receiver markers for '${redacted(publicKey)}' during '$reason'",
it,
context = TAG,
)
logPrivateReceiverPathSelectionFailure(publicKey, reason, it)
}
val linkedReceiverPathInspection = inspectLinkedReceiverPaths(
publicKey,
linkedReceiverPathsSnapshot,
reason,
)
firstError = firstError ?: linkedReceiverPathInspection.error
val cleanupReceiverPaths = receiverPathsForPrivateEndpointCleanup(
publicKey = publicKey,
excludedReceiverPaths = publicationReceiverPaths + receiverPathSelection.cleanupProtectedReceiverPaths,
linkedReceiverPaths = linkedReceiverPathInspection.receiverPaths,
)

(linkableReceiverPaths + cleanupReceiverPaths).distinct().forEach { receiverPath ->
linkRetryKeys += PrivateMessageDrainRetryKey(publicKey, receiverPath)
runSuspendCatching { paykitSdkService.ensureLinkWithPeer(publicKey, receiverPath) }.onFailure {
Logger.warn(
"Failed to prepare private Paykit link for '${redacted(publicKey)}' during '$reason'",
it,
context = TAG,
)
}
}
linkRetryKeys += preparePrivateLinks(publicKey, linkableReceiverPaths + cleanupReceiverPaths, reason)

cleanupReceiverPaths.forEach { receiverPath ->
updates += PrivatePaymentListReservationUpdateInput(
Expand Down Expand Up @@ -791,6 +802,21 @@ class PrivatePaykitRepo @Inject constructor(
drainAndSchedulePrivateLinkRetries(reason, retryKeys.distinct())
}

private suspend fun preparePrivateLinks(
publicKey: String,
receiverPaths: Collection<String>,
reason: String,
): List<PrivateMessageDrainRetryKey> = receiverPaths.distinct().map { receiverPath ->

@jvsena42 jvsena42 Jul 31, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Nit: .map { } here is really a side-effecting loop — the lambda's actual work is ensureLinkWithPeer, and the returned value is unrelated to it. onEach/an explicit loop building the list would read more honestly about the intent. Behavior is preserved (the retry key is added regardless of whether the link succeeded), so this is purely stylistic.

runSuspendCatching { paykitSdkService.ensureLinkWithPeer(publicKey, receiverPath) }.onFailure {
Logger.warn(
"Failed to prepare private Paykit link for '${redacted(publicKey)}' during '$reason'",
it,
context = TAG,
)
}
PrivateMessageDrainRetryKey(publicKey, receiverPath)
}

private suspend fun drainAndSchedulePrivateLinkRetries(
reason: String,
retryKeys: Collection<PrivateMessageDrainRetryKey>,
Expand Down Expand Up @@ -837,6 +863,18 @@ class PrivatePaykitRepo @Inject constructor(
}
}

private fun logPrivateReceiverPathSelectionFailure(
publicKey: String,
reason: String,
error: Throwable,
) {
Logger.warn(
"Failed to inspect private Paykit receiver markers for '${redacted(publicKey)}' during '$reason'",
error,
context = TAG,
)
}

private suspend fun applyPrivatePaymentListDeliveryReport(
report: PrivatePaymentListDeliveryReport,
reason: String,
Expand Down Expand Up @@ -1190,41 +1228,96 @@ class PrivatePaykitRepo @Inject constructor(
}

private suspend fun removePublishedEndpoints(): Result<Unit> = withContext(serializedDispatcher) {
runSuspendCatching {
val keys = (knownSavedContactKeys + ensureState().contacts.keys + pendingDeletedContactCleanupPublicKeys())
.distinct()
val firstError = keys.mapNotNull { publicKey ->
removePublishedEndpoints(publicKey).exceptionOrNull()
}.firstOrNull()
if (firstError != null) throw firstError
}
val keys = (knownSavedContactKeys + ensureState().contacts.keys + pendingDeletedContactCleanupPublicKeys())
.distinct()
removePublishedEndpoints(keys)
}

private suspend fun removePublishedEndpoints(publicKey: String): Result<Unit> = withContext(serializedDispatcher) {
runSuspendCatching {
var firstError: Throwable? = null
receiverPathsForCleanup(publicKey).forEach { receiverPath ->
val result = runSuspendCatching {
val report = paykitSdkService.clearPrivatePaymentList(
counterparty = publicKey,
receiverPath = receiverPath,
private suspend fun removePublishedEndpoints(publicKey: String): Result<Unit> =
removePublishedEndpoints(listOf(publicKey))

@jvsena42 jvsena42 Jul 31, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

retryPendingDeletedContactEndpointRemoval (line 1563) still loops this single-key overload per pending contact, so each one gets its own linkedPeers() snapshot and its own drain — exactly the multi-contact pattern this PR set out to batch. Passing the whole pending set to the Collection overload would get the same win there.

Separately (pre-existing, but it interacts badly with the invalid-key issue): the .getOrThrow() inside that forEach aborts the remaining pending keys on the first failure, so one bad key blocks every key after it in iteration order.


private suspend fun removePublishedEndpoints(publicKeys: Collection<String>): Result<Unit> =
withContext(serializedDispatcher) {
runSuspendCatching {
val normalizedBatch = normalizedPublicKeyBatch(publicKeys)
val publicKeys = normalizedBatch.publicKeys

@jvsena42 jvsena42 Jul 31, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Nit: this shadows the function parameter. It's consistent with the existing val retryKeys = retryKeys.toSet() in pendingPrivateMessageDrainKeys, so take it or leave it — but a distinct name (normalizedKeys) reads better and makes it obvious the raw input is intentionally no longer used.

var firstError: Throwable? = PrivatePaykitError.InvalidPublicKey.takeIf {

@jvsena42 jvsena42 Jul 31, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Invalid pending key now poisons every cleanup pass, permanently. (Android-only, new in 0b91ae2 — no iOS counterpart.)

Seeding firstError with InvalidPublicKey means it is rethrown at the end of the batch. If a non-normalizable key ever lands in deletedContactCleanupPendingPublicKeys, it can never be cleared, so:

  • retryPendingEndpointRemoval fails on every call forever → updateContactSharingCleanupPending(false) (line 221) never runs;
  • removePublishedEndpointsForCleanup sets cleanupPending = true forever — affects ProfileViewModel:110, EditProfileViewModel:230, WipeWalletUseCase:57;
  • disableSharingAndPruneUnsavedContactState throws permanently.

Pre-PR, the un-normalized single-key path silently no-oped and then cleared the marker via updateDeletedContactCleanupPending(key, false) — it self-healed. Current writers all normalize, so this only bites on legacy/corrupt persisted data, but the failure is unrecoverable and takes every other contact's cleanup down with it.

An unnormalizable key cannot correspond to any SDK state (the SDK is keyed by normalized pubkeys), so I'd drop it from the pending set with a warning rather than retaining it forever. At minimum, don't let it feed the batch-wide firstError.

normalizedBatch.hadInvalidKey
}
if (publicKeys.isEmpty()) {
firstError?.let { throw it }
return@runSuspendCatching Unit
}

val linkedReceiverPathsSnapshot = linkedReceiverPathsSnapshotOrNull("private endpoint cleanup")
val preparation = clearPrivatePaymentLists(publicKeys, linkedReceiverPathsSnapshot)
val failedPublicKeys = preparation.failedPublicKeys.toMutableSet()
firstError = firstError ?: preparation.firstError

@jvsena42 jvsena42 Jul 31, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

firstError precedence hides the real failure: because InvalidPublicKey is seeded above, this ?: discards the actual SDK error the batch hit. Callers log and surface only the first one, so diagnostics for a genuinely failing cleanup get replaced by "Contact public key is invalid." Prefer surfacing the SDK failure and tracking the invalid-key case separately.


if (preparation.clearedRetryKeys.isNotEmpty()) {
drainPendingPrivateMessages(
reason = "private endpoint cleanup",
advancingLinksFor = preparation.clearedRetryKeys,
)
if (report.failedToQueue.isNotEmpty() || report.failedToDeliver.isNotEmpty()) {
throw PrivatePaykitError.PrivateUnavailable
val pendingRetryKeys = pendingPrivateMessageDrainKeys(preparation.clearedRetryKeys)

@jvsena42 jvsena42 Jul 31, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Blast radius of a transient drain-check failure grew from one contact to the whole batch — same as iOS #638.

pendingPrivateMessageDrainKeys returns all passed keys from both of its getOrElse branches (lines 1030 and 1042) when linkedPeers() or pendingOutboundPrivateCounterparties() throws. With the check now run once per batch, a single transient SDK hiccup marks every successfully-cleared contact as failed, so none of them get their cache cleared or their retry marker cleared.

The retry path covers it, so not a correctness bug — but it's a real reduction in transient-failure granularity that comes with the batching, and it's the kind of thing that surfaces as "cleanup never converges" if the SDK is flaky during a full pass. Worth noting in the description alongside the perf win, and no test covers it.

if (pendingRetryKeys.isNotEmpty()) {
failedPublicKeys += pendingRetryKeys.map { it.publicKey }
firstError = firstError ?: PrivatePaykitError.PrivateUnavailable
}
val retryKey = PrivateMessageDrainRetryKey(publicKey, receiverPath)
drainPendingPrivateMessages("private endpoint cleanup", advancingLinksFor = listOf(retryKey))
if (retryKey in pendingPrivateMessageDrainKeys(listOf(retryKey))) {
}

val successfulPublicKeys = publicKeys.filterNot { it in failedPublicKeys }
clearPublishedEndpointCache(successfulPublicKeys)

firstError?.let { throw it }
Unit
}
}

private suspend fun clearPrivatePaymentLists(
publicKeys: Collection<String>,
linkedReceiverPathsSnapshot: Map<String, Set<String>>?,
): PrivateEndpointCleanupPreparation {
val failedPublicKeys = mutableSetOf<String>()
val clearedRetryKeys = mutableListOf<PrivateMessageDrainRetryKey>()
var firstError: Throwable? = null

publicKeys.forEach { publicKey ->
val contactLinkedReceiverPaths = linkedReceiverPaths(

@jvsena42 jvsena42 Jul 31, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Duplicated snapshot-or-fallback block (mirrors the iOS #638 duplication finding, just in a different spot).

This hand-rolls exactly what inspectLinkedReceiverPaths (line 1398) already does — resolve, log on failure, default to empty — differing only in the log wording ("during cleanup" vs "during '$reason'") and the extra failedPublicKeys bookkeeping. Per the repo's reuse convention, call the helper with reason = "private endpoint cleanup" and branch on LinkedReceiverPathInspection.error for the failure marking, so the two paths can't drift.

publicKey = publicKey,
snapshot = linkedReceiverPathsSnapshot,
).onFailure {
failedPublicKeys += publicKey
firstError = firstError ?: it
Logger.warn(
"Failed to inspect private Paykit links for '${redacted(publicKey)}' during cleanup",
it,
context = TAG,
)
}.getOrDefault(emptySet())
receiverPathsForCleanup(

@jvsena42 jvsena42 Jul 31, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Behavior change worth calling out — the description says "no user-facing behavior changes are intended." Same finding as iOS #638.

Previously receiverPathsForCleanup was suspend and called linkedPeers() directly, so a throw propagated out of removePublishedEndpoints(publicKey) and zero clearPrivatePaymentList calls were made for that contact. Now a link-inspection failure yields an empty contactLinkedReceiverPaths and execution falls through to here, so the contact still gets cleared for its locally-published paths (asserted by cleanup keeps pending state when linked receiver inspection fails).

This is arguably an improvement — partial progress, and the contact is marked failed either way so it retries — but it's a real change in what hits the SDK on the error path. Worth a line in the PR description.

publicKey = publicKey,
linkedReceiverPaths = contactLinkedReceiverPaths,
).forEach { receiverPath ->
runSuspendCatching {
val report = paykitSdkService.clearPrivatePaymentList(publicKey, receiverPath)
if (report.failedToQueue.isNotEmpty() || report.failedToDeliver.isNotEmpty()) {
throw PrivatePaykitError.PrivateUnavailable
}
}
if (result.isFailure) {
firstError = firstError ?: result.exceptionOrNull()
}.onSuccess {
clearedRetryKeys += PrivateMessageDrainRetryKey(publicKey, receiverPath)
}.onFailure {
failedPublicKeys += publicKey
firstError = firstError ?: it
}
}
}

firstError?.let { throw it }
return PrivateEndpointCleanupPreparation(clearedRetryKeys, failedPublicKeys, firstError)
}

private suspend fun clearPublishedEndpointCache(publicKeys: Collection<String>) {

@jvsena42 jvsena42 Jul 31, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Reentrancy window for the cache wipe is now much wider — Android analogue of the iOS actor-reentrancy finding.

serializedDispatcher is ioDispatcher.limitedParallelism(1), which serializes execution but releases the slot at every suspension point — same semantics as the iOS actor. And removePublishedEndpoints is not run under publicationMutex, while publishLocalEndpoints is (line 674).

Before this PR each contact's wipe happened immediately after that contact's own clear+drain. Now every wipe is deferred to here, after the whole batch's clears and the full drain. Scenario: cleanup runs while a foreground refreshKnownSavedContactEndpoints is in flight; contact A is cleared early in the batch, and during the many subsequent awaits the concurrent publish calls recordPublishedPrivatePaymentListCache for A (or an incoming invoice writes localInvoicesByReceiverPath); this loop then wipes remoteEndpoints / localInvoicesByReceiverPath / publishedPrivatePaymentReceiverPaths for A, desyncing local cache from what the SDK actually holds.

The hazard pre-exists, but the batching widens it from one-contact latency to whole-batch latency. Cheapest mitigations: wrap the cleanup pass in publicationMutex.withLock { } too, or snapshot the relevant contactState fields before the clear and skip the wipe if they changed.

publicKeys.forEach { publicKey ->
state?.contacts?.get(publicKey)?.let { contactState ->
contactState.remoteEndpoints = emptyList()
contactState.localInvoicesByReceiverPath = emptyMap()
Expand All @@ -1234,6 +1327,9 @@ class PrivatePaykitRepo @Inject constructor(
}
}
updateDeletedContactCleanupPending(publicKey, isPending = false)

@jvsena42 jvsena42 Jul 31, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Missed batching in a batching PR: persistState below is correctly hoisted out of the loop, but updateDeletedContactCleanupPending still runs per key — and each call is a DataStore.updateData (PrivatePaykitStores.kt:35), i.e. serialize + file write. That's N disk writes for N contacts, plausibly more expensive than the linkedPeers() call this PR batched away.

Collapse into a single update, e.g.:

cacheStore.update {
    it.copy(deletedContactCleanupPendingPublicKeys = it.deletedContactCleanupPendingPublicKeys - publicKeys.toSet())
}

}

if (publicKeys.isNotEmpty()) {
persistState(markWalletBackup = true)
}
}
Expand All @@ -1246,42 +1342,85 @@ class PrivatePaykitRepo @Inject constructor(
return paths.ifEmpty { listOf(PaykitReceiverPaths.WALLET) }
}

private suspend fun receiverPathsForPrivateEndpointCleanup(
private fun receiverPathsForPrivateEndpointCleanup(
publicKey: String,
excludedReceiverPaths: List<String>,
linkedReceiverPaths: Collection<String>,
): List<String> {
val publishedPaths = publishedPrivatePaymentReceiverPaths(publicKey)
val linkedPaths = linkedReceiverPaths(publicKey)
return (publishedPaths + linkedPaths)
return (publishedPaths + linkedReceiverPaths)
.filter { it in PaykitReceiverPaths.supported }
.filterNot { it in excludedReceiverPaths }
.distinct()
.sorted()
}

private suspend fun receiverPathsForCleanup(publicKey: String): List<String> {
val paths = (
linkedReceiverPaths(publicKey) +
publishedPrivatePaymentReceiverPaths(publicKey)
)
private fun receiverPathsForCleanup(
publicKey: String,
linkedReceiverPaths: Collection<String>,
): List<String> {
return (linkedReceiverPaths + publishedPrivatePaymentReceiverPaths(publicKey))
.filter { it in PaykitReceiverPaths.supported }
.distinct()
.sorted()
return paths
}

private suspend fun linkedReceiverPaths(publicKey: String): List<String> {
val normalizedKey = normalizedPublicKey(publicKey) ?: return emptyList()
val paths = paykitSdkService.linkedPeers()
.mapNotNull { peer ->
val peerKey = normalizedPublicKey(peer.counterparty)
peer.counterpartyReceiverPath.takeIf {
peerKey == normalizedKey && it in PaykitReceiverPaths.supported
}
private suspend fun linkedReceiverPathsByPublicKey(): Map<String, Set<String>> {
val linkedPaths = mutableMapOf<String, MutableSet<String>>()
paykitSdkService.linkedPeers().forEach { peer ->
val publicKey = normalizedPublicKey(peer.counterparty) ?: return@forEach
if (peer.counterpartyReceiverPath in PaykitReceiverPaths.supported) {
linkedPaths.getOrPut(publicKey, ::mutableSetOf) += peer.counterpartyReceiverPath
}
.distinct()
.sorted()
return paths
}
return linkedPaths
}

private suspend fun linkedReceiverPathsSnapshotOrNull(reason: String): Map<String, Set<String>>? =
runSuspendCatching { linkedReceiverPathsByPublicKey() }
.onFailure {
Logger.warn(
"Failed to inspect private Paykit links during '$reason'; retrying per contact",
it,
context = TAG,
)
}.getOrNull()

private suspend fun linkedReceiverPaths(
publicKey: String,
snapshot: Map<String, Set<String>>?,
): Result<Set<String>> = if (snapshot != null) {
Result.success(snapshot[publicKey].orEmpty())
} else {
runSuspendCatching { linkedReceiverPathsByPublicKey()[publicKey].orEmpty() }

@jvsena42 jvsena42 Jul 31, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This fallback re-issues the exact call that just failed — same finding as iOS #638.

linkedReceiverPathsByPublicKey() is the same paykitSdkService.linkedPeers() that already threw when linkedReceiverPathsSnapshotOrNull took the snapshot. So on snapshot failure the pass makes N+1 linkedPeers() calls instead of the pre-PR N — a regression in exactly the dimension this PR optimizes. The new cleanup falls back per contact when shared linked receiver inspection fails test encodes it (linkedPeerReads <= 2 = snapshot + contact 1's re-fetch).

Suggest picking one: on snapshot failure either fail the pass once (set firstError, mark all keys failed, return) or continue with an empty snapshot — and drop the per-contact re-fetch entirely. The "; retrying per contact" wording in the warning above also oversells what the retry can accomplish. Applies equally to preparePrivatePaymentListReservations via inspectLinkedReceiverPaths.

}

private suspend fun inspectLinkedReceiverPaths(
publicKey: String,
snapshot: Map<String, Set<String>>?,
reason: String,
): LinkedReceiverPathInspection {
val result = linkedReceiverPaths(publicKey, snapshot)
val error = result.exceptionOrNull()
if (error != null) {
Logger.warn(
"Failed to inspect private Paykit links for '${redacted(publicKey)}' during '$reason'",
error,
context = TAG,
)
}
return LinkedReceiverPathInspection(result.getOrDefault(emptySet()), error)
}

private fun normalizedPublicKeyBatch(publicKeys: Collection<String>): NormalizedPublicKeyBatch {
var hadInvalidKey = false
val normalizedKeys = publicKeys.mapNotNull { publicKey ->
normalizedPublicKey(publicKey) ?: run {
hadInvalidKey = true
null
}
}.distinct()
return NormalizedPublicKeyBatch(normalizedKeys, hadInvalidKey)
}

private fun publishedPrivatePaymentReceiverPaths(publicKey: String): List<String> {
Expand Down
Loading
Loading