(Known bug with a release that was fixed)

R

Ray•12/10/24, 12:33 PM

https://github.com/opraying/worker-workflow-test case someone is interested in the integration of Effects and Workflows, here are some cool things. Use a Schema (like zod) to ensure your data types are correct (since there seem to be some DataCloneError with current Workers). Workflows is very cool!!

:cloudflare:

class UserEventSchema extends Schema.TaggedRequest<UserEventSchema>()("UserEventSchema", {
  failure: Schema.Defect,
  success: Schema.Option(Schema.Struct({ name: Schema.String })),
  payload: {
    name: Schema.String,
    payload: Schema.Option(Schema.Struct({ ... })),
  },
}) {}

const ComplexSteps = Workflow.schema(
  // encode | decode
  UserEventSchema,
  ({ name, payload }) =>
    Effect.gen(function* () {
      yield* Effect.logInfo(name, Option.getOrNull(payload));

      return Option.some({ name });
    }),
  {
    retries: { limit: 3, delay: "1 seconds" },
  },
);

const SimpleStep = Workflow.do("step1", Effect.log("hello"));

const MyWorkflow = makeWorkflowEntrypoint(
  { name: "MyWorkflow", binding: "MY_WORKFLOW", schema: WorkflowParams },
  Effect.gen(function* () {
    yield* SimpleStep;
    yield* Workflow.do("step2", Effect.log("world"));
    yield* Workflow.sleep("sleep", "1 minute");
    yield* ComplexSteps({ name: "cloudflare", payload: Option.some({ ... }) }); // return: Option
  }),
);
const workflow = yield* Workflows.getWorkflow<WorkflowsBinding>("MyWorkflow");
const instance = yield* workflow.create({ params: { ... } }); // typesafe create

class UserEventSchema extends Schema.TaggedRequest<UserEventSchema>()("UserEventSchema", {
  failure: Schema.Defect,
  success: Schema.Option(Schema.Struct({ name: Schema.String })),
  payload: {
    name: Schema.String,
    payload: Schema.Option(Schema.Struct({ ... })),
  },
}) {}

const ComplexSteps = Workflow.schema(
  // encode | decode
  UserEventSchema,
  ({ name, payload }) =>
    Effect.gen(function* () {
      yield* Effect.logInfo(name, Option.getOrNull(payload));

      return Option.some({ name });
    }),
  {
    retries: { limit: 3, delay: "1 seconds" },
  },
);

const SimpleStep = Workflow.do("step1", Effect.log("hello"));

const MyWorkflow = makeWorkflowEntrypoint(
  { name: "MyWorkflow", binding: "MY_WORKFLOW", schema: WorkflowParams },
  Effect.gen(function* () {
    yield* SimpleStep;
    yield* Workflow.do("step2", Effect.log("world"));
    yield* Workflow.sleep("sleep", "1 minute");
    yield* ComplexSteps({ name: "cloudflare", payload: Option.some({ ... }) }); // return: Option
  }),
);
const workflow = yield* Workflows.getWorkflow<WorkflowsBinding>("MyWorkflow");
const instance = yield* workflow.create({ params: { ... } }); // typesafe create

class UserEventSchema extends Schema.TaggedRequest<UserEventSchema>()("UserEventSchema", {
  failure: Schema.Defect,
  success: Schema.Option(Schema.Struct({ name: Schema.String })),
  payload: {
    name: Schema.String,
    payload: Schema.Option(Schema.Struct({ ... })),
  },
}) {}

const ComplexSteps = Workflow.schema(
  // encode | decode
  UserEventSchema,
  ({ name, payload }) =>
    Effect.gen(function* () {
      yield* Effect.logInfo(name, Option.getOrNull(payload));

      return Option.some({ name });
    }),
  {
    retries: { limit: 3, delay: "1 seconds" },
  },
);

const SimpleStep = Workflow.do("step1", Effect.log("hello"));

const MyWorkflow = makeWorkflowEntrypoint(
  { name: "MyWorkflow", binding: "MY_WORKFLOW", schema: WorkflowParams },
  Effect.gen(function* () {
    yield* SimpleStep;
    yield* Workflow.do("step2", Effect.log("world"));
    yield* Workflow.sleep("sleep", "1 minute");
    yield* ComplexSteps({ name: "cloudflare", payload: Option.some({ ... }) }); // return: Option
  }),
);
const workflow = yield* Workflows.getWorkflow<WorkflowsBinding>("MyWorkflow");
const instance = yield* workflow.create({ params: { ... } }); // typesafe create

