       (>!! status-check-ch (munge-write-batch event))
       (recur))))
 
-(defn status-check-loop [status-ch commit-tx-ch reset-payload-node-ch]
+(defn status-check-loop [status-ch ack-ch reset-payload-node-ch]
   (loop []
     (when-let [event (<!! status-ch)]
       (let [event (munge-status-check event)]
         (if (:commit? event)
-          (>!! commit-tx-ch event)
+          (>!! ack-ch event)
           (>!! reset-payload-node-ch event))
         (recur)))))
 
-(defn commit-tx-loop [commit-ch ack-ch]
+(defn ack-loop [ack-ch commit-ch]
+  (loop []
+    (when-let [event (<!! ack-ch)]
+      (>!! commit-ch (munge-ack event))
+      (recur))))
+
+(defn commit-tx-loop [commit-ch close-resources-ch]
   (loop []
     (when-let [event (<!! commit-ch)]
-      (>!! ack-ch (munge-commit-tx event))
-      (recur))))
-
-(defn ack-loop [ack-ch close-resources-ch]
-  (loop []
-    (when-let [event (<!! ack-ch)]
-      (>!! close-resources-ch (munge-ack event))
+      (>!! close-resources-ch (munge-commit-tx event))
       (recur))))
 
 (defn close-resources-loop [close-ch complete-task-ch]
   (loop []
     (when-let [event (<!! close-ch)]
+      (taoensso.timbre/info "Closing out resources for " (:session event))
       (>!! complete-task-ch (munge-close-resources event))
       (recur))))
 

          compress-batch-ch (chan 1)
          status-check-ch (chan 1)
          write-batch-ch (chan 1)
          commit-tx-ch (chan 1)
          ack-ch (chan 1)
          close-resources-ch (chan 1)
          complete-task-ch (chan 1)
          reset-payload-node-ch (chan 1)

        java.lang.Exception
        (fn [e & _] (.printStackTrace e)))

      (dire/with-handler! #'ack-loop
        java.lang.Exception
        (fn [e & args] (.printStackTrace e)))

      (dire/with-handler! #'commit-tx-loop
        java.lang.Exception
        (fn [e & _] (.printStackTrace e)))

      (dire/with-handler! #'close-resources-loop
        java.lang.Exception
