<!-- greptile_comment -->

<h3>Greptile Summary</h3>

This PR refactors the four day-parallel processor modules (`ip_db`, `protocol_db`, `spectrum_db`, `structure_db`) from a WAL-concurrent-writer model to a single-writer model where worker subprocesses only compute and serialize data, then return it to the parent thread for all DB writes under explicit `BEGIN`/`COMMIT` transactions. `common.py` is updated to use `isolation_level=None` (autocommit) so callers own all transaction boundaries, and `discovery.batch_mark_processed` gains a `commit=False` opt-out so it can participate in the parent's transaction. `restore_spectrum_from_backup.py` (a one-off recovery utility) is deleted.

Key observations:

- The architecture is sound: workers now do CPU-bound work only, and a single sequential DB writer avoids WAL write conflicts.
- `ip_db.py`'s worker log message still says `"5m, {inserted_agg} agg, {errors} errors"` (without `"prepared"`), inconsistent with the other three modules which were correctly updated.
- `ip_db.py` passes the raw `compute_aggregates()` output directly as `rows_agg` (already flat dicts with scalar counts), while `protocol_db`, `spectrum_db`, and `structure_db` build a separate serialized `rows_agg` list. Both approaches are correct; `ip_db` is simply more concise because it has no set-to-string conversion needed for aggregates.
- Per-row insert failures inside `insert_results` are silently swallowed and not reflected in `stats['errors']`; this is pre-existing behavior unchanged by this PR.

<h3>Confidence Score: 4/5</h3>

- This PR is safe to merge; the single-writer refactor is logically sound and the one identified issue is a minor cosmetic inconsistency in a log message.
- The core architectural change is well-implemented and consistent across all four processor modules. The only issue found is that `ip_db.py`'s worker log message omits the "prepared" qualifier present in the other three modules, which was the explicit intent of a previous review comment. No logic errors, data loss paths, or regressions were found.
- netflow-db/ip_db.py — worker log message inconsistency at line 282.

<h3>Important Files Changed</h3>




| Filename | Overview |
|----------|----------|
| netflow-db/common.py | Switches get_db_connection to isolation_level=None (autocommit) to support explicit BEGIN/COMMIT transaction control by callers; existing conn.commit() calls in discovery.py and init functions become no-ops but are harmless. |
| netflow-db/discovery.py | Adds optional commit parameter to batch_mark_processed so callers can defer commit when wrapping calls in their own explicit transaction; existing default commit=True is backward-compatible. |
| netflow-db/ip_db.py | Moves DB writes to parent thread using imap_unordered with explicit BEGIN/COMMIT; worker log message inconsistently omits "prepared" unlike other modules; rows_agg is passed as raw aggregates dict (correct, but implicit). |
| netflow-db/protocol_db.py | Fully pre-serializes 5m and aggregate rows in the worker before returning; worker log now correctly says "prepared"; parent does explicit BEGIN/COMMIT for each day's batch. |
| netflow-db/restore_spectrum_from_backup.py | File deleted; was a one-off utility script for restoring spectrum_stats from a named backup. Deletion is intentional and straightforward. |
| netflow-db/spectrum_db.py | Worker pre-serializes spectrum JSON in the subprocess; parent thread handles all DB writes with explicit transactions. Pattern is consistent with protocol_db.py. |
| netflow-db/structure_db.py | Follows the same pattern as spectrum_db.py; worker pre-serializes structure function JSON and returns rows for parent-thread insertion. Consistent with the new single-writer model. |

</details>



<h3>Sequence Diagram</h3>

```mermaid
sequenceDiagram
    participant M as main()
    participant P as process_pending_files(conn)
    participant W as Worker Process (N workers)
    participant DB as SQLite DB (single conn)

    M->>P: call with autocommit conn
    P->>P: init_*_stats_table(conn)
    P->>P: handle_stale_days(conn, table)
    P->>P: get_files_needing_processing()
    P->>P: group_files_by_day()

    loop For each completed day result (imap_unordered)
        P->>W: dispatch day_task (router, day_start, files)
        W->>W: process_file() × N (nfdump, parse)
        W->>W: compute_aggregates()
        W->>W: build rows_5m[], rows_agg[], mark_results[]
        W-->>P: return {rows_5m, rows_agg, mark_results, errors, ...}

        P->>DB: conn.execute("BEGIN")
        P->>DB: insert_results(conn, rows_5m, rows_agg)
        P->>DB: batch_mark_processed(conn, table, mark_results, commit=False)
        P->>DB: conn.commit()
        Note over P,DB: On exception: conn.rollback()
    end

    P-->>M: return stats
```

<sub>Last reviewed commit: 129bfc8</sub>

<!-- greptile_other_comments_section -->

<!-- /greptile_comment -->