CORE-3968 Add new types for flow event retry and exception reporting (#342)

- Add new types for flow event retry and reporting exceptions to user code
- Add flow wake up schedule to checkpoint

Author: owenstanford

data/avro-schema/src/main/resources/avro/net/corda/data/flow/event/WakeUpWithException.avsc

@@ -0,0 +1,13 @@
+{
+  "type": "record",
+  "name": "WakeUpWithException",
+  "namespace": "net.corda.data.flow.event",
+  "doc": "The WakeUpWithException event is used to report an exception back to user code",
+  "fields": [
+    {
+      "name": "error",
+      "type": "net.corda.data.ExceptionEnvelope",
+      "doc": "The error to be thrown back to the user code"
+    }
+  ]
+}

data/avro-schema/src/main/resources/avro/net/corda/data/flow/output/FlowStatus.avsc

@@ -33,6 +33,7 @@
         "symbols": [
           "START_REQUESTED",
           "RUNNING",
+          "RETRYING",
           "COMPLETED",
           "FAILED"
         ]

data/avro-schema/src/main/resources/avro/net/corda/data/flow/state/Checkpoint.avsc

@@ -39,6 +39,17 @@
         "items": "net.corda.data.flow.FlowStackItem"
       },
       "doc": "Used to track calls to sub-flows and their relationship to sessions."
+    },
+    {
+      "name": "retryState",
+      "type": ["null", "net.corda.data.flow.state.RetryState"],
+      "default": null,
+      "doc": "Optional retry information for a failed flow event. Setting this field marks the flow as retrying."
+    },
+    {
+      "name": "maxFlowSleepDuration",
+      "type": "int",
+      "doc": "The maximum time a flow can sleep, before a Wakeup event is generated (milliseconds)"
     }
   ]
 }

data/avro-schema/src/main/resources/avro/net/corda/data/flow/state/RetryState.avsc

@@ -0,0 +1,39 @@
+{
+  "type": "record",
+  "name": "RetryState",
+  "namespace": "net.corda.data.flow.state",
+  "doc": "The Retry State records the need to retry a failed event on the flow Checkpoint",
+  "fields": [
+    {
+      "name": "retryCount",
+      "type": "int",
+      "doc": "The current retry count, set to 0 for the initial failure"
+    },
+    {
+      "name": "failedEvent",
+      "type": "net.corda.data.flow.event.FlowEvent",
+      "doc": "Copy of the event that caused the failure"
+    },
+    {
+      "name": "error",
+      "type": "net.corda.data.ExceptionEnvelope",
+      "doc": "The original error that caused the retry"
+    },
+    {
+      "name": "firstFailureTimestamp",
+      "type": {
+        "type": "long",
+        "logicalType": "timestamp-millis"
+      },
+      "doc": "The timestamp of when the first exception occurred that triggered a retry"
+    },
+    {
+      "name": "lastFailureTimestamp",
+      "type": {
+        "type": "long",
+        "logicalType": "timestamp-millis"
+      },
+      "doc": "The timestamp of when the last exception occurred that triggered a retry (this will be the same as firstRetryTimestamp for a first time failure"
+    }
+  ]
+}

data/config-schema/src/main/kotlin/net/corda/schema/configuration/FlowConfig.kt

@@ -4,4 +4,7 @@ object FlowConfig {
         const val SESSION_MESSAGE_RESEND_WINDOW = "session.messageResendWindow"
         const val SESSION_HEARTBEAT_TIMEOUT_WINDOW = "session.heartbeatTimeout"
         const val SESSION_P2P_TTL = "session.p2pTTL"
+        const val PROCESSING_MAX_RETRY_ATTEMPTS = "processing.maxRetryAttempts"
+        const val PROCESSING_MAX_RETRY_DELAY = "processing.maxRetryDelay"
+        const val PROCESSING_MAX_FLOW_SLEEP_DURATION = "processing.maxFlowSleepDuration"
 }

data/config-schema/src/main/resources/net/corda/schema/configuration/flow/1.0/corda.flow.json

@@ -5,6 +5,30 @@
   "description": "Configuration schema for the flow subsection.",
   "type": "object",
   "properties": {
+    "processing": {
+      "description": "Settings for flow pipeline processing",
+      "type": "object",
+      "properties": {
+        "maxRetryAttempts": {
+          "description": "The maximum number of retry attempts a transient error will be retried before failing the flow. a value of zero disables retries.",
+          "type": "integer",
+          "minimum": 0,
+          "default": 5
+        },
+        "maxRetryDelay": {
+          "description": "The maximum delay before a retry is scheduled, in milliseconds",
+          "type": "integer",
+          "minimum": 1000,
+          "default": 16000
+        },
+        "maxFlowSleepDuration": {
+          "description": "The maximum delay before a periodic WakeUp is scheduled, in milliseconds",
+          "type": "integer",
+          "minimum": 1000,
+          "default": 60000
+        }
+      }
+    },
     "session": {
       "description": "Settings for flow sessions",
       "type": "object",

gradle.properties

@@ -7,7 +7,7 @@ cordaProductVersion = 5.0.0
 # NOTE: update this each time this module contains a breaking change
 ## NOTE: currently this is a top level revision, so all API versions will line up, but this could be moved to
 ##   a per module property in which case module versions can change independently.
-cordaApiRevision = 96
+cordaApiRevision = 97
 
 # Main
 kotlinVersion = 1.6.21