Async tasks

Flexible, durable scheduling across processes and time.

Write resilient scheduling logic with the flexibility of code: delay execution, re-attach to ongoing tasks, fan out/in, etc.
Restate runs your tasks to completion exactly once.

Durable timers

Register timers with Restate to make them durable. Restate makes sure that they get fired when they should, whether it's millis or months.

Message queue

Schedule tasks asynchronously, by using Restate as message queue. Tasks execute with workflow-like semantics and durability. Restate handles retries and recovery of progress.

Switch between async and sync

Schedule an async task, by using Restate as queue. Let another process latch on to it later and wait for the result, or peek at the output to see if it has finished.

Async tasks with Restate

Schedule tasks for now or later with the Restate SDK.

TypeScript
Java
Kotlin

Execute any handler async

Every handler in Restate is executed asynchronously and can be treated as a reliable asynchronous task. No matter whether it is a simple function, or a complex workflow. Restate persists the requests to this handler and makes sure they run to completion. Restate handles retries and recovery upon failures.

task_submitter.ts

async_task_service.ts


const asyncTaskService = restate.service({
  name: "taskWorker",
  handlers: {
    runTask: async (ctx: Context, params: TaskOpts) => {
      return someHeavyWork(params);
    },
  },
});
export type AsyncTaskService = typeof asyncTaskService;
const endpoint = restate.endpoint().bind(asyncTaskService).listen(9080);

Schedule tasks reliably

Schedule tasks asynchronously, by using Restate as message queue. Restate reliably queues them, also under backpressure/load.

Handlers can be called asynchronously from anywhere. This returns a task handle once the call in enqueued.

task_submitter.ts

async_task_service.ts


async function submitAndAwaitTask(task: TaskOpts) {
  const rs = restate.connect({ url: RESTATE_URL });
  const taskHandle = await rs
    .serviceSendClient<AsyncTaskService>({ name: "taskWorker" })
    .runTask(
      task,
      SendOpts.from({ idempotencyKey: "dQw4w9WgXcQ" })
    );
  // await the handler's result
  const result = await rs.result(taskHandle);
}
async function attachToTask(taskHandle: string) {
  const rs = restate.connect({ url: RESTATE_URL });
  const result2 = await rs.result<string>(JSON.parse(taskHandle));
}

Idempotent task submission

Use an idempotency key to ensure that the task is only scheduled once. Restate will deduplicate the request and return the previous response.

task_submitter.ts

async_task_service.ts


async function submitAndAwaitTask(task: TaskOpts) {
  const rs = restate.connect({ url: RESTATE_URL });
  const taskHandle = await rs
    .serviceSendClient<AsyncTaskService>({ name: "taskWorker" })
    .runTask(
      task,
      SendOpts.from({ idempotencyKey: "dQw4w9WgXcQ" })
    );
  // await the handler's result
  const result = await rs.result(taskHandle);
}
async function attachToTask(taskHandle: string) {
  const rs = restate.connect({ url: RESTATE_URL });
  const result2 = await rs.result<string>(JSON.parse(taskHandle));
}

Latch on to the task

For requests with an idempotency key, you can use this task handle to latch on to the task later and retrieve the result, or wait for it to finish.

task_submitter.ts

async_task_service.ts


async function submitAndAwaitTask(task: TaskOpts) {
  const rs = restate.connect({ url: RESTATE_URL });
  const taskHandle = await rs
    .serviceSendClient<AsyncTaskService>({ name: "taskWorker" })
    .runTask(
      task,
      SendOpts.from({ idempotencyKey: "dQw4w9WgXcQ" })
    );
  // await the handler's result
  const result = await rs.result(taskHandle);
}
async function attachToTask(taskHandle: string) {
  const rs = restate.connect({ url: RESTATE_URL });
  const result2 = await rs.result<string>(JSON.parse(taskHandle));
}

This works across processes, so you can have a separate process latch on to the task later.

task_submitter.ts

async_task_service.ts


async function submitAndAwaitTask(task: TaskOpts) {
  const rs = restate.connect({ url: RESTATE_URL });
  const taskHandle = await rs
    .serviceSendClient<AsyncTaskService>({ name: "taskWorker" })
    .runTask(
      task,
      SendOpts.from({ idempotencyKey: "dQw4w9WgXcQ" })
    );
  // await the handler's result
  const result = await rs.result(taskHandle);
}
async function attachToTask(taskHandle: string) {
  const rs = restate.connect({ url: RESTATE_URL });
  const result2 = await rs.result<string>(JSON.parse(taskHandle));
}

Execute any handler async

Schedule tasks reliably

Schedule tasks asynchronously, by using Restate as message queue. Restate reliably queues them, also under backpressure/load.

Handlers can be called asynchronously from anywhere. This returns a task handle once the call in enqueued.

Idempotent task submission

Use an idempotency key to ensure that the task is only scheduled once. Restate will deduplicate the request and return the previous response.

Latch on to the task

For requests with an idempotency key, you can use this task handle to latch on to the task later and retrieve the result, or wait for it to finish.

This works across processes, so you can have a separate process latch on to the task later.

task_submitter.ts

async_task_service.ts


const asyncTaskService = restate.service({
  name: "taskWorker",
  handlers: {
    runTask: async (ctx: Context, params: TaskOpts) => {
      return someHeavyWork(params);
    },
  },
});
export type AsyncTaskService = typeof asyncTaskService;
const endpoint = restate.endpoint().bind(asyncTaskService).listen(9080);

