blockworx_server/
writer.rs

1//! The single-writer task: the only thing that folds, mints revs, and
2//! appends. Everything else — connections, sockets, framing — talks to it
3//! through one channel, so rev assignment and persistence cannot race
4//! without any locking to get wrong.
5
6use ahash::HashMap;
7use blockworx_doc::{
8    commit::CommitEnvelope,
9    protocol::{Nonce, ServerMsg},
10    rev::Rev,
11    session::Host,
12};
13use tokio::sync::{mpsc, oneshot};
14
15use crate::store::Store;
16
17#[derive(Clone, Copy, PartialEq, Eq, Hash, Debug)]
18pub struct ConnectionId(u64);
19
20/// A connection's end of the fanout. Unbounded: a client that stops
21/// reading grows this queue rather than stalling the writer, which would
22/// stall every other client with it. Bounding it is a back-pressure
23/// decision to make when there is a real client to measure.
24pub type Outbox = mpsc::UnboundedSender<ServerMsg>;
25
26pub enum Request {
27    /// The reply carries the id back so the connection can name itself in
28    /// later requests. `Welcome` is pushed to `outbox` from inside the
29    /// writer, which is what makes it impossible for a commit to fall
30    /// between the snapshot of the log and the start of the subscription.
31    Connect {
32        outbox: Outbox,
33        reply: oneshot::Sender<ConnectionId>,
34    },
35    Submit {
36        from: ConnectionId,
37        nonce: Nonce,
38        commit: CommitEnvelope,
39    },
40    Disconnect(ConnectionId),
41}
42
43pub struct Writer {
44    host: Host,
45    store: Store,
46    connections: HashMap<ConnectionId, Outbox>,
47    next_connection: u64,
48}
49
50impl Writer {
51    pub fn new(host: Host, store: Store) -> Self {
52        Self {
53            host,
54            store,
55            connections: HashMap::default(),
56            next_connection: 0,
57        }
58    }
59
60    pub async fn run(mut self, mut requests: mpsc::UnboundedReceiver<Request>) {
61        while let Some(request) = requests.recv().await {
62            match request {
63                Request::Connect { outbox, reply } => self.connect(outbox, reply),
64                Request::Submit {
65                    from,
66                    nonce,
67                    commit,
68                } => self.submit(from, nonce, commit),
69                Request::Disconnect(id) => {
70                    self.connections.remove(&id);
71                }
72            }
73        }
74    }
75
76    fn connect(&mut self, outbox: Outbox, reply: oneshot::Sender<ConnectionId>) {
77        let id = ConnectionId(self.next_connection);
78        self.next_connection += 1;
79
80        let welcome = ServerMsg::Welcome {
81            rev: self.host.rev(),
82            commits: self
83                .host
84                .commits_after(Rev::ZERO)
85                .iter()
86                .cloned()
87                .map(CommitEnvelope::CommitV1)
88                .collect(),
89        };
90        // A send failure means the connection died between the upgrade and
91        // here; dropping the outbox is the whole cleanup.
92        if outbox.send(welcome).is_ok() {
93            self.connections.insert(id, outbox);
94        }
95        let _ = reply.send(id);
96    }
97
98    fn submit(&mut self, from: ConnectionId, nonce: Nonce, envelope: CommitEnvelope) {
99        let CommitEnvelope::CommitV1(commit) = envelope;
100
101        let accepted = match self.host.accept(commit) {
102            Ok(accepted) => accepted,
103            Err(refusal) => {
104                self.send(
105                    from,
106                    &ServerMsg::Rejected {
107                        nonce,
108                        reason: refusal.to_string(),
109                    },
110                );
111                return;
112            }
113        };
114
115        // Durability before visibility: persist, then publish. A storage
116        // failure therefore surfaces as an ordinary rejection — the client
117        // reverts an edit that did not happen — rather than as a commit
118        // the clients hold and the log does not.
119        if let Err(error) = self.store.append(accepted.rev(), accepted.commit()) {
120            tracing::error!("could not append to the log: {error:#}");
121            self.send(
122                from,
123                &ServerMsg::Rejected {
124                    nonce,
125                    reason: "the server could not persist the commit".into(),
126                },
127            );
128            return;
129        }
130
131        let broadcast = CommitEnvelope::CommitV1(accepted.commit().clone());
132        let rev = self.host.publish(accepted);
133        self.send(from, &ServerMsg::Committed { nonce, rev });
134        for (id, outbox) in &self.connections {
135            if *id != from {
136                let _ = outbox.send(ServerMsg::Apply {
137                    rev,
138                    commit: broadcast.clone(),
139                });
140            }
141        }
142    }
143
144    fn send(&self, to: ConnectionId, message: &ServerMsg) {
145        if let Some(outbox) = self.connections.get(&to) {
146            let _ = outbox.send(message.clone());
147        }
148    }
149}
blockworx_server/writer.rs

blockworx_server/
writer.rs
