Writing Declarative Merge Rules in YAML #

Your conflict-resolution policy lives in a Python if/elif ladder, and every time operations wants orders.* to switch from last-write-wins to a field-union merge they have to file a pull request, wait for review, and ship a deploy. That coupling is the problem this page fixes: it shows how to externalize which resolver runs for which document into a version-controlled YAML file, so a policy change is a config edit and a hot-reload rather than a code release. You will design a small rules schema that maps a document type or key pattern to a strategy (lww, field-union, crdt, or manual) with per-field overrides and an explicit fallback, validate it with pyyaml, and build a RuleEngine that parses the file and dispatches the right resolver per conflicted document. This is the configuration surface that sits on top of the auto-merge rule engines section; the resolver functions it dispatches to are built in Writing Custom Conflict Resolver Functions in Python.

Immediate Triage / Prerequisites #

Before externalizing anything, confirm the pain is really policy churn and not a resolver bug. If your deploy history shows commits that only change a merge branch selector — “route inventory.* to field-union”, “make session.* manual” — that logic belongs in data, not code. Grep your repository for the tell-tale dispatch ladder:

# find hard-coded strategy selection worth externalizing
grep -rn "doc\[.type.\]\|startswith(\|== .lww.\|== .field" resolvers/ | head -20

Prerequisites for the steps below: Python 3.8+, pyyaml (pip install pyyaml), and a set of resolver callables that already work — the engine only selects a resolver, it does not implement merges. If you do not yet have those functions, build them first in Writing Custom Conflict Resolver Functions in Python, and decide which algorithm each document class needs using algorithm selection for merge. One rule to hold onto throughout: YAML never contains executable merge logic. It only names a strategy from a fixed, code-defined registry. A rules file that could inject behaviour would turn an ops config edit into remote code execution.

Step-by-Step Implementation #

Follow these steps to move policy out of Python and into a validated, reloadable YAML document. Each step includes a check so you can confirm state before moving on.

1. Define the YAML schema. Fix a small, closed vocabulary: a top-level version, an ordered rules list where each rule has a matcher (type, id_prefix, or glob), a strategy from the enum, an optional fields map of per-field strategy overrides, and a required top-level fallback. Keeping matchers in an ordered list (not a dict) makes specificity explicit — the first match wins.

   version: 1
   fallback: manual          # any doc that matches nothing goes to review
   rules:
     - id_prefix: "invoice-" # most specific matcher first
       strategy: lww
     - type: order
       strategy: field-union
       fields:
         status: lww         # override one field inside a field-union merge
     - type: sensor_reading
       strategy: crdt

   Verify the file loads as plain data with no surprises: python -c "import yaml,sys; print(type(yaml.safe_load(open('rules.yaml'))))" must print <class 'dict'>. Always use yaml.safe_load, never yaml.load, so the file cannot construct arbitrary Python objects.

2. Validate against the schema and its version. Parsing is not validating. Reject unknown strategy names, unknown keys, a missing fallback, and a version your engine does not understand — a typo like strategy: filed-union must fail loudly at load time, not silently route to the fallback in production. Assert the shape immediately after loading:

   raw = yaml.safe_load(open("rules.yaml"))
   assert raw["version"] == 1, "unsupported rules schema version"
   assert raw["fallback"] in {"lww", "field-union", "crdt", "manual"}

   A rule whose strategy is not in the registry should raise, naming the offending line so operations can fix it before the file is ever activated.

3. Build the rule engine and dispatch. Load the validated rules once into a RuleEngine, then for each conflicted document call select(doc) to get the winning rule and resolve(doc, branches) to run it. The engine maps each strategy name to a resolver callable from a registry defined in code; per-field overrides are passed through so a field-union merge can treat status as last-write-wins. A document that matches no rule uses the file’s fallback, and manual routes to a Manual Review Sync Queues queue rather than forcing a lossy write.

   Confirm dispatch with a dry run that prints the selected strategy per sample document without writing to CouchDB — the next section provides that harness. Testing a rule change safely before activation is covered in Testing Conflict Resolvers with Synthetic Conflicts.

Complete Working Example #

