Conflict-Resolution Metrics & SLOs for CouchDB Sync #

A sync fleet can look healthy on every dashboard while silently accumulating tens of thousands of unresolved conflicts, because the default CouchDB metrics never count documents carrying a _conflicts array — they count requests, not divergence. When an operator finally reads a document with ?conflicts=true and finds a backlog days deep, there is no history to say when it started or how fast it grew. This page defines the metrics that actually describe conflict-resolution health — conflict rate, backlog depth, resolution latency, auto-resolution success ratio, and manual-queue depth — shows how to count conflicts at scale with a design-document view (because Mango _find cannot select on _conflicts), and turns those numbers into service-level objectives with error budgets. It is the observability layer for the whole of Conflict Detection & Automated Resolution Strategies: the strategies chosen in algorithm selection for merge and dispatched by auto-merge rule engines only earn trust once you can measure what they do to the backlog.

Configuration Schema & Required Parameters #

Conflict counting hangs off two artifacts you deploy once: a design document whose view emits every document currently carrying a _conflicts array, and a collector configuration that names the databases to scan, the poll interval, and the SLO thresholds to evaluate against. The view is the scalable counting path because a CouchDB map function is passed the _conflicts field when a document is forked — so it can index exactly the divergent documents — whereas Mango _find can only return conflicts with its conflicts option and cannot use _conflicts in a selector. Deploy the design document to each database you sync:

{
  "_id": "_design/conflict_metrics",
  "views": {
    "conflicted": {
      "map": "function (doc) { if (doc._conflicts) { emit(doc._id, doc._conflicts.length); } }",
      "reduce": "_count"
    }
  },
  "options": { "partitioned": false }
}

The collector reads that view and evaluates thresholds. Drive it entirely from configuration so one image serves every environment:

couch_url: "https://central-db.example:5984"
databases: ["orders", "inventory", "iot_telemetry"]
poll_interval_seconds: 30
view: "_design/conflict_metrics/_view/conflicted"
slo:
  backlog_max: 500              # docs currently carrying _conflicts (per database)
  p95_resolution_latency_seconds: 300
  auto_resolution_ratio_min: 0.95
  manual_queue_max: 50

Key callout: never build a conflict count by scanning _all_docs and re-reading each document with ?conflicts=true — that is one HTTP round trip per document and melts under load. The reduced view returns the backlog depth in a single query, and its keys give you the exact document IDs when you need them.

Because conflict metadata is computed rather than stored, treat the view as eventually consistent: a freshly born conflict appears in the index only after the view is updated, so read with ?stale=update_after (or the modern ?update=lazy) when you want a fast, slightly-behind gauge, and force a fresh build only when auditing. The authoritative semantics for finding conflicts through a view are in the Apache CouchDB replication conflicts documentation.

Streaming Detection / Monitoring Setup #

Backlog depth is a level; conflict rate is a flow, and you measure a flow at the edge where conflicts are born. The _changes feed with style=all_docs&conflicts=true surfaces each document the moment a new leaf lands, letting you increment a born-counter and stamp a first-seen timestamp used later for latency. The minimal listener below tracks the rate without holding the whole backlog in memory:

import json
import time

import httpx


def stream_conflict_births(db_url: str, since: str = "now"):
    """Yield (doc_id, timestamp) the first time each forked document is seen.

    style=all_docs reports every leaf and conflicts=true attaches the computed
    _conflicts array, so a document appears here exactly when it is divergent.
    """
    params = {
        "feed": "continuous",
        "since": since,
        "style": "all_docs",
        "include_docs": "true",
        "conflicts": "true",
        "heartbeat": "10000",
    }
    with httpx.stream("GET", f"{db_url}/_changes", params=params, timeout=None) as r:
        for line in r.iter_lines():
            if not line:                       # heartbeat keep-alive
                continue
            doc = json.loads(line).get("doc") or {}
            if doc.get("_conflicts"):          # only forked docs carry this key
                yield doc["_id"], time.time()


if __name__ == "__main__":
    seen: dict[str, float] = {}
    window_start = time.time()
    births = 0
    for doc_id, ts in stream_conflict_births("http://localhost:5984/orders"):
        if doc_id not in seen:                 # count each fork once per window
            seen[doc_id] = ts
            births += 1
        if ts - window_start >= 60:            # emit conflict rate per minute
            print(f"conflict_rate_per_min={births}")
            births, window_start = 0, time.time()

