## The problem I kept running into

I was building a small system made of multiple services (one API, one worker, one “data fixer” job). The goal was simple: run end-to-end integration tests locally, with minimal friction.

The frustrating part was *state drift*. Every time I changed code, tests either:

1. reused stale artifacts (old build outputs, old container images, old configs), or  
2. needed a manual “clean rebuild” step that killed my feedback loop.

So I set out to design a very specific solution: a **Git-aware hot reload orchestrator** for local integration tests that automatically rebuilds *only when the relevant parts changed*—and does so by inspecting the current Git diff.

## The design: a tiny orchestrator with three contracts

I ended up with a pipeline split into three clear steps:

1. **Detect what changed** using `git diff --name-only`.
2. **Map file changes to build/test targets** using a small rules table.
3. **Run tasks in the right order** and write “done markers” so we don’t rebuild unnecessarily.

### Key design choices

- **No magic**: change detection is explicit and based on file paths.
- **Deterministic**: task selection is purely a function of the diff.
- **Fast feedback**: we avoid running everything on every change.
- **Local-only**: this is meant for developer machines (so we don’t need heavy distributed coordination).

## What counts as “relevant change”?

To keep this niche and practical, I hard-coded rules that reflect the structure of a typical repo:

- `api/**` → rebuild API and rerun API-related tests
- `worker/**` → rebuild worker and rerun worker-related tests
- `shared/**` → rebuild both + rerun all integration tests
- `migrations/**` → run migrations before tests
- `docker/**` or `Dockerfile*` → rebuild everything because runtime environment changes

This mapping is the heart of the system design. It’s “system architecture” in miniature: a policy layer on top of execution.

## Step-by-step implementation (working code)

Below is a working Python script that:

- figures out the files changed since a base commit,
- determines which tasks to run,
- runs them,
- and caches outcomes using a marker file in `.cache/`.

### 1) `git_diff_hot_reload.py`

```python
#!/usr/bin/env python3
import hashlib
import json
import os
import subprocess
import sys
from dataclasses import dataclass
from pathlib import Path
from typing import Dict, List, Set


REPO_ROOT = Path(__file__).resolve().parent
CACHE_DIR = REPO_ROOT / ".cache"
CACHE_DIR.mkdir(exist_ok=True)

MARKER_DIR = CACHE_DIR / "markers"
MARKER_DIR.mkdir(exist_ok=True)


@dataclass(frozen=True)
class Rule:
    prefixes: List[str]
    targets: List[str]  # symbolic task names


RULES: List[Rule] = [
    Rule(prefixes=["api/"], targets=["build_api", "test_integration_api"]),
    Rule(prefixes=["worker/"], targets=["build_worker", "test_integration_worker"]),
    Rule(prefixes=["shared/"], targets=["build_api", "build_worker", "test_integration_all"]),
    Rule(prefixes=["migrations/"], targets=["run_migrations", "test_integration_all"]),
    Rule(prefixes=["docker/", "Dockerfile", "Dockerfile.", ".github/workflows/"], targets=["build_all", "test_integration_all"]),
    # fallback safety: if anything unknown changes, run everything
    Rule(prefixes=["."], targets=["build_all", "test_integration_all"]),
]


TASKS: Dict[str, List[str]] = {
    # In a real project these would be actual commands; for a working demo
    # they are shell-friendly placeholders.
    "build_api": ["bash", "-lc", "echo '[build_api] compiling api...' && sleep 0.2"],
    "build_worker": ["bash", "-lc", "echo '[build_worker] compiling worker...' && sleep 0.2"],
    "build_all": ["bash", "-lc", "echo '[build_all] building full system...' && sleep 0.3"],
    "run_migrations": ["bash", "-lc", "echo '[run_migrations] applying migrations...' && sleep 0.2"],
    "test_integration_api": ["bash", "-lc", "echo '[tests] running API integration tests...' && sleep 0.2"],
    "test_integration_worker": ["bash", "-lc", "echo '[tests] running worker integration tests...' && sleep 0.2"],
    "test_integration_all": ["bash", "-lc", "echo '[tests] running full integration suite...' && sleep 0.25"],
}


def run(cmd: List[str], cwd: Path = REPO_ROOT) -> str:
    proc = subprocess.run(cmd, cwd=str(cwd), text=True, capture_output=True)
    if proc.returncode != 0:
        raise RuntimeError(f"Command failed: {cmd}\nSTDOUT:\n{proc.stdout}\nSTDERR:\n{proc.stderr}")
    return proc.stdout.strip()


def get_changed_files(base_ref: str, head_ref: str) -> List[str]:
    # name-only lists paths changed between the two refs
    out = run(["git", "diff", "--name-only", f"{base_ref}..{head_ref}"])
    files = [line.strip() for line in out.splitlines() if line.strip()]
    return files


def decide_targets(changed_files: List[str]) -> Set[str]:
    selected: Set[str] = set()

    # Track whether we matched any specific prefix.
    matched_specific = False

    for f in changed_files:
        # Special: ignore deleted/renamed path ambiguity for simplicity in this local tool.
        for rule in RULES:
            # If a rule prefix is ".", it acts as a fallback.
            if rule.prefixes == ["."]:
                continue
            if any(f.startswith(pfx) for pfx in rule.prefixes):
                matched_specific = True
                selected.update(rule.targets)
                # Break after first match to keep behavior predictable.
                break

    if not matched_specific:
        # If we changed files but they didn't match known prefixes, be safe.
        selected.update(["build_all", "test_integration_all"])

    # If there's a fallback rule, keep it as the safety net.
    if not selected:
        selected.update(["build_all", "test_integration_all"])

    return selected


def cache_key_for(changed_files: List[str], targets: Set[str]) -> str:
    payload = {
        "changed_files": sorted(changed_files),
        "targets": sorted(targets),
    }
    raw = json.dumps(payload, sort_keys=True).encode("utf-8")
    return hashlib.sha256(raw).hexdigest()


def marker_path(task: str, key: str) -> Path:
    safe_task = task.replace("/", "_")
    return MARKER_DIR / f"{safe_task}.{key}.done"


def run_tasks_if_needed(changed_files: List[str], targets: Set[str], key: str) -> None:
    for task in sorted(targets):
        marker = marker_path(task, key)
        if marker.exists():
            print(f"[skip] {task} (cached)")
            continue

        cmd = TASKS.get(task)
        if not cmd:
            raise KeyError(f"No command configured for task '{task}'")

        print(f"[run] {task}")
        run(cmd)

        marker.write_text("done\n", encoding="utf-8")


def main(argv: List[str]) -> int:
    base_ref = "origin/main"
    head_ref = "HEAD"

    # Small CLI: allow overriding base ref for local workflows
    if len(argv) >= 2:
        base_ref = argv[1]
    if len(argv) >= 3:
        head_ref = argv[2]

    changed_files = get_changed_files(base_ref, head_ref)
    print(f"[info] changed files between {base_ref}..{head_ref}:")
    for f in changed_files:
        print(f"  - {f}")

    targets = decide_targets(changed_files)
    print(f"[info] selected targets: {sorted(targets)}")

    key = cache_key_for(changed_files, targets)
    run_tasks_if_needed(changed_files, targets, key)

    print("[done] hot reload orchestration finished successfully.")
    return 0


if __name__ == "__main__":
    raise SystemExit(main(sys.argv))
```

### 2) How it works (and what happens on a run)

When you run:

```bash
python3 git_diff_hot_reload.py
```

the script:

1. Calls Git to list changed files:
   - `git diff --name-only origin/main..HEAD`
2. Assigns tasks using path prefixes.
3. Computes a cache key from:
   - the file list
   - the selected task set
4. For each selected task:
   - checks for a marker file like:
     - `.cache/markers/build_api.<hash>.done`
   - if it exists, it skips the task
   - otherwise it runs the configured shell command and writes the marker

That means: **re-running the script without new relevant changes becomes instant**.

## Making it feel like “hot reload” (tight feedback loop)

To use this as a workflow, I wired it into a pre-test step.

A common pattern is:

- rebuild/test only what changed,
- then run a stable test runner.

Here’s an example “integration test runner” that uses the orchestrator output implicitly (via markers) and then starts tests that rely on the rebuilt services.

```python
#!/usr/bin/env python3
import subprocess
import sys

def main() -> int:
    # First run the orchestrator to ensure the right components are up-to-date.
    subprocess.check_call(["python3", "git_diff_hot_reload.py", "origin/main", "HEAD"])

    # Then run a single stable command.
    subprocess.check_call(["bash", "-lc", "echo '[tests] running docker-compose integration checks...' && sleep 0.2"])
    return 0

if __name__ == "__main__":
    raise SystemExit(main())
```

In my local setup, the “rebuild” commands were hooked into whatever tooling I used (Make, docker build, mvn/gradle, etc.). The important part wasn’t the build tool—it was the *diff-driven selection*.

## System design notes I wish I had known earlier

### 1) Path-based rules are an architecture, not a hack
The rules table is where the system’s “domain knowledge” lives. It’s a policy layer:
- detector (diff)
- classifier (paths → targets)
- executor (targets → commands)

I treated that like a first-class design artifact, and everything got easier to reason about.

### 2) Cache markers should be keyed to intent, not time
A common mistake is caching “last build” by timestamp. My version keys to the *changed inputs*.
So if two unrelated diffs happen to arrive close together, the cache won’t lie.

### 3) Safety beats cleverness when rules don’t match
If a change doesn’t match known prefixes, I default to:
- `build_all`
- `test_integration_all`

That’s less efficient, but it prevents a subtle failure mode I hit once: tests passing against an environment that was missing a required migration or rebuild.

## Conclusion

I built a Git-aware hot reload orchestrator for local system integration tests by designing a small three-part pipeline: detect changed files with `git diff`, map those paths to explicit build/test targets via rules, and execute tasks with deterministic caching markers. The result was a much tighter feedback loop and a system that avoids stale artifacts without the “always rebuild everything” tax.

Building A Git-Aware “Hot Reload” Plugin For Local System Integration Tests

I ran into a frustrating problem while building a little “best trade-offs” engine: I needed a live **Pareto front** (set of non-dominated solutions) that updates as new candidate points arrive. The catch was practical: I needed the result to be **deterministic** and **fast**, even when many points share equal coordinates.

### What problem I was actually solving

Each candidate solution is a point in 2D: `(cost, quality)`. A point `A` **dominates** point `B` if:

- `cost_A <= cost_B` (no worse cost)
- `quality_A >= quality_B` (no worse quality)
- and at least one of those is strictly better

The **Pareto front** is the set of points that aren’t dominated by any other.

The “gotchas” that made this niche (and annoying) were:

1. **Incremental updates:** points arrive one-by-one, and I need to update the Pareto front without recomputing everything.
2. **Determinism with ties:** if two points have the same `cost` and different `quality`, only the best survives; if they have equal quality, order must not affect the result.
3. **Avoiding O(n²):** naive “check against every front point” can degrade quickly.

I ended up using a deterministic structure I call **incremental dominance buckets**: I maintain a set of “front” points sorted by cost, and I only scan the portion that could possibly be affected by the incoming point.

---

## The core idea: monotonic frontier scanning

If the Pareto front is sorted by increasing `cost`, then **quality along the frontier is strictly increasing** (when you remove dominated points).