The script below is self-contained and runnable. It defines the strategy registry in code, loads and validates a YAML rules file, and exposes a RuleEngine that selects a rule per document and dispatches to the matching resolver. It writes a sample rules.yaml, then resolves two documents to prove routing works — including a per-field override and a fallback to manual review.

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

import yaml

SCHEMA_VERSION = 1
VALID_STRATEGIES = {"lww", "field-union", "crdt", "manual"}


# --- Strategy registry: the ONLY place merge behaviour is defined -----------
def _lww(branches, overrides=None):
    """Keep the branch with the highest application timestamp."""
    return max(branches, key=lambda d: d.get("updated_at", ""))


def _field_union(branches, overrides=None):
    """Union non-null fields; honour per-field overrides (e.g. status->lww)."""
    overrides = overrides or {}
    merged = dict(branches[0])
    for branch in branches[1:]:
        for key, value in branch.items():
            if key.startswith("_") or value is None:
                continue
            if overrides.get(key) == "lww":
                # this field resolves by timestamp, not by union
                if branch.get("updated_at", "") >= merged.get("updated_at", ""):
                    merged[key] = value
            elif merged.get(key) is None:
                merged[key] = value
    return merged


def _crdt(branches, overrides=None):
    """Placeholder: sum a commutative counter across branches."""
    total = sum(b.get("counter", 0) for b in branches)
    merged = dict(max(branches, key=lambda d: d.get("updated_at", "")))
    merged["counter"] = total
    return merged


# manual has no merge function: it signals escalation, not a write.
RESOLVERS: dict[str, Optional[Callable]] = {
    "lww": _lww,
    "field-union": _field_union,
    "crdt": _crdt,
    "manual": None,
}


class RulesValidationError(ValueError):
    """Raised when a rules file is structurally or semantically invalid."""


@dataclass
class Rule:
    strategy: str
    matcher_kind: str            # "type" | "id_prefix" | "glob"
    matcher_value: str
    fields: dict = field(default_factory=dict)

    def matches(self, doc: dict) -> bool:
        if self.matcher_kind == "type":
            return doc.get("type") == self.matcher_value
        if self.matcher_kind == "id_prefix":
            return str(doc.get("_id", "")).startswith(self.matcher_value)
        if self.matcher_kind == "glob":
            return fnmatch.fnmatch(str(doc.get("_id", "")), self.matcher_value)
        return False


class RuleEngine:
    """Select and run a conflict resolver per document from a YAML policy."""

    def __init__(self, rules: list[Rule], fallback: str):
        self.rules = rules              # evaluated in order; first match wins
        self.fallback = fallback

    @classmethod
    def from_yaml(cls, path: str) -> "RuleEngine":
        raw = yaml.safe_load(open(path, encoding="utf-8"))
        return cls._validate(raw)

    @staticmethod
    def _validate(raw: dict) -> "RuleEngine":
        if not isinstance(raw, dict):
            raise RulesValidationError("top level must be a mapping")
        if raw.get("version") != SCHEMA_VERSION:
            raise RulesValidationError(
                f"unsupported version {raw.get('version')!r}; expected {SCHEMA_VERSION}")
        fallback = raw.get("fallback")
        if fallback not in VALID_STRATEGIES:
            raise RulesValidationError(f"fallback must be one of {VALID_STRATEGIES}")
        parsed: list[Rule] = []
        for i, entry in enumerate(raw.get("rules", [])):
            strategy = entry.get("strategy")
            if strategy not in VALID_STRATEGIES:
                raise RulesValidationError(
                    f"rules[{i}]: unknown strategy {strategy!r}")
            kinds = [k for k in ("type", "id_prefix", "glob") if k in entry]
            if len(kinds) != 1:
                raise RulesValidationError(
                    f"rules[{i}]: exactly one matcher (type|id_prefix|glob) required")
            for fld, fstrat in (entry.get("fields") or {}).items():
                if fstrat not in VALID_STRATEGIES:
                    raise RulesValidationError(
                        f"rules[{i}].fields.{fld}: unknown strategy {fstrat!r}")
            parsed.append(Rule(strategy, kinds[0], entry[kinds[0]],
                               entry.get("fields", {})))
        return RuleEngine(parsed, fallback)

    def select(self, doc: dict) -> Rule:
        """Return the first matching rule, or a synthetic fallback rule."""
        for rule in self.rules:
            if rule.matches(doc):
                return rule
        return Rule(self.fallback, "type", "*")  # fallback carries no matcher

    def resolve(self, doc: dict, branches: list[dict]):
        """Dispatch the selected strategy; None means escalate to manual review."""
        rule = self.select(doc)
        resolver = RESOLVERS[rule.strategy]
        if resolver is None:                       # strategy == "manual"
            return None
        return resolver(branches, rule.fields)


