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.muiznieksOP•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;
  }
}

Nnil Did you find an easy way to see your local store with data in R2 to use with wor...

C

Chris Jayden•2/9/25, 9:49 PM

I ended up testing on production

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

C

Chris Jayden•2/9/25, 9:50 PM

Yeah it’s frustrating when you need workflows + R2 or whatever else cross thing. To be honest, I’m currently rewriting to Inngest, API’s are almost the same. I feel workflows can’tsolve my challenges right now

R

Rakhim•2/10/25, 6:53 AM

hey! thanks for the detailed info!
I've been using queues extensively for about 10 months before switching to workflows. It was working fine, but I had to build a somewhat complex system of inter-dependent queue invocations with custom retry and backoff logic. With workflows, I was able to re-create the logic in a simpler manner and with additional observability. I'd rather not migrate back to queues, but I guess I will have to :/

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

M

Matt Silverlock•2/10/25, 2:19 PM

We’re working on deploying some fixes to this today.

RRakhim hey! thanks for the detailed info! I've been using queues extensively for about ...

S

Seekerdasbatatas•2/10/25, 3:09 PM

hey your account should be getting unstuck atm

it will probably take a bit (since your account is a bit big haha), but I'm monitoring it

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

S

Seekerdasbatatas•2/10/25, 3:09 PM

this should be fixed atm

SSeekerdasbatatas hey your account should be getting unstuck atm 🙂 - it will probably take a bit ...

R

Rakhim•2/10/25, 3:27 PM

Cool! Fingers crossed. So far, the Dashboard still shows

processing.

RRakhim Cool! Fingers crossed. So far, the Dashboard still shows `9` processing.

R

Rakhim•2/10/25, 3:49 PM

But the metrics dashboard already shows 425 new processed workflows!

RRakhim But the metrics dashboard already shows 425 new processed workflows!

S

Seekerdasbatatas•2/10/25, 3:56 PM

yhep, i'll try to push some fixes to make it faster :soontm:

!

!AH!•2/10/25, 8:40 PM

Hey guys i have a problem with the NonRetryableError thing. I have the following code

step.do('Validate params', async () => {
            if (!event.payload.email) {
                throw new NonRetryableError('No email provided');
            }
            if (!event.payload.url) {
                throw new NonRetryableError('No url provided');
            }
            if (!isValidUrl(event.payload.url)) {
                throw new NonRetryableError('Invalid URL');
            }
            if (!isValidEmail(event.payload.email)) {
                throw new NonRetryableError('Invalid email provided');
            }
        });

step.do('Validate params', async () => {
            if (!event.payload.email) {
                throw new NonRetryableError('No email provided');
            }
            if (!event.payload.url) {
                throw new NonRetryableError('No url provided');
            }
            if (!isValidUrl(event.payload.url)) {
                throw new NonRetryableError('Invalid URL');
            }
            if (!isValidEmail(event.payload.email)) {
                throw new NonRetryableError('Invalid email provided');
            }
        });

step.do('Validate params', async () => {
            if (!event.payload.email) {
                throw new NonRetryableError('No email provided');
            }
            if (!event.payload.url) {
                throw new NonRetryableError('No url provided');
            }
            if (!isValidUrl(event.payload.url)) {
                throw new NonRetryableError('Invalid URL');
            }
            if (!isValidEmail(event.payload.email)) {
                throw new NonRetryableError('Invalid email provided');
            }
        });

step.do('Validate params', async () => {
            if (!event.payload.email) {
                throw new NonRetryableError('No email provided');
            }
            if (!event.payload.url) {
                throw new NonRetryableError('No url provided');
            }
            if (!isValidUrl(event.payload.url)) {
                throw new NonRetryableError('Invalid URL');
            }
            if (!isValidEmail(event.payload.email)) {
                throw new NonRetryableError('Invalid email provided');
            }
        });

and everything is fine until the error gets thrown and the workflow doesn't stop there but instead executes all other steps as shown in the screenshot.
Does anyone have the same problem or is it just a skill issue on my end?

!!AH!Hey guys i have a problem with the NonRetryableError thing. I have the following...

D

D Trombett•2/10/25, 9:33 PM

You're not awaiting your step

D

D Trombett•2/10/25, 9:33 PM

Add

await

await

before

step.do

step.do

DD Trombett You're not awaiting your step

!

!AH!•2/11/25, 5:58 AM

Thank you very much this fixed it! I had no idea every step had to be awaited.

J

JonasDoesThings•2/11/25, 7:45 AM

