A few days' work redoing syndicate-rs - still WIP

.gitignore

@@ -1,3 +1,4 @@
 /target
 **/*.rs.bk
 scratch/
+src/gen/**/*.rs

Cargo.toml

@@ -1,11 +1,12 @@
 [package]
 name = "syndicate-rs"
-version = "0.1.0"
+version = "0.2.0"
 authors = ["Tony Garnock-Jones <tonyg@leastfixedpoint.com>"]
 edition = "2018"
 
 [patch.crates-io]
 preserves = { path = "/home/tonyg/src/preserves/implementations/rust/preserves" }
+preserves-schema = { path = "/home/tonyg/src/preserves/implementations/rust/preserves-schema" }
 
 [features]
 vendored-openssl = ["openssl/vendored"]
@@ -17,22 +18,26 @@ lto = true
 [lib]
 name = "syndicate"
 
+[build-dependencies]
+preserves-schema = "0.2.0"
+
 [dependencies]
-preserves = "0.13.0"
+preserves = "0.15.0"
+preserves-schema = "0.2.0"
 
 serde = { version = "1.0", features = ["derive", "rc"] }
 serde_bytes = "0.11"
 
-tokio = { version = "0.2.21", features = ["macros", "rt-threaded", "sync", "dns", "tcp", "time", "stream"] }
-tokio-util = { version = "0.3.1", features = ["codec"] }
-bytes = "0.5.4"
+tokio = { version = "1.7.1", features = ["macros", "sync", "net", "rt", "rt-multi-thread", "time"] }
+tokio-util = { version = "0.6.7", features = ["codec"] }
+bytes = "1.0.1"
 
 futures = "0.3.5"
 
 structopt = "0.3.14"
 
-tungstenite = "0.10.1"
-tokio-tungstenite = "0.10.1"
+tungstenite = "0.13.0"
+tokio-tungstenite = "0.14.0"
 
 tracing = "0.1.14"
 tracing-subscriber = "0.2.5"

src/.gitignore

@@ -0,0 +1 @@
+schemas/

src/actor.rs

@@ -0,0 +1,389 @@
+use super::Assertion;
+use super::ActorId;
+use super::Handle;
+use super::schemas::internal_protocol::*;
+use super::error::Error;
+
+use preserves::value::Domain;
+use preserves::value::IOResult;
+use preserves::value::IOValue;
+use preserves::value::Map;
+use preserves::value::NestedValue;
+
+use std::boxed::Box;
+use std::cell::Cell;
+use std::collections::hash_map::HashMap;
+use std::future::Future;
+use std::future::ready;
+use std::sync::Arc;
+use std::sync::atomic::{AtomicUsize, Ordering};
+
+use tokio::select;
+use tokio::sync::mpsc::{unbounded_channel, UnboundedSender, UnboundedReceiver};
+use tokio_util::sync::CancellationToken;
+
+use tracing::{Instrument, trace, error};
+
+pub type ActorResult = Result<(), Error>;
+pub type ActorHandle = tokio::task::JoinHandle<ActorResult>;
+
+pub trait Entity {
+    fn assert(&mut self, _t: &mut Activation, _a: Assertion, _h: Handle) -> ActorResult {
+        Ok(())
+    }
+    fn retract(&mut self, _t: &mut Activation, _h: Handle) -> ActorResult {
+        Ok(())
+    }
+    fn message(&mut self, _t: &mut Activation, _m: Assertion) -> ActorResult {
+        Ok(())
+    }
+    fn sync(&mut self, t: &mut Activation, peer: Arc<Ref>) -> ActorResult {
+        t.message(peer, Assertion::new(true));
+        Ok(())
+    }
+}
+
+type OutboundAssertions = Map<Handle, Arc<Ref>>;
+
+// This is what other implementations call a "Turn", renamed here to
+// avoid conflicts with schemas::internal_protocol::Turn.
+pub struct Activation<'activation> {
+    outbound_assertions: &'activation mut OutboundAssertions,
+    queues: HashMap<ActorId, Vec<(Arc<Ref>, Event)>>,
+}
+
+enum SystemMessage {
+    Release,
+    ReleaseOid(Oid),
+    Turn(Turn),
+    Crash(Error),
+}
+
+pub struct Mailbox {
+    pub actor_id: ActorId,
+    pub mailbox_id: u64,
+    tx: UnboundedSender<SystemMessage>,
+    pub queue_depth: Arc<AtomicUsize>,
+    pub mailbox_count: Arc<AtomicUsize>,
+}
+
+pub struct Actor {
+    pub template_mailbox: Mailbox,
+    rx: UnboundedReceiver<SystemMessage>,
+    pub outbound_assertions: OutboundAssertions,
+    pub oid_map: Map<Oid, Cell<Box<dyn Entity + Send>>>,
+    pub next_task_id: u64,
+    pub linked_tasks: Map<u64, CancellationToken>,
+}
+
+#[derive(Debug, PartialEq, Eq, Hash, PartialOrd, Ord)]
+pub struct Ref {
+    pub relay: Mailbox,
+    pub target: Oid,
+    /* TODO: attenuation */
+}
+
+//---------------------------------------------------------------------------
+
+impl<'activation> Activation<'activation> {
+    pub fn for_actor(actor: &'activation mut Actor) -> Self {
+        Self::for_actor_details(&mut actor.outbound_assertions)
+    }
+
+    pub fn for_actor_details(outbound_assertions: &'activation mut OutboundAssertions) -> Self {
+        Activation {
+            outbound_assertions,
+            queues: HashMap::new(),
+        }
+    }
+
+    pub fn assert<M>(&mut self, r: Arc<Ref>, a: M) -> Handle where M: Into<Assertion> {
+        let handle = crate::next_handle();
+        self.queue_for(&r).push((Arc::clone(&r), Event::Assert(Box::new(
+            Assert { assertion: Assertion(a.into()), handle: handle.clone() }))));
+        self.outbound_assertions.insert(handle.clone(), r);
+        handle
+    }
+
+    pub fn retract(&mut self, handle: Handle) {
+        if let Some(r) = self.outbound_assertions.remove(&handle) {
+            self.retract_known_ref(r, handle)
+        }
+    }
+
+    pub fn retract_known_ref(&mut self, r: Arc<Ref>, handle: Handle) {
+        self.queue_for(&r).push((r, Event::Retract(Box::new(Retract { handle }))));
+    }
+
+    pub fn message<M>(&mut self, r: Arc<Ref>, m: M) where M: Into<Assertion> {
+        self.queue_for(&r).push((r, Event::Message(Box::new(
+            Message { body: Assertion(m.into()) }))))
+    }
+
+    fn queue_for(&mut self, r: &Arc<Ref>) -> &mut Vec<(Arc<Ref>, Event)> {
+        self.queues.entry(r.relay.actor_id).or_default()
+    }
+
+    pub fn deliver(&mut self) {
+        for (_actor_id, turn) in std::mem::take(&mut self.queues).into_iter() {
+            if turn.len() == 0 { continue; }
+            let first_ref = Arc::clone(&turn[0].0);
+            let target = &first_ref.relay;
+            target.send(Turn(turn.into_iter().map(
+                |(r, e)| TurnEvent { oid: r.target.clone(), event: e }).collect()));
+        }
+    }
+}
+
+impl<'activation> Drop for Activation<'activation> {
+    fn drop(&mut self) {
+        self.deliver()
+    }
+}
+
+impl Mailbox {
+    pub fn send(&self, t: Turn) {
+        let _ = self.tx.send(SystemMessage::Turn(t));
+        self.queue_depth.fetch_add(1, Ordering::Relaxed);
+    }
+}
+
+impl std::fmt::Debug for Mailbox {
+    fn fmt(&self, f: &mut std::fmt::Formatter) -> Result<(), std::fmt::Error> {
+        write!(f, "#<Mailbox {}:{}>", self.actor_id, self.mailbox_id)
+    }
+}
+
+impl std::hash::Hash for Mailbox {
+    fn hash<H: std::hash::Hasher>(&self, state: &mut H) {
+        self.mailbox_id.hash(state)
+    }
+}
+
+impl Eq for Mailbox {}
+impl PartialEq for Mailbox {
+    fn eq(&self, other: &Mailbox) -> bool {
+        self.mailbox_id == other.mailbox_id
+    }
+}
+
+impl Ord for Mailbox {
+    fn cmp(&self, other: &Mailbox) -> std::cmp::Ordering {
+        return self.mailbox_id.cmp(&other.mailbox_id)
+    }
+}
+
+impl PartialOrd for Mailbox {
+    fn partial_cmp(&self, other: &Mailbox) -> Option<std::cmp::Ordering> {
+        return Some(self.cmp(&other))
+    }
+}
+
+impl Clone for Mailbox {
+    fn clone(&self) -> Self {
+        let Mailbox { actor_id, tx, queue_depth, mailbox_count, .. } = self;
+        mailbox_count.fetch_add(1, Ordering::SeqCst);
+        Mailbox {
+            actor_id: *actor_id,
+            mailbox_id: crate::next_mailbox_id(),
+            tx: tx.clone(),
+            queue_depth: Arc::clone(queue_depth),
+            mailbox_count: Arc::clone(mailbox_count),
+        }
+    }
+}
+
+impl Drop for Mailbox {
+    fn drop(&mut self) {
+        if self.mailbox_count.fetch_sub(1, Ordering::SeqCst) == 1 {
+            let _ = self.tx.send(SystemMessage::Release);
+            ()
+        }
+    }
+}
+
+impl Actor {
+    pub fn new() -> Self {
+        let (tx, rx) = unbounded_channel();
+        Actor {
+            template_mailbox: Mailbox {
+                actor_id: crate::next_actor_id(),
+                mailbox_id: crate::next_mailbox_id(),
+                tx,
+                queue_depth: Arc::new(AtomicUsize::new(0)),
+                mailbox_count: Arc::new(AtomicUsize::new(0)),
+            },
+            rx,
+            outbound_assertions: Map::new(),
+            oid_map: Map::new(),
+            next_task_id: 0,
+            linked_tasks: Map::new(),
+        }
+    }
+
+    pub fn id(&self) -> ActorId {
+        self.template_mailbox.actor_id
+    }
+
+    pub fn create<E: Entity + Send + 'static>(&mut self, e: E) -> Arc<Ref> {
+        let r = Ref {
+            relay: self.template_mailbox.clone(),
+            target: crate::next_oid(),
+        };
+        self.oid_map.insert(r.target.clone(), Cell::new(Box::new(e)));
+        Arc::new(r)
+    }
+
+    pub fn boot<F: Future<Output = ActorResult> + Send + 'static>(
+        mut self,
+        name: tracing::Span,
+        boot: F,
+    ) -> ActorHandle {
+        tokio::spawn(async move {
+            trace!("start");
+            let run_future = self.run(boot);
+            let result = run_future.await;
+            match &result {
+                Ok(()) => trace!("normal stop"),
+                Err(e) => error!("{}", e),
+            }
+            result
+        }.instrument(name))
+    }
+
+    pub fn start(self, name: tracing::Span) -> ActorHandle {
+        self.boot(name, ready(Ok(())))
+    }
+
+    async fn run<F: Future<Output = ActorResult>>(&mut self, boot: F) -> ActorResult {
+        boot.await?;
+        loop {
+            match self.rx.recv().await {
+                None =>
+                    Err(Error {
+                        message: "Unexpected channel close".to_owned(),
+                        detail: _Any::new(false),
+                    })?,
+                Some(m) => {
+                    if self.handle(m)? {
+                        return Ok(());
+                    }
+                    // We would have a loop calling try_recv until it answers "no more at
+                    // present" here, to avoid decrementing queue_depth for every message
+                    // (instead zeroing it on queue empty - it only needs to be approximate),
+                    // but try_recv has been removed from mpsc at the time of writing. See
+                    // https://github.com/tokio-rs/tokio/issues/3350 .
+                    self.template_mailbox.queue_depth.fetch_sub(1, Ordering::Relaxed);
+                }
+            }
+        }
+    }
+
+    fn handle(&mut self, m: SystemMessage) -> Result<bool, Error> {
+        match m {
+            SystemMessage::Release =>
+                Ok(true),
+            SystemMessage::ReleaseOid(oid) => {
+                self.oid_map.remove(&oid);
+                Ok(false)
+            }
+            SystemMessage::Turn(Turn(events)) => {
+                for TurnEvent { oid, event } in events.into_iter() {
+                    if let Some(e) = self.oid_map.get_mut(&oid) {
+                        let mut t = Activation::for_actor_details(&mut self.outbound_assertions);
+                        let e = e.get_mut();
+                        match event {
+                            Event::Assert(b) => {
+                                let Assert { assertion: Assertion(assertion), handle } = *b;
+                                e.assert(&mut t, assertion, handle)?;
+                            }
+                            Event::Retract(b) => {
+                                let Retract { handle } = *b;
+                                e.retract(&mut t, handle)?;
+                            }
+                            Event::Message(b) => {
+                                let Message { body: Assertion(body) } = *b;
+                                e.message(&mut t, body)?;
+                            }
+                            Event::Sync(b) => {
+                                let Sync { peer } = *b;
+                                e.sync(&mut t, peer)?;
+                            }
+                        }
+                    }
+                }
+                Ok(false)
+            }
+            SystemMessage::Crash(e) =>
+                Err(e)?
+        }
+    }
+
+    pub fn linked_task<F: Future<Output = ActorResult> + Send + 'static>(
+        &mut self,
+        name: tracing::Span,
+        boot: F,
+    ) {
+        let mailbox = self.template_mailbox.clone();
+        let token = CancellationToken::new();
+        let task_id = self.next_task_id;
+        self.next_task_id += 1;
+        {
+            let token = token.clone();
+            tokio::spawn(async move {
+                trace!("linked task start");
+                select! {
+                    _ = token.cancelled() => (),
+                    result = boot => match result {
+                        Ok(()) => trace!("linked task normal stop"),
+                        Err(e) => {
+                            error!("linked task error: {}", e);
+                            let _ = mailbox.tx.send(SystemMessage::Crash(e));
+                            ()
+                        }
+                    }
+                }
+            }.instrument(name));
+        }
+        self.linked_tasks.insert(task_id, token);
+    }
+}
+
+impl Drop for Actor {
+    fn drop(&mut self) {
+        for (_task_id, token) in std::mem::take(&mut self.linked_tasks).into_iter() {
+            token.cancel();
+        }
+
+        let to_clear = std::mem::take(&mut self.outbound_assertions);
+        let mut t = Activation::for_actor(self);
+        for (handle, r) in to_clear.into_iter() {
+            t.retract_known_ref(r, handle);
+        }
+    }
+}
+
+impl Drop for Ref {
+    fn drop(&mut self) {
+        let _ = self.relay.tx.send(SystemMessage::ReleaseOid(self.target.clone()));
+        ()
+    }
+}
+
+impl Domain for Ref {
+    fn from_preserves(v: IOValue) -> IOResult<Self> {
+        panic!("aiee")
+    }
+    fn as_preserves(&self) -> IOValue {
+        panic!("aiee")
+    }
+}
+
+impl Domain for super::schemas::sturdy::WireRef {
+    fn from_preserves(v: IOValue) -> IOResult<Self> {
+        panic!("aiee")
+    }
+    fn as_preserves(&self) -> IOValue {
+        panic!("aiee")
+    }
+}

