fix(app): stop serving when the outbox relay stops
ZhenYi committed
Modified app/src/main.rs app/src/main.rs +27 −3 View app/src/main.rs
@@ -95,6 +95,27 @@
9595 Some(queue::Publisher::new(conn.clone())),
9696 vec![("apihost", state.database.primary().clone())],
9797 );
98 let relay_cancel = cancel.clone();
99 let (relay_stop, relay_stop_rx) = tokio::sync::oneshot::channel::<()>();
100 let relay_task = tokio::spawn(async move {
101 let mut relay = relay;
102 tokio::select! {
103 result = relay.wait() => {
104 match result {
105 Ok(()) => tracing::error!("event outbox relay stopped unexpectedly"),
106 Err(error) => tracing::error!(%error, "event outbox relay failed"),
107 }
108 relay_cancel.cancel();
109 true
110 }
111 _ = relay_stop_rx => {
112 if let Err(error) = relay.shutdown().await {
113 warn!(%error, "shutting down the event outbox relay failed");
114 }
115 false
116 }
117 }
118 });
98119 info!("event outbox relay started");
99120
100121 let mut push_tasks = Vec::new();
@@ -167,9 +188,8 @@
167188 }
168189 };
169190 let (api_result, ()) = tokio::join!(api, signal);
170 if let Err(error) = relay.shutdown().await {
171 warn!(%error, "shutting down the event outbox relay failed");
172 }
191 let _ = relay_stop.send(());
192 let relay_failed = relay_task.await.unwrap_or(true);
173193 for task in push_tasks {
174194 let _ = task.await;
175195 }
@@ -180,6 +200,10 @@
180200 warn!(%error, "flushing telemetry failed");
181201 }
182202
203 if relay_failed {
204 return Err(anyhow::anyhow!("event outbox relay stopped unexpectedly"));
205 }
206
183207 info!("apihost shut down");
184208 Ok(())
185209}