Improve concurrency control between plugin hooks (ArchiveBox#1721)

<!-- IMPORTANT: Do not submit PRs with only formatting / PEP8 / line
length changes. -->

# Summary

<!--e.g. This PR fixes ABC or adds the ability to do XYZ...-->

# Related issues

<!-- e.g. ArchiveBox#123 or Roadmap goal #
https://github.com/pirate/ArchiveBox/wiki/Roadmap -->

# Changes these areas

- [ ] Bugfixes
- [ ] Feature behavior
- [ ] Command line interface
- [ ] Configuration options
- [ ] Internal architecture
- [ ] Snapshot data layout on disk

Author: pirate

TODO_hook_concurrency.md

@@ -310,37 +310,40 @@ archivebox/plugins/{plugin_name}/
 ## Implementation Checklist
 
 ### Phase 1: Schema Migration ✅
-- [ ] Add `Snapshot.current_step` (IntegerField 0-9, default=0)
-- [ ] Add `ArchiveResult.hook_name` (CharField, nullable) - just filename
-- [ ] Create migration: `0033_snapshot_current_step_archiveresult_hook_name.py`
+- [x] Add `Snapshot.current_step` (IntegerField 0-9, default=0)
+- [x] Add `ArchiveResult.hook_name` (CharField, nullable) - just filename
+- [x] Create migration: `0034_snapshot_current_step.py`
 
-### Phase 2: Core Logic Updates
-- [ ] Add `extract_step(hook_name)` utility in `archivebox/hooks.py`
+### Phase 2: Core Logic Updates ✅
+- [x] Add `extract_step(hook_name)` utility in `archivebox/hooks.py`
   - Extract first digit from `__XX_` pattern
   - Default to 9 for unnumbered hooks
-- [ ] Update `Snapshot.create_pending_archiveresults()` in `archivebox/core/models.py`:
+- [x] Add `is_background_hook(hook_name)` utility in `archivebox/hooks.py`
+  - Check for `.bg.` in filename
+- [x] Update `Snapshot.create_pending_archiveresults()` in `archivebox/core/models.py`:
   - Discover all hooks (not plugins)
   - Create one AR per hook with `hook_name` set
-- [ ] Update `ArchiveResult.run()` in `archivebox/core/models.py`:
+- [x] Update `ArchiveResult.run()` in `archivebox/core/models.py`:
   - If `hook_name` set: run single hook
   - If `hook_name` None: discover all plugin hooks (existing behavior)
-- [ ] Add `Snapshot.advance_step_if_ready()` method:
+- [x] Add `Snapshot.advance_step_if_ready()` method:
   - Check if all foreground ARs in current step finished
   - Increment `current_step` if ready
   - Ignore background hooks (.bg) in completion check
-- [ ] Integrate with `SnapshotMachine.is_finished()` in `archivebox/core/statemachines.py`:
+- [x] Integrate with `SnapshotMachine.is_finished()` in `archivebox/core/statemachines.py`:
   - Call `advance_step_if_ready()` before checking if done
 
-### Phase 3: Worker Coordination
-- [ ] Update worker AR claiming query in `archivebox/workers/worker.py`:
+### Phase 3: Worker Coordination ✅
+- [x] Update worker AR claiming query in `archivebox/workers/worker.py`:
   - Filter: `extract_step(ar.hook_name) <= snapshot.current_step`
-  - Note: May need to denormalize or use clever query since step is derived
-  - Alternative: Claim any AR in QUEUED state, check step in Python before processing
+  - Claims ARs in QUEUED state, checks step in Python before processing
+  - Orders by hook_name for deterministic execution within step
 
-### Phase 4: Hook Renumbering
-- [ ] Renumber hooks per renumbering map below
-- [ ] Add `.bg` suffix to long-running hooks
-- [ ] Test all hooks still work after renumbering
+### Phase 4: Hook Renumbering ✅
+- [x] Renumber hooks per renumbering map below
+- [x] Add `.bg` suffix to long-running hooks (media, gallerydl, forumdl, papersdl)
+- [x] Move parse_* hooks to step 7 (70-79)
+- [x] Test all hooks still work after renumbering
 
 ## Migration Path
 
@@ -353,25 +356,34 @@ No special migration needed:
 
 ### Renumbering Map
 
-**Current → New:**
-```
-git/on_Snapshot__12_git.py                    → git/on_Snapshot__62_git.py
-media/on_Snapshot__51_media.py                → media/on_Snapshot__63_media.bg.py
-gallerydl/on_Snapshot__52_gallerydl.py        → gallerydl/on_Snapshot__64_gallerydl.bg.py
-forumdl/on_Snapshot__53_forumdl.py            → forumdl/on_Snapshot__65_forumdl.bg.py
-papersdl/on_Snapshot__54_papersdl.py          → papersdl/on_Snapshot__66_papersdl.bg.py
-
-readability/on_Snapshot__52_readability.py    → readability/on_Snapshot__55_readability.py
-mercury/on_Snapshot__53_mercury.py            → mercury/on_Snapshot__56_mercury.py
-
-singlefile/on_Snapshot__37_singlefile.py      → singlefile/on_Snapshot__50_singlefile.py
-screenshot/on_Snapshot__34_screenshot.js      → screenshot/on_Snapshot__51_screenshot.js
-pdf/on_Snapshot__35_pdf.js                    → pdf/on_Snapshot__52_pdf.js
-dom/on_Snapshot__36_dom.js                    → dom/on_Snapshot__53_dom.js
-title/on_Snapshot__32_title.js                → title/on_Snapshot__54_title.js
-headers/on_Snapshot__33_headers.js            → headers/on_Snapshot__55_headers.js
-
-wget/on_Snapshot__50_wget.py                  → wget/on_Snapshot__61_wget.py
+**Completed Renames:**
+```
+# Step 5: DOM Extraction (sequential, non-background)
+singlefile/on_Snapshot__37_singlefile.py      → singlefile/on_Snapshot__50_singlefile.py ✅
+screenshot/on_Snapshot__34_screenshot.js      → screenshot/on_Snapshot__51_screenshot.js ✅
+pdf/on_Snapshot__35_pdf.js                    → pdf/on_Snapshot__52_pdf.js ✅
+dom/on_Snapshot__36_dom.js                    → dom/on_Snapshot__53_dom.js ✅
+title/on_Snapshot__32_title.js                → title/on_Snapshot__54_title.js ✅
+readability/on_Snapshot__52_readability.py    → readability/on_Snapshot__55_readability.py ✅
+headers/on_Snapshot__33_headers.js            → headers/on_Snapshot__55_headers.js ✅
+mercury/on_Snapshot__53_mercury.py            → mercury/on_Snapshot__56_mercury.py ✅
+htmltotext/on_Snapshot__54_htmltotext.py      → htmltotext/on_Snapshot__57_htmltotext.py ✅
+
+# Step 6: Post-DOM Extraction (background for long-running)
+wget/on_Snapshot__50_wget.py                  → wget/on_Snapshot__61_wget.py ✅
+git/on_Snapshot__12_git.py                    → git/on_Snapshot__62_git.py ✅
+media/on_Snapshot__51_media.py                → media/on_Snapshot__63_media.bg.py ✅
+gallerydl/on_Snapshot__52_gallerydl.py        → gallerydl/on_Snapshot__64_gallerydl.bg.py ✅
+forumdl/on_Snapshot__53_forumdl.py            → forumdl/on_Snapshot__65_forumdl.bg.py ✅
+papersdl/on_Snapshot__54_papersdl.py          → papersdl/on_Snapshot__66_papersdl.bg.py ✅
+
+# Step 7: URL Extraction (parse_* hooks moved from step 6)
+parse_html_urls/on_Snapshot__60_parse_html_urls.py      → parse_html_urls/on_Snapshot__70_parse_html_urls.py ✅
+parse_txt_urls/on_Snapshot__62_parse_txt_urls.py        → parse_txt_urls/on_Snapshot__71_parse_txt_urls.py ✅
+parse_rss_urls/on_Snapshot__61_parse_rss_urls.py        → parse_rss_urls/on_Snapshot__72_parse_rss_urls.py ✅
+parse_netscape_urls/on_Snapshot__63_parse_netscape_urls.py → parse_netscape_urls/on_Snapshot__73_parse_netscape_urls.py ✅
+parse_jsonl_urls/on_Snapshot__64_parse_jsonl_urls.py    → parse_jsonl_urls/on_Snapshot__74_parse_jsonl_urls.py ✅
+parse_dom_outlinks/on_Snapshot__40_parse_dom_outlinks.js → parse_dom_outlinks/on_Snapshot__75_parse_dom_outlinks.js ✅
 ```
 
 ## Testing Strategy

archivebox/cli/archivebox_status.py

@@ -11,18 +11,6 @@
 from archivebox.config import DATA_DIR, CONSTANTS, ARCHIVE_DIR
 from archivebox.config.common import SHELL_CONFIG
 from archivebox.misc.legacy import parse_json_links_details
-from archivebox.misc.folders import (
-    get_indexed_folders,
-    get_archived_folders,
-    get_invalid_folders,
-    get_unarchived_folders,
-    get_present_folders,
-    get_valid_folders,
-    get_duplicate_folders,
-    get_orphaned_folders,
-    get_corrupted_folders,
-    get_unrecognized_folders,
-)
 from archivebox.misc.system import get_dir_size
 from archivebox.misc.logging_util import printable_filesize
 
@@ -55,42 +43,40 @@ def status(out_dir: Path=DATA_DIR) -> None:
     size = printable_filesize(num_bytes)
     print(f'    Size: {size} across {num_files} files in {num_dirs} directories')
 
-    num_indexed = len(get_indexed_folders(links, out_dir=out_dir))
-    num_archived = len(get_archived_folders(links, out_dir=out_dir))
-    num_unarchived = len(get_unarchived_folders(links, out_dir=out_dir))
-    print(f'    > indexed: {num_indexed}'.ljust(36), f'({get_indexed_folders.__doc__})')
-    print(f'      > archived: {num_archived}'.ljust(36), f'({get_archived_folders.__doc__})')
-    print(f'      > unarchived: {num_unarchived}'.ljust(36), f'({get_unarchived_folders.__doc__})')
-    
-    num_present = len(get_present_folders(links, out_dir=out_dir))
-    num_valid = len(get_valid_folders(links, out_dir=out_dir))
+    # Use DB as source of truth for snapshot status
+    num_indexed = links.count()
+    num_archived = links.filter(status='archived').count() or links.exclude(downloaded_at=None).count()
+    num_unarchived = links.filter(status='queued').count() or links.filter(downloaded_at=None).count()
+    print(f'    > indexed: {num_indexed}'.ljust(36), '(total snapshots in DB)')
+    print(f'      > archived: {num_archived}'.ljust(36), '(snapshots with archived content)')
+    print(f'      > unarchived: {num_unarchived}'.ljust(36), '(snapshots pending archiving)')
+
+    # Count directories on filesystem
+    num_present = 0
+    orphaned_dirs = []
+    if ARCHIVE_DIR.exists():
+        for entry in ARCHIVE_DIR.iterdir():
+            if entry.is_dir():
+                num_present += 1
+                if not links.filter(timestamp=entry.name).exists():
+                    orphaned_dirs.append(str(entry))
+
+    num_valid = min(num_present, num_indexed)  # approximate
     print()
-    print(f'    > present: {num_present}'.ljust(36), f'({get_present_folders.__doc__})')
-    print(f'      > [green]valid:[/green] {num_valid}'.ljust(36), f'               ({get_valid_folders.__doc__})')
-    
-    duplicate = get_duplicate_folders(links, out_dir=out_dir)
-    orphaned = get_orphaned_folders(links, out_dir=out_dir)
-    corrupted = get_corrupted_folders(links, out_dir=out_dir)
-    unrecognized = get_unrecognized_folders(links, out_dir=out_dir)
-    num_invalid = len({**duplicate, **orphaned, **corrupted, **unrecognized})
-    print(f'      > [red]invalid:[/red] {num_invalid}'.ljust(36), f'           ({get_invalid_folders.__doc__})')
-    print(f'        > duplicate: {len(duplicate)}'.ljust(36), f'({get_duplicate_folders.__doc__})')
-    print(f'        > orphaned: {len(orphaned)}'.ljust(36), f'({get_orphaned_folders.__doc__})')
-    print(f'        > corrupted: {len(corrupted)}'.ljust(36), f'({get_corrupted_folders.__doc__})')
-    print(f'        > unrecognized: {len(unrecognized)}'.ljust(36), f'({get_unrecognized_folders.__doc__})')
+    print(f'    > present: {num_present}'.ljust(36), '(directories in archive/)')
+    print(f'      > [green]valid:[/green] {num_valid}'.ljust(36), '               (directories with matching DB entry)')
+
+    num_orphaned = len(orphaned_dirs)
+    print(f'      > [red]orphaned:[/red] {num_orphaned}'.ljust(36), '         (directories without matching DB entry)')
 
     if num_indexed:
-        print('    [violet]Hint:[/violet] You can list link data directories by status like so:')
-        print('        [green]archivebox list --status=<status>  (e.g. indexed, corrupted, archived, etc.)[/green]')
+        print('    [violet]Hint:[/violet] You can list snapshots by status like so:')
+        print('        [green]archivebox list --status=<status>  (e.g. archived, queued, etc.)[/green]')
 
-    if orphaned:
+    if orphaned_dirs:
         print('    [violet]Hint:[/violet] To automatically import orphaned data directories into the main index, run:')
         print('        [green]archivebox init[/green]')
 
-    if num_invalid:
-        print('    [violet]Hint:[/violet] You may need to manually remove or fix some invalid data directories, afterwards make sure to run:')
-        print('        [green]archivebox init[/green]')
-    
     print()
     print('[green]\\[*] Scanning recent archive changes and user logins:[/green]')
     print(f'[yellow]   {CONSTANTS.LOGS_DIR}/*[/yellow]')

archivebox/core/migrations/0032_alter_archiveresult_binary_and_more.py

@@ -1,7 +1,7 @@
 # Generated by Django 6.0 on 2025-12-28 05:12
 
 import django.db.models.deletion
-import uuid
+from archivebox import uuid_compat
 from django.conf import settings
 from django.db import migrations, models
 
@@ -49,7 +49,7 @@ class Migration(migrations.Migration):
         migrations.AlterField(
             model_name='archiveresult',
             name='uuid',
-            field=models.UUIDField(blank=True, db_index=True, default=uuid.uuid7, null=True),
+            field=models.UUIDField(blank=True, db_index=True, default=uuid_compat.uuid7, null=True),
         ),
         migrations.AddConstraint(
             model_name='snapshot',

archivebox/core/migrations/0034_snapshot_current_step.py

@@ -0,0 +1,23 @@
+# Generated by Django 6.0 on 2025-12-28
+# Add Snapshot.current_step field for hook step-based execution
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('core', '0033_rename_extractor_add_hook_name'),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name='snapshot',
+            name='current_step',
+            field=models.PositiveSmallIntegerField(
+                default=0,
+                db_index=True,
+                help_text='Current hook step being executed (0-9). Used for sequential hook execution.'
+            ),
+        ),
+    ]