Persistence with a snapshot

Step 6 of 10. Step 5 ended at the tag mini-ch05 with a deadline on every key: EXPIRE and TTL, lazy expiry on the read path, and a sweeper reclaiming the rest once a second. This step starts from that tag and ends at mini-ch06. A SAVE command writes the whole store to disk, the next boot reads the file back before it opens its port, and deadlines survive the trip — a key with 100 seconds left keeps shrinking through a restart instead of coming back fresh. The server can die now, and the keys do not.

The deadline that cannot go to disk

Step 5’s data model said it outright: the deadline is an Instant, and an Instant is process-relative. It counts from an arbitrary point chosen when the machine booted, so its raw value means nothing to the next process — write one to disk and the file holds a number no reader can place on any clock. Storing “seconds left” instead does not survive contact with the real server either: 100 seconds of remaining TTL written at save time would come back as 100, and the real server comes back with less.

What the real server actually does, captured on redis-server 7.2.4 before any of the Rust below was written:

Event	Real server
`SAVE`	`+OK\r\n`, and the reply waits for the dump to finish
`BGSAVE`	`+Background saving started\r\n`, and the reply returns at once
`TTL` 100, `SAVE`, 3 seconds of downtime, restart	`TTL` answers 97
restart with a saved database	`Done loading RDB, keys loaded: 1, keys expired: 0.`
corrupt dump file at startup	`# Short read or OOM loading DB. Unrecoverable error, aborting now.` — the process exits

The 97 is the whole design in one line. The file carries a moment on the shared timeline, not a remaining duration: the real server’s RDB stores deadlines as milliseconds since the Unix epoch, and downtime comes out of every TTL because the shared clock did not stop.

The clone copies that. The live Entry keeps its monotonic Instant — step 5’s reason, that an adjusted system clock cannot move a key’s expiry, still holds for everything the running server does. The conversion happens only at the edge:

on the way out, the remaining time is measured on the monotonic clock and pinned to the wall clock: now_ms + (deadline - now), in milliseconds since the epoch;
on the way in, the reverse: Instant::now() + (stored_ms - now_ms).

Both directions take the same (now, now_ms) pair, one from each clock, captured at the same instant — that pairing is what keeps the two scales consistent.

The file format

Length-prefixed throughout, like the RESP frames of step 2:

MRDB          4 bytes  magic
01            1 byte   format version
xx xx xx xx   4 bytes  key count, u32 little-endian

then, per key:
xx xx xx xx   4 bytes  key length, u32 LE
....          N bytes  key
00            1 byte   value kind: 0 = string, 1 = list
  kind 0:     4 bytes length, then the bytes
  kind 1:     4 bytes item count, then per item 4 bytes length + bytes
00|01         1 byte   deadline flag: 0 = none, 1 = deadline follows
xx × 8        8 bytes  deadline: milliseconds since the Unix epoch, u64 LE

The magic and the version byte exist for the file’s reader, not its writer. The writer is this binary and knows the format; the reader might be a future one — or the file might never have been a snapshot at all. Both are refused before a single record is parsed.

snapshot and restore

The whole round trip, run over three keys — one string with no deadline, one list, one string with 100 seconds left — and then over a file with one byte missing:

use std::collections::{HashMap, VecDeque};
use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};

/// What a key holds: a string, or a list of strings.
#[derive(PartialEq, Debug)]
enum Value {
    String(Vec<u8>),
    List(VecDeque<Vec<u8>>),
}

/// One key in the store: the value, and the instant it stops being
/// there.
///
/// The deadline is an `Instant`, not a wall-clock time. It is monotonic,
/// so adjusting the system clock cannot move a key's expiry, and it is
/// process-relative: a snapshot cannot carry it, so it goes to disk as
/// wall-clock milliseconds and comes back re-anchored to the new boot's
/// clocks.
#[derive(PartialEq, Debug)]
struct Entry {
    value: Value,
    /// `None` is a key with no deadline: it lives until it is
    /// overwritten or deleted, which is what `SET` leaves behind.
    expires_at: Option<Instant>,
}

/// Whether `entry`'s deadline has arrived. A deadline of exactly `now`
/// counts as arrived, the way the real server treats a key whose
/// millisecond timestamp has passed.
fn expired(entry: &Entry, now: Instant) -> bool {
    matches!(entry.expires_at, Some(at) if now >= at)
}

