//! Durable descriptor log: append-only relation-shape history, captured at

//! catalog-commit boundaries, read interval-scoped by decode.

//!

//! Decode asks `descriptor(rfn, L)` / `descriptor(oid, L)` — a point-in-time

//! question. The log stores per-key version chains (`valid_from` ascending);

//! lookup binary-searches the last entry at or below `L`. Capture appends one

//! batch per catalog boundary keyed `captured_at` = the commit's EndRecPtr,

//! fsyncs, then publishes to the in-memory index — records reaching decode

//! always have coverage.

//!

//! ## On-disk layout

//!

//! Two files under the spill dir: `desc_log.ckpt` (compacted snapshot,

//! replaced atomically at GC) + `desc_log.tail` (framed appends since).

//! Both open with the same header:

//!

//! ```text

//! [2 "WL"] [u16 LE version] [u32 pg_major] [str system_id] [u32 timeline]

//! [u32 db_oid] [u32 wal_seg_size]

//! ```

//!

//! then frames `[u32 LE len] [u32 LE crc32c(body)] [body]`; `body[0]` tags

//! `0` meta (ckpt only: `covered_through`, `floor_at_write`) or `1` batch.

//! Identity mismatch is fatal (foreign spill dir), version mismatch is fatal

//! (no cross-version compatibility by design).

//!

//! Tail repair: a frame extending past EOF or failing CRC as the final frame

//! is a torn append — truncate + `sync_data`. CRC failure with valid data

//! after is interior corruption — fail closed. GC writes the ckpt first,

//! then truncates the tail; a crash between leaves duplicate batches, which

//! load dedupes by `captured_at` (byte-equal skip, divergent fail-closed).

//!

//! ## Interval semantics

//!

//! `Present` opens an interval at `valid_from` (bias-early: the descriptor is

//! a backward-compatible reader of older tuples, never the reverse).

//! `Dropped` tombstones relation + rfn. `Retired` closes an rfn's interval on

//! filenode rotation without relation-level effects: rewrite/TRUNCATE hold

//! AccessExclusiveLock so no decode query lands past it — the entry exists so

//! GC can drop rotated-away chains and buggy callers fail closed, not open.

//!

//! ## Ambiguity intervals

//!

//! A boundary whose shape change cannot be proven safe for one descriptor

//! records an [`Ambiguity`]: a `[from_lsn, through_lsn)` interval scoped to

//! an rfn, oid, or whole database. Lookup consults ambiguity intervals

//! before the descriptor chain — a covered LSN answers

//! [`LookupResult::Ambiguous`] even when a chain entry exists, so callers

//! fail closed instead of decoding under an unproven layout. The final

//! post-commit `Present` entry stays usable past `through_lsn`

//!

//! Each interval files under exactly one key, and each lookup consults its

//! own key plus the database scope, so producers publish one interval per

//! identity key they can name — capture emits an `Rfn` and an `Oid` interval

//! for the same verdict. Deferred records resolve at the commit's

//! `next_lsn`, past their own interval's end, so the drain fences per record

//! against [`DescriptorLog::ambiguities_intersecting`] rather than relying

//! on the point lookup

//!

//! Identity keys the full physical `RelFileNode`: PG guarantees

//! relfilenumber uniqueness only per database of one tablespace

//! (`GetNewRelFileNumber`, PG `src/backend/catalog/catalog.c`), so

//! `(db_node, rel_node)` alone can alias two live relations after OID

//! wraparound. Capture resolves the `pg_class.reltablespace` 0 sentinel to

//! the database's `dattablespace`, so stored rfns compare directly against

//! WAL locators' physical spcOid. See `plans/tablespaces.md`

use std::collections::BTreeMap;

use std::io::SeekFrom;

use std::path::{Path, PathBuf};

use std::sync::atomic::AtomicU64;

use std::sync::{Arc, RwLock};

use thiserror::Error;

use tokio::fs::{File, OpenOptions};

use tokio::io::{AsyncSeekExt, AsyncWriteExt};

use tokio_postgres::types::Oid;

use walrus::pg::walparser::RelFileNode;

use crate::pos::{Floor, Pos};

use crate::schema::{MissingDefault, RelAttr, RelDescriptor, RelName, ReplIdent};

use ahash::{HashMap, HashSet, HashSetExt};

pub const CKPT_FILE: &str = "desc_log.ckpt";

pub const TAIL_FILE: &str = "desc_log.tail";

const MAGIC: &[u8; 2] = b"WL";

/// 3: ambiguity intervals fence per record at the drain. A v2 log's

/// intervals were published for read-identical drift too and span past the

/// layout change, so replaying them under the fence would fail closed on

/// rows v2 decoded fine — reject the log instead

const VERSION: u16 = 3;

/// Reject frames past this before allocating: no realistic batch (thousands

