I'm unsure of the best way to implement a fan-out pattern using Workflows. I fetch a list of files,

1

1984 Ford Laser•11/19/24, 2:14 AM

I've only just started looking at Workflows, I was wondering if it would be suitable for a process where one step requires collecting data from dozens of endpoints (DOs in this case) and waiting until all are returned before going to the next step. Would this be possible?

Mmartinhoest I'm unsure of the best way to implement a fan-out pattern using Workflows. I fet...

M

makrdev•11/19/24, 7:14 AM

You can do both actually and both have different trade-offs. There is a hard limit (256) for maximum number of steps in a single run. So if the list you want to process is bigger than it you can’t. There is no hard limit for workflows (you can run 100 workflows concurrently tho) however you won’t be able to share state between them without external storage (kv etc.)

11984 Ford Laser I've only just started looking at Workflows, I was wondering if it would be suit...

M

makrdev•11/19/24, 7:18 AM

Yes, you can do. It’s recommended to make a single fetch call in a step in order not to lose its retry logic and persistency.

Mmakrdev You can do both actually and both have different trade-offs. There is a hard lim...

M

martinhoestOP•11/19/24, 7:26 AM

thanks, didn't immediately see this hard limit in the docs - but it is there, I just missed it. I have a few thousand, so need to trigger a workflow for each.

Mmartinhoest I'm unsure of the best way to implement a fan-out pattern using Workflows. I fet...

P

pao ramen•11/19/24, 10:35 AM

unluckily, workflows have limited amount of steps. So only do step.do for each file if you are 100% sure you are not going to have more than 256 files. Regarding triggering a new process for each, it can be done but it does not support batching nor concurrency. What I do to workaround this is one step, and inside I chunk and send batches to a queue. That queue then triggers the workflows.

Also, since having multiple workflows locally does not work, I recommend having a single workflow that dispatches work.

P

pao ramen•11/19/24, 10:40 AM

This is how I connected the pieces myself. Basically a queue endpoint that will forward events to a workflow. The workflow then dispatches with a given id. It's based on convention (each workflow has an id, and the queue name is that id + '-queue') and it allows me to have some type safety. I then pass an object around, so I can compose workflows as you expect.

MMatt Silverlock Most of these (bar sharing state across Workflows) will likely land in the next ...

E

Evan•11/19/24, 3:55 PM

Any comment on whether or not workflows will ever support workers for platforms?

S

sdr•11/19/24, 7:33 PM

Bug in the workflow list versions docs, result should be a collection
https://developers.cloudflare.com/api/operations/wor-list-workflow-versions

Cloudflare API Documentation

Interact with Cloudflare's products and services via the Cloudflare API

Ggotfredsen Question . When should one use Queues and when should on use Workflows ?

C

cloudkite•11/19/24, 8:18 PM

Curious about this as well. Seems a bit of overlap between queues and workflows. Would be nice if there was a guide which outlined reasons to use either

O

oof2win2•11/20/24, 2:31 AM

so i have a workflow where i have an input pdf and i need to run a claude call. the intermediate step however is to transcribe that pdf via a poll-based service. my question therefore is how do i add sleeps into the polling part - the execution should be under 10-15s, so should i just make it a setTimeout function that is awaited?

O

oof2win2•11/20/24, 2:32 AM

like i could probably split into two workflow steps - generate the poll URL and then poll to get the transcript

O

oof2win2•11/20/24, 2:32 AM

but how should i do the sleeps inside the polling

O

oof2win2•11/20/24, 3:18 AM

actually nvm polling it's a promise.all -> pdfs.map -> polling inside bc it allows for caching of the individual pdf steps

Ccloudkite Curious about this as well. Seems a bit of overlap between queues and workflows....

Q

Quinn•11/20/24, 4:34 AM

