anchor.diff:  
   handle_info({cache_remote_cluster, UUID, RemoteCluster0}, State) ->
       RemoteCluster = last_cache_request(cache_remote_cluster, UUID, RemoteCluster0),
  -
  -    true = ets:insert(?CACHE, {UUID, RemoteCluster}),
  -
  -    [{clusters, Clusters}] = ets:lookup(?CACHE, clusters),
  -    NewClusters = ordsets:add_element(UUID, Clusters),
  -    true = ets:insert(?CACHE, {clusters, NewClusters}),
  -
  -    ets:insert_new(?CACHE, {{buckets, UUID}, []}),
  -    {noreply, maybe_schedule_cluster_config_update(RemoteCluster, State)};
  -handle_info({cache_remote_bucket, {UUID, Bucket} = Id, RemoteBucket0}, State) ->
  -    [{_, Buckets}] = ets:lookup(?CACHE, {buckets, UUID}),
  -    NewBuckets = ordsets:add_element(Bucket, Buckets),
  -    true = ets:insert(?CACHE, {{buckets, UUID}, NewBuckets}),
  -
  -    RemoteBucket = last_cache_request(cache_remote_bucket, Id, RemoteBucket0),
  -    true = ets:insert(?CACHE, {{bucket, UUID, Bucket}, RemoteBucket}),
  -
  -    {noreply, State};
  +    {noreply, cache_remote_cluster(UUID, RemoteCluster, State)};
  +handle_info({remote_bucket_request_result, TargetNode, {UUID, Bucket} = Id, R},
  +            #state{remote_bucket_requests=Requests,
  +                   remote_bucket_waiters=Waiters,
  +                   remote_bucket_waiters_trefs=TRefs} = State) ->
  +    BucketWaiters = misc:dict_get(Id, Waiters, []),
  +    BucketTRef = misc:dict_get(Id, TRefs, make_ref()),
  +
  +    BucketRequests0 = dict:fetch(Id, Requests),
  +    BucketRequests1 = dict:erase(TargetNode, BucketRequests0),
  +    Requests1 = case dict:size(BucketRequests1) of
  +                    0 ->
  +                        dict:erase(Id, Requests);
  +                    _ ->
  +                        dict:store(Id, BucketRequests1, Requests)
  +                end,
  +
  +    {NewState0, Reply} =
  +        case R of
  +            {ok, {RemoteCluster, RemoteBucket}} ->
  +                State1 = cache_remote_cluster(UUID, RemoteCluster, State),
  +                State2 = cache_remote_bucket(Id, RemoteBucket, State1),
  +
  +                {State2, {ok, RemoteBucket}};
  +            {error, Type, _} when Type =:= not_present;
  +                                  Type =:= not_capable ->
  +                {State, R};
  +            Error ->
  +                ?log_error("Failed to grab remote bucket `~s`: ~p", [Bucket, Error]),
  +                case dict:size(BucketRequests1) of
  +                    0 ->
  +                        %% no more ongoing request so we just return an error to
  +                        %% the caller
  +                        Msg = io_lib:format("Failed to grab remote bucket `~s`"
  +                                            "from any of known nodes", [Bucket]),
  +
  +                        {State, {error, all_nodes_failed, iolist_to_binary(Msg)}};
  +                    _ ->
  +                        %% let's try to wait for the other requests to return
  +                        {State, undefined}
  +                end
  +        end,
  +
  +    NewState1 = NewState0#state{remote_bucket_requests=Requests1},
  +    NewState2 = case Reply of
  +                    undefined ->
  +                        NewState1;
  +                    _ ->
  +                        cancel_bucket_request_tref(Id, BucketTRef),
  +
  +                        lists:foreach(
  +                          fun ({_, Waiter}) ->
  +                                  gen_server:reply(Waiter, Reply)
  +                          end, BucketWaiters),
  +
  +                        Waiters1 = dict:erase(Id, Waiters),
  +                        TRefs1 = dict:erase(Id, TRefs),
  +                        %% we don't drop other requests to prevent overloading of
  +                        %% possibly busy nodes
  +                        NewState1#state{remote_bucket_waiters=Waiters1,
  +                                        remote_bucket_waiters_trefs=TRefs1}
  +                end,
  +
  +    {noreply, NewState2};
  +handle_info({remote_bucket_request_timeout, Id},
  +            #state{remote_bucket_waiters=Waiters,
  +                   remote_bucket_waiters_trefs=TRefs} = State) ->
  +    Now = misc:time_to_epoch_ms_int(os:timestamp()),
  +
  +    BucketWaiters = dict:fetch(Id, Waiters),
  +    BucketTRef = dict:fetch(Id, TRefs),
  +
  +    %% tref must already canceled
  +    false = erlang:read_timer(BucketTRef),
  +
  +    [{_, From} | BucketWaiters1] = BucketWaiters,
  +    gen_server:reply(From, {error, timeout}),
  +
  +    NewState =
  +        case BucketWaiters1 of
  +            [] ->
  +                Waiters1 = dict:erase(Id, Waiters),
  +                TRefs1 = dict:erase(Id, TRefs),
  +
  +                State#state{remote_bucket_waiters=Waiters1,
  +                            remote_bucket_waiters_trefs=TRefs1};
  +            [{Expiration, _} | _] ->
  +                Waiters1 = dict:store(Id, BucketWaiters1, Waiters),
  +
  +                BucketTRef1 = erlang:send_after(max(Expiration - Now, 0), self(),
  +                                                {remote_bucket_request_timeout, Id}),
  +                TRefs1 = dict:store(Id, BucketTRef1, TRefs),
  +
  +                State#state{remote_bucket_waiters=Waiters1,
  +                            remote_bucket_waiters_trefs=TRefs1}
  +        end,
  +
  +    {noreply, NewState};
   handle_info(gc, #state{cache_path=CachePath} = State) ->
       gc(),
       dump_table(?CACHE, CachePath),
  
positive.code:             end
      end.
  
  remote_cluster_and_bucket(#remote_cluster{nodes=Nodes,
                                            uuid=UUID},
                            BucketStr, Username, Password) ->
      Bucket = list_to_binary(BucketStr),
      ShuffledNodes = misc:shuffle(Nodes),
  
      remote_bucket_from_nodes(ShuffledNodes, Bucket, Username, Password, UUID).
  
  remote_bucket_from_nodes([], _, _, _, _) ->
      {error, all_nodes_failed,
       <<"Failed to grab remote bucket info from any of known nodes">>};
  remote_bucket_from_nodes([Node | Rest], Bucket, Username, Password, UUID) ->
      R = remote_bucket(Node, Bucket, Username, Password, UUID),
  
      case R of
          {ok, _} ->
              R;
          {error, Type, _} when Type =:= not_present;
                                Type =:= not_capable ->
              R;
          Other ->
              ?log_info("Failed to get bucket info from node ~p:~n~p", [Node, Other]),
              remote_bucket_from_nodes(Rest, Bucket, Username, Password, UUID)
      end.
  
  remote_bucket(#remote_node{host=Host, port=Port},
                Bucket, Username, Password, UUID) ->
      Creds = {Username, Password},
  
      JsonGet = mk_json_get(Host, Port, Username, Password),
  
negative.code:                 FoundCluster
          end,
  
      State1 = enqueue_waiter({UUID, Bucket}, From, Timeout, State),
      State2 = request_remote_bucket(RemoteCluster, Bucket,
                                     Username, Password, State1),
  
      {noreply, State2};
  handle_call(Request, From, State) ->
      ?log_warning("Got unexpected call request: ~p", [{Request, From}]),
      {reply, unhandled, State}.
  