taler/exchange
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

new aggregator mega transaction logic

Browse Source
This commit is contained in:
Christian Grothoff 2022-03-27 13:48:25 +02:00
parent d0a69da895
commit b9a9af3a59
No known key found for this signature in database
GPG Key ID: 939E6BE1E29FC3CC
3 changed files with 159 additions and 505 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

541
src/exchange/taler-exchange-aggregator.c
View File

@ -28,18 +28,6 @@
#include "taler_json_lib.h" #include "taler_json_lib.h"
#include "taler_bank_service.h" #include "taler_bank_service.h"
struct AdditionalDeposit
{
/**
* Public key of the coin.
*/
struct TALER_CoinSpendPublicKeyP coin_pub;
/**
* Row of the deposit.
*/
uint64_t row;
};
/** /**
* Information about one aggregation process to be executed. There is * Information about one aggregation process to be executed. There is
@ -54,11 +42,6 @@ struct AggregationUnit
*/ */
struct TALER_MerchantPublicKeyP merchant_pub; struct TALER_MerchantPublicKeyP merchant_pub;
/**
* Public key of the coin.
*/
struct TALER_CoinSpendPublicKeyP coin_pub;
/** /**
* Total amount to be transferred, before subtraction of @e fees.wire and rounding down. * Total amount to be transferred, before subtraction of @e fees.wire and rounding down.
*/ */
@ -79,11 +62,6 @@ struct AggregationUnit
*/ */
struct TALER_WireTransferIdentifierRawP wtid; struct TALER_WireTransferIdentifierRawP wtid;
/**
* Row ID of the transaction that started it all.
*/
uint64_t row_id;
/** /**
* The current time (which triggered the aggregation and * The current time (which triggered the aggregation and
* defines the wire fee). * defines the wire fee).
@ -100,33 +78,12 @@ struct AggregationUnit
*/ */
struct TALER_PaytoHashP h_payto; struct TALER_PaytoHashP h_payto;
/**
* Serial number of the wire target.
*/
uint64_t wire_target;
/** /**
* Exchange wire account to be used for the preparation and * Exchange wire account to be used for the preparation and
* eventual execution of the aggregate wire transfer. * eventual execution of the aggregate wire transfer.
*/ */
const struct TALER_EXCHANGEDB_AccountInfo *wa; const struct TALER_EXCHANGEDB_AccountInfo *wa;
/**
* Array of row_ids from the aggregation.
*/
struct AdditionalDeposit
additional_rows[TALER_EXCHANGEDB_MATCHING_DEPOSITS_LIMIT];
/**
* Offset specifying how many @e additional_rows are in use.
*/
unsigned int rows_offset;
/**
* Set to true if we encountered a refund during #refund_by_coin_cb.
* Used to wave the deposit fee.
*/
bool have_refund;
}; };
@ -340,331 +297,6 @@ parse_wirewatch_config (void)
} }
/**
* Callback invoked with information about refunds applicable
* to a particular coin. Subtract refunded amount(s) from
* the aggregation unit's total amount.
*
* @param cls closure with a `struct AggregationUnit *`
* @param amount_with_fee what was the refunded amount with the fee
* @return #GNUNET_OK to continue to iterate, #GNUNET_SYSERR to stop
*/
static enum GNUNET_GenericReturnValue
refund_by_coin_cb (void *cls,
const struct TALER_Amount *amount_with_fee)
{
struct AggregationUnit *aux = cls;
GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
"Aggregator subtracts applicable refund of amount %s\n",
TALER_amount2s (amount_with_fee));
aux->have_refund = true;
if (0 >
TALER_amount_subtract (&aux->total_amount,
&aux->total_amount,
amount_with_fee))
{
GNUNET_break (0);
return GNUNET_SYSERR;
}
return GNUNET_OK;
}
/**
* Function called with details about deposits that have been made,
* with the goal of executing the corresponding wire transaction.
*
* @param cls a `struct AggregationUnit`
* @param row_id identifies database entry
* @param merchant_pub public key of the merchant
* @param coin_pub public key of the coin
* @param amount_with_fee amount that was deposited including fee
* @param deposit_fee amount the exchange gets to keep as transaction fees
* @param h_contract_terms hash of the proposal data known to merchant and customer
* @param wire_target target account for the wire transfer
* @param payto_uri URI of the target account
* @return transaction status code, #GNUNET_DB_STATUS_SUCCESS_ONE_RESULT to continue to iterate
*/
static enum GNUNET_DB_QueryStatus
deposit_cb (void *cls,
uint64_t row_id,
const struct TALER_MerchantPublicKeyP *merchant_pub,
const struct TALER_CoinSpendPublicKeyP *coin_pub,
const struct TALER_Amount *amount_with_fee,
const struct TALER_Amount *deposit_fee,
const struct TALER_PrivateContractHashP *h_contract_terms,
uint64_t wire_target,
const char *payto_uri)
{
struct AggregationUnit *au = cls;
enum GNUNET_DB_QueryStatus qs;
au->merchant_pub = *merchant_pub;
au->coin_pub = *coin_pub;
GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
"Aggregator processing payment %s with amount %s\n",
TALER_B2S (coin_pub),
TALER_amount2s (amount_with_fee));
au->row_id = row_id;
au->total_amount = *amount_with_fee;
au->have_refund = false;
qs = db_plugin->select_refunds_by_coin (db_plugin->cls,
coin_pub,
&au->merchant_pub,
h_contract_terms,
&refund_by_coin_cb,
au);
if (0 > qs)
{
GNUNET_break (GNUNET_DB_STATUS_SOFT_ERROR == qs);
return qs;
}
if (! au->have_refund)
{
struct TALER_Amount ntotal;
GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
"Non-refunded transaction, subtracting deposit fee %s\n",
TALER_amount2s (deposit_fee));
if (0 >
TALER_amount_subtract (&ntotal,
amount_with_fee,
deposit_fee))
{
/* This should never happen, issue a warning, but continue processing
with an amount of zero, least we hang here for good. */
GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
"Fatally malformed record at row %llu over %s (deposit fee exceeds deposited value)\n",
(unsigned long long) row_id,
TALER_amount2s (amount_with_fee));
GNUNET_assert (GNUNET_OK ==
TALER_amount_set_zero (au->total_amount.currency,
&au->total_amount));
}
else
{
au->total_amount = ntotal;
}
}
GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
"Amount after fee is %s\n",
TALER_amount2s (&au->total_amount));
GNUNET_assert (NULL == au->payto_uri);
au->payto_uri = GNUNET_strdup (payto_uri);
TALER_payto_hash (payto_uri,
&au->h_payto);
au->wire_target = wire_target;
GNUNET_CRYPTO_random_block (GNUNET_CRYPTO_QUALITY_NONCE,
&au->wtid,
sizeof (au->wtid));
GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
"Starting aggregation under H(WTID)=%s, starting amount %s at %llu\n",
TALER_B2S (&au->wtid),
TALER_amount2s (amount_with_fee),
(unsigned long long) row_id);
au->wa = TALER_EXCHANGEDB_find_account_by_payto_uri (payto_uri);
if (NULL == au->wa)
{
GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
"No exchange account configured for `%s', please fix your setup to continue!\n",
payto_uri);
return GNUNET_DB_STATUS_HARD_ERROR;
}
/* make sure we have current fees */
au->execution_time = GNUNET_TIME_timestamp_get ();
{
struct GNUNET_TIME_Timestamp start_date;
struct GNUNET_TIME_Timestamp end_date;
struct TALER_MasterSignatureP master_sig;
enum GNUNET_DB_QueryStatus qs;
qs = db_plugin->get_wire_fee (db_plugin->cls,
au->wa->method,
au->execution_time,
&start_date,
&end_date,
&au->fees,
&master_sig);
if (0 >= qs)
{
GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
"Could not get wire fees for %s at %s. Aborting run.\n",
au->wa->method,
GNUNET_TIME_timestamp2s (au->execution_time));
return GNUNET_DB_STATUS_HARD_ERROR;
}
}
GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
"Aggregator starts aggregation for deposit %llu to %s with wire fee %s\n",
(unsigned long long) row_id,
TALER_B2S (&au->wtid),
TALER_amount2s (&au->fees.wire));
qs = db_plugin->insert_aggregation_tracking (db_plugin->cls,
&au->wtid,
row_id);
if (qs <= 0)
{
GNUNET_break (GNUNET_DB_STATUS_SOFT_ERROR == qs);
return qs;
}
GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
"Aggregator marks deposit %llu as done\n",
(unsigned long long) row_id);
qs = db_plugin->mark_deposit_done (db_plugin->cls,
coin_pub,
row_id);
if (GNUNET_DB_STATUS_SUCCESS_ONE_RESULT != qs)
{
GNUNET_break (GNUNET_DB_STATUS_SOFT_ERROR == qs);
return qs;
}
return qs;
}
/**
* Function called with details about another deposit we
* can aggregate into an existing aggregation unit.
*
* @param cls a `struct AggregationUnit`
* @param row_id identifies database entry
* @param coin_pub public key of the coin
* @param amount_with_fee amount that was deposited including fee
* @param deposit_fee amount the exchange gets to keep as transaction fees
* @param h_contract_terms hash of the proposal data known to merchant and customer
* @return transaction status code
*/
static enum GNUNET_DB_QueryStatus
aggregate_cb (void *cls,
uint64_t row_id,
const struct TALER_CoinSpendPublicKeyP *coin_pub,
const struct TALER_Amount *amount_with_fee,
const struct TALER_Amount *deposit_fee,
const struct TALER_PrivateContractHashP *h_contract_terms)
{
struct AggregationUnit *au = cls;
struct TALER_Amount old;
enum GNUNET_DB_QueryStatus qs;
if (row_id == au->row_id)
return GNUNET_DB_STATUS_SUCCESS_ONE_RESULT;
if (au->rows_offset >= TALER_EXCHANGEDB_MATCHING_DEPOSITS_LIMIT)
{
/* Bug: we asked for at most #TALER_EXCHANGEDB_MATCHING_DEPOSITS_LIMIT results! */
GNUNET_break (0);
/* Skip this one, but keep going with the overall transaction */
return GNUNET_DB_STATUS_SUCCESS_ONE_RESULT;
}
/* add to total */
GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
"Adding transaction amount %s from row %llu to aggregation\n",
TALER_amount2s (amount_with_fee),
(unsigned long long) row_id);
/* save the existing total aggregate in 'old', for later */
old = au->total_amount;
/* we begin with the total contribution of the current coin */
au->total_amount = *amount_with_fee;
/* compute contribution of this coin (after fees) */
au->have_refund = false;
qs = db_plugin->select_refunds_by_coin (db_plugin->cls,
coin_pub,
&au->merchant_pub,
h_contract_terms,
&refund_by_coin_cb,
au);
if (0 > qs)
{
GNUNET_break (GNUNET_DB_STATUS_SOFT_ERROR == qs);
return qs;
}
if (! au->have_refund)
{
struct TALER_Amount tmp;
GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
"Subtracting deposit fee %s for non-refunded coin\n",
TALER_amount2s (deposit_fee));
if (0 >
TALER_amount_subtract (&tmp,
&au->total_amount,
deposit_fee))
{
GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
"Fatally malformed record at %llu over amount %s (deposit fee exceeds deposited value)\n",
(unsigned long long) row_id,
TALER_amount2s (&au->total_amount));
GNUNET_assert (GNUNET_OK ==
TALER_amount_set_zero (old.currency,
&au->total_amount));
}
else
{
au->total_amount = tmp;
}
}
/* now add the au->total_amount with the (remaining) contribution of
the current coin to the 'old' value with the current aggregate value */
{
struct TALER_Amount tmp;
if (0 >
TALER_amount_add (&tmp,
&au->total_amount,
&old))
{
GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
"Overflow or currency incompatibility during aggregation at %llu\n",
(unsigned long long) row_id);
/* Skip this one, but keep going! */
au->total_amount = old;
return GNUNET_DB_STATUS_SUCCESS_ONE_RESULT;
}
au->total_amount = tmp;
}
/* "append" to our list of rows */
au->additional_rows[au->rows_offset].coin_pub = *coin_pub;
au->additional_rows[au->rows_offset].row = row_id;
au->rows_offset++;
/* insert into aggregation tracking table */
GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
"Adding %llu to aggregate %s\n",
(unsigned long long) row_id,
TALER_B2S (&au->wtid));
qs = db_plugin->insert_aggregation_tracking (db_plugin->cls,
&au->wtid,
row_id);
if (GNUNET_DB_STATUS_SUCCESS_ONE_RESULT != qs)
{
GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
"Failed to add %llu to aggregate %s: %d\n",
(unsigned long long) row_id,
TALER_B2S (&au->wtid),
qs);
GNUNET_break (GNUNET_DB_STATUS_SOFT_ERROR == qs);
return qs;
}
qs = db_plugin->mark_deposit_done (db_plugin->cls,
coin_pub,
row_id);
if (GNUNET_DB_STATUS_SUCCESS_ONE_RESULT != qs)
{
GNUNET_break (GNUNET_DB_STATUS_SOFT_ERROR == qs);
return qs;
}
GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
"Aggregator marked deposit %llu as DONE\n",
(unsigned long long) row_id);
return GNUNET_DB_STATUS_SUCCESS_ONE_RESULT;
}
/** /**
* Perform a database commit. If it fails, print a warning. * Perform a database commit. If it fails, print a warning.
* *
@ -727,10 +359,17 @@ run_aggregation (void *cls)
struct Shard *s = cls; struct Shard *s = cls;
struct AggregationUnit au_active; struct AggregationUnit au_active;
enum GNUNET_DB_QueryStatus qs; enum GNUNET_DB_QueryStatus qs;
struct TALER_Amount trans;
bool have_transient;
task = NULL; task = NULL;
GNUNET_log (GNUNET_ERROR_TYPE_INFO, GNUNET_log (GNUNET_ERROR_TYPE_INFO,
"Checking for ready deposits to aggregate\n"); "Checking for ready deposits to aggregate\n");
/* make sure we have current fees */
memset (&au_active,
0,
sizeof (au_active));
au_active.execution_time = GNUNET_TIME_timestamp_get ();
if (GNUNET_OK != if (GNUNET_OK !=
db_plugin->start_deferred_wire_out (db_plugin->cls)) db_plugin->start_deferred_wire_out (db_plugin->cls))
{ {
@ -741,16 +380,13 @@ run_aggregation (void *cls)
release_shard (s); release_shard (s);
return; return;
} }
memset (&au_active,
0,
sizeof (au_active));
qs = db_plugin->get_ready_deposit ( qs = db_plugin->get_ready_deposit (
db_plugin->cls, db_plugin->cls,
s->shard_start, s->shard_start,
s->shard_end, s->shard_end,
kyc_off ? true : false, kyc_off ? true : false,
&deposit_cb, &au_active.merchant_pub,
&au_active); &au_active.payto_uri);
switch (qs) switch (qs)
{ {
case GNUNET_DB_STATUS_HARD_ERROR: case GNUNET_DB_STATUS_HARD_ERROR:
@ -808,22 +444,98 @@ run_aggregation (void *cls)
/* continued below */ /* continued below */
break; break;
} }
au_active.wa = TALER_EXCHANGEDB_find_account_by_payto_uri (
au_active.payto_uri);
if (NULL == au_active.wa)
{
GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
"No exchange account configured for `%s', please fix your setup to continue!\n",
au_active.payto_uri);
global_ret = EXIT_FAILURE;
GNUNET_SCHEDULER_shutdown ();
release_shard (s);
return;
}
{
struct GNUNET_TIME_Timestamp start_date;
struct GNUNET_TIME_Timestamp end_date;
struct TALER_MasterSignatureP master_sig;
qs = db_plugin->get_wire_fee (db_plugin->cls,
au_active.wa->method,
au_active.execution_time,
&start_date,
&end_date,
&au_active.fees,
&master_sig);
if (0 >= qs)
{
GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
"Could not get wire fees for %s at %s. Aborting run.\n",
au_active.wa->method,
GNUNET_TIME_timestamp2s (au_active.execution_time));
global_ret = EXIT_FAILURE;
GNUNET_SCHEDULER_shutdown ();
release_shard (s);
return;
}
}
/* Now try to find other deposits to aggregate */ /* Now try to find other deposits to aggregate */
GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
"Found ready deposit for %s, aggregating by target %llu\n", "Found ready deposit for %s, aggregating by target %s\n",
TALER_B2S (&au_active.merchant_pub), TALER_B2S (&au_active.merchant_pub),
(unsigned long long) au_active.wire_target); au_active.payto_uri);
qs = db_plugin->iterate_matching_deposits (db_plugin->cls, TALER_payto_hash (au_active.payto_uri,
&au_active.h_payto);
qs = db_plugin->select_aggregation_transient (db_plugin->cls,
&au_active.h_payto,
au_active.wa->section_name,
&au_active.wtid,
&trans);
switch (qs)
{
case GNUNET_DB_STATUS_HARD_ERROR:
GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
"Failed to lookup transient aggregates!\n");
cleanup_au (&au_active);
db_plugin->rollback (db_plugin->cls);
global_ret = EXIT_FAILURE;
GNUNET_SCHEDULER_shutdown ();
release_shard (s);
return;
case GNUNET_DB_STATUS_SOFT_ERROR:
/* serializiability issue, try again */
GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
"Serialization issue, trying again later!\n");
db_plugin->rollback (db_plugin->cls);
cleanup_au (&au_active);
GNUNET_assert (NULL == task);
task = GNUNET_SCHEDULER_add_now (&run_aggregation,
s);
return;
case GNUNET_DB_STATUS_SUCCESS_NO_RESULTS:
GNUNET_CRYPTO_random_block (GNUNET_CRYPTO_QUALITY_NONCE,
&au_active.wtid,
sizeof (au_active.wtid));
have_transient = false;
break;
case GNUNET_DB_STATUS_SUCCESS_ONE_RESULT:
have_transient = true;
break;
}
qs = db_plugin->aggregate (db_plugin->cls,
&au_active.h_payto, &au_active.h_payto,
&au_active.merchant_pub, &au_active.merchant_pub,
&aggregate_cb, &au_active.wtid,
&au_active, &au_active.total_amount);
TALER_EXCHANGEDB_MATCHING_DEPOSITS_LIMIT);
if (GNUNET_DB_STATUS_HARD_ERROR == qs) if (GNUNET_DB_STATUS_HARD_ERROR == qs)
{ {
GNUNET_log (GNUNET_ERROR_TYPE_ERROR, GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
"Failed to execute deposit iteration!\n"); "Failed to execute aggregation!\n");
cleanup_au (&au_active); cleanup_au (&au_active);
db_plugin->rollback (db_plugin->cls); db_plugin->rollback (db_plugin->cls);
global_ret = EXIT_FAILURE; global_ret = EXIT_FAILURE;
@ -844,13 +556,17 @@ run_aggregation (void *cls)
return; return;
} }
GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
"Found %d other deposits to combine into wire transfer with fee %s.\n", "Aggregation total is %s.\n",
qs, TALER_amount2s (&au_active.total_amount));
TALER_amount2s (&au_active.fees.wire));
/* Subtract wire transfer fee and round to the unit supported by the /* Subtract wire transfer fee and round to the unit supported by the
wire transfer method; Check if after rounding down, we still have wire transfer method; Check if after rounding down, we still have
an amount to transfer, and if not mark as 'tiny'. */ an amount to transfer, and if not mark as 'tiny'. */
if (have_transient)
GNUNET_assert (0 <=
TALER_amount_add (&au_active.total_amount,
&au_active.total_amount,
&trans));
GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
"Rounding aggregate of %s\n", "Rounding aggregate of %s\n",
TALER_amount2s (&au_active.total_amount)); TALER_amount2s (&au_active.total_amount));
@ -867,45 +583,17 @@ run_aggregation (void *cls)
"Aggregate value too low for transfer (%d/%s)\n", "Aggregate value too low for transfer (%d/%s)\n",
qs, qs,
TALER_amount2s (&au_active.final_amount)); TALER_amount2s (&au_active.final_amount));
/* Rollback ongoing transaction, as we will not use the respective if (have_transient)
WTID and thus need to remove the tracking data */ qs = db_plugin->update_aggregation_transient (db_plugin->cls,
db_plugin->rollback (db_plugin->cls); &au_active.h_payto,
&au_active.wtid,
/* There were results, just the value was too low. Start another &au_active.total_amount);
transaction to mark all* of the selected deposits as minor! */ else
if (GNUNET_OK != qs = db_plugin->create_aggregation_transient (db_plugin->cls,
db_plugin->start (db_plugin->cls, &au_active.h_payto,
"aggregator mark tiny transactions")) au_active.wa->section_name,
{ &au_active.wtid,
GNUNET_log (GNUNET_ERROR_TYPE_ERROR, &au_active.total_amount);
"Failed to start database transaction!\n");
global_ret = EXIT_FAILURE;
cleanup_au (&au_active);
GNUNET_SCHEDULER_shutdown ();
release_shard (s);
return;
}
/* Mark transactions by row_id as minor */
GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
"Marking %s (%llu) as tiny\n",
TALER_B2S (&au_active.coin_pub),
(unsigned long long) au_active.row_id);
qs = db_plugin->mark_deposit_tiny (db_plugin->cls,
&au_active.coin_pub,
au_active.row_id);
if (0 < qs)
{
for (unsigned int i = 0; i<au_active.rows_offset; i++)
{
qs = db_plugin->mark_deposit_tiny (db_plugin->cls,
&au_active.additional_rows[i].
coin_pub,
au_active.additional_rows[i].row);
if (0 >= qs)
break;
}
}
GNUNET_break (GNUNET_DB_STATUS_SUCCESS_NO_RESULTS != qs);
if (GNUNET_DB_STATUS_SOFT_ERROR == qs) if (GNUNET_DB_STATUS_SOFT_ERROR == qs)
{ {
GNUNET_log (GNUNET_ERROR_TYPE_INFO, GNUNET_log (GNUNET_ERROR_TYPE_INFO,
@ -962,8 +650,7 @@ run_aggregation (void *cls)
buf_size); buf_size);
GNUNET_free (buf); GNUNET_free (buf);
} }
/* Commit the WTID data to 'wire_out' to finally satisfy aggregation /* Commit the WTID data to 'wire_out' */
table constraints */
if (qs >= 0) if (qs >= 0)
qs = db_plugin->store_wire_transfer_out (db_plugin->cls, qs = db_plugin->store_wire_transfer_out (db_plugin->cls,
au_active.execution_time, au_active.execution_time,
@ -971,6 +658,12 @@ run_aggregation (void *cls)
&au_active.h_payto, &au_active.h_payto,
au_active.wa->section_name, au_active.wa->section_name,
&au_active.final_amount); &au_active.final_amount);
if ( (qs >= 0) &&
have_transient)
qs = db_plugin->delete_aggregation_transient (db_plugin->cls,
&au_active.h_payto,
&au_active.wtid);
cleanup_au (&au_active); cleanup_au (&au_active);
if (GNUNET_DB_STATUS_SOFT_ERROR == qs) if (GNUNET_DB_STATUS_SOFT_ERROR == qs)

