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:24:39 UTC
(opendal-yinyang) branch xuanwo/managed-sync-namespace created (now c13d1b8)
This is an automated email from the ASF dual-hosted git repository.
Xuanwo pushed a change to branch xuanwo/managed-sync-namespace
in repository https://gitbox.apache.org/repos/asf/opendal-yinyang.git
at c13d1b8 fix: preserve uncertain namespace outcomes
This branch includes the following new commits:
new 752e19b feat: coordinate durable Sync namespace intents
new c11bba4 Merge managed sync status documentation
new c13d1b8 fix: preserve uncertain namespace outcomes
The 3 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(opendal-yinyang) 02/03: Merge managed sync status documentation
Posted by xu...@apache.org.
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
commit c11bba4ee1ea2715ee8a2442bcd7c1d88d6b6235
Merge: 752e19b cd93967
Author: Xuanwo <gi...@xuanwo.io>
AuthorDate: Tue Oct 27 00:21:56 2026 +0800
Merge managed sync status documentation
Co-authored-by: Chojan Shang <ps...@apache.org>
CONTRIBUTING.md | 8 ++++++++
README.md | 33 +++++++++++++++++++++++++++++----
specs/README.md | 14 ++++++++++----
3 files changed, 47 insertions(+), 8 deletions(-)
(opendal-yinyang) 03/03: fix: preserve uncertain namespace outcomes
Posted by xu...@apache.org.
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
commit c13d1b825509d935989f933be5ace8f094e5f42c
Author: Xuanwo <gi...@xuanwo.io>
AuthorDate: Tue Oct 27 00:24:22 2026 +0800
fix: preserve uncertain namespace outcomes
Co-authored-by: Chojan Shang <ps...@apache.org>
---
crates/core/tests/support/mod.rs | 27 ++++++++++++
specs/sync-runtime.md | 15 ++++++-
src/sync.rs | 10 ++++-
src/sync/namespace.rs | 90 +++++++++++++++++++++++++++++++++-------
tests/sync.rs | 53 +++++++++++++++++++++++
5 files changed, 178 insertions(+), 17 deletions(-)
diff --git a/crates/core/tests/support/mod.rs b/crates/core/tests/support/mod.rs
index d9c9986..aed2670 100644
--- a/crates/core/tests/support/mod.rs
+++ b/crates/core/tests/support/mod.rs
@@ -45,6 +45,8 @@ pub struct TestState {
pub delay_next_head: bool,
pub pending_head: Option<(Vec<u8>, Option<String>)>,
pub file_to_change_on_data_close: Option<std::path::PathBuf>,
+ pub offline: bool,
+ pub fail_next_head_write_and_go_offline: bool,
}
#[derive(Clone, Debug)]
@@ -74,6 +76,13 @@ impl TestBackend {
self.state.lock().unwrap().fail_after_head_write = true;
}
+ pub fn lose_next_head_response_and_go_offline(&self) {
+ self.state
+ .lock()
+ .unwrap()
+ .fail_next_head_write_and_go_offline = true;
+ }
+
pub fn stat_calls(&self) -> u64 {
self.state.lock().unwrap().stat_calls
}
@@ -134,6 +143,9 @@ impl Service for TestService {
async fn stat(&self, _: &OperationContext, path: &str, _: OpStat) -> opendal::Result<RpStat> {
let mut state = self.state.lock().unwrap();
+ if state.offline {
+ return Err(unavailable());
+ }
state.stat_calls += 1;
if path == ".yinyang/head" {
state.head_stat_calls += 1;
@@ -144,6 +156,9 @@ impl Service for TestService {
fn read(&self, _: &OperationContext, path: &str, _: OpRead) -> opendal::Result<Self::Reader> {
let mut state = self.state.lock().unwrap();
+ if state.offline {
+ return Err(unavailable());
+ }
state.read_paths.push(path.to_owned());
let object = state.objects.get(path).cloned().ok_or_else(not_found)?;
Ok(TestReader {
@@ -332,6 +347,14 @@ impl oio::Write for TestWriter {
"publication response was lost",
));
}
+ if self.path == ".yinyang/head" && state.fail_next_head_write_and_go_offline {
+ state.fail_next_head_write_and_go_offline = false;
+ state.offline = true;
+ return Err(opendal::Error::new(
+ opendal::ErrorKind::Unexpected,
+ "publication response was lost and authority went offline",
+ ));
+ }
Ok(metadata)
}
@@ -360,3 +383,7 @@ fn unsupported() -> opendal::Error {
"operation is not supported by the test backend",
)
}
+
+fn unavailable() -> opendal::Error {
+ opendal::Error::new(opendal::ErrorKind::Unexpected, "test authority is offline")
+}
diff --git a/specs/sync-runtime.md b/specs/sync-runtime.md
index 1fe5340..bf457eb 100644
--- a/specs/sync-runtime.md
+++ b/specs/sync-runtime.md
@@ -55,7 +55,11 @@ Creation allocates its stable NodeId in the same Planner that produces the
persisted transaction. The intent exposes that NodeId; callers must not resolve
the path again and adopt a concurrent replacement. `CreateFile` also exposes
the prepared `ContentDescriptor`, allowing a local adapter to compare its
-current bytes without a fresh remote lookup.
+current bytes without a fresh remote lookup. `read` on that intent reads the
+verified immutable descriptor from the authority, including before publication;
+this requires authority access. Other namespace intents return `Unsupported`
+for `read`. An adapter that needs offline export of a creation conflict must
+keep its durable local input separately from Sync.
`FileMetadata` contains only the executable bit. File creation and subsequent
`SetExecutable` support that bit. Directory modes, ownership, ACLs, timestamps,
@@ -131,6 +135,15 @@ does not prove that a delayed attempt failed. Conflict resolution is explicit:
keep or export the relevant local and pinned baseline bytes, then create a new
intent with suitable current predicates.
+Recovery preserves what prior dispatch established. For `Unknown`, `publish`
+checks the original receipt first. If the authority is unavailable or restoring
+the retained transaction fails, the intent remains `Unknown` with the new
+diagnostic; failure to observe is not evidence that the commit failed. Once the
+authority recovers, receipt lookup can attach the original outcome without a
+second dispatch. A transient restore failure for `Prepared` or `Retryable`
+retains `Retryable`; only invalid, corrupt, or unsupported retained state becomes
+`Failed`.
+
Operations within one Sync instance serialize to protect intent creation,
staging leases, and state transitions. This is not a background worker,
directory reconciler, OS cache, or cross-process concurrent uploader. It does
diff --git a/src/sync.rs b/src/sync.rs
index c98c39f..118e7c7 100644
--- a/src/sync.rs
+++ b/src/sync.rs
@@ -324,9 +324,17 @@ impl Sync {
handle.fsync().await?;
Ok(handle.status().await?.remote_revision)
}
- /// Export retained local bytes without claiming remote completion.
+ /// Export retained bytes without claiming remote completion.
+ ///
+ /// Content edits read their local staging. A file-creation namespace intent
+ /// reads its immutable prepared descriptor from the authority; other
+ /// namespace intents return Unsupported. Adapters that need offline export
+ /// of a creation conflict must retain their durable local input separately.
pub async fn read(&self, edit: EditId, offset: u64, length: usize) -> Result<Vec<u8>> {
let _gate = self.inner.gate.lock().await;
+ if let Some(bytes) = namespace::read(self, edit, offset, length).await? {
+ return Ok(bytes);
+ }
self.inner
.runtime
.recover(self.handle_id(edit).await?)
diff --git a/src/sync/namespace.rs b/src/sync/namespace.rs
index ac032eb..7fe6d47 100644
--- a/src/sync/namespace.rs
+++ b/src/sync/namespace.rs
@@ -20,6 +20,7 @@ use super::{EditId, Sync};
use crate::runtime::{Error, ErrorKind, Failure, Result};
use borsh::{BorshDeserialize, BorshSerialize};
use rusqlite::{Connection, OptionalExtension, TransactionBehavior, params};
+use std::ops::Range;
use std::pin::Pin;
use std::task::{Context, Poll};
use tokio::io::{AsyncRead, ReadBuf};
@@ -886,12 +887,31 @@ pub(super) async fn statuses(sync: &Sync) -> Result<Vec<NamespaceStatus>> {
rows.iter().map(IntentRow::status).collect()
}
+async fn receipt_for(sync: &Sync, intent: &NamespaceIntent) -> Result<Option<Receipt>> {
+ let receipt = sync
+ .authority()
+ .observe_latest()
+ .await?
+ .receipt(intent.commit_id)
+ .await?;
+ if let Some(receipt) = &receipt
+ && receipt.request_digest != intent.request_digest
+ {
+ return Err(Error::local_kind(
+ ErrorKind::Corrupt,
+ "commit identity returned a different request",
+ ));
+ }
+ Ok(receipt)
+}
+
pub(super) async fn publish(sync: &Sync, id: EditId) -> Result<Revision> {
let row = load_async(sync, id)
.await?
.ok_or_else(|| Error::local_kind(ErrorKind::NotFound, "unknown namespace intent"))?;
let status = row.status()?;
- match status.state {
+ let state = status.state.clone();
+ match &state {
NamespaceState::Committed(receipt) => return Ok(receipt.cursor.revision),
NamespaceState::Conflict | NamespaceState::Failed => {
return Err(Error::Retained(status.failure.ok_or_else(|| {
@@ -900,11 +920,35 @@ pub(super) async fn publish(sync: &Sync, id: EditId) -> Result<Revision> {
}
NamespaceState::Prepared | NamespaceState::Unknown | NamespaceState::Retryable => {}
}
+ if matches!(&state, NamespaceState::Unknown) {
+ match receipt_for(sync, &status.intent).await {
+ Ok(Some(receipt)) => {
+ let revision = receipt.cursor.revision;
+ save_state_async(sync, id, COMMITTED, Some(receipt), None).await?;
+ return Ok(revision);
+ }
+ Ok(None) => {}
+ Err(error) => {
+ let failure = Failure::capture(&error);
+ save_state_async(sync, id, UNKNOWN, None, Some(failure.clone())).await?;
+ return Err(Error::Retained(failure));
+ }
+ }
+ }
let request = match sync.authority().restore_transaction(&row.request).await {
Ok(request) => request,
Err(error) => {
- save_state_async(sync, id, FAILED, None, Some(Failure::capture(&error))).await?;
- return Err(error);
+ let state = match state {
+ NamespaceState::Unknown => UNKNOWN,
+ NamespaceState::Prepared | NamespaceState::Retryable => match error.kind() {
+ ErrorKind::Invalid | ErrorKind::Corrupt | ErrorKind::Unsupported => FAILED,
+ _ => RETRYABLE,
+ },
+ _ => unreachable!("terminal states returned above"),
+ };
+ let failure = Failure::capture(&error);
+ save_state_async(sync, id, state, None, Some(failure.clone())).await?;
+ return Err(Error::Retained(failure));
}
};
// The original request is durable. Recording the possibly-dispatched state
@@ -956,20 +1000,36 @@ pub(super) async fn resolve(sync: &Sync, id: EditId) -> Result<Option<Receipt>>
if let NamespaceState::Committed(receipt) = status.state {
return Ok(Some(receipt));
}
- let receipt = sync
- .authority()
- .observe_latest()
- .await?
- .receipt(status.intent.commit_id)
- .await?;
+ let receipt = receipt_for(sync, &status.intent).await?;
if let Some(receipt) = &receipt {
- if receipt.request_digest != status.intent.request_digest {
- return Err(Error::local_kind(
- ErrorKind::Corrupt,
- "commit identity returned a different request",
- ));
- }
save_state_async(sync, id, COMMITTED, Some(receipt.clone()), None).await?;
}
Ok(receipt)
}
+
+pub(super) async fn read(
+ sync: &Sync,
+ id: EditId,
+ offset: u64,
+ length: usize,
+) -> Result<Option<Vec<u8>>> {
+ let Some(row) = load_async(sync, id).await? else {
+ return Ok(None);
+ };
+ let status = row.status()?;
+ let NamespaceOperation::CreateFile { content, .. } = status.intent.operation else {
+ return Err(Error::State(
+ ErrorKind::Unsupported,
+ "namespace intent does not retain file bytes",
+ ));
+ };
+ let size = content.content_id().length();
+ let start = offset.min(size);
+ let end = offset.saturating_add(length as u64).min(size);
+ let mut bytes = Vec::new();
+ sync.authority()
+ .data()
+ .read_range(&content, Range { start, end }, &mut bytes)
+ .await?;
+ Ok(Some(bytes))
+}
diff --git a/tests/sync.rs b/tests/sync.rs
index 9f5cb02..1952e89 100644
--- a/tests/sync.rs
+++ b/tests/sync.rs
@@ -236,6 +236,7 @@ async fn exercise_namespace(authority: Arc<dyn Authority>) {
);
let status = sync.namespace_status(created).await.unwrap();
assert_eq!(status.state, NamespaceState::Prepared);
+ assert_eq!(sync.read(created, 0, 64).await.unwrap(), b"created bytes");
let NamespaceOperation::CreateFile {
node: created_node,
metadata,
@@ -326,6 +327,10 @@ async fn exercise_namespace(authority: Arc<dyn Authority>) {
.stage_remove(created_node, baseline.revision())
.await
.unwrap();
+ assert_eq!(
+ sync.read(remove, 0, 1).await.unwrap_err().kind(),
+ ErrorKind::Unsupported
+ );
let revision = sync.publish(remove).await.unwrap();
assert!(
authority
@@ -637,3 +642,51 @@ async fn namespace_unknown_retries_the_original_request_after_restart() {
NamespaceState::Committed(_)
));
}
+
+#[tokio::test]
+async fn namespace_unknown_survives_unavailable_recovery_and_resolves_once() {
+ let backend = TestBackend::default();
+ let authority: Arc<dyn Authority> = Arc::new(
+ Fs::create(backend.operator(), BackendProfile::Minio)
+ .await
+ .unwrap(),
+ );
+ let temp = tempfile::tempdir().unwrap();
+ let sync = Sync::open(authority.clone(), temp.path().join("sync"))
+ .await
+ .unwrap();
+ let baseline = authority.observe_latest().await.unwrap();
+ let intent = sync
+ .stage_create_directory(authority.root(), "published-then-lost", baseline.revision())
+ .await
+ .unwrap();
+ backend.lose_next_head_response_and_go_offline();
+ let error = sync.publish(intent).await.unwrap_err();
+ assert_eq!(error.kind(), ErrorKind::Unknown);
+ assert_eq!(
+ sync.namespace_status(intent).await.unwrap().state,
+ NamespaceState::Unknown
+ );
+ drop(sync);
+
+ let sync = Sync::open(authority.clone(), temp.path().join("sync"))
+ .await
+ .unwrap();
+ assert!(sync.resolve_namespace(intent).await.is_err());
+ assert_eq!(
+ sync.namespace_status(intent).await.unwrap().state,
+ NamespaceState::Unknown
+ );
+
+ backend.state.lock().unwrap().offline = false;
+ let receipt = sync.resolve_namespace(intent).await.unwrap().unwrap();
+ assert!(matches!(
+ sync.namespace_status(intent).await.unwrap().state,
+ NamespaceState::Committed(_)
+ ));
+ assert_eq!(
+ authority.observe_latest().await.unwrap().revision(),
+ receipt.cursor.revision
+ );
+ assert_eq!(sync.publish(intent).await.unwrap(), receipt.cursor.revision);
+}
(opendal-yinyang) 01/03: feat: coordinate durable Sync namespace intents
Posted by xu...@apache.org.
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
commit 752e19b95f522df7c78a45ad966d9574866015cf
Author: Xuanwo <gi...@xuanwo.io>
AuthorDate: Tue Oct 27 00:21:46 2026 +0800
feat: coordinate durable Sync namespace intents
Co-authored-by: Chojan Shang <ps...@apache.org>
---
specs/sync-runtime.md | 157 ++++++--
src/runtime/error.rs | 2 +-
src/sync.rs | 119 +++++-
src/sync/namespace.rs | 975 ++++++++++++++++++++++++++++++++++++++++++++++++++
tests/sync.rs | 477 +++++++++++++++++++++++-
5 files changed, 1695 insertions(+), 35 deletions(-)
diff --git a/specs/sync-runtime.md b/specs/sync-runtime.md
index d5d82b5..1fe5340 100644
--- a/specs/sync-runtime.md
+++ b/specs/sync-runtime.md
@@ -1,20 +1,23 @@
-# Conditional Sync edits
+# Durable Sync intents
-**Status: experimental content coordinator, independent of Mount.**
+**Status: experimental content and namespace coordinator, independent of Mount.**
-`yinyang::sync::Sync` accepts full-file replacement bytes against an explicit
-common remote baseline (NodeId and retained Revision). It never substitutes a
-fresh observation after an offline edit. Downloads use `Snapshot::open_file`
-and its immutable verified ranges; enumeration uses the same Snapshot APIs.
-Ordinary local directories and OS-managed provider trees supply their own
-platform callbacks, local edit tracking and materialization.
+`yinyang::sync::Sync` accepts content replacement and namespace operations
+against an explicit common remote baseline. It never substitutes a fresh
+observation after an offline edit. Downloads use `Snapshot::open_file` and its
+immutable verified ranges; enumeration uses the same Snapshot APIs. Ordinary
+local directories and OS-managed provider trees supply their own platform
+callbacks, local edit tracking and materialization.
-`stage` streams bytes into durable FileHandle staging, in 64 KiB chunks. It
-currently materializes the baseline before replacing it. The returned EditId is
-BLAKE3 with derive-key context `yinyang-sync-content-edit-1` over filesystem ID,
-node ID, 24-byte baseline revision and all replacement bytes. Repeating this
-exact intent returns the same local edit, including after restart. An empty
-replacement is still conditional even when the baseline was also empty.
+## Content edits
+
+`stage` streams replacement bytes into durable FileHandle staging, in 64 KiB
+chunks. It currently materializes the baseline before replacing it. The returned
+EditId is BLAKE3 with derive-key context `yinyang-sync-content-edit-1` over
+filesystem ID, node ID, 24-byte baseline revision and all replacement bytes.
+Repeating this exact intent returns the same local edit, including after
+restart. An empty replacement is still conditional even when the baseline was
+also empty.
`publish` uses that edit's original staged baseline and frozen request. Remote
success returns the revision that accepted the edit, never a later observation.
@@ -28,17 +31,117 @@ An adapter may mark a file in-sync only if the currently observed local version
still matches the completed edit. Completion of v1 must not acknowledge a newer
v2. In particular local fsync inside a File Provider domain is not remote fsync.
-The dedicated directory contains `sync.db` (SQLite WAL, synchronous FULL) and
+## Namespace intents
+
+The namespace API accepts only operations expressible by the shared
+Authority/Planner contract:
+
+```text
+stage_create_file(parent, name, baseline, FileMetadata, bytes)
+stage_create_directory(parent, name, baseline)
+stage_remove(node, baseline)
+stage_rename(node, baseline, new_parent, new_name, RenameMode)
+stage_set_executable(node, baseline, executable)
+```
+
+`RenameMode` is explicitly `NoReplace` or `Replace`. `NoReplace` rejects an
+occupied destination. `Replace` removes exactly the destination observed in
+the same baseline snapshot and requires the same kind; a directory destination
+must be empty. The source keeps its NodeId. Existing pinned versions and staged
+handles of the replaced node remain readable but cannot publish back to the
+removed identity.
+
+Creation allocates its stable NodeId in the same Planner that produces the
+persisted transaction. The intent exposes that NodeId; callers must not resolve
+the path again and adopt a concurrent replacement. `CreateFile` also exposes
+the prepared `ContentDescriptor`, allowing a local adapter to compare its
+current bytes without a fresh remote lookup.
+
+`FileMetadata` contains only the executable bit. File creation and subsequent
+`SetExecutable` support that bit. Directory modes, ownership, ACLs, timestamps,
+and other POSIX metadata are outside this contract; adapters must report them
+unsupported before creating an intent. There is no recursive removal. Removing
+a directory requires the observed empty directory and conflicts with a
+concurrent child creation.
+
+Every namespace intent is planned against `Authority::observe_revision` for its
+supplied baseline and persisted with `Transaction::to_bytes` before its first
+commit. Publication restores that exact transaction through the selected
+Authority. It does not call current-head Runtime namespace helpers, rerun a
+callback, choose a newer baseline, or allocate a different NodeId.
+
+The Planner guards each selected name, parent kind, source link, destination,
+and relevant ancestry. Rename additionally guards the complete logical state of
+the source and of an observed replacement target. Remove guards the complete
+source node, its link and directory membership. These dependencies distinguish
+destructive content or executable changes from a safe namespace-only move.
+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.
+- Concurrent renames of one source publish at most one move.
+- Child creation and removal of its observed empty parent publish at most one
+ operation.
+- Replacing a destination conflicts with a concurrent edit, replacement, or new
+ child of that destination.
+
+## Durable state and resolution
+
+The dedicated directory contains `sync.db` (SQLite WAL, `synchronous=FULL`) and
`content/` (the existing runtime's durable staging and exclusive process lock).
-Profile `yinyang-sync-1` binds the journal to one filesystem. The journal maps
-32-byte EditIds to 16-byte staged handle IDs. A mapping is committed only after
-all local bytes are accepted; publication is forbidden before that mapping.
-Interrupted staging can leave unreferenced, never-dispatched content, collected
-on restart or before the next stage. Missing referenced content is an error,
-not permission to replan. Keep both databases and their WALs together.
-
-Operations within one Sync instance serialize to protect intent creation and
-staging leases. This is not a background worker, directory reconciler, OS cache,
-or cross-process concurrent uploader. It does not yet coordinate namespace edits.
-Tests cover both authorities, repeated intent, empty replacement conflicts,
-pinned downloads, retained conflicting bytes, and Unknown recovery after restart.
+Profile `yinyang-sync-2` binds the journal to one filesystem. Opening a valid
+`yinyang-sync-1` journal upgrades it in place without dropping content edit
+records. A mismatched filesystem or unknown profile is an error. Keep both
+databases and their WALs together.
+
+The content edit table maps 32-byte EditIds to 16-byte staged handle IDs. A
+mapping is committed only after all local bytes are accepted; publication is
+forbidden before that mapping. Interrupted staging can leave unreferenced,
+never-dispatched content, collected on restart or before the next stage.
+Missing referenced content is an error, not permission to replan.
+
+The namespace table stores the logical operation, exact `CommitId`, canonical
+request digest, baseline, transaction bytes, durable state, receipt, and typed
+failure. One `EditId` identifies either a content edit or namespace intent.
+`namespace_status` and `namespace_intents` expose the retained logical intent;
+`edit_status` distinguishes the two edit families. Namespace state is:
+
+| State | Meaning |
+| --- | --- |
+| Prepared | Accepted locally and durable, with no dispatch recorded. |
+| Unknown | A dispatch may have published; receipt absence is not failure. |
+| Committed | The original receipt is attached. |
+| Conflict | The baseline-bound predicates no longer hold. |
+| Retryable | The attempt is known not to have published. |
+| Failed | Validation, readiness, corruption, or another definitive failure. |
+
+Before dispatch, Sync changes `Prepared`/`Retryable` to `Unknown` in a durable
+SQLite transaction. A crash or lost response therefore cannot turn a possibly
+published request into a safe abort. `publish` retries `Unknown` and `Retryable`
+with the same transaction and CommitId. It does not dispatch `Conflict` or
+`Failed` again, and never replans either one. A caller that wants a different
+intent must create a new baseline-bound EditId.
+
+`resolve_namespace` checks only the original CommitId's receipt and verifies its
+request digest before attaching it. It never dispatches. Returning no receipt
+does not prove that a delayed attempt failed. Conflict resolution is explicit:
+keep or export the relevant local and pinned baseline bytes, then create a new
+intent with suitable current predicates.
+
+Operations within one Sync instance serialize to protect intent creation,
+staging leases, and state transitions. This is not a background worker,
+directory reconciler, OS cache, or cross-process concurrent uploader. It does
+not implement the ordinary-directory scanner, materialization policy, CLI,
+File Provider callbacks, or a macOS frontend.
+
+## Validation
+
+Tests cover both publication authorities, repeated intent, empty replacement
+conflicts, pinned downloads, retained conflicting bytes, and Unknown recovery
+after restart. Namespace tests cover create file/directory, stable-identity
+rename, no-replace and replace, executable metadata, create/create,
+delete/edit, rename/edit, rename/rename, parent removal, and response-loss
+replay after process reopen.
diff --git a/src/runtime/error.rs b/src/runtime/error.rs
index 137863f..84ddba2 100644
--- a/src/runtime/error.rs
+++ b/src/runtime/error.rs
@@ -72,7 +72,7 @@ impl Failure {
commit: None,
}
}
- pub(super) fn capture(error: &Error) -> Self {
+ pub(crate) fn capture(error: &Error) -> Self {
match error {
Error::Retained(failure) | Error::Local(failure) => failure.clone(),
_ => Self {
diff --git a/src/sync.rs b/src/sync.rs
index f257949..c98c39f 100644
--- a/src/sync.rs
+++ b/src/sync.rs
@@ -23,7 +23,12 @@ use std::path::Path;
use std::sync::Arc;
use tokio::io::{AsyncRead, AsyncReadExt};
use tokio::sync::Mutex;
-use yinyang_core::{Authority, NodeId, Revision};
+use yinyang_core::{Authority, NodeId, Receipt, Revision};
+
+mod namespace;
+pub use namespace::{
+ FileMetadata, NamespaceIntent, NamespaceOperation, NamespaceState, NamespaceStatus, RenameMode,
+};
/// Content-addressed local intent, distinct from the frozen remote commit ID.
/// It binds the filesystem, node, observed revision and exact replacement bytes.
@@ -59,11 +64,38 @@ impl Sync {
c.execute_batch("PRAGMA journal_mode=WAL; PRAGMA synchronous=FULL;
CREATE TABLE IF NOT EXISTS binding(singleton INTEGER PRIMARY KEY CHECK(singleton=1), profile TEXT NOT NULL, filesystem BLOB NOT NULL);
CREATE TABLE IF NOT EXISTS edits(id BLOB PRIMARY KEY, handle BLOB NOT NULL UNIQUE);").map_err(Error::local)?;
+ namespace::initialize(&c)?;
let tx = c.transaction().map_err(Error::local)?;
- tx.execute("INSERT OR IGNORE INTO binding VALUES(1,'yinyang-sync-1',?1)", params![filesystem.as_slice()]).map_err(Error::local)?;
- let (profile, bound): (String, Vec<u8>) = tx.query_row("SELECT profile,filesystem FROM binding", [], |r| Ok((r.get(0)?,r.get(1)?))).map_err(Error::local)?;
- if profile != "yinyang-sync-1" || bound != filesystem {
- return Err(Error::Invalid("sync journal belongs to another filesystem or profile"));
+ let binding: Option<(String, Vec<u8>)> = tx
+ .query_row("SELECT profile,filesystem FROM binding", [], |r| {
+ Ok((r.get(0)?, r.get(1)?))
+ })
+ .optional()
+ .map_err(Error::local)?;
+ match binding {
+ None => {
+ tx.execute(
+ "INSERT INTO binding VALUES(1,'yinyang-sync-2',?1)",
+ params![filesystem.as_slice()],
+ )
+ .map_err(Error::local)?;
+ }
+ Some((profile, bound)) if bound != filesystem => {
+ return Err(Error::Invalid(
+ "sync journal belongs to another filesystem or profile",
+ ));
+ }
+ Some((profile, _)) if profile == "yinyang-sync-1" => {
+ tx.execute("UPDATE binding SET profile='yinyang-sync-2'", [])
+ .map_err(Error::local)?;
+ }
+ Some((profile, _)) => {
+ if profile != "yinyang-sync-2" {
+ return Err(Error::Invalid(
+ "sync journal belongs to another filesystem or profile",
+ ));
+ }
+ }
}
tx.commit().map_err(Error::local)?;
Ok(c)
@@ -193,6 +225,13 @@ impl Sync {
.status()
.await
}
+ /// Inspect either a content edit or a namespace intent.
+ pub async fn edit_status(&self, edit: EditId) -> Result<EditStatus> {
+ if let Some(status) = namespace::status(self, edit).await? {
+ return Ok(EditStatus::Namespace(status));
+ }
+ Ok(EditStatus::Content(self.status(edit).await?))
+ }
pub async fn edits(&self) -> Result<Vec<EditId>> {
self.call(|c| {
let mut q = c
@@ -210,10 +249,73 @@ impl Sync {
})
.await
}
+ /// Durably accept a file creation with its immutable prepared content.
+ pub async fn stage_create_file(
+ &self,
+ parent: NodeId,
+ name: &str,
+ baseline: Revision,
+ metadata: FileMetadata,
+ source: &mut (dyn AsyncRead + Unpin + Send),
+ ) -> Result<EditId> {
+ namespace::stage_create_file(self, parent, name, baseline, metadata, source).await
+ }
+ /// Durably accept directory creation.
+ pub async fn stage_create_directory(
+ &self,
+ parent: NodeId,
+ name: &str,
+ baseline: Revision,
+ ) -> Result<EditId> {
+ namespace::stage_create_directory(self, parent, name, baseline).await
+ }
+ /// Durably accept removal of an observed stable identity.
+ pub async fn stage_remove(&self, node: NodeId, baseline: Revision) -> Result<EditId> {
+ namespace::stage_remove(self, node, baseline).await
+ }
+ /// Durably accept a rename with explicit destination replacement policy.
+ pub async fn stage_rename(
+ &self,
+ node: NodeId,
+ baseline: Revision,
+ new_parent: NodeId,
+ new_name: &str,
+ mode: RenameMode,
+ ) -> Result<EditId> {
+ namespace::stage_rename(self, node, baseline, new_parent, new_name, mode).await
+ }
+ /// Durably accept an executable-bit change to one observed file identity.
+ pub async fn stage_set_executable(
+ &self,
+ node: NodeId,
+ baseline: Revision,
+ executable: bool,
+ ) -> Result<EditId> {
+ namespace::stage_set_executable(self, node, baseline, executable).await
+ }
+ /// Inspect one durable namespace intent and its exact logical request.
+ pub async fn namespace_status(&self, edit: EditId) -> Result<NamespaceStatus> {
+ namespace::status(self, edit)
+ .await?
+ .ok_or_else(|| Error::Invalid("unknown namespace intent"))
+ }
+ /// List all durable namespace intents in stable identity order.
+ pub async fn namespace_intents(&self) -> Result<Vec<NamespaceStatus>> {
+ namespace::statuses(self).await
+ }
+ /// Resolve a namespace intent through the original commit receipt.
+ ///
+ /// Absence is not proof of failure. This method never dispatches or replans.
+ pub async fn resolve_namespace(&self, edit: EditId) -> Result<Option<Receipt>> {
+ namespace::resolve(self, edit).await
+ }
/// Return the revision that accepted this edit, never a later observation.
/// Conflicts and Unknown retain all bytes and their original frozen request.
pub async fn publish(&self, edit: EditId) -> Result<Revision> {
let _gate = self.inner.gate.lock().await;
+ if namespace::exists(self, edit).await? {
+ return namespace::publish(self, edit).await;
+ }
let mut handle = self
.inner
.runtime
@@ -233,3 +335,10 @@ impl Sync {
.await
}
}
+
+/// Status of the two durable edit families supported by Sync.
+#[derive(Clone, Debug)]
+pub enum EditStatus {
+ Content(HandleStatus),
+ Namespace(NamespaceStatus),
+}
diff --git a/src/sync/namespace.rs b/src/sync/namespace.rs
new file mode 100644
index 0000000..ac032eb
--- /dev/null
+++ b/src/sync/namespace.rs
@@ -0,0 +1,975 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Durable namespace intents on top of the shared Authority/Planner contract.
+use super::{EditId, Sync};
+use crate::runtime::{Error, ErrorKind, Failure, Result};
+use borsh::{BorshDeserialize, BorshSerialize};
+use rusqlite::{Connection, OptionalExtension, TransactionBehavior, params};
+use std::pin::Pin;
+use std::task::{Context, Poll};
+use tokio::io::{AsyncRead, ReadBuf};
+use yinyang_core::{ContentDescriptor, NodeId, Planner, Receipt, Revision, Transaction};
+
+const PREPARED: u8 = 0;
+const UNKNOWN: u8 = 1;
+const COMMITTED: u8 = 2;
+const CONFLICT: u8 = 3;
+const RETRYABLE: u8 = 4;
+const FAILED: u8 = 5;
+
+/// Creation metadata represented by the current namespace contract.
+///
+/// The core persists only the executable bit. Adapters must reject ownership,
+/// timestamps, ACLs, or richer mode bits before creating a sync intent.
+#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+pub struct FileMetadata {
+ pub executable: bool,
+}
+
+impl FileMetadata {
+ pub const fn new(executable: bool) -> Self {
+ Self { executable }
+ }
+}
+
+/// Destination behavior for a rename.
+#[derive(Clone, Copy, Debug, Eq, PartialEq, BorshSerialize, BorshDeserialize)]
+pub enum RenameMode {
+ NoReplace,
+ Replace,
+}
+
+/// The logical request retained by a namespace intent.
+#[derive(Clone, Debug, Eq, PartialEq)]
+pub enum NamespaceOperation {
+ CreateFile {
+ node: NodeId,
+ parent: NodeId,
+ name: String,
+ metadata: FileMetadata,
+ content: ContentDescriptor,
+ },
+ CreateDirectory {
+ node: NodeId,
+ parent: NodeId,
+ name: String,
+ },
+ Remove {
+ node: NodeId,
+ },
+ Rename {
+ node: NodeId,
+ new_parent: NodeId,
+ new_name: String,
+ mode: RenameMode,
+ replaced: Option<NodeId>,
+ },
+ SetExecutable {
+ node: NodeId,
+ executable: bool,
+ },
+}
+
+impl NamespaceOperation {
+ /// The created or source identity this intent operates on.
+ pub const fn node_id(&self) -> NodeId {
+ match self {
+ Self::CreateFile { node, .. }
+ | Self::CreateDirectory { node, .. }
+ | Self::Remove { node }
+ | Self::Rename { node, .. }
+ | Self::SetExecutable { node, .. } => *node,
+ }
+ }
+}
+
+/// Stable identity and baseline-bound request for one namespace operation.
+#[derive(Clone, Debug, Eq, PartialEq)]
+pub struct NamespaceIntent {
+ pub id: EditId,
+ pub baseline: Revision,
+ pub commit_id: yinyang_core::CommitId,
+ pub request_digest: [u8; 32],
+ pub operation: NamespaceOperation,
+}
+
+/// Durable local outcome of a namespace intent.
+#[derive(Clone, Debug, Eq, PartialEq)]
+pub enum NamespaceState {
+ /// Accepted locally and persisted before its first dispatch.
+ Prepared,
+ /// A dispatch may have published and no definitive outcome is stored.
+ Unknown,
+ Committed(Receipt),
+ Conflict,
+ Retryable,
+ /// Validation, readiness, corruption, or another definitive failure.
+ Failed,
+}
+
+/// Reopen-safe status for a namespace intent.
+#[derive(Clone, Debug, Eq, PartialEq)]
+pub struct NamespaceStatus {
+ pub intent: NamespaceIntent,
+ pub state: NamespaceState,
+ pub failure: Option<Failure>,
+}
+
+#[derive(BorshSerialize)]
+enum RequestKey {
+ CreateFile {
+ parent: [u8; 16],
+ name: String,
+ executable: bool,
+ length: u64,
+ digest: [u8; 32],
+ },
+ CreateDirectory {
+ parent: [u8; 16],
+ name: String,
+ },
+ Remove {
+ node: [u8; 16],
+ },
+ Rename {
+ node: [u8; 16],
+ new_parent: [u8; 16],
+ new_name: String,
+ mode: RenameMode,
+ },
+ SetExecutable {
+ node: [u8; 16],
+ executable: bool,
+ },
+}
+
+#[derive(BorshSerialize, BorshDeserialize)]
+enum OperationWire {
+ CreateFile {
+ node: [u8; 16],
+ parent: [u8; 16],
+ name: String,
+ executable: bool,
+ content: Vec<u8>,
+ },
+ CreateDirectory {
+ node: [u8; 16],
+ parent: [u8; 16],
+ name: String,
+ },
+ Remove {
+ node: [u8; 16],
+ },
+ Rename {
+ node: [u8; 16],
+ new_parent: [u8; 16],
+ new_name: String,
+ mode: RenameMode,
+ replaced: Option<[u8; 16]>,
+ },
+ SetExecutable {
+ node: [u8; 16],
+ executable: bool,
+ },
+}
+
+impl OperationWire {
+ fn encode(operation: &NamespaceOperation) -> Result<Vec<u8>> {
+ let wire = match operation {
+ NamespaceOperation::CreateFile {
+ node,
+ parent,
+ name,
+ metadata,
+ content,
+ } => Self::CreateFile {
+ node: *node.as_bytes(),
+ parent: *parent.as_bytes(),
+ name: name.clone(),
+ executable: metadata.executable,
+ content: content.to_bytes(),
+ },
+ NamespaceOperation::CreateDirectory { node, parent, name } => Self::CreateDirectory {
+ node: *node.as_bytes(),
+ parent: *parent.as_bytes(),
+ name: name.clone(),
+ },
+ NamespaceOperation::Remove { node } => Self::Remove {
+ node: *node.as_bytes(),
+ },
+ NamespaceOperation::Rename {
+ node,
+ new_parent,
+ new_name,
+ mode,
+ replaced,
+ } => Self::Rename {
+ node: *node.as_bytes(),
+ new_parent: *new_parent.as_bytes(),
+ new_name: new_name.clone(),
+ mode: *mode,
+ replaced: replaced.map(|id| *id.as_bytes()),
+ },
+ NamespaceOperation::SetExecutable { node, executable } => Self::SetExecutable {
+ node: *node.as_bytes(),
+ executable: *executable,
+ },
+ };
+ borsh::to_vec(&wire).map_err(Error::local)
+ }
+
+ fn decode(bytes: &[u8]) -> Result<NamespaceOperation> {
+ let wire = borsh::from_slice(bytes)
+ .map_err(|error| Error::local_kind(ErrorKind::Corrupt, error))?;
+ match wire {
+ Self::CreateFile {
+ node,
+ parent,
+ name,
+ executable,
+ content,
+ } => Ok(NamespaceOperation::CreateFile {
+ node: NodeId::from_bytes(node),
+ parent: NodeId::from_bytes(parent),
+ name,
+ metadata: FileMetadata { executable },
+ content: ContentDescriptor::from_bytes(&content)?,
+ }),
+ Self::CreateDirectory { node, parent, name } => {
+ Ok(NamespaceOperation::CreateDirectory {
+ node: NodeId::from_bytes(node),
+ parent: NodeId::from_bytes(parent),
+ name,
+ })
+ }
+ Self::Remove { node } => Ok(NamespaceOperation::Remove {
+ node: NodeId::from_bytes(node),
+ }),
+ Self::Rename {
+ node,
+ new_parent,
+ new_name,
+ mode,
+ replaced,
+ } => Ok(NamespaceOperation::Rename {
+ node: NodeId::from_bytes(node),
+ new_parent: NodeId::from_bytes(new_parent),
+ new_name,
+ mode,
+ replaced: replaced.map(NodeId::from_bytes),
+ }),
+ Self::SetExecutable { node, executable } => Ok(NamespaceOperation::SetExecutable {
+ node: NodeId::from_bytes(node),
+ executable,
+ }),
+ }
+ }
+}
+
+pub(super) fn initialize(connection: &Connection) -> Result<()> {
+ connection
+ .execute_batch(
+ "CREATE TABLE IF NOT EXISTS namespaces(
+ id BLOB PRIMARY KEY,
+ intent_key BLOB NOT NULL,
+ baseline BLOB NOT NULL,
+ commit_id BLOB NOT NULL UNIQUE,
+ request_digest BLOB NOT NULL,
+ operation BLOB NOT NULL,
+ request BLOB NOT NULL,
+ state INTEGER NOT NULL,
+ receipt_revision BLOB,
+ receipt_ordinal INTEGER,
+ failure BLOB
+ );",
+ )
+ .map_err(Error::local)
+}
+
+#[derive(Clone)]
+struct IntentRow {
+ id: Vec<u8>,
+ intent_key: Vec<u8>,
+ baseline: Vec<u8>,
+ commit_id: Vec<u8>,
+ request_digest: Vec<u8>,
+ operation: Vec<u8>,
+ request: Vec<u8>,
+ state: u8,
+ receipt_revision: Option<Vec<u8>>,
+ receipt_ordinal: Option<i64>,
+ failure: Option<Vec<u8>>,
+}
+
+impl IntentRow {
+ fn status(&self) -> Result<NamespaceStatus> {
+ let id: [u8; 32] = self
+ .id
+ .as_slice()
+ .try_into()
+ .map_err(|_| Error::local_kind(ErrorKind::Corrupt, "invalid intent identity"))?;
+ let baseline: [u8; 24] = self
+ .baseline
+ .as_slice()
+ .try_into()
+ .map_err(|_| Error::local_kind(ErrorKind::Corrupt, "invalid intent baseline"))?;
+ let commit_id: [u8; 16] = self
+ .commit_id
+ .as_slice()
+ .try_into()
+ .map_err(|_| Error::local_kind(ErrorKind::Corrupt, "invalid commit identity"))?;
+ let request_digest: [u8; 32] = self
+ .request_digest
+ .as_slice()
+ .try_into()
+ .map_err(|_| Error::local_kind(ErrorKind::Corrupt, "invalid request digest"))?;
+ let state = match self.state {
+ PREPARED => NamespaceState::Prepared,
+ UNKNOWN => NamespaceState::Unknown,
+ COMMITTED => {
+ let revision = self.receipt_revision.as_deref().ok_or_else(|| {
+ Error::local_kind(ErrorKind::Corrupt, "missing receipt revision")
+ })?;
+ let ordinal = self.receipt_ordinal.ok_or_else(|| {
+ Error::local_kind(ErrorKind::Corrupt, "missing receipt ordinal")
+ })?;
+ let revision: [u8; 24] = revision.try_into().map_err(|_| {
+ Error::local_kind(ErrorKind::Corrupt, "invalid receipt revision")
+ })?;
+ let ordinal = u32::try_from(ordinal).map_err(|_| {
+ Error::local_kind(ErrorKind::Corrupt, "invalid receipt ordinal")
+ })?;
+ NamespaceState::Committed(Receipt {
+ commit_id: yinyang_core::CommitId::from_bytes(commit_id),
+ request_digest,
+ cursor: yinyang_core::Cursor {
+ revision: Revision::from_bytes(revision),
+ ordinal,
+ },
+ })
+ }
+ CONFLICT => NamespaceState::Conflict,
+ RETRYABLE => NamespaceState::Retryable,
+ FAILED => NamespaceState::Failed,
+ _ => {
+ return Err(Error::local_kind(
+ ErrorKind::Corrupt,
+ "invalid namespace state",
+ ));
+ }
+ };
+ if !matches!(state, NamespaceState::Committed(_))
+ && (self.receipt_revision.is_some() || self.receipt_ordinal.is_some())
+ {
+ return Err(Error::local_kind(
+ ErrorKind::Corrupt,
+ "receipt fields exist outside a committed state",
+ ));
+ }
+ let failure = self
+ .failure
+ .as_deref()
+ .map(borsh::from_slice)
+ .transpose()
+ .map_err(|error| Error::local_kind(ErrorKind::Corrupt, error))?;
+ Ok(NamespaceStatus {
+ intent: NamespaceIntent {
+ id: EditId::from_bytes(id),
+ baseline: Revision::from_bytes(baseline),
+ commit_id: yinyang_core::CommitId::from_bytes(commit_id),
+ request_digest,
+ operation: OperationWire::decode(&self.operation)?,
+ },
+ state,
+ failure,
+ })
+ }
+}
+
+fn row_from_sql(row: &rusqlite::Row<'_>) -> rusqlite::Result<IntentRow> {
+ Ok(IntentRow {
+ id: row.get(0)?,
+ intent_key: row.get(1)?,
+ baseline: row.get(2)?,
+ commit_id: row.get(3)?,
+ request_digest: row.get(4)?,
+ operation: row.get(5)?,
+ request: row.get(6)?,
+ state: row.get(7)?,
+ receipt_revision: row.get(8)?,
+ receipt_ordinal: row.get(9)?,
+ failure: row.get(10)?,
+ })
+}
+
+fn load(connection: &Connection, id: EditId) -> Result<Option<IntentRow>> {
+ connection
+ .query_row(
+ "SELECT id,intent_key,baseline,commit_id,request_digest,operation,request,
+ state,receipt_revision,receipt_ordinal,failure
+ FROM namespaces WHERE id=?1",
+ params![id.as_bytes().as_slice()],
+ row_from_sql,
+ )
+ .optional()
+ .map_err(Error::local)
+}
+
+async fn load_async(sync: &Sync, id: EditId) -> Result<Option<IntentRow>> {
+ sync.call(move |connection| load(connection, id)).await
+}
+
+fn save_state(
+ connection: &Connection,
+ id: EditId,
+ state: u8,
+ receipt: Option<&Receipt>,
+ failure: Option<&Failure>,
+) -> Result<()> {
+ let receipt_revision = receipt.map(|receipt| receipt.cursor.revision.to_bytes().to_vec());
+ let receipt_ordinal = receipt.map(|receipt| i64::from(receipt.cursor.ordinal));
+ let failure = failure
+ .map(borsh::to_vec)
+ .transpose()
+ .map_err(Error::local)?;
+ let changed = connection
+ .execute(
+ "UPDATE namespaces
+ SET state=?2,receipt_revision=?3,receipt_ordinal=?4,failure=?5
+ WHERE id=?1",
+ params![
+ id.as_bytes().as_slice(),
+ state,
+ receipt_revision,
+ receipt_ordinal,
+ failure
+ ],
+ )
+ .map_err(Error::local)?;
+ if changed != 1 {
+ return Err(Error::local_kind(
+ ErrorKind::NotFound,
+ "namespace intent disappeared",
+ ));
+ }
+ Ok(())
+}
+
+async fn save_state_async(
+ sync: &Sync,
+ id: EditId,
+ state: u8,
+ receipt: Option<Receipt>,
+ failure: Option<Failure>,
+) -> Result<()> {
+ sync.call(move |connection| {
+ let tx = connection
+ .transaction_with_behavior(TransactionBehavior::Immediate)
+ .map_err(Error::local)?;
+ save_state(&tx, id, state, receipt.as_ref(), failure.as_ref())?;
+ tx.commit().map_err(Error::local)
+ })
+ .await
+}
+
+fn request_key(key: &RequestKey) -> Result<Vec<u8>> {
+ borsh::to_vec(key).map_err(Error::local)
+}
+
+fn intent_id(baseline: Revision, key: &[u8], filesystem: NodeId) -> EditId {
+ let mut hash = blake3::Hasher::new_derive_key("yinyang-sync-namespace-intent-1");
+ hash.update(filesystem.as_bytes());
+ hash.update(&baseline.to_bytes());
+ hash.update(key);
+ EditId::from_bytes(*hash.finalize().as_bytes())
+}
+
+fn commit_id(id: EditId) -> yinyang_core::CommitId {
+ let digest = blake3::derive_key("yinyang-sync-namespace-commit-1", id.as_bytes());
+ yinyang_core::CommitId::from_bytes(digest[..16].try_into().expect("fixed commit prefix"))
+}
+
+struct ContentDigest {
+ hash: blake3::Hasher,
+ length: u64,
+}
+
+struct HashingReader<'a> {
+ source: &'a mut (dyn AsyncRead + Unpin + Send),
+ digest: ContentDigest,
+}
+
+impl<'a> HashingReader<'a> {
+ fn new(source: &'a mut (dyn AsyncRead + Unpin + Send)) -> Self {
+ Self {
+ source,
+ digest: ContentDigest {
+ hash: blake3::Hasher::new(),
+ length: 0,
+ },
+ }
+ }
+
+ fn finish(self) -> (u64, [u8; 32]) {
+ (self.digest.length, *self.digest.hash.finalize().as_bytes())
+ }
+}
+
+impl AsyncRead for HashingReader<'_> {
+ fn poll_read(
+ self: Pin<&mut Self>,
+ cx: &mut Context<'_>,
+ buffer: &mut ReadBuf<'_>,
+ ) -> Poll<std::io::Result<()>> {
+ let this = self.get_mut();
+ let before = buffer.filled().len();
+ match Pin::new(&mut *this.source).poll_read(cx, buffer) {
+ Poll::Ready(Ok(())) => {
+ let bytes = &buffer.filled()[before..];
+ this.digest.hash.update(bytes);
+ this.digest.length = this
+ .digest
+ .length
+ .checked_add(bytes.len() as u64)
+ .ok_or_else(|| std::io::Error::other("namespace file length overflowed"))?;
+ Poll::Ready(Ok(()))
+ }
+ other => other,
+ }
+ }
+}
+
+async fn insert(
+ sync: &Sync,
+ id: EditId,
+ key: Vec<u8>,
+ baseline: Revision,
+ commit_id: yinyang_core::CommitId,
+ operation: Vec<u8>,
+ request: &Transaction,
+) -> Result<()> {
+ let request_digest = request.digest();
+ let request = request.to_bytes()?;
+ sync.call(move |connection| {
+ let existing = load(connection, id)?;
+ if let Some(existing) = existing {
+ if existing.intent_key != key {
+ return Err(Error::local_kind(
+ ErrorKind::Invalid,
+ "namespace intent identity collision",
+ ));
+ }
+ return Ok(());
+ }
+ let tx = connection
+ .transaction_with_behavior(TransactionBehavior::Immediate)
+ .map_err(Error::local)?;
+ tx.execute(
+ "INSERT INTO namespaces(
+ id,intent_key,baseline,commit_id,request_digest,operation,request,state
+ ) VALUES(?1,?2,?3,?4,?5,?6,?7,?8)",
+ params![
+ id.as_bytes().as_slice(),
+ key,
+ baseline.to_bytes().as_slice(),
+ commit_id.as_bytes().as_slice(),
+ request_digest.as_slice(),
+ operation,
+ request,
+ PREPARED
+ ],
+ )
+ .map_err(Error::local)?;
+ tx.commit().map_err(Error::local)
+ })
+ .await
+}
+
+/// Durably accept a file creation intent and its immutable bytes.
+pub async fn stage_create_file(
+ sync: &Sync,
+ parent: NodeId,
+ name: &str,
+ baseline: Revision,
+ metadata: FileMetadata,
+ source: &mut (dyn AsyncRead + Unpin + Send),
+) -> Result<EditId> {
+ let _gate = sync.inner.gate.lock().await;
+ sync.collect_incomplete().await?;
+ let snapshot = sync.authority().observe_revision(baseline).await?;
+ let mut reader = HashingReader::new(source);
+ let prepared = sync.authority().prepare(&mut reader).await?;
+ let (length, digest) = reader.finish();
+ let key = RequestKey::CreateFile {
+ parent: *parent.as_bytes(),
+ name: name.to_owned(),
+ executable: metadata.executable,
+ length,
+ digest,
+ };
+ let key = request_key(&key)?;
+ let id = intent_id(baseline, &key, sync.authority().filesystem());
+ if let Some(existing) = load_async(sync, id).await? {
+ if existing.intent_key != key {
+ return Err(Error::local_kind(
+ ErrorKind::Invalid,
+ "namespace intent identity collision",
+ ));
+ }
+ return Ok(id);
+ }
+ let mut planner = Planner::new(&snapshot, commit_id(id));
+ let node = planner
+ .create_file(parent, name, prepared.clone(), metadata.executable)
+ .await?;
+ let request = planner.finish()?;
+ let operation = NamespaceOperation::CreateFile {
+ node,
+ parent,
+ name: name.to_owned(),
+ metadata,
+ content: prepared.descriptor().clone(),
+ };
+ insert(
+ sync,
+ id,
+ key,
+ baseline,
+ commit_id(id),
+ OperationWire::encode(&operation)?,
+ &request,
+ )
+ .await?;
+ Ok(id)
+}
+
+/// Durably accept a directory creation intent.
+pub async fn stage_create_directory(
+ sync: &Sync,
+ parent: NodeId,
+ name: &str,
+ baseline: Revision,
+) -> Result<EditId> {
+ let _gate = sync.inner.gate.lock().await;
+ sync.collect_incomplete().await?;
+ let snapshot = sync.authority().observe_revision(baseline).await?;
+ let key = RequestKey::CreateDirectory {
+ parent: *parent.as_bytes(),
+ name: name.to_owned(),
+ };
+ let key = request_key(&key)?;
+ let id = intent_id(baseline, &key, sync.authority().filesystem());
+ if let Some(existing) = load_async(sync, id).await? {
+ if existing.intent_key != key {
+ return Err(Error::local_kind(
+ ErrorKind::Invalid,
+ "namespace intent identity collision",
+ ));
+ }
+ return Ok(id);
+ }
+ let mut planner = Planner::new(&snapshot, commit_id(id));
+ let node = planner.create_directory(parent, name).await?;
+ let request = planner.finish()?;
+ let operation = NamespaceOperation::CreateDirectory {
+ node,
+ parent,
+ name: name.to_owned(),
+ };
+ insert(
+ sync,
+ id,
+ key,
+ baseline,
+ commit_id(id),
+ OperationWire::encode(&operation)?,
+ &request,
+ )
+ .await?;
+ Ok(id)
+}
+
+/// Durably accept removal of an observed stable identity.
+pub async fn stage_remove(sync: &Sync, node: NodeId, baseline: Revision) -> Result<EditId> {
+ let _gate = sync.inner.gate.lock().await;
+ sync.collect_incomplete().await?;
+ let snapshot = sync.authority().observe_revision(baseline).await?;
+ let key = RequestKey::Remove {
+ node: *node.as_bytes(),
+ };
+ let key = request_key(&key)?;
+ let id = intent_id(baseline, &key, sync.authority().filesystem());
+ if let Some(existing) = load_async(sync, id).await? {
+ if existing.intent_key != key {
+ return Err(Error::local_kind(
+ ErrorKind::Invalid,
+ "namespace intent identity collision",
+ ));
+ }
+ return Ok(id);
+ }
+ let mut planner = Planner::new(&snapshot, commit_id(id));
+ // Remove already guards state, link, entry, and emptiness. The logical
+ // guard also makes the whole retained node state an explicit dependency.
+ let _ = planner.node(node).await?;
+ planner.remove(node).await?;
+ let request = planner.finish()?;
+ let operation = NamespaceOperation::Remove { node };
+ insert(
+ sync,
+ id,
+ key,
+ baseline,
+ commit_id(id),
+ OperationWire::encode(&operation)?,
+ &request,
+ )
+ .await?;
+ Ok(id)
+}
+
+/// Durably accept a rename, optionally replacing one observed destination.
+pub async fn stage_rename(
+ sync: &Sync,
+ node: NodeId,
+ baseline: Revision,
+ new_parent: NodeId,
+ new_name: &str,
+ mode: RenameMode,
+) -> Result<EditId> {
+ let _gate = sync.inner.gate.lock().await;
+ sync.collect_incomplete().await?;
+ let snapshot = sync.authority().observe_revision(baseline).await?;
+ let key = RequestKey::Rename {
+ node: *node.as_bytes(),
+ new_parent: *new_parent.as_bytes(),
+ new_name: new_name.to_owned(),
+ mode,
+ };
+ let key = request_key(&key)?;
+ let id = intent_id(baseline, &key, sync.authority().filesystem());
+ if let Some(existing) = load_async(sync, id).await? {
+ if existing.intent_key != key {
+ return Err(Error::local_kind(
+ ErrorKind::Invalid,
+ "namespace intent identity collision",
+ ));
+ }
+ return Ok(id);
+ }
+ let mut planner = Planner::new(&snapshot, commit_id(id));
+ // Rename's built-in guards preserve reachability. Guarding the complete
+ // source state prevents a concurrent content or executable edit from
+ // being silently carried across the move.
+ let _ = planner.node(node).await?;
+ let destination = planner.lookup(new_parent, new_name).await?;
+ let replaced = destination
+ .as_ref()
+ .filter(|entry| entry.node_id != node)
+ .map(|entry| entry.node_id);
+ if let Some(replaced) = replaced {
+ let _ = planner.node(replaced).await?;
+ }
+ match mode {
+ RenameMode::NoReplace => planner.rename(node, new_parent, new_name).await?,
+ RenameMode::Replace => planner.rename_replace(node, new_parent, new_name).await?,
+ }
+ let request = planner.finish()?;
+ let operation = NamespaceOperation::Rename {
+ node,
+ new_parent,
+ new_name: new_name.to_owned(),
+ mode,
+ replaced,
+ };
+ insert(
+ sync,
+ id,
+ key,
+ baseline,
+ commit_id(id),
+ OperationWire::encode(&operation)?,
+ &request,
+ )
+ .await?;
+ Ok(id)
+}
+
+/// Durably accept an executable-bit change to one observed file identity.
+pub async fn stage_set_executable(
+ sync: &Sync,
+ node: NodeId,
+ baseline: Revision,
+ executable: bool,
+) -> Result<EditId> {
+ let _gate = sync.inner.gate.lock().await;
+ sync.collect_incomplete().await?;
+ let snapshot = sync.authority().observe_revision(baseline).await?;
+ let key = RequestKey::SetExecutable {
+ node: *node.as_bytes(),
+ executable,
+ };
+ let key = request_key(&key)?;
+ let id = intent_id(baseline, &key, sync.authority().filesystem());
+ if let Some(existing) = load_async(sync, id).await? {
+ if existing.intent_key != key {
+ return Err(Error::local_kind(
+ ErrorKind::Invalid,
+ "namespace intent identity collision",
+ ));
+ }
+ return Ok(id);
+ }
+ let mut planner = Planner::new(&snapshot, commit_id(id));
+ let _ = planner.node(node).await?;
+ planner.set_executable(node, executable).await?;
+ let request = planner.finish()?;
+ let operation = NamespaceOperation::SetExecutable { node, executable };
+ insert(
+ sync,
+ id,
+ key,
+ baseline,
+ commit_id(id),
+ OperationWire::encode(&operation)?,
+ &request,
+ )
+ .await?;
+ Ok(id)
+}
+
+pub(super) async fn status(sync: &Sync, id: EditId) -> Result<Option<NamespaceStatus>> {
+ let _gate = sync.inner.gate.lock().await;
+ load_async(sync, id)
+ .await?
+ .map(|row| row.status())
+ .transpose()
+}
+
+pub(super) async fn exists(sync: &Sync, id: EditId) -> Result<bool> {
+ Ok(load_async(sync, id).await?.is_some())
+}
+
+pub(super) async fn statuses(sync: &Sync) -> Result<Vec<NamespaceStatus>> {
+ let _gate = sync.inner.gate.lock().await;
+ let rows = sync
+ .call(|connection| {
+ let mut statement = connection
+ .prepare(
+ "SELECT id,intent_key,baseline,commit_id,request_digest,operation,request,
+ state,receipt_revision,receipt_ordinal,failure
+ FROM namespaces ORDER BY id",
+ )
+ .map_err(Error::local)?;
+ let rows = statement
+ .query_map([], row_from_sql)
+ .map_err(Error::local)?;
+ rows.collect::<rusqlite::Result<Vec<_>>>()
+ .map_err(Error::local)
+ })
+ .await?;
+ rows.iter().map(IntentRow::status).collect()
+}
+
+pub(super) async fn publish(sync: &Sync, id: EditId) -> Result<Revision> {
+ let row = load_async(sync, id)
+ .await?
+ .ok_or_else(|| Error::local_kind(ErrorKind::NotFound, "unknown namespace intent"))?;
+ let status = row.status()?;
+ match status.state {
+ NamespaceState::Committed(receipt) => return Ok(receipt.cursor.revision),
+ NamespaceState::Conflict | NamespaceState::Failed => {
+ return Err(Error::Retained(status.failure.ok_or_else(|| {
+ Error::local_kind(ErrorKind::Corrupt, "terminal intent has no failure")
+ })?));
+ }
+ NamespaceState::Prepared | NamespaceState::Unknown | NamespaceState::Retryable => {}
+ }
+ let request = match sync.authority().restore_transaction(&row.request).await {
+ Ok(request) => request,
+ Err(error) => {
+ save_state_async(sync, id, FAILED, None, Some(Failure::capture(&error))).await?;
+ return Err(error);
+ }
+ };
+ // The original request is durable. Recording the possibly-dispatched state
+ // before dispatch means a process crash cannot turn this into a safe abort.
+ save_state_async(sync, id, UNKNOWN, None, None).await?;
+ match sync.authority().commit(&request).await {
+ Ok(yinyang_core::CommitOutcome::Committed(receipt)) => {
+ save_state_async(sync, id, COMMITTED, Some(receipt.clone()), None).await?;
+ Ok(receipt.cursor.revision)
+ }
+ Ok(yinyang_core::CommitOutcome::Conflict) => {
+ let failure = Failure::capture(&Error::State(
+ ErrorKind::Conflict,
+ "namespace predicates changed; the original intent is retained",
+ ));
+ save_state_async(sync, id, CONFLICT, None, Some(failure.clone())).await?;
+ Err(Error::Retained(failure))
+ }
+ Ok(yinyang_core::CommitOutcome::Retryable) => {
+ let failure = Failure::capture(&Error::Retryable);
+ save_state_async(sync, id, RETRYABLE, None, Some(failure.clone())).await?;
+ Err(Error::Retained(failure))
+ }
+ Ok(yinyang_core::CommitOutcome::Unknown(commit)) => {
+ let failure = Failure::capture(&Error::Unknown(commit));
+ save_state_async(sync, id, UNKNOWN, None, Some(failure.clone())).await?;
+ 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),
+ };
+ let failure = Failure::capture(&failure);
+ save_state_async(sync, id, state, None, Some(failure.clone())).await?;
+ Err(Error::Retained(failure))
+ }
+ }
+}
+
+pub(super) async fn resolve(sync: &Sync, id: EditId) -> Result<Option<Receipt>> {
+ let _gate = sync.inner.gate.lock().await;
+ let row = load_async(sync, id)
+ .await?
+ .ok_or_else(|| Error::local_kind(ErrorKind::NotFound, "unknown namespace intent"))?;
+ let status = row.status()?;
+ if let NamespaceState::Committed(receipt) = status.state {
+ return Ok(Some(receipt));
+ }
+ let receipt = sync
+ .authority()
+ .observe_latest()
+ .await?
+ .receipt(status.intent.commit_id)
+ .await?;
+ if let Some(receipt) = &receipt {
+ if receipt.request_digest != status.intent.request_digest {
+ return Err(Error::local_kind(
+ ErrorKind::Corrupt,
+ "commit identity returned a different request",
+ ));
+ }
+ save_state_async(sync, id, COMMITTED, Some(receipt.clone()), None).await?;
+ }
+ Ok(receipt)
+}
diff --git a/tests/sync.rs b/tests/sync.rs
index df89bd9..9f5cb02 100644
--- a/tests/sync.rs
+++ b/tests/sync.rs
@@ -20,9 +20,11 @@ mod support;
use std::sync::Arc;
use support::TestBackend;
use yinyang::core::service::{MetadataService, ServiceClient, serve};
-use yinyang::core::{Authority, BackendProfile, Fs};
+use yinyang::core::{Authority, BackendProfile, Fs, NodeId, Revision};
use yinyang::runtime::{ErrorKind, Runtime};
-use yinyang::sync::Sync;
+use yinyang::sync::{
+ EditStatus, FileMetadata, NamespaceOperation, NamespaceState, RenameMode, Sync,
+};
async fn exercise(authority: Arc<dyn Authority>) {
let temp = tempfile::tempdir().unwrap();
@@ -164,3 +166,474 @@ async fn original_upload_survives_unknown_and_restart() {
.unwrap();
assert_eq!(revision, receipt.cursor.revision);
}
+
+async fn create_file(
+ authority: &Arc<dyn Authority>,
+ runtime: &Runtime,
+ parent: NodeId,
+ name: &str,
+) -> (Revision, NodeId) {
+ let (node, _) = runtime
+ .create_file_with_node(parent, name, false)
+ .await
+ .unwrap();
+ let baseline = authority.observe_latest().await.unwrap();
+ (baseline.revision(), node)
+}
+
+async fn create_directory(
+ authority: &Arc<dyn Authority>,
+ runtime: &Runtime,
+ parent: NodeId,
+ name: &str,
+) -> (Revision, NodeId) {
+ let (node, _) = runtime
+ .create_directory_with_node(parent, name)
+ .await
+ .unwrap();
+ let baseline = authority.observe_latest().await.unwrap();
+ (baseline.revision(), node)
+}
+
+fn assert_conflict(result: yinyang::runtime::Result<Revision>) {
+ assert_eq!(result.unwrap_err().kind(), ErrorKind::Conflict);
+}
+
+async fn exercise_namespace(authority: Arc<dyn Authority>) {
+ let temp = tempfile::tempdir().unwrap();
+ let runtime = Runtime::open(authority.clone(), temp.path().join("editor"))
+ .await
+ .unwrap();
+ let sync = Sync::open(authority.clone(), temp.path().join("sync"))
+ .await
+ .unwrap();
+ let root = authority.root();
+
+ // Create with bytes and executable metadata. The durable identity survives
+ // restart and the exact intent resolves to the same commit.
+ let baseline = authority.observe_latest().await.unwrap();
+ let created = sync
+ .stage_create_file(
+ root,
+ "created",
+ baseline.revision(),
+ FileMetadata::new(true),
+ &mut &b"created bytes"[..],
+ )
+ .await
+ .unwrap();
+ assert_eq!(
+ sync.stage_create_file(
+ root,
+ "created",
+ baseline.revision(),
+ FileMetadata::new(true),
+ &mut &b"created bytes"[..],
+ )
+ .await
+ .unwrap(),
+ created
+ );
+ let status = sync.namespace_status(created).await.unwrap();
+ assert_eq!(status.state, NamespaceState::Prepared);
+ let NamespaceOperation::CreateFile {
+ node: created_node,
+ metadata,
+ ..
+ } = status.intent.operation
+ else {
+ panic!("expected file creation intent");
+ };
+ assert!(metadata.executable);
+ assert!(matches!(
+ sync.edit_status(created).await.unwrap(),
+ EditStatus::Namespace(_)
+ ));
+ assert!(
+ sync.namespace_intents()
+ .await
+ .unwrap()
+ .iter()
+ .any(|intent| intent.intent.id == created)
+ );
+ let revision = sync.publish(created).await.unwrap();
+ let committed = authority.observe_revision(revision).await.unwrap();
+ let file = committed.open_file(created_node).await.unwrap();
+ assert!(file.node().executable());
+ assert_eq!(file.read(0, 64).await.unwrap(), b"created bytes");
+ let status = sync.namespace_status(created).await.unwrap();
+ assert!(matches!(status.state, NamespaceState::Committed(_)));
+ drop(sync);
+ let sync = Sync::open(authority.clone(), temp.path().join("sync"))
+ .await
+ .unwrap();
+ assert!(matches!(
+ sync.namespace_status(created).await.unwrap().state,
+ NamespaceState::Committed(_)
+ ));
+ assert_eq!(sync.publish(created).await.unwrap(), revision);
+
+ // Directory creation and a stable-identity rename.
+ let baseline = authority.observe_latest().await.unwrap();
+ let directory = sync
+ .stage_create_directory(root, "dir", baseline.revision())
+ .await
+ .unwrap();
+ let revision = sync.publish(directory).await.unwrap();
+ let directory_node = match sync
+ .namespace_status(directory)
+ .await
+ .unwrap()
+ .intent
+ .operation
+ {
+ NamespaceOperation::CreateDirectory { node, .. } => node,
+ operation => panic!("expected directory creation, got {operation:?}"),
+ };
+ assert!(
+ authority
+ .observe_revision(revision)
+ .await
+ .unwrap()
+ .node(directory_node)
+ .await
+ .unwrap()
+ .unwrap()
+ .is_directory()
+ );
+ let baseline = authority.observe_latest().await.unwrap();
+ let rename = sync
+ .stage_rename(
+ created_node,
+ baseline.revision(),
+ directory_node,
+ "moved",
+ RenameMode::NoReplace,
+ )
+ .await
+ .unwrap();
+ let revision = sync.publish(rename).await.unwrap();
+ let observed = authority.observe_revision(revision).await.unwrap();
+ assert_eq!(
+ observed.resolve("dir/moved").await.unwrap().unwrap().id(),
+ created_node
+ );
+ assert!(observed.resolve("created").await.unwrap().is_none());
+
+ // Removing a live file cannot make its pinned baseline disappear.
+ let baseline = authority.observe_latest().await.unwrap();
+ let remove = sync
+ .stage_remove(created_node, baseline.revision())
+ .await
+ .unwrap();
+ let revision = sync.publish(remove).await.unwrap();
+ assert!(
+ authority
+ .observe_revision(revision)
+ .await
+ .unwrap()
+ .resolve("dir/moved")
+ .await
+ .unwrap()
+ .is_none()
+ );
+ assert_eq!(
+ baseline
+ .open_file(created_node)
+ .await
+ .unwrap()
+ .read(0, 64)
+ .await
+ .unwrap(),
+ b"created bytes"
+ );
+
+ // No-replace rejects an occupied destination while replace removes exactly
+ // the observed destination and keeps its old version readable.
+ let (_, source) = create_file(&authority, &runtime, root, "source").await;
+ let (baseline, target) = create_file(&authority, &runtime, root, "target").await;
+ let baseline = authority.observe_revision(baseline).await.unwrap();
+ assert_eq!(
+ sync.stage_rename(
+ source,
+ baseline.revision(),
+ root,
+ "target",
+ RenameMode::NoReplace,
+ )
+ .await
+ .unwrap_err()
+ .kind(),
+ ErrorKind::AlreadyExists
+ );
+ let replace = sync
+ .stage_rename(
+ source,
+ baseline.revision(),
+ root,
+ "target",
+ RenameMode::Replace,
+ )
+ .await
+ .unwrap();
+ assert!(matches!(
+ sync.namespace_status(replace).await.unwrap().intent.operation,
+ NamespaceOperation::Rename {
+ replaced: Some(id),
+ ..
+ } if id == target
+ ));
+ let revision = sync.publish(replace).await.unwrap();
+ let observed = authority.observe_revision(revision).await.unwrap();
+ assert_eq!(
+ observed.resolve("target").await.unwrap().unwrap().id(),
+ source
+ );
+ baseline.open_file(target).await.unwrap();
+
+ // Executable changes use a baseline-bound metadata transaction.
+ let (baseline, executable) = create_file(&authority, &runtime, root, "executable").await;
+ let set_executable = sync
+ .stage_set_executable(executable, baseline, true)
+ .await
+ .unwrap();
+ let revision = sync.publish(set_executable).await.unwrap();
+ assert!(
+ authority
+ .observe_revision(revision)
+ .await
+ .unwrap()
+ .node(executable)
+ .await
+ .unwrap()
+ .unwrap()
+ .executable()
+ );
+
+ // Create/create: the second intent keeps its conflict.
+ let baseline = authority.observe_latest().await.unwrap();
+ let first = sync
+ .stage_create_file(
+ root,
+ "same",
+ baseline.revision(),
+ FileMetadata::new(false),
+ &mut &b"first"[..],
+ )
+ .await
+ .unwrap();
+ let second = sync
+ .stage_create_file(
+ root,
+ "same",
+ baseline.revision(),
+ FileMetadata::new(false),
+ &mut &b"second"[..],
+ )
+ .await
+ .unwrap();
+ sync.publish(first).await.unwrap();
+ assert_conflict(sync.publish(second).await);
+ assert_eq!(
+ sync.namespace_status(second).await.unwrap().state,
+ NamespaceState::Conflict
+ );
+ assert_conflict(sync.publish(second).await);
+
+ // Content/edit -> rename conflict. The rename guards the source's complete
+ // logical state rather than only its old parent/name topology.
+ let (baseline, renamed) = create_file(&authority, &runtime, root, "renamed").await;
+ let content = sync
+ .stage(renamed, baseline, &mut &b"new content"[..])
+ .await
+ .unwrap();
+ let rename = sync
+ .stage_rename(
+ renamed,
+ baseline,
+ root,
+ "renamed-new",
+ RenameMode::NoReplace,
+ )
+ .await
+ .unwrap();
+ sync.publish(content).await.unwrap();
+ assert_conflict(sync.publish(rename).await);
+ assert_eq!(
+ sync.namespace_status(rename).await.unwrap().state,
+ NamespaceState::Conflict
+ );
+
+ // 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;
+ let content = sync
+ .stage(deleted, baseline, &mut &b"edit before remove"[..])
+ .await
+ .unwrap();
+ let remove = sync.stage_remove(deleted, baseline).await.unwrap();
+ sync.publish(content).await.unwrap();
+ assert_conflict(sync.publish(remove).await);
+
+ let (baseline, deleted) = create_file(&authority, &runtime, root, "removed-first").await;
+ let content = sync
+ .stage(deleted, baseline, &mut &b"edit after remove"[..])
+ .await
+ .unwrap();
+ let remove = sync.stage_remove(deleted, baseline).await.unwrap();
+ sync.publish(remove).await.unwrap();
+ assert_conflict(sync.publish(content).await);
+ assert_eq!(
+ sync.read(content, 0, 64).await.unwrap(),
+ b"edit after remove"
+ );
+
+ // Rename/rename: only one move of the common baseline may publish.
+ let (baseline, moved) = create_file(&authority, &runtime, root, "moved-twice").await;
+ let left = sync
+ .stage_rename(moved, baseline, root, "left", RenameMode::NoReplace)
+ .await
+ .unwrap();
+ let right = sync
+ .stage_rename(moved, baseline, root, "right", RenameMode::NoReplace)
+ .await
+ .unwrap();
+ sync.publish(left).await.unwrap();
+ assert_conflict(sync.publish(right).await);
+
+ // Parent removal versus child creation, in both publication orders.
+ let (baseline, parent) = create_directory(&authority, &runtime, root, "parent-a").await;
+ let remove = sync.stage_remove(parent, baseline).await.unwrap();
+ let child = sync
+ .stage_create_file(
+ parent,
+ "child",
+ baseline,
+ FileMetadata::new(false),
+ &mut &b""[..],
+ )
+ .await
+ .unwrap();
+ sync.publish(child).await.unwrap();
+ assert_conflict(sync.publish(remove).await);
+
+ let (baseline, parent) = create_directory(&authority, &runtime, root, "parent-b").await;
+ let remove = sync.stage_remove(parent, baseline).await.unwrap();
+ let child = sync
+ .stage_create_file(
+ parent,
+ "child",
+ baseline,
+ FileMetadata::new(false),
+ &mut &b""[..],
+ )
+ .await
+ .unwrap();
+ sync.publish(remove).await.unwrap();
+ assert_conflict(sync.publish(child).await);
+
+ // A content edit conflicts with an executable change against the same
+ // baseline; the executable intent remains inspectable.
+ let (baseline, metadata) = create_file(&authority, &runtime, root, "metadata-race").await;
+ let content = sync
+ .stage(metadata, baseline, &mut &b"edited"[..])
+ .await
+ .unwrap();
+ let mode = sync
+ .stage_set_executable(metadata, baseline, true)
+ .await
+ .unwrap();
+ sync.publish(content).await.unwrap();
+ assert_conflict(sync.publish(mode).await);
+ assert_eq!(
+ sync.namespace_status(mode).await.unwrap().state,
+ NamespaceState::Conflict
+ );
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn namespace_intents_are_durable_and_conflict_safe_in_both_modes() {
+ let backend = TestBackend::default();
+ exercise_namespace(Arc::new(
+ Fs::create(backend.operator(), BackendProfile::Minio)
+ .await
+ .unwrap(),
+ ))
+ .await;
+
+ let backend = TestBackend::default();
+ let temp = tempfile::tempdir().unwrap();
+ let service = MetadataService::create(temp.path().join("metadata.db"), backend.operator())
+ .await
+ .unwrap();
+ let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
+ let address = listener.local_addr().unwrap();
+ let token = "sync-test-only-service-token-at-least-32-bytes";
+ let server = tokio::spawn(serve(listener, service, token.into()));
+ let client = ServiceClient::connect(address, token.into(), backend.operator())
+ .await
+ .unwrap();
+ exercise_namespace(Arc::new(client)).await;
+ server.abort();
+}
+
+#[tokio::test]
+async fn namespace_unknown_retries_the_original_request_after_restart() {
+ let backend = TestBackend::default();
+ let authority: Arc<dyn Authority> = Arc::new(
+ Fs::create(backend.operator(), BackendProfile::Minio)
+ .await
+ .unwrap(),
+ );
+ let temp = tempfile::tempdir().unwrap();
+ let sync = Sync::open(authority.clone(), temp.path().join("sync"))
+ .await
+ .unwrap();
+ let baseline = authority.observe_latest().await.unwrap();
+ let intent = sync
+ .stage_create_directory(authority.root(), "uncertain", baseline.revision())
+ .await
+ .unwrap();
+ let node = sync
+ .namespace_status(intent)
+ .await
+ .unwrap()
+ .intent
+ .operation
+ .node_id();
+ backend.state.lock().unwrap().delay_next_head = true;
+ let error = sync.publish(intent).await.unwrap_err();
+ assert_eq!(error.kind(), ErrorKind::Unknown);
+ let status = sync.namespace_status(intent).await.unwrap();
+ assert_eq!(status.state, NamespaceState::Unknown);
+ assert_eq!(status.failure.unwrap().commit_id(), error.commit_id());
+ drop(sync);
+
+ let sync = Sync::open(authority.clone(), temp.path().join("sync"))
+ .await
+ .unwrap();
+ assert_eq!(
+ sync.namespace_status(intent).await.unwrap().state,
+ NamespaceState::Unknown
+ );
+ assert!(sync.resolve_namespace(intent).await.unwrap().is_none());
+ let revision = sync.publish(intent).await.unwrap();
+ let receipt = sync.resolve_namespace(intent).await.unwrap().unwrap();
+ assert_eq!(receipt.cursor.revision, revision);
+ assert_eq!(
+ authority
+ .observe_revision(revision)
+ .await
+ .unwrap()
+ .node(node)
+ .await
+ .unwrap()
+ .unwrap()
+ .id(),
+ node
+ );
+ assert!(matches!(
+ sync.namespace_status(intent).await.unwrap().state,
+ NamespaceState::Committed(_)
+ ));
+}