                         (rank-jobs (db conn) (d/db conn) offensive-job-filter))))))
 
 (meters/defmeter [cook-mesos scheduler mesos-error])
+(meters/defmeter [cook-mesos scheduler offer-chan-full-error])
 
 (defn make-fenzo-scheduler
   [driver offer-incubate-time-ms]

(defn handle-status-update
  "Takes a status update from mesos."
  [conn driver ^TaskScheduler fenzo status]
  (future
    (log/info "Mesos status is: " status)
    (timers/time!
      handle-status-update-duration
      (try (let [db (db conn)
                 {:keys [task-id reason] task-state :state} status
                 [job instance prior-instance-status] (first (q '[:find ?j ?i ?status
                                                                  :in $ ?task-id
                                                                  :where
                                                                  [?i :instance/task-id ?task-id]
                                                                  [?i :instance/status ?s]
                                                                  [?s :db/ident ?status]
                                                                  [?j :job/instance ?i]]
                                                                db task-id))
                 job-ent (d/entity db job)
                 instance-ent (d/entity db instance)
                 retries-so-far (count (:job/instance job-ent))
                 previous-reason (reason/instance-entity->reason-entity db instance-ent)
                 instance-status (condp contains? task-state
                                   #{:task-staging} :instance.status/unknown
                                   #{:task-starting
                                     :task-running} :instance.status/running
                                   #{:task-finished} :instance.status/success
                                   #{:task-failed
                                     :task-killed
                                     :task-lost
                                     :task-error} :instance.status/failed)
                 prior-job-state (:job/state (d/entity db job))
                 progress (try
                              (when (:data status)
                                  (:percent (read-string (String. (:data status)))))
                          (catch Exception e
                              (log/debug e "Error parse mesos status data. Is it in the format we expect?")))
                 instance-ent (d/entity db instance)
                 instance-runtime (- (.getTime (now)) ; Used for reporting
                                     (.getTime (:instance/start-time instance-ent)))
                 job-resources (util/job-ent->resources job-ent)]
             (when (#{:instance.status/success :instance.status/failed} instance-status)
               (log/debug "Unassigning task" task-id "from" (:instance/hostname instance-ent))
               (try
                 (.. fenzo
                     (getTaskUnAssigner)
                     (call task-id (:instance/hostname instance-ent)))
                 (catch Exception e
                   (log/error e "Failed to unassign task" task-id "from" (:instance/hostname instance-ent)))))
             (when (= instance-status :instance.status/success)
               (handle-throughput-metrics success-throughput-metrics
                                          job-resources
                                          instance-runtime)
               (handle-throughput-metrics complete-throughput-metrics
                                          job-resources
                                          instance-runtime))
             (when (= instance-status :instance.status/failed)
               (handle-throughput-metrics fail-throughput-metrics
                                          job-resources
                                          instance-runtime)
               (handle-throughput-metrics complete-throughput-metrics
                                          job-resources
                                          instance-runtime))
             ;; This code kills any task that "shouldn't" be running
             (when (and
                     (or (nil? instance) ; We could know nothing about the task, meaning a DB error happened and it's a waste to finish
                         (= prior-job-state :job.state/completed) ; The task is attached to a failed job, possibly due to instances running on multiple hosts
                         (= prior-instance-status :instance.status/failed)) ; The kill-task message could've been glitched on the network
                     (contains? #{:task-running
                                  :task-staging
                                  :task-starting}
                                task-state)) ; killing an unknown task causes a TASK_LOST message. Break the cycle! Only kill non-terminal tasks
               (log/warn "Attempting to kill task" task-id
                         "as instance" instance "with" prior-job-state "and" prior-instance-status
                         "should've been put down already")
               (meters/mark! tasks-killed-in-status-update)
               (mesos/kill-task driver task-id))
             (when-not (nil? instance)
               ;; (println "update:" task-id task-state job instance instance-status prior-job-state)
               (log/debug "Transacting updated state for instance" instance "to status" instance-status)
               (transact-with-retries conn
                                      (concat
                                        [[:instance/update-state instance instance-status]]
                                        (when (and (#{:instance.status/failed} instance-status) (not previous-reason) reason)
                                          [[:db/add instance :instance/reason (reason/mesos-reason->cook-reason-entity-id db reason)]])
                                        (when (#{:instance.status/success
                                                 :instance.status/failed} instance-status)
                                          [[:db/add instance :instance/end-time (now)]])
                                        (when progress
                                          [[:db/add instance :instance/progress progress]])))))
           (catch Exception e
             (log/error e "Mesos scheduler status update error"))))))

(timers/deftimer [cook-mesos scheduler tx-report-queue-processing-duration])
(meters/defmeter [cook-mesos scheduler tx-report-queue-datoms])

        (let [next-considerable
              (try
                (let [offers (async/alt!!
                               offers-chan ([offers]
                                            (counters/dec! offer-chan-depth)
                                            offers)
                               timer-chan ([_] [])
                               :priority true)
                      ;; Try to clear the channel
                      offers (->> (repeatedly chan-length #(async/poll! offers-chan))
                                  (filter nil?)
                                  (apply concat offers))
                      _ (log/debug "Passing following offers to handle-resource-offers!" offers)
                      matched-head? (handle-resource-offers! conn @driver-atom fenzo @fid-atom pending-jobs-atom num-considerable offers-chan offers)]
                  (when (seq offers)
                    (reset! resources-atom (view-incubating-offers fenzo)))