The main use case difference I see is that queues seem preferable when I have a lot of one task that I need to do for many objects (e.g., I need to send 1000s of notifications but need to respect the rate limits of a provider, I can queue them all up and then chug along and delay as needed.) Workflows seem to be suited for when you have many different tasks that you need to do for one object (e.g., A user subscribed to my PaaS app, so I need to (1) charge their credit card, (2) spin up some infrastructure, (3) send them a welcome email, and so on while appropriately handling errors/retries at every step). You might be using both together if, for instance, your PaaS app has many users signing up; you might have a queue that triggers a workflow. This is all just based off of my own thoughts as I've read through the docs though so I could be way off

T

telliks•11/20/24, 7:25 AM

Pretty sure I'm missing something obvious. I've been struggling to understand how the first workflow is created. For other primitives like KV, you first have to create them via wrangler or the dashboard before you can reference their binding in the

wrangler.toml

wrangler.toml

file. I'm not seeing the wrangler command to create new workflows. When I add the bindings and deploy it out, it is just undefined. I'm running the latest version of wrangler with a current compatibility date. What am I missing?

Ttelliks Pretty sure I'm missing something obvious. I've been struggling to understand ho...

S

skip•11/20/24, 8:27 AM

https://duckduckgo.com/?q=bowie+all+dudes&t=osx

In this example, you can call one programatically await env.MY_WORKFLOW.create() or hit the URL

And there is also trigger from wrangler CLI https://developers.cloudflare.com/workers/wrangler/commands/#trigger

I'm not sure how to automatically trigger based on queues, though

bowie all dudes at DuckDuckGo

DuckDuckGo. Privacy, Simplified.

Sskip https://duckduckgo.com/?q=bowie+all+dudes&t=osx In this example, you can call o...

T

telliks•11/20/24, 8:41 AM

Back one step. I'm not getting to the point of creating a workflow instance. I'm having problem creating the workflow binding itself. Using your example

await env.MY_WORKFLOW.create()

await env.MY_WORKFLOW.create()

,

env.MY_WORKFLOW

env.MY_WORKFLOW

itself is

undefined

undefined

. Do you need to do anything more than simply adding the

[[workflows]]

[[workflows]]

config in the wrangler.toml? And export the relevant workflow class from the main entry point?

P

pao ramen•11/20/24, 9:00 AM

More feedback on workflows, this time, the UI: the list takes more than 15s to load.

P

pao ramen•11/20/24, 9:01 AM

and one question: how do we track errors that happen within a step? It seems like they are swallowed. You can see the error on the workflow through the UI, but how would you send them to sentry?

N

Nick•11/20/24, 9:38 AM

I would like to see webhook support integrated into a workflow, allowing it to wait for one or more webhooks to be fulfilled before proceeding to the next step.

Ttelliks Back one step. I'm not getting to the point of creating a workflow instance. I'm...

D

Diogo Ferreira•11/20/24, 10:50 AM

The wrangler.toml config is enough for wrangler to create the bindings when the Workflow get deployed. Can you ensure that you are setting the bindings to env?

type Env = {
    // Add your bindings here, e.g. Workers KV, D1, Workers AI, etc.
    MY_WORKFLOW: Workflow;
};

type Env = {
    // Add your bindings here, e.g. Workers KV, D1, Workers AI, etc.
    MY_WORKFLOW: Workflow;
};

type Env = {
    // Add your bindings here, e.g. Workers KV, D1, Workers AI, etc.
    MY_WORKFLOW: Workflow;
};

type Env = {
    // Add your bindings here, e.g. Workers KV, D1, Workers AI, etc.
    MY_WORKFLOW: Workflow;
};

DDiogo Ferreira The wrangler.toml config is enough for wrangler to create the bindings when the ...

T

telliks•11/20/24, 11:18 AM

It looks correct to me. Could it be that I'm trying to access it from an email worker?

Ttelliks It looks correct to me. Could it be that I'm trying to access it from an email w...

M

Matt Silverlock•11/20/24, 11:25 AM

Can you share your code? Did you deploy the Workflow?

Workflows don’t need to be created separately: deploying a Workflow registers it. This section explains it: https://developers.cloudflare.com/workflows/get-started/guide/#5-deploy-your-workflow

Cloudflare Docs

Guide | Cloudflare Workflows docs

Workflows allow you to build durable, multi-step applications using the Workers platform. A Workflow can automatically retry, persist state, run for hours or days, and coordinate between third-party APIs.