Execute any handler async

TaskSubmitter.java

AsyncTaskService.java


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

Schedule tasks reliably

Schedule tasks asynchronously, by using Restate as message queue. Restate reliably queues them, also under backpressure/load.

Handlers can be called asynchronously from anywhere. This returns a task handle once the call in enqueued.

TaskSubmitter.java

AsyncTaskService.java


public class TaskSubmitter {
  private static final Client rs = Client.connect("http://localhost:8080");
  public void submitAndAwaitTasks(TaskOpts taskOpts) {
    SendResponse handle =
        AsyncTaskServiceClient.fromClient(rs)
            .send()
            .runTask(
                taskOpts,
                CallRequestOptions.DEFAULT.withIdempotency("dQw4w9WgXcQ")
                );
    // await the handler's result
    String result = rs.invocationHandle(handle.getInvocationId(), JsonSerdes.STRING).attach();
  }
}

Idempotent task submission

Use an idempotency key to ensure that the task is only scheduled once. Restate will deduplicate the request and return the previous response.

TaskSubmitter.java

AsyncTaskService.java


public class TaskSubmitter {
  private static final Client rs = Client.connect("http://localhost:8080");
  public void submitAndAwaitTasks(TaskOpts taskOpts) {
    SendResponse handle =
        AsyncTaskServiceClient.fromClient(rs)
            .send()
            .runTask(
                taskOpts,
                CallRequestOptions.DEFAULT.withIdempotency("dQw4w9WgXcQ")
                );
    // await the handler's result
    String result = rs.invocationHandle(handle.getInvocationId(), JsonSerdes.STRING).attach();
  }
}

Latch on to the task

For requests with an idempotency key, you can use this task handle to latch on to the task later and retrieve the result, or wait for it to finish.

This works across processes, so you can have a separate process latch on to the task later.

TaskSubmitter.java

AsyncTaskService.java


public class TaskSubmitter {
  private static final Client rs = Client.connect("http://localhost:8080");
  public void submitAndAwaitTasks(TaskOpts taskOpts) {
    SendResponse handle =
        AsyncTaskServiceClient.fromClient(rs)
            .send()
            .runTask(
                taskOpts,
                CallRequestOptions.DEFAULT.withIdempotency("dQw4w9WgXcQ")
                );
    // await the handler's result
    String result = rs.invocationHandle(handle.getInvocationId(), JsonSerdes.STRING).attach();
  }
}

Execute any handler async

Schedule tasks reliably

Schedule tasks asynchronously, by using Restate as message queue. Restate reliably queues them, also under backpressure/load.

Handlers can be called asynchronously from anywhere. This returns a task handle once the call in enqueued.

Idempotent task submission

Use an idempotency key to ensure that the task is only scheduled once. Restate will deduplicate the request and return the previous response.

Latch on to the task

For requests with an idempotency key, you can use this task handle to latch on to the task later and retrieve the result, or wait for it to finish.

This works across processes, so you can have a separate process latch on to the task later.

TaskSubmitter.java

AsyncTaskService.java


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

Execute any handler async

TaskSubmitter.kt

AsyncTaskService.kt


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

Schedule tasks reliably

Schedule tasks asynchronously, by using Restate as message queue. Restate reliably queues them, also under backpressure/load.

Handlers can be called asynchronously from anywhere. This returns a task handle once the call in enqueued.

TaskSubmitter.kt

AsyncTaskService.kt


class TaskSubmitter {
    companion object {
        private val rs: Client = Client.connect("http://localhost:8080")
    }
    suspend fun submitAndAwaitTasks(taskOpts: TaskOpts) {
        val handle = AsyncTaskServiceClient.fromClient(rs)
                .send()
                .runTask(
                    taskOpts,
                    CallRequestOptions.DEFAULT.withIdempotency("dQw4w9WgXcQ")
                )
        // await the handler's result
        val result = rs.invocationHandle(
                handle.invocationId,
                KtSerdes.json<String>()
            )
            .attach()
    }
}

Idempotent task submission

Use an idempotency key to ensure that the task is only scheduled once. Restate will deduplicate the request and return the previous response.

TaskSubmitter.kt

AsyncTaskService.kt


class TaskSubmitter {
    companion object {
        private val rs: Client = Client.connect("http://localhost:8080")
    }
    suspend fun submitAndAwaitTasks(taskOpts: TaskOpts) {
        val handle = AsyncTaskServiceClient.fromClient(rs)
                .send()
                .runTask(
                    taskOpts,
                    CallRequestOptions.DEFAULT.withIdempotency("dQw4w9WgXcQ")
                )
        // await the handler's result
        val result = rs.invocationHandle(
                handle.invocationId,
                KtSerdes.json<String>()
            )
            .attach()
    }
}

Latch on to the task

For requests with an idempotency key, you can use this task handle to latch on to the task later and retrieve the result, or wait for it to finish.

This works across processes, so you can have a separate process latch on to the task later.

TaskSubmitter.kt

AsyncTaskService.kt


