Extraction worker¶

The extraction worker turns one uploaded episode archive into playable video, frame rows, and episode metadata. It is a run-to-completion Fargate task: EventBridge starts it when a raw archive lands in S3, it processes one queue message, and it exits.

The archive is a zstd-compressed tar (.tar.zst). Only objects whose key ends in .zst wake the worker. The key must be orgs/<org_id>/episodes/<episode_id>/…. The episode row already exists; the worker updates it. It does not create the episode.

How a run works¶

 robot upload (.tar.zst)
        │
        ▼
 data bucket ──► EventBridge ──► one Fargate task
        │
        └──► SQS queue ──► worker long-polls one message
                                │
                                ├─ download, zstd, untar
                                ├─ .mp4 / .parquet  →  …/processed/…
                                ├─ .mp4             →  HLS under …/processed/hls/<camera>/
                                ├─ data/*.parquet   →  frames
                                └─ meta/*           →  episode columns + episode_metadata
                                │
                                ▼
                    status = processed, then delete the SQS message

The task connects to Aurora. If that fails, the process exits and the message stays on the queue.
It long-polls SQS for up to five attempts (20 seconds each, 3 seconds between attempts). The visibility timeout is 15 minutes. With no message, it exits.
An s3:TestEvent is deleted and ignored. Any other body is an S3 event. The first record's key is decoded and split. A key that is not orgs/<org_id>/episodes/<episode_id>/… is logged and left on the queue.
The episode status becomes processing. The object is downloaded, decompressed with zstd, and streamed out of the tar. The whole archive is not held expanded on disk.
Each .mp4 and .parquet is uploaded again under the same prefix plus /processed/, keeping the path it had inside the tar. Other files are skipped, except meta/info.json and meta/tasks.parquet.
When that work finishes, the episode is marked processed and the SQS message is deleted. The temp directory is removed.

EventBridge passes the triggering bucket and key into the task as S3_BUCKET and S3_OBJECT_KEY. The worker does not read those variables. It takes whichever message SQS returns first.

What it writes¶

Episode row (episodes), set together with status = 'processed' and processed_at:

Column	Source
`mp4_s3_path`	S3 key of the first `.mp4` uploaded
`parquet_s3_path`	S3 key of the first `data/` parquet uploaded
`frame_count`	`total_frames` in `meta/info.json`
`fps`	`fps` in `meta/info.json`
`duration_seconds`	`frame_count / fps`, rounded to an integer

Frames. Every parquet whose path contains /data/ is read in order. Rows are inserted into frames in batches of 1,000.

`frames` column	Parquet field
`timestamp_ms`	`timestamp` (seconds), converted to milliseconds
`observation`	`observation.state`, stored as `{ "state": […] }`
`action`	`action`

Video. Each .mp4 is transcoded to HLS with FFmpeg: baseline profile, level 3.0, 4-second segments, playlist playlist.m3u8. The camera name is the folder two levels above the file inside the tar, or cam_main when the path is shorter. Segments go to:

orgs/<org_id>/episodes/<episode_id>/processed/hls/<camera>/

The playlist prefix and the probed resolution are not stored on the episode. A video or parquet error is logged and the episode can still be marked processed.

Metadata (episode_metadata), written after the episode is already processed. A failure here does not change the episode status. Re-processing the same episode replaces these keys.

`meta_key`	Value	Status
`episode_title`	First row of `meta/tasks.parquet`, column `__index_level_0__`	`non_editable`
`robot_type`	`robot_type` in `meta/info.json`	`non_editable`
`codebase_version`	`codebase_version` in `meta/info.json`	`non_editable`
`operator_name`, `operator_id`, `task`, `group`, `setting`, `lighting`, `clutter`, `success_score`, `descriptions`	Fixed placeholder strings	`editable`

The same metadata list is stored as version 1 of the episode in entity_versions (operation = 'create'). A later run overwrites that version 1 row.

When it fails¶

The message is deleted only after a successful run. A thrown error sets the episode to failed and leaves the message on the queue. After 3 receives (the queue's max_receive_count) SQS moves it to the dead-letter queue. A CloudWatch alarm emails the shared alerts topic when that queue is not empty. Message retention is 4 days on the main queue and 14 days on the dead-letter queue.

Per-file video and parquet errors, and the metadata write, are caught inside the run. They do not set failed and they do not keep the message.

Deploy¶

A push to dev that touches extraction-worker/** builds the image and pushes :latest and :<git sha> to ECR (toborverse-dev-extraction-worker in us-west-2). The task definition uses :latest, so the next EventBridge run picks up the new image. ECR keeps the three most recent images.

The image is Node 20 plus FFmpeg and zstd. The default task is 1 vCPU and 2 GB, on private subnets, with logs in /ecs/<name-prefix>-worker for 7 days. DB_PASSWORD is injected from the Aurora secret at start. When DB_IAM_AUTH=true the worker ignores that password and connects with an IAM auth token, which is what dev requires because the database user has rds_iam. The other database settings are plain environment variables: DB_HOST, DB_PORT, DB_NAME, DB_USER, DB_SSL=true. The queue URL is SQS_QUEUE_URL.

Where the code is¶

Path	Contents
`extraction-worker/index.js`	Poll, extract, upload, status updates
`extraction-worker/src/services/parquetService.js`	Frame insert and episode title
`extraction-worker/src/services/videoService.js`	HLS transcode and upload
`extraction-worker/src/models/`	`episodes`, `frames`, `episode_metadata`, `entity_versions`
`extraction-worker/Dockerfile`	Node 18, FFmpeg, zstd
`.github/workflows/deploy-extraction-worker-dev.yml`	Build and push to ECR
`infra/terraform/modules/data_ingestion/`	Bucket notification, EventBridge, ECS task, SQS, DLQ alarm