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 16:47:35 UTC
(opendal-yinyang) branch xuanwo/managed-sync-namespace updated: fix: keep post-dispatch namespace errors uncertain
This is an automated email from the ASF dual-hosted git repository.
Xuanwo pushed a commit to branch xuanwo/managed-sync-namespace
in repository https://gitbox.apache.org/repos/asf/opendal-yinyang.git
The following commit(s) were added to refs/heads/xuanwo/managed-sync-namespace by this push:
new 89e327b fix: keep post-dispatch namespace errors uncertain
89e327b is described below
commit 89e327ba9c94c67f1a29077840474fa21f6b0db7
Author: Xuanwo <gi...@xuanwo.io>
AuthorDate: Tue Oct 27 00:47:28 2026 +0800
fix: keep post-dispatch namespace errors uncertain
Co-authored-by: Chojan Shang <ps...@apache.org>
---
specs/sync-runtime.md | 7 ++-
src/sync.rs | 4 +-
src/sync/namespace.rs | 11 ++--
tests/sync.rs | 138 +++++++++++++++++++++++++++++++++++++++++++++++++-
4 files changed, 149 insertions(+), 11 deletions(-)
diff --git a/specs/sync-runtime.md b/specs/sync-runtime.md
index bf457eb..6f42d9b 100644
--- a/specs/sync-runtime.md
+++ b/specs/sync-runtime.md
@@ -84,8 +84,11 @@ The required interleavings are:
- Concurrent creates of one name publish at most one intent; the other is
`Conflict`.
- A delete or rename conflicts when a content/executable edit against its
- baseline publishes first. If the namespace operation publishes first, a later
- content edit conflicts because its original node state is no longer live.
+ baseline publishes first, because the edit changes the source node state the
+ namespace operation guards. The reverse order is safe for rename: the moved
+ identity keeps its content, so a same-baseline edit still commits under the
+ new name. Delete first makes the later edit conflict because its node is no
+ longer live.
- Concurrent renames of one source publish at most one move.
- Child creation and removal of its observed empty parent publish at most one
operation.
diff --git a/src/sync.rs b/src/sync.rs
index cac26df..6b35481 100644
--- a/src/sync.rs
+++ b/src/sync.rs
@@ -297,7 +297,7 @@ impl Sync {
pub async fn namespace_status(&self, edit: EditId) -> Result<NamespaceStatus> {
namespace::status(self, edit)
.await?
- .ok_or_else(|| Error::Invalid("unknown namespace intent"))
+ .ok_or(Error::Invalid("unknown namespace intent"))
}
/// List all durable namespace intents in stable identity order.
pub async fn namespace_intents(&self) -> Result<Vec<NamespaceStatus>> {
@@ -348,5 +348,5 @@ impl Sync {
#[derive(Clone, Debug)]
pub enum EditStatus {
Content(HandleStatus),
- Namespace(NamespaceStatus),
+ Namespace(Box<NamespaceStatus>),
}
diff --git a/src/sync/namespace.rs b/src/sync/namespace.rs
index d659c79..1cda5ad 100644
--- a/src/sync/namespace.rs
+++ b/src/sync/namespace.rs
@@ -983,12 +983,11 @@ pub(super) async fn publish(sync: &Sync, id: EditId) -> Result<Revision> {
Err(Error::Retained(failure))
}
Err(error) => {
- let (state, failure) = match error.kind() {
- ErrorKind::Retryable => (RETRYABLE, Error::Retryable),
- ErrorKind::Unknown => (UNKNOWN, error),
- ErrorKind::Conflict => (CONFLICT, Error::Conflict),
- _ => (FAILED, error),
- };
+ // A response or transport error after dispatch cannot prove that
+ // the commit did not publish. Only an authoritative outcome below
+ // the wrapper can establish Conflict/Retryable; anything else stays
+ // Unknown so the original receipt can resolve it later.
+ let (state, failure) = (UNKNOWN, error);
let failure = Failure::capture(&failure);
save_state_async(sync, id, state, None, Some(failure.clone())).await?;
Err(Error::Retained(failure))
diff --git a/tests/sync.rs b/tests/sync.rs
index 35cb182..85ccb6a 100644
--- a/tests/sync.rs
+++ b/tests/sync.rs
@@ -17,10 +17,13 @@
#[path = "../crates/core/tests/support/mod.rs"]
mod support;
+use std::future::Future;
+use std::pin::Pin;
use std::sync::Arc;
+use std::sync::atomic::{AtomicBool, Ordering};
use support::TestBackend;
use yinyang::core::service::{MetadataService, ServiceClient, serve};
-use yinyang::core::{Authority, BackendProfile, Fs, NodeId, Revision};
+use yinyang::core::{Authority, BackendProfile, Fs, NodeId, Outcome, Revision, Transaction};
use yinyang::runtime::{ErrorKind, Runtime};
use yinyang::sync::{
EditStatus, FileMetadata, NamespaceOperation, NamespaceState, RenameMode, Sync,
@@ -199,6 +202,67 @@ fn assert_conflict(result: yinyang::runtime::Result<Revision>) {
assert_eq!(result.unwrap_err().kind(), ErrorKind::Conflict);
}
+/// Delegating authority that substitutes a non-Unknown error after the inner
+/// authority already published, modeling a lost or malformed response.
+#[derive(Debug)]
+struct FaultAuthority {
+ inner: Arc<dyn Authority>,
+ fail_next: Arc<AtomicBool>,
+}
+
+impl FaultAuthority {
+ fn new(inner: Arc<dyn Authority>) -> Self {
+ Self {
+ inner,
+ fail_next: Arc::new(AtomicBool::new(false)),
+ }
+ }
+
+ fn fail_next_response(&self) {
+ self.fail_next.store(true, Ordering::SeqCst);
+ }
+}
+
+impl Authority for FaultAuthority {
+ fn data(&self) -> &yinyang::core::DataStore {
+ self.inner.data()
+ }
+ fn filesystem(&self) -> NodeId {
+ self.inner.filesystem()
+ }
+ fn root(&self) -> NodeId {
+ self.inner.root()
+ }
+ fn observe_latest<'a>(
+ &'a self,
+ ) -> Pin<Box<dyn Future<Output = yinyang::core::Result<yinyang::core::Snapshot>> + Send + 'a>>
+ {
+ self.inner.observe_latest()
+ }
+ fn observe_revision<'a>(
+ &'a self,
+ revision: Revision,
+ ) -> Pin<Box<dyn Future<Output = yinyang::core::Result<yinyang::core::Snapshot>> + Send + 'a>>
+ {
+ self.inner.observe_revision(revision)
+ }
+ fn commit_batch<'a>(
+ &'a self,
+ requests: &'a [Transaction],
+ ) -> Pin<Box<dyn Future<Output = yinyang::core::Result<Vec<Outcome>>> + Send + 'a>> {
+ Box::pin(async move {
+ let outcomes = self.inner.commit_batch(requests).await?;
+ if self.fail_next.swap(false, Ordering::SeqCst) {
+ return Err(yinyang::core::Error::invalid(
+ "fault authority",
+ "publication response was malformed",
+ ));
+ }
+ Ok(outcomes)
+ })
+ }
+}
+
async fn exercise_namespace(authority: Arc<dyn Authority>) {
let temp = tempfile::tempdir().unwrap();
let runtime = Runtime::open(authority.clone(), temp.path().join("editor"))
@@ -469,6 +533,47 @@ async fn exercise_namespace(authority: Arc<dyn Authority>) {
NamespaceState::Conflict
);
+ // Rename-first is safe: the same baseline content edit still commits under
+ // the moved identity because its content predicates were unchanged.
+ let (baseline, always_safe) =
+ create_file(&authority, &runtime, root, "moved-then-edited").await;
+ let content = sync
+ .stage(always_safe, baseline, &mut b"moved edit"[..])
+ .await
+ .unwrap();
+ let rename = sync
+ .stage_rename(
+ always_safe,
+ baseline,
+ root,
+ "moved-then-edited-2",
+ RenameMode::NoReplace,
+ )
+ .await
+ .unwrap();
+ sync.publish(rename).await.unwrap();
+ let revision = sync.publish(content).await.unwrap();
+ let observed = authority.observe_revision(revision).await.unwrap();
+ assert_eq!(
+ observed
+ .resolve("moved-then-edited-2")
+ .await
+ .unwrap()
+ .unwrap()
+ .id(),
+ always_safe
+ );
+ assert_eq!(
+ observed
+ .open_file(always_safe)
+ .await
+ .unwrap()
+ .read(0, 64)
+ .await
+ .unwrap(),
+ b"moved edit"
+ );
+
// Delete/edit conflicts in both orders. Neither the local replacement nor
// the pinned pre-delete version can be silently discarded.
let (baseline, deleted) = create_file(&authority, &runtime, root, "deleted-first").await;
@@ -725,3 +830,34 @@ async fn namespace_restore_failure_keeps_a_prepared_intent_retryable() {
NamespaceState::Committed(_)
));
}
+
+#[tokio::test]
+async fn namespace_post_dispatch_error_stays_unknown_and_resolves() {
+ let backend = TestBackend::default();
+ let authority: Arc<dyn Authority> = Arc::new(
+ Fs::create(backend.operator(), BackendProfile::Minio)
+ .await
+ .unwrap(),
+ );
+ let fault = Arc::new(FaultAuthority::new(authority.clone()));
+ let temp = tempfile::tempdir().unwrap();
+ let sync = Sync::open(fault.clone(), temp.path().join("sync"))
+ .await
+ .unwrap();
+ let baseline = fault.observe_latest().await.unwrap();
+ let intent = sync
+ .stage_create_directory(fault.root(), "lost-malformed", baseline.revision())
+ .await
+ .unwrap();
+ fault.fail_next_response();
+ let error = sync.publish(intent).await.unwrap_err();
+ assert_eq!(error.kind(), ErrorKind::Invalid);
+ let status = sync.namespace_status(intent).await.unwrap();
+ assert_eq!(status.state, NamespaceState::Unknown);
+ assert_eq!(status.failure.unwrap().kind(), ErrorKind::Invalid);
+ let receipt = sync.resolve_namespace(intent).await.unwrap().unwrap();
+ assert_eq!(
+ authority.observe_latest().await.unwrap().revision(),
+ receipt.cursor.revision
+ );
+}