more resilience

crates/weaver-index/src/clickhouse/client.rs

···
             .inserter(table)
             .with_max_rows(1000)
             .with_period_bias(0.1)
-            .with_period(Some(Duration::from_secs(1)))
-            .with_max_bytes(1_048_576)
+            .with_period(Some(Duration::from_secs(2)))
+            .with_max_bytes(1_048_576 * 2)
     }
 
     /// Query table sizes from system.parts

crates/weaver-index/src/clickhouse/resilient_inserter.rs

···
 impl Default for InserterConfig {
     fn default() -> Self {
         Self {
-            max_rows: 1000,
-            max_bytes: 1_048_576, // 1MB
-            period: Some(Duration::from_secs(1)),
+            max_rows: 10000,
+            max_bytes: 1_048_576 * 2, // 1MB
+            period: Some(Duration::from_secs(2)),
             period_bias: 0.1,
         }
     }
···
         self.pending.len()
     }
 
-    /// Handle a batch failure by retrying rows individually
+    /// Get time remaining until the next scheduled flush
+    pub fn time_left(&mut self) -> Option<std::time::Duration> {
+        self.inner.time_left()
+    }
+
+    /// Handle a batch failure by retrying rows
+    ///
+    /// Attempts to extract the failing row number from the error message.
+    /// If found, batches rows before/after the failure point for efficiency.
+    /// Falls back to individual retries if row number unavailable or sub-batches fail.
     async fn handle_batch_failure(
         &mut self,
         original_error: clickhouse::error::Error,
···
         // Create fresh inserter (old one is poisoned after error)
         self.inner = Self::create_inserter(&self.client, &self.config);
 
+        // Try to extract failing row number for smart retry
+        if let Some(failing_row) = extract_failing_row(&original_error) {
+            // Subtract 2 for safety margin (1-indexed to 0-indexed, plus buffer)
+            let safe_row = failing_row.saturating_sub(2);
+
+            if safe_row > 0 && safe_row < total {
+                debug!(
+                    failing_row,
+                    safe_row, total, "extracted failing row, attempting smart retry"
+                );
+                return self.smart_retry(rows, safe_row, &original_error).await;
+            }
+        }
+
+        // Fall back to individual retries
+        debug!(total, "no row number found, retrying individually");
+        self.retry_individually(rows).await
+    }
+
+    /// Smart retry: batch rows before failure, DLQ the bad row, batch rows after
+    async fn smart_retry(
+        &mut self,
+        rows: Vec<RawRecordInsert>,
+        failing_idx: usize,
+        original_error: &clickhouse::error::Error,
+    ) -> Result<Quantities, IndexError> {
+        let total = rows.len();
         let mut succeeded = 0u64;
         let mut failed = 0u64;
 
-        for row in rows {
-            match self.try_single_insert(&row).await {
+        // Try to batch insert rows before the failure point
+        if failing_idx > 0 {
+            let before = &rows[..failing_idx];
+            debug!(count = before.len(), "batch inserting rows before failure");
+
+            match self.batch_insert(before).await {
+                Ok(count) => {
+                    succeeded += count;
+                    debug!(count, "pre-failure batch succeeded");
+                }
+                Err(e) => {
+                    // Sub-batch failed, fall back to individual for this chunk
+                    warn!(error = ?e, "pre-failure batch failed, retrying individually");
+                    let (s, f) = self.retry_individually_slice(before).await?;
+                    succeeded += s;
+                    failed += f;
+                }
+            }
+        }
+
+        // Send the failing row (and a couple around it) to DLQ
+        let dlq_start = failing_idx;
+        let dlq_end = (failing_idx + 3).min(total); // failing row + 2 more for safety
+        for row in &rows[dlq_start..dlq_end] {
+            warn!(
+                did = %row.did,
+                collection = %row.collection,
+                rkey = %row.rkey,
+                seq = row.seq,
+                "sending suspected bad row to DLQ"
+            );
+            self.send_to_dlq(row, original_error).await?;
+            failed += 1;
+        }
+
+        // Try to batch insert rows after the failure point
+        if dlq_end < total {
+            let after = &rows[dlq_end..];
+            debug!(count = after.len(), "batch inserting rows after failure");
+
+            match self.batch_insert(after).await {
+                Ok(count) => {
+                    succeeded += count;
+                    debug!(count, "post-failure batch succeeded");
+                }
+                Err(e) => {
+                    // Sub-batch failed, fall back to individual for this chunk
+                    warn!(error = ?e, "post-failure batch failed, retrying individually");
+                    let (s, f) = self.retry_individually_slice(after).await?;
+                    succeeded += s;
+                    failed += f;
+                }
+            }
+        }
+
+        debug!(total, succeeded, failed, "smart retry complete");
+
+        Ok(Quantities {
+            rows: succeeded,
+            bytes: 0,
+            transactions: 0,
+        })
+    }
+
+    /// Batch insert a slice of rows using a fresh one-shot inserter
+    async fn batch_insert(
+        &mut self,
+        rows: &[RawRecordInsert],
+    ) -> Result<u64, clickhouse::error::Error> {
+        batch_insert_rows(&self.client, rows).await
+    }
+
+    /// Retry a vec of rows individually, returning (succeeded, failed) counts
+    async fn retry_individually(
+        &mut self,
+        rows: Vec<RawRecordInsert>,
+    ) -> Result<Quantities, IndexError> {
+        let (succeeded, failed) = self.retry_individually_slice(&rows).await?;
+
+        if failed > 0 {
+            warn!(
+                succeeded,
+                failed, "individual retry had failures sent to DLQ"
+            );
+        }
+
+        Ok(Quantities {
+            rows: succeeded,
+            bytes: 0,
+            transactions: 0,
+        })
+    }
+
+    /// Retry a slice of rows individually, returning (succeeded, failed) counts
+    async fn retry_individually_slice(
+        &mut self,
+        rows: &[RawRecordInsert],
+    ) -> Result<(u64, u64), IndexError> {
+        let total = rows.len();
+        let mut succeeded = 0u64;
+        let mut failed = 0u64;
+
+        let client = self.client.clone();
+
+        for (i, row) in rows.iter().enumerate() {
+            debug!(i, total, did = %row.did, "retrying row individually");
+            match try_single_insert(&client, row).await {
                 Ok(()) => {
                     succeeded += 1;
+                    debug!(i, "row succeeded");
                 }
                 Err(e) => {
                     failed += 1;
···
                         error = ?e,
                         "row insert failed, sending to DLQ"
                     );
-                    self.send_to_dlq(&row, &e).await?;
+                    debug!(i, "sending to DLQ");
+                    self.send_to_dlq(row, &e).await?;
+                    debug!(i, "DLQ write complete");
                 }
             }
         }
 
-        debug!(total, succeeded, failed, "batch failure recovery complete");
-
-        Ok(Quantities {
-            rows: succeeded,
-            bytes: 0,
-            transactions: 0,
-        })
-    }
-
-    /// Try to insert a single row using a fresh one-shot inserter
-    async fn try_single_insert(
-        &self,
-        row: &RawRecordInsert,
-    ) -> Result<(), clickhouse::error::Error> {
-        let mut inserter: Inserter<RawRecordInsert> =
-            self.client.inserter(Tables::RAW_RECORDS).with_max_rows(1);
-
-        inserter.write(row).await?;
-        inserter.end().await?;
-        Ok(())
+        debug!(total, succeeded, failed, "individual retry complete");
+        Ok((succeeded, failed))
     }
 
     /// Send a failed row to the dead-letter queue
···
         let raw_data = serde_json::to_string(row)
             .unwrap_or_else(|e| format!("{{\"serialization_error\": \"{}\"}}", e));
 
+        self.write_raw_to_dlq(row.operation.clone(), raw_data, error.to_string(), row.seq)
+            .await
+    }
+
+    /// Write a pre-insert failure directly to the DLQ
+    ///
+    /// Use this for failures that happen before we even have a valid RawRecordInsert,
+    /// like JSON serialization errors.
+    pub async fn write_raw_to_dlq(
+        &mut self,
+        event_type: SmolStr,
+        raw_data: String,
+        error_message: String,
+        seq: u64,
+    ) -> Result<(), IndexError> {
         let dlq_row = RawEventDlq {
-            event_type: row.operation.clone(),
+            event_type,
             raw_data: raw_data.to_smolstr(),
-            error_message: error.to_smolstr(),
-            seq: row.seq,
+            error_message: error_message.to_smolstr(),
+            seq,
         };
 
         self.dlq
···
     }
 }
 
