components/job-orchestration/job_orchestration/executor/compress/compression_task.py (1)

425-551: ⚠️ Potential issue | 🟠 Major

Make temporary cleanup exception-safe.

Line 551 now covers the normal non-zero return-code path, but exceptions before it still bypass cleanup. For example, subprocess.Popen, json.loads, DB updates, or archive post-processing can raise after logs_list_path / converted_inputs_dir have been created, leaving the same /var/tmp artifacts this PR is trying to prevent. Wrap the compression/conversion work in try/finally, and make cleanup idempotent so early returns and exceptions share the same path.

Suggested cleanup pattern

 def cleanup_temporary_files():
-    if logs_list_path is not None:
-        logs_list_path.unlink()
+    logs_list_path.unlink(missing_ok=True)
     if converted_inputs_dir is not None:
-        shutil.rmtree(converted_inputs_dir)
+        shutil.rmtree(converted_inputs_dir, ignore_errors=True)

 # Open stderr log file
 stderr_log_path = logs_dir / f"{instance_id_str}-stderr.log"
 stderr_log_file = open(stderr_log_path, "w")
+try:
+    # conversion/compression subprocess handling and result construction
+    ...
+finally:
+    cleanup_temporary_files()
+    stderr_log_file.close()

This also lets the conversion failure return path drop its separate cleanup call without risking double-unlink failures.

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In
`@components/job-orchestration/job_orchestration/executor/compress/compression_task.py`
around lines 425 - 551, The code currently performs conversion, streaming JSON
parsing, DB updates, S3 uploads and indexing (via
conversion_cmd/conversion_proc, proc, json.loads, _upload_archive_to_s3,
update_archive_metadata, update_job_metadata and the indexer subprocess) without
guaranteeing cleanup_temporary_files() runs on exceptions; refactor by wrapping
the entire conversion+compression loop and subsequent wait/return-code handling
in a try/finally (so any exception from subprocess.Popen, json.loads, DB calls,
S3 upload, or indexing will still run cleanup_temporary_files()), remove the
duplicate cleanup call in the conversion-failure early return, and ensure
cleanup_temporary_files() is idempotent (safe to call multiple times) so early
returns and the finally block won’t error if files were already removed.

components/job-orchestration/job_orchestration/executor/compress/compression_task.py (1)

456-680: ⚠️ Potential issue | 🔴 Critical

Critical: the compression block is duplicated — compression will run twice and the second run will fail because cleanup ran first.

Lines 456–559 (first compression block) and lines 562–677 (second compression block, inside the new try:/finally:) are two copies of the same logic. As written, run_clp will:

1. Execute Popen(compression_cmd, …) at line 459, drain proc.stdout, wait for the child, and set compression_successful at 555.
2. Call cleanup_temporary_files() at line 557, which unlinks logs_list_path and removes converted_inputs_dir (the inputs the compressor needs).
3. Enter the try: at line 562, re-run the dead conversion-failure handler at 563–572 (already handled at 443–454 with a return), then execute Popen(compression_cmd, …) again at line 577 — now with its input list / converted inputs already deleted — so that run fails (or produces nothing), compression_successful is reset to False at 576 and overwritten, and the uncompressed/compressed size counters from the first run are clobbered by the reset at 582–585.
4. Finally, cleanup_temporary_files() runs a second time and stderr_log_file is closed (680).

Beyond running compression twice, this also breaks the PR's stated goal: on a real failure path, the state reported to the caller reflects the second (broken) run, not the first, and database/job metadata may have already been committed for archives that subsequently appear to fail.

This looks like an editing/merge artifact from applying the earlier review suggestion on top of the original body without removing the original. The intended structure is to keep only one compression block, wrapped by the new try:/finally: that guarantees cleanup and stderr closure.

🛠️ Proposed fix — delete the first, unguarded copy so only the try/finally-wrapped version remains

