#!/usr/bin/env python3

"""Verify exports, advance MySQL watermarks, and block incomplete clusters."""

import argparse
import concurrent.futures
import json
import re
import sys
import threading
from dataclasses import dataclass
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Callable, Dict, List, Optional, Sequence, Tuple
from urllib.parse import urlsplit

import pymysql
from boto3.session import Session
from botocore.config import Config


ACTIVE_CLUSTER_IDS_TABLE = "active_cluster_ids"
CLUSTER_BLOCKLIST_TABLE = "cluster_blocklist"
CLUSTER_BLOCKLIST_USED_BY = "br_export"
COMPLETE_MARKER_JOB_INSTANCE_METADATA = "job-instance-name"
COMPLETE_MARKER_BACKUP_TIME_METADATA = "backup-time"
COMPLETE_MARKER_BACKUP_TIME_FORMAT = "%Y-%m-%dT%H:%M:%SZ"
DEFAULT_PARALLEL = 8
TERMINATION_MESSAGE_PATH = Path("/dev/termination-log")
TERMINATION_MESSAGE_MAX_BYTES = 4 * 1024
TERMINATION_RESULT_FIELDS = (
    "status",
    "watermark_instance",
    "export_instance",
    "backup_time",
    "packed_clusters",
    "packed_meta_objects",
    "verified_complete_markers",
    "unverified_complete_marker_clusters",
    "mysql_updated_clusters",
    "newer_complete_marker_clusters",
    "newer_mysql_watermark_clusters",
    "mysql_update_failed_clusters",
)


class WatermarkError(Exception):
    pass


class CompleteMarkerNotVerified(WatermarkError):
    """A missing or mismatched marker that must not advance its watermark."""


@dataclass(frozen=True)
class WatermarkAdvanceSummary:
    verified_markers: int
    updated_clusters: int
    newer_backup_clusters: int
    newer_mysql_watermark_clusters: int
    unverified_clusters: int
    update_failed_clusters: int
    first_error: Optional[str]


def log(message: str) -> None:
    timestamp = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
    print(f"{timestamp} {message}", file=sys.stderr, flush=True)


def non_empty(value: str) -> str:
    if not value:
        raise argparse.ArgumentTypeError("must not be empty")
    return value


def positive_integer(value: str) -> int:
    try:
        parsed = int(value)
    except ValueError as exc:
        raise argparse.ArgumentTypeError("must be a positive integer") from exc
    if parsed <= 0:
        raise argparse.ArgumentTypeError("must be a positive integer")
    return parsed


def kubernetes_name(value: str) -> str:
    if (
        len(value) > 63
        or re.fullmatch(r"[a-z0-9]([-a-z0-9]*[a-z0-9])?", value) is None
    ):
        raise argparse.ArgumentTypeError(
            "must be a Kubernetes DNS label of at most 63 characters"
        )
    return value


def s3_prefix(value: str) -> str:
    if re.fullmatch(r"s3://[^/\s]+/.+", value) is None:
        raise argparse.ArgumentTypeError(
            "must be an s3://<bucket>/<prefix> URI"
        )
    return value


def utc_backup_time(value: str) -> datetime:
    try:
        parsed = datetime.strptime(value, COMPLETE_MARKER_BACKUP_TIME_FORMAT)
    except ValueError as exc:
        raise argparse.ArgumentTypeError(
            f"must use {COMPLETE_MARKER_BACKUP_TIME_FORMAT}"
        ) from exc
    return parsed.replace(tzinfo=timezone.utc)


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(
        description=(
            "List packed metadata, verify complete markers, and update MySQL "
            "watermarks or block incomplete clusters. The container "
            "termination message is a JSON result."
        ),
        formatter_class=argparse.ArgumentDefaultsHelpFormatter,
        allow_abbrev=False,
    )
    parser.add_argument(
        "--watermark-instance",
        type=kubernetes_name,
        required=True,
        metavar="NAME",
    )
    parser.add_argument(
        "--export-instance",
        type=kubernetes_name,
        required=True,
        metavar="NAME",
    )
    parser.add_argument(
        "--tasks-json",
        type=non_empty,
        required=True,
        metavar="PATH",
    )
    parser.add_argument(
        "--out-prefix",
        type=s3_prefix,
        required=True,
        metavar="S3_URI",
    )
    parser.add_argument(
        "--backup-time",
        type=utc_backup_time,
        required=True,
        metavar="UTC_TIME",
    )
    parser.add_argument(
        "--mysql-defaults-file",
        type=non_empty,
        required=True,
        metavar="PATH",
    )
    parser.add_argument(
        "--aws-region",
        type=non_empty,
        required=True,
        metavar="REGION",
    )
    parser.add_argument(
        "--parallel",
        type=positive_integer,
        default=DEFAULT_PARALLEL,
        metavar="COUNT",
        help=(
            "maximum number of cluster reconciliations and MySQL "
            "connections in flight"
        ),
    )
    return parser


