forked from
microcosm.blue/Allegedly
Server tools to backfill, tail, mirror, and verify PLC logs
fjall: increment cursor +1 so we dont send the same op again on next wake up
+19
-22
+18
-22
src/mirror/fjall.rs
+18
-22
src/mirror/fjall.rs
···
1
-
use std::sync::Arc;
2
-
3
1
use super::*;
4
2
use futures::{SinkExt as _, StreamExt as _};
5
3
use poem::IntoResponse;
···
306
304
Data(FjallState { fjall, .. }): Data<&FjallState>,
307
305
) -> poem::Result<impl IntoResponse> {
308
306
use poem::web::websocket::Message;
309
-
use tokio::sync::Notify;
310
307
311
308
let db = fjall.clone();
312
309
···
358
355
};
359
356
360
357
Ok(ws.on_upgrade(move |mut socket| async move {
361
-
let errored = Arc::new(Notify::new());
362
-
363
358
loop {
364
359
let (tx, mut op_rx) = tokio::sync::mpsc::channel(64);
365
360
366
-
tokio::task::spawn_blocking({
361
+
let read = tokio::task::spawn_blocking({
367
362
let db = db.clone();
368
-
let errored = errored.clone();
369
-
move || {
370
-
let iter = match db.export_ops(cursor..) {
371
-
Ok(it) => it,
372
-
Err(e) => {
373
-
log::error!("read failed: {e}");
374
-
errored.notify_one();
375
-
return;
376
-
}
377
-
};
378
-
for op in iter.flatten() {
363
+
move || -> anyhow::Result<()> {
364
+
for op in db.export_ops(cursor..)?.flatten() {
379
365
if tx.blocking_send(op).is_err() {
380
-
return;
366
+
return Ok(());
381
367
}
382
368
}
369
+
Ok(())
383
370
}
384
371
});
385
372
···
389
376
log::warn!("closing export stream: {e}");
390
377
return;
391
378
}
392
-
cursor = op.seq;
379
+
cursor = op.seq + 1;
393
380
}
394
381
395
-
tokio::select! {
396
-
_ = db.subscribe() => {},
397
-
_ = errored.notified() => return,
382
+
match read.await {
383
+
Ok(Err(e)) => {
384
+
log::error!("stream read failed: {e}");
385
+
return;
386
+
}
387
+
Err(e) => {
388
+
log::error!("stream read task panicked: {e}");
389
+
return;
390
+
}
391
+
Ok(Ok(())) => {}
398
392
}
393
+
394
+
db.subscribe().await;
399
395
}
400
396
}))
401
397
}
+1
src/plc_fjall.rs
+1
src/plc_fjall.rs
···
841
841
}
842
842
843
843
impl Op {
844
+
// todo: we should probably just have this in Op tbh as a `r#type: SequencedOpType` or something
844
845
/// adds the `type` field to the op
845
846
pub fn to_sequenced_json(&self) -> serde_json::Value {
846
847
let mut val = serde_json::to_value(self).expect("Op is serializable");