mod file_store;
pub use file_store::FileAttachmentStore;
use std::collections::{BTreeSet, HashMap};
use std::path::PathBuf;
use std::sync::{Arc, Mutex};
use lash_sansio::{AttachmentCreateMeta, AttachmentId, AttachmentMeta, AttachmentRef};
use sha2::{Digest, Sha256};
use crate::store::{AttachmentIntent, AttachmentManifest, StoreError};
#[derive(Debug, thiserror::Error)]
pub enum AttachmentStoreError {
#[error("attachment `{0}` was not found")]
NotFound(AttachmentId),
#[error("attachment store I/O failed at {path}: {source}")]
Io {
path: PathBuf,
#[source]
source: std::io::Error,
},
#[error("attachment manifest write failed: {0}")]
ManifestRecordFailed(String),
#[error("attachment store backend failed: {0}")]
Backend(String),
}
#[derive(Clone, Debug)]
pub struct StoredAttachment {
pub bytes: Vec<u8>,
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct StoredBlobRef {
pub id: AttachmentId,
pub last_modified_epoch_ms: Option<u64>,
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum AttachmentStorePersistence {
Ephemeral,
Durable,
}
impl AttachmentStorePersistence {
pub fn durability_tier(self) -> crate::DurabilityTier {
match self {
Self::Ephemeral => crate::DurabilityTier::Inline,
Self::Durable => crate::DurabilityTier::Durable,
}
}
}
#[async_trait::async_trait]
pub trait AttachmentStore: Send + Sync {
fn persistence(&self) -> AttachmentStorePersistence {
AttachmentStorePersistence::Ephemeral
}
async fn put(
&self,
bytes: Vec<u8>,
meta: AttachmentCreateMeta,
) -> Result<AttachmentRef, AttachmentStoreError>;
async fn get(&self, id: &AttachmentId) -> Result<StoredAttachment, AttachmentStoreError>;
async fn delete(&self, id: &AttachmentId) -> Result<(), AttachmentStoreError>;
async fn list(&self) -> Result<Vec<StoredBlobRef>, AttachmentStoreError>;
async fn head(&self, id: &AttachmentId) -> Result<Option<StoredBlobRef>, AttachmentStoreError> {
Ok(self.list().await?.into_iter().find(|blob| &blob.id == id))
}
}
#[async_trait::async_trait]
pub trait AttachmentRootSet: Send + Sync {
async fn live_attachment_refs(
&self,
intent_grace_cutoff_epoch_ms: u64,
) -> Result<BTreeSet<AttachmentId>, StoreError>;
async fn has_live_attachment_ref(
&self,
id: &AttachmentId,
intent_grace_cutoff_epoch_ms: u64,
) -> Result<bool, StoreError> {
Ok(self
.live_attachment_refs(intent_grace_cutoff_epoch_ms)
.await?
.contains(id))
}
}
#[async_trait::async_trait]
impl<T: crate::SessionStoreFactory + ?Sized> AttachmentRootSet for T {
async fn live_attachment_refs(
&self,
intent_grace_cutoff_epoch_ms: u64,
) -> Result<BTreeSet<AttachmentId>, StoreError> {
crate::SessionStoreFactory::live_attachment_refs(self, intent_grace_cutoff_epoch_ms).await
}
async fn has_live_attachment_ref(
&self,
id: &AttachmentId,
intent_grace_cutoff_epoch_ms: u64,
) -> Result<bool, StoreError> {
crate::SessionStoreFactory::has_live_attachment_ref(self, id, intent_grace_cutoff_epoch_ms)
.await
}
}
#[derive(Clone, Debug, Default, PartialEq, Eq)]
pub struct AttachmentReclamationReport {
pub scanned_blob_count: usize,
pub reclaimed_count: usize,
pub failed_ids: Vec<AttachmentId>,
pub deleted_while_referenced: Vec<AttachmentId>,
}
pub async fn reclaim_unreferenced_attachments<R>(
root_set: &R,
backend: &dyn AttachmentStore,
grace_period_ms: u64,
) -> Result<AttachmentReclamationReport, AttachmentStoreError>
where
R: AttachmentRootSet + ?Sized,
{
let now = now_epoch_ms();
let intent_grace_cutoff = now.saturating_sub(grace_period_ms);
let live = root_set
.live_attachment_refs(intent_grace_cutoff)
.await
.map_err(|err| {
AttachmentStoreError::Backend(format!(
"failed to enumerate live attachment refs: {err}"
))
})?;
let blobs = backend.list().await?;
let mut report = AttachmentReclamationReport::default();
for blob in blobs {
report.scanned_blob_count += 1;
if live.contains(&blob.id) {
continue;
}
if within_grace(blob.last_modified_epoch_ms, now, grace_period_ms) {
continue;
}
match backend.head(&blob.id).await {
Ok(Some(fresh)) => {
if within_grace(
fresh.last_modified_epoch_ms,
now_epoch_ms(),
grace_period_ms,
) {
continue;
}
}
Ok(None) => continue,
Err(_) => {
report.failed_ids.push(blob.id);
continue;
}
}
match root_set
.has_live_attachment_ref(&blob.id, intent_grace_cutoff)
.await
{
Ok(true) => continue,
Ok(false) => {}
Err(_) => {
report.failed_ids.push(blob.id);
continue;
}
}
match backend.delete(&blob.id).await {
Ok(()) => {
report.reclaimed_count += 1;
if let Ok(true) = root_set
.has_live_attachment_ref(&blob.id, intent_grace_cutoff)
.await
{
tracing::error!(
attachment_id = %blob.id,
"attachment GC deleted a blob that was re-referenced in the \
delete window; bytes are unrecoverable but a subsequent put \
self-heals"
);
report.deleted_while_referenced.push(blob.id);
}
}
Err(_) => report.failed_ids.push(blob.id),
}
}
Ok(report)
}
fn within_grace(last_modified_epoch_ms: Option<u64>, now: u64, grace_period_ms: u64) -> bool {
last_modified_epoch_ms.is_some_and(|modified| now.saturating_sub(modified) < grace_period_ms)
}
struct InMemoryBlob {
stored: StoredAttachment,
stored_at_epoch_ms: u64,
}
#[derive(Default)]
pub struct InMemoryAttachmentStore {
attachments: Mutex<HashMap<AttachmentId, InMemoryBlob>>,
}
impl InMemoryAttachmentStore {
pub fn new() -> Self {
Self::default()
}
}
#[async_trait::async_trait]
impl AttachmentStore for InMemoryAttachmentStore {
async fn put(
&self,
bytes: Vec<u8>,
meta: AttachmentCreateMeta,
) -> Result<AttachmentRef, AttachmentStoreError> {
let meta = stored_meta(&bytes, meta);
let reference = meta.as_ref();
let now = now_epoch_ms();
let mut attachments = self.attachments.lock().expect("attachment store lock");
match attachments.entry(reference.id.clone()) {
std::collections::hash_map::Entry::Occupied(mut existing) => {
existing.get_mut().stored_at_epoch_ms = now;
}
std::collections::hash_map::Entry::Vacant(slot) => {
slot.insert(InMemoryBlob {
stored: StoredAttachment { bytes },
stored_at_epoch_ms: now,
});
}
}
Ok(reference)
}
async fn get(&self, id: &AttachmentId) -> Result<StoredAttachment, AttachmentStoreError> {
self.attachments
.lock()
.expect("attachment store lock")
.get(id)
.map(|blob| blob.stored.clone())
.ok_or_else(|| AttachmentStoreError::NotFound(id.clone()))
}
async fn delete(&self, id: &AttachmentId) -> Result<(), AttachmentStoreError> {
self.attachments
.lock()
.expect("attachment store lock")
.remove(id);
Ok(())
}
async fn list(&self) -> Result<Vec<StoredBlobRef>, AttachmentStoreError> {
Ok(self
.attachments
.lock()
.expect("attachment store lock")
.iter()
.map(|(id, blob)| StoredBlobRef {
id: id.clone(),
last_modified_epoch_ms: Some(blob.stored_at_epoch_ms),
})
.collect())
}
async fn head(&self, id: &AttachmentId) -> Result<Option<StoredBlobRef>, AttachmentStoreError> {
Ok(self
.attachments
.lock()
.expect("attachment store lock")
.get(id)
.map(|blob| StoredBlobRef {
id: id.clone(),
last_modified_epoch_ms: Some(blob.stored_at_epoch_ms),
}))
}
}
pub fn content_id(bytes: &[u8]) -> AttachmentId {
AttachmentId::new(format!("{:x}", Sha256::digest(bytes)))
}
pub struct SessionAttachmentStore {
backend: Arc<dyn AttachmentStore>,
manifest: Arc<dyn AttachmentManifest>,
session_id: String,
pending_manifest_commit_ids: Mutex<BTreeSet<AttachmentId>>,
}
impl SessionAttachmentStore {
pub fn new(
backend: Arc<dyn AttachmentStore>,
manifest: Arc<dyn AttachmentManifest>,
session_id: impl Into<String>,
) -> Self {
Self::new_with_pending(backend, manifest, session_id, std::iter::empty())
}
pub fn new_with_pending(
backend: Arc<dyn AttachmentStore>,
manifest: Arc<dyn AttachmentManifest>,
session_id: impl Into<String>,
pending_manifest_commit_ids: impl IntoIterator<Item = AttachmentId>,
) -> Self {
Self {
backend,
manifest,
session_id: session_id.into(),
pending_manifest_commit_ids: Mutex::new(
pending_manifest_commit_ids.into_iter().collect(),
),
}
}
pub fn ephemeral(backend: Arc<dyn AttachmentStore>) -> Self {
Self::new(backend, Arc::new(NoopAttachmentManifest), String::new())
}
pub fn in_memory() -> Self {
Self::ephemeral(Arc::new(InMemoryAttachmentStore::new()))
}
pub fn backend(&self) -> &Arc<dyn AttachmentStore> {
&self.backend
}
pub fn manifest(&self) -> &Arc<dyn AttachmentManifest> {
&self.manifest
}
pub fn session_id(&self) -> &str {
&self.session_id
}
pub fn persistence(&self) -> AttachmentStorePersistence {
self.backend.persistence()
}
pub fn pending_manifest_commit_ids(&self) -> Vec<AttachmentId> {
self.pending_manifest_commit_ids
.lock()
.expect("attachment manifest commit tracker lock")
.iter()
.cloned()
.collect()
}
pub fn mark_manifest_committed(&self, ids: &[AttachmentId]) {
if ids.is_empty() {
return;
}
let mut pending = self
.pending_manifest_commit_ids
.lock()
.expect("attachment manifest commit tracker lock");
for id in ids {
pending.remove(id);
}
}
pub async fn put(
&self,
bytes: Vec<u8>,
meta: AttachmentCreateMeta,
) -> Result<AttachmentRef, AttachmentStoreError> {
let attachment_id = content_id(&bytes);
let intent = AttachmentIntent {
attachment_id: attachment_id.clone(),
session_id: self.session_id.clone(),
canonical_uri: attachment_uri(&attachment_id),
intent_at_epoch_ms: now_epoch_ms(),
};
self.manifest.record_intent(intent).map_err(|err| {
AttachmentStoreError::ManifestRecordFailed(format!(
"failed to record attachment intent for `{attachment_id}`: {err}"
))
})?;
let reference = self.backend.put(bytes, meta).await?;
if reference.id != attachment_id {
return Err(AttachmentStoreError::Backend(format!(
"attachment store returned id `{}` after manifest intent for `{attachment_id}`",
reference.id
)));
}
self.pending_manifest_commit_ids
.lock()
.expect("attachment manifest commit tracker lock")
.insert(reference.id.clone());
Ok(reference)
}
pub async fn get(&self, id: &AttachmentId) -> Result<StoredAttachment, AttachmentStoreError> {
let holds_ref = self
.manifest
.holds_ref(&self.session_id, id)
.map_err(|err| {
AttachmentStoreError::Backend(format!(
"failed to check attachment manifest for `{id}`: {err}"
))
})?;
if !holds_ref {
return Err(AttachmentStoreError::NotFound(id.clone()));
}
self.backend.get(id).await
}
pub async fn delete(&self, id: &AttachmentId) -> Result<(), AttachmentStoreError> {
self.pending_manifest_commit_ids
.lock()
.expect("attachment manifest commit tracker lock")
.remove(id);
self.manifest.forget(&self.session_id, id).map_err(|err| {
AttachmentStoreError::ManifestRecordFailed(format!(
"failed to forget attachment ref for `{id}`: {err}"
))
})?;
Ok(())
}
}
pub struct NoopAttachmentManifest;
impl AttachmentManifest for NoopAttachmentManifest {
fn record_intent(&self, _intent: AttachmentIntent) -> Result<(), StoreError> {
Ok(())
}
fn commit_refs(
&self,
_session_id: &str,
_attachment_ids: &[AttachmentId],
) -> Result<(), StoreError> {
Ok(())
}
fn list_uncommitted(
&self,
_older_than_epoch_ms: u64,
) -> Result<Vec<crate::AttachmentManifestEntry>, StoreError> {
Ok(Vec::new())
}
fn forget(&self, _session_id: &str, _attachment_id: &AttachmentId) -> Result<(), StoreError> {
Ok(())
}
fn holds_ref(
&self,
_session_id: &str,
_attachment_id: &AttachmentId,
) -> Result<bool, StoreError> {
Ok(true)
}
fn list_all_refs(&self) -> Result<Vec<AttachmentId>, StoreError> {
Ok(Vec::new())
}
}
fn attachment_uri(attachment_id: &AttachmentId) -> String {
format!("lash-attachment://sha256/{attachment_id}")
}
fn now_epoch_ms() -> u64 {
<crate::SystemClock as crate::Clock>::timestamp_ms(&crate::SystemClock)
}
pub(crate) struct PersistenceManifestAdapter(pub Arc<dyn crate::RuntimePersistence>);
impl AttachmentManifest for PersistenceManifestAdapter {
fn record_intent(&self, intent: AttachmentIntent) -> Result<(), crate::StoreError> {
AttachmentManifest::record_intent(&*self.0, intent)
}
fn commit_refs(
&self,
session_id: &str,
attachment_ids: &[AttachmentId],
) -> Result<(), crate::StoreError> {
AttachmentManifest::commit_refs(&*self.0, session_id, attachment_ids)
}
fn list_uncommitted(
&self,
older_than_epoch_ms: u64,
) -> Result<Vec<crate::AttachmentManifestEntry>, crate::StoreError> {
AttachmentManifest::list_uncommitted(&*self.0, older_than_epoch_ms)
}
fn forget(
&self,
session_id: &str,
attachment_id: &AttachmentId,
) -> Result<(), crate::StoreError> {
AttachmentManifest::forget(&*self.0, session_id, attachment_id)
}
fn holds_ref(
&self,
session_id: &str,
attachment_id: &AttachmentId,
) -> Result<bool, crate::StoreError> {
AttachmentManifest::holds_ref(&*self.0, session_id, attachment_id)
}
fn list_all_refs(&self) -> Result<Vec<AttachmentId>, crate::StoreError> {
AttachmentManifest::list_all_refs(&*self.0)
}
}
fn stored_meta(bytes: &[u8], meta: AttachmentCreateMeta) -> AttachmentMeta {
AttachmentMeta::new(
content_id(bytes),
meta.media_type,
bytes.len() as u64,
meta.width,
meta.height,
meta.label,
)
}
pub async fn resolve_llm_request_attachments(
mut request: crate::llm::types::LlmRequest,
store: &SessionAttachmentStore,
) -> Result<crate::llm::types::LlmRequest, AttachmentStoreError> {
for attachment in &mut request.attachments {
let Some(reference) = attachment.reference.as_ref() else {
continue;
};
if !attachment.data.is_empty() {
continue;
}
let stored = store.get(&reference.id).await?;
attachment.mime = reference.canonical_mime().to_string();
attachment.data = stored.bytes;
}
Ok(request)
}
#[cfg(test)]
mod tests {
use super::*;
use lash_sansio::{ImageMediaType, MediaType};
#[derive(Default)]
struct RecordingManifest {
entries: Mutex<HashMap<(String, AttachmentId), crate::AttachmentManifestEntry>>,
}
impl AttachmentManifest for RecordingManifest {
fn record_intent(&self, intent: AttachmentIntent) -> Result<(), crate::StoreError> {
let key = (intent.session_id.clone(), intent.attachment_id.clone());
self.entries
.lock()
.expect("lock entries")
.entry(key)
.or_insert(crate::AttachmentManifestEntry {
attachment_id: intent.attachment_id,
session_id: intent.session_id,
canonical_uri: intent.canonical_uri,
intent_at_epoch_ms: intent.intent_at_epoch_ms,
committed_at_epoch_ms: None,
});
Ok(())
}
fn commit_refs(
&self,
session_id: &str,
attachment_ids: &[AttachmentId],
) -> Result<(), crate::StoreError> {
let mut entries = self.entries.lock().expect("lock entries");
for attachment_id in attachment_ids {
if let Some(entry) =
entries.get_mut(&(session_id.to_string(), attachment_id.clone()))
{
entry.committed_at_epoch_ms.get_or_insert(1);
}
}
Ok(())
}
fn list_uncommitted(
&self,
older_than_epoch_ms: u64,
) -> Result<Vec<crate::AttachmentManifestEntry>, crate::StoreError> {
Ok(self
.entries
.lock()
.expect("lock entries")
.values()
.filter(|entry| {
entry.committed_at_epoch_ms.is_none()
&& entry.intent_at_epoch_ms <= older_than_epoch_ms
})
.cloned()
.collect())
}
fn forget(
&self,
session_id: &str,
attachment_id: &AttachmentId,
) -> Result<(), crate::StoreError> {
self.entries
.lock()
.expect("lock entries")
.remove(&(session_id.to_string(), attachment_id.clone()));
Ok(())
}
fn holds_ref(
&self,
session_id: &str,
attachment_id: &AttachmentId,
) -> Result<bool, crate::StoreError> {
Ok(self
.entries
.lock()
.expect("lock entries")
.contains_key(&(session_id.to_string(), attachment_id.clone())))
}
fn list_all_refs(&self) -> Result<Vec<AttachmentId>, crate::StoreError> {
Ok(self
.entries
.lock()
.expect("lock entries")
.values()
.map(|entry| entry.attachment_id.clone())
.collect())
}
}
struct RecordingRootSet {
manifests: Vec<Arc<RecordingManifest>>,
}
#[async_trait::async_trait]
impl AttachmentRootSet for RecordingRootSet {
async fn live_attachment_refs(
&self,
intent_grace_cutoff_epoch_ms: u64,
) -> Result<BTreeSet<AttachmentId>, crate::StoreError> {
let mut refs = BTreeSet::new();
for manifest in &self.manifests {
for aged in manifest.list_uncommitted(intent_grace_cutoff_epoch_ms)? {
manifest.forget(&aged.session_id, &aged.attachment_id)?;
}
refs.extend(manifest.list_all_refs()?);
}
Ok(refs)
}
}
fn meta() -> AttachmentCreateMeta {
AttachmentCreateMeta::new(
MediaType::Image(ImageMediaType::Png),
Some(1),
Some(1),
Some("pixel".to_string()),
)
}
#[tokio::test]
async fn memory_store_dedupes_by_bytes() {
let store = InMemoryAttachmentStore::new();
let a = store.put(vec![1, 2, 3], meta()).await.expect("put a");
let b = store.put(vec![1, 2, 3], meta()).await.expect("put b");
assert_eq!(a.id, b.id);
assert_eq!(a.byte_len, 3);
assert_eq!(store.get(&a.id).await.expect("get").bytes, vec![1, 2, 3]);
}
#[tokio::test]
async fn memory_store_assigns_identity_and_byte_len_from_bytes() {
let store = InMemoryAttachmentStore::new();
let reference = store.put(vec![4, 5, 6, 7], meta()).await.expect("put");
assert_eq!(reference.id, content_id(&[4, 5, 6, 7]));
assert_eq!(reference.byte_len, 4);
}
#[tokio::test]
async fn memory_store_lists_stored_blobs() {
let store = InMemoryAttachmentStore::new();
let a = store.put(vec![1], meta()).await.expect("put a");
let b = store.put(vec![2], meta()).await.expect("put b");
let listed: BTreeSet<AttachmentId> = store
.list()
.await
.expect("list")
.into_iter()
.map(|blob| blob.id)
.collect();
assert!(listed.contains(&a.id));
assert!(listed.contains(&b.id));
assert_eq!(listed.len(), 2);
}
#[tokio::test]
async fn facade_get_is_gated_by_manifest_ownership() {
let backend: Arc<dyn AttachmentStore> = Arc::new(InMemoryAttachmentStore::new());
let manifest: Arc<dyn AttachmentManifest> = Arc::new(RecordingManifest::default());
let session_a = SessionAttachmentStore::new(backend.clone(), manifest.clone(), "session-a");
let session_b = SessionAttachmentStore::new(backend.clone(), manifest.clone(), "session-b");
let reference = session_a.put(vec![7, 7, 7], meta()).await.expect("put a");
assert_eq!(
session_a.get(&reference.id).await.expect("a reads").bytes,
vec![7, 7, 7]
);
assert!(matches!(
session_b.get(&reference.id).await,
Err(AttachmentStoreError::NotFound(_))
));
}
#[tokio::test]
async fn facade_delete_drops_ref_but_keeps_backend_bytes() {
let backend: Arc<dyn AttachmentStore> = Arc::new(InMemoryAttachmentStore::new());
let manifest: Arc<dyn AttachmentManifest> = Arc::new(RecordingManifest::default());
let session = SessionAttachmentStore::new(backend.clone(), manifest, "session-1");
let reference = session.put(vec![9, 9], meta()).await.expect("put");
session.delete(&reference.id).await.expect("delete ref");
assert!(matches!(
session.get(&reference.id).await,
Err(AttachmentStoreError::NotFound(_))
));
assert_eq!(
backend
.get(&reference.id)
.await
.expect("bytes remain")
.bytes,
vec![9, 9]
);
}
#[tokio::test]
async fn shared_bytes_survive_until_all_refs_released_then_gc_collects() {
let backend: Arc<dyn AttachmentStore> = Arc::new(InMemoryAttachmentStore::new());
let manifest_a = Arc::new(RecordingManifest::default());
let manifest_b = Arc::new(RecordingManifest::default());
let session_a = SessionAttachmentStore::new(
backend.clone(),
manifest_a.clone() as Arc<dyn AttachmentManifest>,
"session-a",
);
let session_b = SessionAttachmentStore::new(
backend.clone(),
manifest_b.clone() as Arc<dyn AttachmentManifest>,
"session-b",
);
let ref_a = session_a.put(vec![5, 5, 5], meta()).await.expect("put a");
let ref_b = session_b.put(vec![5, 5, 5], meta()).await.expect("put b");
assert_eq!(ref_a.id, ref_b.id);
manifest_a
.commit_refs("session-a", std::slice::from_ref(&ref_a.id))
.expect("commit a");
manifest_b
.commit_refs("session-b", std::slice::from_ref(&ref_b.id))
.expect("commit b");
assert_eq!(backend.list().await.expect("list").len(), 1);
let root_set = RecordingRootSet {
manifests: vec![manifest_a.clone(), manifest_b.clone()],
};
session_a.delete(&ref_a.id).await.expect("a releases");
let report = reclaim_unreferenced_attachments(&root_set, &*backend, 0)
.await
.expect("sweep with b holding a ref");
assert_eq!(report.reclaimed_count, 0, "b still references the blob");
assert_eq!(
backend.get(&ref_b.id).await.expect("blob alive").bytes,
vec![5, 5, 5]
);
session_b.delete(&ref_b.id).await.expect("b releases");
let report = reclaim_unreferenced_attachments(&root_set, &*backend, 0)
.await
.expect("sweep with no refs");
assert_eq!(report.reclaimed_count, 1);
assert!(matches!(
backend.get(&ref_b.id).await,
Err(AttachmentStoreError::NotFound(_))
));
}
#[tokio::test]
async fn gc_spares_fresh_in_flight_intents_as_refs() {
let backend: Arc<dyn AttachmentStore> = Arc::new(InMemoryAttachmentStore::new());
let manifest = Arc::new(RecordingManifest::default());
let session = SessionAttachmentStore::new(
backend.clone(),
manifest.clone() as Arc<dyn AttachmentManifest>,
"session-1",
);
let reference = session.put(vec![3, 1, 4], meta()).await.expect("put");
let root_set = RecordingRootSet {
manifests: vec![manifest.clone()],
};
const GRACE_MS: u64 = 60 * 60 * 1000;
let report = reclaim_unreferenced_attachments(&root_set, &*backend, GRACE_MS)
.await
.expect("sweep");
assert_eq!(report.reclaimed_count, 0, "a fresh intent is a live ref");
assert_eq!(
backend.get(&reference.id).await.expect("kept").bytes,
vec![3, 1, 4]
);
}
#[tokio::test]
async fn gc_collects_aged_uncommitted_intent_orphan() {
let backend: Arc<dyn AttachmentStore> = Arc::new(InMemoryAttachmentStore::new());
let manifest = Arc::new(RecordingManifest::default());
let session = SessionAttachmentStore::new(
backend.clone(),
manifest.clone() as Arc<dyn AttachmentManifest>,
"session-1",
);
let orphan = session
.put(vec![9, 9, 9], meta())
.await
.expect("put orphan");
let root_set = RecordingRootSet {
manifests: vec![manifest.clone()],
};
let report = reclaim_unreferenced_attachments(&root_set, &*backend, 0)
.await
.expect("sweep");
assert_eq!(
report.reclaimed_count, 1,
"an aged, never-committed intent is a collectable orphan"
);
assert!(matches!(
backend.get(&orphan.id).await,
Err(AttachmentStoreError::NotFound(_))
));
assert!(manifest.list_all_refs().expect("refs").is_empty());
}
#[tokio::test]
async fn gc_delete_recheck_spares_blob_refreshed_after_snapshot() {
struct StaleSnapshotStore {
id: AttachmentId,
list_mtime: u64,
head_mtime: u64,
deleted: Mutex<bool>,
}
#[async_trait::async_trait]
impl AttachmentStore for StaleSnapshotStore {
async fn put(
&self,
_bytes: Vec<u8>,
_meta: AttachmentCreateMeta,
) -> Result<AttachmentRef, AttachmentStoreError> {
unreachable!("test does not put through this store")
}
async fn get(
&self,
id: &AttachmentId,
) -> Result<StoredAttachment, AttachmentStoreError> {
Err(AttachmentStoreError::NotFound(id.clone()))
}
async fn delete(&self, _id: &AttachmentId) -> Result<(), AttachmentStoreError> {
*self.deleted.lock().expect("lock") = true;
Ok(())
}
async fn list(&self) -> Result<Vec<StoredBlobRef>, AttachmentStoreError> {
Ok(vec![StoredBlobRef {
id: self.id.clone(),
last_modified_epoch_ms: Some(self.list_mtime),
}])
}
async fn head(
&self,
id: &AttachmentId,
) -> Result<Option<StoredBlobRef>, AttachmentStoreError> {
Ok((id == &self.id).then(|| StoredBlobRef {
id: id.clone(),
last_modified_epoch_ms: Some(self.head_mtime),
}))
}
}
let now = now_epoch_ms();
const GRACE_MS: u64 = 60 * 60 * 1000;
let backend = StaleSnapshotStore {
id: AttachmentId::new("recheck"),
list_mtime: now.saturating_sub(GRACE_MS * 2),
head_mtime: now,
deleted: Mutex::new(false),
};
let root_set = RecordingRootSet { manifests: vec![] };
let report = reclaim_unreferenced_attachments(&root_set, &backend, GRACE_MS)
.await
.expect("sweep");
assert_eq!(
report.reclaimed_count, 0,
"the delete-time re-check must spare a blob refreshed after the snapshot"
);
assert!(
!*backend.deleted.lock().expect("lock"),
"the freshly-refreshed blob must not be deleted"
);
}
struct StaleHeadStore {
id: AttachmentId,
mtime: u64,
deleted: Mutex<bool>,
}
#[async_trait::async_trait]
impl AttachmentStore for StaleHeadStore {
async fn put(
&self,
_bytes: Vec<u8>,
_meta: AttachmentCreateMeta,
) -> Result<AttachmentRef, AttachmentStoreError> {
unreachable!("test does not put through this store")
}
async fn get(&self, id: &AttachmentId) -> Result<StoredAttachment, AttachmentStoreError> {
Err(AttachmentStoreError::NotFound(id.clone()))
}
async fn delete(&self, _id: &AttachmentId) -> Result<(), AttachmentStoreError> {
*self.deleted.lock().expect("lock") = true;
Ok(())
}
async fn list(&self) -> Result<Vec<StoredBlobRef>, AttachmentStoreError> {
Ok(vec![StoredBlobRef {
id: self.id.clone(),
last_modified_epoch_ms: Some(self.mtime),
}])
}
async fn head(
&self,
id: &AttachmentId,
) -> Result<Option<StoredBlobRef>, AttachmentStoreError> {
Ok((id == &self.id).then(|| StoredBlobRef {
id: id.clone(),
last_modified_epoch_ms: Some(self.mtime),
}))
}
}
struct ScriptedRootSet {
answers: Mutex<std::collections::VecDeque<bool>>,
}
#[async_trait::async_trait]
impl AttachmentRootSet for ScriptedRootSet {
async fn live_attachment_refs(
&self,
_intent_grace_cutoff_epoch_ms: u64,
) -> Result<BTreeSet<AttachmentId>, crate::StoreError> {
Ok(BTreeSet::new())
}
async fn has_live_attachment_ref(
&self,
_id: &AttachmentId,
_intent_grace_cutoff_epoch_ms: u64,
) -> Result<bool, crate::StoreError> {
Ok(self
.answers
.lock()
.expect("lock answers")
.pop_front()
.unwrap_or(false))
}
}
#[tokio::test]
async fn gc_pre_delete_root_recheck_spares_reappeared_ref() {
let now = now_epoch_ms();
const GRACE_MS: u64 = 60 * 60 * 1000;
let backend = StaleHeadStore {
id: AttachmentId::new("reappeared"),
mtime: now.saturating_sub(GRACE_MS * 2),
deleted: Mutex::new(false),
};
let root_set = ScriptedRootSet {
answers: Mutex::new([true].into_iter().collect()),
};
let report = reclaim_unreferenced_attachments(&root_set, &backend, GRACE_MS)
.await
.expect("sweep");
assert_eq!(
report.reclaimed_count, 0,
"pre-delete root re-check must spare the blob"
);
assert!(report.deleted_while_referenced.is_empty());
assert!(
!*backend.deleted.lock().expect("lock"),
"a blob re-referenced before delete must not be deleted"
);
}
#[tokio::test]
async fn gc_post_delete_root_recheck_alarms_on_window_ref() {
let now = now_epoch_ms();
const GRACE_MS: u64 = 60 * 60 * 1000;
let id = AttachmentId::new("window-ref");
let backend = StaleHeadStore {
id: id.clone(),
mtime: now.saturating_sub(GRACE_MS * 2),
deleted: Mutex::new(false),
};
let root_set = ScriptedRootSet {
answers: Mutex::new([false, true].into_iter().collect()),
};
let report = reclaim_unreferenced_attachments(&root_set, &backend, GRACE_MS)
.await
.expect("sweep");
assert_eq!(report.reclaimed_count, 1, "the blob is deleted");
assert!(*backend.deleted.lock().expect("lock"), "delete happened");
assert_eq!(
report.deleted_while_referenced,
vec![id],
"a ref appearing in the delete window is recorded for the operator alarm"
);
}
#[tokio::test]
async fn session_facade_tracks_successful_puts_until_commit_mark() {
let manifest: Arc<dyn AttachmentManifest> = Arc::new(RecordingManifest::default());
let store = SessionAttachmentStore::new(
Arc::new(InMemoryAttachmentStore::new()),
manifest,
"session-1",
);
let reference = store.put(vec![8, 9, 10], meta()).await.expect("put");
assert_eq!(
store.pending_manifest_commit_ids(),
vec![reference.id.clone()]
);
store.mark_manifest_committed(&[AttachmentId::new("other")]);
assert_eq!(
store.pending_manifest_commit_ids(),
vec![reference.id.clone()]
);
store.mark_manifest_committed(std::slice::from_ref(&reference.id));
assert!(store.pending_manifest_commit_ids().is_empty());
}
#[tokio::test]
async fn ephemeral_facade_passes_reads_through_without_a_guard() {
let store = SessionAttachmentStore::in_memory();
let reference = store.put(vec![1, 2, 3], meta()).await.expect("put");
assert_eq!(
store.get(&reference.id).await.expect("get").bytes,
vec![1, 2, 3]
);
}
#[test]
fn persistence_manifest_adapter_forwards_holds_ref() {
let runtime: Arc<dyn crate::RuntimePersistence> =
Arc::new(crate::InMemorySessionStore::new());
let adapter = PersistenceManifestAdapter(runtime);
let attachment_id = AttachmentId::new("adapter-forwarding");
let intent = AttachmentIntent {
attachment_id: attachment_id.clone(),
session_id: "adapter-session".to_string(),
canonical_uri: attachment_uri(&attachment_id),
intent_at_epoch_ms: 10,
};
adapter.record_intent(intent).expect("record intent");
assert!(
adapter
.holds_ref("adapter-session", &attachment_id)
.expect("holds ref")
);
assert!(
!adapter
.holds_ref("other-session", &attachment_id)
.expect("no ref for other session")
);
assert_eq!(
adapter.list_all_refs().expect("list all refs"),
vec![attachment_id]
);
}
}