105
src/exchangedb/plugin_exchangedb_postgres.c
View File

@ -1160,29 +1160,17 @@ prepare_statements (struct PostgresClosure *pg)
GNUNET_PQ_make_prepare ( GNUNET_PQ_make_prepare (
"deposits_get_ready", "deposits_get_ready",
"SELECT" "SELECT"
" dep.deposit_serial_id" " payto_uri"
",amount_with_fee_val"
",amount_with_fee_frac"
",denom.fee_deposit_val"
",denom.fee_deposit_frac"
",h_contract_terms"
",payto_uri"
",wire_target_serial_id"
",merchant_pub" ",merchant_pub"
",kc.coin_pub"
" FROM deposits_by_ready dbr" " FROM deposits_by_ready dbr"
" JOIN deposits dep" " JOIN deposits dep"
" ON (dbr.coin_pub = dep.coin_pub AND dbr.deposit_serial_id = dep.deposit_serial_id)" " ON (dbr.coin_pub = dep.coin_pub AND dbr.deposit_serial_id = dep.deposit_serial_id)"
" JOIN wire_targets " " JOIN wire_targets wt"
" USING (wire_target_h_payto)" " USING (wire_target_h_payto)"
" JOIN known_coins kc"
" ON (kc.coin_pub = dep.coin_pub)"
" JOIN denominations denom"
" USING (denominations_serial)"
" WHERE dbr.wire_deadline<=$1" " WHERE dbr.wire_deadline<=$1"
" AND dbr.shard >= $2" " AND dbr.shard >= $2"
" AND dbr.shard <= $3" " AND dbr.shard <= $3"
" AND (kyc_ok OR $4)" " AND (wt.kyc_ok OR $4)"
" ORDER BY " " ORDER BY "
" dbr.wire_deadline ASC" " dbr.wire_deadline ASC"
" ,dbr.shard ASC" " ,dbr.shard ASC"
@ -1218,22 +1206,23 @@ prepare_statements (struct PostgresClosure *pg)
/* Used in #postgres_aggregate() */ /* Used in #postgres_aggregate() */
GNUNET_PQ_make_prepare ( GNUNET_PQ_make_prepare (
"aggregate", "aggregate",
"WITH rdy AS (" /* find deposits ready */ "WITH rdy AS (" /* find deposits ready by merchant */
" SELECT" " SELECT"
" coin_pub" " coin_pub"
" FROM deposits_for_matching" " FROM deposits_for_matching"
" WHERE refund_deadline<$1" " WHERE refund_deadline<$1" /* filter by shard, only actually executable deposits */
" AND merchant_pub=$2" " AND merchant_pub=$2" /* filter by target merchant */
" ORDER BY refund_deadline ASC" /* ordering is not critical */ " ORDER BY refund_deadline ASC" /* ordering is not critical */
" LIMIT " " LIMIT "
TALER_QUOTE (TALER_EXCHANGEDB_MATCHING_DEPOSITS_LIMIT) TALER_QUOTE (TALER_EXCHANGEDB_MATCHING_DEPOSITS_LIMIT) /* limits transaction size */
" )" " )"
" ,dep AS (" /* restrict to our merchant and account */ " ,dep AS (" /* restrict to our merchant and account and mark as done */
" UPDATE deposits" " UPDATE deposits"
" SET done=TRUE" " SET done=TRUE"
" WHERE coin_pub IN (SELECT coin_pub FROM rdy)" " WHERE coin_pub IN (SELECT coin_pub FROM rdy)"
" AND merchant_pub=$2" " AND merchant_pub=$2" /* theoretically, same coin could be spent at another merchant */
" AND wire_target_h_payto=$3" " AND wire_target_h_payto=$3" /* merchant could have a 2nd bank account */
" AND done=FALSE" /* theoretically, same coin could be spend at the same merchant a 2nd time */
" RETURNING" " RETURNING"
" deposit_serial_id" " deposit_serial_id"
" ,coin_pub" " ,coin_pub"
@ -1244,18 +1233,26 @@ prepare_statements (struct PostgresClosure *pg)
" amount_with_fee_val AS refund_val" " amount_with_fee_val AS refund_val"
" ,amount_with_fee_frac AS refund_frac" " ,amount_with_fee_frac AS refund_frac"
" ,coin_pub" " ,coin_pub"
" ,deposit_serial_id" /* theoretically, coin could be in multiple refunded transactions */
" FROM refunds" " FROM refunds"
" WHERE coin_pub IN (SELECT coin_pub FROM dep)" " WHERE coin_pub IN (SELECT coin_pub FROM dep)"
" AND deposit_serial_id IN (SELECT deposit_serial_id FROM dep))" " AND deposit_serial_id IN (SELECT deposit_serial_id FROM dep))"
" ,coins_with_fees AS (" /* find coins for which deposit fees apply */
" SELECT"
" coin_pub"
" ,deposit_serial_id" /* ensures that if the same coin is deposited twice, it is in the list twice */
" FROM dep"
" WHERE deposit_serial_id NOT IN (SELECT deposit_serial_id FROM ref))"
" ,fees AS (" /* find deposit fees for non-refunded deposits */ " ,fees AS (" /* find deposit fees for non-refunded deposits */
" SELECT" " SELECT"
" denom.fee_deposit_val AS fee_val" " denom.fee_deposit_val AS fee_val"
" ,denom.fee_deposit_frac AS fee_frac" " ,denom.fee_deposit_frac AS fee_frac"
" FROM known_coins kc" " ,cs.deposit_serial_id" /* ensures we get the fee for each coin, not once per denomination */
" FROM coins_with_fees cs"
" JOIN known_coins kc"
" USING (coin_pub)"
" JOIN denominations denom" " JOIN denominations denom"
" USING (denominations_serial)" " USING (denominations_serial))"
" WHERE coin_pub IN (SELECT coin_pub FROM dep)"
" AND coin_pub NOT IN (SELECT coin_pub FROM ref))"
" ,dummy AS (" /* add deposits to aggregation_tracking */ " ,dummy AS (" /* add deposits to aggregation_tracking */
" INSERT INTO aggregation_tracking" " INSERT INTO aggregation_tracking"
" (deposit_serial_id" " (deposit_serial_id"
@ -1263,14 +1260,14 @@ prepare_statements (struct PostgresClosure *pg)
" SELECT deposit_serial_id,$4" " SELECT deposit_serial_id,$4"
" FROM dep)" " FROM dep)"
"SELECT" /* calculate totals (deposits, refunds and fees) */ "SELECT" /* calculate totals (deposits, refunds and fees) */
" CAST(COALESCE(SUM(dep.amount_val),0) AS INT8) AS sum_deposit_value" " CAST(COALESCE(SUM(dep.amount_val),0) AS INT8) AS sum_deposit_value" /* cast needed, otherwise we get NUMBER */
" ,COALESCE(SUM(dep.amount_frac),0) AS sum_deposit_fraction" " ,COALESCE(SUM(dep.amount_frac),0) AS sum_deposit_fraction" /* SUM over INT returns INT8 */
" ,CAST(COALESCE(SUM(ref.refund_val),0) AS INT8) AS sum_refund_value" " ,CAST(COALESCE(SUM(ref.refund_val),0) AS INT8) AS sum_refund_value"
" ,COALESCE(SUM(ref.refund_frac),0) AS sum_refund_fraction" " ,COALESCE(SUM(ref.refund_frac),0) AS sum_refund_fraction"
" ,CAST(COALESCE(SUM(fees.fee_val),0) AS INT8) AS sum_fee_value" " ,CAST(COALESCE(SUM(fees.fee_val),0) AS INT8) AS sum_fee_value"
" ,COALESCE(SUM(fees.fee_frac),0) AS sum_fee_fraction" " ,COALESCE(SUM(fees.fee_frac),0) AS sum_fee_fraction"
" FROM dep " " FROM dep "
" FULL OUTER JOIN ref ON (FALSE)" " FULL OUTER JOIN ref ON (FALSE)" /* We just want all sums */
" FULL OUTER JOIN fees ON (FALSE);", " FULL OUTER JOIN fees ON (FALSE);",
4), 4),
@ -6270,8 +6267,8 @@ postgres_mark_deposit_done (void *cls,
* @param end_shard_row maximum shard row to select (inclusive) * @param end_shard_row maximum shard row to select (inclusive)
* @param kyc_off true if we should not check the KYC status because * @param kyc_off true if we should not check the KYC status because
* this exchange does not need/support KYC checks. * this exchange does not need/support KYC checks.
* @param deposit_cb function to call for ONE such deposit * @param[out] merchant_pub set to the public key of a merchant with a ready deposit
* @param deposit_cb_cls closure for @a deposit_cb * @param[out] payto_uri set to the account of the merchant, to be freed by caller
* @return transaction status code * @return transaction status code
*/ */
static enum GNUNET_DB_QueryStatus static enum GNUNET_DB_QueryStatus
@ -6279,8 +6276,8 @@ postgres_get_ready_deposit (void *cls,
uint64_t start_shard_row, uint64_t start_shard_row,
uint64_t end_shard_row, uint64_t end_shard_row,
bool kyc_off, bool kyc_off,
TALER_EXCHANGEDB_DepositIterator deposit_cb, struct TALER_MerchantPublicKeyP *merchant_pub,
void *deposit_cb_cls) char **payto_uri)
{ {
struct PostgresClosure *pg = cls; struct PostgresClosure *pg = cls;
struct GNUNET_TIME_Absolute now = {0}; struct GNUNET_TIME_Absolute now = {0};
@ -6291,34 +6288,13 @@ postgres_get_ready_deposit (void *cls,
GNUNET_PQ_query_param_bool (kyc_off), GNUNET_PQ_query_param_bool (kyc_off),
GNUNET_PQ_query_param_end GNUNET_PQ_query_param_end
}; };
struct TALER_Amount amount_with_fee;
struct TALER_Amount deposit_fee;
struct TALER_PrivateContractHashP h_contract_terms;
struct TALER_MerchantPublicKeyP merchant_pub;
struct TALER_CoinSpendPublicKeyP coin_pub;
uint64_t serial_id;
uint64_t wire_target;
char *payto_uri;
struct GNUNET_PQ_ResultSpec rs[] = { struct GNUNET_PQ_ResultSpec rs[] = {
GNUNET_PQ_result_spec_uint64 ("deposit_serial_id",
&serial_id),
GNUNET_PQ_result_spec_uint64 ("wire_target_serial_id",
&wire_target),
TALER_PQ_RESULT_SPEC_AMOUNT ("amount_with_fee",
&amount_with_fee),
TALER_PQ_RESULT_SPEC_AMOUNT ("fee_deposit",
&deposit_fee),
GNUNET_PQ_result_spec_auto_from_type ("h_contract_terms",
&h_contract_terms),
GNUNET_PQ_result_spec_auto_from_type ("merchant_pub", GNUNET_PQ_result_spec_auto_from_type ("merchant_pub",
&merchant_pub), merchant_pub),
GNUNET_PQ_result_spec_auto_from_type ("coin_pub",
&coin_pub),
GNUNET_PQ_result_spec_string ("payto_uri", GNUNET_PQ_result_spec_string ("payto_uri",
&payto_uri), payto_uri),
GNUNET_PQ_result_spec_end GNUNET_PQ_result_spec_end
}; };
enum GNUNET_DB_QueryStatus qs;
now = GNUNET_TIME_absolute_round_down (GNUNET_TIME_absolute_get (), now = GNUNET_TIME_absolute_round_down (GNUNET_TIME_absolute_get (),
pg->aggregator_shift); pg->aggregator_shift);
@ -6328,25 +6304,10 @@ postgres_get_ready_deposit (void *cls,
"Finding ready deposits by deadline %s (%llu)\n", "Finding ready deposits by deadline %s (%llu)\n",
GNUNET_TIME_absolute2s (now), GNUNET_TIME_absolute2s (now),
(unsigned long long) now.abs_value_us); (unsigned long long) now.abs_value_us);
return GNUNET_PQ_eval_prepared_singleton_select (pg->conn,
qs = GNUNET_PQ_eval_prepared_singleton_select (pg->conn,
"deposits_get_ready", "deposits_get_ready",
params, params,
rs); rs);
if (qs <= 0)
return qs;
qs = deposit_cb (deposit_cb_cls,
serial_id,
&merchant_pub,
&coin_pub,
&amount_with_fee,
&deposit_fee,
&h_contract_terms,
wire_target,
payto_uri);
GNUNET_PQ_cleanup_result (rs);
return qs;
} }