Deduplicate by document ID inside the window: during network flapping the same fork reappears on the feed, and counting each reappearance would inflate the rate. The first-seen timestamp captured here is the start clock for resolution latency, closed when the resolver later reports the document converged.

Core Implementation #

The collector below is production-grade: it reads the reduced conflicts view for backlog depth, lists offending IDs on demand, tracks resolution latency against a first-seen registry, and exposes every metric as a labelled gauge dictionary ready to hand to a Prometheus client. It uses structured logging and bounded retries, and it never re-reads documents one at a time — the reduced view is a single request per database.

import logging
import time
from dataclasses import dataclass, field
from typing import Optional

import requests

logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
logger = logging.getLogger("conflict-metrics")

VIEW = "_design/conflict_metrics/_view/conflicted"


@dataclass
class Sample:
    """One scrape's worth of conflict-resolution gauges for a single database."""
    database: str
    backlog_depth: int = 0
    manual_queue_depth: int = 0
    auto_success_ratio: float = 1.0
    p95_resolution_latency_s: float = 0.0
    conflicted_ids: list = field(default_factory=list)


class ConflictMetrics:
    """Collect conflict-resolution health gauges from a CouchDB database.

    Backlog depth comes from the reduced conflicts view (one request). Latency
    is derived from a first-seen registry updated as forks are observed and
    cleared as documents converge. Success ratio and manual-queue depth are fed
    by the resolver via record_resolution().
    """

    def __init__(self, base_url: str, database: str,
                 auth: Optional[tuple] = None, max_retries: int = 4):
        self.base_url = base_url.rstrip("/")
        self.database = database
        self.session = requests.Session()
        if auth:
            self.session.auth = auth
        self.max_retries = max_retries
        self._first_seen: dict[str, float] = {}       # doc_id -> fork timestamp
        self._latencies: list[float] = []             # recent resolved latencies (s)
        self._auto, self._manual = 0, 0               # rolling outcome counters

    def _get(self, path: str, params: dict) -> dict:
        for attempt in range(1, self.max_retries + 1):
            resp = self.session.get(f"{self.base_url}/{self.database}/{path}",
                                    params=params, timeout=30)
            if resp.status_code < 500:
                resp.raise_for_status()
                return resp.json()
            backoff = min(2 ** attempt, 20)
            logger.warning("%s returned %s; retry in %ss", path, resp.status_code, backoff)
            time.sleep(backoff)
        raise RuntimeError(f"exhausted retries reading {path}")

    def backlog_depth(self) -> int:
        """Reduced view -> count of documents currently carrying _conflicts."""
        data = self._get(VIEW, {"reduce": "true", "update": "lazy"})
        rows = data.get("rows", [])
        return int(rows[0]["value"]) if rows else 0    # _count reduce -> single row

    def conflicted_ids(self, limit: int = 200) -> list:
        """Un-reduced view -> the actual forked document ids (bounded)."""
        data = self._get(VIEW, {"reduce": "false", "limit": str(limit)})
        return [row["id"] for row in data.get("rows", [])]

    def observe_fork(self, doc_id: str, ts: Optional[float] = None) -> None:
        """Register a newly detected fork so its latency can be measured later."""
        self._first_seen.setdefault(doc_id, ts or time.time())

    def record_resolution(self, doc_id: str, *, automatic: bool) -> None:
        """Close a fork: compute latency and update outcome counters."""
        born = self._first_seen.pop(doc_id, None)
        if born is not None:
            self._latencies.append(time.time() - born)
            self._latencies = self._latencies[-1000:]   # bounded rolling window
        if automatic:
            self._auto += 1
        else:
            self._manual += 1

    def _p95_latency(self) -> float:
        if not self._latencies:
            return 0.0
        ordered = sorted(self._latencies)
        idx = min(len(ordered) - 1, int(round(0.95 * (len(ordered) - 1))))
        return round(ordered[idx], 2)

    def _auto_ratio(self) -> float:
        total = self._auto + self._manual
        return round(self._auto / total, 4) if total else 1.0

    def collect(self) -> Sample:
        """Produce one Sample of all gauges for this scrape."""
        sample = Sample(
            database=self.database,
            backlog_depth=self.backlog_depth(),
            manual_queue_depth=self._manual,           # replace with real queue length
            auto_success_ratio=self._auto_ratio(),
            p95_resolution_latency_s=self._p95_latency(),
        )
        logger.info("db=%s backlog=%d p95_latency=%.1fs auto_ratio=%.3f",
                    sample.database, sample.backlog_depth,
                    sample.p95_resolution_latency_s, sample.auto_success_ratio)
        return sample

    def evaluate_slo(self, sample: Sample, slo: dict) -> dict:
        """Return each SLO's breach state; a True value means the budget is burning."""
        return {
            "backlog_breached": sample.backlog_depth > slo["backlog_max"],
            "latency_breached": sample.p95_resolution_latency_s
                                > slo["p95_resolution_latency_seconds"],
            "auto_ratio_breached": sample.auto_success_ratio
                                   < slo["auto_resolution_ratio_min"],
            "manual_queue_breached": sample.manual_queue_depth > slo["manual_queue_max"],
        }


