blockworx/storage/

writer.rs

//! The background writer: a thread that takes documents and puts them on disk.
//!
//! It exists because of latency, not throughput. Serializing a document costs
//! well under a millisecond even at the scale-test extreme; the `fsync` behind a
//! durable write costs about 11 ms, which is two thirds of a 60 fps frame (see
//! `todo.md` P0). Autosaving on the UI thread would therefore hitch visibly on
//! every settled edit, however small.
//!
//! Native only. The browser has no threads of this kind — the web writer is a
//! worker the page talks to by message, which arrives with OPFS.

use std::sync::mpsc::{Receiver, Sender, channel};

use super::container::Container;
use super::fs::FsStorage;
use super::history;
use crate::document::Document;

/// One document to write, and the history entry to record it as.
struct Job {
    document: Document,
    seq: u64,
    meta: history::Entry,
}

enum Message {
    Write(Box<Job>),
    /// Report back once everything queued ahead of this has been written.
    Drained(Sender<()>),
}

pub struct Writer {
    tx: Sender<Message>,
    thread: Option<std::thread::JoinHandle<()>>,
}

impl Writer {
    /// Start a writer for the container rooted at `root`.
    pub fn spawn(root: std::path::PathBuf) -> Self {
        let (tx, rx) = channel();
        let thread = std::thread::Builder::new()
            .name("blockworx-writer".to_string())
            .spawn(move || run(&Container::open(&root), &rx))
            .ok();
        Self { tx, thread }
    }

    /// Queue `document` to be written and recorded as entry `seq`. Returns
    /// immediately; the write happens on the writer's thread.
    pub fn write(&self, document: Document, seq: u64, meta: history::Entry) {
        // A closed channel means the writer thread died, which it only does by
        // panicking — already reported there.
        let _ = self.tx.send(Message::Write(Box::new(Job {
            document,
            seq,
            meta,
        })));
    }

    /// Block until everything queued so far has been written.
    ///
    /// The point of the whole arrangement is that saving does not block, so this
    /// is for the one moment where it must: the app is going away and whatever
    /// is still queued would go with it.
    pub fn drain(&self) {
        let (tx, rx) = channel();
        if self.tx.send(Message::Drained(tx)).is_ok() {
            let _ = rx.recv();
        }
    }
}

impl Drop for Writer {
    fn drop(&mut self) {
        self.drain();
        // Dropping the sender ends the loop; joining then guarantees the last
        // write completed before the process is allowed to leave.
        let (dead_tx, _) = channel();
        let _ = std::mem::replace(&mut self.tx, dead_tx);
        if let Some(thread) = self.thread.take() {
            let _ = thread.join();
        }
    }
}

fn run(container: &Container<FsStorage>, rx: &Receiver<Message>) {
    while let Ok(message) = rx.recv() {
        match message {
            Message::Write(job) => {
                if let Err(e) = container.save_and_record(&job.document, job.seq, &job.meta) {
                    tracing::error!("Autosave failed:\n{e:?}");
                }
            }
            Message::Drained(reply) => {
                let _ = reply.send(());
            }
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::storage::Storage as _;
    use crate::storage::atomic::tests::TempDir;
    use crate::storage::container::ROOT;

    fn meta(changed: &[&str]) -> history::Entry {
        history::Entry {
            ts: history::now_millis(),
            command: None,
            changed: changed.iter().map(|s| (*s).to_string()).collect(),
        }
    }

    #[test]
    fn a_queued_document_reaches_the_disk() {
        let dir = TempDir::new("writer-writes");
        let root = dir.join("d.bwx");
        let writer = Writer::spawn(root.clone());

        writer.write(Document::default(), 0, meta(&["b1"]));
        writer.drain();

        assert!(root.join(ROOT).is_file());
        let container = Container::open(&root);
        assert_eq!(history::records(container.storage()).unwrap().len(), 1);
    }

    /// Entries queued in a burst all land, in order — the writer is behind the
    /// UI thread, not instead of it.
    #[test]
    fn a_burst_of_writes_all_land_in_order() {
        let dir = TempDir::new("writer-burst");
        let root = dir.join("d.bwx");
        let writer = Writer::spawn(root.clone());

        for seq in 0..8 {
            writer.write(Document::default(), seq, meta(&[&format!("b{seq}")]));
        }
        writer.drain();

        let container = Container::open(&root);
        let records = history::records(container.storage()).unwrap();
        let seqs: Vec<u64> = records.iter().map(|r| r.seq).collect();
        assert_eq!(seqs, (0..8).collect::<Vec<_>>());
        for (seq, record) in records.iter().enumerate() {
            let meta = record.meta.as_ref().expect("a sidecar");
            assert_eq!(meta.changed, vec![format!("b{seq}")]);
        }
    }

    /// What `drain` is for: dropping the writer must not abandon queued work.
    #[test]
    fn dropping_the_writer_finishes_what_was_queued() {
        let dir = TempDir::new("writer-drop");
        let root = dir.join("d.bwx");
        {
            let writer = Writer::spawn(root.clone());
            for seq in 0..4 {
                writer.write(Document::default(), seq, meta(&[]));
            }
            // No explicit drain: the drop has to do it.
        }
        let container = Container::open(&root);
        assert_eq!(history::records(container.storage()).unwrap().len(), 4);
        assert!(container.storage().exists(ROOT));
    }
}