tests/si: use snapshots to find removed segments

Previously, the shadow indexing end-to-end asserted against the current
number of segments when checking for segment removal. This approach has
the downside that a restart/failure of a redpanda node causes a segment
roll, which makes the assertion unreliable in a context with simulated
failures.

This commit changes the assertion to use the
wait_for_removal_of_n_segments helper method which uses segment
snapshots to determine how many segments were removed. The change
deflakes the test against a high number of injected node failures.

tests/rptest/tests/e2e_shadow_indexing_test.py

@@ -27,7 +27,7 @@
 from rptest.util import Scale
 from rptest.util import (
     produce_until_segments,
-    wait_for_segments_removal,
+    wait_for_removal_of_n_segments,
 )
 
 
@@ -89,16 +89,29 @@ def test_write(self):
             count=10,
         )
 
+        # Get a snapshot of the current segments, before tightening the
+        # retention policy.
+        original_snapshot = self.redpanda.storage(
+            all_nodes=True).segments_by_node("kafka", self.topic, 0)
+
+        for node, node_segments in original_snapshot.items():
+            assert len(
+                node_segments
+            ) >= 10, f"Expected at least 10 segments, but got {len(node_segments)} on {node}"
+
         self.kafka_tools.alter_topic_config(
             self.topic,
             {
                 TopicSpec.PROPERTY_RETENTION_BYTES: 5 * self.segment_size,
             },
         )
-        wait_for_segments_removal(redpanda=self.redpanda,
-                                  topic=self.topic,
-                                  partition_idx=0,
-                                  count=6)
+
+        wait_for_removal_of_n_segments(redpanda=self.redpanda,
+                                       topic=self.topic,
+                                       partition_idx=0,
+                                       n=6,
+                                       original_snapshot=original_snapshot)
+
         self.start_consumer()
         self.run_validation()
 
@@ -120,16 +133,28 @@ def test_write_with_node_failures(self):
             count=10,
         )
 
+        # Get a snapshot of the current segments, before tightening the
+        # retention policy.
+        original_snapshot = self.redpanda.storage(
+            all_nodes=True).segments_by_node("kafka", self.topic, 0)
+
+        for node, node_segments in original_snapshot.items():
+            assert len(
+                node_segments
+            ) >= 10, f"Expected at least 10 segments, but got {len(node_segments)} on {node}"
+
         self.kafka_tools.alter_topic_config(
             self.topic,
             {TopicSpec.PROPERTY_RETENTION_BYTES: 5 * self.segment_size},
         )
 
         with random_process_kills(self.redpanda) as ctx:
-            wait_for_segments_removal(redpanda=self.redpanda,
-                                      topic=self.topic,
-                                      partition_idx=0,
-                                      count=6)
+            wait_for_removal_of_n_segments(redpanda=self.redpanda,
+                                           topic=self.topic,
+                                           partition_idx=0,
+                                           n=6,
+                                           original_snapshot=original_snapshot)
+
             self.start_consumer()
             self.run_validation()
         ctx.assert_actions_triggered()