tests/pipeline/test_timeout.py

"""Sprint A14-S8 — timeout depuis le début d'exécution **réelle**.

Le bug critique de l'ancien runner : un document pouvait être marqué
``timeout`` parce qu'il avait passé N secondes en queue, pas N
secondes en train de tourner.  Le nouveau ``CorpusRunner`` mesure
le timeout depuis ``time.monotonic()`` au moment où le worker
démarre réellement (cf. ``CorpusRunner._run_one`` qui écrit
``started_at[doc.id]`` en première instruction).
"""

from __future__ import annotations

import time


from picarones.domain import Artifact, ArtifactType, DocumentRef
from picarones.pipeline import (
    CorpusRunner,
    PipelineExecutor,
    PipelineSpec,
    PipelineStep,
    RunContext,
)


class _SlowAdapter:
    """Adapter qui dort un certain temps avant de retourner."""

    name = "slow"
    input_types = frozenset({ArtifactType.IMAGE})
    output_types = frozenset({ArtifactType.RAW_TEXT})

    def __init__(self, sleep_seconds: float) -> None:
        self._sleep = sleep_seconds

    def execute(self, inputs, params, context, control):
        time.sleep(self._sleep)
        return {
            ArtifactType.RAW_TEXT: Artifact(
                id=f"{context.document_id}:raw_text",
                document_id=context.document_id,
                type=ArtifactType.RAW_TEXT,
            ),
        }


def _build(adapter, *, timeout: float, max_in_flight: int = 2):
    registry = {"slow": adapter}
    exe = PipelineExecutor(adapter_resolver=lambda n: registry[n])
    runner = CorpusRunner(
        exe,
        max_in_flight=max_in_flight,
        timeout_seconds_per_doc=timeout,
        poll_interval_seconds=0.01,
    )
    spec = PipelineSpec(
        name="t", initial_inputs=(ArtifactType.IMAGE,),
        steps=(PipelineStep(
            id="s", kind="ocr", adapter_name="slow",
            input_types=(ArtifactType.IMAGE,),
            output_types=(ArtifactType.RAW_TEXT,),
        ),),
    )
    return runner, spec


def _factories():
    def inputs(doc):
        return {ArtifactType.IMAGE: Artifact(
            id=f"{doc.id}:image",
            document_id=doc.id,
            type=ArtifactType.IMAGE,
        )}

    def ctx(doc):
        return RunContext(
            document_id=doc.id, code_version="1.0.0", pipeline_name="t",
        )
    return inputs, ctx


def test_doc_timed_out_when_exceeds_timeout() -> None:
    """Step qui dort 0.5s, timeout 0.1s → status timed_out."""
    adapter = _SlowAdapter(sleep_seconds=0.5)
    runner, spec = _build(adapter, timeout=0.1, max_in_flight=1)
    inputs, ctx = _factories()
    docs = [DocumentRef(id="slow_one", image_uri="/tmp/x.png")]

    t0 = time.perf_counter()
    result = runner.run(spec, docs, inputs, ctx)
    elapsed = time.perf_counter() - t0

    assert result.n_timed_out == 1
    assert result.outcomes[0].status == "timed_out"
    assert "timeout" in (result.outcomes[0].error or "")
    # Le run principal a rendu la main rapidement (ne s'est pas bloqué
    # sur le sleep complet — le thread continue mais on n'attend plus).
    assert elapsed < 0.3, f"runner s'est bloqué : {elapsed:.2f}s"


def test_timeout_measured_from_real_start_not_submission() -> None:
    """Bug historique : avec un seul worker (max_in_flight=1) et 4
    documents, les 3 derniers attendent en queue.  L'ancien runner
    aurait marqué ces 3 docs timeout dès que la queue dépassait le
    timeout.  Le nouveau runner ne marque timeout que les docs qui
    ont **réellement** dépassé le délai en exécution.

    Marges de robustesse cross-OS
    ------------------------------
    - Adapter dort **0.05s** par doc.
    - Timeout par doc : **2.0s** (40× la durée d'exécution).
    - 4 docs en série = ~0.2s d'exécution totale + overhead du
      ``concurrent.futures`` worker.

    L'ancienne version utilisait timeout=0.5s (10× la durée), ce
    qui flaquait sur les runners GitHub Actions sous charge (le
    scheduler prenait parfois >0.45s pour démarrer un doc en
    queue).  Avec un ratio de 40×, le test ne flaque plus.
    """
    adapter = _SlowAdapter(sleep_seconds=0.05)
    runner, spec = _build(adapter, timeout=2.0, max_in_flight=1)
    inputs, ctx = _factories()
    docs = [DocumentRef(id=f"d{i}") for i in range(4)]

    result = runner.run(spec, docs, inputs, ctx)

    # Les 4 docs auraient pris ~0.2s en série, ce qui dépasse le
    # timeout de 2.0s **si** le runner mesurait depuis la submission
    # du dernier doc.  Mais comme on mesure depuis le début réel
    # de chaque doc, aucun ne devrait timeout.
    assert result.n_succeeded == 4
    assert result.n_timed_out == 0


