Initial Streaming Implementation

[ptpaterson]

Ticket(s): FE-4969

This is an initial go at implementing streaming events for the v10 API. Anything here is up for review and comment.

Problem

Implement the v10 Fauna streaming API

We need to support browsers and Node. We would also like to support other runtimes if possible.

Solution

Provide a StreamClient class which provides the primary user interface to establish a stream and handle events.

Provide a lower-level HTTPStreamClient interface that can be implemented for different environments.

The StreamClient contains an instance of an HTTPStreamClient, which mirrors the existing lower-level HTTPClient interface that enables different backend implementations for the Client class.

The HTTPStreamClient is responsible for initiating a request and returning an object that implements the StreamAdapter interface. The StreamAdapter contains an asyncronous generator function that yields Fauna streaming events as raw strings for the StreamClient to use. The StreamAdapter also provides a close method to provide a mechanism to close the stream.

The StreamClient is responsible for parsing the raw-string events into valid types. Streaming events always return data in the "tagged" format. The StreamClient must be configured with how to handle decoding of Long types.

An instance of StreamClient is created on each call to Client.stream. Each instance tracks it's own handlers for events. The underlying HTTP connections should be reused when possible.

Result

Streams can be used in the browser and Node.js.

This does not work in public Fauna environments or the docker image yet.

Usage

Two ways to use the stream are enabled:

1. Async Iterator
2. Callbacks

Async Iterator example

import { Client, fql } from "fauna"
const client = new Client({ secret: FAUNA_SECRET })

const response = await client.query(fql`MyCollection.all().toStream()`);
const streamToken = response.data;

const stream = client.stream(streamToken)

try {
  for await (const event of stream) {
    switch (event.type) {
      case "update":
      case "add":
      case "remove":
        console.log("Stream update:", event);
        // ...
        break;
    }
  }
} catch (error) {
  // An error will be handled here if Fauna returns a terminal, "error" event, or
  // if Fauna returns a non-200 response when trying to connect, or
  // if the max number of retries on network errors is reached.

  // ... handle fatal error
}

Callbacks example

import { Client, fql } from "fauna"
const client = new Client({ secret: FAUNA_SECRET })

const response = await client.query(fql`MyCollection.all().toStream()`);
const streamToken = response.data;

const stream = client.stream(streamToken)
  
stream.start(
  function onEvent(event) {
    switch (event.type) {
      case "update":
      case "add":
      case "remove":
        console.log("Stream update:", event);
        // ...
        break;
    }
  },
  function onFatalError(error) {
    // An error will be handled here if Fauna returns a terminal, "error" event, or
    // if Fauna returns a non-200 response when trying to connect, or
    // if the max number of retries on network errors is reached.

    // ... handle fatal error
  }
);

The driver will take care of the initial request to convert to a stream if you provide a Query

import { Client, fql } from "fauna"
const client = new Client({ secret: FAUNA_SECRET })

const stream = await client.stream(fql`MyCollection.all().toStream()`)

for await (const event of stream) {
  // ...
}

Error handling

The driver tries to reestablish streaming connections when there are network errors. With current defaults, retries start quickly but have exponential backoff, capped at 10s, and will continue trying to reestablish for about 10 minutes. That's generous enough to let your internet connection drop and come back.

Network errors are hidden from the user while retries are still being attempted

Errors from Fauna are assumed fatal. Any error events coming from Fauna will cause an error to be thrown.

Out of scope

N/A

Testing

tests have been added and can be run on the most recent docker image using yarn test.

---

By submitting this pull request, I confirm that you can use, modify, copy, and redistribute this contribution, under the terms of your choice.

[erickpintor]

This link returns a 404 to me.

[ptpaterson]

Ah. Looks like all of the links should be updated!

[fauna-chase]

one thing I wonder, and we could always wait and see if people want this, but would it be worth in addition to our two step
1 - create stream token
2 - call stream

to also add an API in the drivers that allowed
client.stream(fql'query here that produces a stream').on( event => ..)
purely as a form of convenience. Under the hood we'd still send the same queries.

[erickpintor]

one thing I wonder, and we could always wait and see if people want this, but would it be worth in addition to our two step 1 - create stream token 2 - call stream

to also add an API in the drivers that allowed client.stream(fql'query here that produces a stream').on( event => ..) purely as a form of convenience. Under the hood we'd still send the same queries.

I think it's a good idea.

[ptpaterson]

one thing I wonder, and we could always wait and see if people want this, but would it be worth in addition to our two step 1 - create stream token 2 - call stream
to also add an API in the drivers that allowed client.stream(fql'query here that produces a stream').on( event => ..) purely as a form of convenience. Under the hood we'd still send the same queries.

I think it's a good idea.

I've implemented this. It required Client.stream to be async, but I think that is okay. It will only make the initial request for the stream token. The stream connection still waits for you to call stream.start or start iterating. Client.stream is still a sync function.

[ptpaterson]

I've updated error handling, so that retry-able network errors create events which can be handled. In the event of a retry, the user should expect at least two events back-to-back:

1. One or more "error" events with info about the network error(s), and
2. assuming a new connection is made, another "start" event

[erickpintor]

I've updated error handling, so that retry-able network errors create events which can be handled. In the event of a retry, the user should expect at least two events back-to-back:

1. One or more "error" events with info about the network error(s), and

2. assuming a new connection is made, another "start" event

Does that mean that drivers are syntactically producing error events that mimic events from the server? If so, are driver errors distinguishable from stream error events? I'm concerned that if we are co-opting the event format we can end up in a situation where a new error event in the server shadows a driver specific event. Ideally we would want to keep these dimensions separate so we don't have to check every driver before introducing new error codes in the server.

[ptpaterson]

@pnwpedro I've rebased on main to include the concourse pipeline changes, and this PR is now set to merge into the beta branch.