diff --git a/seaweed-volume/src/server/grpc_server.rs b/seaweed-volume/src/server/grpc_server.rs
index 369022e3f..3c82b680e 100644
--- a/seaweed-volume/src/server/grpc_server.rs
+++ b/seaweed-volume/src/server/grpc_server.rs
@@ -2326,7 +2326,7 @@ impl VolumeServer for VolumeGrpcService {
             // Write needle to local volume
             let mut store = state.store.write().unwrap();
             store
-                .write_volume_needle(vid, &mut n)
+                .write_volume_needle(vid, &mut n, false)
                 .map_err(|e| Status::internal(format!("write needle: {}", e)))?;
         }
 
@@ -3987,7 +3987,7 @@ impl VolumeServer for VolumeGrpcService {
         let local_handle = tokio::task::spawn_blocking(move || {
             let mut store = state_clone.store.write().unwrap();
             store
-                .write_volume_needle(vid, &mut n_clone)
+                .write_volume_needle(vid, &mut n_clone, false)
                 .map(|_| ())
                 .map_err(|e| format!("local write needle {} size {}: {}", needle_id, size, e))
         });
@@ -5218,7 +5218,7 @@ mod tests {
                 data_size: "remote-incremental-copy".len() as u32,
                 ..Needle::default()
             };
-            volume.write_needle(&mut needle, true).unwrap();
+            volume.write_needle(&mut needle, true, false).unwrap();
             volume.sync_to_disk().unwrap();
             (
                 std::fs::read(volume.file_name(".dat")).unwrap(),
@@ -5386,7 +5386,7 @@ mod tests {
                 data_size: b"ec-generate".len() as u32,
                 ..Needle::default()
             };
-            volume.write_needle(&mut needle, true).unwrap();
+            volume.write_needle(&mut needle, true, false).unwrap();
             volume.sync_to_disk().unwrap();
         }
 
@@ -5609,7 +5609,7 @@ mod tests {
                 data: payload,
                 ..Needle::default()
             };
-            v.write_needle(&mut needle, true).unwrap();
+            v.write_needle(&mut needle, true, false).unwrap();
             v.sync_to_disk().unwrap();
         }
         let dat_path = {
diff --git a/seaweed-volume/src/server/handlers.rs b/seaweed-volume/src/server/handlers.rs
index 429a13cdd..f1bf9da36 100644
--- a/seaweed-volume/src/server/handlers.rs
+++ b/seaweed-volume/src/server/handlers.rs
@@ -2580,11 +2580,17 @@ pub async fn post_handler(
         n.set_has_name();
     }
 
+    // A durable write flushes before it is acked. Read it the way Go's
+    // r.FormValue does, off the decoded fields, so a percent-encoded value is
+    // honored here too. ReplicatedWrite forwards the parameter, so a replica
+    // sees it the same way the primary did.
+    let fsync = form_value("fsync").as_deref() == Some("true");
+
     let write_result = if let Some(wq) = state.write_queue.get() {
-        wq.submit(vid, n.clone()).await
+        wq.submit(vid, n.clone(), fsync).await
     } else {
         let mut store = state.store.write().unwrap();
-        store.write_volume_needle(vid, &mut n)
+        store.write_volume_needle(vid, &mut n, fsync)
     };
 
     // Replicate to remote volume servers if this volume has replicas.
@@ -3852,6 +3858,26 @@ fn parse_content_disposition_filename(value: &str) -> Option<String> {
 mod tests {
     use super::*;
 
+    /// The upload handler reads fsync off the decoded query fields rather than
+    /// matching the raw string, because Go's r.FormValue decodes and a raw
+    /// match would silently drop a percent-encoded value.
+    #[test]
+    fn test_encoded_query_field_decodes() {
+        let raw = "fsync=%74rue";
+        assert!(
+            !raw.split('&').any(|p| p == "fsync=true"),
+            "a raw match is exactly what misses this"
+        );
+        let fields: Vec<(String, String)> = serde_urlencoded::from_str(raw).unwrap();
+        assert_eq!(
+            fields
+                .iter()
+                .find(|(k, _)| k == "fsync")
+                .map(|(_, v)| v.as_str()),
+            Some("true")
+        );
+    }
+
     #[test]
     fn test_parse_url_path_comma() {
         let (vid, nid, cookie) = parse_url_path("/3,01637037d6").unwrap();
diff --git a/seaweed-volume/src/server/write_queue.rs b/seaweed-volume/src/server/write_queue.rs
index df809db38..7c8e84285 100644
--- a/seaweed-volume/src/server/write_queue.rs
+++ b/seaweed-volume/src/server/write_queue.rs
@@ -1,9 +1,10 @@
 //! Async batched write processing for the volume server.
 //!
-//! Instead of each upload handler directly calling `write_needle` and syncing,
-//! writes are submitted to a queue. A background worker drains the queue in
-//! batches (up to 128 entries), groups them by volume ID, processes them
-//! together, and syncs once per volume for the entire batch.
+//! Instead of each upload handler directly calling `write_needle`, writes are
+//! submitted to a queue. A background worker drains the queue in batches (up to
+//! 128 entries), groups them by volume ID, and processes them together under a
+//! single store lock. Requests that asked for `fsync` are flushed by
+//! `write_needle` itself, one flush per durable write.
 
 use std::sync::Arc;
 
@@ -19,10 +20,14 @@ use super::volume_server::VolumeServerState;
 /// Result of a single write operation: (offset, size, is_unchanged).
 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<WriteResult>)>;
+
 /// A request to write a needle, submitted to the write queue.
 pub struct WriteRequest {
     pub volume_id: VolumeId,
     pub needle: Needle,
+    pub fsync: bool,
     pub response_tx: oneshot::Sender<WriteResult>,
 }
 
@@ -54,11 +59,12 @@ impl WriteQueue {
     /// Submit a write request and wait for the result.
     ///
     /// Returns `Err` if the worker has shut down or the response channel was dropped.
-    pub async fn submit(&self, volume_id: VolumeId, needle: Needle) -> WriteResult {
+    pub async fn submit(&self, volume_id: VolumeId, needle: Needle, fsync: bool) -> WriteResult {
         let (response_tx, response_rx) = oneshot::channel();
         let request = WriteRequest {
             volume_id,
             needle,
+            fsync,
             response_tx,
         };
 
@@ -138,14 +144,14 @@ fn process_batch(state: Arc<VolumeServerState>, batch: Vec<WriteRequest>) {
     // Group requests by volume ID for efficient processing.
     // We use a Vec of (VolumeId, Vec<(Needle, Sender)>) to preserve order
     // and avoid requiring Hash on VolumeId.
-    let mut groups: Vec<(VolumeId, Vec<(Needle, oneshot::Sender<WriteResult>)>)> = Vec::new();
+    let mut groups: Vec<(VolumeId, VolumeBatch)> = Vec::new();
 
     for req in batch {
         let vid = req.volume_id;
         if let Some(group) = groups.iter_mut().find(|(v, _)| *v == vid) {
-            group.1.push((req.needle, req.response_tx));
+            group.1.push((req.needle, req.fsync, req.response_tx));
         } else {
-            groups.push((vid, vec![(req.needle, req.response_tx)]));
+            groups.push((vid, vec![(req.needle, req.fsync, req.response_tx)]));
         }
     }
 
@@ -153,8 +159,8 @@ fn process_batch(state: Arc<VolumeServerState>, batch: Vec<WriteRequest>) {
     let mut store = state.store.write().unwrap();
 
     for (vid, entries) in groups {
-        for (mut needle, response_tx) in entries {
-            let result = store.write_volume_needle(vid, &mut needle);
+        for (mut needle, fsync, response_tx) in entries {
+            let result = store.write_volume_needle(vid, &mut needle, fsync);
             // Send result back; ignore error if receiver dropped.
             let _ = response_tx.send(result);
         }
@@ -239,7 +245,7 @@ mod tests {
             ..Needle::default()
         };
 
-        let result = queue.submit(VolumeId(999), needle).await;
+        let result = queue.submit(VolumeId(999), needle, false).await;
         assert!(result.is_err());
         match result {
             Err(VolumeError::NotFound) => {} // expected
@@ -264,7 +270,7 @@ mod tests {
                     data_size: 10,
                     ..Needle::default()
                 };
-                q.submit(VolumeId(1), needle).await
+                q.submit(VolumeId(1), needle, false).await
             }));
         }
 
@@ -293,7 +299,7 @@ mod tests {
                     data_size: 4,
                     ..Needle::default()
                 };
-                q.submit(VolumeId(42), needle).await
+                q.submit(VolumeId(42), needle, false).await
             }));
         }
 
@@ -327,7 +333,7 @@ mod tests {
             data_size: 0,
             ..Needle::default()
         };
-        let result = queue2.submit(VolumeId(1), needle).await;
+        let result = queue2.submit(VolumeId(1), needle, false).await;
         assert!(result.is_err()); // NotFound is fine -- the point is it doesn't panic
     }
 }
diff --git a/seaweed-volume/src/storage/erasure_coding/ec_decoder.rs b/seaweed-volume/src/storage/erasure_coding/ec_decoder.rs
index 428184538..38a7de561 100644
--- a/seaweed-volume/src/storage/erasure_coding/ec_decoder.rs
+++ b/seaweed-volume/src/storage/erasure_coding/ec_decoder.rs
@@ -400,7 +400,7 @@ mod tests {
                 data_size: data.len() as u32,
                 ..Needle::default()
             };
-            v.write_needle(&mut n, true).unwrap();
+            v.write_needle(&mut n, true, false).unwrap();
         }
         v.sync_to_disk().unwrap();
         let original_dat_size = v.dat_file_size().unwrap();
