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

working on #5010 for aggregator

Browse Source
This commit is contained in:
Christian Grothoff 2017-06-24 00:41:41 +02:00
parent 6acb0271d6
commit 9aa2e94812
No known key found for this signature in database
GPG Key ID: 939E6BE1E29FC3CC
4 changed files with 210 additions and 151 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -812,6 +812,8 @@ prepare_close_cb (void *cls,
const char *buf,
size_t buf_size)
{
enum GNUNET_DB_QueryStatus qs;
GNUNET_assert (cls == ctc);
GNUNET_log (GNUNET_ERROR_TYPE_INFO,
@ -832,14 +834,25 @@ prepare_close_cb (void *cls,
}
/* Commit our intention to execute the wire transfer! */
if (GNUNET_OK !=
db_plugin->wire_prepare_data_insert (db_plugin->cls,
ctc->session,
ctc->type,
buf,
buf_size))
qs = db_plugin->wire_prepare_data_insert (db_plugin->cls,
ctc->session,
ctc->type,
buf,
buf_size);
if (GNUNET_DB_STATUS_HARD_ERROR == qs)
{
GNUNET_break (0);
db_plugin->rollback (db_plugin->cls,
ctc->session);
global_ret = GNUNET_SYSERR;
GNUNET_SCHEDULER_shutdown ();
GNUNET_free (ctc->type);
GNUNET_free (ctc);
ctc = NULL;
return;
}
if (GNUNET_DB_STATUS_SOFT_ERROR == qs)
{
GNUNET_break (0); /* why? how to best recover? */
db_plugin->rollback (db_plugin->cls,
ctc->session);
/* start again */
@ -863,33 +876,53 @@ prepare_close_cb (void *cls,
}
/**
* Closure for #expired_reserve_cb().
*/
struct ExpiredReserveContext
{
/**
* Database session we are using.
*/
struct TALER_EXCHANGEDB_Session *session;
/**
* Set to #GNUNET_YES if the transaction continues
* asynchronously.
*/
int async_cont;
};
/**
* Function called with details about expired reserves.
* We trigger the reserve closure by inserting the respective
* closing record and prewire instructions into the respective
* tables.
*
* @param cls a `struct TALER_EXCHANGEDB_Session *`
* @param cls a `struct ExpiredReserveContext *`
* @param reserve_pub public key of the reserve
* @param left amount left in the reserve
* @param account_details information about the reserve's bank account
* @param expiration_date when did the reserve expire
* @return #GNUNET_OK to continue to iterate, #GNUNET_SYSERR to stop
* @return transaction status code
*/
static int
static enum GNUNET_DB_QueryStatus
expired_reserve_cb (void *cls,
const struct TALER_ReservePublicKeyP *reserve_pub,
const struct TALER_Amount *left,
const json_t *account_details,
struct GNUNET_TIME_Absolute expiration_date)
{
struct TALER_EXCHANGEDB_Session *session = cls;
struct ExpiredReserveContext *erc = cls;
struct TALER_EXCHANGEDB_Session *session = erc->session;
struct GNUNET_TIME_Absolute now;
struct TALER_WireTransferIdentifierRawP wtid;
struct TALER_Amount amount_without_fee;
const struct TALER_Amount *closing_fee;
int ret;
int iret;
enum GNUNET_DB_QueryStatus qs;
const char *type;
struct WirePlugin *wp;
@ -901,21 +934,17 @@ expired_reserve_cb (void *cls,
if (NULL == type)
{
GNUNET_break (0);
db_plugin->rollback (db_plugin->cls,
session);
global_ret = GNUNET_SYSERR;
GNUNET_SCHEDULER_shutdown ();
return GNUNET_SYSERR;
return GNUNET_DB_STATUS_HARD_ERROR;
}
wp = find_plugin (type);
if (NULL == wp)
{
GNUNET_break (0);
db_plugin->rollback (db_plugin->cls,
session);
global_ret = GNUNET_SYSERR;
GNUNET_SCHEDULER_shutdown ();
return GNUNET_SYSERR;
return GNUNET_DB_STATUS_HARD_ERROR;
}
/* lookup `closing_fee` */
@ -925,11 +954,9 @@ expired_reserve_cb (void *cls,
session))
{
GNUNET_break (0);
db_plugin->rollback (db_plugin->cls,
session);
global_ret = GNUNET_SYSERR;
GNUNET_SCHEDULER_shutdown ();
return GNUNET_SYSERR;
return GNUNET_DB_STATUS_HARD_ERROR;
}
closing_fee = &wp->af->closing_fee;
@ -956,22 +983,22 @@ expired_reserve_cb (void *cls,
reserve_pub,
GNUNET_MIN (sizeof (wtid),
sizeof (*reserve_pub)));
iret = db_plugin->insert_reserve_closed (db_plugin->cls,
session,
reserve_pub,
now,
account_details,
&wtid,
left,
closing_fee);
qs = db_plugin->insert_reserve_closed (db_plugin->cls,
session,
reserve_pub,
now,
account_details,
&wtid,
left,
closing_fee);
GNUNET_log (GNUNET_ERROR_TYPE_INFO,
"Closing reserve %s over %s (%d, %d)\n",
TALER_B2S (reserve_pub),
TALER_amount2s (left),
ret,
iret);
qs);
if ( (GNUNET_OK == ret) &&
(GNUNET_OK == iret) )
(GNUNET_DB_STATUS_SUCCESS_ONE_RESULT == qs) )
{
/* success, perform wire transfer */
if (GNUNET_SYSERR ==
@ -979,11 +1006,9 @@ expired_reserve_cb (void *cls,
&amount_without_fee))
{
GNUNET_break (0);
db_plugin->rollback (db_plugin->cls,
session);
global_ret = GNUNET_SYSERR;
GNUNET_SCHEDULER_shutdown ();
return GNUNET_SYSERR;
return GNUNET_DB_STATUS_HARD_ERROR;
}
ctc = GNUNET_new (struct CloseTransferContext);
ctc->wp = wp;
@ -1000,33 +1025,29 @@ expired_reserve_cb (void *cls,
if (NULL == ctc->ph)
{
GNUNET_break (0);
db_plugin->rollback (db_plugin->cls,
session);
global_ret = GNUNET_SYSERR;
GNUNET_SCHEDULER_shutdown ();
GNUNET_free (ctc->type);
GNUNET_free (ctc);
ctc = NULL;
return GNUNET_DB_STATUS_HARD_ERROR;
}
return GNUNET_SYSERR;
erc->async_cont = GNUNET_YES;
return GNUNET_DB_STATUS_SUCCESS_ONE_RESULT;
}
/* Check for hard failure */
if (GNUNET_SYSERR == iret)
if ( (GNUNET_SYSERR == ret) ||
(GNUNET_DB_STATUS_HARD_ERROR == qs) )
{
GNUNET_break (0);
db_plugin->rollback (db_plugin->cls,
session);
global_ret = GNUNET_SYSERR;
GNUNET_SCHEDULER_shutdown ();
return GNUNET_SYSERR;
return GNUNET_DB_STATUS_HARD_ERROR;
}
/* Reserve balance was almost zero; just commit */
/* Reserve balance was almost zero OR soft error */
GNUNET_log (GNUNET_ERROR_TYPE_INFO,
"Reserve was virtually empty, moving on\n");
(void) commit_or_warn (session);
task = GNUNET_SCHEDULER_add_now (&run_reserve_closures,
NULL);
return GNUNET_SYSERR;
return qs;
}
@ -1040,9 +1061,10 @@ static void
run_reserve_closures (void *cls)
{
struct TALER_EXCHANGEDB_Session *session;
int ret;
enum GNUNET_DB_QueryStatus qs;
const struct GNUNET_SCHEDULER_TaskContext *tc;
struct ExpiredReserveContext erc;
task = NULL;
reserves_idle = GNUNET_NO;
tc = GNUNET_SCHEDULER_get_task_context ();
@ -1068,22 +1090,29 @@ run_reserve_closures (void *cls)
GNUNET_SCHEDULER_shutdown ();
return;
}
ret = db_plugin->get_expired_reserves (db_plugin->cls,
session,
GNUNET_TIME_absolute_get (),
&expired_reserve_cb,
session);
if (GNUNET_SYSERR == ret)
erc.session = session;
erc.async_cont = GNUNET_NO;
qs = db_plugin->get_expired_reserves (db_plugin->cls,
session,
GNUNET_TIME_absolute_get (),
&expired_reserve_cb,
&erc);
switch (qs)
{
case GNUNET_DB_STATUS_HARD_ERROR:
GNUNET_break (0);
db_plugin->rollback (db_plugin->cls,
session);
global_ret = GNUNET_SYSERR;
GNUNET_SCHEDULER_shutdown ();
return;
}
if (GNUNET_NO == ret)
{
case GNUNET_DB_STATUS_SOFT_ERROR:
db_plugin->rollback (db_plugin->cls,
session);
task = GNUNET_SCHEDULER_add_now (&run_reserve_closures,
NULL);
return;
case GNUNET_DB_STATUS_SUCCESS_NO_RESULTS:
GNUNET_log (GNUNET_ERROR_TYPE_INFO,
"No more idle reserves, going back to aggregation\n");
reserves_idle = GNUNET_YES;
@ -1092,6 +1121,13 @@ run_reserve_closures (void *cls)
task = GNUNET_SCHEDULER_add_now (&run_aggregation,
NULL);
return;
case GNUNET_DB_STATUS_SUCCESS_ONE_RESULT:
if (GNUNET_YES == erc.async_cont)
break;
(void) commit_or_warn (session);
task = GNUNET_SCHEDULER_add_now (&run_reserve_closures,
NULL);
return;
}
}

188
src/exchangedb/plugin_exchangedb_postgres.c
View File

@ -1454,7 +1454,9 @@ postgres_prepare (PGconn *db_conn)
" FROM reserves"
" WHERE expiration_date<=$1"
" AND (current_balance_val != 0 "
" OR current_balance_frac != 0);",
" OR current_balance_frac != 0)"
" ORDER BY expiration_date ASC"
" LIMIT 1;",
1),
/* Used in #postgres_get_coin_transactions() to obtain payback transactions
for a coin */
@ -2884,8 +2886,8 @@ postgres_get_ready_deposit (void *cls,
* @param deposit_cb function to call for each deposit
* @param deposit_cb_cls closure for @a deposit_cb
* @param limit maximum number of matching deposits to return
* @return number of rows processed, 0 if none exist,
* #GNUNET_SYSERR on error
* @return transaction status code, if positive:
* number of rows processed, 0 if none exist
*/
static int
postgres_iterate_matching_deposits (void *cls,
@ -4633,57 +4635,50 @@ postgres_insert_wire_fee (void *cls,
/**
* Obtain information about expired reserves and their
* remaining balances.
*
* @param cls closure of the plugin
* @param session database connection
* @param now timestamp based on which we decide expiration
* @param rec function to call on expired reserves
* @param rec_cls closure for @a rec
* @return #GNUNET_SYSERR on database error
* #GNUNET_NO if there are no expired non-empty reserves
* #GNUNET_OK on success
* Closure for #reserve_expired_cb().
*/
static int
postgres_get_expired_reserves (void *cls,
struct TALER_EXCHANGEDB_Session *session,
struct GNUNET_TIME_Absolute now,
TALER_EXCHANGEDB_ReserveExpiredCallback rec,
void *rec_cls)
struct ExpiredReserveContext
{
struct GNUNET_PQ_QueryParam params[] = {
GNUNET_PQ_query_param_absolute_time (&now),
GNUNET_PQ_query_param_end
};
PGresult *result;
int nrows;
/**
* Function to call for each expired reserve.
*/
TALER_EXCHANGEDB_ReserveExpiredCallback rec;
result = GNUNET_PQ_exec_prepared (session->conn,
"get_expired_reserves",
params);
if (PGRES_TUPLES_OK !=
PQresultStatus (result))
{
BREAK_DB_ERR (result, session->conn);
PQclear (result);
return GNUNET_SYSERR;
}
nrows = PQntuples (result);
if (0 == nrows)
{
/* no matches found */
PQclear (result);
return GNUNET_NO;
}
/**
* Closure to give to @e rec.
*/
void *rec_cls;
for (int i=0;i<nrows;i++)
/**
* Set to #GNUNET_SYSERR on error.
*/
int status;
};
/**
* Function to be called with the results of a SELECT statement
* that has returned @a num_results results.
*
* @param cls closure
* @param result the postgres result
* @param num_result the number of results in @a result
*/
static void
reserve_expired_cb (void *cls,
PGresult *result,
unsigned int num_results)
{
struct ExpiredReserveContext *erc = cls;
int ret;
ret = GNUNET_OK;
for (unsigned int i=0;i<num_results;i++)
{
struct GNUNET_TIME_Absolute exp_date;
json_t *account_details;
struct TALER_ReservePublicKeyP reserve_pub;
struct TALER_Amount remaining_balance;
int ret;
struct GNUNET_PQ_ResultSpec rs[] = {
GNUNET_PQ_result_spec_absolute_time ("expiration_date",
&exp_date),
@ -4701,21 +4696,59 @@ postgres_get_expired_reserves (void *cls,
rs,
i))
{
PQclear (result);
GNUNET_break (0);
return GNUNET_SYSERR;
ret = GNUNET_SYSERR;
break;
}
ret = rec (rec_cls,
&reserve_pub,
&remaining_balance,
account_details,
exp_date);
ret = erc->rec (erc->rec_cls,
&reserve_pub,
&remaining_balance,
account_details,
exp_date);
GNUNET_PQ_cleanup_result (rs);
if (GNUNET_OK != ret)
break;
}
PQclear (result);
return GNUNET_OK;
erc->status = ret;
}
/**
* Obtain information about expired reserves and their
* remaining balances.
*
* @param cls closure of the plugin
* @param session database connection
* @param now timestamp based on which we decide expiration
* @param rec function to call on expired reserves
* @param rec_cls closure for @a rec
* @return transaction status
*/
static enum GNUNET_DB_QueryStatus
postgres_get_expired_reserves (void *cls,
struct TALER_EXCHANGEDB_Session *session,
struct GNUNET_TIME_Absolute now,
TALER_EXCHANGEDB_ReserveExpiredCallback rec,
void *rec_cls)
{
struct GNUNET_PQ_QueryParam params[] = {
GNUNET_PQ_query_param_absolute_time (&now),
GNUNET_PQ_query_param_end
};
struct ExpiredReserveContext ectx;
enum GNUNET_DB_QueryStatus qs;
ectx.rec = rec;
ectx.rec_cls = rec_cls;
ectx.status = GNUNET_OK;
qs = GNUNET_PQ_eval_prepared_multi_select (session->conn,
"get_expired_reserves",
params,
&reserve_expired_cb,
&ectx);
if (GNUNET_OK != ectx.status)
return GNUNET_DB_STATUS_HARD_ERROR;
return qs;
}
@ -4730,11 +4763,9 @@ postgres_get_expired_reserves (void *cls,
* @param wtid wire transfer details
* @param amount_with_fee amount we charged to the reserve
* @param closing_fee how high is the closing fee
* @return #GNUNET_OK on success,
* #GNUNET_NO if the record exists or on transient errors
* #GNUNET_SYSERR on failure
* @return transaction status code
*/
static int
static enum GNUNET_DB_QueryStatus
postgres_insert_reserve_closed (void *cls,
struct TALER_EXCHANGEDB_Session *session,
const struct TALER_ReservePublicKeyP *reserve_pub,
@ -4757,11 +4788,11 @@ postgres_insert_reserve_closed (void *cls,
int ret;
enum GNUNET_DB_QueryStatus qs;
ret = execute_prepared_non_select (session,
"reserves_close_insert",
params);
if (GNUNET_OK != ret)
return ret;
qs = GNUNET_PQ_eval_prepared_non_select (session->conn,
"reserves_close_insert",
params);
if (GNUNET_DB_STATUS_SUCCESS_ONE_RESULT != qs)
return qs;
/* update reserve balance */
reserve.pub = *reserve_pub;
@ -4770,10 +4801,11 @@ postgres_insert_reserve_closed (void *cls,
session,
&reserve)))
{
/* FIXME: #5010 */
/* Existence should have been checked before we got here... */
GNUNET_break (GNUNET_DB_STATUS_SUCCESS_NO_RESULTS == qs);
return (GNUNET_DB_STATUS_SUCCESS_NO_RESULTS == qs) ? GNUNET_NO : GNUNET_SYSERR;
GNUNET_break (GNUNET_DB_STATUS_SOFT_ERROR == qs);
if (GNUNET_DB_STATUS_SUCCESS_NO_RESULTS == qs)
qs = GNUNET_DB_STATUS_HARD_ERROR;
return qs;
}
ret = TALER_amount_subtract (&reserve.balance,
&reserve.balance,
@ -4786,18 +4818,12 @@ postgres_insert_reserve_closed (void *cls,
GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
"Closing of reserve `%s' refused due to balance missmatch. Retrying.\n",
TALER_B2S (reserve_pub));
return GNUNET_NO;
return GNUNET_DB_STATUS_HARD_ERROR;
}
GNUNET_break (GNUNET_NO == ret);
qs = reserves_update (cls,
session,
&reserve);
if (0 >= qs)
{
GNUNET_break (0);
return GNUNET_SYSERR;
}
return GNUNET_OK;
return reserves_update (cls,
session,
&reserve);
}
@ -4809,9 +4835,9 @@ postgres_insert_reserve_closed (void *cls,
* @param type type of the wire transfer (i.e. "sepa")
* @param buf buffer with wire transfer preparation data
* @param buf_size number of bytes in @a buf
* @return #GNUNET_OK on success, #GNUNET_SYSERR on DB errors
* @return query status code
*/
static int
static enum GNUNET_DB_QueryStatus
postgres_wire_prepare_data_insert (void *cls,
struct TALER_EXCHANGEDB_Session *session,
const char *type,
@ -4824,9 +4850,9 @@ postgres_wire_prepare_data_insert (void *cls,
GNUNET_PQ_query_param_end
};
return execute_prepared_non_select (session,
"wire_prepare_data_insert",
params);
return GNUNET_PQ_eval_prepared_non_select (session->conn,
"wire_prepare_data_insert",
params);
}

4
src/exchangedb/test_exchangedb.c
View File

@ -119,7 +119,7 @@ test_wire_prepare (struct TALER_EXCHANGEDB_Session *session)
session,
&dead_prepare_cb,
NULL));
FAILIF (GNUNET_OK !=
FAILIF (GNUNET_DB_STATUS_SUCCESS_ONE_RESULT !=
plugin->wire_prepare_data_insert (plugin->cls,
session,
"testcase",
@ -1666,7 +1666,7 @@ run (void *cls)
GNUNET_assert (GNUNET_OK ==
TALER_string_to_amount (CURRENCY ":0.000010",
&fee_closing));
FAILIF (GNUNET_OK !=
FAILIF (GNUNET_DB_STATUS_SUCCESS_ONE_RESULT !=
plugin->insert_reserve_closed (plugin->cls,
session,
&reserve_pub,

19
src/include/taler_exchangedb_plugin.h
View File

@ -1009,9 +1009,9 @@ typedef int
* @param left amount left in the reserve
* @param account_details information about the reserve's bank account
* @param expiration_date when did the reserve expire
* @return #GNUNET_OK to continue to iterate, #GNUNET_SYSERR to stop
* @return transaction status code to pass on
*/
typedef int
typedef enum GNUNET_DB_QueryStatus
(*TALER_EXCHANGEDB_ReserveExpiredCallback)(void *cls,
const struct TALER_ReservePublicKeyP *reserve_pub,
const struct TALER_Amount *left,
@ -1821,11 +1821,9 @@ struct TALER_EXCHANGEDB_Plugin
* @param now timestamp based on which we decide expiration
* @param rec function to call on expired reserves
* @param rec_cls closure for @a rec
* @return #GNUNET_SYSERR on database error
* #GNUNET_NO if there are no expired non-empty reserves
* #GNUNET_OK on success
* @return transaction status
*/
int
enum GNUNET_DB_QueryStatus
(*get_expired_reserves)(void *cls,
struct TALER_EXCHANGEDB_Session *session,
struct GNUNET_TIME_Absolute now,
@ -1844,10 +1842,9 @@ struct TALER_EXCHANGEDB_Plugin
* @param wtid identifier for the wire transfer
* @param amount_with_fee amount we charged to the reserve
* @param closing_fee how high is the closing fee
* @return #GNUNET_OK on success, #GNUNET_NO if the record exists,
* #GNUNET_SYSERR on failure
* @return transaction status code
*/
int
enum GNUNET_DB_QueryStatus
(*insert_reserve_closed)(void *cls,
struct TALER_EXCHANGEDB_Session *session,
const struct TALER_ReservePublicKeyP *reserve_pub,
@ -1866,9 +1863,9 @@ struct TALER_EXCHANGEDB_Plugin
* @param type type of the wire transfer (i.e. "sepa")
* @param buf buffer with wire transfer preparation data
* @param buf_size number of bytes in @a buf
* @return #GNUNET_OK on success, #GNUNET_SYSERR on DB errors
* @return query status code
*/
int
enum GNUNET_DB_QueryStatus
(*wire_prepare_data_insert)(void *cls,
struct TALER_EXCHANGEDB_Session *session,
const char *type,