# Result Backend (/integrations/celery/how-to/result-backend)
By default a Celery worker on KubeMQ runs with results disabled — tasks execute, but their return values are discarded. To capture and read those return values, enable the KubeMQ **queue-peek result backend**. It stores each result as a message on KubeMQ [Queues](/learn/queues) and reads it back with a non-destructive peek, so you get a pure-KubeMQ stack with no external Redis instance or database to operate.
## Enable the Result Backend [#enable-the-result-backend]
Add `result_backend="kubemq://localhost:50000"` alongside your broker URL. Both the broker and the backend can point at the same KubeMQ instance, so one broker serves the entire stack.
```python title="tasks.py"
import kubemq_celery # registers the kubemq:// transport
from celery import Celery
app = Celery(
"myapp",
broker="kubemq://localhost:50000",
result_backend="kubemq://localhost:50000",
)
@app.task
def add(x, y):
return x + y
```
You need a reachable KubeMQ broker. For local development, run one in Docker:
Port `50000` is the gRPC port the transport and result backend connect over. Port `9090` is the shared HTTP server (REST and the `/health` probe) — `curl http://localhost:9090/health` confirms the broker is up. The result backend talks to KubeMQ over native gRPC — no HTTP connector needs to be enabled.
`import kubemq_celery` must run before Celery resolves the URLs — it registers the `kubemq://` scheme with Kombu for both the broker and the result backend. Without it, Celery raises an "unknown transport" error.
## How It Works [#how-it-works]
Each task result is written to its own KubeMQ Queue channel named `celery-result-{task_id}`. Retrieval uses `peek_queue_messages()`, a **non-destructive** read — the message stays on the channel after it is read, so multiple callers can fetch the same result independently without one consumer stealing it from another.
This is what lets a web request handler and a background poller both read the outcome of the same task, and it is also how Celery's group and chord machinery polls for completion.
## Store and Retrieve a Result [#store-and-retrieve-a-result]
With the backend enabled, dispatch a task with `.delay()` and block for its return value with `result.get()`:
```python title="store_and_retrieve.py"
from tasks import add
result = add.delay(4, 6)
print(result.id) # the task_id, also the result channel suffix
print(result.get(timeout=10)) # 10
print(result.state) # SUCCESS
```
Because retrieval is a peek, you can re-read the same result later from anywhere that knows the task ID — construct an `AsyncResult` and call `.get()` again:
```python title="second_reader.py"
from celery.result import AsyncResult
from tasks import app
second_reader = AsyncResult(task_id, app=app)
print(second_reader.get(timeout=10)) # same value, message not consumed
```
## Result Expiration [#result-expiration]
`result_expires` controls how long a stored result lives before KubeMQ removes it. KubeMQ caps message expiration at **86400 seconds (24 hours)** — this is a KubeMQ limitation, and any larger value is silently capped. Celery's own default `result_expires` is also 24 hours, so the two line up with no surprises.
```python title="result_expiration.py"
app.conf.update(
result_backend="kubemq://localhost:50000",
result_expires=86400, # 24 hours — the KubeMQ maximum
)
```
`result_expires` accepts either an integer number of seconds or a `timedelta`:
```python title="timedelta_expires.py"
from datetime import timedelta
app.conf.result_expires = timedelta(hours=1) # 3600 seconds
```
The table below summarizes how each value maps to an effective expiration:
| `result_expires` | Effective expiration |
| ----------------------- | -------------------- |
| `3600` | 1 hour |
| `timedelta(minutes=30)` | 30 minutes |
| `86400` | 24 hours (maximum) |
| `172800` (48 hours) | Capped to 24 hours |
| `None` or `0` | Defaults to 24 hours |
Results are stored with KubeMQ's `expiration_in_seconds`, so they vanish from the channel once expired. After expiration, `result.get()` sees no message and the task reads back as `PENDING`. If you need result retention beyond 24 hours, KubeMQ's result backend is not the right fit — see the [troubleshooting guide](/integrations/celery/how-to/troubleshooting) for when Redis may still be appropriate.
## Result Backend Transport Options [#result-backend-transport-options]
The result backend opens its own KubeMQ client, configured independently of the broker via `result_backend_transport_options`. Use it to set authentication and TLS for the backend connection.
```python title="result_backend_options.py"
app.conf.update(
result_backend="kubemq://localhost:50000",
result_expires=3600,
result_backend_transport_options={
"auth_token": "my-token",
"tls_enabled": False,
"tls_cert_file": "/path/to/cert.pem",
"tls_key_file": "/path/to/key.pem",
"tls_ca_file": "/path/to/ca.pem",
},
)
```
| Option | Type | Description |
| --------------- | ------ | --------------------------------------------------------------------------------------------------- |
| `auth_token` | `str` | KubeMQ authentication token for the result backend client (alternative to embedding it in the URL). |
| `tls_enabled` | `bool` | Enable TLS for the result backend gRPC connection. |
| `tls_cert_file` | `str` | Path to the client certificate for mTLS. |
| `tls_key_file` | `str` | Path to the client private key for mTLS. |
| `tls_ca_file` | `str` | Path to the CA certificate for custom certificate authority verification. |
## Task States [#task-states]
The result backend tracks a task across its lifecycle: `PENDING → STARTED → SUCCESS`. Each transition **purges and rewrites** the message on the `celery-result-{task_id}` channel, so a peek always returns the latest state rather than a backlog of historical entries.
To observe the `STARTED` state, enable `task_track_started`. Bound tasks can also publish a custom `PROGRESS` state with `self.update_state()`, which is written to the same channel:
```python title="task_states.py"
app.conf.update(
result_backend="kubemq://localhost:50000",
task_track_started=True, # required for the STARTED state
)
@app.task(bind=True)
def process_items(self, items: list[str]) -> dict:
total = len(items)
for i, item in enumerate(items):
self.update_state(
state="PROGRESS",
meta={"current": i + 1, "total": total},
)
# ... process item ...
return {"processed": total, "status": "completed"}
```
Clients poll the live state from the same `AsyncResult`:
```python title="poll_state.py"
result = process_items.delay(["alpha", "beta", "gamma"])
while not result.ready():
print(result.state, result.info) # PROGRESS, {"current": .., "total": ..}
print(result.get(timeout=10)) # final SUCCESS result
```
## Fire-and-Forget Tasks [#fire-and-forget-tasks]
For tasks whose return value nobody reads — notifications, logging, cleanup — set `ignore_result=True`. The worker skips the backend entirely, so no `celery-result-{task_id}` channel is created and you avoid the overhead of writing a result that would only be discarded.
```python title="ignore_result.py"
@app.task(ignore_result=True)
def send_notification(user_id: str, message: str) -> None:
# No celery-result-{task_id} channel is created.
# The caller cannot retrieve a return value.
print(f"Notification sent to {user_id}: {message}")
```
A task marked `ignore_result=True` always reports `result = None` and stays in the `PENDING` state from the caller's perspective — never call `.get()` on it, as there is no result message to read.
Use `ignore_result=True` for high-volume fire-and-forget work to reduce KubeMQ channel churn. Keep the backend for tasks whose outcome a caller, group, or chord needs to read.
## Group Results and Custom Serializers [#group-results-and-custom-serializers]
Group results work the same way. Dispatch a `group` of tasks in parallel and collect every return value with a single `.get()`; group metadata is stored on `celery-group-{group_id}` channels.
```python title="group_results.py"
from celery import chord, group
from tasks import square, aggregate
# Group: run square(1..5) in parallel and collect all results
g = group(square.s(i) for i in range(1, 6))
print(g.apply_async().get(timeout=60)) # [1, 4, 9, 16, 25]
# Chord: run the group, then call aggregate(results) once all finish
c = chord([square.s(i) for i in range(1, 5)], aggregate.s())
print(c.apply_async().get(timeout=60))
```
Results are serialized with Celery's `result_serializer`. The default `json` is human-readable and cross-language; switch to `pickle` for full Python-type support or `msgpack` for a compact binary encoding. Workers and clients must agree on `accept_content`.
```python title="custom_result_serializer.py"
app.conf.update(
result_serializer="json", # or "pickle", "msgpack"
task_serializer="json",
accept_content=["json"],
)
```
`pickle` can execute arbitrary code on deserialization — only use it with fully trusted producers and consumers. Prefer `json` (default) or `msgpack` for untrusted boundaries.
## Chord Dependency [#chord-dependency]
KubeMQ does not provide native chord support, so `kubemq-celery` falls back to Celery's polling mechanism: the `chord_unlock` task repeatedly peeks the group result channels until every member completes, then fires the callback. This fallback **requires the result backend to be configured** — a chord with no result backend has nothing to poll and will never fire its callback.
If you use chords (or groups whose completion you depend on), the result backend is mandatory, not optional. See [Canvas Workflows](/integrations/celery/how-to/canvas-workflows) for composing chains, groups, and chords, and the [troubleshooting guide](/integrations/celery/how-to/troubleshooting) if `chord_unlock` appears to hang.
## Related [#related]