{"record":{"id":"94ef741b4e6015d6","repo":"risingwavelabs/risingwave","slug":"locality-provider-upstream-ended-unexpectedly-duri","errorCode":null,"errorMessage":"locality provider upstream ended unexpectedly during backfill","messagePattern":"locality provider upstream ended unexpectedly during backfill","errorType":"exception","errorClass":null,"httpStatus":null,"severity":"critical","filePath":"src/stream/src/executor/locality_provider.rs","lineNumber":600,"sourceCode":"                    let snapshot_next = snapshot_stream_ref.next();\n                    pin_mut!(upstream_next);\n                    pin_mut!(snapshot_next);\n\n                    match select(upstream_next, snapshot_next).await {\n                        FutureEither::Left((msg, _)) => match msg.transpose()? {\n                            Some(Message::Barrier(barrier)) => {\n                                // Process the barrier after draining the snapshot builders.\n                                break barrier;\n                            }\n                            Some(Message::Chunk(chunk)) => {\n                                // Buffer the upstream chunk.\n                                upstream_chunk_buffer.push(chunk.compact_vis());\n                            }\n                            Some(Message::Watermark(_)) => {\n                                // Ignore watermark during backfill.\n                            }\n                            None => {\n                                return Err(anyhow::anyhow!(\n                                    \"locality provider upstream ended unexpectedly during backfill\"\n                                )\n                                .into());\n                            }\n                        },\n                        FutureEither::Right((msg, _)) => match msg.transpose()? {\n                            Some((vnode, row)) => {\n                                // Use builder to batch rows efficiently\n                                let builder = builders.get_mut(&vnode).unwrap();\n                                if let Some(data_chunk) = builder.append_one_row(row) {\n                                    // Builder is full, handle the chunk\n                                    let chunk = Self::handle_snapshot_chunk(\n                                        data_chunk,\n                                        vnode,\n                                        &pk_indices,\n                                        &mut backfill_state,\n                                        &mut cur_barrier_snapshot_processed_rows,\n                                    )?;","sourceCodeStart":582,"sourceCodeEnd":618,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/stream/src/executor/locality_provider.rs#L582-L618","documentation":"During backfill, the locality provider executor selects between its upstream message stream and the snapshot stream, expecting the upstream to stay alive until a barrier is delivered. If the upstream stream terminates (yields None) before any barrier arrives, the provider cannot make progress — the backfill loop has no barrier to break on — and aborts with this error. A stream ending without a barrier is a protocol violation for a running actor.","triggerScenarios":"execute_inner's backfill loop selects on upstream.next(); the upstream yields None (actor finished, failed, or was cancelled) before emitting Message::Barrier. Chunks are buffered and watermarks ignored, but stream end without a barrier is fatal.","commonSituations":"Upstream actor crash/panic during a backfill; rescale or fragment rewrite removing the upstream mid-backfill; upstream reaching a terminal error state; scheduler bugs that stop an actor without emitting a final barrier.","solutions":["Check the upstream actor's logs for the original failure/panic that ended its stream and fix that root cause.","Retry recovery — RisingWave's barrier-based recovery restarts the fragment and the backfill resumes from its persisted state.","Verify no concurrent rescale/schema change removed the upstream mid-backfill; re-run the backfill after the topology is stable.","If reproducible with a healthy upstream, file a bug with the fragment graph — it indicates an executor/scheduler protocol violation."],"exampleFix":null,"handlingStrategy":"try-catch","validationCode":null,"typeGuard":null,"tryCatchPattern":"match locality_provider.run().await {\n    Err(e) if e.to_string().contains(\"locality provider upstream ended unexpectedly during backfill\") => {\n        // upstream died mid-backfill: surface root cause, then restart the fragment\n        tracing::error!(%e, \"locality provider upstream lost during backfill; recovering\");\n        recover_fragment().await?;\n    }\n    other => other?,\n}","preventionTips":["Avoid rescaling or DDL/fragment rewrites while a backfill is in progress.","Monitor upstream actor health and restart policies so stream failures are recovered promptly.","Confirm the upstream always emits barriers (never terminates silently) in custom executors feeding the locality provider.","Enable structured logging on upstream actors to correlate their termination with the backfill failure."],"tags":["streaming","backfill","upstream","barrier","rust"],"backgroundTag":"upstream-terminated-unexpectedly","analyzedSha":"6469eb736d691e8e9b8a419a57edd6429ca77417","analyzedAt":"2026-09-11T21:06:21.487Z","contentChangedAt":"2026-09-11T21:06:21.487Z","schemaVersion":2},"datasetVersion":"2026-09-14T11:17:12.474Z"}