Skip to content

fix(swift-sdk): make a changeset round linear without giving up atomicity #4595

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
romchornyi wants to merge 2 commits into
base: v4.2-dev
Choose a base branch
from
+123 −49
Open

fix(swift-sdk): make a changeset round linear without giving up atomicity #4595

Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
c1c0514
fix(swift-sdk): save a changeset round in bounded batches, certify th…
jeanpierreroma Sep 4, 2026
590dfb5
fix(swift-sdk): keep the changeset round atomic — bound lookups with …
jeanpierreroma Sep 4, 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
172 changes: 123 additions & 49 deletions ...ages/swift-sdk/Sources/SwiftDashSDK/PlatformWallet/PlatformWalletPersistenceHandler.swift
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -146,6 +146,103 @@ public final class PlatformWalletPersistenceHandler: @unchecked Sendable {
/// atomically.
private var inChangeset = false

// MARK: Round-scoped row registry
//
// A changeset round is one SwiftData transaction: every row it touches
// stays unsaved until `endChangeset` (the `atomicChangesets` contract —
// a failed round rolls back as a unit, which invitation creation relies
// on). The price used to be quadratic: each per-row lookup by txid /
// outpoint ran a `fetch()` whose predicate SwiftData evaluates against
// the context's pending changes too, hashing the whole unsaved set every
// time. On a mixing-heavy wallet the catch-up round after a backward
// re-walk (thousands of TXO/transaction upserts) turned into 10+ minutes
// of compute on the persistence queue and read as a hang; every sample
// sat in the pending-merge under `upsertUtxo`.
//
// Instead, while a round is open, lookups go through the registry
// below: rows this round has fetched or created are served from these
// dictionaries, and a miss falls through to a STORE-ONLY fetch
// (`includePendingChanges = false`), which skips the pending-merge and
// is an index lookup. Correctness rests on one rule — every row of
// these entities created inside a round is registered at creation, so
// a store-only miss never means "not created yet this round". Rows
// deleted this round are filtered out (`isDeleted`), since the store
// still has them until the save. Outside a round the helpers behave
// exactly like the plain fetches they replaced.
//
// Keyed by the immutable row identity (txid / outpoint), which is why
// predicate evaluation against store values is safe.
private var roundTransactions: [Data: PersistentTransaction] = [:]
private var roundTxos: [Data: PersistentTxo] = [:]
private var roundPendingInputs: [Data: [PersistentPendingInput]] = [:]

private func resetRoundRegistry() {
roundTransactions.removeAll(keepingCapacity: true)
roundTxos.removeAll(keepingCapacity: true)
roundPendingInputs.removeAll(keepingCapacity: true)
}

/// Fetch with the pending-merge skipped while a round is open. Only
/// valid for lookups whose in-round rows are tracked by the registry.
private func fetchStoreOnlyInRound<T: PersistentModel>(_ descriptor: FetchDescriptor<T>) -> [T] {
var descriptor = descriptor
descriptor.includePendingChanges = !inChangeset
let rows = (try? backgroundContext.fetch(descriptor)) ?? []
return inChangeset ? rows.filter { !$0.isDeleted } : rows
}

private func lookupTransaction(txid: Data) -> PersistentTransaction? {
if inChangeset, let hit = roundTransactions[txid] {
return hit.isDeleted ? nil : hit
}
var descriptor = FetchDescriptor<PersistentTransaction>(
predicate: #Predicate { $0.txid == txid }
)
descriptor.fetchLimit = 1
guard let row = fetchStoreOnlyInRound(descriptor).first else { return nil }
if inChangeset { roundTransactions[txid] = row }
return row
}

private func registerRoundTransaction(_ row: PersistentTransaction) {
if inChangeset { roundTransactions[row.txid] = row }
}

private func lookupTxo(outpoint: Data) -> PersistentTxo? {
if inChangeset, let hit = roundTxos[outpoint] {
return hit.isDeleted ? nil : hit
}
var descriptor = FetchDescriptor<PersistentTxo>(
predicate: #Predicate { $0.outpoint == outpoint }
)
descriptor.fetchLimit = 1
guard let row = fetchStoreOnlyInRound(descriptor).first else { return nil }
if inChangeset { roundTxos[outpoint] = row }
return row
}

private func registerRoundTxo(_ row: PersistentTxo) {
if inChangeset { roundTxos[row.outpoint] = row }
}

/// All pending-input rows for `outpoint`: the store's (minus in-round
/// deletions) plus the ones created this round.
private func lookupPendingInputs(outpoint: Data) -> [PersistentPendingInput] {
let descriptor = FetchDescriptor<PersistentPendingInput>(
predicate: #Predicate { $0.outpoint == outpoint }
)
let stored = fetchStoreOnlyInRound(descriptor)
guard inChangeset else { return stored }
let created = (roundPendingInputs[outpoint] ?? []).filter { !$0.isDeleted }
// A row created this round is unsaved, so it cannot also come back
// from the store-only fetch; no dedupe needed.
return stored + created
}

private func registerRoundPendingInput(_ row: PersistentPendingInput) {
if inChangeset { roundPendingInputs[row.outpoint, default: []].append(row) }
}

/// Breadcrumb backfills that arrived on the serial queue while a
/// changeset round was open. The backfill both mutates
/// `backgroundContext` and saves it, so running it mid-round would
Expand Down Expand Up @@ -1234,9 +1331,6 @@ public final class PlatformWalletPersistenceHandler: @unchecked Sendable {
//
let resolvedWalletId: Data = account.wallet.walletId
let txidData = hashData(tx.txid)
let descriptor = FetchDescriptor<PersistentTransaction>(
predicate: #Predicate { $0.txid == txidData }
)

// The FFI projection always serializes the transaction body
// (`dashcore::consensus::encode::serialize` upstream), so
Expand All @@ -1259,7 +1353,7 @@ public final class PlatformWalletPersistenceHandler: @unchecked Sendable {
tx.first_seen != 0 ? tx.first_seen : UInt64(Date().timeIntervalSince1970)

let record: PersistentTransaction
if let existing = try? backgroundContext.fetch(descriptor).first {
if let existing = lookupTransaction(txid: txidData) {
record = existing
} else {
record = PersistentTransaction(
Expand All @@ -1273,6 +1367,7 @@ public final class PlatformWalletPersistenceHandler: @unchecked Sendable {
firstSeen: firstSeen
)
backgroundContext.insert(record)
registerRoundTransaction(record)
}

record.context = tx.context
Expand Down Expand Up @@ -1389,10 +1484,7 @@ public final class PlatformWalletPersistenceHandler: @unchecked Sendable {
spendingTxid: Data,
walletId: Data
) {
let txoDescriptor = FetchDescriptor<PersistentTxo>(
predicate: #Predicate { $0.outpoint == outpoint }
)
if let txo = try? backgroundContext.fetch(txoDescriptor).first {
if let txo = lookupTxo(outpoint: outpoint) {
// Flag and link move together — see
// `reconcileSpendObservation` for the finality rule.
let verdict = Self.reconcileSpendObservation(
Expand Down Expand Up @@ -1432,10 +1524,9 @@ public final class PlatformWalletPersistenceHandler: @unchecked Sendable {
// of the same transaction would otherwise produce
// duplicate pending rows that all resolve to the same
// TXO, wasting fetch work on the resolve side.
let pendingDescriptor = FetchDescriptor<PersistentPendingInput>(
predicate: #Predicate { $0.outpoint == outpoint && $0.spendingTxid == spendingTxid }
)
if (try? backgroundContext.fetch(pendingDescriptor).first) == nil {
let alreadyPending = lookupPendingInputs(outpoint: outpoint)
.contains { $0.spendingTxid == spendingTxid }
if !alreadyPending {
let pending = PersistentPendingInput(
outpoint: outpoint,
inputIndex: inputIndex,
Expand All @@ -1444,6 +1535,7 @@ public final class PlatformWalletPersistenceHandler: @unchecked Sendable {
walletId: walletId
)
backgroundContext.insert(pending)
registerRoundPendingInput(pending)
}
}
}
Expand All @@ -1454,15 +1546,14 @@ public final class PlatformWalletPersistenceHandler: @unchecked Sendable {
/// `upsertUtxo`'s resolve path so a freshly-arrived TXO doesn't
/// keep its corresponding pending row alive.
private func removePendingInputs(for outpoint: Data) {
let descriptor = FetchDescriptor<PersistentPendingInput>(
predicate: #Predicate { $0.outpoint == outpoint }
)
guard let rows = try? backgroundContext.fetch(descriptor), !rows.isEmpty else {
return
}
let rows = lookupPendingInputs(outpoint: outpoint)
guard !rows.isEmpty else { return }
for row in rows {
backgroundContext.delete(row)
}
// Deleted rows drop out via `isDeleted`; drop the bucket so the
// next lookup for this outpoint doesn't rescan them.
roundPendingInputs[outpoint] = nil
}

private func upsertUtxo(account: PersistentAccount, utxo: UtxoEntryFFI) {
Expand All @@ -1473,11 +1564,8 @@ public final class PlatformWalletPersistenceHandler: @unchecked Sendable {

let txidData = hashData(utxo.outpoint.txid)
let outpoint = PersistentTxo.makeOutpoint(txid: txidData, vout: utxo.outpoint.vout)
let descriptor = FetchDescriptor<PersistentTxo>(
predicate: #Predicate { $0.outpoint == outpoint }
)
let record: PersistentTxo
if let existing = try? backgroundContext.fetch(descriptor).first {
if let existing = lookupTxo(outpoint: outpoint) {
record = existing
// Backfill if the account or wallet linkage is missing —
// the per-wallet query path filters on TXO.walletId, so
Expand All @@ -1496,11 +1584,8 @@ public final class PlatformWalletPersistenceHandler: @unchecked Sendable {
// arrives. Note we no longer set `parentTx.account` —
// transactions don't carry account linkage anymore (they
// can span multiple accounts).
let txDescriptor = FetchDescriptor<PersistentTransaction>(
predicate: #Predicate { $0.txid == txidData }
)
let parentTx: PersistentTransaction
if let existingTx = try? backgroundContext.fetch(txDescriptor).first {
if let existingTx = lookupTransaction(txid: txidData) {
parentTx = existingTx
} else {
// Stub row — `transactionData` is left as empty
Expand All @@ -1512,6 +1597,7 @@ public final class PlatformWalletPersistenceHandler: @unchecked Sendable {
// treats as miss.
parentTx = PersistentTransaction(txid: txidData, transactionData: Data())
backgroundContext.insert(parentTx)
registerRoundTransaction(parentTx)
}

let script: Data = {
Expand All @@ -1530,6 +1616,7 @@ public final class PlatformWalletPersistenceHandler: @unchecked Sendable {
record.account = account
record.walletId = resolvedWalletId
backgroundContext.insert(record)
registerRoundTxo(record)
}

record.amount = utxo.amount
Expand Down Expand Up @@ -1565,12 +1652,8 @@ public final class PlatformWalletPersistenceHandler: @unchecked Sendable {
// `upsertTransaction`, so the spend signal is order-
// independent at this layer regardless of which side arrives
// first.
let outpointKey = record.outpoint
let pendingDescriptor = FetchDescriptor<PersistentPendingInput>(
predicate: #Predicate { $0.outpoint == outpointKey }
)
if let pendingRows = try? backgroundContext.fetch(pendingDescriptor),
!pendingRows.isEmpty {
let pendingRows = lookupPendingInputs(outpoint: record.outpoint)
if !pendingRows.isEmpty {
// Reconcile EVERY deferred observation, not just the newest —
// the rows are about to be deleted, and picking one would let
// a mempool competitor recorded after a confirmed spender
Expand All @@ -1587,11 +1670,7 @@ public final class PlatformWalletPersistenceHandler: @unchecked Sendable {
if let spending = pending.spendingTransaction {
resolvedSpending = spending
} else {
let spendingTxid = pending.spendingTxid
let txDescriptor = FetchDescriptor<PersistentTransaction>(
predicate: #Predicate { $0.txid == spendingTxid }
)
resolvedSpending = try? backgroundContext.fetch(txDescriptor).first
resolvedSpending = lookupTransaction(txid: pending.spendingTxid)
}
guard let spending = resolvedSpending else { continue }
// Flag and link move together — see
Expand Down Expand Up @@ -1668,10 +1747,7 @@ public final class PlatformWalletPersistenceHandler: @unchecked Sendable {
txid: hashData(entry.outpoint.txid),
vout: entry.outpoint.vout
)
let descriptor = FetchDescriptor<PersistentTxo>(
predicate: #Predicate { $0.outpoint == outpoint }
)
guard let txo = try? backgroundContext.fetch(descriptor).first else {
guard let txo = lookupTxo(outpoint: outpoint) else {
return
}
// Link the spending transaction. The FFI now carries
Expand All @@ -1689,10 +1765,7 @@ public final class PlatformWalletPersistenceHandler: @unchecked Sendable {
if txo.spendingTransaction?.txid == spendingTxid {
spendingTx = txo.spendingTransaction
} else {
let txDescriptor = FetchDescriptor<PersistentTransaction>(
predicate: #Predicate { $0.txid == spendingTxid }
)
spendingTx = try? backgroundContext.fetch(txDescriptor).first
spendingTx = lookupTransaction(txid: spendingTxid)
}
}
// When the spending tx isn't resolved this flush, leave the row
Expand Down Expand Up @@ -1728,10 +1801,7 @@ public final class PlatformWalletPersistenceHandler: @unchecked Sendable {

private func markUtxoInstantLocked(_ op: OutPointFFI) {
let outpoint = PersistentTxo.makeOutpoint(txid: hashData(op.txid), vout: op.vout)
let descriptor = FetchDescriptor<PersistentTxo>(
predicate: #Predicate { $0.outpoint == outpoint }
)
if let txo = try? backgroundContext.fetch(descriptor).first {
if let txo = lookupTxo(outpoint: outpoint) {
txo.isInstantLocked = true
txo.lastUpdated = Date()
}
Expand Down Expand Up @@ -1860,6 +1930,7 @@ public final class PlatformWalletPersistenceHandler: @unchecked Sendable {
func beginChangeset(walletId: Data) {
onQueue {
self.inChangeset = true
self.resetRoundRegistry()
SDKLogger.event(
"persistence_changeset_started",
category: .persistence,
Expand Down Expand Up @@ -1893,6 +1964,9 @@ public final class PlatformWalletPersistenceHandler: @unchecked Sendable {
// (clear, then drain) is load-bearing.
defer {
self.inChangeset = false
// The registry only ever held rows of this round's context;
// after commit or rollback they are either saved or gone.
self.resetRoundRegistry()
self.drainDeferredBackfills()
}
if success {
Expand Down
Loading