tlog.rs 90.0%

609 of 677 lines run, 113 of 172 functions · run #494 on main · run missed
//! The hub's transparency log (oxygenhub#238, docs/transparency.md, Oxygen
//! spec §14.8): one entry per signed object the hub accepts, in the order
//! it accepts them, kept as append-only files and served as C2SP
//! tlog-tiles with a signed checkpoint, so anyone can check the hub shows
//! everyone the same history.
//!
//! Files in `<data>/tlog/`:
//! - `entries`: each entry as a big-endian u16 length and its bytes;
//! - `offsets`: each entry's offset in `entries`, a u64 (little-endian);
//! - `hashes`: every hash of the tree in stored order (Oxygen spec §14.8);
//! - `checkpoint`: the latest signed checkpoint note (replaced, not
//!   appended, as it is only a signature over the others).
//!
//! Nothing is rewritten: a crash mid-append leaves at most a tail that is
//! cut back to the last whole entry on open, and the hashes are rebuilt
//! from the entries. The database's `tlog_entries` points from objects to
//! entry numbers.

use crate::{db, Shared};
use axum::extract::{Path as UrlPath, State};
use axum::http::{header, StatusCode};
use axum::response::{IntoResponse, Response};
use oxy_core::hash::{KeyId, ObjectRef};
use oxy_core::identity::Identity;
use oxy_core::repo::{PushReport, Repo};
use oxy_core::tlog::{self, Checkpoint, Entry, Hash, HashReader, Tile, TlogError, TILE_WIDTH};
use std::fs::{File, OpenOptions};
use std::io::Write;
use std::os::unix::fs::FileExt;
use std::path::{Path, PathBuf};
use std::sync::Mutex;

/// How often a new checkpoint is signed while entries arrive.
pub const CHECKPOINT_EVERY: std::time::Duration = std::time::Duration::from_secs(5);

struct Files {
    entries: File,
    offsets: File,
    hashes: File,
    /// Entries in the log.
    size: u64,
    /// Bytes in `entries`.
    entries_len: u64,
}

/// The stored hashes in a file, by stored index.
struct FileHashes<'a>(&'a File);

impl HashReader for FileHashes<'_> {
    fn read_hashes(&self, indexes: &[u64]) -> Result<Vec<Hash>, TlogError> {
        indexes
            .iter()
            .map(|&i| {
                let mut h = [0u8; 32];
                self.0.read_exact_at(&mut h, i * 32).map_err(|e| TlogError::Read(format!("stored hash {i}: {e}")))?;
                Ok(h)
            })
            .collect()
    }
}

/// The log on disk.
pub struct Log {
    dir: PathBuf,
    /// The log's name in its checkpoints: the hub's host and `/log`.
    pub origin: String,
    files: Mutex<Files>,
    /// `append.lock`: held, exclusively, around each append, so several
    /// processes sharing the data directory append one at a time (#274).
    lock: File,
    /// The latest signed checkpoint: its size and its note.
    checkpoint: Mutex<Option<(u64, String)>>,
    /// Witnesses this log asks to cosign its checkpoints.
    witnesses: std::sync::RwLock<Vec<Peer>>,
    /// Logs this hub witnesses: their origins and keys.
    witness_for: std::sync::RwLock<Vec<(String, Identity)>>,
    /// Whether this process writes the log: the web role's (#272). A
    /// process without it only reads, catching up with the writers.
    writer: bool,
}

/// A witness of this hub's log.
#[derive(Clone, Debug)]
pub struct Peer {
    pub name: String,
    /// Its add-checkpoint address.
    pub url: String,
    pub identity: Identity,
}

fn read_u64_at(f: &File, at: u64) -> std::io::Result<u64> {
    let mut b = [0u8; 8];
    f.read_exact_at(&mut b, at)?;
    Ok(u64::from_le_bytes(b))
}

/// The origin of a hub's log: its host (and port) and `/log`.
pub fn origin_of(base_url: &str) -> String {
    let host = base_url.split_once("://").map(|(_, h)| h).unwrap_or(base_url);
    format!("{}/log", host.trim_end_matches('/'))
}

impl Log {
    /// Open (or start) the log in `dir`, recovering from an interrupted
    /// append.
    pub fn open(dir: &Path, origin: &str) -> anyhow::Result<Log> {
        Self::open_as(dir, origin, true)
    }

    /// Open the log to read only, as a process without the web role does
    /// (#272): it neither appends nor signs, and leaves the files as they
    /// are, so it never cuts short an append a writer is making. Several
    /// web processes may share the data directory and write (#274).
    pub fn open_reader(dir: &Path, origin: &str) -> anyhow::Result<Log> {
        Self::open_as(dir, origin, false)
    }

    fn open_as(dir: &Path, origin: &str, writer: bool) -> anyhow::Result<Log> {
        std::fs::create_dir_all(dir)?;
        let open = |name: &str| OpenOptions::new().read(true).append(true).create(true).open(dir.join(name));
        let (entries, offsets, hashes) = (open("entries")?, open("offsets")?, open("hashes")?);
        let lock = OpenOptions::new().read(true).write(true).create(true).truncate(false).open(dir.join("append.lock"))?;
        let mut files = Files { entries, offsets, hashes, size: 0, entries_len: 0 };
        if writer {
            lock.lock()?;
            let done = Self::recover(&mut files, true);
            lock.unlock()?;
            done?;
        } else {
            Self::recover(&mut files, false)?;
        }
        let size = files.size;
        Ok(Log {
            dir: dir.to_path_buf(),
            origin: origin.to_string(),
            files: Mutex::new(files),
            lock,
            checkpoint: Mutex::new(Self::read_checkpoint(dir, size)),
            witnesses: Default::default(),
            witness_for: Default::default(),
            writer,
        })
    }

