Add local state query mini-protocol naive implementation

pallas-localstate/.gitignore

@@ -0,0 +1,2 @@
+/target
+Cargo.lock

pallas-localstate/Cargo.toml

@@ -0,0 +1,25 @@
+[package]
+name = "pallas-localstate"
+version = "0.1.0"
+edition = "2021"
+repository = "https://github.com/txpipe/pallas"
+license = "Apache-2.0"
+authors = [
+    "Santiago Carmuega <santiago@carmuega.me>"
+]
+
+# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
+
+[dependencies]
+pallas-multiplexer = { path = "../pallas-multiplexer/" }
+pallas-machines = { path = "../pallas-machines/" }
+minicbor = { version="0.11.4", features=["half"] }
+minicbor-io = "0.6.0"
+log = "0.4.14"
+hex = "0.4.3"
+
+[dev-dependencies]
+net2 = "0.2.37"
+env_logger = "0.9.0"
+pallas-handshake = { path = "../pallas-handshake/" }
+pallas-txsubmission = { path = "../pallas-txsubmission/" }

pallas-localstate/examples/chainpoint.rs

@@ -0,0 +1,99 @@
+use minicbor::data::Cbor;
+use pallas_localstate::{OneShotClient, Point, Query};
+use pallas_handshake::n2c::{Client, VersionTable};
+use pallas_handshake::{MAINNET_MAGIC};
+use pallas_machines::{DecodePayload, EncodePayload, run_agent};
+use pallas_multiplexer::Multiplexer;
+use std::net::TcpStream;
+use std::os::unix::net::UnixStream;
+use net2::*;
+
+#[derive(Debug, Clone)]
+struct BlockQuery {}
+
+#[derive(Debug, Clone)]
+enum Request {
+    BlockQuery(BlockQuery),
+    GetSystemStart,
+    GetChainBlockNo,
+    GetChainPoint,
+}
+
+
+impl EncodePayload for Request {
+    fn encode_payload(&self, e: &mut pallas_machines::PayloadEncoder) -> Result<(), Box<dyn std::error::Error>> {
+        match self {
+            Request::BlockQuery(block_query) => {
+                e.u16(0)?;
+                e.array(0)?;
+                Ok(())
+            }
+            Request::GetSystemStart => {
+                e.u16(1)?;
+                Ok(())
+            }
+            Request::GetChainBlockNo => {
+                e.u16(2)?;
+                Ok(())
+            }
+            Request::GetChainPoint => {
+                e.u16(3)?;
+                Ok(())
+            }
+        }
+    }
+}
+
+impl DecodePayload for Request {
+    fn decode_payload(d: &mut pallas_machines::PayloadDecoder) -> Result<Self, Box<dyn std::error::Error>> {
+        todo!()
+    }
+}
+
+#[derive(Debug, Clone)]
+enum Response {
+    Generic(Vec<u8>),
+}
+
+impl EncodePayload for Response {
+    fn encode_payload(&self, e: &mut pallas_machines::PayloadEncoder) -> Result<(), Box<dyn std::error::Error>> {
+        todo!()
+    }
+}
+
+impl DecodePayload for Response {
+    fn decode_payload(d: &mut pallas_machines::PayloadDecoder) -> Result<Self, Box<dyn std::error::Error>> {
+        let cbor: Cbor = d.decode()?;
+        let slice = cbor.as_ref();
+        let vec = slice.to_vec();
+        Ok(Response::Generic(vec))
+    }
+}
+
+#[derive(Debug, Clone)]
+struct ShelleyQuery {}
+
+impl Query for ShelleyQuery {
+    type Request = Request;
+    type Response = Response;
+}
+
+fn main() {
+    env_logger::init();
+
+    // we connect to the unix socket of the local node. Make sure you have the right
+    // path for your environment
+    let bearer = UnixStream::connect("/tmp/node.socket").unwrap();
+
+    let mut muxer = Multiplexer::try_setup(bearer, &vec![0, 7]).unwrap();
+
+    let (rx, tx) = muxer.use_channel(0);
+    let versions = VersionTable::only_v10(MAINNET_MAGIC);
+    let last = run_agent(Client::initial(versions), rx, &tx).unwrap();
+    println!("last hanshake state: {:?}", last);
+
+    let (cs_rx, cs_tx) = muxer.use_channel(7);
+    let cs = OneShotClient::<ShelleyQuery>::initial(None, Request::GetChainPoint);
+    let cs = run_agent(cs, cs_rx, &cs_tx).unwrap();
+    println!("{:?}", cs);
+}

pallas-localstate/src/codec.rs

