Function process_upsert_state_error

src/storage/src/upsert.rs:1281–1294

Emit the given error, and stall till the dataflow is restarted.

(
    context: String,
    e: anyhow::Error,
    health_output: &AsyncOutputHandle<
        T,
        CapacityContainerBuilder<Vec<(Option<GlobalId>, HealthStatusUpdate)>>,
    >,
    health_cap: &Ca

/// Emit the given error, and stall till the dataflow is restarted.
async fn process_upsert_state_error<T: Timestamp>(
    context: String,
    e: anyhow::Error,
    health_output: &AsyncOutputHandle<
        T,
        CapacityContainerBuilder<Vec<(Option<GlobalId>, HealthStatusUpdate)>>,
    >,
    health_cap: &Capability<T>,
) {
    let update = HealthStatusUpdate::halting(e.context(context).to_string_with_causes(), None);
    health_output.give(health_cap, (None, update));
    std::future::pending::<()>().await;
    unreachable!("pending future never returns");
}