/// The seconds left on `entry`, the way `TTL` reports them: the
/// remaining time rounded to the nearest second. `None` is a key with
/// no deadline at all.
fn ttl_of(entry: &Entry, now: Instant) -> Option<i64> {
    let at = entry.expires_at?;
    let left = at.saturating_duration_since(now).as_millis() as i64;
    Some((left + 500) / 1000)
}

/// Milliseconds since the Unix epoch, right now. This is the wall-clock
/// side of the deadline conversion: an `Instant` cannot go to disk, but
/// a moment on the shared timeline can.
fn unix_ms() -> u64 {
    SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .expect("the clock is past 1970")
        .as_millis() as u64
}

fn put_u32(out: &mut Vec<u8>, n: u32) {
    out.extend_from_slice(&n.to_le_bytes());
}

fn put_u64(out: &mut Vec<u8>, n: u64) {
    out.extend_from_slice(&n.to_le_bytes());
}

fn end_of_snapshot() -> String {
    "unexpected end of snapshot".to_string()
}

fn get_byte(bytes: &[u8], pos: &mut usize) -> Result<u8, String> {
    let byte = bytes
        .get(*pos)
        .copied()
        .ok_or_else(end_of_snapshot)?;
    *pos += 1;
    Ok(byte)
}

fn get_u32(bytes: &[u8], pos: &mut usize) -> Result<u32, String> {
    let end = (*pos).checked_add(4).ok_or_else(end_of_snapshot)?;
    let slice = bytes.get(*pos..end).ok_or_else(end_of_snapshot)?;
    *pos = end;
    Ok(u32::from_le_bytes(slice.try_into().unwrap()))
}

fn get_u64(bytes: &[u8], pos: &mut usize) -> Result<u64, String> {
    let end = (*pos).checked_add(8).ok_or_else(end_of_snapshot)?;
    let slice = bytes.get(*pos..end).ok_or_else(end_of_snapshot)?;
    *pos = end;
    Ok(u64::from_le_bytes(slice.try_into().unwrap()))
}

fn take_bytes<'a>(bytes: &'a [u8], pos: &mut usize, len: usize) -> Result<&'a [u8], String> {
    let end = (*pos).checked_add(len).ok_or_else(end_of_snapshot)?;
    let slice = bytes.get(*pos..end).ok_or_else(end_of_snapshot)?;
    *pos = end;
    Ok(slice)
}

/// The store as bytes: a five-byte header, then one length-prefixed
/// record per key. Returns the buffer and how many keys went into it.
///
/// Deadlines go out as wall-clock milliseconds — the shared-timeline
/// moment the key stops existing, not the seconds left. The remaining
/// time is measured on the monotonic clock, where it cannot be moved by
/// an adjusted system clock, and then pinned to the wall clock with the
/// same `now` used for the check.
///
/// Keys whose deadline has passed are not written: the read path would
/// remove them on sight, so carrying them would only defer the removal
/// to load time.
fn snapshot(
    store: &HashMap<Vec<u8>, Entry>,
    now: Instant,
    now_ms: u64,
) -> (Vec<u8>, usize) {
    let mut out = Vec::new();
    out.extend_from_slice(b"MRDB");
    out.push(1);
    // The count is patched in at the end, when it is known.
    put_u32(&mut out, 0);
    let mut count = 0usize;
    for (key, entry) in store {
        // Expired keys are dead on arrival: the read path would remove
        // them on sight, so the snapshot does not carry them. The skip
        // also guarantees the `at - now` below cannot panic.
        if expired(entry, now) {
            continue;
        }
        put_u32(&mut out, key.len() as u32);
        out.extend_from_slice(key);
        match &entry.value {
            Value::String(value) => {
                out.push(0);
                put_u32(&mut out, value.len() as u32);
                out.extend_from_slice(value);
            }
            Value::List(list) => {
                out.push(1);
                put_u32(&mut out, list.len() as u32);
                for value in list {
                    put_u32(&mut out, value.len() as u32);
                    out.extend_from_slice(value);
                }
            }
        }
        match entry.expires_at {
            None => out.push(0),
            Some(at) => {
                out.push(1);
                let remaining = (at - now).as_millis() as u64;
                put_u64(&mut out, now_ms + remaining);
            }
        }
        count += 1;
    }
    out[5..9].copy_from_slice(&(count as u32).to_le_bytes());
    (out, count)
}

