Limit reconciliation apply mutations

dns_sync/cli.py

@@ -17,6 +17,20 @@ from .registry import load_adapter, load_provider, load_source
 from .reporting import print_endpoint_plans, print_plan
 
 
+def _nonnegative_int(value: str) -> int:
+    try:
+        parsed = int(value)
+    except ValueError as exc:
+        raise argparse.ArgumentTypeError(
+            "must be a non-negative integer"
+        ) from exc
+    if parsed < 0:
+        raise argparse.ArgumentTypeError(
+            "must be a non-negative integer"
+        )
+    return parsed
+
+
 def build_parser():
     parser = argparse.ArgumentParser(prog="dns-sync", description="Synchronize reverse-proxy hostnames to a pluggable DNS provider")
     sub = parser.add_subparsers(dest="command", required=True)
@@ -27,7 +41,7 @@ def build_parser():
 
     reconcile = sub.add_parser(
         "reconcile",
-        help="Run the persistent generic engine without applying changes",
+        help="Run the persistent generic reconciliation engine",
     )
     reconcile.add_argument(
         "--state-db",
@@ -47,6 +61,12 @@ def build_parser():
         action="store_true",
         help="Apply and verify planned changes; default is read-only",
     )
+    reconcile.add_argument(
+        "--max-changes",
+        type=_nonnegative_int,
+        default=None,
+        help="Maximum CREATE/UPDATE actions allowed with --apply",
+    )
 
     sub.add_parser("plugins", help="Show the configured source/provider plugins")
     return parser
@@ -175,6 +195,7 @@ def run_reconcile(args):
         for plan in result.plans
         for action in plan.actions
     )
+    change_count = create_count + update_count
     noop_count = sum(
         action.operation.value == "noop"
         for plan in result.plans
@@ -198,6 +219,18 @@ def run_reconcile(args):
             "Refusing to apply while discovery failures are present"
         )
 
+    if (
+        args.max_changes is not None
+        and change_count > args.max_changes
+    ):
+        raise RuntimeError(
+            f"Refusing to apply {change_count} change(s); "
+            f"--max-changes allows {args.max_changes}"
+        )
+
+    if args.max_changes is not None:
+        print(f"Apply mutation budget: {args.max_changes}")
+
     with SQLiteStateStore(args.state_db) as store:
         execution = execute_endpoint_plans(
             adapters,