test(ilm): isolate manual transition backpressure

crates/e2e_test/src/inline_fast_path_cluster_test.rs

@@ -2234,7 +2234,6 @@ async fn four_node_manual_transition_distributed_admission_conflict_reports_stat
     let bucket = format!("distributed-admission-{}", Uuid::new_v4().simple());
     let prefix = "transition/distributed-admission/";
     hot_client.create_bucket().bucket(&bucket).send().await?;
-    put_lifecycle_with_transition_retry(&hot_client, &bucket, &tier_name).await?;
     for index in 0u8..64 {
         let key = format!("{prefix}object-{index:02}.bin");
         hot_client
@@ -2245,6 +2244,7 @@ async fn four_node_manual_transition_distributed_admission_conflict_reports_stat
             .send()
             .await?;
     }
+    put_lifecycle_with_transition_retry(&hot_client, &bucket, &tier_name).await?;
 
     let (node0, node1) = tokio::join!(
         start_manual_transition_job_on_node(&hot, 0, &bucket, prefix, &tier_name, false, 64),

crates/ecstore/src/bucket/replication/replication_resyncer.rs

@@ -8229,7 +8229,6 @@ mod tests {
         #[derive(Debug)]
         struct Source {
             body: Bytes,
-            stored: Option<Bytes>,
             info: ObjectInfo,
             ranges: StdMutex<Vec<(i64, i64)>>,
             full_reads: std::sync::atomic::AtomicUsize,
@@ -8258,18 +8257,6 @@ mod tests {
                     self.info.version_id.map(|id| id.to_string()),
                     "every read retains the selected source version"
                 );
-                if let Some(stored) = &self.stored {
-                    if let Some(range) = &range {
-                        self.ranges.lock().expect("range journal lock").push((range.start, range.end));
-                    } else {
-                        self.full_reads.fetch_add(1, Ordering::Relaxed);
-                    }
-                    let plan =
-                        crate::object_api::ReadPlan::build_for_request(range, &self.info, opts, &HeaderMap::new(), None).await?;
-                    let start = plan.storage_offset();
-                    let end = start + usize::try_from(plan.storage_length()).expect("nonnegative storage length");
-                    return plan.into_object_reader(Box::new(std::io::Cursor::new(stored.slice(start..end))), &self.info);
-                }
                 if range.is_none() {
                     self.full_reads.fetch_add(1, Ordering::Relaxed);
                     return Ok(GetObjectReader {
@@ -8428,7 +8415,6 @@ mod tests {
                     ..Default::default()
                 },
                 body: body.clone(),
-                stored: Some(Bytes::from(stored)),
                 ranges: StdMutex::new(Vec::new()),
                 full_reads: std::sync::atomic::AtomicUsize::new(0),
             });
@@ -8670,7 +8656,6 @@ mod tests {
                     ..Default::default()
                 },
                 body: body.clone(),
-                stored: None,
                 ranges: StdMutex::new(Vec::new()),
                 full_reads: std::sync::atomic::AtomicUsize::new(0),
             });