diff --git a/seaweed-volume/src/server/handlers.rs b/seaweed-volume/src/server/handlers.rs index 54eacac554..df5599a970 100644 --- a/seaweed-volume/src/server/handlers.rs +++ b/seaweed-volume/src/server/handlers.rs @@ -3419,10 +3419,14 @@ async fn write_upload_needle( Ok(vol) => vol, Err(refused) => return Ok(Err(refused)), }; + let pending = vol.track_pending_write(&n); if fsync { - vol.write_needle(&mut n, true, fsync).map(Ok) + vol.write_pending_needles(&mut [(n, true)], &[Some(pending)]) + .pop() + .unwrap_or(Err(crate::storage::volume::VolumeError::NotFound)) + .map(Ok) } else { - vol.write_needle_unsynced_or_refused(&mut n) + vol.write_pending_needle_unsynced(&mut n, Some(&pending)) } } @@ -3615,15 +3619,29 @@ pub async fn delete_handler( // Read needle first to validate cookie (matching Go behavior) let original_cookie = cookie; + let is_replicate = serde_urlencoded::from_str::>(&del_query) + .unwrap_or_default() + .into_iter() + .find(|(name, _)| name == "type") + .is_some_and(|(_, value)| value == "replicate"); { let vol = { let store = state.store.read().unwrap(); store.find_volume(vid).map(|(_, vol)| vol) }; - match vol + let mut found = vol + .as_ref() .ok_or(crate::storage::volume::VolumeError::NotFound) - .and_then(|vol| vol.read_needle_with_option(&mut n, &mut delete_read_option())) + .and_then(|vol| vol.read_needle_with_option(&mut n, &mut delete_read_option())); + if found.is_err() + && is_replicate + && let Some(vol) = &vol { + // the write this delete undoes can still be queued here + vol.cancel_pending_writes(needle_id, original_cookie); + found = vol.read_needle_with_option(&mut n, &mut delete_read_option()); + } + match found { Ok(_) => {} Err(_) => { let result = DeleteResult { size: 0 }; @@ -3711,7 +3729,6 @@ pub async fn delete_handler( } }; - let is_replicate = del_query.split('&').any(|p| p == "type=replicate"); if !is_replicate && delete_result.is_ok() { let needs_replication = { let store = state.store.read().unwrap(); @@ -9050,6 +9067,98 @@ mod tests { } } + /// A write queued behind another write's fsync is not indexed yet, so a + /// replicate delete of it finds nothing. The delete cancels the write, so + /// it is taken back when it lands instead of reviving the needle on this + /// copy alone, even when its handler was dropped while it waited in the + /// durable queue. A plain delete that finds nothing is not sent on to the + /// replicas, so it leaves the write alone. + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + #[allow(clippy::await_holding_lock)] + async fn test_replicate_delete_cancels_a_queued_write() { + use crate::server::write_queue::{SyncBatchOption, WriteQueue}; + use crate::storage::volume::VolumeError; + use std::time::{Duration, Instant}; + + for (id, fsync, query, abandon) in [ + (0x6e7a_0c01u64, false, "?type=replicate", false), + (0x6e7a_0c02, true, "?type=replicate", false), + (0x6e7a_0c03, true, "?type=replicate", true), + (0x6e7a_0c04, false, "", false), + (0x6e7a_0c05, false, "?%74ype=replicat%65", false), + ] { + let replicate = !query.is_empty(); + let tmp = tempfile::TempDir::new().unwrap(); + let state = volume_test_state(&tmp); + assert!( + state + .write_queue + .set(WriteQueue::new( + state.clone(), + 128, + SyncBatchOption::default() + )) + .is_ok() + ); + let vol = state + .store + .read() + .unwrap() + .find_volume(VolumeId(1)) + .unwrap() + .1; + let held = vol.sync_lock_for_test(); + + let mut n = Needle { + id: NeedleId(id), + cookie: Cookie(TEST_COOKIE), + data: b"late write".to_vec(), + data_size: 10, + ..Needle::default() + }; + n.checksum = CRC::new(&n.data); + let write = tokio::spawn({ + let state = state.clone(); + async move { write_upload_needle(&state, VolumeId(1), n, fsync).await } + }); + let deadline = Instant::now() + Duration::from_secs(10); + while vol.pending_writes_for_test(NeedleId(id)) == 0 { + assert!(Instant::now() < deadline, "the write never registered"); + tokio::time::sleep(Duration::from_millis(1)).await; + } + + let path = format!("/1,{:x}{:08x}{}", id, TEST_COOKIE, query); + assert_eq!(send_delete(&state, &path).await, StatusCode::NOT_FOUND); + if abandon { + write.abort(); + drop(held); + while vol.pending_writes_for_test(NeedleId(id)) > 0 { + assert!(Instant::now() < deadline, "the queued write never settled"); + tokio::time::sleep(Duration::from_millis(1)).await; + } + } else { + drop(held); + let written = write.await.unwrap(); + if replicate { + assert!(matches!(written, Err(VolumeError::Deleted)), "{id:x}"); + } else { + assert!(matches!(written, Ok(Ok(_))), "{id:x}"); + } + assert_eq!(vol.pending_writes_for_test(NeedleId(id)), 0); + } + + let read = vol.read_needle(&mut Needle { + id: NeedleId(id), + ..Needle::default() + }); + if replicate { + assert!(read.is_err(), "{id:x}: the cancelled write is live"); + } else { + assert!(read.is_ok(), "{id:x}: {read:?}"); + } + } + } + /// Issue #801: an fsync upload takes the per-volume worker so its sync is /// shared across the batch (Go's `writeNeedle2` split); a non-fsync write, /// or an fsync write landing while the server is stopping, takes the diff --git a/seaweed-volume/src/server/write_queue.rs b/seaweed-volume/src/server/write_queue.rs index e141734111..6286bc81c2 100644 --- a/seaweed-volume/src/server/write_queue.rs +++ b/seaweed-volume/src/server/write_queue.rs @@ -18,7 +18,7 @@ use tracing::debug; use crate::storage::needle::needle::Needle; use crate::storage::types::{Size, VolumeId}; -use crate::storage::volume::VolumeError; +use crate::storage::volume::{PendingWrite, VolumeError}; use super::volume_server::VolumeServerState; @@ -26,7 +26,12 @@ use super::volume_server::VolumeServerState; pub type WriteResult = Result<(u64, Size, bool), VolumeError>; /// The needles queued for one volume, each with the durability it asked for. -type VolumeBatch = Vec<(Needle, bool, oneshot::Sender)>; +type VolumeBatch = Vec<( + Needle, + bool, + oneshot::Sender, + Option, +)>; /// A request to write a needle, submitted to the write queue. pub struct WriteRequest { @@ -38,6 +43,8 @@ pub struct WriteRequest { /// that cancels `submit` after the send must not free the slot while the /// worker still holds the needle. pub queue_permit: Option, + /// Tracked from submission, so a delete cancels the write while it waits. + pub pending: Option, } /// Maximum number of write requests to batch together. @@ -167,12 +174,13 @@ impl WriteQueue { #[cfg(test)] self.submits .fetch_add(1, std::sync::atomic::Ordering::Relaxed); - { + let pending = { let store = self.state.store.read().unwrap(); - if store.find_volume(volume_id).is_none() { - return Err(VolumeError::NotFound); + match store.find_volume(volume_id) { + Some((_, vol)) => vol.track_pending_write(&needle), + None => return Err(VolumeError::NotFound), } - } + }; // The permit travels inside the request, so the global queued bound // covers the request until the worker drops it — including after a // canceled submit. @@ -189,6 +197,7 @@ impl WriteQueue { fsync, response_tx, queue_permit: Some(queue_permit), + pending: Some(pending), }; // A send fails only against a retiring worker's closed channel; the @@ -345,10 +354,11 @@ fn process_batch(state: Arc, batch: Vec) { for req in batch { let vid = req.volume_id; + let entry = (req.needle, req.fsync, req.response_tx, req.pending); if let Some(group) = groups.iter_mut().find(|(v, _)| *v == vid) { - group.1.push((req.needle, req.fsync, req.response_tx)); + group.1.push(entry); } else { - groups.push((vid, vec![(req.needle, req.fsync, req.response_tx)])); + groups.push((vid, vec![entry])); } } @@ -356,16 +366,28 @@ fn process_batch(state: Arc, batch: Vec) { // the group's I/O with the guard released — one volume's write must not // block admission on another volume's. for (vid, entries) in groups { - let (mut writes, senders): (Vec<_>, Vec<_>) = entries - .into_iter() - .map(|(needle, fsync, response_tx)| ((needle, fsync), response_tx)) - .unzip(); let vol = { let store = state.store.read().unwrap(); store.writable_volume(vid) }; - let results = match vol { - Ok(vol) => vol.write_needles_grouped(&mut writes), + let mut writes = Vec::with_capacity(entries.len()); + let mut senders = Vec::with_capacity(entries.len()); + let mut pendings = Vec::with_capacity(entries.len()); + for (needle, fsync, response_tx, pending) in entries { + // A delete reaching a remounted volume cannot cancel a write + // queued for the old one. + if let (Ok(vol), Some(pending)) = (&vol, &pending) + && !vol.tracks_pending_write(pending) + { + let _ = response_tx.send(Err(VolumeError::NotFound)); + continue; + } + writes.push((needle, fsync)); + senders.push(response_tx); + pendings.push(pending); + } + let results: Vec = match vol { + Ok(vol) => vol.write_pending_needles(&mut writes, &pendings), Err(e) => writes .iter() .map(|_| match e { @@ -577,6 +599,7 @@ mod tests { fsync: true, response_tx, queue_permit: None, + pending: None, } }) .collect(); @@ -669,6 +692,7 @@ mod tests { fsync: false, response_tx, queue_permit: None, + pending: None, }]; let writer = { let state = state.clone(); diff --git a/seaweed-volume/src/storage/volume.rs b/seaweed-volume/src/storage/volume.rs index 614d52a115..b351a246e2 100644 --- a/seaweed-volume/src/storage/volume.rs +++ b/seaweed-volume/src/storage/volume.rs @@ -1404,6 +1404,8 @@ pub struct Volume { /// `data_file_sync_lock` -> `data_file_access_control` -> `io`. data_file_sync_lock: std::sync::Mutex<()>, + pending_writes: Arc, + /// Go's superBlockAccessLock. super_block: RwLock, @@ -1453,6 +1455,36 @@ pub struct Volume { const NO_WRITE_OR_DELETE: u8 = 1; const NO_WRITE_CAN_DELETE: u8 = 2; +#[derive(Default)] +struct PendingWrites(Mutex>>>); + +struct CancelFlag { + cookie: Cookie, + cancelled: AtomicBool, +} + +/// A write that has entered the volume but not returned yet. A delete of its +/// needle in that window cancels it: the write may still be queued behind the +/// volume locks, and landing after the delete would revive the needle on this +/// copy alone. Dropping it untracks the write. +pub struct PendingWrite { + writes: Arc, + id: NeedleId, + flag: Arc, +} + +impl Drop for PendingWrite { + fn drop(&mut self) { + let mut by_id = self.writes.0.lock().unwrap(); + if let Some(writes) = by_id.get_mut(&self.id) { + writes.retain(|w| !Arc::ptr_eq(w, &self.flag)); + if writes.is_empty() { + by_id.remove(&self.id); + } + } + } +} + /// What a volume is created with beyond its id, directories and index kind: /// the tail of Go's `NewVolume` argument list. The default is an empty /// collection with no replication, no TTL and no preallocation at the @@ -1527,6 +1559,7 @@ impl Volume { needle_map_kind, data_file_access_control: Arc::new(DataFileAccessControl::default()), data_file_sync_lock: std::sync::Mutex::new(()), + pending_writes: Arc::default(), super_block: RwLock::new(SuperBlock { replica_placement: replica_placement.unwrap_or_default(), ttl: ttl.unwrap_or(crate::storage::needle::ttl::TTL::EMPTY), @@ -1584,6 +1617,7 @@ impl Volume { needle_map_kind: NeedleMapKind::InMemory, data_file_access_control: Arc::new(DataFileAccessControl::default()), data_file_sync_lock: std::sync::Mutex::new(()), + pending_writes: Arc::default(), super_block: RwLock::new(SuperBlock::default()), no_write: AtomicU8::new(0), io_unavailable: Arc::default(), @@ -2776,6 +2810,15 @@ impl Volume { pub fn write_needle_unsynced_or_refused( &self, n: &mut Needle, + ) -> Result, VolumeError> { + self.write_pending_needle_unsynced(n, None) + } + + /// write_needle_unsynced_or_refused for a write a delete can cancel. + pub fn write_pending_needle_unsynced( + &self, + n: &mut Needle, + pending: Option<&PendingWrite>, ) -> Result, VolumeError> { let _sync_guard = self.data_file_sync_lock.lock().unwrap(); let _guard = self.data_file_access_control.write_lock(); @@ -2783,7 +2826,9 @@ impl Volume { if let Err(refused) = self.check_writable() { return Ok(Err(refused)); } - self.do_write_request(&mut io, n, true, false).map(Ok) + let written = self.do_write_request(&mut io, n, true, false)?; + self.drop_cancelled_write(&mut io, pending, n, &written)?; + Ok(Ok(written)) } /// Write a batch of needles the way Go's processBatch does: one .dat @@ -2797,6 +2842,16 @@ impl Volume { pub fn write_needles_grouped( &self, writes: &mut [(Needle, bool)], + ) -> Vec> { + self.write_pending_needles(writes, &[]) + } + + /// write_needles_grouped for writes a delete can cancel, `pending[i]` + /// tracking `writes[i]`. + pub fn write_pending_needles( + &self, + writes: &mut [(Needle, bool)], + pending: &[Option], ) -> Vec> { // The sync lock serializes a .dat mutation with the flush that makes // it durable: a failed sync's rollback can never clobber an append @@ -2811,16 +2866,21 @@ impl Volume { let mut ids = HashSet::new(); let len = rest.iter().take_while(|(n, _)| ids.insert(n.id)).count(); let (run, tail) = std::mem::take(&mut rest).split_at_mut(len); + let tracked = pending.get(results.len()..).unwrap_or_default(); if run.iter().any(|(_, fsync)| *fsync) { - results.extend(self.write_synced_run(run)); + results.extend(self.write_synced_run(run, tracked)); } else { let _guard = self.data_file_access_control.write_lock(); let mut io = self.io.write().unwrap(); - for (n, _) in run.iter_mut() { - results.push( - self.check_writable() - .and_then(|()| self.do_write_request(&mut io, n, true, false)), - ); + for (i, (n, _)) in run.iter_mut().enumerate() { + let written = self + .check_writable() + .and_then(|()| self.do_write_request(&mut io, n, true, false)); + results.push(written.and_then(|written| { + let pending = tracked.get(i).and_then(Option::as_ref); + self.drop_cancelled_write(&mut io, pending, n, &written) + .map(|()| written) + })); } } rest = tail; @@ -2831,6 +2891,7 @@ impl Volume { fn write_synced_run( &self, run: &mut [(Needle, bool)], + pending: &[Option], ) -> Vec> { // Per entry: Some(offset) once appended, None when it dedups. let mut staged = Vec::with_capacity(run.len()); @@ -2874,6 +2935,7 @@ impl Volume { } } + let mut took_back = false; { let _guard = self.data_file_access_control.write_lock(); let mut io = self.io.write().unwrap(); @@ -2890,24 +2952,37 @@ impl Volume { // writes, so the entries after it are refused the way a lone // write would be. let mut refused = false; - for ((n, fsync), r) in run.iter().zip(staged.iter_mut()) { + for (i, ((n, fsync), r)) in run.iter().zip(staged.iter_mut()).enumerate() { if refused { *r = Err(self .check_writable() .err() .unwrap_or(VolumeError::ReadOnly(self.id))); - } else if let Ok(Some(offset)) = *r - && let Err(e) = self.publish_write(&mut io, n, offset, *fsync) - { - *r = Err(e); - refused = *fsync; + } else if let Ok(Some(offset)) = *r { + if let Err(e) = self.publish_write(&mut io, n, offset, *fsync) { + *r = Err(e); + refused = *fsync; + } else if let Err(e) = self.drop_cancelled_write( + &mut io, + pending.get(i).and_then(Option::as_ref), + n, + &(offset, n.size, false), + ) { + took_back |= matches!(e, VolumeError::Deleted); + *r = Err(e); + } } } } if sync { let io = self.io.read().unwrap(); - if let Err(e) = self.flush_idx(&io) { + let flushed = if took_back { + self.flush_taken_back(&io) + } else { + Ok(()) + }; + if let Err(e) = flushed.and_then(|()| self.flush_idx(&io)) { return staged .into_iter() .map(|r| r.and_then(|_| Err(run_error(&e)))) @@ -3028,6 +3103,17 @@ impl Volume { Ok(()) } + /// Flush the tombstones drop_cancelled_write appended after the .dat sync, + /// so their .idx rows never outlive them. Like flush_idx, a failure cannot + /// be undone and stops the volume taking writes. + fn flush_taken_back(&self, io: &VolumeIo) -> Result<(), VolumeError> { + self.flush_dat(io).map_err(|e| { + self.check_read_write_error(Some(&e)); + self.set_no_write_flag(NO_WRITE_OR_DELETE, true); + VolumeError::Io(e) + }) + } + /// Take an unflushed append back off the .dat after its sync failed. pub(crate) fn truncate_dat(&self, io: &VolumeIo, len: u64) -> io::Result<()> { #[cfg(test)] @@ -3410,10 +3496,79 @@ impl Volume { Ok((offset, n.size, actual_size)) } + pub fn track_pending_write(&self, n: &Needle) -> PendingWrite { + let flag = Arc::new(CancelFlag { + cookie: n.cookie, + cancelled: AtomicBool::new(false), + }); + self.pending_writes + .0 + .lock() + .unwrap() + .entry(n.id) + .or_default() + .push(flag.clone()); + PendingWrite { + writes: self.pending_writes.clone(), + id: n.id, + flag, + } + } + + /// Whether the write was tracked on this volume rather than on one that + /// has since been unmounted under the same id. + pub fn tracks_pending_write(&self, pending: &PendingWrite) -> bool { + Arc::ptr_eq(&pending.writes, &self.pending_writes) + } + + /// Cancels the writes of the needle that are in flight. A caller that found + /// the needle missing must look it up again afterwards: a write that + /// finished in between is no longer pending. + pub fn cancel_pending_writes(&self, id: NeedleId, cookie: Cookie) { + if let Some(writes) = self.pending_writes.0.lock().unwrap().get(&id) { + for w in writes.iter().filter(|w| w.cookie == cookie) { + w.cancelled.store(true, Ordering::Release); + } + } + } + + /// Tombstones a cancelled write right after it is published, under the + /// same lock hold, so no reader, later write or compaction sees it live + /// in between. + fn drop_cancelled_write( + &self, + io: &mut VolumeIo, + pending: Option<&PendingWrite>, + n: &Needle, + written: &(u64, Size, bool), + ) -> Result<(), VolumeError> { + let (offset, _, unchanged) = *written; + if unchanged || !pending.is_some_and(|p| p.flag.cancelled.load(Ordering::Acquire)) { + return Ok(()); + } + let live = match io.nm.as_ref() { + Some(nm) => nm.get(n.id)?.is_some_and(|nv| { + !nv.size.is_deleted() && nv.offset == Offset::from_actual_offset(offset as i64) + }), + None => false, + }; + if !live { + return Ok(()); + } + let mut tombstone = Needle { + id: n.id, + cookie: n.cookie, + ..Needle::default() + }; + self.do_delete_request(io, &mut tombstone)?; + Err(VolumeError::Deleted) + } + // ---- Delete ---- /// Delete a needle from the volume. pub fn delete_needle(&self, n: &mut Needle) -> Result { + self.cancel_pending_writes(n.id, n.cookie); let _sync_guard = self.data_file_sync_lock.lock().unwrap(); let _guard = self.data_file_access_control.write_lock(); let mut io = self.io.write().unwrap(); @@ -6158,6 +6313,22 @@ impl Volume { self.io.read().unwrap() } + /// Holds the lock a write waits on behind another write's fsync. + #[cfg(test)] + pub(crate) fn sync_lock_for_test(&self) -> std::sync::MutexGuard<'_, ()> { + self.data_file_sync_lock.lock().unwrap() + } + + #[cfg(test)] + pub(crate) fn pending_writes_for_test(&self, id: NeedleId) -> usize { + self.pending_writes + .0 + .lock() + .unwrap() + .get(&id) + .map_or(0, Vec::len) + } + #[cfg(test)] pub(crate) fn fail_next_fsync_for_test(&self, fail: bool) { self.io.write().unwrap().fail_fsync_for_test = fail; @@ -8484,6 +8655,171 @@ mod tests { assert!(matches!(err, VolumeError::Deleted)); } + fn pending_needle(id: u64, data: &[u8]) -> Needle { + Needle { + id: NeedleId(id), + cookie: Cookie(0xcc), + data: data.to_vec(), + data_size: data.len() as u32, + ..Needle::default() + } + } + + fn write_pending( + v: &Volume, + n: Needle, + fsync: bool, + pending: PendingWrite, + ) -> Result<(u64, Size, bool), VolumeError> { + v.write_pending_needles(&mut [(n, fsync)], &[Some(pending)]) + .pop() + .unwrap() + } + + /// A delete that finds nothing can still have the write it undoes in + /// flight. That write must not land live afterwards. + #[test] + fn test_delete_of_missing_needle_cancels_pending_write() { + for fsync in [false, true] { + let tmp = TempDir::new().unwrap(); + let dir = tmp.path().to_str().unwrap(); + let v = make_test_volume(dir); + + let n = pending_needle(1, b"late write"); + let pending = v.track_pending_write(&n); + v.delete_needle(&mut pending_needle(1, b"")).unwrap(); + let err = write_pending(&v, n, fsync, pending).unwrap_err(); + assert!(matches!(err, VolumeError::Deleted)); + + let err = v.read_needle(&mut pending_needle(1, b"")).unwrap_err(); + assert!(matches!(err, VolumeError::Deleted)); + assert!(v.pending_writes.0.lock().unwrap().is_empty()); + + v.close(); + let v = Volume::new( + dir, + dir, + VolumeId(1), + NeedleMapKind::InMemory, + &VolumeSpec::default(), + ) + .unwrap(); + assert!(v.read_needle(&mut pending_needle(1, b"")).is_err()); + } + } + + /// The unsynced write path takes a cancelled write back the same way. + #[test] + fn test_unsynced_write_cancelled_by_delete() { + let tmp = TempDir::new().unwrap(); + let v = make_test_volume(tmp.path().to_str().unwrap()); + + let mut n = pending_needle(1, b"late write"); + let pending = v.track_pending_write(&n); + v.cancel_pending_writes(n.id, n.cookie); + let err = v + .write_pending_needle_unsynced(&mut n, Some(&pending)) + .unwrap_err(); + assert!(matches!(err, VolumeError::Deleted)); + let err = v.read_needle(&mut pending_needle(1, b"")).unwrap_err(); + assert!(matches!(err, VolumeError::Deleted)); + } + + /// A delete that finds a copy also cancels a retry of the same write that + /// is still in flight, so the retry does not bring the needle back. + #[test] + fn test_delete_cancels_pending_rewrite() { + let tmp = TempDir::new().unwrap(); + let dir = tmp.path().to_str().unwrap(); + let v = make_test_volume(dir); + + v.write_needle(&mut pending_needle(1, b"first"), true, false) + .unwrap(); + let rewrite = pending_needle(1, b"retry"); + let pending = v.track_pending_write(&rewrite); + v.delete_needle(&mut pending_needle(1, b"")).unwrap(); + let err = write_pending(&v, rewrite, false, pending).unwrap_err(); + assert!(matches!(err, VolumeError::Deleted)); + let err = v.read_needle(&mut pending_needle(1, b"")).unwrap_err(); + assert!(matches!(err, VolumeError::Deleted)); + } + + /// Only writes already in flight and carrying the delete's cookie are + /// cancelled. + #[test] + fn test_pending_write_cancel_leaves_later_writes() { + let tmp = TempDir::new().unwrap(); + let dir = tmp.path().to_str().unwrap(); + let v = make_test_volume(dir); + + v.delete_needle(&mut pending_needle(1, b"")).unwrap(); + let n = pending_needle(1, b"after the delete"); + let pending = v.track_pending_write(&n); + write_pending(&v, n, false, pending).unwrap(); + let mut got = pending_needle(1, b""); + v.read_needle(&mut got).unwrap(); + assert_eq!(got.data, b"after the delete"); + + let n = pending_needle(2, b"other cookie"); + let pending = v.track_pending_write(&n); + v.cancel_pending_writes(n.id, Cookie(n.cookie.0 + 1)); + write_pending(&v, n, false, pending).unwrap(); + let mut got = pending_needle(2, b""); + v.read_needle(&mut got).unwrap(); + assert_eq!(got.data, b"other cookie"); + } + + /// The cancelled write is taken back before the next write of the needle + /// runs, so an identical write sent after the delete appends its own copy + /// instead of matching the record about to be deleted. + #[test] + fn test_write_after_cancelled_write_stays_live() { + for fsync in [false, true] { + let tmp = TempDir::new().unwrap(); + let v = make_test_volume(tmp.path().to_str().unwrap()); + + let first = pending_needle(1, b"same bytes"); + let first_pending = v.track_pending_write(&first); + v.cancel_pending_writes(first.id, first.cookie); + let retry = pending_needle(1, b"same bytes"); + let retry_pending = v.track_pending_write(&retry); + let mut results = v.write_pending_needles( + &mut [(first, fsync), (retry, fsync)], + &[Some(first_pending), Some(retry_pending)], + ); + let retried = results.pop().unwrap().unwrap(); + assert!( + !retried.2, + "fsync={fsync}: the retry matched the cancelled write" + ); + assert!(matches!(results.pop().unwrap(), Err(VolumeError::Deleted))); + + let mut got = pending_needle(1, b""); + v.read_needle(&mut got).unwrap(); + assert_eq!(got.data, b"same bytes"); + } + } + + /// A durable write taken back appends its tombstone after the .dat sync; + /// the tombstone is flushed before the .idx row that points at it. + #[test] + fn test_taken_back_tombstone_is_synced() { + let tmp = TempDir::new().unwrap(); + let v = make_test_volume(tmp.path().to_str().unwrap()); + + let n = pending_needle(1, b"late write"); + let pending = v.track_pending_write(&n); + v.cancel_pending_writes(n.id, n.cookie); + let err = write_pending(&v, n, true, pending).unwrap_err(); + assert!(matches!(err, VolumeError::Deleted)); + assert_eq!(v.sync_counts_for_test(), (2, 1)); + + let n = pending_needle(2, b"kept"); + let pending = v.track_pending_write(&n); + write_pending(&v, n, true, pending).unwrap(); + assert_eq!(v.sync_counts_for_test(), (3, 2)); + } + // Guard the Rust integrity-check tombstone path against the Go regression // where verifyDeletedNeedleIntegrity forwarded TombstoneFileSize into the // needle-size check, mismatched against the on-disk Size=0 header, and diff --git a/weed/server/volume_grpc_batch_delete.go b/weed/server/volume_grpc_batch_delete.go index 06a2eeb3ea..6b4ab1af69 100644 --- a/weed/server/volume_grpc_batch_delete.go +++ b/weed/server/volume_grpc_batch_delete.go @@ -10,6 +10,7 @@ import ( "github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb" "github.com/seaweedfs/seaweedfs/weed/storage" "github.com/seaweedfs/seaweedfs/weed/storage/needle" + "github.com/seaweedfs/seaweedfs/weed/storage/types" ) func (vs *VolumeServer) BatchDelete(ctx context.Context, req *volume_server_pb.BatchDeleteRequest) (*volume_server_pb.BatchDeleteResponse, error) { @@ -37,8 +38,9 @@ func (vs *VolumeServer) BatchDelete(ctx context.Context, req *volume_server_pb.B n := new(needle.Needle) volumeId, _ := needle.NewVolumeId(vid) + var fidCookie types.Cookie if req.SkipCookieCheck { - n.Id, _, err = needle.ParseNeedleIdCookie(id_cookie) + n.Id, fidCookie, err = needle.ParseNeedleIdCookie(id_cookie) if err != nil { resp.Results = append(resp.Results, &volume_server_pb.DeleteResult{ FileId: fid, @@ -99,6 +101,10 @@ func (vs *VolumeServer) BatchDelete(ctx context.Context, req *volume_server_pb.B n.LastModified = now if !isEcVolume { + if req.SkipCookieCheck { + // the delete cancels in-flight writes carrying this cookie + n.Cookie = fidCookie + } if size, err := vs.store.DeleteVolumeNeedle(volumeId, n); err != nil { resp.Results = append(resp.Results, &volume_server_pb.DeleteResult{ FileId: fid, diff --git a/weed/server/volume_server_handlers_write.go b/weed/server/volume_server_handlers_write.go index 87f0f0d9b8..564d4a8029 100644 --- a/weed/server/volume_server_handlers_write.go +++ b/weed/server/volume_server_handlers_write.go @@ -123,6 +123,13 @@ func (vs *VolumeServer) DeleteHandler(w http.ResponseWriter, r *http.Request) { } _, ok := vs.store.ReadVolumeNeedle(volumeId, n, deleteReadOption(), nil) + if ok != nil && r.FormValue("type") == "replicate" { + // the write this delete undoes can still be queued here + if v := vs.store.GetVolume(volumeId); v != nil { + v.CancelPendingWrites(n.Id, cookie) + _, ok = vs.store.ReadVolumeNeedle(volumeId, n, deleteReadOption(), nil) + } + } if ok != nil { m := make(map[string]uint32) m["size"] = 0 diff --git a/weed/server/volume_server_handlers_write_test.go b/weed/server/volume_server_handlers_write_test.go new file mode 100644 index 0000000000..8ff91c5aaa --- /dev/null +++ b/weed/server/volume_server_handlers_write_test.go @@ -0,0 +1,141 @@ +package weed_server + +import ( + "bytes" + "context" + "net/http" + "net/http/httptest" + "runtime" + "sync" + "testing" + "time" + + "github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb" + "github.com/seaweedfs/seaweedfs/weed/security" + "github.com/seaweedfs/seaweedfs/weed/stats" + "github.com/seaweedfs/seaweedfs/weed/storage" + "github.com/seaweedfs/seaweedfs/weed/storage/backend" + "github.com/seaweedfs/seaweedfs/weed/storage/needle" + "github.com/seaweedfs/seaweedfs/weed/storage/types" + "github.com/seaweedfs/seaweedfs/weed/util" + "github.com/stretchr/testify/require" +) + +// heldSyncBackend parks the first .dat fsync until released: the window where +// a durable write is appended but not indexed yet. +type heldSyncBackend struct { + backend.BackendStorageFile + once sync.Once + entered chan struct{} + release chan struct{} +} + +func (b *heldSyncBackend) Sync() error { + b.once.Do(func() { + close(b.entered) + <-b.release + }) + return b.BackendStorageFile.Sync() +} + +// startHeldWrite starts a durable write of needle 1 on a fresh volume and +// returns once it is parked in its .dat fsync, appended but not indexed. +func startHeldWrite(t *testing.T) (store *storage.Store, fid string, release func(), written <-chan error) { + store = storage.NewStore(nil, "localhost", 8080, 18080, "http://localhost:8080", "store-id", + []string{t.TempDir()}, []int32{10}, []util.MinFreeSpace{{}}, "", storage.NeedleMapInMemory, + []types.DiskType{types.HardDriveType}, nil, 0, stats.DefaultDiskIOProbeConfig()) + t.Cleanup(store.Close) + require.NoError(t, store.AddVolume(1, "", storage.NeedleMapInMemory, "000", "", 0, needle.Version3, 0, types.HardDriveType, 0)) + v := store.GetVolume(1) + held := &heldSyncBackend{BackendStorageFile: v.DataBackend, entered: make(chan struct{}), release: make(chan struct{})} + v.DataBackend = held + + n := &needle.Needle{Id: 1, Cookie: 0x1234, Data: []byte("late write")} + n.Checksum = needle.NewCRC(n.Data) + done := make(chan error, 1) + go func() { + _, err := store.WriteVolumeNeedle(1, n, true, true) + done <- err + }() + <-held.entered + var releaseOnce sync.Once + release = func() { releaseOnce.Do(func() { close(held.release) }) } + t.Cleanup(release) + return store, needle.NewFileIdFromNeedle(1, n).String(), release, done +} + +func requireTakenBack(t *testing.T, store *storage.Store, written <-chan error, takenBack bool) { + t.Helper() + err := <-written + _, readErr := store.ReadVolumeNeedle(1, &needle.Needle{Id: 1}, nil, nil) + if takenBack { + require.ErrorIs(t, err, storage.ErrorDeleted) + require.ErrorIs(t, readErr, storage.ErrorDeleted) + } else { + require.NoError(t, err) + require.NoError(t, readErr) + } +} + +// A replica delete that arrives while the write it undoes is in its fsync +// finds nothing, yet the write must not land live afterwards. A plain delete +// that finds nothing is not sent on to the replicas, so it leaves the write. +func TestDeleteHandlerCancelsAWriteInItsFsyncWindow(t *testing.T) { + for _, tc := range []struct { + query string + live bool + }{ + {"?type=replicate", false}, + {"", true}, + } { + t.Run(tc.query, func(t *testing.T) { + store, fid, release, written := startHeldWrite(t) + resp := httptest.NewRecorder() + vs := &VolumeServer{store: store, guard: security.NewGuard(nil, "", 0, "", 0)} + vs.DeleteHandler(resp, httptest.NewRequest(http.MethodDelete, "/"+fid+tc.query, nil)) + require.Equal(t, http.StatusNotFound, resp.Code) + release() + requireTakenBack(t, store, written, !tc.live) + }) + } +} + +// A batch delete that skips the cookie check still cancels the in-flight +// writes of the file id it names, and only those. +func TestBatchDeleteCancelsAWriteInItsFsyncWindow(t *testing.T) { + for _, tc := range []struct { + cookie string + takenBack bool + }{ + {"", true}, + {"00004321", false}, + } { + t.Run(tc.cookie, func(t *testing.T) { + store, fid, release, written := startHeldWrite(t) + if tc.cookie != "" { + fid = fid[:len(fid)-8] + tc.cookie + } + deleted := make(chan error, 1) + go func() { + _, err := (&VolumeServer{store: store}).BatchDelete(context.Background(), &volume_server_pb.BatchDeleteRequest{ + FileIds: []string{fid}, + SkipCookieCheck: true, + }) + deleted <- err + }() + // the delete waits behind the held write's sync lock + require.Eventually(t, func() bool { + buf := make([]byte, 1<<20) + return bytes.Contains(buf[:runtime.Stack(buf, true)], []byte("(*Volume).syncDelete")) + }, 10*time.Second, time.Millisecond) + release() + require.NoError(t, <-deleted) + err := <-written + if tc.takenBack { + require.ErrorIs(t, err, storage.ErrorDeleted) + } else { + require.NoError(t, err) + } + }) + } +} diff --git a/weed/storage/volume.go b/weed/storage/volume.go index fd615bb452..d2b03fcf30 100644 --- a/weed/storage/volume.go +++ b/weed/storage/volume.go @@ -59,6 +59,8 @@ type Volume struct { asyncRequestsChan chan *needle.AsyncRequest asyncWorkerClosed bool + pendingWrites pendingWrites + lastModifiedTsSeconds uint64 // unix time in seconds lastAppendAtNs uint64 // unix time in nanoseconds lastWriteAppendAtNs uint64 // AppendAtNs of the newest write; tombstones don't move it diff --git a/weed/storage/volume_pending_write.go b/weed/storage/volume_pending_write.go new file mode 100644 index 0000000000..43830a3536 --- /dev/null +++ b/weed/storage/volume_pending_write.go @@ -0,0 +1,95 @@ +package storage + +import ( + "fmt" + "sync" + "time" + + "github.com/seaweedfs/seaweedfs/weed/storage/needle" + . "github.com/seaweedfs/seaweedfs/weed/storage/types" +) + +// pendingWrite is a write that has entered the volume but not returned yet. +// A delete of its needle in that window cancels it: the write may still be +// queued behind the volume locks, and landing after the delete would revive +// the needle on this copy alone. +type pendingWrite struct { + n *needle.Needle + cookie Cookie + cancelled bool +} + +type pendingWrites struct { + sync.Mutex + byId map[NeedleId][]*pendingWrite +} + +func (v *Volume) trackPendingWrite(n *needle.Needle) *pendingWrite { + w := &pendingWrite{n: n, cookie: n.Cookie} + v.pendingWrites.Lock() + defer v.pendingWrites.Unlock() + if v.pendingWrites.byId == nil { + v.pendingWrites.byId = make(map[NeedleId][]*pendingWrite) + } + v.pendingWrites.byId[n.Id] = append(v.pendingWrites.byId[n.Id], w) + return w +} + +func (v *Volume) untrackPendingWrite(w *pendingWrite) { + v.pendingWrites.Lock() + defer v.pendingWrites.Unlock() + id := w.n.Id + writes := v.pendingWrites.byId[id] + for i, x := range writes { + if x == w { + writes = append(writes[:i], writes[i+1:]...) + break + } + } + if len(writes) == 0 { + delete(v.pendingWrites.byId, id) + } else { + v.pendingWrites.byId[id] = writes + } +} + +// CancelPendingWrites cancels the writes of the needle that are in flight. A +// caller that found the needle missing must look it up again afterwards: a +// write that finished in between is no longer pending. +func (v *Volume) CancelPendingWrites(id NeedleId, cookie Cookie) { + v.pendingWrites.Lock() + defer v.pendingWrites.Unlock() + for _, w := range v.pendingWrites.byId[id] { + if w.cookie == cookie { + w.cancelled = true + } + } +} + +func (v *Volume) isCancelledWrite(n *needle.Needle) bool { + v.pendingWrites.Lock() + defer v.pendingWrites.Unlock() + for _, w := range v.pendingWrites.byId[n.Id] { + if w.n == n { + return w.cancelled + } + } + return false +} + +// dropCancelledWrite tombstones a cancelled write right after it is published, +// under the same dataFileSyncLock and dataFileAccessLock hold, so no reader, +// later write or compaction sees it live in between. +func (v *Volume) dropCancelledWrite(n *needle.Needle, offset uint64) error { + if !v.isCancelledWrite(n) { + return nil + } + if nv, ok := v.nm.Get(n.Id); !ok || nv.Size.IsDeleted() || nv.Offset != ToOffset(int64(offset)) { + return nil + } + tombstone := &needle.Needle{Id: n.Id, Cookie: n.Cookie, LastModified: uint64(time.Now().Unix())} + if _, err := v.doDeleteRequest(tombstone); err != nil { + return err + } + return fmt.Errorf("needle %s: %w", needle.NewFileIdFromNeedle(v.Id, n), ErrorDeleted) +} diff --git a/weed/storage/volume_pending_write_test.go b/weed/storage/volume_pending_write_test.go new file mode 100644 index 0000000000..0ca6a84bf9 --- /dev/null +++ b/weed/storage/volume_pending_write_test.go @@ -0,0 +1,229 @@ +package storage + +import ( + "fmt" + "sync/atomic" + "testing" + "time" + + "github.com/seaweedfs/seaweedfs/weed/storage/backend" + "github.com/seaweedfs/seaweedfs/weed/storage/needle" + "github.com/seaweedfs/seaweedfs/weed/storage/super_block" + . "github.com/seaweedfs/seaweedfs/weed/storage/types" + "github.com/stretchr/testify/require" +) + +func waitForPendingWrites(t *testing.T, v *Volume, id NeedleId, count int) { + t.Helper() + deadline := time.Now().Add(10 * time.Second) + for { + v.pendingWrites.Lock() + got := len(v.pendingWrites.byId[id]) + v.pendingWrites.Unlock() + if got >= count { + return + } + if time.Now().After(deadline) { + t.Fatalf("%d of %d writes of needle %d registered", got, count, id) + } + time.Sleep(time.Millisecond) + } +} + +func requireNeedleDeleted(t *testing.T, v *Volume, id NeedleId) { + t.Helper() + _, err := v.readNeedle(&needle.Needle{Id: id}, nil, nil) + require.ErrorIs(t, err, ErrorDeleted) +} + +// A replicate delete that finds nothing can still have the write it undoes +// queued behind the volume locks. That write must not land live afterwards. +func TestDeleteOfMissingNeedleCancelsQueuedWrite(t *testing.T) { + for _, fsync := range []bool{false, true} { + t.Run(fmt.Sprintf("fsync=%v", fsync), func(t *testing.T) { + dir := t.TempDir() + v, err := NewVolume(dir, dir, "", 1, NeedleMapInMemory, &super_block.ReplicaPlacement{}, &needle.TTL{}, 0, needle.GetCurrentVersion(), 0, 0) + require.NoError(t, err) + + n := newRandomNeedle(1) + v.dataFileSyncLock.Lock() + written := make(chan error, 1) + go func() { + _, _, _, err := v.writeNeedle2(n, true, fsync, false) + written <- err + }() + waitForPendingWrites(t, v, n.Id, 1) + + _, err = v.readNeedle(&needle.Needle{Id: n.Id}, nil, nil) + require.ErrorIs(t, err, ErrorNotFound) + v.CancelPendingWrites(n.Id, n.Cookie) + v.dataFileSyncLock.Unlock() + + require.ErrorIs(t, <-written, ErrorDeleted) + requireNeedleDeleted(t, v, n.Id) + require.Empty(t, v.pendingWrites.byId) + + v.Close() + v, err = NewVolume(dir, dir, "", 1, NeedleMapInMemory, nil, nil, 0, needle.GetCurrentVersion(), 0, 0) + require.NoError(t, err) + defer v.Close() + requireNeedleDeleted(t, v, n.Id) + }) + } +} + +// A delete that finds a copy also cancels a retry of the same write that is +// still queued, so the retry does not bring the needle back. +func TestDeleteCancelsQueuedRewrite(t *testing.T) { + v := newWorkerTestVolume(t) + n := newRandomNeedle(1) + _, _, _, err := v.writeNeedle2(n, true, false, false) + require.NoError(t, err) + + rewrite := newRandomNeedle(1) + rewrite.Cookie = n.Cookie + // holds the durable rewrite after it registers, before it reaches the worker + v.asyncWorkerLock.Lock() + written := make(chan error, 1) + go func() { + _, _, _, err := v.writeNeedle2(rewrite, true, true, false) + written <- err + }() + waitForPendingWrites(t, v, n.Id, 1) + + _, err = v.deleteNeedle2(&needle.Needle{Id: n.Id, Cookie: n.Cookie}) + require.NoError(t, err) + requireNeedleDeleted(t, v, n.Id) + v.asyncWorkerLock.Unlock() + + require.ErrorIs(t, <-written, ErrorDeleted) + requireNeedleDeleted(t, v, n.Id) + require.Empty(t, v.pendingWrites.byId) +} + +// Only writes already in flight are cancelled: writing the needle again after +// the delete works as before. +func TestWriteAfterDeleteIsNotCancelled(t *testing.T) { + v := newWorkerTestVolume(t) + n := newRandomNeedle(1) + v.CancelPendingWrites(n.Id, n.Cookie) + _, err := v.deleteNeedle2(&needle.Needle{Id: n.Id, Cookie: n.Cookie}) + require.NoError(t, err) + + _, _, _, err = v.writeNeedle2(n, true, false, false) + require.NoError(t, err) + got := &needle.Needle{Id: n.Id} + _, err = v.readNeedle(got, nil, nil) + require.NoError(t, err) + require.Equal(t, n.Data, got.Data) +} + +// A delete only cancels writes that carry its cookie. +func TestCancelWithOtherCookieKeepsWrite(t *testing.T) { + v := newWorkerTestVolume(t) + n := newRandomNeedle(1) + v.dataFileSyncLock.Lock() + written := make(chan error, 1) + go func() { + _, _, _, err := v.writeNeedle2(n, true, false, false) + written <- err + }() + waitForPendingWrites(t, v, n.Id, 1) + v.CancelPendingWrites(n.Id, n.Cookie+1) + v.dataFileSyncLock.Unlock() + + require.NoError(t, <-written) + got := &needle.Needle{Id: n.Id} + _, err := v.readNeedle(got, nil, nil) + require.NoError(t, err) + require.Equal(t, n.Data, got.Data) +} + +// The cancelled write is taken back before the next write of the needle runs, +// so an identical write sent after the delete appends its own copy instead of +// matching the record about to be deleted, even in the same batch. +func TestWriteAfterCancelledWriteStaysLive(t *testing.T) { + v := newWorkerTestVolume(t) + v.location = &DiskLocation{syncBatch: SyncBatchOption{ + MinBytes: 1 << 30, + MinRequests: syncBatchMaxRequests, + Wait: time.Minute, + }} + armed := make(chan chan time.Time, 1) + prevAfter := syncBatchAfter + syncBatchAfter = func(time.Duration) <-chan time.Time { + window := make(chan time.Time, 1) + armed <- window + return window + } + defer func() { syncBatchAfter = prevAfter }() + + queue := func(n *needle.Needle) *needle.AsyncRequest { + w := v.trackPendingWrite(n) + t.Cleanup(func() { v.untrackPendingWrite(w) }) + r := needle.NewAsyncRequest(n, true) + r.ActualSize = needle.GetActualSize(Size(len(n.Data)), v.Version()) + require.True(t, v.asyncRequestAppend(r)) + return r + } + + n := newRandomNeedle(1) + first := queue(n) + window := <-armed + v.CancelPendingWrites(n.Id, n.Cookie) + retry := queue(&needle.Needle{Id: n.Id, Cookie: n.Cookie, Data: n.Data, Checksum: n.Checksum}) + require.Eventually(t, func() bool { return len(v.asyncRequestsChan) == 0 }, 10*time.Second, time.Millisecond) + window <- time.Now() + + _, _, _, err := first.WaitComplete() + require.ErrorIs(t, err, ErrorDeleted) + _, _, isUnchanged, err := retry.WaitComplete() + require.NoError(t, err) + require.False(t, isUnchanged, "the retry matched the cancelled write") + got := &needle.Needle{Id: n.Id} + _, err = v.readNeedle(got, nil, nil) + require.NoError(t, err) + require.Equal(t, n.Data, got.Data) +} + +type syncedSizeBackend struct { + backend.BackendStorageFile + synced atomic.Int64 +} + +func (b *syncedSizeBackend) Sync() error { + size, _, err := b.GetStat() + if err != nil { + return err + } + b.synced.Store(size) + return b.BackendStorageFile.Sync() +} + +// A durable write taken back during its fsync appends the tombstone after +// that fsync; it must be flushed before the .idx row that points at it. +func TestTakenBackTombstoneIsSynced(t *testing.T) { + for _, isStopping := range []bool{false, true} { + t.Run(fmt.Sprintf("inline=%v", isStopping), func(t *testing.T) { + v := newWorkerTestVolume(t) + synced := &syncedSizeBackend{BackendStorageFile: v.DataBackend} + v.DataBackend = synced + + n := newRandomNeedle(1) + v.dataFileSyncLock.Lock() + written := make(chan error, 1) + go func() { + _, _, _, err := v.writeNeedle2(n, true, true, isStopping) + written <- err + }() + waitForPendingWrites(t, v, n.Id, 1) + v.CancelPendingWrites(n.Id, n.Cookie) + v.dataFileSyncLock.Unlock() + + require.ErrorIs(t, <-written, ErrorDeleted) + size, _, err := v.DataBackend.GetStat() + require.NoError(t, err) + require.Equal(t, size, synced.synced.Load(), "the tombstone was not flushed") + }) + } +} diff --git a/weed/storage/volume_write.go b/weed/storage/volume_write.go index f4597b2937..bd3bc39c4f 100644 --- a/weed/storage/volume_write.go +++ b/weed/storage/volume_write.go @@ -236,19 +236,42 @@ func (v *Volume) syncWrite(n *needle.Needle, checkCookie bool, fsync bool) (offs return 0, 0, false, syncErr } if !isUnchanged { - err = v.publishWriteRequest(n, offset, fsync) + if err = v.publishWriteRequest(n, offset, fsync); err == nil { + err = v.dropCancelledWrite(n, offset) + } } // The .idx row goes down before the write is acknowledged, same as the // .dat record above; the access lock is dropped for the fsync the same // way, since the row is already published. v.dataFileAccessLock.Unlock() - if idxErr := v.syncIndex(); err == nil { + var idxErr error + if errors.Is(err, ErrorDeleted) { + idxErr = v.syncTakenBack(backend) + } + if idxErr == nil { + idxErr = v.syncIndex() + } + if err == nil { err = idxErr } v.dataFileAccessLock.Lock() return } +// syncTakenBack flushes the tombstones dropCancelledWrite appended after the +// .dat sync, so their .idx rows never outlive them. Like syncIndex, a failure +// cannot be undone and stops the volume taking writes. +func (v *Volume) syncTakenBack(backend backend.BackendStorageFile) error { + if err := backend.Sync(); err != nil { + v.checkReadWriteError(err) + v.noWriteLock.Lock() + v.noWriteOrDelete = true + v.noWriteLock.Unlock() + return fmt.Errorf("sync volume %d tombstone: %w", v.Id, err) + } + return nil +} + // syncIndex flushes the .idx, the second half of a durable write: the row is // already published and cannot be taken back, so a failed sync stops the // volume taking writes rather than acknowledge an index entry that may not @@ -278,6 +301,8 @@ func (v *Volume) writeNeedle2(n *needle.Needle, checkCookie bool, fsync bool, is return 0, 0, false, err } + defer v.untrackPendingWrite(v.trackPendingWrite(n)) + if n.Ttl == needle.EMPTY_TTL && v.Ttl != needle.EMPTY_TTL { n.SetHasTtl() n.Ttl = v.Ttl @@ -304,7 +329,9 @@ func (v *Volume) doWriteRequest(n *needle.Needle, checkCookie bool, fsync bool) if err != nil || isUnchanged { return } - err = v.publishWriteRequest(n, offset, fsync) + if err = v.publishWriteRequest(n, offset, fsync); err == nil { + err = v.dropCancelledWrite(n, offset) + } return } @@ -433,6 +460,8 @@ func (v *Volume) syncDelete(n *needle.Needle) (Size, error) { } func (v *Volume) deleteNeedle2(n *needle.Needle) (Size, error) { + v.CancelPendingWrites(n.Id, n.Cookie) + // todo: delete info is always appended no fsync, it may need fsync in future fsync := false @@ -629,6 +658,7 @@ func (v *Volume) processBatchRun(run []*needle.AsyncRequest) { } refused := false + tookBack := false for i := 0; i < len(run); i++ { if refused { run[i].UpdateResult(0, 0, false, fmt.Errorf("volume %d is read only", v.Id)) @@ -639,6 +669,9 @@ func (v *Volume) processBatchRun(run []*needle.AsyncRequest) { if perr := v.publishWriteRequest(run[i].N, stagedWriteOffsets[i], true); perr != nil { run[i].UpdateResult(0, 0, false, perr) refused = true + } else if derr := v.dropCancelledWrite(run[i].N, stagedWriteOffsets[i]); derr != nil { + run[i].UpdateResult(0, 0, false, derr) + tookBack = tookBack || errors.Is(derr, ErrorDeleted) } } } else if stagedDeleteOk[i] { @@ -652,7 +685,14 @@ func (v *Volume) processBatchRun(run []*needle.AsyncRequest) { // rows cannot be rolled back, so a failed sync fails every request in the // run instead of acknowledging an index entry that may not survive. if needIndexSync { - if idxErr := v.syncIndex(); idxErr != nil { + var idxErr error + if tookBack { + idxErr = v.syncTakenBack(backend) + } + if idxErr == nil { + idxErr = v.syncIndex() + } + if idxErr != nil { for i := 0; i < len(run); i++ { run[i].UpdateResult(0, 0, false, idxErr) }