                                       _ (when (= timeout-ch source-ch)
                                           (log/warn "timeout while reading from error-chan"))
                                       ex (ex-info "Unable to stream, back pressure in resp-chan. Is connection live?"
-                                                  {:bytes-pending bytes-read
+                                                  {:body-source-closed? (protocols/closed? body)
+                                                   :body-target-closed? (protocols/closed? resp-chan)
+                                                   :bytes-pending bytes-read
                                                    :bytes-streamed bytes-streamed
-                                                   :correlation-id (cid/get-correlation-id)}
+                                                   :correlation-id (cid/get-correlation-id)
+                                                   :error-cause :client-error}
                                                   error)]
                                   (meters/mark! stream-back-pressure)
                                   (deliver reservation-status-promise :client-error)

                              (log/info "done processing request" status)
                              (when (= :success status)
                                (counters/inc! (metrics/service-counter service-id "request-counts" "successful")))
                              (when (= :generic-error status)
                                (log/error "there was a generic error in processing the request;"
                                            "if this is a client or server related issue, the code needs to be updated."))
                              (when (not= :success-async status)
                                (counters/dec! (metrics/service-counter service-id "request-counts" "outstanding"))
                                (statsd/gauge-delta! metric-group "request_outstanding" -1))

                            (log/error e "error releasing instance!"))))
      instance)))

(defn- classify-error
  "Classifies the error responses from the backend into the following vector:
   - error cause (:client-error, :instance-error or :generic-error),
   - associated error message, and
   - the http status code."
  [error]
  (let [classification (cond (instance? ExceptionInfo error)
                             (let [[error-cause message status] (classify-error (ex-cause error))
                                   error-cause (or (-> error ex-data :error-cause) error-cause)]
                               [error-cause message status])
                             (instance? IllegalStateException error)
                             [:generic-error (.getMessage error) 400]
                             (instance? EofException error)
                             [:client-error "Connection unexpectedly closed while streaming request" 400]
                             (instance? TimeoutException error)
                             [:instance-error (utils/message :backend-request-timed-out) 504]
                             :else
                             [:instance-error (utils/message :backend-request-failed) 502])
        [error-cause _ _] classification]
    (log/info (-> error .getClass .getCanonicalName) (.getMessage error) "identified as" error-cause)
    classification))

(defn- handle-response-error
  "Handles error responses from the backend."
  [error reservation-status-promise service-id request]
  (let [[error-cause message status] (classify-error error)
        metrics-map (metrics/retrieve-local-stats-for-service service-id)]
    (deliver reservation-status-promise error-cause)
    (utils/exception->response (ex-info message (assoc metrics-map :status status) error) request)))

(let [min-buffer-size 1024
