Ray developer on retainer: ray.get() 30-second timeout cuts batch job to 8% of expected records — partial results written as complete — 6-month monthly undercount — Ray developer on monthly retainer

Ray 2.9.3 • AWS EC2 m5.2xlarge head + 6× m5.xlarge workers • Python 3.11 • @ray.remote(num_cpus=0.5) • ray.get(refs, timeout=30) • GetTimeoutError caught silently • partial results written as complete • 6-month monthly undercount • retainer billing

On October 10, 2026, the ML team at Apex Compute Analytics LLC loaded six months of monthly feature files into their annual model retraining pipeline and immediately flagged an anomaly: the combined record count across all six files was 4,800,000 instead of the expected 60,000,000. The files spanned April through September 2026 and were produced each month by run_monthly_feature_extraction.py, a Ray batch job that had been running without modification since March 2026. No error had been reported. No alert had fired. The monthly job had completed with exit code 0 every time. A Ray developer retainer was engaged on October 10 to investigate. The root cause was a single argument on a single function call: timeout=30 in ray.get(refs, timeout=30).

Apex Compute Analytics ran a Ray 2.9.3 cluster on AWS EC2 — a head node on an m5.2xlarge and six worker nodes on m5.xlarge instances, giving the cluster 24 total CPUs (6 workers × 4 cores). The monthly batch job submitted 100 tasks using a @ray.remote-decorated function extract_features with num_cpus=0.5 fractional CPU allocation. With num_cpus=0.5, Ray could schedule up to 48 tasks concurrently across 24 CPUs. The job prepared 100 record batches of 10,000 records each — 1,000,000 records total per monthly run — and submitted all 100 tasks at once. Each extract_features task performed feature engineering on its batch: numerical normalization, categorical encoding, interaction term computation, and writing intermediate results to the Ray object store. On a warm cluster, most tasks completed in 45–90 seconds depending on batch composition.

# run_monthly_feature_extraction.py — Ray 2.9.3, Python 3.11
# As it existed from March 2026 through October 2026

import ray
import boto3
import pandas as pd
from datetime import datetime

RAY_ADDRESS = "ray://ray-head.internal:10001"
S3_BUCKET = "apex-compute-features"

@ray.remote(num_cpus=0.5)
def extract_features(record_batch: pd.DataFrame) -> pd.DataFrame:
    """Feature extraction for a single record batch (10,000 records)."""
    # Numerical normalization
    numeric_cols = record_batch.select_dtypes("number").columns
    record_batch[numeric_cols] = (
        record_batch[numeric_cols] - record_batch[numeric_cols].mean()
    ) / record_batch[numeric_cols].std()
    # Categorical encoding, interaction terms, etc.
    # ... (45–90 seconds of compute per batch on m5.xlarge workers)
    return record_batch

def run_monthly_extraction(year: int, month: int) -> None:
    ray.init(address=RAY_ADDRESS)

    s3 = boto3.client("s3")
    record_batches = load_record_batches(s3, year, month)  # returns list of 100 DataFrames

    # Submit all 100 tasks
    refs = [extract_features.remote(batch) for batch in record_batches]

    # Collect results — with a 30-second timeout
    results = []
    try:
        results = ray.get(refs, timeout=30)  # ← THE BUG
    except ray.exceptions.GetTimeoutError:
        results = [r for r in results if r is not None]  # ← silently discards timed-out refs

    # Write results to S3 — no record count check
    output_df = pd.concat(results, ignore_index=True)
    output_key = f"monthly_features_{year}_{month:02d}.parquet"
    output_df.to_parquet(f"/tmp/{output_key}")
    s3.upload_file(f"/tmp/{output_key}", S3_BUCKET, output_key)
    print(f"Wrote {output_key} ({len(output_df):,} records)")

if __name__ == "__main__":
    now = datetime.utcnow()
    run_monthly_extraction(now.year, now.month)

