Core implement ws server listener interface for subs and adverts #189

eloff · 2025-02-05T22:44:03Z

Add new callbacks to ServerListener:

fn on_subscribe(&self, _channel_id: ChannelId) {}
fn on_unsubscribe(&self, _channel_id: ChannelId) {}
fn on_client_advertise(&self, _channel: &ClientChannel) {}
fn on_client_unadvertise(&self, _channel_id: ClientChannelId) {}

This mirrors how the callbacks are defined in the Python implementation, and they function in a similar way (e.g. not firing for duplicate or erroneous requests)

I added RecordingServerListener to testutil to allow more easily testing ServerListener.

I modified existing tests to verify these callbacks are called at the appropriate times with the expected arguments, and not called for duplicate requests.

…nvoked

linear · 2025-02-05T22:44:06Z

FG-9723 [core] Implement WS server listener interface for subs & adverts

…r-interface-for-subs-adverts

bryfox

Looking good to me. As discussed, I think we can probably bring the server listener out from behind the feature now.

There's a client-publish example which uses part of this — it might be good to augment that with the advertisement handling.

bryfox · 2025-02-06T17:12:35Z

rust/foxglove/src/cow_vec.rs

        let final_state = vec.get().to_vec();

        // Old snapshot should still be valid and have original length
        assert_eq!(old_snapshot.len(), 3);
        // Both threads should see 5 items in their final state
        assert_eq!(final_state.len(), 5);
-        assert_eq!(thread_state.len(), 5);


I'm not sure why this is no longer desired, but the comment above is now stale.

I'll fix the comment, this was a race condition. It would be 4 items if the thread runs before the push(5) after starting the thread.

bryfox · 2025-02-06T17:14:34Z

rust/foxglove/src/tests/websocket.rs

    client_sender
        .send(Message::text(subscribe.to_string()))
        .await
        .expect("Failed to send");

    // Allow the server to process the subscription
    // FG-9723: replace this with an on_subscribe callback
+    // (whoops, that won't work either, unless we do something like polling the recording_listener)


Or maybe a way to flush the queues?

I think we should file an issue so we can clean up the comment above (not referencing a completed ticket) — whatever approach you think is best.

There doesn't seem to be a way to wait for the queue to be drained, other than polling it, checking is_empty - but that still doesn't mean the data was acted on, just that it was removed from the queue. So I think it's the wrong thing to do.

I'm not sure what a sensible way to do this would look like. It's easy enough to come up with something just for tests where we poll somehow and wait for the messages to be processed.

Is that just for tests, or do we need something more polished that we could expose to the user?

In my opinion, just for tests. I guess we don't necessarily need to poll the queue — we could have a test helper that polls, with a timeout, whatever future we want to wait on in the test (e.g. client_receiver.next())

A pattern I've used in the past is assert_eventually(conf: impl Fn() -> bool). Imagine we had something like:

let client: Arc<TestClient> = ...; tokio::spawn(client.clone().receive_forever()); for msg in [/* requests */] { client.send(msg).await.unwrap(); } let expected = vec![ /* responses */]; dbg!(&expected); assert_eventually(|| { expect == dbg!(client.get_received()) }).await; client.reset_received();

That future is buried at the bottom of Server::handle_connection, which is what makes it tricky. I'll create ticket to track it, but I'd have to experiment and think about it as to how we could implement it in a sensible way.

rust/foxglove/src/tests/websocket.rs

bryfox · 2025-02-06T17:48:35Z

rust/foxglove/src/websocket.rs

+    /// Callback invoked when a client advertises a client channel. Requires the "clientPublish" capability.
+    fn on_client_advertise(&self, _channel: &ClientChannel) {}
+    /// Callback invoked when a client unadvertises a client channel. Requires the "clientPublish" capability.
+    fn on_client_unadvertise(&self, _channel_id: ClientChannelId) {}


Not for this PR, but I'm wondering if we can design a trait in a way that if you implement one of the 'clientPublish' functions, it forces you to implement all — to help guide correct usage.

