You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@opendal.apache.org by xu...@apache.org on 2026/10/26 20:42:29 UTC
(opendal-yinyang) branch xuanwo/managed-sync-reconciliation updated: fix: apply resolution receipts from the selected artifact bytes
This is an automated email from the ASF dual-hosted git repository.
Xuanwo pushed a commit to branch xuanwo/managed-sync-reconciliation
in repository https://gitbox.apache.org/repos/asf/opendal-yinyang.git
The following commit(s) were added to refs/heads/xuanwo/managed-sync-reconciliation by this push:
new caf1a7e fix: apply resolution receipts from the selected artifact bytes
caf1a7e is described below
commit caf1a7e0aa39907aaef5369b425365a3637d5fed
Author: Xuanwo <gi...@xuanwo.io>
AuthorDate: Tue Oct 27 04:42:23 2026 +0800
fix: apply resolution receipts from the selected artifact bytes
A crash after a conditional resolution publication must install the retained
selected bytes against the exact pre-dispatch local fingerprint, confirm the
baseline only for the verified result, and keep a newer live local version
pending without adopting it. Also keep the remote helper TempDir alive with
its Sync client.
Co-authored-by: Chojan Shang <ps...@apache.org>
---
src/replica/directory.rs | 81 ++++++++++++++++++++++++++++++++-----
src/replica/directory/resolution.rs | 2 +-
tests/directory_resolution.rs | 15 +++----
3 files changed, 79 insertions(+), 19 deletions(-)
diff --git a/src/replica/directory.rs b/src/replica/directory.rs
index 926744d..21ffc7e 100644
--- a/src/replica/directory.rs
+++ b/src/replica/directory.rs
@@ -940,16 +940,28 @@ impl DirectorySync {
// Apply baseline and clear the linked conflict before
// marking the work complete. Publication is idempotent, so
// a crash here is recovered by re-running this retry.
- self.apply_retry_receipt(&record, revision).await?;
- if let Some(conflict) = &record.conflict {
- self.clear_resolved_conflict(conflict)?;
+ let applied = self.apply_retry_receipt(&record, revision).await?;
+ if applied {
+ if let Some(conflict) = &record.conflict {
+ self.clear_resolved_conflict(conflict)?;
+ }
+ self.replica.store_mut().update_work(
+ &record.id,
+ "committed",
+ Some(revision),
+ None,
+ )?;
+ } else {
+ // The remote receipt committed but the live local
+ // version moved on. Keep the association unresolved so
+ // the next run retries the idempotent decision.
+ self.replica.store_mut().update_work(
+ &record.id,
+ "prepared",
+ Some(revision),
+ None,
+ )?;
}
- self.replica.store_mut().update_work(
- &record.id,
- "committed",
- Some(revision),
- None,
- )?;
}
Err(error) if error.kind() == crate::runtime::ErrorKind::Unknown => {
outcome.unknown += 1;
@@ -995,7 +1007,11 @@ impl DirectorySync {
/// operation kind: a removal drops the subtree, a rename relocates the
/// subtree preserving accepted fingerprints, and other edits save the
/// accepted version at the path.
- async fn apply_retry_receipt(&mut self, record: &WorkRecord, revision: Revision) -> Result<()> {
+ async fn apply_retry_receipt(
+ &mut self,
+ record: &WorkRecord,
+ revision: Revision,
+ ) -> Result<bool> {
match record.operation.as_str() {
"remove" => {
self.replica
@@ -1007,6 +1023,49 @@ impl DirectorySync {
self.migrate_baseline_from_work(&from, &record.path, revision)?;
}
}
+ "content" if record.source_artifact.is_some() => {
+ // A resolution receipt publishes selected bytes that are not
+ // necessarily the live local content. Install them only when the
+ // live file is still the version this decision was dispatched
+ // against, then confirm the baseline from the verified result.
+ let Some(artifact) = &record.source_artifact else {
+ return Ok(false);
+ };
+ let mut file = self.replica.open_artifact(artifact)?;
+ let mut bytes = Vec::new();
+ std::io::copy(&mut file, &mut bytes).map_err(|e| e.to_string())?;
+ let expected = accepted_local(record)?;
+ self.install_chosen_bytes(
+ &record.path,
+ &bytes,
+ record.local_executable.unwrap_or(false),
+ expected.as_ref(),
+ )?;
+ let digest = ContentHash::from_bytes(*blake3::hash(&bytes).as_bytes());
+ let installed = self
+ .replica
+ .scan()?
+ .get(&record.path)
+ .cloned()
+ .filter(|node| {
+ matches!(node.kind, LocalKind::File { length, digest: live }
+ if length == bytes.len() as u64 && live == digest)
+ });
+ if let (Some(node), Some(installed)) =
+ (self.accepted_node_at(record, revision).await?, installed)
+ {
+ self.replica.store_mut().save_baseline_path(
+ &record.path,
+ &node,
+ Some(&installed),
+ revision,
+ )?;
+ } else {
+ // A newer local write or deletion stays live and pending;
+ // keep the conflict active and retry later.
+ return Ok(false);
+ }
+ }
_ => {
if let Some(node) = self.accepted_node_at(record, revision).await? {
let accepted = accepted_local(record)?;
@@ -1019,7 +1078,7 @@ impl DirectorySync {
}
}
}
- Ok(())
+ Ok(true)
}
/// Relocate a whole subtree recorded by a committed rename receipt without
diff --git a/src/replica/directory/resolution.rs b/src/replica/directory/resolution.rs
index 5b1f9b4..743d64c 100644
--- a/src/replica/directory/resolution.rs
+++ b/src/replica/directory/resolution.rs
@@ -164,7 +164,7 @@ impl DirectorySync {
/// version this decision was dispatched against. A newer local write stays
/// live and pending; it is never overwritten by the chosen version.
/// Returns the local fingerprint the baseline may accept for this path.
- fn install_chosen_bytes(
+ pub(super) fn install_chosen_bytes(
&mut self,
path: &str,
bytes: &[u8],
diff --git a/tests/directory_resolution.rs b/tests/directory_resolution.rs
index 860a1bd..dd68dfe 100644
--- a/tests/directory_resolution.rs
+++ b/tests/directory_resolution.rs
@@ -110,15 +110,16 @@ async fn remote_tree(authority: &Arc<dyn Authority>) -> BTreeMap<String, TreeEnt
}
/// A short-lived public Sync journal used for an independent remote edit.
-async fn remote_client(authority: &Arc<dyn Authority>) -> Sync {
+async fn remote_client(authority: &Arc<dyn Authority>) -> (Sync, tempfile::TempDir) {
let temp = tempfile::tempdir().unwrap();
- Sync::open(authority.clone(), temp.path().join("seed"))
+ let sync = Sync::open(authority.clone(), temp.path().join("seed"))
.await
- .unwrap()
+ .unwrap();
+ (sync, temp)
}
async fn remote_write(authority: &Arc<dyn Authority>, path: &str, bytes: &[u8]) {
- let sync = remote_client(authority).await;
+ let (sync, _temp) = remote_client(authority).await;
let baseline = authority.observe_latest().await.unwrap();
let node = baseline.resolve(path).await.unwrap().unwrap().id();
let edit = sync
@@ -129,7 +130,7 @@ async fn remote_write(authority: &Arc<dyn Authority>, path: &str, bytes: &[u8])
}
async fn remote_create_file(authority: &Arc<dyn Authority>, path: &str, bytes: &[u8]) {
- let sync = remote_client(authority).await;
+ let (sync, _temp) = remote_client(authority).await;
let (parent, name) = path.rsplit_once('/').unwrap_or(("", path));
let baseline = authority.observe_latest().await.unwrap();
let parent_node = baseline.resolve(parent).await.unwrap().unwrap().id();
@@ -147,7 +148,7 @@ async fn remote_create_file(authority: &Arc<dyn Authority>, path: &str, bytes: &
}
async fn remote_remove(authority: &Arc<dyn Authority>, path: &str) {
- let sync = remote_client(authority).await;
+ let (sync, _temp) = remote_client(authority).await;
let baseline = authority.observe_latest().await.unwrap();
let node = baseline.resolve(path).await.unwrap().unwrap().id();
let edit = sync.stage_remove(node, baseline.revision()).await.unwrap();
@@ -155,7 +156,7 @@ async fn remote_remove(authority: &Arc<dyn Authority>, path: &str) {
}
async fn remote_create_directory(authority: &Arc<dyn Authority>, path: &str) {
- let sync = remote_client(authority).await;
+ let (sync, _temp) = remote_client(authority).await;
let (parent, name) = path.rsplit_once('/').unwrap_or(("", path));
let baseline = authority.observe_latest().await.unwrap();
let parent_node = baseline.resolve(parent).await.unwrap().unwrap().id();