diff --git a/seaweed-volume/src/storage/erasure_coding/ec_encoder.rs b/seaweed-volume/src/storage/erasure_coding/ec_encoder.rs
index 15b944483..346b2fd1f 100644
--- a/seaweed-volume/src/storage/erasure_coding/ec_encoder.rs
+++ b/seaweed-volume/src/storage/erasure_coding/ec_encoder.rs
@@ -746,7 +746,7 @@ mod tests {
                 data_size: data.len() as u32,
                 ..Needle::default()
             };
-            v.write_needle(&mut n, true).unwrap();
+            v.write_needle(&mut n, true, false).unwrap();
         }
         v.sync_to_disk().unwrap();
         v.close();
@@ -798,7 +798,7 @@ mod tests {
                 data_size: data.len() as u32,
                 ..Needle::default()
             };
-            v.write_needle(&mut needle, true).unwrap();
+            v.write_needle(&mut needle, true, false).unwrap();
         }
         v.sync_to_disk().unwrap();
         v.close();
@@ -875,7 +875,7 @@ mod tests {
                 data_size: data.len() as u32,
                 ..Needle::default()
             };
-            v.write_needle(&mut n, true).unwrap();
+            v.write_needle(&mut n, true, false).unwrap();
         }
         v.sync_to_disk().unwrap();
         v.close();
@@ -1116,7 +1116,7 @@ mod tests {
                 data_size: data.len() as u32,
                 ..Needle::default()
             };
-            v.write_needle(&mut n, true).unwrap();
+            v.write_needle(&mut n, true, false).unwrap();
         }
         v.sync_to_disk().unwrap();
         v.close();
@@ -1198,7 +1198,7 @@ mod tests {
             data_size: 5,
             ..Needle::default()
         };
-        v.write_needle(&mut n, true).unwrap();
+        v.write_needle(&mut n, true, false).unwrap();
         v.sync_to_disk().unwrap();
         v.close();
 
