Module of the Week - Cluster, Part 5

In the previous installment of Module of the Week, we finally got Dad to stop telling the same joke twice, but we made a mess of his code. Most of it is now about surviving crashes instead of telling jokes. We need to write down the result of each step he takes, then read it all back after a crash to find out where he got to.

It works, but I’m not proud of it. And I never want to write anything like it again.

This week we’re bringing in Effect Workflow, and most of that code is going in the bin.

By the end of this post, you’ll know:

1. What a workflow is, and how it remembers what it’s done
2. How to split work into activities
3. Why a workflow doesn’t need to be running to be in progress

Note to Self

Remember the code we hid at the end of Part 4? If you were brave enough to expand it, you’ll know there wasn’t much joke-telling in there. It was more than eighty lines of bookkeeping, all to answer one question: after a crash, what did I already do?

What we really want is for something else to do that bookkeeping for us.

Recipes

Let’s forget about computers for a minute and make pancakes.

You’ve got a recipe with five steps. Halfway through, the doorbell rings. You answer it, sign for a package, and come back to the kitchen. Where do you start?

If you’ve been ticking off each step as you go, it’s easy. Go back to the top of the recipe and work your way down, skipping anything with a tick. The first step without a tick mark is where you pick back up.

Click Step to work through the recipe, and keep an ear out for the doorbell.

Without the ticks, you might still figure out where you left off. But now you’re inspecting the batter and trying to remember how many eggs you cracked. Writing it down as you go makes coming back much easier.

We’ve just described a durable workflow: a program that saves the result of each step so it can recover from an interruption without repeating work.

Dad’s Recipe

Dad’s recipe has just two steps. When a batch of messages arrives, he:

1. Comes up with a joke he hasn’t told before
2. Posts it to the chat

Let’s turn that into a durable workflow. We’ll start by giving it a name and defining the data it receives.

import { Schema } from "effect"
import { Workflow } from "effect/workflow"

const Message = Schema.Struct({
  messageId: Schema.String,
  from: Schema.String,
  text: Schema.String,
})

// Dad couldn't come up with a joke
export class JokeError extends Schema.TaggedError<JokeError>()("JokeError", {
  cause: Schema.Defect(),
}) {}

export const TellJoke = Workflow.make("TellJoke", {
  payload: {
    // Which chat, and the batch of messages the joke is about
    chat: Schema.String,
    messages: Schema.Array(Message),
  },
  error: JokeError,
  // Same batch, same execution
  idempotencyKey: ({ chat, messages }) => `${chat}/${messages[0]!.messageId}`,
})

Like Entity.make, Workflow.make takes a name and schemas describing the data it needs to save. Our workflow receives a chat ID and a batch of messages. If it fails, the JokeError schema describes the error to save.

The idempotencyKey tells Workflow when two calls should count as the same run. Workflow combines this key with the workflow’s name to produce an execution ID. Call TellJoke again with the same key and it refers to the execution we already started. It’s the elevator button from Part 4 again. Pressing it twice still calls one elevator.

We could, but Dad might crash after saving the workflow and before acknowledging the messages. Those messages would be delivered again, possibly with a few new ones joining the batch. If we included every ID in the key, those additions would cause Dad to start another workflow.

Using the first message’s ID gives Dad the same key either way, so he gets back to the workflow he already started. That workflow keeps its original batch, though, so Dad will acknowledge any new messages without including them in the joke.

Activities

An activity is a named effect whose result gets saved to the workflow’s journal. When the workflow runs again, it can reuse that result instead of running the effect again. We’ll use one activity for each step in Dad’s recipe.

import { Effect, Schema } from "effect"
import { ClusterSchema } from "effect/cluster"
import { SqlClient } from "effect/sql"
import { Activity } from "effect/workflow"
import { JokeError, TellJoke } from "./workflow"

