Help: How do I trigger workflows using events

[oEscal]

As the title says, I don't understand how I can trigger a workflow or a set of tasks from a specific event. From the Synapse architecture, I noticed that the Correlator entity performs event processing into CloudEvents. However, how do I link a specific source of events with this Correlator?
I am very new to this project, and I have already tried to find this issue in the wiki and the repo's issues. However, I could not find what I was looking for.

[cdavernas]

Hi @oEscal! You are on the right track: (cloud)events need to be posted using the structured JSON format on a correlator's API, at /api/v1/events/pub. This will trigger any matching correlation that exist, which you can view on the dashboard's dedicated interface.

[oEscal]

Thank you for your prompt response, @cdavernas! Could you direct me to the documentation on the correlator's API?

[cdavernas]

@oEscal You're very welcome!
As for the correlator's API documentation, I just realized it's not in the wiki.

You can, however, consult the Swagger by launching a correlator and navigating to /api/doc.

[oEscal]

I am very sorry @cdavernas, but I continue to have multiple issues.

First, is Synapse supposed to automatically create a correlation when I define a schedule based on events in a workflow? I have the following workflow (just for me to get things started in Synapse):

document:
  dsl: 1.0.0-alpha5
  namespace: default
  name: new-workflow2
  version: 0.1.4
schedule:
  "on":
    one:
      with:
        type: com.petstore.order.placed.v1
do:
- function-a:
    call: http
    with:
      method: get
      endpoint: http://function-a.simple-conditional.svc.cluster.local?value=true
    output:
      schema:
        format: json
        document:
          type: object
          properties:
            result:
              type: bool
          required:
          - result
      as:
        result: ${ .result }
- decide:
    switch:
    - case-true:
        when: .result == true
        then: function-b
    - case-false:
        when: .result == false
        then: function-c
- function-b:
    do:
    - function:
        call: http
        with:
          method: get
          endpoint: http://function-b.simple-conditional.svc.cluster.local
    then: exit
- function-c:
    do:
    - function:
        call: http
        with:
          method: get
          endpoint: http://function-c.simple-conditional.svc.cluster.local

But no correlation is created (at least, I don't see any in the Web UI)

Second, when I post a new event with this type in the correlator API endpoint we discussed, nothing happens.

[cdavernas]

is Synapse supposed to automatically create a correlation when I define a schedule based on events in a workflow?

Yes, it should create it. That's one of the operator's responsibilities, so make sure one is running when creating your flow and check its logs for potential errors. And yes, it should show up in the UI, else somethings not working as expected.

[oEscal]

I tried to do a little more debugging. I restarted all Pods (the api, operator, and correlator), and created the workflow all over again. Then, I made the following request to the correlator multiple times in a row (all fields but type have the default value from the API documentation):

curl -X 'POST'   '<correlator_url>/api/v1/events/pub'   -H 'accept: */*'   -H 'Content-Type: application/json'   -d '{
  "id": "string",
  "specversion": "1.0",
  "time": "2025-05-05T18:12:37.725Z",
  "source": "string",
  "type": "com.petstore.order.placed.v1",
  "subject": "string",
  "datacontenttype": "application/json",
  "dataschema": "string",
  "data": "string",
  "data_base64": "string",
  "additionalProp1": "string",
  "additionalProp2": "string",
  "additionalProp3": "string"
}'

And everything run as expected.

However, when I removed all non-required fields (according to the API documentation), like so:

curl -X 'POST'   '<correlator_url>/api/v1/events/pub'   -H 'accept: */*'   -H 'Content-Type: application/json'  -d '{
  "id": "string",
  "source": "string",
  "type": "com.petstore.order.placed.v1",
}' 

Although I don't get any error in any of the three Synapse Pods, the workflow is never instantiated again from the event creation. Even when I go back to use the first POST request above, it does not work. Moreover, some features seem to break: when I remove the workflow, its correlation is not removed and, when I create a new one, Synapse does not create the related correlation.
To get everything working back again, I need to restart the operator. However, it never throws any error.

@cdavernas is this a known behavior?

Also, although it does not seem related to this issue, any time I make the POST request that works (the first one above), the `api` Pod gives the following error (although everything works in that case):

