wallet-core: emit DD37 self-transition notifications with errors

1 files changed, 504 insertions, 77 deletions

diff --git a/packages/taler-wallet-core/src/operations/common.ts b/packages/taler-wallet-core/src/operations/common.ts
index ad18767c4..293870a18 100644
--- a/packages/taler-wallet-core/src/operations/common.ts
+++ b/packages/taler-wallet-core/src/operations/common.ts
@@ -18,42 +18,56 @@
  * Imports.
  */
 import {
+  AbsoluteTime,
   AgeRestriction,
   AmountJson,
   Amounts,
   CancellationToken,
   CoinRefreshRequest,
   CoinStatus,
+  Duration,
+  ErrorInfoSummary,
   ExchangeEntryStatus,
   ExchangeListItem,
   ExchangeTosStatus,
   getErrorDetailFromException,
   j2s,
   Logger,
+  NotificationType,
   OperationErrorInfo,
   RefreshReason,
   TalerErrorCode,
   TalerErrorDetail,
   TombstoneIdStr,
   TransactionIdStr,
+  TransactionType,
+  WalletNotification,
 } from "@gnu-taler/taler-util";
 import {
   WalletStoresV1,
   CoinRecord,
   ExchangeDetailsRecord,
   ExchangeRecord,
+  BackupProviderRecord,
+  DepositGroupRecord,
+  PeerPullPaymentIncomingRecord,
+  PeerPullPaymentInitiationRecord,
+  PeerPushPaymentIncomingRecord,
+  PeerPushPaymentInitiationRecord,
+  PurchaseRecord,
+  RecoupGroupRecord,
+  RefreshGroupRecord,
+  TipRecord,
+  WithdrawalGroupRecord,
 } from "../db.js";
 import { makeErrorDetail, TalerError } from "@gnu-taler/taler-util";
 import { InternalWalletState } from "../internal-wallet-state.js";
 import { checkDbInvariant, checkLogicInvariant } from "../util/invariants.js";
-import { GetReadWriteAccess } from "../util/query.js";
-import {
-  OperationAttemptResult,
-  OperationAttemptResultType,
-  RetryInfo,
-} from "../util/retries.js";
+import { GetReadOnlyAccess, GetReadWriteAccess } from "../util/query.js";
 import { CryptoApiStoppedError } from "../crypto/workers/crypto-dispatcher.js";
-import { TaskId } from "../pending-types.js";
+import { PendingTaskType, TaskId } from "../pending-types.js";
+import { assertUnreachable } from "../util/assertUnreachable.js";
+import { constructTransactionIdentifier } from "./transactions.js";
 
 const logger = new Logger("operations/common.ts");
 
@@ -197,68 +211,185 @@ export async function spendCoins(
   );
 }
 
