Is there a way to limit concurrent workflows based on their id? For example if the id starts with 't

Aapolodoro Is there a way to limit concurrent workflows based on their id? For example if t...

M

Not yet. Looking at controls here by Workflow - e.g. a

max_concurrent_instances

max_concurrent_instances

max_concurrent_instances

max_concurrent_instances config.

Ccrunchy Also what sort of concurrency is likely to be allowed in future, are these thing...

M

Matt Silverlock•2/2/25, 2:42 PM

Next big step is 4,500 concurrent (very soon) and we want to get to 10k (at least) over the next few weeks/this quarter.

Making sure it all works + nothing bottlenecks.

MMatt Silverlock Not yet. Looking at controls here by Workflow - e.g. a `max_concurrent_instances...

A

apolodoroOP•2/2/25, 2:56 PM

Nice but not yet implemented right?

Aapolodoro Nice but not yet implemented right?

S

Seekerdasbatatas•2/3/25, 11:53 AM

correct

SSeekerdasbatatas correct

J

Jadugar_Jaggu•2/3/25, 4:13 PM

Can I know the limit of memory per workflow ?

For workers it's 128 MB shared per invocation right?

JJadugar_Jaggu Can I know the limit of memory per workflow ? For workers it's 128 MB shared p...

S

Seekerdasbatatas•2/3/25, 4:15 PM

it's the same 128MiB on Workflows too (and depending on scheduling, it maybe be shared with other instances)

SSeekerdasbatatas it's the same 128MiB on Workflows too (and depending on scheduling, it maybe be ...

J

Jadugar_Jaggu•2/3/25, 4:16 PM

Ok it means every step will get 128 MiB or a whole workerflow ?

JJadugar_Jaggu Ok it means every step will get 128 MiB or a whole workerflow ?

S

Seekerdasbatatas•2/3/25, 4:17 PM

the whole workflow

SSeekerdasbatatas the whole workflow

J

Jadugar_Jaggu•2/3/25, 4:20 PM

Ok thanks, well this is somewhat concerning if there are some tasks which use memory too much.

That's why may be my workflow steps are throwing an error of cpu/memory limit after doing work of around 128 MiB ( nearly )

But on retry it's again working till 128 MiB

what's this behaviour?

D

D Trombett•2/3/25, 6:07 PM

Docs say

step.sleep

step.sleep

step.sleep

step.sleep duration is in seconds if it's a number but it seems to me that it's actually ms?

D

D Trombett•2/3/25, 7:26 PM

ah I see, thanks

V

Victor•2/4/25, 9:44 AM

I'm unable to use cross-script calls to workflows with wrangler dev (local)

Getting

Worker "workflows:AfterContractActivated-development"'s binding "USER_WORKFLOW" refers to a service "core:user:delivery-development", but no such service is defined.

Worker "workflows:AfterContractActivated-development"'s binding "USER_WORKFLOW" refers to a service "core:user:delivery-development", but no such service is defined.

Worker "workflows:AfterContractActivated-development"'s binding "USER_WORKFLOW" refers to a service "core:user:delivery-development", but no such service is defined.

Worker "workflows:AfterContractActivated-development"'s binding "USER_WORKFLOW" refers to a service "core:user:delivery-development", but no such service is defined.

Is that not supported? It doesn't seem specified in the docs

L

laurynas•2/4/25, 1:31 PM

has there been any discussion on setting error callbacks for workflow steps? I’d love an

onError

onError

onError

onError callback i could pass where i could set some custom logic to try when a step errors. E.g.:

export class TextGenerator extends WorkflowEntrypoint<CloudflareBindings> {
  async run(
    event: Readonly<WorkflowEvent<{ prompt: string }>>,
    step: WorkflowStep,
  ) {
    const { prompt } = event.payload;

    return await step.do(
      "generate-text",
      {
        retries: { limit: 3, delay: 1000, backoff: "exponential" },
        onError: async (error, retry) => {
          if (error instanceof AITimeoutError) {
            // Complex model timed out, try with a faster one first
            try {
              return await generateText({
                model: "gpt-3.5-turbo",
                prompt,
                timeout: 5000,
              });
            } catch {
              // If faster model fails, retry with original complex model (retry is the original callback in step.do)
              return retry();
            }
          }
          throw error;
        }
      },

export class TextGenerator extends WorkflowEntrypoint<CloudflareBindings> {
  async run(
    event: Readonly<WorkflowEvent<{ prompt: string }>>,
    step: WorkflowStep,
  ) {
    const { prompt } = event.payload;

    return await step.do(
      "generate-text",
      {
        retries: { limit: 3, delay: 1000, backoff: "exponential" },
        onError: async (error, retry) => {
          if (error instanceof AITimeoutError) {
            // Complex model timed out, try with a faster one first
            try {
              return await generateText({
                model: "gpt-3.5-turbo",
                prompt,
                timeout: 5000,
              });
            } catch {
              // If faster model fails, retry with original complex model (retry is the original callback in step.do)
              return retry();
            }
          }
          throw error;
        }
      },

export class TextGenerator extends WorkflowEntrypoint<CloudflareBindings> {
  async run(
    event: Readonly<WorkflowEvent<{ prompt: string }>>,
    step: WorkflowStep,
  ) {
    const { prompt } = event.payload;

    return await step.do(
      "generate-text",
      {
        retries: { limit: 3, delay: 1000, backoff: "exponential" },
        onError: async (error, retry) => {
          if (error instanceof AITimeoutError) {
            // Complex model timed out, try with a faster one first
            try {
              return await generateText({
                model: "gpt-3.5-turbo",
                prompt,
                timeout: 5000,
              });
            } catch {
              // If faster model fails, retry with original complex model (retry is the original callback in step.do)
              return retry();
            }
          }
          throw error;
        }
      },

export class TextGenerator extends WorkflowEntrypoint<CloudflareBindings> {
  async run(
    event: Readonly<WorkflowEvent<{ prompt: string }>>,
    step: WorkflowStep,
  ) {
    const { prompt } = event.payload;

    return await step.do(
      "generate-text",
      {
        retries: { limit: 3, delay: 1000, backoff: "exponential" },
        onError: async (error, retry) => {
          if (error instanceof AITimeoutError) {
            // Complex model timed out, try with a faster one first
            try {
              return await generateText({
                model: "gpt-3.5-turbo",
                prompt,
                timeout: 5000,
              });
            } catch {
              // If faster model fails, retry with original complex model (retry is the original callback in step.do)
              return retry();
            }
          }
          throw error;
        }
      },

Llaurynas has there been any discussion on setting error callbacks for workflow steps? I’d...

M

Matt Silverlock•2/4/25, 4:46 PM

Isn’t that just another step? Write an outer step that encapsulates both.

Any

onError

onError

onError

onError callback would still have to be treated like another step, but ends up being another not-quite-a-step API with its own semantics.

Run stepA, catch the error, call optional stepB, etc.

D

Dom•2/5/25, 4:10 AM

Hello!
I'm working on my first workflow. Its supposed to send a couple of emails via Resend.
The workflow itself works.

But I want to call it from a Pages project.

In my wrangler.jsonc I have:

  "services": [
    {
      "binding": "WORKFLOW_SERVICE",
      "service": "wf-portcity-ai"
    }
  ],

  "services": [
    {
      "binding": "WORKFLOW_SERVICE",
      "service": "wf-portcity-ai"
    }
  ],

  "services": [
    {
      "binding": "WORKFLOW_SERVICE",
      "service": "wf-portcity-ai"
    }
  ],

  "services": [
    {
      "binding": "WORKFLOW_SERVICE",
      "service": "wf-portcity-ai"
    }
  ],

But what should I type the

WORKFLOW_SERVICE

WORKFLOW_SERVICE

WORKFLOW_SERVICE

WORKFLOW_SERVICE Env variabe as?
And how do I trigger my workflow?

D

Dom•2/5/25, 4:37 AM

https://developers.cloudflare.com/workflows/build/call-workflows-from-pages/

This would really benefit from a complete source example

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.

D

Dom•2/5/25, 4:45 AM

Whatever I seem to do - I always get:

✘ [ERROR] TypeError: The RPC receiver does not implement the method "createInstance"

✘ [ERROR] TypeError: The RPC receiver does not implement the method "createInstance"

✘ [ERROR] TypeError: The RPC receiver does not implement the method "createInstance"

✘ [ERROR] TypeError: The RPC receiver does not implement the method "createInstance"

Nnil Alternatively.. if that breaks rules of global state.. wondering if this works? ...

S

Seekerdasbatatas•2/5/25, 10:42 AM

that could work, but

yieldEvent

yieldEvent

yieldEvent

yieldEvent is outside of a step - meaning that these calls can be repeated multiple times (not sure if that's you want or not)

SSushidata Hi all, day 2 now of not being able to access Workflows 😦

M

Matt Silverlock•2/5/25, 6:18 PM

Can you clear all of your cookies? cc @Diogo Ferreira

DDom Whatever I seem to do - I always get: `✘ [ERROR] TypeError: The RPC receiver doe...