That means: when a new point `(c, q)` arrives, the only way it can get onto the front is if it beats the current frontier’s quality at the appropriate position.

Concretely, I keep:

- A sorted list of frontier points by `cost`
- A list of the corresponding `quality` values
- A `dominates` rule to decide membership
- A “scan and splice” step to remove newly dominated neighbors

This is conceptually similar to maintaining a monotonic skyline, but the determinism comes from handling equal `cost` carefully.

---

## Implementation (Python): incremental dominance buckets

This version is built to be **repeatable** regardless of insertion order for equal coordinates.

```python
from bisect import bisect_left
from dataclasses import dataclass
from typing import List, Tuple, Optional


@dataclass(frozen=True, order=True)
class Point:
    cost: int
    quality: int


def pareto_front_insert(front: List[Point], p: Point) -> List[Point]:
    """
    Incrementally update Pareto front in 2D where:
      - lower cost is better
      - higher quality is better

    front is maintained as:
      - sorted by cost ascending
      - with strictly increasing quality across successive costs
    """
    # 1) Handle exact tie on cost deterministically:
    #    Among equal-cost points, only keep the one with highest quality.
    costs = [pt.cost for pt in front]
    i = bisect_left(costs, p.cost)

    # If there is already a point with the same cost:
    if i < len(front) and front[i].cost == p.cost:
        # If existing dominates p (same cost, higher/equal quality), p is discarded.
        if front[i].quality >= p.quality:
            return front
        # Otherwise p dominates the equal-cost one; replace it.
        front = front[:i] + front[i + 1:]

    # 2) Check whether p is dominated by the neighbor to the left.
    #    Since front qualities are increasing, we only need to compare with the
    #    predecessor by cost.
    costs = [pt.cost for pt in front]
    i = bisect_left(costs, p.cost)

    # predecessor exists?
    if i - 1 >= 0:
        pred = front[i - 1]
        # pred has cost <= p.cost. If pred also has quality >= p.quality,
        # then pred dominates p.
        if pred.quality >= p.quality:
            return front

    # 3) Insert p into the correct position.
    front = front[:i] + [p] + front[i:]

    # 4) Remove points dominated by p to the right.
    #    Because qualities increase with cost on the current frontier,
    #    as soon as we find a point with quality > p.quality, p no longer dominates further points.
    j = i + 1
    while j < len(front) and front[j].quality <= p.quality:
        j += 1

    # Points in [i+1, j) are dominated by p (their quality <= p.quality)
    front = front[:i + 1] + front[j:]

    # 5) Remove points dominated by the left neighbor due to replacement gaps.
    #    In this algorithm, this typically isn't needed after the initial checks,
    #    but it keeps invariants tight if the data contains duplicates or replacements.
    #    Ensure frontier quality is increasing.
    #    Remove any unexpected violations (quality not increasing).
    k = 1
    while k < len(front):
        if front[k - 1].quality >= front[k].quality:
            # left dominates right (since cost strictly increases, left has <= cost and >= quality)
            front.pop(k)
        else:
            k += 1

    return front


def build_front(points: List[Tuple[int, int]]) -> List[Point]:
    front: List[Point] = []
    for cost, quality in points:
        front = pareto_front_insert(front, Point(cost, quality))
    return front


def main():
    # Designed to stress:
    # - equal costs
    # - dominated points
    # - insertion order changes
    stream = [
        (5, 1),
        (2, 2),
        (3, 1),  # dominated by (2,2) because cheaper and better quality? Actually cost 2<=3 and quality 2>=1 => yes dominated
        (4, 2),  # not dominated: (2,2) has quality ==2 but cost 2<4; since quality not worse, (2,2) dominates only if quality >= and cost <=, yes dominates (4,2) with equal quality
        (4, 3),  # replaces frontier around cost 4, dominates (4,2) (but (4,2) never survived)
        (1, 1),  # dominated by none? (1,1) might be dominated by (1,1) itself? Check: (2,2) has cost 2>1 so cannot dominate (cost must be <=). So (1,1) stays only if not dominated by predecessor (none) => stays, but later (2,2) might dominate it? yes if quality >= => (2,2) dominates (1,1) would require cost 2<=1 which is false. So both can stay.
        (1, 2),  # same cost as (1,1) but better quality; replaces deterministically
    ]

    front = build_front(stream)
    print("Final Pareto front (cost asc, quality asc):")
    for pt in front:
        print(pt)


if __name__ == "__main__":
    main()
```

### What happens when I run it

- Points with the same `cost` replace the earlier one only if they have higher `quality`.
- For a new point, I first check the immediate predecessor (the only one that can dominate on the left).
- Then I insert it.
- Finally, I scan forward removing any points whose `quality` is `<=` the new point’s `quality`.

This avoids checking every old front point every time.

---

## Step-by-step walkthrough on a small sequence

Take this pair of inserts:

1. Insert `(5, 1)` → front is `[(5,1)]`
2. Insert `(2, 2)`:
   - predecessor doesn’t exist
   - insert position is front start
   - remove dominated points on the right:
     - `(5,1)` has quality `1 <= 2`, so it gets removed
   - front becomes `[(2,2)]`

3. Insert `(3, 1)`:
   - predecessor is `(2,2)` with quality `2 >= 1`, so it’s dominated → discarded

This matches the dominance definition exactly, but does so with only local checks.

---

## Why this is deterministic (the tie rules)

Without special handling, insertion order can create different front sets when points share the same `cost` or when “equal quality” interactions occur.

The determinism is enforced by this rule:

- For equal `cost`, keep only the point with **maximum `quality`**
- If the incoming point isn’t strictly better at that cost, it’s discarded immediately

So `(4,2)` will always lose to `(4,3)` even if `(4,2)` arrived first.

---

## Complexity notes (practical, not academic)

Let `m` be the current size of the Pareto front.

- The predecessor check is effectively `O(log m)` for the binary search in the sorted costs list.
- The removal scan after insertion can be `O(m)` in the worst case, but each dominated point is removed when it becomes invalid.
- This is usually far faster than checking all candidates against all front points.

For streaming workloads where the frontier stays small compared to total points, it performs well.

---

## Conclusion

I built an incremental Pareto front updater using deterministic **incremental dominance buckets**: maintain a sorted frontier with monotonic quality, reject points dominated by their predecessor, insert, then scan rightward to remove any newly dominated neighbors. The key “engineering detail” that made results stable was explicit handling of equal `cost` so ties can’t create insertion-order-dependent front sets.

Deterministic Pareto Front Updates Via Incremental Dominance Buckets

## The weird bug that sent me into JSON-LD debugging mode

I hit an annoying mismatch while building an API that returned JSON-LD (JSON for Linked Data). Everything looked correct when I inspected the payload in a browser—but downstream services kept treating parts of my JSON as different “identities” depending on whether they fetched the context via an HTTP `Link` header or embedded it directly.

The root cause ended up being *not* JSON itself, but how the remote system resolved the JSON-LD “context” (a document that tells JSON-LD how to interpret terms like `"name"` or `"schema:Person"`). In JSON-LD, the *context resolution* step affects how the final expanded graph is produced, and that means different context resolution can lead to different output.

I wanted something deterministic: given the same semantic data, I wanted the same expanded result—even when different clients fetch contexts differently.

So I built a pattern I now use: **advertise a JSON-LD context via HTTP `Link` headers and also include a deterministic hash of that context in a dedicated response header**, so clients can verify they’re using the exact context version.

Below is what I found and the small working implementation I used to make it reliable.

---

## Key pieces: JSON-LD context, HTTP Link headers, and a context hash

### JSON-LD context (plain-language)
A JSON-LD *context* maps short property names to full IRIs (Internationalized Resource Identifiers). For example, you might map `"type"` to `"@type"` or `"name"` to a schema.org property.

### HTTP `Link` header
HTTP allows servers to attach “metadata links” using the `Link` header. A common pattern is:

- `rel="http://www.w3.org/ns/json-ld#context"` to point to a JSON-LD context resource.
- The client can then fetch the context from the URL.

### Deterministic context hash (the reliability trick)
Different services sometimes fetch and resolve contexts in different ways, especially if you later change the context document.

So the trick is:
1. Compute a stable hash of the context document content.
2. Send that hash in a response header.
3. Require (or at least detect) that the client resolved the same context content.

For hashing JSON documents deterministically, I used a canonicalization strategy based on “sort keys and encode consistently”. (The goal is to avoid “same JSON, different whitespace/order” producing different hashes.)

---

## A working Node.js example: API response + Link header + context hash

This example:
- Hosts a JSON-LD context at `/contexts/person.jsonld`
- Returns JSON-LD data from `/api/person/:id`
- Adds:
  - a `Link` header advertising the context URL
  - an `X-JSONLD-Context-SHA256` header with the deterministic hash

### Project setup

```bash
npm init -y
npm i express
```

### Server: `server.js`

```javascript
// server.js
import express from "express";
import crypto from "crypto";

const app = express();

// 1) Define a JSON-LD context document (served as-is)
const personContext = {
  "@context": {
    "@version": 1.1,
    "name": "http://schema.org/name",
    "type": "@type",
    "knows": {
      "@id": "http://schema.org/knows",
      "@type": "@id"
    }
  }
};

// 2) A deterministic JSON stringifier:
//    - sorts object keys
//    - produces stable UTF-8 encoding
function stableStringify(value) {
  if (value === null) return "null";
  if (typeof value === "number" || typeof value === "boolean") return String(value);
  if (typeof value === "string") return JSON.stringify(value);

  if (Array.isArray(value)) {
    return `[${value.map(stableStringify).join(",")}]`;
  }

  if (typeof value === "object") {
    const keys = Object.keys(value).sort();
    return `{${keys.map((k) => `${JSON.stringify(k)}:${stableStringify(value[k])}`).join(",")}}`;
  }

  // Fallback (shouldn't happen for this example)
  return JSON.stringify(value);
}

// 3) Compute the deterministic SHA-256 hash of the context content
const contextCanonical = stableStringify(personContext);
const contextHash = crypto.createHash("sha256").update(contextCanonical, "utf8").digest("hex");

// 4) Serve the context at a stable URL
app.get("/contexts/person.jsonld", (req, res) => {
  res.type("application/ld+json").json(personContext);
});

// 5) Return JSON-LD data with Link header and hash header
app.get("/api/person/:id", (req, res) => {
  const { id } = req.params;

  // JSON-LD data: note the compact terms that rely on the external context
  const doc = {
    "@context": "https://example.test/contexts/person.jsonld",
    "@id": `https://example.test/person/${id}`,
    "type": "http://schema.org/Person",
    "name": `Person ${id}`,
    "knows": [`https://example.test/person/${Number(id) + 1}`]
  };

  res.setHeader(
    "Link",
    `<https://example.test/contexts/person.jsonld>; rel="http://www.w3.org/ns/json-ld#context"; type="application/ld+json"; anchor="{}"`
  );

  res.setHeader("X-JSONLD-Context-SHA256", contextHash);
  res.setHeader("Content-Type", "application/ld+json");

  res.status(200).json(doc);
});

