#!/usr/bin/env python3 from __future__ import annotations import argparse import hashlib import json import os import ssl import sys import time from concurrent.futures import ThreadPoolExecutor from datetime import datetime, timezone from pathlib import Path from typing import Any from urllib.parse import urlencode from urllib.error import HTTPError from urllib.request import Request, urlopen try: import pymysql except ModuleNotFoundError: # Keep contract/profile inspection usable without live DB dependencies. pymysql = None try: from hubspot_pressure_matrix_contract import build_scale_gate, load_pressure_matrix_contract except ModuleNotFoundError: from scripts.hubspot_pressure_matrix_contract import build_scale_gate, load_pressure_matrix_contract try: from lib.workitem_evidence import ( active_lease_sample_violations, explain_has_broad_scan, valid_active_lease_width, ) except ModuleNotFoundError: from scripts.lib.workitem_evidence import ( active_lease_sample_violations, explain_has_broad_scan, valid_active_lease_width, ) ROOT = Path(__file__).resolve().parents[1] DEPLOYMENT_CONFIG = ROOT / "deployments" / "deployment-config.json" NAMESPACE = "hubspot-crm-connector" WEBHOOK_PATH = "/webhooks/hubspot-crm-connector/webhook" WEBHOOK_SECRET_NAME = WEBHOOK_PATH.rsplit("/", 1)[-1] PORTAL_ID = 0 APP_ID = 0 SUBSCRIPTION_ID = 0 TERMINAL_EVENT_STATES = {"processed", "skipped", "dead"} TERMINAL_WORK_STATES = {"processed", "skipped", "dead"} LEAD_UPSERT_EPOCH_MILLIS_PROPERTY = "hs_v2_date_entered_68053680" FIXTURE_MANIFEST_SCHEMA_VERSION = "staging-hubspot-workitem-fixtures/v1" PRESSURE_POLL_INTERVAL_SECONDS = 1.0 class GateBlocked(Exception): def __init__(self, verdict: str, message: str) -> None: super().__init__(message) self.verdict = verdict self.message = message def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser(description="Run HubSpot WorkItem pressure gates against a deployed test/staging environment.") parser.add_argument("--target-env", default=os.environ.get("ENV_NAME", "staging")) parser.add_argument("--base-url", default=os.environ.get("PUBLIC_BASE_URL", "")) parser.add_argument("--client-secret-env", default="EXP_HUBSPOT_CLIENT_SECRET") parser.add_argument("--output", required=True) parser.add_argument("--run-id", default=os.environ.get("GITHUB_RUN_ID", str(int(time.time())))) parser.add_argument("--profile", choices=["pilot", "full", "noop-pilot", "noop-full", "scale"], default="full") parser.add_argument("--pressure-contract", default=os.environ.get("HUBSPOT_PRESSURE_MATRIX_CONTRACT", "")) parser.add_argument("--events", type=int, default=0) parser.add_argument("--lanes", type=int, default=0) parser.add_argument("--workers", type=int, default=16) parser.add_argument("--timeout-seconds", type=int, default=0) parser.add_argument("--hubspot-token-env", default="") parser.add_argument("--expected-portal-id", default=os.environ.get("HUBSPOT_EXPECTED_PORTAL_ID", "")) parser.add_argument("--portal-id", type=int, default=int(os.environ.get("EXP_HUBSPOT_PORTAL_ID", str(PORTAL_ID)))) parser.add_argument("--app-id", type=int, default=int(os.environ.get("EXP_HUBSPOT_APP_ID", str(APP_ID)))) parser.add_argument("--subscription-id", type=int, default=int(os.environ.get("EXP_HUBSPOT_SUBSCRIPTION_ID", str(SUBSCRIPTION_ID)))) parser.add_argument("--fixture-manifest", default="") parser.add_argument("--candidate-id", default="") parser.add_argument("--single-gate", default="") return parser.parse_args() def load_target(target_env: str) -> dict[str, Any]: password = os.environ.get("TIDB_PASSWORD", "") if password and os.environ.get("TIDB_HOST") and os.environ.get("TIDB_USER"): return { "host": os.environ.get("TIDB_HOST", ""), "port": int(os.environ.get("TIDB_PORT", "4000")), "user": os.environ.get("TIDB_USER", ""), "password": password, "database": os.environ.get("TIDB_DATABASE", "nexuscrm"), } cfg = json.loads(DEPLOYMENT_CONFIG.read_text()) target = cfg["environments"]["test"]["topology"]["target_config"][target_env] tidb = target["tidb"] password_env = str(tidb["password_env"]) password = os.environ.get(password_env, "") if not password: raise GateBlocked("BLOCK_PRESSURE_GATE_DB_PASSWORD_MISSING", f"missing TiDB password env {password_env}") return { "host": str(tidb["host"]), "port": int(tidb["port"]), "user": str(tidb["user"]), "password": password, "database": "nexuscrm", } def connect(target: dict[str, Any]) -> pymysql.connections.Connection: if pymysql is None: raise GateBlocked( "BLOCK_PRESSURE_GATE_DB_DRIVER_MISSING", "PyMySQL is required to run a live HubSpot WorkItem pressure gate", ) db_timeout = int(os.environ.get("XIAOPENG_PRESSURE_DB_READ_TIMEOUT_SECONDS", "300")) 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=db_timeout, write_timeout=db_timeout, ssl=ssl.create_default_context(), cursorclass=pymysql.cursors.DictCursor, ) def qident(name: str) -> str: return "`" + name.replace("`", "``") + "`" def table_columns(conn: pymysql.connections.Connection, table: str) -> set[str]: with conn.cursor() as cur: cur.execute(f"SHOW COLUMNS FROM {qident(table)}") return {str(row["Field"]) for row in cur.fetchall()} def has_column(conn: pymysql.connections.Connection, table: str, column: str) -> bool: return column in table_columns(conn, table) def request_json(method: str, url: str, body: Any | None = None, headers: dict[str, str] | None = None) -> tuple[int, str, float]: raw = b"" if body is None else json.dumps(body, separators=(",", ":"), sort_keys=True).encode() req = Request(url, data=raw if body is not None else None, headers=headers or {}, method=method) start = time.perf_counter() try: with urlopen(req, timeout=60) as resp: text = resp.read().decode(errors="replace") return resp.status, text, (time.perf_counter() - start) * 1000.0 except HTTPError as exc: text = exc.read().decode(errors="replace") return exc.code, text, (time.perf_counter() - start) * 1000.0 def signed_headers(_base_url: str, body: Any, secret: str) -> dict[str, str]: raw = json.dumps(body, separators=(",", ":"), sort_keys=True) digest = hashlib.sha256((secret + raw).encode()).hexdigest() return { "Content-Type": "application/json", "X-HubSpot-Signature": digest, } def env_any(names: list[str]) -> str: for name in names: value = os.environ.get(name, "").strip() if value: return value return "" def hubspot_json(access_token: str, path: str, params: dict[str, Any] | None = None) -> dict[str, Any]: query = "" if not params else "?" + urlencode(params) status, body, _latency_ms = request_json( "GET", "https://api.hubapi.com" + path + query, headers={"Authorization": f"Bearer {access_token}", "Content-Type": "application/json"}, ) if not (200 <= status < 300): raise GateBlocked("BLOCK_FIXTURE_NOT_A_VALID_PRESSURE_HARNESS", f"HubSpot read-only contact discovery returned HTTP {status}") return json.loads(body or "{}") def verify_hubspot_sandbox_scope( access_token: str, *, expected_portal_id: str, event_portal_id: int, ) -> dict[str, str]: if not access_token: raise GateBlocked( "BLOCK_HUBSPOT_PRESSURE_ACCESS_TOKEN_MISSING", "missing HubSpot access token for sandbox account verification", ) expected = str(expected_portal_id or "").strip() if not expected: raise GateBlocked( "BLOCK_HUBSPOT_PRESSURE_EXPECTED_PORTAL_ID_REQUIRED", "expected portal ID is required before HubSpot pressure admission", ) details = hubspot_json(access_token, "/account-info/v3/details") account_type = str(details.get("accountType") or "").strip() observed_portal_id = str(details.get("portalId") or "").strip() if account_type != "SANDBOX": raise GateBlocked( "BLOCK_HUBSPOT_PRESSURE_SANDBOX_REQUIRED", f"HubSpot pressure requires a SANDBOX account; observed account type {account_type or '(missing)'}", ) if observed_portal_id != expected: raise GateBlocked( "BLOCK_HUBSPOT_PRESSURE_PORTAL_MISMATCH", f"HubSpot sandbox portal {observed_portal_id or '(missing)'} does not match expected portal {expected}", ) if str(event_portal_id) != expected: raise GateBlocked( "BLOCK_HUBSPOT_PRESSURE_EVENT_PORTAL_MISMATCH", f"pressure event portal {event_portal_id} does not match verified sandbox portal {expected}", ) return {"account_type": account_type, "portal_id": observed_portal_id} def is_optional_epoch_millis(value: Any) -> bool: normalized = str(value or "").strip() return not normalized or (normalized.isascii() and normalized.isdecimal()) def hubspot_batch_read_contacts(access_token: str, ids: list[str]) -> list[dict[str, Any]]: if not access_token: raise GateBlocked("BLOCK_FIXTURE_NOT_A_VALID_PRESSURE_HARNESS", "missing HubSpot access token for read-only contact discovery") contacts: list[dict[str, Any]] = [] for offset in range(0, len(ids), 100): chunk = ids[offset : offset + 100] if not chunk: continue status, body, _latency_ms = request_json( "POST", "https://api.hubapi.com/crm/v3/objects/contacts/batch/read", body={ "properties": ["email", LEAD_UPSERT_EPOCH_MILLIS_PROPERTY], "inputs": [{"id": contact_id} for contact_id in chunk], }, headers={"Authorization": f"Bearer {access_token}", "Content-Type": "application/json"}, ) if not (200 <= status < 300): raise GateBlocked( "BLOCK_FIXTURE_NOT_A_VALID_PRESSURE_HARNESS", f"HubSpot mapped-contact batch read returned HTTP {status}", ) payload = json.loads(body or "{}") for item in payload.get("results") or []: contact_id = str(item.get("id") or "").strip() if not contact_id: continue props = item.get("properties") or {} contacts.append( { "id": contact_id, "email": str(props.get("email") or "").strip().lower(), "lead_upsert_epoch_millis_valid": is_optional_epoch_millis( props.get(LEAD_UPSERT_EPOCH_MILLIS_PROPERTY) ), } ) return contacts def active_filter(conn: pymysql.connections.Connection, table: str) -> str: columns = table_columns(conn, table) parts = ["1=1"] if "IsDeleted" in columns: parts.append("(IsDeleted IS NULL OR IsDeleted = 0)") if table == "Lead" and "IsConverted" in columns: parts.append("(IsConverted IS NULL OR IsConverted = 0)") return " AND ".join(parts) def count_values( conn: pymysql.connections.Connection, table: str, field: str, values: list[str], *, lower: bool = False, ) -> dict[str, int]: if not values or not has_column(conn, table, field): return {} counts: dict[str, int] = {} field_expr = f"LOWER({qident(field)})" if lower else qident(field) normalized = [value.lower() if lower else value for value in values if value] for offset in range(0, len(normalized), 200): chunk = normalized[offset : offset + 200] if not chunk: continue placeholders = ",".join(["%s"] * len(chunk)) rows = fetch_all( conn, f""" SELECT {field_expr} AS k, COUNT(*) AS c FROM {qident(table)} WHERE {field_expr} IN ({placeholders}) AND {active_filter(conn, table)} GROUP BY {field_expr} """, chunk, ) for row in rows: key = str(row.get("k") or "").strip().lower() if lower else str(row.get("k") or "").strip() counts[key] = counts.get(key, 0) + int(row.get("c") or 0) return counts def unambiguous_active_hubspot_ids(conn: pymysql.connections.Connection, required: int) -> list[str]: candidate_limit = max(required * 10, 200) counts: dict[str, int] = {} recency: dict[str, str] = {} for table in ("Lead", "Contact"): if not has_column(conn, table, "HubSpot_Id__c"): continue last_modified_expr = qident("LastModifiedDate") if has_column(conn, table, "LastModifiedDate") else "NULL" rows = fetch_all( conn, f""" SELECT {qident("HubSpot_Id__c")} AS k, COUNT(*) AS c, MAX({last_modified_expr}) AS last_modified FROM {qident(table)} WHERE {qident("HubSpot_Id__c")} IS NOT NULL AND TRIM({qident("HubSpot_Id__c")}) <> '' AND {active_filter(conn, table)} GROUP BY {qident("HubSpot_Id__c")} ORDER BY last_modified DESC LIMIT %s """, [candidate_limit], ) for row in rows: contact_id = str(row.get("k") or "").strip() if not contact_id: continue counts[contact_id] = counts.get(contact_id, 0) + int(row.get("c") or 0) last_modified = str(row.get("last_modified") or "") if last_modified > recency.get(contact_id, ""): recency[contact_id] = last_modified return sorted( (contact_id for contact_id, count in counts.items() if count == 1), key=lambda contact_id: (recency.get(contact_id, ""), contact_id), reverse=True, ) def discover_mapped_hubspot_contacts( conn: pymysql.connections.Connection, access_token: str, required: int, ) -> list[dict[str, Any]]: if required <= 0: return [] mapped_ids = unambiguous_active_hubspot_ids(conn, required) contacts = hubspot_batch_read_contacts(access_token, mapped_ids) if len(contacts) < required: raise GateBlocked( "BLOCK_FIXTURE_NOT_A_VALID_PRESSURE_HARNESS", "HubSpot mapped-contact discovery could not find enough active remote contacts " f"after filtering: returned={len(contacts)} required={required} mapped_candidates={len(mapped_ids)}", ) return contacts def fixture_manifest_contacts(path: Path, candidate_id: str, required: int) -> list[dict[str, str]]: if not candidate_id: raise GateBlocked("BLOCK_FIXTURE_MANIFEST_CANDIDATE_MISSING", "candidate id is required with a fixture manifest") try: payload = json.loads(path.read_text(encoding="utf-8")) except FileNotFoundError as exc: raise GateBlocked("BLOCK_FIXTURE_MANIFEST_MISSING", f"fixture manifest is missing: {path}") from exc except json.JSONDecodeError as exc: raise GateBlocked("BLOCK_FIXTURE_MANIFEST_INVALID", "fixture manifest is not valid JSON") from exc if payload.get("schema_version") != FIXTURE_MANIFEST_SCHEMA_VERSION: raise GateBlocked("BLOCK_FIXTURE_MANIFEST_INVALID", "fixture manifest has an unsupported schema") identity = payload.get("candidate_identity") if not isinstance(identity, dict) or str(identity.get("candidate_id") or "") != candidate_id: raise GateBlocked("BLOCK_FIXTURE_MANIFEST_IDENTITY_MISMATCH", "fixture manifest does not belong to the active candidate") contacts = payload.get("contacts") if not isinstance(contacts, list) or len(contacts) < required: raise GateBlocked( "BLOCK_FIXTURE_MANIFEST_INSUFFICIENT", f"fixture manifest has fewer than required contacts: available={len(contacts) if isinstance(contacts, list) else 0} required={required}", ) seen: set[str] = set() normalized: list[dict[str, str]] = [] for item in contacts: if not isinstance(item, dict): raise GateBlocked("BLOCK_FIXTURE_MANIFEST_INVALID", "fixture manifest contact is invalid") contact_id = str(item.get("id") or "").strip() email = str(item.get("email") or "").strip().lower() if not contact_id or not email or contact_id in seen: raise GateBlocked("BLOCK_FIXTURE_MANIFEST_INVALID", "fixture manifest contacts must contain unique ids and emails") seen.add(contact_id) normalized.append({"id": contact_id, "email": email}) return normalized[:required] def select_unambiguous_contact_ids( conn: pymysql.connections.Connection, contacts: list[dict[str, Any]], required: int, ) -> list[str]: ids = [contact["id"] for contact in contacts if contact.get("id")] emails = sorted({contact["email"] for contact in contacts if contact.get("email")}) remote_counts: dict[str, int] = {} email_counts: dict[str, int] = {} for table in ("Lead", "Contact"): for key, count in count_values(conn, table, "HubSpot_Id__c", ids).items(): remote_counts[key] = remote_counts.get(key, 0) + count for key, count in count_values(conn, table, "Email", emails, lower=True).items(): email_counts[key] = email_counts.get(key, 0) + count selected: list[str] = [] skipped_unmapped_remote = 0 skipped_remote = 0 skipped_email = 0 skipped_invalid_epoch_millis = 0 for contact in contacts: contact_id = contact.get("id", "") email = contact.get("email", "") if not contact.get("lead_upsert_epoch_millis_valid", True): skipped_invalid_epoch_millis += 1 continue remote_count = remote_counts.get(contact_id, 0) if remote_count == 0: skipped_unmapped_remote += 1 continue if remote_count > 1: skipped_remote += 1 continue if email and email_counts.get(email, 0) > 1: skipped_email += 1 continue selected.append(contact_id) if len(selected) >= required: break if len(selected) < required: raise GateBlocked( "BLOCK_FIXTURE_NOT_A_VALID_PRESSURE_HARNESS", "HubSpot read-only discovery could not find enough clean mapped contact lanes " f"after filtering: selected={len(selected)} required={required} " f"candidates={len(contacts)} skipped_invalid_epoch_millis={skipped_invalid_epoch_millis} " f"skipped_unmapped_remote={skipped_unmapped_remote} skipped_remote={skipped_remote} " f"skipped_email={skipped_email}", ) return selected def pressure_run_marker(prefix: str) -> str: return hashlib.sha256(prefix.encode("utf-8")).hexdigest()[:8] def make_events( prefix: str, count: int, lanes: int, lane_source_ids: list[str], app_id: int, subscription_id: int, portal_id: int, *, non_mutating: bool = False, ) -> list[dict[str, Any]]: now_ms = int(time.time() * 1000) if len(lane_source_ids) < lanes: raise GateBlocked( "BLOCK_FIXTURE_NOT_A_VALID_PRESSURE_HARNESS", f"only {len(lane_source_ids)} valid source IDs available for {lanes} lanes", ) events: list[dict[str, Any]] = [] run_marker = pressure_run_marker(prefix) base_events_per_lane, remainder = divmod(count, lanes) event_index = 0 for lane in range(lanes): lane_event_count = base_events_per_lane + (1 if lane < remainder else 0) # The no-op profile exercises signed admission, WorkItem materialization, # row-CAS claim, and drain while the active L1 flow terminates on its # explicit Salesforce-echo branch before any Lead or Contact write. object_id = lane_source_ids[lane] property_name = "first_campaign" if non_mutating else "firstname" change_source = "SALESFORCE" if non_mutating else "CRM_UI" for lane_sequence in range(1, lane_event_count + 1): property_value = "" if non_mutating else f"Pressure-{run_marker}-{lane:04d}-{lane_sequence:04d}" events.append( { "portalId": portal_id, "appId": app_id, "subscriptionId": subscription_id, "eventId": f"{prefix}-{event_index:06d}", "occurredAt": str(now_ms + event_index), "subscriptionType": "object.propertyChange", "objectTypeId": "0-1", "attemptNumber": 0, "objectId": str(object_id), "propertyName": property_name, "propertyValue": property_value, "changeSource": change_source, } ) event_index += 1 return events def expected_latest_first_names(events: list[dict[str, Any]]) -> dict[str, str]: expected: dict[str, str] = {} for event in events: if str(event.get("propertyName") or "").strip().lower() != "firstname": continue source_id = str(event.get("objectId") or "").strip() if source_id: expected[source_id] = str(event.get("propertyValue") or "") return expected def evaluate_business_projection( expected: dict[str, str], observations: list[dict[str, str]], ) -> dict[str, Any]: by_source: dict[str, list[dict[str, str]]] = {} for observation in observations: source_id = str(observation.get("source_id") or "").strip() if source_id in expected: by_source.setdefault(source_id, []).append(observation) matching = 0 missing = 0 duplicate = 0 mismatched = 0 samples: list[dict[str, Any]] = [] for source_id, expected_value in expected.items(): rows = by_source.get(source_id, []) issue = "" if not rows: missing += 1 issue = "missing" elif len(rows) != 1: duplicate += 1 issue = "duplicate_active_identity" elif str(rows[0].get("first_name") or "") != expected_value: mismatched += 1 issue = "latest_value_mismatch" else: matching += 1 if issue and len(samples) < 10: samples.append( { "source_id_sha256": hashlib.sha256(source_id.encode("utf-8")).hexdigest(), "issue": issue, "active_record_count": len(rows), "tables": sorted({str(row.get("table") or "") for row in rows}), } ) expected_count = len(expected) return { "business_oracle": "firstname_latest_per_source", "business_expected_source_count": expected_count, "business_observed_record_count": sum(len(rows) for rows in by_source.values()), "business_matching_source_count": matching, "missing_source_count": missing, "duplicate_source_count": duplicate, "mismatched_source_count": mismatched, "business_mismatch_samples": samples, "business_oracle_ok": matching == expected_count and missing == 0 and duplicate == 0 and mismatched == 0, } def build_event_batches(events: list[dict[str, Any]], batch_size: int = 100) -> list[list[dict[str, Any]]]: if batch_size < 1: raise ValueError("batch_size must be positive") lane_groups: list[list[dict[str, Any]]] = [] seen_lanes: set[str] = set() for event in events: lane = str(event.get("objectId") or "").strip() if not lane: raise GateBlocked("BLOCK_PRESSURE_EVENT_SHAPE_INVALID", "pressure event is missing objectId") if not lane_groups or str(lane_groups[-1][0].get("objectId")) != lane: if lane in seen_lanes: raise GateBlocked( "BLOCK_PRESSURE_EVENT_SHAPE_INVALID", f"source lane {lane} is not contiguous in the pressure event stream", ) seen_lanes.add(lane) lane_groups.append([]) lane_groups[-1].append(event) batches: list[list[dict[str, Any]]] = [] current: list[dict[str, Any]] = [] for lane_group in lane_groups: if len(lane_group) > batch_size: raise GateBlocked( "BLOCK_PRESSURE_EVENT_SHAPE_INVALID", f"one source lane has {len(lane_group)} events, above webhook batch size {batch_size}", ) if current and len(current) + len(lane_group) > batch_size: batches.append(current) current = [] current.extend(lane_group) if current: batches.append(current) return batches def post_event_batches( base_url: str, secret: str, events: list[dict[str, Any]], batch_size: int = 100, concurrency: int = 1, ) -> dict[str, Any]: if concurrency < 1: raise ValueError("concurrency must be positive") samples: list[dict[str, Any]] = [] accepted = 0 rejected = 0 unbound_accepted = 0 receipt_ids: list[str] = [] inline_event_ids: list[str] = [] accepted_event_ids: list[str] = [] started = time.perf_counter() batches = build_event_batches(events, batch_size=batch_size) requested_event_ids = [str(event.get("eventId") or "").strip() for event in events] if any(not event_id for event_id in requested_event_ids) or len(set(requested_event_ids)) != len(requested_event_ids): raise GateBlocked( "BLOCK_PRESSURE_EVENT_SHAPE_INVALID", "pressure events must carry unique non-empty eventId values", ) def post_one(batch: list[dict[str, Any]]) -> tuple[list[dict[str, Any]], int, str, float]: status, response_body, latency_ms = request_json( "POST", base_url.rstrip("/") + WEBHOOK_PATH, batch, signed_headers(base_url, batch, secret), ) return batch, status, response_body, latency_ms with ThreadPoolExecutor(max_workers=min(concurrency, max(1, len(batches)))) as executor: responses = list(executor.map(post_one, batches)) for batch, status, response_body, latency_ms in responses: # Normal HubSpot receipt admission is durable only when the endpoint # returns 202. Other 2xx modes (for example ack-drop maintenance mode) # must not be counted as pressure input accepted for execution. if status == 202: accepted += len(batch) accepted_event_ids.extend(str(event["eventId"]).strip() for event in batch) response: dict[str, Any] = {} try: decoded = json.loads(response_body or "{}") if isinstance(decoded, dict): response = decoded except json.JSONDecodeError: pass receipt_id = str(response.get("receiptId") or "").strip() response_event_ids = response.get("eventIds") if receipt_id: receipt_ids.append(receipt_id) admission_mode = "verified_receipt" elif isinstance(response_event_ids, list) and response_event_ids: inline_event_ids.extend(str(value).strip() for value in response_event_ids if str(value).strip()) admission_mode = "inline_event" else: unbound_accepted += len(batch) admission_mode = "unbound_202" else: rejected += len(batch) admission_mode = "rejected" samples.append( { "status": status, "count": len(batch), "latency_ms": round(latency_ms, 3), "admission_mode": admission_mode, } ) admission_seconds = time.perf_counter() - started request_latencies = [float(sample["latency_ms"]) for sample in samples] return { "accepted": accepted, "rejected": rejected, "unbound_accepted": unbound_accepted, "receipt_ids": receipt_ids, "inline_event_ids": inline_event_ids, "accepted_event_ids": accepted_event_ids, "admission_seconds": round(admission_seconds, 3), "admission_events_per_second": round(len(events) / admission_seconds, 3) if events and admission_seconds > 0 else 0.0, "admission_request_p95_ms": round(percentile(request_latencies, 95), 3), "admission_request_max_ms": round(percentile(request_latencies, 100), 3), "admission_concurrency": concurrency, "samples": samples, } def event_filter_sql(event_ids: list[str]) -> tuple[str, list[Any]]: normalized = list(dict.fromkeys(str(value).strip() for value in event_ids if str(value).strip())) if not normalized: return "e.Namespace = %s AND 1 = 0", [NAMESPACE] placeholders = ",".join(["%s"] * len(normalized)) return ( f"e.Namespace = %s AND e.Id IN ({placeholders})", [NAMESPACE, *normalized], ) def fetch_one(conn: pymysql.connections.Connection, sql: str, params: list[Any]) -> dict[str, Any]: with conn.cursor() as cur: cur.execute(sql, params) row = cur.fetchone() return row or {} def fetch_all(conn: pymysql.connections.Connection, sql: str, params: list[Any]) -> list[dict[str, Any]]: with conn.cursor() as cur: cur.execute(sql, params) return list(cur.fetchall()) def business_projection_readback( conn: pymysql.connections.Connection, expected: dict[str, str], ) -> dict[str, Any]: observations: list[dict[str, str]] = [] missing_schema: list[str] = [] source_ids = list(expected) for table in ("Lead", "Contact"): if not has_column(conn, table, "HubSpot_Id__c") or not has_column(conn, table, "FirstName"): missing_schema.append(table) continue for offset in range(0, len(source_ids), 200): chunk = source_ids[offset : offset + 200] if not chunk: continue placeholders = ",".join(["%s"] * len(chunk)) rows = fetch_all( conn, f""" SELECT {qident("HubSpot_Id__c")} AS source_id, {qident("FirstName")} AS first_name FROM {qident(table)} WHERE {qident("HubSpot_Id__c")} IN ({placeholders}) AND {active_filter(conn, table)} """, chunk, ) observations.extend( { "table": table, "source_id": str(row.get("source_id") or "").strip(), "first_name": str(row.get("first_name") or ""), } for row in rows ) evaluated = evaluate_business_projection(expected, observations) evaluated["business_schema_ready"] = not missing_schema evaluated["business_missing_schema_tables"] = missing_schema if missing_schema: evaluated["business_oracle_ok"] = False return evaluated def receipt_materialization_snapshot( rows: list[dict[str, Any]], *, expected_receipt_ids: list[str], inline_event_ids: list[str], expected_events: int, ) -> dict[str, Any]: row_by_id = {str(row.get("Id") or "").strip(): row for row in rows} status_counts: dict[str, int] = {} event_ids = [value for value in inline_event_ids if value] invalid_event_id_json = 0 terminal_failure_receipts = 0 for receipt_id in expected_receipt_ids: row = row_by_id.get(receipt_id) if row is None: continue status = str(row.get("Status") or "").strip().lower() status_counts[status] = status_counts.get(status, 0) + 1 if status in {"dead", "config_terminal"}: terminal_failure_receipts += 1 raw_event_ids = row.get("EnqueuedEventIds") try: decoded = json.loads(str(raw_event_ids or "[]")) except json.JSONDecodeError: decoded = None if not isinstance(decoded, list): invalid_event_id_json += 1 continue event_ids.extend(str(value).strip() for value in decoded if str(value).strip()) unique_event_ids = list(dict.fromkeys(event_ids)) missing_receipts = sum(1 for receipt_id in expected_receipt_ids if receipt_id not in row_by_id) all_receipts_enqueued = ( missing_receipts == 0 and status_counts.get("enqueued", 0) == len(expected_receipt_ids) ) return { "complete": ( terminal_failure_receipts == 0 and invalid_event_id_json == 0 and all_receipts_enqueued and len(unique_event_ids) == expected_events and len(event_ids) == expected_events ), "event_ids": unique_event_ids, "status_counts": status_counts, "missing_receipt_count": missing_receipts, "invalid_event_id_json_count": invalid_event_id_json, "terminal_failure_receipt_count": terminal_failure_receipts, "duplicate_event_id_count": len(event_ids) - len(unique_event_ids), } def scoped_idempotency_key(secret_name: str, raw_event_id: str) -> str: # Match connector_webhooks.scopeIdempotencyKey: receipt Event IDs are # internal UUIDs, so the original HubSpot eventId is proven through this # endpoint-scoped persisted key rather than by comparing unlike ID domains. digest = hashlib.sha256( (str(secret_name).strip().lower() + "\x00" + str(raw_event_id)).encode("utf-8") ).hexdigest() return f"ep:{digest}" def fetch_event_identity_rows( conn: pymysql.connections.Connection, event_ids: list[str], ) -> list[dict[str, Any]]: normalized = list( dict.fromkeys(str(value).strip() for value in event_ids if str(value).strip()) ) rows: list[dict[str, Any]] = [] for offset in range(0, len(normalized), 200): chunk = normalized[offset : offset + 200] placeholders = ",".join(["%s"] * len(chunk)) rows.extend( fetch_all( conn, f""" SELECT Id, IdempotencyKey FROM `_System_ConnectorEvent` WHERE Namespace = %s AND Id IN ({placeholders}) """, [NAMESPACE, *chunk], ) ) return rows def event_identity_snapshot( rows: list[dict[str, Any]], *, receipt_event_ids: list[str], generated_event_ids: list[str], secret_name: str, ) -> dict[str, Any]: normalized_receipt_ids = [ str(value).strip() for value in receipt_event_ids if str(value).strip() ] normalized_generated_ids = [ str(value).strip() for value in generated_event_ids if str(value).strip() ] receipt_id_set = set(normalized_receipt_ids) expected_keys = { scoped_idempotency_key(secret_name, event_id) for event_id in normalized_generated_ids } observed_rows = [ row for row in rows if str(row.get("Id") or "").strip() in receipt_id_set ] observed_row_ids = {str(row.get("Id") or "").strip() for row in observed_rows} observed_keys = { str(row.get("IdempotencyKey") or "").strip() for row in observed_rows if str(row.get("IdempotencyKey") or "").strip() } missing_event_rows = receipt_id_set - observed_row_ids unexpected_keys = observed_keys - expected_keys missing_expected_keys = expected_keys - observed_keys subset_valid = ( not missing_event_rows and not unexpected_keys and len(observed_rows) == len(observed_row_ids) and len(observed_keys) == len(observed_rows) ) complete = ( subset_valid and len(receipt_id_set) == len(normalized_generated_ids) and len(expected_keys) == len(normalized_generated_ids) and not missing_expected_keys ) return { "event_identity_subset_valid": subset_valid, "event_identity_complete": complete, "event_identity_row_count": len(observed_rows), "missing_receipt_event_row_count": len(missing_event_rows), "missing_generated_idempotency_key_count": len(missing_expected_keys), "unexpected_idempotency_key_count": len(unexpected_keys), } def fetch_receipt_rows( conn: pymysql.connections.Connection, receipt_ids: list[str], ) -> list[dict[str, Any]]: if not receipt_ids: return [] rows: list[dict[str, Any]] = [] for offset in range(0, len(receipt_ids), 200): chunk = receipt_ids[offset : offset + 200] placeholders = ",".join(["%s"] * len(chunk)) rows.extend( fetch_all( conn, f""" SELECT Id, Status, COALESCE(EnqueuedEventIds, '[]') AS EnqueuedEventIds FROM `_System_ConnectorIngressReceipt` WHERE Id IN ({placeholders}) """, chunk, ) ) return rows def wait_for_receipt_materialization( conn: pymysql.connections.Connection, admission: dict[str, Any], timeout_seconds: int, ) -> dict[str, Any]: if int(admission.get("unbound_accepted") or 0) != 0: raise GateBlocked( "BLOCK_PRESSURE_DURABLE_ADMISSION_UNBOUND", "a 202 response returned neither a verified receipt id nor inline event ids", ) receipt_ids = [str(value) for value in admission.get("receipt_ids") or []] inline_event_ids = [str(value) for value in admission.get("inline_event_ids") or []] expected_events = int(admission.get("accepted") or 0) expected_event_ids = [str(value) for value in admission.get("accepted_event_ids") or []] if ( len(expected_event_ids) != expected_events or len(set(expected_event_ids)) != expected_events or any(not event_id.strip() for event_id in expected_event_ids) ): raise GateBlocked( "BLOCK_PRESSURE_DURABLE_ADMISSION_EVENT_ID_MISMATCH", "accepted pressure batches are not bound to exact unique generated event ids", ) started = time.perf_counter() deadline = time.time() + max(1, timeout_seconds) last: dict[str, Any] = {} lease_evidence: dict[str, Any] = { "max_observed_leased_work_items": 0, "max_observed_leased_lanes": 0, "max_observed_lease_owners": 0, "max_observed_lease_tokens": 0, "max_valid_simultaneous_leases": 0, "lease_observations": 0, "lease_invariant_violations": 0, "lease_invariant_violation_counts": {}, } while time.time() < deadline: rows = fetch_receipt_rows(conn, receipt_ids) last = receipt_materialization_snapshot( rows, expected_receipt_ids=receipt_ids, inline_event_ids=inline_event_ids, expected_events=expected_events, ) scoped_event_ids = list(last.get("event_ids") or []) receipt_complete = bool(last.get("complete")) identity = event_identity_snapshot( fetch_event_identity_rows(conn, scoped_event_ids), receipt_event_ids=scoped_event_ids, generated_event_ids=expected_event_ids, secret_name=WEBHOOK_SECRET_NAME, ) last["receipt_complete"] = receipt_complete last.update(identity) last["complete"] = receipt_complete and bool(identity["event_identity_complete"]) if scoped_event_ids and identity["event_identity_subset_valid"]: counts = state_counts(conn, scoped_event_ids) active = active_claim_snapshot(conn, scoped_event_ids) lease_evidence["max_observed_leased_work_items"] = max( int(lease_evidence["max_observed_leased_work_items"]), active["leased_work_items"], ) lease_evidence["max_observed_leased_lanes"] = max( int(lease_evidence["max_observed_leased_lanes"]), active["leased_lanes"], ) lease_evidence["max_observed_lease_owners"] = max( int(lease_evidence["max_observed_lease_owners"]), active["lease_owners"], ) lease_evidence["max_observed_lease_tokens"] = max( int(lease_evidence["max_observed_lease_tokens"]), active["lease_tokens"], ) if active["leased_work_items"] > 0: lease_evidence["lease_observations"] = int(lease_evidence["lease_observations"]) + 1 expired_processing = int(counts.get("expired_processing") or 0) violations = active_lease_sample_violations(active, expired_processing=expired_processing) if violations: lease_evidence["lease_invariant_violations"] = int( lease_evidence["lease_invariant_violations"] ) + 1 violation_counts = lease_evidence["lease_invariant_violation_counts"] for violation in violations: violation_counts[violation] = int(violation_counts.get(violation) or 0) + 1 lease_evidence["max_valid_simultaneous_leases"] = max( int(lease_evidence["max_valid_simultaneous_leases"]), valid_active_lease_width(active, expired_processing=expired_processing), ) if int(last.get("terminal_failure_receipt_count") or 0) != 0: raise GateBlocked( "BLOCK_PRESSURE_RECEIPT_TERMINAL_FAILURE", f"verified ingress receipt reached terminal failure: status_counts={last.get('status_counts')}", ) if receipt_complete and not identity["event_identity_complete"]: raise GateBlocked( "BLOCK_PRESSURE_DURABLE_ADMISSION_EVENT_ID_MISMATCH", "verified ingress receipts resolved Event rows outside the exact generated event identity set", ) if last.get("complete"): last["receipt_materialization_seconds"] = round(time.perf_counter() - started, 3) last.update(lease_evidence) return last if not receipt_ids: raise GateBlocked( "BLOCK_PRESSURE_DURABLE_ADMISSION_EVENT_ID_MISMATCH", "inline 202 response did not bind exactly the accepted pressure events", ) time.sleep(PRESSURE_POLL_INTERVAL_SECONDS) raise GateBlocked( "BLOCK_PRESSURE_RECEIPT_MATERIALIZATION_INCOMPLETE", f"verified ingress receipts did not resolve exactly {expected_events} event ids: " f"status_counts={last.get('status_counts', {})} event_ids={len(last.get('event_ids') or [])} " f"missing_identity_keys={last.get('missing_generated_idempotency_key_count', 0)} " f"unexpected_identity_keys={last.get('unexpected_idempotency_key_count', 0)}", ) def state_counts(conn: pymysql.connections.Connection, event_ids: list[str]) -> dict[str, Any]: where, params = event_filter_sql(event_ids) row = fetch_one( conn, f""" SELECT COUNT(*) AS materialized, COALESCE(SUM(e.Status = 'processed'), 0) AS processed, COALESCE(SUM(e.Status = 'skipped'), 0) AS skipped, COALESCE(SUM(e.Status = 'pending'), 0) AS pending, COALESCE(SUM(e.Status = 'processing'), 0) AS processing, COALESCE(SUM(e.Status = 'failed'), 0) AS failed, COALESCE(SUM(e.Status = 'retry'), 0) AS retry, COALESCE(SUM(e.Status = 'dead'), 0) AS dead, COALESCE(SUM(e.Status NOT IN ('processed', 'skipped', 'dead')), 0) AS nonterminal FROM _System_ConnectorEvent e WHERE {where} """, params, ) work = fetch_one( conn, f""" SELECT COUNT(*) AS work_materialized, COALESCE(SUM(w.State != 'pending'), 0) AS work_claimed, COALESCE(SUM(w.State = 'pending'), 0) AS work_pending, COALESCE(SUM(w.State = 'processing'), 0) AS work_processing, COALESCE(SUM(w.State NOT IN ('processed', 'skipped', 'dead')), 0) AS work_nonterminal, COUNT(DISTINCT NULLIF(w.LeaseOwner, '')) AS lease_owner_count, COUNT(DISTINCT NULLIF(w.LeaseToken, '')) AS lease_token_count, COALESCE(SUM( w.State = 'processing' AND (w.LeaseUntil IS NULL OR w.LeaseUntil <= UTC_TIMESTAMP()) ), 0) AS expired_processing, MIN(TIMESTAMPDIFF(MICROSECOND, w.CreatedDate, COALESCE(w.LeaseUntil, w.LastModifiedDate))) AS first_claim_us, AVG(TIMESTAMPDIFF(MICROSECOND, w.CreatedDate, w.LastModifiedDate)) AS avg_lifecycle_us FROM _System_ConnectorEvent e JOIN _System_ConnectorWorkItem w ON w.EventId = e.Id WHERE {where} """, params, ) return {**row, **work} def active_claim_snapshot(conn: pymysql.connections.Connection, event_ids: list[str]) -> dict[str, int]: where, params = event_filter_sql(event_ids) row = fetch_one( conn, f""" SELECT COUNT(*) AS leased_work_items, COUNT(DISTINCT NULLIF( COALESCE(NULLIF(w.TargetMutationLaneKey, ''), NULLIF(w.SourceLaneKey, ''), ''), '' )) AS leased_lanes, COUNT(DISTINCT NULLIF(TRIM(w.LeaseOwner), '')) AS lease_owners, COUNT(DISTINCT NULLIF(TRIM(w.LeaseToken), '')) AS lease_tokens, COALESCE(SUM( COALESCE(NULLIF(w.TargetMutationLaneKey, ''), NULLIF(w.SourceLaneKey, ''), '') = '' ), 0) AS missing_effective_lane_count, COALESCE(SUM(TRIM(COALESCE(w.LeaseOwner, '')) = ''), 0) AS missing_lease_owner_count, COALESCE(SUM(TRIM(COALESCE(w.LeaseToken, '')) = ''), 0) AS missing_lease_token_count, COALESCE(SUM(COALESCE(w.OwnershipEpoch, 0) <= 0), 0) AS nonpositive_ownership_epoch_count FROM _System_ConnectorEvent e JOIN _System_ConnectorWorkItem w ON w.EventId = e.Id WHERE {where} AND w.State = 'processing' AND w.LeaseUntil > UTC_TIMESTAMP() """, params, ) return { "leased_work_items": int(row.get("leased_work_items") or 0), "leased_lanes": int(row.get("leased_lanes") or 0), "lease_owners": int(row.get("lease_owners") or 0), "lease_tokens": int(row.get("lease_tokens") or 0), "missing_effective_lane_count": int(row.get("missing_effective_lane_count") or 0), "missing_lease_owner_count": int(row.get("missing_lease_owner_count") or 0), "missing_lease_token_count": int(row.get("missing_lease_token_count") or 0), "nonpositive_ownership_epoch_count": int(row.get("nonpositive_ownership_epoch_count") or 0), } def wait_for_gate( conn: pymysql.connections.Connection, event_ids: list[str], expected: int, timeout_seconds: int, *, initial_evidence: dict[str, Any] | None = None, ) -> dict[str, Any]: initial_evidence = initial_evidence or {} deadline = time.time() + timeout_seconds last = {} max_leased_work_items = int(initial_evidence.get("max_observed_leased_work_items") or 0) max_leased_lanes = int(initial_evidence.get("max_observed_leased_lanes") or 0) max_lease_owners = int(initial_evidence.get("max_observed_lease_owners") or 0) max_lease_tokens = int(initial_evidence.get("max_observed_lease_tokens") or 0) max_valid_simultaneous_leases = int(initial_evidence.get("max_valid_simultaneous_leases") or 0) lease_observations = int(initial_evidence.get("lease_observations") or 0) lease_invariant_violations = int(initial_evidence.get("lease_invariant_violations") or 0) lease_invariant_violation_counts: dict[str, int] = dict( initial_evidence.get("lease_invariant_violation_counts") or {} ) def attach_evidence(result: dict[str, Any]) -> dict[str, Any]: result["max_observed_leased_work_items"] = max_leased_work_items result["max_observed_leased_lanes"] = max_leased_lanes result["max_observed_lease_owners"] = max_lease_owners result["max_observed_lease_tokens"] = max_lease_tokens result["max_valid_simultaneous_leases"] = max_valid_simultaneous_leases result["lease_observations"] = lease_observations result["lease_invariant_violations"] = lease_invariant_violations result["lease_invariant_violation_counts"] = dict(lease_invariant_violation_counts) return result while time.time() < deadline: last = state_counts(conn, event_ids) active = active_claim_snapshot(conn, event_ids) max_leased_work_items = max(max_leased_work_items, active["leased_work_items"]) max_leased_lanes = max(max_leased_lanes, active["leased_lanes"]) max_lease_owners = max(max_lease_owners, active["lease_owners"]) max_lease_tokens = max(max_lease_tokens, active["lease_tokens"]) if active["leased_work_items"] > 0: lease_observations += 1 expired_processing = int(last.get("expired_processing") or 0) violations = active_lease_sample_violations( active, expired_processing=expired_processing, ) if violations: lease_invariant_violations += 1 for violation in violations: lease_invariant_violation_counts[violation] = ( lease_invariant_violation_counts.get(violation, 0) + 1 ) max_valid_simultaneous_leases = max( max_valid_simultaneous_leases, valid_active_lease_width(active, expired_processing=expired_processing), ) materialized = int(last.get("materialized") or 0) nonterminal = int(last.get("nonterminal") or 0) + int(last.get("work_nonterminal") or 0) if materialized >= expected and nonterminal == 0: return attach_evidence(last) time.sleep(PRESSURE_POLL_INTERVAL_SECONDS) last = state_counts(conn, event_ids) return attach_evidence(last) def completion_concurrency(conn: pymysql.connections.Connection, event_ids: list[str]) -> dict[str, int]: where, params = event_filter_sql(event_ids) row = fetch_one( conn, f""" SELECT COALESCE(MAX(lanes_completed), 0) AS max_lanes_completed_same_second FROM ( SELECT completed_at, COUNT(DISTINCT NULLIF(lane, '')) AS lanes_completed FROM ( SELECT DATE_FORMAT(w.LastModifiedDate, '%%Y-%%m-%%d %%H:%%i:%%s') AS completed_at, w.SourceLaneKey AS lane FROM _System_ConnectorEvent e JOIN _System_ConnectorWorkItem w ON w.EventId = e.Id WHERE {where} AND w.State IN ('processed', 'skipped', 'dead') AND w.LastModifiedDate IS NOT NULL ) completed GROUP BY completed_at ) t """, params, ) return {"max_lanes_completed_same_second": int(row.get("max_lanes_completed_same_second") or 0)} def independent_lane_capacity_ok(worker_count: int, counts: dict[str, Any]) -> bool: return worker_count == 1 or int(counts.get("max_valid_simultaneous_leases") or 0) > 1 def throughput_stats(*, accepted: int, drain_seconds: float) -> dict[str, float]: if accepted <= 0 or drain_seconds <= 0: return { "terminal_events_per_second": 0.0, "terminal_events_per_minute": 0.0, } per_second = accepted / drain_seconds return { "terminal_events_per_second": round(per_second, 3), "terminal_events_per_minute": round(per_second * 60, 3), } def percentile(values: list[float], pct: float) -> float: if not values: return 0.0 ordered = sorted(values) idx = int(round((pct / 100.0) * (len(ordered) - 1))) return ordered[max(0, min(idx, len(ordered) - 1))] def claim_latency_stats(conn: pymysql.connections.Connection, event_ids: list[str]) -> dict[str, float]: where, params = event_filter_sql(event_ids) rows = fetch_all( conn, f""" SELECT TIMESTAMPDIFF(MICROSECOND, w.CreatedDate, w.LastModifiedDate) / 1000.0 AS ms FROM _System_ConnectorEvent e JOIN _System_ConnectorWorkItem w ON w.EventId = e.Id WHERE {where} AND w.LastModifiedDate IS NOT NULL """, params, ) values = [float(row["ms"] or 0) for row in rows] return { "p50_claim_latency_ms": round(percentile(values, 50), 3), "p95_claim_latency_ms": round(percentile(values, 95), 3), "max_claim_latency_ms": round(percentile(values, 100), 3), } def lane_ordering(conn: pymysql.connections.Connection, event_ids: list[str]) -> dict[str, Any]: where, params = event_filter_sql(event_ids) rows = fetch_all( conn, f""" SELECT COALESCE(w.SourceLaneKey, '') AS lane, w.LaneOrdinal, e.Status, w.LastModifiedDate, w.WorkId FROM _System_ConnectorEvent e JOIN _System_ConnectorWorkItem w ON w.EventId = e.Id WHERE {where} ORDER BY lane, w.LastModifiedDate, w.LaneOrdinal, w.WorkId """, params, ) last_by_lane: dict[str, int] = {} samples: list[str] = [] for row in rows: lane = str(row["lane"]) ordinal = int(row["LaneOrdinal"] or 0) last = last_by_lane.get(lane, 0) if ordinal < last and len(samples) < 10: samples.append(f"{lane}: ordinal {ordinal} after {last} ({row['WorkId']})") last_by_lane[lane] = max(last, ordinal) return {"lane_ordering_ok": not samples, "lane_order_violations": len(samples), "lane_order_samples": samples} def explain_claim(conn: pymysql.connections.Connection, event_ids: list[str]) -> dict[str, Any]: where, params = event_filter_sql(event_ids) scope = fetch_one( conn, f""" SELECT MIN(w.InstalledPackId) AS InstalledPackId, MIN(w.Generation) AS Generation, COUNT(*) AS workitem_count, COALESCE(SUM(w.InstalledPackId IS NULL OR w.Generation IS NULL), 0) AS invalid_scope_count, COUNT(DISTINCT w.InstalledPackId, w.Generation) AS scope_count, COALESCE(SUM( p.PackInstallId IS NULL OR p.Name != %s OR p.Status != 'active' OR COALESCE(p.IsDeleted, 0) != 0 ), 0) AS invalid_active_pack_count FROM _System_ConnectorEvent e JOIN _System_ConnectorWorkItem w ON w.EventId = e.Id LEFT JOIN _System_InstalledPack p ON p.PackInstallId = w.InstalledPackId WHERE {where} """, [NAMESPACE, *params], ) if ( int(scope.get("workitem_count") or 0) == 0 or int(scope.get("invalid_scope_count") or 0) != 0 or int(scope.get("scope_count") or 0) != 1 or int(scope.get("invalid_active_pack_count") or 0) != 0 ): return { "candidate_claim_plan": "workitem_row_cas", "active_pack_match": False, "explain_available": False, "explain_has_broad_scan": False, "scheduler_lease_read_path_used": False, "explain_access_summary": ["pressure run does not resolve to exactly one WorkItem pack generation"], } claim_sql = """ EXPLAIN SELECT w.WorkId FROM _System_ConnectorWorkItem w WHERE w.InstalledPackId = %s AND w.Generation = %s AND w.QueueClass IN ('default', 'bulk') AND ( (w.State IN ('pending', 'retry') AND w.RunAfter <= UTC_TIMESTAMP()) OR (w.State = 'processing' AND w.LeaseUntil IS NOT NULL AND w.LeaseUntil <= UTC_TIMESTAMP()) ) AND ( NULLIF(COALESCE(w.ProjectionKey, ''), '') IS NULL OR NULLIF(COALESCE(w.ProjectionActiveKey, ''), '') IS NOT NULL ) AND ( w.WorkType NOT IN ('connector_event', 'workflow_action_apply') OR EXISTS ( SELECT 1 FROM _System_ConnectorEvent immutable_event WHERE immutable_event.Id = w.EventId ) ) AND NOT EXISTS ( SELECT 1 FROM _System_ConnectorWorkItem older USE INDEX (idx_connector_work_item_source_lane_gen) WHERE older.InstalledPackId = w.InstalledPackId AND older.Generation = w.Generation AND older.WorkId <> w.WorkId AND ( NULLIF(COALESCE(older.ProjectionKey, ''), '') IS NULL OR NULLIF(COALESCE(older.ProjectionActiveKey, ''), '') IS NOT NULL ) AND ( older.State = 'processing' OR ( older.State IN ('pending', 'retry', 'blocked', 'blocked_retry') AND older.RunAfter <= UTC_TIMESTAMP() ) ) AND COALESCE(NULLIF(older.TargetMutationLaneKey, ''), older.SourceLaneKey, '') = COALESCE(NULLIF(w.TargetMutationLaneKey, ''), w.SourceLaneKey, '') AND ( older.LaneOrdinal < w.LaneOrdinal OR ( older.LaneOrdinal = w.LaneOrdinal AND ( older.CreatedDate < w.CreatedDate OR (older.CreatedDate = w.CreatedDate AND older.WorkId < w.WorkId) ) ) ) ) ORDER BY w.LaneOrdinal ASC, w.RunAfter ASC, w.CreatedDate ASC, w.WorkId ASC LIMIT 1024 """ scheduler_lease_read_path_used = "_System_ConnectorSchedulerLease" in claim_sql if scheduler_lease_read_path_used: raise GateBlocked( "BLOCK_WORKITEM_CLAIM_PATH_USES_SCHEDULER_LEASE", "WorkItem claim evidence unexpectedly depends on SchedulerLease", ) rows = fetch_all( conn, claim_sql, [scope["InstalledPackId"], scope["Generation"]], ) summary = [json.dumps(row, default=str, sort_keys=True)[:500] for row in rows] broad = explain_has_broad_scan(rows) return { "candidate_claim_plan": "workitem_row_cas", "claim_scope": { "installed_pack_id_sha256": hashlib.sha256( str(scope["InstalledPackId"]).encode("utf-8") ).hexdigest(), "generation": int(scope["Generation"]), "workitem_count": int(scope["workitem_count"]), }, "active_pack_match": True, "explain_available": True, "explain_has_broad_scan": broad, "scheduler_lease_read_path_used": scheduler_lease_read_path_used, "explain_access_summary": summary, } def gate_diagnostics(conn: pymysql.connections.Connection, event_ids: list[str]) -> dict[str, Any]: where, params = event_filter_sql(event_ids) event_states = fetch_all( conn, f""" SELECT e.Status, COUNT(*) AS count, MIN(LEFT(COALESCE(e.LastError, ''), 500)) AS sample_error, MIN(e.CreatedDate) AS first_created, MAX(e.LastModifiedDate) AS last_modified FROM _System_ConnectorEvent e WHERE {where} GROUP BY e.Status ORDER BY e.Status """, params, ) work_states = fetch_all( conn, f""" SELECT w.State, w.WorkType, w.Generation, w.QueueClass, COUNT(*) AS count, COALESCE(SUM(w.LeaseOwner IS NOT NULL AND w.LeaseOwner != ''), 0) AS leased_count, COUNT(DISTINCT NULLIF(w.LeaseOwner, '')) AS lease_owner_count, COUNT(DISTINCT NULLIF(w.LeaseToken, '')) AS lease_token_count, COALESCE(MAX(w.OwnershipEpoch), 0) AS max_ownership_epoch, MIN(w.RunAfter) AS min_run_after, MAX(w.RunAfter) AS max_run_after, MIN(w.LeaseUntil) AS min_lease_until, MAX(w.LeaseUntil) AS max_lease_until FROM _System_ConnectorEvent e JOIN _System_ConnectorWorkItem w ON w.EventId = e.Id WHERE {where} GROUP BY w.State, w.WorkType, w.Generation, w.QueueClass ORDER BY w.Generation, w.QueueClass, w.State, w.WorkType """, params, ) row = fetch_one( conn, f""" SELECT COALESCE(SUM( w.State IN ('pending', 'retry') AND w.RunAfter <= UTC_TIMESTAMP() AND NOT EXISTS ( SELECT 1 FROM _System_ConnectorWorkItem older WHERE older.InstalledPackId = w.InstalledPackId AND older.Generation = w.Generation AND older.WorkId <> w.WorkId AND COALESCE(NULLIF(older.TargetMutationLaneKey, ''), older.SourceLaneKey, '') = COALESCE(NULLIF(w.TargetMutationLaneKey, ''), w.SourceLaneKey, '') AND ( older.State = 'processing' OR ( older.State IN ('pending', 'retry', 'blocked', 'blocked_retry') AND older.RunAfter <= UTC_TIMESTAMP() ) ) AND ( older.LaneOrdinal < w.LaneOrdinal OR ( older.LaneOrdinal = w.LaneOrdinal AND ( older.CreatedDate < w.CreatedDate OR (older.CreatedDate = w.CreatedDate AND older.WorkId < w.WorkId) ) ) ) ) ), 0) AS eligible, COALESCE(SUM( w.State = 'processing' AND w.LeaseUntil IS NOT NULL AND w.LeaseUntil <= UTC_TIMESTAMP() ), 0) AS expired_processing FROM _System_ConnectorEvent e JOIN _System_ConnectorWorkItem w ON w.EventId = e.Id WHERE {where} """, params, ) return { "event_state_details": event_states, "work_state_details": work_states, "candidate_claim_plan": "workitem_row_cas", "claim_eligible_count": int(row.get("eligible") or 0), "expired_processing_count": int(row.get("expired_processing") or 0), } def run_gate( conn: pymysql.connections.Connection, base_url: str, secret: str, run_id: str, gate: dict[str, Any], lane_source_ids: list[str], app_id: int, subscription_id: int, portal_id: int, fixture_source: str, ) -> dict[str, Any]: prefix = f"hswi-{run_id}-{gate['name']}" events = make_events( prefix, int(gate["events"]), int(gate["lanes"]), lane_source_ids, app_id, subscription_id, portal_id, non_mutating=bool(gate.get("non_mutating")), ) started = time.perf_counter() admission = post_event_batches( base_url, secret, events, batch_size=int(gate.get("webhook_batch_size") or 100), concurrency=int(gate.get("admission_concurrency") or 1), ) gate_timeout_seconds = int(gate.get("timeout_seconds", 600)) elapsed_seconds = time.perf_counter() - started receipt_materialization = wait_for_receipt_materialization( conn, admission, max(1, int(gate_timeout_seconds - elapsed_seconds)), ) event_ids = list(receipt_materialization["event_ids"]) elapsed_seconds = time.perf_counter() - started counts = wait_for_gate( conn, event_ids, admission["accepted"], max(1, int(gate_timeout_seconds - elapsed_seconds)), initial_evidence=receipt_materialization, ) terminal_elapsed_seconds = time.perf_counter() - started post_admission_drain_seconds = max( 0.0, terminal_elapsed_seconds - float(admission["admission_seconds"]), ) latencies = claim_latency_stats(conn, event_ids) ordering = lane_ordering(conn, event_ids) completion = completion_concurrency(conn, event_ids) explain = explain_claim(conn, event_ids) if gate.get("explain") else {} diagnostics = gate_diagnostics(conn, event_ids) business = {} if gate.get("business_oracle") == "firstname_latest_per_source": business = business_projection_readback(conn, expected_latest_first_names(events)) max_active = int(counts.get("max_valid_simultaneous_leases") or 0) workitem_lease_single_winner_observed = ( int(counts.get("lease_observations") or 0) > 0 and int(counts.get("lease_invariant_violations") or 0) == 0 ) result = { "gate": gate, "prefix": prefix, "fixture_source": fixture_source, "fixture_source_id_count": len(lane_source_ids), "accepted": admission["accepted"], "rejected": admission["rejected"], "unbound_accepted": admission["unbound_accepted"], "verified_receipt_count": len(admission["receipt_ids"]), "inline_event_id_count": len(admission["inline_event_ids"]), "event_primary_key_count": len(event_ids), "receipt_status_counts": receipt_materialization["status_counts"], "receipt_materialization_seconds": receipt_materialization["receipt_materialization_seconds"], "admission_seconds": admission["admission_seconds"], "admission_events_per_second": admission["admission_events_per_second"], "admission_request_p95_ms": admission["admission_request_p95_ms"], "admission_request_max_ms": admission["admission_request_max_ms"], "admission_concurrency": admission["admission_concurrency"], "status_codes": admission["samples"], "materialized": int(counts.get("materialized") or 0), "work_materialized": int(counts.get("work_materialized") or 0), "processed": int(counts.get("processed") or 0), "skipped": int(counts.get("skipped") or 0), "claimed": int(counts.get("work_claimed") or 0), "pending": int(counts.get("pending") or 0), "processing": int(counts.get("processing") or 0), "work_pending": int(counts.get("work_pending") or 0), "work_processing": int(counts.get("work_processing") or 0), "failed": int(counts.get("failed") or 0), "retry": int(counts.get("retry") or 0), "dead": int(counts.get("dead") or 0), "nonterminal": int(counts.get("nonterminal") or 0) + int(counts.get("work_nonterminal") or 0), "first_claim_latency_ms": round((float(counts.get("first_claim_us") or 0) / 1000.0), 3), **latencies, # Keep drain_seconds for historical report readers; it is the same # request-start-to-terminal span now named explicitly below. "drain_seconds": round(terminal_elapsed_seconds, 3), "terminal_elapsed_seconds": round(terminal_elapsed_seconds, 3), "post_admission_drain_seconds": round(post_admission_drain_seconds, 3), **throughput_stats(accepted=admission["accepted"], drain_seconds=terminal_elapsed_seconds), "max_active_workers": max_active, "max_observed_leased_work_items": int(counts.get("max_observed_leased_work_items") or 0), "max_observed_leased_lanes": int(counts.get("max_observed_leased_lanes") or 0), "max_observed_lease_owners": int(counts.get("max_observed_lease_owners") or 0), "max_observed_lease_tokens": int(counts.get("max_observed_lease_tokens") or 0), "max_valid_simultaneous_leases": max_active, "lease_observations": int(counts.get("lease_observations") or 0), "lease_invariant_violations": int(counts.get("lease_invariant_violations") or 0), "lease_invariant_violation_counts": counts.get("lease_invariant_violation_counts") or {}, "independent_lane_concurrency_ok": independent_lane_capacity_ok(int(gate["workers"]), counts), "workitem_lease_single_winner_observed": workitem_lease_single_winner_observed, **ordering, **completion, **explain, **diagnostics, **business, } result["verdict"] = classify_gate_result(result) return result def classify_gate_result(result: dict[str, Any]) -> str: requested_events = int(result["gate"].get("events") or 0) if requested_events > 0 and int(result.get("accepted") or 0) != requested_events: return "BLOCK_PRESSURE_ADMISSION_INCOMPLETE" if requested_events > 0 and int(result.get("materialized") or 0) != requested_events: return "BLOCK_PRESSURE_MATERIALIZATION_INCOMPLETE" if result["accepted"] == 0 or result["materialized"] == 0: return "BLOCK_FIXTURE_NOT_A_VALID_PRESSURE_HARNESS" if result["processed"] == 0 and result["failed"] > 0: return "BLOCK_FIXTURE_NOT_A_VALID_PRESSURE_HARNESS" if result["processed"] == 0 and int(result["gate"]["workers"]) >= 16: return "BLOCK_PRESSURE_ARCHITECTURE_NOT_DRAINING" if result["gate"].get("explain"): if result.get("active_pack_match") is not True: return "BLOCK_CLAIM_ACTIVE_PACK_SCOPE_MISMATCH" if result.get("explain_available") is not True: return "BLOCK_CLAIM_HOTPATH_EXPLAIN_UNAVAILABLE" if result.get("scheduler_lease_read_path_used"): return "BLOCK_WORKITEM_CLAIM_PATH_USES_SCHEDULER_LEASE" if result.get("explain_has_broad_scan"): return "BLOCK_CLAIM_HOTPATH_STILL_SCANNING" if result["nonterminal"] != 0: return "BLOCK_PRESSURE_ARCHITECTURE_NOT_DRAINING" expected_work_items = int(result["gate"].get("lanes") or 0) if ( result["gate"].get("source_projection") and int(result["gate"].get("settle_seconds") or 0) > 0 and int(result.get("work_materialized") or 0) > expected_work_items ): return "BLOCK_PRESSURE_FIXTURE_SETTLE_WINDOW_SPLIT" if result["gate"].get("require_capacity_evidence"): if not result.get("workitem_lease_single_winner_observed"): return "BLOCK_WORKITEM_LEASE_SINGLE_WINNER_UNPROVED" if not result.get("independent_lane_concurrency_ok"): return "BLOCK_WORKITEM_MULTI_WORKER_CAPACITY_UNPROVED" expected_bulk_capacity = int(result["gate"].get("expected_bulk_eligible_workers") or 0) if ( expected_bulk_capacity > 0 and int(result.get("max_valid_simultaneous_leases") or 0) < expected_bulk_capacity ): return "BLOCK_WORKER_BULK_CAPACITY_NOT_SATURATED" if result["gate"].get("source_projection"): if result["failed"] != 0 or result["retry"] != 0 or result["dead"] != 0: return "BLOCK_PRESSURE_ARCHITECTURE_NOT_DRAINING" if result["processed"] != expected_work_items: return "BLOCK_SOURCE_PROJECTION_CONTRACT" if result["skipped"] != result["accepted"] - expected_work_items: return "BLOCK_SOURCE_PROJECTION_CONTRACT" if result["work_materialized"] != expected_work_items or result["claimed"] != expected_work_items: return "BLOCK_SOURCE_PROJECTION_CONTRACT" elif result["processed"] != result["accepted"]: return "BLOCK_FIXTURE_NOT_A_VALID_PRESSURE_HARNESS" if result["gate"].get("business_oracle") and not result.get("business_oracle_ok"): return "BLOCK_PRESSURE_BUSINESS_ORACLE_MISMATCH" if not result["lane_ordering_ok"]: return "BLOCK_PRESSURE_ARCHITECTURE_NOT_DRAINING" return "PASS" def health_ok(base_url: str) -> bool: status, _body, _latency = request_json("GET", base_url.rstrip("/") + "/health") return status == 200 def legacy_worker16_name(legacy_name: str, workers: int) -> str: if workers == 16: return legacy_name return f"{legacy_name}_worker{workers}" def gates_for_profile( profile: str, *, workers: int = 16, events: int = 0, lanes: int = 0, timeout_seconds: int = 0, pressure_contract: dict[str, Any] | None = None, ) -> list[dict[str, Any]]: if profile == "scale": return [ build_scale_gate( workers=workers, events=events, source_lanes=lanes or None, timeout_seconds=timeout_seconds or None, contract=pressure_contract, ) ] if profile == "pilot": return [{"name": legacy_worker16_name("pilot_20_events_5_lanes", workers), "events": 20, "lanes": 5, "workers": workers, "timeout_seconds": 180, "explain": True, "source_projection": True}] if profile == "noop-pilot": return [{"name": legacy_worker16_name("noop_pilot_20_events_20_lanes", workers), "events": 20, "lanes": 20, "workers": workers, "timeout_seconds": 180, "explain": True, "non_mutating": True}] if profile == "noop-full": return [{"name": f"noop_worker{workers}_800_events_120_lanes", "events": 800, "lanes": 120, "workers": workers, "timeout_seconds": 600, "explain": True, "require_capacity_evidence": True, "non_mutating": True}] return [ {"name": f"worker{workers}_800_events_120_lanes", "events": 800, "lanes": 120, "workers": workers, "timeout_seconds": 600, "source_projection": True, "require_capacity_evidence": True}, {"name": "worker1_500_backlog", "events": 500, "lanes": 120, "workers": 1, "timeout_seconds": 600, "source_projection": True}, {"name": legacy_worker16_name("large_pending_1000_events_501_lanes", workers), "events": 1000, "lanes": 501, "workers": workers, "timeout_seconds": 600, "explain": True, "source_projection": True, "require_capacity_evidence": True}, ] def apply_timeout_override(gates: list[dict[str, Any]]) -> list[dict[str, Any]]: raw = os.environ.get("XIAOPENG_PRESSURE_GATE_TIMEOUT_SECONDS", "").strip() if not raw: return gates try: timeout_seconds = int(raw) except ValueError: return gates if timeout_seconds <= 0: return gates adjusted: list[dict[str, Any]] = [] for gate in gates: copied = dict(gate) copied["timeout_seconds"] = timeout_seconds adjusted.append(copied) return adjusted def validate_scale_execution_scope( *, profile: str, target_env: str, fixture_manifest: str, candidate_id: str, ) -> None: if profile != "scale": return if str(target_env).strip().lower() != "test1": raise GateBlocked( "BLOCK_PRESSURE_SCALE_TEST1_ONLY", "the 2k/4k/6k scale profile is restricted to Test1", ) if not str(fixture_manifest).strip(): raise GateBlocked( "BLOCK_PRESSURE_SCALE_FIXTURE_MANIFEST_REQUIRED", "the scale profile requires a candidate-scoped fixture manifest", ) if not str(candidate_id).strip(): raise GateBlocked( "BLOCK_PRESSURE_SCALE_CANDIDATE_REQUIRED", "the scale profile requires the candidate identity bound by its fixture manifest", ) def main() -> int: args = parse_args() validate_scale_execution_scope( profile=args.profile, target_env=args.target_env, fixture_manifest=args.fixture_manifest, candidate_id=args.candidate_id, ) global PORTAL_ID, APP_ID, SUBSCRIPTION_ID PORTAL_ID = int(args.portal_id) APP_ID = int(args.app_id) SUBSCRIPTION_ID = int(args.subscription_id) secret = os.environ.get(args.client_secret_env, "").strip() if not secret: raise GateBlocked("BLOCK_PRESSURE_GATE_CLIENT_SECRET_MISSING", f"missing client secret env {args.client_secret_env}") token_names = [args.hubspot_token_env] if args.hubspot_token_env else [] token_names.extend([ f"{args.target_env.upper().replace('-', '_')}_HUBSPOT_ACCESS_TOKEN", "HUBSPOT_STAGING_TOKEN", "EXP_HUBSPOT_ACCESS_TOKEN", "HUBSPOT_ACCESS_TOKEN", ]) report: dict[str, Any] = { "generated_at": datetime.now(timezone.utc).isoformat(), "target_env": args.target_env, "base_url": args.base_url, "run_id": args.run_id, "profile": args.profile, "fixture_source": "hubspot_candidate_scoped_manifest" if args.fixture_manifest else "hubspot_readonly_existing_contacts", "health_ok": False, "results": [], } try: pressure_contract = ( load_pressure_matrix_contract(args.pressure_contract) if args.profile == "scale" else None ) gates = apply_timeout_override( gates_for_profile( args.profile, workers=args.workers, events=args.events, lanes=args.lanes, timeout_seconds=args.timeout_seconds, pressure_contract=pressure_contract, ) ) if args.single_gate: gates = [gate for gate in gates if str(gate.get("name")) == args.single_gate] if not gates: raise GateBlocked("BLOCK_PRESSURE_GATE_UNKNOWN_SINGLE_GATE", f"single gate not found: {args.single_gate}") report["health_ok"] = health_ok(args.base_url) if not report["health_ok"]: raise GateBlocked("BLOCK_PRESSURE_GATE_HEALTH_NOT_200", "health endpoint did not return 200") access_token = env_any(token_names) report["hubspot_scope"] = verify_hubspot_sandbox_scope( access_token, expected_portal_id=args.expected_portal_id, event_portal_id=PORTAL_ID, ) target = load_target(args.target_env) required_lanes = max(int(gate["lanes"]) for gate in gates) report["fixture_required_lanes"] = required_lanes with connect(target) as conn: fixture_source = "hubspot_readonly_existing_contacts" if args.fixture_manifest: fixture_source = "hubspot_candidate_scoped_manifest" report["fixture_manifest"] = str(Path(args.fixture_manifest).resolve()) report["fixture_candidate_id"] = args.candidate_id manifest_contacts = fixture_manifest_contacts(Path(args.fixture_manifest), args.candidate_id, required_lanes) contact_candidates = hubspot_batch_read_contacts(access_token, [contact["id"] for contact in manifest_contacts]) if len(contact_candidates) != len(manifest_contacts): raise GateBlocked( "BLOCK_FIXTURE_MANIFEST_REMOTE_MISSING", f"candidate fixture contacts are no longer readable from HubSpot: returned={len(contact_candidates)} required={len(manifest_contacts)}", ) elif args.candidate_id: raise GateBlocked( "BLOCK_FIXTURE_MANIFEST_REQUIRED", "candidate-scoped pressure validation requires its fixture manifest", ) else: contact_candidates = discover_mapped_hubspot_contacts(conn, access_token, required_lanes) report["fixture_candidate_count"] = len(contact_candidates) lane_source_ids = select_unambiguous_contact_ids(conn, contact_candidates, required_lanes) report["fixture_source_id_count"] = len(lane_source_ids) for gate in gates: result = run_gate( conn, args.base_url, secret, args.run_id, gate, lane_source_ids, APP_ID, SUBSCRIPTION_ID, PORTAL_ID, fixture_source, ) report["results"].append(result) if result["verdict"] != "PASS": raise GateBlocked(result["verdict"], f"{gate['name']} returned {result['verdict']}") report["verdict"] = "PASS_HUBSPOT_WORKITEM_PRESSURE_GATE" rc = 0 except GateBlocked as exc: report["verdict"] = exc.verdict report["blocker"] = exc.message rc = 1 except Exception as exc: report["verdict"] = "BLOCK_HUBSPOT_WORKITEM_PRESSURE_GATE_FAILED" report["blocker"] = f"{type(exc).__name__}: {exc}" rc = 1 Path(args.output).parent.mkdir(parents=True, exist_ok=True) Path(args.output).write_text(json.dumps(report, indent=2, sort_keys=True, default=str) + "\n") print(json.dumps({"verdict": report["verdict"], "output": args.output}, sort_keys=True)) return rc if __name__ == "__main__": sys.exit(main())