chore(epoxy): add replica reconfigure endpoint

Author: NathanFlurry

packages/core/api-peer/src/internal.rs

@@ -11,9 +11,9 @@ pub struct CachePurgeRequest {
 }
 
 #[derive(Serialize)]
+#[serde(deny_unknown_fields)]
 pub struct CachePurgeResponse {}
 
-#[tracing::instrument(skip_all)]
 pub async fn cache_purge(
 	ctx: ApiCtx,
 	_path: (),
@@ -30,9 +30,9 @@ pub async fn cache_purge(
 }
 
 #[derive(Serialize)]
+#[serde(deny_unknown_fields)]
 pub struct BumpServerlessAutoscalerResponse {}
 
-#[tracing::instrument(skip_all)]
 pub async fn bump_serverless_autoscaler(
 	ctx: ApiCtx,
 	_path: (),
@@ -82,3 +82,24 @@ pub async fn set_tracing_config(
 
 	Ok(SetTracingConfigResponse {})
 }
+
+#[derive(Deserialize)]
+#[serde(deny_unknown_fields)]
+pub struct ReplicaReconfigureRequest {}
+
+#[derive(Serialize)]
+#[serde(deny_unknown_fields)]
+pub struct ReplicaReconfigureResponse {}
+
+pub async fn epoxy_replica_reconfigure(
+	ctx: ApiCtx,
+	_path: (),
+	_query: (),
+	_body: ReplicaReconfigureRequest,
+) -> Result<ReplicaReconfigureResponse> {
+	ctx.signal(epoxy::workflows::coordinator::ReplicaReconfigure {})
+	.send()
+	.await?;
+
+	Ok(ReplicaReconfigureResponse {})
+}

packages/core/api-peer/src/router.rs

@@ -35,7 +35,10 @@ pub async fn router(
 				"/bump-serverless-autoscaler",
 				post(internal::bump_serverless_autoscaler),
 			)
-			// MARK: Debug
+			.route(
+				"/epoxy/replica-reconfigure",
+				post(internal::epoxy_replica_reconfigure),
+			)
 			.route("/debug/tracing/config", put(internal::set_tracing_config))
 	})
 	.await

packages/services/epoxy/src/workflows/coordinator/mod.rs

@@ -37,6 +37,9 @@ pub async fn epoxy_coordinator(ctx: &mut WorkflowCtx, _input: &Input) -> Result<
 				Main::ReplicaStatusChange(sig) => {
 					replica_status_change::replica_status_change(ctx, sig).await?;
 				}
+				Main::ReplicaReconfigure(_) => {
+					replica_status_change::replica_reconfigure(ctx).await?;
+				}
 			}
 
 			Ok(Loop::<()>::Continue)
@@ -81,7 +84,11 @@ pub struct ReplicaStatusChange {
 	pub status: types::ReplicaStatus,
 }
 
+#[signal("epoxy_coordinator_replica_reconfigure")]
+pub struct ReplicaReconfigure {}
+
 join_signal!(Main {
 	Reconfigure,
 	ReplicaStatusChange,
+	ReplicaReconfigure,
 });

packages/services/epoxy/src/workflows/coordinator/replica_status_change.rs

@@ -37,6 +37,23 @@ pub async fn replica_status_change(
 	Ok(())
 }
 
+#[tracing::instrument(skip_all)]
+pub async fn replica_reconfigure(
+	ctx: &mut WorkflowCtx,
+) -> Result<()> {
+	let notify_out = ctx.activity(NotifyAllReplicasInput {}).await?;
+
+	let replica_id = ctx.config().epoxy_replica_id();
+	ctx.msg(super::ConfigChangeMessage {
+		config: notify_out.config,
+	})
+	.tag("replica", replica_id)
+	.send()
+	.await?;
+
+	Ok(())
+}
+
 #[derive(Debug, Clone, Serialize, Deserialize, Hash)]
 pub struct UpdateReplicaStatusInput {
 	pub replica_id: protocol::ReplicaId,