M

Matt Silverlock•2/5/25, 6:19 PM

Can you share your code?

MMatt Silverlock Can you clear all of your cookies? cc @Diogo Ferreira

V

Victor•2/5/25, 6:20 PM

https://discord.com/channels/595317990191398933/1299119533319131209/1336492428353474601

Nnil The motivation is when you do workflows, you would want to incrementally update ...

F

Focux•2/5/25, 6:26 PM

It'd be cool if you could update some state from within a workflow and then be able to access that state by calling the workflow binding from a worker

F

Focux•2/5/25, 6:26 PM

I guess you can implement this on userland, but it'd be cool if it was built-in

T

Teddy•2/5/25, 9:52 PM

Hi

I'm following the RAGWorkflow tutorial in the docs, and somehow I'm getting an

internal error

internal error

internal error

internal error before the workflow tries to execute any steps. Any ideas?

export class RAGWorkflow {
    async run(event, step) {
        const { text } = event.params;

        const record = await step.do('create database record', async () => {
            const query = 'INSERT INTO notes (text) VALUES (?) RETURNING *';

            const { results } = await  env.DB.prepare(query).bind(text).run();

            const record = results[0];
            if (!record) throw new Error('Failed to create note');
            return record;
        });

        const embedding = await step.do('generate embedding', async () => {
            const embeddings = await env.AI.run('@cf/baai/bge-base-en-v1.5', { text: text });
            const values = embeddings.data[0];
            if (!values) throw new Error('Failed to generate vector embedding');
            return values;
        });

        await step.do('insert vector', async () => {
            return env.VECTOR_INDEX.upsert([
                {
                    id: record.id.toString(),
                    values: embedding,
                },
            ]);
        });
    }
}