P

pao ramen•11/20/24, 11:42 AM

Sometimes I get this error: "A call to blockConcurrencyWhile() in a Durable Object waited for too long. The call was canceled and the Durable Object was reset."

P

pao ramen•11/20/24, 11:43 AM

and also "Error: Durable Object exceeded its CPU time limit and was reset."

MMatt Silverlock Can you share your code? Did you deploy the Workflow? Workflows don’t need to ...

T

telliks•11/20/24, 12:06 PM

Sure, I can share some pieces that I reckon is relevant. Appreciate your help. I did deploy the worker, yes.

wrangler.toml

wrangler.toml

[[workflows]]
name = "process-incoming-emails"
binding = "PROCESS_INCOMING_EMAILS_WORKFLOW"
class_name = "ProcessIncomingEmailsWorkflow"

[[workflows]]
name = "process-incoming-emails"
binding = "PROCESS_INCOMING_EMAILS_WORKFLOW"
class_name = "ProcessIncomingEmailsWorkflow"

[[workflows]]
name = "process-incoming-emails"
binding = "PROCESS_INCOMING_EMAILS_WORKFLOW"
class_name = "ProcessIncomingEmailsWorkflow"

[[workflows]]
name = "process-incoming-emails"
binding = "PROCESS_INCOMING_EMAILS_WORKFLOW"
class_name = "ProcessIncomingEmailsWorkflow"

types.ts

types.ts

import type { Workflow } from '@cloudflare/workers-types';

export type Bindings = {
  // ...other bindings
  PROCESS_INCOMING_EMAILS_WORKFLOW: Workflow;
};

import type { Workflow } from '@cloudflare/workers-types';

export type Bindings = {
  // ...other bindings
  PROCESS_INCOMING_EMAILS_WORKFLOW: Workflow;
};

import type { Workflow } from '@cloudflare/workers-types';

export type Bindings = {
  // ...other bindings
  PROCESS_INCOMING_EMAILS_WORKFLOW: Workflow;
};

import type { Workflow } from '@cloudflare/workers-types';

export type Bindings = {
  // ...other bindings
  PROCESS_INCOMING_EMAILS_WORKFLOW: Workflow;
};

src/index.ts

src/index.ts

import { app } from './app'; // Hono app
import type { Bindings } from './types';
import { handleForwardingIncomingEmails } from './incoming-emails/forward-email-handler';

export { ProcessIncomingEmailsWorkflow } from './incoming-emails/workflow';

const handler = {
  fetch: app.fetch,
  async email(message, env, ctx) {
    await handleForwardingIncomingEmails(message, env, ctx);
  },
} satisfies ExportedHandler<Bindings>;

import { app } from './app'; // Hono app
import type { Bindings } from './types';
import { handleForwardingIncomingEmails } from './incoming-emails/forward-email-handler';

export { ProcessIncomingEmailsWorkflow } from './incoming-emails/workflow';

const handler = {
  fetch: app.fetch,
  async email(message, env, ctx) {
    await handleForwardingIncomingEmails(message, env, ctx);
  },
} satisfies ExportedHandler<Bindings>;

import { app } from './app'; // Hono app
import type { Bindings } from './types';
import { handleForwardingIncomingEmails } from './incoming-emails/forward-email-handler';

export { ProcessIncomingEmailsWorkflow } from './incoming-emails/workflow';

const handler = {
  fetch: app.fetch,
  async email(message, env, ctx) {
    await handleForwardingIncomingEmails(message, env, ctx);
  },
} satisfies ExportedHandler<Bindings>;

import { app } from './app'; // Hono app
import type { Bindings } from './types';
import { handleForwardingIncomingEmails } from './incoming-emails/forward-email-handler';

export { ProcessIncomingEmailsWorkflow } from './incoming-emails/workflow';

const handler = {
  fetch: app.fetch,
  async email(message, env, ctx) {
    await handleForwardingIncomingEmails(message, env, ctx);
  },
} satisfies ExportedHandler<Bindings>;

./incoming-emails/forward-email-handler

