# `src/worker/` — SQS + SFN waitForTaskToken + ECS task protection framework

A long-running worker framework for ECS Fargate services that consume from SQS and report back to Step Functions via `sqs:sendMessage.waitForTaskToken`. Owns the SQS poll loop, ECS task protection lifecycle, SFN heartbeat thread, and subprocess spawn/lifecycle/cancellation. Generic over the consumer's request schema; treats the subprocess result as an opaque JSON object forwarded to SFN.

This package has zero imports from outside itself. The intent is to extract it to its own repo and publish to the internal PyPI server once a second consumer adopts the same pattern. Until then, it lives here.

## Public API

```python
from src.worker import (
    Worker,                      # parent-side: SQS poll, dispatcher, subprocess lifecycle
    WorkerRequest,               # base class for typed request schemas
    NonEmptyStr,                 # `Annotated[str, StringConstraints(min_length=1, strip_whitespace=True)]`
    run_handler,                 # subprocess-side: parse env-var, run handler, emit result/error
    AbortedStaleTask,            # framework exceptions
    MalformedMessage,
    MalformedSubprocessOutput,
    SubprocessCrashError,
    UnknownSubprocessError,
)
```

## Architecture

The framework splits the work across two processes:

- **Parent** (long-lived ECS task): `Worker` polls SQS, enables/disables ECS task protection per message, spawns a subprocess per message, runs a heartbeat thread, sends `send_task_success` / `send_task_failure` to SFN.
- **Subprocess** (one per message, fresh Python process): runs `run_handler(request_schema, handler)`, which parses the request from an env var, calls the consumer's handler, writes the result JSON to stdout (or an error envelope on exception).

The split exists so the parent can `killpg(SIGKILL)` the entire subprocess tree if SFN's task token goes stale mid-validation — cooperative cancellation can't be enforced against blocking I/O (e.g. an HTTP call inside a validator), so a hard kill of a separate process group is the only reliable mechanism.

## Minimal consumer example

**`app.py`** — parent-side wiring:

```python
import logging
from pydantic import ConfigDict
from src.worker import Worker, SubprocessCrashError, NonEmptyStr, WorkerRequest


class MyRequest(WorkerRequest):
    model_config = ConfigDict(frozen=True, extra="ignore")

    asset_upload_type: NonEmptyStr


class UnsupportedAssetUploadType(Exception):
    pass


def _select_subprocess_module(request: MyRequest) -> str:
    match request.asset_upload_type:
        case "atmos":
            return "src.atmos.validator"
        case _:
            raise UnsupportedAssetUploadType(f"unsupported: {request.asset_upload_type}")


def make_worker() -> Worker[MyRequest]:
    return Worker(
        request_schema=MyRequest,
        select_subprocess_module=_select_subprocess_module,
        sqs_queue_url=...,
        heartbeat_interval_seconds=15.0,
        message_processing_timeout_minutes=30,
        exception_levels={
            UnsupportedAssetUploadType: logging.ERROR,
            SubprocessCrashError: logging.WARNING,
            # ... domain exceptions raised by handler ...
        },
    )


def main() -> None:
    make_worker().run()
```

**`atmos/validator.py`** — subprocess-side handler:

```python
from pydantic import BaseModel
from src.worker import run_handler, WorkerRequest, NonEmptyStr


class MyDomainRequest(WorkerRequest):
    asset_upload_type: NonEmptyStr
    bucket: NonEmptyStr
    key: NonEmptyStr


class MyDomainResult(BaseModel):
    is_valid: bool


def _handle(request: MyDomainRequest) -> MyDomainResult:
    # ... domain work ...
    return MyDomainResult(is_valid=True)


def main() -> None:
    run_handler(MyDomainRequest, _handle)


if __name__ == "__main__":
    main()
```

Parent `request_schema` needs only the routing fields. Use `extra="ignore"` so domain fields pass through to the subprocess for strict validation.

## Protocol

The parent and subprocess communicate over an environment variable (request) and stdout (response), with stderr reserved for human-readable diagnostics:

| Direction | Channel | Format |
|---|---|---|
| parent → subprocess | env var `WORKER_REQUEST` | raw SQS message body |
| subprocess → parent (success) | stdout | any JSON object, exit 0 |
| subprocess → parent (failure) | stdout | `ErrorOutput(error=ClassName, message=str(exc)).model_dump_json()`, exit 1 |
| subprocess → parent (diagnostics) | stderr | traceback (on failure) + free-form text the parent logs |

## Key invariants

1. **SFN is the sole retry source.** When the SFN heartbeat times out, SFN re-drives the state with a new task token (and a new SQS message). The worker never reuses an SQS message and never re-processes one. SQS visibility timeout is *not* used as a retry mechanism.
2. **Delete-on-pickup.** Once the worker holds task protection for a message, it deletes the SQS message immediately. No safety net of redelivery — but no risk of two workers racing on the same task token either.
3. **Subprocess cancellation is hard kill.** The heartbeat thread calls `os.killpg(os.getpgid(process.pid), signal.SIGKILL)` if SFN reports the token stale. The subprocess runs in its own process group (`start_new_session=True`) so the kill takes the whole subprocess tree, not just the Python orchestrator.
4. **Task protection wraps message handling.** `enable()` is called before the subprocess spawn; `disable()` runs in a `finally` after. ECS won't terminate the task while protection is held, so SIGTERM during a rolling deploy/scale-in is deferred to the next loop iteration.
5. **Worker-owned exceptions are filtered out of subprocess dispatch.** A buggy or malicious subprocess emitting `{"error": "AbortedStaleTask"}` cannot fake a worker-internal control-flow exception — those names are excluded from the name → class registry at construction time.

## Failure taxonomy

Framework-owned exceptions (raised by `Worker`, never dispatched from subprocess output):

| Exception | When | Behavior |
|---|---|---|
| `MalformedMessage` | SQS body fails `request_schema` validation | Best-effort `send_task_failure` if a non-empty `task_token` string is present in the JSON body; message deleted regardless |
| `AbortedStaleTask` | Heartbeat thread killed the subprocess after SFN reported stale token | Silent return — SFN already moved on |
| `SubprocessCrashError` | Subprocess exited via signal (other than the heartbeat-driven kill) | `send_task_failure` |
| `MalformedSubprocessOutput` | Subprocess stdout was not a JSON object, or the `ErrorOutput` envelope on exit-1 failed to parse | `send_task_failure` |
| `UnknownSubprocessError` | Subprocess emitted an `error` name not in the consumer's `exception_levels` registry | `send_task_failure` |

Consumer-defined exceptions (registered via `exception_levels`, dispatched from subprocess output by class name):

- Anything the handler raises. The class name in the error envelope is looked up in `_subprocess_exception_for_name` (built from `exception_levels` minus framework exceptions), and the corresponding class is re-raised in the parent before being routed to `send_task_failure`. Log level for each exception comes from the same map.

## Extraction notes

- No imports from outside `src.worker.*` — verifiable via `grep -rn "from src\." src/worker/ | grep -v "from src.worker"` (returns zero matches).
- Third-party deps: `boto3`, `mypy_boto3_sqs`, `pydantic`. Standard pip packages.
- After extraction, the package becomes the library's root and this README becomes the library's README.