forked from
microcosm.blue/microcosm-rs
Constellation, Spacedust, Slingshot, UFOs: atproto crates and services for microcosm
Merge pull request #19 from at-microcosm/ufos-jetstream-backfill
UFOs: jetstream backfill
+263
-70
+6
-9
Cargo.lock
+6
-9
Cargo.lock
···
225
225
226
226
[[package]]
227
227
name = "atrium-api"
228
-
version = "0.25.2"
229
-
source = "registry+https://github.com/rust-lang/crates.io-index"
230
-
checksum = "0d4eb9b4787aba546015c8ccda1d3924c157cee13d67848997fba74ac8144a07"
228
+
version = "0.25.3"
229
+
source = "git+https://github.com/uniphil/atrium?branch=fix%2Fnsid-allow-nonleading-name-digits#c4364f318d337bbc3e3e3aaf97c9f971e95f5f7e"
231
230
dependencies = [
232
231
"atrium-common",
233
232
"atrium-xrpc",
···
245
244
246
245
[[package]]
247
246
name = "atrium-common"
248
-
version = "0.1.1"
249
-
source = "registry+https://github.com/rust-lang/crates.io-index"
250
-
checksum = "ba30d2f9e1a8b3db8fc97d0a5f91ee5a28f8acdddb771ad74c1b08eda357ca3d"
247
+
version = "0.1.2"
248
+
source = "git+https://github.com/uniphil/atrium?branch=fix%2Fnsid-allow-nonleading-name-digits#c4364f318d337bbc3e3e3aaf97c9f971e95f5f7e"
251
249
dependencies = [
252
250
"dashmap",
253
251
"lru",
···
260
258
261
259
[[package]]
262
260
name = "atrium-xrpc"
263
-
version = "0.12.2"
264
-
source = "registry+https://github.com/rust-lang/crates.io-index"
265
-
checksum = "18a9e526cb2ed3e0a2ca78c3ce2a943d9041a68e067dadf42923b523771e07df"
261
+
version = "0.12.3"
262
+
source = "git+https://github.com/uniphil/atrium?branch=fix%2Fnsid-allow-nonleading-name-digits#c4364f318d337bbc3e3e3aaf97c9f971e95f5f7e"
266
263
dependencies = [
267
264
"http",
268
265
"serde",
+1
-1
jetstream/Cargo.toml
+1
-1
jetstream/Cargo.toml
···
10
10
11
11
[dependencies]
12
12
async-trait = "0.1.83"
13
-
atrium-api = { version = "0.25.2", default-features = false, features = [
13
+
atrium-api = { git = "https://github.com/uniphil/atrium", branch = "fix/nsid-allow-nonleading-name-digits", default-features = false, features = [
14
14
"namespace-appbsky",
15
15
] }
16
16
tokio = { version = "1.44.2", features = ["full", "sync", "time"] }
+2
-2
jetstream/src/lib.rs
+2
-2
jetstream/src/lib.rs
···
439
439
440
440
if let Some(last) = last_cursor {
441
441
if event_cursor <= *last {
442
-
log::warn!("event cursor {event_cursor:?} was older than the last one: {last:?}. dropping event.");
442
+
log::warn!("event cursor {event_cursor:?} was not newer than the last one: {last:?}. dropping event.");
443
443
continue;
444
444
}
445
445
}
···
475
475
476
476
if let Some(last) = last_cursor {
477
477
if event_cursor <= *last {
478
-
log::warn!("event cursor {event_cursor:?} was older than the last one: {last:?}. dropping event.");
478
+
log::warn!("event cursor {event_cursor:?} was not newer than the last one: {last:?}. dropping event.");
479
479
continue;
480
480
}
481
481
}
+32
-12
ufos/src/consumer.rs
+32
-12
ufos/src/consumer.rs
···
8
8
use std::mem;
9
9
use std::time::Duration;
10
10
use tokio::sync::mpsc::{channel, Receiver, Sender};
11
+
use tokio::time::{timeout, Interval};
11
12
12
13
use crate::error::{BatchInsertError, FirehoseEventError};
13
14
use crate::{DeleteAccount, EventBatch, UFOsCommit};
···
34
35
batch_sender: Sender<LimitedBatch>,
35
36
current_batch: CurrentBatch,
36
37
sketch_secret: SketchSecretPrefix,
38
+
rate_limit: Interval,
37
39
}
38
40
39
41
pub async fn consume(
···
64
66
.await?;
65
67
let (batch_sender, batch_reciever) = channel::<LimitedBatch>(BATCH_QUEUE_SIZE);
66
68
let mut batcher = Batcher::new(jetstream_receiver, batch_sender, sketch_secret);
67
-
tokio::task::spawn(async move { batcher.run().await });
69
+
tokio::task::spawn(async move {
70
+
let r = batcher.run().await;
71
+
log::info!("batcher ended: {r:?}");
72
+
});
68
73
Ok(batch_reciever)
69
74
}
70
75
···
74
79
batch_sender: Sender<LimitedBatch>,
75
80
sketch_secret: SketchSecretPrefix,
76
81
) -> Self {
82
+
let mut rate_limit = tokio::time::interval(std::time::Duration::from_millis(5));
83
+
rate_limit.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
77
84
Self {
78
85
jetstream_receiver,
79
86
batch_sender,
80
87
current_batch: Default::default(),
81
88
sketch_secret,
89
+
rate_limit,
82
90
}
83
91
}
84
92
85
93
pub async fn run(&mut self) -> anyhow::Result<()> {
94
+
// TODO: report errors *from here* probably, since this gets shipped off into a spawned task that might just vanish
86
95
loop {
87
-
if let Some(event) = self.jetstream_receiver.recv().await {
88
-
self.handle_event(event).await?
89
-
} else {
90
-
anyhow::bail!("channel closed");
96
+
match timeout(Duration::from_millis(9_000), self.jetstream_receiver.recv()).await {
97
+
Err(_elapsed) => self.no_events_step().await?,
98
+
Ok(Some(event)) => self.handle_event(event).await?,
99
+
Ok(None) => anyhow::bail!("channel closed"),
91
100
}
92
101
}
93
102
}
94
103
104
+
async fn no_events_step(&mut self) -> anyhow::Result<()> {
105
+
let empty = self.current_batch.batch.is_empty();
106
+
log::info!("no events received, stepping batcher (empty? {empty})");
107
+
if !empty {
108
+
self.send_current_batch_now(true, "no events step").await?;
109
+
}
110
+
Ok(())
111
+
}
112
+
95
113
async fn handle_event(&mut self, event: JetstreamEvent) -> anyhow::Result<()> {
96
114
if let Some(earliest) = &self.current_batch.initial_cursor {
97
115
if event.cursor.duration_since(earliest)? > Duration::from_secs_f64(MAX_BATCH_SPAN_SECS)
98
116
{
99
-
self.send_current_batch_now(false).await?;
117
+
self.send_current_batch_now(false, "time since event")
118
+
.await?;
100
119
}
101
120
} else {
102
121
self.current_batch.initial_cursor = Some(event.cursor);
···
126
145
if event.cursor.duration_since(earliest)?.as_secs_f64() > MIN_BATCH_SPAN_SECS
127
146
&& self.batch_sender.capacity() == BATCH_QUEUE_SIZE
128
147
{
129
-
self.send_current_batch_now(true).await?;
148
+
self.send_current_batch_now(true, "available queue").await?;
130
149
}
131
150
}
132
151
Ok(())
···
141
160
);
142
161
143
162
if let Err(BatchInsertError::BatchFull(commit)) = optimistic_res {
144
-
self.send_current_batch_now(false).await?;
163
+
self.send_current_batch_now(false, "handle commit").await?;
145
164
self.current_batch.batch.insert_commit_by_nsid(
146
165
&collection,
147
166
commit,
···
157
176
158
177
async fn handle_delete_account(&mut self, did: Did, cursor: Cursor) -> anyhow::Result<()> {
159
178
if self.current_batch.batch.account_removes.len() >= MAX_ACCOUNT_REMOVES {
160
-
self.send_current_batch_now(false).await?;
179
+
self.send_current_batch_now(false, "delete account").await?;
161
180
}
162
181
self.current_batch
163
182
.batch
···
168
187
169
188
// holds up all consumer progress until it can send to the channel
170
189
// use this when the current batch is too full to add more to it
171
-
async fn send_current_batch_now(&mut self, small: bool) -> anyhow::Result<()> {
190
+
async fn send_current_batch_now(&mut self, small: bool, referrer: &str) -> anyhow::Result<()> {
172
191
let beginning = match self.current_batch.initial_cursor.map(|c| c.elapsed()) {
173
192
None => "unknown".to_string(),
174
193
Some(Ok(t)) => format!("{:?}", t),
175
194
Some(Err(e)) => format!("+{:?}", e.duration()),
176
195
};
177
-
log::info!(
178
-
"sending batch now from {beginning}, {}, queue capacity: {}",
196
+
log::trace!(
197
+
"sending batch now from {beginning}, {}, queue capacity: {}, referrer: {referrer}",
179
198
if small { "small" } else { "full" },
180
199
self.batch_sender.capacity(),
181
200
);
182
201
let current = mem::take(&mut self.current_batch);
202
+
self.rate_limit.tick().await;
183
203
self.batch_sender
184
204
.send_timeout(current.batch, Duration::from_secs_f64(SEND_TIMEOUT_S))
185
205
.await?;
+24
-8
ufos/src/file_consumer.rs
+24
-8
ufos/src/file_consumer.rs
···
12
12
async fn read_jsonl(f: File, sender: Sender<JetstreamEvent>) -> Result<()> {
13
13
let mut lines = BufReader::new(f).lines();
14
14
while let Some(line) = lines.next_line().await? {
15
-
let event: JetstreamEvent =
16
-
serde_json::from_str(&line).map_err(JetstreamEventError::ReceivedMalformedJSON)?;
17
-
if sender.send(event).await.is_err() {
18
-
log::warn!("All receivers for the jsonl fixture have been dropped, bye.");
19
-
return Err(JetstreamEventError::ReceiverClosedError.into());
15
+
match serde_json::from_str::<JetstreamEvent>(&line) {
16
+
Ok(event) => match sender.send(event).await {
17
+
Ok(_) => {}
18
+
Err(e) => {
19
+
log::warn!("All receivers for the jsonl fixture have been dropped, bye: {e:?}");
20
+
return Err(JetstreamEventError::ReceiverClosedError.into());
21
+
}
22
+
},
23
+
Err(parse_err) => {
24
+
log::warn!("failed to parse event: {parse_err:?} from event:\n{line}");
25
+
continue;
26
+
}
20
27
}
21
28
}
22
-
Ok(())
29
+
log::info!("reached end of jsonl file, looping on noop to keep server alive.");
30
+
loop {
31
+
tokio::time::sleep(std::time::Duration::from_secs_f64(10.)).await;
32
+
}
23
33
}
24
34
25
35
pub async fn consume(
···
30
40
let (jsonl_sender, jsonl_receiver) = channel::<JetstreamEvent>(16);
31
41
let (batch_sender, batch_reciever) = channel::<LimitedBatch>(BATCH_QUEUE_SIZE);
32
42
let mut batcher = Batcher::new(jsonl_receiver, batch_sender, sketch_secret);
33
-
tokio::task::spawn(async move { read_jsonl(f, jsonl_sender).await });
34
-
tokio::task::spawn(async move { batcher.run().await });
43
+
tokio::task::spawn(async move {
44
+
let r = read_jsonl(f, jsonl_sender).await;
45
+
log::info!("read_jsonl finished: {r:?}");
46
+
});
47
+
tokio::task::spawn(async move {
48
+
let r = batcher.run().await;
49
+
log::info!("batcher finished: {r:?}");
50
+
});
35
51
Ok(batch_reciever)
36
52
}
+1
ufos/src/lib.rs
+1
ufos/src/lib.rs
+143
-4
ufos/src/main.rs
+143
-4
ufos/src/main.rs
···
1
1
use clap::Parser;
2
2
use jetstream::events::Cursor;
3
3
use std::path::PathBuf;
4
+
use std::time::{Duration, SystemTime};
4
5
use ufos::consumer;
5
6
use ufos::file_consumer;
6
7
use ufos::server;
···
8
9
use ufos::storage_fjall::FjallStorage;
9
10
use ufos::storage_mem::MemStorage;
10
11
use ufos::store_types::SketchSecretPrefix;
12
+
use ufos::ConsumerInfo;
11
13
12
14
#[cfg(not(target_env = "msvc"))]
13
15
use tikv_jemallocator::Jemalloc;
···
36
38
#[arg(long)]
37
39
data: PathBuf,
38
40
/// DEBUG: don't start the jetstream consumer or its write loop
39
-
/// todo: restore this
40
41
#[arg(long, action)]
41
42
pause_writer: bool,
43
+
/// Adjust runtime settings like background task intervals for efficient backfill
44
+
#[arg(long, action)]
45
+
backfill: bool,
42
46
/// DEBUG: force the rw loop to fall behind by pausing it
43
47
/// todo: restore this
44
48
#[arg(long, action)]
···
46
50
/// DEBUG: use an in-memory store instead of fjall
47
51
#[arg(long, action)]
48
52
in_mem: bool,
53
+
/// reset the rollup cursor, scrape through missed things in the past (backfill)
54
+
#[arg(long, action)]
55
+
reroll: bool,
49
56
/// DEBUG: interpret jetstream as a file fixture
50
57
#[arg(long, action)]
51
58
jetstream_fixture: bool,
···
68
75
args.jetstream,
69
76
args.jetstream_fixture,
70
77
args.pause_writer,
78
+
args.backfill,
79
+
args.reroll,
71
80
read_store,
72
81
write_store,
73
82
cursor,
···
85
94
args.jetstream,
86
95
args.jetstream_fixture,
87
96
args.pause_writer,
97
+
args.backfill,
98
+
args.reroll,
88
99
read_store,
89
100
write_store,
90
101
cursor,
···
96
107
Ok(())
97
108
}
98
109
110
+
#[allow(clippy::too_many_arguments)]
99
111
async fn go<B: StoreBackground>(
100
112
jetstream: String,
101
113
jetstream_fixture: bool,
102
114
pause_writer: bool,
103
-
read_store: impl StoreReader + 'static,
115
+
backfill: bool,
116
+
reroll: bool,
117
+
read_store: impl StoreReader + 'static + Clone,
104
118
mut write_store: impl StoreWriter<B> + 'static,
105
119
cursor: Option<Cursor>,
106
120
sketch_secret: SketchSecretPrefix,
107
121
) -> anyhow::Result<()> {
108
122
println!("starting server with storage...");
109
-
let serving = server::serve(read_store);
123
+
let serving = server::serve(read_store.clone());
110
124
111
125
if pause_writer {
112
126
log::info!("not starting jetstream or the write loop.");
···
125
139
consumer::consume(&jetstream, cursor, false, sketch_secret).await?
126
140
};
127
141
128
-
let rolling = write_store.background_tasks()?.run();
142
+
let rolling = write_store.background_tasks(reroll)?.run(backfill);
129
143
let storing = write_store.receive_batches(batches);
144
+
145
+
let stating = do_update_stuff(read_store);
130
146
131
147
tokio::select! {
132
148
z = serving => log::warn!("serve task ended: {z:?}"),
133
149
z = rolling => log::warn!("rollup task ended: {z:?}"),
134
150
z = storing => log::warn!("storage task ended: {z:?}"),
151
+
z = stating => log::warn!("status task ended: {z:?}"),
135
152
};
136
153
137
154
println!("bye!");
138
155
139
156
Ok(())
140
157
}
158
+
159
+
async fn do_update_stuff(read_store: impl StoreReader) {
160
+
let started_at = std::time::SystemTime::now();
161
+
let mut first_cursor = None;
162
+
let mut first_rollup = None;
163
+
let mut last_at = std::time::SystemTime::now();
164
+
let mut last_cursor = None;
165
+
let mut last_rollup = None;
166
+
loop {
167
+
tokio::time::sleep(std::time::Duration::from_secs_f64(4.)).await;
168
+
match read_store.get_consumer_info().await {
169
+
Err(e) => log::warn!("failed to get jetstream consumer info: {e:?}"),
170
+
Ok(ConsumerInfo::Jetstream {
171
+
latest_cursor,
172
+
rollup_cursor,
173
+
..
174
+
}) => {
175
+
let now = std::time::SystemTime::now();
176
+
let latest_cursor = latest_cursor.map(Cursor::from_raw_u64);
177
+
let rollup_cursor = rollup_cursor.map(Cursor::from_raw_u64);
178
+
backfill_info(
179
+
latest_cursor,
180
+
rollup_cursor,
181
+
last_cursor,
182
+
last_rollup,
183
+
last_at,
184
+
first_cursor,
185
+
first_rollup,
186
+
started_at,
187
+
now,
188
+
);
189
+
first_cursor = first_cursor.or(latest_cursor);
190
+
first_rollup = first_rollup.or(rollup_cursor);
191
+
last_cursor = latest_cursor;
192
+
last_rollup = rollup_cursor;
193
+
last_at = now;
194
+
}
195
+
}
196
+
}
197
+
}
198
+
199
+
#[allow(clippy::too_many_arguments)]
200
+
fn backfill_info(
201
+
latest_cursor: Option<Cursor>,
202
+
rollup_cursor: Option<Cursor>,
203
+
last_cursor: Option<Cursor>,
204
+
last_rollup: Option<Cursor>,
205
+
last_at: SystemTime,
206
+
first_cursor: Option<Cursor>,
207
+
first_rollup: Option<Cursor>,
208
+
started_at: SystemTime,
209
+
now: SystemTime,
210
+
) {
211
+
let nice_duration = |dt: Duration| {
212
+
let secs = dt.as_secs_f64();
213
+
if secs < 1. {
214
+
return format!("{:.0}ms", secs * 1000.);
215
+
}
216
+
if secs < 60. {
217
+
return format!("{secs:.02}s");
218
+
}
219
+
let mins = (secs / 60.).floor();
220
+
let rsecs = secs - (mins * 60.);
221
+
if mins < 60. {
222
+
return format!("{mins:.0}m{rsecs:.0}s");
223
+
}
224
+
let hrs = (mins / 60.).floor();
225
+
let rmins = mins - (hrs * 60.);
226
+
if hrs < 24. {
227
+
return format!("{hrs:.0}h{rmins:.0}m{rsecs:.0}s");
228
+
}
229
+
let days = (hrs / 24.).floor();
230
+
let rhrs = hrs - (days * 24.);
231
+
format!("{days:.0}d{rhrs:.0}h{rmins:.0}m{rsecs:.0}s")
232
+
};
233
+
234
+
let nice_dt_two_maybes = |earlier: Option<Cursor>, later: Option<Cursor>| match (earlier, later)
235
+
{
236
+
(Some(earlier), Some(later)) => match later.duration_since(&earlier) {
237
+
Ok(dt) => nice_duration(dt),
238
+
Err(e) => {
239
+
let rev_dt = e.duration();
240
+
format!("+{}", nice_duration(rev_dt))
241
+
}
242
+
},
243
+
_ => "unknown".to_string(),
244
+
};
245
+
246
+
let rate = |mlatest: Option<Cursor>, msince: Option<Cursor>, real: Duration| {
247
+
mlatest
248
+
.zip(msince)
249
+
.map(|(latest, since)| {
250
+
latest
251
+
.duration_since(&since)
252
+
.unwrap_or(Duration::from_millis(1))
253
+
})
254
+
.map(|dtc| format!("{:.2}", dtc.as_secs_f64() / real.as_secs_f64()))
255
+
.unwrap_or("??".into())
256
+
};
257
+
258
+
let dt_real = now
259
+
.duration_since(last_at)
260
+
.unwrap_or(Duration::from_millis(1));
261
+
262
+
let dt_real_total = now
263
+
.duration_since(started_at)
264
+
.unwrap_or(Duration::from_millis(1));
265
+
266
+
let cursor_rate = rate(latest_cursor, last_cursor, dt_real);
267
+
let cursor_avg = rate(latest_cursor, first_cursor, dt_real_total);
268
+
269
+
let rollup_rate = rate(rollup_cursor, last_rollup, dt_real);
270
+
let rollup_avg = rate(rollup_cursor, first_rollup, dt_real_total);
271
+
272
+
log::info!(
273
+
"cursor: {} behind (→{}, {cursor_rate}x, {cursor_avg}x avg). rollup: {} behind (→{}, {rollup_rate}x, {rollup_avg}x avg).",
274
+
latest_cursor.map(|c| c.elapsed().map(nice_duration).unwrap_or("++".to_string())).unwrap_or("?".to_string()),
275
+
nice_dt_two_maybes(last_cursor, latest_cursor),
276
+
rollup_cursor.map(|c| c.elapsed().map(nice_duration).unwrap_or("++".to_string())).unwrap_or("?".to_string()),
277
+
nice_dt_two_maybes(last_rollup, rollup_cursor),
278
+
);
279
+
}
+4
-3
ufos/src/storage.rs
+4
-3
ufos/src/storage.rs
···
26
26
where
27
27
Self: 'static,
28
28
{
29
-
fn background_tasks(&mut self) -> StorageResult<B>;
29
+
fn background_tasks(&mut self, reroll: bool) -> StorageResult<B>;
30
30
31
31
fn receive_batches<const LIMIT: usize>(
32
32
mut self,
···
53
53
54
54
fn step_rollup(&mut self) -> StorageResult<(usize, HashSet<Nsid>)>;
55
55
56
-
fn trim_collection(&mut self, collection: &Nsid, limit: usize) -> StorageResult<()>;
56
+
fn trim_collection(&mut self, collection: &Nsid, limit: usize)
57
+
-> StorageResult<(usize, usize)>;
57
58
58
59
fn delete_account(&mut self, did: &Did) -> StorageResult<usize>;
59
60
}
60
61
61
62
#[async_trait]
62
63
pub trait StoreBackground: Send + Sync {
63
-
async fn run(mut self) -> StorageResult<()>;
64
+
async fn run(mut self, backfill: bool) -> StorageResult<()>;
64
65
}
65
66
66
67
#[async_trait]
+41
-27
ufos/src/storage_fjall.rs
+41
-27
ufos/src/storage_fjall.rs
···
149
149
let keyspace = {
150
150
let config = Config::new(path);
151
151
152
-
#[cfg(not(test))]
153
-
let config = config.fsync_ms(Some(4_000));
152
+
// #[cfg(not(test))]
153
+
// let config = config.fsync_ms(Some(4_000));
154
154
155
155
config.open()?
156
156
};
···
187
187
)?;
188
188
} else {
189
189
return Err(StorageError::InitError(format!(
190
-
"stored js_endpoint {stored:?} differs from provided {endpoint:?}, refusing to start.")));
190
+
"stored js_endpoint {stored:?} differs from provided {endpoint:?}, refusing to start without --jetstream-force.")));
191
191
}
192
192
}
193
193
stored_secret
···
360
360
get_snapshot_static_neu::<JetstreamCursorKey, JetstreamCursorValue>(&global)?
361
361
.map(|c| c.to_raw_u64());
362
362
363
+
let rollup_cursor =
364
+
get_snapshot_static_neu::<NewRollupCursorKey, NewRollupCursorValue>(&global)?
365
+
.map(|c| c.to_raw_u64());
366
+
363
367
Ok(ConsumerInfo::Jetstream {
364
368
endpoint,
365
369
started_at,
366
370
latest_cursor,
371
+
rollup_cursor,
367
372
})
368
373
}
369
374
···
801
806
}
802
807
803
808
impl StoreWriter<FjallBackground> for FjallWriter {
804
-
fn background_tasks(&mut self) -> StorageResult<FjallBackground> {
809
+
fn background_tasks(&mut self, reroll: bool) -> StorageResult<FjallBackground> {
805
810
if self.bg_taken.swap(true, Ordering::SeqCst) {
806
811
Err(StorageError::BackgroundAlreadyStarted)
807
812
} else {
813
+
if reroll {
814
+
insert_static_neu::<NewRollupCursorKey>(&self.global, Cursor::from_start())?;
815
+
}
808
816
Ok(FjallBackground(self.clone()))
809
817
}
810
818
}
···
948
956
Ok((cursors_stepped, dirty_nsids))
949
957
}
950
958
951
-
fn trim_collection(&mut self, collection: &Nsid, limit: usize) -> StorageResult<()> {
959
+
fn trim_collection(
960
+
&mut self,
961
+
collection: &Nsid,
962
+
limit: usize,
963
+
) -> StorageResult<(usize, usize)> {
952
964
let mut dangling_feed_keys_cleaned = 0;
953
965
let mut records_deleted = 0;
954
966
···
967
979
let mut live_records_found = 0;
968
980
let mut latest_expired_feed_cursor = None;
969
981
let mut batch = self.keyspace.batch();
970
-
for kv in self.feeds.range(live_range).rev() {
982
+
for (i, kv) in self.feeds.range(live_range).rev().enumerate() {
983
+
if i > 1_000_000 {
984
+
log::info!("stopping collection trim early: already scanned 1M elements");
985
+
break;
986
+
}
971
987
let (key_bytes, val_bytes) = kv?;
972
988
let feed_key = db_complete::<NsidRecordFeedKey>(&key_bytes)?;
973
989
let feed_val = db_complete::<NsidRecordFeedVal>(&val_bytes)?;
···
1022
1038
1023
1039
batch.commit()?;
1024
1040
1025
-
log::info!("trim_collection ({collection:?}) removed {dangling_feed_keys_cleaned} dangling feed entries and {records_deleted} records");
1026
-
Ok(())
1041
+
log::trace!("trim_collection ({collection:?}) removed {dangling_feed_keys_cleaned} dangling feed entries and {records_deleted} records");
1042
+
Ok((dangling_feed_keys_cleaned, records_deleted))
1027
1043
}
1028
1044
1029
1045
fn delete_account(&mut self, did: &Did) -> Result<usize, StorageError> {
···
1048
1064
1049
1065
#[async_trait]
1050
1066
impl StoreBackground for FjallBackground {
1051
-
async fn run(mut self) -> StorageResult<()> {
1067
+
async fn run(mut self, backfill: bool) -> StorageResult<()> {
1052
1068
let mut dirty_nsids = HashSet::new();
1053
1069
1054
-
let mut rollup = tokio::time::interval(Duration::from_millis(81));
1070
+
let mut rollup =
1071
+
tokio::time::interval(Duration::from_millis(if backfill { 1 } else { 81 }));
1055
1072
rollup.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
1056
1073
1057
-
let mut trim = tokio::time::interval(Duration::from_millis(6_000));
1074
+
let mut trim =
1075
+
tokio::time::interval(Duration::from_millis(if backfill { 3_000 } else { 6_000 }));
1058
1076
trim.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
1059
1077
1060
1078
loop {
···
1065
1083
rollup.reset_after(Duration::from_millis(1_200)); // we're caught up, take a break
1066
1084
}
1067
1085
dirty_nsids.extend(dirty);
1068
-
log::info!("rolled up {n} items ({} collections now dirty)", dirty_nsids.len());
1086
+
log::trace!("rolled up {n} items ({} collections now dirty)", dirty_nsids.len());
1069
1087
},
1070
1088
_ = trim.tick() => {
1071
-
log::info!("trimming {} nsids: {dirty_nsids:?}", dirty_nsids.len());
1089
+
let n = dirty_nsids.len();
1090
+
log::trace!("trimming {n} nsids: {dirty_nsids:?}");
1072
1091
let t0 = Instant::now();
1092
+
let (mut total_danglers, mut total_deleted) = (0, 0);
1073
1093
for collection in &dirty_nsids {
1074
-
self.0.trim_collection(collection, 512).inspect_err(|e| log::error!("trim error: {e:?}"))?;
1094
+
let (danglers, deleted) = self.0.trim_collection(collection, 512).inspect_err(|e| log::error!("trim error: {e:?}"))?;
1095
+
total_danglers += danglers;
1096
+
total_deleted += deleted;
1097
+
if total_deleted > 1_000_000 {
1098
+
log::info!("trim stopped early, more than 1M records already deleted.");
1099
+
break;
1100
+
}
1075
1101
}
1076
-
log::info!("finished trimming in {:?}", t0.elapsed());
1102
+
log::info!("finished trimming {n} nsids in {:?}: {total_danglers} dangling and {total_deleted} total removed.", t0.elapsed());
1077
1103
dirty_nsids.clear();
1078
1104
},
1079
1105
};
···
1154
1180
}
1155
1181
1156
1182
////////// temp stuff to remove:
1157
-
1158
-
// fn summarize_batch<const LIMIT: usize>(batch: &EventBatch<LIMIT>) -> String {
1159
-
// format!(
1160
-
// "batch of {: >3} samples from {: >4} records in {: >2} collections from ~{: >4} DIDs, {} acct removes, cursor {: <12?}",
1161
-
// batch.total_records(),
1162
-
// batch.total_seen(),
1163
-
// batch.total_collections(),
1164
-
// batch.estimate_dids(),
1165
-
// batch.account_removes(),
1166
-
// batch.latest_cursor().map(|c| c.elapsed()),
1167
-
// )
1168
-
// }
1169
1183
1170
1184
#[cfg(test)]
1171
1185
mod tests {
+9
-4
ufos/src/storage_mem.rs
+9
-4
ufos/src/storage_mem.rs
···
441
441
get_snapshot_static_neu::<JetstreamCursorKey, JetstreamCursorValue>(&global)?
442
442
.map(|c| c.to_raw_u64());
443
443
444
+
let rollup_cursor =
445
+
get_snapshot_static_neu::<NewRollupCursorKey, JetstreamCursorValue>(&global)?
446
+
.map(|c| c.to_raw_u64());
447
+
444
448
Ok(ConsumerInfo::Jetstream {
445
449
endpoint,
446
450
started_at,
447
451
latest_cursor,
452
+
rollup_cursor,
448
453
})
449
454
}
450
455
···
797
802
}
798
803
799
804
impl StoreWriter<MemBackground> for MemWriter {
800
-
fn background_tasks(&mut self) -> StorageResult<MemBackground> {
805
+
fn background_tasks(&mut self, _reroll: bool) -> StorageResult<MemBackground> {
801
806
Ok(MemBackground {})
802
807
}
803
808
···
959
964
collection: &Nsid,
960
965
limit: usize,
961
966
// TODO: could add a start cursor limit to avoid iterating deleted stuff at the start (/end)
962
-
) -> StorageResult<()> {
967
+
) -> StorageResult<(usize, usize)> {
963
968
let mut dangling_feed_keys_cleaned = 0;
964
969
let mut records_deleted = 0;
965
970
···
1015
1020
batch.commit()?;
1016
1021
1017
1022
log::info!("trim_collection ({collection:?}) removed {dangling_feed_keys_cleaned} dangling feed entries and {records_deleted} records");
1018
-
Ok(())
1023
+
Ok((dangling_feed_keys_cleaned, records_deleted))
1019
1024
}
1020
1025
1021
1026
fn delete_account(&mut self, did: &Did) -> Result<usize, StorageError> {
···
1040
1045
1041
1046
#[async_trait]
1042
1047
impl StoreBackground for MemBackground {
1043
-
async fn run(mut self) -> StorageResult<()> {
1048
+
async fn run(mut self, _backfill: bool) -> StorageResult<()> {
1044
1049
// noop for mem (is there a nicer way to do this?)
1045
1050
loop {
1046
1051
tokio::time::sleep(std::time::Duration::from_secs_f64(10.)).await;