commit 3e47690be266b15c9f61478f8914c9a89e461774
parent e9aedc01fa0725d6b0aa01ec5bf8c1b996706fc8
Author: Antoine A <>
Date: Thu, 15 Oct 2026 11:05:06 +0200
common: improve metrics
Diffstat:
10 files changed, 233 insertions(+), 26 deletions(-)
diff --git a/adapters/taler-cyclos/src/api.rs b/adapters/taler-cyclos/src/api.rs
@@ -16,7 +16,11 @@
use compact_str::CompactString;
use jiff::Timestamp;
-use prometheus_client::{metrics::gauge::Gauge, registry::Registry};
+use prometheus_client::{
+ encoding::EncodeLabelSet,
+ metrics::{family::Family, gauge::Gauge},
+ registry::{Registry, Unit},
+};
use sqlx::PgPool;
use taler_api::{
api::{
@@ -64,9 +68,16 @@ pub struct CyclosApi {
registry: Registry,
}
+#[derive(Clone, Debug, Hash, PartialEq, Eq, EncodeLabelSet)]
+struct TaskLabel {
+ name: &'static str,
+}
+
#[derive(Default)]
struct Metrics {
db_access: Gauge,
+ task_execution: Family<TaskLabel, Gauge>,
+ task_success: Family<TaskLabel, Gauge>,
}
impl Metrics {
@@ -78,12 +89,52 @@ impl Metrics {
"Whether the last database metrics refresh succeeded",
self.db_access.clone(),
);
+ registry.register_with_unit(
+ "task_execution_timestamp_seconds",
+ "Unix timestamp of the last task execution",
+ Unit::Seconds,
+ self.task_execution.clone(),
+ );
+ registry.register_with_unit(
+ "task_success_timestamp_seconds",
+ "Unix timestamp of the last successful task execution",
+ Unit::Seconds,
+ self.task_success.clone(),
+ );
registry
}
pub async fn sync(&self, db: &PgPool) {
- let test = sqlx::query("SELECT 1").fetch_one(db).await.is_ok();
- self.db_access.set(if test { 1 } else { 0 });
+ let result = sqlx::query_as::<_, (String, sqlx::types::Json<db::TaskStatus>)>(
+ "SELECT key, value FROM kv WHERE key IN ('sync', 'submit')",
+ )
+ .fetch_all(db)
+ .await;
+ let success = match result {
+ Ok(rows) => {
+ for name in ["sync", "submit"] {
+ let label = TaskLabel { name };
+ let status = rows.iter().find(|(key, _)| key == name).map(|(_, s)| &s.0);
+ if let Some(time) = status.and_then(|s| s.last_execution) {
+ self.task_execution
+ .get_or_create(&label)
+ .set(time.as_second());
+ } else {
+ self.task_execution.remove(&label);
+ }
+ if let Some(time) = status.and_then(|s| s.last_success) {
+ self.task_success
+ .get_or_create(&label)
+ .set(time.as_second());
+ } else {
+ self.task_success.remove(&label);
+ }
+ }
+ true
+ }
+ Err(_) => false,
+ };
+ self.db_access.set(if success { 1 } else { 0 });
}
}
diff --git a/adapters/taler-cyclos/src/db.rs b/adapters/taler-cyclos/src/db.rs
@@ -18,7 +18,7 @@ use std::fmt::Display;
use compact_str::CompactString;
use jiff::Timestamp;
-use serde::{Serialize, de::DeserializeOwned};
+use serde::{Deserialize, Serialize, de::DeserializeOwned};
use sqlx::{PgConnection, PgPool, QueryBuilder, Row, postgres::PgRow};
use taler_api::{
db::{BindHelper, TypeHelper, history, page},
@@ -1656,3 +1656,31 @@ mod test {
assert_eq!(pendings.len(), 83);
}
}
+
+#[derive(Debug, Default, Deserialize)]
+pub struct TaskStatus {
+ pub last_execution: Option<Timestamp>,
+ pub last_success: Option<Timestamp>,
+}
+
+pub async fn task_start(db: &mut PgConnection, key: &str) -> sqlx::Result<()> {
+ task_update(db, key, "last_execution").await
+}
+
+pub async fn task_success(db: &mut PgConnection, key: &str) -> sqlx::Result<()> {
+ task_update(db, key, "last_success").await
+}
+
+async fn task_update(db: &mut PgConnection, key: &str, field: &str) -> sqlx::Result<()> {
+ serialized!(
+ sqlx::query(
+ "INSERT INTO kv (key, value) VALUES ($1, jsonb_build_object($2::text, $3::jsonb))
+ ON CONFLICT (key) DO UPDATE SET value = kv.value || EXCLUDED.value",
+ )
+ .bind(key)
+ .bind(field)
+ .bind(sqlx::types::Json(Timestamp::now()))
+ .execute(&mut *db)
+ )?;
+ Ok(())
+}
diff --git a/adapters/taler-cyclos/src/worker.rs b/adapters/taler-cyclos/src/worker.rs
@@ -195,6 +195,8 @@ impl Worker<'_> {
return Err(WorkerError::Concurrency);
}
+ db::task_start(&mut *self.db, "sync").await?;
+
// Sync transactions
let mut cursor: Timestamp = kv_get(&mut *self.db, SYNC_CURSOR_KEY)
.await?
@@ -223,6 +225,9 @@ impl Worker<'_> {
}
}
+ db::task_success(&mut *self.db, "sync").await?;
+ db::task_start(&mut *self.db, "submit").await?;
+
// Send transactions
let start = Timestamp::now();
loop {
@@ -270,6 +275,7 @@ impl Worker<'_> {
}
}
}
+ db::task_success(&mut *self.db, "submit").await?;
Ok(())
}
diff --git a/adapters/taler-magnet-bank/src/api.rs b/adapters/taler-magnet-bank/src/api.rs
@@ -15,7 +15,11 @@
*/
use jiff::Timestamp;
-use prometheus_client::{metrics::gauge::Gauge, registry::Registry};
+use prometheus_client::{
+ encoding::EncodeLabelSet,
+ metrics::{family::Family, gauge::Gauge},
+ registry::{Registry, Unit},
+};
use sqlx::PgPool;
use taler_api::{
api::{
@@ -62,9 +66,16 @@ pub struct MagnetApi {
registry: Registry,
}
+#[derive(Clone, Debug, Hash, PartialEq, Eq, EncodeLabelSet)]
+struct TaskLabel {
+ name: &'static str,
+}
+
#[derive(Default)]
struct Metrics {
db_access: Gauge,
+ task_execution: Family<TaskLabel, Gauge>,
+ task_success: Family<TaskLabel, Gauge>,
}
impl Metrics {
@@ -76,12 +87,59 @@ impl Metrics {
"Whether the last database metrics refresh succeeded",
self.db_access.clone(),
);
+ registry.register_with_unit(
+ "task_execution_timestamp_seconds",
+ "Unix timestamp of the last task execution",
+ Unit::Seconds,
+ self.task_execution.clone(),
+ );
+ registry.register_with_unit(
+ "task_success_timestamp_seconds",
+ "Unix timestamp of the last successful task execution",
+ Unit::Seconds,
+ self.task_success.clone(),
+ );
registry
}
pub async fn sync(&self, db: &PgPool) {
- let test = sqlx::query("SELECT 1").fetch_one(db).await.is_ok();
- self.db_access.set(if test { 1 } else { 0 });
+ let success = match sqlx::query_as::<
+ _,
+ (
+ Option<sqlx::types::Json<db::TaskStatus>>,
+ Option<sqlx::types::Json<db::TaskStatus>>,
+ ),
+ >(
+ "SELECT (SELECT value FROM kv WHERE key = 'sync'),
+ (SELECT value FROM kv WHERE key = 'submit')",
+ )
+ .fetch_one(db)
+ .await
+ {
+ Ok((sync, submit)) => {
+ for (name, status) in [("sync", sync), ("submit", submit)] {
+ let label = TaskLabel { name };
+ let status = status.map(|s| s.0).unwrap_or_default();
+ if let Some(time) = status.last_execution {
+ self.task_execution
+ .get_or_create(&label)
+ .set(time.as_second());
+ } else {
+ self.task_execution.remove(&label);
+ }
+ if let Some(time) = status.last_success {
+ self.task_success
+ .get_or_create(&label)
+ .set(time.as_second());
+ } else {
+ self.task_success.remove(&label);
+ }
+ }
+ true
+ }
+ Err(_) => false,
+ };
+ self.db_access.set(if success { 1 } else { 0 });
}
}
diff --git a/adapters/taler-magnet-bank/src/db.rs b/adapters/taler-magnet-bank/src/db.rs
@@ -18,7 +18,7 @@ use std::fmt::Display;
use compact_str::CompactString;
use jiff::{Timestamp, civil::Date, tz::TimeZone};
-use serde::{Serialize, de::DeserializeOwned};
+use serde::{Deserialize, Serialize, de::DeserializeOwned};
use sqlx::{PgConnection, PgPool, QueryBuilder, Row, postgres::PgRow};
use taler_api::{
db::{BindHelper, TypeHelper, history, page},
@@ -1685,3 +1685,31 @@ mod test {
assert_eq!(pendings.len(), 83);
}
}
+
+#[derive(Debug, Default, Deserialize)]
+pub struct TaskStatus {
+ pub last_execution: Option<Timestamp>,
+ pub last_success: Option<Timestamp>,
+}
+
+pub async fn task_start(db: &mut PgConnection, key: &str) -> sqlx::Result<()> {
+ task_update(db, key, "last_execution").await
+}
+
+pub async fn task_success(db: &mut PgConnection, key: &str) -> sqlx::Result<()> {
+ task_update(db, key, "last_success").await
+}
+
+async fn task_update(db: &mut PgConnection, key: &str, field: &str) -> sqlx::Result<()> {
+ serialized!(
+ sqlx::query(
+ "INSERT INTO kv (key, value) VALUES ($1, jsonb_build_object($2::text, $3::jsonb))
+ ON CONFLICT (key) DO UPDATE SET value = kv.value || EXCLUDED.value",
+ )
+ .bind(key)
+ .bind(field)
+ .bind(sqlx::types::Json(Timestamp::now()))
+ .execute(&mut *db)
+ )?;
+ Ok(())
+}
diff --git a/adapters/taler-magnet-bank/src/worker.rs b/adapters/taler-magnet-bank/src/worker.rs
@@ -188,6 +188,8 @@ impl Worker<'_> {
return Err(WorkerError::Concurrency);
};
+ db::task_start(&mut *self.db, "sync").await?;
+
// Sync transactions
let mut next: Option<Next> = None; //kv_get(&mut *self.db, TXS_CURSOR_KEY).await?; TODO cursor logic is broken and cannot be stored & reused
let mut all_final = true;
@@ -469,6 +471,9 @@ impl Worker<'_> {
}
}
+ db::task_success(&mut *self.db, "sync").await?;
+ db::task_start(&mut *self.db, "submit").await?;
+
// Send transactions
let start = Timestamp::now();
let now = Zoned::now();
@@ -482,6 +487,7 @@ impl Worker<'_> {
self.init_tx(&tx, &now).await?;
}
}
+ db::task_success(&mut *self.db, "submit").await?;
Ok(())
}
diff --git a/adapters/taler-wise/db/wise-0001.sql b/adapters/taler-wise/db/wise-0001.sql
@@ -27,11 +27,13 @@ CREATE TABLE balances (
early_cursor INT8 NOT NULL,
sync_cursor INT8 NOT NULL,
checkpoint_cursor INT8 NOT NULL,
+ sync_execution_at INT8,
early_success_at INT8,
sync_success_at INT8,
checkpoint_success_at INT8
);
COMMENT ON TABLE balances IS 'Balance creation time, transaction cursors and successful scan times (UTC microseconds)';
+COMMENT ON COLUMN balances.sync_execution_at IS 'Last sync attempt (UTC microseconds)';
CREATE TABLE tx_in (
tx_in_id INT8 PRIMARY KEY GENERATED ALWAYS AS IDENTITY,
@@ -83,4 +85,4 @@ CREATE TABLE pending_recurrent_in(
);
CREATE INDEX pending_recurrent_inc_auth_pub
ON pending_recurrent_in (authorization_pub);
-COMMENT ON TABLE pending_recurrent_in IS 'Pending recurrent incoming transaction';
-\ No newline at end of file
+COMMENT ON TABLE pending_recurrent_in IS 'Pending recurrent incoming transaction';
diff --git a/adapters/taler-wise/src/api.rs b/adapters/taler-wise/src/api.rs
@@ -183,21 +183,33 @@ impl Metrics {
}
pub async fn sync(&self, db: &PgPool, balance_id: u32) {
- let test = sqlx::query("SELECT 1").fetch_one(db).await.is_ok();
- if let Ok(Some(state)) = db::load_sync_state(db, balance_id).await {
- let label = TaskLabel { name: "sync" };
- let success = state
- .early_success
- .max(state.sync_success)
- .max(state.checkpoint_success);
-
- if let Some(time) = success {
- self.task_success
- .get_or_create(&label)
- .set(time.as_second());
+ let success = match db::load_sync_state(db, balance_id).await {
+ Ok(state) => {
+ let label = TaskLabel { name: "sync" };
+ if let Some(time) = state.as_ref().and_then(|s| s.execution) {
+ self.task_execution
+ .get_or_create(&label)
+ .set(time.as_second());
+ } else {
+ self.task_execution.remove(&label);
+ }
+ let success = state.and_then(|s| {
+ s.early_success
+ .max(s.sync_success)
+ .max(s.checkpoint_success)
+ });
+ if let Some(time) = success {
+ self.task_success
+ .get_or_create(&label)
+ .set(time.as_second());
+ } else {
+ self.task_success.remove(&label);
+ }
+ true
}
- }
- self.db_access.set(if test { 1 } else { 0 });
+ Err(_) => false,
+ };
+ self.db_access.set(if success { 1 } else { 0 });
}
}
diff --git a/adapters/taler-wise/src/db.rs b/adapters/taler-wise/src/db.rs
@@ -64,6 +64,7 @@ pub async fn dbinit(cfg: &Config, reset: bool) -> anyhow::Result<PgPool> {
/// Durable transaction high-water marks and completed scan schedules.
#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) struct SyncState {
+ pub execution: Option<Timestamp>,
pub early: Timestamp,
pub sync: Timestamp,
pub checkpoint: Timestamp,
@@ -78,12 +79,13 @@ pub(crate) async fn load_sync_state(
) -> sqlx::Result<Option<SyncState>> {
sqlx::query(
"SELECT early_cursor, sync_cursor, checkpoint_cursor,
- early_success_at, sync_success_at, checkpoint_success_at
+ early_success_at, sync_success_at, checkpoint_success_at, sync_execution_at
FROM balances WHERE balance_id = $1",
)
.bind(i64::from(balance_id))
.try_map(|r: PgRow| {
Ok(SyncState {
+ execution: r.try_get_opt_timestamp(6)?,
early: r.try_get_timestamp(0)?,
sync: r.try_get_timestamp(1)?,
checkpoint: r.try_get_timestamp(2)?,
@@ -118,6 +120,7 @@ pub(crate) async fn reset_sync_state(db: &PgPool) -> sqlx::Result<()> {
serialized!(
sqlx::query(
"UPDATE balances SET
+ sync_execution_at = NULL,
early_cursor = created_at,
sync_cursor = created_at,
checkpoint_cursor = created_at,
@@ -425,6 +428,16 @@ pub async fn transfer_unregister(db: &PgPool, req: &Unregistration) -> sqlx::Res
)
}
+pub async fn record_sync_execution(db: &PgPool, balance_id: u32) -> sqlx::Result<()> {
+ serialized!(
+ sqlx::query("UPDATE balances SET sync_execution_at = $2 WHERE balance_id = $1")
+ .bind(i64::from(balance_id))
+ .bind_timestamp(&Timestamp::now())
+ .execute(db)
+ )?;
+ Ok(())
+}
+
#[cfg(test)]
mod test {
diff --git a/adapters/taler-wise/src/worker.rs b/adapters/taler-wise/src/worker.rs
@@ -32,7 +32,10 @@ use tracing::{debug, error, info, trace, warn};
use crate::{
config::{WiseBalance, WorkerCfg},
constants::{DAY, HOUR},
- db::{AddIncomingResult, SyncState, TxIn, load_sync_state, register_tx_in, save_sync_state},
+ db::{
+ AddIncomingResult, SyncState, TxIn, load_sync_state, record_sync_execution, register_tx_in,
+ save_sync_state,
+ },
payto::WiseAccount,
wise_api::{
client::{Client, WiseErr},
@@ -268,6 +271,7 @@ async fn sync_balance(
if !force && next > now {
return Ok(next);
}
+ record_sync_execution(pool, balance.id).await?;
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 {
