Phase 7 (partial): CPU thread/profile config + verified multi-worker scaling

Add worker/src/config.rs resolving RENDER_THREADS (auto | manual override)
and RENDER_PROFILE (server | consumer) into a rayon thread-pool size and a
per-xread batch size, unit tested. Restage main.rs's processing loop into
three stages per batch: async fetch, CPU-bound rasterize+mesh parallelized
across the batch on a sized rayon pool, then async store+ack — the
parallelism target is many chunks in flight at once, since a single 16x16
tile is too small for rayon to help within itself (per the pre-existing
doc comment in render/cpu.rs).

Verified locally: 3 worker instances against the same Redis stream split
24 queued dirty-chunk jobs with zero duplicates and zero drops (confirmed
via worker logs and tile_pointers rows), validating the consumer-group
design ahead of a real remote-worker deployment.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_015tKdPZt78zbPUZMXWzKEKt

worker/.env.example

@@ -11,3 +11,17 @@ MINIO_SECRET_KEY=changeme-set-in-untracked-env
 
 # cpu | gpu | hybrid — see render::backend. Only `cpu` exists so far (Phase 8 adds gpu/hybrid).
 RENDER_BACKEND=cpu
+
+# Thread count for the rayon pool that renders a batch of dirty chunks in parallel (see
+# config.rs/main.rs — a single 16x16 tile is too small for rayon to help within itself, so the
+# parallelism target is many chunks in flight at once). `auto` (default) uses
+# std::thread::available_parallelism(); set a positive integer to override (0/garbage also falls
+# back to auto).
+RENDER_THREADS=auto
+
+# server | consumer — tunes how many dirty-chunk entries are pulled off the Redis stream per
+# xread cycle before being rendered in parallel. `server` (default) batches more aggressively —
+# many small batches so every core on a many-core Xeon-style box stays fed; `consumer` uses
+# fewer, larger batches, less scheduling overhead per chunk on fewer/faster cores. Unrecognized
+# values fall back to `server`, matching this project's own primary deployment target.
+RENDER_PROFILE=server

worker/src/config.rs

