     else
       @_workers[key].resume()
 
+  _multi: ->
+    @_kv.redis.multi()
+
+  # on pop job key
+  #   record which queue for key
+  #   if dirty flag
+  #     enqueue key
+  #   else
+  #     as multi
+  #       set self as owner
+  #       add to set of owned jobs
+  #       when done
+  #         start running job
+  _handle_job_key: (queue, key) ->
+    @_queue_for_key[key] = queue
+    if @_dirty
+      @_enqueue_job queue, key
+    else
+      @_multi()
+        .set('_lock:'+key, @queue_id)
+        .sadd('_active:'+@queue_id, key)
+        .exec (err) =>
+          return @error err if err
+          @_run key
+    @_watch_queues()
+
+  # on pop dirty
+  #   set dirty flag
+  #   delete info key
+  #   look up targets
+  #     push targets to dirty
+  #       reset dirty timeout
+  #       call popper
+  _handle_dirty_key: (key) ->
+    @_dirty = true
+    @_kv.del '_lock:'+key
+    @_kv.smembers '_targets:'+key, (err, targets) =>
+      return @error err if err
+      @_out_queue.rpush_all 'dirty', targets, (err) =>
+        @error err if err
+        @_reset_dirty_timeout()
+        @_watch_queues()
+
+  # on enqueue key
+  #   acquire lock
+  #     if success
+  #       push key to job queue
+  _enqueue_job: (queue, key) ->
+    @_kv.setnx '_lock:'+key, 'queued', (err, set) =>
+      return @error err if err
+      return unless set
+      @_out_queue.lpush queue, key
+
+  # on dirty timeout
+  #   clear dirty flag
+  #   take/clear all pending keys
+  #   get info for pending keys
+  #     for each key
+  #       if missing
+  #         enqueue key
+  #       else
+  #         next tick: call resume function
+  _post_dirty: ->
+    @_dirty = false
+    @_kv.del '_dirty'
+    pending = @_pending
+    @_pending = []
+    keys = _.map pending, (p) -> '_lock:'+p
+    @_kv.mget keys, (err, locks) ->
+      return @error err if err
+      for lock, i in locks
+        f = @_resume_for_key keys[i]
+        delete @_resume_for_key keys[i]
+        if lock
+          process.nextTick f
+        else
+          @_enqueue_job @_queue_for_key[key], key
+
+
+  _reset_dirty_timeout: ->
+    clearTimeout @_dirty_timeout
+    @_dirty_timeout = setTimeout (=> @_post_dirty()), 1000
+
   _handle:
+
+    dirty: ->
+      @_dirty = true
+      @_reset_dirty_timeout()
 
     ready: (key) ->
       clearTimeout @_cycle_timeouts[key]


  # Run the work queue, calling the given callback on completion
  run: (@callback) ->
    @_connect()

  # Add an accessor that @gets an input
  input: (prefix, params...) ->

    @_watch_queues()

  _watch_queues: ->
    @_in_queue.pop_any 'dirty', @_job_queues..., (err, response) =>
      return @error err if err
      [type, value] = response
      if type == 'dirty'
        @_handle_dirty_key value
      else
        @_handle_job_key type, value

  # cleanup procedure
  #   find keys which are done, but which have no targets
  #     if none, return
  #     delete them and their state
  #     get/delete their sources
  #     remove key from each source's target
  #   repeat
  cleanup: ->
    # TODO

  # Send a log message over redis pubsub
  log: (key, label, payload) ->
