use std::sync::LazyLock;
use tokio::sync::Mutex;
use super::KeyspaceHandle;
use crate::error::AppError;
static COUNTER_LOCK: LazyLock<Mutex<()>> = LazyLock::new(|| Mutex::new(()));
pub async fn allocate_u32(ks: &KeyspaceHandle, counter_key: &str) -> Result<u32, AppError> {
allocate_u32_block(ks, counter_key, 1, None).await
}
pub async fn allocate_u32_block(
ks: &KeyspaceHandle,
counter_key: &str,
count: u32,
expected: Option<u32>,
) -> Result<u32, AppError> {
let _guard = COUNTER_LOCK.lock().await;
let current = read_u32(ks, counter_key).await?;
if let Some(expected) = expected
&& current != expected
{
return Err(AppError::Conflict(format!(
"derivation counter at `{counter_key}` moved from {expected} to {current} while a \
decision was pending — the keys this would derive are not the keys that were approved"
)));
}
if count == 0 {
return Ok(current);
}
ks.insert_raw(counter_key, (current + count).to_le_bytes().to_vec())
.await?;
ks.persist().await?;
crate::integrity::reseal_if_active().await?;
Ok(current)
}
pub async fn peek_u32(ks: &KeyspaceHandle, counter_key: &str) -> Result<u32, AppError> {
let _guard = COUNTER_LOCK.lock().await;
read_u32(ks, counter_key).await
}
async fn read_u32(ks: &KeyspaceHandle, counter_key: &str) -> Result<u32, AppError> {
match ks.get_raw(counter_key).await? {
Some(bytes) => {
let arr: [u8; 4] = bytes
.try_into()
.map_err(|_| AppError::Internal(format!("corrupt counter at {counter_key}")))?;
Ok(u32::from_le_bytes(arr))
}
None => Ok(0),
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::config::StoreConfig;
use crate::store::Store;
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn allocate_u32_is_collision_free_under_concurrency() {
let dir = tempfile::tempdir().expect("tempdir");
let store = Store::open(&StoreConfig {
data_dir: dir.path().to_path_buf(),
})
.expect("open store");
let ks = store.keyspace("test").expect("keyspace");
let n = 64usize;
let mut handles = Vec::with_capacity(n);
for _ in 0..n {
let ks = ks.clone();
handles.push(tokio::spawn(async move {
allocate_u32(&ks, "counter:x").await.expect("alloc")
}));
}
let mut seen = std::collections::HashSet::with_capacity(n);
for h in handles {
let v = h.await.expect("join");
assert!(seen.insert(v), "duplicate counter value {v}");
}
assert_eq!(seen.len(), n);
}
#[tokio::test]
async fn allocate_u32_starts_at_zero_and_is_sequential() {
let dir = tempfile::tempdir().expect("tempdir");
let store = Store::open(&StoreConfig {
data_dir: dir.path().to_path_buf(),
})
.expect("open store");
let ks = store.keyspace("test").expect("keyspace");
for expect in 0..3u32 {
assert_eq!(allocate_u32(&ks, "counter:y").await.unwrap(), expect);
}
}
}
#[cfg(test)]
mod block_tests {
use super::*;
use crate::config::StoreConfig;
use crate::store::Store;
async fn ks() -> (crate::store::KeyspaceHandle, tempfile::TempDir) {
let dir = tempfile::tempdir().unwrap();
let store = Store::open(&StoreConfig {
data_dir: dir.path().to_path_buf(),
})
.unwrap();
(store.keyspace("keys").unwrap(), dir)
}
#[tokio::test]
async fn a_block_is_contiguous() {
let (ks, _d) = ks().await;
assert_eq!(allocate_u32_block(&ks, "c", 3, None).await.unwrap(), 0);
assert_eq!(allocate_u32(&ks, "c").await.unwrap(), 3);
assert_eq!(allocate_u32_block(&ks, "c", 2, None).await.unwrap(), 4);
}
#[tokio::test]
async fn an_expectation_that_no_longer_holds_is_refused() {
let (ks, _d) = ks().await;
assert_eq!(peek_u32(&ks, "c").await.unwrap(), 0);
assert_eq!(allocate_u32(&ks, "c").await.unwrap(), 0);
let err = allocate_u32_block(&ks, "c", 2, Some(0)).await.unwrap_err();
assert!(
matches!(err, AppError::Conflict(_)),
"a moved counter must be a Conflict, got: {err:?}"
);
assert_eq!(peek_u32(&ks, "c").await.unwrap(), 1);
assert_eq!(allocate_u32_block(&ks, "c", 2, Some(1)).await.unwrap(), 1);
}
#[tokio::test]
async fn a_matching_expectation_allocates() {
let (ks, _d) = ks().await;
assert_eq!(allocate_u32_block(&ks, "c", 2, Some(0)).await.unwrap(), 0);
}
}