wallet-core: implement retries for peer push payments

packages/taler-wallet-core/src/crypto/cryptoImplementation.ts

@@ -445,17 +445,19 @@ export interface SignPurseCreationRequest {
   minAge: number;
 }
 
-export interface SignPurseDepositsRequest {
+export interface SpendCoinDetails {
-  pursePub: string;
+  coinPub: string;
-  exchangeBaseUrl: string;
-  coins: {
-    coinPub: string;
     coinPriv: string;
     contribution: AmountString;
     denomPubHash: string;
     denomSig: UnblindedSignature;
     ageCommitmentProof: AgeCommitmentProof | undefined;
-  }[];
+}
+
+export interface SignPurseDepositsRequest {
+  pursePub: string;
+  exchangeBaseUrl: string;
+  coins: SpendCoinDetails[];
 }
 
 export interface SignPurseDepositsResponse {
@@ -1451,25 +1453,24 @@ export const nativeCryptoR: TalerCryptoInterfaceR = {
     tci: TalerCryptoInterfaceR,
     req: EncryptContractRequest,
   ): Promise<EncryptContractResponse> {
-    const contractKeyPair = await this.createEddsaKeypair(tci, {});
+
     const enc = await encryptContractForMerge(
       decodeCrock(req.pursePub),
-      decodeCrock(contractKeyPair.priv),
+      decodeCrock(req.contractPriv),
       decodeCrock(req.mergePriv),
       req.contractTerms,
     );
     const sigBlob = buildSigPS(TalerSignaturePurpose.WALLET_PURSE_ECONTRACT)
       .put(hash(enc))
-      .put(decodeCrock(contractKeyPair.pub))
+      .put(decodeCrock(req.contractPub))
       .build();
     const sig = eddsaSign(sigBlob, decodeCrock(req.pursePriv));
     return {
       econtract: {
-        contract_pub: contractKeyPair.pub,
+        contract_pub: req.contractPub,
         econtract: encodeCrock(enc),
         econtract_sig: encodeCrock(sig),
       },
-      contractPriv: contractKeyPair.priv,
     };
   },
   async decryptContractForMerge(

packages/taler-wallet-core/src/crypto/cryptoTypes.ts

@@ -176,17 +176,15 @@ export interface EncryptedContract {
 
 export interface EncryptContractRequest {
   contractTerms: any;
-
+  contractPriv: string;
+  contractPub: string;
   pursePub: string;
   pursePriv: string;
-
   mergePriv: string;
 }
 
 export interface EncryptContractResponse {
   econtract: EncryptedContract;
-
-  contractPriv: string;
 }
 
 export interface EncryptContractForDepositRequest {

packages/taler-wallet-core/src/db.ts

@@ -57,6 +57,7 @@ import {
   AttentionInfo,
   AbsoluteTime,
   Logger,
+  CoinPublicKeyString,
 } from "@gnu-taler/taler-util";
 import {
   DbAccess,
@@ -1692,6 +1693,11 @@ export enum PeerPushPaymentInitiationStatus {
   PurseCreated = 50 /* DORMANT_START */,
 }
 
+export interface PeerPushPaymentCoinSelection {
+  contributions: AmountString[];
+  coinPubs: CoinPublicKeyString[];
+}
+
 /**
  * Record for a push P2P payment that this wallet initiated.
  */
@@ -1701,8 +1707,13 @@ export interface PeerPushPaymentInitiationRecord {
    */
   exchangeBaseUrl: string;
 
+  /**
+   * Instructed amount.
+   */
   amount: AmountString;
 
+  coinSel: PeerPushPaymentCoinSelection;
+
   contractTermsHash: HashCodeString;
 
   /**
@@ -1727,6 +1738,9 @@ export interface PeerPushPaymentInitiationRecord {
   mergePriv: string;
 
   contractPriv: string;
+  contractPub: string;
+
+  contractTerms: PeerContractTerms;
 
   purseExpiration: TalerProtocolTimestamp;
 

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

@@ -68,9 +68,11 @@ import {
   UnblindedSignature,
   WalletAccountMergeFlags,
 } from "@gnu-taler/taler-util";
+import { SpendCoinDetails } from "../crypto/cryptoImplementation.js";
 import {
   OperationStatus,
   PeerPullPaymentIncomingStatus,
+  PeerPushPaymentCoinSelection,
   PeerPushPaymentIncomingRecord,
   PeerPushPaymentInitiationStatus,
   ReserveRecord,
@@ -80,17 +82,26 @@ import {
 } from "../db.js";
 import { TalerError } from "../errors.js";
 import { InternalWalletState } from "../internal-wallet-state.js";
-import { makeTransactionId, spendCoins } from "../operations/common.js";
+import {
+  makeTransactionId,
+  runOperationWithErrorReporting,
+  spendCoins,
+} from "../operations/common.js";
 import { readSuccessResponseJsonOrThrow } from "../util/http.js";
 import { checkDbInvariant } from "../util/invariants.js";
 import { GetReadOnlyAccess } from "../util/query.js";
+import {
+  OperationAttemptResult,
+  OperationAttemptResultType,
+  RetryTags,
+} from "../util/retries.js";
 import { getPeerPaymentBalanceDetailsInTx } from "./balance.js";
 import { updateExchangeFromUrl } from "./exchanges.js";
 import { internalCreateWithdrawalGroup } from "./withdraw.js";
 
 const logger = new Logger("operations/peer-to-peer.ts");
 
-export interface PeerCoinSelection {
+export interface PeerCoinSelectionDetails {
   exchangeBaseUrl: string;
 
   /**
@@ -111,6 +122,9 @@ export interface PeerCoinSelection {
   depositFees: AmountJson;
 }
 
+/**
+ * Information about a selected coin for peer to peer payments.
+ */
 interface CoinInfo {
   /**
    * Public key of the coin.
@@ -131,16 +145,52 @@ interface CoinInfo {
   denomSig: UnblindedSignature;
 
   maxAge: number;
+
   ageCommitmentProof?: AgeCommitmentProof;
 }
 
 export type SelectPeerCoinsResult =
-  | { type: "success"; result: PeerCoinSelection }
+  | { type: "success"; result: PeerCoinSelectionDetails }
   | {
       type: "failure";
       insufficientBalanceDetails: PayPeerInsufficientBalanceDetails;
     };
 
+export async function queryCoinInfosForSelection(
+  ws: InternalWalletState,
+  csel: PeerPushPaymentCoinSelection,
+): Promise<SpendCoinDetails[]> {
+  let infos: SpendCoinDetails[] = [];
+  await ws.db
+    .mktx((x) => [x.coins, x.denominations])
+    .runReadOnly(async (tx) => {
+      for (let i = 0; i < csel.coinPubs.length; i++) {
+        const coin = await tx.coins.get(csel.coinPubs[i]);
+        if (!coin) {
+          throw Error("coin not found anymore");
+        }
+        const denom = await ws.getDenomInfo(
+          ws,
+          tx,
+          coin.exchangeBaseUrl,
+          coin.denomPubHash,
+        );
+        if (!denom) {
+          throw Error("denom for coin not found anymore");
+        }
+        infos.push({
+          coinPriv: coin.coinPriv,
+          coinPub: coin.coinPub,
+          denomPubHash: coin.denomPubHash,
+          denomSig: coin.denomSig,
+          ageCommitmentProof: coin.ageCommitmentProof,
+          contribution: csel.contributions[i],
+        });
+      }
+    });
+  return infos;
+}
+
 export async function selectPeerCoins(
   ws: InternalWalletState,
   tx: GetReadOnlyAccess<{
@@ -228,7 +278,7 @@ export async function selectPeerCoins(
       lastDepositFee = coin.feeDeposit;
     }
     if (Amounts.cmp(amountAcc, instructedAmount) >= 0) {
-      const res: PeerCoinSelection = {
+      const res: PeerCoinSelectionDetails = {
         exchangeBaseUrl: exch.baseUrl,
         coins: resCoins,
         depositFees: depositFeesAcc,
@@ -290,6 +340,94 @@ export async function preparePeerPushPayment(
   };
 }
 
+export async function processPeerPushOutgoing(
+  ws: InternalWalletState,
+  pursePub: string,
+): Promise<OperationAttemptResult> {
+  const peerPushInitiation = await ws.db
+    .mktx((x) => [x.peerPushPaymentInitiations])
+    .runReadOnly(async (tx) => {
+      return tx.peerPushPaymentInitiations.get(pursePub);
+    });
+  if (!peerPushInitiation) {
+    throw Error("peer push payment not found");
+  }
+
+  const purseExpiration = peerPushInitiation.purseExpiration;
+  const hContractTerms = peerPushInitiation.contractTermsHash;
+
+  const purseSigResp = await ws.cryptoApi.signPurseCreation({
+    hContractTerms,
+    mergePub: peerPushInitiation.mergePub,
+    minAge: 0,
+    purseAmount: peerPushInitiation.amount,
+    purseExpiration,
+    pursePriv: peerPushInitiation.pursePriv,
+  });
+
+  const coins = await queryCoinInfosForSelection(
+    ws,
+    peerPushInitiation.coinSel,
+  );
+
+  const depositSigsResp = await ws.cryptoApi.signPurseDeposits({
+    exchangeBaseUrl: peerPushInitiation.exchangeBaseUrl,
+    pursePub: peerPushInitiation.pursePub,
+    coins,
+  });
+
+  const econtractResp = await ws.cryptoApi.encryptContractForMerge({
+    contractTerms: peerPushInitiation.contractTerms,
+    mergePriv: peerPushInitiation.mergePriv,
+    pursePriv: peerPushInitiation.pursePriv,
+    pursePub: peerPushInitiation.pursePub,
+    contractPriv: peerPushInitiation.contractPriv,
+    contractPub: peerPushInitiation.contractPub,
+  });
+
+  const createPurseUrl = new URL(
+    `purses/${peerPushInitiation.pursePub}/create`,
+    peerPushInitiation.exchangeBaseUrl,
+  );
+
+  const httpResp = await ws.http.postJson(createPurseUrl.href, {
+    amount: peerPushInitiation.amount,
+    merge_pub: peerPushInitiation.mergePub,
+    purse_sig: purseSigResp.sig,
+    h_contract_terms: hContractTerms,
+    purse_expiration: purseExpiration,
+    deposits: depositSigsResp.deposits,
+    min_age: 0,
+    econtract: econtractResp.econtract,
+  });
+
+  const resp = await httpResp.json();
+
+  logger.info(`resp: ${j2s(resp)}`);
+
+  if (httpResp.status !== 200) {
+    throw Error("got error response from exchange");
+  }
+
+  await ws.db
+    .mktx((x) => [x.peerPushPaymentInitiations])
+    .runReadWrite(async (tx) => {
+      const ppi = await tx.peerPushPaymentInitiations.get(pursePub);
+      if (!ppi) {
+        return;
+      }
+      ppi.status = PeerPushPaymentInitiationStatus.PurseCreated;
+    });
+
+  return {
+    type: OperationAttemptResultType.Finished,
+    result: undefined,
+  };
+}
+
+/**
+ * Initiate sending a peer-to-peer push payment.
+ */
 export async function initiatePeerToPeerPush(
   ws: InternalWalletState,
   req: InitiatePeerPushPaymentRequest,
@@ -305,13 +443,7 @@ export async function initiatePeerToPeerPush(
 
   const hContractTerms = ContractTermsUtil.hashContractTerms(contractTerms);
 
-  const econtractResp = await ws.cryptoApi.encryptContractForMerge({
+  const contractKeyPair = await ws.cryptoApi.createEddsaKeypair({});
-    contractTerms,
-    mergePriv: mergePair.priv,
-    pursePriv: pursePair.priv,
-    pursePub: pursePair.pub,
-  });
-
   const coinSelRes: SelectPeerCoinsResult = await ws.db
     .mktx((x) => [
       x.exchanges,
@@ -320,7 +452,6 @@ export async function initiatePeerToPeerPush(
       x.coinAvailability,
       x.denominations,
       x.refreshGroups,
-      x.peerPullPaymentInitiations,
       x.peerPushPaymentInitiations,
     ])
     .runReadWrite(async (tx) => {
@@ -342,7 +473,8 @@ export async function initiatePeerToPeerPush(
 
       await tx.peerPushPaymentInitiations.add({
         amount: Amounts.stringify(instructedAmount),
-        contractPriv: econtractResp.contractPriv,
+        contractPriv: contractKeyPair.priv,
+        contractPub: contractKeyPair.pub,
         contractTermsHash: hContractTerms,
         exchangeBaseUrl: sel.exchangeBaseUrl,
         mergePriv: mergePair.priv,
@@ -351,8 +483,12 @@ export async function initiatePeerToPeerPush(
         pursePriv: pursePair.priv,
         pursePub: pursePair.pub,
         timestampCreated: TalerProtocolTimestamp.now(),
-        // FIXME: Only set the later when the purse is actually created!
+        status: PeerPushPaymentInitiationStatus.Initiated,
-        status: PeerPushPaymentInitiationStatus.PurseCreated,
+        contractTerms: contractTerms,
+        coinSel: {
+          coinPubs: sel.coins.map((x) => x.coinPub),
+          contributions: sel.coins.map((x) => x.contribution),
+        },
       });
 
       await tx.contractTerms.put({
@@ -373,53 +509,22 @@ export async function initiatePeerToPeerPush(
     );
   }
 
-  const purseSigResp = await ws.cryptoApi.signPurseCreation({
+  await runOperationWithErrorReporting(
-    hContractTerms,
+    ws,
-    mergePub: mergePair.pub,
+    RetryTags.byPeerPushPaymentInitiationPursePub(pursePair.pub),
-    minAge: 0,
+    async () => {
-    purseAmount: Amounts.stringify(instructedAmount),
+      return await processPeerPushOutgoing(ws, pursePair.pub);
-    purseExpiration,
+    },
-    pursePriv: pursePair.priv,
-  });
-
-  const depositSigsResp = await ws.cryptoApi.signPurseDeposits({
-    exchangeBaseUrl: coinSelRes.result.exchangeBaseUrl,
-    pursePub: pursePair.pub,
-    coins: coinSelRes.result.coins,
-  });
-
-  const createPurseUrl = new URL(
-    `purses/${pursePair.pub}/create`,
-    coinSelRes.result.exchangeBaseUrl,
   );
 
-  const httpResp = await ws.http.postJson(createPurseUrl.href, {
-    amount: Amounts.stringify(instructedAmount),
-    merge_pub: mergePair.pub,
-    purse_sig: purseSigResp.sig,
-    h_contract_terms: hContractTerms,
-    purse_expiration: purseExpiration,
-    deposits: depositSigsResp.deposits,
-    min_age: 0,
-    econtract: econtractResp.econtract,
-  });
-
-  const resp = await httpResp.json();
-
-  logger.info(`resp: ${j2s(resp)}`);
-
-  if (httpResp.status !== 200) {
-    throw Error("got error response from exchange");
-  }
-
   return {
-    contractPriv: econtractResp.contractPriv,
+    contractPriv: contractKeyPair.priv,
     mergePriv: mergePair.priv,
     pursePub: pursePair.pub,
     exchangeBaseUrl: coinSelRes.result.exchangeBaseUrl,
     talerUri: constructPayPushUri({
       exchangeBaseUrl: coinSelRes.result.exchangeBaseUrl,
-      contractPriv: econtractResp.contractPriv,
+      contractPriv: contractKeyPair.priv,
     }),
     transactionId: makeTransactionId(
       TransactionType.PeerPushDebit,

packages/taler-wallet-core/src/pending-types.ts

@@ -24,11 +24,7 @@
 /**
  * Imports.
  */
-import {
+import { TalerErrorDetail, AbsoluteTime } from "@gnu-taler/taler-util";
-  TalerErrorDetail,
-  AbsoluteTime,
-  TalerProtocolTimestamp,
-} from "@gnu-taler/taler-util";
 import { RetryInfo } from "./util/retries.js";
 
 export enum PendingTaskType {
@@ -41,6 +37,7 @@ export enum PendingTaskType {
   Withdraw = "withdraw",
   Deposit = "deposit",
   Backup = "backup",
+  PeerPushOutgoing = "peer-push-outgoing",
 }
 
 /**
@@ -57,6 +54,7 @@ export type PendingTaskInfo = PendingTaskInfoCommon &
     | PendingRecoupTask
     | PendingDepositTask
     | PendingBackupTask
+    | PendingPeerPushOutgoingTask
   );
 
 export interface PendingBackupTask {
@@ -74,6 +72,14 @@ export interface PendingExchangeUpdateTask {
   lastError: TalerErrorDetail | undefined;
 }
 
+/**
+ * The wallet wants to send a peer push payment.
+ */
+export interface PendingPeerPushOutgoingTask {
+  type: PendingTaskType.PeerPushOutgoing;
+  pursePub: string;
+}
+
 /**
  * The wallet should check whether coins from this exchange
  * need to be auto-refreshed.

packages/taler-wallet-core/src/util/retries.ts

@@ -30,6 +30,7 @@ import {
   BackupProviderRecord,
   DepositGroupRecord,
   ExchangeRecord,
+  PeerPushPaymentInitiationRecord,
   PurchaseRecord,
   RecoupGroupRecord,
   RefreshGroupRecord,
@@ -200,9 +201,17 @@ export namespace RetryTags {
   export function forBackup(backupRecord: BackupProviderRecord): string {
     return `${PendingTaskType.Backup}:${backupRecord.baseUrl}`;
   }
+  export function forPeerPushPaymentInitiation(
+    ppi: PeerPushPaymentInitiationRecord,
+  ): string {
+    return `${PendingTaskType.PeerPushOutgoing}:${ppi.pursePub}`;
+  }
   export function byPaymentProposalId(proposalId: string): string {
     return `${PendingTaskType.Purchase}:${proposalId}`;
   }
+  export function byPeerPushPaymentInitiationPursePub(pursePub: string): string {
+    return `${PendingTaskType.PeerPushOutgoing}:${pursePub}`;
+  }
 }
 
 export async function scheduleRetryInTx(

packages/taler-wallet-core/src/wallet.ts

@@ -198,6 +198,7 @@ import {
   initiatePeerToPeerPush,
   preparePeerPullPayment,
   preparePeerPushPayment,
+  processPeerPushOutgoing,
 } from "./operations/pay-peer.js";
 import { getPendingOperations } from "./operations/pending.js";
 import {
@@ -317,6 +318,8 @@ async function callOperationHandler(
     }
     case PendingTaskType.Backup:
       return await processBackupForProvider(ws, pending.backupProviderBaseUrl);
+    case PendingTaskType.PeerPushOutgoing:
+      return await processPeerPushOutgoing(ws, pending.pursePub);
     default:
       return assertUnreachable(pending);
   }