diff --git a/server/src/bin/storage_bench.rs b/server/src/bin/storage_bench.rs index 3d988416ad36b27a7d3dc84280cfdbdcafa43e69..cc5cd78ba1e02a1e6f92aa503677f2d329bb5993 100644 --- a/server/src/bin/storage_bench.rs +++ b/server/src/bin/storage_bench.rs @@ -29,14 +29,7 @@ async fn main() -> Result<(), Box> { for chunk_start in (0..opts.events).step_by(opts.batch_size) { let chunk_end = (chunk_start + opts.batch_size).min(opts.events); let events = (chunk_start..chunk_end) - .map(|i| Event::VoteRecorded { - ts: i as i64, - a: format!("item-{i}"), - b: format!("item-{}", i + 1), - ratio_left: 2, - ratio_right: 1, - scope: String::new(), - }) + .map(|i| Event::vote_recorded(i as i64, format!("item-{i}"), format!("item-{}", i + 1), 2, 1, "")) .collect(); journal.append_many(events).await?; } diff --git a/server/src/event_log.rs b/server/src/event_log.rs index 2c9290b6fdbf2c2ad1c0f1ffd7374b2d9cc97f36..8e5684cb1378e80c45bcf743d2fdc48402dd4dc8 100644 --- a/server/src/event_log.rs +++ b/server/src/event_log.rs @@ -206,14 +206,7 @@ mod tests { .unwrap(); log.append(&sample_record( 2, - Event::VoteRecorded { - ts: 1, - a: "a".into(), - b: "b".into(), - ratio_left: 2, - ratio_right: 1, - scope: String::new(), - }, + Event::vote_recorded(1, "a", "b", 2, 1, ""), )) .await .unwrap(); diff --git a/server/src/events.rs b/server/src/events.rs index d76c3bb4277216b0d39c9422ba7a50db10a95e05..8a166d49b4f26835fbc2b58cb1f4bdbf002763b8 100644 --- a/server/src/events.rs +++ b/server/src/events.rs @@ -1,7 +1,7 @@ use serde::{Deserialize, Serialize}; /// Schema version for JSONL log records. Bump when event semantics change. -pub const CURRENT_LOG_SCHEMA: u32 = 1; +pub const CURRENT_LOG_SCHEMA: u32 = 2; /// One JSONL line: schema envelope around a payload event. #[derive(Debug, Clone, Serialize, Deserialize)] @@ -51,9 +51,33 @@ pub enum Event { b: String, ratio_left: i32, ratio_right: i32, - #[serde(default)] scope: String, + pseudonym: String, + trust_weight: f64, }, /// Register a node path in the fractal tree (no external fetch). NodeEnsured { id: String }, } + +impl Event { + /// Construct a vote event with the default dev pseudonym (tests and benches). + pub fn vote_recorded( + ts: i64, + a: impl Into, + b: impl Into, + ratio_left: i32, + ratio_right: i32, + scope: impl Into, + ) -> Self { + Self::VoteRecorded { + ts, + a: a.into(), + b: b.into(), + ratio_left, + ratio_right, + scope: scope.into(), + pseudonym: crate::identity::DEFAULT_PSEUDONYM.to_string(), + trust_weight: 1.0, + } + } +} diff --git a/server/src/html/vote.rs b/server/src/html/vote.rs index 8cf2b0d4d8bd58cc51cc27c9046fa7f7728a7017..bf82aef3ad9c5f4e9e877c47dab27beb29a80b8f 100644 --- a/server/src/html/vote.rs +++ b/server/src/html/vote.rs @@ -336,6 +336,7 @@ pub async fn vote_page( #[cfg(test)] mod polarity_tests { use super::*; + use crate::identity::{DEFAULT_PSEUDONYM, TEST_ACTOR_UUID}; use crate::ranking::ranked_items; use crate::reducer::GlobalTree; @@ -351,11 +352,11 @@ mod polarity_tests { let right = id("right_item"); // Stored a == page left: keep order. - let v1 = VoteData::from_recorded(1, left.as_str(), right.as_str(), 9, 1).unwrap(); + let v1 = VoteData::from_event(1, left.as_str(), right.as_str(), 9, 1, DEFAULT_PSEUDONYM.to_string(), 1.0).unwrap(); assert_eq!(ratios_for_page(&v1, &left, &right), (9, 1)); // Stored a == page right: swap so left stays left. - let v2 = VoteData::from_recorded(2, right.as_str(), left.as_str(), 9, 1).unwrap(); + let v2 = VoteData::from_event(2, right.as_str(), left.as_str(), 9, 1, DEFAULT_PSEUDONYM.to_string(), 1.0).unwrap(); assert_eq!(ratios_for_page(&v2, &left, &right), (1, 9)); } @@ -383,9 +384,9 @@ mod polarity_tests { let right = id("right_item"); // Slider dragged left yields e.g. 9:1 with a = left item. - let vote = VoteData::from_recorded(1, left.as_str(), right.as_str(), 9, 1).unwrap(); + let vote = VoteData::from_event(1, left.as_str(), right.as_str(), 9, 1, DEFAULT_PSEUDONYM.to_string(), 1.0).unwrap(); let mut tree = GlobalTree::new(); - tree.apply_vote(&parent, vote); + tree.apply_vote(&parent, vote, TEST_ACTOR_UUID); let group = &tree.get(&parent).unwrap().local_ranking; let ranked = ranked_items(group); diff --git a/server/src/identity.rs b/server/src/identity.rs new file mode 100644 index 0000000000000000000000000000000000000000..621d505436912246ec0830ff20c3473f0c4665ce --- /dev/null +++ b/server/src/identity.rs @@ -0,0 +1,35 @@ +//! Actor identity: pseudonym display names mapped to stable UUIDs for vote dedup. + +use durable::Db; + +use crate::storage_schema::{Store, StoreFields}; + +/// Default pseudonym until session/auth UI exists. +pub const DEFAULT_PSEUDONYM: &str = "anon"; + +/// UUID for the default single-user dev principal. +pub const DEFAULT_ACTOR_UUID: &str = "00000000-0000-0000-0000-000000000001"; + +/// UUID for in-memory unit tests. +pub const TEST_ACTOR_UUID: &str = "00000000-0000-0000-0000-000000000099"; + +/// Resolve the trust anchor for a pseudonym (must exist in the pseudonyms map). +pub fn resolve_actor_uuid(db: &Db, pseudonym: &str) -> Result { + Store::root() + .pseudonyms() + .key(&pseudonym.to_string()) + .get(db) + .map_err(|e| e.to_string())? + .ok_or_else(|| format!("unknown pseudonym: {pseudonym}")) +} + +/// Ensure the default pseudonym → UUID mapping exists (operational seed, not event-logged). +pub fn seed_default_pseudonym(db: &Db) -> Result<(), durable::Error> { + let path = Store::root() + .pseudonyms() + .key(&DEFAULT_PSEUDONYM.to_string()); + if path.get(db)?.is_none() { + db.run(path.set(&DEFAULT_ACTOR_UUID.to_string()), durable::Durability::SyncWal)?; + } + Ok(()) +} diff --git a/server/src/lib.rs b/server/src/lib.rs index 3dfc7c8acb8ed61bb73ade63e72768e402042cc5..da6e33e3dd6d79967bbee7a2708a7d982c93b88c 100644 --- a/server/src/lib.rs +++ b/server/src/lib.rs @@ -4,6 +4,7 @@ pub mod events; pub mod fetch; pub mod form_template; pub mod html; +pub mod identity; pub mod journal; pub mod pair; pub mod parser; diff --git a/server/src/pair.rs b/server/src/pair.rs index 2277e32edf6e0024687d6fe2b9d1a9b84c0b34c8..42a1b1eb2adf16730d34d0fe23c13d5a75d7ba27 100644 --- a/server/src/pair.rs +++ b/server/src/pair.rs @@ -360,8 +360,17 @@ impl PairError { #[cfg(test)] mod tests { use super::*; + use crate::identity::{DEFAULT_PSEUDONYM, TEST_ACTOR_UUID}; use crate::reducer::{GlobalTree, VoteData}; + fn test_vote(ts: i64, a: &str, b: &str, l: i32, r: i32) -> VoteData { + VoteData::from_event(ts, a, b, l, r, DEFAULT_PSEUDONYM.to_string(), 1.0).unwrap() + } + + fn apply(tree: &mut GlobalTree, parent: &ItemId, vote: VoteData) { + tree.apply_vote(parent, vote, TEST_ACTOR_UUID); + } + fn seed_children(parent: &ItemId, ids: &[&str]) -> GlobalTree { let mut tree = GlobalTree::new(); tree.ensure_path(parent); @@ -382,22 +391,13 @@ mod tests { #[test] fn zero_weight_vote_leaves_pair_available_for_suggestion() { let parent = ItemId::parse("https://reddit.com/r/rust").unwrap(); - let mut tree = seed_children( + let tree = seed_children( &parent, &[ "https://reddit.com/r/rust/a", "https://reddit.com/r/rust/b", ], ); - let noop = VoteData::from_recorded( - 1, - "https://reddit.com/r/rust/a", - "https://reddit.com/r/rust/b", - 0, - 0, - ) - .unwrap(); - tree.apply_vote(&parent, noop); let group = tree.get(&parent).unwrap().local_ranking.clone(); let pool = children_of(&tree, &parent); assert!(!pair_is_voted(&group, &pool[0], &pool[1])); @@ -415,9 +415,8 @@ mod tests { "https://reddit.com/r/rust/c", ], ); - let vote = - VoteData::from_recorded(1, "https://reddit.com/r/rust/a", "https://reddit.com/r/rust/b", 2, 1).unwrap(); - tree.apply_vote(&parent, vote); + let vote = test_vote(1, "https://reddit.com/r/rust/a", "https://reddit.com/r/rust/b", 2, 1); + apply(&mut tree, &parent, vote); let group = tree.get(&parent).unwrap().local_ranking.clone(); let pool = children_of(&tree, &parent); let (l, r) = suggest_next_pair_in_pool(&group, &pool, None).unwrap(); @@ -438,12 +437,10 @@ mod tests { "https://reddit.com/r/rust/d", ], ); - let ab = - VoteData::from_recorded(1, "https://reddit.com/r/rust/a", "https://reddit.com/r/rust/b", 2, 1).unwrap(); - let cd = - VoteData::from_recorded(2, "https://reddit.com/r/rust/c", "https://reddit.com/r/rust/d", 2, 1).unwrap(); - tree.apply_vote(&parent, ab); - tree.apply_vote(&parent, cd); + let ab = test_vote(1, "https://reddit.com/r/rust/a", "https://reddit.com/r/rust/b", 2, 1); + let cd = test_vote(2, "https://reddit.com/r/rust/c", "https://reddit.com/r/rust/d", 2, 1); + apply(&mut tree, &parent, ab); + apply(&mut tree, &parent, cd); let group = tree.get(&parent).unwrap().local_ranking.clone(); let pool = children_of(&tree, &parent); let pair = suggest_next_pair_in_pool(&group, &pool, None).unwrap(); @@ -468,9 +465,8 @@ mod tests { "https://reddit.com/r/rust/e", ], ); - let ab = - VoteData::from_recorded(1, "https://reddit.com/r/rust/a", "https://reddit.com/r/rust/b", 2, 1).unwrap(); - tree.apply_vote(&parent, ab); + let ab = test_vote(1, "https://reddit.com/r/rust/a", "https://reddit.com/r/rust/b", 2, 1); + apply(&mut tree, &parent, ab); let group = tree.get(&parent).unwrap().local_ranking.clone(); let pool = children_of(&tree, &parent); let pair = suggest_next_pair_in_pool(&group, &pool, None).unwrap(); @@ -498,9 +494,8 @@ mod tests { "https://reddit.com/r/rust/c", ], ); - let ab = - VoteData::from_recorded(1, "https://reddit.com/r/rust/a", "https://reddit.com/r/rust/b", 2, 1).unwrap(); - tree.apply_vote(&parent, ab); + let ab = test_vote(1, "https://reddit.com/r/rust/a", "https://reddit.com/r/rust/b", 2, 1); + apply(&mut tree, &parent, ab); let group = tree.get(&parent).unwrap().local_ranking.clone(); let pool = children_of(&tree, &parent); let pair = suggest_next_pair_in_pool(&group, &pool, None).unwrap(); @@ -524,8 +519,8 @@ mod tests { ("https://reddit.com/r/rust/a", "https://reddit.com/r/rust/b", 3, 1), ("https://reddit.com/r/rust/a", "https://reddit.com/r/rust/c", 2, 1), ] { - let v = VoteData::from_recorded(1, a, b, l, r).unwrap(); - tree.apply_vote(&parent, v); + let v = test_vote(1, a, b, l, r); + apply(&mut tree, &parent, v); } let group = tree.get(&parent).unwrap().local_ranking.clone(); let pool = children_of(&tree, &parent); @@ -552,8 +547,8 @@ mod tests { ("https://reddit.com/r/rust/b", "https://reddit.com/r/rust/c", 2, 1), ("https://reddit.com/r/rust/a", "https://reddit.com/r/rust/c", 2, 1), ] { - let v = VoteData::from_recorded(1, a, b, l, r).unwrap(); - tree.apply_vote(&parent, v); + let v = test_vote(1, a, b, l, r); + apply(&mut tree, &parent, v); } let group = tree.get(&parent).unwrap().local_ranking.clone(); let pool = children_of(&tree, &parent); diff --git a/server/src/projection_apply.rs b/server/src/projection_apply.rs index ad404bacb8bcdd5ae0e682cff97f974fd44528ea..f2859c7e0e718c5bd3d332224eb05d7a8c65a816 100644 --- a/server/src/projection_apply.rs +++ b/server/src/projection_apply.rs @@ -1,14 +1,13 @@ //! 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) 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. +//! Each batch of records lowers to reified durable writes (uuid vote upserts, +//! child links, recent-vote appends) plus a cursor advance, all committed in +//! one atomic `DisableWal` batch. use crate::{ event_log::EventLogError, events::{Event, EventRecord}, + identity::resolve_actor_uuid, path_types::ItemId, projection_store::ProjectionStore, reducer::VoteData, @@ -53,20 +52,31 @@ pub fn apply_records( ratio_left, ratio_right, scope, + pseudonym, + trust_weight, } => { - let vote = VoteData::from_recorded(*ts, a, b, *ratio_left, *ratio_right) - .ok_or_else(|| EventLogError::Apply(format!("invalid vote event: {a} vs {b}")))?; - let parent = parent_from_event_scope(scope); - vote_writes( - &mut batch, - &parent, - vote.a.as_str(), - vote.b.as_str(), - *ratio_left, - *ratio_right, + let left = (*ratio_left).max(0); + let right = (*ratio_right).max(0); + if left == 0 && right == 0 { + return Err(EventLogError::Apply(format!( + "invalid vote event: zero weights ({a} vs {b})" + ))); + } + let vote = VoteData::from_event( *ts, + a, + b, + left, + right, + pseudonym.clone(), + *trust_weight, ) - .map_err(|e| EventLogError::Apply(e.to_string()))?; + .ok_or_else(|| EventLogError::Apply(format!("invalid vote event: {a} vs {b}")))?; + let actor_uuid = resolve_actor_uuid(db, pseudonym) + .map_err(|e| EventLogError::Apply(e))?; + let parent = parent_from_event_scope(scope); + vote_writes(&mut batch, &parent, &vote, &actor_uuid) + .map_err(|e| EventLogError::Apply(e.to_string()))?; } Event::NodeEnsured { id } => { let parsed = parse_event_id(id)?; diff --git a/server/src/projection_store.rs b/server/src/projection_store.rs index 9a8953d010029d3639dc3987687554bab8b7663e..27fc0bb0519d9f035dceec1fc8b0fa74b00b6b40 100644 --- a/server/src/projection_store.rs +++ b/server/src/projection_store.rs @@ -18,7 +18,7 @@ use crate::{ const PROJECTION_CURSOR_KEY: &str = "cursor"; const PROJECTION_SCHEMA_KEY: &str = "schema_version"; -const PROJECTION_SCHEMA_VERSION: u64 = 4; +const PROJECTION_SCHEMA_VERSION: u64 = 5; #[derive(Debug, thiserror::Error)] pub enum ProjectionStoreError { @@ -51,6 +51,8 @@ impl ProjectionStore { if version != Some(PROJECTION_SCHEMA_VERSION) { store.reset()?; } + crate::identity::seed_default_pseudonym(db) + .map_err(ProjectionStoreError::Durable)?; Ok(store) } @@ -206,14 +208,7 @@ mod tests { let db = Db::open(tmp.path()).unwrap(); let store = ProjectionStore::from_db(&db).unwrap(); - let event = Event::VoteRecorded { - ts: 1, - a: "alpha".into(), - b: "beta".into(), - ratio_left: 2, - ratio_right: 1, - scope: String::new(), - }; + let event = Event::vote_recorded(1, "alpha", "beta", 2, 1, ""); projection_apply::apply_records(&store, &[record(1, event)]).unwrap(); assert_eq!(store.last_applied_event_count().unwrap(), 1); @@ -229,14 +224,7 @@ mod tests { let db = Db::open(tmp.path()).unwrap(); let store = ProjectionStore::from_db(&db).unwrap(); - let event = Event::VoteRecorded { - ts: 1, - a: "alpha".into(), - b: "beta".into(), - ratio_left: 2, - ratio_right: 1, - scope: String::new(), - }; + let event = Event::vote_recorded(1, "alpha", "beta", 2, 1, ""); projection_apply::apply_records(&store, &[record(1, event)]).unwrap(); let scoped = store.scope_tree(&ItemId::root()).unwrap(); diff --git a/server/src/ranking.rs b/server/src/ranking.rs index a83f7be120d9420db042479222ca9efc230b6da9..1fc3298d2b2864e9711cd262a034677e45c7090c 100644 --- a/server/src/ranking.rs +++ b/server/src/ranking.rs @@ -271,6 +271,7 @@ pub fn group_summary_scores(group: &GroupState) -> HashMap { #[cfg(test)] mod tests { use super::*; + use crate::identity::{DEFAULT_PSEUDONYM, TEST_ACTOR_UUID}; use crate::reducer::VoteData; fn mk_group() -> GroupState { @@ -278,18 +279,11 @@ mod tests { } fn vote(ts: i64, a: &str, b: &str, l: i32, r: i32) -> VoteData { - use crate::path_types::ItemId; - VoteData { - ts, - a: ItemId::parse(a).unwrap(), - b: ItemId::parse(b).unwrap(), - ratio_left: l, - ratio_right: r, - body: "because".to_string(), - principal: "test".to_string(), - delegate: Some("00000000-0000-0000-0000-000000000000:test:local/test".to_string()), - thread_tag: "untagged".to_string(), - } + VoteData::from_event(ts, a, b, l, r, DEFAULT_PSEUDONYM.to_string(), 1.0).unwrap() + } + + fn apply(g: &mut GroupState, v: VoteData) { + g.apply_vote(v, TEST_ACTOR_UUID); } /// Regression for issue #146: pure forward star at default `>` ratio (2:1). @@ -301,8 +295,8 @@ mod tests { #[test] fn star_topology_winner_at_top_via_subset() { let mut g = mk_group(); - g.apply_vote(vote(1, "zebra", "alpha", 2, 1)); - g.apply_vote(vote(2, "zebra", "beta", 2, 1)); + g.apply_vote(vote(1, "zebra", "alpha", 2, 1), TEST_ACTOR_UUID); + g.apply_vote(vote(2, "zebra", "beta", 2, 1), TEST_ACTOR_UUID); let mut items: Vec<(usize, String)> = g .idx_to_item @@ -329,7 +323,7 @@ mod tests { let mut g = mk_group(); // Chain a > b > c > d > e > f so ranks are well separated. for (hi, lo) in [("a", "b"), ("b", "c"), ("c", "d"), ("d", "e"), ("e", "f")] { - g.apply_vote(vote(1, hi, lo, 2, 1)); + apply(&mut g, vote(1, hi, lo, 2, 1)); } let (top, bottom) = top_bottom(&g, 2); assert_eq!(top.len(), 2); @@ -345,7 +339,7 @@ mod tests { #[test] fn top_bottom_small_group_has_empty_bottom() { let mut g = mk_group(); - g.apply_vote(vote(1, "a", "b", 2, 1)); + apply(&mut g, vote(1, "a", "b", 2, 1)); let (top, bottom) = top_bottom(&g, 5); assert_eq!(top.len(), 2); assert!(bottom.is_empty()); @@ -355,8 +349,8 @@ mod tests { fn connected_components_split_disconnected_pairs() { let mut g = mk_group(); // Two disconnected edges: (a,b) and (c,d) - g.apply_vote(vote(1, "a", "b", 3, 1)); - g.apply_vote(vote(2, "c", "d", 3, 1)); + apply(&mut g, vote(1, "a", "b", 3, 1)); + apply(&mut g, vote(2, "c", "d", 3, 1)); let n = g.idx_to_item.len(); let (mut comps, isolates) = @@ -401,13 +395,16 @@ mod tests { let i = *perm[..k].choose(&mut rng).unwrap(); let j = perm[k]; let (a, b) = (letters[i], letters[j]); - g.apply_vote(vote( - k as i64, - &a.to_string(), - &b.to_string(), - (i + 1) as i32, - (j + 1) as i32, - )); + apply( + &mut g, + vote( + k as i64, + &a.to_string(), + &b.to_string(), + (i + 1) as i32, + (j + 1) as i32, + ), + ); } let ranked = ranked_items(&g); @@ -426,8 +423,8 @@ mod tests { #[test] fn subset_ranking_ranks_within_component_only() { let mut g = mk_group(); - g.apply_vote(vote(1, "a", "b", 3, 1)); // a > b - g.apply_vote(vote(2, "c", "d", 1, 4)); // d > c + apply(&mut g, vote(1, "a", "b", 3, 1)); // a > b + apply(&mut g, vote(2, "c", "d", 1, 4)); // d > c let (comps, _) = connected_components_from_voted_pairs( g.idx_to_item.len(), diff --git a/server/src/reducer.rs b/server/src/reducer.rs index 759918b8c0eb8f8bf1ed0911d8877adaa55c8ea6..8c4c9f83635cbcbb037d680dbae2aa99cb2dc785 100644 --- a/server/src/reducer.rs +++ b/server/src/reducer.rs @@ -5,27 +5,27 @@ use serde::{Deserialize, Serialize}; use crate::path_types::ItemId; /// Parsed pairwise vote (internal representation). -#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] pub struct VoteData { pub ts: i64, pub a: ItemId, pub b: ItemId, pub ratio_left: i32, pub ratio_right: i32, - pub body: String, - pub principal: String, - pub delegate: Option, - pub thread_tag: String, + pub pseudonym: String, + pub trust_weight: f64, } impl VoteData { - /// Build a vote from persisted event fields (web UI / replay). - pub fn from_recorded( + /// Build a vote from validated event fields (replay / tests). + pub fn from_event( ts: i64, a: &str, b: &str, ratio_left: i32, ratio_right: i32, + pseudonym: String, + trust_weight: f64, ) -> Option { let a = ItemId::from_storage(a)?; let b = ItemId::from_storage(b)?; @@ -38,10 +38,8 @@ impl VoteData { b, ratio_left, ratio_right, - body: String::new(), - principal: "web".to_string(), - delegate: None, - thread_tag: "default".to_string(), + pseudonym, + trust_weight, }) } } @@ -52,6 +50,8 @@ pub struct GroupState { pub idx_to_item: Vec, pub edges: HashMap<(usize, usize), f64>, pub voted_pairs: HashSet<(usize, usize)>, + /// Latest vote per `(actor_uuid, min_idx, max_idx)` — Sybil dedup anchor. + pub uuid_votes: HashMap<(String, usize, usize), VoteData>, pub recent_votes: Vec, } @@ -62,6 +62,7 @@ impl GroupState { idx_to_item: Vec::new(), edges: HashMap::new(), voted_pairs: HashSet::new(), + uuid_votes: HashMap::new(), recent_votes: Vec::new(), } } @@ -83,36 +84,78 @@ impl GroupState { *self.edges.entry((src, dst)).or_insert(0.0) += w; } - pub fn apply_vote(&mut self, mut vote: VoteData) { - vote.a = ItemId::from_storage(vote.a.as_str()).unwrap_or(vote.a.clone()); - vote.b = ItemId::from_storage(vote.b.as_str()).unwrap_or(vote.b.clone()); - if vote.ratio_left < 0 { - vote.ratio_left = 0; - } - if vote.ratio_right < 0 { - vote.ratio_right = 0; - } - if vote.ratio_left == 0 && vote.ratio_right == 0 { + fn subtract_edge_weight(&mut self, src: usize, dst: usize, w: f64) { + if w <= 0.0 { return; } - let a_idx = self.ensure_item(&vote.a); - let b_idx = self.ensure_item(&vote.b); + if let Some(entry) = self.edges.get_mut(&(src, dst)) { + *entry -= w; + if *entry <= 0.0 { + self.edges.remove(&(src, dst)); + } + } + } + fn apply_weights(&mut self, vote: &VoteData, a_idx: usize, b_idx: usize) { + let w_a = vote.ratio_left as f64 * vote.trust_weight; + let w_b = vote.ratio_right as f64 * vote.trust_weight; let (i, j) = if a_idx < b_idx { (a_idx, b_idx) } else { (b_idx, a_idx) }; - - let w_a = vote.ratio_left as f64; - let w_b = vote.ratio_right as f64; - self.voted_pairs.insert((i, j)); self.add_edge_weight(b_idx, a_idx, w_a); self.add_edge_weight(a_idx, b_idx, w_b); + } + fn rollback_weights(&mut self, vote: &VoteData) { + let a_idx = match self.item_to_idx.get(&vote.a) { + Some(&i) => i, + None => return, + }; + let b_idx = match self.item_to_idx.get(&vote.b) { + Some(&i) => i, + None => return, + }; + let w_a = vote.ratio_left as f64 * vote.trust_weight; + let w_b = vote.ratio_right as f64 * vote.trust_weight; + self.subtract_edge_weight(b_idx, a_idx, w_a); + self.subtract_edge_weight(a_idx, b_idx, w_b); + } + + /// Apply a validated vote, deduplicating by `actor_uuid` per unordered pair. + pub fn apply_vote(&mut self, vote: VoteData, actor_uuid: &str) { + let a_idx = self.ensure_item(&vote.a); + let b_idx = self.ensure_item(&vote.b); + let (i, j) = if a_idx < b_idx { + (a_idx, b_idx) + } else { + (b_idx, a_idx) + }; + + let dedupe_key = (actor_uuid.to_string(), i, j); + if let Some(old) = self.uuid_votes.get(&dedupe_key).cloned() { + self.rollback_weights(&old); + } + + self.apply_weights(&vote, a_idx, b_idx); + self.uuid_votes.insert(dedupe_key, vote.clone()); self.recent_votes.push(vote); } + + /// Rebuild edge weights from deduped uuid votes (load path — no rollback). + pub fn ingest_uuid_vote(&mut self, vote: VoteData, actor_uuid: &str) { + let a_idx = self.ensure_item(&vote.a); + let b_idx = self.ensure_item(&vote.b); + let (i, j) = if a_idx < b_idx { + (a_idx, b_idx) + } else { + (b_idx, a_idx) + }; + self.apply_weights(&vote, a_idx, b_idx); + self.uuid_votes.insert((actor_uuid.to_string(), i, j), vote); + } } /// Structured data imported from Reddit or elsewhere. @@ -192,14 +235,14 @@ impl GlobalTree { self.nodes.get(id) } - pub fn apply_vote(&mut self, parent: &ItemId, vote: VoteData) { + pub fn apply_vote(&mut self, parent: &ItemId, vote: VoteData, actor_uuid: &str) { self.ensure_path(parent); self.ensure_path(&vote.a); self.ensure_path(&vote.b); if let Some(node) = self.nodes.get_mut(parent) { node.children.insert(vote.a.clone()); node.children.insert(vote.b.clone()); - node.local_ranking.apply_vote(vote); + node.local_ranking.apply_vote(vote, actor_uuid); } } @@ -231,36 +274,62 @@ impl GlobalTree { } #[cfg(test)] -mod from_recorded_tests { +mod tests { use super::*; + fn vote(ts: i64, a: &str, b: &str, l: i32, r: i32, pseudonym: &str) -> VoteData { + VoteData { + ts, + a: ItemId::opaque(a), + b: ItemId::opaque(b), + ratio_left: l, + ratio_right: r, + pseudonym: pseudonym.to_string(), + trust_weight: 1.0, + } + } + #[test] - fn rejects_same_item() { - assert!(VoteData::from_recorded(1, "a", "a", 2, 1).is_none()); + fn from_event_rejects_same_item() { + assert!(VoteData::from_event( + 1, + "a", + "a", + 2, + 1, + "anon".into(), + 1.0 + ) + .is_none()); } #[test] - fn rejects_empty_pair() { - assert!(VoteData::from_recorded(1, "", "b", 2, 1).is_none()); + fn from_event_rejects_empty_pair() { + assert!(VoteData::from_event(1, "", "b", 2, 1, "anon".into(), 1.0).is_none()); + } + + #[test] + fn same_uuid_replaces_prior_vote_on_pair() { + let mut g = GroupState::new(); + let uuid = "u1"; + g.apply_vote(vote(1, "a", "b", 2, 1, "alice"), uuid); + let first_total: f64 = g.edges.values().sum(); + assert_eq!(first_total, 3.0); + + g.apply_vote(vote(2, "a", "b", 0, 1, "bob"), uuid); + let second_total: f64 = g.edges.values().sum(); + assert_eq!(second_total, 1.0); + assert_eq!(g.uuid_votes.len(), 1); } #[test] - fn zero_weight_vote_does_not_mark_pair_or_edges() { + fn different_uuids_both_count() { let mut g = GroupState::new(); - g.apply_vote(VoteData { - ts: 1, - a: ItemId::opaque("a"), - b: ItemId::opaque("b"), - ratio_left: 0, - ratio_right: 0, - body: String::new(), - principal: "test".to_string(), - delegate: None, - thread_tag: "untagged".to_string(), - }); - assert!(g.voted_pairs.is_empty()); - assert!(g.edges.is_empty()); - assert!(g.recent_votes.is_empty()); + g.apply_vote(vote(1, "a", "b", 2, 1, "alice"), "u1"); + g.apply_vote(vote(2, "a", "b", 0, 1, "bob"), "u2"); + let total: f64 = g.edges.values().sum(); + assert_eq!(total, 4.0); + assert_eq!(g.uuid_votes.len(), 2); } #[test] diff --git a/server/src/state.rs b/server/src/state.rs index 247b9047a57956f76c4b6bef691662101e62a8f9..8dabc93e95c39cbe18d69596dee79683b384ef86 100644 --- a/server/src/state.rs +++ b/server/src/state.rs @@ -9,7 +9,7 @@ use crate::{ projection_apply, projection_store::ProjectionStore, reddit::{RedditApiConfig, RedditBroker, REDDIT_CONTENT_TTL}, - reducer::{GlobalTree, VoteData}, + reducer::GlobalTree, view_log::ViewLog, views::ViewStore, }; @@ -234,23 +234,31 @@ impl AppState { if a_raw.is_empty() || b_raw.is_empty() || a_raw == b_raw { return Err("invalid vote: need two distinct non-empty items".to_string()); } - if ratio_left.max(0) == 0 && ratio_right.max(0) == 0 { + let left = ratio_left.max(0); + let right = ratio_right.max(0); + if left == 0 && right == 0 { return Err( "invalid vote: need a positive preference on at least one side".to_string(), ); } - // Validate items canonicalize (or are opaque keys) before append. - let _ = VoteData::from_recorded(ts, a_raw, b_raw, ratio_left, ratio_right) - .ok_or_else(|| "invalid vote: need two distinct parseable items".to_string())?; + let a_id = ItemId::from_storage(a_raw) + .or_else(|| ItemId::parse(a_raw)) + .ok_or_else(|| "invalid vote: unparseable item a".to_string())?; + let b_id = ItemId::from_storage(b_raw) + .or_else(|| ItemId::parse(b_raw)) + .ok_or_else(|| "invalid vote: unparseable item b".to_string())?; + if a_id == b_id { + return Err("invalid vote: need two distinct items".to_string()); + } - let event = Event::VoteRecorded { + let event = Event::vote_recorded( ts, - a: a_raw.to_string(), - b: b_raw.to_string(), - ratio_left, - ratio_right, - scope: parent.as_str().to_string(), - }; + a_id.as_str(), + b_id.as_str(), + left, + right, + parent.as_str(), + ); self.journal.append(event).await } @@ -333,17 +341,7 @@ mod tests { id: "https://reddit.com/r/rust".into(), }, ), - event_record( - 2, - Event::VoteRecorded { - ts: 2, - a: "alpha".into(), - b: "beta".into(), - ratio_left: 2, - ratio_right: 1, - scope: String::new(), - }, - ), + event_record(2, Event::vote_recorded(2, "alpha", "beta", 2, 1, "")), ]) .await .unwrap(); @@ -430,14 +428,7 @@ mod tests { let log = EventLog::new(log_path.to_string_lossy().into_owned()); log.append(&event_record( 1, - Event::VoteRecorded { - ts: 1, - a: "alpha".into(), - b: "beta".into(), - ratio_left: 2, - ratio_right: 1, - scope: String::new(), - }, + Event::vote_recorded(1, "alpha", "beta", 2, 1, ""), )) .await .unwrap(); @@ -601,14 +592,7 @@ mod tests { let log = EventLog::new(format!("{data_dir}/events.jsonl")); log.append(&event_record( 1, - Event::VoteRecorded { - ts: 1, - a: "alpha".into(), - b: "beta".into(), - ratio_left: 2, - ratio_right: 1, - scope: String::new(), - }, + Event::vote_recorded(1, "alpha", "beta", 2, 1, ""), )) .await .unwrap(); diff --git a/server/src/storage_dto.rs b/server/src/storage_dto.rs index 3fd6db5cb909ac4896bd8a3ecace796de5f08781..22f5ac498ae3a5db4347f3fe94830539c7c01e3e 100644 --- a/server/src/storage_dto.rs +++ b/server/src/storage_dto.rs @@ -1,9 +1,8 @@ //! Versioned leaf value DTOs persisted in durable collections. //! -//! 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: -//! ephemeral entity views and individual votes. +//! Node structure (children, uuid votes, recent votes) lives as point-addressable +//! durable collections (see [`crate::storage_schema`]). This module defines the +//! small leaf values: ephemeral entity views and individual votes. use serde::{Deserialize, Serialize}; @@ -12,7 +11,7 @@ use crate::{ reducer::{EntityData, VoteData}, }; -pub const VOTE_RECORD_VERSION: u32 = 1; +pub const VOTE_RECORD_VERSION: u32 = 2; pub const ENTITY_DATA_VERSION: u32 = 1; #[derive(Debug, Clone, Serialize, Deserialize)] @@ -39,7 +38,7 @@ pub struct StoredEntityDataV1 { pub link_url: Option, } -/// One vote stored in a node's `recent_votes` list. +/// One vote stored in a node's `recent_votes` list or `uuid_votes` map. #[derive(Debug, Clone, Serialize, Deserialize)] pub struct StoredVoteV1 { pub version: u32, @@ -48,10 +47,8 @@ pub struct StoredVoteV1 { pub b: String, pub ratio_left: i32, pub ratio_right: i32, - pub body: String, - pub principal: String, - pub delegate: Option, - pub thread_tag: String, + pub pseudonym: String, + pub trust_weight: f64, } pub fn encode_entity_data(data: &EntityData) -> StoredEntityDataV1 { @@ -85,10 +82,8 @@ pub fn encode_vote(vote: &VoteData) -> StoredVoteV1 { b: vote.b.as_str().to_string(), ratio_left: vote.ratio_left, ratio_right: vote.ratio_right, - body: vote.body.clone(), - principal: vote.principal.clone(), - delegate: vote.delegate.clone(), - thread_tag: vote.thread_tag.clone(), + pseudonym: vote.pseudonym.clone(), + trust_weight: vote.trust_weight, } } @@ -99,10 +94,8 @@ pub fn decode_vote(vote: StoredVoteV1) -> Result { b: parse_stored_id(&vote.b)?, ratio_left: vote.ratio_left, ratio_right: vote.ratio_right, - body: vote.body, - principal: vote.principal, - delegate: vote.delegate, - thread_tag: vote.thread_tag, + pseudonym: vote.pseudonym, + trust_weight: vote.trust_weight, }) } diff --git a/server/src/storage_schema.rs b/server/src/storage_schema.rs index 5d2bb1d56927fb61c7c6d2d8602bd6882327f862..67c9f4ab2e1865f8da81b6735dd2a05b87e5e366 100644 --- a/server/src/storage_schema.rs +++ b/server/src/storage_schema.rs @@ -1,14 +1,13 @@ //! Durable schema for sorter2 — the projection laid out as point-addressable //! durable collections instead of one blob per node. //! -//! A vote updates a handful of keys: a few edge-weight merges, a voted-pair flag, -//! a recent-vote list append, and child-link set entries. The in-memory -//! [`crate::reducer::GroupState`] is reconstructed from these keys on read for -//! rank-centrality. +//! Votes are stored as deduped `uuid_votes` entries plus an append-only +//! `recent_votes` audit list. Edge weights for rank centrality are derived +//! from `uuid_votes` on read, not incrementally merged in RocksDB. -use std::collections::{BTreeSet, HashMap, HashSet}; +use std::collections::{HashSet}; -use durable::{Batch, Db, Durable, Leaf, List, Map, Sum}; +use durable::{Batch, Db, Durable, Leaf, List, Map}; use crate::{ path_types::ItemId, @@ -19,10 +18,8 @@ use crate::{ }, }; -/// Directed edge key `(from_id, to_id)`. -pub type EdgeKey = (String, String); -/// Unordered voted-pair key, stored canonically as `(min, max)` by string. -pub type PairKey = (String, String); +/// `(actor_uuid, min_item_id, max_item_id)` — one vote slot per human per pair. +pub type UuidVoteKey = (String, String, String); /// One node in the fractal tree, exploded into precisely-updatable collections. #[derive(Durable)] @@ -34,22 +31,21 @@ pub struct NodeSchema { pub data: Leaf, /// Child ids (a set; value is always `true`). pub children: Map>, - /// Directed edge weights `(from, to) -> weight`, updated by blind merges. - pub edges: Map>, - /// Voted pairs `(min, max) -> true`. - pub voted_pairs: Map>, + /// Latest vote per actor per unordered pair; edges are derived from this on read. + pub uuid_votes: Map>, /// Recent votes, append-only oldest-first (cap applied on read). pub recent_votes: List>, /// When ephemeral Reddit display content was last fetched (ms); absent after eviction. pub fetched_at: Leaf, } -/// The single database root: nodes, view counts, and per-concern metadata maps -/// (cursors and schema versions). +/// The single database root: nodes, identity maps, view counts, and metadata. #[derive(Durable)] #[allow(dead_code)] pub struct Store { pub nodes: Map, + /// Global pseudonym → actor UUID (Sybil dedup anchor). + pub pseudonyms: Map>, pub proj_meta: Map>, pub view_counts: Map>, pub view_meta: Map>, @@ -62,6 +58,21 @@ fn id_key(id: &ItemId) -> String { id.as_str().to_string() } +fn pair_keys(a: &ItemId, b: &ItemId) -> (String, String) { + let ak = id_key(a); + let bk = id_key(b); + if ak <= bk { + (ak, bk) + } else { + (bk, ak) + } +} + +pub fn uuid_vote_key(actor_uuid: &str, a: &ItemId, b: &ItemId) -> UuidVoteKey { + let (lo, hi) = pair_keys(a, b); + (actor_uuid.to_string(), lo, hi) +} + /// Path to a node by id. pub fn node(id: &ItemId) -> durable::Path { Store::root().nodes().key(&id_key(id)) @@ -77,16 +88,10 @@ pub fn load_node_state(db: &Db, id: &ItemId) -> durable::Result durable::Result durable::Result, - voted: Vec, - edges_raw: Vec<(EdgeKey, f64)>, -) -> durable::Result { - // Item universe = every endpoint that appears in a voted pair or an edge. - let mut item_strs: BTreeSet = BTreeSet::new(); - for (a, b) in &voted { - item_strs.insert(a.clone()); - item_strs.insert(b.clone()); - } - for ((a, b), _) in &edges_raw { - item_strs.insert(a.clone()); - item_strs.insert(b.clone()); - } - - let mut idx_to_item: Vec = Vec::with_capacity(item_strs.len()); - let mut item_to_idx: HashMap = HashMap::with_capacity(item_strs.len()); - let mut str_to_idx: HashMap = HashMap::with_capacity(item_strs.len()); - for s in item_strs { - let id = parse_storage_id(&s)?; - let idx = idx_to_item.len(); - str_to_idx.insert(s, idx); - item_to_idx.insert(id.clone(), idx); - idx_to_item.push(id); - } +fn build_group_state(db: &Db, np: &durable::Path) -> durable::Result { + let mut group = GroupState::new(); - let mut edges: HashMap<(usize, usize), f64> = HashMap::with_capacity(edges_raw.len()); - for ((a, b), w) in edges_raw { - if let (Some(&ai), Some(&bi)) = (str_to_idx.get(&a), str_to_idx.get(&b)) { - edges.insert((ai, bi), w); - } - } - - let mut voted_pairs: HashSet<(usize, usize)> = HashSet::with_capacity(voted.len()); - for (a, b) in voted { - if let (Some(&ai), Some(&bi)) = (str_to_idx.get(&a), str_to_idx.get(&b)) { - let (i, j) = if ai < bi { (ai, bi) } else { (bi, ai) }; - voted_pairs.insert((i, j)); - } + for (key, stored) in np.uuid_votes().iter(db)? { + let (actor_uuid, _lo, _hi) = key; + let vote = decode_vote(stored).map_err(durable::Error::Deserialize)?; + group.ingest_uuid_vote(vote, &actor_uuid); } - // List is index order (oldest first); keep the newest RECENT_VOTES_CAP entries. let stored = np.recent_votes().iter(db)?; let cap = RECENT_VOTES_CAP as usize; let start = stored.len().saturating_sub(cap); - let recent_votes = stored[start..] + group.recent_votes = stored[start..] .iter() .map(|s| decode_vote(s.clone()).map_err(durable::Error::Deserialize)) .collect::, _>>()?; - Ok(GroupState { - item_to_idx, - idx_to_item, - edges, - voted_pairs, - recent_votes, - }) + Ok(group) } fn parse_storage_id(s: &str) -> durable::Result { @@ -197,73 +161,24 @@ pub fn ensure_path_writes(batch: &mut Batch, id: &ItemId) { } } -/// Reified writes for a recorded vote under `parent`. Mirrors -/// [`crate::reducer::GroupState::apply_vote`] as point updates. +/// Reified writes for a validated vote under `parent`. pub fn vote_writes( batch: &mut Batch, parent: &ItemId, - a: &str, - b: &str, - ratio_left: i32, - ratio_right: i32, - ts: i64, + vote: &VoteData, + actor_uuid: &str, ) -> durable::Result<()> { - let left = ratio_left.max(0); - let right = ratio_right.max(0); - if left == 0 && right == 0 { - return Ok(()); - } - - let a_id = ItemId::from_storage(a).unwrap_or_else(|| ItemId::opaque(a)); - let b_id = ItemId::from_storage(b).unwrap_or_else(|| ItemId::opaque(b)); - ensure_path_writes(batch, parent); - ensure_path_writes(batch, &a_id); - ensure_path_writes(batch, &b_id); + ensure_path_writes(batch, &vote.a); + ensure_path_writes(batch, &vote.b); let pnode = node(parent); - batch.write(pnode.children().key(&id_key(&a_id)).set(&true)); - batch.write(pnode.children().key(&id_key(&b_id)).set(&true)); - - // Edge weights: edge (b,a) += left, edge (a,b) += right (positive only). - if left > 0 { - batch.write( - pnode - .edges() - .key(&(id_key(&b_id), id_key(&a_id))) - .add(left as f64), - ); - } - if right > 0 { - batch.write( - pnode - .edges() - .key(&(id_key(&a_id), id_key(&b_id))) - .add(right as f64), - ); - } + batch.write(pnode.children().key(&id_key(&vote.a)).set(&true)); + batch.write(pnode.children().key(&id_key(&vote.b)).set(&true)); - // Voted pair, canonicalized. - let (lo, hi) = if id_key(&a_id) <= id_key(&b_id) { - (id_key(&a_id), id_key(&b_id)) - } else { - (id_key(&b_id), id_key(&a_id)) - }; - batch.write(pnode.voted_pairs().key(&(lo, hi)).set(&true)); - - // Recent votes (append-only; cap on read). - let stored = encode_vote(&VoteData { - ts, - a: a_id, - b: b_id, - ratio_left: left, - ratio_right: right, - body: String::new(), - principal: "web".to_string(), - delegate: None, - thread_tag: "default".to_string(), - }); - batch.push(&pnode.recent_votes(), &stored)?; + let key = uuid_vote_key(actor_uuid, &vote.a, &vote.b); + batch.write(pnode.uuid_votes().key(&key).set(&encode_vote(vote))); + batch.push(&pnode.recent_votes(), &encode_vote(vote))?; Ok(()) } @@ -283,15 +198,30 @@ pub fn entity_content_clear_writes(batch: &mut Batch, id: &ItemId) { #[cfg(test)] mod tests { use super::*; + use crate::identity::{seed_default_pseudonym, DEFAULT_ACTOR_UUID, DEFAULT_PSEUDONYM}; + + fn sample_vote(ts: i64, a: &str, b: &str, l: i32, r: i32) -> VoteData { + VoteData { + ts, + a: ItemId::opaque(a), + b: ItemId::opaque(b), + ratio_left: l, + ratio_right: r, + pseudonym: DEFAULT_PSEUDONYM.to_string(), + trust_weight: 1.0, + } + } #[test] fn vote_roundtrip_reconstructs_group_state() { let dir = tempfile::tempdir().unwrap(); let db = Db::open(dir.path()).unwrap(); + seed_default_pseudonym(&db).unwrap(); let parent = ItemId::root(); + let vote = sample_vote(1, "alpha", "beta", 2, 1); let mut batch = db.batch(); - vote_writes(&mut batch, &parent, "alpha", "beta", 2, 1, 1).unwrap(); + vote_writes(&mut batch, &parent, &vote, DEFAULT_ACTOR_UUID).unwrap(); batch.commit().unwrap(); let node_state = load_node_state(&db, &parent).unwrap().unwrap(); @@ -305,16 +235,38 @@ mod tests { } #[test] - fn zero_weight_vote_writes_nothing() { + fn uuid_vote_replace_updates_edges() { let dir = tempfile::tempdir().unwrap(); let db = Db::open(dir.path()).unwrap(); let parent = ItemId::root(); + let uuid = "u1"; let mut batch = db.batch(); - vote_writes(&mut batch, &parent, "alpha", "beta", 0, 0, 1).unwrap(); + vote_writes( + &mut batch, + &parent, + &sample_vote(1, "alpha", "beta", 2, 1), + uuid, + ) + .unwrap(); + vote_writes( + &mut batch, + &parent, + &VoteData { + pseudonym: "alias2".into(), + ratio_left: 0, + ratio_right: 1, + ..sample_vote(2, "alpha", "beta", 0, 1) + }, + uuid, + ) + .unwrap(); batch.commit().unwrap(); - assert!(load_node_state(&db, &parent).unwrap().is_none()); + let g = &load_node_state(&db, &parent).unwrap().unwrap().local_ranking; + let edge_total: f64 = g.edges.values().sum(); + assert_eq!(edge_total, 1.0); + assert_eq!(g.uuid_votes.len(), 1); } #[test] @@ -325,7 +277,13 @@ mod tests { let mut batch = db.batch(); for i in 0..RECENT_VOTES_CAP + 10 { - vote_writes(&mut batch, &parent, "alpha", "beta", 1, 0, i as i64).unwrap(); + vote_writes( + &mut batch, + &parent, + &sample_vote(i as i64, "alpha", "beta", 1, 0), + "u1", + ) + .unwrap(); } batch.commit().unwrap(); @@ -337,13 +295,13 @@ mod tests { let node_state = load_node_state(&db, &parent).unwrap().unwrap(); assert_eq!(node_state.local_ranking.recent_votes.len(), RECENT_VOTES_CAP as usize); assert_eq!( - node_state.local_ranking.recent_votes.first().map(|v| v.ts), + node_state + .local_ranking + .recent_votes + .first() + .map(|v| v.ts), Some(10) ); - assert_eq!( - node_state.local_ranking.recent_votes.last().map(|v| v.ts), - Some(RECENT_VOTES_CAP as i64 + 9) - ); } #[test]