use std::fs;
use std::path::Path;
use std::time::{SystemTime, UNIX_EPOCH};
use chio_kernel::{ExecutionNonceStore, KernelError};
use r2d2::Pool;
use r2d2_sqlite::SqliteConnectionManager;
use rusqlite::params;
const RETENTION_GRACE_SECS: i64 = 60;
#[derive(Debug)]
pub struct SqliteExecutionNonceStoreError(String);
impl std::fmt::Display for SqliteExecutionNonceStoreError {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
write!(f, "sqlite execution nonce store error: {}", self.0)
}
}
impl std::error::Error for SqliteExecutionNonceStoreError {}
impl From<rusqlite::Error> for SqliteExecutionNonceStoreError {
fn from(e: rusqlite::Error) -> Self {
Self(e.to_string())
}
}
impl From<std::io::Error> for SqliteExecutionNonceStoreError {
fn from(e: std::io::Error) -> Self {
Self(e.to_string())
}
}
impl From<r2d2::Error> for SqliteExecutionNonceStoreError {
fn from(e: r2d2::Error) -> Self {
Self(e.to_string())
}
}
pub struct SqliteExecutionNonceStore {
pool: Pool<SqliteConnectionManager>,
}
const EXECUTION_NONCE_STORE_SUPPORTED_SCHEMA_VERSION: i32 = 0;
const EXECUTION_NONCE_STORE_SCHEMA_KEY: &str = "execution_nonce";
const EXECUTION_NONCE_STORE_LEGACY_ANCHOR_TABLES: &[&str] = &["chio_execution_nonces"];
impl SqliteExecutionNonceStore {
pub fn open(path: impl AsRef<Path>) -> Result<Self, SqliteExecutionNonceStoreError> {
let path = path.as_ref();
if let Some(parent) = crate::sqlite_parent_dir_to_create(path) {
fs::create_dir_all(&parent)?;
}
let manager = SqliteConnectionManager::file(path);
let pool = Pool::builder().max_size(8).build(manager)?;
let store = Self { pool };
store.run_migrations()?;
Ok(store)
}
pub fn open_in_memory() -> Result<Self, SqliteExecutionNonceStoreError> {
let manager = SqliteConnectionManager::memory();
let pool = Pool::builder().max_size(1).build(manager)?;
let store = Self { pool };
store.run_migrations()?;
Ok(store)
}
fn run_migrations(&self) -> Result<(), SqliteExecutionNonceStoreError> {
let conn = self
.pool
.get()
.map_err(|e| SqliteExecutionNonceStoreError(format!("pool acquire: {e}")))?;
crate::check_schema_version(
&conn,
EXECUTION_NONCE_STORE_SCHEMA_KEY,
EXECUTION_NONCE_STORE_SUPPORTED_SCHEMA_VERSION,
EXECUTION_NONCE_STORE_LEGACY_ANCHOR_TABLES,
)
.map_err(|error| SqliteExecutionNonceStoreError(error.to_string()))?;
conn.execute_batch(
r#"
PRAGMA journal_mode = WAL;
PRAGMA synchronous = FULL;
PRAGMA busy_timeout = 5000;
CREATE TABLE IF NOT EXISTS chio_execution_nonces (
nonce_id TEXT PRIMARY KEY,
consumed_at INTEGER NOT NULL,
expires_at INTEGER NOT NULL
);
CREATE INDEX IF NOT EXISTS idx_chio_execution_nonces_expires_at
ON chio_execution_nonces(expires_at);
"#,
)?;
crate::stamp_schema_version(
&conn,
EXECUTION_NONCE_STORE_SCHEMA_KEY,
EXECUTION_NONCE_STORE_SUPPORTED_SCHEMA_VERSION,
)
.map_err(|error| SqliteExecutionNonceStoreError(error.to_string()))?;
Ok(())
}
pub fn try_reserve(
&self,
nonce_id: &str,
now: i64,
expires_at: i64,
) -> Result<bool, SqliteExecutionNonceStoreError> {
if nonce_id.trim().is_empty() || nonce_id.trim() != nonce_id {
return Err(SqliteExecutionNonceStoreError(
"nonce_id must be non-empty and unpadded".to_string(),
));
}
let mut conn = self
.pool
.get()
.map_err(|e| SqliteExecutionNonceStoreError(format!("pool acquire: {e}")))?;
let tx = conn.transaction()?;
tx.execute(
"DELETE FROM chio_execution_nonces WHERE expires_at <= ?1",
params![now],
)?;
let rows = tx.execute(
r#"
INSERT INTO chio_execution_nonces (nonce_id, consumed_at, expires_at)
VALUES (?1, ?2, ?3)
ON CONFLICT(nonce_id) DO NOTHING
"#,
params![nonce_id, now, expires_at],
)?;
tx.commit()?;
Ok(rows > 0)
}
}
fn now_secs() -> i64 {
i64::try_from(
SystemTime::now()
.duration_since(UNIX_EPOCH)
.map(|d| d.as_secs())
.unwrap_or(0),
)
.unwrap_or(i64::MAX)
}
impl ExecutionNonceStore for SqliteExecutionNonceStore {
fn reserve(&self, nonce_id: &str) -> Result<bool, KernelError> {
let now = now_secs();
let estimated_nonce_expiry = now.saturating_add(
i64::try_from(chio_kernel::DEFAULT_EXECUTION_NONCE_TTL_SECS).unwrap_or(0),
);
self.reserve_until(nonce_id, estimated_nonce_expiry)
}
fn reserve_until(&self, nonce_id: &str, nonce_expires_at: i64) -> Result<bool, KernelError> {
let now = now_secs();
let retention = nonce_expires_at.saturating_add(RETENTION_GRACE_SECS);
let baseline = now.saturating_add(RETENTION_GRACE_SECS);
let expires_at = retention.max(baseline);
self.try_reserve(nonce_id, now, expires_at)
.map_err(|e| KernelError::Internal(format!("sqlite execution nonce store: {e}")))
}
fn is_consumed(&self, nonce_id: &str) -> Result<bool, KernelError> {
let now = now_secs();
let conn = self.pool.get().map_err(|error| {
KernelError::Internal(format!(
"sqlite execution nonce store pool acquire: {error}"
))
})?;
conn.query_row(
r#"
SELECT EXISTS (
SELECT 1
FROM chio_execution_nonces
WHERE nonce_id = ?1 AND expires_at > ?2
)
"#,
params![nonce_id, now],
|row| row.get(0),
)
.map_err(|error| {
KernelError::Internal(format!(
"sqlite execution nonce store consumed lookup: {error}"
))
})
}
}
#[cfg(test)]
#[allow(clippy::expect_used, clippy::unwrap_used)]
mod tests {
use super::*;
fn unique_db_path(prefix: &str) -> std::path::PathBuf {
let nonce = SystemTime::now()
.duration_since(UNIX_EPOCH)
.expect("time before epoch")
.as_nanos();
std::env::temp_dir().join(format!("{prefix}-{nonce}.sqlite3"))
}
#[test]
fn fresh_nonce_is_reserved() {
let store = SqliteExecutionNonceStore::open_in_memory().unwrap();
assert!(<SqliteExecutionNonceStore as ExecutionNonceStore>::reserve(&store, "a").unwrap());
}
#[test]
fn duplicate_nonce_is_rejected() {
let store = SqliteExecutionNonceStore::open_in_memory().unwrap();
assert!(store.try_reserve("a", 1_000, 1_100).unwrap());
assert!(!store.try_reserve("a", 1_001, 1_100).unwrap());
}
#[test]
fn padded_nonce_id_is_rejected() {
let store = SqliteExecutionNonceStore::open_in_memory().unwrap();
let error = store.try_reserve(" nonce", 1_000, 1_100).unwrap_err();
assert!(
error.to_string().contains("nonce_id"),
"expected nonce_id validation error, got {error}"
);
}
#[test]
fn expired_row_is_pruned_and_slot_reusable() {
let store = SqliteExecutionNonceStore::open_in_memory().unwrap();
assert!(store.try_reserve("a", 1_000, 1_030).unwrap());
assert!(store.try_reserve("a", 2_000, 2_030).unwrap());
}
#[test]
fn persists_across_reopen() {
let path = unique_db_path("chio-exec-nonce");
let now = now_secs();
let expires_at = now.saturating_add(120);
{
let store = SqliteExecutionNonceStore::open(&path).unwrap();
assert!(store
.try_reserve("persistent-nonce", now, expires_at)
.unwrap());
assert!(store.is_consumed("persistent-nonce").unwrap());
}
let reopened = SqliteExecutionNonceStore::open(&path).unwrap();
assert!(reopened.is_consumed("persistent-nonce").unwrap());
assert!(!reopened
.try_reserve("persistent-nonce", now, expires_at)
.unwrap());
let _ = fs::remove_file(path);
}
#[test]
fn consumed_lookup_ignores_expired_rows() {
let store = SqliteExecutionNonceStore::open_in_memory().unwrap();
assert!(store.try_reserve("expired-nonce", 1_000, 1_100).unwrap());
assert!(!store.is_consumed("expired-nonce").unwrap());
}
}