    /// Read the log's extent from its files, as another process may have
    /// appended since (#274). The whole entries count: those whose offset
    /// and bytes were written, and whose stored hashes all were. A writer,
    /// holding the append lock, also finishes what an append cut short left
    /// (it truncates partial bytes and computes missing hashes again); a
    /// reader changes nothing and takes the whole entries only.
    fn recover(files: &mut Files, writer: bool) -> anyhow::Result<()> {
        let entries_on_disk = files.entries.metadata()?.len();
        let mut size = files.offsets.metadata()?.len() / 8;
        let mut entries_len = 0;
        while size > 0 {
            let off = read_u64_at(&files.offsets, (size - 1) * 8)?;
            let mut len = [0u8; 2];
            if off + 2 <= entries_on_disk && files.entries.read_exact_at(&mut len, off).is_ok() {
                let end = off + 2 + u16::from_be_bytes(len) as u64;
                if end <= entries_on_disk {
                    entries_len = end;
                    break;
                }
            }
            size -= 1;
        }
        // The hashes of every leaf whose hashes were all written; the rest
        // are computed again from the entries.
        let have = files.hashes.metadata()?.len() / 32;
        let mut whole = size;
        while tlog::stored_hash_count(whole) > have {
            whole -= 1;
        }
        if !writer {
            files.size = whole;
            files.entries_len = entries_len;
            return Ok(());
        }
        files.offsets.set_len(size * 8)?;
        files.entries.set_len(entries_len)?;
        files.hashes.set_len(tlog::stored_hash_count(whole) * 32)?;
        files.entries_len = entries_len;
        for n in whole..size {
            let bytes = Self::entry_bytes(files, n)?;
            let add = tlog::stored_hashes_for_leaf(n, tlog::leaf_hash(&bytes), &FileHashes(&files.hashes))?;
            for h in add {
                files.hashes.write_all(&h)?;
            }
            files.size = n + 1;
        }
        files.size = size;
        files.hashes.sync_data()?;
        Ok(())
    }

    /// Catch up with what other processes appended (#274).
    fn catch_up(&self, f: &mut Files) {
        if let Err(e) = Self::recover(f, false) {
            tracing::warn!("tlog: reading the log's extent: {e:#}");
        }
    }

    /// The signed checkpoint kept in `dir`, if it is of at most `size` entries.
    fn read_checkpoint(dir: &Path, size: u64) -> Option<(u64, String)> {
        std::fs::read_to_string(dir.join("checkpoint")).ok().and_then(|note| {
            let n = tlog::Note::parse(&note).ok()?;
            let c = Checkpoint::parse(&n.text).ok()?;
            (c.size <= size).then_some((c.size, note))
        })
    }

    fn entry_bytes(f: &Files, n: u64) -> anyhow::Result<Vec<u8>> {
        let off = read_u64_at(&f.offsets, n * 8)?;
        let mut len = [0u8; 2];
        f.entries.read_exact_at(&mut len, off)?;
        let mut b = vec![0u8; u16::from_be_bytes(len) as usize];
        f.entries.read_exact_at(&mut b, off + 2)?;
        Ok(b)
    }

    /// Entry `n`'s bytes.
    pub fn entry(&self, n: u64) -> anyhow::Result<Vec<u8>> {
        let mut f = self.files.lock().unwrap();
        if n >= f.size {
            self.catch_up(&mut f);
        }
        anyhow::ensure!(n < f.size, "entry {n} is not in the log");
        Self::entry_bytes(&f, n)
    }

    /// Entries in the log.
    pub fn size(&self) -> u64 {
        let mut f = self.files.lock().unwrap();
        self.catch_up(&mut f);
        f.size
    }

    /// Append entries, durably; their numbers.
    pub fn append(&self, entries: &[Entry]) -> anyhow::Result<Vec<u64>> {
        if !self.writer {
            anyhow::bail!("this process does not write the log: it has no web role");
        }
        let mut f = self.files.lock().unwrap();
        // One append at a time across the processes sharing the directory
        // (#274), each from the log as the last left it.
        self.lock.lock()?;
        let appended = Self::recover(&mut f, true).and_then(|()| Self::append_locked(&mut f, entries));
        self.lock.unlock()?;
        appended
    }

    fn append_locked(f: &mut Files, entries: &[Entry]) -> anyhow::Result<Vec<u64>> {
        let mut out = Vec::with_capacity(entries.len());
        for e in entries {
            let bytes = e.encode();
            let n = f.size;
            let add = tlog::stored_hashes_for_leaf(n, tlog::leaf_hash(&bytes), &FileHashes(&f.hashes))?;
            let off = f.entries_len;
            let mut rec = (bytes.len() as u16).to_be_bytes().to_vec();
            rec.extend_from_slice(&bytes);
            f.entries.write_all(&rec)?;
            f.entries_len += rec.len() as u64;
            f.offsets.write_all(&off.to_le_bytes())?;
            let mut hs = Vec::with_capacity(add.len() * 32);
            for h in add {
                hs.extend_from_slice(&h);
            }
            f.hashes.write_all(&hs)?;
            f.size = n + 1;
            out.push(n);
        }
        f.entries.sync_data()?;
        f.offsets.sync_data()?;
        f.hashes.sync_data()?;
        Ok(out)
    }

