From cc38151be4e74552b2224fdabc02d21c9e85d92f Mon Sep 17 00:00:00 2001 From: tongfengyuan <71140753@chinatelecom.cn> Date: Wed, 2 Sep 2026 08:50:59 +0800 Subject: [PATCH 1/3] feat(sipflow): count raw-file flushes and record flush batch size The raw capture file already recorded write latency (sipflow_raw_write_seconds); add the flush counter and a per-flush byte-size histogram so the data.raw write path can be correlated with sqlite flush metrics (sipflow_flush_rows_total / flush_batch_size / flush_db_seconds) against cgroup IO counters. --- crates/rustpbx-sipflow/src/storage.rs | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/crates/rustpbx-sipflow/src/storage.rs b/crates/rustpbx-sipflow/src/storage.rs index 4a111e53a..a0dddc3b1 100644 --- a/crates/rustpbx-sipflow/src/storage.rs +++ b/crates/rustpbx-sipflow/src/storage.rs @@ -341,6 +341,7 @@ impl StorageManager { .current_offset .saturating_sub(self.write_buf.len() as u64); let write_start = Instant::now(); + let batch_bytes = self.write_buf.len(); if !self.pos_known { file.seek(SeekFrom::Start(offset)).await?; } @@ -350,6 +351,9 @@ impl StorageManager { } self.pos_known = true; self.write_buf.clear(); + metrics::counter!("sipflow_raw_flush_total", "component" => "sipflow").increment(1); + metrics::histogram!("sipflow_raw_flush_bytes", "component" => "sipflow") + .record(batch_bytes as f64); metrics::histogram!("sipflow_raw_write_seconds", "component" => "sipflow") .record(write_start.elapsed().as_secs_f64()); Ok(()) From 84b677483a18a6c8020c88c43d97c3ec0e38800c Mon Sep 17 00:00:00 2001 From: tongfengyuan <71140753@chinatelecom.cn> Date: Wed, 2 Sep 2026 09:42:13 +0800 Subject: [PATCH 2/3] feat(sipflow): add general sqlite_* pipeline metrics alongside sipflow_* ones MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Instrument the sqlite usage with a general taxonomy — connections (opened/open/errors/latency, role=read|write), statements (statements_total, rows read/written, latency by kind select/insert/pragma/ddl), transactions (commit/rollback + latency), WAL checkpoints (passive/truncate + busy) and file gauges (db/wal bytes, page_count, freelist). All carry a database label (value: sipflow) so the same taxonomy generalizes to other sqlite usage later. Existing domain metrics (sipflow_flush_*, sipflow_wal_checkpoint_*, sipflow_db_file_bytes) are unchanged. --- crates/rustpbx-sipflow/src/flusher.rs | 158 +++++++++++---- crates/rustpbx-sipflow/src/lib.rs | 1 + crates/rustpbx-sipflow/src/sqlite_metrics.rs | 200 +++++++++++++++++++ crates/rustpbx-sipflow/src/storage.rs | 72 +++++-- 4 files changed, 372 insertions(+), 59 deletions(-) create mode 100644 crates/rustpbx-sipflow/src/sqlite_metrics.rs diff --git a/crates/rustpbx-sipflow/src/flusher.rs b/crates/rustpbx-sipflow/src/flusher.rs index 0c55ac4e3..ce8cbec8f 100644 --- a/crates/rustpbx-sipflow/src/flusher.rs +++ b/crates/rustpbx-sipflow/src/flusher.rs @@ -1,4 +1,5 @@ use crate::protocol::MsgType; +use crate::sqlite_metrics; use anyhow::Result; use lru::LruCache; use sqlx::sqlite::SqliteConnectOptions; @@ -333,13 +334,16 @@ async fn handle_flush_command( // WAL and shrink its page cache so the previous bucket's WAL // doesn't linger on disk and its pages are released from RSS. if let Some(conn) = db_conn.as_mut() { + let ckpt_start = Instant::now(); let _ = sqlx::query("PRAGMA wal_checkpoint(TRUNCATE)") .execute(&mut *conn) .await; + sqlite_metrics::record_checkpoint("truncate", ckpt_start.elapsed(), 0); let _ = sqlx::query("PRAGMA shrink_memory") .execute(&mut *conn) .await; } + sqlite_metrics::ConnectionGuard::release("write"); drop(db_conn.take()); *db_conn = Some(open_db_with_pragmas(&new_db_path).await); *db_path = Some(new_db_path); @@ -359,12 +363,15 @@ async fn handle_flush_command( } async fn open_db_with_pragmas(db_path: &PathBuf) -> SqliteConnection { + let open_start = Instant::now(); let mut conn = SqliteConnectOptions::new() .filename(db_path) .create_if_missing(true) .connect() .await + .inspect_err(|e| sqlite_metrics::record_error("connect", e)) .expect("failed to open sipflow sqlite db"); + sqlite_metrics::record_write_open(open_start.elapsed()); for pragma in [ "PRAGMA journal_mode=WAL", @@ -374,27 +381,44 @@ async fn open_db_with_pragmas(db_path: &PathBuf) -> SqliteConnection { "PRAGMA busy_timeout=5000", "PRAGMA page_size=4096", ] { - if let Err(e) = sqlx::query(pragma).execute(&mut conn).await { - tracing::warn!("sipflow flusher: PRAGMA failed: {e}"); + let start = Instant::now(); + match sqlx::query(pragma).execute(&mut conn).await { + Ok(r) => sqlite_metrics::record_statement("pragma", 0, r.rows_affected(), start.elapsed()), + Err(e) => { + sqlite_metrics::record_error("pragma", &e); + tracing::warn!("sipflow flusher: PRAGMA failed: {e}"); + } + } + } + + // Schema bootstrap (bucket creation). Each DDL runs once per bucket; + // failures stay non-fatal as before but are now visible in metrics. + async fn exec_ddl(conn: &mut SqliteConnection, sql: &'static str) { + let start = Instant::now(); + match sqlx::query(sql).execute(conn).await { + Ok(r) => sqlite_metrics::record_statement("ddl", 0, r.rows_affected(), start.elapsed()), + Err(e) => { + sqlite_metrics::record_error("ddl", &e); + tracing::warn!("sipflow flusher: DDL failed: {e}"); + } } } - sqlx::query( + exec_ddl( + &mut conn, "CREATE TABLE IF NOT EXISTS call_meta ( id INTEGER PRIMARY KEY AUTOINCREMENT, callid TEXT UNIQUE NOT NULL )", ) - .execute(&mut conn) - .await - .ok(); - - sqlx::query("CREATE INDEX IF NOT EXISTS idx_callid ON call_meta(callid)") - .execute(&mut conn) - .await - .ok(); - - sqlx::query( + .await; + exec_ddl( + &mut conn, + "CREATE INDEX IF NOT EXISTS idx_callid ON call_meta(callid)", + ) + .await; + exec_ddl( + &mut conn, "CREATE TABLE IF NOT EXISTS sip_msgs ( id INTEGER PRIMARY KEY, call_id INTEGER NOT NULL, @@ -405,16 +429,14 @@ async fn open_db_with_pragmas(db_path: &PathBuf) -> SqliteConnection { size INTEGER NOT NULL )", ) - .execute(&mut conn) - .await - .ok(); - - sqlx::query("CREATE INDEX IF NOT EXISTS idx_sip_call ON sip_msgs(call_id)") - .execute(&mut conn) - .await - .ok(); - - sqlx::query( + .await; + exec_ddl( + &mut conn, + "CREATE INDEX IF NOT EXISTS idx_sip_call ON sip_msgs(call_id)", + ) + .await; + exec_ddl( + &mut conn, "CREATE TABLE IF NOT EXISTS media_msgs ( id INTEGER PRIMARY KEY, call_id INTEGER NOT NULL, @@ -425,26 +447,20 @@ async fn open_db_with_pragmas(db_path: &PathBuf) -> SqliteConnection { size INTEGER NOT NULL )", ) - .execute(&mut conn) - .await - .ok(); + .await; // `idx_media_call_timestamp` (call_id, timestamp) already covers all // call_id-prefix lookups, so the standalone `idx_media_call` index is // redundant. Every media row insert updated two B-trees; dropping the // redundant one measurably raises sustained write throughput on large // DBs. Existing databases are migrated (one-time) by the DROP below. - sqlx::query("DROP INDEX IF EXISTS idx_media_call") - .execute(&mut conn) - .await - .ok(); + exec_ddl(&mut conn, "DROP INDEX IF EXISTS idx_media_call").await; - sqlx::query( + exec_ddl( + &mut conn, "CREATE INDEX IF NOT EXISTS idx_media_call_timestamp ON media_msgs(call_id, timestamp)", ) - .execute(&mut conn) - .await - .ok(); + .await; conn } @@ -488,6 +504,10 @@ async fn flush_to_db( { metrics::gauge!("sipflow_db_file_bytes", "component" => "sipflow") .set(md.len() as f64); + sqlite_metrics::set_gauge("db", md.len()); + if let Ok(wmd) = std::fs::metadata(format!("{}-wal", path.display())) { + sqlite_metrics::set_gauge("wal", wmd.len()); + } } tracing::trace!( batch_size, @@ -527,6 +547,25 @@ async fn flush_to_db( ) .increment(1); } + sqlite_metrics::record_checkpoint( + "passive", + ckpt_start.elapsed(), + row.busy as u64, + ); + // Piggyback the page gauges on the throttled checkpoint + // cadence — two cheap PRAGMAs against the write conn. + if let (Some(pc), Some(fl)) = ( + sqlx::query_scalar::<_, i64>("PRAGMA page_count") + .fetch_one(&mut *conn) + .await + .ok(), + sqlx::query_scalar::<_, i64>("PRAGMA freelist_count") + .fetch_one(&mut *conn) + .await + .ok(), + ) { + sqlite_metrics::set_page_gauges(pc.max(0) as u64, fl.max(0) as u64); + } } *last_checkpoint = Instant::now(); metrics::histogram!("sipflow_wal_checkpoint_seconds", "component" => "sipflow") @@ -597,7 +636,18 @@ async fn insert_callids( b.push_bind(c); }); qb.push(" ON CONFLICT(callid) DO UPDATE SET callid=callid RETURNING id"); - let rows = qb.build().fetch_all(&mut **tx).await?; + let start = Instant::now(); + let rows = qb + .build() + .fetch_all(&mut **tx) + .await + .inspect(|rows| { + sqlite_metrics::record_statement("insert", 0, rows.len() as u64, start.elapsed()) + }) + .map_err(|e| { + sqlite_metrics::record_error("insert", &e); + e + })?; for (row, callid) in rows.iter().zip(chunk.iter()) { let id: i32 = row.try_get("id")?; out.insert(callid.clone(), id); @@ -650,6 +700,28 @@ async fn flush_slice( conn: &mut SqliteConnection, metas: Vec, call_id_cache: &mut LruCache, +) -> Result<(usize, usize, usize)> { + let tx_timer = sqlite_metrics::TxTimer::begin(); + let res = flush_slice_inner(conn, metas, call_id_cache).await; + match &res { + Ok(_) => tx_timer.finish(true), + Err(e) => { + tx_timer.finish(false); + if let Some(sqlx_err) = e.downcast_ref::() { + sqlite_metrics::record_error("transaction", sqlx_err); + } else { + metrics::counter!("sqlite_errors_total", "database" => sqlite_metrics::DATABASE, "kind" => "transaction", "error" => "other") + .increment(1); + } + } + } + res +} + +async fn flush_slice_inner( + conn: &mut SqliteConnection, + metas: Vec, + call_id_cache: &mut LruCache, ) -> Result<(usize, usize, usize)> { let mut tx = conn.begin().await?; @@ -705,7 +777,13 @@ async fn flush_slice( .push_bind(r.offset) .push_bind(r.size); }); - qb.build().execute(&mut *tx).await?; + let start = Instant::now(); + if let Err(e) = qb.build().execute(&mut *tx).await.inspect(|r| { + sqlite_metrics::record_statement("insert", 0, r.rows_affected(), start.elapsed()) + }) { + sqlite_metrics::record_error("insert", &e); + return Err(e.into()); + } } for chunk in rtp_rows.chunks(INSERT_CHUNK_ROWS) { @@ -720,7 +798,13 @@ async fn flush_slice( .push_bind(r.offset) .push_bind(r.size); }); - qb.build().execute(&mut *tx).await?; + let start = Instant::now(); + if let Err(e) = qb.build().execute(&mut *tx).await.inspect(|r| { + sqlite_metrics::record_statement("insert", 0, r.rows_affected(), start.elapsed()) + }) { + sqlite_metrics::record_error("insert", &e); + return Err(e.into()); + } } tx.commit().await?; diff --git a/crates/rustpbx-sipflow/src/lib.rs b/crates/rustpbx-sipflow/src/lib.rs index 2e3ce3e16..f67a0765d 100644 --- a/crates/rustpbx-sipflow/src/lib.rs +++ b/crates/rustpbx-sipflow/src/lib.rs @@ -7,6 +7,7 @@ pub mod protocol; pub mod rtp_stats; pub mod sdp_utils; pub mod shard; +pub mod sqlite_metrics; pub mod storage; pub mod wav_utils; diff --git a/crates/rustpbx-sipflow/src/sqlite_metrics.rs b/crates/rustpbx-sipflow/src/sqlite_metrics.rs new file mode 100644 index 000000000..08e492450 --- /dev/null +++ b/crates/rustpbx-sipflow/src/sqlite_metrics.rs @@ -0,0 +1,200 @@ +//! General-purpose sqlite instrumentation. +//! +//! Complements the domain-specific `sipflow_*` metrics with a `sqlite_*` +//! taxonomy so the same questions (connections, statements, transactions, +//! WAL health) can be asked of any sqlite usage in this binary. Every +//! metric carries a `database` label; sipflow always reports `sipflow`. +//! Statement-level only: multi-row/batched operations, never per-row. + +use std::path::Path; +use std::time::{Duration, Instant}; +use sqlx::{Connection, SqliteConnection}; + +pub const DATABASE: &str = "sipflow"; + +/// Classify a SQL statement by its first keyword. +pub fn classify_kind(sql: &str) -> &'static str { + let kw = sql.trim_start(); + let starts = |p: &str| kw.len() >= p.len() && kw[..p.len()].eq_ignore_ascii_case(p); + if starts("select") { + "select" + } else if starts("insert") { + "insert" + } else if starts("update") { + "update" + } else if starts("delete") { + "delete" + } else if starts("pragma") { + "pragma" + } else if starts("create") { + "ddl" + } else if starts("checkpoint") { + "checkpoint" + } else { + "other" + } +} + +/// Drop guard decrementing the open-connections gauge. Tie its lifetime to +/// the connection it tracks. +pub struct ConnectionGuard { + database: &'static str, + role: &'static str, +} + +impl ConnectionGuard { + fn acquire(role: &'static str) -> Self { + metrics::gauge!("sqlite_connections_open", "database" => DATABASE, "role" => role) + .increment(1.0); + Self { + database: DATABASE, + role, + } + } + + /// Explicit release for long-lived connections not tied to a guard. + pub fn release(role: &'static str) { + metrics::gauge!("sqlite_connections_open", "database" => DATABASE, "role" => role) + .decrement(1.0); + } +} + +impl Drop for ConnectionGuard { + fn drop(&mut self) { + metrics::gauge!("sqlite_connections_open", "database" => self.database, "role" => self.role) + .decrement(1.0); + } +} + +fn error_class(err: &sqlx::Error) -> &'static str { + match err { + sqlx::Error::Database(db) => match db.code().and_then(|c| c.parse::().ok()).unwrap_or(0) { + 5 => "busy", + 6 => "locked", + 19 | 2067 => "constraint", + _ => "other", + }, + _ => "other", + } +} + +pub fn record_error(kind: &'static str, err: &sqlx::Error) { + metrics::counter!("sqlite_errors_total", "database" => DATABASE, "kind" => kind, "error" => error_class(err)) + .increment(1); +} + +pub fn record_statement( + kind: &'static str, + rows_read: u64, + rows_written: u64, + elapsed: Duration, +) { + metrics::counter!("sqlite_statements_total", "database" => DATABASE, "kind" => kind) + .increment(1); + if rows_read > 0 { + metrics::counter!("sqlite_statement_rows_total", "database" => DATABASE, "kind" => kind, "direction" => "read") + .increment(rows_read); + } + if rows_written > 0 { + metrics::counter!("sqlite_statement_rows_total", "database" => DATABASE, "kind" => kind, "direction" => "written") + .increment(rows_written); + } + metrics::histogram!("sqlite_statement_seconds", "database" => DATABASE, "kind" => kind) + .record(elapsed.as_secs_f64()); +} + +/// A SELECT result set: count returned rows as reads. +pub fn record_select(rows: usize, elapsed: Duration) { + record_statement("select", rows as u64, 0, elapsed); +} + +/// Open a per-query read connection (sipflow opens one per hour bucket per +/// query). Returns the connection plus a guard that keeps +/// `sqlite_connections_open` accurate until the connection drops. +pub async fn connect_read( + db_path: &Path, +) -> Result<(SqliteConnection, ConnectionGuard), sqlx::Error> { + let start = Instant::now(); + let res = SqliteConnection::connect(&format!("sqlite:{}", db_path.display())).await; + match &res { + Ok(_) => { + metrics::counter!("sqlite_connections_opened_total", "database" => DATABASE, "role" => "read") + .increment(1); + metrics::histogram!("sqlite_connection_open_seconds", "database" => DATABASE, "role" => "read") + .record(start.elapsed().as_secs_f64()); + } + Err(e) => { + metrics::counter!("sqlite_connection_errors_total", "database" => DATABASE, "phase" => "connect") + .increment(1); + record_error("connect", e); + } + } + res.map(|c| (c, ConnectionGuard::acquire("read"))) +} + +/// Open metrics for the long-lived per-bucket write connection. +pub fn record_write_open(elapsed: Duration) { + metrics::counter!("sqlite_connections_opened_total", "database" => DATABASE, "role" => "write") + .increment(1); + metrics::histogram!("sqlite_connection_open_seconds", "database" => DATABASE, "role" => "write") + .record(elapsed.as_secs_f64()); + ConnectionGuard::acquire("write"); +} + +/// Begin-to-end timer for a write transaction. +pub struct TxTimer { + start: Instant, +} + +impl TxTimer { + pub fn begin() -> Self { + Self { + start: Instant::now(), + } + } + + pub fn finish(self, committed: bool) { + metrics::counter!("sqlite_transactions_total", "database" => DATABASE, "outcome" => if committed { "commit" } else { "rollback" }) + .increment(1); + metrics::histogram!("sqlite_transaction_seconds", "database" => DATABASE) + .record(self.start.elapsed().as_secs_f64()); + } +} + +pub fn record_checkpoint(kind: &'static str, elapsed: Duration, busy: u64) { + metrics::counter!("sqlite_wal_checkpoint_total", "database" => DATABASE, "kind" => kind) + .increment(1); + metrics::histogram!("sqlite_wal_checkpoint_seconds", "database" => DATABASE, "kind" => kind) + .record(elapsed.as_secs_f64()); + if busy > 0 { + metrics::counter!("sqlite_wal_checkpoint_busy_total", "database" => DATABASE) + .increment(busy); + } +} + +pub fn set_gauge(name_file: &str, bytes: u64) { + match name_file { + "db" => metrics::gauge!("sqlite_db_bytes", "database" => DATABASE).set(bytes as f64), + "wal" => metrics::gauge!("sqlite_wal_bytes", "database" => DATABASE).set(bytes as f64), + _ => {} + } +} + +pub fn set_page_gauges(page_count: u64, freelist_pages: u64) { + metrics::gauge!("sqlite_page_count", "database" => DATABASE).set(page_count as f64); + metrics::gauge!("sqlite_freelist_pages", "database" => DATABASE).set(freelist_pages as f64); +} + +#[cfg(test)] +mod tests { + use super::classify_kind; + + #[test] + fn classify_first_keyword() { + assert_eq!(classify_kind("SELECT * FROM t"), "select"); + assert_eq!(classify_kind(" insert INTO t VALUES (1)"), "insert"); + assert_eq!(classify_kind("PRAGMA wal_checkpoint(PASSIVE)"), "pragma"); + assert_eq!(classify_kind("CREATE INDEX i ON t(c)"), "ddl"); + assert_eq!(classify_kind("explain select 1"), "other"); + } +} diff --git a/crates/rustpbx-sipflow/src/storage.rs b/crates/rustpbx-sipflow/src/storage.rs index a0dddc3b1..f9c20f8aa 100644 --- a/crates/rustpbx-sipflow/src/storage.rs +++ b/crates/rustpbx-sipflow/src/storage.rs @@ -3,17 +3,18 @@ use crate::flusher::{FlushCommand, FlushMeta}; use crate::protocol::{MsgType, Packet}; use crate::rtp_stats::{MediaStatsAccumulator, parse_rtp_stats_header}; use crate::shard::{RouterState, bucket_query_dirs, detect_bucket_layout}; +use crate::sqlite_metrics; use crate::{SipFlowItem, SipFlowMediaStats, SipFlowMsgType}; use anyhow::Result; use bytes::{BufMut, Bytes}; use chrono::{DateTime, Datelike, Local, Timelike}; use futures::TryStreamExt; -use sqlx::{Connection, SqliteConnection}; +use sqlx::SqliteConnection; use std::io::{Read, SeekFrom, Write}; +use std::time::Instant; use std::path::{Path, PathBuf}; use std::sync::Arc; use std::sync::atomic::{AtomicU64, Ordering}; -use std::time::Instant; use tokio::fs::{File, OpenOptions}; use tokio::io::{AsyncReadExt, AsyncSeekExt, AsyncWriteExt}; use tokio::sync::mpsc; @@ -482,7 +483,14 @@ impl StorageManager { "PRAGMA busy_timeout=5000", "PRAGMA query_only=1", ] { - let _ = sqlx::query(pragma).execute(&mut *conn).await; + let start = Instant::now(); + match sqlx::query(pragma).execute(&mut *conn).await { + Ok(_) => sqlite_metrics::record_statement("pragma", 0, 0, start.elapsed()), + Err(e) => { + sqlite_metrics::record_error("pragma", &e); + tracing::debug!("read-conn pragma failed: {pragma}: {e}"); + } + } } } @@ -508,14 +516,13 @@ impl StorageManager { continue; } - let mut conn = - SqliteConnection::connect(&format!("sqlite:{}", db_path.to_string_lossy())) - .await?; + let (mut conn, _conn_guard) = sqlite_metrics::connect_read(&db_path).await?; Self::configure_read_conn(&mut conn).await; let mut raw_file = File::open(raw_path).await?; let mut current_pos = None; // Query using JOIN with call_meta + let q_start = Instant::now(); let rows = sqlx::query_as::<_, SipPacketRow>( "SELECT s.src AS src, s.dst AS dst, @@ -533,7 +540,12 @@ impl StorageManager { .bind(start_ts) .bind(end_ts) .fetch_all(&mut conn) - .await?; + .await + .inspect(|rows| sqlite_metrics::record_select(rows.len(), q_start.elapsed())) + .map_err(|e| { + sqlite_metrics::record_error("select", &e); + e + })?; for row in rows { let offset = u64::try_from(row.offset)?; @@ -578,13 +590,12 @@ impl StorageManager { continue; } - let mut conn = - SqliteConnection::connect(&format!("sqlite:{}", db_path.to_string_lossy())) - .await?; + let (mut conn, _conn_guard) = sqlite_metrics::connect_read(&db_path).await?; Self::configure_read_conn(&mut conn).await; let mut raw_file = File::open(raw_path).await?; let mut current_pos = None; + let q_start = Instant::now(); let rows = sqlx::query_as::<_, SipPacketRow>( "SELECT s.src AS src, s.dst AS dst, @@ -599,7 +610,12 @@ impl StorageManager { .bind(start_ts) .bind(end_ts) .fetch_all(&mut conn) - .await?; + .await + .inspect(|rows| sqlite_metrics::record_select(rows.len(), q_start.elapsed())) + .map_err(|e| { + sqlite_metrics::record_error("select", &e); + e + })?; for row in rows { let offset = u64::try_from(row.offset)?; @@ -668,11 +684,10 @@ impl StorageManager { continue; } - let mut conn = - SqliteConnection::connect(&format!("sqlite:{}", db_path.to_string_lossy())) - .await?; + let (mut conn, _conn_guard) = sqlite_metrics::connect_read(&db_path).await?; Self::configure_read_conn(&mut conn).await; + let q_start = Instant::now(); let mut rows = sqlx::query_as::<_, MediaSourceRow>( "SELECT m.leg AS leg, m.src AS src @@ -688,11 +703,18 @@ impl StorageManager { .bind(end_ts) .fetch(&mut conn); - while let Some(row) = rows.try_next().await? { + let mut n_rows = 0usize; + while let Some(row) = rows + .try_next() + .await + .inspect_err(|e| sqlite_metrics::record_error("select", e))? + { + n_rows += 1; if seen.insert((row.leg, row.src.clone())) { results.push(row); } } + sqlite_metrics::record_select(n_rows, q_start.elapsed()); } } @@ -721,13 +743,12 @@ impl StorageManager { continue; } - let mut conn = - SqliteConnection::connect(&format!("sqlite:{}", db_path.to_string_lossy())) - .await?; + let (mut conn, _conn_guard) = sqlite_metrics::connect_read(&db_path).await?; Self::configure_read_conn(&mut conn).await; let mut raw_file = File::open(&raw_path).await?; let mut current_pos = None; + let q_start = Instant::now(); let rows = sqlx::query_as::<_, MediaPacketRow>( "SELECT s.leg AS leg, s.src AS src, @@ -745,7 +766,12 @@ impl StorageManager { .bind(start_ts) .bind(end_ts) .fetch_all(&mut conn) - .await?; + .await + .inspect(|rows| sqlite_metrics::record_select(rows.len(), q_start.elapsed())) + .map_err(|e| { + sqlite_metrics::record_error("select", &e); + e + })?; if !rows.is_empty() { tracing::info!( @@ -1615,9 +1641,11 @@ mod tests { let db_path = dir.path().join("sipflow.db"); let raw_path = dir.path().join("data.raw"); - let mut conn = SqliteConnection::connect(&format!("sqlite:{}", db_path.display())) - .await - .expect("open sipflow.db"); + let mut conn = ::connect( + &format!("sqlite:{}", db_path.display()), + ) + .await + .expect("open sipflow.db"); let rows = sqlx::query_as::<_, (i32, i64, i64, i64)>( "SELECT m.leg, m.timestamp, m.offset, m.size FROM media_msgs m From 7853c0616a7cf410e720e8cc7ee7da0216797465 Mon Sep 17 00:00:00 2001 From: tongfengyuan <71140753@chinatelecom.cn> Date: Wed, 2 Sep 2026 09:57:22 +0800 Subject: [PATCH 3/3] fix(sipflow): only release the write-connection gauge when a conn existed The bucket-rotate path released sqlite_connections_open{role=write} unconditionally; on the first rotate with no open connection (startup layout pass) that drove the gauge negative. --- crates/rustpbx-sipflow/src/flusher.rs | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/crates/rustpbx-sipflow/src/flusher.rs b/crates/rustpbx-sipflow/src/flusher.rs index ce8cbec8f..56b9d8f4f 100644 --- a/crates/rustpbx-sipflow/src/flusher.rs +++ b/crates/rustpbx-sipflow/src/flusher.rs @@ -343,8 +343,9 @@ async fn handle_flush_command( .execute(&mut *conn) .await; } - sqlite_metrics::ConnectionGuard::release("write"); - drop(db_conn.take()); + if db_conn.take().is_some() { + sqlite_metrics::ConnectionGuard::release("write"); + } *db_conn = Some(open_db_with_pragmas(&new_db_path).await); *db_path = Some(new_db_path); call_id_cache.clear();