cluster_link/replication: retry deferred prefix truncation on commit

When a source partition is prefix-trimmed, the shadow partition has to
truncate to the same start offset. maybe_synchronize_start_offset()
defers that while the shadow hasn't replicated up to the source start
offset yet and (before this commit) relied on the next fetch_next() to
retry.

It's possible that after deferring the prefix-trim, we replicate up to
the source HWM so there isn't a next fetch_next(), and so our start
offset never proceeds.

This commit retries the synchronization when a batch commits, which is
exactly when the shadow high watermark advances past the offset the
truncation was waiting on, so it no longer depends on new source data
arriving.

This attempts to fix
ShadowLinkingReplicationTests.test_auto_prefix_trimming which is quite
flaky.

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