mvcc-snapshot-isolation

Answer

The implementation (mvcc_store.py) builds a thread-safe MVCC key-value store with three core components:

1. Versioned Storage

Each key maps to a sorted list of Version objects. Every version carries: - value – the stored data - begin_ts – commit timestamp when this version was created - end_ts – timestamp when this version was superseded (or ∞ for the latest)

A transaction with start_ts = S reads the version with the largest begin_ts ≤ S and S < end_ts. Writes are buffered locally until commit.

2. First-Committer-Wins Conflict Detection

On commit, for every key in the transaction's write-set, the store checks whether any committed version has begin_ts > start_ts. If so, a concurrent transaction already wrote to that key and committed first → the current transaction aborts. This is checked atomically under a global lock.

3. Garbage Collection

GC computes min_active_start_ts (the smallest start_ts among all active transactions). A version is safe to remove when end_ts ≤ min_active_start_ts because no active snapshot can see it. As a safety net, the latest version per key is always preserved so the key list never becomes empty (every key retains at least a tombstone history).

"""mvcc_store.py – MVCC key-value store with snapshot isolation."""

import threading
import uuid

INF = float("inf")

class Version:
    __slots__ = ("value", "begin_ts", "end_ts")
    def __init__(self, value, begin_ts, end_ts=INF):
        self.value = value
        self.begin_ts = begin_ts
        self.end_ts = end_ts

class MVCCStore:
    def __init__(self):
        self._lock = threading.Lock()
        self._data = {}          # key -> [Version]  (sorted by begin_ts)
        self._clock = 0
        self._active_txns = {}   # id -> Transaction

    def _next_ts(self):
        self._clock += 1
        return self._clock

    def _min_active_start_ts(self):
        min_ts = INF
        for t in self._active_txns.values():
            if t.status == "active":
                min_ts = min(min_ts, t.start_ts)
        return min_ts

    def _read_version(self, key, start_ts):
        versions = self._data.get(key, [])
        if not versions:
            return None
        # binary search for rightmost begin_ts <= start_ts
        lo, hi, idx = 0, len(versions) - 1, -1
        while lo <= hi:
            mid = (lo + hi) // 2
            if versions[mid].begin_ts <= start_ts:
                idx, lo = mid, mid + 1
            else:
                hi = mid - 1
        if idx == -1:
            return None
        c = versions[idx]
        return c if start_ts < c.end_ts else None

    def _install_version(self, key, value, commit_ts):
        if key not in self._data:
            self._data[key] = []
        vs = self._data[key]
        if vs:
            vs[-1].end_ts = commit_ts
        vs.append(Version(value, commit_ts))

    def _has_write_conflict(self, keys, start_ts):
        for key in keys:
            vs = self._data.get(key, [])
            if vs and vs[-1].begin_ts > start_ts:
                return True
        return False

    def begin(self):
        t = Transaction(self, self._next_ts())
        with self._lock:
            self._active_txns[t.id] = t
        return t

    def gc(self):
        with self._lock:
            min_ts = self._min_active_start_ts()
            for key in list(self._data.keys()):
                kept = [v for v in self._data[key] if v.end_ts > min_ts]
                if not kept and self._data[key]:
                    kept.append(self._data[key][-1])
                self._data[key] = kept

class Transaction:
    def __init__(self, store, start_ts):
        self.id = uuid.uuid4().hex
        self.store = store
        self.start_ts = start_ts
        self.status = "active"
        self._write_set = {}

    def read(self, key):
        v = self.store._read_version(key, self.start_ts)
        return v.value if v else None

    def write(self, key, value):
        self._write_set[key] = value

    def commit(self):
        if self.status != "active":
            return False
        with self.store._lock:
            if self.store._has_write_conflict(self._write_set.keys(), self.start_ts):
                self.status = "aborted"
                self.store._active_txns.pop(self.id, None)
                return False
            commit_ts = self.store._next_ts()
            for k, v in self._write_set.items():
                self.store._install_version(k, v, commit_ts)
            self.status = "committed"
            self.store._active_txns.pop(self.id, None)
        self.store.gc()
        return True

    def rollback(self):
        if self.status != "active":
            return
        self.status = "aborted"
        with self.store._lock:
            self.store._active_txns.pop(self.id, None)

Key design decisions: - Binary search for _read_version gives O(log N) lookup per key. - Global lock simplifies concurrency; contention is minimal for KV workloads. - GC triggered on every commit keeps the version list tidy without a background thread. - Empty-list guard ensures every key has at least one version post-GC, preserving correct deletion/non-existence semantics.