Async tasks

Flexible, durable scheduling across processes and time.

Restate lets you write resilient scheduling logic with the flexibility of code: delay execution, re-attach to ongoing tasks, fan-out/in patterns, and more. Restate guarantees that your tasks run to completion exactly once.

Durable timers

Persist timers in Restate. Restate makes sure that they get fired when they should, whether it's millis or months later.

(Delayed) task queue

Use Restate as message queue and schedule tasks asynchronously. Tasks execute with workflow-like durability and reliability. Restate handles retries and recovery of progress.

Switch between async and sync

Schedule an async task through Restate, then let any process latch onto it later to wait for the result or check its completion status, all with built-in durability.

Scheduling async tasks with Restate

Restate as message queue

Any function becomes a durable async task by using Restate as your message queue. Restate persists all requests and ensures they run to completion with automatic retries and recovery upon failures.

Learn more

Restate as message queue

Any function becomes a durable async task by using Restate as your message queue. Restate persists all requests and ensures they run to completion with automatic retries and recovery upon failures.

Learn more

Schedule async tasks

You can schedule tasks with the SDK clients or via HTTP, and you can postpone execution via the optional delay parameter.

Learn more

Schedule async tasks

You can schedule tasks with the SDK clients or via HTTP, and you can postpone execution via the optional delay parameter.

Learn more

Deduplicate requests

Use an idempotency key to ensure that the task is only scheduled once. For duplicate requests, Restate returns the previous response.

Deduplicate requests

Use an idempotency key to ensure that the task is only scheduled once. For duplicate requests, Restate returns the previous response.

Latch on to the task

Use the idempotency key to latch onto the task later and retrieve the result. You can also let another process latch on.

Learn more

Latch on to the task

Use the idempotency key to latch onto the task later and retrieve the result. You can also let another process latch on.

Learn more

TypeScript
Java
Kotlin
Go
Python

const asyncTaskService = restate.service({
  name: "taskWorker",
  handlers: {
    runTask: async (ctx: Context, params: TaskOpts) => {
      return someHeavyWork(params);
    },
  },
});

export type AsyncTaskService = typeof asyncTaskService;

// The TypeScript SDK includes a client to send requests to services
const restateClient = restate.connect({ url: RESTATE_URL });
const taskHandle = await restateClient
  .serviceSendClient<AsyncTaskService>({ name: "taskWorker" })
  .runTask(
    task,
    SendOpts.from({ idempotencyKey: "dQw4w9WgXcQ", delay: 1000 })
  );

// Attach to the async task to get the result
const result = await restateClient.result(taskHandle);

@Service
public class AsyncTaskService {

  @Handler
  public String runTask(Context ctx, TaskOpts params) {
    return someHeavyWork(params);
  }
}

// The Java SDK generates clients for each service
Client restateClient = Client.connect(RESTATE_URL);
SendResponse handle =
    AsyncTaskServiceClient.fromClient(restateClient)
        .send(Duration.ofDays(5))
        .runTask(
            taskOpts,
            CallRequestOptions.DEFAULT.withIdempotency("dQw4w9WgXcQ")
            );

// Attach to the async task to get the result
String result =
    restateClient
        .invocationHandle(handle.getInvocationId(), JsonSerdes.STRING)
        .attach();

@Service
class AsyncTaskService {
  @Handler
  suspend fun runTask(ctx: Context, params: TaskOpts): String {
    return params.someHeavyWork()
  }
}

// The Kotlin SDK generates clients for each service
val restateClient: Client = Client.connect(RESTATE_URL)
val handle =
    AsyncTaskServiceClient.fromClient(restateClient)
        .send(5.days)
        .runTask(
            taskOpts,
            CallRequestOptions.DEFAULT.withIdempotency("dQw4w9WgXcQ"),
        )

// Attach to the async task to get the result
val result =
    restateClient
        .invocationHandle(
            handle.invocationId,
            KtSerdes.json<String>(),
        )
        .attach()