    /// The root of the first `n` entries.
    pub fn tree_hash(&self, n: u64) -> anyhow::Result<Hash> {
        let f = self.files.lock().unwrap();
        Ok(tlog::tree_hash(n, &FileHashes(&f.hashes))?)
    }

    /// Entry `n`'s inclusion proof in the tree of `t` entries.
    pub fn prove_inclusion(&self, t: u64, n: u64) -> anyhow::Result<Vec<Hash>> {
        let f = self.files.lock().unwrap();
        Ok(tlog::prove_inclusion(t, n, &FileHashes(&f.hashes))?)
    }

    /// The consistency proof from `n` entries to `t`.
    pub fn prove_consistency(&self, t: u64, n: u64) -> anyhow::Result<Vec<Hash>> {
        let f = self.files.lock().unwrap();
        Ok(tlog::prove_consistency(t, n, &FileHashes(&f.hashes))?)
    }

    /// A tile's hashes, when the log holds all of them.
    pub fn tile(&self, t: &Tile) -> anyhow::Result<Option<Vec<u8>>> {
        let f = self.files.lock().unwrap();
        let at_level = f.size >> (tlog::TILE_HEIGHT * t.level);
        if at_level < (t.n << tlog::TILE_HEIGHT) + t.width as u64 {
            return Ok(None);
        }
        Ok(Some(t.read_data(&FileHashes(&f.hashes))?))
    }

    /// The entry bundle beside level-0 tile `t`, when the log holds it.
    pub fn bundle(&self, t: &Tile) -> anyhow::Result<Option<Vec<u8>>> {
        let f = self.files.lock().unwrap();
        let first = t.n << tlog::TILE_HEIGHT;
        if f.size < first + t.width as u64 {
            return Ok(None);
        }
        let entries: Vec<Vec<u8>> = (first..first + t.width as u64).map(|n| Self::entry_bytes(&f, n)).collect::<anyhow::Result<_>>()?;
        Ok(Some(tlog::encode_bundle(&entries)?))
    }

    /// Take `[tlog]` from the configuration: witnesses and witnessed logs.
    pub fn configure(&self, c: &crate::config::Tlog) -> anyhow::Result<()> {
        let key = |b64: &str, what: &str| -> anyhow::Result<Identity> {
            let bytes = oxy_core::dsse::base64_decode(b64.trim()).ok_or_else(|| anyhow::anyhow!("[tlog] {what}: the key is not base64"))?;
            Identity::decode(&bytes).map_err(|e| anyhow::anyhow!("[tlog] {what}: {e}"))
        };
        let mut ws = Vec::new();
        for w in &c.witnesses {
            ws.push(Peer { name: w.name.clone(), url: w.url.clone(), identity: key(&w.key, &w.name)? });
        }
        let mut wf = Vec::new();
        for l in &c.witness_for {
            wf.push((l.origin.clone(), key(&l.key, &l.origin)?));
        }
        self.set_witnesses(ws);
        for (o, k) in wf {
            self.add_witness_for(&o, k);
        }
        Ok(())
    }

    pub fn set_witnesses(&self, peers: Vec<Peer>) {
        *self.witnesses.write().unwrap() = peers;
    }

    pub fn witnesses(&self) -> Vec<Peer> {
        self.witnesses.read().unwrap().clone()
    }

    /// Witness the log named `origin`, signed with `key`.
    pub fn add_witness_for(&self, origin: &str, key: Identity) {
        let mut w = self.witness_for.write().unwrap();
        w.retain(|(o, _)| o != origin);
        w.push((origin.to_string(), key));
    }

    fn witnessed_key(&self, origin: &str) -> Option<Identity> {
        self.witness_for.read().unwrap().iter().find(|(o, _)| o == origin).map(|(_, k)| k.clone())
    }

    /// This hub's name as a witness: its log's host.
    pub fn witness_name(&self) -> &str {
        self.origin.trim_end_matches("/log")
    }

    /// Replace the latest checkpoint's note with `note` (the same
    /// checkpoint, with cosignatures added), unless a newer one was signed.
    fn set_note(&self, size: u64, note: String) -> anyhow::Result<()> {
        let mut cur = self.checkpoint.lock().unwrap();
        if cur.as_ref().map(|(s, _)| *s) != Some(size) {
            return Ok(());
        }
        let tmp = self.dir.join("checkpoint.new");
        std::fs::write(&tmp, &note)?;
        File::open(&tmp)?.sync_all()?;
        std::fs::rename(&tmp, self.dir.join("checkpoint"))?;
        *cur = Some((size, note));
        Ok(())
    }

    /// The latest signed checkpoint note: the newer of this process's and
    /// the one kept in the directory, which the process leading the
    /// checkpointer signs (#274).
    pub fn checkpoint(&self) -> Option<String> {
        let mut held = self.checkpoint.lock().unwrap();
        if let Some(kept) = Self::read_checkpoint(&self.dir, u64::MAX) {
            if held.as_ref().is_none_or(|(s, _)| kept.0 > *s) {
                *held = Some(kept);
            }
        }
        held.as_ref().map(|(_, n)| n.clone())
    }

