diff --git a/Cargo.lock b/Cargo.lock index 05bd8b6e1b6..fdbe964be67 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -6509,11 +6509,14 @@ dependencies = [ "dash-network", "dashcore", "dashpay-contract", + "hex", "jni 0.21.1", "key-wallet-ffi", "log", "platform-wallet-ffi", "rs-sdk-ffi", + "serde", + "serde_json", "zeroize", ] diff --git a/packages/kotlin-sdk/sdk/src/main/kotlin/org/dashfoundation/dashsdk/ffi/WalletManagerNative.kt b/packages/kotlin-sdk/sdk/src/main/kotlin/org/dashfoundation/dashsdk/ffi/WalletManagerNative.kt index 090bca5a5a1..8d9409700bd 100644 --- a/packages/kotlin-sdk/sdk/src/main/kotlin/org/dashfoundation/dashsdk/ffi/WalletManagerNative.kt +++ b/packages/kotlin-sdk/sdk/src/main/kotlin/org/dashfoundation/dashsdk/ffi/WalletManagerNative.kt @@ -145,6 +145,75 @@ internal object WalletManagerNative { gapLimit: Int, ) + /** + * One bounded page of the engine's UTXO inventory for one wallet, + * across every funds account, as JSON + * `{"utxos":[...],"cursor":,"hasMore":}` — the + * source of truth the TXO-store reconciler + * ([PlatformWalletManager.reconcileTxoStore]) diffs against the Room + * `txos` mirror. A serialization shim over one Rust call + * (`platform_wallet_wallet_utxos_page`): account ordering, cursor + * semantics, the page bound and the address rendering all live in + * `platform-wallet`, so the Swift host walks the identical inventory. + * + * Paged, not swept whole: a wallet's UTXO count is chain-controlled + * (anyone who knows a watched address can keep sending dust to it), so + * a full-inventory read would let a remote party decide how much this + * process allocates on every SYNCED transition and every 30-minute + * pass. Pass [cursor] `null` to start, then hand back the returned + * `cursor` verbatim while `hasMore` is true — a cursor this export did + * not produce throws. [limit] caps the rows in one page; non-positive + * asks for the engine's own default, and the engine clamps its own + * maximum, so no bound is mirrored on this side. + * + * Contact watch-only chains (`DashpayExternalAccount`) are absent by + * construction — those coins are the contact's money, not this + * wallet's — so the inventory never offers one to heal. + * + * Each `utxos` row is a + * [org.dashfoundation.dashsdk.persistence.PlatformWalletPersistenceHandler.EngineUtxoRow]: + * the owning account tuple, the txid hex in the same byte order the + * changeset path hands the handler (so hex→bytes reproduces the + * `txos.txid` blob), vout, amount (duffs), the engine's own + * Base58Check address (empty when the script has no address form), + * scriptHex, height, and the engine's own `isConfirmed` / + * `isInstantLocked` / `isCoinbase` / `isLocked` flags. + */ + external fun walletManagerUtxosPageJson( + managerHandle: Long, + walletId: ByteArray, + cursor: String?, + limit: Int, + ): String? + + /** + * Classify a batch of store rows against the engine's live state: the + * reverse half of the reconcile transport, and the reason the paged + * inventory above carries no spent-outpoint list. The caller pages its + * OWN mirror rows and asks about them a batch at a time, so neither + * side ever builds a set over the whole engine inventory. + * + * [queriesJson] is a JSON array of + * [org.dashfoundation.dashsdk.persistence.PlatformWalletPersistenceHandler.OutpointQuery]: + * for each row, the account tuple the STORE files the coin under, its + * txid hex, vout, and the scriptHex the store recorded. The account + * and script are the store's claim — the engine checks both against + * its own pools rather than trusting them, which is why a bare + * outpoint is not enough to ask the question. + * + * Returns one verdict byte per query, positionally — see + * `OUTPOINT_CLASS_*` on + * [org.dashfoundation.dashsdk.persistence.PlatformWalletPersistenceHandler]: + * 0 unknown, 1 unspent, 2 known-uncredited, 3 not-owned. A query whose + * account tag this build cannot map keeps `unknown`; the rest are + * still answered. + */ + external fun walletManagerClassifyOutpoints( + managerHandle: Long, + walletId: ByteArray, + queriesJson: String, + ): ByteArray? + // ── Core transaction builder (1:1 over `core_wallet_tx_builder_*`) ─ // // Each step is a thin extern (one export = one FFI call, per diff --git a/packages/kotlin-sdk/sdk/src/main/kotlin/org/dashfoundation/dashsdk/persistence/PlatformWalletPersistenceHandler.kt b/packages/kotlin-sdk/sdk/src/main/kotlin/org/dashfoundation/dashsdk/persistence/PlatformWalletPersistenceHandler.kt index 88a080308a0..3fd604ddc60 100644 --- a/packages/kotlin-sdk/sdk/src/main/kotlin/org/dashfoundation/dashsdk/persistence/PlatformWalletPersistenceHandler.kt +++ b/packages/kotlin-sdk/sdk/src/main/kotlin/org/dashfoundation/dashsdk/persistence/PlatformWalletPersistenceHandler.kt @@ -13,6 +13,16 @@ import kotlinx.coroutines.flow.update import kotlinx.coroutines.runBlocking import kotlinx.coroutines.sync.Mutex import kotlinx.coroutines.sync.withLock +import kotlinx.serialization.Serializable +import kotlinx.serialization.json.Json +import kotlinx.serialization.json.boolean +import kotlinx.serialization.json.booleanOrNull +import kotlinx.serialization.json.contentOrNull +import kotlinx.serialization.json.int +import kotlinx.serialization.json.jsonArray +import kotlinx.serialization.json.jsonObject +import kotlinx.serialization.json.jsonPrimitive +import kotlinx.serialization.json.long import org.dashfoundation.dashsdk.errors.DashSdkError import org.dashfoundation.dashsdk.ffi.AccountSpecData import org.dashfoundation.dashsdk.ffi.ContactProfileRestoreData @@ -1038,111 +1048,878 @@ class PlatformWalletPersistenceHandler( isLocked: Boolean, ): Int = guarded { stage(walletId) { db -> - val outpoint = makeOutpoint(txid, vout) - // Ensure a parent transaction row exists (stub if missing, so - // the TXO FK holds; the real tx upsert overwrites it later). - if (db.transactionDao().getByTxid(txid) == null) { - db.transactionDao().upsert( - TransactionEntity(txid = txid, transactionData = ByteArray(0)), + upsertUtxoRow( + db, walletId, txid, vout, amount, address, scriptPubKey, + height, isCoinbase, isConfirmed, isInstantLocked, isLocked, + ) + } + 0 + } + + /** + * The single TXO-insert discipline, shared by the changeset callback + * ([onWalletChangesetUtxoAdded]) and the reconcile sweep + * ([reconcileTxos]): stub the parent transaction row so the FK holds, + * upsert the TXO preserving any existing spend linkage, then drain + * pending-input rows staged before this funding TXO existed — a 1:1 + * port of the Swift upsertUtxo drain + * (PlatformWalletPersistenceHandler.swift). A spend that arrived first + * was deferred (see onWalletChangesetTransaction); now that the funding + * output is here, resolve the claim and clear the rows so the + * UTXO-restore path won't hand this consumed output back to Rust as + * spendable. + */ + private suspend fun upsertUtxoRow( + db: DashDatabase, + walletId: ByteArray, + txid: ByteArray, + vout: Int, + amount: Long, + address: String, + scriptPubKey: ByteArray, + height: Int, + isCoinbase: Boolean, + isConfirmed: Boolean, + isInstantLocked: Boolean, + isLocked: Boolean, + // The Room account this output belongs to, when the CALLER could + // resolve it (the reconcile resolves it from the engine inventory's + // account tags). Stamped on the row so ownership survives even when + // the address projection is absent — a heal into a store that lost + // BOTH the TXO and its address row must not produce a row the + // restore loader cannot attribute (it would be skipped at the next + // mirror-reload, recreating the fund loss the heal repaired). An + // existing row's accountId always wins; changeset callbacks pass + // null and keep their address-projection behavior. + resolvedAccountId: Long? = null, + ) { + val outpoint = makeOutpoint(txid, vout) + // Ensure a parent transaction row exists (stub if missing, so + // the TXO FK holds; the real tx upsert overwrites it later). + if (db.transactionDao().getByTxid(txid) == null) { + db.transactionDao().upsert( + TransactionEntity(txid = txid, transactionData = ByteArray(0)), + ) + } + val existing = db.txoDao().getByOutpoint(outpoint) + val coreAddressId = if (address.isNotEmpty()) address else null + // A materialised coin the wallet re-delivers unspent follows the + // wallet — the mirror of the SQLite store's upsert valve, which + // holds only never-materialised placeholders. The wallet knows + // this coin, and any network-final spender of a coin it knows is + // wallet-relevant by BIP158 prevout matching, so its own scan + // re-discovers the spend; refusing the re-delivery would lock a + // real coin out forever after a reorg of the winner, and on this + // side of the FFI a row at `isSpent = true` is never restored to + // Rust again. So an UNLINKED row — a sweep hold with its stamp, or + // a legacy flag with nothing behind it — is cleared, stamp + // included. A LINKED row keeps its flag and stamp: the link is + // this store's recorded spend attribution, the pending drain + // below and the sweep pass own that transition, and a spender + // that reached a block is confirmed evidence a re-delivery never + // displaces. + val linked = existing?.spendingTxid != null + val row = TxoEntity( + outpoint = outpoint, + vout = vout, + amount = amount, + address = address, + scriptPubKey = scriptPubKey, + height = height, + isCoinbase = isCoinbase, + isConfirmed = isConfirmed, + isInstantLocked = isInstantLocked, + isLocked = isLocked, + isSpent = linked && existing!!.isSpent, + walletId = walletId, + txid = txid, + spendingTxid = existing?.spendingTxid, + spendingInputIndex = existing?.spendingInputIndex, + accountId = existing?.accountId ?: resolvedAccountId, + coreAddressId = existing?.coreAddressId ?: coreAddressIdIfPresent(db, coreAddressId), + createdAt = existing?.createdAt ?: java.util.Date(), + lastUpdated = now(), + supersededByTxid = if (linked) existing!!.supersededByTxid else null, + ) + db.txoDao().upsert(row) + // Drain any pending-input rows staged before this funding TXO + // existed — a port of the Swift `upsertUtxo` drain + // (PlatformWalletPersistenceHandler.swift). A spend that arrived + // first was deferred (see onWalletChangesetTransaction); now that + // the funding output is here, resolve the claim and clear the rows + // so the UTXO-restore path won't hand this consumed output back to + // Rust as spendable. + val pending = db.documentDao().getPendingInputsByOutpoint(outpoint) + if (pending.isNotEmpty()) { + // A tombstone outranks every ordinary row regardless of age: + // ordinary rows are competing *observations*, a tombstone is + // the sweep's settled verdict that its winner consumed this + // coin. Prefer the tombstone tagged with the delivering + // wallet; failing that any tombstone on the outpoint still + // holds — the stamp is a txid fact, not a per-wallet one. + val tombstones = pending.filter { it.isSweptTombstone } + val tombstone = tombstones.filter { it.walletId.contentEquals(walletId) } + .maxByOrNull { it.createdAt } + ?: tombstones.maxByOrNull { it.createdAt } + if (tombstone != null) { + // A drained tombstone STAMPS, it never mints a spender + // link: the winner need not have its own `transactions` + // row, and a link would make the coin non-releasable + // (the release pass frees stamped, unlinked rows) when a + // later sweep proves the winner never took it. The + // existing link, if any, is carried as it was. + db.txoDao().upsert( + row.copy( + isSpent = true, + supersededByTxid = tombstone.spendingTxid, + lastUpdated = now(), + ), + ) + } else { + // Competing ordinary observations: a network-final spender + // outranks a newer mempool one (its row is the settled + // claim the link guard protects); among equals the newest + // wins, as before (reorg / double-spend: newest wins). + val ranked = pending.map { p -> p to db.transactionDao().getByTxid(p.spendingTxid) } + val (chosen, spending) = ranked.maxWithOrNull( + compareBy>( + { it.second?.context ?: 0 }, + { it.first.createdAt }, + ), + )!! + val spendingContext = spending?.context ?: 0 + val keepExistingLink = + keepSettledSpenderLink(db, row, chosen.spendingTxid, spendingContext) + db.txoDao().upsert( + linkSpender(row, chosen.spendingTxid, chosen.inputIndex, spendingContext, keepExistingLink), ) } - val existing = db.txoDao().getByOutpoint(outpoint) - val coreAddressId = if (address.isNotEmpty()) address else null - // A materialised coin the wallet re-delivers unspent follows the - // wallet — the mirror of the SQLite store's upsert valve, which - // holds only never-materialised placeholders. The wallet knows - // this coin, and any network-final spender of a coin it knows is - // wallet-relevant by BIP158 prevout matching, so its own scan - // re-discovers the spend; refusing the re-delivery would lock a - // real coin out forever after a reorg of the winner, and on this - // side of the FFI a row at `isSpent = true` is never restored to - // Rust again. So an UNLINKED row — a sweep hold with its stamp, or - // a legacy flag with nothing behind it — is cleared, stamp - // included. A LINKED row keeps its flag and stamp: the link is - // this store's recorded spend attribution, the pending drain - // below and the sweep pass own that transition, and a spender - // that reached a block is confirmed evidence a re-delivery never - // displaces. - val linked = existing?.spendingTxid != null - val row = TxoEntity( - outpoint = outpoint, - vout = vout, - amount = amount, - address = address, - scriptPubKey = scriptPubKey, - height = height, - isCoinbase = isCoinbase, - isConfirmed = isConfirmed, - isInstantLocked = isInstantLocked, - isLocked = isLocked, - isSpent = linked && existing!!.isSpent, - walletId = walletId, - txid = txid, - spendingTxid = existing?.spendingTxid, - spendingInputIndex = existing?.spendingInputIndex, - accountId = existing?.accountId, - coreAddressId = existing?.coreAddressId ?: coreAddressIdIfPresent(db, coreAddressId), - createdAt = existing?.createdAt ?: java.util.Date(), - lastUpdated = now(), - supersededByTxid = if (linked) existing!!.supersededByTxid else null, + for (p in pending) db.documentDao().deletePendingInput(p) + } + } + + /** + * One row of the engine's paged UTXO inventory, exactly as + * `walletManagerUtxosPageJson` serializes it (rs-unified-sdk-jni + * `UtxoPageRow`, serde on that side, kotlinx on this one). The contract + * is typed on BOTH ends so a renamed or missing key fails the decode + * loudly instead of healing a row with a defaulted owner or a zero + * amount into the mirror the engine reloads from. The account tuple + * (`typeTag`…`friendIdentityId`) is the routing context [fetchAccount] + * resolves the owning Room account by; the DashPay identity halves are + * absent on every non-DashPay account. + */ + @Serializable + data class EngineUtxoRow( + val typeTag: Int, + val standardTag: Int = 0, + val index: Int = 0, + val registrationIndex: Int = 0, + val keyClass: Int = 0, + val userIdentityId: String? = null, + val friendIdentityId: String? = null, + val txid: String, + val vout: Int, + val amount: Long, + val address: String = "", + val scriptHex: String = "", + val height: Int, + /** The engine's own flags for the coin. Stamped onto a healed row + * as given: guessing them (a coinbase row filed as non-coinbase) + * would misstate maturity in the mirror the engine reloads from. */ + val isConfirmed: Boolean = false, + val isInstantlocked: Boolean = false, + val isCoinbase: Boolean = false, + val isLocked: Boolean = false, + ) + + /** + * One store row handed to `walletManagerClassifyOutpoints`: the account + * tuple the STORE files the coin under, its outpoint, and the script + * the store recorded for it. The account and script are the store's + * claim — the engine checks both against its own pools rather than + * trusting them, which is why a bare outpoint cannot ask the question. + * Serialized against the same field names the Rust `OutpointQuery` + * deserializes. + */ + @Serializable + data class OutpointQuery( + val typeTag: Int, + val standardTag: Int = 0, + val index: Int = 0, + val registrationIndex: Int = 0, + val keyClass: Int = 0, + val userIdentityId: String? = null, + val friendIdentityId: String? = null, + val txid: String, + val vout: Int, + val scriptHex: String = "", + ) + + /** + * One page of the engine's UTXO inventory: the rows, the opaque resume + * cursor (absent on the last page) and whether more pages follow. + */ + @Serializable + data class EngineUtxoPage( + val utxos: List = emptyList(), + val cursor: String? = null, + val hasMore: Boolean = false, + ) + + /** + * Outcome of one [reconcileTxos] sweep. [inserted]/[insertedDuffs] + * are the healed holes; a non-zero value after a completed sync means + * a changeset failed to deliver an owned output (the + * CoinJoin-funded-send change-drop class) and would have become a + * fund-loss on the next engine reload from this store. + */ + data class TxoReconcileReport( + val engineUtxos: Int, + val inserted: Int, + val insertedDuffs: Long, + /** Healed TXOs whose pre-existing record's netAmount MAY be short by + * the healed amount. LOG-ONLY: the record can already carry the + * corrected net (a corrective callback racing this sweep), and + * blind addition double-credits. The event pipeline owns net + * correctness. */ + val netAmountSuspects: Int, + /** Healed rows whose owning Room account could not be resolved from + * the inventory's account tuple — ownership rides on the address + * projection alone, and if that row is also missing the healed TXO + * will not survive the next mirror-reload. */ + val healedUnowned: Int = 0, + val skippedImmature: Int, + val skippedNoAddress: Int, + /** Store rows marked unspent for which the engine holds a MINED + * record spending the outpoint — the lost-spend-update class + * (dashpay/platform#4425), on durable evidence rather than a bare + * absence. LOG-ONLY all the same: flipping spent state is the one + * direction where a reconciler bug spends a coin the user still + * owns, and this pass is insert-only by contract. */ + val wouldFlipSpent: Int = 0, + val wouldFlipSpentDuffs: Long = 0, + /** Store rows marked unspent the engine has no verdict for — + * swept/abandoned residue (pre-rust-dashcore#971 stores), a + * funding transaction this session never processed, or an engine + * gap. After a restart the engine's finalized set is empty, so + * this is the ordinary answer rather than the alarming one. + * LOG-ONLY: counted and named in the log, never removed. */ + val wouldRemove: Int = 0, + val wouldRemoveDuffs: Long = 0, + /** Store rows filed under an account whose pools do not monitor the + * row's script — the engine could never have credited the coin + * there, so this is a store attribution error rather than a + * missing coin. LOG-ONLY: counted, never re-filed. */ + val misattributed: Int = 0, + val misattributedDuffs: Long = 0, + /** Unspent store rows whose owning account could not be resolved + * from either the explicit link or the address projection. They + * carry no claim for the engine to check, so they are counted + * rather than asked about — and the same missing link is what + * would stop such a row surviving the next mirror-reload. */ + val unresolvedAccount: Int = 0, + /** Watch-only DIP-15 contact coins excluded on both sides — engine + * rows the insert pass refused to heal as ours, and unspent store + * rows the reverse pass did not classify (the engine's own accounts + * never report them, so their absence is expected, not + * divergence). */ + val skippedForeign: Int = 0, + /** Store rows marked spent for a coin the engine lists UNSPENT — + * either a released coin from a swept transaction whose release + * event a pre-rust-dashcore#971 build lost, or a live spend the + * store wrote moments before the engine settled. The two cannot + * be told apart safely, so this is LOG-ONLY: un-marking a coin + * mid-payment would let the wallet double-spend it. */ + val stuckSpent: Int = 0, + val stuckSpentDuffs: Long = 0, + /** Transport reads that failed mid-sweep — an engine inventory page + * after the first, or an outpoint-classification batch, that came + * back null. The pass stops at the first one; whatever it already + * applied stands (insert-only, idempotent) and the rest waits for + * the next cadence tick. A persistently non-zero value means the + * sweep never finishes, so the report's other counters are a + * partial view. */ + val transportFailures: Int = 0, + ) + + /** The insert pass's half of a [TxoReconcileReport]. */ + private data class HealPass( + val engineUtxos: Int, + val inserted: Int, + val insertedDuffs: Long, + val netAmountSuspects: Int, + val healedUnowned: Int, + val skippedImmature: Int, + val skippedNoAddress: Int, + val skippedForeign: Int, + val transportFailures: Int, + ) + + /** The classification pass's half of a [TxoReconcileReport]. */ + private data class ClassifyPass( + val wouldFlipSpent: Int, + val wouldFlipSpentDuffs: Long, + val wouldRemove: Int, + val wouldRemoveDuffs: Long, + val stuckSpent: Int, + val stuckSpentDuffs: Long, + val misattributed: Int, + val misattributedDuffs: Long, + val unresolvedAccount: Int, + val skippedForeign: Int, + val transportFailures: Int, + ) + + /** + * Reconcile the Room `txos` mirror against the engine's live UTXO + * inventory, healing rows a changeset failed to deliver. The mirror is + * write-behind with no other feedback loop: a changeset that fails to + * deliver an owned output leaves a permanent hole, and because the + * engine is REBUILT from this mirror on restart (buildUtxoRestoreData), + * the hole graduates to a fund-loss on the next launch. Observed in the + * field as the job-flower 106.43→86.33 restart drop: rescan + * nondeterministically drops the change outputs of sends funded from + * CoinJoin-account outputs. + * + * Two passes, each bounded, sharing only the set of watch-only contact + * accounts they both exclude: + * + * * [healMissingTxos] pages the ENGINE ([engineUtxoPage]: `cursor` null + * to start, then the cursor the previous page returned while its + * `hasMore` is true) and inserts the rows the store lacks, one Room + * transaction per page. Insert-only and idempotent, so a sweep is + * many small commits rather than one giant one — that is the point. + * * [classifyStoreRows] pages the STORE and asks [classifyOutpoints] + * about one page at a time (0 unknown, 1 unspent, 2 spent). It + * writes nothing; every verdict is log-only. + * + * Neither side ever holds a set over a whole inventory: a wallet's UTXO + * count is chain-controlled (anyone who knows a watched address can + * keep sending dust to it), so a pass that materialized it would hand a + * remote party control over how much this process allocates on every + * SYNCED transition and every cadence tick. + * + * Rows the mirror holds and the engine lacks are LEFT ALONE (the mirror + * may legitimately be ahead — a live spend marks rows spent here before + * the engine's map settles — and it also carries watch-only contact + * outputs the engine's own accounts never report). Spent-state repair + * is deliberately out of scope. + * + * [minConfirmations] (default 100): the engine snapshot cannot carry + * `isCoinbase`/`isInstantLocked`, so inserted rows get + * `isConfirmed=true` and both flags false — inert for any output at or + * beyond coinbase maturity, which the gate guarantees. Fresher holes + * age into a later sweep. + * + * `netAmount` is reported, never repaired: a record born blind to one + * of its own outputs persisted a net short by exactly that output's + * value, but the record may equally have been corrected already by a + * callback racing this sweep, and blind addition double-credits. + * + * Returns null when the FIRST engine page is unavailable — there is + * nothing to reconcile against, so there is no report to make. A page + * or classification batch failing later truncates the sweep instead, + * which the report's `transportFailures` records. A page that does not + * decode as an [EngineUtxoPage] is a contract violation between the JNI + * emitter and this reader, and it throws rather than being absorbed. + * + * Must NOT be called from the handler's own [dispatcher] (it takes + * [callbackExclusion] and runs Room transactions). + */ + suspend fun reconcileTxos( + walletId: ByteArray, + tipHeight: Int, + minConfirmations: Int = 100, + pageSize: Int = TXO_RECONCILE_PAGE_SIZE, + engineUtxoPage: suspend (cursor: String?, limit: Int) -> String?, + classifyOutpoints: suspend (queriesJson: String) -> ByteArray?, + ): TxoReconcileReport? { + val limit = pageSize.coerceAtLeast(1) + + // Watch-only DIP-15 contact (external) accounts, resolved once up + // front because BOTH passes need the exclusion and the account set + // does not move under a sweep. The engine's UTXO inventory export + // includes these accounts' coins — it tracks them to show payments + // TO contacts — but they are the CONTACT's money and must never be + // healed into the store as ours. Before this check lived on the + // insert pass, a fresh restore's post-backfill reconcile healed + // every contact-payment coin into the store (12 rows / 0.05692493 + // tDASH on the large-wallet validation run of 2026-08-25) while the + // reverse pass — the only place the exclusion existed — dutifully + // counted the same rows as foreign. + val foreignAccountIds = database.accountDao() + .observeByWallet(walletId).first() + .filter { it.accountType == ACCOUNT_TYPE_TAG_DASHPAY_EXTERNAL } + .map { it.id } + .toSet() + + val heal = healMissingTxos( + walletId, tipHeight, minConfirmations, limit, foreignAccountIds, engineUtxoPage, + ) ?: return null + val classify = classifyStoreRows(walletId, limit, foreignAccountIds, classifyOutpoints) + + val report = TxoReconcileReport( + engineUtxos = heal.engineUtxos, + inserted = heal.inserted, + insertedDuffs = heal.insertedDuffs, + netAmountSuspects = heal.netAmountSuspects, + healedUnowned = heal.healedUnowned, + skippedImmature = heal.skippedImmature, + skippedNoAddress = heal.skippedNoAddress, + wouldFlipSpent = classify.wouldFlipSpent, + wouldFlipSpentDuffs = classify.wouldFlipSpentDuffs, + wouldRemove = classify.wouldRemove, + wouldRemoveDuffs = classify.wouldRemoveDuffs, + misattributed = classify.misattributed, + misattributedDuffs = classify.misattributedDuffs, + unresolvedAccount = classify.unresolvedAccount, + skippedForeign = heal.skippedForeign + classify.skippedForeign, + stuckSpent = classify.stuckSpent, + stuckSpentDuffs = classify.stuckSpentDuffs, + transportFailures = heal.transportFailures + classify.transportFailures, + ) + if (report.inserted > 0 || report.wouldFlipSpent > 0 || report.wouldRemove > 0 || + report.stuckSpent > 0 || report.misattributed > 0 || + report.unresolvedAccount > 0 || report.transportFailures > 0 + ) { + Log.w( + TAG, + "txos reconcile: healed ${report.inserted} missing TXO(s) " + + "(${report.insertedDuffs} duffs), " + + "${report.netAmountSuspects} netAmount suspect(s) (log-only), " + + "healedUnowned=${report.healedUnowned}, " + + "wouldFlipSpent=${report.wouldFlipSpent} " + + "(${report.wouldFlipSpentDuffs} duffs, log-only), " + + "wouldRemove=${report.wouldRemove} " + + "(${report.wouldRemoveDuffs} duffs, log-only), " + + "stuckSpent=${report.stuckSpent} " + + "(${report.stuckSpentDuffs} duffs, log-only), " + + "misattributed=${report.misattributed} " + + "(${report.misattributedDuffs} duffs, log-only), " + + "unresolvedAccount=${report.unresolvedAccount}, " + + "engine=${report.engineUtxos} " + + "skipped immature=${report.skippedImmature} " + + "noAddress=${report.skippedNoAddress} " + + "foreign=${report.skippedForeign} " + + "transportFailures=${report.transportFailures} — a non-zero " + + "heal after a completed sync means a changeset dropped an owned output", ) - db.txoDao().upsert(row) - // Drain any pending-input rows staged before this funding TXO - // existed — a port of the Swift `upsertUtxo` drain - // (PlatformWalletPersistenceHandler.swift). A spend that arrived - // first was deferred (see onWalletChangesetTransaction); now that - // the funding output is here, resolve the claim and clear the rows - // so the UTXO-restore path won't hand this consumed output back to - // Rust as spendable. - val pending = db.documentDao().getPendingInputsByOutpoint(outpoint) - if (pending.isNotEmpty()) { - // A tombstone outranks every ordinary row regardless of age: - // ordinary rows are competing *observations*, a tombstone is - // the sweep's settled verdict that its winner consumed this - // coin. Prefer the tombstone tagged with the delivering - // wallet; failing that any tombstone on the outpoint still - // holds — the stamp is a txid fact, not a per-wallet one. - val tombstones = pending.filter { it.isSweptTombstone } - val tombstone = tombstones.filter { it.walletId.contentEquals(walletId) } - .maxByOrNull { it.createdAt } - ?: tombstones.maxByOrNull { it.createdAt } - if (tombstone != null) { - // A drained tombstone STAMPS, it never mints a spender - // link: the winner need not have its own `transactions` - // row, and a link would make the coin non-releasable - // (the release pass frees stamped, unlinked rows) when a - // later sweep proves the winner never took it. The - // existing link, if any, is carried as it was. - db.txoDao().upsert( - row.copy( - isSpent = true, - supersededByTxid = tombstone.spendingTxid, - lastUpdated = now(), - ), - ) - } else { - // Competing ordinary observations: a network-final spender - // outranks a newer mempool one (its row is the settled - // claim the link guard protects); among equals the newest - // wins, as before (reorg / double-spend: newest wins). - val ranked = pending.map { p -> p to db.transactionDao().getByTxid(p.spendingTxid) } - val (chosen, spending) = ranked.maxWithOrNull( - compareBy>( - { it.second?.context ?: 0 }, - { it.first.createdAt }, - ), - )!! - val spendingContext = spending?.context ?: 0 - val keepExistingLink = - keepSettledSpenderLink(db, row, chosen.spendingTxid, spendingContext) - db.txoDao().upsert( - linkSpender(row, chosen.spendingTxid, chosen.inputIndex, spendingContext, keepExistingLink), + } else { + Log.i( + TAG, + "txos reconcile: mirror consistent (${report.engineUtxos} engine UTXOs, " + + "skipped immature=${report.skippedImmature} " + + "noAddress=${report.skippedNoAddress} foreign=${report.skippedForeign})", + ) + } + return report + } + + /** + * Insert pass of [reconcileTxos]: one engine page at a time, one Room + * transaction each. Each page is fetched OUTSIDE the exclusion lock — + * the fetch is a native call into the engine, and the lock exists to + * keep changeset callbacks out of our writes, not out of the engine. + * Returns null when the first page is unavailable (nothing to reconcile + * against); a later null page truncates the pass and is counted. + */ + private suspend fun healMissingTxos( + walletId: ByteArray, + tipHeight: Int, + minConfirmations: Int, + limit: Int, + foreignAccountIds: Set, + engineUtxoPage: suspend (cursor: String?, limit: Int) -> String?, + ): HealPass? { + var engineUtxos = 0 + var inserted = 0 + var insertedDuffs = 0L + var netAmountSuspects = 0 + var healedUnowned = 0 + var skippedImmature = 0 + var skippedNoAddress = 0 + var skippedForeign = 0 + var transportFailures = 0 + + var cursor: String? = null + while (true) { + val pageJson = engineUtxoPage(cursor, limit) + if (pageJson == null) { + // No first page: nothing to reconcile against, no report. + if (cursor == null) return null + // The transport died mid-sweep. Everything already applied + // stands (insert-only, idempotent); the rest waits for the + // next cadence tick. + transportFailures++ + break + } + // Strict decode: an unknown key, a missing required field or a + // mistyped value throws — the row would otherwise be healed with + // a defaulted owner or a zero amount. + val page = engineJson.decodeFromString(EngineUtxoPage.serializer(), pageJson) + engineUtxos += page.utxos.size + + if (page.utxos.isNotEmpty()) { + callbackExclusion.withLock { + database.withTransaction { + for (row in page.utxos) { + when (healEngineRow(walletId, row, tipHeight, minConfirmations, foreignAccountIds)) { + HealOutcome.SKIPPED_IMMATURE -> skippedImmature++ + HealOutcome.SKIPPED_NO_ADDRESS -> skippedNoAddress++ + HealOutcome.SKIPPED_FOREIGN -> skippedForeign++ + HealOutcome.ALREADY_PRESENT -> {} + HealOutcome.HEALED -> { + inserted++ + insertedDuffs += row.amount + } + HealOutcome.HEALED_UNOWNED -> { + inserted++ + insertedDuffs += row.amount + healedUnowned++ + } + HealOutcome.HEALED_NET_SUSPECT -> { + inserted++ + insertedDuffs += row.amount + netAmountSuspects++ + } + HealOutcome.HEALED_UNOWNED_NET_SUSPECT -> { + inserted++ + insertedDuffs += row.amount + healedUnowned++ + netAmountSuspects++ + } + } + } + } + } + } + if (!page.hasMore || page.cursor == null) break + cursor = page.cursor + } + return HealPass( + engineUtxos = engineUtxos, + inserted = inserted, + insertedDuffs = insertedDuffs, + netAmountSuspects = netAmountSuspects, + healedUnowned = healedUnowned, + skippedImmature = skippedImmature, + skippedNoAddress = skippedNoAddress, + skippedForeign = skippedForeign, + transportFailures = transportFailures, + ) + } + + private enum class HealOutcome { + SKIPPED_IMMATURE, + SKIPPED_NO_ADDRESS, + SKIPPED_FOREIGN, + ALREADY_PRESENT, + HEALED, + HEALED_UNOWNED, + HEALED_NET_SUSPECT, + HEALED_UNOWNED_NET_SUSPECT, + } + + /** + * Apply one engine inventory row to the store (inside the caller's + * exclusion lock and Room transaction). Provable-only discipline: it + * neither mutates nor suppresses on guesswork. + */ + private suspend fun healEngineRow( + walletId: ByteArray, + row: EngineUtxoRow, + tipHeight: Int, + minConfirmations: Int, + foreignAccountIds: Set, + ): HealOutcome { + if (row.height <= 0 || tipHeight - row.height + 1 < minConfirmations) { + return HealOutcome.SKIPPED_IMMATURE + } + if (row.address.isEmpty()) return HealOutcome.SKIPPED_NO_ADDRESS + // The inventory tags every UTXO with its owning account tuple. The + // tag is the authoritative foreign check — a watch-only external + // account's coin is the CONTACT's money whether or not its address + // row survived persistence. The address-based check (through + // core_addresses.accountId, the same second path rowIsForeign uses + // for store rows) stays as a fallback. An unresolvable address is + // NOT provably foreign — those proceed. + if (row.typeTag == ACCOUNT_TYPE_TAG_DASHPAY_EXTERNAL) return HealOutcome.SKIPPED_FOREIGN + val addressOwner = database.coreAddressDao().getByAddress(row.address)?.accountId + if (addressOwner != null && addressOwner in foreignAccountIds) { + return HealOutcome.SKIPPED_FOREIGN + } + val txid = row.txid.hexToByteArray() + require(txid.size == 32) { "engine inventory row carries a ${txid.size}-byte txid" } + if (database.txoDao().getByOutpoint(makeOutpoint(txid, row.vout)) != null) { + return HealOutcome.ALREADY_PRESENT + } + // Resolve the Room account from the tuple and stamp it on the + // healed row. Ownership must not depend on the address projection: + // the two things persistence loses together are the TXO and its + // address row, and a healed row with neither link is skipped by + // the restore loader at the next mirror-reload — recreating the + // fund loss the heal repaired. + val ownerAccountId = fetchAccount( + database, walletId, row.typeTag, row.index, row.standardTag, + row.registrationIndex, row.keyClass, + row.userIdentityId?.hexToByteArray() ?: ByteArray(32), + row.friendIdentityId?.hexToByteArray() ?: ByteArray(32), + )?.id + if (ownerAccountId == null) { + // Heal anyway — the address projection may still attribute it — + // but surface the unresolved owner: if the address row is also + // gone, this row will not survive the next mirror-reload. + Log.w( + TAG, + "txos reconcile: healing TXO with UNRESOLVED account " + + "(typeTag=${row.typeTag} index=${row.index} address=${row.address}) — " + + "ownership rides on the address projection alone", + ) + } + upsertUtxoRow( + database, walletId, txid, row.vout, row.amount, row.address, + row.scriptHex.hexToByteArray(), row.height, + isCoinbase = row.isCoinbase, + isConfirmed = row.isConfirmed, + isInstantLocked = row.isInstantlocked, + isLocked = row.isLocked, + resolvedAccountId = ownerAccountId, + ) + // netAmount is NOT mutated here. The record's net may already be + // correct (a corrective record callback can land while its TXO + // delivery races this sweep), and adding the healed amount to an + // already-corrected net double-credits. The event pipeline owns + // net correctness; this pass only reports the suspicion. + val priorTx = database.transactionDao().getByTxid(txid) + val netSuspect = priorTx != null && priorTx.transactionData.isNotEmpty() + if (netSuspect) { + Log.w( + TAG, + "txos reconcile: healed TXO ${row.txid}:${row.vout} (${row.amount} duffs) " + + "has a pre-existing record whose netAmount may be short by that " + + "amount — LOG-ONLY, storedNet=${priorTx?.netAmount}", + ) + } + return when { + ownerAccountId == null && netSuspect -> HealOutcome.HEALED_UNOWNED_NET_SUSPECT + ownerAccountId == null -> HealOutcome.HEALED_UNOWNED + netSuspect -> HealOutcome.HEALED_NET_SUSPECT + else -> HealOutcome.HEALED + } + } + + /** + * Classification pass of [reconcileTxos] (the widened scope from the + * #4425 / pre-#971 review). Inverted relative to the insert pass — it + * pages the STORE and asks the engine about each page — so that neither + * side has to hold a set over a whole inventory. + * + * Every query carries the account tuple the store files the coin under + * and the script the store recorded. Those are the store's CLAIM; the + * engine checks both against its own pools rather than trusting them, + * which is why a bare outpoint cannot ask the question and why a row + * whose owning account this store cannot resolve is counted rather + * than asked about. + * + * Read-only by construction: it writes nothing, so it runs outside both + * the exclusion lock and any transaction. A row a concurrent callback + * moves under it is at worst a stale log line, and every verdict here + * is log-only anyway. + * + * Watch-only DIP-15 contact rows are excluded via [foreignAccountIds]. + * Production changeset writes leave txos.accountId null and route + * ownership through coreAddressId -> core_addresses.accountId, so the + * exclusion resolves BOTH paths — an accountId-only check silently + * classifies every contact row. + */ + private suspend fun classifyStoreRows( + walletId: ByteArray, + limit: Int, + foreignAccountIds: Set, + classifyOutpoints: suspend (queriesJson: String) -> ByteArray?, + ): ClassifyPass { + var wouldFlipSpent = 0 + var wouldFlipSpentDuffs = 0L + var wouldRemove = 0 + var wouldRemoveDuffs = 0L + var stuckSpent = 0 + var stuckSpentDuffs = 0L + var misattributed = 0 + var misattributedDuffs = 0L + var unresolvedAccount = 0 + var skippedForeign = 0 + var transportFailures = 0 + + // Owning Room account for a store row: the explicit link first, + // then the address projection production writes actually use. + suspend fun ownerAccount( + row: org.dashfoundation.dashsdk.persistence.entities.TxoEntity, + ): org.dashfoundation.dashsdk.persistence.entities.AccountEntity? { + row.accountId?.let { return database.accountDao().getById(it) } + val addr = row.coreAddressId ?: return null + val ownerId = database.coreAddressDao().getByAddress(addr)?.accountId ?: return null + return database.accountDao().getById(ownerId) + } + // An empty BLOB sorts before every real outpoint, so this starts at + // the first row. + var after = ByteArray(0) + while (true) { + val storeRows = database.txoDao().pageByWallet(walletId, after, limit) + if (storeRows.isEmpty()) break + after = storeRows.last().outpoint + + // Rows worth asking the engine about. A row still mid-insert + // (no txid yet) is not classifiable, a foreign row's absence + // from the engine is expected rather than divergence — counted, + // as before, only when it is unspent — and a row whose owning + // account cannot be resolved carries no claim to check. + val classifiable = + ArrayList( + storeRows.size, + ) + val queries = ArrayList(storeRows.size) + for (row in storeRows) { + if (row.txid == null || row.outpoint.size != OUTPOINT_BYTES) continue + val owner = ownerAccount(row) + if (owner != null && owner.id in foreignAccountIds) { + if (!row.isSpent) skippedForeign++ + continue + } + if (owner == null) { + // No account to name in the query, so the engine has + // nothing to check the claim against. Reported rather + // than guessed: the same missing link is what makes a + // healed row fail to survive the next mirror-reload. + if (!row.isSpent) unresolvedAccount++ + continue + } + classifiable.add(row) + queries.add( + OutpointQuery( + typeTag = owner.accountType, + standardTag = owner.standardTag, + index = owner.accountIndex, + registrationIndex = owner.registrationIndex, + keyClass = owner.keyClass, + userIdentityId = owner.userIdentityId + .takeIf { it.size == 32 && it.any { b -> b != 0.toByte() } }?.toHex(), + friendIdentityId = owner.friendIdentityId + .takeIf { it.size == 32 && it.any { b -> b != 0.toByte() } }?.toHex(), + txid = row.txid.toHex(), + vout = row.vout, + scriptHex = row.scriptPubKey.toHex(), + ), + ) + } + if (classifiable.isEmpty()) continue + + val verdicts = classifyOutpoints( + engineJson.encodeToString(kotlinx.serialization.builtins.ListSerializer(OutpointQuery.serializer()), queries), + ) + if (verdicts == null || verdicts.size != classifiable.size) { + // No verdicts, no classification. Log-only either way, so + // the sweep stops rather than guessing. + transportFailures++ + break + } + for ((i, row) in classifiable.withIndex()) { + val verdict = verdicts[i] + val key = "${row.txid?.toHex()}:${row.vout}" + if (verdict == OUTPOINT_CLASS_NOT_OWNED) { + // The account this store filed the coin under does not + // monitor its script, so the engine could never have + // credited it there. A store attribution error, not a + // missing coin — and the one verdict that says nothing + // about spentness, so it is reported for spent and + // unspent rows alike. + misattributed++ + misattributedDuffs += row.amount + Log.w( + TAG, + "txos reconcile: store row filed under an account that does " + + "not own its script outpoint=$key amount=${row.amount} " + + "accountId=${row.accountId} — LOG-ONLY, not re-filed", ) + continue + } + if (row.isSpent) { + // Rows marked spent for coins the engine still lists + // unspent. Either lost-release residue (pre-#971) or a + // live spend racing the engine — never un-marked, only + // reported: un-marking a coin mid-payment would let the + // wallet double-spend it. + if (verdict == OUTPOINT_CLASS_UNSPENT) { + stuckSpent++ + stuckSpentDuffs += row.amount + Log.w( + TAG, + "txos reconcile: store row spent but engine lists it " + + "unspent outpoint=$key amount=${row.amount} — LOG-ONLY " + + "(lost release, or a live spend racing the engine)", + ) + } + continue + } + when (verdict) { + OUTPOINT_CLASS_UNSPENT -> {} + OUTPOINT_CLASS_KNOWN_UNCREDITED -> { + // Lost spend update (#4425), on durable evidence: + // this verdict requires the owning account to know + // the funding txid, to recognise the script, not to + // hold the coin, AND a funds account to hold a MINED + // record spending the outpoint — so unlike the old + // spent-set answer it cannot be a mempool artifact. + // Still LOG-ONLY here: flipping spent state is the + // one direction where a reconciler bug spends a + // coin the user still owns, and this pass is + // insert-only by contract. Acting on it is a + // separate, reviewable change. + wouldFlipSpent++ + wouldFlipSpentDuffs += row.amount + Log.w( + TAG, + "txos reconcile: store row unspent but the engine holds a " + + "mined record spending it outpoint=$key " + + "amount=${row.amount} — LOG-ONLY, not flipped", + ) + } + else -> { + // OUTPOINT_CLASS_UNKNOWN: the engine has no opinion. + // Swept/abandoned residue (pre-#971 stores), a + // funding transaction this session never processed, + // or an engine gap — and after a restart the + // finalized set is empty, so this is the common + // answer rather than the alarming one. Deliberately + // LOG-ONLY: removal by reconciliation is the one + // direction where a bug destroys user-visible data. + wouldRemove++ + wouldRemoveDuffs += row.amount + Log.w( + TAG, + "txos reconcile: engine has no verdict for store row " + + "outpoint=$key amount=${row.amount} — ambiguous " + + "(swept/abandoned residue, funding not seen this " + + "session, or a dropped engine record) — LOG-ONLY, " + + "not removed", + ) + } } - for (p in pending) db.documentDao().deletePendingInput(p) } } - 0 + return ClassifyPass( + wouldFlipSpent = wouldFlipSpent, + wouldFlipSpentDuffs = wouldFlipSpentDuffs, + wouldRemove = wouldRemove, + wouldRemoveDuffs = wouldRemoveDuffs, + stuckSpent = stuckSpent, + stuckSpentDuffs = stuckSpentDuffs, + misattributed = misattributed, + misattributedDuffs = misattributedDuffs, + unresolvedAccount = unresolvedAccount, + skippedForeign = skippedForeign, + transportFailures = transportFailures, + ) } override fun onWalletChangesetUtxoSpent( @@ -3980,6 +4757,48 @@ class PlatformWalletPersistenceHandler( runBlocking(dispatcher) { block() } companion object { + /** `AccountTypeTagFFI::DashpayExternalAccount` — watch-only DIP-15 + * contact accounts the engine's inventories never report. */ + internal const val ACCOUNT_TYPE_TAG_DASHPAY_EXTERNAL = 13 + + /** Bytes of a `txos.outpoint` key: 32-byte txid (wire order) plus + * the vout as a little-endian `Int`. Also the wire format of the + * reconcile's outpoint-classification batch. */ + internal const val OUTPOINT_BYTES = 36 + + /** + * Rows per page in both directions of [reconcileTxos] — engine + * UTXOs coming in, store rows going out for classification. + * + * The number itself is not delicate; that there IS one is the + * point. Inventory size is chain-controlled, so an unpaged sweep + * would let anyone who knows a watched address decide how much a + * phone allocates at every SYNCED transition and cadence tick. + */ + const val TXO_RECONCILE_PAGE_SIZE = 512 + + /** + * Decoder for the engine inventory transport. Deliberately the + * strict default (`ignoreUnknownKeys = false`, no coercion): the JNI + * emitter and [EngineUtxoPage] are one contract, and a drift between + * them must fail the decode, not heal a defaulted row. + */ + private val engineJson = Json + + /** [reconcileTxos] classification verdicts, mirroring + * `platform_wallet::manager::accessors::OutpointClass`. Only + * [OUTPOINT_CLASS_KNOWN_UNCREDITED] is positive evidence: the + * owning account knows the funding txid, recognises the script, + * does not hold the coin, AND a funds account holds a MINED + * record that spends the outpoint. Plain absence from the live + * UTXO set is NOT evidence — a mempool spend that never confirms + * and a conflict sweep both leave a coin absent with its funding + * known — so those arrive as [OUTPOINT_CLASS_UNKNOWN]. */ + internal const val OUTPOINT_CLASS_UNKNOWN: Byte = 0 + internal const val OUTPOINT_CLASS_UNSPENT: Byte = 1 + internal const val OUTPOINT_CLASS_KNOWN_UNCREDITED: Byte = 2 + internal const val OUTPOINT_CLASS_NOT_OWNED: Byte = 3 + internal const val PERSISTENCE_CAPABILITIES_VERSION: Int = 1 internal const val CAPABILITY_ATOMIC_CHANGESETS: Long = 0x01 internal const val CAPABILITY_INVITATIONS: Long = 0x02 diff --git a/packages/kotlin-sdk/sdk/src/main/kotlin/org/dashfoundation/dashsdk/persistence/dao/TxoDao.kt b/packages/kotlin-sdk/sdk/src/main/kotlin/org/dashfoundation/dashsdk/persistence/dao/TxoDao.kt index ac484917836..67b9b6de722 100644 --- a/packages/kotlin-sdk/sdk/src/main/kotlin/org/dashfoundation/dashsdk/persistence/dao/TxoDao.kt +++ b/packages/kotlin-sdk/sdk/src/main/kotlin/org/dashfoundation/dashsdk/persistence/dao/TxoDao.kt @@ -21,6 +21,23 @@ interface TxoDao { @Query("SELECT * FROM txos WHERE walletId = :walletId") fun observeByWallet(walletId: ByteArray): Flow> + /** + * One outpoint-ordered page of a wallet's TXOs, for a pass that must + * not hold the whole table at once (the store reconcile's reverse + * half). Pass an empty [after] to start — an empty BLOB sorts before + * every real 36-byte outpoint — then the previous page's last + * `outpoint` to continue. + * + * `outpoint` is the primary key, so the order is an index walk and the + * cursor is exact: no row can be visited twice or skipped because + * another one was inserted or deleted mid-sweep. + */ + @Query( + "SELECT * FROM txos WHERE walletId = :walletId AND outpoint > :after " + + "ORDER BY outpoint LIMIT :limit", + ) + suspend fun pageByWallet(walletId: ByteArray, after: ByteArray, limit: Int): List + /** WalletMemoryExplorer: `txo.walletId == walletId && txo.isSpent == false`. */ @Query("SELECT * FROM txos WHERE walletId = :walletId AND isSpent = 0") fun observeUnspentByWallet(walletId: ByteArray): Flow> diff --git a/packages/kotlin-sdk/sdk/src/main/kotlin/org/dashfoundation/dashsdk/wallet/PlatformWalletManager.kt b/packages/kotlin-sdk/sdk/src/main/kotlin/org/dashfoundation/dashsdk/wallet/PlatformWalletManager.kt index 7ef17c2d23e..8b332d1f23d 100644 --- a/packages/kotlin-sdk/sdk/src/main/kotlin/org/dashfoundation/dashsdk/wallet/PlatformWalletManager.kt +++ b/packages/kotlin-sdk/sdk/src/main/kotlin/org/dashfoundation/dashsdk/wallet/PlatformWalletManager.kt @@ -1243,6 +1243,81 @@ class PlatformWalletManager( mapNativeErrors { DashpayNative.walletManagerAccountBalances(managerHandle, walletId) } } + /** + * Reconcile the Room `txos` mirror against the engine's live UTXO + * inventory, healing rows a changeset failed to deliver. The mirror is + * write-behind with no other feedback loop, and the engine is REBUILT + * from it on restart — an unhealed hole becomes a fund-loss on the next + * launch (the job-flower 106.43→86.33 restart drop: rescan + * nondeterministically drops the change outputs of sends funded from + * CoinJoin-account outputs). Insert-only; never flips spent state or + * deletes. + * + * Both directions of the sweep are paged, so neither this process nor + * the engine ever holds a whole wallet's inventory: UTXO counts are + * chain-controlled, and a periodic full-inventory read would let anyone + * who knows a watched address decide how much a phone allocates. See + * [PlatformWalletPersistenceHandler.reconcileTxos]. + * + * Call it after the L1 scan settles and again on a slow cadence; + * [tipHeight] is the synced chain height — only outputs at least + * [minConfirmations] deep are healed (immature holes age into the next + * sweep). Returns null when the engine inventory read failed at the + * FIRST page: there is nothing to reconcile against, so there is no + * report to make. A page or classification batch failing later + * truncates the sweep instead, which the report's `transportFailures` + * records. Native failures surface through [mapNativeErrors] as + * exceptions; both transport lambdas turn them into the null the + * handler's truncate-and-report contract is written against (and log + * them), so a mid-sweep fault yields a partial report rather than + * discarding the sweep. A malformed page is NOT absorbed: the handler's + * strict decode throws, and that propagates. + */ + suspend fun reconcileTxoStore( + walletId: ByteArray, + tipHeight: Int, + minConfirmations: Int = 100, + ): PlatformWalletPersistenceHandler.TxoReconcileReport? = + persistenceHandler.reconcileTxos( + walletId = walletId, + tipHeight = tipHeight, + minConfirmations = minConfirmations, + engineUtxoPage = { cursor, limit -> + withContext(Dispatchers.IO) { + runCatching { + mapNativeErrors { + WalletManagerNative.walletManagerUtxosPageJson( + managerHandle, walletId, cursor, limit, + ) + } + }.onFailure { t -> + android.util.Log.w( + "PlatformWalletManager", + "txos reconcile: engine inventory page failed (cursor=$cursor)", + t, + ) + }.getOrNull() + } + }, + classifyOutpoints = { queriesJson -> + withContext(Dispatchers.IO) { + runCatching { + mapNativeErrors { + WalletManagerNative.walletManagerClassifyOutpoints( + managerHandle, walletId, queriesJson, + ) + } + }.onFailure { t -> + android.util.Log.w( + "PlatformWalletManager", + "txos reconcile: outpoint classification failed", + t, + ) + }.getOrNull() + } + }, + ) + /** * Refresh the persisted DashPay payment history for one identity: * one FFI read (`managed_identity_get_dashpay_payments`) + one Room @@ -2146,6 +2221,7 @@ class PlatformWalletManager( if (running) { runCatching { spvSyncProgress() }.getOrNull()?.let { next -> if (next != _spvProgress.value) _spvProgress.value = next + maybeReconcileTxoStores(next) } runCatching { spvTipUnixSeconds() }.getOrNull()?.let { tip -> if (tip != _spvTipUnixSeconds.value) _spvTipUnixSeconds.value = tip @@ -2159,6 +2235,61 @@ class PlatformWalletManager( } } + private var lastTxoReconcileAtMs = 0L + private var txoReconcileWasSynced = false + + /** + * SDK-internal trigger for [reconcileTxoStore] — runs on the SYNCED + * transition of the SPV progress poll and again every + * [TXO_RECONCILE_INTERVAL_MS] while synced, for every loaded wallet. + * Lives here rather than in the host apps so Android and iOS-parity + * hosts both get the heal without wiring anything: the mirror hole it + * repairs (rescan dropping change outputs of CoinJoin-funded sends) + * becomes a fund-loss on the next engine reload if any host forgets + * to call it. Failures are logged and re-tried on the next cadence + * tick — never allowed to kill the progress poll. + */ + private fun maybeReconcileTxoStores(progress: SpvSyncProgressData) { + val synced = progress.overallState == SpvSyncState.SYNCED + val transitioned = synced && !txoReconcileWasSynced + txoReconcileWasSynced = synced + if (!synced) return + val now = System.currentTimeMillis() + if (!transitioned && now - lastTxoReconcileAtMs < TXO_RECONCILE_INTERVAL_MS) return + // The filter sub-phase is the wallet-relevant height; if dash-spv + // reports SYNCED without one (filter sync disabled, or the phase + // dropped after completion) fall back to the header tip rather than + // silently skipping the heal this exists to deliver. + val tipHeight = ( + progress.filters?.currentHeight?.takeIf { it > 0L } + ?: progress.headers?.currentHeight + ?: 0L + ).toInt() + if (tipHeight <= 0) { + android.util.Log.w( + "PlatformWalletManager", + "txos reconcile skipped: SYNCED progress carries no tip height " + + "(filters=${progress.filters?.currentHeight} headers=${progress.headers?.currentHeight})", + ) + return + } + val walletIds = wallets.value.values.map { it.walletId } + if (walletIds.isEmpty()) return + lastTxoReconcileAtMs = now + scope.launch { + for (walletId in walletIds) { + runCatching { reconcileTxoStore(walletId, tipHeight) } + .onFailure { t -> + android.util.Log.w( + "PlatformWalletManager", + "txos reconcile failed for wallet ${walletId.toHex()}", + t, + ) + } + } + } + } + // ── DashPay sync + seedless unlock ──────────────────────────────── // // Port of `PlatformWalletManagerDashPaySync.swift` + the unlock flow @@ -2572,6 +2703,14 @@ class PlatformWalletManager( /** SPV progress poll cadence — matches Swift's 1 Hz `startProgressPolling`. */ const val POLL_INTERVAL_MS = 1_000L + /** + * Cadence of the steady-state TXO-store reconcile + * ([maybeReconcileTxoStores]) while SPV reports SYNCED. The + * SYNCED transition itself always triggers a pass regardless of + * this interval. + */ + const val TXO_RECONCILE_INTERVAL_MS = 30 * 60 * 1_000L + /** De-offset `PlatformWalletFFIResultCode::ErrorInvalidParameter`. */ const val PWFFI_INVALID_PARAMETER = 2 } diff --git a/packages/kotlin-sdk/sdk/src/test/kotlin/org/dashfoundation/dashsdk/persistence/PlatformWalletPersistenceHandlerTest.kt b/packages/kotlin-sdk/sdk/src/test/kotlin/org/dashfoundation/dashsdk/persistence/PlatformWalletPersistenceHandlerTest.kt index 0aaacf099e1..6205fa0c7c0 100644 --- a/packages/kotlin-sdk/sdk/src/test/kotlin/org/dashfoundation/dashsdk/persistence/PlatformWalletPersistenceHandlerTest.kt +++ b/packages/kotlin-sdk/sdk/src/test/kotlin/org/dashfoundation/dashsdk/persistence/PlatformWalletPersistenceHandlerTest.kt @@ -13,6 +13,10 @@ import androidx.test.core.app.ApplicationProvider import kotlinx.coroutines.Dispatchers import kotlinx.coroutines.flow.first import kotlinx.coroutines.test.runTest +import kotlinx.serialization.json.int +import kotlinx.serialization.json.jsonArray +import kotlinx.serialization.json.jsonObject +import kotlinx.serialization.json.jsonPrimitive import org.dashfoundation.dashsdk.Network import org.dashfoundation.dashsdk.errors.DashSdkError import org.dashfoundation.dashsdk.ffi.NativePersistenceBridge @@ -6044,6 +6048,66 @@ class PlatformWalletPersistenceHandlerTest { assertTrue(row.isSweptTombstone) } + // ── TXO-store reconcile (the job-flower change-drop repair) ─────── + + private val changeTxid = ByteArray(32) { 7 } + private val reconcileTip = 1_536_950 + + private fun engineUtxoJson( + txidHex: String, + vout: Int, + amount: Long, + address: String = "yStxXHHzhAx58JhaPBNhn3xsH93UwBM2nd", + height: Int = 1_534_921, + isConfirmed: Boolean = true, + isCoinbase: Boolean = false, + ): String = + """{"utxos":[{"typeTag":0,"standardTag":0,"index":0,"txid":"$txidHex","vout":$vout,""" + + """"amount":$amount,"address":"$address","scriptHex":"76a914000088ac",""" + + """"height":$height,"isConfirmed":$isConfirmed,"isInstantlocked":false,""" + + """"isCoinbase":$isCoinbase,"isLocked":false}]}""" + + private fun ByteArray.toHexLower() = joinToString("") { "%02x".format(it) } + + @Test + fun reconcileHealsMissingChangeTxoAndRepairsNetAmount() = runTest { + // A send record born blind to its own change output: netAmount + // persisted as the full input value (the job-flower 6cef55ab… + // shape) and NO txos row for the change. + db.transactionDao().upsert( + org.dashfoundation.dashsdk.persistence.entities.TransactionEntity( + txid = changeTxid, + transactionData = byteArrayOf(1, 2, 3), + netAmount = -1_000_010_000L, + ), + ) + + val report = handler.reconcileFromInventory( + walletId, + engineUtxoJson(changeTxid.toHexLower(), vout = 1, amount = 989_009_773L), + tipHeight = reconcileTip, + ) + + assertEquals(1, report.inserted) + assertEquals(989_009_773L, report.insertedDuffs) + assertEquals(1, report.netAmountSuspects) + + val row = db.txoDao().getByOutpoint(makeOutpoint(changeTxid, 1)) + assertNotNull(row) + assertFalse(row!!.isSpent) + assertEquals(989_009_773L, row.amount) + assertTrue(row.isConfirmed) + + // The stored netAmount is NOT mutated: the record may already carry + // the corrected net (a corrective callback racing this sweep), and + // blind addition double-credits. The suspicion is logged; the event + // pipeline owns net correctness. + assertEquals( + -1_000_010_000L, + db.transactionDao().getByTxid(changeTxid)!!.netAmount, + ) + } + @Test fun aRepointedTombstoneIsRestampedToTheLaterSweep() = runTest { // A chained sweep that re-points a still-unfunded claim to a new @@ -6548,6 +6612,791 @@ class PlatformWalletPersistenceHandlerTest { chainLockHeightRound(handler, 600) assertEquals(600, db.walletDao().getByWalletId(walletId)!!.lastAppliedChainLockHeight) } + + @Test + fun reconcileIsIdempotentAndNeverDoubleCredits() = runTest { + db.transactionDao().upsert( + org.dashfoundation.dashsdk.persistence.entities.TransactionEntity( + txid = changeTxid, + transactionData = byteArrayOf(1), + netAmount = -1_000_010_000L, + ), + ) + val json = engineUtxoJson(changeTxid.toHexLower(), vout = 1, amount = 989_009_773L) + + handler.reconcileFromInventory(walletId, json, tipHeight = reconcileTip) + val second = handler.reconcileFromInventory(walletId, json, tipHeight = reconcileTip) + + assertEquals(0, second.inserted) + assertEquals(0, second.netAmountSuspects) + assertEquals( + -1_000_010_000L, + db.transactionDao().getByTxid(changeTxid)!!.netAmount, + ) + } + + @Test + fun reconcileSkipsImmatureOutputsAndPreservesSpentRows() = runTest { + // Immature: inside the 100-conf gate (flags on the engine snapshot + // can't carry coinbase/IS-lock, so fresh rows wait for a later + // sweep) — nothing inserted. + val fresh = handler.reconcileFromInventory( + walletId, + engineUtxoJson(changeTxid.toHexLower(), vout = 0, amount = 5L, height = reconcileTip - 3), + tipHeight = reconcileTip, + ) + assertEquals(0, fresh.inserted) + assertEquals(1, fresh.skippedImmature) + assertNull(db.txoDao().getByOutpoint(makeOutpoint(changeTxid, 0))) + + // A row the mirror already holds — even marked spent while the + // engine still lists it — is left untouched: reconcile is + // insert-only and never flips spend state. + assertEquals( + 0, + handler.onWalletChangesetUtxoAdded( + walletId, changeTxid, 2, 42L, "yTestAddr", byteArrayOf(0x51), 1_500_000, + false, true, false, false, + ), + ) + val seeded = db.txoDao().getByOutpoint(makeOutpoint(changeTxid, 2))!! + db.txoDao().upsert(seeded.copy(isSpent = true)) + + val report = handler.reconcileFromInventory( + walletId, + engineUtxoJson(changeTxid.toHexLower(), vout = 2, amount = 42L, height = 1_500_000), + tipHeight = reconcileTip, + ) + assertEquals(0, report.inserted) + assertTrue(db.txoDao().getByOutpoint(makeOutpoint(changeTxid, 2))!!.isSpent) + } + + /** Engine inventory JSON with both halves: unspent rows and spent outpoints. */ + private fun engineInventoryJson(unspent: List>, spent: List>): String { + val utxoRows = unspent.joinToString(",") { (txid, vout, amount) -> + """{"typeTag":0,"standardTag":0,"index":0,"txid":"$txid","vout":$vout,""" + + """"amount":$amount,"address":"yStxXHHzhAx58JhaPBNhn3xsH93UwBM2nd",""" + + """"scriptHex":"76a914000088ac","height":1400000,"isConfirmed":true,""" + + """"isInstantlocked":false,"isCoinbase":false,"isLocked":false}""" + } + val spentRows = spent.joinToString(",") { (txid, vout) -> + """{"txid":"$txid","vout":$vout}""" + } + return """{"utxos":[$utxoRows],"spent":[$spentRows]}""" + } + + @Test + fun reconcileLogsButNeverFlipsLostSpendRows() = runTest { + // A store row still marked unspent for a coin the engine records as + // spent (dashpay/platform#4425). The engine's spent set includes + // MEMPOOL spends and carries no context, so persisting the flip + // would settle an unconfirmed spend — counted and logged only. + handler.onWalletChangesetUtxoAdded( + walletId, changeTxid, 3, 500_000L, "yTestAddr", byteArrayOf(0x51), 1_400_000, + false, true, false, false, + ) + val report = handler.reconcileFromInventory( + walletId, + engineInventoryJson(unspent = emptyList(), spent = listOf(changeTxid.toHexLower() to 3)), + tipHeight = reconcileTip, + ) + assertEquals(1, report.wouldFlipSpent) + assertEquals(500_000L, report.wouldFlipSpentDuffs) + assertEquals(0, report.wouldRemove) + val row = db.txoDao().getByOutpoint(makeOutpoint(changeTxid, 3))!! + assertFalse("the row must stay unspent — the flip is log-only", row.isSpent) + assertNull(row.spendingTxid) + } + + @Test + fun reconcileLogsButNeverRemovesEngineUnknownRows() = runTest { + // A store row for a coin the engine has in NEITHER inventory — + // residue of a swept/abandoned transaction (pre-rust-dashcore#971 + // stores). Counted and logged, NEVER removed. + handler.onWalletChangesetUtxoAdded( + walletId, changeTxid, 4, 250_000L, "yTestAddr", byteArrayOf(0x51), 1_400_000, + false, true, false, false, + ) + val report = handler.reconcileFromInventory( + walletId, + engineInventoryJson(unspent = emptyList(), spent = emptyList()), + tipHeight = reconcileTip, + ) + assertEquals(1, report.wouldRemove) + assertEquals(250_000L, report.wouldRemoveDuffs) + assertEquals(0, report.wouldFlipSpent) + val row = db.txoDao().getByOutpoint(makeOutpoint(changeTxid, 4))!! + assertFalse(row.isSpent) + assertEquals(250_000L, row.amount) + } + + @Test + fun reconcileStampsTheEnginesOwnFlagsOnAHealedRow() = runTest { + // The inventory carries the engine's own isCoinbase / isConfirmed / + // isInstantLocked. They are stamped as given, not assumed: a + // coinbase output filed as an ordinary one misstates maturity in + // the mirror the engine reloads from, and the old transport had no + // field to carry the truth in. + val report = handler.reconcileFromInventory( + walletId, + engineUtxoJson( + changeTxid.toHexLower(), + vout = 20, + amount = 5_000_000_000L, + isConfirmed = true, + isCoinbase = true, + ), + tipHeight = reconcileTip, + ) + + assertEquals(1, report.inserted) + val row = db.txoDao().getByOutpoint(makeOutpoint(changeTxid, 20))!! + assertTrue("a coinbase output must be filed as one", row.isCoinbase) + assertTrue(row.isConfirmed) + } + + @Test + fun reconcileReportsRowsFiledUnderAnAccountThatDoesNotOwnThem() = runTest { + // NOT_OWNED: the account the store filed the coin under does not + // monitor its script, so the engine could never have credited it + // there. That is a store attribution error rather than a missing + // coin, and it is the one verdict that says nothing about + // spentness — so it is reported on its own counter and the row is + // left exactly as it was. + handler.onWalletChangesetUtxoAdded( + walletId, changeTxid, 21, 777_000L, "yTestAddr", byteArrayOf(0x51), 1_400_000, + false, true, false, false, + ) + val json = """{"utxos":[],"spent":[],"notOwned":[""" + + """{"txid":"${changeTxid.toHexLower()}","vout":21}]}""" + + val report = handler.reconcileFromInventory(walletId, json, tipHeight = reconcileTip) + + assertEquals(1, report.misattributed) + assertEquals(777_000L, report.misattributedDuffs) + assertEquals("a not-owned row is not also counted as removable", 0, report.wouldRemove) + assertEquals(0, report.wouldFlipSpent) + val row = db.txoDao().getByOutpoint(makeOutpoint(changeTxid, 21))!! + assertFalse("the row is left alone — the verdict is log-only", row.isSpent) + assertEquals(777_000L, row.amount) + } + + @Test + fun reconcileCountsRowsWithNoResolvableOwnerInsteadOfAskingAboutThem() = runTest { + // A query names the account the store filed the coin under, because + // the engine checks that claim against its own pools. A row whose + // owner resolves through neither the explicit link nor the + // core_addresses projection carries no claim to check, so it is + // counted rather than asked about — and the same missing link is + // what would stop the row surviving the next mirror-reload. + db.coreAddressDao().upsert( + org.dashfoundation.dashsdk.persistence.entities.CoreAddressEntity( + address = "yOrphanAddr", + publicKey = ByteArray(33), + poolTypeTag = 0, + addressIndex = 0, + derivationPath = "m/44'/1'/0'/0/7", + isUsed = true, + accountId = null, + ), + ) + handler.onWalletChangesetUtxoAdded( + walletId, changeTxid, 22, 31_000L, "yOrphanAddr", byteArrayOf(0x51), 1_400_000, + false, true, false, false, + ) + assertEquals( + "the row must hang off the address projection for this test to mean anything", + "yOrphanAddr", + db.txoDao().getByOutpoint(makeOutpoint(changeTxid, 22))!!.coreAddressId, + ) + + val report = handler.reconcileFromInventory( + walletId, + engineInventoryJson(unspent = emptyList(), spent = emptyList()), + tipHeight = reconcileTip, + ) + + assertEquals(1, report.unresolvedAccount) + assertEquals("an unasked row is not counted as engine-unknown", 0, report.wouldRemove) + } + + @Test + fun reconcileReversePassIsSilentOnConsistentStore() = runTest { + // Rows the engine also holds unspent — including a YOUNG coin the + // insert pass would skip as immature — are consistent, not + // divergence. Every reverse-pass counter must be zero. + handler.onWalletChangesetUtxoAdded( + walletId, changeTxid, 5, 42L, "yTestAddr", byteArrayOf(0x51), reconcileTip - 3, + false, true, false, false, + ) + val json = + """{"utxos":[{"typeTag":0,"standardTag":0,"index":0,""" + + """"txid":"${changeTxid.toHexLower()}","vout":5,"amount":42,""" + + """"address":"yTestAddr","scriptHex":"51",""" + + """"height":${reconcileTip - 3},"isConfirmed":true,""" + + """"isInstantlocked":false,"isCoinbase":false,"isLocked":false}],"spent":[]}""" + val report = handler.reconcileFromInventory(walletId, json, tipHeight = reconcileTip) + assertEquals(0, report.wouldFlipSpent) + assertEquals(0, report.wouldRemove) + assertEquals(1, report.skippedImmature) + assertFalse(db.txoDao().getByOutpoint(makeOutpoint(changeTxid, 5))!!.isSpent) + } + + @Test + fun reconcileExcludesWatchOnlyContactRowsFromReversePass() = runTest { + // Watch-only DIP-15 contact rows are never in the engine's + // inventories; flagging them would be a false positive on every + // wallet with contact payments. + db.walletDao().upsert(WalletEntity(walletId, networkRaw = Network.TESTNET.ffiValue)) + val foreignAccountId = db.accountDao().insert( + org.dashfoundation.dashsdk.persistence.entities.AccountEntity( + walletId = walletId, + accountType = PlatformWalletPersistenceHandler.ACCOUNT_TYPE_TAG_DASHPAY_EXTERNAL, + accountIndex = 0, + accountTypeName = "DashpayExternalAccount", + ), + ) + handler.onWalletChangesetUtxoAdded( + walletId, changeTxid, 6, 1_230_000L, "yContactAddr", byteArrayOf(0x51), 1_400_000, + false, true, false, false, + ) + val seeded = db.txoDao().getByOutpoint(makeOutpoint(changeTxid, 6))!! + db.txoDao().upsert(seeded.copy(accountId = foreignAccountId)) + + val report = handler.reconcileFromInventory( + walletId, + engineInventoryJson(unspent = emptyList(), spent = emptyList()), + tipHeight = reconcileTip, + ) + assertEquals(1, report.skippedForeign) + assertEquals(0, report.wouldRemove) + assertFalse(db.txoDao().getByOutpoint(makeOutpoint(changeTxid, 6))!!.isSpent) + } + + @Test + fun reconcileResolvesContactOwnershipThroughCoreAddressId() = runTest { + // Production changeset writes leave txos.accountId null and route + // ownership through coreAddressId -> core_addresses.accountId. The + // exclusion must resolve that path, or every contact row gets + // classified as divergence. + db.walletDao().upsert(WalletEntity(walletId, networkRaw = Network.TESTNET.ffiValue)) + val foreignAccountId = db.accountDao().insert( + org.dashfoundation.dashsdk.persistence.entities.AccountEntity( + walletId = walletId, + accountType = PlatformWalletPersistenceHandler.ACCOUNT_TYPE_TAG_DASHPAY_EXTERNAL, + accountIndex = 1, + accountTypeName = "DashpayExternalAccount", + ), + ) + db.coreAddressDao().upsert( + org.dashfoundation.dashsdk.persistence.entities.CoreAddressEntity( + address = "yContactRouted", + publicKey = ByteArray(33), + poolTypeTag = 0, + addressIndex = 0, + derivationPath = "m/9'/1'/15'/0'/x/y/0", + isUsed = true, + accountId = foreignAccountId, + ), + ) + handler.onWalletChangesetUtxoAdded( + walletId, changeTxid, 8, 990_000L, "yContactRouted", byteArrayOf(0x51), 1_400_000, + false, true, false, false, + ) + val seeded = db.txoDao().getByOutpoint(makeOutpoint(changeTxid, 8))!! + assertNull("production shape: accountId is null", seeded.accountId) + + val report = handler.reconcileFromInventory( + walletId, + engineInventoryJson(unspent = emptyList(), spent = emptyList()), + tipHeight = reconcileTip, + ) + assertEquals(1, report.skippedForeign) + assertEquals(0, report.wouldRemove) + } + + @Test + fun reconcileInsertPassNeverHealsContactAccountCoins() = runTest { + // The engine's UTXO inventory export includes the watch-only DIP-15 + // external accounts' coins — it tracks them to show payments TO + // contacts, but they are the CONTACT's money. With the foreign + // exclusion living only on the reverse pass, a fresh restore's + // post-backfill reconcile healed every contact-payment coin into the + // store as an ownerless row (12 rows / 5,692,493 duffs on the + // 2026-08-25 large-wallet validation run) while the reverse pass + // counted the very same rows as foreign — and the mirror-reload path + // hands such rows back to the engine on the next launch. The insert + // pass must skip any engine UTXO whose address resolves to an + // external account, and count it as foreign, not healed. + db.walletDao().upsert(WalletEntity(walletId, networkRaw = Network.TESTNET.ffiValue)) + val foreignAccountId = db.accountDao().insert( + org.dashfoundation.dashsdk.persistence.entities.AccountEntity( + walletId = walletId, + accountType = PlatformWalletPersistenceHandler.ACCOUNT_TYPE_TAG_DASHPAY_EXTERNAL, + accountIndex = 2, + accountTypeName = "DashpayExternalAccount", + ), + ) + db.coreAddressDao().upsert( + org.dashfoundation.dashsdk.persistence.entities.CoreAddressEntity( + address = "yContactPaid", + publicKey = ByteArray(33), + poolTypeTag = 0, + addressIndex = 0, + derivationPath = "m/9'/1'/15'/0'/x/y/1", + isUsed = true, + accountId = foreignAccountId, + ), + ) + + val json = + """{"utxos":[{"typeTag":0,"standardTag":0,"index":0,""" + + """"txid":"${changeTxid.toHexLower()}","vout":9,"amount":10000,""" + + """"address":"yContactPaid","scriptHex":"51",""" + + """"height":1400000,"isLocked":false}],"spent":[]}""" + val report = handler.reconcileFromInventory(walletId, json, tipHeight = reconcileTip) + + assertEquals(0, report.inserted) + assertEquals(0L, report.insertedDuffs) + assertEquals(1, report.skippedForeign) + assertNull( + "the contact's coin must not enter the store", + db.txoDao().getByOutpoint(makeOutpoint(changeTxid, 9)), + ) + } + + @Test + fun reconcileStampsResolvedAccountOnHealedRows() = runTest { + // The blocking scenario from review: persistence lost BOTH the TXO + // and its address row. The heal must resolve the owning Room account + // from the inventory's account tuple and stamp it on the inserted + // row — a healed row with neither accountId nor a resolvable address + // is skipped by the restore loader at the next mirror-reload, + // recreating the fund loss the heal repaired. + db.walletDao().upsert(WalletEntity(walletId, networkRaw = Network.TESTNET.ffiValue)) + // Production shape: onPersistAccountRegistration stores the FFI's + // 32-zero-byte identity ids verbatim (the entity ctor default of an + // EMPTY array never occurs on persisted rows). + val bip44Id = db.accountDao().insert( + org.dashfoundation.dashsdk.persistence.entities.AccountEntity( + walletId = walletId, + accountType = 0, + accountIndex = 0, + accountTypeName = "standardBip44", + userIdentityId = ByteArray(32), + friendIdentityId = ByteArray(32), + ), + ) + // Deliberately NO core_addresses row for this address. + val report = handler.reconcileFromInventory( + walletId, + engineUtxoJson(changeTxid.toHexLower(), vout = 11, amount = 70_000L, address = "yOrphanAddr"), + tipHeight = reconcileTip, + ) + assertEquals(1, report.inserted) + assertEquals(0, report.healedUnowned) + assertEquals( + "the healed row must carry the account resolved from the inventory tuple", + bip44Id, db.txoDao().getByOutpoint(makeOutpoint(changeTxid, 11))!!.accountId, + ) + } + + @Test + fun reconcileCountsHealsWhoseAccountCannotBeResolved() = runTest { + // No matching Room account row at all (a store damaged past the + // account registrations): the heal proceeds — the address projection + // may still attribute it — but the unresolved owner is surfaced. + db.walletDao().upsert(WalletEntity(walletId, networkRaw = Network.TESTNET.ffiValue)) + val report = handler.reconcileFromInventory( + walletId, + engineUtxoJson(changeTxid.toHexLower(), vout = 12, amount = 5_000L), + tipHeight = reconcileTip, + ) + assertEquals(1, report.inserted) + assertEquals(1, report.healedUnowned) + assertNull(db.txoDao().getByOutpoint(makeOutpoint(changeTxid, 12))!!.accountId) + } + + @Test + fun reconcileForeignSkipKeysOffTheInventoryTagWithoutAddressRow() = runTest { + // The tag is the authoritative foreign check: a contact's coin must + // be skipped even when its address row never survived persistence + // (the case the address-based fallback cannot see). + val json = + """{"utxos":[{"typeTag":13,"standardTag":0,"index":0,""" + + """"userIdentityId":"${"11".repeat(32)}","friendIdentityId":"${"22".repeat(32)}",""" + + """"txid":"${changeTxid.toHexLower()}","vout":13,"amount":10000,""" + + """"address":"yContactNoRow","scriptHex":"51",""" + + """"height":1400000,"isLocked":false}],"spent":[]}""" + val report = handler.reconcileFromInventory(walletId, json, tipHeight = reconcileTip) + + assertEquals(0, report.inserted) + assertEquals(1, report.skippedForeign) + assertNull(db.txoDao().getByOutpoint(makeOutpoint(changeTxid, 13))) + } + + @Test + fun reconcileNeverUnmarksSpentRowsEvenWhenEngineDisagrees() = runTest { + // A row marked spent while the engine lists the coin unspent: either + // a lost release event (pre-rust-dashcore#971) or a live spend the + // store wrote before the engine settled. Un-marking a coin + // mid-payment would let the wallet double-spend it, so this is + // counted and logged but NEVER changed. + handler.onWalletChangesetUtxoAdded( + walletId, changeTxid, 7, 77_000L, "yTestAddr", byteArrayOf(0x51), 1_400_000, + false, true, false, false, + ) + val seeded = db.txoDao().getByOutpoint(makeOutpoint(changeTxid, 7))!! + db.txoDao().upsert(seeded.copy(isSpent = true)) + + val report = handler.reconcileFromInventory( + walletId, + engineInventoryJson( + unspent = listOf(Triple(changeTxid.toHexLower(), 7, 77_000L)), + spent = emptyList(), + ), + tipHeight = reconcileTip, + ) + assertEquals(1, report.stuckSpent) + assertEquals(77_000L, report.stuckSpentDuffs) + assertTrue( + "the row must stay spent — un-marking is never done by reconciliation", + db.txoDao().getByOutpoint(makeOutpoint(changeTxid, 7))!!.isSpent, + ) + } + + // ── Paged reconcile transport ───────────────────────────────────── + + @Test + fun reconcileWalksEveryPageOfTheEngineInventory() = runTest { + // The engine inventory is chain-controlled in size, so the sweep + // reads it a page at a time. Every page must be applied — a sweep + // that healed only the first one would leave most of a damaged + // mirror unrepaired, and silently. + val engine = FakeEngine( + engineInventoryJson( + unspent = (0 until 5).map { Triple(changeTxid.toHexLower(), 20 + it, 1_000L) }, + spent = emptyList(), + ), + ) + val report = handler.reconcileTxos( + walletId = walletId, + tipHeight = reconcileTip, + pageSize = 2, + engineUtxoPage = engine::page, + classifyOutpoints = engine::classify, + )!! + + assertEquals("3 pages for 5 rows at 2 per page", 3, engine.pages) + assertEquals(5, report.engineUtxos) + assertEquals(5, report.inserted) + assertEquals(5_000L, report.insertedDuffs) + for (vout in 20 until 25) { + assertNotNull( + "the row at vout=$vout must be healed whichever page carried it", + db.txoDao().getByOutpoint(makeOutpoint(changeTxid, vout)), + ) + } + } + + @Test + fun reconcileStopsAtAFailedPageAndKeepsWhatItAlreadyHealed() = runTest { + // A transport that dies mid-sweep must not discard the pages that + // already landed — the pass is insert-only and idempotent, so they + // are already correct — and must not report a clean run either. + val engine = FakeEngine( + engineInventoryJson( + unspent = (0 until 4).map { Triple(changeTxid.toHexLower(), 30 + it, 500L) }, + spent = emptyList(), + ), + ) + val report = handler.reconcileTxos( + walletId = walletId, + tipHeight = reconcileTip, + pageSize = 2, + engineUtxoPage = { cursor, limit -> + if (cursor == null) engine.page(cursor, limit) else null + }, + classifyOutpoints = engine::classify, + )!! + + assertEquals(1, report.transportFailures) + assertEquals("only the page that arrived", 2, report.inserted) + assertNotNull(db.txoDao().getByOutpoint(makeOutpoint(changeTxid, 30))) + assertNull(db.txoDao().getByOutpoint(makeOutpoint(changeTxid, 32))) + } + + @Test + fun reconcileClassifiesStoreRowsInBoundedBatches() = runTest { + // The reverse direction is inverted: the STORE is paged and the + // engine is asked about one page at a time, so neither side builds + // a set over a whole inventory. Every batch must still be answered + // and counted. + for (vout in 40 until 43) { + handler.onWalletChangesetUtxoAdded( + walletId, changeTxid, vout, 100L, "yTestAddr", byteArrayOf(0x51), 1_400_000, + false, true, false, false, + ) + } + // These rows must carry a resolvable owner: a query names the + // account the store filed the coin under, and a row without one is + // counted rather than sent to the classifier. + attributeUnownedTxos(walletId) + val engine = FakeEngine(engineInventoryJson(unspent = emptyList(), spent = emptyList())) + val report = handler.reconcileTxos( + walletId = walletId, + tipHeight = reconcileTip, + pageSize = 1, + engineUtxoPage = engine::page, + classifyOutpoints = engine::classify, + )!! + + assertEquals("one classification batch per store page", 3, engine.batches) + assertEquals("every store row reached the classifier", 3, engine.classified) + assertEquals(3, report.wouldRemove) + assertEquals(300L, report.wouldRemoveDuffs) + } + + @Test + fun reconcileStopsWhenAClassificationBatchFails() = runTest { + // No verdicts, no classification: the reverse pass stops rather + // than guessing at rows it could not ask the engine about. + for (vout in 50 until 53) { + handler.onWalletChangesetUtxoAdded( + walletId, changeTxid, vout, 100L, "yTestAddr", byteArrayOf(0x51), 1_400_000, + false, true, false, false, + ) + } + // These rows must carry a resolvable owner: a query names the + // account the store filed the coin under, and a row without one is + // counted rather than sent to the classifier. + attributeUnownedTxos(walletId) + val engine = FakeEngine(engineInventoryJson(unspent = emptyList(), spent = emptyList())) + val report = handler.reconcileTxos( + walletId = walletId, + tipHeight = reconcileTip, + pageSize = 1, + engineUtxoPage = engine::page, + classifyOutpoints = { null }, + )!! + + assertEquals(1, report.transportFailures) + assertEquals(0, report.wouldRemove) + } + + @Test + fun reconcileReturnsNoReportWhenTheFirstPageIsUnavailable() = runTest { + // No first page means nothing to reconcile against — the contract + // callers had before the sweep was paged, now enforced inside the + // handler rather than by a prefetch in the caller. + var classifyCalls = 0 + val report = handler.reconcileTxos( + walletId = walletId, + tipHeight = reconcileTip, + engineUtxoPage = { _, _ -> null }, + classifyOutpoints = { queriesJson -> + classifyCalls++ + ByteArray( + kotlinx.serialization.json.Json.parseToJsonElement(queriesJson).jsonArray.size, + ) + }, + ) + assertNull("a dead transport yields no report", report) + assertEquals("and the classification pass never runs", 0, classifyCalls) + } + + @Test + fun reconcileRejectsAMalformedInventoryRowInsteadOfHealingIt() = runTest { + // The transport is a typed contract on both ends. A row missing its + // amount (or carrying a key this reader does not know) must fail the + // decode loudly — the alternative is a defaulted row written into + // the mirror the engine reloads from. + val malformed = """{"utxos":[{"typeTag":0,"txid":"${changeTxid.toHexLower()}",""" + + """"vout":40,"address":"yStxXHHzhAx58JhaPBNhn3xsH93UwBM2nd","height":1400000}],""" + + """"cursor":null,"hasMore":false}""" + val thrown = runCatching { + handler.reconcileTxos( + walletId = walletId, + tipHeight = reconcileTip, + engineUtxoPage = { _, _ -> malformed }, + classifyOutpoints = { queriesJson -> + ByteArray( + kotlinx.serialization.json.Json + .parseToJsonElement(queriesJson).jsonArray.size, + ) + }, + ) + }.exceptionOrNull() + assertTrue( + "strict decode must throw, got $thrown", + thrown is kotlinx.serialization.SerializationException, + ) + assertNull( + "nothing was healed from the malformed page", + db.txoDao().getByOutpoint(makeOutpoint(changeTxid, 40)), + ) + } + + /** + * Drive the paged reconcile from one whole-inventory JSON blob — the + * shape these tests describe an engine in, and the shape the native + * side used to hand over in a single unbounded call. + * + * The blob is served the way the transport now serves it: sliced into + * bounded pages behind an opaque cursor, with a separate positional + * classifier for the outpoints the store asks about. [pageSize] + * defaults to 2, so a test describing more than a couple of rows walks + * the real cursor loop rather than a single page. + */ + private suspend fun PlatformWalletPersistenceHandler.reconcileFromInventory( + walletId: ByteArray, + inventoryJson: String, + tipHeight: Int, + minConfirmations: Int = 100, + pageSize: Int = 2, + ): PlatformWalletPersistenceHandler.TxoReconcileReport { + attributeUnownedTxos(walletId) + val engine = FakeEngine(inventoryJson) + return checkNotNull( + reconcileTxos( + walletId = walletId, + tipHeight = tipHeight, + minConfirmations = minConfirmations, + pageSize = pageSize, + engineUtxoPage = engine::page, + classifyOutpoints = engine::classify, + ), + ) { "the fake engine always serves a first page" } + } + + /** + * File every still-unattributed TXO of [walletId] under a default + * BIP44 account, seeding that account on first use. + * + * The classification pass asks the engine about a row by naming the + * account the STORE filed it under — the engine checks that claim + * against its own pools rather than trusting it — so a row with no + * resolvable owner carries no claim and is counted instead of asked + * about. Production rows always resolve one, through the explicit link + * or the `core_addresses` projection; these tests write TXOs straight + * through the changeset callbacks without the account rows a real + * wallet load creates, so the link is seeded here rather than in each + * test. Rows a test attributed itself (the watch-only contact case) + * keep their own account. + */ + private suspend fun attributeUnownedTxos(walletId: ByteArray): Long { + db.walletDao().upsert(WalletEntity(walletId, networkRaw = Network.TESTNET.ffiValue)) + val existing = db.accountDao().getByKey(walletId, accountType = 0, accountIndex = 0) + val ownerId = existing.firstOrNull()?.id + ?: db.accountDao().insert( + org.dashfoundation.dashsdk.persistence.entities.AccountEntity( + walletId = walletId, + accountType = 0, + accountIndex = 0, + accountTypeName = "Standard", + ), + ) + for (row in db.txoDao().pageByWallet(walletId, ByteArray(0), limit = 1_000)) { + if (row.accountId == null && row.coreAddressId == null) { + db.txoDao().upsert(row.copy(accountId = ownerId)) + } + } + return ownerId + } + + /** + * A stand-in for the engine's paged inventory transport, built from the + * whole-inventory JSON a test writes out. Pages come back behind an + * opaque ordinal cursor — the real cursor is opaque too, the handler + * only ever hands back what it was given — and classification answers + * positionally out of the same two inventories: 1 unspent, + * 2 known-uncredited (the blob's `spent` list: the engine holds a + * mined record spending the coin), 3 not-owned (its optional + * `notOwned` list), 0 neither. + */ + private class FakeEngine(inventoryJson: String) { + private val utxos: List + private val unspentKeys: Set + private val spentKeys: Set + private val notOwnedKeys: Set + + /** Inventory pages served, classification batches answered, and + * outpoints classified across those batches. */ + var pages = 0 + private set + var batches = 0 + private set + var classified = 0 + private set + + init { + val root = kotlinx.serialization.json.Json + .parseToJsonElement(inventoryJson).jsonObject + utxos = root["utxos"]?.jsonArray?.map { it.jsonObject } ?: emptyList() + unspentKeys = utxos.map { + key( + it["txid"]!!.jsonPrimitive.content, + it["vout"]!!.jsonPrimitive.int, + ) + }.toSet() + spentKeys = (root["spent"]?.jsonArray?.toList() ?: emptyList()).map { + key( + it.jsonObject["txid"]!!.jsonPrimitive.content, + it.jsonObject["vout"]!!.jsonPrimitive.int, + ) + }.toSet() + notOwnedKeys = (root["notOwned"]?.jsonArray?.toList() ?: emptyList()).map { + key( + it.jsonObject["txid"]!!.jsonPrimitive.content, + it.jsonObject["vout"]!!.jsonPrimitive.int, + ) + }.toSet() + } + + fun page(cursor: String?, limit: Int): String { + pages++ + val start = cursor?.toInt() ?: 0 + val slice = utxos.drop(start).take(limit) + val next = start + slice.size + val hasMore = next < utxos.size + return """{"utxos":[${slice.joinToString(",") { it.toString() }}],""" + + """"cursor":${if (hasMore) "\"$next\"" else "null"},"hasMore":$hasMore}""" + } + + fun classify(queriesJson: String): ByteArray { + batches++ + val queries = kotlinx.serialization.json.Json + .parseToJsonElement(queriesJson).jsonArray + classified += queries.size + val verdicts = ByteArray(queries.size) + for ((i, query) in queries.withIndex()) { + val q = query.jsonObject + // Every query must name the account the store filed the + // coin under and the script it recorded — that claim is + // the whole reason the engine can answer NOT_OWNED. + requireNotNull(q["typeTag"]) { "classification query carries no account tag" } + requireNotNull(q["scriptHex"]) { "classification query carries no script" } + val k = key( + q["txid"]!!.jsonPrimitive.content, + q["vout"]!!.jsonPrimitive.int, + ) + verdicts[i] = when { + k in notOwnedKeys -> 3 + k in unspentKeys -> 1 + k in spentKeys -> 2 + else -> 0 + } + } + return verdicts + } + + private fun key(txidHex: String, vout: Int) = "$txidHex:$vout" + + private companion object { + /** txid (32 bytes, wire order) + vout (4 bytes, little-endian). */ + const val OUTPOINT_SIZE = 36 + } + } } /** diff --git a/packages/rs-platform-wallet-ffi/src/persistence.rs b/packages/rs-platform-wallet-ffi/src/persistence.rs index 7491999e283..4bc4389cef1 100644 --- a/packages/rs-platform-wallet-ffi/src/persistence.rs +++ b/packages/rs-platform-wallet-ffi/src/persistence.rs @@ -4549,6 +4549,7 @@ unsafe fn restore_core_address_pools( pool_entries: &[AccountAddressPoolFFI], network: Network, wallet_id: &[u8; 32], + signing_wallet: Option<&Wallet>, ) -> Result { use key_wallet::managed_account::managed_account_trait::ManagedAccountTrait; let mut pools_routed = 0usize; @@ -4735,11 +4736,111 @@ unsafe fn restore_core_address_pools( } } } + // Resolve the pool's key source BEFORE taking the mutable pool + // borrow, for the hole-repair pass below. Degrades to NoKeySource + // for anything unresolvable (no signing wallet handle, an account + // this wallet does not carry, a pool with no public derivation at + // all) — repair is then skipped. + // + // The concrete account comes first, because it is the only source + // that covers DashPay. `key_source_for_account_type` routes through + // `extended_public_key_for_account_type`, which has no arm for + // either DashPay variant and so answers NoKeySource for both — yet + // `DashpayReceivingFunds` is wallet-owned, funds-bearing, and its + // account carries exactly the xpub the missing addresses derive + // from. Skipping its repair leaves a contact's payments to us + // unrecognizable in the same rescan-proof way the repair exists to + // prevent. The helper stays as the fallback for the account types + // `account_of_type` deliberately does not return: the BLS provider + // operator account, whose key source is a BLS public key rather + // than an xpub, and the Ed25519 platform-node account. + let key_source = signing_wallet + .and_then(|wallet| { + wallet + .accounts + .account_of_type(account_type) + .map(|account| key_wallet::KeySource::Public(account.account_xpub)) + .or_else(|| { + key_wallet::transaction_checking::transaction_router::AccountTypeToCheck::try_from( + &*managed_type, + ) + .ok() + .map(|check_type| { + let account_index = match &account_type { + AccountType::Standard { index, .. } + | AccountType::CoinJoin { index } + | AccountType::DashpayReceivingFunds { index, .. } + | AccountType::DashpayExternalAccount { index, .. } => Some(*index), + AccountType::IdentityTopUp { registration_index } => { + Some(*registration_index) + } + _ => None, + }; + wallet.key_source_for_account_type(&check_type, account_index) + }) + }) + }) + .unwrap_or(key_wallet::KeySource::NoKeySource); + let mut managed_pools = managed_type.address_pools_mut(); match managed_pools.iter_mut().find(|p| p.pool_type == pool_type) { Some(pool) => { pools_routed += infos.len(); restore_address_pool(pool, infos); + // Hole repair: mirrors have been observed dropping address + // rows (2026-08-19 field wallet: BIP44-change rows 875..=890 + // absent between surviving rows), and ingesting the sparse + // list as-is makes outputs paying the missing addresses + // permanently unrecognizable — a rescan-proof fund loss — + // while the row-derived `highest_generated` suppresses the + // gap-limit re-derivation that would repair it. Derivation + // is pure key arithmetic, so re-derive every missing index + // up to the persisted watermark. Never fatal: a failed + // repair restores exactly what the rows carried (the + // pre-repair behavior). + let repairable = !matches!(key_source, key_wallet::KeySource::NoKeySource) + && !matches!(pool_type, AddressPoolType::AbsentHardened); + if !repairable { + // Announce the skip instead of silently claiming full + // coverage. What lands here now is a pool with no public + // derivation at all: a hardened pool, the Ed25519 + // platform-node account, or an account this wallet does + // not carry (a watch-only load with no signing wallet + // handle). DashPay pools no longer land here — they + // resolve through their own account's xpub above. + tracing::debug!( + wallet_id = %hex::encode(wallet_id), + ?account_type, + ?pool_type, + "load: address-pool hole repair skipped (no public key source); \ + pool restored as persisted" + ); + } else if let Some(max_idx) = pool.highest_generated { + match pool.ensure_contiguous_to(max_idx, &key_source) { + Ok(0) => {} + Ok(filled) => { + tracing::warn!( + wallet_id = %hex::encode(wallet_id), + ?account_type, + ?pool_type, + filled, + "load: repaired address-pool holes left by dropped \ + persisted rows; outputs paying these addresses are \ + recognizable again" + ); + } + Err(e) => { + tracing::warn!( + wallet_id = %hex::encode(wallet_id), + ?account_type, + ?pool_type, + error = %e, + "load: address-pool hole repair failed; pool restored \ + as persisted (sparse)" + ); + } + } + } } None => { pools_dropped += 1; @@ -5219,7 +5320,13 @@ fn build_wallet_start_state( // SAFETY: `pool_entries` is a valid slice (checked above) and each // row's `addresses_ptr` follows the load-callback contract. unsafe { - restore_core_address_pools(&mut wallet_info, pool_entries, network, &entry.wallet_id)?; + restore_core_address_pools( + &mut wallet_info, + pool_entries, + network, + &entry.wallet_id, + Some(&wallet), + )?; } } @@ -8642,6 +8749,165 @@ mod tests { ManagedWalletInfo::from_wallet(&wallet, 0) } + /// A wallet carrying exactly one account of `account_type`, returned + /// alongside its managed view so a test can hand the SIGNING wallet to + /// [`restore_core_address_pools`] — the handle the hole repair resolves + /// a key source from. + fn test_wallet_and_info_with_account(account_type: AccountType) -> (ManagedWalletInfo, Wallet) { + let mnemonic = Mnemonic::from_phrase( + "abandon abandon abandon abandon abandon abandon abandon abandon abandon abandon abandon about", + ) + .expect("static BIP-39 vector must parse"); + let seed = mnemonic.to_seed(""); + let master = ExtendedPrivKey::new_master(Network::Testnet, &seed) + .expect("master derivation must succeed"); + let secp = Secp256k1::new(); + let xpub = ExtendedPubKey::from_priv(&secp, &master); + let account = Account::from_xpub(None, account_type, xpub, Network::Testnet) + .expect("Account::from_xpub on a valid xpub must succeed"); + let mut accounts = key_wallet::AccountCollection::new(); + accounts + .insert(account) + .expect("inserting the single account must succeed"); + let wallet = Wallet::new_external_signable(Network::Testnet, [0u8; 32], accounts); + let info = ManagedWalletInfo::from_wallet(&wallet, 0); + (info, wallet) + } + + /// The hole repair must cover DashPay receiving pools. + /// + /// `Wallet::key_source_for_account_type` answers `NoKeySource` for both + /// DashPay variants — it routes through + /// `extended_public_key_for_account_type`, which has no DashPay arm — so + /// resolving the key source through that helper alone skipped the repair + /// on exactly the account type that needs it most: a receiving account is + /// wallet-owned and funds-bearing, and an unrepaired hole makes a + /// contact's payments to us unrecognizable in a way no rescan fixes. The + /// concrete account carries the xpub the missing addresses derive from, + /// so the resolver consults `accounts.account_of_type` first. + /// + /// Shape: a persisted row at index 50 lifts `highest_generated` far past + /// the pre-derived gap window, leaving every index between as a hole. + #[test] + fn dashpay_receiving_pool_holes_are_repaired_from_the_account_xpub() { + use key_wallet::managed_account::managed_account_trait::ManagedAccountTrait; + use std::ffi::CString; + + let account_type = AccountType::DashpayReceivingFunds { + index: 2, + user_identity_id: [0x11u8; 32], + friend_identity_id: [0x22u8; 32], + }; + let (mut wallet_info, wallet) = test_wallet_and_info_with_account(account_type); + + let dashpay_key = key_wallet::account::account_collection::DashpayAccountKey { + index: 2, + user_identity_id: [0x11u8; 32], + friend_identity_id: [0x22u8; 32], + }; + let pool_type = wallet_info + .accounts + .dashpay_receival_accounts + .get_mut(&dashpay_key) + .expect("managed DashPay receiving account must exist") + .managed_account_type_mut() + .address_pools_mut()[0] + .pool_type; + let pool_type_tag: u8 = match pool_type { + AddressPoolType::External => 0, + AddressPoolType::Internal => 1, + AddressPoolType::Absent => 2, + AddressPoolType::AbsentHardened => 3, + }; + + const RESTORED_INDEX: u32 = 50; + // The gap window the fresh account pre-derived. Everything above it + // and below the restored row is a hole the persisted rows do not + // carry. + let pre_derived_top = { + let account = wallet_info + .accounts + .dashpay_receival_accounts + .get_mut(&dashpay_key) + .expect("managed DashPay receiving account must exist"); + let mut pools = account.managed_account_type_mut().address_pools_mut(); + let pool = pools + .iter_mut() + .find(|p| p.pool_type == pool_type) + .expect("the DashPay receiving pool must exist"); + pool.highest_generated + .expect("a fresh pool pre-derives its gap window") + }; + assert!( + pre_derived_top < RESTORED_INDEX - 1, + "the fixture only means something if the restored row leaves holes \ + (pre-derived to {pre_derived_top}, restoring {RESTORED_INDEX})" + ); + + let addr_c = CString::new("yMqShkrgjTRuReBGFpQr7FozEF1QcNBBYA").unwrap(); + let path_c = CString::new("m/9'/1'/15'/50").unwrap(); + let row = CoreAddressEntryFFI { + public_key: [0u8; 48], + public_key_len: 0, + key_type_tag: 0, + pool_type_tag, + address_index: RESTORED_INDEX, + is_used: true, + balance: 0, + address_base58: addr_c.as_ptr(), + derivation_path: path_c.as_ptr(), + }; + let no_xpub: &[u8] = &[]; + let pools = [AccountAddressPoolFFI { + account: build_account_spec_ffi(&account_type, no_xpub), + pool_type_tag, + addresses_ptr: &row, + addresses_count: 1, + }]; + + // SAFETY: `row` / `addr_c` / `path_c` outlive the call below. + let stats = unsafe { + restore_core_address_pools( + &mut wallet_info, + &pools, + Network::Testnet, + &[0u8; 32], + Some(&wallet), + ) + } + .expect("restore must succeed for a well-formed DashPay pool"); + assert_eq!( + stats, + PoolRestoreStats { + routed: 1, + dropped: 0 + }, + "the DashPay row must route into the managed pool, not drop" + ); + + let account = wallet_info + .accounts + .dashpay_receival_accounts + .get_mut(&dashpay_key) + .expect("managed DashPay receiving account must exist"); + let mut pools_mut = account.managed_account_type_mut().address_pools_mut(); + let restored = pools_mut + .iter_mut() + .find(|p| p.pool_type == pool_type) + .expect("the DashPay receiving pool must exist"); + assert!( + restored.used_indices.contains(&RESTORED_INDEX), + "the used index must be restored into the pool" + ); + let holes: Vec = (0..=RESTORED_INDEX) + .filter(|i| !restored.addresses.contains_key(i)) + .collect(); + assert!( + holes.is_empty(), + "every index up to the watermark must be derivable again; holes left: {holes:?}" + ); + } + /// Restore-arm coverage (PR #4120): a persisted core-address-pool row /// targeting a PROVIDER account (`ProviderOwnerKeys`) must rehydrate /// its used-flag + beyond-gap index into @@ -8698,7 +8964,7 @@ mod tests { // SAFETY: `row` / `addr_c` / `path_c` outlive the call below. let stats = unsafe { - restore_core_address_pools(&mut wallet_info, &pools, Network::Testnet, &[0u8; 32]) + restore_core_address_pools(&mut wallet_info, &pools, Network::Testnet, &[0u8; 32], None) } .expect("restore must succeed for a well-formed provider pool"); assert_eq!( diff --git a/packages/rs-unified-sdk-jni/Cargo.toml b/packages/rs-unified-sdk-jni/Cargo.toml index e2604b7ab2d..b3ea2782f9e 100644 --- a/packages/rs-unified-sdk-jni/Cargo.toml +++ b/packages/rs-unified-sdk-jni/Cargo.toml @@ -18,6 +18,14 @@ platform-wallet-ffi = { path = "../rs-platform-wallet-ffi" } key-wallet-ffi = { workspace = true } dash-network = { workspace = true, features = ["ffi"] } log = "0.4" +# Serialization for the TXO-reconcile transport (the paged wallet +# inventory and the outpoint-classification batch). Both exports are +# shims over one platform-wallet call each: serde shapes the JSON +# against the same field names the Kotlin side decodes, so a renamed +# key fails the decode loudly instead of defaulting a row. +serde = { version = "1", features = ["derive"] } +serde_json = "1" +hex = "0.4" zeroize = "1" [target.'cfg(target_os = "android")'.dependencies] diff --git a/packages/rs-unified-sdk-jni/src/wallet_manager.rs b/packages/rs-unified-sdk-jni/src/wallet_manager.rs index 3a0b2be5ffe..52afd70b452 100644 --- a/packages/rs-unified-sdk-jni/src/wallet_manager.rs +++ b/packages/rs-unified-sdk-jni/src/wallet_manager.rs @@ -3218,6 +3218,443 @@ fn core_selection_strategy( } } + +// ── TXO-store reconcile transport ───────────────────────────────────── +// +// Two exports, each a serialization shim over exactly ONE platform-wallet +// call: `platform_wallet_wallet_utxos_page` and +// `platform_wallet_classify_outpoints`. The account ordering, the cursor +// semantics, the page bound and every classification rule live in +// `platform-wallet`, so the Swift host walks the identical inventory and +// gets the identical verdicts. Nothing here decides anything; it moves +// serde-shaped JSON in and out. + +/// The account tuple that owns one inventory row — the raw `AccountSpecFFI` +/// tag layout minus the xpub, in the field names the Kotlin `EngineUtxoRow` +/// and `PlatformWalletPersistenceHandler.fetchAccount` resolve a Room +/// account by. The DashPay identity halves are emitted only when set +/// (all-zero on every non-DashPay account); the Kotlin side defaults them. +#[derive(serde::Serialize, serde::Deserialize)] +#[serde(rename_all = "camelCase")] +struct UtxoAccountTuple { + type_tag: u8, + standard_tag: u8, + index: u32, + registration_index: u32, + key_class: u32, + #[serde(default, skip_serializing_if = "Option::is_none")] + user_identity_id: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + friend_identity_id: Option, +} + +impl UtxoAccountTuple { + fn from_entry(e: &platform_wallet_ffi::WalletUtxoEntryFFI) -> Self { + let identity = |id: &[u8; 32]| (*id != [0u8; 32]).then(|| hex::encode(id)); + UtxoAccountTuple { + type_tag: e.type_tag as u8, + standard_tag: e.standard_tag as u8, + index: e.index, + registration_index: e.registration_index, + key_class: e.key_class, + user_identity_id: identity(&e.user_identity_id), + friend_identity_id: identity(&e.friend_identity_id), + } + } + + /// The two 32-byte identity halves, or `None` when a hex string the + /// host handed back is malformed — a tuple this layer did not emit. + fn identity_ids(&self) -> Option<([u8; 32], [u8; 32])> { + fn id32(hex_id: &Option) -> Option<[u8; 32]> { + match hex_id { + None => Some([0u8; 32]), + Some(h) => hex::decode(h).ok()?.try_into().ok(), + } + } + Some((id32(&self.user_identity_id)?, id32(&self.friend_identity_id)?)) + } +} + +/// One row of an inventory page — the typed contract the Kotlin +/// `EngineUtxoRow` decodes, strictly (an unknown or missing key throws +/// there rather than healing a defaulted row). `txid` is lower hex in the +/// same byte order the changeset path hands Kotlin, so hex→bytes +/// reproduces the `txos.txid` blob; `address` is the engine's own +/// Base58Check rendering, empty when the script has no address form. +#[derive(serde::Serialize)] +#[serde(rename_all = "camelCase")] +struct UtxoPageRow { + #[serde(flatten)] + account: UtxoAccountTuple, + txid: String, + vout: u32, + amount: u64, + address: String, + script_hex: String, + height: u32, + is_confirmed: bool, + is_instantlocked: bool, + is_coinbase: bool, + is_locked: bool, +} + +/// One page: the rows, the opaque resume cursor (absent on the last page) +/// and whether more pages follow. +#[derive(serde::Serialize)] +#[serde(rename_all = "camelCase")] +struct UtxoPage { + utxos: Vec, + cursor: Option, + has_more: bool, +} + +/// Where a paged inventory sweep left off — the last row's account tuple +/// and outpoint, which is exactly what `platform_wallet_wallet_utxos_page` +/// resumes from. Serialized as JSON and handed to the host as an opaque +/// string; the host only ever gives back what a previous page returned. +#[derive(serde::Serialize, serde::Deserialize)] +#[serde(rename_all = "camelCase")] +struct UtxoPageCursor { + #[serde(flatten)] + account: UtxoAccountTuple, + txid: String, + vout: u32, +} + +/// One store row the host asks the engine to classify: the account the +/// STORE files the coin under, the outpoint, and the script the store +/// recorded. The engine checks that claim against its own pools rather +/// than trusting it, which is why the tuple and script travel with every +/// query instead of the outpoint alone. +#[derive(serde::Deserialize)] +#[serde(rename_all = "camelCase")] +struct OutpointQuery { + #[serde(flatten)] + account: UtxoAccountTuple, + txid: String, + vout: u32, + #[serde(default)] + script_hex: String, +} + +/// One bounded page of the engine's UTXO inventory across every funds +/// account of one wallet — the source of truth +/// `PlatformWalletManager.reconcileTxoStore` diffs against the Room `txos` +/// mirror (dropped change outputs of CoinJoin-funded sends leave the +/// mirror short; the engine reloads from that mirror on restart, so an +/// un-reconciled hole becomes a fund loss). +/// +/// Paged rather than swept whole because inventory size is +/// chain-controlled: anyone who knows a watched address can keep sending +/// dust outputs to it, and a periodic full-inventory read would let them +/// decide how much a phone allocates at every SYNCED transition and every +/// 30-minute pass. Nothing bigger than one page is ever formatted, copied +/// across JNI, or parsed. +/// +/// Returns a JSON object `{"utxos":[...],"cursor":,"hasMore":}`. +/// `cursor` is opaque: hand it back verbatim on the next call +/// (`null`/absent starts from the beginning) and keep going while +/// `hasMore` is true. A cursor this export did not produce is rejected +/// with an SDK exception rather than silently restarting the sweep. +/// `limit` caps the rows in one page; non-positive passes 0, which means +/// the engine's own default, and the engine clamps its own maximum — this +/// layer holds no copy of either bound. +/// +/// Rows carry the engine's own address and flags. Nothing is derived +/// here: an address computed in this shim could disagree with the one the +/// engine credited the coin under, and the store keys its address rows by +/// that string. +#[no_mangle] +pub extern "system" fn Java_org_dashfoundation_dashsdk_ffi_WalletManagerNative_walletManagerUtxosPageJson( + mut env: JNIEnv, + _class: JClass, + manager_handle: jlong, + wallet_id: JByteArray, + cursor: JString, + limit: jni::sys::jint, +) -> jni::sys::jstring { + guard(&mut env, ptr::null_mut(), |env| { + let Some(wid) = read_id32(env, &wallet_id) else { + return ptr::null_mut(); + }; + // 0 means "the engine's default page"; the engine also clamps its + // own maximum, so a host asking for more simply gets less. + let page_limit = if limit <= 0 { 0usize } else { limit as usize }; + + // Resume point, decoded from the opaque cursor. Kept alive across + // the FFI call — it is passed by pointer. + let resume: Option = if cursor.is_null() { + None + } else { + let raw = match env.get_string(&cursor) { + Ok(s) => String::from(s), + Err(_) => { + throw_sdk_exception(env, 1, "inventory cursor must be a String"); + return ptr::null_mut(); + } + }; + let parsed = serde_json::from_str::(&raw) + .ok() + .and_then(|c| { + let (user_identity_id, friend_identity_id) = c.account.identity_ids()?; + let txid: [u8; 32] = hex::decode(&c.txid).ok()?.try_into().ok()?; + Some(platform_wallet_ffi::WalletUtxoCursorFFI { + type_tag: c.account.type_tag, + standard_tag: c.account.standard_tag, + index: c.account.index, + registration_index: c.account.registration_index, + key_class: c.account.key_class, + user_identity_id, + friend_identity_id, + outpoint: platform_wallet_ffi::OutPointFFI { + txid, + vout: c.vout, + }, + }) + }); + match parsed { + Some(r) => Some(r), + None => { + throw_sdk_exception( + env, + 1, + "malformed inventory cursor: only a cursor returned by a previous page may be handed back", + ); + return ptr::null_mut(); + } + } + }; + + let mut entries: *const platform_wallet_ffi::WalletUtxoEntryFFI = ptr::null(); + let mut count: usize = 0; + let mut has_more = false; + let result = unsafe { + platform_wallet_ffi::platform_wallet_wallet_utxos_page( + manager_handle as Handle, + wid.as_ptr(), + resume.as_ref().map_or(ptr::null(), |c| c as *const _), + page_limit, + &mut entries, + &mut count, + &mut has_more, + ) + }; + if take_pwffi_error(env, result) { + return ptr::null_mut(); + } + let mut rows: Vec = Vec::with_capacity(count); + if !entries.is_null() && count > 0 { + let items = unsafe { std::slice::from_raw_parts(entries, count) }; + for e in items { + let script: &[u8] = if e.script_pubkey.is_null() || e.script_pubkey_len == 0 { + &[] + } else { + unsafe { std::slice::from_raw_parts(e.script_pubkey, e.script_pubkey_len) } + }; + // Never null per the accessor's contract; treated as empty + // if it ever is, which the Kotlin heal skips rather than + // filing a row it cannot key an address by. + let address = if e.address.is_null() { + String::new() + } else { + unsafe { CStr::from_ptr(e.address) } + .to_string_lossy() + .into_owned() + }; + rows.push(UtxoPageRow { + account: UtxoAccountTuple::from_entry(e), + txid: hex::encode(e.outpoint.txid), + vout: e.outpoint.vout, + amount: e.value_duffs, + address, + script_hex: hex::encode(script), + height: e.height, + is_confirmed: e.is_confirmed, + is_instantlocked: e.is_instantlocked, + is_coinbase: e.is_coinbase, + is_locked: e.is_locked, + }); + } + unsafe { + platform_wallet_ffi::platform_wallet_wallet_utxos_page_free( + entries as *mut platform_wallet_ffi::WalletUtxoEntryFFI, + count, + ) + }; + } + // The cursor is the last row itself; a page with no rows has + // nowhere to resume from, and the accessor reports no more in that + // case. + let cursor_json = if has_more { + match rows.last() { + Some(last) => { + let next = UtxoPageCursor { + account: UtxoAccountTuple { + type_tag: last.account.type_tag, + standard_tag: last.account.standard_tag, + index: last.account.index, + registration_index: last.account.registration_index, + key_class: last.account.key_class, + user_identity_id: last.account.user_identity_id.clone(), + friend_identity_id: last.account.friend_identity_id.clone(), + }, + txid: last.txid.clone(), + vout: last.vout, + }; + match serde_json::to_string(&next) { + Ok(c) => Some(c), + Err(e) => { + throw_sdk_exception( + env, + 1, + &format!("inventory cursor encode failed: {e}"), + ); + return ptr::null_mut(); + } + } + } + None => None, + } + } else { + None + }; + let page = UtxoPage { + utxos: rows, + has_more: has_more && cursor_json.is_some(), + cursor: cursor_json, + }; + match serde_json::to_string(&page) { + Ok(json) => env + .new_string(json) + .map(|s| s.into_raw()) + .unwrap_or(ptr::null_mut()), + Err(e) => { + throw_sdk_exception(env, 1, &format!("inventory page encode failed: {e}")); + ptr::null_mut() + } + } + }) +} + +/// Classify a batch of store rows against the engine's live state — the +/// reverse half of the reconcile transport, and the reason the paged +/// inventory carries no spent-outpoint list. The host pages its OWN mirror +/// rows and asks about them a batch at a time, so neither side builds a +/// set over the whole engine inventory. +/// +/// `queriesJson` is a JSON array of store rows: the account tuple the +/// store files each coin under, its `txid` (lower hex, wire order) and +/// `vout`, and the `scriptHex` the store recorded for it. The account and +/// script are the store's CLAIM — the engine checks both against its own +/// pools rather than trusting them, which is why a bare outpoint is not +/// enough. +/// +/// Returns one verdict byte per query, positionally: see the +/// `OUTPOINT_CLASS_*` constants (0 unknown, 1 unspent, 2 known-uncredited, +/// 3 not-owned). A query whose account tag this build cannot map keeps +/// `unknown` and the rest are still answered. +#[no_mangle] +pub extern "system" fn Java_org_dashfoundation_dashsdk_ffi_WalletManagerNative_walletManagerClassifyOutpoints( + mut env: JNIEnv, + _class: JClass, + manager_handle: jlong, + wallet_id: JByteArray, + queries_json: JString, +) -> jbyteArray { + guard(&mut env, ptr::null_mut(), |env| { + let Some(wid) = read_id32(env, &wallet_id) else { + return ptr::null_mut(); + }; + let raw = match env.get_string(&queries_json) { + Ok(s) => String::from(s), + Err(_) => { + throw_sdk_exception(env, 1, "outpoint queries must be a String"); + return ptr::null_mut(); + } + }; + let parsed: Vec = match serde_json::from_str(&raw) { + Ok(q) => q, + Err(e) => { + throw_sdk_exception(env, 1, &format!("malformed outpoint queries: {e}")); + return ptr::null_mut(); + } + }; + if parsed.is_empty() { + return env + .byte_array_from_slice(&[]) + .map(|a| a.into_raw()) + .unwrap_or(ptr::null_mut()); + } + // Scripts are owned here for the duration of the call — the query + // struct borrows them by pointer. + let mut scripts: Vec> = Vec::with_capacity(parsed.len()); + let mut queries: Vec = + Vec::with_capacity(parsed.len()); + for q in &parsed { + let Some((user_identity_id, friend_identity_id)) = q.account.identity_ids() else { + throw_sdk_exception(env, 1, "outpoint query carries a malformed identity id"); + return ptr::null_mut(); + }; + let txid: [u8; 32] = match hex::decode(&q.txid).ok().and_then(|b| b.try_into().ok()) { + Some(t) => t, + None => { + throw_sdk_exception(env, 1, "outpoint query carries a malformed txid"); + return ptr::null_mut(); + } + }; + let script = match hex::decode(&q.script_hex) { + Ok(s) => s, + Err(_) => { + throw_sdk_exception(env, 1, "outpoint query carries a malformed script"); + return ptr::null_mut(); + } + }; + scripts.push(script); + queries.push(platform_wallet_ffi::OutpointOwnershipQueryFFI { + type_tag: q.account.type_tag, + standard_tag: q.account.standard_tag, + index: q.account.index, + registration_index: q.account.registration_index, + key_class: q.account.key_class, + user_identity_id, + friend_identity_id, + outpoint: platform_wallet_ffi::OutPointFFI { txid, vout: q.vout }, + // Filled below, once `scripts` has stopped reallocating. + script_pubkey: ptr::null(), + script_pubkey_len: 0, + }); + } + for (query, script) in queries.iter_mut().zip(scripts.iter()) { + query.script_pubkey = if script.is_empty() { + ptr::null() + } else { + script.as_ptr() + }; + query.script_pubkey_len = script.len(); + } + + // Caller-allocated answers, one byte per query. + let mut classes = vec![platform_wallet_ffi::OUTPOINT_CLASS_UNKNOWN; queries.len()]; + let result = unsafe { + platform_wallet_ffi::platform_wallet_classify_outpoints( + manager_handle as Handle, + wid.as_ptr(), + queries.as_ptr(), + queries.len(), + classes.as_mut_ptr(), + ) + }; + // `scripts` must outlive the call above; this keeps that explicit + // against a future reorder. + drop(scripts); + if take_pwffi_error(env, result) { + return ptr::null_mut(); + } + env.byte_array_from_slice(&classes) + .map(|a| a.into_raw()) + .unwrap_or(ptr::null_mut()) + }) +} /// Read a 32-byte id from a Java `byte[]`; throws + returns None on the /// wrong length or a JNI error. fn read_id32(env: &mut JNIEnv, arr: &JByteArray) -> Option<[u8; 32]> {