export class RAGWorkflow {
    async run(event, step) {
        const { text } = event.params;

        const record = await step.do('create database record', async () => {
            const query = 'INSERT INTO notes (text) VALUES (?) RETURNING *';

            const { results } = await  env.DB.prepare(query).bind(text).run();

            const record = results[0];
            if (!record) throw new Error('Failed to create note');
            return record;
        });

        const embedding = await step.do('generate embedding', async () => {
            const embeddings = await env.AI.run('@cf/baai/bge-base-en-v1.5', { text: text });
            const values = embeddings.data[0];
            if (!values) throw new Error('Failed to generate vector embedding');
            return values;
        });

        await step.do('insert vector', async () => {
            return env.VECTOR_INDEX.upsert([
                {
                    id: record.id.toString(),
                    values: embedding,
                },
            ]);
        });
    }
}

export class RAGWorkflow {
    async run(event, step) {
        const { text } = event.params;

        const record = await step.do('create database record', async () => {
            const query = 'INSERT INTO notes (text) VALUES (?) RETURNING *';

            const { results } = await  env.DB.prepare(query).bind(text).run();

            const record = results[0];
            if (!record) throw new Error('Failed to create note');
            return record;
        });

        const embedding = await step.do('generate embedding', async () => {
            const embeddings = await env.AI.run('@cf/baai/bge-base-en-v1.5', { text: text });
            const values = embeddings.data[0];
            if (!values) throw new Error('Failed to generate vector embedding');
            return values;
        });

        await step.do('insert vector', async () => {
            return env.VECTOR_INDEX.upsert([
                {
                    id: record.id.toString(),
                    values: embedding,
                },
            ]);
        });
    }
}

