Implement flush policy

Introduce a configurable FlushPolicy to control when the TLS layer flushes its
underlying transport.

This change adds a FlushPolicy enum (Relaxed or Strict) and wires it into TlsConnection
and TlsWriter. The policy determines whether the transport’s flush() is called after
writing a TLS record.

With Strict (the default), the transport is always flushed, ensuring that data is acknowledged
or committed before continuing. This mode is compatible with existing behavior.

With Relaxed, the TLS layer closes the record and hands bytes to the transport without
forcing a flush, allowing buffered writes to improve performance and reduce latency.

This gives callers explicit control over how aggressively the transport flushes data.
It's particularly important for transports like embassy-net TCP, where flush() blocks
waiting for ACKs. The default remains Strict to preserve compatibility with embedded-tls 0.17.0,
while the Relaxed mode can be selected together with larger socket window for improved throughput.

Author: dmitry-tarnyagin

src/asynch.rs

@@ -4,6 +4,7 @@ use crate::TlsError;
 use crate::common::decrypted_buffer_info::DecryptedBufferInfo;
 use crate::common::decrypted_read_handler::DecryptedReadHandler;
 use crate::connection::{Handshake, State, decrypt_record};
+use crate::flush_policy::FlushPolicy;
 use crate::key_schedule::KeySchedule;
 use crate::key_schedule::{ReadKeySchedule, WriteKeySchedule};
 use crate::read_buffer::ReadBuffer;
@@ -30,6 +31,7 @@ where
     record_reader: RecordReader<'a>,
     record_write_buf: WriteBuffer<'a>,
     decrypted: DecryptedBufferInfo,
+    flush_policy: FlushPolicy,
 }
 
 impl<'a, Socket, CipherSuite> TlsConnection<'a, Socket, CipherSuite>
@@ -63,9 +65,28 @@ where
             record_reader: RecordReader::new(record_read_buf),
             record_write_buf: WriteBuffer::new(record_write_buf),
             decrypted: DecryptedBufferInfo::default(),
+            flush_policy: FlushPolicy::default(),
         }
     }
 
+    /// Returns a reference to the current flush policy.
+    ///
+    /// The flush policy controls whether the underlying transport is flushed
+    /// (via its `flush()` method) after writing a TLS record.
+    #[inline]
+    pub fn flush_policy(&self) -> FlushPolicy {
+        self.flush_policy
+    }
+
+    /// Replace the current flush policy with the provided one.
+    ///
+    /// This sets how and when the connection will call `flush()` on the
+    /// underlying transport after writing records.
+    #[inline]
+    pub fn set_flush_policy(&mut self, policy: FlushPolicy) {
+        self.flush_policy = policy;
+    }
+
     /// Open a TLS connection, performing the handshake with the configuration provided when
     /// creating the connection instance.
     ///
@@ -152,15 +173,22 @@ where
 
             key_schedule.increment_counter();
 
-            self.delegate
-                .flush()
-                .await
-                .map_err(|e| TlsError::Io(e.kind()))?;
+            if self.flush_policy.flush_transport() {
+                self.flush_transport().await?;
+            }
         }
 
         Ok(())
     }
 
+    #[inline]
+    async fn flush_transport(&mut self) -> Result<(), TlsError> {
+        self.delegate
+            .flush()
+            .await
+            .map_err(|e| TlsError::Io(e.kind()))
+    }
+
     fn create_read_buffer(&mut self) -> ReadBuffer {
         self.decrypted.create_read_buffer(self.record_reader.buf)
     }
@@ -231,7 +259,7 @@ where
 
         self.key_schedule.write_state().increment_counter();
 
-        self.flush().await
+        self.flush_transport().await
     }
 
     /// Close a connection instance, returning the ownership of the async I/O provider.
@@ -265,6 +293,7 @@ where
             delegate: self.delegate.clone(),
             key_schedule: wks,
             record_write_buf: self.record_write_buf.reborrow_mut(),
+            flush_policy: self.flush_policy,
         };
 
         (reader, writer)
@@ -391,6 +420,21 @@ where
     delegate: Socket,
     key_schedule: &'a mut WriteKeySchedule<CipherSuite>,
     record_write_buf: WriteBufferBorrowMut<'a>,
+    flush_policy: FlushPolicy,
+}
+
+impl<'a, Socket, CipherSuite> TlsWriter<'a, Socket, CipherSuite>
+where
+    Socket: AsyncWrite + 'a,
+    CipherSuite: TlsCipherSuite + 'static,
+{
+    #[inline]
+    async fn flush_transport(&mut self) -> Result<(), TlsError> {
+        self.delegate
+            .flush()
+            .await
+            .map_err(|e| TlsError::Io(e.kind()))
+    }
 }
 
 impl<Socket, CipherSuite> AsRef<Socket> for TlsWriter<'_, Socket, CipherSuite>
@@ -487,10 +531,9 @@ where
 
             self.key_schedule.increment_counter();
 
-            self.delegate
-                .flush()
-                .await
-                .map_err(|e| TlsError::Io(e.kind()))?;
+            if self.flush_policy.flush_transport() {
+                self.flush_transport().await?;
+            }
         }
 
         Ok(())

src/blocking.rs

@@ -3,6 +3,7 @@ use core::sync::atomic::Ordering;
 use crate::common::decrypted_buffer_info::DecryptedBufferInfo;
 use crate::common::decrypted_read_handler::DecryptedReadHandler;
 use crate::connection::{Handshake, State, decrypt_record};
+use crate::flush_policy::FlushPolicy;
 use crate::key_schedule::KeySchedule;
 use crate::key_schedule::{ReadKeySchedule, WriteKeySchedule};
 use crate::read_buffer::ReadBuffer;
