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