[ADR] RPC streaming design doc#952
Conversation
|
We should consider the scenario where the executor restarts while it has only received half of the request stream, and call out the executor handling in that case explicitly. |
| /// If the stream has not been cancelled, this will return null. If the stream has been cancelled, but no user properties were | ||
| /// provided in that cancellation request, this will return null. | ||
| /// </remarks> | ||
| Dictionary<string, string>? GetCancellationRequestUserProperties(); |
There was a problem hiding this comment.
why not have these as some information provided with the cancellation? (Or is this just a dotnet limitation)
There was a problem hiding this comment.
This is kind of a .NET limitation. For all things cancellation, we use the cancellation token class. But the cancellation token class has no user properties.
There was a problem hiding this comment.
I am curious how you would approach this in Rust though because I'm not in love with my current .NET approach
There was a problem hiding this comment.
I don't have something exact, but I imagine we'd do some sort of wrapper around a standard cancellation token to expose these fns. Here's a very rough psuedo code that matches what you have here in dotnet (although I skipped the timeout for nowl)
pub struct RPCStream {
...
}
impl RPCStream {
pub async fn recv_next_entry() -> Option<T>;
pub async fn cancel(user_properties) -> Result;
// not sure if this would be needed in rust, but left it for now in case
pub fn is_cancelled() -> bool;
// this is the rust cancellation pattern that allows you to `await` until this is cancelled
pub async fn cancelled() -> UserProperties;
}
| - Allow for an arbitrary number of command requests and responses for a single command invocation | ||
| - The total number of requests and responses does not need to be known before the first request/response is sent | ||
| - The total number of entries in a stream is allowed to be 1 | ||
| - When exposed to the user, each request and response includes an index of where it was in the stream |
There was a problem hiding this comment.
Aren't they already ordered by __ts and __ft ?
| /// <param name="streamExchangeTimeout">The timeout between the beginning of the request stream and the end of both the request and response stream.</param> | ||
| /// <param name="cancellationToken">Cancellation token. Signalling this will also make a single attempt to notify the executor of the cancellation.</param> | ||
| /// <returns>The stream of responses.</returns> | ||
| public async Task<IStreamContext<StreamingExtendedResponse<TResp?>> InvokeStreamingCommandAsync( |
There was a problem hiding this comment.
Can we please use IAsyncEnumerable<StreamingExtendedResponse<TResp?>> as return value?
There was a problem hiding this comment.
I believe the reason I got away from that design was because we need a way to cancel the stream using the returned type, but we will follow up on that again
| - API design is messy because a command invoker/executor should not expose streaming command APIs if they have no streaming commands | ||
| - Caching behavior of normal RPC doesn't fit well with streamed RPCs which may grow indefinitely large | ||
|
|
||
|
|
There was a problem hiding this comment.
In case of protocol error or exception in executor, will be __stat not equal to 200 automatically seen as last message?
| - Caching behavior of normal RPC doesn't fit well with streamed RPCs which may grow indefinitely large | ||
|
|
||
|
|
||
| ## Appendix |
There was a problem hiding this comment.
Can each streaming message contain application level errors?
| - *The only limitation is that the invoker must initiate the RPC streaming with a request | ||
| - Allow for invoker/executor to end their own request/response stream gracefully at any time | ||
| - For instance, if the executor doesn't know if a response will be the last one prior to sending it, the executor should still be capable of ending the response stream later without sending another fully-fledged payload | ||
|
|
There was a problem hiding this comment.
Do we want to make it a requirement that a hypothetical chunking implementation could be built on this proposed streaming API?
There was a problem hiding this comment.
I think that would be a good goal!
| - *The only limitation is that the invoker must initiate the RPC streaming with a request | ||
| - Allow for invoker/executor to end their own request/response stream gracefully at any time | ||
| - For instance, if the executor doesn't know if a response will be the last one prior to sending it, the executor should still be capable of ending the response stream later without sending another fully-fledged payload | ||
|
|
There was a problem hiding this comment.
I think that would be a good goal!
| /// <returns>The stream of responses.</returns> | ||
| public async Task<IStreamContext<StreamingExtendedResponse<TResp?>> InvokeStreamingCommandAsync( | ||
| IAsyncEnumerable<StreamingExtendedRequest<TReq>> requests, | ||
| StreamRequestMetadata? streamRequestMetadata = null, |
There was a problem hiding this comment.
would this metadata (here and in the executor fn as well) be per item in the stream or only one for the entire stream?
|
|
||
| ```0:false:false:10000```: The first (and not last) message in a request stream where the RPC should timeout beyond 10 seconds | ||
|
|
||
| ```3:true:false```: The third and final message in a response stream |
There was a problem hiding this comment.
what happens if the executor sends a response with isLast = true, but the invoker either has another request in flight or would like to send more requests on the same stream?
There was a problem hiding this comment.
or should the executor not send isLast=true until the invoker has sent a request with isLast=true?
|
|
||
| Upon receiving an MQTT message in the response stream with the 'isLast' flag set in the '__stream' metadata, the streaming command invoker should notify the user that the stream of responses has ended. This particular message should not contain any payload or other user properties, so the message _should not_ be propagated to the user as if it were part of the response stream. [See here for more details on why this ```isLast``` flag is an independent message](#islast-message-being-its-own-message). | ||
|
|
||
| By default, the streaming command invoker will acknowledge all request messages it receives as soon as they are given to the user. Users may opt into manual acknowledgements, though. Opting into manual acknowledgements allows the user time to "process" each response as necessary before forgoing re-delivery from the broker if the invoker crashes unexpectedly. |
There was a problem hiding this comment.
if the invoker crashes unexpectedly, it won't be able to process the rest of the stream anyways since it will have lost the context of the request(s) that it sent
There was a problem hiding this comment.
related question for both invoker and executor, because I remember .net might be slightly different than rust - do the current .net invoker and executor both expose manual acking? I know rust doesn't for either
| - The appropriate streaming metadata [see above](#streaming-user-property) | ||
| - The serialized payload as provided by the user's response object | ||
| - Any user-definied metadata as specified in the ```ExtendedStreamingResponse``` | ||
| - QoS 1 |
There was a problem hiding this comment.
similar comment as on the invoker, would be good to specify if any of the additional standard rpc response metadata is used (for example, are the error properties supported in the same way or not?) https://github.com/Azure/iot-operations-sdks/blob/10f55dcab314135a35e7ef362c223d02d066b98c/doc/reference/message-metadata.md#response-message
|
|
||
| Also unlike normal RPC, the streaming command executor will not provide any re-play cache support. This is because streams may grow indefinitely in length and size so re-playing a response stream isn't feasible. | ||
|
|
||
| The streaming command executor will provide de-dupe caching of received requests to account for QoS 1 messages potentially being re-delivered. The streaming command invoker will de-dup using a the combination of the correlationId of the stream and the index of the message within that stream. The de-dup cache entries for a stream should be cleared once the stream has finished (gracefully or otherwise). |
There was a problem hiding this comment.
is this just a copy and paste typo, or is there something from the invoker that should be relevant here?
And from the timeout section, shouldn't the cache be maintained for some time after the timeout to ignore any additional requests that come in late?
| The streaming command executor will provide de-dupe caching of received requests to account for QoS 1 messages potentially being re-delivered. The streaming command invoker will de-dup using a the combination of the correlationId of the stream and the index of the message within that stream. The de-dup cache entries for a stream should be cleared once the stream has finished (gracefully or otherwise). | |
| The streaming command executor will provide de-dupe caching of received requests to account for QoS 1 messages potentially being re-delivered. The streaming command executor will de-dup using the combination of the correlationId of the stream and the index of the message within that stream. The de-dup cache entries for a stream should be cleared once the stream has finished (gracefully or otherwise) plus some buffer if it has no rpc level timeout or after 2x the rpc timeout plus buffer to not treat post-timeout messages as initiating a new stream. |
| - If the invoker's session is lost, then the RPC will timeout | ||
| - Executor side isn't connected when invoker sends first request | ||
| - Depending on broker behavior, invoker will receive a "no matching subscribers" puback | ||
| - Seems like a scenario we would want to retry? |
There was a problem hiding this comment.
probably best to give back to the user to decide whether to retry or not - if things just weren't configured properly, there might never be an executor available and we might be retrying forever
|
|
||
| - Allow the command executor to decide at run time of each command if it will stream responses independent of the command invoker's request | ||
| - This would force users to always call the ```InvokeCommandWithStreaming``` API on the command invoker side and that returned object isn't as easy to use for single responses | ||
| - Treat streaming RPC as the same protocol as RPC |
There was a problem hiding this comment.
being able to support chunking as part of regular RPC (and using streaming under the hood) might require bringing some of these alternatives back into the discussion, but let's see!
TODO: #1107 (comment)