Skip to content

Commit 308d246

Browse files
tot19claudepront
authored
fix(sources): prevent windows_event_log permanent freeze from signal-event lost wakeup (#25195)
* fix(windows_event_log): fix pre-drain ResetEvent race and add lost-wakeup regression tests 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, any SetEvent that fires between the last EvtNext call and the post-drain ResetEvent is silently lost — the subscription then hangs until the next OS event arrives (#25194). Fix: reset the handle *before* entering the drain loop. Signals raised during the drain are preserved because SetEvent on an already-signaled handle is a no-op. Re-arm (SetEvent) on early exits so the next pull_events revisits the channel without waiting for a fresh OS notification: - budget exhaustion - bookmark failure mid-batch - transient EvtNext error Regression tests: - test_pull_events_preserves_setevent_during_drain: installs DRAIN_STEP_HOOK to fire SetEvent mid-drain and asserts wait_for_events_blocking returns EventsAvailable, not Timeout. - test_speculative_pull_recovers_without_signal: manually clears the channel signal via ResetEvent, confirms wait times out, then asserts pull_events still returns events — proving the speculative timeout pull in mod.rs self-heals independently of signal state. Also: comment re-subscription break paths (ERROR_EVT_QUERY_RESULT_STALE and INVALID_POSITION) noting the speculative pull as a safety net if the re-subbed channel does not immediately re-signal; add serialization note to DRAIN_STEP_HOOK. * fix(windows_event_log): add speculative timeout pull, deduplicate processing, fix error handling Four related improvements to mod.rs: 1. Speculative pull on WaitResult::Timeout: call pull_events on every timeout cycle as a belt-and-suspenders self-heal. EvtNext returns ERROR_NO_MORE_ITEMS immediately on an empty channel (near-zero cost). If events are recovered a warning is emitted. Guarantees recovery within one timeout period regardless of the root cause of the lost wakeup. 2. Extract with_subscription_blocking helper: wraps the spawn_blocking ownership-transfer pattern (move subscription in, run blocking fn, return subscription + result). All three blocking calls (wait, normal pull, speculative pull) now use this helper instead of inlining spawn_blocking. 3. Extract process_event_batch helper: the parse/emit/send_batch/finalize sequence was duplicated verbatim between the EventsAvailable arm and the speculative-timeout arm. Extracted into a shared free async function. Rate limiting is applied consistently in both paths via the helper. 4. Fix error-handling asymmetry: the speculative pull Err branch previously only logged warn! and continued, so a non-recoverable error (access denied, channel not found) would spam warnings indefinitely. Now mirrors the EventsAvailable path: emit WindowsEventLogQueryError, break on non-recoverable errors, apply exponential backoff on recoverable ones. * test(windows_event_log source): harden lost-wakeup regression tests against flakiness Address two independent flakiness sources in the #25194 regression tests so the suite is stable on real Windows CI runners. test_pull_events_preserves_setevent_during_drain: - Replaced a 1000ms blocking wait with an immediate 0ms poll after pull_events returns, so the check measures only the reset/preserve behavior of pull_events and is not contaminated by unrelated Windows system events signaling the handle during a nonzero wait window. - Keyed the DRAIN_STEP_HOOK fire-once to the subscription's own signal handle so concurrent pull_events calls from other tests can't flip the hook first and SetEvent the wrong handle. test_speculative_pull_recovers_without_signal: - Same 500ms→0ms poll change, opposite direction: real events arriving during the wait would have re-signaled the manually-cleared handle and flipped the expected Timeout into a real signal result. - Seed a deterministic record via 'eventcreate' before subscription creation so the non-empty-events assertion is independent of whatever backlog the runner happens to have. Freshly provisioned images can have an empty Application log, which would otherwise cause pull_events(100) to legitimately return empty and false-fail the test. * fix(windows_event_log): implement Sync for subscription types to satisfy Send bound on source future EventLogSubscription and ChannelSubscription had unsafe impl Send but no Sync impl. Since &T: Send requires T: Sync, process_event_batch holding &EventLogSubscription across an .await made the entire source future !Send, causing a compile error (ICE + future-not-Send) in release builds. All mutation on these types requires &mut self; &self methods are read-only or delegate to already-Sync types (RateLimiter). The underlying Windows kernel handles are safe for concurrent access. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com> * chore(changelog): simplify windows_event_log fix fragment to user-focused one-liner Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com> * fix(windows_event_log): prioritize shutdown signal * fix(windows_event_log): lighten speculative timeout pulls --------- Co-authored-by: Claude Sonnet 4.6 <noreply@anthropic.com> Co-authored-by: Pavlos Rontidis <pavlos.rontidis@gmail.com>
1 parent 9c617a7 commit 308d246

3 files changed

Lines changed: 546 additions & 112 deletions

File tree

‎changelog.d/25194_windows_event_log_lost_wakeup.fix.md‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,3 @@
1+
The `windows_event_log` source no longer freezes after periods of inactivity.
2+
3+
authors: tot19

‎src/sources/windows_event_log/mod.rs‎

Lines changed: 192 additions & 89 deletions
Original file line numberDiff line numberDiff line change
@@ -29,7 +29,7 @@ cfg_if::cfg_if! {
2929
use vector_lib::EstimatedJsonEncodedSizeOf;
3030
use vector_lib::finalizer::OrderedFinalizer;
3131
use vector_lib::internal_event::{
32-
ByteSize, BytesReceived, CountByteSize, InternalEventHandle as _, Protocol,
32+
ByteSize, BytesReceived, CountByteSize, InternalEventHandle, Protocol,
3333
};
3434
use windows::Win32::Foundation::{DUPLICATE_SAME_ACCESS, DuplicateHandle, HANDLE};
3535
use windows::Win32::System::Threading::GetCurrentProcess;
@@ -48,6 +48,7 @@ cfg_if::cfg_if! {
4848
error::WindowsEventLogError,
4949
parser::EventLogParser,
5050
subscription::{EventLogSubscription, WaitResult},
51+
xml_parser::WindowsEvent,
5152
};
5253
}
5354
}
@@ -157,6 +158,107 @@ impl Finalizer {
157158
}
158159
}
159160

