Router publisher advances latestNonce before transfers are stored

[chenshj73]

Title: Router publisher advances latestNonce before transfers are stored

Repository: connext/monorepo
Commit reviewed: 7758e62

Summary

getXCalls() advances the persistent latestNonce watermark before the newly discovered transfers are enriched with destination data and written to the pending-transfer cache.

If subgraph.getDestinationXCalls(...) or cache.transfers.storeTransfers(...) throws after the watermark is advanced, the affected origin xcalls have not been added to the pending queue, but the next polling round starts from latestNonce + 1. Those xcalls can then be skipped indefinitely and never published to the sequencer by retryXCalls().

This is a lifecycle bug in the xcall discovery flow: the durable progress marker is committed before the durable work item is committed.

Code

In packages/agents/router/src/tasks/publisher/operations/getXCalls.ts, latestNonce is loaded and used as the lower bound for the subgraph query:

37	      let latestNonce = await cache.transfers.getLatestNonce(domain);
38	      latestNonce = Math.max(latestNonce, config.chains[domain].startNonce ?? 0);
39	      logger.debug("Selected latestNonce", requestContext, methodContext, { domain, latestNonce });
40	
41	      subgraphQueryMetaParams.set(domain, {
42	        maxBlockNumber: latestBlockNumber - safeConfirmations,
43	        latestNonce: latestNonce == 0 ? 0 : latestNonce + 1, // queries at >= latest nonce, so use 1 larger than whats in the cache
44	        destinationDomains,
45	        orderDirection: "asc",
46	      });

After getOriginXCalls(...) returns, the code persists the new nonce watermark immediately:

