Batch standalone star write durability syncs

docs/chunk-consensus.md

@@ -400,7 +400,13 @@ One batch response is successful only after every member has a confirmed reply;
 code 16 exposes no partial digest vector. An uncertain member or mixed leadership
 result refuses the whole response.
 The members are not an atomic transaction. Failed publication never replays an
-uncertain write. Standalone journal nodes apply and sync members one at a time.
+uncertain write. A standalone journal node applies the bounded batch under one
+lock, appends one binary outcome per accepted page, and performs one durability
+sync before replying. At the maximum batch width this changes eight durability
+syncs into one without changing the eight recoverable outcomes. If a later
+member is refused, the accepted prefix is synced before the error is returned.
+Recovery can therefore observe that prefix;
+the shared sync reduces write amplification but does not make the pages atomic.
 This reduces request admission overhead for small object pages; it does not
 reduce replicated page payload bytes or physical storage writes.
 

src/cluster/replica_journal.rs

@@ -18,6 +18,8 @@ pub(super) struct ReplicaJournal {
     _lock: Option<File>,
     path: PathBuf,
     bytes_since_checkpoint: u64,
+    #[cfg(test)]
+    sync_count: u64,
 }
 
 fn invalid(message: impl ToString) -> io::Error {
@@ -151,6 +153,8 @@ impl ReplicaJournal {
             _lock: Some(lock),
             path,
             bytes_since_checkpoint: offset - base_end,
+            #[cfg(test)]
+            sync_count: 0,
         };
         if offset != 0 {
             journal.mark_initialized()?;
@@ -187,16 +191,37 @@ impl ReplicaJournal {
     }
 
     pub(super) fn append(&mut self, record: &[u8]) -> io::Result<()> {
+        self.append_deferred(record)?;
+        self.sync_deferred()
+    }
+
+    /// Append one complete outcome without acknowledging its durability yet.
+    /// The bounded standalone batch path uses this for every member, then calls
+    /// `sync_deferred` once before it returns a reply.
+    pub(super) fn append_deferred(&mut self, record: &[u8]) -> io::Result<()> {
         self.file.write_all(record)?;
-        self.file.sync_all()?;
-        self.mark_initialized()?;
         self.bytes_since_checkpoint = self
             .bytes_since_checkpoint
             .checked_add(record.len() as u64)
             .ok_or_else(|| invalid("journal length overflow"))?;
         Ok(())
     }
 
+    pub(super) fn sync_deferred(&mut self) -> io::Result<()> {
+        self.file.sync_all()?;
+        self.mark_initialized()?;
+        #[cfg(test)]
+        {
+            self.sync_count = self.sync_count.saturating_add(1);
+        }
+        Ok(())
+    }
+
+    #[cfg(test)]
+    pub(super) fn sync_count(&self) -> u64 {
+        self.sync_count
+    }
+
     fn mark_initialized(&self) -> io::Result<()> {
         if let Some(lock) = &self._lock {
             if lock.metadata()?.len() == 0 {
@@ -362,6 +387,7 @@ impl ReplicaJournal {
             _lock: None,
             path: path.to_path_buf(),
             bytes_since_checkpoint: 0,
+            sync_count: 0,
         }
     }
 }

src/cluster/replica_rpc_server.rs

@@ -441,10 +441,12 @@ fn frame_response(
                 };
                 #[cfg(not(feature = "chunk-consensus"))]
                 let result = match request {
-                    ReplicaRequest::WriteChunkBatchIfSequence { writes } =>
-                        apply_batch(replica, writes, options.max_payload_bytes),
-                    ReplicaRequest::WriteChunkBatchIfSequenceDigest { writes } =>
-                        apply_batch_digests(replica, writes, options.max_payload_bytes),
+                    ReplicaRequest::WriteChunkBatchIfSequence { writes } => {
+                        apply_batch(replica, writes, options.max_payload_bytes)
+                    }
+                    ReplicaRequest::WriteChunkBatchIfSequenceDigest { writes } => {
+                        apply_batch_digests(replica, writes, options.max_payload_bytes)
+                    }
                     request => apply(replica, request, options.max_payload_bytes),
                 };
                 match result.and_then(|reply| {
@@ -470,14 +472,32 @@ fn apply_batch(
     writes: Vec<super::message::ConditionalChunkWrite>,
     max_payload_bytes: usize,
 ) -> Result<ReplicaReply, ObjectError> {
+    if writes.is_empty() {
+        return Ok(ReplicaReply::Applied);
+    }
+    let mut guard = replica.lock().map_err(|_| {
+        ObjectError::Corruption(
+            "this replica's state was left locked by a request that panicked, so it can no longer be trusted to answer"
+                .to_string(),
+        )
+    })?;
     for write in writes {
-        apply(replica, ReplicaRequest::WriteChunkIfSequence {
-            page_id: write.page_id,
-            offset: write.offset,
-            data: write.data,
-            expected_sequence: write.expected_sequence,
-        }, max_payload_bytes)?;
+        if let Err(error) = apply_locked(
+            &mut guard,
+            ReplicaRequest::WriteChunkIfSequence {
+                page_id: write.page_id,
+                offset: write.offset,
+                data: write.data,
+                expected_sequence: write.expected_sequence,
+            },
+            max_payload_bytes,
+            false,
+        ) {
+            finish_deferred_journal(&mut guard)?;
+            return Err(error);
+        }
     }
+    finish_deferred_journal(&mut guard)?;
     Ok(ReplicaReply::Applied)
 }
 
@@ -486,33 +506,76 @@ fn apply_batch_digests(
     writes: Vec<super::message::ConditionalChunkWrite>,
     max_payload_bytes: usize,
 ) -> Result<ReplicaReply, ObjectError> {
+    if writes.is_empty() {
+        return Ok(ReplicaReply::Digests(Vec::new()));
+    }
+    let mut guard = replica.lock().map_err(|_| {
+        ObjectError::Corruption(
+            "this replica's state was left locked by a request that panicked, so it can no longer be trusted to answer"
+                .to_string(),
+        )
+    })?;
     let mut digests = Vec::with_capacity(writes.len());
     for write in writes {
-        match apply(
-            replica,
+        let reply = match apply_locked(
+            &mut guard,
             ReplicaRequest::WriteChunkIfSequenceDigest {
                 page_id: write.page_id,
                 offset: write.offset,
                 data: write.data,
                 expected_sequence: write.expected_sequence,
             },
             max_payload_bytes,
-        )? {
+            false,
+        ) {
+            Ok(reply) => reply,
+            Err(error) => {
+                finish_deferred_journal(&mut guard)?;
+                return Err(error);
+            }
+        };
+        match reply {
             ReplicaReply::Digest {
                 write_sequence,
                 length,
                 sha256,
             } => digests.push((write_sequence, length, sha256)),
             other => {
+                finish_deferred_journal(&mut guard)?;
                 return Err(ObjectError::Corruption(format!(
                     "unexpected conditional write digest reply: {other:?}"
                 )))
             }
         }
     }
+    finish_deferred_journal(&mut guard)?;
     Ok(ReplicaReply::Digests(digests))
 }
 
+fn finish_deferred_journal(guard: &mut NodeReplica) -> Result<(), ObjectError> {
+    let Some(journal) = guard.journal.as_mut() else {
+        return Ok(());
+    };
+    if let Err(error) = journal.sync_deferred() {
+        guard.failed = true;
+        return Err(ObjectError::Corruption(format!(
+            "replica outcome persistence failed: {error}"
+        )));
+    }
+    if journal.checkpoint_due(guard.journal_checkpoint_after_bytes) {
+        let NodeReplica {
+            service, journal, ..
+        } = guard;
+        if let Err(error) = journal.as_mut().unwrap().checkpoint(service) {
+            guard.failed = true;
+            return Err(ObjectError::Corruption(format!(
+                "replica outcome persistence failed: {error}"
+            )));
+        }
+    }
+    Ok(())
+}
+
 /// Run one request against the replica.
 ///
 /// Every arm is a call into [`MatrixReplicaService`]. Nothing here decides anything about
@@ -534,6 +597,15 @@ fn apply(
                 .to_string(),
         )
     })?;
+    apply_locked(&mut guard, request, max_payload_bytes, true)
+}
+
+fn apply_locked(
+    guard: &mut NodeReplica,
+    request: ReplicaRequest,
+    max_payload_bytes: usize,
+    sync_journal: bool,
+) -> Result<ReplicaReply, ObjectError> {
     if guard.failed {
         return Err(ObjectError::Corruption(
             "replica journal failed; restart and recover before serving requests".to_string(),
@@ -557,7 +629,7 @@ fn apply(
         ));
     }
     if let ReplicaRequest::StateInstall(request) = request {
-        return apply_state_install(&mut guard, request);
+        return apply_state_install(guard, request);
     }
     if let ReplicaRequest::ReadChunkStateRange { length, .. } = &request {
         let size = usize::try_from(*length)
@@ -651,14 +723,18 @@ fn apply(
                 );
             let record = ReplicaJournal::encode_outcome(&outcome)
                 .map_err(|err| ObjectError::Corruption(err.to_string()))?;
-            guard
-                .journal
-                .as_mut()
-                .unwrap()
-                .append(&record)
-                .map_err(|err| ObjectError::Corruption(err.to_string()))
+            let journal = guard.journal.as_mut().unwrap();
+            if sync_journal {
+                journal.append(&record)
+            } else {
+                journal.append_deferred(&record)
+            }
+            .map_err(|err| ObjectError::Corruption(err.to_string()))
         })();
         let persist = persist.and_then(|()| {
+            if !sync_journal {
+                return Ok(());
+            }
             if guard
                 .journal
                 .as_ref()
@@ -1051,6 +1127,98 @@ mod journal_failure_tests {
         assert_eq!(service.read_chunk(&page, 0, 8).unwrap().data, b"original");
     }
 
+    #[test]
+    fn standalone_star_batches_sync_once_and_recover_accepted_prefixes() {
+        let directory = std::env::temp_dir().join(format!(
+            "replica-star-batch-{}-{}",
+            std::process::id(),
+            std::time::SystemTime::now()
+                .duration_since(std::time::UNIX_EPOCH)
+                .unwrap()
+                .as_nanos()
+        ));
+        std::fs::create_dir(&directory).unwrap();
+        let path = directory.join("replica.wal");
+        let mut service = MatrixReplicaService::new();
+        let journal = ReplicaJournal::open(&path, &mut service).unwrap();
+        let replica = Arc::new(Mutex::new(NodeReplica {
+            used_bytes: 0,
+            service,
+            journal: Some(journal),
+            journal_checkpoint_after_bytes: 0,
+            staged: None,
+            maintenance: false,
+            max_pages: 65_536,
+            #[cfg(all(feature = "chunk-consensus", feature = "meta-server"))]
+            applied_membership: None,
+            failed: false,
+        }));
+        let make_page = |index| PageId {
+            chunk_id: ChunkId::new(7, 9).unwrap(),
+            page_index: index,
+        };
+        let write = |index, expected_sequence| super::super::message::ConditionalChunkWrite {
+            page_id: make_page(index),
+            offset: 0,
+            data: vec![index as u8 + 1; 16],
+            expected_sequence,
+        };
+
+        assert_eq!(
+            apply_batch(
+                &replica,
+                (0..8).map(|index| write(index, 0)).collect(),
+                MAX_PAYLOAD_BYTES,
+            )
+            .unwrap(),
+            ReplicaReply::Applied
+        );
+        assert_eq!(
+            replica
+                .lock()
+                .unwrap()
+                .journal
+                .as_ref()
+                .unwrap()
+                .sync_count(),
+            1
+        );
+
+        assert!(matches!(
+            apply_batch(
+                &replica,
+                vec![write(8, 0), write(9, 1)],
+                MAX_PAYLOAD_BYTES,
+            ),
+            Err(ObjectError::PreconditionFailed(_))
+        ));
+        assert_eq!(
+            replica
+                .lock()
+                .unwrap()
+                .journal
+                .as_ref()
+                .unwrap()
+                .sync_count(),
+            2
+        );
+        drop(replica);
+
+        let mut recovered = MatrixReplicaService::new();
+        drop(ReplicaJournal::open(&path, &mut recovered).unwrap());
+        for index in 0..9 {
+            assert_eq!(
+                recovered.read_chunk(&make_page(index), 0, 16).unwrap().data,
+                vec![index as u8 + 1; 16]
+            );
+        }
+        assert!(matches!(
+            recovered.head_chunk(&make_page(9)),
+            Err(ObjectError::NotFound(_))
+        ));
+        std::fs::remove_dir_all(directory).unwrap();
+    }
+
     #[test]
     fn failed_persistence_blocks_acknowledgment_and_all_later_requests() {
         let path = std::env::temp_dir().join(format!(