app.listen(3000, () => {
  console.log("Listening on http://localhost:3000");
  console.log("Context is at http://localhost:3000/contexts/person.jsonld");
});
```

### Run it

```bash
node server.js
```

---

## What happens when you call the API

### Fetch the context

```bash
curl -s http://localhost:3000/contexts/person.jsonld | jq
```

You’ll see the JSON-LD context structure.

### Fetch the person JSON-LD document

```bash
curl -i http://localhost:3000/api/person/1
```

You should see headers like:

- `Link: <.../contexts/person.jsonld>; rel="http://www.w3.org/ns/json-ld#context"; ...`
- `X-JSONLD-Context-SHA256: <hex>`

And a JSON-LD body containing:

- `@context` pointing to the external context URL
- compact terms like `"name"` and `"knows"` that depend on that context

---

## Why the context hash matters (and where it actually saved me)

The downstream systems I was integrating with did something like:

1. Fetch JSON document
2. Resolve context (either from `@context` or from a `Link` header)
3. Expand JSON-LD into a canonical-ish internal graph
4. Normalize or hash the expanded graph for deduplication

If the remote service fetched a *different* version of the context (even if it was “close enough” semantically), the expanded representation could differ, and deduplication broke.

By sending:

- a clear `Link` header for discovery
- a hash for verification

…I could detect when resolution didn’t match.

---

## Client-side verification: compare the received context hash

Here’s a small Node snippet that:
- downloads the context from the `Link` header
- computes the deterministic hash
- checks it matches `X-JSONLD-Context-SHA256`

### `verify.js`

```javascript
// verify.js
import crypto from "crypto";

function stableStringify(value) {
  if (value === null) return "null";
  if (typeof value === "number" || typeof value === "boolean") return String(value);
  if (typeof value === "string") return JSON.stringify(value);

  if (Array.isArray(value)) {
    return `[${value.map(stableStringify).join(",")}]`;
  }

  if (typeof value === "object") {
    const keys = Object.keys(value).sort();
    return `{${keys.map((k) => `${JSON.stringify(k)}:${stableStringify(value[k])}`).join(",")}}`;
  }

  return JSON.stringify(value);
}

function sha256Hex(s) {
  return crypto.createHash("sha256").update(s, "utf8").digest("hex");
}

function extractContextUrlFromLinkHeader(linkHeader) {
  // Very small parser for this specific pattern:
  // <URL>; rel="http://www.w3.org/ns/json-ld#context";
  const match = linkHeader.match(/<([^>]+)>;\s*rel="http:\/\/www\.w3\.org\/ns\/json-ld#context"/);
  return match ? match[1] : null;
}

async function run() {
  const personUrl = "http://localhost:3000/api/person/1";

  const resp = await fetch(personUrl);
  const headers = resp.headers;

  const link = headers.get("Link") || "";
  const expectedHash = headers.get("X-JSONLD-Context-SHA256");

  const contextUrl = extractContextUrlFromLinkHeader(link);
  if (!contextUrl) throw new Error("Could not find JSON-LD context URL in Link header");

  const contextResp = await fetch(contextUrl);
  const contextJson = await contextResp.json();

  const canonical = stableStringify(contextJson);
  const actualHash = sha256Hex(canonical);

  console.log("Context URL:", contextUrl);
  console.log("Expected hash:", expectedHash);
  console.log("Actual hash:  ", actualHash);

  if (expectedHash !== actualHash) {
    throw new Error("Context hash mismatch: client resolved different context content");
  }

  console.log("Context hash matches. Resolution is deterministic for this context version.");
}

run().catch((e) => {
  console.error(e);
  process.exit(1);
});
```

### Run it

```bash
node verify.js
```

This confirms that the context content the client resolved is exactly what the server intended to advertise.

---

## Practical notes I learned the hard way

### Hashing JSON needs a stable canonicalization rule
If I simply did `JSON.stringify(personContext)`, the hash could change if the object key insertion order changed (or if different languages serialized differently). Sorting keys fixed that for this example.

In production, I’d typically standardize on a well-defined JSON canonicalization approach (or a canonical RDF dataset hashing approach) depending on how strict I needed to be. The point is: **hash what you mean**, consistently.

### Separate “discovery” from “verification”
The `Link` header answers: *where can I find the context?*  
The `X-JSONLD-Context-SHA256` answers: *is it the exact one you expect?*

Together, they prevent silent drift.

---

## Conclusion

I learned that JSON-LD bugs often aren’t about the JSON payload—they’re about how the JSON-LD context gets resolved. By pairing a JSON-LD context `Link` header (for discovery) with a deterministic SHA-256 hash of the context document (for verification), I made context resolution deterministic enough for downstream deduplication and normalization. The result was a much more stable API/data contract when multiple systems interpreted my data.

Specifying Json-Ld Contexts With Http Link Headers And A Deterministic Hash

I got bitten by a bug that looked “nondeterministic” at first: sometimes my API returned the *exact same data* but the downstream system rejected it anyway. No stack traces, no obvious validation errors—just a polite “invalid payload.”

What finally cracked it wasn’t deeper logging or more retries. It was a byte-exact diff harness that compared the raw JSON bytes (including whitespace and key order) between “expected” and “actual.” Once I could see the differences at the byte level, the culprit became obvious: the JSON encoder was changing formatting and key ordering in a way the downstream system treated as significant.

Below is the small deterministic harness I built to make those issues impossible to miss.

## The core problem: “valid JSON” can still be rejected

JSON validity is about syntax (brackets/quotes/commas), but some systems also do strict checks on the exact serialized representation—commonly:

- **Key ordering**: some consumers expect keys in a particular order.
- **Whitespace**: some legacy parsers normalize incorrectly (or even compare raw strings).
- **Escaping rules**: Unicode escaping differences can matter.
- **Number formatting**: `1` vs `1.0`, exponent notation, etc.

My payloads were valid JSON, but the *string form* wasn’t what the consumer expected.

## What I built: a byte-exact diff harness

The idea is simple:

1. Generate an “expected” JSON string.
2. Generate an “actual” JSON string from the code under test.
3. Compare the raw bytes and show the *first byte where they differ*.
4. Also parse into a stable structure (so I can distinguish “semantic difference” from “serialization difference”).

### A tiny Python script that pinpoints the first differing byte

```python
# json_byte_diff.py
import json
import difflib

def to_bytes(s: str) -> bytes:
    # Use UTF-8 since JSON strings are typically UTF-8 on the wire.
    return s.encode("utf-8")

def describe_first_difference(expected: bytes, actual: bytes) -> str:
    max_len = min(len(expected), len(actual))
    for i in range(max_len):
        if expected[i] != actual[i]:
            # Show a small window around the mismatch for context.
            exp_slice = expected[max(0, i - 10): i + 10]
            act_slice = actual[max(0, i - 10): i + 10]
            return (
                f"First mismatch at byte index {i}\n"
                f"Expected byte: {expected[i]} (0x{expected[i]:02x})\n"
                f"Actual byte:   {actual[i]} (0x{actual[i]:02x})\n"
                f"Expected context (utf-8 with replacement): {exp_slice.decode('utf-8', 'replace')}\n"
                f"Actual context   (utf-8 with replacement): {act_slice.decode('utf-8', 'replace')}\n"
            )
    if len(expected) != len(actual):
        return (
            f"No mismatch in first {max_len} bytes, but length differs:\n"
            f"Expected length: {len(expected)}\n"
            f"Actual length:   {len(actual)}\n"
        )
    return "Byte-for-byte identical."

def semantic_compare(expected_json: str, actual_json: str) -> str:
    # Parse JSON into Python objects. This ignores formatting differences
    # and normalizes key order at the object level.
    expected_obj = json.loads(expected_json)
    actual_obj = json.loads(actual_json)

    if expected_obj == actual_obj:
        return "Semantic equality: parsed objects match (differences are purely serialization/formatting)."
    return "Semantic inequality: parsed objects differ."

def show_unified_diff(expected_json: str, actual_json: str) -> None:
    expected_lines = expected_json.splitlines(keepends=True)
    actual_lines = actual_json.splitlines(keepends=True)
    diff = difflib.unified_diff(
        expected_lines,
        actual_lines,
        fromfile="expected.json",
        tofile="actual.json",
    )
    print("---- Unified diff (line-based) ----")
    for line in diff:
        print(line.rstrip("\n"))

def main():
    # This simulates the “expected” payload the consumer strictly wants.
    # Note key order and whitespace are part of the raw bytes.
    expected_json = '{"a":1,"b":2,"nested":{"x":true,"y":"ok"}}'

    # This simulates a different serialization produced by the code under test.
    # json.dumps() with sort_keys=False preserves insertion order of dicts,
    # but if you build dicts differently, key order can change.
    actual_obj = {
        "b": 2,
        "a": 1,
        "nested": {"y": "ok", "x": True},
    }

    # Example of a common “looks fine but bytes differ” issue:
    # - separators add spaces
    # - ensure_ascii changes escaping for non-ascii (not shown here)
    actual_json = json.dumps(actual_obj, ensure_ascii=False, separators=(", ", ": "))

    expected_bytes = to_bytes(expected_json)
    actual_bytes = to_bytes(actual_json)

    print("---- Byte-level report ----")
    print(describe_first_difference(expected_bytes, actual_bytes))

    print("---- Semantic report ----")
    print(semantic_compare(expected_json, actual_json))

    show_unified_diff(expected_json, actual_json)

if __name__ == "__main__":
    main()
```

Run it:

```bash
python json_byte_diff.py
```

You’ll see:

- The **first mismatching byte index**
- A small **context window** around that mismatch
- A **semantic comparison** result that usually says the objects match (meaning the consumer is sensitive to serialization, not meaning)

In the common case above, the semantic objects match, but the bytes differ due to whitespace and key order.

## A step-by-step “root cause” workflow

I used this exact flow when the consumer rejected my payloads:

### 1) Confirm whether it’s semantic or serialization mismatch

The semantic comparison uses `json.loads()`, which parses JSON into objects (dicts/lists). That ignores whitespace, and key order generally won’t matter *for equality of parsed objects*.

- If semantic equality is **true**, the bug is almost certainly about **encoding/serialization format**.
- If semantic equality is **false**, then the data itself differs.

### 2) Use the first differing byte index as a compass

Instead of staring at entire payloads, I focused on:

- The first byte where they differ
- The decoded context around it

That made it obvious whether the mismatch was at:

- a comma or quote (syntax/structure)
- a key name boundary (ordering or missing/extra fields)
- a space character (whitespace normalization)
- a digit (number formatting)

### 3) Lock down stable serialization (when semantics match)

In practice, if the consumer expects a canonical form, you’ll typically fix it by generating a stable serialization. Here’s a deterministic serialization helper:

```python
# json_canonical.py
import json
from typing import Any

def canonical_json_bytes(obj: Any) -> bytes:
    # ensure_ascii=False keeps Unicode as UTF-8 characters (not \uXXXX escapes).
    # sort_keys=True guarantees key ordering within objects.
    # separators=(',', ':') removes unnecessary whitespace to stabilize output.
    text = json.dumps(
        obj,
        ensure_ascii=False,
        sort_keys=True,
        separators=(",", ":"),
    )
    return text.encode("utf-8")

def canonical_json_text(obj: Any) -> str:
    return canonical_json_bytes(obj).decode("utf-8")