if __name__ == "__main__":
    with open("rules.yaml", "w", encoding="utf-8") as fh:
        fh.write(
            "version: 1\n"
            "fallback: manual\n"
            "rules:\n"
            "  - id_prefix: \"invoice-\"\n"
            "    strategy: lww\n"
            "  - type: order\n"
            "    strategy: field-union\n"
            "    fields:\n"
            "      status: lww\n"
        )

    engine = RuleEngine.from_yaml("rules.yaml")

    order = {"_id": "order-77", "type": "order"}
    branches = [
        {"_id": "order-77", "_rev": "3-a", "customer": "Acme",
         "status": "packed", "updated_at": "2026-07-18T09:00:00Z"},
        {"_id": "order-77", "_rev": "3-b", "note": "rush",
         "status": "shipped", "updated_at": "2026-07-18T10:30:00Z"},
    ]
    merged = engine.resolve(order, branches)
    print("order ->", engine.select(order).strategy, "| status:", merged["status"],
          "| note:", merged.get("note"))

    session = {"_id": "session-5", "type": "session"}  # matches nothing -> manual
    print("session ->", engine.select(session).strategy,
          "| resolve returns:", engine.resolve(session, []))
    sys.exit(0)

Running it prints that order-77 resolved via field-union with status taken from the newer branch (shipped) while the disjoint note field survived, and that the unmatched session document routed to manual (resolve returns None). Change one line of rules.yaml and rerun — no code changed, yet the policy did.

Gotchas & Edge Cases #

• Validation must run before activation, not on first conflict. If you only discover an invalid strategy when a document finally conflicts, a typo can sit dormant for weeks and then silently divert live traffic to the fallback. Validate on load and in CI so a bad file never reaches production; treat a RulesValidationError as a hard startup failure.
• Matcher order is the policy. Because the first match wins, a broad type: order rule placed above a narrow id_prefix: "order-vip-" rule will shadow the specific one forever. Order matchers most-specific first, and add a test that asserts the expected rule is chosen for representative IDs.
• Unknown document types must hit the fallback, never crash. A document with no type and no matching prefix should resolve deterministically to the file’s fallback — usually manual — so a new document class introduced by an app update degrades safely instead of raising mid-stream.
• Hot-reload needs atomic swaps. When you reload rules.yaml on a running worker, parse and validate the new file into a new RuleEngine object first, then swap the reference in one assignment. Never mutate the live engine’s rule list in place, or an in-flight select() can read a half-updated policy.
• Version the schema explicitly. Bumping version and refusing unknown versions lets you change the file format later without a fleet of workers silently misreading an old-shaped file. Keep old versions loadable only if you write an explicit upgrader.

Verification & Observability #

Confirm the engine routes as intended before and after any policy change. First, dry-run the current rules against a sample of real document IDs and assert the selected strategy matches expectation:

# print the strategy each sampled doc would resolve with, no writes
curl -s "http://localhost:5984/orders/_all_docs?limit=50" | \
  python3 -c "import sys,json; ids=[r['id'] for r in json.load(sys.stdin)['rows']]; print('\n'.join(ids))"

Feed those IDs through engine.select() in a scratch script and eyeball the strategy distribution — a sudden jump in documents routing to manual after an edit means a matcher stopped matching. In production, emit a counter labelled by chosen strategy each time resolve() runs, and alert if the manual share crosses a threshold; that ratio is one of the health signals defined in Conflict-Resolution Metrics & SLOs for CouchDB Sync. After a hot-reload, log the loaded version and rule count so you can prove the worker picked up the new file rather than silently running the old one.

FAQ #

Related #

• Writing Custom Conflict Resolver Functions in Python
• Testing Conflict Resolvers with Synthetic Conflicts
• Designing Fallback Resolution Chains in Python

Part of: Auto-Merge Rule Engines