# Cancellation (/integrations/rayserve/how-to/cancellation)



## Overview [#overview]

Cancellation lets a producer mark an enqueued task as no longer needed. `cancel_task_sync(task_id)`
is a **soft** cancel: it overwrites the task's stored result with a `CANCELLED` status and always
returns `True`. It does not interrupt a handler that is already running — the work still completes
on the consumer — but the producer-visible result reads `CANCELLED`. Because the result lives in the
queue-peek backend, cancellation is just another write to that result channel → see
[Queues](/learn/queues).

<Callout type="info">
  Soft cancel is **best-effort**. If the handler finishes and stores its outcome *after* the cancel
  write lands, the later write wins, so the task can still end `SUCCESS`. There is no signal sent to
  the consumer to stop the in-progress handler.
</Callout>

## The cancel\_task\_sync method [#the-cancel_task_sync-method]

`cancel_task_sync` writes a `CANCELLED` result record for the given `task_id` and returns `True`.
The `cancel_task` alias points at the same implementation.

```python title="signature"
def cancel_task_sync(self, task_id: str) -> bool:
    ...
```

| Argument  | Type  | Description                                                    |
| --------- | ----- | -------------------------------------------------------------- |
| `task_id` | `str` | The id returned by `enqueue_task_sync` for the task to cancel. |

The call resolves the task's result channel (`{result_channel_prefix}{task_id}`) and stores a record
with `status="CANCELLED"`, using the same purge-then-write path the result backend uses for every
state transition. A subsequent `get_task_status_sync(task_id)` reads `CANCELLED` — unless the handler
already stored a terminal result first.

## Cancel a running task [#cancel-a-running-task]

Enqueue a long-running task, cancel it before the handler finishes, and confirm the result reads
`CANCELLED`. This mirrors the packaged `examples/cancel_task/soft_cancel.py`.

```python title="soft_cancel.py"
from __future__ import annotations

import os
import time
import uuid

from kubemq_rayserve import KubeMQAdapterConfig, KubeMQTaskProcessorAdapter

BROKER = os.environ.get("KUBEMQ_ADDRESS", "localhost:50000")


def slow_handler(data: str) -> dict:
    """Slow handler to give time for cancellation."""
    time.sleep(10)
    return {"data": data, "completed": True}


def main():
    channel = f"example-cancel-{uuid.uuid4().hex[:8]}"

    config = KubeMQAdapterConfig(address=BROKER)
    adapter = KubeMQTaskProcessorAdapter(config)

    class _Cfg:
        queue_name = channel
        max_retries = 0
        failed_task_queue_name = ""
        unprocessable_task_queue_name = ""

    adapter.initialize(consumer_concurrency=1, task_processor_config=_Cfg())
    adapter.register_task_handler(slow_handler, name="slow_handler")
    adapter.start_consumer()

    try:
        # Enqueue a long-running task
        result = adapter.enqueue_task_sync("slow_handler", args=["work-item"])
        task_id = result.id

        # Cancel before it completes — overwrites the result with CANCELLED
        cancelled = adapter.cancel_task(task_id)
        print(f"cancel_task() returned: {cancelled}")  # True

        status = adapter.get_task_status_sync(task_id)
        print(f"status={status.status}")  # CANCELLED (unless the handler already finished)
    finally:
        adapter.stop_consumer()


if __name__ == "__main__":
    main()
```

`cancel_task()` returns `True` immediately. The 10-second handler keeps running on the consumer, but
the producer's next poll reads `CANCELLED` because the cancel write replaced the `PENDING` record.

## Observe the status transition [#observe-the-status-transition]

To watch the full lifecycle, poll the task before and after the cancel. The
`examples/cancel_task/cancel_and_check_status.py` example drives a `PENDING → CANCELLED` transition
and confirms it by polling.

```python title="cancel_and_check_status.py"
# 1. Enqueue a 15-second task
result = adapter.enqueue_task_sync("slow_handler", args=["work-item"])
task_id = result.id

# 2. Confirm the initial status
status = adapter.get_task_status_sync(task_id)
print(f"initial status={status.status}")  # PENDING

# 3. Cancel
cancelled = adapter.cancel_task(task_id)
print(f"cancel_task() returned: {cancelled}")  # True

# 4. Poll until CANCELLED (or a terminal status if the handler won the race)
for i in range(10):
    status = adapter.get_task_status_sync(task_id)
    if status.status == "CANCELLED":
        print("Task confirmed as CANCELLED.")
        break
    if status.status in ("SUCCESS", "FAILURE"):
        print(f"Task reached {status.status} before cancel took effect.")
        break
    time.sleep(0.5)
```

The polling loop also guards against the race: if the handler stored `SUCCESS` or `FAILURE` before
the cancel write landed, the loop reports that terminal status instead of `CANCELLED`.

<Callout type="warn">
  Cancellation only changes what the producer reads — it never reclaims the compute already spent on
  an in-flight handler. For tasks where stopping the work matters, keep handlers short and rely on the
  [retry and DLQ controls](/integrations/rayserve/how-to/dlq-and-retries) to bound how long a single task can run.
</Callout>

## Next steps [#next-steps]

<Cards>
  <Card title="Async Inference" href="/integrations/rayserve/how-to/async-inference" description="Enqueue tasks onto KubeMQ Queues and poll the queue-peek result backend for outcomes." />

  <Card title="Retries & DLQ" href="/integrations/rayserve/how-to/dlq-and-retries" description="Bound retries, route exhausted tasks to a dead-letter queue, and alert with on_dlq." />

  <Card title="API Reference" href="/integrations/rayserve/reference/api" description="The cancel_task_sync signature and the full TaskResult status model." />
</Cards>