From 0b0d073bc5c23d05a338e2fe7b19b2c1e3939540 Mon Sep 17 00:00:00 2001
From: Junhao Liao <junhao@junhao.ca>
Date: Tue, 21 Jul 2026 21:24:01 -0400
Subject: [PATCH 1/7] feat(log-ingestor): Make database connection pool size
 configurable

---
 .../clp-py-utils/clp_py_utils/clp_config.py   |  2 ++
 .../src/clp_config/package/config.rs          | 27 ++++++++++++++++++-
 .../log-ingestor/src/ingestion_job_manager.rs |  4 ---
 .../ingestion_job_manager/clp_ingestion.rs    |  9 ++++++-
 .../src/etc/clp-config.template.json.yaml     |  1 +
 5 files changed, 37 insertions(+), 6 deletions(-)

diff --git a/components/clp-py-utils/clp_py_utils/clp_config.py b/components/clp-py-utils/clp_py_utils/clp_config.py
index 46ea77940c..1b4f29b854 100644
--- a/components/clp-py-utils/clp_py_utils/clp_config.py
+++ b/components/clp-py-utils/clp_py_utils/clp_config.py
@@ -97,6 +97,7 @@
 # Specific types
 # TODO: Replace this with pydantic_extra_types.domain.DomainStr.
 DomainStr = NonEmptyStr
+DatabaseConnectionPoolSize = Annotated[int, Field(gt=0, lt=2**32)]
 Port = Annotated[int, Field(gt=0, lt=2**16)]
 SerializablePath = Annotated[pathlib.Path, PlainSerializer(serialize_path)]
 ZstdCompressionLevel = Annotated[int, Field(ge=1, le=19)]
@@ -775,6 +776,7 @@ class ApiServer(BaseModel):
 class LogIngestor(BaseModel):
     host: DomainStr = "localhost"
     port: Port = 3002
+    database_connection_pool_size: DatabaseConnectionPoolSize = 100
     logging_level: LoggingLevelRust = "INFO"
 
 
diff --git a/components/clp-rust-utils/src/clp_config/package/config.rs b/components/clp-rust-utils/src/clp_config/package/config.rs
index 98c62c55a5..8418faccf2 100644
--- a/components/clp-rust-utils/src/clp_config/package/config.rs
+++ b/components/clp-rust-utils/src/clp_config/package/config.rs
@@ -1,3 +1,5 @@
+use std::num::NonZeroU32;
+
 use serde::Deserialize;
 
 use crate::clp_config::{AwsAuthentication, S3Config};
@@ -218,6 +220,9 @@ impl Default for StreamOutputStorage {
     }
 }
 
+const DEFAULT_LOG_INGESTOR_DATABASE_CONNECTION_POOL_SIZE: NonZeroU32 =
+    NonZeroU32::new(100).unwrap();
+
 /// Mirror of `clp_py_utils.clp_config.LogIngestor`.
 ///
 /// # NOTE