class TaskSubmitter {
    companion object {
        private val rs: Client = Client.connect("http://localhost:8080")
    }
    suspend fun submitAndAwaitTasks(taskOpts: TaskOpts) {
        val handle = AsyncTaskServiceClient.fromClient(rs)
                .send()
                .runTask(
                    taskOpts,
                    CallRequestOptions.DEFAULT.withIdempotency("dQw4w9WgXcQ")
                )
        // await the handler's result
        val result = rs.invocationHandle(
                handle.invocationId,
                KtSerdes.json<String>()
            )
            .attach()
    }
}

Execute any handler async

Schedule tasks reliably

Schedule tasks asynchronously, by using Restate as message queue. Restate reliably queues them, also under backpressure/load.

Handlers can be called asynchronously from anywhere. This returns a task handle once the call in enqueued.

Idempotent task submission

Use an idempotency key to ensure that the task is only scheduled once. Restate will deduplicate the request and return the previous response.

Latch on to the task

For requests with an idempotency key, you can use this task handle to latch on to the task later and retrieve the result, or wait for it to finish.

This works across processes, so you can have a separate process latch on to the task later.

TaskSubmitter.kt

AsyncTaskService.kt


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

LOW-LATENCY

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

DURABLE EXECUTION

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

Parallelizing work with Restate

Write flexible scheduling logic via durable building blocks.

TypeScript
Java
Kotlin

Restate makes it easy to parallelize async work by fanning out tasks. Afterwards, you can collect the result by fanning in the partial results. Durable Execution ensures that the fan-out and fan-in steps happen reliably exactly once.

fan_out_worker.ts


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();

Fan out

Fan out tasks by calling the subtask handler for each subtask. Every handler is an asynchronous task, for which Restate serves as the queue.

The subtasks might run in different processes, if this is deployed in a parallel setup.

fan_out_worker.ts


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();

Fan in

Invocations produce durable promises that can be awaited and combined. Fan in by simply awaiting the combined promise. Invocation promises recover from failures, re-connect to running subtasks.

fan_out_worker.ts


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();

Server(less)

Deploy this service on an platform like Kubernetes or AWS Lambda to automatically get parallel scale out.

fan_out_worker.ts


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();

Fan out

Fan out tasks by calling the subtask handler for each subtask. Every handler is an asynchronous task, for which Restate serves as the queue.

The subtasks might run in different processes, if this is deployed in a parallel setup.

Fan in

Invocations produce durable promises that can be awaited and combined. Fan in by simply awaiting the combined promise. Invocation promises recover from failures, re-connect to running subtasks.

Server(less)

Deploy this service on an platform like Kubernetes or AWS Lambda to automatically get parallel scale out.

fan_out_worker.ts


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();

FanOutWorker.java


@Service
public class FanOutWorker {
  @Handler
  public Result run(Context ctx, Task task) {
    // Split the task in subtasks
    SubTask[] subTasks = ctx.run(JacksonSerdes.of(new TypeReference<>() {}), () -> split(task));
    List<Awaitable<?>> resultFutures = new ArrayList<>();
    for (SubTask subTask : subTasks) {
      Awaitable<SubTaskResult> subResultFuture =
          FanOutWorkerClient.fromContext(ctx).runSubtask(subTask);
      resultFutures.add(subResultFuture);
    }
    Awaitable.all(resultFutures).await();
    SubTaskResult[] results =
        (SubTaskResult[]) resultFutures.stream().map(Awaitable::await).toArray();
    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 new SubTaskResult();
  }
}

Fan out

Fan out tasks by calling the subtask handler for each subtask. Every handler is an asynchronous task, for which Restate serves as the queue.

The subtasks might run in different processes, if this is deployed in a parallel setup.

FanOutWorker.java


@Service
public class FanOutWorker {
  @Handler
  public Result run(Context ctx, Task task) {
    // Split the task in subtasks
    SubTask[] subTasks = ctx.run(JacksonSerdes.of(new TypeReference<>() {}), () -> split(task));
    List<Awaitable<?>> resultFutures = new ArrayList<>();
    for (SubTask subTask : subTasks) {
      Awaitable<SubTaskResult> subResultFuture =
          FanOutWorkerClient.fromContext(ctx).runSubtask(subTask);
      resultFutures.add(subResultFuture);
    }
    Awaitable.all(resultFutures).await();
    SubTaskResult[] results =
        (SubTaskResult[]) resultFutures.stream().map(Awaitable::await).toArray();
    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 new SubTaskResult();
  }
}

Fan in

Invocations produce durable promises that can be awaited and combined. Fan in by simply awaiting the combined promise. Invocation promises recover from failures, re-connect to running subtasks.

FanOutWorker.java


@Service
public class FanOutWorker {
  @Handler
  public Result run(Context ctx, Task task) {
    // Split the task in subtasks
    SubTask[] subTasks = ctx.run(JacksonSerdes.of(new TypeReference<>() {}), () -> split(task));
    List<Awaitable<?>> resultFutures = new ArrayList<>();
    for (SubTask subTask : subTasks) {
      Awaitable<SubTaskResult> subResultFuture =
          FanOutWorkerClient.fromContext(ctx).runSubtask(subTask);
      resultFutures.add(subResultFuture);
    }
    Awaitable.all(resultFutures).await();
    SubTaskResult[] results =
        (SubTaskResult[]) resultFutures.stream().map(Awaitable::await).toArray();
    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 new SubTaskResult();
  }
}

Server(less)

Deploy this service on an platform like Kubernetes or AWS Lambda to automatically get parallel scale out.

FanOutWorker.java