161+
/// Parse, emit metrics for, send, and finalize a non-empty batch of pulled Windows events.
162+
///
163+
/// Both the `EventsAvailable` path and the speculative-timeout path share this
164+
/// logic. Returns `true` if the downstream pipeline closed and the caller
165+
/// should break out of the main event loop.
166+
async fn process_event_batch(
167+
events: Vec<WindowsEvent>,
168+
parser: &EventLogParser,
169+
acknowledgements: bool,
170+
subscription: &EventLogSubscription,
171+
out: &mut SourceSender,
172+
finalizer: &Finalizer,
173+
events_received: &impl InternalEventHandle<Data = CountByteSize>,
174+
bytes_received: &impl InternalEventHandle<Data = ByteSize>,
175+
) -> bool {
176+
// Rate limiting between batches (async-compatible).
177+
if let Some(limiter) = subscription.rate_limiter() {
178+
limiter.until_ready().await;
179+
}
180+
181+
let (batch, receiver) = BatchNotifier::maybe_new_with_receiver(acknowledgements);
182+
let mut log_events = Vec::new();
183+
let mut total_byte_size = 0usize;
184+
let mut channels_in_batch = std::collections::HashSet::new();
185+
186+
for event in events {
187+
let channel = event.channel.clone();
188+
channels_in_batch.insert(channel.clone());
189+
let event_id = event.event_id;
190+
match parser.parse_event(event) {
191+
Ok(mut log_event) => {
192+
let byte_size = log_event.estimated_json_encoded_size_of();
193+
total_byte_size += byte_size.get();
194+
if let Some(ref batch) = batch {
195+
log_event = log_event.with_batch_notifier(batch);
196+
}
197+
log_events.push(log_event);
198+
}
199+
Err(e) => {
200+
emit!(WindowsEventLogParseError {
201+
error: e.to_string(),
202+
channel,
203+
event_id: Some(event_id),
204+
});
205+
}
206+
}
207+
}
208+
209+
if !log_events.is_empty() {
210+
let count = log_events.len();
211+
events_received.emit(CountByteSize(count, total_byte_size.into()));
212+
bytes_received.emit(ByteSize(total_byte_size));
213+
214+
// BACK PRESSURE: block until the pipeline accepts the batch.
215+
// We don't call EvtNext again until this completes.
216+
if let Err(_error) = out.send_batch(log_events).await {
217+
emit!(StreamClosedError { count });
218+
return true; // signal: break the main loop
219+
}
220+
221+
// Register checkpoint entry with the finalizer.
222+
let bookmarks: Vec<(String, String)> = channels_in_batch
223+
.into_iter()
224+
.filter_map(|channel| {
225+
subscription
226+
.get_bookmark_xml(&channel)
227+
.map(|xml| (channel, xml))
228+
})
229+
.collect();
230+
231+
if !bookmarks.is_empty() {
232+
let entry = FinalizerEntry { bookmarks };
233+
finalizer.finalize(entry, receiver).await;
234+
}
235+
}
236+
237+
false // pipeline still open
238+
}
239+
240+
/// Transfer ownership of `subscription` into a `spawn_blocking` task, run `f`
241+
/// on it, then return both the subscription and the result.
242+
///
243+
/// All blocking Windows APIs (`WaitForMultipleObjects`, `EvtNext`, `EvtRender`)
244+
/// must run in `spawn_blocking` to avoid stalling the async runtime. The
245+
/// ownership-transfer pattern ensures only one thread holds the subscription
246+
/// at a time, preventing data races without requiring locks.
247+
async fn with_subscription_blocking<F, R>(
248+
subscription: EventLogSubscription,
249+
f: F,
250+
) -> Result<(EventLogSubscription, R), WindowsEventLogError>
251+
where
252+
F: FnOnce(EventLogSubscription) -> (EventLogSubscription, R) + Send + 'static,
253+
R: Send + 'static,
254+
{
255+
tokio::task::spawn_blocking(move || f(subscription))
256+
.await
257+
.map_err(|e| WindowsEventLogError::ConfigError {
258+
message: format!("Blocking subscription task panicked: {e}"),
259+
})
260+
}
261+
160262
/// Windows Event Log source implementation
161263
pub struct WindowsEventLogSource {
162264
config: WindowsEventLogConfig,
@@ -281,42 +383,25 @@ impl WindowsEventLogSource {
281383
// Ownership transfer ensures no data races between the blocking thread
282384
// and async code. The shutdown watcher uses a raw HANDLE value (just an
283385
// integer) to signal shutdown without needing access to the subscription.
284-
let (returned_sub, wait_result) = tokio::task::spawn_blocking({
285-
let sub = subscription;
286-
move || {
386+
let (returned_sub, wait_result) =
387+
with_subscription_blocking(subscription, move |sub| {
287388
let result = sub.wait_for_events_blocking(timeout_ms);
288389
(sub, result)
289-
}
290-
})
291-
.await
292-
.map_err(|e| WindowsEventLogError::ConfigError {
293-
message: format!("Wait task panicked: {e}"),
294-
})?;
295-
390+
})
391+
.await?;
296392
subscription = returned_sub;
297393

298394
match wait_result {
299395
WaitResult::EventsAvailable => {
300396
// Pull events via spawn_blocking (EvtNext/EvtRender are blocking APIs)
301-
let (returned_sub, events_result) = tokio::task::spawn_blocking({
302-
let mut sub = subscription;
303-
move || {
397+
let (returned_sub, events_result) =
398+
with_subscription_blocking(subscription, move |mut sub| {
304399
let result = sub.pull_events(batch_size);
305400
(sub, result)
306-
}
307-
})
308-
.await
309-
.map_err(|e| WindowsEventLogError::ConfigError {
310-
message: format!("Pull task panicked: {e}"),
311-
})?;
312-
401+
})
402+
.await?;
313403
subscription = returned_sub;
314404

315-
// Rate limiting between batches (async-compatible)
316-
if let Some(limiter) = subscription.rate_limiter() {
317-
limiter.until_ready().await;
318-
}
319-
320405
match events_result {
321406
Ok(events) if events.is_empty() => {
322407
error_backoff = std::time::Duration::from_millis(100);
@@ -328,65 +413,19 @@ impl WindowsEventLogSource {
328413
message = "Pulled Windows Event Log events.",
329414
event_count = events.len()
330415
);
331-
332-
let (batch, receiver) =
333-
BatchNotifier::maybe_new_with_receiver(acknowledgements);
334-
335-
let mut log_events = Vec::new();
336-
let mut total_byte_size = 0;
337-
let mut channels_in_batch = std::collections::HashSet::new();
338-
339-
for event in events {
340-
let channel = event.channel.clone();
341-
channels_in_batch.insert(channel.clone());
342-
let event_id = event.event_id;
343-
match parser.parse_event(event) {
344-
Ok(mut log_event) => {
345-
let byte_size = log_event.estimated_json_encoded_size_of();
346-
total_byte_size += byte_size.get();
347-
348-
if let Some(ref batch) = batch {
349-
log_event = log_event.with_batch_notifier(batch);
350-
}
351-
352-
log_events.push(log_event);
353-
}
354-
Err(e) => {
355-
emit!(WindowsEventLogParseError {
356-
error: e.to_string(),
357-
channel,
358-
event_id: Some(event_id),
359-
});
360-
}
361-
}
362-
}
363-
364-
if !log_events.is_empty() {
365-
let count = log_events.len();
366-
events_received.emit(CountByteSize(count, total_byte_size.into()));
367-
bytes_received.emit(ByteSize(total_byte_size));
368-
369-
// BACK PRESSURE: block here until the pipeline accepts
370-
// the batch. We don't call EvtNext again until this completes.
371-
if let Err(_error) = out.send_batch(log_events).await {
372-
emit!(StreamClosedError { count });
373-
break;
374-
}
375-
376-
// Register checkpoint entry with finalizer
377-
let bookmarks: Vec<(String, String)> = channels_in_batch
378-
.into_iter()
379-
.filter_map(|channel| {
380-
subscription
381-
.get_bookmark_xml(&channel)
382-
.map(|xml| (channel, xml))
383-
})
384-
.collect();
385-
386-
if !bookmarks.is_empty() {
387-
let entry = FinalizerEntry { bookmarks };
388-
finalizer.finalize(entry, receiver).await;
389-
}
416+
if process_event_batch(
417+
events,
418+
&parser,
419+
acknowledgements,
420+
&subscription,
421+
&mut out,
422+
&finalizer,
423+
&events_received,
424+
&bytes_received,
425+
)
426+
.await
427+
{
428+
break;
390429
}
391430
}
392431
Err(e) => {
@@ -415,10 +454,6 @@ impl WindowsEventLogSource {
415454
}
416455

417456
WaitResult::Timeout => {
418-
// A full wait cycle without errors means the system is healthy;
419-
// reset backoff so the next transient error starts fresh.
420-
error_backoff = std::time::Duration::from_millis(100);
421-
422457
// Periodic checkpoint flush (sync mode only)
423458
if !acknowledgements && last_checkpoint.elapsed() >= checkpoint_interval {
424459
if let Err(e) = subscription.flush_bookmarks().await {
@@ -448,6 +483,74 @@ impl WindowsEventLogSource {
448483
);
449484
}
450485
}
486+
487+
// Speculative pull: self-heal against any lost-wakeup scenario,
488+
// regardless of root cause. If the OS signal was lost through any
489+
// mechanism (not just the pre-drain race fixed in #25194), this
490+
// ensures the source recovers within one timeout period.
491+
// Use the speculative pull variant so idle timeout cycles don't
492+
// refresh per-channel record-count gauges via EvtOpenLog /
493+
// EvtGetLogInfo on every configured channel.
494+
let (returned_sub, speculative_result) =
495+
with_subscription_blocking(subscription, move |mut sub| {
496+
let result = sub.pull_events_speculative(batch_size);
497+
(sub, result)
498+
})
499+
.await?;
500+
subscription = returned_sub;
501+
502+
match speculative_result {
503+
Ok(events) if events.is_empty() => {
504+
// Healthy cycle: reset backoff so the next transient
505+
// error starts fresh.
506+
error_backoff = std::time::Duration::from_millis(100);
507+
}
508+
Ok(events) => {
509+
// Healthy cycle: reset backoff so the next transient
510+
// error starts fresh.
511+
error_backoff = std::time::Duration::from_millis(100);
512+
warn!(
513+
message = "Speculative timeout pull recovered events; possible lost wakeup detected.",
514+
event_count = events.len(),
515+
);
516+
if process_event_batch(
517+
events,
518+
&parser,
519+
acknowledgements,
520+
&subscription,
521+
&mut out,
522+
&finalizer,
523+
&events_received,
524+
&bytes_received,
525+
)
526+
.await
527+
{
528+
break;
529+
}
530+
}
531+
Err(e) => {
532+
emit!(WindowsEventLogQueryError {
533+
channel: "all".to_string(),
534+
query: None,
535+
error: e.to_string(),
536+
});
537+
if !e.is_recoverable() {
538+
error!(
539+
message = "Non-recoverable speculative pull error, shutting down.",
540+
error = %e
541+
);
542+
break;
543+
}
544+
// Exponential backoff mirrors the EventsAvailable error path.
545+
warn!(
546+
message = "Recoverable speculative pull error, backing off.",
547+
backoff_ms = error_backoff.as_millis() as u64,
548+
error = %e
549+
);
550+
tokio::time::sleep(error_backoff).await;
551+
error_backoff = (error_backoff * 2).min(MAX_ERROR_BACKOFF);
552+
}
553+
}
451554
}
452555

453556
WaitResult::Shutdown => {

0 commit comments

Comments
 (0)