Support legacy daemons (< v0.14.0) in ServerPoll mode

mayanayza · mayanayza · commit 134b129086ab · 2026-02-02T19:31:28.000-05:00
- Add supports_full_server_poll() method to check daemon version
- Skip polling for legacy daemons (they don't have /api/status, /api/poll, etc.)
- Make API key optional in post_to_daemon and discovery request methods
- Allow subscriber to send discovery commands without auth for legacy daemons
- Add info-level logging for HubSpot org sync on startup

Legacy daemons stay alive via their own heartbeat calls and can still
receive discovery initiate/cancel commands without authentication.
diff --git a/backend/src/bin/server.rs b/backend/src/bin/server.rs
@@ -286,11 +286,14 @@ async fn main() -> anyhow::Result<()> {
 
     // Sync existing organizations to HubSpot if configured
     if let Some(hubspot_service) = state.services.hubspot_service.clone() {
+        tracing::info!(target: LOG_TARGET, "  Spawning HubSpot organization sync task");
         tokio::spawn(async move {
             if let Err(e) = hubspot_service.sync_existing_organizations().await {
                 tracing::error!(target: LOG_TARGET, error = %e, "Failed to sync existing organizations to HubSpot");
             }
         });
+    } else {
+        tracing::info!(target: LOG_TARGET, "  HubSpot service not configured, skipping org sync");
     }
 
     // Configuration summary
diff --git a/backend/src/server/daemons/impl/base.rs b/backend/src/server/daemons/impl/base.rs
@@ -77,6 +77,23 @@ impl Daemon {
             && self.base.network_id == other.base.network_id
             && self.base.host_id == other.base.host_id
     }
+
+    /// Check if daemon supports full ServerPoll mode (v0.14.0+).
+    ///
+    /// Legacy daemons (< v0.14.0) only support `/api/discovery/initiate` and
+    /// `/api/discovery/cancel` endpoints without authentication.
+    /// They don't support the newer endpoints: `/api/status`, `/api/poll`,
+    /// `/api/first-contact`, `/api/discovery/entities-created`.
+    ///
+    /// Returns `false` for daemons without a version (assume legacy).
+    pub fn supports_full_server_poll(&self) -> bool {
+        const SERVER_POLL_VERSION: Version = Version::new(0, 14, 0);
+        self.base
+            .version
+            .as_ref()
+            .map(|v| v >= &SERVER_POLL_VERSION)
+            .unwrap_or(false)
+    }
 }
 
 impl Display for Daemon {
diff --git a/backend/src/server/daemons/service.rs b/backend/src/server/daemons/service.rs
@@ -224,29 +224,34 @@ impl DaemonService {
         .await
     }
 
-    /// Send POST request to daemon with auth and retry.
+    /// Send POST request to daemon with optional auth and retry.
     /// Uses exponential backoff: 5 retries, 5-30s delays.
     /// Returns `Option<T>` - `Some(data)` if response contains data, `None` otherwise.
     /// For endpoints that don't return data, use `::<serde_json::Value>` and ignore result.
+    ///
+    /// If `api_key` is `None`, the request is sent without an Authorization header.
+    /// This is used for legacy daemons (< v0.14.0) that don't require authentication.
     async fn post_to_daemon<T: serde::de::DeserializeOwned>(
         &self,
         daemon: &Daemon,
-        api_key: &str,
+        api_key: Option<&str>,
         path: &str,
         body: &impl serde::Serialize,
     ) -> Result<Option<T>> {
         let url = format!("{}{}", daemon.base.url, path);
         let daemon_id = daemon.id;
         let body_json = serde_json::to_value(body)?;
+        let api_key_owned = api_key.map(|s| s.to_owned());
 
         (|| async {
-            let response = self
-                .client
-                .post(&url)
-                .header("Authorization", format!("Bearer {}", api_key))
-                .json(&body_json)
-                .send()
-                .await?;
+            let mut request = self.client.post(&url).json(&body_json);
+
+            // Only add auth header if API key provided (v0.14.0+ daemons)
+            if let Some(ref key) = api_key_owned {
+                request = request.header("Authorization", format!("Bearer {}", key));
+            }
+
+            let response = request.send().await?;
 
             if !response.status().is_success() {
                 anyhow::bail!("POST {} failed: HTTP {}", path, response.status());
@@ -317,7 +322,7 @@ impl DaemonService {
         let _: Option<serde_json::Value> = self
             .post_to_daemon(
                 daemon,
-                api_key,
+                Some(api_key),
                 "/api/discovery/entities-created",
                 &created_entities,
             )
@@ -333,11 +338,14 @@ impl DaemonService {
         Ok(())
     }
 
-    /// Send discovery request to daemon (HTTP only, no event publishing)
+    /// Send discovery request to daemon (HTTP only, no event publishing).
+    ///
+    /// If `api_key` is `None`, the request is sent without authentication.
+    /// This is used for legacy daemons (< v0.14.0) that don't require auth.
     pub async fn send_discovery_request_to_daemon(
         &self,
         daemon: &Daemon,
-        api_key: &str,
+        api_key: Option<&str>,
         request: DaemonDiscoveryRequest,
     ) -> Result<(), Error> {
         tracing::info!(
@@ -359,11 +367,14 @@ impl DaemonService {
         Ok(())
     }
 
-    /// Send discovery cancellation to daemon (HTTP only, no event publishing)
+    /// Send discovery cancellation to daemon (HTTP only, no event publishing).
+    ///
+    /// If `api_key` is `None`, the request is sent without authentication.
+    /// This is used for legacy daemons (< v0.14.0) that don't require auth.
     pub async fn send_discovery_cancellation_to_daemon(
         &self,
         daemon: &Daemon,
-        api_key: &str,
+        api_key: Option<&str>,
         session_id: Uuid,
     ) -> Result<(), Error> {
         let _: Option<serde_json::Value> = self
@@ -394,7 +405,7 @@ impl DaemonService {
             server_capabilities,
         };
 
-        self.post_to_daemon(daemon, api_key, "/api/first-contact", &request)
+        self.post_to_daemon(daemon, Some(api_key), "/api/first-contact", &request)
             .await?
             .ok_or_else(|| anyhow::anyhow!("First contact response missing daemon status"))
     }
@@ -1067,7 +1078,23 @@ impl DaemonService {
     /// Poll a single daemon for status and discovery data.
     /// Uses backon for retry with exponential backoff.
     /// Marks daemon unreachable after UNREACHABLE_THRESHOLD failures.
+    ///
+    /// Legacy daemons (< v0.14.0) are skipped entirely - they don't support
+    /// the polling endpoints (/api/status, /api/poll, /api/first-contact,
+    /// /api/discovery/entities-created). Legacy daemons stay alive via their
+    /// own heartbeat calls to the server's backward-compat endpoint.
     async fn poll_daemon(&self, daemon: &Daemon) -> Result<()> {
+        // Skip polling for legacy daemons - they don't have the new endpoints
+        if !daemon.supports_full_server_poll() {
+            tracing::debug!(
+                daemon_id = %daemon.id,
+                daemon_name = %daemon.base.name,
+                version = ?daemon.base.version,
+                "Skipping poll for legacy daemon (< v0.14.0) - polling endpoints not supported"
+            );
+            return Ok(());
+        }
+
         tracing::debug!(
             daemon_id = %daemon.id,
             daemon_name = %daemon.base.name,
@@ -1296,7 +1323,7 @@ impl DaemonService {
                 discovery_type: work.discovery_type,
             };
             if let Err(e) = self
-                .send_discovery_request_to_daemon(daemon, &api_key, request)
+                .send_discovery_request_to_daemon(daemon, Some(&api_key), request)
                 .await
             {
                 tracing::warn!(
diff --git a/backend/src/server/daemons/subscriber.rs b/backend/src/server/daemons/subscriber.rs
@@ -45,17 +45,28 @@ impl EventSubscriber for DaemonService {
                     continue;
                 }
 
-                // Get the API key for this daemon
-                let api_key = match self.get_daemon_api_key(&daemon).await {
-                    Ok(key) => key,
-                    Err(e) => {
-                        tracing::error!(
-                            error = ?e,
-                            daemon_id = %discovery_event.daemon_id,
-                            "Failed to get API key for daemon, skipping event"
-                        );
-                        continue;
+                // Get the API key - optional for legacy daemons (< v0.14.0)
+                // Legacy daemons only have /api/discovery/initiate and /api/discovery/cancel
+                // and they don't require authentication
+                let api_key = if daemon.supports_full_server_poll() {
+                    match self.get_daemon_api_key(&daemon).await {
+                        Ok(key) => Some(key),
+                        Err(e) => {
+                            tracing::error!(
+                                error = ?e,
+                                daemon_id = %discovery_event.daemon_id,
+                                "Failed to get API key for daemon, skipping event"
+                            );
+                            continue;
+                        }
                     }
+                } else {
+                    tracing::debug!(
+                        daemon_id = %discovery_event.daemon_id,
+                        version = ?daemon.base.version,
+                        "Legacy daemon (< v0.14.0) - sending discovery command without auth"
+                    );
+                    None
                 };
 
                 match discovery_event.phase {
@@ -72,7 +83,7 @@ impl EventSubscriber for DaemonService {
                         };
 
                         if let Err(e) = self
-                            .send_discovery_request_to_daemon(&daemon, &api_key, request)
+                            .send_discovery_request_to_daemon(&daemon, api_key.as_deref(), request)
                             .await
                         {
                             tracing::error!(
@@ -93,7 +104,7 @@ impl EventSubscriber for DaemonService {
                         if let Err(e) = self
                             .send_discovery_cancellation_to_daemon(
                                 &daemon,
-                                &api_key,
+                                api_key.as_deref(),
                                 discovery_event.session_id,
                             )
                             .await
diff --git a/backend/src/server/hubspot/service.rs b/backend/src/server/hubspot/service.rs
@@ -586,6 +586,8 @@ impl HubSpotService {
     /// Sync all organizations that don't have a HubSpot company ID.
     /// Called on server startup.
     pub async fn sync_existing_organizations(&self) -> Result<()> {
+        tracing::info!("Starting HubSpot organization sync check");
+
         let filter = StorableFilter::<Organization>::new_without_hubspot_company_id();
         let orgs = self.organization_service.get_all(filter).await?;
 
@@ -600,15 +602,22 @@ impl HubSpotService {
         );
 
         for org in orgs {
+            tracing::info!(
+                organization_id = %org.id,
+                org_name = %org.base.name,
+                "Syncing organization to HubSpot"
+            );
             if let Err(e) = self.sync_organization(org).await {
                 tracing::error!(
-                    organization_id = %e,
+                    error = %e,
                     "Failed to sync organization to HubSpot"
                 );
                 // Continue with other orgs
             }
         }
 
+        tracing::info!("HubSpot organization sync check complete");
+
         Ok(())
     }
 
diff --git a/backend/tests/integration/main.rs b/backend/tests/integration/main.rs
@@ -106,38 +106,18 @@ async fn integration_tests() {
     println!("✅ ServerPoll daemon provisioned: {}", serverpoll_daemon_id);
 
     // =========================================================================
-    // Phase 4: API Compatibility Tests
+    // Phase 4: ServerPoll Discovery
     // =========================================================================
+    // Run discovery BEFORE compat tests to ensure the daemon has clean state.
+    // Compat tests send fixture requests that can leave stale sessions in the
+    // server's daemon_sessions map, causing discovery to skip publishing events.
     println!("\n============================================================");
-    println!("Phase 4: API Compatibility Tests");
-    println!("============================================================");
-
-    compat::run_compat_tests(
-        daemon.id, // Use DaemonPoll daemon ID (like original)
-        network.id,
-        organization.id,
-        user.id,
-        &serverpoll_api_key,
-    )
-    .await
-    .expect("Compatibility tests failed");
-
-    // =========================================================================
-    // Phase 5: Full Integration Verification
-    // =========================================================================
-    println!("\n============================================================");
-    println!("Phase 5: Full Integration Verification");
+    println!("Phase 4: ServerPoll Discovery");
     println!("============================================================\n");
 
-    // Clear DB data and in-memory session state
+    // Clear any leftover discovery data from previous test runs
     clear_discovery_data().expect("Failed to clear discovery data");
 
-    // Cancel any stale sessions from compat tests that may still be in memory.
-    // This prevents the daemon from processing old fixture sessions instead of the new one.
-    compat::cancel_server_discovery_sessions(&client)
-        .await
-        .expect("Failed to clear stale discovery sessions");
-
     // Trigger discovery for the ServerPoll daemon and get the session_id
     let session_id = discovery::trigger_discovery(&client, serverpoll_daemon_id, network.id)
         .await
@@ -168,7 +148,26 @@ async fn integration_tests() {
         .await
         .expect("Failed to create group");
 
-    println!("\n✅ ServerPoll integration flow completed!");
+    println!("\n✅ ServerPoll discovery completed!");
+
+    // =========================================================================
+    // Phase 5: API Compatibility Tests
+    // =========================================================================
+    // Runs after discovery so any stale sessions left by fixture playback
+    // don't affect the real discovery flow.
+    println!("\n============================================================");
+    println!("Phase 5: API Compatibility Tests");
+    println!("============================================================");
+
+    compat::run_compat_tests(
+        daemon.id, // Use DaemonPoll daemon ID (like original)
+        network.id,
+        organization.id,
+        user.id,
+        &serverpoll_api_key,
+    )
+    .await
+    .expect("Compatibility tests failed");
 
     // =========================================================================
     // Phase 6: CRUD Endpoint Tests

Original file line number	Diff line number	Diff line change
`@@ -286,11 +286,14 @@ async fn main() -> anyhow::Result<()> {`
`286`	`286`
`287`	`287`	`// Sync existing organizations to HubSpot if configured`
`288`	`288`	`if let Some(hubspot_service) = state.services.hubspot_service.clone() {`
	`289`	`+ tracing::info!(target: LOG_TARGET, " Spawning HubSpot organization sync task");`
`289`	`290`	`tokio::spawn(async move {`
`290`	`291`	`if let Err(e) = hubspot_service.sync_existing_organizations().await {`
`291`	`292`	`tracing::error!(target: LOG_TARGET, error = %e, "Failed to sync existing organizations to HubSpot");`
`292`	`293`	`}`
`293`	`294`	`});`
	`295`	`+ } else {`
	`296`	`+ tracing::info!(target: LOG_TARGET, " HubSpot service not configured, skipping org sync");`
`294`	`297`	`}`
`295`	`298`
`296`	`299`	`// Configuration summary`