anchor.diff:          pending = _delegate.handle(request).single;
           pending
             .catchError((_) {}) // Ignore errors here, they are caught separately below.
  -          .then((entity) => _handleRpcReply(request, entity));
  +          .then((entity) => _handleRpcReply(request, entity))
  +          .whenComplete(() {
  +            _inFlightRequests.remove(request);
  +          });
           _inFlightRequests[request] = pending;
         } else {
           pending = _inFlightRequests[request];
  
positive.code:   _removeActive(_ActiveStream stream) =>
        _activeIndex.removeValue(stream.request, stream);
  
    _maybeRecordRpcData(entity) {
      entity.streamy
        ..ts = new DateTime.now().millisecondsSinceEpoch
        ..source = 'RPC';
  
negative.code:         })
          .then((entity) {
            if (entity != _INTERNAL_ERROR) {
              _recordRpcData(entity);
              active.submit(entity);
            }
          })
  