diff options
Diffstat (limited to 'packages/taler-wallet-core/src/refresh.ts')
| -rw-r--r-- | packages/taler-wallet-core/src/refresh.ts | 661 |
1 files changed, 470 insertions, 191 deletions
diff --git a/packages/taler-wallet-core/src/refresh.ts b/packages/taler-wallet-core/src/refresh.ts index e6013938d..fb2cdba93 100644 --- a/packages/taler-wallet-core/src/refresh.ts +++ b/packages/taler-wallet-core/src/refresh.ts @@ -1,6 +1,6 @@ /* This file is part of GNU Taler - (C) 2019 GNUnet e.V. + (C) 2019-2024 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 @@ -14,6 +14,14 @@ GNU Taler; see the file COPYING. If not, see <http://www.gnu.org/licenses/> */ +/** + * @fileoverview + * Implementation of the refresh transaction. + */ + +/** + * Imports. + */ import { AgeCommitment, AgeRestriction, @@ -23,6 +31,7 @@ import { assertUnreachable, AsyncFlag, checkDbInvariant, + codecForCoinHistoryResponse, codecForExchangeMeltResponse, codecForExchangeRevealResponse, CoinPublicKeyString, @@ -61,7 +70,6 @@ import { import { readSuccessResponseJsonOrThrow, readTalerErrorResponse, - readUnexpectedResponseDetails, throwUnexpectedRequestError, } from "@gnu-taler/taler-util/http"; import { @@ -73,6 +81,8 @@ import { TaskRunResultType, TombstoneTag, TransactionContext, + TransitionResult, + TransitionResultType, } from "./common.js"; import { TalerCryptoInterface } from "./crypto/cryptoImplementation.js"; import { @@ -93,11 +103,13 @@ import { timestampPreciseToDb, WalletDbReadOnlyTransaction, WalletDbReadWriteTransaction, + WalletDbStoresArr, } from "./db.js"; import { selectWithdrawalDenominations } from "./denomSelection.js"; import { constructTransactionIdentifier, notifyTransition, + TransitionInfo, } from "./transactions.js"; import { EXCHANGE_COINS_LOCK, @@ -108,6 +120,23 @@ import { getCandidateWithdrawalDenomsTx } from "./withdraw.js"; const logger = new Logger("refresh.ts"); +/** + * Update the materialized refresh transaction based + * on the refresh group record. + */ +async function updateRefreshTransaction( + ctx: RefreshTransactionContext, + tx: WalletDbReadWriteTransaction< + [ + "refreshGroups", + "transactions", + "operationRetries", + "exchanges", + "exchangeDetails", + ] + >, +): Promise<void> {} + export class RefreshTransactionContext implements TransactionContext { readonly transactionId: TransactionIdStr; readonly taskId: TaskIdStr; @@ -126,56 +155,112 @@ export class RefreshTransactionContext implements TransactionContext { }); } - async deleteTransaction(): Promise<void> { - const refreshGroupId = this.refreshGroupId; - await this.wex.db.runReadWriteTx( - ["refreshGroups", "tombstones"], + /** + * Transition a withdrawal transaction. + * Extra object stores may be accessed during the transition. + */ + async transition<StoreNameArray extends WalletDbStoresArr = []>( + opts: { extraStores?: StoreNameArray; transactionLabel?: string }, + f: ( + rec: RefreshGroupRecord | undefined, + tx: WalletDbReadWriteTransaction< + [ + "refreshGroups", + "transactions", + "operationRetries", + "exchanges", + "exchangeDetails", + ...StoreNameArray, + ] + >, + ) => Promise<TransitionResult<RefreshGroupRecord>>, + ): Promise<TransitionInfo | undefined> { + const baseStores = [ + "refreshGroups" as const, + "transactions" as const, + "operationRetries" as const, + "exchanges" as const, + "exchangeDetails" as const, + ]; + let stores = opts.extraStores + ? [...baseStores, ...opts.extraStores] + : baseStores; + const transitionInfo = await this.wex.db.runReadWriteTx( + stores, async (tx) => { - const rg = await tx.refreshGroups.get(refreshGroupId); - if (rg) { - await tx.refreshGroups.delete(refreshGroupId); - await tx.tombstones.put({ - id: TombstoneTag.DeleteRefreshGroup + ":" + refreshGroupId, - }); + const wgRec = await tx.refreshGroups.get(this.refreshGroupId); + let oldTxState: TransactionState; + if (wgRec) { + oldTxState = computeRefreshTransactionState(wgRec); + } else { + oldTxState = { + major: TransactionMajorState.None, + }; + } + const res = await f(wgRec, tx); + switch (res.type) { + case TransitionResultType.Transition: { + await tx.refreshGroups.put(res.rec); + await updateRefreshTransaction(this, tx); + const newTxState = computeRefreshTransactionState(res.rec); + return { + oldTxState, + newTxState, + }; + } + case TransitionResultType.Delete: + await tx.refreshGroups.delete(this.refreshGroupId); + await updateRefreshTransaction(this, tx); + return { + oldTxState, + newTxState: { + major: TransactionMajorState.None, + }, + }; + default: + return undefined; } }, ); + notifyTransition(this.wex, this.transactionId, transitionInfo); + return transitionInfo; } - async suspendTransaction(): Promise<void> { - const { wex, refreshGroupId, transactionId } = this; - let transitionInfo = await wex.db.runReadWriteTx( - ["refreshGroups"], - async (tx) => { - const dg = await tx.refreshGroups.get(refreshGroupId); - if (!dg) { - logger.warn( - `can't suspend refresh group, refreshGroupId=${refreshGroupId} not found`, - ); - return undefined; - } - const oldState = computeRefreshTransactionState(dg); - switch (dg.operationStatus) { - case RefreshOperationStatus.Finished: - case RefreshOperationStatus.Suspended: - case RefreshOperationStatus.Failed: - return undefined; - case RefreshOperationStatus.Pending: { - dg.operationStatus = RefreshOperationStatus.Suspended; - await tx.refreshGroups.put(dg); - break; - } - default: - assertUnreachable(dg.operationStatus); + async deleteTransaction(): Promise<void> { + await this.transition( + { + extraStores: ["tombstones"], + }, + async (rec, tx) => { + if (!rec) { + return TransitionResult.stay(); } - return { - oldTxState: oldState, - newTxState: computeRefreshTransactionState(dg), - }; + await tx.tombstones.put({ + id: TombstoneTag.DeleteRefreshGroup + ":" + this.refreshGroupId, + }); + return TransitionResult.delete(); }, ); - wex.taskScheduler.stopShepherdTask(this.taskId); - notifyTransition(wex, transactionId, transitionInfo); + } + + async suspendTransaction(): Promise<void> { + await this.transition({}, async (rec, tx) => { + if (!rec) { + return TransitionResult.stay(); + } + switch (rec.operationStatus) { + case RefreshOperationStatus.Finished: + case RefreshOperationStatus.Suspended: + case RefreshOperationStatus.Failed: + return TransitionResult.stay(); + case RefreshOperationStatus.Pending: { + rec.operationStatus = RefreshOperationStatus.Suspended; + return TransitionResult.transition(rec); + } + default: + assertUnreachable(rec.operationStatus); + } + }); } async abortTransaction(): Promise<void> { @@ -184,78 +269,43 @@ export class RefreshTransactionContext implements TransactionContext { } async resumeTransaction(): Promise<void> { - const { wex, refreshGroupId, transactionId } = this; - const transitionInfo = await wex.db.runReadWriteTx( - ["refreshGroups"], - async (tx) => { - const dg = await tx.refreshGroups.get(refreshGroupId); - if (!dg) { - logger.warn( - `can't resume refresh group, refreshGroupId=${refreshGroupId} not found`, - ); - return; - } - const oldState = computeRefreshTransactionState(dg); - switch (dg.operationStatus) { - case RefreshOperationStatus.Finished: - return; - case RefreshOperationStatus.Pending: { - return; - } - case RefreshOperationStatus.Suspended: - dg.operationStatus = RefreshOperationStatus.Pending; - await tx.refreshGroups.put(dg); - return { - oldTxState: oldState, - newTxState: computeRefreshTransactionState(dg), - }; + await this.transition({}, async (rec, tx) => { + if (!rec) { + return TransitionResult.stay(); + } + switch (rec.operationStatus) { + case RefreshOperationStatus.Finished: + case RefreshOperationStatus.Failed: + case RefreshOperationStatus.Pending: + return TransitionResult.stay(); + case RefreshOperationStatus.Suspended: { + rec.operationStatus = RefreshOperationStatus.Pending; + return TransitionResult.transition(rec); } - return undefined; - }, - ); - notifyTransition(wex, transactionId, transitionInfo); - wex.taskScheduler.startShepherdTask(this.taskId); + default: + assertUnreachable(rec.operationStatus); + } + }); } async failTransaction(): Promise<void> { - const { wex, refreshGroupId, transactionId } = this; - const transitionInfo = await wex.db.runReadWriteTx( - ["refreshGroups"], - async (tx) => { - const dg = await tx.refreshGroups.get(refreshGroupId); - if (!dg) { - logger.warn( - `can't resume refresh group, refreshGroupId=${refreshGroupId} not found`, - ); - return; - } - const oldState = computeRefreshTransactionState(dg); - let newStatus: RefreshOperationStatus | undefined; - switch (dg.operationStatus) { - case RefreshOperationStatus.Finished: - break; - case RefreshOperationStatus.Pending: - case RefreshOperationStatus.Suspended: - newStatus = RefreshOperationStatus.Failed; - break; - case RefreshOperationStatus.Failed: - break; - default: - assertUnreachable(dg.operationStatus); - } - if (newStatus) { - dg.operationStatus = newStatus; - await tx.refreshGroups.put(dg); + await this.transition({}, async (rec, tx) => { + if (!rec) { + return TransitionResult.stay(); + } + switch (rec.operationStatus) { + case RefreshOperationStatus.Finished: + case RefreshOperationStatus.Failed: + return TransitionResult.stay(); + case RefreshOperationStatus.Pending: + case RefreshOperationStatus.Suspended: { + rec.operationStatus = RefreshOperationStatus.Failed; + return TransitionResult.transition(rec); } - return { - oldTxState: oldState, - newTxState: computeRefreshTransactionState(dg), - }; - }, - ); - wex.taskScheduler.stopShepherdTask(this.taskId); - notifyTransition(wex, transactionId, transitionInfo); - wex.taskScheduler.startShepherdTask(this.taskId); + default: + assertUnreachable(rec.operationStatus); + } + }); } } @@ -301,7 +351,7 @@ export function getTotalRefreshCost( return totalCost; } -export async function getCoinAvailabilityForDenom( +async function getCoinAvailabilityForDenom( wex: WalletExecutionContext, tx: WalletDbReadWriteTransaction< ["coins", "coinAvailability", "denominations"] @@ -392,6 +442,7 @@ async function initRefreshSession( )} too small`, ); refreshGroup.statusPerCoin[coinIndex] = RefreshCoinStatus.Finished; + return; } for (let i = 0; i < newCoinDenoms.selectedDenoms.length; i++) { @@ -427,6 +478,45 @@ async function initRefreshSession( await tx.refreshSessions.put(newSession); } +/** + * Uninitialize a refresh session. + * + * Adjust the coin availability of involved coins. + */ +async function destroyRefreshSession( + wex: WalletExecutionContext, + tx: WalletDbReadWriteTransaction< + ["denominations", "coinAvailability", "coins"] + >, + refreshGroup: RefreshGroupRecord, + refreshSession: RefreshSessionRecord, +): Promise<void> { + for (let i = 0; i < refreshSession.newDenoms.length; i++) { + const oldCoin = await tx.coins.get( + refreshGroup.oldCoinPubs[refreshSession.coinIndex], + ); + if (!oldCoin) { + continue; + } + const dph = refreshSession.newDenoms[i].denomPubHash; + const denom = await getDenomInfo(wex, tx, oldCoin.exchangeBaseUrl, dph); + if (!denom) { + logger.error(`denom ${dph} not in DB`); + continue; + } + const car = await getCoinAvailabilityForDenom( + wex, + tx, + denom, + oldCoin.maxAge, + ); + checkDbInvariant(car.pendingRefreshOutputCount != null); + car.pendingRefreshOutputCount = + car.pendingRefreshOutputCount - refreshSession.newDenoms[i].count; + await tx.coinAvailability.put(car); + } +} + function getRefreshRequestTimeout(rg: RefreshGroupRecord): Duration { return Duration.fromSpec({ seconds: 5, @@ -447,7 +537,6 @@ async function refreshMelt( coinIndex: number, ): Promise<void> { const ctx = new RefreshTransactionContext(wex, refreshGroupId); - const transactionId = ctx.transactionId; const d = await wex.db.runReadWriteTx( ["refreshGroups", "refreshSessions", "coins", "denominations"], async (tx) => { @@ -567,80 +656,34 @@ async function refreshMelt( }, ); - if (resp.status === HttpStatusCode.NotFound) { - const errDetails = await readUnexpectedResponseDetails(resp); - await wex.db.runReadWriteTx( - ["refreshGroups", "refreshSessions", "coins", "coinAvailability"], - async (tx) => { - const rg = await tx.refreshGroups.get(refreshGroupId); - if (!rg) { - return; - } - if (rg.timestampFinished) { - return; - } - if (rg.statusPerCoin[coinIndex] !== RefreshCoinStatus.Pending) { - return; - } - rg.statusPerCoin[coinIndex] = RefreshCoinStatus.Failed; - const refreshSession = await tx.refreshSessions.get([ - refreshGroupId, - coinIndex, - ]); - if (!refreshSession) { - throw Error( - "db invariant failed: missing refresh session in database", - ); - } - refreshSession.lastError = errDetails; - await tx.refreshGroups.put(rg); - await tx.refreshSessions.put(refreshSession); - }, - ); - return; - } - - if (resp.status === HttpStatusCode.Gone) { - const errDetail = await readTalerErrorResponse(resp); - - // FIXME(#7935): Remove coin from refresh group, but allow the whole group to finish. - - throwUnexpectedRequestError(resp, errDetail); - } - - if (resp.status === HttpStatusCode.Conflict) { - // Just log for better diagnostics here, error status - // will be handled later. - logger.error( - `melt request for ${Amounts.stringify( - derived.meltValueWithFee, - )} failed in refresh group ${refreshGroupId} due to conflict`, - ); - - const historySig = await wex.cryptoApi.signCoinHistoryRequest({ - coinPriv: oldCoin.coinPriv, - coinPub: oldCoin.coinPub, - startOffset: 0, - }); - - const historyUrl = new URL( - `coins/${oldCoin.coinPub}/history`, - oldCoin.exchangeBaseUrl, - ); - - const historyResp = await wex.http.fetch(historyUrl.href, { - method: "GET", - headers: { - "Taler-Coin-History-Signature": historySig.sig, - }, - cancellationToken: wex.cancellationToken, - }); - - const historyJson = await historyResp.json(); - logger.info(`coin history: ${j2s(historyJson)}`); - - // FIXME(#7935): Before failing and re-trying, analyse response and adjust amount. - // If response seems wrong, report to auditor (in the future!). + switch (resp.status) { + case HttpStatusCode.NotFound: { + const errDetail = await readTalerErrorResponse(resp); + await handleRefreshMeltNotFound(ctx, coinIndex, errDetail); + return; + } + case HttpStatusCode.Gone: { + const errDetail = await readTalerErrorResponse(resp); + await handleRefreshMeltGone(ctx, coinIndex, errDetail); + return; + } + case HttpStatusCode.Conflict: { + const errDetail = await readTalerErrorResponse(resp); + await handleRefreshMeltConflict( + ctx, + coinIndex, + errDetail, + derived, + oldCoin, + ); + return; + } + case HttpStatusCode.Ok: + break; + default: { + const errDetail = await readTalerErrorResponse(resp); + throwUnexpectedRequestError(resp, errDetail); + } } const meltResponse = await readSuccessResponseJsonOrThrow( @@ -675,6 +718,186 @@ async function refreshMelt( ); } +async function handleRefreshMeltGone( + ctx: RefreshTransactionContext, + coinIndex: number, + errDetails: TalerErrorDetail, +): Promise<void> { + // const expiredMsg = codecForDenominationExpiredMessage().decode(errDetails); + + // FIXME: Validate signature. + + await ctx.wex.db.runReadWriteTx( + [ + "refreshGroups", + "refreshSessions", + "coins", + "denominations", + "coinAvailability", + ], + async (tx) => { + const rg = await tx.refreshGroups.get(ctx.refreshGroupId); + if (!rg) { + return; + } + if (rg.timestampFinished) { + return; + } + if (rg.statusPerCoin[coinIndex] !== RefreshCoinStatus.Pending) { + return; + } + rg.statusPerCoin[coinIndex] = RefreshCoinStatus.Failed; + const refreshSession = await tx.refreshSessions.get([ + ctx.refreshGroupId, + coinIndex, + ]); + if (!refreshSession) { + throw Error("db invariant failed: missing refresh session in database"); + } + refreshSession.lastError = errDetails; + await destroyRefreshSession(ctx.wex, tx, rg, refreshSession); + await tx.refreshGroups.put(rg); + await tx.refreshSessions.put(refreshSession); + }, + ); +} + +async function handleRefreshMeltConflict( + ctx: RefreshTransactionContext, + coinIndex: number, + errDetails: TalerErrorDetail, + derived: DerivedRefreshSession, + oldCoin: CoinRecord, +): Promise<void> { + // Just log for better diagnostics here, error status + // will be handled later. + logger.error( + `melt request for ${Amounts.stringify( + derived.meltValueWithFee, + )} failed in refresh group ${ctx.refreshGroupId} due to conflict`, + ); + + const historySig = await ctx.wex.cryptoApi.signCoinHistoryRequest({ + coinPriv: oldCoin.coinPriv, + coinPub: oldCoin.coinPub, + startOffset: 0, + }); + + const historyUrl = new URL( + `coins/${oldCoin.coinPub}/history`, + oldCoin.exchangeBaseUrl, + ); + + const historyResp = await ctx.wex.http.fetch(historyUrl.href, { + method: "GET", + headers: { + "Taler-Coin-History-Signature": historySig.sig, + }, + cancellationToken: ctx.wex.cancellationToken, + }); + + const historyJson = await readSuccessResponseJsonOrThrow( + historyResp, + codecForCoinHistoryResponse(), + ); + logger.info(`coin history: ${j2s(historyJson)}`); + + // FIXME: If response seems wrong, report to auditor (in the future!); + + await ctx.wex.db.runReadWriteTx( + [ + "refreshGroups", + "refreshSessions", + "denominations", + "coins", + "coinAvailability", + ], + async (tx) => { + const rg = await tx.refreshGroups.get(ctx.refreshGroupId); + if (!rg) { + return; + } + if (rg.timestampFinished) { + return; + } + if (rg.statusPerCoin[coinIndex] !== RefreshCoinStatus.Pending) { + return; + } + if (Amounts.isZero(historyJson.balance)) { + rg.statusPerCoin[coinIndex] = RefreshCoinStatus.Failed; + const refreshSession = await tx.refreshSessions.get([ + ctx.refreshGroupId, + coinIndex, + ]); + if (!refreshSession) { + throw Error( + "db invariant failed: missing refresh session in database", + ); + } + refreshSession.lastError = errDetails; + await tx.refreshGroups.put(rg); + await tx.refreshSessions.put(refreshSession); + } else { + // Try again with new denoms! + rg.inputPerCoin[coinIndex] = historyJson.balance; + const refreshSession = await tx.refreshSessions.get([ + ctx.refreshGroupId, + coinIndex, + ]); + if (!refreshSession) { + throw Error( + "db invariant failed: missing refresh session in database", + ); + } + await destroyRefreshSession(ctx.wex, tx, rg, refreshSession); + await tx.refreshSessions.delete([ctx.refreshGroupId, coinIndex]); + await initRefreshSession(ctx.wex, tx, rg, coinIndex); + } + }, + ); +} + +async function handleRefreshMeltNotFound( + ctx: RefreshTransactionContext, + coinIndex: number, + errDetails: TalerErrorDetail, +): Promise<void> { + // FIXME: Validate the exchange's error response + await ctx.wex.db.runReadWriteTx( + [ + "refreshGroups", + "refreshSessions", + "coins", + "denominations", + "coinAvailability", + ], + async (tx) => { + const rg = await tx.refreshGroups.get(ctx.refreshGroupId); + if (!rg) { + return; + } + if (rg.timestampFinished) { + return; + } + if (rg.statusPerCoin[coinIndex] !== RefreshCoinStatus.Pending) { + return; + } + rg.statusPerCoin[coinIndex] = RefreshCoinStatus.Failed; + const refreshSession = await tx.refreshSessions.get([ + ctx.refreshGroupId, + coinIndex, + ]); + if (!refreshSession) { + throw Error("db invariant failed: missing refresh session in database"); + } + await destroyRefreshSession(ctx.wex, tx, rg, refreshSession); + refreshSession.lastError = errDetails; + await tx.refreshGroups.put(rg); + await tx.refreshSessions.put(refreshSession); + }, + ); +} + export async function assembleRefreshRevealRequest(args: { cryptoApi: TalerCryptoInterface; derived: DerivedRefreshSession; @@ -741,6 +964,7 @@ async function refreshReveal( logger.trace( `doing refresh reveal for ${refreshGroupId} (old coin ${coinIndex})`, ); + const ctx = new RefreshTransactionContext(wex, refreshGroupId); const d = await wex.db.runReadOnlyTx( ["refreshGroups", "refreshSessions", "coins", "denominations"], async (tx) => { @@ -868,6 +1092,21 @@ async function refreshReveal( }, ); + switch (resp.status) { + case HttpStatusCode.Ok: + break; + case HttpStatusCode.Conflict: + case HttpStatusCode.Gone: { + const errDetail = await readTalerErrorResponse(resp); + await handleRefreshRevealError(ctx, coinIndex, errDetail); + return; + } + default: { + const errDetail = await readTalerErrorResponse(resp); + throwUnexpectedRequestError(resp, errDetail); + } + } + const reveal = await readSuccessResponseJsonOrThrow( resp, codecForExchangeRevealResponse(), @@ -975,6 +1214,46 @@ async function refreshReveal( logger.trace("refresh finished (end of reveal)"); } +async function handleRefreshRevealError( + ctx: RefreshTransactionContext, + coinIndex: number, + errDetails: TalerErrorDetail, +): Promise<void> { + await ctx.wex.db.runReadWriteTx( + [ + "refreshGroups", + "refreshSessions", + "coins", + "denominations", + "coinAvailability", + ], + async (tx) => { + const rg = await tx.refreshGroups.get(ctx.refreshGroupId); + if (!rg) { + return; + } + if (rg.timestampFinished) { + return; + } + if (rg.statusPerCoin[coinIndex] !== RefreshCoinStatus.Pending) { + return; + } + rg.statusPerCoin[coinIndex] = RefreshCoinStatus.Failed; + const refreshSession = await tx.refreshSessions.get([ + ctx.refreshGroupId, + coinIndex, + ]); + if (!refreshSession) { + throw Error("db invariant failed: missing refresh session in database"); + } + refreshSession.lastError = errDetails; + await destroyRefreshSession(ctx.wex, tx, rg, refreshSession); + await tx.refreshGroups.put(rg); + await tx.refreshSessions.put(refreshSession); + }, + ); +} + export async function processRefreshGroup( wex: WalletExecutionContext, refreshGroupId: string, @@ -1444,7 +1723,7 @@ export async function forceRefresh( wex: WalletExecutionContext, req: ForceRefreshRequest, ): Promise<ForceRefreshResult> { - if (req.coinPubList.length == 0) { + if (req.refreshCoinSpecs.length == 0) { throw Error("refusing to create empty refresh group"); } const res = await wex.db.runReadWriteTx( @@ -1457,8 +1736,8 @@ export async function forceRefresh( ], async (tx) => { let coinPubs: CoinRefreshRequest[] = []; - for (const c of req.coinPubList) { - const coin = await tx.coins.get(c); + for (const c of req.refreshCoinSpecs) { + const coin = await tx.coins.get(c.coinPub); if (!coin) { throw Error(`coin (pubkey ${c}) not found`); } @@ -1470,8 +1749,8 @@ export async function forceRefresh( ); checkDbInvariant(!!denom); coinPubs.push({ - coinPub: c, - amount: denom?.value, + coinPub: c.coinPub, + amount: c.amount ?? denom.value, }); } return await createRefreshGroup( |