use super::{RawCommitMarker, marker_envelope::encode_commit_marker_bytes};
use crate::{
db::{
MutationJobAdvanceRequest, MutationJobId, MutationJobIdempotencyKey, MutationJobPhase,
MutationJobStatus,
commit::marker::{
COMMIT_MARKER_FORMAT_VERSION_CURRENT, CommitMarker, DatabaseControlOp,
MAX_COMMIT_BYTES, commit_marker_payload_capacity, decode_commit_marker_payload,
encode_commit_marker_payload,
},
data::{DecodedDataStoreKey, RawDataStoreKey},
integrity::MutationProgressRecordOp,
journal::{
DatabaseCommitSequence, JournalBatch, JournalRecord, JournalSequence,
encode_journal_batch,
},
key_taxonomy::{PrimaryKeyComponent, PrimaryKeyValue},
mutation_job::{MutationJobRecord, MutationJobTransition},
},
error::{ErrorClass, ErrorOrigin},
testing::test_memory,
types::EntityTag,
};
use ic_stable_structures::Memory;
fn encode_test_marker_payload(marker: &CommitMarker) -> Vec<u8> {
let payload =
encode_commit_marker_payload(marker).expect("test marker payload encode should succeed");
encode_commit_marker_bytes(COMMIT_MARKER_FORMAT_VERSION_CURRENT, &payload)
.expect("test marker envelope encode should succeed")
}
fn raw_data_store_key(fill: u8) -> RawDataStoreKey {
DecodedDataStoreKey::new_primary_key_value(
EntityTag::new(1),
&PrimaryKeyValue::from(PrimaryKeyComponent::Nat64(u64::from(fill))),
)
.to_raw()
.expect("test key should materialize")
}
fn maximal_mutation_progress_operation() -> MutationProgressRecordOp {
let job_id = MutationJobId::try_from_bytes([0x61; 32])
.expect("nonzero marker mutation job id should admit");
let initial = MutationJobRecord::new(
job_id,
vec![1; crate::db::MAX_MUTATION_JOB_INTENT_BYTES],
vec![2; crate::db::MAX_MUTATION_JOB_CONTINUATION_BYTES],
)
.expect("maximum initial marker record should admit");
let transition = |sequence, key: &str| {
MutationJobAdvanceRequest::new(
job_id,
sequence,
MutationJobIdempotencyKey::new(key).expect("maximum marker replay key should admit"),
)
};
let key = "k".repeat(crate::db::MAX_MUTATION_JOB_IDEMPOTENCY_KEY_BYTES);
let (before, _) = initial
.apply_transition(
&transition(0, &key),
MutationJobTransition::new(
MutationJobStatus::Active,
MutationJobPhase::Forward,
vec![3; crate::db::MAX_MUTATION_JOB_CONTINUATION_BYTES],
crate::db::MAX_MUTATION_JOB_STEP_KEYS_SCANNED,
crate::db::MAX_MUTATION_JOB_STEP_ROWS_UPDATED,
0,
),
)
.expect("maximum predecessor marker record should admit");
let (after, _) = before
.apply_transition(
&transition(1, &key),
MutationJobTransition::new(
MutationJobStatus::Active,
MutationJobPhase::Forward,
vec![4; crate::db::MAX_MUTATION_JOB_CONTINUATION_BYTES],
crate::db::MAX_MUTATION_JOB_STEP_KEYS_SCANNED,
crate::db::MAX_MUTATION_JOB_STEP_ROWS_UPDATED,
0,
),
)
.expect("maximum successor marker record should admit");
MutationProgressRecordOp::replace(&before, &after)
.expect("maximum marker progress replacement should admit")
}
#[test]
fn commit_control_slot_rejects_corrupt_magic() {
let store = super::CommitStore::init(test_memory(233));
let mut malformed = super::CommitStore::encode_raw_control_slot_for_tests(Vec::new())
.expect("current empty commit-control slot should encode");
malformed[0] = b'X';
store.set_raw_marker_bytes_for_tests(malformed);
let err = store
.marker_is_empty()
.expect_err("corrupt control-slot magic should fail closed");
assert_eq!(err.class, ErrorClass::IncompatiblePersistedFormat);
assert_eq!(err.origin, ErrorOrigin::Serialize);
}
#[test]
fn commit_marker_empty_bytes_decode_as_absent_marker() {
let decoded = RawCommitMarker(Vec::new())
.try_decode()
.expect("empty marker bytes should decode as marker absence");
assert!(
decoded.is_none(),
"empty marker bytes are the explicit no-marker durable state",
);
}
#[test]
fn commit_marker_rejects_truncated_envelope_header() {
let err = RawCommitMarker(vec![COMMIT_MARKER_FORMAT_VERSION_CURRENT])
.try_decode()
.expect_err("truncated marker envelope header should fail closed");
assert_eq!(err.class, ErrorClass::Corruption);
assert_eq!(err.origin, ErrorOrigin::Store);
}
#[test]
fn commit_marker_rejects_truncated_envelope_payload() {
let mut bytes = Vec::new();
bytes.push(COMMIT_MARKER_FORMAT_VERSION_CURRENT);
bytes.extend_from_slice(&4u32.to_le_bytes());
bytes.push(0xAA);
let err = RawCommitMarker(bytes)
.try_decode()
.expect_err("truncated marker envelope payload should fail closed");
assert_eq!(err.class, ErrorClass::Corruption);
assert_eq!(err.origin, ErrorOrigin::Store);
}
#[test]
fn commit_marker_rejects_trailing_payload_bytes() {
let marker = CommitMarker {
id: [0u8; 16],
journal_batches: Vec::new(),
database_control: Vec::new(),
};
let mut bytes = encode_test_marker_payload(&marker);
bytes.push(0xFF);
let err = RawCommitMarker(bytes)
.try_decode()
.expect_err("trailing payload bytes should be rejected");
assert_eq!(err.class, ErrorClass::Corruption);
assert_eq!(err.origin, ErrorOrigin::Store);
}
#[test]
fn commit_marker_current_version_round_trip_succeeds() {
let marker = CommitMarker {
id: [9u8; 16],
journal_batches: Vec::new(),
database_control: Vec::new(),
};
let encoded = encode_test_marker_payload(&marker);
let decoded = RawCommitMarker(encoded)
.try_decode()
.expect("current-version marker envelope should decode")
.expect("marker payload should be present");
assert_eq!(decoded.id, marker.id);
assert!(decoded.journal_batches().is_empty());
}
#[test]
fn current_commit_marker_round_trips_one_bounded_mutation_progress_effect() {
assert_eq!(COMMIT_MARKER_FORMAT_VERSION_CURRENT, 1);
let operation = maximal_mutation_progress_operation();
assert_eq!(operation.before_bytes().len(), 18_842);
assert_eq!(operation.after_bytes().len(), 18_842);
let empty = CommitMarker::from_parts([0x71; 16], Vec::new())
.expect("empty comparison marker should admit");
let marker =
CommitMarker::from_parts_with_mutation_progress([0x71; 16], Vec::new(), operation.clone())
.expect("bounded mutation progress marker should admit");
assert_eq!(
commit_marker_payload_capacity(&marker) - commit_marker_payload_capacity(&empty),
37_797,
"the maximum current replacement contribution is frozen",
);
let encoded = encode_test_marker_payload(&marker);
let decoded = RawCommitMarker(encoded)
.try_decode()
.expect("current marker should decode")
.expect("current marker should remain present");
let [DatabaseControlOp::MutationProgress(decoded)] = decoded.database_control() else {
panic!("marker must contain exactly one mutation progress replacement");
};
assert_eq!(decoded.key(), operation.key());
assert_eq!(decoded.job_id(), operation.job_id());
assert_eq!(decoded.expected_sequence(), operation.expected_sequence());
assert_eq!(decoded.before_bytes(), operation.before_bytes());
assert_eq!(decoded.after_bytes(), operation.after_bytes());
}
#[test]
fn commit_marker_rejects_more_than_four_database_control_operations_before_allocation() {
let mut payload = vec![0u8; 16];
payload.extend_from_slice(&0u32.to_le_bytes());
payload.push(5);
let err = decode_commit_marker_payload(&payload)
.expect_err("an oversized database-control transaction must fail closed");
assert_eq!(err.class, ErrorClass::Corruption);
}
#[test]
fn commit_marker_embeds_marker_bound_journal_batches() {
let marker_id = [0xAB; 16];
let journal_batch = JournalBatch::new(
[0x44; 16],
marker_id,
JournalSequence::new(1),
vec![
JournalRecord::row_put(
"test::Entity",
raw_data_store_key(4),
vec![0x77; 3],
[0x55; 16],
)
.expect("journal row record should build"),
],
)
.expect("journal batch should build");
let marker = CommitMarker::from_parts(marker_id, vec![journal_batch.clone()])
.expect("marker-bound journal batch should build");
let bytes =
encode_commit_marker_payload(&marker).expect("marker payload should encode journal batch");
let decoded = decode_commit_marker_payload(&bytes)
.expect("marker payload should decode embedded journal batch");
assert_eq!(decoded.journal_batches(), &[journal_batch]);
}
#[test]
fn persisted_marker_retains_exact_journal_envelopes_for_live_append() {
let marker_id = [0xAC; 16];
let journal_batch = JournalBatch::new(
[0x45; 16],
marker_id,
JournalSequence::new(1),
vec![
JournalRecord::row_put(
"test::Entity",
raw_data_store_key(5),
vec![0x78; 3],
[0x56; 16],
)
.expect("journal row record should build"),
],
)
.expect("journal batch should build");
let second_batch = JournalBatch::new(
[0x46; 16],
marker_id,
JournalSequence::new(1),
vec![
JournalRecord::row_put(
"test::SecondEntity",
raw_data_store_key(6),
vec![0x79; 3],
[0x57; 16],
)
.expect("second journal row record should build"),
],
)
.expect("second journal batch should build");
let expected_first =
encode_journal_batch(&journal_batch).expect("first journal batch should encode");
let expected_second =
encode_journal_batch(&second_batch).expect("second journal batch should encode");
let marker = CommitMarker::from_parts(marker_id, vec![journal_batch, second_batch])
.expect("marker-bound journal batch should build");
let store = super::CommitStore::init(test_memory(235));
let persisted = store
.set_if_empty(&marker)
.expect("marker should persist once");
assert_eq!(
persisted
.journal_batch_bytes(0)
.expect("persisted marker should retain its journal range"),
expected_first,
);
assert_eq!(
persisted
.journal_batch_bytes(1)
.expect("persisted marker should retain its second journal range"),
expected_second,
);
}
#[test]
fn marker_publication_owns_one_database_sequence_without_consuming_rejections() {
let store = super::CommitStore::init(test_memory(224));
let first_marker_id = [0xB1; 16];
let first = JournalBatch::new_with_database_commit_sequence(
[0xB2; 16],
first_marker_id,
JournalSequence::new(7),
DatabaseCommitSequence::new(1),
Vec::new(),
)
.unwrap();
let second = JournalBatch::new_with_database_commit_sequence(
[0xB3; 16],
first_marker_id,
JournalSequence::new(3),
DatabaseCommitSequence::new(1),
Vec::new(),
)
.unwrap();
let marker = CommitMarker::from_parts(first_marker_id, vec![first, second]).unwrap();
store.set_if_empty(&marker).unwrap();
store.clear_verified().unwrap();
assert_eq!(store.next_database_commit_sequence().unwrap(), 2);
let rejected_marker_id = [0xB4; 16];
let rejected = JournalBatch::new_with_database_commit_sequence(
[0xB5; 16],
rejected_marker_id,
JournalSequence::new(8),
DatabaseCommitSequence::new(3),
Vec::new(),
)
.unwrap();
let rejected_marker = CommitMarker::from_parts(rejected_marker_id, vec![rejected]).unwrap();
assert!(store.set_if_empty(&rejected_marker).is_err());
assert!(store.marker_is_empty().unwrap());
assert_eq!(store.next_database_commit_sequence().unwrap(), 2);
}
#[test]
fn commit_marker_accepts_equal_sequences_from_distinct_journal_tails() {
let marker_id = [0xAB; 16];
let first = JournalBatch::new(
[0x41; 16],
marker_id,
JournalSequence::new(1),
vec![
JournalRecord::row_put(
"test::FirstEntity",
raw_data_store_key(1),
vec![0x11],
[0x31; 16],
)
.expect("first row record should build"),
],
)
.expect("first journal batch should build");
let second = JournalBatch::new(
[0x42; 16],
marker_id,
JournalSequence::new(1),
vec![
JournalRecord::row_put(
"test::SecondEntity",
raw_data_store_key(2),
vec![0x21],
[0x32; 16],
)
.expect("second row record should build"),
],
)
.expect("second journal batch should build");
let marker = CommitMarker::from_parts(marker_id, vec![first.clone(), second.clone()])
.expect("tail-local sequences may coincide inside one marker");
let encoded =
encode_commit_marker_payload(&marker).expect("multi-tail marker payload should encode");
let decoded =
decode_commit_marker_payload(&encoded).expect("multi-tail marker payload should decode");
assert_eq!(decoded.journal_batches(), &[first, second]);
}
#[test]
fn commit_marker_rejects_unbound_journal_batch() {
let marker_id = [0xAB; 16];
let journal_batch = JournalBatch::new(
[0x44; 16],
[0xCD; 16],
JournalSequence::new(1),
vec![
JournalRecord::row_delete("test::Entity", raw_data_store_key(4), [0x55; 16])
.expect("journal row-delete record should build"),
],
)
.expect("journal batch should build");
let err = CommitMarker::from_parts(marker_id, vec![journal_batch])
.expect_err("journal batch must be bound to enclosing marker id");
assert_eq!(err.class, ErrorClass::Corruption);
assert_eq!(err.origin, ErrorOrigin::Store);
}
#[test]
fn commit_marker_future_version_fails_closed() {
let marker = CommitMarker {
id: [6u8; 16],
journal_batches: Vec::new(),
database_control: Vec::new(),
};
let marker_payload = encode_commit_marker_payload(&marker)
.expect("marker payload encode for future-version test should work");
let future_version = COMMIT_MARKER_FORMAT_VERSION_CURRENT.saturating_add(1);
let encoded = encode_commit_marker_bytes(future_version, &marker_payload)
.expect("future-version marker envelope encode should succeed");
let err = RawCommitMarker(encoded)
.try_decode()
.expect_err("future marker versions must fail closed");
assert_eq!(err.class, ErrorClass::IncompatiblePersistedFormat);
assert_eq!(err.origin, ErrorOrigin::Serialize);
}
#[test]
fn commit_marker_rejects_oversized_stored_payload_as_corruption() {
let len = (MAX_COMMIT_BYTES as usize).saturating_add(1);
let err = RawCommitMarker(vec![0; len])
.try_decode()
.expect_err("oversized persisted marker should be rejected");
assert_eq!(err.class, ErrorClass::Corruption);
assert_eq!(err.origin, ErrorOrigin::Store);
}
#[test]
fn clear_verified_rejects_malformed_control_slot() {
let store = super::CommitStore::init(test_memory(232));
let mut malformed = super::CommitStore::encode_raw_control_slot_for_tests(Vec::new())
.expect("current empty commit-control slot should encode");
let marker_length_offset = malformed.len() - size_of::<u32>();
malformed[marker_length_offset..].copy_from_slice(&1_u32.to_le_bytes());
store.set_raw_marker_bytes_for_tests(malformed.clone());
let err = store
.clear_verified()
.expect_err("malformed control slot should fail closed");
assert_eq!(err.class, ErrorClass::Corruption);
assert_eq!(err.origin, ErrorOrigin::Store);
assert_eq!(store.raw_control_slot_bytes_for_tests(), malformed);
}
#[test]
fn commit_slot_writes_and_clears_preserve_database_boot_record() {
let memory = test_memory(231);
let store = super::CommitStore::init(memory.clone());
let mut boot_before = [0_u8; crate::db::database_format::DATABASE_BOOT_RECORD_BYTES];
memory.read(0, &mut boot_before);
let control_slot = super::CommitStore::encode_raw_control_slot_for_tests(vec![0xaa])
.expect("test control slot should encode");
store.set_raw_marker_bytes_for_tests(control_slot);
store.clear_raw_for_tests();
let mut boot_after = [0_u8; crate::db::database_format::DATABASE_BOOT_RECORD_BYTES];
memory.read(0, &mut boot_after);
assert_eq!(boot_after, boot_before);
}
#[test]
fn commit_marker_transitions_preserve_database_incarnation() {
let store = super::CommitStore::init(test_memory(227));
let incarnation_before = store
.database_incarnation_id()
.expect("current control slot should carry an incarnation");
let cursor_key_before = store
.cursor_authentication_key()
.expect("current control slot should carry a cursor key");
let proof_before = store
.proof_identity()
.expect("empty current control slot should fingerprint");
let marker = CommitMarker {
id: [0xA7; 16],
journal_batches: Vec::new(),
database_control: Vec::new(),
};
store
.set_if_empty(&marker)
.expect("marker publication should preserve control metadata");
assert_eq!(
store
.database_incarnation_id()
.expect("marker-bearing control slot should carry an incarnation"),
incarnation_before,
);
assert_eq!(
store
.cursor_authentication_key()
.expect("marker-bearing control slot should carry a cursor key"),
cursor_key_before,
);
assert_ne!(
store
.proof_identity()
.expect("marker-bearing current control slot should fingerprint"),
proof_before,
"marker publication must invalidate the database-control proof",
);
store
.clear_verified()
.expect("marker clear should preserve control metadata");
assert_eq!(
store
.database_incarnation_id()
.expect("cleared control slot should carry an incarnation"),
incarnation_before,
);
assert_eq!(
store
.cursor_authentication_key()
.expect("cleared control slot should carry a cursor key"),
cursor_key_before,
);
assert_ne!(
store
.proof_identity()
.expect("cleared current control slot should fingerprint"),
proof_before,
"advancing the database sequence must change the empty control proof",
);
assert_eq!(
store
.next_database_commit_sequence()
.expect("cleared control should preview its successor"),
2,
);
}
#[test]
fn ordinary_reopen_preserves_database_incarnation() {
let memory = test_memory(225);
let first = super::CommitStore::init(memory.clone());
let incarnation = first
.database_incarnation_id()
.expect("initial current control slot should carry an incarnation");
let cursor_key = first
.cursor_authentication_key()
.expect("initial current control slot should carry a cursor key");
let reopened = super::CommitStore::open(memory)
.expect("ordinary reopen should admit the same current control state");
assert_eq!(
reopened
.database_incarnation_id()
.expect("reopened current control slot should carry an incarnation"),
incarnation,
);
assert_eq!(
reopened
.cursor_authentication_key()
.expect("reopened current control slot should carry a cursor key"),
cursor_key,
);
}
#[test]
fn database_control_rejects_zero_incarnation() {
let store = super::CommitStore::init(test_memory(226));
let mut control_slot = super::CommitStore::encode_raw_control_slot_for_tests(Vec::new())
.expect("current empty commit-control slot should encode");
control_slot[5..21].fill(0);
store.set_raw_marker_bytes_for_tests(control_slot);
let err = store
.database_incarnation_id()
.expect_err("zero database incarnation must fail closed");
assert_eq!(err.class, ErrorClass::Corruption);
assert_eq!(err.origin, ErrorOrigin::Store);
}
#[test]
fn database_control_rejects_zero_cursor_authentication_key() {
let store = super::CommitStore::init(test_memory(234));
let mut control_slot = super::CommitStore::encode_raw_control_slot_for_tests(Vec::new())
.expect("current empty commit-control slot should encode");
control_slot[21..53].fill(0);
store.set_raw_marker_bytes_for_tests(control_slot);
let err = store
.cursor_authentication_key()
.expect_err("zero cursor authentication key must fail closed");
assert_eq!(err.class, ErrorClass::Corruption);
assert_eq!(err.origin, ErrorOrigin::Store);
}
#[test]
fn database_control_frame_checksum_corruption_fails_closed() {
let memory = test_memory(230);
let store = super::CommitStore::init(memory.clone());
let checksum_offset = super::DATABASE_CONTROL_SLOT_FRAME_OFFSET
+ super::DATABASE_CONTROL_SLOT_FRAME_CHECKSUM_OFFSET as u64;
let mut checksum_byte = [0_u8; 1];
memory.read(checksum_offset, &mut checksum_byte);
checksum_byte[0] ^= 0xff;
memory.write(checksum_offset, &checksum_byte);
let err = store
.marker_is_empty()
.expect_err("corrupt database-control checksum should fail closed");
assert_eq!(err.class, ErrorClass::Corruption);
assert_eq!(err.origin, ErrorOrigin::Store);
}
#[test]
fn database_control_frame_future_version_fails_closed() {
let memory = test_memory(229);
let store = super::CommitStore::init(memory.clone());
let version_offset = super::DATABASE_CONTROL_SLOT_FRAME_OFFSET
+ super::DATABASE_CONTROL_SLOT_FRAME_MAGIC.len() as u64;
memory.write(
version_offset,
&[super::DATABASE_CONTROL_SLOT_FRAME_VERSION.saturating_add(1)],
);
let err = store
.marker_is_empty()
.expect_err("future database-control frame must fail closed");
assert_eq!(err.class, ErrorClass::IncompatiblePersistedFormat);
assert_eq!(err.origin, ErrorOrigin::Serialize);
}
#[test]
fn commit_control_slot_future_version_fails_closed() {
let store = super::CommitStore::init(test_memory(228));
let mut control_slot = super::CommitStore::encode_raw_control_slot_for_tests(Vec::new())
.expect("current empty commit-control slot should encode");
control_slot[4] = control_slot[4].saturating_add(1);
store.set_raw_marker_bytes_for_tests(control_slot);
let err = store
.marker_is_empty()
.expect_err("future commit-control slot must fail closed");
assert_eq!(err.class, ErrorClass::IncompatiblePersistedFormat);
assert_eq!(err.origin, ErrorOrigin::Serialize);
}