blockworx_server/
lib.rs

1//! The authority: one process that owns the document, sequences every
2//! commit by arrival, and broadcasts the result.
3//!
4//! The concurrency story is one sentence: **one task writes.** It owns the
5//! folded document, the log, and the fanout, and every socket reaches it
6//! through a single channel. That is what makes rev assignment total
7//! without a lock, and what lets `Welcome` be served from the same task
8//! that appends, so no commit can fall between the snapshot of the log and
9//! the start of a subscription.
10
11pub mod store;
12pub mod writer;
13
14use std::path::Path;
15
16use anyhow::{Context, Result};
17use axum::{
18    Router,
19    extract::{
20        State, WebSocketUpgrade,
21        ws::{Message, WebSocket},
22    },
23    response::Response,
24    routing::any,
25};
26use blockworx_doc::{encode, protocol::ClientMsg, session::Host};
27use futures_util::{SinkExt, StreamExt};
28use tokio::{net::TcpListener, sync::mpsc};
29
30use crate::{
31    store::{Store, replay_into},
32    writer::{Request, Writer},
33};
34
35type Writers = mpsc::UnboundedSender<Request>;
36
37/// Open the log, fold it, and start the writer task.
38///
39/// # Errors
40/// The log cannot be opened, or a row will not decode or fold — all hard
41/// startup errors, because a log this build cannot reproduce is one it
42/// must not serve.
43pub fn start(database: &Path) -> Result<Writers> {
44    let store = Store::open(database)?;
45    let mut host = Host::default();
46    replay_into(&store, &mut host).context("replaying the log")?;
47    tracing::info!("replayed to rev {}", host.rev().get());
48
49    let (requests, receiver) = mpsc::unbounded_channel();
50    tokio::spawn(Writer::new(host, store).run(receiver));
51    Ok(requests)
52}
53
54/// # Errors
55/// The listener fails while serving.
56pub async fn serve(listener: TcpListener, writer: Writers) -> Result<()> {
57    let app = Router::new().route("/ws", any(upgrade)).with_state(writer);
58    axum::serve(listener, app).await.context("serving")
59}
60
61async fn upgrade(State(writer): State<Writers>, ws: WebSocketUpgrade) -> Response {
62    ws.on_upgrade(move |socket| connection(socket, writer))
63}
64
65/// One connection: a pump forwarding the writer's messages to the socket,
66/// and this loop forwarding the socket's submissions to the writer.
67async fn connection(socket: WebSocket, writer: Writers) {
68    let (mut sink, mut stream) = socket.split();
69    let (outbox, mut inbox) = mpsc::unbounded_channel();
70    let (reply, id) = tokio::sync::oneshot::channel();
71
72    if writer.send(Request::Connect { outbox, reply }).is_err() {
73        return;
74    }
75    let Ok(id) = id.await else { return };
76
77    let pump = tokio::spawn(async move {
78        while let Some(message) = inbox.recv().await {
79            if sink
80                .send(Message::Binary(encode::to_bytes(&message).into()))
81                .await
82                .is_err()
83            {
84                break;
85            }
86        }
87    });
88
89    while let Some(Ok(message)) = stream.next().await {
90        let Message::Binary(bytes) = message else {
91            continue;
92        };
93        match encode::from_bytes::<ClientMsg>(&bytes) {
94            Ok(ClientMsg::Submit { nonce, commit }) => {
95                if writer
96                    .send(Request::Submit {
97                        from: id,
98                        nonce,
99                        commit,
100                    })
101                    .is_err()
102                {
103                    break;
104                }
105            }
106            // A frame this build cannot read is refused, not skipped: the
107            // client and server disagree about the protocol, and carrying
108            // on would mean guessing which edits were dropped.
109            Err(refusal) => {
110                tracing::warn!("closing a connection that sent an unreadable frame: {refusal}");
111                break;
112            }
113        }
114    }
115
116    let _ = writer.send(Request::Disconnect(id));
117    pump.abort();
118}
blockworx_server/lib.rs

blockworx_server/
lib.rs
