From 4e76edf129229823281c2a662392249c32a1c7a2 Mon Sep 17 00:00:00 2001 From: Florian Dold Date: Fri, 6 Mar 2020 19:39:55 +0530 Subject: [PATCH] include (pending) wallet balance in pending ops response --- src/operations/balance.ts | 148 ++++++++++++++++++++------------------ src/operations/pending.ts | 45 +++++++----- src/types/pending.ts | 14 +++- src/wallet.ts | 30 ++++---- 4 files changed, 137 insertions(+), 100 deletions(-) diff --git a/src/operations/balance.ts b/src/operations/balance.ts index d12fbaf7c..03d1b2a9f 100644 --- a/src/operations/balance.ts +++ b/src/operations/balance.ts @@ -18,7 +18,7 @@ * Imports. */ import { WalletBalance, WalletBalanceEntry } from "../types/walletTypes"; -import { Database } from "../util/query"; +import { Database, TransactionHandle } from "../util/query"; import { InternalWalletState } from "./state"; import { Stores, TipRecord, CoinStatus } from "../types/dbTypes"; import * as Amounts from "../util/amounts"; @@ -28,13 +28,14 @@ import { Logger } from "../util/logging"; const logger = new Logger("withdraw.ts"); /** - * Get detailed balance information, sliced by exchange and by currency. + * Get balance information. */ -export async function getBalances( +export async function getBalancesInsideTransaction( ws: InternalWalletState, + tx: TransactionHandle, ): Promise { - logger.trace("starting to compute balance"); - /** + + /** * Add amount to a balance field, both for * the slicing by exchange and currency. */ @@ -73,76 +74,85 @@ export async function getBalances( byExchange: {}, }; - await ws.db.runWithReadTransaction( - [Stores.coins, Stores.refreshGroups, Stores.reserves, Stores.purchases, Stores.withdrawalSession], - async tx => { - await tx.iter(Stores.coins).forEach(c => { - if (c.suspended) { - return; - } - if (c.status === CoinStatus.Fresh) { - addTo(balanceStore, "available", c.currentAmount, c.exchangeBaseUrl); - } - }); - await tx.iter(Stores.refreshGroups).forEach(r => { - // Don't count finished refreshes, since the refresh already resulted - // in coins being added to the wallet. - if (r.timestampFinished) { - return; - } - for (let i = 0; i < r.oldCoinPubs.length; i++) { - const session = r.refreshSessionPerCoin[i]; - if (session) { - addTo( - balanceStore, - "pendingIncoming", - session.amountRefreshOutput, - session.exchangeBaseUrl, - ); - addTo( - balanceStore, - "pendingIncomingRefresh", - session.amountRefreshOutput, - session.exchangeBaseUrl, - ); - } - } - }); - - await tx.iter(Stores.withdrawalSession).forEach(wds => { - let w = wds.totalCoinValue; - for (let i = 0; i < wds.planchets.length; i++) { - if (wds.withdrawn[i]) { - const p = wds.planchets[i]; - if (p) { - w = Amounts.sub(w, p.coinValue).amount; - } - } - } + await tx.iter(Stores.coins).forEach(c => { + if (c.suspended) { + return; + } + if (c.status === CoinStatus.Fresh) { + addTo(balanceStore, "available", c.currentAmount, c.exchangeBaseUrl); + } + }); + await tx.iter(Stores.refreshGroups).forEach(r => { + // Don't count finished refreshes, since the refresh already resulted + // in coins being added to the wallet. + if (r.timestampFinished) { + return; + } + for (let i = 0; i < r.oldCoinPubs.length; i++) { + const session = r.refreshSessionPerCoin[i]; + if (session) { addTo( balanceStore, "pendingIncoming", - w, - wds.exchangeBaseUrl, + session.amountRefreshOutput, + session.exchangeBaseUrl, ); - }); + addTo( + balanceStore, + "pendingIncomingRefresh", + session.amountRefreshOutput, + session.exchangeBaseUrl, + ); + } + } + }); - await tx.iter(Stores.purchases).forEach(t => { - if (t.timestampFirstSuccessfulPay) { - return; + await tx.iter(Stores.withdrawalSession).forEach(wds => { + let w = wds.totalCoinValue; + for (let i = 0; i < wds.planchets.length; i++) { + if (wds.withdrawn[i]) { + const p = wds.planchets[i]; + if (p) { + w = Amounts.sub(w, p.coinValue).amount; } - for (const c of t.payReq.coins) { - addTo( - balanceStore, - "pendingPayment", - Amounts.parseOrThrow(c.contribution), - c.exchange_url, - ); - } - }); - }, - ); + } + } + addTo(balanceStore, "pendingIncoming", w, wds.exchangeBaseUrl); + }); + + await tx.iter(Stores.purchases).forEach(t => { + if (t.timestampFirstSuccessfulPay) { + return; + } + for (const c of t.payReq.coins) { + addTo( + balanceStore, + "pendingPayment", + Amounts.parseOrThrow(c.contribution), + c.exchange_url, + ); + } + }); - logger.trace("computed balances:", balanceStore); return balanceStore; } + +/** + * Get detailed balance information, sliced by exchange and by currency. + */ +export async function getBalances( + ws: InternalWalletState, +): Promise { + logger.trace("starting to compute balance"); + + return await ws.db.runWithReadTransaction([ + Stores.coins, + Stores.refreshGroups, + Stores.reserves, + Stores.purchases, + Stores.withdrawalSession, + ], + async tx => { + return getBalancesInsideTransaction(ws, tx); + }); +} diff --git a/src/operations/pending.ts b/src/operations/pending.ts index 9e2ff6b32..fce9a3bfb 100644 --- a/src/operations/pending.ts +++ b/src/operations/pending.ts @@ -28,9 +28,16 @@ import { PendingOperationType, ExchangeUpdateOperationStage, } from "../types/pending"; -import { Duration, getTimestampNow, Timestamp, getDurationRemaining, durationMin } from "../util/time"; +import { + Duration, + getTimestampNow, + Timestamp, + getDurationRemaining, + durationMin, +} from "../util/time"; import { TransactionHandle } from "../util/query"; import { InternalWalletState } from "./state"; +import { getBalances, getBalancesInsideTransaction } from "./balance"; function updateRetryDelay( oldDelay: Duration, @@ -38,7 +45,7 @@ function updateRetryDelay( retryTimestamp: Timestamp, ): Duration { const remaining = getDurationRemaining(retryTimestamp, now); - const nextDelay = durationMin(oldDelay, remaining); + const nextDelay = durationMin(oldDelay, remaining); return nextDelay; } @@ -110,14 +117,14 @@ async function gatherExchangePending( }); break; case ExchangeUpdateStatus.FinalizeUpdate: - resp.pendingOperations.push({ - type: PendingOperationType.ExchangeUpdate, - givesLifeness: false, - stage: ExchangeUpdateOperationStage.FinalizeUpdate, - exchangeBaseUrl: e.baseUrl, - lastError: e.lastError, - reason: e.updateReason || "unknown", - }); + resp.pendingOperations.push({ + type: PendingOperationType.ExchangeUpdate, + givesLifeness: false, + stage: ExchangeUpdateOperationStage.FinalizeUpdate, + exchangeBaseUrl: e.baseUrl, + lastError: e.lastError, + reason: e.updateReason || "unknown", + }); break; default: resp.pendingOperations.push({ @@ -400,15 +407,10 @@ async function gatherPurchasePending( export async function getPendingOperations( ws: InternalWalletState, - onlyDue: boolean = false, + { onlyDue = false } = {}, ): Promise { - const resp: PendingOperationsResponse = { - nextRetryDelay: { d_ms: Number.MAX_SAFE_INTEGER }, - onlyDue: onlyDue, - pendingOperations: [], - }; const now = getTimestampNow(); - await ws.db.runWithReadTransaction( + return await ws.db.runWithReadTransaction( [ Stores.exchanges, Stores.reserves, @@ -420,6 +422,13 @@ export async function getPendingOperations( Stores.purchases, ], async tx => { + const walletBalance = await getBalancesInsideTransaction(ws, tx); + const resp: PendingOperationsResponse = { + nextRetryDelay: { d_ms: Number.MAX_SAFE_INTEGER }, + onlyDue: onlyDue, + walletBalance, + pendingOperations: [], + }; await gatherExchangePending(tx, now, resp, onlyDue); await gatherReservePending(tx, now, resp, onlyDue); await gatherRefreshPending(tx, now, resp, onlyDue); @@ -427,7 +436,7 @@ export async function getPendingOperations( await gatherProposalPending(tx, now, resp, onlyDue); await gatherTipPending(tx, now, resp, onlyDue); await gatherPurchasePending(tx, now, resp, onlyDue); + return resp; }, ); - return resp; } diff --git a/src/types/pending.ts b/src/types/pending.ts index 3c169c2c4..b86c7797b 100644 --- a/src/types/pending.ts +++ b/src/types/pending.ts @@ -21,7 +21,7 @@ /** * Imports. */ -import { OperationError } from "./walletTypes"; +import { OperationError, WalletBalance } from "./walletTypes"; import { WithdrawalSource, RetryInfo, ReserveRecordStatus } from "./dbTypes"; import { Timestamp, Duration } from "../util/time"; @@ -231,7 +231,19 @@ export interface PendingOperationInfoCommon { * Response returned from the pending operations API. */ export interface PendingOperationsResponse { + /** + * List of pending operations. + */ pendingOperations: PendingOperationInfo[]; + + /** + * Current wallet balance, including pending balances. + */ + walletBalance: WalletBalance; + + /** + * When is the next pending operation due to be re-tried? + */ nextRetryDelay: Duration; /** diff --git a/src/wallet.ts b/src/wallet.ts index 32a92cee0..12bc2ccbb 100644 --- a/src/wallet.ts +++ b/src/wallet.ts @@ -82,7 +82,10 @@ import { getExchangePaytoUri, acceptExchangeTermsOfService, } from "./operations/exchanges"; -import { processReserve, createTalerWithdrawReserve } from "./operations/reserves"; +import { + processReserve, + createTalerWithdrawReserve, +} from "./operations/reserves"; import { InternalWalletState } from "./operations/state"; import { createReserve, confirmReserve } from "./operations/reserves"; @@ -111,7 +114,6 @@ import { } from "./operations/refund"; import { durationMin, Duration } from "./util/time"; - const builtinCurrencies: CurrencyRecord[] = [ { auditors: [ @@ -225,7 +227,7 @@ export class Wallet { */ public async runPending(forceNow: boolean = false): Promise { const onlyDue = !forceNow; - const pendingOpsResponse = await this.getPendingOperations(onlyDue); + const pendingOpsResponse = await this.getPendingOperations({ onlyDue }); for (const p of pendingOpsResponse.pendingOperations) { try { await this.processOnePendingOperation(p, forceNow); @@ -260,7 +262,7 @@ export class Wallet { await p; } - /** + /** * Run the wallet until there are no more pending operations that give * liveness left. The wallet will be in a stopped state when this function * returns without resolving to an exception. @@ -304,10 +306,10 @@ export class Wallet { private async runRetryLoopImpl(): Promise { while (!this.stopped) { console.log("running wallet retry loop iteration"); - let pending = await this.getPendingOperations(true); + let pending = await this.getPendingOperations({ onlyDue: true }); console.log("pending ops", JSON.stringify(pending, undefined, 2)); if (pending.pendingOperations.length === 0) { - const allPending = await this.getPendingOperations(false); + const allPending = await this.getPendingOperations({ onlyDue: false }); let numPending = 0; let numGivingLiveness = 0; for (const p of allPending.pendingOperations) { @@ -324,7 +326,7 @@ export class Wallet { // Wait for 5 seconds dt = { d_ms: 5000 }; } else { - dt = durationMin({ d_ms: 5000}, allPending.nextRetryDelay); + dt = durationMin({ d_ms: 5000 }, allPending.nextRetryDelay); } const timeout = this.timerGroup.resolveAfter(dt); this.ws.notify({ @@ -524,11 +526,11 @@ export class Wallet { return getHistory(this.ws, historyQuery); } - async getPendingOperations( - onlyDue: boolean = false, - ): Promise { + async getPendingOperations({ onlyDue = false } = {}): Promise< + PendingOperationsResponse + > { return this.ws.memoGetPending.memo(() => - getPendingOperations(this.ws, onlyDue), + getPendingOperations(this.ws, { onlyDue }), ); } @@ -702,7 +704,11 @@ export class Wallet { selectedExchange: string, ): Promise { try { - return createTalerWithdrawReserve(this.ws, talerWithdrawUri, selectedExchange); + return createTalerWithdrawReserve( + this.ws, + talerWithdrawUri, + selectedExchange, + ); } finally { this.latch.trigger(); }