diff --git a/seaweed-volume/src/storage/erasure_coding/ec_volume.rs b/seaweed-volume/src/storage/erasure_coding/ec_volume.rs
index 261574073..3f1717a79 100644
--- a/seaweed-volume/src/storage/erasure_coding/ec_volume.rs
+++ b/seaweed-volume/src/storage/erasure_coding/ec_volume.rs
@@ -1509,7 +1509,7 @@ mod tests {
                 data_size: data.len() as u32,
                 ..Needle::default()
             };
-            v.write_needle(&mut n, true).unwrap();
+            v.write_needle(&mut n, true, false).unwrap();
         }
         v.sync_to_disk().unwrap();
         v.close();
@@ -1585,7 +1585,7 @@ mod tests {
                 data_size: data.len() as u32,
                 ..Needle::default()
             };
-            v.write_needle(&mut n, true).unwrap();
+            v.write_needle(&mut n, true, false).unwrap();
         }
         v.sync_to_disk().unwrap();
         v.close();
@@ -1632,7 +1632,7 @@ mod tests {
                 data_size: data.len() as u32,
                 ..Needle::default()
             };
-            v.write_needle(&mut n, true).unwrap();
+            v.write_needle(&mut n, true, false).unwrap();
         }
         v.sync_to_disk().unwrap();
         v.close();
diff --git a/seaweed-volume/src/storage/store.rs b/seaweed-volume/src/storage/store.rs
index 789e9a330..e51d2e989 100644
--- a/seaweed-volume/src/storage/store.rs
+++ b/seaweed-volume/src/storage/store.rs
@@ -519,11 +519,13 @@ impl Store {
         vol.re_lookup_needle_data_offset(needle_id)
     }
 
-    /// Write a needle to a volume.
+    /// Write a needle to a volume. With `fsync` the volume flushes its .dat
+    /// before returning, so the caller can ack a durable write.
     pub fn write_volume_needle(
         &mut self,
         vid: VolumeId,
         n: &mut Needle,
+        fsync: bool,
     ) -> Result<(u64, Size, bool), VolumeError> {
         // Check disk space on the location containing this volume.
         // We do this before the mutable borrow to avoid borrow conflicts.
@@ -539,7 +541,7 @@ impl Store {
         }
 
         let (_, vol) = self.find_volume_mut(vid).ok_or(VolumeError::NotFound)?;
-        vol.write_needle(n, true)
+        vol.write_needle(n, true, fsync)
     }
 
     /// Delete a needle from a volume.
@@ -1385,7 +1387,9 @@ mod tests {
             data_size: 11,
             ..Needle::default()
         };
-        let (offset, _size, unchanged) = store.write_volume_needle(VolumeId(1), &mut n).unwrap();
+        let (offset, _size, unchanged) = store
+            .write_volume_needle(VolumeId(1), &mut n, false)
+            .unwrap();
         assert!(!unchanged);
         assert!(offset > 0);
 
@@ -1455,7 +1459,9 @@ mod tests {
             data_size: 10,
             ..Needle::default()
         };
-        store.write_volume_needle(VolumeId(1), &mut n).unwrap();
+        store
+            .write_volume_needle(VolumeId(1), &mut n, false)
+            .unwrap();
 
         // add_volume already placed the index in the -dir.idx directory, so
         // consolidation has nothing to move and leaves the volume readable.
diff --git a/seaweed-volume/src/storage/volume.rs b/seaweed-volume/src/storage/volume.rs
index 363ac0cb4..cb4ef1208 100644
--- a/seaweed-volume/src/storage/volume.rs
+++ b/seaweed-volume/src/storage/volume.rs
@@ -22,7 +22,9 @@ use tracing::{error, info, warn};
 #[cfg(test)]
 use crate::storage::idx;
 use crate::storage::needle::needle::{self, get_actual_size, Needle, NeedleError};
-use crate::storage::needle_map::{CompactNeedleMap, NeedleMap, NeedleMapKind, RedbNeedleMap};
+use crate::storage::needle_map::{
+    CompactNeedleMap, NeedleMap, NeedleMapKind, NeedleValue, RedbNeedleMap,
+};
 use crate::storage::super_block::{ReplicaPlacement, SuperBlock, SUPER_BLOCK_SIZE};
 use crate::storage::types::*;
 