The timeout=30 parameter had been added in the original March 2026 implementation by an engineer who was unfamiliar with Ray’s ray.get() timeout semantics. The intent was defensive: prevent the job from hanging indefinitely if the cluster became unresponsive. A 30-second ceiling seemed reasonable — the engineer assumed that if most tasks had not completed within 30 seconds, something was wrong with the cluster and partial results would be a reasonable fallback. What was not understood was how Ray’s timeout interacts with task completion timing on a cluster that is still healthy and actively processing all 100 tasks.

With num_cpus=0.5 and 24 cluster CPUs, Ray can schedule up to 48 tasks concurrently. The scheduler assigns tasks to workers in order of worker availability. Workers with faster network connections to the object store tend to receive tasks first and complete them first. In the Apex cluster, the first 8 tasks — assigned to workers that started immediately and had no object store contention — completed within the 30-second window. The remaining 92 tasks were in active execution when the 30-second timeout fired. Those 92 tasks were not stuck or hung. They were running correctly and would have completed in 45–90 seconds. But ray.get() does not wait for them. It raises ray.exceptions.GetTimeoutError for all remaining refs simultaneously. The exception handler then filters out None values from the partially-populated results list and returns a list of 8 completed DataFrames. Those 8 DataFrames — 80,000 records — were concatenated and written to S3 as the complete monthly feature file.

Why ray.get() timeout fires on all refs simultaneously

Ray’s ray.get(refs, timeout=N) behavior is frequently misunderstood. The timeout is not a per-task deadline — it is a wall-clock deadline for the entire ray.get() call. When the timeout fires, GetTimeoutError is raised immediately for the entire list of refs, regardless of how many tasks have completed. The tasks that have not completed are not cancelled — they continue running on the cluster, consuming worker CPU and object store memory. Their results will eventually be placed in the object store when they finish. But since the calling code is no longer tracking those ObjectRefs (the refs list is out of scope after the exception handler exits), those results are never retrieved. They sit in the object store until Ray’s object store garbage collector reclaims them.

The correct pattern for collecting results from a long-running batch where some tasks may be slower than others is ray.wait(), which allows incremental result collection as tasks complete, with optional timeout per individual task rather than a global timeout on the entire batch. Alternatively, if the goal is truly to bound total execution time, ray.get() can be called without a timeout parameter, and a separate monitoring mechanism (a watchdog thread, an external timeout on the job runner) can terminate the job if the entire batch does not complete within a reasonable wall-clock window — but in that case, the job should fail explicitly rather than write a partial result.

# Diagnostic: check S3 monthly feature file sizes and record counts
# Run on October 10, 2026 after ML team flags the record count anomaly

import boto3
import pandas as pd

s3 = boto3.client("s3")
BUCKET = "apex-compute-features"

months = [
    (2026, 4), (2026, 5), (2026, 6),
    (2026, 7), (2026, 8), (2026, 9),
]

print(f"{'File':<45} {'Size (MB)':>10} {'Records':>12} {'Expected':>12}")
print("-" * 82)

total_records = 0
for year, month in months:
    key = f"monthly_features_{year}_{month:02d}.parquet"

    # Check S3 object size
    attrs = s3.get_object_attributes(
        Bucket=BUCKET,
        Key=key,
        ObjectAttributes=["ObjectSize"],
    )
    size_mb = attrs["ObjectSize"] / (1024 * 1024)

    # Count records in parquet
    obj = s3.get_object(Bucket=BUCKET, Key=key)
    df = pd.read_parquet(obj["Body"])
    record_count = len(df)
    total_records += record_count

    flag = " ← UNDERSIZED" if size_mb < 20 else ""
    print(f"{key:<45} {size_mb:>9.1f}  {record_count:>12,}  {1_000_000:>12,}{flag}")

print("-" * 82)
print(f"{'TOTAL':<45} {'':>10} {total_records:>12,}  {6_000_000:>12,}")

