port forward dd722fdaf6 (flush recording arcs outside SmgrOpTimerInner)

This reverts commit 949fbc15a0.

.dockerignore

@@ -24,4 +24,3 @@
 !storage_controller/
 !vendor/postgres-*/
 !workspace_hack/
-!build_tools/patches

build-tools.Dockerfile

@@ -12,8 +12,6 @@ RUN echo 'Acquire::Retries "5";' > /etc/apt/apt.conf.d/80-retries && \
     echo -e "retry_connrefused = on\ntimeout=15\ntries=5\n" > /root/.wgetrc && \
     echo -e "--retry-connrefused\n--connect-timeout 15\n--retry 5\n--max-time 300\n" > /root/.curlrc
 
-COPY build_tools/patches/pgcopydbv017.patch /pgcopydbv017.patch
-
 RUN if [ "${DEBIAN_VERSION}" = "bookworm" ]; then \
         set -e && \
         apt update && \
@@ -46,7 +44,6 @@ RUN if [ "${DEBIAN_VERSION}" = "bookworm" ]; then \
         mkdir /tmp/pgcopydb && \
         tar -xzf /tmp/pgcopydb.tar.gz -C /tmp/pgcopydb --strip-components=1 && \
         cd /tmp/pgcopydb && \
-        patch -p1 < /pgcopydbv017.patch && \
         make -s clean && \
         make -s -j12 install && \
         libpq_path=$(find /lib /usr/lib -name "libpq.so.5" | head -n 1) && \

build_tools/patches/pgcopydbv017.patch