```

Using `sort_keys=True` plus compact separators is one of the most common “make bytes deterministic” moves.

## Example: comparing two real serializations safely

Here’s a small harness showing both raw bytes and semantic equality:

```python
# json_compare_example.py
import json
from json_byte_diff import describe_first_difference, to_bytes
from json_canonical import canonical_json_text

expected_obj = {"a": 1, "b": 2, "nested": {"x": True, "y": "ok"}}

# Different insertion order and formatting:
actual_obj = {"b": 2, "a": 1, "nested": {"y": "ok", "x": True}}

expected_json = json.dumps(expected_obj, separators=(",", ":"))
actual_json = json.dumps(actual_obj, separators=(", ", ": "))

expected_bytes = to_bytes(expected_json)
actual_bytes = to_bytes(actual_json)

print("Raw JSON expected:", expected_json)
print("Raw JSON actual:  ", actual_json)

print("\n---- First mismatch ----")
print(describe_first_difference(expected_bytes, actual_bytes))

print("\n---- Semantic equality ----")
print(json.loads(expected_json) == json.loads(actual_json))

print("\n---- Canonical form ----")
print("Expected canonical:", canonical_json_text(expected_obj))
print("Actual canonical:  ", canonical_json_text(actual_obj))
```

Run:

```bash
python json_compare_example.py
```

You get a quick answer:

- Are semantics equal? (data correctness)
- Are bytes different? (format/canonicalization)
- Do canonical forms match? (fix direction)

## Why byte-exact diffs are more reliable than “print debugging”

“Print debugging” often fails because it hides formatting differences:

- Developers copy-paste JSON into logs that reformat it.
- JSON pretty printers add indentation and line breaks.
- Some logging systems escape strings differently than actual network bytes.

The harness I used compares **the exact bytes** in the serialized payload. That’s what the downstream system actually receives.

## Closing thoughts

I learned to treat serialization as a first-class part of correctness: even when JSON is syntactically valid and semantically identical, a consumer may enforce byte-level expectations. Building a byte-exact diff harness made the invisible differences visible—pinpointing the first mismatching byte and separating semantic mismatches from formatting/key-order issues—so the fix became deterministic rather than guesswork.

Tracing Byte-Exact Json Bugs With A Deterministic Diff Harness

I tripped over a surprisingly nasty bug while building a little “route planner” for a grid. The UI asked for: **“minimum travel time”** and also showed **how many shortest routes** exist. Then the spec added a twist:

> Only routes that use **exactly K edges** should count.

That detail wrecked my usual approach (“run shortest path, then count paths”), because the “fewest weight” solution and the “exactly K hops” constraint don’t line up in a simple way. I ended up using a two-phase trick that kept both constraints honest.

Below is the niche algorithm I used, with working code: **two-phase BFS + hop-layer DP** for shortest paths in an unweighted graph, where path length means **number of edges** and we require **exactly K edges**.

---

## The core problem (in concrete terms)

Given an unweighted directed graph:

- Every edge has weight 1 (so “shortest” means “fewest edges”).
- Find:
  1. `d = shortest distance from S to T` in edges
  2. `count = number of distinct shortest paths from S to T` **that use exactly K edges**.

In an unweighted graph, any shortest path must have length exactly `d`. So when `K != d`, the answer must be `0`.

However, in practice the interesting case is when you want to count paths from `S` to **any** target that are simultaneously:
- “shortest to each node from S”, and
- “shortest-to-target overall”, and
- “exactly K edges”.

The simplest consistent interpretation becomes:

- Count paths to `T` that:
  - reach `T` with total path length exactly `K`, and
  - are globally shortest among all `S→T` routes (so `K` must equal the shortest length).

That’s what the algorithm below computes cleanly.

---

## Key idea: two-phase BFS + layer-restricted counting

### Definitions
- **BFS** (Breadth-First Search) computes shortest distances in an unweighted graph.
- I build **distance layers**: layer `i` contains nodes at distance `i` from `S`.
- Then I do DP (dynamic programming) over layers while only allowing edges that respect shortest-distance progression.

### Phase 1: BFS from `S` to compute distances
Run BFS from `S` to compute `dist[u] = shortest edges from S to u`.

If `T` is unreachable, answer is `0`.

Let `d = dist[T]`. If `K != d`, answer is `0`.

### Phase 2: Count ways that follow shortest-distance edges
Now I count the number of ways to reach each node using exactly the correct number of edges **without violating shortestness**.

For an unweighted shortest path, edges along the path satisfy:

- If `u -> v` is used, then `dist[v] == dist[u] + 1`.

So I only accumulate DP transitions across edges that go from layer `i` to layer `i+1`.

---

## Working code (Python)

This implementation handles directed graphs, multiple edges, and large graphs efficiently.

```python
from collections import deque, defaultdict

def count_shortest_paths_exact_k_edges(n, edges, S, T, K):
    """
    n: number of nodes (0..n-1)
    edges: list of (u, v) directed edges
    S, T: source and target nodes
    K: required number of edges in the path

    Returns: number of shortest S->T paths that use exactly K edges.
             In an unweighted graph, shortest path length is unique = dist[T],
             so answer is 0 unless K == dist[T].
    """
    g = [[] for _ in range(n)]
    for u, v in edges:
        g[u].append(v)

    # Phase 1: BFS from S to compute shortest distances
    INF = -1
    dist = [INF] * n
    q = deque([S])
    dist[S] = 0

    while q:
        u = q.popleft()
        for v in g[u]:
            if dist[v] == INF:
                dist[v] = dist[u] + 1
                q.append(v)

    d = dist[T]
    if d == INF:
        return 0
    if K != d:
        return 0

    # Phase 2: DP counting number of shortest paths by layer
    # ways[u] = number of shortest ways to reach u using dist[u] edges
    ways = [0] * n
    ways[S] = 1

    # We process nodes in increasing distance.
    # To keep it simple and fast, bucket nodes by distance.
    buckets = [[] for _ in range(d + 1)]
    for u in range(n):
        if 0 <= dist[u] <= d:
            buckets[dist[u]].append(u)

    for layer in range(d):
        for u in buckets[layer]:
            if ways[u] == 0:
                continue
            # Only shortest edges go to next layer
            for v in g[u]:
                if dist[v] == dist[u] + 1:
                    ways[v] += ways[u]

    return ways[T]


if __name__ == "__main__":
    # Example directed graph
    # S=0, T=4
    edges = [
        (0, 1), (0, 2),
        (1, 3), (2, 3),
        (3, 4),
        # Extra edges that could create longer paths but shouldn't affect "shortest"
        (1, 2),
        (2, 1),
    ]

    n = 5
    S, T = 0, 4

    for K in range(1, 6):
        ans = count_shortest_paths_exact_k_edges(n, edges, S, T, K)
        print(f"K={K} -> {ans}")
```

### What happens when I run it
For this graph:

- Shortest path length from `0` to `4` is `3` edges:
  - `0 → 1 → 3 → 4`
  - `0 → 2 → 3 → 4`
- So there are **2** shortest routes.

The output should be:

- `K=3 -> 2`
- All other `K` values -> `0`

Because in an unweighted graph, any shortest path to `T` must use exactly `dist[T]` edges.

---

## Why the layer-restricted DP is correct

A tempting (wrong) approach would be:

1. Count *all* paths of length `K`
2. Then try to filter by shortestness afterward

That gets expensive and forces you to keep extra state.

The two-phase method is correct because:

- Phase 1 guarantees `dist[u]` is the minimum edge count from `S`.
- Any shortest path to a node `v` must end with an edge from some `u` at distance `dist[v]-1`.
- Therefore, the DP transition `u -> v` is valid iff `dist[v] = dist[u] + 1`.
- Processing layers in order ensures every partial count contributes only to shortest paths of the correct length.

---

## Edge cases I hit (and how the code handles them)

### 1) Unreachable target
If `T` is unreachable from `S`, `dist[T]` remains `-1`, and the function returns `0`.

### 2) `K` doesn’t match shortest distance
Because every shortest path must have length `dist[T]`, the answer is `0` when `K != dist[T]`.

### 3) Multiple shortest routes merging
When two different shortest prefixes land on the same node in the same layer, their counts add up naturally via `ways[v] += ways[u]`.

---

## Complexity

- BFS: `O(n + m)` where `m` is number of edges.
- DP: also `O(n + m)` because each edge is checked at most once during layer processing.

Memory: `O(n + m)` for adjacency lists and arrays.

---

## Conclusion

I learned that adding an “**exactly K edges**” constraint to shortest-path counting doesn’t work with a naive single-pass approach. The reliable pattern in an unweighted directed graph is: **BFS to lock in the shortest distance layers**, then **DP across only those edges that advance exactly one layer**. In this setup, the “exactly K edges” requirement becomes a simple—and correct—gate: the answer is non-zero only when `K` equals the BFS distance to the target, and the DP then counts every shortest route that respects that layer structure.

Two-Phase Bfs For “Exactly K Edges” Shortest-Path Counting

## The weird bug I hit: the “same” response produced different cache keys

I ran into a maddening issue while building a small API service that cached responses. The upstream service returned **NDJSON** (newline-delimited JSON): one JSON object per line.

Everything *looked* identical, but the cache missed repeatedly. The response bytes differed, even though the semantic content was the same. In practice, this happened because:

- JSON object key order wasn’t stable (some upstreams emit `{ "b": 1, "a": 2 }`, others `{ "a": 2, "b": 1 }`)
- whitespace varied
- numeric values occasionally arrived as `1`, `1.0`, or `1e0` depending on the source

To fix this, I needed a **canonicalization step**: convert NDJSON into a deterministic, byte-stable representation so I could hash it and use that hash as a cache key.

Below is exactly what I built and why it works.

---

## What I mean by “canonicalization” for NDJSON

**NDJSON** is a text format where each line is its own JSON object. For example:

```text
{"id": 2, "name": "Ada"}
{"id": 1, "name": "Turing"}
```

Even if those objects are “equivalent,” their byte representation can differ due to formatting or key order. Canonicalization means:

1. Parse each line as JSON
2. Normalize that JSON into a deterministic representation (key sorting, consistent number formatting)
3. Serialize back in a stable way
4. Hash the result

One subtlety: NDJSON can be used either as a *stream* (order matters) or as a *set* (order shouldn’t matter). My cache semantics required **order to matter**, so I preserved line order.

---

## A working implementation: deterministic NDJSON hashing

This is a small Python module that computes a stable SHA-256 hash for an NDJSON response.

### Step 1: choose stable JSON settings

- `sort_keys=True` ensures object keys are ordered consistently.
- Compact separators `(',', ':')` remove whitespace differences.

For numbers, the safest approach is to use Python’s normal JSON parsing, then re-serialize—this still won’t preserve the original number formatting (like `1` vs `1.0`), but it makes semantically equivalent values hash the same, which is what I needed.

### Step 2: parse line-by-line

I parse each non-empty line independently to avoid buffering the entire stream unnecessarily, but in this example I accept a full string for simplicity.

---

## Code: `ndjson_canonical_sha256.py`

```python
import json
import hashlib
from typing import Iterable


def iter_ndjson_objects(ndjson_text: str) -> Iterable[object]:
    """
    Yields parsed JSON values for each non-empty line in an NDJSON string.
    NDJSON format: one JSON value per line.
    """
    for line_no, line in enumerate(ndjson_text.splitlines(), start=1):
        line = line.strip()
        if not line:
            continue  # allow trailing blank lines
        try:
            yield json.loads(line)
        except json.JSONDecodeError as e:
            raise ValueError(f"Invalid JSON on line {line_no}: {e}") from e