/// of descriptors) approaches it, garbage lengths do

const MAX_FRAME: u32 = 256 * 1024 * 1024;

const TAG_META: u8 = 0;

const TAG_BATCH: u8 = 1;

/// GC when this many entries are droppable below the floor

const GC_DEAD_ENTRIES: usize = 512;

/// or when the tail outgrows this

const GC_TAIL_BYTES: u64 = 8 * 1024 * 1024;

pub type Result<T> = std::result::Result<T, DescLogError>;

#[derive(Debug, Error)]

pub enum DescLogError {

#[error("descriptor log io: {0}")]

Io(#[from] std::io::Error),

#[error("unsupported descriptor log version {0} (this build expects {VERSION})")]

Version(u16),

#[error("foreign descriptor log: {field} mismatch (log has {log}, this source is {ours})")]

ForeignLog {

field: &'static str,

log: String,

ours: String,

},

#[error("descriptor log corrupt at {file}:{offset}: {detail}")]

Corrupt {

file: &'static str,

offset: u64,

detail: String,

},

}

/// Binds log files to one source + shadow pairing; any mismatch at open is

/// fatal, mirroring `ManifestError::ForeignSource`.

#[derive(Debug, Clone, PartialEq, Eq)]

pub struct DescLogIdentity {

pub pg_major: u32,

pub system_id: String,

pub timeline: u32,

pub db_oid: Oid,

pub wal_seg_size: u32,

}

#[derive(Debug, Clone, PartialEq)]

pub enum LogValue {

Present(Arc<RelDescriptor>),

/// Relation dropped: tombstones oid + rfn chains

Dropped,

/// Filenode rotated away (rewrite/TRUNCATE/SET TABLESPACE): closes the

/// rfn chain only, relation lives on under its new rfn

Retired,

}

#[derive(Debug, Clone, PartialEq)]

pub struct LogEntry {

pub valid_from: u64,

pub oid: Oid,

/// Full physical rfn (tablespace sentinel resolved at capture). For

/// `Present` this is the descriptor's rfn; for tombstones the chain

/// being closed

pub rfn: RelFileNode,

pub value: LogValue,

}

/// Interval `[from_lsn, through_lsn)` where no single descriptor provably

/// decodes rows in `scope`

#[derive(Debug, Clone, PartialEq, Eq)]

pub struct Ambiguity {

pub scope: AmbiguityScope,

pub from_lsn: u64,

pub through_lsn: u64,

pub reason: AmbiguityReason,

}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]

pub enum AmbiguityScope {

Rfn(RelFileNode),

Oid(Oid),

/// Conservative fallback when affected relations cannot be enumerated

Database(Oid),

}

/// Discriminants are the wire tags; only produced variants exist, so an

/// operator never sees a label that cannot fire

#[derive(Debug, Clone, Copy, PartialEq, Eq)]

#[repr(u8)]

pub enum AmbiguityReason {

/// Descriptor changed in place and the dirty interval's exact mutation

/// position is unknown (only the first catalog touch is tracked)

UnknownMutationPosition = 1,

}

#[derive(Debug, Clone, PartialEq)]

pub struct BatchRecord {

/// Boundary commit EndRecPtr (or seed LSN); the replay-from-log key

pub captured_at: u64,

/// Commit record start LSN; 0 for seed batches

pub commit_lsn: u64,

/// Evidence the boundary verdict derives from, kept so replay

/// reproduces the verdict instead of reinferring it from current

/// catalog. Deterministic order (capture sorts)

pub observations: Vec<RelationObservation>,

/// Intervals this boundary could not prove decodable under one

/// descriptor

pub ambiguities: Vec<Arc<Ambiguity>>,

/// Empty = stub: boundary produced no shape change, recorded so boot

/// replay distinguishes "captured, nothing changed" from "never captured"

pub entries: Vec<Arc<LogEntry>>,

}

impl BatchRecord {

/// Digest over the deterministic encoding; divergence diagnostics

pub fn digest(&self) -> u32 {

crc32c::crc32c(&encode_batch(self))

}

}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]

pub struct RelationObservation {

pub oid: Option<Oid>,

pub rfn: Option<RelFileNode>,

pub first_touch_lsn: u64,

/// Main-fork smgr create marker: new generation lower bound

pub smgr_create_lsn: Option<u64>,

pub kind: ObservationKind,

}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]

pub enum ObservationKind {

/// Dirty-tracker pg_class decode or commit relcache inval named the oid

AffectedOid,

/// Smgr create marker registered for the filenode

SmgrCreate,

/// Enumeration incomplete, capture fell back to full catalog scan

FullScan,

}

#[derive(Debug, Clone, PartialEq)]

