Key-Value Store

কেন এই প্রজেক্ট

lsm-trees আর wal-recovery লেসনে আপনি diagram-এ memtable, SSTable, tombstone দেখেছেন। কিন্তু ওই diagram-গুলো একটা uncomfortable সত্য লুকিয়ে রাখে: storage engine-এর প্রতিটা সিদ্ধান্ত durability-র সঙ্গে দর-কষাকষি। কোন মুহূর্তে fsync করবেন, crash-এ কতটুকু হারানো গ্রহণযোগ্য, পড়ার পথে কতটা কাজ করতে রাজি — এগুলো slide-এ বোঝা যায় না, নিজের engine লিখলেই প্রতিদিন মুখোমুখি হতে হয়।

আর সবচেয়ে মজার অংশটা এখানেই শেষ হয় না — আপনি নিজের process নিজে মারবেন। Write-এর ঠিক মাঝপথে SIGKILL, তারপর খুলে দেখবেন কী বাঁচলো। এই crash-injection অভ্যাসটাই production database debug করার আসল দক্ষতা।

ধাপে ধাপে

১. Append-only WAL — প্রতি record-এ CRC

Record layout: [crc32(4)][key_len(2)][val_len(4)][key][value] — tombstone হলে val_len = -1:

import struct, zlib

H = struct.Struct("<HIi")                 # key_len | val_len | reserved-free

def encode_record(key: bytes, value: bytes | None) -> bytes:
    vlen = -1 if value is None else len(value)     # -1 = tombstone
    payload = struct.pack("<Hi", len(key), vlen) + key
    if value is not None:
        payload += value
    crc = zlib.crc32(payload) & 0xFFFFFFFF
    return struct.pack("<I", crc) + payload

CRC কেন? ডিস্ক sector write-এর মাঝপথে বিদ্যুৎ গেলে আংশিক record (torn write) পড়া যায় — crc mismatch বা অসম্পূর্ণ length-ই সেটা ধরে। Replay করার সময় প্রথম খারাপ record-এ থেমে যান, বাকিটা truncate।

২. Index rebuild — open-এ replay

def load_index(path):
    data = open(path, "rb").read()
    index, off = {}, 0
    while off < len(data):
        rec = try_decode(data, off)          # crc/length verify করে
        if rec is None:
            break                            # torn tail — এখানেই শেষ
        key, value, size = rec
        index[key] = (off, value)            # পরের record-ই newer, overwrite
        off += size
    return index, off                        # off = valid prefix

Hashmap-এ ভ্যালুর বদলে offset রাখলে বড় value-এর জন্য memory বাঁচে; সরলতার জন্য value-ও রাখতে পারেন — দুটোই বৈধ, trade-off লিখে রাখুন।

Checkpoint: ১০০০ key লিখে kill -9, আবার খুলে দেখুন — ঠিক ১০০০টাই আছে, একটাও কম বা বেশি নয়। এবার write-এর মাঝে kill করুন — সংখ্যা যত পূর্ণ record হয় ততটাই হবে।

৩. Memtable → SSTable flush

Memtable একটা সাধারণ dict; সাইজ threshold ছুঁলে sorted ফাইলে ঢালুন:

import os

def flush(memtable: dict, path: str) -> list:
    sparse = []                              # [(key, file_offset)]
    with open(path, "wb") as f:
        for key in sorted(memtable):
            sparse.append((key, f.tell()))
            f.write(encode_record(key, memtable[key]))
        f.flush()
        os.fsync(f.fileno())                 # flush ছাড়া durable নয়!
    return sparse

Read path এখন নতুন segment থেকে পুরনোর দিকে হাঁটে — প্রথম hit-ই উত্তর, কারণ পরে লেখা record-এর age কম। Segment তালিকা newest-first রাখুন।

৪. Tombstone delete

Append-only জগতে delete মানে আগের record মুছে নয় — নতুন ধরনের record লেখা: value-এর জায়গায় marker।

def delete(self, key: str):
    self.wal_append(key.encode(), None)      # tombstone
    self.memtable[key.encode()] = None

কেন এমন? কারণ পুরনো SSTable-এ key-টা এখনো আছে — tombstone না থাকলে read path পুরনো value-টাই ফিরিয়ে দেবে। Get করার সময় tombstone পেলে উত্তর “নেই”, হাঁটা থামান।

৫. Size-tiered compaction — newest wins

Segment সংখ্যা বাড়লে merge করুন; ভালো করে খেয়াল করুন ক্রমটা:

def compact(segments_newest_first: list[dict]) -> dict:
    merged = {}
    for seg in reversed(segments_newest_first):   # পুরনো আগে
        merged.update(seg)                         # নতুনটা পরে — জেতে
    return {k: v for k, v in merged.items()
            if v is not None}                      # নিচতলায় tombstone ফেলে দিন

