Skip to content

refactor: batch paykit contact cleanup #638

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
Choose a base branch
from
+110 −58
Open

refactor: batch paykit contact cleanup #638

Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
cb633e5
refactor: batch paykit contact cleanup
ben-kaufman Jul 22, 2026
0dd87ec
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
168 changes: 110 additions & 58 deletions Bitkit/Services/PrivatePaykitService+Contacts.swift
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -87,54 +87,86 @@ extension PrivatePaykitService {
}

func removePublishedEndpoints(for publicKeys: [String]) async throws {
let publicKeys = normalizedSavedContactKeys(publicKeys)
guard !publicKeys.isEmpty else { return }

let linkedReceiverPathsSnapshot: [String: Set<String>]?
do {
linkedReceiverPathsSnapshot = try await linkedReceiverPathsByPublicKey()
} catch {
linkedReceiverPathsSnapshot = nil
Logger.warn(
"Failed to inspect private Paykit links for cleanup; retrying per contact: \(error)",
context: "PrivatePaykit"
)
}

var firstError: Error?
var failedPublicKeys = Set<String>()
var clearedRetryKeys = [PrivateMessageDrainRetryKey]()
var didChangeState = false
for publicKey in normalizedSavedContactKeys(publicKeys) {
var didFail = false
let cleanupReceiverPaths: [String]
do {
cleanupReceiverPaths = try await receiverPathsForCleanup(publicKey: publicKey)
} catch {
firstError = firstError ?? error
Self.markDeletedContactCleanupPending([publicKey])
continue
for publicKey in publicKeys {
let contactLinkedReceiverPaths: Set<String>
if let linkedReceiverPathsSnapshot {
contactLinkedReceiverPaths = linkedReceiverPathsSnapshot[publicKey, default: []]
} else {
do {
contactLinkedReceiverPaths = try await linkedReceiverPaths(publicKey: 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.

This fallback re-issues the exact call that just failed.

linkedReceiverPaths(publicKey:) now calls linkedReceiverPathsByPublicKey(), which is the same PaykitSdkService.shared.linkedPeers() that just threw when the snapshot was taken. So when the snapshot fails, this pass makes N+1 linkedPeers() calls instead of the pre-PR N — a regression in exactly the dimension the PR optimizes — and the "retrying per contact" wording in the warning above oversells what the retry can accomplish.

Suggest picking one: on snapshot failure either fail the pass once (set firstError, mark all keys pending, return) or continue with an empty snapshot, and drop the per-contact re-fetch entirely. Same applies to the mirrored block in syncLocalEndpointPublicationLocked.

} catch {
contactLinkedReceiverPaths = []
failedPublicKeys.insert(publicKey)
firstError = firstError ?? error
Logger.warn(
"Failed to inspect private Paykit links for \(PubkyPublicKeyFormat.redacted(publicKey)) during cleanup: \(error)",
context: "PrivatePaykit"
)
}
}

let cleanupReceiverPaths = 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").

Previously, when receiverPathsForCleanup threw for a contact, the loop continued and made zero clear attempts for it. Now a link-inspection failure yields an empty contactLinkedReceiverPaths and execution falls through to here, so the contact still gets clearPrivatePaymentList called for its locally-published paths.

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. Probably worth a line in the PR description.

publicKey: publicKey,
linkedReceiverPaths: contactLinkedReceiverPaths
)
for receiverPath in cleanupReceiverPaths {
do {
let report = try await PaykitSdkService.shared.clearPrivatePaymentList(to: publicKey, receiverPath: receiverPath)
if !report.failedToQueue.isEmpty || !report.failedToDeliver.isEmpty {
throw PrivatePaykitError.privateUnavailable
}
let retryKey = PrivateMessageDrainRetryKey(publicKey: publicKey, receiverPath: receiverPath)
await drainPendingPrivateMessages(reason: "cleanup", advancing: [retryKey])
if await pendingPrivateMessageDrainKeys([retryKey]).contains(retryKey) {
throw PrivatePaykitError.privateUnavailable
}
clearedRetryKeys.append(PrivateMessageDrainRetryKey(publicKey: publicKey, receiverPath: receiverPath))
} catch {
didFail = true
failedPublicKeys.insert(publicKey)
firstError = firstError ?? error
Self.markDeletedContactCleanupPending([publicKey])
}
}
}

if !didFail {
if var contactState = state.contacts[publicKey] {
let hadStateToClear = !contactState.cachedResolvedEndpoints.isEmpty ||
!contactState.localInvoicesByReceiverPath.isEmpty ||
!contactState.publishedPrivatePaymentReceiverPaths.isEmpty
contactState.cachedResolvedEndpoints = []
contactState.localInvoicesByReceiverPath = [:]
contactState.publishedPrivatePaymentReceiverPaths = []
let shouldRemoveContact = !contactState.hasCacheState
state.contacts[publicKey] = shouldRemoveContact ? nil : contactState
didChangeState = didChangeState || hadStateToClear || shouldRemoveContact
}
if !clearedRetryKeys.isEmpty {
await drainPendingPrivateMessages(reason: "cleanup", advancing: clearedRetryKeys)
let pendingRetryKeys = await pendingPrivateMessageDrainKeys(clearedRetryKeys)
if !pendingRetryKeys.isEmpty {

@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.

pendingPrivateMessageDrainKeys returns all the passed keys when linkedPeers() or pendingOutboundPrivateCounterparties() throws (see its two catch blocks). 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 state cleared or their retry marker cleared.

The retry path covers it, so this isn't 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 shows up as "cleanup never converges" if the SDK is flaky during a full cleanup pass. Worth noting in the description alongside the perf win.

failedPublicKeys.formUnion(pendingRetryKeys.map(\.publicKey))
firstError = firstError ?? PrivatePaykitError.privateUnavailable
}
}

Self.clearDeletedContactCleanupPending([publicKey])
let successfulPublicKeys = Set(publicKeys).subtracting(failedPublicKeys)

@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.

Minor: publicKeys is already deduplicated by normalizedSavedContactKeys at the top of the function, so this Set(publicKeys) is rebuilding a set that could be constructed once above the loop and reused (it's also implicitly the membership source for the loop). Purely cosmetic.

for publicKey in successfulPublicKeys {

@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.

Actor-reentrancy window for the state wipe is now much wider.

PrivatePaykitService is an actor (reentrant), and removePublishedEndpoints is not run under withPublicationLock — only syncLocalEndpointPublication is. Before this PR each contact's state.contacts[X] 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: removePublishedEndpoints() (the no-arg variant, which unions all known + cached + pending keys) runs while a foreground refreshKnownSavedContactEndpoints is in flight. Contact A is cleared early in the batch; during the many subsequent awaits the concurrent publish calls recordPublishedPrivatePaymentList for A (or an incoming invoice writes localInvoicesByReceiverPath); this loop then wipes cachedResolvedEndpoints / localInvoicesByReceiverPath / publishedPrivatePaymentReceiverPaths for A, desyncing local state 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: snapshot the relevant state.contacts[publicKey] fields before the clear and skip the wipe if they changed, or run the cleanup pass under withPublicationLock as well.

if var contactState = state.contacts[publicKey] {
let hadStateToClear = !contactState.cachedResolvedEndpoints.isEmpty ||
!contactState.localInvoicesByReceiverPath.isEmpty ||
!contactState.publishedPrivatePaymentReceiverPaths.isEmpty
contactState.cachedResolvedEndpoints = []
contactState.localInvoicesByReceiverPath = [:]
contactState.publishedPrivatePaymentReceiverPaths = []
let shouldRemoveContact = !contactState.hasCacheState
state.contacts[publicKey] = shouldRemoveContact ? nil : contactState
didChangeState = didChangeState || hadStateToClear || shouldRemoveContact
}
}

Self.markDeletedContactCleanupPending(Array(failedPublicKeys))

@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.

Failure markers are now written once at the end instead of eagerly.

The old code called markDeletedContactCleanupPending([publicKey]) at the moment each contact failed, so a crash mid-loop still left the marker persisted. Now nothing is written until the batch completes.

Mostly recoverable, since removePublishedEndpoints() and pruneUnsavedContactState re-derive keys from state.contacts (which isn't cleared for failed contacts). The narrow gap is a contact deleted via removeSavedContact that has no local cache state — its only record would be this marker, and a crash before this line loses it. Low severity, but it is a durability regression relative to the eager write.

Self.clearDeletedContactCleanupPending(Array(successfulPublicKeys))

if didChangeState {
persistState(markWalletBackup: true)
}
Expand Down Expand Up @@ -265,6 +297,17 @@ extension PrivatePaykitService {
var firstError: Error?
var updates = [PrivatePaymentListReservationUpdateInput]()
var linkRetryKeys = [PrivateMessageDrainRetryKey]()
let linkedReceiverPathsSnapshot: [String: Set<String>]?
do {
linkedReceiverPathsSnapshot = try await linkedReceiverPathsByPublicKey()
} catch {
linkedReceiverPathsSnapshot = nil
Logger.warn(
"Failed to inspect private Paykit links during \(reason); retrying per contact: \(error)",
context: "PrivatePaykit"
)
}

for publicKey in publicKeys {
let receiverPaths: [String]
do {
Expand Down Expand Up @@ -295,20 +338,26 @@ extension PrivatePaykitService {
context: "PrivatePaykit"
)
}
let cleanupReceiverPaths: [String]
do {
cleanupReceiverPaths = try await receiverPathsForPrivateEndpointCleanup(
publicKey: publicKey,
excluding: publicationReceiverPaths + receiverPathSelection.cleanupProtectedReceiverPaths
)
} catch {
cleanupReceiverPaths = []
firstError = firstError ?? error
Logger.warn(
"Failed to inspect private Paykit links for \(PubkyPublicKeyFormat.redacted(publicKey)) during \(reason): \(error)",
context: "PrivatePaykit"
)
let contactLinkedReceiverPaths: Set<String>
if let linkedReceiverPathsSnapshot {
contactLinkedReceiverPaths = linkedReceiverPathsSnapshot[publicKey, default: []]
} else {
do {
contactLinkedReceiverPaths = try await linkedReceiverPaths(publicKey: 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.

Duplicated snapshot-or-fallback block.

This is the same ~14 lines as the block in removePublishedEndpoints (lines 109–124), differing only in the log's reason. Per the repo's reuse convention, worth extracting into a single helper — e.g. linkedReceiverPaths(for publicKey: String, snapshot: [String: Set<String>]?, reason: String) async -> (paths: Set<String>, error: Error?) — so the two call sites can't drift.

(Same redundant-refetch caveat as the other block applies here too.)

} catch {
contactLinkedReceiverPaths = []
firstError = firstError ?? error
Logger.warn(
"Failed to inspect private Paykit links for \(PubkyPublicKeyFormat.redacted(publicKey)) during \(reason): \(error)",
context: "PrivatePaykit"
)
}
}
let cleanupReceiverPaths = receiverPathsForPrivateEndpointCleanup(
publicKey: publicKey,
excluding: publicationReceiverPaths + receiverPathSelection.cleanupProtectedReceiverPaths,
linkedReceiverPaths: contactLinkedReceiverPaths
)

for receiverPath in Set(linkableReceiverPaths).union(cleanupReceiverPaths) {
linkRetryKeys.append(PrivateMessageDrainRetryKey(publicKey: publicKey, receiverPath: receiverPath))
Expand Down Expand Up @@ -652,35 +701,38 @@ extension PrivatePaykitService {

private func receiverPathsForPrivateEndpointCleanup(
publicKey: String,
excluding publicationReceiverPaths: [String]
) async throws -> [String] {
excluding publicationReceiverPaths: [String],
linkedReceiverPaths: Set<String>
) -> [String] {
let publishedPaths = publishedPrivatePaymentReceiverPaths(publicKey: publicKey)
let linkedPaths = try await linkedReceiverPaths(publicKey: publicKey)
let excluded = Set(publicationReceiverPaths)
return Array(Set(publishedPaths).union(linkedPaths).subtracting(excluded))
return Array(Set(publishedPaths).union(linkedReceiverPaths).subtracting(excluded))
.filter { PaykitReceiverPath.supported.contains($0) }
.sorted()
}

private func receiverPathsForCleanup(publicKey: String) async throws -> [String] {
let linkedPaths = try await linkedReceiverPaths(publicKey: publicKey)
private func receiverPathsForCleanup(publicKey: String, linkedReceiverPaths: Set<String>) -> [String] {
let publishedPaths = publishedPrivatePaymentReceiverPaths(publicKey: publicKey)
return Array(Set(linkedPaths).union(publishedPaths))
return Array(linkedReceiverPaths.union(publishedPaths))
.filter { PaykitReceiverPath.supported.contains($0) }
.sorted()
}

private func linkedReceiverPaths(publicKey: String) async throws -> [String] {
guard let publicKey = PubkyPublicKeyFormat.normalized(publicKey) else { return [] }
private func linkedReceiverPathsByPublicKey() async throws -> [String: Set<String>] {
let peers = try await PaykitSdkService.shared.linkedPeers()

let linkedPaths = peers.compactMap { peer -> String? in
guard PubkyPublicKeyFormat.normalized(peer.counterparty) == publicKey,
var linkedPaths = [String: Set<String>]()
for peer in peers {
guard let publicKey = PubkyPublicKeyFormat.normalized(peer.counterparty),
PaykitReceiverPath.supported.contains(peer.counterpartyReceiverPath)
else { return nil }
return peer.counterpartyReceiverPath
else { continue }
linkedPaths[publicKey, default: []].insert(peer.counterpartyReceiverPath)
}
return Array(Set(linkedPaths)).sorted()
return linkedPaths
}

private func linkedReceiverPaths(publicKey: String) async throws -> Set<String> {
guard let publicKey = PubkyPublicKeyFormat.normalized(publicKey) else { return [] }
return try await linkedReceiverPathsByPublicKey()[publicKey, default: []]
}

private func publishedPrivatePaymentReceiverPaths(publicKey: String) -> [String] {
Expand Down
Loading