commit 43a28f56d4dfd8237c6c29c19de14ea526ec5245
parent 16b9ff784bda6409ded13a8cfb3d0ce1b0c759b0
Author: Antoine A <>
Date: Wed, 14 Oct 2026 13:16:19 +0200
wise: improve worker to handle inconsistent statement pagination
Diffstat:
14 files changed, 531 insertions(+), 347 deletions(-)
diff --git a/Cargo.lock b/Cargo.lock
@@ -741,9 +741,9 @@ dependencies = [
[[package]]
name = "equivalent"
-version = "1.0.2"
+version = "1.0.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "877a4ace8713b0bcf2a4e7eec82529c029f1d0619886d18145fea96c3ffe5c0f"
+checksum = "00d174d5400e5e8fd687ad1049e2f578285fa914201b1af7e8b112a4546bd826"
[[package]]
name = "errno"
@@ -1950,9 +1950,9 @@ dependencies = [
[[package]]
name = "rustls"
-version = "0.23.45"
+version = "0.23.46"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "0d41d731c7d2f962d1ccc364cec258de3c0e93b38c2fb3ba97ac74513048d634"
+checksum = "48e13bd8c0e9365c43cfa5c9e8f9ad49d3c8444926c9aac819e0e4dc503c8fdf"
dependencies = [
"aws-lc-rs",
"log",
@@ -1965,9 +1965,9 @@ dependencies = [
[[package]]
name = "rustls-native-certs"
-version = "0.8.4"
+version = "0.8.5"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "dab5152771c58876a2146916e53e35057e1a4dfa2b9df0f0305b07f611fdea4d"
+checksum = "4a073f5dc7469f984c52ad2752b63b0807745133b6de880b7b64c1ac4c48aec4"
dependencies = [
"openssl-probe",
"rustls-pki-types",
@@ -2716,7 +2716,6 @@ dependencies = [
"aws-lc-rs",
"clap",
"compact_str",
- "futures-util",
"http-client",
"hyper",
"jiff",
@@ -2809,9 +2808,9 @@ checksum = "fd3ca314f692efd6c868f8408f53fe444634a845f96c028b97d35f6a1f79f0ee"
[[package]]
name = "tokio"
-version = "1.53.2"
+version = "1.53.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "e95f91fcc7a621e8b030f6aa23c71fe9838ae2fb4d8118b75602a328f5144044"
+checksum = "ce3335fa71841cda333a58d7615b03901380ecf09d59b3296d21f8bbac0dde4e"
dependencies = [
"bytes",
"libc",
@@ -3493,18 +3492,18 @@ dependencies = [
[[package]]
name = "zerocopy"
-version = "0.8.62"
+version = "0.8.63"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "86502bf56ac7c77571a32e2647bb2a15894565e981fb2a48d7bde2d91c965a9d"
+checksum = "e5fe1f8f1b06191a00962174c61aa5005e0bb391a6d80d07e24d115c01a92ed8"
dependencies = [
"zerocopy-derive",
]
[[package]]
name = "zerocopy-derive"
-version = "0.8.62"
+version = "0.8.63"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "5457206954b06561e2608c7e19cf58b1926586d999c246eebe4502f7e2039d1a"
+checksum = "863ad3ac83293fb4d740aedbfdc9240dd8d1a50c1099acd76ce80ce7c7230c7f"
dependencies = [
"proc-macro2",
"quote",
diff --git a/adapters/taler-cyclos/src/setup.rs b/adapters/taler-cyclos/src/setup.rs
@@ -180,6 +180,6 @@ pub async fn setup(cfg: &Config, _reset: bool, client: &http_client::Client) ->
Ok(())
} else {
error!(target: "setup", "Setup failed, config need editing");
- std::process::exit(0);
+ std::process::exit(6);
}
}
diff --git a/adapters/taler-wise/Cargo.toml b/adapters/taler-wise/Cargo.toml
@@ -34,7 +34,5 @@ regex.workspace = true
prometheus-client.workspace = true
uuid = { version = "1.24.0", features = ["serde"] }
-futures-util = { version = "0.3", default-features = false, features = ["alloc"] }
-
[dev-dependencies]
tokio = { workspace = true, features = ["test-util"] }
diff --git a/adapters/taler-wise/db/wise-0001.sql b/adapters/taler-wise/db/wise-0001.sql
@@ -23,10 +23,15 @@ COMMENT ON TYPE taler_amount IS 'Stores an amount, fraction is in units of 1/100
CREATE TABLE balances (
balance_id INT8 PRIMARY KEY,
- last_sync_at INT8 NOT NULL,
- last_sync_success_at INT8 NOT NULL
+ created_at INT8 NOT NULL,
+ early_cursor INT8 NOT NULL,
+ sync_cursor INT8 NOT NULL,
+ checkpoint_cursor INT8 NOT NULL,
+ early_success_at INT8,
+ sync_success_at INT8,
+ checkpoint_success_at INT8
);
-COMMENT ON TABLE balances IS 'Synchronized balances accounts';
+COMMENT ON TABLE balances IS 'Balance creation time, transaction cursors and successful scan times (UTC microseconds)';
CREATE TABLE tx_in (
tx_in_id INT8 PRIMARY KEY GENERATED ALWAYS AS IDENTITY,
diff --git a/adapters/taler-wise/src/config.rs b/adapters/taler-wise/src/config.rs
@@ -102,36 +102,37 @@ pub struct WiseBalance {
}
fn balances(cfg: &Config) -> Result<Vec<WiseBalance>, CfgErr> {
- cfg.sections()
- .filter_map(|s| {
- (|| {
- Ok(if s.name.starts_with("wise-balance-") {
- Some(WiseBalance {
- id: s.number("id").require()?,
- currency: s.currency("currency").require()?,
- payto: map_config!(s, "payto type", "PAYTO_TYPE",
- "iban" => {
- WiseAccount::IBAN(BankID {
- iban: s.parse("IBAN", "iban").require()?,
- bic: s.parse("BIC", "bic").opt()?
- })
- },
- "ach" => {
- WiseAccount::ACH(ACH {
- routing_number: s.parse("ACH routing number", "routing_number").require()?,
- account_number: s.parse("ACH account_number", "account_number").require()?
- })
- },
- )
- .require()?,
+ let mut balances = Vec::new();
+ let mut ids = std::collections::BTreeSet::new();
+ for s in cfg
+ .sections()
+ .filter(|s| s.name.starts_with("wise-balance-"))
+ {
+ let id = s.number("ID").require()?;
+ if !ids.insert(id) {
+ return Err(s.invalid("ID", format!("duplicate balance ID {id}")));
+ }
+ balances.push(WiseBalance {
+ id,
+ currency: s.currency("CURRENCY").require()?,
+ payto: map_config!(s, "payto type", "PAYTO_TYPE",
+ "iban" => {
+ WiseAccount::IBAN(BankID {
+ iban: s.parse("IBAN", "iban").require()?,
+ bic: s.parse("BIC", "bic").opt()?
})
- } else {
- None
- })
- })()
- .transpose()
- })
- .collect()
+ },
+ "ach" => {
+ WiseAccount::ACH(ACH {
+ routing_number: s.parse("ACH routing number", "routing_number").require()?,
+ account_number: s.parse("ACH account_number", "account_number").require()?
+ })
+ },
+ )
+ .require()?,
+ });
+ }
+ Ok(balances)
}
/// taler-wise setup config
@@ -203,6 +204,14 @@ impl WorkerCfg {
if cfg.balances.is_empty() {
config_bail!("Missing configured balance in a section [wise-balance-*]");
}
+ if cfg.frequency.is_zero() {
+ return Err(s.invalid("FREQUENCY", "must be positive"));
+ }
+ if cfg.token.is_empty()
+ || hyper::header::HeaderValue::from_str(&format!("Bearer {}", cfg.token)).is_err()
+ {
+ return Err(s.invalid("TOKEN", "must be a nonempty valid HTTP header value"));
+ }
Ok(cfg)
}
}
diff --git a/adapters/taler-wise/src/constants.rs b/adapters/taler-wise/src/constants.rs
@@ -14,6 +14,11 @@
TALER; see the file COPYING. If not, see <http://www.gnu.org/licenses/>
*/
+use std::time::Duration;
+
use taler_common::config::parser::ConfigSource;
pub const CONFIG_SOURCE: ConfigSource = ConfigSource::new("taler-wise", "wise", "taler-wise");
+
+pub(crate) static HOUR: Duration = Duration::from_secs(60 * 60);
+pub(crate) static DAY: Duration = Duration::from_secs(24 * 60 * 60);
diff --git a/adapters/taler-wise/src/db.rs b/adapters/taler-wise/src/db.rs
@@ -61,6 +61,105 @@ pub async fn dbinit(cfg: &Config, reset: bool) -> anyhow::Result<PgPool> {
Ok(pool)
}
+/// Durable transaction high-water marks and completed scan schedules.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub(crate) struct SyncState {
+ pub early: Timestamp,
+ pub sync: Timestamp,
+ pub checkpoint: Timestamp,
+ pub early_success: Option<Timestamp>,
+ pub sync_success: Option<Timestamp>,
+ pub checkpoint_success: Option<Timestamp>,
+}
+
+pub(crate) async fn load_sync_state(
+ db: &PgPool,
+ balance_id: u32,
+) -> sqlx::Result<Option<SyncState>> {
+ sqlx::query(
+ "SELECT early_cursor, sync_cursor, checkpoint_cursor,
+ early_success_at, sync_success_at, checkpoint_success_at
+ FROM balances WHERE balance_id = $1",
+ )
+ .bind(i64::from(balance_id))
+ .try_map(|r: PgRow| {
+ Ok(SyncState {
+ early: r.try_get_timestamp(0)?,
+ sync: r.try_get_timestamp(1)?,
+ checkpoint: r.try_get_timestamp(2)?,
+ early_success: r.try_get_opt_timestamp(3)?,
+ sync_success: r.try_get_opt_timestamp(4)?,
+ checkpoint_success: r.try_get_opt_timestamp(5)?,
+ })
+ })
+ .fetch_optional(db)
+ .await
+}
+
+pub(crate) async fn initialize_sync_state(
+ db: &PgPool,
+ balance_id: u32,
+ created_at: Timestamp,
+) -> sqlx::Result<()> {
+ serialized!(
+ sqlx::query(
+ "INSERT INTO balances (balance_id, created_at, early_cursor, sync_cursor, checkpoint_cursor)
+ VALUES ($1, $2, $2, $2, $2)
+ ON CONFLICT (balance_id) DO NOTHING",
+ )
+ .bind(i64::from(balance_id))
+ .bind_timestamp(&created_at)
+ .execute(db)
+ )?;
+ Ok(())
+}
+
+pub(crate) async fn reset_sync_state(db: &PgPool) -> sqlx::Result<()> {
+ serialized!(
+ sqlx::query(
+ "UPDATE balances SET
+ early_cursor = created_at,
+ sync_cursor = created_at,
+ checkpoint_cursor = created_at,
+ early_success_at = NULL,
+ sync_success_at = NULL,
+ checkpoint_success_at = NULL",
+ )
+ .execute(db)
+ )?;
+ Ok(())
+}
+
+/// Called after all registrations in the chunk have committed. A failed write
+/// leaves the old cursor available for an idempotent replay on restart.
+pub(crate) async fn save_sync_state(
+ db: &PgPool,
+ balance_id: u32,
+ state: &SyncState,
+) -> sqlx::Result<()> {
+ serialized!(
+ sqlx::query(
+ "UPDATE balances SET
+ early_cursor = GREATEST(early_cursor, $2),
+ sync_cursor = GREATEST(sync_cursor, $3),
+ checkpoint_cursor = GREATEST(checkpoint_cursor, $4),
+ early_success_at = GREATEST(early_success_at, $5),
+ sync_success_at = GREATEST(sync_success_at, $6),
+ checkpoint_success_at = GREATEST(checkpoint_success_at, $7)
+ WHERE balance_id = $1",
+ )
+ .bind(i64::from(balance_id))
+ .bind_timestamp(&state.early)
+ .bind_timestamp(&state.sync)
+ .bind_timestamp(&state.checkpoint)
+ .bind(state.early_success.map(|t| t.as_microsecond()))
+ .bind(state.sync_success.map(|t| t.as_microsecond()))
+ .bind(state.checkpoint_success.map(|t| t.as_microsecond()))
+ .execute(db)
+ )?;
+ Ok(())
+}
+
pub async fn notification_listener(
pool: PgPool,
in_channels: Arc<BTreeMap<u32, Sender<i64>>>,
@@ -330,7 +429,6 @@ pub async fn transfer_unregister(db: &PgPool, req: &Unregistration) -> sqlx::Res
mod test {
use compact_str::{CompactString, format_compact};
- use jiff::Span;
use sqlx::{PgConnection, PgPool, postgres::PgRow};
use taler_api::{db::TypeHelper, notification::dummy_listen, subject::IncomingKey};
use taler_common::{
@@ -345,6 +443,7 @@ mod test {
use taler_macros::db_test;
use crate::{
+ constants::HOUR,
db::{AddIncomingResult, TxIn, incoming_history, register_tx_in, revenue_history},
payto::WiseAccount,
};
@@ -361,7 +460,7 @@ mod test {
.await
.unwrap();
let now = now_sql_stable_ts();
- let later = now + Span::new().hours(1);
+ let later = now + HOUR;
let tx = TxIn {
balance_id: 42,
name: CompactString::const_new("Name"),
diff --git a/adapters/taler-wise/src/main.rs b/adapters/taler-wise/src/main.rs
@@ -45,8 +45,12 @@ enum Command {
#[arg(short, long)]
reset: bool,
},
- /// Check taler-wise config
- Setup,
+ /// Check configuration and initialize balance state
+ Setup {
+ /// Reset all synchronization cursors and schedules to rescan from balance creation
+ #[arg(long)]
+ reset: bool,
+ },
/// Run taler-wise worker
Worker {
/// Execute once and return
@@ -69,9 +73,10 @@ async fn run(cmd: Command, cfg: &Config) -> anyhow::Result<()> {
Command::Dbinit { reset } => {
dbinit(cfg, reset).await?;
}
- Command::Setup => {
+ Command::Setup { reset } => {
+ let pool = pool(cfg).await?;
let client = http_client::client();
- setup::setup(cfg, &client).await?
+ setup::setup(cfg, &pool, &client, reset).await?
}
Command::Serve { check } => {
if check {
diff --git a/adapters/taler-wise/src/setup.rs b/adapters/taler-wise/src/setup.rs
@@ -14,15 +14,23 @@
TALER; see the file COPYING. If not, see <http://www.gnu.org/licenses/>
*/
+use anyhow::Context;
+use sqlx::PgPool;
use taler_common::config::Config;
use tracing::{debug, error, info, warn};
use crate::{
config::{SetupCfg, WiseBalance},
+ db::{initialize_sync_state, reset_sync_state},
wise_api::client::Client,
};
-pub async fn setup(cfg: &Config, client: &http_client::Client) -> anyhow::Result<()> {
+pub async fn setup(
+ cfg: &Config,
+ pool: &PgPool,
+ client: &http_client::Client,
+ reset: bool,
+) -> anyhow::Result<()> {
let cfg = SetupCfg::parse(cfg)?;
let client = Client::new(client, cfg.api, &cfg.token);
@@ -40,7 +48,7 @@ pub async fn setup(cfg: &Config, client: &http_client::Client) -> anyhow::Result
if let Some(id) = cfg.profile_id {
if let Some(profile) = profiles.iter().find(|it| it.id == id) {
debug!(target: "setup", "PROFILE_ID {id} in config is one of:{profiles_fmt}");
- match cfg.name {
+ match &cfg.name {
Some(name) => {
if name != profile.full_name {
warn!(
@@ -72,18 +80,12 @@ pub async fn setup(cfg: &Config, client: &http_client::Client) -> anyhow::Result
let account = accounts
.iter()
.find(|it| it.profile_id == profile_id)
- .unwrap();
-
- let mapped: Vec<_> = account
- .balances
- .iter()
- .map(|it| (it.id, it.currency, it.bank_details.account()))
- .collect();
+ .context("Missing Wise account for configured profile")?;
let balances_fmt = std::fmt::from_fn(|w| {
- for (id, currency, account) in &mapped {
- if let Some(account) = account {
- write!(w, "\n{id} {currency} ({account})")?;
+ for balance in &account.balances {
+ if let Some(payto) = balance.bank_details.account() {
+ write!(w, "\n{} {} ({payto})", balance.id, balance.currency)?;
}
}
Ok(())
@@ -92,44 +94,62 @@ pub async fn setup(cfg: &Config, client: &http_client::Client) -> anyhow::Result
if cfg.balances.is_empty() {
error!(target: "setup", "No balances are configured");
- ready = false
- } else {
- for WiseBalance {
- id,
- currency,
- payto,
- } in &cfg.balances
- {
- match mapped.iter().find(|it| it.0 == *id as u64) {
- Some((_, curr, acc)) => {
- if curr != currency {
- error!(target: "setup", "Balance {id} in config use currency {currency} expected {curr}");
- ready = false;
- }
- if let Some(acc) = acc {
- if acc != payto {
- error!(target: "setup", "Balance {id} in config use payto ({payto}) expected ({acc})");
- ready = false
- }
- } else {
- error!(target: "setup", "Balance {id} in config use payto ({payto}) but it's not supported");
- ready = false
- }
- }
- None => {
- error!(target: "setup", "Unknown balance {id} {currency}");
- ready = false
- }
+ ready = false;
+ }
+ for WiseBalance {
+ id,
+ currency,
+ payto,
+ } in &cfg.balances
+ {
+ let Some(balance) = account.balances.iter().find(|b| b.id == u64::from(*id)) else {
+ error!(target: "setup", "Unknown balance {id} {currency}");
+ ready = false;
+ continue;
+ };
+ if balance.currency != *currency {
+ error!(target: "setup", "Balance {id} in config use currency {currency} expected {}", balance.currency);
+ ready = false;
+ }
+ match balance.bank_details.account() {
+ Some(account) if account == *payto => {}
+ Some(account) => {
+ error!(target: "setup", "Balance {id} in config use payto ({payto}) expected ({account})");
+ ready = false;
+ }
+ None => {
+ error!(target: "setup", "Balance {id} in config use payto ({payto}) but it's not supported");
+ ready = false;
}
}
}
}
- if ready {
- info!(target: "setup", "Setup finished");
- Ok(())
- } else {
+ if !ready {
error!(target: "setup", "Setup failed, config need editing");
- std::process::exit(0);
+ std::process::exit(6);
+ }
+
+ let profile_id = cfg.profile_id.context("Missing PROFILE_ID")?;
+ let balances = client.standard_balances(profile_id).await?;
+ for configured in &cfg.balances {
+ let remote = balances
+ .iter()
+ .find(|remote| {
+ remote.id == u64::from(configured.id) && remote.currency == configured.currency
+ })
+ .with_context(|| {
+ format!(
+ "Missing Wise balance {} {}",
+ configured.id, configured.currency
+ )
+ })?;
+ initialize_sync_state(pool, configured.id, remote.creation_time).await?;
+ }
+ if reset {
+ reset_sync_state(pool).await?;
+ info!(target: "setup", "Synchronization state reset to balance creation");
}
+ info!(target: "setup", "Setup finished");
+ Ok(())
}
diff --git a/adapters/taler-wise/src/wise_api/client.rs b/adapters/taler-wise/src/wise_api/client.rs
@@ -14,14 +14,14 @@
TALER; see the file COPYING. If not, see <http://www.gnu.org/licenses/>
*/
-use std::borrow::Cow;
+use std::{borrow::Cow, time::Duration};
use http_client::{ApiErr, ClientErr, builder::Req, headers::HeaderParser as _};
use hyper::{
Method, StatusCode,
header::{AUTHORIZATION, HeaderValue, RETRY_AFTER},
};
-use jiff::Timestamp;
+use jiff::{SignedDuration, Timestamp, fmt::rfc2822::DateTimeParser};
use serde::de::DeserializeOwned;
use taler_common::types::amount::Currency;
use thiserror::Error;
@@ -37,35 +37,49 @@ pub enum WiseErr {
Status(StatusCode),
#[error("status {0} '{1}'")]
StatusCause(StatusCode, String),
- #[error("Rate limited for {0}")]
- TooManyRequest(String),
+ #[error("rate limited (retry after {0:?})")]
+ TooManyRequest(Option<Duration>),
#[error(transparent)]
Client(#[from] ClientErr),
}
pub type ApiResult<R> = std::result::Result<R, ApiErr<WiseErr>>;
+/// Parse both forms of Retry-After. Past dates are already satisfied.
+fn retry_after(value: &str) -> Option<Duration> {
+ let value = value.trim();
+ let delay = if let Ok(seconds) = value.parse() {
+ Duration::from_secs(seconds)
+ } else {
+ let date = DateTimeParser::new().parse_timestamp(value).ok()?;
+ date.duration_since(Timestamp::now())
+ .max(SignedDuration::ZERO)
+ .unsigned_abs()
+ };
+ std::time::Instant::now().checked_add(delay).map(|_| delay)
+}
+
pub struct Client<'a> {
- env: ApiEnv,
+ base_url: &'a url::Url,
client: &'a http_client::Client,
token: HeaderValue,
}
impl<'a> Client<'a> {
- pub fn new(client: &'a http_client::Client, env: ApiEnv, token: &'a str) -> Self {
+ pub fn new(client: &'a http_client::Client, env: ApiEnv, token: &str) -> Self {
Self {
client,
- env,
+ base_url: env.api(),
token: HeaderValue::from_str(&format!("Bearer {token}")).unwrap(),
}
}
fn request(&self, method: Method, path: impl Into<Cow<'static, str>>) -> Req {
- Req::new(self.client, method, self.env.api(), path)
+ Req::new(self.client, method, self.base_url, path)
}
async fn send<T: DeserializeOwned>(&self, req: Req) -> ApiResult<T> {
let (ctx, res) = req
- .header(AUTHORIZATION, self.token.clone())
+ .sensitive_header(AUTHORIZATION, self.token.clone())
.send()
.await
.map_err(|(ctx, e)| ctx.wrap(e.into()))?;
@@ -74,9 +88,9 @@ impl<'a> Client<'a> {
// Parse json
res.json().await.map_err(|e| ctx.wrap(e.into()))
}
- StatusCode::TOO_MANY_REQUESTS => Err(ctx.wrap(WiseErr::TooManyRequest(
- res.str_header(RETRY_AFTER).unwrap_or_default(),
- ))),
+ StatusCode::TOO_MANY_REQUESTS => Err(ctx.wrap(WiseErr::TooManyRequest(retry_after(
+ &res.str_header(RETRY_AFTER).unwrap_or_default(),
+ )))),
other => Err(ctx.wrap(WiseErr::Status(other))),
}
}
@@ -124,6 +138,7 @@ impl<'a> Client<'a> {
Method::GET,
format!("/v1/profiles/{profile_id}/balance-statements/{balance_id}/statement.json"),
)
+ .query("type", "COMPACT")
.query("currency", currency)
.query("intervalStart", start)
.query("intervalEnd", end),
diff --git a/adapters/taler-wise/src/wise_api/types.rs b/adapters/taler-wise/src/wise_api/types.rs
@@ -168,12 +168,14 @@ pub struct WiseAmount {
pub currency: Currency,
}
-impl From<WiseAmount> for Amount {
- fn from(val: WiseAmount) -> Self {
- Amount::new_decimal(
+impl TryFrom<WiseAmount> for Amount {
+ type Error = taler_common::types::amount::ParseDecimalErr;
+
+ fn try_from(val: WiseAmount) -> Result<Self, Self::Error> {
+ Ok(Amount::new_decimal(
&val.currency,
- Decimal::from_str(&val.value.to_compact_string()).unwrap(),
- )
+ Decimal::from_str(&val.value.to_compact_string())?,
+ ))
}
}
@@ -195,7 +197,10 @@ pub struct Details {
#[serde(rename = "type")]
pub ty: CompactString,
pub description: CompactString,
+ #[serde(default)]
pub sender_account: CompactString,
+ #[serde(default)]
pub sender_name: CompactString,
+ #[serde(default)]
pub payment_reference: CompactString,
}
diff --git a/adapters/taler-wise/src/worker.rs b/adapters/taler-wise/src/worker.rs
@@ -16,22 +16,27 @@
use std::{sync::LazyLock, time::Duration};
-use futures_util::future::{join_all, try_join_all};
use http_client::ApiErr;
-use jiff::Timestamp;
+use jiff::{SignedDuration, Span, Timestamp, tz::TimeZone};
use regex::Regex;
use sqlx::PgPool;
use taler_api::subject::{IncomingSubject, parse_incoming_unstructured};
-use taler_common::{ExpoBackoffDecorr, config::Config, types::payto::BankID};
-use tracing::{error, info, trace, warn};
+use taler_common::{
+ ExpoBackoffDecorr,
+ config::Config,
+ types::{payto::BankID, utils::now_sql_stable_ts},
+};
+use tokio::time::Instant;
+use tracing::{debug, error, info, trace, warn};
use crate::{
config::{WiseBalance, WorkerCfg},
- db::{AddIncomingResult, TxIn, register_tx_in},
+ constants::{DAY, HOUR},
+ db::{AddIncomingResult, SyncState, TxIn, load_sync_state, register_tx_in, save_sync_state},
payto::WiseAccount,
wise_api::{
client::{Client, WiseErr},
- types::Direction,
+ types::{Direction, Tx},
},
};
@@ -41,57 +46,69 @@ pub enum WorkerError {
Db(#[from] sqlx::Error),
#[error(transparent)]
Api(#[from] ApiErr<WiseErr>),
+ #[error("invalid Wise statement: {0}")]
+ InvalidStatement(String),
+ #[error("balance {0} has no synchronization state; run taler-wise setup")]
+ UninitializedBalance(u32),
}
-pub type WorkerResult = Result<(), WorkerError>;
+pub type WorkerResult<T = ()> = Result<T, WorkerError>;
-fn parse_account(account_str: &str) -> Option<WiseAccount> {
- static IBAN_BIC_PATTERN: LazyLock<Regex> =
- LazyLock::new(|| Regex::new(r"^\(([A-Z0-9]{8,11})\) ([A-Z0-9]+)$").unwrap());
+#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)]
+enum Tier {
+ Early,
+ Sync,
+ Checkpoint,
+}
- if let Some(caps) = IBAN_BIC_PATTERN.captures(account_str) {
- let bic = caps[1].parse().ok()?;
- let iban = caps[2].parse().ok()?;
+fn remaining(last: Option<Timestamp>, period: Duration, now: Timestamp) -> Duration {
+ last.map_or(Duration::ZERO, |last| {
+ let deadline = last + period;
+ now.duration_until(deadline)
+ .max(SignedDuration::ZERO)
+ .unsigned_abs()
+ })
+}
- return Some(WiseAccount::IBAN(BankID {
- iban,
- bic: Some(bic),
- }));
+impl SyncState {
+ fn advance(&mut self, tier: Tier, latest: Timestamp) {
+ self.early = self.early.max(latest);
+ if tier >= Tier::Sync {
+ self.sync = self.sync.max(latest - 2 * HOUR);
+ }
+ if tier >= Tier::Checkpoint {
+ self.checkpoint = self.checkpoint.max(latest - 2 * DAY);
+ }
}
- None
-}
-
-/// Each balance owns its retry schedule. A slow or failing balance must not
-/// delay other balances, and success elsewhere must not reset its backoff.
-async fn poll_balance(
- balance_id: u32,
- frequency: Duration,
- transient: bool,
- mut sync: impl AsyncFnMut() -> WorkerResult,
-) -> WorkerResult {
- let mut jitter = ExpoBackoffDecorr::default();
- loop {
- let result = sync().await;
- if transient {
- return result;
+ fn complete(&mut self, tier: Tier, now: Timestamp) {
+ self.early_success = Some(now);
+ if tier >= Tier::Sync {
+ self.sync_success = Some(now);
}
- let delay = match result {
- Ok(()) => {
- jitter.reset();
- frequency
- }
- Err(err @ WorkerError::Db(sqlx::Error::PoolClosed)) => return Err(err),
- Err(err) => {
- let delay = jitter.backoff();
- error!(target: "worker", balance_id, ?delay, "balance synchronization failed: {err}");
- delay
- }
- };
- tokio::time::sleep(delay).await;
+ if tier >= Tier::Checkpoint {
+ self.checkpoint_success = Some(now);
+ }
+ }
+
+ fn delay(&self, now: Timestamp, frequency: Duration) -> Duration {
+ remaining(self.early_success, frequency, now)
+ .min(remaining(self.sync_success, HOUR, now))
+ .min(remaining(self.checkpoint_success, DAY, now))
}
}
+fn parse_account(account_str: &str) -> Option<WiseAccount> {
+ static IBAN_BIC_PATTERN: LazyLock<Regex> =
+ LazyLock::new(|| Regex::new(r"^\(([A-Z0-9]{8,11})\) ([A-Z0-9]+)$").unwrap());
+
+ let caps = IBAN_BIC_PATTERN.captures(account_str)?;
+ Some(WiseAccount::IBAN(BankID {
+ iban: caps[2].parse().ok()?,
+ bic: Some(caps[1].parse().ok()?),
+ }))
+}
+
pub async fn run_worker(
cfg: &Config,
pool: &PgPool,
@@ -100,24 +117,47 @@ pub async fn run_worker(
) -> anyhow::Result<()> {
let cfg = WorkerCfg::parse(cfg)?;
let client = Client::new(client, cfg.api, &cfg.token);
- let cfg = &cfg;
- let client = &client;
- let workers = cfg.balances.iter().map(|balance| {
- poll_balance(balance.id, cfg.frequency, transient, async move || {
- sync_balance(cfg, balance, pool, client).await
- })
- });
- if transient {
- // Attempt every balance once, even if another balance fails.
- for result in join_all(workers).await {
- result?;
+ work_balances(&cfg, pool, &client, transient).await?;
+ Ok(())
+}
+
+async fn work_balances(
+ cfg: &WorkerCfg,
+ pool: &PgPool,
+ client: &Client<'_>,
+ transient: bool,
+) -> WorkerResult {
+ let mut next = Instant::now();
+ let mut jitter = ExpoBackoffDecorr::default();
+ loop {
+ let mut delay = Duration::MAX;
+ let mut first_error = None;
+ for balance in &cfg.balances {
+ tokio::time::sleep_until(next).await;
+ match sync_balance(cfg, balance, pool, client, transient).await {
+ Ok(wait) => delay = delay.min(wait),
+ Err(err @ WorkerError::Db(sqlx::Error::PoolClosed)) => return Err(err),
+ Err(err) => {
+ let mut wait = jitter.backoff();
+ if let WorkerError::Api(api) = &err
+ && let WiseErr::TooManyRequest(retry_after) = api.err.as_ref()
+ {
+ wait = wait.max(retry_after.unwrap_or_default());
+ }
+ next = Instant::now() + wait;
+ error!(target: "worker", "synchronization pass for {} failed: {err}", balance.id);
+ first_error.get_or_insert(err);
+ }
+ }
+ }
+ if transient {
+ return first_error.map_or(Ok(()), Err);
+ }
+ if first_error.is_none() {
+ jitter.reset();
+ next = Instant::now() + delay;
}
- } else {
- // These are operation loops, not detached tasks: panics and terminal
- // failures still reach the process, cancelling the remaining loops.
- try_join_all(workers).await?;
}
- Ok(())
}
async fn sync_balance(
@@ -125,163 +165,134 @@ async fn sync_balance(
balance: &WiseBalance,
pool: &PgPool,
client: &Client<'_>,
-) -> WorkerResult {
- let stmt = client
- .balance_statement(
- cfg.profile_id,
- balance.id,
- &balance.currency,
- "2026-07-16T00:00:00.000Z".parse().unwrap(),
- Timestamp::now(),
+ transient: bool,
+) -> WorkerResult<Duration> {
+ let mut state = load_sync_state(pool, balance.id)
+ .await?
+ .ok_or(WorkerError::UninitializedBalance(balance.id))?;
+ let now = now_sql_stable_ts();
+ if !transient {
+ let delay = state.delay(now, cfg.frequency);
+ if !delay.is_zero() {
+ return Ok(delay);
+ }
+ }
+ // Statements can miss transactions: early tracks the latest seen, hourly/daily
+ // scans replay with 2h/2d lag. The broadest due scan advances newer tiers only
+ // after inserts succeed. Idempotency makes replay safe.
+ // We currently warn on recovered payments to see if this mechanism is usefull.
+ let (tier, cursor, last, overlap) = if remaining(state.checkpoint_success, DAY, now).is_zero() {
+ (
+ Tier::Checkpoint,
+ state.checkpoint,
+ state.checkpoint_success,
+ 2 * DAY,
+ )
+ } else if remaining(state.sync_success, HOUR, now).is_zero() {
+ (Tier::Sync, state.sync, state.sync_success, 2 * HOUR)
+ } else {
+ (
+ Tier::Early,
+ state.early,
+ state.early_success,
+ Duration::ZERO,
)
- .await?;
- let now = Timestamp::now();
- for tx in stmt.transactions {
- match tx.direction {
- Direction::Debit => {
- // TODO support outgoing transaction
+ };
+ let mut start = cursor.max(last.map_or(cursor, |last| last - overlap));
+ let previous_early = state.early_success.map(|last| state.early.max(last));
+ while start < now {
+ // Fetch at most one month at a time
+ let end = (start.to_zoned(TimeZone::UTC) + Span::new().months(1))
+ .timestamp()
+ .min(now);
+ debug!(target: "worker", "fetching statement for {} {tier:?} from {start} to {end}", balance.id);
+ let stmt = client
+ .balance_statement(cfg.profile_id, balance.id, &balance.currency, start, end)
+ .await?;
+ let latest = stmt.transactions.iter().map(|tx| tx.date).max();
+ for tx in stmt.transactions {
+ if tx.reference_number.trim().is_empty() {
+ return Err(WorkerError::InvalidStatement(
+ "transaction has an empty Wise reference".into(),
+ ));
+ } else if tx.amount.currency != balance.currency {
+ return Err(WorkerError::InvalidStatement(format!(
+ "transaction {} has currency {}, expected {}",
+ tx.reference_number, tx.amount.currency, balance.currency,
+ )));
}
- Direction::Credit => {
- let subject = parse_incoming_unstructured(&tx.details.payment_reference);
- // Parse sender account
- let payto = parse_account(&tx.details.sender_account);
- //
- let t = TxIn {
- balance_id: balance.id,
- wise_ref: Some(tx.reference_number),
- amount: tx.amount.into(),
- subject: tx.details.payment_reference,
- name: tx.details.sender_name,
- debtor: payto,
- value_at: tx.date,
- };
- let subject = &match subject {
- Ok(IncomingSubject::Key(key)) => Some(key),
- Ok(IncomingSubject::AdminBalanceAdjust) | Err(_) => None,
- };
- let failure = match register_tx_in(pool, &t, subject, &now).await? {
- AddIncomingResult::Success { new, .. } => {
- if new {
- info!(target: "worker", "in {t}");
- if t.debtor.is_none() {
- warn!(target: "worker", "Couldn't parse creditor account from '{}'", tx.details.sender_account)
- }
- } else {
- trace!(target: "worker", "in {t} already seen");
- }
- continue;
- }
- AddIncomingResult::ReservePubReuse => "reserve pub reuse",
- AddIncomingResult::UnknownMapping => "unknown mapping",
- AddIncomingResult::MappingReuse => "mapping reuse",
- };
-
- match register_tx_in(pool, &t, &None, &now).await? {
- AddIncomingResult::Success { new, .. } => {
- if new {
- info!(target: "worker", "in {t}: {failure}");
- if t.debtor.is_none() {
- warn!(target: "worker", "Couldn't parse creditor account from '{}'", tx.details.sender_account)
- }
- } else {
- trace!(target: "worker", "in {t} already seen: {failure}");
- }
- continue;
- }
- AddIncomingResult::ReservePubReuse
- | AddIncomingResult::UnknownMapping
- | AddIncomingResult::MappingReuse => unreachable!(),
- };
+ let Some(tx) = register_transaction(pool, balance, tx, now).await? else {
+ continue;
+ };
+ if tier >= Tier::Sync
+ && let Some(early) = previous_early
+ && tx.value_at <= early
+ {
+ warn!(target: "worker", "in (recovered) {tx}");
}
}
+ if let Some(latest) = latest {
+ state.advance(tier, latest);
+ save_sync_state(pool, balance.id, &state).await?;
+ trace!(target: "worker", "saved statement cursors for {}", balance.id);
+ }
+ start = end;
}
- Ok(())
+ state.complete(tier, now);
+ save_sync_state(pool, balance.id, &state).await?;
+ Ok(state.delay(Timestamp::now(), cfg.frequency))
}
-#[cfg(test)]
-mod tests {
- use std::cell::Cell;
-
- use futures_util::FutureExt;
- use tokio::time::Instant;
-
- use super::*;
-
- fn unavailable() -> WorkerError {
- sqlx::Error::Io(std::io::ErrorKind::ConnectionRefused.into()).into()
+async fn register_transaction(
+ pool: &PgPool,
+ balance: &WiseBalance,
+ tx: Tx,
+ now: Timestamp,
+) -> WorkerResult<Option<TxIn>> {
+ if tx.direction == Direction::Debit {
+ // Its timestamp advances the cursors, but outgoing registration is not supported.
+ return Ok(None);
}
-
- #[tokio::test(start_paused = true)]
- async fn failed_balance_retries_without_stalling_other_balances() {
- let attempts = Cell::new(0);
- let healthy = Cell::new(0);
- let start = Instant::now();
- let bad = poll_balance(1, Duration::from_secs(60), false, async || {
- attempts.set(attempts.get() + 1);
- if attempts.get() == 3 {
- Err(sqlx::Error::PoolClosed.into())
- } else {
- Err(unavailable())
- }
- });
- let good = poll_balance(2, Duration::from_millis(100), false, async || {
- healthy.set(healthy.get() + 1);
- Ok(())
- });
- let result = try_join_all([bad.boxed_local(), good.boxed_local()]).await;
- assert!(matches!(
- result,
- Err(WorkerError::Db(sqlx::Error::PoolClosed))
- ));
- assert_eq!(attempts.get(), 3);
- assert!(healthy.get() > attempts.get());
- assert!(start.elapsed() >= Duration::from_millis(800));
+ let subject = match parse_incoming_unstructured(&tx.details.payment_reference) {
+ Ok(IncomingSubject::Key(key)) => Some(key),
+ Ok(IncomingSubject::AdminBalanceAdjust) | Err(_) => None,
+ };
+ let t = TxIn {
+ balance_id: balance.id,
+ wise_ref: Some(tx.reference_number),
+ amount: tx
+ .amount
+ .try_into()
+ .map_err(|err| WorkerError::InvalidStatement(format!("invalid amount: {err}")))?,
+ subject: tx.details.payment_reference,
+ name: tx.details.sender_name,
+ debtor: parse_account(&tx.details.sender_account),
+ value_at: tx.date,
+ };
+ let mut result = register_tx_in(pool, &t, &subject, &now).await?;
+ let failure = match result {
+ AddIncomingResult::Success { .. } => None,
+ AddIncomingResult::ReservePubReuse => Some("reserve pub reuse"),
+ AddIncomingResult::UnknownMapping => Some("unknown mapping"),
+ AddIncomingResult::MappingReuse => Some("mapping reuse"),
+ };
+ if failure.is_some() {
+ result = register_tx_in(pool, &t, &None, &now).await?;
}
-
- #[tokio::test(start_paused = true)]
- async fn success_resets_only_this_balances_backoff() {
- let attempts = Cell::new(0);
- let failed_at = Cell::new(Instant::now());
- let result = poll_balance(1, Duration::from_millis(10), false, async || {
- attempts.set(attempts.get() + 1);
- match attempts.get() {
- 11 => Ok(()),
- 12 => {
- failed_at.set(Instant::now());
- Err(unavailable())
- }
- 13 => Err(sqlx::Error::PoolClosed.into()),
- _ => Err(unavailable()),
- }
- })
- .await;
- assert!(result.is_err());
- let delay = failed_at.get().elapsed();
- assert!(delay >= Duration::from_millis(400));
- assert!(delay < Duration::from_secs(1));
+ let AddIncomingResult::Success { new, .. } = result else {
+ unreachable!("registration without a Taler subject cannot fail subject checks");
+ };
+ if !new {
+ trace!(target: "worker", "in {t} already seen");
+ return Ok(None);
}
-
- #[tokio::test(start_paused = true)]
- async fn transient_attempts_each_balance_once_and_reports_failure() {
- let attempts = Cell::new(0);
- let results = join_all([
- poll_balance(
- 1,
- Duration::from_secs(60),
- true,
- async || Err(unavailable()),
- )
- .boxed_local(),
- poll_balance(2, Duration::from_secs(60), true, async || {
- tokio::time::sleep(Duration::from_millis(100)).await;
- attempts.set(attempts.get() + 1);
- Ok(())
- })
- .boxed_local(),
- ])
- .await;
- assert!(results[0].is_err());
- assert!(results[1].is_ok());
- assert_eq!(attempts.get(), 1);
+ if let Some(reason) = failure {
+ info!(target: "worker", "in {t} malformed: {reason}");
+ } else {
+ info!(target: "worker", "in {t}");
+ }
+ if t.debtor.is_none() {
+ warn!(target: "worker", "in {t} cannot parse debtor account '{}'", tx.details.sender_account);
}
+ Ok(Some(t))
}
diff --git a/common/taler-common/src/config.rs b/common/taler-common/src/config.rs
@@ -19,7 +19,7 @@ use std::{
sync::Arc, time::Duration,
};
-use compact_str::CompactString;
+use compact_str::{CompactString, ToCompactString};
use indexmap::IndexMap;
use jiff::{SignedDuration, Span};
use url::Url;
@@ -477,14 +477,14 @@ pub enum CfgErr {
IO(Cow<'static, str>, PathBuf, std::io::Error),
Line(Cow<'static, str>, PathBuf, usize, Option<String>),
Missing {
- ty: String,
- section: String,
- option: String,
+ ty: CompactString,
+ section: CompactString,
+ option: CompactString,
},
Invalid {
- ty: String,
- section: String,
- option: String,
+ ty: Option<CompactString>,
+ section: CompactString,
+ option: CompactString,
err: String,
},
Custom(String),
@@ -518,10 +518,13 @@ impl std::fmt::Display for CfgErr {
section,
option,
err,
- } => write!(
- f,
- "invalid {ty} option {option} in section [{section}]: {err}"
- ),
+ } => {
+ f.write_str("invalid ")?;
+ if let Some(ty) = ty {
+ write!(f, "{ty} ")?;
+ }
+ write!(f, "option {option} in section [{section}]: {err}")
+ }
CfgErr::Custom(e) => e.fmt(f),
}
}
@@ -787,6 +790,16 @@ pub enum MapErr {
}
impl<'cfg, 'arg> Section<'cfg, 'arg> {
+ /// Report a logical validation error for an option in this section.
+ pub fn invalid(&self, option: &str, err: impl std::fmt::Display) -> CfgErr {
+ CfgErr::Invalid {
+ ty: None,
+ section: self.name.to_compact_string().to_lowercase(),
+ option: option.to_compact_string().to_uppercase(),
+ err: err.to_string(),
+ }
+ }
+
#[doc(hidden)]
fn inner<T>(
&self,
@@ -836,9 +849,9 @@ impl<'cfg, 'arg> Section<'cfg, 'arg> {
buf.push_str(v);
buf.push('\'');
CfgErr::Invalid {
- ty: ty.to_owned(),
- section: self.name.to_lowercase(),
- option: option.to_uppercase(),
+ ty: Some(ty.to_compact_string()),
+ section: self.name.to_compact_string().to_lowercase(),
+ option: option.to_compact_string().to_uppercase(),
err: buf,
}
}
@@ -856,9 +869,9 @@ impl<'cfg, 'arg> Section<'cfg, 'arg> {
) -> Value<'arg, T> {
self.inner(ty, option, |v| {
transform(v).map_err(|e| CfgErr::Invalid {
- ty: ty.to_owned(),
- section: self.name.to_lowercase(),
- option: option.to_uppercase(),
+ ty: Some(ty.to_compact_string()),
+ section: self.name.to_compact_string().to_lowercase(),
+ option: option.to_compact_string().to_uppercase(),
err: e.to_string(),
})
})
@@ -1042,9 +1055,9 @@ impl<T> Value<'_, T> {
/** Converted value or throw if missing */
pub fn require(self) -> Result<T, CfgErr> {
self.value?.ok_or_else(|| CfgErr::Missing {
- ty: self.ty.to_owned(),
- section: self.section.to_lowercase(),
- option: self.option.to_uppercase(),
+ ty: self.ty.to_compact_string(),
+ section: self.section.to_compact_string().to_lowercase(),
+ option: self.option.to_compact_string().to_uppercase(),
})
}
}
diff --git a/taler-apns-relay/src/apns.rs b/taler-apns-relay/src/apns.rs
@@ -193,7 +193,7 @@ impl Client {
.expect("failed to install the default TLS provider");
let invalid_key = |message: String| taler_common::config::CfgErr::Invalid {
- ty: "PKCS#8 private key file".into(),
+ ty: Some("PKCS#8 private key file".into()),
section: "apns-relay-worker".into(),
option: "KEY_FILE".into(),
err: message,
