Prevent LFC overflow by skipping write when not unpinned pages are available

.github/workflows/build_and_test.yml

@@ -1078,6 +1078,12 @@ jobs:
               console.log(`Tag ${tag} created successfully.`);
             }
 
+            // TODO: check how GitHub releases looks for proxy/compute releases and enable them if they're ok
+            if (context.ref !== 'refs/heads/release') {
+              console.log(`GitHub release skipped for ${context.ref}.`);
+              return;
+            }
+
             try {
               const existingRelease = await github.rest.repos.getReleaseByTag({
                 owner: context.repo.owner,
@@ -1096,8 +1102,7 @@ jobs:
                 owner: context.repo.owner,
                 repo: context.repo.repo,
                 tag_name: tag,
-                // TODO: Automate release notes properly
-                generate_release_notes: false,
+                generate_release_notes: true,
               });
               console.log(`Release for tag ${tag} created successfully.`);
             }

compute_tools/src/bin/fast_import.rs

@@ -58,8 +58,6 @@ struct Args {
     pg_bin_dir: Utf8PathBuf,
     #[clap(long)]
     pg_lib_dir: Utf8PathBuf,
-    #[clap(long)]
-    pg_port: Option<u16>, // port to run postgres on, 5432 is default
 }
 
 #[serde_with::serde_as]
@@ -76,13 +74,6 @@ enum EncryptionSecret {
     KMS { key_id: String },
 }
 
