anchor.diff:      {cy_cache, [node:node_type()]} |
       {clustering_shuffle, comm:mypid(), dc_centroids:centroids(), non_neg_integer()} |
       {clustering_shuffle_reply, comm:mypid(), dc_centroids:centroids(), non_neg_integer()} |
  -    {query_clustering, comm:mypid()}).
  +    {query_clustering, comm:mypid()} |
  +    {query_epoch, comm:mypid()}
  +).
   
   %% @doc Sends an initialization message to the node's dc_clustering process.
   -spec activate() -> ok | ignore.
  
positive.code: init({ResetTrigger, ClusterTrigger}) ->
      ResetTriggerState = trigger:init(ResetTrigger, fun get_clustering_reset_interval/0, reset_clustering),
      ClusterTriggerState = trigger:init(ClusterTrigger, fun get_clustering_interval/0, start_clustering_shuffle),
      {inactive, msg_queue:new(), ResetTriggerState, ClusterTriggerState}.
  
  %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
  % Message Loop
  
negative.code:                 ignore -> ok;
                  R -> comm:send(R, {clustering_shuffle_reply, comm:this(), Centroids, Epoch})
              end,
              {Epoch, cluster(Centroids, RemoteCentroids, Radius)}
      end,
      case NewEpoch of % reset needed?
          Epoch -> State#state_active{centroids=NewCentroids};
          _Else ->
              NewResetTriggerState = trigger:next(ResetTriggerState),
              NewClusterTriggerState = trigger:next(ClusterTriggerState),
              % Note: NewEpoch - 1 because we will increment it in the handler for the
              % vivaldi reply
              State#state_active{
                  centroids=NewCentroids
                  , local_epoch=NewEpoch - 1
                  , reset_trigger_state=NewResetTriggerState
                  , cluster_trigger_state=NewClusterTriggerState
              }
      end
      .
  