Fix manifest sync: zombie prevention, template persistence, and sender race

- Populate all known_ids from local sub-services at startup (init from
  reality) to prevent pruning of pre-existing resources on first manifest
- Include completed templates in node manifest so they aren't pruned
- Register manifest sender before building initial manifest to close race
  window where events during build would be missed

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

Author: MalteJ

mvirt-api/src/grpc/manifest.rs

@@ -152,7 +152,24 @@ async fn build_volume_specs(store: &Arc<dyn DataStore>, node_id: &str) -> Vec<Pr
 async fn build_template_specs(store: &Arc<dyn DataStore>, node_id: &str) -> Vec<ProtoTemplateSpec> {
     let mut specs = Vec::new();
 
-    // Pending/running import jobs — these should be sent to the target node
+    // 1. Completed templates on this node (persistent)
+    if let Ok(templates) = store.list_templates(Some(node_id)).await {
+        for tpl in &templates {
+            specs.push(ProtoTemplateSpec {
+                meta: Some(ResourceMeta {
+                    id: tpl.id.clone(),
+                    name: tpl.name.clone(),
+                    project_id: tpl.project_id.clone(),
+                    node_id: Some(tpl.node_id.clone()),
+                    labels: Default::default(),
+                }),
+                url: String::new(),
+                checksum: None,
+            });
+        }
+    }
+
+    // 2. Pending/running import jobs — these should be sent to the target node
     if let Ok(jobs) = store.list_import_jobs(None).await {
         for job in &jobs {
             if job.state == ImportJobState::Completed || job.state == ImportJobState::Failed {

mvirt-api/src/grpc/server.rs

@@ -237,7 +237,13 @@ impl NodeService for NodeServiceImpl {
         };
         let _ = out_tx.send(Ok(register_result)).await;
 
-        // 4. Build and send initial manifest
+        // 4. Register manifest sender (before building initial manifest to avoid race window)
+        {
+            let mut senders = self.manifest_senders.write().await;
+            senders.insert(registered_id.clone(), manifest_tx);
+        }
+
+        // 5. Build and send initial manifest
         let revision = self.next_revision().await;
         let initial_manifest = build_manifest(&self.store, &registered_id, revision).await;
         let _ = out_tx
@@ -246,12 +252,6 @@ impl NodeService for NodeServiceImpl {
             }))
             .await;
 
-        // 5. Register manifest sender
-        {
-            let mut senders = self.manifest_senders.write().await;
-            senders.insert(registered_id.clone(), manifest_tx);
-        }
-
         tracing::info!("Node {} connected via Sync stream", registered_id);
 
         // 6. Spawn reader task for incoming messages

mvirt-node/src/agent.rs

@@ -119,6 +119,8 @@ pub struct NodeAgent {
     security_group_reconciler: SecurityGroupReconciler,
     route_reconciler: RouteReconciler,
     // Clients for init-from-reality
+    vmm_client: VmmClient,
+    net_client: NetClient,
     zfs_client: ZfsClient,
 }
 
@@ -143,13 +145,15 @@ impl NodeAgent {
             resources,
             audit,
             known_ids: HashMap::new(),
-            vm_reconciler: VmReconciler::new(vmm_client, zfs_client.clone()),
+            vm_reconciler: VmReconciler::new(vmm_client.clone(), zfs_client.clone()),
             network_reconciler: NetworkReconciler::new(net_client.clone()),
             nic_reconciler: NicReconciler::new(net_client.clone()),
             template_reconciler: TemplateReconciler::new(zfs_client.clone()),
             volume_reconciler: VolumeReconciler::new(zfs_client.clone()),
-            security_group_reconciler: SecurityGroupReconciler::new(net_client),
+            security_group_reconciler: SecurityGroupReconciler::new(net_client.clone()),
             route_reconciler: RouteReconciler::new(),
+            vmm_client,
+            net_client,
             zfs_client,
         }
     }
@@ -168,6 +172,72 @@ impl NodeAgent {
 
     /// Initialize known_ids from local sub-services (init from reality).
     async fn init_from_reality(&mut self) {
+        // VMs from VMM
+        match self.vmm_client.list_vms().await {
+            Ok(vms) => {
+                let ids: HashSet<String> = vms.iter().map(|v| v.id.clone()).collect();
+                info!("Init from reality: {} VMs", ids.len());
+                self.known_ids.insert("vm".to_string(), ids);
+            }
+            Err(e) => {
+                warn!("Failed to list VMs from VMM: {}", e);
+                self.known_ids.insert("vm".to_string(), HashSet::new());
+            }
+        }
+
+        // Networks from net
+        match self.net_client.list_networks().await {
+            Ok(networks) => {
+                let ids: HashSet<String> = networks.iter().map(|n| n.id.clone()).collect();
+                info!("Init from reality: {} networks", ids.len());
+                self.known_ids.insert("network".to_string(), ids);
+            }
+            Err(e) => {
+                warn!("Failed to list networks from net: {}", e);
+                self.known_ids.insert("network".to_string(), HashSet::new());
+            }
+        }
+
+        // NICs from net
+        match self.net_client.list_nics().await {
+            Ok(nics) => {
+                let ids: HashSet<String> = nics.iter().map(|n| n.id.clone()).collect();
+                info!("Init from reality: {} NICs", ids.len());
+                self.known_ids.insert("nic".to_string(), ids);
+            }
+            Err(e) => {
+                warn!("Failed to list NICs from net: {}", e);
+                self.known_ids.insert("nic".to_string(), HashSet::new());
+            }
+        }
+
+        // Security groups from net
+        match self.net_client.list_security_groups().await {
+            Ok(sgs) => {
+                let ids: HashSet<String> = sgs.iter().map(|s| s.id.clone()).collect();
+                info!("Init from reality: {} security groups", ids.len());
+                self.known_ids.insert("security_group".to_string(), ids);
+            }
+            Err(e) => {
+                warn!("Failed to list security groups from net: {}", e);
+                self.known_ids
+                    .insert("security_group".to_string(), HashSet::new());
+            }
+        }
+
+        // Volumes from ZFS
+        match self.zfs_client.list_volumes().await {
+            Ok(volumes) => {
+                let ids: HashSet<String> = volumes.iter().map(|v| v.name.clone()).collect();
+                info!("Init from reality: {} volumes", ids.len());
+                self.known_ids.insert("volume".to_string(), ids);
+            }
+            Err(e) => {
+                warn!("Failed to list volumes from ZFS: {}", e);
+                self.known_ids.insert("volume".to_string(), HashSet::new());
+            }
+        }
+
         // Templates from ZFS
         match self.zfs_client.list_templates().await {
             Ok(templates) => {
@@ -182,10 +252,8 @@ impl NodeAgent {
             }
         }
 
-        // Other resource types start empty — the first manifest will populate them
-        for key in &["vm", "network", "nic", "volume", "security_group", "route"] {
-            self.known_ids.entry(key.to_string()).or_default();
-        }
+        // Routes start empty (no local daemon to query)
+        self.known_ids.entry("route".to_string()).or_default();
     }
 
     /// Main agent loop.

mvirt-node/src/clients/net.rs

@@ -8,7 +8,8 @@ use crate::proto::net::{
     net_service_client::NetServiceClient, AddSecurityGroupRuleRequest, AttachNicRequest,
     AttachSecurityGroupRequest, CreateNetworkRequest, CreateNicRequest, CreateSecurityGroupRequest,
     DeleteNetworkRequest, DeleteNicRequest, DeleteSecurityGroupRequest, DetachSecurityGroupRequest,
-    GetNicRequest, RemoveSecurityGroupRuleRequest,
+    GetNicRequest, ListNetworksRequest, ListNicsRequest, ListSecurityGroupsRequest,
+    RemoveSecurityGroupRuleRequest,
 };
 
 pub use crate::proto::net::{
@@ -30,6 +31,43 @@ impl NetClient {
         Ok(Self { client })
     }
 
+    // === List operations ===
+
+    /// List all networks.
+    pub async fn list_networks(&mut self) -> Result<Vec<Network>> {
+        debug!("Listing networks from mvirt-net");
+        let resp = self
+            .client
+            .list_networks(ListNetworksRequest {})
+            .await
+            .context("Failed to list networks")?;
+        Ok(resp.into_inner().networks)
+    }
+
+    /// List all NICs.
+    pub async fn list_nics(&mut self) -> Result<Vec<Nic>> {
+        debug!("Listing NICs from mvirt-net");
+        let resp = self
+            .client
+            .list_nics(ListNicsRequest {
+                network_id: String::new(),
+            })
+            .await
+            .context("Failed to list NICs")?;
+        Ok(resp.into_inner().nics)
+    }
+
+    /// List all security groups.
+    pub async fn list_security_groups(&mut self) -> Result<Vec<SecurityGroup>> {
+        debug!("Listing security groups from mvirt-net");
+        let resp = self
+            .client
+            .list_security_groups(ListSecurityGroupsRequest {})
+            .await
+            .context("Failed to list security groups")?;
+        Ok(resp.into_inner().security_groups)
+    }
+
     // === Network operations ===
 
     /// Create a network.

mvirt-node/src/clients/vmm.rs

@@ -6,12 +6,13 @@ use tracing::debug;
 
 use crate::proto::vmm::{
     vm_service_client::VmServiceClient, CreateVmRequest, DeleteVmRequest, GetVmRequest,
-    KillVmRequest, StartVmRequest, StopVmRequest,
+    KillVmRequest, ListVmsRequest, StartVmRequest, StopVmRequest,
 };
 
 pub use crate::proto::vmm::{BootMode, DiskConfig, NicConfig, Vm, VmConfig, VmState};
 
 /// Client for interacting with mvirt-vmm.
+#[derive(Clone)]
 pub struct VmmClient {
     client: VmServiceClient<Channel>,
 }
@@ -88,6 +89,17 @@ impl VmmClient {
         Ok(resp.into_inner())
     }
 
+    /// List all VMs.
+    pub async fn list_vms(&mut self) -> Result<Vec<Vm>> {
+        debug!("Listing VMs from mvirt-vmm");
+        let resp = self
+            .client
+            .list_vms(ListVmsRequest {})
+            .await
+            .context("Failed to list VMs")?;
+        Ok(resp.into_inner().vms)
+    }
+
     /// Delete a VM.
     pub async fn delete_vm(&mut self, id: &str) -> Result<()> {
         debug!("Deleting VM {} in mvirt-vmm", id);

mvirt-node/src/clients/zfs.rs

@@ -7,7 +7,7 @@ use tracing::debug;
 use crate::proto::zfs::{
     zfs_service_client::ZfsServiceClient, CloneFromTemplateRequest, CreateVolumeRequest,
     DeleteTemplateRequest, DeleteVolumeRequest, GetImportJobRequest, GetVolumeRequest,
-    ImportTemplateRequest, ListTemplatesRequest,
+    ImportTemplateRequest, ListTemplatesRequest, ListVolumesRequest,
 };
 
 pub use crate::proto::zfs::{ImportJob, Template, Volume};
@@ -26,6 +26,17 @@ impl ZfsClient {
         Ok(Self { client })
     }
 
+    /// List all volumes.
+    pub async fn list_volumes(&mut self) -> Result<Vec<Volume>> {
+        debug!("Listing volumes from mvirt-zfs");
+        let resp = self
+            .client
+            .list_volumes(ListVolumesRequest {})
+            .await
+            .context("Failed to list volumes")?;
+        Ok(resp.into_inner().volumes)
+    }
+
     /// Get volume by name.
     pub async fn get_volume(&mut self, name: &str) -> Result<Option<Volume>> {
         debug!("Getting volume {} from mvirt-zfs", name);