@@ -507,6 +509,10 @@ pub struct Volume {
     dat_file: Option<File>,
     remote_dat_file: Option<RemoteDatFile>,
     nm: Option<NeedleMap>,
+    /// Makes the next .dat flush fail, so tests can exercise the durable
+    /// write's failure path without a real disk fault.
+    #[cfg(test)]
+    fail_fsync_for_test: bool,
     needle_map_kind: NeedleMapKind,
     data_file_access_control: Arc<DataFileAccessControl>,
 
@@ -581,6 +587,8 @@ impl Volume {
             collection: collection.to_string(),
             dat_file: None,
             remote_dat_file: None,
+            #[cfg(test)]
+            fail_fsync_for_test: false,
             nm: None,
             needle_map_kind,
             data_file_access_control: Arc::new(DataFileAccessControl::default()),
@@ -619,6 +627,8 @@ impl Volume {
             collection: collection.to_string(),
             dat_file: None,
             remote_dat_file: None,
+            #[cfg(test)]
+            fail_fsync_for_test: false,
             nm: None,
             needle_map_kind: NeedleMapKind::InMemory,
             data_file_access_control: Arc::new(DataFileAccessControl::default()),
@@ -1523,23 +1533,45 @@ impl Volume {
     // ---- Write ----
 
     /// Write a needle to the volume (synchronous path).
+    /// Write a needle to the volume. With `fsync` the .dat is flushed before
+    /// returning, so an ack means the data is on disk and not just in the page
+    /// cache. The flush happens under the same lock as the append, so a failed
+    /// one can take its own append back off the end and nobody else's.
     pub fn write_needle(
         &mut self,
         n: &mut Needle,
         check_cookie: bool,
+        fsync: bool,
     ) -> Result<(u64, Size, bool), VolumeError> {
         let _guard = self.data_file_access_control.write_lock();
         if self.is_read_only() {
             return Err(VolumeError::ReadOnly);
         }
 
-        self.do_write_request(n, check_cookie)
+        self.do_write_request(n, check_cookie, fsync)
+    }
+
+    /// Flush the .dat. The index is left out on purpose: it is rebuilt from the
+    /// .dat, so a durable write only has to get the data down.
+    fn flush_dat(&self) -> io::Result<()> {
+        #[cfg(test)]
+        if self.fail_fsync_for_test {
+            return Err(io::Error::new(
+                io::ErrorKind::Other,
+                "injected fsync failure",
+            ));
+        }
+        match self.dat_file.as_ref() {
+            Some(dat_file) => dat_file.sync_all(),
+            None => Err(io::Error::new(io::ErrorKind::Other, "dat file not open")),
+        }
     }
 
     fn do_write_request(
         &mut self,
         n: &mut Needle,
         check_cookie: bool,
+        fsync: bool,
     ) -> Result<(u64, Size, bool), VolumeError> {
         // TTL inheritance from volume (matching Go's writeNeedle2)
         {
@@ -1559,6 +1591,14 @@ impl Volume {
         // Dedup check (matches Go: n.DataSize = oldNeedle.DataSize on dedup)
         if let Some(old_data_size) = self.is_file_unchanged(n) {
             n.data_size = old_data_size;
+            // Nothing to append, but an earlier write may have left this content
+            // in the page cache, and the caller is asking for it to be on disk.
+            if fsync {
+                self.flush_dat().map_err(|e| {
+                    self.check_read_write_error(Some(&e));
+                    VolumeError::Io(e)
+                })?;
+            }
             return Ok((0, Size(n.data_size as i32), true));
         }
 
@@ -1583,6 +1623,34 @@ impl Volume {
 
         // Append to .dat file
         let (offset, _body_size, _actual_size) = self.append_needle(n)?;
+
+        // Nothing is published until the bytes are down: an index entry for an
+        // unflushed append would resolve past the end of the file after a crash,
+        // and undoing it afterwards would double-count the volume's metrics.
+        if fsync {
+            if let Err(e) = self.flush_dat() {
+                self.check_read_write_error(Some(&e));
+                let truncated = match self.dat_file.as_ref() {
+                    Some(dat_file) => dat_file.set_len(offset),
+                    None => Ok(()),
+                };
+                if let Err(te) = truncated {
+                    // The rejected record is still on the end. A later append
+                    // would bury it mid-file, where the .dat tail check cannot
+                    // see it, so stop taking writes instead.
+                    self.no_write_or_delete = true;
+                    tracing::error!(
+                        "volume {}: failed to truncate back to {} after a failed fsync, \
+                         marking read only: {}",
+                        self.id.0,
+                        offset,
+                        te
+                    );
+                }
+                return Err(VolumeError::Io(e));
+            }
+        }
+
         self.last_append_at_ns = n.append_at_ns;
 
         // Update needle map (uses n.size = full body size, matching Go's nm.Put)
@@ -1601,6 +1669,19 @@ impl Volume {
             }
         }
 
+        // load() rebuilds the map from .idx, not from .dat, so the row has to be
+        // down before the write is acked. Lose it and the volume comes back with
+        // a .dat tail the integrity check cannot account for, and loads read only
+        // - a worse outcome than losing the write.
+        if fsync {
+            if let Some(nm) = self.nm.as_ref() {
+                if let Err(e) = nm.sync() {
+                    self.check_read_write_error(Some(&e));
+                    return Err(VolumeError::Io(e));
+                }
+            }
+        }
+
         if self.last_modified_ts_seconds < n.last_modified {
             self.last_modified_ts_seconds = n.last_modified;
         }
@@ -3673,6 +3754,11 @@ impl Volume {
         }
     }
 
+    #[cfg(test)]
+    pub(crate) fn fail_next_fsync_for_test(&mut self, fail: bool) {
+        self.fail_fsync_for_test = fail;
+    }
+
     #[cfg(test)]
     pub(crate) fn set_last_modified_ts_for_test(&mut self, ts_seconds: u64) {
         self.last_modified_ts_seconds = ts_seconds;
@@ -4057,7 +4143,7 @@ mod tests {
             flags: 0,
             ..Needle::default()
         };
-        let (offset, size, unchanged) = v.write_needle(&mut n, true).unwrap();
+        let (offset, size, unchanged) = v.write_needle(&mut n, true, false).unwrap();
         assert!(!unchanged);
         assert!(offset > 0); // after superblock
         assert!(size.0 > 0);
@@ -4074,6 +4160,149 @@ mod tests {
         assert_eq!(read_n.cookie, Cookie(0x12345678));
     }
 
+    /// A durable write goes down the same path and lands the same data; the
+    /// flush is extra work, not different work.
+    #[test]
+    fn test_volume_write_needle_fsync() {
+        let tmp = TempDir::new().unwrap();
+        let dir = tmp.path().to_str().unwrap();
+        let mut v = make_test_volume(dir);
+
+        let mut n = Needle {
+            id: NeedleId(1),
+            cookie: Cookie(0x12345678),
+            data: b"durable payload".to_vec(),
+            data_size: 15,
+            flags: 0,
+            ..Needle::default()
+        };
+        let (offset, _, unchanged) = v.write_needle(&mut n, true, true).unwrap();
+        assert!(!unchanged);
+        assert!(offset > 0);
+
+        let mut read_n = Needle {
+            id: NeedleId(1),
+            ..Needle::default()
+        };
+        assert_eq!(v.read_needle(&mut read_n).unwrap(), 15);
+        assert_eq!(read_n.data, b"durable payload");
+
+        // rewriting the same content still dedups, so there is nothing to flush
+        let mut same = Needle {
+            id: NeedleId(1),
+            cookie: Cookie(0x12345678),
+            data: b"durable payload".to_vec(),
+            data_size: 15,
+            flags: 0,
+            ..Needle::default()
+        };
+        let (_, _, unchanged) = v.write_needle(&mut same, true, true).unwrap();
+        assert!(unchanged);
+    }
+
+    /// A durable write whose flush fails must leave nothing behind: the append
+    /// comes off the .dat and the mapping it would have replaced still stands,
+    /// with the volume's own accounting untouched.
+    #[test]
+    fn test_write_needle_failed_fsync_keeps_prior_mapping() {
+        let tmp = TempDir::new().unwrap();
+        let dir = tmp.path().to_str().unwrap();
+        let mut v = make_test_volume(dir);
+
+        let mut kept = Needle {
+            id: NeedleId(1),
+            cookie: Cookie(0xaa),
+            data: b"first-copy".to_vec(),
+            data_size: 10,
+            ..Needle::default()
+        };
+        v.write_needle(&mut kept, true, true).unwrap();
+        let prior = v.nm.as_ref().unwrap().get(NeedleId(1)).unwrap();
+        let dat_len_before = std::fs::metadata(v.file_name(".dat")).unwrap().len();
+        let file_count_before = v.file_count();
+        let content_size_before = v.content_size();
+
+        v.fail_next_fsync_for_test(true);
+        let mut replacement = Needle {
+            id: NeedleId(1),
+            cookie: Cookie(0xaa),
+            data: b"second-copy".to_vec(),
+            data_size: 11,
+            ..Needle::default()
+        };
+        v.write_needle(&mut replacement, true, true).unwrap_err();
+        v.fail_next_fsync_for_test(false);
+
+        assert_eq!(
+            std::fs::metadata(v.file_name(".dat")).unwrap().len(),
+            dat_len_before,
+            "the unflushed append should be off the .dat"
+        );
+        let now = v.nm.as_ref().unwrap().get(NeedleId(1)).unwrap();
+        assert_eq!(
+            now.offset, prior.offset,
+            "the mapping should never have moved"
+        );
+        assert_eq!(now.size, prior.size);
+        assert_eq!(
+            v.file_count(),
+            file_count_before,
+            "a rejected write must not count towards the volume"
+        );
+        assert_eq!(v.content_size(), content_size_before);
+
+        let mut read_n = Needle {
+            id: NeedleId(1),
+            ..Needle::default()
+        };
+        assert_eq!(v.read_needle(&mut read_n).unwrap(), 10);
+        assert_eq!(read_n.data, b"first-copy");
+    }
+
+    /// Same for a needle the failed write introduced: it never becomes visible,
+    /// rather than being published and then tombstoned back out.
+    #[test]
+    fn test_write_needle_failed_fsync_publishes_nothing() {
+        let tmp = TempDir::new().unwrap();
+        let dir = tmp.path().to_str().unwrap();
+        let mut v = make_test_volume(dir);
+
+        let dat_len_before = std::fs::metadata(v.file_name(".dat")).unwrap().len();
+        let file_count_before = v.file_count();
+
+        v.fail_next_fsync_for_test(true);
+        let mut fresh = Needle {
+            id: NeedleId(7),
+            cookie: Cookie(0x77),
+            data: b"never-landed".to_vec(),
+            data_size: 12,
+            ..Needle::default()
+        };
+        v.write_needle(&mut fresh, true, true).unwrap_err();
+        v.fail_next_fsync_for_test(false);
+
+        assert_eq!(
+            std::fs::metadata(v.file_name(".dat")).unwrap().len(),
+            dat_len_before
+        );
+        assert!(
+            v.nm.as_ref().unwrap().get(NeedleId(7)).is_none(),
+            "a needle that never reached the disk must not be indexed at all"
+        );
+        assert_eq!(v.file_count(), file_count_before);
+        assert_eq!(
+            v.deleted_count(),
+            0,
+            "nothing was written, so nothing was deleted"
+        );
+
+        let mut read_n = Needle {
+            id: NeedleId(7),
+            ..Needle::default()
+        };
+        assert!(v.read_needle(&mut read_n).is_err());
+    }
+
     #[test]
     fn test_volume_write_dedup() {
         let tmp = TempDir::new().unwrap();
@@ -4087,7 +4316,7 @@ mod tests {
             data_size: 9,
             ..Needle::default()
         };
-        v.write_needle(&mut n, true).unwrap();
+        v.write_needle(&mut n, true, false).unwrap();
 
         // Write same needle again — should be unchanged
         let mut n2 = Needle {
@@ -4098,7 +4327,7 @@ mod tests {
             ..Needle::default()
         };
         n2.checksum = CRC::new(&n2.data);
-        let (_, _, unchanged) = v.write_needle(&mut n2, true).unwrap();
+        let (_, _, unchanged) = v.write_needle(&mut n2, true, false).unwrap();
         assert!(unchanged);
     }
 
@@ -4115,7 +4344,7 @@ mod tests {
             data_size: 9,
             ..Needle::default()
         };
-        v.write_needle(&mut n, true).unwrap();
+        v.write_needle(&mut n, true, false).unwrap();
         assert_eq!(v.file_count(), 1);
 
         let deleted_size = v
@@ -4161,7 +4390,7 @@ mod tests {
                     data_size: data.len() as u32,
                     ..Needle::default()
                 };
-                v.write_needle(&mut n, true).unwrap();
+                v.write_needle(&mut n, true, false).unwrap();
             }
             v.delete_needle(&mut Needle {
                 id: NeedleId(2),
@@ -4201,7 +4430,7 @@ mod tests {
                 data_size: data.len() as u32,
                 ..Needle::default()
             };
-            v.write_needle(&mut n, true).unwrap();
+            v.write_needle(&mut n, true, false).unwrap();
         }
         v.sync_to_disk().unwrap();
     }
@@ -4274,7 +4503,7 @@ mod tests {
                     data_size: data.len() as u32,
                     ..Needle::default()
                 };
-                v.write_needle(&mut n, true).unwrap();
+                v.write_needle(&mut n, true, false).unwrap();
             }
             v.sync_to_disk().unwrap();
         }
@@ -4309,7 +4538,7 @@ mod tests {
                 data_size: data.len() as u32,
                 ..Needle::default()
             };
-            v.write_needle(&mut n, true).unwrap();
+            v.write_needle(&mut n, true, false).unwrap();
             v.delete_needle(&mut Needle {
                 id: NeedleId(1),
                 cookie: Cookie(1),
@@ -4430,7 +4659,7 @@ mod tests {
                 data_size: body.len() as u32,
                 ..Needle::default()
             };
-            v.write_needle(&mut n, true).unwrap();
+            v.write_needle(&mut n, true, false).unwrap();
         }
         v.sync_to_disk().unwrap();
 
@@ -4460,7 +4689,7 @@ mod tests {
             data_size: data.len() as u32,
             ..Needle::default()
         };
-        v.write_needle(&mut n, true).unwrap();
+        v.write_needle(&mut n, true, false).unwrap();
         v.sync_to_disk().unwrap();
 
         // Append an offset-0 logical tombstone to the on-disk .idx.
@@ -4498,7 +4727,7 @@ mod tests {
             data_size: data.len() as u32,
             ..Needle::default()
         };
-        v.write_needle(&mut n, true).unwrap();
+        v.write_needle(&mut n, true, false).unwrap();
         v.sync_to_disk().unwrap();
         assert!(v.dat_file_size().unwrap() > SUPER_BLOCK_SIZE as u64);
 
@@ -4531,7 +4760,7 @@ mod tests {
                 data_size: data.len() as u32,
                 ..Needle::default()
             };
-            v.write_needle(&mut n, true).unwrap();
+            v.write_needle(&mut n, true, false).unwrap();
         }
         v.sync_to_disk().unwrap();
 
@@ -4559,7 +4788,7 @@ mod tests {
                 data_size: data.len() as u32,
                 ..Needle::default()
             };
-            v.write_needle(&mut n, true).unwrap();
+            v.write_needle(&mut n, true, false).unwrap();
         }
 
         assert_eq!(v.file_count(), 10);
@@ -4591,7 +4820,7 @@ mod tests {
                     data_size: data.len() as u32,
                     ..Needle::default()
                 };
-                v.write_needle(&mut n, true).unwrap();
+                v.write_needle(&mut n, true, false).unwrap();
             }
             v.sync_to_disk().unwrap();
         }
@@ -4651,7 +4880,7 @@ mod tests {
             data_size: payload.len() as u32,
             ..Needle::default()
         };
-        v.write_needle(&mut n, true).unwrap();
+        v.write_needle(&mut n, true, false).unwrap();
         v.sync_to_disk().unwrap();
 
         let data_idx = format!("{data}/7.idx");
@@ -4695,7 +4924,7 @@ mod tests {
             data_size: 1,
             ..Needle::default()
         };
-        v.write_needle(&mut n, true).unwrap();
+        v.write_needle(&mut n, true, false).unwrap();
         v.sync_to_disk().unwrap();
 
         // Data and index share a directory, so there is nothing to move.
@@ -4722,7 +4951,7 @@ mod tests {
             data_size: 8,
             ..Needle::default()
         };
-        v.write_needle(&mut n, true).unwrap();
+        v.write_needle(&mut n, true, false).unwrap();
 
         // Write with wrong cookie
         let mut n2 = Needle {
@@ -4732,7 +4961,7 @@ mod tests {
             data_size: 9,
             ..Needle::default()
         };
-        let err = v.write_needle(&mut n2, true).unwrap_err();
+        let err = v.write_needle(&mut n2, true, false).unwrap_err();
         assert!(matches!(err, VolumeError::CookieMismatch(_)));
     }
 
@@ -4752,7 +4981,7 @@ mod tests {
             ..Needle::default()
         };
         n.checksum = CRC::new(&n.data);
-        let (offset, _, _) = v.write_needle(&mut n, true).unwrap();
+        let (offset, _, _) = v.write_needle(&mut n, true, false).unwrap();
         let blob = v.read_needle_blob(offset as i64, n.size).unwrap();
 
         let dat_size_before = v.dat_file_size().unwrap();
@@ -4801,7 +5030,7 @@ mod tests {
             data_size: 5,
             ..Needle::default()
         };
-        v.write_needle(&mut first, true).unwrap();
+        v.write_needle(&mut first, true, false).unwrap();
 
         let mut second = Needle {
             id: NeedleId(20),
@@ -4810,7 +5039,7 @@ mod tests {
             data_size: 6,
             ..Needle::default()
         };
-        v.write_needle(&mut second, true).unwrap();
+        v.write_needle(&mut second, true, false).unwrap();
 
         let mut first_overwrite = Needle {
             id: NeedleId(10),
@@ -4819,7 +5048,7 @@ mod tests {
             data_size: 15,
             ..Needle::default()
         };
-        v.write_needle(&mut first_overwrite, true).unwrap();
+        v.write_needle(&mut first_overwrite, true, false).unwrap();
 
         let needles = v.read_all_needles().unwrap();
         let ids: Vec<u64> = needles.iter().map(|n| u64::from(n.id)).collect();
@@ -4856,7 +5085,7 @@ mod tests {
                 data_size: format!("data-{}", i).len() as u32,
                 ..Needle::default()
             };
-            v.write_needle(&mut n, true).unwrap();
+            v.write_needle(&mut n, true, false).unwrap();
         }
         assert_eq!(v.file_count(), 3);
 
@@ -4943,7 +5172,7 @@ mod tests {
                 data_size: format!("payload-{}", i).len() as u32,
                 ..Needle::default()
             };
-            v.write_needle(&mut n, true).unwrap();
+            v.write_needle(&mut n, true, false).unwrap();
         }
         v.sync_to_disk().unwrap();
 
@@ -5010,7 +5239,7 @@ mod tests {
                 data_size: format!("data-{}", i).len() as u32,
                 ..Needle::default()
             };
-            v.write_needle(&mut n, true).unwrap();
+            v.write_needle(&mut n, true, false).unwrap();
         }
         v.sync_to_disk().unwrap();
 
@@ -5072,7 +5301,7 @@ mod tests {
             data_size: 17,
             ..Needle::default()
         };
-        v.write_needle(&mut n1, true).unwrap();
+        v.write_needle(&mut n1, true, false).unwrap();
 
         let mut n2 = Needle {
             id: NeedleId(2),
@@ -5081,7 +5310,7 @@ mod tests {
             data_size: 18,
             ..Needle::default()
         };
-        v.write_needle(&mut n2, true).unwrap();
+        v.write_needle(&mut n2, true, false).unwrap();
 
         // Get initial revision and offset for needle 1
         let initial_rev = v.super_block.compaction_revision;
@@ -5147,7 +5376,7 @@ mod tests {
             data_size: data.len() as u32,
             ..Needle::default()
         };
-        v.write_needle(&mut n, true).unwrap();
+        v.write_needle(&mut n, true, false).unwrap();
 
         // Read stream info
         let mut read_n = Needle {
@@ -5178,7 +5407,7 @@ mod tests {
                 data_size: 6,
                 ..Needle::default()
             };
-            v.write_needle(&mut n, true).unwrap();
+            v.write_needle(&mut n, true, false).unwrap();
 
             let vif = VifVolumeInfo {
                 files: vec![VifRemoteFile {
@@ -5250,6 +5479,7 @@ mod tests {
                     ..Needle::default()
                 },
                 true,
+                false,
             )
             .unwrap_err();
         assert!(matches!(err, VolumeError::ReadOnly));
@@ -5312,7 +5542,7 @@ mod tests {
                 data_size: 7,
                 ..Needle::default()
             };
-            v.write_needle(&mut n, true).unwrap();
+            v.write_needle(&mut n, true, false).unwrap();
             v.set_read_only_persist(false, true).unwrap();
             v.sync_to_disk().unwrap();
         }
@@ -5352,7 +5582,7 @@ mod tests {
             data_size: 19,
             ..Needle::default()
         };
-        v.write_needle(&mut n, true).unwrap();
+        v.write_needle(&mut n, true, false).unwrap();
         v.sync_to_disk().unwrap();
 
         // Reload one more time — the .idx must contain the post-mark-writable
@@ -5395,7 +5625,7 @@ mod tests {
                     data_size: 7,
                     ..Needle::default()
                 };
-                v.write_needle(&mut n, true).unwrap();
+                v.write_needle(&mut n, true, false).unwrap();
             }
             v.set_read_only_persist(true, true).unwrap();
             assert!(v.no_write_can_delete);
@@ -5412,6 +5642,7 @@ mod tests {
                         ..Needle::default()
                     },
                     true,
+                    false,
                 )
                 .unwrap_err();
             assert!(matches!(err, VolumeError::ReadOnly));
@@ -5456,6 +5687,7 @@ mod tests {
                     ..Needle::default()
                 },
                 true,
+                false,
             )
             .unwrap_err();
         assert!(matches!(err, VolumeError::ReadOnly));
@@ -5484,7 +5716,7 @@ mod tests {
             data_size: 14,
             ..Needle::default()
         };
-        v.write_needle(&mut n, true).unwrap();
+        v.write_needle(&mut n, true, false).unwrap();
     }
 
     // Upgrading a volume that booted plain persisted-readonly to canDelete
