fix(sources): Fix acknowledgment checkpoint race in Windows Event Log source

The background checkpoint saver task was updating checkpoints every 5 seconds
regardless of acknowledgment status, bypassing the end-to-end acknowledgment
system. This caused data loss when Vector was killed because events could be
checkpointed before the sink confirmed delivery.

Changes:
- Disable background checkpoint saver when acknowledgements are enabled
- Checkpoints are now managed solely by the Finalizer when acks are enabled
- Update changelog to document acknowledgment support and wildcard limitation

When acks are disabled, the background checkpoint saver continues to operate
as before for performance (checkpoint every 5 seconds). When acks are enabled,
checkpoints only update after sink confirmation, ensuring at-least-once delivery.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude <noreply@anthropic.com>

Author: tot19

changelog.d/windows_eventlog_source.feature.md

@@ -6,6 +6,7 @@ This Windows-specific source uses event-driven subscriptions to stream events in
 - Real-time event-driven subscriptions using EvtSubscribe API
 - XPath query filtering for selective event collection
 - Checkpoint persistence for reliable resumption after restarts (similar to journald)
+- End-to-end acknowledgment support for at-least-once delivery guarantees
 - Configurable rate limiting to prevent overwhelming downstream systems
 - Configurable field truncation for storage optimization (Winlogbeat-compatible)
 - Enhanced provider name extraction using proper XML parsing
@@ -18,3 +19,5 @@ New configuration options:
 - `events_per_second`: Rate limit event processing (0 = unlimited)
 - `max_event_data_length`: Truncate event data values (0 = no truncation, Winlogbeat default)
 - `data_dir`: Directory for checkpoint persistence
+
+Note: Wildcard channel patterns (e.g., `Microsoft-Windows-*`) are not supported. Specify exact channel names.

src/sources/windows_eventlog/mod.rs

@@ -178,7 +178,7 @@ impl WindowsEventLogSource {
         );
 
         let mut subscription =
-            EventLogSubscription::new(&self.config, Arc::clone(&checkpointer)).await?;
+            EventLogSubscription::new(&self.config, Arc::clone(&checkpointer), self.acknowledgements).await?;
         let parser = EventLogParser::new(&self.config);
 
         let events_received = register!(EventsReceived);

src/sources/windows_eventlog/subscription.rs