type AsyncTaskWorker struct{}

func (AsyncTaskWorker) RunTask(ctx restate.Context, task TaskOpts) (Result, error) {
  return someHeavyWork(task)
}

client := &http.Client{}
url := fmt.Sprintf("%s/AsyncTaskWorker/RunTask/Send", RESTATE_URL)
taskData, _ := json.Marshal(task)
req, err := http.NewRequest("POST", url, bytes.NewBuffer(taskData))
if err != nil {
  return err
}
req.Header.Set("Content-Type", "application/json")
req.Header.Set("idempotency-key", "dQw4w9WgXcQ")

resp, err := client.Do(req)
if err != nil {
  return err
}
defer resp.Body.Close()

// ... do other things while the task is being processed ...

attachUrl := fmt.Sprintf(
  "%s/restate/invocation/AsyncTaskWorker/RunTask/%s/attach",
  RESTATE_URL,
  "dQw4w9WgXcQ")
resp, err = http.DefaultClient.Get(attachUrl)
if err != nil {
  return err
}
defer resp.Body.Close()

async_task_service = Service("taskWorker")


@async_task_service.handler("runTask")
async def run_task(ctx: Context, params: TaskOpts):
    return some_heavy_work(params)

app = restate.app([async_task_service])

requests.post(
    f"{RESTATE_URL}/taskWorker/runTask/send",
      json=json.dumps(task),
      headers={
          "idempotency-key": "dQw4w9WgXcQ",
          "Content-Type": "application/json"
      })

# Attach to the async task to retrieve the result
attach_url = f"{RESTATE_URL}/restate/invocation/taskWorker/runTask/dQw4w9WgXcQ/attach"
response = requests.get(attach_url)

Restate as sophisticated task queue

Restate is built as an event-driven foundation, and therefore supports task queues by design.
Async tasks run like any other function in your infrastructure: on K8S, FaaS, or mix-and-match.
No need to spin up extra infrastructure or message queues.

Parallelizing work with Restate

Fan out

Write flexible scheduling logic with Restate's durable building blocks. Fan out tasks with resilient RPC calls. Restate makes sure all tasks run to completion, with retries and recovery upon failures.

Fan out

Fan in

Invocations produce durable promises that can be awaited and combined. These durable promises can be recovered on other processes after a failure.

Fan in

Invocations produce durable promises that can be awaited and combined. These durable promises can be recovered on other processes after a failure.

Server(less)

Deploy this service or its subtask processors on a platform like Kubernetes or AWS Lambda to automatically get parallel scale out.

Learn more

Server(less)

Deploy this service or its subtask processors on a platform like Kubernetes or AWS Lambda to automatically get parallel scale out.

Learn more

TypeScript
Java
Kotlin
Go
Python

const workerService = restate.service({
  name: "worker",
  handlers: {
    run: async (ctx: Context, task: Task) => {
      // Split the task in subtasks
      const subtasks: SubTask[] = await ctx.run("split task", () =>
        split(task)
      );

      const resultPromises = [];
      for (const subtask of subtasks) {
        const subResultPromise = ctx
          .serviceClient(workerService)
          .runSubtask(subtask);
        resultPromises.push(subResultPromise);
      }

      const results = await CombineablePromise.all(resultPromises);
      return aggregate(results);
    },

    runSubtask: async (ctx: Context, subtask: SubTask) => {
      // Processing logic goes here ...
      // Can be moved to a separate service to scale independently
    },
  },
});

export const handler = restate.endpoint().bind(workerService).handler();

@Service
public class FanOutWorker {

  @Handler
  public Result run(Context ctx, Task task) {

    // Split the task in subtasks
    List<SubTask> subTasks =
        ctx.run(
            JacksonSerdes.of(new TypeReference<>() {}),
            () -> split(task));

    List<Awaitable<?>> resultFutures = new ArrayList<>();
    for (SubTask subTask : subTasks) {
      resultFutures.add(FanOutWorkerClient.fromContext(ctx).runSubtask(subTask));
    }

    Awaitable.all(resultFutures).await();

    var results = resultFutures.stream().map(future -> (SubTaskResult) future.await()).toList();

    return aggregate(results);
  }