pub enum LookupResult {

Present(Arc<RelDescriptor>),

Dropped,

Retired,

/// LSN falls inside a recorded ambiguity interval: no descriptor is

/// proven safe, callers fail closed

Ambiguous(Arc<Ambiguity>),

NotCovered,

/// Foreign `db_node`: preserves the `ForeignDatabase` row-skip control

/// flow — never a stash or a fatal

ForeignDb,

}

/// One key's version chain slot; `captured_at` denormalized for

/// [`DescriptorLog::predecessor_before`]

#[derive(Debug, Clone)]

struct Slot {

valid_from: u64,

captured_at: u64,

entry: Arc<LogEntry>,

}

#[derive(Debug, Default)]

struct Index {

by_rfn: HashMap<RelFileNode, Vec<Slot>>,

by_oid: HashMap<Oid, Vec<Slot>>,

amb_rfn: HashMap<RelFileNode, Vec<Arc<Ambiguity>>>,

amb_oid: HashMap<Oid, Vec<Arc<Ambiguity>>>,

amb_db: HashMap<Oid, Vec<Arc<Ambiguity>>>,

batches: BTreeMap<u64, Arc<BatchRecord>>,

covered_through: u64,

floor_at_write: Pos<Floor>,

entries_total: usize,

}

impl Index {

/// Files each ambiguity under exactly the key its scope names. Producers

/// publish one interval per identity key so the rfn-keyed and oid-keyed

/// lookups see the same fence without either consulting the other's map

fn insert_batch(&mut self, batch: Arc<BatchRecord>) {

for amb in &batch.ambiguities {

let list = match amb.scope {

AmbiguityScope::Rfn(rfn) => self.amb_rfn.entry(rfn).or_default(),

AmbiguityScope::Oid(oid) => self.amb_oid.entry(oid).or_default(),

AmbiguityScope::Database(db) => self.amb_db.entry(db).or_default(),

};

let key = (amb.from_lsn, amb.through_lsn);

let pos = list.partition_point(|a| (a.from_lsn, a.through_lsn) <= key);

list.insert(pos, amb.clone());

}

for entry in &batch.entries {

let slot = Slot {

valid_from: entry.valid_from,

captured_at: batch.captured_at,

entry: entry.clone(),

};

let rfn_chain = self.by_rfn.entry(entry.rfn).or_default();

// Retired closes an rfn chain only; the relation lives on under

// its new rfn, whose Present entry shares this oid + batch — a

// Retired slot in by_oid would shadow it

if !matches!(entry.value, LogValue::Retired) {

let oid_chain = self.by_oid.entry(entry.oid).or_default();

insert_sorted(oid_chain, slot.clone());

}

insert_sorted(rfn_chain, slot);

self.entries_total += 1;

}

self.batches.insert(batch.captured_at, batch);

}

}

/// Chains stay `(valid_from, captured_at)`-ascending. Appends hit the tail

/// (DDL on one rel serializes under AccessExclusiveLock, so per-key order is

/// total); boot replay re-inserts nothing (batch dedupe upstream).

fn insert_sorted(chain: &mut Vec<Slot>, slot: Slot) {

let key = (slot.valid_from, slot.captured_at);

let pos = chain.partition_point(|s| (s.valid_from, s.captured_at) <= key);

chain.insert(pos, slot);

}

#[derive(Debug)]

struct Writer {

tail: File,

tail_len: u64,

header_len: u64,

dir: PathBuf,

}

crate::atomic_stats! {

pub struct DescLogStats {

pub lookups_present,

pub lookups_dropped,

pub lookups_retired,

pub lookups_ambiguous,

pub lookups_not_covered,

pub lookups_foreign_db,

pub batches_appended,

pub gc_runs,

pub gc_dropped_entries,

}

}

#[derive(Debug)]

pub struct DescriptorLog {

identity: DescLogIdentity,

index: RwLock<Index>,

writer: tokio::sync::Mutex<Writer>,

/// `tail_len - header_len` mirror, readable without the writer lock

tail_bytes: AtomicU64,

stats: Arc<DescLogStats>,

}

