dekaf: apply a 30 second idle session timeout

Require that clients send a request within thirty seconds,
or close the connection.

crates/dekaf/src/main.rs

@@ -17,7 +17,7 @@ use std::{
     path::{Path, PathBuf},
     sync::Arc,
 };
-use tokio::io::{split, AsyncRead, AsyncWrite, AsyncWriteExt};
+use tokio::io::{AsyncRead, AsyncWrite, AsyncWriteExt};
 use tracing_subscriber::{filter::LevelFilter, EnvFilter};
 use url::Url;
 
@@ -261,41 +261,43 @@ where
     S: AsyncRead + AsyncWrite + Unpin,
 {
     tracing::info!("accepted client connection");
+
+    let (r, mut w) = tokio::io::split(socket);
+    let mut r = tokio_util::codec::FramedRead::new(
+        r,
+        tokio_util::codec::LengthDelimitedCodec::builder()
+            .big_endian()
+            .length_field_length(4)
+            .max_frame_length(1 << 27) // 128 MiB
+            .new_codec(),
+    );
+    let mut out = bytes::BytesMut::new();
+    let mut raw_sasl_auth = false;
+
     metrics::gauge!("dekaf_total_connections").increment(1);
+
     let result = async {
-        let (r, mut w) = split(socket);
-
-        let mut r = tokio_util::codec::FramedRead::new(
-            r,
-            tokio_util::codec::LengthDelimitedCodec::builder()
-                .big_endian()
-                .length_field_length(4)
-                .max_frame_length(1 << 27) // 128 MiB
-                .new_codec(),
-        );
-
-        let mut out = bytes::BytesMut::new();
-        let mut raw_sasl_auth = false;
-        let mut res = Ok(());
-        while let Some(frame) = r.try_next().await? {
-            if let err @ Err(_) =
-                dekaf::dispatch_request_frame(&mut session, &mut raw_sasl_auth, frame, &mut out)
-                    .await
-            {
-                // Close the connection on error
-                w.shutdown().await?;
-                res = err;
-            }
+        loop {
+            let Some(frame) = tokio::time::timeout(SESSION_TIMEOUT, r.try_next())
+                .await
+                .context("timeout waiting for next session request")?
+                .context("failed to read next session request")?
+            else {
+                return Ok(());
+            };
+
+            dekaf::dispatch_request_frame(&mut session, &mut raw_sasl_auth, frame, &mut out)
+                .await?;
+
             () = w.write_all(&mut out).await?;
             out.clear();
         }
-
-        res
     }
     .await;
 
     metrics::gauge!("dekaf_total_connections").decrement(1);
 
+    w.shutdown().await?;
     result
 }
 
@@ -334,3 +336,5 @@ fn validate_certificate_name(
     }
     return Ok(false);
 }
+
+const SESSION_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(30);