/// The store from bytes, re-anchored: each wall-clock deadline becomes
/// an `Instant` on the new boot's monotonic clock,
/// `now + deadline - now_ms`. A key whose deadline fell while the
/// server was down is dropped here — the real server skips expired keys
/// at load too, counting them in its `keys expired` log line.
/// Captured on redis-server 7.2.4.
fn restore(
    bytes: &[u8],
    now: Instant,
    now_ms: u64,
) -> Result<HashMap<Vec<u8>, Entry>, String> {
    if bytes.len() < 9 || &bytes[..4] != b"MRDB" {
        return Err("not a mini-redis snapshot".to_string());
    }
    if bytes[4] != 1 {
        return Err(format!("unsupported snapshot version {}", bytes[4]));
    }
    let count = u32::from_le_bytes(bytes[5..9].try_into().unwrap()) as usize;
    let mut pos = 9;
    let mut store = HashMap::with_capacity(count);
    for _ in 0..count {
        let key_len = get_u32(bytes, &mut pos)? as usize;
        let key = take_bytes(bytes, &mut pos, key_len)?.to_vec();
        let value = match get_byte(bytes, &mut pos)? {
            0 => {
                let len = get_u32(bytes, &mut pos)? as usize;
                Value::String(take_bytes(bytes, &mut pos, len)?.to_vec())
            }
            1 => {
                let items = get_u32(bytes, &mut pos)? as usize;
                let mut list = VecDeque::with_capacity(items);
                for _ in 0..items {
                    let len = get_u32(bytes, &mut pos)? as usize;
                    list.push_back(take_bytes(bytes, &mut pos, len)?.to_vec());
                }
                Value::List(list)
            }
            other => return Err(format!("unknown value kind {other}")),
        };
        let expires_at = match get_byte(bytes, &mut pos)? {
            0 => None,
            1 => {
                let at_ms = get_u64(bytes, &mut pos)?;
                if at_ms <= now_ms {
                    // The deadline fell while the server was down.
                    continue;
                }
                Some(now + Duration::from_millis(at_ms - now_ms))
            }
            other => return Err(format!("unknown deadline flag {other}")),
        };
        store.insert(key, Entry { value, expires_at });
    }
    Ok(store)
}

fn main() {
    let now = Instant::now();
    let now_ms = unix_ms();
    let mut store = HashMap::new();
    store.insert(
        b"name".to_vec(),
        Entry {
            value: Value::String(b"mini".to_vec()),
            expires_at: None,
        },
    );
    let mut list = VecDeque::new();
    list.push_back(b"one".to_vec());
    list.push_back(b"two".to_vec());
    store.insert(
        b"queue".to_vec(),
        Entry {
            value: Value::List(list),
            expires_at: None,
        },
    );
    store.insert(
        b"session".to_vec(),
        Entry {
            value: Value::String(b"token".to_vec()),
            expires_at: Some(now + Duration::from_secs(100)),
        },
    );
    let (bytes, count) = snapshot(&store, now, now_ms);
    println!("{count} key(s), {} bytes", bytes.len());

    // The process dies here. Forty seconds later, a new one reads the
    // same bytes with its own clocks: `later` is forty seconds on from
    // `now`, and the wall clock moved by the same amount.
    let later = now + Duration::from_secs(40);
    let back = restore(&bytes, later, now_ms + 40_000).expect("restores");
    println!("loaded back after 40 seconds of downtime:");
    let mut keys: Vec<&Vec<u8>> = back.keys().collect();
    keys.sort();
    for key in keys {
        let entry = &back[key];
        let ttl = ttl_of(entry, later).unwrap_or(-1);
        println!("  {:<8} {}", String::from_utf8_lossy(key), ttl);
    }

    // One byte missing: the file stops one byte short of the end.
    let truncated = &bytes[..bytes.len() - 1];
    if let Err(err) = restore(truncated, later, now_ms + 40_000) {
        println!("truncated: {err}");
    }
}

