diff --git a/.github/workflows/integration_windows.yml b/.github/workflows/integration_windows.yml index 39289d786fd24..218b659691ff7 100644 --- a/.github/workflows/integration_windows.yml +++ b/.github/workflows/integration_windows.yml @@ -32,7 +32,7 @@ jobs: run-test-integration-windows: needs: changes if: needs.changes.outputs.windows == 'true' - runs-on: windows-2025-8core + runs-on: windows-latest timeout-minutes: 60 steps: - name: Checkout branch diff --git a/.github/workflows/unit_windows.yml b/.github/workflows/unit_windows.yml index a4553b4a0b751..112ad148476ec 100644 --- a/.github/workflows/unit_windows.yml +++ b/.github/workflows/unit_windows.yml @@ -7,13 +7,22 @@ on: description: "Git ref to checkout" required: false type: string + workflow_dispatch: + pull_request: + paths: + - "src/sources/windows_event_log/**" + - "src/internal_events/windows_event_log.rs" + - "src/internal_events/windows.rs" + - "src/vector_windows.rs" + - "src/service.rs" + - ".github/workflows/unit_windows.yml" permissions: contents: read jobs: test-windows: - runs-on: windows-2025-8core + runs-on: windows-latest timeout-minutes: 60 steps: - name: Checkout branch diff --git a/Makefile b/Makefile index 34da1302487d9..bf7eb2264f1b4 100644 --- a/Makefile +++ b/Makefile @@ -392,7 +392,7 @@ test-integration: test-integration-datadog-traces test-integration-shutdown .PHONY: test-integration-windows-event-log test-integration-windows-event-log: ## Runs Windows Event Log integration tests (Windows only) ifeq ($(OS),Windows_NT) - ${MAYBE_ENVIRONMENT_EXEC} cargo test -p vector --no-default-features --features sources-windows_event_log-integration-tests windows_event_log::integration_tests + ${MAYBE_ENVIRONMENT_EXEC} cargo test -p vector --no-default-features --features sources-windows_event_log-integration-tests windows_event_log::integration_tests -- --test-threads=1 else @echo "Skipping windows-event-log integration tests (Windows only)" endif diff --git a/changelog.d/25194_windows_event_log_lost_wakeup.fix.md b/changelog.d/25194_windows_event_log_lost_wakeup.fix.md new file mode 100644 index 0000000000000..acf560f54635d --- /dev/null +++ b/changelog.d/25194_windows_event_log_lost_wakeup.fix.md @@ -0,0 +1,3 @@ +The `windows_event_log` source no longer freezes after periods of inactivity. A race between the signal-handle reset and incoming event notifications could cause new events to be silently dropped from the wakeup path; the signal is now reset before draining so notifications arriving mid-drain are preserved. + +authors: tot19 diff --git a/src/sources/windows_event_log/subscription.rs b/src/sources/windows_event_log/subscription.rs index dc92713f691d9..8a83351ef146a 100644 --- a/src/sources/windows_event_log/subscription.rs +++ b/src/sources/windows_event_log/subscription.rs @@ -18,9 +18,9 @@ use windows::Win32::System::EventLog::{ EvtSubscribeStartAfterBookmark, EvtSubscribeStartAtOldestRecord, EvtSubscribeStrict, EvtSubscribeToFutureEvents, }; -#[cfg(test)] -use windows::Win32::System::Threading::SetEvent; -use windows::Win32::System::Threading::{CreateEventW, ResetEvent, WaitForMultipleObjects}; +use windows::Win32::System::Threading::{ + CreateEventW, ResetEvent, SetEvent, WaitForMultipleObjects, +}; use windows::core::HSTRING; use super::{ @@ -30,6 +30,15 @@ use super::{ use crate::internal_events::WindowsEventLogBookmarkError; +/// Test-only hook called inside the `pull_events` drain loop after each +/// `EvtNext` invocation. Used by the lost-wakeup regression test +/// (see `test_pull_events_preserves_setevent_during_drain`) to race a +/// `SetEvent` against the drain without relying on thread-timing. +/// No-op and zero-cost in non-test builds. +#[cfg(test)] +static DRAIN_STEP_HOOK: std::sync::Mutex>> = + std::sync::Mutex::new(None); + /// Maximum number of entries in the EvtFormatMessage result cache. pub const FORMAT_CACHE_CAPACITY: usize = 10_000; /// Maximum number of cached publisher metadata handles. @@ -479,9 +488,25 @@ impl EventLogSubscription { let mut bookmark_failed = false; let mut channel_count = 0usize; - // Drain loop: keep calling EvtNext until ERROR_NO_MORE_ITEMS or channel budget. - // Only reset the signal once the channel is fully drained; if we hit the - // budget limit the signal stays set so WaitForMultipleObjects returns immediately. + // Reset the signal BEFORE draining to avoid a lost-wakeup race + // (see vectordotdev/vector#25194). The Windows Event Log service + // signals this manual-reset event via SetEvent each time a new + // matching event is recorded; SetEvent on an already-signaled + // event is a no-op, so if we reset AFTER draining, any signal + // that arrives between our last EvtNext and ResetEvent is lost + // — the subscription then hangs until the next event arrives. + // Resetting first means any signal raised during the drain is + // preserved, causing the next WaitForMultipleObjects to return + // immediately. + // + // If we exit the drain loop early (channel budget exhausted or + // bookmark update failed mid-batch), we re-SetEvent at the end + // of this iteration so the next pull_events call revisits this + // channel without waiting for a fresh OS signal. + unsafe { + let _ = ResetEvent(channel_sub.signal_event); + } + 'drain: loop { if channel_count >= channel_limit { break; @@ -501,6 +526,17 @@ impl EventLogSubscription { ) }; + // Test-only hook: lets the lost-wakeup regression test race + // a SetEvent against the drain without thread-timing. No-op + // and zero-cost in non-test builds. + #[cfg(test)] + { + let hook = DRAIN_STEP_HOOK.lock().unwrap().clone(); + if let Some(h) = hook { + h(channel_sub.signal_event); + } + } + if let Err(err) = result { let code = (err.code().0 as u32) & 0xFFFF; if code == ERROR_NO_MORE_ITEMS { @@ -542,6 +578,17 @@ impl EventLogSubscription { } } } + // Re-arm the signal before returning. We reset it pre-drain + // but are bailing out without confirming the drain completed, + // so if events were left un-drained the next pull_events must + // still revisit this channel without waiting for a fresh OS + // signal. This mirrors the `else` branch below that handles + // budget-exhaustion and bookmark-failure early breaks, and + // avoids the same lost-wakeup symptom (vectordotdev/vector#25194) + // on transient EvtNext failures. + unsafe { + let _ = SetEvent(channel_sub.signal_event); + } return Err(WindowsEventLogError::PullEventsError { channel: channel_sub.channel.clone(), source: err, @@ -697,15 +744,20 @@ impl EventLogSubscription { } if channel_drained && !bookmark_failed { - unsafe { - let _ = ResetEvent(channel_sub.signal_event); - } - // Update channel record count gauge for lag detection. super::render::update_channel_records( &channel_sub.channel, &channel_sub.channel_records_gauge, ); + } else { + // Drain exited early (budget exhausted or bookmark_failed + // mid-batch). Re-arm the signal so the next pull_events + // revisits this channel immediately without waiting for a + // fresh OS notification. Pairs with the pre-drain ResetEvent + // above. + unsafe { + let _ = SetEvent(channel_sub.signal_event); + } } } @@ -816,6 +868,15 @@ impl EventLogSubscription { self.shutdown_event.0 } + /// Test-only accessor for the first channel's signal event handle. Used + /// by the lost-wakeup regression test to scope its drain-loop hook to + /// exactly this subscription, so it does not fire on concurrent + /// `pull_events` calls from other tests in the same process. + #[cfg(test)] + pub(super) fn first_channel_signal_raw(&self) -> isize { + self.channels[0].signal_event.0 as isize + } + /// Returns a reference to the rate limiter, if configured. pub const fn rate_limiter( &self, @@ -1272,4 +1333,121 @@ mod tests { // that the subscription is functional. let _events = subscription.pull_events(100).unwrap_or_default(); } + + /// Regression test for vectordotdev/vector#25194. + /// + /// The Windows Event Log service signals the pull-mode wait handle via + /// `SetEvent` each time a new matching event is recorded. Because the + /// handle is manual-reset, `SetEvent` on an already-signaled handle is + /// a no-op. If `pull_events` resets the signal *after* draining events + /// via `EvtNext`, any signal that fires between the last `EvtNext` and + /// the `ResetEvent` call is silently lost — the subscription then + /// permanently hangs until a subsequent event arrives. + /// + /// The fix is to reset the signal *before* the drain loop, so signals + /// raised during the drain are preserved and the next wait returns + /// immediately. + /// + /// This test pins that invariant by driving the real `pull_events` + /// against a real `EvtSubscribe` handle. It installs a + /// `DRAIN_STEP_HOOK` that runs inside the drain loop after each + /// `EvtNext` and fires `SetEvent` on the subscription's signal + /// handle — simulating the OS signaling a new event arrival during + /// the drain window. After `pull_events` returns, the signal must + /// still be set (the subsequent `wait_for_events_blocking` must + /// return `EventsAvailable`, not `Timeout`). Under the old + /// post-drain `ResetEvent` order, the hook's `SetEvent` would be + /// clobbered by the reset and the wait would time out — which is + /// exactly what #25194 reports. + #[tokio::test] + async fn test_pull_events_preserves_setevent_during_drain() { + use std::sync::Arc as StdArc; + + let mut config = WindowsEventLogConfig::default(); + config.channels = vec!["Application".to_string()]; + config.read_existing_events = true; + config.event_timeout_ms = 1000; + + let (checkpointer, _temp_dir) = create_test_checkpointer().await; + + let mut subscription = EventLogSubscription::new(&config, checkpointer, false) + .await + .expect("Subscription creation should succeed"); + + // Capture THIS subscription's signal handle so the hook can scope + // itself to this test. DRAIN_STEP_HOOK is a process-global, and + // cargo runs tests in parallel by default; without handle-keying, + // a concurrent test's pull_events could trigger our one-shot + // hook first, flip `fired`, and SetEvent on the wrong handle. + let target_signal_raw = subscription.first_channel_signal_raw(); + + // Install the drain-loop hook: every EvtNext call inside + // pull_events fires SetEvent on the subscription's signal + // handle. This simulates the OS signaling a fresh event + // mid-drain, which is exactly the race window #25194 exposes. + // The hook only needs to fire once to prove the invariant; we + // use an AtomicBool to keep it deterministic. The hook is keyed + // to `target_signal_raw` so concurrent pull_events calls from + // other tests no-op here. + let fired = StdArc::new(std::sync::atomic::AtomicBool::new(false)); + { + let fired = StdArc::clone(&fired); + let hook: StdArc = StdArc::new(move |signal: HANDLE| { + if signal.0 as isize != target_signal_raw { + return; + } + if !fired.swap(true, std::sync::atomic::Ordering::SeqCst) { + unsafe { + let _ = SetEvent(signal); + } + } + }); + *DRAIN_STEP_HOOK.lock().unwrap() = Some(hook); + } + + // Drop-guard: clear the hook even if the test panics, so it + // doesn't contaminate other tests in the same process. + struct HookGuard; + impl Drop for HookGuard { + fn drop(&mut self) { + *DRAIN_STEP_HOOK.lock().unwrap() = None; + } + } + let _guard = HookGuard; + + // Drive pull_events with a very large budget so the drain + // exits via ERROR_NO_MORE_ITEMS (channel_drained = true), + // which is the path that ran the post-drain ResetEvent in the + // old buggy code. Exiting via budget exhaustion would skip + // that reset and cause this test to false-pass against the + // pre-fix code. + let _ = subscription.pull_events(usize::MAX).unwrap_or_default(); + + assert!( + fired.load(std::sync::atomic::Ordering::SeqCst), + "drain-loop hook never ran — pull_events must call EvtNext \ + at least once even on an empty channel" + ); + + // With the fix, the SetEvent fired by the hook during the + // drain is preserved; the next wait returns EventsAvailable + // immediately. Under the old post-drain ResetEvent order, it + // would time out — that is the #25194 freeze. + let (_subscription, result) = tokio::task::spawn_blocking(move || { + let r = subscription.wait_for_events_blocking(500); + (subscription, r) + }) + .await + .unwrap(); + + match result { + WaitResult::EventsAvailable => {} + WaitResult::Timeout => panic!( + "signal set during the drain window was lost — this is the \ + lost-wakeup race from vectordotdev/vector#25194. \ + pull_events must call ResetEvent BEFORE draining, not after." + ), + WaitResult::Shutdown => panic!("unexpected shutdown"), + } + } }