def parse_json(text: str, description: str) -> Any:
    try:
        return json.loads(text)
    except json.JSONDecodeError as exc:
        raise WatermarkError(
            f"{description} contains invalid JSON at line {exc.lineno} "
            f"column {exc.colno}: {exc.msg}"
        ) from exc


def read_text_file(path: str, description: str) -> str:
    try:
        return Path(path).read_text(encoding="utf-8")
    except OSError as exc:
        raise WatermarkError(
            f"cannot read {description} ({type(exc).__name__})"
        ) from exc


def load_tasks(path: str) -> List[Dict[str, str]]:
    payload = parse_json(read_text_file(path, "Pack task file"), "Pack task file")
    if not isinstance(payload, list) or not payload:
        raise WatermarkError("Pack task file must contain a non-empty array")

    tasks = []
    for item in payload:
        if not isinstance(item, dict):
            raise WatermarkError("Pack task file contains a non-object task")
        task_prefix = item.get("pack_backup_prefix")
        if (
            not isinstance(task_prefix, str)
            or re.fullmatch(r"s3://[^/]+/.+/", task_prefix) is None
        ):
            raise WatermarkError(
                "Pack task contains an invalid pack_backup_prefix"
            )
        tasks.append({"pack_backup_prefix": task_prefix})
    return tasks


def parse_s3_prefix_uri(value: str, description: str) -> Tuple[str, str]:
    parsed = urlsplit(value)
    if parsed.scheme != "s3" or not parsed.netloc:
        raise WatermarkError(f"{description} must be an s3:// URI")
    if parsed.query or parsed.fragment:
        raise WatermarkError(
            f"{description} must not contain a query or fragment"
        )
    object_prefix = parsed.path.lstrip("/").rstrip("/")
    if not object_prefix:
        raise WatermarkError(f"{description} must include an object prefix")
    return parsed.netloc, object_prefix


def s3_error_code(exc: Exception) -> Optional[str]:
    response = getattr(exc, "response", None)
    if isinstance(response, dict):
        error = response.get("Error")
        if isinstance(error, dict) and error.get("Code") is not None:
            return str(error["Code"])
    return None


def s3_operation_error(operation: str, exc: Exception) -> WatermarkError:
    code = s3_error_code(exc)
    suffix = f", code={code}" if code is not None else ""
    return WatermarkError(
        f"{operation} failed ({type(exc).__name__}{suffix})"
    )


def create_s3_client(region: str, max_pool_connections: int) -> Any:
    try:
        session = Session(region_name=region)
        return session.client(
            "s3",
            region_name=region,
            config=Config(
                retries={"total_max_attempts": 10},
                max_pool_connections=max_pool_connections,
            ),
        )
    except Exception as exc:
        raise s3_operation_error("S3 client initialization", exc) from exc


def list_s3_keys(
    s3_client: Any,
    bucket: str,
    object_prefix: str,
    description: str,
) -> List[str]:
    object_keys = []
    try:
        paginator = s3_client.get_paginator("list_objects_v2")
        pages = paginator.paginate(Bucket=bucket, Prefix=object_prefix)
        for page in pages:
            if not isinstance(page, dict):
                raise WatermarkError(
                    f"{description} returned an unexpected S3 page"
                )
            contents = page.get("Contents", [])
            if contents is None:
                contents = []
            if not isinstance(contents, list):
                raise WatermarkError(
                    f"{description} returned unexpected S3 contents"
                )
            for item in contents:
                if not isinstance(item, dict) or not isinstance(
                    item.get("Key"),
                    str,
                ):
                    raise WatermarkError(
                        f"{description} returned an object without a key"
                    )
                object_keys.append(item["Key"])
    except WatermarkError:
        raise
    except Exception as exc:
        raise s3_operation_error(description, exc) from exc
    if len(object_keys) != len(set(object_keys)):
        raise WatermarkError(f"{description} returned duplicate object keys")
    return sorted(object_keys)


