Add independent pipeline auto-tuning tool

docs/pipeline-auto-tuning.md

@@ -0,0 +1,20 @@
+# Pipeline Auto-Tuning
+
+`tools/tune_pipeline.py` independently sweeps worker threads, writer threads, and commit counts. It measures external wall time with `time.perf_counter()`, runs each candidate 3 times by default, and uses median seeds/second. The recommendation is the smallest candidate at or above `95.00%` of the peak median; the output also reports the efficient ceiling.
+
+```bash
+# Worker-only: executes workloads but changes 0 persistent files or database rows.
+python3 tools/tune_pipeline.py --test worker --test-only
+
+# Preview a Docker writer sweep: 0 workloads, 0 database writes, 0 files changed.
+python3 tools/tune_pipeline.py --test writer --dry-run \
+  --command "docker compose -f compose/compose.yaml run --rm {docker_env} seedfinder"
+
+# Run writer and commit sweeps against a dedicated database range.
+python3 tools/tune_pipeline.py --test writer --test commit \
+  --allow-db-writes --start-seed 8000000 --apply-to tuning.env
+```
+
+Worker sweeps force `BENCHMARK=1`. Writer and commit sweeps force `BENCHMARK=0`. Persistent writer/commit sweeps require `--allow-db-writes` and an explicit `--start-seed`, and leave inserted rows in PostgreSQL. Use a dedicated database or unused seed range.
+
+`--test-only` sets `TEST_ONLY=1`: it requires an existing schema, skips schema creation and checkpoint loading, sets `CHECKPOINT_FREQUENCY=none` when checkpoint controls are available, and rolls every writer transaction back after `COPY`. With older binaries, checkpoint output stays in disposable trial storage. It changes 0 persistent files and 0 PostgreSQL rows, but cannot measure durable commit cost. `--dry-run` executes 0 workloads and changes 0 files. `--apply-to ENV_FILE` offers a separate confirmation prompt for every recommendation; rejecting one setting leaves it unchanged and continues with the remaining settings. The `{docker_env}` command token expands to Docker `-e NAME=VALUE` arguments.

inserter/src/main.rs

@@ -4,6 +4,7 @@ mod generate_csv;
 mod logging;
 mod metrics;
 mod misc;
+mod test_mode;
 mod threads;
 
 use crate::checkpoint::{load_workloads, write_checkpoints};