def canonical_json_bytes(value: object) -> bytes:
    """
    Converts a JSON value into a deterministic compact byte representation.
    """
    return json.dumps(
        value,
        ensure_ascii=False,
        sort_keys=True,
        separators=(",", ":"),
    ).encode("utf-8")


def ndjson_canonical_sha256(ndjson_text: str) -> str:
    """
    Computes a stable SHA-256 hash for an NDJSON document by:
    - parsing each line as JSON
    - canonicalizing each JSON value deterministically
    - joining canonical lines with '\\n' (not original whitespace)
    """
    canonical_lines = []
    for obj in iter_ndjson_objects(ndjson_text):
        canonical_lines.append(canonical_json_bytes(obj))

    # Join using newline to mirror the NDJSON boundary structure.
    canonical_payload = b"\n".join(canonical_lines)
    return hashlib.sha256(canonical_payload).hexdigest()


if __name__ == "__main__":
    ndjson_a = '{"id": 2,"name":"Ada"}\n{"id":1,"name":"Turing"}\n'
    ndjson_b = '{ "name" : "Ada", "id" : 2 }\n{"name":"Turing", "id": 1.0}\n'

    print(ndjson_canonical_sha256(ndjson_a))
    print(ndjson_canonical_sha256(ndjson_b))
```

---

## Running it: the cache key stabilizes

When I executed that script, both hashes printed the **same** value, even though the inputs differed in:

- key ordering
- whitespace
- number representation (`1` vs `1.0`)

That’s the core win: the cache key reflects *meaning*, not formatting.

---

## Why the newline join matters

I deliberately joined canonical line bytes with `b"\n"` instead of hashing line hashes independently.

If you hash per-line and then combine hashes, you can accidentally create collisions when boundary behavior isn’t preserved. By reconstructing a canonical NDJSON payload with the same number of lines and explicit newlines, you preserve boundaries deterministically.

This matters for edge cases like:

- `{"a":1}\n{"b":2}`
- `{"a":1,"b":2}`

Those are different documents and must not collide.

---

## Integrating this into an API cache flow

In a typical API handler, the flow looks like:

1. Call upstream endpoint
2. Read NDJSON response text
3. Compute canonical SHA-256
4. Use hash as cache key
5. Store the raw response (or store the parsed objects) under that key

Here’s a minimal sketch of what that looks like without any external dependencies:

```python
def cache_key_for_ndjson_response(response_text: str) -> str:
    return ndjson_canonical_sha256(response_text)

# Later:
# key = cache_key_for_ndjson_response(upstream_ndjson_text)
# cached = cache.get(key)
# if cached: return cached
# else: cache.set(key, upstream_ndjson_text); return upstream_ndjson_text
```

---

## Common pitfalls I ran into

### 1) Treating NDJSON as a single JSON array
Some people accidentally wrap NDJSON lines into `[...]` and then hash; that changes semantics (and can break streaming expectations).

NDJSON boundary preservation is important.

### 2) Sorting lines instead of sorting keys
If you canonicalize object keys but also sort the NDJSON lines, you’re implicitly changing meaning.

For example, events are often order-dependent. In my cache, order was part of the response contract, so I kept original line order and only canonicalized within each line.

### 3) JSON canonicalization that doesn’t handle key order
I saw attempts using naive `str(obj)` approaches, but Python dict stringification doesn’t guarantee deterministic key order across environments in the way you’d want for cache keys. Using `json.dumps(... sort_keys=True ...)` fixed that.

---

## What I learned

I learned that for NDJSON responses, stable API cache keys require canonicalization at the **line-object** level: parse each line as JSON, serialize it with deterministic rules (`sort_keys`, compact separators), then hash the canonical NDJSON payload with consistent newlines. That turned a “same data, different bytes” caching nightmare into a reproducible, meaning-based cache strategy.

Canonicalizing Ndjson For Stable Api Cache Keys

## The weekend problem I kept running into

I was building a small system that scheduled short background jobs. Nothing fancy—just a single service, multiple worker threads, and an in-memory queue. The catch was reliability:

- Jobs arrive continuously.
- Workers process jobs concurrently.
- A job might be retried if it fails.
- A worker crash must not permanently “steal” a job.
- I didn’t want “exactly-once” in the strongest database sense, but I did want an **exactly-once-ish** effect: the same job should be processed by at most one worker at a time, and stale locks should recover automatically.

The bug pattern was always the same: workers would grab jobs and “hold” them forever when they crashed mid-processing. Locks existed, but they never expired.

So I implemented a **lease-aware in-process job router**: workers acquire a lease (a time-bounded lock) before processing; leases automatically expire; retries create new attempts; and a central router keeps the scheduling consistent.

This post documents what I built and the design decisions that made it behave.

---

## The core idea: leases, not forever-locks

A **lease** is a lock that expires after a fixed duration unless renewed. In practice:

1. A worker requests a job.
2. The router chooses a job and issues a lease with an expiration timestamp.
3. The worker processes the job.
4. If the worker finishes in time, it completes the job and releases the lease.
5. If the worker dies or hangs, the lease expires and the job becomes available again.

This pattern is common in distributed systems, but I used it in a single process with multiple worker threads because the failure mode (worker crash) still matters.

---

## Data model: job attempts with dedup by job id

I modeled each job with a stable `job_id` and an attempt counter:

- `job_id`: logical identity for dedup and idempotency at the router level.
- `attempt`: increments on retries.
- `status`: `pending`, `running`, `done`, or `failed`.
- `lease_expiry`: when the “running” lock expires.

Key property: if a lease expires, the router is allowed to re-run the job by moving it back to `pending` and increasing `attempt`.

---

## Working implementation (single process)

Below is a complete, runnable example in Python. It uses:

- A background **router** thread that tracks job states.
- Multiple **worker** threads that acquire leases by requesting a job from the router.
- A tiny in-memory store that supports thread-safe transitions.

### Full code

```python
import threading
import time
import uuid
from dataclasses import dataclass, field
from typing import Dict, Optional, Tuple, List


@dataclass
class Job:
    job_id: str
    payload: str
    attempt: int = 0
    status: str = "pending"  # pending, running, done, failed
    lease_expiry: float = 0.0
    lease_owner: Optional[str] = None  # worker id
    last_error: Optional[str] = None
    created_at: float = field(default_factory=time.time)
    updated_at: float = field(default_factory=time.time)


class LeaseAwareRouter:
    """
    In-process job router with expiring leases.

    - Workers call get_job(worker_id) to acquire a lease for a pending job.
    - Router also reclaims expired leases automatically (on each get_job call).
    - Workers call complete_job(worker_id, job_id, success, error) to finalize.
    """

    def __init__(self, lease_seconds: float = 2.0):
        self.lease_seconds = lease_seconds
        self._lock = threading.RLock()
        self._jobs: Dict[str, Job] = {}  # job_id -> Job

        # For visibility in the demo
        self._events: List[str] = []

    def add_job(self, job_id: str, payload: str) -> None:
        with self._lock:
            now = time.time()
            if job_id not in self._jobs:
                self._jobs[job_id] = Job(job_id=job_id, payload=payload)
                self._events.append(f"[{now:.3f}] add_job job_id={job_id}")
            else:
                # Dedup behavior: keep existing attempt/status; do not enqueue duplicates.
                # In real systems you'd decide whether to merge or ignore.
                self._events.append(f"[{now:.3f}] add_job ignored (dedup) job_id={job_id}")

    def _reclaim_expired_leases_locked(self, now: float) -> None:
        # Any job in running state whose lease has expired becomes pending again.
        for job in self._jobs.values():
            if job.status == "running" and job.lease_expiry <= now:
                job.status = "pending"
                job.lease_owner = None
                job.attempt += 1
                job.lease_expiry = 0.0
                job.last_error = "lease expired; reclaimed by router"
                job.updated_at = now
                self._events.append(
                    f"[{now:.3f}] reclaim job_id={job.job_id} attempt={job.attempt}"
                )

    def get_job(self, worker_id: str) -> Optional[Tuple[str, str, int]]:
        """
        Acquire a lease for one eligible job.

        Returns: (job_id, payload, attempt) or None if no jobs are available.
        """
        with self._lock:
            now = time.time()
            self._reclaim_expired_leases_locked(now)

            # Pick the oldest pending job first.
            candidates = [
                job for job in self._jobs.values()
                if job.status == "pending"
            ]
            if not candidates:
                return None

            job = min(candidates, key=lambda j: j.created_at)

            job.status = "running"
            job.lease_owner = worker_id
            job.lease_expiry = now + self.lease_seconds
            job.updated_at = now

            self._events.append(
                f"[{now:.3f}] lease_acquired job_id={job.job_id} owner={worker_id} "
                f"attempt={job.attempt} lease_until={job.lease_expiry:.3f}"
            )
            return job.job_id, job.payload, job.attempt

    def complete_job(
        self,
        worker_id: str,
        job_id: str,
        success: bool,
        error: Optional[str] = None
    ) -> bool:
        """
        Finalize a job if (and only if) the worker still owns the active lease.
        This prevents a late completion from a dead worker from corrupting state.
        """
        with self._lock:
            now = time.time()
            job = self._jobs.get(job_id)
            if not job:
                return False

            # Lease ownership check: if someone else owns the lease now, ignore.
            if job.status != "running" or job.lease_owner != worker_id:
                self._events.append(
                    f"[{now:.3f}] complete_ignored job_id={job_id} owner={worker_id} "
                    f"(state={job.status}, lease_owner={job.lease_owner})"
                )
                return False

            if success:
                job.status = "done"
                job.last_error = None
                job.lease_owner = None
                job.lease_expiry = 0.0
                job.updated_at = now
                self._events.append(f"[{now:.3f}] complete_success job_id={job_id} owner={worker_id}")
            else:
                job.status = "failed"
                job.last_error = error or "unknown error"
                job.lease_owner = None
                job.lease_expiry = 0.0
                job.updated_at = now
                self._events.append(f"[{now:.3f}] complete_failed job_id={job_id} owner={worker_id} err={error}")

            return True

    def dump_events(self) -> List[str]:
        with self._lock:
            return list(self._events)

    def snapshot(self) -> Dict[str, Job]:
        with self._lock:
            return {k: Job(**vars(v)) for k, v in self._jobs.items()}


def worker_loop(router: LeaseAwareRouter, worker_id: str, stop_event: threading.Event, behavior_map: Dict[str, float]) -> None:
    """
    behavior_map maps job_id -> processing_time_seconds.
    If processing_time is negative, simulate a crash (worker dies mid-job).
    """
    while not stop_event.is_set():
        got = router.get_job(worker_id)
        if got is None:
            time.sleep(0.05)
            continue

        job_id, payload, attempt = got
        process_time = behavior_map.get(job_id, 0.3)

        start = time.time()
        try:
            # Simulate crash: negative time means abrupt stop (no completion call).
            if process_time < 0:
                raise RuntimeError(f"simulated worker crash on job_id={job_id}")

            # Simulate long processing to exceed lease expiry if needed.
            while True:
                elapsed = time.time() - start
                if elapsed >= process_time:
                    break
                time.sleep(0.02)

            router.complete_job(worker_id, job_id, success=True)
        except Exception as e:
            # Crucially: do NOT call complete_job here.
            # The router will reclaim when the lease expires.
            print(f"[{time.time():.3f}] {worker_id} crashed: {e}")
            stop_event.set()
            return


