Merge pull request #30996 from andrwng/shadow-linking-ci

[CORE-15451] cluster_link/replication: retry deferred prefix truncation on commit

Author: andrwng

src/v/cluster_link/replication/partition_replicator.cc

@@ -121,6 +121,9 @@ ss::future<bool> partition_replicator::handle_replication_result(
         // source to sink
         // We only intend to backoff on consecutive failures.
         _backoff_policy.reset();
+        // The shadow high watermark just advanced, so a prefix truncation that
+        // was deferred waiting for the shadow to catch up can now proceed.
+        co_await maybe_synchronize_start_offset();
         co_return true;
     } catch (...) {
         auto eptr = std::current_exception();

src/v/cluster_link/replication/tests/partition_replicator_tests.cc

@@ -408,4 +408,22 @@ TEST_F_CORO(PartitionReplicatorFixture, TestNoStartOffsetOvershootOnResume) {
       5s, [&] { return _sink->start_offset() == kafka::offset(50); });
 }
 
+// A shadow partition that catches up to a prefix-trimmed source in a single
+// batch must still sync its start offset. When the batch is fetched the shadow
+// HWM is still behind the trim point, so the truncation is deferred; and once
+// the batch is replicated no further source data arrives to drive another
+// fetch. The truncation must therefore be retried when the batch commits.
+TEST_F_CORO(PartitionReplicatorFixture, TestTrimSyncAfterSingleCatchUpFetch) {
+    // Source is prefix-trimmed to 5 but still has a moving tail (lso > 5), so
+    // truncation stays deferred until the shadow catches up to the trim point.
+    _source->set_reported_start_offset(kafka::offset(5));
+
+    // A single batch [0, 9] catches the shadow up past the trim point. After it
+    // commits the source is idle, so the commit itself must retry the sync.
+    co_await push_data();
+
+    RPTEST_REQUIRE_EVENTUALLY_CORO(
+      5s, [&] { return _sink->start_offset() == kafka::offset(5); });
+}
+
 } // namespace cluster_link::replication