Fix split-topology result retrieval

[charlesbluca]

Description

Fix silent retained-result loss in split-topology deployments with multiple worker replicas (NVBug 6348464).

Previously, completed rows remained in worker-local memory while the gateway retrieved them through a load-balanced worker Service. If that request reached a replica that did not process the document, it returned 404 and clients could receive incomplete result data even though processing completed successfully.

What changed

• Workers retain completed rows locally until the gateway acknowledges their handoff.
• Completion callbacks identify the exact worker that owns the rows. The gateway validates that identity, authenticates the internal request, pulls the rows directly from that worker, and persists them atomically before marking the document complete.
• Transient callback, worker, and result-store failures return retryable responses. Bounded deferred retries continue the handoff without rerunning the pipeline, and worker-local rows are discarded only after gateway acknowledgement.
• Gateway status routes read retained results idempotently from the gateway-owned store and move blocking filesystem reads off the async event loop.
• Result generations remain recoverable after filesystem or decoding failures, while configurable TTL cleanup bounds retained files, interrupted writes, in-memory results, and job-tracker state.
• Startup validation checks that the configured result store supports the filesystem operations required for safe publication and recovery.
• Tests cover concurrent and idempotent access, callback retries, failed and orphaned handoffs, internal authentication, worker-address validation, lifecycle bounds, storage recovery, gateway routes, and Helm rendering.

Deployment impact

In split mode, only the gateway mounts the retriever-results PVC. Realtime and batch workers keep pending handoffs in local memory, so they remain independently schedulable across nodes.

The default ReadWriteOnce access mode supports the current single-gateway topology, including multi-node worker deployments. Multiple gateway replicas require shared job-state and result-store coordination beyond the current topology contract. Deployments without a configured filesystem result store retain the in-memory fallback.

Validation

• 86 focused and adjacent pytest tests passed.
• Helm lint and split-mode template rendering passed.
• git diff --check passed.

Checklist

• I am familiar with the Contributing Guidelines.
• New or existing tests cover these changes.
• The documentation is up to date with these changes.

[greptile-apps]

Greptile Summary

This PR replaces the racy load-balanced ClusterIP result-fetch with a shared-filesystem result store on the gateway, fixing silent result loss in multi-worker split-topology deployments.

• Shared result store (worker_result_store.py): introduces an immutable-generation filesystem protocol under NEMO_RETRIEVER_RESULTS_DIR; reads are idempotent, writes are atomic via os.replace, and a background TTL sweep evicts expired files.
• Callback-driven pull (job_callback): gateway fetches retained rows directly from the completing worker pod IP before calling mark_completed, so results land in the shared store before the tracker marks the document terminal.
• Helm wiring: gateway gets the PVC mount and both env vars; realtime/batch workers get only NEMO_RETRIEVER_RESULTS_TTL_SECONDS so they remain independently schedulable on RWO nodes.

Confidence Score: 5/5

Safe to merge — the core result-retrieval fix is correctly implemented with atomic writes, idempotent reads, and properly threaded filesystem I/O in all async handlers.

The shared-filesystem protocol is well-designed: immutable generations, atomic os.replace, per-document SHA-256 directories, and a best-effort TTL sweep. The callback handler correctly orders filesystem write before tracker state transition, ensuring a terminal document always has accessible result data. All filesystem calls in async handlers are correctly wrapped with asyncio.to_thread. The only open items are annotation and style suggestions.

No files require special attention beyond the annotation fix in ingest.py and the values.yaml comment cleanup.

Important Files Changed

Filename	Overview
nemo_retriever/src/nemo_retriever/service/services/worker_result_store.py	New module implementing shared-filesystem result storage with immutable generations; clean error handling via ResultStoreTemporarilyUnavailable; TTL sweep logic is correct.
nemo_retriever/src/nemo_retriever/service/routers/ingest.py	Gateway callback handler wired to pull-and-store before mark_completed; status routes updated to read from shared store; blocking reads correctly wrapped with asyncio.to_thread; one httpx client created per pull request (no pooling).
nemo_retriever/src/nemo_retriever/service/services/job_tracker.py	Eviction now runs before job existence checks; consume_result_data replaced by idempotent get_result_data; changes look correct.
nemo_retriever/helm/templates/deployment.yaml	Gateway gets RESULTS_DIR + PVC mount; workers get only TTL_SECONDS; conditional logic is correctly scoped by role.
nemo_retriever/helm/values.yaml	ttlSeconds added; comment block updated with RWX/RWO guidance; pre-existing 'use local/host-path storage' comment is now potentially misleading for multi-node operators.
nemo_retriever/tests/test_service_worker_callback.py	New tests cover callback omits result_data, idempotent reads, filesystem concurrency, TTL sweep, and validate_result_store; good coverage of the happy and concurrent-write scenarios.