8
src/include/taler_exchangedb_plugin.h
View File

@ -3097,8 +3097,8 @@ struct TALER_EXCHANGEDB_Plugin
* @param end_shard_row maximum shard row to select (inclusive) * @param end_shard_row maximum shard row to select (inclusive)
* @param kyc_off true if we should not check the KYC status because * @param kyc_off true if we should not check the KYC status because
* this exchange does not need/support KYC checks. * this exchange does not need/support KYC checks.
* @param deposit_cb function to call for ONE such deposit * @param[out] merchant_pub set to the public key of a merchant with a ready deposit
* @param deposit_cb_cls closure for @a deposit_cb * @param[out] payto_uri set to the account of the merchant, to be freed by caller
* @return transaction status code * @return transaction status code
*/ */
enum GNUNET_DB_QueryStatus enum GNUNET_DB_QueryStatus
@ -3106,8 +3106,8 @@ struct TALER_EXCHANGEDB_Plugin
uint64_t start_shard_row, uint64_t start_shard_row,
uint64_t end_shard_row, uint64_t end_shard_row,
bool kyc_off, bool kyc_off,
TALER_EXCHANGEDB_DepositIterator deposit_cb, struct TALER_MerchantPublicKeyP *merchant_pub,
void *deposit_cb_cls); char **payto_uri);
/** /**