{"record":{"id":"ef0149f93799c7a0","repo":"risingwavelabs/risingwave","slug":"batch-write-failure-with-context","errorCode":null,"errorMessage":"{batch write failure, with context()}","messagePattern":"\\{batch write failure, with context\\(\\)\\}","errorType":"exception","errorClass":"SinkError","httpStatus":null,"severity":"critical","filePath":"src/connector/src/sink/postgres.rs","lineNumber":678,"sourceCode":"        .await;\n        if let Err(e) = result {\n            // Retry any failed batch row by row: keys distinct to RisingWave but equal to\n            // PostgreSQL fail a multi-row upsert with SQLSTATE 21000 yet apply cleanly one row at\n            // a time; other errors recover on retry or resurface localized to a single row.\n            let context = || {\n                format!(\n                    \"failed to execute batched {} statements ({} delete rows, {} write rows)\",\n                    write_kind.as_str(),\n                    deletes.len(),\n                    upserts.len()\n                )\n            };\n            if let Err(rollback_err) = transaction.rollback().await {\n                tracing::warn!(\n                    error = %rollback_err.as_report(),\n                    \"failed to roll back failed batch\"\n                );\n                return Err(anyhow::Error::new(e).context(context()).into());\n            }\n            tracing::warn!(error = %e.as_report(), \"{}, retrying row by row\", context());\n            return self.flush_row_by_row(&deletes, &upserts).await;\n        }\n        transaction.commit().await?;\n\n        Ok(())\n    }\n\n    /// Fallback for a failed batched flush: batched deletes first, then one upsert per statement.\n    async fn flush_row_by_row(&mut self, deletes: &[PgRow], upserts: &[PgRow]) -> Result<()> {\n        let delete_batches = self.prepare_batches(StatementKind::Delete, deletes).await?;\n        let statement = self.cached_statement(self.write_kind(), 1).await?;\n\n        let transaction = self.client.transaction().await?;\n        execute_batches(&transaction, &delete_batches)\n            .await\n            .with_context(|| {","sourceCodeStart":660,"sourceCodeEnd":696,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/connector/src/sink/postgres.rs#L660-L696","documentation":"When a batched transaction write to Postgres fails and the subsequent rollback also fails, the original error `e` is re-wrapped with a context() description (which sink/table/batch) and returned — surfacing as a batch write failure. If rollback succeeds, the sink falls back to flushing row by row. This error means a whole-batch transaction commit failed AND a clean rollback could not be confirmed.","triggerScenarios":"A batch INSERT/DELETE fails inside the transaction (constraint violation, connection drop, serialization failure) and `transaction.rollback().await` returns Err.","commonSituations":"Broken network connection to Postgres mid-batch (both the statement and rollback fail); PG server restarting; dead connections after idle timeouts.","solutions":["Check connectivity to Postgres (network, PG logs, max_connections) — both the write and rollback failing usually means the connection is dead.","Rely on RisingWave's retry: the sink retries failed batches downstream; verify sink retry/backoff config and let it recover.","Reduce `max_batch_rows`/batch size if constraint conflicts within large batches are implicated.","Inspect the wrapped original error (via context) in the logs to find the root cause of the failed batch."],"exampleFix":null,"handlingStrategy":"retry","validationCode":"// preflight: ensure PG is reachable and credentials valid\npsql 'postgres://user@host/db' -c 'SELECT 1';","typeGuard":null,"tryCatchPattern":"match sink.write_err {\n    e if e.to_string().contains(\"batch write failure\") => {\n        // check PG connectivity, then rely on sink retry / reduce batch size\n    }\n    _ => {}\n}","preventionTips":["Monitor PG connection health and idle timeouts.","Reduce max_batch_rows when constraint conflicts are frequent.","Keep sink retry enabled so failed batches are replayed."],"tags":["rust","postgres","sink","network","transaction"],"backgroundTag":"database-write-failed","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"}