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