./incoming-emails/forward-email-handler

export const handleForwardingIncomingEmails = async (
  message: ForwardableEmailMessage,
  env: Bindings,
  _ctx: ExecutionContext,
) => {
  return withLogTags(
    {
      tags: {
        handler: 'email',
        requestId: crypto.randomUUID(),
      },
    },
    async () => {
      const [address] = addressParser(message.to);

      if (!address?.address) {
        logger.warn('No email address found in `to` field');
        return message.setReject('No email address found in `to` field');
      }

      logger.log('Workflow defined', {
        defined: !!env.PROCESS_INCOMING_EMAILS_WORKFLOW, // <-- Not defined here!
      });

      // await env.PROCESS_INCOMING_EMAILS_WORKFLOW.create({
      //   params: {
      //     message,
      //   },
      // });
    }
  )
}

export const handleForwardingIncomingEmails = async (
  message: ForwardableEmailMessage,
  env: Bindings,
  _ctx: ExecutionContext,
) => {
  return withLogTags(
    {
      tags: {
        handler: 'email',
        requestId: crypto.randomUUID(),
      },
    },
    async () => {
      const [address] = addressParser(message.to);

      if (!address?.address) {
        logger.warn('No email address found in `to` field');
        return message.setReject('No email address found in `to` field');
      }

      logger.log('Workflow defined', {
        defined: !!env.PROCESS_INCOMING_EMAILS_WORKFLOW, // <-- Not defined here!
      });

      // await env.PROCESS_INCOMING_EMAILS_WORKFLOW.create({
      //   params: {
      //     message,
      //   },
      // });
    }
  )
}

export const handleForwardingIncomingEmails = async (
  message: ForwardableEmailMessage,
  env: Bindings,
  _ctx: ExecutionContext,
) => {
  return withLogTags(
    {
      tags: {
        handler: 'email',
        requestId: crypto.randomUUID(),
      },
    },
    async () => {
      const [address] = addressParser(message.to);

      if (!address?.address) {
        logger.warn('No email address found in `to` field');
        return message.setReject('No email address found in `to` field');
      }

      logger.log('Workflow defined', {
        defined: !!env.PROCESS_INCOMING_EMAILS_WORKFLOW, // <-- Not defined here!
      });

      // await env.PROCESS_INCOMING_EMAILS_WORKFLOW.create({
      //   params: {
      //     message,
      //   },
      // });
    }
  )
}

export const handleForwardingIncomingEmails = async (
  message: ForwardableEmailMessage,
  env: Bindings,
  _ctx: ExecutionContext,
) => {
  return withLogTags(
    {
      tags: {
        handler: 'email',
        requestId: crypto.randomUUID(),
      },
    },
    async () => {
      const [address] = addressParser(message.to);

      if (!address?.address) {
        logger.warn('No email address found in `to` field');
        return message.setReject('No email address found in `to` field');
      }

      logger.log('Workflow defined', {
        defined: !!env.PROCESS_INCOMING_EMAILS_WORKFLOW, // <-- Not defined here!
      });

      // await env.PROCESS_INCOMING_EMAILS_WORKFLOW.create({
      //   params: {
      //     message,
      //   },
      // });
    }
  )
}

./incoming-emails/workflow

./incoming-emails/workflow

export class ProcessIncomingEmailsWorkflow extends WorkflowEntrypoint<
  Bindings,
  Params
> {}

export class ProcessIncomingEmailsWorkflow extends WorkflowEntrypoint<
  Bindings,
  Params
> {}

export class ProcessIncomingEmailsWorkflow extends WorkflowEntrypoint<
  Bindings,
  Params
> {}

export class ProcessIncomingEmailsWorkflow extends WorkflowEntrypoint<
  Bindings,
  Params
> {}

Ttelliks Sure, I can share some pieces that I reckon is relevant. Appreciate your help. I...

M

Matt Silverlock•11/20/24, 12:52 PM

And what does

npx wrangler workflows list

npx wrangler workflows list

show? Do your Workflow names match?

Do you pass your

Bindings

Bindings

type when you’re instantiating your Hono app instances?

I don’t see your app definition.