@@ -96,6 +97,9 @@ fn run() -> Result<()> {
     if *BENCHMARK {
         log_info!("Benchmark mode enabled; database writes are disabled");
     }
+    if test_mode::enabled() {
+        log_info!("Test-only mode enabled; database transactions will be rolled back");
+    }
 
     if !check_db_connection() {
         if *BENCHMARK {
@@ -107,7 +111,11 @@ fn run() -> Result<()> {
 
     // Prepare thread resources
     log_info!("Loading workloads...");
-    let workloads = load_workloads().context("failed to load workloads")?;
+    let workloads = if test_mode::enabled() {
+        test_mode::workloads()
+    } else {
+        load_workloads().context("failed to load workloads")?
+    };
     let (entry_sender, entry_reciever): (Sender<(String, String)>, Receiver<(String, String)>) = bounded(*CHANNEL_SIZE);
 
     let mut work_handles = vec![];
@@ -132,7 +140,13 @@ fn run() -> Result<()> {
             commit_handles.push(
                 thread::Builder::new()
                     .name(format!("writer_{}", id))
-                    .spawn(move || commit_thread(thread_receiver))
+                    .spawn(move || {
+                        if test_mode::enabled() {
+                            test_mode::rollback_writer(thread_receiver)
+                        } else {
+                            commit_thread(thread_receiver)
+                        }
+                    })
                     .with_context(|| format!("failed to spawn writer thread {}", id))?,
             );
         }

inserter/src/test_mode.rs

@@ -0,0 +1,65 @@
+use crate::misc::{split_chunks, COPY_PLANET, COPY_STAR};
+use crate::{COMMITTED_SEEDS, COMMIT_COUNT, DB_STR, END_SEED, START_SEED, WORKER_THREADS};
+use anyhow::Result;
+use crossbeam_channel::{Receiver, RecvTimeoutError};
+use postgres::{Client, NoTls};
+use std::io::Write;
+use std::ops::Range;
+use std::sync::atomic::Ordering::SeqCst;
+use std::sync::LazyLock;
+use std::time::Duration;
+
+static ENABLED: LazyLock<bool> = LazyLock::new(|| {
+    std::env::var("TEST_ONLY")
+        .map(|value| value == "1")
+        .unwrap_or(false)
+});
+
+pub fn enabled() -> bool {
+    *ENABLED
+}
+
+pub fn workloads() -> Vec<Range<i32>> {
+    split_chunks(*START_SEED..*END_SEED, *WORKER_THREADS)
+}
+
+pub fn rollback_writer(rec: Receiver<(String, String)>) -> Result<()> {
+    let mut client = Client::connect(&*DB_STR.as_str(), NoTls)?;
+
+    'outer: loop {
+        let mut batch: Vec<(String, String)> = Vec::with_capacity(*COMMIT_COUNT);
+        'inner: for index in 0..*COMMIT_COUNT {
+            match rec.recv_timeout(Duration::from_secs(1)) {
+                Ok(message) => batch.push(message),
+                Err(RecvTimeoutError::Timeout) => {
+                    panic!("rollback_writer: channel stall detected (>1s lull)")
+                }
+                Err(RecvTimeoutError::Disconnected) => {
+                    if index == 0 {
+                        break 'outer;
+                    }
+                    break 'inner;
+                }
+            }
+        }
+
+        let mut transaction = client.transaction()?;
+        {
+            let mut copy = transaction.copy_in(COPY_STAR)?;
+            for (star, _) in &batch {
+                copy.write_all(star.as_bytes())?;
+            }
+            copy.finish()?;
+        }
+        {
+            let mut copy = transaction.copy_in(COPY_PLANET)?;
+            for (_, planet) in &batch {
+                copy.write_all(planet.as_bytes())?;
+            }
+            copy.finish()?;
+        }
+        transaction.rollback()?;
+        COMMITTED_SEEDS.fetch_add(batch.len() as i32, SeqCst);
+    }
+    Ok(())
+}

inserter/src/threads.rs

@@ -65,4 +65,4 @@ pub fn writer_sink(rec: Receiver<(String, String)>) -> Result<()> {
         }
     }
     Ok(())
-}
+}

tools/test_tune_pipeline.py

@@ -0,0 +1,62 @@
+import tempfile
+import unittest
+from pathlib import Path
+
+from tune_pipeline import build_command, confirmed_updates, recommend, update_env_file
+
+
+class TunePipelineTests(unittest.TestCase):
+    def test_recommend_uses_smallest_candidate_within_threshold(self):
+        recommended, peak = recommend(
+            {1: [80.0, 81.0], 2: [96.0, 98.0], 4: [100.0, 99.0], 8: [90.0]},
+            0.95,
+        )
+
+        self.assertEqual(2, recommended)
+        self.assertEqual(4, peak)
+
+    def test_docker_environment_placeholder_expands_to_arguments(self):
+        command = build_command(
+            "docker run --rm {docker_env} image",
+            {"WORKER_THREADS": "8", "BENCHMARK": "1"},
+        )
+
+        self.assertEqual(
+            [
+                "docker",
+                "run",
+                "--rm",
+                "-e",
+                "BENCHMARK=1",
+                "-e",
+                "WORKER_THREADS=8",
+                "image",
+            ],
+            command,
+        )
+
+    def test_env_update_preserves_unrelated_content(self):
+        with tempfile.TemporaryDirectory() as directory:
+            path = Path(directory) / "tuning.env"
+            path.write_text("# retained\nWORKER_THREADS=4\nPG_PORT=5432\n")
+
+            update_env_file(path, {"WORKER_THREADS": 8, "COMMIT_COUNT": 1000})
+
+            self.assertEqual(
+                "# retained\nWORKER_THREADS=8\nPG_PORT=5432\n\nCOMMIT_COUNT=1000\n",
+                path.read_text(),
+            )
+
+    def test_rejecting_one_update_does_not_skip_later_updates(self):
+        answers = iter(("no", "yes"))
+        accepted = confirmed_updates(
+            {"WORKER_THREADS": 8, "WRITER_THREADS": 4},
+            Path("tuning.env"),
+            lambda _: next(answers),
+        )
+
+        self.assertEqual({"WRITER_THREADS": 4}, accepted)
+
+
+if __name__ == "__main__":
+    unittest.main()