missionary.Cancelled not thrown about missionary HOT 6 CLOSED

(time
  (let [delays [{:id    1
                 :delay 1000}
                {:id    2
                 :delay 2000}
                {:id    1
                 :delay 500}
                {:id    1
                 :delay 200}]]
    (m/?
      (->>
        (m/ap
          (let [[k vs] (m/?> ##Inf (m/group-by :id (m/seed delays)))
                m (m/?< vs)]
            (try
              (m/!) ; <---- fixed
              (m/?
                (m/via m/blk (Thread/sleep (:delay m)) (assoc m :slept true)))
              (catch missionary.Cancelled c
                (println :cancelled m)
                (m/amb)))))
        (m/reduce
          (fn [out v]
            (println v)
            (conj out v))
          [])))))

(declare put)
(def c
  (let [observe-flow (m/observe (fn [!]
                                  (defn put [x]
                                    (! x))
                                  #(do)))
        task         (->>
                       (m/ap
                         (let [[k vs] (m/?> ##Inf (m/group-by :id observe-flow))
                               m (m/?< vs)]
                           (try
                             (m/!)                          ; <---- fixed
                             (m/?
                               (m/via m/blk (Thread/sleep (:delay m)) (assoc m :slept true)))
                             (catch missionary.Cancelled c
                               (println :cancelled m)
                               (m/amb))
                             (catch Throwable t
                               (println :throwing-instead-of-cancelling)
                               (m/amb)
                               ))))
                       (m/reduce
                         (fn [out v]
                           (println v)
                           (conj out v))
                         []))
        cancel       (task (fn [s] (println :success s))
                           (fn [f] (println :fail f)))]
    cancel

    ))

(put {:id    1
      :delay 200})
=> nil
{:id 1, :delay 200, :slept true}

(do (put {:id    1
          :delay 200})
    (put {:id    1
          :delay 500}))
:throwing-instead-of-cancelling
=> nil
{:id 1, :delay 500, :slept true}