anchor.diff:      self() ! async_init,
       mqtt_sub_repo:claim_filter(Filter,self()),
       Seq = 0,
  -    {ok,#state{filter = Filter,
  -               live_subs = dict:new(),
  -               queue = shared_queue:new(Seq),
  -               garbage = ?MONOID:id()}}.
  +    {ok,#state{filter         = Filter,
  +               live_subs      = dict:new(),
  +               queue          = shared_queue:new(Seq),
  +               client_offsets = min_val_tree:new(),
  +               garbage        = ?MONOID:id()}}.
   
   %%--------------------------------------------------------------------
   %% @private
  
positive.code:                                live_subs = Subs,
                                 queue = Q}) ->
      SQ1 = shared_queue:pushr(Packet,Q),
      %%NewSeq = shared_queue:max_offset(SQ1),
      {SendTo,Subs1} = dict:fold(update_waiting_subs(SQ1),{[],Subs},Subs),
      error_logger:info_msg("About to send packet ~p for filter ~p to ~p out of subs ~p~n",[Packet,Filter,SendTo,dict:to_list(Subs)]),
      {SendTo,S#state{live_subs = Subs1,queue = SQ1}}.
  
negative.code: 
  process_unsub(ClientId,S = #state{live_subs = Subs,
                                   queue = SQ,
                                   garbage = Garbage,
                                   client_offsets = Offsets}) ->
      Offsets1 = min_val_tree:remove(ClientId,Offsets),
      MinVal = min_val_tree:min(Offsets),
      {MoreGarbage,SQ1} = shared_queue:truncate(MinVal,SQ),
      S#state{live_subs = dict:erase(ClientId,Subs),
              queue = SQ1,
              client_offsets = Offsets1,
              garbage = ?MONOID:as(Garbage,MoreGarbage)}.
  
  handle_resume({ClientId,CSeq,QoS,ResumeFrom,WSize},Pid,S = #state{filter = Filter,
  