{"record":{"id":"480219e4ee170f28","repo":"xai-org/x-algorithm","slug":"failed-to-get-num-servers-or-num-processors-from-g","errorCode":null,"errorMessage":"Failed to get num_servers or num_processors from gRPC service","messagePattern":"Failed to get num_servers or num_processors from gRPC service","errorType":"validation","errorClass":"ValueError","httpStatus":null,"severity":"warning","filePath":"phoenix/xrex/data/streaming/kafkadispatcherloader.py","lineNumber":76,"sourceCode":"    async with get_grpc_channel(master_node_host, grpc_port) as channel:\n        stub = KafkaDispatcherStub(channel)\n        response = await stub.CheckState(CheckStateRequest())\n        num_servers = response.num_shards\n        num_processors = response.num_processors\n        if num_servers is None or num_processors is None:\n            raise ValueError(f\"Invalid {num_servers=} or {num_processors=} in CheckStateResponse\")\n        return num_servers, num_processors\n\n\nasync def get_server_assignments(\n    grpc_host_template: str,\n    grpc_port: int,\n    num_clients: int,\n    client_ix: int,\n) -> tuple[int, int, list[int]]:\n    num_servers, num_processors = await get_service_dimensions(grpc_host_template, grpc_port)\n    if num_servers is None or num_processors is None:\n        raise ValueError(\"Failed to get num_servers or num_processors from gRPC service\")\n\n    assert num_clients % (num_servers * num_processors) == 0, (\n        f\"Num dataloaders {num_clients} must be a multiple of number of servers {num_servers=} * {num_processors=} = {num_servers * num_processors}\"\n    )\n\n    global_num_processors = num_servers * num_processors\n    num_clients_per_processor = num_clients // global_num_processors\n    global_processor_ix = client_ix // num_clients_per_processor\n    server_index = global_processor_ix // num_processors\n    processor_index = global_processor_ix % num_processors\n    clients_assigned_to_processor = [\n        a for a in range(num_clients) if a // num_clients_per_processor == global_processor_ix\n    ]\n    assert len(clients_assigned_to_processor) == num_clients_per_processor, (\n        f\"Number of clients assigned to server {global_processor_ix} is not equal to num_clients_per_processor {num_clients_per_processor}\"\n    )\n    return server_index, processor_index, clients_assigned_to_processor\n","sourceCodeStart":58,"sourceCodeEnd":94,"githubUrl":"https://github.com/xai-org/x-algorithm/blob/24c60942c5c5fdad3a6addffb4c6e6d2f228f04f/phoenix/xrex/data/streaming/kafkadispatcherloader.py#L58-L94","documentation":"Defensive re-check in get_server_assignments after get_service_dimensions already validates the gRPC response. It fires only if dimensions came back as None without the earlier ValueError being raised (defensive dead-code guard in normal operation).","triggerScenarios":"Effectively unreachable when get_service_dimensions raises first; could fire only if get_service_dimensions is monkeypatched or a modified version returns Nones without validating.","commonSituations":"Custom forks/tests stubbing get_service_dimensions to return (None, None); refactors that bypass the validation inside get_service_dimensions.","solutions":["Fix the stub/mock in tests to return valid (num_servers, num_processors) tuples.","Keep validation in one place: rely on get_service_dimensions' ValueError and remove/ignore this redundant check in forks.","If hit in production, check for monkeypatches or duplicated modified code paths."],"exampleFix":"# before (test stub)\nmonkeypatch.setattr(loader, \"get_service_dimensions\", async lambda h, p: (None, None))\n\n# after\nmonkeypatch.setattr(loader, \"get_service_dimensions\", async lambda h, p: (2, 4))","handlingStrategy":"validation","validationCode":"num_servers, num_processors = await get_service_dimensions(host, port)\nif num_servers is None or num_processors is None:\n    raise RuntimeError(\"dispatcher returned invalid dimensions\")","typeGuard":null,"tryCatchPattern":null,"preventionTips":["Keep single-point validation in get_service_dimensions.","In tests, stub get_service_dimensions with realistic tuples."],"tags":["grpc","defensive-check","kafka-dispatcher"],"backgroundTag":"grpc-unset-response-field","analyzedSha":"24c60942c5c5fdad3a6addffb4c6e6d2f228f04f","analyzedAt":"2026-08-28T11:40:14.686Z","schemaVersion":2},"datasetVersion":"2026-08-28T16:17:29.566Z"}