src/bin/syndicate-server.rs

@@ -1,133 +1,33 @@
-use syndicate::{config, spaces, packets, ConnId};
-use syndicate::peer::Peer;
-
-use std::sync::{Mutex, Arc};
 use futures::{SinkExt, StreamExt};
 
-use tracing::{Level, error, info, trace};
-use tracing_futures::Instrument;
+use preserves::value::PackedReader;
+use preserves::value::PackedWriter;
+use preserves::value::Reader;
+use preserves::value::Writer;
+
+use std::convert::TryFrom;
+use std::future::Ready;
+use std::future::ready;
+use std::sync::Arc;
+
+use structopt::StructOpt; // for from_args in main
+
+use syndicate::actor::*;
+use syndicate::dataspace::*;
+use syndicate::error::Error;
+use syndicate::error::error;
+use syndicate::config;
+use syndicate::packets;
+use syndicate::peer::Peer;
 
 use tokio::net::TcpListener;
 use tokio::net::TcpStream;
 use tokio_util::codec::Framed;
 
+use tracing::{Level, info, trace};
+
 use tungstenite::Message;
 
-use structopt::StructOpt; // for from_args in main
-
-type UnitAsyncResult = Result<(), std::io::Error>;
-
-fn message_error<E: std::fmt::Display>(e: E) -> packets::Error {
-    packets::Error::Message(e.to_string())
-}
-
-fn encode_message(p: packets::S2C) ->
-    Result<Message, packets::Error>
-{
-    let mut bs = Vec::with_capacity(128);
-    preserves::ser::to_writer(&mut preserves::value::PackedWriter::new(&mut bs), &p)?;
-    Ok(Message::Binary(bs))
-}
-
-fn message_encoder(p: packets::S2C) -> futures::future::Ready<Result<Message, packets::Error>>
-{
-    futures::future::ready(encode_message(p))
-}
-
-async fn message_decoder(r: Result<Message, tungstenite::Error>) -> Option<Result<packets::C2S, packets::Error>>
-{
-    match r {
-        Ok(ref m) => match m {
-            Message::Text(_) =>
-                Some(Err(preserves::error::syntax_error("Text websocket frames are not accepted"))),
-            Message::Binary(ref bs) =>
-                match preserves::de::from_bytes(bs) {
-                    Ok(p) => Some(Ok(p)),
-                    Err(e) => Some(Err(e.into())),
-                },
-            Message::Ping(_) =>
-                None, // pings are handled by tungstenite before we see them
-            Message::Pong(_) =>
-                None, // unsolicited pongs are to be ignored
-            Message::Close(_) =>
-                Some(Err(preserves::error::eof())),
-        }
-        Err(tungstenite::Error::Io(e)) =>
-            Some(Err(e.into())),
-        Err(e) =>
-            Some(Err(message_error(e))),
-    }
-}
-
-async fn run_connection(connid: ConnId,
-                        mut stream: TcpStream,
-                        spaces: Arc<Mutex<spaces::Spaces>>,
-                        addr: std::net::SocketAddr,
-                        config: config::ServerConfigRef) ->
-    UnitAsyncResult
-{
-    let mut buf = [0; 1]; // peek at the first byte to see what kind of connection to expect
-    match stream.peek(&mut buf).await? {
-        1 => match buf[0] {
-            71 /* ASCII 'G' for "GET" */ => {
-                info!(protocol = display("websocket"), peer = debug(addr));
-                let s = tokio_tungstenite::accept_async(stream).await
-                    .map_err(|e| std::io::Error::new(std::io::ErrorKind::Other, e))?;
-                let (o, i) = s.split();
-                let i = i.filter_map(message_decoder);
-                let o = o.sink_map_err(message_error).with(message_encoder);
-                let mut p = Peer::new(connid, i, o);
-                p.run(spaces, &config).await?
-            },
-            _ => {
-                info!(protocol = display("raw"), peer = debug(addr));
-                let (o, i) = Framed::new(stream, packets::Codec::new()).split();
-                let mut p = Peer::new(connid, i, o);
-                p.run(spaces, &config).await?
-            }
-        }
-        0 => return Err(std::io::Error::new(std::io::ErrorKind::UnexpectedEof,
-                                            "closed before starting")),
-        _ => unreachable!()
-    }
-    Ok(())
-}
-
-static NEXT_ID: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(1);
-
-async fn run_listener(spaces: Arc<Mutex<spaces::Spaces>>, port: u16, config: config::ServerConfigRef) ->
-    UnitAsyncResult
-{
-    let mut listener = TcpListener::bind(format!("0.0.0.0:{}", port)).await?;
-    loop {
-        let (stream, addr) = listener.accept().await?;
-        let id = NEXT_ID.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
-        let spaces = Arc::clone(&spaces);
-        let config = Arc::clone(&config);
-        if let Some(n) = config.recv_buffer_size { stream.set_recv_buffer_size(n)?; }
-        if let Some(n) = config.send_buffer_size { stream.set_send_buffer_size(n)?; }
-        tokio::spawn(async move {
-            match run_connection(id, stream, spaces, addr, config).await {
-                Ok(()) => info!("closed"),
-                Err(e) => info!(error = display(e), "closed"),
-            }
-        }.instrument(tracing::info_span!("connection", id)));
-    }
-}
-
-async fn periodic_tasks(spaces: Arc<Mutex<spaces::Spaces>>) -> UnitAsyncResult {
-    let interval = core::time::Duration::from_secs(10);
-    let mut delay = tokio::time::interval(interval);
-    loop {
-        delay.next().await.unwrap();
-        {
-            let mut spaces = spaces.lock().unwrap();
-            spaces.cleanup();
-            spaces.dump_stats(interval);
-        }
-    }
-}
-
 #[tokio::main]
 async fn main() -> Result<(), Box<dyn std::error::Error>> {
     let filter = tracing_subscriber::filter::EnvFilter::from_default_env()
@@ -175,30 +75,112 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
 
     let config = Arc::new(config::ServerConfig::from_args());
 
-    let spaces = Arc::new(Mutex::new(spaces::Spaces::new()));
     let mut daemons = Vec::new();
 
-    {
-        let spaces = Arc::clone(&spaces);
-        tokio::spawn(async move {
-            periodic_tasks(spaces).await
-        });
-    }
-
     trace!("startup");
 
+    let ds = {
+        let ac = Actor::new();
+        let ds = ac.create(Dataspace::new());
+        daemons.push(ac.start(tracing::info_span!("dataspace")));
+        ds
+    };
+
     for port in config.ports.clone() {
-        let spaces = Arc::clone(&spaces);
+        let ds = Arc::clone(&ds);
         let config = Arc::clone(&config);
-        daemons.push(tokio::spawn(async move {
-            info!(port, "listening");
-            match run_listener(spaces, port, config).await {
-                Ok(()) => (),
-                Err(e) => error!("{}", e),
-            }
-        }.instrument(tracing::info_span!("listener", port))));
+        let ac = Actor::new();
+        ac.linked_task(tracing::info_span!("listener", port), run_listener(ds, port, config));
     }
 
     futures::future::join_all(daemons).await;
     Ok(())
 }