def packed_meta_identities(
    s3_client: Any,
    tasks: Sequence[Dict[str, str]],
) -> List[str]:
    identities = []
    seen_task_prefixes = set()
    for task in tasks:
        task_prefix = task["pack_backup_prefix"]
        bucket, object_prefix = parse_s3_prefix_uri(
            task_prefix,
            "packed-backup task prefix",
        )
        meta_prefix = f"{object_prefix}/meta/"
        task_uri = f"s3://{bucket}/{meta_prefix}"
        if task_uri in seen_task_prefixes:
            raise WatermarkError(
                f"Pack tasks contain duplicate prefix {task_uri}"
            )
        seen_task_prefixes.add(task_uri)
        object_keys = list_s3_keys(
            s3_client,
            bucket,
            meta_prefix,
            f"packed meta listing for {task_uri}",
        )
        for object_key in object_keys:
            if not object_key.endswith(".meta"):
                continue
            if not object_key.startswith(meta_prefix):
                raise WatermarkError(
                    f"packed meta listing returned an out-of-prefix object "
                    f"s3://{bucket}/{object_key}"
                )
            relative_key = object_key[len(meta_prefix) :]
            if not relative_key or "/" in relative_key:
                raise WatermarkError(
                    f"packed meta object is not a direct child of {task_uri}: "
                    f"s3://{bucket}/{object_key}"
                )
            identity = relative_key[: -len(".meta")]
            if not identity:
                raise WatermarkError(
                    f"packed meta object has an empty identity: "
                    f"s3://{bucket}/{object_key}"
                )
            identities.append(identity)
    if len(identities) != len(set(identities)):
        raise WatermarkError(
            "packed meta listing contains duplicate identities"
        )
    return sorted(identities)


def map_packed_clusters(
    meta_identities: Sequence[str],
) -> Dict[str, List[str]]:
    mapping: Dict[str, List[str]] = {}
    for identity in meta_identities:
        # Pack defines every meta identity as <cluster_id>_<keyspace_name> and
        # keyspace names cannot contain "_". Split from the right so cluster
        # IDs may contain "_" without requiring the original cluster ID list.
        cluster_id, separator, keyspace_name = identity.rpartition("_")
        if not separator or not cluster_id or not keyspace_name:
            raise WatermarkError(
                f"packed meta identity {identity!r} does not match "
                "<cluster_id>_<keyspace_name>"
            )
        mapping.setdefault(cluster_id, []).append(identity)
    return {
        cluster_id: sorted(identities)
        for cluster_id, identities in mapping.items()
    }


def complete_marker_backup_time(
    metadata: Dict[str, Any],
    marker_uri: str,
) -> Optional[datetime]:
    value = metadata.get(COMPLETE_MARKER_BACKUP_TIME_METADATA)
    if value is None:
        return None
    if not isinstance(value, str):
        raise CompleteMarkerNotVerified(
            f"complete marker has invalid "
            f"{COMPLETE_MARKER_BACKUP_TIME_METADATA!r} metadata: {marker_uri}"
        )
    try:
        parsed = datetime.strptime(value, COMPLETE_MARKER_BACKUP_TIME_FORMAT)
    except ValueError as exc:
        raise CompleteMarkerNotVerified(
            f"complete marker has invalid "
            f"{COMPLETE_MARKER_BACKUP_TIME_METADATA!r} metadata: {marker_uri}"
        ) from exc
    return parsed.replace(tzinfo=timezone.utc)