@@ -30,6 +31,7 @@ where
     record_reader: RecordReader<'a>,
     record_write_buf: WriteBuffer<'a>,
     decrypted: DecryptedBufferInfo,
+    flush_policy: FlushPolicy,
 }
 
 impl<'a, Socket, CipherSuite> TlsConnection<'a, Socket, CipherSuite>
@@ -64,9 +66,28 @@ where
             record_reader: RecordReader::new(record_read_buf),
             record_write_buf: WriteBuffer::new(record_write_buf),
             decrypted: DecryptedBufferInfo::default(),
+            flush_policy: FlushPolicy::default(),
         }
     }
 
+    /// Returns a reference to the current flush policy.
+    ///
+    /// The flush policy controls whether the underlying transport is flushed
+    /// (via its `flush()` method) after writing a TLS record.
+    #[inline]
+    pub fn flush_policy(&self) -> FlushPolicy {
+        self.flush_policy
+    }
+
+    /// Replace the current flush policy with the provided one.
+    ///
+    /// This sets how and when the connection will call `flush()` on the
+    /// underlying transport after writing records.
+    #[inline]
+    pub fn set_flush_policy(&mut self, policy: FlushPolicy) {
+        self.flush_policy = policy;
+    }
+
     /// Open a TLS connection, performing the handshake with the configuration provided when
     /// creating the connection instance.
     ///
@@ -147,12 +168,21 @@ where
 
             key_schedule.increment_counter();
 
-            self.delegate.flush().map_err(|e| TlsError::Io(e.kind()))?;
+            if self.flush_policy.flush_transport() {
+                self.flush_transport()?;
+            }
         }
 
         Ok(())
     }
 
+    #[inline]
+    fn flush_transport(&mut self) -> Result<(), TlsError> {
+        self.delegate
+            .flush()
+            .map_err(|e| TlsError::Io(e.kind()))
+    }
+
     fn create_read_buffer(&mut self) -> ReadBuffer {
         self.decrypted.create_read_buffer(self.record_reader.buf)
     }
@@ -219,7 +249,7 @@ where
 
         self.key_schedule.write_state().increment_counter();
 
-        self.flush()?;
+        self.flush_transport()?;
 
         Ok(())
     }
@@ -255,6 +285,7 @@ where
             delegate: self.delegate.clone(),
             key_schedule: wks,
             record_write_buf: self.record_write_buf.reborrow_mut(),
+            flush_policy: self.flush_policy,
         };
 
         (reader, writer)
@@ -380,6 +411,19 @@ where
     delegate: Socket,
     key_schedule: &'a mut WriteKeySchedule<CipherSuite>,
     record_write_buf: WriteBufferBorrowMut<'a>,
+    flush_policy: FlushPolicy,
+}
+
+impl<'a, Socket, CipherSuite> TlsWriter<'a, Socket, CipherSuite>
+where
+    Socket: Write + 'a,
+    CipherSuite: TlsCipherSuite + 'static,
+{
+    fn flush_transport(&mut self) -> Result<(), TlsError> {
+        self.delegate
+            .flush()
+            .map_err(|e| TlsError::Io(e.kind()))
+    }
 }
 
 impl<Socket, CipherSuite> AsRef<Socket> for TlsWriter<'_, Socket, CipherSuite>
@@ -475,7 +519,10 @@ where
 
             self.key_schedule.increment_counter();
 
-            self.delegate.flush().map_err(|e| TlsError::Io(e.kind()))?;
+            if self.flush_policy.flush_transport() {
+                self.flush_transport()?;
+            }
+
         }
 
         Ok(())

src/flush_policy.rs

@@ -0,0 +1,37 @@
+//! Flush policy for TLS sockets.
+//!
+//! Two strategies are provided:
+//! - `Relaxed`: close the TLS encryption buffer and hand the data to the transport
+//!   delegate without forcing a transport-level flush.
+//! - `Strict`: in addition to handing the data to the transport delegate, also
+//!   request a flush of the transport. For TCP transports this typically means
+//!   waiting for an ACK (e.g. on embassy TCP sockets) before considering the
+//!   data fully flushed.
+
+/// Policy controlling how TLS layer flushes encrypted data to the transport.
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub enum FlushPolicy {
+    /// Close the TLS encryption buffer and pass bytes to the transport delegate.
+    /// Do not force a transport-level flush or wait for an ACK.
+    Relaxed,
+
+    /// In addition to passing bytes to the transport delegate, request a
+    /// transport-level flush and wait for confirmation (ACK) before returning.
+    Strict,
+}
+
+impl FlushPolicy {
+    /// Returns true when the transport delegate should be explicitly flushed.
+    ///
+    /// Relaxed -> false, Strict -> true.
+    pub fn flush_transport(&self) -> bool {
+        matches!(self, Self::Strict)
+    }
+}
+
+impl Default for FlushPolicy {
+    /// Default to `Strict` for compatibility with embedded-tls 0.17.0.
+    fn default() -> Self {
+        FlushPolicy::Strict
+    }
+}

src/lib.rs

@@ -63,6 +63,7 @@ mod connection;
 mod content_types;
 mod crypto_engine;
 mod extensions;
+pub mod flush_policy;
 mod handshake;
 mod key_schedule;
 mod parse_buffer;
@@ -82,6 +83,8 @@ pub mod webpki;
 mod asynch;
 pub use asynch::*;
 
+pub use flush_policy::*;
+
 #[derive(Debug, Copy, Clone)]
 #[cfg_attr(feature = "defmt", derive(defmt::Format))]
 pub enum TlsError {