+
+//---------------------------------------------------------------------------
+
+fn message_error<E: std::fmt::Display>(e: E) -> Error {
+    error(&e.to_string(), false)
+}
+
+fn encode_message(p: packets::Packet) -> Result<Message, Error> {
+    let mut bs = Vec::with_capacity(128);
+    PackedWriter::new(&mut bs).write(&(&p).into())?;
+    Ok(Message::Binary(bs))
+}
+
+fn message_encoder(p: packets::Packet) -> Ready<Result<Message, Error>>
+{
+    ready(encode_message(p))
+}
+
+fn message_decoder_inner(
+    r: Result<Message, tungstenite::Error>,
+) -> Result<Option<packets::Packet>, Error> {
+    match r {
+        Ok(m) => match m {
+            Message::Text(_) =>
+                Err("Text websocket frames are not accepted")?,
+            Message::Binary(bs) => {
+                let iov = PackedReader::decode_bytes(&bs).demand_next(false)?;
+                let p = packets::Packet::try_from(&iov)?;
+                Ok(Some(p))
+            }
+            Message::Ping(_) =>
+                Ok(None), // pings are handled by tungstenite before we see them
+            Message::Pong(_) =>
+                Ok(None), // unsolicited pongs are to be ignored
+            Message::Close(_) =>
+                Err("EOF")?,
+        },
+        Err(e) => Err(message_error(e)),
+    }
+}
+
+fn message_decoder(r: Result<Message, tungstenite::Error>) -> Ready<Option<Result<packets::Packet, Error>>> {
+    ready(message_decoder_inner(r).transpose())
+}
+
+async fn run_connection(
+    mut stream: TcpStream,
+    ds: Arc<Ref>,
+    addr: std::net::SocketAddr,
+    config: Arc<config::ServerConfig>,
+) -> ActorResult {
+    let mut buf = [0; 1]; // peek at the first byte to see what kind of connection to expect
+    match stream.peek(&mut buf).await? {
+        1 => match buf[0] {
+            71 /* ASCII 'G' for "GET" */ => {
+                info!(protocol = display("websocket"), peer = debug(addr));
+                let s = tokio_tungstenite::accept_async(stream).await
+                    .map_err(|e| std::io::Error::new(std::io::ErrorKind::Other, e))?;
+                let (o, i) = s.split();
+                let i = i.filter_map(message_decoder);
+                let o = o.sink_map_err(message_error).with(message_encoder);
+                let mut p = Peer::new(i, o, ds, config);
+                p.run().await?
+            },
+            _ => {
+                info!(protocol = display("raw"), peer = debug(addr));
+                let (o, i) = Framed::new(stream, packets::Codec).split();
+                let mut p = Peer::new(i, o, ds, config);
+                p.run().await?
+            }
+        }
+        0 => Err(error("closed before starting", false))?,
+        _ => unreachable!()
+    }
+    Ok(())
+}
+
+async fn run_listener(ds: Arc<Ref>, port: u16, config: Arc<config::ServerConfig>) -> ActorResult {
+    let listener = TcpListener::bind(format!("0.0.0.0:{}", port)).await?;
+    loop {
+        let (stream, addr) = listener.accept().await?;
+        let mut ac = Actor::new();
+        let ds = Arc::clone(&ds);
+        let config = Arc::clone(&config);
+        ac.linked_task(tracing::info_span!("connection", id = (ac.id())),
+                       run_connection(stream, ds, addr, config));
+    }
+}

src/config.rs

@@ -5,15 +5,8 @@ pub struct ServerConfig {
     #[structopt(short = "p", long = "port", default_value = "8001")]
     pub ports: Vec<u16>,
 
-    #[structopt(long)]
-    pub recv_buffer_size: Option<usize>,
-    #[structopt(long)]
-    pub send_buffer_size: Option<usize>,
-
     #[structopt(long, default_value = "10000")]
     pub overload_threshold: usize,
     #[structopt(long, default_value = "5")]
     pub overload_turn_limit: usize,
 }
-
-pub type ServerConfigRef = std::sync::Arc<ServerConfig>;

src/dataspace.rs

@@ -1,36 +1,21 @@
-use super::V;
-use super::ConnId;
-use super::packets::{self, Assertion, EndpointName};
+use super::Assertion;
+use super::Handle;
 use super::skeleton;
+use super::actor::*;
+use super::schemas::dataspace::*;
 
-use preserves::value::{self, Map, NestedValue};
-use std::sync::{Arc, RwLock, atomic::{AtomicUsize, Ordering}};
-use tokio::sync::mpsc::UnboundedSender;
+use preserves::value::Map;
 
-pub type DataspaceRef = Arc<RwLock<Dataspace>>;
-pub type DataspaceError = (String, V);
-
-#[derive(Debug)]
-struct Actor {
-    tx: UnboundedSender<packets::S2C>,
-    queue_depth: Arc<AtomicUsize>,
-    endpoints: Map<EndpointName, ActorEndpoint>,
-}
-
-#[derive(Debug)]
-struct ActorEndpoint {
-    analysis_results: Option<skeleton::AnalysisResults>,
-    assertion: Assertion,
-}
+use std::convert::TryFrom;
 
 #[derive(Debug)]
 pub struct Churn {
-    pub peers_added: usize,
-    pub peers_removed: usize,
     pub assertions_added: usize,
     pub assertions_removed: usize,
     pub endpoints_added: usize,
     pub endpoints_removed: usize,
+    pub observers_added: usize,
+    pub observers_removed: usize,
     pub messages_injected: usize,
     pub messages_delivered: usize,
 }
