harmos_storage/
deletion.rs1use std::sync::Arc;
18
19use sha2::{Digest, Sha256};
21
22use super::config::MAX_MUTATIONS;
24use super::manager::{Inner, State, Target};
25use super::receipts::ReceiptState;
26use super::{MutationId, MutationReceipt, MutationValue, StagedFile, StorageManager};
27use crate::{Error, Identifier, MemorySize, Result};
28
29pub(super) struct Deletion {
38 pub(super) work: Arc<DeleteWork>,
39 pub(super) bytes: usize,
40 failure: Option<Arc<Error>>,
41}
42
43pub(super) struct DeleteWork {
44 target: Target,
45 pub(super) file: StagedFile,
46 pub(super) receipt: MutationReceipt,
47}
48
49impl<C> StorageManager<C> {
56 pub async fn delete_file_once(&self, mutation: &MutationId, file: &StagedFile) -> Result<StagedFile> {
57 self.inner.primary.history().check(&file.history)?;
58 if !self.inner.primary.supports_mutation_receipts() || !self.inner.primary.supports_file_deletion() {
59 return Err(invalid("backend must support durable deletion"));
60 }
61 let fingerprint = fingerprint(file);
62 let mut retry = true;
63 let mut completion = None;
64 let mut changed = self.inner.changed.subscribe();
65 loop {
66 let mut notify = false;
67 {
68 let mut state = self.inner.state.lock().unwrap();
69 if state.status.closing {
70 return Err(Error::Closing { domain: "file deletion" });
71 }
72 if !state.ready || !state.status.running {
73 return Err(Error::NotReady { domain: "file deletion" });
74 }
75 match state.receipts.lookup(mutation, &fingerprint, completion.as_ref())? {
76 ReceiptState::Complete(value) => return value.deleted(),
77 ReceiptState::Pending => {
78 if completion.is_none() {
79 let owner = state.receipts.completion(mutation);
80 completion = Some(owner);
81 }
82 let deletion = state.deletions.get(mutation).ok_or_else(|| invalid("pending identity owns different work"))?;
83 if let Some(source) = deletion.failure.clone() {
84 if !retry {
85 return Err(Error::Persistence { source });
86 }
87 if let Some(other) = &state.status.failure
88 && !Arc::ptr_eq(other, &source)
89 {
90 return Err(Error::Persistence { source: other.clone() });
91 }
92 state.deletions.get_mut(mutation).unwrap().failure = None;
93 state.status.failure = None;
94 state.retry = state.retry.wrapping_add(1);
95 notify = true;
96 } else if let Some(source) = &state.status.failure {
97 return Err(Error::Persistence { source: source.clone() });
98 }
99 retry = false;
100 }
101 ReceiptState::Missing => {
102 if let Some(source) = &state.status.failure {
103 return Err(Error::Persistence { source: source.clone() });
104 }
105 if state.files.values().any(|active| active.identifies(file)) {
106 tracing::error!(file = %file.id, reason = "active producer", "file deletion rejected");
107 return Err(invalid("active production must finish or abandon before deletion"));
108 }
109 if state.deleting(&file.id) {
110 return Err(invalid("another deletion owns this file"));
111 }
112 if state.deletions.len() >= MAX_MUTATIONS.get() {
113 return Err(exhausted());
114 }
115 let bytes = crate::btree_map_memory::<MutationId, Deletion>(1)
116 .saturating_add(std::mem::size_of::<DeleteWork>())
117 .saturating_add(std::mem::size_of::<MutationReceipt>())
118 .saturating_add(std::mem::size_of::<MutationValue>())
119 .saturating_add(mutation.memory_size().saturating_mul(2))
120 .saturating_add(file.memory_size().saturating_mul(2))
121 .saturating_add(2 * std::mem::size_of::<usize>());
122 state.collect();
123 let capacity = self.inner.settings.budget.read_memory_bytes.get();
124 if bytes > capacity || state.status.read_bytes > capacity - bytes {
125 return Err(exhausted());
126 }
127 let outcome = MutationValue::Deleted(file.clone());
128 state.receipts.insert(mutation, fingerprint, outcome.memory_size())?;
129 let owner = state.receipts.completion(mutation);
130 completion = Some(owner);
131 let receipt = state.receipts.prepared(mutation, outcome);
132 let target = state.data_accepted;
133 state.deletions.insert(
134 mutation.clone(),
135 Deletion { work: Arc::new(DeleteWork { target, file: file.clone(), receipt }), bytes, failure: None },
136 );
137 retry = false;
138 notify = true;
139 }
140 }
141 state.collect();
142 }
143 if notify {
144 self.inner.notify();
145 }
146 changed.changed().await.map_err(|_| Error::Closing { domain: "file deletion" })?;
147 }
148 }
149
150 pub(super) async fn settle_deletions(&self) -> Result<()> {
151 let mut changed = self.inner.changed.subscribe();
152 {
153 let mut state = self.inner.state.lock().unwrap();
154 if state.deletions.is_empty() {
155 return Ok(());
156 }
157 for deletion in state.deletions.values_mut() {
158 deletion.failure = None;
159 }
160 state.status.failure = None;
161 state.retry = state.retry.wrapping_add(1);
162 }
163 self.inner.notify();
164 loop {
165 {
166 let state = self.inner.state.lock().unwrap();
167 if state.deletions.is_empty() {
168 return Ok(());
169 }
170 if let Some(source) = &state.status.failure {
171 return Err(Error::Persistence { source: source.clone() });
172 }
173 if !state.status.running {
174 return Err(Error::NotReady { domain: "file deletion worker" });
175 }
176 }
177 changed.changed().await.map_err(|_| Error::Closing { domain: "file deletion" })?;
178 }
179 }
180}
181
182impl State {
183 pub(super) fn deleting(&self, file: &Identifier) -> bool {
184 self.deletions.values().any(|deletion| deletion.work.file.id == *file)
185 }
186
187 pub(super) fn request_deletions(&mut self) {
188 for deletion in self.deletions.values().filter(|deletion| deletion.failure.is_none()) {
189 let target = deletion.work.target;
190 if target.index > self.data_flushed && self.data_request.is_none_or(|current| target.index < current.index) {
191 self.data_request = Some(target);
192 }
193 }
194 }
195
196 pub(super) fn deletion_action(&self) -> Option<Arc<DeleteWork>> {
197 self.deletions
198 .values()
199 .find(|deletion| deletion.failure.is_none() && deletion.work.target.index <= self.data_flushed)
200 .map(|deletion| deletion.work.clone())
201 }
202
203 pub(super) fn deleting_data(&self, index: u64) -> bool {
204 self.deletions.values().any(|deletion| deletion.failure.is_none() && deletion.work.target.index >= index)
205 }
206
207 pub(super) fn deletion_failed(&mut self, mutation: &MutationId, source: Arc<Error>) {
208 if let Some(deletion) = self.deletions.get_mut(mutation) {
209 tracing::error!(file = %deletion.work.file.id, window = %mutation.window, mutation = %mutation.key, error = %source, reason = "file deletion failed", "file deletion retained for retry");
210 deletion.failure = Some(source);
211 }
212 }
213
214 pub(super) fn deletion_data_failed(&mut self, index: u64, source: Arc<Error>) {
215 for deletion in self.deletions.values_mut().filter(|deletion| deletion.work.target.index >= index) {
216 deletion.failure = Some(source.clone());
217 }
218 }
219}
220
221impl DeleteWork {
222 pub(super) async fn perform(&self, inner: &Arc<Inner>) -> Result<()> {
223 inner.bounded(inner.primary.delete_file(&self.file)).await?;
224 inner.bounded(inner.primary.save_mutations(std::slice::from_ref(&self.receipt))).await?;
225 tracing::info!(file = %self.file.id, mutation = %self.receipt.mutation.key, reason = "explicit deletion", "file deleted");
226 Ok(())
227 }
228}
229
230fn fingerprint(file: &StagedFile) -> [u8; 32] {
235 let mut hash = Sha256::new();
236 hash.update(b"storage.delete_file:v1\0");
237 for bytes in [file.history.store.as_str().as_bytes(), file.history.epoch.as_str().as_bytes(), file.id.as_str().as_bytes()] {
238 hash.update((bytes.len() as u64).to_le_bytes());
239 hash.update(bytes);
240 }
241 hash.finalize().into()
242}
243fn invalid(reason: &'static str) -> Error {
244 Error::InvalidInput { field: "file deletion", reason }
245}
246fn exhausted() -> Error {
247 Error::ResourceExhausted { resource: "file deletion requests" }
248}
249
250