MongoDB Oplog Batch Validation

Point-in-time recovery on MongoDB is only as trustworthy as the oplog that replays the interval after a snapshot, so this page builds — as a concrete instance of async batching for large datasets — a Python pipeline that validates the oplog window backing a point-in-time recovery target: it reads oplog entries between two BSON timestamps in bounded async batches, proves the ts sequence is strictly monotonic with no gap or rollback, verifies each operation is safely replayable more than once, and confirms the captured window actually spans the recovery point you intend to restore to. Concurrency is bounded with an asyncio.Semaphore so a busy replica set’s oplog — millions of entries between backups — is validated without buffering the whole window in memory, the coverage check ties directly to the recovery objective in your RTO/RPO mapping, and every anomaly it raises is reported with a reason the surrounding checksum validation pipeline and error-routing layers consume.

Architecture and Execution Model

Figure. The oplog window is read between two timestamps, checked for a strictly increasing ts sequence, validated for idempotent operations under bounded concurrency, confirmed to span the recovery target, and reduced to a gating exit code and coverage metric.

The oplog is a capped collection, and that shape drives the whole validation. Because it is capped, the oldest entries are continuously overwritten as new writes arrive, so the first thing worth knowing is whether the window still reaches back far enough to cover the recovery point — a snapshot taken at time T is only restorable to a later target if the oplog still holds every entry from T forward, and a cap sized too small for the write rate silently erodes that guarantee between drills. Each entry carries a ts field, a BSON timestamp of (seconds, increment) that is unique and strictly increasing across the replica set, so a monotonic-sequence check catches both a gap (entries aged out of the cap) and a rollback (a demoted primary’s un-replicated writes disappearing when it rejoins). The pipeline reads the window in bounded async batches rather than one query so a multi-million-entry oplog never materializes in memory, and it checks operation idempotency because MongoDB’s own recovery replays oplog ops at least once — an op that is not safe to reapply is a latent corruption that only manifests during an actual restore, long after the drill that should have caught it.

The order of the four checks is chosen so the cheapest, most decisive test runs first. Monotonicity is a single linear pass over timestamps and immediately disqualifies a reordered or rolled-back window; coverage is two comparisons against the snapshot and target epochs; only then does the pipeline spend effort on the per-operation idempotency inspection that must touch every entry’s payload. This ordering mirrors the WAL sibling’s continuity-first strategy, but the failure semantics differ in one crucial way: a write-ahead log may legitimately switch timelines after a promotion, whereas the oplog’s ts must never decrease within a validated window, so there is no benign case to reconcile — every non-increase is a hard fault.

Prerequisites

• Python 3.10+ — the pipeline uses asyncio.gather and async iteration; both are stable from 3.9/3.10 onward.
• A driver or an export: pip install motor==3.4.* for a live async connection to a replica-set member, or a JSON export of oplog entries (mongodump of local.oplog.rs, converted) for offline validation and the built-in self-test.
• A read-only account with the read role on the local database so it can query oplog.rs. Point it at a secondary, never the primary, and prefer a member reading with majority concern.
• The recovery target as a Unix epoch second, plus the snapshot timestamp the oplog must reach back to — these define the coverage assertion.
• The replica-set context: know whether the deployment uses majority write concern, because entries not yet majority-committed can roll back and must be excluded from a trustworthy window.

Production Implementation

The pipeline reads the window, verifies monotonicity as a cheap gating pass, validates operation idempotency under a semaphore, and asserts coverage of the recovery target. It works offline from a JSON export (and ships a self-test main) so it runs as-is under python3 oplog_validate.py. It exits 0 when the window is monotonic, idempotent, and covers the target, 1 on any gap, rollback, non-idempotent op, or coverage shortfall, and 2 on a usage or input error.

#!/usr/bin/env python3
"""oplog_validate.py — validate a MongoDB oplog window for point-in-time recovery.

Read oplog entries between two timestamps in bounded async batches, prove the ts
sequence is strictly monotonic, verify each op is idempotent, and confirm the
window covers the recovery target. Exit codes gate the DR drill:
    0  monotonic, idempotent, and covers the recovery point -> window is sound
    1  gap, rollback, non-idempotent op, or coverage short   -> halt recovery
    2  usage / unreadable input                              -> abort
"""
import argparse
import asyncio
import json
import logging
import sys
from typing import AsyncIterator, Dict, List, Optional, Tuple

EXIT_OK = 0
EXIT_FAILED = 1
EXIT_USAGE = 2

# Oplog op codes that are inherently safe to reapply. "i" (insert) is idempotent
# only when it targets a fixed _id; "n" is a no-op heartbeat; "c" (command) and
# unindexed updates need per-op inspection.
IDEMPOTENT_OPS = {"n", "d"}  # no-op and delete-by-_id always reapply cleanly

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s | %(levelname)s | %(message)s",
)
logger = logging.getLogger("oplog_validate")


