config.py

"""Validated runtime configuration for every workflow task.

Environment variables are strings and every Workflow task runs on a fresh
instance, so configuration is parsed at the task boundary rather than cached at
module import time. Invalid values fail before a run row or watermark can be
created.
"""

from __future__ import annotations

import os
from dataclasses import dataclass


def _integer(
    name: str,
    default: int,
    *,
    minimum: int,
    maximum: int | None = None,
) -> int:
    raw = os.getenv(name, str(default))
    try:
        value = int(raw)
    except ValueError as exc:
        raise ValueError(f"{name} must be an integer, got {raw!r}") from exc

    if value < minimum:
        raise ValueError(f"{name} must be at least {minimum}, got {value}")
    if maximum is not None and value > maximum:
        raise ValueError(f"{name} must be at most {maximum}, got {value}")
    return value


@dataclass(frozen=True)
class SyncConfig:
    """The source-independent settings that define one sync run."""

    slice_count: int
    overlap_seconds: int
    initial_watermark: str
    page_size: int
    statement_timeout_ms: int
    lease_seconds: int

    @classmethod
    def from_env(cls) -> SyncConfig:
        initial_watermark = os.getenv("SYNC_INITIAL_WATERMARK", "7d").strip()
        if not initial_watermark:
            raise ValueError("SYNC_INITIAL_WATERMARK must not be empty")

        return cls(
            # Keep the ceiling explicit: an accidental extra zero should fail
            # before it consumes a workspace's entire concurrent-run allowance.
            slice_count=_integer("SYNC_SLICE_COUNT", 8, minimum=1, maximum=128),
            # A negative overlap moves the watermark into the future and creates
            # a permanent gap. Zero is valid for sources with no late arrivals.
            overlap_seconds=_integer("SYNC_OVERLAP_SECONDS", 300, minimum=0),
            initial_watermark=initial_watermark,
            page_size=_integer("SYNC_PAGE_SIZE", 100, minimum=1, maximum=10_000),
            statement_timeout_ms=_integer(
                "SYNC_STATEMENT_TIMEOUT_MS",
                120_000,
                minimum=1,
            ),
            # Longer than any leaf task timeout (the longest checkpointed fetch
            # task is 6,000 seconds) plus headroom. This makes expiry evidence
            # that the owner cannot still be writing, not a guess based on age.
            lease_seconds=_integer(
                "SYNC_LEASE_SECONDS",
                9_000,
                minimum=7_500,
            ),
        )


def sync_config() -> SyncConfig:
    """Parse a fresh immutable snapshot for the current task invocation."""
    return SyncConfig.from_env()