anchor.diff:                                              end),
       erlang:start_timer(30000, self(), log_tap_stats),
   
  -    {ok, BucketConfig} = ns_bucket:get_bucket(Bucket),
  -    BucketType = ns_bucket:bucket_type(BucketConfig),
  -
       {ok, #state{bucket=Bucket,
  -                bucket_type=BucketType,
                   disco_events_subscription=Subscription,
                   initial_counts=count_moves(MoveDict),
                   max_per_node=?MAX_MOVES_PER_NODE,
  
positive.code:     maybe_terminate(spawn_workers(State));
  handle_info({move_done, {Node, VBucket, OldChain, [NewNode|_] = NewChain}},
              #state{movers=Movers,
                     bucket=Bucket, bucket_type=BucketType} = State) ->
      master_activity_events:note_move_done(Bucket, VBucket),
      %% Update replication
      update_replication_post_move(VBucket, OldChain, NewChain),
  
negative.code: -type progress_callback() :: fun((dict()) -> any()).
  
  -record(state, {bucket::nonempty_string(),
                  disco_events_subscription::pid(),
                  initial_counts::dict(),
                  max_per_node::pos_integer(),
  