diff options
author | Florian Dold <florian@dold.me> | 2024-04-02 18:19:11 +0200 |
---|---|---|
committer | Florian Dold <florian@dold.me> | 2024-04-02 18:40:02 +0200 |
commit | 651db75296bfe7c35dc7d29e39f25e6dacb72930 (patch) | |
tree | e508ebc04f93e652d6fbbfa4b0193878247ac413 /packages/taler-wallet-core/src/refresh.ts | |
parent | b4185a6ef452fba47e481264af06e67524391584 (diff) | |
download | wallet-core-651db75296bfe7c35dc7d29e39f25e6dacb72930.tar.xz |
wallet-core: refresh cleanup, preparations for #8568
Diffstat (limited to 'packages/taler-wallet-core/src/refresh.ts')
-rw-r--r-- | packages/taler-wallet-core/src/refresh.ts | 407 |
1 files changed, 208 insertions, 199 deletions
diff --git a/packages/taler-wallet-core/src/refresh.ts b/packages/taler-wallet-core/src/refresh.ts index 516d5e3da..490b1b5f5 100644 --- a/packages/taler-wallet-core/src/refresh.ts +++ b/packages/taler-wallet-core/src/refresh.ts @@ -66,7 +66,6 @@ import { } from "@gnu-taler/taler-util/http"; import { constructTaskIdentifier, - makeCoinAvailable, makeCoinsVisible, PendingTaskType, TaskIdStr, @@ -82,6 +81,7 @@ import { } from "./crypto/cryptoTypes.js"; import { CryptoApiStoppedError } from "./crypto/workers/crypto-dispatcher.js"; import { + CoinAvailabilityRecord, CoinRecord, CoinSourceType, DenominationRecord, @@ -305,192 +305,130 @@ export function getTotalRefreshCost( return totalCost; } -function updateGroupStatus(rg: RefreshGroupRecord): { final: boolean } { - const allFinal = fnutil.all( - rg.statusPerCoin, - (x) => x === RefreshCoinStatus.Finished || x === RefreshCoinStatus.Failed, - ); - const anyFailed = fnutil.any( - rg.statusPerCoin, - (x) => x === RefreshCoinStatus.Failed, - ); - if (allFinal) { - if (anyFailed) { - rg.timestampFinished = timestampPreciseToDb(TalerPreciseTimestamp.now()); - rg.operationStatus = RefreshOperationStatus.Failed; - } else { - rg.timestampFinished = timestampPreciseToDb(TalerPreciseTimestamp.now()); - rg.operationStatus = RefreshOperationStatus.Finished; - } - return { final: true }; +export async function getCoinAvailabilityForDenom( + wex: WalletExecutionContext, + tx: WalletDbReadWriteTransaction< + ["coins", "coinAvailability", "denominations"] + >, + denom: DenominationInfo, + ageRestriction: number, +): Promise<CoinAvailabilityRecord> { + checkDbInvariant(!!denom); + let car = await tx.coinAvailability.get([ + denom.exchangeBaseUrl, + denom.denomPubHash, + ageRestriction, + ]); + if (!car) { + car = { + maxAge: ageRestriction, + value: denom.value, + currency: Amounts.currencyOf(denom.value), + denomPubHash: denom.denomPubHash, + exchangeBaseUrl: denom.exchangeBaseUrl, + freshCoinCount: 0, + visibleCoinCount: 0, + }; } - return { final: false }; + return car; } /** * Create a refresh session for one particular coin inside a refresh group. - * - * If the session already exists, return the existing one. - * - * If the session doesn't need to be created (refresh group gone or session already - * finished), return undefined. */ -async function provideRefreshSession( +async function initRefreshSession( wex: WalletExecutionContext, - refreshGroupId: string, + tx: WalletDbReadWriteTransaction< + ["refreshSessions", "coinAvailability", "coins", "denominations"] + >, + refreshGroup: RefreshGroupRecord, coinIndex: number, -): Promise<RefreshSessionRecord | undefined> { +): Promise<void> { + const refreshGroupId = refreshGroup.refreshGroupId; logger.trace( `creating refresh session for coin ${coinIndex} in refresh group ${refreshGroupId}`, ); - - const d = await wex.db.runReadWriteTx( - ["coins", "refreshGroups", "refreshSessions"], - async (tx) => { - const refreshGroup = await tx.refreshGroups.get(refreshGroupId); - if (!refreshGroup) { - return; - } - if ( - refreshGroup.statusPerCoin[coinIndex] === RefreshCoinStatus.Finished - ) { - return; - } - const existingRefreshSession = await tx.refreshSessions.get([ - refreshGroupId, - coinIndex, - ]); - const oldCoinPub = refreshGroup.oldCoinPubs[coinIndex]; - const coin = await tx.coins.get(oldCoinPub); - if (!coin) { - throw Error("Can't refresh, coin not found"); - } - return { refreshGroup, coin, existingRefreshSession }; - }, - ); - - if (!d) { - return undefined; - } - - if (d.existingRefreshSession) { - return d.existingRefreshSession; + const oldCoinPub = refreshGroup.oldCoinPubs[coinIndex]; + const oldCoin = await tx.coins.get(oldCoinPub); + if (!oldCoin) { + throw Error("Can't refresh, coin not found"); } - const { refreshGroup, coin } = d; - - const exch = await fetchFreshExchange(wex, coin.exchangeBaseUrl); + const exchangeBaseUrl = oldCoin.exchangeBaseUrl; - // FIXME: use helper functions from withdraw.ts - // to update and filter withdrawable denoms. + const sessionSecretSeed = encodeCrock(getRandomBytes(64)); - const { availableAmount, availableDenoms } = await wex.db.runReadOnlyTx( - ["denominations"], - async (tx) => { - const oldDenom = await getDenomInfo( - wex, - tx, - exch.exchangeBaseUrl, - coin.denomPubHash, - ); + const oldDenom = await getDenomInfo( + wex, + tx, + exchangeBaseUrl, + oldCoin.denomPubHash, + ); - if (!oldDenom) { - throw Error("db inconsistent: denomination for coin not found"); - } + if (!oldDenom) { + throw Error("db inconsistent: denomination for coin not found"); + } - // FIXME: Use denom groups instead of querying all denominations! - const availableDenoms: DenominationRecord[] = - await tx.denominations.indexes.byExchangeBaseUrl.getAll( - exch.exchangeBaseUrl, - ); + const currency = refreshGroup.currency; - const availableAmount = Amounts.sub( - refreshGroup.inputPerCoin[coinIndex], - oldDenom.feeRefresh, - ).amount; - return { availableAmount, availableDenoms }; - }, + const availableDenoms = await getCandidateWithdrawalDenomsTx( + wex, + tx, + exchangeBaseUrl, + currency, ); + const availableAmount = Amounts.sub( + refreshGroup.inputPerCoin[coinIndex], + oldDenom.feeRefresh, + ).amount; + const newCoinDenoms = selectWithdrawalDenominations( availableAmount, availableDenoms, wex.ws.config.testing.denomselAllowLate, ); - const transactionId = constructTransactionIdentifier({ - tag: TransactionType.Refresh, - refreshGroupId, - }); - if (newCoinDenoms.selectedDenoms.length === 0) { logger.trace( `not refreshing, available amount ${amountToPretty( availableAmount, )} too small`, ); - const transitionInfo = await wex.db.runReadWriteTx( - ["refreshGroups", "coins", "coinAvailability"], - async (tx) => { - const rg = await tx.refreshGroups.get(refreshGroupId); - if (!rg) { - return; - } - const oldTxState = computeRefreshTransactionState(rg); - rg.statusPerCoin[coinIndex] = RefreshCoinStatus.Finished; - const updateRes = updateGroupStatus(rg); - if (updateRes.final) { - await makeCoinsVisible(wex, tx, transactionId); - } - await tx.refreshGroups.put(rg); - const newTxState = computeRefreshTransactionState(rg); - return { oldTxState, newTxState }; - }, - ); - wex.ws.notify({ - type: NotificationType.BalanceChange, - hintTransactionId: transactionId, - }); - notifyTransition(wex, transactionId, transitionInfo); - return; + refreshGroup.statusPerCoin[coinIndex] = RefreshCoinStatus.Finished; } - const sessionSecretSeed = encodeCrock(getRandomBytes(64)); + for (let i = 0; i < newCoinDenoms.selectedDenoms.length; i++) { + const dph = newCoinDenoms.selectedDenoms[i].denomPubHash; + const denom = await getDenomInfo(wex, tx, oldDenom.exchangeBaseUrl, dph); + if (!denom) { + logger.error(`denom ${dph} not in DB`); + continue; + } + const car = await getCoinAvailabilityForDenom( + wex, + tx, + denom, + oldCoin.maxAge, + ); + car.pendingRefreshOutputCount = + (car.pendingRefreshOutputCount ?? 0) + + newCoinDenoms.selectedDenoms[i].count; + await tx.coinAvailability.put(car); + } - // Store refresh session for this coin in the database. - const mySession = await wex.db.runReadWriteTx( - ["refreshGroups", "refreshSessions"], - async (tx) => { - const rg = await tx.refreshGroups.get(refreshGroupId); - if (!rg) { - return; - } - const existingSession = await tx.refreshSessions.get([ - refreshGroupId, - coinIndex, - ]); - if (existingSession) { - return existingSession; - } - const newSession: RefreshSessionRecord = { - coinIndex, - refreshGroupId, - norevealIndex: undefined, - sessionSecretSeed: sessionSecretSeed, - newDenoms: newCoinDenoms.selectedDenoms.map((x) => ({ - count: x.count, - denomPubHash: x.denomPubHash, - })), - amountRefreshOutput: Amounts.stringify(newCoinDenoms.totalCoinValue), - }; - await tx.refreshSessions.put(newSession); - return newSession; - }, - ); - logger.trace( - `found/created refresh session for coin #${coinIndex} in ${refreshGroupId}`, - ); - return mySession; + const newSession: RefreshSessionRecord = { + coinIndex, + refreshGroupId, + norevealIndex: undefined, + sessionSecretSeed: sessionSecretSeed, + newDenoms: newCoinDenoms.selectedDenoms.map((x) => ({ + count: x.count, + denomPubHash: x.denomPubHash, + })), + amountRefreshOutput: Amounts.stringify(newCoinDenoms.totalCoinValue), + }; + await tx.refreshSessions.put(newSession); } function getRefreshRequestTimeout(rg: RefreshGroupRecord): Duration { @@ -499,6 +437,14 @@ function getRefreshRequestTimeout(rg: RefreshGroupRecord): Duration { }); } +/** + * Run the melt step of a refresh session. + * + * If the melt step succeeds or fails permanently, + * the status in the refresh group is updated. + * + * When a transient error occurs, an exception is thrown. + */ async function refreshMelt( wex: WalletExecutionContext, refreshGroupId: string, @@ -627,7 +573,7 @@ async function refreshMelt( if (resp.status === HttpStatusCode.NotFound) { const errDetails = await readUnexpectedResponseDetails(resp); - const transitionInfo = await wex.db.runReadWriteTx( + await wex.db.runReadWriteTx( ["refreshGroups", "refreshSessions", "coins", "coinAvailability"], async (tx) => { const rg = await tx.refreshGroups.get(refreshGroupId); @@ -640,7 +586,6 @@ async function refreshMelt( if (rg.statusPerCoin[coinIndex] !== RefreshCoinStatus.Pending) { return; } - const oldTxState = computeRefreshTransactionState(rg); rg.statusPerCoin[coinIndex] = RefreshCoinStatus.Failed; const refreshSession = await tx.refreshSessions.get([ refreshGroupId, @@ -652,24 +597,10 @@ async function refreshMelt( ); } refreshSession.lastError = errDetails; - const updateRes = updateGroupStatus(rg); - if (updateRes.final) { - await makeCoinsVisible(wex, tx, transactionId); - } await tx.refreshGroups.put(rg); await tx.refreshSessions.put(refreshSession); - const newTxState = computeRefreshTransactionState(rg); - return { - oldTxState, - newTxState, - }; }, ); - wex.ws.notify({ - type: NotificationType.BalanceChange, - hintTransactionId: transactionId, - }); - notifyTransition(wex, transactionId, transitionInfo); return; } @@ -718,6 +649,7 @@ async function refreshMelt( const input = Amounts.parseOrThrow(rg.inputPerCoin[rs.coinIndex]); const newSel = selectWithdrawalDenominations(input, candidates); rs.amountRefreshOutput = newSel.totalCoinValue; + // FIXME: This is wrong! When denoms are re-selected, the melt commitment breaks. rs.newDenoms = newSel.selectedDenoms.map((x) => ({ count: x.count, denomPubHash: x.denomPubHash, @@ -1043,7 +975,7 @@ async function refreshReveal( } } - const transitionInfo = await wex.db.runReadWriteTx( + await wex.db.runReadWriteTx( [ "coins", "denominations", @@ -1061,19 +993,32 @@ async function refreshReveal( if (!rs) { return; } - const oldTxState = computeRefreshTransactionState(rg); rg.statusPerCoin[coinIndex] = RefreshCoinStatus.Finished; - updateGroupStatus(rg); for (const coin of coins) { - await makeCoinAvailable(wex, tx, coin); + const denomInfo = await getDenomInfo( + wex, + tx, + coin.exchangeBaseUrl, + coin.denomPubHash, + ); + checkDbInvariant(!!denomInfo); + const car = await getCoinAvailabilityForDenom( + wex, + tx, + denomInfo, + coin.maxAge, + ); + checkDbInvariant( + car.pendingRefreshOutputCount != null && + car.pendingRefreshOutputCount > 0, + ); + car.pendingRefreshOutputCount--; + car.freshCoinCount++; + await tx.coinAvailability.put(car); } - await makeCoinsVisible(wex, tx, transactionId); await tx.refreshGroups.put(rg); - const newTxState = computeRefreshTransactionState(rg); - return { oldTxState, newTxState }; }, ); - notifyTransition(wex, transactionId, transitionInfo); logger.trace("refresh finished (end of reveal)"); } @@ -1121,17 +1066,71 @@ export async function processRefreshGroup( errors.push(getErrorDetailFromException(x)); }), ); - try { - logger.info("waiting for refreshes"); - await Promise.all(ps); - logger.info("refresh group finished"); - } catch (e) { - logger.warn("process refresh sessions got exception"); - logger.warn(`exception: ${e}`); - } + await Promise.all(ps); if (inShutdown) { - return TaskRunResult.backoff(); + return TaskRunResult.finished(); } + + const ctx = new RefreshTransactionContext(wex, refreshGroupId); + + // We've processed all refresh session and can now update the + // status of the whole refresh group. + + const transitionInfo = await wex.db.runReadWriteTx( + ["coins", "coinAvailability", "refreshGroups"], + async (tx) => { + const rg = await tx.refreshGroups.get(refreshGroupId); + if (!rg) { + return; + } + switch (rg.operationStatus) { + case RefreshOperationStatus.Pending: + break; + default: + return undefined; + } + const oldTxState = computeRefreshTransactionState(rg); + const allFinal = fnutil.all( + rg.statusPerCoin, + (x) => + x === RefreshCoinStatus.Finished || x === RefreshCoinStatus.Failed, + ); + const anyFailed = fnutil.any( + rg.statusPerCoin, + (x) => x === RefreshCoinStatus.Failed, + ); + if (allFinal) { + if (anyFailed) { + rg.timestampFinished = timestampPreciseToDb( + TalerPreciseTimestamp.now(), + ); + rg.operationStatus = RefreshOperationStatus.Failed; + } else { + rg.timestampFinished = timestampPreciseToDb( + TalerPreciseTimestamp.now(), + ); + rg.operationStatus = RefreshOperationStatus.Finished; + } + } + if (allFinal) { + await makeCoinsVisible(wex, tx, ctx.transactionId); + await tx.refreshGroups.put(rg); + const newTxState = computeRefreshTransactionState(rg); + return { + oldTxState, + newTxState, + }; + } else { + return undefined; + } + }, + ); + + if (transitionInfo) { + notifyTransition(wex, ctx.transactionId, transitionInfo); + return TaskRunResult.progress(); + } + if (errors.length > 0) { return { type: TaskRunResultType.Error, @@ -1174,16 +1173,7 @@ async function processRefreshSession( return; } if (!refreshSession) { - refreshSession = await provideRefreshSession( - wex, - refreshGroupId, - coinIndex, - ); - } - if (!refreshSession) { - // We tried to create the refresh session, but didn't get a result back. - // This means that either the session is finished, or that creating - // one isn't necessary. + // No refresh session for that coin. return; } if (refreshSession.norevealIndex === undefined) { @@ -1268,7 +1258,7 @@ export async function calculateRefreshOutput( }; } -async function applyRefresh( +async function applyRefreshToOldCoins( wex: WalletExecutionContext, tx: WalletDbReadWriteTransaction< ["denominations", "coins", "refreshGroups", "coinAvailability"] @@ -1347,20 +1337,29 @@ export interface CreateRefreshGroupResult { export async function createRefreshGroup( wex: WalletExecutionContext, tx: WalletDbReadWriteTransaction< - ["denominations", "coins", "refreshGroups", "coinAvailability"] + [ + "denominations", + "coins", + "refreshGroups", + "refreshSessions", + "coinAvailability", + ] >, currency: string, oldCoinPubs: CoinRefreshRequest[], refreshReason: RefreshReason, originatingTransactionId: string | undefined, ): Promise<CreateRefreshGroupResult> { + // FIXME: Check that involved exchanges are reasonably up-to-date. + // Otherwise, error out. + const refreshGroupId = encodeCrock(getRandomBytes(32)); const outInfo = await calculateRefreshOutput(wex, tx, currency, oldCoinPubs); const estimatedOutputPerCoin = outInfo.outputPerCoin; - await applyRefresh(wex, tx, oldCoinPubs, refreshGroupId); + await applyRefreshToOldCoins(wex, tx, oldCoinPubs, refreshGroupId); const refreshGroup: RefreshGroupRecord = { operationStatus: RefreshOperationStatus.Pending, @@ -1387,6 +1386,10 @@ export async function createRefreshGroup( refreshGroup.operationStatus = RefreshOperationStatus.Finished; } + for (let i = 0; i < oldCoinPubs.length; i++) { + await initRefreshSession(wex, tx, refreshGroup, i); + } + await tx.refreshGroups.put(refreshGroup); const newTxState = computeRefreshTransactionState(refreshGroup); @@ -1487,7 +1490,13 @@ export async function forceRefresh( throw Error("refusing to create empty refresh group"); } const res = await wex.db.runReadWriteTx( - ["refreshGroups", "coinAvailability", "denominations", "coins"], + [ + "refreshGroups", + "coinAvailability", + "refreshSessions", + "denominations", + "coins", + ], async (tx) => { let coinPubs: CoinRefreshRequest[] = []; for (const c of req.coinPubList) { |