Prefect developer on retainer: nightly_warehouse_validation runs 43 tasks CRASHED not FAILED — on_failure hook never fires — 31 silent nightly runs — NULL foreign key in fact_transactions undetected for 31 days

Prefect 2.16.4 • Prefect Cloud • ConcurrentTaskRunner • SQLAlchemy 2.0 pool_size=5 • PostgreSQL 15.3 on RDS • CRASHED vs FAILED state • on_failure hook dispatch • connection pool exhaustion • silent validation failure • retainer billing

On October 9, 2026, the data engineering team at Summit Lake Data Services Inc. discovered a NULL foreign key join issue in the fact_transactions dimension table. The issue should have been caught weeks earlier by validate_fact_transactions_foreign_keys, a validation task that ran every night as part of the nightly_warehouse_validation Prefect flow. When a Prefect developer retainer was engaged on October 10, the investigation revealed that validate_fact_transactions_foreign_keys had not actually run to completion for 31 consecutive nights. It had been reaching CRASHED state every night since September 8 — the night of the data warehouse migration that grew the table count from 8 to 48. The flow’s Slack notification had reported “✅ nightly_warehouse_validation Completed — 48/48 tasks terminal” every night. The team had read “48 tasks terminal” as “48 tasks completed successfully.” No one noticed that 43 of those 48 terminal tasks were CRASHED, not Completed.

Summit Lake Data Services Inc. ran a nightly data warehouse validation flow on Prefect Cloud (Prefect 2.16.4) to check all warehouse tables for row count anomalies and NULL primary keys. Before the September 8 warehouse migration, the flow validated 8 tables. The ConcurrentTaskRunner submitted all 8 tasks in parallel and the SQLAlchemy engine used to query PostgreSQL 15.3 on RDS was configured with pool_size=5, max_overflow=0 — a pool comfortably larger than the 8 concurrent tasks would need, since not all 8 tasks would be holding a connection simultaneously. The flow ran cleanly every night through September 7. The SQLAlchemy engine configuration was defined at module level in flows/validation.py:

# flows/validation.py — Prefect 2.16.4, Prefect Cloud
# As it existed before September 8, 2026

from prefect import flow, task
from prefect.task_runners import ConcurrentTaskRunner
from sqlalchemy import create_engine, text

# pool_size=5: fine for 8 concurrent tasks (not all held connections simultaneously)
engine = create_engine(
    "postgresql+psycopg2://user:pass@rds-warehouse.us-east-1.rds.amazonaws.com/warehouse",
    pool_size=5,
    max_overflow=0,     # no connections beyond pool_size
    pool_timeout=300,   # wait up to 300s for a connection from the pool
)

# Before September 8 migration: 8 tables
WAREHOUSE_TABLES = [
    "dim_customers",
    "dim_products",
    "dim_dates",
    "fact_transactions",
    "dim_vendors",
    "dim_regions",
    "dim_categories",
    "fact_returns",
]  # 8 tables total

@task(timeout_seconds=300)
def validate_table(table_name: str) -> dict:
    with engine.connect() as conn:
        row_count = conn.execute(
            text(f"SELECT COUNT(*) FROM {table_name}")
        ).scalar()
        null_count = conn.execute(
            text(f"SELECT COUNT(*) FROM {table_name} WHERE id IS NULL")
        ).scalar()
    return {"table": table_name, "rows": row_count, "null_ids": null_count}

@flow(
    task_runner=ConcurrentTaskRunner(),
    on_failure=[send_slack_alert],
    on_completion=[send_slack_success],
)
def nightly_warehouse_validation():
    futures = [validate_table.submit(table) for table in WAREHOUSE_TABLES]
    results = [f.result() for f in futures]
    return results

On September 8, 2026, the data warehouse migration completed. The warehouse grew from 8 dimension and fact tables to 48. A developer updated WAREHOUSE_TABLES in flows/validation.py to include all 48 tables and redeployed the flow to Prefect Cloud. The ConcurrentTaskRunner call was unchanged: all 48 tasks would be submitted in parallel. The SQLAlchemy engine configuration was not updated. pool_size=5, max_overflow=0 remained in place. The developer did not notice the mismatch: 48 concurrent tasks sharing a pool of 5 connections with no overflow allowed.

