Extending execution context for retries

Worker.nuspec

@@ -2,7 +2,7 @@
 <package xmlns="http://schemas.microsoft.com/packaging/2012/06/nuspec.xsd">
   <metadata>
     <id>Microsoft.Azure.Functions.NodeJsWorker</id>
-    <version>2.1.0$version$</version>
+    <version>2.1.1$version$</version>
     <authors>Microsoft</authors>
     <owners>Microsoft</owners>
     <requireLicenseAcceptance>false</requireLicenseAcceptance>

azure-functions-language-worker-protobuf/README.md

@@ -61,7 +61,7 @@ mkdir %MSGDIR%
 
 set OUTDIR=%MSGDIR%\DotNet
 mkdir %OUTDIR%
-%GRPC_TOOLS_PATH%\protoc.exe %PROTO% --csharp_out %OUTDIR% --grpc_out=%OUTDIR% --plugin=protoc-gen-grpc=%GRPC_TOOLS_PATH%\grpc_csharp_plugin.exe --proto_path=%PROTO_PATH% --proto_path=%PROTOBUF_TOOLS% 
+%GRPC_TOOLS_PATH%\protoc.exe %PROTO% --csharp_out %OUTDIR% --grpc_out=%OUTDIR% --plugin=protoc-gen-grpc=%GRPC_TOOLS_PATH%\grpc_csharp_plugin.exe --proto_path=%PROTO_PATH% --proto_path=%PROTOBUF_TOOLS%
 ```
 ## JavaScript
 In package.json, add to the build script the following commands to build .js files and to build .ts files. Use and install npm package `protobufjs`.
@@ -81,7 +81,10 @@ In pom.xml add following under configuration for this plugin
 <protoSourceRoot>${basedir}/<path to this repo>/azure-functions-language-worker-protobuf/src/proto</protoSourceRoot>
 
 ## Python
---TODO
+```
+python -m pip install -e .[dev] -U
+python setup.py build
+```
 
 ## Contributing
 

azure-functions-language-worker-protobuf/src/proto/FunctionRpc.proto

@@ -71,6 +71,10 @@ message StreamingMessage {
     FunctionEnvironmentReloadRequest function_environment_reload_request = 25;
 
     FunctionEnvironmentReloadResponse function_environment_reload_response = 26;
+
+    // Ask the worker to close any open shared memory resources for a given invocation
+    CloseSharedMemoryResourcesRequest close_shared_memory_resources_request = 27;
+    CloseSharedMemoryResourcesResponse close_shared_memory_resources_response = 28;
   }
 }
 
@@ -201,6 +205,17 @@ message FunctionEnvironmentReloadResponse {
   StatusResult result = 3;
 }
 
+// Tell the out-of-proc worker to close any shared memory maps it allocated for given invocation
+message CloseSharedMemoryResourcesRequest {
+  repeated string map_names = 1;
+}
+
+// Response from the worker indicating which of the shared memory maps have been successfully closed and which have not been closed
+// The key (string) is the map name and the value (bool) is true if it was closed, false if not
+message CloseSharedMemoryResourcesResponse {
+  map<string, bool> close_map_results = 1;
+}
+
 // Host tells the worker to load a Function
 message FunctionLoadRequest {
   // unique function identifier (avoid name collisions, facilitate reload case)
@@ -263,6 +278,9 @@ message InvocationRequest {
 
   // Populates activityId, tracestate and tags from host
   RpcTraceContext trace_context = 5;
+
+  // Current retry context
+  RetryContext retry_context = 6;
 }
 
 // Host sends ActivityId, traceStateString and Tags from host
@@ -277,6 +295,18 @@ message RpcTraceContext {
 	map<string, string> attributes = 3;
 }
 
+// Host sends retry context for a function invocation
+message RetryContext {
+  // Current retry count
+  int32 retry_count = 1;
+
+  // Max retry count
+  int32 max_retry_count = 2;
+
+  // Exception that caused the retry
+  RpcException exception = 3;
+}
+
 // Host requests worker to cancel invocation
 message InvocationCancel {
   // Unique id for invocation
@@ -318,6 +348,34 @@ message TypedData {
   }
 }
 
+// Specify which type of data is contained in the shared memory region being read
+enum RpcDataType {
+  unknown = 0;
+  string = 1;
+  json = 2;
+  bytes = 3;
+  stream = 4;
+  http = 5;
+  int = 6;
+  double = 7;
+  collection_bytes = 8;
+  collection_string = 9;
+  collection_double = 10;
+  collection_sint64 = 11;
+}
+
+// Used to provide metadata about shared memory region to read data from
+message RpcSharedMemory {
+  // Name of the shared memory map containing data
+  string name = 1;
+  // Offset in the shared memory map to start reading data from
+  int64 offset = 2;
+  // Number of bytes to read (starting from the offset)
+  int64 count = 3;
+  // Final type to which the read data (in bytes) is to be interpreted as
+  RpcDataType type = 4;
+}
+
 // Used to encapsulate collection string
 message CollectionString {
   repeated string string = 1;
@@ -343,8 +401,13 @@ message ParameterBinding {
   // Name for the binding
   string name = 1;
 
-  // Data for the binding
-  TypedData data = 2;
+  oneof rpc_data {
+    // Data for the binding
+    TypedData data = 2;
+
+    // Metadata about the shared memory region to read data from
+    RpcSharedMemory rpc_shared_memory = 3;
+  }
 }
 
 // Used to describe a given binding on load

azure-pipelines.yml

@@ -1,5 +1,5 @@
 variables: {
-  WORKER_VERSION: '2.1.0',
+  WORKER_VERSION: '2.1.1',
   NODE_8: '8.x',
   NODE_10: '10.x',
   NODE_12: '12.x',

e2e-nightly-cli-azure-pipelines.yml

@@ -1,5 +1,5 @@
 variables: {
-  WORKER_VERSION: '2.1.0',
+  WORKER_VERSION: '2.1.1',
   NODE_12: '12.x'
 }
 name: $(WORKER_VERSION)-$(Date:yyyyMMdd)$(Rev:.r)

e2e-nightly-site-azure-pipelines.yml

@@ -1,5 +1,5 @@
 variables: {
-  WORKER_VERSION: '2.1.0',
+  WORKER_VERSION: '2.1.1',
   NODE_12: '12.x'
 }
 name: $(WORKER_VERSION)-$(Date:yyyyMMdd)$(Rev:.r)

src/Context.ts

@@ -72,10 +72,11 @@ class InvocationContext implements Context {
     constructor(info: FunctionInfo, request: rpc.IInvocationRequest, logCallback: LogCallback, callback: ResultCallback) {
         this.invocationId = <string>request.invocationId;
         this.traceContext = fromRpcTraceContext(request.traceContext);
-        const executionContext = {
+        const executionContext = <ExecutionContext>{
             invocationId: this.invocationId,
             functionName: <string>info.name,
-            functionDirectory: <string>info.directory
+            functionDirectory: <string>info.directory,
+            retryContext: request.retryContext
         };
         this.executionContext = executionContext;
         this.bindings = {};

src/WorkerChannel.ts

@@ -80,7 +80,7 @@ export class WorkerChannel implements IWorkerChannel {
           systemError(`Worker ${workerId} malformed message`, msgError);
           throw new InternalException(msgError);
         }
-        oldWrite.apply(eventStream, arguments);
+        oldWrite.apply(eventStream, [msg]);
     }
   }
 
@@ -212,7 +212,7 @@ export class WorkerChannel implements IWorkerChannel {
       this.log({
         invocationId: msg.invocationId,
         category: `${info.name}.Invocation`,
-        message: format.apply(null, args),
+        message: format.apply(null, <[any, any[]]>args),
         level: level,
         logCategory: category
       });

src/public/Interfaces.ts

@@ -177,6 +177,37 @@ export interface ExecutionContext {
      * The directory your function is in (this is the parent directory of this function's function.json).
      */
     functionDirectory: string;
+    /**
+     * The retry context of the current funciton execution. The retry context of the current function execution. Equals null if retry policy is not defined or it's the first function execution.
+     */
+    retryContext?: RetryContext;
+}
+
+export interface RetryContext {
+    /**
+     * Current retry count of the function executions.
+     */
+    retryCount: number;
+    /**
+     * Max retry count is the maximum number of times an execution is retried before eventual failure. A value of -1 means to retry indefinitely.
+     */
+    maxRetryCount: number;
+    /**
+     * Exception that caused the retry
+     */
+    exception?: Exception;
+}
+
+export interface Exception {
+
+    /** Exception source */
+    source?: (string|null);
+
+    /** Exception stackTrace */
+    stackTrace?: (string|null);
+
+    /** Exception message */
+    message?: (string|null);
 }
 
 /**

types/public/Interfaces.d.ts

@@ -160,6 +160,32 @@ export interface ExecutionContext {
      * The directory your function is in (this is the parent directory of this function's function.json).
      */
     functionDirectory: string;
+    /**
+     * The retry context of the current funciton execution. The retry context of the current function execution. Equals null if retry policy is not defined or it's the first function execution.
+     */
+    retryContext?: RetryContext;
+}
+export interface RetryContext {
+    /**
+     * Current retry count of the function executions.
+     */
+    retryCount: number;
+    /**
+     * Max retry count is the maximum number of times an execution is retried before eventual failure. A value of -1 means to retry indefinitely.
+     */
+    maxRetryCount: number;
+    /**
+     * Exception that caused the retry
+     */
+    exception?: Exception;
+}
+export interface Exception {
+    /** Exception source */
+    source?: (string | null);
+    /** Exception stackTrace */
+    stackTrace?: (string | null);
+    /** Exception message */
+    message?: (string | null);
 }
 /**
  * TraceContext information to enable distributed tracing scenarios.