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