@@ -5503,7 +5735,7 @@ mod tests {
                 data_size: 7,
                 ..Needle::default()
             };
-            v.write_needle(&mut n, true).unwrap();
+            v.write_needle(&mut n, true, false).unwrap();
             v.set_read_only_persist(false, true).unwrap();
             v.sync_to_disk().unwrap();
         }
@@ -5678,7 +5910,7 @@ mod tests {
                 data_size: 11,
                 ..Needle::default()
             };
-            v.write_needle(&mut n, true).unwrap();
+            v.write_needle(&mut n, true, false).unwrap();
             v.sync_to_disk().unwrap();
             std::fs::read(v.file_name(".dat")).unwrap()
         };
@@ -5789,7 +6021,7 @@ mod tests {
             data_size: 4,
             ..Needle::default()
         };
-        v.write_needle(&mut n, true).unwrap();
+        v.write_needle(&mut n, true, false).unwrap();
 
         // Write a .vif file (as EC encode would)
         let vif_path = format!("{}/1.vif", dir);
@@ -5849,7 +6081,7 @@ mod tests {
             data_size: 5,
             ..Needle::default()
         };
-        v.write_needle(&mut n, true).unwrap();
+        v.write_needle(&mut n, true, false).unwrap();
 
         // Write .vif in data dir (as EC encode would)
         let vif_path = format!("{}/1.vif", dat_dir);
