commit 248a7e70dade362cff8b12140696baf91acdbeca
parent b3043819e7e2af8f18b2c434b7266c25dcace578
Author: Antoine A <>
Date: Tue, 8 Sep 2026 13:03:59 +0200
ebisync: many fixes
Diffstat:
10 files changed, 121 insertions(+), 131 deletions(-)
diff --git a/debian/etc/libeufin-ebisync/secrets/ebisync-fetch.secret.conf b/debian/etc/libeufin-ebisync/secrets/ebisync-fetch.secret.conf
@@ -10,4 +10,4 @@
# AZURE_ACCOUNT_KEY = Eby8vdM02xNOcqFlqUwJPLlmEtlCDXJ1OUzFT50uSRZ6IFsuFq2UVErCz4I6tq/K1SZFPTOtr/KBHBeksoGMGw==
# Which Azure Blob Storage container to use for azure-blob-storage
-# AZURE_COUNTAINER = mycontainer
+# AZURE_CONTAINER = mycontainer
diff --git a/libeufin-bank/src/api.rs b/libeufin-bank/src/api.rs
@@ -27,9 +27,7 @@ use axum::{
};
use serde::{Deserialize, Serialize};
use sqlx::PgPool;
-use taler_api::{
- api::TalerRouter as _, error::ApiResult, extract::Query, notification::NotificationChannel,
-};
+use taler_api::{error::ApiResult, extract::Query, notification::NotificationChannel};
use taler_common::{
api::LibtoolVersion,
types::amount::{Amount, Currency},
@@ -249,7 +247,6 @@ pub fn bank_api(state: Arc<BankState>) -> Router {
.merge(prepared_api())
.merge(revenue_api())
.with_state(state)
- .finalize()
}
#[cfg(any(test, feature = "test-utils"))]
diff --git a/libeufin-bank/src/api/token.rs b/libeufin-bank/src/api/token.rs
@@ -216,7 +216,7 @@ pub mod test {
use std::{str::FromStr, time::Duration};
use axum::http::{Method, header::AUTHORIZATION};
- use jiff::Timestamp;
+ use jiff::{SignedDuration, Timestamp};
use taler_api::db::TypeHelper;
use taler_common::{
api::ErrorDetail,
@@ -288,7 +288,11 @@ pub mod test {
match token.expiration {
TalerTimestamp::Never => unreachable!(),
TalerTimestamp::Timestamp(expiration) => {
- let lifetime = token.creation.duration_until(expiration).unsigned_abs();
+ let lifetime = token
+ .creation
+ .duration_until(expiration)
+ .max(SignedDuration::ZERO)
+ .unsigned_abs();
assert_eq!(lifetime, TOKEN_DEFAULT_DURATION);
}
}
diff --git a/libeufin-ebics/src/ebics.rs b/libeufin-ebics/src/ebics.rs
@@ -24,7 +24,6 @@ use aws_lc_rs::{
encoding::AsDer,
error::KeyRejected,
rsa::{self, PrivateDecryptingKey, PublicKey},
- signature::KeyPair,
};
use compact_str::CompactString;
use flate2::write::ZlibDecoder;
@@ -45,8 +44,8 @@ use crate::{
cli::EbicsLogs,
config::EbicsHostCfg,
crypto::{
- decrypt_ebics_e002, decrypt_ebics_e002_key, digest_ebics_order_a006, ebics_pub_key_hash,
- encrypt_ebics_e002, gen_ebics_e002_key, sign_ebics_a006,
+ decrypt_ebics_e002, decrypt_ebics_e002_key, digest_ebics_order_a006, encrypt_ebics_e002,
+ gen_ebics_e002_key, sign_ebics_a006,
},
db::{ebics_first, ebics_register, ebics_remove},
ebics::{
diff --git a/libeufin-ebisync/ebisync.conf b/libeufin-ebisync/ebisync.conf
@@ -50,7 +50,7 @@ DESTINATION = none
# AZURE_ACCOUNT_KEY = Eby8vdM02xNOcqFlqUwJPLlmEtlCDXJ1OUzFT50uSRZ6IFsuFq2UVErCz4I6tq/K1SZFPTOtr/KBHBeksoGMGw==
# Which Azure Blob Storage container to use for azure-blob-storage
-# AZURE_COUNTAINER = mycontainer
+# AZURE_CONTAINER = mycontainer
[ebisync-submit]
# Where does the ebics file come from? This his can either can be ebisync-api or none
diff --git a/libeufin-ebisync/src/api.rs b/libeufin-ebisync/src/api.rs
@@ -36,7 +36,7 @@ use reqwest::StatusCode;
use serde::{Deserialize, Serialize};
use sqlx::PgPool;
use taler_api::{
- api::{RouterUtils, TalerRouter as _},
+ api::RouterUtils,
auth::AuthMethod,
error::{ApiResult, bad_request, failure, failure_status},
};
@@ -216,5 +216,4 @@ pub fn sync_api(state: Arc<EbisyncState>, spa: &str, auth: AuthMethod) -> Router
)
.with_state(state)
.auth(auth, "realm")
- .finalize()
}
diff --git a/libeufin-ebisync/src/lib.rs b/libeufin-ebisync/src/lib.rs
@@ -23,14 +23,13 @@ use std::{
fmt::Write as _,
io::{Cursor, Read as _},
sync::Arc,
- time::Duration,
};
use anyhow::{anyhow, bail};
use axum::{body::Bytes, http::HeaderValue};
use compact_str::CompactStringExt;
use http_client::Client;
-use jiff::{Timestamp, Zoned, civil::Date, tz::TimeZone};
+use jiff::{SignedDuration, Timestamp, Zoned, civil::Date, tz::TimeZone};
use libeufin_ebics::{
cli::{EbicsArgs, EbicsLogs},
db::{get_task_status, update_task_status},
@@ -105,7 +104,7 @@ pub enum Cmd {
ebics: EbicsArgs,
/// Only supported in --transient mode, this option lets specify the earliest timestamp of the downloaded documents
- #[arg(long, value_name = "YYYY-MM-DD")]
+ #[arg(long, value_name = "YYYY-MM-DD", requires = "transient")]
pinned_start: Option<Date>,
/// Only supported in --transient mode, do not consume fetched documents
@@ -353,7 +352,7 @@ pub async fn ebics_fetch(
}
} else {
// We never ran, we must checkpoint now
- now
+ Timestamp::UNIX_EPOCH
}
};
@@ -430,15 +429,16 @@ pub async fn ebics_fetch(
}
}
- let delay = now.duration_until(next_fetch.min(next_checkpoint));
- let tx = timeout(
- Duration::from_millis(delay.abs().as_millis() as u64),
- receiver.recv(),
- )
- .await;
+ let delay = now
+ .duration_until(next_fetch.min(next_checkpoint))
+ .max(SignedDuration::ZERO)
+ .unsigned_abs();
+ let tx = timeout(delay, receiver.recv()).await;
if let Ok(Some(notification)) = tx {
info!(target: "fetch", "Running at real-time notifications reception");
- upload(¬ification, None).await?;
+ if let Err(e) = upload(¬ification, None).await {
+ error!(target: "fetch", "{e}");
+ }
}
}
};
diff --git a/libeufin-nexus/src/fetch.rs b/libeufin-nexus/src/fetch.rs
@@ -20,12 +20,11 @@
use std::{
collections::BTreeMap,
io::{Cursor, Read as _},
- time::Duration,
};
use anyhow::{anyhow, bail};
use compact_str::CompactStringExt;
-use jiff::{Timestamp, Zoned, tz::TimeZone};
+use jiff::{SignedDuration, Timestamp, Zoned, tz::TimeZone};
use libeufin_ebics::{
db::{get_task_status, update_task_status},
ebics::{
@@ -272,12 +271,11 @@ pub async fn ebics_fetch(
}
}
- let delay = now.duration_until(next_fetch.min(next_checkpoint));
- let tx = timeout(
- Duration::from_millis(delay.abs().as_millis() as u64),
- receiver.recv(),
- )
- .await;
+ let delay = now
+ .duration_until(next_fetch.min(next_checkpoint))
+ .max(SignedDuration::ZERO)
+ .unsigned_abs();
+ let tx = timeout(delay, receiver.recv()).await;
if let Ok(Some(mut notification)) = tx {
notification.retain(|order| orders.iter().find(|it| order.eq(it)).is_some());
if !notification.is_empty() {
diff --git a/libeufin-nexus/src/lib.rs b/libeufin-nexus/src/lib.rs
@@ -23,7 +23,7 @@ use std::{fmt::Write, sync::Arc};
use anyhow::{anyhow, bail};
use compact_str::{CompactString, CompactStringExt, ToCompactString};
-use jiff::{Timestamp, civil::Date};
+use jiff::{SignedDuration, Timestamp, civil::Date};
use libeufin_ebics::{
cli::{EbicsArgs, EbicsLogs},
db::update_task_status,
@@ -302,12 +302,13 @@ pub async fn ebics_submit(
if let Err(e) = update_task_status(db, SUBMIT_TASK_KEY, &now, success).await {
warn!(target: "submit", "{e}");
}
- let wait = Timestamp::now().duration_until(now + submit_cfg.frequency);
- if wait.is_positive() {
- tokio::time::sleep(wait.unsigned_abs()).await;
- } else {
- tokio::task::yield_now().await;
- }
+ tokio::time::sleep(
+ Timestamp::now()
+ .duration_until(now + submit_cfg.frequency)
+ .max(SignedDuration::ZERO)
+ .unsigned_abs(),
+ )
+ .await;
}
}
}
@@ -584,7 +585,7 @@ pub async fn run(cfg: &Config, cmd: Cmd) -> anyhow::Result<()> {
if let Some(it) = cfg.revenue_cfg {
server = server.revenue(api.clone(), it.auth.method())
}
- server.finalize().serve(&cfg.serve_cfg, None).await?;
+ server.serve(&cfg.serve_cfg, None).await?;
}
}
Cmd::Manual(cmd) => {
diff --git a/testbench/src/main.rs b/testbench/src/main.rs
@@ -307,113 +307,105 @@ async fn main() -> anyhow::Result<()> {
Component::Nexus => {
let mut repl = Repl::new(".nexus_history");
auto_setup().await;
- loop {
- if let Some(cmd) = repl.read_line(&prompt, "") {
- match cmd {
- NexusCmd::Setup => {
- run(&format!("ebics-setup {ebics_log}")).await;
- }
- NexusCmd::ResetDb => {
- run("dbinit -r").await;
- }
- NexusCmd::Compact => {
- compact(format!("testbench/test/{platform}"));
- }
- NexusCmd::Fetch { raw_args } => {
- run(&format!("ebics-fetch {ebics_log} {}", raw_args.join(" "))).await;
- }
- NexusCmd::Submit { raw_args } => {
- run(&format!("ebics-submit {ebics_log} {}", raw_args.join(" "))).await;
- }
- NexusCmd::Tx => {
- run(&format!(
- "initiate-payment --amount={}:0.1 --subject=\"single {}\" {payto}",
- cfg.currency,
- Timestamp::now()
- ))
- .await;
- }
- NexusCmd::List { raw_args } => {
- run(&format!("list {}", raw_args.join(" "))).await;
- }
- NexusCmd::ResetKeys => {
- if test && std::fs::exists(&ebics.client).unwrap() {
- std::fs::remove_file(&ebics.client).unwrap();
- }
- if std::fs::exists(&ebics.bank).unwrap() {
- std::fs::remove_file(&ebics.bank).unwrap();
- }
+ while let Some(cmd) = repl.read_line(&prompt, "") {
+ match cmd {
+ NexusCmd::Setup => {
+ run(&format!("ebics-setup {ebics_log}")).await;
+ }
+ NexusCmd::ResetDb => {
+ run("dbinit -r").await;
+ }
+ NexusCmd::Compact => {
+ compact(format!("testbench/test/{platform}"));
+ }
+ NexusCmd::Fetch { raw_args } => {
+ run(&format!("ebics-fetch {ebics_log} {}", raw_args.join(" "))).await;
+ }
+ NexusCmd::Submit { raw_args } => {
+ run(&format!("ebics-submit {ebics_log} {}", raw_args.join(" "))).await;
+ }
+ NexusCmd::Tx => {
+ run(&format!(
+ "initiate-payment --amount={}:0.1 --subject=\"single {}\" {payto}",
+ cfg.currency,
+ Timestamp::now()
+ ))
+ .await;
+ }
+ NexusCmd::List { raw_args } => {
+ run(&format!("list {}", raw_args.join(" "))).await;
+ }
+ NexusCmd::ResetKeys => {
+ if test && std::fs::exists(&ebics.client).unwrap() {
+ std::fs::remove_file(&ebics.client).unwrap();
}
- NexusCmd::TxCheck => {
- run(&format!("testing tx-check {ebics_log}")).await;
+ if std::fs::exists(&ebics.bank).unwrap() {
+ std::fs::remove_file(&ebics.bank).unwrap();
}
- NexusCmd::Wss => {
- run(&format!("testing wss {ebics_log}")).await;
+ }
+ NexusCmd::TxCheck => {
+ run(&format!("testing tx-check {ebics_log}")).await;
+ }
+ NexusCmd::Wss => {
+ run(&format!("testing wss {ebics_log}")).await;
+ }
+ NexusCmd::Import { path, glob } => {
+ let mut b = GlobSetBuilder::new();
+ for glob in glob {
+ b.add(glob);
}
- NexusCmd::Import { path, glob } => {
- let mut b = GlobSetBuilder::new();
- for glob in glob {
- b.add(glob);
- }
- let matcher = b.build().unwrap();
- let mut paths = Vec::new();
- for entry in WalkDir::new(&path).into_iter().filter_map(Result::ok) {
- let entry = entry.path();
+ let matcher = b.build().unwrap();
+ let mut paths = Vec::new();
+ for entry in WalkDir::new(&path).into_iter().filter_map(Result::ok) {
+ let entry = entry.path();
- // usually match against relative path
- let rel = entry.strip_prefix(&path).unwrap_or(entry);
+ // usually match against relative path
+ let rel = entry.strip_prefix(&path).unwrap_or(entry);
- if matcher.is_match(rel) {
- paths.push(entry.to_string_lossy().to_string());
- }
+ if matcher.is_match(rel) {
+ paths.push(entry.to_string_lossy().to_string());
}
-
- run(&format!("manual import {}", paths.join(" "))).await;
- }
- NexusCmd::Export { path } => {
- run(&format!("manual export {}", path.to_string_lossy())).await;
}
- NexusCmd::Exit => break,
+
+ run(&format!("manual import {}", paths.join(" "))).await;
}
- } else {
- break;
+ NexusCmd::Export { path } => {
+ run(&format!("manual export {}", path.to_string_lossy())).await;
+ }
+ NexusCmd::Exit => break,
}
}
}
Component::Ebisync => {
let mut repl = Repl::new(".ebisync_history");
auto_setup().await;
- loop {
- if let Some(cmd) = repl.read_line(&prompt, "") {
- match cmd {
- SyncCmd::Compact => {
- compact(format!("testbench/test/{platform}"));
- }
- SyncCmd::Setup => {
- run(&format!("setup {ebics_log}")).await;
- }
- SyncCmd::ResetDb => {
- run("dbinit -r").await;
- }
- SyncCmd::Fetch { raw_args } => {
- run(&format!("fetch {ebics_log} {}", raw_args.join(" "))).await;
- }
- SyncCmd::Testing { raw_args } => {
- run(&format!("testing {}", raw_args.join(" "))).await;
- }
- SyncCmd::Serve => {
- run("serve").await;
- }
- SyncCmd::ResetKeys => {
- if test {
- std::fs::remove_file(&ebics.client).unwrap();
- }
- std::fs::remove_file(&ebics.bank).unwrap();
+ while let Some(cmd) = repl.read_line(&prompt, "") {
+ match cmd {
+ SyncCmd::Compact => {
+ compact(format!("testbench/test/{platform}"));
+ }
+ SyncCmd::Setup => {
+ run(&format!("setup {ebics_log}")).await;
+ }
+ SyncCmd::ResetDb => {
+ run("dbinit -r").await;
+ }
+ SyncCmd::Fetch { raw_args } => {
+ run(&format!("fetch {ebics_log} {}", raw_args.join(" "))).await;
+ }
+ SyncCmd::Testing { raw_args } => {
+ run(&format!("testing {}", raw_args.join(" "))).await;
+ }
+ SyncCmd::Serve => {
+ run("serve").await;
+ }
+ SyncCmd::ResetKeys => {
+ if test {
+ std::fs::remove_file(&ebics.client).unwrap();
}
- SyncCmd::Exit => break,
+ std::fs::remove_file(&ebics.bank).unwrap();
}
- } else {
- break;
+ SyncCmd::Exit => break,
}
}
}