def ts_tuple(entry: dict) -> Tuple[int, int]:
    """Normalize a BSON-timestamp-shaped ts into a comparable (seconds, inc)."""
    ts = entry["ts"]
    if isinstance(ts, dict):  # extended-JSON export form: {"$timestamp": {...}}
        inner = ts.get("$timestamp", ts)
        return int(inner["t"]), int(inner["i"])
    return int(ts[0]), int(ts[1])  # plain [seconds, increment] pair


def is_idempotent(entry: dict) -> bool:
    """A replayable op either is in the safe set or carries a fixed _id target."""
    op = entry.get("op")
    if op in IDEMPOTENT_OPS:
        return True
    if op == "i":  # insert is idempotent iff it pins a concrete _id
        return "_id" in entry.get("o", {})
    if op == "u":  # update is idempotent iff it fully replaces or uses $set only
        upd = entry.get("o", {})
        return "$set" in upd or not any(k.startswith("$inc") for k in upd)
    return False  # commands and anything unrecognized are treated as unsafe


async def stream_batches(entries: List[dict], batch_size: int) -> AsyncIterator[List[dict]]:
    """Yield the window in fixed-size batches without holding it all at once.

    In production this wraps a Motor cursor over local.oplog.rs; offline it
    slices a decoded export so the same validation runs without a live cluster.
    """
    for start in range(0, len(entries), batch_size):
        await asyncio.sleep(0)  # cooperative yield between batches
        yield entries[start:start + batch_size]


async def validate_batch(batch: List[dict], sem: asyncio.Semaphore) -> List[Tuple[Tuple[int, int], str]]:
    """Return (ts, reason) for every non-idempotent op in one batch."""
    async with sem:
        return [(ts_tuple(e), "NON_IDEMPOTENT") for e in batch if not is_idempotent(e)]


async def run(args: argparse.Namespace) -> int:
    try:
        with open(args.export, "r", encoding="utf-8") as fh:
            entries = json.load(fh)
    except (OSError, json.JSONDecodeError) as exc:
        logger.error("cannot load oplog export %s: %s", args.export, exc)
        return EXIT_USAGE
    if not entries:
        logger.error("empty oplog window")
        return EXIT_USAGE

    # Monotonicity: a cheap, gating single pass over the ordered window.
    prev: Optional[Tuple[int, int]] = None
    breaks: List[str] = []
    for entry in entries:
        cur = ts_tuple(entry)
        if prev is not None:
            if cur <= prev:
                breaks.append(f"non-monotonic ts {prev} -> {cur} (rollback or reorder)")
        prev = cur
    if breaks:
        for b in breaks:
            logger.error("SEQUENCE %s", b)
        return EXIT_FAILED

    first_ts, last_ts = ts_tuple(entries[0]), ts_tuple(entries[-1])
    logger.info("window monotonic: %d entries from %s to %s",
                len(entries), first_ts, last_ts)

    # Coverage: the window must start at or before the snapshot and reach the target.
    if first_ts[0] > args.snapshot_epoch:
        logger.error("COVERAGE oldest entry %s newer than snapshot %d; oplog aged out",
                     first_ts, args.snapshot_epoch)
        return EXIT_FAILED
    if last_ts[0] < args.target_epoch:
        logger.error("COVERAGE newest entry %s does not reach target %d",
                     last_ts, args.target_epoch)
        return EXIT_FAILED

    sem = asyncio.Semaphore(args.concurrency)
    tasks = []
    async for batch in stream_batches(entries, args.batch_size):
        tasks.append(validate_batch(batch, sem))
    results = await asyncio.gather(*tasks)
    non_idempotent = [item for batch in results for item in batch]

    coverage_s = last_ts[0] - first_ts[0]
    logger.info("idempotency-checked %d entries across %.0fs of oplog",
                len(entries), float(coverage_s))

    if non_idempotent:
        for ts, reason in non_idempotent[:20]:
            logger.error("FAULT %s at ts=%s", reason, ts)
        logger.error("%d non-idempotent op(s) in window", len(non_idempotent))
        return EXIT_FAILED
    return EXIT_OK


def _write_selftest(path: str) -> None:
    """Emit a small valid oplog export so the script runs with no arguments."""
    sample = [
        {"ts": [1720000000, 1], "op": "i", "o": {"_id": 1, "v": "a"}},
        {"ts": [1720000000, 2], "op": "u", "o": {"$set": {"v": "b"}}},
        {"ts": [1720000005, 1], "op": "d", "o": {"_id": 1}},
        {"ts": [1720000010, 1], "op": "n", "o": {}},
    ]
    with open(path, "w", encoding="utf-8") as fh:
        json.dump(sample, fh)


