we're seeing an issue where some workflows are getting stuck in the queue. I only have two queued wo

FFocux @skip i asked that some days ago, and currently, you can't. they said they will ...

S

skip•12/6/24, 6:43 AM

Ah thanks for the reply. This is a feature that would make workflows hugely useful, imo!

Ooof2win2 would this be valid workflows code? everything is deterministic as all results a...

D

DeChill•12/6/24, 8:11 AM

According to https://developers.cloudflare.com/workflows/build/rules-of-workflows/, awaiting multiple steps with

Promise.all

Promise.all

is fine.

But nested steps seem like they might cause issues with the retry logic? I would suggest just placing the

saveNotesStep

saveNotesStep

saveNotesStep

saveNotesStep after the

Promise.all

Promise.all

and reading the output of the step. In that case if the "save" step fails, the

generateNotesStep

generateNotesStep

generateNotesStep

generateNotesStep will have cached the response. It might only be an issue if the serialized data exceeds the limits.

DDeChill According to https://developers.cloudflare.com/workflows/build/rules-of-workflow...

O

oof2win2•12/6/24, 8:14 AM

the issue is that generatesNotesStep might complete earlier than generateTextStep, so i want to save it to the db as soon as it generates

O

oof2win2•12/6/24, 8:15 AM

or should i just make it two nested steps? or an async function that awaits the first step and then saves it (two await step.do calls)

O

oof2win2•12/6/24, 8:15 AM

the last one kinda makes sense

D

DeChill•12/6/24, 8:21 AM

If you nest the save step inside the generate step, then I don't think it will cache the generate response if the inner save fails. So I think you need to keep them separate. 2 await steps are probably most idiomatic, but the second will only run when the first is finished.

SSeekerdasbatatas Or via the binding

D

DeChill•12/6/24, 3:19 PM

I really hope getting the workflow output by binding will be implemented in the near future! Is there any roadmap or feedback site for workflows?

D

D Trombett•12/6/24, 7:33 PM

From the workflows get started guide:

Because Workflows can be long running, it is possible to have running instances that represent different versions of your Workflow code over time.

Does it mean that if, for example, I update my code while the workflow is waiting a

sleep()

sleep()

sleep()

sleep() step, when the sleep is over it will execute the old code and not the updated steps (considering that the step names are the same if it matters)?

DD Trombett From the workflows get started guide: > Because Workflows can be long running, i...

H

Hello, I’m Allie!•12/7/24, 12:45 AM

Yeah, that is correct, since Workflows itself doesn’t know whether the names are the same or not, or whether it is safe to mix a run like that

W

Warren•12/7/24, 2:44 AM

I am using Workers in Workflows and I am running up against the 1000 subrequests limit. my question is is there a way to clearly see what are the subrequests made in a Worker?

so for context, my worker is trying to do long running job in batches. the issue is I am using another Durable Object as a rate limiter. the problem with rate limiting is I can’t estimate how frequently the rate limiter will be called in a single worker run.

now that I’ve gotten rid of the rate limiter, it seems to not run into the subrequest issue anymore.

my feedback would be:

if a Worker in a Workflow has hit the subrequest limit, it seems pointless to retry since any retries would also by definition hit the same limit
I personally feel that this subrequest limit really dampens the promise of Cloudflare platform being serverless and scales infinitely, esp if it includes calling internal Cloudflare resources like Durable

WWarren I am using Workers in Workflows and I am running up against the 1000 subrequests...

H

Hello, I’m Allie!•12/7/24, 3:30 AM

On your first point, not necessarily, since if the first run makes the 1,000 requests, stores its state, and is restarted, it will skip those first 1k requests

HHello, I’m Allie!Yeah, that is correct, since Workflows itself doesn’t know whether the names are...

D

D Trombett•12/7/24, 12:46 PM

I see, thanks! So, if I want to "switch" to the new code, should I implement it myself like cancelling the sleeping workflow and running a new one?

D

D Trombett•12/7/24, 12:49 PM

Oh, wait, is it possible to call a worker from a workflow with rpc, like we do from workers? This way I could implement the logic that may change in the worker and call it from the workflow

D

D Trombett•12/7/24, 3:09 PM

this seems much better, thanks a lot!

W

Warren•12/7/24, 3:15 PM

I have observed that the try-catch behavior described in the docs do not reliably happen. i.e. sometimes when the workflow fails, the catch part of the code does not run. this is quite bad when I build systems that rely on the catch to maintain state...

Cloudflare Docs

Sleeping and retrying · Cloudflare Workflows docs

This guide details how to sleep a Workflow and/or configure retries for a Workflow step.

D

Dmitry•12/7/24, 10:56 PM

Any ideas about when performance issues will be fixed? (In relation to starting a bunch in a short time span and dashboard just loading)? Assuming with holidays right around the corner.. prob some time Jan?

D

Dmitry•12/7/24, 10:56 PM

Would love for the dashboard for Workflows to be more real time

D

Dmitry•12/7/24, 10:57 PM

I know it refreshes every x seconds.. but would be nice for things to stream in/auto update in realtime

D

Dmitry•12/7/24, 10:57 PM

Also adding a refresh button so I don't have to reload the entire page would be awesome too

I

inaridiy•12/8/24, 4:55 AM

Is there an upper limit to the size of the Workflow Payload? If I pass a payload that exceeds 1 MiB in size, the instance will not be triggered.

O

oof2win2•12/8/24, 5:44 AM

do queued instances get killed after a certain time? they aren't getting triggered and are constantly being pushed to the front of the dashboard so it's just annoying

