i KNEW there are problems, probably

This commit is contained in:
lda
2026-05-03 17:43:41 +07:00 Verified
parent b573e9f27f
commit 0c22fb5e41
16 changed files with 25 additions and 168 deletions
@@ -1,12 +0,0 @@
{
"db_name": "SQLite",
"query": "INSERT OR REPLACE INTO agents (agent_id, agent_token) VALUES (?1, ?2)",
"describe": {
"columns": [],
"parameters": {
"Right": 2
},
"nullable": []
},
"hash": "1d4677e5e37a32a8c0f8ab29ec07986142a621c099a62d3c5973f368e664991c"
}
@@ -1,12 +0,0 @@
{
"db_name": "SQLite",
"query": "INSERT OR REPLACE INTO active_alerts\n (alert_id, kind, severity, status, agent_id, message, value, threshold,\n last_seen_unix, metadata_json)\n VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10)",
"describe": {
"columns": [],
"parameters": {
"Right": 10
},
"nullable": []
},
"hash": "a69a1ea43b1a59254cb29b549b5cd3ebfcab8130941dc2d9a78814bc2072c0d5"
}
@@ -1,12 +0,0 @@
{
"db_name": "SQLite",
"query": "INSERT OR REPLACE INTO agent_meta (agent_id, nickname) VALUES (?1, ?2)",
"describe": {
"columns": [],
"parameters": {
"Right": 2
},
"nullable": []
},
"hash": "b69fb8f030b9745db1d8c9db8225426334b4d1edea5d8a8c9b289c17f194c153"
}
@@ -1,12 +0,0 @@
{
"db_name": "SQLite",
"query": "INSERT OR REPLACE INTO audit_events\n (event_key, event_id, ts_unix, actor_type, actor_id, agent_id, request_id,\n event_type, outcome, latency_ms, message, metadata_json)\n VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12)",
"describe": {
"columns": [],
"parameters": {
"Right": 12
},
"nullable": []
},
"hash": "c496c7a8ada53d6d88ea1a29d9e7bf178fe5f3ee9067357dba05384fa1e2772c"
}
@@ -1,12 +0,0 @@
{
"db_name": "SQLite",
"query": "INSERT OR REPLACE INTO alert_transitions\n (transition_key, transition_id, ts_unix, alert_id, kind, agent_id,\n from_status, to_status, message, metadata_json)\n VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10)",
"describe": {
"columns": [],
"parameters": {
"Right": 10
},
"nullable": []
},
"hash": "e508aeb39a57b9defc7cec1bf854669150b8f3a02a7d8312e8fea5951391c1bb"
}
@@ -1,12 +0,0 @@
{
"db_name": "SQLite",
"query": "INSERT OR REPLACE INTO agents (agent_id, agent_token) VALUES (?1, ?2)",
"describe": {
"columns": [],
"parameters": {
"Right": 2
},
"nullable": []
},
"hash": "1d4677e5e37a32a8c0f8ab29ec07986142a621c099a62d3c5973f368e664991c"
}
@@ -1,12 +0,0 @@
{
"db_name": "SQLite",
"query": "INSERT OR REPLACE INTO active_alerts\n (alert_id, kind, severity, status, agent_id, message, value, threshold,\n last_seen_unix, metadata_json)\n VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10)",
"describe": {
"columns": [],
"parameters": {
"Right": 10
},
"nullable": []
},
"hash": "a69a1ea43b1a59254cb29b549b5cd3ebfcab8130941dc2d9a78814bc2072c0d5"
}
@@ -1,12 +0,0 @@
{
"db_name": "SQLite",
"query": "INSERT OR REPLACE INTO agent_meta (agent_id, nickname) VALUES (?1, ?2)",
"describe": {
"columns": [],
"parameters": {
"Right": 2
},
"nullable": []
},
"hash": "b69fb8f030b9745db1d8c9db8225426334b4d1edea5d8a8c9b289c17f194c153"
}
@@ -1,12 +0,0 @@
{
"db_name": "SQLite",
"query": "INSERT OR REPLACE INTO audit_events\n (event_key, event_id, ts_unix, actor_type, actor_id, agent_id, request_id,\n event_type, outcome, latency_ms, message, metadata_json)\n VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12)",
"describe": {
"columns": [],
"parameters": {
"Right": 12
},
"nullable": []
},
"hash": "c496c7a8ada53d6d88ea1a29d9e7bf178fe5f3ee9067357dba05384fa1e2772c"
}
@@ -1,12 +0,0 @@
{
"db_name": "SQLite",
"query": "INSERT OR REPLACE INTO alert_transitions\n (transition_key, transition_id, ts_unix, alert_id, kind, agent_id,\n from_status, to_status, message, metadata_json)\n VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10)",
"describe": {
"columns": [],
"parameters": {
"Right": 10
},
"nullable": []
},
"hash": "e508aeb39a57b9defc7cec1bf854669150b8f3a02a7d8312e8fea5951391c1bb"
}
+13 -2
View File
@@ -17,10 +17,21 @@ pub struct Store {
db_path: PathBuf, db_path: PathBuf,
pool: SqlitePool, pool: SqlitePool,
} }
impl Store {
/// Begins a new transaction with an IMMEDIATE write lock.
///
/// Use this for all transactions that perform writes to avoid SQLite upgrade deadlocks.
pub async fn begin_write(&self) -> Result<Transaction<'static, Sqlite>> {
self.pool
.begin_with("BEGIN IMMEDIATE")
.await
.context("failed to begin write transaction")
}
}
const SCHEMA_VERSION_KEY: &str = "schema_version"; pub(crate) const SCHEMA_VERSION_KEY: &str = "schema_version";
const SEEDED_ENROLL_TOKEN_PREFIX: &str = "seeded_enroll_token:"; const SEEDED_ENROLL_TOKEN_PREFIX: &str = "seeded_enroll_token:";
const SCHEMA_VERSION: u32 = 2; pub(crate) const SCHEMA_VERSION: u32 = 2;
pub(crate) mod agent_devices; pub(crate) mod agent_devices;
mod alerts; mod alerts;
@@ -15,11 +15,7 @@ impl Store {
agent_id: &str, agent_id: &str,
devices: &[Device], devices: &[Device],
) -> Result<usize> { ) -> Result<usize> {
let mut tx = self let mut tx = self.begin_write().await?;
.pool
.begin()
.await
.context("failed starting device snapshot transaction")?;
let mut incoming_keys = std::collections::HashSet::with_capacity(devices.len()); let mut incoming_keys = std::collections::HashSet::with_capacity(devices.len());
let snapshot_time = now_unix(); let snapshot_time = now_unix();
@@ -1,3 +1,4 @@
use super::helpers::core::*;
use super::*; use super::*;
impl Store { impl Store {
@@ -5,6 +6,8 @@ impl Store {
&self, &self,
current: &[AlertState], current: &[AlertState],
) -> Result<Vec<AlertTransition>> { ) -> Result<Vec<AlertTransition>> {
let mut tx = self.begin_write().await?;
let mut previous = std::collections::HashMap::<String, AlertState>::new(); let mut previous = std::collections::HashMap::<String, AlertState>::new();
let rows = sqlx::query_as!( let rows = sqlx::query_as!(
AlertStateRow, AlertStateRow,
@@ -13,7 +16,7 @@ impl Store {
value, threshold, last_seen_unix, metadata_json as "metadata_json!" value, threshold, last_seen_unix, metadata_json as "metadata_json!"
FROM active_alerts"#, FROM active_alerts"#,
) )
.fetch_all(&self.pool) .fetch_all(&mut *tx)
.await .await
.context("failed iterating active_alerts table")?; .context("failed iterating active_alerts table")?;
for row in rows { for row in rows {
@@ -63,11 +66,6 @@ impl Store {
}); });
} }
let mut tx = self
.pool
.begin()
.await
.context("failed starting alert transaction")?;
sqlx::query!("DELETE FROM active_alerts") sqlx::query!("DELETE FROM active_alerts")
.execute(&mut *tx) .execute(&mut *tx)
.await .await
+1 -5
View File
@@ -115,11 +115,7 @@ impl Store {
} }
let expires_at = now_unix().saturating_add(seed_ttl.as_secs().max(1)); let expires_at = now_unix().saturating_add(seed_ttl.as_secs().max(1));
let mut tx = self let mut tx = self.begin_write().await?;
.pool
.begin()
.await
.context("failed starting bootstrap token transaction")?;
let expires_at_i64 = i64::try_from(expires_at).context("token expiry overflow")?; let expires_at_i64 = i64::try_from(expires_at).context("token expiry overflow")?;
sqlx::query!( sqlx::query!(
"INSERT OR REPLACE INTO enroll_tokens (token, expires_at_unix) VALUES (?1, ?2)", "INSERT OR REPLACE INTO enroll_tokens (token, expires_at_unix) VALUES (?1, ?2)",
+4 -20
View File
@@ -11,11 +11,7 @@ impl Store {
let device_id = format!("dev-{}", Uuid::new_v4()); let device_id = format!("dev-{}", Uuid::new_v4());
let now = now_unix(); let now = now_unix();
let mut tx = self let mut tx = self.begin_write().await?;
.pool
.begin()
.await
.context("failed starting known device transaction")?;
let pinned = if input.pinned { 1_i64 } else { 0_i64 }; let pinned = if input.pinned { 1_i64 } else { 0_i64 };
let now_i64 = i64::try_from(now).context("known device timestamp overflow")?; let now_i64 = i64::try_from(now).context("known device timestamp overflow")?;
sqlx::query!( sqlx::query!(
@@ -84,11 +80,7 @@ impl Store {
let now = now_unix(); let now = now_unix();
let now_i64 = i64::try_from(now).context("known device timestamp overflow")?; let now_i64 = i64::try_from(now).context("known device timestamp overflow")?;
let mut tx = self let mut tx = self.begin_write().await?;
.pool
.begin()
.await
.context("failed starting known device merge transaction")?;
let target_exists = sqlx::query_scalar!( let target_exists = sqlx::query_scalar!(
r#"SELECT COUNT(*) as "count!: i64" FROM known_devices WHERE device_id = ?1"#, r#"SELECT COUNT(*) as "count!: i64" FROM known_devices WHERE device_id = ?1"#,
@@ -149,11 +141,7 @@ impl Store {
) -> Result<Option<KnownDevice>> { ) -> Result<Option<KnownDevice>> {
let identifier = normalize_device_identifier(input)?; let identifier = normalize_device_identifier(input)?;
let now = now_unix(); let now = now_unix();
let mut tx = self let mut tx = self.begin_write().await?;
.pool
.begin()
.await
.context("failed starting device identifier transaction")?;
let exists = sqlx::query_scalar!( let exists = sqlx::query_scalar!(
r#"SELECT COUNT(*) as "count!: i64" FROM known_devices WHERE device_id = ?1"#, r#"SELECT COUNT(*) as "count!: i64" FROM known_devices WHERE device_id = ?1"#,
device_id device_id
@@ -188,11 +176,7 @@ impl Store {
) -> Result<Option<KnownDevice>> { ) -> Result<Option<KnownDevice>> {
let now = now_unix(); let now = now_unix();
let now_i64 = i64::try_from(now).context("known device timestamp overflow")?; let now_i64 = i64::try_from(now).context("known device timestamp overflow")?;
let mut tx = self let mut tx = self.begin_write().await?;
.pool
.begin()
.await
.context("failed starting device identifier detach transaction")?;
let exists = sqlx::query_scalar!( let exists = sqlx::query_scalar!(
r#"SELECT COUNT(*) as "count!: i64" FROM known_devices WHERE device_id = ?1"#, r#"SELECT COUNT(*) as "count!: i64" FROM known_devices WHERE device_id = ?1"#,
device_id device_id
@@ -2,11 +2,7 @@ use super::*;
impl Store { impl Store {
pub async fn enroll(&self, enroll_token: &str) -> Result<IssuedAgent> { pub async fn enroll(&self, enroll_token: &str) -> Result<IssuedAgent> {
let mut tx = self let mut tx = self.begin_write().await?;
.pool
.begin()
.await
.context("failed starting enroll transaction")?;
let expires_at_unix = sqlx::query_scalar!( let expires_at_unix = sqlx::query_scalar!(
"SELECT expires_at_unix FROM enroll_tokens WHERE token = ?1", "SELECT expires_at_unix FROM enroll_tokens WHERE token = ?1",
enroll_token enroll_token
@@ -102,11 +98,7 @@ impl Store {
} }
pub async fn revoke_agent(&self, agent_id: &str) -> Result<bool> { pub async fn revoke_agent(&self, agent_id: &str) -> Result<bool> {
let mut tx = self let mut tx = self.begin_write().await?;
.pool
.begin()
.await
.context("failed starting revoke transaction")?;
let result = sqlx::query!("DELETE FROM agents WHERE agent_id = ?1", agent_id) let result = sqlx::query!("DELETE FROM agents WHERE agent_id = ?1", agent_id)
.execute(&mut *tx) .execute(&mut *tx)
.await .await