Designing Fallback Resolution Chains in Python #

You have three different conflict resolvers — a schema-aware auto-merge, a field-union merger, a last-write-wins shortcut — and a review queue for everything else, but they are wired together with a tangle of if statements that nobody dares touch. A new document type needs a fourth strategy and the branching logic collapses. The fix is a Chain of Responsibility: order your resolvers from safest to lossiest, ask each one in turn, and let a link either resolve the conflict, pass it to the next link, or escalate it straight to a human. This page builds that chain in Python — a ResolutionChain that threads a conflicted document through pluggable Resolver links and always terminates, because the last link is a manual review queue that never passes. It is the coordinating pattern under Fallback Resolution Chains, and it turns strategy selection from nested conditionals into an ordered, testable list.

Immediate Triage & Prerequisites #

Before designing links, look at what your current resolution logic actually does on real conflicts — the answer is usually “silently drops a branch.” Pull a conflicted document and inspect how many fields genuinely disagree, because that number decides which links can safely act:

curl -s "http://localhost:5984/inventory/widget-55?open_revs=all" -H "Accept: application/json" | \
  python3 -c "import sys,json;leaves=[r['ok'] for r in json.load(sys.stdin) if 'ok' in r];\
keys={k for l in leaves for k in l if not k.startswith('_')};\
print('leaves:',len(leaves),'disputed:',[k for k in keys if len({repr(l.get(k)) for l in leaves})>1])"

If only non-overlapping fields differ, a field-union merge is lossless and should win before anything lossy runs. Prerequisites: Python 3.8+ (requests only for the commit path). Decide the safety ordering before you write code — this is the one design decision the whole pattern hinges on. The individual strategies each link wraps are covered in Algorithm Selection for Merge; when every automated link declines, the chain’s terminal link hands off to the queue built in Building a Conflict Review Queue in Python.

Step-by-Step Implementation #

Each step ends with an assertion you can run.

1. Define one resolver interface. Every link implements the same method, resolve(doc_id, leaves) -> Outcome, where Outcome is one of resolved(body), pass_(), or escalate(reason). Verify each link returns exactly one of the three; an accidental None return must fail loudly, not be treated as pass.

2. Order links by safety, lossless first. Place non-destructive merges (schema-aware, field-union) ahead of lossy ones (last-write-wins), and put the human queue last. Assert in a unit test that no lossy link precedes a lossless one — the ordering is a contract, not a suggestion.

3. Thread confidence through the chain. A link that resolves attaches a confidence to its outcome, and a link may itself pass when its own confidence is below a floor, deferring to a later strategy or the human. Verify a low-confidence field-union outcome falls through rather than committing.

4. Guarantee termination with a terminal link. The last link — the manual queue — must never return pass; it always resolves (by enqueuing) or is unreachable because an earlier link acted. Assert that running the chain over any input reaches an outcome, never falling off the end.

5. Log which link resolved. Record the resolving link’s name on every outcome so metrics can attribute resolutions per strategy. Verify each committed resolution carries a resolved_by field feeding your metrics pipeline.

Complete Working Example #

The script defines the Outcome type, a Resolver protocol, three automated links plus a terminal manual link, and a ResolutionChain that runs them in order. The __main__ block walks a conflicted document through the chain and prints which link resolved it.

import sys
from dataclasses import dataclass, field
from typing import Callable, Optional


@dataclass
class Outcome:
    """A link's verdict: exactly one of resolved / pass / escalate."""
    kind: str                         # "resolved" | "pass" | "escalate"
    body: Optional[dict] = None       # winning document body when resolved
    confidence: float = 0.0
    reason: str = ""
    resolved_by: str = ""


def resolved(body, confidence, by):
    return Outcome("resolved", body=body, confidence=confidence, resolved_by=by)


def pass_():
    return Outcome("pass")


def escalate(reason):
    return Outcome("escalate", reason=reason)


def disputed_fields(leaves):
    """Fields whose values differ across leaves (ignoring CouchDB metadata)."""
    keys = {k for leaf in leaves for k in leaf if not k.startswith("_")}
    return {k for k in keys if len({repr(l.get(k)) for l in leaves}) > 1}


@dataclass
class Resolver:
    """A named link wrapping one strategy function returning an Outcome."""
    name: str
    fn: Callable[[str, list], Outcome]

    def resolve(self, doc_id: str, leaves: list) -> Outcome:
        out = self.fn(doc_id, leaves)
        if out.kind not in ("resolved", "pass", "escalate"):
            raise TypeError(f"{self.name} returned invalid outcome {out.kind!r}")
        return out


# --- Link 1: schema-aware merge (lossless): resolve only if fields are additive
def schema_merge(doc_id, leaves):
    disputed = disputed_fields(leaves)
    # Only act when disagreement is confined to a known additive field.
    if disputed and disputed <= {"tags"}:
        merged = dict(max(leaves, key=lambda d: d["_rev"]))
        merged["tags"] = sorted({t for l in leaves for t in l.get("tags", [])})
        return resolved(merged, confidence=0.97, by="schema_merge")
    return pass_()  # not our shape; defer to the next link


# --- Link 2: field-union merge (lossless): safe when leaves touch disjoint keys
def field_union(doc_id, leaves):
    disputed = disputed_fields(leaves)
    base = dict(max(leaves, key=lambda d: d["_rev"]))
    # If two leaves changed the *same* field to different values it is not
    # a clean union; hand on rather than pick arbitrarily.
    for k in disputed:
        vals = {repr(l.get(k)) for l in leaves if k in l}
        if len(vals) > 1:
            return pass_()
    for leaf in leaves:
        for k, v in leaf.items():
            if not k.startswith("_"):
                base.setdefault(k, v)
    return resolved(base, confidence=0.9, by="field_union")


