storcon: add use_local_compute_notifications flag (#11333)

## Problem

While working on bulk import, I want to use the `control-plane-url` flag
for a different request.
Currently, the local compute hook is used whenever no control plane is
specified in the config.
My test requires local compute notifications and a configured
`control-plane-url` which isn't supported.

## Summary of changes

Add a `use-local-compute-notifications` flag. When this is set, we use
the local flow regardless of other config values.
It's enabled by default in neon_local and disabled by default in all
other envs. I had to turn the flag off in tests
that wish to bypass the local flow, but that's expected.

---------

Co-authored-by: Arpad Müller <arpad-m@users.noreply.github.com>

storage_controller/src/compute_hook.rs

@@ -624,16 +624,19 @@ impl ComputeHook {
             MaybeSendResult::Transmit((request, lock)) => (request, lock),
         };
 
-        let compute_hook_url = if let Some(control_plane_url) = &self.config.control_plane_url {
-            Some(if control_plane_url.ends_with('/') {
-                format!("{control_plane_url}notify-attach")
+        let result = if !self.config.use_local_compute_notifications {
+            let compute_hook_url = if let Some(control_plane_url) = &self.config.control_plane_url {
+                Some(if control_plane_url.ends_with('/') {
+                    format!("{control_plane_url}notify-attach")
+                } else {
+                    format!("{control_plane_url}/notify-attach")
+                })
             } else {
-                format!("{control_plane_url}/notify-attach")
-            })
-        } else {
-            self.config.compute_hook_url.clone()
-        };
-        let result = if let Some(notify_url) = &compute_hook_url {
+                self.config.compute_hook_url.clone()
+            };
+
+            // We validate this at startup
+            let notify_url = compute_hook_url.as_ref().unwrap();
             self.do_notify(notify_url, &request, cancel).await
         } else {
             self.do_notify_local(&request).await.map_err(|e| {

storage_controller/src/main.rs

@@ -203,6 +203,11 @@ struct Cli {
     /// Trusted root CA certificates to use in https APIs.
     #[arg(long)]
     ssl_ca_file: Option<PathBuf>,
+
+    /// Neon local specific flag. When set, ignore [`Cli::control_plane_url`] and deliver
+    /// the compute notification directly (instead of via control plane).
+    #[arg(long, default_value = "false")]
+    use_local_compute_notifications: bool,
 }
 
 enum StrictMode {
@@ -368,6 +373,9 @@ async fn async_main() -> anyhow::Result<()> {
                 "neither `--compute-hook-url` nor `--control-plane-url` are set: this is only permitted in `--dev` mode"
             );
         }
+        StrictMode::Strict if args.use_local_compute_notifications => {
+            anyhow::bail!("`--use-local-compute-notifications` is only permitted in `--dev` mode");
+        }
         StrictMode::Strict => {
             tracing::info!("Starting in strict mode: configuration is OK.")
         }
@@ -427,6 +435,7 @@ async fn async_main() -> anyhow::Result<()> {
         use_https_safekeeper_api: args.use_https_safekeeper_api,
         ssl_ca_certs,
         timelines_onto_safekeepers: args.timelines_onto_safekeepers,
+        use_local_compute_notifications: args.use_local_compute_notifications,
     };
 
     // Validate that we can connect to the database

storage_controller/src/service.rs

@@ -448,6 +448,8 @@ pub struct Config {
     pub ssl_ca_certs: Vec<Certificate>,
 
     pub timelines_onto_safekeepers: bool,
+
+    pub use_local_compute_notifications: bool,
 }
 
 impl From<DatabaseError> for ApiError {