anchor.diff:  epoch() ->
       {Mega, Sec, _} = os:timestamp(),
       (Mega * 1000000 + Sec).
  +
  +unregister_abnormal(MRefs) ->
  +    %% should only happen if a session does not propperly unregister
  +    {atomic, Ret} = mnesia:transaction(fun unregister_abnormal/2, [MRefs, []]),
  +    unregister_abnormal_cleanup(Ret).
  +
  +unregister_abnormal([MRef|Rest], Acc) ->
  +    Ret =
  +    case mnesia:match_object(vmq_session, #session{monitor=MRef, _='_'}, read) of
  +        [#session{client_id=ClientId, clean=CleanSession} = Obj] ->
  +            unregister_client_tx(Obj),
  +            {CleanSession, ClientId};
  +        [] ->
  +            {error, not_found}
  +    end,
  +    unregister_abnormal(Rest, [Ret|Acc]);
  +unregister_abnormal([], Ret) -> Ret.
  +
  +unregister_abnormal_cleanup([{true, ClientId}|Rest]) ->
  +    vmq_msg_store:clean_session(ClientId),
  +    unregister_abnormal_cleanup(Rest);
  +unregister_abnormal_cleanup([{false, _ClientId}|Rest]) ->
  +    vmq_systree:incr_inactive_clients(),
  +    unregister_abnormal_cleanup(Rest);
  +unregister_abnormal_cleanup([{error, not_found}|Rest]) ->
  +    unregister_abnormal_cleanup(Rest);
  +unregister_abnormal_cleanup([]) -> ok.
  +
  +cancel_timer(undefined) -> ok;
  +cancel_timer(TRef) -> erlang:cancel_timer(TRef).
  
positive.code: -hook({on_subscribe, all, 3}).
  -hook({on_unsubscribe, all, 3}).
  -hook({filter_subscribers, every, 5}).
  
  -record(topic, {name, node}).
  -record(trie, {edge, node_id, vclock=unsplit_vclock:fresh()}).
  
negative.code:                                     end
                              end, Sessions)
        end),
      {ok, #state{}}.
  
  -spec handle_call(_, _, []) -> {reply, ok, []}.
  handle_call({monitor, ClientPid, MaybeMonitor}, _From, State) ->
  