From 0fc811b700a0b777a8db263b06b2f861fc084d03 Mon Sep 17 00:00:00 2001
From: daisuke8000 <daisuke12108000@gmail.com>
Date: Sun, 5 Jul 2026 17:50:07 +0900
Subject: [PATCH] Replace exit_watch .then(async) with .map (DSK-130)

Use synchronous map for exit-on-message and exit-on-level stages instead of per-item futures.

Co-authored-by: Cursor <cursoragent@cursor.com>
---
 .../rustern-core/src/pipeline/exit_watch.rs   | 33 ++++++++-----------
 1 file changed, 13 insertions(+), 20 deletions(-)

diff --git a/crates/rustern-core/src/pipeline/exit_watch.rs b/crates/rustern-core/src/pipeline/exit_watch.rs
index 781b229..65f2572 100644
--- a/crates/rustern-core/src/pipeline/exit_watch.rs
+++ b/crates/rustern-core/src/pipeline/exit_watch.rs
@@ -103,17 +103,13 @@ where
     P: Into<Arc<[Regex]>>,
 {
     let patterns: Arc<[Regex]> = patterns.into();
-    inner.then(move |r| {
-        let state = state.clone();
-        let patterns = Arc::clone(&patterns);
-        async move {
-            if let Ok(ref ev) = r
-                && patterns.iter().any(|re| re.is_match(ev.message.as_ref()))
-            {
-                state.fire();
-            }
-            r
+    inner.map(move |r| {
+        if let Ok(ref ev) = r
+            && patterns.iter().any(|re| re.is_match(ev.message.as_ref()))
+        {
+            state.fire();
         }
+        r
     })
 }
 
@@ -126,17 +122,14 @@ where
     S: Stream<Item = Result<LogEvent, LogSourceError>> + Send + 'static,
 {
     let threshold = min_level.rank();
-    inner.then(move |r| {
-        let state = state.clone();
-        async move {
-            if let Ok(ref ev) = r
-                && let Some(ref lv) = ev.level
-                && event_level_rank(lv) >= threshold
-            {
-                state.fire();
-            }
-            r
+    inner.map(move |r| {
+        if let Ok(ref ev) = r
+            && let Some(ref lv) = ev.level
+            && event_level_rank(lv) >= threshold
+        {
+            state.fire();
         }
+        r
     })
 }