fix: address PR nearai#403 code review feedback

jrevillard · claude · jrevillard · commit 021d81e6ad48 · 2026-03-04T13:32:39.000+01:00
- Remove .serena/ directory (committed by mistake)
- Remove is_webhook_message_processed (TOCTOU race condition)
  - Use atomic INSERT-first pattern in record_webhook_message_processed
  - Return bool to indicate new vs duplicate message
- Fix pending ACK leak on timeout (cleanup before removal)
- Fix metadata parsing efficiency (from_value vs to_string+from_str)
- Consolidate WhatsAppMetadata structs (Option&lt;String&gt; for all fields)

Co-Authored-By: Claude Opus 4.6 &lt;noreply@anthropic.com&gt;
diff --git a/.gitignore b/.gitignore
@@ -19,3 +19,4 @@ bench-results/
 # WASM build artifacts (loaded from disk, not bundled)
 *.wasm
 
+.serena/
diff --git a/src/agent/thread_ops.rs b/src/agent/thread_ops.rs
@@ -279,15 +279,20 @@ impl Agent {
             // For WhatsApp, metadata contains: phone_number_id, sender_phone, message_id, timestamp
             #[derive(serde::Deserialize)]
             struct WhatsAppMetadata {
-                message_id: String,
+                message_id: Option<String>,
             }
 
             let ack_key = if let Ok(meta) =
-                serde_json::from_str::<WhatsAppMetadata>(&message.metadata.to_string())
+                serde_json::from_value::<WhatsAppMetadata>(message.metadata.clone())
             {
-                format!("{}:{}", message.channel, meta.message_id)
+                if let Some(msg_id) = meta.message_id {
+                    format!("{}:{}", message.channel, msg_id)
+                } else {
+                    // Fallback to user_id if no message_id in metadata
+                    format!("{}:{}", message.channel, message.user_id)
+                }
             } else {
-                // Fallback to user_id if no message_id in metadata
+                // Fallback to user_id if metadata parsing fails
                 format!("{}:{}", message.channel, message.user_id)
             };
 
diff --git a/src/channels/wasm/router.rs b/src/channels/wasm/router.rs
@@ -819,41 +819,33 @@ async fn webhook_handler(
                         (format!("{}:{}", channel_name, msg.user_id), None)
                     };
 
-                    // Check for duplicate messages if database is available
-                    // Record immediately after check to prevent race conditions
+                    // Atomic deduplication: try to INSERT first
+                    // Returns true if inserted (new message), false if duplicate
+                    // This eliminates TOCTOU race condition between SELECT and INSERT
                     if let Some(ref db) = state.db
                         && let Some(msg_id) = &external_msg_id
                     {
-                        match db.is_webhook_message_processed(channel_name, msg_id).await {
-                            Ok(is_processed) => {
-                                if is_processed {
+                        match db
+                            .record_webhook_message_processed(channel_name, msg_id)
+                            .await
+                        {
+                            Ok(was_inserted) => {
+                                if !was_inserted {
                                     tracing::info!(
                                         channel = %channel_name,
                                         message_id = %msg_id,
                                         "Duplicate webhook message detected, skipping"
                                     );
                                     continue; // Skip this duplicate message
                                 }
-                                // Not a duplicate - record immediately to prevent races
-                                // This ensures concurrent webhooks for the same message will be caught
-                                if let Err(e) = db
-                                    .record_webhook_message_processed(channel_name, msg_id)
-                                    .await
-                                {
-                                    tracing::warn!(
-                                        channel = %channel_name,
-                                        message_id = %msg_id,
-                                        error = %e,
-                                        "Failed to record message as processed, dedup may not work on retry"
-                                    );
-                                }
+                                // New message - will process below
                             }
                             Err(e) => {
                                 tracing::warn!(
                                     channel = %channel_name,
                                     message_id = %msg_id,
                                     error = %e,
-                                    "Failed to check dedup, proceeding anyway"
+                                    "Failed to record message, proceeding anyway (fail open)"
                                 );
                                 // Continue processing on error (fail open)
                             }
@@ -894,6 +886,12 @@ async fn webhook_handler(
                 // Using join_all ensures we don't accumulate timeouts (3 messages ≠ 30s wait)
                 let ack_timeout = state.webhook_ack_timeout;
 
+                // Keep track of ack_keys for cleanup on timeout
+                let ack_keys: Vec<String> = ack_receivers
+                    .iter()
+                    .map(|(key, _, _)| key.clone())
+                    .collect();
+
                 let ack_futures: Vec<_> = ack_receivers
                     .into_iter()
                     .map(|(ack_key, ack_rx, _)| async move {
@@ -921,6 +919,10 @@ async fn webhook_handler(
                                 timeout_secs = ack_timeout.as_secs(),
                                 "Webhook ACK wait timed out, returning 500 to trigger retry"
                             );
+                            // Clean up pending ACKs that were registered but never signaled
+                            for ack_key in &ack_keys {
+                                state.router.pending_acks.write().await.remove(ack_key);
+                            }
                             // Return 500 so WhatsApp retries - deduplication will handle duplicates
                             return (
                                 StatusCode::INTERNAL_SERVER_ERROR,
diff --git a/src/channels/wasm/wrapper.rs b/src/channels/wasm/wrapper.rs
@@ -1276,11 +1276,9 @@ impl WasmChannel {
         let workspace_store = self.workspace_store.clone();
 
         // Pre-resolve host credentials for automatic injection (before spawn_blocking)
-        let host_credentials = resolve_channel_host_credentials(
-            &self.capabilities,
-            self.secrets_store.as_deref(),
-        )
-        .await;
+        let host_credentials =
+            resolve_channel_host_credentials(&self.capabilities, self.secrets_store.as_deref())
+                .await;
 
         // Prepare request data
         let method = method.to_string();
diff --git a/src/db/libsql/mod.rs b/src/db/libsql/mod.rs
@@ -464,46 +464,47 @@ mod tests {
     use crate::db::WebhookDedupStore;
 
     #[tokio::test]
-    async fn test_webhook_dedup_not_processed_initially() {
+    async fn test_webhook_dedup_first_record_returns_true() {
         // Use a temp file so connections share state (in-memory DBs are connection-local)
         let dir = tempfile::tempdir().unwrap();
-        let db_path = dir.path().join("test_dedup_initial.db");
+        let db_path = dir.path().join("test_dedup_first.db");
         let backend = LibSqlBackend::new_local(&db_path).await.unwrap();
         backend.run_migrations().await.unwrap();
 
-        // Message should not be marked as processed initially
-        let is_processed = backend
-            .is_webhook_message_processed("whatsapp", "wamid.test123")
+        // First record should return true (new message inserted)
+        let was_inserted = backend
+            .record_webhook_message_processed("whatsapp", "wamid.test123")
             .await
             .unwrap();
         assert!(
-            !is_processed,
-            "New message should not be marked as processed"
+            was_inserted,
+            "First record should return true (new message)"
         );
     }
 
     #[tokio::test]
-    async fn test_webhook_dedup_record_and_check() {
+    async fn test_webhook_dedup_duplicate_returns_false() {
         // Use a temp file so connections share state (in-memory DBs are connection-local)
         let dir = tempfile::tempdir().unwrap();
-        let db_path = dir.path().join("test_dedup_record.db");
+        let db_path = dir.path().join("test_dedup_duplicate.db");
         let backend = LibSqlBackend::new_local(&db_path).await.unwrap();
         backend.run_migrations().await.unwrap();
 
-        // Record message as processed
-        backend
-            .record_webhook_message_processed("whatsapp", "wamid.test456")
+        // First record returns true
+        let was_inserted = backend
+            .record_webhook_message_processed("whatsapp", "wamid.duplicate")
             .await
             .unwrap();
+        assert!(was_inserted, "First record should return true");
 
-        // Now it should be marked as processed
-        let is_processed = backend
-            .is_webhook_message_processed("whatsapp", "wamid.test456")
+        // Second record (duplicate) returns false
+        let was_inserted = backend
+            .record_webhook_message_processed("whatsapp", "wamid.duplicate")
             .await
             .unwrap();
         assert!(
-            is_processed,
-            "Recorded message should be marked as processed"
+            !was_inserted,
+            "Duplicate record should return false (already processed)"
         );
     }
 
@@ -515,53 +516,31 @@ mod tests {
         let backend = LibSqlBackend::new_local(&db_path).await.unwrap();
         backend.run_migrations().await.unwrap();
 
-        // Record message for whatsapp
-        backend
+        // Record message for whatsapp - should return true
+        let was_inserted = backend
             .record_webhook_message_processed("whatsapp", "msg789")
             .await
             .unwrap();
+        assert!(was_inserted, "First channel record should return true");
 
-        // Same message_id on different channel should NOT be processed
-        let is_processed = backend
-            .is_webhook_message_processed("telegram", "msg789")
+        // Same message_id on different channel should also return true (independent)
+        let was_inserted = backend
+            .record_webhook_message_processed("telegram", "msg789")
             .await
             .unwrap();
         assert!(
-            !is_processed,
+            was_inserted,
             "Different channel should have independent dedup state"
         );
 
-        // Original channel should be processed
-        let is_processed = backend
-            .is_webhook_message_processed("whatsapp", "msg789")
-            .await
-            .unwrap();
-        assert!(is_processed, "Original channel should be processed");
-    }
-
-    #[tokio::test]
-    async fn test_webhook_dedup_duplicate_record_is_idempotent() {
-        // Use a temp file so connections share state (in-memory DBs are connection-local)
-        let dir = tempfile::tempdir().unwrap();
-        let db_path = dir.path().join("test_dedup_idempotent.db");
-        let backend = LibSqlBackend::new_local(&db_path).await.unwrap();
-        backend.run_migrations().await.unwrap();
-
-        // Record same message twice (simulating webhook retry)
-        backend
-            .record_webhook_message_processed("whatsapp", "wamid.duplicate")
-            .await
-            .unwrap();
-        backend
-            .record_webhook_message_processed("whatsapp", "wamid.duplicate")
-            .await
-            .unwrap();
-
-        // Should still be marked as processed (no error, no duplicate row)
-        let is_processed = backend
-            .is_webhook_message_processed("whatsapp", "wamid.duplicate")
+        // Duplicate on original channel should return false
+        let was_inserted = backend
+            .record_webhook_message_processed("whatsapp", "msg789")
             .await
             .unwrap();
-        assert!(is_processed);
+        assert!(
+            !was_inserted,
+            "Duplicate on original channel should return false"
+        );
     }
 }
diff --git a/src/db/libsql/webhook_dedup.rs b/src/db/libsql/webhook_dedup.rs
@@ -10,45 +10,23 @@ use crate::error::DatabaseError;
 
 #[async_trait]
 impl WebhookDedupStore for LibSqlBackend {
-    async fn is_webhook_message_processed(
+    async fn record_webhook_message_processed(
         &self,
         channel: &str,
         external_message_id: &str,
     ) -> Result<bool, DatabaseError> {
         let conn = self.connect().await?;
-        let mut rows = conn
-            .query(
-                "SELECT 1 FROM webhook_message_dedup \
-                 WHERE channel = ?1 AND external_message_id = ?2",
-                params![channel, external_message_id],
+        let id = Uuid::new_v4().to_string();
+        let rows_affected = conn
+            .execute(
+                "INSERT INTO webhook_message_dedup (id, channel, external_message_id) \
+                 VALUES (?1, ?2, ?3) \
+                 ON CONFLICT(channel, external_message_id) DO NOTHING",
+                params![id, channel, external_message_id],
             )
             .await
             .map_err(|e| DatabaseError::Query(e.to_string()))?;
-
-        let exists = rows
-            .next()
-            .await
-            .map_err(|e| DatabaseError::Query(e.to_string()))?
-            .is_some();
-        Ok(exists)
-    }
-
-    async fn record_webhook_message_processed(
-        &self,
-        channel: &str,
-        external_message_id: &str,
-    ) -> Result<(), DatabaseError> {
-        let conn = self.connect().await?;
-        let id = Uuid::new_v4().to_string();
-        conn.execute(
-            "INSERT INTO webhook_message_dedup (id, channel, external_message_id) \
-             VALUES (?1, ?2, ?3) \
-             ON CONFLICT(channel, external_message_id) DO NOTHING",
-            params![id, channel, external_message_id],
-        )
-        .await
-        .map_err(|e| DatabaseError::Query(e.to_string()))?;
-        Ok(())
+        Ok(rows_affected > 0)
     }
 
     async fn cleanup_old_webhook_dedup_records(&self) -> Result<u64, DatabaseError> {
diff --git a/src/db/mod.rs b/src/db/mod.rs
@@ -340,24 +340,14 @@ pub trait SettingsStore: Send + Sync {
 /// when platforms like WhatsApp retry after a 500 response.
 #[async_trait]
 pub trait WebhookDedupStore: Send + Sync {
-    /// Check if a message has already been processed.
-    ///
-    /// Returns `true` if the message was already processed (duplicate),
-    /// `false` if this is a new message.
-    async fn is_webhook_message_processed(
-        &self,
-        channel: &str,
-        external_message_id: &str,
-    ) -> Result<bool, DatabaseError>;
-
-    /// Record that a message has been successfully processed.
-    ///
-    /// Should be called AFTER the message is fully processed (persisted to DB).
+    /// Try to record that a message is processed, atomically.
+    /// Returns `true` if this is a new message (was inserted), `false` if it was a duplicate.
+    /// Uses INSERT ... ON CONFLICT DO NOTHING pattern for atomic dedup with no race condition.
     async fn record_webhook_message_processed(
         &self,
         channel: &str,
         external_message_id: &str,
-    ) -> Result<(), DatabaseError>;
+    ) -> Result<bool, DatabaseError>;
 
     /// Clean up old dedup records (older than 7 days).
     ///
diff --git a/src/db/postgres.rs b/src/db/postgres.rs
@@ -673,30 +673,11 @@ impl WorkspaceStore for PgBackend {
 
 #[async_trait]
 impl WebhookDedupStore for PgBackend {
-    async fn is_webhook_message_processed(
-        &self,
-        channel: &str,
-        external_message_id: &str,
-    ) -> Result<bool, DatabaseError> {
-        let client = self.store.pool().get().await?;
-        let stmt = client
-            .prepare(
-                "SELECT 1 FROM webhook_message_dedup \
-                 WHERE channel = $1 AND external_message_id = $2",
-            )
-            .await?;
-
-        let rows = client
-            .query(&stmt, &[&channel, &external_message_id])
-            .await?;
-        Ok(!rows.is_empty())
-    }
-
     async fn record_webhook_message_processed(
         &self,
         channel: &str,
         external_message_id: &str,
-    ) -> Result<(), DatabaseError> {
+    ) -> Result<bool, DatabaseError> {
         let client = self.store.pool().get().await?;
         let stmt = client
             .prepare(
@@ -706,10 +687,10 @@ impl WebhookDedupStore for PgBackend {
             )
             .await?;
 
-        client
+        let rows_affected = client
             .execute(&stmt, &[&channel, &external_message_id])
             .await?;
-        Ok(())
+        Ok(rows_affected > 0)
     }
 
     async fn cleanup_old_webhook_dedup_records(&self) -> Result<u64, DatabaseError> {

Original file line number	Diff line number	Diff line change
`@@ -19,3 +19,4 @@ bench-results/`
`19`	`19`	`# WASM build artifacts (loaded from disk, not bundled)`
`20`	`20`	`*.wasm`
`21`	`21`
	`22`	`+.serena/`