+/// Try to insert a single row using a fresh one-shot inserter
+///
+/// Free function to avoid &self borrow across await points (Sync issues)
+async fn try_single_insert(
+    client: &clickhouse::Client,
+    row: &RawRecordInsert,
+) -> Result<(), clickhouse::error::Error> {
+    let mut inserter: Inserter<RawRecordInsert> =
+        client.inserter(Tables::RAW_RECORDS).with_max_rows(1);
+
+    inserter.write(row).await?;
+    inserter.force_commit().await?;
+    inserter.end().await?;
+    Ok(())
+}
+
+/// Batch insert rows using a fresh inserter
+///
+/// Free function to avoid &self borrow across await points (Sync issues)
+async fn batch_insert_rows(
+    client: &clickhouse::Client,
+    rows: &[RawRecordInsert],
+) -> Result<u64, clickhouse::error::Error> {
+    let mut inserter: Inserter<RawRecordInsert> = client
+        .inserter(Tables::RAW_RECORDS)
+        .with_max_rows(rows.len() as u64);
+
+    for row in rows {
+        inserter.write(row).await?;
+    }
+    inserter.end().await?;
+    Ok(rows.len() as u64)
+}
+
+/// Extract the failing row number from a ClickHouse error message
+///
+/// Looks for patterns like "(at row 791)" in the error text.
+/// Returns 1-indexed row number if found.
+fn extract_failing_row(error: &clickhouse::error::Error) -> Option<usize> {
+    let msg = error.to_string();
+    // Look for "(at row N)"
+    if let Some(start) = msg.find("(at row ") {
+        let rest = &msg[start + 8..];
+        if let Some(end) = rest.find(')') {
+            return rest[..end].parse().ok();
+        }
+    }
+    None
+}
+
 #[cfg(test)]
 mod tests {
-    // TODO: Add tests with mock clickhouse client
+    use super::*;
+
+    #[test]
+    fn test_extract_failing_row() {
+        // Simulate the error message format from ClickHouse
+        let msg = "Code: 117. DB::Exception: Cannot parse JSON object here: : (at row 791)\n: While executing BinaryRowInputFormat.";
+
+        // We can't easily construct a clickhouse::error::Error, but we can test the parsing logic
+        assert!(msg.contains("(at row "));
+        let start = msg.find("(at row ").unwrap();
+        let rest = &msg[start + 8..];
+        let end = rest.find(')').unwrap();
+        let row: usize = rest[..end].parse().unwrap();
+        assert_eq!(row, 791);
+    }
 }