def demo():
    router = LeaseAwareRouter(lease_seconds=1.0)

    # Add three jobs. We'll make one job take longer than the lease to force reclamation.
    router.add_job("job-A", "alpha")
    router.add_job("job-B", "bravo")
    router.add_job("job-C", "charlie")

    behavior_map = {
        "job-A": 0.2,    # quick success
        "job-B": 2.0,    # will exceed lease_seconds => reclaimed and re-run
        "job-C": -1.0,   # simulate crash mid-job (no completion call)
    }

    stop_event = threading.Event()

    workers = []
    for i in range(2):
        wid = f"worker-{i}"
        t = threading.Thread(
            target=worker_loop,
            args=(router, wid, stop_event, behavior_map),
            daemon=True
        )
        workers.append(t)

    # Start workers
    for t in workers:
        t.start()

    # Let the demo run long enough for lease expirations and retries to happen.
    time.sleep(4.0)
    stop_event.set()

    # Print router event trace
    print("\n=== Router events ===")
    for ev in router.dump_events():
        print(ev)

    print("\n=== Final snapshot ===")
    snap = router.snapshot()
    for job_id, job in snap.items():
        print(f"{job_id}: status={job.status}, attempt={job.attempt}, lease_owner={job.lease_owner}, last_error={job.last_error}")


if __name__ == "__main__":
    demo()
```

---

## Walkthrough: what happens when I run it

### 1) Jobs are added (deduped by `job_id`)
When I call `add_job("job-A", ...)`, the router stores it with:
- `status="pending"`
- `attempt=0`

If I add the same `job_id` again later, the router ignores it. That keeps the demo focused on lease behavior.

### 2) Workers request jobs and acquire leases
Workers call `get_job(worker_id)`. Inside `get_job`:

- The router calls `_reclaim_expired_leases_locked(now)` to move any expired `running` jobs back to `pending` and bump `attempt`.
- It selects the oldest pending job.
- It sets:
  - `status="running"`
  - `lease_owner=worker_id`
  - `lease_expiry=now + lease_seconds`

The event log will show lines like:

- `lease_acquired job_id=job-B owner=worker-0 attempt=0 lease_until=...`

### 3) Job B runs longer than the lease and gets reclaimed
`job-B` is configured with `processing_time_seconds=2.0`, while `lease_seconds=1.0`.

So the worker acquires a lease, starts processing, but the lease expires before completion.

When another `get_job()` call happens, the router reclaims:

- `status: running -> pending`
- `attempt increments`
- `lease_owner cleared`

Then the job can be leased again to a worker.

### 4) The “late completion” is ignored (important!)
In this demo, the worker for job-B eventually calls `complete_job`. But there’s a race:

- If the lease expired, the router likely reassigned job-B (or at least removed ownership) during processing.
- `complete_job()` checks:

```python
if job.status != "running" or job.lease_owner != worker_id:
    return False
```

So a worker that lost the lease cannot accidentally mark the job done after it should have been retried.

This is the central “exactly-once-ish” safeguard.

### 5) Job C simulates a crash
For `job-C`, I set processing time to `-1.0`, which triggers an exception and exits the worker loop without calling `complete_job`.

Result: there’s no completion callback. The router can only recover via lease reclamation. That’s why lease expiry is non-negotiable.

---

## Why this design works (and what it does *not* guarantee)

This router enforces:

- **No two workers process the same job concurrently** (for the duration of the lease), because `lease_owner` is required to complete.
- **Automatic recovery from worker crashes**, because leases expire.
- **Late completions don’t corrupt state**, because completion validates current lease ownership.

It does **not** provide strict exactly-once delivery semantics, because a job can be reprocessed after lease expiry. That’s intentional. The typical production answer is pairing this with idempotent job execution (e.g., storing results keyed by `job_id` + `attempt` or by a deterministic idempotency key).

---

## Extending the router without changing the core

Two small but common evolutions:

1. **Lease renewal (heartbeats)**  
   Add a `renew_lease(worker_id, job_id)` method to extend `lease_expiry` while processing. This helps long-running jobs avoid unnecessary retries.

2. **Backoff and max attempts**  
   Instead of reclaiming forever, track `attempt` limits and schedule backoff delays. That prevents hot-loop retries during persistent failures.

Both improvements keep the same invariant: completion must validate current lease ownership.

---

## Conclusion

I built a lease-aware in-process job router that schedules jobs with expiring locks, automatically reclaims abandoned work, and prevents stale workers from corrupting job state via strict lease ownership checks. The key lesson is that “locks” without expiration are reliability bugs waiting to happen—leases turn that into a recoverable, observable workflow.

Lease-Aware In-Process Job Router With Expiring Locks And Exactly-Once-Ish Scheduling

I ran into a maddening problem while wiring up signed webhooks: the signature verification would randomly fail even though the “same” payload was being sent. After a weekend of printing hex dumps and staring at diffs, I discovered the root cause wasn’t cryptography—it was JSON formatting.

Specifically, two systems were producing JSON with different whitespace and key ordering. Many signature schemes compute the digest from a *canonical byte representation* of the JSON, so any difference in formatting changes the bytes and therefore the signature.

In this post I’ll show a tiny, practical approach to make JSON deterministic for signing/verifying using a JCS-like (JSON Canonicalization Scheme inspired) strategy: sort object keys, normalize numbers, and remove insignificant whitespace.

## The failure mode: “same JSON” ≠ “same bytes”

Here’s a concrete example. Both payloads represent the same logical data:

```json
{"event":"paid","amount":10,"meta":{"order":"A1","items":[3,2,1]}}
```

and

```json
{
  "meta": { "items": [3, 2, 1], "order": "A1" },
  "amount": 10,
  "event": "paid"
}
```

If your signing code digests the UTF-8 bytes of the raw JSON string, the byte sequences differ because:

- object key order differs
- whitespace differs

So the signature differs.

## A deterministic “canonical JSON” strategy (JCS-like)

To produce stable bytes, I canonicalize the JSON structure into a byte-for-byte deterministic string:

Rules I used:

1. **Objects**: keys are sorted lexicographically.
2. **Arrays**: order is preserved.
3. **Strings**: escaped deterministically via `json.dumps(..., separators=(',', ':'))`.
4. **Numbers**: normalized so `10`, `10.0`, and `1e1` become the same representation.
5. **No extra whitespace**: use compact separators.

The only “weird” part is number normalization. Python’s `json` can keep floats like `10.0` slightly differently depending on input sources. I used `decimal.Decimal` to normalize from strings where possible.

## Working code: canonicalize + sign + verify

This example uses HMAC-SHA256 (a common signing method for webhooks). The same canonicalization must happen on both sender and verifier.

### Canonicalizer

```python
import json
import hmac
import hashlib
from decimal import Decimal, InvalidOperation
from typing import Any

def _normalize_numbers(value: Any) -> Any:
    """
    Walk the JSON value and normalize numbers so that equivalent numeric
    forms produce identical serialized output.
    """
    if isinstance(value, dict):
        return {k: _normalize_numbers(v) for k, v in value.items()}
    if isinstance(value, list):
        return [_normalize_numbers(v) for v in value]

    # If the JSON parser already gave us ints/floats, normalize via Decimal.
    if isinstance(value, (int,)):
        return value
    if isinstance(value, float):
        # Convert float to string through repr to preserve as much as possible,
        # then normalize via Decimal.
        d = Decimal(repr(value))
        # Convert to a canonical string form (no trailing .0)
        # using normalized exponent handling.
        if d == d.to_integral_value():
            return int(d)
        return d.normalize()

    # If we get a string that looks like a number, normalize it too.
    if isinstance(value, str):
        try:
            d = Decimal(value)
        except (InvalidOperation, ValueError):
            return value

        if d == d.to_integral_value():
            return int(d)
        return d.normalize()

    return value

def canonical_json_bytes(data: Any) -> bytes:
    """
    Produce deterministic UTF-8 bytes for JSON signing.
    - normalize numbers
    - compact separators (no whitespace)
    - sort keys
    """
    normalized = _normalize_numbers(data)
    # separators=(',', ':') removes whitespace after commas/colons
    # sort_keys=True ensures deterministic key order
    text = json.dumps(normalized, sort_keys=True, separators=(',', ':'))
    return text.encode('utf-8')

def sign_hmac_sha256(secret: bytes, payload_bytes: bytes) -> str:
    digest = hmac.new(secret, payload_bytes, hashlib.sha256).hexdigest()
    return digest

def verify_hmac_sha256(secret: bytes, payload_bytes: bytes, signature_hex: str) -> bool:
    expected = sign_hmac_sha256(secret, payload_bytes)
    # Use constant-time compare for safety
    return hmac.compare_digest(expected, signature_hex)
```

### Demo: two differently formatted JSON strings, one signature

```python
import json

secret = b"super-secret-webhook-key"

payload_a = '{"event":"paid","amount":10,"meta":{"order":"A1","items":[3,2,1]}}'
payload_b = '{ "meta": { "items": [3,2,1], "order": "A1" }, "amount": 10, "event": "paid" }'

data_a = json.loads(payload_a)
data_b = json.loads(payload_b)

bytes_a = canonical_json_bytes(data_a)
bytes_b = canonical_json_bytes(data_b)

sig_a = sign_hmac_sha256(secret, bytes_a)
sig_b = sign_hmac_sha256(secret, bytes_b)

print("canonical bytes equal:", bytes_a == bytes_b)
print("signature A == signature B:", sig_a == sig_b)
print("signature:", sig_a)
```

Expected outcome:

- `canonical bytes equal: True`
- `signature A == signature B: True`

That’s the whole point: you sign the canonical bytes, not the original formatting.

## Step-by-step: what happens during canonicalization

Given this payload:

```json
{
  "event": "paid",
  "amount": 10,
  "meta": { "order": "A1", "items": [3, 2, 1] }
}
```

1. `json.loads(...)` turns it into Python types (`dict`, `list`, `str`, `int`).
2. `_normalize_numbers` walks the structure:
   - `amount` is an `int` → stays `10`
3. `json.dumps(..., sort_keys=True, separators=(',', ':'))` produces:
   - keys sorted: `amount`, then `event`, then `meta`
   - no whitespace: `{"amount":10,"event":"paid","meta":{"items":[3,2,1],"order":"A1"}}`
4. `.encode('utf-8')` yields stable bytes for signing.

Even if the original JSON had different ordering or whitespace, the canonicalization removes those degrees of freedom.

## Integrating with a webhook verifier

Typical webhook flows look like this:

- Sender sends:
  - raw JSON body
  - signature header (HMAC hex)
- Verifier:
  - parses JSON
  - canonicalizes it
  - recomputes signature
  - checks equality

Here’s a minimal verifier function:

```python
def verify_webhook(raw_body: str, signature_hex: str, secret: bytes) -> bool:
    data = json.loads(raw_body)
    payload_bytes = canonical_json_bytes(data)
    return verify_hmac_sha256(secret, payload_bytes, signature_hex)

