wallet-core: implement coin selection repair for p2p payments

packages/taler-wallet-core/src/operations/pay-peer-common.ts

@@ -158,6 +158,12 @@ export async function queryCoinInfosForSelection(
   return infos;
 }
 
+export interface PeerCoinRepair {
+  exchangeBaseUrl: string;
+  coinPubs: CoinPublicKeyString[];
+  contribs: AmountJson[];
+}
+
 export interface PeerCoinSelectionRequest {
   instructedAmount: AmountJson;
 
@@ -165,11 +171,7 @@ export interface PeerCoinSelectionRequest {
    * Instruct the coin selection to repair this coin
    * selection instead of selecting completely new coins.
    */
-  repair?: {
+  repair?: PeerCoinRepair;
-    exchangeBaseUrl: string;
-    coinPubs: CoinPublicKeyString[];
-    contribs: AmountJson[];
-  };
 }
 
 export async function selectPeerCoins(

packages/taler-wallet-core/src/operations/pay-peer-pull-debit.ts

@@ -29,6 +29,7 @@ import {
   TalerError,
   TalerErrorCode,
   TalerPreciseTimestamp,
+  TalerProtocolViolationError,
   TransactionAction,
   TransactionMajorState,
   TransactionMinorState,
@@ -44,7 +45,11 @@ import {
   j2s,
   parsePayPullUri,
 } from "@gnu-taler/taler-util";
-import { readSuccessResponseJsonOrThrow } from "@gnu-taler/taler-util/http";
+import {
+  HttpResponse,
+  readSuccessResponseJsonOrThrow,
+  readTalerErrorResponse,
+} from "@gnu-taler/taler-util/http";
 import {
   InternalWalletState,
   PeerPullDebitRecordStatus,
@@ -62,6 +67,7 @@ import {
 } from "../util/retries.js";
 import { runOperationWithErrorReporting, spendCoins } from "./common.js";
 import {
+  PeerCoinRepair,
   codecForExchangePurseStatus,
   getTotalPeerPaymentCost,
   queryCoinInfosForSelection,
@@ -76,6 +82,84 @@ import { checkLogicInvariant } from "../util/invariants.js";
 
 const logger = new Logger("pay-peer-pull-debit.ts");
 
+async function handlePurseCreationConflict(
+  ws: InternalWalletState,
+  peerPullInc: PeerPullPaymentIncomingRecord,
+  resp: HttpResponse,
+): Promise<OperationAttemptResult> {
+  const pursePub = peerPullInc.pursePub;
+  const errResp = await readTalerErrorResponse(resp);
+  if (errResp.code !== TalerErrorCode.EXCHANGE_GENERIC_INSUFFICIENT_FUNDS) {
+    await failPeerPullDebitTransaction(ws, pursePub);
+    return OperationAttemptResult.finishedEmpty();
+  }
+
+  // FIXME: Properly parse!
+  const brokenCoinPub = (errResp as any).coin_pub;
+  logger.trace(`excluded broken coin pub=${brokenCoinPub}`);
+
+  if (!brokenCoinPub) {
+    // FIXME: Details!
+    throw new TalerProtocolViolationError();
+  }
+
+  const instructedAmount = Amounts.parseOrThrow(
+    peerPullInc.contractTerms.amount,
+  );
+
+  const sel = peerPullInc.coinSel;
+  if (!sel) {
+    throw Error("invalid state (coin selection expected)");
+  }
+
+  const repair: PeerCoinRepair = {
+    coinPubs: sel.coinPubs,
+    contribs: sel.contributions.map((x) => Amounts.parseOrThrow(x)),
+    exchangeBaseUrl: peerPullInc.exchangeBaseUrl,
+  };
+
+  const coinSelRes = await selectPeerCoins(ws, { instructedAmount, repair });
+
+  if (coinSelRes.type == "failure") {
+    // FIXME: Details!
+    throw Error(
+      "insufficient balance to re-select coins to repair double spending",
+    );
+  }
+
+  const totalAmount = await getTotalPeerPaymentCost(
+    ws,
+    coinSelRes.result.coins,
+  );
+
+  await ws.db
+    .mktx((x) => [x.peerPullPaymentIncoming])
+    .runReadWrite(async (tx) => {
+      const myPpi = await tx.peerPullPaymentIncoming.get(
+        peerPullInc.peerPullPaymentIncomingId,
+      );
+      if (!myPpi) {
+        return;
+      }
+      switch (myPpi.status) {
+        case PeerPullDebitRecordStatus.PendingDeposit:
+        case PeerPullDebitRecordStatus.SuspendedDeposit: {
+          const sel = coinSelRes.result;
+          myPpi.coinSel = {
+            coinPubs: sel.coins.map((x) => x.coinPub),
+            contributions: sel.coins.map((x) => x.contribution),
+            totalCost: Amounts.stringify(totalAmount),
+          };
+          break;
+        }
+        default:
+          return;
+      }
+      await tx.peerPullPaymentIncoming.put(myPpi);
+    });
+  return OperationAttemptResult.finishedEmpty();
+}
+
 async function processPeerPullDebitPendingDeposit(
   ws: InternalWalletState,
   peerPullInc: PeerPullPaymentIncomingRecord,
@@ -118,81 +202,98 @@ async function processPeerPullDebitPendingDeposit(
     method: "POST",
     body: depositPayload,
   });
-  if (httpResp.status === HttpStatusCode.Gone) {
+  switch (httpResp.status) {
-    const transitionInfo = await ws.db
+    case HttpStatusCode.Ok: {
-      .mktx((x) => [
+      const resp = await readSuccessResponseJsonOrThrow(
-        x.peerPullPaymentIncoming,
+        httpResp,
-        x.refreshGroups,
+        codecForAny(),
-        x.denominations,
+      );
-        x.coinAvailability,
+      logger.trace(`purse deposit response: ${j2s(resp)}`);
-        x.coins,
-      ])
-      .runReadWrite(async (tx) => {
-        const pi = await tx.peerPullPaymentIncoming.get(
-          peerPullPaymentIncomingId,
-        );
-        if (!pi) {
-          throw Error("peer pull payment not found anymore");
-        }
-        if (pi.status !== PeerPullDebitRecordStatus.PendingDeposit) {
-          return;
-        }
-        const oldTxState = computePeerPullDebitTransactionState(pi);
 
-        const currency = Amounts.currencyOf(pi.totalCostEstimated);
+      const transitionInfo = await ws.db
-        const coinPubs: CoinRefreshRequest[] = [];
+        .mktx((x) => [x.peerPullPaymentIncoming])
+        .runReadWrite(async (tx) => {
+          const pi = await tx.peerPullPaymentIncoming.get(
+            peerPullPaymentIncomingId,
+          );
+          if (!pi) {
+            throw Error("peer pull payment not found anymore");
+          }
+          if (pi.status !== PeerPullDebitRecordStatus.PendingDeposit) {
+            return;
+          }
+          const oldTxState = computePeerPullDebitTransactionState(pi);
+          pi.status = PeerPullDebitRecordStatus.DonePaid;
+          const newTxState = computePeerPullDebitTransactionState(pi);
+          await tx.peerPullPaymentIncoming.put(pi);
+          return { oldTxState, newTxState };
+        });
+      notifyTransition(ws, transactionId, transitionInfo);
+      break;
+    }
+    case HttpStatusCode.Gone: {
+      const transitionInfo = await ws.db
+        .mktx((x) => [
+          x.peerPullPaymentIncoming,
+          x.refreshGroups,
+          x.denominations,
+          x.coinAvailability,
+          x.coins,
+        ])
+        .runReadWrite(async (tx) => {
+          const pi = await tx.peerPullPaymentIncoming.get(
+            peerPullPaymentIncomingId,
+          );
+          if (!pi) {
+            throw Error("peer pull payment not found anymore");
+          }
+          if (pi.status !== PeerPullDebitRecordStatus.PendingDeposit) {
+            return;
+          }
+          const oldTxState = computePeerPullDebitTransactionState(pi);
 
-        if (!pi.coinSel) {
+          const currency = Amounts.currencyOf(pi.totalCostEstimated);
-          throw Error("invalid db state");
+          const coinPubs: CoinRefreshRequest[] = [];
-        }
 
-        for (let i = 0; i < pi.coinSel.coinPubs.length; i++) {
+          if (!pi.coinSel) {
-          coinPubs.push({
+            throw Error("invalid db state");
-            amount: pi.coinSel.contributions[i],
+          }
-            coinPub: pi.coinSel.coinPubs[i],
-          });
-        }
 
-        const refresh = await createRefreshGroup(
+          for (let i = 0; i < pi.coinSel.coinPubs.length; i++) {
-          ws,
+            coinPubs.push({
-          tx,
+              amount: pi.coinSel.contributions[i],
-          currency,
+              coinPub: pi.coinSel.coinPubs[i],
-          coinPubs,
+            });
-          RefreshReason.AbortPeerPushDebit,
+          }
-        );
 
-        pi.status = PeerPullDebitRecordStatus.AbortingRefresh;
+          const refresh = await createRefreshGroup(
-        pi.abortRefreshGroupId = refresh.refreshGroupId;
+            ws,
-        const newTxState = computePeerPullDebitTransactionState(pi);
+            tx,
-        await tx.peerPullPaymentIncoming.put(pi);
+            currency,
-        return { oldTxState, newTxState };
+            coinPubs,
-      });
+            RefreshReason.AbortPeerPushDebit,
-    notifyTransition(ws, transactionId, transitionInfo);
+          );
-  } else {
-    const resp = await readSuccessResponseJsonOrThrow(httpResp, codecForAny());
-    logger.trace(`purse deposit response: ${j2s(resp)}`);
 
-    const transitionInfo = await ws.db
+          pi.status = PeerPullDebitRecordStatus.AbortingRefresh;
-      .mktx((x) => [x.peerPullPaymentIncoming])
+          pi.abortRefreshGroupId = refresh.refreshGroupId;
-      .runReadWrite(async (tx) => {
+          const newTxState = computePeerPullDebitTransactionState(pi);
-        const pi = await tx.peerPullPaymentIncoming.get(
+          await tx.peerPullPaymentIncoming.put(pi);
-          peerPullPaymentIncomingId,
+          return { oldTxState, newTxState };
-        );
+        });
-        if (!pi) {
+      notifyTransition(ws, transactionId, transitionInfo);
-          throw Error("peer pull payment not found anymore");
+      break;
-        }
+    }
-        if (pi.status !== PeerPullDebitRecordStatus.PendingDeposit) {
+    case HttpStatusCode.Conflict: {
-          return;
+      return handlePurseCreationConflict(ws, peerPullInc, httpResp);
-        }
+    }
-        const oldTxState = computePeerPullDebitTransactionState(pi);
+    default: {
-        pi.status = PeerPullDebitRecordStatus.DonePaid;
+      const errResp = await readTalerErrorResponse(httpResp);
-        const newTxState = computePeerPullDebitTransactionState(pi);
+      return {
-        await tx.peerPullPaymentIncoming.put(pi);
+        type: OperationAttemptResultType.Error,
-        return { oldTxState, newTxState };
+        errorDetail: errResp,
-      });
+      };
-    notifyTransition(ws, transactionId, transitionInfo);
+    }
   }
-
   return {
     type: OperationAttemptResultType.Finished,
     result: undefined,
@@ -434,7 +535,7 @@ export async function preparePeerPullDebit(
 
   const getPurseUrl = new URL(`purses/${pursePub}/merge`, exchangeBaseUrl);
 
-  const purseHttpResp = await ws.http.get(getPurseUrl.href);
+  const purseHttpResp = await ws.http.fetch(getPurseUrl.href);
 
   const purseStatus = await readSuccessResponseJsonOrThrow(
     purseHttpResp,

packages/taler-wallet-core/src/operations/pay-peer-push-debit.ts

@@ -28,6 +28,7 @@ import {
   TalerError,
   TalerErrorCode,
   TalerPreciseTimestamp,
+  TalerProtocolViolationError,
   TalerUriAction,
   TransactionAction,
   TransactionMajorState,
@@ -47,8 +48,13 @@ import {
   getTotalPeerPaymentCost,
   codecForExchangePurseStatus,
   queryCoinInfosForSelection,
+  PeerCoinRepair,
 } from "./pay-peer-common.js";
-import { readSuccessResponseJsonOrThrow } from "@gnu-taler/taler-util/http";
+import {
+  HttpResponse,
+  readSuccessResponseJsonOrThrow,
+  readTalerErrorResponse,
+} from "@gnu-taler/taler-util/http";
 import {
   PeerPushPaymentInitiationRecord,
   PeerPushPaymentInitiationStatus,
@@ -97,6 +103,73 @@ export async function checkPeerPushDebit(
   };
 }
 
+async function handlePurseCreationConflict(
+  ws: InternalWalletState,
+  peerPushInitiation: PeerPushPaymentInitiationRecord,
+  resp: HttpResponse,
+): Promise<OperationAttemptResult> {
+  const pursePub = peerPushInitiation.pursePub;
+  const errResp = await readTalerErrorResponse(resp);
+  if (errResp.code !== TalerErrorCode.EXCHANGE_GENERIC_INSUFFICIENT_FUNDS) {
+    await failPeerPushDebitTransaction(ws, pursePub);
+    return OperationAttemptResult.finishedEmpty();
+  }
+
+  // FIXME: Properly parse!
+  const brokenCoinPub = (errResp as any).coin_pub;
+  logger.trace(`excluded broken coin pub=${brokenCoinPub}`);
+
+  if (!brokenCoinPub) {
+    // FIXME: Details!
+    throw new TalerProtocolViolationError();
+  }
+
+  const instructedAmount = Amounts.parseOrThrow(peerPushInitiation.amount);
+
+  const repair: PeerCoinRepair = {
+    coinPubs: peerPushInitiation.coinSel.coinPubs,
+    contribs: peerPushInitiation.coinSel.contributions.map((x) =>
+      Amounts.parseOrThrow(x),
+    ),
+    exchangeBaseUrl: peerPushInitiation.exchangeBaseUrl,
+  };
+
+  const coinSelRes = await selectPeerCoins(ws, { instructedAmount, repair });
+
+  if (coinSelRes.type == "failure") {
+    // FIXME: Details!
+    throw Error(
+      "insufficient balance to re-select coins to repair double spending",
+    );
+  }
+
+  await ws.db
+    .mktx((x) => [x.peerPushPaymentInitiations])
+    .runReadWrite(async (tx) => {
+      const myPpi = await tx.peerPushPaymentInitiations.get(
+        peerPushInitiation.pursePub,
+      );
+      if (!myPpi) {
+        return;
+      }
+      switch (myPpi.status) {
+        case PeerPushPaymentInitiationStatus.PendingCreatePurse:
+        case PeerPushPaymentInitiationStatus.SuspendedCreatePurse: {
+          const sel = coinSelRes.result;
+          myPpi.coinSel =  {
+            coinPubs: sel.coins.map((x) => x.coinPub),
+            contributions: sel.coins.map((x) => x.contribution),
+          }
+          break;
+        }
+        default:
+          return;
+      }
+      await tx.peerPushPaymentInitiations.put(myPpi);
+    });
+  return OperationAttemptResult.finishedEmpty();
+}
+
 async function processPeerPushDebitCreateReserve(
   ws: InternalWalletState,
   peerPushInitiation: PeerPushPaymentInitiationRecord,
@@ -175,6 +248,27 @@ async function processPeerPushDebitCreateReserve(
 
   logger.info(`resp: ${j2s(resp)}`);
 
+  switch (httpResp.status) {
+    case HttpStatusCode.Ok:
+      break;
+    case HttpStatusCode.Forbidden: {
+      // FIXME: Store this error!
+      await failPeerPushDebitTransaction(ws, pursePub);
+      return OperationAttemptResult.finishedEmpty();
+    }
+    case HttpStatusCode.Conflict: {
+      // Handle double-spending
+      return handlePurseCreationConflict(ws, peerPushInitiation, resp);
+    }
+    default: {
+      const errResp = await readTalerErrorResponse(resp);
+      return {
+        type: OperationAttemptResultType.Error,
+        errorDetail: errResp,
+      };
+    }
+  }
+
   if (httpResp.status !== HttpStatusCode.Ok) {
     // FIXME: do proper error reporting
     throw Error("got error response from exchange");
@@ -710,17 +804,17 @@ export async function failPeerPushDebitTransaction(
       switch (pushDebitRec.status) {
         case PeerPushPaymentInitiationStatus.AbortingRefresh:
         case PeerPushPaymentInitiationStatus.SuspendedAbortingRefresh:
-          // FIXME: We also need to abort the refresh group!
+          // FIXME: What to do about the refresh group?
-          newStatus = PeerPushPaymentInitiationStatus.Aborted;
+          newStatus = PeerPushPaymentInitiationStatus.Failed;
           break;
         case PeerPushPaymentInitiationStatus.AbortingDeletePurse:
         case PeerPushPaymentInitiationStatus.SuspendedAbortingDeletePurse:
-          newStatus = PeerPushPaymentInitiationStatus.Aborted;
-          break;
         case PeerPushPaymentInitiationStatus.PendingReady:
         case PeerPushPaymentInitiationStatus.SuspendedReady:
         case PeerPushPaymentInitiationStatus.SuspendedCreatePurse:
         case PeerPushPaymentInitiationStatus.PendingCreatePurse:
+          newStatus = PeerPushPaymentInitiationStatus.Failed;
+          break;
         case PeerPushPaymentInitiationStatus.Done:
         case PeerPushPaymentInitiationStatus.Aborted:
         case PeerPushPaymentInitiationStatus.Failed: