# Composable flows

**Knative Eventing Working Group**

**Last Updated:** Jan 27th, 2022

**Main GitHub Issue**: [1521](https://github.com/knative/eventing/issues/1521)

**Contributor(s)**:
- Lionel Villard ([e-mail])

## Motivation / Abstract

Knative Eventing flows constructs (Sequence and Parallel) are not composable, considerably limiting their usefulness. For instance, a Parallel branch cannot reference an existing Sequence and forward the result to the next step to, let's say, aggregate the results. This limitation cannot be easily solved at the application layer (for instance by calling a subflow) mostly due to the dependency on inner workings, such as reference resolution, to forward the result to an object somewhere in the middle of a flow. The only reasonable solution is for Knative Eventing to support this pattern out-of-the-box because 1. it is a very common pattern 2. it not something that can easily be done at the application level.

## Background

Knative Eventing provides two flows constructs, [Sequence](https://knative.dev/docs/eventing/flows/sequence/) and [Parallel](https://knative.dev/docs/eventing/flows/parallel/). Both constructs allow references to external sinks with the implicit assumption that those sinks are [callable](https://knative.dev/docs/eventing/sinks/) (i.e. return 0 or 1 event) in order for the flow to *keep going* when an event is received and to be interrupted when no event is received.

For instance, consider this sequence:

apiVersion: flows.knative.dev/v1 
kind: Sequence
metadata:
  name: oneseq
spec:
  steps:
  - ref:
      apiVersion: serving.knative.dev/v1 
      kind: Service
      name: identity
  - ref:
      apiVersion: serving.knative.dev/v1
      kind: Service
      name: last-step

The second step is only executed when the service in the first step synchronously returns a non-empty event. When the first step is a reference to a flow construct, never immediately returning an event, the second step is consequently never executed.

The proposal below leverages existing Knative Eventing capabilities, specifically [error handling](https://knative.dev/docs/eventing/event-delivery/) and delivery guarantees.

Under the cover, both Sequence and Parallel get realized as a set of Channel and Subscription objects by their respective reconciler.

## Proposal Design / Approach

Current flow constructs are not Callable, but they *eventually* produce zero or one event via [`reply`](https://knative.dev/docs/reference/api/eventing-api/#flows.knative.dev/v1.SequenceSpec). We propose to call this type of objects **composable**:

- Composable objects are Addressable, i.e. an event can be delivered over HTTP to an address defined in their `status.address.url` field.
- Composable objects can asynchronously forward events over HTTP to an address specified in their `spec.reply` field.

In addition, sink objects can potentially invoked in two different ways:
- async mode: The current event is forwarded to both the sink and the next step. Any event returned by the sink ingress is dropped.
- sync mode: The event(s) produced by the sink (synchronously or asynchronously) are forwarded to the next step.

Asynchronous invocation of both Callable and Composable objects is straightforward. The real challenge is synchronous invocation of composable objects, as discussed below.

A simple approach is to clone the referenced composable object and to inject the address of the next step into the `spec.reply` field. While this solution could work (modulo some open questions like what happens when `spec.reply` is already specified?), it leads to a lot of additional objects being created, all connected the same way (same topology) except for the last `spec.reply`.

An alternative solution is to dynamic dispatch flow construct results to the next step. This can be achieved by maintaining a stack of callers (e.g. by using a CloudEvent extension attribute) and by adding stack manipulation functions, one for the flow entry (push) and one for the flow exit (pop). This is how most compilers handle [function calls](https://en.wikipedia.org/wiki/Subroutine#Call_stack).

TODO: explain how errors are handled, maybe from a continuation passing style PoV.

## Implementation

### Composable duck-type

The Composable duck-type is defined as follows:

type Composable struct {
	metav1.TypeMeta   `json:",inline"`
	metav1.ObjectMeta `json:"metadata,omitempty"`

	Spec ComposableSpec `json:"spec,omitempty"`

	Status ComposableStatus `json:"status,omitempty"`
}

// ComposableSpec contains Spec of the Composable object
type ComposableSpec struct {
	
	// Reply is a Reference to where the Composable result 
	// is sent to
	// +optional
	Reply *duckv1.Destination `json:"reply,omitempty"`
}

// ComposableStatus contains the Status of a Composable object.
type ComposableStatus struct {
	// AddressStatus is the part where the Composable fulfills the Addressable contract.
	// +optional
	duckv1.AddressStatus `json:",inline"`
}

### Asynchronous Composable Invocation

#### Call Stack

The call stack is represented by a CloudEvent attribute. Let's call it `knativeflowcallstack`. Its value is a space-separated list of resolved URLs.

##### Maximum size

Knative Eventing should enforce the limit on the maximum number of nested calls. A global configuration parameter will be added.

##### Security and integrity

In this proposal the flow state is stored within the CloudEvent `knativeflowcallstack` extension attribute, which is visible to applications, possibly leading to meta-data corruption. A solution to this problem is to remove and reinject this attribute before and after a non-composable function is invoked.

#### Data-Plane

This solution relies on two managed stateless functions: `flow-push` and `flow-pop`. The first one manipulates the received `knativeflowcallstack` by adding a new URL, passed as an HTTP query parameter, to the end of the list and directly forwards the modified event to the specified reference.

The second service removes the last URL from the received CloudEvent and directly forwards the modified event to the URL.

Side comment: these services could be removed if Knative Eventing Subscription would support `ceOverrides` over list, or if it would support capabilities negociation.


#### Control-Plane

When the referenced composable is asynchronously invoked, the flow reconciler configures the underlying Subscription to point to the built-in `flow-push` function. Here an example of a generated Subscription:

apiVersion: messaging.knative.dev/v1
kind: Subscription
metadata:
  name: <subscription-name> # Name of the Subscription.
spec:
  subscriber:
    uri: https://flow-push.knative-eventing?URL=<resolved URI>

On the egress, the flow operator checks the `spec.reply` field is properly set to send event to the `flow-pop` built-in function. If not, the reconciliation process fails.

## Integration Checklist

### Operations

TBD

### Observability

TBD

### Test Plan

TBD

### Documentation

TBD

### User Experience
TBD

## Alternative Proposals

### Delivery Contract Extension

As suggested [in this comment](https://docs.google.com/document/d/1j0DsiyTtET6QNyz3wcUrXhapJk602snurZBQm8Ru07k/edit?disco=AAAAHe6n54M), extend the delivery contract to include an optional `Reply-Location` header that indicates a URL where reply events may be POSTed (instead of, or in addition to, being processed from the HTTP Response).

`Reply-Location` would _not_ be a CloudEvents attribute; it would be a hop-by-hop HTTP option on the delivery, similar to the `Prefer: Reply` header (and possibly set in similar circumstances).

_Pros:_
* Possibly simpler implementation vs a call stack
* Also applies to long-lived delivery to a Trigger or Subscription Target
* Potentially provides an avenue for returning multiple response events to a single event.

_Cons_:
* Recipients using `Reply-Location` would need to ensure that the event was persisted to stable storage before returning a 200.
* `Reply-Location` information might need to be persisted across multiple Channel hops in a Sequence or Parallel, which is not supported in the current implementation.
* This does not avoid the conflict between having `Reply-Location` set on a request and `spec.reply` in Sequence or Parallel. (It seems to me that Broker and Channel would reasonably ignore this parameter.)