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] tune restart intervals
+38
-18
+38
-18
src/db/mod.rs
+38
-18
src/db/mod.rs
···
2
2
use crate::db::compaction::DropPrefixFilterFactory;
3
3
use crate::types::{BroadcastEvent, RepoState};
4
4
5
-
use fjall::config::{BlockSizePolicy, CompressionPolicy};
5
+
use fjall::config::{BlockSizePolicy, CompressionPolicy, RestartIntervalPolicy};
6
6
use fjall::{
7
7
CompressionType, Database, Keyspace, KeyspaceCreateOptions, OwnedWriteBatch, PersistMode, Slice,
8
8
};
···
152
152
.expect_point_read_hits(false)
153
153
.max_memtable_size(mb(cfg.db_repos_memtable_size_mb))
154
154
// did -> repo state, not gonna be compressable well because dids are random
155
-
// and repo state doesnt have repeats really
155
+
// and repo state doesnt have repeats really..
156
+
// these block sizes work fine since we insert into repos constantly anyway
157
+
// whenever we update anything related to a repo, so the repos that arent
158
+
// being updated will be compacted away!
156
159
.data_block_size_policy(BlockSizePolicy::new([kb(4), kb(4), kb(8), kb(64)]))
157
160
.data_block_compression_policy(CompressionPolicy::new([
158
161
CompressionType::None,
159
162
CompressionType::None,
160
163
compression,
161
-
])),
164
+
]))
165
+
// did plc are random so the interval wont rlly matter
166
+
.data_block_restart_interval_policy(RestartIntervalPolicy::new([2, 4])),
162
167
)?;
163
168
let pending = open_ks(
164
169
"pending",
···
167
172
.expect_point_read_hits(true)
168
173
.max_memtable_size(mb(8))
169
174
// its just index of id (int) -> did, and dids arent compressable (especially with the ids being random)
170
-
.data_block_size_policy(BlockSizePolicy::all(kb(1)))
175
+
.data_block_size_policy(BlockSizePolicy::all(kb(8)))
171
176
// and we'll transition from pending to synced anyway, no point trying to compress
172
-
.data_block_compression_policy(CompressionPolicy::disabled()),
177
+
.data_block_compression_policy(CompressionPolicy::disabled())
178
+
// ids are sequential and share prefix so we can use large interval to save space
179
+
.data_block_restart_interval_policy(RestartIntervalPolicy::all(64)),
173
180
)?;
174
181
let resync = open_ks(
175
182
"resync",
···
181
188
// did -> error state, so its gonna be basically random, cant compress well
182
189
.data_block_size_policy(BlockSizePolicy::all(kb(4)))
183
190
// and we arent going to have many of these anyway, no point trying
184
-
.data_block_compression_policy(CompressionPolicy::disabled()),
191
+
.data_block_compression_policy(CompressionPolicy::disabled())
192
+
.data_block_restart_interval_policy(RestartIntervalPolicy::all(4)),
185
193
)?;
186
194
let blocks = open_ks(
187
195
"blocks",
···
189
197
// point reads are used a lot by stream, we know the blocks exist though
190
198
.expect_point_read_hits(true)
191
199
.max_memtable_size(mb(cfg.db_blocks_memtable_size_mb))
192
-
// 16 - 128 kb is probably fine, as the newer blocks will be in the first levels
200
+
// 16 - 128 kb, as the newer blocks will be in the first level
193
201
// and any consumers will probably be streaming the newer events...
194
-
// and blocks are pretty big-ish like around 5kb usually so this helps i think
202
+
// and blocks are pretty big-ish like around 5kb...
203
+
// replaying will hit later levels so it will be slower but thats honestly
204
+
// an acceptable tradeoff to save more space...
205
+
// todo: we can probably decrease these when we get zstd dict compression?
195
206
.data_block_size_policy(BlockSizePolicy::new([kb(16), kb(64), kb(128)]))
196
207
// lets not compress first level so the reads for new blocks are faster
197
208
// since we will be streaming them to consumers
198
209
.data_block_compression_policy(CompressionPolicy::new([
199
210
CompressionType::None,
200
211
compression,
201
-
])),
212
+
]))
213
+
.data_block_restart_interval_policy(RestartIntervalPolicy::new([8, 16, 32])),
202
214
)?;
203
215
let records = open_ks(
204
216
"records",
···
208
220
// and since this keyspace is big, turning off bloom filters will help a lot
209
221
.expect_point_read_hits(true)
210
222
.max_memtable_size(mb(cfg.db_records_memtable_size_mb))
211
-
// its just did|col|rkey -> cid, very small
223
+
// its just did|col|rkey -> cid, very small (84 bytes for bsky post)
212
224
.data_block_size_policy(BlockSizePolicy::new([kb(8), kb(16)]))
213
225
// cids arent compressable, most rkeys are TIDs so they will get compressed
214
226
// by prefix truncation anyway
215
-
.data_block_compression_policy(CompressionPolicy::disabled()),
227
+
.data_block_compression_policy(CompressionPolicy::disabled())
228
+
.data_block_restart_interval_policy(RestartIntervalPolicy::all(9)),
216
229
)?;
217
230
let cursors = open_ks(
218
231
"cursors",
···
222
235
.max_memtable_size(mb(4))
223
236
// its just cursors...
224
237
.data_block_size_policy(BlockSizePolicy::all(kb(1)))
225
-
.data_block_compression_policy(CompressionPolicy::disabled()),
238
+
.data_block_compression_policy(CompressionPolicy::disabled())
239
+
.data_block_restart_interval_policy(RestartIntervalPolicy::all(1)),
226
240
)?;
227
241
let resync_buffer = open_ks(
228
242
"resync_buffer",
···
232
246
.max_memtable_size(mb(16))
233
247
.data_block_size_policy(BlockSizePolicy::all(kb(32)))
234
248
// dont have to compress here since resync buffer will be emptied at some point anyway
235
-
.data_block_compression_policy(CompressionPolicy::disabled()),
249
+
.data_block_compression_policy(CompressionPolicy::disabled())
250
+
.data_block_restart_interval_policy(RestartIntervalPolicy::all(16)),
236
251
)?;
237
252
let events = open_ks(
238
253
"events",
···
249
264
.data_block_compression_policy(CompressionPolicy::new([
250
265
CompressionType::None,
251
266
compression,
252
-
])),
267
+
]))
268
+
// ids are int, we can prefix truncate a lot
269
+
.data_block_restart_interval_policy(RestartIntervalPolicy::new([64, 128])),
253
270
)?;
254
271
let counts = open_ks(
255
272
"counts",
···
258
275
.expect_point_read_hits(true)
259
276
.max_memtable_size(mb(16))
260
277
// the data is very small
261
-
// this is at worst did|col -> u64, so its tiny
278
+
// this is at worst did|col -> u64, so its tiny (40 bytes)
262
279
.data_block_size_policy(BlockSizePolicy::all(kb(2)))
263
-
.data_block_compression_policy(CompressionPolicy::disabled()),
280
+
.data_block_compression_policy(CompressionPolicy::disabled())
281
+
.data_block_restart_interval_policy(RestartIntervalPolicy::all(7)),
264
282
)?;
265
283
266
284
// filter handles high-volume point reads (repo excludes) so it needs the bloom filter
···
270
288
.max_memtable_size(mb(16))
271
289
// dids arent compressable so this is fine, and we have nothing as the value
272
290
.data_block_size_policy(BlockSizePolicy::all(kb(1)))
273
-
.data_block_compression_policy(CompressionPolicy::disabled()),
291
+
.data_block_compression_policy(CompressionPolicy::disabled())
292
+
.data_block_restart_interval_policy(RestartIntervalPolicy::all(2)),
274
293
)?;
275
294
276
295
let crawler = open_ks(
···
281
300
.max_memtable_size(mb(16))
282
301
// did -> failed state, not very compressable
283
302
.data_block_size_policy(BlockSizePolicy::all(kb(2)))
284
-
.data_block_compression_policy(CompressionPolicy::disabled()),
303
+
.data_block_compression_policy(CompressionPolicy::disabled())
304
+
.data_block_restart_interval_policy(RestartIntervalPolicy::all(2)),
285
305
)?;
286
306
287
307
// when adding new keyspaces, make sure to add them to the /stats endpoint