forked from
microcosm.blue/microcosm-rs
Constellation, Spacedust, Slingshot, UFOs: atproto crates and services for microcosm
try not to die for longer
+4
-4
+2
-2
ufos/src/consumer.rs
+2
-2
ufos/src/consumer.rs
···
18
18
pub const MAX_BATCHED_COLLECTIONS: usize = 64; // hard limit, MAX_BATCHED_RECORDS applies per-collection
19
19
pub const MIN_BATCH_SPAN_SECS: f64 = 2.; // breathe
20
20
pub const MAX_BATCH_SPAN_SECS: f64 = 60.; // hard limit, pause consumer if we're unable to send by now
21
-
pub const SEND_TIMEOUT_S: f64 = 15.; // if the channel is blocked longer than this, something is probably up
22
-
pub const BATCH_QUEUE_SIZE: usize = 1; // nearly-rendez-vous
21
+
pub const SEND_TIMEOUT_S: f64 = 150.; // if the channel is blocked longer than this, something is probably up
22
+
pub const BATCH_QUEUE_SIZE: usize = 64; // used to be 1, but sometimes inserts are just really slow????????
23
23
24
24
pub type LimitedBatch = EventBatch<MAX_BATCHED_RECORDS>;
25
25
+2
-2
ufos/src/storage.rs
+2
-2
ufos/src/storage.rs
···
43
43
let mut concerned = false;
44
44
loop {
45
45
tokio::select! {
46
-
_ = tokio::time::sleep(Duration::from_secs_f64(1.)) => {
46
+
_ = tokio::time::sleep(Duration::from_secs_f64(3.)) => {
47
47
log::warn!("taking a long time to insert an event batch ({:?})...", started.elapsed());
48
48
concerned = true;
49
49
}
50
50
_ = cancelled.cancelled() => {
51
51
if concerned {
52
-
log::warn!("finally inserted slow event batch after {:?}", started.elapsed());
52
+
log::warn!("finally inserted slow event batch (or failed) after {:?}", started.elapsed());
53
53
}
54
54
break
55
55
}