def verify_complete_markers(
    s3_client: Any,
    out_prefix: str,
    cluster_meta_identities: Dict[str, List[str]],
    export_instance: str,
    backup_time: datetime,
) -> Tuple[int, bool]:
    bucket, object_prefix = parse_s3_prefix_uri(
        out_prefix,
        "--out-prefix",
    )
    verified = 0
    for cluster_id in sorted(cluster_meta_identities):
        for identity in cluster_meta_identities[cluster_id]:
            marker_key = f"{object_prefix}/{identity}/complete"
            marker_uri = f"s3://{bucket}/{marker_key}"
            try:
                response = s3_client.head_object(
                    Bucket=bucket,
                    Key=marker_key,
                )
            except Exception as exc:
                if s3_error_code(exc) in {"404", "NoSuchKey", "NotFound"}:
                    raise CompleteMarkerNotVerified(
                        f"complete marker does not exist: {marker_uri}"
                    ) from exc
                raise s3_operation_error(
                    f"complete marker verification for {marker_uri}",
                    exc,
                ) from exc
            if not isinstance(response, dict):
                raise WatermarkError(
                    f"complete marker verification returned an unexpected "
                    f"response for {marker_uri}"
                )
            metadata = response.get("Metadata")
            if not isinstance(metadata, dict):
                raise CompleteMarkerNotVerified(
                    f"complete marker has no user metadata: {marker_uri}"
                )
            marker_backup_time = complete_marker_backup_time(
                metadata,
                marker_uri,
            )
            if marker_backup_time is None:
                raise CompleteMarkerNotVerified(
                    f"complete marker has no "
                    f"{COMPLETE_MARKER_BACKUP_TIME_METADATA!r} metadata: "
                    f"{marker_uri}"
                )
            if marker_backup_time > backup_time:
                log(
                    f"skipping newer complete marker: marker={marker_uri} "
                    f"marker_backup_time="
                    f"{marker_backup_time.strftime(COMPLETE_MARKER_BACKUP_TIME_FORMAT)} "
                    f"current_backup_time="
                    f"{backup_time.strftime(COMPLETE_MARKER_BACKUP_TIME_FORMAT)}"
                )
                return verified, True
            if marker_backup_time < backup_time:
                raise CompleteMarkerNotVerified(
                    f"complete marker belongs to an earlier backup: "
                    f"{marker_uri}"
                )
            if (
                metadata.get(COMPLETE_MARKER_JOB_INSTANCE_METADATA)
                != export_instance
            ):
                raise CompleteMarkerNotVerified(
                    f"complete marker job instance does not match "
                    f"{export_instance!r}: {marker_uri}"
                )
            verified += 1
    return verified, False


def mysql_operation_error(operation: str, exc: Exception) -> WatermarkError:
    return WatermarkError(f"{operation} failed ({type(exc).__name__})")


def quietly_rollback_mysql(connection: Any) -> None:
    try:
        connection.rollback()
    except Exception:
        pass


def quietly_close_mysql(connection: Any) -> None:
    try:
        connection.close()
    except Exception:
        pass


def open_mysql_connection(defaults_file: str) -> Any:
    connection = None
    try:
        connection = pymysql.connect(
            read_default_file=str(Path(defaults_file).expanduser()),
            read_default_group="client",
            connect_timeout=10,
            read_timeout=30,
            write_timeout=30,
            autocommit=False,
            charset="utf8mb4",
        )
        with connection.cursor() as cursor:
            cursor.execute("SET time_zone = '+00:00'")
            cursor.execute("SELECT 1")
        connection.rollback()
        return connection
    except Exception as exc:
        if connection is not None:
            quietly_close_mysql(connection)
        raise mysql_operation_error("MySQL connection", exc) from exc


def read_cluster_export_watermark(
    connection: Any,
    cluster_id: str,
) -> Optional[datetime]:
    query = (
        f"SELECT s3_data_updated_at FROM {ACTIVE_CLUSTER_IDS_TABLE} "
        "WHERE cluster_id = %s"
    )
    try:
        with connection.cursor() as cursor:
            cursor.execute(query, (cluster_id,))
            rows = cursor.fetchall()
        connection.rollback()
    except Exception as exc:
        quietly_rollback_mysql(connection)
        raise mysql_operation_error(
            "active cluster watermark read",
            exc,
        ) from exc
    if len(rows) != 1:
        raise WatermarkError(
            f"active cluster watermark read expected one row for "
            f"cluster {cluster_id!r}; found {len(rows)}"
        )
    value = rows[0][0]
    if value is None:
        return None
    if not isinstance(value, datetime):
        raise WatermarkError(
            f"active cluster watermark has an unexpected type for "
            f"cluster {cluster_id!r}"
        )
    if value.tzinfo is not None:
        value = value.astimezone(timezone.utc).replace(tzinfo=None)
    return value