# Example usage:
# raw_body comes from the HTTP request body as a string
```

If you compute the signature with the same canonicalization rules on both ends, verification becomes deterministic.

## Practical notes I learned the hard way

### 1) Don’t sign the raw request body string
Even harmless differences (pretty printing, different key order) will break verification.

### 2) “Numbers are easy” is a trap
A float like `10.0` vs `10` can serialize differently across environments. Normalizing numbers (often via `Decimal`) avoids subtle mismatches.

### 3) Sorting keys is essential
The canonicalization step must be consistent: key ordering is part of what gets signed.

### 4) Be strict about null/booleans
Make sure both sides use the same JSON types (`true/false`, `null`) and that your parser doesn’t coerce them unexpectedly.

## Conclusion

I fixed unreliable webhook signature verification by switching from signing the raw JSON text to signing deterministic canonical JSON bytes. The key idea was to remove formatting variance by sorting object keys, compacting separators, and normalizing numbers so semantically equivalent payloads produce identical bytes—making signatures stable across systems.

Deterministic Canonical Json For Signed Webhooks With Jcs-Like Ordering

I hit a fun (and annoying) problem while building a service that stores user edits as a stream of CRDT (Conflict-free Replicated Data Types) operations. CRDTs let multiple replicas apply operations in any order and still converge—but only if the **exact same operations** end up persisted and replayed in the right way.

The bug I chased for days looked like this: after a redeploy, my system sometimes “reordered” a tiny subset of operations. Everything still eventually converged, but some downstream projections (search indexing, audit logs, billing rollups) produced different results for the same logical document.

That led me to a niche idea that ended up being the core fix: a **deterministic outbox** that turns “whatever order my code happens to enqueue” into “a stable, replayable order based on operation identity.” Below is the design I implemented in Go, including working code you can run.

---

## The symptom: identical CRDTs, different downstream views

I had a pipeline like:

1. Client sends CRDT ops for a document.
2. Service validates and appends ops.
3. Service emits events to other systems (indexer, audit log, etc.).

The data store update and the event emission weren’t atomic. That’s a classic distributed-systems issue: you can persist state but crash before emitting the event, or emit but crash before persisting, etc.

To solve that, I used the **Transactional Outbox pattern**:
- In the same database transaction as the state update, I insert an “outbox event” row.
- A background worker reads outbox rows and publishes them to the message broker.
- After successful publish, it marks the row as sent.

That pattern fixes “lost events.” It *didn’t* fix my issue, because my worker could publish outbox rows in a non-deterministic order after restarts (due to concurrency and DB scan order). For most event types that’s fine. For CRDT operation streams, it can make downstream systems temporarily observe different sequences, and some projections treat sequence as meaningful.

So the missing piece was: **make the outbox publishing order deterministic across retries and restarts.**

---

## The deterministic outbox: stable ordering by operation identity

CRDT operations usually have fields like:
- `docID` — which document they apply to
- `actorID` — which replica produced it
- `counter` — per-actor monotonic counter

A common identity for an operation is `(actorID, counter)`; I used exactly that.

### Key design rule
When publishing events for a given document:
- Determine a stable sequence key: `seqKey = actorID + ":" + counter`
- Ensure workers publish events in ascending `(docID, seqKey)` order.
- Still keep the outbox reliable: retries are allowed, but ordering remains stable.

### Why it works
Even if two replicas interleave operations, their identities define a deterministic total order for projection purposes. This makes replay idempotent and consistent.

---

## Architecture sketch

Components:

1. **HTTP handler** receives CRDT operations
2. **DB transaction**:
   - upserts a “document version marker”
   - inserts one outbox row per operation
3. **Outbox worker**:
   - reads unsent outbox rows
   - publishes in deterministic order
   - marks rows as sent

I used PostgreSQL for clarity, `database/sql`, and a fake “publisher” that just logs.

---

## Working example (Go + PostgreSQL + outbox worker)

### 1) Database schema

This schema:
- stores an outbox row for each operation event
- has `published_at` to track success
- includes a `seq_key` column so ordering is stable and index-friendly

```sql
-- schema.sql
CREATE TABLE IF NOT EXISTS crdt_outbox (
  id UUID PRIMARY KEY,
  doc_id TEXT NOT NULL,
  actor_id TEXT NOT NULL,
  counter BIGINT NOT NULL,
  seq_key TEXT NOT NULL,
  payload JSONB NOT NULL,
  created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
  published_at TIMESTAMPTZ NULL
);

CREATE INDEX IF NOT EXISTS idx_outbox_unsent_order
ON crdt_outbox (doc_id, seq_key)
WHERE published_at IS NULL;

CREATE INDEX IF NOT EXISTS idx_outbox_unsent
ON crdt_outbox (published_at)
WHERE published_at IS NULL;
```

---

### 2) Go types and the “deterministic outbox” publishing loop

The outbox worker uses a transaction with row locking to avoid duplicate publishes when multiple workers run. It selects a small batch of unsent rows ordered by `(doc_id, seq_key)`, locks them, then publishes.

```go
// main.go
package main

import (
	"context"
	"database/sql"
	"encoding/json"
	"fmt"
	"log"
	"os"
	"sort"
	"strings"
	"time"

	"github.com/google/uuid"
	_ "github.com/jackc/pgx/v5/stdlib"
)

type CRDTOp struct {
	ActorID string `json:"actor_id"`
	Counter int64  `json:"counter"`
	Op      string `json:"op"`
}

type OutboxEvent struct {
	DocID   string    `json:"doc_id"`
	Op      CRDTOp    `json:"op"`
	SentAt  time.Time `json:"sent_at,omitempty"`
	SeqKey  string    `json:"seq_key"`
	OpID    string    `json:"op_id"`
}

func seqKey(actorID string, counter int64) string {
	// Deterministic identity -> deterministic ordering.
	// I kept it simple: actorID lexical order, then counter numeric order.
	return actorID + ":" + fmt.Sprintf("%020d", counter)
}

func opID(actorID string, counter int64) string {
	return actorID + ":" + fmt.Sprintf("%d", counter)
}

type Publisher interface {
	Publish(ctx context.Context, event OutboxEvent) error
}

// A fake publisher for demonstration. Replace with Kafka/NATS/etc in real systems.
type LogPublisher struct{}

func (p LogPublisher) Publish(ctx context.Context, event OutboxEvent) error {
	b, _ := json.Marshal(event)
	log.Printf("[publish] %s", string(b))
	return nil
}

func ensureSchema(ctx context.Context, db *sql.DB) error {
	// For the blog, I keep schema creation explicit.
	_, err := db.ExecContext(ctx, `
CREATE TABLE IF NOT EXISTS crdt_outbox (
  id UUID PRIMARY KEY,
  doc_id TEXT NOT NULL,
  actor_id TEXT NOT NULL,
  counter BIGINT NOT NULL,
  seq_key TEXT NOT NULL,
  payload JSONB NOT NULL,
  created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
  published_at TIMESTAMPTZ NULL
);

CREATE INDEX IF NOT EXISTS idx_outbox_unsent_order
ON crdt_outbox (doc_id, seq_key)
WHERE published_at IS NULL;

CREATE INDEX IF NOT EXISTS idx_outbox_unsent
ON crdt_outbox (published_at)
WHERE published_at IS NULL;
`)
	return err
}

func enqueueOpsTx(ctx context.Context, db *sql.DB, docID string, ops []CRDTOp) error {
	// One transaction:
	// - In a real system, you would also update your CRDT state here.
	// - Then you insert outbox rows for each operation event.
	tx, err := db.BeginTx(ctx, &sql.TxOptions{Isolation: sql.LevelSerializable})
	if err != nil {
		return err
	}
	defer tx.Rollback()

	for _, op := range ops {
		skey := seqKey(op.ActorID, op.Counter)
		e := OutboxEvent{
			DocID:  docID,
			Op:     op,
			SeqKey: skey,
			OpID:   opID(op.ActorID, op.Counter),
		}

		payload, err := json.Marshal(e)
		if err != nil {
			return err
		}

		_, err = tx.ExecContext(ctx, `
INSERT INTO crdt_outbox (id, doc_id, actor_id, counter, seq_key, payload)
VALUES ($1, $2, $3, $4, $5, $6)
`, uuid.New(), docID, op.ActorID, op.Counter, skey, payload)
		if err != nil {
			return err
		}
	}

	if err := tx.Commit(); err != nil {
		return err
	}
	return nil
}

func publishOutboxDeterministically(ctx context.Context, db *sql.DB, publisher Publisher, workerID string, batchSize int) error {
	// This worker loop:
	// 1) Select unsent outbox rows ordered by (doc_id, seq_key) and lock them.
	// 2) Publish them in the exact order they were selected.
	// 3) Mark them as published.
	tx, err := db.BeginTx(ctx, &sql.TxOptions{Isolation: sql.LevelReadCommitted})
	if err != nil {
		return err
	}

	rows, err := tx.QueryContext(ctx, `
SELECT id, payload
FROM crdt_outbox
WHERE published_at IS NULL
ORDER BY doc_id, seq_key
LIMIT $1
FOR UPDATE SKIP LOCKED
`, batchSize)
if err != nil {
		tx.Rollback()
		return err
	}
	defer rows.Close()

	type candidate struct {
		id      string
		payload []byte
	}
	var candidates []candidate

	for rows.Next() {
		var id string
		var payload []byte
		if err := rows.Scan(&id, &payload); err != nil {
			tx.Rollback()
			return err
		}
		candidates = append(candidates, candidate{id: id, payload: payload})
	}
	if err := rows.Err(); err != nil {
		tx.Rollback()
		return err
	}

	// Nothing to do
	if len(candidates) == 0 {
		_ = tx.Rollback()
		return nil
	}

	// Publish in deterministic order: we rely on ORDER BY and keeping the slice order.
	for _, c := range candidates {
		var evt OutboxEvent
		if err := json.Unmarshal(c.payload, &evt); err != nil {
			tx.Rollback()
			return err
		}

		evt.SentAt = time.Now().UTC()

		log.Printf("[worker %s] publishing %s for doc=%s seq=%s",
			workerID, evt.OpID, evt.DocID, evt.SeqKey)

		if err := publisher.Publish(ctx, evt); err != nil {
			tx.Rollback()
			return err
		}

		_, err := tx.ExecContext(ctx, `
UPDATE crdt_outbox
SET published_at = now()
WHERE id = $1 AND published_at IS NULL
`, c.id)
		if err != nil {
			tx.Rollback()
			return err
		}
	}

	if err := tx.Commit(); err != nil {
		return err
	}
	return nil
}

func parseOpsFromJSON(s string) ([]CRDTOp, error) {
	var ops []CRDTOp
	if err := json.Unmarshal([]byte(s), &ops); err != nil {
		return nil, err
	}
	return ops, nil
}