Run this in the Rust Playground

3 key(s), 86 bytes
loaded back after 40 seconds of downtime:
  name     -1
  queue    -1
  session  60
truncated: unexpected end of snapshot

Three things worth pausing on. The session row is the point of the whole step: 100 seconds saved, 40 seconds of downtime, 60 left — the same answer the real server gave above, scaled. The truncation is caught by the bounds checks in get_byte and friends, not by trusting the header’s count: every read either fits in the buffer or ends the load with an error. And the name and queue rows answer -1, not 0, because a snapshot with no deadline is a key with no deadline — the Option from step 5 survives the file.

SAVE in the command layer

One new variant, at the end of the enum:

enum Command {
    Ping,
    Echo(Vec<u8>),
    Set(Vec<u8>, Vec<u8>),
    Get(Vec<u8>),
    Del(Vec<Vec<u8>>),
    LPush(Vec<u8>, Vec<Vec<u8>>),
    LRange(Vec<u8>, i64, i64),
    Expire(Vec<u8>, i64),
    Ttl(Vec<u8>),
    Save,
}

And because SAVE takes nothing, its arm can only match the empty argument list:

            // No arguments, ever: the real server takes none and dumps
            // the whole keyspace. Captured on redis-server 7.2.4 —
            // `SAVE` with an argument answers the wrong-arity error.
            b"SAVE" => match args {
                [] => Ok(Command::Save),
                _ => Err(wrong_args(name)),
            },

The command arm. SAVE is synchronous on purpose, like the real server’s: the dump happens inside the command and every other client waits for it:

        Command::Save => {
            // Synchronous on purpose, like the real server's SAVE: the
            // dump happens inside the command, and every other client
            // waits for it. Captured on redis-server 7.2.4 — `SAVE`
            // answers +OK, `BGSAVE` answers
            // `Background saving started` and returns at once.
            match save_snapshot(&db, SNAPSHOT_PATH) {
                Ok(count) => {
                    println!("saved {count} key(s) to {SNAPSHOT_PATH}");
                    Frame::Simple("OK".to_string())
                }
                Err(err) => Frame::Error(format!("ERR {err}")),
            }
        }

The lock is held only while the entries are read into the buffer; the file write happens after it drops, so the critical section stays a couple of map operations long. And a missing file at startup is Ok(None), not an error:

/// Where the snapshot lives, relative to the working directory. The real
/// server's default is `dump.rdb`; the clone keeps its own name so it
/// cannot clobber a real server's snapshot when both run in the same
/// directory.
const SNAPSHOT_PATH: &str = "mini.rdb";

/// `SAVE`, end to end. The entries are read under the lock and the file
/// is written after it drops, so the critical section stays a couple of
/// map operations long.
fn save_snapshot(db: &Db, path: &str) -> Result<usize, String> {
    let (bytes, count) = {
        let store = db.lock().unwrap();
        snapshot(&store, Instant::now(), unix_ms())
    };
    std::fs::write(path, &bytes).map_err(|err| err.to_string())?;
    Ok(count)
}

/// The startup half. `Ok(None)` is no snapshot yet — the normal state
/// of a server's first boot, not an error.
fn load_snapshot(path: &str) -> Result<Option<HashMap<Vec<u8>, Entry>>, String> {
    let bytes = match std::fs::read(path) {
        Ok(bytes) => bytes,
        Err(err) if err.kind() == ErrorKind::NotFound => return Ok(None),
        Err(err) => return Err(err.to_string()),
    };
    restore(&bytes, Instant::now(), unix_ms()).map(Some)
}

And main, which loads before the port opens — the real server also loads before it accepts. A missing file is the normal case for a first boot and costs nothing; a corrupt one costs a warning, not a crash:

    let db = new_db();
    // The startup half of persistence: whatever the last run saved comes
    // back, deadlines re-anchored to this boot's clocks, and the port
    // only opens after — the real server also loads before it accepts.
    // A corrupt file costs a warning, not a crash. The real server
    // refuses to start instead ("Short read or OOM loading DB.
    // Unrecoverable error, aborting now.", captured on 7.2.4); the
    // divergence is deliberate, because a dev server whose snapshot is
    // lost should still come up.
    match load_snapshot(SNAPSHOT_PATH) {
        Ok(Some(entries)) => {
            let count = entries.len();
            db.lock().unwrap().extend(entries);
            println!("loaded {count} key(s) from {SNAPSHOT_PATH}");
        }
        Ok(None) => {}
        Err(err) => eprintln!("ignoring snapshot: {err}"),
    }

