///|
/// WriteBatch stores a sequence of database operations to be committed atomically.
pub struct WriteBatch {
db : DB
// (key, value, ttl, has_ttl); None value means Delete.
ops : Array[(String, Bytes?, Int64, Bool)]
}
///|
/// Creates a new WriteBatch for the database.
pub fn DB::new_write_batch(self : DB) -> WriteBatch {
{ db: self, ops: Array::new() }
}
///|
/// Adds a Put operation to the batch.
pub fn WriteBatch::put(self : WriteBatch, key : String, value : Bytes) -> Unit {
self.ops.push((key, Some(value), 0L, false))
}
///|
/// Adds a Put operation with TTL to the batch.
pub fn WriteBatch::put_ttl(
self : WriteBatch,
key : String,
value : Bytes,
ttl : Int64,
) -> Unit {
self.ops.push((key, Some(value), ttl, true))
}
///|
/// Adds a Delete operation to the batch.
pub fn WriteBatch::delete(self : WriteBatch, key : String) -> Unit {
self.ops.push((key, None, 0L, false))
}
///|
/// Commits all operations in the batch atomically to the WAL log.
pub fn WriteBatch::commit(self : WriteBatch) -> Unit raise MoonKVError {
if self.ops.length() == 0 {
return
}
// Validate the complete batch before writing its first record. This keeps
// invalid input from creating a partially committed transaction.
for op in self.ops {
let (key, _, ttl, has_ttl) = op
validate_user_key(key)
if has_ttl && ttl <= 0L {
raise MoonKVError::InvalidRecord("ttl must be greater than zero")
}
}
// 1. Allocate a unique transaction ID from the logical clock
let tx_id = self.db.logical_clock
self.db.logical_clock = self.db.logical_clock + 1L
let entries = Array::new()
// 2. Write all operations prefixed with transaction markers
for op in self.ops {
let (key, val_opt, ttl, has_ttl) = op
let tx_key = "__tx__:" + tx_id.to_string() + ":" + key
let timestamp = self.db.logical_clock
self.db.logical_clock = self.db.logical_clock + 1L
let record = match val_opt {
Some(val) =>
if has_ttl {
Record::new_ttl(string_to_bytes(tx_key), val, timestamp, ttl)
} else {
Record::new(string_to_bytes(tx_key), val, timestamp)
}
None => Record::new_tombstone(string_to_bytes(tx_key), timestamp)
}
let record_bytes = record.serialize()
let record_len = record_bytes.length()
// Roll file if size limit exceeded
if self.db.active_file_offset + record_len > self.db.max_file_size {
self.db.active_file_id = self.db.active_file_id + 1
self.db.active_file_offset = 0
}
let path = self.db.dir + "/" + self.db.active_file_id.to_string() + ".data"
append_bytes_to_file(path, record_bytes)
// Buffer KeyEntry metadata
let is_put = match val_opt {
Some(_) => true
None => false
}
let val_len = match val_opt {
Some(v) => v.length()
None => 0
}
let expires_at = if has_ttl { timestamp + ttl } else { 0L }
let entry = KeyEntry::new(
self.db.active_file_id,
val_len,
self.db.active_file_offset.to_int64(),
timestamp,
expires_at,
)
entries.push((key, is_put, entry))
self.db.active_file_offset = self.db.active_file_offset + record_len
}
// 3. Write the commit marker record to make the transaction active
let commit_key = "__tx_commit__:" + tx_id.to_string()
let timestamp = self.db.logical_clock
self.db.logical_clock = self.db.logical_clock + 1L
let commit_record = Record::new(
string_to_bytes(commit_key),
Bytes::make(0, (0).to_byte()),
timestamp,
)
let commit_bytes = commit_record.serialize()
let commit_len = commit_bytes.length()
if self.db.active_file_offset + commit_len > self.db.max_file_size {
self.db.active_file_id = self.db.active_file_id + 1
self.db.active_file_offset = 0
}
let path = self.db.dir + "/" + self.db.active_file_id.to_string() + ".data"
append_bytes_to_file(path, commit_bytes)
self.db.active_file_offset = self.db.active_file_offset + commit_len
// 4. Safely update the in-memory Keydir now that it's committed on disk
for ent in entries {
let (key, is_put, entry) = ent
if is_put {
self.db.keydir.put(key, entry)
} else {
self.db.keydir.delete(key)
}
}
}