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;
  }
}

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

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

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

9

9 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

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

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

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

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)

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

Similar Threads

Similar Threads

Similar Threads