Building a Conflict Review Queue in Python #

Your auto-merge stage just returned None on a document because two edits touched the same billing field, and there is no safe way for a machine to pick a winner without losing money. Dropping the write is unacceptable and guessing is worse, so the document has to reach a person. This page builds the plumbing that gets it there: a Python worker that subscribes to _changes with conflicts=true, and for every unresolved fork writes a durable review task capturing the document id, the competing leaves, and the field-level diffs between them. A companion resolve_task() then applies a human’s decision — committing the chosen body and tombstoning the losing revisions in one atomic _bulk_docs call. It is the hands-on layer under Manual Review Sync Queues, and it exists precisely because not every conflict is machine-resolvable.

Immediate Triage & Prerequisites #

Before building anything, confirm that the conflicts you see are genuinely stuck rather than being cleaned up by a resolver you forgot was running. Ask the source database for every document that currently carries a live fork:

# every doc id with at least one conflicting leaf, via a built-in view
curl -s "http://localhost:5984/orders/_all_docs?conflicts=true&include_docs=true" | \
  python3 -c "import sys,json;[print(r['id'], r['doc'].get('_conflicts')) for r in json.load(sys.stdin)['rows'] if r['doc'].get('_conflicts')]"

If that list is long and static, nothing downstream is resolving them — which is exactly the gap a review queue fills. Prerequisites for the code below: Python 3.8+ with requests (pip install requests), write access to the source database, and a second database dedicated to review tasks so operator metadata never pollutes production documents. Create it once with curl -X PUT http://localhost:5984/conflict_reviews. Keep the review database separate from your merge logic: the queue must survive a resolver restart, so its state lives in CouchDB, not in worker memory. The strategy each task ultimately applies still comes from Algorithm Selection for Merge; this page only routes the cases where no strategy is safe.

Step-by-Step Implementation #

Each step below includes a verification you can run before moving on.

1. Subscribe to the feed with conflicts surfaced. Open a continuous or longpoll _changes request with include_docs=true&conflicts=true so every change arrives with its _conflicts array populated when a fork exists:

   curl -s "http://localhost:5984/orders/_changes?feed=longpoll&include_docs=true&conflicts=true&since=now&timeout=60000" | \
     python3 -c "import sys,json;d=json.load(sys.stdin);print('changes:',len(d['results']),'last_seq:',d['last_seq'])"

   Verify you receive a last_seq cursor; you will persist it so a restart resumes rather than replays the whole history.

2. Fetch the competing leaves. For a conflicted change, read every branch body — not just the winner — with open_revs, so the task captures what the human must actually compare:

   curl -s "http://localhost:5984/orders/order-9182?open_revs=all" \
     -H "Accept: application/json"
   # returns one {"ok": {...leaf body...}} object per live leaf

   Assert the number of returned leaves equals len(_conflicts) + 1 (the winner plus each loser).

3. Derive a stable fork fingerprint. Sort the leaf revision ids and hash them into a deterministic id. This fingerprint is the idempotency key: the same fork produces the same task id no matter how many times the feed replays it.

4. Write the review task idempotently. PUT the task into the review database using the fingerprint as _id. A 409 from that write is a success signal — it means the task already exists — so swallow it rather than retrying blindly.

5. Resolve on a human decision. When an operator picks or merges a winner, commit the chosen body against the current winning _rev and a _deleted tombstone for each losing leaf in a single _bulk_docs batch, then flip the task to resolved with an audit record. Verify afterward that ?conflicts=true returns an empty array for the document.

Complete Working Example #

The script below runs two roles from one file: enqueue_from_feed() scans the feed and creates idempotent tasks, and resolve_task() applies a human decision atomically. It never guesses a winner; the decision is passed in.

import hashlib
import sys
import time

import requests

SOURCE = "http://localhost:5984/orders"          # production documents
REVIEW = "http://localhost:5984/conflict_reviews"  # durable task queue