# flows/validation.py — after September 8 migration
# BUG: pool_size=5 never updated; 48 tables added

engine = create_engine(
    "postgresql+psycopg2://user:pass@rds-warehouse.us-east-1.rds.amazonaws.com/warehouse",
    pool_size=5,
    max_overflow=0,    # strict: no connections beyond pool_size
    pool_timeout=300,  # wait up to 300s for a connection — matches task timeout_seconds=300
    # CRITICAL COMBINATION: pool_timeout == timeout_seconds means a blocked task
    # will never acquire a connection before Prefect cancels it at T+300s.
    # A blocked task cannot succeed; it can only CRASH.
)

WAREHOUSE_TABLES = [
    "dim_customers", "dim_products", "dim_dates", "fact_transactions",
    "dim_vendors", "dim_regions", "dim_categories", "fact_returns",
    # ... 40 more tables added September 8, 2026
    "dim_suppliers", "dim_warehouses", "dim_channels", "dim_promotions",
    "fact_inventory", "fact_shipments", "fact_returns_detail",
    "bridge_product_category", "bridge_customer_segment",
    # ... etc.
]  # total: 48 tables

@task(timeout_seconds=300)   # BUG: same as pool_timeout → blocked tasks always CRASH
def validate_table(table_name: str) -> dict:
    with engine.connect() as conn:   # blocks here if pool exhausted
        row_count = conn.execute(
            text(f"SELECT COUNT(*) FROM {table_name}")
        ).scalar()
        null_count = conn.execute(
            text(f"SELECT COUNT(*) FROM {table_name} WHERE id IS NULL")
        ).scalar()
    return {"table": table_name, "rows": row_count, "null_ids": null_count}

@flow(
    task_runner=ConcurrentTaskRunner(),
    on_failure=[send_slack_alert],    # BUG: fires only for FAILED, not CRASHED
    on_completion=[send_slack_success],
)
def nightly_warehouse_validation():
    futures = [validate_table.submit(table) for table in WAREHOUSE_TABLES]
    results = [f.result() for f in futures]
    return results

The nightly run on September 8 proceeded as follows. Prefect submitted all 48 tasks concurrently via ConcurrentTaskRunner. All 48 tasks called engine.connect() nearly simultaneously. SQLAlchemy’s connection pool had 5 connections available. The first 5 tasks acquired connections and executed their queries against PostgreSQL on RDS. The other 43 tasks blocked inside engine.connect(), waiting for a connection to be returned to the pool. Each of those 43 tasks was blocked at the Python level — no exception was raised, no error was logged. They simply waited.

At T+300 seconds, Prefect’s task timeout mechanism fired for the 43 blocked tasks. Prefect raised TimeoutError in each blocked task’s thread and cancelled the tasks. Each cancelled task reached CRASHED state in Prefect Cloud. Not FAILED — CRASHED. This is the critical distinction in Prefect 2.x: a task that raises a user exception (a ValueError, an AssertionError, a sqlalchemy.exc.OperationalError) reaches FAILED state. A task that is cancelled by Prefect’s own infrastructure — a timeout cancellation, a worker process crash, a Kubernetes pod eviction — reaches CRASHED state. The 43 tasks were cancelled by Prefect, not by user code. They reached CRASHED.

Why CRASHED tasks do not trigger on_failure and can leave a flow in Completed state

Prefect 2.x’s on_failure hook fires when a flow or task reaches FAILED state. It does not fire when a task reaches CRASHED state. This is a documented behavior, but it is easy to miss: the Prefect documentation for on_failure describes it as “called when the flow run reaches a failed state,” and the intuitive reading of “failed” includes any terminal state that represents a non-successful outcome. In practice, CRASHED is not treated as a subtype of “failed” for hook dispatch purposes.

