coven_database/store/
local_blob_cleanup_intents.rs

1//! Durable deletion of obsolete logical-source and exact-locator copies.
2
3use crate::DbError;
4
5pub fn intents_from_changes(
6    blob_decls: &crate::BlobDecls,
7    old_changes: &[coven_foundation::changeset::RowChange],
8    new_changes: &[coven_foundation::changeset::RowChange],
9) -> Result<Vec<LocalBlobCleanupIntent>, crate::BlobDeclError> {
10    if old_changes.len() != new_changes.len() {
11        return Err(crate::BlobDeclError::ChangesetWalkMismatch {
12            old_count: old_changes.len(),
13            new_count: new_changes.len(),
14        });
15    }
16    let mut intents = Vec::new();
17    for (old, new) in old_changes.iter().zip(new_changes) {
18        let old_blob_to_drop = match old.op {
19            coven_foundation::changeset::ChangeOp::Delete => blob_decls.ref_from_change(old)?,
20            coven_foundation::changeset::ChangeOp::Update => {
21                let Some(old_blob) = blob_decls.ref_from_change(old)? else {
22                    continue;
23                };
24                let should_drop = match blob_decls.ref_from_change(new)? {
25                    Some(new_blob) => {
26                        old_blob.namespace != new_blob.namespace || old_blob.id != new_blob.id
27                    }
28                    None => true,
29                };
30                should_drop.then_some(old_blob)
31            }
32            coven_foundation::changeset::ChangeOp::Insert => None,
33        };
34        if let Some(blob) = old_blob_to_drop {
35            let row_id =
36                old.pk()
37                    .ok_or_else(|| crate::BlobDeclError::MissingPublicationPrimaryKey {
38                        table: old.table.clone(),
39                    })?;
40            intents.push(LocalBlobCleanupIntent::for_row(
41                blob.namespace,
42                blob.id,
43                old.table.clone(),
44                row_id,
45            ));
46        }
47    }
48    Ok(intents)
49}
50use coven_foundation::store_dir::StoreDir;
51
52/// A transaction-local request that resolves into committed copy-specific cleanup
53/// obligations.
54#[derive(Clone, Debug, PartialEq, Eq)]
55pub struct LocalBlobCleanupIntent {
56    namespace: String,
57    blob_id: String,
58    identity: LocalBlobCleanupIdentity,
59}
60
61#[derive(Clone, Debug, PartialEq, Eq)]
62pub enum LocalBlobCleanupIdentity {
63    Local,
64    Row { table: String, row_id: String },
65    Exact(coven_protocol::store_commit::ObjectHash),
66}
67
68impl LocalBlobCleanupIntent {
69    pub fn local(namespace: impl Into<String>, blob_id: impl Into<String>) -> Self {
70        Self {
71            namespace: namespace.into(),
72            blob_id: blob_id.into(),
73            identity: LocalBlobCleanupIdentity::Local,
74        }
75    }
76
77    pub fn for_row(
78        namespace: impl Into<String>,
79        blob_id: impl Into<String>,
80        table: impl Into<String>,
81        row_id: impl Into<String>,
82    ) -> Self {
83        Self {
84            namespace: namespace.into(),
85            blob_id: blob_id.into(),
86            identity: LocalBlobCleanupIdentity::Row {
87                table: table.into(),
88                row_id: row_id.into(),
89            },
90        }
91    }
92
93    pub fn exact(
94        namespace: impl Into<String>,
95        blob_id: impl Into<String>,
96        locator_hash: coven_protocol::store_commit::ObjectHash,
97    ) -> Self {
98        Self {
99            namespace: namespace.into(),
100            blob_id: blob_id.into(),
101            identity: LocalBlobCleanupIdentity::Exact(locator_hash),
102        }
103    }
104
105    pub fn persisted_identity(&self) -> Result<String, DbError> {
106        match &self.identity {
107            LocalBlobCleanupIdentity::Local => Ok("local".to_string()),
108            LocalBlobCleanupIdentity::Exact(locator_hash) => Ok(locator_hash.to_string()),
109            LocalBlobCleanupIdentity::Row { .. } => Err(DbError::Message(
110                "row-bound local cleanup identity is not durable".to_string(),
111            )),
112        }
113    }
114
115    pub fn from_persisted(
116        namespace: String,
117        blob_id: String,
118        identity: String,
119    ) -> Result<Self, coven_foundation::object_hash::InvalidObjectHash> {
120        if identity == "local" {
121            return Ok(Self::local(namespace, blob_id));
122        }
123        let locator_hash = identity.parse()?;
124        Ok(Self::exact(namespace, blob_id, locator_hash))
125    }
126
127    pub fn namespace(&self) -> &str {
128        &self.namespace
129    }
130
131    pub fn blob_id(&self) -> &str {
132        &self.blob_id
133    }
134
135    pub fn identity(&self) -> &LocalBlobCleanupIdentity {
136        &self.identity
137    }
138
139    pub async fn apply(&self, store_dir: &StoreDir) -> Result<(), DbError> {
140        match &self.identity {
141            LocalBlobCleanupIdentity::Local => store_dir
142                .remove_local_blob(self.namespace(), self.blob_id())
143                .await
144                .map(|_| ())
145                .map_err(DbError::from),
146            LocalBlobCleanupIdentity::Exact(locator_hash) => store_dir
147                .remove_cached_locator(self.namespace(), *locator_hash)
148                .await
149                .map_err(DbError::from),
150            LocalBlobCleanupIdentity::Row { .. } => Err(DbError::Message(
151                "persisted local cleanup intent is row-bound".to_string(),
152            )),
153        }
154    }
155}
coven_database/store/local_blob_cleanup_intents.rs

coven_database/store/
local_blob_cleanup_intents.rs
