fix python tests

This commit is contained in:
boqiu 2024-08-05 16:07:31 +08:00
parent c85d8ff7ed
commit 73c56629a5
2 changed files with 11 additions and 5 deletions

View File

@ -284,7 +284,12 @@ impl SerialBatcher {
let origin = self.next_tx_seq_in_db.load(Ordering::Relaxed);
let mut current = origin;
while let Some(&sync_result) = self.pending_completed_txs.read().await.get(&current) {
loop {
let sync_result = match self.pending_completed_txs.read().await.get(&current) {
Some(&v) => v,
None => break,
};
// downgrade to random sync if file sync failed or timeout
if matches!(sync_result, SyncResult::Failed | SyncResult::Timeout) {
self.sync_store.add_pending_tx(current).await?;

View File

@ -48,15 +48,16 @@ impl AutoSyncManager {
);
// sync randomly
let random = RandomBatcher::new(config, store.clone(), sync_send.clone(), sync_store);
let random = RandomBatcher::new(config, store, sync_send, sync_store);
executor.spawn(random.clone().start(catched_up.clone()), "auto_sync_random");
// handle on catched up notification
executor.spawn(
async move {
catch_up_end_recv.await.expect("Catch up sender dropped");
if catch_up_end_recv.await.is_ok() {
info!("log entry catched up");
catched_up.store(true, Ordering::Relaxed);
}
},
"auto_sync_wait_for_catchup",
);