# Sync Inference (/integrations/rayserve/how-to/sync-inference)
## Overview [#overview]
Sync inference is for low-latency, request-response calls where the caller blocks until the result is ready, instead of enqueueing a task and polling for it. It is built on **[KubeMQ Queries](/learn/rpc)** through the SDK `CQClient` — a feature the Celery `TaskProcessorAdapter` does not offer.
Use sync inference when:
* A single request needs an immediate answer (an HTTP handler waiting on a model prediction, an interactive UI, a synchronous RPC).
* The expected handler duration is short enough to wait on inline.
* You want to avoid the polling-interval overhead that async inference incurs.
Prefer [async inference](/integrations/rayserve/how-to/async-inference) for fire-and-forget submission, batch workloads, or long-running jobs where blocking a caller is wasteful.
Sync inference runs over KubeMQ Queries, a separate channel from the async task queue. The same `start_consumer()` call serves both paths, so a single deployed adapter handles enqueued tasks and sync queries concurrently.
The following diagram shows the sync path. `query_task_sync` sends a Query through the `CQClient`; the query subscription thread on the server dispatches it to the registered handler and returns the response in line — no queue, no result backend, no polling.
## How It Works [#how-it-works]
### The `query_task_sync` method [#the-query_task_sync-method]
`query_task_sync(task_name, args, kwargs, timeout)` serializes a sync query payload of the form `{task_name, args, kwargs}`, sends a `QueryMessage` with `timeout_in_seconds`, blocks until the response or timeout, and returns a `TaskResult(status, result)`.
```python title="signature"
def query_task_sync(
self,
task_name: str,
args: Any = None,
kwargs: Any = None,
timeout: int = 30,
**options: Any,
) -> TaskResult:
...
```
The wire payload contains only `task_name`, `args`, and `kwargs` — no `task_id` and no `created_at`, because the response travels back through the Query channel rather than being stored.
```json title="sync query payload"
{
"task_name": "greet",
"args": ["World"],
"kwargs": {}
}
```
### Server side [#server-side]
`start_consumer()` starts the queue-polling consumer thread **and** a Query subscription. The subscription opens a `QueriesSubscription` on the adapter's `queue_name` and routes every incoming query through `_handle_sync_query`, which:
1. Deserializes the query body into `task_name`, `args`, and `kwargs`.
2. Looks up the registered handler by `task_name`.
3. Calls `handler(*args, **kwargs)` and serializes the return value as a `SUCCESS` response.
4. Sends a `QueryResponse` back through the `CQClient`.
If the handler name is unknown, or the handler raises, the response carries `status="FAILURE"` and an error string instead of crashing the subscription:
```python title="server-side dispatch (adapter.py)"
handler = self._task_handlers.get(task_name)
if handler is None:
response_body = serialize_sync_response(
"FAILURE", error=f"Unknown task handler: {task_name}"
)
else:
try:
result = handler(*args, **kwargs)
response_body = serialize_sync_response("SUCCESS", result=result)
except Exception as exc:
response_body = serialize_sync_response(
"FAILURE", error=f"{type(exc).__name__}: {exc}"
)
response = QueryResponse(
query_received=query,
body=response_body,
is_executed=True,
)
self._cq_client.send_response_message(response)
```
Sync responses are **not** stored in the result backend. The handler's return value is encoded directly into the `QueryResponse` body and returned in line through the Query channel — so there is nothing to poll and nothing to expire. This is the key difference from [async inference](/integrations/rayserve/how-to/async-inference), where results are written to the queue-peek backend and read back with `get_task_status_sync`.
## Prerequisites [#prerequisites]
Sync inference needs a running KubeMQ broker (default address `localhost:50000`). Start one locally with Docker:
Port `50000` is KubeMQ's gRPC endpoint used by the adapter's SDK clients. Port `9090` exposes the shared HTTP server used by connector endpoints such as KEDA scaling. Install the adapter:
```bash title="terminal"
uv pip install kubemq-rayserve
```
## Minimal Example [#minimal-example]
Register a `greet` handler, send one sync query with `query_task_sync`, and print the result. No polling loop is needed — the call blocks and returns the answer directly.
```python title="sync_hello_world.py"
from __future__ import annotations
import os
import uuid
from kubemq_rayserve import KubeMQAdapterConfig, KubeMQTaskProcessorAdapter
BROKER = os.environ.get("KUBEMQ_ADDRESS", "localhost:50000")
def greet(name: str) -> dict:
return {"greeting": f"Hello, {name}!"}
def main():
channel = f"example-quickstart-{uuid.uuid4().hex[:8]}"
config = KubeMQAdapterConfig(address=BROKER, sync_inference_timeout=10)
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(greet, name="greet")
adapter.start_consumer()
try:
print("Sending sync inference query...")
result = adapter.query_task_sync("greet", args=["World"], timeout=10)
print(f" status={result.status}")
print(f" result={result.result}")
finally:
adapter.stop_consumer()
print("Example complete.")
if __name__ == "__main__":
main()
```
Expected output:
```text title="output"
Sending sync inference query...
status=SUCCESS
result={'greeting': 'Hello, World!'}
Example complete.
```
In production you deploy the adapter inside a Ray Serve `@task_consumer` and call `query_task_sync` from your request path. The standalone setup above (`initialize` + `register_task_handler` + `start_consumer`) is used here so the example runs as a single script against a local broker.
## Timeout Behavior [#timeout-behavior]
The `timeout` argument is the per-call query deadline in seconds. When you pass a value it overrides `config.sync_inference_timeout` (default `30`); when omitted or falsy it falls back to that config value:
```python title="adapter.py"
query = QueryMessage(
channel=self._queue_name,
body=body,
timeout_in_seconds=timeout or self._config.sync_inference_timeout,
)
response = self._cq_client.send_query(query)
```
If no response arrives before the deadline, the SDK raises `KubeMQTimeoutError`. Catch it on the caller side and decide whether to retry, fail the request, or fall back to async submission:
```python title="timeout_handling.py"
from kubemq.core.exceptions import KubeMQTimeoutError
def slow_handler(data: str) -> dict:
import time
time.sleep(5) # longer than the query timeout below
return {"data": data, "processed": True}
# slow_handler takes 5s, but the query deadline is only 1s
try:
result = adapter.query_task_sync("slow_handler", args=["slow-data"], timeout=1)
print(f"Unexpected success: {result.result}")
except KubeMQTimeoutError as exc:
print(f"Timed out: {exc}")
# The handler needs more time than the query allows — retry with a
# larger timeout, or enqueue it asynchronously instead.
```
Set the timeout to your expected handler duration plus a margin. As a rule of thumb: fast handlers `timeout=5-10`, slower handlers `timeout=30-60`. If a handler routinely needs more than a minute, use [async inference](/integrations/rayserve/how-to/async-inference) instead of blocking the caller.
## Sync vs Async Tradeoffs [#sync-vs-async-tradeoffs]
The same task can run either way. Sync blocks the caller and returns the result inline; async enqueues the task and polls the result backend until it completes.
| | Sync (`query_task_sync`) | Async (`enqueue_task_sync` + poll) |
| ---------------- | ----------------------------------------- | ------------------------------------------------- |
| Transport | KubeMQ Query (`CQClient`) | KubeMQ Queue (`QueuesClient`) |
| Caller | Blocks until result or timeout | Returns immediately, polls later |
| Result delivery | In line through the Query channel | Stored in the queue-peek result backend |
| Polling overhead | None | Bounded by the poll interval |
| Best for | Interactive, single, low-latency requests | Fire-and-forget, batch, long-running jobs |
| Retries / DLQ | None — failure returns `FAILURE` | `max_retries`, `failed_task_queue_name`, `on_dlq` |
The comparison example runs both paths against the same handler and prints their latencies side by side:
```python title="sync_vs_async_comparison.py"
input_text = "The quick brown fox jumps over the lazy dog"
# --- Sync approach: query_task_sync ---
t0 = time.time()
sync_result = adapter.query_task_sync("summarize", args=[input_text], timeout=10)
sync_elapsed = time.time() - t0
print(f"Sync: status={sync_result.status} elapsed={sync_elapsed:.3f}s")
# --- Async approach: enqueue + poll ---
t0 = time.time()
enqueued = adapter.enqueue_task_sync("summarize", args=[input_text])
for _ in range(30):
status = adapter.get_task_status_sync(enqueued.id)
if status.status in ("SUCCESS", "FAILURE"):
break
time.sleep(0.5)
async_elapsed = time.time() - t0
print(f"Async: status={status.status} elapsed={async_elapsed:.3f}s")
```
Sync is typically faster for single requests because it avoids the polling-interval overhead; async is the better fit for fire-and-forget or batch workloads where the caller should not block.
## Multiple Handlers Over Sync [#multiple-handlers-over-sync]
A single adapter can register many handlers and dispatch sync queries to each by `task_name`. Every handler shares the same Query channel; the `task_name` in the payload selects which one runs.
```python title="sync_multiple_handlers.py"
def sentiment(text: str) -> dict:
positive_words = {"good", "great", "excellent", "happy", "love"}
words = set(text.lower().split())
score = len(words & positive_words) / max(len(words), 1)
return {"label": "positive" if score > 0.2 else "neutral", "score": round(score, 2)}
def tokenize(text: str) -> dict:
tokens = text.split()
return {"tokens": tokens, "count": len(tokens)}
def language_detect(text: str) -> dict:
spanish = {"el", "la", "de", "en", "es", "hola"}
words = set(text.lower().split())
return {"language": "es" if words & spanish else "en", "confidence": 0.95}
adapter.register_task_handler(sentiment, name="sentiment")
adapter.register_task_handler(tokenize, name="tokenize")
adapter.register_task_handler(language_detect, name="language_detect")
adapter.start_consumer()
text = "This is a great and excellent product"
# Dispatch the same input to three different handlers by name.
for task in ("sentiment", "tokenize", "language_detect"):
result = adapter.query_task_sync(task, args=[text], timeout=10)
print(f"{task}: status={result.status} result={result.result}")
```
A query whose `task_name` does not match any registered handler returns a `TaskResult` with `status="FAILURE"` and an `Unknown task handler` error — it never blocks indefinitely.
## Next Steps [#next-steps]