forked from
microcosm.blue/Allegedly
Server tools to backfill, tail, mirror, and verify PLC logs
more logging
+29
-5
+23
-5
src/plc_fjall.rs
+23
-5
src/plc_fjall.rs
···
1031
1031
};
1032
1032
1033
1033
for err in &errors {
1034
-
log::warn!("failed to parse operation {} {}: {err}", op.did, op.cid);
1034
+
log::warn!("dropping op {} {} (seq {seq}) parse error: {err}", op.did, op.cid);
1035
1035
}
1036
1036
if !errors.is_empty() {
1037
1037
// if parse failed but not fatal, we just dont store it
···
1069
1069
.map(|e| e.to_string())
1070
1070
.collect::<Vec<_>>()
1071
1071
.join("\n");
1072
-
log::warn!("dropping op {} {} (invalid sig):\n{msg}", op.did, op.cid);
1072
+
log::warn!("dropping op {} {} (seq {seq}) invalid sig:\n{msg}", op.did, op.cid);
1073
1073
return Ok(0);
1074
1074
}
1075
1075
}
1076
1076
Err(e) => {
1077
-
log::warn!("dropping op {} {}: {e}", op.did, op.cid);
1077
+
log::warn!("dropping op {} {} (seq {seq}): {e}", op.did, op.cid);
1078
1078
return Ok(0);
1079
1079
}
1080
1080
}
1081
-
log::debug!("verified op {} {}", op.did, op.cid);
1081
+
log::debug!("verified op {} {} (seq {seq})", op.did, op.cid);
1082
1082
}
1083
1083
1084
1084
let db_op = DbOp {
···
1104
1104
1105
1105
self.inner.notify_stream.notify_waiters();
1106
1106
1107
+
log::debug!("inserted op {} {} (seq {seq})", op.did, op.cid);
1107
1108
Ok(1)
1108
1109
}
1109
1110
···
1520
1521
let mut ops_inserted: usize = 0;
1521
1522
1522
1523
while let Some(page) = pages.recv().await {
1523
-
log::trace!("writing seq page with {} ops", page.ops.len());
1524
+
let first_seq = page.ops.first().map(|op| op.seq);
1525
+
let last_seq = page.ops.last().map(|op| op.seq);
1526
+
log::debug!(
1527
+
"seq_pages: received page with {} ops, seq {:?}..{:?}",
1528
+
page.ops.len(),
1529
+
first_seq,
1530
+
last_seq
1531
+
);
1532
+
let page_len = page.ops.len();
1524
1533
let db = db.clone();
1525
1534
let count = tokio::task::spawn_blocking(move || -> anyhow::Result<usize> {
1526
1535
let mut count: usize = 0;
1527
1536
for seq_op in &page.ops {
1537
+
log::debug!("seq_pages: processing op {} {} (seq {})", seq_op.did, seq_op.cid, seq_op.seq);
1528
1538
let common_op = CommonOp {
1529
1539
did: seq_op.did.clone(),
1530
1540
cid: seq_op.cid.clone(),
···
1538
1548
Ok(count)
1539
1549
})
1540
1550
.await??;
1551
+
if count < page_len {
1552
+
log::warn!(
1553
+
"seq_pages: page seq {:?}..{:?} inserted {count}/{page_len} ops ({} dropped)",
1554
+
first_seq,
1555
+
last_seq,
1556
+
page_len - count
1557
+
);
1558
+
}
1541
1559
ops_inserted += count;
1542
1560
}
1543
1561
+6
src/poll.rs
+6
src/poll.rs
···
339
339
}
340
340
341
341
if !page.is_empty() {
342
+
log::debug!(
343
+
"seq poll: page with {} ops, seq {}..{}",
344
+
page.ops.len(),
345
+
page.ops.first().map(|op| op.seq).unwrap_or(0),
346
+
last_seq
347
+
);
342
348
match dest.try_send(page) {
343
349
Ok(()) => {}
344
350
Err(mpsc::error::TrySendError::Full(page)) => {