Remove committing mode in ingest_record

pageserver/src/import_datadir.rs

@@ -306,8 +306,9 @@ async fn import_wal(
         while last_lsn <= endpoint {
             if let Some((lsn, recdata)) = waldecoder.poll_decode()? {
                 walingest
-                    .ingest_record(recdata, lsn, &mut modification, &mut decoded, ctx, true)
+                    .ingest_record(recdata, lsn, &mut modification, &mut decoded, ctx)
                     .await?;
+                modification.commit().await?;
                 last_lsn = lsn;
 
                 nrecords += 1;
@@ -442,8 +443,9 @@ pub async fn import_wal_from_tar(
         while last_lsn <= end_lsn {
             if let Some((lsn, recdata)) = waldecoder.poll_decode()? {
                 walingest
-                    .ingest_record(recdata, lsn, &mut modification, &mut decoded, ctx, true)
+                    .ingest_record(recdata, lsn, &mut modification, &mut decoded, ctx)
                     .await?;
+                modification.commit().await?;
                 last_lsn = lsn;
 
                 debug!("imported record at {} (end {})", lsn, end_lsn);

pageserver/src/pgdatadir_mapping.rs

@@ -1285,6 +1285,12 @@ impl<'a> DatadirModification<'a> {
     }
 }
 
+/// This struct facilitates accessing either a committed key from the timeline at a
+/// specific LSN, or the latest uncommitted key from a pending modification.
+/// During WAL ingestion, the records from multiple LSNs may be batched in the same
+/// modification before being flushed to the timeline. Hence, the routines in WalIngest
+/// need to look up the keys in the modification first before looking them up in the
+/// timeline to not miss the latest updates.
 #[derive(Clone, Copy)]
 pub enum Version<'a> {
     Lsn(Lsn),

pageserver/src/tenant/timeline.rs

@@ -1567,7 +1567,6 @@ impl Timeline {
         let max_lsn_wal_lag = tenant_conf_guard
             .max_lsn_wal_lag
             .unwrap_or(self.conf.default_tenant_conf.max_lsn_wal_lag);
-        let ingest_batch_size = self.conf.ingest_batch_size;
         drop(tenant_conf_guard);
 
         let mut guard = self.walreceiver.lock().unwrap();
@@ -1583,7 +1582,7 @@ impl Timeline {
                 max_lsn_wal_lag,
                 auth_token: crate::config::SAFEKEEPER_AUTH_TOKEN.get().cloned(),
                 availability_zone: self.conf.availability_zone.clone(),
-                ingest_batch_size,
+                ingest_batch_size: self.conf.ingest_batch_size,
             },
             broker_client,
             ctx,

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

@@ -332,14 +332,7 @@ pub(super) async fn handle_walreceiver_connection(
 
                         // Ingest the records without immediately committing them.
                         walingest
-                            .ingest_record(
-                                recdata,
-                                lsn,
-                                &mut modification,
-                                &mut decoded,
-                                &ctx,
-                                false,
-                            )
+                            .ingest_record(recdata, lsn, &mut modification, &mut decoded, &ctx)
                             .await
                             .with_context(|| format!("could not ingest record at {lsn}"))?;
 

pageserver/src/walingest.rs

@@ -56,7 +56,7 @@ impl WalIngest {
     pub async fn new(
         timeline: &Timeline,
         startpoint: Lsn,
-        ctx: &'_ RequestContext,
+        ctx: &RequestContext,
     ) -> anyhow::Result<WalIngest> {
         // Fetch the latest checkpoint into memory, so that we can compare with it
         // quickly in `ingest_record` and update it when it changes.
@@ -85,7 +85,6 @@ impl WalIngest {
         modification: &mut DatadirModification<'_>,
         decoded: &mut DecodedWALRecord,
         ctx: &RequestContext,
-        commit: bool,
     ) -> anyhow::Result<()> {
         let pg_version = modification.tline.pg_version;
 
@@ -353,11 +352,9 @@ impl WalIngest {
             self.checkpoint_modified = false;
         }
 
-        // Now that this record has been fully handled, including updating the
-        // checkpoint data, let the repository know that it is up-to-date to this LSN
-        if commit {
-            modification.commit().await?;
-        }
+        // Note that at this point this record is only cached in the modification
+        // until commit() is called to flush the data into the repository and update
+        // the latest LSN.
 
         Ok(())
     }