if __name__ == "__main__":
    slo = {"backlog_max": 500, "p95_resolution_latency_seconds": 300,
           "auto_resolution_ratio_min": 0.95, "manual_queue_max": 50}
    collector = ConflictMetrics("http://localhost:5984", "orders")
    # simulate the resolver reporting outcomes it observed this interval
    collector.observe_fork("order-1", ts=time.time() - 42)
    collector.record_resolution("order-1", automatic=True)
    snap = collector.collect()
    print(snap)
    print("SLO state:", collector.evaluate_slo(snap, slo))

The manual-queue depth here is fed by the resolver’s outcome stream; in production replace it with the live length of the review queue described in Manual Review Sync Queues. The collect() output maps one-to-one onto Prometheus gauges, which is exactly the shape the Prometheus Metrics & Exporter Integration for CouchDB Replication exporter expects, and it complements the request-level signals in Replication Observability & Metrics in CouchDB.

Strategy Variants & Trade-offs #

Counting conflicts at scale is a choice between three approaches, and the right one depends on how many documents you sync and how fresh the number must be. Each trades accuracy, freshness, and load differently:

The _changes scan is unbeatable for the rate metric and for stamping when a fork was born, but reconstructing an exact backlog total from it means holding every open conflict in memory — fragile across restarts. The design-doc view is the scalable source of truth for depth: the reduced _count returns the total in one request and the un-reduced rows hand you the exact IDs, at the price of index-build cost on every write and a few seconds of index lag. Sampling trades accuracy for near-zero cost — read a random slice with ?conflicts=true, compute the conflicted fraction, multiply by doc_count — and suits databases with millions of documents where an exact figure is not worth the index. Most production fleets run the _changes scan and the view together: the feed for rate and latency, the view for depth and audit. Note again that Mango _find is not on this list: its conflicts option only decorates returned rows and cannot select or index on _conflicts, so it cannot count conflicts at all.

Deployment & Orchestration #

Run the collector as a small stateless service, one replica per database namespace it scrapes, scheduled on the same cadence as the rest of your monitoring. Two collectors scraping the same view double the query load and can double-count latency samples, so shard by database rather than cloning workers onto one target. Drive everything through the environment so one image serves every environment:

# Collector environment (one replica per database namespace)
COUCH_URL=https://central-db.example:5984
COUCH_DATABASES=orders,inventory,iot_telemetry
POLL_INTERVAL_SECONDS=30
SLO_BACKLOG_MAX=500
SLO_P95_LATENCY_S=300
SLO_AUTO_RATIO_MIN=0.95
METRICS_PORT=9105               # Prometheus scrape endpoint

Expose the gauges on a /metrics endpoint for Prometheus and a /healthz that confirms the collector can reach CouchDB and that its _changes cursor for the rate metric is advancing — a stalled cursor reports a falsely flat conflict rate, which is worse than no metric. Deploy the _design/conflict_metrics document ahead of the collector and let its first read trigger the index build in a low-traffic window; the initial build touches every document, so on a large database it can be slow and I/O-heavy. Pin the _changes since checkpoint in durable storage so a restart resumes the rate count rather than replaying history and briefly spiking the rate. Because the view is eventually consistent, alert on the backlog gauge with a short for: duration so a single lagging scrape does not page.

Troubleshooting & Common Errors #

FAQ #

Related #

• Algorithm Selection for Merge
• Auto-Merge Rule Engines
• Manual Review Sync Queues
• Prometheus Metrics & Exporter Integration for CouchDB Replication
• Replication Observability & Metrics in CouchDB

Part of: Conflict Detection & Automated Resolution Strategies