     :onyx/batch-size batch-size
     :onyx/doc "Reads segments from a core.async channel"}
 
-   {:onyx/name :inc
+   {:onyx/name :intermediate-1
+    :onyx/fn :onyx.peer.fault-tolerance/process-middle
+    :onyx/type :function
+    :onyx/batch-size batch-size}
+
+   {:onyx/name :intermediate-2
     :onyx/fn :onyx.peer.fault-tolerance/process-middle
     :onyx/type :function
     :onyx/batch-size batch-size}

      (fact after-done => #{}))

    (close! kill-ch)
    (helper-env/remove-n-peers test-env 9))

    (finally 

                      (map (juxt :name :id))
                      (into {}))

        non-input-task-ids (set (vals (dissoc task-ids :in)))

        remove-completed-ch (go-loop []
                                     (let [v (<!! out-chan)]
