fix(pxe): cap fresh secret pending tag indexes to the probed window

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

@@ -12,8 +12,8 @@ import {
 import { randomAppTaggingSecret } from '@aztec/stdlib/testing';
 import { TxEffect, TxHash } from '@aztec/stdlib/tx';
 
-import { UNFINALIZED_TAGGING_INDEXES_WINDOW_LEN } from '../../tagging/constants.js';
-import { SenderTaggingStore } from './sender_tagging_store.js';
+import { UNFINALIZED_TAGGING_INDEXES_WINDOW_LEN, unfinalizedTaggingIndexesWindowEnd } from '../../tagging/constants.js';
+import { SenderTaggingStore, windowExceededError } from './sender_tagging_store.js';
 
 /** Helper to create a single-index range (lowestIndex === highestIndex). */
 function range(secret: AppTaggingSecret, lowest: number, highest?: number): TaggingIndexRange {
@@ -196,7 +196,7 @@ describe('SenderTaggingStore', () => {
         await expect(
           taggingStore.storePendingIndexes([range(secret1, indexBeyondWindow)], txHash2, 'test'),
         ).rejects.toThrow(
-          `Highest used index ${indexBeyondWindow} is further than window length from the highest finalized index ${finalizedIndex}`,
+          windowExceededError(indexBeyondWindow, unfinalizedTaggingIndexesWindowEnd(finalizedIndex), finalizedIndex),
         );
       });
 
@@ -241,16 +241,36 @@ describe('SenderTaggingStore', () => {
 
       it('throws after pending txs exhaust window', async () => {
         // One single-index pending tx per index, mirroring how an un-mined backlog accumulates one log per tx on a
-        // shared secret (e.g. the self-send chain in bench_build_block). A fresh secret treats the
-        // finalized floor as 0, so indexes 0..WINDOW fit...
-        for (let i = 0; i <= UNFINALIZED_TAGGING_INDEXES_WINDOW_LEN; i++) {
+        // shared secret (e.g. the self-send chain in bench_build_block). With no index finalized yet, exactly
+        // WINDOW_LEN indexes (0..WINDOW_LEN - 1) fit...
+        for (let i = 0; i < UNFINALIZED_TAGGING_INDEXES_WINDOW_LEN; i++) {
           await taggingStore.storePendingIndexes([range(secret1, i)], TxHash.random(), 'test');
         }
 
         // ...and the next tx throws, even with a single additional tag.
         await expect(
           taggingStore.storePendingIndexes(
-            [range(secret1, UNFINALIZED_TAGGING_INDEXES_WINDOW_LEN + 1)],
+            [range(secret1, UNFINALIZED_TAGGING_INDEXES_WINDOW_LEN)],
+            TxHash.random(),
+            'test',
+          ),
+        ).rejects.toThrow(/no index finalized yet/);
+      });
+
+      it('permits exactly WINDOW_LEN pending indexes for a fresh secret', async () => {
+        // Fresh-secret counterpart of the two boundary tests above: with no index finalized yet, the last permitted
+        // pending index is WINDOW_LEN - 1, the same WINDOW_LEN-sized allowance as after any real finalization.
+        await expect(
+          taggingStore.storePendingIndexes(
+            [range(secret1, 0, UNFINALIZED_TAGGING_INDEXES_WINDOW_LEN - 1)],
+            TxHash.random(),
+            'test',
+          ),
+        ).resolves.not.toThrow();
+
+        await expect(
+          taggingStore.storePendingIndexes(
+            [range(secret1, UNFINALIZED_TAGGING_INDEXES_WINDOW_LEN)],
             TxHash.random(),
             'test',
           ),

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

@@ -3,7 +3,7 @@ import { AppTaggingSecret, SiloedTag, type TaggingIndexRange } from '@aztec/stdl
 import { TxEffect, TxHash } from '@aztec/stdlib/tx';
 
 import type { StagedStore } from '../../job_coordinator/job_coordinator.js';
-import { UNFINALIZED_TAGGING_INDEXES_WINDOW_LEN } from '../../tagging/constants.js';
+import { UNFINALIZED_TAGGING_INDEXES_WINDOW_LEN, unfinalizedTaggingIndexesWindowEnd } from '../../tagging/constants.js';
 
 /** Internal representation of a pending index range entry. */
 type PendingIndexesEntry = { lowestIndex: number; highestIndex: number; txHash: string };
@@ -195,13 +195,9 @@ export class SenderTaggingStore implements StagedStore {
 
       // Process in memory and validate
       for (const { range, secretStr, pendingData, finalizedIndex } of rangeData) {
-        // Check that the highest index is not further than window length from the highest finalized index.
-        if (range.highestIndex > (finalizedIndex ?? 0) + UNFINALIZED_TAGGING_INDEXES_WINDOW_LEN) {
-          throw new Error(
-            `Highest used index ${range.highestIndex} is further than window length from the highest finalized index ${finalizedIndex ?? 0}.
-            Tagging window length ${UNFINALIZED_TAGGING_INDEXES_WINDOW_LEN} is configured too low. Contact the Aztec team
-            to increase it!`,
-          );
+        const windowEnd = unfinalizedTaggingIndexesWindowEnd(finalizedIndex);
+        if (range.highestIndex >= windowEnd) {
+          throw windowExceededError(range.highestIndex, windowEnd, finalizedIndex);
         }
 
         // Throw if the lowest index is lower than or equal to the last finalized index
@@ -521,3 +517,18 @@ export class SenderTaggingStore implements StagedStore {
     }
   }
 }
+
+/** Builds the error thrown when a pending tag index is at or past the unfinalized tagging window end. */
+export function windowExceededError(
+  highestIndex: number,
+  windowEnd: number,
+  finalizedIndex: number | undefined,
+): Error {
+  const finalizedDescription =
+    finalizedIndex === undefined ? 'no index finalized yet' : `highest finalized index ${finalizedIndex}`;
+  return new Error(
+    `Highest used index ${highestIndex} is at or past the window end ${windowEnd} (${finalizedDescription}). ` +
+      `Tagging window length ${UNFINALIZED_TAGGING_INDEXES_WINDOW_LEN} is configured too low. ` +
+      `Contact the Aztec team to increase it!`,
+  );
+}

yarn-project/pxe/src/tagging/constants.ts

@@ -19,6 +19,18 @@ import { MAX_PRIVATE_LOGS_PER_TX } from '@aztec/constants';
 // MAX_PRIVATE_LOGS_PER_TX. No fixed window value closes that gap.
 export const UNFINALIZED_TAGGING_INDEXES_WINDOW_LEN = MAX_PRIVATE_LOGS_PER_TX + 20;
 
+/**
+ * Exclusive upper bound of the tag indexes that can exist for a secret whose highest finalized index is
+ * `finalizedIndex` (undefined when nothing is finalized yet). The sender store refuses pending indexes at or past it,
+ * and both sender and recipient sync scan exactly up to it. Every absolute window bound must come from this helper:
+ * a site with a wider or narrower bound lets a tx land at an index the syncs never scan, so two stores sharing the
+ * secret could later pick a colliding index.
+ */
+export function unfinalizedTaggingIndexesWindowEnd(finalizedIndex: number | undefined): number {
+  const windowStart = finalizedIndex === undefined ? 0 : finalizedIndex + 1;
+  return windowStart + UNFINALIZED_TAGGING_INDEXES_WINDOW_LEN;
+}
+
 // The number of tags probed per constrained secret in the first round.
 //
 // The probe doubles each round (2, 4, 8, ..., capped at UNFINALIZED_TAGGING_INDEXES_WINDOW_LEN) while every probed

yarn-project/pxe/src/tagging/recipient_sync/sync_tagged_private_logs.test.ts

@@ -132,9 +132,7 @@ describe('syncTaggedPrivateLogs', () => {
     await sync(secrets);
 
     const expectedTags = (
-      await Promise.all(
-        secrets.map(secret => computeSiloedTagRange(secret, UNFINALIZED_TAGGING_INDEXES_WINDOW_LEN + 1)),
-      )
+      await Promise.all(secrets.map(secret => computeSiloedTagRange(secret, UNFINALIZED_TAGGING_INDEXES_WINDOW_LEN)))
     ).flat();
     const asStrings = (tags: SiloedTag[]) => tags.map(t => t.toString()).sort();
 
@@ -191,10 +189,10 @@ describe('syncTaggedPrivateLogs', () => {
   it('updates store correctly when multiple iterations are needed', async () => {
     const secret = await randomAppTaggingSecret(AppTaggingSecretKind.UNCONSTRAINED);
 
-    // A log at the last index of the initial window [0, WINDOW_LEN] moves the finalized index to WINDOW_LEN,
+    // A log at the last index of the initial window [0, WINDOW_LEN) moves the finalized index to WINDOW_LEN - 1,
     // which shifts the next window forward and triggers a second iteration. A second log sits in the advanced
     // window, only reachable in the second iteration.
-    const lastIndexInInitialWindow = UNFINALIZED_TAGGING_INDEXES_WINDOW_LEN;
+    const lastIndexInInitialWindow = UNFINALIZED_TAGGING_INDEXES_WINDOW_LEN - 1;
     const newWindowIndex = lastIndexInInitialWindow + 3;
     mockNodeWithLogs(await computeSiloedTags(secret, [lastIndexInInitialWindow, newWindowIndex]));
 
@@ -514,11 +512,11 @@ describe('syncTaggedPrivateLogs', () => {
       expect(await taggingStore.getHighestFinalizedIndex(secret, JOB_ID)).toBe(totalLogs - 1);
       expect(await taggingStore.getHighestAgedIndex(secret, JOB_ID)).toBe(totalLogs - 1);
 
-      // The first round spans the full cold-start window (WINDOW_LEN + 1). Because every index hit, the next round
-      // re-anchors to another full WINDOW_LEN window ahead of the new finalized index: no small initial probe and no
-      // doubling, in contrast to the constrained scan.
+      // The first round spans the full cold-start window (WINDOW_LEN, the same bound the sender store permits fresh
+      // pending indexes under). Because every index hit, the next round re-anchors to another full WINDOW_LEN window
+      // ahead of the new finalized index: no small initial probe and no doubling, in contrast to the constrained scan.
       expect(callSizes().slice(0, 2)).toEqual([
-        UNFINALIZED_TAGGING_INDEXES_WINDOW_LEN + 1,
+        UNFINALIZED_TAGGING_INDEXES_WINDOW_LEN,
         UNFINALIZED_TAGGING_INDEXES_WINDOW_LEN,
       ]);
     });

yarn-project/pxe/src/tagging/recipient_sync/sync_tagged_private_logs.ts

@@ -7,7 +7,11 @@ import { AppTaggingSecretKind, SiloedTag } from '@aztec/stdlib/logs';
 import type { BlockHeader } from '@aztec/stdlib/tx';
 
 import type { RecipientTaggingStore } from '../../storage/tagging_store/recipient_tagging_store.js';
-import { INITIAL_CONSTRAINED_PROBE_LEN, UNFINALIZED_TAGGING_INDEXES_WINDOW_LEN } from '../constants.js';
+import {
+  INITIAL_CONSTRAINED_PROBE_LEN,
+  UNFINALIZED_TAGGING_INDEXES_WINDOW_LEN,
+  unfinalizedTaggingIndexesWindowEnd,
+} from '../constants.js';
 import { getAllPrivateLogsByTags } from '../get_all_logs_by_tags.js';
 import { findHighestIndexes } from './utils/find_highest_indexes.js';
 
@@ -149,7 +153,7 @@ function getIndexRangesForSecrets(
   return Promise.all(
     secrets.map(async (secret): Promise<PendingSecret> => {
       const currentHighestFinalizedIndex = await taggingStore.getHighestFinalizedIndex(secret, jobId);
-      const boundEnd = (currentHighestFinalizedIndex ?? 0) + UNFINALIZED_TAGGING_INDEXES_WINDOW_LEN + 1;
+      const boundEnd = unfinalizedTaggingIndexesWindowEnd(currentHighestFinalizedIndex);
 
       if (secret.kind === AppTaggingSecretKind.CONSTRAINED) {
         // Constrained streams are gapless and resume at the finalized index, so probe a small initial window and stop
@@ -257,7 +261,7 @@ async function processConstrainedResults(
   const probeFullyConsumed = firstMissingIndex >= pending.end;
   const boundEnd =
     highestFinalizedIndex !== undefined
-      ? Math.max(pending.boundEnd, highestFinalizedIndex + UNFINALIZED_TAGGING_INDEXES_WINDOW_LEN + 1)
+      ? Math.max(pending.boundEnd, unfinalizedTaggingIndexesWindowEnd(highestFinalizedIndex))
       : pending.boundEnd;
 
   // Double the probe each round, capped at the window (see INITIAL_CONSTRAINED_PROBE_LEN in ../constants.ts).
@@ -319,7 +323,7 @@ async function processUnconstrainedResults(
 
   // For the next iteration we want to look only at indexes for which we have not yet fetched logs while
   // ensuring that we do not look further than WINDOW_LEN ahead of the highest finalized index.
-  const end = highestFinalizedIndex + UNFINALIZED_TAGGING_INDEXES_WINDOW_LEN + 1;
+  const end = unfinalizedTaggingIndexesWindowEnd(highestFinalizedIndex);
   return {
     kind: AppTaggingSecretKind.UNCONSTRAINED,
     secret: pending.secret,

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

@@ -3,7 +3,7 @@ import type { AztecNode } from '@aztec/stdlib/interfaces/server';
 import type { AppTaggingSecret } from '@aztec/stdlib/logs';
 
 import type { SenderTaggingStore } from '../../storage/tagging_store/sender_tagging_store.js';
-import { UNFINALIZED_TAGGING_INDEXES_WINDOW_LEN } from '../constants.js';
+import { unfinalizedTaggingIndexesWindowEnd } from '../constants.js';
 import {
   EMPTY_STATUS_CHANGE,
   getStatusChangeOfPending,
@@ -49,7 +49,9 @@ export async function syncSenderTaggingIndexes(
   const finalizedIndex = await taggingStore.getLastFinalizedIndex(secret, jobId);
 
   let start = finalizedIndex === undefined ? 0 : finalizedIndex + 1;
-  let end = start + UNFINALIZED_TAGGING_INDEXES_WINDOW_LEN;
+  // The loop only extends the window when the finalized index moves,
+  // so this first window must cover the entire permitted range on its own.
+  let end = unfinalizedTaggingIndexesWindowEnd(finalizedIndex);
 
   let previousFinalizedIndex = finalizedIndex;
   let newFinalizedIndex = undefined;
@@ -122,8 +124,7 @@ export async function syncSenderTaggingIndexes(
       //    New window:                                             [21, 22, 23]
 
       const previousEnd = end;
-      // Add 1 because `end` is exclusive and the known finalized index is not included in the window.
-      end = newFinalizedIndex! + UNFINALIZED_TAGGING_INDEXES_WINDOW_LEN + 1;
+      end = unfinalizedTaggingIndexesWindowEnd(newFinalizedIndex);
       start = previousEnd;
       previousFinalizedIndex = newFinalizedIndex;
     } else {