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 19:01:47 UTC
(opendal-yinyang) branch xuanwo/managed-sync-reconciliation updated: feat: surface retained-inode late writes as resolvable conflicts
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 4f38f48 feat: surface retained-inode late writes as resolvable conflicts
4f38f48 is described below
commit 4f38f483f3a54db806f69325ab627b843b63fa62
Author: Xuanwo <gi...@xuanwo.io>
AuthorDate: Tue Oct 27 03:01:40 2026 +0800
feat: surface retained-inode late writes as resolvable conflicts
Report a retained install inode that gained late writer bytes as a read-only
conflict, resolve it through the same durable decision path, and acknowledge
the retained inode once a keep-local decision commits so the binding can
return to in_sync.
Co-authored-by: Chojan Shang <ps...@apache.org>
---
src/replica/directory.rs | 91 +++++++++++++++++++++++++++++++++++++++++++++---
src/replica/linux.rs | 26 ++++++++++++++
src/replica/state.rs | 20 +++++++++++
3 files changed, 133 insertions(+), 4 deletions(-)
diff --git a/src/replica/directory.rs b/src/replica/directory.rs
index d753f11..4614450 100644
--- a/src/replica/directory.rs
+++ b/src/replica/directory.rs
@@ -408,7 +408,45 @@ impl DirectorySync {
}
paths.remove("");
- let conflicts: Vec<Conflict> = status.conflicts.clone();
+ let mut conflicts: Vec<Conflict> = status.conflicts.clone();
+ // A retained install inode that gained late writer bytes is an
+ // unresolved conflict even though nothing durable was written before.
+ for artifact in &status.artifacts {
+ let install_kind =
+ artifact.kind == "replaced-file" || artifact.kind == "rollback-displaced";
+ if artifact.phase != "conflict"
+ || !install_kind
+ || artifact.origin.is_empty()
+ || conflicts.iter().any(|conflict| {
+ conflict.local_artifact.as_deref() == Some(artifact.id.as_str())
+ })
+ {
+ continue;
+ }
+ let path = artifact.origin.clone();
+ let remote = match remote_map.and_then(|map| map.get(&path)) {
+ Some(node) if node.is_directory() => "current_directory",
+ Some(_) => "current_file",
+ None => "deleted",
+ };
+ conflicts.push(Conflict {
+ id: artifact.id.clone(),
+ path: path.clone(),
+ original_path: Some(path.clone()),
+ desired_path: Some(path),
+ local: "retained_artifact".to_owned(),
+ remote: remote.to_owned(),
+ local_artifact: Some(artifact.id.clone()),
+ remote_node: remote_map
+ .and_then(|map| map.get(&artifact.origin))
+ .map(Node::id),
+ remote_revision: remote_status.revision.map(Revision::to_bytes),
+ error: artifact
+ .error
+ .clone()
+ .unwrap_or_else(|| "retained inode changed after installation".to_owned()),
+ });
+ }
let mut entries = Vec::with_capacity(paths.len());
let mut pending = 0_u64;
for path in &paths {
@@ -658,13 +696,16 @@ impl DirectorySync {
/// Apply one durable decision to an active conflict, then reconcile again.
/// The conflict stays active until its derived request actually commits.
pub async fn resolve(&mut self, conflict_id: &str, decision: Resolution) -> Result<RunOutcome> {
- let record = self
+ let record = match self
.replica
.store()
.conflicts()?
.into_iter()
.find(|record| record.id == conflict_id)
- .ok_or_else(|| crate::replica::Error::invalid("unknown conflict").to_string())?;
+ {
+ Some(record) => record,
+ None => self.synthetic_conflict(conflict_id)?,
+ };
let observed = self.sync.authority().observe_latest().await?;
let remote = crate::directory_nodes(&observed).await?;
let baseline = self.replica.store().load_baseline()?;
@@ -679,11 +720,53 @@ impl DirectorySync {
}
};
if applied {
- self.replica.store_mut().remove_conflict(conflict_id)?;
+ let is_artifact = self
+ .replica
+ .store()
+ .artifacts()?
+ .iter()
+ .any(|artifact| artifact.id == conflict_id);
+ if is_artifact {
+ self.replica.acknowledge_retained_artifact(conflict_id)?;
+ } else {
+ self.replica.store_mut().remove_conflict(conflict_id)?;
+ }
}
self.run().await
}
+ /// Build the conflict record for a retained install inode that gained late
+ /// writer bytes. This conflict is reported by inspection but is not durable
+ /// until an explicit decision is applied.
+ fn synthetic_conflict(&self, artifact_id: &str) -> Result<ConflictRecord> {
+ let artifact = self
+ .replica
+ .store()
+ .artifacts()?
+ .into_iter()
+ .find(|artifact| artifact.id == artifact_id)
+ .ok_or_else(|| crate::replica::Error::invalid("unknown conflict").to_string())?;
+ if artifact.phase != "conflict" || artifact.origin.is_empty() {
+ return Err(crate::replica::Error::invalid("unknown conflict")
+ .to_string()
+ .into());
+ }
+ Ok(ConflictRecord {
+ id: artifact.id.clone(),
+ path: artifact.origin.clone(),
+ original_path: Some(artifact.origin.clone()),
+ desired_path: Some(artifact.origin.clone()),
+ local: "retained_artifact".to_owned(),
+ remote: "current_file".to_owned(),
+ local_artifact: Some(artifact.id),
+ remote_node: None,
+ remote_revision: None,
+ error: artifact
+ .error
+ .unwrap_or_else(|| "retained inode changed after installation".to_owned()),
+ })
+ }
+
/// Publish the chosen local bytes through the same durable work/receipt
/// path as normal publication, linked to the conflict. The conflict is
/// only cleared once the receipt is applied.
diff --git a/src/replica/linux.rs b/src/replica/linux.rs
index 54ffdae..36a848b 100644
--- a/src/replica/linux.rs
+++ b/src/replica/linux.rs
@@ -775,6 +775,32 @@ impl Replica {
}
}
+ /// Acknowledge a retained install artifact whose inode gained late writer
+ /// bytes: re-baseline its expected digest to the current bytes so a
+ /// deliberate keep-local resolution is no longer reported as a conflict.
+ pub fn acknowledge_retained_artifact(&mut self, id: &str) -> Result<()> {
+ let artifacts = self.store.artifacts()?;
+ let artifact = artifacts
+ .iter()
+ .find(|artifact| artifact.id == id)
+ .ok_or_else(|| Error::invalid("unknown retained artifact"))?;
+ let installs = self.store.installs()?;
+ if let Some(mut install) = installs
+ .into_iter()
+ .find(|install| install.artifact == artifact.name)
+ {
+ let (length, digest, _, _) =
+ hash_file_at(&self.quarantine_fd, &artifact.name, &self.state)?;
+ install.expected_digest = *digest.as_bytes();
+ install.expected_length = length;
+ install.phase = InstallPhase::Retained;
+ install.error = None;
+ self.store.retarget_install(&install)?;
+ }
+ self.store.update_artifact(id, "retained", None)?;
+ Ok(())
+ }
+
/// Create one local directory with a no-replace mkdirat; the entry only
/// becomes visible after a successful directory entry creation.
pub(super) fn create_directory_local(&self, path: &str) -> Result<()> {
diff --git a/src/replica/state.rs b/src/replica/state.rs
index af5c8de..2890e67 100644
--- a/src/replica/state.rs
+++ b/src/replica/state.rs
@@ -1004,6 +1004,26 @@ impl Store {
Ok(())
}
+ /// Re-baseline the expected inode digest of a retained install so a late
+ /// writer into a retained inode can be explicitly acknowledged.
+ pub fn retarget_install(&mut self, install: &InstallRecord) -> Result<()> {
+ self.connection
+ .execute(
+ "UPDATE installs SET expected_digest=?2,expected_length=?3,
+ phase=?4,error=?5 WHERE id=?1",
+ params![
+ install.id,
+ install.expected_digest.as_slice(),
+ i64::try_from(install.expected_length)
+ .map_err(|_| Error::invalid("install length exceeds SQLite range"))?,
+ install.phase.as_str(),
+ install.error,
+ ],
+ )
+ .map_err(|e| Error::corrupt(format!("retarget install journal: {e}")))?;
+ Ok(())
+ }
+
pub fn incomplete_installs(&self) -> Result<Vec<InstallRecord>> {
self.installs_where("WHERE phase NOT IN ('retained','conflict','failed')")
}