From e8b3c1457a6861541a8ceabfe2569a180599275e Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?=E4=BA=8C=E4=B8=AB=E8=AE=B2=E6=A2=B5?= <eryajf@163.com>
Date: Fri, 11 Sep 2026 23:18:28 +0800
Subject: [PATCH] fix(mq): prevent stale cursor resets from affecting
 subscriptions

---
 .../src/components/mq/SubscriptionsPanel.vue  | 96 ++++++++++++-------
 .../mq/__tests__/SubscriptionsPanel.spec.ts   | 36 +++++++
 2 files changed, 100 insertions(+), 32 deletions(-)

diff --git a/apps/desktop/src/components/mq/SubscriptionsPanel.vue b/apps/desktop/src/components/mq/SubscriptionsPanel.vue
index 55a82bae93..c34a0b5684 100644
--- a/apps/desktop/src/components/mq/SubscriptionsPanel.vue
+++ b/apps/desktop/src/components/mq/SubscriptionsPanel.vue
@@ -1,6 +1,6 @@
 <script setup lang="ts">
 import { formatError } from "@/lib/backend/errorUtils";
-import { ref, watch, computed } from "vue";
+import { ref, watch, computed, onBeforeUnmount } from "vue";
 import { useI18n } from "vue-i18n";
 import type { TopicRef, TopicInfo, SubscriptionInfo, ResetPosition, SkipCount, PeekedMessage, MqSystemKind } from "@/types/mq";
 import { mqListSubscriptions, mqEnrichSubscriptions, mqCreateSubscription, mqDeleteSubscription, mqResetCursor, mqSkipMessages, mqClearBacklog, mqPeekMessages, mqExpireMessages } from "@/lib/backend/api";
@@ -42,6 +42,10 @@ const error = ref<string>();
 let loadSeq = 0;
 const showCreateDialog = ref(false);
 const showResetDialog = ref(false);
+const resetError = ref<string>();
+const resetting = ref(false);
+let resetRequestVersion = 0;
+let subscriptionContextVersion = 0;
 const showSkipDialog = ref(false);
 const showPeekDialog = ref(false);
 const showExpireDialog = ref(false);