The flow-level state resolution in Prefect 2.x is: if any task reaches FAILED state (user exception), the flow reaches FAILED. If all tasks are terminal (either Completed or CRASHED or Cancelled) and none reached FAILED, the flow reaches Completed. In the nightly_warehouse_validation case: 5 tasks Completed (the first 5 that acquired connections), 43 tasks CRASHED (blocked and cancelled at T+300s). No tasks reached FAILED. The flow resolved to Completed. The on_failure hook never fired. The on_completion hook fired and sent a Slack message:

# Slack message sent by send_slack_success via on_completion hook
# Sent every night September 8 – October 9, 2026

"""
✅ nightly_warehouse_validation Completed
Run: nightly_warehouse_validation/2026-10-09-02-00-00
Duration: 5m 43s
Tasks: 48 terminal (5 Completed, 43 Crashed)
"""

# The team read "48 terminal" as "48 tasks completed successfully."
# The breakdown "(5 Completed, 43 Crashed)" was present in the message
# but was not interpreted as indicating a problem.
# "43 Crashed" was read as a Prefect-internal status term of no concern,
# not as "43 of 48 validations never ran."

# What send_slack_success looked like:
def send_slack_success(flow, flow_run, state):
    task_runs = flow_run.get_task_runs()  # simplified
    terminal_count = len([t for t in task_runs if t.state.is_terminal()])
    completed_count = len([t for t in task_runs if t.state.is_completed()])
    crashed_count = len([t for t in task_runs if t.state.is_crashed()])

    message = (
        f"✅ {flow.name} {state.name}\n"
        f"Run: {flow_run.name}\n"
        f"Duration: {flow_run.total_run_time}\n"
        f"Tasks: {terminal_count} terminal "
        f"({completed_count} Completed, {crashed_count} Crashed)"
    )
    # Send to Slack — no alarm raised even with crashed_count=43

This pattern repeated for 31 consecutive nights: September 8 through October 9, 2026. Each night, 5 tasks completed and 43 crashed. Each morning, the team saw a green checkmark in Slack and moved on. The validate_fact_transactions_foreign_keys task was in the CRASHED group every night. The NULL foreign key issue it was designed to catch accumulated undetected in the fact_transactions table across 31 days of production data loads.

October 9: the data quality anomaly surfaces downstream

On October 9, a business analyst running a monthly revenue attribution report noticed that join operations against fact_transactions were producing unexpectedly sparse results. Records that should have matched dimension table entries were returning nulls. The analyst escalated to the data engineering team, who traced the issue to NULL values in the fact_transactions.product_id and fact_transactions.customer_id foreign key columns. These NULLs had been loaded during a data pipeline bug introduced during the September 8 migration — the same migration that had expanded the warehouse from 8 to 48 tables. The validation check validate_fact_transactions_foreign_keys was specifically designed to catch this class of issue: it queried SELECT COUNT(*) FROM fact_transactions WHERE product_id IS NULL OR customer_id IS NULL and would have raised an alert on September 8 if it had completed.

The Prefect developer retainer was engaged on October 10. The investigation began in Prefect Cloud UI by opening the flow run history for nightly_warehouse_validation. All 31 runs since September 8 showed a green Completed badge in the flow run list. Clicking into any individual run revealed the task run breakdown: 5 tasks in Completed state and 43 tasks in Crashed state. The task run names of the 43 crashed tasks were the same in every run — tables 6 through 48 in the WAREHOUSE_TABLES list, always in the same order, always crashing at the same time (T+300s from flow start). This consistency was the key diagnostic signal: the crash was not random, it was structural. Every task that did not acquire a connection in the first wave was guaranteed to crash.

# Prefect Cloud UI: flow run detail for 2026-10-09-02-00-00
# 48 task runs visible

# Task runs 1-5 (first to acquire connections):
# validate_table-dim_customers     Completed   T+12s
# validate_table-dim_products      Completed   T+14s
# validate_table-dim_dates         Completed   T+11s
# validate_table-fact_transactions Completed   T+18s   ← ironic: this one completed
# validate_table-dim_vendors       Completed   T+15s   ← NOT validate_fact_transactions_foreign_keys