export const TellJokeLayer = TellJoke.toLayer(
  Effect.fn(function* ({ chat, messages }) {
    const punGenerator = yield* PunGenerator
    const slack = yield* Chat
    const sql = yield* SqlClient.SqlClient

    // Step 1: Write a joke
    const joke = yield* Activity.make({
      // The name of the activity
      name: "WriteJoke",
      // The journal saves the joke...
      success: Schema.String,
      // ...or why there isn't one
      error: JokeError,
      // The logic the activity runs
      execute: Effect.gen(function* () {
        // Every joke Dad has already told in this chat
        const told = yield* sql<{ joke: string }>`
          SELECT joke FROM dad_jokes WHERE chat = ${chat}
        `

        // A fresh joke based on the input messages
        const joke = yield* punGenerator.generate(messages, {
          told: new Set(told.map((row) => row.joke)),
        })

        // Write it down, so he never tells it in this chat again
        yield* sql`INSERT INTO dad_jokes ${sql.insert({ chat, joke })}`

        // The activity result which is persisted to the journal
        return joke
      }).pipe(
        // Anything that goes wrong becomes a JokeError
        Effect.mapError((cause) => new JokeError({ cause })),
      ),
    }).annotate(ClusterSchema.WithTransaction, true)

    // Step 2: Tell the joke
    yield* Activity.make({
      name: "PostJoke",
      execute: slack.reply(chat, joke),
    })
  }),
)

Dad now writes a joke and posts it in two activities, one after the other. The bookkeeping we hid in that collapsed code block last week is mostly taken care of for us.

In the WriteJoke activity, Dad reads the jokes he’s already told, asks the generator for a new one, and adds it to dad_jokes. The chosen joke is then saved as the activity’s result, and once saved, the workflow can reuse it without re-running the logic associated with the activity.

But what if Dad crashes after inserting the joke into dad_jokes, before the activity’s result is saved? On the next attempt, he’d treat that joke as one he’d already told and choose another, even though he never posted it.

That’s why we add the WithTransaction annotation. It saves the joke in dad_jokes and the activity’s result in the journal as one transaction. If Dad crashes before the transaction commits, neither write takes effect, so he can safely try again. If it commits, both records are saved and the workflow can reuse the joke. As in Part 4, this requires the dad_jokes table to be in the same database that is used by the cluster.

Replaying

What happens if the workflow is interrupted before it finishes?

When it starts again, Workflow replays the function, running it from the beginning and reusing saved activity results as it goes. It’s the same approach we took with the pancake recipe. Start at the top, check what’s already done, and carry on with what remains.

Click Step to follow the workflow through a crash and restart. Watch which lines run again and which results come from the journal.

Dad crashed after WriteJoke saved its result, before PostJoke had a chance to run. On replay, WriteJoke returned the saved joke without running its logic again. PostJoke had no result in the journal yet, so it posted that joke to the chat and saved its own result.

However, reading the clock wasn’t wrapped in an activity, so the workflow read it again and got a new time. If we’d left the pun generator outside WriteJoke, it would have run again too, potentially giving Dad a different joke.

Starting the Workflow

We have a workflow that can write and post a joke. Now Dad’s entity needs to pass it some messages. He still collects them until the chat has been quiet for 2.5 seconds, then calls TellJoke.execute with the batch:

export const DadLayer = Dad.toLayerQueue(
  Effect.gen(function* () {
    const address = yield* Entity.CurrentAddress

    return (mailbox, replier) => {
      // Wait for the chat to go quiet for 2.5 seconds, same as Part 3
      const nextBatch = Effect.gen(function* () {
8 collapsed lines
        const batch = [yield* Queue.take(mailbox)]
        while (true) {
          const next = yield* Queue.take(mailbox).pipe(
            Effect.timeoutOption("2500 millis"),
          )
          if (Option.isNone(next)) return batch
          batch.push(next.value)
        }
      })

      return Effect.gen(function* () {
        const batch = yield* nextBatch

        // Hand the joke to a workflow. With `discard: true`, this returns as
        // soon as the workflow is saved, not when the joke is told.
        yield* TellJoke.execute(
          {
            chat: address.entityId,
            messages: batch.map((request) => request.payload),
          },
          { discard: true },
        )

        // The workflow's saved, so it's safe to mark the batch processed
        yield* Effect.forEach(batch, (request) =>
          replier.succeed(request, undefined),
        )
      }).pipe(Effect.forever)
    }
  }),
)

The discard: true option lets Dad acknowledge the messages as soon as the workflow is saved, without waiting for it to finish. Those acknowledgements tell Cluster that he’s handled the messages. The workflow takes care of writing and posting the joke from there.