crates/weaver-index/src/indexer.rs

···
 use chrono::DateTime;
 
 use crate::clickhouse::{
-    AccountRevState, Client, FirehoseCursor, RawAccountEvent, RawIdentityEvent, RawRecordInsert,
+    AccountRevState, Client, FirehoseCursor, InserterConfig, RawAccountEvent, RawIdentityEvent,
+    RawRecordInsert, ResilientRecordInserter,
 };
 use crate::config::IndexerConfig;
 use crate::config::TapConfig;
 use crate::error::{ClickHouseError, IndexError, Result};
 use crate::firehose::{
-    Account, Commit, ExtractedRecord, FirehoseConsumer, Identity, MessageStream,
-    SubscribeReposMessage, extract_records,
+    Account, ExtractedRecord, FirehoseConsumer, Identity, MessageStream, SubscribeReposMessage,
+    extract_records,
 };
 use crate::tap::{TapConfig as TapConsumerConfig, TapConsumer, TapEvent};
 
···
         let mut stream: MessageStream = self.consumer.connect().await?;
 
         // Inserters handle batching internally based on config
-        let mut records = self.client.inserter::<RawRecordInsert>("raw_records");
+        // Use resilient inserter for records since that's where untrusted JSON enters
+        let mut records =
+            ResilientRecordInserter::new(self.client.inner().clone(), InserterConfig::default());
         let mut identities = self
             .client
             .inserter::<RawIdentityEvent>("raw_identity_events");
···
             let accounts_time = accounts.time_left().unwrap_or(Duration::from_secs(10));
             let time_left = records_time.min(identities_time).min(accounts_time);
 