def test_some_docs_succeed_others_timeout() -> None:
    """Mix : la moitié des docs sont rapides, l'autre lente.  Avec
    un timeout intermédiaire, les rapides réussissent et les lents
    timeout.

    Marges de robustesse cross-OS
    ------------------------------
    - Timeout : **0.5s**.
    - Docs pairs dorment **0.05s** (10× sous le timeout) — ne ratent
      pas même sur runners macOS lents avec scheduler imprécis.
    - Docs impairs dorment **2.0s** (4× au-dessus) — timeout
      garanti.

    L'ancienne version utilisait timeout=0.1s / sleep pair=0.01s
    qui était à 10 ms du timeout — le jitter du scheduler macOS sur
    runners GitHub Actions le faisait basculer aléatoirement.
    """

    class _ConditionalSlow:
        name = "cond"
        input_types = frozenset({ArtifactType.IMAGE})
        output_types = frozenset({ArtifactType.RAW_TEXT})

        def execute(self, inputs, params, context, control):
            # Les docs avec id pair sont rapides.
            if int(context.document_id.removeprefix("d")) % 2 == 0:
                time.sleep(0.05)  # 10× sous le timeout (0.5s)
            else:
                time.sleep(2.0)  # 4× au-dessus du timeout
            return {
                ArtifactType.RAW_TEXT: Artifact(
                    id=f"{context.document_id}:raw_text",
                    document_id=context.document_id,
                    type=ArtifactType.RAW_TEXT,
                ),
            }

    adapter = _ConditionalSlow()
    runner, spec = _build(adapter, timeout=0.5, max_in_flight=2)
    inputs, ctx = _factories()
    docs = [DocumentRef(id=f"d{i}") for i in range(6)]

    result = runner.run(spec, docs, inputs, ctx)
    assert result.n_succeeded == 3, (
        f"pairs (d0/d2/d4) auraient dû réussir, "
        f"obtenu n_succeeded={result.n_succeeded}, "
        f"n_timed_out={result.n_timed_out}"
    )
    assert result.n_timed_out == 3


# ──────────────────────────────────────────────────────────────────────
# Régressions backpressure × timeout (zombies)
# ──────────────────────────────────────────────────────────────────────
#
# Bug historique : quand un doc dépassait son timeout, l'orchestrateur
# enregistrait le ``timed_out``, décrémentait ``in_flight`` et soumettait
# immédiatement le doc suivant — alors que le thread worker continuait
# à occuper son slot du pool (Python ne tue pas un thread).  Le doc
# suivant queue derrière le zombie, sans pouvoir démarrer ni timeout
# (pas de ``started_at``), donc le runner pouvait hang dès que tous
# les slots étaient zombifiés.
#
# Le fix : ``zombie_futures`` compte la capacité réellement occupée,
# et la backpressure refuse de soumettre tant que le slot n'est pas
# libéré (par la fin naturelle du thread zombie).


def test_zombie_does_not_block_runner_when_no_more_docs() -> None:
    """Régression : avec 1 seul doc qui timeout, le runner doit
    retourner sans attendre que le thread zombie finisse.

    Pas de doc à soumettre derrière → on sort tout de suite après
    le timeout enregistré.  Le thread se détache via
    ``pool.shutdown(wait=False)``.
    """
    adapter = _SlowAdapter(sleep_seconds=2.0)
    runner, spec = _build(adapter, timeout=0.1, max_in_flight=1)
    inputs, ctx = _factories()
    docs = [DocumentRef(id="solo")]

    t0 = time.perf_counter()
    result = runner.run(spec, docs, inputs, ctx)
    elapsed = time.perf_counter() - t0

    assert result.n_timed_out == 1
    assert elapsed < 0.5, (
        f"runner a attendu le zombie : {elapsed:.2f}s — devrait "
        f"retourner ~0.1s après le timeout"
    )