-export async function storeOperationError(
+/**
+ * Convert the task ID for a task that processes a transaction int
+ * the ID for the transaction.
+ */
+function convertTaskToTransactionId(
+  taskId: string,
+): TransactionIdStr | undefined {
+  const parsedTaskId = parseTaskIdentifier(taskId);
+  switch (parsedTaskId.tag) {
+    case PendingTaskType.PeerPullCredit:
+      return constructTransactionIdentifier({
+        tag: TransactionType.PeerPullCredit,
+        pursePub: parsedTaskId.pursePub,
+      });
+    case PendingTaskType.PeerPullDebit:
+      return constructTransactionIdentifier({
+        tag: TransactionType.PeerPullDebit,
+        peerPullPaymentIncomingId: parsedTaskId.peerPullPaymentIncomingId,
+      });
+    // FIXME: This doesn't distinguish internal-withdrawal.
+    // Maybe we should have a different task type for that as well?
+    // Or maybe transaction IDs should be valid task identifiers?
+    case PendingTaskType.Withdraw:
+      return constructTransactionIdentifier({
+        tag: TransactionType.Withdrawal,
+        withdrawalGroupId: parsedTaskId.withdrawalGroupId,
+      });
+    case PendingTaskType.PeerPushCredit:
+      return constructTransactionIdentifier({
+        tag: TransactionType.PeerPushCredit,
+        peerPushPaymentIncomingId: parsedTaskId.peerPushPaymentIncomingId,
+      });
+    case PendingTaskType.Deposit:
+      return constructTransactionIdentifier({
+        tag: TransactionType.Deposit,
+        depositGroupId: parsedTaskId.depositGroupId,
+      });
+    case PendingTaskType.Refresh:
+      return constructTransactionIdentifier({
+        tag: TransactionType.Refresh,
+        refreshGroupId: parsedTaskId.refreshGroupId,
+      });
+    case PendingTaskType.TipPickup:
+      return constructTransactionIdentifier({
+        tag: TransactionType.Tip,
+        walletTipId: parsedTaskId.walletTipId,
+      });
+    case PendingTaskType.PeerPushDebit:
+      return constructTransactionIdentifier({
+        tag: TransactionType.PeerPushDebit,
+        pursePub: parsedTaskId.pursePub,
+      });
+    case PendingTaskType.Purchase:
+      return constructTransactionIdentifier({
+        tag: TransactionType.Payment,
+        proposalId: parsedTaskId.proposalId,
+      });
+    default:
+      return undefined;
+  }
+}
+
+/**
+ * For tasks that process a transaction,
+ * generate a state transition notification.
+ */
+async function taskToTransactionNotification(
+  ws: InternalWalletState,
+  tx: GetReadOnlyAccess<typeof WalletStoresV1>,
+  pendingTaskId: string,
+  e: TalerErrorDetail | undefined,
+): Promise<WalletNotification | undefined> {
+  const txId = convertTaskToTransactionId(pendingTaskId);
+  if (!txId) {
+    return undefined;
+  }
+  const txState = await ws.getTransactionState(ws, tx, txId);
+  if (!txState) {
+    return undefined;
+  }
+  const notif: WalletNotification = {
+    type: NotificationType.TransactionStateTransition,
+    transactionId: txId,
+    oldTxState: txState,
+    newTxState: txState,
+  };
+  if (e) {
+    notif.errorInfo = {
+      code: e.code as number,
+      hint: e.hint,
+    };
+  }
+  return notif;
+}
+
+async function storePendingTaskError(
   ws: InternalWalletState,
   pendingTaskId: string,
   e: TalerErrorDetail,
 ): Promise<void> {
-  await ws.db
-    .mktx((x) => [x.operationRetries])
-    .runReadWrite(async (tx) => {
-      let retryRecord = await tx.operationRetries.get(pendingTaskId);
-      if (!retryRecord) {
-        retryRecord = {
-          id: pendingTaskId,
-          lastError: e,
-          retryInfo: RetryInfo.reset(),
-        };
-      } else {
-        retryRecord.lastError = e;
-        retryRecord.retryInfo = RetryInfo.increment(retryRecord.retryInfo);
-      }
+  logger.info(`storing pending task error for ${pendingTaskId}`);
+  const maybeNotification = await ws.db.mktxAll().runReadWrite(async (tx) => {
+    let retryRecord = await tx.operationRetries.get(pendingTaskId);
+    if (!retryRecord) {
+      retryRecord = {
+        id: pendingTaskId,
+        lastError: e,
+        retryInfo: RetryInfo.reset(),
+      };
+    } else {
+      retryRecord.lastError = e;
+      retryRecord.retryInfo = RetryInfo.increment(retryRecord.retryInfo);
+    }
+    await tx.operationRetries.put(retryRecord);
+    return taskToTransactionNotification(ws, tx, pendingTaskId, e);
+  });
+  if (maybeNotification) {
+    ws.notify(maybeNotification);
+  }
+}
+
+export async function resetPendingTaskTimeout(
+  ws: InternalWalletState,
+  pendingTaskId: string,
+): Promise<void> {
+  const maybeNotification = await ws.db.mktxAll().runReadWrite(async (tx) => {
+    let retryRecord = await tx.operationRetries.get(pendingTaskId);
+    if (retryRecord) {
+      // Note that we don't reset the lastError, it should still be visible
+      // while the retry runs.
+      retryRecord.retryInfo = RetryInfo.reset();
       await tx.operationRetries.put(retryRecord);
-    });
+    }
+    return taskToTransactionNotification(ws, tx, pendingTaskId, undefined);
+  });
+  if (maybeNotification) {
+    ws.notify(maybeNotification);
+  }
 }
 
-export async function resetOperationTimeout(
+async function storePendingTaskPending(
   ws: InternalWalletState,
   pendingTaskId: string,
 ): Promise<void> {
-  await ws.db
-    .mktx((x) => [x.operationRetries])
-    .runReadWrite(async (tx) => {
-      let retryRecord = await tx.operationRetries.get(pendingTaskId);
-      if (retryRecord) {
-        // Note that we don't reset the lastError, it should still be visible
-        // while the retry runs.
-        retryRecord.retryInfo = RetryInfo.reset();
-        await tx.operationRetries.put(retryRecord);
+  const maybeNotification = await ws.db.mktxAll().runReadWrite(async (tx) => {
+    let retryRecord = await tx.operationRetries.get(pendingTaskId);
+    let hadError = false;
+    if (!retryRecord) {
+      retryRecord = {
+        id: pendingTaskId,
+        retryInfo: RetryInfo.reset(),
+      };
+    } else {
+      if (retryRecord.lastError) {
+        hadError = true;
       }
-    });
+      delete retryRecord.lastError;
+      retryRecord.retryInfo = RetryInfo.increment(retryRecord.retryInfo);
+    }
+    await tx.operationRetries.put(retryRecord);
+    return taskToTransactionNotification(ws, tx, pendingTaskId, undefined);
+  });
+  if (maybeNotification) {
+    ws.notify(maybeNotification);
+  }
 }
 
-export async function storeOperationPending(
+async function storePendingTaskFinished(
   ws: InternalWalletState,
   pendingTaskId: string,
 ): Promise<void> {
   await ws.db
     .mktx((x) => [x.operationRetries])
     .runReadWrite(async (tx) => {
-      let retryRecord = await tx.operationRetries.get(pendingTaskId);
-      if (!retryRecord) {
-        retryRecord = {
-          id: pendingTaskId,
-          retryInfo: RetryInfo.reset(),
-        };
-      } else {
-        delete retryRecord.lastError;
-        retryRecord.retryInfo = RetryInfo.increment(retryRecord.retryInfo);
-      }
-      await tx.operationRetries.put(retryRecord);
+      await tx.operationRetries.delete(pendingTaskId);
     });
 }
 
-export async function runOperationWithErrorReporting<T1, T2>(
+export async function runTaskWithErrorReporting<T1, T2>(
   ws: InternalWalletState,
   opId: TaskId,
   f: () => Promise<OperationAttemptResult<T1, T2>>,
@@ -268,13 +399,13 @@ export async function runOperationWithErrorReporting<T1, T2>(
     const resp = await f();
     switch (resp.type) {
       case OperationAttemptResultType.Error:
-        await storeOperationError(ws, opId, resp.errorDetail);
+        await storePendingTaskError(ws, opId, resp.errorDetail);
         return resp;
       case OperationAttemptResultType.Finished:
-        await storeOperationFinished(ws, opId);
+        await storePendingTaskFinished(ws, opId);
         return resp;
       case OperationAttemptResultType.Pending:
-        await storeOperationPending(ws, opId);
+        await storePendingTaskPending(ws, opId);
         return resp;
       case OperationAttemptResultType.Longpoll:
         return resp;
@@ -297,7 +428,7 @@ export async function runOperationWithErrorReporting<T1, T2>(
       logger.warn("operation processed resulted in error");
       logger.warn(`error was: ${j2s(e.errorDetail)}`);
       maybeError = e.errorDetail;
-      await storeOperationError(ws, opId, maybeError!);
+      await storePendingTaskError(ws, opId, maybeError!);
       return {
         type: OperationAttemptResultType.Error,
         errorDetail: e.errorDetail,
@@ -315,7 +446,7 @@ export async function runOperationWithErrorReporting<T1, T2>(
         },
         `unexpected exception (message: ${e.message})`,
       );
-      await storeOperationError(ws, opId, maybeError);
+      await storePendingTaskError(ws, opId, maybeError);
       return {
         type: OperationAttemptResultType.Error,
         errorDetail: maybeError,
@@ -327,7 +458,7 @@ export async function runOperationWithErrorReporting<T1, T2>(
         {},
         `unexpected exception (not even an error)`,
       );
-      await storeOperationError(ws, opId, maybeError);
+      await storePendingTaskError(ws, opId, maybeError);
       return {
         type: OperationAttemptResultType.Error,
         errorDetail: maybeError,
@@ -336,17 +467,6 @@ export async function runOperationWithErrorReporting<T1, T2>(
   }
 }
 
-export async function storeOperationFinished(
-  ws: InternalWalletState,
-  pendingTaskId: string,
-): Promise<void> {
-  await ws.db
-    .mktx((x) => [x.operationRetries])
-    .runReadWrite(async (tx) => {
-      await tx.operationRetries.delete(pendingTaskId);
-    });
-}
-
 export enum TombstoneTag {
   DeleteWithdrawalGroup = "delete-withdrawal-group",
   DeleteReserve = "delete-reserve",
@@ -361,15 +481,6 @@ export enum TombstoneTag {
   DeletePeerPushCredit = "delete-peer-push-credit",
 }
 
-/**
- * Create an event ID from the type and the primary key for the event.
- *
- * @deprecated use constructTombstone instead
- */
-export function makeTombstoneId(type: TombstoneTag, ...args: string[]): string {
-  return `tmb:${type}:${args.map((x) => encodeURIComponent(x)).join(":")}`;
-}
-
 export function getExchangeTosStatus(
   exchangeDetails: ExchangeDetailsRecord,
 ): ExchangeTosStatus {
@@ -432,7 +543,7 @@ export function runLongpollAsync(
   const asyncFn = async () => {
     if (ws.stopped) {
       logger.trace("not long-polling reserve, wallet already stopped");
-      await storeOperationPending(ws, retryTag);
+      await storePendingTaskPending(ws, retryTag);
       return;
     }
     const cts = CancellationToken.create();
@@ -446,13 +557,13 @@ export function runLongpollAsync(
       };
       res = await reqFn(cts.token);
     } catch (e) {
-      await storeOperationError(ws, retryTag, getErrorDetailFromException(e));
+      await storePendingTaskError(ws, retryTag, getErrorDetailFromException(e));
       return;
     } finally {
       delete ws.activeLongpoll[retryTag];
     }
     if (!res.ready) {
-      await storeOperationPending(ws, retryTag);
+      await storePendingTaskPending(ws, retryTag);
     }
     ws.workAvailable.trigger();
   };
@@ -464,7 +575,11 @@ export type ParsedTombstone =
       tag: TombstoneTag.DeleteWithdrawalGroup;
       withdrawalGroupId: string;
     }
-  | { tag: TombstoneTag.DeleteRefund; refundGroupId: string };
+  | { tag: TombstoneTag.DeleteRefund; refundGroupId: string }
+  | { tag: TombstoneTag.DeleteReserve; reservePub: string }
+  | { tag: TombstoneTag.DeleteRefreshGroup; refreshGroupId: string }
+  | { tag: TombstoneTag.DeleteTip; walletTipId: string }
+  | { tag: TombstoneTag.DeletePayment; proposalId: string };
 
 export function constructTombstone(p: ParsedTombstone): TombstoneIdStr {
   switch (p.tag) {
@@ -472,6 +587,16 @@ export function constructTombstone(p: ParsedTombstone): TombstoneIdStr {
       return `tmb:${p.tag}:${p.withdrawalGroupId}` as TombstoneIdStr;
     case TombstoneTag.DeleteRefund:
       return `tmb:${p.tag}:${p.refundGroupId}` as TombstoneIdStr;
+    case TombstoneTag.DeleteReserve:
+      return `tmb:${p.tag}:${p.reservePub}` as TombstoneIdStr;
+    case TombstoneTag.DeletePayment:
+      return `tmb:${p.tag}:${p.proposalId}` as TombstoneIdStr;
+    case TombstoneTag.DeleteRefreshGroup:
+      return `tmb:${p.tag}:${p.refreshGroupId}` as TombstoneIdStr;
+    case TombstoneTag.DeleteTip:
+      return `tmb:${p.tag}:${p.walletTipId}` as TombstoneIdStr;
+    default:
+      assertUnreachable(p);
   }
 }
 
@@ -487,3 +612,305 @@ export interface TransactionManager {
   resume(): Promise<void>;
   process(): Promise<OperationAttemptResult>;
 }
+
+export enum OperationAttemptResultType {
+  Finished = "finished",
+  Pending = "pending",
+  Error = "error",
+  Longpoll = "longpoll",
+}
+
+export type OperationAttemptResult<TSuccess = unknown, TPending = unknown> =
+  | OperationAttemptFinishedResult<TSuccess>
+  | OperationAttemptErrorResult
+  | OperationAttemptLongpollResult
+  | OperationAttemptPendingResult<TPending>;
+
+export namespace OperationAttemptResult {
+  export function finishedEmpty(): OperationAttemptResult<unknown, unknown> {
+    return {
+      type: OperationAttemptResultType.Finished,
+      result: undefined,
+    };
+  }
+  export function pendingEmpty(): OperationAttemptResult<unknown, unknown> {
+    return {
+      type: OperationAttemptResultType.Pending,
+      result: undefined,
+    };
+  }
+  export function longpoll(): OperationAttemptResult<unknown, unknown> {
+    return {
+      type: OperationAttemptResultType.Longpoll,
+    };
+  }
+}
+
+export interface OperationAttemptFinishedResult<T> {
+  type: OperationAttemptResultType.Finished;
+  result: T;
+}
+
+export interface OperationAttemptPendingResult<T> {
+  type: OperationAttemptResultType.Pending;
+  result: T;
+}
+
+export interface OperationAttemptErrorResult {
+  type: OperationAttemptResultType.Error;
+  errorDetail: TalerErrorDetail;
+}
+
+export interface OperationAttemptLongpollResult {
+  type: OperationAttemptResultType.Longpoll;
+}
+
+export interface RetryInfo {
+  firstTry: AbsoluteTime;
+  nextRetry: AbsoluteTime;
+  retryCounter: number;
+}
+
+export interface RetryPolicy {
+  readonly backoffDelta: Duration;
+  readonly backoffBase: number;
+  readonly maxTimeout: Duration;
+}
+
+const defaultRetryPolicy: RetryPolicy = {
+  backoffBase: 1.5,
+  backoffDelta: Duration.fromSpec({ seconds: 1 }),
+  maxTimeout: Duration.fromSpec({ minutes: 2 }),
+};
+
+function updateTimeout(
+  r: RetryInfo,
+  p: RetryPolicy = defaultRetryPolicy,
+): void {
+  const now = AbsoluteTime.now();
+  if (now.t_ms === "never") {
+    throw Error("assertion failed");
+  }
+  if (p.backoffDelta.d_ms === "forever") {
+    r.nextRetry = AbsoluteTime.never();
+    return;
+  }
+
+  const nextIncrement =
+    p.backoffDelta.d_ms * Math.pow(p.backoffBase, r.retryCounter);
+
+  const t =
+    now.t_ms +
+    (p.maxTimeout.d_ms === "forever"
+      ? nextIncrement
+      : Math.min(p.maxTimeout.d_ms, nextIncrement));
+  r.nextRetry = AbsoluteTime.fromMilliseconds(t);
+}
+
+export namespace RetryInfo {
+  export function getDuration(
+    r: RetryInfo | undefined,
+    p: RetryPolicy = defaultRetryPolicy,
+  ): Duration {
+    if (!r) {
+      // If we don't have any retry info, run immediately.
+      return { d_ms: 0 };
+    }
+    if (p.backoffDelta.d_ms === "forever") {
+      return { d_ms: "forever" };
+    }
+    const t = p.backoffDelta.d_ms * Math.pow(p.backoffBase, r.retryCounter);
+    return {
+      d_ms:
+        p.maxTimeout.d_ms === "forever" ? t : Math.min(p.maxTimeout.d_ms, t),
+    };
+  }
+
+  export function reset(p: RetryPolicy = defaultRetryPolicy): RetryInfo {
+    const now = AbsoluteTime.now();
+    const info = {
+      firstTry: now,
+      nextRetry: now,
+      retryCounter: 0,
+    };
+    updateTimeout(info, p);
+    return info;
+  }
+
+  export function increment(
+    r: RetryInfo | undefined,
+    p: RetryPolicy = defaultRetryPolicy,
+  ): RetryInfo {
+    if (!r) {
+      return reset(p);
+    }
+    const r2 = { ...r };
+    r2.retryCounter++;
+    updateTimeout(r2, p);
+    return r2;
+  }
+}
+
+/**
+ * Parsed representation of task identifiers.
+ */
+export type ParsedTaskIdentifier =
+  | {
+      tag: PendingTaskType.Withdraw;
+      withdrawalGroupId: string;
+    }
+  | { tag: PendingTaskType.ExchangeUpdate; exchangeBaseUrl: string }
+  | { tag: PendingTaskType.Backup; backupProviderBaseUrl: string }
+  | { tag: PendingTaskType.Deposit; depositGroupId: string }
+  | { tag: PendingTaskType.ExchangeCheckRefresh; exchangeBaseUrl: string }
+  | { tag: PendingTaskType.ExchangeUpdate; exchangeBaseUrl: string }
+  | { tag: PendingTaskType.PeerPullDebit; peerPullPaymentIncomingId: string }
+  | { tag: PendingTaskType.PeerPullCredit; pursePub: string }
+  | { tag: PendingTaskType.PeerPushCredit; peerPushPaymentIncomingId: string }
+  | { tag: PendingTaskType.PeerPushDebit; pursePub: string }
+  | { tag: PendingTaskType.Purchase; proposalId: string }
+  | { tag: PendingTaskType.Recoup; recoupGroupId: string }
+  | { tag: PendingTaskType.TipPickup; walletTipId: string }
+  | { tag: PendingTaskType.Refresh; refreshGroupId: string };
+
+export function parseTaskIdentifier(x: string): ParsedTaskIdentifier {
+  const task = x.split(":");
+
+  if (task.length < 2) {
+    throw Error("task id should have al least 2 parts separated by ':'");
+  }
+
+  const [type, ...rest] = task;
+  switch (type) {
+    case PendingTaskType.Backup:
+      return { tag: type, backupProviderBaseUrl: rest[0] };
+    case PendingTaskType.Deposit:
+      return { tag: type, depositGroupId: rest[0] };
+    case PendingTaskType.ExchangeCheckRefresh:
+      return { tag: type, exchangeBaseUrl: rest[0] };
+    case PendingTaskType.ExchangeUpdate:
+      return { tag: type, exchangeBaseUrl: rest[0] };
+    case PendingTaskType.PeerPullCredit:
+      return { tag: type, pursePub: rest[0] };
+    case PendingTaskType.PeerPullDebit:
+      return { tag: type, peerPullPaymentIncomingId: rest[0] };
+    case PendingTaskType.PeerPushCredit:
+      return { tag: type, peerPushPaymentIncomingId: rest[0] };
+    case PendingTaskType.PeerPushDebit:
+      return { tag: type, pursePub: rest[0] };
+    case PendingTaskType.Purchase:
+      return { tag: type, proposalId: rest[0] };
+    case PendingTaskType.Recoup:
+      return { tag: type, recoupGroupId: rest[0] };
+    case PendingTaskType.Refresh:
+      return { tag: type, refreshGroupId: rest[0] };
+    case PendingTaskType.TipPickup:
+      return { tag: type, walletTipId: rest[0] };
+    case PendingTaskType.Withdraw:
+      return { tag: type, withdrawalGroupId: rest[0] };
+    default:
+      throw Error("invalid task identifier");
+  }
+}
+
+export function constructTaskIdentifier(p: ParsedTaskIdentifier): TaskId {
+  switch (p.tag) {
+    case PendingTaskType.Backup:
+      return `${p.tag}:${p.backupProviderBaseUrl}` as TaskId;
+    case PendingTaskType.Deposit:
+      return `${p.tag}:${p.depositGroupId}` as TaskId;
+    case PendingTaskType.ExchangeCheckRefresh:
+      return `${p.tag}:${p.exchangeBaseUrl}` as TaskId;
+    case PendingTaskType.ExchangeUpdate:
+      return `${p.tag}:${p.exchangeBaseUrl}` as TaskId;
+    case PendingTaskType.PeerPullDebit:
+      return `${p.tag}:${p.peerPullPaymentIncomingId}` as TaskId;
+    case PendingTaskType.PeerPushCredit:
+      return `${p.tag}:${p.peerPushPaymentIncomingId}` as TaskId;
+    case PendingTaskType.PeerPullCredit:
+      return `${p.tag}:${p.pursePub}` as TaskId;
+    case PendingTaskType.PeerPushDebit:
+      return `${p.tag}:${p.pursePub}` as TaskId;
+    case PendingTaskType.Purchase:
+      return `${p.tag}:${p.proposalId}` as TaskId;
+    case PendingTaskType.Recoup:
+      return `${p.tag}:${p.recoupGroupId}` as TaskId;
+    case PendingTaskType.Refresh:
+      return `${p.tag}:${p.refreshGroupId}` as TaskId;
+    case PendingTaskType.TipPickup:
+      return `${p.tag}:${p.walletTipId}` as TaskId;
+    case PendingTaskType.Withdraw:
+      return `${p.tag}:${p.withdrawalGroupId}` as TaskId;
+    default:
+      assertUnreachable(p);
+  }
+}
+
+export namespace TaskIdentifiers {
+  export function forWithdrawal(wg: WithdrawalGroupRecord): TaskId {
+    return `${PendingTaskType.Withdraw}:${wg.withdrawalGroupId}` as TaskId;
+  }
+  export function forExchangeUpdate(exch: ExchangeRecord): TaskId {
+    return `${PendingTaskType.ExchangeUpdate}:${exch.baseUrl}` as TaskId;
+  }
+  export function forExchangeUpdateFromUrl(exchBaseUrl: string): TaskId {
+    return `${PendingTaskType.ExchangeUpdate}:${exchBaseUrl}` as TaskId;
+  }
+  export function forExchangeCheckRefresh(exch: ExchangeRecord): TaskId {
+    return `${PendingTaskType.ExchangeCheckRefresh}:${exch.baseUrl}` as TaskId;
+  }
+  export function forTipPickup(tipRecord: TipRecord): TaskId {
+    return `${PendingTaskType.TipPickup}:${tipRecord.walletTipId}` as TaskId;
+  }
+  export function forRefresh(refreshGroupRecord: RefreshGroupRecord): TaskId {
+    return `${PendingTaskType.Refresh}:${refreshGroupRecord.refreshGroupId}` as TaskId;
+  }
+  export function forPay(purchaseRecord: PurchaseRecord): TaskId {
+    return `${PendingTaskType.Purchase}:${purchaseRecord.proposalId}` as TaskId;
+  }
+  export function forRecoup(recoupRecord: RecoupGroupRecord): TaskId {
+    return `${PendingTaskType.Recoup}:${recoupRecord.recoupGroupId}` as TaskId;
+  }
+  export function forDeposit(depositRecord: DepositGroupRecord): TaskId {
+    return `${PendingTaskType.Deposit}:${depositRecord.depositGroupId}` as TaskId;
+  }
+  export function forBackup(backupRecord: BackupProviderRecord): TaskId {
+    return `${PendingTaskType.Backup}:${backupRecord.baseUrl}` as TaskId;
+  }
+  export function forPeerPushPaymentInitiation(
+    ppi: PeerPushPaymentInitiationRecord,
+  ): TaskId {
+    return `${PendingTaskType.PeerPushDebit}:${ppi.pursePub}` as TaskId;
+  }
+  export function forPeerPullPaymentInitiation(
+    ppi: PeerPullPaymentInitiationRecord,
+  ): TaskId {
+    return `${PendingTaskType.PeerPullCredit}:${ppi.pursePub}` as TaskId;
+  }
+  export function forPeerPullPaymentDebit(
+    ppi: PeerPullPaymentIncomingRecord,
+  ): TaskId {
+    return `${PendingTaskType.PeerPullDebit}:${ppi.peerPullPaymentIncomingId}` as TaskId;
+  }
+  export function forPeerPushCredit(
+    ppi: PeerPushPaymentIncomingRecord,
+  ): TaskId {
+    return `${PendingTaskType.PeerPushCredit}:${ppi.peerPushPaymentIncomingId}` as TaskId;
+  }
+}
+
+/**
+ * Run an operation handler, expect a success result and extract the success value.
+ */
+export async function unwrapOperationHandlerResultOrThrow<T>(
+  res: OperationAttemptResult<T>,
+): Promise<T> {
+  switch (res.type) {
+    case OperationAttemptResultType.Finished:
+      return res.result;
+    case OperationAttemptResultType.Error:
+      throw TalerError.fromUncheckedDetail(res.errorDetail);
+    default:
+      throw Error(`unexpected operation result (${res.type})`);
+  }
+}