It's not possible to "dynamically spawn" steps in a one-to-many relation, right?

I've been thinking about an use-case, where I want to send out weekly report e-mails.

The most safe way of building this without risking double sends, would (I think) be to have one starting-step in the workflow that queries all users that should receive an report, and then I would want to spawn a further step/task for each user where the personalized report is generated and sent.

This is (currently) not possible with a workflow alone, right? I would need to use a queue instead

R

Rakhim•2/11/25, 8:04 AM

Is there a way to discard the backlog queue on workflows (i.e. simply delete or cancel all the "queued" workflows)?

RRakhim Is there a way to discard the backlog queue on workflows (i.e. simply delete or ...

S

Seekerdasbatatas•2/11/25, 10:12 AM

not natively, but you could use a script for it:

#!/bin/bash

# Exit immediately if a command exits with a non-zero status
set -e

# Check if workflow name and status are provided
if [ -z "$1" ] || [ -z "$2" ]; then
  echo "Usage: $0 <workflow-name> <status>"
  echo "Status can be: queued, running, succeeded, failed"
  exit 1
fi

WORKFLOW_NAME="$1"
STATUS="$2"

while true; do
  # Get workflow instances with specified status
  output=$(npx wrangler workflows instances list "$WORKFLOW_NAME" --status="$STATUS" 2>/dev/null)
  ids=$(echo "$output" | awk -F '│' '{gsub(/ /, "", $2); if ($2 ~ /^[a-f0-9-]+$/) print $2}')
  
  instance_count=$(echo "$ids" | wc -w | tr -d ' ')
  
  if [ "$instance_count" -eq 0 ]; then
    echo "No more instances found with status '$STATUS'"
    break
  fi
  
  echo "Found $instance_count instances with status '$STATUS'"

  # Process in batches
  BATCH_SIZE=5
  echo "$ids" | xargs -n$BATCH_SIZE -P$BATCH_SIZE -I{} sh -c '
    id="$1"
    workflow="$2"
    if ! npx wrangler workflows instances terminate "$workflow" "$id" >/dev/null 2>&1; then
      echo "Failed to terminate instance $id"
      exit 1
    fi
    echo "Terminated $id"
  ' -- {} "$WORKFLOW_NAME"

  echo "Batch complete, checking for remaining instances..."
done

echo "Done terminating all instances"

#!/bin/bash

# Exit immediately if a command exits with a non-zero status
set -e

# Check if workflow name and status are provided
if [ -z "$1" ] || [ -z "$2" ]; then
  echo "Usage: $0 <workflow-name> <status>"
  echo "Status can be: queued, running, succeeded, failed"
  exit 1
fi

WORKFLOW_NAME="$1"
STATUS="$2"

while true; do
  # Get workflow instances with specified status
  output=$(npx wrangler workflows instances list "$WORKFLOW_NAME" --status="$STATUS" 2>/dev/null)
  ids=$(echo "$output" | awk -F '│' '{gsub(/ /, "", $2); if ($2 ~ /^[a-f0-9-]+$/) print $2}')
  
  instance_count=$(echo "$ids" | wc -w | tr -d ' ')
  
  if [ "$instance_count" -eq 0 ]; then
    echo "No more instances found with status '$STATUS'"
    break
  fi
  
  echo "Found $instance_count instances with status '$STATUS'"

  # Process in batches
  BATCH_SIZE=5
  echo "$ids" | xargs -n$BATCH_SIZE -P$BATCH_SIZE -I{} sh -c '
    id="$1"
    workflow="$2"
    if ! npx wrangler workflows instances terminate "$workflow" "$id" >/dev/null 2>&1; then
      echo "Failed to terminate instance $id"
      exit 1
    fi
    echo "Terminated $id"
  ' -- {} "$WORKFLOW_NAME"

  echo "Batch complete, checking for remaining instances..."
done

echo "Done terminating all instances"

#!/bin/bash

# Exit immediately if a command exits with a non-zero status
set -e

# Check if workflow name and status are provided
if [ -z "$1" ] || [ -z "$2" ]; then
  echo "Usage: $0 <workflow-name> <status>"
  echo "Status can be: queued, running, succeeded, failed"
  exit 1
fi

WORKFLOW_NAME="$1"
STATUS="$2"

while true; do
  # Get workflow instances with specified status
  output=$(npx wrangler workflows instances list "$WORKFLOW_NAME" --status="$STATUS" 2>/dev/null)
  ids=$(echo "$output" | awk -F '│' '{gsub(/ /, "", $2); if ($2 ~ /^[a-f0-9-]+$/) print $2}')
  
  instance_count=$(echo "$ids" | wc -w | tr -d ' ')
  
  if [ "$instance_count" -eq 0 ]; then
    echo "No more instances found with status '$STATUS'"
    break
  fi
  
  echo "Found $instance_count instances with status '$STATUS'"

  # Process in batches
  BATCH_SIZE=5
  echo "$ids" | xargs -n$BATCH_SIZE -P$BATCH_SIZE -I{} sh -c '
    id="$1"
    workflow="$2"
    if ! npx wrangler workflows instances terminate "$workflow" "$id" >/dev/null 2>&1; then
      echo "Failed to terminate instance $id"
      exit 1
    fi
    echo "Terminated $id"
  ' -- {} "$WORKFLOW_NAME"

  echo "Batch complete, checking for remaining instances..."
done

echo "Done terminating all instances"

#!/bin/bash

# Exit immediately if a command exits with a non-zero status
set -e

# Check if workflow name and status are provided
if [ -z "$1" ] || [ -z "$2" ]; then
  echo "Usage: $0 <workflow-name> <status>"
  echo "Status can be: queued, running, succeeded, failed"
  exit 1
fi

WORKFLOW_NAME="$1"
STATUS="$2"

while true; do
  # Get workflow instances with specified status
  output=$(npx wrangler workflows instances list "$WORKFLOW_NAME" --status="$STATUS" 2>/dev/null)
  ids=$(echo "$output" | awk -F '│' '{gsub(/ /, "", $2); if ($2 ~ /^[a-f0-9-]+$/) print $2}')
  
  instance_count=$(echo "$ids" | wc -w | tr -d ' ')
  
  if [ "$instance_count" -eq 0 ]; then
    echo "No more instances found with status '$STATUS'"
    break
  fi
  
  echo "Found $instance_count instances with status '$STATUS'"

  # Process in batches
  BATCH_SIZE=5
  echo "$ids" | xargs -n$BATCH_SIZE -P$BATCH_SIZE -I{} sh -c '
    id="$1"
    workflow="$2"
    if ! npx wrangler workflows instances terminate "$workflow" "$id" >/dev/null 2>&1; then
      echo "Failed to terminate instance $id"
      exit 1
    fi
    echo "Terminated $id"
  ' -- {} "$WORKFLOW_NAME"

  echo "Batch complete, checking for remaining instances..."
done

echo "Done terminating all instances"

JJonasDoesThings It's not possible to "dynamically spawn" steps in a one-to-many relation, right?...

A

adeola13.•2/11/25, 3:59 PM

I would if I were you store the emails you need to send in a KV or any other temporary storage from cloudflare, do that in a seperate step, and then in the next step first check if you've already sent it to that email, then send the email, then mark it as send in storage all in the same step. This makes it idempodent. Then in the final step do a clean up for the storage.

SSeekerdasbatatas not natively, but you could use a script for it: ```bash #!/bin/bash # Exit imm...

R

Rakhim•2/11/25, 4:18 PM

woah, thanks a lot!

F

Focux•2/11/25, 7:38 PM

What's the best way to implement "wait until signal" on userland? sleeping and pulling?

A

Ashley•2/11/25, 7:56 PM

Do you mean the user (browser) polling to check if the Workflow has finished, or the Workflow waiting for “something” to happen?

F

Focux•2/11/25, 8:10 PM

I mean something like this:

while (true) {
  const isDone = await checkIfSomethingIsDone()
  if (isDone) break;
  step.sleep('wait for event', '5 minutes');
}

while (true) {
  const isDone = await checkIfSomethingIsDone()
  if (isDone) break;
  step.sleep('wait for event', '5 minutes');
}

while (true) {
  const isDone = await checkIfSomethingIsDone()
  if (isDone) break;
  step.sleep('wait for event', '5 minutes');
}

while (true) {
  const isDone = await checkIfSomethingIsDone()
  if (isDone) break;
  step.sleep('wait for event', '5 minutes');
}

F

Focux•2/11/25, 8:11 PM

Figuring out the right interval would be tricky in my case. I have a workflow that has several steps, and one of those is sending an email. I want to wait for that email to be opened before continuing with the workflow execution.

A

Ashley•2/11/25, 8:39 PM

Yeah, there’s no native way to do this yet (maybe it will come later - it seems a no brainer to me) so I think a repeated sleep is the only way

F

Focux•2/11/25, 9:25 PM

Yeah, I think I read on Twitter that it's on the roadmap

F

Focux•2/11/25, 9:27 PM

I'm wondering if it's possible to

pause

pause

the workflow from within a workflow instance, like calling the workflow binding and then resuming the workflow from a webhook or something like that

A

Ashley•2/11/25, 10:06 PM

You could as the workflow ID is available inside the Workflow class, you’d then call resume I believe on it (using Workflow binding for example)

F

Focux•2/11/25, 10:54 PM

yeah, that's what i was thinking, i'll try that and see how it goes

币

币成•2/12/25, 3:16 AM

我用谷歌帐号登录的，验证邮箱时收不到验证码如何处理？

A

Ashley•2/12/25, 11:26 AM

For sleeping workflows, I think I saw that they don’t contribute to the concurrent count, but is there a “minimum” sleep before this is true? (e.g. if a Workflow sleeps for 10 seconds, will it still not be counted in the concurrent count - asking as I thought I saw something that suggested this was the case)

AAshley For sleeping workflows, I think I saw that they don’t contribute to the concurre...

S

Seekerdasbatatas•2/12/25, 2:08 PM

at the moment yes, but that's an implementation detail of Workflows so the "minimum" sleep might (and probably will) change over time

SSeekerdasbatatas at the moment yes, but that's an implementation detail of Workflows so the "mini...

A

Ashley•2/12/25, 2:11 PM

Is there a rough guide? Just trying to consider whether hacking around a “wait for event” is worthwhile as I wouldn’t want to have say a 5 second sleep that is effectively polling for something happening and it eat into my concurrency

AAshley Is there a rough guide? Just trying to consider whether hacking around a “wait f...

S

Seekerdasbatatas•2/12/25, 2:14 PM

the minimum sleep required atm to pass it to waiting is in the minutes range (and the transition happens after that time has passed) - so I don't know if it's worthwhile hacking around that. I've personally seen people pausing and resuming an instance to "emulate" events, but on a lighter note, we're working on native events :soontm:

A

Ashley•2/12/25, 2:22 PM

The use case is let’s say it takes 10-30 seconds for something to happen, I don’t want to sleep for say 45 (to be safe) or “poll” at 10 seconds or it would eat concurrency. I could set it to 5 mins, let’s say, but then there’s an unneccessary gap. I have a little solution using a DO (pause workflow -> call DO that sets alarm that effectively does the polling -> DO resumes workflow when event happens) and it works nicely until native events, but wondered if it was worthwhile (but if it’s minutes, I think so for now)

AAshley The use case is let’s say it takes 10-30 seconds for something to happen, I don’...

S

Seekerdasbatatas•2/12/25, 2:25 PM

Considering that you probably don't have stuff running parallel to the wait for event call, that seems to work nicely

A

Ashley•2/12/25, 2:28 PM

Yeah, what would have been super cool was if I could define the callback (well, the logic to determine if to continue) in the Workflow and pass it to the DO, but alas the “connection” doesn’t naturally survive hibernation

A

Ashley•2/12/25, 3:48 PM

https://github.com/apeacock1991/workflows-wait-for-action/ this was surprisingly easy (I wrote it at the weekend, hadn't had the time to check if it was useful until just then) for anyone that wants to pause/resume workflow using a DO and "protect" your concurrency limit

GitHub

GitHub - apeacock1991/workflows-wait-for-action

Contribute to apeacock1991/workflows-wait-for-action development by creating an account on GitHub.

W

wesbos•2/12/25, 6:15 PM

Hi folks - having some trouble with throwing errors inside a workflow.

First, it seems throwing a NonRetryableError, is retried as many times as specified

Second, I can't seem to find out where the throw error messages are being returned in the workflow status. The docs seem to say

const wf = await MY_WORKFLOW.get(id)

const wf = await MY_WORKFLOW.get(id)

and then

await wf.status()

await wf.status()

should return errors, but they are always empty.

here is the code I'm using to reproduce:

export class MyWorkflow extends WorkflowEntrypoint<Env, Params> {
    async run(event: WorkflowEvent<Params>, step: WorkflowStep) {
        step.do('Step 1', async () => { return 'Step 1 finished!';  });

        step.do('Step 2', {
            retries: { limit: 2, delay: '1 second', backoff: 'linear', },
        }, async () => {
            console.log('Trying Step 2');
            throw new NonRetryableError('Step 3 failed. Non-retryable');
        });
    }
}

export class MyWorkflow extends WorkflowEntrypoint<Env, Params> {
    async run(event: WorkflowEvent<Params>, step: WorkflowStep) {
        step.do('Step 1', async () => { return 'Step 1 finished!';  });

        step.do('Step 2', {
            retries: { limit: 2, delay: '1 second', backoff: 'linear', },
        }, async () => {
            console.log('Trying Step 2');
            throw new NonRetryableError('Step 3 failed. Non-retryable');
        });
    }
}

export class MyWorkflow extends WorkflowEntrypoint<Env, Params> {
    async run(event: WorkflowEvent<Params>, step: WorkflowStep) {
        step.do('Step 1', async () => { return 'Step 1 finished!';  });

        step.do('Step 2', {
            retries: { limit: 2, delay: '1 second', backoff: 'linear', },
        }, async () => {
            console.log('Trying Step 2');
            throw new NonRetryableError('Step 3 failed. Non-retryable');
        });
    }
}

export class MyWorkflow extends WorkflowEntrypoint<Env, Params> {
    async run(event: WorkflowEvent<Params>, step: WorkflowStep) {
        step.do('Step 1', async () => { return 'Step 1 finished!';  });

        step.do('Step 2', {
            retries: { limit: 2, delay: '1 second', backoff: 'linear', },
        }, async () => {
            console.log('Trying Step 2');
            throw new NonRetryableError('Step 3 failed. Non-retryable');
        });
    }
}

This logs "trying step 2" three times

Aadeola13.I would if I were you store the emails you need to send in a KV or any other tem...

J

JonasDoesThings•2/12/25, 6:59 PM

Yeah, that would def. work-out, but is not that elegant / simple again

Sticking with using upstash's queueing service for now, where I have 1 scheduled task that calls my api/tasks/generate-reports endpoint every day, which then queries my database and batch-enqueues one task for each report I want to generate & send out

N

Nathan Scheele•2/13/25, 10:41 PM

Hi all! I'm running into the following error when using Workflows:

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.

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.

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.

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.

I see that

step.do

step.do

expects to return

Rpc.Serializable<unknown>

Rpc.Serializable<unknown>

, so I did a little bit of reading about the RPC lifecycle here and found that in the context of Worker Service Bindings, you should either use the experimental

using

using

syntax, or explicitly call

object[Symbol.dispose]()

object[Symbol.dispose]()

.

I'm a little confused because this error is only thrown after some calls to

step.do

step.do

, and not others. If someone could provide some insight as to what's going on, that would be great. Happy to provide more details also

Cloudflare Docs

Workers RPC — Lifecycle · Cloudflare Workers docs

Memory management, resource management, and the lifecycle of RPC stubs.

T

topmass•2/15/25, 4:17 PM

can anyone help me understand the cost difference of a workflow, while the workflow is running does this add all cpu time during the total run time or just during events that trigger CPU time effectively making the cost that same as using wokers native or roughly the same

I'm curious for scale if I implemented in two ways

Option 1. I send post req with a worker webhook url and then when the job is completed webhook is hit generating new worker instance to handle post job completion duties

Option 2. I create a workflow that for a particular "generation" or request tracks/handles everything, initial http post req as well as waiting for job completion

thank you in advance if anyone can help!

A

Ashley•2/15/25, 8:01 PM

https://blog.cloudflare.com/building-workflows-durable-execution-on-workers/#so-how-much-does-it-cost it’s exactly the same as Workers pricing

The Cloudflare Blog

Build durable applications on Cloudflare Workers: you write the Wor...

Cloudflare Workflows is now in open beta! Workflows allows you to build reliable, repeatable, long-lived multi-step applications that can automatically retry, persist state, and scale out. Read on to learn how Workflows works, how we built it on top of Durable Objects, and how you can deploy your first Workflows application.

A

Ashley•2/16/25, 9:53 PM

If I’m understanding correctly, that’s how Workflows are designed, you can just start a Workflow per file (as an example) and they will run in parallel (up to the max concurrency limit, which is current 100 Workflows at any one time - will likely be raised during beta)

F

Focux•2/17/25, 3:50 PM

When using

sleep

sleep

locally with a number value, it treats it like ms instead of seconds

F

Focux•2/17/25, 3:51 PM

I'm not sure if that's a local development issue or if the documentation is outdated, but it says in the doc that when passing a number value to

sleep

sleep

it's seconds not ms

FFocux I'm not sure if that's a local development issue or if the documentation is outd...

D

D Trombett•2/17/25, 3:58 PM

docs outdated (ref. https://canary.discord.com/channels/595317990191398933/1296923707792560189/1336055153945739456)

F

Focux•2/17/25, 4:01 PM

good to know, thank you

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

Similar Threads

Similar Threads

Similar Threads