anchor.diff:      _inputSubscription?.cancel();
   
       for (final pending in _pendingRequests.values) {
  -      pending.completeError(const ConnectionClosedException());
  +      pending.completeWithError(const ConnectionClosedException());
       }
       _pendingRequests.clear();
     }
  
positive.code:     }
  
      if (msg is SuccessResponse) {
        final completer = _pendingRequests[msg.requestId];
        completer?.complete(msg.response);
        _pendingRequests.remove(msg.requestId);
      } else if (msg is ErrorResponse) {
        final requestId = msg.requestId;
        final completer = _pendingRequests[requestId];
        final backgroundTrace = msg.stackTrace != null
            ? StackTrace.fromString(msg.stackTrace!)
            : null;
        final foregroundTrace = _requestTraces[requestId];
  
        completer?.completeError(
            DriftRemoteException._(msg.error, backgroundTrace), foregroundTrace);
        _pendingRequests.remove(msg.requestId);
      } else if (msg is Request) {
        _incomingRequests.add(msg);
      } else if (msg is CancelledResponse) {
        final completer = _pendingRequests[msg.requestId];
        completer?.completeError(const CancellationException());
      }
    }
  
  
negative.code:   }
  }
  
  class _PendingRequest {
    final Completer completer;
  
    /// We capture the current stack trace when `request` is called so that, if
    /// an exception occurs on the remote peer, we can throw exceptions with a
    /// proper stack trace pointing torwards the causing invocation.
    final StackTrace requestTrace;
  
    _PendingRequest(this.completer, this.requestTrace);
  
    void completeWithError(Object error) {
      completer.completeError(error, requestTrace);
    }
  }
  
  /// Exception thrown when there are outstanding pending requests at the time the
  /// isolate connection was cancelled.
  class ConnectionClosedException implements Exception {
  