[ADR] RPC streaming design doc

doc/dev/adr/0025-rpc-streaming.md

@@ -13,16 +13,30 @@
  - When exposed to the user, each response includes an index of where it was in the stream
  - Allow for multiple separate commands to be streamed simultaneously
    - Even the same command can be executed in parallel to itself?
+ - Allow for invoker to cancel streamed responses mid-stream
 
 ## Non-requirements
  - Different payload shapes per command response 
+ - "Client Streaming" RPC (multiples requests -> One command response)
+ - Bi-directional streaming RPC (multiples requests -> multiple responses)
+ - Allow for executor to cancel streamed responses mid-stream
 
 ## State of the art
 
-What does gRPC do?
+gRPC supports these patterns for RPC:
+- [Unary RPC](https://grpc.io/docs/what-is-grpc/core-concepts/#unary-rpc) (1 request message, 1 response message)
+- [Server streaming RPC](https://grpc.io/docs/what-is-grpc/core-concepts/#server-streaming-rpc) (1 request message, many response messages)
+- [Client streaming RPC](https://grpc.io/docs/what-is-grpc/core-concepts/#server-streaming-rpc) (many request messages, one response message)
+- [Bi-directional streaming RPC](https://grpc.io/docs/what-is-grpc/core-concepts/#bidirectional-streaming-rpc) (many request messages, many response messages)
+
+gRPC relies on the HTTP streaming protocol to delineate each message in the stream and to indicate the end of the stream.
+
+[gRPC also allows for either the client or server to cancel an RPC at any time](https://grpc.io/docs/what-is-grpc/core-concepts/#cancelling-an-rpc)
 
 ## Decision
 
+### API design, .NET
+
 Our command invoker base class will now include a new method ```InvokeCommandWithStreaming``` to go with the existing ```InvokeCommand``` method. 
 
 This new method will take the same parameters as ```InvokeCommand``` but will return an asynchronously iterable list (or callback depending on language?) of command response objects. 
@@ -95,11 +109,7 @@
 
 With this design, commands that use streaming are defined at codegen time. Codegen layer changes will be defined in a separate ADR, though.
 
-## Example with code gen
-
-TODO which existing client works well for long-running commands? Mem mon ("Report usage for 10 seconds at 1 second intervals")?
-
-### MQTT layer implementation
+### MQTT layer protocol
 
 #### Command invoker side
 
@@ -117,24 +127,47 @@
   - Each streamed response may contain an MQTT user property with name "__streamRespId" and value equal to that response's streaming response Id. This is an optional and user-provided value.
   - The final command response will include an MQTT user property "__isLastResp" with value "true" to signal that it is the final response in the stream.
     - A streaming command is allowed to have a single response. It must include the "__isLastResp" flag in that first/final response
-  - All **completed** streamed command responses will be added to the command response cache
-    - If we cache incompleted commands, will the cache hit just wait on cache additions to get the remaining responses?
-    - Cache exists for de-duplication, and we want that even for long-running RPC, right?
-    - Separate cache for data structure purposes?
+  - Cache is only updated once the stream has completed and it is updated to include all of the responses (in order) for the command so they can be re-played if the streaming command is invoked again by the same client
 
 - The command executor receives a command **without** "__streamResp" flag set to "true"
   - The command must be responded to without streaming
 
+### Cancellation support
+
+To avoid scenarios where long-running streaming responses are no longer wanted, we will want to support cancelling RPC calls. This feature is moreso applicable for RPC streaming, but the design allows for it to work for non-streaming RPC as well.
+
+#### Invoker side
+
+- The command invoker may cancel a normal or streaming RPC call at an arbitrary time by sending an MQTT message with: 
+  - The same MQTT topic as the invoked method
+  - The same correlation data as the invoked method 
+  - The user property "__cancelRpc" set to "true".
+  - No payload
+  - TODO what would API look like? gRPC uses cancellation token
+- The command invoker should still listen on the response topic for a response from the executor which may still contain a successful response (if cancellation was received after the command completed successfully)
+
+#### Executor side
+
+Regardless of if an RPC is streaming or not, upon receiving an MQTT message with the "__cancelRpc" flag set to "true", the command executor should:
+ - Notify the application layer that that RPC has been canceled if it is still running
+ - Send an MQTT message to the appropriate response topic with error code "canceled" to notify the invoker that the RPC has stopped and no further responses will be sent.
+
+If the executor receives a cancellation request for a command that has already completed, then the cancellation request should be ignored.
+
 ### Protocol version update
 
-This feature is not backwards compatible (old invoker can't communicate with new executor that may try to stream a response), so it requires a bump in our RPC protocol version from "1.0" to "2.0".
+This RPC streaming feature is not backwards compatible (new invoker can't initiate what it believes is a streaming RPC call on an old executor), so it requires a bump in our RPC protocol version from "1.0" to "2.0".
 
 TODO: Start defining a doc in our repo that defines what features are present in what protocol version.
 
+## Example with code gen
+
+TODO which existing client works well for long-running commands? Mem mon ("Report usage for 10 seconds at 1 second intervals")?
+
 ## Alternative designs considered
 
  - 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 call the ```InvokeCommandWithStreaming``` API on the command invoker side and that returned object isn't as easy to use for single responses
+   - 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 a separate protocol from RPC, give it its own client like ```CommandInvoker``` and ```TelemetrySender```
    - There is a lot of code re-use between RPC and streaming RPC so this would make implementation very inconvenient
    - This would introduce another protocol to version. Future RPC changes would likely be relevant to RPC streaming anyways, so this feels redundant.
@@ -149,15 +182,12 @@
    - RPC executor treats it like a non-streaming command, but adds the "__isLastResp" flag to the one and only response
  - RPC invoker tries to invoke a non-streaming command that the executor requires streaming on
    - Atypical case since codegen will prevent this
-   - But, for the sake of non-codegen users, a new error code "StreamingRequired" would be returned by the executor
-     - Or should this just be "invalid header" error since the executor expects the "__streamResp" header?
- - timeout per response vs overall?
+   - But, for the sake of non-codegen users, executor returns "invalid header" error pointing to the "__streamResp" header
+     - Invoker understands that, if the "invalid header" value is "__streamResp", it attempted a invoke a streaming method
+ - timeout per response vs overall? Both?
 
  ## Open Questions
 
-- Do we need to include response index user property on each streamed response?
-  - MQTT message ordering suggests this information can just be inferred by the command invoker
-- Command timeout/cancellation tokens in single vs streaming?
 - When to ack the streaming request?
-  - In normal RPC, request is Ack'd only after the method finishes invocation, but this would likely clog up Acks since streaming requests can take a while.
+  - In normal RPC, request is Ack'd only after the method finishes invocation. Waiting until a streamed RPC finishes could clog up Acks since streaming requests can take a while.
     - Ack after first response is generated?

doc/reference/error-model.md

@@ -151,7 +151,7 @@
     ExecutionException,
     MqttError,
     UnsupportedVersion,
-    StreamingRequired
+    Canceled,
 }
 ```
 
@@ -264,7 +264,7 @@
     EXECUTION_EXCEPTION,
     MQTT_ERROR,
     UNSUPPORTED_VERSION,
-    STREAMING_REQUIRED,
+    CANCELED,
 }
 ```
 
@@ -329,7 +329,7 @@
     ExecutionException,
     MqttError,
     UnsupportedVersion,
-    StreamingRequired,
+    Canceled,
 }
 ```
 
@@ -403,7 +403,7 @@
     ExecutionError
     MqttError
     UnsupportedVersion
-    StreamingRequired
+    Canceled
 }
 ```
 
@@ -428,7 +428,7 @@
 }
 ```
 
 The `AkriMqttError` struct must provide an `Error()` method, but the detais are omitted from this document:
 
 ```go
 func (err AkriMqttError) Error() string {
@@ -459,7 +459,7 @@
     EXECUTION_EXCEPTION = 10
     MQTT_ERROR = 11
     UNSUPPORTED_VERSION = 12
-    STREAMING_REQUIRED = 13
+    CANCELED = 13
 ```
 
 The Akri.Mqtt error type is defined as follows:
@@ -575,7 +575,7 @@
 | 400 | Bad Request | false | no | | invalid payload |
 | 408 | Request Timeout | false | yes | yes | timeout |
 | 415 | Unsupported Media Type | false | yes | yes | invalid header |
-| 452 | Streaming Required | false | no | no | streaming required |
+| 452 | Request Cancelled | false | no | no | canceled |
 | 500 | Internal Server Error | false | no | | unknown error |
 | 500 | Internal Server Error | false | yes | | internal logic error |
 | 500 | Internal Server Error | true | maybe | | execution error |

dotnet/src/Azure.Iot.Operations.Protocol/AkriSystemProperties.cs

@@ -80,9 +80,14 @@
         internal const string CommandInvokerId = ReservedPrefix + "invId";
 
         /// <summary>
-        /// Inidicates that an RPC request expects the executor to 
+        /// Inidicates that an RPC request expects the executor to stream one or many responses.
         /// </summary>
-        internal const string IsStreamingCommand = ReservedPrefix + "stream";
+        internal const string IsStreamingCommand = ReservedPrefix + "streamResp";
+
+        /// <summary>
+        /// Inidicates that an RPC request should be cancelled if it is still executing
+        /// </summary>
+        internal const string CancelCommand = ReservedPrefix + "cancelRpc";
 
         internal static bool IsReservedUserProperty(string name)
         {