@@ -0,0 +1,105 @@
+/// Resolves `RENDER_THREADS`/`RENDER_PROFILE` into a concrete thread count + per-xread batch
+/// size for the rayon pool that parallelizes tile/mesh rendering across a batch of dirty chunks
+/// (see main.rs — a single 16x16 tile is too small for rayon to help within itself, so the
+/// parallelism target is "many chunks in flight at once", tuned by these two knobs). Pure/testable
+/// on purpose — env var reading and rayon pool construction stay in main.rs.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum Profile {
+    /// Many-core Xeon-style deployment (the user's own production box): many small batches so
+    /// every core stays fed without one slow chunk blocking a large batch's completion.
+    Server,
+    /// Fewer/faster consumer cores: fewer, larger batches — less scheduling overhead per chunk.
+    Consumer,
+}
+
+impl Profile {
+    pub fn parse(raw: &str) -> Self {
+        match raw {
+            "consumer" => Profile::Consumer,
+            // Unrecognized falls back to `server`, not `consumer` — matches this project's own
+            // primary deployment target (RTX 3090 + Xeons), see the plan's worker section.
+            _ => Profile::Server,
+        }
+    }
+}
+
+pub struct WorkerConfig {
+    pub threads: usize,
+    pub batch_size: usize,
+}
+
+/// `raw` is `RENDER_THREADS`'s value: `None`/`"auto"` uses `available`, anything else must parse
+/// to a positive integer or it also falls back to `available` (never zero threads).
+pub fn resolve_thread_count(raw: Option<&str>, available: usize) -> usize {
+    match raw {
+        None => available,
+        Some(v) => match v.trim() {
+            "auto" | "" => available,
+            n => match n.parse::<usize>() {
+                Ok(0) | Err(_) => available,
+                Ok(n) => n,
+            },
+        },
+    }
+}
+
+/// How many dirty-chunk stream entries to pull (and then render in parallel) per xread cycle.
+pub fn batch_size_for(profile: Profile) -> usize {
+    match profile {
+        Profile::Server => 32,
+        Profile::Consumer => 4,
+    }
+}
+
+pub fn resolve(threads_raw: Option<&str>, profile_raw: Option<&str>, available: usize) -> WorkerConfig {
+    let profile = Profile::parse(profile_raw.unwrap_or("server"));
+    WorkerConfig { threads: resolve_thread_count(threads_raw, available), batch_size: batch_size_for(profile) }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn auto_or_unset_uses_available_parallelism() {
+        assert_eq!(resolve_thread_count(None, 8), 8);
+        assert_eq!(resolve_thread_count(Some("auto"), 8), 8);
+        assert_eq!(resolve_thread_count(Some(""), 8), 8);
+    }
+
+    #[test]
+    fn a_manual_positive_integer_overrides_available_parallelism() {
+        assert_eq!(resolve_thread_count(Some("4"), 32), 4);
+        assert_eq!(resolve_thread_count(Some("64"), 4), 64);
+    }
+
+    #[test]
+    fn zero_or_garbage_falls_back_to_available_parallelism_rather_than_hanging_with_no_threads() {
+        assert_eq!(resolve_thread_count(Some("0"), 8), 8);
+        assert_eq!(resolve_thread_count(Some("not-a-number"), 8), 8);
+        assert_eq!(resolve_thread_count(Some("-1"), 8), 8);
+    }
+
+    #[test]
+    fn server_profile_batches_more_aggressively_than_consumer() {
+        assert!(batch_size_for(Profile::Server) > batch_size_for(Profile::Consumer));
+    }
+
+    #[test]
+    fn unrecognized_profile_falls_back_to_server_matching_the_primary_deployment_target() {
+        assert_eq!(Profile::parse("laptop"), Profile::Server);
+        assert_eq!(Profile::parse(""), Profile::Server);
+    }
+
+    #[test]
+    fn consumer_profile_is_recognized_case_sensitively() {
+        assert_eq!(Profile::parse("consumer"), Profile::Consumer);
+    }
+
+    #[test]
+    fn resolve_combines_both_knobs() {
+        let cfg = resolve(Some("6"), Some("consumer"), 32);
+        assert_eq!(cfg.threads, 6);
+        assert_eq!(cfg.batch_size, batch_size_for(Profile::Consumer));
+    }
+}

worker/src/main.rs

@@ -1,3 +1,4 @@
+mod config;
 mod db;
 mod mesh;
 mod palette;
@@ -8,6 +9,7 @@ use std::collections::hash_map::DefaultHasher;
 use std::hash::{Hash, Hasher};
 
 use base64::{engine::general_purpose::STANDARD, Engine as _};
+use rayon::prelude::*;
 use redis::streams::{StreamReadOptions, StreamReadReply};
 use redis::AsyncCommands;
 use render::{ColumnPixel, CpuRenderBackend, RenderBackend};
@@ -57,13 +59,30 @@ async fn main() -> anyhow::Result<()> {
         );
     }
     let backend: Box<dyn RenderBackend> = Box::new(CpuRenderBackend::new());
+
+    // A single 16x16 tile is too small for rayon to help within itself, so the parallelism
+    // target is "many chunks in flight at once" — see config.rs's doc comment. RENDER_THREADS/
+    // RENDER_PROFILE (server=many small batches for many-core Xeons, consumer=fewer/larger
+    // batches) tune the pool size and how many stream entries are pulled per xread cycle.
+    let available = std::thread::available_parallelism().map(|n| n.get()).unwrap_or(4);
+    let worker_cfg = config::resolve(
+        std::env::var("RENDER_THREADS").ok().as_deref(),
+        std::env::var("RENDER_PROFILE").ok().as_deref(),
+        available,
+    );
+    let pool = rayon::ThreadPoolBuilder::new().num_threads(worker_cfg.threads).build()?;
+
     println!("[worker] render backend: {}", backend.name());
