diff --git a/docs/chunk-consensus.md b/docs/chunk-consensus.md index 95d7bfac..baa9f361 100644 --- a/docs/chunk-consensus.md +++ b/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. diff --git a/src/cluster/replica_journal.rs b/src/cluster/replica_journal.rs index 9c0436d5..f5940e22 100644 --- a/src/cluster/replica_journal.rs +++ b/src/cluster/replica_journal.rs @@ -18,6 +18,8 @@ pub(super) struct ReplicaJournal { _lock: Option, 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,9 +191,15 @@ 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) @@ -197,6 +207,21 @@ impl ReplicaJournal { 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, } } } diff --git a/src/cluster/replica_rpc_server.rs b/src/cluster/replica_rpc_server.rs index 0e27d9da..926b5fa4 100644 --- a/src/cluster/replica_rpc_server.rs +++ b/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, max_payload_bytes: usize, ) -> Result { + 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,10 +506,19 @@ fn apply_batch_digests( writes: Vec, max_payload_bytes: usize, ) -> Result { + 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, @@ -497,22 +526,56 @@ fn apply_batch_digests( 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 { 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!(