-// copied from pageserver_api::config::defaults::DEFAULT_LOCALE to avoid dependency just for a constant
-const DEFAULT_LOCALE: &str = if cfg!(target_os = "macos") {
-    "C"
-} else {
-    "C.UTF-8"
-};
-
 #[tokio::main]
 pub(crate) async fn main() -> anyhow::Result<()> {
     utils::logging::init(
@@ -106,10 +97,6 @@ pub(crate) async fn main() -> anyhow::Result<()> {
     let working_directory = args.working_directory;
     let pg_bin_dir = args.pg_bin_dir;
     let pg_lib_dir = args.pg_lib_dir;
-    let pg_port = args.pg_port.unwrap_or_else(|| {
-        info!("pg_port not specified, using default 5432");
-        5432
-    });
 
     // Initialize AWS clients only if s3_prefix is specified
     let (aws_config, kms_client) = if args.s3_prefix.is_some() {
@@ -193,7 +180,7 @@ pub(crate) async fn main() -> anyhow::Result<()> {
     let superuser = "cloud_admin"; // XXX: this shouldn't be hard-coded
     postgres_initdb::do_run_initdb(postgres_initdb::RunInitdbArgs {
         superuser,
-        locale: DEFAULT_LOCALE, // XXX: this shouldn't be hard-coded,
+        locale: "en_US.UTF-8", // XXX: this shouldn't be hard-coded,
         pg_version,
         initdb_bin: pg_bin_dir.join("initdb").as_ref(),
         library_search_path: &pg_lib_dir, // TODO: is this right? Prob works in compute image, not sure about neon_local.
@@ -210,7 +197,6 @@ pub(crate) async fn main() -> anyhow::Result<()> {
     let mut postgres_proc = tokio::process::Command::new(pgbin)
         .arg("-D")
         .arg(&pgdata_dir)
-        .args(["-p", &format!("{pg_port}")])
         .args(["-c", "wal_level=minimal"])
         .args(["-c", "shared_buffers=10GB"])
         .args(["-c", "max_wal_senders=0"])
@@ -230,7 +216,6 @@ pub(crate) async fn main() -> anyhow::Result<()> {
             ),
         ])
         .env_clear()
-        .env("LD_LIBRARY_PATH", &pg_lib_dir)
         .stdout(std::process::Stdio::piped())
         .stderr(std::process::Stdio::piped())
         .spawn()
@@ -247,7 +232,7 @@ pub(crate) async fn main() -> anyhow::Result<()> {
 
     // Create neondb database in the running postgres
     let restore_pg_connstring =
-        format!("host=localhost port={pg_port} user={superuser} dbname=postgres");
+        format!("host=localhost port=5432 user={superuser} dbname=postgres");
 
     let start_time = std::time::Instant::now();
 
@@ -329,7 +314,6 @@ pub(crate) async fn main() -> anyhow::Result<()> {
             .arg(&source_connection_string)
             // how we run it
             .env_clear()
-            .env("LD_LIBRARY_PATH", &pg_lib_dir)
             .kill_on_drop(true)
             .stdout(std::process::Stdio::piped())
             .stderr(std::process::Stdio::piped())
@@ -363,7 +347,6 @@ pub(crate) async fn main() -> anyhow::Result<()> {
             .arg(&dumpdir)
             // how we run it
             .env_clear()
-            .env("LD_LIBRARY_PATH", &pg_lib_dir)
             .kill_on_drop(true)
             .stdout(std::process::Stdio::piped())
             .stderr(std::process::Stdio::piped())

control_plane/src/pageserver.rs

@@ -418,11 +418,6 @@ impl PageServerNode {
                 .map(serde_json::from_str)
                 .transpose()
                 .context("parse `wal_receiver_protocol_override` from json")?,
-            rel_size_v2_enabled: settings
-                .remove("rel_size_v2_enabled")
-                .map(|x| x.parse::<bool>())
-                .transpose()
-                .context("Failed to parse 'rel_size_v2_enabled' as bool")?,
         };
         if !settings.is_empty() {
             bail!("Unrecognized tenant settings: {settings:?}")

docs/rfcs/041-sharded-ingest.md

@@ -1,255 +0,0 @@
-# 
-Created on Aug 2024
-Implemented on Jan 2025
-
-## Summary
-
-Data in large tenants is split up between multiple pageservers according to key hashes, as
-introduced in the [sharding RFC](031-sharding-static.md) and [shard splitting RFC](032-shard-splitting.md).
-
-Whereas currently we send all WAL to all pageserver shards, and each shard filters out the data that it needs,
-in this RFC we add a mechanism to filter the WAL on the safekeeper, so that each shard receives
-only the data it needs.
-
-This will place some extra CPU load on the safekeepers, in exchange for reducing the network bandwidth
-for ingesting WAL back to scaling as O(1) with shard count, rather than O(N_shards).
-
-## Motivation
-
-1. Large databases require higher shard counts.  Whereas currently we run with up to 8 shards for tenants
-with a few TB of storage, the next order of magnitude capacity increase will require tens of shards, such
-that sending all WAL to all shards is impractical in terms of bandwidth.
-2. For contemporary database sizes (~2TB), the pageserver is the bottleneck for ingest: since each
-   shard has to decode and process the whole WAL, sharding doesn't fully relieve this bottleneck.  To achieve significantly higher ingest speeds, we need to filter the WAL earlier so that each pageserver
-   only has to process relevant parts.
-
-## Non Goals (if relevant)
-
-We do not seek to introduce multiple WALs per timeline, or to share the work of handling a timeline's
-WAL across safekeepers (beyond simple 3x replication).  This RFC may be thought of as an incremental
-move of the ingestion bottleneck up the stack: instead of high write rates bottlenecking on the
-pageserver, they will bottleneck on the safekeeper.
-
-## Impacted components (e.g. pageserver, safekeeper, console, etc)
-
-Safekeeper, pageserver.
-
-There will be no control plane or storage controller coordination needed, as pageservers will directly
-indicate their sharding parameters to the safekeeper when subscribing for WAL.
-
-## Proposed implementation
-
-Terminology:
-- "Data pages" refers to postgres relation blocks, and SLRU blocks.
-- "Metadata pages" refers to everything else the pageserver stores, such as relation sizes and
-  directories of relations.
-
-### Phase 1: Refactor ingest
-
-Currently, pageserver ingest code is structured approximately as follows:
-1. `handle_walreceiver_connection` reads a stream of binary WAL records off a network
-   socket
-2. `WalIngest::ingest_record` to translate the record into a series of page-level modifications
-3. `DatadirModification` accumulates page updates from several `ingest_record` calls, and when
-   its `commit()` method is called, flushes these into a Timeline's open `InMemoryLayer`.
-
-This process currently assumes access to a pageserver `Timeline` throughout `ingest_record` and
-from `DatadirModification`, which is used to do read-modify-write cycles on metadata pages
-such as relation sizes and the master DBDIR page.  It also assumes that records are ingested
-strictly one after the other: they cannot be ingested in parallel because each record assumes
-that earlier records' changes have already been applied to `Timeline`.
-
-This code will be refactored to disentangle the simple, fast decode of relation page writes
-from the more complex logic for updating internal metadata.  An intermediate representation
-called `InterpretedWalRecords` will be introduced.  This is similar to the internal state of
-a `DatadirModification`, but does not require access to a Timeline.  Instead of storing
-metadata updates as materialized writes to pages, it will accumulate these as abstract operations,
-for example rather than including a write to a relation size key, this structure will include
-an operation that indicates "Update relation _foo_'s size to the max of its current value and
-_bar_", such that these may be applied later to a real Timeline.
-
-The `DatadirModification` will be aware of the `EphemeralFile` format, so that as it accumulates
-simple page writes of relation blocks, it can write them directly into a buffer in the serialized
-format.  This will avoid the need to later deserialize/reserialize this data when passing the
-structure between safekeeper and pageserver.
-
-The new pipeline will be:
-1. `handle_walreceiver_connection` reads a stream of binary WAL records off a network
-2. A `InterpretedWalRecords` is generated from the incoming WAL records.  This does not
-   require a reference to a Timeline.
-3. The logic that is current spread between `WalIngest` and `DatadirModification` for updating
-   metadata will be refactored to consume the metadata operations from the `InterpretedWalRecords`
-   and turn them into literal writes to metadata pages.  This part must be done sequentially.
-4. The resulting buffer of metadata page writes is combined with the buffer of relation block
-   writes, and written into the `InMemoryLayer`.
-
-Implemented in:
-1. https://github.com/neondatabase/neon/pull/9472
-2. https://github.com/neondatabase/neon/pull/9504
-3. https://github.com/neondatabase/neon/pull/9524
-
-### Phase 2: Decode & filter on safekeeper
-
-In the previous phase, the ingest code was modified to be able to do most of its work without access to
-a Timeline: this first stage of ingest simply converts a series of binary wal records into
-a buffer of relation/SLRU page writes, and a buffer of abstract metadata writes.
-
-The modified ingest code may be transplanted from pageserver to safekeeper (probably via a
-shared crate).  The safekeeper->pageserver network protocol is modified to:
- - in subscription requests, send the `ShardIdentity` from the pageserver to the safekeeper
- - in responses, transmit a `InterpretedWalRecords` instead of a raw `WalRecord`.
- - use the `ShardIdentity` to filter the `ProcessedWalIngest` to relevant content for
-   the subscribing shard before transmitting it.
-
-The overall behavior of the pageserver->safekeeper interaction remains the same, in terms of
-consistent LSN feedback, and connection management.  Only the payload of the subscriptions
-changes, to express an LSN range of WAL as a filtered `ProcessedWalIngest` instead of the
-raw data.
-
-The ingest code on the pageserver can now skip the part where it does the first phase of
-processing, as it will receive pre-processed, compressed data off the wire.
-
-Note that `InterpretedWalRecord` batches multiple `InterpretedWalRecord(s)` in the same network
-message. Safekeeper reads WAL in chunks of 16 blocks and then decodes as many Postgres WAL records
-as possible. Each Postgres WAL record maps to one `InterpretedWalRecord` for potentially multiple shards.
-Hence, the size of the batch is given by the number of Postgres WAL records that fit in 16 blocks.
-
-The protocol needs to support evolution. Protobuf was chosen here with the view that, in the future,
-we may migrate it to GRPC altogether
-
-Implemented in:
-1. https://github.com/neondatabase/neon/pull/9746
-2. https://github.com/neondatabase/neon/pull/9821
-
-### Phase 3: Fan out interpreted WAL
-
-In the previous phase, the initial processing of WAL was moved to the safekeeper, but it is still
-done once for each shard: this will generate O(N_shards) CPU work on the safekeeper (especially
-when considering converting to Protobuf format and compression).
-
-To avoid this, we fan-out WAL from one (tenant, timeline, shard) to all other shards subscribed on
-the same safekeeper. Under normal operation, the WAL will be read from disk, decoded and interpreted
-_only_ once per (safekeeper, timeline).
-
-When the first shard of a sharded timeline subscribes to a given safekeeper a task is spawned
-for the WAL reader (`InterpretedWalReader`). This task reads WAL, decodes, interprets it and sends
-it to the sender (`InterpretedWalSender`). The sender is a future that is polled from the connection
-task. When further shards subscribe on the safekeeper they will attach themselves to the existing WAL reader.
-There's two cases to consider:
-1. The shard's requested `start_lsn` is ahead of the current position of the WAL reader. In this case, the shard
-will start receiving data when the reader reaches that LSN. The intuition here is that there's little to gain
-by letting shards "front-run" since compute backpressure is based on the laggard LSN.
-2. The shard's requested `start_lsn` is below the current position of the WAL reader. In this case, the WAL reader
-gets reset to this requested position (same intuition). Special care is taken such that advanced shards do not receive
-interpreted WAL records below their current position.
-
-The approach above implies that there is at most one WAL reader per (tenant, timeline) on a given safekeeper at any point in time.
-If this turns out to be operationally problematic, there's a trick we can deploy: `--max-delta-for-fanout` is an optional safekeeper
-argument that controls the max absolute delta between a new shard and the current WAL position of the WAL reader. If the absolute
-delta is above that value, a new reader is spawned. Note that there's currently no concurrency control on the number of WAL readers,
-so it's recommended to use large values to avoid pushing CPU utilisation too high.
-
-Unsharded tenants do not spawn a separate task for the interpreted WAL reader since there's no benefit to it. Instead they poll
-the reader and sender concurrently from the connection task.
-
-Shard splits are interesting here because it is the only case when the same shard might have two subscriptions at the same time.
-This is handled by giving readers a unique identifier. Both shards will receive the same data while respecting their requested start
-position.
-
-Implemented in:
-1. https://github.com/neondatabase/neon/pull/10190
-
-## Deployment
-
-Each phase shall be deployed independently. Special care should be taken around protocol changes.
-
-## Observability Tips
-
-* The safekeeper logs the protocol requested by the pageserver
-along with the pageserver ID, tenant, timeline and shard: `starting streaming from`.
-* There's metrics for the number of wal readers:
-  * `safekeeper_wal_readers{kind="task", target=~"pageserver.*"}` gives the number of wal reader tasks for each SK
-  * `safekeeper_wal_readers{kind="future", target=~"pageserver.*"}` gives the numer of wal readers polled inline by each SK
-  * `safekeeper_interpreted_wal_reader_tasks` gives the number of wal reader tasks per tenant, timeline
-* Interesting log lines for the fan-out reader:
-  * `Spawning interpreted`: first shard creates the interpreted wal reader
-  * `Fanning out`: a subsequent shard attaches itself to an interpreted wal reader
-  * `Aborting interpreted`: all senders have finished and the reader task is being aborted
-
-## Future Optimizations
-
-This sections describes some improvement areas which may be revisited in the future.
-
-### Buffering of Interpreted WAL
-
-The interpreted WAL reader may buffer interpreted WAL records in user space to help with serving
-subscriptions that are lagging behind the current position of the reader.
-
-Counterpoints:
-* Safekeepers serve many thousands of timelines and allocating a buffer for each might be wasteful,
-especially given that it would go unused on the happy path.
-* WAL is buffered in the kernel page cache. Usually we'd only pay the CPU cost of decoding and interpreting.
-
-### Tweaking the Pagserver Safekeeper Selection Algorithm
-
-We could make the pageserver aware of which safekeeper's already host shards for the timeline along
-with their current WAL positions. The pageserver should then prefer safkeepers that are in the same
-AZ _and_ already have a shard with a position close to the desired start position.
-
-We currently run one safekeeper per AZ, so the point is mute until that changes.
-
-### Pipelining first ingest phase
-
-The first ingest phase is a stateless transformation of a binary WAL record into a pre-processed
-output per shard.  To put multiple CPUs to work, we may pipeline this processing up to some defined buffer
-depth.
-
-## Alternatives considered
-
-### Give safekeepers enough state to fully decode WAL
-
-In this RFC, we only do the first phase of ingest on the safekeeper, because this is
-the phase that is stateless.  Subsequent changes then happen on the pageserver, with
-access to the `Timeline` state.
-
-We could do more work on the safekeeper if we transmitted metadata state to the safekeeper
-when subscribing to the WAL: for example, by telling the safekeeper all the relation sizes,
-so that it could then generate all the metadata writes for relation sizes.
-
-We avoid doing this for several reasons:
-1. Complexity: it's a more invasive protocol change
-2. Decoupling: having the safekeeper understand the `ProcessedWalIngest` already somewhat
-   infects it with knowledge of the pageserver, but this is mainly an abstract structure
-   that describes postgres writes.  However, if we taught the safekeeper about the exact
-   way that pageserver deals with metadata keys, this would be a much tighter coupling.
-3. Load: once the WAL has been processed to the point that it can be split between shards,
-   it is preferable to share out work on the remaining shards rather than adding extra CPU
-   load to the safekeeper.
-
-### Do pre-processing on the compute instead of the safekeeper
-
-Since our first stage of ingest is stateless, it could be done at any stage in the pipeline,
-all the way up to the compute.
-
-We choose not to do this, because it is useful for the safekeeper to store the raw WAL rather
-than just the preprocessed WAL:
-- The safekeeper still needs to be able to serve raw WAL back to postgres for e.g. physical replication
-- It simplifies our paxos implementation to have the offset in the write log be literally
-  the same as the LSN
-- Raw WAL must have a stable protocol since we might have to re-ingest it at arbitrary points in the future.
-  Storing raw WAL give us more flexibility to evolve the pageserver, safekeeper protocol.
-
-### Do wal pre-processing on shard 0 or a separate service, send it to other shards from there
-
-If we wanted to keep the safekeepers as entirely pure stores of raw WAL bytes, then
-we could do the initial decode and shard-splitting in some other location:
-- Shard 0 could subscribe to the full WAL and then send writes to other shards
-- A new intermediate service between the safekeeper and pageserver could do the splitting.
-
-So why not?
-- Extra network hop from shard 0 to the final destination shard
-- Clearly there is more infrastructure involved here compared with doing it inline on the safekeeper.
-- Safekeepers already have very light CPU load: typical cloud instances shapes with appropriate
-  disks for the safekeepers effectively have "free" CPU resources.
-- Doing extra work on shard 0 would complicate scheduling of shards on pageservers, because
-  shard 0 would have significantly higher CPU load under write workloads than other shards.

libs/pageserver_api/src/config.rs

@@ -301,10 +301,6 @@ pub struct TenantConfigToml {
     pub timeline_offloading: bool,
 
     pub wal_receiver_protocol_override: Option<PostgresClientProtocol>,
-
-    /// Enable rel_size_v2 for this tenant. Once enabled, the tenant will persist this information into
-    /// `index_part.json`, and it cannot be reversed.
-    pub rel_size_v2_enabled: Option<bool>,
 }
 
 pub mod defaults {
@@ -542,7 +538,6 @@ impl Default for TenantConfigToml {
             lsn_lease_length_for_ts: LsnLease::DEFAULT_LENGTH_FOR_TS,
             timeline_offloading: false,
             wal_receiver_protocol_override: None,
-            rel_size_v2_enabled: None,
         }
     }
 }

libs/pageserver_api/src/models.rs

@@ -497,8 +497,6 @@ pub struct TenantConfigPatch {
     pub timeline_offloading: FieldPatch<bool>,
     #[serde(skip_serializing_if = "FieldPatch::is_noop")]
     pub wal_receiver_protocol_override: FieldPatch<PostgresClientProtocol>,
-    #[serde(skip_serializing_if = "FieldPatch::is_noop")]
-    pub rel_size_v2_enabled: FieldPatch<bool>,
 }
 
 /// An alternative representation of `pageserver::tenant::TenantConf` with
@@ -530,7 +528,6 @@ pub struct TenantConfig {
     pub lsn_lease_length_for_ts: Option<String>,
     pub timeline_offloading: Option<bool>,
     pub wal_receiver_protocol_override: Option<PostgresClientProtocol>,
-    pub rel_size_v2_enabled: Option<bool>,
 }
 
 impl TenantConfig {
@@ -560,7 +557,6 @@ impl TenantConfig {
             mut lsn_lease_length_for_ts,
             mut timeline_offloading,
             mut wal_receiver_protocol_override,
-            mut rel_size_v2_enabled,
         } = self;
 
         patch.checkpoint_distance.apply(&mut checkpoint_distance);
@@ -605,7 +601,6 @@ impl TenantConfig {
         patch
             .wal_receiver_protocol_override
             .apply(&mut wal_receiver_protocol_override);
-        patch.rel_size_v2_enabled.apply(&mut rel_size_v2_enabled);
 
         Self {
             checkpoint_distance,
@@ -632,7 +627,6 @@ impl TenantConfig {
             lsn_lease_length_for_ts,
             timeline_offloading,
             wal_receiver_protocol_override,
-            rel_size_v2_enabled,
         }
     }
 }

pageserver/src/tenant.rs

@@ -5475,7 +5475,6 @@ pub(crate) mod harness {
                 lsn_lease_length_for_ts: Some(tenant_conf.lsn_lease_length_for_ts),
                 timeline_offloading: Some(tenant_conf.timeline_offloading),
                 wal_receiver_protocol_override: tenant_conf.wal_receiver_protocol_override,
-                rel_size_v2_enabled: tenant_conf.rel_size_v2_enabled,
             }
         }
     }

pageserver/src/tenant/config.rs

@@ -357,9 +357,6 @@ pub struct TenantConfOpt {
 
     #[serde(skip_serializing_if = "Option::is_none")]
     pub wal_receiver_protocol_override: Option<PostgresClientProtocol>,
-
-    #[serde(skip_serializing_if = "Option::is_none")]
-    pub rel_size_v2_enabled: Option<bool>,
 }
 
 impl TenantConfOpt {
@@ -428,7 +425,6 @@ impl TenantConfOpt {
             wal_receiver_protocol_override: self
                 .wal_receiver_protocol_override
                 .or(global_conf.wal_receiver_protocol_override),
-            rel_size_v2_enabled: self.rel_size_v2_enabled.or(global_conf.rel_size_v2_enabled),
         }
     }
 
@@ -458,7 +454,6 @@ impl TenantConfOpt {
             mut lsn_lease_length_for_ts,
             mut timeline_offloading,
             mut wal_receiver_protocol_override,
-            mut rel_size_v2_enabled,
         } = self;
 
         patch.checkpoint_distance.apply(&mut checkpoint_distance);
@@ -527,7 +522,6 @@ impl TenantConfOpt {
         patch
             .wal_receiver_protocol_override
             .apply(&mut wal_receiver_protocol_override);
-        patch.rel_size_v2_enabled.apply(&mut rel_size_v2_enabled);
 
         Ok(Self {
             checkpoint_distance,
@@ -554,7 +548,6 @@ impl TenantConfOpt {
             lsn_lease_length_for_ts,
             timeline_offloading,
             wal_receiver_protocol_override,
-            rel_size_v2_enabled,
         })
     }
 }
@@ -610,7 +603,6 @@ impl From<TenantConfOpt> for models::TenantConfig {
             lsn_lease_length_for_ts: value.lsn_lease_length_for_ts.map(humantime),
             timeline_offloading: value.timeline_offloading,
             wal_receiver_protocol_override: value.wal_receiver_protocol_override,
-            rel_size_v2_enabled: value.rel_size_v2_enabled,
         }
     }
 }

pageserver/src/tenant/remote_timeline_client/index.rs

@@ -79,24 +79,6 @@ pub struct IndexPart {
     /// when this flag is introduced.
     #[serde(skip_serializing_if = "Option::is_none", default)]
     pub(crate) last_aux_file_policy: Option<AuxFilePolicy>,
-
-    #[serde(skip_serializing_if = "Option::is_none", default)]
-    pub(crate) rel_size_migration: Option<RelSizeMigration>,
-}
-
-#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
-#[serde(rename_all = "camelCase")]
-pub enum RelSizeMigration {
-    /// The tenant is using the old rel_size format.
-    /// Note that this enum is persisted as `Option<RelSizeMigration>` in the index part, so
-    /// `None` is the same as `Some(RelSizeMigration::Legacy)`.
-    Legacy,
-    /// The tenant is migrating to the new rel_size format. Both old and new rel_size format are
-    /// persisted in the index part. The read path will read both formats and merge them.
-    Migrating,
-    /// The tenant has migrated to the new rel_size format. Only the new rel_size format is persisted
-    /// in the index part, and the read path will not read the old format.
-    Migrated,
 }
 
 impl IndexPart {
@@ -115,11 +97,10 @@ impl IndexPart {
     /// - 8: added `archived_at`
     /// - 9: +gc_blocking
     /// - 10: +import_pgdata
-    /// - 11: +rel_size_migration
-    const LATEST_VERSION: usize = 11;
+    const LATEST_VERSION: usize = 10;
 
     // Versions we may see when reading from a bucket.
-    pub const KNOWN_VERSIONS: &'static [usize] = &[1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11];
+    pub const KNOWN_VERSIONS: &'static [usize] = &[1, 2, 3, 4, 5, 6, 7, 8, 9, 10];
 
     pub const FILE_NAME: &'static str = "index_part.json";
 
@@ -135,7 +116,6 @@ impl IndexPart {
             gc_blocking: None,
             last_aux_file_policy: None,
             import_pgdata: None,
-            rel_size_migration: None,
         }
     }
 
@@ -436,7 +416,6 @@ mod tests {
             gc_blocking: None,
             last_aux_file_policy: None,
             import_pgdata: None,
-            rel_size_migration: None,
         };
 
         let part = IndexPart::from_json_bytes(example.as_bytes()).unwrap();
@@ -482,7 +461,6 @@ mod tests {
             gc_blocking: None,
             last_aux_file_policy: None,
             import_pgdata: None,
-            rel_size_migration: None,
         };
 
         let part = IndexPart::from_json_bytes(example.as_bytes()).unwrap();
@@ -529,7 +507,6 @@ mod tests {
             gc_blocking: None,
             last_aux_file_policy: None,
             import_pgdata: None,
-            rel_size_migration: None,
         };
 
         let part = IndexPart::from_json_bytes(example.as_bytes()).unwrap();
@@ -579,7 +556,6 @@ mod tests {
             gc_blocking: None,
             last_aux_file_policy: None,
             import_pgdata: None,
-            rel_size_migration: None,
         };
 
         let empty_layers_parsed = IndexPart::from_json_bytes(empty_layers_json.as_bytes()).unwrap();
@@ -624,7 +600,6 @@ mod tests {
             gc_blocking: None,
             last_aux_file_policy: None,
             import_pgdata: None,
-            rel_size_migration: None,
         };
 
         let part = IndexPart::from_json_bytes(example.as_bytes()).unwrap();
@@ -672,7 +647,6 @@ mod tests {
             gc_blocking: None,
             last_aux_file_policy: None,
             import_pgdata: None,
-            rel_size_migration: None,
         };
 
         let part = IndexPart::from_json_bytes(example.as_bytes()).unwrap();
@@ -725,7 +699,6 @@ mod tests {
             gc_blocking: None,
             last_aux_file_policy: Some(AuxFilePolicy::V2),
             import_pgdata: None,
-            rel_size_migration: None,
         };
 
         let part = IndexPart::from_json_bytes(example.as_bytes()).unwrap();
@@ -783,7 +756,6 @@ mod tests {
             gc_blocking: None,
             last_aux_file_policy: Default::default(),
             import_pgdata: None,
-            rel_size_migration: None,
         };
 
         let part = IndexPart::from_json_bytes(example.as_bytes()).unwrap();
@@ -842,7 +814,6 @@ mod tests {
             gc_blocking: None,
             last_aux_file_policy: Default::default(),
             import_pgdata: None,
-            rel_size_migration: None,
         };
 
         let part = IndexPart::from_json_bytes(example.as_bytes()).unwrap();
@@ -906,7 +877,6 @@ mod tests {
             last_aux_file_policy: Default::default(),
             archived_at: None,
             import_pgdata: None,
-            rel_size_migration: None,
         };
 
         let part = IndexPart::from_json_bytes(example.as_bytes()).unwrap();
@@ -982,86 +952,7 @@ mod tests {
                 started_at: parse_naive_datetime("2024-11-13T09:23:42.123000000"),
                 finished_at: parse_naive_datetime("2024-11-13T09:42:23.123000000"),
                 idempotency_key: import_pgdata::index_part_format::IdempotencyKey::new("specified-by-client-218a5213-5044-4562-a28d-d024c5f057f5".to_string()),
-            }))),
-            rel_size_migration: None,
-        };
-
-        let part = IndexPart::from_json_bytes(example.as_bytes()).unwrap();
-        assert_eq!(part, expected);
-    }
-
-    #[test]
-    fn v11_rel_size_migration_is_parsed() {
-        let example = r#"{
-            "version": 11,
-            "layer_metadata":{
-                "000000000000000000000000000000000000-FFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFF__0000000001696070-00000000016960E9": { "file_size": 25600000 },
-                "000000000000000000000000000000000000-FFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFF__00000000016B59D8-00000000016B5A51": { "file_size": 9007199254741001 }
-            },
-            "disk_consistent_lsn":"0/16960E8",
-            "metadata": {
-                "disk_consistent_lsn": "0/16960E8",
-                "prev_record_lsn": "0/1696070",
-                "ancestor_timeline": "e45a7f37d3ee2ff17dc14bf4f4e3f52e",
-                "ancestor_lsn": "0/0",
-                "latest_gc_cutoff_lsn": "0/1696070",
-                "initdb_lsn": "0/1696070",
-                "pg_version": 14
-            },
-            "gc_blocking": {
-                "started_at": "2024-07-19T09:00:00.123",
-                "reasons": ["DetachAncestor"]
-            },
-            "import_pgdata": {
-                "V1": {
-                    "Done": {
-                        "idempotency_key": "specified-by-client-218a5213-5044-4562-a28d-d024c5f057f5",
-                        "started_at": "2024-11-13T09:23:42.123",
-                        "finished_at": "2024-11-13T09:42:23.123"
-                    }
-                }
-            },
-            "rel_size_migration": "legacy"
-        }"#;
-
-        let expected = IndexPart {
-            version: 11,
-            layer_metadata: HashMap::from([
-                ("000000000000000000000000000000000000-FFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFF__0000000001696070-00000000016960E9".parse().unwrap(), LayerFileMetadata {
-                    file_size: 25600000,
-                    generation: Generation::none(),
-                    shard: ShardIndex::unsharded()
-                }),
-                ("000000000000000000000000000000000000-FFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFF__00000000016B59D8-00000000016B5A51".parse().unwrap(), LayerFileMetadata {
-                    file_size: 9007199254741001,
-                    generation: Generation::none(),
-                    shard: ShardIndex::unsharded()
-                })
-            ]),
-            disk_consistent_lsn: "0/16960E8".parse::<Lsn>().unwrap(),
-            metadata: TimelineMetadata::new(
-                Lsn::from_str("0/16960E8").unwrap(),
-                Some(Lsn::from_str("0/1696070").unwrap()),
-                Some(TimelineId::from_str("e45a7f37d3ee2ff17dc14bf4f4e3f52e").unwrap()),
-                Lsn::INVALID,
-                Lsn::from_str("0/1696070").unwrap(),
-                Lsn::from_str("0/1696070").unwrap(),
-                14,
-            ).with_recalculated_checksum().unwrap(),
-            deleted_at: None,
-            lineage: Default::default(),
-            gc_blocking: Some(GcBlocking {
-                started_at: parse_naive_datetime("2024-07-19T09:00:00.123000000"),
-                reasons: enumset::EnumSet::from_iter([GcBlockingReason::DetachAncestor]),
-            }),
-            last_aux_file_policy: Default::default(),
-            archived_at: None,
-            import_pgdata: Some(import_pgdata::index_part_format::Root::V1(import_pgdata::index_part_format::V1::Done(import_pgdata::index_part_format::Done{
-                started_at: parse_naive_datetime("2024-11-13T09:23:42.123000000"),
-                finished_at: parse_naive_datetime("2024-11-13T09:42:23.123000000"),
-                idempotency_key: import_pgdata::index_part_format::IdempotencyKey::new("specified-by-client-218a5213-5044-4562-a28d-d024c5f057f5".to_string()),
-            }))),
-            rel_size_migration: Some(RelSizeMigration::Legacy),
+            })))
         };
 
         let part = IndexPart::from_json_bytes(example.as_bytes()).unwrap();

