   _run: (key) ->
     try
       @_worker_count++
-      @_kv.del "sources:#{key}", "targets:#{key}"
       @_workers[key] = new Worker key, this
       @_workers[key].run()
     catch e


  # 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

    @_worker_count = 0
    @_params = {}
    @_is_input = {}
    @_pending = []
    @_resume_for_key = {}
    @_as = {}
    @_listeners = {}
    @_triggers = {}
    @_cycle_timeouts = {}
    @_queue_for_key = {}

  _random_queue_id: ->
    '' + Math.random()

  _connect: (callback) ->
    @_kv.connect =>
      @_in_queue.connect =>
        @_out_queue.connect =>
          @_pubsub.connect =>
            @_worker_kv.connect =>
              @_worker_pubsub.connect, =>
                callback?()

  # Subscribe to channels and queue events
