anchor.diff:  
   %% @doc The archive is the accumulator.
   -spec archive(state()) -> {ok, list()}.
  -archive(#state{acc=Acc}) ->
  +archive(#state{acc=Acc, inacc=InAcc}) ->
       %% just send state of reduce so far
  -    {ok, Acc}.
  +    {ok, Acc ++ lists:reverse(InAcc)}.
   
   %% @doc Handoff simply concatenates the accumulator from the remote
   %% worker with the accumulator from this worker, and immediately
  
positive.code: -include("riak_kv_js_pools.hrl").
  
  -record(state, {acc :: list(),
                  delay :: integer(),
                  delay_max :: integer(),
                  p :: riak_pipe_vnode:partition(),
  
negative.code:               _ ->
                    []
            end,
      {ok, #state{acc=Acc, inacc=[], delay=0, delay_max = DelayMax,
                  p=Partition, fd=FittingDetails}}.
  
  %% @doc Evaluate the function if the batch is ready.
  -spec process(term(), boolean(), state()) -> {ok, state()}.
  process(Input, _Last,
          #state{acc=PrevAcc, inacc=OldInAcc, delay=Delay, delay_max=DelayMax}=State) ->
      InAcc = [Input|OldInAcc],
      if Delay + 1 >= DelayMax ->
              OutAcc = reduce(PrevAcc ++ lists:reverse(InAcc), State, "reducing"),
              {ok, State#state{acc=OutAcc, inacc=[], delay=0}};
         true ->
              {ok, State#state{inacc=InAcc, delay=Delay + 1}}
      end.
  
  %% @doc Reduce any unreduced inputs, and then send on the outputs.
  -spec done(state()) -> ok.
  done(#state{acc=Acc0, inacc=InAcc, delay=Delay, p=Partition, fd=FittingDetails} = S) ->
      Acc = if Delay == 0 ->
                    Acc0;
               true ->
                    reduce(Acc0 ++ lists:reverse(InAcc), S, "done()")
            end,
      [ riak_pipe_vnode_worker:send_output(O, Partition, FittingDetails)
        || O <- Acc ],
  