pageserver/src/tenant/timeline/handle.rs

@@ -588,40 +588,32 @@ impl<T: Types> Drop for Cache<T> {
             let Some(handle_inner_arc) = handle_inner_weak.upgrade() else {
                 continue;
             };
-            let Some(handle_timeline) = handle_inner_arc
+            let handle_timeline = handle_inner_arc
                 // locking rules: drop lock before acquiring other lock below
                 .lock()
                 .expect("poisoned")
-                .shutdown()
-            else {
-                // Concurrent PerTimelineState::shutdown.
-                continue;
-            };
-            // Clean up per_timeline_state so the HandleInner allocation can be dropped.
+                .shutdown();
             let per_timeline_state = handle_timeline.per_timeline_state();
             let mut handles_lock_guard = per_timeline_state.handles.lock().expect("mutex poisoned");
             let Some(handles) = &mut *handles_lock_guard else {
                 continue;
             };
             let Some(removed_handle_inner_arc) = handles.remove(&self.id) else {
-                // Concurrent PerTimelineState::shutdown.
+                // There could have been a shutdown inbetween us upgrading the weak and locking the mutex.
                 continue;
             };
-            drop(handles_lock_guard); // locking rules!
-            assert!(Arc::ptr_eq(&removed_handle_inner_arc, &handle_inner_arc));
+            drop(handles_lock_guard); // locking rules: remember them when!
+            assert!(Arc::ptr_eq(&removed_handle_inner_arc, &handle_inner_arc,));
         }
     }
 }
 
 impl<T: Types> HandleInner<T> {
-    fn shutdown(&mut self) -> Option<Arc<T::Timeline>> {
+    fn shutdown(&mut self) -> Arc<T::Timeline> {
         match std::mem::replace(self, HandleInner::ShutDown) {
-            HandleInner::KeepingTimelineGateOpen { timeline, .. } => Some(timeline),
+            HandleInner::KeepingTimelineGateOpen { timeline, .. } => timeline,
             HandleInner::ShutDown => {
-                // Duplicate shutdowns are possible because both Cache::drop and PerTimelineState::shutdown
-                // may do it concurrently, but locking rules disallow holding per-timeline-state lock and
-                // the handle lock at the same time.
-                None
+                unreachable!("handles are only shut down once in their lifetime");
             }
         }
     }

pgxn/neon/file_cache.c

@@ -911,85 +911,55 @@ lfc_writev(NRelFileInfo rinfo, ForkNumber forkNum, BlockNumber blkno,
 			if (entry->access_count++ == 0)
 				dlist_delete(&entry->list_node);
 		}
-		/*-----------
-		 * If the chunk wasn't already in the LFC then we have these
-		 * options, in order of preference:
-		 *
-		 * Unless there is no space available, we can:
-		 *  1. Use an entry from the `holes` list, and
-		 *  2. Create a new entry.
-		 * We can always, regardless of space in the LFC:
-		 *  3. evict an entry from LRU, and
-		 *  4. ignore the write operation (the least favorite option)
-		 */
-		else if (lfc_ctl->used < lfc_ctl->limit)
+		else
 		{
-			if (!dlist_is_empty(&lfc_ctl->holes))
+			if (lfc_ctl->used >= lfc_ctl->limit)
+			{
+				/* Cache overflow: evict least recently used chunk */
+				FileCacheEntry *victim;
+
+				if (dlist_is_empty(&lfc_ctl->lru))
+				{
+					/*
+					 * No unpinned entries are available: it should very rarely happen in real life
+					 * because maximal number of backends should be smaller than LFC size.
+					 * So just refuse to update LFC in this case.
+					 */
+					neon_log(LOG, "LFC overflow: no unpinned entries available");
+					hash_search_with_hash_value(lfc_hash, &tag, hash, HASH_REMOVE, NULL);
+					LWLockRelease(lfc_lock);
+					return;
+				}
+				victim = dlist_container(FileCacheEntry, list_node, dlist_pop_head_node(&lfc_ctl->lru));
+
+				for (int i = 0; i < BLOCKS_PER_CHUNK; i++)
+				{
+					lfc_ctl->used_pages -= (victim->bitmap[i >> 5] >> (i & 31)) & 1;
+				}
+				CriticalAssert(victim->access_count == 0);
+				entry->offset = victim->offset; /* grab victim's chunk */
+				hash_search_with_hash_value(lfc_hash, &victim->key, victim->hash, HASH_REMOVE, NULL);
+				neon_log(DEBUG2, "Swap file cache page");
+			}
+			else if (!dlist_is_empty(&lfc_ctl->holes))
 			{
 				/* We can reuse a hole that was left behind when the LFC was shrunk previously */
-				FileCacheEntry *hole = dlist_container(FileCacheEntry, list_node,
-													   dlist_pop_head_node(&lfc_ctl->holes));
-				uint32 offset = hole->offset;
-				bool hole_found;
-
-				hash_search_with_hash_value(lfc_hash, &hole->key,
-											hole->hash, HASH_REMOVE, &hole_found);
+				FileCacheEntry *hole = dlist_container(FileCacheEntry, list_node, dlist_pop_head_node(&lfc_ctl->holes));
+				uint32		offset = hole->offset;
+				bool		hole_found;
+	
+				hash_search_with_hash_value(lfc_hash, &hole->key, hole->hash, HASH_REMOVE, &hole_found);
 				CriticalAssert(hole_found);
-
+	
 				lfc_ctl->used += 1;
-				entry->offset = offset;			/* reuse the hole */
+				entry->offset = offset;	/* reuse the hole */
 			}
 			else
 			{
 				lfc_ctl->used += 1;
-				entry->offset = lfc_ctl->size++;/* allocate new chunk at end
-												 * of file */
+				entry->offset = lfc_ctl->size++;	/* allocate new chunk at end
+													 * of file */
 			}
-		}
-		/*
-		 * We've already used up all allocated LFC entries.
-		 *
-		 * If we can clear an entry from the LRU, do that.
-		 * If we can't (e.g. because all other slots are being accessed)
-		 * then we will remove this entry from the hash and continue
-		 * on to the next chunk, as we may not exceed the limit.
-		 */
-		else if (!dlist_is_empty(&lfc_ctl->lru))
-		{
-			/* Cache overflow: evict least recently used chunk */
-			FileCacheEntry *victim = dlist_container(FileCacheEntry, list_node,
-													 dlist_pop_head_node(&lfc_ctl->lru));
-
-			for (int i = 0; i < BLOCKS_PER_CHUNK; i++)
-			{
-				lfc_ctl->used_pages -= (victim->bitmap[i >> 5] >> (i & 31)) & 1;
-			}
-
-			CriticalAssert(victim->access_count == 0);
-			entry->offset = victim->offset; /* grab victim's chunk */
-			hash_search_with_hash_value(lfc_hash, &victim->key,
-										victim->hash, HASH_REMOVE, NULL);
-			neon_log(DEBUG2, "Swap file cache page");
-		}
-		else
-		{
-			/* Can't add this chunk - we don't have the space for it */
-			hash_search_with_hash_value(lfc_hash, &entry->key, hash,
-										HASH_REMOVE, NULL);
-
-			/*
-			 * We can't process this chunk due to lack of space in LFC,
-			 * so skip to the next one
-			 */
-			LWLockRelease(lfc_lock);
-			blkno += blocks_in_chunk;
-			buf_offset += blocks_in_chunk;
-			nblocks -= blocks_in_chunk;
-			continue;
-		}
-
-		if (!found)
-		{
 			entry->access_count = 1;
 			entry->hash = hash;
 			memset(entry->bitmap, 0, sizeof entry->bitmap);

storage_scrubber/src/pageserver_physical_gc.rs

@@ -8,8 +8,6 @@ use crate::checks::{
 };
 use crate::metadata_stream::{stream_tenant_timelines, stream_tenants};
 use crate::{init_remote, BucketConfig, NodeKind, RootTarget, TenantShardTimelineId, MAX_RETRIES};
-use async_stream::try_stream;
-use futures::future::Either;
 use futures_util::{StreamExt, TryStreamExt};
 use pageserver::tenant::remote_timeline_client::index::LayerFileMetadata;
 use pageserver::tenant::remote_timeline_client::manifest::OffloadedTimelineManifest;
@@ -580,7 +578,7 @@ async fn gc_timeline(
     target: &RootTarget,
     mode: GcMode,
     ttid: TenantShardTimelineId,
-    accumulator: &std::sync::Mutex<TenantRefAccumulator>,
+    accumulator: &Arc<std::sync::Mutex<TenantRefAccumulator>>,
     tenant_manifest_info: Arc<Option<RemoteTenantManifestInfo>>,
 ) -> anyhow::Result<GcSummary> {
     let mut summary = GcSummary::default();
@@ -723,9 +721,9 @@ pub async fn pageserver_physical_gc(
 
     let remote_client = Arc::new(remote_client);
     let tenants = if tenant_shard_ids.is_empty() {
-        Either::Left(stream_tenants(&remote_client, &target))
+        futures::future::Either::Left(stream_tenants(&remote_client, &target))
     } else {
-        Either::Right(futures::stream::iter(tenant_shard_ids.into_iter().map(Ok)))
+        futures::future::Either::Right(futures::stream::iter(tenant_shard_ids.into_iter().map(Ok)))
     };
 
     // How many tenants to process in parallel.  We need to be mindful of pageservers
@@ -733,16 +731,16 @@ pub async fn pageserver_physical_gc(
     const CONCURRENCY: usize = 32;
 
     // Accumulate information about each tenant for cross-shard GC step we'll do at the end
-    let accumulator = std::sync::Mutex::new(TenantRefAccumulator::default());
-
-    // Accumulate information about how many manifests we have GCd
-    let manifest_gc_summary = std::sync::Mutex::new(GcSummary::default());
+    let accumulator = Arc::new(std::sync::Mutex::new(TenantRefAccumulator::default()));
 
     // Generate a stream of TenantTimelineId
+    enum GcSummaryOrContent<T> {
+        Content(T),
+        GcSummary(GcSummary),
+    }
     let timelines = tenants.map_ok(|tenant_shard_id| {
         let target_ref = &target;
         let remote_client_ref = &remote_client;
-        let manifest_gc_summary_ref = &manifest_gc_summary;
         async move {
             let gc_manifest_result = gc_tenant_manifests(
                 remote_client_ref,
@@ -759,48 +757,55 @@ pub async fn pageserver_physical_gc(
                     (GcSummary::default(), None)
                 }
             };
-            manifest_gc_summary_ref
-                .lock()
-                .unwrap()
-                .merge(summary_from_manifest);
             let tenant_manifest_arc = Arc::new(tenant_manifest_opt);
-            let mut timelines = Box::pin(
-                stream_tenant_timelines(remote_client_ref, target_ref, tenant_shard_id).await?,
-            );
-            Ok(try_stream! {
-                while let Some(ttid_res) = timelines.next().await {
-                    let ttid = ttid_res?;
-                    yield (ttid, tenant_manifest_arc.clone());
-                }
-            })
+            let summary_from_manifest = Ok(GcSummaryOrContent::<(_, _)>::GcSummary(
+                summary_from_manifest,
+            ));
+            stream_tenant_timelines(remote_client_ref, target_ref, tenant_shard_id)
+                .await
+                .map(|stream| {
+                    stream
+                        .zip(futures::stream::iter(std::iter::repeat(
+                            tenant_manifest_arc,
+                        )))
+                        .map(|(ttid_res, tenant_manifest_arc)| {
+                            ttid_res.map(move |ttid| {
+                                GcSummaryOrContent::Content((ttid, tenant_manifest_arc))
+                            })
+                        })
+                        .chain(futures::stream::iter([summary_from_manifest].into_iter()))
+                })
         }
     });
+    let timelines = std::pin::pin!(timelines.try_buffered(CONCURRENCY));
+    let timelines = timelines.try_flatten();
 
     let mut summary = GcSummary::default();
-    {
-        let timelines = std::pin::pin!(timelines.try_buffered(CONCURRENCY));
-        let timelines = timelines.try_flatten();
 
-        let timelines = timelines.map_ok(|(ttid, tenant_manifest_arc)| {
-            gc_timeline(
-                &remote_client,
-                &min_age,
-                &target,
-                mode,
-                ttid,
-                &accumulator,
-                tenant_manifest_arc,
-            )
+    // Drain futures for per-shard GC, populating accumulator as a side effect
+    {
+        let timelines = timelines.map_ok(|summary_or_ttid| match summary_or_ttid {
+            GcSummaryOrContent::Content((ttid, tenant_manifest_arc)) => {
+                futures::future::Either::Left(gc_timeline(
+                    &remote_client,
+                    &min_age,
+                    &target,
+                    mode,
+                    ttid,
+                    &accumulator,
+                    tenant_manifest_arc,
+                ))
+            }
+            GcSummaryOrContent::GcSummary(gc_summary) => {
+                futures::future::Either::Right(futures::future::ok(gc_summary))
+            }
         });
         let mut timelines = std::pin::pin!(timelines.try_buffered(CONCURRENCY));
 
-        // Drain futures for per-shard GC, populating accumulator as a side effect
         while let Some(i) = timelines.next().await {
             summary.merge(i?);
         }
     }
-    // Streams are lazily evaluated, so only now do we have access to the inner object
-    summary.merge(manifest_gc_summary.into_inner().unwrap());
 
     // Execute cross-shard GC, using the accumulator's full view of all the shards built in the per-shard GC
     let Some(client) = controller_client else {
@@ -808,7 +813,8 @@ pub async fn pageserver_physical_gc(
         return Ok(summary);
     };
 
-    let (ancestor_shards, ancestor_refs) = accumulator
+    let (ancestor_shards, ancestor_refs) = Arc::into_inner(accumulator)
+        .unwrap()
         .into_inner()
         .unwrap()
         .into_gc_ancestors(client, &mut summary)

test_runner/conftest.py

@@ -15,5 +15,4 @@ pytest_plugins = (
     "fixtures.compare_fixtures",
     "fixtures.slow",
     "fixtures.reruns",
-    "fixtures.fast_import",
 )

test_runner/fixtures/fast_import.py

@@ -1,104 +0,0 @@
-import os
-import shutil
-import subprocess
-import tempfile
-from collections.abc import Iterator
-from pathlib import Path
-
-import pytest
-
-from fixtures.log_helper import log
-from fixtures.neon_cli import AbstractNeonCli
-from fixtures.pg_version import PgVersion
-
-
-class FastImport(AbstractNeonCli):
-    COMMAND = "fast_import"
-    cmd: subprocess.CompletedProcess[str] | None = None
-
-    def __init__(
-        self,
-        extra_env: dict[str, str] | None,
-        binpath: Path,
-        pg_distrib_dir: Path,
-        pg_version: PgVersion,
-        workdir: Path,
-    ):
-        if extra_env is None:
-            env_vars = {}
-        else:
-            env_vars = extra_env.copy()
-
-        if not (binpath / self.COMMAND).exists():
-            raise Exception(f"{self.COMMAND} binary not found at '{binpath}'")
-        super().__init__(env_vars, binpath)
-
-        pg_dir = pg_distrib_dir / pg_version.v_prefixed
-        self.pg_distrib_dir = pg_distrib_dir
-        self.pg_version = pg_version
-        self.pg_bin = pg_dir / "bin"
-        if not (self.pg_bin / "postgres").exists():
-            raise Exception(f"postgres binary was not found at '{self.pg_bin}'")
-        self.pg_lib = pg_dir / "lib"
-        if env_vars.get("LD_LIBRARY_PATH") is not None:
-            self.pg_lib = Path(env_vars["LD_LIBRARY_PATH"])
-        elif os.getenv("LD_LIBRARY_PATH") is not None:
-            self.pg_lib = Path(str(os.getenv("LD_LIBRARY_PATH")))
-        if not workdir.exists():
-            raise Exception(f"Working directory '{workdir}' does not exist")
-        self.workdir = workdir
-
-    def run(
-        self,
-        pg_port: int,
-        source_connection_string: str | None = None,
-        s3prefix: str | None = None,
-        interactive: bool = False,
-    ) -> subprocess.CompletedProcess[str]:
-        if self.cmd is not None:
-            raise Exception("Command already executed")
-        args = [
-            f"--pg-bin-dir={self.pg_bin}",
-            f"--pg-lib-dir={self.pg_lib}",
-            f"--pg-port={pg_port}",
-            f"--working-directory={self.workdir}",
-        ]
-        if source_connection_string is not None:
-            args.append(f"--source-connection-string={source_connection_string}")
-        if s3prefix is not None:
-            args.append(f"--s3-prefix={s3prefix}")
-        if interactive:
-            args.append("--interactive")
-
-        self.cmd = self.raw_cli(args)
-        return self.cmd
-
-    def __enter__(self):
-        return self
-
-    def __exit__(self, *args):
-        if self.workdir.exists():
-            shutil.rmtree(self.workdir)
-
-
-@pytest.fixture(scope="function")
-def fast_import(
-    pg_version: PgVersion,
-    test_output_dir: Path,
-    neon_binpath: Path,
-    pg_distrib_dir: Path,
-) -> Iterator[FastImport]:
-    workdir = Path(tempfile.mkdtemp())
-    with FastImport(None, neon_binpath, pg_distrib_dir, pg_version, workdir) as fi:
-        yield fi
-
-        if fi.cmd is None:
-            return
-
-        # dump stdout & stderr into test log dir
-        with open(test_output_dir / "fast_import.stdout", "w") as f:
-            f.write(fi.cmd.stdout)
-        with open(test_output_dir / "fast_import.stderr", "w") as f:
-            f.write(fi.cmd.stderr)
-
-        log.info("Written logs to %s", test_output_dir)

test_runner/regress/test_attach_tenant_config.py

@@ -176,7 +176,6 @@ def test_fully_custom_config(positive_env: NeonEnv):
             "type": "interpreted",
             "args": {"format": "bincode", "compression": {"zstd": {"level": 1}}},
         },
-        "rel_size_v2_enabled": True,
     }
 
     vps_http = env.storage_controller.pageserver_api()

test_runner/regress/test_import_pgdata.py

@@ -7,15 +7,13 @@ import psycopg2
 import psycopg2.errors
 import pytest
 from fixtures.common_types import Lsn, TenantId, TenantShardId, TimelineId
-from fixtures.fast_import import FastImport
 from fixtures.log_helper import log
-from fixtures.neon_fixtures import NeonEnvBuilder, PgBin, PgProtocol, VanillaPostgres
+from fixtures.neon_fixtures import NeonEnvBuilder, VanillaPostgres
 from fixtures.pageserver.http import (
     ImportPgdataIdemptencyKey,
     PageserverApiException,
 )
 from fixtures.pg_version import PgVersion
-from fixtures.port_distributor import PortDistributor
 from fixtures.remote_storage import RemoteStorageKind
 from fixtures.utils import run_only_on_postgres
 from pytest_httpserver import HTTPServer
@@ -315,41 +313,3 @@ def test_pgdata_import_smoke(
     validate_vanilla_equivalence(br_initdb_endpoint)
     with pytest.raises(psycopg2.errors.UndefinedTable):
         br_initdb_endpoint.safe_psql("select * from othertable")
-
-
-@run_only_on_postgres(
-    [PgVersion.V14, PgVersion.V15, PgVersion.V16],
-    "newer control file catalog version and struct format isn't supported",
-)
-def test_fast_import_binary(
-    test_output_dir,
-    vanilla_pg: VanillaPostgres,
-    port_distributor: PortDistributor,
-    fast_import: FastImport,
-):
-    vanilla_pg.start()
-    vanilla_pg.safe_psql("CREATE TABLE foo (a int); INSERT INTO foo SELECT generate_series(1, 10);")
-
-    pg_port = port_distributor.get_port()
-    fast_import.run(pg_port, vanilla_pg.connstr())
-    vanilla_pg.stop()
-
-    pgbin = PgBin(test_output_dir, fast_import.pg_distrib_dir, fast_import.pg_version)
-    with VanillaPostgres(
-        fast_import.workdir / "pgdata", pgbin, pg_port, False
-    ) as new_pgdata_vanilla_pg:
-        new_pgdata_vanilla_pg.start()
-
-        # database name and user are hardcoded in fast_import binary, and they are different from normal vanilla postgres
-        conn = PgProtocol(dsn=f"postgresql://cloud_admin@localhost:{pg_port}/neondb")
-        res = conn.safe_psql("SELECT count(*) FROM foo;")
-        log.info(f"Result: {res}")
-        assert res[0][0] == 10
-
-
-# TODO: Maybe test with pageserver?
-# 1. run whole neon env
-# 2. create timeline with some s3 path???
-# 3. run fast_import with s3 prefix
-# 4. ??? mock http where pageserver will report progress
-# 5. run compute on this timeline and check if data is there

test_runner/regress/test_local_file_cache.py

@@ -7,78 +7,9 @@ import threading
 import time
 
 import pytest
-from fixtures.neon_fixtures import NeonEnv, NeonEnvBuilder
+from fixtures.neon_fixtures import NeonEnvBuilder
 from fixtures.utils import USE_LFC, query_scalar
 
-"""
-Test whether LFC doesn't error out when the LRU is empty, but the LFC is
-already at its maximum size.
-
-If we don't handle this safely, we might allocate more hash entries than
-otherwise considered safe, thus causing ERRORs in hash_search(HASH_ENTER) once
-we hit lfc->used >= lfc->limit.
-"""
-
-
-@pytest.mark.skipif(not USE_LFC, reason="LFC is disabled, skipping")
-def test_local_file_cache_all_pinned(neon_simple_env: NeonEnv):
-    env = neon_simple_env
-    endpoint = env.endpoints.create_start(
-        "main",
-        config_lines=[
-            "neon.max_file_cache_size='1MB'",
-            "neon.file_cache_size_limit='1MB'",
-        ],
-    )
-    top_cur = endpoint.connect().cursor()
-
-    stop = threading.Event()
-    n_rows = 10000
-    n_threads = 5
-    n_updates_per_connection = 1000
-
-    top_cur.execute("CREATE TABLE lfctest (id int4 PRIMARY KEY, n int) WITH (fillfactor=10)")
-    top_cur.execute(f"INSERT INTO lfctest SELECT g, 1 FROM generate_series(1, {n_rows}) g")
-
-    # Start threads that will perform random UPDATEs. Each UPDATE
-    # increments the counter on the row, so that we can check at the
-    # end that the sum of all the counters match the number of updates
-    # performed (plus the initial 1 on each row).
-    #
-    # Furthermore, each thread will reconnect between every 1000 updates.
-    def run_updates(n_updates_performed_q: queue.Queue[int]):
-        n_updates_performed = 0
-        conn = endpoint.connect()
-        cur = conn.cursor()
-        while not stop.is_set():
-            id = random.randint(1, n_rows)
-            cur.execute(f"UPDATE lfctest SET n = n + 1 WHERE id = {id}")
-            n_updates_performed += 1
-            if n_updates_performed % n_updates_per_connection == 0:
-                cur.close()
-                conn.close()
-                conn = endpoint.connect()
-                cur = conn.cursor()
-        n_updates_performed_q.put(n_updates_performed)
-
-    n_updates_performed_q: queue.Queue[int] = queue.Queue()
-    threads: list[threading.Thread] = []
-    for _i in range(n_threads):
-        thread = threading.Thread(target=run_updates, args=(n_updates_performed_q,), daemon=True)
-        thread.start()
-        threads.append(thread)
-
-    time.sleep(15)
-
-    stop.set()
-
-    n_updates_performed = 0
-    for thread in threads:
-        thread.join()
-        n_updates_performed += n_updates_performed_q.get()
-
-    assert query_scalar(top_cur, "SELECT SUM(n) FROM lfctest") == n_rows + n_updates_performed
-
 
 @pytest.mark.skipif(not USE_LFC, reason="LFC is disabled, skipping")
 def test_local_file_cache_unlink(neon_env_builder: NeonEnvBuilder):

test_runner/regress/test_storage_scrubber.py

@@ -227,9 +227,7 @@ def test_scrubber_physical_gc_ancestors(neon_env_builder: NeonEnvBuilder, shard_
     new_shard_count = 4
     assert shard_count is None or new_shard_count > shard_count
     shards = env.storage_controller.tenant_shard_split(tenant_id, shard_count=new_shard_count)
-    env.storage_controller.reconcile_until_idle(
-        timeout_secs=120
-    )  # Move shards to their final locations immediately
+    env.storage_controller.reconcile_until_idle()  # Move shards to their final locations immediately
 
     # Create a timeline after split, to ensure scrubber can handle timelines that exist in child shards but not ancestors
     env.storage_controller.pageserver_api().timeline_create(
@@ -271,8 +269,6 @@ def test_scrubber_physical_gc_ancestors(neon_env_builder: NeonEnvBuilder, shard_
         ps.http_client().timeline_compact(
             shard, timeline_id, force_image_layer_creation=True, wait_until_uploaded=True
         )
-        # Add some WAL so that we don't gc at the latest remote consistent lsn
-        workload.churn_rows(1)
         ps.http_client().timeline_gc(shard, timeline_id, 0)
 
     # We will use a min_age_secs=1 threshold for deletion, let it pass