DiscordBlogTwitterConsole
OverviewRedisVectorQStashWorkflowSearchBoxRealtimeDeveloper API
Getting Started

Caveats

Introduction

In this guide, we'll look at best practices and caveats for using Upstash Workflow.

Core Principles

Execute business logic in context.run

Your workflow endpoint will be called multiple times during a workflow run. Therefore:

  • Place your business logic code inside the context.run function for each step
  • Code outside context.run only serves to connect steps

Example:

api/workflow/route.ts
export const { POST } = serve<string>(async (context) => {  const input = context.requestPayload  const result = await context.run("step-1", () => {    return { success: true }  })  console.log("This log will appear multiple times")  await context.run("step-2", () => {    console.log("This log will appear just once")    console.log("Step 1 status is:", result.success)  })})
main.py
@serve.post("/api/example")async def example(context: AsyncWorkflowContext[str]) -> None:    input = context.request_payload    async def _step_1() -> Dict:        return {"success": True}    result = await context.run("step-1", _step_1)    print("This log will appear multiple times")    async def _step_2() -> None:        print("This log will appear just once")        print("Step 1 status is:", result["success"])    await context.run("step-2", _step_2)

Return Results from context.run for Later Use

Always return step results if needed in subsequent steps.

❌ Incorrect - TypeScript
export const { POST } = serve<string>(async (context) => {  const input = context.requestPayload  let result  await context.run("step-1", async () => {    result = await someWork(input)  })  await context.run("step-2", async () => {    await someOtherWork(result)  })})
✅ Correct - TypeScript
export const { POST } = serve<string>(async (context) => {  const input = context.requestPayload  const result = await context.run("step-1", async () => {    return await someWork(input)  })  await context.run("step-2", async () => {    someOtherWork(result)  })})
❌ Incorrect - Python
@serve.post("/api/example")async def example(context: AsyncWorkflowContext[str]) -> None:    input = context.request_payload    result = None    async def _step_1() -> Dict:        nonlocal result        result = await some_work(input)    await context.run("step-1", _step_1)    async def _step_2() -> None:        await some_other_work(result)    await context.run("step-2", _step_2)
✅ Correct - Python
@serve.post("/api/example")async def example(context: AsyncWorkflowContext[str]) -> None:    input = context.request_payload    async def _step_1() -> Dict:        return await some_work(input)    result = await context.run("step-1", _step_1)    async def _step_2() -> None:        await some_other_work(result)    await context.run("step-2", _step_2)

Because your workflow endpoint is called multiple times, result will be unitialized when the endpoint is called again to run step-2.

If you are curious about why an endpoint is called multiple times, see how Workflow works.

Avoiding Common Pitfalls

Avoid Non-deterministic Code Outside context.run

A workflow endpoint should always produce the same results, even if it's called multiple times. Avoid:

  • Non-idempotent functions
  • Time-dependent code
  • Randomness

Example of what to avoid:

