Method lease_coordinator

crates/openshell-server/src/compute/mod.rs:878–913

(self: Arc<Self>, mut shutdown_rx: watch::Receiver<bool>)

    }

    async fn lease_coordinator(self: Arc<Self>, mut shutdown_rx: watch::Receiver<bool>) {
        use lease::{LEASE_ACQUIRE_INTERVAL, LEASE_TTL, ReconcilerLease};

        let lease = ReconcilerLease::new(self.store.clone(), self.replica_id.clone(), LEASE_TTL);
        info!(replica = %lease.replica_id(), "reconciler lease coordinator started");

        loop {
            if *shutdown_rx.borrow() {
                break;
            }

            match lease.acquire_or_steal().await {
                Ok(guard) => {
                    info!(replica = %lease.replica_id(), "acquired reconciler lease");
                    self.run_as_holder(&lease, guard, &mut shutdown_rx).await;
                }
                Err(e) => {
                    debug!(
                        replica = %lease.replica_id(),
                        error = %e,
                        "reconciler lease acquisition attempt failed"
                    );
                    tokio::select! {
                        () = tokio::time::sleep(LEASE_ACQUIRE_INTERVAL) => {}
                        _ = shutdown_rx.changed() => {
                            if *shutdown_rx.borrow() {
                                break;
                            }
                        }
                    }
                }
            }
        }

        info!(replica = %lease.replica_id(), "reconciler lease coordinator stopped");
    }

    async fn run_as_holder(
        self: &Arc<Self>,