feat: implement support for volatile events with new VolatileSocket w… by LiamCarPer · Pull Request #744 · Totodore/socketioxide

LiamCarPer · 2026-06-20T19:08:11Z

Motivation

Socket.io's Node.js implementation supports a volatile flag on emits that drops events when the underlying connection is not ready. This is highly useful for high-frequency, non-critical data like game position updates or telemetry, as it prevents buffer buildup on unstable connections.

This feature was requested in #602 and is documented in the [Socket.io volatile events spec](https://www.google.com/search?q=https://socket.io/docs/v4/emitting-events/%23volatile-events). Currently, socketioxide has no equivalent, forcing users to either accept buffer pressure or implement their own messaging layer.

Solution

Added a Volatile variant to BroadcastFlags (0x04) that flows through the existing adapter broadcast pipeline. On the local adapter, when volatile is set, errors from send_many are silently discarded rather than propagated, perfectly matching fire-and-forget semantics.

User-Facing API

Three entry points are provided to mirror the Node.js patterns:

Socket::volatile() Returns a VolatileSocket<'a, A> wrapper with a direct emit() that silently drops events when the socket is disconnected, the internal buffer is full, or encoding fails. The wrapper also exposes chain methods (to, within, except, local, broadcast, timeout) that delegate to BroadcastOperators with the volatile flag pre-set.
**BroadcastOperators::volatile() and ConfOperators::volatile()** Sets the flag on the operator chain, enabling room-based broadcasting patterns like:

io.to("room").volatile().emit(...);

SocketIo::volatile() A convenience alias on the default namespace for global emits:

io.volatile().emit(...);

Adapter Propagation

The flag propagates directly through the BroadcastOptions struct. This ensures remote adapters (Redis, Postgres, MongoDB) receive it and can handle volatile semantics on their own nodes out of the box, without requiring any adapter-specific changes.

codspeed-hq · 2026-06-20T19:10:16Z

Merging this PR will not alter performance

✅ 87 untouched benchmarks

_{Comparing LiamCarPer:feat/volatile-flag (e40d1ad) with main (6f0774a)}

Totodore · 2026-06-20T19:35:34Z

Please check this comment regarding the implementation:
#602 (comment)

you are missing a way to bypass the mpsc channel in engineio. Volatile packets will still be buffered, it is not what we want.

I might be wrong but, before Claude-ing something please try to understand the issue details before implementing something. Feel free to ask more details here or on the issue.

LiamCarPer · 2026-06-20T20:00:29Z

Thanks for the Review. You are right, my current approach still goes through the mpcs channel so volatile packets can buffer.
i am planning to add a separate volatile send path in engineio, a second mpsc channel on Socket with capacity 1, paired with a send_volatile() method that does try _send iton it (drops if full). Then update both transporters to drain it with priority. Then thread the volatile flag BroadcastOptions through send_many so individual socket sends in broadcast flows also use the volatile engine.io path.

Does that sound good? Want to make sure i am in the right track before spending more time on it

LiamCarPer · 2026-06-21T10:10:47Z

Is this what you were looking for?

Totodore · 2026-06-21T10:35:40Z

Yes! One thing that we are missing with this solution is that volatile packets are out of order with classic packets. It might be ok though. We can simply document this specific case.
It would be nice to check what are the ordering constraints in the js implementation though and explore if there is any solution for this. I'll review this whenever I can.

LiamCarPer · 2026-06-21T11:17:53Z

Perfect, let me look into it

LiamCarPer · 2026-06-21T12:20:47Z

I checked js docs and they dont specify ordering constraints for volatile vs regular event.
I have added doc comments on both VolatileSocket and the engine.io emit.volatile() methods.

Totodore

This is the good direction!
We need to rethink how the polling payload encoding is done though. You can pick some changes from #555 regarding the use of Stream in the encoder.

…rapper and operator flags

…ile events

…tile message loop in transport

