diff --git a/noq-proto/src/connection/mod.rs b/noq-proto/src/connection/mod.rs
index 3275d18f1..ea8ca7857 100644
--- a/noq-proto/src/connection/mod.rs
+++ b/noq-proto/src/connection/mod.rs
@@ -2421,13 +2421,7 @@ impl Connection {
             match timer {
                 Timer::Conn(timer) => match timer {
                     ConnTimer::Close => {
-                        let was_draining = self.state.move_to_drained(None);
-                        if !was_draining {
-                            self.endpoint_events.push_back(EndpointEventInner::Draining);
-                        }
-                        // move_to_drained checks that we weren't in drained before.
-                        // Adding events to endpoint_events is only legal if `Drained` was never queued before.
-                        self.endpoint_events.push_back(EndpointEventInner::Drained);
+                        self.state.move_to_drained(None, &mut self.endpoint_events);
                     }
                     ConnTimer::Idle => {
                         self.kill(ConnectionError::TimedOut);
@@ -4407,9 +4401,9 @@ impl Connection {
                     code: TransportErrorCode::AEAD_LIMIT_REACHED,
                     ..
                 }) => {
-                    let was_draining = self.state.move_to_drained(Some(conn_err));
-                    if !was_draining && !was_drained {
-                        self.endpoint_events.push_back(EndpointEventInner::Draining);
+                    if !self.state.is_drained() {
+                        self.state
+                            .move_to_drained(Some(conn_err), &mut self.endpoint_events);
                     }
                 }
                 ConnectionError::TimedOut => {
@@ -4420,8 +4414,8 @@ impl Connection {
                     self.state.move_to_closed(err);
                 }
                 ConnectionError::VersionMismatch => {
-                    self.state.move_to_draining(Some(conn_err));
-                    self.endpoint_events.push_back(EndpointEventInner::Draining);
+                    self.state
+                        .move_to_draining(Some(conn_err), &mut self.endpoint_events);
                 }
                 ConnectionError::LocallyClosed => {
                     unreachable!("LocallyClosed isn't generated by packet processing");
@@ -4439,7 +4433,6 @@ impl Connection {
             }
         }
         if !was_drained && self.state.is_drained() {
-            self.endpoint_events.push_back(EndpointEventInner::Drained);
             // Close timer may have been started previously, e.g. if we sent a close and got a
             // stateless reset in response
             self.timers
@@ -4537,8 +4530,7 @@ impl Connection {
                     self.path_stats.get_mut(path_id).frame_rx.record(frame.ty());
 
                     if let Frame::Close(_error) = frame {
-                        self.state.move_to_draining(None);
-                        self.endpoint_events.push_back(EndpointEventInner::Draining);
+                        self.state.move_to_draining(None, &mut self.endpoint_events);
                         break;
                     }
                 }
@@ -4864,8 +4856,8 @@ impl Connection {
                     self.on_path_ack_received(now, packet.header.space().into(), ack)?;
                 }
                 Frame::Close(reason) => {
-                    self.state.move_to_draining(Some(reason.into()));
-                    self.endpoint_events.push_back(EndpointEventInner::Draining);
+                    self.state
+                        .move_to_draining(Some(reason.into()), &mut self.endpoint_events);
                     return Ok(());
                 }
                 _ => {
@@ -5549,8 +5541,8 @@ impl Connection {
         self.streams.queue_max_stream_id(pending);
 
         if let Some(reason) = close {
-            self.state.move_to_draining(Some(reason.into()));
-            self.endpoint_events.push_back(EndpointEventInner::Draining);
+            self.state
+                .move_to_draining(Some(reason.into()), &mut self.endpoint_events);
             self.connection_close_pending = true;
         }
 
@@ -6972,13 +6964,8 @@ impl Connection {
     /// Terminate the connection instantly, without sending a close packet
     fn kill(&mut self, reason: ConnectionError) {
         self.close_common();
-        let was_draining = self.state.move_to_drained(Some(reason));
-        if !was_draining {
-            self.endpoint_events.push_back(EndpointEventInner::Draining);
-        }
-        // move_to_drained checks that we were never in drained before, so we
-        // never sent a `Drained` event before (it's illegal to send more events after drained).
-        self.endpoint_events.push_back(EndpointEventInner::Drained);
+        self.state
+            .move_to_drained(Some(reason), &mut self.endpoint_events);
     }
 
     /// Storage size required for the largest packet that can be transmitted on all currently
diff --git a/noq-proto/src/connection/state.rs b/noq-proto/src/connection/state.rs
index 06b593711..0a9c422e4 100644
--- a/noq-proto/src/connection/state.rs
+++ b/noq-proto/src/connection/state.rs
@@ -2,6 +2,7 @@ use bytes::Bytes;
 use tracing::trace;
 
 use crate::frame::Close;
+use crate::shared::EndpointEventInner;
 use crate::{ApplicationClose, ConnectionClose, ConnectionError, TransportError, TransportErrorCode};
 
 #[allow(unreachable_pub)] // fuzzing only
@@ -67,8 +68,12 @@ impl State {
     ///
     /// Panics if the state was already drained.
     ///
-    /// Returns whether we were in the draining state before.
-    pub(super) fn move_to_drained(&mut self, error: Option<ConnectionError>) -> bool {
+    /// Emits the appropriate `Draining` and `Drained` endpoint events into `events`.
+    pub(super) fn move_to_drained(
+        &mut self,
+        error: Option<ConnectionError>,
+        events: &mut impl Extend<EndpointEventInner>,
+    ) {
         let (error, is_local, was_draining) = if let Some(error) = error {
             (
                 Some(error),
@@ -103,13 +108,23 @@ impl State {
         };
         self.inner = InnerState::Drained { error, is_local };
         trace!("connection state: drained");
-        was_draining
+
+        if !was_draining {
+            events.extend([EndpointEventInner::Draining]);
+        }
+        events.extend([EndpointEventInner::Drained]);
     }
 
     /// Moves to a draining state.
     ///
     /// Panics if the state is already draining or drained.
-    pub(super) fn move_to_draining(&mut self, error: Option<ConnectionError>) {
+    ///
+    /// Emits a `Draining` endpoint event into `events`.
+    pub(super) fn move_to_draining(
+        &mut self,
+        error: Option<ConnectionError>,
+        events: &mut impl Extend<EndpointEventInner>,
+    ) {
         assert!(
             matches!(
                 self.inner,
@@ -139,6 +154,7 @@ impl State {
 
         self.inner = InnerState::Draining { error, is_local };
         trace!("connection state: draining");
+        events.extend([EndpointEventInner::Draining]);
     }
 
     fn is_local_close(&self) -> bool {