anchor.diff:  %% @doc Buffer a produce request.
   %% Respond to caller immediately if the buffer limit is not yet reached.
   %% @end
  --spec add(buf(), brod_call_ref(), key(), value()) -> {ok, buf()}.
  +-spec add(buf(), brod:brod_call_ref(), brod:key(), brod:value()) -> {ok, buf()}.
   add(#buf{pending = Pending} = Buf, CallRef, Key, Value) ->
     Req = #req{ call_ref = CallRef
               , data     = fun() -> {Key, Value} end
  
positive.code:         , onwire_count   = 0            :: non_neg_integer()
          , pending        = ?EMPTY_QUEUE :: queue:queue(#req{})
          , buffer         = ?EMPTY_QUEUE :: queue:queue(#req{})
          , onwire         = []           :: [{corr_id(), [#req{}]}]
          }).
  
  -opaque buf() :: #buf{}.
  
negative.code: %% reached maximum retry limit.
  %% Unknown corr-id:s are ignored.
  %% @end
  -spec nack(buf(), brod:corr_id(), any()) -> {ok, buf()} | {error, ignored}.
  nack(#buf{onwire = [{CorrId, _Reqs} | _]} = Buf, CorrId, Reason) ->
    nack_all(Buf, Reason);
  nack(#buf{onwire = OnWire}, CorrIdReceived, _Reason) ->
  