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