the whole workflow

J

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

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

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

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

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"

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

S

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

that could work, but

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

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

and bindings are accessed via

this.env

this.env

rather than just

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-_]*$

(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

SeekerdasbatatasOP•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

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

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

supposed to work for workflows?
wrangler.toml:

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

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

wrangler version:

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.

Mm.muiznieks I noticed some patterns - the workflow instance processing slows down proportion...

M

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

Additional note about batch processing - be careful if you make other requests during the step processing. In a different case, I easily hit the Worker 1000 subrequest limit and instances failed.

E

Eric•2/9/25, 7:04 AM

Maybe this helps someone- to terminate all running instances, run this:

wrangler workflows instances list <YOUR_WORKFLOW_NAME> | grep "▶ Running" -B1 | grep -Eo '([a-f0-9\-]{36})' | xargs -I {} wrangler workflows instances terminate <YOUR_WORKFLOW_NAME> {}

wrangler workflows instances list <YOUR_WORKFLOW_NAME> | grep "▶ Running" -B1 | grep -Eo '([a-f0-9\-]{36})' | xargs -I {} wrangler workflows instances terminate <YOUR_WORKFLOW_NAME> {}

wrangler workflows instances list <YOUR_WORKFLOW_NAME> | grep "▶ Running" -B1 | grep -Eo '([a-f0-9\-]{36})' | xargs -I {} wrangler workflows instances terminate <YOUR_WORKFLOW_NAME> {}

wrangler workflows instances list <YOUR_WORKFLOW_NAME> | grep "▶ Running" -B1 | grep -Eo '([a-f0-9\-]{36})' | xargs -I {} wrangler workflows instances terminate <YOUR_WORKFLOW_NAME> {}

definitly helped me. You can also do

xargs -I {} -P 10 sh -c

xargs -I {} -P 10 sh -c

to parallelize

R

Rakhim•2/9/25, 11:05 AM

The situation with growing backlogs is getting out of control, my website is basically non-functional. Is there an estimated date for the release of the fix?
I have migrated from queues to workflows and now I think it was a bad idea and I should just migrate back until workflows are out of beta.

RRakhim The situation with growing backlogs is getting out of control, my website is bas...

R

Rakhim•2/9/25, 11:18 AM

Actually, despite

9 running

9 running

in the screenshot above, no workflows had run in since Friday. Workflow metrics show 0 instances and 0 steps. Please, help!

RRakhim Actually, despite `9 running` in the screenshot above, no workflows had run in s...

M

m.muiznieks•2/9/25, 1:45 PM

I solved this problem with the queues what I mentioned above.

At first you should stop processing and triggering any new instances and let them finish. It looks like if you "bombard" the Workflows with large amounts of new requests, they can't properly run them, only put them in the queue state. I don't know why.
To do that, you should switch the new incoming event processing outside of workflows, so they can get some time to "breathe" and you will see that processing will start. In my case, it was almost 48,000 queued instances at one point had to wait almost 24 hours to finish them.
See this article - I literally went through it. https://developers.cloudflare.com/queues/get-started/
Create a layer of Queue, that accumulates received events ('messages'). I set them of max 100 with wait time 60s. That means - which one comes first - the events are sent.

In short you will need to create, two new, simple workers. Really, if you got through Workflows - they are simple. Maybe it can be done even without creating additional workers, but I wasn't in the mood for FAFO.

First is the
```
producer
```
```
producer
```
worker where you send the events, and it then places them into the queue.
Second is the `consumer` worker - who receives the queued messages. I believe this can be the same as the worker where you process currently your instances, but I made this as a separate for future use cases. From here you directly trigger the workflows.
Now your workflows, instead of being triggered by 100 events 100 times, will be triggered 1 time with 100 events, that are packed as batch.

In the workflow - I created one

main

main

step, which contains all the rest of the steps as substep. That is - it iterates over the batch of received events one by one and all the rest is the same as before.

This approach works, received event batches are processed quickly and the workflow does not build any queue walls.

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.

RRakhim Actually, despite `9 running` in the screenshot above, no workflows had run in s...

M

m.muiznieks•2/9/25, 1:51 PM

I will upload in a moment the producer/consumer workers in GitHub.

Mm.muiznieks I will upload in a moment the producer/consumer workers in GitHub.

M

m.muiznieks•2/9/25, 3:09 PM

So this is the example I use and it works (my code is **it and I'm proud of it, yes), @Rakhim

This is where you send all the incoming events, NOT directly to the Workflow. https://github.com/m-muiznieks/cloudflare-queue-producer-worker
This is where the event batches are sent to: https://github.com/m-muiznieks/cloudflare-queue-consumer-worker

Bind your queue and workflows in wrangler.json.
At the consumer you will set the batch size and timing. I think 100 and 60 seconds are good enough, but read at the Queues
https://developers.cloudflare.com/queues/ for possible limits.

From the consumer you can trigger directly the Workflow you want, but I believe the consumer can be your current worker, where you manage your workflows.

You will have to adjust your Workflows.

Here is a rough example how I do that:

export class WORKFLOW_WITH_BATCHED_EVENTS extends WorkflowEntrypoint<Env, BatchPayload> {
    async run(event: WorkflowEvent<BatchPayload>, step: WorkflowStep) {
        // Can access bindings on `this.env`
        // Can access params on `event.payload`
        
        const ctx = this.ctx;
        const events = event.payload.data;

        // THIS IS THE MAIN STEP, WHICH CONTAINS THE ACTUAL SUB-STEPS, WHAT I MENTIONED BEFORE. 
        const process_events = await step.do("process_events",
            async () => { 
                
                  /// ITERATE OVER THE EVENTS ONE-BY-ONE
                 for(const key in events){
                    const e = events[key];
                   
                    //SUBSTEPS - Process as usual.
                    const fbcapi = await step.do(`your_substep_name`, 
                        async () => { ... }
                    )
                    //.... the rest of your wf logic
                  }
              //...
          })
}

export class WORKFLOW_WITH_BATCHED_EVENTS extends WorkflowEntrypoint<Env, BatchPayload> {
    async run(event: WorkflowEvent<BatchPayload>, step: WorkflowStep) {
        // Can access bindings on `this.env`
        // Can access params on `event.payload`
        
        const ctx = this.ctx;
        const events = event.payload.data;

        // THIS IS THE MAIN STEP, WHICH CONTAINS THE ACTUAL SUB-STEPS, WHAT I MENTIONED BEFORE. 
        const process_events = await step.do("process_events",
            async () => { 
                
                  /// ITERATE OVER THE EVENTS ONE-BY-ONE
                 for(const key in events){
                    const e = events[key];
                   
                    //SUBSTEPS - Process as usual.
                    const fbcapi = await step.do(`your_substep_name`, 
                        async () => { ... }
                    )
                    //.... the rest of your wf logic
                  }
              //...
          })
}

export class WORKFLOW_WITH_BATCHED_EVENTS extends WorkflowEntrypoint<Env, BatchPayload> {
    async run(event: WorkflowEvent<BatchPayload>, step: WorkflowStep) {
        // Can access bindings on `this.env`
        // Can access params on `event.payload`
        
        const ctx = this.ctx;
        const events = event.payload.data;

        // THIS IS THE MAIN STEP, WHICH CONTAINS THE ACTUAL SUB-STEPS, WHAT I MENTIONED BEFORE. 
        const process_events = await step.do("process_events",
            async () => { 
                
                  /// ITERATE OVER THE EVENTS ONE-BY-ONE
                 for(const key in events){
                    const e = events[key];
                   
                    //SUBSTEPS - Process as usual.
                    const fbcapi = await step.do(`your_substep_name`, 
                        async () => { ... }
                    )
                    //.... the rest of your wf logic
                  }
              //...
          })
}

export class WORKFLOW_WITH_BATCHED_EVENTS extends WorkflowEntrypoint<Env, BatchPayload> {
    async run(event: WorkflowEvent<BatchPayload>, step: WorkflowStep) {
        // Can access bindings on `this.env`
        // Can access params on `event.payload`
        
        const ctx = this.ctx;
        const events = event.payload.data;

        // THIS IS THE MAIN STEP, WHICH CONTAINS THE ACTUAL SUB-STEPS, WHAT I MENTIONED BEFORE. 
        const process_events = await step.do("process_events",
            async () => { 
                
                  /// ITERATE OVER THE EVENTS ONE-BY-ONE
                 for(const key in events){
                    const e = events[key];
                   
                    //SUBSTEPS - Process as usual.
                    const fbcapi = await step.do(`your_substep_name`, 
                        async () => { ... }
                    )
                    //.... the rest of your wf logic
                  }
              //...
          })
}

I hope this will help you or anyone else who faces big queues.

Maybe this case is already described somewhere, sorry for spamming, then.

T

Trojan•2/9/25, 5:10 PM

Is there any risk of this pattern losing state on a shutdown?

interface CycleParams<
  I extends Rpc.Serializable<I> = Rpc.Serializable<any>,
  O extends Rpc.Serializable<O> = Rpc.Serializable<any>,
> {
  name: string;
  do: {
    name: string;
    callback: (data: I) => Promise<O>;
  };
  sleep?: {
    name: string;
    duration: WorkflowSleepDuration;
  };
}

class Cycle<T> extends WorkflowEntrypoint<Env, CycleParams[]> {
  async run(event: WorkflowEvent<CycleParams[]>, step: WorkflowStep) {
    const { instanceId, payload, timestamp } = event;

    const data: any[] = [null];
    for (const [index, cycleStep] of payload.entries()) {
      if (cycleStep.sleep)
        await step.sleep(cycleStep.sleep.name, cycleStep.sleep.duration);
      
      data[index + 1] = await step.do(cycleStep.do.name, () => {
        if (data.length > index + 1) return data[index + 1];
        return cycleStep.do.callback(data[index]);
      });
    }

    return data[data.length - 1] as T;
  }
}

interface CycleParams<
  I extends Rpc.Serializable<I> = Rpc.Serializable<any>,
  O extends Rpc.Serializable<O> = Rpc.Serializable<any>,
> {
  name: string;
  do: {
    name: string;
    callback: (data: I) => Promise<O>;
  };
  sleep?: {
    name: string;
    duration: WorkflowSleepDuration;
  };
}

class Cycle<T> extends WorkflowEntrypoint<Env, CycleParams[]> {
  async run(event: WorkflowEvent<CycleParams[]>, step: WorkflowStep) {
    const { instanceId, payload, timestamp } = event;

    const data: any[] = [null];
    for (const [index, cycleStep] of payload.entries()) {
      if (cycleStep.sleep)
        await step.sleep(cycleStep.sleep.name, cycleStep.sleep.duration);
      
      data[index + 1] = await step.do(cycleStep.do.name, () => {
        if (data.length > index + 1) return data[index + 1];
        return cycleStep.do.callback(data[index]);
      });
    }

    return data[data.length - 1] as T;
  }
}

interface CycleParams<
  I extends Rpc.Serializable<I> = Rpc.Serializable<any>,
  O extends Rpc.Serializable<O> = Rpc.Serializable<any>,
> {
  name: string;
  do: {
    name: string;
    callback: (data: I) => Promise<O>;
  };
  sleep?: {
    name: string;
    duration: WorkflowSleepDuration;
  };
}

class Cycle<T> extends WorkflowEntrypoint<Env, CycleParams[]> {
  async run(event: WorkflowEvent<CycleParams[]>, step: WorkflowStep) {
    const { instanceId, payload, timestamp } = event;

    const data: any[] = [null];
    for (const [index, cycleStep] of payload.entries()) {
      if (cycleStep.sleep)
        await step.sleep(cycleStep.sleep.name, cycleStep.sleep.duration);
      
      data[index + 1] = await step.do(cycleStep.do.name, () => {
        if (data.length > index + 1) return data[index + 1];
        return cycleStep.do.callback(data[index]);
      });
    }

    return data[data.length - 1] as T;
  }
}

interface CycleParams<
  I extends Rpc.Serializable<I> = Rpc.Serializable<any>,
  O extends Rpc.Serializable<O> = Rpc.Serializable<any>,
> {
  name: string;
  do: {
    name: string;
    callback: (data: I) => Promise<O>;
  };
  sleep?: {
    name: string;
    duration: WorkflowSleepDuration;
  };
}

class Cycle<T> extends WorkflowEntrypoint<Env, CycleParams[]> {
  async run(event: WorkflowEvent<CycleParams[]>, step: WorkflowStep) {
    const { instanceId, payload, timestamp } = event;

    const data: any[] = [null];
    for (const [index, cycleStep] of payload.entries()) {
      if (cycleStep.sleep)
        await step.sleep(cycleStep.sleep.name, cycleStep.sleep.duration);
      
      data[index + 1] = await step.do(cycleStep.do.name, () => {
        if (data.length > index + 1) return data[index + 1];
        return cycleStep.do.callback(data[index]);
      });
    }

    return data[data.length - 1] as T;
  }
}

Similar Threads

Similar Threads

Similar Threads