anchor.diff:           get_meta/2, update_meta/3]).	 
   
   % @type riak_ring(). The opaque data type used for partition ownership.
  --record(hstate, {
  -    nodename, % the Node responsible for this hstate
  -	vclock,   % for this hstate object, entries are {Node, Ctr}
  -	ring,     % chash ring of {IndexAsInt, Node} mappings
  +-record(chstate, {
  +    nodename, % the Node responsible for this chstate
  +    vclock,   % for this chstate object, entries are {Node, Ctr}
  +    chring,   % chash ring of {IndexAsInt, Node} mappings
       meta      % dict of cluster-wide other data (primarily bucket N-value, etc)
   }). 
   
  -% @type meta_entry(). Record for each entry in #hstate.meta
  +% @type meta_entry(). Record for each entry in #chstate.meta
   -record(meta_entry, {
       value,    % The value stored under this entry
       lastmod   % The last modified time of this entry, 
  -              % from calendar:datetime_to_gregorian_seconds(calendar:universal_time()), 
  +              %  from calendar:datetime_to_gregorian_seconds(
  +              %                             calendar:universal_time()), 
   }).
   
   % @doc This is used only when this node is creating a brand new cluster.
  -% @spec fresh() -> hstate()
  +% @spec fresh() -> chstate()
   fresh() ->
       % use this when starting a new cluster via this node
       fresh(node()).
   
   % @doc Equivalent to fresh/0 but allows specification of the local node name.
   %      Called by fresh/0, and otherwise only intended for testing purposes.
  -% @spec fresh(NodeName :: term()) -> hstate()
  +% @spec fresh(NodeName :: term()) -> chstate()
   fresh(NodeName) ->
       fresh(riak:get_app_env(ring_creation_size), NodeName).
   
   % @doc Equivalent to fresh/1 but allows specification of the ring size.
   %      Called by fresh/1, and otherwise only intended for testing purposes.
  -% @spec fresh(RingSize :: integer(), NodeName :: term()) -> hstate()
  +% @spec fresh(RingSize :: integer(), NodeName :: term()) -> chstate()
   fresh(RingSize, NodeName) ->
  -    #hstate{nodename=NodeName,
  +    #chstate{nodename=NodeName,
   	    vclock=vclock:fresh(),
  -	    ring=chash:fresh(RingSize, NodeName),
  +	    chring=chash:fresh(RingSize, NodeName),
               meta=dict:new()}.
   
   % @doc Return all partition indices owned by the node executing this function.
  -% @spec my_indices(State :: hstate()) -> [integer()]
  +% @spec my_indices(State :: chstate()) -> [integer()]
   my_indices(State) ->
       [I || {I,Owner} <- ?MODULE:all_owners(State), Owner =:= node()].
   
   % @doc Return a partition index not owned by the node executing this function.
   %      If this node owns all partitions, return any index.
  -% @spec random_other_index(State :: hstate()) -> integer()
  +% @spec random_other_index(State :: chstate()) -> integer()
   random_other_index(State) ->
       L = [I || {I,Owner} <- ?MODULE:all_owners(State), Owner =/= node()],
       case L of
  
positive.code:     end.
  
  % @doc Transfer ownership of partition at Idx to Node.
  % @spec transfer_node(Idx :: integer(), Node :: term(), MyState :: hstate()) ->
  %           hstate()
  transfer_node(Idx, Node, MyState) ->
      case chash:lookup(Idx, MyState#hstate.ring) of
  	Node ->
  	    MyState;
  	_ ->
  	    Me = MyState#hstate.nodename,
  	    VClock = vclock:increment(Me, MyState#hstate.vclock),
  	    Ring = chash:update(Idx, Node, MyState#hstate.ring),
  	    #hstate{nodename=Me,vclock=VClock,ring=Ring,
                      meta=MyState#hstate.meta}
      end.
  
  ancestors(RingStates) ->
      Ancest = [[O2 || O2 <- RingStates,
       vclock:descends(O1#hstate.vclock,O2#hstate.vclock),
       (vclock:descends(O2#hstate.vclock,O1#hstate.vclock) == false)]
  		|| O1 <- RingStates],
      lists:flatten(Ancest).
  
  % @doc Incorporate another node's state into our view of the Riak world.
  % @spec reconcile(ExternState :: hstate(), MyState :: hstate()) ->
  %       {no_change, hstate()} | {new_ring, hstate()}
  reconcile(ExternState, MyState) ->
      case vclock:equal(MyState#hstate.vclock, vclock:fresh()) of
          true -> 
              {new_ring, #hstate{nodename=MyState#hstate.nodename,
                                 vclock=ExternState#hstate.vclock,
                                 ring=ExternState#hstate.ring,
                                 meta=ExternState#hstate.meta}};
          false ->
              case ancestors([ExternState, MyState]) of
                  [OlderState] ->
                      case vclock:equal(OlderState#hstate.vclock,
                                        MyState#hstate.vclock) of
                          true ->
                              {new_ring, #hstate{nodename=MyState#hstate.nodename,
                                                 vclock=ExternState#hstate.vclock,
                                                 ring=ExternState#hstate.ring,
                                                 meta=ExternState#hstate.meta}};
                          false -> {no_change, MyState}
                      end;
                  [] -> 
                      case equal_rings(ExternState,MyState) of
                          true -> {no_change, MyState};
                          false -> {new_ring, reconcile(MyState#hstate.nodename,
                                                        ExternState, MyState)}
                      end
              end
      end.
  
  % @private
  equal_rings(_A=#hstate{ring=RA,meta=MA},_B=#hstate{ring=RB,meta=MB}) ->
      MDA = lists:sort(dict:to_list(MA)),
      MDB = lists:sort(dict:to_list(MB)),
      case MDA =:= MDB of
  
negative.code:     
  
  % @doc Return a value from the cluster metadata dict
  % @spec get_meta(Key :: term(), State :: chstate()) -> 
  %    {ok, term()} | error | undefined
  get_meta(Key, State) -> 
      case dict:find(Key, State#chstate.meta) of
          error -> undefined;
          {ok, M} -> {ok, M#meta_entry.value}
      end.
  
                  
  % @doc Set a key in the cluster metadata dict
  % @spec update_meta(Key :: term(), Val :: term(), State :: chstate()) ->
  %     State :: chstate() | error
  update_meta(Key, Val, State) ->
      M = #meta_entry { 
          lastmod = calendar:datetime_to_gregorian_seconds(
                      calendar:universal_time()),
          value = Val
      },    
      VClock = vclock:increment(State#chstate.nodename, State#chstate.vclock),
      State#chstate{vclock=VClock, meta=dict:store(Key, M, State#chstate.meta)}.
  
  sequence_test() ->
      I1 = 365375409332725729550921208179070754913983135744,
      I2 = 730750818665451459101842416358141509827966271488,
      A = fresh(4,a),
      B1 = A#chstate{nodename=b},
      B2 = transfer_node(I1, b, B1),
      ?assertEqual(B2, transfer_node(I1, b, B2)),
      {new_ring, A1} = reconcile(B1,A),
      C1 = A#chstate{nodename=c},
      C2 = transfer_node(I1, c, C1),
      {new_ring, A2} = reconcile(C2,A1),
      {new_ring, A3} = reconcile(B2,A2),
  