def update_cluster_export_watermark(
    connection: Any,
    cluster_id: str,
    backup_time: datetime,
) -> bool:
    """Return True when MySQL already contains a newer watermark."""
    cutoff = backup_time.astimezone(timezone.utc).replace(tzinfo=None)
    query = (
        f"UPDATE {ACTIVE_CLUSTER_IDS_TABLE} "
        "SET s3_data_updated_at = %s "
        "WHERE cluster_id = %s "
        "AND (s3_data_updated_at IS NULL OR s3_data_updated_at < %s)"
    )
    try:
        with connection.cursor() as cursor:
            cursor.execute(query, (cutoff, cluster_id, cutoff))
            affected_rows = cursor.rowcount
        connection.commit()
    except Exception as exc:
        quietly_rollback_mysql(connection)
        raise mysql_operation_error(
            "active cluster watermark update",
            exc,
        ) from exc

    if affected_rows == 1:
        return False
    observed = read_cluster_export_watermark(connection, cluster_id)
    if observed is not None and observed >= cutoff:
        return observed > cutoff
    raise WatermarkError(
        f"active cluster watermark update changed no row for "
        f"cluster {cluster_id!r}"
    )


def add_cluster_to_blocklist(
    connection: Any,
    cluster_id: str,
    export_instance: str,
    backup_time: datetime,
) -> None:
    description = (
        f"export_instance={export_instance}, "
        f"backup_time={backup_time.strftime(COMPLETE_MARKER_BACKUP_TIME_FORMAT)}"
    )
    query = (
        f"INSERT IGNORE INTO {CLUSTER_BLOCKLIST_TABLE} "
        "(cluster_id, used_by, description) VALUES (%s, %s, %s)"
    )
    try:
        with connection.cursor() as cursor:
            cursor.execute(
                query,
                (cluster_id, CLUSTER_BLOCKLIST_USED_BY, description),
            )
        connection.commit()
    except Exception as exc:
        quietly_rollback_mysql(connection)
        raise mysql_operation_error(
            "cluster blocklist update",
            exc,
        ) from exc


def advance_verified_cluster_watermark(
    s3_client: Any,
    out_prefix: str,
    cluster_id: str,
    identities: List[str],
    export_instance: str,
    mysql_connection: Callable[[], Any],
    backup_time: datetime,
) -> WatermarkAdvanceSummary:
    try:
        marker_count, has_newer_backup = verify_complete_markers(
            s3_client,
            out_prefix,
            {cluster_id: identities},
            export_instance,
            backup_time,
        )
    except CompleteMarkerNotVerified as exc:
        try:
            add_cluster_to_blocklist(
                mysql_connection(),
                cluster_id,
                export_instance,
                backup_time,
            )
        except WatermarkError as mysql_exc:
            return WatermarkAdvanceSummary(
                0,
                0,
                0,
                0,
                1,
                1,
                str(mysql_exc),
            )
        log(
            "warning: blocklisted and skipped MySQL watermark update for "
            f"cluster {cluster_id!r}: {exc}"
        )
        return WatermarkAdvanceSummary(0, 0, 0, 0, 1, 0, None)

    if has_newer_backup:
        return WatermarkAdvanceSummary(
            marker_count,
            0,
            1,
            0,
            0,
            0,
            None,
        )

    try:
        has_newer_watermark = update_cluster_export_watermark(
            mysql_connection(),
            cluster_id,
            backup_time,
        )
    except WatermarkError as exc:
        return WatermarkAdvanceSummary(
            marker_count,
            0,
            0,
            0,
            0,
            1,
            str(exc),
        )

    return WatermarkAdvanceSummary(
        marker_count,
        0 if has_newer_watermark else 1,
        0,
        1 if has_newer_watermark else 0,
        0,
        0,
        None,
    )


