From 96e699cade28dc532db768a2154dc2e4b680580e Mon Sep 17 00:00:00 2001
From: Moritz Heidkamp <moritz.heidkamp@bevuta.com>
Date: Thu, 8 Jan 2026 10:25:00 +0100
Subject: [PATCH 1/8] Centralize callback invocation

To that end, extract `execute-callback` helper which is used in all places where callbacks are
invoked. As a side-effect, all callback invocation sites now also catch and log errors in executor
callbacks (before, only invocations via `set-deferred` did so).
---
 src/manifold/deferred.clj | 41 +++++++++++++++++----------------------
 1 file changed, 18 insertions(+), 23 deletions(-)

diff --git a/src/manifold/deferred.clj b/src/manifold/deferred.clj
index 81487599..f170cdc4 100644
--- a/src/manifold/deferred.clj
+++ b/src/manifold/deferred.clj
@@ -399,7 +399,19 @@
   [^IMutableDeferred deferred listener]
   (.cancelListener deferred listener))
 
+(defn execute-callback [^Executor executor callback]
+  (if (nil? executor)
+    (callback)
+    (.execute executor
+              (fn []
+                (try
+                  (callback)
+                  (catch Throwable e
+                    (log/error e "error in deferred handler")))))))
+
 (defmacro ^:private set-deferred [val token success? claimed? executor]
+  ;; Relies on all arguments being safe against double-evaluation, i.e. they're local variables at
+  ;; all callsites.
   `(if (utils/with-lock* ~'lock
          (when (and
                  (identical? ~(if claimed? ::claimed ::unset) ~'state)
@@ -412,17 +424,8 @@
        (clojure.core/loop []
          (when-let [^IDeferredListener l# (.poll ~'listeners)]
            (try
-             (if (nil? ~executor)
-               (~(if success? `.onSuccess `.onError) ^IDeferredListener l# ~val)
-               (.execute ~(with-meta executor {:tag "java.util.concurrent.Executor"})
-                         (fn []
-                           (try
-                             (~(if success? `.onSuccess `.onError) l# ~val)
-                             (catch Throwable e#
-                               #_(.printStackTrace e#)
-                               (log/error e# "error in deferred handler"))))))
+             (execute-callback ~executor #(~(if success? `.onSuccess `.onError) l# ~val))
              (catch Throwable e#
-               #_(.printStackTrace e#)
                (log/error e# "error in deferred handler")))
            (recur)))
        true)
@@ -518,9 +521,7 @@
                        (do
                          (.add listeners listener)
                          nil)))]
-        (if executor
-          (.execute executor f)
-          (f)))
+        (execute-callback executor f))
       true)
     (cancelListener [_ listener]
       (utils/with-lock* lock
@@ -595,9 +596,7 @@
   IMutableDeferred
   (claim [_] false)
   (addListener [_ listener]
-    (if (nil? executor)
-      (.onSuccess ^IDeferredListener listener val)
-      (.execute executor #(.onSuccess ^IDeferredListener listener val)))
+    (execute-callback executor #(.onSuccess ^IDeferredListener listener val))
     true)
   (cancelListener [_ listener] false)
   (success [_ x] false)
@@ -613,9 +612,7 @@
   (executor [_] executor)
   (realized [this] true)
   (onRealized [this on-success on-error]
-    (if executor
-      (.execute executor #(on-success val))
-      (on-success val)))
+    (execute-callback executor #(on-success val)))
   (successValue [_ default-value]
     val)
   (errorValue [_ default-value]
@@ -654,7 +651,7 @@
   (claim [_] false)
   (addListener [_ listener]
     (set! consumed? true)
-    (.onError ^IDeferredListener listener error)
+    (execute-callback executor #(.onError ^IDeferredListener listener error))
     true)
   (cancelListener [_ listener] false)
   (success [_ x] false)
@@ -671,9 +668,7 @@
   (realized [_] true)
   (onRealized [this on-success on-error]
     (set! consumed? true)
-    (if (nil? executor)
-      (on-error error)
-      (.execute executor #(on-error error))))
+    (execute-callback executor #(on-error error)))
   (successValue [_ default-value]
     default-value)
   (errorValue [_ default-value]

From b35d2cfe2977e202089df8f37ee6f32f743ae88b Mon Sep 17 00:00:00 2001
From: Moritz Heidkamp <moritz.heidkamp@bevuta.com>
Date: Thu, 8 Jan 2026 12:25:51 +0100
Subject: [PATCH 2/8] Also use executor in chained callbacks when deferred is
 already realized

* For `chain[']`, change `unwrap[']` so that it also stops when reaching a
  realized deferred with an explicit executor.
* Adjust `catch[']` and `finally[']` accordingly.
* Fill in `executor` method for all `IDeferred` implementations
* Add tests
---
 src/manifold/deferred.clj       | 165 +++++++++++++++++---------------
 test/manifold/deferred_test.clj |  66 ++++++++++++-
 2 files changed, 154 insertions(+), 77 deletions(-)

diff --git a/src/manifold/deferred.clj b/src/manifold/deferred.clj
index f170cdc4..d121b954 100644
--- a/src/manifold/deferred.clj
+++ b/src/manifold/deferred.clj
@@ -185,6 +185,10 @@
   [x]
   `(.realized ~(with-meta x {:tag "manifold.deferred.IDeferred"})))
 
