use std::collections::{HashMap, VecDeque};
use thiserror::Error;
#[derive(Debug, Error, PartialEq, Eq, Clone)]
pub enum TxError {
#[error("transaction {0} not found")]
TransactionNotFound(u64),
#[error("transaction {0} is not active")]
TransactionNotActive(u64),
#[error("transaction {0} has already ended")]
TransactionAlreadyEnded(u64),
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)]
pub struct TransactionId(pub u64);
impl TransactionId {
#[inline]
pub fn as_u64(self) -> u64 {
self.0
}
}
impl std::fmt::Display for TransactionId {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
write!(f, "Tx({})", self.0)
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum TxOperation {
Put {
key: String,
value: Vec<u8>,
prev_value: Option<Vec<u8>>,
},
Delete { key: String, prev_value: Vec<u8> },
BatchPut { entries: Vec<(String, Vec<u8>)> },
}
impl TxOperation {
pub fn key_count(&self) -> usize {
match self {
TxOperation::Put { .. } => 1,
TxOperation::Delete { .. } => 1,
TxOperation::BatchPut { entries } => entries.len(),
}
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
pub enum TransactionStatus {
Active,
Committed,
RolledBack,
Aborted,
}
impl std::fmt::Display for TransactionStatus {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
TransactionStatus::Active => write!(f, "Active"),
TransactionStatus::Committed => write!(f, "Committed"),
TransactionStatus::RolledBack => write!(f, "RolledBack"),
TransactionStatus::Aborted => write!(f, "Aborted"),
}
}
}
#[derive(Debug, Clone)]
pub struct Transaction {
pub id: TransactionId,
pub operations: Vec<TxOperation>,
pub status: TransactionStatus,
pub started_at: u64,
pub ended_at: Option<u64>,
}
impl Transaction {
fn new(id: TransactionId, started_at: u64) -> Self {
Self {
id,
operations: Vec::new(),
status: TransactionStatus::Active,
started_at,
ended_at: None,
}
}
#[inline]
pub fn is_active(&self) -> bool {
self.status == TransactionStatus::Active
}
#[inline]
pub fn operation_count(&self) -> usize {
self.operations.len()
}
}
#[derive(Debug, Clone, PartialEq)]
pub struct TxStats {
pub active_count: usize,
pub committed_count: u64,
pub rolled_back_count: u64,
pub total_operations: u64,
pub avg_ops_per_tx: f64,
}
pub struct StorageTransactionLog {
transactions: VecDeque<Transaction>,
active: HashMap<TransactionId, Transaction>,
next_id: u64,
max_committed: usize,
committed_count: u64,
rolled_back_count: u64,
}
impl StorageTransactionLog {
pub fn new(max_committed: usize) -> Self {
Self {
transactions: VecDeque::new(),
active: HashMap::new(),
next_id: 1,
max_committed,
committed_count: 0,
rolled_back_count: 0,
}
}
pub fn begin(&mut self, now: u64) -> TransactionId {
let id = TransactionId(self.next_id);
self.next_id += 1;
let tx = Transaction::new(id, now);
self.active.insert(id, tx);
id
}
pub fn append(&mut self, tx_id: TransactionId, op: TxOperation) -> Result<(), TxError> {
let tx = self
.active
.get_mut(&tx_id)
.ok_or(TxError::TransactionNotFound(tx_id.0))?;
if !tx.is_active() {
return Err(TxError::TransactionNotActive(tx_id.0));
}
tx.operations.push(op);
Ok(())
}
pub fn commit(&mut self, tx_id: TransactionId, now: u64) -> Result<(), TxError> {
let mut tx = self
.active
.remove(&tx_id)
.ok_or(TxError::TransactionNotFound(tx_id.0))?;
if !tx.is_active() {
self.active.insert(tx_id, tx);
return Err(TxError::TransactionAlreadyEnded(tx_id.0));
}
tx.status = TransactionStatus::Committed;
tx.ended_at = Some(now);
self.committed_count += 1;
if self.max_committed > 0 && self.transactions.len() >= self.max_committed {
self.transactions.pop_front();
}
if self.max_committed > 0 {
self.transactions.push_back(tx);
}
Ok(())
}
pub fn rollback(
&mut self,
tx_id: TransactionId,
now: u64,
) -> Result<Vec<TxOperation>, TxError> {
let mut tx = self
.active
.remove(&tx_id)
.ok_or(TxError::TransactionNotFound(tx_id.0))?;
if !tx.is_active() {
self.active.insert(tx_id, tx);
return Err(TxError::TransactionAlreadyEnded(tx_id.0));
}
tx.status = TransactionStatus::RolledBack;
tx.ended_at = Some(now);
self.rolled_back_count += 1;
let mut ops = tx.operations.clone();
ops.reverse();
Ok(ops)
}
pub fn abort(&mut self, tx_id: TransactionId, now: u64) -> Result<(), TxError> {
let mut tx = self
.active
.remove(&tx_id)
.ok_or(TxError::TransactionNotFound(tx_id.0))?;
if !tx.is_active() {
self.active.insert(tx_id, tx);
return Err(TxError::TransactionAlreadyEnded(tx_id.0));
}
tx.status = TransactionStatus::Aborted;
tx.ended_at = Some(now);
Ok(())
}
pub fn replay_committed(&self, since_id: TransactionId) -> Vec<&Transaction> {
self.transactions
.iter()
.filter(|tx| tx.id > since_id)
.collect()
}
pub fn get_transaction(&self, tx_id: TransactionId) -> Option<&Transaction> {
if let Some(tx) = self.active.get(&tx_id) {
return Some(tx);
}
self.transactions.iter().find(|tx| tx.id == tx_id)
}
pub fn active_transactions(&self) -> Vec<&Transaction> {
let mut txs: Vec<&Transaction> = self.active.values().collect();
txs.sort_by_key(|tx| tx.id);
txs
}
#[inline]
pub fn active_count(&self) -> usize {
self.active.len()
}
#[inline]
pub fn committed_count_total(&self) -> u64 {
self.committed_count
}
#[inline]
pub fn rolled_back_count_total(&self) -> u64 {
self.rolled_back_count
}
pub fn stats(&self) -> TxStats {
let total_operations: u64 = self
.transactions
.iter()
.map(|tx| tx.operations.len() as u64)
.sum();
let deque_len = self.transactions.len();
let avg_ops_per_tx = if deque_len == 0 {
0.0
} else {
total_operations as f64 / deque_len as f64
};
TxStats {
active_count: self.active.len(),
committed_count: self.committed_count,
rolled_back_count: self.rolled_back_count,
total_operations,
avg_ops_per_tx,
}
}
#[inline]
pub fn committed_retained(&self) -> usize {
self.transactions.len()
}
#[inline]
pub fn max_committed(&self) -> usize {
self.max_committed
}
#[inline]
pub fn next_transaction_id(&self) -> TransactionId {
TransactionId(self.next_id)
}
pub fn drain_committed(&mut self) -> Vec<Transaction> {
self.transactions.drain(..).collect()
}
}
#[cfg(test)]
mod tests {
use crate::transaction_log::{
StorageTransactionLog, TransactionId, TransactionStatus, TxError, TxOperation, TxStats,
};
fn put_op(key: &str, value: &[u8]) -> TxOperation {
TxOperation::Put {
key: key.to_string(),
value: value.to_vec(),
prev_value: None,
}
}
fn put_op_with_prev(key: &str, value: &[u8], prev: &[u8]) -> TxOperation {
TxOperation::Put {
key: key.to_string(),
value: value.to_vec(),
prev_value: Some(prev.to_vec()),
}
}
fn delete_op(key: &str, prev: &[u8]) -> TxOperation {
TxOperation::Delete {
key: key.to_string(),
prev_value: prev.to_vec(),
}
}
fn batch_put_op(entries: &[(&str, &[u8])]) -> TxOperation {
TxOperation::BatchPut {
entries: entries
.iter()
.map(|(k, v)| (k.to_string(), v.to_vec()))
.collect(),
}
}
#[test]
fn test_new_log_is_empty() {
let log = StorageTransactionLog::new(100);
assert_eq!(log.active_count(), 0);
assert_eq!(log.committed_count_total(), 0);
assert_eq!(log.rolled_back_count_total(), 0);
assert_eq!(log.committed_retained(), 0);
}
#[test]
fn test_max_committed_preserved() {
let log = StorageTransactionLog::new(42);
assert_eq!(log.max_committed(), 42);
}
#[test]
fn test_next_id_starts_at_one() {
let log = StorageTransactionLog::new(10);
assert_eq!(log.next_transaction_id(), TransactionId(1));
}
#[test]
fn test_begin_returns_incrementing_ids() {
let mut log = StorageTransactionLog::new(10);
let id1 = log.begin(0);
let id2 = log.begin(1);
let id3 = log.begin(2);
assert_eq!(id1, TransactionId(1));
assert_eq!(id2, TransactionId(2));
assert_eq!(id3, TransactionId(3));
}
#[test]
fn test_begin_increments_active_count() {
let mut log = StorageTransactionLog::new(10);
log.begin(0);
assert_eq!(log.active_count(), 1);
log.begin(0);
assert_eq!(log.active_count(), 2);
}
#[test]
fn test_begin_transaction_is_active_status() {
let mut log = StorageTransactionLog::new(10);
let id = log.begin(100);
let tx = log.get_transaction(id).expect("tx must exist");
assert_eq!(tx.status, TransactionStatus::Active);
assert_eq!(tx.started_at, 100);
assert!(tx.ended_at.is_none());
}
#[test]
fn test_append_put_operation() {
let mut log = StorageTransactionLog::new(10);
let id = log.begin(0);
let result = log.append(id, put_op("k", b"v"));
assert!(result.is_ok());
let tx = log.get_transaction(id).expect("tx must exist");
assert_eq!(tx.operations.len(), 1);
}
#[test]
fn test_append_multiple_operations() {
let mut log = StorageTransactionLog::new(10);
let id = log.begin(0);
log.append(id, put_op("a", b"1")).unwrap();
log.append(id, put_op("b", b"2")).unwrap();
log.append(id, delete_op("c", b"old")).unwrap();
let tx = log.get_transaction(id).expect("tx must exist");
assert_eq!(tx.operations.len(), 3);
}
#[test]
fn test_append_unknown_tx_returns_not_found() {
let mut log = StorageTransactionLog::new(10);
let err = log
.append(TransactionId(99), put_op("k", b"v"))
.unwrap_err();
assert_eq!(err, TxError::TransactionNotFound(99));
}
#[test]
fn test_append_batch_put_operation() {
let mut log = StorageTransactionLog::new(10);
let id = log.begin(0);
let op = batch_put_op(&[("x", b"1"), ("y", b"2"), ("z", b"3")]);
log.append(id, op).unwrap();
let tx = log.get_transaction(id).unwrap();
assert_eq!(tx.operations.len(), 1);
if let TxOperation::BatchPut { entries } = &tx.operations[0] {
assert_eq!(entries.len(), 3);
} else {
panic!("expected BatchPut");
}
}
#[test]
fn test_commit_moves_tx_to_committed() {
let mut log = StorageTransactionLog::new(10);
let id = log.begin(0);
log.append(id, put_op("k", b"v")).unwrap();
log.commit(id, 1).unwrap();
assert_eq!(log.active_count(), 0);
assert_eq!(log.committed_count_total(), 1);
assert_eq!(log.committed_retained(), 1);
}
#[test]
fn test_commit_sets_ended_at() {
let mut log = StorageTransactionLog::new(10);
let id = log.begin(0);
log.commit(id, 42).unwrap();
let tx = log.get_transaction(id).expect("committed tx visible");
assert_eq!(tx.ended_at, Some(42));
assert_eq!(tx.status, TransactionStatus::Committed);
}
#[test]
fn test_commit_unknown_tx_returns_not_found() {
let mut log = StorageTransactionLog::new(10);
let err = log.commit(TransactionId(77), 0).unwrap_err();
assert_eq!(err, TxError::TransactionNotFound(77));
}
#[test]
fn test_commit_evicts_oldest_when_at_capacity() {
let mut log = StorageTransactionLog::new(3);
let ids: Vec<_> = (0..4).map(|t| log.begin(t)).collect();
for (i, &id) in ids.iter().enumerate() {
log.commit(id, i as u64 + 10).unwrap();
}
assert_eq!(log.committed_retained(), 3);
assert!(log.get_transaction(TransactionId(1)).is_none());
assert!(log.get_transaction(TransactionId(2)).is_some());
assert!(log.get_transaction(TransactionId(3)).is_some());
assert!(log.get_transaction(TransactionId(4)).is_some());
}
#[test]
fn test_commit_with_max_zero_retains_nothing() {
let mut log = StorageTransactionLog::new(0);
let id = log.begin(0);
log.commit(id, 1).unwrap();
assert_eq!(log.committed_retained(), 0);
assert_eq!(log.committed_count_total(), 1);
}
#[test]
fn test_rollback_returns_ops_in_reverse() {
let mut log = StorageTransactionLog::new(10);
let id = log.begin(0);
log.append(id, put_op("a", b"1")).unwrap();
log.append(id, put_op("b", b"2")).unwrap();
log.append(id, delete_op("c", b"old")).unwrap();
let ops = log.rollback(id, 5).unwrap();
assert_eq!(ops.len(), 3);
assert_eq!(ops[0], delete_op("c", b"old"));
assert_eq!(ops[1], put_op("b", b"2"));
assert_eq!(ops[2], put_op("a", b"1"));
}
#[test]
fn test_rollback_increments_rolled_back_count() {
let mut log = StorageTransactionLog::new(10);
let id = log.begin(0);
log.rollback(id, 1).unwrap();
assert_eq!(log.rolled_back_count_total(), 1);
}
#[test]
fn test_rollback_removes_from_active() {
let mut log = StorageTransactionLog::new(10);
let id = log.begin(0);
log.rollback(id, 1).unwrap();
assert_eq!(log.active_count(), 0);
}
#[test]
fn test_rollback_does_not_add_to_committed_deque() {
let mut log = StorageTransactionLog::new(10);
let id = log.begin(0);
log.rollback(id, 1).unwrap();
assert_eq!(log.committed_retained(), 0);
}
#[test]
fn test_rollback_unknown_tx_returns_not_found() {
let mut log = StorageTransactionLog::new(10);
let err = log.rollback(TransactionId(5), 0).unwrap_err();
assert_eq!(err, TxError::TransactionNotFound(5));
}
#[test]
fn test_rollback_empty_tx_returns_empty_ops() {
let mut log = StorageTransactionLog::new(10);
let id = log.begin(0);
let ops = log.rollback(id, 1).unwrap();
assert!(ops.is_empty());
}
#[test]
fn test_abort_removes_from_active() {
let mut log = StorageTransactionLog::new(10);
let id = log.begin(0);
log.abort(id, 1).unwrap();
assert_eq!(log.active_count(), 0);
}
#[test]
fn test_abort_does_not_increment_committed_count() {
let mut log = StorageTransactionLog::new(10);
let id = log.begin(0);
log.abort(id, 1).unwrap();
assert_eq!(log.committed_count_total(), 0);
}
#[test]
fn test_abort_does_not_increment_rolled_back_count() {
let mut log = StorageTransactionLog::new(10);
let id = log.begin(0);
log.abort(id, 1).unwrap();
assert_eq!(log.rolled_back_count_total(), 0);
}
#[test]
fn test_abort_unknown_tx_returns_not_found() {
let mut log = StorageTransactionLog::new(10);
let err = log.abort(TransactionId(9), 0).unwrap_err();
assert_eq!(err, TxError::TransactionNotFound(9));
}
#[test]
fn test_abort_does_not_retain_in_deque() {
let mut log = StorageTransactionLog::new(10);
let id = log.begin(0);
log.abort(id, 1).unwrap();
assert_eq!(log.committed_retained(), 0);
}
#[test]
fn test_replay_committed_since_zero_returns_all() {
let mut log = StorageTransactionLog::new(10);
let id1 = log.begin(0);
log.commit(id1, 1).unwrap();
let id2 = log.begin(2);
log.commit(id2, 3).unwrap();
let replayed = log.replay_committed(TransactionId(0));
assert_eq!(replayed.len(), 2);
}
#[test]
fn test_replay_committed_filters_by_id() {
let mut log = StorageTransactionLog::new(10);
let id1 = log.begin(0);
log.commit(id1, 1).unwrap();
let id2 = log.begin(2);
log.commit(id2, 3).unwrap();
let id3 = log.begin(4);
log.commit(id3, 5).unwrap();
let replayed = log.replay_committed(id1);
assert_eq!(replayed.len(), 2);
assert_eq!(replayed[0].id, id2);
assert_eq!(replayed[1].id, id3);
}
#[test]
fn test_replay_committed_returns_empty_for_high_since_id() {
let mut log = StorageTransactionLog::new(10);
let id = log.begin(0);
log.commit(id, 1).unwrap();
let replayed = log.replay_committed(TransactionId(999));
assert!(replayed.is_empty());
}
#[test]
fn test_replay_committed_order_is_ascending() {
let mut log = StorageTransactionLog::new(10);
for t in 0..5u64 {
let id = log.begin(t);
log.commit(id, t + 1).unwrap();
}
let replayed = log.replay_committed(TransactionId(0));
let ids: Vec<u64> = replayed.iter().map(|tx| tx.id.0).collect();
let mut sorted = ids.clone();
sorted.sort_unstable();
assert_eq!(ids, sorted);
}
#[test]
fn test_get_transaction_active() {
let mut log = StorageTransactionLog::new(10);
let id = log.begin(7);
let tx = log.get_transaction(id).expect("should find active tx");
assert_eq!(tx.id, id);
assert_eq!(tx.status, TransactionStatus::Active);
}
#[test]
fn test_get_transaction_committed() {
let mut log = StorageTransactionLog::new(10);
let id = log.begin(0);
log.commit(id, 1).unwrap();
let tx = log.get_transaction(id).expect("should find committed tx");
assert_eq!(tx.status, TransactionStatus::Committed);
}
#[test]
fn test_get_transaction_returns_none_for_unknown() {
let log = StorageTransactionLog::new(10);
assert!(log.get_transaction(TransactionId(42)).is_none());
}
#[test]
fn test_active_transactions_sorted_by_id() {
let mut log = StorageTransactionLog::new(10);
let _id3 = log.begin(2);
let _id1 = log.begin(0);
let _id2 = log.begin(1);
let active = log.active_transactions();
let ids: Vec<u64> = active.iter().map(|tx| tx.id.0).collect();
let mut sorted = ids.clone();
sorted.sort_unstable();
assert_eq!(ids, sorted);
}
#[test]
fn test_active_transactions_empty_after_all_committed() {
let mut log = StorageTransactionLog::new(10);
let ids: Vec<_> = (0..3).map(|t| log.begin(t)).collect();
for id in ids {
log.commit(id, 99).unwrap();
}
assert!(log.active_transactions().is_empty());
}
#[test]
fn test_stats_initial() {
let log = StorageTransactionLog::new(10);
let s = log.stats();
assert_eq!(
s,
TxStats {
active_count: 0,
committed_count: 0,
rolled_back_count: 0,
total_operations: 0,
avg_ops_per_tx: 0.0,
}
);
}
#[test]
fn test_stats_reflects_active_and_committed() {
let mut log = StorageTransactionLog::new(10);
let id1 = log.begin(0);
log.append(id1, put_op("a", b"1")).unwrap();
log.append(id1, put_op("b", b"2")).unwrap();
log.commit(id1, 1).unwrap();
let _id2 = log.begin(2);
let s = log.stats();
assert_eq!(s.active_count, 1);
assert_eq!(s.committed_count, 1);
assert_eq!(s.rolled_back_count, 0);
assert_eq!(s.total_operations, 2);
assert!((s.avg_ops_per_tx - 2.0).abs() < f64::EPSILON);
}
#[test]
fn test_stats_avg_ops_zero_when_no_committed_in_deque() {
let mut log = StorageTransactionLog::new(0); let id = log.begin(0);
log.append(id, put_op("k", b"v")).unwrap();
log.commit(id, 1).unwrap();
let s = log.stats();
assert_eq!(s.avg_ops_per_tx, 0.0);
assert_eq!(s.total_operations, 0);
}
#[test]
fn test_tx_operation_key_count_put() {
assert_eq!(put_op("k", b"v").key_count(), 1);
}
#[test]
fn test_tx_operation_key_count_delete() {
assert_eq!(delete_op("k", b"v").key_count(), 1);
}
#[test]
fn test_tx_operation_key_count_batch_put() {
let op = batch_put_op(&[("a", b"1"), ("b", b"2"), ("c", b"3"), ("d", b"4")]);
assert_eq!(op.key_count(), 4);
}
#[test]
fn test_drain_committed_returns_and_clears() {
let mut log = StorageTransactionLog::new(10);
let id1 = log.begin(0);
log.commit(id1, 1).unwrap();
let id2 = log.begin(2);
log.commit(id2, 3).unwrap();
let drained = log.drain_committed();
assert_eq!(drained.len(), 2);
assert_eq!(log.committed_retained(), 0);
}
#[test]
fn test_interleaved_transactions() {
let mut log = StorageTransactionLog::new(10);
let id_a = log.begin(0);
let id_b = log.begin(0);
log.append(id_a, put_op("a", b"va")).unwrap();
log.append(id_b, put_op("b", b"vb")).unwrap();
log.append(id_a, put_op("a2", b"va2")).unwrap();
log.commit(id_b, 1).unwrap();
log.rollback(id_a, 2).unwrap();
assert_eq!(log.committed_count_total(), 1);
assert_eq!(log.rolled_back_count_total(), 1);
assert_eq!(log.active_count(), 0);
}
#[test]
fn test_rollback_with_prev_values() {
let mut log = StorageTransactionLog::new(10);
let id = log.begin(0);
log.append(id, put_op_with_prev("k", b"new", b"old"))
.unwrap();
let ops = log.rollback(id, 1).unwrap();
assert_eq!(ops.len(), 1);
if let TxOperation::Put { prev_value, .. } = &ops[0] {
assert_eq!(prev_value.as_deref(), Some(b"old".as_slice()));
} else {
panic!("expected Put op");
}
}
#[test]
fn test_commit_does_not_affect_other_active_txs() {
let mut log = StorageTransactionLog::new(10);
let id_a = log.begin(0);
let id_b = log.begin(0);
log.commit(id_a, 1).unwrap();
assert_eq!(log.active_count(), 1);
assert!(log.get_transaction(id_b).is_some());
}
#[test]
fn test_large_batch_eviction_boundary() {
let mut log = StorageTransactionLog::new(5);
let ids: Vec<_> = (0..10u64).map(|t| log.begin(t)).collect();
for (i, &id) in ids.iter().enumerate() {
log.commit(id, i as u64 + 100).unwrap();
}
assert_eq!(log.committed_retained(), 5);
assert_eq!(log.committed_count_total(), 10);
let retained = log.replay_committed(TransactionId(0));
let min_id = retained.iter().map(|tx| tx.id.0).min().unwrap_or(0);
assert_eq!(min_id, 6);
}
#[test]
fn test_transaction_id_ordering() {
assert!(TransactionId(1) < TransactionId(2));
assert!(TransactionId(100) > TransactionId(50));
assert_eq!(TransactionId(5), TransactionId(5));
}
#[test]
fn test_transaction_id_display() {
let id = TransactionId(7);
assert_eq!(format!("{id}"), "Tx(7)");
}
#[test]
fn test_tx_error_display() {
assert_eq!(
format!("{}", TxError::TransactionNotFound(3)),
"transaction 3 not found"
);
assert_eq!(
format!("{}", TxError::TransactionNotActive(5)),
"transaction 5 is not active"
);
assert_eq!(
format!("{}", TxError::TransactionAlreadyEnded(9)),
"transaction 9 has already ended"
);
}
#[test]
fn test_many_operations_in_single_tx() {
let mut log = StorageTransactionLog::new(10);
let id = log.begin(0);
for i in 0u64..500 {
log.append(id, put_op(&format!("key-{i}"), b"value"))
.unwrap();
}
let tx = log.get_transaction(id).unwrap();
assert_eq!(tx.operation_count(), 500);
log.commit(id, 1).unwrap();
let s = log.stats();
assert_eq!(s.total_operations, 500);
}
}