def advance_verified_cluster_watermarks(
    s3_client: Any,
    out_prefix: str,
    cluster_meta_identities: Dict[str, List[str]],
    export_instance: str,
    defaults_file: str,
    backup_time: datetime,
    parallel: int,
) -> WatermarkAdvanceSummary:
    verified_markers = 0
    updated_clusters = 0
    newer_backup_clusters = 0
    newer_mysql_watermark_clusters = 0
    unverified_clusters = 0
    update_failed_clusters = 0
    first_error = None
    if not cluster_meta_identities:
        return WatermarkAdvanceSummary(0, 0, 0, 0, 0, 0, None)

    worker_state = threading.local()
    connections: List[Any] = []
    connections_lock = threading.Lock()

    def mysql_connection() -> Any:
        connection = getattr(worker_state, "mysql_connection", None)
        if connection is not None:
            return connection
        connection_error = getattr(worker_state, "mysql_connection_error", None)
        if connection_error is not None:
            raise WatermarkError(connection_error)
        try:
            connection = open_mysql_connection(defaults_file)
        except WatermarkError as exc:
            worker_state.mysql_connection_error = str(exc)
            raise
        worker_state.mysql_connection = connection
        with connections_lock:
            connections.append(connection)
        return connection

    cluster_iterator = iter(sorted(cluster_meta_identities.items()))
    in_flight: Dict[concurrent.futures.Future, str] = {}
    first_error_cluster = None

    def fill_workers(executor: concurrent.futures.ThreadPoolExecutor) -> None:
        while len(in_flight) < parallel:
            try:
                cluster_id, identities = next(cluster_iterator)
            except StopIteration:
                return
            future = executor.submit(
                advance_verified_cluster_watermark,
                s3_client,
                out_prefix,
                cluster_id,
                identities,
                export_instance,
                mysql_connection,
                backup_time,
            )
            in_flight[future] = cluster_id

    try:
        with concurrent.futures.ThreadPoolExecutor(
            max_workers=parallel
        ) as executor:
            fill_workers(executor)
            while in_flight:
                done, _ = concurrent.futures.wait(
                    tuple(in_flight),
                    return_when=concurrent.futures.FIRST_COMPLETED,
                )
                for future in sorted(done, key=lambda item: in_flight[item]):
                    cluster_id = in_flight.pop(future)
                    advance = future.result()
                    verified_markers += advance.verified_markers
                    updated_clusters += advance.updated_clusters
                    newer_backup_clusters += advance.newer_backup_clusters
                    newer_mysql_watermark_clusters += (
                        advance.newer_mysql_watermark_clusters
                    )
                    unverified_clusters += advance.unverified_clusters
                    update_failed_clusters += advance.update_failed_clusters
                    if advance.first_error is not None and (
                        first_error_cluster is None
                        or cluster_id < first_error_cluster
                    ):
                        first_error_cluster = cluster_id
                        first_error = advance.first_error
                fill_workers(executor)
    finally:
        for connection in connections:
            quietly_close_mysql(connection)

    return WatermarkAdvanceSummary(
        verified_markers=verified_markers,
        updated_clusters=updated_clusters,
        newer_backup_clusters=newer_backup_clusters,
        newer_mysql_watermark_clusters=newer_mysql_watermark_clusters,
        unverified_clusters=unverified_clusters,
        update_failed_clusters=update_failed_clusters,
        first_error=first_error,
    )