def fork_fingerprint(doc_id: str, leaf_revs) -> str:
    """Deterministic task id: same fork -> same id, so enqueue is idempotent."""
    joined = doc_id + "|" + "|".join(sorted(leaf_revs))
    return "task:" + hashlib.sha1(joined.encode()).hexdigest()[:20]


def fetch_leaves(doc_id: str, session: requests.Session):
    """Return [winner, *losers] bodies for a conflicted document."""
    resp = session.get(f"{SOURCE}/{doc_id}",
                        params={"open_revs": "all"},
                        headers={"Accept": "application/json"}, timeout=30)
    resp.raise_for_status()
    return [row["ok"] for row in resp.json() if "ok" in row]


def field_diffs(leaves):
    """Map each field to the distinct values across leaves that disagree."""
    keys = {k for leaf in leaves for k in leaf if not k.startswith("_")}
    diffs = {}
    for k in sorted(keys):
        values = [leaf.get(k) for leaf in leaves]
        if len({repr(v) for v in values}) > 1:  # leaves disagree on this field
            diffs[k] = values
    return diffs


def enqueue_from_feed(session: requests.Session, since="now", timeout_ms=60000):
    """One longpoll pass: create a pending task per unresolved fork."""
    resp = session.get(f"{SOURCE}/_changes", params={
        "feed": "longpoll", "include_docs": "true", "conflicts": "true",
        "since": since, "timeout": timeout_ms}, timeout=timeout_ms / 1000 + 10)
    resp.raise_for_status()
    feed = resp.json()
    for row in feed["results"]:
        doc = row.get("doc") or {}
        conflicts = doc.get("_conflicts")
        if not conflicts:
            continue  # no fork on this change; nothing to review
        leaf_revs = [doc["_rev"]] + conflicts
        task_id = fork_fingerprint(doc["_id"], leaf_revs)
        leaves = fetch_leaves(doc["_id"])
        task = {
            "_id": task_id, "type": "conflict_review", "state": "pending",
            "doc_id": doc["_id"], "source_db": "orders",
            "leaf_revs": sorted(leaf_revs),
            "diffs": field_diffs(leaves),
            "created_at": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
            "audit": [],
        }
        r = session.put(f"{REVIEW}/{task_id}", json=task, timeout=30)
        if r.status_code == 409:
            print(f"skip {task_id}: already queued")  # idempotent: same fork
        else:
            r.raise_for_status()
            print(f"queued {task_id} for {doc['_id']} "
                  f"({len(leaf_revs)} leaves, {len(task['diffs'])} diffs)")
    return feed["last_seq"]


def resolve_task(session: requests.Session, task_id: str, winner_body: dict,
                 operator: str):
    """Apply a human decision: commit winner, tombstone losers, one batch."""
    task = session.get(f"{REVIEW}/{task_id}", timeout=30).json()
    if task["state"] == "resolved":
        print(f"{task_id} already resolved; no-op")  # guard against double apply
        return
    doc_id = task["doc_id"]
    # Re-read live leaves: the fork may have shifted since the task was written.
    live = session.get(f"{SOURCE}/{doc_id}",
                       params={"open_revs": "all"},
                       headers={"Accept": "application/json"}, timeout=30).json()
    live_leaves = [row["ok"] for row in live if "ok" in row]
    live_revs = {leaf["_rev"] for leaf in live_leaves}
    if live_revs != set(task["leaf_revs"]):
        # Someone resolved this elsewhere; do not clobber a newer state.
        task["state"] = "stale"
        session.put(f"{REVIEW}/{task_id}", json=task, timeout=30)
        raise RuntimeError(f"{task_id} is stale: fork changed under review")
    winner_rev = max(live_revs)  # extend the current winning branch
    batch = [{**winner_body, "_id": doc_id, "_rev": winner_rev}]
    for rev in live_revs - {winner_rev}:
        batch.append({"_id": doc_id, "_rev": rev, "_deleted": True})
    br = session.post(f"{SOURCE}/_bulk_docs", json={"docs": batch}, timeout=30)
    br.raise_for_status()
    task["state"] = "resolved"
    task["audit"].append({"by": operator, "at": time.strftime(
        "%Y-%m-%dT%H:%M:%SZ", time.gmtime()), "action": "resolved"})
    session.put(f"{REVIEW}/{task_id}", json=task, timeout=30)
    print(f"{task_id} resolved by {operator}; {len(batch) - 1} losers tombstoned")