export class RAGWorkflow {
    async run(event, step) {
        const { text } = event.params;

        const record = await step.do('create database record', async () => {
            const query = 'INSERT INTO notes (text) VALUES (?) RETURNING *';

            const { results } = await  env.DB.prepare(query).bind(text).run();

            const record = results[0];
            if (!record) throw new Error('Failed to create note');
            return record;
        });

        const embedding = await step.do('generate embedding', async () => {
            const embeddings = await env.AI.run('@cf/baai/bge-base-en-v1.5', { text: text });
            const values = embeddings.data[0];
            if (!values) throw new Error('Failed to generate vector embedding');
            return values;
        });

        await step.do('insert vector', async () => {
            return env.VECTOR_INDEX.upsert([
                {
                    id: record.id.toString(),
                    values: embedding,
                },
            ]);
        });
    }
}

T

Teddy•2/5/25, 9:54 PM

T

Teddy•2/5/25, 10:37 PM

Nevermind, figured it out! Tutorial I was following here: https://developers.cloudflare.com/workers-ai/tutorials/build-a-retrieval-augmented-generation-ai/#5-creating-a-workflow

It's not mentioned, but the workflow class needs to extend

WorkflowEntrypoint

WorkflowEntrypoint

WorkflowEntrypoint

WorkflowEntrypoint and bindings are accessed via

this.env

this.env

this.env

this.env rather than just

env

env

env

env

D

D Trombett•2/6/25, 11:17 AM

It needs to match the pattern

^[a-zA-Z0-9_][a-zA-Z0-9-_]*$

^[a-zA-Z0-9_][a-zA-Z0-9-_]*$

^[a-zA-Z0-9_][a-zA-Z0-9-_]*$