def reconcile(args: argparse.Namespace, result: Dict[str, Any]) -> None:
    tasks = load_tasks(args.tasks_json)
    s3_client = create_s3_client(args.aws_region, args.parallel)
    meta_identities = packed_meta_identities(s3_client, tasks)
    cluster_meta_identities = map_packed_clusters(meta_identities)
    result.update(
        {
            "packed_clusters": len(cluster_meta_identities),
            "packed_meta_objects": len(meta_identities),
        }
    )

    log(
        f"validated {len(meta_identities)} packed meta object(s) for "
        f"{len(cluster_meta_identities)} cluster(s)"
    )

    advance = advance_verified_cluster_watermarks(
        s3_client,
        args.out_prefix,
        cluster_meta_identities,
        args.export_instance,
        args.mysql_defaults_file,
        args.backup_time,
        args.parallel,
    )
    result.update(
        {
            "verified_complete_markers": advance.verified_markers,
            "unverified_complete_marker_clusters": (
                advance.unverified_clusters
            ),
            "mysql_updated_clusters": advance.updated_clusters,
            "newer_complete_marker_clusters": (
                advance.newer_backup_clusters
            ),
            "newer_mysql_watermark_clusters": (
                advance.newer_mysql_watermark_clusters
            ),
            "mysql_update_failed_clusters": (
                advance.update_failed_clusters
            ),
        }
    )
    log(
        f"verified {advance.verified_markers} complete marker(s); "
        f"updated {advance.updated_clusters} and skipped "
        f"{advance.newer_backup_clusters} newer-marker and "
        f"{advance.newer_mysql_watermark_clusters} newer-MySQL cluster(s) "
        "out of "
        f"{len(cluster_meta_identities)} MySQL watermark(s); "
        f"unverified={advance.unverified_clusters} "
        f"update_failed={advance.update_failed_clusters}"
    )

    if advance.unverified_clusters:
        log(
            "warning: left MySQL watermarks unchanged for "
            f"{advance.unverified_clusters} cluster(s) with missing or "
            "mismatched complete markers"
        )

    if advance.update_failed_clusters:
        raise WatermarkError(
            f"{advance.update_failed_clusters} cluster(s) failed to update "
            f"their MySQL reconciliation state; first error: "
            f"{advance.first_error}"
        )


def serialize_result(result: Dict[str, Any]) -> Tuple[str, bool]:
    payload = json.dumps(
        result,
        ensure_ascii=False,
        sort_keys=True,
        separators=(",", ":"),
    )
    if len(payload.encode("utf-8")) <= TERMINATION_MESSAGE_MAX_BYTES:
        return payload, False

    compact_result = {
        field: result[field]
        for field in TERMINATION_RESULT_FIELDS
    }
    compact_result["status"] = "failed"
    compact_result["error"] = (
        "Watermark result exceeded the termination message size limit"
    )
    compact_payload = json.dumps(
        compact_result,
        ensure_ascii=False,
        sort_keys=True,
        separators=(",", ":"),
    )
    if len(compact_payload.encode("utf-8")) > TERMINATION_MESSAGE_MAX_BYTES:
        raise RuntimeError("compact Watermark result exceeds size limit")
    return compact_payload, True


def emit_result(result: Dict[str, Any]) -> bool:
    payload, replaced = serialize_result(result)
    if replaced:
        log(
            "error: Watermark result exceeded the termination message "
            "size limit; wrote a compact failure result"
        )
    TERMINATION_MESSAGE_PATH.write_text(payload, encoding="utf-8")
    return not replaced


def main(argv: Optional[Sequence[str]] = None) -> int:
    args = build_parser().parse_args(argv)
    backup_time = args.backup_time.strftime(
        COMPLETE_MARKER_BACKUP_TIME_FORMAT
    )
    result: Dict[str, Any] = {
        "status": "running",
        "watermark_instance": args.watermark_instance,
        "export_instance": args.export_instance,
        "backup_time": backup_time,
        "packed_clusters": 0,
        "packed_meta_objects": 0,
        "verified_complete_markers": 0,
        "unverified_complete_marker_clusters": 0,
        "mysql_updated_clusters": 0,
        "newer_complete_marker_clusters": 0,
        "newer_mysql_watermark_clusters": 0,
        "mysql_update_failed_clusters": 0,
    }
    try:
        reconcile(args, result)
        result["status"] = "succeeded"
        return 0 if emit_result(result) else 1
    except WatermarkError as exc:
        result["status"] = "failed"
        result["error"] = str(exc)
        log(f"error: {exc}")
        emit_result(result)
        return 1
    except Exception as exc:
        result["status"] = "failed"
        result["error"] = f"unexpected failure ({type(exc).__name__})"
        log(f"error: {result['error']}")
        emit_result(result)
        return 1
    except KeyboardInterrupt:
        result["status"] = "failed"
        result["error"] = "interrupted"
        log("error: interrupted")
        emit_result(result)
        return 130


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