use super::*;
use kcode_k1_transaction::{GENESIS_PARENT, SubsystemId, build_signed_transaction};
use std::fs::{self, File, OpenOptions};
use std::io::Write;
use std::thread;
use tempfile::TempDir;
const SUBSYSTEM: &str = "audio-classification";
const PAYLOAD: &[u8] = &[1, 2, 255];
fn id(value: u8) -> TxId {
TxId::from_bytes([value; 12])
}
fn commit(ordering: &K1TxnOrdering, parent: &mut TxId, payload: &[u8]) -> TxId {
let bytes = build_signed_transaction(
*parent,
1,
[1; 32],
SubsystemId::from_str(SUBSYSTEM).unwrap(),
payload,
|_| Ok([2; 64]),
)
.unwrap();
let callback = TxId::for_transaction(&bytes);
ordering.submit_txn(&bytes).unwrap();
*parent = callback;
callback
}
fn log(root: &Path, fragment: TxId) -> PathBuf {
root.join(LOGS).join(fragment_filename(fragment))
}
fn seed(base: &Path, name: &str, ordering: &K1TxnOrdering, callback: TxId) -> PathBuf {
let root = base.join(name);
let mut cache = ProjectionCache::open(&root, ordering).unwrap().0;
cache.append(callback, id(1), PAYLOAD).unwrap();
root
}
fn assert_reset(root: &Path, ordering: &K1TxnOrdering) {
let (_, records, cursor) = ProjectionCache::open(root, ordering).unwrap();
assert!(records.is_empty());
assert_eq!(cursor, None);
assert_eq!(fs::read_dir(root.join(LOGS)).unwrap().count(), 0);
}
fn format_record(sequence: u64, callback: TxId, payload: &[u8]) -> CacheRecord {
CacheRecord {
sequence,
callback_txid: callback,
fragment_id: id(1),
payload: payload.to_vec(),
}
}
fn append_record(root: &Path, record: CacheRecord) {
let bytes = encode_record(&record).unwrap();
OpenOptions::new()
.append(true)
.open(log(root, id(1)))
.unwrap()
.write_all(&bytes)
.unwrap();
}
fn rewrite_cursor(root: &Path, sequence: u64, callback: TxId) {
let bytes = encode_cursor(CacheCursor {
last_sequence: sequence,
last_callback_txid: callback,
})
.unwrap();
fs::write(root.join(CURSOR), bytes).unwrap();
}
fn corrupt(
base: &Path,
name: &str,
ordering: &K1TxnOrdering,
callback: TxId,
edit: impl FnOnce(&Path),
) {
let root = seed(base, name, ordering, callback);
edit(&root);
assert_reset(&root, ordering);
}
#[test]
fn layout_round_trip_sqlite_deletion_and_clear() {
let temp = TempDir::new().unwrap();
let ordering = K1TxnOrdering::open(&temp.path().join("ordering")).unwrap();
let root = temp.path().join("cache");
fs::create_dir(&root).unwrap();
for suffix in ["", "-wal", "-shm"] {
fs::write(root.join(format!("{SQLITE}{suffix}")), b"never read").unwrap();
}
let (mut cache, records, cursor) = ProjectionCache::open(&root, &ordering).unwrap();
assert!(records.is_empty() && cursor.is_none());
for suffix in ["", "-wal", "-shm"] {
assert!(!root.join(format!("{SQLITE}{suffix}")).exists());
}
let mut parent = GENESIS_PARENT;
let first = commit(&ordering, &mut parent, PAYLOAD);
let second = commit(&ordering, &mut parent, b"\x07\x08");
cache.append(first, id(2), PAYLOAD).unwrap();
cache.append(second, id(1), b"\x07\x08").unwrap();
assert_eq!(fs::read_dir(root.join(LOGS)).unwrap().count(), 2);
let disk_cursor = decode_cursor(&fs::read(root.join(CURSOR)).unwrap()).unwrap();
assert_eq!(disk_cursor.last_sequence, 2);
assert_eq!(disk_cursor.last_callback_txid, second);
drop(cache);
let (mut cache, records, cursor) = ProjectionCache::open(&root, &ordering).unwrap();
assert_eq!(records.len(), 2);
assert_eq!(records[0].callback_txid, first);
assert_eq!(records[0].fragment_id, id(2));
assert_eq!(records[0].payload.as_slice(), PAYLOAD);
assert_eq!(records[1].callback_txid, second);
assert_eq!(records[1].fragment_id, id(1));
assert_eq!(records[1].payload.as_slice(), b"\x07\x08");
assert_eq!(cursor, Some(second));
cache.clear().unwrap();
cache.ensure_healthy().unwrap();
assert!(!root.join(CURSOR).exists());
assert!(!root.join(TEMP).exists());
assert_reset(&root, &ordering);
}
#[test]
fn representative_corruption_and_absent_callback_reset() {
let temp = TempDir::new().unwrap();
let ordering = K1TxnOrdering::open(&temp.path().join("ordering")).unwrap();
let mut parent = GENESIS_PARENT;
let callback = commit(&ordering, &mut parent, PAYLOAD);
let second = commit(&ordering, &mut parent, b"second");
let canonical = encode_record(&format_record(1, callback, PAYLOAD)).unwrap();
let mut noncanonical = canonical.clone();
noncanonical.insert(1, b' ');
for (name, bytes) in [
("malformed", b"{bad}\n".to_vec()),
("noncanonical", noncanonical),
("partial", canonical[..canonical.len() - 1].to_vec()),
] {
corrupt(temp.path(), name, &ordering, callback, move |root| {
fs::write(log(root, id(1)), bytes).unwrap();
});
}
let root = seed(temp.path(), "gap", &ordering, callback);
let mut record = decode_record(&fs::read(log(&root, id(1))).unwrap()).unwrap();
record.sequence = 2;
fs::write(log(&root, id(1)), encode_record(&record).unwrap()).unwrap();
assert_reset(&root, &ordering);
let root = seed(temp.path(), "duplicate", &ordering, callback);
append_record(&root, format_record(2, callback, PAYLOAD));
rewrite_cursor(&root, 2, callback);
assert_reset(&root, &ordering);
let root = seed(temp.path(), "extra", &ordering, callback);
append_record(&root, format_record(2, second, b"second"));
assert_reset(&root, &ordering);
let root = seed(temp.path(), "cursor", &ordering, callback);
rewrite_cursor(&root, 1, second);
assert_reset(&root, &ordering);
corrupt(temp.path(), "temp", &ordering, callback, |root| {
fs::write(root.join(TEMP), b"stale").unwrap();
});
corrupt(temp.path(), "filename", &ordering, callback, |root| {
fs::rename(log(root, id(1)), log(root, id(2))).unwrap();
});
corrupt(temp.path(), "entry", &ordering, callback, |root| {
fs::write(root.join(LOGS).join("unknown"), b"entry").unwrap();
});
let root = seed(temp.path(), "absent-kto", &ordering, id(90));
assert_reset(&root, &ordering);
}
#[test]
fn poison_and_independent_root_concurrency() {
let temp = TempDir::new().unwrap();
let ordering = K1TxnOrdering::open(&temp.path().join("ordering")).unwrap();
let root = temp.path().join("poison");
let mut cache = ProjectionCache::open(&root, &ordering).unwrap().0;
fs::remove_dir(root.join(LOGS)).unwrap();
File::create(root.join(LOGS)).unwrap();
assert!(cache.append(id(1), id(2), b"payload").is_err());
assert!(cache.ensure_healthy().is_err());
assert!(cache.append(id(1), id(2), b"payload").is_err());
assert!(cache.clear().is_err());
drop(cache);
assert_reset(&root, &ordering);
let mut first = ProjectionCache::open(&temp.path().join("first"), &ordering)
.unwrap()
.0;
let mut second = ProjectionCache::open(&temp.path().join("second"), &ordering)
.unwrap()
.0;
let one = thread::spawn(move || first.append(id(10), id(1), b"one"));
let two = thread::spawn(move || second.append(id(20), id(2), b"two"));
one.join().unwrap().unwrap();
two.join().unwrap().unwrap();
}