Skip to content

fix(clp-package): Write compression task failure errors to a log, and store the log path (instead of the errors) in status_msg (fixes #716). #1425

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
quinntaylormitchell merged 14 commits into from
Oct 28, 2025
Merged

fix(clp-package): Write compression task failure errors to a log, and store the log path (instead of the errors) in status_msg (fixes #716). #1425

Changes from 7 commits
Commits
Show all changes
14 commits
Select commit Hold shift + click to select a range
14d5b6c
Write failed compression log and store path in status_msg.
quinntaylormitchell Oct 16, 2025
bc3f8cc
Lint
quinntaylormitchell Oct 16, 2025
039e547
Rabbit
quinntaylormitchell Oct 16, 2025
9a1bce3
Merge branch 'main' into feature branch
quinntaylormitchell Oct 23, 2025
1936756
Combine user log write functions; address Kirk comments.
quinntaylormitchell Oct 27, 2025
5a99b0d
Only report the relative path to the user log instead of the absolute…
quinntaylormitchell Oct 28, 2025
45bc548
Address Kirk comments.
quinntaylormitchell Oct 28, 2025
1f579f2
Rabbit.
quinntaylormitchell Oct 28, 2025
d17868e
Reorder variables in _write_user_failure_log.
quinntaylormitchell Oct 28, 2025
3e08c7b
Lint
quinntaylormitchell Oct 28, 2025
97a06f8
Kirk
quinntaylormitchell Oct 28, 2025
10efcd7
Pluralize variable name.
quinntaylormitchell Oct 28, 2025
dabf01f
Lint.
quinntaylormitchell Oct 28, 2025
f1d4439
Merge branch 'main' into feature branch
quinntaylormitchell Oct 28, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
97 changes: 64 additions & 33 deletions components/job-orchestration/job_orchestration/scheduler/compress/compression_scheduler.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -7,7 +7,7 @@
import time
from contextlib import closing
from pathlib import Path
from typing import Any, Dict, List, Optional, Set
from typing import Any, Dict, List, Optional, Set, Union
Comment thread
quinntaylormitchell marked this conversation as resolved.
Resolved

import brotli
import msgpack
Expand Down Expand Up @@ -180,36 +180,45 @@ def _process_s3_input(
paths_to_compress_buffer.add_file(object_metadata)


def _write_failed_path_log(
invalid_path_messages: List[str], logs_directory: Path, job_id: Any
def _write_user_failure_log(
content: List[str],
logs_directory: Path,
job_id: Any,
filename_suffix: str,
title: str,
) -> Optional[Path]:
"""
Writes the error messages in `invalid_path_messages` to a log file,
`{logs_directory}/user/failed_paths_{job_id}.txt`. The directory will be created if it doesn't
already exist.
:param invalid_path_messages:
Writes a user-oriented failure log to
`{logs_directory}/user/job_{job_id}_{filename_suffix}.txt`. The `{logs_directory}/user`
directory will be created if it does not already exist.

:param content:
:param logs_directory:
:param job_id:
:return: Path to the written log file or `None` if error is encountered.
:param filename_suffix:
:param title:
:return: Path to the written log file relative to `logs_directory`, or `None` on error.
"""
Comment on lines +183 to 201

@coderabbitai coderabbitai Bot Oct 28, 2025

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🛠️ Refactor suggestion | 🟠 Major

Narrow job_id type; align with lint and intent.

job_id: Any trips ANN401 and weakens the API. Use an integer.

 def _write_user_failure_log(
-    content: List[str],
-    logs_directory: Path,
-    job_id: Any,
+    content: List[str],
+    logs_directory: Path,
+    job_id: int,
     filename_suffix: str,
     title: str,
 ) -> Optional[Path]:
📝 Committable suggestion

‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.

Suggested change
def _write_user_failure_log(
content: List[str],
logs_directory: Path,
job_id: Any,
filename_suffix: str,
title: str,
) -> Optional[Path]:
"""
Writes the error messages in `invalid_path_messages` to a log file,
`{logs_directory}/user/failed_paths_{job_id}.txt`. The directory will be created if it doesn't
already exist.
:param invalid_path_messages:
Writes a user-oriented failure log to
`{logs_directory}/user/job_{job_id}_{filename_suffix}.txt`. The `{logs_directory}/user`
directory will be created if it does not already exist.
:param content:
:param logs_directory:
:param job_id:
:return: Path to the written log file or `None` if error is encountered.
:param filename_suffix:
:param title:
:return: Path to the written log file relative to `logs_directory`, or `None` on error.
"""
def _write_user_failure_log(
content: List[str],
logs_directory: Path,
job_id: int,
filename_suffix: str,
title: str,
) -> Optional[Path]:
"""
Writes a user-oriented failure log to
`{logs_directory}/user/job_{job_id}_{filename_suffix}.txt`. The `{logs_directory}/user`
directory will be created if it does not already exist.
:param content:
:param logs_directory:
:param job_id:
:param filename_suffix:
:param title:
:return: Path to the written log file relative to `logs_directory`, or `None` on error.
"""
🧰 Tools
🪛 Ruff (0.14.2)

186-186: Dynamically typed expressions (typing.Any) are disallowed in job_id

(ANN401)

🤖 Prompt for AI Agents 
In
components/job-orchestration/job_orchestration/scheduler/compress/compression_scheduler.py
around lines 183 to 201, change the job_id parameter's type from Any to int to
satisfy ANN401 and reflect intent; update the function signature to job_id: int,
adjust any internal code that formats or concatenates job_id (use str(job_id) or
f-strings) if necessary, and update the docstring/type hints accordingly so the
API and typing are consistent.


user_logs_dir = Path(logs_directory) / "user"
relative_log_path = Path("user") / f"job_{job_id}_{filename_suffix}.txt"
user_logs_dir = logs_directory / relative_log_path.parent
try:
user_logs_dir.mkdir(parents=True, exist_ok=True)
except Exception:
except Exception as e:
logger.error("Failed to create user logs directory: '%s' - %s", user_logs_dir, e)
return None

log_path = user_logs_dir / f"failed_paths_{job_id}.txt"
log_path = logs_directory / relative_log_path
try:
with log_path.open("w", encoding="utf-8") as f:
timestamp = datetime.datetime.now().isoformat(timespec="seconds")
f.write(f"Failed input paths log.\nGenerated at {timestamp}.\n\n")
for msg in invalid_path_messages:
f.write(f"{msg.rstrip()}\n")
except Exception:
f.write(f"{title}\nGenerated at {timestamp}.\n\n")
for item in content:
f.write(f"{item.rstrip()}\n")
except Exception as e:
logger.error("Failed to write compression failure user log: '%s' - %s", log_path, e)
return None

return log_path
return relative_log_path
Comment on lines +183 to +221

@coderabbitai coderabbitai Bot Oct 28, 2025

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🛠️ Refactor suggestion | 🟠 Major

Use logger.exception for exception handlers.

The exception handlers currently use logger.error, but logger.exception is preferred as it automatically includes the stack trace, which is valuable for debugging log writing failures.

Apply this diff:

     try:
         user_logs_dir.mkdir(parents=True, exist_ok=True)
     except Exception as e:
-        logger.error("Failed to create user logs directory: '%s' - %s", user_logs_dir, e)
+        logger.exception("Failed to create user logs directory: '%s'", user_logs_dir)
         return None
 
     log_path = Path(logs_directory) / relative_log_path
     try:
         with log_path.open("w", encoding="utf-8") as f:
             timestamp = datetime.datetime.now().isoformat(timespec="seconds")
             f.write(f"{header_title}\nGenerated at {timestamp}.\n\n")
             if isinstance(content, str):
                 f.write(f"{content.rstrip()}\n")
             else:
                 for item in content:
                     f.write(f"{str(item).rstrip()}\n")
     except Exception as e:
-        logger.error("Failed to write compression failure user log: '%s' - %s", log_path, e)
+        logger.exception("Failed to write compression failure user log: '%s'", log_path)
         return None

Note: logger.exception automatically includes the exception details, so the e parameter can be removed from the format string.

🧰 Tools
🪛 Ruff (0.14.1)

184-184: Use X | Y for type annotations

Convert to X | Y

(UP007)

186-186: Dynamically typed expressions (typing.Any) are disallowed in job_id

(ANN401)

206-206: Do not catch blind exception: Exception

(BLE001)

207-207: Use logging.exception instead of logging.error

Replace with exception

(TRY400)

213-213: datetime.datetime.now() called without a tz argument

(DTZ005)

220-220: Do not catch blind exception: Exception

(BLE001)

221-221: Use logging.exception instead of logging.error

Replace with exception

(TRY400)

🤖 Prompt for AI Agents 
components/job-orchestration/job_orchestration/scheduler/compress/compression_scheduler.py
lines 183-224: the except handlers use logger.error and pass the exception
object into the format string; change both to logger.exception and remove the
exception argument from the format string so the stack trace is automatically
logged (i.e., call logger.exception("Failed to create user logs directory:
'%s'", user_logs_dir) and logger.exception("Failed to write compression failure
user log: '%s'", log_path)).



def search_and_schedule_new_tasks(
Expand Down Expand Up @@ -274,18 +283,22 @@ def search_and_schedule_new_tasks(
if input_type == InputType.FS.value:
invalid_path_messages = _process_fs_input_paths(input_config, paths_to_compress_buffer)
if len(invalid_path_messages) > 0:
base_msg = "At least one of your input paths could not be processed."

user_log_path = _write_failed_path_log(
invalid_path_messages, clp_config.logs_directory, job_id
user_log_relative_path = _write_user_failure_log(
invalid_path_messages,
clp_config.logs_directory,
job_id,
filename_suffix="failed_paths",
title="Failed input paths log.",
)
if user_log_relative_path is None:
err_msg = "Failed to write user log for invalid input paths."
raise RuntimeError(err_msg)

error_msg = (
"At least one of your input paths could not be processed."
f" See the error log at '{user_log_relative_path}' inside your configured logs"
" directory (`logs_directory`) for more details."
)
Comment on lines +286 to 301

@coderabbitai coderabbitai Bot Oct 28, 2025

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue | 🟠 Major

Don’t raise on log-write failure; fail job gracefully and continue.

Raising here aborts the scheduler loop (main returns -1), stalling all jobs. Update job status with a short fallback message instead, commit, and continue.

-                if user_log_relative_path is None:
-                    err_msg = "Failed to write user log for invalid input paths."
-                    raise RuntimeError(err_msg)
+                if user_log_relative_path is None:
+                    fallback_msg = (
+                        "At least one of your input paths could not be processed."
+                        " Additionally, writing a user error log failed. See the scheduler logs"
+                        " for details."
+                    )
+                    update_compression_job_metadata(
+                        db_cursor,
+                        job_id,
+                        {
+                            "status": CompressionJobStatus.FAILED,
+                            "status_msg": fallback_msg,
+                        },
+                    )
+                    db_conn.commit()
+                    continue

Based on learnings.

🤖 Prompt for AI Agents 
In
components/job-orchestration/job_orchestration/scheduler/compress/compression_scheduler.py
around lines 286 to 301, the current code raises a RuntimeError when
_write_user_failure_log returns None which aborts the scheduler loop; instead,
replace the raise with logic that sets the job status to a graceful failure
using a short fallback message (e.g., "Could not write user failure log; see
scheduler logs for details"), persist/commit that status update to the
datastore, and then continue execution without throwing so the scheduler loop is
not interrupted.

if user_log_path is None:
error_msg = base_msg + (
f" Check the compression scheduler logs in {clp_config.logs_directory} for"
" more details."
)
else:
error_msg = base_msg + f" Check {user_log_path} for more details."

update_compression_job_metadata(
db_cursor,
Expand Down Expand Up @@ -384,7 +397,7 @@ def search_and_schedule_new_tasks(
scheduled_jobs[job_id] = job


def poll_running_jobs(db_conn, db_cursor):
def poll_running_jobs(logs_directory: Path, db_conn, db_cursor):
"""
Poll for running jobs and update their status.
"""
Expand All @@ -395,7 +408,7 @@ def poll_running_jobs(db_conn, db_cursor):
for job_id, job in scheduled_jobs.items():
job_success = True
duration = 0.0
error_message = ""
error_message: List[str] = []
Comment thread
coderabbitai[bot] marked this conversation as resolved.
Resolved

try:
returned_results = job.result_handle.get_result()
Expand All @@ -412,7 +425,7 @@ def poll_running_jobs(db_conn, db_cursor):
)
else:
job_success = False
error_message += f"task {task_result.task_id}: {task_result.error_message}\n"
error_message.append(f"task {task_result.task_id}: {task_result.error_message}")
logger.error(
f"Compression task job-{job_id}-task-{task_result.task_id} failed with"
f" error: {task_result.error_message}."
Expand All @@ -434,12 +447,30 @@ def poll_running_jobs(db_conn, db_cursor):
)
else:
logger.error(f"Job {job_id} failed. See worker logs or status_msg for details.")

error_log_relative_path = _write_user_failure_log(
error_message,
logs_directory,
job_id,
filename_suffix="task_errors",
title="Compression task errors.",
)
if error_log_relative_path is None:
err_msg = "Failed to write user log for failed compression job."
raise RuntimeError(err_msg)

error_msg = (
"One or more compression tasks failed."
f" See the error log at '{error_log_relative_path}' inside your configured logs"
" directory (`logs_directory`) for more details."
)

Comment on lines +453 to +469

@coderabbitai coderabbitai Bot Oct 28, 2025

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue | 🟠 Major

Avoid crashing the scheduler on task-error log-write failure.

Same concern as the invalid-path branch: don’t raise. Mark job FAILED with a compact message, commit, add to jobs_to_delete, and continue.

-            error_log_relative_path = _write_user_failure_log(
-                error_message,
+            error_log_relative_path = _write_user_failure_log(
+                error_messages,
                 logs_directory,
                 job_id,
                 filename_suffix="task_errors",
                 title="Compression task errors.",
             )
-            if error_log_relative_path is None:
-                err_msg = "Failed to write user log for failed compression job."
-                raise RuntimeError(err_msg)
+            if error_log_relative_path is None:
+                fallback_msg = (
+                    "One or more compression tasks failed."
+                    " Additionally, writing the user error log failed. See worker logs for details."
+                )
+                update_compression_job_metadata(
+                    db_cursor,
+                    job_id,
+                    dict(
+                        status=CompressionJobStatus.FAILED,
+                        status_msg=fallback_msg,
+                    ),
+                )
+                db_conn.commit()
+                jobs_to_delete.append(job_id)
+                continue
📝 Committable suggestion

‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.

Suggested change
error_log_relative_path = _write_user_failure_log(
error_message,
logs_directory,
job_id,
filename_suffix="task_errors",
title="Compression task errors.",
)
if error_log_relative_path is None:
err_msg = "Failed to write user log for failed compression job."
raise RuntimeError(err_msg)
error_msg = (
"One or more compression tasks failed."
f" See the error log at '{error_log_relative_path}' inside your configured logs"
" directory (`logs_directory`) for more details."
)
error_log_relative_path = _write_user_failure_log(
error_messages,
logs_directory,
job_id,
filename_suffix="task_errors",
title="Compression task errors.",
)
if error_log_relative_path is None:
fallback_msg = (
"One or more compression tasks failed."
" Additionally, writing the user error log failed. See worker logs for details."
)
update_compression_job_metadata(
db_cursor,
job_id,
dict(
status=CompressionJobStatus.FAILED,
status_msg=fallback_msg,
),
)
db_conn.commit()
jobs_to_delete.append(job_id)
continue
error_msg = (
"One or more compression tasks failed."
f" See the error log at '{error_log_relative_path}' inside your configured logs"
" directory (`logs_directory`) for more details."
)
🤖 Prompt for AI Agents 
In
components/job-orchestration/job_orchestration/scheduler/compress/compression_scheduler.py
around lines 451 to 467, the code raises a RuntimeError when writing the
user-facing task-error log fails; instead follow the same non-crashing behavior
used for the invalid-path branch: do not raise, set the job state to FAILED with
a compact message noting the inability to write the user error log, commit the
job state change, append the job_id to jobs_to_delete, and continue execution;
ensure the compact message is concise, persist the update via the same commit
method used elsewhere, and keep control flow consistent with the surrounding
error-handling pattern.

update_compression_job_metadata(
db_cursor,
job_id,
dict(
status=CompressionJobStatus.FAILED,
status_msg=error_message,
status_msg=error_msg,
),
)
db_conn.commit()
Expand Down Expand Up @@ -515,7 +546,7 @@ def main(argv):
clp_metadata_db_connection_config,
task_manager,
)
poll_running_jobs(db_conn, db_cursor)
poll_running_jobs(clp_config.logs_directory, db_conn, db_cursor)
time.sleep(clp_config.compression_scheduler.jobs_poll_delay)
except KeyboardInterrupt:
logger.info("Forcefully shutting down")
Expand Down
Loading