@Service
public class FanOutWorker {
  @Handler
  public Result run(Context ctx, Task task) {
    // Split the task in subtasks
    SubTask[] subTasks = ctx.run(JacksonSerdes.of(new TypeReference<>() {}), () -> split(task));
    List<Awaitable<?>> resultFutures = new ArrayList<>();
    for (SubTask subTask : subTasks) {
      Awaitable<SubTaskResult> subResultFuture =
          FanOutWorkerClient.fromContext(ctx).runSubtask(subTask);
      resultFutures.add(subResultFuture);
    }
    Awaitable.all(resultFutures).await();
    SubTaskResult[] results =
        (SubTaskResult[]) resultFutures.stream().map(Awaitable::await).toArray();
    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 new SubTaskResult();
  }
}

Fan out

Fan out tasks by calling the subtask handler for each subtask. Every handler is an asynchronous task, for which Restate serves as the queue.

The subtasks might run in different processes, if this is deployed in a parallel setup.

Fan in

Invocations produce durable promises that can be awaited and combined. Fan in by simply awaiting the combined promise. Invocation promises recover from failures, re-connect to running subtasks.

Server(less)

Deploy this service on an platform like Kubernetes or AWS Lambda to automatically get parallel scale out.

FanOutWorker.java


@Service
public class FanOutWorker {
  @Handler
  public Result run(Context ctx, Task task) {
    // Split the task in subtasks
    SubTask[] subTasks = ctx.run(JacksonSerdes.of(new TypeReference<>() {}), () -> split(task));
    List<Awaitable<?>> resultFutures = new ArrayList<>();
    for (SubTask subTask : subTasks) {
      Awaitable<SubTaskResult> subResultFuture =
          FanOutWorkerClient.fromContext(ctx).runSubtask(subTask);
      resultFutures.add(subResultFuture);
    }
    Awaitable.all(resultFutures).await();
    SubTaskResult[] results =
        (SubTaskResult[]) resultFutures.stream().map(Awaitable::await).toArray();
    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 new SubTaskResult();
  }
}

FanOutWorker.kt


@Service
class FanOutWorker {
    @Handler
    suspend fun run(ctx: Context, task: Task): Result {
        val subTasks = ctx.runBlock { split(task) }
        val resultFutures: MutableList<Awaitable<SubTaskResult>> = mutableListOf()
        for (subTask in subTasks) {
            val subResultFuture = FanOutWorkerClient.fromContext(ctx)
                .runSubtask(subTask)
            resultFutures.add(subResultFuture)
        }
        val results = resultFutures.awaitAll()
        return aggregate(results)
    }
    @Handler
    suspend fun runSubtask(ctx: Context?, subTask: SubTask?): SubTaskResult {
        // Processing logic goes here ...
        // Can be moved to a separate service to scale independently
        return SubTaskResult()
    }
}

Fan out

Fan out tasks by calling the subtask handler for each subtask. Every handler is an asynchronous task, for which Restate serves as the queue.

The subtasks might run in different processes, if this is deployed in a parallel setup.

FanOutWorker.kt


@Service
class FanOutWorker {
    @Handler
    suspend fun run(ctx: Context, task: Task): Result {
        val subTasks = ctx.runBlock { split(task) }
        val resultFutures: MutableList<Awaitable<SubTaskResult>> = mutableListOf()
        for (subTask in subTasks) {
            val subResultFuture = FanOutWorkerClient.fromContext(ctx)
                .runSubtask(subTask)
            resultFutures.add(subResultFuture)
        }
        val results = resultFutures.awaitAll()
        return aggregate(results)
    }
    @Handler
    suspend fun runSubtask(ctx: Context?, subTask: SubTask?): SubTaskResult {
        // Processing logic goes here ...
        // Can be moved to a separate service to scale independently
        return SubTaskResult()
    }
}

Fan in

Invocations produce durable promises that can be awaited and combined. Fan in by simply awaiting the combined promise. Invocation promises recover from failures, re-connect to running subtasks.

FanOutWorker.kt


@Service
class FanOutWorker {
    @Handler
    suspend fun run(ctx: Context, task: Task): Result {
        val subTasks = ctx.runBlock { split(task) }
        val resultFutures: MutableList<Awaitable<SubTaskResult>> = mutableListOf()
        for (subTask in subTasks) {
            val subResultFuture = FanOutWorkerClient.fromContext(ctx)
                .runSubtask(subTask)
            resultFutures.add(subResultFuture)
        }
        val results = resultFutures.awaitAll()
        return aggregate(results)
    }
    @Handler
    suspend fun runSubtask(ctx: Context?, subTask: SubTask?): SubTaskResult {
        // Processing logic goes here ...
        // Can be moved to a separate service to scale independently
        return SubTaskResult()
    }
}

Server(less)

Deploy this service on an platform like Kubernetes or AWS Lambda to automatically get parallel scale out.

FanOutWorker.kt


@Service
class FanOutWorker {
    @Handler
    suspend fun run(ctx: Context, task: Task): Result {
        val subTasks = ctx.runBlock { split(task) }
        val resultFutures: MutableList<Awaitable<SubTaskResult>> = mutableListOf()
        for (subTask in subTasks) {
            val subResultFuture = FanOutWorkerClient.fromContext(ctx)
                .runSubtask(subTask)
            resultFutures.add(subResultFuture)
        }
        val results = resultFutures.awaitAll()
        return aggregate(results)
    }
    @Handler
    suspend fun runSubtask(ctx: Context?, subTask: SubTask?): SubTaskResult {
        // Processing logic goes here ...
        // Can be moved to a separate service to scale independently
        return SubTaskResult()
    }
}