T

telliks•11/20/24, 2:12 PM

It comes back with

There are no deployed Workflows in this account

There are no deployed Workflows in this account

. This is why I thought that you need something more than just the config in

wrangler.toml

wrangler.toml

for the deployment to work.

As far as I can tell my workflow names match:

[[workflows]].binding

[[workflows]].binding

matches what I use for

env.PROCESS_INCOMING_EMAILS_WORKFLOW

env.PROCESS_INCOMING_EMAILS_WORKFLOW

as well as what I set on the

Bindings

Bindings

type. And

[[workflows]].class_name

[[workflows]].class_name

matches the name of my workflows class export,

ProcessIncomingEmailsWorkflow

ProcessIncomingEmailsWorkflow

.

Can

[[workflows]].name

[[workflows]].name

be any arbitrary name or what should that match?

I'm not sure how the types would affect the runtime bindings and this worker app has many other bindings (KV, queues, AI, etc) which all work.

Here is the relevant code that instantiates the Hono app (it uses the factory setup):

app.ts

app.ts

import { createFactory } from 'hono/factory';

import { Env } from './types';

export const factory = createFactory<Env>({
  initApp: (app) => {
    app.use(async (c, next) => {
      // Make supabase client available to all requests
      const supabase = createClient(c.env.SUPABASE_URL, c.env.SUPABASE_KEY);
      c.set('supabase', supabase);

      await next();
    });
  },
});

export const app = factory.createApp();

import { createFactory } from 'hono/factory';

import { Env } from './types';

export const factory = createFactory<Env>({
  initApp: (app) => {
    app.use(async (c, next) => {
      // Make supabase client available to all requests
      const supabase = createClient(c.env.SUPABASE_URL, c.env.SUPABASE_KEY);
      c.set('supabase', supabase);

      await next();
    });
  },
});

export const app = factory.createApp();

import { createFactory } from 'hono/factory';

import { Env } from './types';

export const factory = createFactory<Env>({
  initApp: (app) => {
    app.use(async (c, next) => {
      // Make supabase client available to all requests
      const supabase = createClient(c.env.SUPABASE_URL, c.env.SUPABASE_KEY);
      c.set('supabase', supabase);

      await next();
    });
  },
});

export const app = factory.createApp();

import { createFactory } from 'hono/factory';

import { Env } from './types';

export const factory = createFactory<Env>({
  initApp: (app) => {
    app.use(async (c, next) => {
      // Make supabase client available to all requests
      const supabase = createClient(c.env.SUPABASE_URL, c.env.SUPABASE_KEY);
      c.set('supabase', supabase);

      await next();
    });
  },
});

export const app = factory.createApp();

types.ts

types.ts

import type { Workflow } from '@cloudflare/workers-types';

export type Bindings = {
  // ...other bindings
  PROCESS_INCOMING_EMAILS_WORKFLOW: Workflow;
};

export type Env = {
  Bindings: Bindings;
  Variables: Variables;
};

import type { Workflow } from '@cloudflare/workers-types';

export type Bindings = {
  // ...other bindings
  PROCESS_INCOMING_EMAILS_WORKFLOW: Workflow;
};

export type Env = {
  Bindings: Bindings;
  Variables: Variables;
};

import type { Workflow } from '@cloudflare/workers-types';

export type Bindings = {
  // ...other bindings
  PROCESS_INCOMING_EMAILS_WORKFLOW: Workflow;
};

export type Env = {
  Bindings: Bindings;
  Variables: Variables;
};

import type { Workflow } from '@cloudflare/workers-types';

export type Bindings = {
  // ...other bindings
  PROCESS_INCOMING_EMAILS_WORKFLOW: Workflow;
};

export type Env = {
  Bindings: Bindings;
  Variables: Variables;
};

Do you think the factory setup could be problematic?

Ttelliks It comes back with `There are no deployed Workflows in this account`. This is wh...

M

Matt Silverlock•11/20/24, 5:19 PM

If the Workflows API (incl. wrangler) doesn't report any deployed Workflows, then you need to fix that first. Deploying (registering) the Workflow is needed to bind to it.