That last branch is a divergence from the real server, stated rather than hidden: a corrupt dump makes redis-server 7.2.4 refuse to start — a production server with data worth keeping should not silently come up empty — while the clone prints ignoring snapshot: … and serves. A dev server that lost its snapshot should still come up.

Seeing it run

The binary hardcodes 127.0.0.1:6379, which the real redis-server owns, so the same binary was rebuilt on a scratch port and run from an empty directory, with redis-cli pointed at it:

$ redis-cli -p 7781 set session token
OK

$ redis-cli -p 7781 lpush queue a b
(integer) 2

$ redis-cli -p 7781 expire session 100
(integer) 1

$ redis-cli -p 7781 ttl session
(integer) 100

$ redis-cli -p 7781 save
OK

The server’s own stdout shows the dump:

listening on 127.0.0.1:7781
saved 2 key(s) to mini.rdb

The file is 64 bytes — nine of header, thirty for session, twenty-five for queue. Then the process is killed — the clone has no SHUTDOWN command — and started again from the same directory:

loaded 2 key(s) from mini.rdb
listening on 127.0.0.1:7781

$ redis-cli -p 7781 get session
"token"

$ redis-cli -p 7781 ttl session
(integer) 98

$ redis-cli -p 7781 lrange queue 0 -1
1) "b"
2) "a"

Two seconds of downtime came out of the deadline, exactly as the real server’s did. The list came back whole, including its order: LPUSH put b at the head before the save, and it is still at the head after the load.

Tests

Sixty-two tests, up from fifty: the step-5 set carried over unchanged, plus twelve for the snapshot. The round trip is pinned at three levels — values, deadlines, and the file itself: every value kind survives snapshot then restore, a 100-second deadline comes back as 100 and as 60 after forty simulated seconds of downtime, a key expired while the server was down is dropped at load, an already-expired key is not written, truncated bytes and a wrong magic and an unsupported version are all errors, and save_snapshot/load_snapshot round-trip through a real file in the system temp directory. SAVE itself is tested through execute — it answers +OK, writes the file, and refuses an argument.

$ cargo test
   Compiling mini-redis v0.1.0 (/Users/rango/www/hackingwithrust/code/mini-redis)
    Finished `test` profile [unoptimized + debuginfo] target(s) in 0.75s
     Running unittests src/main.rs (/Volumes/RustBuild/tracks/debug/deps/mini_redis-be061b86284cc959)