…lling payload test

LiamCarPer · 2026-06-26T09:56:17Z

Hey man, sorry for being this late ran out of claude usage, jajjajaj just joking, was busy with work.

I removed VolatileSocket, used ConfOperators with volatile flag, removed tokio macros feature, documented return values, moved volatile docs to external file. Test results, 79/79 unit tests, 106/109 doc tests, 0 clippy warnings, clean.

You had a concern about polling encoder, there are no .await points between the volatile snapshot and the encoder call, so the concern doesnt apply to the specific code,

Check and tell me is everything as you want it!

… polling in encoders to capture packets arriving during encoding

Totodore

Hey, it is good, however we are missing integration tests for engineioxide and socketioxide and unit tests for the encoder functions in engineioxide.
We need to test every combination possible of volatile vs non-volatile packets to ensure payloads are still correct.

LiamCarPer · 2026-06-29T20:01:51Z

I went ahead and added a bunch of comprehensive test coverage.

For the encoder unit tests in encoder.rs, I added 10 new tests checking the core logic for all three versions (v4, v3 string, and v3 binary). That covers volatile-only payloads, mixed normal and volatile ordering both ways, making sure the latest volatile data overwrites the old stuff, and testing what happens if volatile data drops in right while a buffer is draining.

Then for engineioxide integration tests in tests/volatile.rs, I added 3 tests to check the transport layer. It confirms volatile messages make it all the way through the full polling transport, double-checks the ordering when mixing volatile and normal messages, and makes sure those overwrite semantics hold up at the engine level too.

Lastly for socketioxide integration tests in tests/volatile.rs, i added 4 tests for the user-facing API and broadcasting. It ensures socket.volatile().emit() returns Ok(()), makes sure broadcasting with the volatile flag on doesn't panic, and checks that io.volatile() plays nice with the rest of our tools.

I have left two edge cases uncovered, Parked encoder + volatile arrives and Max payload boundary with volatile. What do you want me to do with it?

Totodore

Thanks for the tests, yes it would be nice to add parked encoder and max payload checks (especially because these are edge cases that may easily results in bugs).

…r tests for volatile packets

LiamCarPer · 2026-06-29T21:38:12Z

I handled the WS flush consolidation in crates/engineioxide/src/transport/ws.rs. I removed that double-flush pattern where it was flushing once after the main channel drain and then again after volatile. Now there is just a single unconditional tx.flush().await.ok() at the end of each loop iteration, so it processes both the main and volatile packets before doing a single flush.

I also added 3 new parked encoder tests. These cover scenarios where volatile arrives while the encoder is blocked on recv_packet() and confirms that volatile isn't captured in that specific payload, only on the next poll. I explicitly tested this across v4, v3 string, and v3 binary encoders.

I added 3 new max payload volatile tests. These make sure that if volatile pushes data.len() over the limit, it correctly leaves the normal packets in the channel for the next poll. Like the others, this is fully tested for v4, v3 string, and v3 binary.

Totodore · 2026-06-29T21:59:14Z

I cant make it work on the whiteboard example, no events is being emitted. Try on your side to emit the drawing event with volatile and check if you get something. Same if I force polling transport, it doesn't work. Once it works with the whiteboard example check also with remote adapters (redis/postgres to find the issue).

Once you find the bug in the whiteboard example, please add regression tests.

Tip:
Maybe add more logs to track engineioxide behaviors, you can check with RUST_LOG=socketioxide=trace,engineioxide=trace.

…g transport polling stalls

LiamCarPer · 2026-06-29T23:20:42Z

Found it, volatile packets arriving while the encoder is parked on recv_packet() were not captured. The encoder checks volatile in the initial check and in the drain loop, but after parking, volatile arriving during that .await is missed, it only gets captured on the next poll.

I have added new regression tests "volatile_broadcast_arrives_via_polling_transport".

Thank you for the tip, it was useful

Totodore

