anchor.diff:  
   init([BucketName]) ->
       {ok, #state{bucket_name = BucketName,
  +                tap_info = #safeness_info{},
  +                upr_info = #safeness_info{},
                   last_update_local_clock = misc:time_to_epoch_ms_int(now())}}.
   
   terminate(_Reason, _State)     -> ok.
   code_change(_OldVsn, State, _) -> {ok, State}.
   
  -handle_event({stats, StatsBucket, #stat_entry{timestamp = TS,
  -                                              values = Values}},
  -             #state{bucket_name = StatsBucket} = State) ->
  -    case {orddict:find(ep_tap_replica_total_backlog_size, Values),
  -          orddict:find(ep_tap_replica_queue_drain, Values)} of
  -        {{ok, BacklogSize},
  -         {ok, DrainRate}} ->
  -            handle_stats_sample(BacklogSize, DrainRate, TS, State);
  -        _ ->
  -            {ok, mark_update(State#state{last_ts = TS})}
  -    end;
  +handle_event({stats, StatsBucket, #stat_entry{timestamp = TS} = Stats},
  +             #state{bucket_name = StatsBucket,
  +                    tap_info = TapInfo,
  +                    upr_info = UprInfo,
  +                    last_ts = LastTS} = State) ->
  +    NewTapInfo =
  +         new_safeness_info(ep_tap_replica_total_backlog_size,
  +                           ep_tap_replica_queue_drain,
  +                           Stats,
  +                           LastTS,
  +                           TapInfo),
  +
  +    NewUprInfo =
  +         new_safeness_info(ep_upr_replica_items_remaining,
  +                           ep_upr_replica_items_sent,
  +                           Stats,
  +                           LastTS,
  +                           UprInfo),
  +
  +    {ok, State#state{last_ts = TS,
  +                     tap_info = NewTapInfo,
  +                     upr_info = NewUprInfo,
  +                     last_update_local_clock = misc:time_to_epoch_ms_int(now())}};
   
   handle_event(_, State) ->
       {ok, State}.
  
positive.code: -export([init/1, handle_event/2, handle_call/2,
           handle_info/2, terminate/2, code_change/3]).
  
  -record(state,
          {bucket_name :: bucket_name(),
           backlog_size :: undefined | number(),
           drain_rate :: undefined | number(),
           safeness_level = unknown :: yellow | green | unknown,
           last_ts :: undefined | non_neg_integer(),
           last_update_local_clock :: non_neg_integer()
          }).
  
negative.code:             end
      end.
  
  new_safeness_info(BacklogSizeStat, DrainRateStat,
                    #stat_entry{timestamp = TS,
                                values = Values},
                    LastTS,
                    #safeness_info{backlog_size = OldBacklogSize,
                                   drain_rate = OldDrainRate,
                                   safeness_level = SafenessLevel} = Info) ->
      case {orddict:find(BacklogSizeStat, Values),
            orddict:find(DrainRateStat, Values)} of
          {{ok, BacklogSize},
           {ok, DrainRate}} ->
  
              NewBacklogSize = average_value(OldBacklogSize, BacklogSize,
                                             LastTS, TS),
              NewDrainRate = average_value(OldDrainRate, DrainRate,
                                           LastTS, TS),
              NewSafenessLevel = new_safeness_level(SafenessLevel, NewBacklogSize, NewDrainRate),
  
              #safeness_info{backlog_size = NewBacklogSize,
                             drain_rate = NewDrainRate,
                             safeness_level = NewSafenessLevel};
          _ ->
              Info
      end.
  
  pick_level(unknown, _) ->
      unknown;
  pick_level(_, unknown) ->
      unknown;
  pick_level(green, green) ->
      green;
  pick_level(_, _) ->
      yellow.
  
  handle_call(get_level, #state{last_update_local_clock = UpdateTS,
                                tap_info = #safeness_info{safeness_level = TapLevel},
                                upr_info = #safeness_info{safeness_level = UprLevel}} = State) ->
      Now = misc:time_to_epoch_ms_int(now()),
      RV = case Now - UpdateTS > ?STALENESS_THRESHOLD of
               true ->
                   stale;
               _ ->
                   pick_level(TapLevel, UprLevel)
           end,
      {ok, {ok, RV}, State};
  handle_call(get_state, State) ->
  