That replaces the inbox and outbox tables, polling loop, and keepAlive calls from Part 4. Incoming messages are still persisted by Cluster, with Slack’s ts as their primary key to catch duplicate deliveries. We’ve also moved WithTransaction from the NewMessage handler to WriteJoke, where we now write to dad_jokes.

To run the workflow on our existing cluster, we provide ClusterWorkflowEngine.layer alongside the entity and workflow implementations:

const MainLayer = Layer.mergeAll(DadLayer, TellJokeLayer).pipe(
  Layer.provide(ServicesLayer),
  // Runs workflows on the cluster we already have
  Layer.provideMerge(ClusterWorkflowEngine.layer),
  Layer.provideMerge(ClusterLayer),
)

Adding Features

Now that Workflow handles the bookkeeping, let’s give Dad another feature: if nobody laughs within an hour, he explains the joke.

With last week’s code, we’d need to store when Dad should check for laughs, arrange for that check to run, and track whether he’d already posted an explanation. We’d spend much of our time figuring out how to recover each part after a crash.

Here, we can add the steps directly to the workflow:

import { Effect, Schema } from "effect"
import { ClusterSchema } from "effect/cluster"
import { SqlClient } from "effect/sql"
import { Activity, DurableClock } from "effect/workflow"
import { JokeError, TellJoke } from "./workflow"

export const TellJokeLayer = TellJoke.toLayer(
  Effect.fn(function* ({ chat, messages }) {
20 collapsed lines
    const punGenerator = yield* PunGenerator
    const slack = yield* Chat
    const sql = yield* SqlClient.SqlClient

    // Step 1: Write a joke
    const joke = yield* Activity.make({
      name: "WriteJoke",
      success: Schema.String,
      error: JokeError,
      execute: Effect.gen(function* () {
        const told = yield* sql<{ joke: string }>`
          SELECT joke FROM dad_jokes WHERE chat = ${chat}
        `
        const joke = yield* punGenerator.generate(messages, {
          told: new Set(told.map((row) => row.joke)),
        })
        yield* sql`INSERT INTO dad_jokes ${sql.insert({ chat, joke })}`
        return joke
      }).pipe(Effect.mapError((cause) => new JokeError({ cause }))),
    }).annotate(ClusterSchema.WithTransaction, true)

    // Step 2: Tell the joke
    const ts = yield* Activity.make({
      name: "PostJoke",
      success: Schema.String,
      execute: slack.reply(chat, joke),
    })

    // Step 3: give the room an hour
    yield* DurableClock.sleep({
      name: "WaitForLaughs",
      duration: "1 hour",
    })

    // Step 4: count the laughs
    const laughs = yield* Activity.make({
      name: "CountLaughs",
      success: Schema.Number,
      execute: slack.reactionCount(chat, ts),
    })

    // Step 5: if nobody laughed, they must not have got it
    if (laughs === 0) {
      yield* Activity.make({
        name: "ExplainJoke",
        execute: punGenerator
          .explain(joke)
          .pipe(
            Effect.flatMap((explanation) => slack.reply(chat, explanation)),
          ),
      })
    }
  }),
)

The new code follows what we want Dad to do: wait an hour, count the laughs, and explain the joke if there aren’t any. Last week, we’d have needed code to work out which of those steps had finished if the workflow was interrupted. Now we can rely on the workflow’s journal to track that for us.

When a workflow starts a durable sleep, it suspends. It schedules a wake-up message in the database, timed to arrive in an hour, and then it stops running. No fiber is waiting, no memory is held, and no runner is responsible for it.

An hour later, the wake-up message is delivered to whichever runner owns the workflow’s shard at that time. The workflow runs from the top, every step so far comes back from the journal, and it carries on from CountLaughs.

Click Step to run the workflow, and keep stepping.

While Dad was asleep, nothing was running. When the wake-up message arrived, the workflow replayed from the beginning. The journal supplied the earlier results, and Dad continued by counting the laughs.

Wrapping up

That’s it for this week! To recap what we’ve learned:

• Workflows recover from interruptions by replaying and reusing saved activity results.

• Keep randomness and interactions with the outside world inside activities.

• Durable sleeps suspend the workflow until a scheduled message wakes it up.

All of the code from this post is on GitHub, if you’d like to try to catch Dad laughing at his own jokes.

Until then, Happy Effecting!