Constellation, Spacedust, Slingshot, UFOs: atproto crates and services for microcosm
split out an interval for more rw tasks
and sketch out record trimming
author
phil
date
(Mar 27, 2025, 4:55 PM -0400)
commit
1791653e
1791653e15707c17fd720cea667066a714576cab
parent
8879de5a
8879de5a21eb8f4296be77920698cc93af9a0232
+111
-57
+1
-1
ufos/src/consumer.rs
+1
-1
ufos/src/consumer.rs
···
22
22
const MAX_BATCH_SPAN_SECS: f64 = 60.; // hard limit of duration from oldest to latest event cursor within a batch, in seconds.
23
23
24
24
const SEND_TIMEOUT_S: f64 = 60.;
25
-
const BATCH_QUEUE_SIZE: usize = 1024; // 4096 got OOM'd
25
+
const BATCH_QUEUE_SIZE: usize = 512; // 4096 got OOM'd. update: 1024 also got OOM'd during L0 compaction blocking
26
26
27
27
#[derive(Debug)]
28
28
struct Batcher {
+110
-56
ufos/src/store.rs
+110
-56
ufos/src/store.rs
···
16
16
use std::path::{Path, PathBuf};
17
17
use std::time::{Duration, Instant};
18
18
use tokio::sync::mpsc::Receiver;
19
-
use tokio::time::sleep;
19
+
use tokio::time::{interval_at, sleep};
20
20
21
21
/// Commit the RW batch immediately if this number of events have been read off the mod queue
22
22
const MAX_BATCHED_RW_EVENTS: usize = 18;
···
155
155
/// Read-write loop reads from the queue for record-modifying events and does rollups
156
156
pub async fn rw_loop(&self) -> anyhow::Result<()> {
157
157
// TODO: lock so that only one rw loop can possibly be run. or even better, take a mutable resource thing to enforce at compile time.
158
-
loop {
159
-
sleep(Duration::from_secs_f64(0.1)).await; // todo: interval rate-limit instead
160
158
161
-
let db = &self.db;
162
-
let keyspace = db.keyspace.clone();
163
-
let partition = db.partition.clone();
159
+
let now = tokio::time::Instant::now();
160
+
let mut time_to_update_events = interval_at(now, Duration::from_secs_f64(0.051));
161
+
let mut time_to_trim_surplus = interval_at(
162
+
now + Duration::from_secs_f64(1.0),
163
+
Duration::from_secs_f64(3.3),
164
+
);
165
+
let mut time_to_roll_up = interval_at(
166
+
now + Duration::from_secs_f64(0.4),
167
+
Duration::from_secs_f64(0.9),
168
+
);
164
169
165
-
log::trace!("rw: spawn blocking for batch...");
166
-
tokio::task::spawn_blocking(move || -> anyhow::Result<()> {
167
-
log::trace!("rw: getting rw cursor...");
168
-
let mod_cursor = get_static::<ModCursorKey, ModCursorValue>(&partition)?
169
-
.unwrap_or(Cursor::from_start());
170
-
let range = ModQueueItemKey::new(mod_cursor.clone()).range_to_prefix_end()?;
170
+
time_to_update_events.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
171
+
time_to_trim_surplus.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
172
+
time_to_roll_up.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
171
173
172
-
let mut db_batch = keyspace.batch();
173
-
let mut batched_rw_items = 0;
174
-
let mut any_tasks_found = false;
174
+
loop {
175
+
let keyspace = self.db.keyspace.clone();
176
+
let partition = self.db.partition.clone();
177
+
tokio::select! {
178
+
_ = time_to_update_events.tick() => {
179
+
log::debug!("beginning event update task");
180
+
tokio::task::spawn_blocking(move || Self::update_events(keyspace, partition)).await??;
181
+
log::debug!("finished event update task");
182
+
}
183
+
_ = time_to_trim_surplus.tick() => {
184
+
log::debug!("beginning record trim task");
185
+
tokio::task::spawn_blocking(move || Self::trim_old_events(keyspace, partition)).await??;
186
+
log::debug!("finished record trim task");
187
+
}
188
+
_ = time_to_roll_up.tick() => {
189
+
log::debug!("beginning rollup task");
190
+
tokio::task::spawn_blocking(move || Self::roll_up_counts(keyspace, partition)).await??;
191
+
log::debug!("finished rollup task");
192
+
},
193
+
}
194
+
}
195
+
}
175
196
176
-
log::trace!("rw: iterating newer rw items...");
197
+
fn update_events(keyspace: Keyspace, partition: PartitionHandle) -> anyhow::Result<()> {
198
+
// TODO: lock this to prevent concurrent rw
177
199
178
-
for (i, pair) in partition.range(range.clone()).enumerate() {
179
-
log::trace!("rw: iterating {i}");
180
-
any_tasks_found = true;
200
+
log::trace!("rw: getting rw cursor...");
201
+
let mod_cursor =
202
+
get_static::<ModCursorKey, ModCursorValue>(&partition)?.unwrap_or(Cursor::from_start());
203
+
let range = ModQueueItemKey::new(mod_cursor.clone()).range_to_prefix_end()?;
181
204
182
-
if i >= MAX_BATCHED_RW_EVENTS {
183
-
break;
184
-
}
205
+
let mut db_batch = keyspace.batch();
206
+
let mut batched_rw_items = 0;
207
+
let mut any_tasks_found = false;
185
208
186
-
let (key_bytes, val_bytes) = pair?;
187
-
let mod_key = match db_complete::<ModQueueItemKey>(&key_bytes) {
188
-
Ok(k) => k,
189
-
Err(EncodingError::WrongStaticPrefix(_, _)) => {
190
-
panic!("wsp: mod queue empty.");
191
-
}
192
-
otherwise => otherwise?,
193
-
};
209
+
log::trace!("rw: iterating newer rw items...");
194
210
195
-
let mod_value: ModQueueItemValue =
196
-
db_complete::<ModQueueItemStringValue>(&val_bytes)?.try_into()?;
211
+
for (i, pair) in partition.range(range.clone()).enumerate() {
212
+
log::trace!("rw: iterating {i}");
213
+
any_tasks_found = true;
197
214
198
-
log::trace!("rw: iterating {i}: sending to batcher {mod_key:?} => {mod_value:?}");
199
-
batched_rw_items += DBWriter {
200
-
keyspace: keyspace.clone(),
201
-
partition: partition.clone(),
202
-
}
203
-
.write_rw(&mut db_batch, mod_key, mod_value)?;
204
-
log::trace!("rw: iterating {i}: back from batcher.");
215
+
if i >= MAX_BATCHED_RW_EVENTS {
216
+
break;
217
+
}
205
218
206
-
if batched_rw_items >= MAX_BATCHED_RW_ITEMS {
207
-
log::trace!("rw: iterating {i}: batch big enough, breaking out.");
208
-
break;
209
-
}
219
+
let (key_bytes, val_bytes) = pair?;
220
+
let mod_key = match db_complete::<ModQueueItemKey>(&key_bytes) {
221
+
Ok(k) => k,
222
+
Err(EncodingError::WrongStaticPrefix(_, _)) => {
223
+
panic!("wsp: mod queue empty.");
210
224
}
225
+
otherwise => otherwise?,
226
+
};
227
+
228
+
let mod_value: ModQueueItemValue =
229
+
db_complete::<ModQueueItemStringValue>(&val_bytes)?.try_into()?;
211
230
212
-
if !any_tasks_found {
213
-
log::trace!("rw: skipping batch commit since apparently no items were added (this is normal, skipping is new)");
214
-
return Ok(());
215
-
}
231
+
log::trace!("rw: iterating {i}: sending to batcher {mod_key:?} => {mod_value:?}");
232
+
batched_rw_items += DBWriter {
233
+
keyspace: keyspace.clone(),
234
+
partition: partition.clone(),
235
+
}
236
+
.write_rw(&mut db_batch, mod_key, mod_value)?;
237
+
log::trace!("rw: iterating {i}: back from batcher.");
216
238
217
-
log::info!("rw: committing rw batch with {batched_rw_items} items (items != total inserts/deletes)...");
218
-
let r = db_batch.commit();
219
-
log::info!("rw: commit result: {r:?}");
220
-
r?;
221
-
Ok(())
222
-
})
223
-
.await??;
224
-
log::trace!("rw: back from blocking for rw...");
239
+
if batched_rw_items >= MAX_BATCHED_RW_ITEMS {
240
+
log::trace!("rw: iterating {i}: batch big enough, breaking out.");
241
+
break;
242
+
}
225
243
}
226
-
// log::warn!("exited rw loop (rw task)");
244
+
245
+
if !any_tasks_found {
246
+
log::trace!("rw: skipping batch commit since apparently no items were added (this is normal, skipping is new)");
247
+
// TODO: is this missing a chance to update the cursor?
248
+
return Ok(());
249
+
}
250
+
251
+
log::info!("rw: committing rw batch with {batched_rw_items} items (items != total inserts/deletes)...");
252
+
let r = db_batch.commit();
253
+
log::info!("rw: commit result: {r:?}");
254
+
r?;
255
+
Ok(())
256
+
}
257
+
258
+
fn trim_old_events(_keyspace: Keyspace, _partition: PartitionHandle) -> anyhow::Result<()> {
259
+
// we *could* keep a collection dirty list in memory to reduce the amount of searching here
260
+
// actually can we use seen_by_js_cursor_collection??
261
+
// * ["seen_by_js_cursor_collection"|js_cursor|collection] => u64
262
+
// -> the rollup cursor could handle trims.
263
+
264
+
// key structure:
265
+
// * ["by_collection"|collection|js_cursor] => [did|rkey|record]
266
+
267
+
// *new* strategy:
268
+
// 1. collect `collection`s seen during rollup
269
+
// 2. for each collected collection:
270
+
// 3. set up prefix iterator
271
+
// 4. reverse and try to walk back MAX_RETAINED steps
272
+
// 5. if we didn't end iteration yet, start deleting records (and their forward links) until we get to the end
273
+
274
+
// ... we can probably do even better with cursor ranges too, since we'll have a cursor range from rollup and it's in the by_collection key
275
+
276
+
Ok(())
277
+
}
278
+
279
+
fn roll_up_counts(_keyspace: Keyspace, _partition: PartitionHandle) -> anyhow::Result<()> {
280
+
Ok(())
227
281
}
228
282
229
283
pub async fn get_collection_records(