fix todos

Cargo.toml

@@ -42,7 +42,8 @@ codegen-units = 1
 # for any crates using kameo actors on_panic, this would literally break the binary.
 # For specific crates/binaries this might be something we do, but I (caleb) doubt it.
 
-strip = "symbols" # TODO: this we should talk about doing this, because it makes debugging harder, so if we have a panic in production, the backtrace will likely not be as useful.
+# Keep symbol names in production binaries so panic backtraces remain actionable.
+strip = "none"
 
 incremental = false # this is just to get reproducible builds. incremental compiles aren't consistent and can sometimes be broken when combined with optimizations
 

odorobo/src/actors/agent_actor.rs

@@ -121,20 +121,16 @@ impl Actor for AgentActor {
         })
     }
 
-    // async fn on_panic(state: Self::Args, weak_actor_ref: WeakActorRef<Self>, _panic: &PanicError) {
-    //     panic!("Agent panicked: {:?}", _panic);
-    // }
-    //
     async fn on_panic(
         &mut self,
         _actor_ref: WeakActorRef<Self>,
         err: PanicError,
-    ) -> Result<std::ops::ControlFlow<ActorStopReason>> {
-        error!("Agent panicked: {:?}", err);
-
-        // todo: if we panic, we should completely regen the self struct from scratch. The assumption should be that memory corruption could have possibly happened becauew
-
-        Ok(ControlFlow::Continue(()))
+    ) -> Result<ControlFlow<ActorStopReason>> {
+        error!(
+            ?err,
+            "Agent actor panicked; stopping because its state cannot be safely rebuilt here"
+        );
+        Ok(ControlFlow::Break(ActorStopReason::Panicked(err)))
     }
 
     async fn on_link_died(
@@ -165,7 +161,15 @@ impl Message<CreateVM> for AgentActor {
 
     async fn handle(&mut self, msg: CreateVM, ctx: &mut Context<Self, Self::Reply>) -> Self::Reply {
         let vmid = msg.vmid;
-        // spawn AND link at the same time
+        if let Some(existing) = self.vms.get(&vmid) {
+            info!(?vmid, actor_id = ?existing.actor_ref.id(), "VM already exists; treating create as idempotent");
+            return CreateVMReply {
+                config: Some(msg.config),
+                actor_id: Some(existing.actor_ref.id().to_bytes()),
+            };
+        }
+
+        // Spawn and link at the same time.
         let actor_ref =
             VMActor::spawn_link(ctx.actor_ref(), (vmid, Some(msg.config.clone()))).await;
 

odorobo/src/actors/http_actor.rs

@@ -1,6 +1,7 @@
 use crate::messages::vm::{
     AgentListVMs, AgentListVMsReply, CreateVM, CreateVMReply, DeleteVM, DeleteVMReply,
-    GetConsoleHistory, GetConsoleHistoryReply, ShutdownVM, ShutdownVMReply,
+    GetConsoleHistory, GetConsoleHistoryReply, GetVMInfo, GetVMInfoReply, ShutdownVM,
+    ShutdownVMReply,
 };
 use kameo::prelude::*;
 use stable_eyre::{
@@ -102,6 +103,22 @@ impl Message<ShutdownVM> for HTTPActor {
     }
 }
 
+impl Message<GetVMInfo> for HTTPActor {
+    type Reply = Result<GetVMInfoReply, Report>;
+
+    async fn handle(
+        &mut self,
+        msg: GetVMInfo,
+        _ctx: &mut Context<Self, Self::Reply>,
+    ) -> Self::Reply {
+        self.scheduler
+            .ask(msg)
+            .await
+            .map_err(|err| eyre!(err.to_string()))
+            .wrap_err("failed to get VM info via scheduler")
+    }
+}
+
 impl Message<AgentListVMs> for HTTPActor {
     type Reply = Result<AgentListVMsReply, Report>;
 

odorobo/src/actors/scheduler_actor/cache.rs

@@ -14,6 +14,7 @@ use crate::manifest::VmManifest;
 use super::{CachedVMActor, SchedulerActor, VmLifecycle, VmPlacement};
 
 const UNRESOLVED_VM_CACHE_TIMEOUT: Duration = Duration::from_secs(30);
+const UNCONFIRMED_RUNNING_PLACEMENT_TIMEOUT: Duration = Duration::from_secs(30);
 
 impl SchedulerActor {
     /// Releases excess allocation once an entry list no longer represents migration.
@@ -51,11 +52,11 @@ impl SchedulerActor {
     /// Expires pending placements that were never confirmed by agent status.
     ///
     /// Pending entries expire after 30 seconds. The five-second discovery loop
-    /// triggers this maintenance, so a slow-to-report create can be forgotten.
-    /// When the expired placement was the last placement for a VM, all correlated
-    /// manifest, placement, and actor-cache state is removed.
+    /// triggers this maintenance. Expiry removes only the unconfirmed placement:
+    /// the manifest remains scheduler intent, allowing periodic reconciliation to
+    /// dispatch a replacement create request.
     pub(super) fn cleanup_unresolved_vm_cache(
-        manifests: &mut AHashMap<Ulid, VmManifest>,
+        _manifests: &mut AHashMap<Ulid, VmManifest>,
         placements: &mut AHashMap<Ulid, Vec<VmPlacement>>,
         data_cache: &mut AHashMap<Ulid, Vec<CachedVMActor>>,
     ) {
@@ -73,8 +74,16 @@ impl SchedulerActor {
             .collect();
 
         for vmid in empty_vmids {
-            Self::remove_vm_state(vmid, manifests, placements, data_cache);
+            if data_cache
+                .get(&vmid)
+                .is_some_and(|entries| entries.iter().all(|entry| entry.actor_ref.is_none()))
+            {
+                data_cache.remove(&vmid);
+            }
         }
+
+        // Keep manifests and empty placement entries: they represent desired VM
+        // state and are consumed by `ReconcileVmPlacements`.
     }
 
     /// Returns every VM that could be on an agent, deduplicating each source in
@@ -153,27 +162,15 @@ impl SchedulerActor {
         }
     }
 
-    /// Removes placements assigned to a departed agent and drops VM state only
-    /// when no placement remains.
-    // TODO: Preserve VM intent and enqueue replacement placement or recreation
-    // when an agent disappears instead of dropping the last VM state.
+    /// Removes placements assigned to a departed agent while retaining empty
+    /// placement entries as VM intent for periodic reconciliation.
     pub(super) fn remove_agent_placements(
         agent_id: ActorId,
-        manifests: &mut AHashMap<Ulid, VmManifest>,
         placements: &mut AHashMap<Ulid, Vec<VmPlacement>>,
-        data_cache: &mut AHashMap<Ulid, Vec<CachedVMActor>>,
     ) {
-        let empty_vmids: Vec<_> = placements
-            .iter_mut()
-            .filter_map(|(vmid, entries)| {
-                entries.retain(|entry| entry.agent_id != agent_id);
-                Self::shrink_non_migrating_entries(entries);
-                entries.is_empty().then_some(*vmid)
-            })
-            .collect();
-
-        for vmid in empty_vmids {
-            Self::remove_vm_state(vmid, manifests, placements, data_cache);
+        for entries in placements.values_mut() {
+            entries.retain(|entry| entry.agent_id != agent_id);
+            Self::shrink_non_migrating_entries(entries);
         }
     }
 
@@ -209,16 +206,16 @@ impl SchedulerActor {
         self.agent_data_cache.remove(&actor_id);
         self.agent_vm_index.remove(&actor_id);
         self.invalidate_pending_resources();
-        Self::remove_agent_placements(
-            actor_id,
-            &mut self.vm_manifests,
-            &mut self.vm_placements,
-            &mut self.vm_data_cache,
-        );
+        Self::remove_agent_placements(actor_id, &mut self.vm_placements);
     }
 
-    /// Aborts VM polling and removes actor state, retaining a VM only when
-    /// another discovered actor or unresolved placement can still represent it.
+    /// Aborts VM polling and makes an unrepresented VM placement recoverable.
+    ///
+    /// A VM actor is not associated with a particular migration entry. Therefore
+    /// placements are cleared only when no discovered actor remains for that VM;
+    /// pending destination placements are retained to avoid racing an in-flight
+    /// migration. Empty placement entries retain the manifest's desired intent
+    /// for periodic reconciliation.
     pub(super) fn cleanup_vm_actor(&mut self, actor_id: ActorId) {
         if let Some(keepalive_task) = self.vm_keepalive_tasks.remove(&actor_id) {
             trace!(?actor_id, "Aborting VM keepalive task");
@@ -227,29 +224,32 @@ impl SchedulerActor {
         let vmid = self.vm_actorid_ulid_map.remove(&actor_id);
         self.invalidate_pending_resources();
         Self::remove_vm_actor(actor_id, &mut self.vm_data_cache);
-        if let Some(vmid) = vmid
-            && self
-                .vm_data_cache
-                .get(&vmid)
-                .is_none_or(|entries| entries.iter().all(|entry| entry.actor_ref.is_none()))
-        {
-            Self::remove_vm_state(
-                vmid,
-                &mut self.vm_manifests,
-                &mut self.vm_placements,
-                &mut self.vm_data_cache,
-            );
+
+        let Some(vmid) = vmid else {
+            return;
+        };
+        let has_discovered_actor = self
+            .vm_data_cache
+            .get(&vmid)
+            .is_some_and(|entries| entries.iter().any(|entry| entry.actor_ref.is_some()));
+        if !has_discovered_actor {
+            if let Some(entries) = self.vm_placements.get_mut(&vmid) {
+                entries.retain(|entry| entry.lifecycle == VmLifecycle::Pending);
+                Self::shrink_non_migrating_entries(entries);
+            }
+            self.invalidate_pending_resources();
         }
     }
 
-    /// Incorporates additions from a status delta into placement observations.
+    /// Incorporates additions and removals from a status delta into placement observations.
     ///
-    /// Removals deliberately do not delete desired placements: they may be
-    /// transient observations and reconciliation must be able to recreate the VM.
+    /// A removal is authoritative for a confirmed (`Running`) placement on this
+    /// agent. Pending placements are retained through their timeout because a
+    /// create can race the next status delta.
     pub(super) fn reconcile_agent_delta(
         agent_id: ActorId,
         added: &[Ulid],
-        _removed: &[Ulid],
+        removed: &[Ulid],
         manifests: &AHashMap<Ulid, VmManifest>,
         placements: &mut AHashMap<Ulid, Vec<VmPlacement>>,
     ) {
@@ -272,17 +272,22 @@ impl SchedulerActor {
             }
         }
 
-        // A removal is an observation about the agent, not a change to the
-        // scheduler's desired state. Keep the placement so reconciliation can
-        // schedule the VM again. The full status path performs the same
-        // distinction for snapshots.
+        for vmid in removed {
+            if let Some(entries) = placements.get_mut(vmid) {
+                entries.retain(|entry| {
+                    entry.agent_id != agent_id || entry.lifecycle == VmLifecycle::Pending
+                });
+                Self::shrink_non_migrating_entries(entries);
+            }
+        }
     }
 
     /// Reconciles scheduler placement observations with a complete agent snapshot.
     ///
     /// Known, reported VMs gain or refresh `Running` placements. Unknown VMs are
-    /// ignored because the scheduler has no retained intent for them. Absent VMs
-    /// leave existing desired placements intact so future reconciliation can act.
+    /// ignored because the scheduler has no retained intent for them. An absent
+    /// running placement expires after its confirmation timeout; an absent pending
+    /// placement stays reserved through its pending timeout.
     pub(super) fn reconcile_agent_placements(
         agent_id: ActorId,
         status: &AgentStatus,
@@ -313,17 +318,27 @@ impl SchedulerActor {
         let empty_vmids: Vec<_> = placements
             .iter_mut()
             .filter_map(|(vmid, entries)| {
-                // TODO: Expire or repair `Running` placements that remain absent
-                // from repeated full snapshots; they are retained indefinitely now.
-                for entry in entries
-                    .iter_mut()
-                    .filter(|entry| entry.agent_id == agent_id)
-                {
-                    if observed.contains(vmid) {
-                        entry.lifecycle = VmLifecycle::Running;
-                        entry.last_confirmed_at = Some(now);
+                entries.retain_mut(|entry| {
+                    if entry.agent_id != agent_id || observed.contains(vmid) {
+                        if entry.agent_id == agent_id {
+                            entry.lifecycle = VmLifecycle::Running;
+                            entry.last_confirmed_at = Some(now);
+                        }
+                        return true;
                     }
-                }
+
+                    match entry.lifecycle {
+                        VmLifecycle::Pending => {
+                            now.duration_since(entry.created_at) < UNRESOLVED_VM_CACHE_TIMEOUT
+                        }
+                        VmLifecycle::Running => {
+                            entry.last_confirmed_at.is_some_and(|last_confirmed_at| {
+                                now.duration_since(last_confirmed_at)
+                                    < UNCONFIRMED_RUNNING_PLACEMENT_TIMEOUT
+                            })
+                        }
+                    }
+                });
                 Self::shrink_non_migrating_entries(entries);
                 entries.is_empty().then_some(*vmid)
             })

odorobo/src/actors/scheduler_actor/discovery.rs

@@ -378,8 +378,46 @@ impl Message<ReconcileVmPlacements> for SchedulerActor {
             &mut self.vm_data_cache,
         );
         self.invalidate_pending_resources();
-        // TODO: Reconcile desired placements absent from agent status by choosing
-        // a healthy agent and issuing `CreateVM`; this currently only expires
-        // unconfirmed pending reservations.
+
+        let unplaced_vms: Vec<_> = self
+            .vm_manifests
+            .iter()
+            .filter_map(|(vmid, manifest)| {
+                self.vm_placements
+                    .get(vmid)
+                    .is_none_or(Vec::is_empty)
+                    .then_some((*vmid, manifest.clone()))
+            })
+            .collect();
+
+        for (vmid, config) in unplaced_vms {
+            let request = crate::messages::vm::CreateVM { vmid, config };
+            match self.schedule_agent(&request) {
+                Ok(agent) => {
+                    self.vm_placements
+                        .entry(vmid)
+                        .or_default()
+                        .push(super::VmPlacement {
+                            agent_id: agent.id(),
+                            lifecycle: super::VmLifecycle::Pending,
+                            created_at: std::time::Instant::now(),
+                            last_confirmed_at: None,
+                        });
+                    self.vm_data_cache
+                        .entry(vmid)
+                        .or_default()
+                        .push(CachedVMActor { actor_ref: None });
+                    self.invalidate_pending_resources();
+
+                    if let Err(error) = agent.tell(&request).send() {
+                        warn!(?error, %vmid, "failed to recreate unplaced VM");
+                        self.vm_placements.insert(vmid, Vec::new());
+                        self.vm_data_cache.remove(&vmid);
+                        self.invalidate_pending_resources();
+                    }
+                }
+                Err(error) => trace!(?error, %vmid, "no eligible agent to recreate VM"),
+            }
+        }
     }
 }