    /// Sign a checkpoint of the log as it is, unless the last one already
    /// covers it. True when a new one was signed.
    pub fn refresh_checkpoint(&self, identity: &oxy_core::identity::Identity, sk: &oxy_core::identity::SecretKey) -> anyhow::Result<bool> {
        if !self.writer {
            return Ok(false);
        }
        let size = self.size();
        // From the newest kept, whichever process signed it.
        let _ = self.checkpoint();
        if self.checkpoint.lock().unwrap().as_ref().is_some_and(|(s, _)| *s >= size) {
            return Ok(false);
        }
        let c = Checkpoint { origin: self.origin.clone(), size, root: self.tree_hash(size)?, extensions: vec![] };
        let note = c.sign(identity, sk)?.encode();
        let tmp = self.dir.join("checkpoint.new");
        std::fs::write(&tmp, &note)?;
        File::open(&tmp)?.sync_all()?;
        std::fs::rename(&tmp, self.dir.join("checkpoint"))?;
        *self.checkpoint.lock().unwrap() = Some((size, note));
        Ok(true)
    }
}

/// An object to log.
#[derive(Clone, Debug)]
pub struct Pending {
    pub kind: &'static str,
    /// The repository's id (zero for none) and its row on this hub.
    pub repo: [u8; 16],
    pub repo_row: Option<i64>,
    pub object: ObjectRef,
    pub signers: Vec<KeyId>,
}

/// Log what is not logged yet. A failure is reported, never fatal: the
/// object stays accepted (enforcing logging before acceptance comes with
/// inclusion proofs in push responses).
pub async fn log(app: &Shared, items: Vec<Pending>) -> Vec<u64> {
    let mut fresh = Vec::new();
    for p in items {
        let object = p.object.to_hex();
        match db::tlog::entry_of(&app.db, &object, p.kind).await {
            Ok(Some(_)) => continue,
            Ok(None) => {}
            Err(e) => {
                eprintln!("tlog: looking up {} {}: {e}", p.kind, &object[..12]);
                continue;
            }
        }
        if !fresh.iter().any(|q: &Pending| q.kind == p.kind && q.object == p.object) {
            fresh.push(p);
        }
    }
    if fresh.is_empty() {
        return vec![];
    }
    let now = db::now_ms();
    let entries: Vec<Entry> = fresh
        .iter()
        .map(|p| Entry { kind: p.kind.to_string(), repo: p.repo, object: p.object, signers: p.signers.clone(), received: now as u64 })
        .collect();
    let tl = app.tlog.clone();
    let numbers = match tokio::task::spawn_blocking(move || tl.append(&entries)).await {
        Ok(Ok(n)) => n,
        Ok(Err(e)) => {
            eprintln!("tlog: appending {} entries: {e}", fresh.len());
            return vec![];
        }
        Err(e) => {
            eprintln!("tlog: appending: {e}");
            return vec![];
        }
    };
    for (p, n) in fresh.iter().zip(&numbers) {
        if let Err(e) = db::tlog::record(&app.db, *n as i64, &p.object.to_hex(), p.kind, p.repo_row, now).await {
            eprintln!("tlog: indexing entry {n}: {e}");
        }
    }
    numbers
}

fn sig_keys(sigs: &[oxy_core::identity::Signature]) -> Vec<KeyId> {
    let mut keys: Vec<KeyId> = sigs.iter().map(|s| s.key).collect();
    keys.sort();
    keys.dedup();
    keys
}

/// The certificates of short-lived keys among `signers`, to log beside
/// what they signed.
fn certificates(repo: &Repo, rid: [u8; 16], row: Option<i64>, signers: &[KeyId], out: &mut Vec<Pending>) {
    for k in signers {
        if let Ok(Some(c)) = repo.store().certificate_of(k) {
            let issuer = repo.store().get_certificate(&c).map(|sc| vec![sc.cert.issuer]).unwrap_or_default();
            out.push(Pending { kind: "certificate", repo: rid, repo_row: row, object: c, signers: issuer });
        }
    }
}

/// What a push brought that the log keeps: the attestations of the
/// references it moved (and the hub's acceptance of them), tags, seals,
/// the repository's policy, and the certificates of short-lived signers.
pub fn push_entries(repo: &Repo, row: Option<i64>, report: &PushReport) -> Vec<Pending> {
    let Ok(rid) = repo.id() else { return vec![] };
    let mut out = Vec::new();
    let add = |out: &mut Vec<Pending>, kind: &'static str, object: ObjectRef, signers: Vec<KeyId>| {
        certificates(repo, rid, row, &signers, out);
        out.push(Pending { kind, repo: rid, repo_row: row, object, signers });
    };
    for u in &report.updated {
        if !report.results.iter().any(|r| r.ok && r.name == u.name) {
            continue;
        }
        // A tag's withdrawal (#301), and when the hub accepted it.
        if let (Some(tag), true) = (u.name.strip_prefix("tags/"), u.new_state.is_zero()) {
            if let Ok(Some((r, sa))) = repo.tag_withdrawal(tag) {
                add(&mut out, "untag", r, sig_keys(&sa.sigs));
                if let Ok(Some(acc)) = repo.acceptance_of(&r) {
                    add(&mut out, "accepted", acc.attest.hash(), sig_keys(&acc.sigs));
                }
            }
            continue;
        }
        if u.new_state.is_zero() {
            continue;
        }
        if let Some(tag) = u.name.strip_prefix("tags/") {
            if let Ok(Some((r, sa))) = repo.tag_attest(tag) {
                add(&mut out, "tag", r, sig_keys(&sa.sigs));
                // A short-lived key's tag, and when the hub accepted it
                // (oxygen#129).
                if let Ok(Some(acc)) = repo.acceptance_of(&r) {
                    add(&mut out, "accepted", acc.attest.hash(), sig_keys(&acc.sigs));
                }
            }
        } else if let Ok(Some((r, sa))) = repo.latest_ref_attest(&u.name) {
            add(&mut out, "ref", r, sig_keys(&sa.sigs));
            if let Ok(Some(acc)) = repo.acceptance_of(&r) {
                add(&mut out, "accepted", acc.attest.hash(), sig_keys(&acc.sigs));
            }
        }
    }
    let mut seals = std::collections::BTreeSet::new();
    for h in &report.imported {
        for s in repo.store().seals_for(h).unwrap_or_default() {
            seals.insert(s);
        }
    }
    for s in seals {
        let signers = sig_keys(&repo.store().signatures(&s).unwrap_or_default());
        add(&mut out, "seal", s, signers);
    }
    if let Ok(Some((r, sp))) = repo.current_policy() {
        add(&mut out, "policy", r, sig_keys(&sp.sigs));
    }
    out
}

