fixed discov

control-core/src/ethercat/interface_discovery.rs

@@ -92,9 +92,6 @@ pub async fn discover_ethercat_interface() -> Result<String, anyhow::Error> {
     let mut interface: Option<&str> = None;
 
     for i in 0..interfaces.len() {
-        // Avoid NetworkManager hangups
-        set_interface_managed(&interfaces[i].name, false);
-
         match test_interface(&interfaces[i].name) {
             Ok(_) => {
                 // if interface found with ethercat Exit early, we expect only one interface with ethercat
@@ -107,6 +104,7 @@ pub async fn discover_ethercat_interface() -> Result<String, anyhow::Error> {
 
     for i in 0..interfaces.len() {
         if interface.is_some() && interface.unwrap() == &interfaces[i].name {
+            set_interface_managed(&interfaces[i].name, false);
             continue;
         } else {
             set_interface_managed(&interfaces[i].name, true);

control-core/src/socketio/namespace.rs

@@ -76,7 +76,7 @@ impl Namespace {
 
         // Sort by event count (ascending - lowest count first)
         event_groups.sort_by(|a, b| a.1.len().cmp(&b.1.len()));
-
+        tracing::info!("reemit: {:?}", self.events);
         // Emit events in order of lowest count first
         for (_event_name, events) in event_groups {
             for event in events {

server/src/machines/aquapath1/api.rs

@@ -1,4 +1,4 @@
-use std::{sync::Arc, time::Duration};
+use std::sync::Arc;
 
 use super::{AquaPathV1, AquaPathV1Mode};
 use control_core::{
@@ -6,8 +6,7 @@ use control_core::{
     socketio::{
         event::{Event, GenericEvent},
         namespace::{
-            CacheFn, CacheableEvents, Namespace, NamespaceCacheingLogic, cache_duration,
-            cache_first_and_last_event,
+            CacheFn, CacheableEvents, Namespace, NamespaceCacheingLogic, cache_first_and_last_event,
         },
     },
 };
@@ -116,11 +115,10 @@ impl CacheableEvents<AquaPathV1Events> for AquaPathV1Events {
     }
 
     fn event_cache_fn(&self) -> CacheFn {
-        let cache_one_hour = cache_duration(Duration::from_secs(60 * 60), Duration::from_secs(1));
         let cache_first_and_last = cache_first_and_last_event();
 
         match self {
-            AquaPathV1Events::LiveValues(_) => cache_one_hour,
+            AquaPathV1Events::LiveValues(_) => cache_first_and_last,
             AquaPathV1Events::State(_) => cache_first_and_last,
         }
     }

server/src/machines/buffer1/api.rs

@@ -1,4 +1,4 @@
-use std::{sync::Arc, time::Duration};
+use std::sync::Arc;
 
 use super::{BufferV1, BufferV1Mode};
 use control_core::{
@@ -8,10 +8,7 @@ use control_core::{
     },
     socketio::{
         event::{Event, GenericEvent},
-        namespace::{
-            CacheFn, CacheableEvents, Namespace, NamespaceCacheingLogic, cache_duration,
-            cache_one_event,
-        },
+        namespace::{CacheFn, CacheableEvents, Namespace, NamespaceCacheingLogic, cache_one_event},
     },
 };
 use serde::{Deserialize, Serialize};
@@ -103,11 +100,10 @@ impl CacheableEvents<Self> for BufferV1Events {
     }
 
     fn event_cache_fn(&self) -> CacheFn {
-        let cache_one_hour = cache_duration(Duration::from_secs(60 * 60), Duration::from_secs(1));
         let cache_one = cache_one_event();
 
         match self {
-            Self::LiveValues(_) => cache_one_hour,
+            Self::LiveValues(_) => cache_one,
             Self::State(_) => cache_one,
         }
     }

server/src/machines/extruder1/api.rs

@@ -11,8 +11,8 @@ use control_core::machines::api::MachineApi;
 use control_core::socketio::{
     event::{Event, GenericEvent},
     namespace::{
-        CacheFn, CacheableEvents, Namespace, NamespaceCacheingLogic, cache_duration,
-        cache_first_and_last_event,
+        CacheFn, CacheableEvents, Namespace, NamespaceCacheingLogic, cache_first_and_last_event,
+        cache_one_event,
     },
 };
 use control_core_derive::BuildEvent;
@@ -20,7 +20,7 @@ use serde::{Deserialize, Serialize};
 #[cfg(not(feature = "mock-machine"))]
 use serde_json::Value;
 use smol::lock::Mutex;
-use std::{sync::Arc, time::Duration};
+use std::sync::Arc;
 use tracing::instrument;
 use uom::si::{
     angular_velocity::revolution_per_minute, electric_current::ampere, electric_potential::volt,
@@ -263,11 +263,9 @@ impl CacheableEvents<Self> for ExtruderV2Events {
     }
 
     fn event_cache_fn(&self) -> CacheFn {
-        let cache_one_hour = cache_duration(Duration::from_secs(60 * 60), Duration::from_secs(1));
         let cache_first_and_last = cache_first_and_last_event();
-
         match self {
-            Self::LiveValues(_) => cache_one_hour,
+            Self::LiveValues(_) => cache_one_event(),
             Self::State(_) => cache_first_and_last,
         }
     }

server/src/machines/laser/api.rs

@@ -4,8 +4,8 @@ use control_core::{
     socketio::{
         event::{Event, GenericEvent},
         namespace::{
-            CacheFn, CacheableEvents, Namespace, NamespaceCacheingLogic, cache_duration,
-            cache_first_and_last_event,
+            CacheFn, CacheableEvents, Namespace, NamespaceCacheingLogic,
+            cache_first_and_last_event, cache_one_event,
         },
     },
 };
@@ -13,7 +13,7 @@ use control_core_derive::BuildEvent;
 use serde::{Deserialize, Serialize};
 use serde_json::Value;
 use smol::lock::Mutex;
-use std::{sync::Arc, time::Duration};
+use std::sync::Arc;
 use tracing::instrument;
 
 #[derive(Serialize, Debug, Clone, Default)]
@@ -73,11 +73,9 @@ impl CacheableEvents<Self> for LaserEvents {
     }
 
     fn event_cache_fn(&self) -> CacheFn {
-        let cache_one_hour = cache_duration(Duration::from_secs(60 * 60), Duration::from_secs(1));
         let cache_first_and_last = cache_first_and_last_event();
-
         match self {
-            Self::LiveValues(_) => cache_one_hour,
+            Self::LiveValues(_) => cache_one_event(),
             Self::State(_) => cache_first_and_last,
         }
     }

server/src/machines/mock/api.rs

@@ -1,11 +1,11 @@
 use super::MockMachine;
+use control_core::socketio::namespace::cache_one_event;
 use control_core::{
     machines::api::MachineApi,
     socketio::{
         event::{Event, GenericEvent},
         namespace::{
-            CacheFn, CacheableEvents, Namespace, NamespaceCacheingLogic, cache_duration,
-            cache_first_and_last_event,
+            CacheFn, CacheableEvents, Namespace, NamespaceCacheingLogic, cache_first_and_last_event,
         },
     },
 };
@@ -13,9 +13,8 @@ use control_core_derive::BuildEvent;
 use serde::{Deserialize, Serialize};
 use serde_json::Value;
 use smol::lock::Mutex;
-use std::{sync::Arc, time::Duration};
+use std::sync::Arc;
 use tracing::instrument;
-
 #[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Eq)]
 pub enum Mode {
     Standby,
@@ -72,11 +71,9 @@ impl CacheableEvents<Self> for MockEvents {
     }
 
     fn event_cache_fn(&self) -> CacheFn {
-        let cache_one_hour = cache_duration(Duration::from_secs(60 * 60), Duration::from_secs(1));
         let cache_first_and_last = cache_first_and_last_event();
-
         match self {
-            Self::LiveValues(_) => cache_one_hour,
+            Self::LiveValues(_) => cache_one_event(),
             Self::State(_) => cache_first_and_last,
         }
     }

server/src/machines/winder2/api.rs

@@ -53,6 +53,7 @@ mod winder2_imports {
     pub use tracing::instrument;
 }
 
+use control_core::socketio::namespace::cache_one_event;
 pub use winder2_imports::*;
 
 #[derive(Serialize, Deserialize, Debug, Clone, Default)]
@@ -309,11 +310,9 @@ impl CacheableEvents<Self> for Winder2Events {
     }
 
     fn event_cache_fn(&self) -> CacheFn {
-        let cache_one_hour = cache_duration(Duration::from_secs(60 * 60), Duration::from_secs(1));
         let cache_first_and_last = cache_first_and_last_event();
-
         match self {
-            Self::LiveValues(_) => cache_one_hour,
+            Self::LiveValues(_) => cache_one_event(),
             Self::State(_) => cache_first_and_last,
         }
     }

server/src/socketio/init.rs

@@ -7,7 +7,6 @@ use socketioxide::ParserConfig;
 use socketioxide::extract::SocketRef;
 use socketioxide::layer::SocketIoLayer;
 use tracing::info_span;
-use tracing_futures::Instrument;
 
 pub async fn init_socketio(app_state: &Arc<AppState>) -> SocketIoLayer {
     // create
@@ -20,8 +19,8 @@ pub async fn init_socketio(app_state: &Arc<AppState>) -> SocketIoLayer {
     let app_state_main = app_state.clone();
 
     // set the on connect handler for main namespace
-    io.ns("/main", move |socket: SocketRef| {
-        handle_socket_connection(socket, app_state_main.clone());
+    io.ns("/main", async move |socket: SocketRef| {
+        handle_socket_connection(socket, app_state_main.clone()).await;
     });
 
     // Clone app_state for the second handler
@@ -29,8 +28,8 @@ pub async fn init_socketio(app_state: &Arc<AppState>) -> SocketIoLayer {
 
     if let Err(err) = io.dyn_ns(
         "/machine/{vendor}/{machine}/{serial}",
-        move |socket: SocketRef| {
-            handle_socket_connection(socket, app_state_machine.clone());
+        async move |socket: SocketRef| {
+            handle_socket_connection(socket, app_state_machine.clone()).await;
         },
     ) {
         tracing::error!("Failed to detect machine namespace: {}", err);
@@ -43,7 +42,7 @@ pub async fn init_socketio(app_state: &Arc<AppState>) -> SocketIoLayer {
     socketio_layer
 }
 
-fn handle_socket_connection(socket: SocketRef, app_state: Arc<AppState>) {
+async fn handle_socket_connection(socket: SocketRef, app_state: Arc<AppState>) {
     let namespace_id = match NamespaceId::from_str(socket.ns()) {
         Ok(namespace_id) => namespace_id,
         Err(err) => {
@@ -51,53 +50,49 @@ fn handle_socket_connection(socket: SocketRef, app_state: Arc<AppState>) {
             return;
         }
     };
-
-    // Setup disconnection handler
-    setup_disconnection(socket.clone(), namespace_id.clone(), app_state.clone());
-
-    // Setup connection
-    setup_connection(socket, namespace_id, app_state);
+    setup_disconnection(socket.clone(), namespace_id.clone(), app_state.clone()).await;
+    setup_connection(socket, namespace_id, app_state).await;
 }
 
-fn setup_disconnection(socket: SocketRef, namespace_id: NamespaceId, app_state: Arc<AppState>) {
-    socket.on_disconnect(move |socket: SocketRef| {
+async fn setup_disconnection(
+    socket: SocketRef,
+    namespace_id: NamespaceId,
+    app_state: Arc<AppState>,
+) {
+    socket.on_disconnect(async move |socket: SocketRef| {
         let namespace_id = namespace_id.clone();
         let app_state = app_state.clone();
 
-        // Spawn async task to avoid blocking and potential deadlocks
-        smol::spawn(async move {
-            tracing::debug!(
-                "Socket disconnected from namespace socket={:?} namespace={}",
-                socket.id,
-                namespace_id,
-            );
-            let mut socketio_namespaces_guard = app_state.socketio_setup.namespaces.write().await;
+        tracing::info!(
+            "Socket disconnected from namespace socket={:?} namespace={}",
+            socket.id,
+            namespace_id,
+        );
+        let mut socketio_namespaces_guard = app_state.socketio_setup.namespaces.write().await;
 
-            // remove from machine namespace
-            socketio_namespaces_guard
-                .apply_mut(namespace_id.clone(), &app_state, |namespace_interface| {
-                    if let Ok(namespace_interface) = namespace_interface {
-                        namespace_interface.unsubscribe(socket.clone());
-                        tracing::debug!(
-                            "Socket unsubscribed from namespace socket={:?} namespace={}",
-                            socket.id,
-                            namespace_id
-                        );
-                    } else {
-                        tracing::debug!(
-                            "Failed to unsubscribe socket from namespace socket={:?} namespace={}",
-                            socket.id,
-                            namespace_id
-                        );
-                    }
-                })
-                .await;
-        })
-        .detach();
+        // remove from machine namespace
+        socketio_namespaces_guard
+            .apply_mut(namespace_id.clone(), &app_state, |namespace_interface| {
+                if let Ok(namespace_interface) = namespace_interface {
+                    namespace_interface.unsubscribe(socket.clone());
+                    tracing::debug!(
+                        "Socket unsubscribed from namespace socket={:?} namespace={}",
+                        socket.id,
+                        namespace_id
+                    );
+                } else {
+                    tracing::debug!(
+                        "Failed to unsubscribe socket from namespace socket={:?} namespace={}",
+                        socket.id,
+                        namespace_id
+                    );
+                }
+            })
+            .await;
     });
 }
 
-fn setup_connection(socket: SocketRef, namespace_id: NamespaceId, app_state: Arc<AppState>) {
+async fn setup_connection(socket: SocketRef, namespace_id: NamespaceId, app_state: Arc<AppState>) {
     // Spawn async task to avoid blocking and potential deadlocks
     let socket_clone = socket.clone();
     let namespace_id_clone = namespace_id.clone();
@@ -109,12 +104,9 @@ fn setup_connection(socket: SocketRef, namespace_id: NamespaceId, app_state: Arc
         namespace = %namespace_id_clone,
         "Connecting socket to namespace"
     );
-
-    smol::block_on(
-        async move {
-            let mut socketio_namespaces_guard =
-                app_state_clone.socketio_setup.namespaces.write().await;
-            socketio_namespaces_guard
+    let _ = span.enter();
+    let mut socketio_namespaces_guard = app_state_clone.socketio_setup.namespaces.write().await;
+    socketio_namespaces_guard
                 .apply_mut(
                     namespace_id_clone.clone(),
                     &app_state_clone,
@@ -140,9 +132,6 @@ fn setup_connection(socket: SocketRef, namespace_id: NamespaceId, app_state: Arc
                     },
                 )
                 .await;
-        }
-        .instrument(span),
-    );
 
     tracing::info!(
         "Socket connected to namespace socket={:?} namespace={}",

server/src/socketio/queue.rs

@@ -5,12 +5,18 @@ use tracing::{debug, info, instrument, trace};
 
 /// Send a single event with retry logic
 #[instrument(skip_all)]
-async fn send_event_with_retry(
+fn send_event_with_retry(
     socket: &socketioxide::extract::SocketRef,
     event: &Arc<control_core::socketio::event::GenericEvent>,
 ) {
+    let mut i = 0;
+    const MAX_RETRY: i32 = 10;
     // retry loop for each event
     loop {
+        if i >= MAX_RETRY {
+            return;
+        }
+        i += 1;
         // check if socket is still connected
         if !socket.connected() {
             trace!(
@@ -66,9 +72,7 @@ async fn socketio_queue_worker(app_state: Arc<AppState>) {
 
     while let Ok((socket, event)) = app_state.socketio_setup.socket_queue_rx.recv().await {
         event_count += 1;
-
-        send_event_with_retry(&socket, &event).await;
-
+        send_event_with_retry(&socket, &event);
         if batch_start.elapsed().as_secs() >= 5 {
             let elapsed = batch_start.elapsed();
             if event_count > 0 {