cli:ppg2-filter-constraint-violations

pyproject.toml

@@ -67,3 +67,7 @@ THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLI
 [tool.pytest.ini_options]
 pythonpath = "python"
 addopts="--ignore=dev_utils"
+
+
+[project.scripts]
+ppg2-filter-constraint-violations = "pypipegraph2.cli:filter_constraint_violations"

python/pypipegraph2/cli.py

@@ -0,0 +1,48 @@
+"""CLI interface to ppg2"""
+
+import pyzstd
+import re
+import sys
+import pypipegraph2 as ppg2
+
+
+def filter_constraint_violations(filename=None, pipegraph=None):
+    """For when you really need to remove some jobs from the pipegraph's history.
+    Takes a filename with job ids.
+    Defaults to .ppg/errors/latest/constraint_violations.jobs
+
+    If you have your ppg history in a non standard place, pass in a ppg
+    with the correct directories set.
+    """
+    if pipegraph is None:
+        pipegraph = ppg2.new()
+
+    if sys.argv[1:]:
+        filename = sys.argv[1]
+    if filename is None:
+        filename = pipegraph.error_dir / "latest/constraint_violations.jobs"
+
+    history = pipegraph._load_history()
+
+    to_del = set()
+    for key, v in history.items():
+        found = False
+        for x in q:
+            if key.startswith(x) or key.endswith(x):
+                print("Would delete")
+                print("\t", key)
+                print("\t", x)
+                print("")
+                to_del.add(key)
+                found = True
+            if found:
+                break
+    print("Delete {len(to_del)} history entries?")
+    print("type yes<enter> to proceed")
+    print("This tool does not create backups. Make a snapshot of your project first")
+    for x in to_del:
+        del history[x]
+    if input() == "yes":
+        ppg2.global_pipegraph._save_history(history)
+    else:
+        print("aborted")

python/pypipegraph2/runner.py

@@ -752,7 +752,7 @@ def _executing_thread(self):
                                         # enable_logging_to_file("rust_debug.log")
 
                                         self.evaluator.debug_is_finished()
-                                        Path("debug.txt").write_text(
+                                        Path("ppg_evaluator_debug.txt").write_text(
                                             self.evaluator.debug()
                                         )
                                         # self.evaluator.reconsider_all_jobs() # if this helps' we're looking at a propagation failure. Somewhere.
@@ -781,9 +781,7 @@ def _executing_thread(self):
                                 job.waiting = True
                                 job.actual_cores_needed = -1
                                 self._interactive_report()
-                                job.start_time = (
-                                    time.time()
-                                )  # assign it just in case anything fails before acquiring the lock
+                                job.start_time = time.time()  # assign it just in case anything fails before acquiring the lock
                                 job.stop_time = float("nan")
                                 job.run_time = float("nan")
 
@@ -876,9 +874,7 @@ def _executing_thread(self):
                                     outputs = None
                                     raise error
 
-                    except (
-                        SystemExit
-                    ) as e:  # pragma: no cover - happens in spawned process, and we don't get coverage logging for it thanks to os._exit
+                    except SystemExit as e:  # pragma: no cover - happens in spawned process, and we don't get coverage logging for it thanks to os._exit
                         log_trace(
                             "SystemExit in spawned process -> converting to hard exit"
                         )
@@ -953,6 +949,18 @@ def _executing_thread(self):
                                     log_error(
                                         f"Recording job success failed for {job_id}. Likely constraint violation?: Message was '{e}'"
                                     )
+                                    with self.evaluator_lock:  # just so we don't mess up the file.
+                                        log_filename = (
+                                            self.job_graph.dir_config.error_dir
+                                            / self.job_graph.time_str
+                                            / "constraint_violation.jobs"
+                                        )
+                                        log_error(
+                                            f"Job id has been logged to {log_filename}\n. You might want to use ppg-filter-constraint-violations after fixing the problem."
+                                        )
+                                        with open(log_filename, "a") as op:
+                                            op.write(f"{job_id}\n")
+
                                     self.job_outcomes[job_id] = RecordedJobOutcome(
                                         job_id, JobOutcome.Failed, str(e)
                                     )