refactor: Improve error handling and response validation in config fetching and log sending

Author: pigri

src/worker/config.rs

@@ -133,7 +133,8 @@ pub async fn fetch_config(
         .send()
         .await?;
 
-    match response.status() {
+    let status = response.status();
+    match status {
         StatusCode::OK => {
             // Check if response is gzipped by looking at Content-Encoding header first
             let content_encoding = response.headers()
@@ -176,15 +177,44 @@ pub async fn fetch_config(
                     .map_err(|e| format!("Response contains invalid UTF-8: {}", e))?
             };
 
-            let body: ConfigApiResponse = serde_json::from_str(&json_text)
-                .map_err(|e| format!("Failed to parse JSON response: {}", e))?;
+            // Check if response body is empty
+            let json_text = json_text.trim();
+            if json_text.is_empty() {
+                return Err("API returned empty response body".into());
+            }
+
+            let body: ConfigApiResponse = serde_json::from_str(json_text)
+                .map_err(|e| {
+                    let preview = if json_text.len() > 200 {
+                        format!("{}...", &json_text[..200])
+                    } else {
+                        json_text.to_string()
+                    };
+                    format!("Failed to parse JSON response: {}. Response preview: {}", e, preview)
+                })?;
             // Update global config snapshot
             set_global_config(body.config.clone());
             Ok(body)
         }
         StatusCode::BAD_REQUEST | StatusCode::NOT_FOUND | StatusCode::INTERNAL_SERVER_ERROR => {
-            let body: ErrorResponse = serde_json::from_str(&response.text().await?)?;
-            Err(format!("API Error: {}", body.error).into())
+            let response_text = response.text().await?;
+            let trimmed = response_text.trim();
+            let status_code = status.as_u16();
+            if trimmed.is_empty() {
+                return Err(format!("API returned empty response body with status {}", status_code).into());
+            }
+            match serde_json::from_str::<ErrorResponse>(trimmed) {
+                Ok(body) => Err(format!("API Error: {}", body.error).into()),
+                Err(e) => {
+                    let preview = if trimmed.len() > 200 {
+                        format!("{}...", &trimmed[..200])
+                    } else {
+                        trimmed.to_string()
+                    };
+                    Err(format!("API returned status {} but response is not valid JSON: {}. Response preview: {}",
+                        status_code, e, preview).into())
+                }
+            }
         }
 
         status => Err(format!(

src/worker/log.rs

@@ -8,6 +8,9 @@ use chrono::{DateTime, Utc};
 
 use crate::http_client;
 
+/// Maximum batch size allowed by the API server
+const API_MAX_BATCH_SIZE: usize = 1000;
+
 /// Configuration for sending access logs to arxignis server
 #[derive(Debug, Clone)]
 pub struct LogSenderConfig {
@@ -27,7 +30,7 @@ impl LogSenderConfig {
             enabled,
             base_url,
             api_key,
-            batch_size_limit: 5000,        // Default: 5000 logs per batch
+            batch_size_limit: 1000,        // Default: 1000 logs per batch (API limit)
             batch_size_bytes: 5 * 1024 * 1024, // Default: 5MB
             batch_timeout_secs: 10,        // Default: 10 seconds
             include_request_body: false,   // Default: disabled
@@ -206,6 +209,7 @@ pub fn send_event(event: UnifiedEvent) {
 }
 
 /// Send a batch of events to the /events endpoint
+/// Automatically splits large batches into chunks of API_MAX_BATCH_SIZE
 async fn send_event_batch(events: Vec<UnifiedEvent>) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
     if events.is_empty() {
         return Ok(());
@@ -232,27 +236,38 @@ async fn send_event_batch(events: Vec<UnifiedEvent>) -> Result<(), Box<dyn std::
         .map_err(|e| format!("Failed to get global HTTP client: {}", e))?;
 
     let url = format!("{}/events", config.base_url);
-    let json = serde_json::to_string(&events)?;
-
-    log::debug!("Sending {} events to {}", events.len(), url);
-
-    let response = client
-        .post(&url)
-        .header("Authorization", format!("Bearer {}", config.api_key))
-        .header("Content-Type", "application/json")
-        .body(json)
-        .send()
-        .await?;
-
-    if !response.status().is_success() {
-        let status = response.status();
-        let error_text = response.text().await.unwrap_or_else(|_| "Unknown error".to_string());
-        log::warn!("Failed to send event batch to /events endpoint: {} - {} (batch size: {})", status, error_text, events.len());
-        return Err(format!("HTTP {}: {}", status, error_text).into());
-    } else {
-        log::debug!("Successfully sent event batch to /events endpoint (batch size: {})", events.len());
+
+    // Split events into chunks of API_MAX_BATCH_SIZE to respect API limits
+    let chunks: Vec<_> = events.chunks(API_MAX_BATCH_SIZE).collect();
+    let total_events = events.len();
+
+    for (chunk_idx, chunk) in chunks.iter().enumerate() {
+        let json = serde_json::to_string(chunk)?;
+
+        log::debug!("Sending chunk {}/{} ({} events) to {}",
+            chunk_idx + 1, chunks.len(), chunk.len(), url);
+
+        let response = client
+            .post(&url)
+            .header("Authorization", format!("Bearer {}", config.api_key))
+            .header("Content-Type", "application/json")
+            .body(json)
+            .send()
+            .await?;
+
+        if !response.status().is_success() {
+            let status = response.status();
+            let error_text = response.text().await.unwrap_or_else(|_| "Unknown error".to_string());
+            log::warn!("Failed to send event batch chunk {}/{} to /events endpoint: {} - {} (chunk size: {}, total batch: {})",
+                chunk_idx + 1, chunks.len(), status, error_text, chunk.len(), total_events);
+            return Err(format!("HTTP {}: {}", status, error_text).into());
+        } else {
+            log::debug!("Successfully sent event batch chunk {}/{} to /events endpoint (chunk size: {})",
+                chunk_idx + 1, chunks.len(), chunk.len());
+        }
     }
 
+    log::debug!("Successfully sent all {} events in {} chunk(s) to /events endpoint", total_events, chunks.len());
     Ok(())
 }