For WS transport the issue still appears, here are some tests below to reproduce the issue.
For polling transport this is even weirder when I test the whiteboard example I have nothing but then after the next ping I get an infinite loop of messages. Moreover all the payload encoder code is a mess mainly because we are using the wrong abstraction. What I'm going to do is to merge #555 so payload encoding uses streams. With stream it will be possible to merge gracefully the two channels without touching the encoder code.

In the meantime you can still fix the websocket code :).

#[tokio::test]
async fn volatile_only_arrives_via_polling() {
    let (socket_tx, mut socket_rx) = mpsc::channel(10);
    let mut svc = create_server(VolatileHandler { socket_tx }).await;
    let sid = create_polling_connection(&mut svc).await;

    let socket = socket_rx
        .recv()
        .await
        .expect("socket not received from on_connect");

    // Emit ONLY a volatile packet, nothing on the main channel.
    assert!(socket.emit_volatile("volatile_only"));

    let response = send_req(
        &mut svc,
        format!("transport=polling&sid={sid}"),
        http::Method::GET,
        None,
    )
    .await;
    assert_eq!(response, "volatile_only");
}

#[tokio::test]
async fn volatile_only_arrives_via_ws() {
    let (socket_tx, mut socket_rx) = mpsc::channel(10);
    let mut svc = create_server(VolatileHandler { socket_tx }).await;
    let mut stream = create_ws_connection(&mut svc).await;

    let socket = socket_rx
        .recv()
        .await
        .expect("socket not received from on_connect");

    let _open_packet = stream.next().await.unwrap().unwrap();

    let ping = stream.next().await.unwrap().unwrap();
    assert_eq!(ping, Message::Text("2".into()));

    assert!(socket.emit_volatile("volatile_only"));

    // The volatile must be delivered promptly, well before the next heartbeat
    // ping (300ms) would otherwise wake the main channel.
    let volatile = tokio::time::timeout(Duration::from_millis(100), stream.next()).await;

    let msg = volatile
        .expect("timeout: volatile-only packet was never flushed over websocket")
        .unwrap()
        .unwrap();
    assert_eq!(msg, Message::Text("4volatile_only".into()));
}

Totodore · 2026-07-03T21:16:42Z

