#!/usr/bin/env python3 from __future__ import annotations import argparse import json import os import re import signal import ssl import sys import time from contextlib import contextmanager from dataclasses import dataclass from datetime import datetime, timedelta, timezone from pathlib import Path from typing import Any import pymysql ROOT = Path(__file__).resolve().parents[1] DEPLOYMENT_CONFIG = ROOT / "deployments" / "deployment-config.json" CONNECTOR_EVENT_TABLE = "_System_ConnectorEvent" DEFAULT_NAMESPACE = "hubspot-crm-connector" DEFAULT_CONNECTOR_OWNER_KEY = "pingcap-hubspot-companion" DEFAULT_EXPECTED_CLAIMED_BY_RE = ( r"^connector-inbound-hubspot-crm-connector-" r"ip-172-31-33-30\.us-west-2\.compute\.internal-\d+#" ) DEFAULT_EXPECTED_DB_CLIENT_HOST_RE = r"^52\.32\.199\.98$" TERMINAL_EVENT_STATUSES = frozenset({"processed", "skipped", "dead"}) TIMESTAMP_CANDIDATES = ( "ProcessedAt", "UpdatedAt", "LastModifiedDate", "LastErrorAt", "ClaimedAt", "CreatedDate", "CreatedAt", ) SAMPLE_COLUMNS = ( "Id", "Status", "ClaimedBy", "SourceObjectID", "SourceObjectId", "SourceObjectKind", "EventType", "SubscriptionType", "ProcessedAt", "UpdatedAt", "LastModifiedDate", "LastErrorAt", "ClaimedAt", "CreatedDate", "CreatedAt", "LastError", ) @dataclass(frozen=True) class DbTarget: label: str host: str port: int user: str password: str database: str def parse_args(argv: list[str] | None = None) -> argparse.Namespace: parser = argparse.ArgumentParser( description=( "Fail fast when the HubSpot connector queue is already busy, events are claimed by non-runtime " "workers, or processing leases are stale." ) ) parser.add_argument("--target-env", required=True, choices=["staging", "test1", "test2", "test3"]) parser.add_argument( "--slot", default=os.environ.get("HUBSPOT_CONNECTOR_PREFLIGHT_SLOT", ""), help=( "For staging, optionally select one blue/green slot. If omitted, all configured staging targets are checked." ), ) parser.add_argument("--namespace", default=os.environ.get("HUBSPOT_CONNECTOR_NAMESPACE") or DEFAULT_NAMESPACE) parser.add_argument("--connector-owner-key", default=os.environ.get("HUBSPOT_CONNECTOR_OWNER_KEY") or DEFAULT_CONNECTOR_OWNER_KEY) parser.add_argument("--require-idle-pipeline", action="store_true", help="Require one bounded, empty snapshot of the full durable HubSpot pipeline; never wait or retry.") parser.add_argument("--table", default=os.environ.get("HUBSPOT_CONNECTOR_EVENT_TABLE") or CONNECTOR_EVENT_TABLE) parser.add_argument( "--expected-claimed-by-regex", default=os.environ.get("HUBSPOT_CONNECTOR_EXPECTED_CLAIMED_BY_REGEX") or DEFAULT_EXPECTED_CLAIMED_BY_RE, help="Regex that identifies the current staging/test runtime worker identity.", ) parser.add_argument( "--denied-claimed-by-regex", default=os.environ.get("HUBSPOT_CONNECTOR_DENIED_CLAIMED_BY_REGEX"), help=( "Fail-closed regex for known foreign worker identities. This deny-list is checked before " "expected-claimed-by, so a broad override cannot bless a known prod-host worker. Defaults to " "deployments/deployment-config.json hubspot.connector_worker.denied_claimed_by_needles." ), ) parser.add_argument( "--expected-db-client-host-regex", default=os.environ.get("HUBSPOT_CONNECTOR_EXPECTED_DB_CLIENT_HOST_REGEX") or DEFAULT_EXPECTED_DB_CLIENT_HOST_RE, help=( "Regex for TiDB client hosts allowed to hold extra sessions for the target DB user. " "The preflight connection itself is always ignored." ), ) parser.add_argument( "--lookback-minutes", type=int, default=int(os.environ.get("HUBSPOT_CONNECTOR_PREFLIGHT_LOOKBACK_MINUTES", "240")), ) parser.add_argument( "--stale-processing-minutes", type=int, default=int(os.environ.get("HUBSPOT_CONNECTOR_STALE_PROCESSING_MINUTES", "30")), ) parser.add_argument( "--recent-terminal-foreign-minutes", type=int, default=int(os.environ.get("HUBSPOT_CONNECTOR_RECENT_TERMINAL_FOREIGN_MINUTES", "30")), help=( "Terminal-status events claimed by a foreign identity still BLOCK when their latest " "timestamp is within this many minutes (a hit-and-run foreign process leaves fresh " "terminal footprints). Older terminal foreign rows are reported as history only, so a " "reset-from-prod snapshot stops hard-blocking the suite once its cloned timestamps age " "past this window (issue #4100)." ), ) parser.add_argument( "--parked-nonterminal-minutes", type=int, default=int(os.environ.get("HUBSPOT_CONNECTOR_PARKED_NONTERMINAL_MINUTES", "30")), help=( "A nonterminal event is treated as PARKED backlog (reported, not blocking) when it is not " "'processing', carries no ClaimedBy, and its latest timestamp is older than this many minutes. " "Parked rows are inert leftovers (e.g. re-driven dead events without a scheduler pickup) that " "would otherwise hard-block every suite run against max-nonterminal-events=0 (issue #4402). " "Fresh nonterminal activity, live claims, and 'processing' rows always count as active." ), ) parser.add_argument( "--max-nonterminal-events", type=int, default=int(os.environ.get("HUBSPOT_CONNECTOR_PREFLIGHT_MAX_NONTERMINAL_EVENTS", "0")), help=( "Maximum pending, processing, failed/retryable, or unknown connector events allowed before a test run. " "Defaults to zero so shared staging pressure is rejected before it can create false case failures." ), ) parser.add_argument("--max-samples", type=int, default=10) parser.add_argument("--json-output", help="Optional path for a machine-readable diagnostic artifact.") args = parser.parse_args(argv) if not args.namespace.strip() or not args.connector_owner_key.strip(): parser.error("namespace and connector owner key must be nonempty") if args.parked_nonterminal_minutes < 0: # A negative window would classify every nonterminal event as parked # and silently disable the queue-drain gate. Fail closed on both CLI # and env sources. parser.error("--parked-nonterminal-minutes must be >= 0") if args.recent_terminal_foreign_minutes < 0: # A negative window would place the cutoff in the future and classify # fresh foreign terminal footprints as non-blocking history, silently # disabling the hit-and-run safeguard. Fail closed on both CLI and env # sources (the env default bypasses argparse type checks). parser.error("--recent-terminal-foreign-minutes must be >= 0") return args def load_deployment_config() -> dict[str, Any]: return json.loads(DEPLOYMENT_CONFIG.read_text()) def target_config_for(config: dict[str, Any], target_env: str, slot: str) -> list[tuple[str, dict[str, Any]]]: if target_env == "staging": staging_topology = config["environments"]["staging"]["topology"] staging = staging_topology["target_config"] targets = [str(item).strip() for item in staging_topology.get("targets", []) if str(item).strip()] if slot: if slot not in staging: raise ValueError(f"unknown staging target: {slot}") return [(slot, staging[slot])] return [(name, staging[name]) for name in targets] test_targets = config["environments"]["test"]["topology"]["target_config"] if target_env not in test_targets: raise ValueError(f"unknown test target: {target_env}") return [(target_env, test_targets[target_env])] def environment_key_for_target(target_env: str) -> str: return "staging" if target_env == "staging" else "test" def denied_claimed_by_regex_for_config(config: dict[str, Any], target_env: str) -> str: env_key = environment_key_for_target(target_env) worker_config = ( config.get("environments", {}) .get(env_key, {}) .get("hubspot", {}) .get("connector_worker", {}) ) needles = [ str(item).strip() for item in worker_config.get("denied_claimed_by_needles", []) if str(item).strip() ] return "|".join(re.escape(item) for item in needles) def resolve_targets(target_env: str, slot: str, config: dict[str, Any] | None = None) -> list[DbTarget]: config = config or load_deployment_config() targets: list[DbTarget] = [] for label, item in target_config_for(config, target_env, slot): tidb = item["tidb"] password_env = str(tidb["password_env"]).strip() password = os.environ.get(password_env, "") if not password: raise RuntimeError(f"missing required TiDB password env for {target_env}/{label}: {password_env}") targets.append( DbTarget( label=f"{target_env}/{label}", host=str(tidb["host"]).strip(), port=int(tidb["port"]), user=str(tidb["user"]).strip(), password=password, database=os.environ.get("HUBSPOT_CONNECTOR_PREFLIGHT_DATABASE", "nexuscrm").strip() or "nexuscrm", ) ) return targets def connect(target: DbTarget) -> pymysql.connections.Connection: return pymysql.connect( host=target.host, port=target.port, user=target.user, password=target.password, database=target.database, autocommit=True, connect_timeout=10, read_timeout=30, write_timeout=30, ssl=ssl.create_default_context(), cursorclass=pymysql.cursors.DictCursor, ) def quote_ident(name: str) -> str: return "`" + name.replace("`", "``") + "`" def column_map(conn: pymysql.connections.Connection, table: str) -> dict[str, str]: with conn.cursor() as cur: cur.execute(f"SHOW COLUMNS FROM {quote_ident(table)}") rows = cur.fetchall() if not rows: raise RuntimeError(f"table has no columns: {table}") return {str(row["Field"]).lower(): str(row["Field"]) for row in rows} def col(columns: dict[str, str], name: str) -> str | None: return columns.get(name.lower()) def required_col(columns: dict[str, str], name: str) -> str: value = col(columns, name) if not value: raise RuntimeError(f"{CONNECTOR_EVENT_TABLE} is missing required column: {name}") return value def namespace_where(columns: dict[str, str], namespace: str, params: list[Any]) -> list[str]: namespace_col = col(columns, "Namespace") if not namespace_col: return [] params.append(namespace) return [f"{quote_ident(namespace_col)} = %s"] def recent_or_processing_where( columns: dict[str, str], cutoff: datetime, now: datetime, params: list[Any], ) -> list[str]: status_col = required_col(columns, "Status") predicates = [f"LOWER(COALESCE({quote_ident(status_col)}, '')) = 'processing'"] upper_bound = now + timedelta(minutes=5) timestamp_cols = [] for candidate in TIMESTAMP_CANDIDATES: actual = col(columns, candidate) if actual: timestamp_cols.append(actual) predicates.append(f"({quote_ident(actual)} >= %s AND {quote_ident(actual)} <= %s)") params.append(cutoff.strftime("%Y-%m-%d %H:%M:%S")) params.append(upper_bound.strftime("%Y-%m-%d %H:%M:%S")) if timestamp_cols: # A row with every timestamp NULL can never satisfy a window predicate, # so it would otherwise be invisible to the scan — and the classifier's # fail-closed "missing timestamp counts as recent" branch could never # fire. Fetch such rows explicitly. predicates.append("(" + " AND ".join(f"{quote_ident(item)} IS NULL" for item in timestamp_cols) + ")") return ["(" + " OR ".join(predicates) + ")"] def row_time(row: dict[str, Any]) -> str: for key in TIMESTAMP_CANDIDATES: value = row.get(key) if value: return str(value) return "" def row_datetime(row: dict[str, Any]) -> datetime | None: """Latest TIMESTAMP_CANDIDATES value as a naive-UTC datetime, or None. The maximum (not the first) is used so an event whose terminalization is recorded on a later column (e.g. dead-lettering stamps UpdatedAt while ProcessedAt stays old or NULL) still counts as recent activity. """ latest: datetime | None = None for key in TIMESTAMP_CANDIDATES: value = row.get(key) if isinstance(value, datetime): if value.tzinfo is not None: value = value.astimezone(timezone.utc).replace(tzinfo=None) if latest is None or value > latest: latest = value return latest def compact_row(row: dict[str, Any]) -> dict[str, Any]: out: dict[str, Any] = {} for key in ( "Id", "Status", "ClaimedBy", "SourceObjectID", "SourceObjectId", "SourceObjectKind", "EventType", "SubscriptionType", "ProcessedAt", "UpdatedAt", "ClaimedAt", "CreatedDate", "CreatedAt", ): value = row.get(key) if value not in (None, ""): out[key] = str(value) last_error = str(row.get("LastError") or "") if last_error: out["LastError"] = last_error[:240] return out def record_foreign_worker(foreign: dict[str, dict[str, Any]], row: dict[str, Any], claimed_by: str, status: str) -> None: item = foreign.setdefault( claimed_by, {"claimedBy": claimed_by, "count": 0, "statuses": {}, "latestTime": "", "sourceObjectIDs": []}, ) item["count"] += 1 item["statuses"][status or "unknown"] = item["statuses"].get(status or "unknown", 0) + 1 event_time = row_time(row) if event_time and event_time > item["latestTime"]: item["latestTime"] = event_time source_id = row.get("SourceObjectID") or row.get("SourceObjectId") if source_id and str(source_id) not in item["sourceObjectIDs"] and len(item["sourceObjectIDs"]) < 10: item["sourceObjectIDs"].append(str(source_id)) def classify_rows( rows: list[dict[str, Any]], *, expected_claimed_by: re.Pattern[str], denied_claimed_by: re.Pattern[str] | None = None, stale_cutoff: datetime, recent_terminal_cutoff: datetime | None = None, ) -> dict[str, Any]: # Foreign ClaimedBy values are split by event status and freshness: # - non-terminal claims (pending/processing) prove a live foreign worker # and always block; # - terminal rows (processed/skipped/dead) newer than # recent_terminal_cutoff also block — a hit-and-run foreign process # leaves exactly this fresh terminal footprint after exiting; # - older terminal foreign rows are history — most commonly prod terminal # events copied in by a reset-from-prod-snapshot staging deploy — and # are reported as information only, so a snapshot reset stops blocking # the suite once its cloned timestamps age past the window (issue #4100). # A terminal foreign row without any parsable timestamp is treated as # recent (fail closed). foreign_live: dict[str, dict[str, Any]] = {} foreign_recent_terminal: dict[str, dict[str, Any]] = {} foreign_terminal: dict[str, dict[str, Any]] = {} allowed: dict[str, int] = {} stale_processing: list[dict[str, Any]] = [] recent_cutoff_naive = None if recent_terminal_cutoff is not None: recent_cutoff_naive = recent_terminal_cutoff if recent_cutoff_naive.tzinfo is not None: recent_cutoff_naive = recent_cutoff_naive.astimezone(timezone.utc).replace(tzinfo=None) for row in rows: claimed_by = str(row.get("ClaimedBy") or "") status = str(row.get("Status") or "").lower() if status in TERMINAL_EVENT_STATUSES: if recent_cutoff_naive is None: foreign = foreign_terminal else: event_time = row_datetime(row) is_recent = event_time is None or event_time >= recent_cutoff_naive foreign = foreign_recent_terminal if is_recent else foreign_terminal else: foreign = foreign_live if claimed_by: if denied_claimed_by and denied_claimed_by.search(claimed_by): record_foreign_worker(foreign, row, claimed_by, status) elif expected_claimed_by.search(claimed_by): allowed[claimed_by] = allowed.get(claimed_by, 0) + 1 else: record_foreign_worker(foreign, row, claimed_by, status) if status == "processing": timestamp_value = None for key in ("ClaimedAt", "UpdatedAt", "CreatedDate", "CreatedAt", "ProcessedAt"): value = row.get(key) if isinstance(value, datetime): timestamp_value = value break if timestamp_value is None: stale_processing.append(row) else: if timestamp_value.tzinfo is not None: timestamp_value = timestamp_value.astimezone(timezone.utc).replace(tzinfo=None) if timestamp_value < stale_cutoff.replace(tzinfo=None): stale_processing.append(row) return { "foreignWorkers": sorted(foreign_live.values(), key=lambda item: (-item["count"], item["claimedBy"])), "foreignRecentTerminalWorkers": sorted( foreign_recent_terminal.values(), key=lambda item: (-item["count"], item["claimedBy"]) ), "foreignTerminalWorkers": sorted( foreign_terminal.values(), key=lambda item: (-item["count"], item["claimedBy"]) ), "allowedWorkers": [{"claimedBy": key, "count": value} for key, value in sorted(allowed.items())], "staleProcessingRows": [compact_row(row) for row in stale_processing], } def client_host(host: Any) -> str: return str(host or "").split(":", 1)[0].strip() def classify_db_clients( rows: list[dict[str, Any]], *, expected_client_host: re.Pattern[str], current_connection_id: int, ) -> dict[str, Any]: allowed: dict[str, int] = {} foreign: dict[str, dict[str, Any]] = {} for row in rows: try: connection_id = int(row.get("ID") or row.get("Id") or 0) except (TypeError, ValueError): connection_id = 0 if connection_id == current_connection_id: continue host = client_host(row.get("HOST") or row.get("Host")) if not host: continue if expected_client_host.search(host): allowed[host] = allowed.get(host, 0) + 1 continue item = foreign.setdefault( host, {"host": host, "count": 0, "commands": {}, "maxTimeSeconds": 0}, ) item["count"] += 1 command = str(row.get("COMMAND") or row.get("Command") or "unknown") item["commands"][command] = item["commands"].get(command, 0) + 1 try: item["maxTimeSeconds"] = max(item["maxTimeSeconds"], int(row.get("TIME") or row.get("Time") or 0)) except (TypeError, ValueError): pass return { "foreignDbClients": sorted(foreign.values(), key=lambda item: (-item["count"], item["host"])), "allowedDbClients": [{"host": key, "count": value} for key, value in sorted(allowed.items())], } def fetch_status_counts(conn: pymysql.connections.Connection, table: str, columns: dict[str, str], namespace: str) -> list[dict[str, Any]]: status_col = required_col(columns, "Status") params: list[Any] = [] where = namespace_where(columns, namespace, params) where_sql = f" WHERE {' AND '.join(where)}" if where else "" sql = ( f"SELECT COALESCE({quote_ident(status_col)}, '') AS status, COUNT(*) AS count " f"FROM {quote_ident(table)}{where_sql} GROUP BY status ORDER BY count DESC" ) with conn.cursor() as cur: cur.execute(sql, params) return [{"status": str(row["status"] or "unknown"), "count": int(row["count"])} for row in cur.fetchall()] def nonterminal_event_count(status_counts: list[dict[str, Any]]) -> int: return sum( int(item.get("count") or 0) for item in status_counts if str(item.get("status") or "unknown").strip().lower() not in TERMINAL_EVENT_STATUSES ) def active_nonterminal_predicates( columns: dict[str, str], parked_cutoff: datetime, params: list[Any], ) -> str: """Predicate matching ACTIVE nonterminal rows (blocking). Active means any of: status 'processing', a live ClaimedBy, any timestamp at/after the parked cutoff, or no parsable timestamp at all (fail closed). Everything else nonterminal is parked backlog: unclaimed, not processing, and untouched for longer than the parked window — inert leftovers that must not hard-block the suite (issue #4402). """ status_col = required_col(columns, "Status") predicates = [f"LOWER(COALESCE({quote_ident(status_col)}, '')) = 'processing'"] claimed_col = col(columns, "ClaimedBy") if claimed_col: predicates.append(f"COALESCE({quote_ident(claimed_col)}, '') <> ''") timestamp_cols = [actual for name in TIMESTAMP_CANDIDATES if (actual := col(columns, name))] if not timestamp_cols: # Without any timestamp column, age cannot be established, so no row # may be classified as parked: treat every nonterminal row as active # (fail closed). return "(1=1)" for actual in timestamp_cols: # COALESCE keeps the OR-chain NULL-safe: a NULL timestamp must read as # "not fresh" (FALSE), not as SQL NULL, or NOT(active) would drop # mixed-NULL parked rows from the parked sample query. predicates.append(f"COALESCE({quote_ident(actual)}, '1970-01-01 00:00:00') >= %s") params.append(parked_cutoff.strftime("%Y-%m-%d %H:%M:%S")) predicates.append("(" + " AND ".join(f"{quote_ident(item)} IS NULL" for item in timestamp_cols) + ")") return "(" + " OR ".join(predicates) + ")" def fetch_nonterminal_split( conn: pymysql.connections.Connection, table: str, columns: dict[str, str], namespace: str, parked_cutoff: datetime, max_samples: int, workitem_coverage: dict[str, str] | None = None, ) -> dict[str, Any]: status_col = required_col(columns, "Status") terminal_list = ", ".join(f"'{item}'" for item in sorted(TERMINAL_EVENT_STATUSES)) nonterminal_sql = f"LOWER(COALESCE({quote_ident(status_col)}, '')) NOT IN ({terminal_list})" def active_with_coverage(params_out: list[Any]) -> str: if workitem_coverage: # Canonical drain contract (connector_inbound_legacy_event_gap_ # summary_for_packs, common.py): a nonterminal Event is UNCOVERED — # and therefore blocking — only when it has neither a direct # WorkItem nor a source-projection WorkItem whose selected boundary # is at/after it. Once WorkItem exists, Event.Status, ClaimedBy and # timestamps are immutable admission history rather than queue # lifecycle authority. A covered fresh Event must therefore not be # reclassified as active; nonterminal covered work is checked below # through WorkItem.State. The projection subquery is the canonical # fragment verbatim (fixed split-era column names; a missing column # fails the preflight closed rather than mis-classifying). event_pk = col(columns, "Id") if event_pk: wi = quote_ident(workitem_coverage["table"]) t = quote_ident(table) pk = quote_ident(event_pk) uncovered = f"""( NOT EXISTS ( SELECT 1 FROM {wi} direct_work WHERE direct_work.EventId = {t}.{pk} AND direct_work.WorkType IN ('connector_event', 'workflow_action_apply') ) AND NOT EXISTS ( SELECT 1 FROM {wi} projection JOIN {t} selected ON selected.Id = projection.EventId WHERE projection.InstalledPackId = {t}.InstalledPackId AND projection.WorkType IN ('connector_event', 'workflow_action_apply') AND ( NULLIF(COALESCE(projection.ProjectionKey, ''), '') IS NOT NULL OR ( NULLIF(COALESCE(projection.ProjectionKey, ''), '') IS NULL AND projection.WorkId LIKE 'csp:%%' AND projection.State NOT IN ('processed', 'skipped', 'dead') ) ) AND COALESCE(projection.SourceLaneKey, '') = COALESCE({t}.SourceLaneKey, '') AND selected.InstalledPackId = {t}.InstalledPackId AND LOWER(TRIM(selected.EventKind)) = LOWER(TRIM({t}.EventKind)) AND COALESCE(selected.RouteVersion, 0) = COALESCE({t}.RouteVersion, 0) AND COALESCE(selected.CompanionPackInstallId, '') = COALESCE({t}.CompanionPackInstallId, '') AND COALESCE(selected.CompanionFlowVersionId, '') = COALESCE({t}.CompanionFlowVersionId, '') AND ( ( {t}.EnqueueOrdinal <> 0 AND selected.EnqueueOrdinal <> 0 AND ( {t}.EnqueueOrdinal < selected.EnqueueOrdinal OR ({t}.EnqueueOrdinal = selected.EnqueueOrdinal AND {t}.Id <= selected.Id) ) ) OR ( ({t}.EnqueueOrdinal = 0 OR selected.EnqueueOrdinal = 0) AND ( {t}.CreatedDate < selected.CreatedDate OR ( {t}.CreatedDate = selected.CreatedDate AND ( {t}.DeliverySequence < selected.DeliverySequence OR ({t}.DeliverySequence = selected.DeliverySequence AND {t}.Id <= selected.Id) ) ) ) ) ) ) )""" return uncovered # Legacy environments without the WorkItem ledger have no split-era # queue authority, so retain the conservative Event freshness/claim # heuristic there. return active_nonterminal_predicates(columns, parked_cutoff, params_out) params: list[Any] = [] where = [*namespace_where(columns, namespace, params), nonterminal_sql] active_sql = active_with_coverage(params) count_sql = ( f"SELECT COUNT(*) AS active_count FROM {quote_ident(table)} " f"WHERE {' AND '.join(where)} AND {active_sql}" ) with conn.cursor() as cur: cur.execute(count_sql, params) active_count = int(cur.fetchone()["active_count"]) selected = [actual for name in SAMPLE_COLUMNS if (actual := col(columns, name))] samples: dict[str, list[dict[str, Any]]] = {"active": [], "parked": []} for kind, extra in (("active", f"AND {{active}}"), ("parked", f"AND NOT {{active}}")): params2: list[Any] = [] where2 = [*namespace_where(columns, namespace, params2), nonterminal_sql] active_sql2 = active_with_coverage(params2) sql = ( f"SELECT {', '.join(quote_ident(item) for item in selected)} " f"FROM {quote_ident(table)} WHERE {' AND '.join(where2)} " + extra.format(active=active_sql2) + " LIMIT %s" ) params2.append(max_samples) with conn.cursor() as cur: cur.execute(sql, params2) samples[kind] = [compact_row(dict(row)) for row in cur.fetchall()] return { "activeCount": active_count, "activeSamples": samples["active"], "parkedSamples": samples["parked"], } CONNECTOR_WORKITEM_TABLE = "_System_ConnectorWorkItem" WORKITEM_SAMPLE_COLUMNS = ( "WorkId", "State", "InstalledPackId", "Generation", "RunAfter", "SourceLaneKey", "TargetMutationLaneKey", "LeaseOwner", "LeaseExpiresAt", "LastError", "CreatedDate", "LastModifiedDate", ) INSTALLED_PACK_TABLE = "_System_InstalledPack" def pipeline_drain_sql(event_table: str) -> str: """One snapshot of every durable owner, including work from earlier runs. WorkItem is the inbound execution authority (Connectors workqueue/stats.go). The owners hash their scopes differently. Join their shared trusted identity tuple plus the handoff's fixed command-acceptance capability, never their same-named ScopeDigest values or CommandId alone. Runtime outbound PENDING and ACCEPTED cover admission and outstanding completion. This query deliberately has no date, current pack, generation or due filter. Missing tables/columns are errors; drain never falls back to Event history. """ return f""" WITH scoped_work AS ( SELECT w.State, w.LeaseUntil, e.Id AS MatchedEventId FROM _System_ConnectorWorkItem w LEFT JOIN {quote_ident(event_table)} e ON e.Id = w.EventId WHERE {PRIOR_WORK_ITEM_SCOPE_SQL} AND (w.State = 'processing' OR w.LeaseUntil > UTC_TIMESTAMP(6) OR w.State IS NULL OR w.State NOT IN ('pending','retry','blocked','blocked_retry','processing','processed','skipped','dead') OR w.WorkId NOT LIKE 'csp:%%' OR NULLIF(COALESCE(w.ProjectionActiveKey, ''), '') IS NOT NULL OR (NULLIF(COALESCE(w.ProjectionKey, ''), '') IS NULL AND w.State NOT IN ('processed', 'skipped', 'dead'))) ), scoped_runtime AS ( SELECT c.State, c.RecoveryState, c.PublicationIdentity, c.ResultPayload FROM _System_RuntimePlatformCommand c WHERE EXISTS ( SELECT 1 FROM _System_ConnectorRuntimeHandoffReceipt h WHERE h.Namespace = %s AND h.CommandId = c.CommandId AND HEX(h.CommandId) = HEX(c.CommandId) AND h.OrgId = c.OrgId AND HEX(h.OrgId) = HEX(c.OrgId) AND h.AppInstallationId = c.AppInstallationId AND HEX(h.AppInstallationId) = HEX(c.AppInstallationId) AND h.AuthorizedWorkloadId = c.CallerWorkloadId AND HEX(h.AuthorizedWorkloadId) = HEX(c.CallerWorkloadId) AND HEX(c.RuntimeCapabilityId) = HEX('runtime.commands.accept.v1') ) ), scoped_intents AS ( SELECT AdmissionState FROM _System_RuntimeOutboundIntent WHERE ConnectorOwnerKey = %s ), scoped_outbound AS ( SELECT State FROM _System_IntegrationOutboundCommand WHERE ConnectorOwnerKey = %s ) SELECT (SELECT COUNT(*) FROM scoped_work WHERE State IN ('pending','retry','blocked','blocked_retry','processing') OR LeaseUntil > UTC_TIMESTAMP(6)) AS workItemCount, (SELECT COUNT(*) FROM scoped_runtime WHERE State IN ('ACCEPTED','PROCESSING','PUBLISHING','BLOCKED')) AS runtimeCommandCount, (SELECT COUNT(*) FROM scoped_runtime WHERE State='PUBLISHING' OR (State='POISONED' AND RecoveryState='PUBLISHING')) AS runtimePublicationFenceCount, (SELECT COUNT(*) FROM scoped_runtime WHERE State IN ('ACCEPTED','PROCESSING','BLOCKED') AND (COALESCE(PublicationIdentity,'')<>'' OR COALESCE(ResultPayload,'')<>'')) AS runtimeCancellationUnsafeCount, (SELECT COUNT(*) FROM scoped_intents WHERE AdmissionState = 'PENDING') AS runtimeOutboundPendingCount, (SELECT COUNT(*) FROM scoped_intents WHERE AdmissionState = 'ACCEPTED') AS runtimeOutboundAcceptedCount, (SELECT COUNT(*) FROM scoped_outbound WHERE State IN ('pending','claimed','dispatching','retry','writeback')) AS integrationOutboundCount, (SELECT COUNT(*) FROM scoped_outbound WHERE State = 'uncertain') AS uncertainOutboundCount, ((SELECT COUNT(*) FROM scoped_work WHERE State IS NULL OR State NOT IN ('pending','retry','blocked','blocked_retry','processing','processed','skipped','dead')) + (SELECT COUNT(*) FROM scoped_runtime WHERE State IS NULL OR State NOT IN ('ACCEPTED','PROCESSING','PUBLISHING','BLOCKED','COMPLETED','FAILED','POISONED')) + (SELECT COUNT(*) FROM scoped_intents WHERE AdmissionState IS NULL OR AdmissionState NOT IN ('PENDING','ACCEPTED','REJECTED','COMPLETED')) + (SELECT COUNT(*) FROM scoped_outbound WHERE State IS NULL OR State NOT IN ('pending','claimed','dispatching','retry','writeback','uncertain','succeeded','rejected'))) AS unknownStateCount, (SELECT COUNT(*) FROM scoped_work WHERE MatchedEventId IS NULL AND State NOT IN ('processed','skipped','dead')) AS unattributedWorkItemCount """ PIPELINE_ACTIVE_COUNTS = ( "workItemCount", "runtimeCommandCount", "runtimeOutboundPendingCount", "runtimeOutboundAcceptedCount", "integrationOutboundCount", ) PIPELINE_HARD_COUNTS = ( "uncertainOutboundCount", "unknownStateCount", "unattributedWorkItemCount", "runtimePublicationFenceCount", "runtimeCancellationUnsafeCount", ) def fetch_pipeline_drain(conn: pymysql.connections.Connection, event_table: str, namespace: str, connector_owner_key: str) -> dict[str, int]: with conn.cursor() as cur: cur.execute(pipeline_drain_sql(event_table), [namespace, namespace, namespace, connector_owner_key, connector_owner_key]) row = cur.fetchone() if not row: raise RuntimeError("pipeline drain query returned no snapshot") counts = {} for name in (*PIPELINE_ACTIVE_COUNTS, *PIPELINE_HARD_COUNTS): value = row[name] # Missing counts are an error, never an empty queue. if value is None or isinstance(value, bool) or int(value) != value or int(value) < 0: raise RuntimeError(f"pipeline drain query returned invalid count: {name}") counts[name] = int(value) return counts # Event namespace is authoritative when the Event exists. Eventless work uses # the canonical connector installation identity, including removed/deleted # installations because this reset covers prior queue data as well. # Admission resolves the namespace-named connector-pack; the companion is a # separate app-pack identified by Event.CompanionPackInstallId. PRIOR_WORK_ITEM_SCOPE_SQL = """ (w.WorkType IN ('connector_event','workflow_action_apply') AND ((HEX(e.Namespace)=HEX(%s) AND HEX(e.InstalledPackId)=HEX(w.InstalledPackId)) OR (e.Id IS NULL AND EXISTS ( SELECT 1 FROM _System_InstalledPack p WHERE HEX(p.PackInstallId)=HEX(w.InstalledPackId) AND TRIM(COALESCE(p.PackInstallId,''))<>'' AND p.Category='connector-pack' AND HEX(p.Name)=HEX(%s))))) """ def prior_work_purge_statements() -> list[tuple[str, str, list[Any]]]: """Delete the stopped Staging deployment's prior HubSpot queue data once. This administrative test-environment reset includes queue history. Business records, remote identity links, other connector queues, ownership epochs and pack-wide scheduler leases are outside its scope. Delete commands before their handoff authority and dependent attempts before their Events. """ namespace, owner = DEFAULT_NAMESPACE, DEFAULT_CONNECTOR_OWNER_KEY return [ ("runtimeCommands", """ DELETE c FROM _System_RuntimePlatformCommand c WHERE HEX(c.RuntimeCapabilityId)=HEX('runtime.commands.accept.v1') AND EXISTS (SELECT 1 FROM _System_ConnectorRuntimeHandoffReceipt h WHERE HEX(h.Namespace)=HEX(%s) AND h.CommandId=c.CommandId AND HEX(h.CommandId)=HEX(c.CommandId) AND HEX(h.OrgId)=HEX(c.OrgId) AND HEX(h.AppInstallationId)=HEX(c.AppInstallationId) AND HEX(h.AuthorizedWorkloadId)=HEX(c.CallerWorkloadId)) """, [namespace]), ("runtimeOutboundIntents", """ DELETE FROM _System_RuntimeOutboundIntent WHERE HEX(ConnectorOwnerKey)=HEX(%s) """, [owner]), ("integrationOutboundCommands", """ DELETE FROM _System_IntegrationOutboundCommand WHERE HEX(ConnectorOwnerKey)=HEX(%s) """, [owner]), ("workAttempts", f""" DELETE a FROM _System_ConnectorWorkAttempt a LEFT JOIN _System_ConnectorEvent ae ON ae.Id=a.EventId LEFT JOIN _System_ConnectorWorkItem w ON w.WorkId=a.WorkId AND HEX(w.WorkId)=HEX(a.WorkId) LEFT JOIN _System_ConnectorEvent e ON e.Id=w.EventId WHERE (w.WorkId IS NULL OR w.WorkType IN ('connector_event','workflow_action_apply')) AND (HEX(ae.Namespace)=HEX(%s) OR (ae.Id IS NULL AND {PRIOR_WORK_ITEM_SCOPE_SQL})) """, [namespace] * 3), ("runtimeHandoffs", """ DELETE FROM _System_ConnectorRuntimeHandoffReceipt WHERE HEX(Namespace)=HEX(%s) """, [namespace]), ("workItems", f""" DELETE w FROM _System_ConnectorWorkItem w LEFT JOIN _System_ConnectorEvent e ON e.Id=w.EventId WHERE {PRIOR_WORK_ITEM_SCOPE_SQL} """, [namespace] * 2), ("events", """ DELETE FROM _System_ConnectorEvent WHERE HEX(Namespace)=HEX(%s) """, [namespace]), ] def purge_staging_prior_work(conn: pymysql.connections.Connection) -> dict[str, Any]: """One unconditional scoped purge transaction while Staging writers stop.""" stage = "begin transaction" try: conn.begin() with conn.cursor() as cur: stage = "read transaction timestamp" cur.execute("SELECT UTC_TIMESTAMP(6) AS started_at") started_at = cur.fetchone()["started_at"] deleted = {} for name, sql, parameters in prior_work_purge_statements(): stage = name cur.execute(sql, parameters) deleted[name] = cur.rowcount stage = "commit transaction" conn.commit() return {"startedAt": started_at, "deleted": deleted} except BaseException as exc: try: conn.rollback() except Exception: pass # A disconnected socket must not hide the original failure. code = f", database code {exc.args[0]}" if exc.args and isinstance(exc.args[0], int) else "" raise RuntimeError(f"HubSpot queue purge failed at {stage}: {type(exc).__name__}{code}") from exc def fetch_prior_work_remnants(conn: pymysql.connections.Connection) -> dict[str, int]: """One post-purge check; any remaining scoped queue data fails the test.""" with conn.cursor() as cur: cur.execute(f""" SELECT (SELECT COUNT(*) FROM _System_ConnectorEvent WHERE HEX(Namespace)=HEX(%s)) AS events, (SELECT COUNT(*) FROM _System_ConnectorWorkItem w LEFT JOIN _System_ConnectorEvent e ON e.Id=w.EventId WHERE {PRIOR_WORK_ITEM_SCOPE_SQL}) AS workItems, (SELECT COUNT(*) FROM _System_ConnectorWorkAttempt a LEFT JOIN _System_ConnectorEvent ae ON ae.Id=a.EventId LEFT JOIN _System_ConnectorWorkItem w ON w.WorkId=a.WorkId AND HEX(w.WorkId)=HEX(a.WorkId) LEFT JOIN _System_ConnectorEvent e ON e.Id=w.EventId WHERE (w.WorkId IS NULL OR w.WorkType IN ('connector_event','workflow_action_apply')) AND (HEX(ae.Namespace)=HEX(%s) OR (ae.Id IS NULL AND {PRIOR_WORK_ITEM_SCOPE_SQL}))) AS workAttempts, (SELECT COUNT(*) FROM _System_ConnectorRuntimeHandoffReceipt WHERE HEX(Namespace)=HEX(%s)) AS runtimeHandoffs, (SELECT COUNT(*) FROM _System_RuntimePlatformCommand c WHERE HEX(c.RuntimeCapabilityId)=HEX('runtime.commands.accept.v1') AND EXISTS (SELECT 1 FROM _System_ConnectorRuntimeHandoffReceipt h WHERE HEX(h.Namespace)=HEX(%s) AND h.CommandId=c.CommandId AND HEX(h.CommandId)=HEX(c.CommandId) AND HEX(h.OrgId)=HEX(c.OrgId) AND HEX(h.AppInstallationId)=HEX(c.AppInstallationId) AND HEX(h.AuthorizedWorkloadId)=HEX(c.CallerWorkloadId))) AS runtimeCommands, (SELECT COUNT(*) FROM _System_RuntimeOutboundIntent WHERE HEX(ConnectorOwnerKey)=HEX(%s)) AS runtimeOutboundIntents, (SELECT COUNT(*) FROM _System_IntegrationOutboundCommand WHERE HEX(ConnectorOwnerKey)=HEX(%s)) AS integrationOutboundCommands """, [DEFAULT_NAMESPACE] * 8 + [DEFAULT_CONNECTOR_OWNER_KEY] * 2) return {name: int(value) for name, value in cur.fetchone().items()} def tidb_table_exists(conn: pymysql.connections.Connection, table: str) -> bool: with conn.cursor() as cur: cur.execute( "SELECT COUNT(*) AS table_count FROM information_schema.TABLES " "WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = %s", [table], ) return int(cur.fetchone()["table_count"]) > 0 def fetch_workitem_nonterminal( conn: pymysql.connections.Connection, event_table: str, event_columns: dict[str, str], namespace: str, max_samples: int, ) -> dict[str, Any]: """Nonterminal rows in _System_ConnectorWorkItem, the queue-lifecycle authority. Since the WorkItem/Event ledger split, _System_ConnectorWorkItem owns queue lifecycle and Event.Status can be stale, so the drain gate must consult WorkItem State directly. Scoping replicates the canonical drain (connector_inbound_workitem_summary_for_packs, deployments/offline/runtime/ common.py): WorkItems attach to a currently installed, non-deleted pack — connector-pack rows via w.InstalledPackId, app-pack rows via their Event's CompanionPackInstallId — so stale rows from removed packs cannot block, while genuine connector rows count even when their EventId is NULL. Within that scope, rows are limited to the connector work types and to the target namespace (event-attributed rows by e.Namespace; event-less rows by the connector pack name, which carries the namespace string). Only a confirmed missing WorkItem table (environments predating the split) skips the check; other failures propagate so the preflight fails closed. """ if not tidb_table_exists(conn, CONNECTOR_WORKITEM_TABLE): return {"tableExists": False, "nonterminalCount": 0, "samples": [], "eventIdColumn": ""} columns = column_map(conn, CONNECTOR_WORKITEM_TABLE) state_col = required_col(columns, "State") event_id_col = required_col(columns, "EventId") terminal_list = ", ".join(f"'{item}'" for item in sorted(TERMINAL_EVENT_STATUSES)) where = [f"LOWER(COALESCE(w.{quote_ident(state_col)}, '')) NOT IN ({terminal_list})"] params: list[Any] = [] worktype_col = col(columns, "WorkType") if worktype_col: where.append(f"w.{quote_ident(worktype_col)} IN ('connector_event', 'workflow_action_apply')") joins = "" namespace_col = col(event_columns, "Namespace") event_pk = col(event_columns, "Id") companion_col = col(event_columns, "CompanionPackInstallId") if event_pk: joins += f" LEFT JOIN {quote_ident(event_table)} e ON e.{quote_ident(event_pk)} = w.{quote_ident(event_id_col)}" if tidb_table_exists(conn, INSTALLED_PACK_TABLE): pack_columns = column_map(conn, INSTALLED_PACK_TABLE) pack_install_col = required_col(pack_columns, "PackInstallId") pack_category_col = required_col(pack_columns, "Category") pack_status_col = required_col(pack_columns, "Status") pack_name_col = required_col(pack_columns, "Name") pack_deleted_col = col(pack_columns, "IsDeleted") install_id_col = required_col(columns, "InstalledPackId") companion_join = ( f" OR (p.{quote_ident(pack_category_col)} = 'app-pack'" f" AND COALESCE(e.{quote_ident(companion_col)}, '') = p.{quote_ident(pack_install_col)})" if (companion_col and event_pk) else "" ) joins += ( f" JOIN {quote_ident(INSTALLED_PACK_TABLE)} p ON (" f"(p.{quote_ident(pack_category_col)} = 'connector-pack'" f" AND w.{quote_ident(install_id_col)} = p.{quote_ident(pack_install_col)})" f"{companion_join})" ) where.append(f"p.{quote_ident(pack_status_col)} IN ('active', 'installing', 'removing', 'failed')") if pack_deleted_col: where.append(f"COALESCE(p.{quote_ident(pack_deleted_col)}, 0) = 0") where.append(f"TRIM(COALESCE(p.{quote_ident(pack_install_col)}, '')) <> ''") if namespace_col and event_pk: where.append( f"(e.{quote_ident(namespace_col)} = %s" f" OR (e.{quote_ident(event_pk)} IS NULL AND p.{quote_ident(pack_name_col)} = %s))" ) params.extend([namespace, namespace]) else: where.append(f"p.{quote_ident(pack_name_col)} = %s") params.append(namespace) elif namespace_col and event_pk: # No installed-pack ledger on this environment: fall back to # event-namespace scoping and count unattributable rows (fail closed). where.append( f"(e.{quote_ident(namespace_col)} = %s" f" OR w.{quote_ident(event_id_col)} IS NULL" f" OR e.{quote_ident(event_pk)} IS NULL)" ) params.append(namespace) work_id_col = required_col(columns, "WorkId") base = f"FROM {quote_ident(CONNECTOR_WORKITEM_TABLE)} w{joins} WHERE {' AND '.join(where)}" with conn.cursor() as cur: # COUNT(DISTINCT WorkId) like the canonical drain (common.py:3433): # the pack join can match one WorkItem through both the connector-pack # and the companion app-pack row, and COUNT(*) would double-count it. cur.execute( f"SELECT COUNT(DISTINCT w.{quote_ident(work_id_col)}) AS nonterminal_count {base}", params, ) nonterminal_count = int(cur.fetchone()["nonterminal_count"]) samples: list[dict[str, Any]] = [] if nonterminal_count: selected = [actual for name in WORKITEM_SAMPLE_COLUMNS if (actual := col(columns, name))] with conn.cursor() as cur: cur.execute( f"SELECT DISTINCT {', '.join('w.' + quote_ident(item) for item in selected)} {base} LIMIT %s", [*params, max_samples], ) last_error_col = col(columns, "LastError") samples = [] for row in cur.fetchall(): item = {} for k, v in dict(row).items(): if v in (None, ""): continue value = str(v) if last_error_col and k == last_error_col: # Mirror compact_row: never emit full provider error # payloads into CI logs/artifacts. value = value[:240] item[k] = value samples.append(item) return { "tableExists": True, "nonterminalCount": nonterminal_count, "samples": samples, "eventIdColumn": event_id_col, } def fetch_diagnostic_rows( conn: pymysql.connections.Connection, table: str, columns: dict[str, str], namespace: str, cutoff: datetime, now: datetime, max_rows: int = 10000, ) -> list[dict[str, Any]]: params: list[Any] = [] where = [ *namespace_where(columns, namespace, params), *recent_or_processing_where(columns, cutoff, now, params), ] selected = [actual for name in SAMPLE_COLUMNS if (actual := col(columns, name))] order_candidates = [quote_ident(actual) for name in TIMESTAMP_CANDIDATES if (actual := col(columns, name))] order_sql = ", ".join(f"{candidate} DESC" for candidate in order_candidates) or "1" sql = ( f"SELECT {', '.join(quote_ident(item) for item in selected)} " f"FROM {quote_ident(table)} " f"WHERE {' AND '.join(where)} " f"ORDER BY {order_sql} " f"LIMIT %s" ) params.append(max_rows) with conn.cursor() as cur: cur.execute(sql, params) return [dict(row) for row in cur.fetchall()] def fetch_db_clients(conn: pymysql.connections.Connection) -> tuple[int, list[dict[str, Any]]]: with conn.cursor() as cur: cur.execute("SELECT CONNECTION_ID() AS current_connection_id") current_connection_id = int(cur.fetchone()["current_connection_id"]) cur.execute( "SELECT ID, USER, HOST, DB, COMMAND, TIME, STATE " "FROM INFORMATION_SCHEMA.PROCESSLIST " "WHERE (USER = CURRENT_USER() OR USER = SUBSTRING_INDEX(CURRENT_USER(), '@', 1)) " "AND DB = DATABASE() " "ORDER BY HOST, ID" ) rows = [dict(row) for row in cur.fetchall()] return current_connection_id, rows def inspect_target( target: DbTarget, *, table: str, namespace: str, expected_claimed_by: re.Pattern[str], denied_claimed_by: re.Pattern[str] | None, expected_db_client_host: re.Pattern[str], lookback_minutes: int, stale_processing_minutes: int, recent_terminal_foreign_minutes: int, parked_nonterminal_minutes: int, max_nonterminal_events: int, max_samples: int, drain: bool = False, connector_owner_key: str = DEFAULT_CONNECTOR_OWNER_KEY, ) -> dict[str, Any]: now = datetime.now(timezone.utc) # The diagnostic fetch must cover the recent-terminal window even when an # operator tunes it beyond the lookback, or terminal foreign rows inside # the configured window would never be fetched and could evade blocking. fetch_lookback_minutes = max(lookback_minutes, recent_terminal_foreign_minutes) cutoff = now - timedelta(minutes=fetch_lookback_minutes) stale_cutoff = now - timedelta(minutes=stale_processing_minutes) recent_terminal_cutoff = now - timedelta(minutes=recent_terminal_foreign_minutes) parked_cutoff = now - timedelta(minutes=parked_nonterminal_minutes) with connect(target) as conn: if drain: with conn.cursor() as cur: cur.execute("SET TRANSACTION ISOLATION LEVEL REPEATABLE READ") # TiDB rejects the MySQL READ ONLY modifier unless its no-op # feature switch is enabled. Keep the diagnostic read path # SELECT-only and use a normal repeatable-read transaction. conn.begin() pipeline = fetch_pipeline_drain(conn, table, namespace, connector_owner_key) if drain else None columns = column_map(conn, table) status_counts = fetch_status_counts(conn, table, columns, namespace) rows = fetch_diagnostic_rows(conn, table, columns, namespace, cutoff, now) current_connection_id, db_client_rows = fetch_db_clients(conn) workitem_state = ( {"tableExists": True, "nonterminalCount": pipeline["workItemCount"], "samples": [], "eventIdColumn": "EventId"} if pipeline is not None else fetch_workitem_nonterminal(conn, table, columns, namespace, max_samples) ) workitem_coverage = ( {"table": CONNECTOR_WORKITEM_TABLE, "eventIdColumn": workitem_state["eventIdColumn"]} if workitem_state["tableExists"] and workitem_state.get("eventIdColumn") else None ) nonterminal_split = fetch_nonterminal_split( conn, table, columns, namespace, parked_cutoff, max_samples, workitem_coverage ) if drain: conn.rollback() # End the one-shot observation without retaining a transaction. classified = classify_rows( rows, expected_claimed_by=expected_claimed_by, denied_claimed_by=denied_claimed_by, stale_cutoff=stale_cutoff, recent_terminal_cutoff=recent_terminal_cutoff, ) db_clients = classify_db_clients( db_client_rows, expected_client_host=expected_db_client_host, current_connection_id=current_connection_id, ) foreign_samples = [ compact_row(row) for row in rows if row.get("ClaimedBy") and str(row.get("Status") or "").lower() not in TERMINAL_EVENT_STATUSES and ( (denied_claimed_by and denied_claimed_by.search(str(row.get("ClaimedBy") or ""))) or not expected_claimed_by.search(str(row.get("ClaimedBy") or "")) ) ][:max_samples] stale_samples = classified["staleProcessingRows"][:max_samples] nonterminal_events = nonterminal_event_count(status_counts) active_nonterminal = nonterminal_split["activeCount"] parked_nonterminal = max(0, nonterminal_events - active_nonterminal) busy_blockers = [] hard_blockers = [] if drain: max_nonterminal_events = 0 if active_nonterminal > max_nonterminal_events: busy_blockers.append( f"connector queue has {active_nonterminal} active nonterminal events; " f"maximum allowed is {max_nonterminal_events}" ) workitem_count = pipeline["workItemCount"] if pipeline is not None else workitem_state["nonterminalCount"] if workitem_state["tableExists"] and workitem_count > max_nonterminal_events: # WorkItem State is the queue-lifecycle authority since the # WorkItem/Event ledger split; a parked-looking Event can still have # live queue work behind it, so nonterminal WorkItems always block # regardless of the Event-side parked classification. busy_blockers.append( f"connector work-item queue has {workitem_count} nonterminal work items; " f"maximum allowed is {max_nonterminal_events}" ) if classified["foreignWorkers"]: hard_blockers.append("foreign ClaimedBy values on live (non-terminal) connector events") if classified["foreignRecentTerminalWorkers"]: hard_blockers.append( "foreign ClaimedBy values on terminal connector events finalized within " f"{recent_terminal_foreign_minutes} minutes" ) if db_clients["foreignDbClients"]: hard_blockers.append("foreign TiDB client sessions for the target runtime user") if classified["staleProcessingRows"]: hard_blockers.append("stale processing connector-event leases") if pipeline is not None: for name in PIPELINE_ACTIVE_COUNTS[1:]: if pipeline[name]: busy_blockers.append(f"durable HubSpot pipeline has {name}={pipeline[name]}") for name in PIPELINE_HARD_COUNTS: if pipeline[name]: hard_blockers.append(f"durable HubSpot pipeline cannot drain: {name}={pipeline[name]}") blockers = [*busy_blockers, *hard_blockers] return { "target": target.label, "database": target.database, "namespace": namespace, "lookbackMinutes": lookback_minutes, "staleProcessingMinutes": stale_processing_minutes, "recentTerminalForeignMinutes": recent_terminal_foreign_minutes, "maxNonterminalEvents": max_nonterminal_events, "nonterminalEventCount": nonterminal_events, "activeNonterminalEventCount": active_nonterminal, "parkedNonterminalEventCount": parked_nonterminal, "parkedNonterminalMinutes": parked_nonterminal_minutes, "nonterminalSamples": nonterminal_split["activeSamples"], "parkedNonterminalSamples": nonterminal_split["parkedSamples"], "workItemTableExists": workitem_state["tableExists"], "workItemNonterminalCount": workitem_count, "workItemNonterminalSamples": workitem_state["samples"], "statusCounts": status_counts, "allowedWorkers": classified["allowedWorkers"], "foreignWorkers": classified["foreignWorkers"], "foreignRecentTerminalWorkers": classified["foreignRecentTerminalWorkers"], "foreignTerminalWorkers": classified["foreignTerminalWorkers"], "allowedDbClients": db_clients["allowedDbClients"], "foreignDbClients": db_clients["foreignDbClients"], "foreignSamples": foreign_samples, "staleProcessingSamples": stale_samples, "verdict": "BLOCK" if blockers else "PASS", "blockers": blockers, "busyBlockers": busy_blockers, "hardBlockers": hard_blockers, "pipelineCounts": pipeline, } def render_text(result: dict[str, Any]) -> str: lines = [ f"HubSpot connector worker preflight: {result['verdict']}", f"- target_env: {result['targetEnv']}", f"- expected_claimed_by_regex: {result['expectedClaimedByRegex']}", f"- denied_claimed_by_regex: {result['deniedClaimedByRegex']}", f"- expected_db_client_host_regex: {result['expectedDbClientHostRegex']}", ] if result.get("requireIdlePipeline"): lines.append("- pipeline admission: one bounded snapshot; no wait or retry") if result.get("diagnosticError"): lines.append(f"- diagnostic_error: {result['diagnosticError']}") for target in result["targets"]: lines.append(f"- target: {target['target']} database={target['database']} verdict={target['verdict']}") if target["statusCounts"]: counts = ", ".join(f"{item['status']}={item['count']}" for item in target["statusCounts"]) lines.append(f" status_counts: {counts}") lines.append( f" nonterminal_events: {target['nonterminalEventCount']} " f"(active={target.get('activeNonterminalEventCount', target['nonterminalEventCount'])} " f"parked={target.get('parkedNonterminalEventCount', 0)}) " f"max_allowed_active={target['maxNonterminalEvents']}" ) if target.get("workItemTableExists"): lines.append( f" workitem_nonterminal: {target.get('workItemNonterminalCount', 0)} " f"(queue-lifecycle authority; blocks when > max_allowed_active)" ) for sample in target.get("workItemNonterminalSamples") or []: lines.append(f" {json.dumps(sample, sort_keys=True)}") if target.get("nonterminalSamples"): lines.append(" active_nonterminal_samples:") for sample in target["nonterminalSamples"]: lines.append(f" {json.dumps(sample, sort_keys=True)}") if target.get("parkedNonterminalSamples"): lines.append( f" parked_nonterminal_samples (unclaimed, idle > {target.get('parkedNonterminalMinutes', '?')} min; non-blocking):" ) for sample in target["parkedNonterminalSamples"]: lines.append(f" {json.dumps(sample, sort_keys=True)}") if target["allowedWorkers"]: allowed = "; ".join(f"{item['claimedBy']} count={item['count']}" for item in target["allowedWorkers"][:5]) lines.append(f" allowed_workers: {allowed}") if target["foreignWorkers"]: lines.append(" foreign_workers:") for item in target["foreignWorkers"]: lines.append( " " f"claimedBy={item['claimedBy']} count={item['count']} " f"latest={item['latestTime'] or 'unknown'} " f"sourceObjectIDs={','.join(item['sourceObjectIDs']) or 'none'}" ) if target.get("foreignRecentTerminalWorkers"): lines.append(" foreign_recent_terminal_workers (finalized within the recent window; BLOCKING):") for item in target["foreignRecentTerminalWorkers"]: lines.append( " " f"claimedBy={item['claimedBy']} count={item['count']} " f"latest={item['latestTime'] or 'unknown'} " f"sourceObjectIDs={','.join(item['sourceObjectIDs']) or 'none'}" ) if target.get("foreignTerminalWorkers"): lines.append(" foreign_terminal_workers (history only, e.g. cloned prod snapshot; non-blocking):") for item in target["foreignTerminalWorkers"]: lines.append( " " f"claimedBy={item['claimedBy']} count={item['count']} " f"latest={item['latestTime'] or 'unknown'}" ) if target["allowedDbClients"]: allowed_db = "; ".join(f"{item['host']} count={item['count']}" for item in target["allowedDbClients"][:5]) lines.append(f" allowed_db_clients: {allowed_db}") if target["foreignDbClients"]: lines.append(" foreign_db_clients:") for item in target["foreignDbClients"]: commands = ",".join(f"{name}={count}" for name, count in sorted(item["commands"].items())) lines.append( " " f"host={item['host']} count={item['count']} " f"commands={commands or 'unknown'} max_time_seconds={item['maxTimeSeconds']}" ) if target["staleProcessingSamples"]: lines.append(" stale_processing_samples:") for sample in target["staleProcessingSamples"]: lines.append(f" {json.dumps(sample, sort_keys=True)}") if result["verdict"] == "BLOCK": lines.extend( [ "Repair:", "- On Staging, use the active-runtime prior-work purge before a new full or performance suite.", "- Stop any non-staging/local connector worker using this TiDB branch.", "- On the active runtime host, restart only the staging connector/backend service.", "- Re-run this preflight until verdict=PASS before starting HubSpot Integration Test.", ] ) return "\n".join(lines) class OperationTimeoutError(TimeoutError): pass @contextmanager def operation_deadline(timeout_seconds: int): """Bound the complete database operation in this Linux/macOS CLI.""" if not timeout_seconds: yield return def expired(_signum, _frame): raise OperationTimeoutError("HubSpot pipeline operation deadline expired") previous_handler = signal.signal(signal.SIGALRM, expired) previous_timer = signal.setitimer(signal.ITIMER_REAL, timeout_seconds) try: yield finally: signal.setitimer(signal.ITIMER_REAL, *previous_timer) signal.signal(signal.SIGALRM, previous_handler) def main() -> int: args = parse_args() config = load_deployment_config() expected = re.compile(args.expected_claimed_by_regex) denied_claimed_by_regex = args.denied_claimed_by_regex if denied_claimed_by_regex is None: denied_claimed_by_regex = denied_claimed_by_regex_for_config(config, args.target_env) denied = re.compile(denied_claimed_by_regex) if denied_claimed_by_regex else None expected_db_client_host = re.compile(args.expected_db_client_host_regex) targets = resolve_targets(args.target_env, args.slot.strip(), config=config) def inspect(): return [inspect_target( target, table=args.table, namespace=args.namespace, expected_claimed_by=expected, denied_claimed_by=denied, expected_db_client_host=expected_db_client_host, lookback_minutes=args.lookback_minutes, stale_processing_minutes=args.stale_processing_minutes, recent_terminal_foreign_minutes=args.recent_terminal_foreign_minutes, parked_nonterminal_minutes=args.parked_nonterminal_minutes, max_nonterminal_events=args.max_nonterminal_events, max_samples=args.max_samples, drain=args.require_idle_pipeline, connector_owner_key=args.connector_owner_key, ) for target in targets] metadata = { "targetEnv": args.target_env, "expectedClaimedByRegex": args.expected_claimed_by_regex, "deniedClaimedByRegex": denied_claimed_by_regex, "expectedDbClientHostRegex": args.expected_db_client_host_regex, "maxNonterminalEvents": 0 if args.require_idle_pipeline else args.max_nonterminal_events, "connectorOwnerKey": args.connector_owner_key, } metadata["requireIdlePipeline"] = args.require_idle_pipeline started = time.monotonic() try: with operation_deadline(60 if args.require_idle_pipeline else 0): inspected = inspect() if not inspected: raise RuntimeError("pipeline inspection returned no targets") result = {**metadata, "verdict": "BLOCK" if any(item["verdict"] == "BLOCK" for item in inspected) else "PASS", "targets": inspected} except Exception as exc: result = {**metadata, "verdict": "BLOCK", "targets": [], "diagnosticError": f"{type(exc).__name__}: pipeline observation failed"} result["elapsedSeconds"] = round(time.monotonic() - started, 3) if args.json_output: output_path = Path(args.json_output) output_path.parent.mkdir(parents=True, exist_ok=True) temporary = output_path.with_name(output_path.name + ".tmp") temporary.write_text(json.dumps(result, indent=2, sort_keys=True, default=str) + "\n") temporary.replace(output_path) print(render_text(result)) return 2 if result["verdict"] == "BLOCK" else 0 if __name__ == "__main__": try: raise SystemExit(main()) except Exception as exc: print(f"HubSpot connector worker preflight: BLOCK", file=sys.stderr) print(f"- diagnostic_error: {exc}", file=sys.stderr) print("- no credentials were printed; verify target env TiDB secret wiring and retry", file=sys.stderr) raise SystemExit(2)