@@ -1,57 +0,0 @@
-diff --git a/src/bin/pgcopydb/copydb.c b/src/bin/pgcopydb/copydb.c
-index d730b03..69a9be9 100644
---- a/src/bin/pgcopydb/copydb.c
-+++ b/src/bin/pgcopydb/copydb.c
-@@ -44,6 +44,7 @@ GUC dstSettings[] = {
- 	{ "synchronous_commit", "'off'" },
- 	{ "statement_timeout", "0" },
- 	{ "lock_timeout", "0" },
-+	{ "idle_in_transaction_session_timeout", "0" },
- 	{ NULL, NULL },
- };
- 
-diff --git a/src/bin/pgcopydb/pgsql.c b/src/bin/pgcopydb/pgsql.c
-index 94f2f46..e051ba8 100644
---- a/src/bin/pgcopydb/pgsql.c
-+++ b/src/bin/pgcopydb/pgsql.c
-@@ -2319,6 +2319,11 @@ pgsql_execute_log_error(PGSQL *pgsql,
- 
- 	LinesBuffer lbuf = { 0 };
- 
-+	if (message != NULL){
-+		// make sure message is writable by splitLines
-+		message = strdup(message);
-+	}
-+
- 	if (!splitLines(&lbuf, message))
- 	{
- 		/* errors have already been logged */
-@@ -2332,6 +2337,7 @@ pgsql_execute_log_error(PGSQL *pgsql,
- 				  PQbackendPID(pgsql->connection),
- 				  lbuf.lines[lineNumber]);
- 	}
-+        free(message); // free copy of message we created above
- 
- 	if (pgsql->logSQL)
- 	{
-@@ -3174,11 +3180,18 @@ pgcopy_log_error(PGSQL *pgsql, PGresult *res, const char *context)
- 		/* errors have already been logged */
- 		return;
- 	}
--
- 	if (res != NULL)
- 	{
- 		char *sqlstate = PQresultErrorField(res, PG_DIAG_SQLSTATE);
--		strlcpy(pgsql->sqlstate, sqlstate, sizeof(pgsql->sqlstate));
-+		if (sqlstate == NULL)
-+		{
-+			// PQresultErrorField returned NULL!
-+			pgsql->sqlstate[0] = '\0';  // Set to an empty string to avoid segfault
-+		}
-+		else
-+		{
-+			strlcpy(pgsql->sqlstate, sqlstate, sizeof(pgsql->sqlstate));
-+		}
- 	}
- 
- 	char *endpoint =

pageserver/src/metrics.rs

@@ -1253,9 +1253,6 @@ pub(crate) struct SmgrOpTimerInner {
     global_batch_wait_time: Histogram,
     per_timeline_batch_wait_time: Histogram,
 
-    global_flush_in_progress_micros: IntCounter,
-    per_timeline_flush_in_progress_micros: IntCounter,
-
     throttling: Arc<tenant_throttling::Pagestream>,
 
     timings: SmgrOpTimerState,
@@ -1366,20 +1363,18 @@ impl SmgrOpTimer {
     /// The first callers receives Some, subsequent ones None.
     ///
     /// See [`SmgrOpTimerState`] for more context.
-    pub(crate) fn observe_execution_end_flush_start(
-        &mut self,
-        at: Instant,
-    ) -> Option<SmgrOpFlushInProgress> {
+    pub(crate) fn observe_execution_end_flush_start(&mut self, at: Instant) {
         // NB: unlike the other observe_* methods, this one take()s.
         #[allow(clippy::question_mark)] // maintain similar code pattern.
         let Some(mut inner) = self.0.take() else {
-            return None;
+            // NB: this take() isn't needed anymore, maybe we can simplify
+            return;
         };
         let SmgrOpTimerState::Executing {
             execution_started_at,
         } = &inner.timings
         else {
-            return None;
+            return;
         };
         // update metrics
         let execution = at - *execution_started_at;
@@ -1394,36 +1389,9 @@ impl SmgrOpTimer {
 
         // state transition
         inner.timings = SmgrOpTimerState::Flushing;
-
-        // return the flush in progress object which
-        // will do the remaining metrics updates
-        let SmgrOpTimerInner {
-            global_flush_in_progress_micros,
-            per_timeline_flush_in_progress_micros,
-            ..
-        } = inner;
-        Some(SmgrOpFlushInProgress {
-            flush_started_at: at,
-            global_micros: global_flush_in_progress_micros,
-            per_timeline_micros: per_timeline_flush_in_progress_micros,
-        })
     }
 }
 
-/// The last stage of request processing is serializing and flushing the request
-/// into the TCP connection. We want to make slow flushes observable
-/// _while they are occuring_, so this struct provides a wrapper method [`Self::measure`]
-/// to periodically bump the metric.
-///
-/// If in the future we decide that we're not interested in live updates, we can
-/// add another `observe_*` method to [`SmgrOpTimer`], follow the existing pattern there,
-/// and remove this struct from the code base.
-pub(crate) struct SmgrOpFlushInProgress {
-    flush_started_at: Instant,
-    global_micros: IntCounter,
-    per_timeline_micros: IntCounter,
-}
-
 impl Drop for SmgrOpTimer {
     fn drop(&mut self) {
         // In case of early drop, update any of the remaining metrics with
@@ -1442,42 +1410,6 @@ impl Drop for SmgrOpTimer {
     }
 }
 
-impl SmgrOpFlushInProgress {
-    pub(crate) async fn measure<Fut, O>(mut self, mut fut: Fut) -> O
-    where
-        Fut: std::future::Future<Output = O>,
-    {
-        let mut fut = std::pin::pin!(fut);
-
-        // Whenever observe_guard gets called, or dropped,
-        // it adds the time elapsed since its last call to metrics.
-        // Last call is tracked in `now`.
-        let mut observe_guard = scopeguard::guard(
-            || {
-                let now = Instant::now();
-                let elapsed = now - self.flush_started_at;
-                self.global_micros
-                    .inc_by(u64::try_from(elapsed.as_micros()).unwrap());
-                self.per_timeline_micros
-                    .inc_by(u64::try_from(elapsed.as_micros()).unwrap());
-                self.flush_started_at = now;
-            },
-            |mut observe| {
-                observe();
-            },
-        );
-
-        loop {
-            match tokio::time::timeout(Duration::from_secs(10), &mut fut).await {
-                Ok(v) => return v,
-                Err(_timeout) => {
-                    (*observe_guard)();
-                }
-            }
-        }
-    }
-}
-
 #[derive(
     Debug,
     Clone,
@@ -1513,6 +1445,56 @@ pub(crate) struct SmgrQueryTimePerTimeline {
     throttling: Arc<tenant_throttling::Pagestream>,
 }
 
+impl SmgrQueryTimePerTimeline {
+    pub(crate) async fn record_flush_in_progress<Fut, O>(
+        shard: &crate::tenant::timeline::handle::WeakHandle<
+            crate::page_service::TenantManagerTypes,
+        >,
+        start_at: Instant,
+        mut fut: Fut,
+    ) -> O
+    where
+        Fut: std::future::Future<Output = O>,
+    {
+        let mut fut = std::pin::pin!(fut);
+
+        // Whenever observe_guard gets called, or dropped,
+        // it adds the time elapsed since its last call to metrics.
+        // Last call is tracked in `now`.
+        let mut base = start_at;
+        let mut observe_guard = scopeguard::guard(
+            || {
+                let Ok(upgraded) = shard.upgrade() else {
+                    return;
+                };
+                let now = Instant::now();
+                let elapsed = now - base;
+                upgraded
+                    .query_metrics
+                    .global_flush_in_progress_micros
+                    .inc_by(u64::try_from(elapsed.as_micros()).unwrap());
+                upgraded
+                    .query_metrics
+                    .per_timeline_flush_in_progress_micros
+                    .inc_by(u64::try_from(elapsed.as_micros()).unwrap());
+                base = now;
+            },
+            |mut observe| {
+                observe();
+            },
+        );
+
+        loop {
+            match tokio::time::timeout(Duration::from_secs(10), &mut fut).await {
+                Ok(v) => return v,
+                Err(_timeout) => {
+                    (*observe_guard)();
+                }
+            }
+        }
+    }
+}
+
 static SMGR_QUERY_STARTED_GLOBAL: Lazy<IntCounterVec> = Lazy::new(|| {
     register_int_counter_vec!(
         // it's a counter, but, name is prepared to extend it to a histogram of queue depth
@@ -1797,10 +1779,6 @@ impl SmgrQueryTimePerTimeline {
         SmgrOpTimer(Some(SmgrOpTimerInner {
             global_execution_latency_histo: self.global_latency[op as usize].clone(),
             per_timeline_execution_latency_histo: per_timeline_latency_histo,
-            global_flush_in_progress_micros: self.global_flush_in_progress_micros.clone(),
-            per_timeline_flush_in_progress_micros: self
-                .per_timeline_flush_in_progress_micros
-                .clone(),
             global_batch_wait_time: self.global_batch_wait_time.clone(),
             per_timeline_batch_wait_time: self.per_timeline_batch_wait_time.clone(),
             throttling: self.throttling.clone(),

pageserver/src/page_service.rs

@@ -1063,9 +1063,10 @@ impl PageServerHandler {
         };
 
         // invoke handler function
-        let (handler_results, span): (
+        let (handler_results, span, shard): (
             Vec<Result<(PagestreamBeMessage, SmgrOpTimer), BatchedPageStreamError>>,
             _,
+            _,
         ) = match batch {
             BatchedFeMessage::Exists {
                 span,
@@ -1082,6 +1083,7 @@ impl PageServerHandler {
                         .map(|msg| (msg, timer))
                         .map_err(|err| BatchedPageStreamError { err, req: req.hdr })],
                     span,
+                    Some(shard),
                 )
             }
             BatchedFeMessage::Nblocks {
@@ -1099,6 +1101,7 @@ impl PageServerHandler {
                         .map(|msg| (msg, timer))
                         .map_err(|err| BatchedPageStreamError { err, req: req.hdr })],
                     span,
+                    Some(shard),
                 )
             }
             BatchedFeMessage::GetPage {
@@ -1126,6 +1129,7 @@ impl PageServerHandler {
                         res
                     },
                     span,
+                    Some(shard),
                 )
             }
             BatchedFeMessage::DbSize {
@@ -1143,6 +1147,7 @@ impl PageServerHandler {
                         .map(|msg| (msg, timer))
                         .map_err(|err| BatchedPageStreamError { err, req: req.hdr })],
                     span,
+                    Some(shard),
                 )
             }
             BatchedFeMessage::GetSlruSegment {
@@ -1160,6 +1165,7 @@ impl PageServerHandler {
                         .map(|msg| (msg, timer))
                         .map_err(|err| BatchedPageStreamError { err, req: req.hdr })],
                     span,
+                    Some(shard),
                 )
             }
             #[cfg(feature = "testing")]
@@ -1181,12 +1187,13 @@ impl PageServerHandler {
                         res
                     },
                     span,
+                    Some(shard),
                 )
             }
             BatchedFeMessage::RespondError { span, error } => {
                 // We've already decided to respond with an error, so we don't need to
                 // call the handler.
-                (vec![Err(error)], span)
+                (vec![Err(error)], span, None)
             }
         };
 
@@ -1194,7 +1201,7 @@ impl PageServerHandler {
         // Some handler errors cause exit from pagestream protocol.
         // Other handler errors are sent back as an error message and we stay in pagestream protocol.
         for handler_result in handler_results {
-            let (response_msg, timer) = match handler_result {
+            let (response_msg, mut timer) = match handler_result {
                 Err(e) => match &e.err {
                     PageStreamError::Shutdown => {
                         // If we fail to fulfil a request during shutdown, which may be _because_ of
@@ -1250,20 +1257,24 @@ impl PageServerHandler {
             // The timer's underlying metric is used for a storage-internal latency SLO and
             // we don't want to include latency in it that we can't control.
             // And as pointed out above, in this case, we don't control the time that flush will take.
-            let flushing_timer = timer.map(|mut timer| {
-                timer
-                    .observe_execution_end_flush_start(Instant::now())
-                    .expect("we are the first caller")
-            });
+            let start_flushing_at = Instant::now();
+            if let Some(timer) = &mut timer {
+                timer.observe_execution_end_flush_start(start_flushing_at);
+            }
 
             // what we want to do
             let flush_fut = pgb_writer.flush();
-            // metric for how long flushing takes
-            let flush_fut = match flushing_timer {
-                Some(flushing_timer) => {
-                    futures::future::Either::Left(flushing_timer.measure(flush_fut))
-                }
-                None => futures::future::Either::Right(flush_fut),
+            let flush_fut = if let Some(shard) = &shard {
+                // don't hold upgraded handle while flushing!
+                futures::future::Either::Left(
+                    metrics::SmgrQueryTimePerTimeline::record_flush_in_progress(
+                        shard,
+                        start_flushing_at,
+                        flush_fut,
+                    ),
+                )
+            } else {
+                futures::future::Either::Right(flush_fut)
             };
             // do it while respecting cancellation
             let _: () = async move {
@@ -2035,12 +2046,6 @@ impl PageServerHandler {
             .get(tenant_id, timeline_id, ShardSelector::Zero)
             .await?;
 
-        if timeline.is_archived() == Some(true) {
-            // TODO after a grace period, turn this log line into a hard error
-            tracing::warn!("timeline {tenant_id}/{timeline_id} is archived, but got basebackup request for it.");
-            //return Err(QueryError::NotFound("timeline is archived".into()))
-        }
-
         let latest_gc_cutoff_lsn = timeline.get_latest_gc_cutoff_lsn();
         if let Some(lsn) = lsn {
             // Backup was requested at a particular LSN. Wait for it to arrive.

pageserver/src/tenant/timeline.rs

@@ -326,7 +326,7 @@ pub struct Timeline {
 
     // `Timeline` doesn't write these metrics itself, but it manages the lifetime.  Code
     // in `crate::page_service` writes these metrics.
-    pub(crate) query_metrics: crate::metrics::SmgrQueryTimePerTimeline,
+    pub(crate) query_metrics: Arc<crate::metrics::SmgrQueryTimePerTimeline>,
 
     directory_metrics: [AtomicU64; DirectoryKind::KINDS_NUM],
 
@@ -2517,11 +2517,11 @@ impl Timeline {
 
                 metrics,
 
-                query_metrics: crate::metrics::SmgrQueryTimePerTimeline::new(
+                query_metrics: Arc::new(crate::metrics::SmgrQueryTimePerTimeline::new(
                     &tenant_shard_id,
                     &timeline_id,
                     resources.pagestream_throttle_metrics,
-                ),
+                )),
 
                 directory_metrics: array::from_fn(|_| AtomicU64::new(0)),
 

safekeeper/src/send_interpreted_wal.rs

@@ -120,20 +120,6 @@ pub enum InterpretedWalReaderError {
     WalStreamClosed,
 }
 
-enum CurrentPositionUpdate {
-    Reset(Lsn),
-    NotReset(Lsn),
-}
-
-impl CurrentPositionUpdate {
-    fn current_position(&self) -> Lsn {
-        match self {
-            CurrentPositionUpdate::Reset(lsn) => *lsn,
-            CurrentPositionUpdate::NotReset(lsn) => *lsn,
-        }
-    }
-}
-
 impl InterpretedWalReaderState {
     fn current_position(&self) -> Option<Lsn> {
         match self {
@@ -143,26 +129,6 @@ impl InterpretedWalReaderState {
             InterpretedWalReaderState::Done => None,
         }
     }
-
-    // Reset the current position of the WAL reader if the requested starting position
-    // of the new shard is smaller than the current value.
-    fn maybe_reset(&mut self, new_shard_start_pos: Lsn) -> CurrentPositionUpdate {
-        match self {
-            InterpretedWalReaderState::Running {
-                current_position, ..
-            } => {
-                if new_shard_start_pos < *current_position {
-                    *current_position = new_shard_start_pos;
-                    CurrentPositionUpdate::Reset(*current_position)
-                } else {
-                    CurrentPositionUpdate::NotReset(*current_position)
-                }
-            }
-            InterpretedWalReaderState::Done => {
-                panic!("maybe_reset called on finished reader")
-            }
-        }
-    }
 }
 
 pub(crate) struct AttachShardNotification {
@@ -444,24 +410,15 @@ impl InterpretedWalReader {
                         };
 
                         senders.push(ShardSenderState { sender_id: new_sender_id, tx: sender, next_record_lsn: start_pos});
-
-                        // If the shard is subscribing below the current position the we need
-                        // to update the cursor that tracks where we are at in the WAL
-                        // ([`Self::state`]) and reset the WAL stream itself
-                        // (`[Self::wal_stream`]). This must be done atomically from the POV of
-                        // anything outside the select statement.
-                        let position_reset = self.state.write().unwrap().maybe_reset(start_pos);
-                        match position_reset {
-                            CurrentPositionUpdate::Reset(to) => {
-                                self.wal_stream.reset(to).await;
-                                wal_decoder = WalStreamDecoder::new(to, self.pg_version);
-                            },
-                            CurrentPositionUpdate::NotReset(_) => {}
-                        };
+                        let current_pos = self.state.read().unwrap().current_position().unwrap();
+                        if start_pos < current_pos {
+                            self.wal_stream.reset(start_pos).await;
+                            wal_decoder = WalStreamDecoder::new(start_pos, self.pg_version);
+                        }
 
                         tracing::info!(
                             "Added shard sender {} with start_pos={} current_pos={}",
-                            ShardSenderId::new(shard_id, new_sender_id), start_pos, position_reset.current_position()
+                            ShardSenderId::new(shard_id, new_sender_id), start_pos, current_pos
                         );
                     }
                 }
@@ -627,7 +584,7 @@ mod tests {
             .unwrap();
 
         let resident_tli = tli.wal_residence_guard().await.unwrap();
-        let end_watch = Env::write_wal(tli, start_lsn, SIZE, MSG_COUNT, None)
+        let end_watch = Env::write_wal(tli, start_lsn, SIZE, MSG_COUNT)
             .await
             .unwrap();
         let end_pos = end_watch.get();
@@ -758,6 +715,7 @@ mod tests {
         const MSG_COUNT: usize = 200;
         const PG_VERSION: u32 = 17;
         const SHARD_COUNT: u8 = 2;
+        const ATTACHED_SHARDS: u8 = 4;
 
         let start_lsn = Lsn::from_str("0/149FD18").unwrap();
         let env = Env::new(true).unwrap();
@@ -767,11 +725,9 @@ mod tests {
             .unwrap();
 
         let resident_tli = tli.wal_residence_guard().await.unwrap();
-        let mut next_record_lsns = Vec::default();
-        let end_watch =
-            Env::write_wal(tli, start_lsn, SIZE, MSG_COUNT, Some(&mut next_record_lsns))
-                .await
-                .unwrap();
+        let end_watch = Env::write_wal(tli, start_lsn, SIZE, MSG_COUNT)
+            .await
+            .unwrap();
         let end_pos = end_watch.get();
 
         let streaming_wal_reader = StreamingWalReader::new(
@@ -790,71 +746,38 @@ mod tests {
         )
         .unwrap();
 
-        struct Sender {
-            tx: Option<tokio::sync::mpsc::Sender<Batch>>,
-            rx: tokio::sync::mpsc::Receiver<Batch>,
-            shard: ShardIdentity,
-            start_lsn: Lsn,
-            received_next_record_lsns: Vec<Lsn>,
-        }
+        let (tx, rx) = tokio::sync::mpsc::channel::<Batch>(MSG_COUNT * 2);
+        let mut batch_receivers = vec![rx];
 
-        impl Sender {
-            fn new(start_lsn: Lsn, shard: ShardIdentity) -> Self {
-                let (tx, rx) = tokio::sync::mpsc::channel::<Batch>(MSG_COUNT * 2);
-                Self {
-                    tx: Some(tx),
-                    rx,
-                    shard,
-                    start_lsn,
-                    received_next_record_lsns: Vec::default(),
-                }
-            }
-        }
-
-        assert!(next_record_lsns.len() > 7);
-        let start_lsns = vec![
-            next_record_lsns[5],
-            next_record_lsns[1],
-            next_record_lsns[3],
-        ];
-        let mut senders = start_lsns
-            .into_iter()
-            .map(|lsn| Sender::new(lsn, shard_0))
-            .collect::<Vec<_>>();
-
-        let first_sender = senders.first_mut().unwrap();
         let handle = InterpretedWalReader::spawn(
             streaming_wal_reader,
-            first_sender.start_lsn,
-            first_sender.tx.take().unwrap(),
-            first_sender.shard,
+            start_lsn,
+            tx,
+            shard_0,
             PG_VERSION,
             &Some("pageserver".to_string()),
         );
 
-        for sender in senders.iter_mut().skip(1) {
-            handle
-                .fanout(sender.shard, sender.tx.take().unwrap(), sender.start_lsn)
-                .unwrap();
+        for _ in 0..(ATTACHED_SHARDS - 1) {
+            let (tx, rx) = tokio::sync::mpsc::channel::<Batch>(MSG_COUNT * 2);
+            handle.fanout(shard_0, tx, start_lsn).unwrap();
+            batch_receivers.push(rx);
         }
 
-        for sender in senders.iter_mut() {
-            loop {
-                let batch = sender.rx.recv().await.unwrap();
-                tracing::info!(
-                    "Sender with start_lsn={} received batch ending at {} with {} records",
-                    sender.start_lsn,
-                    batch.wal_end_lsn,
-                    batch.records.records.len()
+        loop {
+            let batch = batch_receivers.first_mut().unwrap().recv().await.unwrap();
+            for rx in batch_receivers.iter_mut().skip(1) {
+                let other_batch = rx.recv().await.unwrap();
+
+                assert_eq!(batch.wal_end_lsn, other_batch.wal_end_lsn);
+                assert_eq!(
+                    batch.available_wal_end_lsn,
+                    other_batch.available_wal_end_lsn
                 );
+            }
 
-                for rec in batch.records.records {
-                    sender.received_next_record_lsns.push(rec.next_record_lsn);
-                }
-
-                if batch.wal_end_lsn == batch.available_wal_end_lsn {
-                    break;
-                }
+            if batch.wal_end_lsn == batch.available_wal_end_lsn {
+                break;
             }
         }
 
@@ -869,20 +792,5 @@ mod tests {
         }
 
         assert!(done);
-
-        for sender in senders {
-            tracing::info!(
-                "Validating records received by sender with start_lsn={}",
-                sender.start_lsn
-            );
-
-            assert!(sender.received_next_record_lsns.is_sorted());
-            let expected = next_record_lsns
-                .iter()
-                .filter(|lsn| **lsn > sender.start_lsn)
-                .copied()
-                .collect::<Vec<_>>();
-            assert_eq!(sender.received_next_record_lsns, expected);
-        }
     }
 }

safekeeper/src/test_utils.rs

@@ -122,7 +122,6 @@ impl Env {
         start_lsn: Lsn,
         msg_size: usize,
         msg_count: usize,
-        mut next_record_lsns: Option<&mut Vec<Lsn>>,
     ) -> anyhow::Result<EndWatch> {
         let (msg_tx, msg_rx) = tokio::sync::mpsc::channel(receive_wal::MSG_QUEUE_SIZE);
         let (reply_tx, mut reply_rx) = tokio::sync::mpsc::channel(receive_wal::REPLY_QUEUE_SIZE);
@@ -131,7 +130,7 @@ impl Env {
 
         WalAcceptor::spawn(tli.wal_residence_guard().await?, msg_rx, reply_tx, Some(0));
 
-        let prefix = c"neon-file:";
+        let prefix = c"p";
         let prefixlen = prefix.to_bytes_with_nul().len();
         assert!(msg_size >= prefixlen);
         let message = vec![0; msg_size - prefixlen];
@@ -140,9 +139,6 @@ impl Env {
             &mut WalGenerator::new(LogicalMessageGenerator::new(prefix, &message), start_lsn);
         for _ in 0..msg_count {
             let (lsn, record) = walgen.next().unwrap();
-            if let Some(ref mut lsns) = next_record_lsns {
-                lsns.push(lsn);
-            }
 
             let req = AppendRequest {
                 h: AppendRequestHeader {

safekeeper/src/wal_reader_stream.rs

@@ -246,7 +246,7 @@ mod tests {
             .unwrap();
 
         let resident_tli = tli.wal_residence_guard().await.unwrap();
-        let end_watch = Env::write_wal(tli, start_lsn, SIZE, MSG_COUNT, None)
+        let end_watch = Env::write_wal(tli, start_lsn, SIZE, MSG_COUNT)
             .await
             .unwrap();
         let end_pos = end_watch.get();

test_runner/performance/test_perf_ingest_using_pgcopydb.py

@@ -136,7 +136,7 @@ def run_command_and_log_output(command, log_file_path: Path):
         "LD_LIBRARY_PATH": f"{os.getenv('PGCOPYDB_LIB_PATH')}:{os.getenv('PG_16_LIB_PATH')}",
         "PGCOPYDB_SOURCE_PGURI": cast(str, os.getenv("BENCHMARK_INGEST_SOURCE_CONNSTR")),
         "PGCOPYDB_TARGET_PGURI": cast(str, os.getenv("BENCHMARK_INGEST_TARGET_CONNSTR")),
-        "PGOPTIONS": "-c idle_in_transaction_session_timeout=0 -c maintenance_work_mem=8388608 -c max_parallel_maintenance_workers=7",
+        "PGOPTIONS": "-c maintenance_work_mem=8388608 -c max_parallel_maintenance_workers=7",
     }
     # Combine the current environment with custom variables
     env = os.environ.copy()