aljaz/turso
1
0
Fork 0
mirror of https://github.com/aljazceru/turso.git synced 2026-01-18 15:44:20 +01:00
Code Issues Packages Projects Releases Wiki Activity

Merge 'core/wal: check index header on begin_write_tx' from Pere Diaz Bou

Browse Source 
Fixes a page cache staleness issue where connections could incorrectly
believe the database hasn't changed after checkpointing. This can happen
when writes following a checkpoint resulted in the same `max_frame
value`, causing connections to miss updates since they only checked
`max_frame` to detect changes.

Closes #3502
This commit is contained in:
Pekka Enberg
2025-10-03 13:51:22 +03:00
committed by GitHub
parent 7d45253d03 b5a969933c
commit c98bf9b593
1 changed files with 14 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files

20
core/storage/wal.rs
View File

@@ -849,6 +849,7 @@ impl Wal for WalFile {
.store(lock_0_idx, Ordering::Release);
self.min_frame.store(nbackfills + 1, Ordering::Release);
self.last_checksum = last_checksum;
self.checkpoint_seq.store(checkpoint_seq, Ordering::Release);
return Ok(db_changed);
}
@@ -942,6 +943,8 @@ impl Wal for WalFile {
self.max_frame.store(best_mark as u64, Ordering::Release);
self.max_frame_read_lock_index
.store(best_idx as usize, Ordering::Release);
self.last_checksum = last_checksum;
self.checkpoint_seq.store(checkpoint_seq, Ordering::Release);
tracing::debug!(
"begin_read_tx(min={}, max={}, slot={}, max_frame_in_wal={})",
self.min_frame.load(Ordering::Acquire),
@@ -983,12 +986,17 @@ impl Wal for WalFile {
if !shared.write_lock.write() {
return Err(LimboError::Busy);
}
let (shared_max, nbackfills, last_checksum) = (
shared.max_frame.load(Ordering::Acquire),
shared.nbackfills.load(Ordering::Acquire),
shared.last_checksum,
);
if self.max_frame.load(Ordering::Acquire) == shared_max {
let (shared_max, nbackfills, last_checksum, checkpoint_seq) = {
let mx = shared.max_frame.load(Ordering::Acquire);
let nb = shared.nbackfills.load(Ordering::Acquire);
let ck = shared.last_checksum;
let checkpoint_seq = shared.wal_header.lock().checkpoint_seq;
(mx, nb, ck, checkpoint_seq)
};
let db_changed = shared_max != self.max_frame.load(Ordering::Acquire)
|| last_checksum != self.last_checksum
|| checkpoint_seq != self.checkpoint_seq.load(Ordering::Acquire);
if !db_changed {
// Snapshot still valid; adopt counters
drop(shared);
self.last_checksum = last_checksum;