bobbin-edge-index: list_multi #16

+1

Cargo.lock

Reviewed

··· 373 373 "bobbin-runtime", 374 374 "bobbin-types", 375 375 "either", 376 376 + "itertools 0.14.0", 376 377 "jacquard-common", 377 378 "lasso", 378 379 "scc",

+1

Cargo.toml

Reviewed

··· 57 57 tokio-tungstenite = { version = "0.29", features = ["rustls-tls-webpki-roots"] } 58 58 futures = "0.3" 59 59 either = "1" 60 60 + itertools = "0.14" 60 61 61 62 serde = { version = "1", features = ["derive"] } 62 63 serde_json = { version = "1", features = ["raw_value"] }

+1

bobbin/crates/edge-index/Cargo.toml

Reviewed

··· 9 9 bobbin-runtime = { workspace = true } 10 10 bobbin-types = { workspace = true } 11 11 either = { workspace = true } 12 12 + itertools = { workspace = true } 12 13 jacquard-common = { workspace = true } 13 14 lasso = { workspace = true } 14 15 scc = { workspace = true }

+114

bobbin/crates/edge-index/src/lib.rs

Reviewed

··· 28 28 use bobbin_types::edges::Edge; 29 29 use bobbin_types::ids::EdgeKey; 30 30 use either::Either; 31 31 + use itertools::Itertools; 31 32 use jacquard_common::DefaultStr; 32 33 use jacquard_common::types::string::AtUri; 33 34 use lasso::{Key, Spur, ThreadedRodeo}; ··· 733 734 }) 734 735 } 735 736 737 737 + /// Merge a page across many buckets in one global time order. 738 738 + pub fn list_multi( 739 739 + &self, 740 740 + keys: &[EdgeKey], 741 741 + cursor: PageCursor, 742 742 + limit: PageLimit, 743 743 + dir: SortDir, 744 744 + ) -> EdgePage { 745 745 + let n = limit.get() as usize + 1; 746 746 + let is_first = |x: &BucketKey, y: &BucketKey| match dir { 747 747 + SortDir::Asc => x <= y, 748 748 + SortDir::Desc => x >= y, 749 749 + }; 750 750 + let mut top: Vec<BucketKey> = Vec::with_capacity(n); 751 751 + for key in keys { 752 752 + let Some(id) = self.lookup_key(key) else { 753 753 + continue; 754 754 + }; 755 755 + let bucket: Vec<BucketKey> = self 756 756 + .forward 757 757 + .read_sync(&id, |_, sources| { 758 758 + sources.directed(cursor, dir).take(n).collect() 759 759 + }) 760 760 + .unwrap_or_default(); 761 761 + top = top.into_iter().merge_by(bucket, &is_first).take(n).collect(); 762 762 + } 763 763 + top.dedup_by_key(|k| k.source); 764 764 + 765 765 + let has_more = top.len() > limit.get() as usize; 766 766 + let page = &top[..top.len().min(limit.get() as usize)]; 767 767 + let items = page 768 768 + .iter() 769 769 + .filter_map(|&k| { 770 770 + let spur = k.source.to_spur()?; 771 771 + let stored = self.source_interner.try_resolve(&spur)?; 772 772 + let uri = AtUri::new_owned(self.decode_source(stored)?).ok()?; 773 773 + Some(EdgeItem { 774 774 + uri, 775 775 + sort_micros: k.micros.0, 776 776 + }) 777 777 + }) 778 778 + .collect(); 779 779 + let next = has_more 780 780 + .then(|| page.last().copied()) 781 781 + .flatten() 782 782 + .map(BucketKey::token); 783 783 + EdgePage { items, next } 784 784 + } 785 785 + 736 786 pub fn list_filtered<F>( 737 787 &self, 738 788 key: &EdgeKey, ··· 1063 1113 }); 1064 1114 } 1065 1115 1116 1116 + // Build three subject buckets with interleaved micros so the global time 1117 1117 + // order crosses buckets, and return the three keys. 1118 1118 + fn fill_three_subjects(store: &EdgeStore) -> Vec<EdgeKey> { 1119 1119 + let rows = [ 1120 1120 + ("alpha", "r0", 10u64), 1121 1121 + ("bravo", "r1", 50), 1122 1122 + ("charlie", "r2", 20), 1123 1123 + ("alpha", "r3", 40), 1124 1124 + ("bravo", "r4", 30), 1125 1125 + ("charlie", "r5", 60), 1126 1126 + ]; 1127 1127 + for (subj, rkey, micros) in rows { 1128 1128 + let source = at(&format!("at://did:plc:x/sh.tangled.feed.star/{rkey}")); 1129 1129 + store.add(star_edge_at(source, did(&format!("did:plc:{subj}")), micros)); 1130 1130 + } 1131 1131 + ["alpha", "bravo", "charlie"] 1132 1132 + .iter() 1133 1133 + .map(|s| { 1134 1134 + EdgeKey::new(nsid("sh.tangled.feed.star"), did_subj(&format!("did:plc:{s}"))) 1135 1135 + }) 1136 1136 + .collect() 1137 1137 + } 1138 1138 + 1139 1139 + #[test] 1140 1140 + fn list_multi_merges_subjects_newest_first() { 1141 1141 + let store = store(); 1142 1142 + let keys = fill_three_subjects(&store); 1143 1143 + let page = store.list_multi(&keys, PageCursor::Start, limit(10), SortDir::Desc); 1144 1144 + let micros: Vec<u64> = page.items.iter().map(|it| it.sort_micros).collect(); 1145 1145 + assert_eq!(micros, vec![60, 50, 40, 30, 20, 10]); 1146 1146 + assert!(page.next.is_none()); 1147 1147 + } 1148 1148 + 1149 1149 + #[test] 1150 1150 + fn list_multi_paginates_disjoint_across_buckets() { 1151 1151 + let store = store(); 1152 1152 + let keys = fill_three_subjects(&store); 1153 1153 + // limit=4 with 6 items across 3 buckets exercises the bounded top-N merge. 1154 1154 + let p1 = store.list_multi(&keys, PageCursor::Start, limit(4), SortDir::Desc); 1155 1155 + let m1: Vec<u64> = p1.items.iter().map(|it| it.sort_micros).collect(); 1156 1156 + assert_eq!(m1, vec![60, 50, 40, 30]); 1157 1157 + let tok = p1.next.expect("first page has more"); 1158 1158 + let p2 = store.list_multi(&keys, PageCursor::After(tok), limit(4), SortDir::Desc); 1159 1159 + let m2: Vec<u64> = p2.items.iter().map(|it| it.sort_micros).collect(); 1160 1160 + assert_eq!(m2, vec![20, 10]); 1161 1161 + assert!(p2.next.is_none()); 1162 1162 + } 1163 1163 + 1164 1164 + #[test] 1165 1165 + fn list_multi_dedups_source_under_two_keys() { 1166 1166 + let store = store(); 1167 1167 + // Same record (source) indexed under two subject keys, same createdAt. 1168 1168 + let shared = at("at://did:plc:x/sh.tangled.feed.star/shared"); 1169 1169 + store.add(star_edge_at(shared.clone(), did("did:plc:alpha"), 100)); 1170 1170 + store.add(star_edge_at(shared, did("did:plc:bravo"), 100)); 1171 1171 + let keys = vec![ 1172 1172 + EdgeKey::new(nsid("sh.tangled.feed.star"), did_subj("did:plc:alpha")), 1173 1173 + EdgeKey::new(nsid("sh.tangled.feed.star"), did_subj("did:plc:bravo")), 1174 1174 + ]; 1175 1175 + let page = store.list_multi(&keys, PageCursor::Start, limit(10), SortDir::Desc); 1176 1176 + assert_eq!(page.items.len(), 1); 1177 1177 + assert!(page.next.is_none()); 1178 1178 + } 1179 1179 + 1066 1180 #[test] 1067 1181 fn remove_from_large_bucket_keeps_pagination_exact() { 1068 1182 let store = store();