class UserEventSchema extends Schema.TaggedRequest<UserEventSchema>()("UserEventSchema", {
  failure: Schema.Defect,
  success: Schema.Option(Schema.Struct({ name: Schema.String })),
  payload: {
    name: Schema.String,
    payload: Schema.Option(Schema.Struct({ ... })),
  },
}) {}

const ComplexSteps = Workflow.schema(
  // encode | decode
  UserEventSchema,
  ({ name, payload }) =>
    Effect.gen(function* () {
      yield* Effect.logInfo(name, Option.getOrNull(payload));

      return Option.some({ name });
    }),
  {
    retries: { limit: 3, delay: "1 seconds" },
  },
);

const SimpleStep = Workflow.do("step1", Effect.log("hello"));

const MyWorkflow = makeWorkflowEntrypoint(
  { name: "MyWorkflow", binding: "MY_WORKFLOW", schema: WorkflowParams },
  Effect.gen(function* () {
    yield* SimpleStep;
    yield* Workflow.do("step2", Effect.log("world"));
    yield* Workflow.sleep("sleep", "1 minute");
    yield* ComplexSteps({ name: "cloudflare", payload: Option.some({ ... }) }); // return: Option
  }),
);
const workflow = yield* Workflows.getWorkflow<WorkflowsBinding>("MyWorkflow");
const instance = yield* workflow.create({ params: { ... } }); // typesafe create

GitHub

GitHub - opraying/worker-workflow-test

Contribute to opraying/worker-workflow-test development by creating an account on GitHub.

D

D Trombett•12/10/24, 6:08 PM

Is it allowed to create a worflow with the id of a workflow that previously completed?

D

dave•12/10/24, 8:28 PM

Do Workflows work locally without any catches?

Ddave Do Workflows work locally without any catches?

A

arch•12/10/24, 9:11 PM

From what I've experienced, yep!

S

skorfmann•12/10/24, 9:16 PM

wondering if the workflow instance limit of 100 includes workflows which are in a sleeping step?

Aarch From what I've experienced, yep!

D

dave•12/10/24, 10:09 PM

Hmm, for some reason my workflow instance is always stuck in

queued

queued

queued

queued for the status

D

dave•12/10/24, 10:09 PM

do I need to start it or something?

D

dave•12/10/24, 10:42 PM

Why does

.get(crypto.randomUUID())

.get(crypto.randomUUID())

.get(crypto.randomUUID())

.get(crypto.randomUUID()) always return an existing instance?

D

dave•12/10/24, 10:42 PM

That seems impossible.

D

dave•12/10/24, 10:53 PM

hmm, any ideas on why it would be stuck in

queued

queued

queued

queued?

D

dave•12/10/24, 10:53 PM

getting pretty frustrated.

Dddsgadget had the same issue. hope they fix this. I think this is just a bug where the que...

D

dave•12/10/24, 10:58 PM

is this with local dev?

CCarl Hi. I'm facing issues with workflow instances stuck in a queued state indefinite...

D

dave•12/10/24, 10:58 PM

still facing it?

D

dave•12/10/24, 11:08 PM

pretty bummed, going to write off Workflows indefinitely.

MMatt Silverlock (Known bug with a release that was fixed)

W

Warren•12/11/24, 6:45 AM

I restarted it and now got a

Network connection lost.

Network connection lost.

Network connection lost.

Network connection lost. error. pretty weird!

Ddave pretty bummed, going to write off Workflows indefinitely.

M

Matt SilverlockOP•12/11/24, 11:10 AM

Workflows is in beta, so that’s unfortunate, but I think we’ve been clear that a few bugs are expected.

Ddave Why does `.get(crypto.randomUUID())` always return an existing instance?

M

Matt SilverlockOP•12/11/24, 11:22 AM

What version of wrangler are you using?

Ddave Why does `.get(crypto.randomUUID())` always return an existing instance?

D

Diogo Ferreira•12/11/24, 11:45 AM

Can you please confirm you are seeing this in local dev?

