Feature request: Awaitable Workflow outcomes

[jwulf]

Is your feature request related to a problem? Please describe.
At the moment it is not possible to start a workflow and be notified of the outcome.

Describe the solution you'd like
In the client libs, some way to await the outcome of a workflow.

Describe alternatives you've considered
https://zeebe.io/blog/2019/08/zeebe-rest-affinity/

[berndruecker]

I can add some thoughts of a discussion a while back.

I see the following alternatives. I personally tend to favor approach number 2 – as I think it is the best compromise of decoupling of the Code and the Model but also really powerful.

1. Wait for a certain activity, via API

So you can say something like (naming tbd):

zeebe.workflowClient().newCreateInstance()….send().waitForActivity(“SomeActivityIdFromTheBPMN”, timeout);

Advantage: different clients could wait for different Activities – and you don’t have to change the model at all for this. We could even later on provide

zeebe.workflowClient().newCreateInstance()….send().waitForActivities([“SomeActivityIdFromTheBPMN”, “AndSomeOther”], timeout);

2. Wait for some data condition to be true, via API

zeebe.workflowClient().newCreateInstance()….send().waitForCondition(“resultAvailable==true”, timeout);

Advantage: You are completely decoupled from the BPMN workflow model and just wait e.g. for some data to be set.

3. Add some element in the BPMN model to trigger the response

zeebe.workflowClient().newCreateInstance()….send().waitForResponse(timeout);

Advantage: You can see it in the model graphically (but nobody stops you adding a noop element in your BPMN in alternative 1 or 2 if you need a visual clue).

4. Add some attribute to elements of the BPMN model to trigger the response

Advantage: You could configure the end event to trigger the response

[tmdoit-zz]

Use Zeebe workflows in REST request/response operations in the sync way

Use Case

1. Client is creating new user via sending POST /users with payload

{
    "data": {...}, 
    "meta" : {...,"request_id":"355ecbf0-b92e-11e9-874c-4520384532ec"}
}

or with HTTP header ie.

X-Request-Id: 355ecbf0-b92e-11e9-874c-4520384532ec

request_id - in Zeebe context known as workflowInstanceCorrelationId which reference to workflowInstanceKey, more below
1.A Client in request sends only header "X-Request-Id" or payload with only "meta" object defined (without data object)
1.A.1 REST server is "subscribing" or periodically requesting service GetWorkflowInstanceStatus waiting for workflow instance to finish (and then return response to the client)
2. REST server is starting instance workflow "CreateUser" using below payload (pseudo code) for CreateWorkflowInstanceRequest

{
   "bpmnProcessId": "CreateUser",
   "version": -1,
   "variables": $string(
       {
           "request_id" : <uuid>,
           "request_payload": <data>
       }
   )
}

3. REST server is "subscribing" or periodically requesting service GetWorkflowInstanceStatus (two options presented below) to be informed when instance finished work, payload pseudo code below

{
    "workflowInstanceCorrelationId" : "<uuid (request_id sent by client)>",
    "flowNodeType": "EndEvent"
}

4. Zeebe workers doing their job
5. Workflow instance hit "EndEvent"
6. REST server gets information about finished workflow instance, payload pseudo code for msg GetWorkflowInstanceStatusResponse:

{
    "statusList": [{
        "flowNodeName": "User not Created",
        "startDate": "2011-10-05T14:48:00.000Z",
        "flowNodeType": "endEvent",
        "endDate": "2012-10-05T14:48:00.000Z",
        "variables": {
            "error": "Invalid request schema"
        }
    }]
}

API

Update msg CreateWorkflowInstanceRequest (add field workflowInstanceCorrelationId)