Fan out

Fan out tasks by calling the subtask handler for each subtask. Every handler is an asynchronous task, for which Restate serves as the queue.

The subtasks might run in different processes, if this is deployed in a parallel setup.

Fan in

Invocations produce durable promises that can be awaited and combined. Fan in by simply awaiting the combined promise. Invocation promises recover from failures, re-connect to running subtasks.

Server(less)

Deploy this service on an platform like Kubernetes or AWS Lambda to automatically get parallel scale out.

FanOutWorker.kt


@Service
class FanOutWorker {
    @Handler
    suspend fun run(ctx: Context, task: Task): Result {
        val subTasks = ctx.runBlock { split(task) }
        val resultFutures: MutableList<Awaitable<SubTaskResult>> = mutableListOf()
        for (subTask in subTasks) {
            val subResultFuture = FanOutWorkerClient.fromContext(ctx)
                .runSubtask(subTask)
            resultFutures.add(subResultFuture)
        }
        val results = resultFutures.awaitAll()
        return aggregate(results)
    }
    @Handler
    suspend fun runSubtask(ctx: Context?, subTask: SubTask?): SubTaskResult {
        // Processing logic goes here ...
        // Can be moved to a separate service to scale independently
        return SubTaskResult()
    }
}

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.

Switch between async and sync with Restate

TypeScript
Java
Kotlin

Imagine a data preparation workflow that creates an S3 bucket, uploads a file to it, and then returns the URL.

Let's now kick off this workflow from another process.

data_preparation_service.ts

client.ts


const dataPreparationService = restate.workflow({
  name: "dataPrep",
  handlers: {
    run: async (ctx: WorkflowContext, args: { userId: string }) => {
      const url = await ctx.run(() => createS3Bucket());
      await ctx.run(() => uploadData(url));
      await ctx.promise<URL>("url").resolve(url);
      return url;
    },
    resultAsEmail: async (
      ctx: WorkflowSharedContext,
      req: { email: string }
    ) => {
      const url = await ctx.promise<URL>("url");
      await ctx.run(() => sendEmail(url, req.email));
    },
  },
});
export type DataPrepService = typeof dataPreparationService;

Connect to the Restate server and create a client for the data preparation workflow.

data_preparation_service.ts

client.ts


const rs = restate.connect({ url: RESTATE_URL });
const dataPrepService: DataPrepService = { name: "dataPrep" };
async function downloadData(userId: string) {
  const dataPrep = rs.workflowClient(dataPrepService, userId);
  await dataPrep.workflowSubmit({ userId });
  const result = await withTimeout(dataPrep.workflowAttach(), 30_000);
  if (result === Timeout) {
    const email = await readLine("This takes long... Mail us the link later");
    await dataPrep.resultAsEmail({ email });
    return;
  }
  // ... process directly ...
}

Kick off a new data preparation workflow. This is idempotent per workflow ID.

data_preparation_service.ts

client.ts


const rs = restate.connect({ url: RESTATE_URL });
const dataPrepService: DataPrepService = { name: "dataPrep" };
async function downloadData(userId: string) {
  const dataPrep = rs.workflowClient(dataPrepService, userId);
  await dataPrep.workflowSubmit({ userId });
  const result = await withTimeout(dataPrep.workflowAttach(), 30_000);
  if (result === Timeout) {
    const email = await readLine("This takes long... Mail us the link later");
    await dataPrep.resultAsEmail({ email });
    return;
  }
  // ... process directly ...
}

Wait for the result for 30 seconds.

data_preparation_service.ts

client.ts


const rs = restate.connect({ url: RESTATE_URL });
const dataPrepService: DataPrepService = { name: "dataPrep" };
async function downloadData(userId: string) {
  const dataPrep = rs.workflowClient(dataPrepService, userId);
  await dataPrep.workflowSubmit({ userId });
  const result = await withTimeout(dataPrep.workflowAttach(), 30_000);
  if (result === Timeout) {
    const email = await readLine("This takes long... Mail us the link later");
    await dataPrep.resultAsEmail({ email });
    return;
  }
  // ... process directly ...
}

If it takes longer, rewire the workflow to send an email instead. If returns within 30 seconds, process the URL directly.

data_preparation_service.ts

client.ts


const rs = restate.connect({ url: RESTATE_URL });
const dataPrepService: DataPrepService = { name: "dataPrep" };
async function downloadData(userId: string) {
  const dataPrep = rs.workflowClient(dataPrepService, userId);
  await dataPrep.workflowSubmit({ userId });
  const result = await withTimeout(dataPrep.workflowAttach(), 30_000);
  if (result === Timeout) {
    const email = await readLine("This takes long... Mail us the link later");
    await dataPrep.resultAsEmail({ email });
    return;
  }
  // ... process directly ...
}

This is implemented in the data preparation workflow by letting the workflow signal our handler when it's done. It does this by resolving a shared Durable Promise that we then retrieve in our handler to send the email.

data_preparation_service.ts

client.ts