+    loop {
+        // Priority: wait for and drain the main channel.
+        // Volatile events are checked after each main-channel cycle.
+        let items = match internal_rx.recv().await {


This is still not working if you only send volatile events because internal_rx will wait indefinitely (until next ping...) here and block everything, you need to use tokio::select! to poll concurrently internal_rx and volatile_rx.

Totodore · 2026-07-03T21:18:27Z

+            let val = volatile_rx.borrow_and_update().clone();
+            if let Some(packets) = val {
+                #[cfg(feature = "tracing")]
+                tracing::info!(sid = ?socket.id, "ws volatile check: flushing {:?}", &packets);


Suggested change

tracing::info!(sid = ?socket.id, "ws volatile check: flushing {:?}", &packets);

tracing::debug!(sid = ?socket.id, "ws volatile check: flushing {} packets", packets.len());

We should only log in trace/debug and avoid printing the content of the packets (this would be extremely inefficient).

Totodore · 2026-07-03T21:20:33Z

+async fn volatile_emit_broadcast_does_not_panic() {
+    let (_svc, io) = SocketIo::new_svc();
+
+    let (tx, rx) = std::sync::mpsc::channel();
+    io.ns("/", async move |socket: SocketRef| {
+        // Broadcast with volatile flag should complete without panicking
+        socket
+            .within("room")
+            .volatile()
+            .emit("event", &"data")
+            .await
+            .ok();
+        tx.send(()).unwrap();
+    });
+
+    io.new_dummy_sock("/", ()).await;
+    rx.recv().unwrap();
+}


Suggested change

async fn volatile_emit_broadcast_does_not_panic() {

let (_svc, io) = SocketIo::new_svc();

let (tx, rx) = std::sync::mpsc::channel();

io.ns("/", async move |socket: SocketRef| {

// Broadcast with volatile flag should complete without panicking

socket

.within("room")

.volatile()

.emit("event", &"data")

.await

.ok();

tx.send(()).unwrap();

});

io.new_dummy_sock("/", ()).await;

rx.recv().unwrap();

}

Why would it panic in the first place? What are we testing here?
If there is no panic there is no reasons for this tests which asserts nothing.

Totodore · 2026-07-03T21:21:31Z

+#[tokio::test]
+async fn io_volatile_composes() {
+    let (_svc, io) = SocketIo::new_svc();
+    io.ns("/", |_: SocketRef| async {});
+
+    io.new_dummy_sock("/", ()).await;
+
+    // Volatile on the io handle delegates to the default namespace
+    let _ = io.volatile();
+    let _ = io.of("/").unwrap().volatile();
+}


Suggested change

#[tokio::test]

async fn io_volatile_composes() {

let (_svc, io) = SocketIo::new_svc();

io.ns("/", |_: SocketRef| async {});

io.new_dummy_sock("/", ()).await;

// Volatile on the io handle delegates to the default namespace

let _ = io.volatile();

let _ = io.of("/").unwrap().volatile();

}

Same here, we want assertions, we dont need to test the API in itself.

Totodore · 2026-07-03T21:40:43Z

+    /// Returns a [`ConfOperators`] with the volatile flag set, so that any
+    /// subsequent `emit()` will drop the event instead of buffering it if the
+    /// client is not ready to receive it.
+    ///
+    /// # Example
+    /// ```
+    /// # use socketioxide::{SocketIo, extract::SocketRef};
+    /// # use serde::Serialize;
+    /// #[derive(Serialize)]
+    /// struct GameState { x: f64, y: f64 }
+    ///
+    /// let (_, io) = SocketIo::new_svc();
+    /// io.ns("/", async |socket: SocketRef| {
+    ///     socket.volatile().emit("position", &GameState { x: 1.0, y: 2.0 }).ok();
+    ///     socket.volatile().to("room1").emit("update", &42).await.ok();
+    /// });
+    /// ```


Replace with the md doc.

Totodore · 2026-07-03T21:44:55Z

+    /// Send data to the list of socket ids with volatile semantics.
+    /// Errors are silently discarded; packets may be dropped if the
+    /// transport is not ready.
+    fn send_many_volatile(&self, sids: BroadcastIter<'_>, data: Value) {


The default impl here is useless and really confusing.

Totodore · 2026-07-03T21:46:37Z

+        let sockets = self.sockets.read().unwrap();
+        for sid in sids {
+            if let Some(socket) = sockets.get(&sid) {
+                socket.send_raw_volatile(data.clone());
+            }
+        }


NIT: that's just to match with the functional pattern above.

Suggested change

let sockets = self.sockets.read().unwrap();

for sid in sids {

if let Some(socket) = sockets.get(&sid) {

socket.send_raw_volatile(data.clone());

}

}

sids.filter_map(|sid| sockets.get(&sid))

.for_each(|socket| socket.send_raw_volatile(data.clone()));

Totodore · 2026-07-03T21:51:05Z

+    let (tx, rx) = std::sync::mpsc::channel();
+    io.ns("/", async move |socket: SocketRef| {
+        let result = socket.volatile().emit("test", &json!({"key": "val"}));
+        tx.send(result).unwrap();


It would also be interesting to test that it returns false on the second call to volatile emit.

…sing tokio::select and unify documentation via include_str

LiamCarPer · 2026-07-06T16:20:35Z

Hey, took care of all the feedback:
Fixed the WS transport bug by switching to tokio::select! so volatile-only events don't get blocked waiting on the next ping.

Cleaned up the tests (swapped to tokio mpsc, added the overwrite test, and included your two new integration tests).

Moved the docs for io.volatile() and socket.volatile() to the md include.

Knocked out all the nits (ns.rs refactor, tracing level, default impl removal, etc.).

Everything is green now

Totodore · 2026-07-09T17:53:31Z

Hey @LiamCarPer, I took the liberty to fix some things myself on top of your PR. So #753 will likely superseed this one. In any case you will be in the contributors list.
Don't hesitate to check it.
Again thanks for the contribution!

LiamCarPer mentioned this pull request Jun 20, 2026

add volatile flag for emitting events from the server #602

Closed

Totodore added A-socketioxide Area related to socketioxide A-engineioxide Area related to engineioxide C-Feature-request Request for a feature labels Jun 21, 2026

Totodore requested changes Jun 21, 2026

View reviewed changes

LiamCarPer added 4 commits June 26, 2026 09:27

feat: implement support for volatile events with new VolatileSocket w…

0ed449c

…rapper and operator flags

feat(socketio,engineio): add volatile flag for emitting events

35d867b

docs: add warnings about potential message ordering issues with volat…

78e4f28

…ile events

refactor: replace VolatileSocket with operator flag and optimize vola…

8861111

…tile message loop in transport

LiamCarPer force-pushed the feat/volatile-flag branch from 90575ac to 8861111 Compare June 26, 2026 09:53

test: update v3_string_encoder call to include missing argument in po…

91a04de

…lling payload test

LiamCarPer added 2 commits June 26, 2026 19:05

refactor: replace volatile packet collection with lazy watch receiver…

fabc57b

… polling in encoders to capture packets arriving during encoding

trigger ci

06f547f

Totodore reviewed Jun 28, 2026

View reviewed changes

feat: implement comprehensive test coverage

0a12cca

Totodore requested changes Jun 29, 2026

View reviewed changes

Comment thread crates/engineioxide/src/transport/ws.rs Outdated

fix: move websocket flush after volatile check and add payload encode…

b150bb4

…r tests for volatile packets

fix: ensure volatile packets are correctly captured and flushed durin…

804212c

…g transport polling stalls

Totodore requested changes Jul 3, 2026

View reviewed changes

refactor: improve volatile event handling in engineioxide transport u…

e40d1ad

…sing tokio::select and unify documentation via include_str

Totodore closed this Jul 9, 2026

	tracing::info!(sid = ?socket.id, "ws volatile check: flushing {:?}", &packets);
	tracing::debug!(sid = ?socket.id, "ws volatile check: flushing {} packets", packets.len());

Uh oh!

Conversation

LiamCarPer commented Jun 20, 2026

Motivation

Solution

User-Facing API

Adapter Propagation

Uh oh!

codspeed-hq Bot commented Jun 20, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Merging this PR will not alter performance

Uh oh!

Totodore commented Jun 20, 2026

Uh oh!

LiamCarPer commented Jun 20, 2026

Uh oh!

LiamCarPer commented Jun 21, 2026

Uh oh!

Totodore commented Jun 21, 2026

Uh oh!

LiamCarPer commented Jun 21, 2026

Uh oh!

LiamCarPer commented Jun 21, 2026

Uh oh!

Totodore left a comment • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

LiamCarPer commented Jun 26, 2026

Uh oh!

Totodore left a comment • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

LiamCarPer commented Jun 29, 2026

Uh oh!

Totodore left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

LiamCarPer commented Jun 29, 2026

Uh oh!

Totodore commented Jun 29, 2026

Uh oh!

LiamCarPer commented Jun 29, 2026

Uh oh!

Totodore left a comment • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

LiamCarPer commented Jul 6, 2026

Uh oh!

Totodore commented Jul 9, 2026

Uh oh!

Reviewers

Assignees

codspeed-hq Bot commented Jun 20, 2026 •

edited

Loading

Totodore left a comment •

edited

Loading

Totodore left a comment •

edited

Loading

Totodore left a comment •

edited

Loading