forked from
microcosm.blue/microcosm-rs
Constellation, Spacedust, Slingshot, UFOs: atproto crates and services for microcosm
link reader db stats
+110
-24
+1
link_aggregator/readme.md
+1
link_aggregator/readme.md
···
116
116
- [ ] persist the jetstream server url, error if started with a different one (maybe with --switch-streams or something)
117
117
- [ ] put delete-account tasks into a separate (persisted?) task queue for the writer so it can work on them incrementally.
118
118
- [ ] jetstream: connect retry: only reset counter after some *time* has passed.
119
+
- [x] either count or estimate the total number of links added (distinct from link targets)
119
120
120
121
data fixes
121
122
- [x] add rkey to linkers 🤦♀️
+17
-1
link_aggregator/src/storage/mem_store.rs
+17
-1
link_aggregator/src/storage/mem_store.rs
···
1
-
use super::{LinkReader, LinkStorage, PagedAppendingCollection};
1
+
use super::{LinkReader, LinkStorage, PagedAppendingCollection, StorageStats};
2
2
use anyhow::Result;
3
3
use link_aggregator::{ActionableEvent, Did, RecordId};
4
4
use links::CollectedLink;
···
202
202
}
203
203
}
204
204
Ok(out)
205
+
}
206
+
207
+
fn get_stats(&self) -> Result<StorageStats> {
208
+
let data = self.0.lock().unwrap();
209
+
let dids = data.dids.len() as u64;
210
+
let targetables = data
211
+
.targets
212
+
.values()
213
+
.map(|sources| sources.len())
214
+
.sum::<usize>() as u64;
215
+
let linking_records = data.links.values().map(|recs| recs.len()).sum::<usize>() as u64;
216
+
Ok(StorageStats {
217
+
dids,
218
+
targetables,
219
+
linking_records,
220
+
})
205
221
}
206
222
207
223
fn summarize(&self, qsize: u32) {
+64
-16
link_aggregator/src/storage/mod.rs
+64
-16
link_aggregator/src/storage/mod.rs
···
17
17
pub next: Option<u64>,
18
18
}
19
19
20
+
#[derive(Debug, PartialEq)]
21
+
pub struct StorageStats {
22
+
/// estimate of how many accounts we've seen create links. the _subjects_ of any links are not represented here.
23
+
/// for example: new user A follows users B and C. this count will only increment by one, for A.
24
+
dids: u64,
25
+
26
+
/// estimate targets * distinct (collection, path)s to reference them.
27
+
/// distinct targets alone are currently challenging to estimate.
28
+
targetables: u64,
29
+
30
+
/// estimate of the count of atproto records seen that contain links.
31
+
/// records with multiple links are single-counted.
32
+
/// for LSM stores, deleted links don't decrement this, and updated records with any links will likely increment it.
33
+
linking_records: u64,
34
+
}
35
+
20
36
pub trait LinkStorage: Send + Sync {
21
37
/// jetstream cursor from last saved actions, if available
22
38
fn get_cursor(&mut self) -> Result<Option<u64>> {
···
43
59
44
60
fn get_all_counts(&self, _target: &str) -> Result<HashMap<String, HashMap<String, u64>>>;
45
61
62
+
/// assume all stats are estimates, since exact counts are very challenging for LSMs
63
+
fn get_stats(&self) -> Result<StorageStats>;
64
+
46
65
// todo: remove it
47
66
fn summarize(&self, qsize: u32) {
48
67
println!("queue: {qsize}");
···
53
72
mod tests {
54
73
use super::*;
55
74
use links::{CollectedLink, Link};
75
+
use std::ops::RangeBounds;
56
76
57
77
macro_rules! test_each_storage {
58
78
($test_name:ident, |$storage_label:ident| $test_code:block) => {
···
79
99
};
80
100
}
81
101
102
+
fn assert_stats(
103
+
stats: StorageStats,
104
+
dids: impl RangeBounds<u64>,
105
+
targetables: impl RangeBounds<u64>,
106
+
linking_records: impl RangeBounds<u64>,
107
+
) {
108
+
fn check(name: &str, stat: u64, rb: impl RangeBounds<u64>) {
109
+
assert!(
110
+
rb.contains(&stat),
111
+
"{name:?}: {stat:?} not in range {:?}–{:?}",
112
+
rb.start_bound(),
113
+
rb.end_bound()
114
+
);
115
+
}
116
+
check("dids", stats.dids, dids);
117
+
check("targetables", stats.targetables, targetables);
118
+
check("linking_records", stats.linking_records, linking_records);
119
+
}
120
+
82
121
test_each_storage!(test_empty, |storage| {
83
122
assert_eq!(storage.get_count("", "", "")?, 0);
84
123
assert_eq!(storage.get_count("a", "b", "c")?, 0);
···
90
129
)?,
91
130
0
92
131
);
132
+
assert_eq!(
133
+
storage.get_links("a.com", "app.t.c", ".abc.uri", 100, None)?,
134
+
PagedAppendingCollection {
135
+
version: (0, 0),
136
+
items: vec![],
137
+
next: None,
138
+
}
139
+
);
140
+
assert_eq!(storage.get_all_counts("bad-example.com")?, HashMap::new());
141
+
assert_stats(storage.get_stats()?, 0..=0, 0..=0, 0..=0);
93
142
});
94
143
95
144
test_each_storage!(test_add_link, |storage| {
···
110
159
assert_eq!(storage.get_count("e.com", "app.t.c", ".abc.uri")?, 1);
111
160
assert_eq!(storage.get_count("bad.com", "app.t.c", ".abc.uri")?, 0);
112
161
assert_eq!(storage.get_count("e.com", "app.t.c", ".bad.uri")?, 0);
162
+
assert_stats(storage.get_stats()?, 1..=1, 1..=1, 1..=1);
113
163
});
114
164
115
165
test_each_storage!(test_links, |storage| {
···
222
272
0,
223
273
)?;
224
274
assert_eq!(storage.get_count("e.com", "app.t.c", ".abc.uri")?, 2);
275
+
assert_stats(storage.get_stats()?, 2..=2, 1..=1, 2..=2);
225
276
});
226
277
227
278
test_each_storage!(test_two_user_links_delete_one, |storage| {
···
268
319
}),
269
320
0,
270
321
)?;
271
-
272
322
assert_eq!(storage.get_count("e.com", "app.t.c", ".abc.uri")?, 1);
323
+
assert_stats(storage.get_stats()?, 1..=1, 1..=1, 1..=1);
273
324
});
274
325
275
326
test_each_storage!(test_accounts, |storage| {
···
326
377
storage.push(&ActionableEvent::DeleteAccount("did:plc:asdf".into()), 0)?;
327
378
assert_eq!(storage.get_count("a.com", "app.t.c", ".abc.uri")?, 1);
328
379
assert_eq!(storage.get_count("b.com", "app.t.c", ".abc.uri")?, 0);
380
+
assert_stats(storage.get_stats()?, 1..=2, 2..=2, 1..=1);
329
381
});
330
382
331
383
test_each_storage!(multi_link, |storage| {
···
368
420
assert_eq!(storage.get_count("e.com", "app.t.c", ".abc.uri")?, 0);
369
421
assert_eq!(storage.get_count("f.com", "app.t.c", ".xyz[].uri")?, 0);
370
422
assert_eq!(storage.get_count("g.com", "app.t.c", ".xyz[].uri")?, 0);
423
+
assert_stats(storage.get_stats()?, 1..=1, 3..=3, 0..=0);
371
424
});
372
425
373
426
test_each_storage!(update_link, |storage| {
···
430
483
assert_eq!(storage.get_count("f.com", "app.t.c", ".xyz[].uri")?, 1);
431
484
assert_eq!(storage.get_count("g.com", "app.t.c", ".xyz[].uri")?, 0);
432
485
assert_eq!(storage.get_count("i.com", "app.t.c", ".xyz[].uri")?, 1);
486
+
assert_stats(storage.get_stats()?, 1..=1, 5..=5, 1..=1);
433
487
});
434
488
435
489
test_each_storage!(update_no_links_to_links, |storage| {
···
449
503
0,
450
504
)?;
451
505
assert_eq!(storage.get_count("a.com", "app.t.c", ".abc.uri")?, 1);
506
+
assert_stats(storage.get_stats()?, 1..=1, 1..=1, 1..=1);
452
507
});
453
508
454
509
test_each_storage!(delete_multi_link_same_target, |storage| {
···
485
540
)?;
486
541
assert_eq!(storage.get_count("a.com", "app.t.c", ".abc.uri")?, 0);
487
542
assert_eq!(storage.get_count("a.com", "app.t.c", ".def.uri")?, 0);
488
-
});
489
-
490
-
test_each_storage!(get_links_zero, |storage| {
491
-
assert_eq!(
492
-
storage.get_links("a.com", "app.t.c", ".abc.uri", 100, None)?,
493
-
PagedAppendingCollection {
494
-
version: (0, 0),
495
-
items: vec![],
496
-
next: None,
497
-
}
498
-
);
543
+
assert_stats(storage.get_stats()?, 1..=1, 2..=2, 0..=0);
499
544
});
500
545
501
546
test_each_storage!(get_links_basic, |storage| {
···
525
570
next: None,
526
571
}
527
572
);
573
+
assert_stats(storage.get_stats()?, 1..=1, 1..=1, 1..=1);
528
574
});
529
575
530
576
test_each_storage!(get_links_paged, |storage| {
···
597
643
next: None,
598
644
}
599
645
);
646
+
assert_stats(storage.get_stats()?, 5..=5, 1..=1, 5..=5);
600
647
});
601
648
602
649
test_each_storage!(get_links_exact_multiple, |storage| {
···
656
703
next: None,
657
704
}
658
705
);
706
+
assert_stats(storage.get_stats()?, 4..=4, 1..=1, 4..=4);
659
707
});
660
708
661
709
test_each_storage!(page_links_while_new_links_arrive, |storage| {
···
729
777
next: None,
730
778
}
731
779
);
780
+
assert_stats(storage.get_stats()?, 5..=5, 1..=1, 5..=5);
732
781
});
733
782
734
783
test_each_storage!(page_links_while_some_are_deleted, |storage| {
···
789
838
next: None,
790
839
}
791
840
);
841
+
assert_stats(storage.get_stats()?, 4..=4, 1..=1, 3..=3);
792
842
});
793
843
794
844
test_each_storage!(page_links_accounts_inactive, |storage| {
···
845
895
next: None,
846
896
}
847
897
);
898
+
assert_stats(storage.get_stats()?, 4..=4, 1..=1, 4..=4);
848
899
});
849
900
850
901
test_each_storage!(get_all_counts, |storage| {
···
876
927
counts.insert("app.t.c".into(), t_c_counts);
877
928
counts
878
929
});
879
-
});
880
-
881
-
test_each_storage!(get_all_counts_no_links, |storage| {
882
-
assert_eq!(storage.get_all_counts("bad-example.com")?, HashMap::new());
930
+
assert_stats(storage.get_stats()?, 1..=1, 2..=2, 1..=1);
883
931
});
884
932
}
+28
-7
link_aggregator/src/storage/rocks_store.rs
+28
-7
link_aggregator/src/storage/rocks_store.rs
···
1
-
use super::{ActionableEvent, LinkReader, LinkStorage, PagedAppendingCollection};
1
+
use super::{ActionableEvent, LinkReader, LinkStorage, PagedAppendingCollection, StorageStats};
2
2
use anyhow::{bail, Result};
3
3
use bincode::Options as BincodeOptions;
4
4
use link_aggregator::{Did, RecordId};
···
179
179
batch.put_cf(cf, _rk(orig), _rv(&id_value));
180
180
id_value
181
181
}))
182
+
}
183
+
fn estimate_count(&self) -> u64 {
184
+
self.base.id_seq.load(Ordering::SeqCst) - 1 // -1 because seq zero is reserved
182
185
}
183
186
}
184
187
impl<Orig: Clone, IdVal: IdTableValue> IdTable<Orig, IdVal, true>
···
698
701
}
699
702
700
703
impl LinkReader for RocksStorage {
701
-
fn summarize(&self, qsize: u32) {
702
-
let did_seq = self.did_id_table.base.id_seq.load(Ordering::SeqCst);
703
-
let target_seq = self.target_id_table.base.id_seq.load(Ordering::SeqCst);
704
-
println!("queue: {qsize}. did seq: {did_seq}, target seq: {target_seq}.");
705
-
}
706
-
707
704
fn get_count(&self, target: &str, collection: &str, path: &str) -> Result<u64> {
708
705
let target_key = TargetKey(
709
706
Target(target.to_string()),
···
792
789
.insert(path.clone(), count);
793
790
}
794
791
Ok(out)
792
+
}
793
+
794
+
fn get_stats(&self) -> Result<StorageStats> {
795
+
let dids = self.did_id_table.estimate_count();
796
+
let targetables = self.target_id_table.estimate_count();
797
+
let lr_cf = self.db.cf_handle(LINK_TARGETS_CF).unwrap();
798
+
let linking_records = self
799
+
.db
800
+
.property_value_cf(&lr_cf, rocksdb::properties::ESTIMATE_NUM_KEYS)?
801
+
.map(|s| s.parse::<u64>())
802
+
.transpose()?
803
+
.unwrap_or(0);
804
+
eprintln!("lr {linking_records:?}");
805
+
Ok(StorageStats {
806
+
dids,
807
+
targetables,
808
+
linking_records,
809
+
})
810
+
}
811
+
812
+
fn summarize(&self, qsize: u32) {
813
+
let did_seq = self.did_id_table.base.id_seq.load(Ordering::SeqCst);
814
+
let target_seq = self.target_id_table.base.id_seq.load(Ordering::SeqCst);
815
+
println!("queue: {qsize}. did seq: {did_seq}, target seq: {target_seq}.");
795
816
}
796
817
}
797
818