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:   }
  
    _handleRpcReply(Request request, Entity entity) {
      _inFlightRequests.remove(request);
  
      if (entity == _INTERNAL_ERROR) {
        // An error occurred, no need to handle it here other than removing the in-flight request.
        return;
      }
  
      _maybeRecordRpcData(entity);
  
      // Publish this new entity on every channel.
      _activeIndex[request].forEach((act) => runAsync(() => act.submit(entity)));
  
negative.code:         })
          .then((entity) {
            if (entity != _INTERNAL_ERROR) {
              _recordRpcData(entity);
              active.submit(entity);
            }
          })
  