fix(mobile): batch notification catch-up ledger writes (#34785)

* fix(mobile): batch notification catch-up ledger writes

* test(mobile): use a fixed expiry in catch-up storage test

---------

Co-authored-by: OrcaWin <293788423+OrcaWin@users.noreply.github.com>

mobile/src/notifications/push-dismissal-catchup-storage.test.ts

@@ -0,0 +1,125 @@
+import { beforeEach, expect, it, vi } from 'vitest'
+import * as Notifications from 'expo-notifications'
+import { requestNotificationCatchup } from './push-dismissal-reconciliation'
+import { deriveHostFingerprint } from './push-host-fingerprint'
+import { wasPushDismissed } from './push-dismissal-watermarks'
+import type { MobilePushDismissalRpcSender } from './push-dismissal-operations'
+
+const disk = vi.hoisted(
+  (): {
+    values: Map<string, string>
+    reads: number
+    writtenRows: number[]
+  } => ({
+    values: new Map<string, string>(),
+    reads: 0,
+    writtenRows: []
+  })
+)
+vi.mock('@react-native-async-storage/async-storage', () => ({
+  default: {
+    getItem: async (key: string) => {
+      disk.reads++
+      return disk.values.get(key) ?? null
+    },
+    setItem: async (key: string, value: string) => {
+      const entries: unknown = JSON.parse(value)
+      if (!Array.isArray(entries)) {
+        throw new Error('Expected dismissal ledger')
+      }
+      disk.writtenRows.push(entries.length)
+      disk.values.set(key, value)
+    }
+  }
+}))
+vi.mock('expo-notifications', () => ({
+  getPresentedNotificationsAsync: vi.fn(),
+  dismissNotificationAsync: vi.fn(async () => {})
+}))
+vi.mock('../transport/host-store', () => ({
+  loadHostCatalog: async () => [{ id: 'host-a', publicKeyB64 }]
+}))
+
+const publicKeyB64 = Buffer.alloc(32, 1).toString('base64')
+const hostFingerprint = deriveHostFingerprint(publicKeyB64)!
+const ledgerKey = 'orca:pushDismissalWatermarks:v1'
+
+function payload(index: number) {
+  return {
+    hostFingerprint,
+    notificationId: `note-${index}`,
+    notificationEpoch: 'epoch-a',
+    notificationSeq: 2
+  }
+}
+
+function notification(index: number): Notifications.Notification {
+  return {
+    date: 0,
+    request: {
+      identifier: `tray-${index}`,
+      trigger: null,
+      content: {
+        title: null,
+        subtitle: null,
+        body: null,
+        sound: null,
+        categoryIdentifier: null,
+        data: payload(index)
+      }
+    }
+  }
+}
+
+beforeEach(() => {
+  disk.values.clear()
+  disk.reads = 0
+  disk.writtenRows = []
+  vi.clearAllMocks()
+})
+
+it.each([8, 80])(
+  'persists a confirmed %i-notification catch-up page with one ledger rewrite',
+  async (count) => {
+    const previous = { ...payload(-1), hostFingerprint: 'other-host' }
+    disk.values.set(
+      ledgerKey,
+      JSON.stringify([
+        {
+          key: JSON.stringify([
+            previous.hostFingerprint,
+            previous.notificationEpoch,
+            previous.notificationId
+          ]),
+          seq: previous.notificationSeq,
+          expiresAt: Number.MAX_SAFE_INTEGER
+        }
+      ])
+    )
+    vi.mocked(Notifications.getPresentedNotificationsAsync).mockResolvedValue(
+      Array.from({ length: count }, (_, index) => notification(index))
+    )
+    const sender: MobilePushDismissalRpcSender = {
+      sendRequest: async () => ({
+        id: 'catch-up',
+        ok: true,
+        result: { dismissedPushes: Array.from({ length: count }, (_, index) => payload(index)) }
+      })
+    }
+    await requestNotificationCatchup(sender, 'host-a', () => false)
+    const measured = {
+      ledgerReads: disk.reads,
+      ledgerWrites: disk.writtenRows.length,
+      serializedRows: disk.writtenRows.reduce((total, rows) => total + rows, 0)
+    }
+    expect(measured).toEqual({ ledgerReads: 1, ledgerWrites: 1, serializedRows: count + 1 })
+    expect(vi.mocked(Notifications.dismissNotificationAsync).mock.calls.map(([id]) => id)).toEqual(
+      Array.from({ length: count }, (_, index) => `tray-${index}`)
+    )
+    for (const index of [0, count - 1]) {
+      expect(await wasPushDismissed(payload(index))).toBe(true)
+      expect(await wasPushDismissed({ ...payload(index), notificationSeq: 3 })).toBe(false)
+    }
+    expect(await wasPushDismissed(previous)).toBe(true)
+  }
+)

mobile/src/notifications/push-dismissal-native-races.test.ts

@@ -1,7 +1,11 @@
 import { beforeEach, expect, it, vi } from 'vitest'
 import AsyncStorage from '@react-native-async-storage/async-storage'
 import { nativePushDismissal } from './native-push-dismissal'
-import { rememberPushDismissal, wasPushDismissed } from './push-dismissal-watermarks'
+import {
+  rememberPushDismissal,
+  rememberPushDismissals,
+  wasPushDismissed
+} from './push-dismissal-watermarks'
 import { foregroundNotificationBehavior } from './push-receive'
 import { loadNotificationDeliveryPreferences } from './notification-delivery-preferences'
 
@@ -74,6 +78,17 @@ it('uses only native storage for iOS dismissal reads and writes', async () => {
   expect(AsyncStorage.setItem).not.toHaveBeenCalled()
 })
 
+it('keeps a batch in the native ledger and joins it before checking dismissal', async () => {
+  const other = { ...fence, notificationId: 'second' }
+  const pending = rememberPushDismissals([fence, other])
+  const secondDismissed = wasPushDismissed(other)
+  await pending
+  expect(await secondDismissed).toBe(true)
+  expect(await wasPushDismissed({ ...other, notificationSeq: 22 })).toBe(false)
+  expect(AsyncStorage.getItem).not.toHaveBeenCalled()
+  expect(AsyncStorage.setItem).not.toHaveBeenCalled()
+})
+
 it('rechecks a negative native snapshot overtaken by a dismissal', async () => {
   let finish!: () => void
   vi.mocked(nativePushDismissal!.wasDismissed).mockImplementationOnce(async () => {

mobile/src/notifications/push-dismissal-reconciliation.ts

@@ -5,7 +5,7 @@ import { resolveHostIdForFingerprint } from './push-host-fingerprint'
 import { readNativeNotificationData } from './native-notification-data'
 import { readOrcaPushPayload, type OrcaPushPayload } from './push-payload'
 import { dismissRememberedPushNotifications } from './push-tray-dismissal'
-import { rememberPushDismissal } from './push-dismissal-watermarks'
+import { rememberPushDismissals } from './push-dismissal-watermarks'
 import {
   readPushNotificationIdentity,
   type PushNotificationIdentity
@@ -70,13 +70,15 @@ export async function requestNotificationCatchup(
       const id = readPushNotificationIdentity(raw)
       const payload = id ? requested.get(key(id)) : undefined
       if (payload && id) {
-        await rememberPushDismissal(payload)
         confirmed.push(payload)
         requested.delete(key(id))
       }
     }
-    if (confirmed.length && !isDisposed()) {
-      await dismissRememberedPushNotifications(confirmed[0]!.hostFingerprint, confirmed)
+    if (confirmed.length) {
+      await rememberPushDismissals(confirmed)
+      if (!isDisposed()) {
+        await dismissRememberedPushNotifications(confirmed[0]!.hostFingerprint, confirmed)
+      }
     }
   }
 }

mobile/src/notifications/push-dismissal-watermarks.test.ts

@@ -9,7 +9,11 @@ vi.mock('@react-native-async-storage/async-storage', () => ({
     }
   }
 }))
-import { rememberPushDismissal, wasPushDismissed } from './push-dismissal-watermarks'
+import {
+  rememberPushDismissal,
+  rememberPushDismissals,
+  wasPushDismissed
+} from './push-dismissal-watermarks'
 const payload = {
   hostFingerprint: 'host-a',
   notificationEpoch: 'epoch-a',
@@ -44,6 +48,29 @@ it('serializes concurrent dismissals and never lowers a watermark', async () =>
   expect(await wasPushDismissed({ ...payload, notificationId: 'other' })).toBe(true)
 })
 
+it('merges batched and concurrent fences without lowering duplicates or accepting invalid ones', async () => {
+  await rememberPushDismissal({ ...payload, notificationSeq: 5 })
+  await Promise.all([
+    rememberPushDismissals([
+      payload,
+      { ...payload, notificationSeq: 4 },
+      { ...payload, notificationId: 'other' },
+      { ...payload, notificationId: 'invalid', notificationSeq: -1 }
+    ]),
+    rememberPushDismissal({ ...payload, notificationSeq: 7 })
+  ])
+  expect(await wasPushDismissed({ ...payload, notificationSeq: 7 })).toBe(true)
+  expect(await wasPushDismissed({ ...payload, notificationSeq: 8 })).toBe(false)
+  expect(await wasPushDismissed({ ...payload, notificationId: 'other' })).toBe(true)
+  expect(await wasPushDismissed({ ...payload, notificationId: 'invalid' })).toBe(false)
+})
+
+it('does no storage work for an empty or entirely invalid batch', async () => {
+  await rememberPushDismissals([])
+  await rememberPushDismissals([{ ...payload, notificationEpoch: undefined }])
+  expect(AsyncStorage.getItem).not.toHaveBeenCalled()
+})
+
 it('expires retained metadata and ignores unversioned dismissals', async () => {
   vi.useFakeTimers()
   await rememberPushDismissal(payload)

mobile/src/notifications/push-dismissal-watermarks.ts

@@ -56,26 +56,32 @@ async function readEntries(): Promise<Entry[]> {
 }
 
 export async function rememberPushDismissal(payload: OrcaPushPayload): Promise<void> {
-  const key = eventKey(payload)
-  if (!key) {
+  return rememberPushDismissals([payload])
+}
+
+export async function rememberPushDismissals(payloads: readonly OrcaPushPayload[]): Promise<void> {
+  const valid = payloads.flatMap((payload) => {
+    const key = eventKey(payload)
+    return key === null ? [] : [{ key, payload }]
+  })
+  if (valid.length === 0) {
     return
   }
   return queueDismissalOperation(async () => {
     if (nativePushDismissal) {
-      await nativePushDismissal.remember(payload)
+      for (const { payload } of valid) {
+        await nativePushDismissal.remember(payload)
+      }
       return
     }
-    const entries = await readEntries()
-    const previous = entries.find((entry) => entry.key === key)
-    const entry = {
-      key,
-      seq: Math.max(previous?.seq ?? 0, payload.notificationSeq!),
-      expiresAt: Date.now() + RETENTION_MS
+    const entries = new Map((await readEntries()).map((entry) => [entry.key, entry]))
+    const expiresAt = Date.now() + RETENTION_MS
+    for (const { key, payload } of valid) {
+      const seq = Math.max(entries.get(key)?.seq ?? 0, payload.notificationSeq!)
+      entries.delete(key)
+      entries.set(key, { key, seq, expiresAt })
     }
-    await AsyncStorage.setItem(
-      STORAGE_KEY,
-      JSON.stringify([...entries.filter((item) => item.key !== key), entry])
-    )
+    await AsyncStorage.setItem(STORAGE_KEY, JSON.stringify([...entries.values()]))
   })
 }