feat: Add order reevaluation logic

config.example.yaml

@@ -19,6 +19,8 @@ global:
     maxPendingTransactions: 50          # Maximum number of transactions within the 'submit' pipeline.
 
     # Evaluation properties
+    evaluationRetryInterval: 3600000    # Interval at which to reevaluate whether to relay a message.
+    maxEvaluationDuration: 86400000     # Time after which to drop an undelivered message.
     unrewardedDeliveryGas: '100000'     # Gas amount that will be unrewarded on delivery submission.
     minDeliveryReward: 0.001            # In the 'pricingDenomination' specified below
     relativeMinDeliveryReward: 0.001

src/config/config.schema.ts

@@ -147,6 +147,8 @@ const SUBMITTER_SCHEMA = {
         maxPendingTransactions: { $ref: "positive-number-schema" },
 
         //TODO define 'evaluation' configuration somewhere else?
+        evaluationRetryInterval: { $ref: "positive-number-schema" },
+        maxEvaluationDuration: { $ref: "positive-number-schema" },
         unrewardedDeliveryGas: { $ref: "gas-field-schema" },
         minDeliveryReward: { $ref: "positive-number-schema" },
         relativeMinDeliveryReward: { $ref: "positive-number-schema" },

src/config/config.types.ts

@@ -45,6 +45,8 @@ export interface SubmitterGlobalConfig {
   maxTries?: number;
   maxPendingTransactions?: number;
 
+  evaluationRetryInterval?: number;
+  maxEvaluationDuration?: number;
   unrewardedDeliveryGas?: bigint;
   minDeliveryReward?: number;
   relativeMinDeliveryReward?: number;

src/submitter/queues/eval-queue.ts

@@ -80,6 +80,9 @@ export class EvalQueue extends ProcessingQueue<EvalOrder, SubmitOrder> {
             // Move the order to the submit queue
             return { result: { ...order, gasLimit: gasEstimation, isDelivery } };
         } else {
+            // Request the order to be retried in the future.
+            order.retryEvaluation = true;
+
             return null;
         }
     }

src/submitter/submitter.service.ts

@@ -15,6 +15,8 @@ const PROCESSING_INTERVAL_DEFAULT = 100;
 const MAX_TRIES_DEFAULT = 3;
 const MAX_PENDING_TRANSACTIONS = 50;
 const NEW_ORDERS_DELAY_DEFAULT = 0;
+const EVALUATION_RETRY_INTERVAL_DEFAULT = 60 * 60 * 1000;
+const MAX_EVALUATION_DURATION_DEFAULT = 24 * 60 * 60 * 1000;
 const UNREWARDED_DELIVERY_GAS_DEFAULT = 0n;
 const MIN_DELIVERY_REWARD_DEFAULT = 0;
 const RELATIVE_MIN_DELIVERY_REWARD_DEFAULT = 0;
@@ -29,6 +31,8 @@ interface GlobalSubmitterConfig {
     processingInterval: number;
     maxTries: number;
     maxPendingTransactions: number;
+    evaluationRetryInterval: number;
+    maxEvaluationDuration: number;
     unrewardedDeliveryGas: bigint;
     minDeliveryReward: number;
     relativeMinDeliveryReward: number;
@@ -48,6 +52,8 @@ export interface SubmitterWorkerData {
     processingInterval: number;
     maxTries: number;
     maxPendingTransactions: number;
+    evaluationRetryInterval: number;
+    maxEvaluationDuration: number;
     unrewardedDeliveryGas: bigint;
     minDeliveryReward: number;
     relativeMinDeliveryReward: number;
@@ -132,6 +138,10 @@ export class SubmitterService {
         const maxPendingTransactions =
             submitterConfig.maxPendingTransactions ?? MAX_PENDING_TRANSACTIONS;
 
+        const evaluationRetryInterval =
+            submitterConfig.evaluationRetryInterval ?? EVALUATION_RETRY_INTERVAL_DEFAULT;
+        const maxEvaluationDuration =
+            submitterConfig.maxEvaluationDuration ?? MAX_EVALUATION_DURATION_DEFAULT;
         const unrewardedDeliveryGas = 
             submitterConfig.unrewardedDeliveryGas ?? UNREWARDED_DELIVERY_GAS_DEFAULT;
         const minDeliveryReward =
@@ -155,6 +165,8 @@ export class SubmitterService {
             maxTries,
             maxPendingTransactions,
             walletPublicKey,
+            evaluationRetryInterval,
+            maxEvaluationDuration,
             unrewardedDeliveryGas,
             minDeliveryReward,
             relativeMinDeliveryReward,
@@ -205,9 +217,17 @@ export class SubmitterService {
                 chainConfig.submitter.maxPendingTransactions ??
                 globalConfig.maxPendingTransactions,
 
+            evaluationRetryInterval:
+                chainConfig.submitter.evaluationRetryInterval ??
+                globalConfig.evaluationRetryInterval,
+
             unrewardedDeliveryGas:
                 chainConfig.submitter.unrewardedDeliveryGas ??
                 globalConfig.unrewardedDeliveryGas,
+
+            maxEvaluationDuration:
+                chainConfig.submitter.maxEvaluationDuration ??
+                globalConfig.maxEvaluationDuration,
 
             minDeliveryReward:
                 chainConfig.submitter.minDeliveryReward ??

src/submitter/submitter.types.ts

@@ -10,6 +10,8 @@ export interface Order {
 
 export interface EvalOrder extends Order {
     priority: boolean;
+    evaluationDeadline: number;
+    retryEvaluation?: boolean;
 }
 
 export interface SubmitOrder extends Order {
@@ -24,13 +26,15 @@ export interface SubmitOrderResult extends SubmitOrder {
     txReceipt: TransactionReceipt;
 }
 
-export interface NewOrder<OrderType> {
+export interface PendingOrder<OrderType> {
     order: OrderType;
     processAt: number;
 }
 
 
 export interface BountyEvaluationConfig {
+    evaluationRetryInterval: number,
+    maxEvaluationDuration: number,
     unrewardedDeliveryGas: bigint;
     minDeliveryReward: number;
     relativeMinDeliveryReward: number,

src/submitter/submitter.worker.ts

@@ -10,7 +10,7 @@ import { IncentivizedMessageEscrow__factory } from 'src/contracts/factories/Ince
 import { workerData } from 'worker_threads';
 import { AmbPayload } from 'src/store/types/store.types';
 import { STATUS_LOG_INTERVAL } from 'src/logger/logger.service';
-import { BountyEvaluationConfig, EvalOrder, NewOrder } from './submitter.types';
+import { BountyEvaluationConfig, EvalOrder, PendingOrder } from './submitter.types';
 import { EvalQueue } from './queues/eval-queue';
 import { SubmitQueue } from './queues/submit-queue';
 import { wait } from 'src/common/utils';
@@ -32,7 +32,7 @@ class SubmitterWorker {
     private readonly pricing: PricingInterface;
     private readonly wallet: WalletInterface;
 
-    private readonly newOrdersQueue: NewOrder<EvalOrder>[] = [];
+    private readonly pendingQueue: PendingOrder<EvalOrder>[] = [];
     private readonly evalQueue: EvalQueue;
     private readonly submitQueue: SubmitQueue;
 
@@ -65,6 +65,8 @@ class SubmitterWorker {
                 this.loadIncentivesContracts(this.config.incentivesAddresses),
                 this.config.chainId,
                 {
+                    evaluationRetryInterval: this.config.evaluationRetryInterval,
+                    maxEvaluationDuration: this.config.maxEvaluationDuration,
                     unrewardedDeliveryGas: this.config.unrewardedDeliveryGas,
                     minDeliveryReward: this.config.minDeliveryReward,
                     relativeMinDeliveryReward: this.config.relativeMinDeliveryReward,
@@ -140,7 +142,7 @@ class SubmitterWorker {
         const logStatus = () => {
             const status = {
                 capacity: this.getSubmitterCapacity(),
-                newOrdersQueue: this.newOrdersQueue.length,
+                pendingQueue: this.pendingQueue.length,
                 evalQueue: this.evalQueue.size,
                 evalRetryQueue: this.evalQueue.retryQueue.length,
                 submitQueue: this.submitQueue.size,
@@ -181,21 +183,57 @@ class SubmitterWorker {
         await this.listenForOrders();
 
         while (true) {
-            const evalOrders = await this.processNewOrdersQueue();
+            const evalOrders = await this.processPendingQueue();
 
             await this.evalQueue.addOrders(...evalOrders);
             await this.evalQueue.processOrders();
 
-            const [newSubmitOrders, ,] = this.evalQueue.getFinishedOrders();
+            const [newSubmitOrders, noSubmitEvalOrders,] = this.evalQueue.getFinishedOrders();
+            this.processNoSubmitEvalOrders(noSubmitEvalOrders);
             await this.submitQueue.addOrders(...newSubmitOrders);
             await this.submitQueue.processOrders();
 
+            //TODO process failed submissions in any way?
+
             this.submitQueue.getFinishedOrders(); // Flush the internal queues
 
             await wait(this.config.processingInterval);
         }
     }
 
+    private processNoSubmitEvalOrders(evalOrders: EvalOrder[]): void {
+        for (const order of evalOrders) {
+            if (order.retryEvaluation) {
+
+                // Clear the 'retryEvaluation' flag to avoid unexpected retries.
+                order.retryEvaluation = undefined;
+
+                const nextEvaluationTime = Date.now() + this.config.evaluationRetryInterval;
+                if (nextEvaluationTime > order.evaluationDeadline) {
+                    this.logger.info(
+                        {
+                            messageIdentifier: order.messageIdentifier,
+                            nextEvaluationTime,
+                            evaluationDeadline: order.evaluationDeadline,
+                        },
+                        'Dropping evaluation order.'
+                    );
+                }
+                else {
+                    this.logger.info(
+                        {
+                            messageIdentifier: order.messageIdentifier,
+                            nextEvaluationTime,
+                            evaluationDeadline: order.evaluationDeadline,
+                        },
+                        'Queueing order for reevaluation.'
+                    );
+                    this.addPendingOrder(nextEvaluationTime, order);
+                }
+            }
+        }
+    }
+
     /**
      * Subscribe to the Store to listen for relevant payloads to submit.
      */
@@ -247,48 +285,69 @@ class SubmitterWorker {
             `Submit order received.`,
         );
         if (priority) {
-            // Push directly into the submit queue
+            // Push directly into the eval queue
             await this.evalQueue.addOrders({
                 amb,
                 messageIdentifier,
                 message,
                 messageCtx,
                 priority: true,
+                evaluationDeadline: Date.now() + this.config.maxEvaluationDuration,
                 incentivesPayload,
             });
         } else {
-            // Push into the evaluation queue
-            this.newOrdersQueue.push({
-                processAt: Date.now() + this.config.newOrdersDelay,
-                order: {
+            // Push into the pending queue
+            this.addPendingOrder(
+                Date.now() + this.config.newOrdersDelay,
+                {
                     amb,
                     messageIdentifier,
                     message,
                     messageCtx,
                     priority: false,
+                    evaluationDeadline: Date.now() + this.config.maxEvaluationDuration,
                     incentivesPayload,
                 },
-            });
+            );
         }
     }
 
-    /***************  New Order Queue  ***************/
+    /***************  Pending Orders Queue  ***************/
+
+    private addPendingOrder(processAt: number, order: EvalOrder): void {
+
+        // Insert the new order into the 'pendingQueue' keeping the queue order.
+        const insertIndex = this.pendingQueue.findIndex(order => {
+            return order.processAt > processAt;
+        });
+
+        const pendingOrder: PendingOrder<EvalOrder> = {
+            order,
+            processAt
+        };
+
+        if (insertIndex == -1) {
+            this.pendingQueue.push(pendingOrder);
+        } else {
+            this.pendingQueue.splice(insertIndex, 0, pendingOrder);
+        }
+    }
 
-    private async processNewOrdersQueue(): Promise<EvalOrder[]> {
+    private async processPendingQueue(): Promise<EvalOrder[]> {
         const currentTimestamp = Date.now();
         const capacity = this.getSubmitterCapacity();
 
         let i;
-        for (i = 0; i < this.newOrdersQueue.length; i++) {
-            const nextNewOrder = this.newOrdersQueue[i]!;
+        for (i = 0; i < this.pendingQueue.length; i++) {
+            const nextPendingOrder = this.pendingQueue[i]!;
 
-            if (nextNewOrder.processAt > currentTimestamp || i + 1 > capacity) {
+            if (nextPendingOrder.processAt > currentTimestamp || i + 1 > capacity) {
                 break;
             }
         }
 
-        const ordersToEval = this.newOrdersQueue.splice(0, i).map((newOrder) => {
-            return newOrder.order;
+        const ordersToEval = this.pendingQueue.splice(0, i).map((pendingOrder) => {
+            return pendingOrder.order;
         });
 
         return ordersToEval;