def test_zombie_slot_blocks_new_submissions_until_thread_completes() -> None:
    """Régression critique : si tous les slots sont zombifiés, les
    docs suivants ne doivent pas être soumis (ils queue sans pouvoir
    démarrer ni timeout).  La concurrence observée ne doit JAMAIS
    dépasser ``max_in_flight``, même avec des timeouts en cours.

    Construction : 2 workers, 2 premiers docs lents (timeout), puis
    2 docs rapides.  Si le bug revient, les rapides seraient
    submittés alors que les zombies occupent encore le pool — la
    concurrence réelle atteindrait 4 (2 zombies + 2 rapides).
    """
    import threading as _threading

    class _MixedAdapter:
        name = "mixed"
        input_types = frozenset({ArtifactType.IMAGE})
        output_types = frozenset({ArtifactType.RAW_TEXT})

        def __init__(self) -> None:
            self._lock = _threading.Lock()
            self._current = 0
            self.max_observed = 0

        def execute(self, inputs, params, context, control):
            with self._lock:
                self._current += 1
                if self._current > self.max_observed:
                    self.max_observed = self._current
            try:
                # Les 2 premiers docs sont lents → vont timeout
                # (slot zombifié).  Les suivants sont rapides.
                if context.document_id in {"slow0", "slow1"}:
                    time.sleep(1.5)
                else:
                    time.sleep(0.02)
                return {
                    ArtifactType.RAW_TEXT: Artifact(
                        id=f"{context.document_id}:raw_text",
                        document_id=context.document_id,
                        type=ArtifactType.RAW_TEXT,
                    ),
                }
            finally:
                with self._lock:
                    self._current -= 1

    adapter = _MixedAdapter()
    runner, spec = _build(adapter, timeout=0.2, max_in_flight=2)
    # _build n'enregistre que "slow" ; on remplace le registry.
    registry = {"mixed": adapter}
    runner._executor = type(runner._executor)(
        adapter_resolver=lambda n: registry[n]
    )
    spec = type(spec)(
        name=spec.name,
        initial_inputs=spec.initial_inputs,
        steps=(type(spec.steps[0])(
            id="s", kind="ocr", adapter_name="mixed",
            input_types=(ArtifactType.IMAGE,),
            output_types=(ArtifactType.RAW_TEXT,),
        ),),
    )

    inputs, ctx = _factories()
    docs = [
        DocumentRef(id="slow0"),
        DocumentRef(id="slow1"),
        DocumentRef(id="fast0"),
        DocumentRef(id="fast1"),
    ]

    result = runner.run(spec, docs, inputs, ctx)

    # Invariant critique : la concurrence n'a JAMAIS dépassé max=2.
    # Si le bug revient, on observerait 3 ou 4 (zombies + rapides
    # soumis en parallèle).
    assert adapter.max_observed <= 2, (
        f"backpressure violée par les zombies : "
        f"max_observed={adapter.max_observed}, attendu <= 2"
    )
    # Les 2 lents timeoutent, les 2 rapides finissent par réussir
    # une fois que les threads zombies se libèrent.
    assert result.n_timed_out == 2
    assert result.n_succeeded == 2


def test_zombie_capacity_freed_when_thread_completes_naturally() -> None:
    """Quand un thread zombie finit naturellement (après son
    ``time.sleep`` complet), son slot redevient disponible et le doc
    suivant peut alors démarrer.  Vérifie que le runner ne reste pas
    bloqué dans un état "zombies à vie".
    """
    adapter = _SlowAdapter(sleep_seconds=0.3)  # finit avant le 2e doc
    runner, spec = _build(adapter, timeout=0.1, max_in_flight=1)
    inputs, ctx = _factories()
    docs = [DocumentRef(id="zombie"), DocumentRef(id="follower")]

    result = runner.run(spec, docs, inputs, ctx)

    # Le 1er doc timeout (0.1 < 0.3).  Le 2e doc doit finir par
    # démarrer après que le zombie se termine à t≈0.3s, et tourner
    # 0.3s à son tour → également timeout (l'adapter est le même).
    assert result.n_timed_out == 2, (
        f"attendu 2 timeouts (zombie + follower), obtenu "
        f"n_timed_out={result.n_timed_out}, "
        f"n_succeeded={result.n_succeeded}"
    )
    # Outcomes ordonnés par achèvement : le zombie est enregistré
    # immédiatement à t=0.1, le follower à t≈0.7 (0.1 + 0.3 + 0.3).
    assert [o.document_id for o in result.outcomes] == [
        "zombie", "follower",
    ]