try with unshallow

Co-authored-by: Heikki Linnakangas <heikki@neon.tech>

.github/workflows/build_and_test.yml

@@ -133,38 +133,38 @@ jobs:
 
       - name: Check vendor/postgres-v14 submodule reference
         if: steps.check-if-submodules-changed.outputs.vendor == 'true'
-        uses: jtmullen/submodule-branch-check-action@v1
+        uses: hlinnaka/submodule-branch-check-action@main
         with:
           path: "vendor/postgres-v14"
           fetch_depth: "50"
-          sub_fetch_depth: "50"
+          sub_fetch_depth: ""
           pass_if_unchanged: true
 
       - name: Check vendor/postgres-v15 submodule reference
         if: steps.check-if-submodules-changed.outputs.vendor == 'true'
-        uses: jtmullen/submodule-branch-check-action@v1
+        uses: hlinnaka/submodule-branch-check-action@main
         with:
           path: "vendor/postgres-v15"
           fetch_depth: "50"
-          sub_fetch_depth: "50"
+          sub_fetch_depth: ""
           pass_if_unchanged: true
 
       - name: Check vendor/postgres-v16 submodule reference
         if: steps.check-if-submodules-changed.outputs.vendor == 'true'
-        uses: jtmullen/submodule-branch-check-action@v1
+        uses: hlinnaka/submodule-branch-check-action@main
         with:
           path: "vendor/postgres-v16"
           fetch_depth: "50"
-          sub_fetch_depth: "50"
+          sub_fetch_depth: ""
           pass_if_unchanged: true
 
       - name: Check vendor/postgres-v17 submodule reference
         if: steps.check-if-submodules-changed.outputs.vendor == 'true'
-        uses: jtmullen/submodule-branch-check-action@v1
+        uses: hlinnaka/submodule-branch-check-action@main
         with:
           path: "vendor/postgres-v17"
           fetch_depth: "50"
-          sub_fetch_depth: "50"
+          sub_fetch_depth: ""
           pass_if_unchanged: true
 
   check-codestyle-rust:

Cargo.lock