message CreateWorkflowInstanceRequest {
  // the unique key identifying the workflow definition (e.g. returned from a workflow
  // in the DeployWorkflowResponse message)
  int64 workflowKey = 1;
  // the BPMN process ID of the workflow definition
  string bpmnProcessId = 2;
  // the version of the process; set to -1 to use the latest version
  int32 version = 3;
  // JSON document that will instantiate the variables for the root variable scope of the
  // workflow instance; it must be a JSON object, as variables will be mapped in a
  // key-value fashion. e.g. { "a": 1, "b": 2 } will create two variables, named "a" and
  // "b" respectively, with their associated values. [{ "a": 1, "b": 2 }] would not be a
  // valid argument, as the root of the JSON document is an array and not an object.
  string variables = 4;
  // Reference to the created workflowInstanceKey, you could use it in GetWorkflowInstanceStatus service
  string workflowInstanceCorrelationId = 5;
}

Add new service: GetWorkflowInstanceStatus

/*
    Get info about workflow instance

    Errors:
        WORKFLOW_INSTANCE_NOT_FOUND:
        FLOW_NODE_NAME_NOT_FOUND:
        FLOW_NODE_TYPE_NOT_FOUND:
*/ 
rpc GetWorkflowInstanceStatus (GetWorkflowInstanceStatusRequest) returns (GetWorkflowInstanceStatusResponse) {
}

message GetWorkflowInstanceStatusRequest {
    // the unique identifier of the created workflow instance; to be used wherever a request
    // needs a workflow instance key (e.g. CancelWorkflowInstanceRequest)
    int64 workflowInstanceKey = 1;
    // Value passed in CreateWorkflowInstanceRequest - references to workflowInstanceKey
    string workflowInstanceCorrelationId = 2;
    // Filter by node name ie. "IsRequestValid?"
    string flowNodeName = 3
    // Filter by node type ie. "EndEvent"
    string flowNodeType = 4
    // Filter by condition
    string condition = 5
}

message GetWorkflowInstanceStatusResponse {
    repeated Status statusList = 1;
}

message Status {
    string flowNodeName = 1
    string flowNodeType = 2
    string startDate = 3
    string endDate = 4
    string variables = 5
    string conditionResult = 6
}

OR

Add new service: GetWorkflowInstanceStatus (Stream Response)

/*
    Get info about running workflow instance

    Errors:
        WORKFLOW_INSTANCE_NOT_FOUND:
        FLOW_NODE_NAME_NOT_FOUND:
        FLOW_NODE_TYPE_NOT_FOUND:
*/ 
rpc GetWorkflowInstanceStatus (GetWorkflowInstanceStatusRequest) returns (GetWorkflowInstanceStatusResponse stream) {
}


message GetWorkflowInstanceStatusRequest {
    // the unique identifier of the created workflow instance; to be used wherever a request
    // needs a workflow instance key (e.g. CancelWorkflowInstanceRequest)
    int64 workflowInstanceKey = 1;
    // Value passed in CreateWorkflowInstanceRequest - references to workflowInstanceKey    
    string workflowInstanceCorrelationKey = 2; // Property passed in CreateWorkflowInstanceRequest - references to workflowInstanceKey
     // Filter by node name ie. "IsRequestValid?"   
    string flowNodeName = 3 // ie. "IsRequestValid?"
    // Filter by node type ie. "EndEvent"    
    string flowNodeType = 4 // ie. "EndEvent"
    // After defined timeout server will close stream 
    int64 timeout = 5 
    // Filter by condition
    string condition = 6
}

message GetWorkflowInstanceStatusResponse {
    repeated Status statusList = 1;
}

message Status {
    string flowNodeName = 1
    string flowNodeType = 2
    string startDate = 3
    string endDate = 4
    string variables = 5
    string conditionResult = 6
}

Solution using MQTT

This could be done using MQTT server (correlate msgs), without access to history:

1. Client is creating new user via sending POST /users with payload

{
    "data": {...}, 
    "meta" : {...,"request_id":"355ecbf0-b92e-11e9-874c-4520384532ec"}
}

or with HTTP header ie.

X-Request-Id: 355ecbf0-b92e-11e9-874c-4520384532ec

