Should Rivus Jobs Use Prefect 3?

Feb 27, 2026 · Research + hands-on spike · Consulted: Codex 3, Gemini 3.1 Pro, Grok Code

1. The Problem

The rivus jobs system is a custom async Python pipeline engine: ~15 jobs discover items from external sources (YouTube, Finnhub, web scraping), process them through multi-stage pipelines, and store structured results. The core is ~6,700 LOC across 6 files, with ~14,000 LOC total including handlers and UI.

It works well, but the codebase has grown complex:

• tracker.py — 1,682 LOC, 69 functions (god object)
• discovery.py — 1,805 LOC, 17 strategies with heavy boilerplate
• runner.py — 1,284 LOC, 13-parameter run_job()

The question: would adopting a standard framework like Prefect 3 reduce complexity, or just shift it?

2. What Prefect 3 Actually Is

Prefect is a Python workflow orchestration framework. You decorate functions with @task and @flow, and Prefect handles scheduling, retries, state tracking, and observability.

Architecture

Key Facts (verified Feb 2026)

3. How It Maps to Our Jobs System

4. How Stages Would Work

Each stage becomes a @task. Data flows implicitly — a task that consumes another task's return value automatically waits for it:

@task(retries=3, retry_delay_seconds=[2, 5, 10], tags=["youtube"])
async def fetch_transcript(video: dict) -> dict:
    async with concurrency("youtube", occupy=1):
        vtt = await download_captions(video["video_id"])
    return {**video, "transcript_path": vtt}

@task(retries=1)
async def extract_metadata(video: dict) -> dict:
    # Runs AFTER fetch_transcript because it consumes its output
    return await parse_vtt(video["transcript_path"])

@flow
async def dumb_money_pipeline(videos: list[dict]):
    for video in videos:
        fetched = await fetch_transcript(video)     # stage 1
        extracted = await extract_metadata(fetched)  # stage 2 (waits for 1)

Fan-out with .map()

Process multiple items in parallel across a stage:

@flow
async def pipeline(videos: list[dict]):
    # Fan out: one task per video, all run concurrently
    fetched_futures = fetch_transcript.map(videos)
    # Chain: each extract gets the corresponding fetch result
    extracted_futures = extract_metadata.map(fetched_futures)
    return [f.result() for f in extracted_futures]

How concurrency limits apply

Stage dependencies

Our jobs system supports stage_deps — stage C requires stages A and B to both complete. In Prefect, this is implicit via data flow: if a task's arguments come from two other tasks, it waits for both:

@flow
async def pipeline(item):
    a_result = await stage_a(item)
    b_result = await stage_b(item)       # runs concurrently with A
    c_result = await stage_c(a_result, b_result)  # waits for BOTH

5. How Discovery Would Work

Discovery is about finding new items to process. In our system, each job has a discovery strategy (e.g. youtube_channel, finnhub_calendar) that runs periodically.

Option A: Scheduled discovery flow

@flow
async def discover_dml_videos():
    """Runs every 60 minutes, finds new videos, triggers pipeline."""
    videos = await youtube_channel_strategy("@DumbMoneyLive")
    new_videos = filter_already_processed(videos)  # check tracker DB
    if new_videos:
        await dumb_money_pipeline(new_videos)

# Schedule it
discover_dml_videos.serve(name="dml-discovery", interval=3600)

Option B: Separate discovery + processing

# Discovery writes to tracker DB
@flow(name="discover")
async def discover():
    items = await strategy.discover()
    tracker.upsert_items(items)  # still our SQLite

# Processing reads from tracker DB (separate scheduled flow)
@flow(name="process")
async def process_pending():
    pending = tracker.get_pending("dumb_money_live", limit=10)
    for item in pending:
        tracker.mark_in_progress(item)
        await pipeline(item)
        tracker.mark_done(item)

6. Pushing Jobs & Items In

Three ways to inject work into a Prefect-based system:

A. API trigger (push a flow run)

