fix: existing messages retry guards

src/messageOperations/MessageOperationStatePolicy.ts

@@ -206,6 +206,10 @@ export class MessageOperationStatePolicy {
     return unhandledKind(kind);
   }
 
+  /**
+   * Records a failed request. Resolves `true` when the error shows the request took effect on the
+   * server, so the operation succeeded after all.
+   */
   async failure<K extends OperationKind>({
     error,
     kind,
@@ -220,17 +224,22 @@ export class MessageOperationStatePolicy {
     messageId: string;
     optimistic?: OptimisticOutcome;
     options?: MessageOperationSpec[K]['options'];
-  }) {
+  }): Promise<boolean> {
     const parsed = parseError(error);
 
-    if (isAlreadyExistsError(error, parsed)) {
+    // The server has a message with this ID, so an earlier attempt landed and the send succeeded.
+    if ((kind === 'send' || kind === 'retry') && isAlreadyExistsError(error, parsed)) {
       const existing = this.ctx.get(messageId);
       if (existing?.status === 'sending') {
-        const received: LocalMessage = { ...localMessage, status: 'received' };
+        const received: LocalMessage = {
+          ...localMessage,
+          error: undefined,
+          status: 'received',
+        };
         this.ctx.ingest(received);
         this.ctx.persist(received);
       }
-      return;
+      return true;
     }
 
     // A request waiting in the offline queue has not failed: the optimistic state stays.
@@ -241,11 +250,11 @@ export class MessageOperationStatePolicy {
 
     if (kind === 'delete') {
       if (!queued) this.revertDelete({ messageId, optimistic, options });
-      return;
+      return false;
     }
 
     if (kind === 'update') {
-      if (queued) return;
+      if (queued) return false;
       // The edit is kept and the failure recorded on it; `MessageOperations.retry` replays it as an
       // edit.
       const failed: LocalMessage = {
@@ -255,14 +264,16 @@ export class MessageOperationStatePolicy {
       };
       this.ctx.ingest(failed);
       this.ctx.persist(failed);
-      return;
+      return false;
     }
 
     if (kind === 'send' || kind === 'retry') {
+      // the server's copy arrived while the request was open, so the send succeeded.
+      if (this.ctx.get(messageId)?.status === 'received') return true;
       const failed: LocalMessage = { ...localMessage, error: parsed, status: 'failed' };
       this.ctx.ingest(failed);
       this.ctx.persist(failed);
-      return;
+      return false;
     }
 
     return unhandledKind(kind);

src/messageOperations/MessageOperations.ts

@@ -201,15 +201,15 @@ export class MessageOperations {
         options: params.options,
       });
     } catch (e) {
-      await this.policy.failure({
+      const succeeded = await this.policy.failure({
         error: e,
         kind,
         localMessage: effective.localMessage,
         messageId,
         optimistic,
         options: params.options,
       });
-      throw e;
+      if (!succeeded) throw e;
     }
   }
 

src/offline-support/offline_support_api.ts

@@ -1147,6 +1147,11 @@ export abstract class AbstractOfflineDB implements OfflineDBApi {
           true,
         );
       } catch (e) {
+        const operation = QUEUEABLE_OPERATIONS[task.type] as QueueableOperation<
+          typeof task.type
+        >;
+        operation.onReplayRejected?.({ client: this.client, error: e, task });
+
         const error = e as AxiosError<APIError>;
         if (!this.shouldSkipQueueingTask(error)) {
           continue;

src/offline-support/queueableOperations.ts

@@ -1,7 +1,9 @@
 import { chatLoggerSystem } from '../logger';
 import { formatMessage } from '../utils';
+import { isAlreadyExistsError, parseError } from '../errors';
 import type { Channel } from '../channel';
 import type { StreamChat } from '../client';
+import type { LocalMessage } from '../types';
 import type { PendingTask, PendingTaskOf, QueueableResult, QueueableType } from './types';
 
 const logger = chatLoggerSystem.getLogger('offline-db');
@@ -31,6 +33,12 @@ export type QueueableOperation<T extends QueueableType> = {
     result: QueueableResult<T>;
     task: PendingTaskOf<T>;
   }) => void;
+  /** Settles local state after a replay the server rejected for whatever reason. */
+  onReplayRejected?: (params: {
+    client: StreamChat;
+    error: unknown;
+    task: PendingTaskOf<T>;
+  }) => void;
 };
 
 /**
@@ -102,6 +110,24 @@ export const QUEUEABLE_OPERATIONS: {
       }
       channelOf(client, task).messagePaginator.trackLatestMessage(formatMessage(message));
     },
+    // the server has the message, meaning an earlier attempt landed but its response was lost
+    onReplayRejected: ({ client, error, task }) => {
+      const store = client.messageStore;
+      const held = store.get(task.messageId);
+
+      if (held?.status === 'failed' && isAlreadyExistsError(error, parseError(error))) {
+        const received: LocalMessage = { ...held, error: undefined, status: 'received' };
+        store.upsert(received);
+        store.flushSubscribers(received.id);
+        client.offlineDb?.executeQuerySafely(
+          (db) =>
+            db.upsertMessageWithChannelGuard({
+              message: { ...received, cid: `${task.channelType}:${task.channelId}` },
+            }),
+          { method: 'sendMessage:replayRejected' },
+        );
+      }
+    },
     run: ({ channel, client, task }) =>
       channelOf(client, task, channel).sendMessage(...task.payload),
   },

test/unit/MessageComposer/messageComposer.test.ts

@@ -3379,6 +3379,45 @@ describe('MessageComposer submit lifecycle', () => {
     );
   });
 
+  it('reports a send the server already has as sent', async () => {
+    const { messageComposer, mockChannel } = setup();
+    withText(messageComposer);
+    const addError = vi.spyOn(messageComposer.client.notifications, 'addError');
+    let messageId: string | undefined;
+    stubSendRequest(mockChannel, async ({ localMessage }) => {
+      messageId = localMessage.id;
+      throw Object.assign(
+        new Error(
+          `SendMessage failed with error: "a message with ID ${messageId} already exists"`,
+        ),
+        { code: 4 },
+      );
+    });
+
+    await expect(messageComposer.send()).resolves.toBe('sent');
+
+    expect(addError).not.toHaveBeenCalled();
+    expect(mockChannel.messagePaginator.getItem(messageId!)?.status).toBe('received');
+  });
+
+  it('reports a send whose server copy arrived before a timeout as sent', async () => {
+    const { messageComposer, mockChannel } = setup();
+    withText(messageComposer);
+    const addError = vi.spyOn(messageComposer.client.notifications, 'addError');
+    let messageId: string | undefined;
+    stubSendRequest(mockChannel, async ({ localMessage }) => {
+      messageId = localMessage.id;
+      // The server's `message.new` arrives before the client gives up on the request.
+      mockChannel.messagePaginator.ingestItem({ ...localMessage, status: 'received' });
+      throw new Error('timeout of 3000ms exceeded');
+    });
+
+    await expect(messageComposer.send()).resolves.toBe('sent');
+
+    expect(addError).not.toHaveBeenCalled();
+    expect(mockChannel.messagePaginator.getItem(messageId!)?.status).toBe('received');
+  });
+
   it('returns false and sends nothing when there is nothing to compose', async () => {
     const { messageComposer, mockChannel } = setup();
     const send = vi

test/unit/messageOperations/MessageOperations.test.ts

@@ -127,10 +127,96 @@ describe('MessageOperations', () => {
 
     const localMessage = makeLocalMessage({ id: 'm1', status: 'failed' });
 
-    await expect(ops.send({ localMessage })).rejects.toThrow();
+    await expect(ops.send({ localMessage })).resolves.toBeUndefined();
     expect(store.get('m1')?.status).toBe('received');
   });
 
+  it('resolves a retry that the server rejects with "already exists"', async () => {
+    const store: Store = new Map();
+    let sendCalls = 0;
+
+    const ops = new MessageOperations({
+      ...stateHooks(store),
+      ingest: (m) => store.set(m.id, m),
+      get: (id) => store.get(id),
+      handlers: () => ({}),
+      defaults: {
+        delete: defaultDelete,
+        send: async () => {
+          sendCalls += 1;
+          // The first request landed, but its response was lost.
+          if (sendCalls === 1) throw new Error('timeout');
+          throw Object.assign(new Error('message already exists'), { code: 4 });
+        },
+        update: async () => ({ message: makeMessageResponse({ id: 'm1' }) }),
+      },
+    });
+
+    const localMessage = makeLocalMessage({ id: 'm1', status: 'failed' });
+
+    await expect(ops.send({ localMessage })).rejects.toThrow('timeout');
+    expect(store.get('m1')?.status).toBe('failed');
+
+    await expect(ops.retry({ localMessage: store.get('m1')! })).resolves.toBeUndefined();
+    expect(store.get('m1')?.status).toBe('received');
+    expect(store.get('m1')?.error).toBeUndefined();
+  });
+
+  it('rejects an edit that fails with the "already exists" code', async () => {
+    const store: Store = new Map();
+    store.set('m1', makeLocalMessage({ id: 'm1', status: 'received' }));
+
+    const ops = new MessageOperations({
+      ...stateHooks(store),
+      ingest: (m) => store.set(m.id, m),
+      get: (id) => store.get(id),
+      handlers: () => ({}),
+      defaults: {
+        delete: defaultDelete,
+        send: async () => ({ message: makeMessageResponse({ id: 'm1' }) }),
+        update: async () => {
+          throw Object.assign(new Error('message already exists'), { code: 4 });
+        },
+      },
+    });
+
+    await expect(
+      ops.update({
+        localMessage: makeLocalMessage({ id: 'm1', status: 'received', text: 'edit' }),
+      }),
+    ).rejects.toThrow('already exists');
+    expect(store.get('m1')?.status).toBe('failed');
+  });
+
+  it('resolves a send whose server copy arrived before the request failed', async () => {
+    const store: Store = new Map();
+
+    const ops = new MessageOperations({
+      ...stateHooks(store),
+      ingest: (m) => store.set(m.id, m),
+      get: (id) => store.get(id),
+      handlers: () => ({}),
+      defaults: {
+        delete: defaultDelete,
+        send: async () => {
+          // The server saved the message and its `message.new` arrived, then the client timed out.
+          store.set(
+            'm1',
+            makeLocalMessage({ id: 'm1', status: 'received', text: 'from the server' }),
+          );
+          throw new Error('timeout of 3000ms exceeded');
+        },
+        update: async () => ({ message: makeMessageResponse({ id: 'm1' }) }),
+      },
+    });
+
+    const localMessage = makeLocalMessage({ id: 'm1', status: 'failed' });
+
+    await expect(ops.send({ localMessage })).resolves.toBeUndefined();
+    expect(store.get('m1')?.status).toBe('received');
+    expect(store.get('m1')?.text).toBe('from the server');
+  });
+
   it('marks as failed on non-duplicate error', async () => {
     const store: Store = new Map();
 

test/unit/offline-support/offline_support_api.test.ts

@@ -18,6 +18,7 @@ import {
   StableWSConnection,
 } from '../../../src';
 import { chatLoggerSystem } from '../../../src/logger';
+import { QUEUEABLE_OPERATIONS } from '../../../src/offline-support/queueableOperations';
 
 import { generateChannel } from '../test-utils/generateChannel';
 import { generateReadResponse } from '../test-utils/generateReadResponse';
@@ -3033,6 +3034,20 @@ describe('OfflineSupportApi', () => {
           expect(deletePendingTaskSpy).toHaveBeenCalledWith({ id: 2 });
         });
 
+        it('hands a rejected replay to its operation', async () => {
+          const onReplayRejected = vi.spyOn(
+            QUEUEABLE_OPERATIONS['send-message'],
+            'onReplayRejected',
+          );
+          executeTaskSpy.mockRejectedValueOnce(skippableError);
+
+          await offlineDb.executePendingTasks();
+
+          expect(onReplayRejected).toHaveBeenCalledWith(
+            expect.objectContaining({ error: skippableError, task: task1 }),
+          );
+        });
+
         it('should skip tasks with no id', async () => {
           const taskWithoutId = generatePendingTask('delete-reaction', 1, {
             id: undefined,