impl DescriptorLog {

/// Load `desc_log.ckpt` + `desc_log.tail` under `dir`, repairing a torn

/// tail. Creates an empty tail (header only) when absent. Never creates

/// the ckpt — [`Self::seed`] and GC own it.

pub async fn open(dir: &Path, identity: DescLogIdentity) -> Result<Self> {

Self::open_on_branch(dir, identity, &[]).await

}

/// [`open`](Self::open) with the branches a stored header may name: every

/// timeline the live chain places between the log's origin and

/// `identity.timeline`. A crossing moves the resume branch without moving

/// the log, so the header lags until the next GC rewrites it.

pub async fn open_on_branch(

dir: &Path,

identity: DescLogIdentity,

lineage: &[u32],

) -> Result<Self> {

let mut index = Index::default();

let ckpt_path = dir.join(CKPT_FILE);

if let Ok(bytes) = tokio::fs::read(&ckpt_path).await {

// write_atomic guarantees complete-or-absent: any parse failure

// here is corruption, never a torn write

load_frames(&bytes, CKPT_FILE, &identity, lineage, &mut index, false)?;

}

let tail_path = dir.join(TAIL_FILE);

let header = encode_header(&identity);

let (tail, tail_len) = match tokio::fs::read(&tail_path).await {

Ok(bytes) => {

let good_len =

load_frames(&bytes, TAIL_FILE, &identity, lineage, &mut index, true)?;

let mut f = OpenOptions::new().write(true).open(&tail_path).await?;

if good_len < bytes.len() as u64 {

f.set_len(good_len).await?;

f.sync_data().await?;

}

f.seek(SeekFrom::Start(good_len)).await?;

(f, good_len)

}

Err(e) if e.kind() == std::io::ErrorKind::NotFound => {

let mut f = OpenOptions::new()

.write(true)

.create_new(true)

.open(&tail_path)

.await?;

f.write_all(&header).await?;

f.sync_data().await?;

crate::fs::fsync_dir(dir).await?;

(f, header.len() as u64)

}

Err(e) => return Err(e.into()),

};

Ok(Self {

identity,

index: RwLock::new(index),

writer: tokio::sync::Mutex::new(Writer {

tail,

tail_len,

header_len: header.len() as u64,

dir: dir.to_path_buf(),

}),

tail_bytes: AtomicU64::new(tail_len - header.len() as u64),

stats: Arc::new(DescLogStats::default()),

})

}

fn publish_tail_bytes(&self, w: &Writer) {

self.tail_bytes.store(

w.tail_len - w.header_len,

std::sync::atomic::Ordering::Relaxed,

);

}

pub fn stats_handle(&self) -> Arc<DescLogStats> {

self.stats.clone()

}

/// Database this log describes; every entry's oid is scoped to it

pub fn db_oid(&self) -> Oid {

self.identity.db_oid

}

pub fn covered_through(&self) -> u64 {

self.index.read().unwrap().covered_through

}

/// Floor recorded by the last GC ckpt write; `--start-lsn` below it has

/// no descriptor history

pub fn floor_at_write(&self) -> Pos<Floor> {

self.index.read().unwrap().floor_at_write

}

/// Highest `captured_at`, 0 when empty

pub fn head(&self) -> u64 {

self.index

.read()

.unwrap()

.batches

.keys()

.next_back()

.copied()

.unwrap_or(0)

}

pub fn is_empty(&self) -> bool {

let idx = self.index.read().unwrap();

idx.batches.is_empty() && idx.covered_through == 0

}

pub fn descriptor_at(&self, rfn: RelFileNode, lsn: u64) -> LookupResult {

match self.descriptor_at_spanned(rfn, lsn) {

Ok((rel, _)) => LookupResult::Present(rel),

Err(other) => other,

}

}

/// `Present` descriptor plus its entry's `valid_from` under one index

/// read (two reads could interleave with a bias-early capture append

/// and pair a stale descriptor with a fresh span). Non-Present outcomes

/// return the plain lookup for the caller's arms

pub fn descriptor_at_spanned(

&self,

rfn: RelFileNode,

lsn: u64,

) -> std::result::Result<(Arc<RelDescriptor>, u64), LookupResult> {

use std::sync::atomic::Ordering::Relaxed;

if rfn.db_node != 0 && rfn.db_node != self.identity.db_oid {

self.stats.lookups_foreign_db.fetch_add(1, Relaxed);

return Err(LookupResult::ForeignDb);

}

let idx = self.index.read().unwrap();

// Ambiguity precedes the chain: a chain entry inside an ambiguous

// interval is not proven safe for rows there. The db scope keys on

// the locator's own db_node, so a shared relation (db_node 0) skips

// this database's Database-scoped intervals — shared catalogs carry

// no descriptors, so no such interval can name their rows

let result = ambiguity_covering(idx.amb_rfn.get(&rfn), lsn)

.or_else(|| ambiguity_covering(idx.amb_db.get(&rfn.db_node), lsn))

.map_or_else(

|| lookup_spanned(idx.by_rfn.get(&rfn), lsn),

|a| Err(LookupResult::Ambiguous(a)),

);

self.count_spanned(&result);

result

}

/// Oid lookup for a caller that already proved the oid belongs to this

/// log's database. Production callers take

/// [`Self::descriptor_by_oid_in_db_at_spanned`] instead, which cannot

/// be used without supplying a database

pub fn descriptor_by_oid_at(&self, oid: Oid, lsn: u64) -> LookupResult {

match self.descriptor_by_oid_at_spanned(oid, lsn) {

Ok((rel, _)) => LookupResult::Present(rel),

Err(other) => other,

}

}

/// Oid-keyed twin of [`Self::descriptor_at_spanned`], scoped in one

/// step. Relation OIDs are unique only within one database, and WAL

/// carrying bare OIDs (`xl_heap_truncate`) names its own: comparing it

/// with this log's identity before the OID chain is read keeps a

/// foreign record from resolving a local relation

pub fn descriptor_by_oid_in_db_at_spanned(

&self,

db_oid: Oid,

oid: Oid,

lsn: u64,

) -> std::result::Result<(Arc<RelDescriptor>, u64), LookupResult> {

if db_oid != self.identity.db_oid {

self.stats

.lookups_foreign_db

.fetch_add(1, std::sync::atomic::Ordering::Relaxed);

return Err(LookupResult::ForeignDb);

}

self.descriptor_by_oid_at_spanned(oid, lsn)

}

fn descriptor_by_oid_at_spanned(

&self,

oid: Oid,

lsn: u64,

) -> std::result::Result<(Arc<RelDescriptor>, u64), LookupResult> {

let idx = self.index.read().unwrap();

let result = ambiguity_covering(idx.amb_oid.get(&oid), lsn)

.or_else(|| ambiguity_covering(idx.amb_db.get(&self.identity.db_oid), lsn))

.map_or_else(

|| lookup_spanned(idx.by_oid.get(&oid), lsn),

|a| Err(LookupResult::Ambiguous(a)),

);

self.count_spanned(&result);

result

}

fn count_spanned(&self, result: &std::result::Result<(Arc<RelDescriptor>, u64), LookupResult>) {

use std::sync::atomic::Ordering::Relaxed;

match result {

Ok(_) => {

self.stats.lookups_present.fetch_add(1, Relaxed);

}

Err(other) => self.count(other),

}

}

fn count(&self, result: &LookupResult) {

use std::sync::atomic::Ordering::Relaxed;

match result {

LookupResult::Present(_) => self.stats.lookups_present.fetch_add(1, Relaxed),

LookupResult::Dropped => self.stats.lookups_dropped.fetch_add(1, Relaxed),

LookupResult::Retired => self.stats.lookups_retired.fetch_add(1, Relaxed),

LookupResult::Ambiguous(_) => self.stats.lookups_ambiguous.fetch_add(1, Relaxed),

LookupResult::NotCovered => self.stats.lookups_not_covered.fetch_add(1, Relaxed),

LookupResult::ForeignDb => self.stats.lookups_foreign_db.fetch_add(1, Relaxed),

};

}

/// Intervals overlapping `[from, through)` for a relation, under every

/// key that can name it. The drain's per-record fence: a stashed record

/// resolves at the commit's `next_lsn`, past its own interval's

/// `through_lsn`, so the point lookups above cannot see the fence that

/// covers it.

///

/// Overlap, not containment — an interval opening below `from` still

/// covers records inside the range, and the caller narrows per record

pub fn ambiguities_intersecting(

&self,

rfn: RelFileNode,

oid: Option<Oid>,

from: u64,

through: u64,

) -> Vec<Arc<Ambiguity>> {

if rfn.db_node != 0 && rfn.db_node != self.identity.db_oid {

return Vec::new();

}

let idx = self.index.read().unwrap();

let mut out: Vec<Arc<Ambiguity>> = Vec::new();

let lists = [

idx.amb_rfn.get(&rfn),

oid.and_then(|o| idx.amb_oid.get(&o)),

idx.amb_db.get(&rfn.db_node),

];

for amb in lists.into_iter().flatten().flatten() {

// Siblings of one verdict are distinct allocations; the same Arc

// reaches here only when two keys index it

if amb.from_lsn < through

&& from < amb.through_lsn

&& !out.iter().any(|k| Arc::ptr_eq(k, amb))

{

out.push(amb.clone());

}

}

out.sort_unstable_by_key(|a| (a.from_lsn, a.through_lsn));

out

}

/// Newest `Present` at or below `lsn` on the rfn chain, skipping

/// tombstones. Truncate apply resolves the relation an rfn named even

/// after its own commit retired it: rotation records `Retired` at the

/// new generation's bias-early valid_from, which precedes the truncate

/// record itself. Bypasses the ambiguity fence: the caller is applying a

/// TRUNCATE, which needs the relation's identity, not its tuple layout.

pub fn present_before(&self, rfn: RelFileNode, lsn: u64) -> Option<Arc<RelDescriptor>> {

let idx = self.index.read().unwrap();

let chain = idx.by_rfn.get(&rfn)?;

let upto = chain.partition_point(|s| s.valid_from <= lsn);

chain[..upto]

.iter()

.rev()

.find_map(|s| match &s.entry.value {

LogValue::Present(d) => Some(d.clone()),

_ => None,

})

}

/// Replay-from-log hit test for a boundary at `captured_at`

pub fn batch_at(&self, captured_at: u64) -> Option<Arc<BatchRecord>> {

self.index

.read()

.unwrap()

.batches

.get(&captured_at)

.cloned()

}

/// The oid's entry preceding the batch at `captured_at` in history

/// order — replay derives events against this, never the loaded head

/// (boot loads the full log before the WAL re-read). Bypasses the

/// ambiguity fence by construction: capture compares descriptors here,

/// it does not read tuples

pub fn predecessor_before(&self, oid: Oid, captured_at: u64) -> Option<Arc<LogEntry>> {

let idx = self.index.read().unwrap();

let chain = idx.by_oid.get(&oid)?;

let pos = chain.partition_point(|s| s.captured_at < captured_at);

pos.checked_sub(1).map(|i| chain[i].entry.clone())

}

/// Ambiguity intervals recorded for an rfn, `(from_lsn, through_lsn)`

/// ascending — introspection for diagnostics and tests

pub fn rfn_ambiguities(&self, rfn: RelFileNode) -> Vec<Arc<Ambiguity>> {

let idx = self.index.read().unwrap();

idx.amb_rfn.get(&rfn).cloned().unwrap_or_default()

}

/// Oid-keyed twin of [`Self::rfn_ambiguities`]: the sibling interval a

/// physical in-place verdict files for the same event

pub fn oid_ambiguities(&self, oid: Oid) -> Vec<Arc<Ambiguity>> {

let idx = self.index.read().unwrap();

idx.amb_oid.get(&oid).cloned().unwrap_or_default()

}

/// Oids whose newest entry is `Present` — capture-all diffs its SQL

/// enumeration against this to tombstone vanished relations

pub fn present_oids(&self) -> Vec<Oid> {

let idx = self.index.read().unwrap();

idx.by_oid

.iter()

.filter(|(_, chain)| {

matches!(

chain.last().map(|s| &s.entry.value),

Some(LogValue::Present(_))

)

})

.map(|(oid, _)| *oid)

.collect()

}

/// Descriptors `Present` at `lsn` — boot Added enumeration. Bypasses the

/// ambiguity fence: an ambiguous interval says nothing about whether the

/// relation exists, and the consumer only emits idempotent

/// `CREATE TABLE IF NOT EXISTS`, never reads a tuple

pub fn active_present_at(&self, lsn: u64) -> Vec<Arc<RelDescriptor>> {

let idx = self.index.read().unwrap();

idx.by_oid

.values()

.filter_map(|chain| Some(lookup_spanned(Some(chain), lsn).ok()?.0))

.collect()

}

/// Names of user relations `Present` at `lsn`, TOAST and system schemas

/// dropped — the catalog a name pattern resolves against

pub fn user_rel_names_at(&self, lsn: u64, config_schema: Option<&str>) -> Vec<RelName> {

self.active_present_at(lsn)

.into_iter()

.filter(|d| {

d.kind != 't'

&& !crate::emit::ch_ddl::is_system_namespace(

&d.rel_name.namespace,

config_schema,

)

})

.map(|d| d.rel_name.clone())

.collect()

}

/// One-time baseline on an empty log: writes the ckpt (meta +

/// seed batch) so `covered_through` is durable before any tail append.

/// Boundaries at or below `covered_through` are baked into the seed —

/// no capture, no event replay.

pub async fn seed(&self, batch: BatchRecord, covered_through: u64) -> Result<()> {

let w = self.writer.lock().await;

{

let idx = self.index.read().unwrap();

if !idx.batches.is_empty() || idx.covered_through != 0 {

return Err(DescLogError::Corrupt {

file: CKPT_FILE,

offset: 0,

detail: "seed on a non-empty log".into(),

});

}

}

let batch = Arc::new(batch);

let mut bytes = encode_header(&self.identity);

push_frame(&mut bytes, &encode_meta(covered_through, 0));

push_frame(&mut bytes, &encode_batch(&batch));

crate::fs::write_atomic(&w.dir, CKPT_FILE, &bytes).await?;

let mut idx = self.index.write().unwrap();

idx.covered_through = covered_through;

idx.insert_batch(batch);

Ok(())

}

/// Append one boundary batch: frame → `sync_data` → index publish.

/// Idempotent per `captured_at` (byte-identical replays no-op); a

/// divergent batch at an existing key fails closed.

pub async fn append_batch(&self, batch: BatchRecord) -> Result<()> {

let mut w = self.writer.lock().await;

{

let idx = self.index.read().unwrap();

if let Some(existing) = idx.batches.get(&batch.captured_at) {

if **existing == batch {

return Ok(());

}

return Err(DescLogError::Corrupt {

file: TAIL_FILE,

offset: w.tail_len,

detail: format!(

"append at captured_at {:#x} diverges from stored batch \

(digest {:#010x} vs stored {:#010x})",

batch.captured_at,

batch.digest(),

existing.digest(),

),

});

}

}

let batch = Arc::new(batch);

let mut frame = Vec::new();

push_frame(&mut frame, &encode_batch(&batch));

w.tail.write_all(&frame).await?;

w.tail.sync_data().await?;

w.tail_len += frame.len() as u64;

self.publish_tail_bytes(&w);

self.stats

.batches_appended

.fetch_add(1, std::sync::atomic::Ordering::Relaxed);

self.index.write().unwrap().insert_batch(batch);

Ok(())

}

/// Compact when the tail outgrew `GC_TAIL_BYTES` or at least

/// `GC_DEAD_ENTRIES` entries are droppable below `floor`. Retention

/// keeps, per key, the state active at the floor: the last entry at or

/// below it when `Present`; a `Dropped`/`Retired` there drops the whole

/// at-or-below history (nothing above can reference it — records

/// predate the drop, and the floor never exceeds the re-read start).

/// Batches above the floor survive whole (stubs included) for boot

/// replay; below it they exist only as carriers of retained entries.

pub async fn maybe_gc(&self, floor: Pos<Floor>) -> Result<bool> {

let w = self.writer.lock().await;

let tail_bytes = w.tail_len - w.header_len;

let (retained, dropped_entries) = {

let idx = self.index.read().unwrap();

// Count before rebuilding: this runs at cursor-write cadence and

// usually declines, while compute_retained re-sorts every chain

let dropped = droppable_entries(&idx, floor.get());

if dropped < GC_DEAD_ENTRIES && tail_bytes < GC_TAIL_BYTES {

return Ok(false);

}

(compute_retained(&idx, floor.get()), dropped)

};

self.gc_locked(w, retained, dropped_entries as u64, floor)

.await?;

Ok(true)

}

#[cfg(test)]

pub async fn force_gc(&self, floor: Pos<Floor>) -> Result<()> {

let w = self.writer.lock().await;

let (retained, dropped) = {

let idx = self.index.read().unwrap();

let retained = compute_retained(&idx, floor.get());

let dropped = idx.entries_total - retained.entries_total;

(retained, dropped)

};

self.gc_locked(w, retained, dropped as u64, floor).await

}

async fn gc_locked(

&self,

mut w: tokio::sync::MutexGuard<'_, Writer>,

mut retained: Index,

dropped_entries: u64,

floor: Pos<Floor>,

) -> Result<()> {

use std::sync::atomic::Ordering::Relaxed;

retained.floor_at_write = floor;

let mut bytes = encode_header(&self.identity);

push_frame(

&mut bytes,

&encode_meta(retained.covered_through, floor.get()),

);

for batch in retained.batches.values() {

push_frame(&mut bytes, &encode_batch(batch));

}

crate::fs::write_atomic(&w.dir, CKPT_FILE, &bytes).await?;

// Crash before this truncate leaves ckpt+tail overlapping; load

// dedupes by captured_at

let header_len = w.header_len;

w.tail.set_len(header_len).await?;

w.tail.seek(SeekFrom::Start(header_len)).await?;

w.tail.sync_data().await?;

w.tail_len = w.header_len;

self.publish_tail_bytes(&w);

*self.index.write().unwrap() = retained;

self.stats.gc_runs.fetch_add(1, Relaxed);

self.stats

.gc_dropped_entries

.fetch_add(dropped_entries, Relaxed);

Ok(())

}

/// (entries, bytes-in-tail, batches) gauges. Tail bytes come from the

/// mirror, not the writer: GC runs off the pump task and would otherwise

/// dip the gauge to 0 for the length of a compaction

pub fn gauges(&self) -> (u64, u64, u64) {

let idx = self.index.read().unwrap();

let entries = idx.entries_total as u64;

let batches = idx.batches.len() as u64;

drop(idx);

(

entries,

self.tail_bytes.load(std::sync::atomic::Ordering::Relaxed),

batches,

)

}

}

/// Newest interval covering `lsn` under `[from_lsn, through_lsn)`. Lists are

/// `(from_lsn, through_lsn)`-sorted, so `partition_point` bounds the search

/// to intervals that opened at or below `lsn` and the reverse walk answers

/// from the newest — GC only drops intervals below the floor, so the list

/// tracks floor lag rather than workload and the prefix skip matters

fn ambiguity_covering(list: Option<&Vec<Arc<Ambiguity>>>, lsn: u64) -> Option<Arc<Ambiguity>> {

let list = list?;

let opened = list.partition_point(|a| a.from_lsn <= lsn);

list[..opened]

.iter()

.rev()

.find(|a| lsn < a.through_lsn)

.cloned()

}

fn lookup_spanned(

chain: Option<&Vec<Slot>>,

lsn: u64,

) -> std::result::Result<(Arc<RelDescriptor>, u64), LookupResult> {

let Some(chain) = chain else {

return Err(LookupResult::NotCovered);

};

let pos = chain.partition_point(|s| s.valid_from <= lsn);

let Some(slot) = pos.checked_sub(1).map(|i| &chain[i]) else {

return Err(LookupResult::NotCovered);

};

match &slot.entry.value {

LogValue::Present(d) => Ok((d.clone(), slot.valid_from)),

LogValue::Dropped => Err(LookupResult::Dropped),

LogValue::Retired => Err(LookupResult::Retired),

}

}

/// Per key, the entry whose state is active at `floor`. Entry identity by

/// allocation: each `Arc<LogEntry>` is inserted once and shared between its

/// batch and both chains

fn keep_at_floor(idx: &Index, floor: u64) -> HashSet<*const LogEntry> {

let mut keep: HashSet<*const LogEntry> = HashSet::new();

for chain in idx.by_rfn.values().chain(idx.by_oid.values()) {

let below = &chain[..chain.partition_point(|s| s.valid_from <= floor)];

if let Some(last) = below.last()

&& matches!(last.entry.value, LogValue::Present(_))

{

keep.insert(Arc::as_ptr(&last.entry));

}

}

keep

}

/// Entries [`compute_retained`] would drop at `floor`, without building the

/// retained index — the GC threshold test

fn droppable_entries(idx: &Index, floor: u64) -> usize {

let keep = keep_at_floor(idx, floor);

idx.batches

.range(..=floor)

.flat_map(|(_, batch)| &batch.entries)

.filter(|e| e.valid_from <= floor && !keep.contains(&Arc::as_ptr(e)))

.count()

}

fn compute_retained(idx: &Index, floor: u64) -> Index {

let keep = keep_at_floor(idx, floor);

let mut out = Index {

covered_through: idx.covered_through,

floor_at_write: idx.floor_at_write,

..Default::default()

};

for (&captured_at, batch) in &idx.batches {

if captured_at > floor {

out.insert_batch(batch.clone());

continue;

}

let entries: Vec<Arc<LogEntry>> = batch

.entries

.iter()

.filter(|e| e.valid_from > floor || keep.contains(&Arc::as_ptr(e)))

.cloned()

.collect();

// Half-open interval: through_lsn == floor covers nothing at or

// above the floor

let ambiguities: Vec<Arc<Ambiguity>> = batch

.ambiguities

.iter()

.filter(|a| a.through_lsn > floor)

.cloned()

.collect();

if !entries.is_empty() || !ambiguities.is_empty() {

out.insert_batch(Arc::new(BatchRecord {

captured_at,

commit_lsn: batch.commit_lsn,

observations: batch.observations.clone(),

ambiguities,

entries,

}));

}

}

out

}

// ── encoding ────────────────────────────────────────────────────────

fn push_u8(out: &mut Vec<u8>, v: u8) {

out.push(v);

}

fn push_u16(out: &mut Vec<u8>, v: u16) {

out.extend_from_slice(&v.to_le_bytes());

}

fn push_i16(out: &mut Vec<u8>, v: i16) {

out.extend_from_slice(&v.to_le_bytes());

}

fn push_u32(out: &mut Vec<u8>, v: u32) {

out.extend_from_slice(&v.to_le_bytes());

}

fn push_i32(out: &mut Vec<u8>, v: i32) {

out.extend_from_slice(&v.to_le_bytes());

}

fn push_u64(out: &mut Vec<u8>, v: u64) {

out.extend_from_slice(&v.to_le_bytes());

}

fn push_str(out: &mut Vec<u8>, s: &str) {

push_u32(out, s.len() as u32);

out.extend_from_slice(s.as_bytes());

}

fn push_opt_missing(out: &mut Vec<u8>, m: Option<&MissingDefault>) {

match m {

None => push_u8(out, 0),

Some(MissingDefault::Text(s)) => {

push_u8(out, 1);

push_str(out, s);

}

Some(MissingDefault::Raw(b)) => {

push_u8(out, 2);

push_u32(out, b.len() as u32);

out.extend_from_slice(b);

}

}

}

fn encode_header(id: &DescLogIdentity) -> Vec<u8> {

let mut out = Vec::new();

out.extend_from_slice(MAGIC);

push_u16(&mut out, VERSION);

push_u32(&mut out, id.pg_major);

push_str(&mut out, &id.system_id);

push_u32(&mut out, id.timeline);

push_u32(&mut out, id.db_oid);

push_u32(&mut out, id.wal_seg_size);

out

}

fn encode_meta(covered_through: u64, floor_at_write: u64) -> Vec<u8> {