@@ -5891,7 +6123,7 @@ mod tests {
             data_size: 4,
             ..Needle::default()
         };
-        v.write_needle(&mut n, true).unwrap();
+        v.write_needle(&mut n, true, false).unwrap();
 
         let vif_path = format!("{}/1.vif", dir);
         std::fs::write(&vif_path, r#"{"version":3}"#).unwrap();
@@ -5933,7 +6165,7 @@ mod tests {
                 data_size: format!("data-{}", i).len() as u32,
                 ..Needle::default()
             };
-            v.write_needle(&mut n, true).unwrap();
+            v.write_needle(&mut n, true, false).unwrap();
         }
         for id in [2u64, 5u64] {
             let mut del = Needle {
@@ -5998,7 +6230,7 @@ mod tests {
                 data_size: format!("data-{}", i).len() as u32,
                 ..Needle::default()
             };
-            v.write_needle(&mut n, true).unwrap();
+            v.write_needle(&mut n, true, false).unwrap();
         }
         v.compact_by_index(0, 0, |_| true).unwrap();
 
@@ -6034,7 +6266,7 @@ mod tests {
                 data_size: format!("data-{}", i).len() as u32,
                 ..Needle::default()
             };
-            v.write_needle(&mut n, true).unwrap();
+            v.write_needle(&mut n, true, false).unwrap();
         }
         v.sync_to_disk().unwrap();
 