# From CLI or external script
$ prefect deployment run 'dml-pipeline/production' \
    --param '{"video_id": "abc123", "url": "https://..."}'

# From Python
from prefect.deployments import run_deployment
run_deployment("dml-pipeline/production", parameters={"video_id": "abc123"})

B. Direct function call

# Just call the flow function — works without server
await dumb_money_pipeline([{"video_id": "abc123", ...}])

C. Via our existing jobctl / tracker

# Write to tracker DB, let scheduled flow pick it up
jobctl inject dml --stage fetch --key "abc123" --data '{"url":"..."}'
# The process_pending flow picks it up on next cycle

Option C is what we'd likely use — it preserves our existing jobctl workflow and doesn't require the Prefect server for injection.

7. Multi-Machine Scaling

This was asked specifically: does Prefect scale to multiple machines?

How it works

Setup

# Machine A: start server with Postgres (SQLite won't work multi-machine)
export PREFECT_API_DATABASE_CONNECTION_URL=postgresql+asyncpg://...
prefect server start --host 0.0.0.0

# Machine B: point worker at server
export PREFECT_API_URL=http://machine-a:4200/api
prefect worker start --pool my-pool

# Machine C: same thing
export PREFECT_API_URL=http://machine-a:4200/api
prefect worker start --pool my-pool

What scales and what doesn't

Bottom line: Prefect itself scales to multiple machines easily. But our handlers assume local disk and SQLite, so real multi-machine would require more work than just adding workers.

8. What the Spike Revealed

We built a proof-of-concept migrating the dumb_money_live pipeline to Prefect 3. See jobs/spike_prefect.py for the full code.

What worked well

What didn't work or felt awkward

9. Side-by-Side Comparison

10. The Alternatives (With Redis)

We consulted three frontier models (Codex 3, Gemini 3.1 Pro, Grok Code) with Redis on the table. Here's how the options ranked:

Why the spike changed the picture

The multi-model consensus said Prefect 3 was the clear winner. But the hands-on spike revealed that the theoretical gains evaporate when you account for the tracker gap:

Net: we'd save ~980 LOC of runner code and gain 73MB of dependencies + a server process. The tracker, discovery, doctor, and dashboard all stay custom.

The refactor alternative

All three models also recommended a refactor of the custom system (approach B) that achieves similar or better results with zero new dependencies:

11. Recommendation

What would change my recommendation

• Multi-machine execution becomes a real need → Prefect or arq+Redis
• Team grows and new developers need familiar tools → Prefect (industry standard)
• Observability pressure (need OpenTelemetry traces, better dashboards) → Prefect

What wouldn't change it

• "Prefect is the standard" — standards matter for teams, less for a single developer
• "We should use Redis for queuing" — our SQLite WAL tracker is simpler and faster for single-machine
• "The runner code is ugly" — refactoring is cheaper and less risky than a framework migration

12. Glossary

Global Concurrency Limits

Prefect's server-side rate limiting. Created via prefect gcl create youtube --limit 4. Tasks acquire slots via async with concurrency("youtube"). Coordinates across all workers connected to the same server.

Flow

A Python function decorated with @flow. The top-level unit of work in Prefect. Contains tasks and orchestration logic. Equivalent to our "job".

Task

A Python function decorated with @task. An individual step within a flow. Has its own retry policy, timeout, and failure hooks. Equivalent to our "stage handler".

WAL (Write-Ahead Logging)

SQLite journaling mode that allows concurrent readers and a single writer. Our tracker uses WAL mode for safe concurrent access from the runner and dashboard.

Ephemeral mode

Running Prefect without a server. Flows execute in-process with temporary state. Loses the UI, scheduling, and global concurrency limits. Good for development, not production.

Worker

A Prefect process that polls the server for scheduled flow runs and executes them. Can run on any machine that can reach the server API.

Deployment

A named, configured instance of a flow. Specifies schedule, parameters, and work pool. Created via prefect.yaml or flow.serve().