LLiudmila Hi Everyone, is it possible to use Workflow from Pages?

D

Diogo Ferreira•12/11/24, 12:08 PM

workflows

Sskorfmann wondering if the workflow instance limit of 100 includes workflows which are in ...

D

Diogo Ferreira•12/11/24, 2:51 PM

The 100 instance limit refer to instances running. Regarding instances in a sleep step, they start the sleep in a 'running' state and will eventually change to a 'waiting' state, if the sleep is long enough. 'waiting' instances do not count towards the 100 instance limit.

S

skorfmann•12/11/24, 3:55 PM

anyone has examples for testing workflows?

F

flash716•12/11/24, 3:58 PM

Is there any way to tell in a step that you are in a retry? If not, what's the recommended way to handle a clean-up that only runs after the last retry?

VVinz Can you pls. elaborate on how I would "bind" the workflow I have created to the ...

M

Matt SilverlockOP•12/11/24, 4:52 PM

You would need to call the Worker over HTTP (e.g. make a fetch() call) or a Service Binding (aka "RPC") - https://developers.cloudflare.com/workers/runtime-apis/bindings/service-bindings/#_top

which means only your Pages function can call the Worker and your Worker can reject any direct HTTP requests

Cloudflare Docs

Service bindings - Runtime APIs · Cloudflare Workers docs

Facilitate Worker-to-Worker communication.

M

Matt SilverlockOP•12/11/24, 4:54 PM

Pages specific: https://developers.cloudflare.com/pages/functions/bindings/#service-bindings

Cloudflare Docs

Bindings · Cloudflare Pages docs

A binding enables your Pages Functions to interact with resources on the Cloudflare developer platform. Use bindings to integrate your Pages Functions with Cloudflare resources like KV, Durable Objects, R2, and D1. You can set bindings for both production and preview environments.

DDiogo Ferreira Can you please confirm you are seeing this in local dev?

D

dave•12/11/24, 5:44 PM

Yes.

MMatt Silverlock What version of wrangler are you using?

D

dave•12/11/24, 5:52 PM

3.95.0

M

Miguel Franco•12/11/24, 6:20 PM

When configuring the step retry settings by setting both limit and delay to 0, the application throws an error: "e2.match is not a function". The error is only related to the delay, setting the limit to 0 works fine.

VVinz Can you pls. elaborate on how I would "bind" the workflow I have created to the ...

M

Matt SilverlockOP•12/11/24, 6:34 PM

Published a new (draft) doc that shows some examples: https://silverlock-workflows-pages.cloudflare-docs-7ou.pages.dev/workflows/build/call-workflows-from-pages/

Cloudflare Docs

Call Workflows from Pages · Cloudflare Workflows docs

You can bind and trigger Workflows from Pages Functions by deploying a Workers project with your Workflow definition and then invoking that Worker using service bindings or a standard fetch() call.

Sskorfmann anyone has examples for testing workflows?

S

skorfmann•12/11/24, 8:54 PM

