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
[ingest] fix tokio panic
+9
-4
+9
-4
src/ingest/worker.rs
+9
-4
src/ingest/worker.rs
···
105
105
let mut buf = Vec::<BufferedMessage>::with_capacity(BUF_SIZE);
106
106
let mut failed = Vec::<BufferedMessage>::new();
107
107
108
+
let _g = handle.enter();
109
+
108
110
loop {
109
111
let mut batch = self.state.db.inner.batch();
110
112
let mut deleted = HashSet::new();
···
187
189
// wait until we receive some messages
188
190
// this does mean we will have an up to 1 second delay, before we send events to consumers
189
191
// but thats reasonable imo, could also be configured of course
190
-
let _ = handle.block_on(tokio::time::timeout(
191
-
Duration::from_secs(1),
192
-
self.rx.recv_many(&mut buf, BUF_SIZE),
193
-
));
192
+
let _ = handle.block_on(async {
193
+
tokio::time::timeout(
194
+
Duration::from_secs(1),
195
+
self.rx.recv_many(&mut buf, BUF_SIZE),
196
+
)
197
+
.await
198
+
});
194
199
if buf.is_empty() {
195
200
if self.rx.is_closed() {
196
201
error!("ingestor crashed? shutting down buffer processor");