-    # Start compression
-    logger.debug("Compressing...")
-    compression_successful = False
-    proc = subprocess.Popen(
-        compression_cmd, stdout=subprocess.PIPE, stderr=stderr_log_file, env=compression_env
-    )
-
-    # Compute the total amount of data compressed
-    last_archive_stats = None
-    last_line_decoded = False
-    total_uncompressed_size = 0
-    total_compressed_size = 0
-
-    # Handle job metadata update and S3 write if enabled
-    s3_error = None
-    while not last_line_decoded:
-        stats: dict[str, Any] | None = None
-
-        line = proc.stdout.readline()
-        if not line:
-            last_line_decoded = True
-        else:
-            stats = json.loads(line.decode("utf-8"))
-
-        if last_archive_stats is not None and (
-            None is stats or stats["id"] != last_archive_stats["id"]
-        ):
-            archive_id = last_archive_stats["id"]
-            archive_path = archive_output_dir / archive_id
-            if enable_s3_write:
-                if s3_error is None:
-                    logger.info(f"Uploading archive {archive_id} to S3...")
-                    try:
-                        _upload_archive_to_s3(s3_config, archive_path, archive_id, dataset)
-                        logger.info(f"Finished uploading archive {archive_id} to S3.")
-                    except Exception as err:
-                        logger.exception(f"Failed to upload archive {archive_id}")
-                        s3_error = str(err)
-                        # NOTE: It's possible `proc` finishes before we call `terminate` on it, in
-                        # which case the process will still return success.
-                        proc.terminate()
-
-            if s3_error is None:
-                # We've started a new archive so add the previous archive's last reported size to
-                # the total
-                total_uncompressed_size += last_archive_stats["uncompressed_size"]
-                total_compressed_size += last_archive_stats["size"]
-                with (
-                    closing(sql_adapter.create_connection(True)) as db_conn,
-                    closing(db_conn.cursor(dictionary=True)) as db_cursor,
-                ):
-                    table_prefix = clp_metadata_db_connection_config["table_prefix"]
-                    if StorageEngine.CLP_S == clp_storage_engine:
-                        update_archive_metadata(
-                            db_cursor, table_prefix, dataset, last_archive_stats
-                        )
-                    update_job_metadata(
-                        db_cursor,
-                        job_id,
-                        last_archive_stats,
-                    )
-                    db_conn.commit()
-
-                if StorageEngine.CLP_S == clp_storage_engine:
-                    indexer_cmd = [
-                        str(clp_home / "bin" / "indexer"),
-                        *_get_db_connection_args_for_clp_cmd(clp_metadata_db_connection_config),
-                        dataset,
-                        archive_path,
-                    ]
-
-                    # Set environment variables for database credentials
-                    indexer_env = dict(os.environ)
-                    indexer_env.update(
-                        _get_db_connection_env_vars_for_clp_cmd(clp_metadata_db_connection_config)
-                    )
-
-                    try:
-                        subprocess.run(
-                            indexer_cmd,
-                            stdout=subprocess.DEVNULL,
-                            stderr=stderr_log_file,
-                            check=True,
-                            env=indexer_env,
-                        )
-                    except subprocess.CalledProcessError:
-                        logger.exception("Failed to index archive.")
-
-            if enable_s3_write:
-                archive_path.unlink()
-
-        last_archive_stats = stats
-
-    # Wait for compression to finish
-    return_code = proc.wait()
-
-    if 0 != return_code:
-        logger.error(f"Failed to compress, return_code={return_code!s}")
-    else:
-        compression_successful = True
-
-    cleanup_temporary_files()
-
-    logger.debug("Compressed.")
-
-    # Close stderr log file
     try:
-        if conversion_return_code != 0:
-            logger.error(
-                f"Failed to convert unstructured log text, return_code={conversion_return_code}"
-            )
-            worker_output = {
-                "total_uncompressed_size": 0,
-                "total_compressed_size": 0,
-                "error_message": f"Check logs in {stderr_log_path}",
-            }
-            return CompressionTaskStatus.FAILED, worker_output
-
         # Start compression
         logger.debug("Compressing...")
         compression_successful = False

