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 17:55:40 UTC
(opendal-yinyang) branch xuanwo/managed-directory-sync updated: fix: compile directory coordinator against Sync API
This is an automated email from the ASF dual-hosted git repository.
Xuanwo pushed a commit to branch xuanwo/managed-directory-sync
in repository https://gitbox.apache.org/repos/asf/opendal-yinyang.git
The following commit(s) were added to refs/heads/xuanwo/managed-directory-sync by this push:
new 7be857d fix: compile directory coordinator against Sync API
7be857d is described below
commit 7be857de43c925fc0b630638da07e9a40ef97234
Author: Xuanwo <gi...@xuanwo.io>
AuthorDate: Tue Oct 27 01:55:33 2026 +0800
fix: compile directory coordinator against Sync API
Co-authored-by: Chojan Shang <ps...@apache.org>
---
src/replica/directory.rs | 61 +++++++++++++++++++++++++++++++-----------------
src/replica/linux.rs | 14 +++++------
src/replica/state.rs | 18 ++++++++++----
3 files changed, 60 insertions(+), 33 deletions(-)
diff --git a/src/replica/directory.rs b/src/replica/directory.rs
index 6b456a8..c2c8059 100644
--- a/src/replica/directory.rs
+++ b/src/replica/directory.rs
@@ -21,7 +21,7 @@ use crate::replica::state::{BaselineRecord, ConflictRecord};
use crate::replica::{Conflict, LocalKind, LocalNode, LocalSnapshot, Replica, ReplicaStatus};
use crate::sync::{EditId, FileMetadata, NamespaceState, RenameMode, Sync};
use std::collections::{BTreeMap, BTreeSet};
-use std::io::{Seek as _, SeekFrom, Write, Write as _};
+use std::io::{Seek as _, SeekFrom, Write};
use std::path::Path;
use std::sync::Arc;
use yinyang_core::{Authority, Node, NodeId, NodeKind, Revision, Snapshot};
@@ -75,7 +75,7 @@ impl DirectorySync {
let mut outcome = self.retry_pending().await?;
let mut current = self.replica.scan()?;
let observed = self.sync.authority().observe_latest().await?;
- let mut remote = crate::directory_nodes(&observed)?;
+ let mut remote = crate::directory_nodes(&observed).await?;
let baseline = self.replica.store().load_baseline()?;
match baseline {
Some(baseline) => {
@@ -157,10 +157,11 @@ impl DirectorySync {
Resolution::KeepLocal => {
if let Some(artifact) = record.local_artifact {
let observed = self.sync.authority().observe_latest().await?;
- let remote = crate::directory_nodes(&observed)?;
+ let remote = crate::directory_nodes(&observed).await?;
if let Some(node) = remote.get(&record.path) {
let mut source = self.replica.open_artifact(&artifact)?;
- let mut source = tokio::io::BufReader::new(source);
+ let mut source =
+ tokio::io::BufReader::new(tokio::fs::File::from_std(source));
self.sync
.stage(node.id(), observed.revision(), &mut source)
.await?;
@@ -529,6 +530,7 @@ impl<'a> Context<'a> {
Ok(())
} else {
self.conflict(path, None, Some(path), "create", "create")
+ .await
}
}
(Action::Modify, Action::Modify) => {
@@ -538,6 +540,7 @@ impl<'a> Context<'a> {
Ok(())
} else {
self.conflict(path, Some(path), None, "modify", "modify")
+ .await
}
}
(Action::Delete, Action::Modify)
@@ -547,6 +550,7 @@ impl<'a> Context<'a> {
| (Action::Create, Action::Modify)
| (Action::Create, Action::Delete) => {
self.conflict(path, Some(path), None, "local", "remote")
+ .await
}
}
}
@@ -680,13 +684,16 @@ impl<'a> Context<'a> {
self.accept_remote(path, node).await?;
Ok(())
}
- Err(error) => self
- .conflict(path, None, Some(path), "present", "remote-create")
- .map_err(|_| error.into()),
+ Err(_) => {
+ self.conflict(path, None, Some(path), "present", "remote-create")
+ .await
+ }
},
NodeKind::File(_) => {
if self.current.get(path).is_some() {
- return self.conflict(path, None, Some(path), "present", "remote-create");
+ return self
+ .conflict(path, None, Some(path), "present", "remote-create")
+ .await;
}
let mut staged = self.stage_remote_file(node).await?;
match self
@@ -699,9 +706,10 @@ impl<'a> Context<'a> {
self.accept_remote(path, node).await?;
Ok(())
}
- Err(error) => self
- .conflict(path, None, Some(path), "present", "remote-create")
- .map_err(|_| error.into()),
+ Err(_) => {
+ self.conflict(path, None, Some(path), "present", "remote-create")
+ .await
+ }
}
}
}
@@ -712,7 +720,9 @@ impl<'a> Context<'a> {
crate::replica::Error::invalid("remote modify path is local-absent").to_string()
})?;
if expected.is_directory() || node.is_directory() {
- return self.conflict(path, Some(path), None, "local", "remote-kind");
+ return self
+ .conflict(path, Some(path), None, "local", "remote-kind")
+ .await;
}
let snapshot = self
.sync
@@ -742,15 +752,16 @@ impl<'a> Context<'a> {
self.accept_remote(path, node).await?;
Ok(())
}
- Err(error) => self
- .conflict(
+ Err(_) => {
+ self.conflict(
path,
Some(path),
None,
"changed-during-install",
"remote-modify",
)
- .map_err(|_| error.into()),
+ .await
+ }
}
}
@@ -769,16 +780,23 @@ impl<'a> Context<'a> {
self.outcome.materialized_remote += 1;
Ok(())
}
- Err(_) => self.conflict(path, Some(path), None, "local-changed", "remote-delete"),
+ Err(_) => {
+ self.conflict(path, Some(path), None, "local-changed", "remote-delete")
+ .await
+ }
}
}
async fn remote_rename(&mut self, from: &str, to: &str) -> Result<()> {
if local_action(self.baseline.local.get(from), self.current.get(from)) != Action::None {
- return self.conflict(from, Some(from), Some(to), "local-changed", "remote-rename");
+ return self
+ .conflict(from, Some(from), Some(to), "local-changed", "remote-rename")
+ .await;
}
if self.current.get(to).is_some() {
- return self.conflict(to, Some(from), Some(to), "present", "remote-rename");
+ return self
+ .conflict(to, Some(from), Some(to), "present", "remote-rename")
+ .await;
}
match self.replica.rename_local(from, to) {
Ok(()) => {
@@ -793,9 +811,10 @@ impl<'a> Context<'a> {
self.outcome.materialized_remote += 1;
Ok(())
}
- Err(error) => self
- .conflict(from, Some(from), Some(to), "local-changed", "remote-rename")
- .map_err(|_| error.into()),
+ Err(_) => {
+ self.conflict(from, Some(from), Some(to), "local-changed", "remote-rename")
+ .await
+ }
}
}
diff --git a/src/replica/linux.rs b/src/replica/linux.rs
index c29d20f..5fbd92a 100644
--- a/src/replica/linux.rs
+++ b/src/replica/linux.rs
@@ -50,7 +50,7 @@ pub enum LocalKind {
File { length: u64, digest: ContentHash },
}
-#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+#[derive(Clone, Copy, Debug, Eq, PartialEq, Ord, PartialOrd)]
pub struct Identity {
pub dev: u64,
pub ino: u64,
@@ -417,7 +417,7 @@ impl Replica {
.map_err(|e| Error::errno("open retained artifact directory", &state, e))?;
let mut artifacts: Vec<Artifact> =
store.artifacts()?.into_iter().map(Artifact::from).collect();
- for mut artifact in artifacts.iter_mut() {
+ for artifact in artifacts.iter_mut() {
let record = store
.artifacts()?
.into_iter()
@@ -721,7 +721,7 @@ impl Replica {
let name_check = yinyang_core::namespace::name_key(name)
.map_err(|e| Error::invalid(format!("invalid local name {name:?}: {e}")))?;
let parent_fd = self.open_relative_dir(&parent)?;
- if let Some(stat) = optional_stat(&parent_fd, name, &self.root.join(path))? {
+ if optional_stat(&parent_fd, name, &self.root.join(path))?.is_some() {
return Err(Error::conflict(format!(
"{} already exists before remote creation could install",
path
@@ -1012,7 +1012,7 @@ impl Replica {
Ok(record.into())
}
- pub fn export_artifact(&self, id: &str, output: &mut impl Write) -> Result<()> {
+ pub fn export_artifact(&self, id: &str, output: &mut (impl Write + ?Sized)) -> Result<()> {
let artifact = self
.store
.artifacts()?
@@ -1551,7 +1551,7 @@ impl PreparedInstall<'_> {
where
F: FnOnce(&Self),
{
- let mut file = self.file.take().expect("prepared install is live");
+ let file = self.file.take().expect("prepared install is live");
file.sync_all().map_err(|e| {
Error::io(
"persist replacement file",
@@ -2238,7 +2238,7 @@ mod tests {
let mut replica = Replica::init(&state, &root, NodeId::from_bytes([9; 16])).unwrap();
let snapshot = replica.scan().unwrap();
let expected = snapshot.get("file").unwrap();
- let mut local_writer = std::fs::OpenOptions::new()
+ let local_writer = std::fs::OpenOptions::new()
.write(true)
.open(root.join("file"))
.unwrap();
@@ -2251,7 +2251,7 @@ mod tests {
let target = root.join("file");
let error = install
.finish_with_hook(|_| {
- let mut replacement_writer = std::fs::OpenOptions::new()
+ let replacement_writer = std::fs::OpenOptions::new()
.write(true)
.open(&target)
.unwrap();
diff --git a/src/replica/state.rs b/src/replica/state.rs
index b94e656..d4e40df 100644
--- a/src/replica/state.rs
+++ b/src/replica/state.rs
@@ -310,10 +310,18 @@ impl Store {
let matches = stored.0 == PROFILE
&& stored.1 == binding.filesystem.as_bytes()
&& stored.2 == binding.root
- && number(&stored.3, "root device")? == binding.root_dev
- && number(&stored.4, "root inode")? == binding.root_ino
- && number(&stored.5, "state device")? == binding.state_dev
- && number(&stored.6, "state inode")? == binding.state_ino;
+ && number(&stored.3, "root device")
+ .map_err(|e| Error::corrupt(format!("decode root device: {e}")))?
+ == binding.root_dev
+ && number(&stored.4, "root inode")
+ .map_err(|e| Error::corrupt(format!("decode root inode: {e}")))?
+ == binding.root_ino
+ && number(&stored.5, "state device")
+ .map_err(|e| Error::corrupt(format!("decode state device: {e}")))?
+ == binding.state_dev
+ && number(&stored.6, "state inode")
+ .map_err(|e| Error::corrupt(format!("decode state inode: {e}")))?
+ == binding.state_ino;
Ok(matches)
}
@@ -352,7 +360,7 @@ impl Store {
Some(entry) => (
1_i64,
match entry.kind {
- LocalKind::File { digest, .. } => Some(digest),
+ LocalKind::File { digest, .. } => Some(*digest.as_bytes()),
LocalKind::Directory => None,
},
match entry.kind {