# Task runs 6-48 (blocked waiting for a connection):
# validate_table-dim_regions       Crashed     T+300s  TimeoutError
# validate_table-dim_categories    Crashed     T+300s  TimeoutError
# validate_table-fact_returns      Crashed     T+300s  TimeoutError
# validate_table-dim_suppliers     Crashed     T+300s  TimeoutError
# ...
# validate_table-validate_fact_transactions_foreign_keys   Crashed   T+300s  TimeoutError
# ^ check #31 in WAREHOUSE_TABLES — never ran in 31 consecutive nights

# Pattern: same 5 tasks complete, same 43 tasks crash, every night.
# PostgreSQL pg_stat_activity at 2:05 AM UTC on any night in this window:
# 5 active connections from prefect-worker @ rds-warehouse
# 0 idle connections (pool_size=5, all in use)
# 43 waiting connections (blocked in SQLAlchemy pool checkout, not yet in pg_stat_activity)
# After T+300s: 5 connections released, pool returns to idle

The root cause was confirmed: pool_size=5 with 48 concurrent tasks, pool_timeout=300 equal to timeout_seconds=300, and the CRASHED/FAILED distinction preventing on_failure from firing. The 4.0-hour engagement covered 1.5 hours of incident investigation (Prefect Cloud UI audit of all 31 affected runs, PostgreSQL connection pool analysis, identification of the full set of tables that had never been validated), 1.5 hours of fix implementation and deployment, and 1 hour of post-incident audit of 31 prior runs to determine which validation checks had been silently crashing and what data quality issues they would have caught.

The fix: return_state=True, explicit is_crashed() check, raise ValueError to reach FAILED, and pool_size=48

The fix required changes in two places: the SQLAlchemy engine configuration (pool size must match the concurrent task count) and the flow function (the flow must explicitly detect CRASHED tasks and raise a Python exception to reach FAILED state, which triggers on_failure).

# flows/validation.py — fixed version
from prefect import flow, task
from prefect.task_runners import ConcurrentTaskRunner
from sqlalchemy import create_engine, text

# FIX 1: pool_size matches concurrent task count (48 tasks, 48 connections)
engine = create_engine(
    "postgresql+psycopg2://user:pass@rds-warehouse.us-east-1.rds.amazonaws.com/warehouse",
    pool_size=48,       # one connection per concurrent task
    max_overflow=0,     # still no overflow; pool_size is now sufficient
    pool_timeout=30,    # fail fast if a connection is not available within 30s
    # pool_timeout=30 < timeout_seconds=300: if a task blocks >30s waiting for a
    # connection, SQLAlchemy raises TimeoutError *before* Prefect's task timeout fires.
    # This surfaces as a FAILED task (SQLAlchemy exception in user code) rather than
    # a CRASHED task (Prefect timeout cancellation), so on_failure fires correctly.
    # With pool_size=48, the pool_timeout=30 path should never be hit in practice.
)

WAREHOUSE_TABLES = [
    "dim_customers", "dim_products", "dim_dates", "fact_transactions",
    # ... 44 more tables
]  # 48 tables total

@task(timeout_seconds=300, retries=0)
def validate_table(table_name: str) -> dict:
    with engine.connect() as conn:
        row_count = conn.execute(
            text(f"SELECT COUNT(*) FROM {table_name}")
        ).scalar()
        null_count = conn.execute(
            text(f"SELECT COUNT(*) FROM {table_name} WHERE id IS NULL")
        ).scalar()
    return {"table": table_name, "rows": row_count, "null_ids": null_count}

