#!/usr/bin/env python3 from __future__ import annotations import json import os from itertools import product from pathlib import Path from typing import Any SCHEMA_VERSION = "hubspot-inbound-pressure-matrix/v1" class PressureMatrixContractError(ValueError): pass def bulk_eligible_workers(worker_count: int) -> int: if worker_count < 1: raise PressureMatrixContractError("worker_count must be positive") if worker_count == 1: return 1 latency_only_workers = max(1, worker_count // 4) return worker_count - latency_only_workers def load_pressure_matrix_contract(path: Path | str | None = None) -> dict[str, Any]: raw_path = str(path or os.environ.get("HUBSPOT_PRESSURE_MATRIX_CONTRACT", "")).strip() if not raw_path: raise PressureMatrixContractError( "pressure matrix contract path is required via argument or HUBSPOT_PRESSURE_MATRIX_CONTRACT" ) contract_path = Path(raw_path).expanduser().resolve() if not contract_path.is_file(): raise PressureMatrixContractError(f"pressure matrix contract is not a file: {contract_path}") data = json.loads(contract_path.read_text(encoding="utf-8")) if data.get("schemaVersion") != SCHEMA_VERSION: raise PressureMatrixContractError( f"pressure matrix schemaVersion must be {SCHEMA_VERSION!r}" ) worker_counts = data.get("workerCounts") event_counts = data.get("eventCounts") if worker_counts != [8, 16, 32]: raise PressureMatrixContractError("workerCounts must be exactly [8, 16, 32]") if event_counts != [2000, 4000, 6000]: raise PressureMatrixContractError("eventCounts must be exactly [2000, 4000, 6000]") case_expectations = data.get("caseExpectations") if not isinstance(case_expectations, list): raise PressureMatrixContractError("caseExpectations must be a list") expected_cells = set(product(worker_counts, event_counts)) observed_cells: set[tuple[int, int]] = set() for item in case_expectations: if not isinstance(item, dict): raise PressureMatrixContractError("each case expectation must be an object") cell = (item.get("workers"), item.get("events")) if cell not in expected_cells or cell in observed_cells: raise PressureMatrixContractError(f"case expectation cell is invalid or duplicated: {cell}") observed_cells.add(cell) if item.get("expectation") not in {"must_pass", "should_pass", "diagnostic"}: raise PressureMatrixContractError(f"case expectation tier is invalid for {cell}") if not isinstance(item.get("releaseBlocking"), bool): raise PressureMatrixContractError(f"releaseBlocking must be boolean for {cell}") if item.get("correctnessRequired") is not True: raise PressureMatrixContractError(f"correctnessRequired must be true for {cell}") if not str(item.get("description") or "").strip() or not str(item.get("nonPassPolicy") or "").strip(): raise PressureMatrixContractError(f"case description and nonPassPolicy are required for {cell}") if observed_cells != expected_cells: raise PressureMatrixContractError("caseExpectations must describe the exact 9-cell matrix") source_lanes = data.get("sourceLanes") timeout_seconds = data.get("timeoutSeconds") if not isinstance(source_lanes, int) or source_lanes < max(worker_counts): raise PressureMatrixContractError("sourceLanes must be an integer at least as large as the largest worker count") if not isinstance(timeout_seconds, int) or timeout_seconds < 1: raise PressureMatrixContractError("timeoutSeconds must be a positive integer") if any(event_count % source_lanes != 0 for event_count in event_counts): raise PressureMatrixContractError("every event count must divide evenly across sourceLanes") if data.get("webhookBatchSize") != 100: raise PressureMatrixContractError("webhookBatchSize must match the HubSpot route bound of 100") if data.get("admissionConcurrency") != 4: raise PressureMatrixContractError("admissionConcurrency must remain the calibrated request-side value 4") if max(event_counts) // source_lanes > data["webhookBatchSize"]: raise PressureMatrixContractError("one source burst must fit in one webhook request") if data.get("sourceSettleSeconds") != 5: raise PressureMatrixContractError("sourceSettleSeconds must match the L0 settled-route contract") expected_route = { "subscriptionType": "object.propertyChange", "objectTypeId": "0-1", "eventKind": "contact.propertychange", "flowDeveloperName": "HubSpot_Contact_Lead_Upsert", "invocationMode": "latest_by_source_full_record_preserve_payload_settled", "queueClass": "bulk", "businessOracle": "firstname_latest_per_source", "signatureScheme": "sha256-secret-prefix-template", "signatureHeader": "X-HubSpot-Signature", } for field, expected in expected_route.items(): if data.get(field) != expected: raise PressureMatrixContractError(f"{field} must be {expected!r}") if data.get("verifiedIngressReceipts") is not True: raise PressureMatrixContractError("verifiedIngressReceipts must be true") if data.get("fixtureContract") != "candidate-scoped-manifest": raise PressureMatrixContractError("fixtureContract must be candidate-scoped-manifest") return data def build_scale_gate( *, workers: int, events: int, source_lanes: int | None = None, timeout_seconds: int | None = None, contract: dict[str, Any] | None = None, ) -> dict[str, Any]: contract = contract or load_pressure_matrix_contract() if workers not in contract["workerCounts"]: raise PressureMatrixContractError( f"workers must be one of {contract['workerCounts']}, got {workers}" ) if events not in contract["eventCounts"]: raise PressureMatrixContractError( f"events must be one of {contract['eventCounts']}, got {events}" ) lanes = contract["sourceLanes"] if source_lanes is None else source_lanes timeout = contract["timeoutSeconds"] if timeout_seconds is None else timeout_seconds if lanes != contract["sourceLanes"]: raise PressureMatrixContractError( f"source lanes must remain the matrix control {contract['sourceLanes']}, got {lanes}" ) if not isinstance(timeout, int) or timeout < 1: raise PressureMatrixContractError("timeout_seconds must be a positive integer") expectation = next( item for item in contract["caseExpectations"] if item["workers"] == workers and item["events"] == events ) return { "name": f"inbound_{events}_events_{lanes}_lanes_worker{workers}", "events": events, "lanes": lanes, "workers": workers, "timeout_seconds": timeout, "source_projection": True, "explain": True, "require_capacity_evidence": True, "expected_bulk_eligible_workers": bulk_eligible_workers(workers), "webhook_batch_size": int(contract["webhookBatchSize"]), "admission_concurrency": int(contract["admissionConcurrency"]), "settle_seconds": int(contract["sourceSettleSeconds"]), "subscription_type": str(contract["subscriptionType"]), "object_type_id": str(contract["objectTypeId"]), "event_kind": str(contract["eventKind"]), "flow_developer_name": str(contract["flowDeveloperName"]), "invocation_mode": str(contract["invocationMode"]), "queue_class": str(contract["queueClass"]), "business_oracle": str(contract["businessOracle"]), "verified_ingress_receipts": bool(contract["verifiedIngressReceipts"]), "signature_scheme": str(contract["signatureScheme"]), "signature_header": str(contract["signatureHeader"]), "expectation": str(expectation["expectation"]), "release_blocking": bool(expectation["releaseBlocking"]), "correctness_required": bool(expectation["correctnessRequired"]), "description": str(expectation["description"]), "non_pass_policy": str(expectation["nonPassPolicy"]), } def build_pressure_matrix(contract: dict[str, Any] | None = None) -> list[dict[str, Any]]: contract = contract or load_pressure_matrix_contract() return [ build_scale_gate(workers=workers, events=events, contract=contract) for workers, events in product(contract["workerCounts"], contract["eventCounts"]) ]