Nothing else matters until you have a registered Workflow.

Ttelliks It comes back with `There are no deployed Workflows in this account`. This is wh...

M

Matt Silverlock•11/20/24, 5:20 PM

Can [[workflows]].name be any arbitrary name or what should that match?

It needs to match the name of the Workflow! Per the guide:

MMatt Silverlock > Can [[workflows]].name be any arbitrary name or what should that match? It ne...

T

telliks•11/20/24, 5:49 PM

What does this mean? Is the workflow supposed to have the same name as the worker? What happens then if you want multiple workflows per worker?

Ttelliks What does this mean? Is the workflow supposed to have the same name as the worke...

M

Matt Silverlock•11/20/24, 6:20 PM

The class name distinguishes Workflows within a script - e.g.

Ttelliks What does this mean? Is the workflow supposed to have the same name as the worke...

M

Matt Silverlock•11/20/24, 6:22 PM

Again, if

workflows list

workflows list

shows nothing, you haven't registered your Workflow yet by deploying it.

MMatt Silverlock Again, if `workflows list` shows nothing, you haven't registered your Workflow y...

T

telliks•11/20/24, 6:30 PM

I understand that my workflow is not deployed, I just don’t understand what I am supposed to do extra apart from the config I showed and then deploying the whole worker with “wrangler deploy”. From my understanding, having the config is enough and then wrangler should pick up that it should create the configured workflow. This is what my whole question is about. It is not deploying my configured workflow when I deploy the worker, even though the deployment runs successfully.

Original message was deleted

T

telliks•11/20/24, 7:05 PM

Sure, here is the whole file with some redactions.

#:schema node_modules/wrangler/config-schema.json
name = "my-app"
compatibility_date = "2024-10-22"
compatibility_flags = ["nodejs_compat"]
main = "dist/index.js"
minify = true

upload_source_maps = true

tail_consumers = [{ service = "my-app-tail" }]

[build]
command = "pnpm run build"

[observability]
enabled = true

[ai]
binding = "AI"

[[kv_namespaces]]
binding = "INCOMING_EMAILS_KV"
id = "<redacted>"

[[kv_namespaces]]
binding = "DLQ_KV"
id = "<redacted>"

[triggers]
crons = ["* * * * *", "*/30 * * * *"]

[[queues.consumers]]
queue = "account-emails"
dead_letter_queue = "dlq"
max_batch_size = 20       
max_batch_timeout = 10    
max_retries = 10              

[[queues.consumers]]
queue = "incoming-emails"
dead_letter_queue = "dlq"
max_batch_size = 10       
max_batch_timeout = 10    
max_retries = 10          

[[queues.consumers]]
queue = "dlq"

[[queues.producers]]
binding = "ACCOUNT_EMAILS_QUEUE"
queue = "account-emails"

[[queues.producers]]
binding = "INCOMING_EMAILS_QUEUE"
queue = "incoming-emails"

[[workflows]]
name = "process-incoming-emails"
binding = "PROCESS_INCOMING_EMAILS_WORKFLOW"
class_name = "ProcessIncomingEmailsWorkflow"

#:schema node_modules/wrangler/config-schema.json
name = "my-app"
compatibility_date = "2024-10-22"
compatibility_flags = ["nodejs_compat"]
main = "dist/index.js"
minify = true

upload_source_maps = true

tail_consumers = [{ service = "my-app-tail" }]

[build]
command = "pnpm run build"

[observability]
enabled = true

[ai]
binding = "AI"

[[kv_namespaces]]
binding = "INCOMING_EMAILS_KV"
id = "<redacted>"

[[kv_namespaces]]
binding = "DLQ_KV"
id = "<redacted>"

[triggers]
crons = ["* * * * *", "*/30 * * * *"]

[[queues.consumers]]
queue = "account-emails"
dead_letter_queue = "dlq"
max_batch_size = 20       
max_batch_timeout = 10    
max_retries = 10              

[[queues.consumers]]
queue = "incoming-emails"
dead_letter_queue = "dlq"
max_batch_size = 10       
max_batch_timeout = 10    
max_retries = 10          