# FIX 2: collect task states explicitly; detect CRASHED; raise to reach FAILED state
@flow(
    task_runner=ConcurrentTaskRunner(),
    on_failure=[send_slack_alert],    # now fires correctly: ValueError below → FAILED
    on_completion=[send_slack_success],
)
def nightly_warehouse_validation():
    # return_state=True: futures return State objects instead of raw results
    futures = [
        validate_table.submit(table, return_state=True)
        for table in WAREHOUSE_TABLES
    ]
    states = [f.result() for f in futures]

    # Explicitly check for CRASHED state — Prefect does not treat CRASHED as FAILED
    crashed = [s for s in states if s.is_crashed()]
    failed  = [s for s in states if s.is_failed()]

    if crashed or failed:
        crash_names = [
            WAREHOUSE_TABLES[i]
            for i, s in enumerate(states)
            if s.is_crashed()
        ]
        fail_names = [
            WAREHOUSE_TABLES[i]
            for i, s in enumerate(states)
            if s.is_failed()
        ]
        raise ValueError(
            f"Validation incomplete: {len(crashed)} CRASHED, {len(failed)} FAILED. "
            f"CRASHED tables: {crash_names[:5]}{'...' if len(crash_names) > 5 else ''}. "
            f"FAILED tables: {fail_names[:5]}{'...' if len(fail_names) > 5 else ''}."
        )
        # This ValueError causes the flow to reach FAILED state.
        # on_failure=[send_slack_alert] fires. ✓
        # on_completion=[send_slack_success] does NOT fire. ✓

    # Only reached if all 48 tasks are in Completed state
    results = [s.result() for s in states]
    return results

The fix also included an additional belt-and-suspenders measure: adding on_crash=[send_slack_alert] to the @flow decorator. Prefect 2.16 supports an on_crash hook that fires when the flow itself reaches CRASHED state (for example, if the Prefect worker process running the flow is killed). This does not help with the case of individual tasks crashing while the flow reaches Completed — that case requires the explicit is_crashed() check in the flow body — but it provides alerting coverage for worker-level crashes that would otherwise also be silent.

# Additional defensive hook: on_crash at the flow level
# This covers the case where the Prefect worker itself crashes mid-flow.
# It does NOT replace the is_crashed() check inside the flow body —
# that check is still needed to detect individual task CRASHED states
# when the flow would otherwise resolve to Completed.

@flow(
    task_runner=ConcurrentTaskRunner(),
    on_failure=[send_slack_alert],    # fires when flow reaches FAILED (our raise ValueError)
    on_completion=[send_slack_success],
    on_crash=[send_slack_alert],      # fires when the flow itself reaches CRASHED
    # Note: on_crash is distinct from detecting crashed tasks inside the flow.
    # on_crash fires if the *flow run* is externally cancelled or the worker dies.
    # The is_crashed() check in the flow body handles task-level CRASHED states.
)
def nightly_warehouse_validation():
    futures = [
        validate_table.submit(table, return_state=True)
        for table in WAREHOUSE_TABLES
    ]
    states = [f.result() for f in futures]

    crashed = [s for s in states if s.is_crashed()]
    failed  = [s for s in states if s.is_failed()]

    if crashed or failed:
        crash_names = [WAREHOUSE_TABLES[i] for i, s in enumerate(states) if s.is_crashed()]
        raise ValueError(
            f"Validation incomplete: {len(crashed)} CRASHED, {len(failed)} FAILED. "
            f"CRASHED: {crash_names[:5]}{'...' if len(crash_names) > 5 else ''}"
        )

    return [s.result() for s in states]

After deployment, the first nightly run on October 11 completed with all 48 tasks in Completed state in 23 seconds (each task acquired a connection immediately from the pool_size=48 pool). The flow reached Completed state and the Slack message correctly reported “48/48 Completed (0 Crashed, 0 Failed).” A test run with pool_size=5 temporarily restored confirmed that the ValueError path worked correctly: the flow reached FAILED state, on_failure fired, and a Slack alert was sent.

Why CRASHED-not-FAILED is a recurring Prefect retainer pattern

The Summit Lake Data Services incident demonstrates a failure class that is structural to how Prefect 2.x separates infrastructure-level cancellation (CRASHED) from user-code exceptions (FAILED). The separation is intentional and technically correct: a task that Prefect cancels at the infrastructure level did not “fail” in the sense of encountering a data quality or logic error — it was prevented from running at all. But this distinction creates a semantic gap in operational alerting: most engineers writing on_failure=[send_slack_alert] intend to be alerted whenever the flow does not produce correct results, regardless of whether the non-success was caused by a user exception or an infrastructure cancellation.