@@ -228,6 +233,7 @@ impl Default for StreamOutputStorage {
 pub struct LogIngestor {
     pub host: String,
     pub port: u16,
+    pub database_connection_pool_size: NonZeroU32,
     pub logging_level: String,
 }
 
@@ -236,6 +242,7 @@ impl Default for LogIngestor {
         Self {
             host: "localhost".to_owned(),
             port: 3002,
+            database_connection_pool_size: DEFAULT_LOG_INGESTOR_DATABASE_CONNECTION_POOL_SIZE,
             logging_level: "INFO".to_owned(),
         }
     }
@@ -330,7 +337,25 @@ impl Default for Telemetry {
 
 #[cfg(test)]
 mod tests {
-    use super::LogsInput;
+    use super::{LogIngestor, LogsInput};
+
+    #[test]
+    fn deserialize_log_ingestor_database_connection_pool_size() {
+        let default_config = serde_json::from_str::<LogIngestor>("{}")
+            .expect("failed to deserialize default `LogIngestor` config");
+        assert_eq!(100, default_config.database_connection_pool_size.get());
+
+        let custom_config =
+            serde_json::from_str::<LogIngestor>(r#"{"database_connection_pool_size": 42}"#)
+                .expect("failed to deserialize custom `LogIngestor` config");
+        assert_eq!(42, custom_config.database_connection_pool_size.get());
+    }
+
+    #[test]
+    fn reject_zero_log_ingestor_database_connection_pool_size() {
+        let result = serde_json::from_str::<LogIngestor>(r#"{"database_connection_pool_size": 0}"#);
+        assert!(result.is_err());
+    }
 
     #[test]
     fn deserialize_logs_input_s3_config() {
diff --git a/components/log-ingestor/src/ingestion_job_manager.rs b/components/log-ingestor/src/ingestion_job_manager.rs
index 00639daf2c..07a8aba3ae 100644
--- a/components/log-ingestor/src/ingestion_job_manager.rs
+++ b/components/log-ingestor/src/ingestion_job_manager.rs
@@ -73,10 +73,6 @@ impl IngestionJobManagerState {
     ///
     /// * [`anyhow::Error`] if the logs input type in the CLP configuration is unsupported.
     /// * Forwards [`ClpDbIngestionConnector::connect`]'s return values on failure.
-    ///
-    /// # Panics
-    ///
-    /// Panics if `clp_config.log_ingestor` is `None`.
     pub async fn from_config(
         clp_config: ClpConfig,
         clp_credentials: ClpCredentials,
diff --git a/components/log-ingestor/src/ingestion_job_manager/clp_ingestion.rs b/components/log-ingestor/src/ingestion_job_manager/clp_ingestion.rs
index 7b65b5ae71..d39236254c 100644
--- a/components/log-ingestor/src/ingestion_job_manager/clp_ingestion.rs
+++ b/components/log-ingestor/src/ingestion_job_manager/clp_ingestion.rs
@@ -178,6 +178,7 @@ impl ClpDbIngestionConnector {
     ///
     /// * Forwards [`clp_rust_utils::database::mysql::create_clp_db_mysql_pool`]'s return values on
     ///   failure.
+    /// * [`anyhow::Error`] if the log-ingestor configuration is missing.
     /// * Forwards [`Self::create_tables`]'s return values on failure.
     /// * Forwards [`Self::get_unfinished_compression_jobs`]'s return values on failure.
     /// * Forwards [`Self::load_ingestion_jobs`]'s return values on failure.
@@ -199,10 +200,16 @@ impl ClpDbIngestionConnector {
             }
         };
 
+        let database_connection_pool_size = clp_config
+            .log_ingestor
+            .as_ref()
+            .ok_or_else(|| anyhow::anyhow!("Invalid CLP config: log-ingestor is not configured"))?
+            .database_connection_pool_size
+            .get();
         let mysql_pool = clp_rust_utils::database::mysql::create_clp_db_mysql_pool(
             &clp_config.database,
             &clp_credentials.database,
-            100,
+            database_connection_pool_size,
         )
         .await?;
 
diff --git a/components/package-template/src/etc/clp-config.template.json.yaml b/components/package-template/src/etc/clp-config.template.json.yaml
index 63c8af4336..64abed24f0 100644
--- a/components/package-template/src/etc/clp-config.template.json.yaml
+++ b/components/package-template/src/etc/clp-config.template.json.yaml
@@ -22,6 +22,7 @@ telemetry:
 #log_ingestor:
 #  host: "localhost"
 #  port: 3002
+#  database_connection_pool_size: 100
 #  logging_level: "INFO"
 
 ## Location (e.g., directory) containing any logs you wish to compress. Must be reachable by all

From 09c8109b6d6ae08c310235ed9ae2304e3ace93f7 Mon Sep 17 00:00:00 2001
From: Junhao Liao <junhao@junhao.ca>
Date: Tue, 21 Jul 2026 21:54:01 -0400
Subject: [PATCH 2/7] refactor(log-ingestor): Use context for missing
 configuration

---
 .../log-ingestor/src/ingestion_job_manager/clp_ingestion.rs    | 3 ++-
 1 file changed, 2 insertions(+), 1 deletion(-)

diff --git a/components/log-ingestor/src/ingestion_job_manager/clp_ingestion.rs b/components/log-ingestor/src/ingestion_job_manager/clp_ingestion.rs
index d39236254c..6c6a341c5a 100644
--- a/components/log-ingestor/src/ingestion_job_manager/clp_ingestion.rs
+++ b/components/log-ingestor/src/ingestion_job_manager/clp_ingestion.rs
@@ -1,5 +1,6 @@
 use std::time::Duration;
 
+use anyhow::Context;
 use async_trait::async_trait;
 use clp_rust_utils::{
     clp_config::{
@@ -203,7 +204,7 @@ impl ClpDbIngestionConnector {
         let database_connection_pool_size = clp_config
             .log_ingestor
             .as_ref()
-            .ok_or_else(|| anyhow::anyhow!("Invalid CLP config: log-ingestor is not configured"))?
+            .context("Invalid CLP config: log-ingestor is not configured")?
             .database_connection_pool_size
             .get();
         let mysql_pool = clp_rust_utils::database::mysql::create_clp_db_mysql_pool(

From 41f8bcc7d4fe4f9b351ba5d38ac4e06726c1d21a Mon Sep 17 00:00:00 2001
From: Junhao Liao <junhao@junhao.ca>
Date: Tue, 21 Jul 2026 22:34:11 -0400
Subject: [PATCH 3/7] docs(log-ingestor): Order connection errors consistently

---
 .../log-ingestor/src/ingestion_job_manager/clp_ingestion.rs     | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/components/log-ingestor/src/ingestion_job_manager/clp_ingestion.rs b/components/log-ingestor/src/ingestion_job_manager/clp_ingestion.rs
index 6c6a341c5a..56a099eccb 100644
--- a/components/log-ingestor/src/ingestion_job_manager/clp_ingestion.rs
+++ b/components/log-ingestor/src/ingestion_job_manager/clp_ingestion.rs
@@ -177,9 +177,9 @@ impl ClpDbIngestionConnector {
     ///
     /// Returns an error if:
     ///
+    /// * [`anyhow::Error`] if the log-ingestor configuration is missing.
     /// * Forwards [`clp_rust_utils::database::mysql::create_clp_db_mysql_pool`]'s return values on
     ///   failure.
-    /// * [`anyhow::Error`] if the log-ingestor configuration is missing.
     /// * Forwards [`Self::create_tables`]'s return values on failure.
     /// * Forwards [`Self::get_unfinished_compression_jobs`]'s return values on failure.
     /// * Forwards [`Self::load_ingestion_jobs`]'s return values on failure.

From 0acbd716e8d8901db39a3fd5c38b015383f07397 Mon Sep 17 00:00:00 2001
From: Junhao Liao <junhao@junhao.ca>
Date: Tue, 21 Jul 2026 23:52:48 -0400
Subject: [PATCH 4/7] feat(helm): Expose log-ingestor database pool size

---
 tools/deployment/package-helm/Chart.yaml               | 2 +-
 tools/deployment/package-helm/templates/configmap.yaml | 1 +
 tools/deployment/package-helm/values.yaml              | 1 +
 3 files changed, 3 insertions(+), 1 deletion(-)

diff --git a/tools/deployment/package-helm/Chart.yaml b/tools/deployment/package-helm/Chart.yaml
index 47a8f305dd..12f1fcac68 100644
--- a/tools/deployment/package-helm/Chart.yaml
+++ b/tools/deployment/package-helm/Chart.yaml
@@ -1,6 +1,6 @@
 apiVersion: "v2"
 name: "clp"
-version: "0.4.1-dev.1"
+version: "0.4.1-dev.2"
 description: "A Helm chart for CLP's (Compressed Log Processor) package deployment"
 type: "application"
 appVersion: "0.13.1-dev"
diff --git a/tools/deployment/package-helm/templates/configmap.yaml b/tools/deployment/package-helm/templates/configmap.yaml
index ee49584a06..1ea4bc4eea 100644
--- a/tools/deployment/package-helm/templates/configmap.yaml
+++ b/tools/deployment/package-helm/templates/configmap.yaml
@@ -121,6 +121,7 @@ data:
     {{- end }}{{/* with .Values.clpConfig.logs_input */}}
     {{- with .Values.clpConfig.log_ingestor }}
     log_ingestor:
+      database_connection_pool_size: {{ .database_connection_pool_size | int }}
       host: "localhost"
       logging_level: {{ .logging_level | quote }}
       port: 3002
diff --git a/tools/deployment/package-helm/values.yaml b/tools/deployment/package-helm/values.yaml
index a52dd21729..7e86bcc589 100644
--- a/tools/deployment/package-helm/values.yaml
+++ b/tools/deployment/package-helm/values.yaml
@@ -284,6 +284,7 @@ clpConfig:
 
   # log-ingestor config. Currently, the config is applicable only if `logs_input.type` is "s3".
   log_ingestor:
+    database_connection_pool_size: 100
     port: 30302
     logging_level: "INFO"
 

From b6a8b32b07440415327cbaf489b181cdf320abee Mon Sep 17 00:00:00 2001
From: Junhao Liao <junhao@junhao.ca>
Date: Fri, 24 Jul 2026 18:22:14 -0400
Subject: [PATCH 5/7] refactor(clp-rust-utils): Inline log-ingestor pool
 default

---
 components/clp-rust-utils/src/clp_config/package/config.rs | 6 ++----
 1 file changed, 2 insertions(+), 4 deletions(-)

diff --git a/components/clp-rust-utils/src/clp_config/package/config.rs b/components/clp-rust-utils/src/clp_config/package/config.rs
index 36817f150f..f14481ebf4 100644
--- a/components/clp-rust-utils/src/clp_config/package/config.rs
+++ b/components/clp-rust-utils/src/clp_config/package/config.rs
@@ -229,9 +229,6 @@ impl Default for StreamOutputStorage {
     }
 }
 
-const DEFAULT_LOG_INGESTOR_DATABASE_CONNECTION_POOL_SIZE: NonZeroU32 =
-    NonZeroU32::new(100).unwrap();
-
 /// Mirror of `clp_py_utils.clp_config.LogIngestor`.
 ///
 /// # NOTE
@@ -251,7 +248,8 @@ impl Default for LogIngestor {
         Self {
             host: "localhost".to_owned(),
             port: 3002,
-            database_connection_pool_size: DEFAULT_LOG_INGESTOR_DATABASE_CONNECTION_POOL_SIZE,
+            database_connection_pool_size: NonZeroU32::new(100)
+                .expect("default database connection pool size must be nonzero"),
             logging_level: "INFO".to_owned(),
         }
     }

From 956f27c16d8fe1e6cdc2788248c29357fb712a93 Mon Sep 17 00:00:00 2001
From: Junhao Liao <junhao@junhao.ca>
Date: Fri, 24 Jul 2026 20:41:13 -0400
Subject: [PATCH 6/7] style(rust): Align tests and error messages

---
 Cargo.lock                                    |  1 +
 components/clp-rust-utils/Cargo.toml          |  1 +
 .../src/clp_config/package/config.rs          | 43 ++++++++++---------
 .../log-ingestor/src/ingestion_job_manager.rs | 12 +++---
 .../ingestion_job_manager/clp_ingestion.rs    |  6 +--
 5 files changed, 34 insertions(+), 29 deletions(-)

diff --git a/Cargo.lock b/Cargo.lock
index 640fcda85e..ec1a8e21ec 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -942,6 +942,7 @@ checksum = "c8d4a3bb8b1e0c1050499d1815f5ab16d04f0959b233085fb31653fbfc9d98f9"
 name = "clp-rust-utils"
 version = "0.13.1-dev"
 dependencies = [
+ "anyhow",
  "aws-config",
  "aws-sdk-s3",
  "aws-sdk-sqs",
diff --git a/components/clp-rust-utils/Cargo.toml b/components/clp-rust-utils/Cargo.toml
index fb7445c1eb..f5387e1b2a 100644
--- a/components/clp-rust-utils/Cargo.toml
+++ b/components/clp-rust-utils/Cargo.toml
@@ -27,5 +27,6 @@ tracing-subscriber = { version = "0.3.22", features = ["json", "env-filter", "fm
 utoipa = { version = "5.4.0" }
 
 [dev-dependencies]
+anyhow = "1.0.100"
 hex = "0.4.3"
 serde_json = "1.0.149"
diff --git a/components/clp-rust-utils/src/clp_config/package/config.rs b/components/clp-rust-utils/src/clp_config/package/config.rs
index f14481ebf4..39aec1e381 100644
--- a/components/clp-rust-utils/src/clp_config/package/config.rs
+++ b/components/clp-rust-utils/src/clp_config/package/config.rs
@@ -347,25 +347,28 @@ mod tests {
     use super::{LogIngestor, LogsInput};
 
     #[test]
-    fn deserialize_log_ingestor_database_connection_pool_size() {
-        let default_config = serde_json::from_str::<LogIngestor>("{}")
-            .expect("failed to deserialize default `LogIngestor` config");
+    fn deserialize_log_ingestor_database_connection_pool_size() -> anyhow::Result<()> {
+        let default_config = serde_json::from_str::<LogIngestor>("{}")?;
         assert_eq!(100, default_config.database_connection_pool_size.get());
 
         let custom_config =
-            serde_json::from_str::<LogIngestor>(r#"{"database_connection_pool_size": 42}"#)
-                .expect("failed to deserialize custom `LogIngestor` config");
+            serde_json::from_str::<LogIngestor>(r#"{"database_connection_pool_size": 42}"#)?;
         assert_eq!(42, custom_config.database_connection_pool_size.get());
+        Ok(())
     }
 
     #[test]
-    fn reject_zero_log_ingestor_database_connection_pool_size() {
+    fn reject_zero_log_ingestor_database_connection_pool_size() -> anyhow::Result<()> {
         let result = serde_json::from_str::<LogIngestor>(r#"{"database_connection_pool_size": 0}"#);
-        assert!(result.is_err());
+        anyhow::ensure!(
+            result.is_err(),
+            "zero database connection pool size was accepted"
+        );
+        Ok(())
     }
 
     #[test]
-    fn deserialize_logs_input_s3_config() {
+    fn deserialize_logs_input_s3_config() -> anyhow::Result<()> {
         const ACCESS_KEY_ID: &str = "YSCOPE";
         const SECRET_ACCESS_KEY: &str = "IamSecret";
         let logs_input_config_json = serde_json::json!({
@@ -380,8 +383,7 @@ mod tests {
         });
 
         let deserialized =
-            serde_json::from_str::<LogsInput>(logs_input_config_json.to_string().as_str())
-                .expect("failed to deserialize `LogsInput` from JSON");
+            serde_json::from_str::<LogsInput>(logs_input_config_json.to_string().as_str())?;
 
         match deserialized {
             LogsInput::S3 { config } => match config.aws_authentication {
@@ -390,15 +392,16 @@ mod tests {
                     assert_eq!(credentials.secret_access_key, SECRET_ACCESS_KEY);
                 }
                 crate::clp_config::AwsAuthentication::Default => {
-                    panic!("Expected credentials, got `default`")
+                    panic!("expected credentials, got `default`")
                 }
             },
-            LogsInput::Fs { .. } => panic!("Expected S3"),
+            LogsInput::Fs { .. } => panic!("expected S3"),
         }
+        Ok(())
     }
 
     #[test]
-    fn deserialize_logs_input_s3_default_config() {
+    fn deserialize_logs_input_s3_default_config() -> anyhow::Result<()> {
         let logs_input_config_json = serde_json::json!({
             "type": "s3",
             "aws_authentication": {
@@ -407,8 +410,7 @@ mod tests {
         });
 
         let deserialized =
-            serde_json::from_str::<LogsInput>(logs_input_config_json.to_string().as_str())
-                .expect("failed to deserialize `LogsInput` from JSON");
+            serde_json::from_str::<LogsInput>(logs_input_config_json.to_string().as_str())?;
 
         match deserialized {
             LogsInput::S3 { config } => {
@@ -417,12 +419,13 @@ mod tests {
                     crate::clp_config::AwsAuthentication::Default
                 );
             }
-            LogsInput::Fs { .. } => panic!("Expected S3"),
+            LogsInput::Fs { .. } => panic!("expected S3"),
         }
+        Ok(())
     }
 
     #[test]
-    fn deserialize_logs_input_fs_config() {
+    fn deserialize_logs_input_fs_config() -> anyhow::Result<()> {
         const DIRECTORY: &str = "/var/logs";
 
         let logs_input_config_json = serde_json::json!({
@@ -431,14 +434,14 @@ mod tests {
         });
 
         let deserialized =
-            serde_json::from_str::<LogsInput>(logs_input_config_json.to_string().as_str())
-                .expect("failed to deserialize `LogsInput` from JSON");
+            serde_json::from_str::<LogsInput>(logs_input_config_json.to_string().as_str())?;
 
         match deserialized {
             LogsInput::Fs { config } => {
                 assert_eq!(config.directory, DIRECTORY);
             }
-            LogsInput::S3 { .. } => panic!("Expected Fs"),
+            LogsInput::S3 { .. } => panic!("expected Fs"),
         }
+        Ok(())
     }
 }
diff --git a/components/log-ingestor/src/ingestion_job_manager.rs b/components/log-ingestor/src/ingestion_job_manager.rs
index 07a8aba3ae..a3fe0ec520 100644
--- a/components/log-ingestor/src/ingestion_job_manager.rs
+++ b/components/log-ingestor/src/ingestion_job_manager.rs
@@ -25,22 +25,22 @@ use crate::{
 /// Errors for ingestion job manager operations.
 #[derive(thiserror::Error, Debug)]
 pub enum Error {
-    #[error("Log ingestor internal error: {0}")]
+    #[error("log ingestor internal error: {0}")]
     InternalError(#[from] anyhow::Error),
 
-    #[error("Ingestion job not found: {0}")]
+    #[error("ingestion job not found: {0}")]
     JobNotFound(IngestionJobId),
 
-    #[error("Prefix conflict: {0}")]
+    #[error("prefix conflict: {0}")]
     PrefixConflict(String),
 
-    #[error("Custom endpoint URL not supported: {0}")]
+    #[error("custom endpoint URL not supported: {0}")]
     CustomEndpointUrlNotSupported(String),
 
-    #[error("Invalid job config: {0}")]
+    #[error("invalid job config: {0}")]
     InvalidConfig(#[from] ConfigError),
 
-    #[error("A region code must be specified when using the default AWS endpoint")]
+    #[error("a region code must be specified when using the default AWS endpoint")]
     MissingRegionCode,
 }
 
diff --git a/components/log-ingestor/src/ingestion_job_manager/clp_ingestion.rs b/components/log-ingestor/src/ingestion_job_manager/clp_ingestion.rs
index 56a099eccb..60bd7ad2ed 100644
--- a/components/log-ingestor/src/ingestion_job_manager/clp_ingestion.rs
+++ b/components/log-ingestor/src/ingestion_job_manager/clp_ingestion.rs
@@ -195,8 +195,8 @@ impl ClpDbIngestionConnector {
             LogsInput::S3 { config } => config.aws_authentication,
             LogsInput::Fs { .. } => {
                 panic!(
-                    "Invalid CLP config: Unsupported logs input type. The current implementation \
-                     only supports S3 input."
+                    "invalid CLP config: unsupported logs input type; the current implementation \
+                     only supports S3 input"
                 );
             }
         };
@@ -371,7 +371,7 @@ impl ClpDbIngestionConnector {
                     },
                     compression_job_id,
                     num_object_metadata_submitted: usize::try_from(num_submitted)
-                        .expect("Number of files submitted is not `usize` compatible"),
+                        .expect("number of files submitted is not `usize` compatible"),
                 },
             )
             .collect();

From cce43e87d048d024489680e3ba3e76a0b852d5d9 Mon Sep 17 00:00:00 2001
From: Junhao Liao <junhao@junhao.ca>
Date: Wed, 29 Jul 2026 17:16:58 -0400
Subject: [PATCH 7/7] chore(chart): bump version to 0.4.1-dev.3

---
 tools/deployment/package-helm/Chart.yaml | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/tools/deployment/package-helm/Chart.yaml b/tools/deployment/package-helm/Chart.yaml
index 12f1fcac68..5e6fef4af4 100644
--- a/tools/deployment/package-helm/Chart.yaml
+++ b/tools/deployment/package-helm/Chart.yaml
@@ -1,6 +1,6 @@
 apiVersion: "v2"
 name: "clp"
-version: "0.4.1-dev.2"
+version: "0.4.1-dev.3"
 description: "A Helm chart for CLP's (Compressed Log Processor) package deployment"
 type: "application"
 appVersion: "0.13.1-dev"