Handle request cancellations

A caller can cancel a request that is in progress or waiting in the queue. If the request is still in the queue, it will never be processed. If the request is in progress, the fal platform makes a request to the runner that is processing the request to cancel it.

Handling cancellations

A cancellation is received in the same endpoint that is processing the request with a /cancel suffix.

So for the endpoint /predict you will receive a request to /predict/cancel.

Request id are passed in the x-fal-request-id header for both the request and the cancellation.

Example

import asyncio

import fal
from fal.exceptions import RequestCancelledException
import fastapi
from pydantic import BaseModel


class HelloInput(BaseModel):
    name: str


class MyApp(fal.App):
    _tasks: dict[str, asyncio.Task] = {}

    async def _predict(self, input: HelloInput):
        await asyncio.sleep(10)
        return "Hello, " + input.name

    @fal.endpoint("/predict")
    async def predict(
        self,
        input: HelloInput,
        x_fal_request_id: str | None = fastapi.Header(None),
    ):
        if not x_fal_request_id:
            raise ValueError("x-fal-request-id is required")

        task = asyncio.create_task(self._predict(input))
        self._tasks[x_fal_request_id] = task
        try:
            return await task
        except asyncio.CancelledError:
            # Run any cancellation logic here to leave the app in good state
            print("Request cancelled")
            raise RequestCancelledException("Request cancelled")
        finally:
            self._tasks.pop(x_fal_request_id)

    @fal.endpoint("/predict/cancel")
    async def predict_cancel(
        self,
        x_fal_request_id: str | None = fastapi.Header(None),
    ):
        if not x_fal_request_id:
            raise ValueError("x-fal-request-id is required")

        if x_fal_request_id not in self._tasks:
            raise ValueError("No task to cancel")

        if self._tasks[x_fal_request_id].done():
            raise ValueError("Task is already done")

        print(f"Cancelling request {x_fal_request_id}")
        self._tasks[x_fal_request_id].cancel()