commit de3eab29eebec8d1c1b1bf0ea2adc39ba647691a parent 73c070c5bbe418f95774d8eedf5b43cebe037fb6 Author: Florian Dold <dold@taler.net> Date: Thu, 10 Sep 2026 23:30:20 +0200 kyccheck: stop refresh loops and scans across all instances KYC long-poll wakeups forced another exchange check, whose result notified the pollers even when nothing changed. These notifications also woke other instances, and each refresh scanned every instance schema, multiplying the work as instance count grew. Refresh once per HTTP request and target notifications and refresh work to the affected instance. Requires updating SQL procedures and restarting HTTPD and kyccheck together because their internal APIs and notification formats change. Diffstat:
24 files changed, 1490 insertions(+), 130 deletions(-)
diff --git a/src/backend/meson.build b/src/backend/meson.build @@ -184,7 +184,7 @@ taler_merchant_httpd_SOURCES = [ 'taler-merchant-httpd_delete-private-donau-DONAU_SERIAL.c', ] -executable( +merchant_httpd = executable( 'taler-merchant-httpd', taler_merchant_httpd_SOURCES, dependencies: [ @@ -379,3 +379,40 @@ test( test_merchant_kyc_order, suite: ['backend'], ) + +test_merchant_kyccheck = executable( + 'test_merchant_kyccheck', + ['test_merchant_kyccheck.c'], + dependencies: [ + libtalermerchantutil_dep, + talerutil_dep, + talerjson_dep, + libtalermerchantdb_dep, + talerpq_dep, + talerexchange_dep, + gnunetutil_dep, + gnunetjson_dep, + gnunetcurl_dep, + pq_dep, + curl_dep, + json_dep, + ], + include_directories: [incdir, configuration_inc], + install: false, +) + +test('test_merchant_kyccheck', test_merchant_kyccheck, suite: ['backend']) + +# Uses a disposable PostgreSQL cluster and real HTTP long polls. +test( + 'kyc-refresh', + find_program('python3'), + args: [ + files('../backenddb/test_kyc_refresh.py'), + meson.project_source_root(), + meson.project_build_root(), + ], + depends: [merchant_httpd, test_merchant_kyccheck, gprocedures_sql, iprocedures_sql] + migration_sql, + suite: ['backenddb', 'integrationtests'], + timeout: 180, +) diff --git a/src/backend/taler-merchant-httpd_exchanges.c b/src/backend/taler-merchant-httpd_exchanges.c @@ -23,6 +23,7 @@ #include <taler/taler_json_lib.h> #include <taler/taler_dbevents.h> #include "taler-merchant-httpd_exchanges.h" +#include "taler-merchant-httpd_get-private-kyc.h" #include "taler-merchant-httpd.h" #include "merchant-database/get_kyc_limits.h" #include "merchant-database/set_instance.h" @@ -1003,6 +1004,7 @@ reload_exchange_keys (struct TMH_Exchange *exchange) } } + TMH_kyc_keys_changed (exchange->url); process_find_operations (exchange); } diff --git a/src/backend/taler-merchant-httpd_get-private-kyc.c b/src/backend/taler-merchant-httpd_get-private-kyc.c @@ -321,6 +321,11 @@ struct KycContext * thus must not yet respond? */ bool in_db; + + /** + * The initial database read already requested an exchange refresh. + */ + bool refresh_requested; }; @@ -543,6 +548,48 @@ kyc_change_cb (void *cls, } +void +TMH_kyc_keys_changed (const char *exchange_url) +{ + for (struct KycContext *kc = kc_head; + NULL != kc; + kc = kc->next) + { + size_t off; + const json_t *entry; + bool affected = false; + + if (NULL == kc->eh) + continue; /* no long poll */ + /* Completed account lookups live in the response; other lookups may + still be waiting for keys. Check both to cover updates during reads. */ + json_array_foreach (kc->kycs_data, off, entry) + { + const char *url = json_string_value (json_object_get (entry, + "exchange_url")); + + if ( (NULL != url) && + (0 == strcmp (url, exchange_url)) ) + { + affected = true; + break; + } + } + for (struct ExchangeKycRequest *ekr = kc->exchange_pending_head; + (! affected) && (NULL != ekr); + ekr = ekr->next) + affected = (0 == strcmp (ekr->exchange_url, exchange_url)); + if (affected) + { + /* Keys affect ToS flags, default limits and KYC auth instructions. + Recompute the full response and let its ETag determine whether to + return. Like a status notification, this does not force a refresh. */ + kyc_change_cb (kc, NULL, 0); + } + } +} + + /** * Suspend @a kc until we have a change in the account status. * @@ -586,6 +633,7 @@ phase_determine_long_poll (struct KycContext *kc) .header.type = htons ( TALER_DBEVENT_MERCHANT_EXCHANGE_KYC_STATUS_CHANGED ), + .merchant_pub = kc->mi->merchant_pub, .h_wire = kc->h_wire }; @@ -598,14 +646,15 @@ phase_determine_long_poll (struct KycContext *kc) } else { - struct GNUNET_DB_EventHeaderP hdr = { - .size = htons (sizeof (hdr)), - .type = htons (TALER_DBEVENT_MERCHANT_KYC_STATUS_CHANGED) + struct TALER_MERCHANTDB_InstanceKycStatusChangeEventP hdr = { + .header.size = htons (sizeof (hdr)), + .header.type = htons (TALER_DBEVENT_MERCHANT_KYC_STATUS_CHANGED), + .merchant_pub = kc->mi->merchant_pub }; kc->eh = TALER_MERCHANTDB_event_listen ( TMH_db, - &hdr, + &hdr.header, GNUNET_TIME_absolute_get_remaining (kc->timeout), &kyc_change_cb, kc); @@ -1407,9 +1456,12 @@ phase_database_kyc_check (struct KycContext *kc) ? &kc->h_wire : NULL, kc->exchange_url, + ! kc->refresh_requested, &kyc_status_cb, kc); kc->in_db = false; + if (qs >= 0) + kc->refresh_requested = true; GNUNET_log (GNUNET_ERROR_TYPE_INFO, "iterate_kyc_statuses returned %d records\n", (int) qs); diff --git a/src/backend/taler-merchant-httpd_get-private-kyc.h b/src/backend/taler-merchant-httpd_get-private-kyc.h @@ -36,6 +36,16 @@ TMH_force_kyc_resume (void); /** + * Recheck pending KYC responses using an exchange whose cached keys changed. + * Must be called after the exchange cache has been updated. + * + * @param exchange_url exchange whose keys changed + */ +void +TMH_kyc_keys_changed (const char *exchange_url); + + +/** * Change the instance's kyc settings. * This is the handler called using the instance's own kycentication. * diff --git a/src/backend/taler-merchant-kyccheck.c b/src/backend/taler-merchant-kyccheck.c @@ -199,6 +199,11 @@ struct Account struct Inquiry { /** + * Key in the index of inquiries by instance, account and exchange. + */ + struct GNUNET_HashCode key; + + /** * Kept in a DLL. */ struct Inquiry *next; @@ -411,6 +416,23 @@ static struct GNUNET_DB_EventHandler *keys_rule; static struct GNUNET_SCHEDULER_Task *account_task; /** + * Pending refreshes, coalesced by instance serial. + */ +struct Refresh +{ + struct Refresh *next; + struct Refresh *prev; + struct GNUNET_HashCode key; + uint64_t merchant_serial; +}; + +static struct Refresh *refresh_head; +static struct Refresh *refresh_tail; +static struct GNUNET_CONTAINER_MultiHashMap *refresh_map; +static struct GNUNET_CONTAINER_MultiHashMap *inquiry_map; +static struct GNUNET_SCHEDULER_Task *refresh_task; + +/** * Counter determining how often we have called * "iterate_accounts" on the database. */ @@ -460,6 +482,62 @@ inquiry_work (void *cls); /** + * Hash a fully qualified inquiry identity. Include string terminators to + * keep adjacent components unambiguous. + */ +static void +inquiry_key (const char *instance_id, + const struct TALER_MerchantWireHashP *h_wire, + const char *exchange_url, + struct GNUNET_HashCode *key) +{ + struct GNUNET_HashContext *hc = GNUNET_CRYPTO_hash_context_start (); + + GNUNET_CRYPTO_hash_context_read (hc, + instance_id, + strlen (instance_id) + 1); + GNUNET_CRYPTO_hash_context_read (hc, + h_wire, + sizeof (*h_wire)); + GNUNET_CRYPTO_hash_context_read (hc, + exchange_url, + strlen (exchange_url) + 1); + GNUNET_CRYPTO_hash_context_finish (hc, + key); +} + + +/** + * An inquiry keeps its active slot through automatic ToS acceptance. + */ +static bool +inquiry_busy (const struct Inquiry *i) +{ + return (NULL != i->kyc) || + (NULL != i->kyc_info) || + (NULL != i->tos_upload); +} + + +/** + * Request an immediate check, sharing any active or queued work. + */ +static void +request_inquiry (struct Inquiry *i) +{ + if (inquiry_busy (i)) + return; + i->due = GNUNET_TIME_UNIT_ZERO_ABS; + if (i->limited) + return; + if (NULL != i->task) + GNUNET_SCHEDULER_cancel (i->task); + i->task = GNUNET_SCHEDULER_add_now (&inquiry_work, + i); +} + + +/** * An inquiry finished, check if we should resume others. */ static void @@ -1289,6 +1367,16 @@ start_inquiry (struct Exchange *e, i = GNUNET_new (struct Inquiry); i->e = e; i->a = a; + inquiry_key (a->instance_id, + &a->h_wire, + e->keys->exchange_url, + &i->key); + GNUNET_assert (GNUNET_OK == + GNUNET_CONTAINER_multihashmap_put ( + inquiry_map, + &i->key, + i, + GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY)); GNUNET_CONTAINER_DLL_insert (a->i_head, a->i_tail, i); @@ -1340,6 +1428,14 @@ start_inquiry (struct Exchange *e, } if (qs > 0) i->not_first_time = true; + if (TALER_EC_MERCHANT_PRIVATE_ACCOUNT_NOT_ELIGIBLE_FOR_EXCHANGE == i->last_ec) + { + /* Eligibility was restored. Do not retain the ineligible status's + infinite delay or backoff. */ + i->due = GNUNET_TIME_UNIT_ZERO_ABS; + i->backoff = GNUNET_TIME_UNIT_ZERO; + i->not_first_time = false; + } if (GNUNET_YES == test_mode) i->due = GNUNET_TIME_UNIT_ZERO_ABS; /* immediately */ inquiry_work (i); @@ -1356,6 +1452,10 @@ stop_inquiry (struct Inquiry *i) { struct Account *a = i->a; + GNUNET_assert (GNUNET_YES == + GNUNET_CONTAINER_multihashmap_remove (inquiry_map, + &i->key, + i)); GNUNET_CONTAINER_DLL_remove (a->i_head, a->i_tail, i); @@ -1703,16 +1803,26 @@ find_keys (const char *exchange_url) NULL != a; a = a->next) { - bool was_eligible = is_eligible (old_keys, - a); - bool now_eligible = is_eligible (keys, - a); + bool was_eligible; + bool now_eligible; + + if (a->account_gen != database_gen) + continue; + was_eligible = is_eligible (old_keys, + a); + now_eligible = is_eligible (keys, + a); if (was_eligible == now_eligible) continue; /* no change, do nothing */ if (was_eligible) + { + flag_ineligible (a->instance_id, + keys->exchange_url, + &a->h_wire); stop_inquiry_at (e, a); + } else /* is_eligible */ start_inquiry (e, a); @@ -1867,13 +1977,7 @@ rule_triggered (void *cls, i->e->keys->exchange_url)) continue; i->kyc_ok = false; - i->due = GNUNET_TIME_UNIT_ZERO_ABS; - if (NULL != i->task) - { - GNUNET_SCHEDULER_cancel (i->task); - i->task = NULL; - } - if (NULL != i->kyc) + if (inquiry_busy (i)) { GNUNET_log (GNUNET_ERROR_TYPE_INFO, "/kyc-check already running for %s\n", @@ -1884,9 +1988,7 @@ rule_triggered (void *cls, "Starting %skyc-check for `%s' due to KYC rule trigger\n", exchange_url, i->a->merchant_account_uri.full_payto); - i->task = GNUNET_SCHEDULER_add_at (i->due, - &inquiry_work, - i); + request_inquiry (i); return; } } @@ -1949,6 +2051,19 @@ shutdown_task (void *cls) (void) cls; GNUNET_log (GNUNET_ERROR_TYPE_INFO, "Running shutdown\n"); + if (NULL != refresh_task) + { + GNUNET_SCHEDULER_cancel (refresh_task); + refresh_task = NULL; + } + while (NULL != refresh_head) + { + struct Refresh *r = refresh_head; + + GNUNET_CONTAINER_DLL_remove (refresh_head, refresh_tail, r); + GNUNET_free (r); + } + GNUNET_CONTAINER_multihashmap_destroy (refresh_map); while (NULL != e_head) { struct Exchange *e = e_head; @@ -1975,6 +2090,7 @@ shutdown_task (void *cls) GNUNET_free (a->instance_id); GNUNET_free (a); } + GNUNET_CONTAINER_multihashmap_destroy (inquiry_map); if (NULL != eh_accounts) { TALER_MERCHANTDB_event_listen_cancel (eh_accounts); @@ -2040,74 +2156,55 @@ force_check_now (void *cls, const char *exchange_url, const struct TALER_MerchantWireHashP *h_wire) { - for (struct Account *a = a_head; - NULL != a; - a = a->next) + struct GNUNET_HashCode key; + struct Inquiry *i; + + (void) cls; + inquiry_key (instance_id, h_wire, exchange_url, &key); + i = GNUNET_CONTAINER_multihashmap_get (inquiry_map, &key); + if (NULL == i) { - if (0 != - strcmp (instance_id, - a->instance_id)) - continue; - if (0 != - GNUNET_memcmp (h_wire, - &a->h_wire)) - continue; - for (struct Inquiry *i = a->i_head; - NULL != i; - i = i->next) - { - if (0 != - strcmp (i->e->keys->exchange_url, - exchange_url)) - continue; - /* If we are not actively checking with the exchange, do start - to check immediately */ - if (NULL == i->kyc) - { - i->due = GNUNET_TIME_absolute_get (); /* now! */ - if (NULL != i->task) - GNUNET_SCHEDULER_cancel (i->task); - i->task = GNUNET_SCHEDULER_add_at (i->due, - &inquiry_work, - i); - } - return; - } + /* Account discovery and exchange-key loading recover the persisted due + time. Absence from the index does not prove the account ineligible. */ + return; } - GNUNET_log (GNUNET_ERROR_TYPE_INFO, - "No inquiry at `%s' for exchange `%s' and h_wire `%s'. Likely the account is not eligible.\n", - instance_id, - exchange_url, - TALER_B2S (h_wire)); - /* In this case, set the due date back to FOREVER */ - flag_ineligible (instance_id, - exchange_url, - h_wire); + GNUNET_assert (0 == strcmp (instance_id, i->a->instance_id)); + GNUNET_assert (0 == GNUNET_memcmp (h_wire, &i->a->h_wire)); + GNUNET_assert (0 == strcmp (exchange_url, i->e->keys->exchange_url)); + if (i->a->account_gen != database_gen) + return; + request_inquiry (i); } /** - * Function called when a KYC status update was forced by an - * application checking the KYC status of an account. - * - * @param cls closure (NULL) - * @param extra additional event data provided - * @param extra_size number of bytes in @a extra + * Process one instance per scheduler turn, after pending account discovery. */ static void -update_forced (void *cls, - const void *extra, - size_t extra_size) +process_refresh (void *cls) { + struct Refresh *r = refresh_head; enum GNUNET_DB_QueryStatus qs; (void) cls; - (void) extra; - (void) extra_size; + refresh_task = NULL; + if (NULL != account_task) + { + refresh_task = GNUNET_SCHEDULER_add_now (&process_refresh, NULL); + return; + } + GNUNET_assert (NULL != r); + GNUNET_CONTAINER_DLL_remove (refresh_head, refresh_tail, r); + GNUNET_assert (GNUNET_YES == + GNUNET_CONTAINER_multihashmap_remove (refresh_map, + &r->key, + r)); qs = TALER_MERCHANTDB_iterate_outdated_kyc_statuses ( pg, + r->merchant_serial, &force_check_now, NULL); + GNUNET_free (r); if (qs < 0) { GNUNET_break (0); @@ -2115,10 +2212,55 @@ update_forced (void *cls, GNUNET_SCHEDULER_shutdown (); return; } + if (NULL != refresh_head) + refresh_task = GNUNET_SCHEDULER_add_now (&process_refresh, NULL); } /** + * Queue an instance refresh. The notification payload is exactly one + * unsigned 64-bit merchant serial in network byte order. + */ +static void +update_forced (void *cls, + const void *extra, + size_t extra_size) +{ + uint64_t serial; + struct GNUNET_HashCode key; + struct Refresh *r; + + (void) cls; + if ( (NULL == extra) || (sizeof (serial) != extra_size) ) + { + GNUNET_log (GNUNET_ERROR_TYPE_WARNING, + "Ignoring KYC refresh notification with invalid payload size\n"); + return; + } + memcpy (&serial, extra, sizeof (serial)); + GNUNET_CRYPTO_hash (&serial, sizeof (serial), &key); + serial = GNUNET_ntohll (serial); + if ( (0 == serial) || (serial > INT64_MAX) ) + { + GNUNET_log (GNUNET_ERROR_TYPE_WARNING, + "Ignoring KYC refresh notification with invalid instance serial\n"); + return; + } + if (GNUNET_CONTAINER_multihashmap_contains (refresh_map, &key)) + return; + r = GNUNET_new (struct Refresh); + r->key = key; + r->merchant_serial = serial; + GNUNET_CONTAINER_DLL_insert_tail (refresh_head, refresh_tail, r); + GNUNET_assert (GNUNET_OK == + GNUNET_CONTAINER_multihashmap_put ( + refresh_map, &r->key, r, + GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY)); + if (NULL == refresh_task) + refresh_task = GNUNET_SCHEDULER_add_now (&process_refresh, NULL); +} + +/** * First task. * * @param cls closure, NULL @@ -2136,6 +2278,8 @@ run (void *cls, (void) cfgfile; cfg = c; + inquiry_map = GNUNET_CONTAINER_multihashmap_create (256, GNUNET_YES); + refresh_map = GNUNET_CONTAINER_multihashmap_create (32, GNUNET_YES); TALER_EXCHANGE_setup (enable_h3 ? TALER_EXCHANGE_GO_ENABLE_HTTP3 : TALER_EXCHANGE_GO_FORCE_HTTP1_1); diff --git a/src/backend/test_merchant_kyccheck.c b/src/backend/test_merchant_kyccheck.c @@ -0,0 +1,272 @@ +/* + This file is part of TALER + Copyright (C) 2026 Taler Systems SA + + 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 + Foundation; either version 3, or (at your option) any later version. + + TALER is distributed in the hope that it will be useful, but WITHOUT ANY + WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR + A PARTICULAR PURPOSE. See the GNU General Public License for more details. + + You should have received a copy of the GNU General Public License along with + TALER; see the file COPYING. If not, see <http://www.gnu.org/licenses/> + */ +/** + * @file backend/test_merchant_kyccheck.c + * @brief Exercise refresh scheduling and database-backed status transitions. + */ +/* Inject successive /keys snapshots while using the real database helpers. */ +#define TALER_MERCHANTDB_get_exchange_keys test_get_exchange_keys +#define main kyccheck_main +#include "taler-merchant-kyccheck.c" +#undef main +#undef TALER_MERCHANTDB_get_exchange_keys + +static struct TALER_EXCHANGE_Keys *next_keys; + + +enum GNUNET_DB_QueryStatus +test_get_exchange_keys (struct TALER_MERCHANTDB_PostgresContext *db, + const char *exchange_url, + struct GNUNET_TIME_Absolute *first_retry, + struct TALER_EXCHANGE_Keys **keys) +{ + (void) db; + GNUNET_assert (0 == strcmp (exchange_url, next_keys->exchange_url)); + *first_retry = GNUNET_TIME_UNIT_ZERO_ABS; + *keys = TALER_EXCHANGE_keys_incref (next_keys); + return GNUNET_DB_STATUS_SUCCESS_ONE_RESULT; +} + + +/* Called by the HTTP fixture with its disposable database configuration. */ +static void +check_database_transition (void *cls) +{ + const char *mode = cls; + struct TALER_EXCHANGE_WireAccount wa = { + .fpayto_uri = { + .full_payto = "payto://x-taler-bank/localhost/exchange?receiver-name=Exchange" + } + }; + struct TALER_EXCHANGE_Keys eligible = { + .exchange_url = "http://localhost:1/", + .accounts = &wa, + .accounts_len = 1, + .rc = 1 + }; + struct TALER_EXCHANGE_Keys ineligible = { + .exchange_url = eligible.exchange_url, + .rc = 1 + }; + struct Account a = { + .instance_id = "test-1", + .merchant_account_uri = { + .full_payto = "payto://x-taler-bank/localhost/account-1?receiver-name=Test" + } + }; + + pg = TALER_MERCHANTDB_connect (cfg); + GNUNET_assert (NULL != pg); + memset (&a.h_wire, 42, sizeof (a.h_wire)); + if (0 == strcmp (mode, "tos-conflict")) + { + struct Exchange e = { .keys = &eligible }; + struct Inquiry i = { + .a = &a, + .e = &e, + .tos_etag = GNUNET_strdup ("v2") + }; + const struct TALER_EXCHANGE_PostKycUploadResponse response = { + .hr.http_status = MHD_HTTP_CONFLICT + }; + + active_inquiries = 1; + tos_upload_cb (&i, &response); + GNUNET_assert (0 == active_inquiries); + GNUNET_assert (NULL != i.task); + GNUNET_SCHEDULER_cancel (i.task); + } + else + { + struct Account inactive = { + .instance_id = "test-2", + .merchant_account_uri = a.merchant_account_uri, + .h_wire = a.h_wire, + .account_gen = 1 /* differs from database_gen; discovery removed it */ + }; + + GNUNET_assert (0 == strcmp (mode, "eligibility")); + inquiry_map = GNUNET_CONTAINER_multihashmap_create (8, GNUNET_YES); + ctx = GNUNET_CURL_init (&GNUNET_CURL_gnunet_scheduler_reschedule, &rc); + rc = GNUNET_CURL_gnunet_rc_create (ctx); + a_head = a_tail = &a; + GNUNET_CONTAINER_DLL_insert (a_head, a_tail, &inactive); + /* Pending discovery must not turn an unindexed account ineligible. */ + force_check_now (NULL, "test-2", eligible.exchange_url, &a.h_wire); + next_keys = &eligible; + find_keys (eligible.exchange_url); + GNUNET_assert (NULL != a.i_head); + GNUNET_assert (a.i_head->kyc_ok); + next_keys = &ineligible; + find_keys (eligible.exchange_url); + GNUNET_assert (NULL == a.i_head); + GNUNET_assert (0 == GNUNET_CONTAINER_multihashmap_size (inquiry_map)); + /* Repeated refreshes while ineligible must not recreate an inquiry. */ + force_check_now (NULL, a.instance_id, eligible.exchange_url, &a.h_wire); + GNUNET_assert (NULL == a.i_head); + next_keys = &eligible; + find_keys (eligible.exchange_url); + GNUNET_assert (NULL != a.i_head); + GNUNET_assert (TALER_EC_MERCHANT_PRIVATE_ACCOUNT_NOT_ELIGIBLE_FOR_EXCHANGE + == a.i_head->last_ec); + GNUNET_assert (0 == a.i_head->backoff.rel_value_us); + GNUNET_assert (NULL != a.i_head->kyc); + GNUNET_assert (! a.i_head->not_first_time); + /* Cancel before performing network I/O; keep the persisted status for + the fixture's HTTP assertion. */ + stop_inquiries (&a); + TALER_EXCHANGE_keys_decref (e_head->keys); + GNUNET_free (e_head); + e_tail = NULL; + a_head = a_tail = NULL; + GNUNET_CONTAINER_multihashmap_destroy (inquiry_map); + GNUNET_CURL_gnunet_rc_destroy (rc); + GNUNET_CURL_fini (ctx); + } + GNUNET_assert (EXIT_SUCCESS == global_ret); + TALER_MERCHANTDB_disconnect (pg); + GNUNET_SCHEDULER_shutdown (); +} + + +static void +check_scheduling (void *cls) +{ + struct TALER_EXCHANGE_Keys keys = { + .exchange_url = "http://localhost:1/" + }; + struct Exchange e = { .keys = &keys }; + struct Account a = { + .instance_id = "test", + .merchant_account_uri = { .full_payto = "payto://x-taler-bank/localhost/test" } + }; + struct Inquiry *i = GNUNET_new (struct Inquiry); + struct GNUNET_HashCode other; + uint64_t serial = GNUNET_htonll (42); + uint64_t serial2 = GNUNET_htonll (43); + + (void) cls; + inquiry_map = GNUNET_CONTAINER_multihashmap_create (8, GNUNET_YES); + refresh_map = GNUNET_CONTAINER_multihashmap_create (8, GNUNET_YES); + i->a = &a; + i->e = &e; + inquiry_key (a.instance_id, &a.h_wire, keys.exchange_url, &i->key); + inquiry_key ("different-instance", &a.h_wire, keys.exchange_url, &other); + GNUNET_assert (0 != GNUNET_memcmp (&i->key, &other)); + GNUNET_CONTAINER_DLL_insert (a.i_head, a.i_tail, i); + GNUNET_assert (GNUNET_OK == GNUNET_CONTAINER_multihashmap_put ( + inquiry_map, &i->key, i, + GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY)); + + /* Repeated notifications for an instance are processed once per pending + batch, and distinct instances retain their own requests. */ + update_forced (NULL, &serial, sizeof (serial)); + update_forced (NULL, &serial, sizeof (serial)); + update_forced (NULL, &serial2, sizeof (serial2)); + GNUNET_assert (2 == GNUNET_CONTAINER_multihashmap_size (refresh_map)); + GNUNET_assert (42 == refresh_head->merchant_serial); + GNUNET_assert (43 == refresh_tail->merchant_serial); + update_forced (NULL, NULL, 0); + update_forced (NULL, &serial, 1); + serial = 0; + update_forced (NULL, &serial, sizeof (serial)); + GNUNET_assert (2 == GNUNET_CONTAINER_multihashmap_size (refresh_map)); + GNUNET_assert (EXIT_SUCCESS == global_ret); + GNUNET_SCHEDULER_cancel (refresh_task); + refresh_task = NULL; + + /* Indexed lookup affects only the selected inquiry, with one pending task. */ + force_check_now (NULL, "different-instance", keys.exchange_url, &a.h_wire); + GNUNET_assert (NULL == i->task); + force_check_now (NULL, a.instance_id, keys.exchange_url, &a.h_wire); + GNUNET_assert (NULL != i->task); + force_check_now (NULL, a.instance_id, keys.exchange_url, &a.h_wire); + GNUNET_assert (NULL != i->task); + GNUNET_SCHEDULER_cancel (i->task); + i->task = NULL; + + /* All stages of an active exchange interaction share the same slot. The + sentinels are never dereferenced and are cleared before cleanup. */ + i->kyc = (void *) i; + request_inquiry (i); + GNUNET_assert (NULL == i->task); + i->kyc = NULL; + i->kyc_info = (void *) i; + request_inquiry (i); + GNUNET_assert (NULL == i->task); + i->kyc_info = NULL; + i->tos_upload = (void *) i; + request_inquiry (i); + GNUNET_assert (NULL == i->task); + i->tos_upload = NULL; + + active_inquiries = OPEN_INQUIRY_LIMIT; + inquiry_work (i); + GNUNET_assert (i->limited && at_limit); + request_inquiry (i); + GNUNET_assert (NULL == i->task); + GNUNET_assert (i->limited); + + /* Releasing slots resumes the limited inquiry exactly once. Cancel the + newly created HTTP operation before the scheduler performs network I/O. */ + ctx = GNUNET_CURL_init (&GNUNET_CURL_gnunet_scheduler_reschedule, &rc); + rc = GNUNET_CURL_gnunet_rc_create (ctx); + a_head = &a; + a_tail = &a; + active_inquiries = OPEN_INQUIRY_LIMIT / 2; + end_inquiry (); + GNUNET_assert (! i->limited); + GNUNET_assert (NULL != i->kyc); + GNUNET_assert (OPEN_INQUIRY_LIMIT / 2 == active_inquiries); + request_inquiry (i); + GNUNET_assert (NULL == i->task); + stop_inquiry (i); + GNUNET_assert (0 == GNUNET_CONTAINER_multihashmap_size (inquiry_map)); + GNUNET_CURL_gnunet_rc_destroy (rc); + GNUNET_CURL_fini (ctx); + while (NULL != refresh_head) + { + struct Refresh *r = refresh_head; + + GNUNET_CONTAINER_DLL_remove (refresh_head, refresh_tail, r); + GNUNET_free (r); + } + GNUNET_CONTAINER_multihashmap_destroy (refresh_map); + GNUNET_CONTAINER_multihashmap_destroy (inquiry_map); + a_head = a_tail = NULL; + GNUNET_SCHEDULER_shutdown (); +} + + +int +main (int argc, char *const argv[]) +{ + GNUNET_log_setup (argv[0], "WARNING", NULL); + if (3 == argc) + { + struct GNUNET_CONFIGURATION_Handle *config = + GNUNET_CONFIGURATION_create (TALER_MERCHANT_project_data ()); + + GNUNET_assert (GNUNET_OK == GNUNET_CONFIGURATION_load (config, argv[1])); + cfg = config; + GNUNET_SCHEDULER_run (&check_database_transition, (void *) argv[2]); + GNUNET_CONFIGURATION_destroy (config); + return global_ret; + } + GNUNET_assert (1 == argc); + GNUNET_SCHEDULER_run (&check_scheduling, NULL); + return 0; +} diff --git a/src/backenddb/delete_tos_accepted_early.c b/src/backenddb/delete_tos_accepted_early.c @@ -20,6 +20,7 @@ */ #include "platform.h" #include <taler/taler_pq_lib.h> +#include <taler/taler_dbevents.h> #include "merchant-database/delete_tos_accepted_early.h" #include "helper.h" @@ -30,18 +31,47 @@ TALER_MERCHANTDB_delete_tos_accepted_early ( const char *merchant_id, const char *exchange_url) { + struct TALER_MERCHANTDB_MerchantKycStatusChangeEventP ev = { + .header.size = htons (sizeof (ev)), + .header.type = htons (TALER_DBEVENT_MERCHANT_EXCHANGE_KYC_STATUS_CHANGED), + .merchant_pub = pg->current_merchant_pub + }; + struct TALER_MERCHANTDB_InstanceKycStatusChangeEventP hdr = { + .header.size = htons (sizeof (hdr)), + .header.type = htons (TALER_DBEVENT_MERCHANT_KYC_STATUS_CHANGED), + .merchant_pub = pg->current_merchant_pub + }; + char *notify_s = GNUNET_PQ_get_event_notify_channel (&hdr.header); struct GNUNET_PQ_QueryParam params[] = { GNUNET_PQ_query_param_string (exchange_url), + GNUNET_PQ_query_param_null (), + GNUNET_PQ_query_param_fixed_size ( + &ev, + offsetof (struct TALER_MERCHANTDB_MerchantKycStatusChangeEventP, h_wire)), + GNUNET_PQ_query_param_string (notify_s), GNUNET_PQ_query_param_end }; + bool changed; + struct GNUNET_PQ_ResultSpec rs[] = { + GNUNET_PQ_result_spec_bool ("changed", &changed), + GNUNET_PQ_result_spec_end + }; + enum GNUNET_DB_QueryStatus qs; GNUNET_assert (NULL != pg->current_merchant_id); GNUNET_assert (0 == strcmp (merchant_id, pg->current_merchant_id)); TMH_PQ_prepare_anon (pg, - "DELETE FROM merchant_tos_accepted" - " WHERE exchange_url=$1"); - return GNUNET_PQ_eval_prepared_non_select (pg->conn, - "", - params); + "SELECT merchant_do_set_tos_accepted_early" + " ($1, $2, $3, $4) AS changed"); + qs = GNUNET_PQ_eval_prepared_singleton_select (pg->conn, + "", + params, + rs); + GNUNET_free (notify_s); + if (qs <= 0) + return qs; + return changed + ? GNUNET_DB_STATUS_SUCCESS_ONE_RESULT + : GNUNET_DB_STATUS_SUCCESS_NO_RESULTS; } diff --git a/src/backenddb/insert_kyc_failure.c b/src/backenddb/insert_kyc_failure.c @@ -38,16 +38,18 @@ TALER_MERCHANTDB_insert_kyc_failure ( struct TALER_MERCHANTDB_MerchantKycStatusChangeEventP ev = { .header.size = htons (sizeof (ev)), .header.type = htons (TALER_DBEVENT_MERCHANT_EXCHANGE_KYC_STATUS_CHANGED), + .merchant_pub = pg->current_merchant_pub, .h_wire = *h_wire }; - struct GNUNET_DB_EventHeaderP hdr = { - .size = htons (sizeof (hdr)), - .type = htons (TALER_DBEVENT_MERCHANT_KYC_STATUS_CHANGED) + struct TALER_MERCHANTDB_InstanceKycStatusChangeEventP hdr = { + .header.size = htons (sizeof (hdr)), + .header.type = htons (TALER_DBEVENT_MERCHANT_KYC_STATUS_CHANGED), + .merchant_pub = pg->current_merchant_pub }; char *notify_s = GNUNET_PQ_get_event_notify_channel (&ev.header); char *notify2_s - = GNUNET_PQ_get_event_notify_channel (&hdr); + = GNUNET_PQ_get_event_notify_channel (&hdr.header); uint32_t http_status32 = (uint32_t) exchange_http_status; struct GNUNET_PQ_QueryParam params[] = { GNUNET_PQ_query_param_auto_from_type (h_wire), diff --git a/src/backenddb/insert_kyc_failure.sql b/src/backenddb/insert_kyc_failure.sql @@ -29,6 +29,7 @@ LANGUAGE plpgsql AS $$ DECLARE my_account_serial INT8; + my_changed BOOL; BEGIN out_no_account=FALSE; @@ -44,6 +45,20 @@ THEN RETURN; END IF; +-- Serialize updates and compare status separately from polling bookkeeping. +SELECT ROW(kyc_ok, exchange_http_status, exchange_ec_code) + IS DISTINCT FROM + ROW(in_kyc_ok, in_exchange_http_status, 0) + INTO my_changed + FROM merchant_kyc + WHERE account_serial=my_account_serial + AND exchange_url=in_exchange_url + FOR UPDATE; +IF NOT FOUND +THEN + my_changed := TRUE; +END IF; + UPDATE merchant_kyc SET kyc_timestamp=in_timestamp ,kyc_ok=in_kyc_ok @@ -69,13 +84,11 @@ THEN ,in_exchange_http_status); END IF; -EXECUTE FORMAT ( - 'NOTIFY %s' - ,in_notify_str); - -EXECUTE FORMAT ( - 'NOTIFY %s' - ,in_notify2_str); +IF my_changed +THEN + EXECUTE FORMAT ('NOTIFY %s', in_notify_str); + EXECUTE FORMAT ('NOTIFY %s', in_notify2_str); +END IF; -- Success! diff --git a/src/backenddb/insert_kyc_status.c b/src/backenddb/insert_kyc_status.c @@ -46,16 +46,18 @@ TALER_MERCHANTDB_insert_kyc_status ( struct TALER_MERCHANTDB_MerchantKycStatusChangeEventP ev = { .header.size = htons (sizeof (ev)), .header.type = htons (TALER_DBEVENT_MERCHANT_EXCHANGE_KYC_STATUS_CHANGED), + .merchant_pub = pg->current_merchant_pub, .h_wire = *h_wire }; - struct GNUNET_DB_EventHeaderP hdr = { - .size = htons (sizeof (hdr)), - .type = htons (TALER_DBEVENT_MERCHANT_KYC_STATUS_CHANGED) + struct TALER_MERCHANTDB_InstanceKycStatusChangeEventP hdr = { + .header.size = htons (sizeof (hdr)), + .header.type = htons (TALER_DBEVENT_MERCHANT_KYC_STATUS_CHANGED), + .merchant_pub = pg->current_merchant_pub }; char *notify_s = GNUNET_PQ_get_event_notify_channel (&ev.header); char *notify2_s - = GNUNET_PQ_get_event_notify_channel (&hdr); + = GNUNET_PQ_get_event_notify_channel (&hdr.header); uint32_t http_status32 = (uint32_t) exchange_http_status; uint32_t ec_code32 = (uint32_t) exchange_ec_code; struct GNUNET_PQ_QueryParam params[] = { diff --git a/src/backenddb/insert_kyc_status.sql b/src/backenddb/insert_kyc_status.sql @@ -36,6 +36,7 @@ LANGUAGE plpgsql AS $$ DECLARE my_account_serial INT8; + my_changed BOOL; BEGIN out_no_account=FALSE; @@ -51,6 +52,22 @@ THEN RETURN; END IF; +-- Serialize updates and compare status separately from polling bookkeeping. +SELECT ROW(kyc_ok, jaccount_limits, aml_review, exchange_http_status, + exchange_ec_code, access_token, last_rule_gen) + IS DISTINCT FROM + ROW(in_kyc_ok, in_jlimits, in_aml_active, in_exchange_http_status, + in_exchange_ec_code, in_access_token, in_rule_gen) + INTO my_changed + FROM merchant_kyc + WHERE account_serial=my_account_serial + AND exchange_url=in_exchange_url + FOR UPDATE; +IF NOT FOUND +THEN + my_changed := TRUE; +END IF; + UPDATE merchant_kyc SET kyc_timestamp=in_timestamp ,kyc_ok=in_kyc_ok @@ -96,13 +113,11 @@ THEN ,in_kyc_backoff); END IF; -EXECUTE FORMAT ( - 'NOTIFY %s' - ,in_notify_str); - -EXECUTE FORMAT ( - 'NOTIFY %s' - ,in_notify2_str); +IF my_changed +THEN + EXECUTE FORMAT ('NOTIFY %s', in_notify_str); + EXECUTE FORMAT ('NOTIFY %s', in_notify2_str); +END IF; -- Success! diff --git a/src/backenddb/insert_tos_accepted_early.c b/src/backenddb/insert_tos_accepted_early.c @@ -20,6 +20,7 @@ */ #include "platform.h" #include <taler/taler_pq_lib.h> +#include <taler/taler_dbevents.h> #include "merchant-database/insert_tos_accepted_early.h" #include "helper.h" @@ -31,23 +32,47 @@ TALER_MERCHANTDB_insert_tos_accepted_early ( const char *exchange_url, const char *tos_version) { + struct TALER_MERCHANTDB_MerchantKycStatusChangeEventP ev = { + .header.size = htons (sizeof (ev)), + .header.type = htons (TALER_DBEVENT_MERCHANT_EXCHANGE_KYC_STATUS_CHANGED), + .merchant_pub = pg->current_merchant_pub + }; + struct TALER_MERCHANTDB_InstanceKycStatusChangeEventP hdr = { + .header.size = htons (sizeof (hdr)), + .header.type = htons (TALER_DBEVENT_MERCHANT_KYC_STATUS_CHANGED), + .merchant_pub = pg->current_merchant_pub + }; + char *notify_s = GNUNET_PQ_get_event_notify_channel (&hdr.header); struct GNUNET_PQ_QueryParam params[] = { GNUNET_PQ_query_param_string (exchange_url), GNUNET_PQ_query_param_string (tos_version), + GNUNET_PQ_query_param_fixed_size ( + &ev, + offsetof (struct TALER_MERCHANTDB_MerchantKycStatusChangeEventP, h_wire)), + GNUNET_PQ_query_param_string (notify_s), GNUNET_PQ_query_param_end }; + bool changed; + struct GNUNET_PQ_ResultSpec rs[] = { + GNUNET_PQ_result_spec_bool ("changed", &changed), + GNUNET_PQ_result_spec_end + }; + enum GNUNET_DB_QueryStatus qs; GNUNET_assert (NULL != pg->current_merchant_id); GNUNET_assert (0 == strcmp (merchant_id, pg->current_merchant_id)); TMH_PQ_prepare_anon (pg, - "INSERT INTO merchant_tos_accepted" - " (exchange_url" - " ,tos_version)" - " VALUES ($1, $2)" - " ON CONFLICT (exchange_url)" - " DO UPDATE SET tos_version=$2"); - return GNUNET_PQ_eval_prepared_non_select (pg->conn, - "", - params); + "SELECT merchant_do_set_tos_accepted_early" + " ($1, $2, $3, $4) AS changed"); + qs = GNUNET_PQ_eval_prepared_singleton_select (pg->conn, + "", + params, + rs); + GNUNET_free (notify_s); + if (qs <= 0) + return qs; + return changed + ? GNUNET_DB_STATUS_SUCCESS_ONE_RESULT + : GNUNET_DB_STATUS_SUCCESS_NO_RESULTS; } diff --git a/src/backenddb/iterate_kyc_statuses.c b/src/backenddb/iterate_kyc_statuses.c @@ -161,6 +161,7 @@ TALER_MERCHANTDB_iterate_kyc_statuses ( const char *merchant_id, const struct TALER_MerchantWireHashP *h_wire, const char *exchange_url, + bool request_refresh, TALER_MERCHANTDB_KycCallback kyc_cb, void *kyc_cb_cls) { @@ -178,6 +179,8 @@ TALER_MERCHANTDB_iterate_kyc_statuses ( NULL == h_wire ? GNUNET_PQ_query_param_null () : GNUNET_PQ_query_param_auto_from_type (h_wire), + GNUNET_PQ_query_param_bool (request_refresh), + GNUNET_PQ_query_param_uint64 (&pg->current_merchant_serial), GNUNET_PQ_query_param_end }; enum GNUNET_DB_QueryStatus qs; @@ -197,7 +200,7 @@ TALER_MERCHANTDB_iterate_kyc_statuses ( " ,out_exchange_ec_code AS exchange_ec_code" " ,out_aml_review AS aml_review" " ,out_jaccount_limits::TEXT AS jaccount_limits" - " FROM merchant_do_account_kyc_get_status($1, $2, $3);"); + " FROM merchant_do_account_kyc_get_status($1, $2, $3, $4, $5);"); qs = GNUNET_PQ_eval_prepared_multi_select ( pg->conn, "", diff --git a/src/backenddb/iterate_kyc_statuses.sql b/src/backenddb/iterate_kyc_statuses.sql @@ -19,7 +19,9 @@ DROP FUNCTION IF EXISTS merchant_do_account_kyc_get_status; CREATE FUNCTION merchant_do_account_kyc_get_status ( IN in_now INT8, IN in_exchange_url TEXT, -- can be NULL - IN in_h_wire BYTEA -- can be NULL + IN in_h_wire BYTEA, -- can be NULL + IN in_request_refresh BOOL, + IN in_merchant_serial INT8 ) RETURNS TABLE ( out_h_wire BYTEA, -- never NULL out_payto_uri TEXT, -- never NULL @@ -39,6 +41,7 @@ DECLARE my_h_wire BYTEA; my_payto_uri TEXT; my_kyc_record RECORD; + my_refresh BOOL := FALSE; BEGIN -- Iterate over merchant_accounts @@ -68,10 +71,13 @@ BEGIN ORDER BY mk.kyc_serial_id ASC LOOP -- Ask taler-merchant-kyccheck to get us an update on the status ASAP - UPDATE merchant_kyc - SET next_kyc_poll=in_now - WHERE kyc_serial_id = my_kyc_record.kyc_serial_id; - NOTIFY XJ40P0CFMZ0DT6SFZ70VRQ19KG1HP6AJ1Q6VCDZN4N2FGPSAG4KDG; -- MERCHANT_EXCHANGE_KYC_UPDATE_FORCED + IF in_request_refresh + THEN + UPDATE merchant_kyc + SET next_kyc_poll=in_now + WHERE kyc_serial_id = my_kyc_record.kyc_serial_id; + my_refresh := TRUE; + END IF; RETURN QUERY SELECT my_h_wire, @@ -106,6 +112,15 @@ BEGIN END LOOP; -- loop over merchant_accounts + IF my_refresh + THEN + -- MERCHANT_EXCHANGE_KYC_UPDATE_FORCED. GNUnet decodes the payload + -- from Crockford base32 to an eight-byte serial in network order. + PERFORM pg_notify ( + lower ('XJ40P0CFMZ0DT6SFZ70VRQ19KG1HP6AJ1Q6VCDZN4N2FGPSAG4KDG'), + merchant.base32_crockford (int8send (in_merchant_serial))); + END IF; + END $$; COMMENT ON FUNCTION merchant_do_account_kyc_get_status - IS 'Returns the KYC status of selected exchanges and accounts, but ALSO resets the next_kyc_check time for all returned data points to the current time (in_now argument)'; + IS 'Returns selected KYC statuses. If in_request_refresh, also marks them due and notifies the checker of the affected instance.'; diff --git a/src/backenddb/iterate_outdated_kyc_statuses.c b/src/backenddb/iterate_outdated_kyc_statuses.c @@ -107,6 +107,7 @@ kyc_status_cb (void *cls, enum GNUNET_DB_QueryStatus TALER_MERCHANTDB_iterate_outdated_kyc_statuses ( struct TALER_MERCHANTDB_PostgresContext *pg, + uint64_t merchant_serial, TALER_MERCHANTDB_KycOutdatedCallback kyc_cb, void *kyc_cb_cls) { @@ -118,6 +119,7 @@ TALER_MERCHANTDB_iterate_outdated_kyc_statuses ( = GNUNET_TIME_absolute_get (); struct GNUNET_PQ_QueryParam params[] = { GNUNET_PQ_query_param_absolute_time (&now), + GNUNET_PQ_query_param_uint64 (&merchant_serial), GNUNET_PQ_query_param_end }; enum GNUNET_DB_QueryStatus qs; @@ -128,7 +130,7 @@ TALER_MERCHANTDB_iterate_outdated_kyc_statuses ( " out_merchant_id" " ,out_h_wire" " ,out_exchange_url" - " FROM merchant.account_kyc_get_outdated($1)"); + " FROM merchant.account_kyc_get_outdated($1, $2)"); qs = GNUNET_PQ_eval_prepared_multi_select ( pg->conn, "iterate_outdated_kyc_statuses", diff --git a/src/backenddb/iterate_outdated_kyc_statuses.sql b/src/backenddb/iterate_outdated_kyc_statuses.sql @@ -16,7 +16,8 @@ DROP FUNCTION IF EXISTS merchant.account_kyc_get_outdated; CREATE FUNCTION merchant.account_kyc_get_outdated( - IN in_now INT8 + IN in_now INT8, + IN in_merchant_serial INT8 ) RETURNS TABLE( out_merchant_id TEXT, @@ -33,6 +34,7 @@ BEGIN SELECT merchant_serial ,merchant_id FROM merchant.merchant_instances + WHERE merchant_serial=in_merchant_serial LOOP s := 'merchant_instance_' || rec.merchant_serial::TEXT; BEGIN @@ -41,7 +43,7 @@ BEGIN EXECUTE format('SELECT ma.h_wire AS h_wire, kyc.exchange_url AS exchange_url' ' FROM %I.merchant_kyc kyc' ' JOIN %I.merchant_accounts ma USING (account_serial)' - ' WHERE kyc.next_kyc_poll < $1' + ' WHERE kyc.next_kyc_poll <= $1 AND ma.active' ' ORDER BY kyc.next_kyc_poll ASC', s, s) USING in_now LOOP @@ -58,6 +60,5 @@ BEGIN END LOOP; END $FN$; -COMMENT ON FUNCTION merchant.account_kyc_get_outdated(INT8) - IS 'Returns one row per outdated KYC entry across all instance schemas.' - ' An entry is outdated if its next_kyc_poll value is less than in_now.'; +COMMENT ON FUNCTION merchant.account_kyc_get_outdated(INT8, INT8) + IS 'Returns due KYC entries for active accounts of the specified instance only.'; diff --git a/src/backenddb/set_tos_accepted_early.sql b/src/backenddb/set_tos_accepted_early.sql @@ -0,0 +1,61 @@ +-- +-- This file is part of TALER +-- Copyright (C) 2026 Taler Systems SA +-- +-- 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 +-- Foundation; either version 3, or (at your option) any later version. +-- +-- TALER is distributed in the hope that it will be useful, but WITHOUT ANY +-- WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR +-- A PARTICULAR PURPOSE. See the GNU General Public License for more details. +-- +-- You should have received a copy of the GNU General Public License along with +-- TALER; see the file COPYING. If not, see <http://www.gnu.org/licenses/> +-- + + +CREATE OR REPLACE FUNCTION merchant_do_set_tos_accepted_early ( + IN in_exchange_url TEXT, + IN in_tos_version TEXT, -- NULL clears the early acceptance + IN in_account_event_prefix BYTEA, -- event header and merchant public key + IN in_instance_notify TEXT) +RETURNS BOOL +LANGUAGE plpgsql +AS $$ +DECLARE + my_h_wire BYTEA; +BEGIN + IF in_tos_version IS NULL + THEN + DELETE FROM merchant_tos_accepted + WHERE exchange_url=in_exchange_url; + ELSE + INSERT INTO merchant_tos_accepted (exchange_url, tos_version) + VALUES (in_exchange_url, in_tos_version) + ON CONFLICT (exchange_url) DO UPDATE + SET tos_version=EXCLUDED.tos_version + WHERE merchant_tos_accepted.tos_version IS DISTINCT FROM EXCLUDED.tos_version; + END IF; + IF NOT FOUND + THEN + RETURN FALSE; + END IF; + + -- Acceptance is per exchange and instance, and can appear in the KYC + -- response for any active account, including one without a cached status. + -- Queue notifications in the same transaction as the actual change. + PERFORM pg_notify (lower (in_instance_notify), ''); + FOR my_h_wire IN + SELECT h_wire FROM merchant_accounts WHERE active + LOOP + -- GNUNET_PQ_get_event_notify_channel: X followed by Crockford base32 + -- of the first 32 bytes of SHA-512 over the packed event. The C caller + -- supplies its header and merchant public key; append this account. + PERFORM pg_notify ( + lower ('X' || merchant.base32_crockford ( + substring (sha512 (in_account_event_prefix || my_h_wire) FROM 1 FOR 32))), + ''); + END LOOP; + RETURN TRUE; +END $$; diff --git a/src/backenddb/sql-schema/meson.build b/src/backenddb/sql-schema/meson.build @@ -53,6 +53,7 @@ sql_instance_procedures = [ '../update_product.sql', '../insert_kyc_status.sql', '../insert_kyc_failure.sql', + '../set_tos_accepted_early.sql', '../update_category.sql', '../update_product_group.sql', '../update_money_pot.sql', diff --git a/src/backenddb/test_kyc_refresh.py b/src/backenddb/test_kyc_refresh.py @@ -0,0 +1,600 @@ +#!/usr/bin/env python3 + +# This file is part of TALER +# Copyright (C) 2026 Taler Systems SA +# +# 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 +# Foundation; either version 3, or (at your option) any later version. +# +# TALER is distributed in the hope that it will be useful, but WITHOUT ANY +# WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR +# A PARTICULAR PURPOSE. See the GNU General Public License for more details. +# +# You should have received a copy of the GNU General Public License along with +# TALER; see the file COPYING. If not, see <http://www.gnu.org/licenses/> + +"""Exercise KYC refreshes with real PostgreSQL notifications and merchant HTTPD. + +Requires PostgreSQL server tools and Python psycopg2. All database state and +HTTP listeners belong to the fixture. No installed database is used. +""" + +import base64 +from concurrent.futures import ThreadPoolExecutor +import hashlib +import json +from pathlib import Path +import select +import shutil +import socket +import struct +import subprocess +import sys +import tempfile +import time +import unittest +from urllib.parse import urlencode +from urllib.request import Request, urlopen + +from test_order_sequence_migrations import Database, postgres_cluster, run + + +ALPHABET = str.maketrans("ABCDEFGHIJKLMNOPQRSTUVWXYZ234567", + "0123456789ABCDEFGHJKMNPQRSTVWXYZ") +EXCHANGE = "http://localhost:1/" +OTHER_EXCHANGE = "http://localhost:2/" +# Synthetic exchange master key (private seed is 31 zero bytes followed by 1) +# and its signature over the account below with no restrictions or gateways. +MASTER_PUB = bytes.fromhex( + "4cb5abf6ad79fbf5abbccafcc269d85cd2651ed4b885b5869f241aedf0a5ba29") +ACCOUNT_SIG = bytes.fromhex( + "42063a0cc9152e9896bb23b0446e14d9bf6e072fbbdcbf3097ac640eba3744ee" + "817368fe93ae567c98ad8f5c27ab5022c82bf73cd024a52baa4e2920bca01806") +FOREVER = 9223372036854775807 +WIRE = bytes([42]) * 64 +TOKEN = bytes([43]) * 32 + + +def encode(data): + return base64.b32encode(data).decode().rstrip("=").translate(ALPHABET) + + +def channel(event_type, body=b""): + header = struct.pack("!HH", 4 + len(body), event_type) + return "x" + encode(hashlib.sha512(header + body).digest()[:32]).lower() + + +def public_key(instance): + return instance.to_bytes(32, "big") + + +# Use the existing protocol event's documented channel, independent of its enum. +REFRESH = "xj40p0cfmz0dt6sfz70vrq19kg1hp6aj1q6vcdzn4n2fgpsag4kdg" + + +class KycRefresh(unittest.TestCase): + def sql(self, statement, params=()): + with self.db.cursor() as cur: + cur.execute(statement, params or None) + return cur.fetchall() if cur.description else [] + + def events(self, wait=0): + deadline = time.monotonic() + wait + events = [] + while True: + self.listener.poll() + events.extend(self.listener.notifies) + self.listener.notifies.clear() + remaining = deadline - time.monotonic() + if remaining <= 0: + return events + select.select([self.listener], [], [], remaining) + + def setUp(self): + self.db = self.connect() + self.listener = self.connect() + self.listener.autocommit = True + with self.listener.cursor() as cur: + for event in [REFRESH] + [channel(t, public_key(i) + suffix) + for i in (1, 2) + for t, suffix in ((1115, b""), (1113, WIRE))]: + cur.execute('LISTEN "' + event + '"') + self.sql("SET search_path TO merchant_instance_1") + # Direct reset avoids generating status events before each scenario. + for i in (1, 2): + self.sql(f'SET search_path TO merchant_instance_{i}') + self.sql("DELETE FROM merchant_tos_accepted") + self.sql(f"UPDATE merchant_instance_{i}.merchant_kyc SET " + "kyc_ok=true, exchange_http_status=200, exchange_ec_code=0, " + "aml_review=false, jaccount_limits='[]', last_rule_gen=1, " + "access_token=%s, next_kyc_poll=%s, kyc_backoff=0", + (TOKEN, FOREVER)) + self.sql('SET search_path TO merchant_instance_1') + self.events() + + def tearDown(self): + self.listener.close() + self.db.close() + + def read(self, refresh, instance=1, exchange=EXCHANGE, wire=WIRE): + self.sql(f"SET search_path TO merchant_instance_{instance}") + return self.sql("SELECT * FROM merchant_do_account_kyc_get_status" + "(%s,%s,%s,%s,%s)", + (time.time_ns() // 1000, exchange, wire, refresh, instance)) + + def store(self, instance=1, exchange=EXCHANGE, **changes): + values = dict(http=200, ec=0, token=TOKEN, limits="[]", aml=False, + ok=True, generation=1, due=FOREVER, backoff=0) + values.update(changes) + self.sql(f"SET search_path TO merchant_instance_{instance}") + return self.sql("SELECT * FROM merchant_do_account_kyc_set_status" + "(%s,%s,%s,%s,%s,%s,%s::jsonb,%s,%s,%s,%s,%s,%s,%s)", + (WIRE, exchange, int(time.time()) * 1000000, + values['http'], values['ec'], values['token'], + values['limits'], values['aml'], values['ok'], + channel(1113, public_key(instance) + WIRE), + channel(1115, public_key(instance)), values['generation'], + values['due'], values['backoff'])) + + def test_read_only_and_transactional_refresh(self): + self.assertEqual(1, len(self.read(False))) + self.assertEqual([], self.events()) + self.assertEqual([(FOREVER,)], self.sql("SELECT next_kyc_poll FROM merchant_kyc")) + self.assertIsNone(self.read(True, exchange="http://unmatched.invalid/")[0][2]) + self.assertEqual([], self.events()) + self.db.autocommit = False + self.read(True) + self.assertEqual([], self.events()) + self.db.rollback() + self.assertEqual([], self.events()) + self.db.autocommit = True + self.read(True) + events = self.events(0.1) + self.assertEqual([(REFRESH, encode(struct.pack('!Q', 1)))], + [(e.channel, e.payload) for e in events]) + due = self.sql("SELECT next_kyc_poll FROM merchant_kyc") + self.read(False) + self.assertEqual(due, self.sql("SELECT next_kyc_poll FROM merchant_kyc")) + self.assertEqual([], self.events()) + + def test_status_change_notifications(self): + self.store(due=17, backoff=123) + self.assertEqual([], self.events()) + changes = dict(http=202, ec=123, token=None, limits=None, aml=True, + ok=False, generation=2) + current = {} + for field, value in changes.items(): + with self.subTest(field=field): + current[field] = value + self.store(**current) + self.assertEqual({channel(1113, public_key(1) + WIRE), + channel(1115, public_key(1))}, + {e.channel for e in self.events(0.1)}) + self.store(**current) + self.assertEqual([], self.events()) + self.store() + self.assertEqual(2, len(self.events(0.1))) # Includes NULL -> value. + self.sql("DELETE FROM merchant_kyc") + self.store() + self.assertEqual(2, len(self.events(0.1))) + args = (WIRE, EXCHANGE, int(time.time()) * 1000000, 502, False, + channel(1113, public_key(1) + WIRE), channel(1115, public_key(1))) + for expected in (2, 0): + self.sql("SELECT * FROM merchant_do_account_kyc_set_failed" + "(%s,%s,%s,%s,%s,%s,%s)", args) + self.assertEqual(expected, len(self.events(0.1))) + + def test_refresh_only_accesses_target_schema(self): + self.read(True, instance=1) + self.read(True, instance=2) + self.db.autocommit = False + rows = self.sql("SELECT * FROM merchant.account_kyc_get_outdated(%s,1)", + (time.time_ns() // 1000,)) + self.assertEqual([('test-1', WIRE, EXCHANGE)], + [(i, bytes(w), e) for i, w, e in rows]) + touched = self.sql("SELECT DISTINCT n.nspname FROM pg_locks l " + "JOIN pg_class c ON c.oid=l.relation " + "JOIN pg_namespace n ON n.oid=c.relnamespace " + "WHERE l.pid=pg_backend_pid() " + "AND n.nspname LIKE 'merchant_instance_%'") + self.assertEqual([('merchant_instance_1',)], touched) + self.db.rollback() + self.db.autocommit = True + self.assertEqual([], self.sql( + "SELECT * FROM merchant.account_kyc_get_outdated(%s,999999)", (FOREVER,))) + self.sql("SET search_path TO merchant_instance_1") + self.sql("UPDATE merchant_accounts SET active=false") + try: + self.assertEqual([], self.sql( + "SELECT * FROM merchant.account_kyc_get_outdated(%s,1)", (FOREVER,))) + finally: + self.sql("UPDATE merchant_accounts SET active=true") + + def get(self, instance=1, query=""): + with urlopen(self.url + f"instances/test-{instance}/private/kyc" + query, + timeout=8) as response: + return json.load(response), response.headers['ETag'] + + def wait_for_refresh(self): + deadline = time.monotonic() + 5 + while time.monotonic() < deadline: + if any(e.channel == REFRESH for e in self.events(0.05)): + return + self.fail("HTTP request did not prompt its initial refresh") + + def test_long_poll_rereads_do_not_refresh(self): + self.check_long_poll() + + def test_account_long_poll_rereads_do_not_refresh(self): + self.check_long_poll("&h_wire=" + encode(WIRE)) + + def check_long_poll(self, account_filter=""): + body, etag = self.get() + self.assertEqual('ready', body['kyc_data'][0]['status']) + self.wait_for_refresh() + with ThreadPoolExecutor() as pool: + pending = pool.submit(self.get, 1, "?timeout_ms=5000&lp_not_etag=" + + etag.strip('"') + account_filter) + self.wait_for_refresh() + # A real status change that does not affect the response ETag forces + # an internal reread, which must not request another exchange check. + self.store(generation=2) + self.assertFalse(any(e.channel == REFRESH for e in self.events(0.25))) + self.assertFalse(pending.done()) + # Another instance's change must not wake this long poll either. + self.store(instance=2, http=202, ok=False) + self.assertFalse(any(e.channel == REFRESH for e in self.events(0.25))) + self.assertFalse(pending.done()) + self.store(http=202, ok=False, generation=2) + changed, _ = pending.result(timeout=3) + self.assertEqual('kyc-required', changed['kyc_data'][0]['status']) + self.assertFalse(any(e.channel == REFRESH for e in self.events(0.1))) + self.get() + self.wait_for_refresh() # Separate HTTP requests still refresh. + + def install_keys(self, exchange=EXCHANGE, **changes): + keys = dict( + version='33:0:0', currency='EUR', asset_type='fiat', + master_public_key=encode(MASTER_PUB), + reserve_closing_delay={'d_us': 3600000000}, + list_issue_date={'t_s': int(time.time())}, + global_fees=[], signkeys=[], denominations=[], auditors=[], + wire_fees={}, wads=[], kyc_enabled=True, kyc_swap_tos_acceptance=False, + hard_limits=[], zero_limits=[], stefan_abs='EUR:0', stefan_log='EUR:0', + stefan_lin=0.0, currency_specification=dict( + name='Euro', num_fractional_input_digits=2, + num_fractional_normal_digits=2, num_fractional_trailing_zero_digits=2, + alt_unit_names={'0': 'EUR'}), + accounts=[dict( + payto_uri='payto://x-taler-bank/localhost/exchange?receiver-name=Exchange', + credit_restrictions=[], debit_restrictions=[], master_sig=encode(ACCOUNT_SIG))]) + keys.update(changes) + self.sql("INSERT INTO merchant.merchant_exchange_keys " + "(exchange_url,keys_json,first_retry,expiration_time," + "exchange_http_status,exchange_ec_code) VALUES (%s,%s,0,0,200,0) " + "ON CONFLICT (exchange_url) DO UPDATE SET keys_json=EXCLUDED.keys_json", + (exchange, json.dumps(dict(version=0, exchange_url=exchange, + expire={'t_s': int(time.time()) + 3600}, keys=keys)))) + self.sql("SELECT pg_notify(%s,%s)", + (channel(1110), encode((exchange + '\0').encode()))) + + def test_initial_keys_wake_long_poll(self): + query = '?' + urlencode({'exchange_url': OTHER_EXCHANGE}) + self.store(instance=2, exchange=OTHER_EXCHANGE, + http=404, token=None, limits=None, ok=False, generation=0) + try: + body, etag = self.get(instance=2, query=query) + self.assertTrue(body['kyc_data'][0]['no_keys']) + self.wait_for_refresh() + with ThreadPoolExecutor() as pool: + pending = pool.submit(self.get, 2, query + "&timeout_ms=5000&lp_not_etag=" + + etag.strip('"')) + self.wait_for_refresh() + self.install_keys(exchange=OTHER_EXCHANGE) + body, new_etag = pending.result(timeout=2) + self.assertNotEqual(etag, new_etag) + self.assertFalse(body['kyc_data'][0]['no_keys']) + self.assertEqual('kyc-wire-required', body['kyc_data'][0]['status']) + self.assertEqual([], self.events(0.1)) + finally: + self.sql("SET search_path TO merchant_instance_2") + self.sql("DELETE FROM merchant_kyc WHERE exchange_url=%s", (OTHER_EXCHANGE,)) + + def test_keys_long_poll(self): + self.check_keys_long_poll() + + def test_keys_exchange_long_poll(self): + self.check_keys_long_poll("&" + urlencode({'exchange_url': EXCHANGE})) + + def test_keys_account_long_poll(self): + self.check_keys_long_poll("&h_wire=" + encode(WIRE)) + + def check_keys_long_poll(self, account_filter=""): + status = dict(http=404, token=None, limits=None, ok=False, generation=0) + self.store(**status) + self.install_keys() + # Wait until HTTPD has consumed the key notification. The keys go + # through the production deserializer, including signature validation. + deadline = time.monotonic() + 5 + while True: + body, etag = self.get() + self.wait_for_refresh() + data = body['kyc_data'][0] + if not data['no_keys'] and not data['kyc_swap_tos_acceptance'] and data['limits'] == []: + break + self.assertLess(time.monotonic(), deadline) + time.sleep(0.02) + self.assertEqual('kyc-wire-required', data['status']) + self.assertTrue(data['payto_kycauths']) + cases = [ + ('kyc_swap_tos_acceptance', True), + ('hard_limits', [dict(operation_type='DEPOSIT', threshold='EUR:10', + timeframe={'d_us': 60000000}, soft_limit=False)]), + ('zero_limits', [dict(operation_type='WITHDRAW')]), + ('accounts', []), + ] + changes = {} + with ThreadPoolExecutor() as pool: + for field, value in cases: + with self.subTest(field=field): + pending = pool.submit(self.get, 1, "?timeout_ms=5000&lp_not_etag=" + + etag.strip('"') + account_filter) + self.wait_for_refresh() + # Repeated or unrelated keys do not finish this poll or + # request another exchange KYC check. + self.install_keys(**changes) + self.install_keys(exchange=OTHER_EXCHANGE, kyc_swap_tos_acceptance=True) + self.assertFalse(any(e.channel == REFRESH for e in self.events(0.15))) + self.assertFalse(pending.done()) + changes[field] = value + self.install_keys(**changes) + self.store(**status) # The same cached /kyc-check result. + body, new_etag = pending.result(timeout=2) + self.assertNotEqual(etag, new_etag) + etag = new_etag + data = body['kyc_data'][0] + if field == 'kyc_swap_tos_acceptance': + self.assertTrue(data[field]) + elif field == 'hard_limits': + self.assertEqual('EUR:10', data['limits'][0]['threshold']) + elif field == 'zero_limits': + self.assertTrue(data['limits'][1]['disallowed']) + else: + self.assertEqual('kyc-wire-impossible', data['status']) + self.assertFalse(data.get('payto_kycauths')) + self.assertEqual([], self.events(0.1)) + + def accept_tos(self, version, instance=1): + request = Request(self.url + f"instances/test-{instance}/private/accept-tos-early", + data=json.dumps(dict(exchange_url=EXCHANGE, + tos_version=version)).encode(), + headers={'Content-Type': 'application/json'}) + with urlopen(request, timeout=3) as response: + self.assertEqual(204, response.status) + + def set_tos(self, version): + return self.sql("SELECT merchant_do_set_tos_accepted_early(%s,%s,%s,%s)", + (EXCHANGE, version, struct.pack('!HH', 100, 1113) + public_key(1), + channel(1115, public_key(1)))) + + def test_tos_notifications_are_transactional_and_scoped(self): + # Even accounts without a cached KYC row can report early acceptance. + extra_wire = bytes([44]) * 64 + extra_channel = channel(1113, public_key(1) + extra_wire) + self.sql("INSERT INTO merchant_accounts(h_wire,salt,payto_uri,active) " + "VALUES(%s,%s,'payto://x-taler-bank/localhost/extra?receiver-name=Test',true)", + (extra_wire, bytes(16))) + with self.listener.cursor() as cur: + cur.execute('LISTEN "' + extra_channel + '"') + self.events() + try: + self.db.autocommit = False + self.assertEqual([(True,)], self.set_tos('v1')) + self.assertEqual([], self.events(0.1)) + self.db.rollback() + self.db.autocommit = True + self.assertEqual([], self.events()) + self.assertEqual([], self.sql("SELECT * FROM merchant_tos_accepted")) + for version in ('v1', 'v2', None): + self.assertEqual([(True,)], self.set_tos(version)) + self.assertEqual({channel(1115, public_key(1)), + channel(1113, public_key(1) + WIRE), extra_channel}, + {e.channel for e in self.events(0.1)}) + self.assertEqual([(False,)], self.set_tos(version)) + self.assertEqual([], self.events(0.1)) + finally: + self.sql("DELETE FROM merchant_accounts WHERE h_wire=%s", (extra_wire,)) + + def test_tos_long_poll(self): + self.check_tos_long_poll() + + def test_tos_account_long_poll(self): + self.check_tos_long_poll("&h_wire=" + encode(WIRE)) + + def check_tos_long_poll(self, account_filter=""): + self.store(http=202, ok=False) + self.events(0.1) + body, etag = self.get() + self.assertNotIn('tos_accepted_early', body['kyc_data'][0]) + self.wait_for_refresh() + with ThreadPoolExecutor() as pool: + for version in ('v1', 'v2', None): + with self.subTest(version=version): + pending = pool.submit(self.get, 1, "?timeout_ms=5000&lp_not_etag=" + + etag.strip('"') + account_filter) + self.wait_for_refresh() + if version == 'v2': + self.accept_tos('v1') # An identical acceptance is silent. + self.assertEqual([], self.events(0.1)) + self.accept_tos(version or 'v3', instance=2) + self.assertFalse(any(e.channel == REFRESH for e in self.events(0.1))) + self.assertFalse(pending.done()) + if version is None: + # The actual 409 callback clears acceptance; the following + # unchanged KYC write must not be needed to wake the poll. + self.transition('tos-conflict') + self.store(http=202, ok=False) + else: + self.accept_tos(version) + body, new_etag = pending.result(timeout=2) + self.assertNotEqual(etag, new_etag) + etag = new_etag + self.assertEqual(version, body['kyc_data'][0].get('tos_accepted_early')) + self.assertEqual('kyc-required', body['kyc_data'][0]['status']) + self.assertEqual({channel(1115, public_key(1)), + channel(1113, public_key(1) + WIRE)}, + {e.channel for e in self.events(0.1)}) + + def test_eligibility_loss_wakes_long_poll(self): + body, etag = self.get() + self.assertEqual('ready', body['kyc_data'][0]['status']) + self.wait_for_refresh() + with ThreadPoolExecutor() as pool: + pending = pool.submit(self.get, 1, "?timeout_ms=5000&lp_not_etag=" + + etag.strip('"')) + self.wait_for_refresh() + self.transition('eligibility') + body, new_etag = pending.result(timeout=2) + self.assertNotEqual(etag, new_etag) + # /keys snapshots are injected into kyccheck, while HTTPD has no + # keys in this fixture. Verify that it returns the persisted + # ineligibility error and drops the stale ready status/token. + self.assertEqual('no-exchange-keys', body['kyc_data'][0]['status']) + self.assertEqual(2628, body['kyc_data'][0]['exchange_code']) + self.assertNotIn('access_token', body['kyc_data'][0]) + self.assertEqual({channel(1115, public_key(1)), + channel(1113, public_key(1) + WIRE)}, + {e.channel for e in self.events(0.1)}) + body, _ = self.get(instance=2) + self.assertEqual('ready', body['kyc_data'][0]['status']) + + def test_upgrade_existing_procedure_signature(self): + # Procedure synchronization must replace the old three-argument API. + self.sql("CREATE FUNCTION merchant_do_account_kyc_get_status(bigint,text,bytea) " + "RETURNS integer LANGUAGE sql AS 'SELECT 0'") + self.sql('CALL merchant.sync_instance_procedures(1)') + self.assertEqual([(None,)], self.sql( + "SELECT to_regprocedure('merchant_do_account_kyc_get_status(bigint,text,bytea)')")) + self.assertEqual(1, len(self.read(False))) + + +def main(): + try: + import psycopg2 + except ImportError: + print("KYC refresh integration tests require Python psycopg2") + return 77 + source, build = (Path(arg).resolve() for arg in sys.argv[1:]) + if not shutil.which('pg_config'): + return 77 + bindir = Path(run(['pg_config', '--bindir'])) + with postgres_cluster(bindir, max_locks=1024) as env, tempfile.TemporaryDirectory( + prefix='merchant-kyc-') as tmp: + sql_dir = build / 'src/backenddb/sql-schema' + db = Database('talercheck', bindir, env, sql_dir) + run(['createdb', 'talercheck'], env=env) + db.apply_file(source / 'src/backenddb/sql-schema/versioning.sql') + for migration in sorted(sql_dir.glob('merchant-????.sql')): + db.apply_file(migration) + db.apply_file(sql_dir / 'global_procedures.sql') + db.apply_file(sql_dir / 'instance_procedures.sql') + for i in range(1, 201): + db.sql(f""" + INSERT INTO merchant.merchant_instances + (merchant_serial, merchant_id, merchant_name, merchant_pub, + merchant_priv, address, jurisdiction, default_wire_transfer_delay, + default_pay_delay, use_stefan) + VALUES ({i}, 'test-{i}', 'Test', decode('{public_key(i).hex()}','hex'), + decode('{public_key(i).hex()}','hex'), '{{}}', '{{}}', 1, 1, false); + SET search_path TO merchant_instance_{i}; + INSERT INTO merchant_instance_{i}.merchant_accounts + (h_wire, salt, payto_uri, active) + VALUES (decode('{WIRE.hex()}','hex'), decode(repeat('00',16),'hex'), + 'payto://x-taler-bank/localhost/account-{i}?receiver-name=Test', true); + INSERT INTO merchant_instance_{i}.merchant_kyc + (account_serial,exchange_url,kyc_timestamp,kyc_ok,access_token, + exchange_http_status,jaccount_limits,last_rule_gen,next_kyc_poll) + VALUES (1,'{EXCHANGE}',0,true,decode('{TOKEN.hex()}','hex'),200,'[]',1,{FOREVER}); + """) + def connect(): + conn = psycopg2.connect(dbname='talercheck', user='postgres', + host=env['PGHOST'], port=env['PGPORT']) + conn.autocommit = True + return conn + KycRefresh.connect = staticmethod(connect) + with socket.socket() as sock: + sock.bind(('127.0.0.1', 0)) + port = sock.getsockname()[1] + KycRefresh.url = f'http://127.0.0.1:{port}/' + config = Path(tmp, 'merchant.conf') + config.write_text(f""" +@INLINE@ {source / "src/backend/merchant.conf"} +@INLINE@ {source / "src/util/currencies.conf"} +[merchant] +CURRENCY=EUR +SERVE=tcp +PORT={port} +BIND_TO=127.0.0.1 +BASE_URL={KycRefresh.url} +[merchantdb-postgres] +CONFIG=postgres:///talercheck +SQL_DIR={sql_dir}/ +[taler] +CURRENCY=EUR +[merchant-exchange-kyc-test] +EXCHANGE_BASE_URL={EXCHANGE} +CURRENCY=EUR +MASTER_KEY={encode(MASTER_PUB)} +[merchant-exchange-other-test] +EXCHANGE_BASE_URL={OTHER_EXCHANGE} +CURRENCY=EUR +MASTER_KEY={encode(MASTER_PUB)} +[merchant-exchange-kudos] +DISABLED=YES +[merchant-exchange-chf] +DISABLED=YES +""") + env['LD_LIBRARY_PATH'] = ':'.join(str(build / 'src' / d) + for d in ('backenddb', 'util', 'bank', 'lib')) \ + + ':' + env.get('LD_LIBRARY_PATH', '') + prefix = Path(tmp, 'prefix') + resources = prefix / 'share/taler-merchant' + resources.mkdir(parents=True) + (resources / 'templates').symlink_to(source / 'src/frontend') + (resources / 'spa').symlink_to(source / 'contrib/spa') + env['TALER_MERCHANT_PREFIX'] = str(prefix) + KycRefresh.transition = staticmethod(lambda mode: run( + [str(build / 'src/backend/test_merchant_kyccheck'), str(config), mode], env=env)) + with open(Path(tmp, 'httpd.log'), 'w+') as log: + process = subprocess.Popen([str(build / 'src/backend/taler-merchant-httpd'), + '-c', str(config), '-L', 'INFO'], + env=env, stdout=log, stderr=log) + try: + deadline = time.monotonic() + 10 + while time.monotonic() < deadline: + if process.poll() is not None: + raise RuntimeError('merchant HTTPD exited at startup') + try: + with urlopen(KycRefresh.url + 'config', timeout=0.2): + break + except OSError: + time.sleep(0.05) + else: + raise RuntimeError('merchant HTTPD did not start') + suite = unittest.defaultTestLoader.loadTestsFromTestCase(KycRefresh) + result = unittest.TextTestRunner(verbosity=2).run(suite) + return 0 if result.wasSuccessful() else 1 + finally: + process.terminate() + try: + process.wait(timeout=10) + except subprocess.TimeoutExpired: + process.kill() + process.wait() + log.seek(0) + Path(build / 'kyc-refresh-httpd.log').write_text(log.read()) + + +if __name__ == '__main__': + sys.exit(main()) diff --git a/src/backenddb/test_merchantdb.c b/src/backenddb/test_merchantdb.c @@ -8344,6 +8344,22 @@ kyc_event_cb (void *cls, (*fired)++; } +/** + * Check the notification encoding used by the SQL refresh request. + */ +static void +kyc_refresh_event_cb (void *cls, + const void *extra, + size_t extra_size) +{ + uint64_t serial; + + GNUNET_assert (sizeof (serial) == extra_size); + memcpy (&serial, extra, sizeof (serial)); + GNUNET_assert (pg->current_merchant_serial == GNUNET_ntohll (serial)); + kyc_event_cb (cls, extra, extra_size); +} + /** * Test inserting and updating a failed KYC check. @@ -8364,11 +8380,13 @@ test_insert_kyc_failure ( .header.size = htons (sizeof (account_event)), .header.type = htons ( TALER_DBEVENT_MERCHANT_EXCHANGE_KYC_STATUS_CHANGED), + .merchant_pub = instance->merchant_pub, .h_wire = account->h_wire }; - struct GNUNET_DB_EventHeaderP general_event = { - .size = htons (sizeof (general_event)), - .type = htons (TALER_DBEVENT_MERCHANT_KYC_STATUS_CHANGED) + struct TALER_MERCHANTDB_InstanceKycStatusChangeEventP general_event = { + .header.size = htons (sizeof (general_event)), + .header.type = htons (TALER_DBEVENT_MERCHANT_KYC_STATUS_CHANGED), + .merchant_pub = instance->merchant_pub }; struct GNUNET_DB_EventHeaderP refresh_event = { .size = htons (sizeof (refresh_event)), @@ -8404,7 +8422,7 @@ test_insert_kyc_failure ( } general_eh = TALER_MERCHANTDB_event_listen ( pg, - &general_event, + &general_event.header, GNUNET_TIME_UNIT_FOREVER_REL, &kyc_event_cb, &general_events); @@ -8419,7 +8437,7 @@ test_insert_kyc_failure ( pg, &refresh_event, GNUNET_TIME_UNIT_FOREVER_REL, - &kyc_event_cb, + &kyc_refresh_event_cb, &refresh_events); if (NULL == refresh_eh) { @@ -8460,6 +8478,7 @@ test_insert_kyc_failure ( instance->instance.id, &account->h_wire, exchange_url, + true, &kyc_failure_check, &expected)) { @@ -8525,6 +8544,7 @@ test_insert_kyc_failure ( instance->instance.id, &account->h_wire, exchange_url, + true, &kyc_failure_check, &expected)) { @@ -8550,6 +8570,34 @@ test_insert_kyc_failure ( refresh_events); goto cleanup; } + /* A notification-driven reread must neither request another refresh nor + move the persisted due time. Repeated identical failures are silent. */ + { + uint64_t before; + uint64_t after; + + TEST_RET_ON_FAIL (query_sql_num ( + "SELECT next_kyc_poll AS num FROM merchant_kyc" + " WHERE exchange_url='https://exchange2.com/'", &before)); + expected.called = 0; + GNUNET_assert (1 == TALER_MERCHANTDB_iterate_kyc_statuses ( + pg, instance->instance.id, &account->h_wire, + exchange_url, false, &kyc_failure_check, &expected)); + GNUNET_assert ((! expected.failed) && (1 == expected.called)); + TEST_RET_ON_FAIL (query_sql_num ( + "SELECT next_kyc_poll AS num FROM merchant_kyc" + " WHERE exchange_url='https://exchange2.com/'", &after)); + GNUNET_assert (before == after); + GNUNET_assert (GNUNET_DB_STATUS_SUCCESS_ONE_RESULT == + TALER_MERCHANTDB_insert_kyc_failure ( + pg, instance->instance.id, &account->h_wire, + exchange_url, GNUNET_TIME_timestamp_get (), + expected.exchange_http_status, false)); + GNUNET_PQ_event_do_poll (pg->conn); + GNUNET_assert (2 == refresh_events); + GNUNET_assert (2 == account_events); + GNUNET_assert (2 == general_events); + } ret = 0; cleanup: @@ -8646,6 +8694,7 @@ test_kyc (void) instance.instance.id, &account.h_wire, "https://exchange.net/", + true, &kyc_status_ok, &fail)); TEST_RET_ON_FAIL (fail); @@ -8655,6 +8704,7 @@ test_kyc (void) instance.instance.id, NULL, "https://exchange2.com/", + true, &kyc_status_fail, &fail)); TEST_RET_ON_FAIL (fail); @@ -8664,6 +8714,7 @@ test_kyc (void) instance.instance.id, NULL, NULL, + true, &kyc_status_fail, &fail)); TEST_RET_ON_FAIL (fail); @@ -8673,6 +8724,7 @@ test_kyc (void) instance.instance.id, NULL, NULL, + true, &kyc_status_ok, &fail)); TEST_RET_ON_FAIL (fail); diff --git a/src/backenddb/test_order_sequence_migrations.py b/src/backenddb/test_order_sequence_migrations.py @@ -47,7 +47,7 @@ def run(command, **kwargs): @contextmanager -def postgres_cluster(bindir): +def postgres_cluster(bindir, *, max_locks=64): """Keep all test data in a disposable server, accessible only by Unix socket.""" with tempfile.TemporaryDirectory(prefix="merchant-seq-", dir="/tmp") as tmp: data = Path(tmp) / "data" @@ -73,7 +73,9 @@ def postgres_cluster(bindir): try: run([str(bindir / "pg_ctl"), "-D", str(data), "-l", str(Path(tmp) / "server.log"), - "-o", f"-F -k {tmp} -c listen_addresses=''", "-w", "start"], + "-o", f"-F -k {tmp} -c listen_addresses='' " + f"-c max_locks_per_transaction={max_locks}", + "-w", "start"], env=env, **server_options) yield env finally: diff --git a/src/include/merchant-database/iterate_kyc_statuses.h b/src/include/merchant-database/iterate_kyc_statuses.h @@ -59,9 +59,8 @@ typedef void const json_t *jlimits); /** - * Check an instance's account's KYC status. Triggers - * a request to taler-merchant-kyccheck to get a - * KYC status update as a side-effect! + * Check an instance's accounts' KYC status, optionally requesting an + * exchange refresh. Reads after a notification must not request another refresh. * * @param pg database context * @param merchant_id merchant backend instance ID @@ -69,6 +68,7 @@ typedef void * NULL to check all accounts of the merchant * @param exchange_url base URL of the exchange to check, * NULL to check all exchanges + * @param request_refresh request an exchange check for the returned records * @param kyc_cb KYC status callback to invoke * @param kyc_cb_cls closure for @a kyc_cb * @return database result code @@ -78,6 +78,7 @@ TALER_MERCHANTDB_iterate_kyc_statuses (struct TALER_MERCHANTDB_PostgresContext * const char *merchant_id, const struct TALER_MerchantWireHashP *h_wire, const char *exchange_url, + bool request_refresh, TALER_MERCHANTDB_KycCallback kyc_cb, void *kyc_cb_cls); diff --git a/src/include/merchant-database/iterate_outdated_kyc_statuses.h b/src/include/merchant-database/iterate_outdated_kyc_statuses.h @@ -48,12 +48,14 @@ typedef void * Find accounts requiring KYC checks. * * @param pg database context + * @param merchant_serial instance whose due KYC records to return * @param kyc_cb status callback to invoke * @param kyc_cb_cls closure for @a kyc_cb * @return database result code */ enum GNUNET_DB_QueryStatus TALER_MERCHANTDB_iterate_outdated_kyc_statuses (struct TALER_MERCHANTDB_PostgresContext *pg, + uint64_t merchant_serial, TALER_MERCHANTDB_KycOutdatedCallback kyc_cb, void *kyc_cb_cls); diff --git a/src/include/merchantdb_lib.h b/src/include/merchantdb_lib.h @@ -45,12 +45,28 @@ struct TALER_MERCHANTDB_MerchantKycStatusChangeEventP struct GNUNET_DB_EventHeaderP header; /** + * Instance owning the account. + */ + struct TALER_MerchantPublicKeyP merchant_pub; + + /** * Salted hash of the affected account. */ struct TALER_MerchantWireHashP h_wire; }; /** + * Instance-wide KYC status notification. Type is + * TALER_DBEVENT_MERCHANT_KYC_STATUS_CHANGED. + */ +struct TALER_MERCHANTDB_InstanceKycStatusChangeEventP +{ + struct GNUNET_DB_EventHeaderP header; + + struct TALER_MerchantPublicKeyP merchant_pub; +}; + +/** * Event triggered when an order is paid. */ struct TMH_OrderPayEventP