Deploy workflow workers as an ephemeral ECS Fargate fleet — Closes #33

[conradbzura]

Summary

Add the runtime application substrate the workflow subsystem needs to run on the ECS Fargate fleet outlined in #32: an EcsProvisioner that launches ephemeral worker containers per workflow via RunTask, an EcsDiscovery poll-and-diff loop over ListTasks + DescribeTasks that surfaces healthy tasks to Wool's worker pool, an S3Cache backend that mirrors LocalFsCache semantics over boto3, a worker-container entrypoint, and a workflow heartbeat that lets multi-hour preprocessing runs survive the stale-reclaim predicate. The same boto3-backed code targets real AWS in production and LocalStack in dev — only AWS_ENDPOINT_URL differs.

Two WoolExecutor defaults move with the multi-hour profile: DEFAULT_WORKFLOW_DURATION_CAP_SECONDS rises from 20 min to 4 h (env-driven), and _NO_WORKERS_RETRY_ATTEMPTS widens from 5 to 60 to cover the Fargate cold-start budget. STALE_WORKFLOW_THRESHOLD drops from 1 h to 15 min — paired with a 5-min heartbeat sidecar in the executor, that catches actually-dead workers fast without false-positive reclaiming legitimate long sorts.

Scope intentionally narrows to the runtime application code so it lands self-contained with full unit coverage. Defer for follow-up PRs once the runtime classes have integration coverage and a deploy pipeline exists: LocalStack docker-compose.yml, CloudFormation worker task-def updates, multi-stage worker Dockerfile + SOCI index, README documentation, and LocalStack-backed integration tests.

Closes #33

Proposed changes

S3 cache backend

Add S3Cache (src/cfdb/workflows/cache.py) as a CacheBackend over boto3 with head_object / get_object (with Range) / upload_file / delete_object. Support an optional key prefix for sharing a single bucket across environments, reject path-traversal segments, and yield an empty iterator on missing objects to match LocalFsCache semantics. Centralize client construction in build_s3_client so production, LocalStack-backed dev, and unit tests all produce the same shape of client — only the endpoint differs.

ECS provisioner and worker discovery

Add EcsProvisioner (src/cfdb/workflows/provisioner.py) wrapping RunTask with awsvpc network configuration. Dedupe concurrent calls keyed on the workflow mutex key so a burst of ensure_workflow calls on the same source does not fan out into multiple tasks, and guard the ~20 req/s RunTask rate limit with a semaphore. CapacityException covers both ClientError- and failures[].reason-shaped capacity / ENI errors so the executor can surface them as retryable rather than hanging.

Add EcsDiscovery (src/cfdb/workflows/discovery.py) implementing Wool's DiscoveryLike protocol with a poll-and-diff loop over list_tasks + describe_tasks. Filter on healthStatus: HEALTHY, extract the awsvpc IP, and emit worker-added / worker-dropped events to non-blocking subscribers via per-subscriber asyncio.Queue. Replay state on subscribe so subscribers attached after startup observe the existing healthy fleet.

Worker container entrypoint

Add worker_main.py (src/cfdb/workflows/worker_main.py) starting wool.LocalWorker, exposing a tiny aiohttp /health endpoint that returns 503 during drain so ECS marks the task unhealthy before stop_task kills gRPC, installing SIGTERM/SIGINT handlers, and self-terminating after a configurable idle timeout. Deliberately omit discovery registration — EcsDiscovery polls ECS directly, so the worker only needs to bind its gRPC port and stay HEALTHY.

Workflow heartbeat

Add heartbeat_workflow (src/cfdb/workflows/lock.py) bumping updated_at on an active job record so a multi-hour run does not trip claim_workflow's stale-reclaim predicate. Fence the update on an active status so a heartbeat against a terminal or stale-reclaimed record is a silent no-op — the executor uses the False return as a stop signal. Shorten STALE_WORKFLOW_THRESHOLD from 1 h to 15 min and expose DEFAULT_HEARTBEAT_INTERVAL_SECONDS as the cadence shared between executor and tests.

Executor integration

Extend WoolExecutor with provisioner and heartbeat_interval_seconds ctor args. Invoke the provisioner from ensure_workflow on a fresh claim and surface CapacityException as a terminal FAILED job with a retryable error string. Spawn a heartbeat sidecar task that bumps updated_at while the workflow body runs and cancel it in finally before the terminal release so the next tick cannot observe an active record while status is flipping. Raise DEFAULT_WORKFLOW_DURATION_CAP_SECONDS from 1200 to 4 h and widen _NO_WORKERS_RETRY_ATTEMPTS from 5 to 60 (~60 s window) for Fargate cold-start. Drop the LambdaExecutor mention from the JobExecutor ABC docstring.

API lifespan wiring

Add env vars to src/cfdb/api/__init__.py: AWS_ENDPOINT_URL, AWS_REGION, WORKFLOW_S3_BUCKET / PREFIX, ECS_CLUSTER, ECS_WORKER_TASK_DEFINITION, ECS_WORKER_SUBNETS, ECS_WORKER_SECURITY_GROUPS, ECS_WORKER_ASSIGN_PUBLIC_IP, WORKFLOW_DURATION_CAP_SECONDS. Have the lifespan pick S3Cache over LocalFsCache when WORKFLOW_S3_BUCKET is set and build an EcsProvisioner when ECS env config is present (helper _maybe_build_provisioner). Leave the bare PoC profile unchanged: with none of the new env set, the API continues to use the local filesystem cache plus the in-process WorkerPool.

