fix(chat): cross-tab lock реплея офлайн-очереди + идемпотентность мессенджера (conv_messages)
Клиентская часть гонки (вариант B): Web Locks API (navigator.locks) — кросс-вкладочный мьютекс iistwin-offline-sync-replay на replay-проход в useOfflineSync: первая вкладка под локом реплеит и удаляет записи, остальные видят пустую очередь. Fallback для сред без Web Locks — старое поведение (дубли гасит серверная идемпотентность). Race-тест: без locks → 2 POST, с locks → 1 POST для одной записи очереди. Мессенджер (conv_messages) — та же защита, что у task_messages: - conversation_messages.client_message_id + partial unique (миграция 0087); - POST /api/messenger/conversations/:id/messages: повтор с тем же clientMessageId → существующее сообщение без insert и без SSE/уведомлений, гонка insert'ов ловится по 23505; - useChatController шлёт crypto.randomUUID(), attachment-реплей мессенджера тоже использует id записи очереди.
This commit is contained in:
@@ -964,7 +964,7 @@ export function useChatController() {
|
|||||||
};
|
};
|
||||||
|
|
||||||
const sendMutation = useMutation({
|
const sendMutation = useMutation({
|
||||||
mutationFn: async (payload: { message: string; replyToId?: number; mentionedUserIds?: number[]; attachments?: ChatAttachment[] }) => {
|
mutationFn: async (payload: { message: string; replyToId?: number; mentionedUserIds?: number[]; attachments?: ChatAttachment[]; clientMessageId?: string }) => {
|
||||||
const res = await apiRequest('POST', `/api/messenger/conversations/${activeConvId}/messages`, payload);
|
const res = await apiRequest('POST', `/api/messenger/conversations/${activeConvId}/messages`, payload);
|
||||||
const data = await res.json();
|
const data = await res.json();
|
||||||
if (!data.success) throw new Error(data.error);
|
if (!data.success) throw new Error(data.error);
|
||||||
@@ -1180,6 +1180,9 @@ export function useChatController() {
|
|||||||
replyToId: replyTo?.id,
|
replyToId: replyTo?.id,
|
||||||
mentionedUserIds: mentionedUserIds.length > 0 ? mentionedUserIds : undefined,
|
mentionedUserIds: mentionedUserIds.length > 0 ? mentionedUserIds : undefined,
|
||||||
attachments: undefined,
|
attachments: undefined,
|
||||||
|
// Ключ идемпотентности: реплеи офлайн-очереди (в т.ч. из разных вкладок)
|
||||||
|
// идут с одним ключом → серверный дедуп по client_message_id (миграция 0087)
|
||||||
|
clientMessageId: crypto.randomUUID(),
|
||||||
});
|
});
|
||||||
setMentionedUserIds([]);
|
setMentionedUserIds([]);
|
||||||
setMentionSearch(null);
|
setMentionSearch(null);
|
||||||
|
|||||||
@@ -86,6 +86,17 @@ function updateAppBadge(n: number) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// Кросс-вкладочный мьютекс на реплей офлайн-очереди (Web Locks API). Без него N
|
||||||
|
// открытых вкладок реплеят одну и ту же IndexedDB-очередь НЕЗАВИСИМО (модульный
|
||||||
|
// _syncLock свой у каждой вкладки) → дубли сообщений/мутаций на сервере.
|
||||||
|
// Fallback для сред без Web Locks — старое поведение (дубли гасит серверная
|
||||||
|
// идемпотентность по clientMessageId, миграции 0086/0087).
|
||||||
|
async function withCrossTabSyncLock<T>(fn: () => Promise<T>): Promise<T> {
|
||||||
|
const locks = (navigator as unknown as { locks?: { request: (name: string, cb: () => Promise<T>) => Promise<T> } }).locks;
|
||||||
|
if (!locks?.request) return fn();
|
||||||
|
return locks.request('iistwin-offline-sync-replay', fn);
|
||||||
|
}
|
||||||
|
|
||||||
async function replayRequest(item: OfflineQueueItem): Promise<Response> {
|
async function replayRequest(item: OfflineQueueItem): Promise<Response> {
|
||||||
const headers: Record<string, string> = {
|
const headers: Record<string, string> = {
|
||||||
'Content-Type': 'application/json',
|
'Content-Type': 'application/json',
|
||||||
@@ -257,10 +268,8 @@ async function trySendAttachmentMessage(
|
|||||||
};
|
};
|
||||||
|
|
||||||
// Идемпотентность: id записи очереди стабилен между реплеями (в т.ч. из разных
|
// Идемпотентность: id записи очереди стабилен между реплеями (в т.ч. из разных
|
||||||
// вкладок) — сервер дедупит повторы по client_message_id (миграция 0086).
|
// вкладок) — сервер дедупит повторы по client_message_id (миграции 0086/0087).
|
||||||
if (item.type === 'task') {
|
payload.clientMessageId = (item as { clientMessageId?: string }).clientMessageId ?? item.id;
|
||||||
payload.clientMessageId = (item as { clientMessageId?: string }).clientMessageId ?? item.id;
|
|
||||||
}
|
|
||||||
|
|
||||||
if (item.type === 'messenger' && (item.replyToId || item.replyToMessageId)) {
|
if (item.type === 'messenger' && (item.replyToId || item.replyToMessageId)) {
|
||||||
payload.replyToId = item.replyToId || item.replyToMessageId;
|
payload.replyToId = item.replyToId || item.replyToMessageId;
|
||||||
@@ -355,7 +364,7 @@ export function useOfflineSync(userId?: number): UseOfflineSyncReturn {
|
|||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
|
|
||||||
_syncLock = (async () => {
|
_syncLock = (async () => withCrossTabSyncLock(async () => {
|
||||||
setStatus('syncing');
|
setStatus('syncing');
|
||||||
|
|
||||||
try {
|
try {
|
||||||
@@ -564,7 +573,7 @@ export function useOfflineSync(userId?: number): UseOfflineSyncReturn {
|
|||||||
} finally {
|
} finally {
|
||||||
_syncLock = null;
|
_syncLock = null;
|
||||||
}
|
}
|
||||||
})();
|
}))();
|
||||||
await _syncLock;
|
await _syncLock;
|
||||||
}, []);
|
}, []);
|
||||||
|
|
||||||
|
|||||||
10
migrations/0087_conv_message_idempotency.sql
Normal file
10
migrations/0087_conv_message_idempotency.sql
Normal file
@@ -0,0 +1,10 @@
|
|||||||
|
-- Идемпотентность сообщений мессенджера (conversation_messages): клиентский ключ.
|
||||||
|
-- Та же защита, что у task_messages (миграция 0086): multi-tab replay
|
||||||
|
-- офлайн-очереди, ретраи после сетевых сбоев, двойные клики.
|
||||||
|
-- Partial unique index: NULL-значения (старые сообщения, боты) не конфликтуют.
|
||||||
|
|
||||||
|
ALTER TABLE conversation_messages ADD COLUMN IF NOT EXISTS client_message_id varchar(64);
|
||||||
|
|
||||||
|
CREATE UNIQUE INDEX IF NOT EXISTS conv_messages_client_message_id_unique
|
||||||
|
ON conversation_messages (client_message_id)
|
||||||
|
WHERE client_message_id IS NOT NULL;
|
||||||
@@ -139,6 +139,7 @@ export function registerMessengerMessageRoutes(router: Router): void {
|
|||||||
size: z.number().default(0),
|
size: z.number().default(0),
|
||||||
mimeType: z.string().optional(),
|
mimeType: z.string().optional(),
|
||||||
})).optional(),
|
})).optional(),
|
||||||
|
clientMessageId: z.string().max(64).optional(),
|
||||||
});
|
});
|
||||||
|
|
||||||
router.post("/api/messenger/conversations/:id/messages",
|
router.post("/api/messenger/conversations/:id/messages",
|
||||||
@@ -171,17 +172,60 @@ export function registerMessengerMessageRoutes(router: Router): void {
|
|||||||
|
|
||||||
const orgId = conv?.organizationId ?? req.organizationId!;
|
const orgId = conv?.organizationId ?? req.organizationId!;
|
||||||
|
|
||||||
const [newMsg] = await db
|
// Идемпотентность: повтор с тем же clientMessageId — вернуть уже созданное
|
||||||
.insert(conversationMessages)
|
// сообщение БЕЗ повторного insert и без side-эффектов (SSE/уведомления
|
||||||
.values({
|
// уже отработали при первой записи). Защита от дублей при multi-tab replay
|
||||||
conversationId: convId,
|
// офлайн-очереди (миграция 0087).
|
||||||
authorId: userId,
|
const clientMessageId = parsed.data.clientMessageId ?? null;
|
||||||
message: message.trim(),
|
if (clientMessageId) {
|
||||||
replyToId: replyToId ?? null,
|
const [existing] = await db
|
||||||
mentionedUserIds: mentionedUserIds?.length ? mentionedUserIds : null,
|
.select({ id: conversationMessages.id })
|
||||||
attachments: attachments?.length ? attachments : null,
|
.from(conversationMessages)
|
||||||
})
|
.where(and(
|
||||||
.returning();
|
eq(conversationMessages.conversationId, convId),
|
||||||
|
eq(conversationMessages.clientMessageId, clientMessageId)
|
||||||
|
))
|
||||||
|
.limit(1);
|
||||||
|
if (existing) {
|
||||||
|
const enrichedExisting = await enrichMessage(existing.id);
|
||||||
|
return res.status(200).json({ success: true, message: enrichedExisting, deduplicated: true });
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
let newMsg;
|
||||||
|
try {
|
||||||
|
[newMsg] = await db
|
||||||
|
.insert(conversationMessages)
|
||||||
|
.values({
|
||||||
|
conversationId: convId,
|
||||||
|
authorId: userId,
|
||||||
|
message: message.trim(),
|
||||||
|
replyToId: replyToId ?? null,
|
||||||
|
mentionedUserIds: mentionedUserIds?.length ? mentionedUserIds : null,
|
||||||
|
attachments: attachments?.length ? attachments : null,
|
||||||
|
clientMessageId,
|
||||||
|
})
|
||||||
|
.returning();
|
||||||
|
} catch (err) {
|
||||||
|
// Гонка: параллельный запрос с тем же clientMessageId уже вставил
|
||||||
|
// сообщение (unique index conv_messages_client_message_id_unique).
|
||||||
|
const e = err as { code?: string; cause?: { code?: string } };
|
||||||
|
if (clientMessageId && (e?.code === '23505' || e?.cause?.code === '23505')) {
|
||||||
|
const [existing] = await db
|
||||||
|
.select({ id: conversationMessages.id })
|
||||||
|
.from(conversationMessages)
|
||||||
|
.where(and(
|
||||||
|
eq(conversationMessages.conversationId, convId),
|
||||||
|
eq(conversationMessages.clientMessageId, clientMessageId)
|
||||||
|
))
|
||||||
|
.limit(1);
|
||||||
|
if (existing) {
|
||||||
|
const enrichedExisting = await enrichMessage(existing.id);
|
||||||
|
return res.status(200).json({ success: true, message: enrichedExisting, deduplicated: true });
|
||||||
|
}
|
||||||
|
}
|
||||||
|
throw err;
|
||||||
|
}
|
||||||
|
|
||||||
const enriched = await enrichMessage(newMsg.id);
|
const enriched = await enrichMessage(newMsg.id);
|
||||||
const members = await getMembersForSSE(convId, orgId);
|
const members = await getMembersForSSE(convId, orgId);
|
||||||
|
|||||||
@@ -2975,6 +2975,9 @@ export const conversationMessages = pgTable("conversation_messages", {
|
|||||||
mentionedUserIds: jsonb("mentioned_user_ids").$type<number[]>(),
|
mentionedUserIds: jsonb("mentioned_user_ids").$type<number[]>(),
|
||||||
attachments: jsonb("attachments").$type<Array<{ url: string; name: string; size: number; mimeType?: string }>>(),
|
attachments: jsonb("attachments").$type<Array<{ url: string; name: string; size: number; mimeType?: string }>>(),
|
||||||
botButtons: jsonb("bot_buttons").$type<Array<{ text: string; callbackData: string; style?: 'primary' | 'secondary' | 'destructive' | 'outline'; row?: number }>>(),
|
botButtons: jsonb("bot_buttons").$type<Array<{ text: string; callbackData: string; style?: 'primary' | 'secondary' | 'destructive' | 'outline'; row?: number }>>(),
|
||||||
|
// Идемпотентность: клиентский ключ сообщения (uuid от клиента / id записи очереди).
|
||||||
|
// Partial unique index — см. миграцию 0087.
|
||||||
|
clientMessageId: varchar("client_message_id", { length: 64 }),
|
||||||
createdAt: timestamp("created_at").defaultNow(),
|
createdAt: timestamp("created_at").defaultNow(),
|
||||||
updatedAt: timestamp("updated_at").defaultNow(),
|
updatedAt: timestamp("updated_at").defaultNow(),
|
||||||
isDeleted: boolean("is_deleted").default(false),
|
isDeleted: boolean("is_deleted").default(false),
|
||||||
|
|||||||
@@ -87,6 +87,8 @@ describe('multi-tab offline queue replay race', () => {
|
|||||||
beforeEach(() => {
|
beforeEach(() => {
|
||||||
sharedQueue.length = 0;
|
sharedQueue.length = 0;
|
||||||
replayedPosts = [];
|
replayedPosts = [];
|
||||||
|
// По умолчанию Web Locks НЕТ (fallback-путь) — гонка воспроизводится
|
||||||
|
delete (navigator as unknown as { locks?: unknown }).locks;
|
||||||
vi.stubGlobal('fetch', vi.fn(async (input: RequestInfo | URL, init?: RequestInit) => {
|
vi.stubGlobal('fetch', vi.fn(async (input: RequestInfo | URL, init?: RequestInit) => {
|
||||||
const url = String(input);
|
const url = String(input);
|
||||||
if ((init?.method ?? 'GET') === 'POST') {
|
if ((init?.method ?? 'GET') === 'POST') {
|
||||||
@@ -99,7 +101,7 @@ describe('multi-tab offline queue replay race', () => {
|
|||||||
}));
|
}));
|
||||||
});
|
});
|
||||||
|
|
||||||
it('две вкладки реплеят одну и ту же запись очереди дважды (дубль сообщения)', async () => {
|
it('без Web Locks две вкладки реплеят одну запись дважды (исходная гонка)', async () => {
|
||||||
const useOfflineSyncA = await importFreshHook();
|
const useOfflineSyncA = await importFreshHook();
|
||||||
const useOfflineSyncB = await importFreshHook();
|
const useOfflineSyncB = await importFreshHook();
|
||||||
|
|
||||||
@@ -127,9 +129,57 @@ describe('multi-tab offline queue replay race', () => {
|
|||||||
});
|
});
|
||||||
|
|
||||||
const messagePosts = replayedPosts.filter((p) => p.url === '/api/tasks/1988/messages');
|
const messagePosts = replayedPosts.filter((p) => p.url === '/api/tasks/1988/messages');
|
||||||
console.log(`[race-test] POST /api/tasks/1988/messages выполнен ${messagePosts.length} раз(а) для 1 записи очереди`);
|
console.log(`[race-test] без Web Locks: POST выполнен ${messagePosts.length} раз(а) для 1 записи`);
|
||||||
// БАГ: одна запись очереди уходит дважды (по разу на каждую открытую вкладку).
|
// Исходная гонка (документация бага): без кросс-вкладочной координации
|
||||||
// После фикса (cross-tab lock / атомарное изъятие) ожидается 1.
|
// одна запись уходит по разу на каждую вкладку.
|
||||||
expect(messagePosts.length).toBe(2);
|
expect(messagePosts.length).toBe(2);
|
||||||
});
|
});
|
||||||
|
|
||||||
|
it('с Web Locks (кросс-вкладочный мьютекс) запись реплеится ровно один раз', async () => {
|
||||||
|
// Эмуляция navigator.locks: сериализация запросов по имени (как у браузера)
|
||||||
|
let lockChain: Promise<void> = Promise.resolve();
|
||||||
|
const mockLocks = {
|
||||||
|
request: async (_name: string, cb: () => Promise<unknown>) => {
|
||||||
|
const prev = lockChain;
|
||||||
|
let release!: () => void;
|
||||||
|
lockChain = new Promise<void>((r) => { release = r; });
|
||||||
|
await prev;
|
||||||
|
try {
|
||||||
|
return await cb();
|
||||||
|
} finally {
|
||||||
|
release();
|
||||||
|
}
|
||||||
|
},
|
||||||
|
};
|
||||||
|
Object.defineProperty(navigator, 'locks', { value: mockLocks, configurable: true });
|
||||||
|
|
||||||
|
const useOfflineSyncA = await importFreshHook();
|
||||||
|
const useOfflineSyncB = await importFreshHook();
|
||||||
|
|
||||||
|
sharedQueue.push({
|
||||||
|
id: 'q1',
|
||||||
|
method: 'POST',
|
||||||
|
url: '/api/tasks/1988/messages',
|
||||||
|
body: JSON.stringify({ message: 'тестовое сообщение' }),
|
||||||
|
token: null,
|
||||||
|
orgId: '1',
|
||||||
|
enqueuedAt: Date.now(),
|
||||||
|
});
|
||||||
|
|
||||||
|
const tabA = renderHook(() => useOfflineSyncA(1), { wrapper: makeWrapper() });
|
||||||
|
const tabB = renderHook(() => useOfflineSyncB(1), { wrapper: makeWrapper() });
|
||||||
|
|
||||||
|
await act(async () => {
|
||||||
|
await Promise.all([
|
||||||
|
tabA.result.current.sync(),
|
||||||
|
tabB.result.current.sync(),
|
||||||
|
]);
|
||||||
|
});
|
||||||
|
|
||||||
|
const messagePosts = replayedPosts.filter((p) => p.url === '/api/tasks/1988/messages');
|
||||||
|
console.log(`[race-test] с Web Locks: POST выполнен ${messagePosts.length} раз(а) для 1 записи`);
|
||||||
|
// ФИКС: первая вкладка под локом реплеит и удаляет запись, вторая видит
|
||||||
|
// пустую очередь — ровно один POST.
|
||||||
|
expect(messagePosts.length).toBe(1);
|
||||||
|
});
|
||||||
});
|
});
|
||||||
|
|||||||
Reference in New Issue
Block a user