const dataPreparationService = restate.workflow({
  name: "dataPrep",
  handlers: {
    run: async (ctx: WorkflowContext, args: { userId: string }) => {
      const url = await ctx.run(() => createS3Bucket());
      await ctx.run(() => uploadData(url));
      await ctx.promise<URL>("url").resolve(url);
      return url;
    },
    resultAsEmail: async (
      ctx: WorkflowSharedContext,
      req: { email: string }
    ) => {
      const url = await ctx.promise<URL>("url");
      await ctx.run(() => sendEmail(url, req.email));
    },
  },
});
export type DataPrepService = typeof dataPreparationService;

Imagine a data preparation workflow that creates an S3 bucket, uploads a file to it, and then returns the URL.

Let's now kick off this workflow from another process.

Connect to the Restate server and create a client for the data preparation workflow.

Kick off a new data preparation workflow. This is idempotent per workflow ID.

Wait for the result for 30 seconds.

If it takes longer, rewire the workflow to send an email instead. If returns within 30 seconds, process the URL directly.

This is implemented in the data preparation workflow by letting the workflow signal our handler when it's done. It does this by resolving a shared Durable Promise that we then retrieve in our handler to send the email.

data_preparation_service.ts

client.ts


const dataPreparationService = restate.workflow({
  name: "dataPrep",
  handlers: {
    run: async (ctx: WorkflowContext, args: { userId: string }) => {
      const url = await ctx.run(() => createS3Bucket());
      await ctx.run(() => uploadData(url));
      await ctx.promise<URL>("url").resolve(url);
      return url;
    },
    resultAsEmail: async (
      ctx: WorkflowSharedContext,
      req: { email: string }
    ) => {
      const url = await ctx.promise<URL>("url");
      await ctx.run(() => sendEmail(url, req.email));
    },
  },
});
export type DataPrepService = typeof dataPreparationService;

Imagine a data preparation workflow that creates an S3 bucket, uploads a file to it, and then returns the URL.

Let's now kick off this workflow from another process.

DataPreparationService.java

MyClient.java


@Workflow
public class DataPreparationService {
  private static final DurablePromiseKey<URL> URL_PROMISE =
      DurablePromiseKey.of("url", JacksonSerdes.of(URL.class));
  @Workflow
  public URL run(WorkflowContext ctx, String userId) {
    URL url = ctx.run(JacksonSerdes.of(URL.class), () -> createS3Bucket());
    ctx.run(() -> uploadData(url));
    ctx.promiseHandle(URL_PROMISE).resolve(url);
    return url;
  }
  @Shared
  public void resultAsEmail(SharedWorkflowContext ctx, Email email) {
    URL url = ctx.promise(URL_PROMISE).awaitable().await();
    ctx.run(() -> sendEmail(url, email));
  }
}

Connect to the Restate server and create a client for the data preparation workflow.

DataPreparationService.java

MyClient.java


public class MyClient {
  private static final Client rs = Client.connect("http://localhost:8080");
  public void downloadData(String userId, Email email) {
    IngressClient client = DataPreparationServiceClient.fromClient(rs, userId);
    client.submit(userId);
    try {
      CompletableFuture.anyOf(client.workflowHandle().attachAsync())
          .orTimeout(30, TimeUnit.SECONDS)
          .join();
    } catch (Exception e) {
      client.resultAsEmail(email);
      return;
    }
    // ... process directly ...
  }
}

Kick off a new data preparation workflow. This is idempotent per workflow ID.

DataPreparationService.java

MyClient.java


public class MyClient {
  private static final Client rs = Client.connect("http://localhost:8080");
  public void downloadData(String userId, Email email) {
    IngressClient client = DataPreparationServiceClient.fromClient(rs, userId);
    client.submit(userId);
    try {
      CompletableFuture.anyOf(client.workflowHandle().attachAsync())
          .orTimeout(30, TimeUnit.SECONDS)
          .join();
    } catch (Exception e) {
      client.resultAsEmail(email);
      return;
    }
    // ... process directly ...
  }
}

Wait for the result for 30 seconds.

DataPreparationService.java

MyClient.java


public class MyClient {
  private static final Client rs = Client.connect("http://localhost:8080");
  public void downloadData(String userId, Email email) {
    IngressClient client = DataPreparationServiceClient.fromClient(rs, userId);
    client.submit(userId);
    try {
      CompletableFuture.anyOf(client.workflowHandle().attachAsync())
          .orTimeout(30, TimeUnit.SECONDS)
          .join();
    } catch (Exception e) {
      client.resultAsEmail(email);
      return;
    }
    // ... process directly ...
  }
}

If it takes longer, rewire the workflow to send an email instead. If returns within 30 seconds, process the URL directly.

DataPreparationService.java

MyClient.java


public class MyClient {
  private static final Client rs = Client.connect("http://localhost:8080");
  public void downloadData(String userId, Email email) {
    IngressClient client = DataPreparationServiceClient.fromClient(rs, userId);
    client.submit(userId);
    try {
      CompletableFuture.anyOf(client.workflowHandle().attachAsync())
          .orTimeout(30, TimeUnit.SECONDS)
          .join();
    } catch (Exception e) {
      client.resultAsEmail(email);
      return;
    }
    // ... process directly ...
  }
}

This is implemented in the data preparation workflow by letting the workflow signal our handler when it's done. It does this by resolving a shared Durable Promise that we then retrieve in our handler to send the email.

DataPreparationService.java

MyClient.java