Build dependencies

Add boto3 to runtime deps and moto[ecs,s3] to dev deps so the unit tests can exercise boto3 paths without LocalStack or real AWS.

Test cases

#	Test Suite	Given	When	Then	Coverage Target
1	TestS3Cache	An S3Cache constructed with empty bucket	The constructor runs	ValueError is raised	Constructor input validation
2	TestS3Cache	A bucket containing no object for the key	head is awaited	None is returned	Cache-miss head
3	TestS3Cache	A small payload uploaded via put	head is awaited for the same key	The reported size matches the payload	Round-trip put/head
4	TestS3Cache	A cached object	get is iterated without a byte range	The full payload is yielded	Full-object streaming
5	TestS3Cache	A cached object	get is iterated with an inclusive byte range	Only the slice is yielded	Range-aware streaming
6	TestS3Cache	A bucket with no object for the key	get is iterated	An empty stream is produced	Cache-miss get
7	TestS3Cache	A cached object	delete is awaited	True is returned and the object is gone	Existing-key deletion
8	TestS3Cache	A bucket with no object for the key	delete is awaited	False is returned	Absent-key deletion
9	TestS3Cache	A key containing a .. segment	put is awaited	ValueError is raised	Path-traversal rejection
10	TestS3Cache	An S3Cache constructed with a key prefix	Round-trip put/head is awaited	Object lands under the prefix	Prefix application
11	TestEcsProvisioner	A provisioner constructed without a cluster	The constructor runs	ValueError is raised	Constructor input validation
12	TestEcsProvisioner	A provisioner constructed without a task definition	The constructor runs	ValueError is raised	Constructor input validation
13	TestEcsProvisioner	A provisioner constructed without subnets	The constructor runs	ValueError is raised	Constructor input validation
14	TestEcsProvisioner	A configured provisioner	request is awaited once	RunTask is called once with the expected payload	Single-caller dispatch
15	TestEcsProvisioner	A provisioner with concurrent calls sharing a dedup key	Many request calls await in parallel	RunTask is invoked once and all callers receive the same ARN	In-flight dedup
16	TestEcsProvisioner	A provisioner with concurrent calls under distinct dedup keys	Many request calls await in parallel	RunTask is invoked once per key	Dedup boundary
17	TestEcsProvisioner	A boto3 client raising ClientError for capacity	request is awaited	CapacityException is raised	Capacity error path
18	TestEcsProvisioner	A RunTask response with capacity in failures[]	request is awaited	CapacityException is raised	Failure-payload capacity path
19	TestEcsDiscovery	A discovery constructed without a cluster	The constructor runs	ValueError is raised	Constructor input validation
20	TestEcsDiscovery	A discovery constructed without a task family	The constructor runs	ValueError is raised	Constructor input validation
21	TestEcsDiscovery	An ECS client returning healthy tasks	_poll_once is awaited	worker-added events are emitted with the expected addresses	Initial healthy fleet
22	TestEcsDiscovery	An ECS client returning a non-healthy task	_poll_once is awaited	The unhealthy task is filtered out	Health filter
23	TestEcsDiscovery	A previously seen worker absent on a later poll	_poll_once is awaited again	worker-dropped is emitted for the gone worker	Drop diff
24	TestEcsDiscovery	A steady-state ECS client	_poll_once is awaited twice	No duplicate events are emitted	Idempotent steady state
25	TestEcsDiscovery	A subscriber with a filter rejecting a worker	The discovery emits an add event	The filtered subscriber receives nothing	Subscriber filter
26	TestParseArgs	A clean environment	_parse_args runs without args	Documented defaults are returned	Default arg parsing
27	TestParseArgs	Worker env vars set	_parse_args runs without args	Env values override defaults	Env override
28	TestParseArgs	Worker env vars set and CLI flags passed	_parse_args runs with the flags	CLI values override env	CLI precedence
29	TestHeartbeatWorkflow	An active job whose updated_at has been backdated	heartbeat_workflow is awaited	True is returned and updated_at advances	Heartbeat happy path
30	TestHeartbeatWorkflow	A job already released to COMPLETED	heartbeat_workflow is awaited	False is returned	Heartbeat stop signal
31	TestStaleWorkflowThreshold	The current STALE_WORKFLOW_THRESHOLD constant	Its value is read	It equals 15 minutes	Threshold guard
32	TestWoolExecutorWithProvisioner	An executor with a recording provisioner	ensure_workflow is awaited for a fresh key	The provisioner observes one request keyed by the workflow mutex key	Provisioner dispatch
33	TestWoolExecutorWithProvisioner	A provisioner that always raises CapacityException	ensure_workflow is awaited	The job is FAILED with a "capacity"-prefixed error	Capacity-failure handling
34	TestWoolExecutorHeartbeat	An executor with a sub-second heartbeat cadence and a blocking processor	ensure_workflow is awaited and the test polls during the run	updated_at advances past the original claim timestamp	In-flight heartbeat