/// Add the log to a refs answer (Oxygen spec §14.8): the latest
/// checkpoint, and for each bookmark's and tag's attestation its entry and
/// inclusion proof against it. An attestation the hub serves but never
/// logged (one pushed before the log began) is logged now, and proved on a
/// later fetch.
pub fn add_log(app: &Shared, row: Option<i64>, refs: &mut oxy_core::protocol::RefsResponse) -> Vec<Pending> {
    let mut backfill = Vec::new();
    let Some(note) = app.tlog.checkpoint() else { return backfill };
    let Some(size) = tlog::Note::parse(&note).ok().and_then(|n| Checkpoint::parse(&n.text).ok()).map(|c| c.size) else {
        return backfill;
    };
    refs.log = Some(note);
    // The witnesses a client may pin, for integrated times (#240).
    refs.witnesses = app.tlog.witnesses().iter().map(|p| (p.name.clone(), p.identity.encode())).collect();
    for r in refs.refs.iter_mut() {
        let Some(a) = &r.attestation else { continue };
        if let Some((logged, integration)) = place(app, size, row, "ref", a, &mut backfill) {
            r.logged = Some(logged);
            r.integration = integration;
        }
    }
    // Tags too (oxygen#129): a short-lived key's tag is checked at its
    // integrated time on every later fetch.
    for t in refs.tags.iter_mut() {
        let Some(a) = &t.attestation else { continue };
        if let Some((logged, integration)) = place(app, size, row, "tag", a, &mut backfill) {
            t.logged = Some(logged);
            t.integration = integration;
        }
    }
    backfill
}

/// Attestation `a`'s entry (of `kind`) in a log of `size` entries, its
/// inclusion proof, and the first cosigned checkpoints that hold it; none
/// when it is not in that checkpoint yet, and queued in `backfill` when it
/// was never logged.
fn place(
    app: &Shared,
    size: u64,
    row: Option<i64>,
    kind: &'static str,
    a: &oxy_core::protocol::SignedAttest,
    backfill: &mut Vec<Pending>,
) -> Option<(oxy_core::protocol::Logged, Vec<oxy_core::protocol::Integration>)> {
    let object = a.attest.hash();
    let idx =
        tokio::task::block_in_place(|| tokio::runtime::Handle::current().block_on(db::tlog::entry_of(&app.db, &object.to_hex(), kind)));
    match idx {
        Ok(Some(i)) if (i as u64) < size => {
            let (Ok(entry), Ok(proof)) = (app.tlog.entry(i as u64), app.tlog.prove_inclusion(size, i as u64)) else { return None };
            // When witnesses first saw it (#240): the first cosigned
            // checkpoints that hold it, each with the entry's proof.
            let notes =
                tokio::task::block_in_place(|| tokio::runtime::Handle::current().block_on(db::tlog::cosigned_notes_after(&app.db, i, 2)))
                    .unwrap_or_default();
            let mut integration = Vec::new();
            for (at, text) in notes {
                if let Ok(proof) = app.tlog.prove_inclusion(at as u64, i as u64) {
                    integration.push(oxy_core::protocol::Integration { note: text, proof });
                }
            }
            Some((oxy_core::protocol::Logged { index: i as u64, entry, proof }, integration))
        }
        Ok(Some(_)) => None,
        Ok(None) => {
            backfill.push(Pending { kind, repo: a.attest.repo, repo_row: row, object, signers: sig_keys(&a.sigs) });
            None
        }
        Err(e) => {
            eprintln!("tlog: looking up {}: {e}", &object.to_hex()[..12]);
            None
        }
    }
}