+(definline ^:no-doc executor
+  [x]
+  `(.executor ~(with-meta x {:tag "manifold.deferred.IDeferred"})))
+
 (definline ^:no-doc success-value [x default-value]
   `(.successValue ~(with-meta x {:tag "manifold.deferred.IDeferred"}) ~default-value))
 
@@ -283,6 +287,7 @@
            (realized? this))
          ADeferred
          IDeferred
+         (executor [_])
          (realized [_]
            (or (.isDone x) (.isCancelled x)))
          (onRealized [_ on-success on-error]
@@ -317,6 +322,7 @@
          (.isRealized ^IPending x))
        ADeferred
        IDeferred
+       (executor [_])
        (realized [_]
          (.isRealized ^IPending x))
        (onRealized [_ on-success on-error]
@@ -738,23 +744,25 @@
   "Like unwrap, but does not coerce deferrable values."
   [x]
   (if (deferred? x)
-    (let [val (success-value x ::none)]
-      (if (identical? val ::none)
-        x
-        (recur val)))
+    (if (executor x)
+      x
+      (let [val (success-value x ::none)]
+        (if (identical? val ::none)
+          x
+          (recur val))))
     x))
 
 (defn unwrap
-  "Recursively unwraps a deferred or deferrable until either 1) a non-deferred
-   value is reached, or 2) an unrealized deferrable is reached."
+  "Recursively unwraps a deferred or deferrable until either 1) a non-deferred value is reached, or 2)
+  an unrealized deferrable is reached, or 3) a realized deferrable with an explicit executor is reached."
   [x]
   (let [d (->deferred x nil)]
-    (if (nil? d)
-      x
-      (let [val (success-value d ::none)]
-        (if (identical? ::none val)
-          d
-          (recur val))))))
+    (cond (nil? d) x
+          (executor d) d
+          :else (let [val (success-value d ::none)]
+                  (if (identical? ::none val)
+                    d
+                    (recur val))))))
 
 (defn connect
   "Conveys the realized value of `a` into `b`."
@@ -1069,40 +1077,44 @@
   ([x f g & fs]
    (apply chain- nil x f g fs)))
 
