fix(pxe): widen tracked sender tagging ranges with onchain discovery evidence (#24655)

Fixes the sender tag sync conflict from
https://gist.github.com/nventuro/0aa690736b1d2865e27197723a814e9d,
including the "Window straddle" residual edge.

Discovery could re-derive a different index range for an already tracked
(secret, txHash) pair and hit the `Conflicting range` throw in
`storePendingIndexes`, permanently wedging the secret. Two triggers:

- A same-PXE tx partially reverts: the chain only shows the surviving
non-revertible sub-range of the prove-time entry, and the throw fired
before the finalized receipt step could resolve it.
- A tx straddles a sync window boundary: the window loop assembles its
range piecewise, so window 2 stores a different range than window 1
(latent regardless of reverts).

Fix: discovery passes `mergeExisting` to `storePendingIndexes`, which
widens the stored entry to the union of both ranges (grow-only, driven
by onchain evidence). The finalized receipt step still resolves partial
reverts. The only other writer of pending ranges,
`persistSenderTaggingIndexRangesForTx` (records the indexes a tx sent
from this PXE used, at prove time), does not pass the flag and still
throws on a mismatch: there it indicates a bug, not partial onchain
evidence.

Red-green (all fail on the base with the `Conflicting range` throw):
- partial revert repro: finalizes the surviving index, frees the
squashed ones, repeat sync is a no-op
- cross-sync widen: an entry tracked from an earlier sync grows when
discovery evidences further indexes
- window straddle: drives the actual window advance loop, asserts per
window queried tags and that the widened entry later finalizes cleanly

Store-level tests pin the union semantics: a sub-range keeps the entry,
a range beyond it widens it, an untracked tx still stores.

(cherry picked from commit fd7515d)

Author: vezenovm

yarn-project/pxe/src/storage/tagging_store/sender_tagging_store.test.ts

@@ -31,14 +31,17 @@ describe('SenderTaggingStore', () => {
   });
 
   describe('storePendingIndexes', () => {
-    it('stores a single pending index range', async () => {
+    it.each([
+      ['', 'storePendingIndexes'],
+      [' when merging', 'mergePendingIndexes'],
+    ] as const)('stores a single pending index range for an untracked tx%s', async (_name, method) => {
       const txHash = TxHash.random();
 
-      await taggingStore.storePendingIndexes([range(secret1, 5)], txHash, 'test');
+      await taggingStore[method]([range(secret1, 5)], txHash, 'test');
 
       const txHashes = await taggingStore.getTxHashesOfPendingIndexes(secret1, 0, 10, 'test');
-      expect(txHashes).toHaveLength(1);
-      expect(txHashes[0]).toEqual(txHash);
+      expect(txHashes).toEqual([txHash]);
+      expect(await taggingStore.getLastUsedIndex(secret1, 'test')).toBe(5);
     });
 
     it('stores multiple pending index ranges for different secrets', async () => {
@@ -107,6 +110,32 @@ describe('SenderTaggingStore', () => {
       );
     });
 
+    it('keeps the existing range when merging in a sub-range for the same tx', async () => {
+      const txHash = TxHash.random();
+
+      // Prove-time entry spanning setup and app-logic phase logs.
+      await taggingStore.storePendingIndexes([range(secret1, 4, 6)], txHash, 'test');
+
+      // Discovery of the surviving sub-range of a partially reverted tx must not throw nor shrink the entry.
+      await taggingStore.mergePendingIndexes([range(secret1, 4)], txHash, 'test');
+
+      expect(await taggingStore.getLastUsedIndex(secret1, 'test')).toBe(6);
+    });
+
+    it('widens the existing range to the union when merging in a range beyond it', async () => {
+      const txHash = TxHash.random();
+
+      // A prior window discovered only part of the tx's range.
+      await taggingStore.storePendingIndexes([range(secret1, 4, 6)], txHash, 'test');
+
+      // Discovery evidences further onchain indexes for the same tx — the entry must grow to cover them.
+      await taggingStore.mergePendingIndexes([range(secret1, 7, 8)], txHash, 'test');
+
+      const txHashes = await taggingStore.getTxHashesOfPendingIndexes(secret1, 0, 10, 'test');
+      expect(txHashes).toEqual([txHash]);
+      expect(await taggingStore.getLastUsedIndex(secret1, 'test')).toBe(8);
+    });
+
     it('throws when storing a pending index range lower than the last finalized index', async () => {
       const txHash1 = TxHash.random();
       const txHash2 = TxHash.random();

yarn-project/pxe/src/storage/tagging_store/sender_tagging_store.ts

@@ -128,10 +128,11 @@ export class SenderTaggingStore implements StagedStore {
   }
 
   /**
-   * Stores pending index ranges.
+   * Stores pending index ranges, rejecting any range that disagrees with an already-stored one.
    * @remarks If the same (secret, txHash) pair already exists in the db with an equal range, it's a no-op. This is
    * expected to happen because whenever we start sync we start from the last finalized index and we can have pending
-   * ranges already stored from previous syncs. If the ranges differ, it throws an error as that indicates a bug.
+   * ranges already stored from previous syncs. If the ranges differ, it throws an error as that indicates a bug in
+   * callers that record indexes at prove time. Discovery from onchain logs must use `mergePendingIndexes` instead.
    * @param ranges - The tagging index ranges containing the directional app tagging secrets and the index ranges that are
    * to be stored in the db.
    * @param txHash - The tx in which the tagging indexes were used in private logs.
@@ -141,6 +142,33 @@ export class SenderTaggingStore implements StagedStore {
    * @throws If a different range already exists for the same (secret, txHash) pair.
    */
   storePendingIndexes(ranges: TaggingIndexRange[], txHash: TxHash, jobId: string): Promise<void> {
+    return this.#storePendingIndexes(ranges, txHash, jobId, false);
+  }
+
+  /**
+   * Stores pending index ranges, widening an existing entry for the same (secret, txHash) pair to the union of the
+   * stored and incoming ranges instead of throwing on a mismatch. Discovery from onchain logs needs this: it may see
+   * only the surviving (non-revertible phase) sub-range of a partially reverted tx recorded at prove time (the
+   * finalized receipt step of the sync resolves that difference), or indexes beyond a partially discovered entry
+   * when a tx from another PXE straddles a sync window boundary. Callers that record indexes at prove time must use
+   * `storePendingIndexes` instead, so that a range disagreement surfaces as a bug rather than being absorbed.
+   * @param ranges - The tagging index ranges containing the directional app tagging secrets and the index ranges that are
+   * to be stored in the db.
+   * @param txHash - The tx in which the tagging indexes were used in private logs.
+   * @param jobId - job context for staged writes to this store. See `JobCoordinator` for more details.
+   * @throws If the highestIndex is further than window length from the highest finalized index for the same secret.
+   * @throws If the lowestIndex is lower than or equal to the last finalized index for the same secret.
+   */
+  mergePendingIndexes(ranges: TaggingIndexRange[], txHash: TxHash, jobId: string): Promise<void> {
+    return this.#storePendingIndexes(ranges, txHash, jobId, true);
+  }
+
+  #storePendingIndexes(
+    ranges: TaggingIndexRange[],
+    txHash: TxHash,
+    jobId: string,
+    mergeExisting: boolean,
+  ): Promise<void> {
     if (ranges.length === 0) {
       return Promise.resolve();
     }
@@ -187,21 +215,40 @@ export class SenderTaggingStore implements StagedStore {
         // Check if an entry with the same txHash already exists
         const existingEntry = pendingData.find(entry => entry.txHash === txHashStr);
 
-        if (existingEntry) {
-          // Assert that the ranges are equal — different ranges for the same (secret, txHash) indicates a bug
-          if (existingEntry.lowestIndex !== range.lowestIndex || existingEntry.highestIndex !== range.highestIndex) {
-            throw new Error(
-              `Conflicting range for secret ${secretStr} and txHash ${txHashStr}: ` +
-                `existing [${existingEntry.lowestIndex}, ${existingEntry.highestIndex}] vs ` +
-                `new [${range.lowestIndex}, ${range.highestIndex}]`,
-            );
-          }
-          // Exact duplicate — skip
-        } else {
-          this.#writePendingIndexes(jobId, secretStr, [
+        let updatedPending: PendingIndexesEntry[] | undefined;
+        if (!existingEntry) {
+          updatedPending = [
             ...pendingData,
             { lowestIndex: range.lowestIndex, highestIndex: range.highestIndex, txHash: txHashStr },
-          ]);
+          ];
+        } else if (mergeExisting) {
+          // Widen the entry to the union of both ranges, never shrink it: replacing would drop prove-time
+          // indexes the chain doesn't show (partially reverted tx), and skipping would drop onchain indexes
+          // discovered in a later sync window (tx straddling the window boundary).
+          const lowestIndex = Math.min(existingEntry.lowestIndex, range.lowestIndex);
+          const highestIndex = Math.max(existingEntry.highestIndex, range.highestIndex);
+          if (lowestIndex !== existingEntry.lowestIndex || highestIndex !== existingEntry.highestIndex) {
+            updatedPending = pendingData.map(entry =>
+              entry === existingEntry ? { lowestIndex, highestIndex, txHash: entry.txHash } : entry,
+            );
+          }
+        } else if (
+          existingEntry.lowestIndex !== range.lowestIndex ||
+          existingEntry.highestIndex !== range.highestIndex
+        ) {
+          // Different ranges for the same (secret, txHash) indicate a bug in callers that record indexes at prove
+          // time.
+          throw new Error(
+            `Conflicting range for secret ${secretStr} and txHash ${txHashStr}: ` +
+              `existing [${existingEntry.lowestIndex}, ${existingEntry.highestIndex}] vs ` +
+              `new [${range.lowestIndex}, ${range.highestIndex}]`,
+          );
+        }
+        // Remaining cases (a merge whose union equals the stored range, or an identical range without
+        // mergeExisting): duplicate evidence, nothing to write.
+
+        if (updatedPending) {
+          this.#writePendingIndexes(jobId, secretStr, updatedPending);
         }
       }
     });

yarn-project/pxe/src/tagging/sender_sync/sync_sender_tagging_indexes.test.ts

@@ -394,26 +394,34 @@ describe('syncSenderTaggingIndexes', () => {
     expect(aztecNode.getTxReceipt).toHaveBeenCalledWith(pendingTxHash);
   });
 
-  it('handles a partially reverted transaction', async () => {
+  /**
+   * Same-PXE partial revert: the pending range was recorded at prove time and spans both the non-revertible (setup)
+   * and revertible (app logic) phases. After the tx mines with reverted app logic, only the setup-phase logs are
+   * onchain, so discovery re-derives a narrower range for the same (secret, txHash). That narrower range must not
+   * conflict with the prove-time entry — the finalized receipt step of the sync owns resolving the difference.
+   */
+  it('handles a partially reverted tx whose pending range was recorded at prove time', async () => {
     await setUp();
 
     const revertedTxHash = TxHash.random();
 
-    // Create logs at indexes 4 and 6 for the same (reverted) tx
+    // Prove-time persist: logs at indexes 4 (setup phase) through 6 (app logic phase) under the same secret.
+    await taggingStore.storePendingIndexes(
+      [{ extendedSecret: secret, lowestIndex: 4, highestIndex: 6 }],
+      revertedTxHash,
+      'test',
+    );
+
+    // Only the setup-phase log survived the revert, so the node only knows the tag at index 4.
     const tag4 = await computeSiloedTagForIndex(4);
+    // The app-logic tags at indexes 5 and 6 were squashed by the revert and never reached the chain.
+    const tag5 = await computeSiloedTagForIndex(5);
     const tag6 = await computeSiloedTagForIndex(6);
 
     aztecNode.getPrivateLogsByTags.mockImplementation(query => {
       const tags = query.tags as SiloedTag[];
       return Promise.resolve(
-        tags.map((tag: SiloedTag) => {
-          if (tag.equals(tag4)) {
-            return [makeLog(revertedTxHash, tag4.value)];
-          } else if (tag.equals(tag6)) {
-            return [makeLog(revertedTxHash, tag6.value)];
-          }
-          return [];
-        }),
+        tags.map((tag: SiloedTag) => (tag.equals(tag4) ? [makeLog(revertedTxHash, tag4.value)] : [])),
       );
     });
 
@@ -432,17 +440,157 @@ describe('syncSenderTaggingIndexes', () => {
       [], // contractClassLogs
     );
 
-    // Mock getTxReceipt to return a FINALIZED + REVERTED mined receipt carrying the tx effect. The same receipt
-    // satisfies both the status-classification call and the includeTxEffect follow-up call.
     aztecNode.getTxReceipt.mockResolvedValue(
       mined(revertedTxHash, TxStatus.FINALIZED, 14, TxExecutionResult.REVERTED, txEffect),
     );
 
     await syncSenderTaggingIndexes(secret, aztecNode, taggingStore, MOCK_ANCHOR_BLOCK_HASH, 'test');
 
-    // Index 4 should be finalized (it survived the partial revert)
+    // The surviving index is finalized and the squashed indexes 5-6 are freed for reuse.
     expect(await taggingStore.getLastFinalizedIndex(secret, 'test')).toBe(4);
-    // No pending indexes should remain for this secret
     expect(await taggingStore.getLastUsedIndex(secret, 'test')).toBe(4);
+    // Reconciliation must remove the pending entry entirely — a stale entry would keep resurfacing in later syncs.
+    const pendingAfterSync = await taggingStore.getTxHashesOfPendingIndexes(
+      secret,
+      0,
+      UNFINALIZED_TAGGING_INDEXES_WINDOW_LEN,
+      'test',
+    );
+    expect(pendingAfterSync).toEqual([]);
+
+    // Premise guard: discovery blindly probes every index in the window, so the first sync must have queried the full
+    // prove-time range [4, 6], with only the setup-phase tag getting an onchain answer. If the sync ever stopped
+    // probing these indexes, the discovery merge this test exists to exercise would silently stop happening.
+    const queriedTags = aztecNode.getPrivateLogsByTags.mock.calls.flatMap(([query]) => query.tags as SiloedTag[]);
+    expect(queriedTags.some(tag => tag.equals(tag4))).toBe(true);
+    expect(queriedTags.some(tag => tag.equals(tag5))).toBe(true);
+    expect(queriedTags.some(tag => tag.equals(tag6))).toBe(true);
+
+    // A repeat sync must be a clean no-op: the behavior being pinned here is that the secret will not throw on every
+    // subsequent sync.
+    await syncSenderTaggingIndexes(secret, aztecNode, taggingStore, MOCK_ANCHOR_BLOCK_HASH, 'test');
+
+    expect(await taggingStore.getLastFinalizedIndex(secret, 'test')).toBe(4);
+    expect(await taggingStore.getLastUsedIndex(secret, 'test')).toBe(4);
+  });
+
+  /**
+   * Cross-device straddle: another PXE sharing this directional secret sent a tx, and an earlier window discovered
+   * only part of its index range, so the store already tracks a narrower entry for the same (secret, txHash).
+   * Discovery must widen the entry to cover every index evidenced onchain, so that the next index choice accounts
+   * for them.
+   */
+  it('widens a tracked pending range when discovery evidences further indexes for the same tx', async () => {
+    await setUp();
+
+    const foreignTxHash = TxHash.random();
+
+    // An earlier window discovered only the first index of the foreign tx.
+    await taggingStore.storePendingIndexes(
+      [{ extendedSecret: secret, lowestIndex: 10, highestIndex: 10 }],
+      foreignTxHash,
+      'test',
+    );
+
+    // The chain shows the tx actually used indexes 10 and 11.
+    const tag10 = await computeSiloedTagForIndex(10);
+    const tag11 = await computeSiloedTagForIndex(11);
+
+    aztecNode.getPrivateLogsByTags.mockImplementation(query => {
+      const tags = query.tags as SiloedTag[];
+      return Promise.resolve(
+        tags.map((tag: SiloedTag) => {
+          if (tag.equals(tag10)) {
+            return [makeLog(foreignTxHash, tag10.value)];
+          } else if (tag.equals(tag11)) {
+            return [makeLog(foreignTxHash, tag11.value)];
+          }
+          return [];
+        }),
+      );
+    });
+
+    // The tx is mined but not yet finalized, so no receipt status change resolves the entry during this sync.
+    aztecNode.getTxReceipt.mockResolvedValue(mined(foreignTxHash, TxStatus.PROPOSED, 14));
+
+    await syncSenderTaggingIndexes(secret, aztecNode, taggingStore, MOCK_ANCHOR_BLOCK_HASH, 'test');
+
+    // The next index choice must account for the onchain tag at index 11.
+    expect(await taggingStore.getLastUsedIndex(secret, 'test')).toBe(11);
+    expect(await taggingStore.getLastFinalizedIndex(secret, 'test')).toBeUndefined();
+  });
+
+  /**
+   * Single-sync window straddle: a foreign tx's tags span the boundary between two consecutive sync windows, so the
+   * window loop assembles the tx's range piecewise — window 1 stores the lower index, window 2 evidences the higher
+   * one for the same (secret, txHash). The second write must widen the entry from window 1 rather than conflict with
+   * it, and the final index choice must cover the full onchain range.
+   */
+  it('assembles a pending range piecewise when a tx straddles the sync window boundary', async () => {
+    await setUp();
+
+    // A tx finalized at index 0 makes the finalized index advance during window 1, so the loop proceeds to window 2.
+    const finalizedTxHash = TxHash.random();
+    const finalizedTag = await computeSiloedTagForIndex(0);
+
+    // The straddling tx used the last index of window 1 and the first index of window 2.
+    const straddlingTxHash = TxHash.random();
+    const lowerStraddleIndex = UNFINALIZED_TAGGING_INDEXES_WINDOW_LEN - 1;
+    const upperStraddleIndex = UNFINALIZED_TAGGING_INDEXES_WINDOW_LEN;
+    const lowerStraddleTag = await computeSiloedTagForIndex(lowerStraddleIndex);
+    const upperStraddleTag = await computeSiloedTagForIndex(upperStraddleIndex);
+
+    aztecNode.getPrivateLogsByTags.mockImplementation(query => {
+      const tags = query.tags as SiloedTag[];
+      return Promise.resolve(
+        tags.map((tag: SiloedTag) => {
+          if (tag.equals(finalizedTag)) {
+            return [makeLog(finalizedTxHash, finalizedTag.value)];
+          } else if (tag.equals(lowerStraddleTag)) {
+            return [makeLog(straddlingTxHash, lowerStraddleTag.value)];
+          } else if (tag.equals(upperStraddleTag)) {
+            return [makeLog(straddlingTxHash, upperStraddleTag.value)];
+          }
+          return [];
+        }),
+      );
+    });
+
+    aztecNode.getTxReceipt.mockImplementation((hash: TxHash) => {
+      if (hash.equals(finalizedTxHash)) {
+        return Promise.resolve(mined(hash, TxStatus.FINALIZED, 14));
+      } else if (hash.equals(straddlingTxHash)) {
+        // Mined but not finalized, so no receipt status change resolves the straddling entry during this sync.
+        return Promise.resolve(mined(hash, TxStatus.PROPOSED, 16));
+      }
+      throw new Error(`Unexpected tx hash: ${hash.toString()}`);
+    });
+
+    await syncSenderTaggingIndexes(secret, aztecNode, taggingStore, MOCK_ANCHOR_BLOCK_HASH, 'test');
+
+    // The straddled range must have been assembled piecewise: window 1's logs query covers the lower straddle index
+    // and window 2's the upper. (Each window fits in a single RPC page, so there is one logs call per window.)
+    expect(aztecNode.getPrivateLogsByTags).toHaveBeenCalledTimes(2);
+    const queriedTags = aztecNode.getPrivateLogsByTags.mock.calls.map(([query]) => query.tags as SiloedTag[]);
+    expect(queriedTags[0].some(tag => tag.equals(lowerStraddleTag))).toBe(true);
+    expect(queriedTags[1].some(tag => tag.equals(upperStraddleTag))).toBe(true);
+
+    expect(await taggingStore.getLastFinalizedIndex(secret, 'test')).toBe(0);
+    // The next index choice must account for both straddled onchain tags.
+    expect(await taggingStore.getLastUsedIndex(secret, 'test')).toBe(upperStraddleIndex);
+
+    // The straddling tx later finalizes. A single widened entry finalizes cleanly at the upper index — a duplicate
+    // entry for the same txHash would instead trip the multiple-pending-entries guard during finalization.
+    aztecNode.getTxReceipt.mockImplementation((hash: TxHash) => {
+      if (hash.equals(straddlingTxHash)) {
+        return Promise.resolve(mined(hash, TxStatus.FINALIZED, 18));
+      }
+      throw new Error(`Unexpected tx hash: ${hash.toString()}`);
+    });
+
+    await syncSenderTaggingIndexes(secret, aztecNode, taggingStore, MOCK_ANCHOR_BLOCK_HASH, 'test');
+
+    expect(await taggingStore.getLastFinalizedIndex(secret, 'test')).toBe(upperStraddleIndex);
+    expect(await taggingStore.getLastUsedIndex(secret, 'test')).toBe(upperStraddleIndex);
   });
 });

yarn-project/pxe/src/tagging/sender_sync/sync_sender_tagging_indexes.ts

@@ -74,7 +74,7 @@ export async function syncSenderTaggingIndexes(
     }
 
     // Receipts for pending tx hashes that the logs query just surfaced still need a sequential follow-up call.
-    // `storePendingIndexes` is idempotent on (secret, txHash), so a re-discovered hash stays classified as known
+    // `mergePendingIndexes` is idempotent on (secret, txHash), so a re-discovered hash stays classified as known
     // and is not re-fetched here.
     const knownSet = new Set(knownPendingTxHashes.map(h => h.toString()));
     const newPendingTxHashes = allPendingTxHashes.filter(h => !knownSet.has(h.toString()));