/// Ask each witness to cosign the latest checkpoint, and serve their
/// cosignatures with it. A witness that answers 409 is told again from the
/// size it says it has; one that refuses is logged.
pub async fn collect(app: &Shared) {
    let peers = app.tlog.witnesses();
    let Some(text) = app.tlog.checkpoint() else { return };
    if peers.is_empty() {
        return;
    }
    let Ok(mut note) = tlog::Note::parse(&text) else { return };
    let Ok(cp) = Checkpoint::parse(&note.text) else { return };
    // What a witness checks: the note as the log signed it.
    let own = tlog::Note { text: note.text.clone(), sigs: note.sigs.iter().filter(|s| s.name == cp.origin).cloned().collect() };
    let mut changed = false;
    for p in peers {
        if note.cosigned_by(&p.name, &p.identity).is_ok() {
            continue;
        }
        let mut old = db::tlog::cosigned_size(&app.db, &p.name).await.unwrap_or(0) as u64;
        for _ in 0..2 {
            let proof = if old == 0 || old >= cp.size { Vec::new() } else { app.tlog.prove_consistency(cp.size, old).unwrap_or_default() };
            let body = tlog::AddCheckpoint { old_size: old, proof, note: own.encode() }.encode();
            let client = match crate::net::client(&p.url, std::slice::from_ref(&p.url), std::time::Duration::from_secs(20)).await {
                Ok(c) => c,
                Err(e) => {
                    eprintln!("tlog: witness {}: {e}", p.name);
                    break;
                }
            };
            let res = match client.post(&p.url).header(header::CONTENT_TYPE, "text/plain").body(body).send().await {
                Ok(r) => r,
                Err(e) => {
                    eprintln!("tlog: witness {}: {e}", p.name);
                    break;
                }
            };
            let status = res.status();
            let answer = res.text().await.unwrap_or_default();
            if status == StatusCode::CONFLICT {
                match answer.trim().parse() {
                    Ok(n) if n != old => {
                        old = n;
                        continue;
                    }
                    _ => {
                        eprintln!("tlog: witness {}: a conflict without a size: {answer}", p.name);
                        break;
                    }
                }
            }
            if !status.is_success() {
                eprintln!("tlog: witness {} refused the checkpoint of {} entries ({status}): {}", p.name, cp.size, answer.trim());
                break;
            }
            let lines = tlog::Note::parse(&format!("{}\n{answer}", note.text)).map(|n| n.sigs).unwrap_or_default();
            let theirs: Vec<_> = lines.into_iter().filter(|s| s.name == p.name).collect();
            let check = tlog::Note { text: note.text.clone(), sigs: theirs.clone() };
            match check.cosigned_by(&p.name, &p.identity) {
                Ok(_) => {
                    note.sigs.extend(theirs);
                    db::tlog::set_cosigned(&app.db, &p.name, cp.size as i64).await.ok();
                    changed = true;
                }
                Err(e) => eprintln!("tlog: witness {} answered a cosignature that does not verify: {e}", p.name),
            }
            break;
        }
    }
    if changed {
        // Kept by size too, for integrated times (#240).
        db::tlog::keep_cosigned_note(&app.db, cp.size as i64, &note.encode()).await.ok();
        if let Err(e) = app.tlog.set_note(cp.size, note.encode()) {
            eprintln!("tlog: keeping the cosigned checkpoint: {e}");
        }
    }
}

/// `POST /log/witness/add-checkpoint` (C2SP tlog-witness): cosign a
/// checkpoint of a log this hub witnesses, if it extends the last one
/// cosigned.
pub async fn add_checkpoint(State(app): State<Shared>, body: String) -> Response {
    let Ok(req) = tlog::AddCheckpoint::parse(&body) else {
        return plain(StatusCode::BAD_REQUEST, "not an add-checkpoint request");
    };
    let Some(origin) = tlog::Note::parse(&req.note).ok().and_then(|n| Checkpoint::parse(&n.text).ok()).map(|c| c.origin) else {
        return plain(StatusCode::BAD_REQUEST, "the checkpoint does not parse");
    };
    let Some(key) = app.tlog.witnessed_key(&origin) else {
        return plain(StatusCode::NOT_FOUND, &format!("this hub does not witness {origin}"));
    };
    let latest = match db::tlog::witnessed(&app.db, &origin).await {
        Ok(l) => l.and_then(|(size, root)| Some((size as u64, ObjectRef::from_hex(&root)?.0))),
        Err(e) => return plain(StatusCode::INTERNAL_SERVER_ERROR, &e.to_string()),
    };
    let sk = match crate::keys::load_secret(&app.config.keys_dir()) {
        Ok((_, sk)) => sk,
        Err(e) => return plain(StatusCode::INTERNAL_SERVER_ERROR, &e.to_string()),
    };
    let me = tlog::WitnessKey { name: app.tlog.witness_name(), identity: &app.hub_key.identity, sk: &sk };
    let now = (db::now_ms() / 1000) as u64;
    let conflict = |n: u64| ([(header::CONTENT_TYPE, "text/x.tlog.size")], format!("{n}\n"));
    match tlog::witness(&req, &origin, &key, latest, &me, now) {
        Ok(tlog::Witnessed::Cosigned { checkpoint, lines }) => {
            let root = ObjectRef(checkpoint.root).to_hex();
            match db::tlog::witness_advance(&app.db, &origin, req.old_size as i64, checkpoint.size as i64, &root).await {
                Ok(true) => {
                    let body: String = tlog::Note { text: String::new(), sigs: lines }.encode().trim_start_matches('\n').to_string();
                    ([(header::CONTENT_TYPE, "text/plain; charset=utf-8")], body).into_response()
                }
                Ok(false) => {
                    let now = db::tlog::witnessed(&app.db, &origin).await.ok().flatten().map(|(s, _)| s as u64).unwrap_or(0);
                    (StatusCode::CONFLICT, conflict(now)).into_response()
                }
                Err(e) => plain(StatusCode::INTERNAL_SERVER_ERROR, &e.to_string()),
            }
        }
        Ok(tlog::Witnessed::Conflict(n)) => (StatusCode::CONFLICT, conflict(n)).into_response(),
        Err(tlog::WitnessError::Unsigned(e)) => plain(StatusCode::FORBIDDEN, &e),
        Err(tlog::WitnessError::Inconsistent(e)) => plain(StatusCode::UNPROCESSABLE_ENTITY, &e),
        Err(tlog::WitnessError::Bad(e)) => plain(StatusCode::BAD_REQUEST, &e),
    }
}