fail: Microsoft.AspNetCore.Diagnostics.ExceptionHandlerMiddleware[1]                                                                                                                                                                        
      An unhandled exception has occurred while executing the request.                                                                                                                                                                      
      System.InvalidOperationException: StatusCode cannot be set because the response has already started.                                                                                                                                  
         at Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Http.HttpProtocol.ThrowResponseAlreadyStartedException(String value)                                                                                                          
         at Microsoft.AspNetCore.Mvc.StatusCodeResult.ExecuteResult(ActionContext context)                                                                                                                                                  
         at Microsoft.AspNetCore.Mvc.ActionResult.ExecuteResultAsync(ActionContext context)                                                                                                                                                 
         at Microsoft.AspNetCore.Mvc.Infrastructure.ResourceInvoker.ResultNext[TFilter,TFilterAsync](State& next, Scope& scope, Object& state, Boolean& isCompleted)                                                                        
         at Microsoft.AspNetCore.Mvc.Infrastructure.ResourceInvoker.InvokeResultFilters()                                                                                                                                                   
      --- End of stack trace from previous location ---                                                                                                                                                                                     
         at Microsoft.AspNetCore.Mvc.Infrastructure.ResourceInvoker.<InvokeFilterPipelineAsync>g__Awaited|20_0(ResourceInvoker invoker, Task lastTask, State next, Scope scope, Object state, Boolean isCompleted)                          
         at Microsoft.AspNetCore.Mvc.Infrastructure.ResourceInvoker.<InvokeAsync>g__Awaited|17_0(ResourceInvoker invoker, Task task, IDisposable scope)                                                                                     
         at Microsoft.AspNetCore.Mvc.Infrastructure.ResourceInvoker.<InvokeAsync>g__Awaited|17_0(ResourceInvoker invoker, Task task, IDisposable scope)                                                                                     
         at Swashbuckle.AspNetCore.SwaggerUI.SwaggerUIMiddleware.Invoke(HttpContext httpContext)                                                                                                                                            
         at Swashbuckle.AspNetCore.Swagger.SwaggerMiddleware.Invoke(HttpContext httpContext, ISwaggerProvider swaggerProvider)                                                                                                              
         at Microsoft.AspNetCore.Diagnostics.ExceptionHandlerMiddlewareImpl.<Invoke>g__Awaited|10_0(ExceptionHandlerMiddlewareImpl middleware, HttpContext context, Task task)
warn: Microsoft.AspNetCore.Diagnostics.ExceptionHandlerMiddleware[2]                                                                                                                                                                        
      The response has already started, the error handler will not be executed.                                                                                                                                                             
