# Progress Tracking (/integrations/rayserve/how-to/progress-tracking) ## Overview [#overview] Long-running inference — large language model generations, batch jobs, or multi-stage pipelines — can take seconds to minutes to complete. Rather than leaving callers blind until a result lands, `kubemq-rayserve` lets a task handler publish incremental progress updates that any external client can subscribe to in real time. Progress tracking is built on **[KubeMQ Events](/learn/events)** (the fire-and-forget pub/sub pattern exposed through the SDK's `PubSubClient`). The adapter publishes each update as an `EventMessage` on a dedicated `{queue_name}.progress` channel, and subscribers receive them live. This is one of the capabilities the Celery adapter lacks out of the box — with Celery, progress reporting requires custom signals, whereas here it is a single built-in method. Progress events use KubeMQ Events, which are **fire-and-forget**. They are delivered to whatever subscribers are connected at publish time and are not persisted. The task result itself is always durable through the [queue-peek result backend](/integrations/rayserve/how-to/async-inference) — progress events are an observability stream layered on top of it. ## How `report_progress` works [#how-report_progress-works] `report_progress` is an extension method on `KubeMQTaskProcessorAdapter`. It is meant to be called from inside a task handler to announce how far along the work is. ```python title="signature" def report_progress(self, task_id: str, pct: float, detail: str = "") -> None: ... ``` | Parameter | Type | Description | | --------- | ------- | ---------------------------------------------- | | `task_id` | `str` | Task identifier the progress update belongs to | | `pct` | `float` | Progress percentage, `0`–`100` | | `detail` | `str` | Optional human-readable detail string | When called, the adapter publishes an `EventMessage` on the `{queue_name}.progress` channel. The event body is a JSON object and the event also carries lightweight tags for cheap, body-free filtering: * **Body:** `{"task_id": ..., "pct": ..., "detail": ...}` * **Tags:** `{"task_id": ..., "pct": ...}` (the `pct` tag is the integer percentage as a string) Crucially, publishing is best-effort: if the underlying `send_event` call fails, the adapter logs a warning rather than raising. A failed progress update will never crash or fail the task it is reporting on. ```python title="adapter.py — report_progress" def report_progress(self, task_id: str, pct: float, detail: str = "") -> None: """Publish a progress update via KubeMQ Events. Callable from within task handlers to report progress. """ body = json.dumps({"task_id": task_id, "pct": pct, "detail": detail}).encode("utf-8") event = EventMessage( channel=f"{self._queue_name}.progress", body=body, tags={"task_id": task_id, "pct": str(int(pct))}, ) try: self._pubsub_client.send_event(event) except Exception as exc: logger.warning("Failed to send progress event: %s", exc) ``` **v1.0.0 limitation.** Progress events are fire-and-forget KubeMQ Events — they are not persisted, so a subscriber that connects after an event was published will not see it. In addition, a handler must know its own `task_id` to report against, and in v1.0.0 that id is **threaded explicitly into the handler** as a keyword argument (the `task_id` kwarg workaround shown below) rather than being injected automatically. ## Prerequisites [#prerequisites] Progress tracking requires a running KubeMQ broker. Start one locally with Docker: Port `50000` is KubeMQ's gRPC endpoint used by both the adapter and the `PubSubClient`. Port `9090` exposes the shared HTTP server used by connector endpoints such as KEDA scaling. ## Reporting progress from a handler [#reporting-progress-from-a-handler] A handler receives its `task_id` as an explicit keyword argument and calls `adapter.report_progress` at meaningful stage boundaries. The handler below reports `0`, `33`, `66`, and `100` percent, each with a short detail string describing the phase. The producer pre-generates the `task_id` and passes it into the task via `kwargs` so the handler and any subscriber share the same id. ```python title="progress_subscriber.py — analyze handler" from kubemq_rayserve import KubeMQAdapterConfig, KubeMQTaskProcessorAdapter # Module-level adapter reference so the handler can reach it adapter: KubeMQTaskProcessorAdapter | None = None def analyze(data: str, task_id: str = "") -> dict: """Handler that reports progress via explicit task_id kwarg.""" assert adapter is not None adapter.report_progress(task_id, 0, "Starting analysis") time.sleep(0.3) adapter.report_progress(task_id, 33, "Phase 1 complete") time.sleep(0.3) adapter.report_progress(task_id, 66, "Phase 2 complete") time.sleep(0.3) adapter.report_progress(task_id, 100, "Analysis finished") return {"analyzed": data, "phases": 2} ``` The handler is registered and started like any other task handler, then enqueued with a known `task_id` threaded through `kwargs`: ```python title="progress_subscriber.py — enqueue with a known task_id" import uuid adapter.register_task_handler(analyze, name="analyze") adapter.start_consumer() # Pre-generate a known task_id and pass it as a kwarg known_task_id = f"task-{uuid.uuid4().hex[:8]}" result = adapter.enqueue_task_sync( "analyze", kwargs={"data": "test input", "task_id": known_task_id}, ) print(f"enqueued task_id={result.id}") ``` ## Subscribing to progress events [#subscribing-to-progress-events] An external client observes progress by building a `PubSubClient` and subscribing to the `{channel}.progress` channel with an `EventsSubscription`. The `on_receive_event_callback` decodes `event.body` as the update is delivered. Because progress runs on KubeMQ Events, the subscriber sees updates live as the handler emits them. ```python title="progress_subscriber.py — external subscriber" from kubemq import ( CancellationToken, ClientConfig, EventReceived, EventsSubscription, PubSubClient, ) progress_channel = f"{channel}.progress" received_events: list[str] = [] def on_event(event: EventReceived) -> None: body = event.body.decode("utf-8") if event.body else "" received_events.append(body) print(f" [subscriber] progress event: {body}") def on_error(err: str) -> None: print(f" [subscriber] error: {err}") cancel_token = CancellationToken() pubsub_config = ClientConfig(address=BROKER, client_id=f"progress-sub-{uuid.uuid4().hex[:8]}") pubsub_client = PubSubClient(config=pubsub_config) subscription = EventsSubscription( channel=progress_channel, group="", on_receive_event_callback=on_event, on_error_callback=on_error, ) pubsub_client.subscribe_to_events(subscription, cancel_token) print(f"Subscribed to progress channel: {progress_channel}") ``` Subscribe **before** enqueuing the task. Since the events are not persisted, any update published while no subscriber is connected is lost. The example pauses briefly (`time.sleep(0.5)`) after subscribing to let the subscription establish before the work begins. When the subscriber and adapter run together — as in `examples/quickstart/progress_hello_world.py` — the callback can parse the JSON body directly to read `pct` and `detail`: ```python title="progress_hello_world.py — decoding pct and detail" import json progress_events: list[dict] = [] def on_event(event) -> None: try: data = json.loads(event.body) progress_events.append(data) print(f" [Progress] pct={data.get('pct', 0)}% detail={data.get('detail', '')}") except Exception: pass subscription = EventsSubscription( channel=f"{channel}.progress", on_receive_event_callback=on_event, on_error_callback=on_error, ) pubsub.subscribe_to_events(subscription, cancel_token) ``` ## Rich detail messages [#rich-detail-messages] The `detail` argument is free-form text, so it can carry descriptive, item-level status such as `"Processing image 3/10"`. The handler in `examples/progress_tracking/progress_with_detail.py` walks a simulated batch and computes a percentage per item while emitting a descriptive detail string at each step. ```python title="progress_with_detail.py — per-item detail strings" def process_images(image_count: int = 5, task_id: str = "") -> dict: """Handler that processes images with rich progress detail strings.""" assert adapter is not None adapter.report_progress(task_id, 0, f"Starting batch of {image_count} images") time.sleep(0.2) processed = 0 for i in range(1, image_count + 1): pct = int((i / image_count) * 80) + 10 # 10% to 90% adapter.report_progress(task_id, pct, f"Processing image {i}/{image_count}") time.sleep(0.2) processed += 1 adapter.report_progress(task_id, 95, f"Generating summary for {processed} images") time.sleep(0.2) adapter.report_progress(task_id, 100, "Batch complete") return {"processed_count": processed, "status": "success"} ``` ## Multi-stage pipeline progress [#multi-stage-pipeline-progress] For pipelines that span multiple handlers, each stage reports progress independently against its own `task_id` while contributing to a shared percentage range. The two-stage example in `examples/progress_tracking/multi_stage_progress.py` splits the work so preprocessing covers `0`–`50%` and inference covers `50`–`100%`, producing a single continuous progress arc across the pipeline. ```python title="multi_stage_progress.py — two-stage progress ranges" def preprocess(data: str, task_id: str = "") -> dict: """Stage 1: Preprocessing with progress 0-50%.""" assert adapter is not None adapter.report_progress(task_id, 0, "Stage 1: Starting preprocessing") time.sleep(0.2) adapter.report_progress(task_id, 10, "Stage 1: Validating input") time.sleep(0.2) adapter.report_progress(task_id, 25, "Stage 1: Normalizing data") time.sleep(0.2) adapter.report_progress(task_id, 40, "Stage 1: Tokenizing") time.sleep(0.2) adapter.report_progress(task_id, 50, "Stage 1: Preprocessing complete") return {"preprocessed": data.lower().strip(), "tokens": len(data.split())} def inference(preprocessed: str, token_count: int = 0, task_id: str = "") -> dict: """Stage 2: Inference with progress 50-100%.""" assert adapter is not None adapter.report_progress(task_id, 55, "Stage 2: Loading model") time.sleep(0.2) adapter.report_progress(task_id, 65, "Stage 2: Running inference") time.sleep(0.3) adapter.report_progress(task_id, 80, "Stage 2: Post-processing results") time.sleep(0.2) adapter.report_progress(task_id, 95, "Stage 2: Formatting output") time.sleep(0.1) adapter.report_progress(task_id, 100, "Stage 2: Inference complete") return {"prediction": "positive", "confidence": 0.94, "token_count": token_count} ``` Each stage is enqueued in turn, with the output of stage 1 feeding stage 2. Both stages publish to the same `{queue_name}.progress` channel, so a single subscriber observes the entire pipeline: ```python title="multi_stage_progress.py — chaining the stages" # Stage 1: preprocess stage1_task_id = f"task-{uuid.uuid4().hex[:8]}" result1 = adapter.enqueue_task_sync( "preprocess", kwargs={"data": input_data, "task_id": stage1_task_id}, ) # ... poll result1 until SUCCESS/FAILURE, capturing stage1_output ... # Stage 2: inference, fed by stage 1 output stage2_task_id = f"task-{uuid.uuid4().hex[:8]}" result2 = adapter.enqueue_task_sync( "inference", kwargs={ "preprocessed": stage1_output.get("preprocessed", ""), "token_count": stage1_output.get("tokens", 0), "task_id": stage2_task_id, }, ) ``` ## End-to-end quickstart [#end-to-end-quickstart] `examples/quickstart/progress_hello_world.py` ties the pieces together in a single script: it registers a handler that reports `25`, `50`, `75`, and `100` percent, subscribes to the progress channel, enqueues a task with a known `task_id`, and polls for the final result while progress events stream in. ```python title="progress_hello_world.py — handler" def process_data(data: str, task_id: str = "") -> dict: """Handler that reports progress at each stage.""" if adapter is not None and task_id: adapter.report_progress(task_id, 25, "Starting processing") time.sleep(0.3) if adapter is not None and task_id: adapter.report_progress(task_id, 50, "Halfway done") time.sleep(0.3) if adapter is not None and task_id: adapter.report_progress(task_id, 75, "Almost done") time.sleep(0.3) if adapter is not None and task_id: adapter.report_progress(task_id, 100, "Complete") return {"processed": data, "status": "done"} ``` Run it against a live broker: ```bash title="terminal" uv run python examples/quickstart/progress_hello_world.py ``` Expected output interleaves the streamed progress events with the final result: ```text title="output" Enqueuing task with known task_id=task-1a2b3c4d... enqueued: task_id=... [Progress] pct=25% detail=Starting processing [Progress] pct=50% detail=Halfway done [Progress] pct=75% detail=Almost done [Progress] pct=100% detail=Complete Result: status=SUCCESS result={'processed': 'hello', 'status': 'done'} Total progress events received: 4 ``` ## Next steps [#next-steps]