  @Handler
  public SubTaskResult runSubtask(Context ctx, SubTask subTask) {
    // Processing logic goes here ...
    // Can be moved to a separate service to scale independently
    return executeSubtask(ctx, subTask);
  }

  public static void main(String[] args) {
    RestateHttpEndpointBuilder.builder().bind(new FanOutWorker()).buildAndListen();
  }
}

@Service
class FanOutWorker {
  @Handler
  suspend fun run(ctx: Context, task: Task): TaskResult {
    val subTasks = ctx.runBlock { task.split() }

    val results =
        subTasks
            .map {
              FanOutWorkerClient.fromContext(ctx).runSubtask(it)
            }
            .awaitAll()
    return results.aggregate()
  }

  @Handler
  suspend fun runSubtask(ctx: Context, subTask: SubTask): SubTaskResult {
    // Processing logic goes here ...
    // Can be moved to a separate service to scale independently
    return subTask.execute(ctx)
  }
}

type FanOutWorker struct{}

func (FanOutWorker) Run(ctx restate.Context, task Task) (Result, error) {
  // Split the task in subtassk
  subtasks, err := split(task)
  if err != nil {
    return Result{}, err
  }

  subtaskFutures := make([]restate.Selectable, 0, len(subtasks))
  for _, subtask := range subtasks {
    subtaskFutures = append(subtaskFutures,
      restate.Service[SubTaskResult](ctx, "FanOutWorker", "RunSubtask").
        RequestFuture(subtask))
  }

  selector := restate.Select(ctx, subtaskFutures...)

  subResults := make([]SubTaskResult, 0, len(subtasks))
  for selector.Remaining() {
    response, err := selector.Select().(restate.ResponseFuture[SubTaskResult]).Response()
    if err != nil {
      return Result{}, err
    }
    subResults = append(subResults, response)
  }

  return aggregate(subResults)
}

func (FanOutWorker) RunSubtask(ctx restate.Context, subtask SubTask) (SubTaskResult, error) {
  // Processing logic goes here ...
  // Can be moved to a separate service or FaaS to scale independently
  return executeSubtask(ctx, subtask)
}

worker_service = Service("worker")


@worker_service.handler()
async def run(ctx: Context, task: Task):
    subtasks = await ctx.run("split task", lambda: split(task))

    result_promises = []
    for subtask in subtasks:
        sub_result_promise = ctx.service_call(run_subtask, arg=subtask)
        result_promises.append(sub_result_promise)

    results = [await promise for promise in result_promises]
    return aggregate(results)


@worker_service.handler()
async def run_subtask(ctx: Context, subtask: Subtask):
    # Processing logic goes here...
    # Can be moved to a separate service to scale independently
    pass


app = restate.app([worker_service])

LOW-LATENCY

Restate’s event-driven foundation built in Rust lets you queue events. Restate pushes them to your functions at high speed.

Learn more

DURABLE EXECUTION

Restate guarantees all tasks run to completion. It keeps track of timers, handles retries and recovery upon failures, and ensures that tasks are executed exactly once.

Learn more

Durable webhook processing with Restate

Restate handlers as durable event processors

Point your webhook endpoint to any Restate handler. Restate makes sure all events are persisted and run to completion.

Restate handlers as durable event processors

Point your webhook endpoint to any Restate handler. Restate makes sure all events are persisted and run to completion.

Schedule follow-up tasks for webhook events, like reminders or escalations.

Stateful handlers and event joins

Correlate or join asynchronous events by routing them to the same object.

Restate ensures sequential processing of events for the same key while giving access to durable key-value state.

Stateful handlers and event joins

