Recover ArchiveResults by plugin

archivebox/core/recovery_util.py

@@ -168,7 +168,10 @@ def recover_orchestrator_state(*, include_chrome: bool = False, crawl_id: str |
             orphaned_hook_processes = orphaned_hook_processes.filter(snapshot_pwd_filter)
         else:
             orphaned_hook_processes = orphaned_hook_processes.none()
-    for process in orphaned_hook_processes.only("id", "pwd", "cmd", "process_type", "status"):
+    for process in orphaned_hook_processes.only("id", "pwd", "cmd", "process_type", "status", "started_at", "ended_at").order_by(
+        "-started_at",
+        "-id",
+    ):
         hook_script_name = process.hook_script_name
         if not hook_script_name or not process.pwd:
             continue
@@ -187,18 +190,22 @@ def recover_orchestrator_state(*, include_chrome: bool = False, crawl_id: str |
         result, created = ArchiveResult.objects.get_or_create(
             snapshot=snapshot,
             plugin=plugin_dir.name,
-            hook_name=Path(hook_script_name).stem,
             defaults={
+                "hook_name": Path(hook_script_name).stem,
                 "status": ArchiveResult.StatusChoices.QUEUED,
             },
         )
-        if result.status == ArchiveResult.StatusChoices.QUEUED:
+        process_is_newer = bool(process.started_at and (result.start_ts is None or process.started_at >= result.start_ts))
+        if result.status == ArchiveResult.StatusChoices.QUEUED or process_is_newer:
             requeue_snapshot = False
             # A runner can die after the hook Process exits but before the
             # ProcessCompletedEvent projector links/finalizes ArchiveResult.
-            # Reconstruct only that exact hook row from the durable Process row.
+            # Reconstruct the plugin row from its newest durable Process row.
             output_files, output_size, output_mimetypes = _collect_output_metadata(plugin_dir)
+            result.hook_name = Path(hook_script_name).stem
             result.process = process
+            result.start_ts = process.started_at
+            result.end_ts = process.ended_at
             if _is_signal_interrupted_exit(process.exit_code):
                 # The owning runner died or was asked to stop while the hook was
                 # still active. Keep the work item queued so takeover retries the
@@ -222,7 +229,10 @@ def recover_orchestrator_state(*, include_chrome: bool = False, crawl_id: str |
                 )
             result.save(
                 update_fields=[
+                    "hook_name",
                     "process",
+                    "start_ts",
+                    "end_ts",
                     "output_files",
                     "output_size",
                     "output_mimetypes",

archivebox/tests/test_cli_run.py

@@ -2302,6 +2302,54 @@ class TestRunDueCrawlState:
 
 @pytest.mark.django_db
 class TestRecoverOrchestratorStateRedFailureModes:
+    def test_recovery_uses_newest_orphaned_hook_process_for_one_plugin_result(self):
+        from datetime import timedelta
+
+        from django.utils import timezone
+
+        from archivebox.base_models.models import get_or_create_system_user_pk
+        from archivebox.core.models import ArchiveResult, Snapshot
+        from archivebox.core.recovery_util import recover_orchestrator_state
+        from archivebox.crawls.models import Crawl
+        from archivebox.machine.models import Machine, NetworkInterface, Process
+
+        crawl = Crawl.objects.create(
+            urls="https://example.com",
+            created_by_id=get_or_create_system_user_pk(),
+            status=Crawl.StatusChoices.SEALED,
+            retry_at=None,
+        )
+        snapshot = Snapshot.objects.create(url="https://example.com", crawl=crawl, status=Snapshot.StatusChoices.SEALED, retry_at=None)
+        machine = Machine.current(refresh=True)
+        iface = NetworkInterface.current(refresh=True)
+        older_start = timezone.now() - timedelta(minutes=2)
+        newer_start = timezone.now() - timedelta(minutes=1)
+        for hook_name, started_at in (
+            ("on_Snapshot__01_title.py", older_start),
+            ("on_Snapshot__02_title.py", newer_start),
+        ):
+            Process.objects.create(
+                machine=machine,
+                iface=iface,
+                process_type=Process.TypeChoices.HOOK,
+                worker_type="archiveresult",
+                pwd=str(snapshot.output_dir / "title"),
+                cmd=[hook_name],
+                status=Process.StatusChoices.EXITED,
+                retry_at=None,
+                exit_code=0,
+                started_at=started_at,
+                ended_at=started_at + timedelta(seconds=1),
+            )
+
+        recover_orchestrator_state()
+
+        assert ArchiveResult.objects.filter(snapshot=snapshot, plugin="title").count() == 1
+        result = ArchiveResult.objects.get(snapshot=snapshot, plugin="title")
+        assert result.hook_name == "on_Snapshot__02_title"
+        assert result.start_ts == newer_start
+        assert result.process.started_at == newer_start
+
     def test_recovery_does_not_seal_queued_snapshot_waiting_for_future_retry_even_with_final_results(self):
         from datetime import timedelta