-(defn catch'
-  "Like `catch`, but does not coerce deferrable values."
-  ([x error-handler]
-   (catch' x nil error-handler))
-  ([x error-class error-handler]
-   (let [x      (chain' x)
-         catch? #(or (nil? error-class) (instance? error-class %))]
-     (if-not (deferred? x)
-
-       ;; not a deferred value, skip over it
-       x
-
-       (success-error-unrealized x
-         val x
-
-         err (try
-               (if (catch? err)
-                 (chain' (error-handler err))
-                 (error-deferred err))
-               (catch Throwable e
-                 (error-deferred e)))
-
-         (let [d' (deferred)]
-
-           (on-realized x
-                        #(success! d' %)
-                        #(try
-                           (if (catch? %)
-                             (chain'- d' (error-handler %))
-                             (chain'- d' (error-deferred %)))
-                           (catch Throwable e
-                             (error! d' e))))
+(let [subscribe (fn [x catch? error-handler]
+                  (let [d' (deferred)]
+                    (on-realized x
+                                 #(success! d' %)
+                                 #(try
+                                    (if (catch? %)
+                                      (chain'- d' (error-handler %))
+                                      (chain'- d' (error-deferred %)))
+                                    (catch Throwable e
+                                      (error! d' e))))
+
+                    d'))]
+  (defn catch'
+    "Like `catch`, but does not coerce deferrable values."
+    ([x error-handler]
+     (catch' x nil error-handler))
+    ([x error-class error-handler]
+     (let [catch? #(or (nil? error-class) (instance? error-class %))]
+       (if (not (deferred? x))
+         ;; not a deferred value, skip over it
+         x
+         (let [x' (unwrap' x)]
+           (if (not (deferred? x'))
+             ;; successfully realized, skip over it
+             x
+             (success-error-unrealized x'
+                val x
+
+                err (try
+                      (if (catch? err)
+                        (if (executor x')
+                          (subscribe x catch? error-handler)
+                          (chain' (error-handler err)))
+                        (error-deferred err))
+                      (catch Throwable e
+                        (error-deferred e)))
 
-           d'))))))
+                (subscribe x catch? error-handler)))))))))
 
 (defn catch
   "An equivalent of the catch clause, which takes an `error-handler` function that will be invoked
@@ -1125,36 +1137,39 @@
          chain)
      x)))
 
-(defn finally'
-  "Like `finally`, but doesn't coerce deferrable values."
-  [x f]
-  (success-error-unrealized x
-
-    val (try
-          (f)
-          x
-          (catch Throwable e
-            (error-deferred e)))
-
-    err (try
-          (f)
-          (error-deferred err)
-          (catch Throwable e
-            (error-deferred e)))
-
-    (let [d (deferred)]
-      (on-realized x
-                   #(try
-                      (f)
-                      (success! d %)
-                      (catch Throwable e
-                        (error! d e)))
-                   #(try
-                      (f)
-                      (error! d %)
-                      (catch Throwable e
-                        (error! d e))))
-      d)))
+(let [subscribe (fn [x f]
+                  (let [d (deferred)]
+                    (on-realized x
+                                 #(try
+                                    (f)
+                                    (success! d %)
+                                    (catch Throwable e
+                                      (error! d e)))
+                                 #(try
+                                    (f)
+                                    (error! d %)
+                                    (catch Throwable e
+                                      (error! d e))))
+                    d))]
+  (defn finally'
+    "Like `finally`, but doesn't coerce deferrable values."
+    [x f]
+    (if (executor x)
+      (subscribe x f)
+      (success-error-unrealized x
+        val (try
+              (f)
+              x
+              (catch Throwable e
+                (error-deferred e)))
+
+        err (try
+              (f)
+              (error-deferred err)
+              (catch Throwable e
+                (error-deferred e)))
+
+        (subscribe x f)))))
 
 (defn finally
   "An equivalent of the finally clause, which takes a no-arg side-effecting function that executes
diff --git a/test/manifold/deferred_test.clj b/test/manifold/deferred_test.clj
index d701a4a5..79062642 100644
--- a/test/manifold/deferred_test.clj
+++ b/test/manifold/deferred_test.clj
@@ -79,6 +79,67 @@
               (d/chain #(/ 1 %))
               (d/catch ArithmeticException (constantly :foo))))))
 
+(defn capture-callback-thread
+  ([d attach]
+   (capture-callback-thread d attach (fn [_ _])))
+  ([d attach realize!]
+   (let [t (d/deferred)]
+     (-> d
+         (attach (fn [& _]
+                   (d/success! t (Thread/currentThread))))
+         ;; Silence dropped error detection for finally + error cases
+         (d/catch' identity))
+     (realize! d [attach realize!])
+     @(d/timeout! t 100 ::timeout))))
+
+(deftest test-executors
+  (let [ex (ex/fixed-thread-executor 1)]
+    (doseq [make-deferred [#'d/deferred
+                           #'d/success-deferred
+                           #'d/error-deferred]]
+      (doseq [[attach realize!] [[#'d/chain    #'d/success!]
+                                 [#'d/chain'   #'d/success!]
+                                 [#'d/catch    #'d/error!]
+                                 [#'d/catch'   #'d/error!]
+                                 [#'d/finally  #'d/success!]
+                                 [#'d/finally  #'d/error!]
+                                 [#'d/finally' #'d/success!]
+                                 [#'d/finally' #'d/error!]]]
+        (when (condp = make-deferred
+                #'d/deferred true
+                #'d/success-deferred (or (= realize! #'d/success!)
+                                         (= attach #'d/finally)
+                                         (= attach #'d/finally'))
+                #'d/error-deferred (or (= realize! #'d/error!)
+                                       (= attach #'d/finally)
+                                       (= attach #'d/finally'))
+                false)
+          (testing (str "Using " (:name (meta attach)) ":")
+            (testing "Deferreds without an executor invoke callbacks on the thread which realizes them."
+              (let [d (if (= make-deferred #'d/deferred)
+                        (make-deferred)
+                        (make-deferred ::value))]
+                (is (= (Thread/currentThread)
+                       (capture-callback-thread d attach realize!)))
+                (when (= make-deferred #'d/deferred) ; all other deferred types are immediately realized anyway
+                  (testing "This is also the case for callbacks attached after realization."
+                    (is (= (Thread/currentThread)
+                           (capture-callback-thread d attach)))))))
+            (testing "Deferreds with an executor invoke callbacks on a thread from that executor."
+              (let [d (if (= make-deferred #'d/deferred)
+                        (make-deferred ex)
+                        (make-deferred ::value ex))
+                    t (capture-callback-thread d attach realize!)]
+                (when (is (instance? Thread t))
+                  (is (not= t (Thread/currentThread)))
+                  (is (re-find #"manifold-pool" (.getName t))))
+                (when (= make-deferred #'d/deferred) ; all other deferred types are immediately realized anyway
+                  (testing "This is also the case for callbacks attached after realization."
+                    (let [t (capture-callback-thread d attach)]
+                      (when (is (instance? Thread t))
+                        (is (not= t (Thread/currentThread)))
+                        (is (re-find #"manifold-pool" (.getName t)))))))))))))))
+
 (def ^:dynamic *test-dynamic-var*)
 
 (deftest test-let-flow
@@ -297,11 +358,12 @@
          (d/loop [[x & xs] [1 2 3]] (or (= x 3) (d/recur xs))))))
 
 (deftest test-coercion
-  (is (= 1 (-> 1 clojure.core/future d/->deferred deref)))
+  (is (= 2 (-> (clojure.core/future 1) d/->deferred (d/chain inc) deref)))
+  (is (= 2 (-> (promise) (deliver 1) d/->deferred (d/chain inc) deref)))
 
   (let [f (CompletableFuture.)]
     (.obtrudeValue f 1)
-    (is (= 1 (-> f d/->deferred deref))))
+    (is (= 2 (-> f d/->deferred (d/chain inc) deref))))
 
   (let [f (CompletableFuture.)]
     (.obtrudeException f (Exception.))

From c4d084e1e302abb56a0f5760643173d4b86035ce Mon Sep 17 00:00:00 2001
From: Moritz Heidkamp <moritz.heidkamp@bevuta.com>
Date: Thu, 8 Jan 2026 12:31:36 +0100
Subject: [PATCH 3/8] Don't re-enqueue callbacks when already on the same
 executor

To that end, keep track of the current executor in a new thread local variable.
---
 src/manifold/deferred.clj       | 17 +++++----
 src/manifold/executor.clj       | 62 +++++++++++++++++++--------------
 test/manifold/deferred_test.clj | 50 ++++++++++++++++++++++++++
 3 files changed, 95 insertions(+), 34 deletions(-)

diff --git a/src/manifold/deferred.clj b/src/manifold/deferred.clj
index d121b954..613ae690 100644
--- a/src/manifold/deferred.clj
+++ b/src/manifold/deferred.clj
@@ -406,7 +406,7 @@
   (.cancelListener deferred listener))
 
 (defn execute-callback [^Executor executor callback]
-  (if (nil? executor)
+  (if (or (nil? executor) (identical? executor (ex/current-executor)))
     (callback)
     (.execute executor
               (fn []
@@ -738,13 +738,15 @@
   ([error executor]
    (ErrorDeferred. error nil false executor)))
 
-(declare chain)
+(defn- different-executor? [d]
+  (when-let [ex (executor d)]
+    (not (identical? ex (ex/current-executor)))))
 
 (defn unwrap'
   "Like unwrap, but does not coerce deferrable values."
   [x]
   (if (deferred? x)
-    (if (executor x)
+    (if (different-executor? x)
       x
       (let [val (success-value x ::none)]
         (if (identical? val ::none)
@@ -754,11 +756,12 @@
 
 (defn unwrap
   "Recursively unwraps a deferred or deferrable until either 1) a non-deferred value is reached, or 2)
-  an unrealized deferrable is reached, or 3) a realized deferrable with an explicit executor is reached."
+  an unrealized deferrable is reached, or 3) a realized deferrable with a different executor than
+  the current one is reached."
   [x]
   (let [d (->deferred x nil)]
     (cond (nil? d) x
-          (executor d) d
+          (different-executor? d) d
           :else (let [val (success-value d ::none)]
                   (if (identical? ::none val)
                     d
@@ -1107,7 +1110,7 @@
 
                 err (try
                       (if (catch? err)
-                        (if (executor x')
+                        (if (different-executor? x')
                           (subscribe x catch? error-handler)
                           (chain' (error-handler err)))
                         (error-deferred err))
@@ -1154,7 +1157,7 @@
   (defn finally'
     "Like `finally`, but doesn't coerce deferrable values."
     [x f]
-    (if (executor x)
+    (if (different-executor? x)
       (subscribe x f)
       (success-error-unrealized x
         val (try
diff --git a/src/manifold/executor.clj b/src/manifold/executor.clj
index d5b8cc65..e43e67cc 100644
--- a/src/manifold/executor.clj
+++ b/src/manifold/executor.clj
@@ -25,6 +25,12 @@
 (definline executor []
   `(.get manifold.executor/executor-thread-local))
 
+
+(def ^ThreadLocal current-executor-thread-local (ThreadLocal.))
+
+(definline current-executor []
+  `(.get manifold.executor/current-executor-thread-local))
+
 (defmacro with-executor [executor & body]
   `(let [executor# (executor)]
      (.set executor-thread-local ~executor)
@@ -66,8 +72,10 @@
        (newThread [_ runnable]
          (let [name        (name-generator)
                curr-loader (.getClassLoader (class thread-factory))
-               f           #(do
-                              (.set executor-thread-local @executor-promise)
+               f           #(let [{:keys [executor onto?]} @executor-promise]
+                              (when onto?
+                                (.set executor-thread-local executor))
+                              (.set current-executor-thread-local executor)
                               (.run ^Runnable runnable))
                thread      ^Thread (new-thread nil f name (or stack-size 0))]
            (doto thread
@@ -136,35 +144,35 @@
                                  thread-factory
                                  (manifold.executor/thread-factory
                                    #(str "manifold-pool-" factory "-" (swap! thread-count inc))
-                                   (if onto?
-                                     executor-promise
-                                     (deliver (promise) nil))))
+                                   executor-promise))
         ^Executor$Controller c controller
         metrics                (if (identical? :none metrics)
                                  (EnumSet/noneOf Stats$Metric)
-                                 metrics)]
+                                 metrics)
+        executor             (Executor.
+                              thread-factory
+                              (if (and queue-length (pos? queue-length))
+                                (if (p/<= queue-length 1024)
+                                  (ArrayBlockingQueue. queue-length false)
+                                  (LinkedBlockingQueue. (int queue-length)))
+                                (SynchronousQueue. false))
+                              (if stats-callback
+                                (reify Executor$Controller
+                                  (shouldIncrement [_ n]
+                                    (.shouldIncrement c n))
+                                  (adjustment [_ s]
+                                    (stats-callback (stats->map s))
+                                    (.adjustment c s)))
+                                c)
+                              initial-thread-count
+                              metrics
+                              sample-period
+                              control-period
+                              TimeUnit/MILLISECONDS)]
     (assert controller "must specify :controller")
-    @(deliver executor-promise
-              (Executor.
-                thread-factory
-                (if (and queue-length (pos? queue-length))
-                  (if (p/<= queue-length 1024)
-                    (ArrayBlockingQueue. queue-length false)
-                    (LinkedBlockingQueue. (int queue-length)))
-                  (SynchronousQueue. false))
-                (if stats-callback
-                  (reify Executor$Controller
-                    (shouldIncrement [_ n]
-                      (.shouldIncrement c n))
-                    (adjustment [_ s]
-                      (stats-callback (stats->map s))
-                      (.adjustment c s)))
-                  c)
-                initial-thread-count
-                metrics
-                sample-period
-                control-period
-                TimeUnit/MILLISECONDS))))
+    (deliver executor-promise {:executor executor
+                               :onto? onto?})
+    executor))
 
 (defn fixed-thread-executor
   "Returns an executor which has a fixed number of threads."
diff --git a/test/manifold/deferred_test.clj b/test/manifold/deferred_test.clj
index 79062642..ce6ba93c 100644
--- a/test/manifold/deferred_test.clj
+++ b/test/manifold/deferred_test.clj
@@ -140,6 +140,56 @@
                         (is (not= t (Thread/currentThread)))
                         (is (re-find #"manifold-pool" (.getName t)))))))))))))))
 
+(deftest test-executor-affinity
+  (let [ex1 (ex/fixed-thread-executor 1)
+        ex2 (ex/fixed-thread-executor 1)
+        d (d/deferred ex1)
+        calls (atom {})
+        record-call (fn [f]
+                      (fn [x]
+                        (swap! calls update (ex/current-executor) (fnil inc 0))
+                        (f x)))
+        r (d/chain d
+                   ;; ex1
+                   (record-call inc)
+                   ;; ex1
+                   (record-call
+                    (fn [n]
+                      (d/chain
+                       ;; Will inherit the executor but is not re-enqueued for execution since we're
+                       ;; already on that executor
+                       (d/success-deferred n)
+                       ;; ex1
+                       (record-call inc))))
+                   ;; ex1
+                   (record-call
+                    (fn [n]
+                      (d/chain
+                       ;; Hand over to a different executor (ex2) and then propagate the result back
+                       ;; the the current one (ex1)
+                       (d/success-deferred n ex2)
+                       ;; ex2
+                       (record-call inc))))
+                   ;; ex1
+                   (record-call inc))
+        scheduled-callbacks (atom {})
+        exec d/execute-callback]
+    (with-redefs [d/execute-callback (fn [executor callback]
+                                       (when executor
+                                         (swap! scheduled-callbacks update executor (fnil inc 0)))
+                                       (exec executor callback))]
+      (d/success! d 1)
+      (is (= 5 @r))
+      (is (nil? (d/executor r)))
+      (let [c @calls]
+        (is (= #{ex1 ex2} (set (keys c))))
+        (is (= 5 (get c ex1)))
+        (is (= 1 (get c ex2))))
+      (let [s @scheduled-callbacks]
+        (is (= #{ex1 ex2} (set (keys s))))
+        (is (<= 1 (get s ex1) 2))
+        (is (= 1 (get s ex2)))))))
+
 (def ^:dynamic *test-dynamic-var*)
 
 (deftest test-let-flow

From 53ab721d5adf5680c318ad3bd296d99ecec6cc5f Mon Sep 17 00:00:00 2001
From: Moritz Heidkamp <moritz.heidkamp@bevuta.com>
Date: Thu, 8 Jan 2026 15:13:24 +0100
Subject: [PATCH 4/8] Fix race in `test-window-streams`

`test-window-streams` relied on the source being drained before invoking `stream->seq`. Due to
`dropping-stream` using `let-flow`, the timing has changed now that executors are used in all cases
so that this isn't always the case anymore. To fix this, explicitly await the channel to be drained
first.
---
 test/manifold/stream_test.clj | 7 +++++++
 1 file changed, 7 insertions(+)

diff --git a/test/manifold/stream_test.clj b/test/manifold/stream_test.clj
index ed4491d3..abeeefda 100644
--- a/test/manifold/stream_test.clj
+++ b/test/manifold/stream_test.clj
@@ -466,16 +466,23 @@
     (is (s/closed? sink))
     (is (s/closed? src))))
 
+(defn await-drained [s timeout]
+  (let [d (d/deferred)]
+    (s/on-drained s (fn [] (d/success! d true)))
+    (deref d timeout false)))
+
 (deftest test-window-streams
   (testing "dropping-stream"
     (let [s (s/->source (range 11))
           dropping-s (s/dropping-stream 10 s)]
+      (is (await-drained s 100))
       (is (= (range 10)
              (s/stream->seq dropping-s)))))
 
   (testing "sliding-stream"
     (let [s (s/->source (range 11))
           sliding-s (s/sliding-stream 10 s)]
+      (is (await-drained s 100))
       (is (= (range 1 11)
              (s/stream->seq sliding-s))))))
 

From 986c2e3a928d9e89bbc8e511c8d31a7a25987227 Mon Sep 17 00:00:00 2001
From: Moritz Heidkamp <moritz.heidkamp@bevuta.com>
Date: Thu, 8 Jan 2026 15:54:30 +0100
Subject: [PATCH 5/8] Simplify windowing streams

Specifically, don't use `let-flow` to avoid switching between different executors. Also, use
`d/chain'` instead of `d/chain` since `try-put!` is guaranteed to return a deferred.
---
 src/manifold/stream.clj | 14 ++++++++------
 1 file changed, 8 insertions(+), 6 deletions(-)

diff --git a/src/manifold/stream.clj b/src/manifold/stream.clj
index a7fbf1e4..de8c2182 100644
--- a/src/manifold/stream.clj
+++ b/src/manifold/stream.clj
@@ -1131,11 +1131,13 @@
      (connect-via
        source
        (fn [val]
-         (d/let-flow [put-result (try-put! sink val 0 ::timeout)]
-           (case put-result
-             true     true
-             false    false
-             ::timeout true)))
+         (d/chain'
+          (try-put! sink val 0 ::timeout)
+          (fn [put-result]
+            (case put-result
+              true     true
+              false    false
+              ::timeout true))))
        sink
        {:upstream?   true
         :downstream? true})
@@ -1159,7 +1161,7 @@
        source
        (fn [val]
          (d/loop []
-           (d/chain
+           (d/chain'
              (try-put! sink val 0 ::timeout)
              (fn [put-result]
                (case put-result

From 04f64dae8a604919fd9e65d3d691ac0fd0d3a34b Mon Sep 17 00:00:00 2001
From: Moritz Heidkamp <moritz.heidkamp@bevuta.com>
Date: Sun, 11 Jan 2026 15:11:21 +0100
Subject: [PATCH 6/8] Introduce `manifold.executor/wrap-executor`

See docstring
---
 src/manifold/executor.clj       | 44 ++++++++++++++++++++++++++++-----
 test/manifold/executor_test.clj | 19 ++++++++++++++
 2 files changed, 57 insertions(+), 6 deletions(-)

diff --git a/src/manifold/executor.clj b/src/manifold/executor.clj
index e43e67cc..30971915 100644
--- a/src/manifold/executor.clj
+++ b/src/manifold/executor.clj
@@ -51,6 +51,42 @@
   [group target name stack-size]
   (Thread. group target name stack-size))
 
+(defn- wrap-thread-runnable [runnable executor-promise]
+  #(let [{:keys [executor onto?]} @executor-promise]
+     (when onto?
+       (.set executor-thread-local executor))
+     (.set current-executor-thread-local executor)
+     (.run ^Runnable runnable)))
+
+(defn- wrap-thread-factory [tf executor-promise]
+  (reify ThreadFactory
+    (newThread [_ runnable]
+      (.newThread tf (wrap-thread-runnable runnable executor-promise)))))
+
+(defn wrap-executor
+  "Wraps an executor so that deferred callbacks will not be re-scheduled when they're bound to the
+  same executor.
+
+   |:---|:----
+   | `executor` | a `java.util.concurrent.Executor` or a function which accepts a `java.util.concurrent.ThreadFactory` and returns an executor for it. |
+   | `thread-factory` | an optional `java.util.concurrent.ThreadFactory` that creates the executor's threads. When given, `executor` must be a function. |
+   | `onto?` | if true, all streams and deferred generated in the scope of this executor will also be 'on' this executor. |"
+  ([executor]
+   (wrap-executor executor {}))
+  ([executor
+    {:keys [thread-factory
+            onto?]
+     :or   {onto? true}}]
+   (let [executor-promise (promise)
+         wrapped-executor (if thread-factory
+                            (executor (wrap-thread-factory thread-factory executor-promise))
+                            (reify java.util.concurrent.Executor
+                              (execute [_ runnable]
+                                (.execute executor (wrap-thread-runnable runnable executor-promise)))))]
+     (deliver executor-promise {:onto? onto?
+                                :executor wrapped-executor})
+     wrapped-executor)))
+
 (defn ^ThreadFactory thread-factory
   "Returns a `java.util.concurrent.ThreadFactory`.
 
@@ -72,12 +108,8 @@
        (newThread [_ runnable]
          (let [name        (name-generator)
                curr-loader (.getClassLoader (class thread-factory))
-               f           #(let [{:keys [executor onto?]} @executor-promise]
-                              (when onto?
-                                (.set executor-thread-local executor))
-                              (.set current-executor-thread-local executor)
-                              (.run ^Runnable runnable))
-               thread      ^Thread (new-thread nil f name (or stack-size 0))]
+               runnable    (wrap-thread-runnable runnable executor-promise)
+               thread      ^Thread (new-thread nil runnable name (or stack-size 0))]
            (doto thread
              (.setDaemon daemon?)
              (.setContextClassLoader curr-loader))))))))
diff --git a/test/manifold/executor_test.clj b/test/manifold/executor_test.clj
index 7abce917..c55db227 100644
--- a/test/manifold/executor_test.clj
+++ b/test/manifold/executor_test.clj
@@ -1,6 +1,7 @@
 (ns manifold.executor-test
   (:require
     [clojure.test :refer :all]
+    [manifold.deferred :as d]
     [manifold.executor :as e]
     [manifold.test :refer :all])
   (:import
@@ -70,4 +71,22 @@
         thread (.newThread tf (constantly nil))]
     (is (= "custom-name" (.getName thread)))))
 
+(deftest test-wrap-executor
+  (testing "with thread-factory"
+    (let [ex (e/wrap-executor #(Executors/newFixedThreadPool 1 %)
+                              {:thread-factory (Executors/defaultThreadFactory)})
+          d (d/deferred)]
+      (-> (d/success-deferred ::value ex)
+          (d/chain' (fn [_]
+                      (d/success! d (d/executor (d/deferred))))))
+      (is (= ex (deref d 100 ::timeout)))))
+
+  (testing "without thread-factory"
+    (let [ex (e/wrap-executor (Executors/newFixedThreadPool 1))
+          d (d/deferred)]
+      (-> (d/success-deferred ::value ex)
+          (d/chain' (fn [_]
+                      (d/success! d (d/executor (d/deferred))))))
+      (is (= ex (deref d 100 ::timeout))))))
+
 (instrument-tests-with-dropped-error-detection!)

From a815909d89026dc9ac992f1b95d891464e6f91bc Mon Sep 17 00:00:00 2001
From: Moritz Heidkamp <moritz.heidkamp@bevuta.com>
Date: Sun, 11 Jan 2026 15:13:00 +0100
Subject: [PATCH 7/8] Prevent redundant executor wrapping

---
 src/manifold/executor.clj | 67 ++++++++++++++++++++++-----------------
 1 file changed, 38 insertions(+), 29 deletions(-)

diff --git a/src/manifold/executor.clj b/src/manifold/executor.clj
index 30971915..ece889f6 100644
--- a/src/manifold/executor.clj
+++ b/src/manifold/executor.clj
@@ -63,6 +63,8 @@
     (newThread [_ runnable]
       (.newThread tf (wrap-thread-runnable runnable executor-promise)))))
 
+(definterface IExecutorTracking)
+
 (defn wrap-executor
   "Wraps an executor so that deferred callbacks will not be re-scheduled when they're bound to the
   same executor.
@@ -77,15 +79,22 @@
     {:keys [thread-factory
             onto?]
      :or   {onto? true}}]
-   (let [executor-promise (promise)
-         wrapped-executor (if thread-factory
-                            (executor (wrap-thread-factory thread-factory executor-promise))
-                            (reify java.util.concurrent.Executor
-                              (execute [_ runnable]
-                                (.execute executor (wrap-thread-runnable runnable executor-promise)))))]
-     (deliver executor-promise {:onto? onto?
-                                :executor wrapped-executor})
-     wrapped-executor)))
+   (if (instance? IExecutorTracking executor)
+     executor
+     (let [executor-promise (promise)
+           wrapped-executor (if thread-factory
+                              (let [executor (executor (wrap-thread-factory thread-factory executor-promise))]
+                                (reify java.util.concurrent.Executor
+                                  (execute [_ runnable]
+                                    (.execute executor runnable))
+                                  IExecutorTracking))
+                              (reify java.util.concurrent.Executor
+                                (execute [_ runnable]
+                                  (.execute executor (wrap-thread-runnable runnable executor-promise)))
+                                IExecutorTracking))]
+       (deliver executor-promise {:onto? onto?
+                                  :executor wrapped-executor})
+       wrapped-executor))))
 
 (defn ^ThreadFactory thread-factory
   "Returns a `java.util.concurrent.ThreadFactory`.
@@ -181,26 +190,26 @@
         metrics                (if (identical? :none metrics)
                                  (EnumSet/noneOf Stats$Metric)
                                  metrics)
-        executor             (Executor.
-                              thread-factory
-                              (if (and queue-length (pos? queue-length))
-                                (if (p/<= queue-length 1024)
-                                  (ArrayBlockingQueue. queue-length false)
-                                  (LinkedBlockingQueue. (int queue-length)))
-                                (SynchronousQueue. false))
-                              (if stats-callback
-                                (reify Executor$Controller
-                                  (shouldIncrement [_ n]
-                                    (.shouldIncrement c n))
-                                  (adjustment [_ s]
-                                    (stats-callback (stats->map s))
-                                    (.adjustment c s)))
-                                c)
-                              initial-thread-count
-                              metrics
-                              sample-period
-                              control-period
-                              TimeUnit/MILLISECONDS)]
+        executor             (proxy [Executor IExecutorTracking]
+                               [thread-factory
+                                (if (and queue-length (pos? queue-length))
+                                  (if (p/<= queue-length 1024)
+                                    (ArrayBlockingQueue. queue-length false)
+                                    (LinkedBlockingQueue. (int queue-length)))
+                                  (SynchronousQueue. false))
+                                (if stats-callback
+                                  (reify Executor$Controller
+                                    (shouldIncrement [_ n]
+                                      (.shouldIncrement c n))
+                                    (adjustment [_ s]
+                                      (stats-callback (stats->map s))
+                                      (.adjustment c s)))
+                                  c)
+                                initial-thread-count
+                                metrics
+                                sample-period
+                                control-period
+                                TimeUnit/MILLISECONDS])]
     (assert controller "must specify :controller")
     (deliver executor-promise {:executor executor
                                :onto? onto?})

From 35eb48a3fb4b403fd3a0b27be892bf905753c7ff Mon Sep 17 00:00:00 2001
From: Moritz Heidkamp <moritz.heidkamp@bevuta.com>
Date: Tue, 20 Jan 2026 10:44:56 +0100
Subject: [PATCH 8/8] Fix reflection warnings in `manifold.executor`

---
 src/manifold/executor.clj | 8 +++++---
 1 file changed, 5 insertions(+), 3 deletions(-)

diff --git a/src/manifold/executor.clj b/src/manifold/executor.clj
index ece889f6..6706f519 100644
--- a/src/manifold/executor.clj
+++ b/src/manifold/executor.clj
@@ -18,6 +18,8 @@
      ThreadFactory
      TimeUnit]))
 
+(set! *warn-on-reflection* true)
+
 ;;;
 
 (def ^ThreadLocal executor-thread-local (ThreadLocal.))
@@ -58,7 +60,7 @@
      (.set current-executor-thread-local executor)
      (.run ^Runnable runnable)))
 
-(defn- wrap-thread-factory [tf executor-promise]
+(defn- wrap-thread-factory [^ThreadFactory tf executor-promise]
   (reify ThreadFactory
     (newThread [_ runnable]
       (.newThread tf (wrap-thread-runnable runnable executor-promise)))))
@@ -75,7 +77,7 @@
    | `onto?` | if true, all streams and deferred generated in the scope of this executor will also be 'on' this executor. |"
   ([executor]
    (wrap-executor executor {}))
-  ([executor
+  ([^java.util.concurrent.Executor executor
     {:keys [thread-factory
             onto?]
      :or   {onto? true}}]
@@ -83,7 +85,7 @@
      executor
      (let [executor-promise (promise)
            wrapped-executor (if thread-factory
-                              (let [executor (executor (wrap-thread-factory thread-factory executor-promise))]
+                              (let [^java.util.concurrent.Executor executor (executor (wrap-thread-factory thread-factory executor-promise))]
                                 (reify java.util.concurrent.Executor
                                   (execute [_ runnable]
                                     (.execute executor runnable))