fix(manager): acquire only as many deployments as can be lease-guarded

crates/alien-deployment/src/runner.rs

@@ -38,8 +38,29 @@ use tracing::{debug, info, warn};
 const CHECKPOINT_RETRY_INITIAL_DELAY: Duration = Duration::from_secs(1);
 const CHECKPOINT_RETRY_MAX_DELAY: Duration = Duration::from_secs(30);
 const LEASE_RENEW_INTERVAL: Duration = Duration::from_secs(60);
-const LEASE_RENEW_REQUEST_TIMEOUT: Duration = Duration::from_secs(30);
-const LEASE_RENEW_FAILURE_DEADLINE: Duration = Duration::from_secs(4 * 60);
+const LEASE_RENEW_REQUEST_TIMEOUT: Duration = Duration::from_secs(10);
+const LEASE_RENEW_FAILURE_DEADLINE: Duration = Duration::from_secs(3 * 60 + 30);
+
+/// The shortest confirmed lease a deployment store is expected to grant.
+///
+/// The runner must give up ownership strictly before the store would let another
+/// session acquire the deployment, otherwise two runners could act on the same
+/// target. The worst case is a renewal attempt that starts just before the
+/// failure deadline and then times out, so the latest the runner can stop is
+/// `LEASE_RENEW_FAILURE_DEADLINE + LEASE_RENEW_REQUEST_TIMEOUT` after the last
+/// success. The assertion below keeps that strictly inside the lease.
+const MINIMUM_CONFIRMED_LEASE: Duration = Duration::from_secs(5 * 60);
+
+const _: () = assert!(
+    LEASE_RENEW_FAILURE_DEADLINE.as_secs() + LEASE_RENEW_REQUEST_TIMEOUT.as_secs()
+        < MINIMUM_CONFIRMED_LEASE.as_secs(),
+    "the runner must abandon a deployment before its lease can be reacquired elsewhere",
+);
+
+const _: () = assert!(
+    LEASE_RENEW_INTERVAL.as_secs() < LEASE_RENEW_FAILURE_DEADLINE.as_secs(),
+    "at least one renewal must be attempted before the failure deadline",
+);
 
 /// Policy configuration for the runner.
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]

crates/alien-manager/src/loops/deployment.rs

@@ -39,6 +39,10 @@ use crate::transports::ManagerTransport;
 const MAX_STEPS_PER_TICK: usize = 100;
 const MAX_STEPS_PER_HEARTBEAT: usize = 1;
 /// Maximum number of deployments to process concurrently per tick.
+///
+/// This is also the acquire batch size. A deployment's lease is only renewed
+/// once it is actually being processed, so acquiring more than can be processed
+/// at once would leave the surplus holding leases that nothing is extending.
 pub(crate) const MAX_CONCURRENT_DEPLOYMENTS: usize = 4;
 /// Maximum acquire/process batches before yielding back to the interval sleep.
 const MAX_ACQUIRE_BATCHES_PER_TICK: usize = 16;
@@ -308,7 +312,7 @@ impl DeploymentLoop {
 
             match self
                 .deployment_store
-                .acquire(&caller, &session, &filter, 10)
+                .acquire(&caller, &session, &filter, MAX_CONCURRENT_DEPLOYMENTS as u32)
                 .await
             {
                 Ok(acquired) => {

crates/alien-manager/src/loops/heartbeat.rs

@@ -14,7 +14,8 @@ use crate::loops::deployment::{DeploymentLoop, MAX_CONCURRENT_DEPLOYMENTS};
 use crate::traits::deployment_store::DeploymentFilter;
 use crate::traits::DeploymentStore;
 
-const HEARTBEAT_ACQUIRE_LIMIT: u32 = 100;
+/// Maximum deployments to health-check per tick, across all batches.
+const HEARTBEAT_DEPLOYMENTS_PER_TICK: usize = 100;
 
 pub struct HeartbeatLoop {
     config: Arc<ManagerConfig>,
@@ -65,30 +66,44 @@ impl HeartbeatLoop {
         // empty `bearer_token` — the documented signal to embedders that
         // no caller passthrough is available.
         let caller = Subject::system();
-        let session = uuid::Uuid::new_v4().to_string();
-        match self
-            .deployment_store
-            .acquire(&caller, &session, &filter, HEARTBEAT_ACQUIRE_LIMIT)
-            .await
-        {
-            Ok(acquired) => {
-                if !acquired.is_empty() {
+
+        // Acquire in batches no larger than the processing concurrency. A lease
+        // is only renewed once its deployment is being processed, so a larger
+        // batch would leave the surplus holding leases that nothing extends.
+        let batches = HEARTBEAT_DEPLOYMENTS_PER_TICK.div_ceil(MAX_CONCURRENT_DEPLOYMENTS);
+        for _ in 0..batches {
+            let session = uuid::Uuid::new_v4().to_string();
+            match self
+                .deployment_store
+                .acquire(
+                    &caller,
+                    &session,
+                    &filter,
+                    MAX_CONCURRENT_DEPLOYMENTS as u32,
+                )
+                .await
+            {
+                Ok(acquired) => {
+                    if acquired.is_empty() {
+                        break;
+                    }
                     debug!(
                         count = acquired.len(),
                         session = %session,
                         "Heartbeat: acquired running deployments"
                     );
+                    stream::iter(acquired)
+                        .for_each_concurrent(MAX_CONCURRENT_DEPLOYMENTS, |item| async {
+                            self.deployment_loop
+                                .process_heartbeat_deployment(item.deployment, &session)
+                                .await;
+                        })
+                        .await;
+                }
+                Err(e) => {
+                    error!(error = %e, "Heartbeat: failed to acquire running deployments");
+                    break;
                 }
-                stream::iter(acquired)
-                    .for_each_concurrent(MAX_CONCURRENT_DEPLOYMENTS, |item| async {
-                        self.deployment_loop
-                            .process_heartbeat_deployment(item.deployment, &session)
-                            .await;
-                    })
-                    .await;
-            }
-            Err(e) => {
-                error!(error = %e, "Heartbeat: failed to acquire running deployments");
             }
         }
     }