@@ -0,0 +1,134 @@
+use super::*;
+use pallas_machines::*;
+
+impl EncodePayload for Point {
+    fn encode_payload(&self, e: &mut PayloadEncoder) -> Result<(), Box<dyn std::error::Error>> {
+        e.array(2)?.u64(self.0)?.bytes(&self.1)?;
+        Ok(())
+    }
+}
+
+impl DecodePayload for Point {
+    fn decode_payload(d: &mut PayloadDecoder) -> Result<Self, Box<dyn std::error::Error>> {
+        d.array()?;
+        let slot = d.u64()?;
+        let hash = d.bytes()?;
+
+        Ok(Point(slot, Vec::from(hash)))
+    }
+}
+
+impl EncodePayload for AcquireFailure {
+    fn encode_payload(&self, e: &mut PayloadEncoder) -> Result<(), Box<dyn std::error::Error>> {
+        let code = match self {
+            AcquireFailure::PointTooOld => 0,
+            AcquireFailure::PointNotInChain => 1,
+        };
+
+        e.u16(code)?;
+
+        Ok(())
+    }
+}
+
+impl DecodePayload for AcquireFailure {
+    fn decode_payload(d: &mut PayloadDecoder) -> Result<Self, Box<dyn std::error::Error>> {
+        let code = d.u16()?;
+
+        match code {
+            0 => Ok(AcquireFailure::PointTooOld),
+            1 => Ok(AcquireFailure::PointNotInChain),
+            _ => Err(Box::new(CodecError::UnexpectedCbor("can't infer acquire failure from variant id"))),
+        }
+    }
+}
+
+impl<Q: Query> EncodePayload for Message<Q> {
+    fn encode_payload(&self, e: &mut PayloadEncoder) -> Result<(), Box<dyn std::error::Error>> {
+        match self {
+            Message::Acquire(Some(point)) => {
+                e.array(2)?.u16(0)?;
+                e.encode_payload(point)?;
+                Ok(())
+            }
+            Message::Acquire(None) => {
+                e.array(1)?.u16(8)?;
+                Ok(())
+            }
+            Message::Acquired => {
+                e.array(1)?.u16(1)?;
+                Ok(())
+            }
+            Message::Failure(failure) => {
+                e.array(2)?.u16(2)?;
+                e.encode_payload(failure)?;
+                Ok(())
+            }
+            Message::Query(query) => {
+                e.array(2)?.u16(3)?;
+                e.array(1)?;
+                e.encode_payload(query)?;
+                Ok(())
+            }
+            Message::Result(result) => {
+                e.array(2)?.u16(4)?;
+                e.array(1)?;
+                e.encode_payload(result)?;
+                Ok(())
+            }
+            Message::ReAcquire(Some(point)) => {
+                e.array(2)?.u16(6)?;
+                e.encode_payload(point)?;
+                Ok(())
+            }
+            Message::ReAcquire(None) => {
+                e.array(1)?.u16(9)?;
+                Ok(())
+            }
+            Message::Release => {
+                e.array(1)?.u16(5)?;
+                Ok(())
+            }
+            Message::Done => {
+                e.array(1)?.u16(7)?;
+                Ok(())
+            }
+        }
+    }
+}
+
+impl<Q: Query> DecodePayload for Message<Q> {
+    fn decode_payload(d: &mut PayloadDecoder) -> Result<Self, Box<dyn std::error::Error>> {
+        d.array()?;
+        let label = d.u16()?;
+
+        match label {
+            0 => {
+                let point = d.decode_payload()?;
+                Ok(Message::Acquire(Some(point)))
+            }
+            8 => Ok(Message::Acquire(None)),
+            1 => Ok(Message::Acquired),
+            2 => {
+                let failure = d.decode_payload()?;
+                Ok(Message::Failure(failure))
+            }
+            3 => {
+                let query = d.decode_payload()?;
+                Ok(Message::Query(query))
+            }
+            4 => {
+                let response = d.decode_payload()?;
+                Ok(Message::Result(response))
+            }
+            5 => Ok(Message::Release),
+            6 => {
+                let point = d.decode_payload()?;
+                Ok(Message::ReAcquire(point))
+            }
+            9 => Ok(Message::ReAcquire(None)),
+            7 => Ok(Message::Done),
+            x => Err(Box::new(CodecError::BadLabel(x))),
+        }
+    }
+}

pallas-localstate/src/lib.rs

