                          offer-size-mem
                          (get-in offer [:resources :mem] 0)))
 
-                      (when-not (async/offer! offers-chan offers)
-                        (decline-offers driver offers)))
+                      (if (async/offer! offers-chan offers)
+                        (counters/inc! offer-chan-depth)
+                        (do (log/warn "Offer chan is full. Are we not handling offers fast enough?")
+                            (meters/mark! offer-chan-full-error)
+                            (future
+                              (try
+                                (decline-offers driver (map :id offers))
+                                (catch Exception e
+                                  (log/error e "Unable to decline offers!")))))))
       (statusUpdate [driver status]
-                    (handle-status-update conn driver fenzo status)))
+                    (future (handle-status-update conn driver fenzo status))))
      :view-incubating-offers (fn get-resources-atom [] @resources-atom)}))

                                       id (:id offer)]
                                   (log/debug "Fenzo is declining offer" offer)
                                   (if-let [driver @driver]
                                     (mesos/decline-offer driver id)
                                     (log/error "Unable to decline offer; no current driver"))))))
      (build)))


(defn handle-status-update
  "Takes a status update from mesos."
  [conn driver ^TaskScheduler fenzo status]
  (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} :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])