# Output:
# File                                          Size (MB)      Records     Expected
# ----------------------------------------------------------------------------------
# monthly_features_2026_04.parquet                    2.1       80,000    1,000,000 ← UNDERSIZED
# monthly_features_2026_05.parquet                    2.1       80,000    1,000,000 ← UNDERSIZED
# monthly_features_2026_06.parquet                    2.1       80,000    1,000,000 ← UNDERSIZED
# monthly_features_2026_07.parquet                    2.1       80,000    1,000,000 ← UNDERSIZED
# monthly_features_2026_08.parquet                    2.1       80,000    1,000,000 ← UNDERSIZED
# monthly_features_2026_09.parquet                    2.1       80,000    1,000,000 ← UNDERSIZED
# ----------------------------------------------------------------------------------
# TOTAL                                                         480,000    6,000,000

# Confirm Ray cluster logs show GetTimeoutError on every monthly run:
# ray job logs --address ray://ray-head.internal:10001 --job-id 
# ...
# ray.exceptions.GetTimeoutError: 92 object(s) timed out: [ObjectRef(...), ...]
# (exception caught by except clause — not propagated to job runner)

The diagnostic confirmed the pattern immediately. All six monthly files were exactly 2.1 MB — consistent with 80,000 records at the expected per-record size. Each file contained exactly 8 of the 100 expected record batches (8 × 10,000 = 80,000 records). The consistency across all six months was diagnostic: the same 8 tasks were consistently completing within 30 seconds every month. Ray’s task scheduler assigns tasks to workers in a deterministic order based on worker availability at submission time. The same 8 workers that happened to be idle at the moment the 100 tasks were submitted received their tasks first, started immediately, and finished within the 30-second window. The other 92 workers — which received tasks a few seconds later due to scheduling queue depth — were still in the 10–20 second range of their 45–90 second task duration when the timeout fired.

October 10: ML team detects 4.8 million records instead of 60 million during annual model retraining

The ML team discovered the undercount on October 10, 2026, when they began loading the six monthly feature files for annual model retraining. The retraining pipeline expected 60,000,000 total records (6 months × 1,000,000 records per month) to meet the minimum training set size threshold of 50,000,000 records for the feature model. When the pipeline loaded all six files and counted total records before beginning training, it received 4,800,000 — 8% of the expected total. The pipeline raised a ValueError: training set size 4,800,000 is below minimum threshold 50,000,000 and halted. The ML team initially suspected a data loading bug in the retraining pipeline itself. When they manually downloaded one of the monthly files and counted its rows, they confirmed 80,000 records instead of 1,000,000. They escalated to the data engineering team, which engaged the Ray developer retainer.

The Ray developer retainer investigation began in the Ray cluster job logs, which showed ray.exceptions.GetTimeoutError on every monthly run going back to April 2026. The exception had been raised and caught on every execution. The timeout=30 parameter in the source code was confirmed. Checking the task timing in the Ray Dashboard for the most recent test run confirmed that extract_features tasks were completing in 45–90 seconds under normal cluster load — well beyond the 30-second timeout. The 5.0-hour engagement covered 1.5 hours of investigation (cluster log analysis confirming GetTimeoutError on all 6 monthly runs; S3 file size and record count diagnostic; task timing confirmation from Ray Dashboard; source code review confirming timeout=30 and exception handler pattern; confirmation that all 92 timed-out tasks do complete eventually but results are discarded), 2.0 hours of fix implementation (timeout removed from ray.get(); record count assertion added after ray.get(); post-write S3 file size assertion added; code reviewed and deployed), and 1.5 hours of validation and re-run (all 6 monthly batches re-run with fixed code; all 6 output files confirmed 26 MB and 1,000,000 records; ML team retraining pipeline confirmed loading 60,000,000 total records).

The fix: remove the timeout, add record count validation, add S3 size assertion

The fix required three changes: removing the timeout=30 parameter from ray.get() so that all 100 tasks are collected before proceeding; adding an explicit record count assertion after ray.get() to catch any future partial result scenarios; and adding a post-write S3 file size assertion to detect undersized output files before the job exits successfully. The fix also added an optional ray.wait()-based progress reporting pattern to give visibility into long-running batches without imposing a global timeout.