@@ -210,9 +210,17 @@ impl CallbackContext {
 
 impl EventLogSubscription {
     /// Create a new event-driven subscription using EvtSubscribe with callback
+    ///
+    /// # Arguments
+    /// * `config` - Configuration for the Windows Event Log source
+    /// * `checkpointer` - Shared checkpointer for persisting bookmarks
+    /// * `acknowledgements` - Whether end-to-end acknowledgements are enabled.
+    ///   When true, the background checkpoint saver is disabled because checkpoints
+    ///   are managed by the finalizer after sink acknowledgement.
     pub async fn new(
         config: &WindowsEventLogConfig,
         checkpointer: Arc<Checkpointer>,
+        acknowledgements: bool,
     ) -> Result<Self, WindowsEventLogError> {
         // Create rate limiter if configured
         let rate_limiter = if config.events_per_second > 0 {
@@ -290,75 +298,83 @@ impl EventLogSubscription {
 
             // Start background task to periodically save bookmarks (every 5 seconds)
             // This avoids blocking event processing with serialization/I/O
-            let checkpoint_saver_ctx = Arc::clone(&callback_context);
-            let mut shutdown_rx_clone = shutdown_rx.clone();
-            tokio::spawn(async move {
-                debug!(message = "Checkpoint saver background task started");
-                let mut interval = tokio::time::interval(tokio::time::Duration::from_secs(5));
-                loop {
-                    // Wait for either interval tick or shutdown signal
-                    tokio::select! {
-                        _ = interval.tick() => {}
-                        _ = shutdown_rx_clone.changed() => {
-                            debug!(message = "Checkpoint saver task received shutdown signal");
-                            break;
+            //
+            // IMPORTANT: Only run this when acknowledgements are DISABLED.
+            // When acks are enabled, checkpoints are managed by the Finalizer
+            // which only updates after the sink confirms delivery.
+            if !acknowledgements {
+                let checkpoint_saver_ctx = Arc::clone(&callback_context);
+                let mut shutdown_rx_clone = shutdown_rx.clone();
+                tokio::spawn(async move {
+                    debug!(message = "Checkpoint saver background task started (acks disabled)");
+                    let mut interval = tokio::time::interval(tokio::time::Duration::from_secs(5));
+                    loop {
+                        // Wait for either interval tick or shutdown signal
+                        tokio::select! {
+                            _ = interval.tick() => {}
+                            _ = shutdown_rx_clone.changed() => {
+                                debug!(message = "Checkpoint saver task received shutdown signal");
+                                break;
+                            }
                         }
-                    }
 
-                    // Step 1: Copy handles while holding lock briefly (handles are just integers)
-                    let bookmark_handles: Vec<(
-                        String,
-                        windows::Win32::System::EventLog::EVT_HANDLE,
-                    )> = {
-                        let bookmarks = match checkpoint_saver_ctx.bookmarks.read() {
-                            Ok(guard) => guard,
-                            Err(poisoned) => {
-                                warn!(message = "Bookmark lock poisoned, recovering");
-                                poisoned.into_inner()
-                            }
-                        };
-                        bookmarks
-                            .iter()
-                            .map(|(channel, bookmark)| (channel.clone(), bookmark.as_handle()))
-                            .collect()
-                    }; // Lock released IMMEDIATELY
-
-                    // Step 2: Serialize handles OUTSIDE the lock (slow Windows API calls)
-                    let bookmark_xmls: Vec<(String, String)> = bookmark_handles
-                        .into_iter()
-                        .filter_map(|(channel, handle)| {
-                            match BookmarkManager::serialize_handle(handle) {
-                                Ok(xml) if Self::is_valid_bookmark_xml(&xml) => {
-                                    Some((channel, xml))
+                        // Step 1: Copy handles while holding lock briefly (handles are just integers)
+                        let bookmark_handles: Vec<(
+                            String,
+                            windows::Win32::System::EventLog::EVT_HANDLE,
+                        )> = {
+                            let bookmarks = match checkpoint_saver_ctx.bookmarks.read() {
+                                Ok(guard) => guard,
+                                Err(poisoned) => {
+                                    warn!(message = "Bookmark lock poisoned, recovering");
+                                    poisoned.into_inner()
                                 }
-                                Ok(_) => None, // Empty or invalid = bookmark not yet updated with events
-                                Err(e) => {
-                                    emit!(WindowsEventLogBookmarkError {
-                                        channel: channel.clone(),
-                                        error: e.to_string(),
-                                    });
-                                    None
+                            };
+                            bookmarks
+                                .iter()
+                                .map(|(channel, bookmark)| (channel.clone(), bookmark.as_handle()))
+                                .collect()
+                        }; // Lock released IMMEDIATELY
+
+                        // Step 2: Serialize handles OUTSIDE the lock (slow Windows API calls)
+                        let bookmark_xmls: Vec<(String, String)> = bookmark_handles
+                            .into_iter()
+                            .filter_map(|(channel, handle)| {
+                                match BookmarkManager::serialize_handle(handle) {
+                                    Ok(xml) if Self::is_valid_bookmark_xml(&xml) => {
+                                        Some((channel, xml))
+                                    }
+                                    Ok(_) => None, // Empty or invalid = bookmark not yet updated with events
+                                    Err(e) => {
+                                        emit!(WindowsEventLogBookmarkError {
+                                            channel: channel.clone(),
+                                            error: e.to_string(),
+                                        });
+                                        None
+                                    }
                                 }
+                            })
+                            .collect();
+
+                        // Save all checkpoints in a single batched disk write (no locks held)
+                        if !bookmark_xmls.is_empty() {
+                            if let Err(e) = checkpoint_saver_ctx
+                                .checkpointer()
+                                .set_batch(bookmark_xmls)
+                                .await
+                            {
+                                error!(
+                                    message = "Failed to save bookmark checkpoints",
+                                    error = %e
+                                );
                             }
-                        })
-                        .collect();
-
-                    // Save all checkpoints in a single batched disk write (no locks held)
-                    if !bookmark_xmls.is_empty() {
-                        if let Err(e) = checkpoint_saver_ctx
-                            .checkpointer()
-                            .set_batch(bookmark_xmls)
-                            .await
-                        {
-                            error!(
-                                message = "Failed to save bookmark checkpoints",
-                                error = %e
-                            );
                         }
                     }
-                }
-                debug!(message = "Checkpoint saver task terminated");
-            });
+                    debug!(message = "Checkpoint saver task terminated");
+                });
+            } else {
+                debug!(message = "Background checkpoint saver disabled (acks enabled)");
+            }
 
             Ok(Self {
                 config,
@@ -1641,7 +1657,7 @@ mod tests {
     async fn test_not_supported_error() {
         let config = WindowsEventLogConfig::default();
         let (checkpointer, _temp_dir) = create_test_checkpointer().await;
-        let result = EventLogSubscription::new(&config, checkpointer).await;
+        let result = EventLogSubscription::new(&config, checkpointer, false).await;
 
         assert!(matches!(
             result,
@@ -1674,7 +1690,7 @@ mod tests {
 
         let (checkpointer, _temp_dir) = create_test_checkpointer().await;
 
-        let mut subscription = EventLogSubscription::new(&config, checkpointer)
+        let mut subscription = EventLogSubscription::new(&config, checkpointer, false)
             .await
             .expect("Subscription creation should succeed");
 
@@ -1824,7 +1840,7 @@ mod tests {
         let (checkpointer, _temp_dir) = create_test_checkpointer().await;
 
         // Create subscription - should succeed initially as EvtSubscribe doesn't validate query syntax
-        let mut subscription = EventLogSubscription::new(&config, checkpointer)
+        let mut subscription = EventLogSubscription::new(&config, checkpointer, false)
             .await
             .expect("Subscription creation should succeed");
 
@@ -1866,7 +1882,7 @@ mod tests {
 
         let (checkpointer, _temp_dir) = create_test_checkpointer().await;
 
-        let mut subscription = EventLogSubscription::new(&config, checkpointer)
+        let mut subscription = EventLogSubscription::new(&config, checkpointer, false)
             .await
             .expect("Subscription creation should succeed");
 
@@ -1899,7 +1915,7 @@ mod tests {
 
         let (checkpointer, _temp_dir) = create_test_checkpointer().await;
 
-        let mut subscription = EventLogSubscription::new(&config, checkpointer)
+        let mut subscription = EventLogSubscription::new(&config, checkpointer, false)
             .await
             .expect("Subscription creation should succeed");
 
@@ -1943,13 +1959,13 @@ mod tests {
         let (checkpointer3, _temp_dir3) = create_test_checkpointer().await;
 
         // Create all three subscriptions concurrently
-        let mut sub1 = EventLogSubscription::new(&config1, checkpointer1)
+        let mut sub1 = EventLogSubscription::new(&config1, checkpointer1, false)
             .await
             .expect("Subscription 1 (Application) should succeed");
-        let mut sub2 = EventLogSubscription::new(&config2, checkpointer2)
+        let mut sub2 = EventLogSubscription::new(&config2, checkpointer2, false)
             .await
             .expect("Subscription 2 (System) should succeed");
-        let mut sub3 = EventLogSubscription::new(&config3, checkpointer3)
+        let mut sub3 = EventLogSubscription::new(&config3, checkpointer3, false)
             .await
             .expect("Subscription 3 (Security) should succeed");
 
@@ -2026,7 +2042,7 @@ mod tests {
         // Record the time BEFORE creating the subscription
         let subscription_start_time = Utc::now();
 
-        let mut subscription = EventLogSubscription::new(&config, checkpointer)
+        let mut subscription = EventLogSubscription::new(&config, checkpointer, false)
             .await
             .expect("Subscription creation should succeed");
 
@@ -2085,7 +2101,7 @@ mod tests {
 
         let subscription_start_time = Utc::now();
 
-        let mut subscription = EventLogSubscription::new(&config, checkpointer)
+        let mut subscription = EventLogSubscription::new(&config, checkpointer, false)
             .await
             .expect("Subscription creation should succeed");
 
@@ -2145,7 +2161,7 @@ mod tests {
         config.read_existing_events = true; // Even with this true, checkpoint should take precedence
         config.event_timeout_ms = 500;
 
-        let mut subscription = EventLogSubscription::new(&config, checkpointer)
+        let mut subscription = EventLogSubscription::new(&config, checkpointer, false)
             .await
             .expect("Subscription creation should succeed");