cluster_link: record the SR-sync task start time

totals_since_task_start.start_time is documented as the time the Schema
Registry sync task started, but the task only ever stamped
current_sync.summary.start_time -- the cumulative summary's start_time
was left unset, so the admin API and rpk reported it as empty.

Stamp it once per task instance on the first run. A fresh instance after
a leadership change re-stamps it, so the field reflects the current
leader's task. A mirroring_task unit test asserts the field is set after a
full sync, and the e2e asserts it is present after a sync and carries a
later value on the post-bounce instance.

Author: pgellert

src/v/cluster_link/schema_registry_sync/mirroring_task.cc

@@ -424,6 +424,13 @@ ss::future<task::state_transition> mirroring_task::full_source_sync(
 
 ss::future<task::state_transition>
 mirroring_task::run_impl(ss::abort_source& as) {
+    // Stamp the cumulative summary's start time once per task instance (the
+    // proto documents totals_since_task_start.start_time as the task start).
+    // A fresh instance after a leadership change re-stamps it on its first run.
+    if (!_status.totals_since_task_start.start_time.has_value()) {
+        _status.totals_since_task_start.start_time = ::model::timestamp::now();
+    }
+
     // Consume the config-changed flag before any co_await so a concurrent
     // update_config during this run is not lost (it re-arms for the next run).
     const bool config_changed = std::exchange(_config_changed, false);

src/v/cluster_link/schema_registry_sync/tests/mirroring_task_test.cc

@@ -212,6 +212,9 @@ TEST_F(mirroring_task_test, full_sync_imports_and_reports) {
     EXPECT_EQ(status->last_full_sync->subject_versions_changed, 3);
     EXPECT_EQ(status->last_full_sync->errors, 0);
     EXPECT_EQ(status->totals_since_task_start.subject_versions_changed, 3);
+    // The cumulative summary's start time is stamped once on the task's first
+    // run; it was previously left unset (only current_sync carried a start).
+    EXPECT_TRUE(status->totals_since_task_start.start_time.has_value());
 
     // Create-only replication imports schema versions but never touches
     // compatibility configs, subject modes, or unsupported-feature handling,

tests/rptest/tests/cluster_linking_schema_registry_sync_test.py

@@ -388,7 +388,10 @@ def counters_ready() -> bool:
 
         # Mapped-but-unimplemented counters must be zero (flagged above).
         totals = sr.totals_since_task_start
-        # The cumulative summary is open-ended, so it carries no finish time.
+        # The cumulative summary carries the task start time (stamped once on the
+        # task's first run) and, being open-ended, has no finish time.
+        assert totals.HasField("start_time"), totals
+        assert totals.start_time.seconds > 0, totals
         assert not totals.HasField("finish_time"), totals
         for name in self.UNPOPULATED_COUNTERS:
             assert getattr(totals, name) == 0, (
@@ -718,6 +721,9 @@ def first_synced() -> bool:
             backoff_sec=1,
             err_msg="first task instance did not fully sync",
         )
+        before_start = (
+            self._admin_sr_status().totals_since_task_start.start_time.ToNanoseconds()
+        )
 
         # Move _schemas/0 leadership to another destination broker. The sync
         # task follows leadership, so a fresh instance takes over there.
@@ -738,22 +744,21 @@ def first_synced() -> bool:
         )
 
         # The new instance resets its cumulative counters (documented on
-        # totals_since_task_start), re-derives the destination inventory from
-        # the replicated _schemas log, and completes a full sync that imports
-        # nothing -- everything is already present. The first instance had
-        # imported every version, so a cumulative subject_versions_changed of 0
-        # marks the reset new instance; last_full_sync == 0 confirms the sync
-        # was a no-op. This relies on the destination scan syncing the local
-        # store first: without it, the freshly-elected leader could see a
-        # stale view of its own store, spuriously re-import already-present
-        # versions, and permanently inflate subject_versions_changed for this
-        # instance (it only ever accumulates, so a transient race here would
-        # never recover).
+        # totals_since_task_start): a fresh, later start_time and
+        # subject_versions_changed back at 0. It re-derives the destination
+        # inventory from the replicated _schemas log and completes a full sync
+        # that imports nothing (last_full_sync == 0) -- everything is already
+        # present. This relies on the destination scan syncing the local store
+        # first: without it, the freshly-elected leader could see a stale view
+        # of its own store, spuriously re-import already-present versions, and
+        # permanently inflate subject_versions_changed for this instance (it
+        # only ever accumulates, so a transient race here would never recover).
         def re_derived() -> bool:
             sr = self._admin_sr_status()
             totals = sr.totals_since_task_start
             return (
                 sr.HasField("last_full_sync")
+                and totals.start_time.ToNanoseconds() > before_start
                 and totals.subject_versions_changed == 0
                 and totals.errors == 0
                 and sr.last_full_sync.subject_versions_changed == 0