פתרון יומן רשומות עם התאוששות

הפתרון הבא הוא קובץ מלא בפרויקט Cargo עם libc. הוא משתמש בנעילת OFD בין משתתפים שמשתפים פעולה, מגביל את היומן ל־16 מגה־בייט לצורך התרגיל ומקבל payload של עד 64 קילובייט לרשומה. הרחבת יומן באמצעות segments ו־compaction תהיה שינוי פרוטוקול נפרד.

use std::{env, fs::{File, OpenOptions}, io::{self, Read, Seek, SeekFrom, Write}, path::Path};
use std::os::fd::AsRawFd;
const MAGIC: &[u8; 4] = b"ATL1";
const MAX_RECORD: usize = 64 * 1024;
const MAX_LOG: usize = 16 * 1024 * 1024;
fn invalid(text: &'static str) -> io::Error { io::Error::new(io::ErrorKind::InvalidData, text) }
fn crc32(bytes: &[u8]) -> u32 {
    let mut crc = !0u32;
    for &byte in bytes {
        crc ^= u32::from(byte);
        for _ in 0..8 {
            crc = (crc >> 1) ^ (0xedb88320u32 & 0u32.wrapping_sub(crc & 1));
        }
    }
    !crc
}
fn encode(sequence: u64, payload: &[u8]) -> io::Result<Vec<u8>> {
    if payload.len() > MAX_RECORD { return Err(invalid("record too large")); }
    let mut bytes = Vec::with_capacity(24 + payload.len());
    bytes.extend_from_slice(MAGIC);
    bytes.extend_from_slice(&sequence.to_le_bytes());
    bytes.extend_from_slice(&(payload.len() as u32).to_le_bytes());
    bytes.extend_from_slice(&crc32(&bytes).to_le_bytes());
    bytes.extend_from_slice(payload);
    bytes.extend_from_slice(&crc32(&bytes).to_le_bytes());
    Ok(bytes)
}
struct Recovery { records: Vec<Vec<u8>>, valid_end: usize, next: u64, partial_tail: bool }
fn scan(bytes: &[u8]) -> io::Result<Recovery> {
    if bytes.len() > MAX_LOG { return Err(invalid("log too large")); }
    let mut result = Recovery { records: Vec::new(), valid_end: 0, next: 0, partial_tail: false };
    while result.valid_end < bytes.len() {
        let rest = &bytes[result.valid_end..];
        if rest.len() < 20 { result.partial_tail = true; break; }
        if &rest[..4] != MAGIC { return Err(invalid("bad format magic")); }
        let header_crc = u32::from_le_bytes(rest[16..20].try_into().unwrap());
        if crc32(&rest[..16]) != header_crc { return Err(invalid("header checksum mismatch")); }
        let sequence = u64::from_le_bytes(rest[4..12].try_into().unwrap());
        let length = u32::from_le_bytes(rest[12..16].try_into().unwrap()) as usize;
        if sequence != result.next { return Err(invalid("sequence mismatch")); }
        if length > MAX_RECORD { return Err(invalid("record length exceeds limit")); }
        let end = 24 + length;
        if rest.len() < end { result.partial_tail = true; break; }
        let expected = u32::from_le_bytes(rest[20 + length..end].try_into().unwrap());
        if crc32(&rest[..20 + length]) != expected { return Err(invalid("checksum mismatch")); }
        result.records.push(rest[20..20 + length].to_vec());
        result.valid_end += end;
        result.next = result.next.checked_add(1).ok_or_else(|| invalid("sequence exhausted"))?;
    }
    Ok(result)
}
fn exclusive(file: &File) -> io::Result<()> {
    let mut lock: libc::flock = unsafe { std::mem::zeroed() };
    lock.l_type = libc::F_WRLCK as _;
    lock.l_whence = libc::SEEK_SET as _;
    loop {
        if unsafe { libc::fcntl(file.as_raw_fd(), libc::F_OFD_SETLKW, &lock) } != -1 {
            return Ok(());
        }
        let error = io::Error::last_os_error();
        if error.kind() != io::ErrorKind::Interrupted { return Err(error); }
    }
}
struct Journal { file: File, next: u64, length: usize, poisoned: bool }
impl Journal {
    fn open(path: &Path) -> io::Result<(Self, Recovery)> {
        let mut file = OpenOptions::new().read(true).write(true).open(path)?;
        exclusive(&file)?;
        let mut bytes = Vec::new();
        (&mut file).take((MAX_LOG + 1) as u64).read_to_end(&mut bytes)?;
        let recovery = scan(&bytes)?;
        if recovery.partial_tail {
            file.set_len(recovery.valid_end as u64)?;
        }
        file.sync_all()?;
        file.seek(SeekFrom::Start(recovery.valid_end as u64))?;
        let journal = Self { file, next: recovery.next, length: recovery.valid_end, poisoned: false };
        Ok((journal, recovery))
    }
    fn append(&mut self, payload: &[u8]) -> io::Result<u64> {
        if self.poisoned { return Err(invalid("journal requires recovery")); }
        let next = self.next.checked_add(1).ok_or_else(|| invalid("sequence exhausted"))?;
        let bytes = encode(self.next, payload)?;
        let new_length = self.length.checked_add(bytes.len()).ok_or_else(|| invalid("length overflow"))?;
        if new_length > MAX_LOG { return Err(invalid("log budget exceeded")); }
        self.poisoned = true;
        self.file.write_all(&bytes)?;
        self.file.sync_all()?;
        let committed = self.next;
        self.next = next;
        self.length = new_length;
        self.poisoned = false;
        Ok(committed)
    }
}
fn initialize(path: &Path) -> io::Result<()> {
    let file = OpenOptions::new().read(true).write(true).create_new(true).open(path)?;
    file.sync_all()?;
    let parent = path.parent().filter(|p| !p.as_os_str().is_empty()).unwrap_or(Path::new("."));
    File::open(parent)?.sync_all()
}
fn main() -> io::Result<()> {
    let mut args = env::args_os().skip(1);
    let command = args.next().and_then(|x| x.into_string().ok()).ok_or_else(|| invalid("command"))?;
    let path = args.next().ok_or_else(|| invalid("path"))?;
    let path = Path::new(&path);
    match command.as_str() {
        "init" => initialize(path),
        "append" => {
            let payload = args.next().and_then(|x| x.into_string().ok()).ok_or_else(|| invalid("UTF-8 payload"))?;
            let (mut journal, recovery) = Journal::open(path)?;
            if recovery.partial_tail { eprintln!("discarded incomplete tail"); }
            let sequence = journal.append(payload.as_bytes())?;
            println!("durable sequence={sequence}");
            Ok(())
        }
        "dump" => {
            let (_journal, recovery) = Journal::open(path)?;
            for (sequence, record) in recovery.records.iter().enumerate() {
                println!("{sequence}: {record:?}");
            }
            Ok(())
        }
        _ => Err(invalid("use init, append or dump")),
    }
}
#[cfg(test)]
mod tests {
    use super::*;
    #[test]
    fn every_truncated_prefix() {
        let first = encode(0, b"first").unwrap();
        let second = encode(1, b"second").unwrap();
        let mut bytes = first.clone();
        bytes.extend(second);
        for cut in 0..=bytes.len() {
            let recovered = scan(&bytes[..cut]).unwrap();
            let expected = if cut == bytes.len() { 2 } else if cut >= first.len() { 1 } else { 0 };
            assert_eq!(recovered.records.len(), expected, "cut={cut}");
        }
    }
    #[test]
    fn corruption_is_not_silently_truncated() {
        let bytes = encode(0, b"payload").unwrap();
        for index in 0..bytes.len() {
            let mut damaged = bytes.clone();
            damaged[index] ^= 1;
            assert!(scan(&damaged).is_err(), "damaged byte={index}");
        }
        let mut damaged = bytes;
        damaged[20] ^= 1;
        assert!(scan(&damaged).is_err());
    }
    #[test]
    fn standard_crc_check_vector() { assert_eq!(crc32(b"123456789"), 0xcbf43926); }
}

הפורמט משתמש ב־little endian מפורש, ולכן אינו תלוי בפריסת struct של ראסט. הכותרת כוללת CRC נפרד שנבדק לפני שימוש באורך, והרשומה מסתיימת ב־CRC שמכסה את הכותרת ואת ה־payload. כך שינוי ביט באורך אינו מתחפש לזנב חלקי רק מפני שהאורך הפגום גדול יותר. הבדיקה עדיין אינה הוכחה מול שינוי זדוני או כל צירוף שיבושים אפשרי, ואינה מנגנון תיקון כללי של אחסון פגום.

הפתיחה מבצעת התאוששות ומשנה את הקובץ רק כאשר נמצא זנב חלקי לפי הפורמט. גם פקודת dump בדוגמה משתמשת בפתיחה הזאת, ולכן היא כלי התאוששות ואינה קורא פורנזי שאינו משנה דבר. עבור כלי אבחון נפרד נשתמש ב־scan על עותק ונשמור את המקור לפני כל חיתוך.

הפתיחה מסנכרנת את הקידומת שהתקבלה גם אם לא נדרש חיתוך. רשומה שלמה עם בדיקת תקינות מוצלחת יכולה להגיע מניסיון קודם שכתב את כל הבתים אך נכשל בסנכרון, והבתים עדיין עשויים להימצא במטמון אחרי אתחול התהליך. לכן קריאה מוצלחת אינה לבדה ראיה לעמידות. רק אחרי sync_all מוצלח נפרסם את הרשומות המשוחזרות לשכבה שמורשית לאשר תוצאה ללקוח; אם הסנכרון נכשל, הפתיחה כולה נכשלת.

המצב poisoned נקבע לפני תחילת הכתיבה ונמחק רק אחרי סנכרון מוצלח. כך כשל מאוחר אינו מאפשר לאותו אובייקט להוסיף רשומה חדשה מעל מצב שלא אומת. פתיחה מחדש בודקת מה נמצא בפועל ביומן, ויכולה לגלות גם רשומה שלמה שהלקוח לא קיבל עליה אישור.

הנעילה היא advisory ומגינה רק מול משתתפים שמכבדים את אותו פרוטוקול. היא מוחזקת באמצעות פתיחת הקובץ של Journal עד השמדתה. כותב זר או תהליך שמחליף את הקובץ ב־rename יכולים להפר הנחות נוספות, ולכן תיקיית היומן וההרשאות שלה הן חלק מסביבת המוצר.

מבחני החיתוך בודקים את parser ואת כללי הקבלה של זנב. הם אינם מוכיחים סדר כתיבות של filesystem או עמידות של ההתקן. את סדר write ו־fsync נבדוק בנפרד, ואת מודל הכשל נבדוק במעבדת אחסון. אישור שמודפס אחרי sync_all נשען על החוזה של כל השכבות שמתחתיו.

הפתרון עדיין אינו מסד נתונים מלא: הוא אינו מבצע compaction, אינו מנהל transactions מרובות רשומות ואינו מספק deduplication של בקשות לקוח. כל אחת מההרחבות האלה צריכה להגדיר נקודת commit ומבחני כשל נוספים, במקום להסתמך על כך שכבר קיים קובץ שנקרא journal.