2. REST server is subscribing to the topic from request_id ie. "355ecbf0-b92e-11e9-874c-4520384532ec"
3. Zeebe workers doing their job
4. Delegated workflow node is responsible to send response to the correlated MQTT topic
5. REST server is receiving msg from MQTT topic and sends response to the client

Comments

Naming convention could be misleading, I'm new to Zeebe, gRPC and BPMN.

According to @berndruecker proposals, the last one (4) seems to me to be the simplest solution to make REST request/response keep sync.
If I understood good, Zeebe client will be notified when workflow instance hit "endEvent" and what is important - access to the passed variables?

Modified 08.03

[jwulf]

We would probably need both, ideally.

We cannot keep completed workflow state indefinitely in the broker, otherwise disk / memory usage and performance will be impacted. So we need to emit completion when it happens as a one-time event.

And a workflow may raise an incident - in which case no completion is emitted - and you may want to let the requestor know; or update them on "30% complete" or similar as it progresses. In this case a query API like GetWorkflowInstanceStatusRequest would be useful.

However, such an API introduces a tight-coupling if it requires knowledge of the internal structure of the workflow. All it takes is for an updated workflow to be deployed with a renamed flow node, and your application code breaks.

I think the lowest-orbit solution would be the ability to subscribe to the end event and get the final variable state.

Then some emitter node would reduce / make explicit the coupling between the workflow and application code.

[tmdoit-zz]

We cannot keep completed workflow state indefinitely in the broker, otherwise disk / memory usage and performance will be impacted. So we need to emit completion when it happens as a one-time event.

It's for current state purpose not for history, the solution could be to add "ttl" (time to live after workflow instance finished) field to the CreateWorkflowInstanceRequest msg.

[tmdoit-zz]

Now I'm thinking about new service like GetWorkflowCurrentStatus - it's something I think what @berndruecker propose in the first and second point

/*
    Get info about running workflow instance

    Errors:
        WORKFLOW_INSTANCE_NOT_FOUND:
        FLOW_NODE_NAME_NOT_FOUND:
        FLOW_NODE_TYPE_NOT_FOUND:
*/ 
rpc GetWorkflowCurrentStatus (GetWorkflowCurrentStatusRequest) returns (GetWorkflowCurrentStatusResponse stream) {
}


message GetWorkflowCurrentStatusRequest {
    // the unique identifier of the created workflow instance; to be used wherever a request
    // needs a workflow instance key (e.g. CancelWorkflowInstanceRequest)
    int64 workflowInstanceKey = 1;
    // Value passed in CreateWorkflowInstanceRequest - references to workflowInstanceKey    
    string workflowInstanceCorrelationKey = 2; // Property passed in CreateWorkflowInstanceRequest - references to workflowInstanceKey
     // Filter by node name ie. "IsRequestValid?"   
    string flowNodeName = 3 // ie. "IsRequestValid?"
    // Filter by node type ie. "EndEvent"    
    string flowNodeType = 4 // ie. "EndEvent"
    // Filter by condition
    string condition = 5
    // After defined timeout server will close stream 
    int64 timeout = 6
}

message GetWorkflowCurrentStatusResponse {
    string flowNodeName = 1
    string flowNodeType = 2
    string startDate = 3
    string endDate = 4
    string bpmnError = 5
    string variables = 6
    string conditionResult = 7
}

The problem with this service is that the client will not get status of workflow instance when disconnected ie. for timeout reason or something.
Server will close stream after when instance hit end event or hit a bpmn error.

[ChrisKujawa]

Do you guys had a look at the Hazelcast exporter? https://github.com/zeebe-io/zeebe-hazelcast-exporter You have to register the exporter in the broker, embed the connector in your application and then get notified for specific events.

[berndruecker]

Just to make sure: The feature request scope is to make this possible without any additional 3rd party software you need to operate. I see Hazelcast as a possible workaround for the moment though.

[tmdoit-zz]

Can you set ETA for this feature?

[danshapir]

Would love to see this feature happening!