Yeah, I think that's a nice idea. I don't know exactly how that would look (or if we can do that.) Something to keep in mind.

gasmith

A few early comments on the trait, still reading through the rest.

gasmith · 2025-02-06T19:03:01Z

rust/foxglove/src/websocket.rs

 pub trait ServerListener: Send + Sync {
    /// Callback invoked when a client message is received.
-    fn on_message_data(&self, channel_id: ClientChannelId, payload: &[u8]);
+    fn on_message_data(&self, _channel_id: ClientChannelId, _payload: &[u8]) {}


Is the ClientChannelId globally unique, or locally-unique to a particular client? I presume the latter? Might IDs be reused, from the perspective of an implementer? We should probably update the rust doc for that type, and add some commentary to these methods to help guide user expectations.

We should add a reference to https://github.com/foxglove/ws-protocol/blob/main/docs/spec.md#client-message-data. Same goes for the other methods, I think.

It should only be unique per client, which suggests maybe we should pass in some kind of identifier for the client as well (what? SocketAddr? make up an id?). The Python implementation doesn't address that, and has these same signatures.

The payload is fine, the message only contains the channel id and the payload, and we pass those separately to the callback. I'll double-check the other methods, but I believe that holds as well.

I added a global integer id for clients, so the subscriber can tell them apart

gasmith · 2025-02-06T19:06:02Z

rust/foxglove/src/websocket.rs

    /// Callback invoked when a client message is received.
-    fn on_message_data(&self, channel_id: ClientChannelId, payload: &[u8]);
+    fn on_message_data(&self, _channel_id: ClientChannelId, _payload: &[u8]) {}
+    /// Callback invoked when a client subscribes to a channel.
+    fn on_subscribe(&self, _channel_id: ChannelId) {}


For server-advertised channels, we allocate IDs internally. The ID is recoverable, but we're asking trait implementation to do some extra legwork to map a channel ID back to something they understand (like a topic). Maybe we go a bit further.

Presumably we also need to pass the client identity, whatever that may be (perhaps a struct with a view over client metadata, so we can add getter methods later without breaking compatibility?). Actually, I think this comment applies to all of the methods in this trait.

What if we map the ChannelId back to the Channel and pass that instead to these two particular callbacks? That seems like it should work.

I implemented this

gasmith · 2025-02-06T19:11:53Z

rust/foxglove/src/websocket.rs

-/// Provides a mechanism for registering callbacks for
-/// handling client message events.
+/// Provides a mechanism for registering callbacks for handling client message events.
+/// All methods are optional.
 pub trait ServerListener: Send + Sync {


Seems like we might also want callbacks for client connect and disconnect, so that an implementer knows when to flush client-specific state (about channel advertisements, subscriptions, etc.).

The existing implementations don't go that far, but it seems like a good idea to me

gasmith · 2025-02-06T19:31:31Z

rust/foxglove/src/websocket.rs

+                for subscription_id in subscription_ids {
+                    if let Some((channel_id, _)) = subscriptions.remove_by_right(&subscription_id) {
+                        if let Some(handler) = self.server_listener.as_ref() {
+                            handler.on_unsubscribe(channel_id);


I don't think we should hold the subscriptions lock across the callback. If we do, we need to document that the SDK is non-reentrant.

Yeah, that's a good point

gasmith · 2025-02-06T19:36:57Z

rust/foxglove/src/websocket.rs

+                subscription.id
+            );
+            if let Some(handler) = self.server_listener.as_ref() {
+                handler.on_subscribe(subscription.channel_id);


Probably ought to drop the channels lock before making these callbacks.

The main downside, I suppose, is that we might process a subscription callback after a channel is removed from the log context. But I think that's a straightforward corner case to document, and probably not one that makes much of a difference to an implementation of this trait.

gasmith · 2025-02-06T19:40:19Z

rust/foxglove/src/websocket.rs

+    /// Callback invoked when a client subscribes to a channel.
+    fn on_subscribe(&self, _channel_id: ChannelId) {}
+    /// Callback invoked when a client unsubscribes from a channel.
+    fn on_unsubscribe(&self, _channel_id: ChannelId) {}


We should maybe document the fact that we do not invoke this callback for a subscribed channel that is removed from the log context, and unadvertised to the client. Maybe that's obvious, but it doesn't hurt to call it out.

gasmith · 2025-02-06T19:41:34Z

rust/foxglove/src/websocket.rs

+            for id in channel_ids
+                .iter()
+                .cloned()
+                .filter(|id| !channels_not_found.contains(id))


Since we have to alloc a new vec anyway, we might as well just store the channel IDs we did find and avoid the possible O(N^2) over the client payload.

channels_not_found is basically an error case that's not going to happen often, so I would expect this approach to perform best on average, but with worse worst-case performance (which is why I did it this way)

gasmith · 2025-02-06T19:45:38Z

rust/foxglove/src/websocket.rs

 pub trait ServerListener: Send + Sync {
    /// Callback invoked when a client message is received.
-    fn on_message_data(&self, channel_id: ClientChannelId, payload: &[u8]);
+    fn on_message_data(&self, _channel_id: ClientChannelId, _payload: &[u8]) {}


Seems we lookup the ClientChannel before invoking this callback. Maybe we should pass a (view over) ClientChannel, instead of asking the trait implementation to maintain its own <ClientChannelId, ClientChannel> map.

gasmith · 2025-02-06T19:48:59Z

rust/foxglove/src/websocket.rs

@@ -171,60 +182,30 @@ impl ConnectedClient {
            self.send_error(format!("Invalid message: {message}"));
            return;
        };
+        let Some(server) = self.server.upgrade() else {
+            tracing::error!("Server closed");


Maybe a bit loud? Is this something the user can do anything about?

gasmith · 2025-02-06T19:49:41Z

rust/foxglove/src/testutil.rs

 pub use log_context::GlobalContextTest;
 pub use log_sink::{ErrorSink, MockSink, RecordingSink};
+use parking_lot::Mutex;
+
+#[allow(dead_code)]


Still necessary?

Turns out not needed anymore

gasmith · 2025-02-06T19:58:29Z

rust/foxglove/src/tests/websocket.rs

    client_sender
        .send(Message::text(subscribe.to_string()))
        .await
        .expect("Failed to send");

    // Allow the server to process the subscription
    // FG-9723: replace this with an on_subscribe callback
+    // (whoops, that won't work either, unless we do something like polling the recording_listener)


A pattern I've used in the past is assert_eventually(conf: impl Fn() -> bool). Imagine we had something like:

let client: Arc<TestClient> = ...; tokio::spawn(client.clone().receive_forever()); for msg in [/* requests */] { client.send(msg).await.unwrap(); } let expected = vec![ /* responses */]; dbg!(&expected); assert_eventually(|| { expect == dbg!(client.get_received()) }).await; client.reset_received();

…ework invoking code to not hold locks while invoking callbacks, pass Channel refs instead of channel ids

…r-interface-for-subs-adverts

gasmith

LG, just some minor dangling comments.

gasmith · 2025-02-07T00:02:18Z

rust/foxglove/src/websocket.rs

+    }
+    /// Callback invoked when a client subscribes to a channel.
+    /// Only invoked if the channel is associated with the server and isn't already subscribed to by the client.
+    fn on_subscribe(&self, _client_id: ClientId, _channel_id: Arc<Channel>) {}


Can rename these args as channel now, instead of channel_id.

Also, maybe we just pass a &Channel so that implementations aren't tempted to clone or hold on to the arc.

gasmith · 2025-02-07T00:06:02Z

rust/foxglove/src/websocket.rs

@@ -96,11 +103,23 @@ pub(crate) struct Server {
 /// handling client message events.
 pub trait ServerListener: Send + Sync {
    /// Callback invoked when a client message is received.
-    fn on_message_data(&self, channel_id: ClientChannelId, payload: &[u8]);
+    fn on_message_data(&self, _client_id: ClientId, _channel_id: ClientChannelId, _payload: &[u8]) {


Should we also pass &ClientChannel here (and on_client_unadvertise)? We're already doing the lookup before invoking the callback.

Based on the slack discussion, I went with wrapper view structs for all the argument types, exposing the bare minimum for now.

We'll need to provide some way to lookup a ClientChannel to get the full data, as that is now the one place where we provide less data than the existing callbacks in the original Python implementation.

…r-interface-for-subs-adverts

gasmith

LG. I suspect we'll need to expand ClientChannelView with schema information for it to be useful, but we can cross that bridge when we get there.

eloff added 6 commits February 4, 2025 18:38

add on_subscribe and on_unsubscribe to ServerListener

b6be1aa

implement on_client_advertise and on_client_unadvertise

bdd83e3

don't call on_unadvertise if channel wasn't advertised

338b6c2

update test_client_advertising to check the listener callbacks were i…

4cc637c

…nvoked

modify tests to check for subscribe and unsubscribe callbacks

40f16c3

remove debug printlns

0bae9c4

eloff requested review from gasmith and bryfox February 5, 2025 22:44

eloff added 3 commits February 5, 2025 15:46

Merge branch 'main' into dan/fg-9723-core-implement-ws-server-listene…

b6f81f6

…r-interface-for-subs-adverts

fix flakey concurrent test

0f4812e

Merge branch 'main' into dan/fg-9723-core-implement-ws-server-listene…

5b25049

…r-interface-for-subs-adverts

bryfox reviewed Feb 6, 2025

View reviewed changes

gasmith reviewed Feb 6, 2025

View reviewed changes

eloff added 5 commits February 6, 2025 14:25

update comments around sleeps to point at FG-10395

71f4c47

rework handler callbacks to accept client id (and add a client id), r…

b2e70f5

…ework invoking code to not hold locks while invoking callbacks, pass Channel refs instead of channel ids

Merge branch 'main' into dan/fg-9723-core-implement-ws-server-listene…

89adf5b

…r-interface-for-subs-adverts

tweak docs a little and fix unstable tests

5732c56

Merge branch 'main' into dan/fg-9723-core-implement-ws-server-listene…

53276d2

…r-interface-for-subs-adverts

eloff requested review from bryfox and gasmith February 6, 2025 23:33

gasmith approved these changes Feb 7, 2025

View reviewed changes

eloff added 2 commits February 7, 2025 10:48

use wrapped view types in the ServerListener callbacks

aa999f8

Merge branch 'main' into dan/fg-9723-core-implement-ws-server-listene…

3a8f318

…r-interface-for-subs-adverts

gasmith approved these changes Feb 7, 2025

View reviewed changes

eloff merged commit 62680fe into main Feb 7, 2025
27 checks passed

eloff deleted the dan/fg-9723-core-implement-ws-server-listener-interface-for-subs-adverts branch February 7, 2025 18:55

Core implement ws server listener interface for subs and adverts #189

Core implement ws server listener interface for subs and adverts #189

Conversation

eloff commented Feb 5, 2025 • edited Loading

linear bot commented Feb 5, 2025

bryfox left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

gasmith Feb 6, 2025 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

gasmith left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

eloff Feb 6, 2025 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

gasmith Feb 6, 2025 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

gasmith Feb 6, 2025 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

gasmith Feb 6, 2025 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

gasmith Feb 6, 2025 • edited Loading

Choose a reason for hiding this comment

gasmith left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

eloff Feb 7, 2025 • edited Loading

Choose a reason for hiding this comment

gasmith left a comment

Choose a reason for hiding this comment

eloff commented Feb 5, 2025 •

edited

Loading

gasmith Feb 6, 2025 •

edited

Loading

eloff Feb 6, 2025 •

edited

Loading

gasmith Feb 6, 2025 •

edited

Loading

gasmith Feb 6, 2025 •

edited

Loading

gasmith Feb 6, 2025 •

edited

Loading

gasmith Feb 6, 2025 •

edited

Loading

eloff Feb 7, 2025 •

edited

Loading