Serialization and durability

import struct
import zlib

HEADER = struct.Struct(">II")    # checksum, payload length: 4 bytes each, big-endian
U32 = struct.Struct(">I")


def encode_record(op, key, value=b""):
    # op is b"S" (set) or b"D" (delete); every field is length-prefixed
    payload = op + U32.pack(len(key)) + key + U32.pack(len(value)) + value
    return HEADER.pack(zlib.crc32(payload), len(payload)) + payload


def replay(log):
    """Rebuilds the store from a log. Stops at the first torn or corrupt record."""
    store, pos, records = {}, 0, 0
    while pos + HEADER.size <= len(log):
        checksum, length = HEADER.unpack_from(log, pos)
        end = pos + HEADER.size + length
        if end > len(log):
            break                                  # torn write: the record never finished
        payload = bytes(log[pos + HEADER.size:end])
        if zlib.crc32(payload) != checksum:
            break                                  # corrupt: trust nothing after it
        (klen,) = U32.unpack_from(payload, 1)
        key = payload[5:5 + klen]
        (vlen,) = U32.unpack_from(payload, 5 + klen)
        value = payload[9 + klen:9 + klen + vlen]
        if payload[:1] == b"S":
            store[key] = value
        else:
            store.pop(key, None)
        pos, records = end, records + 1
    return store, records, pos


class DurableKV:
    def __init__(self, log=b""):
        self.log = bytearray(log)
        self.store, self.records, valid = replay(self.log)
        del self.log[valid:]             # cut a torn tail before appending after it

    def set(self, key, value):
        self.log += encode_record(b"S", key, value)    # log first...
        self.store[key] = value                         # ...then memory

    def delete(self, key):
        self.log += encode_record(b"D", key)
        self.store.pop(key, None)

#include <cstdint>
#include <map>
#include <string>
#include <tuple>
using namespace std;

// The standard CRC-32 (the same as zlib.crc32 and java.util.zip.CRC32).
uint32_t crc32(const string& data) {
    uint32_t crc = 0xFFFFFFFFu;
    for (unsigned char byte : data) {
        crc ^= byte;
        for (int bit = 0; bit < 8; bit++) crc = (crc >> 1) ^ (0xEDB88320u & (0u - (crc & 1u)));
    }
    return ~crc;
}

void put_u32(string& out, uint32_t x) {          // 4 bytes, big-endian
    for (int shift = 24; shift >= 0; shift -= 8) out += char((x >> shift) & 0xFF);
}

uint32_t get_u32(const string& s, size_t pos) {
    uint32_t x = 0;
    for (int i = 0; i < 4; i++) x = (x << 8) | (unsigned char)s[pos + i];
    return x;
}

// op is 'S' (set) or 'D' (delete); every field is length-prefixed
string encode_record(char op, const string& key, const string& value = "") {
    string payload(1, op);
    put_u32(payload, key.size());
    payload += key;
    put_u32(payload, value.size());
    payload += value;
    string record;
    put_u32(record, crc32(payload));
    put_u32(record, payload.size());
    return record + payload;
}

// Rebuilds the store from a log. Stops at the first torn or corrupt record.
// Returns {store, records replayed, bytes that were valid}.
tuple<map<string, string>, int, size_t> replay(const string& log) {
    map<string, string> store;
    size_t pos = 0;
    int records = 0;
    while (pos + 8 <= log.size()) {
        uint32_t checksum = get_u32(log, pos), length = get_u32(log, pos + 4);
        if (length > log.size() - pos - 8) break;          // torn write: the record never finished
        string payload = log.substr(pos + 8, length);
        if (crc32(payload) != checksum) break;             // corrupt: trust nothing after it
        uint32_t klen = get_u32(payload, 1);
        string key = payload.substr(5, klen);
        uint32_t vlen = get_u32(payload, 5 + klen);
        string value = payload.substr(9 + klen, vlen);
        if (payload[0] == 'S') store[key] = value;
        else store.erase(key);
        pos += 8 + length;
        records++;
    }
    return {store, records, pos};
}

class DurableKV {
public:
    string log;
    map<string, string> store;
    int records = 0;

    explicit DurableKV(const string& existing = "") {
        size_t valid;
        tie(store, records, valid) = replay(existing);
        log = existing.substr(0, valid);   // cut a torn tail before appending after it
    }

    void set(const string& key, const string& value) {
        log += encode_record('S', key, value);   // log first...
        store[key] = value;                      // ...then memory
    }

    void erase(const string& key) {
        log += encode_record('D', key);
        store.erase(key);
    }
};

import java.io.ByteArrayOutputStream;
import java.nio.ByteBuffer;
import java.nio.charset.StandardCharsets;
import java.util.*;
import java.util.zip.CRC32;

class DurableKV {
    ByteArrayOutputStream log = new ByteArrayOutputStream();
    TreeMap<String, String> store = new TreeMap<>();
    int records = 0;

    static long crc(byte[] data) {
        CRC32 c = new CRC32();
        c.update(data);
        return c.getValue();
    }

    // op is 'S' (set) or 'D' (delete); every field is length-prefixed
    static byte[] encodeRecord(char op, String key, String value) {
        byte[] k = key.getBytes(StandardCharsets.UTF_8), v = value.getBytes(StandardCharsets.UTF_8);
        ByteBuffer payload = ByteBuffer.allocate(9 + k.length + v.length);   // big-endian by default
        payload.put((byte) op).putInt(k.length).put(k).putInt(v.length).put(v);
        byte[] p = payload.array();
        return ByteBuffer.allocate(8 + p.length).putInt((int) crc(p)).putInt(p.length).put(p).array();
    }

    // Rebuilds the store from a log. Stops at the first torn or corrupt record.
    DurableKV(byte[] existing) {
        ByteBuffer buf = ByteBuffer.wrap(existing);
        int pos = 0;
        while (pos + 8 <= existing.length) {
            long checksum = buf.getInt(pos) & 0xFFFFFFFFL;
            long length = buf.getInt(pos + 4) & 0xFFFFFFFFL;
            if (length > existing.length - pos - 8) break;        // torn write: the record never finished
            byte[] p = Arrays.copyOfRange(existing, pos + 8, pos + 8 + (int) length);
            if (crc(p) != checksum) break;                        // corrupt: trust nothing after it
            ByteBuffer payload = ByteBuffer.wrap(p);
            int klen = payload.getInt(1);
            String key = new String(p, 5, klen, StandardCharsets.UTF_8);
            int vlen = payload.getInt(5 + klen);
            String value = new String(p, 9 + klen, vlen, StandardCharsets.UTF_8);
            if (p[0] == 'S') store.put(key, value);
            else store.remove(key);
            pos += 8 + (int) length;
            records++;
        }
        log.write(existing, 0, pos);   // cut a torn tail before appending after it
    }

    void set(String key, String value) {
        log.writeBytes(encodeRecord('S', key, value));   // log first...
        store.put(key, value);                           // ...then memory
    }

    void delete(String key) {
        log.writeBytes(encodeRecord('D', key, ""));
        store.remove(key);
    }
}