                       (map (juxt :name :id))
                       (into {}))
 
-        non-input-task-ids (set (vals (dissoc task-ids :in :out)))
+        non-input-task-ids (set (vals (dissoc task-ids :in)))
 
         remove-completed-ch (go-loop []
                                      (let [v (<!! out-chan)]

    :onyx/batch-size batch-size
    :onyx/doc "Writes segments to a core.async channel"}])

(def workflow [[:in :inc] [:inc :out]])

(def chan-size 10000)


    :onyx/batch-size batch-size
    :onyx/doc "Reads segments from a core.async channel"}

   {: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}