❌ Non-idempotent functions - TypeScript
export const { POST } = serve<{ entryId: string }>(async (context) => {  const { entryId } = context.requestPayload;  // 👇 Problem: Non-idempotent function outside context.run:  const result = await getResultFromDb(entryId);  if (result.return) {    return;  }  // ...})
❌ Time-dependent code - TypeScript
export const { POST } = serve<string>(async (context) => {  const input = context.requestPayload  // 👇 Problem: time-dependent code  if (Date.now() % 5 == 2) {    await context.run("step-1", () => {      // ...    })  } else {    await context.run("step-2", () => {      // ...    })  }})
❌ Random code - TypeScript
export const { POST } = serve<string>(async (context) => {  const input = context.requestPayload  // 👇 Problem: random code  if (Math.floor(Math.random() * 10) % 5 == 2) {    await context.run("step-1", () => {      // ...    })  } else {    await context.run("step-2", () => {      // ...    })  }})
❌ Non-idempotent functions - Python
@serve.post("/api/example")async def example(context: AsyncWorkflowContext[str]) -> None:    entry_id = context.request_payload["entry_id"]    # 👇 Problem: Non-idempotent function outside context.run:    result = await get_result_from_db(entry_id)    if result.should_return:        return    # ...
❌ Time-dependent code - Python
@serve.post("/api/example")async def example(context: AsyncWorkflowContext[str]) -> None:    input = context.request_payload    # 👇 Problem: time-dependent code    if time.time() % 5 == 2:        await context.run("step-1", lambda: ...)    else:        await context.run("step-2", lambda: ...)
❌ Random code - Python
@serve.post("/api/example")async def example(context: AsyncWorkflowContext[str]) -> None:    input = context.request_payload    # 👇 Problem: random code    if random.randint(0, 9) % 5 == 2:        await context.run("step-1", lambda: ...)    else:        await context.run("step-2", lambda: ...)

If you implement a non-idempotent code like the one shown above, you might encounter Failed to authenticate Workflow request. errors. This can happen if you return based on the result of the non-idempotent code before any workflow step.

To prevent this, ensure that the non-idempotent code (such as getResultFromDb in the example) runs within context.run.

TypeScript
const result = await context.run(async () => {  await getResultFromDb(entryId)});if (result.return) {  return;}
Python
async def _get_result_from_db():    return await get_result_from_db(entry_id)result = await context.run("get-result-from-db", _get_result_from_db)if result.should_return:    return

Ensure Idempotency in context.run

Business logic should be idempotent due to potential retries in distributed systems. In other words, when a workflow runs twice with the same input, the end result should be the same as if the workflow only ran once.

In the example below, the someWork function must be idempotent:

api/workflow/route.ts
export const { POST } = serve<string>(async (context) => {  const input = context.requestPayload  await context.run("step-1", async () => {    return someWork(input)  })})
main.py
@serve.post("/api/example")async def example(context: AsyncWorkflowContext[str]) -> None:    input = context.request_payload    async def _step_1() -> None:        return await some_work(input)    await context.run("step-1", _step_1)

Imagine that someWork executes once and makes a change to a database. However, before the database had a chance to respond with the successful change, the connection is lost. Your Workflow cannot know if the database change was successful or not. The caller has no choice but to retry, which will cause someWork to run twice.

If someWork is not idempotent, this could lead to unintended consequences. For example duplicated records or corrupted data. Idempotency is crucial to maintaining the integrity and reliability of your workflow.

Don't Nest Context Methods

Avoid calling context.call, context.sleep, context.sleepFor, or context.run within another context.run.

api/workflow/route.ts
import { serve } from "@upstash/workflow/nextjs"export const { POST } = serve<string>(async (context) => {  const input = context.requestPayload  await context.run("step-1", async () => {    await context.sleep(...) // ❌ INCORRECT    await context.run(...) // ❌ INCORRECT    await context.call(...) // ❌ INCORRECT  })})
main.py
@serve.post("/api/example")async def example(context: AsyncWorkflowContext[str]) -> None:    input = context.request_payload    async def _step_1() -> None:        await context.sleep(...) #  ❌ INCORRECT        await context.run(...) #  ❌ INCORRECT        await context.call(...) #  ❌ INCORRECT    await context.run("step-1", _step_1)

Include At Least One Step in Workflow

Every workflow must include at least one step execution with context.run. If no steps are defined, the workflow will throw a Failed to authenticate Workflow request. error.

❌ Missing steps - TypeScript
export const { POST } = serve<string>(async (context) => {  const input = context.requestPayload    // 👇 Problem: No context.run call  console.log("Processing input:", input)    // This workflow will fail with "Failed to authenticate Workflow request."})
✅ Correct - TypeScript
export const { POST } = serve<string>(async (context) => {  const input = context.requestPayload    // 👇 At least one step is required  await context.run("dummy-step", async () => {    return  })})
❌ Missing steps - Python
@serve.post("/api/example")async def example(context: AsyncWorkflowContext[str]) -> None:    input = context.request_payload        # 👇 Problem: No context.run call    print("Processing input:", input)        # This workflow will fail with "Failed to authenticate Workflow request."
✅ Correct - Python
@serve.post("/api/example")async def example(context: AsyncWorkflowContext[str]) -> None:    input = context.request_payload        # 👇 At least one step is required    async def _dummy_step():        return            await context.run("dummy-step", _dummy_step)

Even for the placeholder implementations, you must include one dummy step for the Workflow authentication mechanism to function properly.

Avoid Promise.any

In workflow-js, you can use Promise.all to run steps in parallel. However, a similar method, Promise.any, is not supported for workflow steps.

While Promise.all works seamlessly, Promise.any does not currently function with workflow steps. We are exploring the possibility of adding support for Promise.any in the future.

If you have a specific use case that requires Promise.any, don't hesitate to reach out to Upstash support.

Loading search…
↑↓ Select↵ Openesc Close