Also, since the conversion_return_code != 0 path at lines 443–454 already returns (and closes stderr_log_file explicitly before doing so), the duplicated conversion-failure branch inside the try: (lines 563–572) is unreachable and should be removed when collapsing the duplicates.

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In
`@components/job-orchestration/job_orchestration/executor/compress/compression_task.py`
around lines 456 - 680, The compression logic is duplicated: remove the first
unguarded compression block (the Popen loop starting at the initial "Start
compression" comment that uses compression_cmd, proc, last_archive_stats,
total_uncompressed_size/total_compressed_size, s3_error,
update_archive_metadata/update_job_metadata, indexer_cmd, proc.wait(), and
cleanup_temporary_files()) so only the try/finally-wrapped version remains; also
delete the redundant conversion_return_code != 0 branch inside the remaining try
(the duplicated early-return handling) because conversion_return_code is already
handled earlier and that branch is unreachable—ensure stderr_log_file is closed
and cleanup_temporary_files() is only called from the finally block.

components/job-orchestration/job_orchestration/executor/compress/compression_task.py (1)

431-561: ⚠️ Potential issue | 🟠 Major

Move conversion subprocess into the guarded scope and explicitly terminate compression subprocess in finally.

The conversion subprocess runs before the try block (lines 435–441), so a Popen failure skips all cleanup. Additionally, the finally block (lines 559–561) doesn't terminate the compression subprocess, leaving it running if an exception occurs during processing or if the parent removes temporary inputs while proc is still active.

Move the conversion subprocess into the try block, initialize proc = None before try, and add explicit termination with timeout handling in the finally block before cleanup.

Suggested lifecycle fix

     # Open stderr log file
     stderr_log_path = logs_dir / f"{instance_id_str}-stderr.log"
     stderr_log_file = open(stderr_log_path, "w")
 
-    conversion_return_code = 0
-    if conversion_cmd is not None:
-        logger.debug("Execute log-converter with command: %s", conversion_cmd)
-        conversion_proc = subprocess.Popen(
-            conversion_cmd, stdout=subprocess.DEVNULL, stderr=stderr_log_file, env=conversion_env
-        )
-        conversion_return_code = conversion_proc.wait()
-
+    proc = None
     try:
+        conversion_return_code = 0
+        if conversion_cmd is not None:
+            logger.debug("Execute log-converter with command: %s", conversion_cmd)
+            conversion_proc = subprocess.Popen(
+                conversion_cmd,
+                stdout=subprocess.DEVNULL,
+                stderr=stderr_log_file,
+                env=conversion_env,
+            )
+            conversion_return_code = conversion_proc.wait()
+
         if conversion_return_code != 0:
             logger.error(
                 f"Failed to convert unstructured log text, return_code={conversion_return_code}"
             )
             worker_output = {
@@
         logger.debug("Compressing...")
         compression_successful = False
         proc = subprocess.Popen(
             compression_cmd, stdout=subprocess.PIPE, stderr=stderr_log_file, env=compression_env
         )
@@
         logger.debug("Compressed.")
     finally:
+        if proc is not None and proc.poll() is None:
+            logger.warning("Terminating compression subprocess after task failure.")
+            proc.terminate()
+            try:
+                proc.wait(timeout=10)
+            except subprocess.TimeoutExpired:
+                logger.warning("Killing compression subprocess after termination timeout.")
+                proc.kill()
+                proc.wait()
         cleanup_temporary_files()
         stderr_log_file.close()

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In
`@components/job-orchestration/job_orchestration/executor/compress/compression_task.py`
around lines 431 - 561, Move the conversion subprocess creation into the try
block and initialize proc = None (and conversion_proc = None if needed) before
the try so any Popen failure is cleaned up; start conversion with conversion_cmd
inside the try and retain conversion_return_code logic; in the finally block
check if proc is not None and is still running then call proc.terminate(), wait
with a short timeout, and if it hasn’t exited call proc.kill() (also do the same
for conversion_proc if present), then proceed to call cleanup_temporary_files()
and close stderr_log_file to ensure no orphaned processes or leaked file
handles; reference symbols: conversion_cmd, conversion_proc,
conversion_return_code, proc, stderr_log_file, cleanup_temporary_files().