diff --git a/server/src/bin/storage_bench.rs b/server/src/bin/storage_bench.rs index 4762d0d23bc8e2df439afe09a3235ba7f72bd486..3d988416ad36b27a7d3dc84280cfdbdcafa43e69 100644 --- a/server/src/bin/storage_bench.rs +++ b/server/src/bin/storage_bench.rs @@ -6,8 +6,8 @@ use std::{ }; use sorter2_server::{ - entity_store::EntityStore, event_log::EventLog, events::Event, journal::JournalClient, - projection_apply, projection_store::ProjectionStore, + event_log::EventLog, events::Event, journal::JournalClient, projection_apply, + projection_store::ProjectionStore, }; #[tokio::main] @@ -18,12 +18,10 @@ async fn main() -> Result<(), Box> { let data_dir = opts.data_dir.to_string_lossy().into_owned(); let event_log = Arc::new(EventLog::new(format!("{data_dir}/events.jsonl"))); let db = durable::Db::open(opts.data_dir.join("store"))?; - let entity_store = EntityStore::from_db(&db)?; let projection_store = ProjectionStore::from_db(&db)?; let journal = JournalClient::spawn( event_log.clone(), - entity_store.clone(), projection_store.clone(), event_log.last_sequence().await? + 1, ); @@ -46,12 +44,9 @@ async fn main() -> Result<(), Box> { drop(journal); let rebuild_start = Instant::now(); - entity_store.reset()?; projection_store.reset()?; let rebuild = event_log - .replay(|record| { - projection_apply::apply_records(&projection_store, &entity_store, &[record]) - }) + .replay(|record| projection_apply::apply_records(&projection_store, &[record])) .await?; let rebuild_elapsed = rebuild_start.elapsed(); diff --git a/server/src/entity_store.rs b/server/src/entity_store.rs deleted file mode 100644 index d29f39aecca6f12cdcf263cf77c3654eb4ee6cfa..0000000000000000000000000000000000000000 --- a/server/src/entity_store.rs +++ /dev/null @@ -1,134 +0,0 @@ -//! Off-heap storage for full entity payloads (Reddit API JSON). -//! -//! Derived [`crate::reducer::EntityData`] is stored on the node; the raw JSON -//! lives here, in the shared durable [`Store`] schema. - -use std::path::Path; - -use durable::{Batch, Db, Durability}; -use serde_json::Value; - -use crate::{ - path_types::ItemId, - storage_dto::{decode_entity_payload, encode_entity_payload}, - storage_schema::{Store, StoreFields}, -}; - -const ENTITY_SCHEMA_KEY: &str = "schema_version"; -const ENTITY_SCHEMA_VERSION: u64 = 2; - -#[derive(Debug, thiserror::Error)] -pub enum EntityStoreError { - #[error("durable error: {0}")] - Durable(#[from] durable::Error), - #[error("json error: {0}")] - Json(#[from] serde_json::Error), - #[error("storage decode error: {0}")] - Storage(String), - #[error("io error: {0}")] - Io(#[from] std::io::Error), -} - -/// Disk-backed map of entity id → raw JSON payload. -#[derive(Clone)] -pub struct EntityStore { - db: Db, -} - -impl EntityStore { - /// Open (or create) the entity database under `dir`. - pub fn open(dir: &Path) -> Result { - std::fs::create_dir_all(dir)?; - let db = Db::open(dir)?; - Self::from_db(&db) - } - - /// Create an entity store backed by an already-open database. - pub fn from_db(db: &Db) -> Result { - let store = Self { db: db.clone() }; - let version = Store::root() - .entity_meta() - .key(&ENTITY_SCHEMA_KEY.to_string()) - .get(db)?; - if version != Some(ENTITY_SCHEMA_VERSION) { - store.reset()?; - } - Ok(store) - } - - /// Clear rebuildable entity payloads and reset storage schema metadata. - pub fn reset(&self) -> Result<(), EntityStoreError> { - let root = Store::root(); - self.db.apply( - &[root.entities().clear(), root.entity_meta().clear()], - Durability::SyncWal, - )?; - self.db.run( - root.entity_meta() - .key(&ENTITY_SCHEMA_KEY.to_string()) - .set(&ENTITY_SCHEMA_VERSION), - Durability::SyncWal, - )?; - Ok(()) - } - - /// Persist a payload for `id` (overwrites any existing entry). - pub fn put(&self, id: &ItemId, payload: &Value) -> Result<(), EntityStoreError> { - self.db.run( - Store::root() - .entities() - .key(&id.as_str().to_string()) - .set(&encode_entity_payload(payload)), - Durability::SyncWal, - )?; - Ok(()) - } - - /// Add a payload write to the caller's batch. - pub fn put_in_batch( - &self, - batch: &mut Batch, - id: &ItemId, - payload: &Value, - ) -> Result<(), EntityStoreError> { - batch.write( - Store::root() - .entities() - .key(&id.as_str().to_string()) - .set(&encode_entity_payload(payload)), - ); - Ok(()) - } - - /// Load a stored payload, if present. - pub fn get(&self, id: &ItemId) -> Result, EntityStoreError> { - match Store::root() - .entities() - .key(&id.as_str().to_string()) - .get(&self.db)? - { - Some(record) => decode_entity_payload(record) - .map(Some) - .map_err(EntityStoreError::Storage), - None => Ok(None), - } - } -} - -#[cfg(test)] -mod tests { - use super::*; - use serde_json::json; - - #[test] - fn round_trip_payload() { - let tmp = tempfile::tempdir().unwrap(); - let store = EntityStore::open(tmp.path()).unwrap(); - let id = ItemId::from_url("https://reddit.com/r/rust").unwrap(); - let payload = json!({"kind": "t5", "data": {"display_name": "rust"}}); - - store.put(&id, &payload).unwrap(); - let loaded = store.get(&id).unwrap().unwrap(); - assert_eq!(loaded, payload); - } -} diff --git a/server/src/events.rs b/server/src/events.rs index a3d88e285c645a96d33e6f0aed1b487b843aceb8..d76c3bb4277216b0d39c9422ba7a50db10a95e05 100644 --- a/server/src/events.rs +++ b/server/src/events.rs @@ -1,5 +1,4 @@ use serde::{Deserialize, Serialize}; -use serde_json::Value; /// Schema version for JSONL log records. Bump when event semantics change. pub const CURRENT_LOG_SCHEMA: u32 = 1; @@ -30,7 +29,7 @@ pub type ViewRecord = LogRecord; /// Wall-clock timestamp carried on the log envelope for domain events. pub fn event_timestamp(event: &Event) -> i64 { match event { - Event::VoteRecorded { ts, .. } | Event::EntityImported { ts, .. } => *ts, + Event::VoteRecorded { ts, .. } => *ts, Event::NodeEnsured { .. } => crate::fetch::now_ms(), } } @@ -57,6 +56,4 @@ pub enum Event { }, /// Register a node path in the fractal tree (no external fetch). NodeEnsured { id: String }, - /// Full upstream API payload for a node (domain-specific view derived at replay/render time). - EntityImported { id: String, ts: i64, payload: Value }, } diff --git a/server/src/journal.rs b/server/src/journal.rs index 50bc89f976edb82b7b0e49e954a8eccbbe82bf87..d50023aca7c3e74068de988449b11baee32d1ecf 100644 --- a/server/src/journal.rs +++ b/server/src/journal.rs @@ -5,7 +5,6 @@ use std::sync::Arc; use tokio::sync::{mpsc, oneshot}; use crate::{ - entity_store::EntityStore, event_log::EventLog, events::{event_timestamp, Event, EventRecord}, projection_apply, @@ -25,7 +24,6 @@ pub struct JournalClient { impl JournalClient { pub fn spawn( event_log: Arc, - entity_store: EntityStore, projection_store: ProjectionStore, next_seq: u64, ) -> Self { @@ -33,7 +31,6 @@ impl JournalClient { tokio::spawn(journal_worker( rx, event_log, - entity_store, projection_store, next_seq, )); @@ -62,7 +59,6 @@ impl JournalClient { async fn journal_worker( mut rx: mpsc::Receiver, event_log: Arc, - entity_store: EntityStore, projection_store: ProjectionStore, mut next_seq: u64, ) { @@ -75,7 +71,6 @@ async fn journal_worker( let result = append_and_project_batch( &event_log, &projection_store, - &entity_store, &mut next_seq, &batch, ) @@ -99,7 +94,6 @@ async fn journal_worker( async fn append_and_project_batch( event_log: &EventLog, projection_store: &ProjectionStore, - entity_store: &EntityStore, next_seq: &mut u64, commands: &[JournalCommand], ) -> Result<(), String> { @@ -117,7 +111,7 @@ async fn append_and_project_batch( .await .map_err(|e| e.to_string())?; *next_seq = seq; - projection_apply::apply_records(projection_store, entity_store, &records) + projection_apply::apply_records(projection_store, &records) .map_err(|e| format!("projection apply failed after durable append: {e}")) } @@ -132,10 +126,9 @@ mod tests { let log_path = tmp.path().join("events.jsonl"); let event_log = Arc::new(EventLog::new(log_path)); let db = durable::Db::open(tmp.path().join("store")).unwrap(); - let entity_store = EntityStore::from_db(&db).unwrap(); let projection_store = ProjectionStore::from_db(&db).unwrap(); - let journal = JournalClient::spawn(event_log, entity_store, projection_store.clone(), 1); + let journal = JournalClient::spawn(event_log, projection_store.clone(), 1); let j1 = journal.clone(); let j2 = journal.clone(); @@ -177,11 +170,9 @@ mod tests { .unwrap(); let db = durable::Db::open(tmp.path().join("store")).unwrap(); - let entity_store = EntityStore::from_db(&db).unwrap(); let projection_store = ProjectionStore::from_db(&db).unwrap(); projection_apply::apply_records( &projection_store, - &entity_store, &[EventRecord::new( 1, 1, @@ -196,7 +187,6 @@ mod tests { let journal = JournalClient::spawn( event_log.clone(), - entity_store, projection_store.clone(), next_seq, ); @@ -219,10 +209,9 @@ mod tests { let log_path = tmp.path().join("events.jsonl"); let event_log = Arc::new(EventLog::new(log_path)); let db = durable::Db::open(tmp.path().join("store")).unwrap(); - let entity_store = EntityStore::from_db(&db).unwrap(); let projection_store = ProjectionStore::from_db(&db).unwrap(); let journal = - JournalClient::spawn(event_log.clone(), entity_store, projection_store.clone(), 1); + JournalClient::spawn(event_log.clone(), projection_store.clone(), 1); journal .append_many(vec![ diff --git a/server/src/lib.rs b/server/src/lib.rs index 5c02c8e704e4664453bad75d819df8a067668176..3dfc7c8acb8ed61bb73ade63e72768e402042cc5 100644 --- a/server/src/lib.rs +++ b/server/src/lib.rs @@ -1,5 +1,4 @@ pub mod api; -pub mod entity_store; pub mod event_log; pub mod events; pub mod fetch; diff --git a/server/src/projection_apply.rs b/server/src/projection_apply.rs index 5644557a41b3e9497c7421b444155ae629fa79f1..9c8990a8af927f35d3344c8d0872a516aba56b86 100644 --- a/server/src/projection_apply.rs +++ b/server/src/projection_apply.rs @@ -1,22 +1,20 @@ //! Apply event-log records to the durable projection as precise point updates. //! //! Each batch of records lowers to reified durable writes (edge merges, child -//! links, voted-pair flags, recent-vote pushes, entity payloads) plus a cursor -//! advance, all committed in one atomic `DisableWal` batch. The cursor moving in -//! the same batch as the (non-idempotent) edge merges guarantees exactly-once -//! application across replay. +//! links, voted-pair flags, recent-vote pushes) plus a cursor advance, all +//! committed in one atomic `DisableWal` batch. The cursor moving in the same +//! batch as the (non-idempotent) edge merges guarantees exactly-once application +//! across replay. use std::collections::BTreeSet; use crate::{ - entity_store::EntityStore, event_log::EventLogError, events::{Event, EventRecord}, path_types::ItemId, projection_store::ProjectionStore, - reddit::entity_view_from_payload, reducer::VoteData, - storage_schema::{ensure_path_writes, entity_view_writes, vote_writes}, + storage_schema::{ensure_path_writes, vote_writes}, }; fn parse_event_id(id: &str) -> Result { @@ -38,7 +36,6 @@ fn parent_from_event_scope(scope: &str) -> ItemId { pub fn apply_records( projection_store: &ProjectionStore, - entity_store: &EntityStore, records: &[EventRecord], ) -> Result<(), EventLogError> { if records.is_empty() { @@ -79,14 +76,6 @@ pub fn apply_records( let parsed = parse_event_id(id)?; ensure_path_writes(&mut batch, &parsed); } - Event::EntityImported { id, payload, .. } => { - let parsed = parse_event_id(id)?; - let view = entity_view_from_payload(&parsed, payload); - entity_view_writes(&mut batch, &parsed, view.as_ref()); - entity_store - .put_in_batch(&mut batch, &parsed, payload) - .map_err(|e| EventLogError::Apply(e.to_string()))?; - } } last_seq = record.seq; } diff --git a/server/src/projection_store.rs b/server/src/projection_store.rs index 30ee478f953e80d7322bdbfa521ae559ff08fa51..8576d671f351004426207894ac35594ddb0f70cf 100644 --- a/server/src/projection_store.rs +++ b/server/src/projection_store.rs @@ -9,13 +9,16 @@ use durable::{Db, Durability, Write}; use crate::{ path_types::ItemId, - reducer::{GlobalTree, NodeState}, - storage_schema::{load_node_state, node, NodeSchemaFields, Store, StoreFields}, + reducer::{EntityData, GlobalTree, NodeState}, + storage_schema::{ + entity_content_clear_writes, entity_content_writes, load_node_state, node, NodeSchemaFields, + Store, StoreFields, + }, }; const PROJECTION_CURSOR_KEY: &str = "cursor"; const PROJECTION_SCHEMA_KEY: &str = "schema_version"; -const PROJECTION_SCHEMA_VERSION: u64 = 2; +const PROJECTION_SCHEMA_VERSION: u64 = 3; #[derive(Debug, thiserror::Error)] pub enum ProjectionStoreError { @@ -148,6 +151,45 @@ impl ProjectionStore { )?; Ok(()) } + + /// Cache Reddit display content outside the event log (must be evicted per policy). + pub fn put_ephemeral_content( + &self, + id: &ItemId, + view: &EntityData, + fetched_at: i64, + ) -> Result<(), ProjectionStoreError> { + let mut batch = self.db.batch(); + entity_content_writes(&mut batch, id, view, fetched_at); + batch + .commit_with(Durability::DisableWal) + .map_err(ProjectionStoreError::from)?; + Ok(()) + } + + /// Drop cached display content older than `cutoff_ms` (votes and tree structure remain). + pub fn evict_content_older_than(&self, cutoff_ms: i64) -> Result { + let keys = Store::root().nodes().keys(&self.db)?; + let mut batch = self.db.batch(); + let mut evicted = 0usize; + for key in keys { + let id = parse_node_key(&key)?; + let np = node(&id); + let Some(fetched_at) = np.fetched_at().get(&self.db)? else { + continue; + }; + if fetched_at > 0 && fetched_at < cutoff_ms { + entity_content_clear_writes(&mut batch, &id); + evicted += 1; + } + } + if evicted > 0 { + batch + .commit_with(Durability::DisableWal) + .map_err(ProjectionStoreError::from)?; + } + Ok(evicted) + } } fn parse_node_key(key: &str) -> Result { @@ -162,7 +204,7 @@ fn parse_node_key(key: &str) -> Result { #[cfg(test)] mod tests { use super::*; - use crate::{entity_store::EntityStore, events::Event, projection_apply}; + use crate::{events::Event, projection_apply, reducer::EntityData}; fn record(seq: u64, event: Event) -> crate::events::EventRecord { crate::events::EventRecord::new(seq, crate::events::event_timestamp(&event), event) @@ -172,7 +214,6 @@ mod tests { fn applies_and_loads_reducer_nodes() { let tmp = tempfile::tempdir().unwrap(); let db = Db::open(tmp.path()).unwrap(); - let entity_store = EntityStore::from_db(&db).unwrap(); let store = ProjectionStore::from_db(&db).unwrap(); let event = Event::VoteRecorded { @@ -183,7 +224,7 @@ mod tests { ratio_right: 1, scope: String::new(), }; - projection_apply::apply_records(&store, &entity_store, &[record(1, event)]).unwrap(); + projection_apply::apply_records(&store, &[record(1, event)]).unwrap(); assert_eq!(store.last_applied_event_count().unwrap(), 1); let loaded = store.load_tree().unwrap(); @@ -196,7 +237,6 @@ mod tests { fn hydrates_scope_with_child_nodes() { let tmp = tempfile::tempdir().unwrap(); let db = Db::open(tmp.path()).unwrap(); - let entity_store = EntityStore::from_db(&db).unwrap(); let store = ProjectionStore::from_db(&db).unwrap(); let event = Event::VoteRecorded { @@ -207,11 +247,36 @@ mod tests { ratio_right: 1, scope: String::new(), }; - projection_apply::apply_records(&store, &entity_store, &[record(1, event)]).unwrap(); + projection_apply::apply_records(&store, &[record(1, event)]).unwrap(); let scoped = store.scope_tree(&ItemId::root()).unwrap(); let root = scoped.get(&ItemId::root()).unwrap(); assert_eq!(root.children.len(), 2); assert!(scoped.get(&ItemId::opaque("alpha")).is_some()); } + + #[test] + fn evicts_stale_ephemeral_content() { + let tmp = tempfile::tempdir().unwrap(); + let db = Db::open(tmp.path()).unwrap(); + let store = ProjectionStore::from_db(&db).unwrap(); + let id = ItemId::from_url("https://reddit.com/r/rust").unwrap(); + store + .put_ephemeral_content( + &id, + &EntityData { + title: "Rust".into(), + author: None, + body_html: None, + thumb_url: None, + image_url: None, + link_url: None, + }, + 1_000, + ) + .unwrap(); + assert!(store.load_node(&id).unwrap().unwrap().data.is_some()); + assert_eq!(store.evict_content_older_than(2_000).unwrap(), 1); + assert!(store.load_node(&id).unwrap().unwrap().data.is_none()); + } } diff --git a/server/src/reddit.rs b/server/src/reddit.rs index 20b7f9e3f8be39268a1767d09f5cf81eaa6ae0df..a874814f8927192ee62cab2d0db1efd27dcd57b7 100644 --- a/server/src/reddit.rs +++ b/server/src/reddit.rs @@ -9,10 +9,13 @@ use serde_json::Value; use tokio::sync::{mpsc, oneshot}; use crate::{ - entity_store::EntityStore, events::Event, fetch::now_ms, journal::JournalClient, - path_types::ItemId, reducer::GlobalTree, + events::Event, fetch::now_ms, journal::JournalClient, + path_types::ItemId, projection_store::ProjectionStore, }; +/// Reddit display content must not be retained longer than this (API policy). +pub const REDDIT_CONTENT_TTL: std::time::Duration = std::time::Duration::from_secs(48 * 3600); + #[derive(Debug, Clone, PartialEq, Eq)] pub enum FetchJobResult { /// Number of entities written (1 for self, N for children). @@ -70,7 +73,11 @@ struct OAuthToken { } impl RedditBroker { - pub fn spawn(journal: JournalClient, config: RedditApiConfig) -> Self { + pub fn spawn( + journal: JournalClient, + projection_store: ProjectionStore, + config: RedditApiConfig, + ) -> Self { let (tx, rx) = mpsc::channel(100); let mut headers = header::HeaderMap::new(); @@ -93,7 +100,7 @@ impl RedditBroker { "reddit worker started" ); - tokio::spawn(reddit_worker(rx, journal, client, config)); + tokio::spawn(reddit_worker(rx, journal, projection_store, client, config)); Self { tx } } @@ -189,27 +196,50 @@ pub fn entity_view_from_payload( None } -pub fn apply_entity_import( - tree: &mut GlobalTree, - store: &EntityStore, - id: &ItemId, - payload: Value, -) -> Result<(), String> { - let view = entity_view_from_payload(id, &payload); - store.put(id, &payload).map_err(|e| e.to_string())?; - tree.apply_entity(id, view); - Ok(()) -} - fn notify(done: Option>, result: FetchJobResult) { if let Some(tx) = done { let _ = tx.send(result); } } +async fn import_fetched_payload( + kind: FetchKind, + fetch_id: &ItemId, + payload: Value, + projection_store: &ProjectionStore, + journal: &JournalClient, +) -> Result { + let fetched_at = now_ms(); + let imports: Vec<(ItemId, Value)> = match kind { + FetchKind::SelfEntity => vec![(fetch_id.clone(), payload)], + FetchKind::Children => parse_children(fetch_id, &payload), + }; + + for (id, child_payload) in &imports { + if let Some(view) = entity_view_from_payload(id, child_payload) { + projection_store + .put_ephemeral_content(id, &view, fetched_at) + .map_err(|e| e.to_string())?; + } + } + + let events: Vec = imports + .iter() + .map(|(id, _)| Event::NodeEnsured { + id: id.as_str().to_string(), + }) + .collect(); + let written = events.len(); + if !events.is_empty() { + journal.append_many(events).await?; + } + Ok(written) +} + async fn reddit_worker( mut rx: mpsc::Receiver, journal: JournalClient, + projection_store: ProjectionStore, client: Client, config: RedditApiConfig, ) { @@ -276,33 +306,20 @@ async fn reddit_worker( match outcome { Ok(FetchOutcome::Payload(payload)) => { - let imports: Vec<(ItemId, Value)> = match kind { - FetchKind::SelfEntity => vec![(fetch_id.clone(), payload)], - FetchKind::Children => parse_children(&fetch_id, &payload), - }; tracing::debug!( item = %fetch_id, ?kind, - count = imports.len(), - "reddit fetch got payload, importing" + "reddit fetch got payload, caching ephemerally" ); - let events: Vec = imports - .into_iter() - .map(|(child_id, child_payload)| Event::EntityImported { - id: child_id.as_str().to_string(), - ts: now_ms(), - payload: child_payload, - }) - .collect(); - let written = events.len(); - - match journal.append_many(events).await { + match import_fetched_payload(kind, &fetch_id, payload, &projection_store, &journal) + .await + { Err(e) => { - tracing::warn!(item = %fetch_id, err = %e, "reddit import journal failed"); + tracing::warn!(item = %fetch_id, err = %e, "reddit import failed"); notify(done, FetchJobResult::Failed(e)); } - Ok(()) => { + Ok(written) => { recently_fetched.insert(key.clone(), Instant::now()); current_delay = Duration::from_millis(600); tracing::info!(item = %fetch_id, ?kind, written, "reddit import complete"); diff --git a/server/src/reducer.rs b/server/src/reducer.rs index 1352a8f0771add3d868a1b30109b087a2a6dba6f..0c75c85150bb9e5f578bbadf58b3e43f8a80be4b 100644 --- a/server/src/reducer.rs +++ b/server/src/reducer.rs @@ -136,8 +136,7 @@ pub struct EntityData { #[derive(Debug, Clone, Default, Serialize, Deserialize)] pub struct NodeState { pub id: ItemId, - /// Domain-specific view derived from imported payload (e.g. Reddit title/author). - /// Raw JSON lives in [`crate::entity_store::EntityStore`]. + /// Ephemeral display view (Reddit title/author/etc.; not event-logged). pub data: Option, pub children: HashSet, pub local_ranking: GroupState, diff --git a/server/src/state.rs b/server/src/state.rs index e44849eec46123072b238afd40a1fdd51ce19bd9..247b9047a57956f76c4b6bef691662101e62a8f9 100644 --- a/server/src/state.rs +++ b/server/src/state.rs @@ -1,14 +1,14 @@ use std::{error::Error, sync::Arc}; use crate::{ - entity_store::EntityStore, event_log::EventLog, events::Event, + fetch::now_ms, journal::JournalClient, path_types::ItemId, projection_apply, projection_store::ProjectionStore, - reddit::{RedditApiConfig, RedditBroker}, + reddit::{RedditApiConfig, RedditBroker, REDDIT_CONTENT_TTL}, reducer::{GlobalTree, VoteData}, view_log::ViewLog, views::ViewStore, @@ -45,7 +45,6 @@ pub fn normalize_scope(raw: &str) -> String { async fn catch_up_projection( event_log: &EventLog, - entity_store: &EntityStore, projection_store: &ProjectionStore, ) -> Result<(), crate::event_log::EventLogError> { let after_seq = projection_store @@ -54,7 +53,7 @@ async fn catch_up_projection( let stats = event_log .replay_from(after_seq, |record| { - projection_apply::apply_records(projection_store, entity_store, &[record]) + projection_apply::apply_records(projection_store, &[record]) }) .await?; if after_seq > stats.last_seq { @@ -67,22 +66,34 @@ async fn catch_up_projection( Ok(()) } +fn spawn_content_evictor(projection_store: ProjectionStore) { + tokio::spawn(async move { + let mut interval = tokio::time::interval(std::time::Duration::from_secs(15 * 60)); + interval.tick().await; + loop { + interval.tick().await; + let cutoff = now_ms() - REDDIT_CONTENT_TTL.as_millis() as i64; + match projection_store.evict_content_older_than(cutoff) { + Ok(0) => {} + Ok(n) => tracing::info!(evicted = n, "reddit display content TTL eviction"), + Err(e) => tracing::warn!(err = %e, "reddit content TTL eviction failed"), + } + } + }); +} + pub async fn rebuild_projection( cfg: &AppConfig, ) -> Result> { let event_log = EventLog::new(cfg.event_log_path.clone()); let store_path = format!("{}/store", cfg.data_dir); let db = durable::Db::open(std::path::Path::new(&store_path))?; - let entity_store = EntityStore::from_db(&db)?; let projection_store = ProjectionStore::from_db(&db)?; - entity_store.reset()?; projection_store.reset()?; let stats = event_log - .replay(|record| { - projection_apply::apply_records(&projection_store, &entity_store, &[record]) - }) + .replay(|record| projection_apply::apply_records(&projection_store, &[record])) .await?; let cursor = projection_store.last_applied_event_count()?; if cursor != stats.last_seq { @@ -132,7 +143,6 @@ pub struct AppState { pub cfg: Arc, pub event_log: Arc, pub view_log: Arc, - pub entity_store: EntityStore, pub projection_store: ProjectionStore, pub views: ViewStore, journal: JournalClient, @@ -145,7 +155,6 @@ impl AppState { let view_log = Arc::new(ViewLog::new(cfg.views_log_path.clone())); let store_path = format!("{}/store", cfg.data_dir); let db = durable::Db::open(std::path::Path::new(&store_path))?; - let entity_store = EntityStore::from_db(&db)?; let projection_store = ProjectionStore::from_db(&db)?; let views = ViewStore::from_db(&db)?; @@ -157,22 +166,25 @@ impl AppState { } views.spawn_worker(view_log.clone()); - catch_up_projection(&event_log, &entity_store, &projection_store).await?; + catch_up_projection(&event_log, &projection_store).await?; let next_seq = event_log.last_sequence().await? + 1; let journal = JournalClient::spawn( event_log.clone(), - entity_store.clone(), projection_store.clone(), next_seq, ); - let reddit = RedditBroker::spawn(journal.clone(), RedditApiConfig::from_env()); + let reddit = RedditBroker::spawn( + journal.clone(), + projection_store.clone(), + RedditApiConfig::from_env(), + ); + spawn_content_evictor(projection_store.clone()); Ok(Self { cfg: Arc::new(cfg), event_log, view_log, - entity_store, projection_store, views, journal, @@ -248,54 +260,72 @@ impl AppState { mod tests { use super::{normalize_scope, parse_item_param, AppConfig, AppState}; use crate::{ - entity_store::EntityStore, event_log::EventLog, events::Event, path_types::ItemId, - projection_apply, projection_store::ProjectionStore, + event_log::EventLog, events::Event, path_types::ItemId, projection_apply, + projection_store::ProjectionStore, reducer::EntityData, }; - use serde_json::json; fn event_record(seq: u64, event: Event) -> crate::events::EventRecord { crate::events::EventRecord::new(seq, crate::events::event_timestamp(&event), event) } #[tokio::test] - async fn replay_entity_imported_restores_view() { + async fn rebuild_projection_drops_ephemeral_content() { let tmp = tempfile::tempdir().unwrap(); - let log_path = tmp.path().join("events.jsonl"); - let log = EventLog::new(log_path.to_string_lossy().into_owned()); - let payload = json!({"kind":"t5","data":{"title":"Rust","display_name":"rust"}}); - let event = Event::EntityImported { - id: "https://reddit.com/r/rust".into(), - ts: 1, - payload: payload.clone(), - }; - log.append(&event_record(1, event)).await.unwrap(); + let data_dir = tmp.path().to_string_lossy().into_owned(); + let log = EventLog::new(format!("{data_dir}/events.jsonl")); + log.append(&event_record( + 1, + Event::NodeEnsured { + id: "https://reddit.com/r/rust".into(), + }, + )) + .await + .unwrap(); let db = durable::Db::open(tmp.path().join("store")).unwrap(); - let entity_store = EntityStore::from_db(&db).unwrap(); let projection_store = ProjectionStore::from_db(&db).unwrap(); - super::catch_up_projection(&log, &entity_store, &projection_store) + super::catch_up_projection(&log, &projection_store) .await .unwrap(); - let tree = projection_store - .scope_tree(&ItemId::parse("https://reddit.com/r/rust").unwrap()) - .unwrap(); - let node = tree - .get(&ItemId::parse("https://reddit.com/r/rust").unwrap()) - .unwrap(); - assert_eq!(node.data.as_ref().unwrap().title, "Rust"); - let stored = entity_store - .get(&ItemId::parse("https://reddit.com/r/rust").unwrap()) - .unwrap() + let id = ItemId::parse("https://reddit.com/r/rust").unwrap(); + projection_store + .put_ephemeral_content( + &id, + &EntityData { + title: "Rust".into(), + author: None, + body_html: None, + thumb_url: None, + image_url: None, + link_url: None, + }, + 1, + ) .unwrap(); - assert_eq!(stored["data"]["display_name"], "rust"); + assert!(projection_store.load_node(&id).unwrap().unwrap().data.is_some()); + drop(projection_store); + drop(db); + + super::rebuild_projection(&AppConfig { + data_dir: data_dir.clone(), + event_log_path: format!("{data_dir}/events.jsonl"), + views_log_path: format!("{data_dir}/views.jsonl"), + port: 0, + }) + .await + .unwrap(); + + let db = durable::Db::open(tmp.path().join("store")).unwrap(); + let projection_store = ProjectionStore::from_db(&db).unwrap(); + let node = projection_store.load_node(&id).unwrap().unwrap(); + assert!(node.data.is_none()); } #[tokio::test] - async fn rebuild_projection_restores_nodes_payloads_and_cursor_from_jsonl() { + async fn rebuild_projection_restores_structure_and_cursor_from_jsonl() { let tmp = tempfile::tempdir().unwrap(); let data_dir = tmp.path().to_string_lossy().into_owned(); let log = EventLog::new(format!("{data_dir}/events.jsonl")); - let payload = json!({"kind":"t5","data":{"title":"Rust","display_name":"rust"}}); log.append_batch(&[ event_record( 1, @@ -305,16 +335,8 @@ mod tests { ), event_record( 2, - Event::EntityImported { - id: "https://reddit.com/r/rust".into(), - ts: 2, - payload: payload.clone(), - }, - ), - event_record( - 3, Event::VoteRecorded { - ts: 3, + ts: 2, a: "alpha".into(), b: "beta".into(), ratio_left: 2, @@ -328,11 +350,9 @@ mod tests { { let db = durable::Db::open(tmp.path().join("store")).unwrap(); - let entity_store = EntityStore::from_db(&db).unwrap(); let projection_store = ProjectionStore::from_db(&db).unwrap(); projection_apply::apply_records( &projection_store, - &entity_store, &[event_record( 1, Event::NodeEnsured { @@ -352,13 +372,12 @@ mod tests { }) .await .unwrap(); - assert_eq!(stats.applied, 3); - assert_eq!(stats.last_seq, 3); + assert_eq!(stats.applied, 2); + assert_eq!(stats.last_seq, 2); let db = durable::Db::open(tmp.path().join("store")).unwrap(); - let entity_store = EntityStore::from_db(&db).unwrap(); let projection_store = ProjectionStore::from_db(&db).unwrap(); - assert_eq!(projection_store.last_applied_event_count().unwrap(), 3); + assert_eq!(projection_store.last_applied_event_count().unwrap(), 2); let tree = projection_store.scope_tree(&ItemId::root()).unwrap(); let root = tree.get(&ItemId::root()).unwrap(); assert!(root.children.contains(&ItemId::parse("alpha").unwrap())); @@ -366,11 +385,6 @@ mod tests { .load_node(&ItemId::parse("https://reddit.com/r/stale").unwrap()) .unwrap() .is_none()); - let stored = entity_store - .get(&ItemId::parse("https://reddit.com/r/rust").unwrap()) - .unwrap() - .unwrap(); - assert_eq!(stored["data"]["display_name"], "rust"); } #[tokio::test] @@ -389,12 +403,9 @@ mod tests { { let db = durable::Db::open(tmp.path().join("store")).unwrap(); - let entity_store = EntityStore::from_db(&db).unwrap(); let projection_store = ProjectionStore::from_db(&db).unwrap(); - // Advance the projection cursor to 2 while the log tail is only 1. projection_apply::apply_records( &projection_store, - &entity_store, &[event_record( 2, Event::NodeEnsured { @@ -403,7 +414,7 @@ mod tests { )], ) .unwrap(); - let err = super::catch_up_projection(&log, &entity_store, &projection_store) + let err = super::catch_up_projection(&log, &projection_store) .await .unwrap_err(); assert!(err @@ -432,10 +443,9 @@ mod tests { .unwrap(); let db = durable::Db::open(tmp.path().join("store")).unwrap(); - let entity_store = EntityStore::from_db(&db).unwrap(); let projection_store = ProjectionStore::from_db(&db).unwrap(); - super::catch_up_projection(&log, &entity_store, &projection_store) + super::catch_up_projection(&log, &projection_store) .await .unwrap(); assert_eq!(projection_store.last_applied_event_count().unwrap(), 1); @@ -444,7 +454,7 @@ mod tests { let first_edge_total: f64 = first_root.local_ranking.edges.values().sum(); assert_eq!(first_edge_total, 3.0); - super::catch_up_projection(&log, &entity_store, &projection_store) + super::catch_up_projection(&log, &projection_store) .await .unwrap(); assert_eq!(projection_store.last_applied_event_count().unwrap(), 1); @@ -556,9 +566,8 @@ mod tests { .unwrap(); { let db = durable::Db::open(tmp.path().join("store")).unwrap(); - let entity_store = EntityStore::from_db(&db).unwrap(); let projection_store = ProjectionStore::from_db(&db).unwrap(); - super::catch_up_projection(&log, &entity_store, &projection_store) + super::catch_up_projection(&log, &projection_store) .await .unwrap(); } @@ -605,9 +614,8 @@ mod tests { .unwrap(); { let db = durable::Db::open(tmp.path().join("store")).unwrap(); - let entity_store = EntityStore::from_db(&db).unwrap(); let projection_store = ProjectionStore::from_db(&db).unwrap(); - super::catch_up_projection(&log, &entity_store, &projection_store) + super::catch_up_projection(&log, &projection_store) .await .unwrap(); } diff --git a/server/src/storage_dto.rs b/server/src/storage_dto.rs index de5ed995797ae306d3a71394a4d9d245ffd5771d..9dfb13c53efe4389277625a6ab3bfc18f566a453 100644 --- a/server/src/storage_dto.rs +++ b/server/src/storage_dto.rs @@ -3,17 +3,15 @@ //! Node structure (children, edges, voted pairs, recent votes) is no longer a //! single blob — it lives as point-addressable durable collections (see //! [`crate::storage_schema`]). This module only defines the small leaf values: -//! the derived entity view, raw entity payloads, and individual votes. +//! ephemeral entity views and individual votes. use serde::{Deserialize, Serialize}; -use serde_json::Value; use crate::{ path_types::ItemId, reducer::{EntityData, VoteData}, }; -pub const ENTITY_RECORD_VERSION: u32 = 1; pub const VOTE_RECORD_VERSION: u32 = 1; pub const ENTITY_DATA_VERSION: u32 = 1; @@ -29,14 +27,7 @@ impl Versioned { } } -pub type StoredEntityRecord = Versioned; - -#[derive(Debug, Clone, Serialize, Deserialize)] -pub struct StoredEntityV1 { - pub json: Value, -} - -/// Derived entity view stored at a node's `data` leaf. +/// Derived entity view stored at a node's `data` leaf (ephemeral; not logged). #[derive(Debug, Clone, Serialize, Deserialize)] pub struct StoredEntityDataV1 { pub version: u32, @@ -63,25 +54,6 @@ pub struct StoredVoteV1 { pub thread_tag: String, } -pub fn encode_entity_payload(payload: &Value) -> StoredEntityRecord { - Versioned::new( - ENTITY_RECORD_VERSION, - StoredEntityV1 { - json: payload.clone(), - }, - ) -} - -pub fn decode_entity_payload(record: StoredEntityRecord) -> Result { - if record.version != ENTITY_RECORD_VERSION { - return Err(format!( - "unsupported entity record version: {}", - record.version - )); - } - Ok(record.payload.json) -} - pub fn encode_entity_data(data: &EntityData) -> StoredEntityDataV1 { StoredEntityDataV1 { version: ENTITY_DATA_VERSION, diff --git a/server/src/storage_schema.rs b/server/src/storage_schema.rs index 76bf7bc74a2c5ef4f78678a333b7661778f5b835..bd26e665e084b95b10fdfff091c31e8dc84d07b8 100644 --- a/server/src/storage_schema.rs +++ b/server/src/storage_schema.rs @@ -15,7 +15,7 @@ use crate::{ reducer::{EntityData, GroupState, NodeState, VoteData}, storage_dto::{ decode_entity_data, decode_vote, encode_entity_data, encode_vote, parse_stored_id, - StoredEntityDataV1, StoredEntityRecord, StoredVoteV1, + StoredEntityDataV1, StoredVoteV1, }, }; @@ -40,17 +40,17 @@ pub struct NodeSchema { pub voted_pairs: Map>, /// Recent votes, newest at the front (capped on write). pub recent_votes: Deque>, + /// When ephemeral Reddit display content was last fetched (ms); absent after eviction. + pub fetched_at: Leaf, } -/// The single database root: nodes, raw payloads, view counts, and per-concern -/// metadata maps (cursors and schema versions). +/// The single database root: nodes, view counts, and per-concern metadata maps +/// (cursors and schema versions). #[derive(Durable)] #[allow(dead_code)] pub struct Store { pub nodes: Map, pub proj_meta: Map>, - pub entities: Map>, - pub entity_meta: Map>, pub view_counts: Map>, pub view_meta: Map>, } @@ -264,12 +264,17 @@ pub fn vote_writes( Ok(()) } -/// Reified writes for an imported entity view (node data + path wiring). -pub fn entity_view_writes(batch: &mut Batch, id: &ItemId, view: Option<&EntityData>) { +/// Reified writes for ephemeral Reddit display content (not event-logged). +pub fn entity_content_writes(batch: &mut Batch, id: &ItemId, view: &EntityData, fetched_at: i64) { ensure_path_writes(batch, id); - if let Some(view) = view { - batch.write(node(id).data().set(&encode_entity_data(view))); - } + batch.write(node(id).data().set(&encode_entity_data(view))); + batch.write(node(id).fetched_at().set(&fetched_at)); +} + +/// Clear cached display content for one node (structure/votes are untouched). +pub fn entity_content_clear_writes(batch: &mut Batch, id: &ItemId) { + batch.write(node(id).data().delete()); + batch.write(node(id).fetched_at().delete()); } #[cfg(test)] diff --git a/test/reddit_import.clj b/test/reddit_import.clj index b476488526252c13fd73bdda76e5201678e4a714..6d8c5bca7ebad0b5abfddecd4ea48cc09d738ebe 100644 --- a/test/reddit_import.clj +++ b/test/reddit_import.clj @@ -59,9 +59,10 @@ "curl" "-sf" browse-url)) log (slurp (io/file log-path))] (is (str/includes? after "The Rust Programming Language")) - (is (str/includes? log "\"type\":\"entity_imported\"")) - (is (str/includes? log "\"subscribers\":350000")) - (is (str/includes? log "\"display_name\":\"rust\""))) + (is (str/includes? log "\"type\":\"node_ensured\"")) + (is (not (str/includes? log "\"subscribers\""))) + (is (not (str/includes? log "\"display_name\""))) + (is (not (str/includes? log "entity_imported")))) (let [children-sse (curl-fetch-ui-sse app-base "reddit.com/r/rust" "children")] (is (zero? (:exit children-sse)) "POST /ui fetch_entity (children) SSE succeeds") (is (str/includes? (:out children-sse) "Idiomorph.morph")) @@ -71,10 +72,12 @@ log2 (slurp (io/file log-path))] (is (str/includes? after-children "Announcing Rust 1.99")) (is (str/includes? after-children "Unranked")) - (is (str/includes? log2 "announcing_rust_199"))))) + (is (str/includes? log2 "\"type\":\"node_ensured\"")) + (is (str/includes? log2 "/comments/")) + (is (not (str/includes? log2 "\"selftext\"")))))) (deftest reddit-fetch-via-mock-api - (testing "Fetch more queues import; event log stores full payload; page shows title" + (testing "Fetch caches display content ephemerally; log records structure only" (let [root (repo-root) fixtures (mock-reddit/fixtures-dir root) data-dir (.getAbsolutePath