@@ -256,10 +260,8 @@ function closeRocketMqDialog() {
 }
 
 function openResetDialog(sub: SubscriptionInfo) {
-  if (props.readOnly) {
-    error.value = t("mqSubscriptions.readOnly");
-    return;
-  }
+  if (props.readOnly) return;
+  closeResetDialog();
   selectedSub.value = sub;
   resetFormData.value = {
     position: "latest",
@@ -270,6 +272,14 @@ function openResetDialog(sub: SubscriptionInfo) {
   showResetDialog.value = true;
 }
 
+function closeResetDialog() {
+  // A closed dialog must not receive results from an earlier reset request.
+  resetRequestVersion += 1;
+  showResetDialog.value = false;
+  resetError.value = undefined;
+  resetting.value = false;
+}
+
 function openSkipDialog(sub: SubscriptionInfo) {
   if (props.readOnly) {
     error.value = t("mqSubscriptions.readOnly");
@@ -361,32 +371,48 @@ async function confirmDelete() {
 }
 
 async function handleResetCursor() {
-  if (!(await guardWritable(t("mqSubscriptions.reset")))) return;
+  if (resetting.value || !showResetDialog.value) return;
+  resetError.value = undefined;
+  if (props.readOnly) {
+    resetError.value = t("mqSubscriptions.readOnly");
+    return;
+  }
   const topicRef = getPulsarTopicRef();
-  if (!selectedSub.value || !topicRef) return;
-  loading.value = true;
-  error.value = undefined;
+  const sub = selectedSub.value;
+  if (!sub || !topicRef) return;
+  const connectionId = props.connectionId;
+  const form = { ...resetFormData.value };
+  const requestVersion = ++resetRequestVersion;
+  const contextVersion = subscriptionContextVersion;
+  resetting.value = true;
   try {
     let pos: ResetPosition;
-    if (resetFormData.value.position === "timestamp") {
-      pos = { kind: "timestamp", timestampMs: resetFormData.value.timestampMs };
-    } else if (resetFormData.value.position === "partitionOffset") {
-      const { partition, offset } = resetFormData.value;
+    if (form.position === "timestamp") {
+      pos = { kind: "timestamp", timestampMs: form.timestampMs };
+    } else if (form.position === "partitionOffset") {
+      const { partition, offset } = form;
       if (!Number.isSafeInteger(partition) || partition < 0 || !Number.isSafeInteger(offset) || offset < 0) {
-        error.value = t("mqSubscriptions.nonNegativeIntegerRequired");
+        resetError.value = t("mqSubscriptions.nonNegativeIntegerRequired");
         return;
       }
       pos = { kind: "partitionOffset", partition, offset };
     } else {
-      pos = { kind: resetFormData.value.position };
+      pos = { kind: form.position };
     }
-    await mqResetCursor(props.connectionId, topicRef, selectedSub.value.name, pos);
-    showResetDialog.value = false;
+    if (!(await confirmMqWrite(t("mqSubscriptions.reset"))) || requestVersion !== resetRequestVersion) return;
+    if (props.readOnly) {
+      resetError.value = t("mqSubscriptions.readOnly");
+      return;
+    }
+    await mqResetCursor(connectionId, topicRef, sub.name, pos);
+    if (contextVersion !== subscriptionContextVersion) return;
+    // Closing the dialog does not cancel a reset already sent to the backend.
+    if (requestVersion === resetRequestVersion) closeResetDialog();
     await loadSubscriptions();
   } catch (e: unknown) {
-    error.value = formatError(e);
+    if (requestVersion === resetRequestVersion) resetError.value = formatError(e);
   } finally {
-    loading.value = false;
+    if (requestVersion === resetRequestVersion) resetting.value = false;
   }
 }
 
@@ -489,14 +515,20 @@ async function handleExpireMessages() {
 }
 
 watch(
-  () => [props.topic, props.tenant, props.namespace, props.mqSystemKind],
+  () => [props.connectionId, props.topic, props.tenant, props.namespace, props.mqSystemKind],
   () => {
+    subscriptionContextVersion += 1;
+    closeResetDialog();
     selectedSub.value = undefined;
     invalidatePeekRequest();
     loadSubscriptions();
   },
   { immediate: true },
 );
+onBeforeUnmount(() => {
+  subscriptionContextVersion += 1;
+  closeResetDialog();
+});
 </script>
 
 <template>
@@ -639,50 +671,50 @@ watch(
     </div>
 
     <!-- Reset Cursor Dialog -->
-    <div v-if="!isRocketMqCluster && showResetDialog" class="dialog-overlay" @click="showResetDialog = false">
+    <div v-if="!isRocketMqCluster && showResetDialog" class="dialog-overlay" @click="closeResetDialog">
       <div class="dialog" @click.stop>
         <div class="dialog-header">
           <h3>{{ t("mqSubscriptions.resetDialogTitle", { name: selectedSub?.name }) }}</h3>
-          <button @click="showResetDialog = false" class="btn-close">×</button>
+          <button @click="closeResetDialog" class="btn-close">×</button>
         </div>
         <div class="dialog-body">
           <div class="form-group">
             <label>{{ t("mqSubscriptions.resetTo") }}</label>
             <div class="radio-group">
               <label class="radio-label">
-                <input type="radio" v-model="resetFormData.position" value="earliest" :disabled="readOnly" />
+                <input type="radio" v-model="resetFormData.position" value="earliest" :disabled="resetting || readOnly" />
                 {{ t("mqSubscriptions.earliest") }}
               </label>
               <label class="radio-label">
-                <input type="radio" v-model="resetFormData.position" value="latest" :disabled="readOnly" />
+                <input type="radio" v-model="resetFormData.position" value="latest" :disabled="resetting || readOnly" />
                 {{ t("mqSubscriptions.latest") }}
               </label>
               <label class="radio-label">
-                <input type="radio" v-model="resetFormData.position" value="timestamp" :disabled="readOnly" />
+                <input type="radio" v-model="resetFormData.position" value="timestamp" :disabled="resetting || readOnly" />
                 {{ t("mqSubscriptions.timestamp") }}
               </label>
               <label v-if="mqSystemKind === 'kafka'" class="radio-label">
-                <input type="radio" v-model="resetFormData.position" value="partitionOffset" :disabled="readOnly" />
+                <input type="radio" v-model="resetFormData.position" value="partitionOffset" :disabled="resetting || readOnly" />
                 {{ t("mqSubscriptions.partitionOffset") }}
               </label>
             </div>
           </div>
           <div v-if="resetFormData.position === 'timestamp'" class="form-group">
             <label>{{ t("mqSubscriptions.timestampMs") }}</label>
-            <input v-model.number="resetFormData.timestampMs" type="number" :disabled="readOnly" />
+            <input v-model.number="resetFormData.timestampMs" type="number" :disabled="resetting || readOnly" />
             <div class="form-hint">{{ t("mqSubscriptions.currentTime", { time: new Date(resetFormData.timestampMs).toLocaleString() }) }}</div>
           </div>
           <div v-if="mqSystemKind === 'kafka' && resetFormData.position === 'partitionOffset'" class="form-group">
             <label>{{ t("mqSubscriptions.partition") }}</label>
-            <input v-model.number="resetFormData.partition" data-testid="reset-partition" type="number" min="0" step="1" :disabled="readOnly" />
+            <input v-model.number="resetFormData.partition" data-testid="reset-partition" type="number" min="0" step="1" :disabled="resetting || readOnly" />
             <label>{{ t("mqSubscriptions.offset") }}</label>
-            <input v-model.number="resetFormData.offset" data-testid="reset-offset" type="number" min="0" step="1" :disabled="readOnly" />
+            <input v-model.number="resetFormData.offset" data-testid="reset-offset" type="number" min="0" step="1" :disabled="resetting || readOnly" />
           </div>
-          <div v-if="error" class="form-error">{{ error }}</div>
+          <div v-if="resetError" class="form-error">{{ resetError }}</div>
         </div>
         <div class="dialog-footer">
-          <button @click="showResetDialog = false" class="btn-secondary">{{ t("mqSubscriptions.cancel") }}</button>
-          <button @click="handleResetCursor" :disabled="loading || readOnly" class="btn-primary">{{ t("mqSubscriptions.reset") }}</button>
+          <button @click="closeResetDialog" class="btn-secondary">{{ t("mqSubscriptions.cancel") }}</button>
+          <button @click="handleResetCursor" :disabled="resetting || readOnly" class="btn-primary">{{ t("mqSubscriptions.reset") }}</button>
         </div>
       </div>
     </div>
diff --git a/apps/desktop/src/components/mq/__tests__/SubscriptionsPanel.spec.ts b/apps/desktop/src/components/mq/__tests__/SubscriptionsPanel.spec.ts
index 16e92227d0..728f7517ca 100644
--- a/apps/desktop/src/components/mq/__tests__/SubscriptionsPanel.spec.ts
+++ b/apps/desktop/src/components/mq/__tests__/SubscriptionsPanel.spec.ts
@@ -241,6 +241,42 @@ describe("SubscriptionsPanel Kafka absolute offset reset", () => {
     expect(backend.mqResetCursor).toHaveBeenCalledWith("mq-1", expect.objectContaining({ topic: "events" }), "orders-consumer", { kind: "partitionOffset", partition: 1, offset: 42 });
   });
 
+  it.each([false, true])("refreshes after a closed reset succeeds without closing a new dialog (reopen: %s)", async (reopen) => {
+    const reset = deferred<void>();
+    backend.mqResetCursor.mockReturnValueOnce(reset.promise);
+    const panel = await mountPanel();
+    await openAbsoluteReset(panel);
+    buttonWithExactText(panel, "mqSubscriptions.reset").click();
+    await vi.waitFor(() => expect(backend.mqResetCursor).toHaveBeenCalledTimes(1));
+
+    buttonWithExactText(panel, "mqSubscriptions.cancel").click();
+    await flushUi();
+    if (reopen) await openAbsoluteReset(panel);
+    backend.mqListSubscriptions.mockResolvedValue([{ ...subscription("orders-consumer", "NORMAL"), msgBacklog: 123 }]);
+    reset.resolve();
+
+    await vi.waitFor(() => {
+      expect(backend.mqListSubscriptions).toHaveBeenCalledTimes(2);
+      expect(panel.querySelector(".subscriptions-table")?.textContent).toContain("123");
+    });
+    expect(panel.querySelector('input[value="partitionOffset"]') !== null).toBe(reopen);
+    expect(panel.querySelector(".form-error")).toBeNull();
+  });
+
+  it("does not refresh after a reset succeeds on an unmounted panel", async () => {
+    const reset = deferred<void>();
+    backend.mqResetCursor.mockReturnValueOnce(reset.promise);
+    const panel = await mountPanel();
+    await openAbsoluteReset(panel);
+    buttonWithExactText(panel, "mqSubscriptions.reset").click();
+    await vi.waitFor(() => expect(backend.mqResetCursor).toHaveBeenCalledTimes(1));
+    app!.unmount();
+    app = null;
+    reset.resolve();
+    await flushUi();
+    expect(backend.mqListSubscriptions).toHaveBeenCalledTimes(1);
+  });
+
   it.each([
     ["partition", "-1"],
     ["partition", "1.5"],