taler/wallet-core
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Activity

wallet-core: implement partial withdrawal batching, don't block when generating planchets

Browse Source
This commit is contained in:
Florian Dold 2023-02-10 13:21:37 +01:00
parent c4180e1290
commit 18c30b9a00
No known key found for this signature in database
GPG Key ID: D2E4F00F29D02A4B
4 changed files with 210 additions and 206 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

10
packages/taler-harness/src/harness/harness.ts
View File

@ -1361,7 +1361,12 @@ export class ExchangeService implements ExchangeServiceInterface {
this.exchangeWirewatchProc = this.globalState.spawnService( this.exchangeWirewatchProc = this.globalState.spawnService(
"taler-exchange-wirewatch", "taler-exchange-wirewatch",
["-c", this.configFilename, ...this.timetravelArgArr], [
"-c",
this.configFilename,
"--longpoll-timeout=5s",
...this.timetravelArgArr,
],
`exchange-wirewatch-${this.name}`, `exchange-wirewatch-${this.name}`,
); );
@ -1951,6 +1956,9 @@ export class WalletService {
], ],
`wallet-${this.opts.name}`, `wallet-${this.opts.name}`,
); );
logger.info(
`hint: connect to wallet using taler-wallet-cli --wallet-connection=${unixPath}`,
);
} }
async pingUntilAvailable(): Promise<void> { async pingUntilAvailable(): Promise<void> {

3
packages/taler-harness/src/integrationtests/test-withdrawal-huge.ts
View File

@ -87,9 +87,10 @@ export async function runWithdrawalHugeTest(t: GlobalTestState) {
exchangeBaseUrl: exchange.baseUrl, exchangeBaseUrl: exchange.baseUrl,
}); });
// Results in about 1K coins withdrawn
await wallet.client.call(WalletApiOperation.WithdrawFakebank, { await wallet.client.call(WalletApiOperation.WithdrawFakebank, {
exchange: exchange.baseUrl, exchange: exchange.baseUrl,
amount: "TESTKUDOS:5000", amount: "TESTKUDOS:10000",
bank: bank.baseUrl, bank: bank.baseUrl,
}); });

19
packages/taler-util/src/taler-types.ts
View File

