{"record":{"id":"7c47a43976805131","repo":"risingwavelabs/risingwave","slug":"invalid-parallelism","errorCode":null,"errorMessage":"invalid parallelism","messagePattern":"invalid parallelism","errorType":"panic","errorClass":null,"httpStatus":null,"severity":"critical","filePath":"src/meta/src/controller/cluster.rs","lineNumber":701,"sourceCode":"                    Set(Some(add_property.resource_group.unwrap_or_else(|| {\n                        tracing::warn!(\n                            \"resource_group is not set for worker {}, fallback to `default`\",\n                            worker.worker_id\n                        );\n                        DEFAULT_RESOURCE_GROUP.to_owned()\n                    })));\n\n                WorkerProperty::update(property).exec(&txn).await?;\n                txn.commit().await?;\n                self.update_worker_ttl(worker.worker_id, ttl)?;\n                Ok(worker.worker_id)\n            } else if worker.worker_type == WorkerType::Frontend && property.is_none() {\n                let worker_property = worker_property::ActiveModel {\n                    worker_id: Set(worker.worker_id),\n                    parallelism: Set(add_property\n                        .parallelism\n                        .try_into()\n                        .expect(\"invalid parallelism\")),\n                    is_streaming: Set(add_property.is_streaming),\n                    is_serving: Set(add_property.is_serving),\n                    is_unschedulable: Set(false),\n                    internal_rpc_host_addr: Set(Some(add_property.internal_rpc_host_addr)),\n                    resource_group: Set(None),\n                    is_iceberg_compactor: Set(false),\n                    resource: Set(Some((&resource).into())),\n                    started_at: Set(Some(started_at as _)),\n                };\n                WorkerProperty::insert(worker_property).exec(&txn).await?;\n                txn.commit().await?;\n                self.update_worker_ttl(worker.worker_id, ttl)?;\n                Ok(worker.worker_id)\n            } else if worker.worker_type == WorkerType::Compactor {\n                if let Some(property) = property {\n                    let mut property: worker_property::ActiveModel = property.into();\n                    property.is_iceberg_compactor = Set(add_property.is_iceberg_compactor);\n                    property.internal_rpc_host_addr =","sourceCodeStart":683,"sourceCodeEnd":719,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/meta/src/controller/cluster.rs#L683-L719","documentation":"When registering a Frontend worker with no existing property record, the code builds a worker_property::ActiveModel and converts add_property.parallelism (unsigned) into the DB integer type with try_into().expect(\"invalid parallelism\"). If parallelism cannot be converted (e.g. exceeds the target integer range), expect panics with this message — it is an unrecoverable invariant, not a returned error.","triggerScenarios":"ADD NODE / worker registration for a Frontend node whose reported parallelism cannot be represented in the storage type (e.g. > i32::MAX due to a malformed or hostile WorkerNode property).","commonSituations":"Misconfigured or buggy client sending absurd parallelism values in AddWorkerProperties; binary/protobuf version mismatch misinterpreting the field.","solutions":["Send a sane parallelism value in the add-node request (fits within i32).","Ensure client and cluster use compatible protobuf versions so the parallelism field is decoded correctly.","Replace the expect with validated/returned error handling in meta code if maintaining a fork."],"exampleFix":"// before\nparallelism: Set(add_property.parallelism.try_into().expect(\"invalid parallelism\")),\n// after\nlet parallelism: i32 = add_property.parallelism.try_into().map_err(|_| MetaError::invalid_parameter(\"parallelism out of range\"))?;\nparallelism: Set(parallelism),","handlingStrategy":"validation","validationCode":"// Rust (client of AddWorker RPC)\nfn valid_parallelism(p: u32) -> bool { p > 0 && p <= i32::MAX as u32 }\nassert!(valid_parallelism(props.parallelism), \"parallelism must fit i32\");","typeGuard":"fn fits_i32(v: u32) -> Option<i32> { i32::try_from(v).ok() }","tryCatchPattern":"// This is a panic (expect), not an Err — guard at the boundary instead.\nlet p = i32::try_from(add_property.parallelism)\n    .expect(\"invalid parallelism\"); // only safe if you validated above","preventionTips":["Validate parallelism at the RPC boundary before constructing worker properties.","Keep meta and worker binaries version-aligned to avoid field misdecoding.","Replace expect with error propagation in any fork/patch of this code."],"tags":["meta","cluster","panic","parallelism"],"backgroundTag":"value-out-of-range","analyzedSha":"6469eb736d691e8e9b8a419a57edd6429ca77417","analyzedAt":"2026-09-11T21:06:21.487Z","contentChangedAt":"2026-09-11T21:06:21.487Z","schemaVersion":2},"datasetVersion":"2026-09-14T16:17:12.679Z"}