fix(pegboard): isolate runner config dc lookup failures (#4625)

# Description

Please include a summary of the changes and the related issue. Please also include relevant motivation and context.

## Type of change

- [ ] Bug fix (non-breaking change which fixes an issue)
- [ ] New feature (non-breaking change which adds functionality)
- [ ] Breaking change (fix or feature that would cause existing functionality to not work as expected)
- [ ] This change requires a documentation update

## How Has This Been Tested?

Please describe the tests that you ran to verify your changes.

## Checklist:

- [ ] My code follows the style guidelines of this project
- [ ] I have performed a self-review of my code
- [ ] I have commented my code, particularly in hard-to-understand areas
- [ ] I have made corresponding changes to the documentation
- [ ] My changes generate no new warnings
- [ ] I have added tests that prove my fix is effective or that my feature works
- [ ] New and existing unit tests pass locally with my changes

Author: NathanFlurry

engine/packages/pegboard/src/ops/runner/list_runner_config_enabled_dcs.rs

@@ -1,6 +1,6 @@
 use anyhow::Result;
 use epoxy_protocol::generated::v2::CachingBehavior;
-use futures_util::{StreamExt, TryStreamExt};
+use futures_util::StreamExt;
 use gas::prelude::*;
 
 use crate::keys;
@@ -59,27 +59,41 @@ async fn list_runner_config_enabled_dcs_inner(
 	ctx: &OperationCtx,
 	input: &Input,
 ) -> Result<Vec<u16>> {
-	futures_util::stream::iter(ctx.config().topology().datacenters.clone())
-		.map(|dc| async move {
-			let runner_config_key = keys::runner_config::GlobalDataKey::new(
-				dc.datacenter_label,
-				input.namespace_id,
-				input.runner_name.clone(),
-			);
-			let res = ctx
-				.op(epoxy::ops::kv::get_optimistic::Input {
-					replica_id: ctx.config().epoxy_replica_id(),
-					key: namespace::keys::subspace().pack(&runner_config_key),
-					caching_behavior: CachingBehavior::Optimistic,
-					target_replicas: None,
-					save_empty: true,
-				})
-				.await?;
+	Ok(
+		futures_util::stream::iter(ctx.config().topology().datacenters.clone())
+			.map(|dc| async move {
+				let runner_config_key = keys::runner_config::GlobalDataKey::new(
+					dc.datacenter_label,
+					input.namespace_id,
+					input.runner_name.clone(),
+				);
+				let res = ctx
+					.op(epoxy::ops::kv::get_optimistic::Input {
+						replica_id: ctx.config().epoxy_replica_id(),
+						key: namespace::keys::subspace().pack(&runner_config_key),
+						caching_behavior: CachingBehavior::Optimistic,
+						target_replicas: None,
+						save_empty: true,
+					})
+					.await;
 
-			Ok(res.value.map(|_| dc.datacenter_label))
-		})
-		.buffer_unordered(512)
-		.try_filter_map(|x| std::future::ready(Ok(x)))
-		.try_collect::<Vec<_>>()
-		.await
+				match res {
+					Ok(res) => res.value.map(|_| dc.datacenter_label),
+					Err(err) => {
+						tracing::warn!(
+							?err,
+							namespace_id=?input.namespace_id,
+							runner_name=%input.runner_name,
+							dc_label=dc.datacenter_label,
+							"failed to read runner config from dc"
+						);
+						None
+					}
+				}
+			})
+			.buffer_unordered(512)
+			.filter_map(std::future::ready)
+			.collect::<Vec<_>>()
+			.await,
+	)
 }