if __name__ == "__main__":
    s = requests.Session()
    last = enqueue_from_feed(s, since="0")  # backfill every existing fork
    print("cursor:", last)
    # A human then decides; here we demo picking the highest-rev leaf verbatim.
    pending = s.get(f"{REVIEW}/_all_docs", params={"include_docs": "true"},
                    timeout=30).json()
    for r in pending["rows"]:
        t = r["doc"]
        if t.get("state") == "pending":
            leaves = fetch_leaves(t["doc_id"], s)
            chosen = {k: v for k, v in max(
                leaves, key=lambda d: d["_rev"]).items() if not k.startswith("_")}
            resolve_task(s, t["_id"], chosen, operator="ops@example.com")
    sys.exit(0)

Run the enqueuer against a database with live forks and it creates one task per fork; run it again and every task logs already queued, proving the idempotency key holds. The demo resolver picks the highest-_rev leaf only to exercise the commit path — in production winner_body is whatever the human assembled in your review UI.

Gotchas & Edge Cases #

• Idempotent enqueue is non-negotiable. A continuous feed redelivers a conflicted document on every subsequent change to it, and a restart from since=0 replays everything. Keying the task _id on the sorted leaf revisions — not on doc_id alone — means the same fork collapses to one task while a new fork on the same document gets its own task.
• Tasks go stale when the conflict resolves elsewhere. Another node’s resolver, or a later human, may tombstone the leaves before your operator acts. resolve_task() re-reads the live leaves and refuses to commit if they no longer match, marking the task stale instead of overwriting a newer state.
• Preserve ordering only where it matters. The queue itself is unordered — pick tasks by age or priority, not feed arrival — but the _bulk_docs commit must always extend the current winning _rev, so re-read the winner at resolve time rather than trusting the rev captured at enqueue time.
• Backpressure is real. A conflict storm can enqueue faster than humans resolve. Cap the pending count, and when it is exceeded, page an operator rather than silently growing the review database — the same escalation path covered in Escalating Unresolvable Conflicts to Operators.
• Keep an audit trail. Every resolution appends {by, at, action} to the task rather than deleting it. A resolved task is your record of who chose what and when — indispensable when a merge is later disputed.

Verification & Observability #

Confirm both halves of the system: that forks become tasks, and that resolved tasks actually clear the conflict. For a resolved document, the source _conflicts array must be empty:

curl -s "http://localhost:5984/orders/order-9182?conflicts=true" | \
  python3 -c "import sys,json;d=json.load(sys.stdin);print('remaining conflicts:', d.get('_conflicts', []))"

For queue health, count tasks by state so you can watch the backlog and detect stalls:

curl -s "http://localhost:5984/conflict_reviews/_all_docs?include_docs=true" | \
  python3 -c "import sys,json,collections;c=collections.Counter(r['doc'].get('state') for r in json.load(sys.stdin)['rows']);print(dict(c))"

A healthy system shows pending trending toward zero and resolved growing. A rising stale count means resolvers are racing each other or the review UI is showing operators data that has already moved on. Emit pending and the age of the oldest pending task as gauges; when either crosses a threshold, feed it into your escalation policy and, for longer-horizon tracking, the objectives in Conflict-Resolution Metrics & SLOs for CouchDB Sync.

FAQ #

Related #

• Escalating Unresolvable Conflicts to Operators
• Designing Fallback Resolution Chains in Python
• Auto-Merge Rule Engines

Part of: Manual Review Sync Queues