{"record":{"id":"a670a1ee791bced6","repo":"tursodatabase/turso","slug":"failed-to-parse-pipeline-request","errorCode":null,"errorMessage":"Failed to parse pipeline request: {}","messagePattern":"Failed to parse pipeline request: (.+?)","errorType":"exception","errorClass":null,"httpStatus":null,"severity":"error","filePath":"cli/sync_server.rs","lineNumber":303,"sourceCode":"                error!(\"Request error: {}\", e);\n                HttpResponse {\n                    status: 500,\n                    content_type: \"text/plain\".to_string(),\n                    body: format!(\"Internal Server Error: {e}\").into_bytes(),\n                }\n            }\n        };\n\n        let response_bytes = format_http_response(&http_response);\n        stream.write_all(&response_bytes)?;\n        stream.flush()?;\n\n        Ok(())\n    }\n\n    fn handle_pipeline(&self, db: &DbHandle, body: &[u8]) -> Result<HttpResponse> {\n        let req: PipelineReqBody = serde_json::from_slice(body)\n            .map_err(|e| anyhow!(\"Failed to parse pipeline request: {}\", e))?;\n\n        debug!(\"Pipeline request: {:?}\", req);\n\n        let conn = db.conn.lock().unwrap();\n\n        let mut results = Vec::new();\n\n        for request in req.requests {\n            let result = match request {\n                StreamRequest::Execute(exec_req) => self.execute_statement(&conn, &exec_req),\n                StreamRequest::Batch(batch_req) => self.execute_batch(&conn, &batch_req),\n                StreamRequest::None => StreamResult::Error {\n                    error: Error {\n                        message: \"Unknown request type\".to_string(),\n                        code: \"UNKNOWN\".to_string(),\n                    },\n                },\n            };","sourceCodeStart":285,"sourceCodeEnd":321,"githubUrl":"https://github.com/tursodatabase/turso/blob/492c4a71cd7c2649e7df83da1471b74f4b1c7aa9/cli/sync_server.rs#L285-L321","documentation":"The test sync server's POST /v2/pipeline endpoint parses the request body with serde_json into PipelineReqBody. This error means the body is not valid JSON for that schema; the serde message is appended and names the offending field. PipelineReqBody expects an object with a requests array whose entries are Execute or Batch stream requests.","triggerScenarios":"POSTing to /v2/pipeline with syntactically invalid JSON, a missing or non-array requests field, entries whose shape matches neither StreamRequest::Execute nor StreamRequest::Batch, or wrong field types (e.g. args not an array).","commonSituations":"Hand-written curl payloads during debugging; client and server built from different commits so the PipelineReqBody JSON contract drifted; a proxy truncating or rewriting bodies; sending the protobuf /pull-updates payload to the JSON pipeline route by mistake.","solutions":["Log the exact request body and compare it field-by-field against the PipelineReqBody / StreamRequest serde types in this server build","Fix the payload: top-level requests array; each entry is an Execute or Batch object with the expected fields","Rebuild client and server from the same revision so the generated JSON shapes match","Send the body produced by the client library's serializer instead of constructing JSON by hand"],"exampleFix":"// before\ncurl -d '{\"request\": [{\"stmt\": {\"sql\": \"SELECT 1\"}}]}' localhost:8080/v2/pipeline\n\n// after\ncurl -d '{\"requests\": [{\"Execute\": {\"stmt\": {\"sql\": \"SELECT 1\"}}}]}' localhost:8080/v2/pipeline","handlingStrategy":"validation","validationCode":"function validatePipelineRequest(body) {\n  const parsed = JSON.parse(body); // throws early on bad JSON\n  if (!parsed || !Array.isArray(parsed.requests)) {\n    throw new Error('requests must be an array');\n  }\n  for (const r of parsed.requests) {\n    if (!(('Execute' in r) || ('Batch' in r))) {\n      throw new Error('each entry must be Execute or Batch');\n    }\n  }\n  return parsed;\n}","typeGuard":"function isPipelineRequest(v: unknown): v is { requests: unknown[] } {\n  return typeof v === 'object' && v !== null && Array.isArray((v as any).requests);\n}","tryCatchPattern":"// server/operator side: surface the serde message with a 400-style response\nmatch self.handle_pipeline(&body) {\n    Ok(resp) => resp,\n    Err(e) if e.to_string().starts_with(\"Failed to parse pipeline request\") => {\n        HttpResponse { status: 400, content_type: \"text/plain\".into(), body: e.to_string().into_bytes() }\n    }\n    Err(e) => /* 500 */,\n}","preventionTips":["Build client and server from the same commit so the JSON contract cannot drift","Always serialize requests with the client library, never hand-build JSON","Add a contract test that posts a serialized PipelineReqBody and asserts 200"],"tags":["sync-server","json","serde","pipeline","request-validation"],"backgroundTag":"json-deserialization-failed","analyzedSha":"492c4a71cd7c2649e7df83da1471b74f4b1c7aa9","analyzedAt":"2026-09-13T18:13:59.796Z","contentChangedAt":"2026-09-13T18:13:59.796Z","schemaVersion":2},"datasetVersion":"2026-09-14T00:17:10.932Z"}