Yeah, that is correct, since Workflows itself doesn’t know whether the names are the same or not, or

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!OP•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

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

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

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.

Yeah, that is correct, since Workflows itself doesn’t know whether the names are the same or not, or

Similar Threads