@@ -38,167 +23,38 @@ pub struct Churn {
 impl Churn {
     pub fn new() -> Self {
         Self {
-            peers_added: 0,
-            peers_removed: 0,
             assertions_added: 0,
             assertions_removed: 0,
             endpoints_added: 0,
             endpoints_removed: 0,
+            observers_added: 0,
+            observers_removed: 0,
             messages_injected: 0,
             messages_delivered: 0,
         }
     }
 
     pub fn reset(&mut self) {
-        self.peers_added = 0;
-        self.peers_removed = 0;
-        self.assertions_added = 0;
-        self.assertions_removed = 0;
-        self.endpoints_added = 0;
-        self.endpoints_removed = 0;
-        self.messages_injected = 0;
-        self.messages_delivered = 0;
+        *self = Churn::new()
     }
 }
 
 #[derive(Debug)]
 pub struct Dataspace {
-    name: V,
-    peers: Map<ConnId, Actor>,
-    index: skeleton::Index,
+    pub index: skeleton::Index,
+    pub handle_map: Map<Handle, (Assertion, Option<Observe>)>,
     pub churn: Churn,
 }
 
 impl Dataspace {
-    pub fn new(name: &V) -> Self {
+    pub fn new() -> Self {
         Self {
-            name: name.clone(),
-            peers: Map::new(),
             index: skeleton::Index::new(),
+            handle_map: Map::new(),
             churn: Churn::new(),
         }
     }
 
-    pub fn new_ref(name: &V) -> DataspaceRef {
-        Arc::new(RwLock::new(Self::new(name)))
-    }
-
-    pub fn register(&mut self, id: ConnId,
-                    tx: UnboundedSender<packets::S2C>,
-                    queue_depth: Arc<AtomicUsize>)
-    {
-        assert!(!self.peers.contains_key(&id));
-        self.peers.insert(id, Actor {
-            tx,
-            queue_depth,
-            endpoints: Map::new(),
-        });
-        self.churn.peers_added += 1;
-    }
-
-    pub fn deregister(&mut self, id: ConnId) {
-        let ac = self.peers.remove(&id).unwrap();
-        self.churn.peers_removed += 1;
-        let mut outbound_turns: Map<ConnId, Vec<packets::Event>> = Map::new();
-        for (epname, ep) in ac.endpoints {
-            self.remove_endpoint(&mut outbound_turns, id, &epname, ep);
-        }
-        outbound_turns.remove(&id);
-        self.deliver_outbound_turns(outbound_turns);
-    }
-
-    fn remove_endpoint(&mut self,
-                       mut outbound_turns: &mut Map<ConnId, Vec<packets::Event>>,
-                       id: ConnId,
-                       epname: &EndpointName,
-                       ep: ActorEndpoint)
-    {
-        let ActorEndpoint{ analysis_results, assertion } = ep;
-        if let Some(ar) = analysis_results {
-            self.index.remove_endpoint(&ar, skeleton::Endpoint {
-                connection: id,
-                name: epname.clone(),
-            });
-        }
-        let old_assertions = self.index.assertion_count();
-        self.index.remove((&assertion).into(), &mut outbound_turns);
-        self.churn.assertions_removed += old_assertions - self.index.assertion_count();
-        self.churn.endpoints_removed += 1;
-    }
-
-    pub fn turn(&mut self, id: ConnId, actions: Vec<packets::Action>) ->
-        Result<(), DataspaceError>
-    {
-        let mut outbound_turns: Map<ConnId, Vec<packets::Event>> = Map::new();
-        for a in actions {
-            tracing::trace!(action = debug(&a), "turn");
-            match a {
-                packets::Action::Assert(ref epname, ref assertion) => {
-                    let ac = self.peers.get_mut(&id).unwrap();
-                    if ac.endpoints.contains_key(&epname) {
-                        return Err(("Duplicate endpoint name".to_string(), value::to_value(a)));
-                    }
-
-                    let ar =
-                        if let Some(fs) = assertion.value().as_simple_record("observe", Some(1)) {
-                            let ar = skeleton::analyze(&fs[0]);
-                            let events = self.index.add_endpoint(&ar, skeleton::Endpoint {
-                                connection: id,
-                                name: epname.clone(),
-                            });
-                            outbound_turns.entry(id).or_insert_with(Vec::new).extend(events);
-                            Some(ar)
-                        } else {
-                            None
-                        };
-
-                    let old_assertions = self.index.assertion_count();
-                    self.index.insert(assertion.into(), &mut outbound_turns);
-                    self.churn.assertions_added += self.index.assertion_count() - old_assertions;
-                    self.churn.endpoints_added += 1;
-
-                    ac.endpoints.insert(epname.clone(), ActorEndpoint {
-                        analysis_results: ar,
-                        assertion: assertion.clone()
-                    });
-                }
-                packets::Action::Clear(ref epname) => {
-                    let ac = self.peers.get_mut(&id).unwrap();
-                    match ac.endpoints.remove(epname) {
-                        None => {
-                            return Err(("Nonexistent endpoint name".to_string(), value::to_value(a)));
-                        }
-                        Some(ep) => {
-                            self.remove_endpoint(&mut outbound_turns, id, epname, ep);
-                            outbound_turns.entry(id).or_insert_with(Vec::new)
-                                .push(packets::Event::End(epname.clone()));
-                        }
-                    }
-                }
-                packets::Action::Message(ref assertion) => {
-                    self.index.send(assertion.into(),
-                                    &mut outbound_turns,
-                                    &mut self.churn.messages_delivered);
-                    self.churn.messages_injected += 1;
-                }
-            }
-        }
-        self.deliver_outbound_turns(outbound_turns);
-        Ok(())
-    }
-
-    fn deliver_outbound_turns(&mut self, outbound_turns: Map<ConnId, Vec<packets::Event>>) {
-        for (target, events) in outbound_turns {
-            let actor = self.peers.get_mut(&target).unwrap();
-            let _ = actor.tx.send(packets::S2C::Turn(events));
-            actor.queue_depth.fetch_add(1, Ordering::Relaxed);
-        }
-    }
-
-    pub fn peer_count(&self) -> usize {
-        self.peers.len()
-    }
-
     pub fn assertion_count(&self) -> usize {
         self.index.assertion_count()
     }
@@ -206,4 +62,49 @@ impl Dataspace {
     pub fn endpoint_count(&self) -> isize {
         self.index.endpoint_count()
     }
+
+    pub fn observer_count(&self) -> usize {
+        self.index.observer_count()
+    }
+}
+
+impl Entity for Dataspace {
+    fn assert(&mut self, t: &mut Activation, a: Assertion, h: Handle) -> ActorResult {
+        tracing::trace!(action = debug(&a), "assert");
+
+        let old_assertions = self.index.assertion_count();
+        self.index.insert(t, &a);
+        self.churn.assertions_added += self.index.assertion_count() - old_assertions;
+        self.churn.endpoints_added += 1;
+
+        if let Ok(o) = Observe::try_from(&a) {
+            self.index.add_observer(t, &o.pattern, &o.observer);
+            self.churn.observers_added += 1;
+            self.handle_map.insert(h, (a, Some(o)));
+        } else {
+            self.handle_map.insert(h, (a, None));
+        }
+        Ok(())
+    }
+
+    fn retract(&mut self, t: &mut Activation, h: Handle) -> ActorResult {
+        if let Some((a, maybe_o)) = self.handle_map.remove(&h) {
+            if let Some(o) = maybe_o {
+                self.index.remove_observer(o.pattern, &o.observer);
+                self.churn.observers_removed += 1;
+            }
+
+            let old_assertions = self.index.assertion_count();
+            self.index.remove(t, &a);
+            self.churn.assertions_removed += old_assertions - self.index.assertion_count();
+            self.churn.endpoints_removed += 1;
+        }
+        Ok(())
+    }
+
+    fn message(&mut self, t: &mut Activation, m: Assertion) -> ActorResult {
+        self.index.send(t, &m, &mut self.churn.messages_delivered);
+        self.churn.messages_injected += 1;
+        Ok(())
+    }
 }

src/error.rs

@@ -0,0 +1,41 @@
+pub use super::schemas::internal_protocol::_Any;
+pub use super::schemas::internal_protocol::_Ptr;
+pub use super::schemas::internal_protocol::Error;
+
+use preserves::value::NestedValue;
+use preserves::value::Value;
+use preserves_schema::support::ParseError;
+
+impl std::error::Error for Error {}
+
+impl std::fmt::Display for Error {
+    fn fmt(&self, f: &mut std::fmt::Formatter) -> Result<(), std::fmt::Error> {
+        write!(f, "Error: {}; detail: {:?}", self.message, self.detail)
+    }
+}
+
+pub fn error<Detail>(message: &str, detail: Detail) -> Error where Value<_Any, _Ptr>: From<Detail> {
+    Error {
+        message: message.to_owned(),
+        detail: _Any::new(detail),
+    }
+}
+
+impl From<&str> for Error {
+    fn from(v: &str) -> Self {
+        error(v, false)
+    }
+}
+
+impl From<std::io::Error> for Error {
+    fn from(v: std::io::Error) -> Self {
+        error(&format!("{}", v), false)
+    }
+}
+
+impl From<ParseError> for Error {
+    fn from(v: ParseError) -> Self {
+        error(&format!("{}", v), false)
+    }
+}
+

src/lib.rs

@@ -1,30 +1,45 @@
 #![recursion_limit="512"]
 
-pub mod bag;
-pub mod config;
-pub mod dataspace;
-pub mod packets;
-pub mod peer;
-pub mod skeleton;
-pub mod spaces;
-
 pub use preserves::value;
 
-// use std::sync::atomic::{AtomicUsize, Ordering};
-//
-// #[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
-// pub enum Syndicate {
-//     Placeholder(usize),
-// }
-//
-// impl value::Domain for Syndicate {}
-//
-// static NEXT_PLACEHOLDER: AtomicUsize = AtomicUsize::new(0);
-// impl Syndicate {
-//     pub fn new_placeholder() -> Self {
-//         Self::Placeholder(NEXT_PLACEHOLDER.fetch_add(1, Ordering::SeqCst))
-//     }
-// }
+pub use schemas::internal_protocol::Handle;
+pub use schemas::internal_protocol::Oid;
 
-pub type ConnId = u64;
-pub type V = value::IOValue; // value::ArcValue<Syndicate>;
+use std::sync::atomic::AtomicU64;
+use std::sync::atomic::Ordering;
+
+pub mod actor;
+pub mod bag;
+pub mod config;
+pub mod dataspace;
+pub mod error;
+pub mod schemas;
+pub mod packets;
+pub mod pattern;
+pub mod peer;
+pub mod skeleton;
+
+pub type Assertion = schemas::dataspace::_Any;
+
+pub type ActorId = u64;
+static NEXT_ACTOR_ID: AtomicU64 = AtomicU64::new(0);
+pub fn next_actor_id() -> ActorId {
+    NEXT_ACTOR_ID.fetch_add(1, Ordering::Relaxed)
+}
+
+static NEXT_OID: AtomicU64 = AtomicU64::new(0);
+pub fn next_oid() -> Oid {
+    Oid(value::signed_integer::SignedInteger::from(
+        NEXT_OID.fetch_add(1, Ordering::Relaxed) as u128))
+}
+
+static NEXT_HANDLE: AtomicU64 = AtomicU64::new(0);
+pub fn next_handle() -> Handle {
+    Handle(value::signed_integer::SignedInteger::from(
+        NEXT_HANDLE.fetch_add(1, Ordering::Relaxed) as u128))
+}
+
+static NEXT_MAILBOX_ID: AtomicU64 = AtomicU64::new(0);
+pub fn next_mailbox_id() -> u64 {
+    NEXT_MAILBOX_ID.fetch_add(1, Ordering::Relaxed)
+}

src/packets.rs

@@ -1,91 +1,35 @@
-use super::V;
+pub use crate::schemas::internal_protocol::*;
 
-use bytes::{Buf, buf::BufMutExt, BytesMut};
-use std::sync::Arc;
-use std::marker::PhantomData;
+use bytes::{Buf, BufMut, BytesMut};
 
-use preserves::{
-    de::Deserializer,
-    error,
-    ser::to_writer,
-    value::{PackedReader, PackedWriter},
-};
+use std::convert::TryFrom;
 
-pub type EndpointName = V;
-pub type Assertion = V;
-pub type Captures = Arc<Vec<Assertion>>;
+use preserves::value::PackedReader;
+use preserves::value::PackedWriter;
+use preserves::value::Reader;
+use preserves::value::Writer;
 
-#[derive(Clone, Debug, serde::Serialize, serde::Deserialize)]
-pub enum Action {
-    Assert(EndpointName, Assertion),
-    Clear(EndpointName),
-    Message(Assertion),
-}
+pub struct Codec;
 
-#[derive(Clone, Debug, serde::Serialize, serde::Deserialize)]
-pub enum Event {
-    Add(EndpointName, Captures),
-    Del(EndpointName, Captures),
-    Msg(EndpointName, Captures),
-    End(EndpointName),
-}
-
-#[derive(Clone, Debug, serde::Serialize, serde::Deserialize)]
-pub enum C2S {
-    Connect(V),
-    Turn(Vec<Action>),
-    Ping(),
-    Pong(),
-}
-
-#[derive(Clone, Debug, serde::Serialize, serde::Deserialize)]
-pub enum S2C {
-    Err(String, V),
-    Turn(Vec<Event>),
-    Ping(),
-    Pong(),
-}
-
-//---------------------------------------------------------------------------
-
-pub type Error = error::Error;
-
-pub struct Codec<InT, OutT> {
-    ph_in: PhantomData<InT>,
-    ph_out: PhantomData<OutT>,
-}
-
-pub type ServerCodec = Codec<C2S, S2C>;
-pub type ClientCodec = Codec<S2C, C2S>;
-
-impl<InT, OutT> Codec<InT, OutT> {
-    pub fn new() -> Self {
-        Codec { ph_in: PhantomData, ph_out: PhantomData }
-    }
-}
-
-impl<InT: serde::de::DeserializeOwned, OutT> tokio_util::codec::Decoder for Codec<InT, OutT> {
-    type Item = InT;
+impl tokio_util::codec::Decoder for Codec {
+    type Item = Packet;
     type Error = Error;
     fn decode(&mut self, bs: &mut BytesMut) -> Result<Option<Self::Item>, Self::Error> {
         let mut r = PackedReader::decode_bytes(bs);
-        let mut d = Deserializer::from_reader(&mut r);
-        match Self::Item::deserialize(&mut d) {
-            Err(e) if error::is_eof_error(&e) => Ok(None),
-            Err(e) => Err(e),
-            Ok(item) => {
-                let count = d.read.source.index;
+        match r.next(false)? {
+            None => Ok(None),
+            Some(item) => {
+                let count = r.source.index;
                 bs.advance(count);
-                Ok(Some(item))
+                Ok(Some(Packet::try_from(&item)?))
             }
         }
     }
 }
 
-impl<InT, OutT: serde::Serialize> tokio_util::codec::Encoder<OutT> for Codec<InT, OutT>
-{
+impl tokio_util::codec::Encoder<&Packet> for Codec {
     type Error = Error;
-    fn encode(&mut self, item: OutT, bs: &mut BytesMut) -> Result<(), Self::Error> {
-        to_writer(&mut PackedWriter::new(&mut bs.writer()), &item)
+    fn encode(&mut self, item: &Packet, bs: &mut BytesMut) -> Result<(), Self::Error> {
+        Ok(PackedWriter::new(&mut bs.writer()).write(&item.into())?)
     }
 }

src/pattern.rs

@@ -0,0 +1,81 @@
+use crate::Assertion;
+use crate::schemas::dataspace_patterns::*;
+
+use preserves::value::NestedValue;
+
+use std::convert::TryFrom;
+
+#[derive(Debug, Clone, PartialOrd, Ord, PartialEq, Eq)]
+pub enum PathStep {
+    Index(usize),
+    Key(Assertion),
+}
+
+pub type Path = Vec<PathStep>;
+pub type Paths = Vec<Path>;
+
+struct Analyzer {
+    pub const_paths: Paths,
+    pub const_values: Vec<_Any>,
+    pub capture_paths: Paths,
+}
+
+pub struct PatternAnalysis {
+    pub const_paths: Paths,
+    pub const_values: _Any,
+    pub capture_paths: Paths,
+}
+
+impl PatternAnalysis {
+    pub fn new(p: &Pattern) -> Self {
+        let mut analyzer = Analyzer {
+            const_paths: Vec::new(),
+            const_values: Vec::new(),
+            capture_paths: Vec::new(),
+        };
+        analyzer.walk(&mut Vec::new(), p);
+        PatternAnalysis {
+            const_paths: analyzer.const_paths,
+            const_values: _Any::new(analyzer.const_values),
+            capture_paths: analyzer.capture_paths,
+        }
+    }
+}
+
+impl Analyzer {
+    fn walk_step(&mut self, path: &mut Path, s: PathStep, p: &Pattern) {
+        path.push(s);
+        self.walk(path, p);
+        path.pop();
+    }
+
+    fn walk(&mut self, path: &mut Path, p: &Pattern) {
+        match p {
+            Pattern::DCompound(b) => match &**b {
+                DCompound::Rec { members, .. } |
+                DCompound::Arr { members, .. } => {
+                    for (i, p) in members {
+                        self.walk_step(path, PathStep::Index(usize::try_from(i).unwrap_or(0)), p);
+                    }
+                }
+                DCompound::Dict { members, .. } => {
+                    for (k, p) in members {
+                        self.walk_step(path, PathStep::Key(k.clone()), p);
+                    }
+                }
+            }
+            Pattern::DBind(b) => {
+                let DBind { pattern, .. } = &**b;
+                self.capture_paths.push(path.clone());
+                self.walk(path, pattern)
+            }
+            Pattern::DDiscard(_) =>
+                (),
+            Pattern::DLit(b) => {
+                let DLit { value } = &**b;
+                self.const_paths.push(path.clone());
+                self.const_values.push(value.clone());
+            }
+        }
+    }
+}

src/peer.rs

@@ -1,68 +1,40 @@
-use super::V;
-use super::ConnId;
-use super::dataspace;
-use super::packets;
-use super::spaces;
-use super::config;
-
-use core::time::Duration;
-use futures::{Sink, SinkExt, Stream};
 use futures::FutureExt;
+use futures::StreamExt;
 use futures::select;
-use preserves::value;
-use std::pin::Pin;
-use std::sync::{Mutex, Arc, atomic::{AtomicUsize, Ordering}};
-use tokio::stream::StreamExt;
-use tokio::sync::mpsc::{unbounded_channel, UnboundedSender, UnboundedReceiver, error::TryRecvError};
-use tokio::time::interval;
+use futures::{Sink, SinkExt, Stream};
 
-pub type ResultC2S = Result<packets::C2S, packets::Error>;
+use preserves::value;
+
+use std::pin::Pin;
+use std::sync::Arc;
+use std::sync::atomic::{AtomicUsize, Ordering};
+
+use super::actor::*;
+use super::config;
+use super::error::Error;
+use super::error::error;
+use super::packets;
 
 pub struct Peer<I, O>
-where I: Stream<Item = ResultC2S> + Send,
-      O: Sink<packets::S2C, Error = packets::Error>,
+where I: Stream<Item = Result<packets::Packet, Error>> + Send,
+      O: Sink<packets::Packet, Error = Error>,
 {
-    id: ConnId,
-    tx: UnboundedSender<packets::S2C>,
-    rx: UnboundedReceiver<packets::S2C>,
     i: Pin<Box<I>>,
     o: Pin<Box<O>>,
-    space: Option<dataspace::DataspaceRef>,
-}
-
-fn err(s: &str, ctx: V) -> packets::S2C {
-    packets::S2C::Err(s.into(), ctx)
+    ds: Arc<Ref>,
+    config: Arc<config::ServerConfig>,
 }
 
 impl<I, O> Peer<I, O>
-where I: Stream<Item = ResultC2S> + Send,
-      O: Sink<packets::S2C, Error = packets::Error>,
+where I: Stream<Item = Result<packets::Packet, Error>> + Send,
+      O: Sink<packets::Packet, Error = Error>,
 {
-    pub fn new(id: ConnId, i: I, o: O) -> Self {
-        let (tx, rx) = unbounded_channel();
-        Peer{ id, tx, rx, i: Box::pin(i), o: Box::pin(o), space: None }
+    pub fn new(i: I, o: O, ds: Arc<Ref>, config: Arc<config::ServerConfig>) -> Self {
+        Peer{ i: Box::pin(i), o: Box::pin(o), ds, config }
     }
 
-    pub async fn run(&mut self, spaces: Arc<Mutex<spaces::Spaces>>, config: &config::ServerConfig) ->
-        Result<(), packets::Error>
-    {
-        let firstpacket = self.i.next().await;
-        let dsname = if let Some(Ok(packets::C2S::Connect(dsname))) = firstpacket {
-            dsname
-        } else {
-            let e = format!("Expected initial Connect, got {:?}", firstpacket);
-            self.o.send(err(&e, value::FALSE.clone())).await?;
-            return Err(preserves::error::syntax_error(&e))
-        };
-
-        self.space = Some(spaces.lock().unwrap().lookup(&dsname));
+    pub async fn run(mut self) -> Result<(), packets::Error> {
         let queue_depth = Arc::new(AtomicUsize::new(0));
-        self.space.as_ref().unwrap().write().unwrap().register(
-            self.id,
-            self.tx.clone(),
-            Arc::clone(&queue_depth));
-
-        let mut ping_timer = interval(Duration::from_secs(60));
 
         let mut running = true;
         let mut overloaded = None;
@@ -71,12 +43,11 @@ where I: Stream<Item = ResultC2S> + Send,
             let mut to_send = Vec::new();
 
             let queue_depth_sample = queue_depth.load(Ordering::Relaxed);
-            if queue_depth_sample > config.overload_threshold {
+            if queue_depth_sample > self.config.overload_threshold {
                 let n = overloaded.unwrap_or(0);
                 tracing::warn!(turns=n, queue_depth=queue_depth_sample, "overloaded");
-                if n == config.overload_turn_limit {
-                    to_send.push(err("Overloaded",
-                                     value::Value::from(queue_depth_sample as u64).wrap()));
+                if n == self.config.overload_turn_limit {
+                    to_send.push(error("Overloaded", queue_depth_sample as u128));
                     running = false;
                 } else {
                     if queue_depth_sample > previous_sample.unwrap_or(0) {
@@ -94,50 +65,22 @@ where I: Stream<Item = ResultC2S> + Send,
             previous_sample = Some(queue_depth_sample);
 
             select! {
-                _instant = ping_timer.next().boxed().fuse() => to_send.push(packets::S2C::Ping()),
                 frame = self.i.next().fuse() => match frame {
                     Some(res) => match res {
                         Ok(p) => {
                             tracing::trace!(packet = debug(&p), "input");
                             match p {
-                                packets::C2S::Turn(actions) => {
-                                    match self.space.as_ref().unwrap().write().unwrap()
-                                        .turn(self.id, actions)
-                                    {
-                                        Ok(()) => (),
-                                        Err((msg, ctx)) => {
-                                            to_send.push(err(&msg, ctx));
-                                            running = false;
-                                        }
-                                    }
+                                packets::Packet::Turn(b) => {
+                                    let packets::Turn(actions) = &*b;
+                                    /* ... */
                                 }
-                                packets::C2S::Ping() =>
-                                    to_send.push(packets::S2C::Pong()),
-                                packets::C2S::Pong() =>
-                                    (),
-                                packets::C2S::Connect(_) => {
-                                    to_send.push(err("Unexpected Connect", value::to_value(p)));
-                                    running = false;
+                                packets::Packet::Error(b) => {
+                                    let e = &*b;
+                                    /* ... */
                                 }
                             }
                         }
-                        Err(e) if preserves::error::is_eof_error(&e) => {
-                            tracing::trace!("eof");
-                            running = false;
-                        }
-                        Err(e) if preserves::error::is_syntax_error(&e) => {
-                            to_send.push(err(&e.to_string(), value::FALSE.clone()));
-                            running = false;
-                        }
-                        Err(e) => {
-                            if preserves::error::is_io_error(&e) {
-                                return Err(e);
-                            } else {
-                                to_send.push(err(&format!("Packet deserialization error: {}", e),
-                                                 value::FALSE.clone()));
-                                running = false;
-                            }
-                        }
+                        Err(e) => return Err(e),
                     }
                     None => {
                         tracing::trace!("remote has closed");
@@ -167,7 +110,7 @@ where I: Stream<Item = ResultC2S> + Send,
                     }
                     if !ok {
                         /* weird. */
-                        to_send.push(err("Outbound channel closed unexpectedly", value::FALSE.clone()));
+                        to_send.push(error("Outbound channel closed unexpectedly", value::FALSE.clone()));
                         running = false;
                     }
                 },
@@ -185,14 +128,3 @@ where I: Stream<Item = ResultC2S> + Send,
         Ok(())
     }
 }
-
-impl<I, O> Drop for Peer<I, O>
-where I: Stream<Item = ResultC2S> + Send,
-      O: Sink<packets::S2C, Error = packets::Error>,
-{
-    fn drop(&mut self) {
-        if let Some(ref s) = self.space {
-            s.write().unwrap().deregister(self.id);
-        }
-    }
-}

src/skeleton.rs

@@ -1,118 +1,99 @@
-use super::ConnId;
+use super::Assertion;
 use super::bag;
-use super::packets::Assertion;
-use super::packets::Captures;
-use super::packets::EndpointName;
-use super::packets::Event;
 
 use preserves::value::{Map, Set, Value, NestedValue};
-use std::cmp::Ordering;
 use std::collections::btree_map::Entry;
+use std::convert::TryFrom;
+use std::convert::TryInto;
 use std::sync::Arc;
 
+use crate::actor::Activation;
+use crate::actor::Ref;
+use crate::schemas::internal_protocol::Handle;
+use crate::schemas::dataspace_patterns as ds;
+use crate::pattern::{self, PathStep, Path, Paths};
+
 type Bag<A> = bag::BTreeBag<A>;
 
-pub type Path = Vec<usize>;
-pub type Paths = Vec<Path>;
-pub type Events = Vec<Event>;
-pub type TurnMap = Map<ConnId, Events>;
+type Captures = Assertion;
 
 #[derive(Debug, PartialEq, Eq, PartialOrd, Ord, Clone)]
-pub struct Endpoint {
-    pub connection: ConnId,
-    pub name: EndpointName,
-}
-
-#[derive(Debug)]
-pub enum Skeleton {
-    Blank,
-    Guarded(Guard, Vec<Skeleton>)
-}
-
-#[derive(Debug)]
-pub struct AnalysisResults {
-    pub skeleton: Skeleton,
-    pub const_paths: Paths,
-    pub const_vals: Captures,
-    pub capture_paths: Paths,
-    pub assertion: Assertion,
+pub enum Guard {
+    Rec(Assertion, usize),
+    Seq(usize),
+    Map,
 }
 
 #[derive(Debug)]
 pub struct Index {
-    all_assertions: Bag<CachedAssertion>,
+    all_assertions: Bag<Assertion>,
+    observer_count: usize,
     root: Node,
 }
 
+#[derive(Debug)]
+struct Node {
+    continuation: Continuation,
+    edges: Map<Selector, Map<Guard, Node>>,
+}
+
+#[derive(Debug)]
+struct Continuation {
+    cached_assertions: Set<Assertion>,
+    leaf_map: Map<Paths, Map<Captures, Leaf>>,
+}
+
+#[derive(Debug, PartialEq, Eq, PartialOrd, Ord)]
+struct Selector {
+    pop_count: usize,
+    step: PathStep,
+}
+
+#[derive(Debug)]
+struct Leaf { // aka Topic
+    cached_assertions: Set<Assertion>,
+    endpoints_map: Map<Paths, Endpoints>,
+}
+
+#[derive(Debug)]
+struct Endpoints {
+    cached_captures: Bag<Captures>,
+    endpoints: Map<Arc<Ref>, Map<Captures, Handle>>,
+}
+
+//---------------------------------------------------------------------------
+
 impl Index {
     pub fn new() -> Self {
-        Index{ all_assertions: Bag::new(), root: Node::new(Continuation::new(Set::new())) }
-    }
-
-    pub fn add_endpoint(&mut self, analysis_results: &AnalysisResults, endpoint: Endpoint) -> Events
-    {
-        let continuation = self.root.extend(&analysis_results.skeleton);
-        let continuation_cached_assertions = &continuation.cached_assertions;
-        let const_val_map =
-            continuation.leaf_map.entry(analysis_results.const_paths.clone()).or_insert_with(|| {
-                let mut cvm = Map::new();
-                for a in continuation_cached_assertions {
-                    let key = project_paths(a.unscope(), &analysis_results.const_paths);
-                    cvm.entry(key).or_insert_with(Leaf::new).cached_assertions.insert(a.clone());
-                }
-                cvm
-            });
-        let capture_paths = &analysis_results.capture_paths;
-        let leaf = const_val_map.entry(analysis_results.const_vals.clone()).or_insert_with(Leaf::new);
-        let leaf_cached_assertions = &leaf.cached_assertions;
-        let endpoints = leaf.endpoints_map.entry(capture_paths.clone()).or_insert_with(|| {
-            let mut b = Bag::new();
-            for a in leaf_cached_assertions {
-                let (restriction_paths, term) = a.unpack();
-                if is_unrestricted(&capture_paths, restriction_paths) {
-                    let captures = project_paths(term, &capture_paths);
-                    *b.entry(captures).or_insert(0) += 1;
-                }
-            }
-            Endpoints::new(b)
-        });
-        let endpoint_name = endpoint.name.clone();
-        endpoints.endpoints.insert(endpoint);
-        endpoints.cached_captures.into_iter()
-            .map(|(cs,_)| Event::Add(endpoint_name.clone(), cs.clone()))
-            .collect()
-    }
-
-    pub fn remove_endpoint(&mut self, analysis_results: &AnalysisResults, endpoint: Endpoint) {
-        let continuation = self.root.extend(&analysis_results.skeleton);
-        if let Entry::Occupied(mut const_val_map_entry)
-            = continuation.leaf_map.entry(analysis_results.const_paths.clone())
-        {
-            let const_val_map = const_val_map_entry.get_mut();
-            if let Entry::Occupied(mut leaf_entry)
-                = const_val_map.entry(analysis_results.const_vals.clone())
-            {
-                let leaf = leaf_entry.get_mut();
-                if let Entry::Occupied(mut endpoints_entry)
-                    = leaf.endpoints_map.entry(analysis_results.capture_paths.clone())
-                {
-                    let endpoints = endpoints_entry.get_mut();
-                    endpoints.endpoints.remove(&endpoint);
-                    if endpoints.endpoints.is_empty() {
-                        endpoints_entry.remove_entry();
-                    }
-                }
-                if leaf.is_empty() {
-                    leaf_entry.remove_entry();
-                }
-            }
-            if const_val_map.is_empty() {
-                const_val_map_entry.remove_entry();
-            }
+        Index {
+            all_assertions: Bag::new(),
+            observer_count: 0,
+            root: Node::new(Continuation::new(Set::new())),
         }
     }
 
-    pub fn insert(&mut self, outer_value: CachedAssertion, outputs: &mut TurnMap) {
+    pub fn add_observer(
+        &mut self,
+        t: &mut Activation,
+        pat: &ds::Pattern,
+        observer: &Arc<Ref>,
+    ) {
+        let analysis = pattern::PatternAnalysis::new(pat);
+        self.root.extend(pat).add_observer(t, &analysis, observer);
+        self.observer_count += 1;
+    }
+
+    pub fn remove_observer(
+        &mut self,
+        pat: ds::Pattern,
+        observer: &Arc<Ref>,
+    ) {
+        let analysis = pattern::PatternAnalysis::new(&pat);
+        self.root.extend(&pat).remove_observer(analysis, observer);
+        self.observer_count -= 1;
+    }
+
+    pub fn insert(&mut self, t: &mut Activation, outer_value: &Assertion) {
         let net = self.all_assertions.change(outer_value.clone(), 1);
         match net {
             bag::Net::AbsentToPresent => {
@@ -123,9 +104,8 @@ impl Index {
                     |l, v| { l.cached_assertions.insert(v.clone()); },
                     |es, cs| {
                         if es.cached_captures.change(cs.clone(), 1) == bag::Net::AbsentToPresent {
-                            for ep in &es.endpoints {
-                                outputs.entry(ep.connection).or_insert_with(Vec::new)
-                                    .push(Event::Add(ep.name.clone(), cs.clone()))
+                            for (observer, capture_map) in &mut es.endpoints {
+                                capture_map.insert(cs.clone(), t.assert(observer.clone(), cs.clone()));
                             }
                         }
                     })
@@ -136,7 +116,7 @@ impl Index {
         }
     }
 
-    pub fn remove(&mut self, outer_value: CachedAssertion, outputs: &mut TurnMap) {
+    pub fn remove(&mut self, t: &mut Activation, outer_value: &Assertion) {
         let net = self.all_assertions.change(outer_value.clone(), -1);
         match net {
             bag::Net::PresentToAbsent => {
@@ -147,9 +127,10 @@ impl Index {
                     |l, v| { l.cached_assertions.remove(v); },
                     |es, cs| {
                         if es.cached_captures.change(cs.clone(), -1) == bag::Net::PresentToAbsent {
-                            for ep in &es.endpoints {
-                                outputs.entry(ep.connection).or_insert_with(Vec::new)
-                                    .push(Event::Del(ep.name.clone(), cs.clone()))
+                            for capture_map in es.endpoints.values_mut() {
+                                if let Some(h) = capture_map.remove(&cs) {
+                                    t.retract(h);
+                                }
                             }
                         }
                     })
@@ -160,11 +141,7 @@ impl Index {
         }
     }
 
-    pub fn send(&mut self,
-                outer_value: CachedAssertion,
-                outputs: &mut TurnMap,
-                delivery_count: &mut usize)
-    {
+    pub fn send(&mut self, t: &mut Activation, outer_value: &Assertion, delivery_count: &mut usize) {
         Modification::new(
             false,
             &outer_value,
@@ -172,9 +149,8 @@ impl Index {
             |_l, _v| (),
             |es, cs| {
                 *delivery_count += es.endpoints.len();
-                for ep in &es.endpoints {
-                    outputs.entry(ep.connection).or_insert_with(Vec::new)
-                        .push(Event::Msg(ep.name.clone(), cs.clone()))
+                for observer in es.endpoints.keys() {
+                    t.message(observer.clone(), cs.clone());
                 }
             }).perform(&mut self.root);
     }
@@ -186,12 +162,10 @@ impl Index {
     pub fn endpoint_count(&self) -> isize {
         return self.all_assertions.total()
     }
-}
 
-#[derive(Debug)]
-struct Node {
-    continuation: Continuation,
-    edges: Map<Selector, Map<Guard, Node>>,
+    pub fn observer_count(&self) -> usize {
+        return self.observer_count
+    }
 }
 
 impl Node {
@@ -199,38 +173,60 @@ impl Node {
         Node { continuation, edges: Map::new() }
     }
 
-    fn extend(&mut self, skeleton: &Skeleton) -> &mut Continuation {
-        let (_pop_count, final_node) = self.extend_walk(&mut Vec::new(), 0, 0, skeleton);
+    fn extend(&mut self, pat: &ds::Pattern) -> &mut Continuation {
+        let (_pop_count, final_node) = self.extend_walk(&mut Vec::new(), 0, PathStep::Index(0), pat);
         &mut final_node.continuation
     }
 
-    fn extend_walk(&mut self, path: &mut Path, pop_count: usize, index: usize, skeleton: &Skeleton)
-                   -> (usize, &mut Node) {
-        match skeleton {
-            Skeleton::Blank => (pop_count, self),
-            Skeleton::Guarded(cls, kids) => {
-                let selector = Selector { pop_count, index };
-                let continuation = &self.continuation;
-                let table = self.edges.entry(selector).or_insert_with(Map::new);
-                let mut next_node = table.entry(cls.clone()).or_insert_with(|| {
-                    Self::new(Continuation::new(
-                        continuation.cached_assertions.iter()
-                            .filter(|a| {
-                                Some(cls) == class_of(project_path(a.unscope(), path)).as_ref() })
-                            .cloned()
-                            .collect()))
-                });
-                let mut pop_count = 0;
-                for (index, kid) in kids.iter().enumerate() {
-                    path.push(index);
-                    let (pc, nn) = next_node.extend_walk(path, pop_count, index, kid);
-                    pop_count = pc;
-                    next_node = nn;
-                    path.pop();
-                }
-                (pop_count + 1, next_node)
+    fn extend_walk(
+        &mut self,
+        path: &mut Path,
+        pop_count: usize,
+        step: PathStep,
+        pat: &ds::Pattern,
+    ) -> (usize, &mut Node) {
+        let (guard, members): (Guard, Vec<(PathStep, &ds::Pattern)>) = match pat {
+            ds::Pattern::DCompound(b) => match &**b {
+                ds::DCompound::Arr { ctor, members } =>
+                    (Guard::Seq(usize::try_from(&ctor.arity).unwrap_or(0)),
+                     members.iter().map(|(i, p)| (PathStep::Index(i.try_into().unwrap_or(0)), p)).collect()),
+                ds::DCompound::Rec { ctor, members } =>
+                    (Guard::Rec(ctor.label.clone(), usize::try_from(&ctor.arity).unwrap_or(0)),
+                     members.iter().map(|(i, p)| (PathStep::Index(i.try_into().unwrap_or(0)), p)).collect()),
+                ds::DCompound::Dict { members, .. } =>
+                    (Guard::Map,
+                     members.iter().map(|(k, p)| (PathStep::Key(k.clone()), p)).collect()),
             }
+            ds::Pattern::DBind(b) => {
+                let ds::DBind { pattern, .. } = &**b;
+                return self.extend_walk(path, pop_count, step, pattern);
+            }
+            ds::Pattern::DDiscard(_) | ds::Pattern::DLit(_) =>
+                return (pop_count, self),
+        };
+
+        let selector = Selector { pop_count, step };
+        let continuation = &self.continuation;
+        let table = self.edges.entry(selector).or_insert_with(Map::new);
+        let mut next_node = table.entry(guard.clone()).or_insert_with(|| {
+            Self::new(Continuation::new(
+                continuation.cached_assertions.iter()
+                    .filter(|a| match project_path(a, path) {
+                        Some(v) => Some(&guard) == class_of(v).as_ref(),
+                        None => false,
+                    })
+                    .cloned()
+                    .collect()))
+        });
+        let mut pop_count = 0;
+        for (step, kid) in members.into_iter() {
+            path.push(step.clone());
+            let (pc, nn) = next_node.extend_walk(path, pop_count, step, kid);
+            pop_count = pc;
+            next_node = nn;
+            path.pop();
         }
+        (pop_count + 1, next_node)
     }
 }
 
@@ -257,35 +253,31 @@ impl<'a, T> Stack<'a, T> {
 }
 
 struct Modification<'op, FCont, FLeaf, FEndpoints>
-where FCont: FnMut(&mut Continuation, &CachedAssertion) -> (),
-      FLeaf: FnMut(&mut Leaf, &CachedAssertion) -> (),
+where FCont: FnMut(&mut Continuation, &Assertion) -> (),
+      FLeaf: FnMut(&mut Leaf, &Assertion) -> (),
       FEndpoints: FnMut(&mut Endpoints, Captures) -> ()
 {
     create_leaf_if_absent: bool,
-    outer_value: &'op CachedAssertion,
-    restriction_paths: Option<&'op Paths>,
-    outer_value_term: &'op Assertion,
+    outer_value: &'op Assertion,
     m_cont: FCont,
     m_leaf: FLeaf,
     m_endpoints: FEndpoints,
 }
 
 impl<'op, FCont, FLeaf, FEndpoints> Modification<'op, FCont, FLeaf, FEndpoints>
-where FCont: FnMut(&mut Continuation, &CachedAssertion) -> (),
-      FLeaf: FnMut(&mut Leaf, &CachedAssertion) -> (),
+where FCont: FnMut(&mut Continuation, &Assertion) -> (),
+      FLeaf: FnMut(&mut Leaf, &Assertion) -> (),
       FEndpoints: FnMut(&mut Endpoints, Captures) -> ()
 {
     fn new(create_leaf_if_absent: bool,
-           outer_value: &'op CachedAssertion,
+           outer_value: &'op Assertion,
            m_cont: FCont,
            m_leaf: FLeaf,
-           m_endpoints: FEndpoints) -> Self {
-        let (restriction_paths, outer_value_term) = outer_value.unpack();
+           m_endpoints: FEndpoints,
+    ) -> Self {
         Modification {
             create_leaf_if_absent,
             outer_value,
-            restriction_paths,
-            outer_value_term,
             m_cont,
             m_leaf,
             m_endpoints,
@@ -293,7 +285,7 @@ where FCont: FnMut(&mut Continuation, &CachedAssertion) -> (),
     }
 
     fn perform(&mut self, n: &mut Node) {
-        self.node(n, &Stack::Item(&Value::from(vec![self.outer_value_term.clone()]).wrap(), &Stack::Empty))
+        self.node(n, &Stack::Item(&Value::from(vec![self.outer_value.clone()]).wrap(), &Stack::Empty))
     }
 
     fn node(&mut self, n: &mut Node, term_stack: &Stack<&Assertion>) {
@@ -301,10 +293,11 @@ where FCont: FnMut(&mut Continuation, &CachedAssertion) -> (),
         for (selector, table) in &mut n.edges {
             let mut next_stack = term_stack;
             for _ in 0..selector.pop_count { next_stack = next_stack.pop() }
-            let next_value = step(next_stack.top(), selector.index);
-            if let Some(next_class) = class_of(next_value) {
-                if let Some(next_node) = table.get_mut(&next_class) {
-                    self.node(next_node, &Stack::Item(next_value, next_stack))
+            if let Some(next_value) = step(next_stack.top(), &selector.step) {
+                if let Some(next_class) = class_of(next_value) {
+                    if let Some(next_node) = table.get_mut(&next_class) {
+                        self.node(next_node, &Stack::Item(next_value, next_stack))
+                    }
                 }
             }
         }
@@ -314,24 +307,24 @@ where FCont: FnMut(&mut Continuation, &CachedAssertion) -> (),
         (self.m_cont)(c, self.outer_value);
         let mut empty_const_paths = Vec::new();
         for (const_paths, const_val_map) in &mut c.leaf_map {
-            let const_vals = project_paths(self.outer_value_term, const_paths);
-            let leaf_opt = if self.create_leaf_if_absent {
-                Some(const_val_map.entry(const_vals.clone()).or_insert_with(Leaf::new))
-            } else {
-                const_val_map.get_mut(&const_vals)
-            };
-            if let Some(leaf) = leaf_opt {
-                (self.m_leaf)(leaf, self.outer_value);
-                for (capture_paths, endpoints) in &mut leaf.endpoints_map {
-                    if is_unrestricted(&capture_paths, self.restriction_paths) {
-                        (self.m_endpoints)(endpoints,
-                                           project_paths(self.outer_value_term, &capture_paths));
+            if let Some(const_vals) = project_paths(self.outer_value, const_paths) {
+                let leaf_opt = if self.create_leaf_if_absent {
+                    Some(const_val_map.entry(const_vals.clone()).or_insert_with(Leaf::new))
+                } else {
+                    const_val_map.get_mut(&const_vals)
+                };
+                if let Some(leaf) = leaf_opt {
+                    (self.m_leaf)(leaf, self.outer_value);
+                    for (capture_paths, endpoints) in &mut leaf.endpoints_map {
+                        if let Some(cs) = project_paths(self.outer_value, &capture_paths) {
+                            (self.m_endpoints)(endpoints, cs);
+                        }
                     }
-                }
-                if leaf.is_empty() {
-                    const_val_map.remove(&const_vals);
-                    if const_val_map.is_empty() {
-                        empty_const_paths.push(const_paths.clone());
+                    if leaf.is_empty() {
+                        const_val_map.remove(&const_vals);
+                        if const_val_map.is_empty() {
+                            empty_const_paths.push(const_paths.clone());
+                        }
                     }
                 }
             }
@@ -344,71 +337,124 @@ where FCont: FnMut(&mut Continuation, &CachedAssertion) -> (),
 
 fn class_of(v: &Assertion) -> Option<Guard> {
     match v.value() {
-        Value::Sequence(ref vs) => Some(Guard::Seq(vs.len())),
-        Value::Record(ref r) => Some(Guard::Rec(r.label().clone(), r.arity())),
+        Value::Sequence(vs) => Some(Guard::Seq(vs.len())),
+        Value::Record(r) => Some(Guard::Rec(r.label().clone(), r.arity())),
+        Value::Dictionary(_) => Some(Guard::Map),
         _ => None,
     }
 }
 
-fn project_path<'a>(v: &'a Assertion, p: &Path) -> &'a Assertion {
+fn project_path<'a>(v: &'a Assertion, p: &Path) -> Option<&'a Assertion> {
     let mut v = v;
     for i in p {
-        v = step(v, *i);
+        match step(v, i) {
+            Some(w) => v = w,
+            None => return None,
+        }
     }
-    v
+    Some(v)
 }
 
-fn project_paths<'a>(v: &'a Assertion, ps: &Paths) -> Captures {
-    Arc::new(ps.iter().map(|p| project_path(v, p)).cloned().collect())
-}
-
-fn step(v: &Assertion, i: usize) -> &Assertion {
-    match v.value() {
-        Value::Sequence(ref vs) => &vs[i],
-        Value::Record(ref r) => &r.fields()[i],
-        _ => panic!("step: non-sequence, non-record {:?}", v)
+fn project_paths<'a>(v: &'a Assertion, ps: &Paths) -> Option<Captures> {
+    let mut vs = Vec::new();
+    for p in ps {
+        match project_path(v, p) {
+            Some(c) => vs.push(c.clone()),
+            None => return None,
+        }
     }
+    Some(Captures::new(vs))
 }
 
-#[derive(Debug)]
-struct Continuation {
-    cached_assertions: Set<CachedAssertion>,
-    leaf_map: Map<Paths, Map<Captures, Leaf>>,
+fn step<'a>(v: &'a Assertion, s: &PathStep) -> Option<&'a Assertion> {
+    match (v.value(), s) {
+        (Value::Sequence(vs), PathStep::Index(i)) =>
+            if *i < vs.len() { Some(&vs[*i]) } else { None },
+        (Value::Record(r), PathStep::Index(i)) =>
+            if *i < r.arity() { Some(&r.fields()[*i]) } else { None },
+        (Value::Dictionary(m), PathStep::Key(k)) =>
+            m.get(k),
+        _ =>
+            None,
+    }
 }
 
 impl Continuation {
-    fn new(cached_assertions: Set<CachedAssertion>) -> Self {
+    fn new(cached_assertions: Set<Assertion>) -> Self {
         Continuation { cached_assertions, leaf_map: Map::new() }
     }
-}
 
-#[derive(Debug, PartialEq, Eq, PartialOrd, Ord)]
-struct Selector {
-    pop_count: usize,
-    index: usize,
-}
+    pub fn add_observer(
+        &mut self,
+        t: &mut Activation,
+        analysis: &pattern::PatternAnalysis,
+        observer: &Arc<Ref>,
+    ) {
+        let cached_assertions = &self.cached_assertions;
+        let const_val_map =
+            self.leaf_map.entry(analysis.const_paths.clone()).or_insert_with({
+                || {
+                    let mut cvm = Map::new();
+                    for a in cached_assertions {
+                        if let Some(key) = project_paths(a, &analysis.const_paths) {
+                            cvm.entry(key).or_insert_with(Leaf::new)
+                                .cached_assertions.insert(a.clone());
+                        }
+                    }
+                    cvm
+                }
+            });
+        let leaf = const_val_map.entry(analysis.const_values.clone()).or_insert_with(Leaf::new);
+        let leaf_cached_assertions = &leaf.cached_assertions;
+        let endpoints = leaf.endpoints_map.entry(analysis.capture_paths.clone()).or_insert_with(|| {
+            let mut b = Bag::new();
+            for term in leaf_cached_assertions {
+                if let Some(captures) = project_paths(term, &analysis.capture_paths) {
+                    *b.entry(captures).or_insert(0) += 1;
+                }
+            }
+            Endpoints { cached_captures: b, endpoints: Map::new() }
+        });
+        let mut capture_map = Map::new();
+        for cs in endpoints.cached_captures.keys() {
+            capture_map.insert(cs.clone(), t.assert(observer.clone(), cs.clone()));
+        }
+        endpoints.endpoints.insert(observer.clone(), capture_map);
+    }
 
-#[derive(Debug, PartialEq, Eq, PartialOrd, Ord, Clone)]
-pub enum Guard {
-    Rec(Assertion, usize),
-    Seq(usize),
-}
-
-impl Guard {
-    fn arity(&self) -> usize {
-        match self {
-            Guard::Rec(_, s) => *s,
-            Guard::Seq(s) => *s
+    pub fn remove_observer(
+        &mut self,
+        analysis: pattern::PatternAnalysis,
+        observer: &Arc<Ref>,
+    ) {
+        if let Entry::Occupied(mut const_val_map_entry)
+            = self.leaf_map.entry(analysis.const_paths)
+        {
+            let const_val_map = const_val_map_entry.get_mut();
+            if let Entry::Occupied(mut leaf_entry)
+                = const_val_map.entry(analysis.const_values)
+            {
+                let leaf = leaf_entry.get_mut();
+                if let Entry::Occupied(mut endpoints_entry)
+                    = leaf.endpoints_map.entry(analysis.capture_paths)
+                {
+                    let endpoints = endpoints_entry.get_mut();
+                    endpoints.endpoints.remove(observer);
+                    if endpoints.endpoints.is_empty() {
+                        endpoints_entry.remove_entry();
+                    }
+                }
+                if leaf.is_empty() {
+                    leaf_entry.remove_entry();
+                }
+            }
+            if const_val_map.is_empty() {
+                const_val_map_entry.remove_entry();
+            }
         }
     }
 }
 
-#[derive(Debug)]
-struct Leaf { // aka Topic
-    cached_assertions: Set<CachedAssertion>,
-    endpoints_map: Map<Paths, Endpoints>,
-}
-
 impl Leaf {
     fn new() -> Self {
         Leaf { cached_assertions: Set::new(), endpoints_map: Map::new() }
@@ -418,192 +464,3 @@ impl Leaf {
         self.cached_assertions.is_empty() && self.endpoints_map.is_empty()
     }
 }
-
-#[derive(Debug)]
-struct Endpoints {
-    cached_captures: Bag<Captures>,
-    endpoints: Set<Endpoint>,
-}
-
-impl Endpoints {
-    fn new(cached_captures: Bag<Captures>) -> Self {
-        Endpoints { cached_captures, endpoints: Set::new() }
-    }
-}
-
-#[derive(Debug, PartialEq, Eq, PartialOrd, Ord, Clone)]
-pub enum CachedAssertion {
-    VisibilityRestricted(Paths, Assertion),
-    Unrestricted(Assertion),
-}
-
-impl From<&Assertion> for CachedAssertion {
-    fn from(a: &Assertion) -> Self {
-        CachedAssertion::Unrestricted(a.clone())
-    }
-}
-
-impl CachedAssertion {
-    fn unscope(&self) -> &Assertion {
-        match self {
-            CachedAssertion::VisibilityRestricted(_, a) => a,
-            CachedAssertion::Unrestricted(a) => a,
-        }
-    }
-
-    fn unpack(&self) -> (Option<&Paths>, &Assertion) {
-        match self {
-            CachedAssertion::VisibilityRestricted(ps, a) => (Some(ps), a),
-            CachedAssertion::Unrestricted(a) => (None, a),
-        }
-    }
-}
-
-fn is_unrestricted(capture_paths: &Paths, restriction_paths: Option<&Paths>) -> bool {
-    // We are "unrestricted" if Set(capture_paths) ⊆ Set(restriction_paths). Since both
-    // variables really hold lists, we operate with awareness of the order the lists are
-    // built here. We know that the lists are built in fringe order; that is, they are
-    // sorted wrt `pathCmp`.
-    match restriction_paths {
-        None => true, // not visibility-restricted in the first place
-        Some(rpaths) => {
-            let mut rpi = rpaths.iter();
-            'outer: for c in capture_paths {
-                'inner: loop {
-                    match rpi.next() {
-                        None => {
-                            // there's at least one capture_paths entry (`c`) that does
-                            // not appear in restriction_paths, so we are restricted
-                            return false;
-                        }
-                        Some(r) => match c.cmp(r) {
-                            Ordering::Less => {
-                                // `c` is less than `r`, but restriction_paths is sorted,
-                                // so `c` does not appear in restriction_paths, and we are
-                                // thus restricted.
-                                return false;
-                            }
-                            Ordering::Equal => {
-                                // `c` is equal to `r`, so we may yet be unrestricted.
-                                // Discard both `c` and `r` and continue.
-                                continue 'outer;
-                            }
-                            Ordering::Greater => {
-                                // `c` is greater than `r`, but capture_paths and
-                                // restriction_paths are sorted, so while we might yet
-                                // come to an `r` that is equal to `c`, we will never find
-                                // another `c` that is less than this `c`. Discard this
-                                // `r` then, keeping the `c`, and compare against the next
-                                // `r`.
-                                continue 'inner;
-                            }
-                        }
-                    }
-                }
-            }
-            // We went all the way through capture_paths without finding any `c` not in
-            // restriction_paths.
-            true
-        }
-    }
-}
-
-pub struct Analyzer {
-    const_paths: Paths,
-    const_vals: Vec<Assertion>,
-    capture_paths: Paths,
-    path: Path,
-}
-
-impl Analyzer {
-    fn walk(&mut self, mut a: &Assertion) -> Skeleton {
-        while let Some(fields) = a.value().as_simple_record("capture", Some(1)) {
-            self.capture_paths.push(self.path.clone());
-            a = &fields[0];
-        }
-
-        if a.value().is_simple_record("discard", Some(0)) {
-            Skeleton::Blank
-        } else {
-            match class_of(a) {
-                Some(cls) => {
-                    let arity = cls.arity();
-                    Skeleton::Guarded(cls,
-                                      (0..arity).map(|i| {
-                                          self.path.push(i);
-                                          let s = self.walk(step(a, i));
-                                          self.path.pop();
-                                          s
-                                      }).collect())
-                }
-                None => {
-                    self.const_paths.push(self.path.clone());
-                    self.const_vals.push(a.clone());
-                    Skeleton::Blank
-                }
-            }
-        }
-    }
-}
-
-pub fn analyze(a: &Assertion) -> AnalysisResults {
-    let mut z = Analyzer {
-        const_paths: Vec::new(),
-        const_vals: Vec::new(),
-        capture_paths: Vec::new(),
-        path: Vec::new(),
-    };
-    let skeleton = z.walk(a);
-    AnalysisResults {
-        skeleton,
-        const_paths: z.const_paths,
-        const_vals: Arc::new(z.const_vals),
-        capture_paths: z.capture_paths,
-        assertion: a.clone(),
-    }
-}
-
-// pub fn instantiate_assertion(a: &Assertion, cs: Captures) -> CachedAssertion {
-//     let mut capture_paths = Vec::new();
-//     let mut path = Vec::new();
-//     let mut vs: Vec<Assertion> = (*cs).clone();
-//     vs.reverse();
-//     let instantiated = instantiate_assertion_walk(&mut capture_paths, &mut path, &mut vs, a);
-//     CachedAssertion::VisibilityRestricted(capture_paths, instantiated)
-// }
-
-// fn instantiate_assertion_walk(capture_paths: &mut Paths,
-//                               path: &mut Path,
-//                               vs: &mut Vec<Assertion>,
-//                               a: &Assertion) -> Assertion {
-//     if let Some(fields) = a.value().as_simple_record("capture", Some(1)) {
-//         capture_paths.push(path.clone());
-//         let v = vs.pop().unwrap();
-//         instantiate_assertion_walk(capture_paths, path, vs, &fields[0]);
-//         v
-//     } else if a.value().is_simple_record("discard", Some(0)) {
-//         Value::Domain(Syndicate::new_placeholder()).wrap()
-//     } else {
-//         let f = |(i, aa)| {
-//             path.push(i);
-//             let vv = instantiate_assertion_walk(capture_paths,
-//                                                 path,
-//                                                 vs,
-//                                                 aa);
-//             path.pop();
-//             vv
-//         };
-//         match class_of(a) {
-//             Some(Guard::Seq(_)) =>
-//                 Value::from(Vec::from_iter(a.value().as_sequence().unwrap()
-//                                            .iter().enumerate().map(f)))
-//                 .wrap(),
-//             Some(Guard::Rec(l, fieldcount)) =>
-//                 Value::record(l, a.value().as_record(Some(fieldcount)).unwrap().1
-//                               .iter().enumerate().map(f).collect())
-//                 .wrap(),
-//             None =>
-//                 a.clone(),
-//         }
-//     }
-// }

src/spaces.rs

@@ -1,54 +0,0 @@
-use super::V;
-use super::dataspace;
-
-use std::sync::Arc;
-
-use tracing::{info, debug};
-
-use preserves::value::Map;
-
-pub struct Spaces {
-    index: Map<V, dataspace::DataspaceRef>,
-}
-
-impl Spaces {
-    pub fn new() -> Self {
-        Self { index: Map::new() }
-    }
-
-    pub fn lookup(&mut self, name: &V) -> dataspace::DataspaceRef {
-        let (is_new, space) = match self.index.get(name) {
-            Some(s) => (false, s.clone()),
-            None => {
-                let s = dataspace::Dataspace::new_ref(name);
-                self.index.insert(name.clone(), s.clone());
-                (true, s)
-            }
-        };
-
-        debug!(name = debug(name),
-               action = display(if is_new { "created" } else { "accessed" }));
-
-        space
-    }
-
-    pub fn cleanup(&mut self) {
-        self.index = self.index.iter()
-            .filter(|s| s.1.read().unwrap().peer_count() > 0)
-            .map(|(k,v)| (k.clone(), Arc::clone(v)))
-            .collect();
-    }
-
-    pub fn dump_stats(&self, delta: core::time::Duration)  {
-        for (dsname, dsref) in &self.index {
-            let mut ds = dsref.write().unwrap();
-            info!(name = debug(dsname),
-                  connections = display(format!("{} (+{}/-{})", ds.peer_count(), ds.churn.peers_added, ds.churn.peers_removed)),
-                  assertions = display(format!("{} (+{}/-{})", ds.assertion_count(), ds.churn.assertions_added, ds.churn.assertions_removed)),
-                  endpoints = display(format!("{} (+{}/-{})", ds.endpoint_count(), ds.churn.endpoints_added, ds.churn.endpoints_removed)),
-                  msg_in_rate = display(ds.churn.messages_injected as f32 / delta.as_secs() as f32),
-                  msg_out_rate = display(ds.churn.messages_delivered as f32 / delta.as_secs() as f32));
-            ds.churn.reset();
-        }
-    }
-}