Function process_upsert_state_error 

async fn process_upsert_state_error<T: Timestamp>(
    context: String,
    e: Error,
    health_output: &AsyncOutputHandle<T, CapacityContainerBuilder<Vec<(Option<GlobalId>, HealthStatusUpdate)>>>,
    health_cap: &Capability<T>,
)

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