forked from
microcosm.blue/microcosm-rs
Constellation, Spacedust, Slingshot, UFOs: atproto crates and services for microcosm
chill out rollups if we're caught up
+6
-13
-10
ufos/src/main.rs
-10
ufos/src/main.rs
···
124
124
let rolling = write_store.background_tasks()?.run();
125
125
let storing = write_store.receive_batches(batches);
126
126
127
-
// let storing = tokio::task::spawn_blocking(move || {
128
-
// while let Some(event_batch) = batches.blocking_recv() {
129
-
// write_store.insert_batch(event_batch)?;
130
-
// write_store
131
-
// .step_rollup()
132
-
// .inspect_err(|e| log::error!("rollup error: {e:?}"))?;
133
-
// }
134
-
// Ok::<(), StorageError>(())
135
-
// });
136
-
137
127
tokio::select! {
138
128
z = serving => log::warn!("serve task ended: {z:?}"),
139
129
z = rolling => log::warn!("rollup task ended: {z:?}"),
+5
-2
ufos/src/storage_fjall.rs
+5
-2
ufos/src/storage_fjall.rs
···
862
862
async fn run(mut self) -> StorageResult<()> {
863
863
let mut dirty_nsids = HashSet::new();
864
864
865
-
let mut rollup = tokio::time::interval(Duration::from_millis(240));
865
+
let mut rollup = tokio::time::interval(Duration::from_millis(81));
866
866
rollup.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
867
867
868
-
let mut trim = tokio::time::interval(Duration::from_millis(3_000));
868
+
let mut trim = tokio::time::interval(Duration::from_millis(6_000));
869
869
trim.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
870
870
871
871
loop {
872
872
tokio::select! {
873
873
_ = rollup.tick() => {
874
874
let (n, dirty) = self.0.step_rollup().inspect_err(|e| log::error!("rollup error: {e:?}"))?;
875
+
if n == 0 {
876
+
rollup.reset_after(Duration::from_millis(1_200)); // we're caught up, take a break
877
+
}
875
878
dirty_nsids.extend(dirty);
876
879
log::info!("rolled up {n} items ({} collections now dirty)", dirty_nsids.len());
877
880
},