/// Sign a new checkpoint every few seconds while entries arrive.
pub async fn checkpointer(app: Shared) {
    let mut tick = tokio::time::interval(CHECKPOINT_EVERY);
    loop {
        tick.tick().await;
        let (identity, keys) = (app.hub_key.identity.clone(), app.config.keys_dir());
        let tl = app.tlog.clone();
        let done = tokio::task::spawn_blocking(move || -> anyhow::Result<bool> {
            let (_, sk) = crate::keys::load_secret(&keys)?;
            tl.refresh_checkpoint(&identity, &sk)
        })
        .await;
        match done {
            Ok(Err(e)) => eprintln!("tlog: signing a checkpoint: {e}"),
            // A new checkpoint, or one some witness has not cosigned yet.
            _ => collect(&app).await,
        }
    }
}

fn plain(status: StatusCode, why: &str) -> Response {
    (status, [(header::CONTENT_TYPE, "text/plain; charset=utf-8")], format!("{why}\n")).into_response()
}

/// `GET /log/checkpoint`: the latest signed checkpoint (C2SP
/// tlog-checkpoint, a signed note).
pub async fn checkpoint(State(app): State<Shared>) -> Response {
    match app.tlog.checkpoint() {
        Some(note) => ([(header::CONTENT_TYPE, "text/plain; charset=utf-8"), (header::CACHE_CONTROL, "no-cache")], note).into_response(),
        None => plain(StatusCode::SERVICE_UNAVAILABLE, "no checkpoint yet: the log signs one within seconds"),
    }
}

/// `GET /log/witnesses` (#240): this log's witnesses, one `NAME BASE64`
/// line each (the identity object), which `oxy tlog verify` pins.
pub async fn witnesses(State(app): State<Shared>) -> Response {
    let body: String =
        app.tlog.witnesses().iter().map(|p| format!("{} {}\n", p.name, oxy_core::dsse::base64(&p.identity.encode()))).collect();
    ([(header::CONTENT_TYPE, "text/plain; charset=utf-8"), (header::CACHE_CONTROL, "no-cache")], body).into_response()
}

