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/09/22 08:41:07 UTC
(opendal-yinyang) 01/01: feat: separate reference release and retain typed runtime failures
This is an automated email from the ASF dual-hosted git repository.
Xuanwo pushed a commit to branch xuanwo/core-completion-errors
in repository https://gitbox.apache.org/repos/asf/opendal-yinyang.git
commit 3cc795d2769e8d5d3fdc86427141ea40180c48ac
Author: Xuanwo <gi...@xuanwo.io>
AuthorDate: Tue Sep 22 16:40:57 2026 +0800
feat: separate reference release and retain typed runtime failures
Co-authored-by: Chojan Shang <ps...@apache.org>
---
crates/core/src/error.rs | 11 +-
crates/core/src/namespace.rs | 11 +-
crates/core/src/service/transport.rs | 37 +++++
crates/core/src/snapshot.rs | 18 +-
crates/core/src/transaction.rs | 40 ++++-
specs/file-runtime.md | 45 ++++-
specs/metadata-service.md | 7 +
specs/yinyang-format.md | 7 +
src/file_cli.rs | 2 +-
src/lib.rs | 4 +
src/runtime.rs | 149 ++++++++---------
src/runtime/error.rs | 178 ++++++++++++++++++++
src/runtime/staging.rs | 312 ++++++++++++++++++++++++++++++-----
src/volume.rs | 2 +-
tests/directory.rs | 2 +-
tests/runtime.rs | 117 ++++++++++++-
16 files changed, 798 insertions(+), 144 deletions(-)
diff --git a/crates/core/src/error.rs b/crates/core/src/error.rs
index ee04a0d..27a0a80 100644
--- a/crates/core/src/error.rs
+++ b/crates/core/src/error.rs
@@ -27,6 +27,11 @@ pub enum ErrorKind {
Unsupported,
Storage,
Io,
+ InvalidName,
+ NotDirectory,
+ IsDirectory,
+ NotEmpty,
+ PermissionDenied,
}
/// Error returned by the YinYang Format core.
@@ -71,7 +76,11 @@ impl Error {
}
pub(crate) fn from_storage(operation: &'static str, error: opendal::Error) -> Self {
- Self::new(ErrorKind::Storage, operation, error.to_string())
+ let kind = match error.kind() {
+ opendal::ErrorKind::PermissionDenied => ErrorKind::PermissionDenied,
+ _ => ErrorKind::Storage,
+ };
+ Self::new(kind, operation, error.to_string())
}
pub(crate) fn from_io(operation: &'static str, error: std::io::Error) -> Self {
diff --git a/crates/core/src/namespace.rs b/crates/core/src/namespace.rs
index 11365db..d81852e 100644
--- a/crates/core/src/namespace.rs
+++ b/crates/core/src/namespace.rs
@@ -16,7 +16,7 @@
// under the License.
use crate::data::ContentDescriptor;
-use crate::{Error, NodeId, Result};
+use crate::{Error, ErrorKind, NodeId, Result};
use unicode_casefold::UnicodeCaseFold as _;
use unicode_normalization::UnicodeNormalization as _;
@@ -167,7 +167,8 @@ pub fn name_key(name: &str) -> Result<String> {
|| matches!(c, '/' | '\\' | ':' | '*' | '?' | '"' | '<' | '>' | '|')
})
{
- return Err(Error::invalid(
+ return Err(Error::new(
+ ErrorKind::InvalidName,
"validate name",
"non-portable or non-NFC component",
));
@@ -184,7 +185,11 @@ pub fn name_key(name: &str) -> Result<String> {
})
})
{
- return Err(Error::invalid("validate name", "reserved device name"));
+ return Err(Error::new(
+ ErrorKind::InvalidName,
+ "validate name",
+ "reserved device name",
+ ));
}
Ok(folded)
}
diff --git a/crates/core/src/service/transport.rs b/crates/core/src/service/transport.rs
index deb8731..6d79270 100644
--- a/crates/core/src/service/transport.rs
+++ b/crates/core/src/service/transport.rs
@@ -66,6 +66,11 @@ fn kind_tag(kind: ErrorKind) -> u8 {
ErrorKind::Unsupported => 4,
ErrorKind::Storage => 5,
ErrorKind::Io => 6,
+ ErrorKind::InvalidName => 7,
+ ErrorKind::NotDirectory => 8,
+ ErrorKind::IsDirectory => 9,
+ ErrorKind::NotEmpty => 10,
+ ErrorKind::PermissionDenied => 11,
}
}
fn tag_kind(tag: u8) -> Result<ErrorKind> {
@@ -77,9 +82,41 @@ fn tag_kind(tag: u8) -> Result<ErrorKind> {
4 => ErrorKind::Unsupported,
5 => ErrorKind::Storage,
6 => ErrorKind::Io,
+ 7 => ErrorKind::InvalidName,
+ 8 => ErrorKind::NotDirectory,
+ 9 => ErrorKind::IsDirectory,
+ 10 => ErrorKind::NotEmpty,
+ 11 => ErrorKind::PermissionDenied,
_ => return Err(protocol()),
})
}
+
+#[cfg(test)]
+mod error_tests {
+ use super::*;
+ #[test]
+ fn domain_error_tags_round_trip_without_changing_existing_tags() {
+ let kinds = [
+ ErrorKind::Invalid,
+ ErrorKind::Corrupt,
+ ErrorKind::NotFound,
+ ErrorKind::AlreadyExists,
+ ErrorKind::Unsupported,
+ ErrorKind::Storage,
+ ErrorKind::Io,
+ ErrorKind::InvalidName,
+ ErrorKind::NotDirectory,
+ ErrorKind::IsDirectory,
+ ErrorKind::NotEmpty,
+ ErrorKind::PermissionDenied,
+ ];
+ for (tag, kind) in kinds.into_iter().enumerate() {
+ assert_eq!(kind_tag(kind), tag as u8);
+ assert_eq!(tag_kind(tag as u8).unwrap(), kind);
+ }
+ assert!(tag_kind(255).is_err());
+ }
+}
async fn read_frame<T: BorshDeserialize>(stream: &mut TcpStream) -> Result<T> {
let length = stream.read_u32().await.map_err(io)? as usize;
if length > MAX_FRAME {
diff --git a/crates/core/src/snapshot.rs b/crates/core/src/snapshot.rs
index fc8a864..b35f86b 100644
--- a/crates/core/src/snapshot.rs
+++ b/crates/core/src/snapshot.rs
@@ -257,7 +257,11 @@ impl Snapshot {
Error::new(ErrorKind::NotFound, "open file version", "node is absent")
})?;
if node.is_directory() {
- return Err(Error::invalid("open file version", "node is a directory"));
+ return Err(Error::new(
+ ErrorKind::IsDirectory,
+ "open file version",
+ "node is a directory",
+ ));
}
Ok(FileVersion {
data: self.data.clone(),
@@ -324,7 +328,11 @@ impl Snapshot {
.await?
.ok_or_else(|| Error::new(ErrorKind::NotFound, "read directory", "node is absent"))?;
if !node.is_directory() {
- return Err(Error::invalid("read directory", "node is not a directory"));
+ return Err(Error::new(
+ ErrorKind::NotDirectory,
+ "read directory",
+ "node is not a directory",
+ ));
}
Ok(node)
}
@@ -392,7 +400,11 @@ impl Snapshot {
.kind
{
NodeKind::File(file) => self.data.published(file),
- _ => Err(Error::invalid("read content", "node is a directory")),
+ _ => Err(Error::new(
+ ErrorKind::IsDirectory,
+ "read content",
+ "node is a directory",
+ )),
}
}
pub async fn receipt(&self, id: CommitId) -> Result<Option<Receipt>> {
diff --git a/crates/core/src/transaction.rs b/crates/core/src/transaction.rs
index 59c8f23..4f2a458 100644
--- a/crates/core/src/transaction.rs
+++ b/crates/core/src/transaction.rs
@@ -528,15 +528,23 @@ impl Working {
self.original(id).await
}
async fn required(&mut self, id: NodeId) -> Result<Node> {
- self.node(id)
- .await?
- .ok_or_else(|| Error::invalid("plan mutation", "node is absent"))
+ self.node(id).await?.ok_or_else(|| {
+ Error::new(
+ crate::ErrorKind::NotFound,
+ "plan mutation",
+ "node is absent",
+ )
+ })
}
async fn directory(&mut self, id: NodeId) -> Result<Node> {
self.guard(id, KIND).await?;
let node = self.required(id).await?;
if !node.is_directory() {
- return Err(Error::invalid("plan mutation", "parent is not a directory"));
+ return Err(Error::new(
+ crate::ErrorKind::NotDirectory,
+ "plan mutation",
+ "parent is not a directory",
+ ));
}
Ok(node)
}
@@ -632,7 +640,11 @@ impl Working {
self.guard_entry(&key).await?;
self.guard(*id, KIND).await?;
if self.entry(&key).await?.is_some() || self.node(*id).await?.is_some() {
- return Err(Error::invalid("create node", "name or identity exists"));
+ return Err(Error::new(
+ crate::ErrorKind::AlreadyExists,
+ "create node",
+ "name or identity exists",
+ ));
}
let kind = match file {
Some(file) => NodeKind::File(self.base.data.accept(file)?),
@@ -661,7 +673,11 @@ impl Working {
self.guard(*id, STATE).await?;
let mut node = self.required(*id).await?;
if node.is_directory() {
- return Err(Error::invalid("set content", "node is a directory"));
+ return Err(Error::new(
+ crate::ErrorKind::IsDirectory,
+ "set content",
+ "node is a directory",
+ ));
}
node.kind = NodeKind::File(self.base.data.accept(file)?);
self.nodes.insert(*id, Some(node));
@@ -689,7 +705,11 @@ impl Working {
.await?
.is_some_and(|v| v.node_id != *id)
{
- return Err(Error::invalid("rename", "destination exists"));
+ return Err(Error::new(
+ crate::ErrorKind::AlreadyExists,
+ "rename",
+ "destination exists",
+ ));
}
if node.is_directory() {
let mut ancestor = *parent;
@@ -732,7 +752,11 @@ impl Working {
if matches!(node.kind, NodeKind::Directory { .. }) {
self.guard(*id, MEMBERSHIP).await?;
if !self.list(*id).await?.is_empty() {
- return Err(Error::invalid("remove", "directory is not empty"));
+ return Err(Error::new(
+ crate::ErrorKind::NotEmpty,
+ "remove",
+ "directory is not empty",
+ ));
}
}
self.nodes.insert(*id, None);
diff --git a/specs/file-runtime.md b/specs/file-runtime.md
index f8866cc..87238de 100644
--- a/specs/file-runtime.md
+++ b/specs/file-runtime.md
@@ -75,6 +75,7 @@ publication. Callers needing frozen namespace replay use Authority/Planner.
| sync_local | Report local durable and remote published positions; does not publish. |
| flush / fsync / commit | Prepare bytes and conditionally publish their metadata; success means remote durability, not just local acceptance. |
| close | Publish dirty data, then release staging. Failure keeps the handle open/recoverable. |
+| release | Consume the caller's handle reference and return its recovery UUID. Do not publish, delete staging, acknowledge errors, or resolve a frozen request. |
| abort | Discard a never-dispatched or definitively conflicted handle. Refuse an unresolved frozen request. |
| drop | Release the in-process lease only. Neither publish nor discard staging. |
@@ -101,7 +102,29 @@ recovers by replaying the original request and reading its original receipt.
## Staging and errors
-The experimental local profile `yinyang-stage-1` stores filesystem binding,
+Local acceptance, remote completion and reference release are separate events.
+After release, a caller may immediately recover the UUID in the same runtime;
+after restart, the same ID recovers the durable state. Release initiates no
+publication or staging mutation and makes no remote completion claim. Even a clean released handle stays in
+staging until explicit close or abort. A failed close can be followed by release
+without losing bytes, the frozen request, or the error ledger.
+
+`Error::kind()` exposes domain categories, not platform status codes. Namespace
+failures distinguish NotFound, AlreadyExists, InvalidName, NotDirectory,
+IsDirectory and NotEmpty. Runtime failures distinguish ReadOnly, Busy, Closed,
+Frozen and TooLarge; publication distinguishes Conflict, Retryable and Unknown.
+Corrupt, Unsupported, Storage, Io, PermissionDenied and NoSpace preserve the
+available failure category without parsing messages. Invalid covers other
+contract violations; unknown backend failures remain Storage/Io rather than
+being guessed as conflicts or safe retries. `Error::commit_id()` retains the
+original Unknown publication identity.
+
+`HandleStatus::error` and `WritebackError::error` contain a typed `Failure`
+with kind, diagnostic message and optional commit ID. Clean fsync/close returns
+the retained category until acknowledgment. Clearing the volume ledger does not
+clear an active handle's error, pending data or frozen request.
+
+The experimental local profile `yinyang-stage-2` stores filesystem binding,
handle records, chunk rows, and an error ledger in `staging.db` with WAL and
`synchronous=FULL`. The process lock is `runtime.lock`. Keep WAL with the
database; a copied live main database alone is not a supported backup.
@@ -109,6 +132,22 @@ Incomplete opens were never exposed and are discarded on reopen. Completed
handles and accepted writes survive process restart. There is no stable
migration promise for this local experimental format.
+Opening stage-1 performs a single SQLite transaction that adds nullable
+`failure BLOB` columns to handles and errors and advances the profile to
+stage-2. Original handle records, chunks, frozen requests and ledger sequences
+are preserved. Inspect supports either profile without upgrading it. Old
+message-only failures report Unclassified; their kind and commit identity are
+not inferred from prose. Older binaries reject stage-2; downgrade is unsupported.
+The upgrade changes only local staging, not remote metadata or content.
+
+Failure blobs use Borsh `(kind: u8, message: String, commit: Option<[u8;16]>)`.
+Kind tags, in order from 0 through 21, are Invalid, InvalidName, NotFound,
+AlreadyExists, NotDirectory, IsDirectory, NotEmpty, ReadOnly, PermissionDenied,
+Busy, Closed, Frozen, TooLarge, Conflict, Retryable, Unknown, Corrupt, Unsupported,
+Storage, Io, NoSpace and Unclassified. New handle records keep the original
+message field as a diagnostic copy; typed details live in the adjacent column.
+An operation failure records both fields and its ledger entry atomically.
+
Failures are returned immediately and retained on the handle and volume ledger.
A successful retry clears the active handle error, not the ledger. A clean
handle with a retained error returns it on fsync/close until
@@ -128,6 +167,10 @@ The file-operation tests exercise public Runtime/FileHandle methods against
object and authenticated service authorities. They cover sparse gaps, append,
cross-chunk truncation/extension, pinned visibility, rename/unlink, conflicting
writers, failed uploads/close, frozen unknown requests, and reopening staging.
+They also cover explicit reference release without publication, typed errors
+through both authorities, retained Unknown identity, and stage-1 upgrades with
+pending data and frozen requests. Range-reader tests verify bounded payload I/O
+and corruption rejection without staging.
Production power-loss testing, capacity benchmarks, incremental upload
optimization, OS-specific mount behavior, and Sync reconciliation remain
separate work.
diff --git a/specs/metadata-service.md b/specs/metadata-service.md
index 26aaa92..0ee704b 100644
--- a/specs/metadata-service.md
+++ b/specs/metadata-service.md
@@ -81,6 +81,13 @@ typed errors. Maximum batch size is 4096; scans request at most 4097 records.
The server handles at most 32 active connections, each with a 60-second timeout.
This wire profile is experimental and is not a stable interoperability promise.
+Error responses retain the existing tag order: Invalid=0, Corrupt=1,
+NotFound=2, AlreadyExists=3, Unsupported=4, Storage=5 and Io=6. Additional
+domain tags are InvalidName=7, NotDirectory=8, IsDirectory=9, NotEmpty=10 and
+PermissionDenied=11. Diagnostics remain strings. Clients reject unknown tags;
+deploy matching client/server versions for these new categories. This extends
+error reporting, not the persisted metadata or transaction encoding.
+
## Failure and recovery
Database contention before commit returns `Retryable`; replay the original
diff --git a/specs/yinyang-format.md b/specs/yinyang-format.md
index cb3301b..6231766 100644
--- a/specs/yinyang-format.md
+++ b/specs/yinyang-format.md
@@ -77,6 +77,13 @@ not turn ambiguity into failure. Callers retain the request and commit identity.
Invalid requests and malformed persistent state return typed errors, not
semantic conflicts.
+Namespace errors distinguish an absent required node (NotFound), an occupied
+destination (AlreadyExists), a non-directory parent (NotDirectory), a directory
+used as a file (IsDirectory), a nonempty removal (NotEmpty), and an invalid
+portable component (InvalidName). Other structural violations remain Invalid.
+Missing optional lookup results remain None. These categories are shared with
+the service authority and do not change persisted node or transaction bytes.
+
## Names and ordering
Components are NFC, 1–255 UTF-8 bytes, with no full-path length invariant.
diff --git a/src/file_cli.rs b/src/file_cli.rs
index 7e3ef12..faa7cf3 100644
--- a/src/file_cli.rs
+++ b/src/file_cli.rs
@@ -107,7 +107,7 @@ fn status(value: RuntimeStatus) {
for e in value.errors {
println!(
"error {} handle {} persisted {}: {}",
- e.sequence, e.handle, e.persisted, e.message
+ e.sequence, e.handle, e.persisted, e.error
);
}
}
diff --git a/src/lib.rs b/src/lib.rs
index c53318a..5d93fe4 100644
--- a/src/lib.rs
+++ b/src/lib.rs
@@ -25,3 +25,7 @@ pub mod volume;
pub use directory::{directory_nodes, prepare_directory, publish_directory, restore_directory};
pub use yinyang_core as core;
+
+#[cfg(test)]
+#[path = "../crates/core/tests/support/mod.rs"]
+mod support;
diff --git a/src/runtime.rs b/src/runtime.rs
index 8da0c73..3a5f96f 100644
--- a/src/runtime.rs
+++ b/src/runtime.rs
@@ -28,43 +28,8 @@ use yinyang_core::{
mod staging;
use staging::{CHUNK, Record, Stage};
-#[derive(Debug)]
-pub enum Error {
- Core(yinyang_core::Error),
- Local(String),
- Invalid(&'static str),
- Conflict,
- Retryable,
- Unknown(CommitId),
-}
-impl std::fmt::Display for Error {
- fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
- match self {
- Self::Core(e) => write!(f, "{e}"),
- Self::Local(e) => write!(f, "local staging: {e}"),
- Self::Invalid(e) => write!(f, "{e}"),
- Self::Conflict => write!(
- f,
- "opened file state changed or was unlinked; staged bytes are retained"
- ),
- Self::Retryable => write!(
- f,
- "publication is retryable; the original request is retained"
- ),
- Self::Unknown(id) => write!(
- f,
- "publication outcome is unknown for {id:?}; the original request is retained"
- ),
- }
- }
-}
-impl std::error::Error for Error {}
-impl From<yinyang_core::Error> for Error {
- fn from(e: yinyang_core::Error) -> Self {
- Self::Core(e)
- }
-}
-pub type Result<T> = std::result::Result<T, Error>;
+mod error;
+pub use error::{Error, ErrorKind, Failure, Result};
/// Local and remote positions are distinct; a local write is not a remote receipt.
#[derive(Clone, Debug)]
@@ -78,13 +43,13 @@ pub struct HandleStatus {
pub pending: bool,
pub frozen: bool,
pub conflict: bool,
- pub error: Option<String>,
+ pub error: Option<Failure>,
}
#[derive(Clone, Debug)]
pub struct WritebackError {
pub sequence: u64,
pub handle: uuid::Uuid,
- pub message: String,
+ pub error: Failure,
pub persisted: bool,
}
#[derive(Clone, Debug)]
@@ -97,7 +62,7 @@ struct Inner {
authority: Arc<dyn Authority>,
stage: Stage,
active: Mutex<BTreeSet<[u8; 16]>>,
- volatile_errors: Mutex<BTreeMap<[u8; 16], String>>,
+ volatile_errors: Mutex<BTreeMap<[u8; 16], Failure>>,
}
/// One staging directory per active runtime. Different clients may have separate
/// staging directories against the same authority; generation checks arbitrate.
@@ -136,7 +101,7 @@ impl Runtime {
}
fn require_write(&self) -> Result<()> {
if self.inner.read_only {
- Err(Error::Invalid("volume is read-only"))
+ Err(Error::State(ErrorKind::ReadOnly, "volume is read-only"))
} else {
Ok(())
}
@@ -152,7 +117,7 @@ impl Runtime {
status.errors.push(WritebackError {
sequence: 0,
handle: uuid::Uuid::from_bytes(*id),
- message: message.clone(),
+ error: message.clone(),
persisted: false,
});
if let Some(handle) = status.handles.iter_mut().find(|h| h.id.as_bytes() == id) {
@@ -184,14 +149,8 @@ impl Runtime {
self.claim(r.id)
}
fn claim(&self, id: [u8; 16]) -> Result<FileHandle> {
- if !self
- .inner
- .active
- .lock()
- .map_err(|e| Error::Local(e.to_string()))?
- .insert(id)
- {
- return Err(Error::Invalid("handle already active"));
+ if !self.inner.active.lock().map_err(Error::local)?.insert(id) {
+ return Err(Error::State(ErrorKind::Busy, "handle already active"));
}
Ok(FileHandle {
runtime: self.clone(),
@@ -209,7 +168,7 @@ impl Runtime {
let node = snapshot
.resolve(path)
.await?
- .ok_or(Error::Invalid("file does not exist"))?;
+ .ok_or(Error::State(ErrorKind::NotFound, "file does not exist"))?;
self.open_snapshot_node(snapshot, node.id(), writable).await
}
/// Opens a stable node identity at latest, independently of its current name.
@@ -243,7 +202,10 @@ impl Runtime {
) -> Result<FileHandle> {
let file = snapshot.open_file(node).await?;
if file.descriptor().content_id().length() > i64::MAX as u64 {
- return Err(Error::Invalid("file exceeds staging length limit"));
+ return Err(Error::State(
+ ErrorKind::TooLarge,
+ "file exceeds staging length limit",
+ ));
}
let mut r = Record {
id: *uuid::Uuid::new_v4().as_bytes(),
@@ -257,6 +219,7 @@ impl Runtime {
plan: None,
conflict: false,
error: None,
+ failure: None,
};
self.inner.stage.save(r.clone()).await?;
let mut offset = 0;
@@ -336,7 +299,7 @@ impl FileHandle {
}
fn live(&self) -> Result<()> {
if self.closed {
- Err(Error::Invalid("handle is closed"))
+ Err(Error::State(ErrorKind::Closed, "handle is closed"))
} else {
Ok(())
}
@@ -353,37 +316,45 @@ impl FileHandle {
}
pub async fn read(&self, offset: u64, length: usize) -> Result<Vec<u8>> {
self.live()?;
- self.runtime.inner.stage.read(self.id, offset, length).await
+ let result = self.runtime.inner.stage.read(self.id, offset, length).await;
+ self.report(result).await
}
pub async fn write(&mut self, offset: u64, bytes: &[u8]) -> Result<()> {
self.live()?;
- self.runtime.require_write()?;
- let result = self
- .runtime
- .inner
- .stage
- .write(self.id, Some(offset), bytes.to_vec())
- .await
- .map(|_| ());
+ let result = async {
+ self.runtime.require_write()?;
+ self.runtime
+ .inner
+ .stage
+ .write(self.id, Some(offset), bytes.to_vec())
+ .await
+ .map(|_| ())
+ }
+ .await;
self.report(result).await
}
/// Appends to this handle's private staged version; concurrent handles
/// conflict at publication rather than silently merging their appends.
pub async fn append(&mut self, bytes: &[u8]) -> Result<u64> {
self.live()?;
- self.runtime.require_write()?;
- let result = self
- .runtime
- .inner
- .stage
- .write(self.id, None, bytes.to_vec())
- .await;
+ let result = async {
+ self.runtime.require_write()?;
+ self.runtime
+ .inner
+ .stage
+ .write(self.id, None, bytes.to_vec())
+ .await
+ }
+ .await;
self.report(result).await
}
pub async fn truncate(&mut self, length: u64) -> Result<()> {
self.live()?;
- self.runtime.require_write()?;
- let result = self.runtime.inner.stage.truncate(self.id, length).await;
+ let result = async {
+ self.runtime.require_write()?;
+ self.runtime.inner.stage.truncate(self.id, length).await
+ }
+ .await;
self.report(result).await
}
async fn report<T>(&self, result: Result<T>) -> Result<T> {
@@ -393,12 +364,12 @@ impl FileHandle {
.runtime
.inner
.stage
- .failure(self.id, error.to_string())
+ .failure(self.id, Failure::capture(error))
.await
.is_err()
&& let Ok(mut errors) = self.runtime.inner.volatile_errors.lock()
{
- errors.insert(self.id, error.to_string());
+ errors.insert(self.id, Failure::capture(error));
}
}
result
@@ -414,6 +385,7 @@ impl FileHandle {
self.live()?;
let mut record = self.runtime.inner.stage.load(self.id).await?;
record.error = None;
+ record.failure = None;
self.runtime.inner.stage.save(record).await?;
if let Ok(mut errors) = self.runtime.inner.volatile_errors.lock() {
errors.remove(&self.id);
@@ -436,7 +408,7 @@ impl FileHandle {
let mut r = stage.load(self.id).await?;
if r.local == r.remote {
if let Some(error) = self.status().await?.error {
- return Err(Error::Local(error));
+ return Err(Error::Retained(error));
}
return Ok(None);
}
@@ -457,16 +429,14 @@ impl FileHandle {
if bytes.is_empty() {
return Err(Error::Invalid("staged data ended early"));
}
- sink.write_all(&bytes)
- .await
- .map_err(|e| Error::Local(e.to_string()))?;
+ sink.write_all(&bytes).await.map_err(Error::local)?;
offset += bytes.len() as u64;
}
Ok::<_, Error>(())
});
let prepared = self.runtime.authority().prepare(&mut source).await;
drop(source);
- let produced = producer.await.map_err(|e| Error::Local(e.to_string()))?;
+ let produced = producer.await.map_err(Error::local)?;
let prepared = prepared?;
produced?;
let snapshot = self
@@ -489,6 +459,7 @@ impl FileHandle {
r.remote = r.local;
r.plan = None;
r.error = None;
+ r.failure = None;
r.conflict = false;
stage.save(r).await?;
if let Ok(mut errors) = self.runtime.inner.volatile_errors.lock() {
@@ -511,18 +482,30 @@ impl FileHandle {
self.live()?;
let r = self.runtime.inner.stage.load(self.id).await?;
if r.plan.is_some() && !r.conflict {
- return Err(Error::Invalid("resolve the frozen request before aborting"));
+ return Err(Error::State(
+ ErrorKind::Frozen,
+ "resolve the frozen request before aborting",
+ ));
}
- self.runtime.inner.stage.remove(self.id).await?;
+ let result = self.runtime.inner.stage.remove(self.id).await;
+ self.report(result).await?;
self.closed = true;
Ok(())
}
- /// Failed close leaves the handle open and recoverable. Drop never swallows
- /// the failure by retrying or silently discarding dirty state.
+ /// Release this caller's lease without publishing or deleting staging.
+ /// Returns the identity for recovery, including pending, frozen or failed
+ /// handles. Errors and data remain inspectable after the reference is gone.
+ pub fn release(self) -> Result<uuid::Uuid> {
+ self.live()?;
+ Ok(self.id())
+ }
+ /// Publish before deleting staging. Failure leaves the handle open and
+ /// recoverable; release or drop never retries or discards the failed data.
pub async fn close(&mut self) -> Result<()> {
self.live()?;
self.fsync().await?;
- self.runtime.inner.stage.remove(self.id).await?;
+ let result = self.runtime.inner.stage.remove(self.id).await;
+ self.report(result).await?;
self.closed = true;
Ok(())
}
diff --git a/src/runtime/error.rs b/src/runtime/error.rs
new file mode 100644
index 0000000..137863f
--- /dev/null
+++ b/src/runtime/error.rs
@@ -0,0 +1,178 @@
+// 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.
+
+//! Domain failures shared by runtime calls and durable error reporting.
+use borsh::{BorshDeserialize, BorshSerialize};
+use yinyang_core::CommitId;
+
+/// Classify errors without parsing diagnostic text. This is not an errno or
+/// platform status: adapters still choose their operation-specific mapping.
+#[derive(Clone, Copy, Debug, Eq, PartialEq, BorshSerialize, BorshDeserialize)]
+pub enum ErrorKind {
+ Invalid,
+ InvalidName,
+ NotFound,
+ AlreadyExists,
+ NotDirectory,
+ IsDirectory,
+ NotEmpty,
+ ReadOnly,
+ PermissionDenied,
+ Busy,
+ Closed,
+ Frozen,
+ TooLarge,
+ Conflict,
+ Retryable,
+ Unknown,
+ Corrupt,
+ Unsupported,
+ Storage,
+ Io,
+ NoSpace,
+ /// An older staging record retained only diagnostic text.
+ Unclassified,
+}
+
+/// A durable failure, including the original uncertain publication identity.
+#[derive(Clone, Debug, Eq, PartialEq, BorshSerialize, BorshDeserialize)]
+pub struct Failure {
+ kind: ErrorKind,
+ message: String,
+ commit: Option<[u8; 16]>,
+}
+impl Failure {
+ pub const fn kind(&self) -> ErrorKind {
+ self.kind
+ }
+ pub fn message(&self) -> &str {
+ &self.message
+ }
+ pub fn commit_id(&self) -> Option<CommitId> {
+ self.commit.map(CommitId::from_bytes)
+ }
+ pub(super) fn legacy(message: String) -> Self {
+ Self {
+ kind: ErrorKind::Unclassified,
+ message,
+ commit: None,
+ }
+ }
+ pub(super) fn capture(error: &Error) -> Self {
+ match error {
+ Error::Retained(failure) | Error::Local(failure) => failure.clone(),
+ _ => Self {
+ kind: error.kind(),
+ message: error.to_string(),
+ commit: error.commit_id().map(|id| *id.as_bytes()),
+ },
+ }
+ }
+}
+impl std::fmt::Display for Failure {
+ fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+ f.write_str(&self.message)
+ }
+}
+
+#[derive(Debug)]
+pub enum Error {
+ Core(yinyang_core::Error),
+ Local(Failure),
+ Invalid(&'static str),
+ State(ErrorKind, &'static str),
+ Retained(Failure),
+ Conflict,
+ Retryable,
+ Unknown(CommitId),
+}
+impl Error {
+ pub fn kind(&self) -> ErrorKind {
+ match self {
+ Self::Core(e) => match e.kind() {
+ yinyang_core::ErrorKind::Invalid => ErrorKind::Invalid,
+ yinyang_core::ErrorKind::InvalidName => ErrorKind::InvalidName,
+ yinyang_core::ErrorKind::NotFound => ErrorKind::NotFound,
+ yinyang_core::ErrorKind::AlreadyExists => ErrorKind::AlreadyExists,
+ yinyang_core::ErrorKind::NotDirectory => ErrorKind::NotDirectory,
+ yinyang_core::ErrorKind::IsDirectory => ErrorKind::IsDirectory,
+ yinyang_core::ErrorKind::NotEmpty => ErrorKind::NotEmpty,
+ yinyang_core::ErrorKind::PermissionDenied => ErrorKind::PermissionDenied,
+ yinyang_core::ErrorKind::Corrupt => ErrorKind::Corrupt,
+ yinyang_core::ErrorKind::Unsupported => ErrorKind::Unsupported,
+ yinyang_core::ErrorKind::Storage => ErrorKind::Storage,
+ yinyang_core::ErrorKind::Io => ErrorKind::Io,
+ },
+ Self::Local(e) | Self::Retained(e) => e.kind(),
+ Self::Invalid(_) => ErrorKind::Invalid,
+ Self::State(kind, _) => *kind,
+ Self::Conflict => ErrorKind::Conflict,
+ Self::Retryable => ErrorKind::Retryable,
+ Self::Unknown(_) => ErrorKind::Unknown,
+ }
+ }
+ pub fn commit_id(&self) -> Option<CommitId> {
+ match self {
+ Self::Unknown(id) => Some(*id),
+ Self::Retained(e) | Self::Local(e) => e.commit_id(),
+ _ => None,
+ }
+ }
+ pub(crate) fn local(message: impl std::fmt::Display) -> Self {
+ Self::local_kind(ErrorKind::Io, message)
+ }
+ pub(crate) fn local_kind(kind: ErrorKind, message: impl std::fmt::Display) -> Self {
+ Self::Local(Failure {
+ kind,
+ message: message.to_string(),
+ commit: None,
+ })
+ }
+}
+impl std::fmt::Display for Error {
+ fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+ match self {
+ Self::Core(e) => write!(f, "{e}"),
+ Self::Local(e) => write!(f, "local staging: {e}"),
+ Self::Invalid(e) | Self::State(_, e) => write!(f, "{e}"),
+ Self::Retained(e) => write!(f, "{e}"),
+ Self::Conflict => write!(f, "observed state changed; any staged bytes are retained"),
+ Self::Retryable => write!(
+ f,
+ "publication is retryable; the original request is retained"
+ ),
+ Self::Unknown(id) => write!(
+ f,
+ "publication outcome is unknown for {id:?}; the original request is retained"
+ ),
+ }
+ }
+}
+impl std::error::Error for Error {
+ fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
+ match self {
+ Self::Core(e) => Some(e),
+ _ => None,
+ }
+ }
+}
+impl From<yinyang_core::Error> for Error {
+ fn from(e: yinyang_core::Error) -> Self {
+ Self::Core(e)
+ }
+}
+pub type Result<T> = std::result::Result<T, Error>;
diff --git a/src/runtime/staging.rs b/src/runtime/staging.rs
index bff3e85..8db7b83 100644
--- a/src/runtime/staging.rs
+++ b/src/runtime/staging.rs
@@ -37,6 +37,9 @@ pub(super) struct Record {
pub plan: Option<Vec<u8>>,
pub conflict: bool,
pub error: Option<String>,
+ // Stored in an additive SQLite column; keep existing staged record bytes readable.
+ #[borsh(skip)]
+ pub failure: Option<Failure>,
}
impl Record {
pub fn status(&self) -> HandleStatus {
@@ -50,7 +53,10 @@ impl Record {
pending: self.local != self.remote,
frozen: self.plan.is_some(),
conflict: self.conflict,
- error: self.error.clone(),
+ error: self
+ .failure
+ .clone()
+ .or_else(|| self.error.clone().map(Failure::legacy)),
}
}
}
@@ -59,29 +65,80 @@ pub(super) struct Stage {
db: Arc<Mutex<Connection>>,
_lock: Arc<File>,
}
-fn local(e: impl std::fmt::Display) -> Error {
- Error::Local(e.to_string())
+fn local(e: impl std::error::Error + 'static) -> Error {
+ let cause: &dyn std::error::Error = &e;
+ let kind = if let Some(error) = cause.downcast_ref::<rusqlite::Error>() {
+ match error {
+ rusqlite::Error::QueryReturnedNoRows => ErrorKind::NotFound,
+ rusqlite::Error::SqliteFailure(code, _) => match code.code {
+ rusqlite::ErrorCode::DiskFull => ErrorKind::NoSpace,
+ rusqlite::ErrorCode::ReadOnly => ErrorKind::ReadOnly,
+ rusqlite::ErrorCode::PermissionDenied => ErrorKind::PermissionDenied,
+ rusqlite::ErrorCode::DatabaseCorrupt | rusqlite::ErrorCode::NotADatabase => {
+ ErrorKind::Corrupt
+ }
+ rusqlite::ErrorCode::DatabaseBusy | rusqlite::ErrorCode::DatabaseLocked => {
+ ErrorKind::Busy
+ }
+ _ => ErrorKind::Io,
+ },
+ _ => ErrorKind::Io,
+ }
+ } else if let Some(error) = cause.downcast_ref::<std::io::Error>() {
+ match error.kind() {
+ std::io::ErrorKind::StorageFull => ErrorKind::NoSpace,
+ std::io::ErrorKind::PermissionDenied => ErrorKind::PermissionDenied,
+ std::io::ErrorKind::ReadOnlyFilesystem => ErrorKind::ReadOnly,
+ _ => ErrorKind::Io,
+ }
+ } else {
+ ErrorKind::Io
+ };
+ Error::local_kind(kind, e)
}
fn decode(bytes: &[u8]) -> Result<Record> {
- borsh::from_slice(bytes).map_err(local)
+ borsh::from_slice(bytes).map_err(|e| Error::local_kind(ErrorKind::Corrupt, e))
}
fn save(c: &Connection, r: &Record) -> Result<()> {
c.execute(
- "INSERT OR REPLACE INTO handles VALUES(?1,?2)",
- params![r.id.as_slice(), borsh::to_vec(r).map_err(local)?],
+ "INSERT OR REPLACE INTO handles(id,record,failure) VALUES(?1,?2,?3)",
+ params![
+ r.id.as_slice(),
+ borsh::to_vec(r).map_err(local)?,
+ r.failure
+ .as_ref()
+ .map(borsh::to_vec)
+ .transpose()
+ .map_err(local)?
+ ],
)
.map_err(local)?;
Ok(())
}
+fn has_failure(c: &Connection, table: &str) -> Result<bool> {
+ c.query_row(
+ "SELECT EXISTS(SELECT 1 FROM pragma_table_info(?1) WHERE name='failure')",
+ [table],
+ |r| r.get(0),
+ )
+ .map_err(local)
+}
+fn decoded_record(bytes: &[u8], failure: Option<Vec<u8>>) -> Result<Record> {
+ let mut r = decode(bytes)?;
+ r.failure = failure
+ .map(|b| borsh::from_slice(&b).map_err(|e| Error::local_kind(ErrorKind::Corrupt, e)))
+ .transpose()?;
+ Ok(r)
+}
fn record(c: &Connection, id: [u8; 16]) -> Result<Record> {
- let bytes: Vec<u8> = c
+ let (bytes, failure): (Vec<u8>, Option<Vec<u8>>) = c
.query_row(
- "SELECT record FROM handles WHERE id=?1",
+ "SELECT record,failure FROM handles WHERE id=?1",
params![id.as_slice()],
- |r| r.get(0),
+ |r| Ok((r.get(0)?, r.get(1)?)),
)
.map_err(local)?;
- decode(&bytes)
+ decoded_record(&bytes, failure)
}
fn chunk(c: &Connection, id: [u8; 16], index: u64) -> Result<Vec<u8>> {
Ok(c.query_row(
@@ -102,39 +159,57 @@ fn put_chunk(c: &Connection, id: [u8; 16], index: u64, bytes: &[u8]) -> Result<(
Ok(())
}
fn records(c: &Connection) -> Result<Vec<Record>> {
- let mut query = c
- .prepare("SELECT record FROM handles ORDER BY id")
- .map_err(local)?;
- let bytes = query
- .query_map([], |r| r.get::<_, Vec<u8>>(0))
- .map_err(local)?
- .collect::<std::result::Result<Vec<_>, _>>()
+ // Inspect must also work before an older staging database is opened/upgraded.
+ let sql = if has_failure(c, "handles")? {
+ "SELECT record,failure FROM handles ORDER BY id"
+ } else {
+ "SELECT record,NULL FROM handles ORDER BY id"
+ };
+ let mut query = c.prepare(sql).map_err(local)?;
+ let rows = query
+ .query_map([], |r| {
+ Ok((r.get::<_, Vec<u8>>(0)?, r.get::<_, Option<Vec<u8>>>(1)?))
+ })
.map_err(local)?;
- bytes
- .iter()
- .map(|b| decode(b))
- .filter(|r| !matches!(r, Ok(record) if !record.ready))
- .collect()
+ let mut result = Vec::new();
+ for row in rows {
+ let (bytes, failure) = row.map_err(local)?;
+ let record = decoded_record(&bytes, failure)?;
+ if record.ready {
+ result.push(record);
+ }
+ }
+ Ok(result)
}
fn errors(c: &Connection) -> Result<Vec<WritebackError>> {
- let mut query = c
- .prepare("SELECT sequence,handle,message FROM errors ORDER BY sequence")
- .map_err(local)?;
+ let sql = if has_failure(c, "errors")? {
+ "SELECT sequence,handle,message,failure FROM errors ORDER BY sequence"
+ } else {
+ "SELECT sequence,handle,message,NULL FROM errors ORDER BY sequence"
+ };
+ let mut query = c.prepare(sql).map_err(local)?;
let rows = query
.query_map([], |r| {
Ok((
r.get::<_, i64>(0)?,
r.get::<_, Vec<u8>>(1)?,
r.get::<_, String>(2)?,
+ r.get::<_, Option<Vec<u8>>>(3)?,
))
})
.map_err(local)?;
rows.map(|r| {
- let (sequence, id, message) = r.map_err(local)?;
+ let (sequence, id, message, failure) = r.map_err(local)?;
+ let error = match failure {
+ Some(bytes) => {
+ borsh::from_slice(&bytes).map_err(|e| Error::local_kind(ErrorKind::Corrupt, e))?
+ }
+ None => Failure::legacy(message),
+ };
Ok(WritebackError {
sequence: sequence as u64,
handle: uuid::Uuid::from_slice(&id).map_err(local)?,
- message,
+ error,
persisted: true,
})
})
@@ -150,7 +225,7 @@ impl Stage {
r.get(0)
})
.map_err(local)?;
- if profile != "yinyang-stage-1" {
+ if profile != "yinyang-stage-1" && profile != "yinyang-stage-2" {
return Err(Error::Invalid("unknown staging profile"));
}
Ok(RuntimeStatus {
@@ -165,7 +240,10 @@ impl Stage {
tokio::task::spawn_blocking(move || {
std::fs::create_dir_all(&path).map_err(local)?;
let lock = std::fs::OpenOptions::new().create(true).truncate(false).read(true).write(true).open(path.join("runtime.lock")).map_err(local)?;
- lock.try_lock().map_err(|_|Error::Invalid("staging directory is already in use"))?;
+ lock.try_lock().map_err(|e| match e {
+ std::fs::TryLockError::WouldBlock => Error::State(ErrorKind::Busy, "staging directory is already in use"),
+ std::fs::TryLockError::Error(e) => local(e),
+ })?;
let mut c = Connection::open(path.join("staging.db")).map_err(local)?;
c.execute_batch("PRAGMA journal_mode=WAL; PRAGMA synchronous=FULL;
CREATE TABLE IF NOT EXISTS authority(singleton INTEGER PRIMARY KEY CHECK(singleton=1), profile TEXT NOT NULL, filesystem BLOB NOT NULL);
@@ -175,7 +253,13 @@ impl Stage {
let tx = c.transaction().map_err(local)?;
tx.execute("INSERT OR IGNORE INTO authority VALUES(1,'yinyang-stage-1',?1)",params![fs.as_bytes().as_slice()]).map_err(local)?;
let (profile,bound): (String,Vec<u8>) = tx.query_row("SELECT profile,filesystem FROM authority WHERE singleton=1",[],|r|Ok((r.get(0)?,r.get(1)?))).map_err(local)?;
- if profile!="yinyang-stage-1" || bound!=fs.as_bytes() { return Err(Error::Invalid("staging belongs to a different filesystem or profile")); }
+ if (profile!="yinyang-stage-1" && profile!="yinyang-stage-2") || bound!=fs.as_bytes() { return Err(Error::Invalid("staging belongs to a different filesystem or profile")); }
+ for table in ["handles", "errors"] {
+ if !has_failure(&tx, table)? {
+ tx.execute_batch(&format!("ALTER TABLE {table} ADD COLUMN failure BLOB")).map_err(local)?;
+ }
+ }
+ tx.execute("UPDATE authority SET profile='yinyang-stage-2' WHERE singleton=1", []).map_err(local)?;
// Incomplete opens have never been exposed or accepted application writes.
let all = {
let mut q=tx.prepare("SELECT record FROM handles").map_err(local)?;
@@ -241,7 +325,7 @@ impl Stage {
let end = offset
.checked_add(bytes.len() as u64)
.filter(|n| *n <= i64::MAX as u64)
- .ok_or(Error::Invalid("file length overflows"))?;
+ .ok_or(Error::State(ErrorKind::TooLarge, "file length overflows"))?;
if bytes.is_empty() {
return Ok(offset);
}
@@ -271,7 +355,7 @@ impl Stage {
pub async fn truncate(&self, id: [u8; 16], length: u64) -> Result<()> {
self.call(move |c| {
if length > i64::MAX as u64 {
- return Err(Error::Invalid("file length overflows"));
+ return Err(Error::State(ErrorKind::TooLarge, "file length overflows"));
}
let tx = c.transaction().map_err(local)?;
let mut r = record(&tx, id)?;
@@ -302,15 +386,20 @@ impl Stage {
})
.await
}
- pub async fn failure(&self, id: [u8; 16], message: String) -> Result<()> {
+ pub async fn failure(&self, id: [u8; 16], failure: Failure) -> Result<()> {
self.call(move |c| {
let tx = c.transaction().map_err(local)?;
let mut r = record(&tx, id)?;
- r.error = Some(message.clone());
+ r.error = Some(failure.to_string());
+ r.failure = Some(failure.clone());
save(&tx, &r)?;
tx.execute(
- "INSERT INTO errors(handle,message) VALUES(?1,?2)",
- params![id.as_slice(), message],
+ "INSERT INTO errors(handle,message,failure) VALUES(?1,?2,?3)",
+ params![
+ id.as_slice(),
+ failure.to_string(),
+ borsh::to_vec(&failure).map_err(local)?
+ ],
)
.map_err(local)?;
tx.commit().map_err(local)?;
@@ -356,10 +445,11 @@ impl Stage {
}
fn mutable(r: &Record) -> Result<()> {
if !r.ready || !r.writable {
- return Err(Error::Invalid("handle is not writable"));
+ return Err(Error::State(ErrorKind::ReadOnly, "handle is not writable"));
}
if r.plan.is_some() {
- return Err(Error::Invalid(
+ return Err(Error::State(
+ ErrorKind::Frozen,
"resolve or abort the frozen request before writing",
));
}
@@ -369,6 +459,150 @@ fn mutable(r: &Record) -> Result<()> {
mod tests {
use super::*;
+ #[tokio::test]
+ async fn release_retains_typed_failure_when_the_ledger_is_unwritable() {
+ let temp = tempfile::tempdir().unwrap();
+ let backend = crate::support::TestBackend::default();
+ let authority = Arc::new(
+ yinyang_core::Fs::create(backend.operator(), yinyang_core::BackendProfile::Minio)
+ .await
+ .unwrap(),
+ );
+ let runtime = Runtime::open(authority.clone(), temp.path()).await.unwrap();
+ runtime.create_file(authority.root(), "file").await.unwrap();
+ let mut file = runtime.open_file("file", false).await.unwrap();
+ runtime
+ .inner
+ .stage
+ .db
+ .lock()
+ .await
+ .execute_batch("PRAGMA query_only=ON")
+ .unwrap();
+ assert_eq!(
+ file.write(0, b"rejected").await.unwrap_err().kind(),
+ ErrorKind::ReadOnly
+ );
+ let id = file.release().unwrap();
+ let status = runtime.status().await.unwrap();
+ assert_eq!(status.errors.len(), 1);
+ assert!(!status.errors[0].persisted);
+ assert_eq!(status.errors[0].error.kind(), ErrorKind::ReadOnly);
+ let mut file = runtime.recover(id).await.unwrap();
+ assert_eq!(file.close().await.unwrap_err().kind(), ErrorKind::ReadOnly);
+ runtime
+ .inner
+ .stage
+ .db
+ .lock()
+ .await
+ .execute_batch("PRAGMA query_only=OFF")
+ .unwrap();
+ file.acknowledge_error().await.unwrap();
+ file.close().await.unwrap();
+ }
+
+ #[tokio::test]
+ async fn stage_one_upgrade_preserves_pending_data_requests_and_legacy_errors() {
+ let temp = tempfile::tempdir().unwrap();
+ let path = temp.path().join("staging.db");
+ let fs = NodeId::generate();
+ let id = [7_u8; 16];
+ // Independently encode stage-1 fields, including an opaque frozen request.
+ let bytes = borsh::to_vec(&(
+ id,
+ [8_u8; 16],
+ [9_u8; 24],
+ 3_u64,
+ 2_u64,
+ 1_u64,
+ true,
+ true,
+ Some(vec![1_u8, 2, 3]),
+ false,
+ Some(String::from("legacy failure")),
+ ))
+ .unwrap();
+ {
+ let c = Connection::open(&path).unwrap();
+ c.execute_batch("CREATE TABLE authority(singleton INTEGER PRIMARY KEY, profile TEXT, filesystem BLOB);
+ CREATE TABLE handles(id BLOB PRIMARY KEY, record BLOB);
+ CREATE TABLE chunks(handle BLOB, position INTEGER, data BLOB, PRIMARY KEY(handle,position));
+ CREATE TABLE errors(sequence INTEGER PRIMARY KEY AUTOINCREMENT, handle BLOB, message TEXT);").unwrap();
+ c.execute(
+ "INSERT INTO authority VALUES(1,'yinyang-stage-1',?1)",
+ [fs.as_bytes().as_slice()],
+ )
+ .unwrap();
+ c.execute(
+ "INSERT INTO handles VALUES(?1,?2)",
+ params![id.as_slice(), &bytes],
+ )
+ .unwrap();
+ c.execute(
+ "INSERT INTO chunks VALUES(?1,0,?2)",
+ params![id.as_slice(), b"abc".as_slice()],
+ )
+ .unwrap();
+ c.execute(
+ "INSERT INTO errors(handle,message) VALUES(?1,'legacy failure')",
+ [id.as_slice()],
+ )
+ .unwrap();
+ }
+ let before = Stage::inspect(path.clone()).await.unwrap();
+ assert_eq!(
+ before.handles[0].error.as_ref().unwrap().kind(),
+ ErrorKind::Unclassified
+ );
+ assert!(before.handles[0].pending && before.handles[0].frozen);
+ let stage = Stage::open(temp.path().to_owned(), fs).await.unwrap();
+ let r = stage.load(id).await.unwrap();
+ assert_eq!(borsh::to_vec(&r).unwrap(), bytes);
+ assert_eq!(stage.read(id, 0, 10).await.unwrap(), b"abc");
+ let commit = CommitId::generate();
+ stage
+ .failure(id, Failure::capture(&Error::Unknown(commit)))
+ .await
+ .unwrap();
+ stage.acknowledge(u64::MAX).await.unwrap();
+ drop(stage);
+ let stage = Stage::open(temp.path().to_owned(), fs).await.unwrap();
+ let status = stage.status().await.unwrap();
+ assert!(status.errors.is_empty());
+ assert_eq!(
+ status.handles[0].error.as_ref().unwrap().commit_id(),
+ Some(commit)
+ );
+ assert_eq!(stage.load(id).await.unwrap().plan, Some(vec![1, 2, 3]));
+ assert_eq!(stage.read(id, 0, 3).await.unwrap(), b"abc");
+ assert_eq!(
+ Stage::inspect(path).await.unwrap().handles[0]
+ .error
+ .as_ref()
+ .unwrap()
+ .kind(),
+ ErrorKind::Unknown
+ );
+ }
+
+ #[test]
+ fn local_failure_classification_does_not_parse_messages() {
+ let full = rusqlite::Error::SqliteFailure(
+ rusqlite::ffi::Error::new(rusqlite::ffi::SQLITE_FULL),
+ Some("opaque".into()),
+ );
+ assert_eq!(local(full).kind(), ErrorKind::NoSpace);
+ assert_eq!(
+ local(std::io::Error::from(std::io::ErrorKind::PermissionDenied)).kind(),
+ ErrorKind::PermissionDenied
+ );
+ assert_eq!(
+ local(rusqlite::Error::QueryReturnedNoRows).kind(),
+ ErrorKind::NotFound
+ );
+ }
+
#[test]
fn cancelled_write_keeps_its_place_before_later_reads() {
let runtime = tokio::runtime::Builder::new_multi_thread()
@@ -416,7 +650,7 @@ mod tests {
let reader = tokio::spawn(async move { stage.status().await.unwrap() });
release.send(()).unwrap();
blocker.await.unwrap();
- assert_eq!(reader.await.unwrap().errors[0].message, "accepted");
+ assert_eq!(reader.await.unwrap().errors[0].error.message(), "accepted");
});
}
}
diff --git a/src/volume.rs b/src/volume.rs
index 8fe17d2..7ae8ded 100644
--- a/src/volume.rs
+++ b/src/volume.rs
@@ -163,7 +163,7 @@ impl Capabilities {
impl VolumeConfig {
pub fn from_json(bytes: &[u8]) -> Result<Self> {
serde_json::from_slice(bytes)
- .map_err(|e| Error::Local(format!("volume configuration: {e}")))
+ .map_err(|e| Error::local(format!("volume configuration: {e}")))
}
/// Negotiation is read-only. Storage capabilities are semantic prerequisites
/// for Managed operations, not claims that raw S3 implements atomic rename.
diff --git a/tests/directory.rs b/tests/directory.rs
index 27c91ed..b0d0241 100644
--- a/tests/directory.rs
+++ b/tests/directory.rs
@@ -350,7 +350,7 @@ async fn invalid_names_fail_before_upload() {
.await
.unwrap_err()
.kind(),
- ErrorKind::Invalid
+ ErrorKind::InvalidName
);
assert_eq!(backend.state.lock().unwrap().objects.len(), count);
}
diff --git a/tests/runtime.rs b/tests/runtime.rs
index bf688cc..06b3778 100644
--- a/tests/runtime.rs
+++ b/tests/runtime.rs
@@ -21,7 +21,7 @@ use std::sync::Arc;
use support::TestBackend;
use yinyang::core::service::{MetadataService, ServiceClient, serve};
use yinyang::core::{Authority, BackendProfile, CommitId, CommitOutcome, Fs, Planner};
-use yinyang::runtime::{Error, Runtime};
+use yinyang::runtime::{Error, ErrorKind, Runtime};
async fn exercise(authority: Arc<dyn Authority>, path: &std::path::Path, backend: &TestBackend) {
let runtime = Runtime::open(authority.clone(), path).await.unwrap();
@@ -87,9 +87,111 @@ async fn exercise(authority: Arc<dyn Authority>, path: &std::path::Path, backend
runtime.acknowledge_errors(sequence).await.unwrap();
assert!(runtime.status().await.unwrap().errors.is_empty());
identity_access(&runtime).await;
+ completion_and_errors(&runtime, backend).await;
version_reads(&runtime, backend).await;
}
+async fn completion_and_errors(runtime: &Runtime, backend: &TestBackend) {
+ let root = runtime.authority().root();
+ assert_eq!(
+ runtime
+ .open_file("missing", false)
+ .await
+ .err()
+ .unwrap()
+ .kind(),
+ ErrorKind::NotFound
+ );
+ assert_eq!(
+ runtime.open_node(root, false).await.err().unwrap().kind(),
+ ErrorKind::IsDirectory
+ );
+ assert_eq!(
+ runtime.create_file(root, "con").await.unwrap_err().kind(),
+ ErrorKind::InvalidName
+ );
+ runtime.create_file(root, "completion").await.unwrap();
+ assert_eq!(
+ runtime
+ .create_file(root, "completion")
+ .await
+ .unwrap_err()
+ .kind(),
+ ErrorKind::AlreadyExists
+ );
+ let mut file = runtime.open_file("completion", true).await.unwrap();
+ let node = file.status().await.unwrap().node;
+ assert_eq!(
+ runtime.create_file(node, "child").await.unwrap_err().kind(),
+ ErrorKind::NotDirectory
+ );
+ let snapshot = runtime.authority().observe_latest().await.unwrap();
+ let dir = snapshot.resolve("dir").await.unwrap().unwrap().id();
+ assert_eq!(
+ runtime.unlink(dir).await.unwrap_err().kind(),
+ ErrorKind::NotEmpty
+ );
+ assert_eq!(
+ runtime.recover(file.id()).await.err().unwrap().kind(),
+ ErrorKind::Busy
+ );
+ file.write(0, b"local only").await.unwrap();
+ let local = file.sync_local().await.unwrap();
+ assert!(local.pending);
+ assert_eq!(local.remote_generation, 0);
+ let writes = backend.state.lock().unwrap().written_bytes;
+ let id = file.release().unwrap();
+ assert_eq!(backend.state.lock().unwrap().written_bytes, writes);
+ assert!(
+ runtime
+ .authority()
+ .observe_latest()
+ .await
+ .unwrap()
+ .open_file(node)
+ .await
+ .unwrap()
+ .read(0, 20)
+ .await
+ .unwrap()
+ .is_empty()
+ );
+ let mut file = runtime.recover(id).await.unwrap();
+ assert_eq!(file.read(0, 20).await.unwrap(), b"local only");
+ backend.state.lock().unwrap().fail_data_close = true;
+ assert_eq!(file.close().await.unwrap_err().kind(), ErrorKind::Storage);
+ let id = file.release().unwrap();
+ let status = runtime.status().await.unwrap();
+ let state = status.handles.iter().find(|s| s.id == id).unwrap();
+ assert!(state.pending);
+ assert_eq!(state.error.as_ref().unwrap().kind(), ErrorKind::Storage);
+ assert!(
+ status
+ .errors
+ .iter()
+ .any(|e| e.handle == id && e.error.kind() == ErrorKind::Storage)
+ );
+ backend.state.lock().unwrap().fail_data_close = false;
+ let mut file = runtime.recover(id).await.unwrap();
+ file.fsync().await.unwrap();
+ assert!(file.status().await.unwrap().error.is_none());
+ file.close().await.unwrap();
+ assert_eq!(file.read(0, 1).await.unwrap_err().kind(), ErrorKind::Closed);
+ let mut read_only = runtime.open_node(node, false).await.unwrap();
+ assert_eq!(
+ read_only.write(0, b"no").await.unwrap_err().kind(),
+ ErrorKind::ReadOnly
+ );
+ let id = read_only.release().unwrap();
+ let mut read_only = runtime.recover(id).await.unwrap();
+ assert_eq!(
+ read_only.close().await.unwrap_err().kind(),
+ ErrorKind::ReadOnly
+ );
+ read_only.acknowledge_error().await.unwrap();
+ read_only.close().await.unwrap();
+}
+
async fn version_reads(runtime: &Runtime, backend: &TestBackend) {
let root = runtime.authority().root();
let bytes: Vec<u8> = (0..16 * 65536).map(|i| (i % 251) as u8).collect();
@@ -331,12 +433,21 @@ async fn upload_failure_and_unknown_publication_keep_frozen_identity() {
let Err(Error::Unknown(id)) = file.fsync().await else {
panic!("expected unknown");
};
+ let unknown = file.status().await.unwrap().error.unwrap();
+ assert_eq!(unknown.kind(), ErrorKind::Unknown);
+ assert_eq!(unknown.commit_id(), Some(id));
assert!(file.status().await.unwrap().frozen);
assert!(file.abort().await.is_err());
assert!(file.write(0, b"different").await.is_err());
- let handle = file.id();
- drop(file);
+ let handle = file.release().unwrap();
drop(runtime);
+ let inspected = Runtime::inspect(temp.path()).await.unwrap();
+ assert!(
+ inspected
+ .errors
+ .iter()
+ .any(|e| e.error.kind() == ErrorKind::Unknown && e.error.commit_id() == Some(id))
+ );
// A later retry fences the delayed object CAS without changing the original request.
let runtime = Runtime::open(authority.clone(), temp.path()).await.unwrap();
let mut file = runtime.recover(handle).await.unwrap();