^[a-zA-Z0-9_][a-zA-Z0-9-_]*$ (https://developers.cloudflare.com/workflows/reference/limits)

Mm.muiznieks I guess this issue is similar as what's already posted here: https://discord.co...

R

Rakhim•2/6/25, 11:30 AM

hey! Have you found a way to resolve these issues? I'm still experiencing a big buildup of the backlog queue, and only 9-10 workflows run simultaneously (account-wide).

RRakhim hey! Have you found a way to resolve these issues? I'm still experiencing a big ...

S

Seekerdasbatatas•2/6/25, 11:50 AM

give me your account id and I'll unstuck it once changes roll out

RRakhim hey! Have you found a way to resolve these issues? I'm still experiencing a big ...

M

m.muiznieks•2/6/25, 1:27 PM

From what @Seekerdasbatatas says, I understand that this bug has been identified, and a fix will be rolled soon.

Nnil There's inconsistencies with what is and isn't supported with remote and local w...

M

Matt Silverlock•2/7/25, 2:10 AM

Working on remote for Workflows. Still some things to work on while we’re in beta.

SSeekerdasbatatas To keep you posted, the bug has been identified and will be rolling out througho...

M

m.muiznieks•2/7/25, 2:51 AM

Is there any possibility that this bug fix will be out at least in next week? It was funny before, but now I am thinking about whether to use the Workflows during its beta period at all. :/

The current backlog is already -12 hours and with this tendency it will go to an even lower rate of workflow processing (and there isn't anything resource-heavy).

By looking into the solutions I even bought the domain subscription, but it went even worse.

I understand that it is hard to hope for any technical support here, except for some little advice, but is it different through ticket support? (this is what I would like to know - simple yes/no will be acceptable).

p.s. if someone again asks for my account ID:

5e3dea7b807381cba36b9292bde14920

5e3dea7b807381cba36b9292bde14920

5e3dea7b807381cba36b9292bde14920

5e3dea7b807381cba36b9292bde14920

SSeekerdasbatatas give me your account id and I'll unstuck it once changes roll out 🙂

R

Rakhim•2/7/25, 8:39 AM

Account number

1ac93aea0842f2d3ef10111b5bf7bb11

1ac93aea0842f2d3ef10111b5bf7bb11

1ac93aea0842f2d3ef10111b5bf7bb11

1ac93aea0842f2d3ef10111b5bf7bb11
Thanks!

Mm.muiznieks Is there any possibility that this bug fix will be out at least in next week? It...

M

m.muiznieks•2/7/25, 3:50 PM

Now I am starting to see at the logs these type of errors.
A wall of them.

Error: Durable Object storage operation exceeded timeout which caused object to be reset.

This, for example, was 56 in a row.

Now I have a queue of 44,000+ events and growing, with an average throughput of 18 instances per minute, account-wide.

Now I think I will stop this, as at this rate it will take almost 48 hours to process those Facebook conversion call api instances.

W

will smith•2/7/25, 5:30 PM

is

npx wrangler types

npx wrangler types

npx wrangler types

npx wrangler types supposed to work for workflows?
wrangler.toml:

services = [{ binding = "WORKFLOW_PROCESS_TRANSACTIONS", service = "money" }]

services = [{ binding = "WORKFLOW_PROCESS_TRANSACTIONS", service = "money" }]

services = [{ binding = "WORKFLOW_PROCESS_TRANSACTIONS", service = "money" }]

services = [{ binding = "WORKFLOW_PROCESS_TRANSACTIONS", service = "money" }]
wrangler version:

wrangler 3.107.3

wrangler 3.107.3

wrangler 3.107.3

wrangler 3.107.3
generated interface, WORKFLOW_PROCESS_TRANSACTIONS should be Workflow?

interface Env {
    SESSION_SECRET: string;
    DB: D1Database;
    WORKFLOW_PROCESS_TRANSACTIONS: Fetcher;
}

interface Env {
    SESSION_SECRET: string;
    DB: D1Database;
    WORKFLOW_PROCESS_TRANSACTIONS: Fetcher;
}

interface Env {
    SESSION_SECRET: string;
    DB: D1Database;
    WORKFLOW_PROCESS_TRANSACTIONS: Fetcher;
}

interface Env {
    SESSION_SECRET: string;
    DB: D1Database;
    WORKFLOW_PROCESS_TRANSACTIONS: Fetcher;
}

RRakhim hey! Have you found a way to resolve these issues? I'm still experiencing a big ...

M

m.muiznieks•2/8/25, 2:39 PM

I noticed some patterns - the workflow instance processing slows down proportionally the number of incoming requests by not only to the workflow as a service, but it looks like the worker where they are processed as a whole.

Switching off new instance processing is raising the "speed" a little bit, but not enough. By looking at the graphs - I see that during the night when the peak hours are over (I am processing website-related events), it raises up to 2,000 instances per hour (which is 33 per minute, still far from the 100). Even if the new workflows are not created, processing speed is dropping when the users are getting active again (request amount increases).

So I followed this article: https://developers.cloudflare.com/queues/get-started/ and implemented additional Queue layer, that batches the requests and each instance is processing them by up to 100. For now I am testing the 50/50 approach, where 1/2 goes through the queues and 1/2 makes direct API calls without the individual instance firing and I see that the drop of speed during active hours are much lower.

So, this is the solution I found and it looks that it works in this situation. Still hope on some wider fix soon.

Cloudflare Docs

Get started · Cloudflare Queues

Cloudflare Queues is a flexible messaging queue that allows you to queue messages for asynchronous processing. By following this guide, you will create your first queue, a Worker to publish messages to that queue, and a consumer Worker to consume messages from that queue.

Is there a way to limit concurrent workflows based on their id? For example if the id starts with 't

Similar Threads