func main() {
	ctx := context.Background()

	dsn := os.Getenv("DATABASE_URL")
	if dsn == "" {
		dsn = "postgres://postgres:postgres@localhost:5432/postgres?sslmode=disable"
	}

	db, err := sql.Open("pgx", dsn)
	if err != nil {
		log.Fatal(err)
	}
	defer db.Close()

	if err := ensureSchema(ctx, db); err != nil {
		log.Fatal(err)
	}

	publisher := LogPublisher{}

	// Demo workload:
	// I enqueue operations in a deliberately scrambled order,
	// then I run the outbox worker and observe the deterministic publish order.
	//
	// The operations are for the same doc. Worker should publish by (actor_id, counter).
	docID := "doc-123"

	opsScrambled := []CRDTOp{
		{ActorID: "b", Counter: 2, Op: "insert:X"},
		{ActorID: "a", Counter: 1, Op: "insert:A"},
		{ActorID: "a", Counter: 3, Op: "insert:C"},
		{ActorID: "b", Counter: 1, Op: "insert:B"},
	}

	// Show that input ordering is not sorted.
	{
		var keys []string
		for _, op := range opsScrambled {
			keys = append(keys, seqKey(op.ActorID, op.Counter))
		}
		log.Printf("input seq keys (scrambled): %v", keys)
	}

	if err := enqueueOpsTx(ctx, db, docID, opsScrambled); err != nil {
		log.Fatal(err)
	}

	// Run one worker pass; in production you'd loop continuously.
	if err := publishOutboxDeterministically(ctx, db, publisher, "W1", 100); err != nil {
		log.Fatal(err)
	}

	// Validate order empirically by reading remaining none.
	var remaining int
	if err := db.QueryRowContext(ctx, `SELECT count(*) FROM crdt_outbox WHERE published_at IS NULL`).Scan(&remaining); err != nil {
		log.Fatal(err)
	}
	log.Printf("remaining unsent outbox rows: %d", remaining)

	// Optional: clean up (keeps demo repeatable).
	_, _ = db.ExecContext(ctx, `DELETE FROM crdt_outbox WHERE doc_id = $1`, docID)

	// Additional sanity check: deterministic ordering function.
	expected := make([]string, 0, len(opsScrambled))
	for _, op := range opsScrambled {
		expected = append(expected, seqKey(op.ActorID, op.Counter))
	}
	sort.Strings(expected)

	log.Printf("expected deterministic order (sorted seq keys): %s", strings.Join(expected, ", "))
}
```

---

## How to run it

### 1) Start PostgreSQL
For local testing:

```bash
docker run --name pg -e POSTGRES_PASSWORD=postgres -e POSTGRES_USER=postgres -e POSTGRES_DB=postgres -p 5432:5432 -d postgres:16
```

### 2) Create the table (optional because the Go code does it)
```bash
psql "postgres://postgres:postgres@localhost:5432/postgres?sslmode=disable" -f schema.sql
```

### 3) Run the Go program
```bash
go mod init example.com/outbox && go get github.com/jackc/pgx/v5/stdlib github.com/google/uuid
go run main.go
```

You’ll see logs like:

- `input seq keys (scrambled): [...]`
- worker publishes events in `(doc_id, seq_key)` order
- remaining unsent rows becomes `0`

The important part is that even though the input operations were scrambled, the publishing order is stable because:
- outbox rows carry `seq_key`
- worker selects with `ORDER BY doc_id, seq_key`
- worker locks rows with `FOR UPDATE SKIP LOCKED` to avoid duplicate publishes across workers

---

## What I learned (and the design constraints)

1. **Transactional Outbox solves “don’t lose events,” not “publish in the same order.”**  
   When ordering matters to projections, you must make ordering explicit.

2. **Determinism needs a key, not a vibe.**  
   I used a concrete operation identity `(actor_id, counter)` to derive `seq_key`. That’s stable across restarts and independent of thread scheduling or DB scan behavior.

3. **Don’t rely on concurrency behavior.**  
   Even with correct transactions, multiple workers and retry paths can reorder delivery. `ORDER BY ...` plus row locking makes the worker’s action order deterministic.

---

## Conclusion

I built a deterministic outbox for CRDT operation event streams by combining the Transactional Outbox pattern with explicit, stable ordering derived from operation identity. The end result is that downstream projections see the same sequence after crashes and redeploys, even when worker concurrency and database scan order would otherwise cause reordering.

Designing A Deterministic Outbox For Crdt Event Streams In Go

## The bug I chased: “login works on one subdomain, breaks on the other”
I ran into a weird authentication failure while building a setup like this:

- I had an **edge-rendered** Next.js app serving `app.example.com`
- A separate **API** lived at `api.example.com`
- Users logged in through the frontend, but after login the API would act like the user was still logged out
- Worse: refreshing sometimes “fixed” it, then broke again

The root cause ended up being **subdomain cookie scope** and **edge caching** interacting in an unexpected way. This post documents the exact architecture I ended up with and the working code I used to make it deterministic.

---

## What was actually happening
A cookie is a small piece of data the browser stores and sends back automatically with HTTP requests. Cookie delivery is controlled by attributes like:

- `Domain`: which hostnames the cookie should be sent to
- `Path`: which URL paths the cookie applies to
- `Secure`: only send over HTTPS
- `HttpOnly`: JS can’t read it (prevents XSS cookie theft)
- `SameSite`: controls whether cookies are sent on cross-site requests

In my case, the login endpoint set a cookie on `app.example.com` without a `Domain` that covered `api.example.com`, so the API requests never included the cookie.

Then edge-rendering added a second layer: the server sometimes served a cached page that didn’t trigger a fresh session check the way I expected.

---

## The niche architecture: “Edge frontend session cookie + API token verification without re-calling user info”
The approach I used:

1. Frontend login sets a cookie that is valid for **both** `app.example.com` and `api.example.com`.
2. API endpoints read that cookie and verify the session using a signed token (no extra user-info fetch).
3. Edge responses use cache headers that ensure auth-sensitive content isn’t incorrectly cached.

To keep it concrete, I used:

- **Next.js App Router** for the frontend
- **Express** for the API
- Cookie-based sessions with a signed JWT stored in an **HttpOnly** cookie

---

## Working implementation

### 1) Frontend: login route sets the cookie with `Domain=.example.com`
This is the critical part: using `Domain=.example.com` so the cookie is sent to both subdomains.

```typescript
// app/api/login/route.ts (Next.js App Router)
import { cookies } from "next/headers";
import jwt from "jsonwebtoken";

export async function POST() {
  // Demo auth: in a real app you would validate user credentials here.
  const userId = "user_123";

  const secret = process.env.SESSION_JWT_SECRET!;
  const token = jwt.sign({ sub: userId }, secret, { expiresIn: "15m" });

  // Domain=.example.com makes the cookie available to:
  // - app.example.com
  // - api.example.com
  // - (and other subdomains too, if you use a wider Domain)
  cookies().set("session", token, {
    httpOnly: true,
    secure: true,
    sameSite: "lax",
    path: "/",
    domain: ".example.com",
    maxAge: 15 * 60, // seconds
  });

  return new Response(JSON.stringify({ ok: true }), {
    status: 200,
    headers: { "content-type": "application/json" },
  });
}
```

**Why this fixes the “split brain” cookie problem:**
- Without `domain: ".example.com"`, the browser will default the cookie to the host that set it.
- That means a cookie set by `app.example.com` won’t automatically be included in requests to `api.example.com`.

---

### 2) API: read cookie and verify the token
Here’s an Express API server that validates the session cookie on every request.

```typescript
// server.ts (Express API)
import express from "express";
import cookieParser from "cookie-parser";
import jwt from "jsonwebtoken";

const app = express();
app.use(cookieParser());

const secret = process.env.SESSION_JWT_SECRET!;

function requireAuth(
  req: express.Request,
  res: express.Response,
  next: express.NextFunction
) {
  const token = req.cookies.session;
  if (!token) return res.status(401).json({ error: "missing session" });

  try {
    const payload = jwt.verify(token, secret) as { sub: string };
    (req as any).userId = payload.sub;
    return next();
  } catch {
    return res.status(401).json({ error: "invalid session" });
  }
}

app.get("/me", requireAuth, (req, res) => {
  return res.json({
    userId: (req as any).userId,
  });
});

app.listen(4000, () => {
  console.log("API listening on http://localhost:4000");
});
```

**What happens when you run this:**
- Your browser hits `app.example.com/api/login` → cookie gets set with `Domain=.example.com`
- Browser then calls `api.example.com/me`
- Cookie `session` is automatically attached to the request
- API verifies JWT and returns the user identity

---

### 3) Prevent edge caching from serving auth-wrong content
I also added explicit cache headers on the frontend route that checks auth. The key is to avoid caching responses that depend on cookies.

In Next.js App Router, for dynamic auth pages I forced “no store”.

```typescript
// app/me/page.tsx (Next.js)
import "server-only";
import { cookies } from "next/headers";
import jwt from "jsonwebtoken";

export const dynamic = "force-dynamic"; // disable static optimization

export default function MePage() {
  const token = cookies().get("session")?.value;

  if (!token) {
    return (
      <main>
        <h1>Not logged in</h1>
      </main>
    );
  }

  const secret = process.env.SESSION_JWT_SECRET!;
  try {
    const payload = jwt.verify(token, secret) as { sub: string };
    return (
      <main>
        <h1>Logged in</h1>
        <p>User id: {payload.sub}</p>
      </main>
    );
  } catch {
    return (
      <main>
        <h1>Session expired</h1>
      </main>
    );
  }
}
```

**Why this matters:**
- Edge-rendered frameworks can try to cache content.
- Auth-dependent pages must not be cached in a way that ignores per-user cookies.

---

## Local testing with hostnames that resemble production
Testing with `localhost` hides the `Domain` behavior because cookie scoping rules differ when you’re not using real subdomains.

A practical way to simulate subdomains is to use `/etc/hosts`:

```bash
# /etc/hosts
127.0.0.1 app.example.com
127.0.0.1 api.example.com
```

Then run:

- Frontend on something like `https://app.example.com:3000` (with HTTPS)
- API on `https://api.example.com:4000` (with HTTPS)

Because the cookie uses `secure: true`, **HTTP won’t send it**. That’s another easy-to-miss failure mode that looked similar to the cookie scoping bug.

---

## Common failure modes I hit (and how I recognized them)

### 1) Cookie set without a shared domain
Symptom:
- `/api/login` returns 200
- Cookie appears in the browser for `app.example.com`
- `/me` on `api.example.com` returns 401

Fix:
- Add `domain: ".example.com"` when setting the cookie.

### 2) Cookie is present but API still sees “missing session”
Symptom:
- Cookie appears in devtools
- API still returns 401 “missing session”

Fix:
- Ensure requests are actually going to `api.example.com` (not a different host)
- Ensure HTTPS is used (`secure: true`)
- Ensure `sameSite` doesn’t block sending cookies for your request type

In my case, `sameSite: "lax"` worked with navigation-style flows.

### 3) Edge caching serves stale auth UI
Symptom:
- Clicking login sometimes “works” only after a manual refresh
- UI claims you’re logged in but API says otherwise (or vice versa)

Fix:
- Use `dynamic = "force-dynamic"` (or equivalent) for auth-sensitive routes
- Ensure cache headers don’t treat auth pages as public static content

---

## Summary
I built an auth architecture for an edge-rendered Next.js frontend paired with an Express API where login sets an `HttpOnly` JWT cookie scoped to `Domain=.example.com`, and the API verifies that cookie on every protected route. The deterministic behavior came from addressing the “split brain” caused by cookie scoping across subdomains and preventing edge caching from serving auth-dependent content out of context.

Core Engineering Posts