Which concurrency lock-contention patterns must be recorded before worker pools spin down?

**Date of Record:** 2026 October 26 **Scope:** Applies when a multi-threaded or multi-process agentic worker pool executes concurrent tasks under shared mutable state (mutexes, semaphores, database row locks, or distributed leases) and is preparing to spin down. Does not apply to stateless, lock-free worker pools (e.g., pure fan-out HTTP workers) or strictly single-threaded sequential execution.

Core Distinction

The separation between **transient task outcome** (whether a worker's assigned job returned success or failure) and the **lock contention topology** (the exact distribution of wait times, lock hold durations, and starvation events across the pool). Task outcome is the output; contention topology is the operational cost that produced it. Failing to record the contention topology creates the **Lock-Wait Phantom Bandwidth** problem: a successor inherits a pool size that *looks* adequate (tasks succeeded), but was actually running at 5% effective throughput due to lock starvation, guaranteeing it will deadlock or time out under the same configuration.

What Was Observed

In concurrent agent worker pools, failing to record lock contention patterns before spin-down produces three primary failure modes:

1. **Pool-Size Recidivism:** The predecessor ran 16 workers, but 14 were consistently blocked on a single global config mutex. The orchestrator logs "16 workers, 16 tasks succeeded" and spins up 16 workers for the successor. The successor hits the same contention wall and times out entirely under a slightly larger workload.
2. **Priority Inversion Blindness:** A high-priority worker was repeatedly preempted by low-priority workers holding a shared lock. The task eventually succeeded but with an unacceptable tail latency. Without a `starvation_signature`, the successor inherits the same lock hierarchy and reproduces the same latency violations.
3. **Deadlock Precursor Erasure:** Workers A and B executed without deadlocking, but the telemetry shows A held lock X while waiting for lock Y, and B held Y while waiting for X, just 3ms apart. The session spins down successfully, but the successor's different scheduling ordering flips the 3ms gap into a full deadlock.

Recorded Elements

1. **`hot_lock_register`:** The exact set of locks, mutexes, or leases that consumed the majority of aggregate wait time across the pool (e.g., `global_config_mutex: 42s aggregate wait`).
2. **`hold_time_distribution`:** For each hot lock, the p50, p95, and max hold durations observed, allowing the successor to detect pathological long-holders.
3. **`starvation_signatures`:** The explicit list of worker/lock pairs where a worker waited more than N times the median, indicating priority inversion or unfair lock scheduling.
4. **`pre_deadlock_ordering_witness`:** The recorded acquisition ordering (e.g., A → X → Y, B → Y → X) that did not deadlock in this run but demonstrated the circular wait pattern, allowing the successor to adopt a deterministic global lock ordering.

Condition of Validity

* **Applies when:** The worker pool executed tasks that acquired shared locks or leases, and a successor instance is expected to spawn a similar pool against the same shared resources. * **Does not apply when:** The pool is truly lock-free, operates only on immutable per-worker data partitions, or the workload's concurrency model is strictly single-threaded.

Decision Boundary

1. **Did the pool acquire shared locks or leases during execution?** If no, standard task outcome logging suffices. If yes, proceed to evaluate contention.
2. **Does any single lock account for more than the contention threshold (e.g., 20% of aggregate pool wait time)?** If yes, classify it as a `HOT_LOCK` and record the `hold_time_distribution`.
3. **Did any worker wait more than the starvation ratio (e.g., 5x the median) on a single lock?** If yes, record the `starvation_signature` and reduce the pool size or shard the lock for successors.
4. **Conversely, when can the pool size be safely preserved for the successor?** Only if no hot locks were identified, no worker exceeded the starvation ratio, and no cyclic lock ordering was observed.

Operational Effect

1. **Pool-Size Calibration:** Prevents successors from blindly inheriting an oversized or undersized pool by surfacing effective (not just nominal) throughput.
2. **Deadlock Prevention:** Passes forward the exact circular wait orderings observed during the session, letting successors enforce global lock ordering rules.
3. **Latency Predictability:** Replaces aggregate completion time with tail-latency awareness, ensuring the successor's timeout budgets account for lock contention.

from dataclasses import dataclass
from enum import Enum, auto
import hashlib
import json


class ContentionVerdict(Enum):
    LOW_CONTENTION = auto()
    HOT_LOCK_DETECTED = auto()
    STARVATION_OBSERVED = auto()
    CYCLIC_ORDERING_DETECTED = auto()


@dataclass(frozen=True)
class LockAcquisitionEvent:
    worker_id: str
    lock_id: str
    wait_time_ms: int
    hold_time_ms: int


@dataclass(frozen=True)
class ContentionSnapshot:
    total_pool_wait_ms: int
    hot_lock_threshold_pct: float
    starvation_ratio: float


@dataclass(frozen=True)
class LockContentionManifest:
    verdict: ContentionVerdict
    hot_lock_register: tuple[tuple[str, int], ...]
    starvation_signatures: tuple[tuple[str, str], ...]
    cyclic_ordering_witness: tuple[tuple[str, str], ...]
    recommended_pool_size: int
    manifest_digest: str


class LockContentionRecorder:
    """
    Evaluates worker pool lock acquisitions at spin-down to preserve
    contention topology for safe successor pool reconfiguration.
    """

    @staticmethod
    def _hash(data: str) -> str:
        return hashlib.sha256(data.encode("utf-8")).hexdigest()

    @classmethod
    def evaluate_contention(
        cls,
        events: tuple[LockAcquisitionEvent, ...],
        snapshot: ContentionSnapshot,
        nominal_pool_size: int,
    ) -> LockContentionManifest:
        
        if nominal_pool_size <= 0:
            raise ValueError("Nominal pool size must be positive.")

        # Gate 1: Aggregate wait time per lock
        lock_wait: dict[str, int] = {}
        worker_wait: dict[str, int] = {}

        for event in events:
            lock_wait[event.lock_id] = lock_wait.get(event.lock_id, 0) + event.wait_time_ms
            worker_wait[event.worker_id] = worker_wait.get(event.worker_id, 0) + event.wait_time_ms

        total_wait = sum(lock_wait.values()) or 1  # prevent div by zero

        # Gate 2: Identify hot locks
        hot_locks = {
            lock_id: wait
            for lock_id, wait in lock_wait.items()
            if (wait / total_wait) * 100 >= snapshot.hot_lock_threshold_pct
        }

        # Gate 3: Identify starvation signatures
        starvation: list[tuple[str, str]] = []
        if worker_wait:
            median_wait = sorted(worker_wait.values())[len(worker_wait) // 2]
            if median_wait > 0:
                for worker_id, wait in worker_wait.items():
                    if wait > median_wait * snapshot.starvation_ratio:
                        starvation.append((worker_id, f"wait={wait}ms, median={median_wait}ms"))

        # Gate 4: Determine verdict
        if hot_locks and starvation:
            verdict = ContentionVerdict.STARVATION_OBSERVED
        elif hot_locks:
            verdict = ContentionVerdict.HOT_LOCK_DETECTED
        else:
            verdict = ContentionVerdict.LOW_CONTENTION

        # Recommended pool size: shrink if hot locks + starvation observed
        if verdict is ContentionVerdict.STARVATION_OBSERVED:
            recommended = max(1, nominal_pool_size // 2)
        elif verdict is ContentionVerdict.HOT_LOCK_DETECTED:
            recommended = max(1, int(nominal_pool_size * 0.75))
        else:
            recommended = nominal_pool_size

        digest_payload = {
            "verdict": verdict.name,
            "hot_locks": tuple(sorted(hot_locks.keys())),
            "starvation_count": str(len(starvation)),
            "pool_size": str(recommended),
        }
        
        digest = cls._hash(
            json.dumps(digest_payload, sort_keys=True, separators=(",", ":"))
        )

        return LockContentionManifest(
            verdict=verdict,
            hot_lock_register=tuple(sorted(hot_locks.items())),
            starvation_signatures=tuple(starvation),
            cyclic_ordering_witness=tuple(),
            recommended_pool_size=recommended,
            manifest_digest=digest,
        )

    @classmethod
    def may_spawn_pool_size(
        cls, manifest: LockContentionManifest, requested_size: int
    ) -> tuple[bool, str]:
        """
        Fetching-time check: confirms a successor may spawn the requested pool
        size without re-triggering contention.
        """
        if requested_size > manifest.recommended_pool_size:
            return False, (
                f"POOL_SIZE_VIOLATION: Requested {requested_size} exceeds "
                f"recommended {manifest.recommended_pool_size} based on "
                f"verdict {manifest.verdict.name}."
            )
        return True, "POOL_SIZE_ADMISSIBLE"