8f8ba256a8
The release's Windows leg failed to compile joule-moq-relay: the DVR read and wrote its segments through std::os::unix::fs::FileExt. On Windows the same two calls now run over seek_read and seek_write. wai-remote-host would have failed next on its Unix file modes; they now apply on Unix only, as the room's keys already do. Co-Authored-By: Claude Opus 5.5 <[email protected]>
2132 lines
98 KiB
Rust
2132 lines
98 KiB
Rust
//! The DVR store: a window of a room's tracks, with their receipts, kept on disk and served
|
|
//! to FETCH below the live edge (WAI `live-recording.md` §8).
|
|
//!
|
|
//! **What it keeps.** Every track whose namespace a `[[dvr.window]]` rule matches (the rule
|
|
//! with the longest prefix): media tracks, their receipts tracks, events tracks, catalogs
|
|
//! and takes alike. A group is stored once the cache holds it whole (`crate::cache`): its
|
|
//! objects as they arrived, with every field a FETCH carries (Subgroup ID or datagram,
|
|
//! publisher priority, properties, status, payload). Group *g* of a receipts track is the
|
|
//! receipts of media group *g* (WAI `moq-streaming-format.md` §4.1), so a viewer that
|
|
//! fetches both checks a recorded object exactly as it checks a live one.
|
|
//!
|
|
//! **The window.** Each rule has an age (`max_age_s`, at most 24 hours) and a byte budget
|
|
//! (`max_bytes`) over its tracks together. Past either, the oldest group goes first, whole,
|
|
//! and group *g* of a media track goes with group *g* of its receipts track. A group at or
|
|
//! below a track's eviction mark is no longer known to the relay: a FETCH that reaches it
|
|
//! is told so with an End of Unknown Range (MOQT §3.2.2, §11.4.1.2), since an evicted
|
|
//! object existed, and an unmarked gap would say it did not.
|
|
//!
|
|
//! **On disk.** One directory per track under `tracks/`, named by the first 16 bytes of the
|
|
//! SHA-256 of its rendered name (MOQT §8.8) in hexadecimal, holding `TRACK` (its namespace,
|
|
//! name and properties, with a SHA-256 check) and its **segments**, `s<n, 20 digits>.wds`:
|
|
//! files a group record is appended to as each group is stored, a new one begun every ten
|
|
//! seconds or 16 MiB. A segment is a sequence of frames, `len(u32) state(u8) record`, the
|
|
//! state 1 for a record that stands and 0 for one a later record replaced; a record is
|
|
//!
|
|
//! ```text
|
|
//! header "WDG1" 1 track(32) group(u64) arrived_us(u64) count(u32) erased(u32) check(32)
|
|
//! object id(u64) subgroup(u64, 2^64-1 for a datagram) priority(u8) status(u8)
|
|
//! props_len(u32) payload_len(u32) digest(32) check(32) props payload
|
|
//! erased id(u64) digest(32) why(u8: 1 redacted, 2 did not verify, 3 taken down)
|
|
//! ```
|
|
//!
|
|
//! Integers are little-endian. `track` is the SHA-256 of the rendered track; a header's
|
|
//! `check` is SHA-256 over the header's bytes before it; an object's `digest` is the content
|
|
//! hash its receipt names it by (WAI `jwp-receipts.md` §2.1), and its `check` SHA-256 over
|
|
//! its fixed fields before `check`, its properties and its payload. A record that replaces
|
|
//! another (a group written again: an object erased, or a subgroup that came late) is
|
|
//! appended, and the old frame's state set to 0 and its record's bytes overwritten with
|
|
//! zeros. A segment is deleted once no group's standing record is in it.
|
|
//!
|
|
//! **Durability.** The writer syncs every segment it wrote, and the directories it made
|
|
//! files in, at least once a second and before it answers a flush or stops; a shutdown
|
|
//! flushes. A crash loses at most the last second's groups. On start the store reads every
|
|
//! segment again: a frame a crash cut short, at a segment's end, is cut off and counted
|
|
//! (`torn_writes`); a record whose header does not check is dropped; an object whose
|
|
//! `check` fails, or, on a track with a receipts track, whose payload does not verify
|
|
//! against group *g*'s receipts (the group receipt's signature under the door's trust for
|
|
//! the namespace, the object's receipt at its position, its Merkle proof to the signed
|
|
//! root), or, on an events track, a record of which does not verify under its principal, is
|
|
//! dropped and counted, and its group written again with the object among the erased. It is
|
|
//! never served: a FETCH that reaches it is told End of Unknown Range. Each object's digest
|
|
//! is checked again as it is read.
|
|
//!
|
|
//! **Moderation.** The relay reads the `decision` events (WAI `live-events.md` §4.11) on the
|
|
//! events tracks it already forwards. A decision is honoured when its record verifies, its
|
|
//! author owns the track it rides and is a moderator the relay knows: a room key it is
|
|
//! configured with (`dvr.room_keys`, the keys of `auth.authorities`), or the principal of a
|
|
//! grant with the moderate power its publishing session proved. The last honoured decision
|
|
//! on a target stands (WAI `live-moderation.md` §4.5): deny and redact withhold the target
|
|
//! from every DVR read, and from every FETCH the cache answers; redact also erases its bytes
|
|
//! from the store (an event record from the object that carries it, the object's other
|
|
//! records kept). Allow restores what deny withheld. The honoured decisions are kept in
|
|
//! `decisions.wdl` and survive a restart.
|
|
//!
|
|
//! **A room whose objects are sealed** (WAI `confidentiality.md`: an end-to-end encrypted
|
|
//! room) is never kept, since such a room is never recorded and a relay keeps nothing of its
|
|
//! tracks beyond delivery (§9). The store knows one by its first sealed object: an object
|
|
//! whose properties carry the secure-objects Key ID (§4.3), which a relay reads without a key.
|
|
//! A group holding one is never written. From then on the store writes nothing of the room,
|
|
//! deletes what it already held (the room's events and receipts tracks, in the clear, can be
|
|
//! whole before its first media group is), and drops the room's decisions from
|
|
//! `decisions.wdl`; it still honours them for what the cache serves. The room is recorded in
|
|
//! `sealed.wdl`, `room(32) check(32)` per room with `check` the SHA-256 of the room, so a
|
|
//! restart keeps refusing it.
|
|
//!
|
|
//! **A room in which a minor publishes** (WAI `age-assurance.md` §8.4) is never kept either.
|
|
//! The relay knows one by a minor's publishing grant: a session whose SETUP proved a grant of
|
|
//! an authority it trusts whose `publish` is not 0 and whose attribute `minor` is `true`
|
|
//! (WAI `live-rooms.md` §4.4). From then on the store treats the room as it treats a sealed
|
|
//! one: it writes nothing of it, deletes what it held, keeps none of its decisions, and
|
|
//! records it in `sealed.wdl`, which so holds every room the store never keeps.
|
|
//!
|
|
//! **Takedowns** (WAI `takedown.md` §10.1). A signed takedown (`crate::takedown`) is honoured
|
|
//! when its signer is a key the store trusts (`dvr.takedown_keys`, and the keys of
|
|
//! `auth.authorities`) and its signature verifies; it rides as an object of the track
|
|
//! `[key, takedown]`, `0`, whose key is the signer's, or is given to [`Dvr::takedown`]. The
|
|
//! store then forgets every covered object, as it forgets a sealed room, whatever its room: an
|
|
//! object whose content hash its `exact` set or its `matched` set names, and every object of a
|
|
//! track its `exact` set names (by the track's digest, WAI live-rooms §6.2) from the group it
|
|
//! names on. Each is erased from every group that holds it (`why` 3), never stored again (a
|
|
//! group that becomes whole later stores it erased), and never served, to a moderator's
|
|
//! session either. The takedowns are kept in `takedowns.wdl`, `len(u32) record check(32)` per
|
|
//! takedown, and verified again on start, which makes again any erasure a crash left undone.
|
|
//! Once an erasure is on disk the store signs a purge with the relay's key and writes it to
|
|
//! `purges/<link>.json`, the takedown's link hash in hexadecimal, once.
|
|
|
|
use crate::auth::pattern_matches;
|
|
use crate::cache::CachedObject;
|
|
use crate::config::DvrConfig;
|
|
use crate::events::{self, Event, REDACT};
|
|
use crate::relay::TrackKey;
|
|
use bytes::Bytes;
|
|
use ed25519_dalek::SigningKey;
|
|
use joule_moq::carriage::ReceiptBundle;
|
|
use joule_moq::draft22::data::ObjectStatus;
|
|
use joule_moq::draft22::kvp::{prop, Properties};
|
|
use joule_moq::draft22::names::render;
|
|
use joule_moq::draft22::Location;
|
|
use joule_moq::receipt::{ObjectReceipt, ReceiptVerifier};
|
|
use joule_moq::secure_objects::{self, Refusal};
|
|
use joule_moq::{GroupId, ObjectId, TrackName, TrackNamespace};
|
|
use std::collections::{BTreeMap, BTreeSet, HashMap, HashSet};
|
|
use std::io::Write;
|
|
#[cfg(unix)]
|
|
use std::os::unix::fs::FileExt;
|
|
#[cfg(windows)]
|
|
use windows_at::FileExt;
|
|
use std::path::{Path, PathBuf};
|
|
use std::sync::atomic::{AtomicU64, Ordering};
|
|
use std::sync::{mpsc, Arc, Mutex};
|
|
use std::time::{Duration, Instant};
|
|
|
|
const GROUP_MAGIC: &[u8; 4] = b"WDG1";
|
|
const TRACK_MAGIC: &[u8; 4] = b"WDT1";
|
|
const VERSION: u8 = 1;
|
|
const HEADER_LEN: usize = 4 + 1 + 32 + 8 + 8 + 4 + 4 + 32;
|
|
const OBJECT_FIXED: usize = 8 + 8 + 1 + 1 + 4 + 4 + 32 + 32;
|
|
/// A frame's length and state, before its record.
|
|
const FRAME: u64 = 5;
|
|
const DATAGRAM: u64 = u64::MAX;
|
|
/// How long a segment takes groups, and how large it grows, before another is begun.
|
|
const SEGMENT_SPAN_US: u64 = 10_000_000;
|
|
const SEGMENT_MAX: u64 = 16 << 20;
|
|
/// How often the writer syncs what it wrote.
|
|
const SYNC_EVERY: Duration = Duration::from_secs(1);
|
|
/// Why an object is among a group's erased.
|
|
pub const ERASED_REDACTED: u8 = 1;
|
|
pub const ERASED_UNVERIFIED: u8 = 2;
|
|
pub const ERASED_TAKEN_DOWN: u8 = 3;
|
|
/// The resolution of the store's metering series.
|
|
const POINT_US: u64 = 1_000_000;
|
|
/// How long the metering series keeps its points.
|
|
const SERIES_KEEP_US: u64 = 48 * 3_600 * 1_000_000;
|
|
|
|
fn sha256(parts: &[&[u8]]) -> [u8; 32] {
|
|
let mut c = ring::digest::Context::new(&ring::digest::SHA256);
|
|
for p in parts {
|
|
c.update(p);
|
|
}
|
|
c.finish().as_ref().try_into().expect("32 bytes")
|
|
}
|
|
|
|
pub fn now_us() -> u64 {
|
|
std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).map_or(0, |d| d.as_micros() as u64)
|
|
}
|
|
|
|
/// The room a namespace is a track of (WAI live-rooms §6.1): its first field, 64 lowercase
|
|
/// hexadecimal digits.
|
|
pub fn room_of(ns: &TrackNamespace) -> Option<[u8; 32]> {
|
|
let f = ns.0.first()?;
|
|
if f.len() != 64 || !f.iter().all(|b| b.is_ascii_digit() || (b'a'..=b'f').contains(b)) {
|
|
return None;
|
|
}
|
|
hex::decode(f).ok()?.try_into().ok()
|
|
}
|
|
|
|
/// A key a namespace field names, 64 lowercase hexadecimal digits.
|
|
fn key_field(ns: &TrackNamespace, i: usize) -> Option<[u8; 32]> {
|
|
let f = ns.0.get(i)?;
|
|
if f.len() != 64 {
|
|
return None;
|
|
}
|
|
hex::decode(f).ok()?.try_into().ok()
|
|
}
|
|
|
|
/// Whether a track is a room's events track (live-events §8.1): `[R, P, events]`, `0`.
|
|
pub fn is_events_track(key: &TrackKey) -> bool {
|
|
key.0 .0.len() == 3 && key.0 .0[2] == b"events" && key.1 .0 == b"0" && room_of(&key.0).is_some()
|
|
}
|
|
|
|
fn track_digest(key: &TrackKey) -> [u8; 32] {
|
|
sha256(&[render(&key.0, Some(&key.1)).as_bytes()])
|
|
}
|
|
|
|
/// A track's digest as a takedown names it (WAI live-rooms §6.2): SHA-256 over the MOQT §8.8
|
|
/// rendering of its namespace, one zero byte, and its name.
|
|
pub fn room_track_digest(key: &TrackKey) -> [u8; 32] {
|
|
sha256(&[render(&key.0, None).as_bytes(), &[0], &key.1 .0])
|
|
}
|
|
|
|
/// Whether a track is one takedowns ride (WAI takedown §10.1): `[K, takedown]`, `0`, with `K`
|
|
/// a key in 64 lowercase hexadecimal digits. The key, if so.
|
|
pub fn takedown_track_key(key: &TrackKey) -> Option<[u8; 32]> {
|
|
(key.0 .0.len() == 2 && key.0 .0[1] == crate::takedown::TRACK_FIELD && key.1 .0 == b"0").then(|| room_of(&key.0)).flatten()
|
|
}
|
|
|
|
// ── The index ──────────────────────────────────────────────────────────────
|
|
|
|
#[derive(Clone, Debug)]
|
|
struct DObject {
|
|
id: u64,
|
|
subgroup: Option<u64>,
|
|
priority: u8,
|
|
status: ObjectStatus,
|
|
/// Where its properties begin in its segment; its payload follows them.
|
|
at: u64,
|
|
props_len: u32,
|
|
payload_len: u32,
|
|
digest: [u8; 32],
|
|
/// The digests of its event records, on an events track.
|
|
records: Vec<[u8; 32]>,
|
|
}
|
|
|
|
#[derive(Clone, Debug)]
|
|
struct DGroup {
|
|
arrived_us: u64,
|
|
/// Payload bytes of its objects.
|
|
bytes: u64,
|
|
objects: Vec<DObject>,
|
|
/// Objects no longer held, by Object ID, with their digests.
|
|
erased: Vec<(u64, [u8; 32], u8)>,
|
|
/// Its standing record: the segment, and the frame's place and record length.
|
|
seg: u64,
|
|
frame_at: u64,
|
|
record_len: u32,
|
|
}
|
|
|
|
#[derive(Debug)]
|
|
struct DTrack {
|
|
dir: PathBuf,
|
|
window: usize,
|
|
room: Option<[u8; 32]>,
|
|
events: bool,
|
|
props: Properties,
|
|
groups: BTreeMap<u64, DGroup>,
|
|
/// Groups known not to exist (a Prior Group ID Gap, MOQT §10.8), `lo ..= hi` by `lo`.
|
|
absent: BTreeMap<u64, u64>,
|
|
/// Groups at or below this are no longer known to the store.
|
|
evicted_through: Option<u64>,
|
|
end_of_track: Option<Location>,
|
|
/// Each segment, with the groups whose standing record is in it.
|
|
segs: BTreeMap<u64, usize>,
|
|
}
|
|
|
|
impl DTrack {
|
|
fn largest(&self) -> Option<Location> {
|
|
let (g, grp) = self.groups.iter().next_back()?;
|
|
let o = grp.objects.iter().map(|o| o.id).chain(grp.erased.iter().map(|e| e.0)).max()?;
|
|
Some(Location::new(*g, o))
|
|
}
|
|
}
|
|
|
|
/// The store's metering of one room (WAI live-metering §3.9): what it wrote, what it holds
|
|
/// over time, what it dropped. Points are kept each second a change falls in.
|
|
#[derive(Clone, Debug, Default)]
|
|
struct Series {
|
|
t_last: u64,
|
|
held: u64,
|
|
acc: u128,
|
|
stored_objects: u64,
|
|
stored_bytes: u64,
|
|
dropped: u64,
|
|
points: Vec<(u64, u128, u64, u64, u64, u64)>,
|
|
}
|
|
|
|
impl Series {
|
|
fn change(&mut self, now: u64, held: i128, objects: u64, bytes: u64, dropped: u64) {
|
|
if self.t_last != 0 && now > self.t_last {
|
|
self.acc += self.held as u128 * (now - self.t_last) as u128;
|
|
}
|
|
self.t_last = self.t_last.max(now);
|
|
self.held = (self.held as i128 + held).max(0) as u64;
|
|
self.stored_objects += objects;
|
|
self.stored_bytes += bytes;
|
|
self.dropped += dropped;
|
|
let p = (self.t_last, self.acc, self.held, self.stored_objects, self.stored_bytes, self.dropped);
|
|
match self.points.last_mut() {
|
|
Some(last) if last.0 / POINT_US == p.0 / POINT_US => *last = p,
|
|
_ => self.points.push(p),
|
|
}
|
|
if self.points.first().is_some_and(|f| f.0 + SERIES_KEEP_US < now) {
|
|
self.points.retain(|x| x.0 + SERIES_KEEP_US >= now);
|
|
}
|
|
}
|
|
|
|
/// (byte-microseconds held, objects stored, bytes stored, dropped) up to `x`.
|
|
fn at(&self, x: u64) -> (u128, u64, u64, u64) {
|
|
if self.t_last != 0 && x >= self.t_last {
|
|
return (self.acc + self.held as u128 * (x - self.t_last) as u128, self.stored_objects, self.stored_bytes, self.dropped);
|
|
}
|
|
match self.points.partition_point(|p| p.0 <= x) {
|
|
0 => (0, 0, 0, 0),
|
|
i => {
|
|
let p = self.points[i - 1];
|
|
(p.1 + p.2 as u128 * (x - p.0) as u128, p.3, p.4, p.5)
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
/// What the store did for one room over a span (WAI live-metering §3.9).
|
|
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
|
|
pub struct RoomFigures {
|
|
pub objects_stored: u64,
|
|
pub bytes_stored: u64,
|
|
/// Payload bytes held, integrated over the span, in byte-seconds, rounded down.
|
|
pub byte_s: u64,
|
|
/// Objects dropped because they did not verify against their receipts.
|
|
pub dropped: u64,
|
|
}
|
|
|
|
#[derive(Default)]
|
|
struct Index {
|
|
tracks: HashMap<TrackKey, DTrack>,
|
|
window_bytes: Vec<u64>,
|
|
/// Where each digest is held: (track, group, Object ID).
|
|
by_digest: HashMap<[u8; 32], Vec<(TrackKey, u64, u64)>>,
|
|
/// The standing decision on each (room, target): its verdict.
|
|
standing: HashMap<([u8; 32], [u8; 32]), u8>,
|
|
decided: HashSet<[u8; 32]>,
|
|
series: HashMap<[u8; 32], Series>,
|
|
/// Rooms whose objects are sealed: nothing of them is kept.
|
|
sealed: HashSet<[u8; 32]>,
|
|
/// The takedowns honoured, by link hash; the objects they cover, by content hash; and
|
|
/// the tracks they cover, by digest, each from a group on.
|
|
takedowns: HashSet<[u8; 32]>,
|
|
taken: HashSet<[u8; 32]>,
|
|
taken_tracks: Vec<([u8; 32], u64)>,
|
|
}
|
|
|
|
impl Index {
|
|
fn withheld(&self, room: Option<[u8; 32]>, digest: &[u8; 32]) -> bool {
|
|
room.is_some_and(|r| self.standing.get(&(r, *digest)).is_some_and(|v| *v != events::ALLOW))
|
|
}
|
|
|
|
/// Whether a takedown covers group `group` of `key`, by its track.
|
|
fn track_gone(&self, key: &TrackKey, group: u64) -> bool {
|
|
if self.taken_tracks.is_empty() {
|
|
return false;
|
|
}
|
|
let d = room_track_digest(key);
|
|
self.taken_tracks.iter().any(|(t, from)| *t == d && group >= *from)
|
|
}
|
|
|
|
/// Record a takedown honoured. False when it was already.
|
|
fn take(&mut self, t: &crate::takedown::Takedown) -> bool {
|
|
if !self.takedowns.insert(t.link) {
|
|
return false;
|
|
}
|
|
self.taken.extend(t.objects.iter().copied());
|
|
for x in &t.tracks {
|
|
if !self.taken_tracks.contains(x) {
|
|
self.taken_tracks.push(*x);
|
|
}
|
|
}
|
|
true
|
|
}
|
|
|
|
fn unindex(&mut self, key: &TrackKey, g: u64, o: &DObject) {
|
|
for d in std::iter::once(&o.digest).chain(o.records.iter()) {
|
|
if let Some(v) = self.by_digest.get_mut(d) {
|
|
v.retain(|x| !(x.0 == *key && x.1 == g && x.2 == o.id));
|
|
if v.is_empty() {
|
|
self.by_digest.remove(d);
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
fn reindex(&mut self, key: &TrackKey, g: u64, o: &DObject) {
|
|
for d in std::iter::once(&o.digest).chain(o.records.iter()) {
|
|
let e = self.by_digest.entry(*d).or_default();
|
|
if !e.iter().any(|x| x.0 == *key && x.1 == g && x.2 == o.id) {
|
|
e.push((key.clone(), g, o.id));
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
/// Counters across the store.
|
|
#[derive(Default)]
|
|
pub struct DvrStats {
|
|
pub groups_written: AtomicU64,
|
|
pub objects_written: AtomicU64,
|
|
pub groups_evicted: AtomicU64,
|
|
/// Groups that became whole after the window had moved past them (their receipts or
|
|
/// media group already evicted): not stored.
|
|
pub groups_late: AtomicU64,
|
|
/// Objects dropped because their check or their receipt did not verify.
|
|
pub objects_dropped: AtomicU64,
|
|
/// Frames a crash cut short at a segment's end, cut off at start.
|
|
pub torn_writes: AtomicU64,
|
|
/// Objects erased by a redaction.
|
|
pub objects_redacted: AtomicU64,
|
|
pub decisions: AtomicU64,
|
|
/// Segments synced, and the syncs that wrote them out.
|
|
pub syncs: AtomicU64,
|
|
/// Takedowns honoured, objects they erased, and purges signed.
|
|
pub takedowns: AtomicU64,
|
|
pub objects_taken_down: AtomicU64,
|
|
pub purges_signed: AtomicU64,
|
|
}
|
|
|
|
/// A snapshot of [`DvrStats`].
|
|
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
|
|
pub struct DvrCounts {
|
|
pub groups_written: u64,
|
|
pub objects_written: u64,
|
|
pub groups_evicted: u64,
|
|
pub groups_late: u64,
|
|
pub objects_dropped: u64,
|
|
pub torn_writes: u64,
|
|
pub objects_redacted: u64,
|
|
pub decisions: u64,
|
|
pub syncs: u64,
|
|
pub groups_held: u64,
|
|
pub bytes_held: u64,
|
|
pub segments: u64,
|
|
/// Rooms whose objects are sealed, of which the store keeps nothing.
|
|
pub rooms_sealed: u64,
|
|
pub takedowns: u64,
|
|
pub objects_taken_down: u64,
|
|
pub purges_signed: u64,
|
|
}
|
|
|
|
/// One object read from the store, or a place it can no longer serve.
|
|
#[derive(Clone, Debug)]
|
|
pub enum Slot {
|
|
Object(StoredObject),
|
|
/// No longer held, or withheld: its Object ID.
|
|
Unknown(u64),
|
|
}
|
|
|
|
#[derive(Clone, Debug)]
|
|
pub struct StoredObject {
|
|
pub id: u64,
|
|
pub subgroup: Option<u64>,
|
|
pub priority: u8,
|
|
pub status: ObjectStatus,
|
|
pub properties: Properties,
|
|
pub payload: Bytes,
|
|
}
|
|
|
|
/// What the store holds of a track over a span of groups, for a FETCH's plan.
|
|
#[derive(Clone, Debug, Default)]
|
|
pub struct Snapshot {
|
|
pub held: BTreeSet<u64>,
|
|
pub absent: Vec<(u64, u64)>,
|
|
pub evicted_through: Option<u64>,
|
|
pub largest: Option<Location>,
|
|
pub end_of_track: Option<Location>,
|
|
pub props: Properties,
|
|
}
|
|
|
|
impl Snapshot {
|
|
/// Whether group `g` is known not to exist, and the last group of its run if so.
|
|
pub fn absent_through(&self, g: u64) -> Option<u64> {
|
|
self.absent.iter().filter(|(lo, hi)| *lo <= g && g <= *hi).map(|(_, hi)| *hi).max()
|
|
}
|
|
}
|
|
|
|
enum Job {
|
|
Seal { key: TrackKey, props: Properties, group: u64, arrived_us: u64, objects: Vec<((u64, u64), CachedObject)> },
|
|
Erase { room: [u8; 32], digest: [u8; 32] },
|
|
Log(Vec<u8>),
|
|
Forget([u8; 32]),
|
|
TakeDown { t: crate::takedown::Takedown, rec: Vec<u8> },
|
|
Flush(mpsc::Sender<()>),
|
|
}
|
|
|
|
struct Window {
|
|
prefix: TrackNamespace,
|
|
max_age_us: u64,
|
|
max_bytes: u64,
|
|
}
|
|
|
|
/// The segment a track appends to.
|
|
struct OpenSeg {
|
|
seq: u64,
|
|
file: std::fs::File,
|
|
started_us: u64,
|
|
len: u64,
|
|
}
|
|
|
|
/// What the writer alone holds: each track's open segment, and what it has to sync.
|
|
#[derive(Default)]
|
|
struct Writer {
|
|
open: HashMap<TrackKey, OpenSeg>,
|
|
next: HashMap<TrackKey, u64>,
|
|
dirty: HashSet<PathBuf>,
|
|
dirs: HashSet<PathBuf>,
|
|
last_sync: Option<Instant>,
|
|
}
|
|
|
|
pub struct Dvr {
|
|
root: PathBuf,
|
|
windows: Vec<Window>,
|
|
suffix: Vec<u8>,
|
|
room_keys: Vec<[u8; 32]>,
|
|
/// The keys whose takedowns the store honours.
|
|
takedown_keys: Vec<[u8; 32]>,
|
|
/// The key the store signs its purges with: the relay's.
|
|
purge_key: Option<SigningKey>,
|
|
index: Mutex<Index>,
|
|
tx: Mutex<Option<mpsc::Sender<Job>>>,
|
|
worker: Mutex<Option<std::thread::JoinHandle<()>>>,
|
|
pub stats: DvrStats,
|
|
}
|
|
|
|
fn put_short_field(o: &mut Vec<u8>, b: &[u8]) {
|
|
o.extend_from_slice(&(b.len() as u16).to_le_bytes());
|
|
o.extend_from_slice(b);
|
|
}
|
|
|
|
fn track_dir_name(key: &TrackKey) -> String {
|
|
hex::encode(&track_digest(key)[..16])
|
|
}
|
|
|
|
fn seg_path(dir: &Path, seq: u64) -> PathBuf {
|
|
dir.join(format!("s{seq:020}.wds"))
|
|
}
|
|
|
|
fn seg_seq(path: &Path) -> Option<u64> {
|
|
let n = path.file_name()?.to_str()?;
|
|
n.strip_prefix('s')?.strip_suffix(".wds")?.parse().ok()
|
|
}
|
|
|
|
fn sync_dir(dir: &Path) {
|
|
if let Ok(f) = std::fs::File::open(dir) {
|
|
let _ = f.sync_all();
|
|
}
|
|
}
|
|
|
|
/// Write `bytes` to `path` durably: a temporary file, synced, renamed into place, and the
|
|
/// directory synced.
|
|
fn write_durably(path: &Path, bytes: &[u8]) -> std::io::Result<()> {
|
|
let tmp = path.with_extension("tmp");
|
|
{
|
|
let mut f = std::fs::File::create(&tmp)?;
|
|
f.write_all(bytes)?;
|
|
f.sync_all()?;
|
|
}
|
|
std::fs::rename(&tmp, path)?;
|
|
if let Some(d) = path.parent() {
|
|
sync_dir(d);
|
|
}
|
|
Ok(())
|
|
}
|
|
|
|
fn status_code(s: ObjectStatus) -> u8 {
|
|
s.code() as u8
|
|
}
|
|
|
|
/// The bytes of a group record (module documentation), and its objects' index, their
|
|
/// places counted from the record's start.
|
|
fn encode_group(track_digest: &[u8; 32], group: u64, arrived_us: u64, objects: &[(DObject, Bytes, Bytes)], erased: &[(u64, [u8; 32], u8)]) -> (Vec<u8>, Vec<DObject>) {
|
|
let mut o = Vec::new();
|
|
o.extend_from_slice(GROUP_MAGIC);
|
|
o.push(VERSION);
|
|
o.extend_from_slice(track_digest);
|
|
o.extend_from_slice(&group.to_le_bytes());
|
|
o.extend_from_slice(&arrived_us.to_le_bytes());
|
|
o.extend_from_slice(&(objects.len() as u32).to_le_bytes());
|
|
o.extend_from_slice(&(erased.len() as u32).to_le_bytes());
|
|
let check = sha256(&[&o]);
|
|
o.extend_from_slice(&check);
|
|
let mut index = Vec::with_capacity(objects.len());
|
|
for (d, props, payload) in objects {
|
|
let mut fixed = Vec::with_capacity(OBJECT_FIXED);
|
|
fixed.extend_from_slice(&d.id.to_le_bytes());
|
|
fixed.extend_from_slice(&d.subgroup.unwrap_or(DATAGRAM).to_le_bytes());
|
|
fixed.push(d.priority);
|
|
fixed.push(status_code(d.status));
|
|
fixed.extend_from_slice(&(props.len() as u32).to_le_bytes());
|
|
fixed.extend_from_slice(&(payload.len() as u32).to_le_bytes());
|
|
fixed.extend_from_slice(&d.digest);
|
|
let check = sha256(&[&fixed, props, payload]);
|
|
o.extend_from_slice(&fixed);
|
|
o.extend_from_slice(&check);
|
|
let at = o.len() as u64;
|
|
o.extend_from_slice(props);
|
|
o.extend_from_slice(payload);
|
|
index.push(DObject { at, props_len: props.len() as u32, payload_len: payload.len() as u32, ..d.clone() });
|
|
}
|
|
for (id, digest, why) in erased {
|
|
o.extend_from_slice(&id.to_le_bytes());
|
|
o.extend_from_slice(digest);
|
|
o.push(*why);
|
|
}
|
|
(o, index)
|
|
}
|
|
|
|
/// A group record that read: its header facts, each object with its properties and payload
|
|
/// and whether its `check` held (places counted from the record's start), and its erased
|
|
/// entries.
|
|
struct Decoded {
|
|
group: u64,
|
|
arrived_us: u64,
|
|
count: u32,
|
|
objects: Vec<(DObject, Bytes, Bytes, bool)>,
|
|
erased: Vec<(u64, [u8; 32], u8)>,
|
|
}
|
|
|
|
/// Read a group record. `None` when its header does not check; objects past a truncation
|
|
/// are left out, and with them the erased entries.
|
|
fn decode_group(b: &[u8], track_digest: &[u8; 32]) -> Option<Decoded> {
|
|
if b.len() < HEADER_LEN || &b[..4] != GROUP_MAGIC || b[4] != VERSION || &b[5..37] != track_digest {
|
|
return None;
|
|
}
|
|
if sha256(&[&b[..HEADER_LEN - 32]])[..] != b[HEADER_LEN - 32..HEADER_LEN] {
|
|
return None;
|
|
}
|
|
let u64_at = |i: usize| u64::from_le_bytes(b[i..i + 8].try_into().expect("8"));
|
|
let u32_at = |i: usize| u32::from_le_bytes(b[i..i + 4].try_into().expect("4"));
|
|
let (group, arrived_us, count, erased_n) = (u64_at(37), u64_at(45), u32_at(53), u32_at(57) as usize);
|
|
let mut p = HEADER_LEN;
|
|
let mut objects = Vec::new();
|
|
for _ in 0..count {
|
|
if b.len() < p + OBJECT_FIXED {
|
|
break;
|
|
}
|
|
let f = &b[p..p + OBJECT_FIXED];
|
|
let id = u64::from_le_bytes(f[..8].try_into().expect("8"));
|
|
let sg = u64::from_le_bytes(f[8..16].try_into().expect("8"));
|
|
let priority = f[16];
|
|
let status = ObjectStatus::from_code(f[17] as u64).unwrap_or(ObjectStatus::Normal);
|
|
let props_len = u32::from_le_bytes(f[18..22].try_into().expect("4"));
|
|
let payload_len = u32::from_le_bytes(f[22..26].try_into().expect("4"));
|
|
let digest: [u8; 32] = f[26..58].try_into().expect("32");
|
|
let at = p + OBJECT_FIXED;
|
|
let end = at + props_len as usize + payload_len as usize;
|
|
if b.len() < end {
|
|
break;
|
|
}
|
|
let props = Bytes::copy_from_slice(&b[at..at + props_len as usize]);
|
|
let payload = Bytes::copy_from_slice(&b[at + props_len as usize..end]);
|
|
let ok = sha256(&[&f[..58], &props, &payload])[..] == f[58..90];
|
|
objects.push((DObject { id, subgroup: (sg != DATAGRAM).then_some(sg), priority, status, at: at as u64, props_len, payload_len, digest, records: Vec::new() }, props, payload, ok));
|
|
p = end;
|
|
}
|
|
let mut erased = Vec::new();
|
|
if objects.len() == count as usize {
|
|
for _ in 0..erased_n {
|
|
if b.len() < p + 41 {
|
|
break;
|
|
}
|
|
erased.push((u64::from_le_bytes(b[p..p + 8].try_into().expect("8")), b[p + 8..p + 40].try_into().expect("32"), b[p + 40]));
|
|
p += 41;
|
|
}
|
|
}
|
|
Some(Decoded { group, arrived_us, count, objects, erased })
|
|
}
|
|
|
|
fn encode_track(key: &TrackKey, props: &Properties) -> Vec<u8> {
|
|
let mut o = TRACK_MAGIC.to_vec();
|
|
o.push(VERSION);
|
|
o.push(key.0 .0.len() as u8);
|
|
for f in &key.0 .0 {
|
|
put_short_field(&mut o, f);
|
|
}
|
|
put_short_field(&mut o, &key.1 .0);
|
|
let p = props.to_bytes().unwrap_or_default();
|
|
o.extend_from_slice(&(p.len() as u32).to_le_bytes());
|
|
o.extend_from_slice(&p);
|
|
let check = sha256(&[&o]);
|
|
o.extend_from_slice(&check);
|
|
o
|
|
}
|
|
|
|
fn decode_track(b: &[u8]) -> Option<(TrackKey, Properties)> {
|
|
if b.len() < 4 + 1 + 1 + 32 || &b[..4] != TRACK_MAGIC || b[4] != VERSION {
|
|
return None;
|
|
}
|
|
let (body, check) = b.split_at(b.len() - 32);
|
|
if sha256(&[body])[..] != *check {
|
|
return None;
|
|
}
|
|
let mut p = 6;
|
|
let mut field = || -> Option<Vec<u8>> {
|
|
let n = u16::from_le_bytes(body.get(p..p + 2)?.try_into().ok()?) as usize;
|
|
let v = body.get(p + 2..p + 2 + n)?.to_vec();
|
|
p += 2 + n;
|
|
Some(v)
|
|
};
|
|
let mut fields = Vec::new();
|
|
for _ in 0..body[5] {
|
|
fields.push(field()?);
|
|
}
|
|
let name = field()?;
|
|
let n = u32::from_le_bytes(body.get(p..p + 4)?.try_into().ok()?) as usize;
|
|
let props = Properties::decode(body.get(p + 4..p + 4 + n)?, "dvr track properties").ok()?;
|
|
if p + 4 + n != body.len() {
|
|
return None;
|
|
}
|
|
Some(((TrackNamespace(fields), TrackName(name)), props))
|
|
}
|
|
|
|
/// The Prior Group ID Gap an object's properties carry (MOQT §10.8): the groups before it
|
|
/// that do not exist.
|
|
/// Whether an object may be sealed (WAI confidentiality §4.3): its properties carry a Key
|
|
/// ID, or do not read (Immutable Properties that do not decode could hide one). Only an
|
|
/// object whose properties read and carry none is in the clear.
|
|
pub fn is_sealed(props: &Properties) -> bool {
|
|
!matches!(secure_objects::key_id_of(props), Err(Refusal::KeyIdMissing))
|
|
}
|
|
|
|
fn prior_groups(props: &Properties, group: u64) -> Option<(u64, u64)> {
|
|
let n = props.find(prop::PRIOR_GROUP_ID_GAP).ok().flatten().and_then(|k| k.as_varint())?;
|
|
(n > 0 && n <= group).then(|| (group - n, group - 1))
|
|
}
|
|
|
|
fn read_at(path: &Path, at: u64, len: usize) -> std::io::Result<Vec<u8>> {
|
|
let f = std::fs::File::open(path)?;
|
|
let mut b = vec![0u8; len];
|
|
f.read_exact_at(&mut b, at)?;
|
|
Ok(b)
|
|
}
|
|
|
|
/// Unix's `read_exact_at` and `write_all_at`, over Windows' `seek_read` and `seek_write`.
|
|
/// These move the file's cursor, which nothing here reads: every segment write is positional.
|
|
#[cfg(windows)]
|
|
mod windows_at {
|
|
use std::io::{Error, ErrorKind, Result};
|
|
use std::os::windows::fs::FileExt as _;
|
|
|
|
pub trait FileExt {
|
|
fn read_exact_at(&self, buf: &mut [u8], at: u64) -> Result<()>;
|
|
fn write_all_at(&self, buf: &[u8], at: u64) -> Result<()>;
|
|
}
|
|
|
|
impl FileExt for std::fs::File {
|
|
fn read_exact_at(&self, mut buf: &mut [u8], mut at: u64) -> Result<()> {
|
|
while !buf.is_empty() {
|
|
match self.seek_read(buf, at) {
|
|
Ok(0) => return Err(Error::from(ErrorKind::UnexpectedEof)),
|
|
Ok(n) => {
|
|
buf = &mut buf[n..];
|
|
at += n as u64;
|
|
}
|
|
Err(e) if e.kind() == ErrorKind::Interrupted => {}
|
|
Err(e) => return Err(e),
|
|
}
|
|
}
|
|
Ok(())
|
|
}
|
|
|
|
fn write_all_at(&self, mut buf: &[u8], mut at: u64) -> Result<()> {
|
|
while !buf.is_empty() {
|
|
match self.seek_write(buf, at) {
|
|
Ok(0) => return Err(Error::from(ErrorKind::WriteZero)),
|
|
Ok(n) => {
|
|
buf = &buf[n..];
|
|
at += n as u64;
|
|
}
|
|
Err(e) if e.kind() == ErrorKind::Interrupted => {}
|
|
Err(e) => return Err(e),
|
|
}
|
|
}
|
|
Ok(())
|
|
}
|
|
}
|
|
}
|
|
|
|
/// A group's record held in memory to be written again: each object with its properties
|
|
/// and payload.
|
|
type Rows = Vec<(DObject, Bytes, Bytes)>;
|
|
|
|
impl Dvr {
|
|
/// Open the store at `cfg.path`: read every track and segment in it, check every object
|
|
/// again, and start its writer. `verifier` gives the receipts verifier of a namespace;
|
|
/// `authorities` are the keys of `auth.authorities`, room keys besides `cfg.room_keys` and
|
|
/// takedown keys besides `cfg.takedown_keys`; `purge_key` signs the store's purges (none
|
|
/// are signed without it).
|
|
pub fn open(cfg: &DvrConfig, suffix: &str, authorities: Vec<[u8; 32]>, purge_key: Option<SigningKey>, verifier: impl Fn(&TrackNamespace) -> ReceiptVerifier) -> Result<Arc<Dvr>, String> {
|
|
cfg.check()?;
|
|
let root = cfg.path.clone();
|
|
std::fs::create_dir_all(root.join("tracks")).map_err(|e| format!("dvr: {}: {e}", root.display()))?;
|
|
let mut room_keys: Vec<[u8; 32]> = cfg.room_keys.iter().filter_map(|k| hex::decode(k).ok()?.try_into().ok()).collect();
|
|
let mut takedown_keys: Vec<[u8; 32]> = cfg.takedown_keys.iter().filter_map(|k| hex::decode(k).ok()?.try_into().ok()).collect();
|
|
takedown_keys.extend(authorities.iter().copied());
|
|
room_keys.extend(authorities);
|
|
let windows = cfg
|
|
.window
|
|
.iter()
|
|
.map(|w| Window { prefix: TrackNamespace::from_strs(w.prefix.iter().map(String::as_str)), max_age_us: w.max_age_s * 1_000_000, max_bytes: w.max_bytes })
|
|
.collect::<Vec<_>>();
|
|
let dvr = Arc::new(Dvr {
|
|
root,
|
|
index: Mutex::new(Index { window_bytes: vec![0; windows.len()], ..Index::default() }),
|
|
windows,
|
|
suffix: suffix.as_bytes().to_vec(),
|
|
room_keys,
|
|
takedown_keys,
|
|
purge_key,
|
|
tx: Mutex::new(None),
|
|
worker: Mutex::new(None),
|
|
stats: DvrStats::default(),
|
|
});
|
|
let mut w = Writer::default();
|
|
dvr.load(&mut w, &verifier)?;
|
|
let (tx, rx) = mpsc::channel();
|
|
*dvr.tx.lock().unwrap() = Some(tx);
|
|
let me = dvr.clone();
|
|
let worker = std::thread::Builder::new().name("joule-moq-relay-dvr".into()).spawn(move || me.run(w, rx)).map_err(|e| e.to_string())?;
|
|
*dvr.worker.lock().unwrap() = Some(worker);
|
|
Ok(dvr)
|
|
}
|
|
|
|
/// The window that keeps `ns`: the rule with the longest prefix that matches it.
|
|
pub fn window_of(&self, ns: &TrackNamespace) -> Option<usize> {
|
|
self.windows.iter().enumerate().filter(|(_, w)| pattern_matches(&w.prefix, ns)).max_by_key(|(_, w)| w.prefix.0.len()).map(|(i, _)| i)
|
|
}
|
|
|
|
/// Whether a window rule matches `ns`. Moderation reads the decisions of every events
|
|
/// track it matches, a sealed room's among them.
|
|
pub fn keeps(&self, ns: &TrackNamespace) -> bool {
|
|
self.window_of(ns).is_some()
|
|
}
|
|
|
|
/// Whether the store writes the tracks of `ns`: a rule matches, and it is no room whose
|
|
/// objects are sealed and no room in which a minor publishes.
|
|
pub fn stores(&self, ns: &TrackNamespace) -> bool {
|
|
self.keeps(ns) && !room_of(ns).is_some_and(|r| self.index.lock().unwrap().sealed.contains(&r))
|
|
}
|
|
|
|
/// Room `room`'s objects are sealed: from now the store keeps nothing of it, and the
|
|
/// writer deletes what it held.
|
|
fn seal_room(&self, room: [u8; 32]) {
|
|
if self.index.lock().unwrap().sealed.insert(room) {
|
|
tracing::info!(target: "joule_moq_relay::dvr", room = %hex::encode(room), "a room's objects are sealed: the store keeps nothing of it");
|
|
self.send(Job::Forget(room));
|
|
}
|
|
}
|
|
|
|
/// A minor publishes in room `room` (WAI age-assurance §8.4): from now the store keeps
|
|
/// nothing of it, as of a room whose objects are sealed, and the writer deletes what it held.
|
|
pub fn minors_room(&self, room: [u8; 32]) {
|
|
if self.index.lock().unwrap().sealed.insert(room) {
|
|
tracing::info!(target: "joule_moq_relay::dvr", room = %hex::encode(room), "a minor publishes in a room: the store keeps nothing of it");
|
|
self.send(Job::Forget(room));
|
|
}
|
|
}
|
|
|
|
fn send(&self, j: Job) {
|
|
if let Some(tx) = self.tx.lock().unwrap().as_ref() {
|
|
let _ = tx.send(j);
|
|
}
|
|
}
|
|
|
|
/// Wait until every write asked for so far is on disk, synced.
|
|
pub fn flush(&self) {
|
|
let (tx, rx) = mpsc::channel();
|
|
self.send(Job::Flush(tx));
|
|
let _ = rx.recv_timeout(Duration::from_secs(60));
|
|
}
|
|
|
|
/// Flush, then stop the writer. The store is closed; a new one can open its path.
|
|
pub fn close(&self) {
|
|
self.flush();
|
|
self.tx.lock().unwrap().take();
|
|
if let Some(w) = self.worker.lock().unwrap().take() {
|
|
let _ = w.join();
|
|
}
|
|
}
|
|
|
|
/// Group `group` of `key` is whole in the cache: store it, unless it is stored with as
|
|
/// many objects already. Cheap: the write is the writer's.
|
|
pub fn seal(&self, key: &TrackKey, props: &Properties, group: u64, arrived_us: u64, objects: Vec<((u64, u64), CachedObject)>) {
|
|
if objects.is_empty() || !self.stores(&key.0) {
|
|
return;
|
|
}
|
|
// A sealed object is never written, and its room is never kept again
|
|
// (confidentiality §9).
|
|
if objects.iter().any(|(_, o)| is_sealed(&o.properties)) {
|
|
if let Some(room) = room_of(&key.0) {
|
|
self.seal_room(room);
|
|
}
|
|
return;
|
|
}
|
|
{
|
|
let ix = self.index.lock().unwrap();
|
|
if let Some(t) = ix.tracks.get(key) {
|
|
if t.evicted_through.is_some_and(|e| group <= e) {
|
|
self.stats.groups_late.fetch_add(1, Ordering::Relaxed);
|
|
return;
|
|
}
|
|
if t.groups.get(&group).is_some_and(|g| g.objects.len() + g.erased.len() >= objects.len()) {
|
|
return;
|
|
}
|
|
}
|
|
}
|
|
self.send(Job::Seal { key: key.clone(), props: props.clone(), group, arrived_us, objects });
|
|
}
|
|
|
|
/// What the store holds of `key` from group `lo` to `hi`.
|
|
pub fn snapshot(&self, key: &TrackKey, lo: u64, hi: u64) -> Option<Snapshot> {
|
|
let ix = self.index.lock().unwrap();
|
|
let t = ix.tracks.get(key)?;
|
|
Some(Snapshot {
|
|
held: t.groups.range(lo..=hi).map(|(g, _)| *g).collect(),
|
|
absent: t.absent.iter().filter(|(l, h)| **h >= lo && **l <= hi).map(|(l, h)| (*l, *h)).collect(),
|
|
evicted_through: t.evicted_through,
|
|
largest: t.largest(),
|
|
end_of_track: t.end_of_track,
|
|
props: t.props.clone(),
|
|
})
|
|
}
|
|
|
|
/// The Largest Object the store holds of `key`.
|
|
pub fn largest(&self, key: &TrackKey) -> Option<Location> {
|
|
self.index.lock().unwrap().tracks.get(key).and_then(DTrack::largest)
|
|
}
|
|
|
|
/// The track's properties as the store recorded them.
|
|
pub fn props(&self, key: &TrackKey) -> Option<Properties> {
|
|
self.index.lock().unwrap().tracks.get(key).map(|t| t.props.clone())
|
|
}
|
|
|
|
/// Whether any decision withholds something in `room`.
|
|
pub fn withholds_in(&self, room: Option<[u8; 32]>) -> bool {
|
|
room.is_some_and(|r| self.index.lock().unwrap().standing.iter().any(|((rr, _), v)| *rr == r && *v != events::ALLOW))
|
|
}
|
|
|
|
/// Whether the standing decision on `digest` in `room` withholds it.
|
|
pub fn withheld(&self, room: Option<[u8; 32]>, digest: &[u8; 32]) -> bool {
|
|
self.index.lock().unwrap().withheld(room, digest)
|
|
}
|
|
|
|
/// The objects of group `group` of `key` from `from` to `to` (Object IDs, inclusive),
|
|
/// read from disk, in Object ID order, with what is no longer held or withheld marked.
|
|
/// An events-track object some of whose records are withheld comes back without them
|
|
/// (`crate::events::withhold`). With `withhold` false (a moderator's session, WAI
|
|
/// live-moderation §6.3) nothing is withheld; what is erased stays unknown. `None` when
|
|
/// the store no longer holds the group.
|
|
pub fn read_group(&self, key: &TrackKey, group: u64, from: u64, to: u64, withhold: bool) -> Option<Vec<Slot>> {
|
|
// A group written again between the index and the read (an erasure) is read again.
|
|
for attempt in 0..2 {
|
|
let (path, place, objs, erased, room, is_events) = {
|
|
let ix = self.index.lock().unwrap();
|
|
let t = ix.tracks.get(key)?;
|
|
let g = t.groups.get(&group)?;
|
|
(seg_path(&t.dir, g.seg), (g.seg, g.frame_at), g.objects.clone(), g.erased.clone(), t.room.filter(|_| withhold), t.events)
|
|
};
|
|
let Ok(f) = std::fs::File::open(&path) else {
|
|
if attempt == 0 {
|
|
continue;
|
|
}
|
|
return None;
|
|
};
|
|
let mut out: Vec<(u64, Slot)> = erased.iter().filter(|e| e.0 >= from && e.0 <= to).map(|e| (e.0, Slot::Unknown(e.0))).collect();
|
|
let mut moved = false;
|
|
// What a takedown covers is never served, to any session (WAI takedown §10.1).
|
|
let gone = self.index.lock().unwrap().track_gone(key, group);
|
|
for o in objs.iter().filter(|o| o.id >= from && o.id <= to) {
|
|
if gone || self.index.lock().unwrap().taken.contains(&o.digest) {
|
|
out.push((o.id, Slot::Unknown(o.id)));
|
|
continue;
|
|
}
|
|
let mut b = vec![0u8; o.props_len as usize + o.payload_len as usize];
|
|
let payload = match f.read_exact_at(&mut b, o.at) {
|
|
Ok(()) => Bytes::from(b.split_off(o.props_len as usize)),
|
|
Err(_) => Bytes::new(),
|
|
};
|
|
if ObjectReceipt::hash_payload(&payload) != o.digest {
|
|
moved = true;
|
|
out.push((o.id, Slot::Unknown(o.id)));
|
|
continue;
|
|
}
|
|
let ix = self.index.lock().unwrap();
|
|
if ix.withheld(room, &o.digest) {
|
|
out.push((o.id, Slot::Unknown(o.id)));
|
|
continue;
|
|
}
|
|
let payload = if is_events && o.records.iter().any(|d| ix.withheld(room, d)) {
|
|
match events::withhold(&payload, |d| ix.withheld(room, d)) {
|
|
Some(p) => Bytes::from(p.into_owned()),
|
|
None => {
|
|
out.push((o.id, Slot::Unknown(o.id)));
|
|
continue;
|
|
}
|
|
}
|
|
} else {
|
|
payload
|
|
};
|
|
drop(ix);
|
|
let properties = if b.is_empty() { Properties::new() } else { Properties::decode(&b, "dvr object properties").unwrap_or_default() };
|
|
out.push((o.id, Slot::Object(StoredObject { id: o.id, subgroup: o.subgroup, priority: o.priority, status: o.status, properties, payload })));
|
|
}
|
|
if moved {
|
|
// Written again since the index was read: read it where it stands now.
|
|
let now = self.index.lock().unwrap().tracks.get(key).and_then(|t| t.groups.get(&group)).map(|g| (g.seg, g.frame_at));
|
|
if attempt == 0 && now.is_some() && now != Some(place) {
|
|
continue;
|
|
}
|
|
tracing::warn!(target: "joule_moq_relay::dvr", file = %path.display(), group, "a stored object whose bytes no longer hold its digest: not served");
|
|
}
|
|
out.sort_by_key(|(id, _)| *id);
|
|
return Some(out.into_iter().map(|(_, s)| s).collect());
|
|
}
|
|
None
|
|
}
|
|
|
|
/// What the store did for `room` between two instants (WAI live-metering §3.9).
|
|
pub fn figures(&self, room: [u8; 32], from_us: u64, to_us: u64) -> RoomFigures {
|
|
let ix = self.index.lock().unwrap();
|
|
let Some(s) = ix.series.get(&room) else { return RoomFigures::default() };
|
|
let (a0, o0, b0, d0) = s.at(from_us);
|
|
let (a1, o1, b1, d1) = s.at(to_us.max(from_us));
|
|
RoomFigures { objects_stored: o1 - o0, bytes_stored: b1 - b0, byte_s: ((a1 - a0) / 1_000_000) as u64, dropped: d1 - d0 }
|
|
}
|
|
|
|
pub fn counts(&self) -> DvrCounts {
|
|
let s = &self.stats;
|
|
let ix = self.index.lock().unwrap();
|
|
DvrCounts {
|
|
groups_written: s.groups_written.load(Ordering::Relaxed),
|
|
objects_written: s.objects_written.load(Ordering::Relaxed),
|
|
groups_evicted: s.groups_evicted.load(Ordering::Relaxed),
|
|
groups_late: s.groups_late.load(Ordering::Relaxed),
|
|
objects_dropped: s.objects_dropped.load(Ordering::Relaxed),
|
|
torn_writes: s.torn_writes.load(Ordering::Relaxed),
|
|
objects_redacted: s.objects_redacted.load(Ordering::Relaxed),
|
|
decisions: s.decisions.load(Ordering::Relaxed),
|
|
syncs: s.syncs.load(Ordering::Relaxed),
|
|
groups_held: ix.tracks.values().map(|t| t.groups.len() as u64).sum(),
|
|
bytes_held: ix.window_bytes.iter().sum(),
|
|
segments: ix.tracks.values().map(|t| t.segs.len() as u64).sum(),
|
|
rooms_sealed: ix.sealed.len() as u64,
|
|
takedowns: s.takedowns.load(Ordering::Relaxed),
|
|
objects_taken_down: s.objects_taken_down.load(Ordering::Relaxed),
|
|
purges_signed: s.purges_signed.load(Ordering::Relaxed),
|
|
}
|
|
}
|
|
|
|
/// The groups the store holds of `key`, ascending.
|
|
pub fn groups(&self, key: &TrackKey) -> Vec<u64> {
|
|
self.index.lock().unwrap().tracks.get(key).map(|t| t.groups.keys().copied().collect()).unwrap_or_default()
|
|
}
|
|
|
|
/// Where group `group` of `key` stands on disk: its segment and the place of its
|
|
/// frame, for tests and tools.
|
|
pub fn group_location(&self, key: &TrackKey, group: u64) -> Option<(PathBuf, u64)> {
|
|
let ix = self.index.lock().unwrap();
|
|
let t = ix.tracks.get(key)?;
|
|
let g = t.groups.get(&group)?;
|
|
Some((seg_path(&t.dir, g.seg), g.frame_at))
|
|
}
|
|
|
|
/// The directory of `key`'s files, for tests and tools.
|
|
pub fn track_dir(&self, key: &TrackKey) -> Option<PathBuf> {
|
|
self.index.lock().unwrap().tracks.get(key).map(|t| t.dir.clone())
|
|
}
|
|
|
|
// ── Takedowns ───────────────────────────────────────────────────────────
|
|
|
|
/// Whether any takedown is in force.
|
|
pub fn takes_down(&self) -> bool {
|
|
!self.index.lock().unwrap().takedowns.is_empty()
|
|
}
|
|
|
|
/// Whether a takedown covers the object of content hash `digest`, or group `group` of
|
|
/// `key` by its track.
|
|
pub fn taken_down(&self, key: &TrackKey, group: u64, digest: &[u8; 32]) -> bool {
|
|
let ix = self.index.lock().unwrap();
|
|
ix.taken.contains(digest) || ix.track_gone(key, group)
|
|
}
|
|
|
|
/// Honour a signed takedown (WAI takedown §10.1): verify it under the keys the store
|
|
/// trusts, record what it covers, and have the writer erase it, log it and sign the
|
|
/// purge. Its link hash, or why it is not honoured. A takedown honoured before is not
|
|
/// applied again.
|
|
pub fn takedown(&self, rec: &[u8]) -> Result<[u8; 32], String> {
|
|
self.takedown_under(rec, &self.takedown_keys)
|
|
}
|
|
|
|
fn takedown_under(&self, rec: &[u8], trusted: &[[u8; 32]]) -> Result<[u8; 32], String> {
|
|
let t = crate::takedown::read(rec, trusted)?;
|
|
if !self.index.lock().unwrap().take(&t) {
|
|
return Ok(t.link);
|
|
}
|
|
self.stats.takedowns.fetch_add(1, Ordering::Relaxed);
|
|
tracing::info!(target: "joule_moq_relay::dvr", takedown = %hex::encode(t.link), objects = t.objects.len(), tracks = t.tracks.len(), "a takedown is honoured");
|
|
let link = t.link;
|
|
self.send(Job::TakeDown { t, rec: rec.to_vec() });
|
|
Ok(link)
|
|
}
|
|
|
|
/// One object of a takedown track `[K, takedown]` the relay forwards: a takedown signed by
|
|
/// `K`, honoured when `K` is a key the store trusts.
|
|
pub fn takedown_object(&self, ns: &TrackNamespace, payload: &[u8]) {
|
|
let Some(k) = room_of(ns).filter(|_| ns.0.len() == 2 && ns.0[1] == crate::takedown::TRACK_FIELD) else { return };
|
|
if !self.takedown_keys.contains(&k) {
|
|
return;
|
|
}
|
|
if let Err(e) = self.takedown_under(payload, &[k]) {
|
|
tracing::warn!(target: "joule_moq_relay::dvr", error = %e, "a takedown is not honoured");
|
|
}
|
|
}
|
|
|
|
/// Where the purge of the takedown `link` is written.
|
|
pub fn purge_path(&self, link: &[u8; 32]) -> PathBuf {
|
|
self.root.join("purges").join(format!("{}.json", hex::encode(link)))
|
|
}
|
|
|
|
/// The writer's part of a takedown: log it, erase what it covers, sync, sign the purge.
|
|
fn apply_takedown(&self, w: &mut Writer, t: &crate::takedown::Takedown, rec: Option<&[u8]>) {
|
|
if let Some(rec) = rec {
|
|
let mut b = (rec.len() as u32).to_le_bytes().to_vec();
|
|
b.extend_from_slice(rec);
|
|
b.extend_from_slice(&sha256(&[rec]));
|
|
let r = std::fs::OpenOptions::new().create(true).append(true).open(self.root.join("takedowns.wdl")).and_then(|mut f| {
|
|
f.write_all(&b)?;
|
|
f.sync_all()
|
|
});
|
|
if let Err(e) = r {
|
|
tracing::error!(target: "joule_moq_relay::dvr", error = %e, "the takedowns log could not be written");
|
|
}
|
|
}
|
|
for d in &t.objects {
|
|
self.erase(w, None, *d, ERASED_TAKEN_DOWN);
|
|
}
|
|
for (track, from) in &t.tracks {
|
|
self.erase_track(w, track, *from);
|
|
}
|
|
self.sync(w);
|
|
self.write_purge(t);
|
|
}
|
|
|
|
/// Sign and write the purge of `t`, once.
|
|
fn write_purge(&self, t: &crate::takedown::Takedown) {
|
|
let Some(key) = &self.purge_key else { return };
|
|
let path = self.purge_path(&t.link);
|
|
if path.exists() {
|
|
return;
|
|
}
|
|
let body = crate::takedown::purge(t, now_us() / 1_000_000, key);
|
|
let r = std::fs::create_dir_all(self.root.join("purges")).and_then(|_| write_durably(&path, body.as_bytes()));
|
|
match r {
|
|
Ok(()) => {
|
|
self.stats.purges_signed.fetch_add(1, Ordering::Relaxed);
|
|
tracing::info!(target: "joule_moq_relay::dvr", takedown = %hex::encode(t.link), path = %path.display(), "a purge is signed");
|
|
}
|
|
Err(e) => tracing::error!(target: "joule_moq_relay::dvr", error = %e, "a purge could not be written"),
|
|
}
|
|
}
|
|
|
|
/// Erase every object of each group, from `from` on, of every track whose digest is
|
|
/// `track` (WAI live-rooms §6.2).
|
|
fn erase_track(&self, w: &mut Writer, track: &[u8; 32], from: u64) {
|
|
let groups: Vec<(TrackKey, PathBuf, u64)> = {
|
|
let ix = self.index.lock().unwrap();
|
|
ix.tracks
|
|
.iter()
|
|
.filter(|(k, _)| room_track_digest(k) == *track)
|
|
.flat_map(|(k, t)| t.groups.range(from..).map(move |(g, _)| (k.clone(), t.dir.clone(), *g)))
|
|
.collect()
|
|
};
|
|
for (key, dir, g) in groups {
|
|
let Some(grp) = self.index.lock().unwrap().tracks.get(&key).and_then(|t| t.groups.get(&g)).cloned() else { continue };
|
|
if grp.objects.is_empty() {
|
|
continue;
|
|
}
|
|
let mut erased = grp.erased.clone();
|
|
erased.extend(grp.objects.iter().map(|o| (o.id, o.digest, ERASED_TAKEN_DOWN)));
|
|
if self.rewrite(w, &key, &dir, g, grp.arrived_us, &Vec::new(), &erased, (grp.seg, grp.frame_at, grp.record_len)).is_none() {
|
|
continue;
|
|
}
|
|
let room = {
|
|
let mut ix = self.index.lock().unwrap();
|
|
for o in &grp.objects {
|
|
ix.unindex(&key, g, o);
|
|
}
|
|
let (window, room) = ix.tracks.get(&key).map_or((None, None), |t| (Some(t.window), t.room));
|
|
if let Some(gr) = ix.tracks.get_mut(&key).and_then(|t| t.groups.get_mut(&g)) {
|
|
gr.bytes = 0;
|
|
}
|
|
if let Some(wi) = window {
|
|
ix.window_bytes[wi] = ix.window_bytes[wi].saturating_sub(grp.bytes);
|
|
}
|
|
room
|
|
};
|
|
self.series_change(room, now_us(), -(grp.bytes as i128), 0, 0, 0);
|
|
self.stats.objects_taken_down.fetch_add(grp.objects.len() as u64, Ordering::Relaxed);
|
|
tracing::info!(target: "joule_moq_relay::dvr", dir = %dir.display(), group = g, objects = grp.objects.len(), "a takedown erased a group of a track");
|
|
}
|
|
}
|
|
|
|
// ── Moderation ──────────────────────────────────────────────────────────
|
|
|
|
/// One object of an events track `ns` the relay forwards: each `decision` record of it
|
|
/// is honoured when it verifies, its author owns the track, and the author is a
|
|
/// moderator (`moderators`: the principals of grants with the moderate power the
|
|
/// publishing session proved, in this room).
|
|
pub fn events_object(&self, ns: &TrackNamespace, payload: &[u8], moderators: &[[u8; 32]]) {
|
|
let (Some(room), Some(owner)) = (room_of(ns), key_field(ns, 1)) else { return };
|
|
if !(ns.0.len() == 3 && ns.0[2] == b"events") {
|
|
return;
|
|
}
|
|
let Some(recs) = events::records(payload) else { return };
|
|
for (_, rec) in recs {
|
|
let Some(e) = Event::read(&rec) else { continue };
|
|
if e.decision().is_none() || e.room != room || e.principal != owner {
|
|
continue;
|
|
}
|
|
if self.index.lock().unwrap().decided.contains(&e.digest) {
|
|
continue;
|
|
}
|
|
if !(self.room_keys.contains(&owner) || moderators.contains(&owner)) || !e.verifies(&rec) {
|
|
continue;
|
|
}
|
|
self.apply_decision(&e);
|
|
if !self.index.lock().unwrap().sealed.contains(&room) {
|
|
self.send(Job::Log(rec));
|
|
}
|
|
}
|
|
}
|
|
|
|
/// Returns the target to erase, for a redaction.
|
|
fn apply_decision(&self, e: &Event) -> Option<[u8; 32]> {
|
|
let (verdict, target) = e.decision()?;
|
|
let mut ix = self.index.lock().unwrap();
|
|
if !ix.decided.insert(e.digest) {
|
|
return None;
|
|
}
|
|
ix.standing.insert((e.room, target), verdict);
|
|
drop(ix);
|
|
self.stats.decisions.fetch_add(1, Ordering::Relaxed);
|
|
tracing::info!(target: "joule_moq_relay::dvr", room = %hex::encode(e.room), target_digest = %hex::encode(target), verdict, "a decision is honoured");
|
|
if verdict == REDACT {
|
|
self.send(Job::Erase { room: e.room, digest: target });
|
|
return Some(target);
|
|
}
|
|
None
|
|
}
|
|
|
|
// ── Segments ────────────────────────────────────────────────────────────
|
|
|
|
/// Append `record` to `key`'s open segment (beginning another past its span or size),
|
|
/// returning the segment and the frame's place.
|
|
fn append(&self, w: &mut Writer, key: &TrackKey, dir: &Path, record: &[u8]) -> std::io::Result<(u64, u64)> {
|
|
let now = now_us();
|
|
let stale = w.open.get(key).is_none_or(|s| now.saturating_sub(s.started_us) >= SEGMENT_SPAN_US || s.len >= SEGMENT_MAX);
|
|
if stale {
|
|
if let Some(old) = w.open.remove(key) {
|
|
self.retire(key, old.seq, None);
|
|
}
|
|
let seq = *w.next.get(key).unwrap_or(&0);
|
|
w.next.insert(key.clone(), seq + 1);
|
|
let path = seg_path(dir, seq);
|
|
let file = std::fs::OpenOptions::new().create(true).truncate(true).read(true).write(true).open(&path)?;
|
|
w.dirs.insert(dir.to_path_buf());
|
|
w.open.insert(key.clone(), OpenSeg { seq, file, started_us: now, len: 0 });
|
|
if let Some(t) = self.index.lock().unwrap().tracks.get_mut(key) {
|
|
t.segs.entry(seq).or_insert(0);
|
|
}
|
|
}
|
|
let s = w.open.get_mut(key).expect("opened above");
|
|
let mut frame = Vec::with_capacity(FRAME as usize + record.len());
|
|
frame.extend_from_slice(&(record.len() as u32).to_le_bytes());
|
|
frame.push(1);
|
|
frame.extend_from_slice(record);
|
|
s.file.write_all_at(&frame, s.len)?;
|
|
let at = s.len;
|
|
s.len += frame.len() as u64;
|
|
w.dirty.insert(seg_path(dir, s.seq));
|
|
Ok((s.seq, at))
|
|
}
|
|
|
|
/// Mark a record replaced: its frame's state 0 and its bytes zeros.
|
|
fn kill(&self, w: &mut Writer, dir: &Path, seq: u64, frame_at: u64, record_len: u32) {
|
|
let path = seg_path(dir, seq);
|
|
let r = std::fs::OpenOptions::new().write(true).open(&path).and_then(|f| {
|
|
f.write_all_at(&[0], frame_at + 4)?;
|
|
f.write_all_at(&vec![0u8; record_len as usize], frame_at + FRAME)
|
|
});
|
|
if r.is_ok() {
|
|
w.dirty.insert(path);
|
|
}
|
|
}
|
|
|
|
/// Delete segment `seq` of `key` when no group's standing record is in it and it is not
|
|
/// the one being appended to (`open`, the writer's).
|
|
fn retire(&self, key: &TrackKey, seq: u64, open: Option<u64>) {
|
|
if open == Some(seq) {
|
|
return;
|
|
}
|
|
let path = {
|
|
let mut ix = self.index.lock().unwrap();
|
|
let Some(t) = ix.tracks.get_mut(key) else { return };
|
|
if t.segs.get(&seq).is_some_and(|n| *n > 0) {
|
|
return;
|
|
}
|
|
t.segs.remove(&seq);
|
|
seg_path(&t.dir, seq)
|
|
};
|
|
let _ = std::fs::remove_file(&path);
|
|
}
|
|
|
|
/// Sync every segment written since the last sync, and the directories files were made
|
|
/// in.
|
|
fn sync(&self, w: &mut Writer) {
|
|
for p in w.dirty.drain() {
|
|
if let Ok(f) = std::fs::OpenOptions::new().write(true).open(&p) {
|
|
let _ = f.sync_data();
|
|
}
|
|
self.stats.syncs.fetch_add(1, Ordering::Relaxed);
|
|
}
|
|
for d in w.dirs.drain() {
|
|
sync_dir(&d);
|
|
}
|
|
w.last_sync = Some(Instant::now());
|
|
}
|
|
|
|
/// Write group `g` of `key` again from `rows` and `erased`: append the new record, kill
|
|
/// the old, and update the index. Returns the group's new objects.
|
|
#[allow(clippy::too_many_arguments)]
|
|
fn rewrite(&self, w: &mut Writer, key: &TrackKey, dir: &Path, g: u64, arrived_us: u64, rows: &Rows, erased: &[(u64, [u8; 32], u8)], old: (u64, u64, u32)) -> Option<Vec<DObject>> {
|
|
let (bytes, mut index) = encode_group(&track_digest(key), g, arrived_us, rows, erased);
|
|
let (seq, at) = match self.append(w, key, dir, &bytes) {
|
|
Ok(x) => x,
|
|
Err(e) => {
|
|
tracing::error!(target: "joule_moq_relay::dvr", error = %e, "a group could not be written again");
|
|
return None;
|
|
}
|
|
};
|
|
for (o, r) in index.iter_mut().zip(rows) {
|
|
o.at += at + FRAME;
|
|
o.records = r.0.records.clone();
|
|
}
|
|
{
|
|
let mut ix = self.index.lock().unwrap();
|
|
if let Some(t) = ix.tracks.get_mut(key) {
|
|
*t.segs.entry(seq).or_insert(0) += 1;
|
|
if let Some(n) = t.segs.get_mut(&old.0) {
|
|
*n = n.saturating_sub(1);
|
|
}
|
|
if let Some(gr) = t.groups.get_mut(&g) {
|
|
gr.objects = index.clone();
|
|
gr.erased = erased.to_vec();
|
|
gr.seg = seq;
|
|
gr.frame_at = at;
|
|
gr.record_len = bytes.len() as u32;
|
|
}
|
|
}
|
|
}
|
|
self.kill(w, dir, old.0, old.1, old.2);
|
|
let open = w.open.get(key).map(|s| s.seq);
|
|
self.retire(key, old.0, open);
|
|
Some(index)
|
|
}
|
|
|
|
// ── Loading ─────────────────────────────────────────────────────────────
|
|
|
|
fn load(&self, w: &mut Writer, verifier: &dyn Fn(&TrackNamespace) -> ReceiptVerifier) -> Result<(), String> {
|
|
let now = now_us();
|
|
// The rooms found sealed before. Nothing of them is loaded: their decisions and the
|
|
// tracks a crash left of them are deleted.
|
|
let path = self.root.join("sealed.wdl");
|
|
if let Ok(b) = std::fs::read(&path) {
|
|
let mut good = 0;
|
|
for c in b.as_chunks::<64>().0 {
|
|
if sha256(&[&c[..32]])[..] != c[32..] {
|
|
break;
|
|
}
|
|
self.index.lock().unwrap().sealed.insert(c[..32].try_into().expect("32"));
|
|
good += 64;
|
|
}
|
|
if good != b.len() {
|
|
let _ = std::fs::OpenOptions::new().write(true).open(&path).and_then(|f| f.set_len(good as u64));
|
|
}
|
|
}
|
|
let sealed = self.index.lock().unwrap().sealed.clone();
|
|
if !sealed.is_empty() {
|
|
self.rewrite_decisions(|room| !sealed.contains(room));
|
|
}
|
|
// The decisions honoured before, each verified again.
|
|
let log = self.root.join("decisions.wdl");
|
|
if let Ok(b) = std::fs::read(&log) {
|
|
let mut p = 0;
|
|
let mut good = 0;
|
|
while b.len() >= p + 4 {
|
|
let n = u32::from_le_bytes(b[p..p + 4].try_into().expect("4")) as usize;
|
|
let Some(rec) = b.get(p + 4..p + 4 + n) else { break };
|
|
let Some(check) = b.get(p + 4 + n..p + 4 + n + 32) else { break };
|
|
if sha256(&[rec])[..] != *check {
|
|
break;
|
|
}
|
|
if let Some(e) = Event::read(rec).filter(|e| e.verifies(rec)) {
|
|
// Erasures are made below, once the groups are read.
|
|
let (verdict, target) = e.decision().unwrap_or((0, [0; 32]));
|
|
let mut ix = self.index.lock().unwrap();
|
|
if verdict != 0 && ix.decided.insert(e.digest) {
|
|
ix.standing.insert((e.room, target), verdict);
|
|
self.stats.decisions.fetch_add(1, Ordering::Relaxed);
|
|
}
|
|
}
|
|
p += 4 + n + 32;
|
|
good = p;
|
|
}
|
|
if good != b.len() {
|
|
// A record a crash cut short: the log keeps what precedes it.
|
|
let _ = std::fs::OpenOptions::new().write(true).open(&log).and_then(|f| f.set_len(good as u64));
|
|
}
|
|
}
|
|
// The takedowns honoured before, each verified again (WAI takedown §10.1).
|
|
let mut takedowns: Vec<crate::takedown::Takedown> = Vec::new();
|
|
let tlog = self.root.join("takedowns.wdl");
|
|
if let Ok(b) = std::fs::read(&tlog) {
|
|
let (mut p, mut good) = (0, 0);
|
|
while b.len() >= p + 4 {
|
|
let n = u32::from_le_bytes(b[p..p + 4].try_into().expect("4")) as usize;
|
|
let Some(rec) = b.get(p + 4..p + 4 + n) else { break };
|
|
let Some(check) = b.get(p + 4 + n..p + 4 + n + 32) else { break };
|
|
if sha256(&[rec])[..] != *check {
|
|
break;
|
|
}
|
|
match crate::takedown::read(rec, &self.takedown_keys) {
|
|
Ok(t) => {
|
|
if self.index.lock().unwrap().take(&t) {
|
|
self.stats.takedowns.fetch_add(1, Ordering::Relaxed);
|
|
takedowns.push(t);
|
|
}
|
|
}
|
|
Err(e) => tracing::warn!(target: "joule_moq_relay::dvr", error = %e, "a logged takedown no longer verifies: not honoured"),
|
|
}
|
|
p += 4 + n + 32;
|
|
good = p;
|
|
}
|
|
if good != b.len() {
|
|
let _ = std::fs::OpenOptions::new().write(true).open(&tlog).and_then(|f| f.set_len(good as u64));
|
|
}
|
|
}
|
|
let tracks_dir = self.root.join("tracks");
|
|
let mut loaded: Vec<(TrackKey, DTrack, HashMap<u64, Rows>)> = Vec::new();
|
|
// Groups whose standing record no longer says what the index holds: written again.
|
|
let mut changed: HashSet<(TrackKey, u64)> = HashSet::new();
|
|
for d in std::fs::read_dir(&tracks_dir).map_err(|e| e.to_string())?.flatten() {
|
|
let dir = d.path();
|
|
let Some((key, props)) = std::fs::read(dir.join("TRACK")).ok().and_then(|b| decode_track(&b)) else {
|
|
tracing::warn!(target: "joule_moq_relay::dvr", dir = %dir.display(), "a track directory whose TRACK does not read: left alone");
|
|
continue;
|
|
};
|
|
if room_of(&key.0).is_some_and(|r| sealed.contains(&r)) {
|
|
let _ = std::fs::remove_dir_all(&dir);
|
|
continue;
|
|
}
|
|
let Some(window) = self.window_of(&key.0) else { continue };
|
|
let digest = track_digest(&key);
|
|
let mut t = DTrack { dir: dir.clone(), window, room: room_of(&key.0), events: is_events_track(&key), props, groups: BTreeMap::new(), absent: BTreeMap::new(), evicted_through: None, end_of_track: None, segs: BTreeMap::new() };
|
|
let mut segs: Vec<(u64, PathBuf)> = std::fs::read_dir(&dir).map_err(|e| e.to_string())?.flatten().filter_map(|e| seg_seq(&e.path()).map(|s| (s, e.path()))).collect();
|
|
segs.sort();
|
|
// Each group's standing record: the last that reads; the earlier ones killed.
|
|
let mut rows: HashMap<u64, Rows> = HashMap::new();
|
|
let mut replaced: Vec<(u64, u64, u32)> = Vec::new();
|
|
for (seq, path) in &segs {
|
|
let Ok(b) = std::fs::read(path) else { continue };
|
|
t.segs.insert(*seq, 0);
|
|
let mut p = 0usize;
|
|
while p < b.len() {
|
|
let whole = b.len() >= p + FRAME as usize && {
|
|
let n = u32::from_le_bytes(b[p..p + 4].try_into().expect("4")) as usize;
|
|
b.len() >= p + FRAME as usize + n
|
|
};
|
|
if !whole {
|
|
// A frame a crash cut short: cut off.
|
|
self.stats.torn_writes.fetch_add(1, Ordering::Relaxed);
|
|
let _ = std::fs::OpenOptions::new().write(true).open(path).and_then(|f| f.set_len(p as u64));
|
|
break;
|
|
}
|
|
let n = u32::from_le_bytes(b[p..p + 4].try_into().expect("4")) as usize;
|
|
let state = b[p + 4];
|
|
let rec = &b[p + FRAME as usize..p + FRAME as usize + n];
|
|
let frame_at = p as u64;
|
|
p += FRAME as usize + n;
|
|
if state != 1 {
|
|
continue;
|
|
}
|
|
let Some(dg) = decode_group(rec, &digest) else {
|
|
self.stats.objects_dropped.fetch_add(1, Ordering::Relaxed);
|
|
self.series_change(t.room, now, 0, 0, 0, 1);
|
|
tracing::warn!(target: "joule_moq_relay::dvr", file = %path.display(), at = frame_at, "a group record whose header does not check: dropped");
|
|
replaced.push((*seq, frame_at, n as u32));
|
|
continue;
|
|
};
|
|
if let Some(prev) = t.groups.remove(&dg.group) {
|
|
replaced.push((prev.seg, prev.frame_at, prev.record_len));
|
|
if let Some(c) = t.segs.get_mut(&prev.seg) {
|
|
*c = c.saturating_sub(1);
|
|
}
|
|
}
|
|
let base = frame_at + FRAME;
|
|
let short = dg.count as usize - dg.objects.len();
|
|
let mut grp = DGroup { arrived_us: dg.arrived_us, bytes: 0, objects: Vec::new(), erased: dg.erased, seg: *seq, frame_at, record_len: n as u32 };
|
|
let mut kept: Rows = Vec::new();
|
|
if short > 0 {
|
|
changed.insert((key.clone(), dg.group));
|
|
}
|
|
for (mut o, props, payload, ok) in dg.objects {
|
|
o.at += base;
|
|
if !ok || ObjectReceipt::hash_payload(&payload) != o.digest {
|
|
self.dropped(&t, dg.group, &o, &mut grp, "its check does not hold");
|
|
changed.insert((key.clone(), dg.group));
|
|
continue;
|
|
}
|
|
if t.events {
|
|
o.records = events::digests(&payload);
|
|
}
|
|
kept.push((o, props, payload));
|
|
}
|
|
if short > 0 {
|
|
self.stats.objects_dropped.fetch_add(short as u64, Ordering::Relaxed);
|
|
self.series_change(t.room, now, 0, 0, 0, short as u64);
|
|
}
|
|
for (o, props, payload) in &kept {
|
|
if let Some(r) = prior_groups(&Properties::decode(props, "dvr").unwrap_or_default(), dg.group) {
|
|
t.absent.insert(r.0, r.1);
|
|
}
|
|
if o.status == ObjectStatus::EndOfTrack {
|
|
t.end_of_track = Some(Location::new(dg.group, o.id));
|
|
}
|
|
grp.bytes += payload.len() as u64;
|
|
grp.objects.push(o.clone());
|
|
}
|
|
*t.segs.entry(*seq).or_insert(0) += 1;
|
|
t.groups.insert(dg.group, grp);
|
|
rows.insert(dg.group, kept);
|
|
}
|
|
}
|
|
for (seq, at, len) in replaced {
|
|
self.kill(w, &dir, seq, at, len);
|
|
}
|
|
w.next.insert(key.clone(), segs.last().map_or(0, |(s, _)| s + 1));
|
|
loaded.push((key, t, rows));
|
|
}
|
|
// Every object of a track with a receipts track, against its group's receipts; every
|
|
// record of an events track, under its principal.
|
|
let held: HashSet<TrackKey> = loaded.iter().map(|(k, _, _)| k.clone()).collect();
|
|
let receipts: HashMap<TrackKey, HashMap<u64, Bytes>> = loaded
|
|
.iter()
|
|
.filter(|(k, _, _)| self.media_of(&k.1).is_some())
|
|
.map(|(k, _, rows)| (k.clone(), rows.iter().filter_map(|(g, r)| r.first().map(|(_, _, p)| (*g, p.clone()))).collect()))
|
|
.collect();
|
|
for (k, t, rows) in &mut loaded {
|
|
let rkey = (k.0.clone(), self.receipts_of(&k.1));
|
|
let check_receipts = self.media_of(&k.1).is_none() && held.contains(&rkey);
|
|
if !(check_receipts || t.events) {
|
|
continue;
|
|
}
|
|
let v = verifier(&k.0);
|
|
let groups: Vec<u64> = t.groups.keys().copied().collect();
|
|
for g in groups {
|
|
let bundle = if check_receipts { receipts.get(&rkey).and_then(|m| m.get(&g)).and_then(|b| ReceiptBundle::from_payload(b).ok()) } else { None };
|
|
let vg = bundle.as_ref().and_then(|b| (b.group.group_id == GroupId(g)).then(|| v.verify_group(&b.group).ok()).flatten());
|
|
let Some(r) = rows.get_mut(&g) else { continue };
|
|
let mut bad = Vec::new();
|
|
for (o, _, payload) in r.iter() {
|
|
let ok = if check_receipts {
|
|
match (&bundle, &vg) {
|
|
(Some(b), Some(vg)) => b.objects.iter().find(|x| x.object_id == ObjectId(o.id)).is_some_and(|x| v.verify_payload(payload, x, vg).is_ok()),
|
|
_ => o.status != ObjectStatus::Normal,
|
|
}
|
|
} else {
|
|
match events::records(payload) {
|
|
Some(rs) => rs.iter().all(|(_, x)| Event::read(x).is_some_and(|e| e.verifies(x))),
|
|
None => true,
|
|
}
|
|
};
|
|
if !ok {
|
|
bad.push(o.clone());
|
|
}
|
|
}
|
|
let mut grp = t.groups.remove(&g).expect("held");
|
|
if !bad.is_empty() {
|
|
changed.insert((k.clone(), g));
|
|
}
|
|
for o in &bad {
|
|
self.dropped(t, g, o, &mut grp, if check_receipts { "it does not verify against its receipt" } else { "a record of it does not verify" });
|
|
r.retain(|x| x.0.id != o.id);
|
|
}
|
|
t.groups.insert(g, grp);
|
|
}
|
|
}
|
|
// Into the index; then every group whose record no longer says what the index holds
|
|
// (an object dropped, or a redaction the log asks for that a crash left undone) is
|
|
// written again.
|
|
let mut again: Vec<(TrackKey, PathBuf, u64)> = Vec::new();
|
|
{
|
|
let mut ix = self.index.lock().unwrap();
|
|
for (k, mut t, _) in std::mem::take(&mut loaded) {
|
|
for (g, grp) in t.groups.iter_mut() {
|
|
// What a takedown covers, erased (a crash can leave it held).
|
|
let gone = ix.track_gone(&k, *g);
|
|
let taken: Vec<DObject> = grp.objects.iter().filter(|o| gone || ix.taken.contains(&o.digest)).cloned().collect();
|
|
if !taken.is_empty() {
|
|
changed.insert((k.clone(), *g));
|
|
}
|
|
for o in &taken {
|
|
grp.objects.retain(|x| x.id != o.id);
|
|
grp.bytes = grp.bytes.saturating_sub(o.payload_len as u64);
|
|
grp.erased.push((o.id, o.digest, ERASED_TAKEN_DOWN));
|
|
self.stats.objects_taken_down.fetch_add(1, Ordering::Relaxed);
|
|
}
|
|
let redacted: Vec<DObject> = grp.objects.iter().filter(|o| t.room.is_some_and(|r| ix.standing.get(&(r, o.digest)) == Some(&REDACT))).cloned().collect();
|
|
if !redacted.is_empty() {
|
|
changed.insert((k.clone(), *g));
|
|
}
|
|
for o in &redacted {
|
|
grp.objects.retain(|x| x.id != o.id);
|
|
grp.bytes = grp.bytes.saturating_sub(o.payload_len as u64);
|
|
grp.erased.push((o.id, o.digest, ERASED_REDACTED));
|
|
self.stats.objects_redacted.fetch_add(1, Ordering::Relaxed);
|
|
}
|
|
ix.window_bytes[t.window] += grp.bytes;
|
|
if let Some(room) = t.room {
|
|
ix.series.entry(room).or_default().change(now, grp.bytes as i128, 0, 0, 0);
|
|
}
|
|
for o in &grp.objects {
|
|
ix.reindex(&k, *g, o);
|
|
}
|
|
if changed.contains(&(k.clone(), *g)) {
|
|
again.push((k.clone(), t.dir.clone(), *g));
|
|
}
|
|
}
|
|
// What is below the first group held is no longer known to the store.
|
|
t.evicted_through = t.groups.keys().next().and_then(|g| g.checked_sub(1));
|
|
ix.tracks.insert(k, t);
|
|
}
|
|
}
|
|
for (k, dir, g) in again {
|
|
self.write_again_if_changed(w, &k, &dir, g);
|
|
}
|
|
// Event records a redaction the log holds targets, erased from the objects that
|
|
// carry them.
|
|
let redactions: Vec<([u8; 32], [u8; 32])> = self.index.lock().unwrap().standing.iter().filter(|(_, v)| **v == REDACT).map(|(k, _)| *k).collect();
|
|
for (room, target) in redactions {
|
|
self.erase(w, Some(room), target, ERASED_REDACTED);
|
|
}
|
|
// The takedowns the log holds: what a crash left of their erasures is made now, and a
|
|
// purge a crash left unsigned is signed.
|
|
for t in &takedowns {
|
|
self.apply_takedown(w, t, None);
|
|
}
|
|
self.sync(w);
|
|
let n: usize = self.index.lock().unwrap().tracks.values().map(|t| t.groups.len()).sum();
|
|
tracing::info!(target: "joule_moq_relay::dvr", path = %self.root.display(), groups = n, dropped = self.stats.objects_dropped.load(Ordering::Relaxed), torn = self.stats.torn_writes.load(Ordering::Relaxed), "the DVR store is open");
|
|
Ok(())
|
|
}
|
|
|
|
/// Write group `g` again when its standing record holds other objects than the index
|
|
/// does (some dropped, or erased).
|
|
fn write_again_if_changed(&self, w: &mut Writer, key: &TrackKey, dir: &Path, g: u64) {
|
|
let Some(grp) = self.index.lock().unwrap().tracks.get(key).and_then(|t| t.groups.get(&g)).cloned() else { return };
|
|
let path = seg_path(dir, grp.seg);
|
|
let Ok(rec) = read_at(&path, grp.frame_at + FRAME, grp.record_len as usize) else { return };
|
|
let Some(dg) = decode_group(&rec, &track_digest(key)) else { return };
|
|
if dg.count as usize == grp.objects.len() && dg.erased.len() == grp.erased.len() {
|
|
return;
|
|
}
|
|
let rows: Rows = grp.objects.iter().filter_map(|o| dg.objects.iter().find(|x| x.0.id == o.id).map(|(_, props, payload, _)| (o.clone(), props.clone(), payload.clone()))).collect();
|
|
self.rewrite(w, key, dir, g, grp.arrived_us, &rows, &grp.erased, (grp.seg, grp.frame_at, grp.record_len));
|
|
}
|
|
|
|
/// An object that did not verify at load: out of the group, among its erased, counted.
|
|
fn dropped(&self, t: &DTrack, g: u64, o: &DObject, grp: &mut DGroup, why: &str) {
|
|
self.stats.objects_dropped.fetch_add(1, Ordering::Relaxed);
|
|
self.series_change(t.room, now_us(), 0, 0, 0, 1);
|
|
tracing::warn!(target: "joule_moq_relay::dvr", dir = %t.dir.display(), group = g, object = o.id, reason = why, "a stored object dropped: never served");
|
|
if grp.objects.iter().any(|x| x.id == o.id) {
|
|
grp.bytes = grp.bytes.saturating_sub(o.payload_len as u64);
|
|
}
|
|
grp.objects.retain(|x| x.id != o.id);
|
|
if !grp.erased.iter().any(|e| e.0 == o.id) {
|
|
grp.erased.push((o.id, o.digest, ERASED_UNVERIFIED));
|
|
}
|
|
}
|
|
|
|
fn series_change(&self, room: Option<[u8; 32]>, now: u64, held: i128, objects: u64, bytes: u64, dropped: u64) {
|
|
if let Some(r) = room {
|
|
self.index.lock().unwrap().series.entry(r).or_default().change(now, held, objects, bytes, dropped);
|
|
}
|
|
}
|
|
|
|
fn media_of(&self, name: &TrackName) -> Option<TrackName> {
|
|
(!self.suffix.is_empty() && name.0.len() > self.suffix.len() && name.0.ends_with(&self.suffix)).then(|| TrackName(name.0[..name.0.len() - self.suffix.len()].to_vec()))
|
|
}
|
|
|
|
fn receipts_of(&self, name: &TrackName) -> TrackName {
|
|
let mut v = name.0.clone();
|
|
v.extend_from_slice(&self.suffix);
|
|
TrackName(v)
|
|
}
|
|
|
|
// ── The writer ──────────────────────────────────────────────────────────
|
|
|
|
fn run(self: Arc<Self>, mut w: Writer, rx: mpsc::Receiver<Job>) {
|
|
loop {
|
|
let job = rx.recv_timeout(SYNC_EVERY);
|
|
match job {
|
|
Ok(Job::Seal { key, props, group, arrived_us, objects }) => self.write_seal(&mut w, key, props, group, arrived_us, objects),
|
|
Ok(Job::Erase { room, digest }) => self.erase(&mut w, Some(room), digest, ERASED_REDACTED),
|
|
Ok(Job::TakeDown { t, rec }) => self.apply_takedown(&mut w, &t, Some(&rec)),
|
|
Ok(Job::Log(rec)) => self.log(&rec),
|
|
Ok(Job::Forget(room)) => self.forget(&mut w, room),
|
|
Ok(Job::Flush(done)) => {
|
|
self.evict_by_age(&mut w, now_us());
|
|
self.sync(&mut w);
|
|
let _ = done.send(());
|
|
}
|
|
Err(mpsc::RecvTimeoutError::Timeout) => {}
|
|
Err(mpsc::RecvTimeoutError::Disconnected) => {
|
|
self.sync(&mut w);
|
|
return;
|
|
}
|
|
}
|
|
if w.last_sync.is_none_or(|t| t.elapsed() >= SYNC_EVERY) {
|
|
self.evict_by_age(&mut w, now_us());
|
|
self.sync(&mut w);
|
|
}
|
|
}
|
|
}
|
|
|
|
fn log(&self, rec: &[u8]) {
|
|
// A room found sealed after the record was queued: its decisions are not kept.
|
|
if Event::read(rec).is_some_and(|e| self.index.lock().unwrap().sealed.contains(&e.room)) {
|
|
return;
|
|
}
|
|
let path = self.root.join("decisions.wdl");
|
|
let mut b = (rec.len() as u32).to_le_bytes().to_vec();
|
|
b.extend_from_slice(rec);
|
|
b.extend_from_slice(&sha256(&[rec]));
|
|
let r = std::fs::OpenOptions::new().create(true).append(true).open(&path).and_then(|mut f| {
|
|
f.write_all(&b)?;
|
|
f.sync_all()
|
|
});
|
|
if let Err(e) = r {
|
|
tracing::error!(target: "joule_moq_relay::dvr", error = %e, "the decisions log could not be written");
|
|
}
|
|
}
|
|
|
|
/// Forget room `room`, whose objects are sealed: record it in `sealed.wdl`, delete every
|
|
/// track of it the store holds, and drop its decisions from `decisions.wdl`.
|
|
fn forget(&self, w: &mut Writer, room: [u8; 32]) {
|
|
let mut rec = room.to_vec();
|
|
rec.extend_from_slice(&sha256(&[&room]));
|
|
let r = std::fs::OpenOptions::new().create(true).append(true).open(self.root.join("sealed.wdl")).and_then(|mut f| {
|
|
f.write_all(&rec)?;
|
|
f.sync_all()
|
|
});
|
|
if let Err(e) = r {
|
|
tracing::error!(target: "joule_moq_relay::dvr", error = %e, "a sealed room could not be recorded");
|
|
}
|
|
sync_dir(&self.root);
|
|
let keys: Vec<TrackKey> = self.index.lock().unwrap().tracks.iter().filter(|(_, t)| t.room == Some(room)).map(|(k, _)| k.clone()).collect();
|
|
for k in &keys {
|
|
self.drop_track(w, k);
|
|
}
|
|
self.rewrite_decisions(|r| *r != room);
|
|
tracing::info!(target: "joule_moq_relay::dvr", room = %hex::encode(room), tracks = keys.len(), "a sealed room is forgotten");
|
|
}
|
|
|
|
/// Delete track `key` from the store: its groups, its directory, its writer state.
|
|
fn drop_track(&self, w: &mut Writer, key: &TrackKey) {
|
|
let t = {
|
|
let mut ix = self.index.lock().unwrap();
|
|
let Some(t) = ix.tracks.remove(key) else { return };
|
|
let held: u64 = t.groups.values().map(|g| g.bytes).sum();
|
|
ix.window_bytes[t.window] = ix.window_bytes[t.window].saturating_sub(held);
|
|
for (g, grp) in &t.groups {
|
|
for o in &grp.objects {
|
|
ix.unindex(key, *g, o);
|
|
}
|
|
}
|
|
if let Some(r) = t.room {
|
|
ix.series.entry(r).or_default().change(now_us(), -(held as i128), 0, 0, 0);
|
|
}
|
|
t
|
|
};
|
|
w.open.remove(key);
|
|
w.next.remove(key);
|
|
w.dirty.retain(|p| !p.starts_with(&t.dir));
|
|
if let Err(e) = std::fs::remove_dir_all(&t.dir) {
|
|
tracing::error!(target: "joule_moq_relay::dvr", error = %e, dir = %t.dir.display(), "a sealed room's track could not be deleted");
|
|
}
|
|
w.dirs.insert(self.root.join("tracks"));
|
|
}
|
|
|
|
/// Write `decisions.wdl` again with only the records whose room `keep` keeps; a record
|
|
/// that does not read as an event stays. Nothing is written when every record stays.
|
|
fn rewrite_decisions(&self, keep: impl Fn(&[u8; 32]) -> bool) {
|
|
let path = self.root.join("decisions.wdl");
|
|
let Ok(b) = std::fs::read(&path) else { return };
|
|
let (mut out, mut p, mut dropped) = (Vec::with_capacity(b.len()), 0, false);
|
|
while b.len() >= p + 4 {
|
|
let n = u32::from_le_bytes(b[p..p + 4].try_into().expect("4")) as usize;
|
|
let Some(frame) = b.get(p..p + 4 + n + 32) else { break };
|
|
if Event::read(&frame[4..4 + n]).is_some_and(|e| !keep(&e.room)) {
|
|
dropped = true;
|
|
} else {
|
|
out.extend_from_slice(frame);
|
|
}
|
|
p += 4 + n + 32;
|
|
}
|
|
if dropped && let Err(e) = write_durably(&path, &out) {
|
|
tracing::error!(target: "joule_moq_relay::dvr", error = %e, "the decisions log could not be written again");
|
|
}
|
|
}
|
|
|
|
fn write_seal(&self, w: &mut Writer, key: TrackKey, props: Properties, group: u64, arrived_us: u64, objects: Vec<((u64, u64), CachedObject)>) {
|
|
let Some(window) = self.window_of(&key.0) else { return };
|
|
let room = room_of(&key.0);
|
|
// A seal queued before its room was found sealed. The room is marked before its
|
|
// forgetting is queued, and this thread takes jobs in order: a seal it takes after the
|
|
// mark is refused here, and one it took before is deleted by the forgetting.
|
|
if room.is_some_and(|r| self.index.lock().unwrap().sealed.contains(&r)) {
|
|
return;
|
|
}
|
|
let events = is_events_track(&key);
|
|
let dir = self.root.join("tracks").join(track_dir_name(&key));
|
|
let known = self.index.lock().unwrap().tracks.contains_key(&key);
|
|
if !known {
|
|
if let Err(e) = std::fs::create_dir_all(&dir).and_then(|_| write_durably(&dir.join("TRACK"), &encode_track(&key, &props))) {
|
|
tracing::error!(target: "joule_moq_relay::dvr", error = %e, "a track could not be stored");
|
|
return;
|
|
}
|
|
w.dirs.insert(self.root.join("tracks"));
|
|
let mut ix = self.index.lock().unwrap();
|
|
ix.tracks.entry(key.clone()).or_insert_with(|| DTrack { dir: dir.clone(), window, room, events, props: props.clone(), groups: BTreeMap::new(), absent: BTreeMap::new(), evicted_through: None, end_of_track: None, segs: BTreeMap::new() });
|
|
}
|
|
// A seal checked against the window when it was queued can reach the writer after
|
|
// the window has moved past its group: it is late here too, and never written below
|
|
// the evicted mark (where it would leave the window two runs).
|
|
if self.index.lock().unwrap().tracks.get(&key).is_some_and(|t| t.evicted_through.is_some_and(|e| group <= e)) {
|
|
self.stats.groups_late.fetch_add(1, Ordering::Relaxed);
|
|
return;
|
|
}
|
|
let mut rows: Rows = Vec::with_capacity(objects.len());
|
|
let mut erased = Vec::new();
|
|
{
|
|
let ix = self.index.lock().unwrap();
|
|
let gone = ix.track_gone(&key, group);
|
|
for ((_, oid), o) in &objects {
|
|
let d = ObjectReceipt::hash_payload(&o.payload);
|
|
// A takedown covers it: it is never stored (WAI takedown §10.1).
|
|
if gone || ix.taken.contains(&d) {
|
|
erased.push((*oid, d, ERASED_TAKEN_DOWN));
|
|
self.stats.objects_taken_down.fetch_add(1, Ordering::Relaxed);
|
|
continue;
|
|
}
|
|
if room.is_some_and(|r| ix.standing.get(&(r, d)) == Some(&REDACT)) {
|
|
erased.push((*oid, d, ERASED_REDACTED));
|
|
continue;
|
|
}
|
|
let records = if events { events::digests(&o.payload) } else { Vec::new() };
|
|
let row = DObject { id: *oid, subgroup: o.subgroup, priority: o.publisher_priority, status: o.status, at: 0, props_len: 0, payload_len: 0, digest: d, records };
|
|
rows.push((row, Bytes::from(o.properties.to_bytes().unwrap_or_default()), o.payload.clone()));
|
|
}
|
|
}
|
|
let (bytes, mut index) = encode_group(&track_digest(&key), group, arrived_us, &rows, &erased);
|
|
let (seq, at) = match self.append(w, &key, &dir, &bytes) {
|
|
Ok(x) => x,
|
|
Err(e) => {
|
|
tracing::error!(target: "joule_moq_relay::dvr", error = %e, "a group could not be stored");
|
|
return;
|
|
}
|
|
};
|
|
for o in &mut index {
|
|
o.at += at + FRAME;
|
|
}
|
|
let payload_bytes: u64 = index.iter().map(|o| o.payload_len as u64).sum();
|
|
let mut ix = self.index.lock().unwrap();
|
|
let Some(t) = ix.tracks.get_mut(&key) else { return };
|
|
if t.evicted_through.is_some_and(|e| group <= e) {
|
|
drop(ix);
|
|
self.kill(w, &dir, seq, at, bytes.len() as u32);
|
|
self.stats.groups_late.fetch_add(1, Ordering::Relaxed);
|
|
return;
|
|
}
|
|
for (_, o) in &objects {
|
|
if let Some(r) = prior_groups(&o.properties, group) {
|
|
t.absent.insert(r.0, r.1);
|
|
}
|
|
}
|
|
if let Some(((g, oo), _)) = objects.iter().find(|(_, x)| x.status == ObjectStatus::EndOfTrack) {
|
|
t.end_of_track = Some(Location::new(*g, *oo));
|
|
}
|
|
*t.segs.entry(seq).or_insert(0) += 1;
|
|
let old = t.groups.insert(group, DGroup { arrived_us, bytes: payload_bytes, objects: index.clone(), erased, seg: seq, frame_at: at, record_len: bytes.len() as u32 });
|
|
if let Some(o) = &old
|
|
&& let Some(n) = t.segs.get_mut(&o.seg)
|
|
{
|
|
*n = n.saturating_sub(1);
|
|
}
|
|
let (old_bytes, old_n) = old.as_ref().map_or((0, 0), |g| (g.bytes, g.objects.len()));
|
|
ix.window_bytes[window] = (ix.window_bytes[window] + payload_bytes).saturating_sub(old_bytes);
|
|
if let Some(o) = &old {
|
|
for x in &o.objects {
|
|
ix.unindex(&key, group, x);
|
|
}
|
|
}
|
|
for o in &index {
|
|
ix.reindex(&key, group, o);
|
|
}
|
|
let new_objects = index.len().saturating_sub(old_n) as u64;
|
|
if let Some(r) = room {
|
|
ix.series.entry(r).or_default().change(now_us(), payload_bytes as i128 - old_bytes as i128, new_objects, payload_bytes.saturating_sub(old_bytes), 0);
|
|
}
|
|
drop(ix);
|
|
if let Some(o) = old {
|
|
self.kill(w, &dir, o.seg, o.frame_at, o.record_len);
|
|
let open = w.open.get(&key).map(|s| s.seq);
|
|
self.retire(&key, o.seg, open);
|
|
}
|
|
self.stats.groups_written.fetch_add(1, Ordering::Relaxed);
|
|
self.stats.objects_written.fetch_add(new_objects, Ordering::Relaxed);
|
|
self.evict_by_bytes(w, window);
|
|
}
|
|
|
|
/// Erase what a redaction targets from every group that holds it: the group is written
|
|
/// again with the object among its erased, and its old record's bytes are overwritten.
|
|
/// An event record is erased from the events-track object that carries it, the object's
|
|
/// other records kept byte for byte; an object left with none is erased whole.
|
|
fn erase(&self, w: &mut Writer, room: Option<[u8; 32]>, digest: [u8; 32], why: u8) {
|
|
let places: Vec<(TrackKey, u64, u64)> = self.index.lock().unwrap().by_digest.get(&digest).cloned().unwrap_or_default();
|
|
for (key, g, oid) in places {
|
|
let (dir, grp, track_room) = {
|
|
let ix = self.index.lock().unwrap();
|
|
let Some(t) = ix.tracks.get(&key).filter(|t| room.is_none_or(|r| t.room == Some(r))) else { continue };
|
|
let Some(grp) = t.groups.get(&g).cloned() else { continue };
|
|
(t.dir.clone(), grp, t.room)
|
|
};
|
|
let Some(o) = grp.objects.iter().find(|o| o.id == oid).cloned() else { continue };
|
|
let Ok(rec) = read_at(&seg_path(&dir, grp.seg), grp.frame_at + FRAME, grp.record_len as usize) else { continue };
|
|
let Some(dg) = decode_group(&rec, &track_digest(&key)) else { continue };
|
|
let mut rows: Rows = grp.objects.iter().filter_map(|x| dg.objects.iter().find(|y| y.0.id == x.id).map(|(_, props, payload, _)| (x.clone(), props.clone(), payload.clone()))).collect();
|
|
let mut erased = grp.erased.clone();
|
|
// The object's new payload, when a record is taken out of it and others stay.
|
|
let replaced: Option<Bytes> = if o.digest == digest {
|
|
None
|
|
} else {
|
|
let Some((_, _, payload)) = rows.iter().find(|x| x.0.id == oid) else { continue };
|
|
events::withhold(payload, |d| *d == digest).map(|p| Bytes::from(p.into_owned()))
|
|
};
|
|
let delta: i128 = match &replaced {
|
|
None => {
|
|
rows.retain(|x| x.0.id != oid);
|
|
erased.push((oid, o.digest, why));
|
|
-(o.payload_len as i128)
|
|
}
|
|
Some(p) => {
|
|
for x in rows.iter_mut().filter(|x| x.0.id == oid) {
|
|
x.0.digest = ObjectReceipt::hash_payload(p);
|
|
x.0.records = events::digests(p);
|
|
x.2 = p.clone();
|
|
}
|
|
p.len() as i128 - o.payload_len as i128
|
|
}
|
|
};
|
|
let Some(index) = self.rewrite(w, &key, &dir, g, grp.arrived_us, &rows, &erased, (grp.seg, grp.frame_at, grp.record_len)) else { continue };
|
|
{
|
|
let mut ix = self.index.lock().unwrap();
|
|
ix.unindex(&key, g, &o);
|
|
if let Some(x) = index.iter().find(|x| x.id == oid) {
|
|
ix.reindex(&key, g, x);
|
|
}
|
|
if let Some(t) = ix.tracks.get_mut(&key) {
|
|
let window = t.window;
|
|
if let Some(gr) = t.groups.get_mut(&g) {
|
|
gr.bytes = (gr.bytes as i128 + delta).max(0) as u64;
|
|
}
|
|
ix.window_bytes[window] = (ix.window_bytes[window] as i128 + delta).max(0) as u64;
|
|
}
|
|
if let Some(r) = track_room {
|
|
ix.series.entry(r).or_default().change(now_us(), delta, 0, 0, 0);
|
|
}
|
|
}
|
|
if why == ERASED_TAKEN_DOWN {
|
|
self.stats.objects_taken_down.fetch_add(1, Ordering::Relaxed);
|
|
} else {
|
|
self.stats.objects_redacted.fetch_add(1, Ordering::Relaxed);
|
|
}
|
|
tracing::info!(target: "joule_moq_relay::dvr", dir = %dir.display(), group = g, object = oid, whole = replaced.is_none(), why, "an object erased from the store");
|
|
}
|
|
}
|
|
|
|
fn evict_by_bytes(&self, w: &mut Writer, window: usize) {
|
|
loop {
|
|
let victim = {
|
|
let ix = self.index.lock().unwrap();
|
|
if ix.window_bytes[window] <= self.windows[window].max_bytes {
|
|
return;
|
|
}
|
|
oldest(&ix, |t| t.window == window)
|
|
};
|
|
match victim {
|
|
Some((k, g)) => self.evict(w, &k, g),
|
|
None => return,
|
|
}
|
|
}
|
|
}
|
|
|
|
fn evict_by_age(&self, w: &mut Writer, now: u64) {
|
|
for (i, win) in self.windows.iter().enumerate() {
|
|
let floor = now.saturating_sub(win.max_age_us);
|
|
loop {
|
|
let victim = {
|
|
let ix = self.index.lock().unwrap();
|
|
oldest(&ix, |t| t.window == i).filter(|(k, g)| ix.tracks[k].groups[g].arrived_us < floor)
|
|
};
|
|
match victim {
|
|
Some((k, g)) => self.evict(w, &k, g),
|
|
None => break,
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
/// Evict group `g` of `key`, and group `g` of its receipts or media track with it. A
|
|
/// segment no group stands in any longer is deleted.
|
|
fn evict(&self, w: &mut Writer, key: &TrackKey, g: u64) {
|
|
let partner = match self.media_of(&key.1) {
|
|
Some(m) => (key.0.clone(), m),
|
|
None => (key.0.clone(), self.receipts_of(&key.1)),
|
|
};
|
|
for k in [key.clone(), partner] {
|
|
// The mark moves to `g`, and every group at or below it goes: none is ever held
|
|
// below a track's evicted mark. Groups are sealed out of order under load and a
|
|
// group's arrival time can be later than a higher group's, so evicting `g` alone
|
|
// could leave a lower group held while higher ones went: a gap in the window.
|
|
let victims: Vec<u64> = {
|
|
let mut ix = self.index.lock().unwrap();
|
|
let Some(t) = ix.tracks.get_mut(&k) else { continue };
|
|
let through = t.evicted_through.map_or(g, |e| e.max(g));
|
|
t.evicted_through = Some(through);
|
|
t.absent.retain(|_, hi| *hi > through);
|
|
t.groups.range(..=through).map(|(gg, _)| *gg).collect()
|
|
};
|
|
for v in victims {
|
|
let seg = {
|
|
let mut ix = self.index.lock().unwrap();
|
|
let Some(t) = ix.tracks.get_mut(&k) else { break };
|
|
let Some(grp) = t.groups.remove(&v) else { continue };
|
|
if let Some(n) = t.segs.get_mut(&grp.seg) {
|
|
*n = n.saturating_sub(1);
|
|
}
|
|
let (window, room) = (t.window, t.room);
|
|
ix.window_bytes[window] = ix.window_bytes[window].saturating_sub(grp.bytes);
|
|
for o in &grp.objects {
|
|
ix.unindex(&k, v, o);
|
|
}
|
|
if let Some(r) = room {
|
|
ix.series.entry(r).or_default().change(now_us(), -(grp.bytes as i128), 0, 0, 0);
|
|
}
|
|
grp.seg
|
|
};
|
|
self.stats.groups_evicted.fetch_add(1, Ordering::Relaxed);
|
|
let open = w.open.get(&k).map(|s| s.seq);
|
|
self.retire(&k, seg, open);
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
/// The oldest group held among the tracks `f` selects: by arrival, then track and group.
|
|
fn oldest(ix: &Index, f: impl Fn(&DTrack) -> bool) -> Option<(TrackKey, u64)> {
|
|
// Within a track, the lowest group goes first, so a track's window stays one run of
|
|
// groups: groups finish, and are sealed, out of order under load, and evicting by
|
|
// arrival left gaps below a late group. Across tracks, the track whose lowest group
|
|
// arrived first.
|
|
ix.tracks
|
|
.iter()
|
|
.filter(|(_, t)| f(t))
|
|
.filter_map(|(k, t)| t.groups.iter().next().map(|(g, grp)| (grp.arrived_us, k.clone(), *g)))
|
|
.min_by(|a, b| a.0.cmp(&b.0).then_with(|| a.2.cmp(&b.2)))
|
|
.map(|(_, k, g)| (k, g))
|
|
}
|
|
|
|
#[cfg(test)]
|
|
mod tests {
|
|
use super::*;
|
|
|
|
#[test]
|
|
fn a_group_record_reads_back_and_a_flipped_byte_is_found() {
|
|
let td = [7u8; 32];
|
|
let o = |id: u64| DObject { id, subgroup: Some(0), priority: 3, status: ObjectStatus::Normal, at: 0, props_len: 0, payload_len: 0, digest: ObjectReceipt::hash_payload(format!("p{id}").as_bytes()), records: vec![] };
|
|
let rows: Rows = (0..3).map(|i| (o(i), Bytes::new(), Bytes::from(format!("p{i}")))).collect();
|
|
let (mut b, index) = encode_group(&td, 5, 99, &rows, &[(3, [1; 32], ERASED_REDACTED)]);
|
|
let dg = decode_group(&b, &td).unwrap();
|
|
assert_eq!((dg.group, dg.arrived_us, dg.count, dg.erased.len()), (5, 99, 3, 1));
|
|
assert!(dg.objects.iter().all(|x| x.3));
|
|
assert_eq!(index[1].at, dg.objects[1].0.at);
|
|
// A flipped payload byte fails that object's check alone.
|
|
let at = index[1].at as usize;
|
|
b[at] ^= 0x40;
|
|
let dg = decode_group(&b, &td).unwrap();
|
|
assert_eq!(dg.objects.iter().map(|x| x.3).collect::<Vec<_>>(), vec![true, false, true]);
|
|
// A flipped header byte drops the group; so does another track's record.
|
|
b[40] ^= 1;
|
|
assert!(decode_group(&b, &td).is_none());
|
|
let (b2, _) = encode_group(&td, 5, 99, &rows, &[]);
|
|
assert!(decode_group(&b2, &[8; 32]).is_none());
|
|
}
|
|
|
|
#[test]
|
|
fn a_track_record_reads_back() {
|
|
let key = (TrackNamespace::from_strs(["r", "p", "video"]), TrackName::from_str("0"));
|
|
let mut p = Properties::new();
|
|
p.push(joule_moq::draft22::kvp::KeyValuePair::varint(prop::DEFAULT_PUBLISHER_PRIORITY, 7));
|
|
let b = encode_track(&key, &p);
|
|
assert_eq!(decode_track(&b), Some((key, p)));
|
|
let mut bad = b.clone();
|
|
bad[7] ^= 1;
|
|
assert!(decode_track(&bad).is_none());
|
|
}
|
|
|
|
#[test]
|
|
fn the_series_integrates_what_is_held() {
|
|
let mut s = Series::default();
|
|
s.change(1_000_000, 100, 1, 100, 0);
|
|
s.change(3_000_000, 50, 1, 50, 0);
|
|
// 100 bytes for 2 s, then 150.
|
|
assert_eq!(s.at(3_000_000).0, 100 * 2_000_000);
|
|
assert_eq!(s.at(4_000_000).0, 100 * 2_000_000 + 150 * 1_000_000);
|
|
assert_eq!(s.at(0), (0, 0, 0, 0));
|
|
assert_eq!((s.at(5_000_000).1, s.at(5_000_000).2), (2, 150));
|
|
}
|
|
|
|
#[test]
|
|
fn segment_names_read_back() {
|
|
assert_eq!(seg_seq(&seg_path(Path::new("/x"), 42)), Some(42));
|
|
assert_eq!(seg_seq(Path::new("/x/TRACK")), None);
|
|
}
|
|
}
|