LCOV - code coverage report
Current view: top level - storage_controller/src - reconciler.rs (source / functions) Coverage Total Hit
Test: 07bee600374ccd486c69370d0972d9035964fe68.info Lines: 0.0 % 740 0
Test Date: 2025-02-20 13:11:02 Functions: 0.0 % 55 0

            Line data    Source code
       1              : use crate::pageserver_client::PageserverClient;
       2              : use crate::persistence::Persistence;
       3              : use crate::{compute_hook, service};
       4              : use pageserver_api::controller_api::{AvailabilityZone, MigrationConfig, PlacementPolicy};
       5              : use pageserver_api::models::{
       6              :     LocationConfig, LocationConfigMode, LocationConfigSecondary, TenantConfig, TenantWaitLsnRequest,
       7              : };
       8              : use pageserver_api::shard::{ShardIdentity, TenantShardId};
       9              : use pageserver_client::mgmt_api;
      10              : use reqwest::StatusCode;
      11              : use std::borrow::Cow;
      12              : use std::collections::HashMap;
      13              : use std::sync::Arc;
      14              : use std::time::{Duration, Instant};
      15              : use tokio_util::sync::CancellationToken;
      16              : use utils::backoff::exponential_backoff;
      17              : use utils::generation::Generation;
      18              : use utils::id::{NodeId, TimelineId};
      19              : use utils::lsn::Lsn;
      20              : use utils::pausable_failpoint;
      21              : use utils::sync::gate::GateGuard;
      22              : 
      23              : use crate::compute_hook::{ComputeHook, NotifyError};
      24              : use crate::node::Node;
      25              : use crate::tenant_shard::{IntentState, ObservedState, ObservedStateDelta, ObservedStateLocation};
      26              : 
      27              : const DEFAULT_HEATMAP_PERIOD: &str = "60s";
      28              : 
      29              : /// Object with the lifetime of the background reconcile task that is created
      30              : /// for tenants which have a difference between their intent and observed states.
      31              : pub(super) struct Reconciler {
      32              :     /// See [`crate::tenant_shard::TenantShard`] for the meanings of these fields: they are a snapshot
      33              :     /// of a tenant's state from when we spawned a reconcile task.
      34              :     pub(super) tenant_shard_id: TenantShardId,
      35              :     pub(crate) shard: ShardIdentity,
      36              :     pub(crate) placement_policy: PlacementPolicy,
      37              :     pub(crate) generation: Option<Generation>,
      38              :     pub(crate) intent: TargetState,
      39              : 
      40              :     /// Nodes not referenced by [`Self::intent`], from which we should try
      41              :     /// to detach this tenant shard.
      42              :     pub(crate) detach: Vec<Node>,
      43              : 
      44              :     /// Configuration specific to this reconciler
      45              :     pub(crate) reconciler_config: ReconcilerConfig,
      46              : 
      47              :     pub(crate) config: TenantConfig,
      48              :     pub(crate) preferred_az: Option<AvailabilityZone>,
      49              : 
      50              :     /// Observed state from the point of view of the reconciler.
      51              :     /// This gets updated as the reconciliation makes progress.
      52              :     pub(crate) observed: ObservedState,
      53              : 
      54              :     /// Snapshot of the observed state at the point when the reconciler
      55              :     /// was spawned.
      56              :     pub(crate) original_observed: ObservedState,
      57              : 
      58              :     pub(crate) service_config: service::Config,
      59              : 
      60              :     /// A hook to notify the running postgres instances when we change the location
      61              :     /// of a tenant.  Use this via [`Self::compute_notify`] to update our failure flag
      62              :     /// and guarantee eventual retries.
      63              :     pub(crate) compute_hook: Arc<ComputeHook>,
      64              : 
      65              :     /// To avoid stalling if the cloud control plane is unavailable, we may proceed
      66              :     /// past failures in [`ComputeHook::notify`], but we _must_ remember that we failed
      67              :     /// so that we can set [`crate::tenant_shard::TenantShard::pending_compute_notification`] to ensure a later retry.
      68              :     pub(crate) compute_notify_failure: bool,
      69              : 
      70              :     /// Reconciler is responsible for keeping alive semaphore units that limit concurrency on how many
      71              :     /// we will spawn.
      72              :     pub(crate) _resource_units: ReconcileUnits,
      73              : 
      74              :     /// A means to abort background reconciliation: it is essential to
      75              :     /// call this when something changes in the original TenantShard that
      76              :     /// will make this reconciliation impossible or unnecessary, for
      77              :     /// example when a pageserver node goes offline, or the PlacementPolicy for
      78              :     /// the tenant is changed.
      79              :     pub(crate) cancel: CancellationToken,
      80              : 
      81              :     /// Reconcilers are registered with a Gate so that during a graceful shutdown we
      82              :     /// can wait for all the reconcilers to respond to their cancellation tokens.
      83              :     pub(crate) _gate_guard: GateGuard,
      84              : 
      85              :     /// Access to persistent storage for updating generation numbers
      86              :     pub(crate) persistence: Arc<Persistence>,
      87              : }
      88              : 
      89              : pub(crate) struct ReconcilerConfigBuilder {
      90              :     config: ReconcilerConfig,
      91              : }
      92              : 
      93              : impl ReconcilerConfigBuilder {
      94              :     /// Priority is special: you must pick one thoughtfully, do not just use 'normal' as the default
      95            0 :     pub(crate) fn new(priority: ReconcilerPriority) -> Self {
      96            0 :         Self {
      97            0 :             config: ReconcilerConfig::new(priority),
      98            0 :         }
      99            0 :     }
     100              : 
     101            0 :     pub(crate) fn secondary_warmup_timeout(self, value: Duration) -> Self {
     102            0 :         Self {
     103            0 :             config: ReconcilerConfig {
     104            0 :                 secondary_warmup_timeout: Some(value),
     105            0 :                 ..self.config
     106            0 :             },
     107            0 :         }
     108            0 :     }
     109              : 
     110            0 :     pub(crate) fn secondary_download_request_timeout(self, value: Duration) -> Self {
     111            0 :         Self {
     112            0 :             config: ReconcilerConfig {
     113            0 :                 secondary_download_request_timeout: Some(value),
     114            0 :                 ..self.config
     115            0 :             },
     116            0 :         }
     117            0 :     }
     118              : 
     119            0 :     pub(crate) fn tenant_creation_hint(self, hint: bool) -> Self {
     120            0 :         Self {
     121            0 :             config: ReconcilerConfig {
     122            0 :                 tenant_creation_hint: hint,
     123            0 :                 ..self.config
     124            0 :             },
     125            0 :         }
     126            0 :     }
     127              : 
     128            0 :     pub(crate) fn build(self) -> ReconcilerConfig {
     129            0 :         self.config
     130            0 :     }
     131              : }
     132              : 
     133              : // Higher priorities are used for user-facing tasks, so that a long backlog of housekeeping work (e.g. reconciling on startup, rescheduling
     134              : // things on node changes) does not starve user-facing tasks.
     135              : #[derive(Debug, Copy, Clone)]
     136              : pub(crate) enum ReconcilerPriority {
     137              :     Normal,
     138              :     High,
     139              : }
     140              : 
     141              : #[derive(Debug, Copy, Clone)]
     142              : pub(crate) struct ReconcilerConfig {
     143              :     pub(crate) priority: ReconcilerPriority,
     144              : 
     145              :     // During live migration give up on warming-up the secondary
     146              :     // after this timeout.
     147              :     secondary_warmup_timeout: Option<Duration>,
     148              : 
     149              :     // During live migrations this is the amount of time that
     150              :     // the pagserver will hold our poll.
     151              :     secondary_download_request_timeout: Option<Duration>,
     152              : 
     153              :     // A hint indicating whether this reconciliation is done on the
     154              :     // creation of a new tenant. This only informs logging behaviour.
     155              :     tenant_creation_hint: bool,
     156              : }
     157              : 
     158              : impl ReconcilerConfig {
     159              :     /// Configs are always constructed with an explicit priority, to force callers to think about whether
     160              :     /// the operation they're scheduling is high-priority or not. Normal priority is not a safe default, because
     161              :     /// scheduling something user-facing at normal priority can result in it getting starved out by background work.
     162            0 :     pub(crate) fn new(priority: ReconcilerPriority) -> Self {
     163            0 :         Self {
     164            0 :             priority,
     165            0 :             secondary_warmup_timeout: None,
     166            0 :             secondary_download_request_timeout: None,
     167            0 :             tenant_creation_hint: false,
     168            0 :         }
     169            0 :     }
     170              : 
     171            0 :     pub(crate) fn get_secondary_warmup_timeout(&self) -> Duration {
     172              :         const SECONDARY_WARMUP_TIMEOUT_DEFAULT: Duration = Duration::from_secs(300);
     173            0 :         self.secondary_warmup_timeout
     174            0 :             .unwrap_or(SECONDARY_WARMUP_TIMEOUT_DEFAULT)
     175            0 :     }
     176              : 
     177            0 :     pub(crate) fn get_secondary_download_request_timeout(&self) -> Duration {
     178              :         const SECONDARY_DOWNLOAD_REQUEST_TIMEOUT_DEFAULT: Duration = Duration::from_secs(20);
     179            0 :         self.secondary_download_request_timeout
     180            0 :             .unwrap_or(SECONDARY_DOWNLOAD_REQUEST_TIMEOUT_DEFAULT)
     181            0 :     }
     182              : 
     183            0 :     pub(crate) fn tenant_creation_hint(&self) -> bool {
     184            0 :         self.tenant_creation_hint
     185            0 :     }
     186              : }
     187              : 
     188              : impl From<&MigrationConfig> for ReconcilerConfig {
     189            0 :     fn from(value: &MigrationConfig) -> Self {
     190            0 :         // Run reconciler at high priority because MigrationConfig comes from human requests that should
     191            0 :         // be presumed urgent.
     192            0 :         let mut builder = ReconcilerConfigBuilder::new(ReconcilerPriority::High);
     193              : 
     194            0 :         if let Some(timeout) = value.secondary_warmup_timeout {
     195            0 :             builder = builder.secondary_warmup_timeout(timeout)
     196            0 :         }
     197              : 
     198            0 :         if let Some(timeout) = value.secondary_download_request_timeout {
     199            0 :             builder = builder.secondary_download_request_timeout(timeout)
     200            0 :         }
     201              : 
     202            0 :         builder.build()
     203            0 :     }
     204              : }
     205              : 
     206              : /// RAII resource units granted to a Reconciler, which it should keep alive until it finishes doing I/O
     207              : pub(crate) struct ReconcileUnits {
     208              :     _sem_units: tokio::sync::OwnedSemaphorePermit,
     209              : }
     210              : 
     211              : impl ReconcileUnits {
     212            0 :     pub(crate) fn new(sem_units: tokio::sync::OwnedSemaphorePermit) -> Self {
     213            0 :         Self {
     214            0 :             _sem_units: sem_units,
     215            0 :         }
     216            0 :     }
     217              : }
     218              : 
     219              : /// This is a snapshot of [`crate::tenant_shard::IntentState`], but it does not do any
     220              : /// reference counting for Scheduler.  The IntentState is what the scheduler works with,
     221              : /// and the TargetState is just the instruction for a particular Reconciler run.
     222              : #[derive(Debug)]
     223              : pub(crate) struct TargetState {
     224              :     pub(crate) attached: Option<Node>,
     225              :     pub(crate) secondary: Vec<Node>,
     226              : }
     227              : 
     228              : impl TargetState {
     229            0 :     pub(crate) fn from_intent(nodes: &HashMap<NodeId, Node>, intent: &IntentState) -> Self {
     230            0 :         Self {
     231            0 :             attached: intent.get_attached().map(|n| {
     232            0 :                 nodes
     233            0 :                     .get(&n)
     234            0 :                     .expect("Intent attached referenced non-existent node")
     235            0 :                     .clone()
     236            0 :             }),
     237            0 :             secondary: intent
     238            0 :                 .get_secondary()
     239            0 :                 .iter()
     240            0 :                 .map(|n| {
     241            0 :                     nodes
     242            0 :                         .get(n)
     243            0 :                         .expect("Intent secondary referenced non-existent node")
     244            0 :                         .clone()
     245            0 :                 })
     246            0 :                 .collect(),
     247            0 :         }
     248            0 :     }
     249              : }
     250              : 
     251              : #[derive(thiserror::Error, Debug)]
     252              : pub(crate) enum ReconcileError {
     253              :     #[error(transparent)]
     254              :     Remote(#[from] mgmt_api::Error),
     255              :     #[error(transparent)]
     256              :     Notify(#[from] NotifyError),
     257              :     #[error("Cancelled")]
     258              :     Cancel,
     259              :     #[error(transparent)]
     260              :     Other(#[from] anyhow::Error),
     261              : }
     262              : 
     263              : impl Reconciler {
     264            0 :     async fn location_config(
     265            0 :         &mut self,
     266            0 :         node: &Node,
     267            0 :         config: LocationConfig,
     268            0 :         flush_ms: Option<Duration>,
     269            0 :         lazy: bool,
     270            0 :     ) -> Result<(), ReconcileError> {
     271            0 :         if !node.is_available() && config.mode == LocationConfigMode::Detached {
     272              :             // [`crate::service::Service::node_activate_reconcile`] will update the observed state
     273              :             // when the node comes back online. At that point, the intent and observed states will
     274              :             // be mismatched and a background reconciliation will detach.
     275            0 :             tracing::info!(
     276            0 :                 "Node {node} is unavailable during detach: proceeding anyway, it will be detached via background reconciliation"
     277              :             );
     278            0 :             return Ok(());
     279            0 :         }
     280            0 : 
     281            0 :         self.observed
     282            0 :             .locations
     283            0 :             .insert(node.get_id(), ObservedStateLocation { conf: None });
     284            0 : 
     285            0 :         // TODO: amend locations that use long-polling: they will hit this timeout.
     286            0 :         let timeout = Duration::from_secs(25);
     287            0 : 
     288            0 :         tracing::info!("location_config({node}) calling: {:?}", config);
     289            0 :         let tenant_shard_id = self.tenant_shard_id;
     290            0 :         let config_ref = &config;
     291            0 :         match node
     292            0 :             .with_client_retries(
     293            0 :                 |client| async move {
     294            0 :                     let config = config_ref.clone();
     295            0 :                     client
     296            0 :                         .location_config(tenant_shard_id, config.clone(), flush_ms, lazy)
     297            0 :                         .await
     298            0 :                 },
     299            0 :                 &self.service_config.jwt_token,
     300            0 :                 1,
     301            0 :                 3,
     302            0 :                 timeout,
     303            0 :                 &self.cancel,
     304            0 :             )
     305            0 :             .await
     306              :         {
     307            0 :             Some(Ok(_)) => {}
     308            0 :             Some(Err(e)) => return Err(e.into()),
     309            0 :             None => return Err(ReconcileError::Cancel),
     310              :         };
     311            0 :         tracing::info!("location_config({node}) complete: {:?}", config);
     312              : 
     313            0 :         match config.mode {
     314            0 :             LocationConfigMode::Detached => {
     315            0 :                 self.observed.locations.remove(&node.get_id());
     316            0 :             }
     317            0 :             _ => {
     318            0 :                 self.observed
     319            0 :                     .locations
     320            0 :                     .insert(node.get_id(), ObservedStateLocation { conf: Some(config) });
     321            0 :             }
     322              :         }
     323              : 
     324            0 :         Ok(())
     325            0 :     }
     326              : 
     327            0 :     fn get_node(&self, node_id: &NodeId) -> Option<&Node> {
     328            0 :         if let Some(node) = self.intent.attached.as_ref() {
     329            0 :             if node.get_id() == *node_id {
     330            0 :                 return Some(node);
     331            0 :             }
     332            0 :         }
     333              : 
     334            0 :         if let Some(node) = self
     335            0 :             .intent
     336            0 :             .secondary
     337            0 :             .iter()
     338            0 :             .find(|n| n.get_id() == *node_id)
     339              :         {
     340            0 :             return Some(node);
     341            0 :         }
     342              : 
     343            0 :         if let Some(node) = self.detach.iter().find(|n| n.get_id() == *node_id) {
     344            0 :             return Some(node);
     345            0 :         }
     346            0 : 
     347            0 :         None
     348            0 :     }
     349              : 
     350            0 :     async fn maybe_live_migrate(&mut self) -> Result<(), ReconcileError> {
     351            0 :         let destination = if let Some(node) = &self.intent.attached {
     352            0 :             match self.observed.locations.get(&node.get_id()) {
     353            0 :                 Some(conf) => {
     354              :                     // We will do a live migration only if the intended destination is not
     355              :                     // currently in an attached state.
     356            0 :                     match &conf.conf {
     357            0 :                         Some(conf) if conf.mode == LocationConfigMode::Secondary => {
     358            0 :                             // Fall through to do a live migration
     359            0 :                             node
     360              :                         }
     361              :                         None | Some(_) => {
     362              :                             // Attached or uncertain: don't do a live migration, proceed
     363              :                             // with a general-case reconciliation
     364            0 :                             tracing::info!("maybe_live_migrate: destination is None or attached");
     365            0 :                             return Ok(());
     366              :                         }
     367              :                     }
     368              :                 }
     369              :                 None => {
     370              :                     // Our destination is not attached: maybe live migrate if some other
     371              :                     // node is currently attached.  Fall through.
     372            0 :                     node
     373              :                 }
     374              :             }
     375              :         } else {
     376              :             // No intent to be attached
     377            0 :             tracing::info!("maybe_live_migrate: no attached intent");
     378            0 :             return Ok(());
     379              :         };
     380              : 
     381            0 :         let mut origin = None;
     382            0 :         for (node_id, state) in &self.observed.locations {
     383            0 :             if let Some(observed_conf) = &state.conf {
     384            0 :                 if observed_conf.mode == LocationConfigMode::AttachedSingle {
     385              :                     // We will only attempt live migration if the origin is not offline: this
     386              :                     // avoids trying to do it while reconciling after responding to an HA failover.
     387            0 :                     if let Some(node) = self.get_node(node_id) {
     388            0 :                         if node.is_available() {
     389            0 :                             origin = Some(node.clone());
     390            0 :                             break;
     391            0 :                         }
     392            0 :                     }
     393            0 :                 }
     394            0 :             }
     395              :         }
     396              : 
     397            0 :         let Some(origin) = origin else {
     398            0 :             tracing::info!("maybe_live_migrate: no origin found");
     399            0 :             return Ok(());
     400              :         };
     401              : 
     402              :         // We have an origin and a destination: proceed to do the live migration
     403            0 :         tracing::info!("Live migrating {}->{}", origin, destination);
     404            0 :         self.live_migrate(origin, destination.clone()).await?;
     405              : 
     406            0 :         Ok(())
     407            0 :     }
     408              : 
     409            0 :     async fn wait_lsn(
     410            0 :         &self,
     411            0 :         node: &Node,
     412            0 :         tenant_shard_id: TenantShardId,
     413            0 :         timelines: HashMap<TimelineId, Lsn>,
     414            0 :     ) -> Result<StatusCode, ReconcileError> {
     415              :         const TIMEOUT: Duration = Duration::from_secs(10);
     416              : 
     417            0 :         let client = PageserverClient::new(
     418            0 :             node.get_id(),
     419            0 :             node.base_url(),
     420            0 :             self.service_config.jwt_token.as_deref(),
     421            0 :         );
     422            0 : 
     423            0 :         client
     424            0 :             .wait_lsn(
     425            0 :                 tenant_shard_id,
     426            0 :                 TenantWaitLsnRequest {
     427            0 :                     timelines,
     428            0 :                     timeout: TIMEOUT,
     429            0 :                 },
     430            0 :             )
     431            0 :             .await
     432            0 :             .map_err(|e| e.into())
     433            0 :     }
     434              : 
     435            0 :     async fn get_lsns(
     436            0 :         &self,
     437            0 :         tenant_shard_id: TenantShardId,
     438            0 :         node: &Node,
     439            0 :     ) -> anyhow::Result<HashMap<TimelineId, Lsn>> {
     440            0 :         let client = PageserverClient::new(
     441            0 :             node.get_id(),
     442            0 :             node.base_url(),
     443            0 :             self.service_config.jwt_token.as_deref(),
     444            0 :         );
     445              : 
     446            0 :         let timelines = client.timeline_list(&tenant_shard_id).await?;
     447            0 :         Ok(timelines
     448            0 :             .into_iter()
     449            0 :             .map(|t| (t.timeline_id, t.last_record_lsn))
     450            0 :             .collect())
     451            0 :     }
     452              : 
     453            0 :     async fn secondary_download(
     454            0 :         &self,
     455            0 :         tenant_shard_id: TenantShardId,
     456            0 :         node: &Node,
     457            0 :     ) -> Result<(), ReconcileError> {
     458            0 :         // This is not the timeout for a request, but the total amount of time we're willing to wait
     459            0 :         // for a secondary location to get up to date before
     460            0 :         let total_download_timeout = self.reconciler_config.get_secondary_warmup_timeout();
     461            0 : 
     462            0 :         // This the long-polling interval for the secondary download requests we send to destination pageserver
     463            0 :         // during a migration.
     464            0 :         let request_download_timeout = self
     465            0 :             .reconciler_config
     466            0 :             .get_secondary_download_request_timeout();
     467            0 : 
     468            0 :         let started_at = Instant::now();
     469              : 
     470              :         loop {
     471            0 :             let (status, progress) = match node
     472            0 :                 .with_client_retries(
     473            0 :                     |client| async move {
     474            0 :                         client
     475            0 :                             .tenant_secondary_download(
     476            0 :                                 tenant_shard_id,
     477            0 :                                 Some(request_download_timeout),
     478            0 :                             )
     479            0 :                             .await
     480            0 :                     },
     481            0 :                     &self.service_config.jwt_token,
     482            0 :                     1,
     483            0 :                     3,
     484            0 :                     request_download_timeout * 2,
     485            0 :                     &self.cancel,
     486            0 :                 )
     487            0 :                 .await
     488              :             {
     489            0 :                 None => Err(ReconcileError::Cancel),
     490            0 :                 Some(Ok(v)) => Ok(v),
     491            0 :                 Some(Err(e)) => {
     492            0 :                     // Give up, but proceed: it's unfortunate if we couldn't freshen the destination before
     493            0 :                     // attaching, but we should not let an issue with a secondary location stop us proceeding
     494            0 :                     // with a live migration.
     495            0 :                     tracing::warn!("Failed to prepare by downloading layers on node {node}: {e})");
     496            0 :                     return Ok(());
     497              :                 }
     498            0 :             }?;
     499              : 
     500            0 :             if status == StatusCode::OK {
     501            0 :                 tracing::info!(
     502            0 :                     "Downloads to {} complete: {}/{} layers, {}/{} bytes",
     503              :                     node,
     504              :                     progress.layers_downloaded,
     505              :                     progress.layers_total,
     506              :                     progress.bytes_downloaded,
     507              :                     progress.bytes_total
     508              :                 );
     509            0 :                 return Ok(());
     510            0 :             } else if status == StatusCode::ACCEPTED {
     511            0 :                 let total_runtime = started_at.elapsed();
     512            0 :                 if total_runtime > total_download_timeout {
     513            0 :                     tracing::warn!("Timed out after {}ms downloading layers to {node}.  Progress so far: {}/{} layers, {}/{} bytes",
     514            0 :                         total_runtime.as_millis(),
     515              :                         progress.layers_downloaded,
     516              :                         progress.layers_total,
     517              :                         progress.bytes_downloaded,
     518              :                         progress.bytes_total
     519              :                     );
     520              :                     // Give up, but proceed: an incompletely warmed destination doesn't prevent migration working,
     521              :                     // it just makes the I/O performance for users less good.
     522            0 :                     return Ok(());
     523            0 :                 }
     524            0 : 
     525            0 :                 // Log and proceed around the loop to retry.  We don't sleep between requests, because our HTTP call
     526            0 :                 // to the pageserver is a long-poll.
     527            0 :                 tracing::info!(
     528            0 :                     "Downloads to {} not yet complete: {}/{} layers, {}/{} bytes",
     529              :                     node,
     530              :                     progress.layers_downloaded,
     531              :                     progress.layers_total,
     532              :                     progress.bytes_downloaded,
     533              :                     progress.bytes_total
     534              :                 );
     535            0 :             }
     536              :         }
     537            0 :     }
     538              : 
     539              :     /// This function does _not_ mutate any state, so it is cancellation safe.
     540              :     ///
     541              :     /// This function does not respect [`Self::cancel`], callers should handle that.
     542            0 :     async fn await_lsn(
     543            0 :         &self,
     544            0 :         tenant_shard_id: TenantShardId,
     545            0 :         node: &Node,
     546            0 :         baseline: HashMap<TimelineId, Lsn>,
     547            0 :     ) -> anyhow::Result<()> {
     548              :         // Signal to the pageserver that it should ingest up to the baseline LSNs.
     549              :         loop {
     550            0 :             match self.wait_lsn(node, tenant_shard_id, baseline.clone()).await {
     551              :                 Ok(StatusCode::OK) => {
     552              :                     // Everything is caught up
     553            0 :                     return Ok(());
     554              :                 }
     555              :                 Ok(StatusCode::ACCEPTED) => {
     556              :                     // Some timelines are not caught up yet.
     557              :                     // They'll be polled below.
     558            0 :                     break;
     559              :                 }
     560              :                 Ok(StatusCode::NOT_FOUND) => {
     561              :                     // None of the timelines are present on the pageserver.
     562              :                     // This is correct if they've all been deleted, but
     563              :                     // let let the polling loop below cross check.
     564            0 :                     break;
     565              :                 }
     566            0 :                 Ok(status_code) => {
     567            0 :                     tracing::warn!(
     568            0 :                         "Unexpected status code ({status_code}) returned by wait_lsn endpoint"
     569              :                     );
     570            0 :                     break;
     571              :                 }
     572            0 :                 Err(e) => {
     573            0 :                     tracing::info!("🕑 Can't trigger LSN wait on {node} yet, waiting ({e})",);
     574            0 :                     tokio::time::sleep(Duration::from_millis(500)).await;
     575            0 :                     continue;
     576              :                 }
     577              :             }
     578              :         }
     579              : 
     580              :         // Poll the LSNs until they catch up
     581              :         loop {
     582            0 :             let latest = match self.get_lsns(tenant_shard_id, node).await {
     583            0 :                 Ok(l) => l,
     584            0 :                 Err(e) => {
     585            0 :                     tracing::info!("🕑 Can't get LSNs on node {node} yet, waiting ({e})",);
     586            0 :                     tokio::time::sleep(Duration::from_millis(500)).await;
     587            0 :                     continue;
     588              :                 }
     589              :             };
     590              : 
     591            0 :             let mut any_behind: bool = false;
     592            0 :             for (timeline_id, baseline_lsn) in &baseline {
     593            0 :                 match latest.get(timeline_id) {
     594            0 :                     Some(latest_lsn) => {
     595            0 :                         tracing::info!(timeline_id = %timeline_id, "🕑 LSN origin {baseline_lsn} vs destination {latest_lsn}");
     596            0 :                         if latest_lsn < baseline_lsn {
     597            0 :                             any_behind = true;
     598            0 :                         }
     599              :                     }
     600            0 :                     None => {
     601            0 :                         // Timeline was deleted in the meantime - ignore it
     602            0 :                     }
     603              :                 }
     604              :             }
     605              : 
     606            0 :             if !any_behind {
     607            0 :                 tracing::info!("✅ LSN caught up.  Proceeding...");
     608            0 :                 break;
     609              :             } else {
     610            0 :                 tokio::time::sleep(Duration::from_millis(500)).await;
     611              :             }
     612              :         }
     613              : 
     614            0 :         Ok(())
     615            0 :     }
     616              : 
     617            0 :     pub async fn live_migrate(
     618            0 :         &mut self,
     619            0 :         origin_ps: Node,
     620            0 :         dest_ps: Node,
     621            0 :     ) -> Result<(), ReconcileError> {
     622            0 :         // `maybe_live_migrate` is responsibble for sanity of inputs
     623            0 :         assert!(origin_ps.get_id() != dest_ps.get_id());
     624              : 
     625            0 :         fn build_location_config(
     626            0 :             shard: &ShardIdentity,
     627            0 :             config: &TenantConfig,
     628            0 :             mode: LocationConfigMode,
     629            0 :             generation: Option<Generation>,
     630            0 :             secondary_conf: Option<LocationConfigSecondary>,
     631            0 :         ) -> LocationConfig {
     632            0 :             LocationConfig {
     633            0 :                 mode,
     634            0 :                 generation: generation.map(|g| g.into().unwrap()),
     635            0 :                 secondary_conf,
     636            0 :                 tenant_conf: config.clone(),
     637            0 :                 shard_number: shard.number.0,
     638            0 :                 shard_count: shard.count.literal(),
     639            0 :                 shard_stripe_size: shard.stripe_size.0,
     640            0 :             }
     641            0 :         }
     642              : 
     643            0 :         tracing::info!("🔁 Switching origin node {origin_ps} to stale mode",);
     644              : 
     645              :         // FIXME: it is incorrect to use self.generation here, we should use the generation
     646              :         // from the ObservedState of the origin pageserver (it might be older than self.generation)
     647            0 :         let stale_conf = build_location_config(
     648            0 :             &self.shard,
     649            0 :             &self.config,
     650            0 :             LocationConfigMode::AttachedStale,
     651            0 :             self.generation,
     652            0 :             None,
     653            0 :         );
     654            0 :         self.location_config(&origin_ps, stale_conf, Some(Duration::from_secs(10)), false)
     655            0 :             .await?;
     656              : 
     657            0 :         let baseline_lsns = Some(self.get_lsns(self.tenant_shard_id, &origin_ps).await?);
     658              : 
     659              :         // If we are migrating to a destination that has a secondary location, warm it up first
     660            0 :         if let Some(destination_conf) = self.observed.locations.get(&dest_ps.get_id()) {
     661            0 :             if let Some(destination_conf) = &destination_conf.conf {
     662            0 :                 if destination_conf.mode == LocationConfigMode::Secondary {
     663            0 :                     tracing::info!("🔁 Downloading latest layers to destination node {dest_ps}",);
     664            0 :                     self.secondary_download(self.tenant_shard_id, &dest_ps)
     665            0 :                         .await?;
     666            0 :                 }
     667            0 :             }
     668            0 :         }
     669              : 
     670            0 :         pausable_failpoint!("reconciler-live-migrate-pre-generation-inc");
     671              : 
     672              :         // Increment generation before attaching to new pageserver
     673              :         self.generation = Some(
     674            0 :             self.persistence
     675            0 :                 .increment_generation(self.tenant_shard_id, dest_ps.get_id())
     676            0 :                 .await?,
     677              :         );
     678              : 
     679            0 :         let dest_conf = build_location_config(
     680            0 :             &self.shard,
     681            0 :             &self.config,
     682            0 :             LocationConfigMode::AttachedMulti,
     683            0 :             self.generation,
     684            0 :             None,
     685            0 :         );
     686            0 : 
     687            0 :         tracing::info!("🔁 Attaching to pageserver {dest_ps}");
     688            0 :         self.location_config(&dest_ps, dest_conf, None, false)
     689            0 :             .await?;
     690              : 
     691            0 :         pausable_failpoint!("reconciler-live-migrate-pre-await-lsn");
     692              : 
     693            0 :         if let Some(baseline) = baseline_lsns {
     694            0 :             tracing::info!("🕑 Waiting for LSN to catch up...");
     695            0 :             tokio::select! {
     696            0 :                 r = self.await_lsn(self.tenant_shard_id, &dest_ps, baseline) => {r?;}
     697            0 :                 _ = self.cancel.cancelled() => {return Err(ReconcileError::Cancel)}
     698              :             };
     699            0 :         }
     700              : 
     701            0 :         tracing::info!("🔁 Notifying compute to use pageserver {dest_ps}");
     702              : 
     703              :         // During a live migration it is unhelpful to proceed if we couldn't notify compute: if we detach
     704              :         // the origin without notifying compute, we will render the tenant unavailable.
     705            0 :         self.compute_notify_blocking(&origin_ps).await?;
     706            0 :         pausable_failpoint!("reconciler-live-migrate-post-notify");
     707              : 
     708              :         // Downgrade the origin to secondary.  If the tenant's policy is PlacementPolicy::Attached(0), then
     709              :         // this location will be deleted in the general case reconciliation that runs after this.
     710            0 :         let origin_secondary_conf = build_location_config(
     711            0 :             &self.shard,
     712            0 :             &self.config,
     713            0 :             LocationConfigMode::Secondary,
     714            0 :             None,
     715            0 :             Some(LocationConfigSecondary { warm: true }),
     716            0 :         );
     717            0 :         self.location_config(&origin_ps, origin_secondary_conf.clone(), None, false)
     718            0 :             .await?;
     719              :         // TODO: we should also be setting the ObservedState on earlier API calls, in case we fail
     720              :         // partway through.  In fact, all location conf API calls should be in a wrapper that sets
     721              :         // the observed state to None, then runs, then sets it to what we wrote.
     722            0 :         self.observed.locations.insert(
     723            0 :             origin_ps.get_id(),
     724            0 :             ObservedStateLocation {
     725            0 :                 conf: Some(origin_secondary_conf),
     726            0 :             },
     727            0 :         );
     728            0 : 
     729            0 :         pausable_failpoint!("reconciler-live-migrate-post-detach");
     730              : 
     731            0 :         tracing::info!("🔁 Switching to AttachedSingle mode on node {dest_ps}",);
     732            0 :         let dest_final_conf = build_location_config(
     733            0 :             &self.shard,
     734            0 :             &self.config,
     735            0 :             LocationConfigMode::AttachedSingle,
     736            0 :             self.generation,
     737            0 :             None,
     738            0 :         );
     739            0 :         self.location_config(&dest_ps, dest_final_conf.clone(), None, false)
     740            0 :             .await?;
     741            0 :         self.observed.locations.insert(
     742            0 :             dest_ps.get_id(),
     743            0 :             ObservedStateLocation {
     744            0 :                 conf: Some(dest_final_conf),
     745            0 :             },
     746            0 :         );
     747            0 : 
     748            0 :         tracing::info!("✅ Migration complete");
     749              : 
     750            0 :         Ok(())
     751            0 :     }
     752              : 
     753            0 :     async fn maybe_refresh_observed(&mut self) -> Result<(), ReconcileError> {
     754              :         // If the attached node has uncertain state, read it from the pageserver before proceeding: this
     755              :         // is important to avoid spurious generation increments.
     756              :         //
     757              :         // We don't need to do this for secondary/detach locations because it's harmless to just PUT their
     758              :         // location conf, whereas for attached locations it can interrupt clients if we spuriously destroy/recreate
     759              :         // the `Timeline` object in the pageserver.
     760              : 
     761            0 :         let Some(attached_node) = self.intent.attached.as_ref() else {
     762              :             // Nothing to do
     763            0 :             return Ok(());
     764              :         };
     765              : 
     766            0 :         if matches!(
     767            0 :             self.observed.locations.get(&attached_node.get_id()),
     768              :             Some(ObservedStateLocation { conf: None })
     769              :         ) {
     770            0 :             let tenant_shard_id = self.tenant_shard_id;
     771            0 :             let observed_conf = match attached_node
     772            0 :                 .with_client_retries(
     773            0 :                     |client| async move { client.get_location_config(tenant_shard_id).await },
     774            0 :                     &self.service_config.jwt_token,
     775            0 :                     1,
     776            0 :                     1,
     777            0 :                     Duration::from_secs(5),
     778            0 :                     &self.cancel,
     779            0 :                 )
     780            0 :                 .await
     781              :             {
     782            0 :                 Some(Ok(observed)) => Some(observed),
     783            0 :                 Some(Err(mgmt_api::Error::ApiError(status, _msg)))
     784            0 :                     if status == StatusCode::NOT_FOUND =>
     785            0 :                 {
     786            0 :                     None
     787              :                 }
     788            0 :                 Some(Err(e)) => return Err(e.into()),
     789            0 :                 None => return Err(ReconcileError::Cancel),
     790              :             };
     791            0 :             tracing::info!("Scanned location configuration on {attached_node}: {observed_conf:?}");
     792            0 :             match observed_conf {
     793            0 :                 Some(conf) => {
     794            0 :                     // Pageserver returned a state: update it in observed.  This may still be an indeterminate (None) state,
     795            0 :                     // if internally the pageserver's TenantSlot was being mutated (e.g. some long running API call is still running)
     796            0 :                     self.observed
     797            0 :                         .locations
     798            0 :                         .insert(attached_node.get_id(), ObservedStateLocation { conf });
     799            0 :                 }
     800            0 :                 None => {
     801            0 :                     // Pageserver returned 404: we have confirmation that there is no state for this shard on that pageserver.
     802            0 :                     self.observed.locations.remove(&attached_node.get_id());
     803            0 :                 }
     804              :             }
     805            0 :         }
     806              : 
     807            0 :         Ok(())
     808            0 :     }
     809              : 
     810              :     /// Reconciling a tenant makes API calls to pageservers until the observed state
     811              :     /// matches the intended state.
     812              :     ///
     813              :     /// First we apply special case handling (e.g. for live migrations), and then a
     814              :     /// general case reconciliation where we walk through the intent by pageserver
     815              :     /// and call out to the pageserver to apply the desired state.
     816              :     ///
     817              :     /// An Ok(()) result indicates that we successfully attached the tenant, but _not_ that
     818              :     /// all locations for the tenant are in the expected state. When nodes that are to be detached
     819              :     /// or configured as secondary are unavailable, we may return Ok(()) but leave the shard in a
     820              :     /// state where it still requires later reconciliation.
     821            0 :     pub(crate) async fn reconcile(&mut self) -> Result<(), ReconcileError> {
     822            0 :         // Prepare: if we have uncertain `observed` state for our would-be attachement location, then refresh it
     823            0 :         self.maybe_refresh_observed().await?;
     824              : 
     825              :         // Special case: live migration
     826            0 :         self.maybe_live_migrate().await?;
     827              : 
     828              :         // If the attached pageserver is not attached, do so now.
     829            0 :         if let Some(node) = self.intent.attached.as_ref() {
     830              :             // If we are in an attached policy, then generation must have been set (null generations
     831              :             // are only present when a tenant is initially loaded with a secondary policy)
     832            0 :             debug_assert!(self.generation.is_some());
     833            0 :             let Some(generation) = self.generation else {
     834            0 :                 return Err(ReconcileError::Other(anyhow::anyhow!(
     835            0 :                     "Attempted to attach with NULL generation"
     836            0 :                 )));
     837              :             };
     838              : 
     839            0 :             let mut wanted_conf = attached_location_conf(
     840            0 :                 generation,
     841            0 :                 &self.shard,
     842            0 :                 &self.config,
     843            0 :                 &self.placement_policy,
     844            0 :             );
     845            0 :             match self.observed.locations.get(&node.get_id()) {
     846            0 :                 Some(conf) if conf.conf.as_ref() == Some(&wanted_conf) => {
     847            0 :                     // Nothing to do
     848            0 :                     tracing::info!(node_id=%node.get_id(), "Observed configuration already correct.")
     849              :                 }
     850            0 :                 observed => {
     851              :                     // In all cases other than a matching observed configuration, we will
     852              :                     // reconcile this location.  This includes locations with different configurations, as well
     853              :                     // as locations with unknown (None) observed state.
     854              : 
     855              :                     // Incrementing generation is the safe general case, but is inefficient for changes that only
     856              :                     // modify some details (e.g. the tenant's config).
     857            0 :                     let increment_generation = match observed {
     858            0 :                         None => true,
     859            0 :                         Some(ObservedStateLocation { conf: None }) => true,
     860              :                         Some(ObservedStateLocation {
     861            0 :                             conf: Some(observed),
     862            0 :                         }) => {
     863            0 :                             let generations_match = observed.generation == wanted_conf.generation;
     864            0 : 
     865            0 :                             // We may skip incrementing the generation if the location is already in the expected mode and
     866            0 :                             // generation.  In principle it would also be safe to skip from certain other modes (e.g. AttachedStale),
     867            0 :                             // but such states are handled inside `live_migrate`, and if we see that state here we're cleaning up
     868            0 :                             // after a restart/crash, so fall back to the universally safe path of incrementing generation.
     869            0 :                             !generations_match || (observed.mode != wanted_conf.mode)
     870              :                         }
     871              :                     };
     872              : 
     873            0 :                     if increment_generation {
     874            0 :                         pausable_failpoint!("reconciler-pre-increment-generation");
     875              : 
     876            0 :                         let generation = self
     877            0 :                             .persistence
     878            0 :                             .increment_generation(self.tenant_shard_id, node.get_id())
     879            0 :                             .await?;
     880            0 :                         self.generation = Some(generation);
     881            0 :                         wanted_conf.generation = generation.into();
     882            0 :                     }
     883            0 :                     tracing::info!(node_id=%node.get_id(), "Observed configuration requires update.");
     884              : 
     885              :                     // Because `node` comes from a ref to &self, clone it before calling into a &mut self
     886              :                     // function: this could be avoided by refactoring the state mutated by location_config into
     887              :                     // a separate type to Self.
     888            0 :                     let node = node.clone();
     889            0 : 
     890            0 :                     // Use lazy=true, because we may run many of Self concurrently, and do not want to
     891            0 :                     // overload the pageserver with logical size calculations.
     892            0 :                     self.location_config(&node, wanted_conf, None, true).await?;
     893            0 :                     self.compute_notify().await?;
     894              :                 }
     895              :             }
     896            0 :         }
     897              : 
     898              :         // Configure secondary locations: if these were previously attached this
     899              :         // implicitly downgrades them from attached to secondary.
     900            0 :         let mut changes = Vec::new();
     901            0 :         for node in &self.intent.secondary {
     902            0 :             let wanted_conf = secondary_location_conf(&self.shard, &self.config);
     903            0 :             match self.observed.locations.get(&node.get_id()) {
     904            0 :                 Some(conf) if conf.conf.as_ref() == Some(&wanted_conf) => {
     905            0 :                     // Nothing to do
     906            0 :                     tracing::info!(node_id=%node.get_id(), "Observed configuration already correct.")
     907              :                 }
     908              :                 _ => {
     909              :                     // Only try and configure secondary locations on nodes that are available.  This
     910              :                     // allows the reconciler to "succeed" while some secondaries are offline (e.g. after
     911              :                     // a node failure, where the failed node will have a secondary intent)
     912            0 :                     if node.is_available() {
     913            0 :                         tracing::info!(node_id=%node.get_id(), "Observed configuration requires update.");
     914            0 :                         changes.push((node.clone(), wanted_conf))
     915              :                     } else {
     916            0 :                         tracing::info!(node_id=%node.get_id(), "Skipping configuration as secondary, node is unavailable");
     917            0 :                         self.observed
     918            0 :                             .locations
     919            0 :                             .insert(node.get_id(), ObservedStateLocation { conf: None });
     920              :                     }
     921              :                 }
     922              :             }
     923              :         }
     924              : 
     925              :         // Detach any extraneous pageservers that are no longer referenced
     926              :         // by our intent.
     927            0 :         for node in &self.detach {
     928            0 :             changes.push((
     929            0 :                 node.clone(),
     930            0 :                 LocationConfig {
     931            0 :                     mode: LocationConfigMode::Detached,
     932            0 :                     generation: None,
     933            0 :                     secondary_conf: None,
     934            0 :                     shard_number: self.shard.number.0,
     935            0 :                     shard_count: self.shard.count.literal(),
     936            0 :                     shard_stripe_size: self.shard.stripe_size.0,
     937            0 :                     tenant_conf: self.config.clone(),
     938            0 :                 },
     939            0 :             ));
     940            0 :         }
     941              : 
     942            0 :         for (node, conf) in changes {
     943            0 :             if self.cancel.is_cancelled() {
     944            0 :                 return Err(ReconcileError::Cancel);
     945            0 :             }
     946            0 :             // We only try to configure secondary locations if the node is available.  This does
     947            0 :             // not stop us succeeding with the reconcile, because our core goal is to make the
     948            0 :             // shard _available_ (the attached location), and configuring secondary locations
     949            0 :             // can be done lazily when the node becomes available (via background reconciliation).
     950            0 :             if node.is_available() {
     951            0 :                 self.location_config(&node, conf, None, false).await?;
     952              :             } else {
     953              :                 // If the node is unavailable, we skip and consider the reconciliation successful: this
     954              :                 // is a common case where a pageserver is marked unavailable: we demote a location on
     955              :                 // that unavailable pageserver to secondary.
     956            0 :                 tracing::info!("Skipping configuring secondary location {node}, it is unavailable");
     957            0 :                 self.observed
     958            0 :                     .locations
     959            0 :                     .insert(node.get_id(), ObservedStateLocation { conf: None });
     960              :             }
     961              :         }
     962              : 
     963              :         // The condition below identifies a detach. We must have no attached intent and
     964              :         // must have been attached to something previously. Pass this information to
     965              :         // the [`ComputeHook`] such that it can update its tenant-wide state.
     966            0 :         if self.intent.attached.is_none() && !self.detach.is_empty() {
     967            0 :             // TODO: Consider notifying control plane about detaches. This would avoid situations
     968            0 :             // where the compute tries to start-up with a stale set of pageservers.
     969            0 :             self.compute_hook
     970            0 :                 .handle_detach(self.tenant_shard_id, self.shard.stripe_size);
     971            0 :         }
     972              : 
     973            0 :         pausable_failpoint!("reconciler-epilogue");
     974              : 
     975            0 :         Ok(())
     976            0 :     }
     977              : 
     978            0 :     pub(crate) async fn compute_notify(&mut self) -> Result<(), NotifyError> {
     979              :         // Whenever a particular Reconciler emits a notification, it is always notifying for the intended
     980              :         // destination.
     981            0 :         if let Some(node) = &self.intent.attached {
     982            0 :             let result = self
     983            0 :                 .compute_hook
     984            0 :                 .notify(
     985            0 :                     compute_hook::ShardUpdate {
     986            0 :                         tenant_shard_id: self.tenant_shard_id,
     987            0 :                         node_id: node.get_id(),
     988            0 :                         stripe_size: self.shard.stripe_size,
     989            0 :                         preferred_az: self.preferred_az.as_ref().map(Cow::Borrowed),
     990            0 :                     },
     991            0 :                     &self.cancel,
     992            0 :                 )
     993            0 :                 .await;
     994            0 :             if let Err(e) = &result {
     995              :                 // Set this flag so that in our ReconcileResult we will set the flag on the shard that it
     996              :                 // needs to retry at some point.
     997            0 :                 self.compute_notify_failure = true;
     998              : 
     999              :                 // It is up to the caller whether they want to drop out on this error, but they don't have to:
    1000              :                 // in general we should avoid letting unavailability of the cloud control plane stop us from
    1001              :                 // making progress.
    1002            0 :                 match e {
    1003            0 :                     // 404s from cplane during tenant creation are expected.
    1004            0 :                     // Cplane only persists the shards to the database after
    1005            0 :                     // creating the tenant and the timeline. If we notify before
    1006            0 :                     // that, we'll get a 404.
    1007            0 :                     //
    1008            0 :                     // This is fine because tenant creations happen via /location_config
    1009            0 :                     // and that returns the list of locations in the response. Hence, we
    1010            0 :                     // silence the error and return Ok(()) here. Reconciliation will still
    1011            0 :                     // be retried because we set [`Reconciler::compute_notify_failure`] above.
    1012            0 :                     NotifyError::Unexpected(hyper::StatusCode::NOT_FOUND)
    1013            0 :                         if self.reconciler_config.tenant_creation_hint() =>
    1014            0 :                     {
    1015            0 :                         return Ok(());
    1016              :                     }
    1017            0 :                     NotifyError::ShuttingDown => {}
    1018              :                     _ => {
    1019            0 :                         tracing::warn!(
    1020            0 :                             "Failed to notify compute of attached pageserver {node}: {e}"
    1021              :                         );
    1022              :                     }
    1023              :                 }
    1024            0 :             }
    1025            0 :             result
    1026              :         } else {
    1027            0 :             Ok(())
    1028              :         }
    1029            0 :     }
    1030              : 
    1031              :     /// Compare the observed state snapshot from when the reconcile was created
    1032              :     /// with the final observed state in order to generate observed state deltas.
    1033            0 :     pub(crate) fn observed_deltas(&self) -> Vec<ObservedStateDelta> {
    1034            0 :         let mut deltas = Vec::default();
    1035              : 
    1036            0 :         for (node_id, location) in &self.observed.locations {
    1037            0 :             let previous_location = self.original_observed.locations.get(node_id);
    1038            0 :             let do_upsert = match previous_location {
    1039              :                 // Location config changed for node
    1040            0 :                 Some(prev) if location.conf != prev.conf => true,
    1041              :                 // New location config for node
    1042            0 :                 None => true,
    1043              :                 // Location config has not changed for node
    1044            0 :                 _ => false,
    1045              :             };
    1046              : 
    1047            0 :             if do_upsert {
    1048            0 :                 deltas.push(ObservedStateDelta::Upsert(Box::new((
    1049            0 :                     *node_id,
    1050            0 :                     location.clone(),
    1051            0 :                 ))));
    1052            0 :             }
    1053              :         }
    1054              : 
    1055            0 :         for node_id in self.original_observed.locations.keys() {
    1056            0 :             if !self.observed.locations.contains_key(node_id) {
    1057            0 :                 deltas.push(ObservedStateDelta::Delete(*node_id));
    1058            0 :             }
    1059              :         }
    1060              : 
    1061            0 :         deltas
    1062            0 :     }
    1063              : 
    1064              :     /// Keep trying to notify the compute indefinitely, only dropping out if:
    1065              :     /// - the node `origin` becomes unavailable -> Ok(())
    1066              :     /// - the node `origin` no longer has our tenant shard attached -> Ok(())
    1067              :     /// - our cancellation token fires -> Err(ReconcileError::Cancelled)
    1068              :     ///
    1069              :     /// This is used during live migration, where we do not wish to detach
    1070              :     /// an origin location until the compute definitely knows about the new
    1071              :     /// location.
    1072              :     ///
    1073              :     /// In cases where the origin node becomes unavailable, we return success, indicating
    1074              :     /// to the caller that they should continue irrespective of whether the compute was notified,
    1075              :     /// because the origin node is unusable anyway.  Notification will be retried later via the
    1076              :     /// [`Self::compute_notify_failure`] flag.
    1077            0 :     async fn compute_notify_blocking(&mut self, origin: &Node) -> Result<(), ReconcileError> {
    1078            0 :         let mut notify_attempts = 0;
    1079            0 :         while let Err(e) = self.compute_notify().await {
    1080            0 :             match e {
    1081            0 :                 NotifyError::Fatal(_) => return Err(ReconcileError::Notify(e)),
    1082            0 :                 NotifyError::ShuttingDown => return Err(ReconcileError::Cancel),
    1083              :                 _ => {
    1084            0 :                     tracing::warn!(
    1085            0 :                         "Live migration blocked by compute notification error, retrying: {e}"
    1086              :                     );
    1087              :                 }
    1088              :             }
    1089              : 
    1090              :             // Did the origin pageserver become unavailable?
    1091            0 :             if !origin.is_available() {
    1092            0 :                 tracing::info!("Giving up on compute notification because {origin} is unavailable");
    1093            0 :                 break;
    1094            0 :             }
    1095            0 : 
    1096            0 :             // Does the origin pageserver still host the shard we are interested in?  We should only
    1097            0 :             // continue waiting for compute notification to be acked if the old location is still usable.
    1098            0 :             let tenant_shard_id = self.tenant_shard_id;
    1099            0 :             match origin
    1100            0 :                 .with_client_retries(
    1101            0 :                     |client| async move { client.get_location_config(tenant_shard_id).await },
    1102            0 :                     &self.service_config.jwt_token,
    1103            0 :                     1,
    1104            0 :                     3,
    1105            0 :                     Duration::from_secs(5),
    1106            0 :                     &self.cancel,
    1107            0 :                 )
    1108            0 :                 .await
    1109              :             {
    1110            0 :                 Some(Ok(Some(location_conf))) => {
    1111            0 :                     if matches!(
    1112            0 :                         location_conf.mode,
    1113              :                         LocationConfigMode::AttachedMulti
    1114              :                             | LocationConfigMode::AttachedSingle
    1115              :                             | LocationConfigMode::AttachedStale
    1116              :                     ) {
    1117            0 :                         tracing::debug!(
    1118            0 :                             "Still attached to {origin}, will wait & retry compute notification"
    1119              :                         );
    1120              :                     } else {
    1121            0 :                         tracing::info!(
    1122            0 :                             "Giving up on compute notification because {origin} is in state {:?}",
    1123              :                             location_conf.mode
    1124              :                         );
    1125            0 :                         return Ok(());
    1126              :                     }
    1127              :                     // Fall through
    1128              :                 }
    1129              :                 Some(Ok(None)) => {
    1130            0 :                     tracing::info!(
    1131            0 :                         "No longer attached to {origin}, giving up on compute notification"
    1132              :                     );
    1133            0 :                     return Ok(());
    1134              :                 }
    1135            0 :                 Some(Err(e)) => {
    1136            0 :                     match e {
    1137              :                         mgmt_api::Error::Cancelled => {
    1138            0 :                             tracing::info!(
    1139            0 :                                 "Giving up on compute notification because {origin} is unavailable"
    1140              :                             );
    1141            0 :                             return Ok(());
    1142              :                         }
    1143              :                         mgmt_api::Error::ApiError(StatusCode::NOT_FOUND, _) => {
    1144            0 :                             tracing::info!(
    1145            0 :                                 "No longer attached to {origin}, giving up on compute notification"
    1146              :                             );
    1147            0 :                             return Ok(());
    1148              :                         }
    1149            0 :                         e => {
    1150            0 :                             // Other API errors are unexpected here.
    1151            0 :                             tracing::warn!("Unexpected error checking location on {origin}: {e}");
    1152              : 
    1153              :                             // Fall through, we will retry compute notification.
    1154              :                         }
    1155              :                     }
    1156              :                 }
    1157            0 :                 None => return Err(ReconcileError::Cancel),
    1158              :             };
    1159              : 
    1160            0 :             exponential_backoff(
    1161            0 :                 notify_attempts,
    1162            0 :                 // Generous waits: control plane operations which might be blocking us usually complete on the order
    1163            0 :                 // of hundreds to thousands of milliseconds, so no point busy polling.
    1164            0 :                 1.0,
    1165            0 :                 10.0,
    1166            0 :                 &self.cancel,
    1167            0 :             )
    1168            0 :             .await;
    1169            0 :             notify_attempts += 1;
    1170              :         }
    1171              : 
    1172            0 :         Ok(())
    1173            0 :     }
    1174              : }
    1175              : 
    1176              : /// We tweak the externally-set TenantConfig while configuring
    1177              : /// locations, using our awareness of whether secondary locations
    1178              : /// are in use to automatically enable/disable heatmap uploads.
    1179            0 : fn ha_aware_config(config: &TenantConfig, has_secondaries: bool) -> TenantConfig {
    1180            0 :     let mut config = config.clone();
    1181            0 :     if has_secondaries {
    1182            0 :         if config.heatmap_period.is_none() {
    1183            0 :             config.heatmap_period = Some(DEFAULT_HEATMAP_PERIOD.to_string());
    1184            0 :         }
    1185            0 :     } else {
    1186            0 :         config.heatmap_period = None;
    1187            0 :     }
    1188            0 :     config
    1189            0 : }
    1190              : 
    1191            0 : pub(crate) fn attached_location_conf(
    1192            0 :     generation: Generation,
    1193            0 :     shard: &ShardIdentity,
    1194            0 :     config: &TenantConfig,
    1195            0 :     policy: &PlacementPolicy,
    1196            0 : ) -> LocationConfig {
    1197            0 :     let has_secondaries = match policy {
    1198              :         PlacementPolicy::Attached(0) | PlacementPolicy::Detached | PlacementPolicy::Secondary => {
    1199            0 :             false
    1200              :         }
    1201            0 :         PlacementPolicy::Attached(_) => true,
    1202              :     };
    1203              : 
    1204            0 :     LocationConfig {
    1205            0 :         mode: LocationConfigMode::AttachedSingle,
    1206            0 :         generation: generation.into(),
    1207            0 :         secondary_conf: None,
    1208            0 :         shard_number: shard.number.0,
    1209            0 :         shard_count: shard.count.literal(),
    1210            0 :         shard_stripe_size: shard.stripe_size.0,
    1211            0 :         tenant_conf: ha_aware_config(config, has_secondaries),
    1212            0 :     }
    1213            0 : }
    1214              : 
    1215            0 : pub(crate) fn secondary_location_conf(
    1216            0 :     shard: &ShardIdentity,
    1217            0 :     config: &TenantConfig,
    1218            0 : ) -> LocationConfig {
    1219            0 :     LocationConfig {
    1220            0 :         mode: LocationConfigMode::Secondary,
    1221            0 :         generation: None,
    1222            0 :         secondary_conf: Some(LocationConfigSecondary { warm: true }),
    1223            0 :         shard_number: shard.number.0,
    1224            0 :         shard_count: shard.count.literal(),
    1225            0 :         shard_stripe_size: shard.stripe_size.0,
    1226            0 :         tenant_conf: ha_aware_config(config, true),
    1227            0 :     }
    1228            0 : }
        

Generated by: LCOV version 2.1-beta