[[queues.consumers]]
queue = "dlq"

[[queues.producers]]
binding = "ACCOUNT_EMAILS_QUEUE"
queue = "account-emails"

[[queues.producers]]
binding = "INCOMING_EMAILS_QUEUE"
queue = "incoming-emails"

[[workflows]]
name = "process-incoming-emails"
binding = "PROCESS_INCOMING_EMAILS_WORKFLOW"
class_name = "ProcessIncomingEmailsWorkflow"

#:schema node_modules/wrangler/config-schema.json
name = "my-app"
compatibility_date = "2024-10-22"
compatibility_flags = ["nodejs_compat"]
main = "dist/index.js"
minify = true

upload_source_maps = true

tail_consumers = [{ service = "my-app-tail" }]

[build]
command = "pnpm run build"

[observability]
enabled = true

[ai]
binding = "AI"

[[kv_namespaces]]
binding = "INCOMING_EMAILS_KV"
id = "<redacted>"

[[kv_namespaces]]
binding = "DLQ_KV"
id = "<redacted>"

[triggers]
crons = ["* * * * *", "*/30 * * * *"]

[[queues.consumers]]
queue = "account-emails"
dead_letter_queue = "dlq"
max_batch_size = 20       
max_batch_timeout = 10    
max_retries = 10              

[[queues.consumers]]
queue = "incoming-emails"
dead_letter_queue = "dlq"
max_batch_size = 10       
max_batch_timeout = 10    
max_retries = 10          

[[queues.consumers]]
queue = "dlq"

[[queues.producers]]
binding = "ACCOUNT_EMAILS_QUEUE"
queue = "account-emails"

[[queues.producers]]
binding = "INCOMING_EMAILS_QUEUE"
queue = "incoming-emails"

[[workflows]]
name = "process-incoming-emails"
binding = "PROCESS_INCOMING_EMAILS_WORKFLOW"
class_name = "ProcessIncomingEmailsWorkflow"

#:schema node_modules/wrangler/config-schema.json
name = "my-app"
compatibility_date = "2024-10-22"
compatibility_flags = ["nodejs_compat"]
main = "dist/index.js"
minify = true

upload_source_maps = true

tail_consumers = [{ service = "my-app-tail" }]

[build]
command = "pnpm run build"

[observability]
enabled = true

[ai]
binding = "AI"

[[kv_namespaces]]
binding = "INCOMING_EMAILS_KV"
id = "<redacted>"

[[kv_namespaces]]
binding = "DLQ_KV"
id = "<redacted>"

[triggers]
crons = ["* * * * *", "*/30 * * * *"]

[[queues.consumers]]
queue = "account-emails"
dead_letter_queue = "dlq"
max_batch_size = 20       
max_batch_timeout = 10    
max_retries = 10              

[[queues.consumers]]
queue = "incoming-emails"
dead_letter_queue = "dlq"
max_batch_size = 10       
max_batch_timeout = 10    
max_retries = 10          

[[queues.consumers]]
queue = "dlq"

[[queues.producers]]
binding = "ACCOUNT_EMAILS_QUEUE"
queue = "account-emails"

[[queues.producers]]
binding = "INCOMING_EMAILS_QUEUE"
queue = "incoming-emails"

[[workflows]]
name = "process-incoming-emails"
binding = "PROCESS_INCOMING_EMAILS_WORKFLOW"
class_name = "ProcessIncomingEmailsWorkflow"

T

telliks•11/20/24, 7:27 PM

@Skye If it helps, I could successfully create a brand new workflow with the

npm create cloudflare@latest workflows-starter -- --template "cloudflare/workflows-starter"

npm create cloudflare@latest workflows-starter -- --template "cloudflare/workflows-starter"

command (apart from the empty object issue when

env.MY_WORKFLOW.create()

env.MY_WORKFLOW.create()

is called locally). Does this command do more than simply scaffolding the worker project on my local file system?

