anchor.diff:      ok = wait_until_ring_converged(Nodes),
       ok = wait_until_master_ready(MainNode).
   
  +%% @doc Ensure that all nodes are the sole owner of their rings
  +-spec check_nodes_own_their_ring(list(atom())) -> ok | {error, atom()}.
  +check_nodes_own_their_ring([]) -> ok;
  +check_nodes_own_their_ring([H | T]) ->
  +    case sorted_ring_owners(H) of
  +        {ok, [H]} ->
  +            check_nodes_own_their_ring(T);
  +        Reason ->
  +            {error, H, Reason}
  +    end.
  +
   %% @doc Retrieve a list of ring-owning physical nodes according to the MainNode
   %%
   %%      A node is ring-owning if a partition is stored on it
   %%
  +-spec sorted_ring_owners(node()) -> {ok, list(node())} | {badrpc, term()}.
  +sorted_ring_owners(Node) ->
  +    case rpc:call(Node, riak_core_ring_manager, get_raw_ring, []) of
  +        {ok, Ring} ->
  +            Owners = [Owner || {_Idx, Owner} <- rpc:call(Node, riak_core_ring, all_owners, [Ring])],
  +            SortedOwners = lists:usort(Owners),
  +            io:format("Owners at ~p: ~p~n", [Node, SortedOwners]),
  +            {ok, SortedOwners};
  +
  +        {badrpc, _}=BadRpc ->
  +            BadRpc
  +    end.
  +
   -spec owners_according_to(node(), node()) -> {ok, list(node())} | {badrpc, term()}.
   owners_according_to(Node, MainNode) ->
       case rpc:call(Node, riak_core_ring_manager, get_raw_ring, []) of
  
positive.code: %% @doc Build clusters out of the given node list
  -spec join_cluster(list(atom())) -> ok.
  join_cluster([MainNode | OtherNodes] = Nodes) ->
      %% Ensure nodes have non-overlapping rings according to the master node,
      %% also, each node should own its entire ring
      true = lists:all(fun(Node) ->
          case owners_according_to(Node, MainNode) of
              {ok, [Node]} ->
                  true;
              _ ->
                  false
          end
      end, Nodes),
  
      ok = case OtherNodes of
          [] ->
              %% no other nodes, nothing to join/plan/commit
              ok;
  
          _ ->
              %% Do a plan/commit staged join, instead of sequential joins
              ok = lists:foreach(fun(N) -> request_join(N, MainNode) end, OtherNodes),
              ok = wait_plan_ready(MainNode),
              ok = commit_plan(MainNode),
              ok = try_cluster_ready(Nodes)
      end,
  
      ok = wait_until_nodes_ready(Nodes),
  
  
negative.code:         error ->
              usage();
  
          {ok, [_SingleNode]} ->
              io:format("Single-node cluster, nothing to join"),
              halt();
  
          {ok, Nodes} ->
              io:format("Starting clustering of nodes ~p~n", [Nodes]),
              lists:foreach(fun(N) -> erlang:set_cookie(N, antidote) end, Nodes),
  