O

oof2win2•12/8/24, 5:49 AM

also is there a limit to how many workflows can be created per second? i have a database webhook that can execute say 40 calls per second and some instances just don't seem to get a payload and execute

O

oof2win2•12/8/24, 5:50 AM

like the input is say 500 bytes so don't think i should have any issues there

O

oof2win2•12/8/24, 6:13 AM

yeah it just happens that the tail end (last ~12 of my ~50 workflow invocations) get empty params and don't execute at all because of that

O

oof2win2•12/8/24, 6:16 AM

code works in all cases except this tail end which is just frustrating. an empty object as a body doesn't make sense here - i check for props on

body.record

body.record

body.record

body.record and they aren't null

const workflowBody: WorkflowParams = {
  coursework: {
    id: body.record.id,
...
  },
  prompt_id: body.record.prompt_id,
};
await env.grading_workflow.create({
  id: submissionId,
  params: workflowBody,
});

const workflowBody: WorkflowParams = {
  coursework: {
    id: body.record.id,
...
  },
  prompt_id: body.record.prompt_id,
};
await env.grading_workflow.create({
  id: submissionId,
  params: workflowBody,
});

const workflowBody: WorkflowParams = {
  coursework: {
    id: body.record.id,
...
  },
  prompt_id: body.record.prompt_id,
};
await env.grading_workflow.create({
  id: submissionId,
  params: workflowBody,
});

const workflowBody: WorkflowParams = {
  coursework: {
    id: body.record.id,
...
  },
  prompt_id: body.record.prompt_id,
};
await env.grading_workflow.create({
  id: submissionId,
  params: workflowBody,
});

Iinaridiy Is there an upper limit to the size of the Workflow Payload? If I pass a payload...

O

oof2win2•12/8/24, 6:59 AM

the failure rate is fairly low but it is there and it scales a lot when i scale the requests - from 12k executions i got around 600 that didn't execute at all, then from like 400 today i got 50. is there a way to debug these issues? i feel like i have the same issue as above but without the 1MB requirement - all payloads are ~400 characters/bytes

Original message was deleted

O

oof2win2•12/8/24, 7:06 PM

thank you! is there any eta on this?

O

oof2win2•12/8/24, 7:28 PM

thank you very much!

W

Warren•12/9/24, 12:45 AM

ah for anyone else facing the

Serialized RPC arguments or return values are limited to 1MiB

Serialized RPC arguments or return values are limited to 1MiB

Serialized RPC arguments or return values are limited to 1MiB

Serialized RPC arguments or return values are limited to 1MiB error

I don’t think it’s possible to catch and handle this error directly since it happens at the infra rather than the app level
something I’ve found that works is to check the return size before returning, in which case you could either throw an error or internalize the error handling somehow

CChaiSloth I've been deploying my workflow with no issues and now I get the error: ✘ [ERROR...

M

Matt Silverlock•12/9/24, 2:04 PM

Can you share your accountId (which is safe to share publicly) - ? cc @Diogo Ferreira @Seekerdasbatatas

CChaiSloth 6c4e51960caa05c7ac6f42d3151666ae

M

Matt Silverlock•12/9/24, 2:34 PM

Thanks, looking into it.

CChaiSloth I've also tried running **npx wrangler workflows list** and I'm also getting a 1...

S

Seekerdasbatatas•12/9/24, 2:53 PM

could you check and see if it's working now?

CChaiSloth thanks very much. What was the issue- had I caused it?

D

Diogo Ferreira•12/9/24, 3:23 PM

The issue was on our side. An upgrade of the base service that caused a race condition for some accounts, yours included. The issue should be fixed by now, and we are monitoring it. We are sorry for that occurrence and the inconvenience.

O

oof2win2•12/9/24, 4:59 PM

did the race condition affect new workflow invocations as well by chance? i had an outage that began ~7-8 hours ago and i can't get to the bottom of why it happened - it was out of nowhere

O

oof2win2•12/9/24, 5:05 PM

account ID

142c8bdb1fea8b57b0fb24ca54327b99

142c8bdb1fea8b57b0fb24ca54327b99

142c8bdb1fea8b57b0fb24ca54327b99

142c8bdb1fea8b57b0fb24ca54327b99 if it helps

P

Patrick•12/9/24, 8:18 PM

Related to workflows I think, does anyone know if there has been any movement with the Pipelines open beta? I'm building a clickstream ingest and don't really want to have to throw it all out if Pipelines is near as I'd much rather be using that

W

Warren•12/10/24, 2:49 AM

I have a workflow that errored out in a weird way:

the error is
```
migrate: latest schema version 1 is less than current DB schema version 2
```
migrate: latest schema version 1 is less than current DB schema version 2
```
migrate: latest schema version 1 is less than current DB schema version 2
```
migrate: latest schema version 1 is less than current DB schema version 2
my
```
try-catch
```
```
try-catch
```
code path was triggered
but the workflow itself was stuck in
```
running
```
running
```
running
```
running
for 16+ hours

I re-ran the exact same workflow and it seems to be working now. the error message doesn’t seem to be related to my app (I did not do any db migration during this period of time), makes me think it’s related to the SQLite backend powering Workfows?

WWarren I have a workflow that errored out in a weird way: - the error is `migrate: late...

M

Matt Silverlock•12/10/24, 9:20 AM

workflows

M

Matt Silverlock•12/10/24, 9:20 AM

(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?

we're seeing an issue where some workflows are getting stuck in the queue. I only have two queued wo

Similar Threads