Fix Mutex unwrap calls to handle poisoned mutex errors
Replaced all .unwrap() calls on Mutex::lock() with proper error handling using map_err to prevent panics when the mutex is poisoned. This prevents silent crashes in production when any thread panics while holding the storage mutex lock.
This commit is contained in:
parent
656992221a
commit
b9db4133be
1 changed files with 28 additions and 7 deletions
|
|
@ -86,7 +86,10 @@ impl Storage {
|
||||||
/// Store a metric in the buffer
|
/// Store a metric in the buffer
|
||||||
pub fn store_metric(&self, metric: &Metric) -> Result<()> {
|
pub fn store_metric(&self, metric: &Metric) -> Result<()> {
|
||||||
let data = serde_json::to_string(metric)?;
|
let data = serde_json::to_string(metric)?;
|
||||||
let conn = self.conn.lock().unwrap();
|
let conn = self
|
||||||
|
.conn
|
||||||
|
.lock()
|
||||||
|
.map_err(|_| StorageError::Database(rusqlite::Error::InvalidQuery))?;
|
||||||
|
|
||||||
conn.execute(
|
conn.execute(
|
||||||
"INSERT INTO metrics (metric_type, data, timestamp) VALUES (?1, ?2, ?3)",
|
"INSERT INTO metrics (metric_type, data, timestamp) VALUES (?1, ?2, ?3)",
|
||||||
|
|
@ -98,7 +101,10 @@ impl Storage {
|
||||||
|
|
||||||
/// Get pending metrics that haven't been sent yet
|
/// Get pending metrics that haven't been sent yet
|
||||||
pub fn get_pending_metrics(&self, limit: usize) -> Result<Vec<(i64, Metric)>> {
|
pub fn get_pending_metrics(&self, limit: usize) -> Result<Vec<(i64, Metric)>> {
|
||||||
let conn = self.conn.lock().unwrap();
|
let conn = self
|
||||||
|
.conn
|
||||||
|
.lock()
|
||||||
|
.map_err(|_| StorageError::Database(rusqlite::Error::InvalidQuery))?;
|
||||||
|
|
||||||
let mut stmt = conn
|
let mut stmt = conn
|
||||||
.prepare("SELECT id, data FROM metrics WHERE sent = 0 ORDER BY created_at LIMIT ?1")?;
|
.prepare("SELECT id, data FROM metrics WHERE sent = 0 ORDER BY created_at LIMIT ?1")?;
|
||||||
|
|
@ -126,7 +132,10 @@ impl Storage {
|
||||||
return Ok(());
|
return Ok(());
|
||||||
}
|
}
|
||||||
|
|
||||||
let conn = self.conn.lock().unwrap();
|
let conn = self
|
||||||
|
.conn
|
||||||
|
.lock()
|
||||||
|
.map_err(|_| StorageError::Database(rusqlite::Error::InvalidQuery))?;
|
||||||
let placeholders = ids.iter().map(|_| "?").collect::<Vec<_>>().join(",");
|
let placeholders = ids.iter().map(|_| "?").collect::<Vec<_>>().join(",");
|
||||||
let query = format!("UPDATE metrics SET sent = 1 WHERE id IN ({})", placeholders);
|
let query = format!("UPDATE metrics SET sent = 1 WHERE id IN ({})", placeholders);
|
||||||
|
|
||||||
|
|
@ -138,7 +147,10 @@ impl Storage {
|
||||||
|
|
||||||
/// Clean up old metrics that have been sent
|
/// Clean up old metrics that have been sent
|
||||||
pub fn cleanup_old_metrics(&self) -> Result<()> {
|
pub fn cleanup_old_metrics(&self) -> Result<()> {
|
||||||
let conn = self.conn.lock().unwrap();
|
let conn = self
|
||||||
|
.conn
|
||||||
|
.lock()
|
||||||
|
.map_err(|_| StorageError::Database(rusqlite::Error::InvalidQuery))?;
|
||||||
|
|
||||||
conn.execute(
|
conn.execute(
|
||||||
"DELETE FROM metrics WHERE sent = 1 AND created_at < datetime('now', '-24 hours')",
|
"DELETE FROM metrics WHERE sent = 1 AND created_at < datetime('now', '-24 hours')",
|
||||||
|
|
@ -151,7 +163,10 @@ impl Storage {
|
||||||
/// Get the last poll time for equipment
|
/// Get the last poll time for equipment
|
||||||
#[allow(dead_code)] // Used in full SNMP implementation
|
#[allow(dead_code)] // Used in full SNMP implementation
|
||||||
pub fn get_last_poll_time(&self, equipment_id: &str) -> Result<Option<Timestamp>> {
|
pub fn get_last_poll_time(&self, equipment_id: &str) -> Result<Option<Timestamp>> {
|
||||||
let conn = self.conn.lock().unwrap();
|
let conn = self
|
||||||
|
.conn
|
||||||
|
.lock()
|
||||||
|
.map_err(|_| StorageError::Database(rusqlite::Error::InvalidQuery))?;
|
||||||
|
|
||||||
let result: std::result::Result<String, _> = conn.query_row(
|
let result: std::result::Result<String, _> = conn.query_row(
|
||||||
"SELECT last_poll_time FROM last_poll_times WHERE equipment_id = ?1",
|
"SELECT last_poll_time FROM last_poll_times WHERE equipment_id = ?1",
|
||||||
|
|
@ -172,7 +187,10 @@ impl Storage {
|
||||||
|
|
||||||
/// Update the last poll time for equipment
|
/// Update the last poll time for equipment
|
||||||
pub fn update_last_poll_time(&self, equipment_id: &str) -> Result<()> {
|
pub fn update_last_poll_time(&self, equipment_id: &str) -> Result<()> {
|
||||||
let conn = self.conn.lock().unwrap();
|
let conn = self
|
||||||
|
.conn
|
||||||
|
.lock()
|
||||||
|
.map_err(|_| StorageError::Database(rusqlite::Error::InvalidQuery))?;
|
||||||
let now = Timestamp::now().to_rfc3339();
|
let now = Timestamp::now().to_rfc3339();
|
||||||
|
|
||||||
conn.execute(
|
conn.execute(
|
||||||
|
|
@ -185,7 +203,10 @@ impl Storage {
|
||||||
|
|
||||||
/// Get all last poll times
|
/// Get all last poll times
|
||||||
pub fn get_all_last_poll_times(&self) -> Result<std::collections::HashMap<String, Timestamp>> {
|
pub fn get_all_last_poll_times(&self) -> Result<std::collections::HashMap<String, Timestamp>> {
|
||||||
let conn = self.conn.lock().unwrap();
|
let conn = self
|
||||||
|
.conn
|
||||||
|
.lock()
|
||||||
|
.map_err(|_| StorageError::Database(rusqlite::Error::InvalidQuery))?;
|
||||||
|
|
||||||
let mut stmt = conn.prepare("SELECT equipment_id, last_poll_time FROM last_poll_times")?;
|
let mut stmt = conn.prepare("SELECT equipment_id, last_poll_time FROM last_poll_times")?;
|
||||||
|
|
||||||
|
|
|
||||||
Loading…
Add table
Reference in a new issue