Fix response listener registration

[sashacmc]

No description provided.

[gregmedd]

@sashacmc - I'm a little concerned about this change. There shouldn't be a reason to swap source and sink if I'm reading the zenoh uri mapping spec correctly.

[sashacmc]

@sashacmc - I'm a little concerned about this change. There shouldn't be a reason to swap source and sink if I'm reading the zenoh uri mapping spec correctly.

Yes, but from the client side of view server side sink and source will be reverted.
And in the rust implementation we have the same:
https://github.com/eclipse-uprotocol/up-transport-zenoh-rust/blob/main/src/utransport.rs#L470

[PLeVasseur]

Tagging @evshary for his feedback

[gregmedd]

@sashacmc - I'm a little concerned about this change. There shouldn't be a reason to swap source and sink if I'm reading the zenoh uri mapping spec correctly.

Yes, but from the client side of view server side sink and source will be reverted. And in the rust implementation we have the same: https://github.com/eclipse-uprotocol/up-transport-zenoh-rust/blob/main/src/utransport.rs#L470

It would be really helpful to know more / have some examples, but I think I get what you are saying. Over Zenoh, the response to a request comes via the same "topic" string that was used for the request. This is different than what uProtocol expects (i.e that source and sink get swapped from request to response), and so the values have to be un-swapped when registering a callback listener for RPC responses. This suggests that the source and sink addresses in the response message are ignored when looking up the callback and, in essence, only the request ID is used.

I have a few concerns about this. For one, we plan on replacing the one-callback-per-request model with a single callback for all requests. This would require non-trivial rework in this transport implementation even though both configurations should have near identical behavior per uProtocol spec. Additionally, this direct 1:1 mapping means that many other use cases in uProtocol may not work correctly; for example, it would not be possible to build an audit logger that listens to RPC requests and responses flowing in and out of a particular entity. The uProtocol spec allows for such a thing to be constructed - so long as it is authorized to do so, that logger would simply listen with the appropriate URI filters to receive copies of the messages. That does not appear possible with this model, particularly if the uProtocol spec is not followed with respect to the structure of the response message URIs.

@stevenhartley - do you have any input on this? Have I misunderstood the spec? Is it expected that transports may optimize certain communication modes at the cost of message-bus-like capabilities (e,g, monitoring other topics)?

[evshary]

For one, we plan on replacing the one-callback-per-request model with a single callback for all requests. This would require non-trivial rework in this transport implementation even though both configurations should have near identical behavior per uProtocol spec.

Perhaps I didn't fully understand the difference between the one-callback-per-request model and a single callback for all requests. I can elaborate on how Zenoh implementation (at least Rust) works now, and then see whether it matches your expectations.

The flow of RPC client:

1. register_listener(sink, source, callback) => register response callback
   • Detect this is response message type based on the combination of sink and source
   • Transform the sink and source to the key "sink/source"
   • Put the key-value pair {"sink/source", callback} into HashMap.
2. send(UMessage) => send the request message
   • Detect this is request message type based on the combination of sink and source
   • Get the callback from HashMap by using the sink and source in UMessage
   • send query (get) in Zenoh with that callback
   • Callback will be triggered while getting the reply
3. unregister_listener(sink, source, callback) => unregister response callback
   • Detect this is response message type based on the combination of sink and source
   • Transform the sink and source to the key "sink/source"
   • Remove the key-value pair {"sink/source", callback} from HashMap.

Additionally, this direct 1:1 mapping means that many other use cases in uProtocol may not work correctly; for example, it would not be possible to build an audit logger that listens to RPC requests and responses flowing in and out of a particular entity. The uProtocol spec allows for such a thing to be constructed - so long as it is authorized to do so, that logger would simply listen with the appropriate URI filters to receive copies of the messages. That does not appear possible with this model, particularly if the uProtocol spec is not followed with respect to the structure of the response message URIs.

RPC requests and responses are not available to record currently. We're using the mechanisms get and queryable in Zenoh, which is not possible to do the recording. @PLeVasseur and I discussed this before, and it might need some workaround here. We can have a further discussion if it is time to implement it.

[gregmedd]

@evshary and @sashacmc - if the swapped positions of the source / sink filters is only used in send() when an RPC request is detected to look up the corresponding listener, then that swap should happen in the send() code and not in registerListener(). At the time registerListener() is called, it can't be known with absolute certainty that the listener will correspond to an RPC request so swapping them there could produce unexpected behaviors.

Additionally, there may be multiple pending listeners that match a specific pattern if a particular RPC method receives a high volume of traffic, so that mechanism of guessing the match may also produce race conditions (unless the transport implementation is grabbing all matching callbacks).

You also mention this as a step for send():

Detect this is request message type based on the combination of sink and source

That seems to be overly complicated for detecting message type when the attributes structure contains a message type field. Shouldn't that be used instead within send()? It seems less error prone.

[evshary]

@evshary and @sashacmc - if the swapped positions of the source / sink filters is only used in send() when an RPC request is detected to look up the corresponding listener, then that swap should happen in the send() code and not in registerListener(). At the time registerListener() is called, it can't be known with absolute certainty that the listener will correspond to an RPC request so swapping them there could produce unexpected behaviors.

Both are fine for me. It's just a key for HashMap and is only used internally, so I don't have a strong opinion on where to do the swap.

Additionally, there may be multiple pending listeners that match a specific pattern if a particular RPC method receives a high volume of traffic, so that mechanism of guessing the match may also produce race conditions (unless the transport implementation is grabbing all matching callbacks).

Now I'm using Mutex to protect the callback hashmap to avoid that. Perhaps we need some benchmarks in the future to ensure this doesn't affect the performance too much.

You also mention this as a step for send():

Detect this is request message type based on the combination of sink and source

That seems to be overly complicated for detecting message type when the attributes structure contains a message type field. Shouldn't that be used instead within send()? It seems less error prone.

You're right. This is a typo. I'm using the attributes structure to check message type in send() now.
Only register_listener and unregister_listener need to detect the message type based on the combination of sink and source

[gregmedd]

I definitely have some questions about how zenoh is being used to transport uProtocol messages, but I think they reach outside the scope of this PR. I'll write up my thoughts / questions to share in the appropriate forum.

For now, this change looks good to me. I'm going to verify that this gets the RPC test passing then I'll move forward with merging it.

[gregmedd]

This still does not appear to pass the RPC client/server tests. I tried swapping out the zenoh implementation to exchange messages only over the pub/sub path and it worked fine, so I think the messages, attributes, and URIs are all correct.

I recommend rebasing to main to get the latest versions of the tests.

[sashacmc]

Hi @gregmedd, fixed, now test passes

[gregmedd]

LGTM. Discussion on use of queries and layer 2 behaviors in layer 1 here: eclipse-uprotocol/up-spec#229