anchor.diff:  
   handle_event({stats, StatsBucket, #stat_entry{timestamp = TS, values = Values}},
                #state{bucket_name = StatsBucket,
  -                    tap_info = TapInfo,
                       dcp_info = DcpInfo,
                       last_ts = LastTS} = State) ->
  -    TapBacklogSize =
  -        case {orddict:find(ep_tap_replica_queue_backfillremaining, Values),
  -              orddict:find(ep_tap_replica_qlen, Values)} of
  -            {{ok, BackfillRemaining}, {ok, Qlen}} ->
  -                {ok, BackfillRemaining + Qlen};
  -            _ ->
  -                false
  -        end,
  -
  -    NewTapInfo =
  -         new_safeness_info(TapBacklogSize,
  -                           orddict:find(ep_tap_replica_queue_drain, Values),
  -                           TS,
  -                           LastTS,
  -                           TapInfo),
  -
       NewDcpInfo =
            new_safeness_info(orddict:find(ep_dcp_replica_items_remaining, Values),
                              orddict:find(ep_dcp_replica_items_sent, Values),
  
positive.code: new_safeness_info(_, _, _, _, Info) ->
      Info.
  
  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},
                                dcp_info = #safeness_info{safeness_level = DcpLevel}} = State) ->
      Now = misc:time_to_epoch_ms_int(now()),
      RV = case Now - UpdateTS > ?STALENESS_THRESHOLD of
               true ->
                   stale;
               _ ->
                   pick_level(TapLevel, DcpLevel)
           end,
      {ok, {ok, RV}, State};
  handle_call(get_state, State) ->
  
negative.code: 
  -record(state,
          {bucket_name :: bucket_name(),
           dcp_info :: #safeness_info{},
           last_ts :: undefined | non_neg_integer(),
           last_update_local_clock :: non_neg_integer()
  