anchor.diff:  
   -spec common_change_callback(Seq :: non_neg_integer(), datastore:document() | stream_ended, model_behaviour:model_type() | undefined) -> ok.
   common_change_callback(_Seq, stream_ended, _Type) ->
  -    Seq = worker_host:state_get(?MODULE, last_seq),
  -    ?error("Changes stream broken - restarting at ~p", [Seq]),
  -    start_changes_stream(Seq),
  +    CurrentSeq = newest_cached_seq(),
  +    ?error("Changes stream broken - restarting at ~p", [CurrentSeq]),
  +    start_changes_stream(CurrentSeq),
       ok;
  +
   common_change_callback(Seq, Doc, Type) ->
  -    cleanup_expired(),
  +    cleanup_expired_callbacks(),
  +    update_cache(Seq, Doc, Type),
       Callbacks = worker_host:state_get(?MODULE, provider_callbacks),
  -    update_last_common_seq(Seq),
       handle_change(Seq, Doc, Type, Callbacks).
  -
  -update_last_common_seq(Seq) ->
  -    worker_host:state_update(?MODULE, last_seq, fun(Val) ->
  -        case Val < Seq of
  -            true -> Seq;
  -            false -> Val
  -        end
  -    end).
   
   handle_change(Seq, Doc, Type, Callbacks) ->
       NoOp = fun(_Seq, _Doc, _Type) -> ok end,
       Providers = get_providers(Seq, Doc, Type),
       ?info("Dispatching {seq=~p, doc=~p, type=~p} to ~p having ~p", [Seq, Doc, Type, Providers, Callbacks]),
  +
       lists:foreach(fun(Provider) ->
           {Callback, _} = maps:get(Provider, Callbacks, NoOp),
           spawn(fun() -> Callback(Seq, Doc, Type) end)
       end, Providers).
   
  -cleanup_expired() ->
  +cleanup_expired_callbacks() ->
       worker_host:state_update(?MODULE, provider_callbacks, fun(Callbacks) ->
           Now = now_seconds(),
           maps:filter(fun(_ID, {_Callback, ExpiresAt}) ->
  
positive.code: init(_Args) ->
      timer:sleep(timer:seconds(2)),% todo wait for couchdb
      process_flag(trap_exit, true),
      LastSeq = get_last_seq(),
      start_changes_stream(LastSeq),
      {ok, #{provider_callbacks => #{}, last_seq => LastSeq}}.
  
  handle(healthcheck) ->
      ok;
  handle({provider_subscribe, ProviderID, Callback, From}) ->
      ?info("Request ~p", [{provider_subscribe, ProviderID, Callback, From}]),
      add_subscription(ProviderID, Callback),
      fetch_history(ProviderID, Callback, From);
  handle(_Request) ->
      ?log_bad_request(_Request).
  
  
negative.code:     ok.
  
  fetch_history(ProviderID, Callback, From) ->
      NewestCached = newest_cached_seq(),
      OldestCached = oldest_cached_seq(),
      ?info("Fetching from cache; {provider:~p, from:~p, cache_start:~p, cache_end:~p}",
          [ProviderID, From, OldestCached, NewestCached]),
  
      case {OldestCached, From >= OldestCached} of
          {cache_empty, _} ->
              fetch_from_db(From, last_seq(), ProviderID, Callback);
          {_, true} ->
              fetch_from_cache(From, NewestCached, ProviderID, Callback);
          {_, false} ->
              fetch_from_cache(From, NewestCached, ProviderID, Callback),
              fetch_from_db(From, OldestCached - 1, ProviderID, Callback)
      end.
  
  
  fetch_from_cache(From, To, ProviderID, Callback) ->
      Cache = worker_host:state_get(?MODULE, cache),
      CachedList = gb_trees:values(Cache),
      Shifted = lists:dropwhile(fun({Seq, _}) -> Seq < From end, CachedList),
      Trimmed = lists:takewhile(fun({Seq, _}) -> Seq =< To end, Shifted),
      lists:foreach(fun({Seq, {Doc, Type}}) ->
          handle_change(Seq, Doc, Type, #{ProviderID => {Callback, infinity}})
      end, Trimmed).
  
  fetch_from_db(From, To, ProviderID, Callback) ->
      couchdb_datastore_driver:changes_start_link(fun
          (_Seq, stream_ended, _Type) -> ok;
          (Seq, Doc, Type) -> handle_change(Seq, Doc, Type, #{ProviderID => {Callback, infinity}})
  