@ -951,12 +951,12 @@ export const codecForBlindedDenominationSignature = () =>
.alternative(DenomKeyType.Rsa, codecForRsaBlindedDenominationSignature()) .alternative(DenomKeyType.Rsa, codecForRsaBlindedDenominationSignature())
.build("BlindedDenominationSignature"); .build("BlindedDenominationSignature");
export class WithdrawResponse { export class ExchangeWithdrawResponse {
ev_sig: BlindedDenominationSignature; ev_sig: BlindedDenominationSignature;
} }
export class WithdrawBatchResponse { export class ExchangeWithdrawBatchResponse {
ev_sigs: WithdrawResponse[]; ev_sigs: ExchangeWithdrawResponse[];
} }
export interface MerchantPayResponse { export interface MerchantPayResponse {
@ -1476,13 +1476,13 @@ export const codecForRecoupConfirmation = (): Codec<RecoupConfirmation> =>
.property("old_coin_pub", codecOptional(codecForString())) .property("old_coin_pub", codecOptional(codecForString()))
.build("RecoupConfirmation"); .build("RecoupConfirmation");
export const codecForWithdrawResponse = (): Codec<WithdrawResponse> => export const codecForWithdrawResponse = (): Codec<ExchangeWithdrawResponse> =>
buildCodecForObject<WithdrawResponse>() buildCodecForObject<ExchangeWithdrawResponse>()
.property("ev_sig", codecForBlindedDenominationSignature()) .property("ev_sig", codecForBlindedDenominationSignature())
.build("WithdrawResponse"); .build("WithdrawResponse");
export const codecForWithdrawBatchResponse = (): Codec<WithdrawBatchResponse> => export const codecForWithdrawBatchResponse = (): Codec<ExchangeWithdrawBatchResponse> =>
buildCodecForObject<WithdrawBatchResponse>() buildCodecForObject<ExchangeWithdrawBatchResponse>()
.property("ev_sigs", codecForList(codecForWithdrawResponse())) .property("ev_sigs", codecForList(codecForWithdrawResponse()))
.build("WithdrawBatchResponse"); .build("WithdrawBatchResponse");
@ -1753,6 +1753,11 @@ export interface ExchangeWithdrawRequest {
coin_ev: CoinEnvelope; coin_ev: CoinEnvelope;
} }
export interface ExchangeBatchWithdrawRequest {
planchets: ExchangeWithdrawRequest[];
}
export interface ExchangeRefreshRevealRequest { export interface ExchangeRefreshRevealRequest {
new_denoms_h: HashCodeString[]; new_denoms_h: HashCodeString[];
coin_evs: CoinEnvelope[]; coin_evs: CoinEnvelope[];

384
packages/taler-wallet-core/src/operations/withdraw.ts
View File

@ -59,9 +59,11 @@ import {
TransactionType, TransactionType,
UnblindedSignature, UnblindedSignature,
URL, URL,
WithdrawBatchResponse, ExchangeWithdrawBatchResponse,
WithdrawResponse, ExchangeWithdrawResponse,
WithdrawUriInfoResponse, WithdrawUriInfoResponse,
ExchangeBatchWithdrawRequest,
WalletNotification,
} from "@gnu-taler/taler-util"; } from "@gnu-taler/taler-util";
import { EddsaKeypair } from "../crypto/cryptoImplementation.js"; import { EddsaKeypair } from "../crypto/cryptoImplementation.js";
import { import {
@ -93,6 +95,7 @@ import {
import { walletCoreDebugFlags } from "../util/debugFlags.js"; import { walletCoreDebugFlags } from "../util/debugFlags.js";
import { import {
HttpRequestLibrary, HttpRequestLibrary,
HttpResponse,
readSuccessResponseJsonOrErrorCode, readSuccessResponseJsonOrErrorCode,
readSuccessResponseJsonOrThrow, readSuccessResponseJsonOrThrow,
throwUnexpectedRequestError, throwUnexpectedRequestError,
@ -455,21 +458,43 @@ async function processPlanchetGenerate(
}); });
} }
interface WithdrawalRequestBatchArgs {
/**
* Use the batched request on the network level.
* Not supported by older exchanges.
*/
useBatchRequest: boolean;
coinStartIndex: number;
batchSize: number;
}
interface WithdrawalBatchResult {
coinIdxs: number[];
batchResp: ExchangeWithdrawBatchResponse;
}
/** /**
* Send the withdrawal request for a generated planchet to the exchange. * Send the withdrawal request for a generated planchet to the exchange.
* *
* The verification of the response is done asynchronously to enable parallelism. * The verification of the response is done asynchronously to enable parallelism.
*/ */
async function processPlanchetExchangeRequest( async function processPlanchetExchangeBatchRequest(
ws: InternalWalletState, ws: InternalWalletState,
wgContext: WithdrawalGroupContext, wgContext: WithdrawalGroupContext,
coinIdx: number, args: WithdrawalRequestBatchArgs,
): Promise<WithdrawResponse | undefined> { ): Promise<WithdrawalBatchResult> {
const withdrawalGroup: WithdrawalGroupRecord = wgContext.wgRecord; const withdrawalGroup: WithdrawalGroupRecord = wgContext.wgRecord;
logger.info( logger.info(
`processing planchet exchange request ${withdrawalGroup.withdrawalGroupId}/${coinIdx}`, `processing planchet exchange batch request ${withdrawalGroup.withdrawalGroupId}, start=${args.coinStartIndex}, len=${args.batchSize}`,
); );
const d = await ws.db
const batchReq: ExchangeBatchWithdrawRequest = { planchets: [] };
// Indices of coins that are included in the batch request
const coinIdxs: number[] = [];
await ws.db
.mktx((x) => [ .mktx((x) => [
x.withdrawalGroups, x.withdrawalGroups,
x.planchets, x.planchets,
@ -477,96 +502,88 @@ async function processPlanchetExchangeRequest(
x.denominations, x.denominations,
]) ])
.runReadOnly(async (tx) => { .runReadOnly(async (tx) => {
let planchet = await tx.planchets.indexes.byGroupAndIndex.get([ for (
withdrawalGroup.withdrawalGroupId, let coinIdx = args.coinStartIndex;
coinIdx, coinIdx < args.coinStartIndex + args.batchSize &&
]); coinIdx < wgContext.numPlanchets;
if (!planchet) { coinIdx++
return; ) {
let planchet = await tx.planchets.indexes.byGroupAndIndex.get([
withdrawalGroup.withdrawalGroupId,
coinIdx,
]);
if (!planchet) {
continue;
}
if (planchet.planchetStatus === PlanchetStatus.WithdrawalDone) {
logger.warn("processPlanchet: planchet already withdrawn");
continue;
}
const denom = await ws.getDenomInfo(
ws,
tx,
withdrawalGroup.exchangeBaseUrl,
planchet.denomPubHash,
);
if (!denom) {
logger.error("db inconsistent: denom for planchet not found");
continue;
}
const planchetReq: ExchangeWithdrawRequest = {
denom_pub_hash: planchet.denomPubHash,
reserve_sig: planchet.withdrawSig,
coin_ev: planchet.coinEv,
};
batchReq.planchets.push(planchetReq);
coinIdxs.push(coinIdx);
} }
if (planchet.planchetStatus === PlanchetStatus.WithdrawalDone) {
logger.warn("processPlanchet: planchet already withdrawn");
return;
}
const exchange = await tx.exchanges.get(withdrawalGroup.exchangeBaseUrl);
if (!exchange) {
logger.error("db inconsistent: exchange for planchet not found");
return;
}
const denom = await ws.getDenomInfo(
ws,
tx,
withdrawalGroup.exchangeBaseUrl,
planchet.denomPubHash,
);
if (!denom) {
logger.error("db inconsistent: denom for planchet not found");
return;
}
logger.trace(
`processing planchet #${coinIdx} in withdrawal ${withdrawalGroup.withdrawalGroupId}`,
);
const reqBody: ExchangeWithdrawRequest = {
denom_pub_hash: planchet.denomPubHash,
reserve_sig: planchet.withdrawSig,
coin_ev: planchet.coinEv,
};
const reqUrl = new URL(
`reserves/${withdrawalGroup.reservePub}/withdraw`,
exchange.baseUrl,
).href;
return { reqUrl, reqBody };
}); });
if (!d) { if (batchReq.planchets.length == 0) {
return; logger.warn("empty withdrawal batch");
return {
batchResp: { ev_sigs: [] },
coinIdxs: [],
};
} }
const { reqUrl, reqBody } = d;
try { async function handleKycRequired(resp: HttpResponse, startIdx: number) {
const resp = await ws.http.postJson(reqUrl, reqBody); logger.info("withdrawal requires KYC");
if (resp.status === HttpStatusCode.UnavailableForLegalReasons) { const respJson = await resp.json();
logger.info("withdrawal requires KYC"); const uuidResp = codecForWalletKycUuid().decode(respJson);
const respJson = await resp.json(); logger.info(`kyc uuid response: ${j2s(uuidResp)}`);
const uuidResp = codecForWalletKycUuid().decode(respJson); await ws.db
logger.info(`kyc uuid response: ${j2s(uuidResp)}`); .mktx((x) => [x.planchets, x.withdrawalGroups])
await ws.db .runReadWrite(async (tx) => {
.mktx((x) => [x.planchets, x.withdrawalGroups]) for (let i = 0; i < startIdx; i++) {
.runReadWrite(async (tx) => {
let planchet = await tx.planchets.indexes.byGroupAndIndex.get([ let planchet = await tx.planchets.indexes.byGroupAndIndex.get([
withdrawalGroup.withdrawalGroupId, withdrawalGroup.withdrawalGroupId,
coinIdx, coinIdxs[i],
]); ]);
if (!planchet) { if (!planchet) {
return; continue;
} }
planchet.planchetStatus = PlanchetStatus.KycRequired; planchet.planchetStatus = PlanchetStatus.KycRequired;
const wg2 = await tx.withdrawalGroups.get(
withdrawalGroup.withdrawalGroupId,
);
if (!wg2) {
return;
}
wg2.kycPending = {
paytoHash: uuidResp.h_payto,
requirementRow: uuidResp.requirement_row,
};
await tx.planchets.put(planchet); await tx.planchets.put(planchet);
await tx.withdrawalGroups.put(wg2); }
}); const wg2 = await tx.withdrawalGroups.get(
return; withdrawalGroup.withdrawalGroupId,
} );
const r = await readSuccessResponseJsonOrThrow( if (!wg2) {
resp, return;
codecForWithdrawResponse(), }
); wg2.kycPending = {
return r; paytoHash: uuidResp.h_payto,
} catch (e) { requirementRow: uuidResp.requirement_row,
};
await tx.withdrawalGroups.put(wg2);
});
return;
}
async function storeCoinError(e: any, coinIdx: number) {
const errDetail = getErrorDetailFromException(e); const errDetail = getErrorDetailFromException(e);
logger.trace("withdrawal request failed", e); logger.trace("withdrawal request failed", e);
logger.trace(String(e)); logger.trace(String(e));
@ -583,101 +600,81 @@ async function processPlanchetExchangeRequest(
planchet.lastError = errDetail; planchet.lastError = errDetail;
await tx.planchets.put(planchet); await tx.planchets.put(planchet);
}); });
return;
} }
}
/** // FIXME: handle individual error codes better!
* Send the withdrawal request for a generated planchet to the exchange.
* if (args.useBatchRequest) {
* The verification of the response is done asynchronously to enable parallelism. const reqUrl = new URL(
*/ `reserves/${withdrawalGroup.reservePub}/batch-withdraw`,
async function processPlanchetExchangeBatchRequest( withdrawalGroup.exchangeBaseUrl,
ws: InternalWalletState, ).href;
wgContext: WithdrawalGroupContext,
): Promise<WithdrawBatchResponse | undefined> { try {
const withdrawalGroup: WithdrawalGroupRecord = wgContext.wgRecord; const resp = await ws.http.postJson(reqUrl, batchReq);
logger.info( if (resp.status === HttpStatusCode.UnavailableForLegalReasons) {
`processing planchet exchange batch request ${withdrawalGroup.withdrawalGroupId}`, await handleKycRequired(resp, 0);
); }
const numTotalCoins = withdrawalGroup.denomsSel.selectedDenoms const r = await readSuccessResponseJsonOrThrow(
.map((x) => x.count) resp,
.reduce((a, b) => a + b); codecForWithdrawBatchResponse(),
const d = await ws.db );
.mktx((x) => [ return {
x.withdrawalGroups, coinIdxs,
x.planchets, batchResp: r,
x.exchanges,
x.denominations,
])
.runReadOnly(async (tx) => {
const reqBody: { planchets: ExchangeWithdrawRequest[] } = {
planchets: [],
}; };
const exchange = await tx.exchanges.get(withdrawalGroup.exchangeBaseUrl); } catch (e) {
if (!exchange) { await storeCoinError(e, coinIdxs[0]);
logger.error("db inconsistent: exchange for planchet not found"); return {
return; batchResp: { ev_sigs: [] },
} coinIdxs: [],
};
for (let coinIdx = 0; coinIdx < numTotalCoins; coinIdx++) { }
let planchet = await tx.planchets.indexes.byGroupAndIndex.get([ } else {
withdrawalGroup.withdrawalGroupId, // We emulate the batch response here by making multiple individual requests
coinIdx, const responses: ExchangeWithdrawBatchResponse = {
]); ev_sigs: [],
if (!planchet) { };
return; for (let i = 0; i < batchReq.planchets.length; i++) {
} try {
if (planchet.planchetStatus === PlanchetStatus.WithdrawalDone) { const p = batchReq.planchets[i];
logger.warn("processPlanchet: planchet already withdrawn"); const reqUrl = new URL(
return; `reserves/${withdrawalGroup.reservePub}/withdraw`,
}
const denom = await ws.getDenomInfo(
ws,
tx,
withdrawalGroup.exchangeBaseUrl, withdrawalGroup.exchangeBaseUrl,
planchet.denomPubHash, ).href;
); const resp = await ws.http.postJson(reqUrl, p);
if (resp.status === HttpStatusCode.UnavailableForLegalReasons) {
if (!denom) { await handleKycRequired(resp, i);
logger.error("db inconsistent: denom for planchet not found"); // We still return blinded coins that we could actually withdraw.
return; return {
coinIdxs,
batchResp: responses,
};
} }
const r = await readSuccessResponseJsonOrThrow(
const planchetReq: ExchangeWithdrawRequest = { resp,
denom_pub_hash: planchet.denomPubHash, codecForWithdrawResponse(),
reserve_sig: planchet.withdrawSig, );
coin_ev: planchet.coinEv, responses.ev_sigs.push(r);
}; } catch (e) {
reqBody.planchets.push(planchetReq); await storeCoinError(e, coinIdxs[i]);
} }
return reqBody; }
}); return {
coinIdxs,
if (!d) { batchResp: responses,
return; };
} }
const reqUrl = new URL(
`reserves/${withdrawalGroup.reservePub}/batch-withdraw`,
withdrawalGroup.exchangeBaseUrl,
).href;
const resp = await ws.http.postJson(reqUrl, d);
const r = await readSuccessResponseJsonOrThrow(
resp,
codecForWithdrawBatchResponse(),
);
return r;
} }
async function processPlanchetVerifyAndStoreCoin( async function processPlanchetVerifyAndStoreCoin(
ws: InternalWalletState, ws: InternalWalletState,
wgContext: WithdrawalGroupContext, wgContext: WithdrawalGroupContext,
coinIdx: number, coinIdx: number,
resp: WithdrawResponse, resp: ExchangeWithdrawResponse,
): Promise<void> { ): Promise<void> {
const withdrawalGroup = wgContext.wgRecord; const withdrawalGroup = wgContext.wgRecord;
logger.info(`checking and storing planchet idx=${coinIdx}`);
const d = await ws.db const d = await ws.db
.mktx((x) => [x.withdrawalGroups, x.planchets, x.denominations]) .mktx((x) => [x.withdrawalGroups, x.planchets, x.denominations])
.runReadOnly(async (tx) => { .runReadOnly(async (tx) => {
@ -791,6 +788,14 @@ async function processPlanchetVerifyAndStoreCoin(
wgContext.planchetsFinished.add(planchet.coinPub); wgContext.planchetsFinished.add(planchet.coinPub);
// We create the notification here, as the async transaction below
// allows other planchet withdrawals to change wgContext.planchetsFinished
const notification: WalletNotification = {
type: NotificationType.CoinWithdrawn,
numTotal: wgContext.numPlanchets,
numWithdrawn: wgContext.planchetsFinished.size,
}
// Check if this is the first time that the whole // Check if this is the first time that the whole
// withdrawal succeeded. If so, mark the withdrawal // withdrawal succeeded. If so, mark the withdrawal
// group as finished. // group as finished.
@ -814,11 +819,7 @@ async function processPlanchetVerifyAndStoreCoin(
}); });
if (firstSuccess) { if (firstSuccess) {
ws.notify({ ws.notify(notification);
type: NotificationType.CoinWithdrawn,
numTotal: wgContext.numPlanchets,
numWithdrawn: wgContext.planchetsFinished.size,
});
} }
} }
@ -1150,8 +1151,6 @@ export async function processWithdrawalGroup(
wgRecord: withdrawalGroup, wgRecord: withdrawalGroup,
}; };
let work: Promise<void>[] = [];
await ws.db await ws.db
.mktx((x) => [x.planchets]) .mktx((x) => [x.planchets])
.runReadOnly(async (tx) => { .runReadOnly(async (tx) => {
@ -1165,44 +1164,35 @@ export async function processWithdrawalGroup(
} }
}); });
// We sequentially generate planchets, so that
// large withdrawal groups don't make the wallet unresponsive.
for (let i = 0; i < numTotalCoins; i++) { for (let i = 0; i < numTotalCoins; i++) {
work.push(processPlanchetGenerate(ws, withdrawalGroup, i)); await processPlanchetGenerate(ws, withdrawalGroup, i);
} }
// Generate coins concurrently (parallelism only happens in the crypto API workers) const maxBatchSize = 100;
await Promise.all(work);
work = []; for (let i = 0; i < numTotalCoins; i += maxBatchSize) {
const resp = await processPlanchetExchangeBatchRequest(ws, wgContext, {
if (ws.batchWithdrawal) { batchSize: maxBatchSize,
const resp = await processPlanchetExchangeBatchRequest(ws, wgContext); coinStartIndex: i,
if (!resp) { useBatchRequest: ws.batchWithdrawal,
throw Error("unable to do batch withdrawal"); });
} let work: Promise<void>[] = [];
for (let coinIdx = 0; coinIdx < numTotalCoins; coinIdx++) { work = [];
for (let j = 0; j < resp.coinIdxs.length; j++) {
work.push( work.push(
processPlanchetVerifyAndStoreCoin( processPlanchetVerifyAndStoreCoin(
ws, ws,
wgContext, wgContext,
coinIdx, resp.coinIdxs[j],
resp.ev_sigs[coinIdx], resp.batchResp.ev_sigs[j],
), ),
); );
} }
} else { await Promise.all(work);
for (let coinIdx = 0; coinIdx < numTotalCoins; coinIdx++) {
const resp = await processPlanchetExchangeRequest(ws, wgContext, coinIdx);
if (!resp) {
continue;
}
work.push(
processPlanchetVerifyAndStoreCoin(ws, wgContext, coinIdx, resp),
);
}
} }
await Promise.all(work);
let numFinished = 0; let numFinished = 0;
let numKycRequired = 0; let numKycRequired = 0;
let finishedForFirstTime = false; let finishedForFirstTime = false;