LCOV - code coverage report
Current view: top level - libs/pageserver_api/src - models.rs (source / functions) Coverage Total Hit
Test: 5fe7fa8d483b39476409aee736d6d5e32728bfac.info Lines: 57.3 % 1114 638
Test Date: 2025-03-12 16:10:49 Functions: 7.4 % 1000 74

            Line data    Source code
       1              : pub mod detach_ancestor;
       2              : pub mod partitioning;
       3              : pub mod utilization;
       4              : 
       5              : use core::ops::Range;
       6              : use std::collections::HashMap;
       7              : use std::fmt::Display;
       8              : use std::io::{BufRead, Read};
       9              : use std::num::{NonZeroU32, NonZeroU64, NonZeroUsize};
      10              : use std::str::FromStr;
      11              : use std::time::{Duration, SystemTime};
      12              : 
      13              : use byteorder::{BigEndian, ReadBytesExt};
      14              : use bytes::{Buf, BufMut, Bytes, BytesMut};
      15              : #[cfg(feature = "testing")]
      16              : use camino::Utf8PathBuf;
      17              : use postgres_ffi::BLCKSZ;
      18              : use serde::{Deserialize, Deserializer, Serialize, Serializer};
      19              : use serde_with::serde_as;
      20              : pub use utilization::PageserverUtilization;
      21              : use utils::id::{NodeId, TenantId, TimelineId};
      22              : use utils::lsn::Lsn;
      23              : use utils::postgres_client::PostgresClientProtocol;
      24              : use utils::{completion, serde_system_time};
      25              : 
      26              : use crate::key::{CompactKey, Key};
      27              : use crate::reltag::RelTag;
      28              : use crate::shard::{ShardCount, ShardStripeSize, TenantShardId};
      29              : 
      30              : /// The state of a tenant in this pageserver.
      31              : ///
      32              : /// ```mermaid
      33              : /// stateDiagram-v2
      34              : ///
      35              : ///     [*] --> Attaching: spawn_attach()
      36              : ///
      37              : ///     Attaching --> Activating: activate()
      38              : ///     Activating --> Active: infallible
      39              : ///
      40              : ///     Attaching --> Broken: attach() failure
      41              : ///
      42              : ///     Active --> Stopping: set_stopping(), part of shutdown & detach
      43              : ///     Stopping --> Broken: late error in remove_tenant_from_memory
      44              : ///
      45              : ///     Broken --> [*]: ignore / detach / shutdown
      46              : ///     Stopping --> [*]: remove_from_memory complete
      47              : ///
      48              : ///     Active --> Broken: cfg(testing)-only tenant break point
      49              : /// ```
      50              : #[derive(
      51              :     Clone,
      52              :     PartialEq,
      53              :     Eq,
      54            0 :     serde::Serialize,
      55            1 :     serde::Deserialize,
      56              :     strum_macros::Display,
      57              :     strum_macros::VariantNames,
      58              :     strum_macros::AsRefStr,
      59              :     strum_macros::IntoStaticStr,
      60              : )]
      61              : #[serde(tag = "slug", content = "data")]
      62              : pub enum TenantState {
      63              :     /// This tenant is being attached to the pageserver.
      64              :     ///
      65              :     /// `set_stopping()` and `set_broken()` do not work in this state and wait for it to pass.
      66              :     Attaching,
      67              :     /// The tenant is transitioning from Loading/Attaching to Active.
      68              :     ///
      69              :     /// While in this state, the individual timelines are being activated.
      70              :     ///
      71              :     /// `set_stopping()` and `set_broken()` do not work in this state and wait for it to pass.
      72              :     Activating(ActivatingFrom),
      73              :     /// The tenant has finished activating and is open for business.
      74              :     ///
      75              :     /// Transitions out of this state are possible through `set_stopping()` and `set_broken()`.
      76              :     Active,
      77              :     /// The tenant is recognized by pageserver, but it is being detached or the
      78              :     /// system is being shut down.
      79              :     ///
      80              :     /// Transitions out of this state are possible through `set_broken()`.
      81              :     Stopping {
      82              :         // Because of https://github.com/serde-rs/serde/issues/2105 this has to be a named field,
      83              :         // otherwise it will not be skipped during deserialization
      84              :         #[serde(skip)]
      85              :         progress: completion::Barrier,
      86              :     },
      87              :     /// The tenant is recognized by the pageserver, but can no longer be used for
      88              :     /// any operations.
      89              :     ///
      90              :     /// If the tenant fails to load or attach, it will transition to this state
      91              :     /// and it is guaranteed that no background tasks are running in its name.
      92              :     ///
      93              :     /// The other way to transition into this state is from `Stopping` state
      94              :     /// through `set_broken()` called from `remove_tenant_from_memory()`. That happens
      95              :     /// if the cleanup future executed by `remove_tenant_from_memory()` fails.
      96              :     Broken { reason: String, backtrace: String },
      97              : }
      98              : 
      99              : impl TenantState {
     100            0 :     pub fn attachment_status(&self) -> TenantAttachmentStatus {
     101              :         use TenantAttachmentStatus::*;
     102              : 
     103              :         // Below TenantState::Activating is used as "transient" or "transparent" state for
     104              :         // attachment_status determining.
     105            0 :         match self {
     106              :             // The attach procedure writes the marker file before adding the Attaching tenant to the tenants map.
     107              :             // So, technically, we can return Attached here.
     108              :             // However, as soon as Console observes Attached, it will proceed with the Postgres-level health check.
     109              :             // But, our attach task might still be fetching the remote timelines, etc.
     110              :             // So, return `Maybe` while Attaching, making Console wait for the attach task to finish.
     111            0 :             Self::Attaching | Self::Activating(ActivatingFrom::Attaching) => Maybe,
     112              :             // We only reach Active after successful load / attach.
     113              :             // So, call atttachment status Attached.
     114            0 :             Self::Active => Attached,
     115              :             // If the (initial or resumed) attach procedure fails, the tenant becomes Broken.
     116              :             // However, it also becomes Broken if the regular load fails.
     117              :             // From Console's perspective there's no practical difference
     118              :             // because attachment_status is polled by console only during attach operation execution.
     119            0 :             Self::Broken { reason, .. } => Failed {
     120            0 :                 reason: reason.to_owned(),
     121            0 :             },
     122              :             // Why is Stopping a Maybe case? Because, during pageserver shutdown,
     123              :             // we set the Stopping state irrespective of whether the tenant
     124              :             // has finished attaching or not.
     125            0 :             Self::Stopping { .. } => Maybe,
     126              :         }
     127            0 :     }
     128              : 
     129            0 :     pub fn broken_from_reason(reason: String) -> Self {
     130            0 :         let backtrace_str: String = format!("{}", std::backtrace::Backtrace::force_capture());
     131            0 :         Self::Broken {
     132            0 :             reason,
     133            0 :             backtrace: backtrace_str,
     134            0 :         }
     135            0 :     }
     136              : }
     137              : 
     138              : impl std::fmt::Debug for TenantState {
     139            2 :     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
     140            2 :         match self {
     141            2 :             Self::Broken { reason, backtrace } if !reason.is_empty() => {
     142            2 :                 write!(f, "Broken due to: {reason}. Backtrace:\n{backtrace}")
     143              :             }
     144            0 :             _ => write!(f, "{self}"),
     145              :         }
     146            2 :     }
     147              : }
     148              : 
     149              : /// A temporary lease to a specific lsn inside a timeline.
     150              : /// Access to the lsn is guaranteed by the pageserver until the expiration indicated by `valid_until`.
     151              : #[serde_as]
     152            0 : #[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
     153              : pub struct LsnLease {
     154              :     #[serde_as(as = "SystemTimeAsRfc3339Millis")]
     155              :     pub valid_until: SystemTime,
     156              : }
     157              : 
     158              : serde_with::serde_conv!(
     159              :     SystemTimeAsRfc3339Millis,
     160              :     SystemTime,
     161            0 :     |time: &SystemTime| humantime::format_rfc3339_millis(*time).to_string(),
     162            0 :     |value: String| -> Result<_, humantime::TimestampError> { humantime::parse_rfc3339(&value) }
     163              : );
     164              : 
     165              : impl LsnLease {
     166              :     /// The default length for an explicit LSN lease request (10 minutes).
     167              :     pub const DEFAULT_LENGTH: Duration = Duration::from_secs(10 * 60);
     168              : 
     169              :     /// The default length for an implicit LSN lease granted during
     170              :     /// `get_lsn_by_timestamp` request (1 minutes).
     171              :     pub const DEFAULT_LENGTH_FOR_TS: Duration = Duration::from_secs(60);
     172              : 
     173              :     /// Checks whether the lease is expired.
     174           12 :     pub fn is_expired(&self, now: &SystemTime) -> bool {
     175           12 :         now > &self.valid_until
     176           12 :     }
     177              : }
     178              : 
     179              : /// The only [`TenantState`] variants we could be `TenantState::Activating` from.
     180              : ///
     181              : /// XXX: We used to have more variants here, but now it's just one, which makes this rather
     182              : /// useless. Remove, once we've checked that there's no client code left that looks at this.
     183            1 : #[derive(Clone, Copy, Debug, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
     184              : pub enum ActivatingFrom {
     185              :     /// Arrived to [`TenantState::Activating`] from [`TenantState::Attaching`]
     186              :     Attaching,
     187              : }
     188              : 
     189              : /// A state of a timeline in pageserver's memory.
     190            0 : #[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
     191              : pub enum TimelineState {
     192              :     /// The timeline is recognized by the pageserver but is not yet operational.
     193              :     /// In particular, the walreceiver connection loop is not running for this timeline.
     194              :     /// It will eventually transition to state Active or Broken.
     195              :     Loading,
     196              :     /// The timeline is fully operational.
     197              :     /// It can be queried, and the walreceiver connection loop is running.
     198              :     Active,
     199              :     /// The timeline was previously Loading or Active but is shutting down.
     200              :     /// It cannot transition back into any other state.
     201              :     Stopping,
     202              :     /// The timeline is broken and not operational (previous states: Loading or Active).
     203              :     Broken { reason: String, backtrace: String },
     204              : }
     205              : 
     206              : #[serde_with::serde_as]
     207            0 : #[derive(Debug, Clone, serde::Deserialize, serde::Serialize)]
     208              : pub struct CompactLsnRange {
     209              :     pub start: Lsn,
     210              :     pub end: Lsn,
     211              : }
     212              : 
     213              : #[serde_with::serde_as]
     214            0 : #[derive(Debug, Clone, serde::Deserialize, serde::Serialize)]
     215              : pub struct CompactKeyRange {
     216              :     #[serde_as(as = "serde_with::DisplayFromStr")]
     217              :     pub start: Key,
     218              :     #[serde_as(as = "serde_with::DisplayFromStr")]
     219              :     pub end: Key,
     220              : }
     221              : 
     222              : impl From<Range<Lsn>> for CompactLsnRange {
     223           12 :     fn from(range: Range<Lsn>) -> Self {
     224           12 :         Self {
     225           12 :             start: range.start,
     226           12 :             end: range.end,
     227           12 :         }
     228           12 :     }
     229              : }
     230              : 
     231              : impl From<Range<Key>> for CompactKeyRange {
     232           32 :     fn from(range: Range<Key>) -> Self {
     233           32 :         Self {
     234           32 :             start: range.start,
     235           32 :             end: range.end,
     236           32 :         }
     237           32 :     }
     238              : }
     239              : 
     240              : impl From<CompactLsnRange> for Range<Lsn> {
     241           20 :     fn from(range: CompactLsnRange) -> Self {
     242           20 :         range.start..range.end
     243           20 :     }
     244              : }
     245              : 
     246              : impl From<CompactKeyRange> for Range<Key> {
     247           32 :     fn from(range: CompactKeyRange) -> Self {
     248           32 :         range.start..range.end
     249           32 :     }
     250              : }
     251              : 
     252              : impl CompactLsnRange {
     253            8 :     pub fn above(lsn: Lsn) -> Self {
     254            8 :         Self {
     255            8 :             start: lsn,
     256            8 :             end: Lsn::MAX,
     257            8 :         }
     258            8 :     }
     259              : }
     260              : 
     261              : #[derive(Debug, Clone, Serialize)]
     262              : pub struct CompactInfoResponse {
     263              :     pub compact_key_range: Option<CompactKeyRange>,
     264              :     pub compact_lsn_range: Option<CompactLsnRange>,
     265              :     pub sub_compaction: bool,
     266              :     pub running: bool,
     267              :     pub job_id: usize,
     268              : }
     269              : 
     270            0 : #[derive(Serialize, Deserialize, Clone)]
     271              : pub struct TimelineCreateRequest {
     272              :     pub new_timeline_id: TimelineId,
     273              :     #[serde(flatten)]
     274              :     pub mode: TimelineCreateRequestMode,
     275              : }
     276              : 
     277              : /// Storage controller specific extensions to [`TimelineInfo`].
     278            0 : #[derive(Serialize, Deserialize, Clone)]
     279              : pub struct TimelineCreateResponseStorcon {
     280              :     #[serde(flatten)]
     281              :     pub timeline_info: TimelineInfo,
     282              : 
     283              :     pub safekeepers: Option<SafekeepersInfo>,
     284              : }
     285              : 
     286              : /// Safekeepers as returned in timeline creation request to storcon or pushed to
     287              : /// cplane in the post migration hook.
     288            0 : #[derive(Serialize, Deserialize, Clone)]
     289              : pub struct SafekeepersInfo {
     290              :     pub tenant_id: TenantId,
     291              :     pub timeline_id: TimelineId,
     292              :     pub generation: u32,
     293              :     pub safekeepers: Vec<SafekeeperInfo>,
     294              : }
     295              : 
     296            0 : #[derive(Serialize, Deserialize, Clone)]
     297              : pub struct SafekeeperInfo {
     298              :     pub id: NodeId,
     299              :     pub hostname: String,
     300              : }
     301              : 
     302            0 : #[derive(Serialize, Deserialize, Clone)]
     303              : #[serde(untagged)]
     304              : pub enum TimelineCreateRequestMode {
     305              :     Branch {
     306              :         ancestor_timeline_id: TimelineId,
     307              :         #[serde(default)]
     308              :         ancestor_start_lsn: Option<Lsn>,
     309              :         // TODO: cplane sets this, but, the branching code always
     310              :         // inherits the ancestor's pg_version. Earlier code wasn't
     311              :         // using a flattened enum, so, it was an accepted field, and
     312              :         // we continue to accept it by having it here.
     313              :         pg_version: Option<u32>,
     314              :     },
     315              :     ImportPgdata {
     316              :         import_pgdata: TimelineCreateRequestModeImportPgdata,
     317              :     },
     318              :     // NB: Bootstrap is all-optional, and thus the serde(untagged) will cause serde to stop at Bootstrap.
     319              :     // (serde picks the first matching enum variant, in declaration order).
     320              :     Bootstrap {
     321              :         #[serde(default)]
     322              :         existing_initdb_timeline_id: Option<TimelineId>,
     323              :         pg_version: Option<u32>,
     324              :     },
     325              : }
     326              : 
     327            0 : #[derive(Serialize, Deserialize, Clone)]
     328              : pub struct TimelineCreateRequestModeImportPgdata {
     329              :     pub location: ImportPgdataLocation,
     330              :     pub idempotency_key: ImportPgdataIdempotencyKey,
     331              : }
     332              : 
     333            0 : #[derive(Serialize, Deserialize, Clone, Debug)]
     334              : pub enum ImportPgdataLocation {
     335              :     #[cfg(feature = "testing")]
     336              :     LocalFs { path: Utf8PathBuf },
     337              :     AwsS3 {
     338              :         region: String,
     339              :         bucket: String,
     340              :         /// A better name for this would be `prefix`; changing requires coordination with cplane.
     341              :         /// See <https://github.com/neondatabase/cloud/issues/20646>.
     342              :         key: String,
     343              :     },
     344              : }
     345              : 
     346            0 : #[derive(Serialize, Deserialize, Clone)]
     347              : #[serde(transparent)]
     348              : pub struct ImportPgdataIdempotencyKey(pub String);
     349              : 
     350              : impl ImportPgdataIdempotencyKey {
     351            0 :     pub fn random() -> Self {
     352              :         use rand::Rng;
     353              :         use rand::distributions::Alphanumeric;
     354            0 :         Self(
     355            0 :             rand::thread_rng()
     356            0 :                 .sample_iter(&Alphanumeric)
     357            0 :                 .take(20)
     358            0 :                 .map(char::from)
     359            0 :                 .collect(),
     360            0 :         )
     361            0 :     }
     362              : }
     363              : 
     364            0 : #[derive(Serialize, Deserialize, Clone)]
     365              : pub struct LsnLeaseRequest {
     366              :     pub lsn: Lsn,
     367              : }
     368              : 
     369            0 : #[derive(Serialize, Deserialize)]
     370              : pub struct TenantShardSplitRequest {
     371              :     pub new_shard_count: u8,
     372              : 
     373              :     // A tenant's stripe size is only meaningful the first time their shard count goes
     374              :     // above 1: therefore during a split from 1->N shards, we may modify the stripe size.
     375              :     //
     376              :     // If this is set while the stripe count is being increased from an already >1 value,
     377              :     // then the request will fail with 400.
     378              :     pub new_stripe_size: Option<ShardStripeSize>,
     379              : }
     380              : 
     381            0 : #[derive(Serialize, Deserialize)]
     382              : pub struct TenantShardSplitResponse {
     383              :     pub new_shards: Vec<TenantShardId>,
     384              : }
     385              : 
     386              : /// Parameters that apply to all shards in a tenant.  Used during tenant creation.
     387            0 : #[derive(Serialize, Deserialize, Debug)]
     388              : #[serde(deny_unknown_fields)]
     389              : pub struct ShardParameters {
     390              :     pub count: ShardCount,
     391              :     pub stripe_size: ShardStripeSize,
     392              : }
     393              : 
     394              : impl ShardParameters {
     395              :     pub const DEFAULT_STRIPE_SIZE: ShardStripeSize = ShardStripeSize(256 * 1024 / 8);
     396              : 
     397            0 :     pub fn is_unsharded(&self) -> bool {
     398            0 :         self.count.is_unsharded()
     399            0 :     }
     400              : }
     401              : 
     402              : impl Default for ShardParameters {
     403          453 :     fn default() -> Self {
     404          453 :         Self {
     405          453 :             count: ShardCount::new(0),
     406          453 :             stripe_size: Self::DEFAULT_STRIPE_SIZE,
     407          453 :         }
     408          453 :     }
     409              : }
     410              : 
     411              : #[derive(Debug, Default, Clone, Eq, PartialEq)]
     412              : pub enum FieldPatch<T> {
     413              :     Upsert(T),
     414              :     Remove,
     415              :     #[default]
     416              :     Noop,
     417              : }
     418              : 
     419              : impl<T> FieldPatch<T> {
     420           70 :     fn is_noop(&self) -> bool {
     421           70 :         matches!(self, FieldPatch::Noop)
     422           70 :     }
     423              : 
     424           35 :     pub fn apply(self, target: &mut Option<T>) {
     425           35 :         match self {
     426            1 :             Self::Upsert(v) => *target = Some(v),
     427            1 :             Self::Remove => *target = None,
     428           33 :             Self::Noop => {}
     429              :         }
     430           35 :     }
     431              : 
     432           10 :     pub fn map<U, E, F: FnOnce(T) -> Result<U, E>>(self, map: F) -> Result<FieldPatch<U>, E> {
     433           10 :         match self {
     434            0 :             Self::Upsert(v) => Ok(FieldPatch::<U>::Upsert(map(v)?)),
     435            0 :             Self::Remove => Ok(FieldPatch::<U>::Remove),
     436           10 :             Self::Noop => Ok(FieldPatch::<U>::Noop),
     437              :         }
     438           10 :     }
     439              : }
     440              : 
     441              : impl<'de, T: Deserialize<'de>> Deserialize<'de> for FieldPatch<T> {
     442            2 :     fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
     443            2 :     where
     444            2 :         D: Deserializer<'de>,
     445            2 :     {
     446            2 :         Option::deserialize(deserializer).map(|opt| match opt {
     447            1 :             None => FieldPatch::Remove,
     448            1 :             Some(val) => FieldPatch::Upsert(val),
     449            2 :         })
     450            2 :     }
     451              : }
     452              : 
     453              : impl<T: Serialize> Serialize for FieldPatch<T> {
     454            2 :     fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
     455            2 :     where
     456            2 :         S: Serializer,
     457            2 :     {
     458            2 :         match self {
     459            1 :             FieldPatch::Upsert(val) => serializer.serialize_some(val),
     460            1 :             FieldPatch::Remove => serializer.serialize_none(),
     461            0 :             FieldPatch::Noop => unreachable!(),
     462              :         }
     463            2 :     }
     464              : }
     465              : 
     466            2 : #[derive(Serialize, Deserialize, Debug, Default, Clone, Eq, PartialEq)]
     467              : #[serde(default)]
     468              : pub struct TenantConfigPatch {
     469              :     #[serde(skip_serializing_if = "FieldPatch::is_noop")]
     470              :     pub checkpoint_distance: FieldPatch<u64>,
     471              :     #[serde(skip_serializing_if = "FieldPatch::is_noop")]
     472              :     pub checkpoint_timeout: FieldPatch<String>,
     473              :     #[serde(skip_serializing_if = "FieldPatch::is_noop")]
     474              :     pub compaction_target_size: FieldPatch<u64>,
     475              :     #[serde(skip_serializing_if = "FieldPatch::is_noop")]
     476              :     pub compaction_period: FieldPatch<String>,
     477              :     #[serde(skip_serializing_if = "FieldPatch::is_noop")]
     478              :     pub compaction_threshold: FieldPatch<usize>,
     479              :     #[serde(skip_serializing_if = "FieldPatch::is_noop")]
     480              :     pub compaction_upper_limit: FieldPatch<usize>,
     481              :     // defer parsing compaction_algorithm, like eviction_policy
     482              :     #[serde(skip_serializing_if = "FieldPatch::is_noop")]
     483              :     pub compaction_algorithm: FieldPatch<CompactionAlgorithmSettings>,
     484              :     #[serde(skip_serializing_if = "FieldPatch::is_noop")]
     485              :     pub compaction_l0_first: FieldPatch<bool>,
     486              :     #[serde(skip_serializing_if = "FieldPatch::is_noop")]
     487              :     pub compaction_l0_semaphore: FieldPatch<bool>,
     488              :     #[serde(skip_serializing_if = "FieldPatch::is_noop")]
     489              :     pub l0_flush_delay_threshold: FieldPatch<usize>,
     490              :     #[serde(skip_serializing_if = "FieldPatch::is_noop")]
     491              :     pub l0_flush_stall_threshold: FieldPatch<usize>,
     492              :     #[serde(skip_serializing_if = "FieldPatch::is_noop")]
     493              :     pub l0_flush_wait_upload: FieldPatch<bool>,
     494              :     #[serde(skip_serializing_if = "FieldPatch::is_noop")]
     495              :     pub gc_horizon: FieldPatch<u64>,
     496              :     #[serde(skip_serializing_if = "FieldPatch::is_noop")]
     497              :     pub gc_period: FieldPatch<String>,
     498              :     #[serde(skip_serializing_if = "FieldPatch::is_noop")]
     499              :     pub image_creation_threshold: FieldPatch<usize>,
     500              :     #[serde(skip_serializing_if = "FieldPatch::is_noop")]
     501              :     pub pitr_interval: FieldPatch<String>,
     502              :     #[serde(skip_serializing_if = "FieldPatch::is_noop")]
     503              :     pub walreceiver_connect_timeout: FieldPatch<String>,
     504              :     #[serde(skip_serializing_if = "FieldPatch::is_noop")]
     505              :     pub lagging_wal_timeout: FieldPatch<String>,
     506              :     #[serde(skip_serializing_if = "FieldPatch::is_noop")]
     507              :     pub max_lsn_wal_lag: FieldPatch<NonZeroU64>,
     508              :     #[serde(skip_serializing_if = "FieldPatch::is_noop")]
     509              :     pub eviction_policy: FieldPatch<EvictionPolicy>,
     510              :     #[serde(skip_serializing_if = "FieldPatch::is_noop")]
     511              :     pub min_resident_size_override: FieldPatch<u64>,
     512              :     #[serde(skip_serializing_if = "FieldPatch::is_noop")]
     513              :     pub evictions_low_residence_duration_metric_threshold: FieldPatch<String>,
     514              :     #[serde(skip_serializing_if = "FieldPatch::is_noop")]
     515              :     pub heatmap_period: FieldPatch<String>,
     516              :     #[serde(skip_serializing_if = "FieldPatch::is_noop")]
     517              :     pub lazy_slru_download: FieldPatch<bool>,
     518              :     #[serde(skip_serializing_if = "FieldPatch::is_noop")]
     519              :     pub timeline_get_throttle: FieldPatch<ThrottleConfig>,
     520              :     #[serde(skip_serializing_if = "FieldPatch::is_noop")]
     521              :     pub image_layer_creation_check_threshold: FieldPatch<u8>,
     522              :     #[serde(skip_serializing_if = "FieldPatch::is_noop")]
     523              :     pub image_creation_preempt_threshold: FieldPatch<usize>,
     524              :     #[serde(skip_serializing_if = "FieldPatch::is_noop")]
     525              :     pub lsn_lease_length: FieldPatch<String>,
     526              :     #[serde(skip_serializing_if = "FieldPatch::is_noop")]
     527              :     pub lsn_lease_length_for_ts: FieldPatch<String>,
     528              :     #[serde(skip_serializing_if = "FieldPatch::is_noop")]
     529              :     pub timeline_offloading: FieldPatch<bool>,
     530              :     #[serde(skip_serializing_if = "FieldPatch::is_noop")]
     531              :     pub wal_receiver_protocol_override: FieldPatch<PostgresClientProtocol>,
     532              :     #[serde(skip_serializing_if = "FieldPatch::is_noop")]
     533              :     pub rel_size_v2_enabled: FieldPatch<bool>,
     534              :     #[serde(skip_serializing_if = "FieldPatch::is_noop")]
     535              :     pub gc_compaction_enabled: FieldPatch<bool>,
     536              :     #[serde(skip_serializing_if = "FieldPatch::is_noop")]
     537              :     pub gc_compaction_initial_threshold_kb: FieldPatch<u64>,
     538              :     #[serde(skip_serializing_if = "FieldPatch::is_noop")]
     539              :     pub gc_compaction_ratio_percent: FieldPatch<u64>,
     540              : }
     541              : 
     542              : /// An alternative representation of `pageserver::tenant::TenantConf` with
     543              : /// simpler types.
     544           40 : #[derive(Serialize, Deserialize, Debug, Default, Clone, Eq, PartialEq)]
     545              : pub struct TenantConfig {
     546              :     pub checkpoint_distance: Option<u64>,
     547              :     #[serde(default)]
     548              :     #[serde(with = "humantime_serde")]
     549              :     pub checkpoint_timeout: Option<Duration>,
     550              :     pub compaction_target_size: Option<u64>,
     551              :     #[serde(default)]
     552              :     #[serde(with = "humantime_serde")]
     553              :     pub compaction_period: Option<Duration>,
     554              :     pub compaction_threshold: Option<usize>,
     555              :     pub compaction_upper_limit: Option<usize>,
     556              :     // defer parsing compaction_algorithm, like eviction_policy
     557              :     pub compaction_algorithm: Option<CompactionAlgorithmSettings>,
     558              :     pub compaction_l0_first: Option<bool>,
     559              :     pub compaction_l0_semaphore: Option<bool>,
     560              :     pub l0_flush_delay_threshold: Option<usize>,
     561              :     pub l0_flush_stall_threshold: Option<usize>,
     562              :     pub l0_flush_wait_upload: Option<bool>,
     563              :     pub gc_horizon: Option<u64>,
     564              :     #[serde(default)]
     565              :     #[serde(with = "humantime_serde")]
     566              :     pub gc_period: Option<Duration>,
     567              :     pub image_creation_threshold: Option<usize>,
     568              :     #[serde(default)]
     569              :     #[serde(with = "humantime_serde")]
     570              :     pub pitr_interval: Option<Duration>,
     571              :     #[serde(default)]
     572              :     #[serde(with = "humantime_serde")]
     573              :     pub walreceiver_connect_timeout: Option<Duration>,
     574              :     #[serde(default)]
     575              :     #[serde(with = "humantime_serde")]
     576              :     pub lagging_wal_timeout: Option<Duration>,
     577              :     pub max_lsn_wal_lag: Option<NonZeroU64>,
     578              :     pub eviction_policy: Option<EvictionPolicy>,
     579              :     pub min_resident_size_override: Option<u64>,
     580              :     #[serde(default)]
     581              :     #[serde(with = "humantime_serde")]
     582              :     pub evictions_low_residence_duration_metric_threshold: Option<Duration>,
     583              :     #[serde(default)]
     584              :     #[serde(with = "humantime_serde")]
     585              :     pub heatmap_period: Option<Duration>,
     586              :     pub lazy_slru_download: Option<bool>,
     587              :     pub timeline_get_throttle: Option<ThrottleConfig>,
     588              :     pub image_layer_creation_check_threshold: Option<u8>,
     589              :     pub image_creation_preempt_threshold: Option<usize>,
     590              :     #[serde(default)]
     591              :     #[serde(with = "humantime_serde")]
     592              :     pub lsn_lease_length: Option<Duration>,
     593              :     #[serde(default)]
     594              :     #[serde(with = "humantime_serde")]
     595              :     pub lsn_lease_length_for_ts: Option<Duration>,
     596              :     pub timeline_offloading: Option<bool>,
     597              :     pub wal_receiver_protocol_override: Option<PostgresClientProtocol>,
     598              :     pub rel_size_v2_enabled: Option<bool>,
     599              :     pub gc_compaction_enabled: Option<bool>,
     600              :     pub gc_compaction_initial_threshold_kb: Option<u64>,
     601              :     pub gc_compaction_ratio_percent: Option<u64>,
     602              : }
     603              : 
     604              : impl TenantConfig {
     605            1 :     pub fn apply_patch(
     606            1 :         self,
     607            1 :         patch: TenantConfigPatch,
     608            1 :     ) -> Result<TenantConfig, humantime::DurationError> {
     609            1 :         let Self {
     610            1 :             mut checkpoint_distance,
     611            1 :             mut checkpoint_timeout,
     612            1 :             mut compaction_target_size,
     613            1 :             mut compaction_period,
     614            1 :             mut compaction_threshold,
     615            1 :             mut compaction_upper_limit,
     616            1 :             mut compaction_algorithm,
     617            1 :             mut compaction_l0_first,
     618            1 :             mut compaction_l0_semaphore,
     619            1 :             mut l0_flush_delay_threshold,
     620            1 :             mut l0_flush_stall_threshold,
     621            1 :             mut l0_flush_wait_upload,
     622            1 :             mut gc_horizon,
     623            1 :             mut gc_period,
     624            1 :             mut image_creation_threshold,
     625            1 :             mut pitr_interval,
     626            1 :             mut walreceiver_connect_timeout,
     627            1 :             mut lagging_wal_timeout,
     628            1 :             mut max_lsn_wal_lag,
     629            1 :             mut eviction_policy,
     630            1 :             mut min_resident_size_override,
     631            1 :             mut evictions_low_residence_duration_metric_threshold,
     632            1 :             mut heatmap_period,
     633            1 :             mut lazy_slru_download,
     634            1 :             mut timeline_get_throttle,
     635            1 :             mut image_layer_creation_check_threshold,
     636            1 :             mut image_creation_preempt_threshold,
     637            1 :             mut lsn_lease_length,
     638            1 :             mut lsn_lease_length_for_ts,
     639            1 :             mut timeline_offloading,
     640            1 :             mut wal_receiver_protocol_override,
     641            1 :             mut rel_size_v2_enabled,
     642            1 :             mut gc_compaction_enabled,
     643            1 :             mut gc_compaction_initial_threshold_kb,
     644            1 :             mut gc_compaction_ratio_percent,
     645            1 :         } = self;
     646            1 : 
     647            1 :         patch.checkpoint_distance.apply(&mut checkpoint_distance);
     648            1 :         patch
     649            1 :             .checkpoint_timeout
     650            1 :             .map(|v| humantime::parse_duration(&v))?
     651            1 :             .apply(&mut checkpoint_timeout);
     652            1 :         patch
     653            1 :             .compaction_target_size
     654            1 :             .apply(&mut compaction_target_size);
     655            1 :         patch
     656            1 :             .compaction_period
     657            1 :             .map(|v| humantime::parse_duration(&v))?
     658            1 :             .apply(&mut compaction_period);
     659            1 :         patch.compaction_threshold.apply(&mut compaction_threshold);
     660            1 :         patch
     661            1 :             .compaction_upper_limit
     662            1 :             .apply(&mut compaction_upper_limit);
     663            1 :         patch.compaction_algorithm.apply(&mut compaction_algorithm);
     664            1 :         patch.compaction_l0_first.apply(&mut compaction_l0_first);
     665            1 :         patch
     666            1 :             .compaction_l0_semaphore
     667            1 :             .apply(&mut compaction_l0_semaphore);
     668            1 :         patch
     669            1 :             .l0_flush_delay_threshold
     670            1 :             .apply(&mut l0_flush_delay_threshold);
     671            1 :         patch
     672            1 :             .l0_flush_stall_threshold
     673            1 :             .apply(&mut l0_flush_stall_threshold);
     674            1 :         patch.l0_flush_wait_upload.apply(&mut l0_flush_wait_upload);
     675            1 :         patch.gc_horizon.apply(&mut gc_horizon);
     676            1 :         patch
     677            1 :             .gc_period
     678            1 :             .map(|v| humantime::parse_duration(&v))?
     679            1 :             .apply(&mut gc_period);
     680            1 :         patch
     681            1 :             .image_creation_threshold
     682            1 :             .apply(&mut image_creation_threshold);
     683            1 :         patch
     684            1 :             .pitr_interval
     685            1 :             .map(|v| humantime::parse_duration(&v))?
     686            1 :             .apply(&mut pitr_interval);
     687            1 :         patch
     688            1 :             .walreceiver_connect_timeout
     689            1 :             .map(|v| humantime::parse_duration(&v))?
     690            1 :             .apply(&mut walreceiver_connect_timeout);
     691            1 :         patch
     692            1 :             .lagging_wal_timeout
     693            1 :             .map(|v| humantime::parse_duration(&v))?
     694            1 :             .apply(&mut lagging_wal_timeout);
     695            1 :         patch.max_lsn_wal_lag.apply(&mut max_lsn_wal_lag);
     696            1 :         patch.eviction_policy.apply(&mut eviction_policy);
     697            1 :         patch
     698            1 :             .min_resident_size_override
     699            1 :             .apply(&mut min_resident_size_override);
     700            1 :         patch
     701            1 :             .evictions_low_residence_duration_metric_threshold
     702            1 :             .map(|v| humantime::parse_duration(&v))?
     703            1 :             .apply(&mut evictions_low_residence_duration_metric_threshold);
     704            1 :         patch
     705            1 :             .heatmap_period
     706            1 :             .map(|v| humantime::parse_duration(&v))?
     707            1 :             .apply(&mut heatmap_period);
     708            1 :         patch.lazy_slru_download.apply(&mut lazy_slru_download);
     709            1 :         patch
     710            1 :             .timeline_get_throttle
     711            1 :             .apply(&mut timeline_get_throttle);
     712            1 :         patch
     713            1 :             .image_layer_creation_check_threshold
     714            1 :             .apply(&mut image_layer_creation_check_threshold);
     715            1 :         patch
     716            1 :             .image_creation_preempt_threshold
     717            1 :             .apply(&mut image_creation_preempt_threshold);
     718            1 :         patch
     719            1 :             .lsn_lease_length
     720            1 :             .map(|v| humantime::parse_duration(&v))?
     721            1 :             .apply(&mut lsn_lease_length);
     722            1 :         patch
     723            1 :             .lsn_lease_length_for_ts
     724            1 :             .map(|v| humantime::parse_duration(&v))?
     725            1 :             .apply(&mut lsn_lease_length_for_ts);
     726            1 :         patch.timeline_offloading.apply(&mut timeline_offloading);
     727            1 :         patch
     728            1 :             .wal_receiver_protocol_override
     729            1 :             .apply(&mut wal_receiver_protocol_override);
     730            1 :         patch.rel_size_v2_enabled.apply(&mut rel_size_v2_enabled);
     731            1 :         patch
     732            1 :             .gc_compaction_enabled
     733            1 :             .apply(&mut gc_compaction_enabled);
     734            1 :         patch
     735            1 :             .gc_compaction_initial_threshold_kb
     736            1 :             .apply(&mut gc_compaction_initial_threshold_kb);
     737            1 :         patch
     738            1 :             .gc_compaction_ratio_percent
     739            1 :             .apply(&mut gc_compaction_ratio_percent);
     740            1 : 
     741            1 :         Ok(Self {
     742            1 :             checkpoint_distance,
     743            1 :             checkpoint_timeout,
     744            1 :             compaction_target_size,
     745            1 :             compaction_period,
     746            1 :             compaction_threshold,
     747            1 :             compaction_upper_limit,
     748            1 :             compaction_algorithm,
     749            1 :             compaction_l0_first,
     750            1 :             compaction_l0_semaphore,
     751            1 :             l0_flush_delay_threshold,
     752            1 :             l0_flush_stall_threshold,
     753            1 :             l0_flush_wait_upload,
     754            1 :             gc_horizon,
     755            1 :             gc_period,
     756            1 :             image_creation_threshold,
     757            1 :             pitr_interval,
     758            1 :             walreceiver_connect_timeout,
     759            1 :             lagging_wal_timeout,
     760            1 :             max_lsn_wal_lag,
     761            1 :             eviction_policy,
     762            1 :             min_resident_size_override,
     763            1 :             evictions_low_residence_duration_metric_threshold,
     764            1 :             heatmap_period,
     765            1 :             lazy_slru_download,
     766            1 :             timeline_get_throttle,
     767            1 :             image_layer_creation_check_threshold,
     768            1 :             image_creation_preempt_threshold,
     769            1 :             lsn_lease_length,
     770            1 :             lsn_lease_length_for_ts,
     771            1 :             timeline_offloading,
     772            1 :             wal_receiver_protocol_override,
     773            1 :             rel_size_v2_enabled,
     774            1 :             gc_compaction_enabled,
     775            1 :             gc_compaction_initial_threshold_kb,
     776            1 :             gc_compaction_ratio_percent,
     777            1 :         })
     778            1 :     }
     779              : }
     780              : 
     781              : /// The policy for the aux file storage.
     782              : ///
     783              : /// It can be switched through `switch_aux_file_policy` tenant config.
     784              : /// When the first aux file written, the policy will be persisted in the
     785              : /// `index_part.json` file and has a limited migration path.
     786              : ///
     787              : /// Currently, we only allow the following migration path:
     788              : ///
     789              : /// Unset -> V1
     790              : ///       -> V2
     791              : ///       -> CrossValidation -> V2
     792              : #[derive(
     793              :     Eq,
     794              :     PartialEq,
     795              :     Debug,
     796              :     Copy,
     797              :     Clone,
     798            0 :     strum_macros::EnumString,
     799              :     strum_macros::Display,
     800            4 :     serde_with::DeserializeFromStr,
     801              :     serde_with::SerializeDisplay,
     802              : )]
     803              : #[strum(serialize_all = "kebab-case")]
     804              : pub enum AuxFilePolicy {
     805              :     /// V1 aux file policy: store everything in AUX_FILE_KEY
     806              :     #[strum(ascii_case_insensitive)]
     807              :     V1,
     808              :     /// V2 aux file policy: store in the AUX_FILE keyspace
     809              :     #[strum(ascii_case_insensitive)]
     810              :     V2,
     811              :     /// Cross validation runs both formats on the write path and does validation
     812              :     /// on the read path.
     813              :     #[strum(ascii_case_insensitive)]
     814              :     CrossValidation,
     815              : }
     816              : 
     817            0 : #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
     818              : #[serde(tag = "kind")]
     819              : pub enum EvictionPolicy {
     820              :     NoEviction,
     821              :     LayerAccessThreshold(EvictionPolicyLayerAccessThreshold),
     822              :     OnlyImitiate(EvictionPolicyLayerAccessThreshold),
     823              : }
     824              : 
     825              : impl EvictionPolicy {
     826            0 :     pub fn discriminant_str(&self) -> &'static str {
     827            0 :         match self {
     828            0 :             EvictionPolicy::NoEviction => "NoEviction",
     829            0 :             EvictionPolicy::LayerAccessThreshold(_) => "LayerAccessThreshold",
     830            0 :             EvictionPolicy::OnlyImitiate(_) => "OnlyImitiate",
     831              :         }
     832            0 :     }
     833              : }
     834              : 
     835              : #[derive(
     836              :     Eq,
     837              :     PartialEq,
     838              :     Debug,
     839              :     Copy,
     840              :     Clone,
     841            0 :     strum_macros::EnumString,
     842              :     strum_macros::Display,
     843            0 :     serde_with::DeserializeFromStr,
     844              :     serde_with::SerializeDisplay,
     845              : )]
     846              : #[strum(serialize_all = "kebab-case")]
     847              : pub enum CompactionAlgorithm {
     848              :     Legacy,
     849              :     Tiered,
     850              : }
     851              : 
     852              : #[derive(
     853            4 :     Debug, Clone, Copy, PartialEq, Eq, serde_with::DeserializeFromStr, serde_with::SerializeDisplay,
     854              : )]
     855              : pub enum ImageCompressionAlgorithm {
     856              :     // Disabled for writes, support decompressing during read path
     857              :     Disabled,
     858              :     /// Zstandard compression. Level 0 means and None mean the same (default level). Levels can be negative as well.
     859              :     /// For details, see the [manual](http://facebook.github.io/zstd/zstd_manual.html).
     860              :     Zstd {
     861              :         level: Option<i8>,
     862              :     },
     863              : }
     864              : 
     865              : impl FromStr for ImageCompressionAlgorithm {
     866              :     type Err = anyhow::Error;
     867            8 :     fn from_str(s: &str) -> Result<Self, Self::Err> {
     868            8 :         let mut components = s.split(['(', ')']);
     869            8 :         let first = components
     870            8 :             .next()
     871            8 :             .ok_or_else(|| anyhow::anyhow!("empty string"))?;
     872            8 :         match first {
     873            8 :             "disabled" => Ok(ImageCompressionAlgorithm::Disabled),
     874            6 :             "zstd" => {
     875            6 :                 let level = if let Some(v) = components.next() {
     876            4 :                     let v: i8 = v.parse()?;
     877            4 :                     Some(v)
     878              :                 } else {
     879            2 :                     None
     880              :                 };
     881              : 
     882            6 :                 Ok(ImageCompressionAlgorithm::Zstd { level })
     883              :             }
     884            0 :             _ => anyhow::bail!("invalid specifier '{first}'"),
     885              :         }
     886            8 :     }
     887              : }
     888              : 
     889              : impl Display for ImageCompressionAlgorithm {
     890           12 :     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
     891           12 :         match self {
     892            3 :             ImageCompressionAlgorithm::Disabled => write!(f, "disabled"),
     893            9 :             ImageCompressionAlgorithm::Zstd { level } => {
     894            9 :                 if let Some(level) = level {
     895            6 :                     write!(f, "zstd({})", level)
     896              :                 } else {
     897            3 :                     write!(f, "zstd")
     898              :                 }
     899              :             }
     900              :         }
     901           12 :     }
     902              : }
     903              : 
     904            0 : #[derive(Eq, PartialEq, Debug, Clone, Serialize, Deserialize)]
     905              : pub struct CompactionAlgorithmSettings {
     906              :     pub kind: CompactionAlgorithm,
     907              : }
     908              : 
     909            8 : #[derive(Debug, PartialEq, Eq, Clone, Deserialize, Serialize)]
     910              : #[serde(tag = "mode", rename_all = "kebab-case", deny_unknown_fields)]
     911              : pub enum L0FlushConfig {
     912              :     #[serde(rename_all = "snake_case")]
     913              :     Direct { max_concurrency: NonZeroUsize },
     914              : }
     915              : 
     916            0 : #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
     917              : pub struct EvictionPolicyLayerAccessThreshold {
     918              :     #[serde(with = "humantime_serde")]
     919              :     pub period: Duration,
     920              :     #[serde(with = "humantime_serde")]
     921              :     pub threshold: Duration,
     922              : }
     923              : 
     924            6 : #[derive(Debug, Serialize, Deserialize, Clone, PartialEq, Eq)]
     925              : pub struct ThrottleConfig {
     926              :     /// See [`ThrottleConfigTaskKinds`] for why we do the serde `rename`.
     927              :     #[serde(rename = "task_kinds")]
     928              :     pub enabled: ThrottleConfigTaskKinds,
     929              :     pub initial: u32,
     930              :     #[serde(with = "humantime_serde")]
     931              :     pub refill_interval: Duration,
     932              :     pub refill_amount: NonZeroU32,
     933              :     pub max: u32,
     934              : }
     935              : 
     936              : /// Before <https://github.com/neondatabase/neon/pull/9962>
     937              : /// the throttle was a per `Timeline::get`/`Timeline::get_vectored` call.
     938              : /// The `task_kinds` field controlled which Pageserver "Task Kind"s
     939              : /// were subject to the throttle.
     940              : ///
     941              : /// After that PR, the throttle is applied at pagestream request level
     942              : /// and the `task_kinds` field does not apply since the only task kind
     943              : /// that us subject to the throttle is that of the page service.
     944              : ///
     945              : /// However, we don't want to make a breaking config change right now
     946              : /// because it means we have to migrate all the tenant configs.
     947              : /// This will be done in a future PR.
     948              : ///
     949              : /// In the meantime, we use emptiness / non-emptsiness of the `task_kinds`
     950              : /// field to determine if the throttle is enabled or not.
     951            1 : #[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Eq)]
     952              : #[serde(transparent)]
     953              : pub struct ThrottleConfigTaskKinds(Vec<String>);
     954              : 
     955              : impl ThrottleConfigTaskKinds {
     956          925 :     pub fn disabled() -> Self {
     957          925 :         Self(vec![])
     958          925 :     }
     959          454 :     pub fn is_enabled(&self) -> bool {
     960          454 :         !self.0.is_empty()
     961          454 :     }
     962              : }
     963              : 
     964              : impl ThrottleConfig {
     965          925 :     pub fn disabled() -> Self {
     966          925 :         Self {
     967          925 :             enabled: ThrottleConfigTaskKinds::disabled(),
     968          925 :             // other values don't matter with emtpy `task_kinds`.
     969          925 :             initial: 0,
     970          925 :             refill_interval: Duration::from_millis(1),
     971          925 :             refill_amount: NonZeroU32::new(1).unwrap(),
     972          925 :             max: 1,
     973          925 :         }
     974          925 :     }
     975              :     /// The requests per second allowed  by the given config.
     976            0 :     pub fn steady_rps(&self) -> f64 {
     977            0 :         (self.refill_amount.get() as f64) / (self.refill_interval.as_secs_f64())
     978            0 :     }
     979              : }
     980              : 
     981              : #[cfg(test)]
     982              : mod throttle_config_tests {
     983              :     use super::*;
     984              : 
     985              :     #[test]
     986            1 :     fn test_disabled_is_disabled() {
     987            1 :         let config = ThrottleConfig::disabled();
     988            1 :         assert!(!config.enabled.is_enabled());
     989            1 :     }
     990              :     #[test]
     991            1 :     fn test_enabled_backwards_compat() {
     992            1 :         let input = serde_json::json!({
     993            1 :             "task_kinds": ["PageRequestHandler"],
     994            1 :             "initial": 40000,
     995            1 :             "refill_interval": "50ms",
     996            1 :             "refill_amount": 1000,
     997            1 :             "max": 40000,
     998            1 :             "fair": true
     999            1 :         });
    1000            1 :         let config: ThrottleConfig = serde_json::from_value(input).unwrap();
    1001            1 :         assert!(config.enabled.is_enabled());
    1002            1 :     }
    1003              : }
    1004              : 
    1005              : /// A flattened analog of a `pagesever::tenant::LocationMode`, which
    1006              : /// lists out all possible states (and the virtual "Detached" state)
    1007              : /// in a flat form rather than using rust-style enums.
    1008            0 : #[derive(Serialize, Deserialize, Debug, Clone, Copy, Eq, PartialEq)]
    1009              : pub enum LocationConfigMode {
    1010              :     AttachedSingle,
    1011              :     AttachedMulti,
    1012              :     AttachedStale,
    1013              :     Secondary,
    1014              :     Detached,
    1015              : }
    1016              : 
    1017            0 : #[derive(Serialize, Deserialize, Debug, Clone, Eq, PartialEq)]
    1018              : pub struct LocationConfigSecondary {
    1019              :     pub warm: bool,
    1020              : }
    1021              : 
    1022              : /// An alternative representation of `pageserver::tenant::LocationConf`,
    1023              : /// for use in external-facing APIs.
    1024            0 : #[derive(Serialize, Deserialize, Debug, Clone, Eq, PartialEq)]
    1025              : pub struct LocationConfig {
    1026              :     pub mode: LocationConfigMode,
    1027              :     /// If attaching, in what generation?
    1028              :     #[serde(default)]
    1029              :     pub generation: Option<u32>,
    1030              : 
    1031              :     // If requesting mode `Secondary`, configuration for that.
    1032              :     #[serde(default)]
    1033              :     pub secondary_conf: Option<LocationConfigSecondary>,
    1034              : 
    1035              :     // Shard parameters: if shard_count is nonzero, then other shard_* fields
    1036              :     // must be set accurately.
    1037              :     #[serde(default)]
    1038              :     pub shard_number: u8,
    1039              :     #[serde(default)]
    1040              :     pub shard_count: u8,
    1041              :     #[serde(default)]
    1042              :     pub shard_stripe_size: u32,
    1043              : 
    1044              :     // This configuration only affects attached mode, but should be provided irrespective
    1045              :     // of the mode, as a secondary location might transition on startup if the response
    1046              :     // to the `/re-attach` control plane API requests it.
    1047              :     pub tenant_conf: TenantConfig,
    1048              : }
    1049              : 
    1050            0 : #[derive(Serialize, Deserialize)]
    1051              : pub struct LocationConfigListResponse {
    1052              :     pub tenant_shards: Vec<(TenantShardId, Option<LocationConfig>)>,
    1053              : }
    1054              : 
    1055              : #[derive(Serialize)]
    1056              : pub struct StatusResponse {
    1057              :     pub id: NodeId,
    1058              : }
    1059              : 
    1060            0 : #[derive(Serialize, Deserialize, Debug)]
    1061              : #[serde(deny_unknown_fields)]
    1062              : pub struct TenantLocationConfigRequest {
    1063              :     #[serde(flatten)]
    1064              :     pub config: LocationConfig, // as we have a flattened field, we should reject all unknown fields in it
    1065              : }
    1066              : 
    1067            0 : #[derive(Serialize, Deserialize, Debug)]
    1068              : #[serde(deny_unknown_fields)]
    1069              : pub struct TenantTimeTravelRequest {
    1070              :     pub shard_counts: Vec<ShardCount>,
    1071              : }
    1072              : 
    1073            0 : #[derive(Serialize, Deserialize, Debug)]
    1074              : #[serde(deny_unknown_fields)]
    1075              : pub struct TenantShardLocation {
    1076              :     pub shard_id: TenantShardId,
    1077              :     pub node_id: NodeId,
    1078              : }
    1079              : 
    1080            0 : #[derive(Serialize, Deserialize, Debug)]
    1081              : #[serde(deny_unknown_fields)]
    1082              : pub struct TenantLocationConfigResponse {
    1083              :     pub shards: Vec<TenantShardLocation>,
    1084              :     // If the shards' ShardCount count is >1, stripe_size will be set.
    1085              :     pub stripe_size: Option<ShardStripeSize>,
    1086              : }
    1087              : 
    1088            2 : #[derive(Serialize, Deserialize, Debug)]
    1089              : #[serde(deny_unknown_fields)]
    1090              : pub struct TenantConfigRequest {
    1091              :     pub tenant_id: TenantId,
    1092              :     #[serde(flatten)]
    1093              :     pub config: TenantConfig, // as we have a flattened field, we should reject all unknown fields in it
    1094              : }
    1095              : 
    1096              : impl std::ops::Deref for TenantConfigRequest {
    1097              :     type Target = TenantConfig;
    1098              : 
    1099            0 :     fn deref(&self) -> &Self::Target {
    1100            0 :         &self.config
    1101            0 :     }
    1102              : }
    1103              : 
    1104              : impl TenantConfigRequest {
    1105            0 :     pub fn new(tenant_id: TenantId) -> TenantConfigRequest {
    1106            0 :         let config = TenantConfig::default();
    1107            0 :         TenantConfigRequest { tenant_id, config }
    1108            0 :     }
    1109              : }
    1110              : 
    1111            3 : #[derive(Serialize, Deserialize, Debug)]
    1112              : #[serde(deny_unknown_fields)]
    1113              : pub struct TenantConfigPatchRequest {
    1114              :     pub tenant_id: TenantId,
    1115              :     #[serde(flatten)]
    1116              :     pub config: TenantConfigPatch, // as we have a flattened field, we should reject all unknown fields in it
    1117              : }
    1118              : 
    1119            0 : #[derive(Serialize, Deserialize, Debug)]
    1120              : pub struct TenantWaitLsnRequest {
    1121              :     #[serde(flatten)]
    1122              :     pub timelines: HashMap<TimelineId, Lsn>,
    1123              :     pub timeout: Duration,
    1124              : }
    1125              : 
    1126              : /// See [`TenantState::attachment_status`] and the OpenAPI docs for context.
    1127            0 : #[derive(Serialize, Deserialize, Clone)]
    1128              : #[serde(tag = "slug", content = "data", rename_all = "snake_case")]
    1129              : pub enum TenantAttachmentStatus {
    1130              :     Maybe,
    1131              :     Attached,
    1132              :     Failed { reason: String },
    1133              : }
    1134              : 
    1135            0 : #[derive(Serialize, Deserialize, Clone)]
    1136              : pub struct TenantInfo {
    1137              :     pub id: TenantShardId,
    1138              :     // NB: intentionally not part of OpenAPI, we don't want to commit to a specific set of TenantState's
    1139              :     pub state: TenantState,
    1140              :     /// Sum of the size of all layer files.
    1141              :     /// If a layer is present in both local FS and S3, it counts only once.
    1142              :     pub current_physical_size: Option<u64>, // physical size is only included in `tenant_status` endpoint
    1143              :     pub attachment_status: TenantAttachmentStatus,
    1144              :     pub generation: u32,
    1145              : 
    1146              :     /// Opaque explanation if gc is being blocked.
    1147              :     ///
    1148              :     /// Only looked up for the individual tenant detail, not the listing.
    1149              :     #[serde(skip_serializing_if = "Option::is_none")]
    1150              :     pub gc_blocking: Option<String>,
    1151              : }
    1152              : 
    1153            0 : #[derive(Serialize, Deserialize, Clone)]
    1154              : pub struct TenantDetails {
    1155              :     #[serde(flatten)]
    1156              :     pub tenant_info: TenantInfo,
    1157              : 
    1158              :     pub walredo: Option<WalRedoManagerStatus>,
    1159              : 
    1160              :     pub timelines: Vec<TimelineId>,
    1161              : }
    1162              : 
    1163            0 : #[derive(Serialize, Deserialize, PartialEq, Eq, Clone, Copy, Debug)]
    1164              : pub enum TimelineArchivalState {
    1165              :     Archived,
    1166              :     Unarchived,
    1167              : }
    1168              : 
    1169            0 : #[derive(Serialize, Deserialize, PartialEq, Eq, Clone)]
    1170              : pub struct TimelineArchivalConfigRequest {
    1171              :     pub state: TimelineArchivalState,
    1172              : }
    1173              : 
    1174            0 : #[derive(Serialize, Deserialize, PartialEq, Eq, Clone)]
    1175              : pub struct TimelinePatchIndexPartRequest {
    1176              :     pub rel_size_migration: Option<RelSizeMigration>,
    1177              :     pub gc_compaction_last_completed_lsn: Option<Lsn>,
    1178              :     pub applied_gc_cutoff_lsn: Option<Lsn>,
    1179              :     #[serde(default)]
    1180              :     pub force_index_update: bool,
    1181              : }
    1182              : 
    1183            0 : #[derive(Debug, Serialize, Deserialize, Clone)]
    1184              : pub struct TimelinesInfoAndOffloaded {
    1185              :     pub timelines: Vec<TimelineInfo>,
    1186              :     pub offloaded: Vec<OffloadedTimelineInfo>,
    1187              : }
    1188              : 
    1189              : /// Analog of [`TimelineInfo`] for offloaded timelines.
    1190            0 : #[derive(Debug, Serialize, Deserialize, Clone)]
    1191              : pub struct OffloadedTimelineInfo {
    1192              :     pub tenant_id: TenantShardId,
    1193              :     pub timeline_id: TimelineId,
    1194              :     /// Whether the timeline has a parent it has been branched off from or not
    1195              :     pub ancestor_timeline_id: Option<TimelineId>,
    1196              :     /// Whether to retain the branch lsn at the ancestor or not
    1197              :     pub ancestor_retain_lsn: Option<Lsn>,
    1198              :     /// The time point when the timeline was archived
    1199              :     pub archived_at: chrono::DateTime<chrono::Utc>,
    1200              : }
    1201              : 
    1202           12 : #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
    1203              : #[serde(rename_all = "camelCase")]
    1204              : pub enum RelSizeMigration {
    1205              :     /// The tenant is using the old rel_size format.
    1206              :     /// Note that this enum is persisted as `Option<RelSizeMigration>` in the index part, so
    1207              :     /// `None` is the same as `Some(RelSizeMigration::Legacy)`.
    1208              :     Legacy,
    1209              :     /// The tenant is migrating to the new rel_size format. Both old and new rel_size format are
    1210              :     /// persisted in the index part. The read path will read both formats and merge them.
    1211              :     Migrating,
    1212              :     /// The tenant has migrated to the new rel_size format. Only the new rel_size format is persisted
    1213              :     /// in the index part, and the read path will not read the old format.
    1214              :     Migrated,
    1215              : }
    1216              : 
    1217              : /// This represents the output of the "timeline_detail" and "timeline_list" API calls.
    1218            0 : #[derive(Debug, Serialize, Deserialize, Clone)]
    1219              : pub struct TimelineInfo {
    1220              :     pub tenant_id: TenantShardId,
    1221              :     pub timeline_id: TimelineId,
    1222              : 
    1223              :     pub ancestor_timeline_id: Option<TimelineId>,
    1224              :     pub ancestor_lsn: Option<Lsn>,
    1225              :     pub last_record_lsn: Lsn,
    1226              :     pub prev_record_lsn: Option<Lsn>,
    1227              : 
    1228              :     /// Legacy field, retained for one version to enable old storage controller to
    1229              :     /// decode (it was a mandatory field).
    1230              :     #[serde(default, rename = "latest_gc_cutoff_lsn")]
    1231              :     pub _unused: Lsn,
    1232              : 
    1233              :     /// The LSN up to which GC has advanced: older data may still exist but it is not available for clients.
    1234              :     /// This LSN is not suitable for deciding where to create branches etc: use [`TimelineInfo::min_readable_lsn`] instead,
    1235              :     /// as it is easier to reason about.
    1236              :     #[serde(default)]
    1237              :     pub applied_gc_cutoff_lsn: Lsn,
    1238              : 
    1239              :     /// The upper bound of data which is either already GC'ed, or elegible to be GC'ed at any time based on PITR interval.
    1240              :     /// This LSN represents the "end of history" for this timeline, and callers should use it to figure out the oldest
    1241              :     /// LSN at which it is legal to create a branch or ephemeral endpoint.
    1242              :     ///
    1243              :     /// Note that holders of valid LSN leases may be able to create branches and read pages earlier
    1244              :     /// than this LSN, but new leases may not be taken out earlier than this LSN.
    1245              :     #[serde(default)]
    1246              :     pub min_readable_lsn: Lsn,
    1247              : 
    1248              :     pub disk_consistent_lsn: Lsn,
    1249              : 
    1250              :     /// The LSN that we have succesfully uploaded to remote storage
    1251              :     pub remote_consistent_lsn: Lsn,
    1252              : 
    1253              :     /// The LSN that we are advertizing to safekeepers
    1254              :     pub remote_consistent_lsn_visible: Lsn,
    1255              : 
    1256              :     /// The LSN from the start of the root timeline (never changes)
    1257              :     pub initdb_lsn: Lsn,
    1258              : 
    1259              :     pub current_logical_size: u64,
    1260              :     pub current_logical_size_is_accurate: bool,
    1261              : 
    1262              :     pub directory_entries_counts: Vec<u64>,
    1263              : 
    1264              :     /// Sum of the size of all layer files.
    1265              :     /// If a layer is present in both local FS and S3, it counts only once.
    1266              :     pub current_physical_size: Option<u64>, // is None when timeline is Unloaded
    1267              :     pub current_logical_size_non_incremental: Option<u64>,
    1268              : 
    1269              :     /// How many bytes of WAL are within this branch's pitr_interval.  If the pitr_interval goes
    1270              :     /// beyond the branch's branch point, we only count up to the branch point.
    1271              :     pub pitr_history_size: u64,
    1272              : 
    1273              :     /// Whether this branch's branch point is within its ancestor's PITR interval (i.e. any
    1274              :     /// ancestor data used by this branch would have been retained anyway).  If this is false, then
    1275              :     /// this branch may be imposing a cost on the ancestor by causing it to retain layers that it would
    1276              :     /// otherwise be able to GC.
    1277              :     pub within_ancestor_pitr: bool,
    1278              : 
    1279              :     pub timeline_dir_layer_file_size_sum: Option<u64>,
    1280              : 
    1281              :     pub wal_source_connstr: Option<String>,
    1282              :     pub last_received_msg_lsn: Option<Lsn>,
    1283              :     /// the timestamp (in microseconds) of the last received message
    1284              :     pub last_received_msg_ts: Option<u128>,
    1285              :     pub pg_version: u32,
    1286              : 
    1287              :     pub state: TimelineState,
    1288              : 
    1289              :     pub walreceiver_status: String,
    1290              : 
    1291              :     // ALWAYS add new fields at the end of the struct with `Option` to ensure forward/backward compatibility.
    1292              :     // Backward compatibility: you will get a JSON not containing the newly-added field.
    1293              :     // Forward compatibility: a previous version of the pageserver will receive a JSON. serde::Deserialize does
    1294              :     // not deny unknown fields by default so it's safe to set the field to some value, though it won't be
    1295              :     // read.
    1296              :     /// Whether the timeline is archived.
    1297              :     pub is_archived: Option<bool>,
    1298              : 
    1299              :     /// The status of the rel_size migration.
    1300              :     pub rel_size_migration: Option<RelSizeMigration>,
    1301              : }
    1302              : 
    1303            0 : #[derive(Debug, Clone, Serialize, Deserialize)]
    1304              : pub struct LayerMapInfo {
    1305              :     pub in_memory_layers: Vec<InMemoryLayerInfo>,
    1306              :     pub historic_layers: Vec<HistoricLayerInfo>,
    1307              : }
    1308              : 
    1309              : /// The residence status of a layer
    1310            0 : #[derive(Debug, Clone, Copy, Serialize, Deserialize)]
    1311              : pub enum LayerResidenceStatus {
    1312              :     /// Residence status for a layer file that exists locally.
    1313              :     /// It may also exist on the remote, we don't care here.
    1314              :     Resident,
    1315              :     /// Residence status for a layer file that only exists on the remote.
    1316              :     Evicted,
    1317              : }
    1318              : 
    1319              : #[serde_as]
    1320            0 : #[derive(Debug, Clone, Serialize, Deserialize)]
    1321              : pub struct LayerAccessStats {
    1322              :     #[serde_as(as = "serde_with::TimestampMilliSeconds")]
    1323              :     pub access_time: SystemTime,
    1324              : 
    1325              :     #[serde_as(as = "serde_with::TimestampMilliSeconds")]
    1326              :     pub residence_time: SystemTime,
    1327              : 
    1328              :     pub visible: bool,
    1329              : }
    1330              : 
    1331            0 : #[derive(Debug, Clone, Serialize, Deserialize)]
    1332              : #[serde(tag = "kind")]
    1333              : pub enum InMemoryLayerInfo {
    1334              :     Open { lsn_start: Lsn },
    1335              :     Frozen { lsn_start: Lsn, lsn_end: Lsn },
    1336              : }
    1337              : 
    1338            0 : #[derive(Debug, Clone, Serialize, Deserialize)]
    1339              : #[serde(tag = "kind")]
    1340              : pub enum HistoricLayerInfo {
    1341              :     Delta {
    1342              :         layer_file_name: String,
    1343              :         layer_file_size: u64,
    1344              : 
    1345              :         lsn_start: Lsn,
    1346              :         lsn_end: Lsn,
    1347              :         remote: bool,
    1348              :         access_stats: LayerAccessStats,
    1349              : 
    1350              :         l0: bool,
    1351              :     },
    1352              :     Image {
    1353              :         layer_file_name: String,
    1354              :         layer_file_size: u64,
    1355              : 
    1356              :         lsn_start: Lsn,
    1357              :         remote: bool,
    1358              :         access_stats: LayerAccessStats,
    1359              :     },
    1360              : }
    1361              : 
    1362              : impl HistoricLayerInfo {
    1363            0 :     pub fn layer_file_name(&self) -> &str {
    1364            0 :         match self {
    1365              :             HistoricLayerInfo::Delta {
    1366            0 :                 layer_file_name, ..
    1367            0 :             } => layer_file_name,
    1368              :             HistoricLayerInfo::Image {
    1369            0 :                 layer_file_name, ..
    1370            0 :             } => layer_file_name,
    1371              :         }
    1372            0 :     }
    1373            0 :     pub fn is_remote(&self) -> bool {
    1374            0 :         match self {
    1375            0 :             HistoricLayerInfo::Delta { remote, .. } => *remote,
    1376            0 :             HistoricLayerInfo::Image { remote, .. } => *remote,
    1377              :         }
    1378            0 :     }
    1379            0 :     pub fn set_remote(&mut self, value: bool) {
    1380            0 :         let field = match self {
    1381            0 :             HistoricLayerInfo::Delta { remote, .. } => remote,
    1382            0 :             HistoricLayerInfo::Image { remote, .. } => remote,
    1383              :         };
    1384            0 :         *field = value;
    1385            0 :     }
    1386            0 :     pub fn layer_file_size(&self) -> u64 {
    1387            0 :         match self {
    1388              :             HistoricLayerInfo::Delta {
    1389            0 :                 layer_file_size, ..
    1390            0 :             } => *layer_file_size,
    1391              :             HistoricLayerInfo::Image {
    1392            0 :                 layer_file_size, ..
    1393            0 :             } => *layer_file_size,
    1394              :         }
    1395            0 :     }
    1396              : }
    1397              : 
    1398            0 : #[derive(Debug, Serialize, Deserialize)]
    1399              : pub struct DownloadRemoteLayersTaskSpawnRequest {
    1400              :     pub max_concurrent_downloads: NonZeroUsize,
    1401              : }
    1402              : 
    1403            0 : #[derive(Debug, Serialize, Deserialize)]
    1404              : pub struct IngestAuxFilesRequest {
    1405              :     pub aux_files: HashMap<String, String>,
    1406              : }
    1407              : 
    1408            0 : #[derive(Debug, Serialize, Deserialize)]
    1409              : pub struct ListAuxFilesRequest {
    1410              :     pub lsn: Lsn,
    1411              : }
    1412              : 
    1413            0 : #[derive(Debug, Serialize, Deserialize, Clone)]
    1414              : pub struct DownloadRemoteLayersTaskInfo {
    1415              :     pub task_id: String,
    1416              :     pub state: DownloadRemoteLayersTaskState,
    1417              :     pub total_layer_count: u64,         // stable once `completed`
    1418              :     pub successful_download_count: u64, // stable once `completed`
    1419              :     pub failed_download_count: u64,     // stable once `completed`
    1420              : }
    1421              : 
    1422            0 : #[derive(Debug, Serialize, Deserialize, Clone)]
    1423              : pub enum DownloadRemoteLayersTaskState {
    1424              :     Running,
    1425              :     Completed,
    1426              :     ShutDown,
    1427              : }
    1428              : 
    1429            0 : #[derive(Debug, Serialize, Deserialize)]
    1430              : pub struct TimelineGcRequest {
    1431              :     pub gc_horizon: Option<u64>,
    1432              : }
    1433              : 
    1434            0 : #[derive(Debug, Clone, Serialize, Deserialize)]
    1435              : pub struct WalRedoManagerProcessStatus {
    1436              :     pub pid: u32,
    1437              : }
    1438              : 
    1439            0 : #[derive(Debug, Clone, Serialize, Deserialize)]
    1440              : pub struct WalRedoManagerStatus {
    1441              :     pub last_redo_at: Option<chrono::DateTime<chrono::Utc>>,
    1442              :     pub process: Option<WalRedoManagerProcessStatus>,
    1443              : }
    1444              : 
    1445              : /// The progress of a secondary tenant.
    1446              : ///
    1447              : /// It is mostly useful when doing a long running download: e.g. initiating
    1448              : /// a download job, timing out while waiting for it to run, and then inspecting this status to understand
    1449              : /// what's happening.
    1450            0 : #[derive(Default, Debug, Serialize, Deserialize, Clone)]
    1451              : pub struct SecondaryProgress {
    1452              :     /// The remote storage LastModified time of the heatmap object we last downloaded.
    1453              :     pub heatmap_mtime: Option<serde_system_time::SystemTime>,
    1454              : 
    1455              :     /// The number of layers currently on-disk
    1456              :     pub layers_downloaded: usize,
    1457              :     /// The number of layers in the most recently seen heatmap
    1458              :     pub layers_total: usize,
    1459              : 
    1460              :     /// The number of layer bytes currently on-disk
    1461              :     pub bytes_downloaded: u64,
    1462              :     /// The number of layer bytes in the most recently seen heatmap
    1463              :     pub bytes_total: u64,
    1464              : }
    1465              : 
    1466            0 : #[derive(Serialize, Deserialize, Debug)]
    1467              : pub struct TenantScanRemoteStorageShard {
    1468              :     pub tenant_shard_id: TenantShardId,
    1469              :     pub generation: Option<u32>,
    1470              : }
    1471              : 
    1472            0 : #[derive(Serialize, Deserialize, Debug, Default)]
    1473              : pub struct TenantScanRemoteStorageResponse {
    1474              :     pub shards: Vec<TenantScanRemoteStorageShard>,
    1475              : }
    1476              : 
    1477            0 : #[derive(Serialize, Deserialize, Debug, Clone)]
    1478              : #[serde(rename_all = "snake_case")]
    1479              : pub enum TenantSorting {
    1480              :     /// Total size of layers on local disk for all timelines in a shard.
    1481              :     ResidentSize,
    1482              :     /// The logical size of the largest timeline within a _tenant_ (not shard). Only tracked on
    1483              :     /// shard 0, contains the sum across all shards.
    1484              :     MaxLogicalSize,
    1485              :     /// The logical size of the largest timeline within a _tenant_ (not shard), divided by number of
    1486              :     /// shards. Only tracked on shard 0, and estimates the per-shard logical size.
    1487              :     MaxLogicalSizePerShard,
    1488              : }
    1489              : 
    1490              : impl Default for TenantSorting {
    1491            0 :     fn default() -> Self {
    1492            0 :         Self::ResidentSize
    1493            0 :     }
    1494              : }
    1495              : 
    1496            0 : #[derive(Serialize, Deserialize, Debug, Clone)]
    1497              : pub struct TopTenantShardsRequest {
    1498              :     // How would you like to sort the tenants?
    1499              :     pub order_by: TenantSorting,
    1500              : 
    1501              :     // How many results?
    1502              :     pub limit: usize,
    1503              : 
    1504              :     // Omit tenants with more than this many shards (e.g. if this is the max number of shards
    1505              :     // that the caller would ever split to)
    1506              :     pub where_shards_lt: Option<ShardCount>,
    1507              : 
    1508              :     // Omit tenants where the ordering metric is less than this (this is an optimization to
    1509              :     // let us quickly exclude numerous tiny shards)
    1510              :     pub where_gt: Option<u64>,
    1511              : }
    1512              : 
    1513            0 : #[derive(Serialize, Deserialize, Debug, PartialEq, Eq)]
    1514              : pub struct TopTenantShardItem {
    1515              :     pub id: TenantShardId,
    1516              : 
    1517              :     /// Total size of layers on local disk for all timelines in this shard.
    1518              :     pub resident_size: u64,
    1519              : 
    1520              :     /// Total size of layers in remote storage for all timelines in this shard.
    1521              :     pub physical_size: u64,
    1522              : 
    1523              :     /// The largest logical size of a timeline within this _tenant_ (not shard). This is only
    1524              :     /// tracked on shard 0, and contains the sum of the logical size across all shards.
    1525              :     pub max_logical_size: u64,
    1526              : 
    1527              :     /// The largest logical size of a timeline within this _tenant_ (not shard) divided by number of
    1528              :     /// shards. This is only tracked on shard 0, and is only an estimate as we divide it evenly by
    1529              :     /// shard count, rounded up.
    1530              :     pub max_logical_size_per_shard: u64,
    1531              : }
    1532              : 
    1533            0 : #[derive(Serialize, Deserialize, Debug, Default)]
    1534              : pub struct TopTenantShardsResponse {
    1535              :     pub shards: Vec<TopTenantShardItem>,
    1536              : }
    1537              : 
    1538              : pub mod virtual_file {
    1539              :     #[derive(
    1540              :         Copy,
    1541              :         Clone,
    1542              :         PartialEq,
    1543              :         Eq,
    1544              :         Hash,
    1545            0 :         strum_macros::EnumString,
    1546              :         strum_macros::Display,
    1547            0 :         serde_with::DeserializeFromStr,
    1548              :         serde_with::SerializeDisplay,
    1549              :         Debug,
    1550              :     )]
    1551              :     #[strum(serialize_all = "kebab-case")]
    1552              :     pub enum IoEngineKind {
    1553              :         StdFs,
    1554              :         #[cfg(target_os = "linux")]
    1555              :         TokioEpollUring,
    1556              :     }
    1557              : 
    1558              :     /// Direct IO modes for a pageserver.
    1559              :     #[derive(
    1560              :         Copy,
    1561              :         Clone,
    1562              :         PartialEq,
    1563              :         Eq,
    1564              :         Hash,
    1565            0 :         strum_macros::EnumString,
    1566              :         strum_macros::Display,
    1567            0 :         serde_with::DeserializeFromStr,
    1568              :         serde_with::SerializeDisplay,
    1569              :         Debug,
    1570              :     )]
    1571              :     #[strum(serialize_all = "kebab-case")]
    1572              :     #[repr(u8)]
    1573              :     pub enum IoMode {
    1574              :         /// Uses buffered IO.
    1575              :         Buffered,
    1576              :         /// Uses direct IO, error out if the operation fails.
    1577              :         #[cfg(target_os = "linux")]
    1578              :         Direct,
    1579              :     }
    1580              : 
    1581              :     impl IoMode {
    1582          488 :         pub const fn preferred() -> Self {
    1583          488 :             Self::Buffered
    1584          488 :         }
    1585              :     }
    1586              : 
    1587              :     impl TryFrom<u8> for IoMode {
    1588              :         type Error = u8;
    1589              : 
    1590         5096 :         fn try_from(value: u8) -> Result<Self, Self::Error> {
    1591         5096 :             Ok(match value {
    1592         5096 :                 v if v == (IoMode::Buffered as u8) => IoMode::Buffered,
    1593              :                 #[cfg(target_os = "linux")]
    1594            0 :                 v if v == (IoMode::Direct as u8) => IoMode::Direct,
    1595            0 :                 x => return Err(x),
    1596              :             })
    1597         5096 :         }
    1598              :     }
    1599              : }
    1600              : 
    1601            0 : #[derive(Debug, Clone, Serialize, Deserialize)]
    1602              : pub struct ScanDisposableKeysResponse {
    1603              :     pub disposable_count: usize,
    1604              :     pub not_disposable_count: usize,
    1605              : }
    1606              : 
    1607              : // Wrapped in libpq CopyData
    1608              : #[derive(PartialEq, Eq, Debug)]
    1609              : pub enum PagestreamFeMessage {
    1610              :     Exists(PagestreamExistsRequest),
    1611              :     Nblocks(PagestreamNblocksRequest),
    1612              :     GetPage(PagestreamGetPageRequest),
    1613              :     DbSize(PagestreamDbSizeRequest),
    1614              :     GetSlruSegment(PagestreamGetSlruSegmentRequest),
    1615              :     #[cfg(feature = "testing")]
    1616              :     Test(PagestreamTestRequest),
    1617              : }
    1618              : 
    1619              : // Wrapped in libpq CopyData
    1620              : #[derive(strum_macros::EnumProperty)]
    1621              : pub enum PagestreamBeMessage {
    1622              :     Exists(PagestreamExistsResponse),
    1623              :     Nblocks(PagestreamNblocksResponse),
    1624              :     GetPage(PagestreamGetPageResponse),
    1625              :     Error(PagestreamErrorResponse),
    1626              :     DbSize(PagestreamDbSizeResponse),
    1627              :     GetSlruSegment(PagestreamGetSlruSegmentResponse),
    1628              :     #[cfg(feature = "testing")]
    1629              :     Test(PagestreamTestResponse),
    1630              : }
    1631              : 
    1632              : // Keep in sync with `pagestore_client.h`
    1633              : #[repr(u8)]
    1634              : enum PagestreamFeMessageTag {
    1635              :     Exists = 0,
    1636              :     Nblocks = 1,
    1637              :     GetPage = 2,
    1638              :     DbSize = 3,
    1639              :     GetSlruSegment = 4,
    1640              :     /* future tags above this line */
    1641              :     /// For testing purposes, not available in production.
    1642              :     #[cfg(feature = "testing")]
    1643              :     Test = 99,
    1644              : }
    1645              : 
    1646              : // Keep in sync with `pagestore_client.h`
    1647              : #[repr(u8)]
    1648              : enum PagestreamBeMessageTag {
    1649              :     Exists = 100,
    1650              :     Nblocks = 101,
    1651              :     GetPage = 102,
    1652              :     Error = 103,
    1653              :     DbSize = 104,
    1654              :     GetSlruSegment = 105,
    1655              :     /* future tags above this line */
    1656              :     /// For testing purposes, not available in production.
    1657              :     #[cfg(feature = "testing")]
    1658              :     Test = 199,
    1659              : }
    1660              : 
    1661              : impl TryFrom<u8> for PagestreamFeMessageTag {
    1662              :     type Error = u8;
    1663            4 :     fn try_from(value: u8) -> Result<Self, u8> {
    1664            4 :         match value {
    1665            1 :             0 => Ok(PagestreamFeMessageTag::Exists),
    1666            1 :             1 => Ok(PagestreamFeMessageTag::Nblocks),
    1667            1 :             2 => Ok(PagestreamFeMessageTag::GetPage),
    1668            1 :             3 => Ok(PagestreamFeMessageTag::DbSize),
    1669            0 :             4 => Ok(PagestreamFeMessageTag::GetSlruSegment),
    1670              :             #[cfg(feature = "testing")]
    1671            0 :             99 => Ok(PagestreamFeMessageTag::Test),
    1672            0 :             _ => Err(value),
    1673              :         }
    1674            4 :     }
    1675              : }
    1676              : 
    1677              : impl TryFrom<u8> for PagestreamBeMessageTag {
    1678              :     type Error = u8;
    1679            0 :     fn try_from(value: u8) -> Result<Self, u8> {
    1680            0 :         match value {
    1681            0 :             100 => Ok(PagestreamBeMessageTag::Exists),
    1682            0 :             101 => Ok(PagestreamBeMessageTag::Nblocks),
    1683            0 :             102 => Ok(PagestreamBeMessageTag::GetPage),
    1684            0 :             103 => Ok(PagestreamBeMessageTag::Error),
    1685            0 :             104 => Ok(PagestreamBeMessageTag::DbSize),
    1686            0 :             105 => Ok(PagestreamBeMessageTag::GetSlruSegment),
    1687              :             #[cfg(feature = "testing")]
    1688            0 :             199 => Ok(PagestreamBeMessageTag::Test),
    1689            0 :             _ => Err(value),
    1690              :         }
    1691            0 :     }
    1692              : }
    1693              : 
    1694              : // A GetPage request contains two LSN values:
    1695              : //
    1696              : // request_lsn: Get the page version at this point in time.  Lsn::Max is a special value that means
    1697              : // "get the latest version present". It's used by the primary server, which knows that no one else
    1698              : // is writing WAL. 'not_modified_since' must be set to a proper value even if request_lsn is
    1699              : // Lsn::Max. Standby servers use the current replay LSN as the request LSN.
    1700              : //
    1701              : // not_modified_since: Hint to the pageserver that the client knows that the page has not been
    1702              : // modified between 'not_modified_since' and the request LSN. It's always correct to set
    1703              : // 'not_modified_since equal' to 'request_lsn' (unless Lsn::Max is used as the 'request_lsn'), but
    1704              : // passing an earlier LSN can speed up the request, by allowing the pageserver to process the
    1705              : // request without waiting for 'request_lsn' to arrive.
    1706              : //
    1707              : // The now-defunct V1 interface contained only one LSN, and a boolean 'latest' flag. The V1 interface was
    1708              : // sufficient for the primary; the 'lsn' was equivalent to the 'not_modified_since' value, and
    1709              : // 'latest' was set to true. The V2 interface was added because there was no correct way for a
    1710              : // standby to request a page at a particular non-latest LSN, and also include the
    1711              : // 'not_modified_since' hint. That led to an awkward choice of either using an old LSN in the
    1712              : // request, if the standby knows that the page hasn't been modified since, and risk getting an error
    1713              : // if that LSN has fallen behind the GC horizon, or requesting the current replay LSN, which could
    1714              : // require the pageserver unnecessarily to wait for the WAL to arrive up to that point. The new V2
    1715              : // interface allows sending both LSNs, and let the pageserver do the right thing. There was no
    1716              : // difference in the responses between V1 and V2.
    1717              : //
    1718              : // V3 version of protocol adds request ID to all requests. This request ID is also included in response
    1719              : // as well as other fields from requests, which allows to verify that we receive response for our request.
    1720              : // We copy fields from request to response to make checking more reliable: request ID is formed from process ID
    1721              : // and local counter, so in principle there can be duplicated requests IDs if process PID is reused.
    1722              : //
    1723              : #[derive(Debug, PartialEq, Eq, Clone, Copy)]
    1724              : pub enum PagestreamProtocolVersion {
    1725              :     V2,
    1726              :     V3,
    1727              : }
    1728              : 
    1729              : pub type RequestId = u64;
    1730              : 
    1731              : #[derive(Debug, PartialEq, Eq, Clone, Copy)]
    1732              : pub struct PagestreamRequest {
    1733              :     pub reqid: RequestId,
    1734              :     pub request_lsn: Lsn,
    1735              :     pub not_modified_since: Lsn,
    1736              : }
    1737              : 
    1738              : #[derive(Debug, PartialEq, Eq, Clone, Copy)]
    1739              : pub struct PagestreamExistsRequest {
    1740              :     pub hdr: PagestreamRequest,
    1741              :     pub rel: RelTag,
    1742              : }
    1743              : 
    1744              : #[derive(Debug, PartialEq, Eq, Clone, Copy)]
    1745              : pub struct PagestreamNblocksRequest {
    1746              :     pub hdr: PagestreamRequest,
    1747              :     pub rel: RelTag,
    1748              : }
    1749              : 
    1750              : #[derive(Debug, PartialEq, Eq, Clone, Copy)]
    1751              : pub struct PagestreamGetPageRequest {
    1752              :     pub hdr: PagestreamRequest,
    1753              :     pub rel: RelTag,
    1754              :     pub blkno: u32,
    1755              : }
    1756              : 
    1757              : #[derive(Debug, PartialEq, Eq, Clone, Copy)]
    1758              : pub struct PagestreamDbSizeRequest {
    1759              :     pub hdr: PagestreamRequest,
    1760              :     pub dbnode: u32,
    1761              : }
    1762              : 
    1763              : #[derive(Debug, PartialEq, Eq, Clone, Copy)]
    1764              : pub struct PagestreamGetSlruSegmentRequest {
    1765              :     pub hdr: PagestreamRequest,
    1766              :     pub kind: u8,
    1767              :     pub segno: u32,
    1768              : }
    1769              : 
    1770              : #[derive(Debug)]
    1771              : pub struct PagestreamExistsResponse {
    1772              :     pub req: PagestreamExistsRequest,
    1773              :     pub exists: bool,
    1774              : }
    1775              : 
    1776              : #[derive(Debug)]
    1777              : pub struct PagestreamNblocksResponse {
    1778              :     pub req: PagestreamNblocksRequest,
    1779              :     pub n_blocks: u32,
    1780              : }
    1781              : 
    1782              : #[derive(Debug)]
    1783              : pub struct PagestreamGetPageResponse {
    1784              :     pub req: PagestreamGetPageRequest,
    1785              :     pub page: Bytes,
    1786              : }
    1787              : 
    1788              : #[derive(Debug)]
    1789              : pub struct PagestreamGetSlruSegmentResponse {
    1790              :     pub req: PagestreamGetSlruSegmentRequest,
    1791              :     pub segment: Bytes,
    1792              : }
    1793              : 
    1794              : #[derive(Debug)]
    1795              : pub struct PagestreamErrorResponse {
    1796              :     pub req: PagestreamRequest,
    1797              :     pub message: String,
    1798              : }
    1799              : 
    1800              : #[derive(Debug)]
    1801              : pub struct PagestreamDbSizeResponse {
    1802              :     pub req: PagestreamDbSizeRequest,
    1803              :     pub db_size: i64,
    1804              : }
    1805              : 
    1806              : #[cfg(feature = "testing")]
    1807              : #[derive(Debug, PartialEq, Eq, Clone)]
    1808              : pub struct PagestreamTestRequest {
    1809              :     pub hdr: PagestreamRequest,
    1810              :     pub batch_key: u64,
    1811              :     pub message: String,
    1812              : }
    1813              : 
    1814              : #[cfg(feature = "testing")]
    1815              : #[derive(Debug)]
    1816              : pub struct PagestreamTestResponse {
    1817              :     pub req: PagestreamTestRequest,
    1818              : }
    1819              : 
    1820              : // This is a cut-down version of TenantHistorySize from the pageserver crate, omitting fields
    1821              : // that require pageserver-internal types.  It is sufficient to get the total size.
    1822            0 : #[derive(Serialize, Deserialize, Debug)]
    1823              : pub struct TenantHistorySize {
    1824              :     pub id: TenantId,
    1825              :     /// Size is a mixture of WAL and logical size, so the unit is bytes.
    1826              :     ///
    1827              :     /// Will be none if `?inputs_only=true` was given.
    1828              :     pub size: Option<u64>,
    1829              : }
    1830              : 
    1831              : impl PagestreamFeMessage {
    1832              :     /// Serialize a compute -> pageserver message. This is currently only used in testing
    1833              :     /// tools. Always uses protocol version 3.
    1834            4 :     pub fn serialize(&self) -> Bytes {
    1835            4 :         let mut bytes = BytesMut::new();
    1836            4 : 
    1837            4 :         match self {
    1838            1 :             Self::Exists(req) => {
    1839            1 :                 bytes.put_u8(PagestreamFeMessageTag::Exists as u8);
    1840            1 :                 bytes.put_u64(req.hdr.reqid);
    1841            1 :                 bytes.put_u64(req.hdr.request_lsn.0);
    1842            1 :                 bytes.put_u64(req.hdr.not_modified_since.0);
    1843            1 :                 bytes.put_u32(req.rel.spcnode);
    1844            1 :                 bytes.put_u32(req.rel.dbnode);
    1845            1 :                 bytes.put_u32(req.rel.relnode);
    1846            1 :                 bytes.put_u8(req.rel.forknum);
    1847            1 :             }
    1848              : 
    1849            1 :             Self::Nblocks(req) => {
    1850            1 :                 bytes.put_u8(PagestreamFeMessageTag::Nblocks as u8);
    1851            1 :                 bytes.put_u64(req.hdr.reqid);
    1852            1 :                 bytes.put_u64(req.hdr.request_lsn.0);
    1853            1 :                 bytes.put_u64(req.hdr.not_modified_since.0);
    1854            1 :                 bytes.put_u32(req.rel.spcnode);
    1855            1 :                 bytes.put_u32(req.rel.dbnode);
    1856            1 :                 bytes.put_u32(req.rel.relnode);
    1857            1 :                 bytes.put_u8(req.rel.forknum);
    1858            1 :             }
    1859              : 
    1860            1 :             Self::GetPage(req) => {
    1861            1 :                 bytes.put_u8(PagestreamFeMessageTag::GetPage as u8);
    1862            1 :                 bytes.put_u64(req.hdr.reqid);
    1863            1 :                 bytes.put_u64(req.hdr.request_lsn.0);
    1864            1 :                 bytes.put_u64(req.hdr.not_modified_since.0);
    1865            1 :                 bytes.put_u32(req.rel.spcnode);
    1866            1 :                 bytes.put_u32(req.rel.dbnode);
    1867            1 :                 bytes.put_u32(req.rel.relnode);
    1868            1 :                 bytes.put_u8(req.rel.forknum);
    1869            1 :                 bytes.put_u32(req.blkno);
    1870            1 :             }
    1871              : 
    1872            1 :             Self::DbSize(req) => {
    1873            1 :                 bytes.put_u8(PagestreamFeMessageTag::DbSize as u8);
    1874            1 :                 bytes.put_u64(req.hdr.reqid);
    1875            1 :                 bytes.put_u64(req.hdr.request_lsn.0);
    1876            1 :                 bytes.put_u64(req.hdr.not_modified_since.0);
    1877            1 :                 bytes.put_u32(req.dbnode);
    1878            1 :             }
    1879              : 
    1880            0 :             Self::GetSlruSegment(req) => {
    1881            0 :                 bytes.put_u8(PagestreamFeMessageTag::GetSlruSegment as u8);
    1882            0 :                 bytes.put_u64(req.hdr.reqid);
    1883            0 :                 bytes.put_u64(req.hdr.request_lsn.0);
    1884            0 :                 bytes.put_u64(req.hdr.not_modified_since.0);
    1885            0 :                 bytes.put_u8(req.kind);
    1886            0 :                 bytes.put_u32(req.segno);
    1887            0 :             }
    1888              :             #[cfg(feature = "testing")]
    1889            0 :             Self::Test(req) => {
    1890            0 :                 bytes.put_u8(PagestreamFeMessageTag::Test as u8);
    1891            0 :                 bytes.put_u64(req.hdr.reqid);
    1892            0 :                 bytes.put_u64(req.hdr.request_lsn.0);
    1893            0 :                 bytes.put_u64(req.hdr.not_modified_since.0);
    1894            0 :                 bytes.put_u64(req.batch_key);
    1895            0 :                 let message = req.message.as_bytes();
    1896            0 :                 bytes.put_u64(message.len() as u64);
    1897            0 :                 bytes.put_slice(message);
    1898            0 :             }
    1899              :         }
    1900              : 
    1901            4 :         bytes.into()
    1902            4 :     }
    1903              : 
    1904            4 :     pub fn parse<R: std::io::Read>(
    1905            4 :         body: &mut R,
    1906            4 :         protocol_version: PagestreamProtocolVersion,
    1907            4 :     ) -> anyhow::Result<PagestreamFeMessage> {
    1908              :         // these correspond to the NeonMessageTag enum in pagestore_client.h
    1909              :         //
    1910              :         // TODO: consider using protobuf or serde bincode for less error prone
    1911              :         // serialization.
    1912            4 :         let msg_tag = body.read_u8()?;
    1913            4 :         let (reqid, request_lsn, not_modified_since) = match protocol_version {
    1914              :             PagestreamProtocolVersion::V2 => (
    1915              :                 0,
    1916            0 :                 Lsn::from(body.read_u64::<BigEndian>()?),
    1917            0 :                 Lsn::from(body.read_u64::<BigEndian>()?),
    1918              :             ),
    1919              :             PagestreamProtocolVersion::V3 => (
    1920            4 :                 body.read_u64::<BigEndian>()?,
    1921            4 :                 Lsn::from(body.read_u64::<BigEndian>()?),
    1922            4 :                 Lsn::from(body.read_u64::<BigEndian>()?),
    1923              :             ),
    1924              :         };
    1925              : 
    1926            4 :         match PagestreamFeMessageTag::try_from(msg_tag)
    1927            4 :             .map_err(|tag: u8| anyhow::anyhow!("invalid tag {tag}"))?
    1928              :         {
    1929              :             PagestreamFeMessageTag::Exists => {
    1930              :                 Ok(PagestreamFeMessage::Exists(PagestreamExistsRequest {
    1931            1 :                     hdr: PagestreamRequest {
    1932            1 :                         reqid,
    1933            1 :                         request_lsn,
    1934            1 :                         not_modified_since,
    1935            1 :                     },
    1936            1 :                     rel: RelTag {
    1937            1 :                         spcnode: body.read_u32::<BigEndian>()?,
    1938            1 :                         dbnode: body.read_u32::<BigEndian>()?,
    1939            1 :                         relnode: body.read_u32::<BigEndian>()?,
    1940            1 :                         forknum: body.read_u8()?,
    1941              :                     },
    1942              :                 }))
    1943              :             }
    1944              :             PagestreamFeMessageTag::Nblocks => {
    1945              :                 Ok(PagestreamFeMessage::Nblocks(PagestreamNblocksRequest {
    1946            1 :                     hdr: PagestreamRequest {
    1947            1 :                         reqid,
    1948            1 :                         request_lsn,
    1949            1 :                         not_modified_since,
    1950            1 :                     },
    1951            1 :                     rel: RelTag {
    1952            1 :                         spcnode: body.read_u32::<BigEndian>()?,
    1953            1 :                         dbnode: body.read_u32::<BigEndian>()?,
    1954            1 :                         relnode: body.read_u32::<BigEndian>()?,
    1955            1 :                         forknum: body.read_u8()?,
    1956              :                     },
    1957              :                 }))
    1958              :             }
    1959              :             PagestreamFeMessageTag::GetPage => {
    1960              :                 Ok(PagestreamFeMessage::GetPage(PagestreamGetPageRequest {
    1961            1 :                     hdr: PagestreamRequest {
    1962            1 :                         reqid,
    1963            1 :                         request_lsn,
    1964            1 :                         not_modified_since,
    1965            1 :                     },
    1966            1 :                     rel: RelTag {
    1967            1 :                         spcnode: body.read_u32::<BigEndian>()?,
    1968            1 :                         dbnode: body.read_u32::<BigEndian>()?,
    1969            1 :                         relnode: body.read_u32::<BigEndian>()?,
    1970            1 :                         forknum: body.read_u8()?,
    1971              :                     },
    1972            1 :                     blkno: body.read_u32::<BigEndian>()?,
    1973              :                 }))
    1974              :             }
    1975              :             PagestreamFeMessageTag::DbSize => {
    1976              :                 Ok(PagestreamFeMessage::DbSize(PagestreamDbSizeRequest {
    1977            1 :                     hdr: PagestreamRequest {
    1978            1 :                         reqid,
    1979            1 :                         request_lsn,
    1980            1 :                         not_modified_since,
    1981            1 :                     },
    1982            1 :                     dbnode: body.read_u32::<BigEndian>()?,
    1983              :                 }))
    1984              :             }
    1985              :             PagestreamFeMessageTag::GetSlruSegment => Ok(PagestreamFeMessage::GetSlruSegment(
    1986              :                 PagestreamGetSlruSegmentRequest {
    1987            0 :                     hdr: PagestreamRequest {
    1988            0 :                         reqid,
    1989            0 :                         request_lsn,
    1990            0 :                         not_modified_since,
    1991            0 :                     },
    1992            0 :                     kind: body.read_u8()?,
    1993            0 :                     segno: body.read_u32::<BigEndian>()?,
    1994              :                 },
    1995              :             )),
    1996              :             #[cfg(feature = "testing")]
    1997              :             PagestreamFeMessageTag::Test => Ok(PagestreamFeMessage::Test(PagestreamTestRequest {
    1998            0 :                 hdr: PagestreamRequest {
    1999            0 :                     reqid,
    2000            0 :                     request_lsn,
    2001            0 :                     not_modified_since,
    2002            0 :                 },
    2003            0 :                 batch_key: body.read_u64::<BigEndian>()?,
    2004              :                 message: {
    2005            0 :                     let len = body.read_u64::<BigEndian>()?;
    2006            0 :                     let mut buf = vec![0; len as usize];
    2007            0 :                     body.read_exact(&mut buf)?;
    2008            0 :                     String::from_utf8(buf)?
    2009              :                 },
    2010              :             })),
    2011              :         }
    2012            4 :     }
    2013              : }
    2014              : 
    2015              : impl PagestreamBeMessage {
    2016            0 :     pub fn serialize(&self, protocol_version: PagestreamProtocolVersion) -> Bytes {
    2017            0 :         let mut bytes = BytesMut::new();
    2018              : 
    2019              :         use PagestreamBeMessageTag as Tag;
    2020            0 :         match protocol_version {
    2021              :             PagestreamProtocolVersion::V2 => {
    2022            0 :                 match self {
    2023            0 :                     Self::Exists(resp) => {
    2024            0 :                         bytes.put_u8(Tag::Exists as u8);
    2025            0 :                         bytes.put_u8(resp.exists as u8);
    2026            0 :                     }
    2027              : 
    2028            0 :                     Self::Nblocks(resp) => {
    2029            0 :                         bytes.put_u8(Tag::Nblocks as u8);
    2030            0 :                         bytes.put_u32(resp.n_blocks);
    2031            0 :                     }
    2032              : 
    2033            0 :                     Self::GetPage(resp) => {
    2034            0 :                         bytes.put_u8(Tag::GetPage as u8);
    2035            0 :                         bytes.put(&resp.page[..])
    2036              :                     }
    2037              : 
    2038            0 :                     Self::Error(resp) => {
    2039            0 :                         bytes.put_u8(Tag::Error as u8);
    2040            0 :                         bytes.put(resp.message.as_bytes());
    2041            0 :                         bytes.put_u8(0); // null terminator
    2042            0 :                     }
    2043            0 :                     Self::DbSize(resp) => {
    2044            0 :                         bytes.put_u8(Tag::DbSize as u8);
    2045            0 :                         bytes.put_i64(resp.db_size);
    2046            0 :                     }
    2047              : 
    2048            0 :                     Self::GetSlruSegment(resp) => {
    2049            0 :                         bytes.put_u8(Tag::GetSlruSegment as u8);
    2050            0 :                         bytes.put_u32((resp.segment.len() / BLCKSZ as usize) as u32);
    2051            0 :                         bytes.put(&resp.segment[..]);
    2052            0 :                     }
    2053              : 
    2054              :                     #[cfg(feature = "testing")]
    2055            0 :                     Self::Test(resp) => {
    2056            0 :                         bytes.put_u8(Tag::Test as u8);
    2057            0 :                         bytes.put_u64(resp.req.batch_key);
    2058            0 :                         let message = resp.req.message.as_bytes();
    2059            0 :                         bytes.put_u64(message.len() as u64);
    2060            0 :                         bytes.put_slice(message);
    2061            0 :                     }
    2062              :                 }
    2063              :             }
    2064              :             PagestreamProtocolVersion::V3 => {
    2065            0 :                 match self {
    2066            0 :                     Self::Exists(resp) => {
    2067            0 :                         bytes.put_u8(Tag::Exists as u8);
    2068            0 :                         bytes.put_u64(resp.req.hdr.reqid);
    2069            0 :                         bytes.put_u64(resp.req.hdr.request_lsn.0);
    2070            0 :                         bytes.put_u64(resp.req.hdr.not_modified_since.0);
    2071            0 :                         bytes.put_u32(resp.req.rel.spcnode);
    2072            0 :                         bytes.put_u32(resp.req.rel.dbnode);
    2073            0 :                         bytes.put_u32(resp.req.rel.relnode);
    2074            0 :                         bytes.put_u8(resp.req.rel.forknum);
    2075            0 :                         bytes.put_u8(resp.exists as u8);
    2076            0 :                     }
    2077              : 
    2078            0 :                     Self::Nblocks(resp) => {
    2079            0 :                         bytes.put_u8(Tag::Nblocks as u8);
    2080            0 :                         bytes.put_u64(resp.req.hdr.reqid);
    2081            0 :                         bytes.put_u64(resp.req.hdr.request_lsn.0);
    2082            0 :                         bytes.put_u64(resp.req.hdr.not_modified_since.0);
    2083            0 :                         bytes.put_u32(resp.req.rel.spcnode);
    2084            0 :                         bytes.put_u32(resp.req.rel.dbnode);
    2085            0 :                         bytes.put_u32(resp.req.rel.relnode);
    2086            0 :                         bytes.put_u8(resp.req.rel.forknum);
    2087            0 :                         bytes.put_u32(resp.n_blocks);
    2088            0 :                     }
    2089              : 
    2090            0 :                     Self::GetPage(resp) => {
    2091            0 :                         bytes.put_u8(Tag::GetPage as u8);
    2092            0 :                         bytes.put_u64(resp.req.hdr.reqid);
    2093            0 :                         bytes.put_u64(resp.req.hdr.request_lsn.0);
    2094            0 :                         bytes.put_u64(resp.req.hdr.not_modified_since.0);
    2095            0 :                         bytes.put_u32(resp.req.rel.spcnode);
    2096            0 :                         bytes.put_u32(resp.req.rel.dbnode);
    2097            0 :                         bytes.put_u32(resp.req.rel.relnode);
    2098            0 :                         bytes.put_u8(resp.req.rel.forknum);
    2099            0 :                         bytes.put_u32(resp.req.blkno);
    2100            0 :                         bytes.put(&resp.page[..])
    2101              :                     }
    2102              : 
    2103            0 :                     Self::Error(resp) => {
    2104            0 :                         bytes.put_u8(Tag::Error as u8);
    2105            0 :                         bytes.put_u64(resp.req.reqid);
    2106            0 :                         bytes.put_u64(resp.req.request_lsn.0);
    2107            0 :                         bytes.put_u64(resp.req.not_modified_since.0);
    2108            0 :                         bytes.put(resp.message.as_bytes());
    2109            0 :                         bytes.put_u8(0); // null terminator
    2110            0 :                     }
    2111            0 :                     Self::DbSize(resp) => {
    2112            0 :                         bytes.put_u8(Tag::DbSize as u8);
    2113            0 :                         bytes.put_u64(resp.req.hdr.reqid);
    2114            0 :                         bytes.put_u64(resp.req.hdr.request_lsn.0);
    2115            0 :                         bytes.put_u64(resp.req.hdr.not_modified_since.0);
    2116            0 :                         bytes.put_u32(resp.req.dbnode);
    2117            0 :                         bytes.put_i64(resp.db_size);
    2118            0 :                     }
    2119              : 
    2120            0 :                     Self::GetSlruSegment(resp) => {
    2121            0 :                         bytes.put_u8(Tag::GetSlruSegment as u8);
    2122            0 :                         bytes.put_u64(resp.req.hdr.reqid);
    2123            0 :                         bytes.put_u64(resp.req.hdr.request_lsn.0);
    2124            0 :                         bytes.put_u64(resp.req.hdr.not_modified_since.0);
    2125            0 :                         bytes.put_u8(resp.req.kind);
    2126            0 :                         bytes.put_u32(resp.req.segno);
    2127            0 :                         bytes.put_u32((resp.segment.len() / BLCKSZ as usize) as u32);
    2128            0 :                         bytes.put(&resp.segment[..]);
    2129            0 :                     }
    2130              : 
    2131              :                     #[cfg(feature = "testing")]
    2132            0 :                     Self::Test(resp) => {
    2133            0 :                         bytes.put_u8(Tag::Test as u8);
    2134            0 :                         bytes.put_u64(resp.req.hdr.reqid);
    2135            0 :                         bytes.put_u64(resp.req.hdr.request_lsn.0);
    2136            0 :                         bytes.put_u64(resp.req.hdr.not_modified_since.0);
    2137            0 :                         bytes.put_u64(resp.req.batch_key);
    2138            0 :                         let message = resp.req.message.as_bytes();
    2139            0 :                         bytes.put_u64(message.len() as u64);
    2140            0 :                         bytes.put_slice(message);
    2141            0 :                     }
    2142              :                 }
    2143              :             }
    2144              :         }
    2145            0 :         bytes.into()
    2146            0 :     }
    2147              : 
    2148            0 :     pub fn deserialize(buf: Bytes) -> anyhow::Result<Self> {
    2149            0 :         let mut buf = buf.reader();
    2150            0 :         let msg_tag = buf.read_u8()?;
    2151              : 
    2152              :         use PagestreamBeMessageTag as Tag;
    2153            0 :         let ok =
    2154            0 :             match Tag::try_from(msg_tag).map_err(|tag: u8| anyhow::anyhow!("invalid tag {tag}"))? {
    2155              :                 Tag::Exists => {
    2156            0 :                     let reqid = buf.read_u64::<BigEndian>()?;
    2157            0 :                     let request_lsn = Lsn(buf.read_u64::<BigEndian>()?);
    2158            0 :                     let not_modified_since = Lsn(buf.read_u64::<BigEndian>()?);
    2159            0 :                     let rel = RelTag {
    2160            0 :                         spcnode: buf.read_u32::<BigEndian>()?,
    2161            0 :                         dbnode: buf.read_u32::<BigEndian>()?,
    2162            0 :                         relnode: buf.read_u32::<BigEndian>()?,
    2163            0 :                         forknum: buf.read_u8()?,
    2164              :                     };
    2165            0 :                     let exists = buf.read_u8()? != 0;
    2166            0 :                     Self::Exists(PagestreamExistsResponse {
    2167            0 :                         req: PagestreamExistsRequest {
    2168            0 :                             hdr: PagestreamRequest {
    2169            0 :                                 reqid,
    2170            0 :                                 request_lsn,
    2171            0 :                                 not_modified_since,
    2172            0 :                             },
    2173            0 :                             rel,
    2174            0 :                         },
    2175            0 :                         exists,
    2176            0 :                     })
    2177              :                 }
    2178              :                 Tag::Nblocks => {
    2179            0 :                     let reqid = buf.read_u64::<BigEndian>()?;
    2180            0 :                     let request_lsn = Lsn(buf.read_u64::<BigEndian>()?);
    2181            0 :                     let not_modified_since = Lsn(buf.read_u64::<BigEndian>()?);
    2182            0 :                     let rel = RelTag {
    2183            0 :                         spcnode: buf.read_u32::<BigEndian>()?,
    2184            0 :                         dbnode: buf.read_u32::<BigEndian>()?,
    2185            0 :                         relnode: buf.read_u32::<BigEndian>()?,
    2186            0 :                         forknum: buf.read_u8()?,
    2187              :                     };
    2188            0 :                     let n_blocks = buf.read_u32::<BigEndian>()?;
    2189            0 :                     Self::Nblocks(PagestreamNblocksResponse {
    2190            0 :                         req: PagestreamNblocksRequest {
    2191            0 :                             hdr: PagestreamRequest {
    2192            0 :                                 reqid,
    2193            0 :                                 request_lsn,
    2194            0 :                                 not_modified_since,
    2195            0 :                             },
    2196            0 :                             rel,
    2197            0 :                         },
    2198            0 :                         n_blocks,
    2199            0 :                     })
    2200              :                 }
    2201              :                 Tag::GetPage => {
    2202            0 :                     let reqid = buf.read_u64::<BigEndian>()?;
    2203            0 :                     let request_lsn = Lsn(buf.read_u64::<BigEndian>()?);
    2204            0 :                     let not_modified_since = Lsn(buf.read_u64::<BigEndian>()?);
    2205            0 :                     let rel = RelTag {
    2206            0 :                         spcnode: buf.read_u32::<BigEndian>()?,
    2207            0 :                         dbnode: buf.read_u32::<BigEndian>()?,
    2208            0 :                         relnode: buf.read_u32::<BigEndian>()?,
    2209            0 :                         forknum: buf.read_u8()?,
    2210              :                     };
    2211            0 :                     let blkno = buf.read_u32::<BigEndian>()?;
    2212            0 :                     let mut page = vec![0; 8192]; // TODO: use MaybeUninit
    2213            0 :                     buf.read_exact(&mut page)?;
    2214            0 :                     Self::GetPage(PagestreamGetPageResponse {
    2215            0 :                         req: PagestreamGetPageRequest {
    2216            0 :                             hdr: PagestreamRequest {
    2217            0 :                                 reqid,
    2218            0 :                                 request_lsn,
    2219            0 :                                 not_modified_since,
    2220            0 :                             },
    2221            0 :                             rel,
    2222            0 :                             blkno,
    2223            0 :                         },
    2224            0 :                         page: page.into(),
    2225            0 :                     })
    2226              :                 }
    2227              :                 Tag::Error => {
    2228            0 :                     let reqid = buf.read_u64::<BigEndian>()?;
    2229            0 :                     let request_lsn = Lsn(buf.read_u64::<BigEndian>()?);
    2230            0 :                     let not_modified_since = Lsn(buf.read_u64::<BigEndian>()?);
    2231            0 :                     let mut msg = Vec::new();
    2232            0 :                     buf.read_until(0, &mut msg)?;
    2233            0 :                     let cstring = std::ffi::CString::from_vec_with_nul(msg)?;
    2234            0 :                     let rust_str = cstring.to_str()?;
    2235            0 :                     Self::Error(PagestreamErrorResponse {
    2236            0 :                         req: PagestreamRequest {
    2237            0 :                             reqid,
    2238            0 :                             request_lsn,
    2239            0 :                             not_modified_since,
    2240            0 :                         },
    2241            0 :                         message: rust_str.to_owned(),
    2242            0 :                     })
    2243              :                 }
    2244              :                 Tag::DbSize => {
    2245            0 :                     let reqid = buf.read_u64::<BigEndian>()?;
    2246            0 :                     let request_lsn = Lsn(buf.read_u64::<BigEndian>()?);
    2247            0 :                     let not_modified_since = Lsn(buf.read_u64::<BigEndian>()?);
    2248            0 :                     let dbnode = buf.read_u32::<BigEndian>()?;
    2249            0 :                     let db_size = buf.read_i64::<BigEndian>()?;
    2250            0 :                     Self::DbSize(PagestreamDbSizeResponse {
    2251            0 :                         req: PagestreamDbSizeRequest {
    2252            0 :                             hdr: PagestreamRequest {
    2253            0 :                                 reqid,
    2254            0 :                                 request_lsn,
    2255            0 :                                 not_modified_since,
    2256            0 :                             },
    2257            0 :                             dbnode,
    2258            0 :                         },
    2259            0 :                         db_size,
    2260            0 :                     })
    2261              :                 }
    2262              :                 Tag::GetSlruSegment => {
    2263            0 :                     let reqid = buf.read_u64::<BigEndian>()?;
    2264            0 :                     let request_lsn = Lsn(buf.read_u64::<BigEndian>()?);
    2265            0 :                     let not_modified_since = Lsn(buf.read_u64::<BigEndian>()?);
    2266            0 :                     let kind = buf.read_u8()?;
    2267            0 :                     let segno = buf.read_u32::<BigEndian>()?;
    2268            0 :                     let n_blocks = buf.read_u32::<BigEndian>()?;
    2269            0 :                     let mut segment = vec![0; n_blocks as usize * BLCKSZ as usize];
    2270            0 :                     buf.read_exact(&mut segment)?;
    2271            0 :                     Self::GetSlruSegment(PagestreamGetSlruSegmentResponse {
    2272            0 :                         req: PagestreamGetSlruSegmentRequest {
    2273            0 :                             hdr: PagestreamRequest {
    2274            0 :                                 reqid,
    2275            0 :                                 request_lsn,
    2276            0 :                                 not_modified_since,
    2277            0 :                             },
    2278            0 :                             kind,
    2279            0 :                             segno,
    2280            0 :                         },
    2281            0 :                         segment: segment.into(),
    2282            0 :                     })
    2283              :                 }
    2284              :                 #[cfg(feature = "testing")]
    2285              :                 Tag::Test => {
    2286            0 :                     let reqid = buf.read_u64::<BigEndian>()?;
    2287            0 :                     let request_lsn = Lsn(buf.read_u64::<BigEndian>()?);
    2288            0 :                     let not_modified_since = Lsn(buf.read_u64::<BigEndian>()?);
    2289            0 :                     let batch_key = buf.read_u64::<BigEndian>()?;
    2290            0 :                     let len = buf.read_u64::<BigEndian>()?;
    2291            0 :                     let mut msg = vec![0; len as usize];
    2292            0 :                     buf.read_exact(&mut msg)?;
    2293            0 :                     let message = String::from_utf8(msg)?;
    2294            0 :                     Self::Test(PagestreamTestResponse {
    2295            0 :                         req: PagestreamTestRequest {
    2296            0 :                             hdr: PagestreamRequest {
    2297            0 :                                 reqid,
    2298            0 :                                 request_lsn,
    2299            0 :                                 not_modified_since,
    2300            0 :                             },
    2301            0 :                             batch_key,
    2302            0 :                             message,
    2303            0 :                         },
    2304            0 :                     })
    2305              :                 }
    2306              :             };
    2307            0 :         let remaining = buf.into_inner();
    2308            0 :         if !remaining.is_empty() {
    2309            0 :             anyhow::bail!(
    2310            0 :                 "remaining bytes in msg with tag={msg_tag}: {}",
    2311            0 :                 remaining.len()
    2312            0 :             );
    2313            0 :         }
    2314            0 :         Ok(ok)
    2315            0 :     }
    2316              : 
    2317            0 :     pub fn kind(&self) -> &'static str {
    2318            0 :         match self {
    2319            0 :             Self::Exists(_) => "Exists",
    2320            0 :             Self::Nblocks(_) => "Nblocks",
    2321            0 :             Self::GetPage(_) => "GetPage",
    2322            0 :             Self::Error(_) => "Error",
    2323            0 :             Self::DbSize(_) => "DbSize",
    2324            0 :             Self::GetSlruSegment(_) => "GetSlruSegment",
    2325              :             #[cfg(feature = "testing")]
    2326            0 :             Self::Test(_) => "Test",
    2327              :         }
    2328            0 :     }
    2329              : }
    2330              : 
    2331            0 : #[derive(Debug, Serialize, Deserialize)]
    2332              : pub struct PageTraceEvent {
    2333              :     pub key: CompactKey,
    2334              :     pub effective_lsn: Lsn,
    2335              :     pub time: SystemTime,
    2336              : }
    2337              : 
    2338              : impl Default for PageTraceEvent {
    2339            0 :     fn default() -> Self {
    2340            0 :         Self {
    2341            0 :             key: Default::default(),
    2342            0 :             effective_lsn: Default::default(),
    2343            0 :             time: std::time::UNIX_EPOCH,
    2344            0 :         }
    2345            0 :     }
    2346              : }
    2347              : 
    2348              : #[cfg(test)]
    2349              : mod tests {
    2350              :     use std::str::FromStr;
    2351              : 
    2352              :     use serde_json::json;
    2353              : 
    2354              :     use super::*;
    2355              : 
    2356              :     #[test]
    2357            1 :     fn test_pagestream() {
    2358            1 :         // Test serialization/deserialization of PagestreamFeMessage
    2359            1 :         let messages = vec![
    2360            1 :             PagestreamFeMessage::Exists(PagestreamExistsRequest {
    2361            1 :                 hdr: PagestreamRequest {
    2362            1 :                     reqid: 0,
    2363            1 :                     request_lsn: Lsn(4),
    2364            1 :                     not_modified_since: Lsn(3),
    2365            1 :                 },
    2366            1 :                 rel: RelTag {
    2367            1 :                     forknum: 1,
    2368            1 :                     spcnode: 2,
    2369            1 :                     dbnode: 3,
    2370            1 :                     relnode: 4,
    2371            1 :                 },
    2372            1 :             }),
    2373            1 :             PagestreamFeMessage::Nblocks(PagestreamNblocksRequest {
    2374            1 :                 hdr: PagestreamRequest {
    2375            1 :                     reqid: 0,
    2376            1 :                     request_lsn: Lsn(4),
    2377            1 :                     not_modified_since: Lsn(4),
    2378            1 :                 },
    2379            1 :                 rel: RelTag {
    2380            1 :                     forknum: 1,
    2381            1 :                     spcnode: 2,
    2382            1 :                     dbnode: 3,
    2383            1 :                     relnode: 4,
    2384            1 :                 },
    2385            1 :             }),
    2386            1 :             PagestreamFeMessage::GetPage(PagestreamGetPageRequest {
    2387            1 :                 hdr: PagestreamRequest {
    2388            1 :                     reqid: 0,
    2389            1 :                     request_lsn: Lsn(4),
    2390            1 :                     not_modified_since: Lsn(3),
    2391            1 :                 },
    2392            1 :                 rel: RelTag {
    2393            1 :                     forknum: 1,
    2394            1 :                     spcnode: 2,
    2395            1 :                     dbnode: 3,
    2396            1 :                     relnode: 4,
    2397            1 :                 },
    2398            1 :                 blkno: 7,
    2399            1 :             }),
    2400            1 :             PagestreamFeMessage::DbSize(PagestreamDbSizeRequest {
    2401            1 :                 hdr: PagestreamRequest {
    2402            1 :                     reqid: 0,
    2403            1 :                     request_lsn: Lsn(4),
    2404            1 :                     not_modified_since: Lsn(3),
    2405            1 :                 },
    2406            1 :                 dbnode: 7,
    2407            1 :             }),
    2408            1 :         ];
    2409            5 :         for msg in messages {
    2410            4 :             let bytes = msg.serialize();
    2411            4 :             let reconstructed =
    2412            4 :                 PagestreamFeMessage::parse(&mut bytes.reader(), PagestreamProtocolVersion::V3)
    2413            4 :                     .unwrap();
    2414            4 :             assert!(msg == reconstructed);
    2415              :         }
    2416            1 :     }
    2417              : 
    2418              :     #[test]
    2419            1 :     fn test_tenantinfo_serde() {
    2420            1 :         // Test serialization/deserialization of TenantInfo
    2421            1 :         let original_active = TenantInfo {
    2422            1 :             id: TenantShardId::unsharded(TenantId::generate()),
    2423            1 :             state: TenantState::Active,
    2424            1 :             current_physical_size: Some(42),
    2425            1 :             attachment_status: TenantAttachmentStatus::Attached,
    2426            1 :             generation: 1,
    2427            1 :             gc_blocking: None,
    2428            1 :         };
    2429            1 :         let expected_active = json!({
    2430            1 :             "id": original_active.id.to_string(),
    2431            1 :             "state": {
    2432            1 :                 "slug": "Active",
    2433            1 :             },
    2434            1 :             "current_physical_size": 42,
    2435            1 :             "attachment_status": {
    2436            1 :                 "slug":"attached",
    2437            1 :             },
    2438            1 :             "generation" : 1
    2439            1 :         });
    2440            1 : 
    2441            1 :         let original_broken = TenantInfo {
    2442            1 :             id: TenantShardId::unsharded(TenantId::generate()),
    2443            1 :             state: TenantState::Broken {
    2444            1 :                 reason: "reason".into(),
    2445            1 :                 backtrace: "backtrace info".into(),
    2446            1 :             },
    2447            1 :             current_physical_size: Some(42),
    2448            1 :             attachment_status: TenantAttachmentStatus::Attached,
    2449            1 :             generation: 1,
    2450            1 :             gc_blocking: None,
    2451            1 :         };
    2452            1 :         let expected_broken = json!({
    2453            1 :             "id": original_broken.id.to_string(),
    2454            1 :             "state": {
    2455            1 :                 "slug": "Broken",
    2456            1 :                 "data": {
    2457            1 :                     "backtrace": "backtrace info",
    2458            1 :                     "reason": "reason",
    2459            1 :                 }
    2460            1 :             },
    2461            1 :             "current_physical_size": 42,
    2462            1 :             "attachment_status": {
    2463            1 :                 "slug":"attached",
    2464            1 :             },
    2465            1 :             "generation" : 1
    2466            1 :         });
    2467            1 : 
    2468            1 :         assert_eq!(
    2469            1 :             serde_json::to_value(&original_active).unwrap(),
    2470            1 :             expected_active
    2471            1 :         );
    2472              : 
    2473            1 :         assert_eq!(
    2474            1 :             serde_json::to_value(&original_broken).unwrap(),
    2475            1 :             expected_broken
    2476            1 :         );
    2477            1 :         assert!(format!("{:?}", &original_broken.state).contains("reason"));
    2478            1 :         assert!(format!("{:?}", &original_broken.state).contains("backtrace info"));
    2479            1 :     }
    2480              : 
    2481              :     #[test]
    2482            1 :     fn test_reject_unknown_field() {
    2483            1 :         let id = TenantId::generate();
    2484            1 :         let config_request = json!({
    2485            1 :             "tenant_id": id.to_string(),
    2486            1 :             "unknown_field": "unknown_value".to_string(),
    2487            1 :         });
    2488            1 :         let err = serde_json::from_value::<TenantConfigRequest>(config_request).unwrap_err();
    2489            1 :         assert!(
    2490            1 :             err.to_string().contains("unknown field `unknown_field`"),
    2491            0 :             "expect unknown field `unknown_field` error, got: {}",
    2492              :             err
    2493              :         );
    2494            1 :     }
    2495              : 
    2496              :     #[test]
    2497            1 :     fn tenantstatus_activating_serde() {
    2498            1 :         let states = [TenantState::Activating(ActivatingFrom::Attaching)];
    2499            1 :         let expected = "[{\"slug\":\"Activating\",\"data\":\"Attaching\"}]";
    2500            1 : 
    2501            1 :         let actual = serde_json::to_string(&states).unwrap();
    2502            1 : 
    2503            1 :         assert_eq!(actual, expected);
    2504              : 
    2505            1 :         let parsed = serde_json::from_str::<Vec<TenantState>>(&actual).unwrap();
    2506            1 : 
    2507            1 :         assert_eq!(states.as_slice(), &parsed);
    2508            1 :     }
    2509              : 
    2510              :     #[test]
    2511            1 :     fn tenantstatus_activating_strum() {
    2512            1 :         // tests added, because we use these for metrics
    2513            1 :         let examples = [
    2514            1 :             (line!(), TenantState::Attaching, "Attaching"),
    2515            1 :             (
    2516            1 :                 line!(),
    2517            1 :                 TenantState::Activating(ActivatingFrom::Attaching),
    2518            1 :                 "Activating",
    2519            1 :             ),
    2520            1 :             (line!(), TenantState::Active, "Active"),
    2521            1 :             (
    2522            1 :                 line!(),
    2523            1 :                 TenantState::Stopping {
    2524            1 :                     progress: utils::completion::Barrier::default(),
    2525            1 :                 },
    2526            1 :                 "Stopping",
    2527            1 :             ),
    2528            1 :             (
    2529            1 :                 line!(),
    2530            1 :                 TenantState::Broken {
    2531            1 :                     reason: "Example".into(),
    2532            1 :                     backtrace: "Looooong backtrace".into(),
    2533            1 :                 },
    2534            1 :                 "Broken",
    2535            1 :             ),
    2536            1 :         ];
    2537              : 
    2538            6 :         for (line, rendered, expected) in examples {
    2539            5 :             let actual: &'static str = rendered.into();
    2540            5 :             assert_eq!(actual, expected, "example on {line}");
    2541              :         }
    2542            1 :     }
    2543              : 
    2544              :     #[test]
    2545            1 :     fn test_image_compression_algorithm_parsing() {
    2546              :         use ImageCompressionAlgorithm::*;
    2547            1 :         let cases = [
    2548            1 :             ("disabled", Disabled),
    2549            1 :             ("zstd", Zstd { level: None }),
    2550            1 :             ("zstd(18)", Zstd { level: Some(18) }),
    2551            1 :             ("zstd(-3)", Zstd { level: Some(-3) }),
    2552            1 :         ];
    2553              : 
    2554            5 :         for (display, expected) in cases {
    2555            4 :             assert_eq!(
    2556            4 :                 ImageCompressionAlgorithm::from_str(display).unwrap(),
    2557              :                 expected,
    2558            0 :                 "parsing works"
    2559              :             );
    2560            4 :             assert_eq!(format!("{expected}"), display, "Display FromStr roundtrip");
    2561              : 
    2562            4 :             let ser = serde_json::to_string(&expected).expect("serialization");
    2563            4 :             assert_eq!(
    2564            4 :                 serde_json::from_str::<ImageCompressionAlgorithm>(&ser).unwrap(),
    2565              :                 expected,
    2566            0 :                 "serde roundtrip"
    2567              :             );
    2568              : 
    2569            4 :             assert_eq!(
    2570            4 :                 serde_json::Value::String(display.to_string()),
    2571            4 :                 serde_json::to_value(expected).unwrap(),
    2572            0 :                 "Display is the serde serialization"
    2573              :             );
    2574              :         }
    2575            1 :     }
    2576              : 
    2577              :     #[test]
    2578            1 :     fn test_tenant_config_patch_request_serde() {
    2579            1 :         let patch_request = TenantConfigPatchRequest {
    2580            1 :             tenant_id: TenantId::from_str("17c6d121946a61e5ab0fe5a2fd4d8215").unwrap(),
    2581            1 :             config: TenantConfigPatch {
    2582            1 :                 checkpoint_distance: FieldPatch::Upsert(42),
    2583            1 :                 gc_horizon: FieldPatch::Remove,
    2584            1 :                 compaction_threshold: FieldPatch::Noop,
    2585            1 :                 ..TenantConfigPatch::default()
    2586            1 :             },
    2587            1 :         };
    2588            1 : 
    2589            1 :         let json = serde_json::to_string(&patch_request).unwrap();
    2590            1 : 
    2591            1 :         let expected = r#"{"tenant_id":"17c6d121946a61e5ab0fe5a2fd4d8215","checkpoint_distance":42,"gc_horizon":null}"#;
    2592            1 :         assert_eq!(json, expected);
    2593              : 
    2594            1 :         let decoded: TenantConfigPatchRequest = serde_json::from_str(&json).unwrap();
    2595            1 :         assert_eq!(decoded.tenant_id, patch_request.tenant_id);
    2596            1 :         assert_eq!(decoded.config, patch_request.config);
    2597              : 
    2598              :         // Now apply the patch to a config to demonstrate semantics
    2599              : 
    2600            1 :         let base = TenantConfig {
    2601            1 :             checkpoint_distance: Some(28),
    2602            1 :             gc_horizon: Some(100),
    2603            1 :             compaction_target_size: Some(1024),
    2604            1 :             ..Default::default()
    2605            1 :         };
    2606            1 : 
    2607            1 :         let expected = TenantConfig {
    2608            1 :             checkpoint_distance: Some(42),
    2609            1 :             gc_horizon: None,
    2610            1 :             ..base.clone()
    2611            1 :         };
    2612            1 : 
    2613            1 :         let patched = base.apply_patch(decoded.config).unwrap();
    2614            1 : 
    2615            1 :         assert_eq!(patched, expected);
    2616            1 :     }
    2617              : }
        

Generated by: LCOV version 2.1-beta