                  (first (first (sort-by second top)))))
         trim (fn trim [top smallest]
                (if (< k (count top))
-                 (dissoc top smallest)
-                 top))]
+                 [(dissoc top smallest) smallest]
+                 [top]))]
     (cond
       ; Expired or irrelevant event
       (or (expired? event)

  Index everything, but tag the top k events with \"top\":

  (top 10 :metric
    (adjust [:tags conj \"top\"]
      index)
    index)

  This implementation of top is lazy, in a sense. It won't proactively expire
  events which are bumped from the top-k set--you have to wait for another
  event with the same host and service to arrive before child streams will know
  it's expired. At some point I (or an enterprising committer) should fix
  that."
  ([k f top-stream]
   (top k f top-stream bit-bucket))
  ([k f top-stream bottom-stream]
   (let [state (atom [nil {}])]
     (dual (fn stream [event]
             (let [top   (second (swap! state top-update k f event))]
               (top [(:host event) (:service event)])))
           top-stream
           bottom-stream))))



      ; In the top set
      :else
      (let [[top out] (trim (assoc top ekey value) smallest)]
        (if (or (nil? (top smallest))
                (neg? (compare value (top smallest))))
          [(scan top) top out]
          [smallest top out])))))

(defn top
  "Bifurcates a stream into a dual pair of streams: one for the top k events,
  and one for the bottom k events. An optional third stream can be provided
  which will receive events which are leaving the top

  f is a function which maps events to comparable values, e.g. numbers. If an
  incoming event e falls in the top k, the top stream receives e and the bottom