diff --git a/seaweed-volume/src/storage/volume_idx_repair.rs b/seaweed-volume/src/storage/volume_idx_repair.rs
index 612341b96..d4dfef3fa 100644
--- a/seaweed-volume/src/storage/volume_idx_repair.rs
+++ b/seaweed-volume/src/storage/volume_idx_repair.rs
@@ -246,7 +246,7 @@ mod tests {
                 data: data.clone(),
                 ..Needle::default()
             };
-            v.write_needle(&mut n, true).unwrap();
+            v.write_needle(&mut n, true, false).unwrap();
             written.push(data);
         }
         written
diff --git a/seaweed-volume/tests/http_integration.rs b/seaweed-volume/tests/http_integration.rs
index 521c25d2b..7a2bea586 100644
--- a/seaweed-volume/tests/http_integration.rs
+++ b/seaweed-volume/tests/http_integration.rs
@@ -385,6 +385,46 @@ async fn write_then_read_needle() {
     assert_eq!(body, payload, "GET body should match written data");
 }
 
+// A durable upload takes the same route through the handler; the flush is not
+// observable from here, but a broken wiring would show up as a failed write.
+#[tokio::test]
+async fn write_with_fsync_then_read_needle() {
+    let (state, _tmp) = test_state();
+
+    let uri = "/1,01637037d6?fsync=true";
+    let payload = b"durable through the handler";
+
+    let app = build_admin_router(state.clone());
+    let response = app
+        .oneshot(
+            Request::builder()
+                .method("POST")
+                .uri(uri)
+                .body(Body::from(payload.to_vec()))
+                .unwrap(),
+        )
+        .await
+        .unwrap();
+    assert_eq!(
+        response.status(),
+        StatusCode::CREATED,
+        "a durable POST should return 201 Created"
+    );
+
+    let app = build_admin_router(state.clone());
+    let response = app
+        .oneshot(
+            Request::builder()
+                .uri("/1,01637037d6")
+                .body(Body::empty())
+                .unwrap(),
+        )
+        .await
+        .unwrap();
+    assert_eq!(response.status(), StatusCode::OK);
+    assert_eq!(body_bytes(response).await, payload);
+}
+
 // ============================================================================
 // 5. DELETE deletes a needle, subsequent GET returns 404
 // ============================================================================