running 62 tests
test tests::a_frame_in_two_pieces_is_incomplete_not_invalid ... ok
test tests::a_snapshot_round_trips_every_value_kind ... ok
test tests::a_key_expired_while_down_is_dropped_at_load ... ok
test tests::a_deadline_survives_the_round_trip ... ok
test tests::a_non_positive_ttl_on_a_missing_key_answers_zero ... ok
test tests::a_key_with_no_deadline_answers_minus_one ... ok
test tests::a_snapshot_with_the_wrong_magic_is_an_error ... ok
test tests::a_non_positive_ttl_deletes_the_key ... ok
test tests::a_deadline_that_has_arrived_expires_the_key_on_read ... ok
test tests::an_empty_array_is_not_a_command ... ok
test tests::a_truncated_snapshot_is_an_error ... ok
test tests::an_expire_time_past_the_cutoff_is_refused ... ok
test tests::an_expired_key_is_not_written ... ok
test tests::an_unknown_command_lists_its_arguments ... ok
test tests::an_unknown_command_caps_its_argument_list_at_128_bytes ... ok
test tests::an_unsupported_version_is_an_error ... ok
test tests::bytes_that_are_not_resp_are_invalid ... ok
test tests::del_counts_only_the_keys_that_existed ... ok
test tests::downtime_shrinks_the_ttl_the_way_it_does_on_the_real_server ... ok
test tests::echo_round_trips_through_the_codec ... ok
test tests::echo_with_the_wrong_number_of_arguments_is_refused ... ok
test tests::error_messages_quote_the_command_name_as_it_was_sent ... ok
test tests::a_buffered_second_frame_needs_no_read ... ok
test tests::every_argument_must_be_a_bulk_string ... ok
test tests::every_frame_type_round_trips ... ok
test tests::expire_on_a_missing_key_answers_zero ... ok
test tests::expire_reads_its_argument_the_way_the_real_server_does ... ok
test tests::expire_replaces_an_earlier_deadline ... ok
test tests::expire_then_ttl_reports_the_seconds_left ... ok
test tests::expire_with_the_wrong_number_of_arguments_is_refused ... ok
test tests::expire_works_on_a_list_key ... ok
test tests::get_on_a_list_is_a_wrongtype_error ... ok
test tests::get_on_a_missing_key_answers_null ... ok
test tests::a_protocol_error_ends_the_connection_without_a_reply ... ok
test tests::load_snapshot_answers_none_when_the_file_is_missing ... ok
test tests::lpush_needs_at_least_one_value ... ok
test tests::lpush_on_a_string_key_is_a_wrongtype_error ... ok
test tests::lpush_on_an_expired_key_starts_a_new_list ... ok
test tests::a_half_sent_command_does_not_block_another_connection ... ok
test tests::lpush_prepends_each_value_in_order ... ok
test tests::lrange_on_a_missing_key_answers_an_empty_array ... ok
test tests::lrange_on_a_string_key_is_a_wrongtype_error ... ok
test tests::lrange_reads_a_slice_the_way_the_real_server_clamps_it ... ok
test tests::one_decode_consumes_exactly_one_frame ... ok
test tests::lrange_with_a_non_integer_index_is_refused ... ok
test tests::ping_with_an_argument_echoes_it ... ok
test tests::save_with_an_argument_is_refused ... ok
test tests::set_clears_the_deadline ... ok
test tests::set_replaces_a_list_without_an_error ... ok
test tests::set_then_get_round_trips ... ok
test tests::the_exact_bytes_the_real_redis_cli_sends_for_ping ... ok
test tests::ping_over_a_real_socket ... ok
test tests::the_sweeper_removes_only_the_keys_whose_deadline_has_passed ... ok
test tests::pipelined_commands_are_all_answered ... ok
test tests::ttl_on_a_missing_key_answers_minus_two ... ok
test tests::save_answers_ok_and_writes_the_file ... ok
test tests::save_then_load_round_trips_through_a_file ... ok
test tests::ttl_rounds_to_the_nearest_second ... ok
test tests::ttl_with_the_wrong_number_of_arguments_is_refused ... ok
test tests::two_connections_share_one_store ... ok
test tests::the_sweeper_task_reclaims_an_expired_key_on_its_own ... ok
test tests::a_frame_split_across_two_reads_is_reassembled ... ok

test result: ok. 62 passed; 0 failed; 0 ignored; 0 measured; 0 filtered out; finished in 0.11s

The order the tests print in varies between runs; the set does not. The Running line names this machine’s shared build cache; elsewhere it reads target/debug/deps/mini_redis-….

What is deliberately not here

BGSAVE. The real server forks a child process and answers at once; the clone has one process and no fork, so SAVE blocks — the same trade the real server’s own SAVE makes.
Auto-save points. A real server configured with save 900 1 dumps on its own after enough changes; the clone saves only when told.
Durability. std::fs::write hands the bytes to the operating system and returns; nothing forces them through to the disk. A power cut between the two can lose the file’s tail. The real server has the same shape, with the fsync policy configurable.
The RDB format itself. Compression, checksums, auxiliary fields — all reduced here to a magic string, a version byte, and length prefixes.
A matching corrupt-file startup. The clone warns and starts empty; the real server refuses to start. Deliberate, and stated above.

Step 7, Pub/sub messaging, adds subscribers: SUBSCRIBE and PUBLISH, channels, and fan-out to every connection listening. It is not written yet, so it gets no link — the track overview carries the full list as articles land.

Storing values with expiry — step 5, the deadlines this step carries across a restart.
Reading and writing files — std::fs::read and std::fs::write, and treating ErrorKind::NotFound as an answer rather than a failure.
Handling errors — why every parse failure here is a Result with a message, never an unwrap.