Correlate or join asynchronous events by routing them to the same object.

Restate ensures sequential processing of events for the same key while giving access to durable key-value state.

TypeScript
Java
Kotlin
Go
Python

const paymentTracker = restate.object({ // one instance per invoice id
    name: "PaymentTracker",
    handlers: {
        // Stripe sends us webhook events for invoice payment attempts
        onPaymentSuccess: async (ctx: restate.ObjectContext, event: StripeEvent) => {
            ctx.set("paid", true);
        },
        onPaymentFailed: async (ctx: restate.ObjectContext, event: StripeEvent) => {
            if (await ctx.get<boolean>("paid")) {
                return;
            }

            const remindersCount = await ctx.get<number>("reminders_count") ?? 0;
            if (remindersCount < 3) {
                ctx.set("reminders_count", remindersCount + 1);
                await ctx.run(() => sendReminderEmail(event));

                // Schedule next reminder via a delayed self call
                await ctx.objectSendClient(
                    PaymentTracker,
                    ctx.key, // this object's invoice id
                    {delay: 24 * 60 * 60 * 1000}
                ).onPaymentFailed(event);
            } else {
                await ctx.run(() => escalateToHuman(event));
            }
        },
    }
})

@VirtualObject
public class PaymentTracker {

  StateKey<Boolean> PAID = StateKey.of("paid", BOOLEAN);
  StateKey<Integer> REMINDER_COUNT = StateKey.of("reminder_count", INT);

  // Stripe sends us webhook events for invoice payment attempts
  @Handler
  public void onPaymentSuccess(ObjectContext ctx, StripeEvent event) {
    ctx.set(PAID, true);
  }

  @Handler
  public void onPaymentFailure(ObjectContext ctx, StripeEvent event) {
    if (ctx.get(PAID).orElse(false)) {
      return;
    }

    int remindersCount = ctx.get(REMINDER_COUNT).orElse(0);
    if (remindersCount < 3) {
      ctx.set(REMINDER_COUNT, remindersCount + 1);
      ctx.run(() -> sendReminderEmail(event));

      // Schedule next reminder via a delayed self call
      PaymentTrackerClient.fromContext(ctx, ctx.key())
          .send(Duration.ofDays(1))
          .onPaymentFailure(event);
    } else {
      ctx.run(() -> escalateToHuman(event));
    }
  }
}

@VirtualObject
class PaymentTracker {

  companion object {
    private val PAID = KtStateKey.json<Boolean>("paid")
    private val REMINDER_COUNT = KtStateKey.json<Int>("reminder_count")
  }

  // Stripe sends us webhook events for invoice payment attempts
  @Handler
  suspend fun onPaymentSuccess(ctx: ObjectContext, event: StripeEvent) {
    ctx.set(PAID, true)
  }

  @Handler
  suspend fun onPaymentFailure(ctx: ObjectContext, event: StripeEvent) {
    if (ctx.get(PAID) == true) {
      return
    }

    val remindersCount = ctx.get(REMINDER_COUNT) ?: 0
    if (remindersCount < 3) {
      ctx.set(REMINDER_COUNT, remindersCount + 1)
      ctx.runBlock { sendReminderEmail(event) }

      // Schedule next reminder via a delayed self call
      PaymentTrackerClient.fromContext(ctx, ctx.key()).send(1.days).onPaymentFailure(event)
    } else {
      ctx.runBlock { escalateToHuman(event) }
    }
  }
}

type PaymentTracker struct{} // one instance per invoice id

// OnPaymentSuccess Stripe sends us webhook events for invoice payment attempts
func (PaymentTracker) OnPaymentSuccess(ctx restate.ObjectContext, event StripeEvent) restate.Void {
  restate.Set[bool](ctx, "paid", true)
  return restate.Void{}
}

