syndicate-lang
/
syndicate-rs
1
0
Fork 0
Code Issues 1 Pull Requests Projects Releases Wiki Activity

Breaking change: much improved error API

This commit is contained in:
Tony Garnock-Jones 2023-01-30 14:25:58 +01:00
parent 3dea29ffe4
commit dbbbc8c1c6
13 changed files with 59 additions and 35 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
syndicate-macros/examples/box-and-client.rs
View File

@ -6,7 +6,7 @@ use syndicate::schemas::dataspace::Observe;
use syndicate::value::NestedValue;
#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
async fn main() -> ActorResult {
syndicate::convenient_logging()?;
Actor::top(None, |t| {
let ds = Cap::new(&t.create(Dataspace::new(None)));

2
syndicate-server/examples/consumer.rs
View File

@ -20,7 +20,7 @@ pub struct Config {
}
#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
async fn main() -> ActorResult {
syndicate::convenient_logging()?;
let config = Config::from_args();
let sturdyref = sturdy::SturdyRef::from_hex(&config.dataspace)?;

2
syndicate-server/examples/pingpong.rs
View File

@ -89,7 +89,7 @@ fn report_latencies(rtt_ns_samples: &Vec<u64>) {
}
#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
async fn main() -> ActorResult {
syndicate::convenient_logging()?;
let config = Config::from_args();
let sturdyref = sturdy::SturdyRef::from_hex(&config.dataspace)?;

2
syndicate-server/examples/producer.rs
View File

@ -21,7 +21,7 @@ pub struct Config {
}
#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
async fn main() -> ActorResult {
syndicate::convenient_logging()?;
let config = Config::from_args();
let sturdyref = sturdy::SturdyRef::from_hex(&config.dataspace)?;

2
syndicate-server/examples/state-consumer.rs
View File

@ -20,7 +20,7 @@ pub struct Config {
}
#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
async fn main() -> ActorResult {
syndicate::convenient_logging()?;
let config = Config::from_args();
let sturdyref = sturdy::SturdyRef::from_hex(&config.dataspace)?;

2
syndicate-server/examples/state-producer.rs
View File

@ -15,7 +15,7 @@ pub struct Config {
}
#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
async fn main() -> ActorResult {
syndicate::convenient_logging()?;
let config = Config::from_args();
let sturdyref = sturdy::SturdyRef::from_hex(&config.dataspace)?;

2
syndicate-server/src/main.rs
View File

@ -59,7 +59,7 @@ struct ServerConfig {
}
#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
async fn main() -> ActorResult {
let config = Arc::new(ServerConfig::from_args());
syndicate::convenient_logging()?;

2
syndicate-server/src/protocol.rs
View File

@ -19,7 +19,7 @@ use tungstenite::Message;
struct ExitListener;
impl Entity<()> for ExitListener {
fn exit_hook(&mut self, _t: &mut Activation, exit_status: &Arc<ActorResult>) {
fn exit_hook(&mut self, _t: &mut Activation, exit_status: &Arc<ExitStatus>) {
tracing::info!(?exit_status, "disconnect");
}
}

36
syndicate/src/actor.rs
View File

@ -82,8 +82,14 @@ pub type FieldId = NonZeroU64;
/// The type of process-unique field observer block IDs.
pub type BlockId = NonZeroU64;
/// Error responses to events must have type `ActorError`.
pub type ActorError = Box<dyn std::error::Error + Sync + Send + 'static>;
/// Responses to events must have type `ActorResult`.
pub type ActorResult = Result<(), Error>;
pub type ActorResult = Result<(), ActorError>;
/// Final exit status of an actor.
pub type ExitStatus = Result<(), Error>;
/// The [`Actor::boot`] method returns an `ActorHandle`, representing
/// the actor's mainloop task.
@ -207,7 +213,7 @@ pub trait Entity<M>: Send {
/// [RunningActor::add_exit_hook].
///
/// The default implementation does nothing.
fn exit_hook(&mut self, turn: &mut Activation, exit_status: &Arc<ActorResult>) {
fn exit_hook(&mut self, turn: &mut Activation, exit_status: &Arc<ExitStatus>) {
}
}
@ -352,7 +358,7 @@ pub enum ActorState {
Terminated {
/// The exit status of the actor: `Ok(())` for normal
/// termination, `Err(_)` for abnormal termination.
exit_status: Arc<ActorResult>,
exit_status: Arc<ExitStatus>,
},
}
@ -368,7 +374,7 @@ pub struct RunningActor {
#[doc(hidden)]
pub fields: HashMap<FieldId, Box<dyn Any + Send>>,
blocks: HashMap<BlockId, (FacetId, Block)>,
exit_hooks: Vec<Box<dyn Send + FnOnce(&mut Activation, &Arc<ActorResult>)>>,
exit_hooks: Vec<Box<dyn Send + FnOnce(&mut Activation, &Arc<ExitStatus>)>>,
#[doc(hidden)]
pub outbound_assertions: OutboundAssertions,
#[doc(hidden)]
@ -665,7 +671,7 @@ impl FacetRef {
match maybe_exit_status {
None => true,
Some(exit_status) => {
g.terminate(exit_status, &self.actor, &account.trace_collector);
g.terminate(exit_status.map_err(|e| e.into()), &self.actor, &account.trace_collector);
false
}
}
@ -1253,7 +1259,7 @@ impl<'activation> Activation<'activation> {
pub fn facet<F: FnOnce(&mut Activation) -> ActorResult>(
&mut self,
boot: F,
) -> Result<FacetId, Error> {
) -> Result<FacetId, ActorError> {
let f = Facet::new(Some(self.facet.facet_id));
self.trace(|t| trace::ActionDescription::FacetStart {
path: t.facet_ids_for(&f).iter().map(|i| trace::FacetId(AnyValue::new(u64::from(*i)))).collect(),
@ -1502,7 +1508,7 @@ impl<'activation> Activation<'activation> {
Ok(())
}
fn repair_dataflow(&mut self) -> Result<bool, Error> {
fn repair_dataflow(&mut self) -> Result<bool, ActorError> {
let mut pass_number = 0;
while !self.state.dataflow.is_clean() {
pass_number += 1;
@ -1688,8 +1694,8 @@ fn send_actions(
t: PendingEventQueue,
) -> ActorResult {
let token_count = t.len();
tx.send(SystemMessage::Turn(caused_by, LoanedItem::new(account, token_count, t)))
.map_err(|_| error("Target actor not running", AnyValue::new(false)))
Ok(tx.send(SystemMessage::Turn(caused_by, LoanedItem::new(account, token_count, t)))
.map_err(|_| error("Target actor not running", AnyValue::new(false)))?)
}
impl std::fmt::Debug for Mailbox {
@ -1836,7 +1842,7 @@ impl Actor {
let root_facet_ref = self.ac_ref.root_facet_ref();
let terminate = |e: Error | {
assert!(!root_facet_ref.activate(&Account::new(None, None), None, |_| Err(e)));
assert!(!root_facet_ref.activate(&Account::new(None, None), None, |_| Err(e)?));
};
if !root_facet_ref.activate(&boot_account, boot_cause, boot) {
@ -1940,7 +1946,7 @@ impl ActorRef {
pub fn exit_status(&self) -> Option<ActorResult> {
self.access(|state| match state {
ActorState::Running(_) => None,
ActorState::Terminated { exit_status } => Some((**exit_status).clone()),
ActorState::Terminated { exit_status } => Some((**exit_status).clone().map_err(|e| e.into())),
})
}
@ -1979,7 +1985,7 @@ impl ActorState {
fn terminate(
&mut self,
exit_status: ActorResult,
exit_status: ExitStatus,
actor: &ActorRef,
trace_collector: &Option<trace::TraceCollector>,
) {
@ -2076,7 +2082,7 @@ impl RunningActor {
fn cleanup(
mut self,
ac_ref: &ActorRef,
exit_status: Arc<ActorResult>,
exit_status: Arc<ExitStatus>,
trace_collector: Option<trace::TraceCollector>,
) {
if exit_status.is_ok() {
@ -2250,7 +2256,7 @@ impl Cap {
Self::new(&Arc::new(Ref {
mailbox: Arc::clone(&underlying.mailbox),
facet_id: underlying.facet_id,
target: Mutex::new(Some(Box::new(Guard { underlying: underlying, literals }))),
target: Mutex::new(Some(Box::new(Guard { underlying, literals }))),
}))
}
@ -2378,7 +2384,7 @@ where
fn stop(&mut self, t: &mut Activation) -> ActorResult {
t.with_entity(&self.underlying, |t, e| e.stop(t))
}
fn exit_hook(&mut self, t: &mut Activation, exit_status: &Arc<ActorResult>) {
fn exit_hook(&mut self, t: &mut Activation, exit_status: &Arc<ExitStatus>) {
self.underlying.internal_with_entity(|e| e.exit_hook(t, exit_status))
}
}

14
syndicate/src/during.rs
View File

@ -17,7 +17,7 @@ where
Fa: 'static + Send + FnMut(&mut E, &mut Activation, M) -> DuringResult<E>,
Fm: 'static + Send + FnMut(&mut E, &mut Activation, M) -> ActorResult,
Fs: 'static + Send + FnMut(&mut E, &mut Activation) -> ActorResult,
Fx: 'static + Send + FnMut(&mut E, &mut Activation, &Arc<ActorResult>),
Fx: 'static + Send + FnMut(&mut E, &mut Activation, &Arc<ExitStatus>),
{
state: E,
assertion_handler: Option<Fa>,
@ -54,7 +54,7 @@ pub fn entity<M: 'static + Send, E>(
fn (&mut E, &mut Activation, M) -> DuringResult<E>,
fn (&mut E, &mut Activation, M) -> ActorResult,
fn (&mut E, &mut Activation) -> ActorResult,
fn (&mut E, &mut Activation, &Arc<ActorResult>)>
fn (&mut E, &mut Activation, &Arc<ExitStatus>)>
where
E: 'static + Send,
{
@ -68,7 +68,7 @@ where
Fa: 'static + Send + FnMut(&mut E, &mut Activation, M) -> DuringResult<E>,
Fm: 'static + Send + FnMut(&mut E, &mut Activation, M) -> ActorResult,
Fs: 'static + Send + FnMut(&mut E, &mut Activation) -> ActorResult,
Fx: 'static + Send + FnMut(&mut E, &mut Activation, &Arc<ActorResult>),
Fx: 'static + Send + FnMut(&mut E, &mut Activation, &Arc<ExitStatus>),
{
pub fn new(
state: E,
@ -154,7 +154,7 @@ where
pub fn on_exit<Fx1>(self, exit_handler: Fx1) -> DuringEntity<M, E, Fa, Fm, Fs, Fx1>
where
Fx1: 'static + Send + FnMut(&mut E, &mut Activation, &Arc<ActorResult>),
Fx1: 'static + Send + FnMut(&mut E, &mut Activation, &Arc<ExitStatus>),
{
DuringEntity {
state: self.state,
@ -187,7 +187,7 @@ where
Fa: 'static + Send + FnMut(&mut E, &mut Activation, AnyValue) -> DuringResult<E>,
Fm: 'static + Send + FnMut(&mut E, &mut Activation, AnyValue) -> ActorResult,
Fs: 'static + Send + FnMut(&mut E, &mut Activation) -> ActorResult,
Fx: 'static + Send + FnMut(&mut E, &mut Activation, &Arc<ActorResult>),
Fx: 'static + Send + FnMut(&mut E, &mut Activation, &Arc<ExitStatus>),
{
pub fn create_cap(self, t: &mut Activation) -> Arc<Cap>
{
@ -202,7 +202,7 @@ where
Fa: 'static + Send + FnMut(&mut E, &mut Activation, M) -> DuringResult<E>,
Fm: 'static + Send + FnMut(&mut E, &mut Activation, M) -> ActorResult,
Fs: 'static + Send + FnMut(&mut E, &mut Activation) -> ActorResult,
Fx: 'static + Send + FnMut(&mut E, &mut Activation, &Arc<ActorResult>),
Fx: 'static + Send + FnMut(&mut E, &mut Activation, &Arc<ExitStatus>),
{
fn assert(&mut self, t: &mut Activation, a: M, h: Handle) -> ActorResult {
match &mut self.assertion_handler {
@ -232,7 +232,7 @@ where
}
}
fn exit_hook(&mut self, t: &mut Activation, exit_status: &Arc<ActorResult>) {
fn exit_hook(&mut self, t: &mut Activation, exit_status: &Arc<ExitStatus>) {
if let Some(handler) = &mut self.exit_handler {
handler(&mut self.state, t, exit_status);
}

18
syndicate/src/error.rs
View File

@ -73,3 +73,21 @@ impl From<preserves::error::Error> for Error {
error(&format!("{}", v), AnyValue::new(false))
}
}
impl From<Box<dyn std::error::Error>> for Error {
fn from(v: Box<dyn std::error::Error>) -> Self {
match v.downcast::<Error>() {
Ok(e) => *e,
Err(v) => error(&format!("{}", v), AnyValue::new(false)),
}
}
}
impl From<Box<dyn std::error::Error + Send + Sync + 'static>> for Error {
fn from(v: Box<dyn std::error::Error + Send + Sync + 'static>) -> Self {
match v.downcast::<Error>() {
Ok(e) => *e,
Err(v) => error(&format!("{}", v), AnyValue::new(false)),
}
}
}

2
syndicate/src/lib.rs
View File

@ -38,7 +38,7 @@ pub use during::entity;
/// Sets up [`tracing`] logging in a reasonable way.
///
/// Useful at the top of `main` functions.
pub fn convenient_logging() -> Result<(), Box<dyn std::error::Error>> {
pub fn convenient_logging() -> actor::ActorResult {
let filter = match std::env::var(tracing_subscriber::filter::EnvFilter::DEFAULT_ENV) {
Err(std::env::VarError::NotPresent) =>
tracing_subscriber::filter::EnvFilter::default()

8
syndicate/src/relay.rs
View File

@ -313,7 +313,7 @@ impl TunnelRelay {
tracing::info!(message = ?b.message.clone(),
detail = ?b.detail.clone(),
"received Error from peer");
Err(*b)
Err(*b)?
}
P::Packet::Turn(b) => {
let P::Turn(events) = *b;
@ -338,7 +338,7 @@ impl TunnelRelay {
&mut |r| Ok(pins.push(self.membranes.lookup_ref(r))))?;
if let Some(local_handle) = target.assert(t, &(), &a) {
if let Some(_) = self.inbound_assertions.insert(remote_handle, (local_handle, pins)) {
return Err(error("Assertion with duplicate handle", AnyValue::new(false)));
return Err(error("Assertion with duplicate handle", AnyValue::new(false)))?;
}
} else {
self.membranes.release(pins);
@ -348,7 +348,7 @@ impl TunnelRelay {
P::Event::Retract(b) => {
let P::Retract { handle: remote_handle } = *b;
let (local_handle, previous_pins) = match self.inbound_assertions.remove(&remote_handle) {
None => return Err(error("Retraction of nonexistent handle", language().unparse(&remote_handle))),
None => return Err(error("Retraction of nonexistent handle", language().unparse(&remote_handle)))?,
Some(wss) => wss,
};
self.membranes.release(previous_pins);
@ -715,7 +715,7 @@ async fn output_loop(
}
impl Entity<()> for TunnelRefEntity {
fn exit_hook(&mut self, t: &mut Activation, exit_status: &Arc<ActorResult>) {
fn exit_hook(&mut self, t: &mut Activation, exit_status: &Arc<ExitStatus>) {
if let Err(e) = &**exit_status {
let e = e.clone();
let mut g = self.relay_ref.lock();