def main() -> int:
    parser = argparse.ArgumentParser(description="Validate a MongoDB oplog window for PITR")
    parser.add_argument("--export", help="JSON export of oplog entries (ordered by ts)")
    parser.add_argument("--snapshot-epoch", type=int, default=1720000000,
                        help="epoch second of the base snapshot the oplog must reach")
    parser.add_argument("--target-epoch", type=int, default=1720000008,
                        help="epoch second of the recovery point to restore to")
    parser.add_argument("--batch-size", type=int, default=1000)
    parser.add_argument("--concurrency", type=int, default=8)
    parser.add_argument("--selftest", action="store_true",
                        help="write and validate a built-in sample window")
    args = parser.parse_args()

    if args.batch_size < 1 or args.concurrency < 1:
        logger.error("--batch-size and --concurrency must be positive")
        return EXIT_USAGE

    if args.selftest or not args.export:
        args.export = "/tmp/oplog_selftest.json"
        _write_selftest(args.export)

    return asyncio.run(run(args))


if __name__ == "__main__":
    sys.exit(main())

Two checks distinguish this from a naive scan. The monotonicity pass treats ts <= prev as fatal because a decrease means either the oplog rolled back (a former primary’s writes were discarded on rejoin) or the export was reordered, and either way the replay sequence is untrustworthy; unlike a WAL timeline, the oplog has no legitimate “reset” within a single window, so any non-increase is a hard fault. The idempotency check encodes what MongoDB’s recovery semantics require: because oplog application is at-least-once, an insert is only safe when it pins a concrete _id, and an update is only safe when it fully replaces the document or uses $set rather than a relative $inc that would compound on replay. Commands and anything unrecognized are treated as unsafe by default, so the pipeline fails closed rather than certifying a window it cannot reason about.

Step-by-Step Execution Walkthrough

1. Run the self-test. python3 oplog_validate.py --selftest writes a small valid window and validates it, exiting 0; this confirms the pipeline is wired correctly before you point it at real data.
2. Export a real window. Dump local.oplog.rs between the snapshot and target timestamps and convert it to an ordered JSON array, or swap stream_batches for a Motor cursor against a secondary.
3. Set the coverage bounds. Pass --snapshot-epoch (when the base snapshot was taken) and --target-epoch (the recovery point); the window must start at or before the first and reach the second.
4. Size the batching. Raise --batch-size for fewer, larger reads on a fast secondary; keep --concurrency bounded so the semaphore caps how many batches validate at once.
5. Branch on the exit code. The orchestrator reads $?: 0 marks the window replayable to the target, 1 halts recovery on a gap, rollback, non-idempotent op, or coverage shortfall, 2 aborts on bad input.

Verification and Expected Output

A sound window logs monotonicity, the idempotency span, and exits 0:

2026-07-05 03:40:02 | INFO | window monotonic: 4 entries from (1720000000, 1) to (1720000010, 1)
2026-07-05 03:40:02 | INFO | idempotency-checked 4 entries across 10s of oplog
$ echo $?
0

A window whose newest entry falls short of the recovery target fails coverage and exits 1:

2026-07-05 03:44:19 | ERROR | COVERAGE newest entry (1720000010, 1) does not reach target 1720000090
$ echo $?
1

Success means an exit of 0, a “window monotonic” line spanning every entry, and an idempotency span that brackets the interval between snapshot and target. Any SEQUENCE, COVERAGE, or FAULT NON_IDEMPOTENT line is a hard stop — the window cannot be trusted to replay cleanly to the recovery point.

Failure Modes and Troubleshooting

Integration Notes

Wire the validator into orchestration by branching on its exit code, and gate any point-in-time restore on a 0. In Airflow, a BashOperator fails the DAG on exit 1 so a window that cannot reach the target never drives a doomed replay; a scheduled run trends the coverage span so you catch an oplog cap that has silently shrunk below the snapshot cadence. Route each SEQUENCE and FAULT line into the error-routing layer so a transient export error is retried while a genuine rollback halts, and reconcile the window’s endpoints against the intended point-in-time recovery target and the objectives in your RTO/RPO mapping so coverage is judged against a real recovery goal. For the authoritative behavior of the oplog, its ts field, and majority write concern, see the MongoDB replica-set oplog documentation; for the async primitives the batching relies on, see the Python asyncio documentation.

Frequently Asked Questions

Related

• Async Batching for Large Datasets — the parent workflow this oplog pipeline instantiates for streaming windows.
• PostgreSQL WAL Streaming Validation — the sibling pipeline that applies the same bounded-batch pattern to a write-ahead log.
• Point-in-Time Recovery Targeting — the recovery-target selection this window’s coverage check must satisfy.
• Checksum Validation Pipelines — the integrity layer that governs the base snapshot the oplog is replayed onto.

This page is one component of the broader Async Batching for Large Datasets workflow; from there you can move up to the Automated Backup Integrity Check Implementation overview.