# run_monthly_feature_extraction.py — fixed version

import ray
import boto3
import pandas as pd
from datetime import datetime

RAY_ADDRESS = "ray://ray-head.internal:10001"
S3_BUCKET = "apex-compute-features"
EXPECTED_BATCHES = 100
EXPECTED_RECORDS_PER_BATCH = 10_000
MIN_OUTPUT_SIZE_BYTES = 20 * 1024 * 1024  # 20 MB minimum (actual expected ~26 MB)

@ray.remote(num_cpus=0.5)
def extract_features(record_batch: pd.DataFrame) -> pd.DataFrame:
    """Feature extraction for a single record batch (10,000 records)."""
    numeric_cols = record_batch.select_dtypes("number").columns
    record_batch[numeric_cols] = (
        record_batch[numeric_cols] - record_batch[numeric_cols].mean()
    ) / record_batch[numeric_cols].std()
    # Categorical encoding, interaction terms, etc.
    return record_batch

def run_monthly_extraction(year: int, month: int) -> None:
    ray.init(address=RAY_ADDRESS)

    s3 = boto3.client("s3")
    record_batches = load_record_batches(s3, year, month)

    if len(record_batches) != EXPECTED_BATCHES:
        raise ValueError(
            f"Expected {EXPECTED_BATCHES} record batches, got {len(record_batches)}. "
            f"Aborting to prevent partial output."
        )

    # Submit all 100 tasks
    refs = [extract_features.remote(batch) for batch in record_batches]

    # FIX 1: Remove timeout=30 — collect ALL results before proceeding.
    # If the cluster is unhealthy and tasks hang, this will block until Ray's
    # internal task timeout (configured separately) fires, or the job runner's
    # external wall-clock timeout terminates the process. Either failure mode
    # is preferable to silently writing partial results.
    #
    # Optional: use ray.wait() for progress reporting without a global timeout:
    done_refs = []
    remaining_refs = list(refs)
    while remaining_refs:
        newly_done, remaining_refs = ray.wait(remaining_refs, num_returns=1, timeout=None)
        done_refs.extend(newly_done)
        print(f"Progress: {len(done_refs)}/{EXPECTED_BATCHES} tasks complete", flush=True)

    # Retrieve all results — all tasks are confirmed done, so this is instantaneous
    results = ray.get(done_refs)  # FIX 1: no timeout parameter

    # FIX 2: Validate result count before writing
    if len(results) != EXPECTED_BATCHES:
        raise ValueError(
            f"ray.get() returned {len(results)} results, expected {EXPECTED_BATCHES}. "
            f"Aborting to prevent partial output being written to S3."
        )

    output_df = pd.concat(results, ignore_index=True)
    expected_records = EXPECTED_BATCHES * EXPECTED_RECORDS_PER_BATCH

    # FIX 3: Validate record count in concatenated output
    if len(output_df) != expected_records:
        raise ValueError(
            f"Concatenated output has {len(output_df):,} records, "
            f"expected {expected_records:,}. "
            f"Aborting to prevent partial output being written to S3."
        )

    output_key = f"monthly_features_{year}_{month:02d}.parquet"
    local_path = f"/tmp/{output_key}"
    output_df.to_parquet(local_path)
    s3.upload_file(local_path, S3_BUCKET, output_key)

    # FIX 4: Post-write S3 file size assertion
    attrs = s3.get_object_attributes(
        Bucket=S3_BUCKET,
        Key=output_key,
        ObjectAttributes=["ObjectSize"],
    )
    actual_size = attrs["ObjectSize"]
    if actual_size < MIN_OUTPUT_SIZE_BYTES:
        raise ValueError(
            f"S3 file {output_key} is {actual_size / 1024 / 1024:.1f} MB, "
            f"below minimum {MIN_OUTPUT_SIZE_BYTES / 1024 / 1024:.0f} MB threshold. "
            f"File may be corrupt or incomplete. Manual review required."
        )

    print(
        f"Wrote {output_key}: {len(output_df):,} records, "
        f"{actual_size / 1024 / 1024:.1f} MB"
    )

