use std::collections::{HashMap, HashSet};
use std::fs::{self, File, OpenOptions};
use std::io::{Write, Seek, SeekFrom};
use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicBool, AtomicU64, AtomicUsize, Ordering};
use std::ops::Deref;
use std::sync::{Arc, Mutex, OnceLock, RwLock, Weak};
use std::thread::JoinHandle;
use g_math::fixed_point::FixedPoint;
use std::time::{Duration, Instant};
use horon_engine::{SemanticOutlier, Store, StoreConfig};
use crate::error::{HoronError, HoronResult};
use crate::format::*;
use crate::gacl::{Credentials, NodeAccessBands};
use crate::header::GeoHeader;
use crate::hilbert::HilbertMapper;
use crate::partial::SnapView;
use crate::quant::SemLayout;
use crate::snapshot::{self, NodeEntry};
use crate::wal::{self, WalEntry, WalPayload};
const _: () = {
fn _assert_send<T: Send>() {}
fn _assert_sync<T: Sync>() {}
fn _check() {
_assert_send::<Horon>();
_assert_sync::<Horon>();
}
};
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum DurabilityMode {
Fsync,
Batched,
Relaxed,
}
impl Default for DurabilityMode {
fn default() -> Self {
Self::Batched
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub enum HistoryRetention {
#[default]
Off,
Archive,
}
#[derive(Debug)]
pub enum WalTail {
Entries(Vec<WalEntry>),
SnapshotRequired {
base_seq: u32,
},
}
pub struct HoronConfig {
pub dimension: u8,
pub semantic_dims: u8,
pub compression: bool,
pub auto_compact_threshold: u32,
pub wal_batch_size: u32,
pub wal_flush_interval_ms: u64,
pub durability: DurabilityMode,
pub gacl: bool,
pub gacl_fail_closed: bool,
pub lazy_geometry: bool,
pub partial_reads: bool,
pub meaning_addressed: bool,
pub semantic_bounds: (f64, f64),
pub history_retention: HistoryRetention,
pub approximate_semantic: bool,
pub quantized_semantic: bool,
}
impl Default for HoronConfig {
fn default() -> Self {
Self {
dimension: 4,
semantic_dims: 16,
compression: true,
auto_compact_threshold: 10_000,
wal_batch_size: 0,
wal_flush_interval_ms: 0,
durability: DurabilityMode::Batched,
gacl: false,
gacl_fail_closed: false,
lazy_geometry: false,
partial_reads: false,
meaning_addressed: false,
semantic_bounds: (0.0, 1.0),
history_retention: HistoryRetention::Off,
approximate_semantic: false,
quantized_semantic: false,
}
}
}
struct PartialState {
view: RwLock<SnapView>,
tombstones: RwLock<HashSet<String>>,
scanned: AtomicUsize,
}
struct WalWriter {
file: File,
next_seq: u32,
entry_count: u32,
wal_data_offset: u64,
pending_serialized: Vec<Vec<u8>>,
pending_count: u32,
batch_size: u32,
flush_interval: Duration,
last_flush: Instant,
durability: DurabilityMode,
wal_compressed: bool,
compression_algo: u8,
layout: SemLayout,
pending_entries: Vec<WalEntry>,
subscribers: Vec<std::sync::mpsc::Sender<WalEntry>>,
}
impl WalWriter {
fn append(&mut self, entry: &WalEntry) -> HoronResult<()> {
if self.next_seq == u32::MAX {
return Err(HoronError::InvalidOperation(
"WAL sequence space exhausted (u32 seq at its maximum); \
rebuild the file to reset the sequence".to_string(),
));
}
let mut buf = Vec::new();
entry.write_to(&mut buf, &self.layout)?;
self.pending_serialized.push(buf);
if !self.subscribers.is_empty() {
self.pending_entries.push(entry.clone());
}
self.pending_count += 1;
self.next_seq += 1;
let should_flush = self.durability == DurabilityMode::Fsync
|| self.batch_size == 0
|| self.pending_count >= self.batch_size
|| (self.flush_interval.as_millis() > 0
&& self.last_flush.elapsed() >= self.flush_interval);
if should_flush {
self.flush_pending()?;
}
Ok(())
}
fn flush_pending(&mut self) -> HoronResult<()> {
if self.pending_serialized.is_empty() {
return Ok(());
}
if self.wal_compressed {
for chunk in self.pending_serialized.chunks(WAL_BLOCK_SIZE) {
let mut block_bytes = Vec::new();
for entry_bytes in chunk {
block_bytes.extend_from_slice(entry_bytes);
}
wal::write_wal_block(
&mut self.file,
&block_bytes,
chunk.len() as u16,
self.compression_algo,
)?;
}
} else {
for entry_bytes in &self.pending_serialized {
self.file.write_all(entry_bytes)?;
}
}
self.entry_count += self.pending_serialized.len() as u32;
let current_pos = self.file.stream_position()?;
self.file.seek(SeekFrom::Start(self.wal_data_offset - WAL_HEADER_SIZE as u64))?;
let base_seq = self.next_seq - self.entry_count;
wal::write_wal_header(&mut self.file, self.entry_count, base_seq)?;
self.file.seek(SeekFrom::Start(current_pos))?;
match self.durability {
DurabilityMode::Fsync | DurabilityMode::Batched => {
self.file.sync_data()?;
}
DurabilityMode::Relaxed => {}
}
self.pending_serialized.clear();
self.pending_count = 0;
self.last_flush = Instant::now();
if !self.subscribers.is_empty() {
for entry in self.pending_entries.drain(..) {
self.subscribers.retain(|tx| tx.send(entry.clone()).is_ok());
}
} else {
self.pending_entries.clear();
}
Ok(())
}
fn total_entries(&self) -> u32 {
self.entry_count + self.pending_serialized.len() as u32
}
}
pub struct Horon {
core: Arc<HoronCore>,
}
impl Deref for Horon {
type Target = HoronCore;
fn deref(&self) -> &HoronCore {
&self.core
}
}
impl Horon {
pub fn open<P: AsRef<Path>>(path: P) -> HoronResult<Self> {
Self::open_with_config(path, HoronConfig::default())
}
pub fn open_with_config<P: AsRef<Path>>(
path: P,
config: HoronConfig,
) -> HoronResult<Self> {
let core = Arc::new(HoronCore::open_core(path.as_ref(), config)?);
let _ = core.self_weak.set(Arc::downgrade(&core));
Ok(Self { core })
}
pub fn semantic_distance(
coords_a: &[u8],
coords_b: &[u8],
dim_range: std::ops::Range<usize>,
) -> FixedPoint {
Store::semantic_distance(coords_a, coords_b, dim_range)
}
}
#[doc(hidden)]
pub struct HoronCore {
store: Store,
path: PathBuf,
header: GeoHeader,
wal: Mutex<WalWriter>,
config: HoronConfig,
compacting: AtomicBool,
credentials: RwLock<Option<Credentials>>,
last_compaction_error: Mutex<Option<String>>,
self_weak: OnceLock<Weak<HoronCore>>,
partial: Option<PartialState>,
ma_bounds: Option<(f64, f64)>,
current_epoch: AtomicU64,
}
const MAX_HILBERT_DIMS: usize = 8;
const HILBERT_BITS: u32 = 12;
fn addressed_dims(semantic_dims: usize) -> usize {
semantic_dims
.saturating_sub(DIM_USER_DEFINED_START)
.min(MAX_HILBERT_DIMS)
}
fn decode_user_dims(sem: &[u8], semantic_dims: usize) -> Vec<FixedPoint> {
(0..addressed_dims(semantic_dims))
.map(|d| {
let start = (DIM_USER_DEFINED_START + d) * 16;
let end = start + 16;
if sem.len() >= end {
FixedPoint::from_raw(i128::from_le_bytes(sem[start..end].try_into().unwrap()))
} else {
FixedPoint::from_int(0)
}
})
.collect()
}
fn hilbert_from_values(vals: &[FixedPoint], bounds: (f64, f64)) -> u128 {
if vals.is_empty() {
return 0;
}
let lo = FixedPoint::from_f64(bounds.0);
let hi = FixedPoint::from_f64(bounds.1);
let range = hi - lo;
let zero = FixedPoint::from_int(0);
let one = FixedPoint::from_int(1);
let half = one / FixedPoint::from_int(2);
let epsilon = FixedPoint::from_f64(1e-12);
let norm: Vec<FixedPoint> = vals
.iter()
.map(|v| {
if range < epsilon {
half
} else {
let n = (*v - lo) / range;
if n < zero { zero } else if n > one { one } else { n }
}
})
.collect();
HilbertMapper::new(vals.len(), HILBERT_BITS)
.coords_to_index_fixed(&norm)
.value()
}
fn decode_dim_slice(coords: &[u8], r: &std::ops::Range<usize>) -> Vec<FixedPoint> {
r.clone()
.map(|dim| {
let start = dim * 16;
let end = start + 16;
if coords.len() >= end {
FixedPoint::from_raw(i128::from_le_bytes(coords[start..end].try_into().unwrap()))
} else {
FixedPoint::from_int(0)
}
})
.collect()
}
fn semantic_distance_sq(
qv: &[FixedPoint],
coords: &[u8],
r: &std::ops::Range<usize>,
) -> FixedPoint {
let cv = decode_dim_slice(coords, r);
g_math::fixed_point::imperative::fused::euclidean_distance_squared(qv, &cv)
}
fn global_hilbert(sem: &[u8], semantic_dims: usize, bounds: (f64, f64)) -> u128 {
hilbert_from_values(&decode_user_dims(sem, semantic_dims), bounds)
}
impl HoronCore {
fn open_core(path: &Path, config: HoronConfig) -> HoronResult<Self> {
if path.exists() {
Self::open_existing(path, &config)
} else {
Self::create_new(path, config)
}
}
fn validate_mode_combos(config: &HoronConfig) -> HoronResult<()> {
if config.dimension != 4 {
return Err(HoronError::Config(format!(
"dimension {} is not supported in this release (only 4)",
config.dimension
)));
}
if config.partial_reads && config.compression {
return Err(HoronError::Config(
"partial_reads requires compression: false (a zstd frame cannot be partially read)".into(),
));
}
if config.partial_reads && config.gacl {
return Err(HoronError::Config(
"partial_reads is not yet compatible with GACL enforcement".into(),
));
}
if config.quantized_semantic
&& config.semantic_dims as usize <= DIM_USER_DEFINED_START
{
return Err(HoronError::Config(format!(
"quantized_semantic requires semantic_dims > {} (user dims to quantize)",
DIM_USER_DEFINED_START
)));
}
if config.meaning_addressed {
if config.compression {
return Err(HoronError::Config(
"meaning_addressed requires compression: false".into(),
));
}
if config.semantic_dims as usize <= DIM_USER_DEFINED_START {
return Err(HoronError::Config(format!(
"meaning_addressed requires semantic_dims > {} (user dims to address by)",
DIM_USER_DEFINED_START
)));
}
let (lo, hi) = config.semantic_bounds;
if !(hi > lo) {
return Err(HoronError::Config(
"semantic_bounds must satisfy max > min".into(),
));
}
}
Ok(())
}
fn bounds_section_bytes(semantic_dims: usize, bounds: (f64, f64)) -> Vec<u8> {
let user_dims = semantic_dims.saturating_sub(DIM_USER_DEFINED_START);
let mut out = Vec::with_capacity(user_dims * 16);
for _ in 0..user_dims {
out.extend_from_slice(&bounds.0.to_le_bytes());
out.extend_from_slice(&bounds.1.to_le_bytes());
}
out
}
fn read_bounds_section(
file: &mut File,
semantic_dims: usize,
) -> HoronResult<(f64, f64)> {
use std::io::Read;
let user_dims = semantic_dims.saturating_sub(DIM_USER_DEFINED_START);
let mut buf = vec![0u8; user_dims * 16];
file.read_exact(&mut buf)?;
if user_dims == 0 {
return Err(HoronError::InvalidFormat(
"meaning-addressed file with no user semantic dims".into(),
));
}
let lo = f64::from_le_bytes(buf[0..8].try_into().unwrap());
let hi = f64::from_le_bytes(buf[8..16].try_into().unwrap());
if !(hi > lo) || !lo.is_finite() || !hi.is_finite() {
return Err(HoronError::InvalidFormat(
"corrupt bounds section (max <= min or non-finite)".into(),
));
}
Ok((lo, hi))
}
fn create_new(path: &Path, config: HoronConfig) -> HoronResult<Self> {
Self::validate_mode_combos(&config)?;
let tau_raw = g_math::fixed_point::FixedPoint::from_int(1).raw();
let mut header = GeoHeader::with_gacl(
config.dimension,
config.semantic_dims,
tau_raw,
config.compression,
config.gacl,
);
if config.meaning_addressed {
header.version = VERSION_MEANING_ADDRESSED;
header.flags |= FLAG_MEANING_ADDRESSED;
}
if config.quantized_semantic {
header.version = VERSION_QUANTIZED;
header.flags |= FLAG_QUANTIZED_SEMANTIC;
}
let layout = SemLayout::from_header(&header);
let mut file = OpenOptions::new()
.read(true)
.write(true)
.create(true)
.open(path)?;
try_lock_exclusive(&file, path)?;
file.set_len(0)?;
cleanup_orphan_tmp(path);
file.write_all(&header.to_bytes())?;
if config.meaning_addressed {
file.write_all(&Self::bounds_section_bytes(
config.semantic_dims as usize,
config.semantic_bounds,
))?;
}
snapshot::write_snapshot(&mut file, &[], false, true, &layout)?;
wal::write_wal_header(&mut file, 0, 1)?;
let wal_data_offset = file.stream_position()?;
file.sync_all()?;
fsync_dir(path)?;
let ma_bounds = config.meaning_addressed.then_some(config.semantic_bounds);
let partial = if config.partial_reads {
let mmap = unsafe { memmap2::Mmap::map(&file)? };
let raw_start = HEADER_SIZE
+ ma_bounds.map_or(0, |_| {
(config.semantic_dims as usize - DIM_USER_DEFINED_START) * 16
})
+ SNAP_HEADER_SIZE;
let view = SnapView::scan(mmap, raw_start, 0, 0, layout, None)?;
Some(PartialState {
view: RwLock::new(view),
tombstones: RwLock::new(HashSet::new()),
scanned: AtomicUsize::new(0),
})
} else {
None
};
let store = Store::with_config(
StoreConfig::new().tau(g_math::fixed_point::FixedPoint::from_raw(tau_raw)),
);
let wal = WalWriter {
file,
next_seq: 1,
entry_count: 0,
wal_data_offset,
pending_serialized: Vec::new(),
pending_entries: Vec::new(),
subscribers: Vec::new(),
pending_count: 0,
batch_size: config.wal_batch_size,
flush_interval: Duration::from_millis(config.wal_flush_interval_ms),
last_flush: Instant::now(),
durability: config.durability,
wal_compressed: config.compression,
compression_algo: if config.compression { ALGO_ZSTD } else { 0 },
layout,
};
Ok(Self {
store,
path: path.to_path_buf(),
header,
wal: Mutex::new(wal),
config,
compacting: AtomicBool::new(false),
credentials: RwLock::new(None),
last_compaction_error: Mutex::new(None),
self_weak: OnceLock::new(),
partial,
ma_bounds,
current_epoch: AtomicU64::new(0),
})
}
fn open_existing(path: &Path, config: &HoronConfig) -> HoronResult<Self> {
let mut file = OpenOptions::new()
.read(true)
.write(true)
.open(path)?;
try_lock_exclusive(&file, path)?;
cleanup_orphan_tmp(path);
let mut header_bytes = [0u8; HEADER_SIZE];
std::io::Read::read_exact(&mut file, &mut header_bytes)?;
let header = GeoHeader::from_bytes(&header_bytes)?;
let layout = SemLayout::from_header(&header);
let compressed = header.compression_enabled();
let snapshot_has_crc = header.version >= 2;
let meaning_addressed = header.flags & FLAG_MEANING_ADDRESSED != 0;
let ma_bounds = if meaning_addressed {
Some(Self::read_bounds_section(&mut file, header.semantic_dims as usize)?)
} else {
None
};
if config.partial_reads {
if compressed {
return Err(HoronError::Config(
"partial_reads requires an uncompressed snapshot".into(),
));
}
if header.gacl_enabled() || config.gacl {
return Err(HoronError::Config(
"partial_reads is not yet compatible with GACL enforcement".into(),
));
}
return Self::open_partial(path, file, header, ma_bounds, config);
}
let snap_entries = {
let mut entries =
snapshot::read_snapshot(&mut file, compressed, &layout, snapshot_has_crc)?;
if meaning_addressed {
entries.sort_by(|a, b| {
let da = a.key.matches('/').count();
let db = b.key.matches('/').count();
da.cmp(&db).then(a.key.cmp(&b.key))
});
}
entries
};
let (wal_entry_count, wal_base_seq) = wal::read_wal_header(&mut file)?;
let wal_data_offset = file.stream_position()?;
let store = Store::with_config(
StoreConfig::new().tau(g_math::fixed_point::FixedPoint::from_raw(header.tau_raw)),
);
load_snapshot_into_store(&store, &snap_entries, config.lazy_geometry)?;
let lazy = config.lazy_geometry;
let replayed_epoch = std::cell::Cell::new(0u64);
let (next_seq, valid_wal_count, wal_valid_end) = scan_wal(
&mut file,
&header,
&layout,
wal_base_seq,
wal_entry_count,
|entry| {
if let WalPayload::Epoch { epoch_id, .. } = &entry.payload {
replayed_epoch.set((*epoch_id).max(replayed_epoch.get()));
}
replay_entry(&store, entry, lazy)
},
)?;
if wal_valid_end < file.metadata()?.len() {
log::warn!(
"truncating torn WAL tail at byte {} (file was {} bytes)",
wal_valid_end,
file.metadata()?.len()
);
file.set_len(wal_valid_end)?;
file.sync_all()?;
}
file.seek(SeekFrom::End(0))?;
let partial: Option<PartialState> = None;
let file_config = HoronConfig {
dimension: header.dimension,
semantic_dims: header.semantic_dims,
compression: compressed,
auto_compact_threshold: config.auto_compact_threshold,
wal_batch_size: config.wal_batch_size,
wal_flush_interval_ms: config.wal_flush_interval_ms,
durability: config.durability,
gacl: header.gacl_enabled(),
gacl_fail_closed: config.gacl_fail_closed,
lazy_geometry: config.lazy_geometry,
approximate_semantic: config.approximate_semantic,
partial_reads: config.partial_reads,
meaning_addressed,
semantic_bounds: ma_bounds.unwrap_or(config.semantic_bounds),
history_retention: config.history_retention,
quantized_semantic: header.quantized_semantic(),
};
let wal = WalWriter {
file,
next_seq,
entry_count: valid_wal_count,
wal_data_offset,
pending_serialized: Vec::new(),
pending_entries: Vec::new(),
subscribers: Vec::new(),
pending_count: 0,
batch_size: config.wal_batch_size,
flush_interval: Duration::from_millis(config.wal_flush_interval_ms),
last_flush: Instant::now(),
durability: config.durability,
wal_compressed: header.wal_compressed(),
compression_algo: header.compression_algo(),
layout,
};
Ok(Self {
store,
path: path.to_path_buf(),
header,
wal: Mutex::new(wal),
config: file_config,
compacting: AtomicBool::new(false),
credentials: RwLock::new(None),
last_compaction_error: Mutex::new(None),
self_weak: OnceLock::new(),
partial,
ma_bounds,
current_epoch: AtomicU64::new(replayed_epoch.get()),
})
}
fn open_partial(
path: &Path,
mut file: File,
header: GeoHeader,
ma_bounds: Option<(f64, f64)>,
config: &HoronConfig,
) -> HoronResult<Self> {
use std::io::Read;
let layout = SemLayout::from_header(&header);
let semantic_dims = header.semantic_dims as usize;
let snapshot_has_crc = header.version >= 2;
let mut b4 = [0u8; 4];
file.read_exact(&mut b4)?;
let snap_byte_len = u32::from_le_bytes(b4) as usize;
if snap_byte_len > MAX_SNAPSHOT_BYTES {
return Err(HoronError::InvalidFormat(format!(
"snapshot byte length {} exceeds maximum {}",
snap_byte_len, MAX_SNAPSHOT_BYTES
)));
}
file.read_exact(&mut b4)?;
let node_count = u32::from_le_bytes(b4) as usize;
if node_count > snap_byte_len / 8 + 1 {
return Err(HoronError::InvalidFormat(format!(
"snapshot node count {} impossible for {} section bytes",
node_count, snap_byte_len
)));
}
let raw_start = file.stream_position()? as usize;
file.seek(SeekFrom::Start((raw_start + snap_byte_len) as u64))?;
let stored_crc = if snapshot_has_crc {
file.read_exact(&mut b4)?;
Some(u32::from_le_bytes(b4))
} else {
None
};
let (wal_entry_count, wal_base_seq) = wal::read_wal_header(&mut file)?;
let wal_data_offset = file.stream_position()?;
let mmap = unsafe { memmap2::Mmap::map(&file)? };
if let Some(stored) = stored_crc {
let region = mmap.get(raw_start..raw_start + snap_byte_len).ok_or_else(|| {
HoronError::InvalidFormat("snapshot region exceeds file length".into())
})?;
let computed = crc32fast::hash(region);
if stored != computed {
return Err(HoronError::ChecksumMismatch {
expected: stored,
actual: computed,
context: "snapshot section".to_string(),
});
}
}
let hilbert_fn;
let hilbert_of: Option<&dyn Fn(&[u8]) -> u128> = match ma_bounds {
Some(b) => {
hilbert_fn = move |sem: &[u8]| {
if layout.quantized {
match layout.decode_tail(sem) {
Ok(full) => global_hilbert(&full, semantic_dims, b),
Err(_) => 0,
}
} else {
global_hilbert(sem, semantic_dims, b)
}
};
Some(&hilbert_fn)
}
None => None,
};
let view = SnapView::scan(
mmap, raw_start, snap_byte_len, node_count, layout, hilbert_of,
)?;
let store = Store::with_config(
StoreConfig::new().tau(g_math::fixed_point::FixedPoint::from_raw(header.tau_raw)),
);
let mut tombstones = HashSet::new();
let replayed_epoch = std::cell::Cell::new(0u64);
let (next_seq, valid_wal_count, wal_valid_end) = scan_wal(
&mut file,
&header,
&layout,
wal_base_seq,
wal_entry_count,
|entry| {
if let WalPayload::Epoch { epoch_id, .. } = &entry.payload {
replayed_epoch.set((*epoch_id).max(replayed_epoch.get()));
}
apply_wal_partial(&store, &view, &mut tombstones, entry)
},
)?;
if wal_valid_end < file.metadata()?.len() {
log::warn!(
"truncating torn WAL tail at byte {} (file was {} bytes)",
wal_valid_end,
file.metadata()?.len()
);
file.set_len(wal_valid_end)?;
file.sync_all()?;
}
file.seek(SeekFrom::End(0))?;
let partial = Some(PartialState {
view: RwLock::new(view),
tombstones: RwLock::new(tombstones),
scanned: AtomicUsize::new(0),
});
let file_config = HoronConfig {
dimension: header.dimension,
semantic_dims: header.semantic_dims,
compression: false,
approximate_semantic: config.approximate_semantic,
auto_compact_threshold: config.auto_compact_threshold,
wal_batch_size: config.wal_batch_size,
wal_flush_interval_ms: config.wal_flush_interval_ms,
durability: config.durability,
gacl: false,
gacl_fail_closed: config.gacl_fail_closed,
lazy_geometry: true,
partial_reads: true,
meaning_addressed: ma_bounds.is_some(),
semantic_bounds: ma_bounds.unwrap_or(config.semantic_bounds),
history_retention: config.history_retention,
quantized_semantic: header.quantized_semantic(),
};
let wal_writer = WalWriter {
file,
next_seq,
entry_count: valid_wal_count,
wal_data_offset,
pending_serialized: Vec::new(),
pending_entries: Vec::new(),
subscribers: Vec::new(),
pending_count: 0,
batch_size: config.wal_batch_size,
flush_interval: Duration::from_millis(config.wal_flush_interval_ms),
last_flush: Instant::now(),
durability: config.durability,
wal_compressed: header.wal_compressed(),
compression_algo: header.compression_algo(),
layout,
};
Ok(Self {
store,
path: path.to_path_buf(),
header,
wal: Mutex::new(wal_writer),
config: file_config,
compacting: AtomicBool::new(false),
credentials: RwLock::new(None),
last_compaction_error: Mutex::new(None),
self_weak: OnceLock::new(),
partial,
ma_bounds,
current_epoch: AtomicU64::new(replayed_epoch.get()),
})
}
fn sem_layout(&self) -> SemLayout {
SemLayout::from_header(&self.header)
}
pub fn set_credentials(&self, creds: Credentials) {
let mut guard = self.credentials.write().unwrap_or_else(|e| e.into_inner());
*guard = Some(creds);
}
pub fn clear_credentials(&self) {
let mut guard = self.credentials.write().unwrap_or_else(|e| e.into_inner());
*guard = None;
}
pub fn gacl_active(&self) -> bool {
if !self.header.gacl_enabled() {
return false;
}
let guard = self.credentials.read().unwrap_or_else(|e| e.into_inner());
guard.is_some() || self.config.gacl_fail_closed
}
fn check_read(&self, key: &str) -> HoronResult<()> {
if !self.header.gacl_enabled() {
return Ok(());
}
let guard = self.credentials.read().unwrap_or_else(|e| e.into_inner());
let creds = match guard.as_ref() {
Some(c) => c,
None => {
return self.no_credentials_result(key);
}
};
let bands = self.node_bands(key);
if creds.can_read(&bands) {
Ok(())
} else {
Err(HoronError::AccessDenied {
key: key.to_string(),
reason: "read access denied by GACL".to_string(),
})
}
}
fn check_write(&self, key: &str) -> HoronResult<()> {
if !self.header.gacl_enabled() {
return Ok(());
}
let guard = self.credentials.read().unwrap_or_else(|e| e.into_inner());
let creds = match guard.as_ref() {
Some(c) => c,
None => return self.no_credentials_result(key),
};
let bands = self.node_bands(key);
if creds.can_access(&bands) {
Ok(())
} else {
Err(HoronError::AccessDenied {
key: key.to_string(),
reason: "write access denied by GACL".to_string(),
})
}
}
fn no_credentials_result(&self, key: &str) -> HoronResult<()> {
if self.config.gacl_fail_closed {
Err(HoronError::AccessDenied {
key: key.to_string(),
reason: "GACL enabled but no credentials set (fail-closed)".to_string(),
})
} else {
Ok(())
}
}
fn node_bands(&self, key: &str) -> NodeAccessBands {
match self.store.get_semantic(key) {
Ok(sem) if sem.len() >= 12 * 16 => {
NodeAccessBands::from_semantic_bytes(&sem)
.unwrap_or_else(NodeAccessBands::public)
}
_ => NodeAccessBands::public(),
}
}
fn retain_readable<T>(&self, items: Vec<T>, key_of: impl Fn(&T) -> &str) -> Vec<T> {
if !self.header.gacl_enabled() {
return items;
}
let guard = self.credentials.read().unwrap_or_else(|e| e.into_inner());
let creds = match guard.as_ref() {
Some(c) => c,
None => return if self.config.gacl_fail_closed { Vec::new() } else { items },
};
items.into_iter().filter(|item| {
let bands = self.node_bands(key_of(item));
creds.can_read(&bands)
}).collect()
}
fn filter_readable_paths(&self, paths: Vec<String>) -> Vec<String> {
self.retain_readable(paths, |k| k.as_str())
}
fn collect_k_readable<T>(
&self,
k: usize,
key_of: impl Fn(&T) -> &str + Copy,
mut fetch: impl FnMut(usize) -> HoronResult<Vec<T>>,
) -> HoronResult<Vec<T>> {
if k == 0 {
return Ok(Vec::new());
}
let total = self.store.len().max(1);
let mut window = k.saturating_mul(3).max(16);
loop {
let raw = fetch(window)?;
let fetched = raw.len();
let filtered = self.retain_readable(raw, key_of);
if filtered.len() >= k || fetched >= total || window >= total {
return Ok(filtered.into_iter().take(k).collect());
}
window = window.saturating_mul(4).min(total);
}
}
fn validate_write(key: &str, data_len: usize, meta: Option<(&str, &str)>) -> HoronResult<()> {
if key.len() > u16::MAX as usize {
return Err(HoronError::InvalidOperation(format!(
"key length {} exceeds format maximum {}",
key.len(),
u16::MAX
)));
}
if data_len > crate::format::MAX_ENTRY_DATA {
return Err(HoronError::InvalidOperation(format!(
"data length {} exceeds format maximum {}",
data_len,
crate::format::MAX_ENTRY_DATA
)));
}
if let Some((mk, mv)) = meta {
if mk.len() > u16::MAX as usize || mv.len() > u16::MAX as usize {
return Err(HoronError::InvalidOperation(format!(
"metadata key/value length {}/{} exceeds format maximum {}",
mk.len(),
mv.len(),
u16::MAX
)));
}
}
Ok(())
}
pub fn put(&self, key: &str, data: &[u8]) -> HoronResult<()> {
self.put_inner(key, data, true)
}
pub fn put_data_only(&self, key: &str, data: &[u8]) -> HoronResult<()> {
self.put_inner(key, data, false)
}
fn ancestor_paths(key: &str) -> Vec<String> {
let parts: Vec<&str> = key.split('/').filter(|p| !p.is_empty()).collect();
(1..parts.len())
.map(|depth| format!("/{}", parts[..depth].join("/")))
.collect()
}
fn put_inner(&self, key: &str, data: &[u8], geometry: bool) -> HoronResult<()> {
Self::validate_write(key, data.len(), None)?;
let is_update = self.exists_inner(key);
let new_ancestors: Vec<String> = if is_update {
Vec::new()
} else {
Self::ancestor_paths(key)
.into_iter()
.filter(|ancestor| !self.exists_inner(ancestor))
.collect()
};
if is_update {
self.check_write(key)?;
}
if let Some(p) = &self.partial {
{
let view = p.view.read().unwrap_or_else(|e| e.into_inner());
let _ = promote_from_view(&self.store, &view, key);
ensure_ancestors(&self.store, &view, key);
}
Store::put_data_only(&self.store, key, data)?;
p.tombstones.write().unwrap_or_else(|e| e.into_inner()).remove(key);
} else if geometry {
self.store.put(key, data)?;
} else {
self.store.put_data_only(key, data)?;
}
let should_compact = {
let mut wal = self.wal.lock().unwrap_or_else(|e| e.into_inner());
let sem_bytes = self.header.semantic_dims as usize * 16;
for ancestor in &new_ancestors {
let mut metadata: Vec<(String, String)> = self
.store
.get_meta(ancestor)
.unwrap_or_default()
.into_iter()
.filter(|(k, _)| {
k != "key" && k != "size" && k != "created_at" && k != "updated_at"
})
.collect();
metadata.sort();
let entry = WalEntry {
seq: wal.next_seq,
op: OP_INSERT,
key: ancestor.clone(),
payload: WalPayload::Insert(NodeEntry {
key: ancestor.clone(),
data: Vec::new(),
metadata,
semantic_coords: vec![0u8; sem_bytes],
}),
};
wal.append(&entry)?;
}
let entry = if is_update {
WalEntry {
seq: wal.next_seq,
op: OP_UPDATE,
key: key.to_string(),
payload: WalPayload::Update {
data: data.to_vec(),
metadata: vec![],
},
}
} else {
WalEntry {
seq: wal.next_seq,
op: OP_INSERT,
key: key.to_string(),
payload: WalPayload::Insert(NodeEntry {
key: key.to_string(),
data: data.to_vec(),
metadata: vec![],
semantic_coords: vec![0u8; sem_bytes],
}),
}
};
wal.append(&entry)?;
self.config.auto_compact_threshold > 0
&& wal.total_entries() >= self.config.auto_compact_threshold
};
if should_compact {
self.spawn_auto_compact();
}
Ok(())
}
pub fn get(&self, key: &str) -> HoronResult<Vec<u8>> {
self.check_read(key)?;
if let Some(p) = &self.partial {
if self.store.exists(key) {
return Ok(self.store.get(key)?);
}
if !p.tombstones.read().unwrap_or_else(|e| e.into_inner()).contains(key) {
let view = p.view.read().unwrap_or_else(|e| e.into_inner());
if let Some(&idx) = view.by_key.get(key) {
return Ok(view.decode(idx)?.data);
}
}
return Err(HoronError::Store(
horon_engine::store::StoreError::NotFound(key.to_string()),
));
}
Ok(self.store.get(key)?)
}
pub fn remove(&self, key: &str) -> HoronResult<()> {
self.check_write(key)?;
if let Some(p) = &self.partial {
let in_store = self.store.exists(key);
let in_snapshot = {
let view = p.view.read().unwrap_or_else(|e| e.into_inner());
view.by_key.contains_key(key)
};
let tombstoned = p
.tombstones
.read()
.unwrap_or_else(|e| e.into_inner())
.contains(key);
if !in_store && (!in_snapshot || tombstoned) {
return Err(HoronError::Store(
horon_engine::store::StoreError::NotFound(key.to_string()),
));
}
if in_store {
self.store.remove(key)?;
}
p.tombstones
.write()
.unwrap_or_else(|e| e.into_inner())
.insert(key.to_string());
} else {
self.store.remove(key)?;
}
let should_compact = {
let mut wal = self.wal.lock().unwrap_or_else(|e| e.into_inner());
let entry = WalEntry {
seq: wal.next_seq,
op: OP_DELETE,
key: key.to_string(),
payload: WalPayload::Delete,
};
wal.append(&entry)?;
self.config.auto_compact_threshold > 0
&& wal.total_entries() >= self.config.auto_compact_threshold
};
if should_compact {
self.spawn_auto_compact();
}
Ok(())
}
pub fn exists(&self, key: &str) -> bool {
if !self.exists_inner(key) {
return false;
}
self.check_read(key).is_ok()
}
fn exists_inner(&self, key: &str) -> bool {
if self.store.exists(key) {
return true;
}
if let Some(p) = &self.partial {
if p.tombstones.read().unwrap_or_else(|e| e.into_inner()).contains(key) {
return false;
}
return p
.view
.read()
.unwrap_or_else(|e| e.into_inner())
.by_key
.contains_key(key);
}
false
}
pub fn set_meta(&self, key: &str, name: &str, value: &str) -> HoronResult<()> {
Self::validate_write(key, 0, Some((name, value)))?;
self.check_write(key)?;
if let Some(p) = &self.partial {
let view = p.view.read().unwrap_or_else(|e| e.into_inner());
promote_from_view(&self.store, &view, key)?;
}
self.store.set_meta(key, name, value)?;
let should_compact = {
let mut wal = self.wal.lock().unwrap_or_else(|e| e.into_inner());
let entry = WalEntry {
seq: wal.next_seq,
op: OP_SET_META,
key: key.to_string(),
payload: WalPayload::SetMeta {
meta_key: name.to_string(),
meta_value: value.to_string(),
},
};
wal.append(&entry)?;
self.config.auto_compact_threshold > 0
&& wal.total_entries() >= self.config.auto_compact_threshold
};
if should_compact {
self.spawn_auto_compact();
}
Ok(())
}
pub fn get_meta(&self, key: &str) -> HoronResult<HashMap<String, String>> {
self.check_read(key)?;
if let Some(p) = &self.partial {
if self.store.exists(key) {
return Ok(self.store.get_meta(key)?);
}
if !p.tombstones.read().unwrap_or_else(|e| e.into_inner()).contains(key) {
let view = p.view.read().unwrap_or_else(|e| e.into_inner());
if let Some(&idx) = view.by_key.get(key) {
let entry = view.decode(idx)?;
let mut map: HashMap<String, String> =
entry.metadata.into_iter().collect();
map.insert("key".to_string(), entry.key.clone());
map.insert("size".to_string(), entry.data.len().to_string());
return Ok(map);
}
}
return Err(HoronError::Store(
horon_engine::store::StoreError::NotFound(key.to_string()),
));
}
Ok(self.store.get_meta(key)?)
}
pub fn set_semantic(&self, key: &str, coords: Vec<u8>) -> HoronResult<()> {
Self::validate_write(key, 0, None)?;
self.check_write(key)?;
let mut coords = coords;
let layout = self.sem_layout();
if layout.quantized {
layout.canonicalize(&mut coords)?;
} else if coords.len() != layout.mem_bytes() {
if coords.len() > layout.mem_bytes() {
return Err(HoronError::InvalidOperation(format!(
"coords cover {} bytes but the file has {} semantic dims ({} bytes)",
coords.len(),
layout.dims,
layout.mem_bytes()
)));
}
coords.resize(layout.mem_bytes(), 0);
}
if coords.iter().all(|&b| b == 0) {
return Err(HoronError::InvalidOperation(
"all-zero semantic coordinates encode \"not set\" and cannot be stored as a \
placement; use clear_semantic() to remove a placement, or offset at least one \
dimension from zero for a placement near the origin"
.to_string(),
));
}
if let Some(p) = &self.partial {
let view = p.view.read().unwrap_or_else(|e| e.into_inner());
promote_from_view(&self.store, &view, key)?;
}
self.store.set_semantic(key, coords.clone())?;
let should_compact = {
let mut wal = self.wal.lock().unwrap_or_else(|e| e.into_inner());
let entry = WalEntry {
seq: wal.next_seq,
op: OP_SET_SEMANTIC,
key: key.to_string(),
payload: WalPayload::SetSemantic {
coords: coords,
},
};
wal.append(&entry)?;
self.config.auto_compact_threshold > 0
&& wal.total_entries() >= self.config.auto_compact_threshold
};
if should_compact {
self.spawn_auto_compact();
}
Ok(())
}
pub fn clear_semantic(&self, key: &str) -> HoronResult<()> {
Self::validate_write(key, 0, None)?;
self.check_write(key)?;
if let Some(p) = &self.partial {
let view = p.view.read().unwrap_or_else(|e| e.into_inner());
promote_from_view(&self.store, &view, key)?;
}
self.store.set_semantic(key, Vec::new())?;
let should_compact = {
let mut wal = self.wal.lock().unwrap_or_else(|e| e.into_inner());
let entry = WalEntry {
seq: wal.next_seq,
op: OP_SET_SEMANTIC,
key: key.to_string(),
payload: WalPayload::SetSemantic {
coords: vec![0u8; self.sem_layout().mem_bytes()],
},
};
wal.append(&entry)?;
self.config.auto_compact_threshold > 0
&& wal.total_entries() >= self.config.auto_compact_threshold
};
if should_compact {
self.spawn_auto_compact();
}
Ok(())
}
pub fn get_semantic(&self, key: &str) -> HoronResult<Vec<u8>> {
self.check_read(key)?;
if let Some(p) = &self.partial {
if self.store.exists(key) {
return Ok(self.store.get_semantic(key)?);
}
if !p.tombstones.read().unwrap_or_else(|e| e.into_inner()).contains(key) {
let view = p.view.read().unwrap_or_else(|e| e.into_inner());
if let Some(&idx) = view.by_key.get(key) {
let sem = view.semantic_of(idx)?; let layout = view.layout();
let full = if layout.quantized {
layout.decode_tail(sem)?
} else {
sem.to_vec()
};
if full.iter().any(|&b| b != 0) {
return Ok(full);
}
return Ok(Vec::new());
}
}
return Err(HoronError::Store(
horon_engine::store::StoreError::NotFound(key.to_string()),
));
}
Ok(self.store.get_semantic(key)?)
}
pub fn children(&self, path: &str) -> HoronResult<Vec<String>> {
if let Some(p) = &self.partial {
let base = if path == "/" { String::new() } else { path.trim_end_matches('/').to_string() };
let direct_child = |k: &str| -> bool {
match k.strip_prefix(&base) {
Some(rest) => rest.len() > 1 && rest.starts_with('/') && !rest[1..].contains('/'),
None => false,
}
};
let mut out: std::collections::BTreeSet<String> = self
.store
.list("/")
.unwrap_or_default()
.into_iter()
.filter(|k| direct_child(k))
.collect();
let view = p.view.read().unwrap_or_else(|e| e.into_inner());
let tombs = p.tombstones.read().unwrap_or_else(|e| e.into_inner());
for e in &view.entries {
if direct_child(&e.key) && !tombs.contains(&e.key) {
out.insert(e.key.clone());
}
}
return Ok(out.into_iter().collect());
}
let kids = self.store.children(path)?;
Ok(self.filter_readable_paths(kids))
}
pub fn list(&self, prefix: &str) -> HoronResult<Vec<String>> {
if let Some(p) = &self.partial {
let mut out: std::collections::BTreeSet<String> =
self.store.list(prefix).unwrap_or_default().into_iter().collect();
let view = p.view.read().unwrap_or_else(|e| e.into_inner());
let tombs = p.tombstones.read().unwrap_or_else(|e| e.into_inner());
for e in &view.entries {
if e.key.starts_with(prefix) && !tombs.contains(&e.key) {
out.insert(e.key.clone());
}
}
return Ok(out.into_iter().collect());
}
let keys = self.store.list(prefix)?;
Ok(self.filter_readable_paths(keys))
}
pub fn nearest(&self, coords: &[FixedPoint]) -> HoronResult<(String, FixedPoint)> {
if !self.gacl_active() {
return Ok(self.store.nearest(coords)?);
}
let closest = self.collect_k_readable(1, |(k, _): &(String, FixedPoint)| k.as_str(), |n| {
Ok(self.store.nearest_k(coords, n)?)
})?;
closest.into_iter().next()
.ok_or_else(|| HoronError::AccessDenied {
key: "(nearest query)".to_string(),
reason: "no accessible nodes found near query point".to_string(),
})
}
pub fn neighbors(&self, path: &str, k: usize) -> HoronResult<Vec<String>> {
if !self.gacl_active() {
return Ok(self.store.neighbors(path, k)?);
}
self.collect_k_readable(k, |p: &String| p.as_str(), |n| {
Ok(self.store.neighbors(path, n)?)
})
}
pub fn nearest_semantic(
&self,
query_coords: &[u8],
k: usize,
dim_range: std::ops::Range<usize>,
) -> HoronResult<Vec<(String, FixedPoint)>> {
if self.partial.is_some() {
return self.nearest_semantic_partial(query_coords, k, dim_range);
}
if !self.gacl_active() {
return Ok(self.store.nearest_semantic(query_coords, k, dim_range)?);
}
self.collect_k_readable(k, |(key, _): &(String, FixedPoint)| key.as_str(), |n| {
Ok(self.store.nearest_semantic(query_coords, n, dim_range.clone())?)
})
}
pub fn neighbors_semantic(
&self,
path: &str,
k: usize,
dim_range: std::ops::Range<usize>,
) -> HoronResult<Vec<(String, FixedPoint)>> {
if self.partial.is_some() {
let coords = self.get_semantic(path)?;
if coords.is_empty() {
return Ok(Vec::new());
}
let results = self.nearest_semantic_partial(&coords, k + 1, dim_range)?;
return Ok(results.into_iter().filter(|(key, _)| key != path).take(k).collect());
}
if !self.gacl_active() {
return Ok(self.store.neighbors_semantic(path, k, dim_range)?);
}
self.collect_k_readable(k, |(key, _): &(String, FixedPoint)| key.as_str(), |n| {
Ok(self.store.neighbors_semantic(path, n, dim_range.clone())?)
})
}
pub fn nearest_k(
&self,
coords: &[FixedPoint],
k: usize,
) -> HoronResult<Vec<(String, FixedPoint)>> {
if !self.gacl_active() {
return Ok(self.store.nearest_k(coords, k)?);
}
self.collect_k_readable(k, |(key, _): &(String, FixedPoint)| key.as_str(), |n| {
Ok(self.store.nearest_k(coords, n)?)
})
}
pub fn find_similar(
&self,
key: &str,
k: usize,
dim_range: std::ops::Range<usize>,
) -> HoronResult<Vec<(String, FixedPoint)>> {
self.neighbors_semantic(key, k, dim_range)
}
pub fn find_outliers(
&self,
prefix: &str,
z_threshold: FixedPoint,
dim_range: std::ops::Range<usize>,
) -> HoronResult<Vec<SemanticOutlier>> {
let outliers = self.store.find_outliers(prefix, z_threshold, dim_range)?;
Ok(self.retain_readable(outliers, |o| o.key.as_str()))
}
pub fn find_within(&self, path: &str, radius: FixedPoint) -> HoronResult<Vec<String>> {
let keys = self.store.find_within(path, radius)?;
Ok(self.filter_readable_paths(keys))
}
pub fn position(&self, key: &str) -> HoronResult<Vec<FixedPoint>> {
self.check_read(key)?;
Ok(self.store.position(key)?)
}
pub fn len(&self) -> usize {
if let Some(p) = &self.partial {
let view = p.view.read().unwrap_or_else(|e| e.into_inner());
let tombs = p.tombstones.read().unwrap_or_else(|e| e.into_inner());
let snapshot_only = view
.entries
.iter()
.filter(|e| !self.store.exists(&e.key) && !tombs.contains(&e.key))
.count();
return self.store.len() + snapshot_only;
}
self.store.len()
}
pub fn is_empty(&self) -> bool {
if self.partial.is_some() {
return self.len() == 0;
}
self.store.is_empty()
}
pub fn store(&self) -> &Store {
&self.store
}
fn nearest_semantic_partial(
&self,
query_coords: &[u8],
k: usize,
dim_range: std::ops::Range<usize>,
) -> HoronResult<Vec<(String, FixedPoint)>> {
let p = self.partial.as_ref().expect("partial mode");
if k == 0 {
return Ok(Vec::new());
}
let view = p.view.read().unwrap_or_else(|e| e.into_inner());
let tombs = p.tombstones.read().unwrap_or_else(|e| e.into_inner());
let n = view.entries.len();
let addressed_end = DIM_USER_DEFINED_START + MAX_HILBERT_DIMS;
let windowed = self.config.approximate_semantic
&& self.ma_bounds.is_some()
&& dim_range.start >= DIM_USER_DEFINED_START
&& dim_range.end <= addressed_end;
let mut best: Vec<(String, FixedPoint)> = Vec::new();
let mut scanned = 0usize;
let qv = decode_dim_slice(query_coords, &dim_range);
let consider = |i: usize, view: &SnapView, best: &mut Vec<(String, FixedPoint)>,
scanned: &mut usize| -> HoronResult<()> {
let e = &view.entries[i];
if tombs.contains(&e.key) || self.store.exists(&e.key) {
return Ok(()); }
let sem = view.semantic_of(i)?; if !sem.iter().any(|&b| b != 0) {
return Ok(()); }
*scanned += 1;
let layout = view.layout();
let dsq = if layout.quantized {
let full = layout.decode_tail(sem)?;
semantic_distance_sq(&qv, &full, &dim_range)
} else {
semantic_distance_sq(&qv, sem, &dim_range)
};
best.push((e.key.clone(), dsq));
Ok(())
};
if windowed && n > 0 {
let bounds = self.ma_bounds.unwrap();
let sem_dims = self.header.semantic_dims as usize;
let qvals = decode_user_dims(query_coords, sem_dims);
let span = bounds.1 - bounds.0;
let mut half = (2 * k).max(16);
let mut delta = FixedPoint::from_f64(span / 256.0);
let mut seen: HashSet<usize> = HashSet::new();
loop {
best.clear();
seen.clear();
scanned = 0;
let mut probes: Vec<Vec<FixedPoint>> = vec![qvals.clone()];
for d in 0..qvals.len() {
for sign in [-1i32, 1] {
let mut v = qvals.clone();
v[d] = v[d] + FixedPoint::from_int(sign) * delta;
probes.push(v);
}
}
let mut covered_all = true;
for probe in &probes {
let center = view.hilbert_position(hilbert_from_values(probe, bounds));
let lo = center.saturating_sub(half);
let hi = (center + half).min(n);
if lo > 0 || hi < n {
covered_all = false;
}
for i in lo..hi {
if seen.insert(i) {
consider(i, &view, &mut best, &mut scanned)?;
}
}
}
if best.len() >= k * 2 || covered_all {
break;
}
half *= 2;
delta = delta * FixedPoint::from_int(2);
}
} else {
for i in 0..n {
consider(i, &view, &mut best, &mut scanned)?;
}
}
p.scanned.store(scanned, Ordering::Relaxed);
best.sort_by(|a, b| a.1.partial_cmp(&b.1).unwrap_or(std::cmp::Ordering::Equal));
best.truncate(k);
for (key, d) in best.iter_mut() {
*d = match view.by_key.get(key).copied() {
Some(i) => {
let sem = view.semantic_of(i)?;
let layout = view.layout();
if layout.quantized {
let full = layout.decode_tail(sem)?;
Store::semantic_distance(query_coords, &full, dim_range.clone())
} else {
Store::semantic_distance(query_coords, sem, dim_range.clone())
}
}
None => d.sqrt(),
};
}
let overlay = self
.store
.nearest_semantic(query_coords, k, dim_range)
.unwrap_or_default();
best.extend(overlay);
best.sort_by(|a, b| a.1.partial_cmp(&b.1).unwrap_or(std::cmp::Ordering::Equal));
best.truncate(k);
Ok(best)
}
pub fn last_semantic_scan_count(&self) -> Option<usize> {
self.partial.as_ref().map(|p| p.scanned.load(Ordering::Relaxed))
}
pub fn subscribe_wal(
&self,
) -> HoronResult<(u32, std::sync::mpsc::Receiver<WalEntry>)> {
let mut wal = self.wal.lock().unwrap_or_else(|e| e.into_inner());
wal.flush_pending()?;
let (tx, rx) = std::sync::mpsc::channel();
wal.subscribers.push(tx);
Ok((wal.next_seq, rx))
}
pub fn wal_entries_since(&self, from_seq: u32) -> HoronResult<WalTail> {
let mut wal = self.wal.lock().unwrap_or_else(|e| e.into_inner());
wal.flush_pending()?;
let base_seq = wal.next_seq - wal.entry_count;
if from_seq < base_seq {
return Ok(WalTail::SnapshotRequired { base_seq });
}
let end_pos = wal.file.stream_position()?;
let data_offset = wal.wal_data_offset;
wal.file.seek(SeekFrom::Start(data_offset))?;
let layout = self.sem_layout();
let mut out = Vec::new();
let wal_compressed = wal.wal_compressed;
let algo = wal.compression_algo;
let entry_count = wal.entry_count;
let scan_result = scan_wal_raw(
&mut wal.file,
wal_compressed,
algo,
&layout,
base_seq,
entry_count,
|entry| {
if entry.seq >= from_seq {
out.push(entry.clone());
}
Ok(())
},
);
wal.file.seek(SeekFrom::Start(end_pos))?;
scan_result?;
Ok(WalTail::Entries(out))
}
pub fn compact(&self) -> HoronResult<bool> {
if self.compacting.compare_exchange(
false, true, Ordering::SeqCst, Ordering::SeqCst,
).is_err() {
return Ok(false);
}
let result = self.compact_inner();
self.compacting.store(false, Ordering::SeqCst);
{
let mut last = self
.last_compaction_error
.lock()
.unwrap_or_else(|e| e.into_inner());
match &result {
Ok(()) => *last = None,
Err(e) => {
log::error!("compaction failed: {} — WAL will keep growing until a compaction succeeds", e);
*last = Some(e.to_string());
}
}
}
result.map(|()| true)
}
pub fn last_compaction_error(&self) -> Option<String> {
self.last_compaction_error
.lock()
.unwrap_or_else(|e| e.into_inner())
.clone()
}
pub fn compact_async(&self) -> JoinHandle<HoronResult<bool>> {
let this = self
.self_weak
.get()
.and_then(Weak::upgrade)
.expect("Horon must be constructed via Horon::open()");
std::thread::spawn(move || this.compact())
}
fn spawn_auto_compact(&self) {
if self.compacting.load(Ordering::SeqCst) {
return;
}
match self.self_weak.get().and_then(Weak::upgrade) {
Some(core) => {
std::thread::spawn(move || {
let _ = core.compact();
});
}
None => {
let _ = self.compact();
}
}
}
fn compact_inner(&self) -> HoronResult<()> {
let layout = self.sem_layout();
let sem_bytes = layout.mem_bytes();
let compressed = self.config.compression;
let (fence_seq, fence_file_pos) = {
let mut wal = self.wal.lock().unwrap_or_else(|e| e.into_inner());
wal.flush_pending()?;
let fence_seq = wal.next_seq;
let fence_file_pos = wal.file.stream_position()?;
(fence_seq, fence_file_pos)
};
let mut entries = Vec::new();
if let Some(p) = &self.partial {
let view = p.view.read().unwrap_or_else(|e| e.into_inner());
let tombs = p.tombstones.read().unwrap_or_else(|e| e.into_inner());
for (i, loc) in view.entries.iter().enumerate() {
if !self.store.exists(&loc.key) && !tombs.contains(&loc.key) {
entries.push(view.decode(i)?);
}
}
}
let all_keys = self.store.list("/")?;
entries.reserve(all_keys.len());
for key in &all_keys {
let data = match self.store.get(key) {
Ok(d) => d,
Err(horon_engine::store::StoreError::NotFound(_)) => continue,
Err(e) => return Err(e.into()),
};
let meta = self.store.get_meta(key)?;
let mut metadata: Vec<(String, String)> = meta
.into_iter()
.filter(|(k, _)| {
k != "key" && k != "size" && k != "created_at" && k != "updated_at"
})
.collect();
metadata.sort();
let semantic_coords = self.store.get_semantic(key)
.unwrap_or_default();
let semantic_coords = if semantic_coords.len() == sem_bytes {
semantic_coords
} else {
let mut padded = vec![0u8; sem_bytes];
let copy_len = semantic_coords.len().min(sem_bytes);
padded[..copy_len].copy_from_slice(&semantic_coords[..copy_len]);
padded
};
entries.push(NodeEntry {
key: key.clone(),
data,
metadata,
semantic_coords,
});
}
let entries: Vec<NodeEntry> = if let Some(bounds) = self.ma_bounds {
let sem_dims = self.header.semantic_dims as usize;
let mut indexed: Vec<(u128, NodeEntry)> = entries
.into_iter()
.map(|e| (global_hilbert(&e.semantic_coords, sem_dims, bounds), e))
.collect();
indexed.sort_by(|(ra, a), (rb, b)| ra.cmp(rb).then(a.key.cmp(&b.key)));
indexed.into_iter().map(|(_, e)| e).collect()
} else {
let ranks = hilbert_snapshot_ranks(&entries, self.header.semantic_dims as usize);
let mut indexed: Vec<(u128, NodeEntry)> = ranks.into_iter().zip(entries).collect();
indexed.sort_by(|(rank_a, a), (rank_b, b)| {
let depth_a = a.key.matches('/').count();
let depth_b = b.key.matches('/').count();
depth_a
.cmp(&depth_b)
.then(rank_a.cmp(rank_b))
.then(a.key.cmp(&b.key))
});
indexed.into_iter().map(|(_, e)| e).collect()
};
let tmp_path = self.path.with_extension("htt.tmp");
let mut tmp_file = OpenOptions::new()
.read(true)
.write(true)
.create(true)
.truncate(true)
.open(&tmp_path)?;
try_lock_exclusive(&tmp_file, &tmp_path)?;
let mut header = self.header.clone();
header.node_count = entries.len() as u32;
header.version = if layout.quantized {
VERSION_QUANTIZED
} else if self.ma_bounds.is_some() {
VERSION_MEANING_ADDRESSED
} else {
VERSION
};
tmp_file.write_all(&header.to_bytes())?;
if let Some(bounds) = self.ma_bounds {
tmp_file.write_all(&Self::bounds_section_bytes(
self.header.semantic_dims as usize,
bounds,
))?;
}
let snap_raw_len: usize = {
let mut probe = Vec::new();
for e in &entries {
e.write_to(&mut probe, &layout)?;
}
probe.len()
};
snapshot::write_snapshot(&mut tmp_file, &entries, compressed, true, &layout)?;
let mut wal = self.wal.lock().unwrap_or_else(|e| e.into_inner());
wal.flush_pending()?;
let concurrent_count = wal.next_seq - fence_seq;
let mut concurrent_entries = Vec::new();
if concurrent_count > 0 {
wal.file.seek(SeekFrom::Start(fence_file_pos))?;
if wal.wal_compressed {
let algo = wal.compression_algo;
while concurrent_entries.len() < concurrent_count as usize {
match wal::read_wal_block(&mut wal.file, algo)? {
Some((block_data, block_entry_count)) => {
let mut cursor = std::io::Cursor::new(&block_data);
for _ in 0..block_entry_count {
if let Some(entry) = WalEntry::read_from(&mut cursor, &layout)? {
concurrent_entries.push(entry);
}
}
}
None => break,
}
}
} else {
for _ in 0..concurrent_count {
match WalEntry::read_from(&mut wal.file, &layout)? {
Some(entry) => concurrent_entries.push(entry),
None => break,
}
}
}
}
if self.config.history_retention == HistoryRetention::Archive {
let old_header_pos = wal.wal_data_offset - 8;
wal.file.seek(SeekFrom::Start(old_header_pos))?;
let (_old_count, old_base_seq) = wal::read_wal_header(&mut wal.file)?;
if fence_seq > old_base_seq {
let mut archived: Vec<WalEntry> = Vec::new();
if wal.wal_compressed {
let algo = wal.compression_algo;
'archive: while wal.file.stream_position()? < fence_file_pos {
match wal::read_wal_block(&mut wal.file, algo)? {
Some((block, n)) => {
let mut cursor = std::io::Cursor::new(&block);
for _ in 0..n {
match WalEntry::read_from(&mut cursor, &layout)? {
Some(e) if e.seq < fence_seq => archived.push(e),
_ => break 'archive,
}
}
}
None => break,
}
}
} else {
while wal.file.stream_position()? < fence_file_pos {
match WalEntry::read_from(&mut wal.file, &layout)? {
Some(e) if e.seq < fence_seq => archived.push(e),
_ => break,
}
}
}
if !archived.is_empty() {
let n = crate::history::next_segment_number(&self.path);
crate::history::write_segment(
&self.path,
n,
layout,
old_base_seq,
fence_seq,
&archived,
)?;
}
}
}
wal::write_wal_header(&mut tmp_file, concurrent_entries.len() as u32, fence_seq)?;
let new_wal_data_offset = tmp_file.stream_position()?;
if wal.wal_compressed && !concurrent_entries.is_empty() {
let algo = wal.compression_algo;
let serialized: Vec<Vec<u8>> = concurrent_entries
.iter()
.map(|e| {
let mut buf = Vec::new();
e.write_to(&mut buf, &layout).unwrap();
buf
})
.collect();
for chunk in serialized.chunks(WAL_BLOCK_SIZE) {
let mut block_bytes = Vec::new();
for entry_bytes in chunk {
block_bytes.extend_from_slice(entry_bytes);
}
wal::write_wal_block(&mut tmp_file, &block_bytes, chunk.len() as u16, algo)?;
}
} else {
for entry in &concurrent_entries {
entry.write_to(&mut tmp_file, &layout)?;
}
}
tmp_file.flush()?;
tmp_file.sync_all()?;
let mut new_file = tmp_file;
new_file.seek(SeekFrom::End(0))?;
let new_view = if self.partial.is_some() {
let mmap = unsafe { memmap2::Mmap::map(&new_file)? };
let sem_dims = self.header.semantic_dims as usize;
let raw_start = HEADER_SIZE
+ self.ma_bounds.map_or(0, |_| {
(sem_dims - DIM_USER_DEFINED_START) * 16
})
+ SNAP_HEADER_SIZE;
let hilbert_fn;
let hilbert_of: Option<&dyn Fn(&[u8]) -> u128> = match self.ma_bounds {
Some(b) => {
hilbert_fn = move |sem: &[u8]| {
if layout.quantized {
match layout.decode_tail(sem) {
Ok(full) => global_hilbert(&full, sem_dims, b),
Err(_) => 0,
}
} else {
global_hilbert(sem, sem_dims, b)
}
};
Some(&hilbert_fn)
}
None => None,
};
Some(SnapView::scan(
mmap,
raw_start,
snap_raw_len,
entries.len(),
layout,
hilbert_of,
)?)
} else {
None
};
fs::rename(&tmp_path, &self.path)?;
if let Some(p) = &self.partial {
if let Some(view) = new_view {
*p.view.write().unwrap_or_else(|e| e.into_inner()) = view;
}
{
let mut tombs = p.tombstones.write().unwrap_or_else(|e| e.into_inner());
let view = p.view.read().unwrap_or_else(|e| e.into_inner());
tombs.retain(|k| view.by_key.contains_key(k));
}
}
wal.file = new_file;
wal.entry_count = concurrent_entries.len() as u32;
wal.wal_data_offset = new_wal_data_offset;
wal.pending_serialized.clear();
wal.pending_count = 0;
fsync_dir(&self.path)?;
let epoch = self.current_epoch.load(Ordering::SeqCst);
if epoch > 0 {
let entry = WalEntry {
seq: wal.next_seq,
op: OP_EPOCH,
key: String::new(),
payload: WalPayload::Epoch { epoch_id: epoch, flags: EPOCH_FLAG_RESTAMP },
};
wal.append(&entry)?;
wal.flush_pending()?;
}
Ok(())
}
pub fn flush(&self) -> HoronResult<()> {
let mut wal = self.wal.lock().unwrap_or_else(|e| e.into_inner());
wal.flush_pending()?;
wal.file.flush()?;
Ok(())
}
pub fn seal_epoch(&self) -> HoronResult<u64> {
self.seal_epoch_inner(0)
}
pub fn seal_speculative_epoch(&self) -> HoronResult<u64> {
self.seal_epoch_inner(EPOCH_FLAG_SPECULATIVE)
}
fn seal_epoch_inner(&self, flags: u8) -> HoronResult<u64> {
let mut wal = self.wal.lock().unwrap_or_else(|e| e.into_inner());
let epoch_id = self.current_epoch.load(Ordering::SeqCst) + 1;
let entry = WalEntry {
seq: wal.next_seq,
op: OP_EPOCH,
key: String::new(),
payload: WalPayload::Epoch { epoch_id, flags },
};
wal.append(&entry)?;
wal.flush_pending()?;
self.current_epoch.store(epoch_id, Ordering::SeqCst);
Ok(epoch_id)
}
pub fn current_epoch(&self) -> u64 {
self.current_epoch.load(Ordering::SeqCst)
}
pub fn embed(&self, key: &str) -> HoronResult<bool> {
self.check_read(key)?;
if let Some(p) = &self.partial {
let view = p.view.read().unwrap_or_else(|e| e.into_inner());
ensure_ancestors(&self.store, &view, key);
promote_from_view(&self.store, &view, key)?;
}
Ok(self.store.embed_existing(key)?)
}
pub fn embed_all(&self, prefix: &str) -> HoronResult<usize> {
let mut upgraded = 0;
let mut keys = self.list(prefix)?;
if !keys.iter().any(|k| k == prefix) {
keys.push(prefix.to_string());
}
keys.sort();
for key in keys {
match self.embed(&key) {
Ok(true) => upgraded += 1,
Ok(false) => {}
Err(HoronError::AccessDenied { .. }) => {}
Err(e) => return Err(e),
}
}
Ok(upgraded)
}
pub fn wal_len(&self) -> u32 {
let wal = self.wal.lock().unwrap_or_else(|e| e.into_inner());
wal.total_entries()
}
pub fn path(&self) -> &Path {
&self.path
}
}
impl Drop for HoronCore {
fn drop(&mut self) {
let mut wal = self.wal.lock().unwrap_or_else(|e| e.into_inner());
let _ = wal.flush_pending();
let _ = wal.file.flush();
}
}
fn fsync_dir(path: &Path) -> HoronResult<()> {
#[cfg(unix)]
{
if let Some(parent) = path.parent() {
let dir = if parent.as_os_str().is_empty() {
Path::new(".")
} else {
parent
};
File::open(dir)?.sync_all()?;
}
}
#[cfg(not(unix))]
let _ = path;
Ok(())
}
fn try_lock_exclusive(file: &File, path: &Path) -> HoronResult<()> {
#[cfg(unix)]
{
use std::os::unix::io::AsRawFd;
let rc = unsafe { libc::flock(file.as_raw_fd(), libc::LOCK_EX | libc::LOCK_NB) };
if rc != 0 {
let err = std::io::Error::last_os_error();
if err.kind() == std::io::ErrorKind::WouldBlock {
return Err(HoronError::Locked(path.display().to_string()));
}
return Err(HoronError::Io(err));
}
}
#[cfg(not(unix))]
let _ = (file, path);
Ok(())
}
fn cleanup_orphan_tmp(path: &Path) {
crate::history::cleanup_orphan_segment_tmp(path);
let tmp = path.with_extension("htt.tmp");
if tmp.exists() {
match fs::remove_file(&tmp) {
Ok(()) => log::warn!(
"removed orphaned compaction tempfile {} (crash during a previous compaction)",
tmp.display()
),
Err(e) => log::warn!("could not remove orphaned tempfile {}: {}", tmp.display(), e),
}
}
}
fn hilbert_snapshot_ranks(entries: &[NodeEntry], semantic_dims: usize) -> Vec<u128> {
const MAX_HILBERT_DIMS: usize = 8;
const HILBERT_BITS: u32 = 12;
let user_dims = semantic_dims
.saturating_sub(DIM_USER_DEFINED_START)
.min(MAX_HILBERT_DIMS);
if user_dims == 0 || entries.is_empty() {
return vec![0; entries.len()];
}
let decode = |coords: &[u8]| -> Vec<FixedPoint> {
(0..user_dims)
.map(|d| {
let start = (DIM_USER_DEFINED_START + d) * 16;
let end = start + 16;
if coords.len() >= end {
FixedPoint::from_raw(i128::from_le_bytes(
coords[start..end].try_into().unwrap(),
))
} else {
FixedPoint::from_int(0)
}
})
.collect()
};
let decoded: Vec<Vec<FixedPoint>> =
entries.iter().map(|e| decode(&e.semantic_coords)).collect();
let one = FixedPoint::from_int(1);
let half = one / FixedPoint::from_int(2);
let epsilon = FixedPoint::from_f64(1e-12);
let mut mins = vec![FixedPoint::from_raw(i128::MAX); user_dims];
let mut maxs = vec![FixedPoint::from_raw(i128::MIN); user_dims];
for coords in &decoded {
for i in 0..user_dims {
if coords[i] < mins[i] {
mins[i] = coords[i];
}
if coords[i] > maxs[i] {
maxs[i] = coords[i];
}
}
}
let mapper = HilbertMapper::new(user_dims, HILBERT_BITS);
decoded
.iter()
.map(|coords| {
let norm: Vec<FixedPoint> = (0..user_dims)
.map(|i| {
let range = maxs[i] - mins[i];
if range < epsilon {
half
} else {
(coords[i] - mins[i]) / range
}
})
.collect();
mapper.coords_to_index_fixed(&norm).value()
})
.collect()
}
fn load_snapshot_into_store(
store: &Store,
snap_entries: &[snapshot::NodeEntry],
lazy_geometry: bool,
) -> HoronResult<()> {
for entry in snap_entries {
if lazy_geometry {
Store::put_data_only(store, &entry.key, &entry.data).map_err(HoronError::from)?;
} else {
let child_index = entry
.metadata
.iter()
.find(|(k, _)| k == "_child_index")
.and_then(|(_, v)| v.parse::<u32>().ok());
if let Some(idx) = child_index {
store
.put_positioned(&entry.key, &entry.data, idx)
.map_err(HoronError::from)?;
} else {
Store::put(store, &entry.key, &entry.data).map_err(HoronError::from)?;
}
}
for (mk, mv) in &entry.metadata {
if mk != "key"
&& mk != "size"
&& mk != "created_at"
&& mk != "updated_at"
&& mk != "_child_index"
{
let _ = store.set_meta(&entry.key, mk, mv);
}
}
if !entry.semantic_coords.is_empty() && entry.semantic_coords.iter().any(|&b| b != 0) {
let _ = store.set_semantic(&entry.key, entry.semantic_coords.clone());
}
}
Ok(())
}
fn scan_wal<F>(
file: &mut File,
header: &GeoHeader,
layout: &SemLayout,
wal_base_seq: u32,
wal_entry_count: u32,
apply: F,
) -> HoronResult<(u32, u32, u64)>
where
F: FnMut(&WalEntry) -> HoronResult<()>,
{
scan_wal_raw(
file,
header.wal_compressed(),
header.compression_algo(),
layout,
wal_base_seq,
wal_entry_count,
apply,
)
}
fn scan_wal_raw<F>(
file: &mut File,
wal_compressed: bool,
algo: u8,
layout: &SemLayout,
wal_base_seq: u32,
wal_entry_count: u32,
mut apply: F,
) -> HoronResult<(u32, u32, u64)>
where
F: FnMut(&WalEntry) -> HoronResult<()>,
{
let mut next_seq = wal_base_seq;
let mut valid: u32 = 0;
let mut expected_seq: Option<u32> = None;
let mut valid_end = file.stream_position()?;
let accept = |entry: WalEntry,
expected_seq: &mut Option<u32>,
next_seq: &mut u32,
valid: &mut u32,
apply: &mut F|
-> HoronResult<bool> {
if let Some(exp) = *expected_seq {
if entry.seq != exp {
log::warn!(
"WAL scan stopped: sequence break (expected {}, found {})",
exp, entry.seq
);
return Ok(false);
}
}
*expected_seq = Some(entry.seq + 1);
*next_seq = entry.seq + 1;
apply(&entry)?;
*valid += 1;
Ok(true)
};
if wal_compressed {
'blocks: loop {
match wal::read_wal_block(file, algo) {
Ok(Some((block_data, block_entry_count))) => {
let mut cursor = std::io::Cursor::new(&block_data);
let mut entries = Vec::with_capacity(block_entry_count as usize);
for _ in 0..block_entry_count {
match WalEntry::read_from(&mut cursor, layout) {
Ok(Some(entry)) => entries.push(entry),
Ok(None) => break 'blocks,
Err(e) => {
log::warn!("WAL scan stopped: torn entry in block ({})", e);
break 'blocks;
}
}
}
let first = match entries.first() {
Some(e) => e.seq,
None => break 'blocks,
};
if let Some(exp) = expected_seq {
if first != exp {
log::warn!(
"WAL scan stopped: sequence break at block boundary (expected {}, found {})",
exp, first
);
break 'blocks;
}
}
if entries.iter().enumerate().any(|(i, e)| e.seq != first + i as u32) {
log::warn!("WAL scan stopped: sequence break inside block");
break 'blocks;
}
for entry in entries {
accept(entry, &mut expected_seq, &mut next_seq, &mut valid, &mut apply)?;
}
valid_end = file.stream_position()?;
}
Ok(None) => break,
Err(e) => {
log::warn!("WAL scan stopped: unreadable block ({})", e);
break;
}
}
}
} else {
loop {
match WalEntry::read_from(file, layout) {
Ok(Some(entry)) => {
if !accept(entry, &mut expected_seq, &mut next_seq, &mut valid, &mut apply)? {
break;
}
valid_end = file.stream_position()?;
}
Ok(None) => break,
Err(e) => {
log::warn!("WAL scan stopped: torn tail entry ({})", e);
break;
}
}
}
}
if valid != wal_entry_count {
log::warn!(
"WAL header count {} differs from scan result {} — header is advisory, scan wins",
wal_entry_count, valid
);
}
Ok((next_seq, valid, valid_end))
}
fn promote_from_view(store: &Store, view: &SnapView, key: &str) -> HoronResult<bool> {
if store.exists(key) {
return Ok(true);
}
let Some(&idx) = view.by_key.get(key) else {
return Ok(false);
};
let entry = view.decode(idx)?;
Store::put_data_only(store, &entry.key, &entry.data).map_err(HoronError::from)?;
for (mk, mv) in &entry.metadata {
if mk != "key" && mk != "size" && mk != "created_at" && mk != "updated_at" {
let _ = store.set_meta(&entry.key, mk, mv);
}
}
if !entry.semantic_coords.is_empty() && entry.semantic_coords.iter().any(|&b| b != 0) {
let _ = store.set_semantic(&entry.key, entry.semantic_coords.clone());
}
Ok(true)
}
fn ensure_ancestors(store: &Store, view: &SnapView, key: &str) {
let mut path = String::new();
let segments: Vec<&str> = key.trim_matches('/').split('/').collect();
for seg in &segments[..segments.len().saturating_sub(1)] {
path.push('/');
path.push_str(seg);
if !store.exists(&path) && !view.by_key.contains_key(&path) {
let _ = Store::put_data_only(store, &path, b"");
}
}
}
fn apply_wal_partial(
store: &Store,
view: &SnapView,
tombstones: &mut HashSet<String>,
entry: &WalEntry,
) -> HoronResult<()> {
match &entry.payload {
WalPayload::Insert(node) => {
ensure_ancestors(store, view, &entry.key);
Store::put_data_only(store, &entry.key, &node.data).map_err(HoronError::from)?;
for (mk, mv) in &node.metadata {
let _ = store.set_meta(&entry.key, mk, mv);
}
if !node.semantic_coords.is_empty() && node.semantic_coords.iter().any(|&b| b != 0) {
let _ = store.set_semantic(&entry.key, node.semantic_coords.clone());
}
tombstones.remove(&entry.key);
}
WalPayload::Update { data, metadata } => {
let _ = promote_from_view(store, view, &entry.key);
ensure_ancestors(store, view, &entry.key);
Store::put_data_only(store, &entry.key, data).map_err(HoronError::from)?;
for (mk, mv) in metadata {
let _ = store.set_meta(&entry.key, mk, mv);
}
tombstones.remove(&entry.key);
}
WalPayload::Delete => {
if store.exists(&entry.key) {
let _ = store.remove(&entry.key);
}
tombstones.insert(entry.key.clone());
}
WalPayload::SetMeta { meta_key, meta_value } => {
if promote_from_view(store, view, &entry.key)? {
let _ = store.set_meta(&entry.key, meta_key, meta_value);
} else {
log::warn!("WAL SetMeta for unknown key '{}' — skipped", entry.key);
}
}
WalPayload::SetSemantic { coords } => {
if promote_from_view(store, view, &entry.key)? {
let _ = store.set_semantic(&entry.key, coords.clone());
} else {
log::warn!("WAL SetSemantic for unknown key '{}' — skipped", entry.key);
}
}
WalPayload::Epoch { .. } => {
}
}
Ok(())
}
fn replay_entry(store: &Store, entry: &WalEntry, lazy_geometry: bool) -> HoronResult<()> {
match &entry.payload {
WalPayload::Insert(node) => {
let result = if lazy_geometry {
Store::put_data_only(store, &entry.key, &node.data)
} else {
let child_index = node.metadata.iter()
.find(|(k, _)| k == "_child_index")
.and_then(|(_, v)| v.parse::<u32>().ok());
if let Some(idx) = child_index {
store.put_positioned(&entry.key, &node.data, idx)
} else {
Store::put(store, &entry.key, &node.data)
}
};
match result {
Ok(()) => {}
Err(horon_engine::StoreError::AlreadyExists(_)) => {
}
Err(e) => return Err(HoronError::from(e)),
}
for (mk, mv) in &node.metadata {
if mk != "_child_index" {
let _ = store.set_meta(&entry.key, mk, mv);
}
}
}
WalPayload::Update { data, metadata } => {
let put = if lazy_geometry { Store::put_data_only } else { Store::put };
put(store, &entry.key, data)?;
for (mk, mv) in metadata {
let _ = store.set_meta(&entry.key, mk, mv);
}
}
WalPayload::Delete => {
let _ = store.remove(&entry.key);
}
WalPayload::SetMeta { meta_key, meta_value } => {
let _ = store.set_meta(&entry.key, meta_key, meta_value);
}
WalPayload::SetSemantic { coords } => {
if coords.iter().all(|&b| b == 0) {
let _ = store.set_semantic(&entry.key, Vec::new());
} else {
let _ = store.set_semantic(&entry.key, coords.clone());
}
}
WalPayload::Epoch { .. } => {
}
}
Ok(())
}
#[cfg(test)]
mod tests {
fn fp(vals: &[f64]) -> Vec<g_math::fixed_point::FixedPoint> {
vals.iter().map(|&v| g_math::fixed_point::FixedPoint::from_f64(v)).collect()
}
use super::*;
use tempfile::NamedTempFile;
fn temp_path() -> PathBuf {
NamedTempFile::new().unwrap().into_temp_path().to_path_buf()
}
#[test]
fn test_create_and_reopen() {
let path = temp_path();
{
let gf = Horon::open(&path).unwrap();
gf.put("/hello", b"world").unwrap();
gf.put("/foo/bar", b"baz").unwrap();
gf.set_meta("/hello", "author", "alice").unwrap();
gf.flush().unwrap();
assert_eq!(gf.len(), 3);
}
{
let gf = Horon::open(&path).unwrap();
assert_eq!(gf.get("/hello").unwrap(), b"world");
assert_eq!(gf.get("/foo/bar").unwrap(), b"baz");
assert_eq!(gf.len(), 3);
}
}
#[test]
fn test_wal_recovery() {
let path = temp_path();
{
let gf = Horon::open(&path).unwrap();
gf.put("/a", b"1").unwrap();
gf.put("/b", b"2").unwrap();
gf.remove("/a").unwrap();
gf.flush().unwrap();
}
{
let gf = Horon::open(&path).unwrap();
assert!(!gf.exists("/a"));
assert_eq!(gf.get("/b").unwrap(), b"2");
assert_eq!(gf.len(), 1);
}
}
#[test]
fn test_compact() {
let path = temp_path();
{
let gf = Horon::open(&path).unwrap();
for i in 0..50 {
gf.put(&format!("/node_{}", i), format!("data_{}", i).as_bytes()).unwrap();
}
assert!(gf.wal_len() > 0);
gf.compact().unwrap();
assert_eq!(gf.wal_len(), 0);
assert_eq!(gf.len(), 50);
}
{
let gf = Horon::open(&path).unwrap();
assert_eq!(gf.len(), 50);
assert_eq!(gf.get("/node_42").unwrap(), b"data_42");
}
}
#[test]
fn test_upsert() {
let path = temp_path();
let gf = Horon::open(&path).unwrap();
gf.put("/key", b"v1").unwrap();
gf.put("/key", b"v2").unwrap();
assert_eq!(gf.get("/key").unwrap(), b"v2");
assert_eq!(gf.len(), 1);
}
#[test]
fn test_nearest() {
let path = temp_path();
let gf = Horon::open(&path).unwrap();
gf.put("/a", b"data").unwrap();
gf.put("/b", b"data").unwrap();
let (nearest_path, dist) = gf.nearest(&fp(&[0.0, 0.0, 0.0, 0.0])).unwrap();
assert_eq!(nearest_path, "/");
assert!(dist.to_f64() < 0.1);
}
#[test]
fn test_empty_file_size() {
let path = temp_path();
{
let gf = Horon::open(&path).unwrap();
gf.flush().unwrap();
drop(gf);
}
let size = std::fs::metadata(&path).unwrap().len();
assert_eq!(size as usize, MIN_FILE_SIZE);
}
#[test]
fn test_send_sync() {
fn assert_send<T: Send>() {}
fn assert_sync<T: Sync>() {}
assert_send::<Horon>();
assert_sync::<Horon>();
}
#[test]
fn test_concurrent_put() {
let path = temp_path();
let gf = Arc::new(Horon::open(&path).unwrap());
let mut handles = vec![];
for t in 0..4 {
let gf = Arc::clone(&gf);
handles.push(std::thread::spawn(move || {
for i in 0..5 {
gf.put(&format!("/t{}/n{}", t, i), b"data").unwrap();
}
}));
}
for h in handles {
h.join().unwrap();
}
for t in 0..4 {
for i in 0..5 {
assert!(gf.exists(&format!("/t{}/n{}", t, i)));
}
}
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum RefreshOutcome {
UpToDate,
Applied(usize),
Reloaded,
}
pub struct HoronReader {
store: Store,
path: PathBuf,
header: GeoHeader,
layout: SemLayout,
next_seq: u32,
}
impl HoronReader {
pub fn open<P: AsRef<Path>>(path: P) -> HoronResult<Self> {
let path = path.as_ref().to_path_buf();
let (store, header, layout, next_seq) = Self::load(&path)?;
Ok(Self { store, path, header, layout, next_seq })
}
fn load(path: &Path) -> HoronResult<(Store, GeoHeader, SemLayout, u32)> {
let mut file = File::open(path)?;
let mut header_bytes = [0u8; HEADER_SIZE];
std::io::Read::read_exact(&mut file, &mut header_bytes)?;
let header = GeoHeader::from_bytes(&header_bytes)?;
let layout = SemLayout::from_header(&header);
let compressed = header.compression_enabled();
let snapshot_has_crc = header.version >= 2;
let meaning_addressed = header.flags & FLAG_MEANING_ADDRESSED != 0;
if meaning_addressed {
let bounds_len = (header.semantic_dims as usize)
.saturating_sub(DIM_USER_DEFINED_START)
* 16;
file.seek(SeekFrom::Current(bounds_len as i64))?;
}
let mut snap_entries =
snapshot::read_snapshot(&mut file, compressed, &layout, snapshot_has_crc)?;
if meaning_addressed {
snap_entries.sort_by(|a, b| {
let da = a.key.matches('/').count();
let db = b.key.matches('/').count();
da.cmp(&db).then(a.key.cmp(&b.key))
});
}
let (wal_entry_count, wal_base_seq) = wal::read_wal_header(&mut file)?;
let store = Store::with_config(
StoreConfig::new().tau(g_math::fixed_point::FixedPoint::from_raw(header.tau_raw)),
);
load_snapshot_into_store(&store, &snap_entries, false)?;
let (next_seq, _valid, _end) = scan_wal(
&mut file,
&header,
&layout,
wal_base_seq,
wal_entry_count,
|entry| replay_entry(&store, entry, false),
)?;
Ok((store, header, layout, next_seq))
}
pub fn refresh(&mut self) -> HoronResult<RefreshOutcome> {
let mut file = File::open(&self.path)?;
let mut header_bytes = [0u8; HEADER_SIZE];
std::io::Read::read_exact(&mut file, &mut header_bytes)?;
let header = GeoHeader::from_bytes(&header_bytes)?;
let compressed = header.compression_enabled();
let meaning_addressed = header.flags & FLAG_MEANING_ADDRESSED != 0;
if meaning_addressed {
let bounds_len = (header.semantic_dims as usize)
.saturating_sub(DIM_USER_DEFINED_START)
* 16;
file.seek(SeekFrom::Current(bounds_len as i64))?;
}
let mut buf4 = [0u8; 4];
std::io::Read::read_exact(&mut file, &mut buf4)?;
let snap_byte_len = u32::from_le_bytes(buf4) as u64;
std::io::Read::read_exact(&mut file, &mut buf4)?;
let node_count = u32::from_le_bytes(buf4);
if node_count > 0 && compressed {
std::io::Read::read_exact(&mut file, &mut buf4)?;
let comp_len = u32::from_le_bytes(buf4) as u64;
file.seek(SeekFrom::Current(comp_len as i64))?;
} else {
file.seek(SeekFrom::Current(snap_byte_len as i64))?;
}
if header.version >= 2 {
file.seek(SeekFrom::Current(4))?; }
let (wal_entry_count, wal_base_seq) = wal::read_wal_header(&mut file)?;
if wal_base_seq > self.next_seq {
let (store, header, layout, next_seq) = Self::load(&self.path)?;
self.store = store;
self.header = header;
self.layout = layout;
self.next_seq = next_seq;
return Ok(RefreshOutcome::Reloaded);
}
let from = self.next_seq;
let mut applied = 0usize;
let (next_seq, _valid, _end) = scan_wal(
&mut file,
&self.header,
&self.layout,
wal_base_seq,
wal_entry_count,
|entry| {
if entry.seq >= from {
applied += 1;
replay_entry(&self.store, entry, false)
} else {
Ok(())
}
},
)?;
self.next_seq = next_seq;
Ok(if applied == 0 {
RefreshOutcome::UpToDate
} else {
RefreshOutcome::Applied(applied)
})
}
pub fn next_seq(&self) -> u32 {
self.next_seq
}
pub fn get(&self, key: &str) -> HoronResult<Vec<u8>> {
Ok(self.store.get(key)?)
}
pub fn exists(&self, key: &str) -> bool {
self.store.exists(key)
}
pub fn get_semantic(&self, key: &str) -> HoronResult<Vec<u8>> {
Ok(self.store.get_semantic(key)?)
}
pub fn children(&self, path: &str) -> HoronResult<Vec<String>> {
Ok(self.store.children(path)?)
}
pub fn list(&self, prefix: &str) -> HoronResult<Vec<String>> {
Ok(self.store.list(prefix)?)
}
pub fn nearest(&self, coords: &[FixedPoint]) -> HoronResult<(String, FixedPoint)> {
Ok(self.store.nearest(coords)?)
}
pub fn neighbors(&self, path: &str, k: usize) -> HoronResult<Vec<String>> {
Ok(self.store.neighbors(path, k)?)
}
pub fn len(&self) -> usize {
self.store.len()
}
pub fn is_empty(&self) -> bool {
self.store.is_empty()
}
}