feat(webrtc): fin-ack functionality

Cargo.toml

@@ -110,7 +110,7 @@ libp2p-tls = { version = "0.6.2", path = "transports/tls" }
 libp2p-uds = { version = "0.43.0", path = "transports/uds" }
 libp2p-upnp = { version = "0.5.0", path = "protocols/upnp" }
 libp2p-webrtc = { version = "0.9.0-alpha.1", path = "transports/webrtc" }
-libp2p-webrtc-utils = { version = "0.4.0", path = "misc/webrtc-utils" }
+libp2p-webrtc-utils = { version = "0.4.1", path = "misc/webrtc-utils" }
 libp2p-webrtc-websys = { version = "0.4.0", path = "transports/webrtc-websys" }
 libp2p-websocket = { version = "0.45.1", path = "transports/websocket" }
 libp2p-websocket-websys = { version = "0.5.0", path = "transports/websocket-websys" }

misc/webrtc-utils/CHANGELOG.md

@@ -1,3 +1,8 @@
+## 0.4.1
+
+- FIN_ACK support for WebRTC streams.
+  See [PR 6084](https://github.com/libp2p/rust-libp2p/pull/6084).
+
 ## 0.4.0
 
 <!-- Update to libp2p-core v0.43.0 -->

misc/webrtc-utils/Cargo.toml

@@ -7,7 +7,7 @@ license = "MIT"
 name = "libp2p-webrtc-utils"
 repository = "https://github.com/libp2p/rust-libp2p"
 rust-version = { workspace = true }
-version = "0.4.0"
+version = "0.4.1"
 publish = true
 
 [dependencies]

misc/webrtc-utils/src/generated/message.proto

@@ -12,9 +12,13 @@ message Message {
     // The sender abruptly terminates the sending part of the stream. The
     // receiver can discard any data that it already received on that stream.
     RESET = 2;
+    // Sending the FIN_ACK flag acknowledges the previous receipt of a message
+    // with the FIN flag set. Receiving a FIN_ACK flag gives the recipient
+    // confidence that the remote has received all sent messages.
+    FIN_ACK = 3;
   }
 
-  optional Flag flag=1;
+  optional Flag flag = 1;
 
   optional bytes message = 2;
 }

misc/webrtc-utils/src/generated/webrtc/pb.rs

@@ -57,6 +57,7 @@ pub enum Flag {
     FIN = 0,
     STOP_SENDING = 1,
     RESET = 2,
+    FIN_ACK = 3,
 }
 
 impl Default for Flag {
@@ -71,6 +72,7 @@ impl From<i32> for Flag {
             0 => Flag::FIN,
             1 => Flag::STOP_SENDING,
             2 => Flag::RESET,
+            3 => Flag::FIN_ACK,
             _ => Self::default(),
         }
     }
@@ -82,6 +84,7 @@ impl<'a> From<&'a str> for Flag {
             "FIN" => Flag::FIN,
             "STOP_SENDING" => Flag::STOP_SENDING,
             "RESET" => Flag::RESET,
+            "FIN_ACK" => Flag::FIN_ACK,
             _ => Self::default(),
         }
     }

misc/webrtc-utils/src/stream.rs

@@ -135,6 +135,17 @@ where
                 return Poll::Ready(Ok(n));
             }
 
+            // Check if we need to send a FIN_ACK
+            if self.state.needs_fin_ack() {
+                ready!(self.io.poll_ready_unpin(cx))?;
+                self.io.start_send_unpin(Message {
+                    flag: Some(Flag::FIN_ACK),
+                    message: None,
+                })?;
+                ready!(self.io.poll_flush_unpin(cx))?;
+                self.state.fin_ack_sent();
+            }
+
             let Self {
                 read_buffer,
                 io,
@@ -177,6 +188,17 @@ where
         cx: &mut Context<'_>,
         buf: &[u8],
     ) -> Poll<io::Result<usize>> {
+        // Check if we need to send a FIN_ACK first
+        if self.state.needs_fin_ack() {
+            ready!(self.io.poll_ready_unpin(cx))?;
+            self.io.start_send_unpin(Message {
+                flag: Some(Flag::FIN_ACK),
+                message: None,
+            })?;
+            ready!(self.io.poll_flush_unpin(cx))?;
+            self.state.fin_ack_sent();
+        }
+
         while self.state.read_flags_in_async_write() {
             // TODO: In case AsyncRead::poll_read encountered an error or returned None earlier, we
             // will poll the underlying I/O resource once more. Is that allowed? How
@@ -193,7 +215,7 @@ where
                 Poll::Ready(Some((Some(flag), message))) => {
                     // Read side is closed. Discard any incoming messages.
                     drop(message);
-                    // But still handle flags, e.g. a `Flag::StopSending`.
+                    // But still handle flags, e.g. a `Flag::StopSending` or `Flag::FIN_ACK`.
                     state.handle_inbound_flag(flag, read_buffer)
                 }
                 Poll::Ready(Some((None, message))) => drop(message),
@@ -216,10 +238,32 @@ where
     }
 
     fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
+        // Check if we need to send a FIN_ACK first
+        if self.state.needs_fin_ack() {
+            ready!(self.io.poll_ready_unpin(cx))?;
+            self.io.start_send_unpin(Message {
+                flag: Some(Flag::FIN_ACK),
+                message: None,
+            })?;
+            ready!(self.io.poll_flush_unpin(cx))?;
+            self.state.fin_ack_sent();
+        }
+
         self.io.poll_flush_unpin(cx).map_err(Into::into)
     }
 
     fn poll_close(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
+        // Check if we need to send a FIN_ACK first
+        if self.state.needs_fin_ack() {
+            ready!(self.io.poll_ready_unpin(cx))?;
+            self.io.start_send_unpin(Message {
+                flag: Some(Flag::FIN_ACK),
+                message: None,
+            })?;
+            ready!(self.io.poll_flush_unpin(cx))?;
+            self.state.fin_ack_sent();
+        }
+
         loop {
             match self.state.close_write_barrier()? {
                 Some(Closing::Requested) => {