Sequence Diagram

%%{init: {'theme': 'neutral'}}%%
sequenceDiagram
    participant Client
    participant Gateway
    participant Worker
    participant SharedFS as Shared PVC (gateway only)
    participant Tracker as Job Tracker (gateway memory)

    Client->>Gateway: "POST /v1/ingest/job/{id}/document"
    Gateway->>Tracker: register_document + mark_processing
    Gateway->>Worker: forward upload (X-Gateway-Callback-Url: pod-IP)

    Note over Worker: Pipeline processes document
    Worker->>Worker: store_result_data(id, rows) in-memory

    Worker->>Gateway: "POST /v1/internal/job-callback {result_worker_ip}"
    Gateway->>Gateway: get_result_data(id) returns None
    Gateway->>Worker: "GET /v1/internal/document-result/{id} direct pod IP"
    Worker-->>Gateway: "{result_data: [...]}"
    Gateway->>SharedFS: store_result_data(id, rows) as UUID.json
    Gateway->>Tracker: "mark_completed(id, result_rows=N)"
    Gateway-->>Worker: 200 OK

    Client->>Gateway: "GET /v1/ingest/status/{id}"
    Gateway->>Tracker: get_result_data(id) returns None
    Gateway->>SharedFS: _get_from_filesystem(id) returns rows
    Gateway-->>Client: "200 {result_data: [...]}"

Loading

%%{init: {'theme': 'base', 'themeVariables': {"darkMode": true, "background": "#0d1117", "primaryColor": "#21262d", "primaryTextColor": "#e6edf3", "primaryBorderColor": "#8b949e", "lineColor": "#8b949e", "textColor": "#e6edf3", "edgeLabelBackground": "#161b22", "actorBkg": "#21262d", "actorBorder": "#8b949e", "actorTextColor": "#e6edf3", "actorLineColor": "#8b949e", "signalColor": "#8b949e", "signalTextColor": "#e6edf3", "noteBkgColor": "#373320", "noteBorderColor": "#d4a72c", "noteTextColor": "#f0e6c0", "labelBoxBkgColor": "#21262d", "labelBoxBorderColor": "#8b949e", "labelTextColor": "#e6edf3", "loopTextColor": "#e6edf3", "activationBkgColor": "#30363d", "activationBorderColor": "#8b949e"}}}%%
sequenceDiagram
    participant Client
    participant Gateway
    participant Worker
    participant SharedFS as Shared PVC (gateway only)
    participant Tracker as Job Tracker (gateway memory)

    Client->>Gateway: "POST /v1/ingest/job/{id}/document"
    Gateway->>Tracker: register_document + mark_processing
    Gateway->>Worker: forward upload (X-Gateway-Callback-Url: pod-IP)

    Note over Worker: Pipeline processes document
    Worker->>Worker: store_result_data(id, rows) in-memory

    Worker->>Gateway: "POST /v1/internal/job-callback {result_worker_ip}"
    Gateway->>Gateway: get_result_data(id) returns None
    Gateway->>Worker: "GET /v1/internal/document-result/{id} direct pod IP"
    Worker-->>Gateway: "{result_data: [...]}"
    Gateway->>SharedFS: store_result_data(id, rows) as UUID.json
    Gateway->>Tracker: "mark_completed(id, result_rows=N)"
    Gateway-->>Worker: 200 OK

    Client->>Gateway: "GET /v1/ingest/status/{id}"
    Gateway->>Tracker: get_result_data(id) returns None
    Gateway->>SharedFS: _get_from_filesystem(id) returns rows
    Gateway-->>Client: "200 {result_data: [...]}"

Loading

_{Reviews (12): Last reviewed commit: "Offload shared result reads" | Re-trigger Greptile}