1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
//! Transaction brackets for the AOF: the begin/commit markers that make
//! a group of appends replay all-or-nothing, plus the group-commit
//! window they pair with. Split from `aof.rs` to keep that file under
//! the 500-LOC house rule.
//!
//! Why markers and not just group commit: group commit defers the
//! *fsync*, but the AOF writes through a 256 KiB buffer, so a longer
//! transaction still hands whole, valid frames to the kernel as that
//! buffer fills — and `kill -9` leaves them there. Measured before these
//! existed: a 20,000-mutation block killed mid-commit replayed 6,393 of
//! them. The markers move "did this transaction finish" out of "how much
//! got flushed" and into the log itself.
use super::Aof;
use crate::Fsync;
use std::io::{self, Write};
use std::time::Instant;
impl Aof {
/// The two markers that make a group of appends replay atomically.
///
/// Group commit alone only defers the *fsync*; the frames still reach
/// the kernel whenever the write buffer fills, so a `kill -9` during
/// a transaction larger than that buffer leaves a durably
/// half-applied block — measured at 6393/20000 before these markers
/// existed. Bracketing the frames means replay can tell a completed
/// transaction from a torn one regardless of size:
/// [`crate::replay_aof`] buffers everything after a begin marker and
/// applies it only on seeing the matching commit marker, discarding
/// the buffer at EOF.
///
/// They ride as ordinary v2 records holding a one-element multibulk,
/// so the on-disk format is unchanged and older readers see a command
/// they will reject rather than a corrupt frame. The names start with
/// a NUL, which no RESP verb can.
pub const TXN_BEGIN: &'static [u8] = b"\0KEVYTXNBEGIN";
/// See [`Self::TXN_BEGIN`].
pub const TXN_COMMIT: &'static [u8] = b"\0KEVYTXNCOMMIT";
/// Append one of the transaction markers as a bare one-element frame.
fn append_marker(&mut self, name: &'static [u8]) -> io::Result<()> {
let mut frame = Vec::with_capacity(name.len() + 24);
frame.extend_from_slice(b"*1\r\n$");
frame.extend_from_slice(name.len().to_string().as_bytes());
frame.extend_from_slice(b"\r\n");
frame.extend_from_slice(name);
frame.extend_from_slice(b"\r\n");
match self.format {
crate::AofFormat::V2 => {
self.file.write_all(&(frame.len() as u32).to_le_bytes())?;
self.file.write_all(&crate::crc32c::crc32c(&frame).to_le_bytes())?;
self.file.write_all(&frame)?;
}
// v1 has no envelope, so it cannot express a torn-transaction
// boundary; a v1 log stays exactly as atomic as it was. The
// first rewrite upgrades it to v2 and it gains this.
crate::AofFormat::V1 => {}
}
Ok(())
}
/// Open a group-commit window (no-op unless the policy is `Always`):
/// subsequent `append`s buffer instead of fsyncing per command, and
/// the run is bracketed by transaction markers so replay applies it
/// all-or-nothing. Pair with [`Self::end_group`] **before** sending
/// the batch's replies.
#[inline]
pub fn begin_group(&mut self) {
if matches!(self.fsync, Fsync::Always) {
self.deferred = true;
}
self.in_txn = self.append_marker(Self::TXN_BEGIN).is_ok();
}
/// Close the group-commit window: one `flush()+sync_data()` for the
/// whole batch (if anything was buffered), then resume per-command
/// fsync. Must be called before the batch's replies leave the shard.
#[inline]
pub fn end_group(&mut self) -> io::Result<()> {
// Commit marker BEFORE the sync: replay treats a transaction as
// committed only once this record is present and intact, so it
// must be inside the same durable run as the frames it closes.
if self.in_txn {
self.in_txn = false;
self.append_marker(Self::TXN_COMMIT)?;
}
if self.deferred {
self.deferred = false;
if self.dirty {
self.file.flush()?;
self.file.get_ref().sync_data()?;
self.dirty = false;
self.last_sync = Instant::now();
}
}
Ok(())
}
}