commit 9abeece6cf73ac6e8baaf5c39f009fcfba0da448
parent d9401613fd72554b2c16ef1b116595fcb8b5789f
Author: Florian Dold <dold@taler.net>
Date: Thu, 20 Aug 2026 19:06:49 +0200
wallet-core: pay authenticated pull-purse remainder
Diffstat:
4 files changed, 717 insertions(+), 73 deletions(-)
diff --git a/packages/taler-wallet-core/src/db-common.ts b/packages/taler-wallet-core/src/db-common.ts
@@ -2378,6 +2378,12 @@ export interface PeerPullPaymentCoinSelection {
contributions: AmountString[];
coinPubs: string[];
totalCost: AmountString | undefined;
+
+ /** Number of leading entries confirmed by a signed exchange response. */
+ depositedCoinCount?: number;
+
+ /** Latest purse balance covered by a verified deposit confirmation. */
+ confirmedPurseBalance?: AmountString;
}
/**
diff --git a/packages/taler-wallet-core/src/dbtx-conformance-cases.ts b/packages/taler-wallet-core/src/dbtx-conformance-cases.ts
@@ -3795,6 +3795,13 @@ export const conformanceCases: ConformanceCase[] = [
async run(t, runner) {
const rec = makePeerPullDebit("ppld-1");
rec.contractPriv = ck("cpriv-pull");
+ rec.coinSel = {
+ coinPubs: [ck("pull-coin-1"), ck("pull-coin-2")],
+ contributions: [amt("TESTKUDOS:1"), amt("TESTKUDOS:2")],
+ totalCost: amt("TESTKUDOS:3.1"),
+ depositedCoinCount: 1,
+ confirmedPurseBalance: amt("TESTKUDOS:1"),
+ };
await runner.runReadWriteTx((tx) => tx.upsertPeerPullDebit(rec));
const got = await runner.runReadWriteTx((tx) =>
tx.getPeerPullDebit("ppld-1"),
diff --git a/packages/taler-wallet-core/src/pay-peer-pull-debit.test.ts b/packages/taler-wallet-core/src/pay-peer-pull-debit.test.ts
@@ -0,0 +1,113 @@
+/*
+ This file is part of GNU Taler
+ (C) 2026 Taler Systems S.A.
+
+ GNU Taler is free software; you can redistribute it and/or modify it under the
+ terms of the GNU General Public License as published by the Free Software
+ Foundation; either version 3, or (at your option) any later version.
+
+ GNU Taler is distributed in the hope that it will be useful, but WITHOUT ANY
+ WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR
+ A PARTICULAR PURPOSE. See the GNU General Public License for more details.
+
+ You should have received a copy of the GNU General Public License along with
+ GNU Taler; see the file COPYING. If not, see <http://www.gnu.org/licenses/>
+ */
+
+import { Amounts } from "@gnu-taler/taler-util";
+import assert from "node:assert/strict";
+import { test } from "node:test";
+import { PeerPullPaymentCoinSelection } from "./db-common.js";
+import {
+ getPeerPullDebitRemainder,
+ getPeerPullDebitUnconfirmedCoins,
+ markPeerPullDebitCoinsDeposited,
+ partitionPeerPullDebitRepair,
+} from "./pay-peer-pull-debit.js";
+
+test("peer pull debit selects only the authenticated purse remainder", () => {
+ assert.strictEqual(
+ Amounts.stringify(
+ getPeerPullDebitRemainder("TESTKUDOS:10", "TESTKUDOS:1")!,
+ ),
+ "TESTKUDOS:9",
+ );
+ assert.strictEqual(
+ Amounts.stringify(
+ getPeerPullDebitRemainder("TESTKUDOS:10", "TESTKUDOS:0")!,
+ ),
+ "TESTKUDOS:10",
+ );
+});
+
+test("a full or overfunded pull purse has no payable remainder", () => {
+ assert.strictEqual(
+ getPeerPullDebitRemainder("TESTKUDOS:10", "TESTKUDOS:10"),
+ undefined,
+ );
+ assert.strictEqual(
+ getPeerPullDebitRemainder("TESTKUDOS:10", "TESTKUDOS:11"),
+ undefined,
+ );
+});
+
+test("pull purse remainder rejects a currency mismatch", () => {
+ assert.throws(
+ () => getPeerPullDebitRemainder("TESTKUDOS:10", "EUR:1"),
+ /currency/i,
+ );
+});
+
+test("signed batch progress advances only in prefix order", () => {
+ const selection: PeerPullPaymentCoinSelection = {
+ coinPubs: ["one", "two", "three"],
+ contributions: ["TESTKUDOS:1", "TESTKUDOS:2", "TESTKUDOS:3"],
+ totalCost: "TESTKUDOS:6",
+ };
+ markPeerPullDebitCoinsDeposited(selection, 0, 2, "TESTKUDOS:3");
+ assert.strictEqual(selection.depositedCoinCount, 2);
+ assert.strictEqual(selection.confirmedPurseBalance, "TESTKUDOS:3");
+ assert.throws(
+ () => markPeerPullDebitCoinsDeposited(selection, 0, 1, "TESTKUDOS:4"),
+ /sequence/,
+ );
+});
+
+test("conflict repair never carries coins beyond the latest remainder", () => {
+ const selection: PeerPullPaymentCoinSelection = {
+ coinPubs: ["accepted", "too-large", "fits", "broken"],
+ contributions: ["TESTKUDOS:2", "TESTKUDOS:6", "TESTKUDOS:3", "TESTKUDOS:2"],
+ totalCost: "TESTKUDOS:13",
+ depositedCoinCount: 1,
+ };
+ const partition = partitionPeerPullDebitRepair(
+ selection,
+ ["TESTKUDOS:1", "TESTKUDOS:1", "TESTKUDOS:1", "TESTKUDOS:1"],
+ "broken",
+ "TESTKUDOS:4",
+ );
+
+ assert.deepStrictEqual(partition.repair, [
+ {
+ coinPub: "fits",
+ contribution: Amounts.parseOrThrow("TESTKUDOS:3"),
+ },
+ ]);
+ assert.deepStrictEqual(partition.recover, [
+ { coinPub: "too-large", amount: "TESTKUDOS:6" },
+ ]);
+});
+
+test("race recovery excludes confirmed and externally broken coins", () => {
+ const selection: PeerPullPaymentCoinSelection = {
+ coinPubs: ["accepted", "unsubmitted", "broken"],
+ contributions: ["TESTKUDOS:1", "TESTKUDOS:2", "TESTKUDOS:3"],
+ totalCost: "TESTKUDOS:6",
+ depositedCoinCount: 1,
+ };
+
+ assert.deepStrictEqual(
+ getPeerPullDebitUnconfirmedCoins(selection, new Set(["broken"])),
+ [{ coinPub: "unsubmitted", amount: "TESTKUDOS:2" }],
+ );
+});
diff --git a/packages/taler-wallet-core/src/pay-peer-pull-debit.ts b/packages/taler-wallet-core/src/pay-peer-pull-debit.ts
@@ -22,6 +22,8 @@
import {
AcceptPeerPullPaymentResponse,
+ AmountJson,
+ AmountLike,
Amounts,
CoinRefreshRequest,
ConfirmPeerPullDebitRequest,
@@ -80,6 +82,7 @@ import {
} from "./common.js";
import {
PeerPullDebitRecordStatus,
+ PeerPullPaymentCoinSelection,
RefreshOperationStatus,
WalletPeerPullDebit,
timestampPreciseFromDb,
@@ -92,6 +95,7 @@ import {
} from "./progress.js";
import {
getTotalPeerPaymentCost,
+ getTotalPeerPaymentCostInTx,
isPurseDeposited,
isPurseGoneByExpiration,
queryCoinInfosForSelection,
@@ -115,6 +119,178 @@ import {
const logger = new Logger("pay-peer-pull-debit.ts");
/**
+ * Amount this payer still needs to contribute to the purse.
+ *
+ * A missing result means that the authenticated purse balance already meets
+ * (or exceeds) the contract target. The caller must then distinguish a
+ * locally proven replay from a purse completed by another wallet.
+ */
+export function getPeerPullDebitRemainder(
+ target: AmountLike,
+ balance: AmountLike,
+): AmountJson | undefined {
+ const targetAmount = Amounts.parseOrThrow(target);
+ const balanceAmount = Amounts.parseOrThrow(balance);
+ if (Amounts.cmp(balanceAmount, targetAmount) >= 0) {
+ return undefined;
+ }
+ return Amounts.sub(targetAmount, balanceAmount).amount;
+}
+
+export function getPeerPullDebitDepositedCoinCount(
+ selection: PeerPullPaymentCoinSelection,
+): number {
+ const count = selection.depositedCoinCount ?? 0;
+ if (
+ !Number.isSafeInteger(count) ||
+ count < 0 ||
+ count > selection.coinPubs.length ||
+ selection.coinPubs.length !== selection.contributions.length
+ ) {
+ throw Error("invalid persisted peer pull-debit deposit progress");
+ }
+ return count;
+}
+
+export function markPeerPullDebitCoinsDeposited(
+ selection: PeerPullPaymentCoinSelection,
+ batchStart: number,
+ batchSize: number,
+ confirmedPurseBalance: AmountLike,
+): void {
+ const oldCount = getPeerPullDebitDepositedCoinCount(selection);
+ if (
+ batchStart !== oldCount ||
+ !Number.isSafeInteger(batchSize) ||
+ batchSize <= 0 ||
+ batchStart + batchSize > selection.coinPubs.length
+ ) {
+ throw Error("invalid peer pull-debit deposit confirmation sequence");
+ }
+ selection.depositedCoinCount = batchStart + batchSize;
+ selection.confirmedPurseBalance = Amounts.stringify(confirmedPurseBalance);
+}
+
+export function getPeerPullDebitUnconfirmedCoins(
+ selection: PeerPullPaymentCoinSelection,
+ unrecoverableCoinPubs: ReadonlySet<string> = new Set(),
+): CoinRefreshRequest[] {
+ const acceptedCount = getPeerPullDebitDepositedCoinCount(selection);
+ const coins: CoinRefreshRequest[] = [];
+ for (let i = acceptedCount; i < selection.coinPubs.length; i++) {
+ if (unrecoverableCoinPubs.has(selection.coinPubs[i])) {
+ continue;
+ }
+ coins.push({
+ amount: selection.contributions[i],
+ coinPub: selection.coinPubs[i],
+ });
+ }
+ return coins;
+}
+
+export function partitionPeerPullDebitRepair(
+ selection: PeerPullPaymentCoinSelection,
+ depositFees: AmountLike[],
+ brokenCoinPub: string,
+ remainder: AmountLike,
+): {
+ repair: PreviousPayCoins;
+ recover: CoinRefreshRequest[];
+} {
+ if (depositFees.length !== selection.coinPubs.length) {
+ throw Error("missing deposit fee for peer pull-debit repair");
+ }
+ let remaining = Amounts.parseOrThrow(remainder);
+ const repair: PreviousPayCoins = [];
+ const recover: CoinRefreshRequest[] = [];
+ const acceptedCount = getPeerPullDebitDepositedCoinCount(selection);
+ for (let i = acceptedCount; i < selection.coinPubs.length; i++) {
+ const coinPub = selection.coinPubs[i];
+ if (coinPub === brokenCoinPub) {
+ continue;
+ }
+ const contribution = Amounts.parseOrThrow(selection.contributions[i]);
+ const netContribution = Amounts.sub(contribution, depositFees[i]);
+ if (
+ !netContribution.saturated &&
+ Amounts.isNonZero(netContribution.amount) &&
+ Amounts.cmp(netContribution.amount, remaining) <= 0
+ ) {
+ repair.push({ coinPub, contribution });
+ remaining = Amounts.sub(remaining, netContribution.amount).amount;
+ } else {
+ recover.push({
+ coinPub,
+ amount: Amounts.stringify(contribution),
+ });
+ }
+ }
+ return { repair, recover };
+}
+
+async function getStoredPeerPullDebitSelectionCostInTx(
+ wex: WalletExecutionContext,
+ tx: WalletDbTransaction,
+ selection: PeerPullPaymentCoinSelection,
+ end: number,
+ currency: string,
+): Promise<AmountJson> {
+ if (end === 0) {
+ return Amounts.zeroOfCurrency(currency);
+ }
+ const pubs = selection.coinPubs.slice(0, end);
+ const coins = await tx.getCoinsByPubs(pubs);
+ const coinsByPub = new Map(coins.map((coin) => [coin.coinPub, coin]));
+ const prospective = pubs.map((coinPub, i) => {
+ const coin = coinsByPub.get(coinPub);
+ checkDbInvariant(!!coin, `coin ${coinPub} missing from peer pull debit`);
+ return {
+ coinPub,
+ contribution: selection.contributions[i],
+ denomPubHash: coin.denomPubHash,
+ exchangeBaseUrl: coin.exchangeBaseUrl,
+ exchangeMasterPub: coin.exchangeMasterPub,
+ };
+ });
+ return getTotalPeerPaymentCostInTx(wex, tx, prospective);
+}
+
+async function retainPeerPullDebitAcceptedSelectionInTx(
+ wex: WalletExecutionContext,
+ tx: WalletDbTransaction,
+ rec: WalletPeerPullDebit,
+): Promise<number> {
+ if (!rec.coinSel) {
+ return 0;
+ }
+ const acceptedCount = getPeerPullDebitDepositedCoinCount(rec.coinSel);
+ if (acceptedCount === 0) {
+ return 0;
+ }
+ rec.coinSel.totalCost = Amounts.stringify(
+ await getStoredPeerPullDebitSelectionCostInTx(
+ wex,
+ tx,
+ rec.coinSel,
+ acceptedCount,
+ Amounts.currencyOf(rec.amount),
+ ),
+ );
+ rec.coinSel.coinPubs = rec.coinSel.coinPubs.slice(0, acceptedCount);
+ rec.coinSel.contributions = rec.coinSel.contributions.slice(0, acceptedCount);
+ return acceptedCount;
+}
+
+function makePeerPullDebitPartialDepositError(): TalerErrorDetail {
+ return makeErrorDetail(
+ TalerErrorCode.WALLET_PEER_PULL_DEBIT_PURSE_GONE,
+ { purseExpired: false },
+ "only part of this wallet's payment was accepted by the purse",
+ );
+}
+
+/**
* Common context for a peer-pull-debit transaction.
*/
export class PeerPullDebitTransactionContext implements TransactionContext {
@@ -216,6 +392,7 @@ export class PeerPullDebitTransactionContext implements TransactionContext {
async failTransaction(
fromSt: PeerPullDebitRecordStatus,
reason?: TalerErrorDetail,
+ unrecoverableCoinPubs: ReadonlySet<string> = new Set(),
): Promise<void> {
const { wex } = this;
await wex.runWalletDbTx(async (tx) => {
@@ -223,6 +400,23 @@ export class PeerPullDebitTransactionContext implements TransactionContext {
if (rec?.status != fromSt) {
return;
}
+ if (rec.coinSel) {
+ const recover = getPeerPullDebitUnconfirmedCoins(
+ rec.coinSel,
+ unrecoverableCoinPubs,
+ );
+ if (recover.length > 0) {
+ await createRefreshGroup(
+ wex,
+ tx,
+ Amounts.currencyOf(rec.amount),
+ recover,
+ RefreshReason.AbortPeerPullDebit,
+ this.transactionId,
+ );
+ }
+ await retainPeerPullDebitAcceptedSelectionInTx(wex, tx, rec);
+ }
rec.status = PeerPullDebitRecordStatus.Failed;
rec.failReason = reason;
await h.update(rec, "fail");
@@ -234,7 +428,10 @@ export class PeerPullDebitTransactionContext implements TransactionContext {
* the invoice anymore, which happens when the invoice lapsed, when the
* payee withdrew it and when another wallet paid it first.
*/
- async purseGoneTransaction(fromSt: PeerPullDebitRecordStatus): Promise<void> {
+ async purseGoneTransaction(
+ fromSt: PeerPullDebitRecordStatus,
+ unrecoverableCoinPubs: ReadonlySet<string> = new Set(),
+ ): Promise<void> {
const { wex } = this;
await wex.runWalletDbTx(async (tx) => {
const [rec, h] = await this.getRecordHandle(tx);
@@ -242,25 +439,31 @@ export class PeerPullDebitTransactionContext implements TransactionContext {
return;
}
if (rec.coinSel) {
- // The purse is gone, so the coins we allocated for it never reached
- // anyone. Without the refresh they would stay allocated to this dead
- // transaction and the money would be lost.
- const coinPubs: CoinRefreshRequest[] = [];
- for (let i = 0; i < rec.coinSel.coinPubs.length; i++) {
- coinPubs.push({
- amount: rec.coinSel.contributions[i],
- coinPub: rec.coinSel.coinPubs[i],
- });
- }
- const refresh = await createRefreshGroup(
- wex,
- tx,
- Amounts.currencyOf(rec.totalCostEstimated),
- coinPubs,
- RefreshReason.AbortPeerPullDebit,
- this.transactionId,
+ const acceptedCount = getPeerPullDebitDepositedCoinCount(rec.coinSel);
+ // Only the suffix without a signed deposit confirmation can be
+ // recovered. Confirmed prefix coins did reach the purse.
+ const coinPubs = getPeerPullDebitUnconfirmedCoins(
+ rec.coinSel,
+ unrecoverableCoinPubs,
);
- rec.abortRefreshGroupId = refresh.refreshGroupId;
+ if (coinPubs.length > 0) {
+ const refresh = await createRefreshGroup(
+ wex,
+ tx,
+ Amounts.currencyOf(rec.totalCostEstimated),
+ coinPubs,
+ RefreshReason.AbortPeerPullDebit,
+ this.transactionId,
+ );
+ rec.abortRefreshGroupId = refresh.refreshGroupId;
+ }
+ if (acceptedCount > 0) {
+ await retainPeerPullDebitAcceptedSelectionInTx(wex, tx, rec);
+ rec.status = PeerPullDebitRecordStatus.Failed;
+ rec.failReason = makePeerPullDebitPartialDepositError();
+ await h.update(rec, "purse-gone-after-partial-deposit");
+ return;
+ }
}
const ct = await tx.getContractTerms(rec.contractTermsHash);
checkDbInvariant(!!ct, "contract terms for P2P payment not found");
@@ -398,11 +601,20 @@ export class PeerPullDebitTransactionContext implements TransactionContext {
pi.status = PeerPullDebitRecordStatus.Aborted;
pi.abortReason = reason;
} else {
- for (let i = 0; i < pi.coinSel.coinPubs.length; i++) {
- coinPubs.push({
- amount: pi.coinSel.contributions[i],
- coinPub: pi.coinSel.coinPubs[i],
- });
+ const acceptedCount = getPeerPullDebitDepositedCoinCount(pi.coinSel);
+ coinPubs.push(...getPeerPullDebitUnconfirmedCoins(pi.coinSel));
+ if (coinPubs.length === 0) {
+ if (acceptedCount > 0) {
+ await retainPeerPullDebitAcceptedSelectionInTx(this.wex, tx, pi);
+ pi.status = PeerPullDebitRecordStatus.Failed;
+ pi.failReason = makePeerPullDebitPartialDepositError();
+ pi.abortReason = reason;
+ } else {
+ pi.status = PeerPullDebitRecordStatus.Aborted;
+ pi.abortReason = reason;
+ }
+ await h.update(pi, "abort-without-refresh");
+ return;
}
const refresh = await createRefreshGroup(
this.wex,
@@ -435,9 +647,8 @@ async function handlePurseCreationConflict(
const brokenCoinPub = conflict.coin_pub;
logger.trace(`excluded broken coin pub=${brokenCoinPub}`);
- const instructedAmount = Amounts.parseOrThrow(peerPullInc.amount);
- const currency = instructedAmount.currency;
const exchangeBaseUrl = peerPullInc.exchangeBaseUrl;
+ const currency = Amounts.currencyOf(peerPullInc.amount);
const sel = peerPullInc.coinSel;
checkDbInvariant(
@@ -445,17 +656,55 @@ async function handlePurseCreationConflict(
`no coin selected for peer pull deposit ${peerPullInc.pursePub}`,
);
- const repair: PreviousPayCoins = [];
+ const statusResp = await walletExchangeClient(
+ exchangeBaseUrl,
+ ctx.wex,
+ ).getPurseStatusAtMerge(peerPullInc.pursePub);
+ switch (statusResp.case) {
+ case "ok":
+ break;
+ case HttpStatusCode.Gone:
+ await ctx.purseGoneTransaction(
+ peerPullInc.status,
+ new Set([brokenCoinPub]),
+ );
+ return TaskRunResult.finished();
+ case HttpStatusCode.NotFound:
+ await ctx.failTransaction(
+ peerPullInc.status,
+ statusResp.detail,
+ new Set([brokenCoinPub]),
+ );
+ return TaskRunResult.finished();
+ default:
+ assertUnreachable(statusResp);
+ }
+ await requireValidExchangePurseStatus(
+ ctx.wex,
+ exchangeBaseUrl,
+ statusResp.body,
+ );
- for (let i = 0; i < sel.coinPubs.length; i++) {
- if (sel.coinPubs[i] != brokenCoinPub) {
- repair.push({
- coinPub: sel.coinPubs[i],
- contribution: Amounts.parseOrThrow(sel.contributions[i]),
- });
- }
+ const instructedAmount = getPeerPullDebitRemainder(
+ peerPullInc.amount,
+ statusResp.body.balance,
+ );
+ if (isPurseDeposited(statusResp.body) || !instructedAmount) {
+ await ctx.purseGoneTransaction(
+ peerPullInc.status,
+ new Set([brokenCoinPub]),
+ );
+ return TaskRunResult.finished();
}
+ const coinDetails = await queryCoinInfosForSelection(ctx.wex, sel);
+ const { repair, recover } = partitionPeerPullDebitRepair(
+ sel,
+ coinDetails.map((x) => x.feeDeposit),
+ brokenCoinPub,
+ instructedAmount,
+ );
+
const coinSelRes = await selectPeerCoins(ctx.wex, {
instructedAmount,
repair,
@@ -467,18 +716,66 @@ async function handlePurseCreationConflict(
});
switch (coinSelRes.type) {
- case "failure":
- // FIXME: Details!
- throw Error(
- "insufficient balance to re-select coins to repair double spending",
- );
+ case "failure": {
+ // Recover every unsubmitted, non-broken allocation. Once those
+ // refreshes produce fresh coins, the normal pending path re-selects
+ // against a newly authenticated remainder.
+ const recoverAfterFailure = [
+ ...recover,
+ ...repair.map((x) => ({
+ coinPub: x.coinPub,
+ amount: Amounts.stringify(x.contribution),
+ })),
+ ];
+ await ctx.wex.runWalletDbTx(async (tx) => {
+ const [rec, h] = await ctx.getRecordHandle(tx);
+ if (
+ rec?.status !== PeerPullDebitRecordStatus.PendingDeposit ||
+ !rec.coinSel
+ ) {
+ return;
+ }
+ const acceptedCount = getPeerPullDebitDepositedCoinCount(rec.coinSel);
+ if (recoverAfterFailure.length > 0) {
+ await createRefreshGroup(
+ ctx.wex,
+ tx,
+ currency,
+ recoverAfterFailure,
+ RefreshReason.AbortPeerPullDebit,
+ ctx.transactionId,
+ );
+ }
+ if (acceptedCount === 0) {
+ rec.coinSel = undefined;
+ } else {
+ rec.coinSel.totalCost = Amounts.stringify(
+ await getStoredPeerPullDebitSelectionCostInTx(
+ ctx.wex,
+ tx,
+ rec.coinSel,
+ acceptedCount,
+ currency,
+ ),
+ );
+ rec.coinSel.coinPubs = rec.coinSel.coinPubs.slice(0, acceptedCount);
+ rec.coinSel.contributions = rec.coinSel.contributions.slice(
+ 0,
+ acceptedCount,
+ );
+ rec.coinSel.confirmedPurseBalance = statusResp.body.balance;
+ }
+ await h.update(rec, "purse-conflict-await-refresh");
+ });
+ return TaskRunResult.backoff();
+ }
case "success":
break;
default:
assertUnreachable(coinSelRes);
}
- const totalAmount = await getTotalPeerPaymentCost(
+ const replacementCost = await getTotalPeerPaymentCost(
ctx.wex,
coinSelRes.result.coins,
);
@@ -491,15 +788,47 @@ async function handlePurseCreationConflict(
switch (rec.status) {
case PeerPullDebitRecordStatus.PendingDeposit:
case PeerPullDebitRecordStatus.SuspendedDeposit: {
- const sel = coinSelRes.result;
+ if (!rec.coinSel) {
+ return;
+ }
+ const acceptedCount = getPeerPullDebitDepositedCoinCount(rec.coinSel);
+ const oldSelection = rec.coinSel;
+ const acceptedCost = await getStoredPeerPullDebitSelectionCostInTx(
+ ctx.wex,
+ tx,
+ oldSelection,
+ acceptedCount,
+ currency,
+ );
+ if (recover.length > 0) {
+ await createRefreshGroup(
+ ctx.wex,
+ tx,
+ currency,
+ recover,
+ RefreshReason.AbortPeerPullDebit,
+ ctx.transactionId,
+ );
+ }
+ const replacement = coinSelRes.result;
rec.coinSel = {
- coinPubs: sel.coins.map((x) => x.coinPub),
- contributions: sel.coins.map((x) => x.contribution),
- totalCost: Amounts.stringify(totalAmount),
+ coinPubs: [
+ ...oldSelection.coinPubs.slice(0, acceptedCount),
+ ...replacement.coins.map((x) => x.coinPub),
+ ],
+ contributions: [
+ ...oldSelection.contributions.slice(0, acceptedCount),
+ ...replacement.coins.map((x) => x.contribution),
+ ],
+ totalCost: Amounts.stringify(
+ Amounts.add(acceptedCost, replacementCost).amount,
+ ),
+ depositedCoinCount: acceptedCount,
+ confirmedPurseBalance: statusResp.body.balance,
};
// The carried-over coins were spent when they were first selected,
// the ones that replace the double-spent coin still have to be.
- const added = coinsAddedByRepair(repair, sel.coins);
+ const added = coinsAddedByRepair(repair, replacement.coins);
await spendCoins(ctx.wex, tx, {
transactionId: ctx.transactionId,
coinPubs: added.map((x) => x.coinPub),
@@ -548,7 +877,10 @@ async function processPeerPullDebitDialogProposed(
resp.body,
);
- if (isPurseDeposited(resp.body)) {
+ if (
+ isPurseDeposited(resp.body) ||
+ !getPeerPullDebitRemainder(pullIni.amount, resp.body.balance)
+ ) {
logger.info("purse completed by another wallet");
await ctx.wex.runWalletDbTx(async (tx) => {
const [rec, h] = await ctx.getRecordHandle(tx);
@@ -579,9 +911,63 @@ async function processPeerPullDebitPendingDeposit(
const exchangeBaseUrl = peerPullInc.exchangeBaseUrl;
- // This can happen when there was a prospective coin selection.
- if (coinSel == null) {
- const instructedAmount = Amounts.parseOrThrow(peerPullInc.amount);
+ if (coinSel) {
+ const depositedCount = getPeerPullDebitDepositedCoinCount(coinSel);
+ if (
+ depositedCount > 0 &&
+ depositedCount === coinSel.coinPubs.length &&
+ coinSel.confirmedPurseBalance &&
+ !getPeerPullDebitRemainder(
+ peerPullInc.amount,
+ coinSel.confirmedPurseBalance,
+ )
+ ) {
+ await wex.runWalletDbTx(async (tx) => {
+ const [rec, h] = await ctx.getRecordHandle(tx);
+ if (rec?.status !== PeerPullDebitRecordStatus.PendingDeposit) {
+ return;
+ }
+ rec.status = PeerPullDebitRecordStatus.Done;
+ await h.update(rec, "deposit-replay-confirmed");
+ });
+ return TaskRunResult.finished();
+ }
+ }
+
+ // This can happen when there was a prospective coin selection, or after a
+ // conflict where the already-confirmed prefix was retained and every
+ // unsubmitted coin was sent to recovery.
+ if (
+ coinSel == null ||
+ getPeerPullDebitDepositedCoinCount(coinSel) === coinSel.coinPubs.length
+ ) {
+ const exchangeClient = walletExchangeClient(exchangeBaseUrl, wex);
+ const statusResp = await exchangeClient.getPurseStatusAtMerge(pursePub);
+ switch (statusResp.case) {
+ case "ok":
+ break;
+ case HttpStatusCode.Gone:
+ await ctx.purseGoneTransaction(peerPullInc.status);
+ return TaskRunResult.finished();
+ case HttpStatusCode.NotFound:
+ await ctx.failTransaction(peerPullInc.status, statusResp.detail);
+ return TaskRunResult.finished();
+ default:
+ assertUnreachable(statusResp);
+ }
+ await requireValidExchangePurseStatus(
+ wex,
+ exchangeBaseUrl,
+ statusResp.body,
+ );
+ const instructedAmount = getPeerPullDebitRemainder(
+ peerPullInc.amount,
+ statusResp.body.balance,
+ );
+ if (isPurseDeposited(statusResp.body) || !instructedAmount) {
+ await ctx.purseGoneTransaction(peerPullInc.status);
+ return TaskRunResult.finished();
+ }
const currency = instructedAmount.currency;
const coinSelRes = await selectPeerCoins(wex, {
@@ -619,9 +1005,18 @@ async function processPeerPullDebitPendingDeposit(
if (!rec) {
return TaskRunResult.finished();
}
+ if (rec.status !== PeerPullDebitRecordStatus.PendingDeposit) {
+ return TaskRunResult.backoff();
+ }
+ const acceptedCount = rec.coinSel
+ ? getPeerPullDebitDepositedCoinCount(rec.coinSel)
+ : 0;
if (
- rec.status !== PeerPullDebitRecordStatus.PendingDeposit ||
- rec.coinSel != null
+ (coinSel == null && rec.coinSel != null) ||
+ (coinSel != null &&
+ (!rec.coinSel ||
+ acceptedCount !== rec.coinSel.coinPubs.length ||
+ acceptedCount !== coinSel.coinPubs.length))
) {
return TaskRunResult.backoff();
}
@@ -633,10 +1028,25 @@ async function processPeerPullDebitPendingDeposit(
),
refreshReason: RefreshReason.PayPeerPull,
});
+ const acceptedCoinPubs = rec.coinSel?.coinPubs ?? [];
+ const acceptedContributions = rec.coinSel?.contributions ?? [];
+ const acceptedCost = rec.coinSel?.totalCost
+ ? Amounts.parseOrThrow(rec.coinSel.totalCost)
+ : Amounts.zeroOfCurrency(currency);
rec.coinSel = {
- coinPubs: coinSelRes.result.coins.map((x) => x.coinPub),
- contributions: coinSelRes.result.coins.map((x) => x.contribution),
- totalCost: Amounts.stringify(totalAmount),
+ coinPubs: [
+ ...acceptedCoinPubs,
+ ...coinSelRes.result.coins.map((x) => x.coinPub),
+ ],
+ contributions: [
+ ...acceptedContributions,
+ ...coinSelRes.result.coins.map((x) => x.contribution),
+ ],
+ totalCost: Amounts.stringify(
+ Amounts.add(acceptedCost, totalAmount).amount,
+ ),
+ depositedCoinCount: acceptedCount,
+ confirmedPurseBalance: statusResp.body.balance,
};
await h.update(rec, "select-coins");
return TaskRunResult.progress();
@@ -657,8 +1067,9 @@ async function processPeerPullDebitPendingDeposit(
const coins = await queryCoinInfosForSelection(wex, coinSel);
const maxBatchSize = 64;
+ const depositedCoinCount = getPeerPullDebitDepositedCoinCount(coinSel);
- for (let i = 0; i < coins.length; i += maxBatchSize) {
+ for (let i = depositedCoinCount; i < coins.length; i += maxBatchSize) {
const batchSize = Math.min(maxBatchSize, coins.length - i);
wex.oc.observe({
@@ -694,6 +1105,77 @@ async function processPeerPullDebitPendingDeposit(
purseExpiration: contractTerms.contractTermsRaw.purse_expiration,
response: resp.body,
});
+ const purseComplete =
+ getPeerPullDebitRemainder(
+ peerPullInc.amount,
+ resp.body.total_deposited,
+ ) === undefined;
+ const confirmationStored = await wex.runWalletDbTx(async (tx) => {
+ const [rec, h] = await ctx.getRecordHandle(tx);
+ if (
+ rec?.status !== PeerPullDebitRecordStatus.PendingDeposit ||
+ !rec.coinSel
+ ) {
+ return false;
+ }
+ const currentCount = getPeerPullDebitDepositedCoinCount(rec.coinSel);
+ if (currentCount !== i) {
+ return false;
+ }
+ markPeerPullDebitCoinsDeposited(
+ rec.coinSel,
+ i,
+ batchSize,
+ resp.body.total_deposited,
+ );
+ if (purseComplete) {
+ const acceptedCount = i + batchSize;
+ const unsubmitted: CoinRefreshRequest[] = [];
+ for (
+ let coinIndex = acceptedCount;
+ coinIndex < rec.coinSel.coinPubs.length;
+ coinIndex++
+ ) {
+ unsubmitted.push({
+ coinPub: rec.coinSel.coinPubs[coinIndex],
+ amount: rec.coinSel.contributions[coinIndex],
+ });
+ }
+ if (unsubmitted.length > 0) {
+ await createRefreshGroup(
+ wex,
+ tx,
+ Amounts.currencyOf(peerPullInc.amount),
+ unsubmitted,
+ RefreshReason.AbortPeerPullDebit,
+ ctx.transactionId,
+ );
+ }
+ rec.coinSel.totalCost = Amounts.stringify(
+ await getStoredPeerPullDebitSelectionCostInTx(
+ wex,
+ tx,
+ rec.coinSel,
+ acceptedCount,
+ Amounts.currencyOf(peerPullInc.amount),
+ ),
+ );
+ rec.coinSel.coinPubs = rec.coinSel.coinPubs.slice(0, acceptedCount);
+ rec.coinSel.contributions = rec.coinSel.contributions.slice(
+ 0,
+ acceptedCount,
+ );
+ rec.status = PeerPullDebitRecordStatus.Done;
+ }
+ await h.update(rec, "deposit-confirmed");
+ return true;
+ });
+ if (!confirmationStored) {
+ return TaskRunResult.progress();
+ }
+ if (purseComplete) {
+ return TaskRunResult.finished();
+ }
continue;
case HttpStatusCode.Gone: {
await ctx.purseGoneTransaction(peerPullInc.status);
@@ -709,17 +1191,9 @@ async function processPeerPullDebitPendingDeposit(
assertUnreachable(resp);
}
}
- // All batches succeeded, we can transition!
- await ctx.wex.runWalletDbTx(async (tx) => {
- const [rec, h] = await ctx.getRecordHandle(tx);
- switch (rec?.status) {
- case PeerPullDebitRecordStatus.PendingDeposit:
- rec.status = PeerPullDebitRecordStatus.Done;
- break;
- default:
- return;
- }
- await h.update(rec, "deposit-done");
+ await ctx.failTransaction(peerPullInc.status, {
+ code: TalerErrorCode.WALLET_TRANSACTION_PROTOCOL_VIOLATION,
+ hint: "exchange accepted every selected coin but reported an incomplete purse",
});
return TaskRunResult.finished();
}
@@ -732,6 +1206,7 @@ async function processPeerPullDebitAbortingRefresh(
const abortRefreshGroupId = peerPullInc.abortRefreshGroupId;
checkLogicInvariant(!!abortRefreshGroupId);
const ctx = new PeerPullDebitTransactionContext(wex, peerPullDebitId);
+ let terminal = false;
await wex.runWalletDbTx(async (tx) => {
const refreshGroup = await tx.getRefreshGroup(abortRefreshGroupId);
const [rec, h] = await ctx.getRecordHandle(tx);
@@ -743,13 +1218,27 @@ async function processPeerPullDebitAbortingRefresh(
// just go into failed.
logger.warn("no aborting refresh group found for deposit group");
rec.status = PeerPullDebitRecordStatus.Failed;
+ terminal = true;
} else {
switch (refreshGroup.operationStatus) {
- case RefreshOperationStatus.Finished:
- rec.status = PeerPullDebitRecordStatus.Aborted;
+ case RefreshOperationStatus.Finished: {
+ const acceptedCount = await retainPeerPullDebitAcceptedSelectionInTx(
+ wex,
+ tx,
+ rec,
+ );
+ if (acceptedCount > 0) {
+ rec.status = PeerPullDebitRecordStatus.Failed;
+ rec.failReason = makePeerPullDebitPartialDepositError();
+ } else {
+ rec.status = PeerPullDebitRecordStatus.Aborted;
+ }
+ terminal = true;
break;
+ }
case RefreshOperationStatus.Failed: {
rec.status = PeerPullDebitRecordStatus.Failed;
+ terminal = true;
break;
}
default:
@@ -758,8 +1247,7 @@ async function processPeerPullDebitAbortingRefresh(
}
await h.update(rec, "aborting-refresh-failed");
});
- // FIXME: Shouldn't this be finished in some cases?!
- return TaskRunResult.backoff();
+ return terminal ? TaskRunResult.finished() : TaskRunResult.backoff();
}
export async function processPeerPullDebit(
@@ -825,7 +1313,32 @@ export async function confirmPeerPullDebit(
const exchangeBaseUrl = peerPullInc.exchangeBaseUrl;
- const instructedAmount = Amounts.parseOrThrow(peerPullInc.amount);
+ const statusResp = await walletExchangeClient(
+ exchangeBaseUrl,
+ wex,
+ ).getPurseStatusAtMerge(peerPullInc.pursePub);
+ switch (statusResp.case) {
+ case "ok":
+ break;
+ case HttpStatusCode.Gone:
+ await ctx.purseGoneTransaction(peerPullInc.status);
+ return { transactionId: ctx.transactionId };
+ case HttpStatusCode.NotFound:
+ await ctx.failTransaction(peerPullInc.status, statusResp.detail);
+ return { transactionId: ctx.transactionId };
+ default:
+ assertUnreachable(statusResp);
+ }
+ await requireValidExchangePurseStatus(wex, exchangeBaseUrl, statusResp.body);
+
+ const instructedAmount = getPeerPullDebitRemainder(
+ peerPullInc.amount,
+ statusResp.body.balance,
+ );
+ if (isPurseDeposited(statusResp.body) || !instructedAmount) {
+ await ctx.purseGoneTransaction(peerPullInc.status);
+ return { transactionId: ctx.transactionId };
+ }
const currency = instructedAmount.currency;
const coinSelRes = await selectPeerCoins(wex, {
@@ -880,6 +1393,8 @@ export async function confirmPeerPullDebit(
coinPubs: coinSelRes.result.coins.map((x) => x.coinPub),
contributions: coinSelRes.result.coins.map((x) => x.contribution),
totalCost: Amounts.stringify(totalAmount),
+ depositedCoinCount: 0,
+ confirmedPurseBalance: statusResp.body.balance,
};
}
rec.status = PeerPullDebitRecordStatus.PendingDeposit;
@@ -1079,7 +1594,11 @@ async function internalPreparePeerPullDebit(
await requireValidExchangePurseStatus(wex, exchangeBaseUrl, resp.body);
- if (isPurseDeposited(resp.body)) {
+ const instructedAmount = getPeerPullDebitRemainder(
+ contractTerms.amount,
+ resp.body.balance,
+ );
+ if (isPurseDeposited(resp.body) || !instructedAmount) {
logger.info("purse completed by another wallet");
throw TalerError.fromDetail(
TalerErrorCode.WALLET_PEER_PULL_DEBIT_ALREADY_PAID,
@@ -1094,7 +1613,6 @@ async function internalPreparePeerPullDebit(
// FIXME: Why don't we compute the totalCost here?!
- const instructedAmount = Amounts.parseOrThrow(contractTerms.amount);
const currency = Amounts.currencyOf(instructedAmount);
const coinSelRes = await selectPeerCoins(wex, {
