         (let [next-considerable
               (try
                 (let [offers (async/alt!!
-                               offers-chan ([offers] offers)
+                               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)))

  [conn driver ^TaskScheduler fenzo fid pending-jobs num-considerable offers-chan offers]
  (log/debug "invoked handle-resource-offers!")
  (let [offer-stash (atom nil)] ;; This is a way to ensure we never lose offers fenzo assigned if an errors occures in the middle of processing
    (timers/time!
      handle-resource-offer!-duration
      (try
        (loop [] ;; This loop is for compare-and-set! below
          (let [scheduler-contents @pending-jobs
                db (db conn)
                _ (log/debug "There are" (count scheduler-contents) "pending jobs")
                considerable (->> scheduler-contents
                                  (filter (fn [job]
                                            (util/job-allowed-to-start? db job)))
                                  (take num-considerable))
                _ (log/debug "We'll consider scheduling" (count considerable) "of those pending jobs (limited to " num-considerable " due to backdown)")
                matches (match-offer-to-schedule fenzo considerable offers db fid)
                _ (reset! offer-stash (doall (for [{:keys [leases]} matches
                                                   lease leases]
                                               (:offer lease))))
                matched-jobs (for [match matches
                                   ^TaskAssignmentResult task-result (:tasks match)
                                   :let [task-request (.getRequest task-result)]]
                               (:job task-request))
                processed-matches (process-matches-for-backfill scheduler-contents (first considerable) matched-jobs)
                new-scheduler-contents (remove (fn [{pending-job-uuid :job/uuid}]
                                                 (or (contains? (:fully-processed processed-matches) pending-job-uuid)
                                                     (contains? (:upgrade-backfill processed-matches) pending-job-uuid)))
                                               scheduler-contents)
                ;; We don't remove backfilled jobs here, because although backfilled
                ;; jobs have already been scheduled in a sense, the scheduler still can't
                ;; adjust the status of backfilled tasks.
                ;; Backfilled tasks can be updgraded to non-backfilled after the jobs
                ;; prioritized above them are also scheduled.
                first-considerable-resources (-> considerable first util/job-ent->resources)
                match-resource-requirements (util/sum-resources-of-jobs matched-jobs)]
            (reset! front-of-job-queue-mem-atom
                    (or (:mem first-considerable-resources) 0))
            (reset! front-of-job-queue-cpus-atom
                    (or (:cpus first-considerable-resources) 0))
            (cond
              ;; Possible inocuous reasons for no matches: no offers, or no pending jobs.
              ;; Even beyond that, if Fenzo fails to match ANYTHING, "penalizing" it in the form of giving
              ;; it fewer jobs to look at is unlikely to improve the situation.
              ;; "Penalization" should only be employed when Fenzo does successfully match,
              ;; but the matches don't align with Cook's priorities.
              (empty? matches) true
              (not (compare-and-set! pending-jobs scheduler-contents new-scheduler-contents))
              (do
                (log/debug "Pending job atom contention encountered, recycling offers:" @offer-stash)
                (async/go
                  (async/>! offers-chan @offer-stash))
                (meters/mark! pending-job-atom-contended)
                (recur))
              :else
              (let [task-txns (for [{:keys [tasks leases]} matches
                                    :let [offers (mapv :offer leases)
                                          slave-id (:slave-id (first offers))]
                                    ^TaskAssignmentResult task tasks
                                    :let [request (.getRequest task)
                                          task-info (:task-info request)
                                          job-id (get-in request [:job :db/id])]]
                                [[:job/allowed-to-start? job-id]
                                 {:db/id (d/tempid :db.part/user)
                                  :job/_instance job-id
                                  :instance/task-id (:task-id task-info)
                                  :instance/hostname (.getHostname task)
                                  :instance/start-time (now)
                                  ;; NB command executor uses the task-id
                                  ;; as the executor-id
                                  :instance/executor-id (get-in
                                                          task-info
                                                          [:executor :executor-id]
                                                          (:task-id task-info))
                                  :instance/backfilled? (contains? (:backfill-jobs processed-matches) (get-in request [:job :job/uuid]))
                                  :instance/slave-id slave-id
                                  :instance/ports (.getAssignedPorts task)
                                  :instance/progress 0
                                  :instance/status :instance.status/unknown
                                  :instance/preempted? false}])
                    upgrade-txns (map (fn [instance-id]
                                        [:db/add instance-id :instance/backfilled? false])
                                      (:upgrade-backfill processed-matches))]
                ;; Note that this transaction can fail if a job was scheduled
                ;; during a race. If that happens, then other jobs that should
                ;; be scheduled will not be eligible for rescheduling until
                ;; the pending-jobs atom is repopulated
                @(d/transact
                   conn
                   (vec (apply concat upgrade-txns task-txns)))
                (log/info "Launching" (count task-txns) "tasks")
                (log/info "Upgrading" (count (:upgrade-backfill processed-matches)) "tasks from backfilled to proper")
                (log/info "Matched tasks" task-txns)
                ;; This launch-tasks MUST happen after the above transaction in
                ;; order to allow a transaction failure (due to failed preconditions)
                ;; to block the launch
                (meters/mark! scheduler-offer-matched
                              (->> matches
                                   (mapcat (comp :id :offer :leases))
                                   (distinct)
                                   (count)))
                (histograms/update! number-offers-matched
                                    (->> matches
                                         (mapcat (comp :id :offer :leases))
                                         (distinct)
                                         (count)))
                (meters/mark! matched-tasks (count task-txns))
                (meters/mark! matched-tasks-cpus (:cpus match-resource-requirements))
                (meters/mark! matched-tasks-mem (:mem match-resource-requirements))
                (doseq [{:keys [tasks leases]} matches
                        :let [offers (mapv :offer leases)
                              task-infos (mapv (fn process-results [^TaskAssignmentResult task]
                                                 (reduce
                                                   (fn add-ports-to-task-info [task-info [index port]]
                                                     (log/debug "task-info" task-info [index port])
                                                     (-> task-info
                                                         (update-in [:resources :ports]
                                                                    (fnil conj [])
                                                                    {:begin port :end port})
                                                         (assoc-in [:command :environment (str "PORT" index)]
                                                                   (str port))))
                                                   (:task-info (.getRequest task))
                                                   (map-indexed (fn [index port] [index port])
                                                                (.getAssignedPorts task))))
                                               tasks)
                              slave-id (:slave-id (first offers))]]
                  (mesos/launch-tasks
                    driver
                    (mapv :id offers)
                    (mapv #(-> %
                               (assoc :slave-id slave-id)
                               (dissoc :num-ports))
                          task-infos))
                  (doseq [^TaskAssignmentResult task tasks]
                    (.. fenzo
                        (getTaskAssigner)
                        (call (.getRequest task) (get-in (first leases) [:offer :hostname])))))
                (:matched-head? processed-matches)))))
      (catch Throwable t
        (meters/mark! handle-resource-offer!-errors)
        (log/error t "Error in match:" (ex-data t))
        (async/go
          (async/>! offers-chan @offer-stash))
        true  ; if an error happened, it doesn't mean we need to penalize Fenzo
        )))))

(defn view-incubating-offers
  [^TaskScheduler fenzo]

(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])
