anchor.diff:  -include("riak_kv_js_pools.hrl").
   
   -record(state, {acc :: list(),
  +                inacc :: list(),
                   delay :: integer(),
                   delay_max :: integer(),
                   p :: riak_pipe_vnode:partition(),
  
positive.code: 
  -spec handoff_acc([term()], [term()], state()) -> [term()].
  handoff_acc(HandoffAcc, LocalAcc, State) ->
      InAcc = HandoffAcc++LocalAcc,
      reduce(InAcc, State, "reducing handoff").
  
  %% @doc Actually evaluate the aggregation function.
  
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 ],
  