But I'd state that it's very reasonable to not want to know the workflow had ended, but rather reached a specific step. This way part of the flow is "online" and part of offline.

[Bizoner]

Waiting for this to happen.

[dk1100]

Can we have this feature? I could be really awesome!!

[oromano]

I will be glad to get this feature!

[StephenOTT]

Any thoughts on #3048

I think it's basically a duplicate of #2896 (comment).

I think it would be great if you could have a client subscribe to a message. It opens up a whole other way of usage and allows easy response that is clear and concise in the Bpmn

[mike7ang1rdz]

this feature for c#, would be nice

[Tails]

I was surprised that this feature doesn't exist. I have a use case where a user uploads a file through an OpenFaas serverless handler. This schedules a workflow of parsing the file which may take a while. While the user is waiting in the frontend, I would like to update the view with the status of the workflow (progressbar or steps). I expected this to be possible by having the serverless handle open a websocket connection with the client, passing through all events it would receive from the workflow.

Is this possible?

Edit: found my use case in the blog post: https://github.com/jwulf/zeebe-affinity-service

[Tails]

See also: #2916 (comment)

[tmdoit-zz]

@Tails as a workaround you can generate UUID for every request and attach it to HTTP header ie. "requestId" and pass it through tasks and handlers. Then you can use "requestId" as MQTT topic name and subscribe to it to get status information which you will send from OpenFAAS functions (publish to the defined topic name). You can use MQTT over web sockets if you wish.

[Tails]

@Tails as a workaround you can generate UUID for every request and attach it to HTTP header ie. "requestId" and pass it through tasks and handlers. Then you can use "requestId" as MQTT topic name and subscribe to it to get status information which you will send from OpenFAAS functions (publish to the defined topic name). You can use MQTT over web sockets if you wish.

Based on the inspiration of the blog post and the example repository this was indeed the idea I got. The OpenFaas handler returns a topic ID for the JS client to listen to. After the workflow invocation using the short-lived serverless handle the JS client opens a connection with a websocket server that, as a worker "spy", listens to workflow progress events and publishes these over the websocket to the frontend view.

Thank you!

[StephenOTT]

Can we get some more details on what that PR does? What is a completed workflow in this context? End event? Wait state? (Like wait for a message). Will this be added for the correlate a message "with result"?

[deepthidevaki]

Can we get some more details on what that PR does? What is a completed workflow in this context? End event? Wait state? (Like wait for a message). Will this be added for the correlate a message "with result"?

@StephenOTT A workflow instance is completed when there is nothing more to be processed related to the workflow. That means the end events are completed. To be more precise, the "process" element has reached ELEMENT_COMPLETED (reference)

[StephenOTT]

Okay. Can a client reconnect for a waiting of a process to complete or if the client loses its connection, it can never retrieve the result? This feature was designed for short lived processes?

[danshapir]

Amazing work guys! Any idea when will this be available for the nodejs library?

[Tails]

Very cool, such a quick response!

Workflows are often invoked from short-lived (REST/serverless) handlers. Can the workflow result be retrieved asynchronously; from another handler? And can the result be queried indefinitely or only in a certain window of time? (until historic event log is purged?)

[deepthidevaki]

At the moment it is not possible to retrieve the results asynchronously. Reason is that all state related to the workflow is deleted when it is completed. Currently this feature is addressed towards short lived processes. TBD if we implement asynchronous query.

[StephenOTT]

If you could implement the ability for a client to await a BPMN message then you would not need to, you can have a client await for that message.

[jwulf]

Amazing work guys! Any idea when will this be available for the nodejs library?

As soon as it drops in an alpha build.

[jwulf]

Okay. Can a client reconnect for a waiting of a process to complete or if the client loses its connection, it can never retrieve the result? This feature was designed for short lived processes?

It depends what you mean by short-lived, but yeah, you are not going to wait for days, because if you lose the gRPC connection you lose everything. You don't even have a workflow instance id to find the outcome in an exporter.