@@ -739,6 +779,76 @@ async fn replicate_write_raw_body_is_stored() {
     assert_eq!(body_bytes(response).await, payload);
 }
 
+/// Go reads fsync through r.FormValue, which decodes the query, so a
+/// percent-encoded value has to reach the write path here too.
+#[tokio::test]
+async fn write_with_percent_encoded_fsync_is_accepted() {
+    let (state, _tmp) = test_state();
+    let payload = b"encoded durable payload";
+
+    let app = build_admin_router(state.clone());
+    let response = app
+        .oneshot(
+            Request::builder()
+                .method("POST")
+                .uri("/1,01637037d6?fsync=%74rue")
+                .body(Body::from(payload.to_vec()))
+                .unwrap(),
+        )
+        .await
+        .unwrap();
+    assert_eq!(response.status(), StatusCode::CREATED);
+
+    let app = build_admin_router(state.clone());
+    let response = app
+        .oneshot(
+            Request::builder()
+                .uri("/1,01637037d6")
+                .body(Body::empty())
+                .unwrap(),
+        )
+        .await
+        .unwrap();
+    assert_eq!(response.status(), StatusCode::OK);
+    assert_eq!(body_bytes(response).await, payload);
+}
+
+/// The fan-out query a Go primary sends for a durable write: `fsync=true` rides
+/// along with `type=replicate`, and the replica has to honor it rather than ack
+/// out of the page cache.
+#[tokio::test]
+async fn replicate_write_with_fsync_is_stored() {
+    let (state, _tmp) = test_state();
+    let uri = "/1,01637037d6?fsync=true&type=replicate";
+    let payload = b"durable replica bytes";
+
+    let app = build_admin_router(state.clone());
+    let response = app
+        .oneshot(
+            Request::builder()
+                .method("POST")
+                .uri(uri)
+                .body(Body::from(payload.to_vec()))
+                .unwrap(),
+        )
+        .await
+        .unwrap();
+    assert_eq!(response.status(), StatusCode::CREATED);
+
+    let app = build_admin_router(state.clone());
+    let response = app
+        .oneshot(
+            Request::builder()
+                .uri("/1,01637037d6")
+                .body(Body::empty())
+                .unwrap(),
+        )
+        .await
+        .unwrap();
+    assert_eq!(response.status(), StatusCode::OK);
+    assert_eq!(body_bytes(response).await, payload);
+}
+
 /// Multipart `type=replicate` write (the shape the Go gateway uploader sends)
 /// is stored and reads back.
 #[tokio::test]
@@ -882,7 +992,7 @@ async fn chunk_manifest_expands_chunk_stored_on_ec_volume() {
             data_size: chunk_data.len() as u32,
             ..Needle::default()
         };
-        v.write_needle(&mut n, true).unwrap();
+        v.write_needle(&mut n, true, false).unwrap();
         v.sync_to_disk().unwrap();
         v.close();
     }
@@ -916,7 +1026,9 @@ async fn chunk_manifest_expands_chunk_stored_on_ec_volume() {
         };
         n.data_size = n.data.len() as u32;
         n.set_is_chunk_manifest();
-        store.write_volume_needle(VolumeId(1), &mut n).unwrap();
+        store
+            .write_volume_needle(VolumeId(1), &mut n, false)
+            .unwrap();
     }
 
     // GET the manifest object; expect the reconstructed chunk bytes.