-            let result =
-                match tokio::time::timeout(time_left, stream.next()).await {
-                    Ok(Some(result)) => result,
-                    Ok(None) => {
-                        // Stream ended
-                        break;
-                    }
-                    Err(_) => {
-                        // Timeout - flush inserters to keep INSERT alive
-                        debug!("flush timeout, committing inserters");
-                        records.commit().await.map_err(|e| {
-                            crate::error::ClickHouseError::Query {
-                                message: "periodic records commit failed".into(),
-                                source: e,
-                            }
+            let result = match tokio::time::timeout(time_left, stream.next()).await {
+                Ok(Some(result)) => result,
+                Ok(None) => {
+                    // Stream ended
+                    break;
+                }
+                Err(_) => {
+                    // Timeout - flush inserters to keep INSERT alive
+                    debug!("flush timeout, committing inserters");
+                    records.commit().await?;
+                    identities.commit().await.map_err(|e| {
+                        crate::error::ClickHouseError::Query {
+                            message: "periodic identities commit failed".into(),
+                            source: e,
+                        }
+                    })?;
+                    accounts
+                        .commit()
+                        .await
+                        .map_err(|e| crate::error::ClickHouseError::Query {
+                            message: "periodic accounts commit failed".into(),
+                            source: e,
                         })?;
-                        identities.commit().await.map_err(|e| {
-                            crate::error::ClickHouseError::Query {
-                                message: "periodic identities commit failed".into(),
-                                source: e,
-                            }
-                        })?;
-                        accounts.commit().await.map_err(|e| {
-                            crate::error::ClickHouseError::Query {
-                                message: "periodic accounts commit failed".into(),
-                                source: e,
-                            }
-                        })?;
-                        continue;
-                    }
-                };
+                    continue;
+                }
+            };
 
             let msg = match result {
                 Ok(msg) => msg,
···
 
             match msg {
                 SubscribeReposMessage::Commit(commit) => {
-                    if self
-                        .process_commit(&commit, &mut records, &mut skipped)
-                        .await?
-                    {
-                        processed += 1;
+                    let did = commit.repo.as_ref();
+                    let rev = commit.rev.as_ref();
+
+                    // Dedup check
+                    if !self.rev_cache.should_process(did, rev) {
+                        skipped += 1;
+                        continue;
+                    }
+
+                    // Extract and write records
+                    for record in extract_records(&commit).await? {
+                        // Collection filter - skip early before JSON conversion
+                        if !self.config.collections.matches(&record.collection) {
+                            continue;
+                        }
+
+                        let json = record.to_json()?.unwrap_or_else(|| "{}".to_string());
+
+                        // Fire and forget delete handling
+                        if record.operation == "delete" {
+                            let client = self.client.clone();
+                            let record_clone = record.clone();
+                            tokio::spawn(async move {
+                                if let Err(e) = handle_delete(&client, record_clone).await {
+                                    warn!(error = ?e, "delete handling failed");
+                                }
+                            });
+                        }
+
+                        records
+                            .write(RawRecordInsert {
+                                did: record.did.clone(),
+                                collection: record.collection.clone(),
+                                rkey: record.rkey.clone(),
+                                cid: record.cid.clone(),
+                                rev: record.rev.clone(),
+                                record: json.to_smolstr(),
+                                operation: record.operation.clone(),
+                                seq: record.seq as u64,
+                                event_time: record.event_time,
+                                is_live: true,
+                            })
+                            .await?;
                     }
+
+                    // Update rev cache
+                    self.rev_cache.update(
+                        &SmolStr::new(did),
+                        &SmolStr::new(rev),
+                        &commit.commit.0.to_smolstr(),
+                    );
+
+                    processed += 1;
                 }
                 SubscribeReposMessage::Identity(identity) => {
                     write_identity(&identity, &mut identities).await?;
···
             }
 
             // commit() flushes if internal thresholds met, otherwise no-op
-            records
-                .commit()
-                .await
-                .map_err(|e| crate::error::ClickHouseError::Query {
-                    message: "commit failed".into(),
-                    source: e,
-                })?;
+            records.commit().await?;
 
             // Periodic stats and cursor save (every 10s)
             if last_stats.elapsed() >= Duration::from_secs(10) {
···
         }
 
         // Final flush
-        records
-            .end()
-            .await
-            .map_err(|e| crate::error::ClickHouseError::Query {
-                message: "final flush failed".into(),
-                source: e,
-            })?;
+        records.end().await?;
         identities
             .end()
             .await
···
         info!(last_seq, "firehose stream ended");
         Ok(())
     }
-
-    async fn process_commit(
-        &self,
-        commit: &Commit<'_>,
-        inserter: &mut clickhouse::inserter::Inserter<RawRecordInsert>,
-        skipped: &mut u64,
-    ) -> Result<bool> {
-        let did = commit.repo.as_ref();
-        let rev = commit.rev.as_ref();
-
-        // Dedup check
-        if !self.rev_cache.should_process(did, rev) {
-            *skipped += 1;
-            return Ok(false);
-        }
-
-        // Extract and write records
-        for record in extract_records(commit).await? {
-            // Collection filter - skip early before JSON conversion
-            if !self.config.collections.matches(&record.collection) {
-                continue;
-            }
-
-            let json = record.to_json()?.unwrap_or_else(|| "{}".to_string());
-
-            // Fire and forget delete handling
-            if record.operation == "delete" {
-                let client = self.client.clone();
-                let record_clone = record.clone();
-                tokio::spawn(async move {
-                    if let Err(e) = handle_delete(&client, record_clone).await {
-                        warn!(error = ?e, "delete handling failed");
-                    }
-                });
-            }
-
-            inserter
-                .write(&RawRecordInsert {
-                    did: record.did.clone(),
-                    collection: record.collection.clone(),
-                    rkey: record.rkey.clone(),
-                    cid: record.cid.clone(),
-                    rev: record.rev.clone(),
-                    record: json.to_smolstr(),
-                    operation: record.operation.clone(),
-                    seq: record.seq as u64,
-                    event_time: record.event_time,
-                    is_live: true,
-                })
-                .await
-                .map_err(|e| crate::error::ClickHouseError::Query {
-                    message: "write failed".into(),
-                    source: e,
-                })?;
-        }
-
-        // Update rev cache
-        self.rev_cache.update(
-            &SmolStr::new(did),
-            &SmolStr::new(rev),
-            &commit.commit.0.to_smolstr(),
-        );
-
-        Ok(true)
-    }
 }
 
 async fn write_identity(
···
 
         let (mut events, ack_tx) = consumer.connect().await?;
 
-        let mut records = self.client.inserter::<RawRecordInsert>("raw_records");
+        // Use resilient inserter for records since that's where untrusted JSON enters
+        let mut records =
+            ResilientRecordInserter::new(self.client.inner().clone(), InserterConfig::default());
         let mut identities = self
             .client
             .inserter::<RawIdentityEvent>("raw_identity_events");
···
                 Err(_) => {
                     // Timeout - flush inserters to keep INSERT alive
                     trace!("flush timeout, committing inserters");
-                    records.commit().await.map_err(|e| ClickHouseError::Query {
-                        message: "periodic records commit failed".into(),
-                        source: e,
-                    })?;
+                    records.commit().await?;
                     identities
                         .commit()
                         .await
···
                         continue;
                     }
 
-                    let json = record
-                        .record
-                        .as_ref()
-                        .map(|v| serde_json::to_string(v).unwrap_or_default())
-                        .unwrap_or_default();
+                    let json = match &record.record {
+                        Some(v) => match serde_json::to_string(v) {
+                            Ok(s) => s,
+                            Err(e) => {
+                                warn!(
+                                    did = %record.did,
+                                    collection = %record.collection,
+                                    rkey = %record.rkey,
+                                    error = ?e,
+                                    "failed to serialize record, sending to DLQ"
+                                );
+                                let raw_data = format!(
+                                    r#"{{"did":"{}","collection":"{}","rkey":"{}","cid":"{}","error":"serialization_failed"}}"#,
+                                    record.did, record.collection, record.rkey, record.cid
+                                );
+                                records
+                                    .write_raw_to_dlq(
+                                        record.action.as_str().to_smolstr(),
+                                        raw_data,
+                                        e.to_string(),
+                                        event_id,
+                                    )
+                                    .await?;
+                                let _ = ack_tx.send(event_id).await;
+                                continue;
+                            }
+                        },
+                        None => "{}".to_string(),
+                    };
 
                     debug!(
                         op = record.action.as_str(),
···
                     );
 
                     records
-                        .write(&RawRecordInsert {
+                        .write(RawRecordInsert {
                             did: record.did.clone(),
                             collection: record.collection.clone(),
                             rkey: record.rkey.clone(),
···
                             event_time: Utc::now(),
                             is_live: record.live,
                         })
-                        .await
-                        .map_err(|e| ClickHouseError::Query {
-                            message: "record write failed".into(),
-                            source: e,
-                        })?;
-                    records.commit().await.map_err(|e| ClickHouseError::Query {
-                        message: format!("record commit failed for id {}:\n{}", event_id, json),
-                        source: e,
-                    })?;
+                        .await?;
+                    records.commit().await?;
 
                     processed += 1;
                 }
···
         }
 
         // Final flush
-        records.end().await.map_err(|e| ClickHouseError::Query {
-            message: "final records flush failed".into(),
-            source: e,
-        })?;
+        records.end().await?;
         identities.end().await.map_err(|e| ClickHouseError::Query {
             message: "final identities flush failed".into(),
             source: e,

docker-compose.yml

···
     ports:
       - "3000:3000"
     environment:
-      RUST_LOG: debug,weaver_index=debug
+      RUST_LOG: debug,weaver_index=debug,hyper_util::client::legacy::pool=info
       # ClickHouse connection (set these for your cloud/homelab instance)
       CLICKHOUSE_URL: ${CLICKHOUSE_URL}
       CLICKHOUSE_DATABASE: ${CLICKHOUSE_DATABASE:-weaver}