github/dbx
4
0 Fork 0
mirror of https://github.com/t8y2/dbx.git synced 2026-11-15 14:52:36 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(mq): prevent stale cursor resets from affecting subscriptions

Browse Source
This commit is contained in:
二丫讲梵 authored and GitHub committed 2026-09-11 23:18:28 +08:00
1 parent 48d6c94aad
commit e8b3c1457a
2 files changed
+100 -32
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

apps/desktop/src/components/mq/SubscriptionsPanel.vue
+64 -32
View File
@@ -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>
apps/desktop/src/components/mq/__tests__/SubscriptionsPanel.spec.ts
Vendored
+36
View File
@@ -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"],