@@ -1229,15 +1229,12 @@ dependencies = [
  "flate2",
  "futures",
  "hyper 0.14.30",
- "metrics",
  "nix 0.27.1",
  "notify",
  "num_cpus",
- "once_cell",
  "opentelemetry",
  "opentelemetry_sdk",
  "postgres",
- "prometheus",
  "regex",
  "remote_storage",
  "reqwest 0.12.4",
@@ -4009,7 +4006,7 @@ dependencies = [
 [[package]]
 name = "postgres"
 version = "0.19.4"
-source = "git+https://github.com/neondatabase/rust-postgres.git?branch=vlad/interpreted-wal-record-replication-support#e619cf8c2c572e71cbc97f1c7f4cab8219f07d55"
+source = "git+https://github.com/neondatabase/rust-postgres.git?rev=20031d7a9ee1addeae6e0968e3899ae6bf01cee2#20031d7a9ee1addeae6e0968e3899ae6bf01cee2"
 dependencies = [
  "bytes",
  "fallible-iterator",
@@ -4022,7 +4019,7 @@ dependencies = [
 [[package]]
 name = "postgres-protocol"
 version = "0.6.4"
-source = "git+https://github.com/neondatabase/rust-postgres.git?branch=vlad/interpreted-wal-record-replication-support#e619cf8c2c572e71cbc97f1c7f4cab8219f07d55"
+source = "git+https://github.com/neondatabase/rust-postgres.git?rev=20031d7a9ee1addeae6e0968e3899ae6bf01cee2#20031d7a9ee1addeae6e0968e3899ae6bf01cee2"
 dependencies = [
  "base64 0.20.0",
  "byteorder",
@@ -4041,7 +4038,7 @@ dependencies = [
 [[package]]
 name = "postgres-types"
 version = "0.2.4"
-source = "git+https://github.com/neondatabase/rust-postgres.git?branch=vlad/interpreted-wal-record-replication-support#e619cf8c2c572e71cbc97f1c7f4cab8219f07d55"
+source = "git+https://github.com/neondatabase/rust-postgres.git?rev=20031d7a9ee1addeae6e0968e3899ae6bf01cee2#20031d7a9ee1addeae6e0968e3899ae6bf01cee2"
 dependencies = [
  "bytes",
  "fallible-iterator",
@@ -5161,7 +5158,6 @@ dependencies = [
  "itertools 0.10.5",
  "metrics",
  "once_cell",
- "pageserver_api",
  "parking_lot 0.12.1",
  "postgres",
  "postgres-protocol",
@@ -5192,7 +5188,6 @@ dependencies = [
  "tracing-subscriber",
  "url",
  "utils",
- "wal_decoder",
  "walproposer",
  "workspace_hack",
 ]
@@ -5665,9 +5660,9 @@ dependencies = [
 
 [[package]]
 name = "smallvec"
-version = "1.13.2"
+version = "1.13.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "3c5e1a9a646d36c3599cd173a41282daf47c44583ad367b8e6837255952e5c67"
+checksum = "e6ecd384b10a64542d77071bd64bd7b231f4ed5940fba55e98c3de13824cf3d7"
 
 [[package]]
 name = "smol_str"
@@ -6076,9 +6071,9 @@ dependencies = [
 
 [[package]]
 name = "tikv-jemalloc-ctl"
-version = "0.6.0"
+version = "0.5.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "f21f216790c8df74ce3ab25b534e0718da5a1916719771d3fec23315c99e468b"
+checksum = "619bfed27d807b54f7f776b9430d4f8060e66ee138a28632ca898584d462c31c"
 dependencies = [
  "libc",
  "paste",
@@ -6087,9 +6082,9 @@ dependencies = [
 
 [[package]]
 name = "tikv-jemalloc-sys"
-version = "0.6.0+5.3.0-1-ge13ca993e8ccb9ba9847cc330696e02839f328f7"
+version = "0.5.4+5.3.0-patched"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "cd3c60906412afa9c2b5b5a48ca6a5abe5736aec9eb48ad05037a677e52e4e2d"
+checksum = "9402443cb8fd499b6f327e40565234ff34dbda27460c5b47db0db77443dd85d1"
 dependencies = [
  "cc",
  "libc",
@@ -6097,9 +6092,9 @@ dependencies = [
 
 [[package]]
 name = "tikv-jemallocator"
-version = "0.6.0"
+version = "0.5.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "4cec5ff18518d81584f477e9bfdf957f5bb0979b0bac3af4ca30b5b3ae2d2865"
+checksum = "965fe0c26be5c56c94e38ba547249074803efd52adfb66de62107d95aab3eaca"
 dependencies = [
  "libc",
  "tikv-jemalloc-sys",
@@ -6229,7 +6224,7 @@ dependencies = [
 [[package]]
 name = "tokio-postgres"
 version = "0.7.7"
-source = "git+https://github.com/neondatabase/rust-postgres.git?branch=vlad/interpreted-wal-record-replication-support#e619cf8c2c572e71cbc97f1c7f4cab8219f07d55"
+source = "git+https://github.com/neondatabase/rust-postgres.git?rev=20031d7a9ee1addeae6e0968e3899ae6bf01cee2#20031d7a9ee1addeae6e0968e3899ae6bf01cee2"
 dependencies = [
  "async-trait",
  "byteorder",
@@ -6783,7 +6778,6 @@ dependencies = [
  "serde_assert",
  "serde_json",
  "serde_path_to_error",
- "serde_with",
  "signal-hook",
  "strum",
  "strum_macros",

Cargo.toml

@@ -168,8 +168,8 @@ sync_wrapper = "0.1.2"
 tar = "0.4"
 test-context = "0.3"
 thiserror = "1.0"
-tikv-jemallocator = { version = "0.6", features = ["stats"] }
-tikv-jemalloc-ctl = { version = "0.6", features = ["stats"] }
+tikv-jemallocator = "0.5"
+tikv-jemalloc-ctl = "0.5"
 tokio = { version = "1.17", features = ["macros"] }
 tokio-epoll-uring = { git = "https://github.com/neondatabase/tokio-epoll-uring.git" , branch = "main" }
 tokio-io-timeout = "1.2.0"
@@ -203,10 +203,21 @@ env_logger = "0.10"
 log = "0.4"
 
 ## Libraries from neondatabase/ git forks, ideally with changes to be upstreamed
-postgres = { git = "https://github.com/neondatabase/rust-postgres.git", branch = "vlad/interpreted-wal-record-replication-support" }
-postgres-protocol = { git = "https://github.com/neondatabase/rust-postgres.git", branch = "vlad/interpreted-wal-record-replication-support" }
-postgres-types = { git = "https://github.com/neondatabase/rust-postgres.git", branch = "vlad/interpreted-wal-record-replication-support" }
-tokio-postgres = { git = "https://github.com/neondatabase/rust-postgres.git", branch = "vlad/interpreted-wal-record-replication-support" }
+
+# We want to use the 'neon' branch for these, but there's currently one
+# incompatible change on the branch. See:
+#
+# - PR #8076 which contained changes that depended on the new changes in
+#   the rust-postgres crate, and
+# - PR #8654 which reverted those changes and made the code in proxy incompatible
+#   with the tip of the 'neon' branch again.
+#
+# When those proxy changes are re-applied (see PR #8747), we can switch using
+# the tip of the 'neon' branch again.
+postgres = { git = "https://github.com/neondatabase/rust-postgres.git", rev = "20031d7a9ee1addeae6e0968e3899ae6bf01cee2" }
+postgres-protocol = { git = "https://github.com/neondatabase/rust-postgres.git", rev = "20031d7a9ee1addeae6e0968e3899ae6bf01cee2" }
+postgres-types = { git = "https://github.com/neondatabase/rust-postgres.git", rev = "20031d7a9ee1addeae6e0968e3899ae6bf01cee2" }
+tokio-postgres = { git = "https://github.com/neondatabase/rust-postgres.git", rev = "20031d7a9ee1addeae6e0968e3899ae6bf01cee2" }
 
 ## Local libraries
 compute_api = { version = "0.1", path = "./libs/compute_api/" }
@@ -244,7 +255,7 @@ tonic-build = "0.12"
 [patch.crates-io]
 
 # Needed to get `tokio-postgres-rustls` to depend on our fork.
-tokio-postgres = { git = "https://github.com/neondatabase/rust-postgres.git", branch = "vlad/interpreted-wal-record-replication-support" }
+tokio-postgres = { git = "https://github.com/neondatabase/rust-postgres.git", rev = "20031d7a9ee1addeae6e0968e3899ae6bf01cee2" }
 
 ################# Binary contents sections
 

compute/etc/neon_collector.jsonnet

@@ -6,7 +6,6 @@
     import 'sql_exporter/compute_backpressure_throttling_seconds.libsonnet',
     import 'sql_exporter/compute_current_lsn.libsonnet',
     import 'sql_exporter/compute_logical_snapshot_files.libsonnet',
-    import 'sql_exporter/compute_max_connections.libsonnet',
     import 'sql_exporter/compute_receive_lsn.libsonnet',
     import 'sql_exporter/compute_subscriptions_count.libsonnet',
     import 'sql_exporter/connection_counts.libsonnet',

compute/etc/sql_exporter/compute_backpressure_throttling_seconds.sql

@@ -1 +1 @@
-SELECT (neon.backpressure_throttling_time()::float8 / 1000000) AS throttled;
+SELECT neon.backpressure_throttling_time()::float8 / 1000 AS throttled;

compute/etc/sql_exporter/compute_max_connections.libsonnet

@@ -1,10 +0,0 @@
-{
-  metric_name: 'compute_max_connections',
-  type: 'gauge',
-  help: 'Max connections allowed for Postgres',
-  key_labels: null,
-  values: [
-    'max_connections',
-  ],
-  query: importstr 'sql_exporter/compute_max_connections.sql',
-}

compute/etc/sql_exporter/compute_max_connections.sql

@@ -1 +0,0 @@
-SELECT current_setting('max_connections') as max_connections;

compute/patches/cloud_regress_pg16.patch

@@ -147,7 +147,7 @@ index 542c2e098c..0062d3024f 100644
  ALTER TABLE ptnowner1 OWNER TO regress_ptnowner;
  ALTER TABLE ptnowner OWNER TO regress_ptnowner;
 diff --git a/src/test/regress/expected/collate.icu.utf8.out b/src/test/regress/expected/collate.icu.utf8.out
-index 3f9a8f539c..0a51b52940 100644
+index 97bbe53b64..eac3d42a79 100644
 --- a/src/test/regress/expected/collate.icu.utf8.out
 +++ b/src/test/regress/expected/collate.icu.utf8.out
 @@ -1016,7 +1016,7 @@ select * from collate_test1 where b ilike 'ABC';
@@ -309,7 +309,7 @@ index b48365ec98..a6ef910055 100644
  -- the wrong partition. This test is *not* guaranteed to trigger that bug, but
  -- does so when shared_buffers is small enough.  To test if we encountered the
 diff --git a/src/test/regress/expected/copy2.out b/src/test/regress/expected/copy2.out
-index 9a74820ee8..22400a5551 100644
+index faf1a4d1b0..a44c97db52 100644
 --- a/src/test/regress/expected/copy2.out
 +++ b/src/test/regress/expected/copy2.out
 @@ -553,8 +553,8 @@ select * from check_con_tbl;
@@ -573,7 +573,7 @@ index 93302a07ef..1a73f083ac 100644
  -- that does not match with what's expected.
  -- This checks all the object types that include schema qualifications.
 diff --git a/src/test/regress/expected/create_view.out b/src/test/regress/expected/create_view.out
-index f551624afb..57f1e432d4 100644
+index f3f8c7b5a2..3e3e54ff4c 100644
 --- a/src/test/regress/expected/create_view.out
 +++ b/src/test/regress/expected/create_view.out
 @@ -18,7 +18,8 @@ CREATE TABLE real_city (
@@ -700,12 +700,12 @@ index 6ed50fdcfa..caa00a345d 100644
  COMMENT ON FOREIGN DATA WRAPPER dummy IS 'useless';
  CREATE FOREIGN DATA WRAPPER postgresql VALIDATOR postgresql_fdw_validator;
 diff --git a/src/test/regress/expected/foreign_key.out b/src/test/regress/expected/foreign_key.out
-index 6b8c2f2414..8e13b7fa46 100644
+index 12e523c737..8872e23935 100644
 --- a/src/test/regress/expected/foreign_key.out
 +++ b/src/test/regress/expected/foreign_key.out
-@@ -1985,7 +1985,7 @@ ALTER TABLE fk_partitioned_fk_6 ATTACH PARTITION fk_partitioned_pk_6 FOR VALUES
- ERROR:  cannot ALTER TABLE "fk_partitioned_pk_61" because it is being used by active queries in this session
- DROP TABLE fk_partitioned_pk_6, fk_partitioned_fk_6;
+@@ -1968,7 +1968,7 @@ ALTER TABLE fk_partitioned_fk ATTACH PARTITION fk_partitioned_fk_2
+   FOR VALUES IN (1600);
+ -- leave these tables around intentionally
  -- test the case when the referenced table is owned by a different user
 -create role regress_other_partitioned_fk_owner;
 +create role regress_other_partitioned_fk_owner PASSWORD NEON_PASSWORD_PLACEHOLDER;
@@ -713,7 +713,7 @@ index 6b8c2f2414..8e13b7fa46 100644
  set role regress_other_partitioned_fk_owner;
  create table other_partitioned_fk(a int, b int) partition by list (a);
 diff --git a/src/test/regress/expected/generated.out b/src/test/regress/expected/generated.out
-index 5881420388..4ae21aa43c 100644
+index 0f623f7119..b48588a54e 100644
 --- a/src/test/regress/expected/generated.out
 +++ b/src/test/regress/expected/generated.out
 @@ -534,7 +534,7 @@ CREATE TABLE gtest10a (a int PRIMARY KEY, b int GENERATED ALWAYS AS (a * 2) STOR
@@ -762,7 +762,7 @@ index a2036a1597..805d73b9d2 100644
  -- fields, leading to long bucket chains and lots of table expansion.
  -- this is therefore a stress test of the bucket overflow code (unlike
 diff --git a/src/test/regress/expected/identity.out b/src/test/regress/expected/identity.out
-index 1b74958de9..078187b542 100644
+index cc7772349f..98a08eb48d 100644
 --- a/src/test/regress/expected/identity.out
 +++ b/src/test/regress/expected/identity.out
 @@ -520,7 +520,7 @@ ALTER TABLE itest7 ALTER COLUMN a SET GENERATED BY DEFAULT;
@@ -775,10 +775,10 @@ index 1b74958de9..078187b542 100644
  GRANT SELECT, INSERT ON itest8 TO regress_identity_user1;
  SET ROLE regress_identity_user1;
 diff --git a/src/test/regress/expected/inherit.out b/src/test/regress/expected/inherit.out
-index 8f831c95c3..ec681b52af 100644
+index 4943429e9b..0257f22b15 100644
 --- a/src/test/regress/expected/inherit.out
 +++ b/src/test/regress/expected/inherit.out
-@@ -2636,7 +2636,7 @@ create index on permtest_parent (left(c, 3));
+@@ -2606,7 +2606,7 @@ create index on permtest_parent (left(c, 3));
  insert into permtest_parent
    select 1, 'a', left(fipshash(i::text), 5) from generate_series(0, 100) i;
  analyze permtest_parent;
@@ -1133,7 +1133,7 @@ index 8475231735..1afae5395f 100644
  SELECT rolname, rolpassword
      FROM pg_authid
 diff --git a/src/test/regress/expected/privileges.out b/src/test/regress/expected/privileges.out
-index 5b9dba7b32..cc408dad42 100644
+index fbb0489a4f..2905194e2c 100644
 --- a/src/test/regress/expected/privileges.out
 +++ b/src/test/regress/expected/privileges.out
 @@ -20,19 +20,19 @@ SELECT lo_unlink(oid) FROM pg_largeobject_metadata WHERE oid >= 1000 AND oid < 3
@@ -1185,7 +1185,7 @@ index 5b9dba7b32..cc408dad42 100644
  GRANT pg_read_all_data TO regress_priv_user6;
  GRANT pg_write_all_data TO regress_priv_user7;
  GRANT pg_read_all_settings TO regress_priv_user8 WITH ADMIN OPTION;
-@@ -212,8 +212,8 @@ REVOKE pg_read_all_settings FROM regress_priv_user8;
+@@ -145,8 +145,8 @@ REVOKE pg_read_all_settings FROM regress_priv_user8;
  DROP USER regress_priv_user10;
  DROP USER regress_priv_user9;
  DROP USER regress_priv_user8;
@@ -1196,7 +1196,7 @@ index 5b9dba7b32..cc408dad42 100644
  ALTER GROUP regress_priv_group1 ADD USER regress_priv_user4;
  GRANT regress_priv_group2 TO regress_priv_user2 GRANTED BY regress_priv_user1;
  SET SESSION AUTHORIZATION regress_priv_user1;
-@@ -239,12 +239,16 @@ GRANT regress_priv_role TO regress_priv_user1 WITH ADMIN OPTION GRANTED BY regre
+@@ -172,12 +172,16 @@ GRANT regress_priv_role TO regress_priv_user1 WITH ADMIN OPTION GRANTED BY regre
  ERROR:  permission denied to grant privileges as role "regress_priv_role"
  DETAIL:  The grantor must have the ADMIN option on role "regress_priv_role".
  GRANT regress_priv_role TO regress_priv_user1 WITH ADMIN OPTION GRANTED BY CURRENT_ROLE;
@@ -1213,7 +1213,7 @@ index 5b9dba7b32..cc408dad42 100644
  DROP ROLE regress_priv_role;
  SET SESSION AUTHORIZATION regress_priv_user1;
  SELECT session_user, current_user;
-@@ -1776,7 +1780,7 @@ SELECT has_table_privilege('regress_priv_user1', 'atest4', 'SELECT WITH GRANT OP
+@@ -1709,7 +1713,7 @@ SELECT has_table_privilege('regress_priv_user1', 'atest4', 'SELECT WITH GRANT OP
  
  -- security-restricted operations
  \c -
@@ -1222,7 +1222,7 @@ index 5b9dba7b32..cc408dad42 100644
  -- Check that index expressions and predicates are run as the table's owner
  -- A dummy index function checking current_user
  CREATE FUNCTION sro_ifun(int) RETURNS int AS $$
-@@ -2668,8 +2672,8 @@ drop cascades to function testns.priv_testagg(integer)
+@@ -2601,8 +2605,8 @@ drop cascades to function testns.priv_testagg(integer)
  drop cascades to function testns.priv_testproc(integer)
  -- Change owner of the schema & and rename of new schema owner
  \c -
@@ -1233,7 +1233,7 @@ index 5b9dba7b32..cc408dad42 100644
  SET SESSION ROLE regress_schemauser1;
  CREATE SCHEMA testns;
  SELECT nspname, rolname FROM pg_namespace, pg_roles WHERE pg_namespace.nspname = 'testns' AND pg_namespace.nspowner = pg_roles.oid;
-@@ -2792,7 +2796,7 @@ DROP USER regress_priv_user7;
+@@ -2725,7 +2729,7 @@ DROP USER regress_priv_user7;
  DROP USER regress_priv_user8; -- does not exist
  ERROR:  role "regress_priv_user8" does not exist
  -- permissions with LOCK TABLE
@@ -1242,7 +1242,7 @@ index 5b9dba7b32..cc408dad42 100644
  CREATE TABLE lock_table (a int);
  -- LOCK TABLE and SELECT permission
  GRANT SELECT ON lock_table TO regress_locktable_user;
-@@ -2874,7 +2878,7 @@ DROP USER regress_locktable_user;
+@@ -2807,7 +2811,7 @@ DROP USER regress_locktable_user;
  -- pg_backend_memory_contexts.
  -- switch to superuser
  \c -
@@ -1251,7 +1251,7 @@ index 5b9dba7b32..cc408dad42 100644
  SELECT has_table_privilege('regress_readallstats','pg_backend_memory_contexts','SELECT'); -- no
   has_table_privilege 
  ---------------------
-@@ -2918,10 +2922,10 @@ RESET ROLE;
+@@ -2851,10 +2855,10 @@ RESET ROLE;
  -- clean up
  DROP ROLE regress_readallstats;
  -- test role grantor machinery
@@ -1266,7 +1266,7 @@ index 5b9dba7b32..cc408dad42 100644
  GRANT regress_group TO regress_group_direct_manager WITH INHERIT FALSE, ADMIN TRUE;
  GRANT regress_group_direct_manager TO regress_group_indirect_manager;
  SET SESSION AUTHORIZATION regress_group_direct_manager;
-@@ -2950,9 +2954,9 @@ DROP ROLE regress_group_direct_manager;
+@@ -2883,9 +2887,9 @@ DROP ROLE regress_group_direct_manager;
  DROP ROLE regress_group_indirect_manager;
  DROP ROLE regress_group_member;
  -- test SET and INHERIT options with object ownership changes
@@ -1813,7 +1813,7 @@ index 5e6969b173..2c4d52237f 100644
  
  -- clean up roles
 diff --git a/src/test/regress/expected/rowsecurity.out b/src/test/regress/expected/rowsecurity.out
-index 218c0c2863..f7af0cfb12 100644
+index 97ca9bf72c..b2a7a6f710 100644
 --- a/src/test/regress/expected/rowsecurity.out
 +++ b/src/test/regress/expected/rowsecurity.out
 @@ -14,13 +14,13 @@ DROP ROLE IF EXISTS regress_rls_group2;
@@ -1917,19 +1917,6 @@ index b79fe9a1c0..e29fab88ab 100644
  ALTER DEFAULT PRIVILEGES FOR ROLE regress_selinto_user
  	  REVOKE INSERT ON TABLES FROM regress_selinto_user;
  GRANT ALL ON SCHEMA selinto_schema TO public;
-diff --git a/src/test/regress/expected/select_parallel.out b/src/test/regress/expected/select_parallel.out
-index afc6ab08c2..dfcd891af3 100644
---- a/src/test/regress/expected/select_parallel.out
-+++ b/src/test/regress/expected/select_parallel.out
-@@ -1220,7 +1220,7 @@ SELECT 1 FROM tenk1_vw_sec
- 
- rollback;
- -- test that function option SET ROLE works in parallel workers.
--create role regress_parallel_worker;
-+create role regress_parallel_worker PASSWORD NEON_PASSWORD_PLACEHOLDER;
- create function set_and_report_role() returns text as
-   $$ select current_setting('role') $$ language sql parallel safe
-   set role = regress_parallel_worker;
 diff --git a/src/test/regress/expected/select_views.out b/src/test/regress/expected/select_views.out
 index 1aeed8452b..7d9427d070 100644
 --- a/src/test/regress/expected/select_views.out
@@ -2382,7 +2369,7 @@ index 6cb9c926c0..5e689e4062 100644
  ALTER TABLE ptnowner1 OWNER TO regress_ptnowner;
  ALTER TABLE ptnowner OWNER TO regress_ptnowner;
 diff --git a/src/test/regress/sql/collate.icu.utf8.sql b/src/test/regress/sql/collate.icu.utf8.sql
-index 8aa902d5ab..24bb823b86 100644
+index 3db9e25913..c66d5aa2c2 100644
 --- a/src/test/regress/sql/collate.icu.utf8.sql
 +++ b/src/test/regress/sql/collate.icu.utf8.sql
 @@ -353,7 +353,7 @@ reset enable_seqscan;
@@ -2545,7 +2532,7 @@ index 43d2e906dd..6c993d70f0 100644
  -- An earlier bug (see commit b1ecb9b3fcf) could end up using a buffer from
  -- the wrong partition. This test is *not* guaranteed to trigger that bug, but
 diff --git a/src/test/regress/sql/copy2.sql b/src/test/regress/sql/copy2.sql
-index cf3828c16e..cf3ca38175 100644
+index d759635068..d58e50dcc5 100644
 --- a/src/test/regress/sql/copy2.sql
 +++ b/src/test/regress/sql/copy2.sql
 @@ -365,8 +365,8 @@ copy check_con_tbl from stdin;
@@ -2787,7 +2774,7 @@ index 1b7064247a..be5b662ce1 100644
  -- Cases where schema creation fails as objects are qualified with a schema
  -- that does not match with what's expected.
 diff --git a/src/test/regress/sql/create_view.sql b/src/test/regress/sql/create_view.sql
-index ae6841308b..47bc792e30 100644
+index 3a78be1b0c..617d2dc8d6 100644
 --- a/src/test/regress/sql/create_view.sql
 +++ b/src/test/regress/sql/create_view.sql
 @@ -23,7 +23,8 @@ CREATE TABLE real_city (
@@ -2914,11 +2901,11 @@ index aa147b14a9..370e0dd570 100644
  CREATE FOREIGN DATA WRAPPER dummy;
  COMMENT ON FOREIGN DATA WRAPPER dummy IS 'useless';
 diff --git a/src/test/regress/sql/foreign_key.sql b/src/test/regress/sql/foreign_key.sql
-index 45c7a534cb..32dd26b8cd 100644
+index 22e177f89b..7138d5e1d4 100644
 --- a/src/test/regress/sql/foreign_key.sql
 +++ b/src/test/regress/sql/foreign_key.sql
-@@ -1435,7 +1435,7 @@ ALTER TABLE fk_partitioned_fk_6 ATTACH PARTITION fk_partitioned_pk_6 FOR VALUES
- DROP TABLE fk_partitioned_pk_6, fk_partitioned_fk_6;
+@@ -1418,7 +1418,7 @@ ALTER TABLE fk_partitioned_fk ATTACH PARTITION fk_partitioned_fk_2
+ -- leave these tables around intentionally
  
  -- test the case when the referenced table is owned by a different user
 -create role regress_other_partitioned_fk_owner;
@@ -2976,7 +2963,7 @@ index 527024f710..de49c0b85f 100644
  -- the data in this file has a lot of duplicates in the index key
  -- fields, leading to long bucket chains and lots of table expansion.
 diff --git a/src/test/regress/sql/identity.sql b/src/test/regress/sql/identity.sql
-index 7537258a75..9041e35e34 100644
+index 91d2e443b4..241c93f373 100644
 --- a/src/test/regress/sql/identity.sql
 +++ b/src/test/regress/sql/identity.sql
 @@ -287,7 +287,7 @@ ALTER TABLE itest7 ALTER COLUMN a RESTART;
@@ -2989,10 +2976,10 @@ index 7537258a75..9041e35e34 100644
  GRANT SELECT, INSERT ON itest8 TO regress_identity_user1;
  SET ROLE regress_identity_user1;
 diff --git a/src/test/regress/sql/inherit.sql b/src/test/regress/sql/inherit.sql
-index b5b554a125..109889ad24 100644
+index fe699c54d5..bdd5993f45 100644
 --- a/src/test/regress/sql/inherit.sql
 +++ b/src/test/regress/sql/inherit.sql
-@@ -958,7 +958,7 @@ create index on permtest_parent (left(c, 3));
+@@ -950,7 +950,7 @@ create index on permtest_parent (left(c, 3));
  insert into permtest_parent
    select 1, 'a', left(fipshash(i::text), 5) from generate_series(0, 100) i;
  analyze permtest_parent;
@@ -3231,7 +3218,7 @@ index 53e86b0b6c..f07cf1ec54 100644
  CREATE ROLE regress_passwd5 PASSWORD 'md5e73a4b11df52a6068f8b39f90be36023';
  
 diff --git a/src/test/regress/sql/privileges.sql b/src/test/regress/sql/privileges.sql
-index 249df17a58..b258e7f26a 100644
+index 3f68cafcd1..004b26831d 100644
 --- a/src/test/regress/sql/privileges.sql
 +++ b/src/test/regress/sql/privileges.sql
 @@ -24,18 +24,18 @@ RESET client_min_messages;
@@ -3282,7 +3269,7 @@ index 249df17a58..b258e7f26a 100644
  
  GRANT pg_read_all_data TO regress_priv_user6;
  GRANT pg_write_all_data TO regress_priv_user7;
-@@ -163,8 +163,8 @@ DROP USER regress_priv_user10;
+@@ -130,8 +130,8 @@ DROP USER regress_priv_user10;
  DROP USER regress_priv_user9;
  DROP USER regress_priv_user8;
  
@@ -3293,7 +3280,7 @@ index 249df17a58..b258e7f26a 100644
  
  ALTER GROUP regress_priv_group1 ADD USER regress_priv_user4;
  
-@@ -1157,7 +1157,7 @@ SELECT has_table_privilege('regress_priv_user1', 'atest4', 'SELECT WITH GRANT OP
+@@ -1124,7 +1124,7 @@ SELECT has_table_privilege('regress_priv_user1', 'atest4', 'SELECT WITH GRANT OP
  
  -- security-restricted operations
  \c -
@@ -3302,7 +3289,7 @@ index 249df17a58..b258e7f26a 100644
  
  -- Check that index expressions and predicates are run as the table's owner
  
-@@ -1653,8 +1653,8 @@ DROP SCHEMA testns CASCADE;
+@@ -1620,8 +1620,8 @@ DROP SCHEMA testns CASCADE;
  -- Change owner of the schema & and rename of new schema owner
  \c -
  
@@ -3313,7 +3300,7 @@ index 249df17a58..b258e7f26a 100644
  
  SET SESSION ROLE regress_schemauser1;
  CREATE SCHEMA testns;
-@@ -1748,7 +1748,7 @@ DROP USER regress_priv_user8; -- does not exist
+@@ -1715,7 +1715,7 @@ DROP USER regress_priv_user8; -- does not exist
  
  
  -- permissions with LOCK TABLE
@@ -3322,7 +3309,7 @@ index 249df17a58..b258e7f26a 100644
  CREATE TABLE lock_table (a int);
  
  -- LOCK TABLE and SELECT permission
-@@ -1836,7 +1836,7 @@ DROP USER regress_locktable_user;
+@@ -1803,7 +1803,7 @@ DROP USER regress_locktable_user;
  -- switch to superuser
  \c -
  
@@ -3331,7 +3318,7 @@ index 249df17a58..b258e7f26a 100644
  
  SELECT has_table_privilege('regress_readallstats','pg_backend_memory_contexts','SELECT'); -- no
  SELECT has_table_privilege('regress_readallstats','pg_shmem_allocations','SELECT'); -- no
-@@ -1856,10 +1856,10 @@ RESET ROLE;
+@@ -1823,10 +1823,10 @@ RESET ROLE;
  DROP ROLE regress_readallstats;
  
  -- test role grantor machinery
@@ -3346,7 +3333,7 @@ index 249df17a58..b258e7f26a 100644
  
  GRANT regress_group TO regress_group_direct_manager WITH INHERIT FALSE, ADMIN TRUE;
  GRANT regress_group_direct_manager TO regress_group_indirect_manager;
-@@ -1881,9 +1881,9 @@ DROP ROLE regress_group_indirect_manager;
+@@ -1848,9 +1848,9 @@ DROP ROLE regress_group_indirect_manager;
  DROP ROLE regress_group_member;
  
  -- test SET and INHERIT options with object ownership changes
@@ -3638,7 +3625,7 @@ index c961b2d730..0859b89c4f 100644
  -- clean up roles
  DROP ROLE regress_test_def_superuser;
 diff --git a/src/test/regress/sql/rowsecurity.sql b/src/test/regress/sql/rowsecurity.sql
-index d3bfd53e23..919ce1d0c6 100644
+index dec7340538..cdbc03a5cc 100644
 --- a/src/test/regress/sql/rowsecurity.sql
 +++ b/src/test/regress/sql/rowsecurity.sql
 @@ -20,13 +20,13 @@ DROP SCHEMA IF EXISTS regress_rls_schema CASCADE;
@@ -3714,19 +3701,6 @@ index 689c448cc2..223ceb1d75 100644
  ALTER DEFAULT PRIVILEGES FOR ROLE regress_selinto_user
  	  REVOKE INSERT ON TABLES FROM regress_selinto_user;
  GRANT ALL ON SCHEMA selinto_schema TO public;
-diff --git a/src/test/regress/sql/select_parallel.sql b/src/test/regress/sql/select_parallel.sql
-index 33d78e16dc..cb193c9b27 100644
---- a/src/test/regress/sql/select_parallel.sql
-+++ b/src/test/regress/sql/select_parallel.sql
-@@ -464,7 +464,7 @@ SELECT 1 FROM tenk1_vw_sec
- rollback;
- 
- -- test that function option SET ROLE works in parallel workers.
--create role regress_parallel_worker;
-+create role regress_parallel_worker PASSWORD NEON_PASSWORD_PLACEHOLDER;
- 
- create function set_and_report_role() returns text as
-   $$ select current_setting('role') $$ language sql parallel safe
 diff --git a/src/test/regress/sql/select_views.sql b/src/test/regress/sql/select_views.sql
 index e742f13699..7bd0255df8 100644
 --- a/src/test/regress/sql/select_views.sql

compute/patches/pg_anon.patch

@@ -1,45 +1,3 @@
-commit 00aa659afc9c7336ab81036edec3017168aabf40
-Author: Heikki Linnakangas <heikki@neon.tech>
-Date:   Tue Nov 12 16:59:19 2024 +0200
-
-    Temporarily disable test that depends on timezone
-
-diff --git a/tests/expected/generalization.out b/tests/expected/generalization.out
-index 23ef5fa..9e60deb 100644
---- a/ext-src/pg_anon-src/tests/expected/generalization.out
-+++ b/ext-src/pg_anon-src/tests/expected/generalization.out
-@@ -284,12 +284,9 @@ SELECT anon.generalize_tstzrange('19041107','century');
-  ["Tue Jan 01 00:00:00 1901 PST","Mon Jan 01 00:00:00 2001 PST")
- (1 row)
- 
--SELECT anon.generalize_tstzrange('19041107','millennium');
--                      generalize_tstzrange                       
-------------------------------------------------------------------
-- ["Thu Jan 01 00:00:00 1001 PST","Mon Jan 01 00:00:00 2001 PST")
--(1 row)
--
-+-- temporarily disabled, see:
-+-- https://gitlab.com/dalibo/postgresql_anonymizer/-/commit/199f0a392b37c59d92ae441fb8f037e094a11a52#note_2148017485
-+--SELECT anon.generalize_tstzrange('19041107','millennium');
- -- generalize_daterange
- SELECT anon.generalize_daterange('19041107');
-   generalize_daterange   
-diff --git a/tests/sql/generalization.sql b/tests/sql/generalization.sql
-index b868344..b4fc977 100644
---- a/ext-src/pg_anon-src/tests/sql/generalization.sql
-+++ b/ext-src/pg_anon-src/tests/sql/generalization.sql
-@@ -61,7 +61,9 @@ SELECT anon.generalize_tstzrange('19041107','month');
- SELECT anon.generalize_tstzrange('19041107','year');
- SELECT anon.generalize_tstzrange('19041107','decade');
- SELECT anon.generalize_tstzrange('19041107','century');
--SELECT anon.generalize_tstzrange('19041107','millennium');
-+-- temporarily disabled, see:
-+-- https://gitlab.com/dalibo/postgresql_anonymizer/-/commit/199f0a392b37c59d92ae441fb8f037e094a11a52#note_2148017485
-+--SELECT anon.generalize_tstzrange('19041107','millennium');
- 
- -- generalize_daterange
- SELECT anon.generalize_daterange('19041107');
-
 commit 7dd414ee75f2875cffb1d6ba474df1f135a6fc6f
 Author: Alexey Masterov <alexeymasterov@neon.tech>
 Date:   Fri May 31 06:34:26 2024 +0000

compute_tools/Cargo.toml

@@ -18,11 +18,9 @@ clap.workspace = true
 flate2.workspace = true
 futures.workspace = true
 hyper0 = { workspace = true, features = ["full"] }
-metrics.workspace = true
 nix.workspace = true
 notify.workspace = true
 num_cpus.workspace = true
-once_cell.workspace = true
 opentelemetry.workspace = true
 opentelemetry_sdk.workspace = true
 postgres.workspace = true
@@ -41,7 +39,6 @@ tracing-subscriber.workspace = true
 tracing-utils.workspace = true
 thiserror.workspace = true
 url.workspace = true
-prometheus.workspace = true
 
 compute_api.workspace = true
 utils.workspace = true

compute_tools/src/http/api.rs

@@ -9,7 +9,6 @@ use crate::catalog::SchemaDumpError;
 use crate::catalog::{get_database_schema, get_dbs_and_roles};
 use crate::compute::forward_termination_signal;
 use crate::compute::{ComputeNode, ComputeState, ParsedSpec};
-use crate::installed_extensions;
 use compute_api::requests::{ConfigurationRequest, ExtensionInstallRequest, SetRoleGrantsRequest};
 use compute_api::responses::{
     ComputeStatus, ComputeStatusResponse, ExtensionInstallResult, GenericAPIError,
@@ -20,8 +19,6 @@ use anyhow::Result;
 use hyper::header::CONTENT_TYPE;
 use hyper::service::{make_service_fn, service_fn};
 use hyper::{Body, Method, Request, Response, Server, StatusCode};
-use metrics::Encoder;
-use metrics::TextEncoder;
 use tokio::task;
 use tracing::{debug, error, info, warn};
 use tracing_utils::http::OtelName;
@@ -68,28 +65,6 @@ async fn routes(req: Request<Body>, compute: &Arc<ComputeNode>) -> Response<Body
             Response::new(Body::from(serde_json::to_string(&metrics).unwrap()))
         }
 
-        // Prometheus metrics
-        (&Method::GET, "/metrics") => {
-            debug!("serving /metrics GET request");
-
-            let mut buffer = vec![];
-            let metrics = installed_extensions::collect();
-            let encoder = TextEncoder::new();
-            encoder.encode(&metrics, &mut buffer).unwrap();
-
-            match Response::builder()
-                .status(StatusCode::OK)
-                .header(CONTENT_TYPE, encoder.format_type())
-                .body(Body::from(buffer))
-            {
-                Ok(response) => response,
-                Err(err) => {
-                    let msg = format!("error handling /metrics request: {err}");
-                    error!(msg);
-                    render_json_error(&msg, StatusCode::INTERNAL_SERVER_ERROR)
-                }
-            }
-        }
         // Collect Postgres current usage insights
         (&Method::GET, "/insights") => {
             info!("serving /insights GET request");

compute_tools/src/http/openapi_spec.yaml

@@ -37,21 +37,6 @@ paths:
               schema:
                 $ref: "#/components/schemas/ComputeMetrics"
 
-  /metrics
-    get:
-      tags:
-      - Info
-      summary: Get compute node metrics in text format.
-      description: ""
-      operationId: getComputeMetrics
-      responses:
-        200:
-          description: ComputeMetrics
-          content:
-            text/plain:
-              schema:
-                type: string
-                description: Metrics in text format.
   /insights:
     get:
       tags:

compute_tools/src/installed_extensions.rs

@@ -1,5 +1,4 @@
 use compute_api::responses::{InstalledExtension, InstalledExtensions};
-use metrics::proto::MetricFamily;
 use std::collections::HashMap;
 use std::collections::HashSet;
 use tracing::info;
@@ -9,10 +8,6 @@ use anyhow::Result;
 use postgres::{Client, NoTls};
 use tokio::task;
 
-use metrics::core::Collector;
-use metrics::{register_uint_gauge_vec, UIntGaugeVec};
-use once_cell::sync::Lazy;
-
 /// We don't reuse get_existing_dbs() just for code clarity
 /// and to make database listing query here more explicit.
 ///
@@ -64,12 +59,6 @@ pub async fn get_installed_extensions(connstr: Url) -> Result<InstalledExtension
 
             for (extname, v) in extensions.iter() {
                 let version = v.to_string();
-
-                // increment the number of databases where the version of extension is installed
-                INSTALLED_EXTENSIONS
-                    .with_label_values(&[extname, &version])
-                    .inc();
-
                 extensions_map
                     .entry(extname.to_string())
                     .and_modify(|e| {
@@ -85,11 +74,9 @@ pub async fn get_installed_extensions(connstr: Url) -> Result<InstalledExtension
             }
         }
 
-        let res = InstalledExtensions {
+        Ok(InstalledExtensions {
             extensions: extensions_map.values().cloned().collect(),
-        };
-
-        Ok(res)
+        })
     })
     .await?
 }
@@ -110,18 +97,6 @@ pub fn get_installed_extensions_sync(connstr: Url) -> Result<()> {
         "[NEON_EXT_STAT] {}",
         serde_json::to_string(&result).expect("failed to serialize extensions list")
     );
+
     Ok(())
 }
-
-static INSTALLED_EXTENSIONS: Lazy<UIntGaugeVec> = Lazy::new(|| {
-    register_uint_gauge_vec!(
-        "installed_extensions",
-        "Number of databases where the version of extension is installed",
-        &["extension_name", "version"]
-    )
-    .expect("failed to define a metric")
-});
-
-pub fn collect() -> Vec<MetricFamily> {
-    INSTALLED_EXTENSIONS.collect()
-}

deny.toml

@@ -37,7 +37,6 @@ allow = [
     "BSD-2-Clause",
     "BSD-3-Clause",
     "CC0-1.0",
-    "CDDL-1.0",
     "ISC",
     "MIT",
     "MPL-2.0",

libs/pageserver_api/src/config.rs

@@ -18,7 +18,7 @@ use std::{
     str::FromStr,
     time::Duration,
 };
-use utils::{logging::LogFormat, postgres_client::PostgresClientProtocol};
+use utils::logging::LogFormat;
 
 use crate::models::ImageCompressionAlgorithm;
 use crate::models::LsnLease;
@@ -109,7 +109,6 @@ pub struct ConfigToml {
     pub virtual_file_io_mode: Option<crate::models::virtual_file::IoMode>,
     #[serde(skip_serializing_if = "Option::is_none")]
     pub no_sync: Option<bool>,
-    pub wal_receiver_protocol: PostgresClientProtocol,
 }
 
 #[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
@@ -318,9 +317,6 @@ pub mod defaults {
     pub const DEFAULT_EPHEMERAL_BYTES_PER_MEMORY_KB: usize = 0;
 
     pub const DEFAULT_IO_BUFFER_ALIGNMENT: usize = 512;
-
-    pub const DEFAULT_WAL_RECEIVER_PROTOCOL: utils::postgres_client::PostgresClientProtocol =
-        utils::postgres_client::PostgresClientProtocol::Interpreted;
 }
 
 impl Default for ConfigToml {
@@ -403,7 +399,6 @@ impl Default for ConfigToml {
             virtual_file_io_mode: None,
             tenant_config: TenantConfigToml::default(),
             no_sync: None,
-            wal_receiver_protocol: DEFAULT_WAL_RECEIVER_PROTOCOL,
         }
     }
 }

libs/pageserver_api/src/key.rs

@@ -24,7 +24,7 @@ pub struct Key {
 
 /// When working with large numbers of Keys in-memory, it is more efficient to handle them as i128 than as
 /// a struct of fields.
-#[derive(Clone, Copy, Hash, PartialEq, Eq, Ord, PartialOrd, Serialize, Deserialize)]
+#[derive(Clone, Copy, Hash, PartialEq, Eq, Ord, PartialOrd)]
 pub struct CompactKey(i128);
 
 /// The storage key size.

libs/pageserver_api/src/record.rs

@@ -41,11 +41,6 @@ pub enum NeonWalRecord {
         file_path: String,
         content: Option<Bytes>,
     },
-    // Truncate visibility map page
-    TruncateVisibilityMap {
-        trunc_byte: usize,
-        trunc_offs: usize,
-    },
 
     /// A testing record for unit testing purposes. It supports append data to an existing image, or clear it.
     #[cfg(feature = "testing")]
@@ -85,18 +80,18 @@ impl NeonWalRecord {
     }
 
     #[cfg(feature = "testing")]
-    pub fn wal_clear(s: impl AsRef<str>) -> Self {
+    pub fn wal_clear() -> Self {
         Self::Test {
-            append: s.as_ref().to_string(),
+            append: "".to_string(),
             clear: true,
             will_init: false,
         }
     }
 
     #[cfg(feature = "testing")]
-    pub fn wal_init(s: impl AsRef<str>) -> Self {
+    pub fn wal_init() -> Self {
         Self::Test {
-            append: s.as_ref().to_string(),
+            append: "".to_string(),
             clear: true,
             will_init: true,
         }

libs/pageserver_api/src/reltag.rs

@@ -24,7 +24,7 @@ use postgres_ffi::Oid;
 // FIXME: should move 'forknum' as last field to keep this consistent with Postgres.
 // Then we could replace the custom Ord and PartialOrd implementations below with
 // deriving them. This will require changes in walredoproc.c.
-#[derive(Debug, PartialEq, Eq, Hash, Clone, Copy, Serialize, Deserialize)]
+#[derive(Debug, PartialEq, Eq, Hash, Clone, Copy, Serialize)]
 pub struct RelTag {
     pub forknum: u8,
     pub spcnode: Oid,

libs/postgres_ffi/src/pg_constants.rs

@@ -243,11 +243,8 @@ const FSM_LEAF_NODES_PER_PAGE: usize = FSM_NODES_PER_PAGE - FSM_NON_LEAF_NODES_P
 pub const SLOTS_PER_FSM_PAGE: u32 = FSM_LEAF_NODES_PER_PAGE as u32;
 
 /* From visibilitymap.c */
-
-pub const VM_MAPSIZE: usize = BLCKSZ as usize - MAXALIGN_SIZE_OF_PAGE_HEADER_DATA;
-pub const VM_BITS_PER_HEAPBLOCK: usize = 2;
-pub const VM_HEAPBLOCKS_PER_BYTE: usize = 8 / VM_BITS_PER_HEAPBLOCK;
-pub const VM_HEAPBLOCKS_PER_PAGE: usize = VM_MAPSIZE * VM_HEAPBLOCKS_PER_BYTE;
+pub const VM_HEAPBLOCKS_PER_PAGE: u32 =
+    (BLCKSZ as usize - SIZEOF_PAGE_HEADER_DATA) as u32 * (8 / 2); // MAPSIZE * (BITS_PER_BYTE / BITS_PER_HEAPBLOCK)
 
 /* From origin.c */
 pub const REPLICATION_STATE_MAGIC: u32 = 0x1257DADE;

libs/postgres_ffi/src/walrecord.rs

@@ -16,7 +16,7 @@ use utils::bin_ser::DeserializeError;
 use utils::lsn::Lsn;
 
 #[repr(C)]
-#[derive(Debug, Serialize, Deserialize)]
+#[derive(Debug)]
 pub struct XlMultiXactCreate {
     pub mid: MultiXactId,
     /* new MultiXact's ID */
@@ -46,7 +46,7 @@ impl XlMultiXactCreate {
 }
 
 #[repr(C)]
-#[derive(Debug, Serialize, Deserialize)]
+#[derive(Debug)]
 pub struct XlMultiXactTruncate {
     pub oldest_multi_db: Oid,
     /* to-be-truncated range of multixact offsets */
@@ -72,7 +72,7 @@ impl XlMultiXactTruncate {
 }
 
 #[repr(C)]
-#[derive(Debug, Serialize, Deserialize)]
+#[derive(Debug)]
 pub struct XlRelmapUpdate {
     pub dbid: Oid,   /* database ID, or 0 for shared map */
     pub tsid: Oid,   /* database's tablespace, or pg_global */
@@ -90,7 +90,7 @@ impl XlRelmapUpdate {
 }
 
 #[repr(C)]
-#[derive(Debug, Serialize, Deserialize)]
+#[derive(Debug)]
 pub struct XlReploriginDrop {
     pub node_id: RepOriginId,
 }
@@ -104,7 +104,7 @@ impl XlReploriginDrop {
 }
 
 #[repr(C)]
-#[derive(Debug, Serialize, Deserialize)]
+#[derive(Debug)]
 pub struct XlReploriginSet {
     pub remote_lsn: Lsn,
     pub node_id: RepOriginId,
@@ -120,7 +120,7 @@ impl XlReploriginSet {
 }
 
 #[repr(C)]
-#[derive(Debug, Clone, Copy, Serialize, Deserialize)]
+#[derive(Debug, Clone, Copy)]
 pub struct RelFileNode {
     pub spcnode: Oid, /* tablespace */
     pub dbnode: Oid,  /* database */
@@ -911,7 +911,7 @@ impl XlSmgrCreate {
 }
 
 #[repr(C)]
-#[derive(Debug, Serialize, Deserialize)]
+#[derive(Debug)]
 pub struct XlSmgrTruncate {
     pub blkno: BlockNumber,
     pub rnode: RelFileNode,
@@ -984,7 +984,7 @@ impl XlDropDatabase {
 /// xl_xact_parsed_abort structs in PostgreSQL, but we use the same
 /// struct for commits and aborts.
 ///
-#[derive(Debug, Serialize, Deserialize)]
+#[derive(Debug)]
 pub struct XlXactParsedRecord {
     pub xid: TransactionId,
     pub info: u8,

libs/pq_proto/src/lib.rs

@@ -562,9 +562,6 @@ pub enum BeMessage<'a> {
         options: &'a [&'a str],
     },
     KeepAlive(WalSndKeepAlive),
-    /// Batch of interpreted, shard filtered WAL records,
-    /// ready for the pageserver to ingest
-    InterpretedWalRecords(InterpretedWalRecordsBody<'a>),
 }
 
 /// Common shorthands.
@@ -675,18 +672,6 @@ pub struct WalSndKeepAlive {
     pub request_reply: bool,
 }
 
-#[derive(Debug)]
-pub struct InterpretedWalRecordsBody<'a> {
-    /// End of raw WAL in [`Self::data`]
-    pub streaming_lsn: u64,
-    /// Current end of WAL on the server
-    pub commit_lsn: u64,
-    /// Start LSN of the next record in PG WAL.
-    /// Is 0 if the portion of PG WAL did not contain any records.
-    pub next_record_lsn: u64,
-    pub data: &'a [u8],
-}
-
 pub static HELLO_WORLD_ROW: BeMessage = BeMessage::DataRow(&[Some(b"hello world")]);
 
 // single text column
@@ -1011,20 +996,6 @@ impl BeMessage<'_> {
                     Ok(())
                 })?
             }
-
-            BeMessage::InterpretedWalRecords(rec) => {
-                // We use the COPY_DATA_TAG for our custom message
-                // since this tag is interpreted as raw bytes.
-                buf.put_u8(b'd');
-                write_body(buf, |buf| {
-                    buf.put_u8(b'0'); // matches INTERPRETED_WAL_RECORD_TAG in postgres-protocol
-                                      // dependency
-                    buf.put_u64(rec.streaming_lsn);
-                    buf.put_u64(rec.commit_lsn);
-                    buf.put_u64(rec.next_record_lsn);
-                    buf.put_slice(rec.data);
-                });
-            }
         }
         Ok(())
     }

libs/remote_storage/src/error.rs

@@ -15,9 +15,6 @@ pub enum DownloadError {
     ///
     /// Concurrency control is not timed within timeout.
     Timeout,
-    /// Some integrity/consistency check failed during download. This is used during
-    /// timeline loads to cancel the load of a tenant if some timeline detects fatal corruption.
-    Fatal(String),
     /// The file was found in the remote storage, but the download failed.
     Other(anyhow::Error),
 }
@@ -32,7 +29,6 @@ impl std::fmt::Display for DownloadError {
             DownloadError::Unmodified => write!(f, "File was not modified"),
             DownloadError::Cancelled => write!(f, "Cancelled, shutting down"),
             DownloadError::Timeout => write!(f, "timeout"),
-            DownloadError::Fatal(why) => write!(f, "Fatal read error: {why}"),
             DownloadError::Other(e) => write!(f, "Failed to download a remote file: {e:?}"),
         }
     }
@@ -45,7 +41,7 @@ impl DownloadError {
     pub fn is_permanent(&self) -> bool {
         use DownloadError::*;
         match self {
-            BadInput(_) | NotFound | Unmodified | Fatal(_) | Cancelled => true,
+            BadInput(_) | NotFound | Unmodified | Cancelled => true,
             Timeout | Other(_) => false,
         }
     }

libs/utils/Cargo.toml

@@ -32,7 +32,6 @@ pin-project-lite.workspace = true
 regex.workspace = true
 routerify.workspace = true
 serde.workspace = true
-serde_with.workspace = true
 serde_json.workspace = true
 signal-hook.workspace = true
 thiserror.workspace = true

libs/utils/scripts/restore_from_wal.sh

@@ -50,8 +50,8 @@ REDO_POS=0x$("$PG_BIN"/pg_controldata -D "$DATA_DIR" | grep -F "REDO location"|
 declare -i WAL_SIZE=$REDO_POS+114
 "$PG_BIN"/pg_ctl -D "$DATA_DIR" -l "$DATA_DIR/logfile.log" start
 "$PG_BIN"/pg_ctl -D "$DATA_DIR" -l "$DATA_DIR/logfile.log" stop -m immediate
-cp "$DATA_DIR"/pg_wal/000000010000000000000001 "$DATA_DIR"
+cp "$DATA_DIR"/pg_wal/000000010000000000000001 .
 cp "$WAL_PATH"/* "$DATA_DIR"/pg_wal/
 for partial in "$DATA_DIR"/pg_wal/*.partial ; do mv "$partial" "${partial%.partial}" ; done
-dd if="$DATA_DIR"/000000010000000000000001 of="$DATA_DIR"/pg_wal/000000010000000000000001 bs=$WAL_SIZE count=1 conv=notrunc
-rm -f "$DATA_DIR"/000000010000000000000001
+dd if=000000010000000000000001 of="$DATA_DIR"/pg_wal/000000010000000000000001 bs=$WAL_SIZE count=1 conv=notrunc
+rm -f 000000010000000000000001

libs/utils/scripts/restore_from_wal_initdb.sh

@@ -14,8 +14,8 @@ REDO_POS=0x$("$PG_BIN"/pg_controldata -D "$DATA_DIR" | grep -F "REDO location"|
 declare -i WAL_SIZE=$REDO_POS+114
 "$PG_BIN"/pg_ctl -D "$DATA_DIR" -l "$DATA_DIR/logfile.log" start
 "$PG_BIN"/pg_ctl -D "$DATA_DIR" -l "$DATA_DIR/logfile.log" stop -m immediate
-cp "$DATA_DIR"/pg_wal/000000010000000000000001 "$DATA_DIR"
+cp "$DATA_DIR"/pg_wal/000000010000000000000001 .
 cp "$WAL_PATH"/* "$DATA_DIR"/pg_wal/
 for partial in "$DATA_DIR"/pg_wal/*.partial ; do mv "$partial" "${partial%.partial}" ; done
-dd if="$DATA_DIR"/000000010000000000000001 of="$DATA_DIR"/pg_wal/000000010000000000000001 bs=$WAL_SIZE count=1 conv=notrunc
-rm -f "$DATA_DIR"/000000010000000000000001
+dd if=000000010000000000000001 of="$DATA_DIR"/pg_wal/000000010000000000000001 bs=$WAL_SIZE count=1 conv=notrunc
+rm -f 000000010000000000000001

libs/utils/src/crashsafe.rs

@@ -123,27 +123,15 @@ pub async fn fsync_async_opt(
     Ok(())
 }
 
-/// Like postgres' durable_rename, renames a file and issues fsyncs to make it durable. After
-/// returning, both the file and rename are guaranteed to be persisted. Both paths must be on the
-/// same file system.
+/// Like postgres' durable_rename, renames file issuing fsyncs do make it
+/// durable. After return, file and rename are guaranteed to be persisted.
 ///
-/// Unlike postgres, it only fsyncs 1) the file to make contents durable, and 2) the directory to
-/// make the rename durable. This sequence ensures the target file will never be incomplete.
-///
-/// Postgres also:
-///
-/// * Fsyncs the target file, if it exists, before the rename, to ensure either the new or existing
-///   file survives a crash. Current callers don't need this as it should already be fsynced if
-///   durability is needed.
-///
-/// * Fsyncs the file after the rename. This can be required with certain OSes or file systems (e.g.
-///   NFS), but not on Linux with most common file systems like ext4 (which we currently use).
-///
-/// An audit of 8 other databases found that none fsynced the file after a rename:
-/// <https://github.com/neondatabase/neon/pull/9686#discussion_r1837180535>
-///
-/// eBPF probes confirmed that this is sufficient with ext4, XFS, and ZFS, but possibly not Btrfs:
-/// <https://github.com/neondatabase/neon/pull/9686#discussion_r1837926218>
+/// Unlike postgres, it only does fsyncs to 1) file to be renamed to make
+/// contents durable; 2) its directory entry to make rename durable 3) again to
+/// already renamed file, which is not required by standards but postgres does
+/// it, let's stick to that. Postgres additionally fsyncs newpath *before*
+/// rename if it exists to ensure that at least one of the files survives, but
+/// current callers don't need that.
 ///
 /// virtual_file.rs has similar code, but it doesn't use vfs.
 ///
@@ -161,6 +149,9 @@ pub async fn durable_rename(
     // Time to do the real deal.
     tokio::fs::rename(old_path.as_ref(), new_path.as_ref()).await?;
 
+    // Postgres'ish fsync of renamed file.
+    fsync_async_opt(new_path.as_ref(), do_fsync).await?;
+
     // Now fsync the parent
     let parent = match new_path.as_ref().parent() {
         Some(p) => p,

libs/utils/src/lsn.rs

@@ -138,11 +138,6 @@ impl Lsn {
         self.0.checked_sub(other).map(Lsn)
     }
 
-    /// Subtract a number, saturating at numeric bounds instead of overflowing.
-    pub fn saturating_sub<T: Into<u64>>(self, other: T) -> Lsn {
-        Lsn(self.0.saturating_sub(other.into()))
-    }
-
     /// Subtract a number, returning the difference as i128 to avoid overflow.
     pub fn widening_sub<T: Into<u64>>(self, other: T) -> i128 {
         let other: u64 = other.into();

libs/utils/src/postgres_client.rs

@@ -7,94 +7,29 @@ use postgres_connection::{parse_host_port, PgConnectionConfig};
 
 use crate::id::TenantTimelineId;
 
-/// Postgres client protocol types
-#[derive(
-    Copy,
-    Clone,
-    PartialEq,
-    Eq,
-    strum_macros::EnumString,
-    strum_macros::Display,
-    serde_with::DeserializeFromStr,
-    serde_with::SerializeDisplay,
-    Debug,
-)]
-#[strum(serialize_all = "kebab-case")]
-#[repr(u8)]
-pub enum PostgresClientProtocol {
-    /// Usual Postgres replication protocol
-    Vanilla,
-    /// Custom shard-aware protocol that replicates interpreted records.
-    /// Used to send wal from safekeeper to pageserver.
-    Interpreted,
-}
-
-impl TryFrom<u8> for PostgresClientProtocol {
-    type Error = u8;
-
-    fn try_from(value: u8) -> Result<Self, Self::Error> {
-        Ok(match value {
-            v if v == (PostgresClientProtocol::Vanilla as u8) => PostgresClientProtocol::Vanilla,
-            v if v == (PostgresClientProtocol::Interpreted as u8) => {
-                PostgresClientProtocol::Interpreted
-            }
-            x => return Err(x),
-        })
-    }
-}
-
-pub struct ConnectionConfigArgs<'a> {
-    pub protocol: PostgresClientProtocol,
-
-    pub ttid: TenantTimelineId,
-    pub shard_number: Option<u8>,
-    pub shard_count: Option<u8>,
-    pub shard_stripe_size: Option<u32>,
-
-    pub listen_pg_addr_str: &'a str,
-
-    pub auth_token: Option<&'a str>,
-    pub availability_zone: Option<&'a str>,
-}
-
-impl<'a> ConnectionConfigArgs<'a> {
-    fn options(&'a self) -> Vec<String> {
-        let mut options = vec![
-            "-c".to_owned(),
-            format!("timeline_id={}", self.ttid.timeline_id),
-            format!("tenant_id={}", self.ttid.tenant_id),
-            format!("protocol={}", self.protocol as u8),
-        ];
-
-        if self.shard_number.is_some() {
-            assert!(self.shard_count.is_some());
-            assert!(self.shard_stripe_size.is_some());
-
-            options.push(format!("shard_count={}", self.shard_count.unwrap()));
-            options.push(format!("shard_number={}", self.shard_number.unwrap()));
-            options.push(format!(
-                "shard_stripe_size={}",
-                self.shard_stripe_size.unwrap()
-            ));
-        }
-
-        options
-    }
-}
-
 /// Create client config for fetching WAL from safekeeper on particular timeline.
 /// listen_pg_addr_str is in form host:\[port\].
 pub fn wal_stream_connection_config(
-    args: ConnectionConfigArgs,
+    TenantTimelineId {
+        tenant_id,
+        timeline_id,
+    }: TenantTimelineId,
+    listen_pg_addr_str: &str,
+    auth_token: Option<&str>,
+    availability_zone: Option<&str>,
 ) -> anyhow::Result<PgConnectionConfig> {
     let (host, port) =
-        parse_host_port(args.listen_pg_addr_str).context("Unable to parse listen_pg_addr_str")?;
+        parse_host_port(listen_pg_addr_str).context("Unable to parse listen_pg_addr_str")?;
     let port = port.unwrap_or(5432);
     let mut connstr = PgConnectionConfig::new_host_port(host, port)
-        .extend_options(args.options())
-        .set_password(args.auth_token.map(|s| s.to_owned()));
+        .extend_options([
+            "-c".to_owned(),
+            format!("timeline_id={}", timeline_id),
+            format!("tenant_id={}", tenant_id),
+        ])
+        .set_password(auth_token.map(|s| s.to_owned()));
 
-    if let Some(availability_zone) = args.availability_zone {
+    if let Some(availability_zone) = availability_zone {
         connstr = connstr.extend_options([format!("availability_zone={}", availability_zone)]);
     }
 

libs/wal_decoder/src/decoder.rs

@@ -19,7 +19,7 @@ impl InterpretedWalRecord {
     pub fn from_bytes_filtered(
         buf: Bytes,
         shard: &ShardIdentity,
-        next_record_lsn: Lsn,
+        record_end_lsn: Lsn,
         pg_version: u32,
     ) -> anyhow::Result<InterpretedWalRecord> {
         let mut decoded = DecodedWALRecord::default();
@@ -32,18 +32,18 @@ impl InterpretedWalRecord {
             FlushUncommittedRecords::No
         };
 
-        let metadata_record = MetadataRecord::from_decoded(&decoded, next_record_lsn, pg_version)?;
+        let metadata_record = MetadataRecord::from_decoded(&decoded, record_end_lsn, pg_version)?;
         let batch = SerializedValueBatch::from_decoded_filtered(
             decoded,
             shard,
-            next_record_lsn,
+            record_end_lsn,
             pg_version,
         )?;
 
         Ok(InterpretedWalRecord {
             metadata_record,
             batch,
-            next_record_lsn,
+            end_lsn: record_end_lsn,
             flush_uncommitted,
             xid,
         })
@@ -53,7 +53,7 @@ impl InterpretedWalRecord {
 impl MetadataRecord {
     fn from_decoded(
         decoded: &DecodedWALRecord,
-        next_record_lsn: Lsn,
+        record_end_lsn: Lsn,
         pg_version: u32,
     ) -> anyhow::Result<Option<MetadataRecord>> {
         // Note: this doesn't actually copy the bytes since
@@ -74,9 +74,7 @@ impl MetadataRecord {
                 Ok(None)
             }
             pg_constants::RM_CLOG_ID => Self::decode_clog_record(&mut buf, decoded, pg_version),
-            pg_constants::RM_XACT_ID => {
-                Self::decode_xact_record(&mut buf, decoded, next_record_lsn)
-            }
+            pg_constants::RM_XACT_ID => Self::decode_xact_record(&mut buf, decoded, record_end_lsn),
             pg_constants::RM_MULTIXACT_ID => {
                 Self::decode_multixact_record(&mut buf, decoded, pg_version)
             }
@@ -88,9 +86,7 @@ impl MetadataRecord {
             //
             // Alternatively, one can make the checkpoint part of the subscription protocol
             // to the pageserver. This should work fine, but can be done at a later point.
-            pg_constants::RM_XLOG_ID => {
-                Self::decode_xlog_record(&mut buf, decoded, next_record_lsn)
-            }
+            pg_constants::RM_XLOG_ID => Self::decode_xlog_record(&mut buf, decoded, record_end_lsn),
             pg_constants::RM_LOGICALMSG_ID => {
                 Self::decode_logical_message_record(&mut buf, decoded)
             }

libs/wal_decoder/src/models.rs

@@ -32,19 +32,16 @@ use postgres_ffi::walrecord::{
     XlSmgrTruncate, XlXactParsedRecord,
 };
 use postgres_ffi::{Oid, TransactionId};
-use serde::{Deserialize, Serialize};
 use utils::lsn::Lsn;
 
 use crate::serialized_batch::SerializedValueBatch;
 
-#[derive(Serialize, Deserialize)]
 pub enum FlushUncommittedRecords {
     Yes,
     No,
 }
 
 /// An interpreted Postgres WAL record, ready to be handled by the pageserver
-#[derive(Serialize, Deserialize)]
 pub struct InterpretedWalRecord {
     /// Optional metadata record - may cause writes to metadata keys
     /// in the storage engine
@@ -52,10 +49,8 @@ pub struct InterpretedWalRecord {
     /// A pre-serialized batch along with the required metadata for ingestion
     /// by the pageserver
     pub batch: SerializedValueBatch,
-    /// Byte offset within WAL for the start of the next PG WAL record.
-    /// Usually this is the end LSN of the current record, but in case of
-    /// XLOG SWITCH records it will be within the next segment.
-    pub next_record_lsn: Lsn,
+    /// Byte offset within WAL for the end of the original PG WAL record
+    pub end_lsn: Lsn,
     /// Whether to flush all uncommitted modifications to the storage engine
     /// before ingesting this record. This is currently only used for legacy PG
     /// database creations which read pages from a template database. Such WAL
@@ -65,21 +60,8 @@ pub struct InterpretedWalRecord {
     pub xid: TransactionId,
 }
 
-impl InterpretedWalRecord {
-    /// Checks if the WAL record is empty
-    ///
-    /// An empty interpreted WAL record has no data or metadata and does not have to be sent to the
-    /// pageserver.
-    pub fn is_empty(&self) -> bool {
-        self.batch.is_empty()
-            && self.metadata_record.is_none()
-            && matches!(self.flush_uncommitted, FlushUncommittedRecords::No)
-    }
-}
-
 /// The interpreted part of the Postgres WAL record which requires metadata
 /// writes to the underlying storage engine.
-#[derive(Serialize, Deserialize)]
 pub enum MetadataRecord {
     Heapam(HeapamRecord),
     Neonrmgr(NeonrmgrRecord),
@@ -95,12 +77,10 @@ pub enum MetadataRecord {
     Replorigin(ReploriginRecord),
 }
 
-#[derive(Serialize, Deserialize)]
 pub enum HeapamRecord {
     ClearVmBits(ClearVmBits),
 }
 
-#[derive(Serialize, Deserialize)]
 pub struct ClearVmBits {
     pub new_heap_blkno: Option<u32>,
     pub old_heap_blkno: Option<u32>,
@@ -108,29 +88,24 @@ pub struct ClearVmBits {
     pub flags: u8,
 }
 
-#[derive(Serialize, Deserialize)]
 pub enum NeonrmgrRecord {
     ClearVmBits(ClearVmBits),
 }
 
-#[derive(Serialize, Deserialize)]
 pub enum SmgrRecord {
     Create(SmgrCreate),
     Truncate(XlSmgrTruncate),
 }
 
-#[derive(Serialize, Deserialize)]
 pub struct SmgrCreate {
     pub rel: RelTag,
 }
 
-#[derive(Serialize, Deserialize)]
 pub enum DbaseRecord {
     Create(DbaseCreate),
     Drop(DbaseDrop),
 }
 
-#[derive(Serialize, Deserialize)]
 pub struct DbaseCreate {
     pub db_id: Oid,
     pub tablespace_id: Oid,
@@ -138,32 +113,27 @@ pub struct DbaseCreate {
     pub src_tablespace_id: Oid,
 }
 
-#[derive(Serialize, Deserialize)]
 pub struct DbaseDrop {
     pub db_id: Oid,
     pub tablespace_ids: Vec<Oid>,
 }
 
-#[derive(Serialize, Deserialize)]
 pub enum ClogRecord {
     ZeroPage(ClogZeroPage),
     Truncate(ClogTruncate),
 }
 
-#[derive(Serialize, Deserialize)]
 pub struct ClogZeroPage {
     pub segno: u32,
     pub rpageno: u32,
 }
 
-#[derive(Serialize, Deserialize)]
 pub struct ClogTruncate {
     pub pageno: u32,
     pub oldest_xid: TransactionId,
     pub oldest_xid_db: Oid,
 }
 
-#[derive(Serialize, Deserialize)]
 pub enum XactRecord {
     Commit(XactCommon),
     Abort(XactCommon),
@@ -172,7 +142,6 @@ pub enum XactRecord {
     Prepare(XactPrepare),
 }
 
-#[derive(Serialize, Deserialize)]
 pub struct XactCommon {
     pub parsed: XlXactParsedRecord,
     pub origin_id: u16,
@@ -181,73 +150,61 @@ pub struct XactCommon {
     pub lsn: Lsn,
 }
 
-#[derive(Serialize, Deserialize)]
 pub struct XactPrepare {
     pub xl_xid: TransactionId,
     pub data: Bytes,
 }
 
-#[derive(Serialize, Deserialize)]
 pub enum MultiXactRecord {
     ZeroPage(MultiXactZeroPage),
     Create(XlMultiXactCreate),
     Truncate(XlMultiXactTruncate),
 }
 
-#[derive(Serialize, Deserialize)]
 pub struct MultiXactZeroPage {
     pub slru_kind: SlruKind,
     pub segno: u32,
     pub rpageno: u32,
 }
 
-#[derive(Serialize, Deserialize)]
 pub enum RelmapRecord {
     Update(RelmapUpdate),
 }
 
-#[derive(Serialize, Deserialize)]
 pub struct RelmapUpdate {
     pub update: XlRelmapUpdate,
     pub buf: Bytes,
 }
 
-#[derive(Serialize, Deserialize)]
 pub enum XlogRecord {
     Raw(RawXlogRecord),
 }
 
-#[derive(Serialize, Deserialize)]
 pub struct RawXlogRecord {
     pub info: u8,
     pub lsn: Lsn,
     pub buf: Bytes,
 }
 
-#[derive(Serialize, Deserialize)]
 pub enum LogicalMessageRecord {
     Put(PutLogicalMessage),
     #[cfg(feature = "testing")]
     Failpoint,
 }
 
-#[derive(Serialize, Deserialize)]
 pub struct PutLogicalMessage {
     pub path: String,
     pub buf: Bytes,
 }
 
-#[derive(Serialize, Deserialize)]
 pub enum StandbyRecord {
     RunningXacts(StandbyRunningXacts),
 }
 
-#[derive(Serialize, Deserialize)]
 pub struct StandbyRunningXacts {
     pub oldest_running_xid: TransactionId,
 }
 
-#[derive(Serialize, Deserialize)]
 pub enum ReploriginRecord {
     Set(XlReploriginSet),
     Drop(XlReploriginDrop),

libs/wal_decoder/src/serialized_batch.rs

@@ -16,7 +16,6 @@ use pageserver_api::shard::ShardIdentity;
 use pageserver_api::{key::CompactKey, value::Value};
 use postgres_ffi::walrecord::{DecodedBkpBlock, DecodedWALRecord};
 use postgres_ffi::{page_is_new, page_set_lsn, pg_constants, BLCKSZ};
-use serde::{Deserialize, Serialize};
 use utils::bin_ser::BeSer;
 use utils::lsn::Lsn;
 
@@ -30,7 +29,6 @@ static ZERO_PAGE: Bytes = Bytes::from_static(&[0u8; BLCKSZ as usize]);
 /// relation sizes. In the case of "observed" values, we only need to know
 /// the key and LSN, so two types of metadata are supported to save on network
 /// bandwidth.
-#[derive(Serialize, Deserialize)]
 pub enum ValueMeta {
     Serialized(SerializedValueMeta),
     Observed(ObservedValueMeta),
@@ -77,7 +75,6 @@ impl PartialEq for OrderedValueMeta {
 impl Eq for OrderedValueMeta {}
 
 /// Metadata for a [`Value`] serialized into the batch.
-#[derive(Serialize, Deserialize)]
 pub struct SerializedValueMeta {
     pub key: CompactKey,
     pub lsn: Lsn,
@@ -89,14 +86,12 @@ pub struct SerializedValueMeta {
 }
 
 /// Metadata for a [`Value`] observed by the batch
-#[derive(Serialize, Deserialize)]
 pub struct ObservedValueMeta {
     pub key: CompactKey,
     pub lsn: Lsn,
 }
 
 /// Batch of serialized [`Value`]s.
-#[derive(Serialize, Deserialize)]
 pub struct SerializedValueBatch {
     /// [`Value`]s serialized in EphemeralFile's native format,
     /// ready for disk write by the pageserver
@@ -137,7 +132,7 @@ impl SerializedValueBatch {
     pub(crate) fn from_decoded_filtered(
         decoded: DecodedWALRecord,
         shard: &ShardIdentity,
-        next_record_lsn: Lsn,
+        record_end_lsn: Lsn,
         pg_version: u32,
     ) -> anyhow::Result<SerializedValueBatch> {
         // First determine how big the buffer needs to be and allocate it up-front.
@@ -161,17 +156,13 @@ impl SerializedValueBatch {
             let key = rel_block_to_key(rel, blk.blkno);
 
             if !key.is_valid_key_on_write_path() {
-                anyhow::bail!(
-                    "Unsupported key decoded at LSN {}: {}",
-                    next_record_lsn,
-                    key
-                );
+                anyhow::bail!("Unsupported key decoded at LSN {}: {}", record_end_lsn, key);
             }
 
             let key_is_local = shard.is_key_local(&key);
 
             tracing::debug!(
-                lsn=%next_record_lsn,
+                lsn=%record_end_lsn,
                 key=%key,
                 "ingest: shard decision {}",
                 if !key_is_local { "drop" } else { "keep" },
@@ -183,7 +174,7 @@ impl SerializedValueBatch {
                     // its blkno in case it implicitly extends a relation.
                     metadata.push(ValueMeta::Observed(ObservedValueMeta {
                         key: key.to_compact(),
-                        lsn: next_record_lsn,
+                        lsn: record_end_lsn,
                     }))
                 }
 
@@ -214,7 +205,7 @@ impl SerializedValueBatch {
                 // that would corrupt the page.
                 //
                 if !page_is_new(&image) {
-                    page_set_lsn(&mut image, next_record_lsn)
+                    page_set_lsn(&mut image, record_end_lsn)
                 }
                 assert_eq!(image.len(), BLCKSZ as usize);
 
@@ -233,12 +224,12 @@ impl SerializedValueBatch {
 
             metadata.push(ValueMeta::Serialized(SerializedValueMeta {
                 key: key.to_compact(),
-                lsn: next_record_lsn,
+                lsn: record_end_lsn,
                 batch_offset: relative_off,
                 len: val_ser_size,
                 will_init: val.will_init(),
             }));
-            max_lsn = std::cmp::max(max_lsn, next_record_lsn);
+            max_lsn = std::cmp::max(max_lsn, record_end_lsn);
             len += 1;
         }
 
@@ -496,16 +487,11 @@ impl SerializedValueBatch {
         }
     }
 
-    /// Checks if the batch contains any serialized or observed values
-    pub fn is_empty(&self) -> bool {
-        !self.has_data() && self.metadata.is_empty()
-    }
-
-    /// Checks if the batch contains data
+    /// Checks if the batch is empty
     ///
-    /// Note that if this returns false, it may still contain observed values or
-    /// a metadata record.
-    pub fn has_data(&self) -> bool {
+    /// A batch is empty when it contains no serialized values.
+    /// Note that it may still contain observed values.
+    pub fn is_empty(&self) -> bool {
         let empty = self.raw.is_empty();
 
         if cfg!(debug_assertions) && empty {
@@ -515,7 +501,7 @@ impl SerializedValueBatch {
                 .all(|meta| matches!(meta, ValueMeta::Observed(_))));
         }
 
-        !empty
+        empty
     }
 
     /// Returns the number of values serialized in the batch

pageserver/benches/bench_ingest.rs

@@ -167,7 +167,6 @@ fn criterion_benchmark(c: &mut Criterion) {
         16384,
         virtual_file::io_engine_for_bench(),
         conf.virtual_file_io_mode,
-        virtual_file::SyncMode::Sync,
     );
     page_cache::init(conf.page_cache_size);
 

pageserver/ctl/src/layer_map_analyzer.rs

@@ -138,7 +138,6 @@ pub(crate) async fn main(cmd: &AnalyzeLayerMapCmd) -> Result<()> {
         10,
         virtual_file::api::IoEngineKind::StdFs,
         IoMode::preferred(),
-        virtual_file::SyncMode::Sync,
     );
     pageserver::page_cache::init(100);
 

pageserver/ctl/src/layers.rs

@@ -51,7 +51,6 @@ async fn read_delta_file(path: impl AsRef<Path>, ctx: &RequestContext) -> Result
         10,
         virtual_file::api::IoEngineKind::StdFs,
         IoMode::preferred(),
-        virtual_file::SyncMode::Sync,
     );
     page_cache::init(100);
     let path = Utf8Path::from_path(path.as_ref()).expect("non-Unicode path");
@@ -66,7 +65,6 @@ async fn read_image_file(path: impl AsRef<Path>, ctx: &RequestContext) -> Result
         10,
         virtual_file::api::IoEngineKind::StdFs,
         IoMode::preferred(),
-        virtual_file::SyncMode::Sync,
     );
     page_cache::init(100);
     let path = Utf8Path::from_path(path.as_ref()).expect("non-Unicode path");
@@ -173,7 +171,6 @@ pub(crate) async fn main(cmd: &LayerCmd) -> Result<()> {
                 10,
                 virtual_file::api::IoEngineKind::StdFs,
                 IoMode::preferred(),
-                virtual_file::SyncMode::Sync,
             );
             pageserver::page_cache::init(100);
 

pageserver/ctl/src/main.rs

@@ -209,7 +209,6 @@ async fn print_layerfile(path: &Utf8Path) -> anyhow::Result<()> {
         10,
         virtual_file::api::IoEngineKind::StdFs,
         IoMode::preferred(),
-        virtual_file::SyncMode::Sync,
     );
     page_cache::init(100);
     let ctx = RequestContext::new(TaskKind::DebugTool, DownloadBehavior::Error);

pageserver/src/bin/pageserver.rs

@@ -126,7 +126,6 @@ fn main() -> anyhow::Result<()> {
     // after setting up logging, log the effective IO engine choice and read path implementations
     info!(?conf.virtual_file_io_engine, "starting with virtual_file IO engine");
     info!(?conf.virtual_file_io_mode, "starting with virtual_file IO mode");
-    info!(?conf.wal_receiver_protocol, "starting with WAL receiver protocol");
 
     // The tenants directory contains all the pageserver local disk state.
     // Create if not exists and make sure all the contents are durable before proceeding.
@@ -172,18 +171,11 @@ fn main() -> anyhow::Result<()> {
     let scenario = failpoint_support::init();
 
     // Basic initialization of things that don't change after startup
-    tracing::info!("Initializing virtual_file...");
     virtual_file::init(
         conf.max_file_descriptors,
         conf.virtual_file_io_engine,
         conf.virtual_file_io_mode,
-        if conf.no_sync {
-            virtual_file::SyncMode::UnsafeNoSync
-        } else {
-            virtual_file::SyncMode::Sync
-        },
     );
-    tracing::info!("Initializing page_cache...");
     page_cache::init(conf.page_cache_size);
 
     start_pageserver(launch_ts, conf).context("Failed to start pageserver")?;

pageserver/src/config.rs

@@ -14,7 +14,6 @@ use remote_storage::{RemotePath, RemoteStorageConfig};
 use std::env;
 use storage_broker::Uri;
 use utils::logging::SecretString;
-use utils::postgres_client::PostgresClientProtocol;
 
 use once_cell::sync::OnceCell;
 use reqwest::Url;
@@ -183,8 +182,6 @@ pub struct PageServerConf {
 
     /// Optionally disable disk syncs (unsafe!)
     pub no_sync: bool,
-
-    pub wal_receiver_protocol: PostgresClientProtocol,
 }
 
 /// Token for authentication to safekeepers
@@ -341,7 +338,6 @@ impl PageServerConf {
             virtual_file_io_engine,
             tenant_config,
             no_sync,
-            wal_receiver_protocol,
         } = config_toml;
 
         let mut conf = PageServerConf {
@@ -381,7 +377,6 @@ impl PageServerConf {
             image_compression,
             timeline_offloading,
             ephemeral_bytes_per_memory_kb,
-            wal_receiver_protocol,
 
             // ------------------------------------------------------------
             // fields that require additional validation or custom handling

pageserver/src/http/routes.rs

@@ -324,7 +324,6 @@ impl From<crate::tenant::DeleteTimelineError> for ApiError {
                     .into_boxed_str(),
             ),
             a @ AlreadyInProgress(_) => ApiError::Conflict(a.to_string()),
-            Cancelled => ApiError::ResourceUnavailable("shutting down".into()),
             Other(e) => ApiError::InternalServerError(e),
         }
     }

pageserver/src/pgdatadir_mapping.rs

@@ -1055,9 +1055,10 @@ impl<'a> DatadirModification<'a> {
     }
 
     pub(crate) fn has_dirty_data(&self) -> bool {
-        self.pending_data_batch
+        !self
+            .pending_data_batch
             .as_ref()
-            .map_or(false, |b| b.has_data())
+            .map_or(true, |b| b.is_empty())
     }
 
     /// Set the current lsn
@@ -1233,7 +1234,7 @@ impl<'a> DatadirModification<'a> {
             Some(pending_batch) => {
                 pending_batch.extend(batch);
             }
-            None if batch.has_data() => {
+            None if !batch.is_empty() => {
                 self.pending_data_batch = Some(batch);
             }
             None => {

pageserver/src/tenant.rs

@@ -39,7 +39,6 @@ use remote_timeline_client::UploadQueueNotReadyError;
 use std::collections::BTreeMap;
 use std::fmt;
 use std::future::Future;
-use std::sync::atomic::AtomicBool;
 use std::sync::Weak;
 use std::time::SystemTime;
 use storage_broker::BrokerClientChannel;
@@ -525,9 +524,6 @@ pub struct OffloadedTimeline {
     /// Prevent two tasks from deleting the timeline at the same time. If held, the
     /// timeline is being deleted. If 'true', the timeline has already been deleted.
     pub delete_progress: TimelineDeleteProgress,
-
-    /// Part of the `OffloadedTimeline` object's lifecycle: this needs to be set before we drop it
-    pub deleted_from_ancestor: AtomicBool,
 }
 
 impl OffloadedTimeline {
@@ -537,16 +533,9 @@ impl OffloadedTimeline {
     /// the timeline is not in a stopped state.
     /// Panics if the timeline is not archived.
     fn from_timeline(timeline: &Timeline) -> Result<Self, UploadQueueNotReadyError> {
-        let (ancestor_retain_lsn, ancestor_timeline_id) =
-            if let Some(ancestor_timeline) = timeline.ancestor_timeline() {
-                let ancestor_lsn = timeline.get_ancestor_lsn();
-                let ancestor_timeline_id = ancestor_timeline.timeline_id;
-                let mut gc_info = ancestor_timeline.gc_info.write().unwrap();
-                gc_info.insert_child(timeline.timeline_id, ancestor_lsn, MaybeOffloaded::Yes);
-                (Some(ancestor_lsn), Some(ancestor_timeline_id))
-            } else {
-                (None, None)
-            };
+        let ancestor_retain_lsn = timeline
+            .get_ancestor_timeline_id()
+            .map(|_timeline_id| timeline.get_ancestor_lsn());
         let archived_at = timeline
             .remote_client
             .archived_at_stopped_queue()?
@@ -554,17 +543,14 @@ impl OffloadedTimeline {
         Ok(Self {
             tenant_shard_id: timeline.tenant_shard_id,
             timeline_id: timeline.timeline_id,
-            ancestor_timeline_id,
+            ancestor_timeline_id: timeline.get_ancestor_timeline_id(),
             ancestor_retain_lsn,
             archived_at,
 
             delete_progress: timeline.delete_progress.clone(),
-            deleted_from_ancestor: AtomicBool::new(false),
         })
     }
     fn from_manifest(tenant_shard_id: TenantShardId, manifest: &OffloadedTimelineManifest) -> Self {
-        // We expect to reach this case in tenant loading, where the `retain_lsn` is populated in the parent's `gc_info`
-        // by the `initialize_gc_info` function.
         let OffloadedTimelineManifest {
             timeline_id,
             ancestor_timeline_id,
@@ -578,7 +564,6 @@ impl OffloadedTimeline {
             ancestor_retain_lsn,
             archived_at,
             delete_progress: TimelineDeleteProgress::default(),
-            deleted_from_ancestor: AtomicBool::new(false),
         }
     }
     fn manifest(&self) -> OffloadedTimelineManifest {
@@ -596,33 +581,6 @@ impl OffloadedTimeline {
             archived_at: *archived_at,
         }
     }
-    /// Delete this timeline's retain_lsn from its ancestor, if present in the given tenant
-    fn delete_from_ancestor_with_timelines(
-        &self,
-        timelines: &std::sync::MutexGuard<'_, HashMap<TimelineId, Arc<Timeline>>>,
-    ) {
-        if let (Some(_retain_lsn), Some(ancestor_timeline_id)) =
-            (self.ancestor_retain_lsn, self.ancestor_timeline_id)
-        {
-            if let Some((_, ancestor_timeline)) = timelines
-                .iter()
-                .find(|(tid, _tl)| **tid == ancestor_timeline_id)
-            {
-                ancestor_timeline
-                    .gc_info
-                    .write()
-                    .unwrap()
-                    .remove_child_offloaded(self.timeline_id);
-            }
-        }
-        self.deleted_from_ancestor.store(true, Ordering::Release);
-    }
-    /// Call [`Self::delete_from_ancestor_with_timelines`] instead if possible.
-    ///
-    /// As the entire tenant is being dropped, don't bother deregistering the `retain_lsn` from the ancestor.
-    fn defuse_for_tenant_drop(&self) {
-        self.deleted_from_ancestor.store(true, Ordering::Release);
-    }
 }
 
 impl fmt::Debug for OffloadedTimeline {
@@ -631,17 +589,6 @@ impl fmt::Debug for OffloadedTimeline {
     }
 }
 
-impl Drop for OffloadedTimeline {
-    fn drop(&mut self) {
-        if !self.deleted_from_ancestor.load(Ordering::Acquire) {
-            tracing::warn!(
-                "offloaded timeline {} was dropped without having cleaned it up at the ancestor",
-                self.timeline_id
-            );
-        }
-    }
-}
-
 #[derive(Copy, Clone, PartialEq, Eq, Hash, Debug)]
 pub enum MaybeOffloaded {
     Yes,
@@ -753,9 +700,6 @@ pub enum DeleteTimelineError {
     #[error("Timeline deletion is already in progress")]
     AlreadyInProgress(Arc<tokio::sync::Mutex<DeleteTimelineFlow>>),
 
-    #[error("Cancelled")]
-    Cancelled,
-
     #[error(transparent)]
     Other(#[from] anyhow::Error),
 }
@@ -766,7 +710,6 @@ impl Debug for DeleteTimelineError {
             Self::NotFound => write!(f, "NotFound"),
             Self::HasChildren(c) => f.debug_tuple("HasChildren").field(c).finish(),
             Self::AlreadyInProgress(_) => f.debug_tuple("AlreadyInProgress").finish(),
-            Self::Cancelled => f.debug_tuple("Cancelled").finish(),
             Self::Other(e) => f.debug_tuple("Other").field(e).finish(),
         }
     }
@@ -1490,12 +1433,6 @@ impl Tenant {
                     info!(%timeline_id, "index_part not found on remote");
                     continue;
                 }
-                Err(DownloadError::Fatal(why)) => {
-                    // If, while loading one remote timeline, we saw an indication that our generation
-                    // number is likely invalid, then we should not load the whole tenant.
-                    error!(%timeline_id, "Fatal error loading timeline: {why}");
-                    anyhow::bail!(why.to_string());
-                }
                 Err(e) => {
                     // Some (possibly ephemeral) error happened during index_part download.
                     // Pretend the timeline exists to not delete the timeline directory,
@@ -1584,7 +1521,7 @@ impl Tenant {
         }
         // Complete deletions for offloaded timeline id's.
         offloaded_timelines_list
-            .retain(|(offloaded_id, offloaded)| {
+            .retain(|(offloaded_id, _offloaded)| {
                 // At this point, offloaded_timeline_ids has the list of all offloaded timelines
                 // without a prefix in S3, so they are inexistent.
                 // In the end, existence of a timeline is finally determined by the existence of an index-part.json in remote storage.
@@ -1592,7 +1529,6 @@ impl Tenant {
                 let delete = offloaded_timeline_ids.contains(offloaded_id);
                 if delete {
                     tracing::info!("Removing offloaded timeline {offloaded_id} from manifest as no remote prefix was found");
-                    offloaded.defuse_for_tenant_drop();
                 }
                 !delete
         });
@@ -1981,15 +1917,9 @@ impl Tenant {
                 )));
             };
             let mut offloaded_timelines = self.timelines_offloaded.lock().unwrap();
-            match offloaded_timelines.remove(&timeline_id) {
-                Some(offloaded) => {
-                    offloaded.delete_from_ancestor_with_timelines(&timelines);
-                }
-                None => warn!("timeline already removed from offloaded timelines"),
+            if offloaded_timelines.remove(&timeline_id).is_none() {
+                warn!("timeline already removed from offloaded timelines");
             }
-
-            self.initialize_gc_info(&timelines, &offloaded_timelines, Some(timeline_id));
-
             Arc::clone(timeline)
         };
 
@@ -2727,7 +2657,7 @@ impl Tenant {
                 .filter(|timeline| !(timeline.is_broken() || timeline.is_stopping()));
 
             // Before activation, populate each Timeline's GcInfo with information about its children
-            self.initialize_gc_info(&timelines_accessor, &timelines_offloaded_accessor, None);
+            self.initialize_gc_info(&timelines_accessor, &timelines_offloaded_accessor);
 
             // Spawn gc and compaction loops. The loops will shut themselves
             // down when they notice that the tenant is inactive.
@@ -2842,14 +2772,8 @@ impl Tenant {
                 let timeline_id = timeline.timeline_id;
                 let span = tracing::info_span!("timeline_shutdown", %timeline_id, ?shutdown_mode);
                 js.spawn(async move { timeline.shutdown(shutdown_mode).instrument(span).await });
-            });
-        }
-        {
-            let timelines_offloaded = self.timelines_offloaded.lock().unwrap();
-            timelines_offloaded.values().for_each(|timeline| {
-                timeline.defuse_for_tenant_drop();
-            });
-        }
+            })
+        };
         // test_long_timeline_create_then_tenant_delete is leaning on this message
         tracing::info!("Waiting for timelines...");
         while let Some(res) = js.join_next().await {
@@ -3833,13 +3757,10 @@ impl Tenant {
         &self,
         timelines: &std::sync::MutexGuard<HashMap<TimelineId, Arc<Timeline>>>,
         timelines_offloaded: &std::sync::MutexGuard<HashMap<TimelineId, Arc<OffloadedTimeline>>>,
-        restrict_to_timeline: Option<TimelineId>,
     ) {
-        if restrict_to_timeline.is_none() {
-            // This function must be called before activation: after activation timeline create/delete operations
-            // might happen, and this function is not safe to run concurrently with those.
-            assert!(!self.is_active());
-        }
+        // This function must be called before activation: after activation timeline create/delete operations
+        // might happen, and this function is not safe to run concurrently with those.
+        assert!(!self.is_active());
 
         // Scan all timelines. For each timeline, remember the timeline ID and
         // the branch point where it was created.
@@ -3872,12 +3793,7 @@ impl Tenant {
         let horizon = self.get_gc_horizon();
 
         // Populate each timeline's GcInfo with information about its child branches
-        let timelines_to_write = if let Some(timeline_id) = restrict_to_timeline {
-            itertools::Either::Left(timelines.get(&timeline_id).into_iter())
-        } else {
-            itertools::Either::Right(timelines.values())
-        };
-        for timeline in timelines_to_write {
+        for timeline in timelines.values() {
             let mut branchpoints: Vec<(Lsn, TimelineId, MaybeOffloaded)> = all_branchpoints
                 .remove(&timeline.timeline_id)
                 .unwrap_or_default();
@@ -7841,13 +7757,13 @@ mod tests {
             (
                 get_key(3),
                 Lsn(0x20),
-                Value::WalRecord(NeonWalRecord::wal_clear("c")),
+                Value::WalRecord(NeonWalRecord::wal_clear()),
             ),
             (get_key(4), Lsn(0x10), Value::Image("0x10".into())),
             (
                 get_key(4),
                 Lsn(0x20),
-                Value::WalRecord(NeonWalRecord::wal_init("i")),
+                Value::WalRecord(NeonWalRecord::wal_init()),
             ),
         ];
         let image1 = vec![(get_key(1), "0x10".into())];
@@ -7996,30 +7912,8 @@ mod tests {
 
     #[cfg(feature = "testing")]
     #[tokio::test]
-    async fn test_simple_bottom_most_compaction_deltas_1() -> anyhow::Result<()> {
-        test_simple_bottom_most_compaction_deltas_helper(
-            "test_simple_bottom_most_compaction_deltas_1",
-            false,
-        )
-        .await
-    }
-
-    #[cfg(feature = "testing")]
-    #[tokio::test]
-    async fn test_simple_bottom_most_compaction_deltas_2() -> anyhow::Result<()> {
-        test_simple_bottom_most_compaction_deltas_helper(
-            "test_simple_bottom_most_compaction_deltas_2",
-            true,
-        )
-        .await
-    }
-
-    #[cfg(feature = "testing")]
-    async fn test_simple_bottom_most_compaction_deltas_helper(
-        test_name: &'static str,
-        use_delta_bottom_layer: bool,
-    ) -> anyhow::Result<()> {
-        let harness = TenantHarness::create(test_name).await?;
+    async fn test_simple_bottom_most_compaction_deltas() -> anyhow::Result<()> {
+        let harness = TenantHarness::create("test_simple_bottom_most_compaction_deltas").await?;
         let (tenant, ctx) = harness.load().await;
 
         fn get_key(id: u32) -> Key {
@@ -8050,16 +7944,6 @@ mod tests {
         let img_layer = (0..10)
             .map(|id| (get_key(id), Bytes::from(format!("value {id}@0x10"))))
             .collect_vec();
-        // or, delta layer at 0x10 if `use_delta_bottom_layer` is true
-        let delta4 = (0..10)
-            .map(|id| {
-                (
-                    get_key(id),
-                    Lsn(0x08),
-                    Value::WalRecord(NeonWalRecord::wal_init(format!("value {id}@0x10"))),
-                )
-            })
-            .collect_vec();
 
         let delta1 = vec![
             (
@@ -8113,61 +7997,21 @@ mod tests {
             ),
         ];
 
-        let tline = if use_delta_bottom_layer {
-            tenant
-                .create_test_timeline_with_layers(
-                    TIMELINE_ID,
-                    Lsn(0x08),
-                    DEFAULT_PG_VERSION,
-                    &ctx,
-                    vec![
-                        DeltaLayerTestDesc::new_with_inferred_key_range(
-                            Lsn(0x08)..Lsn(0x10),
-                            delta4,
-                        ),
-                        DeltaLayerTestDesc::new_with_inferred_key_range(
-                            Lsn(0x20)..Lsn(0x48),
-                            delta1,
-                        ),
-                        DeltaLayerTestDesc::new_with_inferred_key_range(
-                            Lsn(0x20)..Lsn(0x48),
-                            delta2,
-                        ),
-                        DeltaLayerTestDesc::new_with_inferred_key_range(
-                            Lsn(0x48)..Lsn(0x50),
-                            delta3,
-                        ),
-                    ], // delta layers
-                    vec![], // image layers
-                    Lsn(0x50),
-                )
-                .await?
-        } else {
-            tenant
-                .create_test_timeline_with_layers(
-                    TIMELINE_ID,
-                    Lsn(0x10),
-                    DEFAULT_PG_VERSION,
-                    &ctx,
-                    vec![
-                        DeltaLayerTestDesc::new_with_inferred_key_range(
-                            Lsn(0x10)..Lsn(0x48),
-                            delta1,
-                        ),
-                        DeltaLayerTestDesc::new_with_inferred_key_range(
-                            Lsn(0x10)..Lsn(0x48),
-                            delta2,
-                        ),
-                        DeltaLayerTestDesc::new_with_inferred_key_range(
-                            Lsn(0x48)..Lsn(0x50),
-                            delta3,
-                        ),
-                    ], // delta layers
-                    vec![(Lsn(0x10), img_layer)], // image layers
-                    Lsn(0x50),
-                )
-                .await?
-        };
+        let tline = tenant
+            .create_test_timeline_with_layers(
+                TIMELINE_ID,
+                Lsn(0x10),
+                DEFAULT_PG_VERSION,
+                &ctx,
+                vec![
+                    DeltaLayerTestDesc::new_with_inferred_key_range(Lsn(0x10)..Lsn(0x48), delta1),
+                    DeltaLayerTestDesc::new_with_inferred_key_range(Lsn(0x10)..Lsn(0x48), delta2),
+                    DeltaLayerTestDesc::new_with_inferred_key_range(Lsn(0x48)..Lsn(0x50), delta3),
+                ], // delta layers
+                vec![(Lsn(0x10), img_layer)], // image layers
+                Lsn(0x50),
+            )
+            .await?;
         {
             // Update GC info
             let mut guard = tline.gc_info.write().unwrap();
@@ -8277,7 +8121,7 @@ mod tests {
             (
                 key,
                 Lsn(0x10),
-                Value::WalRecord(NeonWalRecord::wal_init("0x10")),
+                Value::Image(Bytes::copy_from_slice(b"0x10")),
             ),
             (
                 key,
@@ -8339,7 +8183,7 @@ mod tests {
                     Lsn(0x20),
                     KeyLogAtLsn(vec![(
                         Lsn(0x20),
-                        Value::Image(Bytes::from_static(b"0x10;0x20")),
+                        Value::Image(Bytes::copy_from_slice(b"0x10;0x20")),
                     )]),
                 ),
                 (
@@ -9321,7 +9165,7 @@ mod tests {
 
             let will_init = will_init_keys.contains(&i);
             if will_init {
-                delta_layer_spec.push((key, lsn, Value::WalRecord(NeonWalRecord::wal_init(""))));
+                delta_layer_spec.push((key, lsn, Value::WalRecord(NeonWalRecord::wal_init())));
 
                 expected_key_values.insert(key, "".to_string());
             } else {
@@ -9724,54 +9568,4 @@ mod tests {
 
         Ok(())
     }
-
-    #[cfg(feature = "testing")]
-    #[tokio::test]
-    async fn test_timeline_offload_retain_lsn() -> anyhow::Result<()> {
-        let harness = TenantHarness::create("test_timeline_offload_retain_lsn")
-            .await
-            .unwrap();
-        let (tenant, ctx) = harness.load().await;
-        let tline_parent = tenant
-            .create_test_timeline(TIMELINE_ID, Lsn(0x10), DEFAULT_PG_VERSION, &ctx)
-            .await
-            .unwrap();
-        let tline_child = tenant
-            .branch_timeline_test(&tline_parent, NEW_TIMELINE_ID, Some(Lsn(0x20)), &ctx)
-            .await
-            .unwrap();
-        {
-            let gc_info_parent = tline_parent.gc_info.read().unwrap();
-            assert_eq!(
-                gc_info_parent.retain_lsns,
-                vec![(Lsn(0x20), tline_child.timeline_id, MaybeOffloaded::No)]
-            );
-        }
-        // We have to directly call the remote_client instead of using the archive function to avoid constructing broker client...
-        tline_child
-            .remote_client
-            .schedule_index_upload_for_timeline_archival_state(TimelineArchivalState::Archived)
-            .unwrap();
-        tline_child.remote_client.wait_completion().await.unwrap();
-        offload_timeline(&tenant, &tline_child)
-            .instrument(tracing::info_span!(parent: None, "offload_test", tenant_id=%"test", shard_id=%"test", timeline_id=%"test"))
-            .await.unwrap();
-        let child_timeline_id = tline_child.timeline_id;
-        Arc::try_unwrap(tline_child).unwrap();
-
-        {
-            let gc_info_parent = tline_parent.gc_info.read().unwrap();
-            assert_eq!(
-                gc_info_parent.retain_lsns,
-                vec![(Lsn(0x20), child_timeline_id, MaybeOffloaded::Yes)]
-            );
-        }
-
-        tenant
-            .get_offloaded_timeline(child_timeline_id)
-            .unwrap()
-            .defuse_for_tenant_drop();
-
-        Ok(())
-    }
 }

pageserver/src/tenant/remote_timeline_client.rs

@@ -243,7 +243,7 @@ use self::index::IndexPart;
 use super::metadata::MetadataUpdate;
 use super::storage_layer::{Layer, LayerName, ResidentLayer};
 use super::upload_queue::{NotInitialized, SetDeletedFlagProgress};
-use super::{DeleteTimelineError, Generation};
+use super::Generation;
 
 pub(crate) use download::{
     download_index_part, download_tenant_manifest, is_temp_download_file,
@@ -574,18 +574,12 @@ impl RemoteTimelineClient {
 
             if latest_index_generation > index_generation {
                 // Unexpected!  Why are we loading such an old index if a more recent one exists?
-                // We will refuse to proceed, as there is no reasonable scenario where this should happen, but
-                // there _is_ a clear bug/corruption scenario where it would happen (controller sets the generation
-                // backwards).
-                tracing::error!(
+                tracing::warn!(
                     ?index_generation,
                     ?latest_index_generation,
                     ?latest_index_mtime,
                     "Found a newer index while loading an old one"
                 );
-                return Err(DownloadError::Fatal(
-                    "Index age exceeds threshold and a newer index exists".into(),
-                ));
             }
         }
 
@@ -1550,17 +1544,15 @@ impl RemoteTimelineClient {
     /// Prerequisites: UploadQueue should be in stopped state and deleted_at should be successfuly set.
     /// The function deletes layer files one by one, then lists the prefix to see if we leaked something
     /// deletes leaked files if any and proceeds with deletion of index file at the end.
-    pub(crate) async fn delete_all(self: &Arc<Self>) -> Result<(), DeleteTimelineError> {
+    pub(crate) async fn delete_all(self: &Arc<Self>) -> anyhow::Result<()> {
         debug_assert_current_span_has_tenant_and_timeline_id();
 
         let layers: Vec<RemotePath> = {
             let mut locked = self.upload_queue.lock().unwrap();
-            let stopped = locked.stopped_mut().map_err(DeleteTimelineError::Other)?;
+            let stopped = locked.stopped_mut()?;
 
             if !matches!(stopped.deleted_at, SetDeletedFlagProgress::Successful(_)) {
-                return Err(DeleteTimelineError::Other(anyhow::anyhow!(
-                    "deleted_at is not set"
-                )));
+                anyhow::bail!("deleted_at is not set")
             }
 
             debug_assert!(stopped.upload_queue_for_deletion.no_pending_work());
@@ -1595,10 +1587,7 @@ impl RemoteTimelineClient {
         };
 
         let layer_deletion_count = layers.len();
-        self.deletion_queue_client
-            .push_immediate(layers)
-            .await
-            .map_err(|_| DeleteTimelineError::Cancelled)?;
+        self.deletion_queue_client.push_immediate(layers).await?;
 
         // Delete the initdb.tar.zst, which is not always present, but deletion attempts of
         // inexistant objects are not considered errors.
@@ -1606,8 +1595,7 @@ impl RemoteTimelineClient {
             remote_initdb_archive_path(&self.tenant_shard_id.tenant_id, &self.timeline_id);
         self.deletion_queue_client
             .push_immediate(vec![initdb_path])
-            .await
-            .map_err(|_| DeleteTimelineError::Cancelled)?;
+            .await?;
 
         // Do not delete index part yet, it is needed for possible retry. If we remove it first
         // and retry will arrive to different pageserver there wont be any traces of it on remote storage
@@ -1615,9 +1603,7 @@ impl RemoteTimelineClient {
 
         // Execute all pending deletions, so that when we proceed to do a listing below, we aren't
         // taking the burden of listing all the layers that we already know we should delete.
-        self.flush_deletion_queue()
-            .await
-            .map_err(|_| DeleteTimelineError::Cancelled)?;
+        self.flush_deletion_queue().await?;
 
         let cancel = shutdown_token();
 
@@ -1680,32 +1666,28 @@ impl RemoteTimelineClient {
         if !remaining_layers.is_empty() {
             self.deletion_queue_client
                 .push_immediate(remaining_layers)
-                .await
-                .map_err(|_| DeleteTimelineError::Cancelled)?;
+                .await?;
         }
 
         fail::fail_point!("timeline-delete-before-index-delete", |_| {
-            Err(DeleteTimelineError::Other(anyhow::anyhow!(
+            Err(anyhow::anyhow!(
                 "failpoint: timeline-delete-before-index-delete"
-            )))?
+            ))?
         });
 
         debug!("enqueuing index part deletion");
         self.deletion_queue_client
             .push_immediate([latest_index].to_vec())
-            .await
-            .map_err(|_| DeleteTimelineError::Cancelled)?;
+            .await?;
 
         // Timeline deletion is rare and we have probably emitted a reasonably number of objects: wait
         // for a flush to a persistent deletion list so that we may be sure deletion will occur.
-        self.flush_deletion_queue()
-            .await
-            .map_err(|_| DeleteTimelineError::Cancelled)?;
+        self.flush_deletion_queue().await?;
 
         fail::fail_point!("timeline-delete-after-index-delete", |_| {
-            Err(DeleteTimelineError::Other(anyhow::anyhow!(
+            Err(anyhow::anyhow!(
                 "failpoint: timeline-delete-after-index-delete"
-            )))?
+            ))?
         });
 
         info!(prefix=%timeline_storage_path, referenced=layer_deletion_count, not_referenced=%not_referenced_count, "done deleting in timeline prefix, including index_part.json");

pageserver/src/tenant/storage_layer/merge_iterator.rs

@@ -562,7 +562,7 @@ mod tests {
             (
                 get_key(0),
                 Lsn(0x10),
-                Value::WalRecord(NeonWalRecord::wal_init("")),
+                Value::WalRecord(NeonWalRecord::wal_init()),
             ),
             (
                 get_key(0),
@@ -572,7 +572,7 @@ mod tests {
             (
                 get_key(5),
                 Lsn(0x10),
-                Value::WalRecord(NeonWalRecord::wal_init("")),
+                Value::WalRecord(NeonWalRecord::wal_init()),
             ),
             (
                 get_key(5),

pageserver/src/tenant/timeline.rs

@@ -477,21 +477,8 @@ impl GcInfo {
         self.retain_lsns.sort_by_key(|i| i.0);
     }
 
-    pub(super) fn remove_child_maybe_offloaded(
-        &mut self,
-        child_id: TimelineId,
-        maybe_offloaded: MaybeOffloaded,
-    ) {
-        self.retain_lsns
-            .retain(|i| !(i.1 == child_id && i.2 == maybe_offloaded));
-    }
-
-    pub(super) fn remove_child_not_offloaded(&mut self, child_id: TimelineId) {
-        self.remove_child_maybe_offloaded(child_id, MaybeOffloaded::No);
-    }
-
-    pub(super) fn remove_child_offloaded(&mut self, child_id: TimelineId) {
-        self.remove_child_maybe_offloaded(child_id, MaybeOffloaded::Yes);
+    pub(super) fn remove_child(&mut self, child_id: TimelineId) {
+        self.retain_lsns.retain(|i| i.1 != child_id);
     }
 }
 
@@ -2416,7 +2403,6 @@ impl Timeline {
         *guard = Some(WalReceiver::start(
             Arc::clone(self),
             WalReceiverConf {
-                protocol: self.conf.wal_receiver_protocol,
                 wal_connect_timeout,
                 lagging_wal_timeout,
                 max_lsn_wal_lag,
@@ -4515,7 +4501,7 @@ impl Drop for Timeline {
             // This lock should never be poisoned, but in case it is we do a .map() instead of
             // an unwrap(), to avoid panicking in a destructor and thereby aborting the process.
             if let Ok(mut gc_info) = ancestor.gc_info.write() {
-                gc_info.remove_child_not_offloaded(self.timeline_id)
+                gc_info.remove_child(self.timeline_id)
             }
         }
     }
@@ -5044,7 +5030,7 @@ impl Timeline {
 
             // 1. Is it newer than GC horizon cutoff point?
             if l.get_lsn_range().end > space_cutoff {
-                info!(
+                debug!(
                     "keeping {} because it's newer than space_cutoff {}",
                     l.layer_name(),
                     space_cutoff,
@@ -5055,7 +5041,7 @@ impl Timeline {
 
             // 2. It is newer than PiTR cutoff point?
             if l.get_lsn_range().end > time_cutoff {
-                info!(
+                debug!(
                     "keeping {} because it's newer than time_cutoff {}",
                     l.layer_name(),
                     time_cutoff,
@@ -5074,7 +5060,7 @@ impl Timeline {
             for retain_lsn in &retain_lsns {
                 // start_lsn is inclusive
                 if &l.get_lsn_range().start <= retain_lsn {
-                    info!(
+                    debug!(
                         "keeping {} because it's still might be referenced by child branch forked at {} is_dropped: xx is_incremental: {}",
                         l.layer_name(),
                         retain_lsn,
@@ -5089,7 +5075,7 @@ impl Timeline {
             if let Some(lsn) = &max_lsn_with_valid_lease {
                 // keep if layer start <= any of the lease
                 if &l.get_lsn_range().start <= lsn {
-                    info!(
+                    debug!(
                         "keeping {} because there is a valid lease preventing GC at {}",
                         l.layer_name(),
                         lsn,
@@ -5121,13 +5107,13 @@ impl Timeline {
             if !layers
                 .image_layer_exists(&l.get_key_range(), &(l.get_lsn_range().end..new_gc_cutoff))
             {
-                info!("keeping {} because it is the latest layer", l.layer_name());
+                debug!("keeping {} because it is the latest layer", l.layer_name());
                 result.layers_not_updated += 1;
                 continue 'outer;
             }
 
             // We didn't find any reason to keep this file, so remove it.
-            info!(
+            debug!(
                 "garbage collecting {} is_dropped: xx is_incremental: {}",
                 l.layer_name(),
                 l.is_incremental(),
@@ -5789,7 +5775,7 @@ impl<'a> TimelineWriter<'a> {
         batch: SerializedValueBatch,
         ctx: &RequestContext,
     ) -> anyhow::Result<()> {
-        if !batch.has_data() {
+        if batch.is_empty() {
             return Ok(());
         }
 

pageserver/src/tenant/timeline/delete.rs

@@ -5,7 +5,6 @@ use std::{
 
 use anyhow::Context;
 use pageserver_api::{models::TimelineState, shard::TenantShardId};
-use remote_storage::DownloadError;
 use tokio::sync::OwnedMutexGuard;
 use tracing::{error, info, info_span, instrument, Instrument};
 use utils::{crashsafe, fs_ext, id::TimelineId, pausable_failpoint};
@@ -17,7 +16,7 @@ use crate::{
         metadata::TimelineMetadata,
         remote_timeline_client::{PersistIndexPartWithDeletedFlagError, RemoteTimelineClient},
         CreateTimelineCause, DeleteTimelineError, MaybeDeletedIndexPart, Tenant,
-        TenantManifestError, TimelineOrOffloaded,
+        TimelineOrOffloaded,
     },
     virtual_file::MaybeFatalIo,
 };
@@ -111,6 +110,13 @@ pub(super) async fn delete_local_timeline_directory(
     info!("finished deleting layer files, releasing locks");
 }
 
+/// Removes remote layers and an index file after them.
+async fn delete_remote_layers_and_index(
+    remote_client: &Arc<RemoteTimelineClient>,
+) -> anyhow::Result<()> {
+    remote_client.delete_all().await.context("delete_all")
+}
+
 /// It is important that this gets called when DeletionGuard is being held.
 /// For more context see comments in [`DeleteTimelineFlow::prepare`]
 async fn remove_maybe_offloaded_timeline_from_tenant(
@@ -141,10 +147,9 @@ async fn remove_maybe_offloaded_timeline_from_tenant(
             );
         }
         TimelineOrOffloaded::Offloaded(timeline) => {
-            let offloaded_timeline = timelines_offloaded
+            timelines_offloaded
                 .remove(&timeline.timeline_id)
                 .expect("timeline that we were deleting was concurrently removed from 'timelines_offloaded' map");
-            offloaded_timeline.delete_from_ancestor_with_timelines(&timelines);
         }
     }
 
@@ -216,24 +221,11 @@ impl DeleteTimelineFlow {
             None => {
                 let remote_client = tenant
                     .build_timeline_client(timeline.timeline_id(), tenant.remote_storage.clone());
-                let result = match remote_client
+                let result = remote_client
                     .download_index_file(&tenant.cancel)
                     .instrument(info_span!("download_index_file"))
                     .await
-                {
-                    Ok(r) => r,
-                    Err(DownloadError::NotFound) => {
-                        // Deletion is already complete
-                        tracing::info!("Timeline already deleted in remote storage");
-                        return Ok(());
-                    }
-                    Err(e) => {
-                        return Err(DeleteTimelineError::Other(anyhow::anyhow!(
-                            "error: {:?}",
-                            e
-                        )));
-                    }
-                };
+                    .map_err(|e| DeleteTimelineError::Other(anyhow::anyhow!("error: {:?}", e)))?;
                 let index_part = match result {
                     MaybeDeletedIndexPart::Deleted(p) => {
                         tracing::info!("Timeline already set as deleted in remote index");
@@ -414,12 +406,7 @@ impl DeleteTimelineFlow {
             "timeline_delete",
             async move {
                 if let Err(err) = Self::background(guard, conf, &tenant, &timeline, remote_client).await {
-                    // Only log as an error if it's not a cancellation.
-                    if matches!(err, DeleteTimelineError::Cancelled) {
-                        info!("Shutdown during timeline deletion");
-                    }else {
-                        error!("Error: {err:#}");
-                    }
+                    error!("Error: {err:#}");
                     if let TimelineOrOffloaded::Timeline(timeline) = timeline {
                         timeline.set_broken(format!("{err:#}"))
                     }
@@ -451,7 +438,7 @@ impl DeleteTimelineFlow {
             Err(anyhow::anyhow!("failpoint: timeline-delete-after-rm"))?
         });
 
-        remote_client.delete_all().await?;
+        delete_remote_layers_and_index(&remote_client).await?;
 
         pausable_failpoint!("in_progress_delete");
 
@@ -462,10 +449,10 @@ impl DeleteTimelineFlow {
         // So indeed, the tenant manifest might refer to an offloaded timeline which has already been deleted.
         // However, we handle this case in tenant loading code so the next time we attach, the issue is
         // resolved.
-        tenant.store_tenant_manifest().await.map_err(|e| match e {
-            TenantManifestError::Cancelled => DeleteTimelineError::Cancelled,
-            _ => DeleteTimelineError::Other(e.into()),
-        })?;
+        tenant
+            .store_tenant_manifest()
+            .await
+            .map_err(|e| DeleteTimelineError::Other(anyhow::anyhow!(e)))?;
 
         *guard = Self::Finished;
 

pageserver/src/tenant/timeline/offload.rs

@@ -66,7 +66,7 @@ pub(crate) async fn offload_timeline(
     let conf = &tenant.conf;
     delete_local_timeline_directory(conf, tenant.tenant_shard_id, &timeline).await;
 
-    let remaining_refcount = remove_timeline_from_tenant(tenant, &timeline, &guard);
+    remove_timeline_from_tenant(tenant, &timeline, &guard);
 
     {
         let mut offloaded_timelines = tenant.timelines_offloaded.lock().unwrap();
@@ -87,20 +87,16 @@ pub(crate) async fn offload_timeline(
     // not our actual state of offloaded timelines.
     tenant.store_tenant_manifest().await?;
 
-    tracing::info!("Timeline offload complete (remaining arc refcount: {remaining_refcount})");
-
     Ok(())
 }
 
 /// It is important that this gets called when DeletionGuard is being held.
 /// For more context see comments in [`DeleteTimelineFlow::prepare`]
-///
-/// Returns the strong count of the timeline `Arc`
 fn remove_timeline_from_tenant(
     tenant: &Tenant,
     timeline: &Timeline,
     _: &DeletionGuard, // using it as a witness
-) -> usize {
+) {
     // Remove the timeline from the map.
     let mut timelines = tenant.timelines.lock().unwrap();
     let children_exist = timelines
@@ -113,9 +109,7 @@ fn remove_timeline_from_tenant(
         panic!("Timeline grew children while we removed layer files");
     }
 
-    let timeline = timelines
+    timelines
         .remove(&timeline.timeline_id)
         .expect("timeline that we were deleting was concurrently removed from 'timelines' map");
-
-    Arc::strong_count(&timeline)
 }

pageserver/src/tenant/timeline/walreceiver.rs

@@ -38,7 +38,6 @@ use storage_broker::BrokerClientChannel;
 use tokio::sync::watch;
 use tokio_util::sync::CancellationToken;
 use tracing::*;
-use utils::postgres_client::PostgresClientProtocol;
 
 use self::connection_manager::ConnectionManagerStatus;
 
@@ -46,7 +45,6 @@ use super::Timeline;
 
 #[derive(Clone)]
 pub struct WalReceiverConf {
-    pub protocol: PostgresClientProtocol,
     /// The timeout on the connection to safekeeper for WAL streaming.
     pub wal_connect_timeout: Duration,
     /// The timeout to use to determine when the current connection is "stale" and reconnect to the other one.

pageserver/src/tenant/timeline/walreceiver/connection_manager.rs

@@ -36,9 +36,7 @@ use postgres_connection::PgConnectionConfig;
 use utils::backoff::{
     exponential_backoff, DEFAULT_BASE_BACKOFF_SECONDS, DEFAULT_MAX_BACKOFF_SECONDS,
 };
-use utils::postgres_client::{
-    wal_stream_connection_config, ConnectionConfigArgs, PostgresClientProtocol,
-};
+use utils::postgres_client::wal_stream_connection_config;
 use utils::{
     id::{NodeId, TenantTimelineId},
     lsn::Lsn,
@@ -986,33 +984,15 @@ impl ConnectionManagerState {
                 if info.safekeeper_connstr.is_empty() {
                     return None; // no connection string, ignore sk
                 }
-
-                let (shard_number, shard_count, shard_stripe_size) = match self.conf.protocol {
-                    PostgresClientProtocol::Vanilla => {
-                        (None, None, None)
+                match wal_stream_connection_config(
+                    self.id,
+                    info.safekeeper_connstr.as_ref(),
+                    match &self.conf.auth_token {
+                        None => None,
+                        Some(x) => Some(x),
                     },
-                    PostgresClientProtocol::Interpreted => {
-                        let shard_identity = self.timeline.get_shard_identity();
-                        (
-                            Some(shard_identity.number.0),
-                            Some(shard_identity.count.0),
-                            Some(shard_identity.stripe_size.0),
-                        )
-                    }
-                };
-
-                let connection_conf_args = ConnectionConfigArgs {
-                    protocol: self.conf.protocol,
-                    ttid: self.id,
-                    shard_number,
-                    shard_count,
-                    shard_stripe_size,
-                    listen_pg_addr_str: info.safekeeper_connstr.as_ref(),
-                    auth_token: self.conf.auth_token.as_ref().map(|t| t.as_str()),
-                    availability_zone: self.conf.availability_zone.as_deref()
-                };
-
-                match wal_stream_connection_config(connection_conf_args) {
+                    self.conf.availability_zone.as_deref(),
+                ) {
                     Ok(connstr) => Some((*sk_id, info, connstr)),
                     Err(e) => {
                         error!("Failed to create wal receiver connection string from broker data of safekeeper node {}: {e:#}", sk_id);
@@ -1116,7 +1096,6 @@ impl ReconnectReason {
 mod tests {
     use super::*;
     use crate::tenant::harness::{TenantHarness, TIMELINE_ID};
-    use pageserver_api::config::defaults::DEFAULT_WAL_RECEIVER_PROTOCOL;
     use url::Host;
 
     fn dummy_broker_sk_timeline(
@@ -1553,7 +1532,6 @@ mod tests {
             timeline,
             cancel: CancellationToken::new(),
             conf: WalReceiverConf {
-                protocol: DEFAULT_WAL_RECEIVER_PROTOCOL,
                 wal_connect_timeout: Duration::from_secs(1),
                 lagging_wal_timeout: Duration::from_secs(1),
                 max_lsn_wal_lag: NonZeroU64::new(1024 * 1024).unwrap(),

pageserver/src/tenant/timeline/walreceiver/walreceiver_connection.rs

@@ -36,7 +36,7 @@ use crate::{
 use postgres_backend::is_expected_io_error;
 use postgres_connection::PgConnectionConfig;
 use postgres_ffi::waldecoder::WalStreamDecoder;
-use utils::{bin_ser::BeSer, id::NodeId, lsn::Lsn};
+use utils::{id::NodeId, lsn::Lsn};
 use utils::{pageserver_feedback::PageserverFeedback, sync::gate::GateError};
 
 /// Status of the connection.
@@ -291,15 +291,6 @@ pub(super) async fn handle_walreceiver_connection(
                 connection_status.latest_connection_update = now;
                 connection_status.commit_lsn = Some(Lsn::from(keepalive.wal_end()));
             }
-            ReplicationMessage::RawInterpretedWalRecords(raw) => {
-                connection_status.latest_connection_update = now;
-                if !raw.data().is_empty() {
-                    connection_status.latest_wal_update = now;
-                }
-
-                connection_status.commit_lsn = Some(Lsn::from(raw.commit_lsn()));
-                connection_status.streaming_lsn = Some(Lsn::from(raw.streaming_lsn()));
-            }
             &_ => {}
         };
         if let Err(e) = events_sender.send(TaskStateUpdate::Progress(connection_status)) {
@@ -307,130 +298,7 @@ pub(super) async fn handle_walreceiver_connection(
             return Ok(());
         }
 
-        async fn commit(
-            modification: &mut DatadirModification<'_>,
-            uncommitted: &mut u64,
-            filtered: &mut u64,
-            ctx: &RequestContext,
-        ) -> anyhow::Result<()> {
-            WAL_INGEST
-                .records_committed
-                .inc_by(*uncommitted - *filtered);
-            modification.commit(ctx).await?;
-            *uncommitted = 0;
-            *filtered = 0;
-            Ok(())
-        }
-
         let status_update = match replication_message {
-            ReplicationMessage::RawInterpretedWalRecords(raw) => {
-                WAL_INGEST.bytes_received.inc_by(raw.data().len() as u64);
-
-                let mut uncommitted_records = 0;
-                let mut filtered_records = 0;
-
-                // This is the end LSN of the raw WAL from which the records
-                // were interpreted.
-                let streaming_lsn = Lsn::from(raw.streaming_lsn());
-                tracing::debug!(
-                    "Received WAL up to {streaming_lsn} with next_record_lsn={}",
-                    Lsn(raw.next_record_lsn().unwrap_or(0))
-                );
-
-                let records = Vec::<InterpretedWalRecord>::des(raw.data()).with_context(|| {
-                    anyhow::anyhow!(
-                        "Failed to deserialize interpreted records ending at LSN {streaming_lsn}"
-                    )
-                })?;
-
-                // We start the modification at 0 because each interpreted record
-                // advances it to its end LSN. 0 is just an initialization placeholder.
-                let mut modification = timeline.begin_modification(Lsn(0));
-
-                for interpreted in records {
-                    if matches!(interpreted.flush_uncommitted, FlushUncommittedRecords::Yes)
-                        && uncommitted_records > 0
-                    {
-                        commit(
-                            &mut modification,
-                            &mut uncommitted_records,
-                            &mut filtered_records,
-                            &ctx,
-                        )
-                        .await?;
-                    }
-
-                    let next_record_lsn = interpreted.next_record_lsn;
-                    let ingested = walingest
-                        .ingest_record(interpreted, &mut modification, &ctx)
-                        .await
-                        .with_context(|| format!("could not ingest record at {next_record_lsn}"))?;
-
-                    if !ingested {
-                        tracing::debug!("ingest: filtered out record @ LSN {next_record_lsn}");
-                        WAL_INGEST.records_filtered.inc();
-                        filtered_records += 1;
-                    }
-
-                    uncommitted_records += 1;
-
-                    // FIXME: this cannot be made pausable_failpoint without fixing the
-                    // failpoint library; in tests, the added amount of debugging will cause us
-                    // to timeout the tests.
-                    fail_point!("walreceiver-after-ingest");
-
-                    // Commit every ingest_batch_size records. Even if we filtered out
-                    // all records, we still need to call commit to advance the LSN.
-                    if uncommitted_records >= ingest_batch_size
-                        || modification.approx_pending_bytes()
-                            > DatadirModification::MAX_PENDING_BYTES
-                    {
-                        commit(
-                            &mut modification,
-                            &mut uncommitted_records,
-                            &mut filtered_records,
-                            &ctx,
-                        )
-                        .await?;
-                    }
-                }
-
-                // Records might have been filtered out on the safekeeper side, but we still
-                // need to advance last record LSN on all shards. If we've not ingested the latest
-                // record, then set the LSN of the modification past it. This way all shards
-                // advance their last record LSN at the same time.
-                let needs_last_record_lsn_advance = match raw.next_record_lsn().map(Lsn::from) {
-                    Some(lsn) if lsn > modification.get_lsn() => {
-                        modification.set_lsn(lsn).unwrap();
-                        true
-                    }
-                    _ => false,
-                };
-
-                if uncommitted_records > 0 || needs_last_record_lsn_advance {
-                    // Commit any uncommitted records
-                    commit(
-                        &mut modification,
-                        &mut uncommitted_records,
-                        &mut filtered_records,
-                        &ctx,
-                    )
-                    .await?;
-                }
-
-                if !caught_up && streaming_lsn >= end_of_wal {
-                    info!("caught up at LSN {streaming_lsn}");
-                    caught_up = true;
-                }
-
-                tracing::debug!(
-                    "Ingested WAL up to {streaming_lsn}. Last record LSN is {}",
-                    timeline.get_last_record_lsn()
-                );
-
-                Some(streaming_lsn)
-            }
-
             ReplicationMessage::XLogData(xlog_data) => {
                 // Pass the WAL data to the decoder, and see if we can decode
                 // more records as a result.
@@ -448,11 +316,26 @@ pub(super) async fn handle_walreceiver_connection(
                     let mut uncommitted_records = 0;
                     let mut filtered_records = 0;
 
-                    while let Some((next_record_lsn, recdata)) = waldecoder.poll_decode()? {
+                    async fn commit(
+                        modification: &mut DatadirModification<'_>,
+                        uncommitted: &mut u64,
+                        filtered: &mut u64,
+                        ctx: &RequestContext,
+                    ) -> anyhow::Result<()> {
+                        WAL_INGEST
+                            .records_committed
+                            .inc_by(*uncommitted - *filtered);
+                        modification.commit(ctx).await?;
+                        *uncommitted = 0;
+                        *filtered = 0;
+                        Ok(())
+                    }
+
+                    while let Some((record_end_lsn, recdata)) = waldecoder.poll_decode()? {
                         // It is important to deal with the aligned records as lsn in getPage@LSN is
                         // aligned and can be several bytes bigger. Without this alignment we are
                         // at risk of hitting a deadlock.
-                        if !next_record_lsn.is_aligned() {
+                        if !record_end_lsn.is_aligned() {
                             return Err(WalReceiverError::Other(anyhow!("LSN not aligned")));
                         }
 
@@ -460,7 +343,7 @@ pub(super) async fn handle_walreceiver_connection(
                         let interpreted = InterpretedWalRecord::from_bytes_filtered(
                             recdata,
                             modification.tline.get_shard_identity(),
-                            next_record_lsn,
+                            record_end_lsn,
                             modification.tline.pg_version,
                         )?;
 
@@ -484,10 +367,10 @@ pub(super) async fn handle_walreceiver_connection(
                             .ingest_record(interpreted, &mut modification, &ctx)
                             .await
                             .with_context(|| {
-                                format!("could not ingest record at {next_record_lsn}")
+                                format!("could not ingest record at {record_end_lsn}")
                             })?;
                         if !ingested {
-                            tracing::debug!("ingest: filtered out record @ LSN {next_record_lsn}");
+                            tracing::debug!("ingest: filtered out record @ LSN {record_end_lsn}");
                             WAL_INGEST.records_filtered.inc();
                             filtered_records += 1;
                         }
@@ -497,7 +380,7 @@ pub(super) async fn handle_walreceiver_connection(
                         // to timeout the tests.
                         fail_point!("walreceiver-after-ingest");
 
-                        last_rec_lsn = next_record_lsn;
+                        last_rec_lsn = record_end_lsn;
 
                         // Commit every ingest_batch_size records. Even if we filtered out
                         // all records, we still need to call commit to advance the LSN.

pageserver/src/virtual_file.rs

@@ -175,16 +175,10 @@ impl VirtualFile {
     }
 
     pub async fn sync_all(&self) -> Result<(), Error> {
-        if SYNC_MODE.load(std::sync::atomic::Ordering::Relaxed) == SyncMode::UnsafeNoSync as u8 {
-            return Ok(());
-        }
         self.inner.sync_all().await
     }
 
     pub async fn sync_data(&self) -> Result<(), Error> {
-        if SYNC_MODE.load(std::sync::atomic::Ordering::Relaxed) == SyncMode::UnsafeNoSync as u8 {
-            return Ok(());
-        }
         self.inner.sync_data().await
     }
 
@@ -239,27 +233,6 @@ impl VirtualFile {
     }
 }
 
-/// Indicates whether to enable fsync, fdatasync, or O_SYNC/O_DSYNC when writing
-/// files. Switching this off is unsafe and only used for testing on machines
-/// with slow drives.
-#[repr(u8)]
-pub enum SyncMode {
-    Sync,
-    UnsafeNoSync,
-}
-
-impl TryFrom<u8> for SyncMode {
-    type Error = u8;
-
-    fn try_from(value: u8) -> Result<Self, Self::Error> {
-        Ok(match value {
-            v if v == (SyncMode::Sync as u8) => SyncMode::Sync,
-            v if v == (SyncMode::UnsafeNoSync as u8) => SyncMode::UnsafeNoSync,
-            x => return Err(x),
-        })
-    }
-}
-
 ///
 /// A virtual file descriptor. You can use this just like std::fs::File, but internally
 /// the underlying file is closed if the system is low on file descriptors,
@@ -1359,13 +1332,12 @@ impl OpenFiles {
 /// server startup.
 ///
 #[cfg(not(test))]
-pub fn init(num_slots: usize, engine: IoEngineKind, mode: IoMode, sync_mode: SyncMode) {
+pub fn init(num_slots: usize, engine: IoEngineKind, mode: IoMode) {
     if OPEN_FILES.set(OpenFiles::new(num_slots)).is_err() {
         panic!("virtual_file::init called twice");
     }
     set_io_mode(mode);
     io_engine::init(engine);
-    SYNC_MODE.store(sync_mode as u8, std::sync::atomic::Ordering::Relaxed);
     crate::metrics::virtual_file_descriptor_cache::SIZE_MAX.set(num_slots as u64);
 }
 
@@ -1407,9 +1379,6 @@ pub(crate) fn set_io_mode(mode: IoMode) {
 pub(crate) fn get_io_mode() -> IoMode {
     IoMode::try_from(IO_MODE.load(Ordering::Relaxed)).unwrap()
 }
-
-static SYNC_MODE: AtomicU8 = AtomicU8::new(SyncMode::Sync as u8);
-
 #[cfg(test)]
 mod tests {
     use crate::context::DownloadBehavior;

pageserver/src/walingest.rs

@@ -154,7 +154,7 @@ impl WalIngest {
         WAL_INGEST.records_received.inc();
         let prev_len = modification.len();
 
-        modification.set_lsn(interpreted.next_record_lsn)?;
+        modification.set_lsn(interpreted.end_lsn)?;
 
         if matches!(interpreted.flush_uncommitted, FlushUncommittedRecords::Yes) {
             // Records of this type should always be preceded by a commit(), as they
@@ -587,29 +587,11 @@ impl WalIngest {
                 forknum: VISIBILITYMAP_FORKNUM,
             };
 
-            // last remaining block, byte, and bit
-            let mut vm_page_no = blkno / (pg_constants::VM_HEAPBLOCKS_PER_PAGE as u32);
-            let trunc_byte = blkno as usize % pg_constants::VM_HEAPBLOCKS_PER_PAGE
-                / pg_constants::VM_HEAPBLOCKS_PER_BYTE;
-            let trunc_offs = blkno as usize % pg_constants::VM_HEAPBLOCKS_PER_BYTE
-                * pg_constants::VM_BITS_PER_HEAPBLOCK;
-
-            // Unless the new size is exactly at a visibility map page boundary, the
-            // tail bits in the last remaining map page, representing truncated heap
-            // blocks, need to be cleared. This is not only tidy, but also necessary
-            // because we don't get a chance to clear the bits if the heap is extended
-            // again.
-            if (trunc_byte != 0 || trunc_offs != 0)
-                && self.shard.is_key_local(&rel_block_to_key(rel, vm_page_no))
-            {
-                modification.put_rel_wal_record(
-                    rel,
-                    vm_page_no,
-                    NeonWalRecord::TruncateVisibilityMap {
-                        trunc_byte,
-                        trunc_offs,
-                    },
-                )?;
+            let mut vm_page_no = blkno / pg_constants::VM_HEAPBLOCKS_PER_PAGE;
+            if blkno % pg_constants::VM_HEAPBLOCKS_PER_PAGE != 0 {
+                // Tail of last remaining vm page has to be zeroed.
+                // We are not precise here and instead of digging in VM bitmap format just clear the whole page.
+                modification.put_rel_page_image_zero(rel, vm_page_no)?;
                 vm_page_no += 1;
             }
             let nblocks = get_relsize(modification, rel, ctx).await?;

pageserver/src/walredo/apply_neon.rs

@@ -42,34 +42,6 @@ pub(crate) fn apply_in_neon(
         } => {
             anyhow::bail!("tried to pass postgres wal record to neon WAL redo");
         }
-        //
-        // Code copied from PostgreSQL `visibilitymap_prepare_truncate` function in `visibilitymap.c`
-        //
-        NeonWalRecord::TruncateVisibilityMap {
-            trunc_byte,
-            trunc_offs,
-        } => {
-            // sanity check that this is modifying the correct relation
-            let (rel, _) = key.to_rel_block().context("invalid record")?;
-            assert!(
-                rel.forknum == VISIBILITYMAP_FORKNUM,
-                "TruncateVisibilityMap record on unexpected rel {}",
-                rel
-            );
-            let map = &mut page[pg_constants::MAXALIGN_SIZE_OF_PAGE_HEADER_DATA..];
-            map[*trunc_byte + 1..].fill(0u8);
-            /*----
-             * Mask out the unwanted bits of the last remaining byte.
-             *
-             * ((1 << 0) - 1) = 00000000
-             * ((1 << 1) - 1) = 00000001
-             * ...
-             * ((1 << 6) - 1) = 00111111
-             * ((1 << 7) - 1) = 01111111
-             *----
-             */
-            map[*trunc_byte] &= (1 << *trunc_offs) - 1;
-        }
         NeonWalRecord::ClearVisibilityMapFlags {
             new_heap_blkno,
             old_heap_blkno,
@@ -281,10 +253,6 @@ pub(crate) fn apply_in_neon(
             use bytes::BufMut;
             if *will_init {
                 assert!(*clear, "init record must be clear to ensure correctness");
-                assert!(
-                    page.is_empty(),
-                    "init record must be the first entry to ensure correctness"
-                );
             }
             if *clear {
                 page.clear();

pgxn/neon/logical_replication_monitor.c

@@ -1,8 +1,7 @@
-#include <dirent.h>
 #include <limits.h>
 #include <string.h>
+#include <dirent.h>
 #include <signal.h>
-#include <sys/stat.h>
 
 #include "postgres.h"
 
@@ -22,35 +21,17 @@
 
 static int	logical_replication_max_snap_files = 300;
 
-/*
- * According to Chi (shyzh), the pageserver _should_ be good with 10 MB worth of
- * snapshot files. Let's use 8 MB since 8 is a power of 2.
- */
-static int	logical_replication_max_logicalsnapdir_size = 8000;
-
-/*
- * A primitive description of a logical snapshot file including the LSN of the
- * file and its size.
- */
-typedef struct SnapDesc {
-	XLogRecPtr	lsn;
-	off_t		sz;
-} SnapDesc;
-
 PGDLLEXPORT void LogicalSlotsMonitorMain(Datum main_arg);
 
-/*
- * Sorts an array of snapshot descriptors by their LSN.
- */
 static int
-SnapDescComparator(const void *a, const void *b)
+LsnDescComparator(const void *a, const void *b)
 {
-	const SnapDesc	*desc1 = a;
-	const SnapDesc	*desc2 = b;
+	XLogRecPtr	lsn1 = *((const XLogRecPtr *) a);
+	XLogRecPtr	lsn2 = *((const XLogRecPtr *) b);
 
-	if (desc1->lsn < desc2->lsn)
+	if (lsn1 < lsn2)
 		return 1;
-	else if (desc1->lsn == desc2->lsn)
+	else if (lsn1 == lsn2)
 		return 0;
 	else
 		return -1;
@@ -62,39 +43,28 @@ SnapDescComparator(const void *a, const void *b)
  * slots having lower restart_lsn should be dropped.
  */
 static XLogRecPtr
-get_snapshots_cutoff_lsn(void)
+get_num_snap_files_lsn_threshold(void)
 {
-/* PG 18 has a constant defined for this, PG_LOGICAL_SNAPSHOTS_DIR */
-#define SNAPDIR "pg_logical/snapshots"
-
 	DIR		   *dirdesc;
-	int			dirdesc_fd;
 	struct dirent *de;
-	size_t		snapshot_index = 0;
-	SnapDesc   *snapshot_descriptors;
-	size_t		descriptors_allocated = 1024;
-	XLogRecPtr	cutoff = 0;
-	off_t		logicalsnapdir_size = 0;
-	const int	logical_replication_max_logicalsnapdir_size_bytes = logical_replication_max_logicalsnapdir_size * 1000;
+	char	   *snap_path = "pg_logical/snapshots/";
+	int			lsns_allocated = 1024;
+	int			lsns_num = 0;
+	XLogRecPtr *lsns;
+	XLogRecPtr	cutoff;
 
-	if (logical_replication_max_snap_files < 0 && logical_replication_max_logicalsnapdir_size < 0)
+	if (logical_replication_max_snap_files < 0)
 		return 0;
 
-	snapshot_descriptors = palloc(sizeof(*snapshot_descriptors) * descriptors_allocated);
-
-	dirdesc = AllocateDir(SNAPDIR);
-	dirdesc_fd = dirfd(dirdesc);
-	if (dirdesc_fd == -1)
-		ereport(ERROR, errmsg("failed to get a file descriptor for " SNAPDIR ": %m"));
+	lsns = palloc(sizeof(XLogRecPtr) * lsns_allocated);
 
 	/* find all .snap files and get their lsns */
-	while ((de = ReadDir(dirdesc, SNAPDIR)) != NULL)
+	dirdesc = AllocateDir(snap_path);
+	while ((de = ReadDir(dirdesc, snap_path)) != NULL)
 	{
+		XLogRecPtr	lsn;
 		uint32		hi;
 		uint32		lo;
-		struct stat	st;
-		XLogRecPtr	lsn;
-		SnapDesc   *desc;
 
 		if (strcmp(de->d_name, ".") == 0 ||
 			strcmp(de->d_name, "..") == 0)
@@ -109,69 +79,28 @@ get_snapshots_cutoff_lsn(void)
 
 		lsn = ((uint64) hi) << 32 | lo;
 		elog(DEBUG5, "found snap file %X/%X", LSN_FORMAT_ARGS(lsn));
-
-		if (fstatat(dirdesc_fd, de->d_name, &st, 0) == -1)
-			ereport(ERROR, errmsg("failed to get the size of " SNAPDIR "/%s: %m", de->d_name));
-
-		if (descriptors_allocated == snapshot_index)
+		if (lsns_allocated == lsns_num)
 		{
-			descriptors_allocated *= 2;
-			snapshot_descriptors = repalloc(snapshot_descriptors, sizeof(*snapshot_descriptors) * descriptors_allocated);
+			lsns_allocated *= 2;
+			lsns = repalloc(lsns, sizeof(XLogRecPtr) * lsns_allocated);
 		}
-
-		desc = &snapshot_descriptors[snapshot_index++];
-		desc->lsn = lsn;
-		desc->sz = st.st_size;
+		lsns[lsns_num++] = lsn;
 	}
-
-	qsort(snapshot_descriptors, snapshot_index, sizeof(*snapshot_descriptors), SnapDescComparator);
-
-	/* Are there more snapshot files than specified? */
-	if (logical_replication_max_snap_files <= snapshot_index)
+	/* sort by lsn desc */
+	qsort(lsns, lsns_num, sizeof(XLogRecPtr), LsnDescComparator);
+	/* and take cutoff at logical_replication_max_snap_files */
+	if (logical_replication_max_snap_files > lsns_num)
+		cutoff = 0;
+	/* have less files than cutoff */
+	else
 	{
-		cutoff = snapshot_descriptors[logical_replication_max_snap_files - 1].lsn;
-		elog(LOG,
-			"ls_monitor: dropping logical slots with restart_lsn lower %X/%X, found %zu snapshot files, limit is %d",
-			LSN_FORMAT_ARGS(cutoff), snapshot_index, logical_replication_max_snap_files);
+		cutoff = lsns[logical_replication_max_snap_files - 1];
+		elog(LOG, "ls_monitor: dropping logical slots with restart_lsn lower %X/%X, found %d .snap files, limit is %d",
+			 LSN_FORMAT_ARGS(cutoff), lsns_num, logical_replication_max_snap_files);
 	}
-
-	/* Is the size of the logical snapshots directory larger than specified?
-	 *
-	 * It's possible we could hit both thresholds, so remove any extra files
-	 * first, and then truncate based on size of the remaining files.
-	 */
-	if (logicalsnapdir_size > logical_replication_max_logicalsnapdir_size_bytes)
-	{
-		/* Unfortunately, iterating the directory does not guarantee any order
-		 * so we can't cache an index in the preceding loop.
-		 */
-
-		off_t		sz;
-		const XLogRecPtr original = cutoff;
-
-		sz = snapshot_descriptors[0].sz;
-		for (size_t i = 1; i < logical_replication_max_snap_files; ++i)
-		{
-			if (sz > logical_replication_max_logicalsnapdir_size_bytes)
-			{
-				cutoff = snapshot_descriptors[i - 1].lsn;
-				break;
-			}
-
-			sz += snapshot_descriptors[i].sz;
-		}
-
-		if (cutoff != original)
-			elog(LOG, "ls_monitor: dropping logical slots with restart_lsn lower than %X/%X, " SNAPDIR " is larger than %d KB",
-					LSN_FORMAT_ARGS(cutoff), logical_replication_max_logicalsnapdir_size);
-	}
-
-	pfree(snapshot_descriptors);
+	pfree(lsns);
 	FreeDir(dirdesc);
-
 	return cutoff;
-
-#undef SNAPDIR
 }
 
 void
@@ -189,16 +118,6 @@ InitLogicalReplicationMonitor(void)
 							0,
 							NULL, NULL, NULL);
 
-	DefineCustomIntVariable(
-							"neon.logical_replication_max_logicalsnapdir_size",
-							"Maximum allowed size of the pg_logical/snapshots directory (KB). When exceeded, slots are dropped until the limit is met. -1 disables the limit.",
-							NULL,
-							&logical_replication_max_logicalsnapdir_size,
-							8000, -1, INT_MAX,
-							PGC_SIGHUP,
-							GUC_UNIT_KB,
-							NULL, NULL, NULL);
-
 	memset(&bgw, 0, sizeof(bgw));
 	bgw.bgw_flags = BGWORKER_SHMEM_ACCESS;
 	bgw.bgw_start_time = BgWorkerStart_RecoveryFinished;
@@ -243,7 +162,7 @@ LogicalSlotsMonitorMain(Datum main_arg)
 		 * If there are too many .snap files, just drop all logical slots to
 		 * prevent aux files bloat.
 		 */
-		cutoff_lsn = get_snapshots_cutoff_lsn();
+		cutoff_lsn = get_num_snap_files_lsn_threshold();
 		if (cutoff_lsn > 0)
 		{
 			for (int i = 0; i < max_replication_slots; i++)

pgxn/neon/neon.c

@@ -512,7 +512,7 @@ neon_shmem_startup_hook(void)
 	if (prev_shmem_startup_hook)
 		prev_shmem_startup_hook();
 
-#if PG_MAJORVERSION_NUM >= 17
+#if PG_PG_MAJORVERSION_NUM >= 17
 	WAIT_EVENT_NEON_LFC_MAINTENANCE = WaitEventExtensionNew("Neon/FileCache_Maintenance");
 	WAIT_EVENT_NEON_LFC_READ = WaitEventExtensionNew("Neon/FileCache_Read");
 	WAIT_EVENT_NEON_LFC_TRUNCATE = WaitEventExtensionNew("Neon/FileCache_Truncate");

safekeeper/Cargo.toml

@@ -28,7 +28,6 @@ hyper0.workspace = true
 futures.workspace = true
 once_cell.workspace = true
 parking_lot.workspace = true
-pageserver_api.workspace = true
 postgres.workspace = true
 postgres-protocol.workspace = true
 rand.workspace = true
@@ -58,7 +57,6 @@ sd-notify.workspace = true
 storage_broker.workspace = true
 tokio-stream.workspace = true
 utils.workspace = true
-wal_decoder.workspace = true
 
 workspace_hack.workspace = true
 

safekeeper/src/handler.rs

@@ -2,15 +2,11 @@
 //! protocol commands.
 
 use anyhow::Context;
-use pageserver_api::models::ShardParameters;
-use pageserver_api::shard::{ShardIdentity, ShardStripeSize};
 use std::future::Future;
 use std::str::{self, FromStr};
 use std::sync::Arc;
 use tokio::io::{AsyncRead, AsyncWrite};
 use tracing::{debug, info, info_span, Instrument};
-use utils::postgres_client::PostgresClientProtocol;
-use utils::shard::{ShardCount, ShardNumber};
 
 use crate::auth::check_permission;
 use crate::json_ctrl::{handle_json_ctrl, AppendLogicalMessage};
@@ -39,8 +35,6 @@ pub struct SafekeeperPostgresHandler {
     pub tenant_id: Option<TenantId>,
     pub timeline_id: Option<TimelineId>,
     pub ttid: TenantTimelineId,
-    pub shard: Option<ShardIdentity>,
-    pub protocol: Option<PostgresClientProtocol>,
     /// Unique connection id is logged in spans for observability.
     pub conn_id: ConnectionId,
     /// Auth scope allowed on the connections and public key used to check auth tokens. None if auth is not configured.
@@ -113,28 +107,11 @@ impl<IO: AsyncRead + AsyncWrite + Unpin + Send> postgres_backend::Handler<IO>
     ) -> Result<(), QueryError> {
         if let FeStartupPacket::StartupMessage { params, .. } = sm {
             if let Some(options) = params.options_raw() {
-                let mut shard_count: Option<u8> = None;
-                let mut shard_number: Option<u8> = None;
-                let mut shard_stripe_size: Option<u32> = None;
-
                 for opt in options {
                     // FIXME `ztenantid` and `ztimelineid` left for compatibility during deploy,
                     // remove these after the PR gets deployed:
                     // https://github.com/neondatabase/neon/pull/2433#discussion_r970005064
                     match opt.split_once('=') {
-                        Some(("protocol", value)) => {
-                            let raw_value = value
-                                .parse::<u8>()
-                                .with_context(|| format!("Failed to parse {value} as protocol"))?;
-
-                            self.protocol = Some(
-                                PostgresClientProtocol::try_from(raw_value).map_err(|_| {
-                                    QueryError::Other(anyhow::anyhow!(
-                                        "Unexpected client protocol type: {raw_value}"
-                                    ))
-                                })?,
-                            );
-                        }
                         Some(("ztenantid", value)) | Some(("tenant_id", value)) => {
                             self.tenant_id = Some(value.parse().with_context(|| {
                                 format!("Failed to parse {value} as tenant id")
@@ -150,54 +127,9 @@ impl<IO: AsyncRead + AsyncWrite + Unpin + Send> postgres_backend::Handler<IO>
                                 metrics.set_client_az(client_az)
                             }
                         }
-                        Some(("shard_count", value)) => {
-                            shard_count = Some(value.parse::<u8>().with_context(|| {
-                                format!("Failed to parse {value} as shard count")
-                            })?);
-                        }
-                        Some(("shard_number", value)) => {
-                            shard_number = Some(value.parse::<u8>().with_context(|| {
-                                format!("Failed to parse {value} as shard number")
-                            })?);
-                        }
-                        Some(("shard_stripe_size", value)) => {
-                            shard_stripe_size = Some(value.parse::<u32>().with_context(|| {
-                                format!("Failed to parse {value} as shard stripe size")
-                            })?);
-                        }
                         _ => continue,
                     }
                 }
-
-                match self.protocol() {
-                    PostgresClientProtocol::Vanilla => {
-                        if shard_count.is_some()
-                            || shard_number.is_some()
-                            || shard_stripe_size.is_some()
-                        {
-                            return Err(QueryError::Other(anyhow::anyhow!(
-                                "Shard params specified for vanilla protocol"
-                            )));
-                        }
-                    }
-                    PostgresClientProtocol::Interpreted => {
-                        match (shard_count, shard_number, shard_stripe_size) {
-                            (Some(count), Some(number), Some(stripe_size)) => {
-                                let params = ShardParameters {
-                                    count: ShardCount(count),
-                                    stripe_size: ShardStripeSize(stripe_size),
-                                };
-                                self.shard =
-                                    Some(ShardIdentity::from_params(ShardNumber(number), &params));
-                            }
-                            _ => {
-                                return Err(QueryError::Other(anyhow::anyhow!(
-                                    "Shard params were not specified"
-                                )));
-                            }
-                        }
-                    }
-                }
             }
 
             if let Some(app_name) = params.get("application_name") {
@@ -218,11 +150,6 @@ impl<IO: AsyncRead + AsyncWrite + Unpin + Send> postgres_backend::Handler<IO>
                     tracing::field::debug(self.appname.clone()),
                 );
 
-            if let Some(shard) = self.shard.as_ref() {
-                tracing::Span::current()
-                    .record("shard", tracing::field::display(shard.shard_slug()));
-            }
-
             Ok(())
         } else {
             Err(QueryError::Other(anyhow::anyhow!(
@@ -331,8 +258,6 @@ impl SafekeeperPostgresHandler {
             tenant_id: None,
             timeline_id: None,
             ttid: TenantTimelineId::empty(),
-            shard: None,
-            protocol: None,
             conn_id,
             claims: None,
             auth,
@@ -340,10 +265,6 @@ impl SafekeeperPostgresHandler {
         }
     }
 
-    pub fn protocol(&self) -> PostgresClientProtocol {
-        self.protocol.unwrap_or(PostgresClientProtocol::Vanilla)
-    }
-
     // when accessing management api supply None as an argument
     // when using to authorize tenant pass corresponding tenant id
     fn check_permission(&self, tenant_id: Option<TenantId>) -> Result<(), QueryError> {

safekeeper/src/lib.rs

@@ -29,7 +29,6 @@ pub mod receive_wal;
 pub mod recovery;
 pub mod remove_wal;
 pub mod safekeeper;
-pub mod send_interpreted_wal;
 pub mod send_wal;
 pub mod state;
 pub mod timeline;
@@ -39,7 +38,6 @@ pub mod timeline_manager;
 pub mod timelines_set;
 pub mod wal_backup;
 pub mod wal_backup_partial;
-pub mod wal_reader_stream;
 pub mod wal_service;
 pub mod wal_storage;
 

safekeeper/src/receive_wal.rs

@@ -562,9 +562,6 @@ impl WalAcceptor {
                     // Don't flush the WAL on every append, only periodically via flush_ticker.
                     // This batches multiple appends per fsync. If the channel is empty after
                     // sending the reply, we'll schedule an immediate flush.
-                    //
-                    // Note that a flush can still happen on segment bounds, which will result
-                    // in an AppendResponse.
                     if let ProposerAcceptorMessage::AppendRequest(append_request) = msg {
                         msg = ProposerAcceptorMessage::NoFlushAppendRequest(append_request);
                         dirty = true;

safekeeper/src/recovery.rs

@@ -17,7 +17,6 @@ use tokio::{
 use tokio_postgres::replication::ReplicationStream;
 use tokio_postgres::types::PgLsn;
 use tracing::*;
-use utils::postgres_client::{ConnectionConfigArgs, PostgresClientProtocol};
 use utils::{id::NodeId, lsn::Lsn, postgres_client::wal_stream_connection_config};
 
 use crate::receive_wal::{WalAcceptor, REPLY_QUEUE_SIZE};
@@ -326,17 +325,7 @@ async fn recovery_stream(
     conf: &SafeKeeperConf,
 ) -> anyhow::Result<String> {
     // TODO: pass auth token
-    let connection_conf_args = ConnectionConfigArgs {
-        protocol: PostgresClientProtocol::Vanilla,
-        ttid: tli.ttid,
-        shard_number: None,
-        shard_count: None,
-        shard_stripe_size: None,
-        listen_pg_addr_str: &donor.pg_connstr,
-        auth_token: None,
-        availability_zone: None,
-    };
-    let cfg = wal_stream_connection_config(connection_conf_args)?;
+    let cfg = wal_stream_connection_config(tli.ttid, &donor.pg_connstr, None, None)?;
     let mut cfg = cfg.to_tokio_postgres_config();
     // It will make safekeeper give out not committed WAL (up to flush_lsn).
     cfg.application_name(&format!("safekeeper_{}", conf.my_id));

safekeeper/src/safekeeper.rs

@@ -947,7 +947,6 @@ where
         // while first connection still gets some packets later. It might be
         // better to not log this as error! above.
         let write_lsn = self.wal_store.write_lsn();
-        let flush_lsn = self.wal_store.flush_lsn();
         if write_lsn > msg.h.begin_lsn {
             bail!(
                 "append request rewrites WAL written before, write_lsn={}, msg lsn={}",
@@ -980,8 +979,7 @@ where
             self.wal_store.flush_wal().await?;
         }
 
-        // Update commit_lsn. It will be flushed to the control file regularly by the timeline
-        // manager, off of the WAL ingest hot path.
+        // Update commit_lsn.
         if msg.h.commit_lsn != Lsn(0) {
             self.update_commit_lsn(msg.h.commit_lsn).await?;
         }
@@ -994,6 +992,15 @@ where
         self.state.inmem.peer_horizon_lsn =
             max(self.state.inmem.peer_horizon_lsn, msg.h.truncate_lsn);
 
+        // Update truncate and commit LSN in control file.
+        // To avoid negative impact on performance of extra fsync, do it only
+        // when commit_lsn delta exceeds WAL segment size.
+        if self.state.commit_lsn + (self.state.server.wal_seg_size as u64)
+            < self.state.inmem.commit_lsn
+        {
+            self.state.flush().await?;
+        }
+
         trace!(
             "processed AppendRequest of len {}, begin_lsn={}, end_lsn={:?}, commit_lsn={:?}, truncate_lsn={:?}, flushed={:?}",
             msg.wal_data.len(),
@@ -1005,9 +1012,7 @@ where
         );
 
         // If flush_lsn hasn't updated, AppendResponse is not very useful.
-        // This is the common case for !require_flush, but a flush can still
-        // happen on segment bounds.
-        if !require_flush && flush_lsn == self.flush_lsn() {
+        if !require_flush {
             return Ok(None);
         }
 

safekeeper/src/send_interpreted_wal.rs

@@ -1,121 +0,0 @@
-use std::time::Duration;
-
-use anyhow::Context;
-use futures::StreamExt;
-use pageserver_api::shard::ShardIdentity;
-use postgres_backend::{CopyStreamHandlerEnd, PostgresBackend};
-use postgres_ffi::MAX_SEND_SIZE;
-use postgres_ffi::{get_current_timestamp, waldecoder::WalStreamDecoder};
-use pq_proto::{BeMessage, InterpretedWalRecordsBody, WalSndKeepAlive};
-use tokio::io::{AsyncRead, AsyncWrite};
-use tokio::time::MissedTickBehavior;
-use utils::bin_ser::BeSer;
-use utils::lsn::Lsn;
-use wal_decoder::models::InterpretedWalRecord;
-
-use crate::send_wal::EndWatchView;
-use crate::wal_reader_stream::{WalBytes, WalReaderStreamBuilder};
-
-/// Shard-aware interpreted record sender.
-/// This is used for sending WAL to the pageserver. Said WAL
-/// is pre-interpreted and filtered for the shard.
-pub(crate) struct InterpretedWalSender<'a, IO> {
-    pub(crate) pgb: &'a mut PostgresBackend<IO>,
-    pub(crate) wal_stream_builder: WalReaderStreamBuilder,
-    pub(crate) end_watch_view: EndWatchView,
-    pub(crate) shard: ShardIdentity,
-    pub(crate) pg_version: u32,
-    pub(crate) appname: Option<String>,
-}
-
-impl<IO: AsyncRead + AsyncWrite + Unpin> InterpretedWalSender<'_, IO> {
-    /// Send interpreted WAL to a receiver.
-    /// Stops when an error occurs or the receiver is caught up and there's no active compute.
-    ///
-    /// Err(CopyStreamHandlerEnd) is always returned; Result is used only for ?
-    /// convenience.
-    pub(crate) async fn run(self) -> Result<(), CopyStreamHandlerEnd> {
-        let mut wal_position = self.wal_stream_builder.start_pos();
-        let mut wal_decoder =
-            WalStreamDecoder::new(self.wal_stream_builder.start_pos(), self.pg_version);
-
-        let stream = self.wal_stream_builder.build(MAX_SEND_SIZE).await?;
-        let mut stream = std::pin::pin!(stream);
-
-        let mut keepalive_ticker = tokio::time::interval(Duration::from_secs(1));
-        keepalive_ticker.set_missed_tick_behavior(MissedTickBehavior::Skip);
-        keepalive_ticker.reset();
-
-        loop {
-            tokio::select! {
-                // Get some WAL from the stream and then: decode, interpret and send it
-                wal = stream.next() => {
-                    let WalBytes { wal, wal_start_lsn, wal_end_lsn, commit_lsn } = match wal {
-                        Some(some) => some?,
-                        None => { break; }
-                    };
-
-                    wal_position = wal_start_lsn;
-                    wal_decoder.feed_bytes(&wal);
-
-                    let mut records = Vec::new();
-                    let mut max_next_record_lsn = None;
-                    while let Some((next_record_lsn, recdata)) = wal_decoder
-                        .poll_decode()
-                        .with_context(|| "Failed to decode WAL")?
-                    {
-                        assert!(next_record_lsn.is_aligned());
-                        max_next_record_lsn = Some(next_record_lsn);
-
-                        // Deserialize and interpret WAL record
-                        let interpreted = InterpretedWalRecord::from_bytes_filtered(
-                            recdata,
-                            &self.shard,
-                            next_record_lsn,
-                            self.pg_version,
-                        )
-                        .with_context(|| "Failed to interpret WAL")?;
-
-                        if !interpreted.is_empty() {
-                            records.push(interpreted);
-                        }
-                    }
-
-                    let mut buf = Vec::new();
-                    records
-                        .ser_into(&mut buf)
-                        .with_context(|| "Failed to serialize interpreted WAL")?;
-
-                    // Reset the keep alive ticker since we are sending something
-                    // over the wire now.
-                    keepalive_ticker.reset();
-
-                    self.pgb
-                        .write_message(&BeMessage::InterpretedWalRecords(InterpretedWalRecordsBody {
-                            streaming_lsn: wal_end_lsn.0,
-                            commit_lsn: commit_lsn.0,
-                            next_record_lsn: max_next_record_lsn.unwrap_or(Lsn::INVALID).0,
-                            data: buf.as_slice(),
-                        })).await?;
-                }
-
-                // Send a periodic keep alive when the connection has been idle for a while.
-                _ = keepalive_ticker.tick() => {
-                    self.pgb
-                        .write_message(&BeMessage::KeepAlive(WalSndKeepAlive {
-                            wal_end: self.end_watch_view.get().0,
-                            timestamp: get_current_timestamp(),
-                            request_reply: true,
-                        }))
-                        .await?;
-                }
-            }
-        }
-
-        // The loop above ends when the receiver is caught up and there's no more WAL to send.
-        Err(CopyStreamHandlerEnd::ServerInitiated(format!(
-            "ending streaming to {:?} at {}, receiver is caughtup and there is no computes",
-            self.appname, wal_position,
-        )))
-    }
-}

safekeeper/src/send_wal.rs

@@ -5,15 +5,12 @@ use crate::handler::SafekeeperPostgresHandler;
 use crate::metrics::RECEIVED_PS_FEEDBACKS;
 use crate::receive_wal::WalReceivers;
 use crate::safekeeper::{Term, TermLsn};
-use crate::send_interpreted_wal::InterpretedWalSender;
 use crate::timeline::WalResidentTimeline;
-use crate::wal_reader_stream::WalReaderStreamBuilder;
 use crate::wal_service::ConnectionId;
 use crate::wal_storage::WalReader;
 use crate::GlobalTimelines;
 use anyhow::{bail, Context as AnyhowContext};
 use bytes::Bytes;
-use futures::future::Either;
 use parking_lot::Mutex;
 use postgres_backend::PostgresBackend;
 use postgres_backend::{CopyStreamHandlerEnd, PostgresBackendReader, QueryError};
@@ -25,7 +22,6 @@ use tokio::io::{AsyncRead, AsyncWrite};
 use utils::failpoint_support;
 use utils::id::TenantTimelineId;
 use utils::pageserver_feedback::PageserverFeedback;
-use utils::postgres_client::PostgresClientProtocol;
 
 use std::cmp::{max, min};
 use std::net::SocketAddr;
@@ -230,7 +226,7 @@ impl WalSenders {
 
     /// Get remote_consistent_lsn reported by the pageserver. Returns None if
     /// client is not pageserver.
-    pub fn get_ws_remote_consistent_lsn(self: &Arc<WalSenders>, id: WalSenderId) -> Option<Lsn> {
+    fn get_ws_remote_consistent_lsn(self: &Arc<WalSenders>, id: WalSenderId) -> Option<Lsn> {
         let shared = self.mutex.lock();
         let slot = shared.get_slot(id);
         match slot.feedback {
@@ -374,16 +370,6 @@ pub struct WalSenderGuard {
     walsenders: Arc<WalSenders>,
 }
 
-impl WalSenderGuard {
-    pub fn id(&self) -> WalSenderId {
-        self.id
-    }
-
-    pub fn walsenders(&self) -> &Arc<WalSenders> {
-        &self.walsenders
-    }
-}
-
 impl Drop for WalSenderGuard {
     fn drop(&mut self) {
         self.walsenders.unregister(self.id);
@@ -454,12 +440,11 @@ impl SafekeeperPostgresHandler {
         }
 
         info!(
-            "starting streaming from {:?}, available WAL ends at {}, recovery={}, appname={:?}, protocol={}",
+            "starting streaming from {:?}, available WAL ends at {}, recovery={}, appname={:?}",
             start_pos,
             end_pos,
             matches!(end_watch, EndWatch::Flush(_)),
-            appname,
-            self.protocol(),
+            appname
         );
 
         // switch to copy
@@ -471,49 +456,19 @@ impl SafekeeperPostgresHandler {
         // not synchronized with sends, so this avoids deadlocks.
         let reader = pgb.split().context("START_REPLICATION split")?;
 
-        let send_fut = match self.protocol() {
-            PostgresClientProtocol::Vanilla => {
-                let sender = WalSender {
-                    pgb,
-                    // should succeed since we're already holding another guard
-                    tli: tli.wal_residence_guard().await?,
-                    appname,
-                    start_pos,
-                    end_pos,
-                    term,
-                    end_watch,
-                    ws_guard: ws_guard.clone(),
-                    wal_reader,
-                    send_buf: vec![0u8; MAX_SEND_SIZE],
-                };
-
-                Either::Left(sender.run())
-            }
-            PostgresClientProtocol::Interpreted => {
-                let pg_version = tli.tli.get_state().await.1.server.pg_version / 10000;
-                let end_watch_view = end_watch.view();
-                let wal_stream_builder = WalReaderStreamBuilder {
-                    tli: tli.wal_residence_guard().await?,
-                    start_pos,
-                    end_pos,
-                    term,
-                    end_watch,
-                    wal_sender_guard: ws_guard.clone(),
-                };
-
-                let sender = InterpretedWalSender {
-                    pgb,
-                    wal_stream_builder,
-                    end_watch_view,
-                    shard: self.shard.unwrap(),
-                    pg_version,
-                    appname,
-                };
-
-                Either::Right(sender.run())
-            }
+        let mut sender = WalSender {
+            pgb,
+            // should succeed since we're already holding another guard
+            tli: tli.wal_residence_guard().await?,
+            appname,
+            start_pos,
+            end_pos,
+            term,
+            end_watch,
+            ws_guard: ws_guard.clone(),
+            wal_reader,
+            send_buf: vec![0u8; MAX_SEND_SIZE],
         };
-
         let mut reply_reader = ReplyReader {
             reader,
             ws_guard: ws_guard.clone(),
@@ -522,7 +477,7 @@ impl SafekeeperPostgresHandler {
 
         let res = tokio::select! {
             // todo: add read|write .context to these errors
-            r = send_fut => r,
+            r = sender.run() => r,
             r = reply_reader.run() => r,
         };
 
@@ -544,22 +499,16 @@ impl SafekeeperPostgresHandler {
     }
 }
 
-/// TODO(vlad): maybe lift this instead
 /// Walsender streams either up to commit_lsn (normally) or flush_lsn in the
 /// given term (recovery by walproposer or peer safekeeper).
-#[derive(Clone)]
-pub(crate) enum EndWatch {
+enum EndWatch {
     Commit(Receiver<Lsn>),
     Flush(Receiver<TermLsn>),
 }
 
 impl EndWatch {
-    pub(crate) fn view(&self) -> EndWatchView {
-        EndWatchView(self.clone())
-    }
-
     /// Get current end of WAL.
-    pub(crate) fn get(&self) -> Lsn {
+    fn get(&self) -> Lsn {
         match self {
             EndWatch::Commit(r) => *r.borrow(),
             EndWatch::Flush(r) => r.borrow().lsn,
@@ -567,44 +516,15 @@ impl EndWatch {
     }
 
     /// Wait for the update.
-    pub(crate) async fn changed(&mut self) -> anyhow::Result<()> {
+    async fn changed(&mut self) -> anyhow::Result<()> {
         match self {
             EndWatch::Commit(r) => r.changed().await?,
             EndWatch::Flush(r) => r.changed().await?,
         }
         Ok(())
     }
-
-    pub(crate) async fn wait_for_lsn(
-        &mut self,
-        lsn: Lsn,
-        client_term: Option<Term>,
-    ) -> anyhow::Result<Lsn> {
-        loop {
-            let end_pos = self.get();
-            if end_pos > lsn {
-                return Ok(end_pos);
-            }
-            if let EndWatch::Flush(rx) = &self {
-                let curr_term = rx.borrow().term;
-                if let Some(client_term) = client_term {
-                    if curr_term != client_term {
-                        bail!("term changed: requested {}, now {}", client_term, curr_term);
-                    }
-                }
-            }
-            self.changed().await?;
-        }
-    }
 }
 
-pub(crate) struct EndWatchView(EndWatch);
-
-impl EndWatchView {
-    pub(crate) fn get(&self) -> Lsn {
-        self.0.get()
-    }
-}
 /// A half driving sending WAL.
 struct WalSender<'a, IO> {
     pgb: &'a mut PostgresBackend<IO>,
@@ -640,7 +560,7 @@ impl<IO: AsyncRead + AsyncWrite + Unpin> WalSender<'_, IO> {
     ///
     /// Err(CopyStreamHandlerEnd) is always returned; Result is used only for ?
     /// convenience.
-    async fn run(mut self) -> Result<(), CopyStreamHandlerEnd> {
+    async fn run(&mut self) -> Result<(), CopyStreamHandlerEnd> {
         loop {
             // Wait for the next portion if it is not there yet, or just
             // update our end of WAL available for sending value, we

safekeeper/src/state.rs

@@ -4,7 +4,6 @@
 use std::{cmp::max, ops::Deref};
 
 use anyhow::{bail, Result};
-use postgres_ffi::WAL_SEGMENT_SIZE;
 use safekeeper_api::models::TimelineTermBumpResponse;
 use serde::{Deserialize, Serialize};
 use utils::{
@@ -145,7 +144,7 @@ impl TimelinePersistentState {
             ServerInfo {
                 pg_version: 170000, /* Postgres server version (major * 10000) */
                 system_id: 0,       /* Postgres system identifier */
-                wal_seg_size: WAL_SEGMENT_SIZE as u32,
+                wal_seg_size: 16 * 1024 * 1024,
             },
             vec![],
             Lsn::INVALID,

safekeeper/src/timeline_manager.rs

@@ -515,12 +515,7 @@ impl Manager {
             return;
         }
 
-        if state.cfile_last_persist_at.elapsed() > self.conf.control_file_save_interval
-            // If the control file's commit_lsn lags more than one segment behind the current
-            // commit_lsn, flush immediately to limit recovery time in case of a crash. We don't do
-            // this on the WAL ingest hot path since it incurs fsync latency.
-            || state.commit_lsn.saturating_sub(state.cfile_commit_lsn).0 >= self.wal_seg_size as u64
-        {
+        if state.cfile_last_persist_at.elapsed() > self.conf.control_file_save_interval {
             let mut write_guard = self.tli.write_shared_state().await;
             // it should be done in the background because it blocks manager task, but flush() should
             // be fast enough not to be a problem now

safekeeper/src/wal_reader_stream.rs

@@ -1,145 +0,0 @@
-use std::sync::Arc;
-
-use async_stream::try_stream;
-use bytes::Bytes;
-use futures::Stream;
-use postgres_backend::CopyStreamHandlerEnd;
-use std::time::Duration;
-use tokio::time::timeout;
-use utils::lsn::Lsn;
-
-use crate::{
-    safekeeper::Term,
-    send_wal::{EndWatch, WalSenderGuard},
-    timeline::WalResidentTimeline,
-};
-
-pub(crate) struct WalReaderStreamBuilder {
-    pub(crate) tli: WalResidentTimeline,
-    pub(crate) start_pos: Lsn,
-    pub(crate) end_pos: Lsn,
-    pub(crate) term: Option<Term>,
-    pub(crate) end_watch: EndWatch,
-    pub(crate) wal_sender_guard: Arc<WalSenderGuard>,
-}
-
-impl WalReaderStreamBuilder {
-    pub(crate) fn start_pos(&self) -> Lsn {
-        self.start_pos
-    }
-}
-
-pub(crate) struct WalBytes {
-    /// Raw PG WAL
-    pub(crate) wal: Bytes,
-    /// Start LSN of [`Self::wal`]
-    pub(crate) wal_start_lsn: Lsn,
-    /// End LSN of [`Self::wal`]
-    pub(crate) wal_end_lsn: Lsn,
-    /// End LSN of WAL available on the safekeeper
-    pub(crate) commit_lsn: Lsn,
-}
-
-impl WalReaderStreamBuilder {
-    /// Builds a stream of Postgres WAL starting from [`Self::start_pos`].
-    /// The stream terminates when the receiver (pageserver) is fully caught up
-    /// and there's no active computes.
-    pub(crate) async fn build(
-        self,
-        buffer_size: usize,
-    ) -> anyhow::Result<impl Stream<Item = Result<WalBytes, CopyStreamHandlerEnd>>> {
-        // TODO(vlad): The code below duplicates functionality from [`crate::send_wal`].
-        // We can make the raw WAL sender use this stream too and remove the duplication.
-        let Self {
-            tli,
-            mut start_pos,
-            mut end_pos,
-            term,
-            mut end_watch,
-            wal_sender_guard,
-        } = self;
-        let mut wal_reader = tli.get_walreader(start_pos).await?;
-        let mut buffer = vec![0; buffer_size];
-
-        const POLL_STATE_TIMEOUT: Duration = Duration::from_secs(1);
-
-        Ok(try_stream! {
-            loop {
-                let have_something_to_send = end_pos > start_pos;
-
-                if !have_something_to_send {
-                    // wait for lsn
-                    let res = timeout(POLL_STATE_TIMEOUT, end_watch.wait_for_lsn(start_pos, term)).await;
-                    match res {
-                        Ok(ok) => {
-                            end_pos = ok?;
-                        },
-                        Err(_) => {
-                            if let EndWatch::Commit(_) = end_watch {
-                                if let Some(remote_consistent_lsn) = wal_sender_guard
-                                    .walsenders()
-                                    .get_ws_remote_consistent_lsn(wal_sender_guard.id())
-                                {
-                                    if tli.should_walsender_stop(remote_consistent_lsn).await {
-                                        // Terminate if there is nothing more to send.
-                                        // Note that "ending streaming" part of the string is used by
-                                        // pageserver to identify WalReceiverError::SuccessfulCompletion,
-                                        // do not change this string without updating pageserver.
-                                        return;
-                                    }
-                                }
-                            }
-
-                            continue;
-                        }
-                    }
-                }
-
-
-                assert!(
-                    end_pos > start_pos,
-                    "nothing to send after waiting for WAL"
-                );
-
-                // try to send as much as available, capped by the buffer size
-                let mut chunk_end_pos = start_pos + buffer_size as u64;
-                // if we went behind available WAL, back off
-                if chunk_end_pos >= end_pos {
-                    chunk_end_pos = end_pos;
-                } else {
-                    // If sending not up to end pos, round down to page boundary to
-                    // avoid breaking WAL record not at page boundary, as protocol
-                    // demands. See walsender.c (XLogSendPhysical).
-                    chunk_end_pos = chunk_end_pos
-                        .checked_sub(chunk_end_pos.block_offset())
-                        .unwrap();
-                }
-                let send_size = (chunk_end_pos.0 - start_pos.0) as usize;
-                let buffer = &mut buffer[..send_size];
-                let send_size: usize;
-                {
-                    // If uncommitted part is being pulled, check that the term is
-                    // still the expected one.
-                    let _term_guard = if let Some(t) = term {
-                        Some(tli.acquire_term(t).await?)
-                    } else {
-                        None
-                    };
-                    // Read WAL into buffer. send_size can be additionally capped to
-                    // segment boundary here.
-                    send_size = wal_reader.read(buffer).await?
-                };
-                let wal = Bytes::copy_from_slice(&buffer[..send_size]);
-
-                yield WalBytes {
-                    wal,
-                    wal_start_lsn: start_pos,
-                    wal_end_lsn: start_pos + send_size as u64,
-                    commit_lsn: end_pos
-                };
-
-                start_pos += send_size as u64;
-            }
-        })
-    }
-}

safekeeper/src/wal_storage.rs

@@ -113,13 +113,6 @@ pub struct PhysicalStorage {
     /// non-aligned chunks of data.
     write_record_lsn: Lsn,
 
-    /// The last LSN flushed to disk. May be in the middle of a record.
-    ///
-    /// NB: when the rest of the system refers to `flush_lsn`, it usually
-    /// actually refers to `flush_record_lsn`. This ambiguity can be dangerous
-    /// and should be resolved.
-    flush_lsn: Lsn,
-
     /// The LSN of the last WAL record flushed to disk.
     flush_record_lsn: Lsn,
 
@@ -134,29 +127,23 @@ pub struct PhysicalStorage {
     /// - doesn't point to the end of the segment
     file: Option<File>,
 
-    /// When true, WAL truncation potentially has been interrupted and we need
-    /// to finish it before allowing WAL writes; see truncate_wal for details.
-    /// In this case [`write_lsn`] can be less than actually written WAL on
-    /// disk. In particular, there can be a case with unexpected .partial file.
+    /// When false, we have just initialized storage using the LSN from find_end_of_wal().
+    /// In this case, [`write_lsn`] can be less than actually written WAL on disk. In particular,
+    /// there can be a case with unexpected .partial file.
     ///
     /// Imagine the following:
     /// - 000000010000000000000001
-    ///   - it was fully written, but the last record is split between 2
-    ///     segments
-    ///   - after restart, `find_end_of_wal()` returned 0/1FFFFF0, which is in
-    ///     the end of this segment
-    ///   - `write_lsn`, `write_record_lsn` and `flush_record_lsn` were
-    ///     initialized to 0/1FFFFF0
+    ///   - it was fully written, but the last record is split between 2 segments
+    ///   - after restart, `find_end_of_wal()` returned 0/1FFFFF0, which is in the end of this segment
+    ///   - `write_lsn`, `write_record_lsn` and `flush_record_lsn` were initialized to 0/1FFFFF0
     /// - 000000010000000000000002.partial
-    ///   - it has only 1 byte written, which is not enough to make a full WAL
-    ///     record
+    ///   - it has only 1 byte written, which is not enough to make a full WAL record
     ///
-    /// Partial segment 002 has no WAL records, and it will be removed by the
-    /// next truncate_wal(). This flag will be set to true after the first
-    /// truncate_wal() call.
+    /// Partial segment 002 has no WAL records, and it will be removed by the next truncate_wal().
+    /// This flag will be set to true after the first truncate_wal() call.
     ///
     /// [`write_lsn`]: Self::write_lsn
-    pending_wal_truncation: bool,
+    is_truncated_after_restart: bool,
 }
 
 impl PhysicalStorage {
@@ -218,11 +205,10 @@ impl PhysicalStorage {
             system_id: state.server.system_id,
             write_lsn,
             write_record_lsn: write_lsn,
-            flush_lsn,
             flush_record_lsn: flush_lsn,
             decoder: WalStreamDecoder::new(write_lsn, state.server.pg_version / 10000),
             file: None,
-            pending_wal_truncation: true,
+            is_truncated_after_restart: false,
         })
     }
 
@@ -303,9 +289,8 @@ impl PhysicalStorage {
         }
     }
 
-    /// Write WAL bytes, which are known to be located in a single WAL segment. Returns true if the
-    /// segment was completed, closed, and flushed to disk.
-    async fn write_in_segment(&mut self, segno: u64, xlogoff: usize, buf: &[u8]) -> Result<bool> {
+    /// Write WAL bytes, which are known to be located in a single WAL segment.
+    async fn write_in_segment(&mut self, segno: u64, xlogoff: usize, buf: &[u8]) -> Result<()> {
         let mut file = if let Some(file) = self.file.take() {
             file
         } else {
@@ -329,24 +314,20 @@ impl PhysicalStorage {
             let (wal_file_path, wal_file_partial_path) =
                 wal_file_paths(&self.timeline_dir, segno, self.wal_seg_size);
             fs::rename(wal_file_partial_path, wal_file_path).await?;
-            Ok(true)
         } else {
             // otherwise, file can be reused later
             self.file = Some(file);
-            Ok(false)
         }
+
+        Ok(())
     }
 
     /// Writes WAL to the segment files, until everything is writed. If some segments
     /// are fully written, they are flushed to disk. The last (partial) segment can
     /// be flushed separately later.
     ///
-    /// Updates `write_lsn` and `flush_lsn`.
+    /// Updates `write_lsn`.
     async fn write_exact(&mut self, pos: Lsn, mut buf: &[u8]) -> Result<()> {
-        // TODO: this shouldn't be possible, except possibly with write_lsn == 0.
-        // Rename this method to `append_exact`, and make it append-only, removing
-        // the `pos` parameter and this check. For this reason, we don't update
-        // `flush_lsn` here.
         if self.write_lsn != pos {
             // need to flush the file before discarding it
             if let Some(file) = self.file.take() {
@@ -368,13 +349,9 @@ impl PhysicalStorage {
                 buf.len()
             };
 
-            let flushed = self
-                .write_in_segment(segno, xlogoff, &buf[..bytes_write])
+            self.write_in_segment(segno, xlogoff, &buf[..bytes_write])
                 .await?;
             self.write_lsn += bytes_write as u64;
-            if flushed {
-                self.flush_lsn = self.write_lsn;
-            }
             buf = &buf[bytes_write..];
         }
 
@@ -388,9 +365,6 @@ impl Storage for PhysicalStorage {
         self.write_lsn
     }
     /// flush_lsn returns LSN of last durably stored WAL record.
-    ///
-    /// TODO: flush_lsn() returns flush_record_lsn, but write_lsn() returns write_lsn: confusing.
-    #[allow(clippy::misnamed_getters)]
     fn flush_lsn(&self) -> Lsn {
         self.flush_record_lsn
     }
@@ -431,22 +405,14 @@ impl Storage for PhysicalStorage {
                 startpos
             );
         }
-        if self.pending_wal_truncation {
-            bail!(
-                "write_wal called with pending WAL truncation, write_lsn={}, startpos={}",
-                self.write_lsn,
-                startpos
-            );
-        }
 
         let write_seconds = time_io_closure(self.write_exact(startpos, buf)).await?;
         // WAL is written, updating write metrics
         self.metrics.observe_write_seconds(write_seconds);
         self.metrics.observe_write_bytes(buf.len());
 
-        // Figure out the last record's end LSN and update `write_record_lsn`
-        // (if we got a whole record). The write may also have closed and
-        // flushed a segment, so update `flush_record_lsn` as well.
+        // figure out last record's end lsn for reporting (if we got the
+        // whole record)
         if self.decoder.available() != startpos {
             info!(
                 "restart decoder from {} to {}",
@@ -457,15 +423,12 @@ impl Storage for PhysicalStorage {
             self.decoder = WalStreamDecoder::new(startpos, pg_version);
         }
         self.decoder.feed_bytes(buf);
-
-        if self.write_record_lsn <= self.flush_lsn {
-            // We may have flushed a previously written record.
-            self.flush_record_lsn = self.write_record_lsn;
-        }
-        while let Some((lsn, _rec)) = self.decoder.poll_decode()? {
-            self.write_record_lsn = lsn;
-            if lsn <= self.flush_lsn {
-                self.flush_record_lsn = lsn;
+        loop {
+            match self.decoder.poll_decode()? {
+                None => break, // no full record yet
+                Some((lsn, _rec)) => {
+                    self.write_record_lsn = lsn;
+                }
             }
         }
 
@@ -482,17 +445,19 @@ impl Storage for PhysicalStorage {
             self.fdatasync_file(&unflushed_file).await?;
             self.file = Some(unflushed_file);
         } else {
-            // We have unflushed data (write_lsn != flush_lsn), but no file. This
-            // shouldn't happen, since the segment is flushed on close.
-            bail!(
-                "unexpected unflushed data with no open file, write_lsn={}, flush_lsn={}",
-                self.write_lsn,
-                self.flush_record_lsn
-            );
+            // We have unflushed data (write_lsn != flush_lsn), but no file.
+            // This should only happen if last file was fully written and flushed,
+            // but haven't updated flush_lsn yet.
+            if self.write_lsn.segment_offset(self.wal_seg_size) != 0 {
+                bail!(
+                    "unexpected unflushed data with no open file, write_lsn={}, flush_lsn={}",
+                    self.write_lsn,
+                    self.flush_record_lsn
+                );
+            }
         }
 
         // everything is flushed now, let's update flush_lsn
-        self.flush_lsn = self.write_lsn;
         self.flush_record_lsn = self.write_record_lsn;
         Ok(())
     }
@@ -514,35 +479,15 @@ impl Storage for PhysicalStorage {
             );
         }
 
-        // Quick exit if nothing to do and we know that the state is clean to
-        // avoid writing up to 16 MiB of zeros on disk (this happens on each
-        // connect).
-        if !self.pending_wal_truncation
+        // Quick exit if nothing to do to avoid writing up to 16 MiB of zeros on
+        // disk (this happens on each connect).
+        if self.is_truncated_after_restart
             && end_pos == self.write_lsn
             && end_pos == self.flush_record_lsn
         {
             return Ok(());
         }
 
-        // Atomicity: we start with LSNs reset because once on disk deletion is
-        // started it can't be reversed. However, we might crash/error in the
-        // middle, leaving garbage above the truncation point. In theory,
-        // concatenated with previous records it might form bogus WAL (though
-        // very unlikely in practice because CRC would guard from that). To
-        // protect, set pending_wal_truncation flag before beginning: it means
-        // truncation must be retried and WAL writes are prohibited until it
-        // succeeds. Flag is also set on boot because we don't know if the last
-        // state was clean.
-        //
-        // Protocol (HandleElected before first AppendRequest) ensures we'll
-        // always try to ensure clean truncation before any writes.
-        self.pending_wal_truncation = true;
-
-        self.write_lsn = end_pos;
-        self.flush_lsn = end_pos;
-        self.write_record_lsn = end_pos;
-        self.flush_record_lsn = end_pos;
-
         // Close previously opened file, if any
         if let Some(unflushed_file) = self.file.take() {
             self.fdatasync_file(&unflushed_file).await?;
@@ -568,7 +513,11 @@ impl Storage for PhysicalStorage {
             fs::rename(wal_file_path, wal_file_partial_path).await?;
         }
 
-        self.pending_wal_truncation = false;
+        // Update LSNs
+        self.write_lsn = end_pos;
+        self.write_record_lsn = end_pos;
+        self.flush_record_lsn = end_pos;
+        self.is_truncated_after_restart = true;
         Ok(())
     }
 

storage_controller/src/service.rs

@@ -3642,7 +3642,6 @@ impl Service {
                 match res {
                     Ok(ok) => Ok(ok),
                     Err(mgmt_api::Error::ApiError(StatusCode::CONFLICT, _)) => Ok(StatusCode::CONFLICT),
-                    Err(mgmt_api::Error::ApiError(StatusCode::SERVICE_UNAVAILABLE, msg)) => Err(ApiError::ResourceUnavailable(msg.into())),
                     Err(e) => {
                         Err(
                             ApiError::InternalServerError(anyhow::anyhow!(
@@ -6356,19 +6355,6 @@ impl Service {
 
         // Pick the biggest tenant to split first
         top_n.sort_by_key(|i| i.resident_size);
-
-        // Filter out tenants in a prohibiting scheduling mode
-        {
-            let locked = self.inner.read().unwrap();
-            top_n.retain(|i| {
-                if let Some(shard) = locked.tenants.get(&i.id) {
-                    matches!(shard.get_scheduling_policy(), ShardSchedulingPolicy::Active)
-                } else {
-                    false
-                }
-            });
-        }
-
         let Some(split_candidate) = top_n.into_iter().next() else {
             tracing::debug!("No split-elegible shards found");
             return;

test_runner/fixtures/endpoint/http.py

@@ -46,8 +46,3 @@ class EndpointHttpClient(requests.Session):
         )
         res.raise_for_status()
         return res.json()
-
-    def metrics(self) -> str:
-        res = self.get(f"http://localhost:{self.port}/metrics")
-        res.raise_for_status()
-        return res.text

test_runner/fixtures/neon_api.py

@@ -5,8 +5,6 @@ from typing import TYPE_CHECKING, cast, final
 
 import requests
 
-from fixtures.log_helper import log
-
 if TYPE_CHECKING:
     from typing import Any, Literal, Optional
 
@@ -32,11 +30,7 @@ class NeonAPI:
             kwargs["headers"] = {}
         kwargs["headers"]["Authorization"] = f"Bearer {self.__neon_api_key}"
 
-        resp = requests.request(method, f"{self.__neon_api_base_url}{endpoint}", **kwargs)
-        log.debug("%s %s returned a %d: %s", method, endpoint, resp.status_code, resp.text)
-        resp.raise_for_status()
-
-        return resp
+        return requests.request(method, f"{self.__neon_api_base_url}{endpoint}", **kwargs)
 
     def create_project(
         self,
@@ -72,6 +66,8 @@ class NeonAPI:
             json=data,
         )
 
+        assert resp.status_code == 201
+
         return cast("dict[str, Any]", resp.json())
 
     def get_project_details(self, project_id: str) -> dict[str, Any]:
@@ -83,7 +79,7 @@ class NeonAPI:
                 "Content-Type": "application/json",
             },
         )
-
+        assert resp.status_code == 200
         return cast("dict[str, Any]", resp.json())
 
     def delete_project(
@@ -99,6 +95,8 @@ class NeonAPI:
             },
         )
 
+        assert resp.status_code == 200
+
         return cast("dict[str, Any]", resp.json())
 
     def start_endpoint(
@@ -114,6 +112,8 @@ class NeonAPI:
             },
         )
 
+        assert resp.status_code == 200
+
         return cast("dict[str, Any]", resp.json())
 
     def suspend_endpoint(
@@ -129,6 +129,8 @@ class NeonAPI:
             },
         )
 
+        assert resp.status_code == 200
+
         return cast("dict[str, Any]", resp.json())
 
     def restart_endpoint(
@@ -144,6 +146,8 @@ class NeonAPI:
             },
         )
 
+        assert resp.status_code == 200
+
         return cast("dict[str, Any]", resp.json())
 
     def create_endpoint(
@@ -174,6 +178,8 @@ class NeonAPI:
             json=data,
         )
 
+        assert resp.status_code == 201
+
         return cast("dict[str, Any]", resp.json())
 
     def get_connection_uri(
@@ -200,6 +206,8 @@ class NeonAPI:
             },
         )
 
+        assert resp.status_code == 200
+
         return cast("dict[str, Any]", resp.json())
 
     def get_branches(self, project_id: str) -> dict[str, Any]:
@@ -211,6 +219,8 @@ class NeonAPI:
             },
         )
 
+        assert resp.status_code == 200
+
         return cast("dict[str, Any]", resp.json())
 
     def get_endpoints(self, project_id: str) -> dict[str, Any]:
@@ -222,6 +232,8 @@ class NeonAPI:
             },
         )
 
+        assert resp.status_code == 200
+
         return cast("dict[str, Any]", resp.json())
 
     def get_operations(self, project_id: str) -> dict[str, Any]:
@@ -234,6 +246,8 @@ class NeonAPI:
             },
         )
 
+        assert resp.status_code == 200
+
         return cast("dict[str, Any]", resp.json())
 
     def wait_for_operation_to_finish(self, project_id: str):

test_runner/fixtures/neon_fixtures.py

@@ -1065,9 +1065,6 @@ class NeonEnv:
                 "http_auth_type": http_auth_type,
                 # Default which can be overriden with `NeonEnvBuilder.pageserver_config_override`
                 "availability_zone": "us-east-2a",
-                # Disable pageserver disk syncs in tests: when running tests concurrently, this avoids
-                # the pageserver taking a long time to start up due to syncfs flushing other tests' data
-                "no_sync": True,
             }
             if self.pageserver_virtual_file_io_engine is not None:
                 ps_cfg["virtual_file_io_engine"] = self.pageserver_virtual_file_io_engine
@@ -2379,17 +2376,6 @@ class NeonPageserver(PgProtocol, LogUtils):
         #
         # The entries in the list are regular experessions.
         self.allowed_errors: list[str] = list(DEFAULT_PAGESERVER_ALLOWED_ERRORS)
-        # Store persistent failpoints that should be reapplied on each start
-        self._persistent_failpoints: dict[str, str] = {}
-
-    def add_persistent_failpoint(self, name: str, action: str):
-        """
-        Add a failpoint that will be automatically reapplied each time the pageserver starts.
-        The failpoint will be set immediately if the pageserver is running.
-        """
-        self._persistent_failpoints[name] = action
-        if self.running:
-            self.http_client().configure_failpoints([(name, action)])
 
     def timeline_dir(
         self,
@@ -2457,15 +2443,6 @@ class NeonPageserver(PgProtocol, LogUtils):
         """
         assert self.running is False
 
-        if self._persistent_failpoints:
-            # Tests shouldn't use this mechanism _and_ set FAILPOINTS explicitly
-            assert extra_env_vars is None or "FAILPOINTS" not in extra_env_vars
-            if extra_env_vars is None:
-                extra_env_vars = {}
-            extra_env_vars["FAILPOINTS"] = ",".join(
-                f"{k}={v}" for (k, v) in self._persistent_failpoints.items()
-            )
-
         storage = self.env.pageserver_remote_storage
         if isinstance(storage, S3Storage):
             s3_env_vars = storage.access_env_vars()
@@ -4542,7 +4519,7 @@ def pytest_addoption(parser: Parser):
 
 
 SMALL_DB_FILE_NAME_REGEX: re.Pattern[str] = re.compile(
-    r"config-v1|heatmap-v1|tenant-manifest|metadata|.+\.(?:toml|pid|json|sql|conf)"
+    r"config-v1|heatmap-v1|metadata|.+\.(?:toml|pid|json|sql|conf)"
 )
 
 

test_runner/performance/test_logical_replication.py

@@ -1,8 +1,6 @@
 from __future__ import annotations
 
 import time
-from collections.abc import Iterator
-from contextlib import contextmanager
 from typing import TYPE_CHECKING, cast
 
 import psycopg2
@@ -20,7 +18,7 @@ if TYPE_CHECKING:
     from fixtures.benchmark_fixture import NeonBenchmarker
     from fixtures.neon_api import NeonApiEndpoint
     from fixtures.neon_fixtures import NeonEnv, PgBin, VanillaPostgres
-    from psycopg2.extensions import connection, cursor
+    from psycopg2.extensions import cursor
 
 
 @pytest.mark.timeout(1000)
@@ -151,16 +149,12 @@ def test_subscriber_lag(
                 check_pgbench_still_running(pub_workload, "pub")
                 check_pgbench_still_running(sub_workload, "sub")
 
-                pub_conn = psycopg2.connect(pub_connstr)
-                sub_conn = psycopg2.connect(sub_connstr)
-                pub_conn.autocommit = True
-                sub_conn.autocommit = True
-
-                with pub_conn.cursor() as pub_cur, sub_conn.cursor() as sub_cur:
-                    lag = measure_logical_replication_lag(sub_cur, pub_cur)
-
-                pub_conn.close()
-                sub_conn.close()
+                with (
+                    psycopg2.connect(pub_connstr) as pub_conn,
+                    psycopg2.connect(sub_connstr) as sub_conn,
+                ):
+                    with pub_conn.cursor() as pub_cur, sub_conn.cursor() as sub_cur:
+                        lag = measure_logical_replication_lag(sub_cur, pub_cur)
 
                 log.info(f"Replica lagged behind master by {lag} seconds")
                 zenbenchmark.record("replica_lag", lag, "s", MetricReport.LOWER_IS_BETTER)
@@ -212,7 +206,6 @@ def test_publisher_restart(
     sub_conn = psycopg2.connect(sub_connstr)
     pub_conn.autocommit = True
     sub_conn.autocommit = True
-
     with pub_conn.cursor() as pub_cur, sub_conn.cursor() as sub_cur:
         pub_cur.execute("SELECT 1 FROM pg_catalog.pg_publication WHERE pubname = 'pub1'")
         pub_exists = len(pub_cur.fetchall()) != 0
@@ -229,7 +222,6 @@ def test_publisher_restart(
             sub_cur.execute(f"create subscription sub1 connection '{pub_connstr}' publication pub1")
 
         initial_sync_lag = measure_logical_replication_lag(sub_cur, pub_cur)
-
     pub_conn.close()
     sub_conn.close()
 
@@ -256,17 +248,12 @@ def test_publisher_restart(
                     ["pgbench", "-c10", pgbench_duration, "-Mprepared"],
                     env=pub_env,
                 )
-
-                pub_conn = psycopg2.connect(pub_connstr)
-                sub_conn = psycopg2.connect(sub_connstr)
-                pub_conn.autocommit = True
-                sub_conn.autocommit = True
-
-                with pub_conn.cursor() as pub_cur, sub_conn.cursor() as sub_cur:
-                    lag = measure_logical_replication_lag(sub_cur, pub_cur)
-
-                pub_conn.close()
-                sub_conn.close()
+                with (
+                    psycopg2.connect(pub_connstr) as pub_conn,
+                    psycopg2.connect(sub_connstr) as sub_conn,
+                ):
+                    with pub_conn.cursor() as pub_cur, sub_conn.cursor() as sub_cur:
+                        lag = measure_logical_replication_lag(sub_cur, pub_cur)
 
                 log.info(f"Replica lagged behind master by {lag} seconds")
                 zenbenchmark.record("replica_lag", lag, "s", MetricReport.LOWER_IS_BETTER)
@@ -294,48 +281,6 @@ def test_snap_files(
     then runs pgbench inserts while generating large numbers of snapfiles. Then restarts
     the node and tries to peek the replication changes.
     """
-
-    @contextmanager
-    def replication_slot(conn: connection, slot_name: str) -> Iterator[None]:
-        """
-        Make sure that the replication slot doesn't outlive the test. Normally
-        we wouldn't want this behavior, but since the test creates and drops
-        the replication slot, we do.
-
-        We've had problems in the past where this slot sticking around caused
-        issues with the publisher retaining WAL during the execution of the
-        other benchmarks in this suite.
-        """
-
-        def __drop_replication_slot(c: cursor) -> None:
-            c.execute(
-                """
-                DO $$
-                BEGIN
-                    IF EXISTS (
-                        SELECT 1
-                        FROM pg_replication_slots
-                        WHERE slot_name = %(slot_name)s
-                    ) THEN
-                        PERFORM pg_drop_replication_slot(%(slot_name)s);
-                    END IF;
-                END $$;
-                """,
-                {"slot_name": slot_name},
-            )
-
-        with conn.cursor() as c:
-            __drop_replication_slot(c)
-            c.execute(
-                "SELECT pg_create_logical_replication_slot(%(slot_name)s, 'test_decoding')",
-                {"slot_name": slot_name},
-            )
-
-        yield
-
-        with conn.cursor() as c:
-            __drop_replication_slot(c)
-
     test_duration_min = 60
     test_interval_min = 5
     pgbench_duration = f"-T{test_duration_min * 60 * 2}"
@@ -343,32 +288,50 @@ def test_snap_files(
     env = benchmark_project_pub.pgbench_env
     connstr = benchmark_project_pub.connstr
 
-    conn = psycopg2.connect(connstr)
-    conn.autocommit = True
-
-    with conn.cursor() as cur:
-        cur.execute("SELECT rolsuper FROM pg_roles WHERE rolname = 'neondb_owner'")
-        is_super = cast("bool", cur.fetchall()[0][0])
-        assert is_super, "This benchmark won't work if we don't have superuser"
-
-    conn.close()
+    with psycopg2.connect(connstr) as conn:
+        conn.autocommit = True
+        with conn.cursor() as cur:
+            cur.execute("SELECT rolsuper FROM pg_roles WHERE rolname = 'neondb_owner'")
+            is_super = cast("bool", cur.fetchall()[0][0])
+            assert is_super, "This benchmark won't work if we don't have superuser"
 
     pg_bin.run_capture(["pgbench", "-i", "-I", "dtGvp", "-s100"], env=env)
 
     conn = psycopg2.connect(connstr)
     conn.autocommit = True
+    cur = conn.cursor()
+    cur.execute("ALTER SYSTEM SET neon.logical_replication_max_snap_files = -1")
 
-    with replication_slot(conn, "slotter"):
-        workload = pg_bin.run_nonblocking(
-            ["pgbench", "-c10", pgbench_duration, "-Mprepared"], env=env
-        )
-        try:
-            start = time.time()
-            prev_measurement = time.time()
-            while time.time() - start < test_duration_min * 60:
-                conn = psycopg2.connect(connstr)
-                conn.autocommit = True
+    with psycopg2.connect(connstr) as conn:
+        conn.autocommit = True
+        with conn.cursor() as cur:
+            cur.execute("SELECT pg_reload_conf()")
 
+    with psycopg2.connect(connstr) as conn:
+        conn.autocommit = True
+        with conn.cursor() as cur:
+            cur.execute(
+                """
+                DO $$
+                    BEGIN
+                    IF EXISTS (
+                        SELECT 1
+                        FROM pg_replication_slots
+                        WHERE slot_name = 'slotter'
+                    ) THEN
+                        PERFORM pg_drop_replication_slot('slotter');
+                    END IF;
+                END $$;
+            """
+            )
+            cur.execute("SELECT pg_create_logical_replication_slot('slotter', 'test_decoding')")
+
+    workload = pg_bin.run_nonblocking(["pgbench", "-c10", pgbench_duration, "-Mprepared"], env=env)
+    try:
+        start = time.time()
+        prev_measurement = time.time()
+        while time.time() - start < test_duration_min * 60:
+            with psycopg2.connect(connstr) as conn:
                 with conn.cursor() as cur:
                     cur.execute(
                         "SELECT count(*) FROM (SELECT pg_log_standby_snapshot() FROM generate_series(1, 10000) g) s"
@@ -378,15 +341,12 @@ def test_snap_files(
                         "SELECT pg_replication_slot_advance('slotter', pg_current_wal_lsn())"
                     )
 
-                conn.close()
+            # Measure storage
+            if time.time() - prev_measurement > test_interval_min * 60:
+                storage = benchmark_project_pub.get_synthetic_storage_size()
+                zenbenchmark.record("storage", storage, "B", MetricReport.LOWER_IS_BETTER)
+                prev_measurement = time.time()
+            time.sleep(test_interval_min * 60 / 3)
 
-                # Measure storage
-                if time.time() - prev_measurement > test_interval_min * 60:
-                    storage = benchmark_project_pub.get_synthetic_storage_size()
-                    zenbenchmark.record("storage", storage, "B", MetricReport.LOWER_IS_BETTER)
-                    prev_measurement = time.time()
-                time.sleep(test_interval_min * 60 / 3)
-        finally:
-            workload.terminate()
-
-    conn.close()
+    finally:
+        workload.terminate()

test_runner/performance/test_physical_replication.py

@@ -102,21 +102,15 @@ def test_ro_replica_lag(
                     check_pgbench_still_running(master_workload)
                     check_pgbench_still_running(replica_workload)
                     time.sleep(sync_interval_min * 60)
-
-                    conn_master = psycopg2.connect(master_connstr)
-                    conn_replica = psycopg2.connect(replica_connstr)
-                    conn_master.autocommit = True
-                    conn_replica.autocommit = True
-
                     with (
-                        conn_master.cursor() as cur_master,
-                        conn_replica.cursor() as cur_replica,
+                        psycopg2.connect(master_connstr) as conn_master,
+                        psycopg2.connect(replica_connstr) as conn_replica,
                     ):
-                        lag = measure_replication_lag(cur_master, cur_replica)
-
-                    conn_master.close()
-                    conn_replica.close()
-
+                        with (
+                            conn_master.cursor() as cur_master,
+                            conn_replica.cursor() as cur_replica,
+                        ):
+                            lag = measure_replication_lag(cur_master, cur_replica)
                     log.info(f"Replica lagged behind master by {lag} seconds")
                     zenbenchmark.record("replica_lag", lag, "s", MetricReport.LOWER_IS_BETTER)
             finally:
@@ -225,15 +219,11 @@ def test_replication_start_stop(
         pg_bin.run_capture(["pgbench", "-i", "-I", "dtGvp", "-s10"], env=master_env)
 
         # Sync replicas
-        conn_master = psycopg2.connect(master_connstr)
-        conn_master.autocommit = True
-
-        with conn_master.cursor() as cur_master:
-            for i in range(num_replicas):
-                conn_replica = psycopg2.connect(replica_connstr[i])
-                measure_replication_lag(cur_master, conn_replica.cursor())
-
-        conn_master.close()
+        with psycopg2.connect(master_connstr) as conn_master:
+            with conn_master.cursor() as cur_master:
+                for i in range(num_replicas):
+                    conn_replica = psycopg2.connect(replica_connstr[i])
+                    measure_replication_lag(cur_master, conn_replica.cursor())
 
         master_pgbench = pg_bin.run_nonblocking(
             [
@@ -287,22 +277,17 @@ def test_replication_start_stop(
 
             time.sleep(configuration_test_time_sec)
 
-            conn_master = psycopg2.connect(master_connstr)
-            conn_master.autocommit = True
-
-            with conn_master.cursor() as cur_master:
-                for ireplica in range(num_replicas):
-                    replica_conn = psycopg2.connect(replica_connstr[ireplica])
-                    lag = measure_replication_lag(cur_master, replica_conn.cursor())
-                    zenbenchmark.record(
-                        f"Replica {ireplica} lag", lag, "s", MetricReport.LOWER_IS_BETTER
-                    )
-                    log.info(
-                        f"Replica {ireplica} lagging behind master by {lag} seconds after configuration {iconfig:>b}"
-                    )
-
-            conn_master.close()
-
+            with psycopg2.connect(master_connstr) as conn_master:
+                with conn_master.cursor() as cur_master:
+                    for ireplica in range(num_replicas):
+                        replica_conn = psycopg2.connect(replica_connstr[ireplica])
+                        lag = measure_replication_lag(cur_master, replica_conn.cursor())
+                        zenbenchmark.record(
+                            f"Replica {ireplica} lag", lag, "s", MetricReport.LOWER_IS_BETTER
+                        )
+                        log.info(
+                            f"Replica {ireplica} lagging behind master by {lag} seconds after configuration {iconfig:>b}"
+                        )
         master_pgbench.terminate()
     except Exception as e:
         error_occurred = True

test_runner/performance/test_sharded_ingest.py

@@ -15,21 +15,16 @@ from fixtures.neon_fixtures import (
 
 @pytest.mark.timeout(600)
 @pytest.mark.parametrize("shard_count", [1, 8, 32])
-@pytest.mark.parametrize("wal_receiver_protocol", ["vanilla", "interpreted"])
 def test_sharded_ingest(
     neon_env_builder: NeonEnvBuilder,
     zenbenchmark: NeonBenchmarker,
     shard_count: int,
-    wal_receiver_protocol: str,
 ):
     """
     Benchmarks sharded ingestion throughput, by ingesting a large amount of WAL into a Safekeeper
     and fanning out to a large number of shards on dedicated Pageservers. Comparing the base case
     (shard_count=1) to the sharded case indicates the overhead of sharding.
     """
-    neon_env_builder.pageserver_config_override = (
-        f"wal_receiver_protocol = '{wal_receiver_protocol}'"
-    )
 
     ROW_COUNT = 100_000_000  # about 7 GB of WAL
 
@@ -55,6 +50,7 @@ def test_sharded_ingest(
     # Start the endpoint.
     endpoint = env.endpoints.create_start("main", tenant_id=tenant_id)
     start_lsn = Lsn(endpoint.safe_psql("select pg_current_wal_lsn()")[0][0])
+
     # Ingest data and measure WAL volume and duration.
     with closing(endpoint.connect()) as conn:
         with conn.cursor() as cur:
@@ -72,48 +68,4 @@ def test_sharded_ingest(
     wal_written_mb = round((end_lsn - start_lsn) / (1024 * 1024))
     zenbenchmark.record("wal_written", wal_written_mb, "MB", MetricReport.TEST_PARAM)
 
-    total_ingested = 0
-    total_records_received = 0
-    ingested_by_ps = []
-    for pageserver in env.pageservers:
-        ingested = pageserver.http_client().get_metric_value(
-            "pageserver_wal_ingest_bytes_received_total"
-        )
-        records_received = pageserver.http_client().get_metric_value(
-            "pageserver_wal_ingest_records_received_total"
-        )
-
-        if ingested is None:
-            ingested = 0
-
-        if records_received is None:
-            records_received = 0
-
-        ingested_by_ps.append(
-            (
-                pageserver.id,
-                {
-                    "ingested": ingested,
-                    "records_received": records_received,
-                },
-            )
-        )
-
-        total_ingested += int(ingested)
-        total_records_received += int(records_received)
-
-    total_ingested_mb = total_ingested / (1024 * 1024)
-    zenbenchmark.record("wal_ingested", total_ingested_mb, "MB", MetricReport.LOWER_IS_BETTER)
-    zenbenchmark.record(
-        "records_received", total_records_received, "records", MetricReport.LOWER_IS_BETTER
-    )
-
-    ingested_by_ps.sort(key=lambda x: x[0])
-    for _, stats in ingested_by_ps:
-        for k in stats:
-            if k != "records_received":
-                stats[k] /= 1024**2
-
-    log.info(f"WAL ingested by each pageserver {ingested_by_ps}")
-
     assert tenant_get_shards(env, tenant_id) == shards, "shards moved"

test_runner/regress/test_installed_extensions.py

@@ -1,14 +1,6 @@
-from __future__ import annotations
-
-import time
 from logging import info
-from typing import TYPE_CHECKING
 
-from fixtures.log_helper import log
-from fixtures.metrics import parse_metrics
-
-if TYPE_CHECKING:
-    from fixtures.neon_fixtures import NeonEnv
+from fixtures.neon_fixtures import NeonEnv
 
 
 def test_installed_extensions(neon_simple_env: NeonEnv):
@@ -93,52 +85,3 @@ def test_installed_extensions(neon_simple_env: NeonEnv):
             assert ext["n_databases"] == 2
             ext["versions"].sort()
             assert ext["versions"] == ["1.2", "1.3"]
-
-    # check that /metrics endpoint is available
-    # ensure that we see the metric before and after restart
-    res = client.metrics()
-    info("Metrics: %s", res)
-    m = parse_metrics(res)
-    neon_m = m.query_all("installed_extensions", {"extension_name": "neon", "version": "1.2"})
-    assert len(neon_m) == 1
-    for sample in neon_m:
-        assert sample.value == 2
-    neon_m = m.query_all("installed_extensions", {"extension_name": "neon", "version": "1.3"})
-    assert len(neon_m) == 1
-    for sample in neon_m:
-        assert sample.value == 1
-
-    endpoint.stop()
-    endpoint.start()
-
-    timeout = 10
-    while timeout > 0:
-        try:
-            res = client.metrics()
-            timeout = -1
-            if len(parse_metrics(res).query_all("installed_extensions")) < 4:
-                # Assume that not all metrics that are collected yet
-                time.sleep(1)
-                timeout -= 1
-                continue
-        except Exception:
-            log.exception("failed to get metrics, assume they are not collected yet")
-            time.sleep(1)
-            timeout -= 1
-            continue
-
-        assert (
-            len(parse_metrics(res).query_all("installed_extensions")) >= 4
-        ), "Not all metrics are collected"
-
-        info("After restart metrics: %s", res)
-        m = parse_metrics(res)
-        neon_m = m.query_all("installed_extensions", {"extension_name": "neon", "version": "1.2"})
-        assert len(neon_m) == 1
-        for sample in neon_m:
-            assert sample.value == 1
-
-        neon_m = m.query_all("installed_extensions", {"extension_name": "neon", "version": "1.3"})
-        assert len(neon_m) == 1
-        for sample in neon_m:
-            assert sample.value == 1

test_runner/regress/test_pageserver_generations.py

@@ -35,10 +35,9 @@ from fixtures.pageserver.utils import (
     wait_for_upload,
 )
 from fixtures.remote_storage import (
-    LocalFsStorage,
     RemoteStorageKind,
 )
-from fixtures.utils import run_only_on_default_postgres, wait_until
+from fixtures.utils import wait_until
 from fixtures.workload import Workload
 
 if TYPE_CHECKING:
@@ -729,68 +728,3 @@ def test_upgrade_generationless_local_file_paths(
     )
     # We should download into the same local path we started with
     assert os.path.exists(victim_path)
-
-
-@run_only_on_default_postgres("Only tests index logic")
-def test_old_index_time_threshold(
-    neon_env_builder: NeonEnvBuilder,
-):
-    """
-    Exercise pageserver's detection of trying to load an ancient non-latest index.
-    (see https://github.com/neondatabase/neon/issues/6951)
-    """
-
-    # Run with local_fs because we will interfere with mtimes by local filesystem access
-    neon_env_builder.enable_pageserver_remote_storage(RemoteStorageKind.LOCAL_FS)
-    env = neon_env_builder.init_start()
-    tenant_id = env.initial_tenant
-    timeline_id = env.initial_timeline
-
-    workload = Workload(env, tenant_id, timeline_id)
-    workload.init()
-    workload.write_rows(32)
-
-    # Remember generation 1's index path
-    assert isinstance(env.pageserver_remote_storage, LocalFsStorage)
-    index_path = env.pageserver_remote_storage.index_path(tenant_id, timeline_id)
-
-    # Increment generation by detaching+attaching, and write+flush some data to get a new remote index
-    env.storage_controller.tenant_policy_update(tenant_id, {"placement": "Detached"})
-    env.storage_controller.tenant_policy_update(tenant_id, {"placement": {"Attached": 0}})
-    env.storage_controller.reconcile_until_idle()
-    workload.churn_rows(32)
-
-    # A new index should have been written
-    assert env.pageserver_remote_storage.index_path(tenant_id, timeline_id) != index_path
-
-    # Hack the mtime on the generation 1 index
-    log.info(f"Setting old mtime on {index_path}")
-    os.utime(index_path, times=(time.time(), time.time() - 30 * 24 * 3600))
-    env.pageserver.allowed_errors.extend(
-        [
-            ".*Found a newer index while loading an old one.*",
-            ".*Index age exceeds threshold and a newer index exists.*",
-        ]
-    )
-
-    # Detach from storage controller + attach in an old generation directly on the pageserver.
-    workload.stop()
-    env.storage_controller.tenant_policy_update(tenant_id, {"placement": "Detached"})
-    env.storage_controller.reconcile_until_idle()
-    env.storage_controller.tenant_policy_update(tenant_id, {"scheduling": "Stop"})
-    env.storage_controller.allowed_errors.append(".*Scheduling is disabled by policy")
-
-    # The controller would not do this (attach in an old generation): we are doing it to simulate
-    # a hypothetical profound bug in the controller.
-    env.pageserver.http_client().tenant_location_conf(
-        tenant_id, {"generation": 1, "mode": "AttachedSingle", "tenant_conf": {}}
-    )
-
-    # The pageserver should react to this situation by refusing to attach the tenant and putting
-    # it into Broken state
-    env.pageserver.allowed_errors.append(".*tenant is broken.*")
-    with pytest.raises(
-        PageserverApiException,
-        match="tenant is broken: Index age exceeds threshold and a newer index exists",
-    ):
-        env.pageserver.http_client().timeline_detail(tenant_id, timeline_id)

test_runner/regress/test_physical_and_logical_replicaiton.py

@@ -53,7 +53,7 @@ def test_aux_not_logged_at_replica(neon_simple_env: NeonEnv, vanilla_pg):
     """Test that AUX files are not saved at replica"""
     env = neon_simple_env
 
-    n_records = 20000
+    n_records = 200000
 
     primary = env.endpoints.create_start(
         branch_name="main",
@@ -91,7 +91,7 @@ def test_aux_not_logged_at_replica(neon_simple_env: NeonEnv, vanilla_pg):
     s_cur.execute("select count(*) from t")
     assert s_cur.fetchall()[0][0] == n_records
 
-    vanilla_pg.stop()
-    secondary.stop()
     primary.stop()
+    time.sleep(1)
+    secondary.stop()
     assert not secondary.log_contains("cannot make new WAL entries during recovery")

test_runner/regress/test_readonly_node.py

@@ -122,7 +122,6 @@ def test_readonly_node(neon_simple_env: NeonEnv):
         )
 
 
-@pytest.mark.skip("See https://github.com/neondatabase/neon/issues/9754")
 def test_readonly_node_gc(neon_env_builder: NeonEnvBuilder):
     """
     Test static endpoint is protected from GC by acquiring and renewing lsn leases.

test_runner/regress/test_tenants.py

@@ -427,7 +427,7 @@ def test_create_churn_during_restart(neon_env_builder: NeonEnvBuilder):
             env.pageserver.start()
 
             for f in futs:
-                f.result(timeout=30)
+                f.result(timeout=10)
 
     # The tenant should end up active
     wait_until_tenant_active(env.pageserver.http_client(), tenant_id, iterations=10, period=1)

test_runner/regress/test_timeline_archive.py

@@ -1,33 +1,22 @@
 from __future__ import annotations
 
 import json
-import random
-import threading
-import time
 from typing import Optional
 
 import pytest
-import requests
-from fixtures.common_types import TenantId, TenantShardId, TimelineArchivalState, TimelineId
+from fixtures.common_types import TenantId, TimelineArchivalState, TimelineId
 from fixtures.log_helper import log
 from fixtures.neon_fixtures import (
     NeonEnvBuilder,
     last_flush_lsn_upload,
 )
 from fixtures.pageserver.http import PageserverApiException
-from fixtures.pageserver.utils import (
-    assert_prefix_empty,
-    assert_prefix_not_empty,
-    list_prefix,
-    wait_until_tenant_active,
-)
-from fixtures.pg_version import PgVersion
+from fixtures.pageserver.utils import assert_prefix_empty, assert_prefix_not_empty, list_prefix
 from fixtures.remote_storage import S3Storage, s3_storage
-from fixtures.utils import run_only_on_default_postgres, wait_until
+from fixtures.utils import wait_until
 from mypy_boto3_s3.type_defs import (
     ObjectTypeDef,
 )
-from psycopg2.errors import IoError, UndefinedTable
 
 
 @pytest.mark.parametrize("shard_count", [0, 4])
@@ -389,279 +378,8 @@ def test_timeline_offload_persist(neon_env_builder: NeonEnvBuilder, delete_timel
     )
 
 
-@run_only_on_default_postgres("this test isn't sensitive to the contents of timelines")
-def test_timeline_archival_chaos(neon_env_builder: NeonEnvBuilder):
-    """
-    A general consistency check on archival/offload timeline state, and its intersection
-    with tenant migrations and timeline deletions.
-    """
-
-    # Offloading is off by default at time of writing: remove this line when it's on by default
-    neon_env_builder.pageserver_config_override = "timeline_offloading = true"
-    neon_env_builder.enable_pageserver_remote_storage(s3_storage())
-
-    # We will exercise migrations, so need multiple pageservers
-    neon_env_builder.num_pageservers = 2
-
-    env = neon_env_builder.init_start(
-        initial_tenant_conf={
-            "compaction_period": "1s",
-        }
-    )
-    tenant_id = env.initial_tenant
-    tenant_shard_id = TenantShardId(tenant_id, 0, 0)
-
-    # Unavailable pageservers during timeline CRUD operations can be logged as errors on the storage controller
-    env.storage_controller.allowed_errors.extend(
-        [
-            ".*error sending request.*",
-            # FIXME: the pageserver should not return 500s on cancellation (https://github.com/neondatabase/neon/issues/97680)
-            ".*InternalServerError(Error deleting timeline .* on .* on .*: pageserver API: error: Cancelled",
-        ]
-    )
-
-    for ps in env.pageservers:
-        # We will do unclean restarts, which results in these messages when cleaning up files
-        ps.allowed_errors.extend(
-            [
-                ".*removing local file.*because it has unexpected length.*",
-                ".*__temp.*",
-                # FIXME: there are still anyhow::Error paths in timeline creation/deletion which
-                # generate 500 results when called during shutdown (https://github.com/neondatabase/neon/issues/9768)
-                ".*InternalServerError.*",
-                # FIXME: there are still anyhow::Error paths in timeline deletion that generate
-                # log lines at error severity (https://github.com/neondatabase/neon/issues/9768)
-                ".*delete_timeline.*Error",
-            ]
-        )
-
-    class TimelineState:
-        def __init__(self):
-            self.timeline_id = TimelineId.generate()
-            self.created = False
-            self.archived = False
-            self.offloaded = False
-            self.deleted = False
-
-    controller_ps_api = env.storage_controller.pageserver_api()
-
-    shutdown = threading.Event()
-
-    violations = []
-
-    timelines_deleted = []
-
-    def list_timelines(tenant_id) -> tuple[set[TimelineId], set[TimelineId]]:
-        """Get the list of active and offloaded TimelineId"""
-        listing = controller_ps_api.timeline_and_offloaded_list(tenant_id)
-        active_ids = set([TimelineId(t["timeline_id"]) for t in listing.timelines])
-        offloaded_ids = set([TimelineId(t["timeline_id"]) for t in listing.offloaded])
-
-        return (active_ids, offloaded_ids)
-
-    def timeline_objects(tenant_shard_id, timeline_id):
-        response = list_prefix(
-            env.pageserver_remote_storage,  # type: ignore
-            prefix="/".join(
-                (
-                    "tenants",
-                    str(tenant_shard_id),
-                    "timelines",
-                    str(timeline_id),
-                )
-            )
-            + "/",
-        )
-
-        return [k["Key"] for k in response.get("Contents", [])]
-
-    def worker():
-        """
-        Background thread which drives timeline lifecycle operations, and checks that between steps
-        it obeys invariants. This should detect errors in pageserver persistence and in errors in
-        concurrent operations on different timelines when it is run many times in parallel.
-        """
-        state = TimelineState()
-
-        # Jitter worker startup, we're not interested in exercising lots of concurrent creations
-        # as we know that's I/O bound.
-        shutdown.wait(random.random() * 10)
-
-        while not shutdown.is_set():
-            # A little wait between actions to jitter out the API calls rather than having them
-            # all queue up at once
-            shutdown.wait(random.random())
-
-            try:
-                if not state.created:
-                    log.info(f"Creating timeline {state.timeline_id}")
-                    controller_ps_api.timeline_create(
-                        PgVersion.NOT_SET, tenant_id=tenant_id, new_timeline_id=state.timeline_id
-                    )
-                    state.created = True
-
-                    if (
-                        timeline_objects(
-                            tenant_shard_id=tenant_shard_id, timeline_id=state.timeline_id
-                        )
-                        == []
-                    ):
-                        msg = f"Timeline {state.timeline_id} unexpectedly not present in remote storage"
-                        violations.append(msg)
-
-                elif state.deleted:
-                    # Try to confirm its deletion completed.
-                    # Deleted timeline should not appear in listing API, either as offloaded or active
-                    (active_ids, offloaded_ids) = list_timelines(tenant_id)
-                    if state.timeline_id in active_ids or state.timeline_id in offloaded_ids:
-                        msg = f"Timeline {state.timeline_id} appeared in listing after deletion was acked"
-                        violations.append(msg)
-                        raise RuntimeError(msg)
-
-                    objects = timeline_objects(tenant_shard_id, state.timeline_id)
-                    if len(objects) == 0:
-                        log.info(f"Confirmed deletion of timeline {state.timeline_id}")
-                        timelines_deleted.append(state.timeline_id)
-                        state = TimelineState()  # A new timeline ID to create on next iteration
-                    else:
-                        # Deletion of objects doesn't have to be synchronous, we will keep polling
-                        log.info(f"Timeline {state.timeline_id} objects still exist: {objects}")
-                        shutdown.wait(random.random())
-                else:
-                    # The main lifetime of a timeline: proceed active->archived->offloaded->deleted
-                    if not state.archived:
-                        log.info(f"Archiving timeline {state.timeline_id}")
-                        controller_ps_api.timeline_archival_config(
-                            tenant_id, state.timeline_id, TimelineArchivalState.ARCHIVED
-                        )
-                        state.archived = True
-                    elif state.archived and not state.offloaded:
-                        log.info(f"Waiting for offload of timeline {state.timeline_id}")
-                        # Wait for offload: this should happen fast because we configured a short compaction interval
-                        while not shutdown.is_set():
-                            (active_ids, offloaded_ids) = list_timelines(tenant_id)
-                            if state.timeline_id in active_ids:
-                                log.info(f"Timeline {state.timeline_id} is still active")
-                                shutdown.wait(0.5)
-                            elif state.timeline_id in offloaded_ids:
-                                log.info(f"Timeline {state.timeline_id} is now offloaded")
-                                state.offloaded = True
-                                break
-                            else:
-                                # Timeline is neither offloaded nor active, this is unexpected: the pageserver
-                                # should ensure that the timeline appears in either the offloaded list or main list
-                                msg = f"Timeline {state.timeline_id} disappeared!"
-                                violations.append(msg)
-                                raise RuntimeError(msg)
-                    elif state.offloaded:
-                        # Once it's offloaded it should only be in offloaded or deleted state: check
-                        # it didn't revert back to active.  This tests that the manfiest is doing its
-                        # job to suppress loading of offloaded timelines as active.
-                        (active_ids, offloaded_ids) = list_timelines(tenant_id)
-                        if state.timeline_id in active_ids:
-                            msg = f"Timeline {state.timeline_id} is active, should be offloaded or deleted"
-                            violations.append(msg)
-                            raise RuntimeError(msg)
-
-                        log.info(f"Deleting timeline {state.timeline_id}")
-                        controller_ps_api.timeline_delete(tenant_id, state.timeline_id)
-                        state.deleted = True
-                    else:
-                        raise RuntimeError("State should be unreachable")
-            except PageserverApiException as e:
-                # This is expected: we are injecting chaos, API calls will sometimes fail.
-                # TODO: can we narrow this to assert we are getting friendly 503s?
-                log.info(f"Iteration error, will retry: {e}")
-                shutdown.wait(random.random())
-            except requests.exceptions.RetryError as e:
-                # Retryable error repeated more times than `requests` is configured to tolerate, this
-                # is expected when a pageserver remains unavailable for a couple seconds
-                log.info(f"Iteration error, will retry: {e}")
-                shutdown.wait(random.random())
-            except Exception as e:
-                log.warning(
-                    f"Unexpected worker exception (current timeline {state.timeline_id}): {e}"
-                )
-            else:
-                # In the non-error case, use a jitterd but small wait, we want to keep
-                # a high rate of operations going
-                shutdown.wait(random.random() * 0.1)
-
-    n_workers = 4
-    threads = []
-    for _i in range(0, n_workers):
-        t = threading.Thread(target=worker)
-        t.start()
-        threads.append(t)
-
-    # Set delay failpoints so that deletions and migrations take some time, and have a good
-    # chance to interact with other concurrent timeline mutations.
-    env.storage_controller.configure_failpoints(
-        [("reconciler-live-migrate-pre-await-lsn", "sleep(1)")]
-    )
-    for ps in env.pageservers:
-        ps.add_persistent_failpoint("in_progress_delete", "sleep(1)")
-
-    # Generate some chaos, while our workers are trying to complete their timeline operations
-    rng = random.Random()
-    try:
-        chaos_rounds = 48
-        for _i in range(0, chaos_rounds):
-            action = rng.choice([0, 1])
-            if action == 0:
-                # Pick a random pageserver to gracefully restart
-                pageserver = rng.choice(env.pageservers)
-
-                # Whether to use a graceful shutdown or SIGKILL
-                immediate = random.choice([True, False])
-                log.info(f"Restarting pageserver {pageserver.id}, immediate={immediate}")
-
-                t1 = time.time()
-                pageserver.restart(immediate=immediate)
-                restart_duration = time.time() - t1
-
-                # Make sure we're up for as long as we spent restarting, to ensure operations can make progress
-                log.info(f"Staying alive for {restart_duration}s")
-                time.sleep(restart_duration)
-            else:
-                # Migrate our tenant between pageservers
-                origin_ps = env.get_tenant_pageserver(tenant_shard_id)
-                dest_ps = rng.choice([ps for ps in env.pageservers if ps.id != origin_ps.id])
-                log.info(f"Migrating {tenant_shard_id} {origin_ps.id}->{dest_ps.id}")
-                env.storage_controller.tenant_shard_migrate(
-                    tenant_shard_id=tenant_shard_id, dest_ps_id=dest_ps.id
-                )
-
-            log.info(f"Full timeline lifecycles so far: {len(timelines_deleted)}")
-    finally:
-        shutdown.set()
-
-        for thread in threads:
-            thread.join()
-
-    # Sanity check that during our run we did exercise some full timeline lifecycles, in case
-    # one of our workers got stuck
-    assert len(timelines_deleted) > 10
-
-    # That no invariant-violations were reported by workers
-    assert violations == []
-
-
-@pytest.mark.parametrize("with_intermediary", [False, True])
-@pytest.mark.parametrize(
-    "offload_child",
-    [
-        "offload",
-        "offload-corrupt",
-        "offload-no-restart",
-        "offload-parent",
-        "archive",
-        None,
-    ],
-)
-def test_timeline_retain_lsn(
-    neon_env_builder: NeonEnvBuilder, with_intermediary: bool, offload_child: Optional[str]
-):
+@pytest.mark.parametrize("offload_child", ["offload", "offload-corrupt", "archive", None])
+def test_timeline_retain_lsn(neon_env_builder: NeonEnvBuilder, offload_child: Optional[str]):
     """
     Ensure that retain_lsn functionality for timelines works, both for offloaded and non-offloaded ones
     """
@@ -669,7 +387,6 @@ def test_timeline_retain_lsn(
         # Our corruption code only works with S3 compatible storage
         neon_env_builder.enable_pageserver_remote_storage(s3_storage())
 
-    neon_env_builder.rust_log_override = "info,[gc_timeline]=debug"
     env = neon_env_builder.init_start()
     ps_http = env.pageserver.http_client()
 
@@ -677,30 +394,22 @@ def test_timeline_retain_lsn(
     tenant_id, root_timeline_id = env.create_tenant(
         conf={
             # small checkpointing and compaction targets to ensure we generate many upload operations
-            "checkpoint_distance": 32 * 1024,
+            "checkpoint_distance": 128 * 1024,
             "compaction_threshold": 1,
-            "compaction_target_size": 32 * 1024,
+            "compaction_target_size": 128 * 1024,
             # set small image creation thresholds so that gc deletes data
-            "image_creation_threshold": 1,
+            "image_creation_threshold": 2,
             # disable background compaction and GC. We invoke it manually when we want it to happen.
             "gc_period": "0s",
             "compaction_period": "0s",
             # Disable pitr, we only want the latest lsn
             "pitr_interval": "0s",
-            "gc_horizon": 0,
             # Don't rely on endpoint lsn leases
             "lsn_lease_length": "0s",
         }
     )
 
-    if with_intermediary:
-        parent_branch_name = "test_archived_parent"
-        parent_timeline_id = env.create_branch("test_archived_parent", tenant_id)
-    else:
-        parent_branch_name = "main"
-        parent_timeline_id = root_timeline_id
-
-    with env.endpoints.create_start(parent_branch_name, tenant_id=tenant_id) as endpoint:
+    with env.endpoints.create_start("main", tenant_id=tenant_id) as endpoint:
         endpoint.safe_psql_many(
             [
                 "CREATE TABLE foo(v int, key serial primary key, t text default 'data_content')",
@@ -710,16 +419,14 @@ def test_timeline_retain_lsn(
         )
         pre_branch_sum = endpoint.safe_psql("SELECT sum(key) from foo where v < 51200")
         log.info(f"Pre branch sum: {pre_branch_sum}")
-        last_flush_lsn_upload(env, endpoint, tenant_id, parent_timeline_id)
+        last_flush_lsn_upload(env, endpoint, tenant_id, root_timeline_id)
 
     # Create a branch and write some additional data to the parent
-    child_timeline_id = env.create_branch(
-        "test_archived_branch", tenant_id, ancestor_branch_name=parent_branch_name
-    )
+    child_timeline_id = env.create_branch("test_archived_branch", tenant_id)
 
-    with env.endpoints.create_start(parent_branch_name, tenant_id=tenant_id) as endpoint:
-        # Do some overwriting churn with compactions in between. This is important so that we can overwrite image layers.
-        for i in range(5):
+    with env.endpoints.create_start("main", tenant_id=tenant_id) as endpoint:
+        # Do some churn of the data. This is important so that we can overwrite image layers.
+        for i in range(10):
             endpoint.safe_psql_many(
                 [
                     f"SELECT setseed(0.23{i})",
@@ -728,9 +435,9 @@ def test_timeline_retain_lsn(
                     "UPDATE foo SET v=(random() * 409600)::int WHERE v % 3 = 0",
                 ]
             )
-            last_flush_lsn_upload(env, endpoint, tenant_id, parent_timeline_id)
         post_branch_sum = endpoint.safe_psql("SELECT sum(key) from foo where v < 51200")
         log.info(f"Post branch sum: {post_branch_sum}")
+        last_flush_lsn_upload(env, endpoint, tenant_id, root_timeline_id)
 
     if offload_child is not None:
         ps_http.timeline_archival_config(
@@ -745,19 +452,9 @@ def test_timeline_retain_lsn(
         assert leaf_detail["is_archived"] is True
         if "offload" in offload_child:
             ps_http.timeline_offload(tenant_id, child_timeline_id)
-        if "offload-parent" in offload_child:
-            # Also offload the parent to ensure the retain_lsn of the child
-            # is entered in the parent at unoffloading
-            ps_http.timeline_archival_config(
-                tenant_id,
-                parent_timeline_id,
-                state=TimelineArchivalState.ARCHIVED,
-            )
-            ps_http.timeline_offload(tenant_id, parent_timeline_id)
 
     # Do a restart to get rid of any in-memory objects (we only init gc info once, at attach)
-    if offload_child is None or "no-restart" not in offload_child:
-        env.pageserver.stop()
+    env.pageserver.stop()
     if offload_child == "offload-corrupt":
         assert isinstance(env.pageserver_remote_storage, S3Storage)
         listing = list_prefix(
@@ -792,21 +489,13 @@ def test_timeline_retain_lsn(
                 ".*page_service_conn_main.*could not find data for key.*",
             ]
         )
-    if offload_child is None or "no-restart" not in offload_child:
-        env.pageserver.start()
-        if offload_child == "offload-parent":
-            wait_until_tenant_active(ps_http, tenant_id=tenant_id)
-            ps_http.timeline_archival_config(
-                tenant_id,
-                parent_timeline_id,
-                state=TimelineArchivalState.UNARCHIVED,
-            )
+    env.pageserver.start()
 
     # Do an agressive gc and compaction of the parent branch
-    ps_http.timeline_gc(tenant_id=tenant_id, timeline_id=parent_timeline_id, gc_horizon=0)
+    ps_http.timeline_gc(tenant_id=tenant_id, timeline_id=root_timeline_id, gc_horizon=0)
     ps_http.timeline_checkpoint(
         tenant_id,
-        parent_timeline_id,
+        root_timeline_id,
         force_l0_compaction=True,
         force_repartition=True,
         wait_until_uploaded=True,
@@ -822,15 +511,10 @@ def test_timeline_retain_lsn(
 
     # Now, after unarchival, the child timeline should still have its data accessible (or corrupted)
     if offload_child == "offload-corrupt":
-        if with_intermediary:
-            error_regex = "(.*could not read .* from page server.*|.*relation .* does not exist)"
-        else:
-            error_regex = ".*failed to get basebackup.*"
-        with pytest.raises((RuntimeError, IoError, UndefinedTable), match=error_regex):
-            with env.endpoints.create_start(
+        with pytest.raises(RuntimeError, match=".*failed to get basebackup.*"):
+            env.endpoints.create_start(
                 "test_archived_branch", tenant_id=tenant_id, basebackup_request_tries=1
-            ) as endpoint:
-                endpoint.safe_psql("SELECT sum(key) from foo where v < 51200")
+            )
     else:
         with env.endpoints.create_start("test_archived_branch", tenant_id=tenant_id) as endpoint:
             sum = endpoint.safe_psql("SELECT sum(key) from foo where v < 51200")

test_runner/regress/test_vm_truncate.py

@@ -1,33 +0,0 @@
-from fixtures.neon_fixtures import NeonEnv
-
-
-#
-# Test that VM is properly truncated
-#
-def test_vm_truncate(neon_simple_env: NeonEnv):
-    env = neon_simple_env
-
-    endpoint = env.endpoints.create_start("main")
-    con = endpoint.connect()
-    cur = con.cursor()
-    cur.execute("CREATE EXTENSION neon_test_utils")
-    cur.execute("CREATE EXTENSION pageinspect")
-
-    cur.execute(
-        "create table t(pk integer primary key, counter integer default 0, filler text default repeat('?', 200))"
-    )
-    cur.execute("insert into t (pk) values (generate_series(1,1000))")
-    cur.execute("delete from t where pk>10")
-    cur.execute("vacuum t")  # truncates the relation, including its VM and FSM
-    # get image of the first block of the VM excluding the page header. It's expected
-    # to still be in the buffer cache.
-    # ignore page header (24 bytes, 48 - it's hex representation)
-    cur.execute("select substr(encode(get_raw_page('t', 'vm', 0), 'hex'), 48)")
-    pg_bitmap = cur.fetchall()[0][0]
-    # flush shared buffers
-    cur.execute("SELECT clear_buffer_cache()")
-    # now download the first block of the VM from the pageserver ...
-    cur.execute("select substr(encode(get_raw_page('t', 'vm', 0), 'hex'), 48)")
-    ps_bitmap = cur.fetchall()[0][0]
-    # and check that content of bitmaps are equal, i.e. PS is producing the same VM page as Postgres
-    assert pg_bitmap == ps_bitmap

vendor/revisions.json

@@ -1,18 +1,18 @@
 {
   "v17": [
-    "17.1",
-    "aa2e29f2b6952140dfe51876bbd11054acae776f"
+    "17.0",
+    "ae4cc30dba24f3910533e5a48e8103c3f2fff300"
   ],
   "v16": [
-    "16.5",
-    "b0b693ea298454e95e6b154780d1fd586a244dfd"
+    "16.4",
+    "03b43900edc5d8d6eecec460bfc89aec7174bd84"
   ],
   "v15": [
-    "15.9",
-    "1feff6b60f07cb71b665d0f5ead71a4320a71743"
+    "15.8",
+    "fd631a959049dfe2b82f67409c8b8b0d3e0016d1"
   ],
   "v14": [
-    "14.14",
-    "c5e0d642efb02e4bfedc283b0a7707fe6c79cc89"
+    "14.13",
+    "de0a000dafc2e66ce2e39282d3aa1c704fe0390e"
   ]
 }

workspace_hack/Cargo.toml

@@ -58,7 +58,7 @@ num-integer = { version = "0.1", features = ["i128"] }
 num-traits = { version = "0.2", features = ["i128", "libm"] }
 once_cell = { version = "1" }
 parquet = { version = "53", default-features = false, features = ["zstd"] }
-postgres-types = { git = "https://github.com/neondatabase/rust-postgres.git", branch = "vlad/interpreted-wal-record-replication-support", default-features = false, features = ["with-serde_json-1"] }
+postgres-types = { git = "https://github.com/neondatabase/rust-postgres.git", rev = "20031d7a9ee1addeae6e0968e3899ae6bf01cee2", default-features = false, features = ["with-serde_json-1"] }
 prost = { version = "0.13", features = ["prost-derive"] }
 rand = { version = "0.8", features = ["small_rng"] }
 regex = { version = "1" }
@@ -75,10 +75,10 @@ smallvec = { version = "1", default-features = false, features = ["const_new", "
 spki = { version = "0.7", default-features = false, features = ["pem", "std"] }
 subtle = { version = "2" }
 sync_wrapper = { version = "0.1", default-features = false, features = ["futures"] }
-tikv-jemalloc-sys = { version = "0.6", features = ["stats"] }
+tikv-jemalloc-sys = { version = "0.5" }
 time = { version = "0.3", features = ["macros", "serde-well-known"] }
 tokio = { version = "1", features = ["fs", "io-std", "io-util", "macros", "net", "process", "rt-multi-thread", "signal", "test-util"] }
-tokio-postgres = { git = "https://github.com/neondatabase/rust-postgres.git", branch = "vlad/interpreted-wal-record-replication-support", features = ["with-serde_json-1"] }
+tokio-postgres = { git = "https://github.com/neondatabase/rust-postgres.git", rev = "20031d7a9ee1addeae6e0968e3899ae6bf01cee2", features = ["with-serde_json-1"] }
 tokio-rustls = { version = "0.26", default-features = false, features = ["logging", "ring", "tls12"] }
 tokio-stream = { version = "0.1", features = ["net"] }
 tokio-util = { version = "0.7", features = ["codec", "compat", "io", "rt"] }