[fix](trx-server): offload history persistence
This commit is contained in:
@@ -185,20 +185,38 @@ pub fn flush_all(db: &mut PickleDb, rig_id: &str, histories: &Arc<DecoderHistori
|
|||||||
let _ = db.dump();
|
let _ = db.dump();
|
||||||
}
|
}
|
||||||
|
|
||||||
|
fn flush_all_rigs(db: &Mutex<PickleDb>, rig_histories: &[(String, Arc<DecoderHistories>)]) {
|
||||||
|
let Ok(mut guard) = db.lock() else {
|
||||||
|
tracing::warn!("history database mutex poisoned; skipping periodic flush");
|
||||||
|
return;
|
||||||
|
};
|
||||||
|
for (rig_id, histories) in rig_histories {
|
||||||
|
flush_all(&mut guard, rig_id, histories);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
/// Spawn a Tokio task that flushes all rigs' histories to disk every 60 seconds.
|
/// Spawn a Tokio task that flushes all rigs' histories to disk every 60 seconds.
|
||||||
|
///
|
||||||
|
/// Snapshot cloning, JSON serialization, and disk I/O run on Tokio's blocking
|
||||||
|
/// pool so a large history database cannot stall an async runtime worker.
|
||||||
pub fn spawn_flush_task(
|
pub fn spawn_flush_task(
|
||||||
db: Arc<Mutex<PickleDb>>,
|
db: Arc<Mutex<PickleDb>>,
|
||||||
rig_histories: Vec<(String, Arc<DecoderHistories>)>,
|
rig_histories: Vec<(String, Arc<DecoderHistories>)>,
|
||||||
) {
|
) {
|
||||||
tokio::spawn(async move {
|
tokio::spawn(async move {
|
||||||
|
let rig_histories = Arc::new(rig_histories);
|
||||||
let mut interval = tokio::time::interval(Duration::from_secs(60));
|
let mut interval = tokio::time::interval(Duration::from_secs(60));
|
||||||
interval.tick().await; // consume the immediate first tick
|
interval.tick().await; // consume the immediate first tick
|
||||||
loop {
|
loop {
|
||||||
interval.tick().await;
|
interval.tick().await;
|
||||||
if let Ok(mut guard) = db.lock() {
|
let db = Arc::clone(&db);
|
||||||
for (rig_id, histories) in &rig_histories {
|
let rig_histories = Arc::clone(&rig_histories);
|
||||||
flush_all(&mut guard, rig_id, histories);
|
if let Err(err) = tokio::task::spawn_blocking(move || {
|
||||||
}
|
flush_all_rigs(&db, rig_histories.as_slice());
|
||||||
|
})
|
||||||
|
.await
|
||||||
|
{
|
||||||
|
tracing::warn!(error = %err, "history flush worker failed");
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
});
|
});
|
||||||
|
|||||||
Reference in New Issue
Block a user