# --- Link 3: last-write-wins (lossy): a shortcut, so it must sit AFTER unions
def last_write_wins(doc_id, leaves):
    dated = [l for l in leaves if l.get("updated_at")]
    if not dated:
        return escalate("no timestamp to order leaves")  # cannot safely pick
    winner = max(dated, key=lambda d: d["updated_at"])
    return resolved(dict(winner), confidence=0.6, by="last_write_wins")


# --- Link 4: terminal manual queue: NEVER passes, so the chain terminates
def manual_queue(doc_id, leaves):
    # In production this calls the review-queue enqueuer; here we just mark it.
    body = {"_queued_for_review": True, "leaf_count": len(leaves)}
    return resolved(body, confidence=0.0, by="manual_queue")


@dataclass
class ResolutionChain:
    links: list = field(default_factory=list)
    min_confidence: float = 0.5   # a resolve below this is downgraded to pass

    def run(self, doc_id: str, leaves: list) -> Outcome:
        for link in self.links:
            out = link.resolve(doc_id, leaves)
            if out.kind == "resolved" and out.confidence >= self.min_confidence:
                return out                     # committed by this link
            if out.kind == "escalate":
                return self.links[-1].resolve(doc_id, leaves)  # jump to terminal
            # "pass", or a resolve too weak to trust: fall through to next link
        # Unreachable: the terminal link always resolves. Guard anyway.
        raise RuntimeError("chain fell through without a terminal resolver")


if __name__ == "__main__":
    chain = ResolutionChain(links=[
        Resolver("schema_merge", schema_merge),
        Resolver("field_union", field_union),
        Resolver("last_write_wins", last_write_wins),
        Resolver("manual_queue", manual_queue),   # terminal, always last
    ])
    conflicted_leaves = [
        {"_rev": "3-aaa", "price": 19.99, "stock": 40,
         "updated_at": "2026-07-10T09:00:00Z"},
        {"_rev": "3-bbb", "price": 24.50, "stock": 40,
         "updated_at": "2026-07-11T14:30:00Z"},  # same field, different value
    ]
    outcome = chain.run("widget-55", conflicted_leaves)
    print(f"resolved_by={outcome.resolved_by} "
          f"confidence={outcome.confidence} kind={outcome.kind}")
    # price disagrees, so schema+union pass; LWW picks the later timestamp.
    assert outcome.resolved_by == "last_write_wins"
    sys.exit(0)

In the demo both leaves changed price to different values, so the lossless links pass (a union cannot pick between conflicting values for one field), and last-write-wins resolves on the later updated_at. Change the leaves so only tags differ and the schema link resolves first at high confidence — the exact same chain, a different link doing the work, and resolved_by proving which one.

Gotchas & Edge Cases #

• Ordering is the whole design; get it wrong and you lose data. A lossy link placed before a lossless one wins prematurely — last-write-wins ahead of field-union will discard a perfectly mergeable branch. Enforce the ordering in a test: assert no lossy resolver’s index is lower than any lossless resolver’s.
• Never let the chain fall off the end. If the terminal link can pass, an unusual document reaches the end with no verdict and either raises or, worse, is silently dropped. Make the manual queue always resolve; the RuntimeError guard exists to catch a misconfigured chain in testing, not in production.
• Thread confidence, do not ignore it. A resolver that technically can merge but at low confidence should defer. The chain downgrades a below-floor resolved to a fall-through, so a shaky field-union yields to last-write-wins or the human rather than committing a guess.
• Per-document-type chains beat one universal chain. A billing document and a telemetry reading want different orderings — the former should reach the human early, the latter tolerates last-write-wins. Key the chain by document type and keep each ordering explicit rather than cramming every case into one list.
• Log the resolving link or you cannot improve. Without resolved_by on every outcome you cannot tell whether last-write-wins is quietly eating conflicts a smarter link should have caught. Feed the per-link counts into Conflict-Resolution Metrics & SLOs for CouchDB Sync.

Verification & Observability #

Confirm the chain terminates and attributes correctly. A cheap property test runs the chain over randomized leaves and asserts every input produces an outcome whose resolved_by is one of the configured links:

# property check: the chain always terminates with an attributed resolution
for _ in range(1000):
    leaves = random_conflicted_leaves()          # your generator
    out = chain.run("probe", leaves)
    assert out.kind == "resolved" and out.resolved_by, "chain did not terminate"

In production, watch the distribution of resolved_by across committed resolutions. A healthy chain resolves most conflicts at the lossless links and sends only a trickle to the manual queue:

curl -s "http://localhost:5984/inventory/_design/metrics/_view/by_resolver?group=true" | \
  python3 -c "import sys,json;[print(r['key'],r['value']) for r in json.load(sys.stdin)['rows']]"

If last_write_wins or manual_queue dominates that view, an earlier lossless link is either mis-ordered or too conservative — tune it, because every conflict that reaches last-write-wins is a branch you chose to discard. Track the fraction resolved per link as an SLO and alert when the lossless share drops.

FAQ #

Related #

• Algorithm Selection for Merge
• Building a Conflict Review Queue in Python
• Manual Review Sync Queues

Part of: Fallback Resolution Chains