if __name__ == "__main__":
    now = datetime.utcnow()
    run_monthly_extraction(now.year, now.month)

After deploying the fix, all six monthly batches (April through September 2026) were re-run against the source data. Each run completed in approximately 95 seconds — all 100 tasks collected via ray.wait() before writing — and each output file was confirmed at 26 MB and 1,000,000 records. The ML team’s annual retraining pipeline was re-run on October 11 and successfully loaded 60,000,000 total records. Model training completed without error.

Why ray.get() timeout with silent exception handling is a recurring Ray retainer pattern

The Apex Compute Analytics incident is an example of a failure class that is structurally invisible in Ray’s default observability surfaces. Ray’s job runner reports job success or failure based on whether the Python process exits with a non-zero exit code. If GetTimeoutError is caught and not re-raised, the job exits cleanly. The Ray Dashboard shows the job as SUCCEEDED. The task completion view in the Dashboard shows 100 tasks submitted and 100 tasks completed — because all 100 extract_features tasks eventually complete on the cluster, they are just not waited on by the calling code. There is no Ray-native signal that distinguishes “100 tasks submitted, 8 results collected before timeout, 92 results discarded” from “100 tasks submitted, 100 results collected successfully.”

The only signals were external to Ray: the S3 output file was 2.1 MB instead of 26 MB, and the record count in the parquet file was 80,000 instead of 1,000,000. Both signals require downstream consumers of the S3 output to notice the anomaly. In the Apex case, the downstream consumer was the annual model retraining pipeline, which ran only once per year. Six months of monthly undercounts accumulated before the retraining pipeline triggered the detection. A Ray developer retainer who audits all ray.get() calls for timeout parameters, reviews GetTimeoutError exception handlers for silent partial-result returns, and validates that post-write output assertions are in place for all batch job S3 outputs prevents this class of silent data loss from accumulating across monthly runs.

The 5.0-hour engagement covered investigation (1.5h — 0.5h S3 diagnostic script: file sizes 2.1 MB × 6 files, record counts 80,000 × 6 files; 0.25h Ray cluster job logs confirming GetTimeoutError on all 6 monthly runs since April; 0.25h source code review: timeout=30 in ray.get() and except GetTimeoutError handler; 0.25h Ray Dashboard task timing: extract_features p95 completion time 45–90s; 0.25h confirmation that 92 timed-out tasks complete on cluster but results discarded), fix implementation (2.0h — 0.25h timeout=30 removed from ray.get(); 0.25h ray.wait()-based progress collection loop implemented; 0.25h record count assertion added post-ray.get(); 0.5h post-write S3 file size assertion added; 0.25h code reviewed and deployed to production job runner; 0.25h test run confirming 100/100 results collected and 26 MB output), and validation and re-run (1.5h — 1.0h all 6 monthly batches April–September 2026 re-run with fixed code; all 6 output files confirmed 26 MB and 1,000,000 records; 0.5h ML team annual retraining pipeline re-run confirming 60,000,000 total records loaded and model training completing successfully).

In a HourTab retainer log, the incident session reads: “run_monthly_feature_extraction.py silent partial-result investigation: Ray 2.9.3 on AWS EC2 m5.2xlarge head + 6× m5.xlarge workers; extract_features @ray.remote num_cpus=0.5; 100 tasks; ray.get(refs, timeout=30) fires after 30s; 8 of 100 tasks complete in time; GetTimeoutError caught silently; 80,000 records instead of 1,000,000 per monthly file; 6 months April–September 2026; discovered October 10 during annual ML model retraining (4.8M vs 60M total records); fix: remove timeout=30, ray.wait() progress loop, record count assert, S3 size assert > 20 MB; re-run all 6 monthly batches; all confirmed 26 MB / 1,000,000 records; 5.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: Ray developer retainers

See also: Apache Airflow developer retainer · Kafka developer retainer · Apache Spark developer retainer