cluster_link: reset SR-sync state on losing _schemas leadership

The mirroring task instance is created once per link and reused across
leadership changes: losing leadership only stops the runner, so its
in-memory status, destination inventory, and last-full-sync timestamp
survived. On an A->B->A handoff the returning leader would report a
prior tenure's stale counters and could skip its first full sync while
_last_full_sync was still recent, deferring catch-up by up to the
full-sync interval.

Reset the derived state in stop() (safe there: task::stop() has closed
the runner gate, so no run_impl is in flight). _config and the pending
_config_changed flag are preserved, so a config change queued while
stopped still takes effect. Every leadership acquisition now re-derives
from the durable destination store, matching a genuinely fresh instance.

Extend the leadership-change e2e to bounce _schemas/0 leadership back to
the original broker and assert the reused instance also starts fresh: a
later start_time, zeroed totals, and a full sync that imports nothing.

Author: pgellert

src/v/cluster_link/schema_registry_sync/mirroring_task.cc

@@ -97,8 +97,22 @@ void mirroring_task::update_config(const model::metadata& link_metadata) {
     _config_changed = true;
 }
 
+void mirroring_task::reset_sync_state() {
+    _status = model::schema_registry_sync_status{};
+    _destination_inventory = inventory{};
+    _reconcile_stats = reconcile_stats{};
+    _last_full_sync.reset();
+}
+
 ss::future<cl_result<void>> mirroring_task::stop() noexcept {
     auto res = co_await task::stop();
+    // task::stop() closed the runner's gate, so no run_impl is in flight and it
+    // is safe to reset the state directly (unlike update_config, which races a
+    // running fiber and defers via _config_changed). Reset so a later leader
+    // starts fresh: this instance may regain _schemas/0 leadership (A->B->A)
+    // and would otherwise report a prior tenure's stale counters/inventory and
+    // skip its first full sync on a still-recent _last_full_sync.
+    reset_sync_state();
     // The run loop has stopped, so no fiber is using the reader; release its
     // HTTP transport. as_future guards the noexcept contract.
     if (_reader) {
@@ -425,9 +439,10 @@ 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.
+    // Stamp the cumulative summary's start time once per leadership acquisition
+    // (the proto documents totals_since_task_start.start_time as the task
+    // start). stop() clears the status on losing leadership, so the next leader
+    // -- a new instance or this same one regaining it -- re-stamps it here.
     if (!_status.totals_since_task_start.start_time.has_value()) {
         _status.totals_since_task_start.start_time = ::model::timestamp::now();
     }

src/v/cluster_link/schema_registry_sync/mirroring_task.h

@@ -87,6 +87,14 @@ class mirroring_task : public task {
     /// new source URL, auth, or TLS setting takes effect on the next run.
     ss::future<> reset_reader();
 
+    /// Clears the in-memory sync state (status counters, destination inventory,
+    /// last-full-sync timestamp) on losing leadership, so the next leader -- a
+    /// new instance or this same one regaining leadership -- re-derives
+    /// everything from the durable destination store instead of a prior
+    /// tenure's view. `_config`/`_config_changed` are preserved: config is
+    /// authoritative and a change queued while stopped must still take effect.
+    void reset_sync_state();
+
     /// Whether a periodic full scan is due (first run, or the full-sync
     /// interval has elapsed). A config change additionally forces one via
     /// `_config_changed`, consumed in `run_impl`.

tests/rptest/tests/cluster_linking_schema_registry_sync_test.py

@@ -792,3 +792,47 @@ def re_derived() -> bool:
             backoff_sec=1,
             err_msg="new leader did not count the post-bounce import",
         )
+
+        # Now bounce leadership back to the original broker (A -> B -> A). Its
+        # task instance ran before, was stopped on losing leadership, and is now
+        # reused -- not a fresh object. stop() clears the in-memory sync state,
+        # so the returning leader must look just as fresh as B did: a later
+        # start_time, zeroed cumulative counters, and an immediate full sync (a
+        # stale _last_full_sync would otherwise defer it up to the full-sync
+        # interval). Without the reset it would report its pre-B totals instead.
+        expected_subjects += 1  # post-bounce-value
+        expected_versions += 1
+        before_return_start = (
+            self._admin_sr_status().totals_since_task_start.start_time.ToNanoseconds()
+        )
+        admin.partition_transfer_leadership(
+            namespace="kafka", topic="_schemas", partition=0, target_id=leader
+        )
+        admin.await_stable_leader(
+            topic="_schemas",
+            partition=0,
+            namespace="kafka",
+            timeout_s=60,
+            check=lambda node_id: node_id == leader,
+        )
+
+        def returning_leader_fresh() -> bool:
+            sr = self._admin_sr_status()
+            totals = sr.totals_since_task_start
+            return (
+                sr.HasField("last_full_sync")
+                and totals.start_time.ToNanoseconds() > before_return_start
+                and totals.subject_versions_changed == 0
+                and totals.errors == 0
+                and sr.last_full_sync.subject_versions_changed == 0
+                and sr.last_full_sync.errors == 0
+                and sr.inventory.selected_source_subjects == expected_subjects
+                and sr.inventory.destination_subject_versions == expected_versions
+            )
+
+        wait_until(
+            returning_leader_fresh,
+            timeout_sec=90,
+            backoff_sec=1,
+            err_msg="reused instance did not reset state on regaining leadership",
+        )