@Workflow
public class DataPreparationService {
  private static final DurablePromiseKey<URL> URL_PROMISE =
      DurablePromiseKey.of("url", JacksonSerdes.of(URL.class));
  @Workflow
  public URL run(WorkflowContext ctx, String userId) {
    URL url = ctx.run(JacksonSerdes.of(URL.class), () -> createS3Bucket());
    ctx.run(() -> uploadData(url));
    ctx.promiseHandle(URL_PROMISE).resolve(url);
    return url;
  }
  @Shared
  public void resultAsEmail(SharedWorkflowContext ctx, Email email) {
    URL url = ctx.promise(URL_PROMISE).awaitable().await();
    ctx.run(() -> sendEmail(url, email));
  }
}

Imagine a data preparation workflow that creates an S3 bucket, uploads a file to it, and then returns the URL.

Let's now kick off this workflow from another process.

Connect to the Restate server and create a client for the data preparation workflow.

Kick off a new data preparation workflow. This is idempotent per workflow ID.

Wait for the result for 30 seconds.

If it takes longer, rewire the workflow to send an email instead. If returns within 30 seconds, process the URL directly.

This is implemented in the data preparation workflow by letting the workflow signal our handler when it's done. It does this by resolving a shared Durable Promise that we then retrieve in our handler to send the email.

DataPreparationService.java

MyClient.java


@Workflow
public class DataPreparationService {
  private static final DurablePromiseKey<URL> URL_PROMISE =
      DurablePromiseKey.of("url", JacksonSerdes.of(URL.class));
  @Workflow
  public URL run(WorkflowContext ctx, String userId) {
    URL url = ctx.run(JacksonSerdes.of(URL.class), () -> createS3Bucket());
    ctx.run(() -> uploadData(url));
    ctx.promiseHandle(URL_PROMISE).resolve(url);
    return url;
  }
  @Shared
  public void resultAsEmail(SharedWorkflowContext ctx, Email email) {
    URL url = ctx.promise(URL_PROMISE).awaitable().await();
    ctx.run(() -> sendEmail(url, email));
  }
}

Imagine a data preparation workflow that creates an S3 bucket, uploads a file to it, and then returns the URL.

Let's now kick off this workflow from another process.

DataPreparationService.kt

MyClient.kt


@Workflow
class DataPreparationService {
    companion object {
        private val URL_PROMISE = DurablePromiseKey.of("url", KtSerdes.json<URL>())
    }
    @Workflow
    suspend fun run(ctx: WorkflowContext, userId: String): URL {
        val url: URL = ctx.runBlock { createS3Bucket() }
        ctx.runBlock { uploadData(url) }
        ctx.promiseHandle(URL_PROMISE).resolve(url)
        return url
    }
    @Shared
    suspend fun resultAsEmail(ctx: SharedWorkflowContext, email: Email) {
        val url: URL = ctx.promise(URL_PROMISE).awaitable().await()
        ctx.runBlock { sendEmail(url, email) }
    }
}

Connect to the Restate server and create a client for the data preparation workflow.

DataPreparationService.kt

MyClient.kt


class MyClient {
    companion object {
        private val rs: Client = Client.connect("http://localhost:8080")
    }
    suspend fun downloadData(userId: String, email: Email) {
        val client = DataPreparationServiceClient.fromClient(rs, userId)
        client.submit(userId)
        try {
            client.workflowHandle()
                .attachAsync()
                .orTimeout(30, TimeUnit.SECONDS)
                .join()
        } catch (e: Exception) {
            client.resultAsEmail(email)
            return
        }
        // ... process directly ...
    }
}

Kick off a new data preparation workflow. This is idempotent per workflow ID.

DataPreparationService.kt

MyClient.kt


class MyClient {
    companion object {
        private val rs: Client = Client.connect("http://localhost:8080")
    }
    suspend fun downloadData(userId: String, email: Email) {
        val client = DataPreparationServiceClient.fromClient(rs, userId)
        client.submit(userId)
        try {
            client.workflowHandle()
                .attachAsync()
                .orTimeout(30, TimeUnit.SECONDS)
                .join()
        } catch (e: Exception) {
            client.resultAsEmail(email)
            return
        }
        // ... process directly ...
    }
}

Wait for the result for 30 seconds.

DataPreparationService.kt

MyClient.kt


class MyClient {
    companion object {
        private val rs: Client = Client.connect("http://localhost:8080")
    }
    suspend fun downloadData(userId: String, email: Email) {
        val client = DataPreparationServiceClient.fromClient(rs, userId)
        client.submit(userId)
        try {
            client.workflowHandle()
                .attachAsync()
                .orTimeout(30, TimeUnit.SECONDS)
                .join()
        } catch (e: Exception) {
            client.resultAsEmail(email)
            return
        }
        // ... process directly ...
    }
}

If it takes longer, rewire the workflow to send an email instead. If returns within 30 seconds, process the URL directly.

DataPreparationService.kt

MyClient.kt


class MyClient {
    companion object {
        private val rs: Client = Client.connect("http://localhost:8080")
    }
    suspend fun downloadData(userId: String, email: Email) {
        val client = DataPreparationServiceClient.fromClient(rs, userId)
        client.submit(userId)
        try {
            client.workflowHandle()
                .attachAsync()
                .orTimeout(30, TimeUnit.SECONDS)
                .join()
        } catch (e: Exception) {
            client.resultAsEmail(email)
            return
        }
        // ... process directly ...
    }
}