func (PaymentTracker) OnPaymentFailure(ctx restate.ObjectContext, event StripeEvent) error {
  paid, err := restate.Get[bool](ctx, "paid")
  if err != nil {
    return err
  }
  if paid {
    return nil
  }

  remindersCount, err := restate.Get[int8](ctx, "reminders_count")
  if err != nil {
    return err
  }
  if remindersCount < 3 {
    restate.Set(ctx, "reminders_count", remindersCount+1)
    if _, err := restate.Run(ctx, func(ctx restate.RunContext) (restate.Void, error) {
      return restate.Void{}, SendReminderEmail(event)
    }); err != nil {
      return err
    }

    // Schedule next reminder via a delayed self call
    restate.ObjectSend(ctx,
      "PaymentTracker",
      restate.Key(ctx), // this object's invoice id
      "OnPaymentFailure").
      Send(event, restate.WithDelay(5*time.Second))
  } else {
    if _, err := restate.Run(ctx, func(ctx restate.RunContext) (restate.Void, error) {
      return restate.Void{}, EscalateToHuman(event)
    }); err != nil {
      return err
    }
  }
  return nil
}

payment_tracker = VirtualObject("PaymentTracker") # one instance per invoice ID


# Stripe sends us webhook events for invoice payment attempts
@payment_tracker.handler()
async def on_payment_success(ctx: ObjectContext, event: StripeEvent):
    ctx.set("paid", True)


@payment_tracker.handler()
async def on_payment_failed(ctx: ObjectContext, event: StripeEvent):
    if await ctx.get("paid"):
        return

    reminder_count = await ctx.get("reminder_count") or 0
    if reminder_count < 3:
        ctx.set("reminder_count", reminder_count + 1)
        await ctx.run("send_reminder", lambda: send_reminder_email(event))

        # Schedule next reminder via a delayed self call
        ctx.object_send(
            on_payment_failed, # this handler
            ctx.key(), # this object invoice id
            event,
            send_delay=timedelta(days=1))
    else:
        await ctx.run("escalate", lambda: escalate_to_human(event))

What you can build with Async Tasks and Restate

(Delayed) task queue

A task queue that can handle both immediate and scheduled tasks. Restate makes sure task submissions are idempotent and that tasks run to completion exactly once. You can attach back to tasks to retrieve their result later.

Durable webhook event processing

A Restate handler can be the endpoint for a webhook. Restate makes sure the events are persisted and retried until they run till completion.

Complex task scheduling

An invoice payment tracker which correlates payment success events and failure events. Until a success event is received, each failure event generates up to 3 daily reminders and then escalates to support.

Payments: Combining sync & async responses

Issue an idempotent payment to Stripe and process the response. The payment provider either responds immediately or notifies us later via a webhook.

Building Durable-Promises-as-a-Service with Restate

Using Restate to build Promises/Futures as a service, that can be exposed to external clients and are durable across processes and failures..

Wondering about a specific use case?

Let’s brainstorm together on Discord or Slack.

Didn't see your use case?

💡 You can invoke any handler asynchronously, so have a look at the other use case pages for more inspiration.

Developer Resources

Docs: Invocations and Scheduling

Docs: Durable Timers

Need help?

Join the Restate Discord or Slack communities

Async tasks

Flexible, durable scheduling across processes and time.

Durable timers

(Delayed) task queue

Switch between async and sync

Scheduling async tasks with Restate​

Restate as sophisticated task queue

Parallelizing work with Restate​

LOW-LATENCY

DURABLE EXECUTION

Durable webhook processing with Restate​

What you can build with Async Tasks and Restate​

(Delayed) task queue

Durable webhook event processing

Complex task scheduling

Payments: Combining sync & async responses

Building Durable-Promises-as-a-Service with Restate

Wondering about a specific use case?

Developer Resources​

Docs: Invocations and Scheduling

Docs: Durable Timers

Need help?

Scheduling async tasks with Restate

Parallelizing work with Restate

Durable webhook processing with Restate

What you can build with Async Tasks and Restate

Developer Resources