feat: model operation outcome guarantees

This commit is contained in:
2026-09-30 05:52:38 +09:00
parent 027555eb4e
commit 6de58da17d
8 changed files with 1105 additions and 150 deletions
+487 -83
View File
@@ -18,11 +18,14 @@
//!
//! A provider is responsible for entering a backend transaction, lock,
//! stable-handle scope, or conditional-update primitive before invoking the
//! callback and retaining it until the callback returns. The runtime does not
//! reopen the path or select another implementation after checking validators,
//! but cannot make an arbitrary external backend atomic on its own. Mutable
//! operations must make object-validator comparison and backend commit atomic;
//! the validator returned after dispatch must describe the committed state.
//! operation callback and retaining it through commit or rollback. The typed
//! [`OperationAttempt`] boundary distinguishes pre-dispatch rejection from started
//! execution. Only an implementation-supplied non-commit guarantee or a
//! provider-confirmed rollback allows a post-invocation failure to become
//! `Internal` or `ResourceLimitExceeded`; every other outcome is
//! `OperationOutcomeUnknown`. Mutable operations must make object-validator
//! comparison and backend commit atomic, and the validator returned after dispatch
//! must describe the state that becomes committed.
#![deny(missing_docs)]
@@ -31,19 +34,23 @@ use std::collections::{BTreeMap, HashMap, HashSet};
use std::error::Error;
use std::fmt::{self, Display, Formatter};
use std::marker::PhantomData;
use std::panic::{AssertUnwindSafe, catch_unwind};
use std::sync::Arc;
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::time::Instant;
use wip_interface::DispatchError;
pub use wip_interface::{InterfaceDefinition, InterfaceImplementation};
pub use wip_interface::{
EffectGuarantee, HostError, HostErrorKind, InterfaceDefinition, InterfaceImplementation,
};
pub use wip_protocol;
pub use wip_protocol::{
CallOperationRequest as CallRequest, CallOperationResponse as CallResult, InterfaceTarget,
Target,
};
use wip_protocol::{
FetchInterfaceResponse, InterfaceDescriptor, Object, ProtocolError, ProtocolErrorCode,
ValidationError, Value,
FetchInterfaceRequest, FetchInterfaceResponse, InterfaceDescriptor, Object, ProtocolError,
ProtocolErrorCode, ValidationError, Value,
};
/// Opaque bytes used as a conditional execution precondition.
@@ -64,6 +71,122 @@ impl Validator {
}
}
/// Runtime-neutral cooperative cancellation signal.
///
/// Clones observe the same atomic state, so transports, providers, and worker
/// runtimes can share the token without depending on a particular async runtime.
#[derive(Debug, Clone, Default)]
pub struct CancellationToken(Arc<AtomicBool>);
impl CancellationToken {
/// Creates a cancellation token in the active state.
#[must_use]
pub fn new() -> Self {
Self::default()
}
/// Requests cooperative cancellation.
pub fn cancel(&self) {
self.0.store(true, Ordering::Release);
}
/// Returns whether cancellation has been requested.
#[must_use]
pub fn is_cancelled(&self) -> bool {
self.0.load(Ordering::Acquire)
}
}
/// Execution controls propagated through object resolution and operation dispatch.
///
/// Cancellation and deadlines are checked immediately before typed implementation
/// invocation. Providers also receive the complete context and remain responsible
/// for enforcing controls while backend work is in progress. `resource_budget` is
/// an opaque Host-defined work-unit ceiling; this crate deliberately does not
/// prescribe an async runtime or accounting mechanism.
#[derive(Debug, Clone)]
pub struct ExecutionContext {
cancellation: CancellationToken,
deadline: Option<Instant>,
resource_budget: Option<u64>,
}
impl Default for ExecutionContext {
fn default() -> Self {
Self {
cancellation: CancellationToken::new(),
deadline: None,
resource_budget: None,
}
}
}
impl ExecutionContext {
/// Creates an execution context without a deadline or resource budget.
#[must_use]
pub fn new() -> Self {
Self::default()
}
/// Uses `cancellation` as this call's cooperative cancellation signal.
#[must_use]
pub fn with_cancellation(mut self, cancellation: CancellationToken) -> Self {
self.cancellation = cancellation;
self
}
/// Sets the monotonic deadline for this call.
#[must_use]
pub fn with_deadline(mut self, deadline: Instant) -> Self {
self.deadline = Some(deadline);
self
}
/// Sets an opaque Host-defined work-unit ceiling for this call.
#[must_use]
pub fn with_resource_budget(mut self, resource_budget: u64) -> Self {
self.resource_budget = Some(resource_budget);
self
}
/// Returns the shared cooperative cancellation token.
#[must_use]
pub fn cancellation(&self) -> &CancellationToken {
&self.cancellation
}
/// Returns the monotonic deadline, if one was supplied.
#[must_use]
pub fn deadline(&self) -> Option<Instant> {
self.deadline
}
/// Returns the opaque Host-defined work-unit ceiling, if one was supplied.
#[must_use]
pub fn resource_budget(&self) -> Option<u64> {
self.resource_budget
}
fn preflight(&self) -> Result<(), CallError> {
if self.cancellation.is_cancelled() {
return Err(CallError::new(
CallErrorKind::ResourceLimitExceeded,
"operation was cancelled before execution",
));
}
if self
.deadline
.is_some_and(|deadline| Instant::now() >= deadline)
{
return Err(CallError::new(
CallErrorKind::ResourceLimitExceeded,
"operation deadline elapsed before execution",
));
}
Ok(())
}
}
/// Opaque host-assigned reference to one registered interface artifact.
#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
pub struct InterfaceRef(String);
@@ -88,14 +211,26 @@ pub struct InterfaceObservation {
}
impl InterfaceObservation {
/// Converts this host observation to the canonical protocol response.
#[must_use]
pub fn protocol_response(&self) -> FetchInterfaceResponse {
FetchInterfaceResponse {
/// Converts this host observation to a validated canonical protocol response.
///
/// A Host-side response construction failure is returned as `Internal`; it is
/// never emitted as a successful response for a Client to reinterpret as a
/// local validation error.
pub fn protocol_response(&self) -> Result<FetchInterfaceResponse, ProtocolError> {
let response = FetchInterfaceResponse {
interface: self.reference.0.clone(),
descriptor: self.descriptor.clone(),
validator: self.validator.as_ref().map(|validator| validator.0.clone()),
}
};
response
.validate_for(&FetchInterfaceRequest {
interface: self.reference.0.clone(),
})
.map_err(|_| ProtocolError {
code: ProtocolErrorCode::Internal,
message: "Host produced an invalid interface response".into(),
})?;
Ok(response)
}
}
@@ -397,26 +532,29 @@ impl<T: WipObject> PublishedObject<T> {
fn dispatch_resolved(
&self,
context: &ExecutionContext,
request: CallRequest,
authorize: impl FnOnce(&T, &InterfaceRef, &str) -> bool,
) -> Result<CallResult, CallError> {
) -> OperationAttempt {
let pre_dispatch_failure =
|kind, message: &str| OperationAttempt::NotStarted(CallError::new(kind, message));
let current_object_validator = self.implementation.validator();
if let Some(supplied) = request.target.validator.as_deref()
&& current_object_validator.map(Validator::as_bytes) != Some(supplied)
{
return Err(CallError::new(
return pre_dispatch_failure(
CallErrorKind::ValidatorMismatch,
"object validator does not match current object state",
));
);
}
let Some(bound) = self.interfaces.iter().find(|bound| {
bound.artifact.reference.as_str() == request.interface.reference.as_str()
}) else {
return Err(CallError::new(
return pre_dispatch_failure(
CallErrorKind::InterfaceMismatch,
"interface is not a member of the current object publication",
));
);
};
match (
@@ -424,16 +562,16 @@ impl<T: WipObject> PublishedObject<T> {
request.interface.validator.as_deref(),
) {
(Some(_), None) => {
return Err(CallError::new(
return pre_dispatch_failure(
CallErrorKind::InterfaceValidatorRequired,
"registered interface requires a validator",
));
);
}
(expected, Some(actual)) if expected.map(Validator::as_bytes) != Some(actual) => {
return Err(CallError::new(
return pre_dispatch_failure(
CallErrorKind::InterfaceValidatorMismatch,
"interface validator does not match the registered descriptor",
));
);
}
_ => {}
}
@@ -445,10 +583,10 @@ impl<T: WipObject> PublishedObject<T> {
.iter()
.any(|operation| operation.name == request.operation)
{
return Err(CallError::new(
return pre_dispatch_failure(
CallErrorKind::OperationNotFound,
format!("operation `{}` is not declared", request.operation),
));
&format!("operation `{}` is not declared", request.operation),
);
}
if request.target.validator.is_none()
@@ -456,45 +594,67 @@ impl<T: WipObject> PublishedObject<T> {
.implementation
.requires_validator(&bound.artifact.reference, &request.operation)
{
return Err(CallError::new(
return pre_dispatch_failure(
CallErrorKind::ValidatorRequired,
"this operation requires an object validator",
));
);
}
bound
if let Err(error) = bound
.artifact
.descriptor
.validate_arguments(&request.operation, &request.arguments)
.map_err(|error| CallError::new(CallErrorKind::InvalidArguments, error.to_string()))?;
{
return pre_dispatch_failure(CallErrorKind::InvalidArguments, &error.to_string());
}
if !authorize(
&self.implementation,
&bound.artifact.reference,
&request.operation,
) {
return Err(CallError::new(
return pre_dispatch_failure(
CallErrorKind::PermissionDenied,
"operation is not permitted for this caller",
));
);
}
let value = (bound.dispatch)(&self.implementation, &request.operation, request.arguments)
.map_err(CallError::from_dispatch)?;
if let Err(error) = context.preflight() {
return OperationAttempt::NotStarted(error);
}
bound
let value =
match (bound.dispatch)(&self.implementation, &request.operation, request.arguments) {
Ok(value) => value,
Err(
error @ (DispatchError::UnknownOperation(_)
| DispatchError::InvalidArguments(_)
| DispatchError::Decode(_)),
) => return OperationAttempt::NotStarted(CallError::from_pre_dispatch(error)),
Err(error) => {
return OperationAttempt::Started(Err(OperationFailure::from_dispatch(error)));
}
};
if bound
.artifact
.descriptor
.validate_result(&request.operation, &value)
.map_err(|error| CallError::new(CallErrorKind::InvalidResult, error.to_string()))?;
.is_err()
{
return OperationAttempt::Started(Err(OperationFailure::internal(
EffectGuarantee::Unknown,
"operation implementation produced an invalid result",
)));
}
Ok(CallResult {
OperationAttempt::Started(Ok(CallResult {
result: value,
validator: self
.implementation
.validator()
.map(|validator| validator.0.clone()),
})
}))
}
/// Consumes the publication and returns its host-private implementation.
@@ -504,24 +664,158 @@ impl<T: WipObject> PublishedObject<T> {
}
}
/// A post-invocation failure awaiting final provider effect disposition.
///
/// Providers may upgrade an unknown failure with [`confirm_rollback`](Self::confirm_rollback)
/// only after the backend transaction has definitely rolled back. Diagnostics are
/// intentionally generic and safe for protocol exposure.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct OperationFailure {
kind: HostErrorKind,
effect_guarantee: EffectGuarantee,
message: String,
}
impl OperationFailure {
fn internal(effect_guarantee: EffectGuarantee, message: impl Into<String>) -> Self {
Self {
kind: HostErrorKind::Internal,
effect_guarantee,
message: bounded_message(message.into()),
}
}
fn from_dispatch(error: DispatchError) -> Self {
match error {
DispatchError::Host(error) => Self {
kind: error.kind(),
effect_guarantee: error.effect_guarantee(),
message: match error.kind() {
HostErrorKind::Internal => "operation implementation failed".into(),
HostErrorKind::ResourceLimitExceeded => {
"operation resource limit was exceeded".into()
}
},
},
DispatchError::Panic => Self::internal(
EffectGuarantee::Unknown,
"operation implementation panicked",
),
DispatchError::Encode(_) => Self::internal(
EffectGuarantee::Unknown,
"operation result could not be encoded",
),
DispatchError::InvalidResult(_) => Self::internal(
EffectGuarantee::Unknown,
"operation implementation produced an invalid result",
),
DispatchError::UnknownOperation(_)
| DispatchError::InvalidArguments(_)
| DispatchError::Decode(_) => {
unreachable!("pre-dispatch failures are classified before OperationFailure")
}
}
}
/// Returns the Host failure category used when effects are not committed.
#[must_use]
pub fn kind(&self) -> HostErrorKind {
self.kind
}
/// Returns the current effect guarantee.
#[must_use]
pub fn effect_guarantee(&self) -> EffectGuarantee {
self.effect_guarantee
}
/// Confirms that the provider rolled back all operation effects.
///
/// Calling this method is a semantic assertion by the provider and must occur
/// only after its atomic backend boundary has completed rollback.
#[must_use]
pub fn confirm_rollback(mut self) -> Self {
self.effect_guarantee = EffectGuarantee::NotCommitted;
self
}
}
/// Dispatch progress returned to the provider's atomic execution boundary.
#[derive(Debug)]
pub enum OperationAttempt {
/// A canonical precondition failed and typed implementation code did not run.
NotStarted(CallError),
/// Typed implementation invocation began, with either a result or a typed
/// post-invocation failure.
Started(Result<CallResult, OperationFailure>),
}
impl OperationAttempt {
/// Marks a post-invocation failure as fully rolled back.
///
/// Successful attempts and pre-dispatch failures are returned unchanged.
#[must_use]
pub fn confirm_rollback(self) -> Self {
match self {
Self::Started(Err(error)) => Self::Started(Err(error.confirm_rollback())),
other => other,
}
}
}
/// Result of one provider resolution or execution boundary.
#[derive(Debug)]
pub enum ProviderResult<T> {
/// The path resolved and the provider callback completed.
Found(T),
/// The path is not currently published to this caller.
NotFound,
/// Provider or backend execution failed.
///
/// During `call_operation`, the embedded effect guarantee determines whether
/// the failure is safe to expose as `Internal`/`ResourceLimitExceeded` or must
/// become `OperationOutcomeUnknown`.
Failure(HostError),
}
/// A request-scoped Worldspace resolver and execution boundary.
///
/// `with_object` must resolve the current object at `path` exactly once and pass
/// a publication that pins its current interface set, implementations, and
/// backend target for the duration of `operation`. Implementations may hold a
/// transaction or lock around the callback. Returning `None` means the path is
/// not published to the caller. The borrowed publication cannot escape the
/// callback, so the default call path cannot outlive that provider boundary.
/// backend target for the duration of `operation`. The execution context is the
/// runtime-neutral carrier for cancellation, deadline, and resource controls.
/// The borrowed publication cannot escape the callback.
///
/// Providers that can roll back mutable operations should override
/// [`ObjectProvider::execute_operation`], hold their transaction across the
/// callback, and call [`OperationAttempt::confirm_rollback`] only after rollback
/// succeeds. The default boundary conservatively preserves unknown outcomes.
pub trait ObjectProvider {
/// Host-private object type produced by this provider.
type Object: WipObject;
/// Runs an operation against the single object snapshot resolved at `path`.
/// Runs a callback against the single object snapshot resolved at `path`.
fn with_object<R>(
&self,
context: &ExecutionContext,
path: &str,
operation: impl FnOnce(&PublishedObject<Self::Object>) -> R,
) -> Option<R>;
) -> ProviderResult<R>;
/// Runs one operation attempt inside the provider's atomic backend boundary.
///
/// The default implementation delegates to [`ObjectProvider::with_object`]
/// and therefore does not add a rollback guarantee. Transactional providers
/// can override this method, inspect the returned [`OperationAttempt`], roll
/// back a failed started attempt, and return its confirmed form.
fn execute_operation(
&self,
context: &ExecutionContext,
path: &str,
operation: impl FnOnce(&PublishedObject<Self::Object>) -> OperationAttempt,
) -> ProviderResult<OperationAttempt> {
self.with_object(context, path, operation)
}
/// Decides whether a declared operation may execute for the current caller.
///
@@ -540,16 +834,43 @@ pub trait ObjectProvider {
/// Fetches the canonical object projection currently published at `path`.
fn fetch(&self, path: &str) -> Result<Object, CallError> {
self.fetch_with_context(&ExecutionContext::default(), path)
}
/// Fetches an object while propagating runtime execution controls.
fn fetch_with_context(
&self,
context: &ExecutionContext,
path: &str,
) -> Result<Object, CallError> {
wip_protocol::validate_path(path)
.map_err(|error| CallError::new(CallErrorKind::InvalidRequest, error.to_string()))?;
self.with_object(path, PublishedObject::protocol_object)
.ok_or_else(|| CallError::new(CallErrorKind::NotFound, "target path is not published"))
.and_then(|object| {
object
.validate_at_path(path)
.map_err(|error| CallError::new(CallErrorKind::Internal, error.to_string()))?;
Ok(object)
})
let resolved = catch_unwind(AssertUnwindSafe(|| {
self.with_object(context, path, PublishedObject::protocol_object)
}))
.map_err(|_| {
CallError::new(
CallErrorKind::Internal,
"provider panicked while fetching an object",
)
})?;
let object = match resolved {
ProviderResult::Found(object) => object,
ProviderResult::NotFound => {
return Err(CallError::new(
CallErrorKind::NotFound,
"target path is not published",
));
}
ProviderResult::Failure(error) => return Err(CallError::from_read_failure(error)),
};
object.validate_at_path(path).map_err(|_| {
CallError::new(
CallErrorKind::Internal,
"Host produced an invalid object response",
)
})?;
Ok(object)
}
/// Resolves and executes a canonical path-targeted operation request.
@@ -559,17 +880,47 @@ pub trait ObjectProvider {
/// supplied object validator mismatch, membership, interface validator
/// requirement and mismatch, operation existence, object validator
/// requirement, arguments, then authorization. No failed precondition
/// invokes the provider callback or typed implementation.
/// invokes typed implementation code.
fn call_operation(&self, request: CallRequest) -> Result<CallResult, CallError> {
self.call_operation_with_context(&ExecutionContext::default(), request)
}
/// Executes an operation while propagating cancellation, deadline, and
/// resource-budget controls to the provider and runtime preflight check.
fn call_operation_with_context(
&self,
context: &ExecutionContext,
request: CallRequest,
) -> Result<CallResult, CallError> {
wip_protocol::validate_path(&request.target.path)
.map_err(|error| CallError::new(CallErrorKind::InvalidRequest, error.to_string()))?;
let path = request.target.path.clone();
self.with_object(&path, |publication| {
publication.dispatch_resolved(request, |object, interface, operation| {
self.authorize(&path, object, interface, operation)
let execution = catch_unwind(AssertUnwindSafe(|| {
self.execute_operation(context, &path, |publication| {
publication.dispatch_resolved(context, request, |object, interface, operation| {
self.authorize(&path, object, interface, operation)
})
})
})
.ok_or_else(|| CallError::new(CallErrorKind::NotFound, "target path is not published"))?
}));
let execution = execution.map_err(|_| {
CallError::from_operation_failure(OperationFailure::internal(
EffectGuarantee::Unknown,
"provider execution panicked",
))
})?;
match execution {
ProviderResult::Found(OperationAttempt::NotStarted(error)) => Err(error),
ProviderResult::Found(OperationAttempt::Started(Ok(result))) => Ok(result),
ProviderResult::Found(OperationAttempt::Started(Err(error))) => {
Err(CallError::from_operation_failure(error))
}
ProviderResult::NotFound => Err(CallError::new(
CallErrorKind::NotFound,
"target path is not published",
)),
ProviderResult::Failure(error) => Err(CallError::from_host_failure(error)),
}
}
}
@@ -594,7 +945,7 @@ impl Display for PublicationError {
impl Error for PublicationError {}
/// Stable host dispatch failure kind.
/// Stable Host dispatch failure kind.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum CallErrorKind {
/// The request is structurally invalid.
@@ -607,10 +958,13 @@ pub enum CallErrorKind {
OperationNotFound,
/// Supplied named values do not satisfy the operation declaration.
InvalidArguments,
/// The Host constructed an invalid response before operation execution.
/// A bounded cancellation, deadline, or resource limit was exceeded and the
/// operation effects are guaranteed not committed.
ResourceLimitExceeded,
/// The Host failed and operation effects are guaranteed not committed.
Internal,
/// The encoded implementation result violates its declaration after execution.
InvalidResult,
/// The Host cannot guarantee whether operation effects were committed.
OperationOutcomeUnknown,
/// This operation requires a validator and the call omitted it.
ValidatorRequired,
/// The supplied object validator is stale or belongs to another object.
@@ -621,11 +975,9 @@ pub enum CallErrorKind {
InterfaceValidatorMismatch,
/// The operation is declared but not authorized for the caller.
PermissionDenied,
/// The typed implementation reported a host/backend failure.
Implementation,
}
/// A dispatch failure with a stable kind and non-normative diagnostic.
/// A dispatch failure with a stable kind and bounded non-normative diagnostic.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct CallError {
kind: CallErrorKind,
@@ -636,11 +988,11 @@ impl CallError {
fn new(kind: CallErrorKind, message: impl Into<String>) -> Self {
Self {
kind,
message: message.into(),
message: bounded_message(message.into()),
}
}
fn from_dispatch(error: DispatchError) -> Self {
fn from_pre_dispatch(error: DispatchError) -> Self {
match error {
DispatchError::UnknownOperation(operation) => Self::new(
CallErrorKind::OperationNotFound,
@@ -649,34 +1001,63 @@ impl CallError {
DispatchError::InvalidArguments(error) => {
Self::new(CallErrorKind::InvalidArguments, error.to_string())
}
DispatchError::Decode(error) => {
Self::new(CallErrorKind::InvalidArguments, error.to_string())
}
DispatchError::Host(error) => {
Self::new(CallErrorKind::Implementation, error.to_string())
}
DispatchError::Encode(error) => {
Self::new(CallErrorKind::InvalidResult, error.to_string())
}
DispatchError::InvalidResult(error) => {
Self::new(CallErrorKind::InvalidResult, error.to_string())
DispatchError::Decode(_) => Self::new(
CallErrorKind::InvalidArguments,
"operation arguments could not be decoded",
),
DispatchError::Host(_)
| DispatchError::Panic
| DispatchError::Encode(_)
| DispatchError::InvalidResult(_) => {
unreachable!("post-invocation failure used as a pre-dispatch failure")
}
}
}
fn from_read_failure(error: HostError) -> Self {
match error.kind() {
HostErrorKind::Internal => Self::new(
CallErrorKind::Internal,
"provider failed while fetching data",
),
HostErrorKind::ResourceLimitExceeded => Self::new(
CallErrorKind::ResourceLimitExceeded,
"provider resource limit was exceeded while fetching data",
),
}
}
fn from_host_failure(error: HostError) -> Self {
let kind = classified_host_kind(error.kind(), error.effect_guarantee());
let message = match error.kind() {
HostErrorKind::Internal => "provider execution failed",
HostErrorKind::ResourceLimitExceeded => {
"provider execution resource limit was exceeded"
}
};
Self::new(kind, message)
}
fn from_operation_failure(error: OperationFailure) -> Self {
Self::new(
classified_host_kind(error.kind, error.effect_guarantee),
error.message,
)
}
/// Returns the stable failure kind.
#[must_use]
pub fn kind(&self) -> CallErrorKind {
self.kind
}
/// Returns the human-readable diagnostic.
/// Returns the bounded protocol-safe human-readable diagnostic.
#[must_use]
pub fn message(&self) -> &str {
&self.message
}
/// Maps this host failure to the canonical protocol error taxonomy.
/// Maps this Host failure to the canonical protocol error taxonomy.
#[must_use]
pub fn protocol_error(&self) -> ProtocolError {
let code = match self.kind {
@@ -685,10 +1066,9 @@ impl CallError {
CallErrorKind::InterfaceMismatch => ProtocolErrorCode::InterfaceMismatch,
CallErrorKind::OperationNotFound => ProtocolErrorCode::OperationNotFound,
CallErrorKind::InvalidArguments => ProtocolErrorCode::InvalidArguments,
CallErrorKind::ResourceLimitExceeded => ProtocolErrorCode::ResourceLimitExceeded,
CallErrorKind::Internal => ProtocolErrorCode::Internal,
CallErrorKind::InvalidResult | CallErrorKind::Implementation => {
ProtocolErrorCode::OperationOutcomeUnknown
}
CallErrorKind::OperationOutcomeUnknown => ProtocolErrorCode::OperationOutcomeUnknown,
CallErrorKind::ValidatorRequired => ProtocolErrorCode::ValidatorRequired,
CallErrorKind::ValidatorMismatch => ProtocolErrorCode::ValidatorMismatch,
CallErrorKind::InterfaceValidatorRequired => {
@@ -706,6 +1086,30 @@ impl CallError {
}
}
fn classified_host_kind(kind: HostErrorKind, effect_guarantee: EffectGuarantee) -> CallErrorKind {
match effect_guarantee {
EffectGuarantee::Unknown => CallErrorKind::OperationOutcomeUnknown,
EffectGuarantee::NotCommitted => match kind {
HostErrorKind::Internal => CallErrorKind::Internal,
HostErrorKind::ResourceLimitExceeded => CallErrorKind::ResourceLimitExceeded,
},
}
}
const MAX_PROTOCOL_MESSAGE_BYTES: usize = 512;
fn bounded_message(mut message: String) -> String {
if message.len() <= MAX_PROTOCOL_MESSAGE_BYTES {
return message;
}
let mut end = MAX_PROTOCOL_MESSAGE_BYTES;
while !message.is_char_boundary(end) {
end -= 1;
}
message.truncate(end);
message
}
impl Display for CallError {
fn fmt(&self, formatter: &mut Formatter<'_>) -> fmt::Result {
write!(
+327 -10
View File
@@ -1,13 +1,16 @@
use std::cell::Cell;
use std::collections::BTreeMap;
use std::rc::Rc;
use std::time::{Duration, Instant};
use wip_host::{
CallError, CallErrorKind, CallRequest, InterfaceRef, InterfaceRegistry, InterfaceTarget,
ObjectProvider, PublishedObject, RegisteredInterface, Target, Validator, WipObject,
CallError, CallErrorKind, CallRequest, CancellationToken, ExecutionContext, InterfaceRef,
InterfaceRegistry, InterfaceTarget, ObjectProvider, OperationAttempt, ProviderResult,
PublishedObject, RegisteredInterface, Target, Validator, WipObject,
};
use wip_interface::{
CodecError, HostError, OperationResult, TypeDeclaration, TypeExpr, Value, WipType, interface,
CodecError, EffectGuarantee, HostError, OperationResult, TypeDeclaration, TypeExpr, Value,
WipType, interface,
};
use wip_protocol::ProtocolErrorCode;
@@ -39,6 +42,24 @@ pub trait Hidden {
pub trait Broken {
/// Returns a deliberately invalid encoded value.
fn broken(&self) -> OperationResult<BadValue>;
/// Produces a result codec failure after implementation invocation.
fn encode_failure(&self) -> OperationResult<EncodeFailure>;
/// Panics after implementation invocation begins.
fn panic_after_call(&self) -> OperationResult<String>;
/// Reports an internal failure without an effect guarantee.
fn unknown_internal(&self) -> OperationResult<String>;
/// Reports an internal failure with a non-commit guarantee.
fn safe_internal(&self) -> OperationResult<String>;
/// Reports a resource failure without an effect guarantee.
fn unknown_resource(&self) -> OperationResult<String>;
/// Reports a resource failure with a non-commit guarantee.
fn safe_resource(&self) -> OperationResult<String>;
}
#[derive(Debug)]
@@ -60,6 +81,25 @@ impl WipType for BadValue {
}
}
#[derive(Debug)]
pub struct EncodeFailure;
impl WipType for EncodeFailure {
fn type_expr() -> TypeExpr {
TypeExpr::String
}
fn collect_declarations(_: &mut Vec<TypeDeclaration>) {}
fn encode(self) -> Result<Value, CodecError> {
Err(CodecError::new("private://backend/result could not encode"))
}
fn decode(_: Value) -> Result<Self, CodecError> {
Ok(Self)
}
}
struct Document {
name: String,
description: Option<String>,
@@ -135,7 +175,10 @@ impl Reader for Document {
fn fail(&self) -> OperationResult<String> {
self.record_call();
Err(HostError::new("backend unavailable"))
Err(HostError::internal(
"backend unavailable",
EffectGuarantee::Unknown,
))
}
}
@@ -158,6 +201,48 @@ impl Broken for Document {
self.record_call();
Ok(BadValue)
}
fn encode_failure(&self) -> OperationResult<EncodeFailure> {
self.record_call();
Ok(EncodeFailure)
}
fn panic_after_call(&self) -> OperationResult<String> {
self.record_call();
panic!("private://backend panic with credential=secret")
}
fn unknown_internal(&self) -> OperationResult<String> {
self.record_call();
Err(HostError::internal(
"private://backend/path credential=secret",
EffectGuarantee::Unknown,
))
}
fn safe_internal(&self) -> OperationResult<String> {
self.record_call();
Err(HostError::internal(
"private://backend/path credential=secret",
EffectGuarantee::NotCommitted,
))
}
fn unknown_resource(&self) -> OperationResult<String> {
self.record_call();
Err(HostError::resource_limit(
"private scheduler deadline details",
EffectGuarantee::Unknown,
))
}
fn safe_resource(&self) -> OperationResult<String> {
self.record_call();
Err(HostError::resource_limit(
"private scheduler deadline details",
EffectGuarantee::NotCommitted,
))
}
}
struct TestProvider {
@@ -169,6 +254,11 @@ struct TestProvider {
authorization_checks: Cell<usize>,
deny: Cell<bool>,
switched: Cell<bool>,
confirm_rollback: Cell<bool>,
provider_failure: Cell<Option<(bool, EffectGuarantee)>>,
observed_budget: Cell<Option<u64>>,
observed_deadline: Cell<Option<Instant>>,
observed_cancelled: Cell<bool>,
}
impl TestProvider {
@@ -186,6 +276,11 @@ impl TestProvider {
authorization_checks: Cell::new(0),
deny: Cell::new(false),
switched: Cell::new(false),
confirm_rollback: Cell::new(false),
provider_failure: Cell::new(None),
observed_budget: Cell::new(None),
observed_deadline: Cell::new(None),
observed_cancelled: Cell::new(false),
}
}
}
@@ -195,9 +290,14 @@ impl ObjectProvider for TestProvider {
fn with_object<R>(
&self,
context: &ExecutionContext,
path: &str,
operation: impl FnOnce(&PublishedObject<Self::Object>) -> R,
) -> Option<R> {
) -> ProviderResult<R> {
self.observed_budget.set(context.resource_budget());
self.observed_deadline.set(context.deadline());
self.observed_cancelled
.set(context.cancellation().is_cancelled());
self.resolutions.set(self.resolutions.get() + 1);
let (document, interfaces) = match path {
"/document" => (
@@ -243,7 +343,7 @@ impl ObjectProvider for TestProvider {
Document::open("broken", Some("broken-ref"), Rc::clone(&self.calls)),
3,
),
_ => return None,
_ => return ProviderResult::NotFound,
};
let publication = PublishedObject::new(document);
@@ -257,7 +357,32 @@ impl ObjectProvider for TestProvider {
3 => publication.expose(&self.broken).unwrap(),
_ => unreachable!(),
};
Some(operation(&publication))
ProviderResult::Found(operation(&publication))
}
fn execute_operation(
&self,
context: &ExecutionContext,
path: &str,
operation: impl FnOnce(&PublishedObject<Self::Object>) -> OperationAttempt,
) -> ProviderResult<OperationAttempt> {
let result = self.with_object(context, path, operation);
if let Some((resource, guarantee)) = self.provider_failure.get() {
let error = if resource {
HostError::resource_limit("private provider scheduler state", guarantee)
} else {
HostError::internal("private provider transaction path", guarantee)
};
return ProviderResult::Failure(error);
}
if self.confirm_rollback.get() {
match result {
ProviderResult::Found(attempt) => ProviderResult::Found(attempt.confirm_rollback()),
other => other,
}
} else {
result
}
}
fn authorize(&self, _: &str, _: &Document, _: &InterfaceRef, _: &str) -> bool {
@@ -320,12 +445,21 @@ fn registry_returns_atomic_canonical_observation_without_rewriting_descriptor()
assert_eq!(observation.reference, *reference);
assert_eq!(observation.descriptor, ReaderDefinition::descriptor());
assert_eq!(observation.validator, Some(Validator::new(b"schema-v1")));
let response = observation.protocol_response();
let response = observation.protocol_response().unwrap();
assert_eq!(response.interface, reference.as_str());
assert_eq!(response.descriptor, ReaderDefinition::descriptor());
assert_eq!(response.validator.as_deref(), Some(b"schema-v1".as_slice()));
response.descriptor.validate().unwrap();
let mut invalid_observation = registry.fetch_interface(reference.as_str()).unwrap();
let duplicate_operation = invalid_observation.descriptor.operations[0].clone();
invalid_observation
.descriptor
.operations
.push(duplicate_operation);
let invalid_response = invalid_observation.protocol_response().unwrap_err();
assert_eq!(invalid_response.code, ProtocolErrorCode::Internal);
let error = registry.fetch_interface("unknown").unwrap_err();
assert_eq!(error.code, ProtocolErrorCode::InterfaceNotFound);
}
@@ -648,7 +782,7 @@ fn invalid_host_responses_and_uncertain_operation_failures_use_canonical_codes()
BTreeMap::new(),
))
.unwrap_err(),
CallErrorKind::InvalidResult,
CallErrorKind::OperationOutcomeUnknown,
ProtocolErrorCode::OperationOutcomeUnknown,
);
assert_failure(
@@ -662,7 +796,190 @@ fn invalid_host_responses_and_uncertain_operation_failures_use_canonical_codes()
BTreeMap::new(),
))
.unwrap_err(),
CallErrorKind::Implementation,
CallErrorKind::OperationOutcomeUnknown,
ProtocolErrorCode::OperationOutcomeUnknown,
);
}
fn call_broken(provider: &TestProvider, operation: &str) -> CallError {
provider
.call_operation(request(
"/broken",
None,
provider.broken.reference(),
None,
operation,
BTreeMap::new(),
))
.unwrap_err()
}
#[test]
fn typed_implementation_failures_map_only_with_a_non_commit_guarantee() {
let (_, provider, _) = registry_and_provider();
let unknown_internal = call_broken(&provider, "unknown_internal");
assert_failure(
unknown_internal.clone(),
CallErrorKind::OperationOutcomeUnknown,
ProtocolErrorCode::OperationOutcomeUnknown,
);
assert!(!unknown_internal.message().contains("private://"));
assert!(
!unknown_internal
.protocol_error()
.message
.contains("credential")
);
assert_failure(
call_broken(&provider, "safe_internal"),
CallErrorKind::Internal,
ProtocolErrorCode::Internal,
);
assert_failure(
call_broken(&provider, "unknown_resource"),
CallErrorKind::OperationOutcomeUnknown,
ProtocolErrorCode::OperationOutcomeUnknown,
);
assert_failure(
call_broken(&provider, "safe_resource"),
CallErrorKind::ResourceLimitExceeded,
ProtocolErrorCode::ResourceLimitExceeded,
);
}
#[test]
fn post_invocation_result_and_panic_failures_require_provider_rollback() {
for operation in ["broken", "encode_failure", "panic_after_call"] {
let (_, provider, _) = registry_and_provider();
let error = call_broken(&provider, operation);
assert_failure(
error.clone(),
CallErrorKind::OperationOutcomeUnknown,
ProtocolErrorCode::OperationOutcomeUnknown,
);
assert!(!error.message().contains("private://"));
assert!(!error.message().contains("credential"));
provider.confirm_rollback.set(true);
assert_failure(
call_broken(&provider, operation),
CallErrorKind::Internal,
ProtocolErrorCode::Internal,
);
}
}
#[test]
fn provider_failures_after_invocation_use_the_provider_effect_guarantee() {
let (_, provider, _) = registry_and_provider();
provider
.provider_failure
.set(Some((false, EffectGuarantee::Unknown)));
assert_failure(
call_broken(&provider, "safe_internal"),
CallErrorKind::OperationOutcomeUnknown,
ProtocolErrorCode::OperationOutcomeUnknown,
);
provider
.provider_failure
.set(Some((false, EffectGuarantee::NotCommitted)));
let internal = call_broken(&provider, "unknown_internal");
assert_failure(
internal.clone(),
CallErrorKind::Internal,
ProtocolErrorCode::Internal,
);
assert!(!internal.message().contains("transaction"));
provider
.provider_failure
.set(Some((true, EffectGuarantee::Unknown)));
assert_failure(
call_broken(&provider, "safe_resource"),
CallErrorKind::OperationOutcomeUnknown,
ProtocolErrorCode::OperationOutcomeUnknown,
);
provider
.provider_failure
.set(Some((true, EffectGuarantee::NotCommitted)));
assert_failure(
call_broken(&provider, "unknown_resource"),
CallErrorKind::ResourceLimitExceeded,
ProtocolErrorCode::ResourceLimitExceeded,
);
}
#[test]
fn execution_context_is_propagated_and_preflight_prevents_invocation() {
let (_, provider, _) = registry_and_provider();
let deadline = Instant::now() + Duration::from_secs(60);
let context = ExecutionContext::new()
.with_deadline(deadline)
.with_resource_budget(17);
provider
.call_operation_with_context(
&context,
request(
"/open",
None,
provider.reader.reference(),
Some(b"schema-v1"),
"label",
BTreeMap::new(),
),
)
.unwrap();
assert_eq!(provider.observed_deadline.get(), Some(deadline));
assert_eq!(provider.observed_budget.get(), Some(17));
assert!(!provider.observed_cancelled.get());
let calls = provider.calls.get();
let cancellation = CancellationToken::new();
cancellation.cancel();
let cancelled = ExecutionContext::new().with_cancellation(cancellation);
assert_failure(
provider
.call_operation_with_context(
&cancelled,
request(
"/open",
None,
provider.reader.reference(),
Some(b"schema-v1"),
"label",
BTreeMap::new(),
),
)
.unwrap_err(),
CallErrorKind::ResourceLimitExceeded,
ProtocolErrorCode::ResourceLimitExceeded,
);
assert_eq!(provider.calls.get(), calls);
assert!(provider.observed_cancelled.get());
let expired = ExecutionContext::new().with_deadline(Instant::now() - Duration::from_secs(1));
assert_failure(
provider
.call_operation_with_context(
&expired,
request(
"/open",
None,
provider.reader.reference(),
Some(b"schema-v1"),
"label",
BTreeMap::new(),
),
)
.unwrap_err(),
CallErrorKind::ResourceLimitExceeded,
ProtocolErrorCode::ResourceLimitExceeded,
);
assert_eq!(provider.calls.get(), calls);
}