This is implemented in the data preparation workflow by letting the workflow signal our handler when it's done. It does this by resolving a shared Durable Promise that we then retrieve in our handler to send the email.

DataPreparationService.kt

MyClient.kt


@Workflow
class DataPreparationService {
    companion object {
        private val URL_PROMISE = DurablePromiseKey.of("url", KtSerdes.json<URL>())
    }
    @Workflow
    suspend fun run(ctx: WorkflowContext, userId: String): URL {
        val url: URL = ctx.runBlock { createS3Bucket() }
        ctx.runBlock { uploadData(url) }
        ctx.promiseHandle(URL_PROMISE).resolve(url)
        return url
    }
    @Shared
    suspend fun resultAsEmail(ctx: SharedWorkflowContext, email: Email) {
        val url: URL = ctx.promise(URL_PROMISE).awaitable().await()
        ctx.runBlock { sendEmail(url, email) }
    }
}

Imagine a data preparation workflow that creates an S3 bucket, uploads a file to it, and then returns the URL.

Let's now kick off this workflow from another process.

Connect to the Restate server and create a client for the data preparation workflow.

Kick off a new data preparation workflow. This is idempotent per workflow ID.

Wait for the result for 30 seconds.

If it takes longer, rewire the workflow to send an email instead. If returns within 30 seconds, process the URL directly.

This is implemented in the data preparation workflow by letting the workflow signal our handler when it's done. It does this by resolving a shared Durable Promise that we then retrieve in our handler to send the email.

DataPreparationService.kt

MyClient.kt


@Workflow
class DataPreparationService {
    companion object {
        private val URL_PROMISE = DurablePromiseKey.of("url", KtSerdes.json<URL>())
    }
    @Workflow
    suspend fun run(ctx: WorkflowContext, userId: String): URL {
        val url: URL = ctx.runBlock { createS3Bucket() }
        ctx.runBlock { uploadData(url) }
        ctx.promiseHandle(URL_PROMISE).resolve(url)
        return url
    }
    @Shared
    suspend fun resultAsEmail(ctx: SharedWorkflowContext, email: Email) {
        val url: URL = ctx.promise(URL_PROMISE).awaitable().await()
        ctx.runBlock { sendEmail(url, email) }
    }
}

What you can build with Async Tasks and Restate

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.

Job scheduler

A job scheduler that can handle both immediate and delayed jobs. The scheduler makes sure job submissions are idempotent and that jobs run to completion exactly once. You can attach back to jobs to retrieve their result later.

Building Durable Promises on top of Restate

Promises that get durably stored in Restate, and can be resolved by any process. Implemented as a Restate service, to use for callbacks, signal and communicate between systems. The SDK offers similar constructs with Awakeables and Durable Promises.

Wondering about a specific use case?

Let’s brainstorm together on Discord

Developer Resources

Docs

Read the docs to learn more.

Schedulers & timers TS / Java/Kotlin
Client library / HTTP

Need help?

Join the Restate Discord channel

Async tasks

Flexible, durable scheduling across processes and time.

Durable timers

Message queue

Switch between async and sync

Async tasks with Restate​

Execute any handler async​

Schedule tasks reliably​

Idempotent task submission​

Latch on to the task​

Execute any handler async​

Schedule tasks reliably​

Idempotent task submission​

Latch on to the task​

Execute any handler async​

Schedule tasks reliably​

Idempotent task submission​

Latch on to the task​

Execute any handler async​

Schedule tasks reliably​

Idempotent task submission​

Latch on to the task​

Execute any handler async​

Schedule tasks reliably​

Idempotent task submission​

Latch on to the task​

Execute any handler async​

Schedule tasks reliably​

Idempotent task submission​

Latch on to the task​

LOW-LATENCY

DURABLE EXECUTION

Parallelizing work with Restate​

Fan out​

Fan in​

Server(less)​

Fan out​

Fan in​

Server(less)​

Fan out​

Fan in​

Server(less)​

Fan out​

Fan in​

Server(less)​

Fan out​

Fan in​

Server(less)​

Fan out​

Fan in​

Server(less)​

Restate as sophisticated task queue

Switch between async and sync with Restate​

What you can build with Async Tasks and Restate​

Payments: Combining sync & async responses

Job scheduler

Building Durable Promises on top of Restate

Wondering about a specific use case?

Developer Resources​

Docs

Need help?

Async tasks with Restate

Execute any handler async

Schedule tasks reliably

Idempotent task submission

Latch on to the task

Execute any handler async

Schedule tasks reliably

Idempotent task submission

Latch on to the task

Execute any handler async

Schedule tasks reliably

Idempotent task submission

Latch on to the task

Execute any handler async

Schedule tasks reliably

Idempotent task submission

Latch on to the task

Execute any handler async

Schedule tasks reliably

Idempotent task submission

Latch on to the task

Execute any handler async

Schedule tasks reliably

Idempotent task submission

Latch on to the task

Parallelizing work with Restate

Fan out

Fan in

Server(less)

Fan out

Fan in

Server(less)

Fan out

Fan in

Server(less)

Fan out

Fan in

Server(less)

Fan out

Fan in

Server(less)

Fan out

Fan in

Server(less)

Switch between async and sync with Restate

What you can build with Async Tasks and Restate

Developer Resources