You are a constitutional council ranking individual git commits for ownership allocation. Compare these two commits. Decide which contributed more lasting value to the project. Judge substance, not spectacle: - Prefer correct, lasting design and real bugfixes over churn, formatting, renames, or generated noise. - Prefer clarity and necessity over sheer line count. A small precise change can beat a large diffuse one. - Do not favor a side merely because its patch is longer or noisier. - Weight what the change does for the project, not the contributor's name. Return ONLY a JSON object: {"winner": "A" or "B", "ratio": "N:M", "explanation": "..."} The explanation must cite concrete differences in the patches (1-3 sentences). Side A — contributor: tommy-mor Side A — commit message: [0366806e] Color rank rows by vote mass within each group, not list position. Min–max normalization keeps similar scores visually close while still using the full gradient as groups grow and absolute mass dilutes. Co-authored-by: Cursor Side A — unified diff (full patch): diff --git a/server/src/html/mod.rs b/server/src/html/mod.rs index a58cbbee3490a08a625cb06df06848c59a615d65..4eff2e19ed4d303ff8e80c1eabd8a15b4990e643 100644 --- a/server/src/html/mod.rs +++ b/server/src/html/mod.rs @@ -264,12 +264,19 @@ pub fn scope_theme_style(parent: &ItemId) -> String { ) } -fn rank_row_style(parent: &ItemId, ordinal: usize, total: usize) -> String { - let t = if total <= 1 { - 0.0 - } else { - ordinal as f64 / (total - 1) as f64 - }; +/// Map vote mass to gradient position using the group's score range, not raw mass or +/// list position. Vote mass sums to 1 across the component, so absolute values dilute +/// as N grows; min–max within the visible list preserves similar scores → similar colors. +fn score_gradient_t(score: f64, min_score: f64, max_score: f64) -> f64 { + let spread = max_score - min_score; + if spread < 1e-9 { + return 0.5; + } + ((max_score - score) / spread).clamp(0.0, 1.0) +} + +fn rank_row_style(parent: &ItemId, score: f64, min_score: f64, max_score: f64) -> String { + let t = score_gradient_t(score, min_score, max_score); let base_hue = scope_base_hue(parent); let hue = (base_hue + 118.0 * t) % 360.0; let lightness = 0.74 - 0.34 * t; @@ -295,14 +302,15 @@ fn rank_list( highlighted: &HashSet, tree: &GlobalTree, ) -> Markup { - let group_len = items.len(); + let min_score = items.iter().map(|r| r.score).fold(f64::INFINITY, f64::min); + let max_score = items.iter().map(|r| r.score).fold(f64::NEG_INFINITY, f64::max); html! { @if !items.is_empty() { h3 class="rank-heading muted small" { (label) } ol class="rank-list" { @for (i, r) in items.iter().enumerate() { @let href = item_href(&r.item); - @let style = rank_row_style(parent, i, group_len); + @let style = rank_row_style(parent, r.score, min_score, max_score); @let class = rank_row_class(&r.item, highlighted); li class=(class) data-rank-item=(r.item.as_str()) @@ -517,19 +525,37 @@ pub async fn browse(State(state): State, uri: Uri) -> impl IntoRespons #[cfg(test)] mod tests { - use super::{rank_row_style, SORTER_UI_JS}; + use super::{rank_row_style, score_gradient_t, SORTER_UI_JS}; use crate::path_types::ItemId; #[test] - fn rank_row_style_gradients_per_group_not_globally() { + fn score_gradient_t_uses_group_range_not_absolute_mass() { + assert!((score_gradient_t(0.12, 0.08, 0.12) - 0.0).abs() < 1e-9); + assert!((score_gradient_t(0.08, 0.08, 0.12) - 1.0).abs() < 1e-9); + // Raw 12% mass would map near the dark end globally; within this group it's the top. + assert!(score_gradient_t(0.12, 0.08, 0.12) < score_gradient_t(0.12, 0.0, 1.0)); + } + + #[test] + fn score_gradient_t_similar_scores_similar_t() { + let a = score_gradient_t(0.41, 0.20, 0.60); + let b = score_gradient_t(0.40, 0.20, 0.60); + assert!((a - b).abs() < 0.05); + assert!((a - score_gradient_t(0.60, 0.20, 0.60)).abs() > 0.3); + } + + #[test] + fn score_gradient_t_tied_scores_neutral() { + assert!((score_gradient_t(0.25, 0.25, 0.25) - 0.5).abs() < 1e-9); + } + + #[test] + fn rank_row_style_same_inputs_same_color() { let parent = ItemId::opaque("test-scope"); - let first_in_four = rank_row_style(&parent, 0, 4); - let last_in_four = rank_row_style(&parent, 3, 4); - let first_in_two = rank_row_style(&parent, 0, 2); - let last_in_two = rank_row_style(&parent, 1, 2); - assert_eq!(first_in_four, first_in_two); - assert_eq!(last_in_four, last_in_two); - assert_ne!(first_in_four, last_in_four); + assert_eq!( + rank_row_style(&parent, 0.33, 0.20, 0.60), + rank_row_style(&parent, 0.33, 0.20, 0.60), + ); } #[test] Side B — contributor: tommy-mor Side B — commit message: [1d14ff09] Ephemeral Reddit content; log structure only (#46) * Keep Reddit content ephemeral; log structure only Remove EntityImported and EntityStore. Reddit fetches write display content directly to the projection with a fetched_at timestamp, while the event log records NodeEnsured for discovered identities only. A background task evicts cached display content after 48 hours. Votes, tree structure, and ItemIds remain in the log and projection. Co-authored-by: tommy * Fix reddit import test assertions and Clojure syntax Co-authored-by: tommy --------- Co-authored-by: Cursor Agent Side B — unified diff (full patch): 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