@@ -0,0 +1,181 @@
+mod codec;
+
+use std::fmt::Debug;
+
+use log::debug;
+
+use pallas_machines::{
+    Agent, DecodePayload, EncodePayload, MachineError, MachineOutput, Transition,
+};
+
+#[derive(Clone)]
+pub struct Point(pub u64, pub Vec<u8>);
+
+impl Debug for Point {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.debug_tuple("Point")
+            .field(&self.0)
+            .field(&hex::encode(&self.1))
+            .finish()
+    }
+}
+
+#[derive(Debug, PartialEq, Clone)]
+pub enum State {
+    Idle,
+    Acquiring,
+    Acquired,
+    Querying,
+    Done,
+}
+
+#[derive(Debug)]
+pub enum AcquireFailure {
+    PointTooOld,
+    PointNotInChain,
+}
+pub trait Query: Debug {
+    type Request: EncodePayload + DecodePayload + Clone + Debug;
+    type Response: EncodePayload + DecodePayload + Clone + Debug;
+}
+
+#[derive(Debug)]
+pub enum Message<Q: Query> {
+    Acquire(Option<Point>),
+    Failure(AcquireFailure),
+    Acquired,
+    Query(Q::Request),
+    Result(Q::Response),
+    ReAcquire(Option<Point>),
+    Release,
+    Done,
+}
+
+pub type Output<QR> = Result<QR, AcquireFailure>;
+
+#[derive(Debug)]
+pub struct OneShotClient<Q: Query> {
+    pub state: State,
+    pub check_point: Option<Point>,
+    pub request: Q::Request,
+    pub output: Option<Output<Q::Response>>,
+}
+
+impl<Q: Query> OneShotClient<Q> {
+    pub fn initial(check_point: Option<Point>, request: Q::Request) -> Self {
+        Self {
+            state: State::Idle,
+            output: None,
+            check_point,
+            request,
+        }
+    }
+
+    fn send_acquire(self, tx: &impl MachineOutput) -> Transition<Self> {
+        let msg = Message::<Q>::Acquire(self.check_point.clone());
+
+        tx.send_msg(&msg)?;
+
+        Ok(Self {
+            state: State::Acquiring,
+            ..self
+        })
+    }
+
+    fn send_query(self, tx: &impl MachineOutput) -> Transition<Self> {
+        let msg = Message::<Q>::Query(self.request.clone());
+
+        tx.send_msg(&msg)?;
+
+        Ok(Self {
+            state: State::Querying,
+            ..self
+        })
+    }
+
+    fn send_release(self, tx: &impl MachineOutput) -> Transition<Self> {
+        let msg = Message::<Q>::Release;
+
+        tx.send_msg(&msg)?;
+
+        Ok(Self {
+            state: State::Idle,
+            ..self
+        })
+    }
+
+    fn on_acquired(self) -> Transition<Self> {
+        debug!("acquired check point for chain state");
+
+        Ok(Self {
+            state: State::Acquired,
+            ..self
+        })
+    }
+
+    fn on_result(self, response: Q::Response) -> Transition<Self> {
+        debug!("query result received: {:?}", response);
+
+        Ok(Self {
+            state: State::Acquired,
+            output: Some(Ok(response)),
+            ..self
+        })
+    }
+
+    fn on_failure(self, failure: AcquireFailure) -> Transition<Self> {
+        debug!("acquire failure: {:?}", failure);
+
+        Ok(Self {
+            state: State::Idle,
+            output: Some(Err(failure)),
+            ..self
+        })
+    }
+
+    fn done(self) -> Transition<Self> {
+        Ok(Self {
+            state: State::Done,
+            ..self
+        })
+    }
+}
+
+impl<Q: Query + 'static> Agent for OneShotClient<Q> {
+    type Message = Message<Q>;
+
+    fn is_done(&self) -> bool {
+        self.state == State::Done
+    }
+
+    fn has_agency(&self) -> bool {
+        match self.state {
+            State::Idle => true,
+            State::Acquired => true,
+            _ => false,
+        }
+    }
+
+    fn send_next(self, tx: &impl MachineOutput) -> Transition<Self> {
+        match (&self.state, &self.output) {
+            // if we're idle and without a result, assume start of flow
+            (State::Idle, None) => self.send_acquire(tx),
+            // if we're idle and with a result, assume end of flow
+            (State::Idle, Some(_)) => self.done(),
+            // if we don't have an output, assume start of query
+            (State::Acquired, None) => self.send_query(tx),
+            // if we have an output but still acquired, release the server
+            (State::Acquired, Some(_)) => self.send_release(tx),
+            _ => panic!("I don't have agency, don't know what to do"),
+        }
+    }
+
+    fn receive_next(self, msg: Self::Message) -> Transition<Self> {
+        match (&self.state, msg) {
+            (State::Acquiring, Message::Acquired) => self.on_acquired(),
+            (State::Acquiring, Message::Failure(failure)) => self.on_failure(failure),
+            (State::Querying, Message::Result(result)) => self.on_result(result),
+            (_, msg) => Err(MachineError::InvalidMsgForState(self.state, msg).into()),
+        }
+    }
+}