+    println!(
+        "[worker] threads={} batch_size={} (available_parallelism={available})",
+        worker_cfg.threads, worker_cfg.batch_size
+    );
     println!("[worker] consuming stream '{DIRTY_CHUNK_STREAM}' as '{consumer_name}'");
 
     loop {
         let opts = StreamReadOptions::default()
             .group(CONSUMER_GROUP, &consumer_name)
-            .count(10)
+            .count(worker_cfg.batch_size)
             .block(5000);
 
         let reply: StreamReadReply = conn
@@ -71,15 +90,51 @@ async fn main() -> anyhow::Result<()> {
             .await
             .unwrap_or_default();
 
+        let mut jobs = Vec::new();
         for key in reply.keys {
             for entry in key.ids {
-                if let Err(err) = process_entry(&pg_pool, &s3_client, backend.as_ref(), &entry).await {
-                    eprintln!("[worker] failed to process {}: {err:#}", entry.id);
+                match parse_job(&entry) {
+                    Ok(job) => jobs.push(job),
+                    Err(err) => eprintln!("[worker] failed to parse {}: {err:#}", entry.id),
+                }
+            }
+        }
+        if jobs.is_empty() {
+            continue;
+        }
+
+        // Stage 1 (async): fetch each job's source data. Sequential — the pg pool is capped at
+        // 5 connections (see db::connect) so unbounded concurrency here wouldn't help, and
+        // keeping this simple leaves the actual parallelism budget for the CPU-bound stage below.
+        let mut fetched = Vec::with_capacity(jobs.len());
+        for job in jobs {
+            match fetch_chunk_data(&pg_pool, &job).await {
+                Ok(data) => fetched.push(data),
+                Err(err) => eprintln!("[worker] failed to fetch {}: {err:#}", job.entry_id),
+            }
+        }
+
+        // Stage 2 (CPU-bound, parallel): rasterize + mesh every chunk in the batch across the
+        // sized rayon pool.
+        let backend_ref = backend.as_ref();
+        let rendered: Vec<anyhow::Result<RenderedChunk>> =
+            pool.install(|| fetched.par_iter().map(|data| render_chunk(data, backend_ref)).collect());
+
+        // Stage 3 (async): write results back and ack, sequentially.
+        for result in rendered {
+            let chunk = match result {
+                Ok(chunk) => chunk,
+                Err(err) => {
+                    eprintln!("[worker] failed to render: {err:#}");
                     continue;
                 }
-                let _: redis::RedisResult<()> =
-                    conn.xack(DIRTY_CHUNK_STREAM, CONSUMER_GROUP, &[&entry.id]).await;
+            };
+            let entry_id = chunk.job.entry_id.clone();
+            if let Err(err) = store_rendered_chunk(&pg_pool, &s3_client, chunk).await {
+                eprintln!("[worker] failed to store {entry_id}: {err:#}");
+                continue;
             }
+            let _: redis::RedisResult<()> = conn.xack(DIRTY_CHUNK_STREAM, CONSUMER_GROUP, &[&entry_id]).await;
         }
     }
 }
@@ -92,52 +147,57 @@ fn field(entry: &redis::streams::StreamId, name: &str) -> anyhow::Result<String>
     }
 }
 
