     @_worker_count--
     delete @_workers[key]
 
-  finish_key: (key, state) ->
-    @_kv.get "info:#{key}", (err, info) =>
-      return @error err if err
-      if info.state == 'working'
-        @_kv.set "info:#{key}", {state}, (err) ->
-          @error err if err
-          @_pubsub.publish @_control_ns, "ready|#{key}"
+  # on job finished
+  #   if dirty flag
+  #     make key pending, with resume to finish job again
+  #   else
+  #     getset sources for key
+  #       find - and + diffs
+  #       as multi
+  #         for each diff
+  #           apply diff to target sets
+  #         set state to done
+  #         remove from our owned jobs
+  #         on completion
+  #           publish done message
+  finish_key: (key) ->
+    if @_dirty
+      @_resume_for_key[key] = =>
+        @finish_key key
+      @_pending.push key
+    else
+      new_sources = @_workers[key].dependencies
+      @_kv.getset '_sources:'+key, new_sources, (err, old_sources) =>
+        return @error err if err
+        [adds, dels] = @_array_diff old_sources, new_sources
+        m = @_multi()
+        for source in dels
+          m.sdel '_targets:'+source, key
+        for source in adds
+          m.sadd '_targets:'+source, key
+        m.set '_lock:'+key, 'done'
+        m.sdel '_active:'+@queue_id, key
+        m.exec (err) =>
+          return @error err if err
+          @_pubsub.publish @_control_ns, 'ready|'+key
+
+  _array_diff: (a, b) ->
+    a = _.clone(a).sort()
+    b = _.clone(b).sort()
+    add = []
+    del = []
+    while a.length && b.length
+      if a[0] == b[0]
+        a.shift(); b.shift()
+      else if a[0] < b[0]
+        del.push a.shift()
+      else
+        add.push b.shift()
+    [add.concat(b), del.concat(a)]
 
   watch_for_cycle: (key) ->
     @_cycle_timeouts[key] = setTimeout (=> @_check_for_cycle key), @_cycle_timeout


  # Register the 'next' event, and listen for 'resume' messages.
  constructor: (@options) ->
    @_workers = {}
    @_mixins = {}
    @_runners = {}
    @_kv = db.key_value @options
    @_in_queue = db.queue @options
    @_out_queue = db.queue @options


  # Run the work queue, calling the given callback on completion
  run: (@callback) ->
    @_connect =>
      @_kv.get '_dirty', (err, val) =>
        return @error err if err
        @perform 'dirty' if val
        @_listen()

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