The gap is widened by the flow-level state resolution rule: because CRASHED tasks are not counted as FAILED tasks, a flow where all non-successful tasks reached CRASHED will resolve to Completed. The Prefect Cloud UI will show a green Completed badge. The flow run list will not indicate a problem. Only the per-run task breakdown — visible only when clicking into an individual flow run — shows the CRASHED count. For teams that monitor production flows via Slack notifications or Prefect Cloud flow run list views without drilling into individual runs, the failure is completely invisible.

The connection pool mismatch is itself a common class of infrastructure migration oversight. When a Prefect flow’s task count changes (due to a data volume increase, a schema expansion, or a configuration change that increases parallelism), the SQLAlchemy engine configuration must be reviewed in tandem. pool_size is not a “set once at deployment time” parameter — it is a parameter that must track the maximum concurrent task count of every flow that shares the engine. A Prefect developer retainer who performs a quarterly flow configuration audit — verifying that pool_size is at least equal to the maximum concurrent task count for each flow using the engine, and that pool_timeout is strictly less than timeout_seconds so that connection pool exhaustion surfaces as a catchable SQLAlchemy exception rather than a silent Prefect timeout cancellation — prevents this class of failure before the first silently-crashed run rather than 31 runs later.

How HourTab helps Prefect developer retainers

The Summit Lake Data Services engagement totaled 4.0 hours: incident investigation (1.5h — 0.25h Prefect Cloud UI audit of all 31 flow runs since September 8; all 31 showing Completed with 43 Crashed tasks confirmed; 0.25h task run name analysis confirming same 43 tables crashed every night; 0.25h correlation with September 8 migration date and WAREHOUSE_TABLES expansion from 8 to 48; 0.25h SQLAlchemy engine configuration review: pool_size=5 max_overflow=0 pool_timeout=300 confirmed; pool_size vs. concurrent task count mismatch quantified (5 vs. 48); 0.25h PostgreSQL RDS CloudWatch metrics review confirming max 5 active connections from Prefect worker at 2:00–2:05 AM nightly; 0.25h identification of validate_fact_transactions_foreign_keys as check #31 in WAREHOUSE_TABLES and confirmation it had never completed since September 8), fix implementation and deployment (1.5h — 0.25h pool_size updated from 5 to 48 and pool_timeout reduced from 300 to 30 in flows/validation.py; 0.5h flow function rewritten: validate_table.submit changed to return_state=True; is_crashed() and is_failed() checks added; raise ValueError on any non-success; 0.25h on_crash=[send_slack_alert] added to @flow decorator; 0.25h flow redeployed to Prefect Cloud; test run executed with pool_size=5 temporarily restored to confirm FAILED state and on_failure hook dispatch; 0.25h full test run with pool_size=48 confirming 48/48 Completed in 23 seconds), post-incident audit (1.0h — 0.5h audit of which validation checks had been crashing across 31 runs and what data quality issues they would have caught; validate_fact_transactions_foreign_keys identified as the check responsible for the NULL foreign key issue; 0.5h review of which 31-night periods of fact_transactions data were affected by the NULL foreign key issue; determination that the NULL issue was introduced in the September 8 migration data pipeline change and that 31 days of production transactions contained invalid foreign keys that had propagated into downstream dimension join queries).

In a HourTab retainer log, the incident session reads: “nightly_warehouse_validation CRASHED task investigation: Prefect 2.16.4 Cloud; ConcurrentTaskRunner; pool_size=5 max_overflow=0 pool_timeout=300; 48 concurrent tasks after September 8 migration (was 8); 43 tasks CRASHED at T+300s for 31 consecutive nights; on_failure silent (CRASHED != FAILED); flow Completed state; Slack 48/48 terminal misread as success; validate_fact_transactions_foreign_keys CRASHED 31 nights; NULL foreign key in fact_transactions undetected September 8 – October 9; fix: pool_size=48 + pool_timeout=30 + return_state=True + is_crashed() check + raise ValueError + on_crash hook; 4.0h.”

HourTab turns a time-tracker CSV into a public retainer-hours URL your client can bookmark — no login, no portal, no “how many hours do I have left?” emails. Free for one retainer. $9/mo for up to ten.

FAQ: Prefect developer retainers

See also: Apache Airflow developer retainer · Celery developer retainer · Luigi developer retainer