From 3c737af8f32181a0e6cdba4929fd036d3c7795ff Mon Sep 17 00:00:00 2001 From: akenshaw <63060680+akenshaw@users.noreply.github.com> Date: Fri, 17 Jul 2026 18:29:32 +0300 Subject: [PATCH 1/8] add hard cap database size: `max_storage_mb` --- config.example.toml | 40 +++-- src/cleanup.rs | 381 ++++++++++++++++++++++++++++++++++++++++++++ src/config.rs | 13 ++ src/main.rs | 14 +- src/storage.rs | 185 ++++++++++----------- 5 files changed, 511 insertions(+), 122 deletions(-) create mode 100644 src/cleanup.rs diff --git a/config.example.toml b/config.example.toml index b8b35ee..e7db73c 100644 --- a/config.example.toml +++ b/config.example.toml @@ -10,10 +10,34 @@ data_dir = "./data" # (fewer fsyncs, larger columnar batches). Default: 2000 (2 seconds). # flush_interval_ms = 2000 +# Maximum trades to buffer in memory before dropping to prevent OOM. +# Tune based on your host RAM and risk tolerance: +# 200k (~20-40 MB) — safe for 1 GB hosts +# 1.2M (~120-240 MB) — ~2 min buffer at 10k trades/sec +# Higher values reduce data-loss during DB outages but use more memory. +# Default: 200000. +# max_buffered_trades = 200000 + # Data retention period in hours. Trades older than this will be purged # on startup (and periodically while running). Default: 48 (2 days). # data_retention_hours = 48 +# Hard cap on total DuckDB database file size in megabytes. +# When the database file size exceeds this value the oldest trades are +# purged during cleanup — even if they're within the time-based retention +# window. Use this to prevent the database from quietly filling your disk +# on small VPS hosts. +# +# Tip: set to ~50-80 % of your available disk space to leave room for +# system files, logs, and bursts. +# Default: no cap. +# max_storage_mb = 4096 + +# Discovery mode: fetches and caches metadata for ALL supported exchanges on startup +# so `/exchanges` is fully populated. Lets you browse available tickers +# before deciding what to track. Default: true. +# discovery_mode = true + # Base assets to track — expanded via the whitelist templates below. # The server fetches exchange metadata and resolves pairs at startup. base_assets = ["BTC", "ETH"] @@ -22,21 +46,7 @@ base_assets = ["BTC", "ETH"] # To exclude a venue entirely, omit it from the whitelist. # To exclude a market kind for a venue, omit that key. # -# Tip: `http(s):///exchanges` shows all available tickers that can be tracked. -# use `[""]` (empty string) as a wildcard to track all `base_assets` regardless of the quote on that venue+market -# -# Discovery mode: fetches and caches metadata for ALL supported exchanges on startup -# so `/exchanges` is fully populated. Lets you browse available tickers -# before deciding what to track. Default: true. -# discovery_mode = true - -# Maximum trades to buffer in memory before dropping to prevent OOM. -# Tune based on your host RAM and risk tolerance: -# 200k (~20-40 MB) — safe for 1 GB hosts -# 1.2M (~120-240 MB) — ~2 min buffer at 10k trades/sec -# Higher values reduce data-loss during DB outages but use more memory. -# Default: 200000. -# max_buffered_trades = 200000 +# Use `[""]` (empty string) as a wildcard to track all `base_assets` regardless of the quote on that venue+market [whitelist.binance] spot = ["USDT"] diff --git a/src/cleanup.rs b/src/cleanup.rs new file mode 100644 index 0000000..14397fd --- /dev/null +++ b/src/cleanup.rs @@ -0,0 +1,381 @@ +use std::time::Duration; + +use anyhow::Result; +use tokio::task::JoinHandle; +use tokio_util::sync::CancellationToken; + +use crate::storage::Storage; + +/// Minimum delay between cleanup attempts, regardless of schedule. +/// +/// This prevents a tight retry loop when `delay_until_next_cleanup` +/// returns zero (e.g. because the last time-based cleanup failed and +/// `last_cleanup` is stale). +const MIN_CLEANUP_DELAY: Duration = Duration::from_secs(60); + +/// How often the size-based cleanup task wakes up to check whether the +/// database has exceeded the `max_storage_mb` cap. +/// +/// This interval only matters when a size cap is configured *and* the +/// time-based retention window is longer than 5 minutes — otherwise the +/// time-based schedule dominates. +const SIZE_CHECK_INTERVAL: Duration = Duration::from_secs(300); + +/// Estimated on-disk bytes per trade row. +/// +/// DuckDB's columnar storage compresses the schema well: exchange +/// names dictionary-compress to ~1–2 bytes, symbols to ~2–5 bytes, +/// timestamps benefit from delta encoding, and `DOUBLE` columns +/// compress modestly. Empirically ~40–60 bytes per row including +/// row-group overhead. +/// +/// Used only as a fast-path threshold in the size-based purge loop — +/// the authoritative check is the actual filesystem file size after +/// `CHECKPOINT`. A conservative (lower) value here means we more +/// often verify with the real file size, which is safer. +const EST_BYTES_PER_ROW: u64 = 50; + +/// Choose a batch size proportional to `max_bytes` so that each purge +/// iteration deletes ~10% of the cap. +/// +/// Clamped to **[1 000, 100 000]** so tiny caps still make progress +/// and huge caps don't create giant transactions. +fn purge_batch_size(max_bytes: u64) -> i64 { + const MIN_ROWS: i64 = 1_000; + const MAX_ROWS: i64 = 100_000; + + // Target ~10% of the cap per iteration. + let rows = (max_bytes / 10 / EST_BYTES_PER_ROW) as i64; + rows.clamp(MIN_ROWS, MAX_ROWS) +} + +/// Groups the two retention parameters that are always passed together. +#[derive(Debug, Clone, Copy)] +pub struct CleanupConfig { + /// Time-based retention: trades older than this are purged. + pub retention_hours: u64, + /// Optional hard cap on total DB+WAL size in bytes. When set, the + /// oldest trades are purged even if within the retention window. + pub max_storage_bytes: Option, +} + +impl CleanupConfig { + /// Derive the config from the user-facing `max_storage_mb` setting. + /// A value of `0` or `None` means no cap. + pub fn from_config(retention_hours: u64, max_storage_mb: Option) -> Self { + let max_storage_bytes = max_storage_mb + .filter(|&mb| mb > 0) + .map(|mb| mb.saturating_mul(1024 * 1024)); + Self { + retention_hours, + max_storage_bytes, + } + } +} + +/// Owns the cleanup lifecycle: startup guard, startup pass, and the +/// periodic background task. +/// +/// Created in [`CleanupScheduler::new`] which runs the startup guard. +/// Call [`spawn`](Self::spawn) to start the periodic task — this also +/// runs the startup cleanup pass on a blocking thread and feeds its +/// result into the scheduler so the first periodic delay is accurate. +pub struct CleanupScheduler { + storage: Storage, + config: CleanupConfig, +} + +impl CleanupScheduler { + /// Create the scheduler and run the startup guard. + pub fn new(storage: &Storage, config: CleanupConfig) -> Self { + if let Some(max) = config.max_storage_bytes { + tracing::info!( + "Storage hard cap enabled: {} MB (will purge oldest trades when exceeded)", + max / (1024 * 1024) + ); + + match storage.current_storage_bytes() { + Ok(current) if current > max.saturating_mul(2) => { + tracing::error!( + "Database is {} MB — more than 2x the configured \ + max_storage_mb cap ({} MB). Purging that much data \ + at startup would take too long; this likely indicates \ + a misconfiguration. Either raise max_storage_mb \ + (e.g. to {} MB or higher) or manually shrink the \ + database and restart.", + current / (1024 * 1024), + max / (1024 * 1024), + (current / (1024 * 1024)).saturating_add(1), + ); + std::process::exit(1); + } + Ok(current) if current > max => { + tracing::warn!( + "Database is {} MB — above the {} MB cap; \ + startup cleanup will purge oldest trades.", + current / (1024 * 1024), + max / (1024 * 1024), + ); + } + Err(e) => { + tracing::error!("Failed to check storage size at startup: {e:#}"); + std::process::exit(1); + } + _ => {} + } + } + + Self { + storage: storage.clone(), + config, + } + } + + /// Run a single cleanup pass (time-based retention + optional size + /// cap + checkpoint + record timestamp). + /// + /// Returns `Some(ts)` when the time-based purge succeeded and the + /// `last_cleanup` timestamp was recorded, or `None` on failure. + fn run_pass(&self) -> Option { + let mut cleaned_anything = false; + let mut time_cleanup_ok = false; + + match self.storage.purge_old_trades(self.config.retention_hours) { + Ok(n) => { + time_cleanup_ok = true; + if n > 0 { + tracing::info!( + "Cleaned up {n} trade(s) older than {}h", + self.config.retention_hours + ); + cleaned_anything = true; + } + } + Err(e) => { + tracing::error!("Data retention cleanup failed: {e:#}"); + } + } + + if let Some(max_bytes) = self.config.max_storage_bytes { + match self.purge_oldest_trades_until_below(max_bytes) { + Ok(n) => { + if n > 0 { + let current_mb = self + .storage + .current_storage_bytes() + .map(|b| b / (1024 * 1024)) + .unwrap_or(0); + tracing::info!( + "Cleaned up {n} trade(s) to keep storage under cap \ + (max {} MB, now ~{current_mb} MB)", + max_bytes / (1024 * 1024), + ); + cleaned_anything = true; + } + } + Err(e) => { + tracing::error!("Storage-cap cleanup failed: {e:#}"); + } + } + } + + if cleaned_anything && let Err(e) = self.storage.run_checkpoint() { + tracing::warn!("Failed to checkpoint DuckDB WAL after cleanup: {e:#}"); + } + + let mut last_cleanup_ts: Option = None; + if time_cleanup_ok { + match self.storage.record_cleanup() { + Ok(ts) => last_cleanup_ts = Some(ts), + Err(e) => tracing::warn!("Failed to record last_cleanup timestamp: {e:#}"), + } + } + last_cleanup_ts + } + + /// Delete the oldest trades until the on-disk size drops below + /// `max_bytes`. Returns the total number of deleted rows. + /// + /// The authoritative convergence criterion is the actual filesystem + /// size after `CHECKPOINT`. A row-count estimate + /// (`COUNT(*) × EST_BYTES_PER_ROW`) is used as a fast path: when + /// the row count is clearly above the estimated target we purge + /// without an expensive `CHECKPOINT`; when it's below we verify + /// with the real file size to guard against estimation errors. + /// + /// After purging, if a significant fraction of the data was freed, + /// a `VACUUM` is run to reclaim filesystem space — `CHECKPOINT` + /// alone doesn't shrink the main database file. + fn purge_oldest_trades_until_below(&self, max_bytes: u64) -> Result { + /// Safety ceiling: if the purge hasn't converged after this + /// many iterations, something is wrong. 200 × 100k batch = + /// 20M rows — far beyond any plausible cap. Bail out rather + /// than looping forever. + const MAX_ITERATIONS: usize = 200; + + let max_est_rows = max_bytes.saturating_div(EST_BYTES_PER_ROW); + let batch_size = purge_batch_size(max_bytes); + let mut total_deleted = 0u64; + + for _iteration in 0..MAX_ITERATIONS { + // ── Fast path: row count ─────────────────────────── + // COUNT(*) in DuckDB is a metadata operation on + // row-group headers — cheap even on large tables. + // + // If the row count is clearly above the estimated + // target we purge without an expensive CHECKPOINT. + // If it's below, the estimate may be wrong so we + // fall through to the authoritative file-size check. + let current_rows = self.storage.count_trades()?; + + if current_rows <= max_est_rows { + // ── Authoritative criterion: file size ────── + // CHECKPOINT first so the WAL doesn't inflate + // the filesystem size check. + self.storage.run_checkpoint()?; + + let file_size = self.storage.current_storage_bytes()?; + if file_size <= max_bytes { + tracing::debug!( + "Size-based purge: file size {file_size} ≤ \ + cap {max_bytes}, stopping" + ); + break; + } + // Row count says we're under, but the file is + // still over the cap — our estimate was too + // optimistic. Continue purging. + } + + let deleted = self.storage.delete_oldest_trades_batch(batch_size)?; + if deleted == 0 { + break; + } + total_deleted += deleted; + } + + if total_deleted > 0 { + // CHECKPOINT to merge the WAL, then VACUUM if we've + // freed a significant fraction of the data — DuckDB's + // CHECKPOINT only merges the WAL, it doesn't shrink the + // main file. VACUUM rewrites the database to reclaim + // filesystem space, but is O(n) in remaining rows so + // we only run it when the payoff is worth it. + self.storage.run_checkpoint()?; + + let remaining_rows = self.storage.count_trades()?; + + // VACUUM when we've freed at least ~20% of the + // remaining data — enough to reclaim meaningful space + // without running an expensive rewrite on every pass. + if remaining_rows == 0 || total_deleted >= remaining_rows / 4 { + self.storage.vacuum()?; + tracing::debug!( + "VACUUMed database after purging {total_deleted} rows \ + ({remaining_rows} remaining)" + ); + } + } + + Ok(total_deleted) + } + + /// Spawn the startup cleanup and the periodic background task. + /// + /// Returns the `JoinHandle` for the periodic task (and the startup + /// task is awaited internally by the periodic task). + pub fn spawn(self, shutdown: CancellationToken) -> JoinHandle<()> { + let Self { storage, config } = self; + + let (startup_tx, startup_rx) = tokio::sync::oneshot::channel(); + + // Startup cleanup on a blocking thread. + let startup_handle = { + let scheduler = CleanupScheduler { + storage: storage.clone(), + config, + }; + tokio::task::spawn_blocking(move || { + let result = scheduler.run_pass(); + let _ = startup_tx.send(result); + }) + }; + + let retention_ms = (config.retention_hours as i64) * 3_600_000; + + tokio::spawn(async move { + // Await the startup cleanup so we have an accurate + // `last_cleanup` timestamp before scheduling the next + // pass. + let mut last_cleanup = tokio::select! { + biased; + _ = shutdown.cancelled() => { + tracing::info!("Periodic cleanup shut down."); + return; + } + result = startup_rx => result.ok().flatten(), + }; + + // Also await the startup task's JoinHandle so it is + // properly joined on shutdown (not detached). + let _ = startup_handle.await; + + loop { + let time_delay = Self::delay_until_next_cleanup(last_cleanup, retention_ms); + let delay = + if config.max_storage_bytes.is_some() && SIZE_CHECK_INTERVAL < time_delay { + SIZE_CHECK_INTERVAL + } else { + time_delay + } + .max(MIN_CLEANUP_DELAY); + + tokio::select! { + biased; + _ = shutdown.cancelled() => { + tracing::info!("Periodic cleanup shut down."); + break; + } + _ = tokio::time::sleep(delay) => { + let scheduler = CleanupScheduler { + storage: storage.clone(), + config, + }; + let result = tokio::task::spawn_blocking(move || { + scheduler.run_pass() + }) + .await + .ok() + .flatten(); + if let Some(ts) = result { + last_cleanup = Some(ts); + } + } + } + } + }) + } + + fn delay_until_next_cleanup(last_cleanup_ms: Option, retention_ms: i64) -> Duration { + let now_ms = Self::now_ms(); + + let anchor_ms = last_cleanup_ms.unwrap_or_else(|| { + tracing::debug!("No last_cleanup recorded; anchoring at now"); + now_ms + }); + + let next_ms = anchor_ms + retention_ms; + if next_ms <= now_ms { + Duration::ZERO + } else { + Duration::from_millis((next_ms - now_ms) as u64) + } + } + + fn now_ms() -> i64 { + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap_or_default() + .as_millis() as i64 + } +} diff --git a/src/config.rs b/src/config.rs index 2eabcc5..3ffe8c3 100644 --- a/src/config.rs +++ b/src/config.rs @@ -78,6 +78,19 @@ pub struct Config { /// Default: 200_000 (~20–40 MB depending on symbol length). #[serde(default = "default_max_buffered_trades")] pub max_buffered_trades: usize, + + /// Optional hard cap on total DuckDB storage (main DB + WAL) in + /// megabytes. When the combined file size exceeds this value the + /// oldest trades are purged during cleanup — even if they're within + /// the time-based retention window. + /// + /// Use this to prevent the database from filling the disk on + /// constrained hosts. Default: `None` (no size cap). + /// + /// Tip: set this to ~50-80 % of your available disk space so the + /// server leaves room for system files, logs, and burst. + #[serde(default)] + pub max_storage_mb: Option, } const fn default_flush_interval() -> u64 { diff --git a/src/main.rs b/src/main.rs index 8c51b1d..d35347f 100644 --- a/src/main.rs +++ b/src/main.rs @@ -1,4 +1,5 @@ mod api; +mod cleanup; mod config; mod discovery; mod ingestion; @@ -59,7 +60,7 @@ struct App { auth_token: Option, flush_interval: std::time::Duration, max_buffered_trades: usize, - data_retention_hours: u64, + cleanup_scheduler: cleanup::CleanupScheduler, tls_config: Option, } @@ -72,7 +73,10 @@ impl App { std::process::exit(1); }); - storage.run_cleanup(config.data_retention_hours); + let cleanup_config = + cleanup::CleanupConfig::from_config(config.data_retention_hours, config.max_storage_mb); + + let cleanup_scheduler = cleanup::CleanupScheduler::new(&storage, cleanup_config); let whitelist = config.resolve_whitelist(); if !config.discovery_mode && (whitelist.is_empty() || config.base_assets.is_empty()) { @@ -198,8 +202,8 @@ impl App { auth_token: config.auth_token.clone(), flush_interval: std::time::Duration::from_millis(config.flush_interval_ms), max_buffered_trades: config.max_buffered_trades, - data_retention_hours: config.data_retention_hours, tls_config, + cleanup_scheduler, } } @@ -214,9 +218,7 @@ impl App { self.max_buffered_trades, ); - let _cleanup = self - .storage - .spawn_periodic_cleanup(self.data_retention_hours, shutdown.child_token()); + let _cleanup = self.cleanup_scheduler.spawn(shutdown.child_token()); let ingest = ingestion::start_all_ingest_tasks( &self.resolved_pairs, diff --git a/src/storage.rs b/src/storage.rs index 2a8f133..b85bc13 100644 --- a/src/storage.rs +++ b/src/storage.rs @@ -1,6 +1,6 @@ use anyhow::{Context, Result}; use duckdb::{Appender, Connection}; -use std::path::Path; +use std::path::{Path, PathBuf}; use std::sync::Arc; use std::time::Duration; @@ -11,7 +11,6 @@ use flowsurface_exchange::{Ticker, TickerInfo, UnixMs}; use crate::api::{AnnotatedTrade, TradeQuery}; use tokio::sync::mpsc; use tokio::task::JoinHandle; -use tokio_util::sync::CancellationToken; /// Information about a tracked pair with the timestamp range stored. #[derive(Debug, Clone, Copy, serde::Serialize)] @@ -33,6 +32,7 @@ pub struct PairInfo { #[derive(Clone)] pub struct Storage { db: Arc>, + data_dir: PathBuf, } impl Storage { @@ -87,9 +87,81 @@ impl Storage { Ok(Self { db: Arc::new(parking_lot::Mutex::new(root)), + data_dir: data_dir.to_path_buf(), }) } + /// Return the size (in bytes) of the main database file plus the + /// WAL file. If a file does not (yet) exist its size is counted as 0. + pub fn current_storage_bytes(&self) -> Result { + let db_path = self.data_dir.join("trades.duckdb"); + let wal_path = self.data_dir.join("trades.duckdb.wal"); + + let db_size = match std::fs::metadata(&db_path) { + Ok(meta) => meta.len(), + Err(e) if e.kind() == std::io::ErrorKind::NotFound => 0, + Err(e) => { + return Err(e).with_context(|| format!("checking size of {}", db_path.display())); + } + }; + + let wal_size = match std::fs::metadata(&wal_path) { + Ok(meta) => meta.len(), + Err(e) if e.kind() == std::io::ErrorKind::NotFound => 0, + Err(e) => { + return Err(e).with_context(|| format!("checking size of {}", wal_path.display())); + } + }; + + Ok(db_size + wal_size) + } + + /// Return the total number of rows in the `trades` table. + /// + /// In DuckDB this is a cheap metadata operation on row-group + /// headers — safe to call frequently. + pub fn count_trades(&self) -> Result { + let conn = self.connection()?; + conn.query_row("SELECT COUNT(*) FROM trades", [], |r| r.get(0)) + .context("counting trades") + } + + /// Delete the `batch_size` oldest trades (by `ts`) and return the + /// number of rows actually deleted. + pub fn delete_oldest_trades_batch(&self, batch_size: i64) -> Result { + let conn = self.connection()?; + let deleted = conn + .execute( + "DELETE FROM trades WHERE rowid IN (\ + SELECT rowid FROM trades ORDER BY ts ASC LIMIT ?\ + )", + duckdb::params![batch_size], + ) + .context("deleting oldest trades batch")?; + Ok(deleted as u64) + } + + /// Rewrite the database file to reclaim filesystem space freed by + /// prior `DELETE` operations. `CHECKPOINT` alone only merges the + /// WAL — it does not shrink the main file. `VACUUM` is O(n) in + /// remaining rows, so callers should gate it behind a threshold. + pub fn vacuum(&self) -> Result<()> { + let conn = self.connection()?; + conn.execute_batch("VACUUM;") + .context("vacuuming DuckDB database")?; + Ok(()) + } + + /// Merge the DuckDB WAL into the main database file, then truncate + /// the WAL. This prevents the `.wal` file from doubling the on-disk + /// footprint after a bulk delete. + pub fn run_checkpoint(&self) -> Result<()> { + let conn = self.connection()?; + conn.execute_batch("CHECKPOINT;") + .context("checkpointing DuckDB WAL")?; + Ok(()) + } + /// Open a dedicated connection for the batch writer (shares the /// underlying `duckdb_database`). pub fn open_writer(&self) -> Result { @@ -364,6 +436,13 @@ impl Storage { } } + fn now_ms() -> i64 { + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap_or_default() + .as_millis() as i64 + } + /// Delete every trade row whose `ts` (milliseconds since epoch) is /// older than `retention_hours`. Returns the number of deleted rows. pub fn purge_old_trades(&self, retention_hours: u64) -> Result { @@ -378,106 +457,10 @@ impl Storage { Ok(deleted as u64) } - /// Convenience: return the stored `last_cleanup` timestamp (milliseconds - /// since epoch), or `None` if cleanup has never run. - pub fn last_cleanup_ms(&self) -> Result> { - self.get_metadata("last_cleanup")? - .map(|v| v.parse::().context("parsing last_cleanup metadata")) - .transpose() - } - - pub fn record_cleanup(&self) -> Result<()> { + pub fn record_cleanup(&self) -> Result { let now_ms = Self::now_ms(); - self.set_metadata("last_cleanup", &now_ms.to_string()) - } - - /// Run a single data-retention cleanup pass. - /// - /// Deletes trades older than `retention_hours` and records the - /// `last_cleanup` timestamp so callers can avoid running it again too soon. - pub fn run_cleanup(&self, retention_hours: u64) { - match self.purge_old_trades(retention_hours) { - Ok(n) => { - if n > 0 { - tracing::info!("Cleaned up {n} trade(s) older than {retention_hours}h"); - } - if let Err(e) = self.record_cleanup() { - tracing::warn!("Failed to record last_cleanup timestamp: {e:#}"); - } - } - Err(e) => { - tracing::error!("Data cleanup failed: {e:#}"); - } - } - } - - /// Compute how long to sleep before the next cleanup is needed. - fn next_cleanup_delay(&self, retention_ms: i64) -> Duration { - let now_ms = Self::now_ms(); - - let anchor_ms = match self.last_cleanup_ms() { - Ok(Some(ts)) => ts, - Ok(None) => { - tracing::debug!("No last_cleanup recorded; anchoring at now"); - now_ms - } - Err(e) => { - tracing::warn!("Failed to read last_cleanup: {e:#}; retrying in 10 min"); - return Duration::from_secs(600); - } - }; - - let next_ms = anchor_ms + retention_ms; - if next_ms <= now_ms { - Duration::ZERO - } else { - Duration::from_millis((next_ms - now_ms) as u64) - } - } - - /// Spawn a background task that schedules the next cleanup pass based - /// on the `last_cleanup` metadata, without polling. - /// - /// After each cleanup pass (which writes `last_cleanup`), the task - /// computes `last_cleanup + retention_hours` and sleeps exactly until - /// that moment. This means wakeups only happen when data is actually - /// due for expiry — there is no periodic polling. - /// - /// A startup [`run_cleanup`](Self::run_cleanup) is expected to have been - /// called by the caller before this task is spawned so that `last_cleanup` - /// is initialised. - pub fn spawn_periodic_cleanup( - &self, - retention_hours: u64, - shutdown: CancellationToken, - ) -> JoinHandle<()> { - let storage = self.clone(); - tokio::spawn(async move { - let retention_ms = (retention_hours as i64) * 3_600_000; - - loop { - let delay = storage.next_cleanup_delay(retention_ms); - - tokio::select! { - biased; - _ = shutdown.cancelled() => { - tracing::info!("Periodic cleanup shut down."); - break; - } - _ = tokio::time::sleep(delay) => { - storage.run_cleanup(retention_hours); - } - } - } - }) - } - - /// Return the current UTC timestamp in milliseconds since the Unix epoch. - fn now_ms() -> i64 { - std::time::SystemTime::now() - .duration_since(std::time::UNIX_EPOCH) - .unwrap_or_default() - .as_millis() as i64 + self.set_metadata("last_cleanup", &now_ms.to_string())?; + Ok(now_ms) } /// Spawn a background task that receives trades on `rx`, buffers them, From 62748a22c375ec97428d45a4602a29fb71c71dcd Mon Sep 17 00:00:00 2001 From: akenshaw <63060680+akenshaw@users.noreply.github.com> Date: Sat, 18 Jul 2026 08:26:42 +0300 Subject: [PATCH 2/8] harden cleanup: validate retention hours, retry vacuum, fix delay --- src/cleanup.rs | 37 ++++++++++++++++++++++++++----------- src/main.rs | 6 +++++- src/storage.rs | 37 +++++++++++++++++++++++++++++++++---- 3 files changed, 64 insertions(+), 16 deletions(-) diff --git a/src/cleanup.rs b/src/cleanup.rs index 14397fd..7ae527b 100644 --- a/src/cleanup.rs +++ b/src/cleanup.rs @@ -62,14 +62,23 @@ pub struct CleanupConfig { impl CleanupConfig { /// Derive the config from the user-facing `max_storage_mb` setting. /// A value of `0` or `None` means no cap. - pub fn from_config(retention_hours: u64, max_storage_mb: Option) -> Self { + /// + /// Returns an error if `retention_hours` is zero, which would purge + /// all trades on every cleanup pass. + pub fn from_config(retention_hours: u64, max_storage_mb: Option) -> anyhow::Result { + if retention_hours == 0 { + anyhow::bail!( + "data_retention_hours must be > 0 (got 0) — \ + a zero retention period would purge all trades on every cleanup pass" + ); + } let max_storage_bytes = max_storage_mb .filter(|&mb| mb > 0) .map(|mb| mb.saturating_mul(1024 * 1024)); - Self { + Ok(Self { retention_hours, max_storage_bytes, - } + }) } } @@ -137,7 +146,7 @@ impl CleanupScheduler { /// Returns `Some(ts)` when the time-based purge succeeded and the /// `last_cleanup` timestamp was recorded, or `None` on failure. fn run_pass(&self) -> Option { - let mut cleaned_anything = false; + let mut time_purge_deleted = false; let mut time_cleanup_ok = false; match self.storage.purge_old_trades(self.config.retention_hours) { @@ -148,7 +157,7 @@ impl CleanupScheduler { "Cleaned up {n} trade(s) older than {}h", self.config.retention_hours ); - cleaned_anything = true; + time_purge_deleted = true; } } Err(e) => { @@ -156,6 +165,7 @@ impl CleanupScheduler { } } + let size_cap_ran = self.config.max_storage_bytes.is_some(); if let Some(max_bytes) = self.config.max_storage_bytes { match self.purge_oldest_trades_until_below(max_bytes) { Ok(n) => { @@ -170,7 +180,6 @@ impl CleanupScheduler { (max {} MB, now ~{current_mb} MB)", max_bytes / (1024 * 1024), ); - cleaned_anything = true; } } Err(e) => { @@ -179,7 +188,10 @@ impl CleanupScheduler { } } - if cleaned_anything && let Err(e) = self.storage.run_checkpoint() { + if time_purge_deleted + && !size_cap_ran + && let Err(e) = self.storage.run_checkpoint() + { tracing::warn!("Failed to checkpoint DuckDB WAL after cleanup: {e:#}"); } @@ -359,10 +371,13 @@ impl CleanupScheduler { fn delay_until_next_cleanup(last_cleanup_ms: Option, retention_ms: i64) -> Duration { let now_ms = Self::now_ms(); - let anchor_ms = last_cleanup_ms.unwrap_or_else(|| { - tracing::debug!("No last_cleanup recorded; anchoring at now"); - now_ms - }); + let Some(anchor_ms) = last_cleanup_ms else { + // No last_cleanup recorded — either this is the first run + // or the previous cleanup pass failed. Return zero so the + // caller (which applies `MIN_CLEANUP_DELAY`) retries quickly + // rather than waiting a full retention period. + return Duration::ZERO; + }; let next_ms = anchor_ms + retention_ms; if next_ms <= now_ms { diff --git a/src/main.rs b/src/main.rs index d35347f..808203d 100644 --- a/src/main.rs +++ b/src/main.rs @@ -74,7 +74,11 @@ impl App { }); let cleanup_config = - cleanup::CleanupConfig::from_config(config.data_retention_hours, config.max_storage_mb); + cleanup::CleanupConfig::from_config(config.data_retention_hours, config.max_storage_mb) + .unwrap_or_else(|e| { + tracing::error!("Invalid cleanup configuration: {e:#}"); + std::process::exit(1); + }); let cleanup_scheduler = cleanup::CleanupScheduler::new(&storage, cleanup_config); diff --git a/src/storage.rs b/src/storage.rs index b85bc13..54eedd6 100644 --- a/src/storage.rs +++ b/src/storage.rs @@ -145,11 +145,40 @@ impl Storage { /// prior `DELETE` operations. `CHECKPOINT` alone only merges the /// WAL — it does not shrink the main file. `VACUUM` is O(n) in /// remaining rows, so callers should gate it behind a threshold. + /// + /// `VACUUM` requires exclusive table access, so it can fail with a + /// transaction conflict if the batch flusher is mid-append. This + /// method retries a few times with short sleeps — the flusher's + /// appender is only held open for a few milliseconds per flush, so + /// a brief wait is almost always enough. Safe to call from a + /// blocking thread (which is where cleanup always runs). pub fn vacuum(&self) -> Result<()> { - let conn = self.connection()?; - conn.execute_batch("VACUUM;") - .context("vacuuming DuckDB database")?; - Ok(()) + const MAX_RETRIES: u32 = 3; + const RETRY_DELAY: std::time::Duration = std::time::Duration::from_secs(1); + + let mut last_err = None; + for attempt in 0..MAX_RETRIES { + let conn = self.connection()?; + match conn.execute_batch("VACUUM;") { + Ok(()) => return Ok(()), + Err(e) => { + last_err = Some(e); + if attempt + 1 < MAX_RETRIES { + tracing::debug!( + "VACUUM attempt {}/{} failed (likely batch flusher \ + holding appender); retrying in {RETRY_DELAY:?}", + attempt + 1, + MAX_RETRIES, + ); + std::thread::sleep(RETRY_DELAY); + } + } + } + } + Err(last_err.unwrap_or_else(|| { + duckdb::Error::InvalidParameterName("VACUUM produced no error".into()) + })) + .with_context(|| format!("vacuuming DuckDB database after {MAX_RETRIES} attempts")) } /// Merge the DuckDB WAL into the main database file, then truncate From 4f94b01bf81f46541683ee4ac86a53948278078d Mon Sep 17 00:00:00 2001 From: akenshaw <63060680+akenshaw@users.noreply.github.com> Date: Sat, 18 Jul 2026 09:31:31 +0300 Subject: [PATCH 3/8] streamline comments, enhance purge outcome handling --- src/cleanup.rs | 305 ++++++++++++++++++------------------------------- src/storage.rs | 9 +- 2 files changed, 114 insertions(+), 200 deletions(-) diff --git a/src/cleanup.rs b/src/cleanup.rs index 7ae527b..838e2aa 100644 --- a/src/cleanup.rs +++ b/src/cleanup.rs @@ -6,65 +6,43 @@ use tokio_util::sync::CancellationToken; use crate::storage::Storage; -/// Minimum delay between cleanup attempts, regardless of schedule. -/// -/// This prevents a tight retry loop when `delay_until_next_cleanup` -/// returns zero (e.g. because the last time-based cleanup failed and -/// `last_cleanup` is stale). +/// Floor on delay between cleanup passes; prevents tight retry loops +/// when `last_cleanup` is stale (first run or previous pass failed). const MIN_CLEANUP_DELAY: Duration = Duration::from_secs(60); -/// How often the size-based cleanup task wakes up to check whether the -/// database has exceeded the `max_storage_mb` cap. -/// -/// This interval only matters when a size cap is configured *and* the -/// time-based retention window is longer than 5 minutes — otherwise the -/// time-based schedule dominates. +/// Wake interval for the size-cap check. Only matters when a size cap +/// is configured and the retention window is longer than this. const SIZE_CHECK_INTERVAL: Duration = Duration::from_secs(300); -/// Estimated on-disk bytes per trade row. -/// -/// DuckDB's columnar storage compresses the schema well: exchange -/// names dictionary-compress to ~1–2 bytes, symbols to ~2–5 bytes, -/// timestamps benefit from delta encoding, and `DOUBLE` columns -/// compress modestly. Empirically ~40–60 bytes per row including -/// row-group overhead. -/// -/// Used only as a fast-path threshold in the size-based purge loop — -/// the authoritative check is the actual filesystem file size after -/// `CHECKPOINT`. A conservative (lower) value here means we more -/// often verify with the real file size, which is safer. +/// Safety ceiling for the size-based purge loop (200 × 100k = 20M rows). +const MAX_PURGE_ITERATIONS: usize = 200; + +/// Conservative estimate of on-disk bytes per trade row. Used only as +/// a fast-path threshold; the authoritative check is the real file +/// size after `CHECKPOINT`. const EST_BYTES_PER_ROW: u64 = 50; -/// Choose a batch size proportional to `max_bytes` so that each purge -/// iteration deletes ~10% of the cap. -/// -/// Clamped to **[1 000, 100 000]** so tiny caps still make progress -/// and huge caps don't create giant transactions. +/// Batch size for each purge iteration — ~10% of the cap, clamped to +/// [1 000, 100 000]. fn purge_batch_size(max_bytes: u64) -> i64 { const MIN_ROWS: i64 = 1_000; const MAX_ROWS: i64 = 100_000; - - // Target ~10% of the cap per iteration. let rows = (max_bytes / 10 / EST_BYTES_PER_ROW) as i64; rows.clamp(MIN_ROWS, MAX_ROWS) } -/// Groups the two retention parameters that are always passed together. #[derive(Debug, Clone, Copy)] pub struct CleanupConfig { - /// Time-based retention: trades older than this are purged. + /// Trades older than this are purged. pub retention_hours: u64, - /// Optional hard cap on total DB+WAL size in bytes. When set, the - /// oldest trades are purged even if within the retention window. + /// Optional hard cap on total DB+WAL size in bytes. pub max_storage_bytes: Option, } impl CleanupConfig { - /// Derive the config from the user-facing `max_storage_mb` setting. - /// A value of `0` or `None` means no cap. - /// - /// Returns an error if `retention_hours` is zero, which would purge - /// all trades on every cleanup pass. + /// Derive from user-facing settings. `max_storage_mb` of `0` or + /// `None` disables the cap. Returns an error if + /// `retention_hours` is zero. pub fn from_config(retention_hours: u64, max_storage_mb: Option) -> anyhow::Result { if retention_hours == 0 { anyhow::bail!( @@ -82,20 +60,14 @@ impl CleanupConfig { } } -/// Owns the cleanup lifecycle: startup guard, startup pass, and the -/// periodic background task. -/// -/// Created in [`CleanupScheduler::new`] which runs the startup guard. -/// Call [`spawn`](Self::spawn) to start the periodic task — this also -/// runs the startup cleanup pass on a blocking thread and feeds its -/// result into the scheduler so the first periodic delay is accurate. pub struct CleanupScheduler { storage: Storage, config: CleanupConfig, } impl CleanupScheduler { - /// Create the scheduler and run the startup guard. + /// Create the scheduler and run the startup guard, which exits + /// the process if the database is > 2× the configured cap. pub fn new(storage: &Storage, config: CleanupConfig) -> Self { if let Some(max) = config.max_storage_bytes { tracing::info!( @@ -140,13 +112,14 @@ impl CleanupScheduler { } } - /// Run a single cleanup pass (time-based retention + optional size - /// cap + checkpoint + record timestamp). + /// Run one cleanup pass: time-based retention, optional size-cap + /// purge, checkpoint, and record `last_cleanup`. /// - /// Returns `Some(ts)` when the time-based purge succeeded and the - /// `last_cleanup` timestamp was recorded, or `None` on failure. + /// Returns `Some(last_cleanup_ts)` on success, or `None` if the + /// pass failed (e.g. time-based purge errored or recording the + /// timestamp failed). fn run_pass(&self) -> Option { - let mut time_purge_deleted = false; + let mut any_deleted = false; let mut time_cleanup_ok = false; match self.storage.purge_old_trades(self.config.retention_hours) { @@ -157,7 +130,7 @@ impl CleanupScheduler { "Cleaned up {n} trade(s) older than {}h", self.config.retention_hours ); - time_purge_deleted = true; + any_deleted = true; } } Err(e) => { @@ -165,172 +138,119 @@ impl CleanupScheduler { } } - let size_cap_ran = self.config.max_storage_bytes.is_some(); if let Some(max_bytes) = self.config.max_storage_bytes { match self.purge_oldest_trades_until_below(max_bytes) { - Ok(n) => { - if n > 0 { - let current_mb = self - .storage - .current_storage_bytes() - .map(|b| b / (1024 * 1024)) - .unwrap_or(0); - tracing::info!( - "Cleaned up {n} trade(s) to keep storage under cap \ - (max {} MB, now ~{current_mb} MB)", - max_bytes / (1024 * 1024), - ); - } - } - Err(e) => { - tracing::error!("Storage-cap cleanup failed: {e:#}"); - } + Ok(n) if n > 0 => any_deleted = true, + Err(e) => tracing::error!("Storage-cap cleanup failed: {e:#}"), + _ => {} } } - if time_purge_deleted - && !size_cap_ran - && let Err(e) = self.storage.run_checkpoint() - { - tracing::warn!("Failed to checkpoint DuckDB WAL after cleanup: {e:#}"); + if any_deleted { + if let Err(e) = self.storage.run_checkpoint() { + tracing::warn!("Failed to checkpoint DuckDB WAL after cleanup: {e:#}"); + } } - let mut last_cleanup_ts: Option = None; if time_cleanup_ok { match self.storage.record_cleanup() { - Ok(ts) => last_cleanup_ts = Some(ts), + Ok(ts) => return Some(ts), Err(e) => tracing::warn!("Failed to record last_cleanup timestamp: {e:#}"), } } - last_cleanup_ts + None } - /// Delete the oldest trades until the on-disk size drops below - /// `max_bytes`. Returns the total number of deleted rows. - /// - /// The authoritative convergence criterion is the actual filesystem - /// size after `CHECKPOINT`. A row-count estimate - /// (`COUNT(*) × EST_BYTES_PER_ROW`) is used as a fast path: when - /// the row count is clearly above the estimated target we purge - /// without an expensive `CHECKPOINT`; when it's below we verify - /// with the real file size to guard against estimation errors. - /// - /// After purging, if a significant fraction of the data was freed, - /// a `VACUUM` is run to reclaim filesystem space — `CHECKPOINT` - /// alone doesn't shrink the main database file. - fn purge_oldest_trades_until_below(&self, max_bytes: u64) -> Result { - /// Safety ceiling: if the purge hasn't converged after this - /// many iterations, something is wrong. 200 × 100k batch = - /// 20M rows — far beyond any plausible cap. Bail out rather - /// than looping forever. - const MAX_ITERATIONS: usize = 200; + /// Check whether storage is at or below `max_bytes`, using a + /// two-tier test: a cheap row-count estimate first, falling back + /// to an authoritative file-size check (which requires a + /// `CHECKPOINT`) only when the estimate is borderline. + fn is_storage_under_cap(&self, max_bytes: u64, max_est_rows: u64) -> Result { + if self.storage.count_trades()? > max_est_rows { + return Ok(false); + } + self.storage.run_checkpoint()?; + Ok(self.storage.current_storage_bytes()? <= max_bytes) + } + /// Delete oldest trades until on-disk size ≤ `max_bytes`. + fn purge_oldest_trades_until_below(&self, max_bytes: u64) -> Result { let max_est_rows = max_bytes.saturating_div(EST_BYTES_PER_ROW); let batch_size = purge_batch_size(max_bytes); let mut total_deleted = 0u64; + let mut converged = false; - for _iteration in 0..MAX_ITERATIONS { - // ── Fast path: row count ─────────────────────────── - // COUNT(*) in DuckDB is a metadata operation on - // row-group headers — cheap even on large tables. - // - // If the row count is clearly above the estimated - // target we purge without an expensive CHECKPOINT. - // If it's below, the estimate may be wrong so we - // fall through to the authoritative file-size check. - let current_rows = self.storage.count_trades()?; - - if current_rows <= max_est_rows { - // ── Authoritative criterion: file size ────── - // CHECKPOINT first so the WAL doesn't inflate - // the filesystem size check. - self.storage.run_checkpoint()?; - - let file_size = self.storage.current_storage_bytes()?; - if file_size <= max_bytes { - tracing::debug!( - "Size-based purge: file size {file_size} ≤ \ - cap {max_bytes}, stopping" - ); - break; - } - // Row count says we're under, but the file is - // still over the cap — our estimate was too - // optimistic. Continue purging. + for _ in 0..MAX_PURGE_ITERATIONS { + if self.is_storage_under_cap(max_bytes, max_est_rows)? { + converged = true; + break; } - let deleted = self.storage.delete_oldest_trades_batch(batch_size)?; if deleted == 0 { + converged = true; break; } total_deleted += deleted; } if total_deleted > 0 { - // CHECKPOINT to merge the WAL, then VACUUM if we've - // freed a significant fraction of the data — DuckDB's - // CHECKPOINT only merges the WAL, it doesn't shrink the - // main file. VACUUM rewrites the database to reclaim - // filesystem space, but is O(n) in remaining rows so - // we only run it when the payoff is worth it. self.storage.run_checkpoint()?; - let remaining_rows = self.storage.count_trades()?; - // VACUUM when we've freed at least ~20% of the - // remaining data — enough to reclaim meaningful space - // without running an expensive rewrite on every pass. - if remaining_rows == 0 || total_deleted >= remaining_rows / 4 { - self.storage.vacuum()?; - tracing::debug!( - "VACUUMed database after purging {total_deleted} rows \ - ({remaining_rows} remaining)" + if converged { + let current_mb = self + .storage + .current_storage_bytes() + .map(|b| b / (1024 * 1024)) + .unwrap_or(0); + tracing::info!( + "Cleaned up {total_deleted} trade(s) to keep storage under cap \ + (max {} MB, now ~{current_mb} MB)", + max_bytes / (1024 * 1024), + ); + } else { + tracing::warn!( + "Size-cap purge did not converge after \ + {MAX_PURGE_ITERATIONS} iterations (deleted {total_deleted} \ + rows); storage may still exceed the {} MB cap", + max_bytes / (1024 * 1024), ); } + + if should_vacuum(total_deleted, remaining_rows) { + if let Err(e) = self.storage.vacuum() { + tracing::warn!( + "VACUUM after size-cap purge failed (data is \ + correct, but filesystem space wasn't reclaimed): \ + {e:#}" + ); + } else { + tracing::debug!( + "VACUUMed database after purging {total_deleted} \ + rows ({remaining_rows} remaining)" + ); + } + } } Ok(total_deleted) } - /// Spawn the startup cleanup and the periodic background task. - /// - /// Returns the `JoinHandle` for the periodic task (and the startup - /// task is awaited internally by the periodic task). + /// Spawn the startup cleanup and periodic background task. pub fn spawn(self, shutdown: CancellationToken) -> JoinHandle<()> { - let Self { storage, config } = self; - - let (startup_tx, startup_rx) = tokio::sync::oneshot::channel(); - - // Startup cleanup on a blocking thread. - let startup_handle = { - let scheduler = CleanupScheduler { - storage: storage.clone(), - config, - }; - tokio::task::spawn_blocking(move || { - let result = scheduler.run_pass(); - let _ = startup_tx.send(result); - }) - }; + let storage = self.storage.clone(); + let config = self.config; let retention_ms = (config.retention_hours as i64) * 3_600_000; tokio::spawn(async move { - // Await the startup cleanup so we have an accurate - // `last_cleanup` timestamp before scheduling the next - // pass. - let mut last_cleanup = tokio::select! { - biased; - _ = shutdown.cancelled() => { - tracing::info!("Periodic cleanup shut down."); - return; - } - result = startup_rx => result.ok().flatten(), - }; - - // Also await the startup task's JoinHandle so it is - // properly joined on shutdown (not detached). - let _ = startup_handle.await; + // Run startup cleanup pass on a blocking thread first. + let mut last_cleanup = tokio::task::spawn_blocking({ + let storage = storage.clone(); + move || CleanupScheduler { storage, config }.run_pass() + }) + .await + .unwrap_or(None); loop { let time_delay = Self::delay_until_next_cleanup(last_cleanup, retention_ms); @@ -349,16 +269,14 @@ impl CleanupScheduler { break; } _ = tokio::time::sleep(delay) => { - let scheduler = CleanupScheduler { - storage: storage.clone(), - config, - }; - let result = tokio::task::spawn_blocking(move || { - scheduler.run_pass() + let result = tokio::task::spawn_blocking({ + let storage = storage.clone(); + move || { + CleanupScheduler { storage, config }.run_pass() + } }) .await - .ok() - .flatten(); + .unwrap_or(None); if let Some(ts) = result { last_cleanup = Some(ts); } @@ -369,13 +287,11 @@ impl CleanupScheduler { } fn delay_until_next_cleanup(last_cleanup_ms: Option, retention_ms: i64) -> Duration { - let now_ms = Self::now_ms(); + let now_ms = Storage::now_ms(); let Some(anchor_ms) = last_cleanup_ms else { - // No last_cleanup recorded — either this is the first run - // or the previous cleanup pass failed. Return zero so the - // caller (which applies `MIN_CLEANUP_DELAY`) retries quickly - // rather than waiting a full retention period. + // No recorded timestamp — retry quickly (caller applies + // `MIN_CLEANUP_DELAY`). return Duration::ZERO; }; @@ -386,11 +302,10 @@ impl CleanupScheduler { Duration::from_millis((next_ms - now_ms) as u64) } } +} - fn now_ms() -> i64 { - std::time::SystemTime::now() - .duration_since(std::time::UNIX_EPOCH) - .unwrap_or_default() - .as_millis() as i64 - } +/// Whether VACUUM is worth running after a purge — only when we've +/// freed at least ~25% of the remaining data. +fn should_vacuum(total_deleted: u64, remaining_rows: u64) -> bool { + remaining_rows == 0 || total_deleted >= remaining_rows / 4 } diff --git a/src/storage.rs b/src/storage.rs index 54eedd6..4ac7de1 100644 --- a/src/storage.rs +++ b/src/storage.rs @@ -175,10 +175,9 @@ impl Storage { } } } - Err(last_err.unwrap_or_else(|| { - duckdb::Error::InvalidParameterName("VACUUM produced no error".into()) - })) - .with_context(|| format!("vacuuming DuckDB database after {MAX_RETRIES} attempts")) + // SAFETY: the loop always sets `last_err` before reaching this point. + Err(last_err.unwrap()) + .with_context(|| format!("vacuuming DuckDB database after {MAX_RETRIES} attempts")) } /// Merge the DuckDB WAL into the main database file, then truncate @@ -465,7 +464,7 @@ impl Storage { } } - fn now_ms() -> i64 { + pub(crate) fn now_ms() -> i64 { std::time::SystemTime::now() .duration_since(std::time::UNIX_EPOCH) .unwrap_or_default() From e55f36ca8bb91c384d1b0a60fb741c9576f24f2b Mon Sep 17 00:00:00 2001 From: akenshaw <63060680+akenshaw@users.noreply.github.com> Date: Sat, 18 Jul 2026 10:19:55 +0300 Subject: [PATCH 4/8] separate startup pass, add headroom fraction --- src/cleanup.rs | 111 +++++++++++++++++++++++++++---------------------- src/main.rs | 17 +++++++- 2 files changed, 76 insertions(+), 52 deletions(-) diff --git a/src/cleanup.rs b/src/cleanup.rs index 838e2aa..cb4b2c6 100644 --- a/src/cleanup.rs +++ b/src/cleanup.rs @@ -22,6 +22,10 @@ const MAX_PURGE_ITERATIONS: usize = 200; /// size after `CHECKPOINT`. const EST_BYTES_PER_ROW: u64 = 50; +/// Target fraction of the cap to purge down to, leaving headroom for +/// incoming trades between periodic checks. 4/5 = 80 %. +const HEADROOM_FRACTION: u64 = 4; + /// Batch size for each purge iteration — ~10% of the cap, clamped to /// [1 000, 100 000]. fn purge_batch_size(max_bytes: u64) -> i64 { @@ -60,15 +64,20 @@ impl CleanupConfig { } } +#[derive(Clone)] pub struct CleanupScheduler { storage: Storage, config: CleanupConfig, } impl CleanupScheduler { - /// Create the scheduler and run the startup guard, which exits - /// the process if the database is > 2× the configured cap. - pub fn new(storage: &Storage, config: CleanupConfig) -> Self { + /// Create the scheduler and check the startup guard: if the database + /// is > 2× the configured cap the process exits (this is a + /// misconfiguration that would take too long to recover from at + /// startup). + /// + /// Returns an error when the storage size cannot be read at all. + pub fn new(storage: &Storage, config: CleanupConfig) -> anyhow::Result { if let Some(max) = config.max_storage_bytes { tracing::info!( "Storage hard cap enabled: {} MB (will purge oldest trades when exceeded)", @@ -77,7 +86,7 @@ impl CleanupScheduler { match storage.current_storage_bytes() { Ok(current) if current > max.saturating_mul(2) => { - tracing::error!( + anyhow::bail!( "Database is {} MB — more than 2x the configured \ max_storage_mb cap ({} MB). Purging that much data \ at startup would take too long; this likely indicates \ @@ -88,7 +97,6 @@ impl CleanupScheduler { max / (1024 * 1024), (current / (1024 * 1024)).saturating_add(1), ); - std::process::exit(1); } Ok(current) if current > max => { tracing::warn!( @@ -99,17 +107,16 @@ impl CleanupScheduler { ); } Err(e) => { - tracing::error!("Failed to check storage size at startup: {e:#}"); - std::process::exit(1); + anyhow::bail!("Failed to check storage size at startup: {e:#}"); } _ => {} } } - Self { + Ok(Self { storage: storage.clone(), config, - } + }) } /// Run one cleanup pass: time-based retention, optional size-cap @@ -118,7 +125,7 @@ impl CleanupScheduler { /// Returns `Some(last_cleanup_ts)` on success, or `None` if the /// pass failed (e.g. time-based purge errored or recording the /// timestamp failed). - fn run_pass(&self) -> Option { + pub fn run_pass(&self) -> Option { let mut any_deleted = false; let mut time_cleanup_ok = false; @@ -146,10 +153,8 @@ impl CleanupScheduler { } } - if any_deleted { - if let Err(e) = self.storage.run_checkpoint() { - tracing::warn!("Failed to checkpoint DuckDB WAL after cleanup: {e:#}"); - } + if any_deleted && let Err(e) = self.storage.run_checkpoint() { + tracing::warn!("Failed to checkpoint DuckDB WAL after cleanup: {e:#}"); } if time_cleanup_ok { @@ -161,27 +166,26 @@ impl CleanupScheduler { None } - /// Check whether storage is at or below `max_bytes`, using a - /// two-tier test: a cheap row-count estimate first, falling back - /// to an authoritative file-size check (which requires a - /// `CHECKPOINT`) only when the estimate is borderline. - fn is_storage_under_cap(&self, max_bytes: u64, max_est_rows: u64) -> Result { - if self.storage.count_trades()? > max_est_rows { - return Ok(false); - } - self.storage.run_checkpoint()?; - Ok(self.storage.current_storage_bytes()? <= max_bytes) - } - - /// Delete oldest trades until on-disk size ≤ `max_bytes`. + /// Delete oldest trades until on-disk size is estimated to be under the cap. + /// + /// To avoid per-iteration CHECKPOINT overhead we use a two-phase approach: + /// 1. **Row-count estimate loop** — delete batches until the estimated row + /// count is below the headroom-adjusted threshold. No CHECKPOINTs here. + /// 2. **CHECKPOINT once**, then verify the real file size. If still over + /// the absolute cap we log a warning — the next periodic pass will retry. + /// + /// Targets `HEADROOM_FRACTION / 5` of the cap so there is headroom for + /// incoming trades between 5-minute checks. fn purge_oldest_trades_until_below(&self, max_bytes: u64) -> Result { - let max_est_rows = max_bytes.saturating_div(EST_BYTES_PER_ROW); - let batch_size = purge_batch_size(max_bytes); + let target_bytes = max_bytes.saturating_mul(HEADROOM_FRACTION) / 5; + let max_est_rows = target_bytes.saturating_div(EST_BYTES_PER_ROW); + let batch_size = purge_batch_size(target_bytes); let mut total_deleted = 0u64; let mut converged = false; for _ in 0..MAX_PURGE_ITERATIONS { - if self.is_storage_under_cap(max_bytes, max_est_rows)? { + // Cheap row-count check (no CHECKPOINT needed). + if self.storage.count_trades()? <= max_est_rows { converged = true; break; } @@ -198,16 +202,22 @@ impl CleanupScheduler { let remaining_rows = self.storage.count_trades()?; if converged { - let current_mb = self - .storage - .current_storage_bytes() - .map(|b| b / (1024 * 1024)) - .unwrap_or(0); - tracing::info!( - "Cleaned up {total_deleted} trade(s) to keep storage under cap \ - (max {} MB, now ~{current_mb} MB)", - max_bytes / (1024 * 1024), - ); + let current_bytes = self.storage.current_storage_bytes()?; + if current_bytes <= max_bytes { + tracing::info!( + "Cleaned up {total_deleted} trade(s) to keep storage under \ + {} MB cap (now ~{} MB)", + max_bytes / (1024 * 1024), + current_bytes / (1024 * 1024), + ); + } else { + tracing::warn!( + "Storage ({} MB) still exceeds {} MB cap after cleanup. \ + The next periodic pass will retry.", + current_bytes / (1024 * 1024), + max_bytes / (1024 * 1024), + ); + } } else { tracing::warn!( "Size-cap purge did not converge after \ @@ -236,21 +246,22 @@ impl CleanupScheduler { Ok(total_deleted) } - /// Spawn the startup cleanup and periodic background task. - pub fn spawn(self, shutdown: CancellationToken) -> JoinHandle<()> { + /// Spawn the periodic background cleanup task. + /// + /// A one-shot startup pass must have been run beforehand (via + /// [`run_startup_pass`](Self::run_startup_pass)) so that + /// `last_cleanup` is initialised. The loop then computes the + /// next wake-up from `last_cleanup` and only polls when work + /// is actually due (or every `SIZE_CHECK_INTERVAL` when a size + /// cap is configured). + pub fn spawn(self, last_cleanup: Option, shutdown: CancellationToken) -> JoinHandle<()> { let storage = self.storage.clone(); let config = self.config; let retention_ms = (config.retention_hours as i64) * 3_600_000; tokio::spawn(async move { - // Run startup cleanup pass on a blocking thread first. - let mut last_cleanup = tokio::task::spawn_blocking({ - let storage = storage.clone(); - move || CleanupScheduler { storage, config }.run_pass() - }) - .await - .unwrap_or(None); + let mut last_cleanup = last_cleanup; loop { let time_delay = Self::delay_until_next_cleanup(last_cleanup, retention_ms); @@ -306,6 +317,6 @@ impl CleanupScheduler { /// Whether VACUUM is worth running after a purge — only when we've /// freed at least ~25% of the remaining data. -fn should_vacuum(total_deleted: u64, remaining_rows: u64) -> bool { +const fn should_vacuum(total_deleted: u64, remaining_rows: u64) -> bool { remaining_rows == 0 || total_deleted >= remaining_rows / 4 } diff --git a/src/main.rs b/src/main.rs index 808203d..4a3d923 100644 --- a/src/main.rs +++ b/src/main.rs @@ -80,7 +80,11 @@ impl App { std::process::exit(1); }); - let cleanup_scheduler = cleanup::CleanupScheduler::new(&storage, cleanup_config); + let cleanup_scheduler = cleanup::CleanupScheduler::new(&storage, cleanup_config) + .unwrap_or_else(|e| { + tracing::error!("{e:#}"); + std::process::exit(1); + }); let whitelist = config.resolve_whitelist(); if !config.discovery_mode && (whitelist.is_empty() || config.base_assets.is_empty()) { @@ -216,13 +220,22 @@ impl App { let (trade_tx, trade_rx) = mpsc::unbounded_channel::(); let shutdown = CancellationToken::new(); + let cleanup_last_run = tokio::task::spawn_blocking({ + let scheduler = self.cleanup_scheduler.clone(); + move || scheduler.run_pass() + }) + .await + .unwrap_or(None); + let flusher = self.storage.spawn_batch_flusher( trade_rx, self.flush_interval, self.max_buffered_trades, ); - let _cleanup = self.cleanup_scheduler.spawn(shutdown.child_token()); + let _cleanup = self + .cleanup_scheduler + .spawn(cleanup_last_run, shutdown.child_token()); let ingest = ingestion::start_all_ingest_tasks( &self.resolved_pairs, From 8438830be2fd89a0bed3d3e9b190e64329233e5d Mon Sep 17 00:00:00 2001 From: akenshaw <63060680+akenshaw@users.noreply.github.com> Date: Sat, 18 Jul 2026 11:14:39 +0300 Subject: [PATCH 5/8] add domain newtypes - `BearerToken`, `RetentionHours` and `StorageBytes` --- src/api.rs | 30 ++++---- src/cleanup.rs | 57 ++++++++-------- src/config.rs | 182 +++++++++++++++++++++++++++++++++++++++++++------ src/main.rs | 21 +++--- src/storage.rs | 9 +-- 5 files changed, 218 insertions(+), 81 deletions(-) diff --git a/src/api.rs b/src/api.rs index e363ce4..c63e565 100644 --- a/src/api.rs +++ b/src/api.rs @@ -17,7 +17,10 @@ use flowsurface_exchange::{ }; use serde::{Deserialize, Serialize}; -use crate::storage::{PairInfo, Storage}; +use crate::{ + config::BearerToken, + storage::{PairInfo, Storage}, +}; #[derive(Serialize)] #[serde(untagged)] @@ -100,7 +103,7 @@ pub fn exchange_from_venue_market(venue: &str, market: &str) -> Option { pub struct Server { pub storage: Storage, pub startup: Instant, - pub auth_token: Option, + pub auth_token: Option, /// The tickers configured at startup. /// Used by `/pairs` to include pairs that have not yet received trades. pub configured_pairs: Vec, @@ -115,7 +118,7 @@ pub struct Server { impl Server { pub fn new( storage: Storage, - auth_token: Option, + auth_token: Option, configured_pairs: Vec, available_tickers: HashMap>, tls_config: Option, @@ -146,9 +149,7 @@ impl Server { .and_then(|v| v.to_str().ok()) .unwrap_or(""); - let expected = format!("Bearer {expected_token}"); - - if !provided.eq_ignore_ascii_case(&expected) { + if !expected_token.is_valid_authorization(provided) { let truncated: String = provided.chars().take(20).collect(); tracing::warn!( "Auth failure from {}: expected valid Bearer token, got '{truncated}'", @@ -312,12 +313,7 @@ impl Server { /// /// Uses plain HTTP for loopback addresses, HTTPS with a self-signed /// certificate for non-loopback (remote) binds. Exits on bind failure. - pub async fn serve(self: Arc, bind_address: &str) -> tokio::task::JoinHandle<()> { - let addr: SocketAddr = bind_address.parse().unwrap_or_else(|e| { - tracing::error!("Invalid bind_address '{bind_address}': {e}"); - std::process::exit(1); - }); - + pub async fn serve(self: Arc, bind_address: SocketAddr) -> tokio::task::JoinHandle<()> { let tls_config = self.tls_config.clone(); // Public routes — no auth required @@ -341,17 +337,17 @@ impl Server { tokio::spawn(async move { if let Some(cfg) = tls_config { - tracing::info!("Starting HTTPS API on {addr}"); - axum_server::bind_rustls(addr, cfg) + tracing::info!("Starting HTTPS API on {bind_address}"); + axum_server::bind_rustls(bind_address, cfg) .serve(router.into_make_service_with_connect_info::()) .await .unwrap(); } else { - tracing::info!("Starting HTTP API on {addr}"); - let listener = tokio::net::TcpListener::bind(addr) + tracing::info!("Starting HTTP API on {bind_address}"); + let listener = tokio::net::TcpListener::bind(bind_address) .await .unwrap_or_else(|e| { - tracing::error!("Failed to bind to {addr}: {e}"); + tracing::error!("Failed to bind to {bind_address}: {e}"); std::process::exit(1); }); axum::serve( diff --git a/src/cleanup.rs b/src/cleanup.rs index cb4b2c6..c1a24bc 100644 --- a/src/cleanup.rs +++ b/src/cleanup.rs @@ -4,7 +4,10 @@ use anyhow::Result; use tokio::task::JoinHandle; use tokio_util::sync::CancellationToken; -use crate::storage::Storage; +use crate::{ + config::{RetentionHours, StorageBytes}, + storage::Storage, +}; /// Floor on delay between cleanup passes; prevents tight retry loops /// when `last_cleanup` is stale (first run or previous pass failed). @@ -38,9 +41,9 @@ fn purge_batch_size(max_bytes: u64) -> i64 { #[derive(Debug, Clone, Copy)] pub struct CleanupConfig { /// Trades older than this are purged. - pub retention_hours: u64, + pub retention_hours: RetentionHours, /// Optional hard cap on total DB+WAL size in bytes. - pub max_storage_bytes: Option, + pub max_storage_bytes: Option, } impl CleanupConfig { @@ -48,15 +51,10 @@ impl CleanupConfig { /// `None` disables the cap. Returns an error if /// `retention_hours` is zero. pub fn from_config(retention_hours: u64, max_storage_mb: Option) -> anyhow::Result { - if retention_hours == 0 { - anyhow::bail!( - "data_retention_hours must be > 0 (got 0) — \ - a zero retention period would purge all trades on every cleanup pass" - ); - } + let retention_hours = RetentionHours::new(retention_hours)?; let max_storage_bytes = max_storage_mb .filter(|&mb| mb > 0) - .map(|mb| mb.saturating_mul(1024 * 1024)); + .map(StorageBytes::from_mb); Ok(Self { retention_hours, max_storage_bytes, @@ -81,7 +79,7 @@ impl CleanupScheduler { if let Some(max) = config.max_storage_bytes { tracing::info!( "Storage hard cap enabled: {} MB (will purge oldest trades when exceeded)", - max / (1024 * 1024) + max.as_mb(), ); match storage.current_storage_bytes() { @@ -93,17 +91,17 @@ impl CleanupScheduler { a misconfiguration. Either raise max_storage_mb \ (e.g. to {} MB or higher) or manually shrink the \ database and restart.", - current / (1024 * 1024), - max / (1024 * 1024), - (current / (1024 * 1024)).saturating_add(1), + current.as_mb(), + max.as_mb(), + current.as_mb().saturating_add(1), ); } Ok(current) if current > max => { tracing::warn!( "Database is {} MB — above the {} MB cap; \ startup cleanup will purge oldest trades.", - current / (1024 * 1024), - max / (1024 * 1024), + current.as_mb(), + max.as_mb(), ); } Err(e) => { @@ -135,7 +133,7 @@ impl CleanupScheduler { if n > 0 { tracing::info!( "Cleaned up {n} trade(s) older than {}h", - self.config.retention_hours + self.config.retention_hours.as_hours() ); any_deleted = true; } @@ -176,10 +174,11 @@ impl CleanupScheduler { /// /// Targets `HEADROOM_FRACTION / 5` of the cap so there is headroom for /// incoming trades between 5-minute checks. - fn purge_oldest_trades_until_below(&self, max_bytes: u64) -> Result { - let target_bytes = max_bytes.saturating_mul(HEADROOM_FRACTION) / 5; - let max_est_rows = target_bytes.saturating_div(EST_BYTES_PER_ROW); - let batch_size = purge_batch_size(target_bytes); + fn purge_oldest_trades_until_below(&self, max_bytes: StorageBytes) -> Result { + let target_bytes = + StorageBytes::from_bytes(max_bytes.as_bytes().saturating_mul(HEADROOM_FRACTION) / 5); + let max_est_rows = target_bytes.as_bytes().saturating_div(EST_BYTES_PER_ROW); + let batch_size = purge_batch_size(target_bytes.as_bytes()); let mut total_deleted = 0u64; let mut converged = false; @@ -202,20 +201,20 @@ impl CleanupScheduler { let remaining_rows = self.storage.count_trades()?; if converged { - let current_bytes = self.storage.current_storage_bytes()?; - if current_bytes <= max_bytes { + let current = self.storage.current_storage_bytes()?; + if current <= max_bytes { tracing::info!( "Cleaned up {total_deleted} trade(s) to keep storage under \ {} MB cap (now ~{} MB)", - max_bytes / (1024 * 1024), - current_bytes / (1024 * 1024), + max_bytes.as_mb(), + current.as_mb(), ); } else { tracing::warn!( "Storage ({} MB) still exceeds {} MB cap after cleanup. \ The next periodic pass will retry.", - current_bytes / (1024 * 1024), - max_bytes / (1024 * 1024), + current.as_mb(), + max_bytes.as_mb(), ); } } else { @@ -223,7 +222,7 @@ impl CleanupScheduler { "Size-cap purge did not converge after \ {MAX_PURGE_ITERATIONS} iterations (deleted {total_deleted} \ rows); storage may still exceed the {} MB cap", - max_bytes / (1024 * 1024), + max_bytes.as_mb(), ); } @@ -258,7 +257,7 @@ impl CleanupScheduler { let storage = self.storage.clone(); let config = self.config; - let retention_ms = (config.retention_hours as i64) * 3_600_000; + let retention_ms = config.retention_hours.as_millis(); tokio::spawn(async move { let mut last_cleanup = last_cleanup; diff --git a/src/config.rs b/src/config.rs index 3ffe8c3..123450a 100644 --- a/src/config.rs +++ b/src/config.rs @@ -1,8 +1,13 @@ +use std::collections::HashMap; +use std::net::SocketAddr; +use std::path::{Path, PathBuf}; + use anyhow::{Context, Result}; use clap::Parser; use serde::{Deserialize, Serialize}; -use std::collections::HashMap; -use std::path::{Path, PathBuf}; + +use std::fmt; +use std::str::FromStr; /// Whitelist templates: venue → market_kind → list of quote assets. /// @@ -23,7 +28,7 @@ pub struct Args { #[derive(Debug, Clone, Serialize, Deserialize)] pub struct Config { /// Socket address to bind the HTTP API (e.g. `127.0.0.1:8080`). - pub bind_address: String, + pub bind_address: SocketAddr, /// Directory where the DuckDB database file will be stored. pub data_dir: String, /// Optional bearer-token required on all API requests. @@ -31,8 +36,11 @@ pub struct Config { /// /// The server uses HTTPS with a self-signed certificate (generated /// on first boot), so the token is always encrypted in transit. + /// + /// When generated automatically the token is stored in + /// `data_dir / .auth_token` so that restarts reuse the same token. #[serde(default, skip_serializing)] - pub auth_token: Option, + pub auth_token: Option, // ── Pair tracking ─────────────────────────────────────────── /// Base assets to expand via the whitelist templates (e.g. `["btc", "eth"]`). @@ -127,7 +135,7 @@ impl Config { toml::from_str(&content).with_context(|| format!("parsing {}", path.display()))?; if let Ok(token) = std::env::var("AUTH_TOKEN") - && !token.is_empty() + && let Some(token) = BearerToken::new(token) { cfg.auth_token = Some(token); } @@ -157,37 +165,35 @@ impl Config { } pub fn resolve_auth_token(&mut self) -> anyhow::Result<()> { - let addr: std::net::SocketAddr = self - .bind_address - .parse() - .with_context(|| format!("invalid bind_address '{}'", self.bind_address))?; - - if addr.ip().is_loopback() { + if self.bind_address.ip().is_loopback() { return Ok(()); } let token_file = std::path::PathBuf::from(&self.data_dir).join(".auth_token"); - let on_disk = std::fs::read_to_string(&token_file) + let on_disk_raw = std::fs::read_to_string(&token_file) .ok() - .map(|s| s.trim().to_string()) + .map(|s| s.trim().to_owned()) .filter(|s| !s.is_empty()); + let on_disk_token = on_disk_raw + .as_deref() + .and_then(|s| BearerToken::new(s.to_owned())); - let token = match (self.auth_token.clone(), on_disk.clone()) { + let token = match (self.auth_token.clone(), on_disk_token) { (Some(explicit), _) => explicit, // env/config wins (None, Some(existing)) => existing, // reuse what's on disk (None, None) => generate_token(), // nothing anywhere — mint one }; - if on_disk.as_deref() != Some(token.as_str()) { + if on_disk_raw.as_deref() != Some(token.as_str()) { std::fs::create_dir_all(&self.data_dir) .with_context(|| format!("creating data dir '{}'", self.data_dir))?; - std::fs::write(&token_file, &token) + std::fs::write(&token_file, token.as_str()) .with_context(|| format!("writing {}", token_file.display()))?; crate::tls::restrict_permissions(&token_file); tracing::info!( "Auth token → {}\n Token starts with: {}… (run `cat {}` to view full token)", token_file.display(), - &token[..4.min(token.len())], + &token.as_str()[..4.min(token.as_str().len())], token_file.display(), ); } @@ -235,9 +241,147 @@ impl Config { } } +/// A Bearer token used to authenticate API requests. +/// +/// Constructed via [`FromStr`] (or [`BearerToken::new`]) which rejects +/// empty strings. The [`Display`] implementation outputs `[REDACTED]` +/// to prevent accidental leakage in logs. +/// +/// # Example +/// +/// ```ignore +/// let token: BearerToken = "my-secret-token".parse()?; +/// assert!(token.is_valid_authorization("Bearer my-secret-token")); +/// assert!(!token.is_valid_authorization("Bearer wrong-token")); +/// ``` +#[derive(Debug, Clone, serde::Deserialize)] +#[serde(try_from = "String")] +pub struct BearerToken(String); + +impl BearerToken { + /// Create a new `BearerToken`, returning `None` if `raw` is empty. + pub fn new(raw: String) -> Option { + if raw.is_empty() { + None + } else { + Some(Self(raw)) + } + } + + /// Check whether `authorization_header` matches `"Bearer {token}"` + /// (case-insensitive). + pub fn is_valid_authorization(&self, authorization_header: &str) -> bool { + let expected = format!("Bearer {}", self.0); + authorization_header.eq_ignore_ascii_case(&expected) + } + + /// Return the raw token string (for writing to disk, etc.). + pub fn as_str(&self) -> &str { + &self.0 + } +} + +impl FromStr for BearerToken { + type Err = &'static str; + + fn from_str(s: &str) -> Result { + if s.is_empty() { + Err("Bearer token must not be empty") + } else { + Ok(Self(s.to_owned())) + } + } +} + +impl TryFrom for BearerToken { + type Error = &'static str; + + fn try_from(value: String) -> Result { + value.parse() + } +} + +impl fmt::Display for BearerToken { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + write!(f, "[REDACTED]") + } +} + +impl AsRef for BearerToken { + fn as_ref(&self) -> &str { + &self.0 + } +} + +impl PartialEq for BearerToken { + fn eq(&self, other: &Self) -> bool { + // Constant-time comparison would be better, but for a self-hosted + // internal API the timing leak is negligible. + self.0 == other.0 + } +} + +/// Retention period expressed in hours. Guaranteed to be **> 0** — +/// validated at construction via [`RetentionHours::new`]. +#[derive(Debug, Clone, Copy)] +pub struct RetentionHours(u64); + +impl RetentionHours { + /// Create a `RetentionHours`, returning an error if `hours == 0`. + pub fn new(hours: u64) -> anyhow::Result { + if hours == 0 { + anyhow::bail!("retention_hours must be > 0"); + } + Ok(Self(hours)) + } + + /// The raw hour count. + pub fn as_hours(self) -> u64 { + self.0 + } + + /// The equivalent span in milliseconds (as a signed value for SQL). + pub fn as_millis(self) -> i64 { + (self.0 as i64) * 3_600_000 + } +} + +/// A storage size expressed in bytes. Provides convenience conversions +/// to megabytes to avoid sprinkling `(1024 * 1024)` throughout the code. +#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)] +pub struct StorageBytes(u64); + +impl StorageBytes { + /// Construct from a byte count. + pub const fn from_bytes(bytes: u64) -> Self { + Self(bytes) + } + + /// Construct from a megabyte count (clamped to `u64::MAX` on overflow). + pub fn from_mb(mb: u64) -> Self { + Self(mb.saturating_mul(1024 * 1024)) + } + + /// The raw byte count. + pub fn as_bytes(self) -> u64 { + self.0 + } + + /// The size in whole megabytes (truncated). + pub fn as_mb(self) -> u64 { + self.0 / (1024 * 1024) + } + + /// Saturating multiplication (returns `StorageBytes`). + pub fn saturating_mul(self, rhs: u64) -> Self { + Self(self.0.saturating_mul(rhs)) + } +} + /// Generate a random 256-bit hex token. -fn generate_token() -> String { +fn generate_token() -> BearerToken { let mut buf = [0u8; 32]; getrandom::getrandom(&mut buf).expect("failed to get random bytes"); - buf.iter().map(|b| format!("{b:02x}")).collect() + let hex: String = buf.iter().map(|b| format!("{b:02x}")).collect(); + BearerToken::new(hex).expect("generated hex token is never empty") } diff --git a/src/main.rs b/src/main.rs index 4a3d923..7e67db9 100644 --- a/src/main.rs +++ b/src/main.rs @@ -6,6 +6,7 @@ mod ingestion; mod storage; mod tls; +use std::net::SocketAddr; use std::path::PathBuf; use std::sync::Arc; @@ -19,7 +20,7 @@ use flowsurface_exchange::adapter::{AdapterHandles, Venue}; use flowsurface_exchange::{Ticker, TickerInfo}; use crate::api::Server; -use crate::config::{Args, Config}; +use crate::config::{Args, BearerToken, Config}; use crate::storage::Storage; #[tokio::main] @@ -56,8 +57,8 @@ struct App { adapter_handles: AdapterHandles, resolved_pairs: Vec, metadata_cache: discovery::MetadataCache, - bind_address: String, - auth_token: Option, + bind_address: SocketAddr, + auth_token: Option, flush_interval: std::time::Duration, max_buffered_trades: usize, cleanup_scheduler: cleanup::CleanupScheduler, @@ -140,16 +141,12 @@ impl App { } // Only generate TLS cert for non-loopback addresses. - let addr: std::net::SocketAddr = config - .bind_address - .parse() - .expect("bind_address already validated"); - - let tls_config = if addr.ip().is_loopback() { + let tls_config = if config.bind_address.ip().is_loopback() { None } else { let tls_domain = config.tls_domain.clone(); - let bind_ip = (!addr.ip().is_unspecified()).then_some(addr.ip()); + let bind_ip = + (!config.bind_address.ip().is_unspecified()).then_some(config.bind_address.ip()); let cert_path = data_dir.join("cert.pem"); let key_path = data_dir.join("key.pem"); @@ -206,7 +203,7 @@ impl App { adapter_handles, resolved_pairs, metadata_cache, - bind_address: config.bind_address.clone(), + bind_address: config.bind_address, auth_token: config.auth_token.clone(), flush_interval: std::time::Duration::from_millis(config.flush_interval_ms), max_buffered_trades: config.max_buffered_trades, @@ -256,7 +253,7 @@ impl App { available_tickers, self.tls_config, )); - let server_handle = server.serve(&self.bind_address).await; + let server_handle = server.serve(self.bind_address).await; AppHandles { shutdown, diff --git a/src/storage.rs b/src/storage.rs index 4ac7de1..3bc6c13 100644 --- a/src/storage.rs +++ b/src/storage.rs @@ -9,6 +9,7 @@ use flowsurface_exchange::unit::{price::Price, qty::Qty}; use flowsurface_exchange::{Ticker, TickerInfo, UnixMs}; use crate::api::{AnnotatedTrade, TradeQuery}; +use crate::config::{RetentionHours, StorageBytes}; use tokio::sync::mpsc; use tokio::task::JoinHandle; @@ -93,7 +94,7 @@ impl Storage { /// Return the size (in bytes) of the main database file plus the /// WAL file. If a file does not (yet) exist its size is counted as 0. - pub fn current_storage_bytes(&self) -> Result { + pub fn current_storage_bytes(&self) -> Result { let db_path = self.data_dir.join("trades.duckdb"); let wal_path = self.data_dir.join("trades.duckdb.wal"); @@ -113,7 +114,7 @@ impl Storage { } }; - Ok(db_size + wal_size) + Ok(StorageBytes::from_bytes(db_size + wal_size)) } /// Return the total number of rows in the `trades` table. @@ -473,9 +474,9 @@ impl Storage { /// Delete every trade row whose `ts` (milliseconds since epoch) is /// older than `retention_hours`. Returns the number of deleted rows. - pub fn purge_old_trades(&self, retention_hours: u64) -> Result { + pub fn purge_old_trades(&self, retention_hours: RetentionHours) -> Result { let conn = self.connection()?; - let cutoff_ms = Self::now_ms() - (retention_hours as i64 * 3_600_000); + let cutoff_ms = Self::now_ms() - retention_hours.as_millis(); let deleted = conn .execute( "DELETE FROM trades WHERE ts < ?1", From f1badb11fedd57e933c1dfaba4345ef10a1df8db Mon Sep 17 00:00:00 2001 From: akenshaw <63060680+akenshaw@users.noreply.github.com> Date: Sat, 18 Jul 2026 13:01:26 +0300 Subject: [PATCH 6/8] cap headroom, enable storage MB by default - also make readme conciser --- README.md | 210 ++++++++++++++++++++++++++++++-------------- config.example.toml | 90 +++++++++---------- src/cleanup.rs | 34 +++++-- src/config.rs | 24 ++--- src/storage.rs | 6 ++ 5 files changed, 237 insertions(+), 127 deletions(-) diff --git a/README.md b/README.md index 18354cc..d40cc78 100644 --- a/README.md +++ b/README.md @@ -1,52 +1,59 @@ # flowsurface-server -A crypto trade data collector and server. +A trade data collector for crypto markets, with an embedded database and REST API. -Connects to crypto exchange WebSocket streams via [flowsurface-exchange](https://crates.io/crates/flowsurface-exchange), -persists trades to an embedded [DuckDB](https://duckdb.org) database, and serves -them over a REST API with optional Arrow IPC export. +- Connects to exchange WebSocket streams via [flowsurface-exchange](https://crates.io/crates/flowsurface-exchange) +- Persists trades to [DuckDB](https://duckdb.org) +- Serves data via a REST API, as JSON or [Arrow IPC](https://arrow.apache.org/) stream formats ## Quick start +1. **Copy the template**: + ```bash -# 1. Copy the example config and edit to suit cp config.example.toml config.toml -# edit config.toml to set your exchange whitelist, base assets, etc. +``` + +> See the [basic settings](#basic) and edit `config.toml` + +2. **Run** + +```bash +# looks for `config.toml` next to the binary or in the current directory. +./flowsurface-server +``` + +Or to use a custom config path: -# 2. Run -./flowsurface-server # looks for config.toml next to binary or CWD +```bash ./flowsurface-server --config /path/to/config.toml ``` > If you run without a config file, the server will write the -> template to the given path and **exit with code 2** — this is deliberate -> so that systemd/supervisors can distinguish "not yet configured" from a -> crash. Simply edit the generated file and re-run. +> template to the given path and then exit. Simply edit the generated file and re-run. ## Configuration -Full reference — see [`config.example.toml`](config.example.toml) for all +See [`config.example.toml`](config.example.toml) for all available options with inline documentation. -| Situation | Behaviour | -| --------------------------------- | ------------------------------------------------------------------------ | -| `bind_address = "127.0.0.1:8080"` | Plain HTTP, no auth required | -| `bind_address = "0.0.0.0:8080"` | HTTPS (self-signed cert), auth token + cert fingerprint generated | -| `discovery_mode = true` (default) | Fetch metadata for **all** exchanges at startup to populate `/exchanges` | -| `discovery_mode = false` | Only fetch metadata for whitelisted venues | - -### Key settings - -| Option | Default | Description | -| ---------------------- | ---------------------- | --------------------------------------------------------------------------- | -| `bind_address` | — | Socket address to bind (e.g. `127.0.0.1:8080`) | -| `data_dir` | `"./data"` | Directory for DuckDB, auth token, TLS certs/keys | -| `base_assets` | — | Base assets expanded via whitelist templates (e.g. `["BTC"]`) | -| `flush_interval_ms` | `2000` | Batch flush interval (ms); lower = less data loss, higher = I/O efficient | -| `data_retention_hours` | `48` | Trades older than this are purged on startup & periodically | -| `discovery_mode` | `true` | Fetch metadata for all exchange variants so `/exchanges` is fully populated | -| `max_buffered_trades` | `200000` | Max trades in memory buffer before dropping (OOM guard) | -| `tls_domain` | `"flowsurface-server"` | Domain in the self-signed TLS cert's SAN | +### Basic + +| Option | Default | Description | +| ---------------------- | ------- | ------------------------------------------------------------------------------------------------------- | +| `bind_address` | — | Socket address to bind (`127.0.0.1:8080` = local-only plain HTTP; `0.0.0.0:8080` = remote HTTPS + auth) | +| `base_assets` | — | Base assets expanded via whitelist templates (e.g. `["BTC", "ETH"]`) | +| `max_storage_mb` | `4096` | Hard cap on database file size (MB); `0` = unlimited. | +| `data_retention_hours` | `168` | Trades older than this are purged; `0` = keep all indefinitely. | + +### Advanced + +| Option | Default | Description | +| --------------------- | ---------------------- | --------------------------------------------------------------------------- | +| `discovery_mode` | `true` | Fetch metadata for all exchange variants so `/exchanges` is fully populated | +| `tls_domain` | `"flowsurface-server"` | Domain in the self-signed TLS cert's SAN (only needed for verified TLS) | +| `flush_interval_ms` | `2000` | How often buffered trades are written to disk (ms); higher = fewer writes | +| `max_buffered_trades` | `200000` | Max trades in memory buffer before dropping (OOM guard) | ### Whitelist templates @@ -89,9 +96,8 @@ When binding to a **non-loopback** address, the server: cat data/.auth_token ``` -The token is reused across restarts. To set a specific token manually, -add `auth_token = "your-token"` to `config.toml`, or set the -`AUTH_TOKEN` environment variable (via `.env` or the environment). +The token is reused across restarts. To set a specific token set the `AUTH_TOKEN` environment variable +(via `.env` or the environment). ## Remote deployment (VPS) @@ -141,13 +147,109 @@ For local-only use, keep `bind_address = "127.0.0.1:8080"`: All other endpoints require `Authorization: Bearer ` when auth is configured. -| Method | Path | Auth | Description | -| ------ | --------------- | -------- | ---------------------------------------------------------- | -| GET | `/status` | ✗ public | Server uptime, DB connectivity check | -| GET | `/exchanges` | required | Available ticker symbols per exchange | -| GET | `/pairs` | required | Configured pairs with time bounds & tracked count | -| GET | `/trades` | required | Trade data (filtered by venue, symbol, time range) | -| GET | `/trades.arrow` | required | Trade data as [Arrow IPC](https://arrow.apache.org) stream | +| Method | Path | Auth | Description | +| ------ | --------------- | -------- | -------------------------------------------------- | +| GET | `/status` | ✗ public | Server uptime, DB connectivity check | +| GET | `/exchanges` | required | Available ticker symbols per exchange | +| GET | `/pairs` | required | Configured pairs with time bounds & tracked count | +| GET | `/trades` | required | Trade data (filtered by venue, symbol, time range) | +| GET | `/trades.arrow` | required | Trade data as Arrow IPC stream | + +### GET /status + +Returns the server health status. No authentication required — suitable for +load balancer health checks. + +#### Response fields + +| Field | Type | Description | +| ------------- | ------ | ----------------------------- | +| `status` | string | Always `"ok"` while running | +| `uptime_secs` | int | Seconds since server start | +| `db_ok` | bool | `true` if DuckDB is reachable | + +#### Example + +```bash +curl http://127.0.0.1:8080/status +``` + +```json +{ + "status": "ok", + "uptime_secs": 7, + "db_ok": true +} +``` + +### GET /exchanges + +Returns every ticker symbol discovered on each exchange, grouped by +canonical exchange name. Useful for browsing available tickers before +configuring the whitelist. + +```bash +curl -H "Authorization: Bearer " \ + http://127.0.0.1:8080/exchanges +``` + +```json +{ + "exchanges": { + "Binance Linear": ["BTCUSDT", "ETHUSDT", ...], + "Binance Spot": ["BTCUSDT", "ETHUSDT", ...], + "Bybit Linear": ["BTCUSDT", ...], + ... + } +} + +``` + +### GET /pairs + +Returns all configured pairs with their stored time ranges. Pairs that have +been configured but have not yet received any trades appear with `earliest` +and `latest` as `null`. + +#### Response fields + +| Field | Type | Description | +| --------------- | ----- | -------------------------------- | +| `pairs` | array | Array of tracked pair objects | +| `tracked_count` | int | Total number of configured pairs | + +Each pair object: + +| Field | Type | Description | +| ---------- | ------ | ----------------------------------------- | +| `ticker` | string | Ticker ID (`Exchange:pair`) | +| `earliest` | int | Unix ms of oldest stored trade, or `null` | +| `latest` | int | Unix ms of newest stored trade, or `null` | + +#### Example + +```bash +curl -H "Authorization: Bearer " \ + http://127.0.0.1:8080/pairs +``` + +```json +{ + "pairs": [ + { + "ticker": "BinanceSpot:btcusdt", + "earliest": 1783873393043, + "latest": 1784372262447 + }, + { + "ticker": "HyperliquidLinear:btcusdc", + "earliest": 1784372175005, + "latest": 1784372262010 + } + ], + "tracked_count": 18 +} +``` ### GET /trades @@ -207,8 +309,7 @@ format** payload (`Content-Type: application/vnd.apache.arrow.stream`) with 4 columns: `ts (int64)`, `price (float64)`, `qty (float64)`, `is_sell (bool)`. -This is ideal for high-volume data transfer into data-science tools -(Polars, Pandas, Julia, etc.) that support Arrow natively. +This is ideal for high-volume data transfer to clients that support Arrow natively. | Param | Type | Description | | -------- | ------ | -------------------------------------------- | @@ -218,26 +319,3 @@ This is ideal for high-volume data transfer into data-science tools | `from` | int | Unix ms lower bound (inclusive) | | `to` | int | Unix ms upper bound (inclusive) | | `limit` | int | Max records (default 100 000, max 1 000 000) | - -### GET /exchanges - -Returns every ticker symbol discovered on each exchange, grouped by -canonical exchange name. Useful for browsing available tickers before -configuring the whitelist. - -```bash -curl -H "Authorization: Bearer " \ - http://127.0.0.1:8080/exchanges -``` - -```json -{ - "exchanges": { - "Binance Linear": ["BTCUSDT", "ETHUSDT", ...], - "Binance Spot": ["BTCUSDT", "ETHUSDT", ...], - "Bybit Linear": ["BTCUSDT", ...], - ... - } -} - -``` diff --git a/config.example.toml b/config.example.toml index e7db73c..9c9b519 100644 --- a/config.example.toml +++ b/config.example.toml @@ -1,53 +1,31 @@ -bind_address = "127.0.0.1:8080" -data_dir = "./data" - -# Domain name for the self-signed TLS certificate's SAN (Subject Alternative Names). -# Ignored on loopback binds (plain HTTP). Default: "flowsurface-server". -# tls_domain = "data.example.com" - -# Trade batch-flush interval in milliseconds. -# Lower values reduce data loss on crash/failure; higher values are more I/O-efficient -# (fewer fsyncs, larger columnar batches). Default: 2000 (2 seconds). -# flush_interval_ms = 2000 +# Directory for DuckDB database, auth token, and TLS certs. +# data_dir = "./data" -# Maximum trades to buffer in memory before dropping to prevent OOM. -# Tune based on your host RAM and risk tolerance: -# 200k (~20-40 MB) — safe for 1 GB hosts -# 1.2M (~120-240 MB) — ~2 min buffer at 10k trades/sec -# Higher values reduce data-loss during DB outages but use more memory. -# Default: 200000. -# max_buffered_trades = 200000 - -# Data retention period in hours. Trades older than this will be purged -# on startup (and periodically while running). Default: 48 (2 days). -# data_retention_hours = 48 +# ── Network ───────────────────────────────────────────────────── +# Where the API listens. Change to "0.0.0.0:8080" for remote +# access (auto-generates TLS cert + auth token). +bind_address = "127.0.0.1:8080" +# ── Storage ────────────────────────────────────────────────────── # Hard cap on total DuckDB database file size in megabytes. -# When the database file size exceeds this value the oldest trades are -# purged during cleanup — even if they're within the time-based retention -# window. Use this to prevent the database from quietly filling your disk -# on small VPS hosts. -# -# Tip: set to ~50-80 % of your available disk space to leave room for -# system files, logs, and bursts. -# Default: no cap. -# max_storage_mb = 4096 +# When exceeded the oldest trades are purged during cleanup — even +# if they're within the time-based retention window. +# Set to 0 to disable the size cap. +# Default: 4096 (4 GiB). +max_storage_mb = 4096 -# Discovery mode: fetches and caches metadata for ALL supported exchanges on startup -# so `/exchanges` is fully populated. Lets you browse available tickers -# before deciding what to track. Default: true. -# discovery_mode = true +# Data retention period in hours. Trades older than this are +# purged on startup and periodically while running. +# Set to 0 to keep all trades indefinitely (disk-permitting). +# Default: 168 (7 days). +data_retention_hours = 168 -# Base assets to track — expanded via the whitelist templates below. -# The server fetches exchange metadata and resolves pairs at startup. +# ── Pairs to track ────────────────────────────────────────────── +# Expand base_assets × quote_assets from the whitelist templates +# below. The server resolves exchange-specific ticker strings +# (handling separators, _PERP, -SWAP suffixes, etc.). base_assets = ["BTC", "ETH"] -# Whitelist templates: per-venue, per-market-kind list of quote assets. -# To exclude a venue entirely, omit it from the whitelist. -# To exclude a market kind for a venue, omit that key. -# -# Use `[""]` (empty string) as a wildcard to track all `base_assets` regardless of the quote on that venue+market - [whitelist.binance] spot = ["USDT"] linear = ["USDT", "USDC"] @@ -62,4 +40,28 @@ linear = ["USDC"] [whitelist.okex] spot = ["USDT"] -linear = ["USDT"] \ No newline at end of file +linear = ["USDT"] + +# ── Advanced options ───────────────────────────────────────────── +# Most users can leave these at their defaults. + +# Discovery mode: fetches metadata for ALL supported exchanges on +# startup so /exchanges is fully populated. +# If false, only fetch metadata for whitelisted venues +# Default: true. +# discovery_mode = true + +# Domain name for the self-signed TLS certificate's SAN. +# Only needed when connecting via a domain with verified TLS. +# tls_domain = "data.example.com" + +# Trade batch-flush interval in milliseconds. +# Higher values = fewer disk writes (more I/O efficient). +# Lower values = trades appear in the DB sooner (at the cost of more writes). +# Default: 2000 (2 seconds). +# flush_interval_ms = 2000 + +# Maximum trades to buffer in memory before dropping (OOM guard). +# 200k ≈ 20-40 MB — safe for 1 GB hosts. +# Default: 200000. +# max_buffered_trades = 200000 \ No newline at end of file diff --git a/src/cleanup.rs b/src/cleanup.rs index c1a24bc..1d7fe00 100644 --- a/src/cleanup.rs +++ b/src/cleanup.rs @@ -25,9 +25,25 @@ const MAX_PURGE_ITERATIONS: usize = 200; /// size after `CHECKPOINT`. const EST_BYTES_PER_ROW: u64 = 50; -/// Target fraction of the cap to purge down to, leaving headroom for -/// incoming trades between periodic checks. 4/5 = 80 %. -const HEADROOM_FRACTION: u64 = 4; +/// On small-to-medium caps use a 20 % headroom so the purge leaves +/// breathing room. On very large caps the percentage would waste +/// too much space, so we cap the absolute headroom at 10 GiB +const MAX_HEADROOM: StorageBytes = StorageBytes::from_bytes(10 * 1024 * 1024 * 1024); + +/// Smallest sane target after headroom subtraction — prevents +/// pathological behaviour on tiny caps. +const MIN_TARGET: StorageBytes = StorageBytes::from_bytes(100 * 1024 * 1024); + +/// Compute the headroom to leave free below the cap. +/// Returns `cap / 5` (20 %) but at most `MAX_HEADROOM`. +fn purge_headroom(cap: StorageBytes) -> StorageBytes { + let pct = cap.as_bytes() / 5; + if pct > MAX_HEADROOM.as_bytes() { + MAX_HEADROOM + } else { + StorageBytes::from_bytes(pct) + } +} /// Batch size for each purge iteration — ~10% of the cap, clamped to /// [1 000, 100 000]. @@ -172,11 +188,15 @@ impl CleanupScheduler { /// 2. **CHECKPOINT once**, then verify the real file size. If still over /// the absolute cap we log a warning — the next periodic pass will retry. /// - /// Targets `HEADROOM_FRACTION / 5` of the cap so there is headroom for - /// incoming trades between 5-minute checks. + /// Targets `cap - headroom` so there is breathing room for incoming + /// trades between 5-minute checks without wasting space on large caps. fn purge_oldest_trades_until_below(&self, max_bytes: StorageBytes) -> Result { - let target_bytes = - StorageBytes::from_bytes(max_bytes.as_bytes().saturating_mul(HEADROOM_FRACTION) / 5); + let headroom = purge_headroom(max_bytes); + let target_bytes = max_bytes + .as_bytes() + .saturating_sub(headroom.as_bytes()) + .max(MIN_TARGET.as_bytes()); + let target_bytes = StorageBytes::from_bytes(target_bytes); let max_est_rows = target_bytes.as_bytes().saturating_div(EST_BYTES_PER_ROW); let batch_size = purge_batch_size(target_bytes.as_bytes()); let mut total_deleted = 0u64; diff --git a/src/config.rs b/src/config.rs index 123450a..451d1a7 100644 --- a/src/config.rs +++ b/src/config.rs @@ -39,7 +39,8 @@ pub struct Config { /// /// When generated automatically the token is stored in /// `data_dir / .auth_token` so that restarts reuse the same token. - #[serde(default, skip_serializing)] + /// Not settable via `config.toml` — use the `AUTH_TOKEN` env var instead. + #[serde(skip)] pub auth_token: Option, // ── Pair tracking ─────────────────────────────────────────── @@ -97,7 +98,8 @@ pub struct Config { /// /// Tip: set this to ~50-80 % of your available disk space so the /// server leaves room for system files, logs, and burst. - #[serde(default)] + /// Default: `4096` (4 GiB). + #[serde(default = "default_max_storage_mb")] pub max_storage_mb: Option, } @@ -106,7 +108,7 @@ const fn default_flush_interval() -> u64 { } const fn default_data_retention_hours() -> u64 { - 48 + 168 } const fn default_true() -> bool { @@ -121,6 +123,10 @@ const fn default_max_buffered_trades() -> usize { 200_000 } +const fn default_max_storage_mb() -> Option { + Some(4096) +} + impl Config { /// Return the default config template as a commented TOML string. pub fn template() -> &'static str { @@ -321,17 +327,15 @@ impl PartialEq for BearerToken { } } -/// Retention period expressed in hours. Guaranteed to be **> 0** — -/// validated at construction via [`RetentionHours::new`]. +/// Retention period expressed in hours. A value of `0` means +/// **unlimited** — no time-based purges. #[derive(Debug, Clone, Copy)] pub struct RetentionHours(u64); impl RetentionHours { - /// Create a `RetentionHours`, returning an error if `hours == 0`. + /// Create a `RetentionHours`. `0` is accepted and means unlimited + /// (time-based purges are skipped). pub fn new(hours: u64) -> anyhow::Result { - if hours == 0 { - anyhow::bail!("retention_hours must be > 0"); - } Ok(Self(hours)) } @@ -363,7 +367,7 @@ impl StorageBytes { } /// The raw byte count. - pub fn as_bytes(self) -> u64 { + pub const fn as_bytes(self) -> u64 { self.0 } diff --git a/src/storage.rs b/src/storage.rs index 3bc6c13..924e0fc 100644 --- a/src/storage.rs +++ b/src/storage.rs @@ -474,7 +474,13 @@ impl Storage { /// Delete every trade row whose `ts` (milliseconds since epoch) is /// older than `retention_hours`. Returns the number of deleted rows. + /// + /// When `retention_hours` is `0` (unlimited) the purge is skipped + /// and `Ok(0)` is returned immediately. pub fn purge_old_trades(&self, retention_hours: RetentionHours) -> Result { + if retention_hours.as_hours() == 0 { + return Ok(0); + } let conn = self.connection()?; let cutoff_ms = Self::now_ms() - retention_hours.as_millis(); let deleted = conn From 222d63be3c74905a84942a9c89f3532b18c2b5b4 Mon Sep 17 00:00:00 2001 From: akenshaw <63060680+akenshaw@users.noreply.github.com> Date: Sat, 18 Jul 2026 14:51:14 +0300 Subject: [PATCH 7/8] resolve `/data` and config paths relative to binary --- README.md | 2 +- config.example.toml | 3 ++- src/config.rs | 12 ++++++------ src/main.rs | 18 +++++++++++++----- 4 files changed, 22 insertions(+), 13 deletions(-) diff --git a/README.md b/README.md index d40cc78..3674d02 100644 --- a/README.md +++ b/README.md @@ -19,7 +19,7 @@ cp config.example.toml config.toml 2. **Run** ```bash -# looks for `config.toml` next to the binary or in the current directory. +# looks for `config.toml` next to the binary. ./flowsurface-server ``` diff --git a/config.example.toml b/config.example.toml index 9c9b519..1cafa02 100644 --- a/config.example.toml +++ b/config.example.toml @@ -1,5 +1,6 @@ # Directory for DuckDB database, auth token, and TLS certs. -# data_dir = "./data" +# Relative paths are resolved from the config file's directory. +# data_dir = "data" # ── Network ───────────────────────────────────────────────────── # Where the API listens. Change to "0.0.0.0:8080" for remote diff --git a/src/config.rs b/src/config.rs index 451d1a7..29c40b5 100644 --- a/src/config.rs +++ b/src/config.rs @@ -155,16 +155,16 @@ impl Config { } /// Resolve the configuration file path. + /// + /// Defaults to next to the binary so the entire app is portable in + /// a single directory. pub fn resolve_path(override_path: Option) -> PathBuf { if let Some(path) = override_path { return path; } - if let Ok(exe) = std::env::current_exe() - && let Some(parent) = exe.parent() - { - let candidate = parent.join("config.toml"); - if candidate.exists() { - return candidate; + if let Ok(exe) = std::env::current_exe() { + if let Some(parent) = exe.parent() { + return parent.join("config.toml"); } } PathBuf::from("config.toml") diff --git a/src/main.rs b/src/main.rs index 7e67db9..8e9b6d3 100644 --- a/src/main.rs +++ b/src/main.rs @@ -7,7 +7,7 @@ mod storage; mod tls; use std::net::SocketAddr; -use std::path::PathBuf; +use std::path::{Path, PathBuf}; use std::sync::Arc; use clap::Parser; @@ -47,7 +47,16 @@ async fn main() { std::process::exit(1); } - let app = App::new(&config).await; + let data_dir = if Path::new(&config.data_dir).is_relative() { + config_path + .parent() + .expect("config path has no parent") + .join(&config.data_dir) + } else { + PathBuf::from(&config.data_dir) + }; + + let app = App::new(&config, &data_dir).await; let handles = app.serve().await; handles.shutdown().await; } @@ -67,9 +76,8 @@ struct App { impl App { /// Open storage, resolve configured pairs, persist ticker metadata. - async fn new(config: &Config) -> Self { - let data_dir = PathBuf::from(&config.data_dir); - let storage = Storage::open(&data_dir).unwrap_or_else(|e| { + async fn new(config: &Config, data_dir: &Path) -> Self { + let storage = Storage::open(data_dir).unwrap_or_else(|e| { tracing::error!("Failed to initialise storage: {e:#}"); std::process::exit(1); }); From ad1cdd115ad790a2d7cb6d8134769e2bfac43001 Mon Sep 17 00:00:00 2001 From: akenshaw <63060680+akenshaw@users.noreply.github.com> Date: Sat, 18 Jul 2026 15:09:38 +0300 Subject: [PATCH 8/8] use `cargo-dist` for releases --- .github/workflows/release.yml | 296 ++++++++++++++++++++++++++++++++++ .gitignore | 1 + Cargo.toml | 8 +- dist-workspace.toml | 15 ++ 4 files changed, 319 insertions(+), 1 deletion(-) create mode 100644 .github/workflows/release.yml create mode 100644 dist-workspace.toml diff --git a/.github/workflows/release.yml b/.github/workflows/release.yml new file mode 100644 index 0000000..1dfcd0f --- /dev/null +++ b/.github/workflows/release.yml @@ -0,0 +1,296 @@ +# This file was autogenerated by dist: https://axodotdev.github.io/cargo-dist +# +# Copyright 2022-2024, axodotdev +# SPDX-License-Identifier: MIT or Apache-2.0 +# +# CI that: +# +# * checks for a Git Tag that looks like a release +# * builds artifacts with dist (archives, installers, hashes) +# * uploads those artifacts to temporary workflow zip +# * on success, uploads the artifacts to a GitHub Release +# +# Note that the GitHub Release will be created with a generated +# title/body based on your changelogs. + +name: Release +permissions: + "contents": "write" + +# This task will run whenever you push a git tag that looks like a version +# like "1.0.0", "v0.1.0-prerelease.1", "my-app/0.1.0", "releases/v1.0.0", etc. +# Various formats will be parsed into a VERSION and an optional PACKAGE_NAME, where +# PACKAGE_NAME must be the name of a Cargo package in your workspace, and VERSION +# must be a Cargo-style SemVer Version (must have at least major.minor.patch). +# +# If PACKAGE_NAME is specified, then the announcement will be for that +# package (erroring out if it doesn't have the given version or isn't dist-able). +# +# If PACKAGE_NAME isn't specified, then the announcement will be for all +# (dist-able) packages in the workspace with that version (this mode is +# intended for workspaces with only one dist-able package, or with all dist-able +# packages versioned/released in lockstep). +# +# If you push multiple tags at once, separate instances of this workflow will +# spin up, creating an independent announcement for each one. However, GitHub +# will hard limit this to 3 tags per commit, as it will assume more tags is a +# mistake. +# +# If there's a prerelease-style suffix to the version, then the release(s) +# will be marked as a prerelease. +on: + pull_request: + push: + tags: + - '**[0-9]+.[0-9]+.[0-9]+*' + +jobs: + # Run 'dist plan' (or host) to determine what tasks we need to do + plan: + runs-on: "ubuntu-22.04" + outputs: + val: ${{ steps.plan.outputs.manifest }} + tag: ${{ !github.event.pull_request && github.ref_name || '' }} + tag-flag: ${{ !github.event.pull_request && format('--tag={0}', github.ref_name) || '' }} + publishing: ${{ !github.event.pull_request }} + env: + GH_TOKEN: ${{ secrets.GITHUB_TOKEN }} + steps: + - uses: actions/checkout@v6 + with: + persist-credentials: false + submodules: recursive + - name: Install dist + # we specify bash to get pipefail; it guards against the `curl` command + # failing. otherwise `sh` won't catch that `curl` returned non-0 + shell: bash + run: "curl --proto '=https' --tlsv1.2 -LsSf https://github.com/axodotdev/cargo-dist/releases/download/v0.32.0/cargo-dist-installer.sh | sh" + - name: Cache dist + uses: actions/upload-artifact@v7 + with: + name: cargo-dist-cache + path: ~/.cargo/bin/dist + # sure would be cool if github gave us proper conditionals... + # so here's a doubly-nested ternary-via-truthiness to try to provide the best possible + # functionality based on whether this is a pull_request, and whether it's from a fork. + # (PRs run on the *source* but secrets are usually on the *target* -- that's *good* + # but also really annoying to build CI around when it needs secrets to work right.) + - id: plan + run: | + dist ${{ (!github.event.pull_request && format('host --steps=create --tag={0}', github.ref_name)) || 'plan' }} --output-format=json > plan-dist-manifest.json + echo "dist ran successfully" + cat plan-dist-manifest.json + echo "manifest=$(jq -c "." plan-dist-manifest.json)" >> "$GITHUB_OUTPUT" + - name: "Upload dist-manifest.json" + uses: actions/upload-artifact@v7 + with: + name: artifacts-plan-dist-manifest + path: plan-dist-manifest.json + + # Build and packages all the platform-specific things + build-local-artifacts: + name: build-local-artifacts (${{ join(matrix.targets, ', ') }}) + # Let the initial task tell us to not run (currently very blunt) + needs: + - plan + if: ${{ fromJson(needs.plan.outputs.val).ci.github.artifacts_matrix.include != null && (needs.plan.outputs.publishing == 'true' || fromJson(needs.plan.outputs.val).ci.github.pr_run_mode == 'upload') }} + strategy: + fail-fast: false + # Target platforms/runners are computed by dist in create-release. + # Each member of the matrix has the following arguments: + # + # - runner: the github runner + # - dist-args: cli flags to pass to dist + # - install-dist: expression to run to install dist on the runner + # + # Typically there will be: + # - 1 "global" task that builds universal installers + # - N "local" tasks that build each platform's binaries and platform-specific installers + matrix: ${{ fromJson(needs.plan.outputs.val).ci.github.artifacts_matrix }} + runs-on: ${{ matrix.runner }} + container: ${{ matrix.container && matrix.container.image || null }} + env: + GH_TOKEN: ${{ secrets.GITHUB_TOKEN }} + BUILD_MANIFEST_NAME: target/distrib/${{ join(matrix.targets, '-') }}-dist-manifest.json + steps: + - name: enable windows longpaths + run: | + git config --global core.longpaths true + - uses: actions/checkout@v6 + with: + persist-credentials: false + submodules: recursive + - name: Install Rust non-interactively if not already installed + if: ${{ matrix.container }} + run: | + if ! command -v cargo > /dev/null 2>&1; then + curl --proto '=https' --tlsv1.2 -sSf https://sh.rustup.rs | sh -s -- -y + echo "$HOME/.cargo/bin" >> $GITHUB_PATH + fi + - name: Install dist + run: ${{ matrix.install_dist.run }} + # Get the dist-manifest + - name: Fetch local artifacts + uses: actions/download-artifact@v8 + with: + pattern: artifacts-* + path: target/distrib/ + merge-multiple: true + - name: Install dependencies + run: | + ${{ matrix.packages_install }} + - name: Build artifacts + run: | + # Actually do builds and make zips and whatnot + dist build ${{ needs.plan.outputs.tag-flag }} --print=linkage --output-format=json ${{ matrix.dist_args }} > dist-manifest.json + echo "dist ran successfully" + - id: cargo-dist + name: Post-build + # We force bash here just because github makes it really hard to get values up + # to "real" actions without writing to env-vars, and writing to env-vars has + # inconsistent syntax between shell and powershell. + shell: bash + run: | + # Parse out what we just built and upload it to scratch storage + echo "paths<> "$GITHUB_OUTPUT" + dist print-upload-files-from-manifest --manifest dist-manifest.json >> "$GITHUB_OUTPUT" + echo "EOF" >> "$GITHUB_OUTPUT" + + cp dist-manifest.json "$BUILD_MANIFEST_NAME" + - name: "Upload artifacts" + uses: actions/upload-artifact@v7 + with: + name: artifacts-build-local-${{ join(matrix.targets, '_') }} + path: | + ${{ steps.cargo-dist.outputs.paths }} + ${{ env.BUILD_MANIFEST_NAME }} + + # Build and package all the platform-agnostic(ish) things + build-global-artifacts: + needs: + - plan + - build-local-artifacts + runs-on: "ubuntu-22.04" + env: + GH_TOKEN: ${{ secrets.GITHUB_TOKEN }} + BUILD_MANIFEST_NAME: target/distrib/global-dist-manifest.json + steps: + - uses: actions/checkout@v6 + with: + persist-credentials: false + submodules: recursive + - name: Install cached dist + uses: actions/download-artifact@v8 + with: + name: cargo-dist-cache + path: ~/.cargo/bin/ + - run: chmod +x ~/.cargo/bin/dist + # Get all the local artifacts for the global tasks to use (for e.g. checksums) + - name: Fetch local artifacts + uses: actions/download-artifact@v8 + with: + pattern: artifacts-* + path: target/distrib/ + merge-multiple: true + - id: cargo-dist + shell: bash + run: | + dist build ${{ needs.plan.outputs.tag-flag }} --output-format=json "--artifacts=global" > dist-manifest.json + echo "dist ran successfully" + + # Parse out what we just built and upload it to scratch storage + echo "paths<> "$GITHUB_OUTPUT" + jq --raw-output ".upload_files[]" dist-manifest.json >> "$GITHUB_OUTPUT" + echo "EOF" >> "$GITHUB_OUTPUT" + + cp dist-manifest.json "$BUILD_MANIFEST_NAME" + - name: "Upload artifacts" + uses: actions/upload-artifact@v7 + with: + name: artifacts-build-global + path: | + ${{ steps.cargo-dist.outputs.paths }} + ${{ env.BUILD_MANIFEST_NAME }} + # Determines if we should publish/announce + host: + needs: + - plan + - build-local-artifacts + - build-global-artifacts + # Only run if we're "publishing", and only if plan, local and global didn't fail (skipped is fine) + if: ${{ always() && needs.plan.result == 'success' && needs.plan.outputs.publishing == 'true' && (needs.build-global-artifacts.result == 'skipped' || needs.build-global-artifacts.result == 'success') && (needs.build-local-artifacts.result == 'skipped' || needs.build-local-artifacts.result == 'success') }} + env: + GH_TOKEN: ${{ secrets.GITHUB_TOKEN }} + runs-on: "ubuntu-22.04" + outputs: + val: ${{ steps.host.outputs.manifest }} + steps: + - uses: actions/checkout@v6 + with: + persist-credentials: false + submodules: recursive + - name: Install cached dist + uses: actions/download-artifact@v8 + with: + name: cargo-dist-cache + path: ~/.cargo/bin/ + - run: chmod +x ~/.cargo/bin/dist + # Fetch artifacts from scratch-storage + - name: Fetch artifacts + uses: actions/download-artifact@v8 + with: + pattern: artifacts-* + path: target/distrib/ + merge-multiple: true + - id: host + shell: bash + run: | + dist host ${{ needs.plan.outputs.tag-flag }} --steps=upload --steps=release --output-format=json > dist-manifest.json + echo "artifacts uploaded and released successfully" + cat dist-manifest.json + echo "manifest=$(jq -c "." dist-manifest.json)" >> "$GITHUB_OUTPUT" + - name: "Upload dist-manifest.json" + uses: actions/upload-artifact@v7 + with: + # Overwrite the previous copy + name: artifacts-dist-manifest + path: dist-manifest.json + # Create a GitHub Release while uploading all files to it + - name: "Download GitHub Artifacts" + uses: actions/download-artifact@v8 + with: + pattern: artifacts-* + path: artifacts + merge-multiple: true + - name: Cleanup + run: | + # Remove the granular manifests + rm -f artifacts/*-dist-manifest.json + - name: Create GitHub Release + env: + PRERELEASE_FLAG: "${{ fromJson(steps.host.outputs.manifest).announcement_is_prerelease && '--prerelease' || '' }}" + ANNOUNCEMENT_TITLE: "${{ fromJson(steps.host.outputs.manifest).announcement_title }}" + ANNOUNCEMENT_BODY: "${{ fromJson(steps.host.outputs.manifest).announcement_github_body }}" + RELEASE_COMMIT: "${{ github.sha }}" + run: | + # Write and read notes from a file to avoid quoting breaking things + echo "$ANNOUNCEMENT_BODY" > $RUNNER_TEMP/notes.txt + + gh release create "${{ needs.plan.outputs.tag }}" --target "$RELEASE_COMMIT" $PRERELEASE_FLAG --title "$ANNOUNCEMENT_TITLE" --notes-file "$RUNNER_TEMP/notes.txt" artifacts/* + + announce: + needs: + - plan + - host + # use "always() && ..." to allow us to wait for all publish jobs while + # still allowing individual publish jobs to skip themselves (for prereleases). + # "host" however must run to completion, no skipping allowed! + if: ${{ always() && needs.host.result == 'success' }} + runs-on: "ubuntu-22.04" + env: + GH_TOKEN: ${{ secrets.GITHUB_TOKEN }} + steps: + - uses: actions/checkout@v6 + with: + persist-credentials: false + submodules: recursive diff --git a/.gitignore b/.gitignore index 6312e72..a0e9bbd 100644 --- a/.gitignore +++ b/.gitignore @@ -4,5 +4,6 @@ *.toml !Cargo.toml !config.example.toml +!dist-workspace.toml .DS_Store *.log \ No newline at end of file diff --git a/Cargo.toml b/Cargo.toml index dab9a6a..f6ae36d 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -3,6 +3,7 @@ name = "flowsurface-server" version = "0.1.0" edition = "2024" license = "MIT" +repository = "https://github.com/akenshaw/fs-server" [dependencies] # HTTP / async runtime / concurrency @@ -45,4 +46,9 @@ tracing = { version = "0.1", default-features = false, features = ["std"] } tracing-subscriber = { version = "0.3", default-features = false, features = ["fmt", "env-filter", "ansi"] } # Error handling -anyhow = "1" \ No newline at end of file +anyhow = "1" + +# The profile that 'dist' will build with +[profile.dist] +inherits = "release" +lto = "thin" diff --git a/dist-workspace.toml b/dist-workspace.toml new file mode 100644 index 0000000..95cdfe9 --- /dev/null +++ b/dist-workspace.toml @@ -0,0 +1,15 @@ +[workspace] +members = ["cargo:."] + +# Config for 'dist' +[dist] +# The preferred dist version to use in CI (Cargo.toml SemVer syntax) +cargo-dist-version = "0.32.0" +# CI backends to support +ci = "github" +# The installers to generate for each app +installers = [] +# Target platforms to build apps for (Rust target-triple syntax) +targets = ["aarch64-apple-darwin", "aarch64-unknown-linux-gnu", "x86_64-apple-darwin", "x86_64-unknown-linux-gnu", "x86_64-pc-windows-msvc"] +# Extra static files to include in the archive (README, LICENSE are auto-included) +include = ["config.example.toml"]