/// `GET /log/tile/{L}/{N}[.p/{W}]` and `/log/tile/entries/{N}[.p/{W}]`:
/// a tile of hashes or an entry bundle (C2SP tlog-tiles). Full tiles never
/// change and are cached for good.
pub async fn tile(State(app): State<Shared>, UrlPath(path): UrlPath<String>) -> Response {
    let (entries, rest) = match path.strip_prefix("entries/") {
        Some(r) => (true, format!("0/{r}")),
        None => (false, path.clone()),
    };
    let Some(t) = Tile::parse(&rest) else {
        return plain(StatusCode::NOT_FOUND, "no such tile");
    };
    let tl = app.tlog.clone();
    let got = tokio::task::spawn_blocking(move || if entries { tl.bundle(&t) } else { tl.tile(&t) }).await;
    match got {
        Ok(Ok(Some(data))) => {
            let cache = if t.width as u64 == TILE_WIDTH { "public, max-age=31536000, immutable" } else { "public, max-age=5" };
            ([(header::CONTENT_TYPE, "application/octet-stream"), (header::CACHE_CONTROL, cache)], data).into_response()
        }
        Ok(Ok(None)) => plain(StatusCode::NOT_FOUND, "the log does not hold that tile yet"),
        Ok(Err(e)) => plain(StatusCode::INTERNAL_SERVER_ERROR, &e.to_string()),
        Err(e) => plain(StatusCode::INTERNAL_SERVER_ERROR, &e.to_string()),
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn a_reader_leaves_the_files_alone_and_cannot_append() {
        let dir = tempfile::tempdir().unwrap();
        let log = Log::open(dir.path(), "hub.example/log").unwrap();
        log.append(&[entry(0), entry(1)]).unwrap();
        // A torn append: an offset whose entry never arrived.
        std::fs::OpenOptions::new().append(true).open(dir.path().join("offsets")).unwrap().write_all(&9999u64.to_le_bytes()).unwrap();
        let before = std::fs::metadata(dir.path().join("offsets")).unwrap().len();
        let reader = Log::open_reader(dir.path(), "hub.example/log").unwrap();
        assert_eq!(reader.size(), 2);
        assert_eq!(std::fs::metadata(dir.path().join("offsets")).unwrap().len(), before, "the writer's files are untouched");
        assert!(reader.append(&[entry(2)]).is_err());
        assert_eq!(reader.tree_hash(2).unwrap(), log.tree_hash(2).unwrap());
    }

    fn entry(i: u64) -> Entry {
        Entry { kind: "ref".into(), repo: [1; 16], object: ObjectRef([i as u8; 32]), signers: vec![], received: i }
    }

    #[test]
    fn the_log_survives_a_torn_append_and_keeps_its_root() {
        let dir = tempfile::tempdir().unwrap();
        let log = Log::open(dir.path(), "hub.example/log").unwrap();
        let first: Vec<Entry> = (0..300).map(entry).collect();
        assert_eq!(log.append(&first).unwrap(), (0..300).collect::<Vec<_>>());
        let root = log.tree_hash(300).unwrap();
        // Leaves alone, as RFC 6962 roots them.
        let mut stored: Vec<Hash> = Vec::new();
        for (n, e) in first.iter().enumerate() {
            stored.extend(tlog::stored_hashes_for_leaf(n as u64, e.leaf_hash(), &stored).unwrap());
        }
        assert_eq!(root, tlog::tree_hash(300, &stored).unwrap());
        drop(log);
        // A crash after the entry and part of the hashes were written.
        let torn = dir.path().join("hashes");
        let len = std::fs::metadata(&torn).unwrap().len();
        OpenOptions::new().write(true).open(&torn).unwrap().set_len(len - 40).unwrap();
        let mut f = OpenOptions::new().append(true).open(dir.path().join("entries")).unwrap();
        f.write_all(&[0, 200, 1, 2]).unwrap(); // a partial entry
        let log = Log::open(dir.path(), "hub.example/log").unwrap();
        assert_eq!(log.size(), 300);
        assert_eq!(log.tree_hash(300).unwrap(), root);
        log.append(&[entry(300)]).unwrap();
        assert_eq!(log.size(), 301);
        let p = log.prove_inclusion(301, 7).unwrap();
        tlog::check_inclusion(&p, 301, &log.tree_hash(301).unwrap(), 7, &first[7].leaf_hash()).unwrap();
        let c = log.prove_consistency(301, 300).unwrap();
        tlog::check_consistency(&c, 301, &log.tree_hash(301).unwrap(), 300, &root).unwrap();
        // Tiles and bundles: the full level-0 tile and the partial one.
        let full = log.tile(&Tile { level: 0, n: 0, width: 256 }).unwrap().unwrap();
        assert_eq!(full.len(), 256 * 32);
        assert!(log.tile(&Tile { level: 0, n: 1, width: 45 }).unwrap().is_some());
        assert!(log.tile(&Tile { level: 0, n: 1, width: 46 }).unwrap().is_none(), "not yet");
        let bundle = log.bundle(&Tile { level: 0, n: 1, width: 45 }).unwrap().unwrap();
        assert_eq!(tlog::decode_bundle(&bundle).unwrap()[0], first[256].encode());
    }

    /// oxygenhub#274: writers sharing the directory (each its own handles,
    /// as separate processes have) append one at a time, each from the log
    /// as the others left it; readers and checkpoints follow.
    #[test]
    fn writers_sharing_the_directory_append_one_at_a_time() {
        let dir = tempfile::tempdir().unwrap();
        let a = std::sync::Arc::new(Log::open(dir.path(), "hub.example/log").unwrap());
        let b = std::sync::Arc::new(Log::open(dir.path(), "hub.example/log").unwrap());
        let reader = Log::open_reader(dir.path(), "hub.example/log").unwrap();
        let spawn = |log: std::sync::Arc<Log>, from: u64| {
            std::thread::spawn(move || {
                let mut got = Vec::new();
                for i in 0..150 {
                    got.extend(log.append(&[entry(from + i)]).unwrap());
                }
                got
            })
        };
        let (ta, tb) = (spawn(a.clone(), 0), spawn(b.clone(), 1000));
        let mut numbers: Vec<u64> = ta.join().unwrap().into_iter().chain(tb.join().unwrap()).collect();
        numbers.sort();
        assert_eq!(numbers, (0..300).collect::<Vec<_>>(), "every entry its own place");
        assert_eq!((a.size(), b.size(), reader.size()), (300, 300, 300));
        let root = a.tree_hash(300).unwrap();
        assert_eq!(b.tree_hash(300).unwrap(), root);
        let fresh = Log::open(dir.path(), "hub.example/log").unwrap();
        assert_eq!(fresh.tree_hash(300).unwrap(), root, "the files are one log");
        assert_eq!(b.entry(299).unwrap(), fresh.entry(299).unwrap());
        assert_eq!(reader.entry(299).unwrap(), fresh.entry(299).unwrap());
        // A checkpoint one signs, the other serves.
        let sk = oxy_core::identity::SecretKey::generate();
        let hub = sk.identity("hub", "[email protected]", 1);
        assert!(a.refresh_checkpoint(&hub, &sk).unwrap());
        assert_eq!(b.checkpoint(), a.checkpoint());
        assert!(!b.refresh_checkpoint(&hub, &sk).unwrap(), "already covered by the other's");
    }

    #[test]
    fn checkpoints_are_signed_when_the_log_grows() {
        let dir = tempfile::tempdir().unwrap();
        let sk = oxy_core::identity::SecretKey::generate();
        let hub = sk.identity("hub", "[email protected]", 1);
        let log = Log::open(dir.path(), &origin_of("https://hub.example:7343/")).unwrap();
        assert_eq!(log.origin, "hub.example:7343/log");
        assert!(log.refresh_checkpoint(&hub, &sk).unwrap(), "the empty log's");
        assert!(!log.refresh_checkpoint(&hub, &sk).unwrap(), "nothing new");
        log.append(&[entry(1), entry(2)]).unwrap();
        assert!(log.refresh_checkpoint(&hub, &sk).unwrap());
        let note = tlog::Note::parse(&log.checkpoint().unwrap()).unwrap();
        let c = Checkpoint::verify(&note, "hub.example:7343/log", &hub).unwrap();
        assert_eq!((c.size, c.root), (2, log.tree_hash(2).unwrap()));
        // Kept across a restart.
        drop(log);
        let log = Log::open(dir.path(), "hub.example:7343/log").unwrap();
        assert_eq!(log.checkpoint().map(|n| n.contains("\n2\n")), Some(true));
    }
}
`/`	Search
`g` `c`	Code
`g` `h`	History
`g` `b`	Bookmarks
`g` `t`	Trust
`y`	Permalink to this state (files and trees)
`?`	This sheet
Coverage / crates/hub/src/tlog.rs 90.0%

Keyboard
