move: improve source service error handling (#14106)

## Description 

Causes the server to shutdown if we see a fatal error when watching for
on-chain upgrades (e.g., a fatal error such as losing a subscription
connection).

## Test Plan 

Tested manually since it's difficult to test forcing a connection drop
to observe the server crashes as expected. I ran a local network and
killed it to see the source server also die. In follow up PR there is
additional testing for the functionality we care about most:
invalidating sources if we observe a upgrade transaction in the
network..

---
If your changes are not user-facing and not a breaking change, you can
skip the following section. Otherwise, please indicate what changed, and
then add to the Release Notes section as highlighted during the release
process.

### Type of Change (Check all that apply)

- [ ] protocol change
- [ ] user-visible impact
- [ ] breaking change for a client SDKs
- [ ] breaking change for FNs (FN binary must upgrade)
- [ ] breaking change for validators or node operators (must upgrade
binaries)
- [ ] breaking change for on-chain data layout
- [ ] necessitate either a data wipe or data migration

### Release notes

Author: rvantonder

crates/sui-source-validation-service/src/main.rs

@@ -27,9 +27,18 @@ pub async fn main() -> anyhow::Result<()> {
     let start = tokio::time::Instant::now();
     let sources = initialize(&package_config, tmp_dir.path()).await?;
     info!("verification complete in {:?}", start.elapsed());
-    info!("serving on {}", host_port());
+
     let app_state = Arc::new(RwLock::new(AppState { sources }));
     let app_state_copy = app_state.clone();
-    tokio::spawn(async move { watch_for_upgrades(&package_config, app_state).await });
-    serve(app_state_copy)?.await.map_err(anyhow::Error::from)
+    let mut threads = vec![];
+    let watcher = tokio::spawn(async move { watch_for_upgrades(&package_config, app_state).await });
+    threads.push(watcher);
+    let server = tokio::spawn(async { serve(app_state_copy)?.await.map_err(anyhow::Error::from) });
+    threads.push(server);
+
+    info!("serving on {}", host_port());
+    for t in threads {
+        t.await.unwrap()?;
+    }
+    Ok(())
 }