Skip to main content

slim_controller/
service.rs

1// Copyright AGNTCY Contributors (https://github.com/agntcy)
2// SPDX-License-Identifier: Apache-2.0
3
4use std::collections::{HashMap, VecDeque};
5use std::pin::Pin;
6use std::sync::Arc;
7
8use std::time::Duration;
9
10use display_error_chain::ErrorChainExt;
11use slim_config::server::ServerConfig;
12use slim_session::subscription_manager::SubscriptionManager;
13use tokio::sync::mpsc;
14use tokio::task::JoinHandle;
15use tokio_stream::{Stream, StreamExt, wrappers::ReceiverStream};
16use tokio_util::sync::CancellationToken;
17use tonic::{Request, Response, Status};
18use tracing::{debug, error, info};
19
20use crate::api::proto::api::v1::control_message::Payload;
21use crate::api::proto::api::v1::controller_service_server::ControllerServiceServer;
22use crate::api::proto::api::v1::{
23    self, AuthMethod, ConnectionDetails, ConnectionDirection, ConnectionListResponse,
24    ConnectionType, RouteListResponse,
25};
26use crate::api::proto::api::v1::{
27    ConnectionEntry, ControlMessage, RouteEntry,
28    controller_service_client::ControllerServiceClient,
29    controller_service_server::ControllerService as GrpcControllerService,
30};
31use crate::errors::ControllerError;
32use prost_types::Struct;
33use slim_config::client::{
34    ClientConfig, RequiredAuthMethod, ServerConnectionConfig, TransportChannel,
35};
36use slim_config::server::AuthenticationConfig;
37use slim_datapath::api::{
38    MessageType::Link as LinkMessageType, MessageType::Subscribe,
39    MessageType::SubscriptionAck as SubscriptionAckType, MessageType::Unsubscribe,
40    ProtoMessage as DataPlaneMessage,
41};
42use slim_datapath::api::{NameId, ProtoName};
43use slim_datapath::message_processing::MessageProcessor;
44use slim_datapath::messages::utils::SlimHeaderFlags;
45use slim_datapath::tables::{ConnType, SubscriptionTable};
46
47type TxChannel = mpsc::Sender<Result<ControlMessage, Status>>;
48type TxChannels = HashMap<String, TxChannel>;
49
50/// Maximum number of queued subscription notifications
51const MAX_QUEUED_NOTIFICATIONS: usize = 1000;
52
53/// Timeout for waiting on a subscription ack from the datapath.
54const SUBSCRIPTION_ACK_TIMEOUT: Duration = Duration::from_secs(30);
55
56use slim_auth::auth_provider::AuthProvider;
57use slim_auth::traits::TokenProvider;
58
59/// Settings struct for creating a ControlPlane instance
60#[derive(Clone)]
61pub struct ControlPlaneSettings {
62    /// Node ID of this SLIM instance
63    pub id: String,
64    /// Optional group name
65    pub group_name: Option<String>,
66    /// Server configurations
67    pub servers: Vec<ServerConfig>,
68    /// Client configurations
69    pub clients: Vec<ClientConfig>,
70    /// Client configurations for server nodes
71    pub outbound_clients: Vec<ClientConfig>,
72    /// Message processor instance
73    pub message_processor: MessageProcessor,
74    /// array of connection details used by the control
75    /// plane to store the connection settings (e.g., TLS settings).
76    pub connection_details: Vec<ConnectionDetails>,
77    /// Optional auth provider for generating registration credentials.
78    pub auth_provider: Option<AuthProvider>,
79}
80
81/// Inner structure for the controller service
82/// This structure holds the internal state of the controller service,
83/// including the ID, message processor, connections, and channels.
84/// It is normally wrapped in an Arc to allow shared ownership across multiple threads.
85struct ControllerServiceInternal {
86    /// Node ID of this SLIM instance
87    id: String,
88
89    /// optional group name
90    group_name: Option<String>,
91
92    /// underlying message processor
93    message_processor: MessageProcessor,
94
95    /// channel to send messages into the datapath
96    tx_slim: mpsc::Sender<Result<DataPlaneMessage, Status>>,
97
98    /// channels to send control messages
99    tx_channels: parking_lot::RwLock<TxChannels>,
100
101    /// cancellation token for graceful shutdown
102    cancellation_tokens: parking_lot::RwLock<HashMap<String, CancellationToken>>,
103
104    /// drain watch channel
105    drain_watch: parking_lot::RwLock<Option<drain::Watch>>,
106
107    /// queue for pending subscription notifications when connections are down
108    pending_notifications: Arc<parking_lot::Mutex<VecDeque<ControlMessage>>>,
109
110    /// Manages pending subscription ack tracking (id generation, registration, resolution).
111    subscription_manager: SubscriptionManager,
112
113    /// connection details used by control plane to store connection settings
114    connection_details: Vec<ConnectionDetails>,
115
116    /// Maps (subscription_name, link_id) → subscription_id for route tracking
117    route_subscription_ids: parking_lot::Mutex<HashMap<(ProtoName, u64), u64>>,
118
119    /// Reverse index: link_id → conn_id for O(1) lookup.
120    /// Populated lazily on first resolve and eagerly on connection create/delete.
121    link_id_to_conn_id: parking_lot::RwLock<HashMap<String, u64>>,
122
123    /// JoinHandles for control-plane stream processing tasks, keyed by endpoint.
124    stream_handles: parking_lot::Mutex<HashMap<String, tokio::task::JoinHandle<()>>>,
125
126    /// connection details for CP reconciled outbound data-plane connections
127    outbound_clients: Vec<ClientConfig>,
128
129    /// Optional auth provider for generating registration credentials.
130    auth_provider: Option<AuthProvider>,
131}
132
133#[derive(Clone)]
134struct ControllerService {
135    /// internal service state
136    inner: Arc<ControllerServiceInternal>,
137}
138
139/// The ControlPlane service is the main entry point for the controller service.
140pub struct ControlPlane {
141    /// servers
142    servers: Vec<ServerConfig>,
143
144    /// clients
145    clients: Vec<ClientConfig>,
146
147    /// drain signal channel
148    drain_signal: parking_lot::RwLock<Option<drain::Signal>>,
149
150    /// controller
151    controller: ControllerService,
152
153    /// channel to receive message from the datapath
154    /// to be used in listen_from_data_plane
155    rx_slim_option: Option<mpsc::Receiver<Result<DataPlaneMessage, Status>>>,
156}
157
158/// ControllerServiceInternal implements Drop trait to cancel all running listeners and
159/// clean up resources.
160impl Drop for ControlPlane {
161    fn drop(&mut self) {
162        // cancel all running listeners
163        for (_endpoint, token) in self.controller.inner.cancellation_tokens.write().drain() {
164            token.cancel();
165        }
166    }
167}
168
169pub(crate) fn from_server_config(server_config: &ServerConfig) -> ConnectionDetails {
170    let mut endpoint = server_config.endpoint.clone();
171    let mut external_endpoint = None;
172    let mut spire_socket_path = None;
173    let mut trust_domain = None;
174    let mut remaining_fields = std::collections::BTreeMap::new();
175
176    if let Some(m) = &server_config.metadata {
177        for (k, v) in &m.inner {
178            match k.as_str() {
179                "local_endpoint" => {
180                    if let Some(s) = v.as_str()
181                        && !s.is_empty()
182                    {
183                        endpoint = s.to_string();
184                    }
185                }
186                "external_endpoint" => {
187                    if let Some(s) = v.as_str()
188                        && !s.is_empty()
189                    {
190                        external_endpoint = Some(s.to_string());
191                    }
192                }
193                "spire_socket_path" => {
194                    if let Some(s) = v.as_str()
195                        && !s.is_empty()
196                    {
197                        spire_socket_path = Some(s.to_string());
198                    }
199                }
200                "trust_domain" => {
201                    if let Some(s) = v.as_str()
202                        && !s.is_empty()
203                    {
204                        trust_domain = Some(s.to_string());
205                    }
206                }
207                _ => {
208                    remaining_fields.insert(k.clone(), prost_types::Value::from(v));
209                }
210            }
211        }
212    }
213
214    let tls_required = !server_config.tls_setting.insecure || spire_socket_path.is_some();
215    let auth_method = if spire_socket_path.is_some()
216        || trust_domain.is_some()
217        || matches!(server_config.auth, AuthenticationConfig::Spire(_))
218    {
219        AuthMethod::Spire
220    } else {
221        match &server_config.auth {
222            AuthenticationConfig::Basic(_) => AuthMethod::Basic,
223            AuthenticationConfig::Jwt(_) => AuthMethod::Jwt,
224            _ => AuthMethod::None,
225        }
226    } as i32;
227    let metadata = if remaining_fields.is_empty() {
228        None
229    } else {
230        Some(Struct {
231            fields: remaining_fields,
232        })
233    };
234
235    ConnectionDetails {
236        endpoint,
237        external_endpoint,
238        tls_required,
239        auth_method,
240        spire_trust_domain: trust_domain,
241        metadata,
242    }
243}
244
245/// ControlPlane implements the service trait for the controller service.
246impl ControlPlane {
247    /// Create a new ControlPlane service instance
248    /// This function initializes the ControlPlane with the given ID, servers, clients, and message processor.
249    /// It also sets up the internal state, including the connections and channels.
250    /// # Arguments
251    /// * `id` - The ID of the SLIM instance.
252    /// * `servers` - A vector of server configurations.
253    /// * `clients` - A vector of client configurations.
254    /// * `drain_rx` - A drain watch channel for graceful shutdown.
255    /// * `message_processor` - An Arc to the message processor instance.
256    /// * `pubsub_servers` - A slice of server configurations for pub/sub connections.
257    /// # Returns
258    /// A new instance of ControlPlane.
259    pub fn new(config: ControlPlaneSettings) -> Self {
260        // create local connection with the message processor
261        let (_, tx_slim, rx_slim) = config
262            .message_processor
263            .register_local_connection(true)
264            .unwrap();
265
266        let (signal, watch) = drain::channel();
267
268        ControlPlane {
269            servers: config.servers,
270            clients: config.clients,
271            controller: ControllerService {
272                inner: Arc::new(ControllerServiceInternal {
273                    id: config.id,
274                    group_name: config.group_name,
275                    message_processor: config.message_processor,
276                    subscription_manager: SubscriptionManager::new(tx_slim.clone()),
277                    tx_slim,
278                    tx_channels: parking_lot::RwLock::new(HashMap::new()),
279                    cancellation_tokens: parking_lot::RwLock::new(HashMap::new()),
280                    drain_watch: parking_lot::RwLock::new(Some(watch)),
281                    pending_notifications: Arc::new(parking_lot::Mutex::new(VecDeque::new())),
282                    connection_details: config.connection_details,
283                    route_subscription_ids: parking_lot::Mutex::new(HashMap::new()),
284                    link_id_to_conn_id: parking_lot::RwLock::new(HashMap::new()),
285                    stream_handles: parking_lot::Mutex::new(HashMap::new()),
286                    outbound_clients: config.outbound_clients,
287                    auth_provider: config.auth_provider,
288                }),
289            },
290            drain_signal: parking_lot::RwLock::new(Some(signal)),
291            rx_slim_option: Some(rx_slim),
292        }
293    }
294
295    /// Take an existing ControlPlane instance and return a new one with the provided clients.
296    pub fn with_clients(mut self, clients: Vec<ClientConfig>) -> Self {
297        self.clients = clients;
298        self
299    }
300
301    /// Take an existing ControlPlane instance and return a new one with the provided servers.
302    pub fn with_servers(mut self, servers: Vec<ServerConfig>) -> Self {
303        self.servers = servers;
304        self
305    }
306
307    /// Run the clients and servers of the ControlPlane service.
308    /// This function starts all the servers and clients defined in the ControlPlane.
309    /// # Returns
310    /// A Result indicating success or failure of the operation.
311    /// # Errors
312    /// If there is an error starting any of the servers or clients, it will return a ControllerError.
313    pub async fn run(&mut self) -> Result<(), ControllerError> {
314        let rx = self
315            .rx_slim_option
316            .take()
317            .ok_or(ControllerError::AlreadyStarted)?;
318
319        // Collect servers to avoid borrowing self both mutably and immutably
320        let servers = self.servers.clone();
321        let clients = self.clients.clone();
322
323        // run all servers
324        for server in servers {
325            self.run_server(server).await?;
326        }
327
328        // run all clients
329        for client in clients {
330            self.run_client(client).await?;
331        }
332
333        self.listen_from_data_plane(rx).await?;
334
335        Ok(())
336    }
337
338    pub async fn deregister(&self) -> Result<(), ControllerError> {
339        let node_id = self.controller.inner.id.clone();
340        let deregister_msg = ControlMessage {
341            message_id: uuid::Uuid::new_v4().to_string(),
342            payload: Some(Payload::DeregisterNodeRequest(v1::DeregisterNodeRequest {
343                node: Some(v1::Node { id: node_id }),
344            })),
345        };
346        let channels: Vec<(String, TxChannel)> = self
347            .controller
348            .inner
349            .tx_channels
350            .read()
351            .iter()
352            .map(|(ep, tx)| (ep.clone(), tx.clone()))
353            .collect();
354        for (endpoint, tx) in channels {
355            if let Err(e) = tx.send(Ok(deregister_msg.clone())).await {
356                error!(%endpoint, error = %e, "failed to send deregister request");
357            }
358        }
359        Ok(())
360    }
361
362    pub async fn shutdown(&self) -> Result<(), ControllerError> {
363        // Get signal drain
364        let signal = self
365            .drain_signal
366            .write()
367            .take()
368            .ok_or(ControllerError::AlreadyStopped)?;
369
370        // Stop everything using the cancellation tokens
371        self.controller
372            .inner
373            .cancellation_tokens
374            .write()
375            .drain()
376            .for_each(|(endpoint, token)| {
377                info!(%endpoint, "stopping");
378                token.cancel();
379            });
380
381        // Drop watch channel
382        self.controller.inner.drain_watch.write().take();
383
384        // Wait for drain to complete
385        signal.drain().await;
386
387        Ok(())
388    }
389
390    async fn listen_from_data_plane(
391        &mut self,
392        mut rx: mpsc::Receiver<Result<DataPlaneMessage, Status>>,
393    ) -> Result<(), ControllerError> {
394        let cancellation_token = CancellationToken::new();
395        let cancellation_token_clone = cancellation_token.clone();
396
397        self.controller
398            .inner
399            .cancellation_tokens
400            .write()
401            .insert("DATA_PLANE".to_string(), cancellation_token_clone);
402
403        let clients = self.clients.clone();
404        let controller = self.controller.clone();
405
406        // Get a drain watch clone
407        let watch = self.controller.drain_watch()?;
408
409        debug!("Starting data plane listener");
410        tokio::spawn(async move {
411            let mut drain_fut = std::pin::pin!(watch.signaled());
412            loop {
413                tokio::select! {
414                    next = rx.recv() => {
415                        match next {
416                            Some(res) => {
417                                match res {
418                                    Ok(msg) => {
419                                        debug!("Received message {:?} from data plane, forwarding to control plane", msg);
420                                        match msg.get_type() {
421                                            Subscribe(_) => {
422                                                controller.handle_subscribe_message(msg.get_dst(), &clients).await;
423                                            }
424                                            Unsubscribe(_) => {
425                                                controller.handle_unsubscribe_message(msg.get_dst(), &clients).await;
426                                            }
427                                            SubscriptionAckType(_) => {
428                                                controller.inner.subscription_manager.resolve_ack(msg.get_subscription_ack());
429                                            }
430                                            LinkMessageType(link) => {
431                                                controller.handle_link_received(link, &clients).await;
432                                            }
433                                            _ => {
434                                                debug!("Ignoring unexpected message type from dataplane: {:?}", msg.get_type());
435                                            }
436                                        }
437                                    }
438                                    Err(e) => {
439                                        error!(error = %e.chain(), "received error from the data plane");
440                                        continue;
441                                    }
442                                }
443                            }
444                            None => {
445                                debug!("Data plane receiver channel closed.");
446                                break;
447                            }
448                        }
449                    }
450                    _ = cancellation_token.cancelled() => {
451                        debug!("shutting down stream on cancellation token");
452                        break;
453                    }
454                    _ = &mut drain_fut => {
455                        debug!("shutting down stream on drain");
456                        break;
457                    }
458                }
459            }
460        });
461        Ok(())
462    }
463
464    /// Stop the ControlPlane service.
465    /// This function stops all running listeners and cancels any ongoing operations.
466    /// It cleans up the internal state and ensures that all resources are released properly.
467    pub fn stop(&mut self) {
468        info!("stopping controller service");
469
470        // cancel all running listeners
471        for (endpoint, token) in self.controller.inner.cancellation_tokens.write().drain() {
472            info!(%endpoint, "stopping");
473            token.cancel();
474        }
475    }
476
477    /// Run a client configuration.
478    /// This function connects to the control plane using the provided client configuration.
479    /// It checks if the client is already running and if not, it starts a new connection.
480    async fn run_client(&mut self, client: ClientConfig) -> Result<(), ControllerError> {
481        if self
482            .controller
483            .inner
484            .cancellation_tokens
485            .read()
486            .contains_key(&client.endpoint)
487        {
488            return Err(ControllerError::ClientAlreadyRunning(client.endpoint));
489        }
490
491        let cancellation_token = CancellationToken::new();
492
493        let tx = self
494            .controller
495            .connect(client.clone(), cancellation_token.clone())
496            .await?;
497
498        // Store the cancellation token in the controller service
499        self.controller
500            .inner
501            .cancellation_tokens
502            .write()
503            .insert(client.endpoint.clone(), cancellation_token);
504
505        // Store the sender in the tx_channels map
506        self.controller
507            .inner
508            .tx_channels
509            .write()
510            .insert(client.endpoint.clone(), tx);
511
512        // return the sender for control messages
513        Ok(())
514    }
515
516    /// Run a server configuration.
517    /// This function starts a server using the provided server configuration.
518    /// It checks if the server is already running and if not, it starts a new server.
519    pub async fn run_server(&mut self, config: ServerConfig) -> Result<(), ControllerError> {
520        // Check if the server is already running
521        if self
522            .controller
523            .inner
524            .cancellation_tokens
525            .read()
526            .contains_key(&config.endpoint)
527        {
528            error!(endpoint = config.endpoint, "server is already running",);
529            return Err(ControllerError::ServerAlreadyRunning(config.endpoint));
530        }
531
532        let token = config
533            .run_grpc_server(
534                &[ControllerServiceServer::new(self.controller.clone())],
535                self.controller.drain_watch()?,
536            )
537            .await?;
538
539        // Store the cancellation token in the controller service
540        self.controller
541            .inner
542            .cancellation_tokens
543            .write()
544            .insert(config.endpoint.clone(), token.clone());
545
546        info!(%config.endpoint, "started controlplane server");
547
548        Ok(())
549    }
550}
551
552impl ControllerService {
553    fn resolve_connection_by_link_id(&self, link_id: &str) -> Result<Option<u64>, String> {
554        let cached = self.inner.link_id_to_conn_id.read().get(link_id).copied();
555        if let Some(conn_id) = cached {
556            if self
557                .inner
558                .message_processor
559                .connection_table()
560                .get(conn_id)
561                .is_some_and(|conn| conn.link_id().as_deref() == Some(link_id))
562            {
563                return Ok(Some(conn_id));
564            }
565            self.inner.link_id_to_conn_id.write().remove(link_id);
566        }
567
568        let mut resolved: Option<u64> = None;
569        self.inner
570            .message_processor
571            .connection_table()
572            .for_each(|id, conn| {
573                if conn.link_id().as_deref() == Some(link_id) && resolved.is_none() {
574                    resolved = Some(id);
575                }
576            });
577
578        if let Some(conn_id) = resolved {
579            self.inner
580                .link_id_to_conn_id
581                .write()
582                .insert(link_id.to_string(), conn_id);
583        }
584
585        Ok(resolved)
586    }
587
588    fn disconnect_connection_by_link_id(&self, link_id: &str) -> Result<(), String> {
589        if link_id.trim().is_empty() {
590            return Err("link_id cannot be empty".to_string());
591        }
592
593        let conn_id = match self.resolve_connection_by_link_id(link_id)? {
594            Some(id) => id,
595            None => {
596                return Err(format!("Connection with link_id {} not found", link_id));
597            }
598        };
599
600        if let Err(e) = self.inner.message_processor.disconnect(conn_id) {
601            // Best-effort delete: local/control-plane connections can lack config_data.
602            info!(
603                link_id = %link_id,
604                conn_id,
605                error = %e,
606                "Disconnect returned an error; continuing delete flow"
607            );
608        }
609
610        self.inner.link_id_to_conn_id.write().remove(link_id);
611        self.inner
612            .route_subscription_ids
613            .lock()
614            .retain(|(_name, cid), _| *cid != conn_id);
615
616        info!(link_id = %link_id, conn_id, "Successfully deleted connection by link_id");
617        Ok(())
618    }
619
620    fn resolve_route_connection(&self, route: &v1::Route) -> Result<Option<u64>, String> {
621        if let Some(link_id) = &route.link_id {
622            let trimmed = link_id.trim();
623            if !trimmed.is_empty() {
624                return self.resolve_connection_by_link_id(trimmed);
625            }
626        }
627
628        Ok(None)
629    }
630
631    async fn diff_connections(
632        &self,
633        desired_connections: &[v1::Connection],
634    ) -> Vec<v1::ConnectionAck> {
635        let mut connections_status = Vec::new();
636
637        let desired_link_ids: std::collections::HashSet<String> = desired_connections
638            .iter()
639            .map(|c| c.link_id.clone())
640            .collect();
641
642        let mut live_outgoing_link_ids: Vec<String> = Vec::new();
643        self.inner
644            .message_processor
645            .connection_table()
646            .for_each(|_id, conn| {
647                // Only manage Remote connections (CP-managed inter-group links).
648                // Peer connections are managed by the peer sync system and must not
649                // be deleted by the reconciler.
650                if conn.is_outgoing()
651                    && conn.connection_type() == slim_datapath::tables::ConnType::Remote
652                    && let Some(lid) = conn.link_id()
653                    && !lid.is_empty()
654                {
655                    live_outgoing_link_ids.push(lid);
656                }
657            });
658
659        for link_id in &live_outgoing_link_ids {
660            if desired_link_ids.contains(link_id) {
661                continue;
662            }
663            info!(link_id = %link_id, "desired state: removing connection");
664            let mut success = true;
665            let mut error_msg = String::new();
666            if let Err(err) = self.disconnect_connection_by_link_id(link_id) {
667                success = false;
668                error_msg = err;
669            }
670            connections_status.push(v1::ConnectionAck {
671                link_id: link_id.clone(),
672                success,
673                error_msg,
674            });
675        }
676
677        for conn in desired_connections {
678            let link_id = &conn.link_id;
679            if link_id.is_empty() {
680                continue;
681            }
682
683            let already_exists = match self.resolve_connection_by_link_id(link_id) {
684                Ok(Some(_)) => true,
685                Ok(None) => false,
686                Err(err) => {
687                    connections_status.push(v1::ConnectionAck {
688                        link_id: link_id.clone(),
689                        success: false,
690                        error_msg: err,
691                    });
692                    continue;
693                }
694            };
695
696            if already_exists {
697                connections_status.push(v1::ConnectionAck {
698                    link_id: link_id.clone(),
699                    success: true,
700                    error_msg: String::new(),
701                });
702                continue;
703            }
704
705            info!(?conn, "desired state: creating connection");
706            let mut success = true;
707            let mut error_msg = String::new();
708
709            match serde_json::from_str::<ServerConnectionConfig>(&conn.config_data) {
710                Err(e) => {
711                    success = false;
712                    error_msg = format!("Failed to parse config: {}", e);
713                }
714                Ok(server_config) => {
715                    let mut client_config = self
716                        .inner
717                        .outbound_clients
718                        .iter()
719                        .find(|c| c.endpoint == server_config.endpoint)
720                        .cloned()
721                        .unwrap_or_default();
722                    match client_config.merge_server_requirements(&server_config) {
723                        Err(err) => {
724                            success = false;
725                            error_msg = format!(
726                                "Failed to merge connection config to client config: {}",
727                                err
728                            );
729                        }
730                        Ok(()) => {
731                            if matches!(
732                                server_config.auth_method,
733                                // Spire config it not mandatory. If not set falls back to default
734                                // socket path.
735                                RequiredAuthMethod::Basic | RequiredAuthMethod::Jwt
736                            ) && !self
737                                .inner
738                                .outbound_clients
739                                .iter()
740                                .any(|c| c.endpoint == server_config.endpoint)
741                            {
742                                success = false;
743                                error_msg = format!(
744                                    "no local credentials configured for {}",
745                                    server_config.endpoint
746                                );
747                            }
748                            if success {
749                                client_config.link_id = link_id.clone();
750                                client_config.connection_type = ConnType::Remote;
751                                match self
752                                    .inner
753                                    .message_processor
754                                    .connect(client_config, None, None)
755                                    .await
756                                {
757                                    Err(e) => {
758                                        success = false;
759                                        error_msg = format!("Connection failed: {}", e);
760                                    }
761                                    Ok(conn_id) => {
762                                        self.inner
763                                            .link_id_to_conn_id
764                                            .write()
765                                            .insert(link_id.clone(), conn_id.1);
766                                        info!(
767                                            link_id = %link_id,
768                                            "Successfully created connection"
769                                        );
770                                    }
771                                }
772                            }
773                        }
774                    }
775                }
776            }
777
778            connections_status.push(v1::ConnectionAck {
779                link_id: link_id.clone(),
780                success,
781                error_msg,
782            });
783        }
784
785        connections_status
786    }
787
788    fn resolve_desired_routes<'a>(
789        &self,
790        desired_routes: &'a [v1::Route],
791    ) -> (HashMap<(ProtoName, u64), &'a v1::Route>, Vec<v1::RouteAck>) {
792        type SubKey = (ProtoName, u64);
793        let mut desired_subs: HashMap<SubKey, &v1::Route> = HashMap::new();
794        let mut failures: Vec<v1::RouteAck> = Vec::new();
795
796        for sub in desired_routes {
797            match self.resolve_route_connection(sub) {
798                Ok(Some(conn_id)) => {
799                    let name = sub.name.clone().unwrap();
800                    desired_subs.insert((name, conn_id), sub);
801                }
802                Ok(None) => {
803                    failures.push(v1::RouteAck {
804                        route: Some(sub.clone()),
805                        success: false,
806                        error_msg: "connection not found".to_string(),
807                    });
808                }
809                Err(err) => {
810                    failures.push(v1::RouteAck {
811                        route: Some(sub.clone()),
812                        success: false,
813                        error_msg: err,
814                    });
815                }
816            }
817        }
818
819        (desired_subs, failures)
820    }
821
822    async fn delete_stale_subscriptions(
823        &self,
824        desired_subs: &HashMap<(ProtoName, u64), &v1::Route>,
825    ) -> Vec<v1::RouteAck> {
826        let active_subs: Vec<((ProtoName, u64), u64)> = self
827            .inner
828            .route_subscription_ids
829            .lock()
830            .iter()
831            .map(|((name, conn_id), sub_id)| ((name.clone(), *conn_id), *sub_id))
832            .collect();
833
834        let stale: Vec<_> = active_subs
835            .into_iter()
836            .filter(|((name, conn_id), _)| !desired_subs.contains_key(&(name.clone(), *conn_id)))
837            .collect();
838
839        let futs = stale.iter().map(|((name, conn_id), sub_id)| {
840            let name = name.clone();
841            let conn_id = *conn_id;
842            let sub_id = *sub_id;
843            async move {
844                let conn_alive = self
845                    .inner
846                    .message_processor
847                    .connection_table()
848                    .get(conn_id)
849                    .is_some();
850
851                let (success, error_msg) = if conn_alive {
852                    let source = name.clone().with_id(0);
853                    let unsub_msg = DataPlaneMessage::builder()
854                        .source(source)
855                        .destination(name.clone())
856                        .identity("")
857                        .flags(SlimHeaderFlags::default().with_recv_from(conn_id))
858                        .build_unsubscribe()
859                        .unwrap();
860
861                    match self
862                        .send_unsubscribe_message_with_ack(unsub_msg, sub_id)
863                        .await
864                    {
865                        Ok(()) => (true, String::new()),
866                        Err(err) => (false, format!("Failed to unsubscribe: {}", err)),
867                    }
868                } else {
869                    (true, String::new())
870                };
871
872                (name, conn_id, success, error_msg)
873            }
874        });
875
876        let results = futures::future::join_all(futs).await;
877
878        let mut routes_status = Vec::with_capacity(results.len());
879        for (name, conn_id, success, error_msg) in results {
880            if success {
881                self.inner
882                    .route_subscription_ids
883                    .lock()
884                    .remove(&(name.clone(), conn_id));
885            }
886
887            routes_status.push(v1::RouteAck {
888                route: Some(v1::Route {
889                    name: Some(name.clone()),
890                    link_id: None,
891                    direction: None,
892                }),
893                success,
894                error_msg,
895            });
896        }
897
898        routes_status
899    }
900
901    async fn create_new_subscriptions(
902        &self,
903        desired_subs: &HashMap<(ProtoName, u64), &v1::Route>,
904    ) -> Vec<v1::RouteAck> {
905        let mut routes_status = Vec::new();
906
907        let to_create: Vec<((ProtoName, u64), v1::Route)> = desired_subs
908            .iter()
909            .filter(|((name, conn_id), _)| {
910                !self
911                    .inner
912                    .route_subscription_ids
913                    .lock()
914                    .contains_key(&(name.clone(), *conn_id))
915            })
916            .map(|((name, conn_id), sub)| ((name.clone(), *conn_id), (*sub).clone()))
917            .collect();
918
919        // Report already-active subscriptions as success.
920        for ((name, conn_id), sub) in desired_subs {
921            let dominated = to_create
922                .iter()
923                .any(|((n, c), _)| n == name && *c == *conn_id);
924            if !dominated {
925                routes_status.push(v1::RouteAck {
926                    route: Some((*sub).clone()),
927                    success: true,
928                    error_msg: String::new(),
929                });
930            }
931        }
932
933        let futs = to_create.iter().map(|((name, conn_id), sub)| {
934            let name = name.clone();
935            let conn_id = *conn_id;
936            let sub = sub.clone();
937            async move {
938                let source = name.clone().with_id(0);
939                let sub_msg = DataPlaneMessage::builder()
940                    .source(source)
941                    .destination(name.clone())
942                    .identity("")
943                    .flags(SlimHeaderFlags::default().with_recv_from(conn_id))
944                    .build_subscribe()
945                    .unwrap();
946
947                let result = self.send_subscribe_message_with_ack(sub_msg).await;
948                (name, conn_id, sub, result)
949            }
950        });
951
952        let results = futures::future::join_all(futs).await;
953
954        for (name, conn_id, sub, result) in results {
955            let (success, error_msg) = match result {
956                Ok(subscription_id) => {
957                    self.inner
958                        .route_subscription_ids
959                        .lock()
960                        .insert((name, conn_id), subscription_id);
961                    info!(?sub, "desired state: created route");
962                    (true, String::new())
963                }
964                Err(err) => (false, format!("Failed to subscribe: {}", err)),
965            };
966
967            routes_status.push(v1::RouteAck {
968                route: Some(sub),
969                success,
970                error_msg,
971            });
972        }
973
974        routes_status
975    }
976
977    /// Handle new control messages.
978    async fn handle_new_control_message(
979        &self,
980        msg: ControlMessage,
981        tx: &mpsc::Sender<Result<ControlMessage, Status>>,
982    ) -> Result<(), ControllerError> {
983        match msg.payload {
984            Some(ref payload) => {
985                match payload {
986                    Payload::ConfigCommand(config) => {
987                        let (connections_status, routes_status) = if config.reconcile {
988                            let connections_status =
989                                self.diff_connections(&config.connections_to_create).await;
990                            let (desired_subs, mut routes_status) =
991                                self.resolve_desired_routes(&config.routes_to_set);
992                            routes_status
993                                .extend(self.delete_stale_subscriptions(&desired_subs).await);
994                            routes_status
995                                .extend(self.create_new_subscriptions(&desired_subs).await);
996                            (connections_status, routes_status)
997                        } else {
998                            let mut connections_status = Vec::new();
999                            let mut routes_status = Vec::new();
1000
1001                            // Process connections to delete by link_id.
1002                            for link_id in &config.connections_to_delete {
1003                                info!(link_id = %link_id, "received a connection to delete");
1004                                let mut connection_success = true;
1005                                let mut connection_error_msg = String::new();
1006
1007                                if let Err(err) = self.disconnect_connection_by_link_id(link_id) {
1008                                    connection_success = false;
1009                                    connection_error_msg = err;
1010                                }
1011
1012                                connections_status.push(v1::ConnectionAck {
1013                                    link_id: link_id.clone(),
1014                                    success: connection_success,
1015                                    error_msg: connection_error_msg,
1016                                });
1017                            }
1018
1019                            // Process connections to create
1020                            for conn in &config.connections_to_create {
1021                                info!(?conn, "received a connection to create");
1022                                let mut connection_success = true;
1023                                let mut connection_error_msg = String::new();
1024
1025                                match serde_json::from_str::<ClientConfig>(&conn.config_data) {
1026                                    Err(e) => {
1027                                        connection_success = false;
1028                                        connection_error_msg =
1029                                            format!("Failed to parse config: {}", e);
1030                                    }
1031                                    Ok(client_config) => {
1032                                        let client_endpoint = &client_config.endpoint;
1033                                        let requested_link_id =
1034                                            if client_config.link_id.trim().is_empty() {
1035                                                String::new()
1036                                            } else {
1037                                                client_config.link_id.clone()
1038                                            };
1039                                        let mut existing_conn_for_link_id = false;
1040
1041                                        if !requested_link_id.is_empty() {
1042                                            match self
1043                                                .resolve_connection_by_link_id(&requested_link_id)
1044                                            {
1045                                                Err(err) => {
1046                                                    connection_success = false;
1047                                                    connection_error_msg = err;
1048                                                }
1049                                                Ok(Some(conn_id)) => {
1050                                                    existing_conn_for_link_id = true;
1051                                                    self.inner
1052                                                        .link_id_to_conn_id
1053                                                        .write()
1054                                                        .insert(requested_link_id.clone(), conn_id);
1055                                                    info!(
1056                                                        link_id = %requested_link_id,
1057                                                        conn_id,
1058                                                        "Connection already exists for link_id"
1059                                                    );
1060                                                }
1061                                                Ok(None) => {}
1062                                            }
1063                                        }
1064
1065                                        if connection_success && !existing_conn_for_link_id {
1066                                            match self
1067                                                .inner
1068                                                .message_processor
1069                                                .connect(client_config.clone(), None, None)
1070                                                .await
1071                                            {
1072                                                Err(e) => {
1073                                                    connection_success = false;
1074                                                    connection_error_msg =
1075                                                        format!("Connection failed: {}", e);
1076                                                }
1077                                                Ok(conn_id) => {
1078                                                    if !requested_link_id.is_empty() {
1079                                                        self.inner
1080                                                            .link_id_to_conn_id
1081                                                            .write()
1082                                                            .insert(
1083                                                                requested_link_id.clone(),
1084                                                                conn_id.1,
1085                                                            );
1086                                                    }
1087                                                    info!(
1088                                                        endpoint = %client_endpoint, "Successfully created connection",
1089                                                    );
1090                                                }
1091                                            }
1092                                        }
1093                                    }
1094                                }
1095
1096                                // Add connection status
1097                                connections_status.push(v1::ConnectionAck {
1098                                    link_id: conn.link_id.clone(),
1099                                    success: connection_success,
1100                                    error_msg: connection_error_msg,
1101                                });
1102                            }
1103
1104                            // Process routes to set
1105                            for route in &config.routes_to_set {
1106                                let mut route_success = true;
1107                                let mut route_error_msg = String::new();
1108
1109                                let conn = self.resolve_route_connection(route);
1110
1111                                if let Ok(Some(conn)) = conn {
1112                                    let name = route.name.clone().unwrap();
1113                                    let source = name.clone().with_id(NameId::NULL_COMPONENT);
1114
1115                                    let msg = DataPlaneMessage::builder()
1116                                        .source(source.clone())
1117                                        .destination(name.clone())
1118                                        .identity("")
1119                                        .flags(SlimHeaderFlags::default().with_recv_from(conn))
1120                                        .build_subscribe()
1121                                        .unwrap();
1122
1123                                    match self.send_subscribe_message_with_ack(msg).await {
1124                                        Ok(subscription_id) => {
1125                                            // Store the subscription_id for later unsubscription
1126                                            self.inner
1127                                                .route_subscription_ids
1128                                                .lock()
1129                                                .insert((name.clone(), conn), subscription_id);
1130                                            info!(?route, "Successfully created route");
1131                                        }
1132                                        Err(err) => {
1133                                            route_success = false;
1134                                            route_error_msg =
1135                                                format!("Failed to subscribe: {}", err);
1136                                        }
1137                                    }
1138                                } else {
1139                                    route_success = false;
1140                                    route_error_msg = match conn {
1141                                        Ok(None) => {
1142                                            format!(
1143                                                "Connection with link_id {} not found",
1144                                                route.link_id.as_deref().unwrap_or("<none>")
1145                                            )
1146                                        }
1147                                        Err(err) => err,
1148                                        _ => "unknown connection lookup error".to_string(),
1149                                    };
1150                                }
1151
1152                                // Add route status
1153                                routes_status.push(v1::RouteAck {
1154                                    route: Some(route.clone()),
1155                                    success: route_success,
1156                                    error_msg: route_error_msg,
1157                                });
1158                            }
1159
1160                            // Process routes to delete
1161                            for route in &config.routes_to_delete {
1162                                let mut route_success = true;
1163                                let mut route_error_msg = String::new();
1164
1165                                let conn = self.resolve_route_connection(route);
1166
1167                                if let Ok(Some(conn)) = conn {
1168                                    let name = route.name.clone().unwrap();
1169                                    let source = name.clone().with_id(NameId::NULL_COMPONENT);
1170
1171                                    let msg = DataPlaneMessage::builder()
1172                                        .source(source.clone())
1173                                        .destination(name.clone())
1174                                        .identity("")
1175                                        .flags(SlimHeaderFlags::default().with_recv_from(conn))
1176                                        .build_unsubscribe()
1177                                        .unwrap();
1178
1179                                    let sub_id = self
1180                                        .inner
1181                                        .route_subscription_ids
1182                                        .lock()
1183                                        .remove(&(name.clone(), conn));
1184                                    let unsubscribe_result = match sub_id {
1185                                        Some(subscription_id) => {
1186                                            self.send_unsubscribe_message_with_ack(
1187                                                msg,
1188                                                subscription_id,
1189                                            )
1190                                            .await
1191                                        }
1192                                        None => {
1193                                            // No tracking ID in the map (e.g. after a CP restart
1194                                            // or for orphan cleanup from the reconciler).  Generate
1195                                            // a fresh ID — the datapath locates the subscription
1196                                            // by Name+connection, not by ID; the ID is only used
1197                                            // for ack correlation.
1198                                            let (ack_id, ack_rx) =
1199                                                self.inner.subscription_manager.register_ack();
1200                                            let mut fresh_msg = msg;
1201                                            fresh_msg.set_subscription_id(ack_id);
1202                                            if let Err(e) =
1203                                                self.send_control_message(fresh_msg).await
1204                                            {
1205                                                self.inner.subscription_manager.cancel_ack(ack_id);
1206                                                Err(format!("datapath send error: {}", e.chain()))
1207                                            } else {
1208                                                match tokio::time::timeout(
1209                                                    SUBSCRIPTION_ACK_TIMEOUT,
1210                                                    ack_rx,
1211                                                )
1212                                                .await
1213                                                {
1214                                                    Ok(Ok(Ok(()))) => Ok(()),
1215                                                    Ok(Ok(Err(err))) => Err(err.to_string()),
1216                                                    Ok(Err(_)) => {
1217                                                        Err("subscription ack channel closed"
1218                                                            .to_string())
1219                                                    }
1220                                                    Err(_) => {
1221                                                        self.inner
1222                                                            .subscription_manager
1223                                                            .cancel_ack(ack_id);
1224                                                        Err("subscription ack timed out"
1225                                                            .to_string())
1226                                                    }
1227                                                }
1228                                            }
1229                                        }
1230                                    };
1231                                    if let Err(err) = unsubscribe_result {
1232                                        route_success = false;
1233                                        route_error_msg = format!("Failed to unsubscribe: {}", err);
1234                                    } else {
1235                                        info!(?route, "Successfully deleted route");
1236                                    }
1237                                } else {
1238                                    route_success = false;
1239                                    route_error_msg = match conn {
1240                                        Ok(None) => {
1241                                            format!(
1242                                                "Connection with link_id {} not found",
1243                                                route.link_id.as_deref().unwrap_or("<none>")
1244                                            )
1245                                        }
1246                                        Err(err) => err,
1247                                        _ => "unknown connection lookup error".to_string(),
1248                                    };
1249                                }
1250
1251                                // Add route status (for deletion)
1252                                routes_status.push(v1::RouteAck {
1253                                    route: Some(route.clone()),
1254                                    success: route_success,
1255                                    error_msg: route_error_msg,
1256                                });
1257                            }
1258
1259                            (connections_status, routes_status)
1260                        };
1261
1262                        let config_ack = v1::ConfigurationCommandAck {
1263                            original_message_id: msg.message_id.clone(),
1264                            connections_status,
1265                            routes_status,
1266                        };
1267
1268                        let reply = ControlMessage {
1269                            message_id: uuid::Uuid::new_v4().to_string(),
1270                            payload: Some(Payload::ConfigCommandAck(config_ack)),
1271                        };
1272
1273                        if let Err(e) = tx.send(Ok(reply)).await {
1274                            error!(error = %e.chain(), "failed to send ConfigurationCommandAck");
1275                        }
1276
1277                        info!(
1278                            connections = %config.connections_to_create.len(),
1279                            connections_to_delete = %config.connections_to_delete.len(),
1280                            routes_to_set = %config.routes_to_set.len(),
1281                            routes_to_del = %config.routes_to_delete.len(),
1282                            "Processed ConfigurationCommand"
1283                        );
1284                    }
1285                    Payload::RouteListRequest(_) => {
1286                        const CHUNK_SIZE: usize = 100;
1287
1288                        let conn_table = self.inner.message_processor.connection_table();
1289                        let mut entries = Vec::new();
1290
1291                        self.inner.message_processor.subscription_table().for_each(
1292                            |name, id, local, remote, peer, edge| {
1293                                let mut entry = RouteEntry {
1294                                    name: Some(name.clone().with_id(id)),
1295                                    ..Default::default()
1296                                };
1297
1298                                for &cid in local {
1299                                    entry.connections.push(ConnectionEntry {
1300                                        id: cid,
1301                                        connection_type: ConnectionType::Local as i32,
1302                                        config_data: "{}".to_string(),
1303                                        link_id: None,
1304                                        direction: ConnectionDirection::Outgoing as i32,
1305                                        peer_node_id: None,
1306                                    });
1307                                }
1308
1309                                let conn_slices = [
1310                                    (remote, ConnectionType::Remote),
1311                                    (peer, ConnectionType::Peer),
1312                                    (edge, ConnectionType::Edge),
1313                                ];
1314                                for (conns, ct) in conn_slices {
1315                                    for &cid in conns {
1316                                        if let Some(conn) = conn_table.get(cid) {
1317                                            entry.connections.push(ConnectionEntry {
1318                                                id: cid,
1319                                                connection_type: ct as i32,
1320                                                config_data: conn
1321                                                    .config_data()
1322                                                    .and_then(|d| serde_json::to_string(d).ok())
1323                                                    .unwrap_or_else(|| "{}".to_string()),
1324                                                link_id: conn.link_id(),
1325                                                direction: if conn.is_outgoing() {
1326                                                    ConnectionDirection::Outgoing as i32
1327                                                } else {
1328                                                    ConnectionDirection::Incoming as i32
1329                                                },
1330                                                peer_node_id: conn
1331                                                    .peer_node_id()
1332                                                    .map(str::to_string),
1333                                            });
1334                                        } else {
1335                                            error!(%cid, ?ct, "no connection entry for id");
1336                                        }
1337                                    }
1338                                }
1339                                entries.push(entry);
1340                            },
1341                        );
1342
1343                        let chunks: Vec<_> = entries.chunks(CHUNK_SIZE).collect();
1344                        if chunks.is_empty() {
1345                            let resp = ControlMessage {
1346                                message_id: uuid::Uuid::new_v4().to_string(),
1347                                payload: Some(Payload::RouteListResponse(RouteListResponse {
1348                                    original_message_id: msg.message_id.clone(),
1349                                    entries: vec![],
1350                                    done: true,
1351                                })),
1352                            };
1353                            if let Err(e) = tx.send(Ok(resp)).await {
1354                                error!(error = %e.chain(), "failed to send route list response");
1355                            }
1356                        } else {
1357                            let n = chunks.len();
1358                            for (i, chunk) in chunks.into_iter().enumerate() {
1359                                let resp = ControlMessage {
1360                                    message_id: uuid::Uuid::new_v4().to_string(),
1361                                    payload: Some(Payload::RouteListResponse(RouteListResponse {
1362                                        original_message_id: msg.message_id.clone(),
1363                                        entries: chunk.to_vec(),
1364                                        done: i + 1 == n,
1365                                    })),
1366                                };
1367                                if let Err(e) = tx.send(Ok(resp)).await {
1368                                    error!(error = %e.chain(), "failed to send route batch");
1369                                    break;
1370                                }
1371                            }
1372                        }
1373                    }
1374                    Payload::ConnectionListRequest(_) => {
1375                        let mut all_entries = Vec::new();
1376                        self.inner
1377                            .message_processor
1378                            .connection_table()
1379                            .for_each(|id, conn| {
1380                                let ct = match conn.connection_type() {
1381                                    slim_datapath::tables::ConnType::Local => ConnectionType::Local,
1382                                    slim_datapath::tables::ConnType::Remote => {
1383                                        ConnectionType::Remote
1384                                    }
1385                                    slim_datapath::tables::ConnType::Peer => ConnectionType::Peer,
1386                                    slim_datapath::tables::ConnType::Edge => ConnectionType::Edge,
1387                                };
1388                                all_entries.push(ConnectionEntry {
1389                                    id,
1390                                    connection_type: ct as i32,
1391                                    config_data: conn
1392                                        .config_data()
1393                                        .and_then(|d| serde_json::to_string(d).ok())
1394                                        .unwrap_or_else(|| "{}".to_string()),
1395                                    link_id: conn.link_id(),
1396                                    direction: if conn.is_outgoing() {
1397                                        ConnectionDirection::Outgoing as i32
1398                                    } else {
1399                                        ConnectionDirection::Incoming as i32
1400                                    },
1401                                    peer_node_id: conn.peer_node_id().map(str::to_string),
1402                                });
1403                            });
1404
1405                        const CHUNK_SIZE: usize = 100;
1406                        let chunks: Vec<_> = all_entries.chunks(CHUNK_SIZE).collect();
1407                        if chunks.is_empty() {
1408                            let resp = ControlMessage {
1409                                message_id: uuid::Uuid::new_v4().to_string(),
1410                                payload: Some(Payload::ConnectionListResponse(
1411                                    ConnectionListResponse {
1412                                        original_message_id: msg.message_id.clone(),
1413                                        entries: vec![],
1414                                        done: true,
1415                                    },
1416                                )),
1417                            };
1418                            if let Err(e) = tx.send(Ok(resp)).await {
1419                                error!(error = %e.chain(), "failed to send connection list response");
1420                            }
1421                        } else {
1422                            let n = chunks.len();
1423                            for (i, chunk) in chunks.into_iter().enumerate() {
1424                                let resp = ControlMessage {
1425                                    message_id: uuid::Uuid::new_v4().to_string(),
1426                                    payload: Some(Payload::ConnectionListResponse(
1427                                        ConnectionListResponse {
1428                                            original_message_id: msg.message_id.clone(),
1429                                            entries: chunk.to_vec(),
1430                                            done: i + 1 == n,
1431                                        },
1432                                    )),
1433                                };
1434                                if let Err(e) = tx.send(Ok(resp)).await {
1435                                    error!(error = %e.chain(), "failed to send connection list batch");
1436                                    break;
1437                                }
1438                            }
1439                        }
1440                    }
1441                    Payload::RegisterNodeRequest(_) => {
1442                        error!("received a register node request");
1443                    }
1444                    Payload::DeregisterNodeRequest(_) => {
1445                        error!("received a deregister node request");
1446                    }
1447                    _ => {
1448                        // received unsupported message type, do nothing
1449                        debug!("received unsupported message type from control - ignoring");
1450                    }
1451                }
1452            }
1453            None => {
1454                error!(
1455                    message_id = %msg.message_id,
1456                    "received control message with no payload",
1457                );
1458            }
1459        }
1460
1461        Ok(())
1462    }
1463
1464    async fn handle_subscribe_message(&self, dst: ProtoName, clients: &[ClientConfig]) {
1465        let mut sub_vec = vec![];
1466
1467        let cmd = v1::Route {
1468            name: Some(dst),
1469            link_id: None,
1470            direction: None,
1471        };
1472
1473        debug!(
1474            "handle_subscribe_message: sending route_to_set to control plane: {:?}",
1475            cmd
1476        );
1477        sub_vec.push(cmd);
1478
1479        let ctrl = ControlMessage {
1480            message_id: uuid::Uuid::new_v4().to_string(),
1481            payload: Some(Payload::ConfigCommand(v1::ConfigurationCommand {
1482                connections_to_create: vec![],
1483                connections_to_delete: vec![],
1484                routes_to_set: sub_vec,
1485                routes_to_delete: vec![],
1486                reconcile: false,
1487                connections_received: vec![],
1488            })),
1489        };
1490
1491        return self.send_or_queue_notification(ctrl, clients).await;
1492    }
1493
1494    /// Called when the data plane completes a link negotiation for a remote
1495    /// incoming connection. Notifies the control-plane so it can claim the link.
1496    async fn handle_link_received(
1497        &self,
1498        link: &slim_datapath::api::ProtoLink,
1499        clients: &[ClientConfig],
1500    ) {
1501        use slim_datapath::api::ProtoLinkType;
1502
1503        let link_id = match &link.link_type {
1504            Some(ProtoLinkType::LinkNegotiation(payload)) => &payload.link_id,
1505            _ => {
1506                debug!("handle_link_received: ignoring link message without negotiation payload");
1507                return;
1508            }
1509        };
1510
1511        if link_id.is_empty() {
1512            debug!("handle_link_received: ignoring link message with empty link_id");
1513            return;
1514        }
1515
1516        debug!(
1517            "handle_link_received: notifying control-plane of new link_id: {}",
1518            link_id
1519        );
1520
1521        let ctrl = ControlMessage {
1522            message_id: uuid::Uuid::new_v4().to_string(),
1523            payload: Some(Payload::ConfigCommand(v1::ConfigurationCommand {
1524                connections_to_create: vec![],
1525                connections_to_delete: vec![],
1526                routes_to_set: vec![],
1527                routes_to_delete: vec![],
1528                reconcile: false,
1529                connections_received: vec![v1::ConnectionEntry {
1530                    id: 0,
1531                    connection_type: ConnectionType::Remote as i32,
1532                    config_data: String::new(),
1533                    link_id: Some(link_id.clone()),
1534                    direction: ConnectionDirection::Incoming as i32,
1535                    peer_node_id: None,
1536                }],
1537            })),
1538        };
1539
1540        self.send_or_queue_notification(ctrl, clients).await;
1541    }
1542
1543    async fn handle_unsubscribe_message(&self, dst: ProtoName, clients: &[ClientConfig]) {
1544        // Remove the subscription from our tracking map so the reconciler
1545        // knows it needs to re-install it if the route is still desired.
1546        // Without this, a reused conn_id would cause create_new_subscriptions
1547        // to skip re-creation because it thinks the subscription is still active.
1548        self.inner
1549            .route_subscription_ids
1550            .lock()
1551            .retain(|(name, _), _| *name != dst);
1552
1553        let mut unsub_vec = vec![];
1554
1555        let cmd = v1::Route {
1556            name: Some(dst),
1557            link_id: None,
1558            direction: None,
1559        };
1560
1561        debug!(
1562            "handle_unsubscribe_message: sending route_to_delete to control plane: {:?}",
1563            cmd
1564        );
1565        unsub_vec.push(cmd);
1566
1567        let ctrl = ControlMessage {
1568            message_id: uuid::Uuid::new_v4().to_string(),
1569            payload: Some(Payload::ConfigCommand(v1::ConfigurationCommand {
1570                connections_to_create: vec![],
1571                connections_to_delete: vec![],
1572                routes_to_set: vec![],
1573                routes_to_delete: unsub_vec,
1574                reconcile: false,
1575                connections_received: vec![],
1576            })),
1577        };
1578
1579        return self.send_or_queue_notification(ctrl, clients).await;
1580    }
1581
1582    /// Send a subscribe message and await the ack. Returns the subscription_id.
1583    async fn send_subscribe_message_with_ack(
1584        &self,
1585        mut msg: DataPlaneMessage,
1586    ) -> Result<u64, String> {
1587        let (ack_id, ack_rx) = self.inner.subscription_manager.register_ack();
1588        msg.set_subscription_id(ack_id);
1589
1590        if let Err(e) = self.send_control_message(msg).await {
1591            self.inner.subscription_manager.cancel_ack(ack_id);
1592            return Err(format!("datapath send error: {}", e.chain()));
1593        }
1594
1595        match tokio::time::timeout(SUBSCRIPTION_ACK_TIMEOUT, ack_rx).await {
1596            Ok(Ok(Ok(()))) => Ok(ack_id),
1597            Ok(Ok(Err(err))) => Err(err.to_string()),
1598            Ok(Err(_)) => Err("subscription ack channel closed".to_string()),
1599            Err(_) => {
1600                self.inner.subscription_manager.cancel_ack(ack_id);
1601                Err("subscription ack timed out".to_string())
1602            }
1603        }
1604    }
1605
1606    /// Send an unsubscribe message with a given subscription_id and await the ack.
1607    async fn send_unsubscribe_message_with_ack(
1608        &self,
1609        mut msg: DataPlaneMessage,
1610        subscription_id: u64,
1611    ) -> Result<(), String> {
1612        let ack_rx = self
1613            .inner
1614            .subscription_manager
1615            .register_ack_with_id(subscription_id);
1616        msg.set_subscription_id(subscription_id);
1617
1618        if let Err(e) = self.send_control_message(msg).await {
1619            self.inner.subscription_manager.cancel_ack(subscription_id);
1620            return Err(format!("datapath send error: {}", e.chain()));
1621        }
1622
1623        match tokio::time::timeout(SUBSCRIPTION_ACK_TIMEOUT, ack_rx).await {
1624            Ok(Ok(Ok(()))) => Ok(()),
1625            Ok(Ok(Err(err))) => Err(err.to_string()),
1626            Ok(Err(_)) => Err("subscription ack channel closed".to_string()),
1627            Err(_) => {
1628                self.inner.subscription_manager.cancel_ack(subscription_id);
1629                Err("subscription ack timed out".to_string())
1630            }
1631        }
1632    }
1633
1634    /// Send a control message to SLIM.
1635    async fn send_control_message(&self, msg: DataPlaneMessage) -> Result<(), ControllerError> {
1636        self.inner.tx_slim.send(Ok(msg)).await.map_err(|e| {
1637            error!(error = %e.chain(), "error sending message into datapath");
1638            ControllerError::Datapath(slim_datapath::errors::DataPathError::ConnectionError)
1639        })
1640    }
1641
1642    /// Send notification to control plane or queue it if no connection is available.
1643    ///
1644    /// Uses `try_send` to avoid blocking the caller when the bounded channel is
1645    /// full (e.g. due to HTTP/2 back-pressure).  Blocking here would stall the
1646    /// data-plane listener task, preventing it from resolving subscription acks
1647    /// that the control-message handler is waiting on — a deadlock.
1648    async fn send_or_queue_notification(&self, ctrl_msg: ControlMessage, clients: &[ClientConfig]) {
1649        let mut sent = false;
1650
1651        for c in clients {
1652            let tx = match self.inner.tx_channels.read().get(&c.endpoint) {
1653                Some(tx) => tx.clone(),
1654                None => continue,
1655            };
1656
1657            match tx.try_send(Ok(ctrl_msg.clone())) {
1658                Ok(()) => {
1659                    sent = true;
1660                }
1661                Err(mpsc::error::TrySendError::Full(_)) => {
1662                    debug!(
1663                        endpoint = %c.endpoint,
1664                        "channel full, queuing notification instead of blocking"
1665                    );
1666                }
1667                Err(mpsc::error::TrySendError::Closed(_)) => {
1668                    debug!(
1669                        endpoint = %c.endpoint,
1670                        "channel closed, queuing notification"
1671                    );
1672                }
1673            }
1674        }
1675
1676        if !sent {
1677            let mut queue = self.inner.pending_notifications.lock();
1678            if queue.len() >= MAX_QUEUED_NOTIFICATIONS {
1679                queue.pop_front();
1680                debug!("queue full, removed oldest notification");
1681            }
1682            queue.push_back(ctrl_msg);
1683        }
1684    }
1685
1686    /// Get a drain watch clone to pass to a task
1687    fn drain_watch(&self) -> Result<drain::Watch, ControllerError> {
1688        self.inner
1689            .drain_watch
1690            .read()
1691            .clone()
1692            .ok_or(ControllerError::AlreadyStopped)
1693    }
1694
1695    /// Send all queued subscription notifications when connection is restored.
1696    async fn send_queued_notifications(
1697        &self,
1698        tx: &mpsc::Sender<Result<ControlMessage, Status>>,
1699        endpoint: &str,
1700    ) {
1701        let notifications = {
1702            let mut queue = self.inner.pending_notifications.lock();
1703            if queue.is_empty() {
1704                return;
1705            }
1706            queue.drain(..).collect::<Vec<_>>()
1707        };
1708
1709        if notifications.is_empty() {
1710            return;
1711        }
1712
1713        debug!(
1714            "sending {} queued subscription notifications to {}",
1715            notifications.len(),
1716            endpoint
1717        );
1718
1719        let mut failed_notifications = Vec::new();
1720        for notification in notifications {
1721            if let Err(e) = tx.send(Ok(notification)).await {
1722                error!(
1723                    error = %e.chain(),
1724                    %endpoint,
1725                    "failed to send queued notification to control plane",
1726                );
1727
1728                // we can unwrap here because we know we sent a Ok(ControlMessage)
1729                failed_notifications.push(e.0.unwrap());
1730            }
1731        }
1732
1733        // Re-queue any failed notifications
1734        if !failed_notifications.is_empty() {
1735            self.inner
1736                .pending_notifications
1737                .lock()
1738                .extend(failed_notifications);
1739        }
1740    }
1741
1742    /// Replay all locally-registered agent subscriptions to the control plane.
1743    /// Called after (re)connecting so the CP can recreate wildcard route templates.
1744    async fn replay_local_subscriptions(&self, clients: &[ClientConfig]) {
1745        let mut routes: Vec<v1::Route> = Vec::new();
1746        self.inner.message_processor.subscription_table().for_each(
1747            |name, id, local, _remote, _peer, edge| {
1748                if local.is_empty() && edge.is_empty() {
1749                    return;
1750                }
1751                routes.push(v1::Route {
1752                    name: Some(name.clone().with_id(id)),
1753                    link_id: None,
1754                    direction: None,
1755                });
1756            },
1757        );
1758
1759        if routes.is_empty() {
1760            return;
1761        }
1762
1763        info!(
1764            count = routes.len(),
1765            "replaying local subscriptions to control plane"
1766        );
1767
1768        let ctrl = ControlMessage {
1769            message_id: uuid::Uuid::new_v4().to_string(),
1770            payload: Some(Payload::ConfigCommand(v1::ConfigurationCommand {
1771                connections_to_create: vec![],
1772                connections_to_delete: vec![],
1773                routes_to_set: routes,
1774                routes_to_delete: vec![],
1775                reconcile: false,
1776                connections_received: vec![],
1777            })),
1778        };
1779
1780        self.send_or_queue_notification(ctrl, clients).await;
1781    }
1782
1783    /// Process the control message stream.
1784    fn process_control_message_stream(
1785        &self,
1786        config: Option<ClientConfig>,
1787        mut stream: impl Stream<Item = Result<ControlMessage, Status>> + Unpin + Send + 'static,
1788        tx: mpsc::Sender<Result<ControlMessage, Status>>,
1789        cancellation_token: CancellationToken,
1790    ) -> Result<JoinHandle<()>, ControllerError> {
1791        let this = self.clone();
1792        let watch = self.drain_watch()?;
1793
1794        let handle = tokio::spawn(async move {
1795            // Send a register message to the control plane
1796            let endpoint = config
1797                .as_ref()
1798                .map(|c| c.endpoint.clone())
1799                .unwrap_or_else(|| "unknown".to_string());
1800            info!(%endpoint, "connected to control plane");
1801
1802            let mut retry_connect = false;
1803
1804            let mut active_connections = Vec::new();
1805            this.inner
1806                .message_processor
1807                .connection_table()
1808                .for_each(|id, conn| {
1809                    active_connections.push(v1::ConnectionEntry {
1810                        id,
1811                        connection_type: v1::ConnectionType::Remote as i32,
1812                        config_data: match conn.config_data() {
1813                            Some(data) => {
1814                                serde_json::to_string(data).unwrap_or_else(|_| "{}".to_string())
1815                            }
1816                            None => "{}".to_string(),
1817                        },
1818                        link_id: conn.link_id(),
1819                        direction: if conn.is_outgoing() {
1820                            v1::ConnectionDirection::Outgoing as i32
1821                        } else {
1822                            v1::ConnectionDirection::Incoming as i32
1823                        },
1824                        peer_node_id: conn.peer_node_id().map(str::to_string),
1825                    });
1826                });
1827
1828            let active_routes = {
1829                let conn_id_to_link_id: HashMap<u64, String> = this
1830                    .inner
1831                    .link_id_to_conn_id
1832                    .read()
1833                    .iter()
1834                    .map(|(lid, cid)| (*cid, lid.clone()))
1835                    .collect();
1836                this.inner
1837                    .route_subscription_ids
1838                    .lock()
1839                    .iter()
1840                    .map(|((name, conn_id), _sub_id)| v1::Route {
1841                        name: Some(name.clone()),
1842                        link_id: conn_id_to_link_id.get(conn_id).cloned(),
1843                        direction: None,
1844                    })
1845                    .collect::<Vec<_>>()
1846            };
1847
1848            let max_attempts = 10;
1849            let mut credentials = None;
1850            let mut i = 0;
1851            while i < max_attempts && credentials.is_none() {
1852                credentials = match &this.inner.auth_provider {
1853                    Some(provider) => match provider.get_token() {
1854                        Ok(token) => Some(token),
1855                        Err(e) => {
1856                            info!(error = %e, attempt = i + 1, max_attempts, "failed to get auth credentials, will retry");
1857                            tokio::time::sleep(Duration::from_secs(2)).await;
1858                            None
1859                        }
1860                    },
1861                    None => Some(String::new()),
1862                };
1863                i += 1;
1864            }
1865
1866            if credentials.is_none() {
1867                error!(
1868                    attempts = max_attempts,
1869                    "failed to obtain auth credentials, aborting registration"
1870                );
1871                return;
1872            }
1873
1874            let register_request = ControlMessage {
1875                message_id: uuid::Uuid::new_v4().to_string(),
1876                payload: Some(Payload::RegisterNodeRequest(v1::RegisterNodeRequest {
1877                    node_id: this.inner.id.clone(),
1878                    group_name: this.inner.group_name.clone(),
1879                    connection_details: this.inner.connection_details.clone(),
1880                    connections: active_connections,
1881                    routes: active_routes,
1882                    credentials: credentials.unwrap(),
1883                })),
1884            };
1885
1886            // send register request if client
1887            if config.is_some()
1888                && let Err(e) = tx.send(Ok(register_request)).await
1889            {
1890                error!(error = %e.chain(), "failed to send register request");
1891                return;
1892            }
1893
1894            // Send queued notifications after the register request so that
1895            // RegisterNodeRequest is always the first message on the stream.
1896            this.send_queued_notifications(&tx, &endpoint).await;
1897
1898            // Wait for RegisterNodeResponse from control plane before processing commands.
1899            if config.is_some() {
1900                let registration_result = tokio::time::timeout(
1901                    Duration::from_secs(10),
1902                    async {
1903                        loop {
1904                            tokio::select! {
1905                                next = stream.next() => {
1906                                    match next {
1907                                        Some(Ok(msg)) => {
1908                                            if let Some(Payload::RegisterNodeResponse(resp)) = msg.payload {
1909                                                if resp.success {
1910                                                    info!(
1911                                                        connections = resp.connections.len(),
1912                                                        routes = resp.routes.len(),
1913                                                        "registration acknowledged by control plane"
1914                                                    );
1915                                                    return Ok(resp);
1916                                                } else {
1917                                                    return Err("registration rejected by control plane".to_string());
1918                                                }
1919                                            }
1920                                        }
1921                                        Some(Err(_)) => return Err("stream error waiting for registration response".to_string()),
1922                                        None => return Err("stream closed before registration response".to_string()),
1923                                    }
1924                                }
1925                                _ = cancellation_token.cancelled() => {
1926                                    return Err("cancelled while waiting for registration response".to_string());
1927                                }
1928                            }
1929                        }
1930                    }
1931                ).await;
1932
1933                match registration_result {
1934                    Ok(Ok(resp)) => {
1935                        // Apply initial desired state from the control plane.
1936                        if !resp.connections.is_empty() || !resp.routes.is_empty() {
1937                            let init_cmd = ControlMessage {
1938                                message_id: uuid::Uuid::new_v4().to_string(),
1939                                payload: Some(Payload::ConfigCommand(v1::ConfigurationCommand {
1940                                    connections_to_create: resp.connections,
1941                                    routes_to_set: resp.routes,
1942                                    routes_to_delete: vec![],
1943                                    connections_to_delete: vec![],
1944                                    reconcile: true,
1945                                    connections_received: vec![],
1946                                })),
1947                            };
1948                            if let Err(e) = this.handle_new_control_message(init_cmd, &tx).await {
1949                                error!(error = %e.chain(), "failed to apply initial state from registration");
1950                            }
1951                        }
1952                    }
1953                    Ok(Err(e)) => {
1954                        error!(%e, "registration handshake failed");
1955                        return;
1956                    }
1957                    Err(_) => {
1958                        error!("registration handshake timed out");
1959                        return;
1960                    }
1961                }
1962            }
1963
1964            // Replay local agent subscriptions so the CP recreates route templates.
1965            if let Some(ref cfg) = config {
1966                this.replay_local_subscriptions(std::slice::from_ref(cfg))
1967                    .await;
1968            }
1969
1970            let mut drain_fut = std::pin::pin!(watch.clone().signaled());
1971
1972            loop {
1973                tokio::select! {
1974                    next = stream.next() => {
1975                        match next {
1976                            Some(Ok(msg)) => {
1977                                if let Err(e) = this.handle_new_control_message(msg, &tx).await {
1978                                    error!(error = %e.chain(), "error processing incoming control message");
1979                                }
1980                            }
1981                            Some(Err(e)) => {
1982                                if let Some(io_err) = Self::match_for_io_error(&e) {
1983                                    if io_err.kind() == std::io::ErrorKind::BrokenPipe {
1984                                        info!("connection closed by peer");
1985                                    } else {
1986                                        // Handle other IO errors (ConnectionAborted, etc.)
1987                                        error!(
1988                                            error = %e.chain(),
1989                                            io_error_kind = ?io_err.kind(),
1990                                            "IO error receiving control messages"
1991                                        );
1992                                    }
1993                                } else {
1994                                    // Handle non-IO errors (e.g., gRPC Canceled, Unavailable, etc.)
1995                                    error!(error = %e.chain(), "error receiving control messages");
1996                                }
1997
1998                                retry_connect = true;
1999                                break;
2000                            }
2001                            None => {
2002                                debug!("end of stream");
2003                                retry_connect = true;
2004                                break;
2005                            }
2006                        }
2007                    }
2008                    _ = cancellation_token.cancelled() => {
2009                        debug!("shutting down stream on cancellation token");
2010                        break;
2011                    }
2012                    _ = &mut drain_fut => {
2013                        debug!("shutting down stream on drain");
2014                        break;
2015                    }
2016                }
2017            }
2018
2019            info!(%endpoint, "control plane stream closed");
2020
2021            if retry_connect && let Some(config) = config {
2022                info!(%config.endpoint, "retrying connection to control plane");
2023                this.connect(config.clone(), cancellation_token)
2024                    .await
2025                    .map_or_else(
2026                        |e| {
2027                            error!(error = %e.chain(), "failed to reconnect to control plane");
2028                        },
2029                        |tx| {
2030                            info!(%config.endpoint, "reconnected to control plane");
2031
2032                            this.inner
2033                                .tx_channels
2034                                .write()
2035                                .insert(config.endpoint.clone(), tx);
2036                        },
2037                    )
2038            }
2039        });
2040
2041        Ok(handle)
2042    }
2043
2044    /// Connect to the control plane using the provided client configuration.
2045    /// This function attempts to establish a connection to the control plane and returns a sender for control messages.
2046    /// It retries the connection a specified number of times if it fails.
2047    async fn connect(
2048        &self,
2049        config: ClientConfig,
2050        cancellation_token: CancellationToken,
2051    ) -> Result<mpsc::Sender<Result<ControlMessage, Status>>, ControllerError> {
2052        info!(%config.endpoint, "connecting to control plane");
2053
2054        // Disconnect all outgoing remote connections — the CP will create
2055        // fresh links after re-registration and nodes will reconnect.
2056        // This prevents stale connections from causing duplicate links.
2057        let mut remote_conn_ids: Vec<u64> = Vec::new();
2058        self.inner
2059            .message_processor
2060            .connection_table()
2061            .for_each(|id, conn| {
2062                if conn.is_outgoing() && matches!(conn.connection_type(), ConnType::Remote) {
2063                    remote_conn_ids.push(id);
2064                }
2065            });
2066        for conn_id in &remote_conn_ids {
2067            if let Err(e) = self.inner.message_processor.disconnect(*conn_id) {
2068                debug!(conn_id, error = %e.chain(), "failed to disconnect remote connection");
2069            }
2070        }
2071        if !remote_conn_ids.is_empty() {
2072            info!(
2073                count = remote_conn_ids.len(),
2074                "disconnected remote connections for clean restart"
2075            );
2076        }
2077
2078        // Reset connection state before establishing a new session. The CP
2079        // will send fresh ConfigCommands after re-registration to rebuild
2080        // these maps.
2081        self.inner.route_subscription_ids.lock().clear();
2082        self.inner.link_id_to_conn_id.write().clear();
2083
2084        // Obtain drain watch to handle graceful shutdown during connection
2085        let watch = self.drain_watch()?;
2086
2087        let connect_fut = async {
2088            let channel = match config.to_channel().await? {
2089                TransportChannel::Grpc(c) => c,
2090                TransportChannel::Websocket(_) => {
2091                    return Err(ControllerError::ConfigError(
2092                        slim_config::errors::ConfigError::GrpcChannelUnsupportedTransport,
2093                    ));
2094                }
2095            };
2096
2097            let mut client = ControllerServiceClient::new(channel.clone());
2098            let (tx, rx) = mpsc::channel::<Result<ControlMessage, Status>>(128);
2099            let out_stream = ReceiverStream::new(rx).filter_map(|res| match res {
2100                Ok(msg) => Some(msg),
2101                Err(e) => {
2102                    error!(error = %e, "dropping outbound control message due to error");
2103                    None
2104                }
2105            });
2106            let stream = client
2107                .open_control_channel(Request::new(out_stream))
2108                .await?;
2109            Ok((tx, stream))
2110        };
2111
2112        let (tx, stream) = tokio::select! {
2113            result = connect_fut => { result? }
2114            _ = cancellation_token.cancelled() => {
2115                debug!("connection cancelled during setup");
2116                return Err(ControllerError::Canceled);
2117            }
2118            _ = watch.signaled() => {
2119                debug!("drain signal received during connection setup");
2120                return Err(ControllerError::Canceled);
2121            }
2122        };
2123
2124        // start processing the incoming stream
2125        let endpoint_key = config.endpoint.clone();
2126        let handle = self.process_control_message_stream(
2127            Some(config),
2128            stream.into_inner(),
2129            tx.clone(),
2130            cancellation_token.clone(),
2131        )?;
2132        self.inner
2133            .stream_handles
2134            .lock()
2135            .insert(endpoint_key, handle);
2136
2137        // return the sender
2138        Ok(tx)
2139    }
2140
2141    fn match_for_io_error(err_status: &Status) -> Option<&std::io::Error> {
2142        let mut err: &(dyn std::error::Error + 'static) = err_status;
2143
2144        loop {
2145            if let Some(io_err) = err.downcast_ref::<std::io::Error>() {
2146                return Some(io_err);
2147            }
2148
2149            // h2::Error do not expose std::io::Error with `source()`
2150            // https://github.com/hyperium/h2/pull/462
2151            if let Some(h2_err) = err.downcast_ref::<h2::Error>()
2152                && let Some(io_err) = h2_err.get_io()
2153            {
2154                return Some(io_err);
2155            }
2156
2157            err = err.source()?;
2158        }
2159    }
2160}
2161
2162#[tonic::async_trait]
2163impl GrpcControllerService for ControllerService {
2164    type OpenControlChannelStream =
2165        Pin<Box<dyn Stream<Item = Result<ControlMessage, Status>> + Send + 'static>>;
2166
2167    async fn open_control_channel(
2168        &self,
2169        request: Request<tonic::Streaming<ControlMessage>>,
2170    ) -> Result<Response<Self::OpenControlChannelStream>, Status> {
2171        // Get the remote endpoint from the request metadata
2172        let remote_endpoint = request
2173            .remote_addr()
2174            .map(|addr| addr.to_string())
2175            .unwrap_or_else(|| "unknown".to_string());
2176
2177        let stream = request.into_inner();
2178        let (tx, rx) = mpsc::channel::<Result<ControlMessage, Status>>(128);
2179
2180        let cancellation_token = CancellationToken::new();
2181
2182        // Server-side connections don't initiate operations requiring acks, so no timer channel needed
2183        let handle = self
2184            .process_control_message_stream(None, stream, tx.clone(), cancellation_token.clone())
2185            .map_err(|e| {
2186                error!(error = %e.chain(), "error processing control message stream");
2187                Status::unavailable("failed to process control message stream")
2188            })?;
2189        self.inner
2190            .stream_handles
2191            .lock()
2192            .insert(remote_endpoint.clone(), handle);
2193
2194        self.inner
2195            .tx_channels
2196            .write()
2197            .insert(remote_endpoint.clone(), tx);
2198
2199        if let Some(old_token) = self
2200            .inner
2201            .cancellation_tokens
2202            .write()
2203            .insert(remote_endpoint.clone(), cancellation_token)
2204        {
2205            old_token.cancel();
2206        }
2207
2208        let out_stream = ReceiverStream::new(rx);
2209        Ok(Response::new(
2210            Box::pin(out_stream) as Self::OpenControlChannelStream
2211        ))
2212    }
2213}
2214
2215#[cfg(test)]
2216mod tests {
2217    use super::*;
2218    use tracing_test::traced_test;
2219
2220    async fn setup_control_planes(
2221        server_endpoint: &str,
2222        server_name: &str,
2223        client_name: &str,
2224    ) -> (ControlPlane, ControlPlane, ClientConfig) {
2225        let server_config = ServerConfig::with_endpoint(server_endpoint)
2226            .with_tls_settings(slim_config::tls::server::TlsServerConfig::insecure());
2227        let client_config = ClientConfig::with_endpoint(&format!("http://{}", server_endpoint))
2228            .with_tls_setting(slim_config::tls::client::TlsClientConfig::insecure());
2229
2230        let message_processor_server = MessageProcessor::new();
2231        let message_processor_client = MessageProcessor::new();
2232
2233        let control_plane_server = ControlPlane::new(ControlPlaneSettings {
2234            id: server_name.to_string(),
2235            group_name: None,
2236            servers: vec![server_config.clone()],
2237            clients: vec![],
2238            outbound_clients: vec![],
2239            message_processor: message_processor_server,
2240            connection_details: vec![from_server_config(&server_config)],
2241            auth_provider: None,
2242        });
2243
2244        let control_plane_client = ControlPlane::new(ControlPlaneSettings {
2245            id: client_name.to_string(),
2246            group_name: None,
2247            servers: vec![],
2248            clients: vec![client_config.clone()],
2249            outbound_clients: vec![],
2250            message_processor: message_processor_client,
2251            connection_details: vec![],
2252            auth_provider: None,
2253        });
2254
2255        (control_plane_server, control_plane_client, client_config)
2256    }
2257
2258    #[tokio::test]
2259    #[traced_test]
2260    async fn test_end_to_end() {
2261        let (mut control_plane_server, mut control_plane_client, _client_cfg) =
2262            setup_control_planes(
2263                "127.0.0.1:50051",
2264                "test-server-instance",
2265                "test-client-instance",
2266            )
2267            .await;
2268
2269        control_plane_server.run().await.unwrap();
2270        tokio::time::sleep(tokio::time::Duration::from_millis(100)).await;
2271        control_plane_client.run().await.unwrap();
2272        tokio::time::sleep(tokio::time::Duration::from_millis(100)).await;
2273
2274        assert!(logs_contain("received a register node request"));
2275    }
2276
2277    #[tokio::test]
2278    #[traced_test]
2279    async fn test_subscription_notification_queue_drain() {
2280        // Reuse common setup with a different port to avoid clash with other tests.
2281        let (mut control_plane_server, mut control_plane_client, client_config) =
2282            setup_control_planes(
2283                "127.0.0.1:50061",
2284                "queue-drain-server",
2285                "queue-drain-client",
2286            )
2287            .await;
2288
2289        let controller = control_plane_client.controller.clone();
2290        assert_eq!(controller.inner.pending_notifications.lock().len(), 0);
2291
2292        const N: usize = 5;
2293        for i in 0..N {
2294            let ctrl_msg = ControlMessage {
2295                message_id: uuid::Uuid::new_v4().to_string(),
2296                payload: Some(Payload::ConfigCommand(v1::ConfigurationCommand {
2297                    connections_to_create: vec![],
2298                    connections_to_delete: vec![],
2299                    routes_to_set: vec![v1::Route {
2300                        name: Some(
2301                            ProtoName::from_strings(["queued", "sub", &format!("name-{i}")])
2302                                .with_id(i as u128),
2303                        ),
2304                        link_id: None,
2305                        direction: None,
2306                    }],
2307                    routes_to_delete: vec![],
2308                    reconcile: false,
2309                    connections_received: vec![],
2310                })),
2311            };
2312            controller
2313                .send_or_queue_notification(ctrl_msg, std::slice::from_ref(&client_config))
2314                .await;
2315        }
2316        assert_eq!(controller.inner.pending_notifications.lock().len(), N);
2317
2318        control_plane_server.run().await.expect("server run failed");
2319        tokio::time::sleep(tokio::time::Duration::from_millis(100)).await;
2320        control_plane_client.run().await.expect("client run failed");
2321        tokio::time::sleep(tokio::time::Duration::from_millis(200)).await;
2322
2323        assert_eq!(controller.inner.pending_notifications.lock().len(), 0);
2324        assert!(
2325            logs_contain(&format!("sending {} queued subscription notifications", N)),
2326            "Expected log about sending queued subscription notifications"
2327        );
2328
2329        drop(controller);
2330        drop(control_plane_server);
2331        drop(control_plane_client);
2332    }
2333
2334    #[tokio::test]
2335    #[traced_test]
2336    async fn test_delete_connection_by_link_id_success_ack() {
2337        let (mut control_plane_server, mut control_plane_client, _client_cfg) =
2338            setup_control_planes(
2339                "127.0.0.1:50081",
2340                "delete-linkid-server",
2341                "delete-linkid-client",
2342            )
2343            .await;
2344
2345        control_plane_server.run().await.unwrap();
2346        tokio::time::sleep(tokio::time::Duration::from_millis(100)).await;
2347        control_plane_client.run().await.unwrap();
2348        tokio::time::sleep(tokio::time::Duration::from_millis(100)).await;
2349
2350        let controller = control_plane_client.controller.clone();
2351        let link_id = "test-delete-link-id".to_string();
2352
2353        // Insert a pre-negotiated remote connection with a known link_id for testing.
2354        let (tx, _rx) = tokio::sync::mpsc::channel(16);
2355        let conn = slim_datapath::connection::Connection::new(
2356            slim_datapath::tables::ConnType::Remote,
2357            slim_datapath::connection::Channel::Server(tx),
2358        )
2359        .with_negotiation(&link_id, "1.0.0");
2360        controller
2361            .inner
2362            .message_processor
2363            .forwarder()
2364            .on_connection_established(conn, None)
2365            .unwrap();
2366
2367        let ctrl_msg = ControlMessage {
2368            message_id: uuid::Uuid::new_v4().to_string(),
2369            payload: Some(Payload::ConfigCommand(v1::ConfigurationCommand {
2370                connections_to_create: vec![],
2371                connections_to_delete: vec![link_id.clone()],
2372                routes_to_set: vec![],
2373                routes_to_delete: vec![],
2374                reconcile: false,
2375                connections_received: vec![],
2376            })),
2377        };
2378        let (tx, mut rx) = mpsc::channel(1);
2379        controller
2380            .handle_new_control_message(ctrl_msg, &tx)
2381            .await
2382            .expect("config command must be handled");
2383
2384        let ack_msg = rx
2385            .recv()
2386            .await
2387            .expect("expected ack message")
2388            .expect("ack should be ok");
2389        let ack = match ack_msg.payload {
2390            Some(Payload::ConfigCommandAck(ack)) => ack,
2391            _ => panic!("expected ConfigCommandAck payload"),
2392        };
2393        assert_eq!(ack.connections_status.len(), 1);
2394        assert_eq!(ack.connections_status[0].link_id, link_id);
2395        assert!(ack.connections_status[0].success);
2396    }
2397
2398    #[tokio::test]
2399    #[traced_test]
2400    async fn test_delete_connection_by_link_id_unknown_fails_ack() {
2401        let (control_plane_server, control_plane_client, _client_cfg) = setup_control_planes(
2402            "127.0.0.1:50082",
2403            "delete-linkid-server-unknown",
2404            "delete-linkid-client-unknown",
2405        )
2406        .await;
2407
2408        let controller = control_plane_client.controller.clone();
2409        let ctrl_msg = ControlMessage {
2410            message_id: uuid::Uuid::new_v4().to_string(),
2411            payload: Some(Payload::ConfigCommand(v1::ConfigurationCommand {
2412                connections_to_create: vec![],
2413                connections_to_delete: vec!["unknown-link-id".to_string()],
2414                routes_to_set: vec![],
2415                routes_to_delete: vec![],
2416                reconcile: false,
2417                connections_received: vec![],
2418            })),
2419        };
2420        let (tx, mut rx) = mpsc::channel(1);
2421        controller
2422            .handle_new_control_message(ctrl_msg, &tx)
2423            .await
2424            .expect("config command must be handled");
2425
2426        let ack_msg = rx
2427            .recv()
2428            .await
2429            .expect("expected ack message")
2430            .expect("ack should be ok");
2431        let ack = match ack_msg.payload {
2432            Some(Payload::ConfigCommandAck(ack)) => ack,
2433            _ => panic!("expected ConfigCommandAck payload"),
2434        };
2435        assert_eq!(ack.connections_status.len(), 1);
2436        assert_eq!(ack.connections_status[0].link_id, "unknown-link-id");
2437        assert!(!ack.connections_status[0].success);
2438        assert!(ack.connections_status[0].error_msg.contains("not found"));
2439
2440        drop(control_plane_server);
2441    }
2442
2443    #[tokio::test]
2444    #[traced_test]
2445    async fn test_create_connection_with_existing_link_id_reuses_connection_ack() {
2446        let (mut control_plane_server, mut control_plane_client, _client_cfg) =
2447            setup_control_planes(
2448                "127.0.0.1:50083",
2449                "create-linkid-server",
2450                "create-linkid-client",
2451            )
2452            .await;
2453
2454        control_plane_server.run().await.unwrap();
2455        tokio::time::sleep(tokio::time::Duration::from_millis(100)).await;
2456        control_plane_client.run().await.unwrap();
2457        tokio::time::sleep(tokio::time::Duration::from_millis(100)).await;
2458
2459        let controller = control_plane_client.controller.clone();
2460        let link_id = "test-create-link-id".to_string();
2461
2462        // Insert a pre-negotiated remote connection with a known link_id for testing.
2463        let (tx, _rx) = tokio::sync::mpsc::channel(16);
2464        let conn = slim_datapath::connection::Connection::new(
2465            slim_datapath::tables::ConnType::Remote,
2466            slim_datapath::connection::Channel::Server(tx),
2467        )
2468        .with_negotiation(&link_id, "1.0.0");
2469        controller
2470            .inner
2471            .message_processor
2472            .forwarder()
2473            .on_connection_established(conn, None)
2474            .unwrap();
2475
2476        let endpoint = "http://127.0.0.1:59999";
2477        let connection_config = serde_json::json!({
2478            "endpoint": endpoint,
2479            "link_id": link_id
2480        })
2481        .to_string();
2482
2483        let ctrl_msg = ControlMessage {
2484            message_id: uuid::Uuid::new_v4().to_string(),
2485            payload: Some(Payload::ConfigCommand(v1::ConfigurationCommand {
2486                connections_to_create: vec![v1::Connection {
2487                    link_id: "reuse-existing-link".to_string(),
2488                    config_data: connection_config,
2489                }],
2490                connections_to_delete: vec![],
2491                routes_to_set: vec![],
2492                routes_to_delete: vec![],
2493                reconcile: false,
2494                connections_received: vec![],
2495            })),
2496        };
2497
2498        let (tx, mut rx) = mpsc::channel(1);
2499        controller
2500            .handle_new_control_message(ctrl_msg, &tx)
2501            .await
2502            .expect("config command must be handled");
2503
2504        let ack_msg = rx
2505            .recv()
2506            .await
2507            .expect("expected ack message")
2508            .expect("ack should be ok");
2509        let ack = match ack_msg.payload {
2510            Some(Payload::ConfigCommandAck(ack)) => ack,
2511            _ => panic!("expected ConfigCommandAck payload"),
2512        };
2513        assert_eq!(ack.connections_status.len(), 1);
2514        assert_eq!(ack.connections_status[0].link_id, "reuse-existing-link");
2515        assert!(ack.connections_status[0].success);
2516
2517        assert!(
2518            controller
2519                .inner
2520                .link_id_to_conn_id
2521                .read()
2522                .contains_key(&link_id),
2523            "expected link_id to be mapped to reused connection id"
2524        );
2525    }
2526
2527    #[tokio::test]
2528    #[traced_test]
2529    async fn test_subscription_set_unknown_link_id_fails_ack() {
2530        let (control_plane_server, control_plane_client, _client_cfg) = setup_control_planes(
2531            "127.0.0.1:50084",
2532            "sub-linkid-server-unknown",
2533            "sub-linkid-client-unknown",
2534        )
2535        .await;
2536
2537        let controller = control_plane_client.controller.clone();
2538        let ctrl_msg = ControlMessage {
2539            message_id: uuid::Uuid::new_v4().to_string(),
2540            payload: Some(Payload::ConfigCommand(v1::ConfigurationCommand {
2541                connections_to_create: vec![],
2542                connections_to_delete: vec![],
2543                routes_to_set: vec![v1::Route {
2544                    name: Some(ProtoName::from_strings(["org", "ns", "agent"]).with_id(1u128)),
2545                    link_id: Some("missing-link-id".to_string()),
2546                    direction: None,
2547                }],
2548                routes_to_delete: vec![],
2549                reconcile: false,
2550                connections_received: vec![],
2551            })),
2552        };
2553        let (tx, mut rx) = mpsc::channel(1);
2554        controller
2555            .handle_new_control_message(ctrl_msg, &tx)
2556            .await
2557            .expect("config command must be handled");
2558
2559        let ack_msg = rx
2560            .recv()
2561            .await
2562            .expect("expected ack message")
2563            .expect("ack should be ok");
2564        let ack = match ack_msg.payload {
2565            Some(Payload::ConfigCommandAck(ack)) => ack,
2566            _ => panic!("expected ConfigCommandAck payload"),
2567        };
2568
2569        assert_eq!(ack.routes_status.len(), 1);
2570        assert!(!ack.routes_status[0].success);
2571        assert!(
2572            ack.routes_status[0]
2573                .error_msg
2574                .contains("Connection with link_id missing-link-id not found")
2575        );
2576
2577        drop(control_plane_server);
2578    }
2579
2580    #[tokio::test]
2581    #[traced_test]
2582    async fn test_create_connection_invalid_config_fails_ack() {
2583        let (_control_plane_server, control_plane_client, _client_cfg) = setup_control_planes(
2584            "127.0.0.1:50085",
2585            "create-invalid-config-server",
2586            "create-invalid-config-client",
2587        )
2588        .await;
2589
2590        let controller = control_plane_client.controller.clone();
2591        let ctrl_msg = ControlMessage {
2592            message_id: uuid::Uuid::new_v4().to_string(),
2593            payload: Some(Payload::ConfigCommand(v1::ConfigurationCommand {
2594                connections_to_create: vec![v1::Connection {
2595                    link_id: "invalid-config-conn".to_string(),
2596                    config_data: "{invalid-json".to_string(),
2597                }],
2598                connections_to_delete: vec![],
2599                routes_to_set: vec![],
2600                routes_to_delete: vec![],
2601                reconcile: false,
2602                connections_received: vec![],
2603            })),
2604        };
2605        let (tx, mut rx) = mpsc::channel(1);
2606        controller
2607            .handle_new_control_message(ctrl_msg, &tx)
2608            .await
2609            .expect("config command must be handled");
2610
2611        let ack_msg = rx
2612            .recv()
2613            .await
2614            .expect("expected ack message")
2615            .expect("ack should be ok");
2616        let ack = match ack_msg.payload {
2617            Some(Payload::ConfigCommandAck(ack)) => ack,
2618            _ => panic!("expected ConfigCommandAck payload"),
2619        };
2620        assert_eq!(ack.connections_status.len(), 1);
2621        assert!(!ack.connections_status[0].success);
2622        assert!(
2623            ack.connections_status[0]
2624                .error_msg
2625                .contains("Failed to parse config")
2626        );
2627    }
2628
2629    #[tokio::test]
2630    #[traced_test]
2631    async fn test_subscription_delete_unknown_link_id_fails_ack() {
2632        let (control_plane_server, control_plane_client, _client_cfg) = setup_control_planes(
2633            "127.0.0.1:50086",
2634            "sub-del-linkid-server-unknown",
2635            "sub-del-linkid-client-unknown",
2636        )
2637        .await;
2638
2639        let controller = control_plane_client.controller.clone();
2640        let ctrl_msg = ControlMessage {
2641            message_id: uuid::Uuid::new_v4().to_string(),
2642            payload: Some(Payload::ConfigCommand(v1::ConfigurationCommand {
2643                connections_to_create: vec![],
2644                connections_to_delete: vec![],
2645                routes_to_set: vec![],
2646                routes_to_delete: vec![v1::Route {
2647                    name: Some(ProtoName::from_strings(["org", "ns", "agent"]).with_id(1u128)),
2648                    link_id: Some("missing-link-id-delete".to_string()),
2649                    direction: None,
2650                }],
2651                reconcile: false,
2652                connections_received: vec![],
2653            })),
2654        };
2655        let (tx, mut rx) = mpsc::channel(1);
2656        controller
2657            .handle_new_control_message(ctrl_msg, &tx)
2658            .await
2659            .expect("config command must be handled");
2660
2661        let ack_msg = rx
2662            .recv()
2663            .await
2664            .expect("expected ack message")
2665            .expect("ack should be ok");
2666        let ack = match ack_msg.payload {
2667            Some(Payload::ConfigCommandAck(ack)) => ack,
2668            _ => panic!("expected ConfigCommandAck payload"),
2669        };
2670
2671        assert_eq!(ack.routes_status.len(), 1);
2672        assert!(!ack.routes_status[0].success);
2673        assert!(
2674            ack.routes_status[0]
2675                .error_msg
2676                .contains("Connection with link_id missing-link-id-delete not found")
2677        );
2678
2679        drop(control_plane_server);
2680    }
2681
2682    #[tokio::test]
2683    #[traced_test]
2684    async fn test_shutdown_drains_resources() {
2685        // Use a unique port to avoid conflicts with other tests.
2686        let (mut control_plane_server, mut control_plane_client, _client_cfg) =
2687            setup_control_planes(
2688                "127.0.0.1:50071",
2689                "shutdown-server-instance",
2690                "shutdown-client-instance",
2691            )
2692            .await;
2693
2694        // Run both ends to populate cancellation tokens.
2695        control_plane_server
2696            .run()
2697            .await
2698            .expect("server should start");
2699        tokio::time::sleep(tokio::time::Duration::from_millis(100)).await;
2700        control_plane_client
2701            .run()
2702            .await
2703            .expect("client should start");
2704        tokio::time::sleep(tokio::time::Duration::from_millis(100)).await;
2705
2706        // Ensure we have at least one cancellation token (server or client side tasks).
2707        let server_tokens_before = control_plane_server
2708            .controller
2709            .inner
2710            .cancellation_tokens
2711            .read()
2712            .len();
2713        assert!(
2714            server_tokens_before > 0,
2715            "expected server to have active cancellation tokens before shutdown"
2716        );
2717
2718        let client_tokens_before = control_plane_client
2719            .controller
2720            .inner
2721            .cancellation_tokens
2722            .read()
2723            .len();
2724        assert!(
2725            client_tokens_before > 0,
2726            "expected client to have active cancellation tokens before shutdown"
2727        );
2728
2729        // Perform shutdown on both.
2730        control_plane_client
2731            .shutdown()
2732            .await
2733            .expect("client shutdown ok");
2734        control_plane_server
2735            .shutdown()
2736            .await
2737            .expect("server shutdown ok");
2738
2739        // After shutdown, all cancellation tokens should be drained.
2740        let server_tokens_after = control_plane_server
2741            .controller
2742            .inner
2743            .cancellation_tokens
2744            .read()
2745            .len();
2746        assert_eq!(
2747            server_tokens_after, 0,
2748            "expected server cancellation tokens to be drained after shutdown"
2749        );
2750
2751        let client_tokens_after = control_plane_client
2752            .controller
2753            .inner
2754            .cancellation_tokens
2755            .read()
2756            .len();
2757        assert_eq!(
2758            client_tokens_after, 0,
2759            "expected client cancellation tokens to be drained after shutdown"
2760        );
2761
2762        // Second shutdown should error because drain_signal has been taken.
2763        assert!(
2764            control_plane_server.shutdown().await.is_err(),
2765            "second shutdown on server should return an error"
2766        );
2767        assert!(
2768            control_plane_client.shutdown().await.is_err(),
2769            "second shutdown on client should return an error"
2770        );
2771    }
2772
2773    #[tokio::test]
2774    #[traced_test]
2775    async fn test_shutdown_without_run() {
2776        // Build a control plane but do NOT call run()
2777        let (control_plane_server, mut _control_plane_client, _client_cfg) = setup_control_planes(
2778            "127.0.0.1:50072",
2779            "shutdown-no-run-server",
2780            "shutdown-no-run-client",
2781        )
2782        .await;
2783
2784        // No tasks should be registered yet
2785        assert_eq!(
2786            control_plane_server
2787                .controller
2788                .inner
2789                .cancellation_tokens
2790                .read()
2791                .len(),
2792            0,
2793            "expected zero cancellation tokens before shutdown when not run"
2794        );
2795
2796        // Shutdown should still succeed gracefully.
2797        control_plane_server
2798            .shutdown()
2799            .await
2800            .expect("shutdown without prior run should succeed");
2801
2802        // Tokens remain zero.
2803        assert_eq!(
2804            control_plane_server
2805                .controller
2806                .inner
2807                .cancellation_tokens
2808                .read()
2809                .len(),
2810            0,
2811            "expected zero cancellation tokens after shutdown when not run"
2812        );
2813
2814        // Second shutdown should fail.
2815        assert!(
2816            control_plane_server.shutdown().await.is_err(),
2817            "second shutdown should error due to missing drain signal"
2818        );
2819    }
2820
2821    fn make_reconcile_msg(link_id: &str, server_config: &ServerConnectionConfig) -> ControlMessage {
2822        ControlMessage {
2823            message_id: uuid::Uuid::new_v4().to_string(),
2824            payload: Some(Payload::ConfigCommand(v1::ConfigurationCommand {
2825                connections_to_create: vec![v1::Connection {
2826                    link_id: link_id.to_string(),
2827                    config_data: serde_json::to_string(server_config).unwrap(),
2828                }],
2829                connections_to_delete: vec![],
2830                routes_to_set: vec![],
2831                routes_to_delete: vec![],
2832                reconcile: true,
2833                connections_received: vec![],
2834            })),
2835        }
2836    }
2837
2838    fn make_controller(outbound_clients: Vec<ClientConfig>) -> ControllerService {
2839        ControlPlane::new(ControlPlaneSettings {
2840            id: "test-node".to_string(),
2841            group_name: None,
2842            servers: vec![],
2843            clients: vec![],
2844            outbound_clients,
2845            message_processor: MessageProcessor::new(),
2846            connection_details: vec![],
2847            auth_provider: None,
2848        })
2849        .controller
2850        .clone()
2851    }
2852
2853    #[tokio::test]
2854    async fn test_reconcile_basic_auth_missing_credentials_fails_ack() {
2855        let controller = make_controller(vec![]);
2856        let server_config = ServerConnectionConfig {
2857            endpoint: "http://target:8080".to_string(),
2858            tls_required: false,
2859            auth_method: RequiredAuthMethod::Basic,
2860            ..Default::default()
2861        };
2862        let (tx, mut rx) = mpsc::channel(1);
2863        controller
2864            .handle_new_control_message(make_reconcile_msg("link-1", &server_config), &tx)
2865            .await
2866            .unwrap();
2867        let ack = match rx.recv().await.unwrap().unwrap().payload {
2868            Some(Payload::ConfigCommandAck(a)) => a,
2869            _ => panic!("expected ConfigCommandAck"),
2870        };
2871        assert!(!ack.connections_status[0].success);
2872        assert!(ack.connections_status[0].error_msg.contains("target:8080"));
2873    }
2874
2875    #[tokio::test]
2876    async fn test_reconcile_jwt_auth_missing_credentials_fails_ack() {
2877        let controller = make_controller(vec![]);
2878        let server_config = ServerConnectionConfig {
2879            endpoint: "http://target:9090".to_string(),
2880            tls_required: false,
2881            auth_method: RequiredAuthMethod::Jwt,
2882            ..Default::default()
2883        };
2884        let (tx, mut rx) = mpsc::channel(1);
2885        controller
2886            .handle_new_control_message(make_reconcile_msg("link-2", &server_config), &tx)
2887            .await
2888            .unwrap();
2889        let ack = match rx.recv().await.unwrap().unwrap().payload {
2890            Some(Payload::ConfigCommandAck(a)) => a,
2891            _ => panic!("expected ConfigCommandAck"),
2892        };
2893        assert!(!ack.connections_status[0].success);
2894        assert!(ack.connections_status[0].error_msg.contains("target:9090"));
2895    }
2896}