Tombstone কখন নিরাপদে বাদ দেওয়া যায়? যখন নিশ্চিত যে তার চেয়ে পুরনো কোনো segment-এ সেই key নেই — এই সরল সংস্করণে সব merge-এই ফেলে দেওয়া যায়, leveled scheme-এ পারে না। Difference-টা মন্তব্যে লিখে রাখুন।

৬. Bloom filter per SSTable

প্রতিটা SSTable-এর সব key থেকে filter বানিয়ে memory-তে রাখুন:

import hashlib, math

class Bloom:
    def __init__(self, n: int, fp: float = 0.01):
        m = int(-n * math.log(fp) / (math.log(2) ** 2))
        self.bits, self.k = bytearray((m + 7) // 8), max(1, round(m / n * math.log(2)))

    def _idx(self, key: bytes):
        h = hashlib.sha256(key).digest()
        h1 = int.from_bytes(h[:8], "big")
        h2 = int.from_bytes(h[8:], "big") | 1          # odd stride
        for i in range(self.k):
            yield (h1 + i * h2) % (len(self.bits) * 8)

    def add(self, key): 
        for i in self._idx(key): self.bits[i // 8] |= 1 << (i % 8)

    def might_contain(self, key) -> bool:
        return all(self.bits[i // 8] >> (i % 8) & 1 for i in self._idx(key))

False positive হয়, false negative কখনো নয় — তাই “নেই” বললেই আপনি ওই ফাইল খোলাই বাদ দিতে পারেন। Read-heavy workload-এ এটাই সবচেয়ে সস্তা জয়।

Checkpoint: ১০ লক্ষ key-এ fp=0.01 হলে bits ≈ 9.6M — প্রতি key-তে প্রায় ৯.৬ bit। মাপ মিলছে কি না দেখুন।

৭. Crash-injection test

import subprocess, time, json

def test_prefix_recovery(tmpdir):
    client = Client(sock_path=f"{tmpdir}/sock")
    p = subprocess.Popen(["python", "-m", "kvstore.server", tmpdir])
    time.sleep(0.37)                          # ইচ্ছেমতো মুহূর্তে ছুরি
    acked = client.write_burst_and_journal(n=500)   # client-side journal
    p.kill()                                   # SIGKILL — graceful কিছু নেই
    store = KVStore.open(f"{tmpdir}")
    missing = [k for k, v in acked.items() if store.get(k) != v]
    assert not missing, f"ack করা কিন্তু হারায়নি হওন: {missing[:5]}"

Contract সুন্দরভাবে দাঁড়ায়: ack-করা write-এর prefix সবসময় বাঁচে, un-acked কিছু বাঁচতেও পারে (ঠিক আছে), কিন্তু ack-এর পরে হারানো মারাত্মক ভুল। fsync প্রতি write-এ দিলে ধীর কিন্তু শক্ত; group commit করলে দ্রুত কিন্তু window-এ হারায় — দুই policy-ই test-এ চালিয়ে দেখুন।

৮. Benchmark — ops/sec report

def bench(ops=200_000, ratio=0.5):
    t0 = time.perf_counter()
    for i in range(ops):
        if random.random() < ratio:
            store.set(f"k{i}", b"v" * 64)
        else:
            store.get(f"k{random.randrange(i + 1)}")
    dt = time.perf_counter() - t0
    return ops / dt

Workload	Bloom off	Bloom on
95% read	?	?
50/50 mix	?	?
95% write	?	?

Compaction চলার সময় write latency p99 কেমন spike করে সেটাও নমুনা দিন — এই একটা টেবিলই আপনার engine-এর চরিত্র।

নিজেকে চ্যালেঞ্জ করুন

1. Group commit — ক’টা write একত্রে এক fsync, throughput vs durability graph
2. Per-block zlib compression — CPU খরচ বনাম ফাইল সাইজ মাপুন
3. Snapshot read — copy-on-write index দিয়ে consistent view, concurrent reader
4. Leveled compaction — size-tiered-এর সঙ্গে write/read amplification তুলনা
5. Range scan — per-segment sorted order কাজে লাগিয়ে k-way merge iterator
6. Checksum upgrade — CRC32 বাদ দিয়ে xxHash64 + length prefix, কেন ভালো?

যা এখানে শিখলেন, পরে যেখানে লাগবে

এখানে	পরে
WAL + CRC + prefix recovery	Level 9 — distributed log, Kafka-র একই প্রাণ
Tombstone + newest-wins merge	CRDT আর event sourcing-এ conflict resolution
Bloom filter	Level 12 — Cassandra/HBase প্রতিটা read path-এ এই একই trick
Compaction trade-off	যেকোনো LSM system tuning — RocksDB-র manual এখন পড়া যায়
Crash-injection habit	সব durability claim-এর একমাত্র বিশ্বস্ত প্রমাণ