saw this blog post (https://blog.cloudflare.com/workers-vitest-integration/) which states: "Test anything exported from a Worker" - Just trying this with workflows, but running into

[vpw:inf] Starting isolated runtimes for vitest.config.mts...
Worker "workflows:shipped"'s binding "USER_WORKFLOW" refers to service "core:user:vitest-pool-workers-runner-" with a named entrypoint "ShippedWorkflow", but "core:user:vitest-pool-workers-runner-" has no such named entrypoint.

running the worker via

npm run dev

npm run dev

npm run dev

npm run dev works as expected. Ideas?

The Cloudflare Blog

Improved Cloudflare Workers testing via Vitest and workerd

Today, we’re excited to announce a new Workers Vitest integration - allowing you to write unit and integration tests via the popular testing framework, Vitest, that execute directly in our runtime, workerd!

Sskorfmann saw this blog post (https://blog.cloudflare.com/workers-vitest-integration/) whi...

S

skorfmann•12/11/24, 8:57 PM

ok, just saw this issue https://github.com/cloudflare/workers-sdk/issues/7414

GitHub

🐛 BUG: Vitest does not work with workflows · Issue #7414 · cloudfla...

Which Cloudflare product(s) does this pertain to? Miniflare, Other What version(s) of the tool(s) are you using? 0.5.32 [@cloudflare/vitest-pool-workers], 3.91.0 [wrangler] What version of Node are...

O

oof2win2•12/12/24, 12:37 AM

is there some way to temporarily fix/workaround the issue of instances getting into the stuck state, where they don't have any input and aren't triggered? is it like a thing with how many instances are created on a per-second basis?

O

oof2win2•12/12/24, 1:58 AM

seems it isn't a creation per second issue. have it in a queue, between each item there is a 1 second delay.

R

Rakhim•12/12/24, 11:10 AM

There're several messages about

An RPC stub was not disposed properly

An RPC stub was not disposed properly

An RPC stub was not disposed properly

An RPC stub was not disposed properly, but I still couldn't figure out the issue. I made a minimal reproducible example. My workflow has 1 step and does nothing except for logging:

export class AddOrUpdateItemWorkflow extends WorkflowEntrypoint<Bindings, Params> {
    async run(event: WorkflowEvent<Params>, step: WorkflowStep) {
        await step.do('assign item sqid', async () => {
            console.log(event.payload.item.title);
        });
    }
}

export class AddOrUpdateItemWorkflow extends WorkflowEntrypoint<Bindings, Params> {
    async run(event: WorkflowEvent<Params>, step: WorkflowStep) {
        await step.do('assign item sqid', async () => {
            console.log(event.payload.item.title);
        });
    }
}

export class AddOrUpdateItemWorkflow extends WorkflowEntrypoint<Bindings, Params> {
    async run(event: WorkflowEvent<Params>, step: WorkflowStep) {
        await step.do('assign item sqid', async () => {
            console.log(event.payload.item.title);
        });
    }
}

export class AddOrUpdateItemWorkflow extends WorkflowEntrypoint<Bindings, Params> {
    async run(event: WorkflowEvent<Params>, step: WorkflowStep) {
        await step.do('assign item sqid', async () => {
            console.log(event.payload.item.title);
        });
    }
}

My worker calls it like so

const instance = await env.ADD_UPDATE_ITEM_WORKFLOW.create({params: { item }});
console.log(instance.id);
console.log(await instance.status());

const instance = await env.ADD_UPDATE_ITEM_WORKFLOW.create({params: { item }});
console.log(instance.id);
console.log(await instance.status());

const instance = await env.ADD_UPDATE_ITEM_WORKFLOW.create({params: { item }});
console.log(instance.id);
console.log(await instance.status());

const instance = await env.ADD_UPDATE_ITEM_WORKFLOW.create({params: { item }});
console.log(instance.id);
console.log(await instance.status());

When testing locally, in about 25% of cases I get:

[ERROR] An RPC stub was not disposed properly. You must call dispose() on all stubs in order to let the other side know that you are no longer using them. You cannot rely on the garbage collector for this because it may take arbitrarily long before actually collecting unreachable objects. As a shortcut, calling dispose() on the result of an RPC call disposes all stubs within it.

[ERROR] An RPC stub was not disposed properly. You must call dispose() on all stubs in order to let the other side know that you are no longer using them. You cannot rely on the garbage collector for this because it may take arbitrarily long before actually collecting unreachable objects. As a shortcut, calling dispose() on the result of an RPC call disposes all stubs within it.

[ERROR] An RPC stub was not disposed properly. You must call dispose() on all stubs in order to let the other side know that you are no longer using them. You cannot rely on the garbage collector for this because it may take arbitrarily long before actually collecting unreachable objects. As a shortcut, calling dispose() on the result of an RPC call disposes all stubs within it.

[ERROR] An RPC stub was not disposed properly. You must call dispose() on all stubs in order to let the other side know that you are no longer using them. You cannot rely on the garbage collector for this because it may take arbitrarily long before actually collecting unreachable objects. As a shortcut, calling dispose() on the result of an RPC call disposes all stubs within it.

Is this a Miniflare issue or I'm missing something fundamental? The workflows seemingly finish correctly without errors.

RRakhim There're several messages about `An RPC stub was not disposed properly`, but I s...

M

Matt SilverlockOP•12/12/24, 11:22 AM

Miniflare/local dev. What version of wrangler are you on?

MMatt Silverlock Miniflare/local dev. What version of wrangler are you on?

R

Rakhim•12/12/24, 11:23 AM

Latest wrangler

3.95.0

3.95.0

3.95.0

3.95.0

M

Matt SilverlockOP•12/12/24, 11:23 AM

Got it. We’ll take a look!

RRakhim There're several messages about `An RPC stub was not disposed properly`, but I s...

M

Matt SilverlockOP•12/12/24, 11:24 AM

@Diogo Ferreira FYI

R

Rakhim•12/12/24, 3:45 PM

I'm now using Workflows in production. Dozens of runs, everything works great! The ability to view each instance in the dashboard is very cool and helps a lot.
I have noticed, however, that each instance run produced an error log like this:

{
    "$cloudflare": {
      "diagnosticsChannelEvents": [],
      "truncated": false,
      "event": {
        "rpcMethod": "run"
      },
      "scriptName": "minifeed",
      "outcome": "exception",
      "eventType": "rpc",
      "entrypoint": "AddOrUpdateItemWorkflow",
      "scriptVersion": {
        "id": "44488eac-<REDACTED>-b1db278bc4cc"
      },
      "executionModel": "stateless",
      "$metadata": {
        "requestId": "L7NQ<REDACTED>U6AF",
        "id": "01JE<REDACTED>VKMK8",
        "type": "cf-worker-event",
        "error": "run"
      }
    },
    "level": "error",
    "message": "run"
  }

{
    "$cloudflare": {
      "diagnosticsChannelEvents": [],
      "truncated": false,
      "event": {
        "rpcMethod": "run"
      },
      "scriptName": "minifeed",
      "outcome": "exception",
      "eventType": "rpc",
      "entrypoint": "AddOrUpdateItemWorkflow",
      "scriptVersion": {
        "id": "44488eac-<REDACTED>-b1db278bc4cc"
      },
      "executionModel": "stateless",
      "$metadata": {
        "requestId": "L7NQ<REDACTED>U6AF",
        "id": "01JE<REDACTED>VKMK8",
        "type": "cf-worker-event",
        "error": "run"
      }
    },
    "level": "error",
    "message": "run"
  }

{
    "$cloudflare": {
      "diagnosticsChannelEvents": [],
      "truncated": false,
      "event": {
        "rpcMethod": "run"
      },
      "scriptName": "minifeed",
      "outcome": "exception",
      "eventType": "rpc",
      "entrypoint": "AddOrUpdateItemWorkflow",
      "scriptVersion": {
        "id": "44488eac-<REDACTED>-b1db278bc4cc"
      },
      "executionModel": "stateless",
      "$metadata": {
        "requestId": "L7NQ<REDACTED>U6AF",
        "id": "01JE<REDACTED>VKMK8",
        "type": "cf-worker-event",
        "error": "run"
      }
    },
    "level": "error",
    "message": "run"
  }

{
    "$cloudflare": {
      "diagnosticsChannelEvents": [],
      "truncated": false,
      "event": {
        "rpcMethod": "run"
      },
      "scriptName": "minifeed",
      "outcome": "exception",
      "eventType": "rpc",
      "entrypoint": "AddOrUpdateItemWorkflow",
      "scriptVersion": {
        "id": "44488eac-<REDACTED>-b1db278bc4cc"
      },
      "executionModel": "stateless",
      "$metadata": {
        "requestId": "L7NQ<REDACTED>U6AF",
        "id": "01JE<REDACTED>VKMK8",
        "type": "cf-worker-event",
        "error": "run"
      }
    },
    "level": "error",
    "message": "run"
  }

But each run's status is

Successful

Successful

Successful

Successful nevertheless.

VVinz Follow up questions: a) if a workflow is completed, can I then create a new one ...

M

Matt SilverlockOP•12/12/24, 6:16 PM

Workflow instance IDs are unique per Workflow. The ID is the "the" identifier for finding the logs, metrics and status of an instance, even after completion. Re-using IDs would make it very hard to know if something was kicked off before.
Why not pass the
```
userId
```
```
userId
```
in the params, generate a composite ID e.g.
```
${userId}-${randomHex}
```
${userId}-${randomHex}
```
${userId}-${randomHex}
```
${userId}-${randomHex}
- and store that ID in D1? What if you had more Workflows you wanted to run for a given userId in the future? Emails? Cleanup? Different message campaigns? etc
Error