fail: Microsoft.AspNetCore.Server.Kestrel[13]                                                                                                                                                                                               
      Connection id "0HNCC13I5FQE0", Request id "0HNCC13I5FQE0:00000003": An unhandled exception was thrown by the application.                                                                                                             
      System.InvalidOperationException: StatusCode cannot be set because the response has already started.                                                                                                                                  
         at Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Http.HttpProtocol.ThrowResponseAlreadyStartedException(String value)                                                                                                          
         at Microsoft.AspNetCore.Mvc.StatusCodeResult.ExecuteResult(ActionContext context)                                                                                                                                                  
         at Microsoft.AspNetCore.Mvc.ActionResult.ExecuteResultAsync(ActionContext context)                                                                                                                                                 
         at Microsoft.AspNetCore.Mvc.Infrastructure.ResourceInvoker.ResultNext[TFilter,TFilterAsync](State& next, Scope& scope, Object& state, Boolean& isCompleted)
         at Microsoft.AspNetCore.Mvc.Infrastructure.ResourceInvoker.InvokeResultFilters()                                                                                                                                                   
      --- End of stack trace from previous location ---                                                                                                                                                                                     
         at Microsoft.AspNetCore.Mvc.Infrastructure.ResourceInvoker.<InvokeFilterPipelineAsync>g__Awaited|20_0(ResourceInvoker invoker, Task lastTask, State next, Scope scope, Object state, Boolean isCompleted)                          
         at Microsoft.AspNetCore.Mvc.Infrastructure.ResourceInvoker.<InvokeAsync>g__Awaited|17_0(ResourceInvoker invoker, Task task, IDisposable scope)                                                                                     
         at Microsoft.AspNetCore.Mvc.Infrastructure.ResourceInvoker.<InvokeAsync>g__Awaited|17_0(ResourceInvoker invoker, Task task, IDisposable scope)                                                                                     
         at Swashbuckle.AspNetCore.SwaggerUI.SwaggerUIMiddleware.Invoke(HttpContext httpContext)                                                                                                                                            
         at Swashbuckle.AspNetCore.Swagger.SwaggerMiddleware.Invoke(HttpContext httpContext, ISwaggerProvider swaggerProvider)                                                                                                              
         at Microsoft.AspNetCore.Diagnostics.ExceptionHandlerMiddlewareImpl.<Invoke>g__Awaited|10_0(ExceptionHandlerMiddlewareImpl middleware, HttpContext context, Task task)
         at Microsoft.AspNetCore.Diagnostics.ExceptionHandlerMiddlewareImpl.HandleException(HttpContext context, ExceptionDispatchInfo edi)                                                                                                 
         at Microsoft.AspNetCore.Diagnostics.ExceptionHandlerMiddlewareImpl.<Invoke>g__Awaited|10_0(ExceptionHandlerMiddlewareImpl middleware, HttpContext context, Task task)
         at Microsoft.AspNetCore.Authorization.AuthorizationMiddleware.Invoke(HttpContext context)                                                                                                                                          
         at Microsoft.AspNetCore.Authentication.AuthenticationMiddleware.Invoke(HttpContext context)                                                                                                                                        
         at IdentityServer4.Hosting.IdentityServerMiddleware.Invoke(HttpContext context, IEndpointRouter router, IUserSession session, IEventService events, IBackChannelLogoutService backChannelLogoutService)                            
         at IdentityServer4.Hosting.MutualTlsEndpointMiddleware.Invoke(HttpContext context, IAuthenticationSchemeProvider schemes)                                                                                                          
         at Microsoft.AspNetCore.Authentication.AuthenticationMiddleware.Invoke(HttpContext context)                                                                                                                                        
         at IdentityServer4.Hosting.BaseUrlMiddleware.Invoke(HttpContext context)                                                                                                                                                           
         at Program.<>c__DisplayClass0_0.<<<Main>$>b__5>d.MoveNext() in /src/src/api/Synapse.Api.Server/Program.cs:line 116                                                                                                                 
      --- End of stack trace from previous location ---                                                                                                                                                                                     
         at Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Http.HttpProtocol.ProcessRequests[TContext](IHttpApplication`1 application)  

[cdavernas]

However, when I removed all non-required fields (according to the API documentation)

I guess that's because there is no specversion/data/datacontentype.
Also, note that context attributes should be lowercased, as per the CloudEvent spec. Finally, make sure to use a different id for each CloudEvent.

when I remove the workflow, its correlation is not removed and, when I create a new one, Synapse does not create the related correlation.

That sounds like a bug, the operator should delete the correlation. Maybe it faced an issue before the workflow's deletion? I'd be very grateful if you could confirm that behavior and, if applicable, open a dedicated issue.

Also, although it does not seem related to this issue, any time I make the POST request that works (the first one above), the api Pod gives the following error (although everything works in that case)

Yes, I actually faced that bug in another project of mine, and it is due to the SSE implementation returning 200 OK after the last event has been sent. I'll fix that, but, as you said, this is a purely cosmetic bug, as it dies not break anything.

[oEscal]

I guess that's because there is no specversion/data/datacontentype.
Also, note that context attributes should be lowercased, as per the CloudEvent spec. Finally, make sure to use a different id for each CloudEvent.

I understand that. However, in the API documentation, it only implies that the id, source, and type are the required keys:

Also, shouldn't the correlator not accept the request if it is malformed? If the CloudEvent lacks some key or has any other issue, I think it should respond with an HTTP error status code and the details of the error.

That sounds like a bug, the operator should delete the correlation. Maybe it faced an issue before the workflow's deletion? I'd be very grateful if you could confirm that behavior and, if applicable, open a dedicated issue.

The problem is that any of the api, correlator, or operator gives an error log.