tor-proto: reject() now waits until the control message is received.

As a result, by the time the `reject` future resolves, the stream has
been removed from the reactor's stream map and the corresponding END
cell has been sent.

Fixes #998.

crates/tor-proto/src/circuit.rs

@@ -1053,14 +1053,20 @@ impl StreamTarget {
     //   called before)
     #[cfg(feature = "hs-service")]
     pub(crate) async fn close(&self, msg: relaymsg::End) -> Result<()> {
+        let (tx, rx) = oneshot::channel();
+
         self.circ
             .control
             .unbounded_send(CtrlMsg::ClosePendingStream {
                 stream_id: self.stream_id,
                 hop_num: self.hop_num,
                 message: msg,
+                done: tx,
             })
             .map_err(|_| Error::CircuitClosed)?;
+
+        // Check whether the ClosePendingStream was processed successfully.
+        rx.await.map_err(|_| Error::CircuitClosed)??;
         Ok(())
     }
 

crates/tor-proto/src/circuit/reactor.rs

@@ -177,6 +177,8 @@ pub(super) enum CtrlMsg {
         stream_id: StreamId,
         /// The END message to send.
         message: End,
+        /// Oneshot channel to notify on completion.
+        done: ReactorResultChannel<()>,
     },
     /// Begin accepting streams on this circuit.
     #[cfg(feature = "hs-service")]
@@ -1394,8 +1396,10 @@ impl Reactor {
                 hop_num,
                 stream_id,
                 message,
+                done,
             } => {
-                self.close_stream(cx, hop_num, stream_id, Some(message))?;
+                let ret = self.close_stream(cx, hop_num, stream_id, Some(message));
+                let _ = done.send(ret); // don't care if sender goes away
             }
             #[cfg(feature = "hs-service")]
             CtrlMsg::AwaitStreamRequest {