use byteorder::{LittleEndian, ReadBytesExt, WriteBytesExt};
use memmap2::{MmapMut, MmapOptions};
use parking_lot::{Mutex, MutexGuard};
use std::fs::OpenOptions;
use std::io::Cursor;
use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::Arc;
pub const PAGE_SIZE: usize = 16 * 1024;
pub const HEADER_SIZE: usize = 64;
pub const MAGIC: u64 = 0x4D31_4353_5031_4E45;
pub const N_LAYERS: usize = 12;
pub const ATTENTION_LAYERS: usize = 7;
pub const SUPERBLOCK_VERSION: u32 = 2;
pub const SUPERBLOCK_BYTES: usize = 56;
pub const SEQ_SLOT: usize = 56;
const _: () = assert!(
SEQ_SLOT.is_multiple_of(8),
"the counter slot must stay 8-byte aligned"
);
const _: () = assert!(
SEQ_SLOT >= SUPERBLOCK_BYTES,
"the payload must end before the counter"
);
pub const SNAPSHOT_PREFIX_BYTES: usize = 128;
pub const SNAPSHOT_MAX_RETRIES: u32 = 65_536;
pub const SNAPSHOT_MAX_WAIT_MS: u64 = 50;
pub const PUBLISH_LOCK_FILE_NAME: &str = "spine.publish";
pub const WRITER_LOCK_FILE_NAME: &str = "spine.lock";
const _: () = assert!(SEQ_SLOT + 8 <= PAGE_SIZE);
const _: () = assert!(SNAPSHOT_PREFIX_BYTES >= StreamHeader::HEADER_BYTES);
const _: () = assert!(SNAPSHOT_PREFIX_BYTES >= HebbianHeader::HEADER_BYTES);
const _: () = assert!(SNAPSHOT_PREFIX_BYTES >= AttentionHeader::HEADER_BYTES);
const _: () = assert!(SNAPSHOT_PREFIX_BYTES >= EmotionHeader::HEADER_BYTES);
const _: () = assert!(SNAPSHOT_PREFIX_BYTES >= ResonanceHeader::HEADER_BYTES);
const _: () = assert!(SNAPSHOT_PREFIX_BYTES >= GraphHeader::HEADER_BYTES);
const _: () = assert!(SNAPSHOT_PREFIX_BYTES >= CacheHeader::HEADER_BYTES);
const _: () = assert!(SNAPSHOT_PREFIX_BYTES >= ArchetypeHeader::HEADER_BYTES);
const _: () = assert!(SNAPSHOT_PREFIX_BYTES >= MirrorHeader::HEADER_BYTES);
const _: () = assert!(SNAPSHOT_PREFIX_BYTES >= MerkleHeader::HEADER_BYTES);
const _: () = assert!(SNAPSHOT_PREFIX_BYTES >= MetaHeader::HEADER_BYTES);
const _: () = assert!(SNAPSHOT_PREFIX_BYTES <= PAGE_SIZE);
const _: () = assert!(SUPERBLOCK_BYTES + 8 <= PAGE_SIZE);
const _: () = assert!(HEADER_SIZE == 64);
const _: () = assert!(11 * PAGE_SIZE + SNAPSHOT_PREFIX_BYTES <= 12 * PAGE_SIZE);
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[repr(u32)]
pub enum Slot {
Superblock = 0,
Stream = 1,
Hebbian = 2,
Attention = 3,
Emotion = 4,
Resonance = 5,
Graph = 6,
Cache = 7,
Archetype = 8,
Mirror = 9,
Merkle = 10,
Meta = 11,
WalStart = 12,
}
impl Slot {
pub fn offset(&self) -> usize {
*self as usize * PAGE_SIZE
}
pub fn from_u32(v: u32) -> Option<Slot> {
Some(match v {
0 => Slot::Superblock,
1 => Slot::Stream,
2 => Slot::Hebbian,
3 => Slot::Attention,
4 => Slot::Emotion,
5 => Slot::Resonance,
6 => Slot::Graph,
7 => Slot::Cache,
8 => Slot::Archetype,
9 => Slot::Mirror,
10 => Slot::Merkle,
11 => Slot::Meta,
12 => Slot::WalStart,
_ => return None,
})
}
}
#[derive(Debug, Clone, Copy)]
#[repr(C, align(64))]
pub struct StreamHeader {
pub cycle: u64,
pub surprise_level: f32,
pub curiosity_level: f32,
pub predicted_query_hash: u64,
pub predicted_confidence: f32,
pub last_query_hash: u64,
pub last_query_ms: u64,
pub hebbian_total_energy: f64,
pub updated_ms: u64,
pub _pad: [u8; PAGE_SIZE - 60],
}
impl StreamHeader {
pub const HEADER_BYTES: usize = 60;
pub fn to_bytes(&self) -> [u8; PAGE_SIZE] {
let mut buf = [0u8; PAGE_SIZE];
let mut c = Cursor::new(&mut buf[..Self::HEADER_BYTES]);
c.write_u64::<LittleEndian>(self.cycle).unwrap();
c.write_f32::<LittleEndian>(self.surprise_level).unwrap();
c.write_f32::<LittleEndian>(self.curiosity_level).unwrap();
c.write_u64::<LittleEndian>(self.predicted_query_hash)
.unwrap();
c.write_f32::<LittleEndian>(self.predicted_confidence)
.unwrap();
c.write_u64::<LittleEndian>(self.last_query_hash).unwrap();
c.write_u64::<LittleEndian>(self.last_query_ms).unwrap();
c.write_f64::<LittleEndian>(self.hebbian_total_energy)
.unwrap();
c.write_u64::<LittleEndian>(self.updated_ms).unwrap();
buf
}
pub fn from_bytes(buf: &[u8]) -> Self {
let mut c = Cursor::new(buf);
Self {
cycle: c.read_u64::<LittleEndian>().unwrap_or(0),
surprise_level: c.read_f32::<LittleEndian>().unwrap_or(0.0),
curiosity_level: c.read_f32::<LittleEndian>().unwrap_or(0.0),
predicted_query_hash: c.read_u64::<LittleEndian>().unwrap_or(0),
predicted_confidence: c.read_f32::<LittleEndian>().unwrap_or(0.0),
last_query_hash: c.read_u64::<LittleEndian>().unwrap_or(0),
last_query_ms: c.read_u64::<LittleEndian>().unwrap_or(0),
hebbian_total_energy: c.read_f64::<LittleEndian>().unwrap_or(0.0),
updated_ms: c.read_u64::<LittleEndian>().unwrap_or(0),
_pad: [0; PAGE_SIZE - Self::HEADER_BYTES],
}
}
}
impl Default for StreamHeader {
fn default() -> Self {
Self {
cycle: 0,
surprise_level: 0.0,
curiosity_level: 0.0,
predicted_query_hash: 0,
predicted_confidence: 0.0,
last_query_hash: 0,
last_query_ms: 0,
hebbian_total_energy: 0.0,
updated_ms: 0,
_pad: [0; PAGE_SIZE - Self::HEADER_BYTES],
}
}
}
#[derive(Debug, Clone, Copy)]
#[repr(C, align(64))]
pub struct HebbianHeader {
pub block_count: u32,
pub activation_count: u32,
pub coactivation_count: u32,
pub fingerprint_count: u32,
pub delta_tail_offset: u64,
pub decay_event_count: u64,
pub last_activation_ms: u64,
pub _pad: [u8; PAGE_SIZE - 40],
}
impl HebbianHeader {
pub const HEADER_BYTES: usize = 40;
pub fn to_bytes(&self) -> [u8; PAGE_SIZE] {
let mut buf = [0u8; PAGE_SIZE];
let mut c = Cursor::new(&mut buf[..Self::HEADER_BYTES]);
c.write_u32::<LittleEndian>(self.block_count).unwrap();
c.write_u32::<LittleEndian>(self.activation_count).unwrap();
c.write_u32::<LittleEndian>(self.coactivation_count)
.unwrap();
c.write_u32::<LittleEndian>(self.fingerprint_count).unwrap();
c.write_u64::<LittleEndian>(self.delta_tail_offset).unwrap();
c.write_u64::<LittleEndian>(self.decay_event_count).unwrap();
c.write_u64::<LittleEndian>(self.last_activation_ms)
.unwrap();
buf
}
pub fn from_bytes(buf: &[u8]) -> Self {
let mut c = Cursor::new(buf);
Self {
block_count: c.read_u32::<LittleEndian>().unwrap_or(0),
activation_count: c.read_u32::<LittleEndian>().unwrap_or(0),
coactivation_count: c.read_u32::<LittleEndian>().unwrap_or(0),
fingerprint_count: c.read_u32::<LittleEndian>().unwrap_or(0),
delta_tail_offset: c.read_u64::<LittleEndian>().unwrap_or(0),
decay_event_count: c.read_u64::<LittleEndian>().unwrap_or(0),
last_activation_ms: c.read_u64::<LittleEndian>().unwrap_or(0),
_pad: [0; PAGE_SIZE - Self::HEADER_BYTES],
}
}
}
impl Default for HebbianHeader {
fn default() -> Self {
Self {
block_count: 0,
activation_count: 0,
coactivation_count: 0,
fingerprint_count: 0,
delta_tail_offset: 0,
decay_event_count: 0,
last_activation_ms: 0,
_pad: [0; PAGE_SIZE - Self::HEADER_BYTES],
}
}
}
#[derive(Debug, Clone, Copy)]
#[repr(C, align(64))]
pub struct AttentionHeader {
pub learned_weights: [f32; ATTENTION_LAYERS],
pub last_recall_ms: u64,
pub total_recalls: u32,
pub quality_signal: f32,
pub _pad: [u8; PAGE_SIZE - (4 * ATTENTION_LAYERS + 16)],
}
impl AttentionHeader {
pub const HEADER_BYTES: usize = 4 * ATTENTION_LAYERS + 16;
pub fn to_bytes(&self) -> [u8; PAGE_SIZE] {
let mut buf = [0u8; PAGE_SIZE];
let mut c = Cursor::new(&mut buf[..Self::HEADER_BYTES]);
for w in &self.learned_weights {
c.write_f32::<LittleEndian>(*w).unwrap();
}
c.write_u64::<LittleEndian>(self.last_recall_ms).unwrap();
c.write_u32::<LittleEndian>(self.total_recalls).unwrap();
c.write_f32::<LittleEndian>(self.quality_signal).unwrap();
buf
}
pub fn from_bytes(buf: &[u8]) -> Self {
let mut c = Cursor::new(buf);
let mut lw = [0.0f32; ATTENTION_LAYERS];
for w in lw.iter_mut() {
*w = c.read_f32::<LittleEndian>().unwrap_or(0.0);
}
Self {
learned_weights: lw,
last_recall_ms: c.read_u64::<LittleEndian>().unwrap_or(0),
total_recalls: c.read_u32::<LittleEndian>().unwrap_or(0),
quality_signal: c.read_f32::<LittleEndian>().unwrap_or(0.0),
_pad: [0; PAGE_SIZE - Self::HEADER_BYTES],
}
}
}
impl Default for AttentionHeader {
fn default() -> Self {
Self {
learned_weights: [1.0 / ATTENTION_LAYERS as f32; ATTENTION_LAYERS],
last_recall_ms: 0,
total_recalls: 0,
quality_signal: 0.0,
_pad: [0; PAGE_SIZE - Self::HEADER_BYTES],
}
}
}
#[derive(Debug, Clone, Copy)]
#[repr(C, align(64))]
pub struct EmotionHeader {
pub current: [f32; 21],
pub head: u32,
pub count: u32,
pub last_update: u64,
pub total_updates: u64,
pub _pad: [u8; PAGE_SIZE - 108],
}
impl EmotionHeader {
pub const HEADER_BYTES: usize = 4 * 21 + 24;
pub fn to_bytes(&self) -> [u8; PAGE_SIZE] {
let mut buf = [0u8; PAGE_SIZE];
let mut c = Cursor::new(&mut buf[..Self::HEADER_BYTES]);
for v in &self.current {
c.write_f32::<LittleEndian>(*v).unwrap();
}
c.write_u32::<LittleEndian>(self.head).unwrap();
c.write_u32::<LittleEndian>(self.count).unwrap();
c.write_u64::<LittleEndian>(self.last_update).unwrap();
c.write_u64::<LittleEndian>(self.total_updates).unwrap();
buf
}
pub fn from_bytes(buf: &[u8]) -> Self {
let mut c = Cursor::new(buf);
let mut cur = [0.0f32; 21];
for v in cur.iter_mut() {
*v = c.read_f32::<LittleEndian>().unwrap_or(0.0);
}
Self {
current: cur,
head: c.read_u32::<LittleEndian>().unwrap_or(0),
count: c.read_u32::<LittleEndian>().unwrap_or(0),
last_update: c.read_u64::<LittleEndian>().unwrap_or(0),
total_updates: c.read_u64::<LittleEndian>().unwrap_or(0),
_pad: [0; PAGE_SIZE - Self::HEADER_BYTES],
}
}
}
impl Default for EmotionHeader {
fn default() -> Self {
Self {
current: [0.0; 21],
head: 0,
count: 0,
last_update: 0,
total_updates: 0,
_pad: [0; PAGE_SIZE - Self::HEADER_BYTES],
}
}
}
#[derive(Debug, Clone, Copy)]
#[repr(C, align(64))]
pub struct ResonanceHeader {
pub instance_id: u64,
pub field_cells: u32,
pub outgoing_count: u32,
pub incoming_count: u32,
pub pulses_emitted: u64,
pub pulses_received: u64,
pub last_decay_ms: u64,
pub _pad: [u8; PAGE_SIZE - 44],
}
impl ResonanceHeader {
pub const HEADER_BYTES: usize = 44;
pub fn to_bytes(&self) -> [u8; PAGE_SIZE] {
let mut buf = [0u8; PAGE_SIZE];
let mut c = Cursor::new(&mut buf[..Self::HEADER_BYTES]);
c.write_u64::<LittleEndian>(self.instance_id).unwrap();
c.write_u32::<LittleEndian>(self.field_cells).unwrap();
c.write_u32::<LittleEndian>(self.outgoing_count).unwrap();
c.write_u32::<LittleEndian>(self.incoming_count).unwrap();
c.write_u64::<LittleEndian>(self.pulses_emitted).unwrap();
c.write_u64::<LittleEndian>(self.pulses_received).unwrap();
c.write_u64::<LittleEndian>(self.last_decay_ms).unwrap();
buf
}
pub fn from_bytes(buf: &[u8]) -> Self {
let mut c = Cursor::new(buf);
Self {
instance_id: c.read_u64::<LittleEndian>().unwrap_or(0),
field_cells: c.read_u32::<LittleEndian>().unwrap_or(0),
outgoing_count: c.read_u32::<LittleEndian>().unwrap_or(0),
incoming_count: c.read_u32::<LittleEndian>().unwrap_or(0),
pulses_emitted: c.read_u64::<LittleEndian>().unwrap_or(0),
pulses_received: c.read_u64::<LittleEndian>().unwrap_or(0),
last_decay_ms: c.read_u64::<LittleEndian>().unwrap_or(0),
_pad: [0; PAGE_SIZE - Self::HEADER_BYTES],
}
}
}
impl Default for ResonanceHeader {
fn default() -> Self {
Self {
instance_id: 0,
field_cells: 0,
outgoing_count: 0,
incoming_count: 0,
pulses_emitted: 0,
pulses_received: 0,
last_decay_ms: 0,
_pad: [0; PAGE_SIZE - Self::HEADER_BYTES],
}
}
}
#[derive(Debug, Clone, Copy)]
#[repr(C, align(64))]
pub struct GraphHeader {
pub node_count: u32,
pub edge_count: u32,
pub pattern_count: u32,
pub current_session_id: u32,
pub next_pattern_id: u32,
pub last_node_ts: u64,
pub _pad: [u8; PAGE_SIZE - 28],
}
impl GraphHeader {
pub const HEADER_BYTES: usize = 28;
pub fn to_bytes(&self) -> [u8; PAGE_SIZE] {
let mut buf = [0u8; PAGE_SIZE];
let mut c = Cursor::new(&mut buf[..Self::HEADER_BYTES]);
c.write_u32::<LittleEndian>(self.node_count).unwrap();
c.write_u32::<LittleEndian>(self.edge_count).unwrap();
c.write_u32::<LittleEndian>(self.pattern_count).unwrap();
c.write_u32::<LittleEndian>(self.current_session_id)
.unwrap();
c.write_u32::<LittleEndian>(self.next_pattern_id).unwrap();
c.write_u64::<LittleEndian>(self.last_node_ts).unwrap();
buf
}
pub fn from_bytes(buf: &[u8]) -> Self {
let mut c = Cursor::new(buf);
Self {
node_count: c.read_u32::<LittleEndian>().unwrap_or(0),
edge_count: c.read_u32::<LittleEndian>().unwrap_or(0),
pattern_count: c.read_u32::<LittleEndian>().unwrap_or(0),
current_session_id: c.read_u32::<LittleEndian>().unwrap_or(0),
next_pattern_id: c.read_u32::<LittleEndian>().unwrap_or(1),
last_node_ts: c.read_u64::<LittleEndian>().unwrap_or(0),
_pad: [0; PAGE_SIZE - Self::HEADER_BYTES],
}
}
}
impl Default for GraphHeader {
fn default() -> Self {
Self {
node_count: 0,
edge_count: 0,
pattern_count: 0,
current_session_id: 0,
next_pattern_id: 1,
last_node_ts: 0,
_pad: [0; PAGE_SIZE - Self::HEADER_BYTES],
}
}
}
#[derive(Debug, Clone, Copy)]
#[repr(C, align(64))]
pub struct CacheHeader {
pub total_predictions: u32,
pub total_hits: u32,
pub total_misses: u32,
pub total_partial_hits: u32,
pub current_predictions: u32,
pub avg_confidence: f32,
pub last_predict_ms: u64,
pub _pad: [u8; PAGE_SIZE - 32],
}
impl CacheHeader {
pub const HEADER_BYTES: usize = 32;
pub fn to_bytes(&self) -> [u8; PAGE_SIZE] {
let mut buf = [0u8; PAGE_SIZE];
let mut c = Cursor::new(&mut buf[..Self::HEADER_BYTES]);
c.write_u32::<LittleEndian>(self.total_predictions).unwrap();
c.write_u32::<LittleEndian>(self.total_hits).unwrap();
c.write_u32::<LittleEndian>(self.total_misses).unwrap();
c.write_u32::<LittleEndian>(self.total_partial_hits)
.unwrap();
c.write_u32::<LittleEndian>(self.current_predictions)
.unwrap();
c.write_f32::<LittleEndian>(self.avg_confidence).unwrap();
c.write_u64::<LittleEndian>(self.last_predict_ms).unwrap();
buf
}
pub fn from_bytes(buf: &[u8]) -> Self {
let mut c = Cursor::new(buf);
Self {
total_predictions: c.read_u32::<LittleEndian>().unwrap_or(0),
total_hits: c.read_u32::<LittleEndian>().unwrap_or(0),
total_misses: c.read_u32::<LittleEndian>().unwrap_or(0),
total_partial_hits: c.read_u32::<LittleEndian>().unwrap_or(0),
current_predictions: c.read_u32::<LittleEndian>().unwrap_or(0),
avg_confidence: c.read_f32::<LittleEndian>().unwrap_or(0.0),
last_predict_ms: c.read_u64::<LittleEndian>().unwrap_or(0),
_pad: [0; PAGE_SIZE - Self::HEADER_BYTES],
}
}
pub fn hit_rate(&self) -> f32 {
let total = self.total_hits + self.total_misses + self.total_partial_hits;
if total == 0 {
return 0.0;
}
self.total_hits as f32 / total as f32
}
}
impl Default for CacheHeader {
fn default() -> Self {
Self {
total_predictions: 0,
total_hits: 0,
total_misses: 0,
total_partial_hits: 0,
current_predictions: 0,
avg_confidence: 0.0,
last_predict_ms: 0,
_pad: [0; PAGE_SIZE - Self::HEADER_BYTES],
}
}
}
#[derive(Debug, Clone, Copy)]
#[repr(C, align(64))]
pub struct ArchetypeHeader {
pub active_count: u32,
pub next_id: u32,
pub emergence_events: u64,
pub decay_events: u64,
pub last_emergence_ms: u64,
pub _pad: [u8; PAGE_SIZE - 32],
}
impl ArchetypeHeader {
pub const HEADER_BYTES: usize = 32;
pub fn to_bytes(&self) -> [u8; PAGE_SIZE] {
let mut buf = [0u8; PAGE_SIZE];
let mut c = Cursor::new(&mut buf[..Self::HEADER_BYTES]);
c.write_u32::<LittleEndian>(self.active_count).unwrap();
c.write_u32::<LittleEndian>(self.next_id).unwrap();
c.write_u64::<LittleEndian>(self.emergence_events).unwrap();
c.write_u64::<LittleEndian>(self.decay_events).unwrap();
c.write_u64::<LittleEndian>(self.last_emergence_ms).unwrap();
buf
}
pub fn from_bytes(buf: &[u8]) -> Self {
let mut c = Cursor::new(buf);
Self {
active_count: c.read_u32::<LittleEndian>().unwrap_or(0),
next_id: c.read_u32::<LittleEndian>().unwrap_or(1),
emergence_events: c.read_u64::<LittleEndian>().unwrap_or(0),
decay_events: c.read_u64::<LittleEndian>().unwrap_or(0),
last_emergence_ms: c.read_u64::<LittleEndian>().unwrap_or(0),
_pad: [0; PAGE_SIZE - Self::HEADER_BYTES],
}
}
}
impl Default for ArchetypeHeader {
fn default() -> Self {
Self {
active_count: 0,
next_id: 1,
emergence_events: 0,
decay_events: 0,
last_emergence_ms: 0,
_pad: [0; PAGE_SIZE - Self::HEADER_BYTES],
}
}
}
#[derive(Debug, Clone, Copy)]
#[repr(C, align(64))]
pub struct MirrorHeader {
pub echo_count: u32,
pub block_resonance_count: u32,
pub strongest_similarity: f32,
pub total_echoes: u64,
pub last_mirror_ms: u64,
pub _pad: [u8; PAGE_SIZE - 28],
}
impl MirrorHeader {
pub const HEADER_BYTES: usize = 28;
pub fn to_bytes(&self) -> [u8; PAGE_SIZE] {
let mut buf = [0u8; PAGE_SIZE];
let mut c = Cursor::new(&mut buf[..Self::HEADER_BYTES]);
c.write_u32::<LittleEndian>(self.echo_count).unwrap();
c.write_u32::<LittleEndian>(self.block_resonance_count)
.unwrap();
c.write_f32::<LittleEndian>(self.strongest_similarity)
.unwrap();
c.write_u64::<LittleEndian>(self.total_echoes).unwrap();
c.write_u64::<LittleEndian>(self.last_mirror_ms).unwrap();
buf
}
pub fn from_bytes(buf: &[u8]) -> Self {
let mut c = Cursor::new(buf);
Self {
echo_count: c.read_u32::<LittleEndian>().unwrap_or(0),
block_resonance_count: c.read_u32::<LittleEndian>().unwrap_or(0),
strongest_similarity: c.read_f32::<LittleEndian>().unwrap_or(0.0),
total_echoes: c.read_u64::<LittleEndian>().unwrap_or(0),
last_mirror_ms: c.read_u64::<LittleEndian>().unwrap_or(0),
_pad: [0; PAGE_SIZE - Self::HEADER_BYTES],
}
}
}
impl Default for MirrorHeader {
fn default() -> Self {
Self {
echo_count: 0,
block_resonance_count: 0,
strongest_similarity: 0.0,
total_echoes: 0,
last_mirror_ms: 0,
_pad: [0; PAGE_SIZE - Self::HEADER_BYTES],
}
}
}
#[derive(Debug, Clone, Copy)]
#[repr(C, align(64))]
pub struct MerkleHeader {
pub root_hash: u64,
pub chain_height: u64,
pub block_count: u32,
pub evidence_audit_entries: u32,
pub last_anchor_ms: u64,
pub _pad: [u8; PAGE_SIZE - 32],
}
impl MerkleHeader {
pub const HEADER_BYTES: usize = 32;
pub fn to_bytes(&self) -> [u8; PAGE_SIZE] {
let mut buf = [0u8; PAGE_SIZE];
let mut c = Cursor::new(&mut buf[..Self::HEADER_BYTES]);
c.write_u64::<LittleEndian>(self.root_hash).unwrap();
c.write_u64::<LittleEndian>(self.chain_height).unwrap();
c.write_u32::<LittleEndian>(self.block_count).unwrap();
c.write_u32::<LittleEndian>(self.evidence_audit_entries)
.unwrap();
c.write_u64::<LittleEndian>(self.last_anchor_ms).unwrap();
buf
}
pub fn from_bytes(buf: &[u8]) -> Self {
let mut c = Cursor::new(buf);
Self {
root_hash: c.read_u64::<LittleEndian>().unwrap_or(0),
chain_height: c.read_u64::<LittleEndian>().unwrap_or(0),
block_count: c.read_u32::<LittleEndian>().unwrap_or(0),
evidence_audit_entries: c.read_u32::<LittleEndian>().unwrap_or(0),
last_anchor_ms: c.read_u64::<LittleEndian>().unwrap_or(0),
_pad: [0; PAGE_SIZE - Self::HEADER_BYTES],
}
}
}
impl Default for MerkleHeader {
fn default() -> Self {
Self {
root_hash: 0,
chain_height: 0,
block_count: 0,
evidence_audit_entries: 0,
last_anchor_ms: 0,
_pad: [0; PAGE_SIZE - Self::HEADER_BYTES],
}
}
}
#[derive(Debug, Clone, Copy)]
#[repr(C, align(64))]
pub struct MetaHeader {
pub block_count: u32,
pub append_log_len: u32,
pub dream_cycle_count: u32,
pub reps_served_count: u32,
pub last_dream_ms: u64,
pub promotion_count: u64,
pub _pad: [u8; PAGE_SIZE - 32],
}
impl MetaHeader {
pub const HEADER_BYTES: usize = 32;
pub fn to_bytes(&self) -> [u8; PAGE_SIZE] {
let mut buf = [0u8; PAGE_SIZE];
let mut c = Cursor::new(&mut buf[..Self::HEADER_BYTES]);
c.write_u32::<LittleEndian>(self.block_count).unwrap();
c.write_u32::<LittleEndian>(self.append_log_len).unwrap();
c.write_u32::<LittleEndian>(self.dream_cycle_count).unwrap();
c.write_u32::<LittleEndian>(self.reps_served_count).unwrap();
c.write_u64::<LittleEndian>(self.last_dream_ms).unwrap();
c.write_u64::<LittleEndian>(self.promotion_count).unwrap();
buf
}
pub fn from_bytes(buf: &[u8]) -> Self {
let mut c = Cursor::new(buf);
Self {
block_count: c.read_u32::<LittleEndian>().unwrap_or(0),
append_log_len: c.read_u32::<LittleEndian>().unwrap_or(0),
dream_cycle_count: c.read_u32::<LittleEndian>().unwrap_or(0),
reps_served_count: c.read_u32::<LittleEndian>().unwrap_or(0),
last_dream_ms: c.read_u64::<LittleEndian>().unwrap_or(0),
promotion_count: c.read_u64::<LittleEndian>().unwrap_or(0),
_pad: [0; PAGE_SIZE - Self::HEADER_BYTES],
}
}
}
impl Default for MetaHeader {
fn default() -> Self {
Self {
block_count: 0,
append_log_len: 0,
dream_cycle_count: 0,
reps_served_count: 0,
last_dream_ms: 0,
promotion_count: 0,
_pad: [0; PAGE_SIZE - Self::HEADER_BYTES],
}
}
}
#[derive(Debug, Clone, Copy)]
pub struct Superblock {
pub magic: u64,
pub version: u32,
pub page_size: u32,
pub wal_start_page: u32,
pub wal_head_page: u32,
pub tick_seq: u64,
pub epoch_ms: i64,
}
impl Default for Superblock {
fn default() -> Self {
Self::ZEROED
}
}
impl Superblock {
const ZEROED: Self = Self {
magic: 0,
version: SUPERBLOCK_VERSION,
page_size: 0,
wal_start_page: 0,
wal_head_page: 0,
tick_seq: 0,
epoch_ms: 0,
};
pub fn new() -> Self {
Self {
magic: MAGIC,
version: SUPERBLOCK_VERSION,
page_size: PAGE_SIZE as u32,
wal_start_page: Slot::WalStart as u32,
wal_head_page: Slot::WalStart as u32,
tick_seq: 0,
epoch_ms: now_ms(),
}
}
pub fn to_payload(&self) -> [u8; SUPERBLOCK_BYTES] {
let mut buf = [0u8; SUPERBLOCK_BYTES];
let mut c = Cursor::new(&mut buf[..]);
c.write_u64::<LittleEndian>(self.magic).unwrap();
c.write_u32::<LittleEndian>(self.version).unwrap();
c.write_u32::<LittleEndian>(self.page_size).unwrap();
c.write_u32::<LittleEndian>(self.wal_start_page).unwrap();
c.write_u32::<LittleEndian>(self.wal_head_page).unwrap();
c.write_u64::<LittleEndian>(self.tick_seq).unwrap();
c.write_i64::<LittleEndian>(self.epoch_ms).unwrap();
buf
}
pub fn from_payload(buf: &[u8]) -> Option<Self> {
let mut c = Cursor::new(buf);
let magic = c.read_u64::<LittleEndian>().ok()?;
if magic != MAGIC {
return None;
}
let version = c.read_u32::<LittleEndian>().ok()?;
if version != SUPERBLOCK_VERSION {
return None;
}
Some(Self {
magic,
version,
page_size: c.read_u32::<LittleEndian>().ok()?,
wal_start_page: c.read_u32::<LittleEndian>().ok()?,
wal_head_page: c.read_u32::<LittleEndian>().ok()?,
tick_seq: c.read_u64::<LittleEndian>().ok()?,
epoch_ms: c.read_i64::<LittleEndian>().ok()?,
})
}
fn from_v1_payload(buf: &[u8]) -> Option<Self> {
let mut c = Cursor::new(buf);
let magic = c.read_u64::<LittleEndian>().ok()?;
if magic != MAGIC {
return None;
}
let version = c.read_u32::<LittleEndian>().ok()?;
if version != 1 {
return None;
}
Some(Self {
magic,
version,
page_size: c.read_u32::<LittleEndian>().ok()?,
wal_start_page: c.read_u32::<LittleEndian>().ok()?,
wal_head_page: c.read_u32::<LittleEndian>().ok()?,
tick_seq: 0,
epoch_ms: c.read_i64::<LittleEndian>().ok()?,
})
}
}
fn now_ms() -> i64 {
std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map(|d| d.as_millis() as i64)
.unwrap_or(0)
}
#[derive(Debug, Clone, Copy)]
pub struct PublishSnapshot {
pub stream: StreamHeader,
pub tick_seq: u64,
pub epoch_ms: i64,
pub sequence: u64,
}
#[derive(Debug, Clone, Copy)]
pub struct StoreSnapshot {
pub stream: StreamHeader,
pub hebbian: HebbianHeader,
pub attention: AttentionHeader,
pub emotion: EmotionHeader,
pub resonance: ResonanceHeader,
pub graph: GraphHeader,
pub cache: CacheHeader,
pub archetype: ArchetypeHeader,
pub mirror: MirrorHeader,
pub merkle: MerkleHeader,
pub meta: MetaHeader,
pub tick_seq: u64,
pub epoch_ms: i64,
pub sequence: u64,
}
struct SnapshotBytes<const N: usize> {
headers: [[u8; SNAPSHOT_PREFIX_BYTES]; N],
payload: [u8; SUPERBLOCK_BYTES],
sequence: u64,
}
#[must_use = "hold the batch while updating headers, then commit or drop it"]
pub struct Publish<'a> {
_advisory: PublishGuard<'a>,
mmap: MutexGuard<'a, MmapMut>,
window: Option<u64>,
}
impl Drop for Publish<'_> {
fn drop(&mut self) {
if let Some(n) = self.window.take() {
seq_end(&mut self.mmap, n);
}
}
}
impl Publish<'_> {
fn begin_write(&mut self) -> u64 {
if let Some(n) = self.window {
return n;
}
let n = seq_begin(&mut self.mmap);
self.window = Some(n);
n
}
fn read_prefix(&self, slot: Slot) -> [u8; SNAPSHOT_PREFIX_BYTES] {
let mut bytes = [0; SNAPSHOT_PREFIX_BYTES];
let off = slot.offset();
let end = (off + SNAPSHOT_PREFIX_BYTES).min(self.mmap.len());
if off < end {
bytes[..end - off].copy_from_slice(&self.mmap[off..end]);
}
bytes
}
fn write_slot(&mut self, slot: Slot, bytes: &[u8; PAGE_SIZE]) {
let off = slot.offset();
let len = if slot == Slot::Superblock {
SUPERBLOCK_BYTES
} else {
PAGE_SIZE
};
if off + len > self.mmap.len() {
return;
}
self.begin_write();
self.mmap[off..off + len].copy_from_slice(&bytes[..len]);
}
pub fn is_advisory(&self) -> bool {
self._advisory.is_advisory()
}
pub fn commit(self) {}
pub fn read_superblock(&self) -> Option<Superblock> {
Superblock::from_payload(&self.mmap[..SUPERBLOCK_BYTES])
}
pub fn write_superblock(&mut self, h: &Superblock) {
self.begin_write();
self.mmap[..SUPERBLOCK_BYTES].copy_from_slice(&h.to_payload());
}
pub fn read_stream(&self) -> StreamHeader {
StreamHeader::from_bytes(&self.read_prefix(Slot::Stream))
}
pub fn write_stream(&mut self, h: &StreamHeader) {
self.write_slot(Slot::Stream, &h.to_bytes());
}
pub fn read_hebbian(&self) -> HebbianHeader {
HebbianHeader::from_bytes(&self.read_prefix(Slot::Hebbian))
}
pub fn write_hebbian(&mut self, h: &HebbianHeader) {
self.write_slot(Slot::Hebbian, &h.to_bytes());
}
pub fn read_attention(&self) -> AttentionHeader {
AttentionHeader::from_bytes(&self.read_prefix(Slot::Attention))
}
pub fn write_attention(&mut self, h: &AttentionHeader) {
self.write_slot(Slot::Attention, &h.to_bytes());
}
pub fn read_emotion(&self) -> EmotionHeader {
EmotionHeader::from_bytes(&self.read_prefix(Slot::Emotion))
}
pub fn write_emotion(&mut self, h: &EmotionHeader) {
self.write_slot(Slot::Emotion, &h.to_bytes());
}
pub fn read_resonance(&self) -> ResonanceHeader {
ResonanceHeader::from_bytes(&self.read_prefix(Slot::Resonance))
}
pub fn write_resonance(&mut self, h: &ResonanceHeader) {
self.write_slot(Slot::Resonance, &h.to_bytes());
}
pub fn read_graph(&self) -> GraphHeader {
GraphHeader::from_bytes(&self.read_prefix(Slot::Graph))
}
pub fn write_graph(&mut self, h: &GraphHeader) {
self.write_slot(Slot::Graph, &h.to_bytes());
}
pub fn read_cache(&self) -> CacheHeader {
CacheHeader::from_bytes(&self.read_prefix(Slot::Cache))
}
pub fn write_cache(&mut self, h: &CacheHeader) {
self.write_slot(Slot::Cache, &h.to_bytes());
}
pub fn read_archetype(&self) -> ArchetypeHeader {
ArchetypeHeader::from_bytes(&self.read_prefix(Slot::Archetype))
}
pub fn write_archetype(&mut self, h: &ArchetypeHeader) {
self.write_slot(Slot::Archetype, &h.to_bytes());
}
pub fn read_mirror(&self) -> MirrorHeader {
MirrorHeader::from_bytes(&self.read_prefix(Slot::Mirror))
}
pub fn write_mirror(&mut self, h: &MirrorHeader) {
self.write_slot(Slot::Mirror, &h.to_bytes());
}
pub fn read_merkle(&self) -> MerkleHeader {
MerkleHeader::from_bytes(&self.read_prefix(Slot::Merkle))
}
pub fn write_merkle(&mut self, h: &MerkleHeader) {
self.write_slot(Slot::Merkle, &h.to_bytes());
}
pub fn read_meta(&self) -> MetaHeader {
MetaHeader::from_bytes(&self.read_prefix(Slot::Meta))
}
pub fn write_meta(&mut self, h: &MetaHeader) {
self.write_slot(Slot::Meta, &h.to_bytes());
}
#[allow(clippy::too_many_arguments)]
pub fn tick_stream(
&mut self,
cycle: u64,
surprise: f32,
curiosity: f32,
predicted_hash: u64,
predicted_confidence: f32,
last_query_hash: u64,
last_query_ms: u64,
hebbian_total_energy: f64,
epoch_ms: i64,
) {
if self.mmap.len() < Slot::Stream.offset() + PAGE_SIZE {
return;
}
let h = StreamHeader {
cycle,
surprise_level: surprise,
curiosity_level: curiosity,
predicted_query_hash: predicted_hash,
predicted_confidence,
last_query_hash,
last_query_ms,
hebbian_total_energy,
updated_ms: epoch_ms as u64,
_pad: [0; PAGE_SIZE - StreamHeader::HEADER_BYTES],
};
let n = self.begin_write();
let payload = Superblock {
tick_seq: n.wrapping_add(1) >> 1,
epoch_ms,
..current_superblock(&self.mmap)
}
.to_payload();
self.mmap[..SUPERBLOCK_BYTES].copy_from_slice(&payload);
self.write_stream(&h);
}
}
pub struct PublishGuard<'a> {
file: Option<&'a std::fs::File>,
held: bool,
}
impl Drop for PublishGuard<'_> {
fn drop(&mut self) {
#[cfg(unix)]
if let (true, Some(file)) = (self.held, self.file) {
use std::os::fd::AsRawFd;
let _ = writer_lock::try_op(file.as_raw_fd(), writer_lock::UN);
}
}
}
impl PublishGuard<'_> {
pub fn is_advisory(&self) -> bool {
self.held
}
}
pub struct Spine {
output_dir: PathBuf,
mmap: Arc<Mutex<MmapMut>>,
lock_file: Option<std::fs::File>,
publish_lock_file: Option<std::fs::File>,
publish_lock_advisory: bool,
}
unsafe fn seq_atomic(mmap: &mut MmapMut) -> &AtomicU64 {
assert!(
mmap.len() >= SEQ_SLOT + 8,
"spine mapping is {} bytes, too short to carry a counter at {SEQ_SLOT}",
mmap.len()
);
let base = mmap.as_mut_ptr();
debug_assert_eq!(
(base as usize + SEQ_SLOT) % std::mem::align_of::<AtomicU64>(),
0,
"publish counter is not AtomicU64-aligned"
);
unsafe { AtomicU64::from_ptr(base.add(SEQ_SLOT).cast::<u64>()) }
}
fn seq_load(mmap: &mut MmapMut) -> u64 {
unsafe { seq_atomic(mmap).load(Ordering::Acquire) }
}
fn snapshot_backoff(attempt: u32) {
if attempt < 64 {
std::hint::spin_loop();
} else {
std::thread::yield_now();
}
}
fn seq_begin(mmap: &mut MmapMut) -> u64 {
unsafe { seq_atomic(mmap).fetch_or(1, Ordering::AcqRel) | 1 }
}
fn seq_end(mmap: &mut MmapMut, n: u64) {
unsafe { seq_atomic(mmap).store(n.wrapping_add(1), Ordering::Release) }
}
#[cfg(unix)]
mod writer_lock {
use std::io;
pub const SH: i32 = 1;
pub const EX: i32 = 2;
pub const NB: i32 = 4;
pub const UN: i32 = 8;
unsafe extern "C" {
fn flock(fd: i32, operation: i32) -> i32;
}
pub fn try_op(fd: i32, operation: i32) -> Result<(), (bool, io::Error)> {
if unsafe { flock(fd, operation) } == 0 {
return Ok(());
}
let err = io::Error::last_os_error();
let contended = operation & NB != 0 && matches!(err.raw_os_error(), Some(11) | Some(35));
Err((contended, err))
}
}
#[cfg(not(unix))]
mod writer_lock {
use std::io;
#[allow(dead_code)]
pub const SH: i32 = 1;
#[allow(dead_code)]
pub const EX: i32 = 2;
#[allow(dead_code)]
pub const NB: i32 = 4;
#[allow(dead_code)]
pub const UN: i32 = 8;
pub fn try_op(_fd: i32, _operation: i32) -> Result<(), (bool, io::Error)> {
Err((
false,
io::Error::other("advisory writer locks are unavailable on this target"),
))
}
}
impl Spine {
fn open_publish_lock(
dir: &Path,
create: bool,
) -> std::io::Result<(Option<std::fs::File>, bool)> {
let path = dir.join(PUBLISH_LOCK_FILE_NAME);
let opened = if create {
OpenOptions::new()
.read(true)
.write(true)
.truncate(false)
.create(true)
.open(&path)
} else {
OpenOptions::new().read(true).write(true).open(&path)
};
match opened {
Ok(file) => Ok((Some(file), cfg!(unix))),
Err(_) if !create => Ok((None, false)),
Err(e) => Err(e),
}
}
pub fn begin_publish(&self) -> Publish<'_> {
let mmap = self.mmap.lock();
let advisory = self.publish_guard();
Publish {
_advisory: advisory,
mmap,
window: None,
}
}
fn publish_guard(&self) -> PublishGuard<'_> {
#[cfg(unix)]
if let Some(file) = self.publish_lock_file.as_ref() {
use std::os::fd::AsRawFd;
if writer_lock::try_op(file.as_raw_fd(), writer_lock::EX).is_ok() {
return PublishGuard {
file: Some(file),
held: true,
};
}
}
PublishGuard {
file: None,
held: false,
}
}
pub fn publish_lock_is_advisory(&self) -> bool {
self.publish_lock_advisory
}
pub fn open(output_dir: impl AsRef<Path>) -> std::io::Result<Self> {
let dir = output_dir.as_ref().to_path_buf();
std::fs::create_dir_all(&dir)?;
let path = dir.join("spine.bin");
let file = OpenOptions::new()
.read(true)
.write(true)
.truncate(false)
.create(true)
.open(&path)?;
let min_len = (Slot::WalStart as usize + 8) * PAGE_SIZE;
if file.metadata()?.len() < min_len as u64 {
file.set_len(min_len as u64)?;
}
let lock_file = OpenOptions::new()
.read(true)
.write(true)
.truncate(false)
.create(true)
.open(dir.join(WRITER_LOCK_FILE_NAME))?;
Self::mark_shared(&lock_file)?;
let (publish_lock_file, publish_lock_advisory) = Self::open_publish_lock(&dir, true)?;
let mmap = unsafe {
MmapOptions::new()
.map_mut(&file)
.map_err(std::io::Error::other)?
};
let spine = Self {
output_dir: dir,
mmap: Arc::new(Mutex::new(mmap)),
lock_file: Some(lock_file),
publish_lock_file,
publish_lock_advisory,
};
spine.init_superblock()?;
Ok(spine)
}
pub fn open_ro(output_dir: impl AsRef<Path>) -> std::io::Result<Self> {
let dir = output_dir.as_ref().to_path_buf();
let file = OpenOptions::new()
.read(true)
.write(true)
.truncate(false)
.open(dir.join("spine.bin"))?;
let mmap = unsafe {
MmapOptions::new()
.map_mut(&file)
.map_err(std::io::Error::other)?
};
if mmap.len() < SEQ_SLOT + 8 {
return Err(std::io::Error::new(
std::io::ErrorKind::InvalidInput,
"spine.bin is shorter than the publish header",
));
}
let (publish_lock_file, publish_lock_advisory) = Self::open_publish_lock(&dir, false)?;
Ok(Self {
output_dir: dir,
mmap: Arc::new(Mutex::new(mmap)),
lock_file: None,
publish_lock_file,
publish_lock_advisory,
})
}
#[cfg(not(unix))]
fn mark_shared(_lock_file: &std::fs::File) -> std::io::Result<()> {
Ok(())
}
#[cfg(unix)]
fn mark_shared(lock_file: &std::fs::File) -> std::io::Result<()> {
use std::os::fd::AsRawFd;
let deadline = std::time::Instant::now() + std::time::Duration::from_millis(500);
loop {
match writer_lock::try_op(lock_file.as_raw_fd(), writer_lock::SH | writer_lock::NB) {
Ok(()) => return Ok(()),
Err((true, _)) if std::time::Instant::now() < deadline => {
std::thread::sleep(std::time::Duration::from_millis(2));
}
Err((true, _)) => {
return Err(std::io::Error::new(
std::io::ErrorKind::WouldBlock,
"another process holds the spine lock exclusively",
))
}
Err((false, e)) => return Err(e),
}
}
}
#[cfg(unix)]
fn acquire_exclusive_until(lock: &std::fs::File) -> std::io::Result<()> {
use std::os::fd::AsRawFd;
let deadline = std::time::Instant::now() + std::time::Duration::from_millis(500);
loop {
match writer_lock::try_op(lock.as_raw_fd(), writer_lock::EX | writer_lock::NB) {
Ok(()) => return Ok(()),
Err((true, _)) if std::time::Instant::now() < deadline => {
std::thread::sleep(std::time::Duration::from_millis(2));
}
Err((true, _)) => {
return Err(std::io::Error::new(
std::io::ErrorKind::WouldBlock,
"another process holds the spine lock; resize would remap its pages",
))
}
Err((false, e)) => return Err(e),
}
}
}
fn init_superblock(&self) -> std::io::Result<()> {
let mut m = self.mmap.lock();
let publish = self.publish_guard();
if Superblock::from_payload(&m[0..SUPERBLOCK_BYTES]).is_some() {
return Ok(()); }
let fresh = match Superblock::from_v1_payload(&m[0..SUPERBLOCK_BYTES]) {
Some(old) => Superblock {
version: SUPERBLOCK_VERSION,
..old
},
None => Superblock::new(),
};
let n = seq_begin(&mut m); m[0..SUPERBLOCK_BYTES].copy_from_slice(&fresh.to_payload());
seq_end(&mut m, n); drop(publish);
drop(m);
self.sync()?; Ok(())
}
fn read_slot(&self, slot: Slot) -> [u8; SNAPSHOT_PREFIX_BYTES] {
let off = slot.offset();
let m = self.mmap.lock();
let mut buf = [0u8; SNAPSHOT_PREFIX_BYTES];
let end = (off + SNAPSHOT_PREFIX_BYTES).min(m.len());
if off < end {
buf[..end - off].copy_from_slice(&m[off..end]);
}
buf
}
pub fn write_slot(&self, slot: Slot, bytes: &[u8; PAGE_SIZE]) {
self.begin_publish().write_slot(slot, bytes);
}
pub fn read_stream(&self) -> StreamHeader {
StreamHeader::from_bytes(&self.read_slot(Slot::Stream))
}
pub fn write_stream(&self, h: &StreamHeader) {
self.write_slot(Slot::Stream, &h.to_bytes());
}
#[allow(clippy::too_many_arguments)]
pub fn tick(
&self,
cycle: u64,
surprise: f32,
curiosity: f32,
predicted_hash: u64,
predicted_confidence: f32,
last_query_hash: u64,
last_query_ms: u64,
hebbian_total_energy: f64,
) {
let epoch_ms = now_ms();
let mut p = self.begin_publish();
p.tick_stream(
cycle,
surprise,
curiosity,
predicted_hash,
predicted_confidence,
last_query_hash,
last_query_ms,
hebbian_total_energy,
epoch_ms,
);
p.commit();
}
pub fn read_snapshot(&self) -> Option<PublishSnapshot> {
let bytes = self.read_prefixes([Slot::Stream])?;
let sb = Superblock::from_payload(&bytes.payload)?;
Some(PublishSnapshot {
stream: StreamHeader::from_bytes(&bytes.headers[0]),
tick_seq: sb.tick_seq,
epoch_ms: sb.epoch_ms,
sequence: bytes.sequence,
})
}
pub fn read_store_snapshot(&self) -> Option<StoreSnapshot> {
let bytes = self.read_prefixes([
Slot::Stream,
Slot::Hebbian,
Slot::Attention,
Slot::Emotion,
Slot::Resonance,
Slot::Graph,
Slot::Cache,
Slot::Archetype,
Slot::Mirror,
Slot::Merkle,
Slot::Meta,
])?;
let sb = Superblock::from_payload(&bytes.payload)?;
Some(StoreSnapshot {
stream: StreamHeader::from_bytes(&bytes.headers[0]),
hebbian: HebbianHeader::from_bytes(&bytes.headers[1]),
attention: AttentionHeader::from_bytes(&bytes.headers[2]),
emotion: EmotionHeader::from_bytes(&bytes.headers[3]),
resonance: ResonanceHeader::from_bytes(&bytes.headers[4]),
graph: GraphHeader::from_bytes(&bytes.headers[5]),
cache: CacheHeader::from_bytes(&bytes.headers[6]),
archetype: ArchetypeHeader::from_bytes(&bytes.headers[7]),
mirror: MirrorHeader::from_bytes(&bytes.headers[8]),
merkle: MerkleHeader::from_bytes(&bytes.headers[9]),
meta: MetaHeader::from_bytes(&bytes.headers[10]),
tick_seq: sb.tick_seq,
epoch_ms: sb.epoch_ms,
sequence: bytes.sequence,
})
}
fn read_prefixes<const N: usize>(&self, slots: [Slot; N]) -> Option<SnapshotBytes<N>> {
let required_len = slots
.iter()
.map(|s| s.offset() + SNAPSHOT_PREFIX_BYTES)
.max()
.unwrap_or(SEQ_SLOT + 8)
.max(SEQ_SLOT + 8);
if self.mmap.lock().len() < required_len {
return None;
}
let mut started = None;
let mut attempt = 0u32;
loop {
if let Some(bytes) = self.read_prefixes_once(&slots) {
return Some(bytes);
}
if self.snapshot_patience(&mut started, attempt) {
break;
}
snapshot_backoff(attempt);
attempt = attempt.saturating_add(1);
}
if self.recover_stalled_publish() {
return self.read_prefixes_once(&slots);
}
None
}
fn read_prefixes_once<const N: usize>(&self, slots: &[Slot; N]) -> Option<SnapshotBytes<N>> {
let mut m = self.mmap.lock();
if m.len() < SEQ_SLOT + 8
|| slots
.iter()
.any(|s| s.offset() + SNAPSHOT_PREFIX_BYTES > m.len())
{
return None;
}
let s1 = seq_load(&mut m);
if s1 & 1 == 1 {
return None;
}
let mut bytes = SnapshotBytes {
headers: [[0; SNAPSHOT_PREFIX_BYTES]; N],
payload: [0; SUPERBLOCK_BYTES],
sequence: s1,
};
for (prefix, slot) in bytes.headers.iter_mut().zip(slots) {
let off = slot.offset();
prefix.copy_from_slice(&m[off..off + SNAPSHOT_PREFIX_BYTES]);
}
bytes.payload.copy_from_slice(&m[..SUPERBLOCK_BYTES]);
std::sync::atomic::fence(Ordering::Acquire);
if seq_load(&mut m) != s1 {
return None;
}
Some(bytes)
}
fn snapshot_patience(&self, started: &mut Option<std::time::Instant>, attempt: u32) -> bool {
if attempt < SNAPSHOT_MAX_RETRIES {
return false;
}
let t0 = *started.get_or_insert_with(std::time::Instant::now);
t0.elapsed() >= std::time::Duration::from_millis(SNAPSHOT_MAX_WAIT_MS)
}
pub fn publish_window_open(&self) -> bool {
self.publish_in_flight()
}
pub fn recover_stalled_publish(&self) -> bool {
let mut m = self.mmap.lock();
if m.len() < SEQ_SLOT + 8 {
return false;
}
if seq_load(&mut m) & 1 == 0 {
return false;
}
let _publish = self.publish_guard();
let n = seq_load(&mut m);
if n & 1 == 0 {
return false; }
if !_publish.is_advisory() {
return false;
}
seq_end(&mut m, n);
true
}
pub fn publish_sequence(&self) -> u64 {
let mut m = self.mmap.lock();
if m.len() < SEQ_SLOT + 8 {
return 0;
}
seq_load(&mut m)
}
pub fn publish_in_flight(&self) -> bool {
let mut m = self.mmap.lock();
if m.len() < SEQ_SLOT + 8 {
return false;
}
seq_load(&mut m) & 1 == 1
}
pub fn read_hebbian(&self) -> HebbianHeader {
HebbianHeader::from_bytes(&self.read_slot(Slot::Hebbian))
}
pub fn write_hebbian(&self, h: &HebbianHeader) {
self.write_slot(Slot::Hebbian, &h.to_bytes());
}
pub fn read_attention(&self) -> AttentionHeader {
AttentionHeader::from_bytes(&self.read_slot(Slot::Attention))
}
pub fn write_attention(&self, h: &AttentionHeader) {
self.write_slot(Slot::Attention, &h.to_bytes());
}
pub fn read_emotion(&self) -> EmotionHeader {
EmotionHeader::from_bytes(&self.read_slot(Slot::Emotion))
}
pub fn write_emotion(&self, h: &EmotionHeader) {
self.write_slot(Slot::Emotion, &h.to_bytes());
}
pub fn read_resonance(&self) -> ResonanceHeader {
ResonanceHeader::from_bytes(&self.read_slot(Slot::Resonance))
}
pub fn write_resonance(&self, h: &ResonanceHeader) {
self.write_slot(Slot::Resonance, &h.to_bytes());
}
pub fn read_graph(&self) -> GraphHeader {
GraphHeader::from_bytes(&self.read_slot(Slot::Graph))
}
pub fn write_graph(&self, h: &GraphHeader) {
self.write_slot(Slot::Graph, &h.to_bytes());
}
pub fn read_cache(&self) -> CacheHeader {
CacheHeader::from_bytes(&self.read_slot(Slot::Cache))
}
pub fn write_cache(&self, h: &CacheHeader) {
self.write_slot(Slot::Cache, &h.to_bytes());
}
pub fn read_archetype(&self) -> ArchetypeHeader {
ArchetypeHeader::from_bytes(&self.read_slot(Slot::Archetype))
}
pub fn write_archetype(&self, h: &ArchetypeHeader) {
self.write_slot(Slot::Archetype, &h.to_bytes());
}
pub fn read_mirror(&self) -> MirrorHeader {
MirrorHeader::from_bytes(&self.read_slot(Slot::Mirror))
}
pub fn write_mirror(&self, h: &MirrorHeader) {
self.write_slot(Slot::Mirror, &h.to_bytes());
}
pub fn read_merkle(&self) -> MerkleHeader {
MerkleHeader::from_bytes(&self.read_slot(Slot::Merkle))
}
pub fn write_merkle(&self, h: &MerkleHeader) {
self.write_slot(Slot::Merkle, &h.to_bytes());
}
pub fn read_meta(&self) -> MetaHeader {
MetaHeader::from_bytes(&self.read_slot(Slot::Meta))
}
pub fn write_meta(&self, h: &MetaHeader) {
self.write_slot(Slot::Meta, &h.to_bytes());
}
pub fn read_superblock(&self) -> Option<Superblock> {
let m = self.mmap.lock();
if m.len() < SUPERBLOCK_BYTES {
return None;
}
Superblock::from_payload(&m[0..SUPERBLOCK_BYTES])
}
pub fn sync(&self) -> std::io::Result<()> {
self.mmap.lock().flush()
}
pub fn size_bytes(&self) -> usize {
self.mmap.lock().len()
}
pub fn output_dir(&self) -> &Path {
&self.output_dir
}
pub fn path(&self) -> PathBuf {
self.output_dir.join("spine.bin")
}
pub fn resize(&self, new_len: usize) -> std::io::Result<()> {
#[cfg(unix)]
use std::os::fd::AsRawFd;
let Some(lock) = self.lock_file.as_ref() else {
return Err(std::io::Error::new(
std::io::ErrorKind::PermissionDenied,
"read-only spine handle cannot resize",
));
};
#[cfg(unix)]
{
let _ = writer_lock::try_op(lock.as_raw_fd(), writer_lock::UN);
if let Err(e) = Self::acquire_exclusive_until(lock) {
let _ = Self::mark_shared(lock);
return Err(e);
}
}
#[cfg(not(unix))]
{
let _ = lock;
return Err(std::io::Error::other(
"resize needs advisory file locks, unavailable on this target",
));
}
let result = (|| -> std::io::Result<()> {
{
let f = OpenOptions::new()
.read(true)
.write(true)
.truncate(false)
.open(self.path())?;
f.set_len(new_len as u64)?;
}
let file = OpenOptions::new()
.read(true)
.write(true)
.truncate(false)
.open(self.path())?;
let new_mmap = unsafe {
MmapOptions::new()
.map_mut(&file)
.map_err(std::io::Error::other)?
};
*self.mmap.lock() = new_mmap;
Ok(())
})();
#[cfg(unix)]
{
let _ = writer_lock::try_op(lock.as_raw_fd(), writer_lock::UN);
let _ = Self::mark_shared(lock);
}
result
}
}
fn current_superblock(mmap: &MmapMut) -> Superblock {
match Superblock::from_payload(&mmap[0..SUPERBLOCK_BYTES]) {
Some(sb) => sb,
None => Superblock::new(), }
}
pub fn format_summary(spine: &Spine) -> String {
let Some(snap) = spine.read_store_snapshot() else {
return "Spine: snapshot unavailable (busy, short or invalid mapping)".to_owned();
};
let s = snap.stream;
let tick_seq = snap.tick_seq;
let epoch_ms = snap.epoch_ms;
let h = snap.hebbian;
let a = snap.attention;
let e = snap.emotion;
let r = snap.resonance;
let g = snap.graph;
let c = snap.cache;
let ar = snap.archetype;
let mi = snap.mirror;
let me = snap.merkle;
let meta = snap.meta;
format!(
"Spine: {} pages mapped ({} KB)\n\
Publish: seq={} epoch_ms={} in_flight={}\n\
Stream: cycle={} surprise={:.3} curiosity={:.3} predict_conf={:.3}\n\
Hebbian: {} activations, {} coactivations, delta_tail={}\n\
Attention: {} recalls, quality={:.2}\n\
Emotion: {} updates, dominant_idx={}\n\
Resonance: {} cells, {} emitted, {} received\n\
Graph: {} nodes, {} edges, {} patterns\n\
Cache: {} predictions, hit_rate={:.1}%\n\
Archetype: {} active, {} emerged\n\
Mirror: {} echoes, strongest={:.3}\n\
Merkle: height={}, {} blocks\n\
Meta: {} blocks, {} dream cycles",
spine.size_bytes() / PAGE_SIZE,
spine.size_bytes() / 1024,
tick_seq,
epoch_ms,
spine.publish_in_flight(),
s.cycle,
s.surprise_level,
s.curiosity_level,
s.predicted_confidence,
h.activation_count,
h.coactivation_count,
h.delta_tail_offset,
a.total_recalls,
a.quality_signal,
e.total_updates,
e.current[0],
r.field_cells,
r.pulses_emitted,
r.pulses_received,
g.node_count,
g.edge_count,
g.pattern_count,
c.total_predictions,
c.hit_rate() * 100.0,
ar.active_count,
ar.emergence_events,
mi.echo_count,
mi.strongest_similarity,
me.chain_height,
me.block_count,
meta.block_count,
meta.dream_cycle_count
)
}
#[cfg(test)]
mod tests {
use super::*;
fn temp_spine(tag: &str) -> Spine {
let dir = std::env::temp_dir().join(format!("m1c_spine_{}_{}", tag, std::process::id()));
let _ = std::fs::remove_dir_all(&dir);
Spine::open(&dir).unwrap()
}
#[test]
fn test_read_only_batch_does_not_publish() {
let spine = temp_spine("read_batch");
let before = spine.publish_sequence();
let p = spine.begin_publish();
assert_eq!(p.read_stream().cycle, 0);
assert!(p.read_superblock().is_some());
p.commit();
assert_eq!(spine.publish_sequence(), before);
}
#[test]
fn test_batch_is_one_window_and_excludes_partial_reads() {
let spine = temp_spine("batch_window");
let peer = Spine::open_ro(spine.output_dir()).unwrap();
let before = spine.publish_sequence();
let mut p = spine.begin_publish();
p.tick_stream(9, 0.0, 0.0, !9, 0.0, 0, 0, 9.0, 1234);
assert_eq!(peer.read_stream().cycle, 9);
assert_eq!(peer.read_graph().node_count, 0);
assert!(peer
.read_prefixes_once(&[Slot::Stream, Slot::Graph])
.is_none());
let mut graph = p.read_graph();
graph.node_count = 9;
p.write_graph(&graph);
let mut emotion = p.read_emotion();
emotion.current[20] = 9.0;
emotion.total_updates = 9;
p.write_emotion(&emotion);
p.commit();
assert_eq!(spine.publish_sequence(), before + 2);
let snap = peer.read_store_snapshot().unwrap();
assert_eq!(snap.stream.cycle, 9);
assert_eq!(snap.graph.node_count, 9);
assert_eq!(snap.emotion.current[20], 9.0);
assert_eq!(snap.emotion.total_updates, 9);
assert_eq!(snap.epoch_ms, 1234);
assert_eq!(snap.stream.updated_ms, 1234);
}
#[test]
fn test_drop_and_unwinding_close_the_batch_without_rollback() {
let spine = temp_spine("batch_drop");
let before = spine.publish_sequence();
{
let mut p = spine.begin_publish();
let mut h = p.read_meta();
h.block_count = 7;
p.write_meta(&h);
}
assert_eq!(spine.publish_sequence(), before + 2);
let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
let mut p = spine.begin_publish();
let mut h = p.read_graph();
h.node_count = 13;
p.write_graph(&h);
panic!("batch interrupted");
}));
assert!(result.is_err());
assert_eq!(spine.publish_sequence(), before + 4);
let snap = spine.read_store_snapshot().unwrap();
assert_eq!(snap.meta.block_count, 7);
assert_eq!(snap.graph.node_count, 13);
spine.tick(2, 0.0, 0.0, 0, 0.0, 0, 0, 0.0);
assert!(!spine.publish_window_open());
}
#[test]
fn test_each_legacy_header_write_opens_a_window() {
let spine = temp_spine("legacy_windows");
let before = spine.publish_sequence();
let h = spine.read_hebbian();
spine.write_hebbian(&h);
let g = spine.read_graph();
spine.write_graph(&g);
let m = spine.read_meta();
spine.write_meta(&m);
assert_eq!(spine.publish_sequence(), before + 6);
assert!(spine.read_store_snapshot().is_some());
}
#[test]
fn test_store_snapshot_rejects_short_and_invalid_mappings() {
let spine = temp_spine("short_snapshot");
let path = spine.output_dir().join("short.bin");
let file = OpenOptions::new()
.read(true)
.write(true)
.create(true)
.truncate(true)
.open(path)
.unwrap();
file.set_len((Slot::Stream.offset() + SNAPSHOT_PREFIX_BYTES) as u64)
.unwrap();
let mut short = unsafe { MmapOptions::new().map_mut(&file).unwrap() };
short[..SUPERBLOCK_BYTES].copy_from_slice(&Superblock::new().to_payload());
*spine.mmap.lock() = short;
assert!(spine.read_snapshot().is_some());
assert!(spine.read_store_snapshot().is_none());
assert!(format_summary(&spine).contains("snapshot unavailable"));
spine.mmap.lock()[..SUPERBLOCK_BYTES].fill(0);
assert!(spine.read_snapshot().is_none());
}
#[test]
fn test_republishing_after_a_stall_never_marks_the_window_even() {
let spine = temp_spine("republish_stall");
let mut m = spine.mmap.lock();
let odd = seq_begin(&mut m);
assert_eq!(seq_begin(&mut m), odd);
assert_eq!(seq_load(&mut m), odd);
seq_end(&mut m, odd);
assert_eq!(seq_load(&mut m), odd + 1);
}
#[test]
fn test_open_creates_superblock() {
let spine = temp_spine("sb");
let sb = spine.read_superblock().unwrap();
assert_eq!(sb.magic, MAGIC);
assert_eq!(sb.page_size as usize, PAGE_SIZE);
}
#[test]
fn test_stream_roundtrip() {
let spine = temp_spine("stream");
spine.tick(42, 0.5, 0.7, 0xdeadbeef, 0.9, 0xcafebabe, 12345, 1337.5);
let s = spine.read_stream();
assert_eq!(s.cycle, 42);
assert!((s.surprise_level - 0.5).abs() < 1e-6);
assert_eq!(s.predicted_query_hash, 0xdeadbeef);
assert!((s.hebbian_total_energy - 1337.5).abs() < 1e-6);
assert!(s.updated_ms > 0);
}
#[test]
fn test_layer_headers_roundtrip() {
let spine = temp_spine("layers");
let mut h = spine.read_hebbian();
h.activation_count = 999;
h.delta_tail_offset = 123456789;
h.decay_event_count = 0;
spine.write_hebbian(&h);
assert_eq!(spine.read_hebbian().activation_count, 999);
assert_eq!(spine.read_hebbian().delta_tail_offset, 123456789);
let mut a = spine.read_attention();
a.learned_weights[0] = 0.25;
a.total_recalls = 17;
spine.write_attention(&a);
assert!((spine.read_attention().learned_weights[0] - 0.25).abs() < 1e-6);
assert_eq!(spine.read_attention().total_recalls, 17);
let mut e = spine.read_emotion();
e.current[3] = 0.8;
e.total_updates = 100;
spine.write_emotion(&e);
assert!((spine.read_emotion().current[3] - 0.8).abs() < 1e-6);
let mut g = spine.read_graph();
g.node_count = 500;
spine.write_graph(&g);
assert_eq!(spine.read_graph().node_count, 500);
let mut c = spine.read_cache();
c.total_hits = 10;
c.total_misses = 2;
spine.write_cache(&c);
let rc = spine.read_cache();
assert!((rc.hit_rate() - 10.0 / 12.0).abs() < 1e-5);
}
#[test]
fn test_all_slots_distinct() {
let spine = temp_spine("slots");
spine.tick(1, 1.0, 1.0, 1, 1.0, 1, 1, 1.0);
let mut g = spine.read_graph();
g.node_count = 42;
spine.write_graph(&g);
let mut m = spine.read_mirror();
m.echo_count = 7;
spine.write_mirror(&m);
assert_eq!(spine.read_stream().cycle, 1);
assert_eq!(spine.read_graph().node_count, 42);
assert_eq!(spine.read_mirror().echo_count, 7);
assert_eq!(spine.read_archetype().active_count, 0);
}
#[test]
fn test_reopen_preserves_all_layers() {
let dir = std::env::temp_dir().join(format!("m1c_spine_reopen_{}", std::process::id()));
let _ = std::fs::remove_dir_all(&dir);
{
let spine = Spine::open(&dir).unwrap();
spine.tick(7, 0.3, 0.4, 5, 6.0, 7, 8, 9.0);
let mut a = spine.read_attention();
a.total_recalls = 55;
spine.write_attention(&a);
spine.sync().unwrap();
}
let spine = Spine::open(&dir).unwrap();
assert_eq!(spine.read_stream().cycle, 7);
assert!((spine.read_stream().surprise_level - 0.3).abs() < 1e-6);
assert_eq!(spine.read_attention().total_recalls, 55);
}
#[test]
fn test_superblock_payload_carries_tick_seq() {
let spine = temp_spine("sb_payload");
assert_eq!(spine.read_superblock().unwrap().tick_seq, 0);
spine.tick(1, 0.0, 0.0, 0, 0.0, 0, 0, 0.0);
let after_first = spine.read_superblock().unwrap().tick_seq;
spine.tick(2, 0.0, 0.0, 0, 0.0, 0, 0, 0.0);
let after_second = spine.read_superblock().unwrap().tick_seq;
assert!(after_first > 0, "first publish did not bump the sequence");
assert_eq!(after_second, after_first + 1);
}
#[test]
fn test_snapshot_pairs_stream_and_superblock() {
let spine = temp_spine("snapshot");
for cycle in 100..110 {
spine.tick(cycle, 0.25, 0.5, cycle, 1.0, cycle, cycle, cycle as f64);
let snap = spine
.read_snapshot()
.expect("seqlock should not starve a same-process reader");
assert_eq!(snap.stream.cycle, cycle);
assert_eq!(snap.epoch_ms, snap.stream.updated_ms as i64);
assert_eq!(
snap.sequence % 2,
0,
"a published snapshot is never taken mid-window"
);
}
}
#[test]
fn test_write_slot_on_superblock_keeps_the_counter() {
let spine = temp_spine("sb_slot");
spine.tick(7, 0.0, 0.0, 0, 0.0, 0, 0, 0.0);
let before = spine.publish_sequence();
assert!(before > 0);
let junk = [0xABu8; PAGE_SIZE];
spine.write_slot(Slot::Superblock, &junk);
let after = spine.publish_sequence();
assert_eq!(after, before + 2, "counter did not advance as a publish");
assert_eq!(after % 2, 0, "counter left inside a publish window");
assert!(spine.read_superblock().is_none());
assert!(!spine.publish_in_flight());
}
#[test]
fn test_open_upgrades_a_version_one_superblock() {
let dir = std::env::temp_dir().join(format!("m1c_spine_v1_{}", std::process::id()));
let _ = std::fs::remove_dir_all(&dir);
{
let spine = Spine::open(&dir).unwrap();
let mut payload = [0u8; SUPERBLOCK_BYTES];
payload[0..8].copy_from_slice(&MAGIC.to_le_bytes());
payload[8..12].copy_from_slice(&1u32.to_le_bytes());
payload[12..16].copy_from_slice(&(PAGE_SIZE as u32).to_le_bytes());
payload[16..20].copy_from_slice(&(Slot::WalStart as u32).to_le_bytes());
payload[20..24].copy_from_slice(&(Slot::WalStart as u32).to_le_bytes());
payload[24..32].copy_from_slice(&1_700_000_000_000i64.to_le_bytes());
let mut page = [0u8; PAGE_SIZE];
page[..SUPERBLOCK_BYTES].copy_from_slice(&payload);
spine.write_slot(Slot::Superblock, &page);
}
let spine = Spine::open(&dir).unwrap();
let sb = spine.read_superblock().unwrap();
assert_eq!(sb.version, SUPERBLOCK_VERSION);
assert_eq!(sb.wal_start_page, Slot::WalStart as u32);
assert_eq!(sb.tick_seq, 0);
assert_eq!(sb.epoch_ms, 1_700_000_000_000);
}
#[test]
fn test_read_only_handle_sees_the_writers_publish() {
let dir = std::env::temp_dir().join(format!("m1c_spine_ro_{}", std::process::id()));
let _ = std::fs::remove_dir_all(&dir);
let writer = Spine::open(&dir).unwrap();
let reader = Spine::open_ro(&dir).unwrap();
for cycle in 1..6 {
writer.tick(cycle, 0.0, 0.0, cycle, 0.0, 0, 0, cycle as f64);
let snap = reader
.read_snapshot()
.expect("read-only handle reads published snapshots");
assert_eq!(snap.stream.cycle, cycle);
assert_eq!(snap.stream.predicted_query_hash, cycle);
}
assert!(
reader.resize(1 << 20).is_err(),
"a read-only handle must not remap the file"
);
}
#[test]
#[should_panic(expected = "too short to carry a counter")]
fn test_counter_access_on_a_short_mapping_panics() {
use std::io::Write;
let dir = std::env::temp_dir().join(format!("m1c_spine_short_{}", std::process::id()));
let _ = std::fs::remove_dir_all(&dir);
std::fs::create_dir_all(&dir).expect("create dir");
let path = dir.join("tiny.bin");
{
let mut f = std::fs::OpenOptions::new()
.read(true)
.write(true)
.create(true)
.truncate(true)
.open(&path)
.expect("create tiny file");
f.write_all(&[0u8; 16]).expect("write");
}
let file = std::fs::OpenOptions::new()
.read(true)
.write(true)
.open(&path)
.expect("open tiny file");
let mut mmap = unsafe { MmapOptions::new().map_mut(&file).expect("map") };
let _ = seq_load(&mut mmap);
let _ = std::fs::remove_dir_all(&dir);
}
#[test]
fn test_stalled_publish_window_is_recoverable() {
let spine = temp_spine("stall");
spine.tick(1, 0.0, 0.0, 0, 0.0, 0, 0, 0.0);
{
let mut m = spine.mmap.lock();
let _n = seq_begin(&mut m);
}
assert!(spine.publish_window_open(), "window should read as open");
let before = spine.read_stream().cycle;
assert!(spine.recover_stalled_publish());
assert!(
!spine.publish_window_open(),
"recovery should close the window"
);
assert!(!spine.recover_stalled_publish(), "nothing to recover twice");
assert_eq!(
spine.read_stream().cycle,
before,
"recovery must not publish"
);
{
let mut m = spine.mmap.lock();
let _n = seq_begin(&mut m);
}
assert!(
spine.read_snapshot().is_some(),
"read_snapshot must heal a stalled window"
);
assert!(!spine.publish_window_open());
spine.tick(2, 0.0, 0.0, 0, 0.0, 0, 0, 0.0);
assert_eq!(spine.read_snapshot().unwrap().stream.cycle, 2);
}
#[test]
fn test_summary_reports_the_publish_record() {
let spine = temp_spine("summary_publish");
spine.tick(11, 0.5, 0.6, 0, 0.7, 0, 0, 100.0);
let out = format_summary(&spine);
assert!(out.contains("Publish:"), "{out}");
let seq = spine.read_snapshot().unwrap().tick_seq;
assert!(
out.contains(&format!("seq={seq}")),
"the summary must print the counter, not a literal: {out}"
);
}
#[test]
fn test_summary_formats() {
let spine = temp_spine("summary");
spine.tick(10, 0.5, 0.6, 0, 0.7, 0, 0, 100.0);
let out = format_summary(&spine);
assert!(out.contains("Spine:"));
assert!(out.contains("cycle=10"));
}
}