forked from
microcosm.blue/Allegedly
Server tools to backfill, tail, mirror, and verify PLC logs
read latest seq from db instead of using current_seq in when polling if we error
+20
+20
src/bin/mirror.rs
+20
src/bin/mirror.rs
···
150
150
// the poll -> stream task: poll until we're caught up, then switch to stream.
151
151
// on stream disconnect, fall back to polling to resync.
152
152
let send_page_bg = send_page.clone();
153
+
let db_for_poll = db.clone();
153
154
tasks.spawn(async move {
154
155
let mut current_seq = latest_seq;
155
156
loop {
···
213
214
Ok(Ok(())) => tracing::info!("stream closed cleanly, resyncing via poll"),
214
215
Ok(Err(e)) => tracing::warn!("stream error: {e}, resyncing via poll"),
215
216
Err(e) => tracing::warn!("stream task join error: {e}"),
217
+
}
218
+
219
+
// rest current_seq to what's actually in the DB. current_seq tracks
220
+
// pages forwarded to seq_pages_to_fjall, which may be ahead of what
221
+
// was actually stored (ops can be dropped by VERIFY=true). polling
222
+
// from the in-memory current_seq would permanently skip those ops.
223
+
let db = db_for_poll.clone();
224
+
match tokio::task::spawn_blocking(move || db.get_latest()).await {
225
+
Ok(Ok(Some((seq, _)))) => {
226
+
if seq < current_seq {
227
+
tracing::info!(
228
+
"resetting poll cursor from {current_seq} to db latest {seq} to avoid skipping dropped ops"
229
+
);
230
+
current_seq = seq;
231
+
}
232
+
}
233
+
Ok(Ok(None)) => {}
234
+
Ok(Err(e)) => tracing::warn!("failed to read db latest for poll reset: {e}"),
235
+
Err(e) => tracing::warn!("spawn_blocking failed for poll reset: {e}"),
216
236
}
217
237
}
218
238
});