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:01:38 UTC
(opendal-yinyang) branch xuanwo/managed-sync-reconciliation updated: fix: apply namespace-only receipts without adopting stale content
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 44dbe49 fix: apply namespace-only receipts without adopting stale content
44dbe49 is described below
commit 44dbe494793b1a6bc05f23688d3d1878da4dc1d5
Author: Xuanwo <gi...@xuanwo.io>
AuthorDate: Tue Oct 27 04:01:32 2026 +0800
fix: apply namespace-only receipts without adopting stale content
- Remove receipts drop the whole accepted baseline subtree; rename receipts
relocate the source subtree preserving accepted content fingerprints, and
never adopt descendant content included only by the rename snapshot.
- Extend the interleaving regression to actually resolve the exposed conflict,
verifying KeepRemote adoption and a final in_sync inspection.
- Add a local-only conflict-side export that reads state without constructing
an Authority.
Co-authored-by: Chojan Shang <ps...@apache.org>
---
specs/directory-sync.md | 103 ++++++++++++++++++++++
src/replica/directory.rs | 221 +++++++++++++++++++++++++++++++++++++++++------
src/replica/linux.rs | 2 +-
src/replica/mod.rs | 2 +-
src/replica/state.rs | 31 ++++++-
tests/directory_sync.rs | 45 +++++++++-
6 files changed, 372 insertions(+), 32 deletions(-)
diff --git a/specs/directory-sync.md b/specs/directory-sync.md
new file mode 100644
index 0000000..c51738e
--- /dev/null
+++ b/specs/directory-sync.md
@@ -0,0 +1,103 @@
+# Linux ordinary-directory Managed Sync
+
+**Status: experimental finite reconciliation on top of the durable Sync intents.**
+
+`yinyang::replica::DirectorySync` reconciles one local ordinary directory with
+one Managed remote root. It owns the common baseline, local installation and
+conflict records; durable conditional publication belongs to
+[`yinyang::sync::Sync`](sync-runtime.md). It is Linux-only and performs explicit
+finite reconciliation runs; no watcher or daemon is implied.
+
+## Common baseline
+
+The baseline records the last agreed state, one entry per synchronized path
+(excluding the root, which is the binding anchor):
+
+- the accepted remote `NodeId`, kind, executable bit and content digest;
+- the accepted remote revision for that entry, used as the staging predicate
+ and as status evidence, never a freshly observed latest revision;
+- the accepted local fingerprint (kind, length, content digest, executable bit)
+ together with its inode and parent identity, captured when the entry was
+ accepted.
+
+The baseline exists only after admission and is stored in the replica database.
+Status never reports `in_sync` for a binding without an accepted common
+baseline.
+
+## Admission
+
+An unbound state directory admits exactly one empty side. Recreation validates
+this before writing any baseline, then persists a root-only common baseline and
+hands the run to the same per-entry reconciler used by every later run. Two
+non-empty sides are an explicit conflict, not an implicit merge.
+
+## Durable work association
+
+Before a staged edit is dispatched, the directory layer persists an association
+keyed by the edit identity: the current path, original and desired path, the
+operation, the accepted local fingerprint and identity, the baseline node and
+revision, any known original commit identity, the linked conflict, and the
+retained input artifact. Retry publishes only associated work and never
+resurrects orphaned staging.
+
+Receipt application is ordered so an interrupted run is recoverable: the common
+baseline (and any linked-conflict clearance) is applied from the returned
+revision before the association is marked complete. Publication is idempotent,
+so re-running the exact request after a crash resolves to the same commit.
+
+## Reconciliation
+
+Each run scans the local tree, observes the remote, and performs a three-way
+comparison of baseline, local and remote per entry, ordering parents before
+children for creation and children before parents for removal. Renames are
+detected by identity and applied as one per-root move with each descendant's
+accepted fingerprint relocated, so a namespace receipt never acknowledges a
+concurrent content edit.
+
+A namespace operation stages against the path's own accepted revision.
+Only paths whose parent directory entry list this run already changed advance to
+the running revision, so a child removal permits its parent's later removal
+without rebasing unrelated paths onto a newer global revision and hiding an
+external edit.
+
+A remote deletion of a directory is quarantined as one unit; when any local
+descendant diverges from the common baseline the directory is retained and the
+unchanged retained entries are recorded as explicit conflicts instead of being
+removed.
+
+## Conflicts and resolution
+
+A conflict has a stable 32-hex identity derived from stable evidence (paths,
+observable sides, baseline and remote node). Detecting the same unresolved
+evidence again refreshes that record rather than inserting a duplicate, and
+reuses the first retained backup. A side is classified by presence
+(`current_file`, `current_directory`, `deleted`, `retained_artifact`,
+`unknown`). `retained_artifact` is reserved for a version that exists only in
+retained state, such as a displaced inode written after an atomic exchange; an
+ordinary content divergence is `current_file` on both sides even when a backup
+copy is preserved separately.
+
+Resolution is durable and conditional: `KeepLocal`, `KeepRemote` and explicit
+`Content` retain the chosen input as an artifact, stage the derived request
+against the recorded predicate, apply the resulting baseline, and clear the
+conflict only after the receipt commits. A stale remote predicate produces a
+new conflict rather than overwriting the remote.
+
+## Read-only inspection
+
+`DirectorySync::inspect` (and `inspect_local` when no authority can be
+constructed) reports the binding, a full local scan classified against the
+baseline, live remote availability, per-entry local/remote/baseline evidence,
+per-entry intent state (including a preserved original commit identity) and
+outcome, active conflicts and retained artifacts. It acquires no writer lease,
+runs no recovery and performs no writes, so it remains usable while a writer
+runs and while the remote is unreachable. The adapter owns the versioned JSON
+projection; the library exposes only domain evidence.
+
+## Export
+
+Retained artifacts export their current bytes, including writes that landed
+after the exchange. A conflict side exports its retained artifact or live local
+bytes (`local`) or the recorded remote node version (`remote`). Directory
+artifacts expose their retained path; byte-stream export of a directory is
+unsupported.
diff --git a/src/replica/directory.rs b/src/replica/directory.rs
index 361087c..c46d0bf 100644
--- a/src/replica/directory.rs
+++ b/src/replica/directory.rs
@@ -675,17 +675,7 @@ impl DirectorySync {
/// it keeps working offline.
pub fn export_conflict_local(&self, conflict_id: &str, output: &mut dyn Write) -> Result<()> {
let record = self.load_conflict_record(conflict_id)?;
- if let Some(artifact) = record.local_artifact {
- return Ok(self.replica.export_artifact(&artifact, output)?);
- }
- let path = record
- .desired_path
- .as_deref()
- .unwrap_or(record.path.as_str());
- let file = self.replica.open_local_file(path)?;
- let mut file = file;
- std::io::copy(&mut file, output).map_err(|error| error.to_string())?;
- Ok(())
+ export_local_side(&self.replica, &record, output)
}
fn load_conflict_record(&self, conflict_id: &str) -> Result<ConflictRecord> {
@@ -1223,15 +1213,7 @@ 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.
- if let Some(node) = self.accepted_node_at(&record, revision).await? {
- let accepted = accepted_local(&record)?;
- self.replica.store_mut().save_baseline_path(
- &record.path,
- &node,
- accepted.as_ref(),
- revision,
- )?;
- }
+ self.apply_retry_receipt(&record, revision).await?;
if let Some(conflict) = &record.conflict {
self.clear_resolved_conflict(conflict)?;
}
@@ -1281,6 +1263,179 @@ impl DirectorySync {
let observed = self.sync.authority().observe_revision(revision).await?;
Ok(observed.resolve(&record.path).await?)
}
+
+ /// Apply an already-committed retry receipt to the common baseline by
+ /// 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<()> {
+ match record.operation.as_str() {
+ "remove" => {
+ self.replica
+ .store_mut()
+ .remove_baseline_subtree(&record.path)?;
+ }
+ "rename" => {
+ if let Some(from) = record.original_path.clone() {
+ self.migrate_baseline_from_work(&from, &record.path, revision)?;
+ }
+ }
+ _ => {
+ if let Some(node) = self.accepted_node_at(record, revision).await? {
+ let accepted = accepted_local(record)?;
+ self.replica.store_mut().save_baseline_path(
+ &record.path,
+ &node,
+ accepted.as_ref(),
+ revision,
+ )?;
+ }
+ }
+ }
+ Ok(())
+ }
+
+ /// Relocate a whole subtree recorded by a committed rename receipt without
+ /// re-reading content: the receipt proves the namespace move only.
+ fn migrate_baseline_from_work(
+ &mut self,
+ from: &str,
+ to: &str,
+ revision: Revision,
+ ) -> Result<()> {
+ let Some(baseline) = self.replica.store().load_baseline()? else {
+ return Ok(());
+ };
+ let prefix = format!("{from}/");
+ let moves: Vec<(
+ String,
+ String,
+ crate::replica::state::BaselineNode,
+ Option<LocalNode>,
+ )> = baseline
+ .remote
+ .iter()
+ .filter(|(path, _)| *path == from || path.starts_with(&prefix))
+ .map(|(path, entry)| {
+ let destination = if path == from {
+ to.to_owned()
+ } else {
+ format!("{to}{}", &path[from.len()..])
+ };
+ (
+ path.clone(),
+ destination,
+ entry.clone(),
+ baseline.local.get(path).cloned(),
+ )
+ })
+ .collect();
+ for (source, destination, entry, previous) in moves {
+ let accepted = accepted_local_before(&self.replica, &destination, previous.as_ref())?;
+ let length = accepted
+ .as_ref()
+ .and_then(|node| match node.kind {
+ LocalKind::File { length, .. } => Some(length),
+ LocalKind::Directory => None,
+ })
+ .unwrap_or(0);
+ self.replica.store_mut().save_baseline_entry(
+ &destination,
+ entry.node,
+ entry.directory,
+ entry.executable,
+ entry.digest,
+ length,
+ accepted.as_ref(),
+ revision,
+ )?;
+ self.replica.store_mut().remove_baseline_path(&source)?;
+ }
+ Ok(())
+ }
+}
+
+/// Export only the local side of a conflict without constructing an Authority.
+/// `conflict_id` is the identity reported by inspection, including a conflict
+/// synthesized from a retained install inode that gained late writer bytes.
+pub fn export_conflict_local(
+ state: &Path,
+ conflict_id: &str,
+ output: &mut dyn Write,
+) -> Result<()> {
+ let replica = Replica::open_read_only(state)?;
+ let record = match replica
+ .store()
+ .conflicts()?
+ .into_iter()
+ .find(|record| record.id == conflict_id)
+ {
+ Some(record) => record,
+ None => {
+ let artifact = replica
+ .store()
+ .artifacts()?
+ .into_iter()
+ .find(|artifact| {
+ artifact_conflict_identity(&artifact.id) == conflict_id
+ && artifact.phase == "conflict"
+ })
+ .ok_or_else(|| crate::replica::Error::invalid("unknown conflict").to_string())?;
+ ConflictRecord {
+ id: conflict_id.to_owned(),
+ 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()),
+ }
+ }
+ };
+ export_local_side(&replica, &record, output)
+}
+
+fn export_local_side(
+ replica: &Replica,
+ record: &ConflictRecord,
+ output: &mut dyn Write,
+) -> Result<()> {
+ if let Some(artifact) = &record.local_artifact {
+ return Ok(replica.export_artifact(artifact, output)?);
+ }
+ let path = record
+ .desired_path
+ .as_deref()
+ .unwrap_or(record.path.as_str());
+ let mut file = replica.open_local_file(path)?;
+ std::io::copy(&mut file, output).map_err(|error| error.to_string())?;
+ Ok(())
+}
+
+fn accepted_local_before(
+ replica: &Replica,
+ to: &str,
+ previous: Option<&LocalNode>,
+) -> Result<Option<LocalNode>> {
+ let Some(previous) = previous else {
+ return Ok(None);
+ };
+ let fresh = replica.scan()?;
+ let Some(current) = fresh.get(to) else {
+ return Ok(None);
+ };
+ Ok(Some(LocalNode {
+ path: to.to_owned(),
+ kind: previous.kind,
+ executable: previous.executable,
+ identity: current.identity,
+ parent: current.parent,
+ }))
}
fn accepted_local(record: &WorkRecord) -> Result<Option<LocalNode>> {
@@ -1601,15 +1756,27 @@ impl<'a> Context<'a> {
let Some(entry) = self.baseline.remote.get(from).cloned() else {
return Ok(());
};
- let snapshot = self.sync.authority().observe_revision(revision).await?;
- let Some(node) = snapshot.node(entry.node).await? else {
- return Ok(());
- };
let previous = self.baseline.local.get(from).cloned();
let accepted = self.accepted_local_at(to, previous.as_ref())?;
- self.replica
- .store_mut()
- .save_baseline_path(to, &node, accepted.as_ref(), revision)?;
+ let length = accepted
+ .as_ref()
+ .and_then(|node| match node.kind {
+ LocalKind::File { length, .. } => Some(length),
+ LocalKind::Directory => None,
+ })
+ .unwrap_or(0);
+ // The rename receipt proves the namespace move, not descendant content.
+ // Keep the previously accepted remote digest and executable bit.
+ self.replica.store_mut().save_baseline_entry(
+ to,
+ entry.node,
+ entry.directory,
+ entry.executable,
+ entry.digest,
+ length,
+ accepted.as_ref(),
+ revision,
+ )?;
self.replica.store_mut().remove_baseline_path(from)?;
self.nodes.remove(from);
self.nodes.insert(to.to_owned(), entry.node);
diff --git a/src/replica/linux.rs b/src/replica/linux.rs
index c396768..271b160 100644
--- a/src/replica/linux.rs
+++ b/src/replica/linux.rs
@@ -394,7 +394,7 @@ impl Replica {
Self::open_read_only(state)?.scan()
}
- fn open_read_only(state: &Path) -> Result<Self> {
+ pub(crate) fn open_read_only(state: &Path) -> Result<Self> {
let state =
std::fs::canonicalize(state).map_err(|e| Error::io("canonicalize", state, e))?;
let (state_fd, state_identity) = open_bound_directory(&state, "open sync state")?;
diff --git a/src/replica/mod.rs b/src/replica/mod.rs
index 2e6d08a..e413893 100644
--- a/src/replica/mod.rs
+++ b/src/replica/mod.rs
@@ -30,7 +30,7 @@ pub use directory::{
BaselineEvidence, BindingEvidence, BindingState, DirectorySync, EntryEvidence, EntryKind,
EntryState, Inspection, IntentEvidence, IntentOperation, IntentState, LocalEvidence,
LocalScanState, LocalStatus, OutcomeEvidence, OutcomeState, Presence, RemoteAvailability,
- RemoteEvidence, RemoteStatus, Resolution, RunOutcome, StatusSummary,
+ RemoteEvidence, RemoteStatus, Resolution, RunOutcome, StatusSummary, export_conflict_local,
};
pub use error::{Error, ErrorKind, Result};
pub use linux::{
diff --git a/src/replica/state.rs b/src/replica/state.rs
index 77e399a..6705926 100644
--- a/src/replica/state.rs
+++ b/src/replica/state.rs
@@ -526,6 +526,33 @@ impl Store {
Some(*file.content_id().digest()),
),
};
+ self.save_baseline_entry(
+ path,
+ node.id(),
+ directory,
+ node.executable(),
+ digest,
+ length,
+ local,
+ revision,
+ )
+ }
+
+ /// Write one baseline entry with explicit accepted remote facts. A
+ /// namespace-only receipt advances path/revision but keeps the previously
+ /// accepted content digest and executable bit.
+ #[allow(clippy::too_many_arguments)]
+ pub fn save_baseline_entry(
+ &mut self,
+ path: &str,
+ node: NodeId,
+ directory: bool,
+ executable: bool,
+ digest: Option<[u8; 32]>,
+ length: u64,
+ local: Option<&LocalNode>,
+ revision: Revision,
+ ) -> Result<()> {
let (
present,
local_digest,
@@ -567,9 +594,9 @@ impl Store {
VALUES(?1,?2,?3,?4,?5,?6,?7,?8,?9,?10,?11,?12,?13,?14,?15)",
params![
path,
- node.id().as_bytes().as_slice(),
+ node.as_bytes().as_slice(),
directory,
- node.executable(),
+ executable,
digest.as_ref().map(|value| value.as_slice()),
i64::try_from(length).map_err(|_| Error::invalid(
"remote baseline length exceeds SQLite range"
diff --git a/tests/directory_sync.rs b/tests/directory_sync.rs
index 32176d7..7738fc6 100644
--- a/tests/directory_sync.rs
+++ b/tests/directory_sync.rs
@@ -33,7 +33,7 @@ use yinyang::core::service::{MetadataService, ServiceClient, serve};
use yinyang::core::{
Authority, BackendProfile, CommitOutcome, Fs, NodeId, Revision, Snapshot, Transaction,
};
-use yinyang::replica::DirectorySync;
+use yinyang::replica::{DirectorySync, Resolution};
use yinyang::sync::{FileMetadata, Sync};
/// An independent snapshot of a real tree on one side of a sync.
@@ -725,6 +725,49 @@ async fn interleaved_remote_edit_is_not_overwritten_by_local_run() {
.iter()
.all(|conflict| !conflict.id.is_empty())
);
+
+ // The conflict is resolvable: keeping the remote side adopts the external
+ // bytes locally, preserves them remotely, and returns to in_sync.
+ let conflict = inspected.conflicts[0].clone();
+ let mut resolver = DirectorySync::open(authority.clone(), &state)
+ .await
+ .unwrap();
+ resolver
+ .resolve(&conflict.id, Resolution::KeepRemote)
+ .await
+ .unwrap();
+ let local = local_tree(&root);
+ let remote = remote_tree(&authority).await;
+ assert_eq!(
+ local["b"],
+ TreeEntry::File {
+ executable: false,
+ bytes: b"external-b".to_vec(),
+ },
+ "keep-remote must adopt the external bytes locally"
+ );
+ assert_eq!(
+ remote["b"],
+ TreeEntry::File {
+ executable: false,
+ bytes: b"external-b".to_vec(),
+ },
+ "the external remote edit must survive resolution"
+ );
+
+ let mut reopened = DirectorySync::open(authority.clone(), &state)
+ .await
+ .unwrap();
+ let stable = reopened.run().await.unwrap();
+ assert!(
+ stable.in_sync,
+ "a resolved interleaving must reach in_sync on the next run"
+ );
+ let final_status = DirectorySync::inspect(authority.clone(), &state)
+ .await
+ .unwrap();
+ assert!(final_status.conflicts.is_empty());
+ assert!(final_status.summary.in_sync);
}
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]