#![allow(dead_code)]
use sha2::{Digest as _, Sha256};
use crate::Error;
pub(crate) fn slice_digest(bytes: &[u8]) -> [u8; 32] {
Sha256::digest(bytes).into()
}
pub(crate) trait ReadCursor: Send {
fn len(&self) -> u64;
async fn read_at(&self, offset: u64, len: u64) -> Result<Vec<u8>, Error>;
async fn read_range(&self, offset: u64, len: u64) -> Result<(Vec<u8>, [u8; 32]), Error> {
let bytes = self.read_at(offset, len).await?;
let digest = slice_digest(&bytes);
Ok((bytes, digest))
}
}
pub(crate) trait StagingCursor: ReadCursor {
async fn append(&mut self, chunk: &[u8]) -> Result<(), Error>;
async fn commit(self, key: &[u8]) -> Result<(), Error>;
async fn discard(self);
}
pub(crate) trait Backend: Send + Sync {
type Cursor: ReadCursor;
type Staging: StagingCursor;
fn name(&self) -> &'static str;
async fn has(&self, key: &[u8]) -> Result<bool, Error>;
async fn get(&self, key: &[u8]) -> Result<Option<Self::Cursor>, Error>;
async fn put(&self, key: &[u8], value: &[u8]) -> Result<(), Error>;
async fn put_staged(&self) -> Result<Self::Staging, Error>;
async fn delete(&self, key: &[u8]) -> Result<(), Error>;
async fn list(&self) -> Result<Vec<Vec<u8>>, Error>;
async fn size(&self, key: &[u8]) -> Result<Option<u64>, Error>;
}
pub(crate) fn clamped(offset: u64, len: u64, total: usize) -> (usize, usize) {
let total = total as u64;
let start = offset.min(total);
let end = start.saturating_add(len).min(total);
(start as usize, end as usize)
}
#[cfg(test)]
mod tests {
use super::*;
use std::collections::BTreeMap;
use std::sync::Arc;
use parking_lot::Mutex;
const CONTENT: &[u8] = b"the quick brown fox jumps over the lazy dog";
fn body_slice(whole: &[u8], offset: u64, len: u64) -> Vec<u8> {
let (start, end) = clamped(offset, len, whole.len());
whole[start..end].to_vec()
}
struct MaterializedCursor {
value: Vec<u8>,
}
impl ReadCursor for MaterializedCursor {
fn len(&self) -> u64 {
self.value.len() as u64
}
async fn read_at(&self, offset: u64, len: u64) -> Result<Vec<u8>, Error> {
Ok(body_slice(&self.value, offset, len))
}
}
struct StreamingCursor {
whole: Vec<u8>,
segment: usize,
}
impl ReadCursor for StreamingCursor {
fn len(&self) -> u64 {
self.whole.len() as u64
}
async fn read_at(&self, offset: u64, len: u64) -> Result<Vec<u8>, Error> {
let (start, end) = clamped(offset, len, self.whole.len());
let mut out = Vec::with_capacity(end - start);
let mut pos = start;
while pos < end {
let next = (((pos / self.segment) + 1) * self.segment).min(end);
out.extend_from_slice(&self.whole[pos..next]);
pos = next;
}
Ok(out)
}
}
async fn assert_cursor_contract<C: ReadCursor>(cursor: &C, whole: &[u8]) {
assert_eq!(
cursor.len() as usize,
whole.len(),
"len is the entry length, metadata-only"
);
let len = whole.len() as u64;
assert_eq!(cursor.read_at(0, len).await.unwrap(), whole, "full read");
assert_eq!(
cursor.read_at(0, len + 17).await.unwrap(),
whole,
"over-read clamps at EOF, pread-style"
);
let mid = (whole.len() / 2) as u64;
assert_eq!(
cursor.read_at(mid, len).await.unwrap(),
&whole[mid as usize..],
"tail read clamps"
);
assert_eq!(
cursor.read_at(len, 5).await.unwrap(),
Vec::<u8>::new(),
"read at EOF is empty, never an error"
);
assert_eq!(
cursor.read_at(u64::MAX, 5).await.unwrap(),
Vec::<u8>::new(),
"an overflowing u64 offset clamps to EOF in u64 arithmetic — never wraps"
);
assert_eq!(
cursor.read_at(0, u64::MAX).await.unwrap(),
whole,
"an overflowing u64 length clamps to the whole body"
);
assert_eq!(
cursor.read_at(0, 0).await.unwrap(),
Vec::<u8>::new(),
"zero-length read"
);
let (start, end) = clamped(mid, 7.min(len.max(1)), whole.len());
let (slice, digest) = cursor.read_range(mid, 7.min(len.max(1))).await.unwrap();
assert_eq!(
slice,
&whole[start..end],
"read_range slice == read_at bytes"
);
let expected: [u8; 32] = Sha256::digest(&slice).into();
assert_eq!(
digest, expected,
"slice digest is SHA-256 over exactly the returned bytes"
);
let (full, full_digest) = cursor.read_range(0, len).await.unwrap();
assert_eq!(full, whole);
assert_eq!(full_digest, slice_digest(whole));
}
#[tokio::test]
async fn cursor_contract_materialized_body() {
assert_cursor_contract(
&MaterializedCursor {
value: CONTENT.to_vec(),
},
CONTENT,
)
.await;
}
#[tokio::test]
async fn cursor_contract_streaming_body() {
assert_cursor_contract(
&StreamingCursor {
whole: CONTENT.to_vec(),
segment: 8,
},
CONTENT,
)
.await;
}
#[tokio::test]
async fn cursor_contract_empty_entry() {
assert_cursor_contract(&MaterializedCursor { value: Vec::new() }, b"").await;
assert_cursor_contract(
&StreamingCursor {
whole: Vec::new(),
segment: 8,
},
b"",
)
.await;
}
#[tokio::test]
async fn both_body_families_serve_the_range_seam_identically() {
let materialized = MaterializedCursor {
value: CONTENT.to_vec(),
};
let streaming = StreamingCursor {
whole: CONTENT.to_vec(),
segment: 9,
};
for (offset, len) in [(0u64, 0u64), (0, 5), (11, 9), (43, 4), (43, 400)] {
assert_eq!(
materialized.read_range(offset, len).await.unwrap(),
streaming.read_range(offset, len).await.unwrap(),
"range {offset}+{len} identical across body families"
);
}
}
#[test]
fn slice_digest_convention_matches_published_vectors() {
assert_eq!(
hex::encode(slice_digest(b"abc")),
"ba7816bf8f01cfea414140de5dae2223b00361a396177a9cb410ff61f20015ad",
"sha256 'abc' (NIST)"
);
assert_eq!(
hex::encode(slice_digest(b"")),
"e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855",
"sha256 empty (NIST)"
);
}
#[derive(Clone, Default)]
struct TestBackend {
rows: Arc<Mutex<BTreeMap<Vec<u8>, Vec<u8>>>>,
fail_commit: bool,
}
impl TestBackend {
fn row(&self, key: &[u8]) -> Option<Vec<u8>> {
self.rows.lock().get(key).cloned()
}
}
struct TestCursor {
value: Vec<u8>,
}
impl ReadCursor for TestCursor {
fn len(&self) -> u64 {
self.value.len() as u64
}
async fn read_at(&self, offset: u64, len: u64) -> Result<Vec<u8>, Error> {
Ok(body_slice(&self.value, offset, len))
}
}
struct TestStaging {
staged: Vec<u8>,
rows: Arc<Mutex<BTreeMap<Vec<u8>, Vec<u8>>>>,
fail_commit: bool,
}
impl ReadCursor for TestStaging {
fn len(&self) -> u64 {
self.staged.len() as u64
}
async fn read_at(&self, offset: u64, len: u64) -> Result<Vec<u8>, Error> {
Ok(body_slice(&self.staged, offset, len))
}
}
impl StagingCursor for TestStaging {
async fn append(&mut self, chunk: &[u8]) -> Result<(), Error> {
self.staged.extend_from_slice(chunk);
Ok(())
}
async fn commit(mut self, key: &[u8]) -> Result<(), Error> {
if self.fail_commit {
self.staged.clear();
return Err(Error::Io("engine commit failure (test double)".to_owned()));
}
let value = std::mem::take(&mut self.staged);
self.rows.lock().entry(key.to_vec()).or_insert(value);
Ok(())
}
async fn discard(mut self) {
self.staged.clear();
}
}
impl Backend for TestBackend {
type Cursor = TestCursor;
type Staging = TestStaging;
fn name(&self) -> &'static str {
"test-double"
}
async fn has(&self, key: &[u8]) -> Result<bool, Error> {
Ok(self.rows.lock().contains_key(key))
}
async fn get(&self, key: &[u8]) -> Result<Option<Self::Cursor>, Error> {
Ok(self
.rows
.lock()
.get(key)
.cloned()
.map(|value| TestCursor { value }))
}
async fn put(&self, key: &[u8], value: &[u8]) -> Result<(), Error> {
self.rows
.lock()
.entry(key.to_vec())
.or_insert_with(|| value.to_vec());
Ok(())
}
async fn put_staged(&self) -> Result<Self::Staging, Error> {
Ok(TestStaging {
staged: Vec::new(),
rows: self.rows.clone(),
fail_commit: self.fail_commit,
})
}
async fn delete(&self, key: &[u8]) -> Result<(), Error> {
self.rows
.lock()
.remove(key)
.map(|_| ())
.ok_or(Error::Missing)
}
async fn list(&self) -> Result<Vec<Vec<u8>>, Error> {
Ok(self.rows.lock().keys().cloned().collect())
}
async fn size(&self, key: &[u8]) -> Result<Option<u64>, Error> {
Ok(self.rows.lock().get(key).map(|value| value.len() as u64))
}
}
#[tokio::test]
async fn backend_method_contract() {
let backend = TestBackend::default();
let key = b"\x01opaque-key-bytes".to_vec();
assert_eq!(backend.name(), "test-double", "reporting identity");
assert!(
!backend.has(&key).await.unwrap(),
"virgin store: has is a no-op read"
);
assert!(
backend.get(&key).await.unwrap().is_none(),
"virgin store: get is a no-op read"
);
assert_eq!(
backend.size(&key).await.unwrap(),
None,
"virgin store: size is a no-op read"
);
assert!(
backend.list().await.unwrap().is_empty(),
"virgin store: list is empty"
);
assert_eq!(
backend.delete(&key).await.unwrap_err(),
Error::Missing,
"virgin store: delete reports Missing"
);
backend.put(&key, CONTENT).await.unwrap();
assert!(backend.has(&key).await.unwrap(), "put lands");
assert_eq!(
backend.size(&key).await.unwrap(),
Some(CONTENT.len() as u64),
"size is metadata-only and exact"
);
backend.put(&key, b"overwritten").await.unwrap();
assert_eq!(
backend.row(&key).as_deref(),
Some(CONTENT),
"CAS first-writer wins: re-put is a no-op"
);
assert_eq!(
backend.list().await.unwrap(),
vec![key.clone()],
"list is complete, keys not values"
);
backend.delete(&key).await.unwrap();
assert!(!backend.has(&key).await.unwrap(), "delete lands");
assert_eq!(
backend.delete(&key).await.unwrap_err(),
Error::Missing,
"delete reports Missing when absent"
);
}
#[tokio::test]
async fn staged_put_round_trips_through_the_read_seam() {
let backend = TestBackend::default();
let key = b"\x01staged-key".to_vec();
let mut staging = backend.put_staged().await.unwrap();
staging.append(&CONTENT[..8]).await.unwrap();
staging.append(&CONTENT[8..16]).await.unwrap();
staging.append(&CONTENT[16..]).await.unwrap();
assert_eq!(
staging.len() as usize,
CONTENT.len(),
"staged length is the appended length"
);
let (bytes, digest) = staging
.read_range(0, staging.len())
.await
.expect("staged bytes readable");
assert_eq!(bytes, CONTENT, "staged artifact readable as a whole");
assert_eq!(digest, slice_digest(CONTENT));
staging.commit(&key).await.expect("commit publishes");
assert!(backend.has(&key).await.unwrap(), "entry published");
let cursor = backend
.get(&key)
.await
.expect("get succeeds")
.expect("entry present");
let (bytes, digest) = cursor
.read_range(0, cursor.len())
.await
.expect("committed read");
assert_eq!(bytes, CONTENT);
assert_eq!(digest, slice_digest(CONTENT));
backend.put(&key, b"overwritten").await.unwrap();
assert_eq!(
backend.row(&key).as_deref(),
Some(CONTENT),
"staged commit also lands CAS-first-writer"
);
}
#[tokio::test]
async fn staged_discard_leaves_zero_residue() {
let backend = TestBackend::default();
let key = b"\x01discard-key".to_vec();
let mut staging = backend.put_staged().await.unwrap();
staging.append(b"residue").await.unwrap();
staging.discard().await;
assert!(
!backend.has(&key).await.unwrap(),
"discarded stage published nothing"
);
assert_eq!(backend.size(&key).await.unwrap(), None, "no entry landed");
assert!(
backend.list().await.unwrap().is_empty(),
"the pool holds exactly zero entries after discard"
);
}
#[tokio::test]
async fn commit_failure_self_discards() {
let backend = TestBackend {
fail_commit: true,
..Default::default()
};
let key = b"\x01failure-key".to_vec();
let mut staging = backend.put_staged().await.unwrap();
staging.append(CONTENT).await.unwrap();
let err = staging.commit(&key).await.unwrap_err();
assert!(matches!(err, Error::Io(_)), "engine media failure");
assert!(
!backend.has(&key).await.unwrap(),
"failed commit published nothing"
);
assert_eq!(backend.size(&key).await.unwrap(), None, "zero residue");
}
}