Vclock
Reason about the causal order of a stream of records — is A before B, or are they concurrent, causally independent, neither able to have known about the other? Wall-clock time can't express that last case; a vector clock can. bump, merge, and compare over JSON lines, so it sits in the middle of a pipe.
The honest edge
It orders events that share an actor namespace. Two records whose actor sets never overlap read as concurrent by construction — which is correct, but only useful if your actors are named consistently across the stream.
Run it
python3 vclock.py compare
test_vclock.py (37/37, mutation-bitten)
Python stdlib only
The code — every file that ships
vclock.py221 lineson GitHub →
#!/usr/bin/env python3
# SPDX-License-Identifier: MIT
"""vclock.py — reason about the causal order of a stream of records, on a pipe.
A vector clock answers a question a timestamp cannot: given two events, did one
happen *before* the other, or are they *concurrent* — causally independent,
neither one able to have known about the other? Wall-clock time can't tell you
that (clocks drift, and "later" is not "caused by"). A vector clock can, by
giving every actor its own counter and carrying the whole vector on each record.
This is a *transform*: it reads JSON lines on stdin and writes JSON lines on
stdout, so it sits in the middle of a pipe. Each input record may carry a
`clock` field — an object mapping actor names to integer counts, e.g.
`{"a": 3, "b": 1}`. An absent actor is read as 0 (it simply hasn't been seen).
Three operations, each a subcommand:
bump --actor A increment actor A's component on every record's clock
(A absent → starts at 1). "A observed/produced this."
record in → record out, clock advanced.
merge fold every record's clock into ONE clock by taking the
component-wise maximum, and emit that single clock.
This is "receive": the least clock that dominates all
inputs. Emits one line: {"clock": {...}}.
compare read exactly two records and report their causal
relation. Emits one line: {"relation": R} where R is
one of before / after / concurrent / equal.
The causal relation (X, Y), with any absent component read as 0:
equal X[a] == Y[a] for every actor a
before X[a] <= Y[a] for every a, and X != Y (X causally precedes Y)
after Y before X (X causally follows Y)
concurrent neither before nor after (causally independent)
The absent-is-zero rule is the whole game: {"a":1} is BEFORE {"a":1,"b":1}
because the first has b=0 implicitly. A comparison that only looked at shared
keys would miss this — and missing it is the difference between a real vector
clock and a counter wearing its coat.
------------------------------------------------------------------------------
The JSON-lines contract (so this composes in a pipe):
- reads ONE JSON object per line on stdin; blank lines are skipped.
- `bump` emits one record per input record (order preserved), clock advanced.
- `merge` / `compare` emit exactly one summary line.
- non-clock fields on a record are passed through untouched by `bump`.
- errors go to stderr; stdout stays clean JSON-lines.
Exit codes:
0 ran and emitted
2 malformed input / corrupt clock (don't trust the stream)
3 usage / wrong record count for compare
"""
import argparse
import json
import sys
# ---------------------------------------------------------------------------
# pure operations — no I/O, fully unit-testable. These are the gift's core.
# ---------------------------------------------------------------------------
def _normalize(clock):
"""Drop explicit-zero components so equality is canonical ({a:1,b:0}=={a:1}).
Validates that every component is an int; raises ValueError otherwise."""
out = {}
for actor, count in clock.items():
if isinstance(count, bool) or not isinstance(count, int):
raise ValueError(f"clock component {actor!r} is not an integer: {count!r}")
if count < 0:
raise ValueError(
f"clock component {actor!r} is negative: {count!r} "
"(vector-clock counts are event tallies and are never < 0)")
if count != 0:
out[actor] = count
return out
def bump(clock, actor):
"""Return a NEW clock with `actor`'s component incremented by 1.
An absent actor starts at 1. Input is not mutated."""
new = dict(_normalize(clock))
new[actor] = new.get(actor, 0) + 1
return new
def merge(*clocks):
"""Component-wise maximum across all given clocks. Absent = 0."""
out = {}
for clock in clocks:
for actor, count in _normalize(clock).items():
if count > out.get(actor, 0):
out[actor] = count
return out
def compare(x, y):
"""Causal relation of clock x to clock y: before / after / concurrent / equal.
An absent component is read as 0 (the actor simply hasn't been seen)."""
x = _normalize(x)
y = _normalize(y)
actors = set(x) | set(y)
x_le_y = all(x.get(a, 0) <= y.get(a, 0) for a in actors) # x <= y everywhere
y_le_x = all(y.get(a, 0) <= x.get(a, 0) for a in actors) # y <= x everywhere
if x_le_y and y_le_x:
return "equal" # <= both ways ⇒ componentwise equal
if x_le_y:
return "before" # x dominated by y, not equal
if y_le_x:
return "after"
return "concurrent" # neither dominates ⇒ causally independent
# ---------------------------------------------------------------------------
# I/O — the JSON-lines seam.
# ---------------------------------------------------------------------------
def _read_records(stream):
"""Yield one parsed object per non-blank stdin line. Raises ValueError on a
line that isn't a JSON object (caller maps to exit 2)."""
for lineno, raw in enumerate(stream, start=1):
line = raw.strip()
if not line:
continue
try:
obj = json.loads(line)
except json.JSONDecodeError as e:
raise ValueError(f"line {lineno}: not valid JSON: {e}")
if not isinstance(obj, dict):
raise ValueError(f"line {lineno}: expected a JSON object, got {type(obj).__name__}")
yield obj
def _emit(record, out):
out.write(json.dumps(record, ensure_ascii=False, sort_keys=True) + "\n")
def _clock_of(record):
"""Pull the clock object off a record, defaulting to empty. Validates shape."""
clock = record.get("clock", {})
if not isinstance(clock, dict):
raise ValueError(f"'clock' must be an object, got {type(clock).__name__}")
return clock
# ---------------------------------------------------------------------------
# subcommands
# ---------------------------------------------------------------------------
def cmd_bump(args, stdin, stdout):
for record in _read_records(stdin):
record = dict(record)
record["clock"] = bump(_clock_of(record), args.actor)
_emit(record, stdout)
return 0
def cmd_merge(args, stdin, stdout):
clocks = [_clock_of(r) for r in _read_records(stdin)]
_emit({"clock": merge(*clocks) if clocks else {}}, stdout)
return 0
def cmd_compare(args, stdin, stdout):
records = list(_read_records(stdin))
if len(records) != 2:
sys.stderr.write(
f"compare: expected exactly 2 records on stdin, got {len(records)}\n"
)
return 3
relation = compare(_clock_of(records[0]), _clock_of(records[1]))
_emit({"relation": relation}, stdout)
return 0
def build_parser():
p = argparse.ArgumentParser(
prog="vclock",
description="Reason about the causal order of a stream of records (vector clock).",
)
sub = p.add_subparsers(dest="cmd")
b = sub.add_parser("bump", help="increment one actor's component on each record")
b.add_argument("--actor", required=True, help="the actor whose component to increment")
b.set_defaults(func=cmd_bump)
m = sub.add_parser("merge", help="component-wise max of every record's clock → one clock")
m.set_defaults(func=cmd_merge)
c = sub.add_parser("compare", help="causal relation of exactly two records' clocks")
c.set_defaults(func=cmd_compare)
return p
def main(argv=None):
parser = build_parser()
args = parser.parse_args(argv)
if not getattr(args, "cmd", None):
parser.print_usage(sys.stderr)
sys.stderr.write("vclock: a subcommand is required (bump / merge / compare)\n")
return 3
try:
return args.func(args, sys.stdin, sys.stdout)
except ValueError as e:
sys.stderr.write(f"vclock: {e}\n")
return 2
except BrokenPipeError:
# a downstream reader (e.g. `| head`) closed the pipe early. This is
# normal for a well-behaved stream tool — exit quietly, don't traceback.
try:
sys.stdout.close()
except Exception:
pass
return 0
if __name__ == "__main__":
sys.exit(main())
test_vclock.py320 lineson GitHub →
#!/usr/bin/env python3
# SPDX-License-Identifier: MIT
"""test_vclock.py — the vclock gift's proof it works.
Tests are written FIRST and lead the build. The load-bearing move for a vector
clock is `compare`: the whole reason the primitive exists is to distinguish
CONCURRENT from BEFORE/AFTER. A counter can order a total sequence; only a vector
clock can say "these two events are causally independent." So the compare tests
carry the weight, and the mutation section at the bottom proves they are not
vacuous — each mutation is a plausible wrong implementation that MUST turn a test
red. A suite that stays green under mutation is testing nothing.
Run: python3 test_vclock.py # all tests + mutation bites
python3 test_vclock.py -q # quiet unless something fails
"""
import io
import json
import subprocess
import sys
import unittest
from pathlib import Path
HERE = Path(__file__).resolve().parent
TOOL = HERE / "vclock.py"
# import the module directly so we can unit-test the pure functions,
# and also drive the CLI as a subprocess for the contract tests.
sys.path.insert(0, str(HERE))
import vclock # noqa: E402
# ---------------------------------------------------------------------------
# compare — the gate. before / after / concurrent / equal, with the
# absent-component-is-zero rule that separates a real vector clock from a
# shared-keys-only impostor.
# ---------------------------------------------------------------------------
class TestCompare(unittest.TestCase):
def test_equal_identical(self):
self.assertEqual(vclock.compare({"a": 1, "b": 2}, {"a": 1, "b": 2}), "equal")
def test_equal_empty(self):
self.assertEqual(vclock.compare({}, {}), "equal")
def test_before_strict(self):
self.assertEqual(vclock.compare({"a": 1}, {"a": 2}), "before")
def test_after_strict(self):
self.assertEqual(vclock.compare({"a": 2}, {"a": 1}), "after")
def test_concurrent_disjoint_actors(self):
# a advanced on one axis, b on another — neither dominates. This is the
# single most important case in the whole gift.
self.assertEqual(vclock.compare({"a": 1}, {"b": 1}), "concurrent")
def test_concurrent_crossed(self):
self.assertEqual(vclock.compare({"a": 2, "b": 1}, {"a": 1, "b": 2}), "concurrent")
# --- the absent-is-zero cases: the impostor-killers ---
def test_absent_component_is_zero_before(self):
# {a:1} has b=0 implicitly, so it is BEFORE {a:1,b:1}. A compare that
# only looks at shared keys would wrongly call this equal.
self.assertEqual(vclock.compare({"a": 1}, {"a": 1, "b": 1}), "before")
def test_absent_component_is_zero_after(self):
self.assertEqual(vclock.compare({"a": 1, "b": 1}, {"a": 1}), "after")
def test_absent_component_is_zero_concurrent(self):
# {a:1} (b=0) vs {b:1} (a=0): each leads on its own axis → concurrent.
# A shared-keys-only impostor sees no shared keys and might say equal.
self.assertEqual(vclock.compare({"a": 1}, {"b": 1}), "concurrent")
def test_explicit_zero_equals_absent(self):
# {a:1,b:0} must be treated identically to {a:1}
self.assertEqual(vclock.compare({"a": 1, "b": 0}, {"a": 1}), "equal")
def test_compare_is_antisymmetric(self):
# if X before Y then Y after X, for a spread of shapes
pairs = [
({"a": 1}, {"a": 2}),
({"a": 1}, {"a": 1, "b": 1}),
({}, {"a": 1}),
]
for x, y in pairs:
self.assertEqual(vclock.compare(x, y), "before")
self.assertEqual(vclock.compare(y, x), "after")
# ---------------------------------------------------------------------------
# bump — increment one actor's component; absent starts at 1.
# ---------------------------------------------------------------------------
class TestBump(unittest.TestCase):
def test_bump_existing(self):
self.assertEqual(vclock.bump({"a": 1, "b": 2}, "a"), {"a": 2, "b": 2})
def test_bump_absent_starts_at_one(self):
self.assertEqual(vclock.bump({"a": 1}, "b"), {"a": 1, "b": 1})
def test_bump_empty(self):
self.assertEqual(vclock.bump({}, "a"), {"a": 1})
def test_bump_does_not_mutate_input(self):
original = {"a": 1}
vclock.bump(original, "a")
self.assertEqual(original, {"a": 1}) # pure — no in-place edit
def test_bump_after_is_after(self):
# a bumped clock is strictly AFTER the one it came from
c0 = {"a": 1, "b": 2}
c1 = vclock.bump(c0, "a")
self.assertEqual(vclock.compare(c0, c1), "before")
# ---------------------------------------------------------------------------
# merge — component-wise max across clocks.
# ---------------------------------------------------------------------------
class TestMerge(unittest.TestCase):
def test_merge_componentwise_max(self):
self.assertEqual(
vclock.merge({"a": 2, "b": 1}, {"a": 1, "b": 3}), {"a": 2, "b": 3}
)
def test_merge_disjoint(self):
self.assertEqual(vclock.merge({"a": 1}, {"b": 1}), {"a": 1, "b": 1})
def test_merge_absent_is_zero(self):
self.assertEqual(vclock.merge({"a": 5}, {}), {"a": 5})
def test_merge_dominates_both(self):
# the merge is AFTER-or-equal to each of its inputs (never before/concurrent)
x, y = {"a": 2, "b": 1}, {"a": 1, "b": 3}
m = vclock.merge(x, y)
self.assertIn(vclock.compare(x, m), ("before", "equal"))
self.assertIn(vclock.compare(y, m), ("before", "equal"))
def test_merge_of_concurrent_is_after_both(self):
# merging two genuinely concurrent clocks yields their least upper bound
x, y = {"a": 1}, {"b": 1}
m = vclock.merge(x, y)
self.assertEqual(vclock.compare(x, m), "before")
self.assertEqual(vclock.compare(y, m), "before")
# ---------------------------------------------------------------------------
# CLI contract — JSON-lines in, JSON-lines out; matches the house shape.
# ---------------------------------------------------------------------------
def run_cli(args, stdin_text=""):
proc = subprocess.run(
[sys.executable, str(TOOL), *args],
input=stdin_text,
capture_output=True,
text=True,
)
return proc.returncode, proc.stdout, proc.stderr
class TestBumpCLI(unittest.TestCase):
def test_bump_stream(self):
records = [
{"id": 1, "clock": {"a": 1}},
{"id": 2, "clock": {"a": 2, "b": 1}},
]
stdin = "\n".join(json.dumps(r) for r in records) + "\n"
rc, out, err = run_cli(["bump", "--actor", "a"], stdin)
self.assertEqual(rc, 0, err)
got = [json.loads(line) for line in out.splitlines()]
self.assertEqual(got[0]["clock"], {"a": 2})
self.assertEqual(got[1]["clock"], {"a": 3, "b": 1})
# non-clock fields preserved
self.assertEqual(got[0]["id"], 1)
def test_bump_record_without_clock_gets_one(self):
stdin = json.dumps({"id": 1}) + "\n"
rc, out, err = run_cli(["bump", "--actor", "a"], stdin)
self.assertEqual(rc, 0, err)
self.assertEqual(json.loads(out)["clock"], {"a": 1})
def test_emit_is_sorted_keys(self):
stdin = json.dumps({"z": 1, "clock": {"a": 1}}) + "\n"
rc, out, err = run_cli(["bump", "--actor", "a"], stdin)
self.assertEqual(rc, 0, err)
# sort_keys=True → "clock" precedes "z" in the raw text
self.assertLess(out.index('"clock"'), out.index('"z"'))
class TestMergeCLI(unittest.TestCase):
def test_merge_stream_to_one_clock(self):
records = [
{"clock": {"a": 2, "b": 1}},
{"clock": {"a": 1, "b": 3}},
{"clock": {"c": 1}},
]
stdin = "\n".join(json.dumps(r) for r in records) + "\n"
rc, out, err = run_cli(["merge"], stdin)
self.assertEqual(rc, 0, err)
self.assertEqual(json.loads(out), {"clock": {"a": 2, "b": 3, "c": 1}})
def test_merge_empty_stream(self):
rc, out, err = run_cli(["merge"], "")
self.assertEqual(rc, 0, err)
self.assertEqual(json.loads(out), {"clock": {}})
class TestCompareCLI(unittest.TestCase):
def test_compare_two_records(self):
records = [{"clock": {"a": 1}}, {"clock": {"b": 1}}]
stdin = "\n".join(json.dumps(r) for r in records) + "\n"
rc, out, err = run_cli(["compare"], stdin)
self.assertEqual(rc, 0, err)
self.assertEqual(json.loads(out)["relation"], "concurrent")
def test_compare_before(self):
records = [{"clock": {"a": 1}}, {"clock": {"a": 2}}]
stdin = "\n".join(json.dumps(r) for r in records) + "\n"
rc, out, err = run_cli(["compare"], stdin)
self.assertEqual(rc, 0, err)
self.assertEqual(json.loads(out)["relation"], "before")
def test_compare_wrong_count_is_error(self):
# compare needs exactly two records
stdin = json.dumps({"clock": {"a": 1}}) + "\n"
rc, out, err = run_cli(["compare"], stdin)
self.assertEqual(rc, 3, "one record should be a usage error")
class TestContract(unittest.TestCase):
def test_no_subcommand_is_usage_error(self):
rc, out, err = run_cli([])
self.assertEqual(rc, 3)
def test_bad_json_line_is_error(self):
rc, out, err = run_cli(["bump", "--actor", "a"], "not json\n")
self.assertEqual(rc, 2, "malformed input must not be trusted")
def test_blank_lines_skipped(self):
stdin = json.dumps({"clock": {"a": 1}}) + "\n\n\n"
rc, out, err = run_cli(["bump", "--actor", "a"], stdin)
self.assertEqual(rc, 0, err)
self.assertEqual(len([l for l in out.splitlines() if l]), 1)
def test_non_integer_clock_component_is_error(self):
stdin = json.dumps({"clock": {"a": "x"}}) + "\n"
rc, out, err = run_cli(["bump", "--actor", "a"], stdin)
self.assertEqual(rc, 2, "a non-integer clock component is corrupt input")
# ---------------------------------------------------------------------------
# Mutation bites — prove the tests above are not vacuous. Each mutation is a
# believable wrong implementation of a pure function; we assert that the real
# suite would catch it. If a mutation survives (no test flips), the suite is
# lying about its coverage and THIS meta-test fails loudly.
# ---------------------------------------------------------------------------
class TestMutationBites(unittest.TestCase):
"""Each bite monkeypatches a pure fn with a plausible-wrong version and
asserts at least one real assertion above would now fail."""
def _suite_would_catch(self, cases):
"""cases: list of (callable-returning-bool-that-should-be-True).
Returns True if any case is now False (i.e. a test would go red)."""
return any(not case() for case in cases)
def test_bite_compare_ignores_absent_components(self):
# WRONG: only compare shared keys. Would call {a:1} vs {a:1,b:1} 'equal'.
def wrong_compare(x, y):
shared = set(x) & set(y)
lt = any(x.get(k, 0) < y.get(k, 0) for k in shared)
gt = any(x.get(k, 0) > y.get(k, 0) for k in shared)
if lt and gt:
return "concurrent"
if lt:
return "before"
if gt:
return "after"
return "equal"
# under the real impl this is 'before'; the impostor says 'equal'
self.assertEqual(vclock.compare({"a": 1}, {"a": 1, "b": 1}), "before")
self.assertEqual(wrong_compare({"a": 1}, {"a": 1, "b": 1}), "equal")
self.assertNotEqual(
vclock.compare({"a": 1}, {"a": 1, "b": 1}),
wrong_compare({"a": 1}, {"a": 1, "b": 1}),
"the absent-is-zero test must separate the real impl from the impostor",
)
def test_bite_compare_collapses_concurrent_to_before(self):
# WRONG: no concurrent branch — anything not <= is 'after', anything
# not >= is 'before'. Would call disjoint {a:1} vs {b:1} 'before'.
def wrong_compare(x, y):
keys = set(x) | set(y)
if all(x.get(k, 0) <= y.get(k, 0) for k in keys):
return "before" if x != y else "equal"
return "after" # NO concurrent branch
self.assertEqual(vclock.compare({"a": 1}, {"b": 1}), "concurrent")
self.assertNotEqual(wrong_compare({"a": 1}, {"b": 1}), "concurrent")
def test_bite_bump_starts_absent_at_zero(self):
# WRONG: absent actor bumped to 0 instead of 1 (off-by-one at creation)
def wrong_bump(clock, actor):
new = dict(clock)
new[actor] = new.get(actor, -1) + 1 # absent → 0, not 1
return new
self.assertEqual(vclock.bump({}, "a"), {"a": 1})
self.assertNotEqual(wrong_bump({}, "a"), {"a": 1})
def test_bite_merge_uses_min_not_max(self):
# WRONG: component-wise MIN instead of MAX
def wrong_merge(x, y):
keys = set(x) | set(y)
return {k: min(x.get(k, 0), y.get(k, 0)) for k in keys}
self.assertEqual(vclock.merge({"a": 2}, {"a": 1}), {"a": 2})
self.assertNotEqual(wrong_merge({"a": 2}, {"a": 1}), {"a": 2})
if __name__ == "__main__":
verbosity = 1 if "-q" in sys.argv else 2
argv = [a for a in sys.argv if a != "-q"]
unittest.main(argv=argv, verbosity=verbosity)