-async fn process_entry(
-    pool: &sqlx::PgPool,
-    s3_client: &aws_sdk_s3::Client,
-    backend: &dyn RenderBackend,
-    entry: &redis::streams::StreamId,
-) -> anyhow::Result<()> {
-    let server_id: Uuid = field(entry, "serverId")?.parse()?;
-    let dimension: i32 = field(entry, "dimension")?.parse()?;
-    let chunk_x: i32 = field(entry, "chunkX")?.parse()?;
-    let chunk_z: i32 = field(entry, "chunkZ")?.parse()?;
+#[derive(Clone)]
+struct ChunkJob {
+    entry_id: String,
+    server_id: Uuid,
+    dimension: i32,
+    chunk_x: i32,
+    chunk_z: i32,
+}
 
-    let columns = db::fetch_chunk_columns(pool, server_id, dimension, chunk_x, chunk_z).await?;
-    let pixels: Vec<ColumnPixel> = columns
-        .iter()
-        .map(|c| ColumnPixel {
-            local_x: (c.x - chunk_x * 16) as u8,
-            local_z: (c.z - chunk_z * 16) as u8,
-            block_id: c.block_id as u16,
-            block_meta: c.block_meta as u8,
-        })
-        .collect();
+fn parse_job(entry: &redis::streams::StreamId) -> anyhow::Result<ChunkJob> {
+    Ok(ChunkJob {
+        entry_id: entry.id.clone(),
+        server_id: field(entry, "serverId")?.parse()?,
+        dimension: field(entry, "dimension")?.parse()?,
+        chunk_x: field(entry, "chunkX")?.parse()?,
+        chunk_z: field(entry, "chunkZ")?.parse()?,
+    })
+}
 
-    let png_bytes = backend.rasterize_tile(&pixels)?;
+struct ChunkData {
+    job: ChunkJob,
+    columns: Vec<db::StoredColumn>,
+    sections: Vec<db::StoredSection>,
+}
 
+async fn fetch_chunk_data(pool: &sqlx::PgPool, job: &ChunkJob) -> anyhow::Result<ChunkData> {
+    let columns =
+        db::fetch_chunk_columns(pool, job.server_id, job.dimension, job.chunk_x, job.chunk_z).await?;
+    let sections =
+        db::fetch_chunk_sections(pool, job.server_id, job.dimension, job.chunk_x, job.chunk_z).await?;
+    Ok(ChunkData { job: job.clone(), columns, sections })
+}
+
+struct RenderedMesh {
+    section_y: i32,
+    bytes: Vec<u8>,
+    content_hash: String,
+}
+
+struct RenderedChunk {
+    job: ChunkJob,
+    png_bytes: Vec<u8>,
+    tile_content_hash: String,
+    column_count: usize,
+    meshes: Vec<RenderedMesh>,
+}
+
+fn content_hash(bytes: &[u8]) -> String {
     let mut hasher = DefaultHasher::new();
-    png_bytes.hash(&mut hasher);
-    let content_hash = format!("{:x}", hasher.finish());
-
-    let storage_key = format!("{server_id}/{dimension}/0/{chunk_x}/{chunk_z}.png");
-    storage::put_object(s3_client, &storage_key, "image/png", png_bytes).await?;
-    db::upsert_tile_pointer(
-        pool,
-        server_id,
-        dimension,
-        0,
-        chunk_x,
-        chunk_z,
-        &storage_key,
-        &content_hash,
-    )
-    .await?;
-
-    println!("[worker] rendered tile {storage_key} ({} columns)", pixels.len());
-
-    mesh_chunk(pool, s3_client, server_id, dimension, chunk_x, chunk_z).await?;
-    Ok(())
+    bytes.hash(&mut hasher);
+    format!("{:x}", hasher.finish())
 }
 
 fn decode_blocks(base64_blocks: &str) -> anyhow::Result<[u16; 4096]> {
@@ -152,56 +212,92 @@ fn decode_blocks(base64_blocks: &str) -> anyhow::Result<[u16; 4096]> {
     Ok(blocks)
 }
 
-async fn mesh_chunk(
-    pool: &sqlx::PgPool,
-    s3_client: &aws_sdk_s3::Client,
-    server_id: Uuid,
-    dimension: i32,
-    chunk_x: i32,
-    chunk_z: i32,
-) -> anyhow::Result<()> {
-    let sections = db::fetch_chunk_sections(pool, server_id, dimension, chunk_x, chunk_z).await?;
-    if sections.is_empty() {
-        return Ok(()); // this server hasn't sent 3D data yet (Phase 2 mod support) — fine, no-op
-    }
+// Pure CPU-bound work — called from inside the rayon pool, one call per chunk in the batch.
+fn render_chunk(data: &ChunkData, backend: &dyn RenderBackend) -> anyhow::Result<RenderedChunk> {
+    let job = &data.job;
+    let pixels: Vec<ColumnPixel> = data
+        .columns
+        .iter()
+        .map(|c| ColumnPixel {
+            local_x: (c.x - job.chunk_x * 16) as u8,
+            local_z: (c.z - job.chunk_z * 16) as u8,
+            block_id: c.block_id as u16,
+            block_meta: c.block_meta as u8,
+        })
+        .collect();
+    let png_bytes = backend.rasterize_tile(&pixels)?;
+    let tile_content_hash = content_hash(&png_bytes);
 
-    for section in sections {
+    let mut meshes = Vec::new();
+    for section in &data.sections {
         let blocks = match decode_blocks(&section.blocks) {
             Ok(b) => b,
             Err(err) => {
                 eprintln!(
-                    "[worker] skipping malformed section ({chunk_x},{chunk_z},{}): {err:#}",
-                    section.section_y
+                    "[worker] skipping malformed section ({},{},{}): {err:#}",
+                    job.chunk_x, job.chunk_z, section.section_y
                 );
                 continue;
             }
         };
-
         let mesh_buf = mesh::mesh_section(&blocks);
         if mesh_buf.is_empty() {
             continue;
         }
-        let mesh_bytes = mesh_buf.encode();
+        let bytes = mesh_buf.encode();
+        let hash = content_hash(&bytes);
+        meshes.push(RenderedMesh { section_y: section.section_y, bytes, content_hash: hash });
+    }
 
-        let mut hasher = DefaultHasher::new();
-        mesh_bytes.hash(&mut hasher);
-        let content_hash = format!("{:x}", hasher.finish());
+    Ok(RenderedChunk {
+        job: job.clone(),
+        png_bytes,
+        tile_content_hash,
+        column_count: pixels.len(),
+        meshes,
+    })
+}
 
-        let storage_key =
-            format!("{server_id}/{dimension}/mesh/{chunk_x}/{chunk_z}/{}.bin", section.section_y);
-        storage::put_object(s3_client, &storage_key, "application/octet-stream", mesh_bytes).await?;
+async fn store_rendered_chunk(
+    pool: &sqlx::PgPool,
+    s3_client: &aws_sdk_s3::Client,
+    chunk: RenderedChunk,
+) -> anyhow::Result<()> {
+    let job = &chunk.job;
+    let storage_key = format!("{}/{}/0/{}/{}.png", job.server_id, job.dimension, job.chunk_x, job.chunk_z);
+    storage::put_object(s3_client, &storage_key, "image/png", chunk.png_bytes).await?;
+    db::upsert_tile_pointer(
+        pool,
+        job.server_id,
+        job.dimension,
+        0,
+        job.chunk_x,
+        job.chunk_z,
+        &storage_key,
+        &chunk.tile_content_hash,
+    )
+    .await?;
+    println!("[worker] rendered tile {storage_key} ({} columns)", chunk.column_count);
+
+    for mesh in chunk.meshes {
+        let storage_key = format!(
+            "{}/{}/mesh/{}/{}/{}.bin",
+            job.server_id, job.dimension, job.chunk_x, job.chunk_z, mesh.section_y
+        );
+        storage::put_object(s3_client, &storage_key, "application/octet-stream", mesh.bytes).await?;
         db::upsert_mesh_pointer(
             pool,
-            server_id,
-            dimension,
-            chunk_x,
-            chunk_z,
-            section.section_y,
+            job.server_id,
+            job.dimension,
+            job.chunk_x,
+            job.chunk_z,
+            mesh.section_y,
             &storage_key,
-            &content_hash,
+            &mesh.content_hash,
         )
         .await?;
         println!("[worker] rendered mesh {storage_key}");
     }
+
     Ok(())
 }