anchor.diff:  trie_delete_path([{NodeId, Word, _}|RestPath]) ->
       Edge = #trie_edge{node_id=NodeId, word=Word},
       mnesia:delete({vmq_trie, Edge}),
  -    case mnesia:wread(vmq_trie_node, NodeId) of
  +    case mnesia:wread({vmq_trie_node, NodeId}) of
           [#trie_node{edge_count=1, topic=undefined}] ->
               mnesia:delete({vmq_trie_node, NodeId}),
               trie_delete_path(RestPath);
  
positive.code:     {Mega, Sec, _} = os:timestamp(),
      (Mega * 1000000 + Sec).
  
  maybe_unregister_multi(MRef, #state{unreg_time=TRef, unreg_queue=UQueue} = State) ->
      cancel_timer(TRef),
      case UQueue =< 1000 of
          true ->
              State#state{unreg_time=erlang:send_after(500, self(), unreg), unreg_queue=[MRef|UQueue]};
          false ->
              unregister_abnormal([MRef|UQueue]),
              State#state{unreg_time=undefined, unreg_queue=[]}
      end.
  
  unregister_abnormal(MRefs) ->
      %% should only happen if a session does not propperly unregister
      Ret = transaction(fun() -> unregister_abnormal(MRefs, []) end),
      unregister_abnormal_cleanup(Ret).
  
  unregister_abnormal([MRef|Rest], Acc) ->
      Ret =
      case mnesia:index_read(vmq_session, MRef, #session.monitor) of
          [#session{client_id=ClientId, clean=true}] ->
              mnesia:delete({vmq_session, ClientId}),
              del_subscriber_tx('_', ClientId),
              {true, ClientId};
          [#session{client_id=ClientId, clean=false, vclock=VClock} = Obj] ->
              mnesia:write(vmq_session, Obj#session{pid=undefined,
                                                    monitor=undefined,
                                                    last_seen=epoch(),
                                                    vclock=unsplit_vclock:increment(node(), VClock)}, write),
              {false, 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).
  
  transaction(TxFun) ->
      %% Making this a sync_transaction allows us to use dirty_read
  
negative.code: -spec register_client__(pid(),client_id(),flag()) -> ok | {error, timeout}.
  register_client__(ClientPid, ClientId, CleanSession) ->
      {Ret, MaybeMonitor} =
      case ets:lookup(vmq_session, ClientId) of
          [] ->
              {new_client, undefined};
          [#session{pid=OldPid, monitor=OldRef, last_seen=LastSeen}]
  