very fast at protocol indexer with flexible filtering, xrpc queries, cursor-backed event stream, and more, built on fjall
rust
fjall
at-protocol
atproto
indexer
[db] some more expect point read hits for certain keyspaces
+26
-29
+25
-28
src/db/mod.rs
+25
-28
src/db/mod.rs
···
3
3
use futures::FutureExt;
4
4
use jacquard::IntoStatic;
5
5
use jacquard_common::types::string::Did;
6
-
use miette::{IntoDiagnostic, Result};
6
+
use miette::{Context, IntoDiagnostic, Result};
7
7
use std::future::Future;
8
8
use std::path::Path;
9
9
use std::sync::Arc;
···
47
47
.into_diagnostic()?;
48
48
let db = Arc::new(db);
49
49
50
-
let opts = || KeyspaceCreateOptions::default();
50
+
let opts = KeyspaceCreateOptions::default;
51
+
let open_ks = |name: &str, opts: KeyspaceCreateOptions| {
52
+
db.keyspace(name, move || opts).into_diagnostic()
53
+
};
51
54
52
-
let repos = db.keyspace("repos", opts).into_diagnostic()?;
53
-
let records = db
54
-
.keyspace("records", || opts().max_memtable_size(32 * 2_u64.pow(20)))
55
-
.into_diagnostic()?;
56
-
57
-
let block_opts = || {
55
+
let repos = open_ks("repos", opts().expect_point_read_hits(true))?;
56
+
let records = open_ks("records", opts().max_memtable_size(32 * 2_u64.pow(20)))?;
57
+
let blocks = open_ks(
58
+
"blocks",
58
59
opts()
59
60
// point reads are used a lot by stream
60
61
.expect_point_read_hits(true)
61
-
.max_memtable_size(32 * 2_u64.pow(20))
62
-
};
63
-
let blocks = db.keyspace("blocks", block_opts).into_diagnostic()?;
64
-
let cursors = db.keyspace("cursors", opts).into_diagnostic()?;
65
-
let buffer = db.keyspace("buffer", opts).into_diagnostic()?;
66
-
let pending = db.keyspace("pending", opts).into_diagnostic()?;
67
-
let resync = db.keyspace("resync", opts).into_diagnostic()?;
68
-
let events = db.keyspace("events", opts).into_diagnostic()?;
69
-
let counts = db.keyspace("counts", opts).into_diagnostic()?;
62
+
.max_memtable_size(32 * 2_u64.pow(20)),
63
+
)?;
64
+
let cursors = open_ks("cursors", opts().expect_point_read_hits(true))?;
65
+
let buffer = open_ks("buffer", opts())?;
66
+
let pending = open_ks("pending", opts())?;
67
+
let resync = open_ks("resync", opts())?;
68
+
let events = open_ks("events", opts())?;
69
+
let counts = open_ks("counts", opts().expect_point_read_hits(true))?;
70
70
71
71
let mut last_id = 0;
72
-
let mut iter = events.iter();
73
-
if let Some(guard) = iter.next_back() {
74
-
let res: Result<fjall::KvPair, _> = guard.into_inner().map_err(|e| miette::miette!(e));
75
-
if let Ok(kv) = res {
76
-
let k = &kv.0;
77
-
let mut buf = [0u8; 8];
78
-
if k.len() == 8 {
79
-
buf.copy_from_slice(k);
80
-
last_id = u64::from_be_bytes(buf);
81
-
}
82
-
}
72
+
if let Some(guard) = events.iter().next_back() {
73
+
let k = guard.key().into_diagnostic()?;
74
+
last_id = u64::from_be_bytes(
75
+
k.as_ref()
76
+
.try_into()
77
+
.into_diagnostic()
78
+
.wrap_err("expected to be id (8 bytes)")?,
79
+
);
83
80
}
84
81
85
82
let (event_tx, _) = broadcast::channel(10000);
+1
-1
src/ops.rs
+1
-1
src/ops.rs
···
93
93
status: crate::types::RepoStatus,
94
94
) -> Result<()> {
95
95
debug!("updating repo status for {did} to {status:?}");
96
-
let (updated, mut batch) =
96
+
let (updated, batch) =
97
97
Db::update_repo_state(db.inner.batch(), &db.repos, did, |state, _val| {
98
98
state.status = status.clone();
99
99
state.last_updated_at = chrono::Utc::now().timestamp();