58	  if ([...subgraphQueryMetaParams.keys()].length > 0) {
59	    const { txIdsByDestinationDomain, allTxById, latestNonces, txByOriginDomain } = await subgraph.getOriginXCalls(
60	      subgraphQueryMetaParams,
61	    );

119	    for (const [domain, nonce] of latestNonces.entries()) {
120	      // set nonce now so we don't requery the same transfers
121	      await cache.transfers.setLatestNonce(domain, nonce ?? 0);
122	      logger.debug("Set latest nonce", requestContext, methodContext, { domain, nonce });
123	    }

Only after that does it fetch destination-side status and write the transfers into cache:

125	    if (txIdsByDestinationDomain.size > 0) {
126	      // filter transfers by unsupported destination domain
127	      for (const destinationDomain of txIdsByDestinationDomain.keys()) {
128	        if (!allowedDomains.includes(destinationDomain)) {
129	          const transferIdsToRemove = txIdsByDestinationDomain.get(destinationDomain);
130	          for (const transferId of transferIdsToRemove ?? []) {
131	            allTxById.delete(transferId);
132	          }
133	
134	          txIdsByDestinationDomain.delete(destinationDomain);
135	        }
136	      }
137	      const transfers = await subgraph.getDestinationXCalls(txIdsByDestinationDomain, allTxById);
138	      if (transfers.length === 0) {
139	        logger.debug("No pending transfers after filtering destination", requestContext, methodContext, {
140	          subgraphQueryMetaParams: [...subgraphQueryMetaParams.entries()],
141	        });
142	      } else {
143	        await cache.transfers.storeTransfers(transfers as XTransfer[], false);
144	        for (const transfer of transfers) {
145	          logger.debug("Added transfer to cache", requestContext, methodContext, { transferId: transfer.transferId });
146	        }
147	      }
148	    } else {

The cache write is what makes a transfer visible to the retry/publish loop. In packages/adapters/cache/src/lib/caches/transfers.ts, storeTransfers() writes the transfer and adds pending origin xcalls to the pending list:

120	  public async storeTransfers(transfers: XTransfer[], cleanup = true): Promise<void> {
121	    const { sanitizeNull } = getHelpers();
122	    const nonceDidIncreaseForDomain: { [domain: string]: boolean } = {};
123	    const highestNonceByDomain: { [domain: string]: number } = {};
124	    for (let transfer of transfers) {
125	      const existing = await this.getTransfer(transfer.transferId);
126	      // Sanity check: no update needed if this transfer is same as the one already stored.
127	      if (JSON.stringify(transfer) === JSON.stringify(existing)) {
128	        continue;
129	      }

153	      const { transferId, xparams, origin, destination } = transfer;
154	      const { originDomain } = xparams;
155	      const nonce = Number(xparams.nonce);
156	      const stringified = JSON.stringify(transfer);
157	
158	      // set transaction data at domain field in hash, hset returns the number of field that were added
159	      // gte(1) => added, 0 => updated,
160	      // reference: https://redis.io/commands/hset
161	      const added = (await this.data.hset(`${this.prefix}:transfers`, transferId, stringified)) >= 1;
162	      if (added && origin?.xcall.transactionHash && !destination) {
163	        // XCall defined but Execute and Reconcile are not defined => pending transfer.
164	        // If the transfer was added (previously not recorded) and it's a pending transfer, add it to the
165	        // pending transfers list.
166	        await this.addPending(originDomain, transferId);
167	      } else if (destination?.execute?.transactionHash || destination?.reconcile?.transactionHash) {

It also contains its own nonce advancement after successful storage:

173	      // Retrieve latest nonce for this domain.
174	      let currentNonce = highestNonceByDomain[originDomain];
175	      if (!currentNonce) {
176	        // If we don't have a nonce recorded yet for this domain, we need to retrieve it from the cache.
177	        currentNonce = (await this.getLatestNonce(originDomain)) ?? 0;
178	        highestNonceByDomain[originDomain] = currentNonce;
179	        nonceDidIncreaseForDomain[originDomain] = true;
180	      }
181	      if (nonce > currentNonce) {
182	        // If the new nonce is higher than the current one, we'll record it to later update the cache.
183	        highestNonceByDomain[originDomain] = nonce;
184	        nonceDidIncreaseForDomain[originDomain] = true;
185	      }
186	    }
187	    // Set the new highest nonce, and publish NewHighestNonce events for any new highest nonces we found.
188	    for (const [domain, nonce] of Object.entries(highestNonceByDomain)) {
189	      if (nonceDidIncreaseForDomain[domain]) {
190	        await this.data.hset(`${this.prefix}:nonce`, domain, nonce);
191	        await this.data.publish(StoreChannel.NewHighestNonce, JSON.stringify({ domain, nonce }));
192	      }

So the publisher path currently has an extra earlier watermark commit that is not coupled to the actual transfer-cache write.

Failure scenario

1. Cache has latestNonce = 100 for origin domain A.
2. getOriginXCalls(...) returns origin xcalls with nonces 101..110 and latestNonces[A] = 110.
3. getXCalls() executes setLatestNonce(A, 110).
4. subgraph.getDestinationXCalls(...) times out, throws, or storeTransfers(...) fails before the transfers are written and before addPending(...) adds them to the pending list.
5. Next polling round calls getLatestNonce(A) and queries with latestNonce: 111.
6. Nonces 101..110 are neither in the pending queue nor in the normal origin query window anymore.

The missingNonces logic does not cover this case because those nonces were present in the successful origin response. They are not gaps between returned nonces.

Impact

Pending source-chain xcalls can be dropped from the router publisher pipeline after a transient destination-subgraph/cache error. The transfer exists on the origin chain, but the off-chain router lifecycle may fail to publish it to the sequencer because the durable scan cursor has moved past it before the durable pending item exists.

This is especially risky because the affected operation spans multiple systems: origin subgraph read, destination status lookup, Redis cache write, pending queue, and later MQ publication.

Suggested fix

Advance latestNonce only after the transfers for that domain have been successfully stored, or rely on storeTransfers(...) to advance the nonce after the cache write. If a separate explicit watermark update is still needed, it should happen after getDestinationXCalls(...) and storeTransfers(...) succeed, ideally per domain and only for transfers that were either stored as pending or proven already completed on the destination.

It would also be useful to add a regression test where getOriginXCalls(...) returns transfers and latestNonces, but getDestinationXCalls(...) or storeTransfers(...) rejects. The expected behavior should be that setLatestNonce(...) is not called before the transfer batch is durably stored.