choir_actor/
lib.rs

1//! Rivet-backed sequencer: the second implementation of the
2//! actor-runtime seam (DECISIONS.md D3).
3//!
4//! The in-process [`choir_sequencer`] gets its single-writer guarantee
5//! from owning an OS thread; here the same guarantee comes from Rivet's
6//! actor model — one [`SequencerActor`] instance per repo, actions
7//! serialized by the runtime, state persisted by the engine. The seam
8//! contract is behavioral: total order, hash chain, per-client FIFO —
9//! proven by `tests/conformance.rs`, the same properties the in-process
10//! implementation's tests assert.
11//!
12//! [`choir_sequencer`]: ../choir_sequencer/index.html
13//!
14//! # Where this sits
15//!
16//! `docs/architecture.md` is the map of the whole workspace.
17//! This crate is the *second* implementation of the actor-runtime seam (D3), on Rivet; it exists so the seam has two implementations passing one conformance suite.
18//!
19//! It builds on [`choir_oplog`].
20
21use std::{future::Future, pin::Pin, sync::Arc};
22
23use async_trait::async_trait;
24use choir_oplog::{OpEntry, Witness, FORMAT_VERSION};
25use rivetkit::{action, Action, Actor, Ctx, Handles};
26use serde::{Deserialize, Serialize};
27
28type BoxFuture<T> = Pin<Box<dyn Future<Output = anyhow::Result<T>> + Send>>;
29
30/// Persisted actor state: the op log entries in order. Rivet owns
31/// durability; the hash chain (each entry's `parent`) stays verifiable
32/// independently of the runtime, per the D16 one-way-door rule.
33#[derive(Debug, Default, Serialize, Deserialize)]
34pub struct LogState {
35    /// Entries in sequence order.
36    pub entries: Vec<OpEntry>,
37}
38
39/// Submit one op for ordering (the analogue of
40/// `SequencerHandle::try_submit`).
41#[derive(Debug, Serialize, Deserialize)]
42pub struct SubmitOp {
43    /// Signature-covered collaboration channel submitting the op.
44    /// Serialized under the v1 `workspace` name for action compatibility.
45    #[serde(rename = "workspace")]
46    pub channel: String,
47    /// Opaque operation body.
48    pub payload: Vec<u8>,
49    /// Author signature over `(channel, payload)`, if signed.
50    pub author_sig: Option<Witness>,
51}
52
53/// The sequencer's acknowledgement.
54#[derive(Debug, Serialize, Deserialize)]
55pub struct SubmitReply {
56    /// Position assigned in the total order.
57    pub seq: u64,
58    /// Hex content hash of the appended entry (the new head).
59    pub hash_hex: String,
60}
61
62impl Action for SubmitOp {
63    type Output = SubmitReply;
64    const NAME: &'static str = "submit";
65}
66
67/// Fetch the full ordered log (conformance verification; a production
68/// reader would page or subscribe instead).
69#[derive(Debug, Serialize, Deserialize)]
70pub struct DumpLog;
71
72impl Action for DumpLog {
73    type Output = Vec<OpEntry>;
74    const NAME: &'static str = "dump";
75}
76
77/// One repo's sequencer, hosted on the Rivet runtime.
78pub struct SequencerActor;
79
80#[async_trait]
81impl Actor for SequencerActor {
82    type State = LogState;
83    type Input = ();
84    type Actions = (SubmitOp, DumpLog);
85    type Events = ();
86    type Queue = ();
87    type ConnParams = ();
88    type ConnState = ();
89    type Action = action::Raw;
90
91    async fn create_state(_ctx: &Ctx<Self>, _input: Self::Input) -> anyhow::Result<Self::State> {
92        Ok(LogState::default())
93    }
94
95    async fn create(_ctx: &Ctx<Self>) -> anyhow::Result<Self> {
96        Ok(Self)
97    }
98}
99
100impl Handles<SubmitOp> for SequencerActor {
101    type Future = BoxFuture<SubmitReply>;
102
103    fn handle(self: Arc<Self>, ctx: Ctx<Self>, op: SubmitOp) -> Self::Future {
104        Box::pin(async move {
105            let mut state = ctx.state_mut();
106            let parent = state.entries.last().map(OpEntry::content_hash);
107            let seq = state.entries.len() as u64;
108            let entry = OpEntry {
109                format_version: FORMAT_VERSION,
110                parent,
111                seq,
112                channel: op.channel,
113                payload: op.payload,
114                witnesses: Vec::new(),
115                author_sig: op.author_sig,
116            };
117            let hash_hex = entry.content_hash().to_hex();
118            state.entries.push(entry);
119            Ok(SubmitReply { seq, hash_hex })
120        })
121    }
122}
123
124impl Handles<DumpLog> for SequencerActor {
125    type Future = BoxFuture<Vec<OpEntry>>;
126
127    fn handle(self: Arc<Self>, ctx: Ctx<Self>, _op: DumpLog) -> Self::Future {
128        Box::pin(async move { Ok(ctx.state().entries.clone()) })
129    }
130}
131
132/// Registers the sequencer under its canonical actor name.
133pub fn register(registry: &mut rivetkit::Registry) {
134    registry.register_actor::<SequencerActor>("choirSequencer");
135}
choir_actor/lib.rs

choir_actor/
lib.rs
