use std::fs::File;
use std::path::Path;
use std::sync::atomic::{AtomicU64, Ordering};
use std::time::Duration;
use tracing::{debug, info, warn};
use crate::util::UnwrapPoison;
pub(crate) const TSHM_MAGIC: [u8; 8] = *b"TSHMWAL\0";
const TSHM_VERSION: u32 = 1;
pub(crate) const TSHM_FRAME_INDEX_LEN_OFFSET: usize = 40;
pub(crate) const TSHM_SNAPSHOT_SEQ_OFFSET: usize = 48;
pub(crate) const TSHM_MAX_FRAME_OFFSET: usize = 56;
pub(crate) const TSHM_NBACKFILLS_OFFSET: usize = 64;
pub(crate) const TSHM_TRANSACTION_COUNT_OFFSET: usize = 72;
pub(crate) const TSHM_CHECKPOINT_SEQ_OFFSET: usize = 88;
pub(crate) const TSHM_PAGE_SIZE_OFFSET: usize = 96;
pub(crate) const TSHM_SALT1_OFFSET: usize = 100;
pub(crate) const TSHM_SALT2_OFFSET: usize = 104;
pub(crate) const TSHM_CHECKSUM1_OFFSET: usize = 108;
pub(crate) const TSHM_CHECKSUM2_OFFSET: usize = 112;
pub(crate) const TSHM_HEADER_READ_LEN: usize = 116;
const WAL_HEADER_SIZE: u64 = 32;
const WAL_FRAME_HEADER_SIZE: u64 = 24;
const WAL_MAGIC_LE: u32 = 0x377f_0682;
const WAL_MAGIC_BE: u32 = 0x377f_0683;
const WAL_GUARD_INTERVAL_SECS: u64 = 60;
const REANNOUNCE_EVERY_CHECKS: u64 = 10;
static TSHM_OPEN_CLOSE_COUNT: AtomicU64 = AtomicU64::new(0);
#[must_use]
pub(crate) fn tshm_open_close_count() -> u64 {
TSHM_OPEN_CLOSE_COUNT.load(Ordering::Relaxed)
}
pub(crate) fn reset_tshm_open_close_count() {
TSHM_OPEN_CLOSE_COUNT.store(0, Ordering::Relaxed);
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct TshmHeader {
pub max_frame: u64,
pub nbackfills: u64,
pub transaction_count: u64,
pub frame_index_len: u32,
pub snapshot_seq: u64,
pub checkpoint_seq: u32,
pub page_size: u32,
pub salt_1: u32,
pub salt_2: u32,
pub checksum_1: u32,
pub checksum_2: u32,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct WalHeaderBytes {
pub page_size: u32,
pub checkpoint_seq: u32,
pub salt_1: u32,
pub salt_2: u32,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct LastFrameHeader {
pub commit_frame_size: u32,
pub salt_1: u32,
pub salt_2: u32,
pub checksum_1: u32,
pub checksum_2: u32,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum WalStateClass {
Healthy,
Orphaned,
OrphanedForeign,
Foreign,
TruncatedMidGen,
Oversized,
TornPre,
Unreadable,
LastFrameMismatch,
}
impl WalStateClass {
#[must_use]
pub(crate) fn blocks_checkpoint(self) -> bool {
matches!(
self,
Self::Orphaned
| Self::OrphanedForeign
| Self::Foreign
| Self::TruncatedMidGen
| Self::LastFrameMismatch
)
}
#[must_use]
pub(crate) fn label(self) -> &'static str {
match self {
Self::Healthy => "healthy",
Self::Orphaned => "orphaned-wal",
Self::OrphanedForeign => "orphaned-foreign-wal",
Self::Foreign => "foreign-wal",
Self::TruncatedMidGen => "truncated-wal",
Self::Oversized => "oversized-wal",
Self::TornPre => "torn-pre-index",
Self::Unreadable => "unreadable-tshm",
Self::LastFrameMismatch => "last-frame-mismatch",
}
}
}
#[derive(Debug, Clone, Copy, Default)]
pub struct StoreFds<'a> {
pub tshm: Option<&'a File>,
pub wal: Option<&'a File>,
}
impl StoreFds<'_> {
#[must_use]
pub(crate) const fn none() -> Self {
Self {
tshm: None,
wal: None,
}
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct StoreArtifactStatus {
pub store: String,
pub tshm: Option<TshmHeader>,
pub wal_size: u64,
pub wal_header: Option<WalHeaderBytes>,
pub class: WalStateClass,
}
#[must_use]
pub(crate) fn parse_tshm_header(bytes: &[u8]) -> Option<TshmHeader> {
if bytes.len() < TSHM_HEADER_READ_LEN {
return None;
}
if bytes[0..8] != TSHM_MAGIC {
return None;
}
let version = u32::from_le_bytes(bytes[8..12].try_into().ok()?);
if version != TSHM_VERSION {
return None;
}
let reader_slots = u32::from_le_bytes(bytes[12..16].try_into().ok()?);
if reader_slots < 64 || reader_slots % 64 != 0 {
return None;
}
Some(TshmHeader {
frame_index_len: u32::from_le_bytes(
bytes[TSHM_FRAME_INDEX_LEN_OFFSET..TSHM_FRAME_INDEX_LEN_OFFSET + 4]
.try_into()
.ok()?,
),
snapshot_seq: u64::from_le_bytes(
bytes[TSHM_SNAPSHOT_SEQ_OFFSET..TSHM_SNAPSHOT_SEQ_OFFSET + 8]
.try_into()
.ok()?,
),
max_frame: u64::from_le_bytes(
bytes[TSHM_MAX_FRAME_OFFSET..TSHM_MAX_FRAME_OFFSET + 8]
.try_into()
.ok()?,
),
nbackfills: u64::from_le_bytes(
bytes[TSHM_NBACKFILLS_OFFSET..TSHM_NBACKFILLS_OFFSET + 8]
.try_into()
.ok()?,
),
transaction_count: u64::from_le_bytes(
bytes[TSHM_TRANSACTION_COUNT_OFFSET..TSHM_TRANSACTION_COUNT_OFFSET + 8]
.try_into()
.ok()?,
),
checkpoint_seq: u32::from_le_bytes(
bytes[TSHM_CHECKPOINT_SEQ_OFFSET..TSHM_CHECKPOINT_SEQ_OFFSET + 4]
.try_into()
.ok()?,
),
page_size: u32::from_le_bytes(
bytes[TSHM_PAGE_SIZE_OFFSET..TSHM_PAGE_SIZE_OFFSET + 4]
.try_into()
.ok()?,
),
salt_1: u32::from_le_bytes(
bytes[TSHM_SALT1_OFFSET..TSHM_SALT1_OFFSET + 4]
.try_into()
.ok()?,
),
salt_2: u32::from_le_bytes(
bytes[TSHM_SALT2_OFFSET..TSHM_SALT2_OFFSET + 4]
.try_into()
.ok()?,
),
checksum_1: u32::from_le_bytes(
bytes[TSHM_CHECKSUM1_OFFSET..TSHM_CHECKSUM1_OFFSET + 4]
.try_into()
.ok()?,
),
checksum_2: u32::from_le_bytes(
bytes[TSHM_CHECKSUM2_OFFSET..TSHM_CHECKSUM2_OFFSET + 4]
.try_into()
.ok()?,
),
})
}
#[must_use]
pub(crate) fn parse_wal_header(bytes: &[u8; 32]) -> Option<WalHeaderBytes> {
let magic = u32::from_be_bytes(bytes[0..4].try_into().ok()?);
if !matches!(magic, WAL_MAGIC_LE | WAL_MAGIC_BE) {
return None;
}
let page_size = u32::from_be_bytes(bytes[8..12].try_into().ok()?);
if !(512..=65_536).contains(&page_size) || !page_size.is_power_of_two() {
return None;
}
let native_endian = cfg!(target_endian = "big") == (magic & 1 != 0);
let (calc1, calc2) = checksum_wal_prefix(&bytes[..24], native_endian);
let stored1 = u32::from_be_bytes(bytes[24..28].try_into().ok()?);
let stored2 = u32::from_be_bytes(bytes[28..32].try_into().ok()?);
if calc1 != stored1 || calc2 != stored2 {
return None;
}
Some(WalHeaderBytes {
page_size,
checkpoint_seq: u32::from_be_bytes(bytes[12..16].try_into().ok()?),
salt_1: u32::from_be_bytes(bytes[16..20].try_into().ok()?),
salt_2: u32::from_be_bytes(bytes[20..24].try_into().ok()?),
})
}
fn checksum_wal_prefix(buf: &[u8], native_endian: bool) -> (u32, u32) {
debug_assert_eq!(buf.len() % 8, 0);
let mut s0 = 0u32;
let mut s1 = 0u32;
let mut i = 0;
while i < buf.len() {
let v0 = u32::from_ne_bytes(buf[i..i + 4].try_into().expect("8-byte aligned"));
let v1 = u32::from_ne_bytes(buf[i + 4..i + 8].try_into().expect("8-byte aligned"));
let (v0, v1) = if native_endian {
(v0, v1)
} else {
(v0.swap_bytes(), v1.swap_bytes())
};
s0 = s0.wrapping_add(v0.wrapping_add(s1));
s1 = s1.wrapping_add(v1.wrapping_add(s0));
i += 8;
}
(s0, s1)
}
#[must_use]
pub(crate) fn classify_wal_state(
tshm: Option<TshmHeader>,
wal_size: u64,
wal_header: Option<WalHeaderBytes>,
last_frame: Option<LastFrameHeader>,
) -> WalStateClass {
let Some(h) = tshm else {
return WalStateClass::Unreadable;
};
if h.max_frame == 0 {
return WalStateClass::Healthy;
}
if wal_size < WAL_HEADER_SIZE {
return WalStateClass::Orphaned;
}
let Some(wal) = wal_header else {
return WalStateClass::OrphanedForeign;
};
if wal.page_size != h.page_size
|| wal.checkpoint_seq != h.checkpoint_seq
|| wal.salt_1 != h.salt_1
|| wal.salt_2 != h.salt_2
{
return WalStateClass::Foreign;
}
let frame_size = WAL_FRAME_HEADER_SIZE + u64::from(wal.page_size);
let expected = WAL_HEADER_SIZE.saturating_add(h.max_frame.saturating_mul(frame_size));
if wal_size < expected {
return WalStateClass::TruncatedMidGen;
}
if wal_size > expected {
return WalStateClass::Oversized;
}
if u64::from(h.frame_index_len) > h.max_frame {
return WalStateClass::TornPre;
}
if let Some(frame) = last_frame
&& (frame.commit_frame_size == 0
|| frame.salt_1 != h.salt_1
|| frame.salt_2 != h.salt_2
|| frame.checksum_1 != h.checksum_1
|| frame.checksum_2 != h.checksum_2)
{
return WalStateClass::LastFrameMismatch;
}
WalStateClass::Healthy
}
fn read_tshm_bytes_path(tshm_path: &Path) -> Option<[u8; TSHM_HEADER_READ_LEN]> {
use std::io::Read;
let mut file = std::fs::File::open(tshm_path).ok()?;
TSHM_OPEN_CLOSE_COUNT.fetch_add(1, Ordering::Relaxed);
let mut bytes = [0u8; TSHM_HEADER_READ_LEN];
file.read_exact(&mut bytes).ok()?;
Some(bytes)
}
fn read_wal_bytes_path(wal_path: &Path) -> Option<[u8; 32]> {
use std::io::Read;
let mut file = std::fs::File::open(wal_path).ok()?;
TSHM_OPEN_CLOSE_COUNT.fetch_add(1, Ordering::Relaxed);
let mut bytes = [0u8; 32];
file.read_exact(&mut bytes).ok()?;
Some(bytes)
}
fn read_last_frame_header(
wal_fd: Option<&File>,
wal_path: &Path,
tshm: &TshmHeader,
wal: &WalHeaderBytes,
) -> Option<LastFrameHeader> {
let frame_size = WAL_FRAME_HEADER_SIZE + u64::from(wal.page_size);
let offset =
WAL_HEADER_SIZE.saturating_add(tshm.max_frame.saturating_sub(1).saturating_mul(frame_size));
let bytes: [u8; 24] = if let Some(f) = wal_fd {
crate::turso::pread_at(f, offset)?
} else {
use std::io::{Read, Seek, SeekFrom};
let mut file = std::fs::File::open(wal_path).ok()?;
TSHM_OPEN_CLOSE_COUNT.fetch_add(1, Ordering::Relaxed);
file.seek(SeekFrom::Start(offset)).ok()?;
let mut buf = [0u8; 24];
file.read_exact(&mut buf).ok()?;
buf
};
Some(LastFrameHeader {
commit_frame_size: u32::from_be_bytes(bytes[4..8].try_into().ok()?),
salt_1: u32::from_be_bytes(bytes[8..12].try_into().ok()?),
salt_2: u32::from_be_bytes(bytes[12..16].try_into().ok()?),
checksum_1: u32::from_be_bytes(bytes[16..20].try_into().ok()?),
checksum_2: u32::from_be_bytes(bytes[20..24].try_into().ok()?),
})
}
fn read_tshm_disciplined(
read: impl Fn() -> Option<[u8; TSHM_HEADER_READ_LEN]>,
) -> Option<TshmHeader> {
for _ in 0..4 {
let first = read()?;
let h1 = parse_tshm_header(&first)?;
if h1.snapshot_seq % 2 != 0 {
continue;
}
let second = read()?;
let h2 = parse_tshm_header(&second)?;
if h2.snapshot_seq == h1.snapshot_seq {
return Some(h2);
}
}
None
}
#[must_use]
pub fn inspect_store_at(db_path: &Path, fds: StoreFds<'_>) -> StoreArtifactStatus {
let sidecars = crate::turso::store_sidecars(db_path);
let tshm = read_tshm_disciplined(|| match fds.tshm {
Some(f) => crate::turso::pread::<TSHM_HEADER_READ_LEN>(f),
None => read_tshm_bytes_path(&sidecars.tshm),
});
let wal_size = std::fs::metadata(&sidecars.wal).map_or(0, |m| m.len());
let wal_header = match fds.wal {
Some(f) => crate::turso::pread::<32>(f),
None => read_wal_bytes_path(&sidecars.wal),
};
let wal_header = wal_header.as_ref().and_then(parse_wal_header);
let mut class = classify_wal_state(tshm, wal_size, wal_header, None);
if class == WalStateClass::Healthy
&& let (Some(h), Some(w)) = (tshm.filter(|h| h.max_frame > 0), wal_header)
{
let last_frame = read_last_frame_header(fds.wal, &sidecars.wal, &h, &w);
class = match last_frame {
Some(frame) => classify_wal_state(tshm, wal_size, wal_header, Some(frame)),
None => WalStateClass::TruncatedMidGen,
};
}
let store = db_path.file_stem().map_or_else(
|| db_path.display().to_string(),
|s| s.to_string_lossy().into_owned(),
);
StoreArtifactStatus {
store,
tshm,
wal_size,
wal_header,
class,
}
}
#[must_use]
pub fn inspect_store(root: &Path, name: &str, fds: StoreFds<'_>) -> StoreArtifactStatus {
inspect_store_at(&crate::turso::store_db_path(root, name), fds)
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum BootDiagnosis {
Healthy,
StaleTail,
DurableB,
BlockedCoordination,
Structural,
}
impl BootDiagnosis {
#[must_use]
pub(crate) fn label(self) -> &'static str {
match self {
Self::Healthy => "healthy",
Self::StaleTail => "stale-tail",
Self::DurableB => "durable-b",
Self::BlockedCoordination => "blocked-coordination",
Self::Structural => "structural",
}
}
}
static BOOT_DIAGNOSES: std::sync::OnceLock<
std::sync::Mutex<std::collections::HashMap<std::path::PathBuf, BootDiagnosis>>,
> = std::sync::OnceLock::new();
fn boot_diagnoses()
-> &'static std::sync::Mutex<std::collections::HashMap<std::path::PathBuf, BootDiagnosis>> {
BOOT_DIAGNOSES.get_or_init(|| std::sync::Mutex::new(std::collections::HashMap::new()))
}
pub(crate) fn set_boot_diagnosis(db_path: &Path, diagnosis: BootDiagnosis) {
boot_diagnoses()
.lock()
.unwrap_poison()
.insert(db_path.to_path_buf(), diagnosis);
}
#[must_use]
pub(crate) fn take_boot_diagnosis(db_path: &Path) -> Option<BootDiagnosis> {
boot_diagnoses().lock().unwrap_poison().remove(db_path)
}
#[must_use]
pub(crate) fn has_boot_diagnosis(db_path: &Path) -> bool {
boot_diagnoses()
.lock()
.unwrap_poison()
.contains_key(db_path)
}
const DB_HEADER_MAGIC: &[u8; 16] = b"SQLite format 3\0";
pub(crate) const DB_HEADER_MIN_SIZE: u64 = 100;
pub(crate) fn read_db_header(db_path: &Path) -> Option<[u8; 18]> {
use std::io::Read;
let mut header = [0u8; 18];
let mut file = std::fs::File::open(db_path).ok()?;
file.read_exact(&mut header).ok()?;
Some(header)
}
#[must_use]
pub(crate) fn db_header_valid(header: &[u8; 18]) -> bool {
if &header[..16] != DB_HEADER_MAGIC {
return false;
}
let raw = u16::from_be_bytes([header[16], header[17]]);
let page_size = if raw == 1 { 65_536 } else { u32::from(raw) };
(512..=65_536).contains(&page_size) && page_size.is_power_of_two()
}
fn classify_main_db(db_path: &Path, wal_exists: bool, wal_size: u64) -> BootDiagnosis {
let Ok(meta) = std::fs::metadata(db_path) else {
return BootDiagnosis::Healthy; };
let size = meta.len();
if size == 0 {
return if wal_exists && wal_size > 0 {
BootDiagnosis::DurableB
} else {
BootDiagnosis::Healthy
};
}
if size < DB_HEADER_MIN_SIZE {
return BootDiagnosis::Structural;
}
let Some(header) = read_db_header(db_path) else {
return BootDiagnosis::Healthy;
};
if db_header_valid(&header) {
BootDiagnosis::Healthy
} else {
BootDiagnosis::Structural
}
}
pub fn diagnose_all_stores(root: &Path) {
for (name, _) in crate::turso::iter_checkpoint_stores() {
let db_path = crate::turso::store_db_path(root, name);
let sidecars = crate::turso::store_sidecars(&db_path);
let status = inspect_store_at(&db_path, StoreFds::none());
let wal_size = status.wal_size;
let diagnosis = match status.class {
WalStateClass::Healthy => {
classify_main_db(&db_path, sidecars.wal.exists(), wal_size)
}
WalStateClass::TornPre => {
match classify_main_db(&db_path, sidecars.wal.exists(), wal_size) {
BootDiagnosis::DurableB => BootDiagnosis::DurableB,
BootDiagnosis::Structural => BootDiagnosis::Structural,
_ => BootDiagnosis::StaleTail,
}
}
WalStateClass::Unreadable => {
if sidecars.tshm.exists() {
BootDiagnosis::Structural
} else {
classify_main_db(&db_path, sidecars.wal.exists(), wal_size)
}
}
WalStateClass::Oversized => {
BootDiagnosis::BlockedCoordination
}
WalStateClass::LastFrameMismatch => {
BootDiagnosis::BlockedCoordination
}
_ => BootDiagnosis::BlockedCoordination,
};
set_boot_diagnosis(&db_path, diagnosis);
if diagnosis != BootDiagnosis::Healthy {
crate::boot::boot_diagnostic(format!(
"boot pre-flight: store '{name}' class {} (max_frame={}, wal_size={}) — \
healing will run before open",
diagnosis.label(),
status.tshm.map_or(0, |h| h.max_frame),
wal_size,
));
}
}
reset_tshm_open_close_count();
}
pub async fn run_wal_guard_loop() {
let root = match crate::config::default_config_dir() {
Ok(root) => root,
Err(e) => {
warn!(error = %e, "wal-guard: cannot resolve storage root; guard disabled");
return;
}
};
let mut seen: std::collections::HashMap<String, (WalStateClass, u64)> =
std::collections::HashMap::new();
let mut identity_seen: std::collections::HashMap<String, (crate::turso::SidecarIdentity, u64)> =
std::collections::HashMap::new();
let mut check_count: u64 = 0;
loop {
if !crate::shutdown::sleep_or_shutdown_or_drain(Duration::from_secs(
WAL_GUARD_INTERVAL_SECS,
))
.await
{
break;
}
check_count += 1;
for (name, conn) in crate::turso::iter_checkpoint_stores() {
let identity = conn.and_then(crate::turso::Connection::check_coordination_identity);
if identity == Some(crate::turso::SidecarIdentity::Replaced) {
announce_identity(
&mut identity_seen,
name,
crate::turso::SidecarIdentity::Replaced,
check_count,
);
continue;
}
let fds = conn.map_or_else(StoreFds::none, crate::turso::Connection::store_fds);
let status = inspect_store(&root, name, fds);
let store = status.store.clone();
if identity == Some(crate::turso::SidecarIdentity::Deleted)
&& status.class == WalStateClass::Healthy
{
announce_identity(
&mut identity_seen,
name,
crate::turso::SidecarIdentity::Deleted,
check_count,
);
} else {
identity_seen.remove(name);
}
if status.class == WalStateClass::Healthy {
seen.remove(&store);
continue;
}
let count = seen
.get(&store)
.filter(|(c, _)| *c == status.class)
.map_or(1, |(_, n)| n + 1);
seen.insert(store, (status.class, count));
let announce =
count == 2 || (count > 2 && check_count.is_multiple_of(REANNOUNCE_EVERY_CHECKS));
if announce {
warn!(
store = %status.store,
class = status.class.label(),
max_frame = status.tshm.map_or(0, |h| h.max_frame),
wal_size = status.wal_size,
"wal-guard: non-healthy coordination state ({}) — {}",
status.class.label(),
class_warning(status.class),
);
}
}
if check_count.is_multiple_of(REANNOUNCE_EVERY_CHECKS) {
let open_close_count = tshm_open_close_count();
if open_close_count == 0 {
debug!(
tshm_open_close_count = open_close_count,
"wal-guard: daemon-side coordination open+close count (0 expected post-boot — \
the daemon-side open+close regression signal)",
);
} else {
info!(
tshm_open_close_count = open_close_count,
"wal-guard: daemon-side coordination open+close count (0 expected post-boot — \
the daemon-side open+close regression signal)",
);
}
}
}
}
fn announce_identity(
identity_seen: &mut std::collections::HashMap<String, (crate::turso::SidecarIdentity, u64)>,
name: &str,
identity: crate::turso::SidecarIdentity,
check_count: u64,
) {
let entry = identity_seen
.entry(name.to_string())
.or_insert((identity, 0));
if entry.0 != identity {
*entry = (identity, 0);
}
entry.1 += 1;
if entry.1 == 2 || (entry.1 > 2 && check_count.is_multiple_of(REANNOUNCE_EVERY_CHECKS)) {
match identity {
crate::turso::SidecarIdentity::Replaced => warn!(
db = %name,
"wal-guard: coordination files replaced by an external process \
(inode mismatch) — checks suspended for this store until restart",
),
crate::turso::SidecarIdentity::Deleted => warn!(
db = %name,
"wal-guard: coordination sidecar deleted by an external process \
(unlinked under the daemon) — the predicate will detect any \
resulting orphaned-WAL state",
),
_ => {}
}
}
}
fn class_warning(class: WalStateClass) -> &'static str {
match class {
WalStateClass::TornPre => {
"frame index exceeds max_frame (crash stale-tail or in-flight write) — \
healed by a TRUNCATE-first checkpoint at the next store open"
}
WalStateClass::Oversized => "WAL is larger than the max_frame-implied size (warn-only)",
WalStateClass::Unreadable => {
".tshm is unreadable — coordination state cannot be classified"
}
WalStateClass::LastFrameMismatch => {
"regrown WAL with a copied header (last frame fails commit/salt/checksum) — \
checkpoints blocked"
}
_ => {
"foreign standard-SQLite activity likely replaced -wal/-tshm under the daemon. \
Reads through .tshm may hit torn-frame errors. Query snapshot copies; never \
delete/recreate -wal/-shm/-tshm while the daemon runs."
}
}
}
#[cfg(test)]
mod tests {
use super::*;
fn write(path: &std::path::Path, bytes: &[u8]) {
std::fs::create_dir_all(path.parent().unwrap()).unwrap();
std::fs::write(path, bytes).unwrap();
}
fn tshm_bytes(max_frame: u64, nbackfills: u64, tx_count: u64) -> Vec<u8> {
let mut bytes = vec![0u8; TSHM_HEADER_READ_LEN];
bytes[0..8].copy_from_slice(&TSHM_MAGIC);
bytes[8..12].copy_from_slice(&1u32.to_le_bytes());
bytes[12..16].copy_from_slice(&64u32.to_le_bytes());
bytes[TSHM_FRAME_INDEX_LEN_OFFSET..TSHM_FRAME_INDEX_LEN_OFFSET + 4]
.copy_from_slice(&max_frame.to_le_bytes()[..4]);
bytes[TSHM_MAX_FRAME_OFFSET..TSHM_MAX_FRAME_OFFSET + 8]
.copy_from_slice(&max_frame.to_le_bytes());
bytes[TSHM_NBACKFILLS_OFFSET..TSHM_NBACKFILLS_OFFSET + 8]
.copy_from_slice(&nbackfills.to_le_bytes());
bytes[TSHM_TRANSACTION_COUNT_OFFSET..TSHM_TRANSACTION_COUNT_OFFSET + 8]
.copy_from_slice(&tx_count.to_le_bytes());
bytes
}
fn wal_header_bytes(page_size: u32, salt_1: u32, salt_2: u32) -> [u8; 32] {
let mut bytes = [0u8; 32];
bytes[0..4].copy_from_slice(&WAL_MAGIC_LE.to_be_bytes());
bytes[8..12].copy_from_slice(&page_size.to_be_bytes());
bytes[16..20].copy_from_slice(&salt_1.to_be_bytes());
bytes[20..24].copy_from_slice(&salt_2.to_be_bytes());
let native_endian = cfg!(target_endian = "big") == (WAL_MAGIC_LE & 1 != 0);
let (c1, c2) = checksum_wal_prefix(&bytes[..24], native_endian);
bytes[24..28].copy_from_slice(&c1.to_be_bytes());
bytes[28..32].copy_from_slice(&c2.to_be_bytes());
bytes
}
#[test]
fn parses_valid_tshm_header() {
let bytes = tshm_bytes(42, 3, 99);
let hdr = parse_tshm_header(&bytes).expect("valid tshm header parses");
assert_eq!(hdr.max_frame, 42);
assert_eq!(hdr.nbackfills, 3);
assert_eq!(hdr.transaction_count, 99);
assert_eq!(hdr.frame_index_len, 42); assert_eq!(hdr.page_size, 0); assert_eq!(hdr.salt_1, 0);
assert_eq!(hdr.salt_2, 0);
}
#[test]
fn rejects_missing_bad_or_short_tshm() {
let mut bytes = tshm_bytes(1, 0, 0);
bytes[0] = b'X';
assert!(parse_tshm_header(&bytes).is_none());
assert!(parse_tshm_header(&bytes[..16]).is_none());
}
#[test]
fn wal_header_validation_rejects_bad_magic_and_checksum() {
let good = wal_header_bytes(4096, 7, 11);
assert!(parse_wal_header(&good).is_some());
let mut bad_magic = good;
bad_magic[0] = 0;
assert!(parse_wal_header(&bad_magic).is_none());
let mut bad_checksum = good;
bad_checksum[24] ^= 0xff;
assert!(parse_wal_header(&bad_checksum).is_none());
let mut bad_size = good;
bad_size[8..12].copy_from_slice(&123u32.to_be_bytes());
assert!(parse_wal_header(&bad_size).is_none());
}
#[test]
fn db_header_valid_decodes_64k_page_size() {
let mut header = [0u8; 18];
header[..16].copy_from_slice(b"SQLite format 3\0");
header[16..18].copy_from_slice(&1u16.to_be_bytes());
assert!(db_header_valid(&header));
header[0] = b'X';
assert!(!db_header_valid(&header));
}
fn tshm(max_frame: u64, frame_index_len: u32) -> TshmHeader {
TshmHeader {
max_frame,
nbackfills: 0,
transaction_count: 0,
frame_index_len,
snapshot_seq: 0,
checkpoint_seq: 0,
page_size: 4096,
salt_1: 7,
salt_2: 11,
checksum_1: 0,
checksum_2: 0,
}
}
fn last_frame(commit: u32, salt_1: u32, salt_2: u32) -> LastFrameHeader {
LastFrameHeader {
commit_frame_size: commit,
salt_1,
salt_2,
checksum_1: 0,
checksum_2: 0,
}
}
#[test]
fn classify_matches_tshm_wal_states() {
let healthy = tshm(120, 120);
let wal = parse_wal_header(&wal_header_bytes(4096, 7, 11)).unwrap();
let frame_size = 4096 + 24;
assert_eq!(
classify_wal_state(Some(healthy), 32 + 120 * frame_size, Some(wal), None),
WalStateClass::Healthy
);
assert_eq!(
classify_wal_state(
Some(healthy),
32 + 120 * frame_size,
Some(wal),
Some(last_frame(120, 7, 11))
),
WalStateClass::Healthy
);
assert_eq!(
classify_wal_state(
Some(healthy),
32 + 120 * frame_size,
Some(wal),
Some(last_frame(0, 7, 11))
),
WalStateClass::LastFrameMismatch
);
assert_eq!(
classify_wal_state(
Some(healthy),
32 + 120 * frame_size,
Some(wal),
Some(last_frame(120, 99, 11))
),
WalStateClass::LastFrameMismatch
);
assert_eq!(
classify_wal_state(Some(tshm(0, 0)), 0, None, None),
WalStateClass::Healthy
);
assert_eq!(
classify_wal_state(Some(tshm(0, 0)), 4096, Some(wal), None),
WalStateClass::Healthy
);
assert_eq!(
classify_wal_state(Some(tshm(120, 130)), 32 + 120 * frame_size, Some(wal), None),
WalStateClass::TornPre
);
assert_eq!(
classify_wal_state(Some(healthy), 0, None, None),
WalStateClass::Orphaned
);
assert_eq!(
classify_wal_state(Some(healthy), 16, None, None),
WalStateClass::Orphaned
);
assert_eq!(
classify_wal_state(Some(healthy), 4096, None, None),
WalStateClass::OrphanedForeign
);
let foreign = parse_wal_header(&wal_header_bytes(4096, 99, 11)).unwrap();
assert_eq!(
classify_wal_state(Some(healthy), 32 + 120 * frame_size, Some(foreign), None),
WalStateClass::Foreign
);
assert_eq!(
classify_wal_state(Some(healthy), 32 + 50 * frame_size, Some(wal), None),
WalStateClass::TruncatedMidGen
);
assert_eq!(
classify_wal_state(Some(healthy), 32 + 121 * frame_size, Some(wal), None),
WalStateClass::Oversized
);
assert_eq!(
classify_wal_state(None, 0, None, None),
WalStateClass::Unreadable
);
}
#[test]
fn blocking_subset_is_only_2_3_4_6_plus_last_frame() {
assert!(WalStateClass::Orphaned.blocks_checkpoint());
assert!(WalStateClass::OrphanedForeign.blocks_checkpoint());
assert!(WalStateClass::Foreign.blocks_checkpoint());
assert!(WalStateClass::TruncatedMidGen.blocks_checkpoint());
assert!(WalStateClass::LastFrameMismatch.blocks_checkpoint());
assert!(!WalStateClass::Healthy.blocks_checkpoint());
assert!(!WalStateClass::Oversized.blocks_checkpoint());
assert!(!WalStateClass::TornPre.blocks_checkpoint());
assert!(!WalStateClass::Unreadable.blocks_checkpoint());
}
#[test]
#[serial_test::serial(tshm_counter)]
fn inspect_store_classifies_synthetic_file_sets() {
let dir = std::env::temp_dir().join(format!("wal_guard_state_{}", std::process::id()));
let _ = std::fs::remove_dir_all(&dir);
write(&dir.join("db/board.db-tshm"), &tshm_bytes(0, 0, 5));
write(&dir.join("db/board.db-wal"), &[]);
let s = inspect_store(&dir, "board", StoreFds::none());
assert_eq!(s.class, WalStateClass::Healthy);
write(
&dir.join("db/sessions.db-tshm"),
&tshm_bytes(356, 0, 710_565),
);
write(&dir.join("db/sessions.db-wal"), &[]);
let s = inspect_store(&dir, "sessions", StoreFds::none());
assert_eq!(s.class, WalStateClass::Orphaned);
assert!(s.tshm.is_some());
assert_eq!(s.tshm.unwrap().max_frame, 356);
write(&dir.join("db/users.db-wal"), &[0u8; 4096]);
let s = inspect_store(&dir, "users", StoreFds::none());
assert_eq!(s.class, WalStateClass::Unreadable);
let _ = std::fs::remove_dir_all(&dir);
}
#[test]
#[serial_test::serial(tshm_counter)]
fn inspect_store_visits_every_store() {
let dir = std::env::temp_dir().join(format!("wal_guard_all_{}", std::process::id()));
let _ = std::fs::remove_dir_all(&dir);
for name in crate::turso::store_names() {
write(
&dir.join(format!("db/{name}.db-tshm")),
&tshm_bytes(0, 0, 0),
);
}
for name in crate::turso::store_names() {
let s = inspect_store(&dir, name, StoreFds::none());
assert_eq!(
s.class,
WalStateClass::Healthy,
"fixture store {name} must be healthy"
);
}
let _ = std::fs::remove_dir_all(&dir);
}
#[test]
#[serial_test::serial(tshm_counter)]
fn path_read_counts_open_close_and_fd_read_does_not() {
let dir = std::env::temp_dir().join(format!("wal_guard_cnt_{}", std::process::id()));
let _ = std::fs::remove_dir_all(&dir);
let tshm_path = dir.join("db/board.db-tshm");
write(&tshm_path, &tshm_bytes(0, 0, 0));
reset_tshm_open_close_count();
let file = File::open(&tshm_path).unwrap();
let before = tshm_open_close_count();
let s = inspect_store_at(
&dir.join("db/board.db"),
StoreFds {
tshm: Some(&file),
wal: None,
},
);
assert_eq!(s.class, WalStateClass::Healthy);
assert_eq!(
tshm_open_close_count(),
before,
"fd read must not open+close"
);
let s = inspect_store_at(&dir.join("db/board.db"), StoreFds::none());
assert_eq!(s.class, WalStateClass::Healthy);
assert_eq!(
tshm_open_close_count(),
before + 2,
"path read must count an open+close pair per disciplined read (two reads)"
);
drop(file);
let _ = std::fs::remove_dir_all(&dir);
}
#[test]
#[serial_test::serial(tshm_counter)]
fn diagnose_all_stores_classifies_stale_tail_and_structural() {
let dir = std::env::temp_dir().join(format!("wal_guard_preflight_{}", std::process::id()));
let _ = std::fs::remove_dir_all(&dir);
let mut stale = tshm_bytes(120, 0, 7);
stale[TSHM_FRAME_INDEX_LEN_OFFSET..TSHM_FRAME_INDEX_LEN_OFFSET + 4]
.copy_from_slice(&130u32.to_le_bytes());
stale[TSHM_PAGE_SIZE_OFFSET..TSHM_PAGE_SIZE_OFFSET + 4]
.copy_from_slice(&4096u32.to_le_bytes());
write(&dir.join("db/sessions.db-tshm"), &stale);
let mut wal = vec![0u8; 32 + 120 * (24 + 4096)];
wal[..32].copy_from_slice(&wal_header_bytes(4096, 0, 0));
write(&dir.join("db/sessions.db-wal"), &wal);
write(&dir.join("db/board.db"), &[0u8; 64]);
write(&dir.join("db/board.db-tshm"), &tshm_bytes(0, 0, 0));
let mut db = vec![0u8; 4096];
db[..16].copy_from_slice(b"SQLite format 3\0");
db[16..18].copy_from_slice(&1u16.to_be_bytes());
write(&dir.join("db/chat_history.db"), &db);
write(&dir.join("db/chat_history.db-tshm"), &tshm_bytes(0, 0, 0));
write(&dir.join("db/users.db-tshm"), &tshm_bytes(0, 0, 0));
crate::wal_guard::diagnose_all_stores(&dir);
assert_eq!(
crate::wal_guard::take_boot_diagnosis(&crate::turso::store_db_path(&dir, "sessions")),
Some(BootDiagnosis::StaleTail)
);
assert_eq!(
crate::wal_guard::take_boot_diagnosis(&crate::turso::store_db_path(&dir, "board")),
Some(BootDiagnosis::Structural)
);
assert_eq!(
crate::wal_guard::take_boot_diagnosis(&crate::turso::store_db_path(
&dir,
"chat_history"
)),
Some(BootDiagnosis::Healthy)
);
assert_eq!(
crate::wal_guard::take_boot_diagnosis(&crate::turso::store_db_path(&dir, "users")),
Some(BootDiagnosis::Healthy)
);
let _ = std::fs::remove_dir_all(&dir);
}
}