For my existing worker, I just added the workflow config to the toml file. Should I create a workflow with a wrangler CLI command first? Based on all the discussions, I'm under the impression that the wrangler.toml config should be enough and that during deployment wrangler will create a workflow if it doesn't exist yet. Please confirm if this is correct. I know this is not how other cloudflare products like KV or Queues work. You have to create them first and then reference them in the toml file.

T

telliks•11/20/24, 7:44 PM

Okay, finally got to the bottom of this. My local wrangler version was on the latest version but the one running in CI was on v3.80.0, so it never deployed the workflow

Thanks Skye, Matt, Diogo Ferreira and sdr for trying to help me

P

pao ramen•11/20/24, 9:41 PM

ok, in the end I gave up. I faced too many durable execution errors, and queues started to get clogged because

Workflow.create

Workflow.create

was taking a whole minute. I will keep an eye when workflows get more mature.

M

makisuo•11/20/24, 10:48 PM

Is it possible to have multiple workflows in one worker? If I try define multiple workflows wrangler crashes without any error logs.

Mmakisuo Is it possible to have multiple workflows in one worker? If I try define multipl...

A

arch•11/20/24, 11:25 PM

A fix was recently merged and will probably be tagged as a release soon https://github.com/cloudflare/workers-sdk/commit/563439bd02c450921b28d721d36be5a70897690d

GitHub

Fix multiple workflows and add proper engine persistance (#7286) · ...

fix: allow multiple workflow definitions
feat: make engine persist logs and status over multiple runs
chore: add persistance test and remove redundant env types
chore: change chang...

Original message was deleted

P

pao ramen•11/21/24, 8:55 AM

awesome! let me know when those improvements land so I can test them out.

Ttelliks @Skye If it helps, I could successfully create a brand new workflow with the `np...

D

Diogo Ferreira•11/21/24, 11:34 AM

^apart from the empty object issue when env.MY_WORKFLOW.create() is called locally

wrangler@3.88.0 fixes that

A

alex (he/him)•11/21/24, 4:31 PM

I have two things which are a bit annoying, but which I can workaround. However any ETA on fixes?

a) The Dashboard trigger doesn't work. The params object get's stringified strangely, so my logic never works. Triggering from the CLI or a Binding with the same params object works without issues
b) The option to delete workflows in the UI & CLI. I changed the name of the workflow, now I have a stale workflow with the old name in the UI

K

kchro3•11/21/24, 8:50 PM

Hi team, I'm seeing an issue when reading & writing from KV in a Workflow.

I am developing with

wrangler dev --remote --env stage

wrangler dev --remote --env stage

& i have a workflow & a KV in a stage env.

I am testing a workflow where it reads from a KV & then writes back to the same KV.

My assumption was that if I stage my worker => trigger the workflow, it will run the stage workflow => read & write from the preview KV in the stage environment because the worker has access to the preview KV.

However, it looks like the stage workflow is actually reading & writing against the non-preview KV in the stage environment, which is inconsistent with the worker behavior. Is this a bug?

I

irvyn•11/22/24, 4:03 PM

Hey based on the limits for workflows (based on workers) does this mean that if I had to do processing of a 2MB file -> 4MB file.

It wont work within just 1 step but I would be able to pass small amounts through multiple steps as long as it stays below the entire limit of "Maximum state that can be persisted per Workflow instance" ?

https://developers.cloudflare.com/workflows/reference/limits/

Maximum persisted state per step    1MiB (2^20 bytes)    1MiB (2^20 bytes)
Maximum state that can be persisted per Workflow instance    100MB    1GB

Maximum persisted state per step    1MiB (2^20 bytes)    1MiB (2^20 bytes)
Maximum state that can be persisted per Workflow instance    100MB    1GB

Cloudflare Docs

Limits | Cloudflare Workflows docs

Limits that apply to authoring, deploying, and running Workflows are detailed below.

森

森@VerseLink (RW)•11/22/24, 11:49 PM

is there a way to check if an workflow id exists?

森

森@VerseLink (RW)•11/22/24, 11:49 PM

say for example if workflow.get(id) and id doesn't exist what is being returned? the documentation wasn't clear about this

I'm unsure of the best way to implement a fan-out pattern using Workflows. I fetch a list of files,

Similar Threads