vacp2p/linea-monorepo
1
0
Fork 0
mirror of https://github.com/vacp2p/linea-monorepo.git synced 2026-01-09 04:08:01 -05:00
Code Issues Packages Projects Releases Wiki Activity

[Feat] Postman Sponsorship Metrics (#902)

Browse Source 
* added claimTxGasUsed attribute

* add test code for message mapper

* add two more table entries

* added isForSponsorship db entity change

* remove isForSponsorship

* working postman tests

* working tests for add isForSponsorship attribute

* fix positioning

* refactor MessageClaimingPersister

* remove bad comment

* added migration:create command

* generate migration docs

* added new migration for sponsorship

* typo

* passing tests for MessageMetricsService & MetricsService refactor

* remove console logs

* added subscriber code for isForSponsorship

* working e2e and local test for sponsorship messages metrics

* add comment

* added SponsorshipFeesWei and SponsorshipFeesGwei metrics

* did sponsorshipfeesubscriber

* register sponsorshipfeessubscriber

* string fixes

* remove claimtxgasused and claimtxgasprice db columns, refactor metricsservice design pattern

* SponsorshipMetricsUpdater.ts

* change isForSponsorship semantics

* remove isForSponsorship from DB

* Revert "remove isForSponsorship from DB"

This reverts commit 79f4887e3492a43443bb5dba78f66f78606241a5.

* remove isForSponsorship public function

* remove isForSponsorship label from message metric

* add getSponsoredMessagesTotal method

* added sponsorshipmetricsupdater to messageclaimingpersister

* amend base postman image for e2e test

* add promql queries in comments

* empty

* remove === true
This commit is contained in:
kyzooghost
2025-05-05 20:26:37 +10:00
committed by GitHub
parent 095dee1f62
commit 19880cd8a9
32 changed files with 755 additions and 380 deletions
Download Patch File Download Diff File Expand all files Collapse all files

15
postman/README.md
View File

@@ -110,6 +110,11 @@ Stop the postman docker container manually.
#### Run the postman locally:
Before the postman can be run and tested locally, we must build the monorepo projects linea-sdk and linea-native-libs
```bash
NATIVE_LIBS_RELEASE_TAG=blob-libs-v1.0.1 pnpm run -F linea-native-libs build && pnpm run -F linea-sdk build
```
From the postman folder run the following commands:
```bash
@@ -133,6 +138,16 @@ pnpm run build
pnpm run test
```
### Database migrations
Create an empty DB migration file with <NAME>
```bash
MIGRATION_NAME=<NAME> pnpm run migration:create
```
We will then implement the migration code manually. We omit scripts for TypeORM migration generation because the CLI tool is unable to generate correct migration code in our case.
## License
This package is licensed under the Apache License, Version 2.0. See [LICENSE](LICENSE) for more information.

3
postman/package.json
View File

@@ -15,7 +15,8 @@
"build": "tsc -p tsconfig.build.json",
"build:runSdk": "tsc ./scripts/runSdk.ts",
"test": "npx jest --bail --detectOpenHandles --forceExit",
"lint:fix": "pnpm run lint:ts:fix && pnpm run prettier:fix"
"lint:fix": "pnpm run lint:ts:fix && pnpm run prettier:fix",
"migration:create": "npx ts-node -P ./tsconfig.json -r tsconfig-paths/register ./node_modules/typeorm/cli.js migration:create ./src/application/postman/persistence/migrations/${MIGRATION_NAME}"
},
"dependencies": {
"@consensys/linea-native-libs": "workspace:*",

71
postman/src/application/postman/api/metrics/MessageMetricsService.ts
View File

@@ -1,71 +0,0 @@
import { EntityManager } from "typeorm";
import { Direction } from "@consensys/linea-sdk";
import { MetricsService } from "./MetricsService";
import { MessageEntity } from "../../persistence/entities/Message.entity";
import { MessageStatus } from "../../../../core/enums";
import { LineaPostmanMetrics } from "../../../../core/metrics/IMetricsService";
export class MessageMetricsService extends MetricsService {
constructor(private readonly entityManager: EntityManager) {
super();
this.createGauge(LineaPostmanMetrics.Messages, "Current number of messages by status and direction", [
"status",
"direction",
]);
}
public async initialize(): Promise<void> {
const fullResult = await this.getMessagesCountFromDatabase();
this.initializeGaugeValues(fullResult);
}
private async getMessagesCountFromDatabase(): Promise<
{ status: MessageStatus; direction: Direction; count: number }[]
> {
const totalNumberOfMessagesByStatusAndDirection = await this.entityManager
.createQueryBuilder(MessageEntity, "message")
.select("message.status", "status")
.addSelect("message.direction", "direction")
.addSelect("COUNT(message.id)", "count")
.groupBy("message.status")
.addGroupBy("message.direction")
.getRawMany();
// MessageStatus => MessageDirection => Count
const resultMap = new Map<string, Map<string, number>>();
totalNumberOfMessagesByStatusAndDirection.forEach((r) => {
if (!resultMap.has(r.status)) {
resultMap.set(r.status, new Map());
}
resultMap.get(r.status)!.set(r.direction, Number(r.count));
});
const results: { status: MessageStatus; direction: Direction; count: number }[] = [];
for (const status of Object.values(MessageStatus)) {
for (const direction of Object.values(Direction)) {
results.push({
status,
direction,
count: resultMap.get(status)?.get(direction) || 0,
});
}
}
return results;
}
private initializeGaugeValues(fullResult: { status: MessageStatus; direction: Direction; count: number }[]): void {
for (const { status, count, direction } of fullResult) {
this.incrementGauge(
LineaPostmanMetrics.Messages,
{
status,
direction,
},
count,
);
}
}
}

101
postman/src/application/postman/api/metrics/MessageMetricsUpdater.ts Normal file
View File

@@ -0,0 +1,101 @@
import { EntityManager } from "typeorm";
import { MessageEntity } from "../../persistence/entities/Message.entity";
import {
IMetricsService,
IMessageMetricsUpdater,
LineaPostmanMetrics,
MessagesMetricsAttributes,
} from "../../../../core/metrics";
import { Direction } from "@consensys/linea-sdk";
import { MessageStatus } from "../../../../core/enums";
export class MessageMetricsUpdater implements IMessageMetricsUpdater {
constructor(
private readonly entityManager: EntityManager,
private readonly metricsService: IMetricsService,
) {
this.metricsService.createGauge(
LineaPostmanMetrics.Messages,
"Current number of messages by status and direction",
["status", "direction"],
);
}
public async initialize(): Promise<void> {
this.initializeMessagesGauges();
}
private async initializeMessagesGauges(): Promise<void> {
const totalNumberOfMessagesByAttributeGroups = await this.entityManager
.createQueryBuilder(MessageEntity, "message")
.select("message.status", "status")
.addSelect("message.direction", "direction")
.addSelect("COUNT(message.id)", "count") // Actually a string type
.groupBy("message.status")
.addGroupBy("message.direction")
.getRawMany();
// JSON.stringify(MessagesMetricsAttributes) => Count
const resultMap = new Map<string, number>();
totalNumberOfMessagesByAttributeGroups.forEach((r) => {
const messageMetricAttributes: MessagesMetricsAttributes = {
status: r.status,
direction: r.direction,
};
const resultMapKey = JSON.stringify(messageMetricAttributes);
resultMap.set(resultMapKey, parseInt(r.count));
});
// Note that we must initialize every attribute combination, or 'incrementGauge' and 'decrementGauge' will not work later on.
for (const status of Object.values(MessageStatus)) {
for (const direction of Object.values(Direction)) {
const attributes: MessagesMetricsAttributes = {
status,
direction,
};
const attributesKey = JSON.stringify(attributes);
this.metricsService.incrementGauge(
LineaPostmanMetrics.Messages,
{
status: attributes.status,
direction: attributes.direction,
},
resultMap.get(attributesKey) ?? 0,
);
}
}
}
public async getMessageCount(messageAttributes: MessagesMetricsAttributes): Promise<number | undefined> {
const { status, direction } = messageAttributes;
return await this.metricsService.getGaugeValue(LineaPostmanMetrics.Messages, {
status,
direction,
});
}
public async incrementMessageCount(messageAttributes: MessagesMetricsAttributes, value: number = 1) {
const { status, direction } = messageAttributes;
await this.metricsService.incrementGauge(
LineaPostmanMetrics.Messages,
{
status,
direction,
},
value,
);
}
public async decrementMessageCount(messageAttributes: MessagesMetricsAttributes, value: number = 1) {
const { status, direction } = messageAttributes;
await this.metricsService.decrementGauge(
LineaPostmanMetrics.Messages,
{
status,
direction,
},
value,
);
}
}

34
postman/src/application/postman/api/metrics/MetricsService.ts → postman/src/application/postman/api/metrics/SingletonMetricsService.ts
View File

@@ -2,10 +2,13 @@ import { Counter, Gauge, MetricObjectWithValues, MetricValue, Registry } from "p
import { IMetricsService, LineaPostmanMetrics } from "../../../../core/metrics/IMetricsService";
/**
* Take care to instantiate as a singleton because there should be only be one instance of prom-client Registry
* TODO - Implement Singleton pattern for this class
*
* MetricsService class that implements the IMetricsService interface.
* This class provides methods to create and manage Prometheus metrics.
*/
export abstract class MetricsService implements IMetricsService {
export class SingletonMetricsService implements IMetricsService {
private readonly registry: Registry;
private readonly counters: Map<LineaPostmanMetrics, Counter<string>>;
private readonly gauges: Map<LineaPostmanMetrics, Gauge<string>>;
@@ -57,8 +60,11 @@ export abstract class MetricsService implements IMetricsService {
}
const metricData = await counter.get();
const metricValueWithMatchingLabels = this.findMetricValueWithExactMatchingLabels(metricData, labels);
return metricValueWithMatchingLabels?.value;
const aggregatedMetricValueWithMatchingLabels = this.aggregateMetricValuesWithExactMatchingLabels(
metricData,
labels,
);
return aggregatedMetricValueWithMatchingLabels?.value;
}
/**
@@ -97,8 +103,11 @@ export abstract class MetricsService implements IMetricsService {
}
const metricData = await gauge.get();
const metricValueWithMatchingLabels = this.findMetricValueWithExactMatchingLabels(metricData, labels);
return metricValueWithMatchingLabels?.value;
const aggregatedMetricValueWithMatchingLabels = this.aggregateMetricValuesWithExactMatchingLabels(
metricData,
labels,
);
return aggregatedMetricValueWithMatchingLabels?.value;
}
/**
@@ -143,10 +152,21 @@ export abstract class MetricsService implements IMetricsService {
}
}
private findMetricValueWithExactMatchingLabels(
private aggregateMetricValuesWithExactMatchingLabels(
metricData: MetricObjectWithValues<MetricValue<string>>,
labels: Record<string, string>,
): MetricValue<string> | undefined {
return metricData.values.find((value) => Object.entries(labels).every(([key, val]) => value.labels[key] === val));
// It is possible to have multiple metric objects with exact matching labels, e.g. if we query for 2 out of the 3 labels being used.
// Hence we should merge all metric objects, and remove labels that were not queried from the merged metric object.
const matchingMetricObjects = metricData.values.filter((value) =>
Object.entries(labels).every(([key, val]) => value.labels[key] === val),
);
if (matchingMetricObjects.length === 0) return undefined;
const mergedMetricObject: MetricValue<string> = {
value: 0,
labels,
};
matchingMetricObjects.forEach((m) => (mergedMetricObject.value += m.value));
return mergedMetricObject;
}
}

50
postman/src/application/postman/api/metrics/SponsorshipMetricsUpdater.ts Normal file
View File

@@ -0,0 +1,50 @@
import { IMetricsService, ISponsorshipMetricsUpdater, LineaPostmanMetrics } from "../../../../core/metrics";
import { Direction } from "@consensys/linea-sdk";
export class SponsorshipMetricsUpdater implements ISponsorshipMetricsUpdater {
constructor(private readonly metricsService: IMetricsService) {
this.metricsService.createCounter(
LineaPostmanMetrics.SponsorshipFeesGwei,
"Gwei component of tx fees paid for sponsored messages by direction",
["direction"],
);
this.metricsService.createCounter(
LineaPostmanMetrics.SponsorshipFeesWei,
"Wei component of tx fees paid for sponsored messages by direction",
["direction"],
);
this.metricsService.createCounter(
LineaPostmanMetrics.SponsoredMessagesTotal,
"Count of sponsored messages by direction",
["direction"],
);
}
public async getSponsoredMessagesTotal(direction: Direction): Promise<number> {
const total = await this.metricsService.getCounterValue(LineaPostmanMetrics.SponsoredMessagesTotal, { direction });
if (total === undefined) return 0;
return total;
}
public async getSponsorshipFeePaid(direction: Direction): Promise<bigint> {
const wei = await this.metricsService.getCounterValue(LineaPostmanMetrics.SponsorshipFeesWei, { direction });
const gwei = await this.metricsService.getCounterValue(LineaPostmanMetrics.SponsorshipFeesGwei, { direction });
if (wei === undefined || gwei === undefined) return 0n;
return BigInt(wei) + BigInt(gwei) * 1_000_000_000n;
}
public async incrementSponsorshipFeePaid(txFee: bigint, direction: Direction) {
const { wei, gwei } = this.convertTxFeesToWeiAndGwei(txFee);
await this.metricsService.incrementCounter(LineaPostmanMetrics.SponsoredMessagesTotal, { direction }, 1);
await this.metricsService.incrementCounter(LineaPostmanMetrics.SponsorshipFeesWei, { direction }, wei);
await this.metricsService.incrementCounter(LineaPostmanMetrics.SponsorshipFeesGwei, { direction }, gwei);
}
private convertTxFeesToWeiAndGwei(txFee: bigint): { gwei: number; wei: number } {
// Last 9 digits
const wei = Number(txFee % BigInt(1_000_000_000));
const gwei = Number(txFee / BigInt(1_000_000_000));
return { wei, gwei };
}
}

64
postman/src/application/postman/api/metrics/__tests__/MessageMetricsService.test.ts
View File

@@ -1,64 +0,0 @@
import { EntityManager, SelectQueryBuilder } from "typeorm";
import { Direction } from "@consensys/linea-sdk";
import { MessageMetricsService } from "../MessageMetricsService";
import { mock, MockProxy } from "jest-mock-extended";
import { MessageStatus } from "../../../../../core/enums";
import { LineaPostmanMetrics } from "../../../../../core/metrics/IMetricsService";
describe("MessageMetricsService", () => {
let messageMetricsService: MessageMetricsService;
let mockEntityManager: MockProxy<EntityManager>;
beforeEach(() => {
mockEntityManager = mock<EntityManager>();
messageMetricsService = new MessageMetricsService(mockEntityManager);
});
it("should update gauges based on message status", async () => {
jest.spyOn(mockEntityManager, "maximum").mockResolvedValue(10);
jest.spyOn(mockEntityManager, "createQueryBuilder").mockReturnValue({
select: jest.fn().mockReturnThis(),
addSelect: jest.fn().mockReturnThis(),
groupBy: jest.fn().mockReturnThis(),
addGroupBy: jest.fn().mockReturnThis(),
getRawMany: jest.fn().mockResolvedValue([
{ status: MessageStatus.SENT, direction: Direction.L1_TO_L2, count: 5 },
{ status: MessageStatus.CLAIMED_SUCCESS, direction: Direction.L1_TO_L2, count: 10 },
]),
// eslint-disable-next-line @typescript-eslint/no-explicit-any
} as unknown as SelectQueryBuilder<any>);
await messageMetricsService.initialize();
// Check if the gauge was updated
expect(
await messageMetricsService.getGaugeValue(LineaPostmanMetrics.Messages, {
status: MessageStatus.SENT,
direction: Direction.L1_TO_L2,
}),
).toBe(5);
expect(
await messageMetricsService.getGaugeValue(LineaPostmanMetrics.Messages, {
status: MessageStatus.CLAIMED_SUCCESS,
direction: Direction.L1_TO_L2,
}),
).toBe(10);
});
it("should return the correct gauge value", async () => {
messageMetricsService.incrementGauge(
LineaPostmanMetrics.Messages,
{
status: "processed",
direction: Direction.L1_TO_L2,
},
10,
);
const gaugeValue = await messageMetricsService.getGaugeValue(LineaPostmanMetrics.Messages, {
status: "processed",
direction: Direction.L1_TO_L2,
});
expect(gaugeValue).toBe(10);
});
});

95
postman/src/application/postman/api/metrics/__tests__/MessageMetricsUpdater.test.ts Normal file
View File

@@ -0,0 +1,95 @@
import { EntityManager, SelectQueryBuilder } from "typeorm";
import { Direction } from "@consensys/linea-sdk";
import { MessageMetricsUpdater } from "../MessageMetricsUpdater";
import { mock, MockProxy } from "jest-mock-extended";
import { MessageStatus } from "../../../../../core/enums";
import { SingletonMetricsService } from "../SingletonMetricsService";
import { IMessageMetricsUpdater } from "../../../../../core/metrics";
describe("MessageMetricsUpdater", () => {
let messageMetricsUpdater: IMessageMetricsUpdater;
let mockEntityManager: MockProxy<EntityManager>;
beforeEach(() => {
mockEntityManager = mock<EntityManager>();
const metricService = new SingletonMetricsService();
messageMetricsUpdater = new MessageMetricsUpdater(mockEntityManager, metricService);
});
const getMessagesCountQueryResp = [
{ status: MessageStatus.SENT, direction: Direction.L1_TO_L2, count: "5" },
{
status: MessageStatus.CLAIMED_SUCCESS,
direction: Direction.L1_TO_L2,
count: "10",
},
];
it("should get correct gauge values after initialization", async () => {
jest.spyOn(mockEntityManager, "maximum").mockResolvedValue(10);
jest.spyOn(mockEntityManager, "createQueryBuilder").mockReturnValue({
select: jest.fn().mockReturnThis(),
addSelect: jest.fn().mockReturnThis(),
groupBy: jest.fn().mockReturnThis(),
addGroupBy: jest.fn().mockReturnThis(),
where: jest.fn().mockReturnThis(),
andWhere: jest.fn().mockReturnThis(),
getRawMany: jest.fn().mockResolvedValue(getMessagesCountQueryResp),
// eslint-disable-next-line @typescript-eslint/no-explicit-any
} as unknown as SelectQueryBuilder<any>);
await messageMetricsUpdater.initialize();
// Check if the gauge was updated
expect(
await messageMetricsUpdater.getMessageCount({
status: MessageStatus.SENT,
direction: Direction.L1_TO_L2,
}),
).toBe(5);
expect(
await messageMetricsUpdater.getMessageCount({
status: MessageStatus.CLAIMED_SUCCESS,
direction: Direction.L1_TO_L2,
}),
).toBe(10);
});
it("should get correct values after incrementMessageCount", async () => {
messageMetricsUpdater.incrementMessageCount(
{
status: MessageStatus.PENDING,
direction: Direction.L1_TO_L2,
},
10,
);
const gaugeValue = await messageMetricsUpdater.getMessageCount({
status: MessageStatus.PENDING,
direction: Direction.L1_TO_L2,
});
expect(gaugeValue).toBe(10);
});
it("should get correct values after decrementMessageCount", async () => {
messageMetricsUpdater.incrementMessageCount(
{
status: MessageStatus.PENDING,
direction: Direction.L1_TO_L2,
},
10,
);
messageMetricsUpdater.decrementMessageCount(
{
status: MessageStatus.PENDING,
direction: Direction.L1_TO_L2,
},
5,
);
const gaugeValue = await messageMetricsUpdater.getMessageCount({
status: MessageStatus.PENDING,
direction: Direction.L1_TO_L2,
});
expect(gaugeValue).toBe(5);
});
});

16
postman/src/application/postman/api/metrics/__tests__/MetricsService.test.ts → postman/src/application/postman/api/metrics/__tests__/SingletonMetricsService.test.ts
View File

@@ -1,18 +1,12 @@
import { Counter, Gauge } from "prom-client";
import { LineaPostmanMetrics } from "../../../../../core/metrics/IMetricsService";
import { MetricsService } from "../MetricsService";
import { IMetricsService, LineaPostmanMetrics } from "../../../../../core/metrics/IMetricsService";
import { SingletonMetricsService } from "../SingletonMetricsService";
class TestMetricService extends MetricsService {
constructor() {
super();
}
}
describe("MetricsService", () => {
let metricService: TestMetricService;
describe("SingletonMetricsService", () => {
let metricService: IMetricsService;
beforeEach(() => {
metricService = new TestMetricService();
metricService = new SingletonMetricsService();
});
it("should create a counter", () => {

23
postman/src/application/postman/api/metrics/__tests__/SponsorshipMetricsUpdater.test.ts Normal file
View File

@@ -0,0 +1,23 @@
import { Direction } from "@consensys/linea-sdk";
import { SponsorshipMetricsUpdater } from "../SponsorshipMetricsUpdater";
import { SingletonMetricsService } from "../SingletonMetricsService";
import { ISponsorshipMetricsUpdater } from "../../../../../core/metrics";
describe("SponsorshipMetricsUpdater", () => {
let sponsorshipMetricsUpdater: ISponsorshipMetricsUpdater;
beforeEach(() => {
const metricService = new SingletonMetricsService();
sponsorshipMetricsUpdater = new SponsorshipMetricsUpdater(metricService);
});
it("should get correct txFee after incrementing txFee", async () => {
const txFeeA = 82821359154819519n;
const txFeeB = 95357651471636n;
await sponsorshipMetricsUpdater.incrementSponsorshipFeePaid(txFeeA, Direction.L1_TO_L2);
await sponsorshipMetricsUpdater.incrementSponsorshipFeePaid(txFeeB, Direction.L1_TO_L2);
const txFees = await sponsorshipMetricsUpdater.getSponsorshipFeePaid(Direction.L1_TO_L2);
expect(txFees).toBe(txFeeA + txFeeB);
expect(await sponsorshipMetricsUpdater.getSponsoredMessagesTotal(Direction.L1_TO_L2)).toBe(2);
});
});

32
postman/src/application/postman/app/PostmanServiceClient.ts
View File

@@ -28,9 +28,16 @@ import { EthereumTransactionValidationService } from "../../../services/Ethereum
import { getConfig } from "./config/utils";
import { Api } from "../api/Api";
import { MessageStatusSubscriber } from "../persistence/subscribers/MessageStatusSubscriber";
import { MessageMetricsService } from "../api/metrics/MessageMetricsService";
import { SingletonMetricsService } from "../api/metrics/SingletonMetricsService";
import { MessageMetricsUpdater } from "../api/metrics/MessageMetricsUpdater";
import { IMessageMetricsUpdater, IMetricsService, ISponsorshipMetricsUpdater } from "postman/src/core/metrics";
import { SponsorshipMetricsUpdater } from "../api/metrics/SponsorshipMetricsUpdater";
export class PostmanServiceClient {
// Metrics services
private singletonMetricsService: IMetricsService;
private messageMetricsUpdater: IMessageMetricsUpdater;
private sponsorshipMetricsUpdater: ISponsorshipMetricsUpdater;
// L1 -> L2 flow
private l1MessageSentEventPoller: IPoller;
private l2MessageAnchoringPoller: IPoller;
@@ -114,6 +121,11 @@ export class PostmanServiceClient {
const lineaMessageDBService = new LineaMessageDBService(messageRepository);
const ethereumMessageDBService = new EthereumMessageDBService(l1GasProvider, messageRepository);
// Metrics services
this.singletonMetricsService = new SingletonMetricsService();
this.messageMetricsUpdater = new MessageMetricsUpdater(this.db.manager, this.singletonMetricsService);
this.sponsorshipMetricsUpdater = new SponsorshipMetricsUpdater(this.singletonMetricsService);
// L1 -> L2 flow
const l1MessageSentEventProcessor = new MessageSentEventProcessor(
@@ -205,6 +217,7 @@ export class PostmanServiceClient {
const l2MessageClaimingPersister = new MessageClaimingPersister(
lineaMessageDBService,
l2MessageServiceClient,
this.sponsorshipMetricsUpdater,
l2Provider,
{
direction: Direction.L1_TO_L2,
@@ -329,6 +342,7 @@ export class PostmanServiceClient {
const l1MessageClaimingPersister = new MessageClaimingPersister(
ethereumMessageDBService,
lineaRollupClient,
this.sponsorshipMetricsUpdater,
l1Provider,
{
direction: Direction.L2_TO_L1,
@@ -377,23 +391,27 @@ export class PostmanServiceClient {
}
}
public async initializeMetrics(): Promise<void> {}
/**
* Initializes metrics, registers subscribers, and configures the API.
* This method expects the database to be connected.
*/
public async initializeMetricsAndApi(): Promise<void> {
try {
const metricService = new MessageMetricsService(this.db.manager);
await metricService.initialize();
await this.messageMetricsUpdater.initialize();
const messageStatusSubscriber = new MessageStatusSubscriber(
metricService,
this.messageMetricsUpdater,
new WinstonLogger(MessageStatusSubscriber.name),
);
this.db.subscribers.push(messageStatusSubscriber);
// Initialize or reinitialize the API using the metrics service.
this.api = new Api({ port: this.config.apiConfig.port }, metricService, new WinstonLogger(Api.name));
this.api = new Api(
{ port: this.config.apiConfig.port },
this.singletonMetricsService,
new WinstonLogger(Api.name),
);
this.logger.info("Metrics and API have been initialized successfully.");
} catch (error) {

8
postman/src/application/postman/app/__tests__/PostmanServiceClient.test.ts
View File

@@ -27,7 +27,7 @@ import { TypeOrmMessageRepository } from "../../persistence/repositories/TypeOrm
import { L2ClaimMessageTransactionSizePoller } from "../../../../services/pollers/L2ClaimMessageTransactionSizePoller";
import { DEFAULT_MAX_CLAIM_GAS_LIMIT } from "../../../../core/constants";
import { MessageStatusSubscriber } from "../../persistence/subscribers/MessageStatusSubscriber";
import { MessageMetricsService } from "../../api/metrics/MessageMetricsService";
import { MessageMetricsUpdater } from "../../api/metrics/MessageMetricsUpdater";
import { Api } from "../../api/Api";
jest.mock("ethers", () => {
@@ -176,7 +176,7 @@ describe("PostmanServiceClient", () => {
describe("connectServices", () => {
it("should initialize API and database", async () => {
jest.spyOn(MessageMetricsService.prototype, "initialize").mockResolvedValueOnce();
jest.spyOn(MessageMetricsUpdater.prototype, "initialize").mockResolvedValueOnce();
const initializeSpy = jest.spyOn(DataSource.prototype, "initialize").mockResolvedValue(
new DataSource({
type: "postgres",
@@ -217,7 +217,7 @@ describe("PostmanServiceClient", () => {
jest.spyOn(TypeOrmMessageRepository.prototype, "getLatestMessageSent").mockImplementationOnce(jest.fn());
jest.spyOn(Api.prototype, "start").mockImplementationOnce(jest.fn());
jest.spyOn(MessageMetricsService.prototype, "initialize").mockResolvedValueOnce();
jest.spyOn(MessageMetricsUpdater.prototype, "initialize").mockResolvedValueOnce();
await postmanServiceClient.initializeMetricsAndApi();
postmanServiceClient.startAllServices();
@@ -237,7 +237,7 @@ describe("PostmanServiceClient", () => {
jest.spyOn(DatabaseCleaningPoller.prototype, "stop").mockImplementationOnce(jest.fn());
jest.spyOn(Api.prototype, "stop").mockImplementationOnce(jest.fn());
jest.spyOn(MessageMetricsService.prototype, "initialize").mockResolvedValueOnce();
jest.spyOn(MessageMetricsUpdater.prototype, "initialize").mockResolvedValueOnce();
await postmanServiceClient.initializeMetricsAndApi();
postmanServiceClient.stopAllServices();

3
postman/src/application/postman/persistence/dataSource.ts
View File

@@ -9,7 +9,7 @@ import { AddUniqueConstraint1709901138056 } from "./migrations/1709901138056-Add
import { DBOptions } from "./config/types";
import { MessageEntity } from "./entities/Message.entity";
import { AddCompressedTxSizeColumn1718026260629 } from "./migrations/1718026260629-AddCompressedTxSizeColumn";
import { AddSponsorshipMetrics1745569276097 } from "./migrations/1745569276097-AddSponsorshipMetrics";
export class DB {
public static create(config: DBOptions): DataSource {
return new DataSource({
@@ -24,6 +24,7 @@ export class DB {
AddNewIndexes1701265652528,
AddUniqueConstraint1709901138056,
AddCompressedTxSizeColumn1718026260629,
AddSponsorshipMetrics1745569276097,
],
migrationsTableName: "migrations",
logging: ["error"],

6
postman/src/application/postman/persistence/entities/Message.entity.ts
View File

@@ -1,4 +1,4 @@
import { IsDate, IsDecimal, IsEnum, IsNumber, IsString } from "class-validator";
import { IsBoolean, IsDate, IsDecimal, IsEnum, IsNumber, IsString } from "class-validator";
import { Entity, PrimaryGeneratedColumn, Column, CreateDateColumn, UpdateDateColumn } from "typeorm";
import { Direction } from "@consensys/linea-sdk";
import { MessageStatus } from "../../../../core/enums";
@@ -88,6 +88,10 @@ export class MessageEntity {
@IsNumber()
compressedTransactionSize?: number;
@Column({ default: false })
@IsBoolean()
isForSponsorship: boolean;
@CreateDateColumn()
public createdAt: Date;

2
postman/src/application/postman/persistence/mappers/__tests__/messageMappers.test.ts
View File

@@ -27,6 +27,7 @@ describe("Message Mappers", () => {
claimTxMaxPriorityFeePerGas: undefined,
claimTxNonce: undefined,
compressedTransactionSize: undefined,
isForSponsorship: false,
contractAddress: TEST_CONTRACT_ADDRESS_2,
createdAt: new Date("2023-08-04"),
destination: TEST_CONTRACT_ADDRESS_1,
@@ -61,6 +62,7 @@ describe("Message Mappers", () => {
claimTxMaxPriorityFeePerGas: undefined,
claimTxNonce: undefined,
contractAddress: TEST_CONTRACT_ADDRESS_2,
isForSponsorship: false,
createdAt: new Date("2023-08-04"),
destination: TEST_CONTRACT_ADDRESS_1,
direction: Direction.L1_TO_L2,

1
postman/src/application/postman/persistence/mappers/messageMappers.ts
View File

@@ -38,6 +38,7 @@ export const mapMessageEntityToMessage = (entity: MessageEntity): Message => {
claimLastRetriedAt: entity.claimLastRetriedAt,
claimGasEstimationThreshold: entity.claimGasEstimationThreshold,
compressedTransactionSize: entity.compressedTransactionSize,
isForSponsorship: entity.isForSponsorship,
createdAt: entity.createdAt,
updatedAt: entity.updatedAt,
});

18
postman/src/application/postman/persistence/migrations/1745569276097-AddSponsorshipMetrics.ts Normal file
View File

@@ -0,0 +1,18 @@
import { MigrationInterface, QueryRunner, TableColumn } from "typeorm";
export class AddSponsorshipMetrics1745569276097 implements MigrationInterface {
public async up(queryRunner: QueryRunner): Promise<void> {
await queryRunner.addColumns("message", [
new TableColumn({
name: "is_for_sponsorship",
type: "boolean",
isNullable: false,
default: false,
}),
]);
}
public async down(queryRunner: QueryRunner): Promise<void> {
await queryRunner.dropColumn("message", "is_for_sponsorship");
}
}

46
postman/src/application/postman/persistence/subscribers/MessageStatusSubscriber.ts
View File

@@ -8,14 +8,14 @@ import {
} from "typeorm";
import { Direction } from "@consensys/linea-sdk";
import { MessageEntity } from "../entities/Message.entity";
import { IMetricsService, LineaPostmanMetrics } from "../../../../core/metrics/IMetricsService";
import { IMessageMetricsUpdater, MessagesMetricsAttributes } from "../../../../core/metrics";
import { ILogger } from "../../../../core/utils/logging/ILogger";
import { MessageStatus } from "../../../../core/enums";
@EventSubscriber()
export class MessageStatusSubscriber implements EntitySubscriberInterface<MessageEntity> {
constructor(
private readonly metricsService: IMetricsService,
private readonly messageMetricsUpdater: IMessageMetricsUpdater,
private readonly logger: ILogger,
) {}
@@ -36,11 +36,7 @@ export class MessageStatusSubscriber implements EntitySubscriberInterface<Messag
const direction = event.databaseEntity.direction;
if (prevStatus !== newStatus) {
await this.swapStatus(
LineaPostmanMetrics.Messages,
{ status: prevStatus, direction },
{ status: newStatus, direction },
);
await this.swapMessageAttributes({ status: prevStatus, direction }, { status: newStatus, direction });
}
}
@@ -53,17 +49,22 @@ export class MessageStatusSubscriber implements EntitySubscriberInterface<Messag
async afterTransactionCommit(event: TransactionCommitEvent): Promise<void> {
const updatedEntity = event.queryRunner?.data?.updatedEntity;
if (updatedEntity) {
await this.swapStatus(
LineaPostmanMetrics.Messages,
{ status: updatedEntity.previousStatus, direction: updatedEntity.direction },
{ status: updatedEntity.newStatus, direction: updatedEntity.direction },
await this.swapMessageAttributes(
{
status: updatedEntity.previousStatus,
direction: updatedEntity.direction,
},
{
status: updatedEntity.newStatus,
direction: updatedEntity.direction,
},
);
}
}
private async updateMessageMetricsOnInsert(messageStatus: MessageStatus, messageDirection: Direction): Promise<void> {
try {
const prevGaugeValue = await this.metricsService.getGaugeValue(LineaPostmanMetrics.Messages, {
const prevGaugeValue = await this.messageMetricsUpdater.getMessageCount({
status: messageStatus,
direction: messageDirection,
});
@@ -72,7 +73,7 @@ export class MessageStatusSubscriber implements EntitySubscriberInterface<Messag
return;
}
this.metricsService.incrementGauge(LineaPostmanMetrics.Messages, {
this.messageMetricsUpdater.incrementMessageCount({
status: messageStatus,
direction: messageDirection,
});
@@ -83,13 +84,13 @@ export class MessageStatusSubscriber implements EntitySubscriberInterface<Messag
private async updateMessageMetricsOnRemove(messageStatus: MessageStatus, messageDirection: Direction): Promise<void> {
try {
const prevGaugeValue = await this.metricsService.getGaugeValue(LineaPostmanMetrics.Messages, {
const prevGaugeValue = await this.messageMetricsUpdater.getMessageCount({
status: messageStatus,
direction: messageDirection,
});
if (prevGaugeValue && prevGaugeValue > 0) {
this.metricsService.decrementGauge(LineaPostmanMetrics.Messages, {
this.messageMetricsUpdater.decrementMessageCount({
status: messageStatus,
direction: messageDirection,
});
@@ -99,23 +100,22 @@ export class MessageStatusSubscriber implements EntitySubscriberInterface<Messag
}
}
private async swapStatus(
name: LineaPostmanMetrics,
previous: Record<string, string>,
next: Record<string, string>,
private async swapMessageAttributes(
previousMessageAttributes: MessagesMetricsAttributes,
nextMessageAttributes: MessagesMetricsAttributes,
): Promise<void> {
try {
const [prevVal, newVal] = await Promise.all([
this.metricsService.getGaugeValue(name, previous),
this.metricsService.getGaugeValue(name, next),
this.messageMetricsUpdater.getMessageCount(previousMessageAttributes),
this.messageMetricsUpdater.getMessageCount(nextMessageAttributes),
]);
if (prevVal && prevVal > 0) {
this.metricsService.decrementGauge(name, previous);
this.messageMetricsUpdater.decrementMessageCount(previousMessageAttributes);
}
if (newVal !== undefined) {
this.metricsService.incrementGauge(name, next);
this.messageMetricsUpdater.incrementMessageCount(nextMessageAttributes);
}
} catch (error) {
this.logger.error("Metrics swap failed:", error);

6
postman/src/core/entities/Message.ts
View File

@@ -24,6 +24,7 @@ export type MessageProps = {
claimLastRetriedAt?: Date;
claimGasEstimationThreshold?: number;
compressedTransactionSize?: number;
isForSponsorship?: boolean;
createdAt?: Date;
updatedAt?: Date;
};
@@ -74,6 +75,7 @@ export class Message {
public claimLastRetriedAt?: Date;
public claimGasEstimationThreshold?: number;
public compressedTransactionSize?: number;
public isForSponsorship: boolean = false;
public createdAt?: Date;
public updatedAt?: Date;
@@ -100,6 +102,7 @@ export class Message {
this.claimLastRetriedAt = props.claimLastRetriedAt;
this.claimGasEstimationThreshold = props.claimGasEstimationThreshold;
this.compressedTransactionSize = props.compressedTransactionSize;
this.isForSponsorship = props.isForSponsorship ?? false;
this.createdAt = props.createdAt;
this.updatedAt = props.updatedAt;
}
@@ -122,6 +125,7 @@ export class Message {
if (newMessage.claimGasEstimationThreshold)
this.claimGasEstimationThreshold = newMessage.claimGasEstimationThreshold;
if (newMessage.compressedTransactionSize) this.compressedTransactionSize = newMessage.compressedTransactionSize;
if (newMessage.isForSponsorship !== undefined) this.isForSponsorship = newMessage.isForSponsorship;
this.updatedAt = new Date();
}
@@ -143,6 +147,6 @@ export class Message {
this.claimGasEstimationThreshold
}, compressedTransactionSize=${
this.compressedTransactionSize
}, createdAt=${this.createdAt?.toISOString()}, updatedAt=${this.updatedAt?.toISOString()})`;
}, isForSponsorship=${this.isForSponsorship}, createdAt=${this.createdAt?.toISOString()}, updatedAt=${this.updatedAt?.toISOString()})`;
}
}

12
postman/src/core/metrics/IMessageMetricsUpdater.ts Normal file
View File

@@ -0,0 +1,12 @@
import { MessagesMetricsAttributes } from "./MessageMetricsAttributes";
export interface IMessageMetricsUpdater {
// Method to initialize the message gauges
initialize(): Promise<void>;
// Method to get the current message count for given attributes
getMessageCount(messageAttributes: MessagesMetricsAttributes): Promise<number | undefined>;
// Method to increment the message count for given attributes and value
incrementMessageCount(messageAttributes: MessagesMetricsAttributes, value?: number): Promise<void>;
// Method to decrement the message count for given attributes and value
decrementMessageCount(messageAttributes: MessagesMetricsAttributes, value?: number): Promise<void>;
}

17
postman/src/core/metrics/IMetricsService.ts
View File

@@ -2,6 +2,23 @@ import { Counter, Gauge, Registry } from "prom-client";
export enum LineaPostmanMetrics {
Messages = "linea_postman_messages",
// Example PromQL query for hourly rate of sponsored messages 'rate(linea_postman_sponsored_messages_total{direction="L1_TO_L2",app="postman"}[60m]) * 3600'
SponsoredMessagesTotal = "linea_postman_sponsored_messages_total",
/**
* Tx fees in wei paid by Postman for sponsored message claims
*
* Workaround for prom-client metrics not supporting bigint type
* - We split txFee into GWEI and WEI components, and accumulate in two separate metrics
* - Note JS limitation of Number.MAX_SAFE_INTEGER = 9007199254740991
* - Given 150,000 sponsored messages a year, we should not reach overflow in <60 years
*
* We do not use separate labels for 'wei' and 'gwei' denominations, because metrics sharing the label should be aggregatable
* - I.e. metric (direction: A, denomination: wei) cannot be aggregated with metric (direction: A, denomination: gwei) because they represent different units
*
* Example PromQL query to get hourly rate of ETH consumed for sponsoring messages - 'rate(linea_postman_sponsorship_fees_gwei_total{direction="L1_TO_L2", app="postman"}[60m]) * 3600 / 1e9 + rate(linea_postman_sponsorship_fees_wei_total{direction="L1_TO_L2", app="postman"}[60m]) * 3600 / 1e18
*/
SponsorshipFeesWei = "linea_postman_sponsorship_fees_wei_total", // Represent up to ~9_007_199 GWEI
SponsorshipFeesGwei = "linea_postman_sponsorship_fees_gwei_total", // Represent up to ~9_007_199 ETH
}
export interface IMetricsService {

7
postman/src/core/metrics/ISponsorshipMetricsUpdater.ts Normal file
View File

@@ -0,0 +1,7 @@
import { Direction } from "@consensys/linea-sdk";
export interface ISponsorshipMetricsUpdater {
getSponsoredMessagesTotal(direction: Direction): Promise<number>;
getSponsorshipFeePaid(direction: Direction): Promise<bigint>;
incrementSponsorshipFeePaid(txFee: bigint, direction: Direction): Promise<void>;
}

7
postman/src/core/metrics/MessageMetricsAttributes.ts Normal file
View File

@@ -0,0 +1,7 @@
import { MessageStatus } from "../enums";
import { Direction } from "@consensys/linea-sdk";
export type MessagesMetricsAttributes = {
status: MessageStatus;
direction: Direction;
};

4
postman/src/core/metrics/index.ts Normal file
View File

@@ -0,0 +1,4 @@
export * from "./IMessageMetricsUpdater";
export * from "./IMetricsService";
export * from "./ISponsorshipMetricsUpdater";
export * from "./MessageMetricsAttributes";

26
postman/src/services/EthereumTransactionValidationService.ts
View File

@@ -77,11 +77,7 @@ export class EthereumTransactionValidationService implements ITransactionValidat
const isUnderPriced = this.isUnderPriced(gasLimit, message.fee, maxFeePerGas);
const hasZeroFee = this.hasZeroFee(message);
const isRateLimitExceeded = await this.isRateLimitExceeded(message.fee, message.value);
const isForSponsorship = this.isForSponsorship(
gasLimit,
this.config.isPostmanSponsorshipEnabled,
this.config.maxPostmanSponsorGasLimit,
);
const isForSponsorship = this.isForSponsorship(gasLimit, hasZeroFee, isUnderPriced);
return {
hasZeroFee,
@@ -156,16 +152,16 @@ export class EthereumTransactionValidationService implements ITransactionValidat
* Determines if the claim transaction is for sponsorship
*
* @param {bigint} gasLimit - The gas limit for the transaction.
* @param {boolean} isPostmanSponsorshipEnabled - `true` if Postman sponsorship is enabled, `false` otherwise
* @param {bigint} maxPostmanSponsorGasLimit - Maximum gas limit for sponsored Postman claim transactions
* @returns {boolean} `true` if the message is for sponsorsing, `false` otherwise.
* @param {boolean} hasZeroFee - `true` if the message has zero fee, `false` otherwise.
* @param {boolean} isUnderPriced - `true` if the transaction is underpriced, `false` otherwise.
* @returns {boolean} `true` if the message is for sponsoring, `false` otherwise.
*/
private isForSponsorship(
gasLimit: bigint,
isPostmanSponsorshipEnabled: boolean,
maxPostmanSponsorGasLimit: bigint,
): boolean {
if (!isPostmanSponsorshipEnabled) return false;
return gasLimit < maxPostmanSponsorGasLimit;
private isForSponsorship(gasLimit: bigint, hasZeroFee: boolean, isUnderPriced: boolean): boolean {
if (!this.config.isPostmanSponsorshipEnabled) return false;
if (gasLimit > this.config.maxPostmanSponsorGasLimit) return false;
if (hasZeroFee) return true;
if (isUnderPriced) return true;
// The message would be claimed regardless of sponsorship settings
return false;
}
}

26
postman/src/services/LineaTransactionValidationService.ts
View File

@@ -85,11 +85,7 @@ export class LineaTransactionValidationService implements ITransactionValidation
const isUnderPriced = await this.isUnderPriced(gasLimit, message.fee, message.compressedTransactionSize!);
const hasZeroFee = this.hasZeroFee(message);
const isRateLimitExceeded = await this.isRateLimitExceeded(message.fee, message.value);
const isForSponsorship = this.isForSponsorship(
gasLimit,
this.config.isPostmanSponsorshipEnabled,
this.config.maxPostmanSponsorGasLimit,
);
const isForSponsorship = this.isForSponsorship(gasLimit, hasZeroFee, isUnderPriced);
return {
hasZeroFee,
@@ -159,17 +155,17 @@ export class LineaTransactionValidationService implements ITransactionValidation
* Determines if the claim transaction is for sponsorship
*
* @param {bigint} gasLimit - The gas limit for the transaction.
* @param {boolean} isPostmanSponsorshipEnabled - `true` if Postman sponsorship is enabled, `false` otherwise
* @param {bigint} maxPostmanSponsorGasLimit - Maximum gas limit for sponsored Postman claim transactions
* @returns {boolean} `true` if the message is for sponsorsing, `false` otherwise.
* @param {boolean} hasZeroFee - `true` if the message has zero fee, `false` otherwise.
* @param {boolean} isUnderPriced - `true` if the transaction is underpriced, `false` otherwise.
* @returns {boolean} `true` if the message is for sponsoring, `false` otherwise.
*/
private isForSponsorship(
gasLimit: bigint,
isPostmanSponsorshipEnabled: boolean,
maxPostmanSponsorGasLimit: bigint,
): boolean {
if (!isPostmanSponsorshipEnabled) return false;
return gasLimit < maxPostmanSponsorGasLimit;
private isForSponsorship(gasLimit: bigint, hasZeroFee: boolean, isUnderPriced: boolean): boolean {
if (!this.config.isPostmanSponsorshipEnabled) return false;
if (gasLimit > this.config.maxPostmanSponsorGasLimit) return false;
if (hasZeroFee) return true;
if (isUnderPriced) return true;
// The message would be claimed regardless of sponsorship settings
return false;
}
/**

71
postman/src/services/processors/MessageClaimingPersister.ts
View File

@@ -21,6 +21,7 @@ import {
} from "../../core/services/processors/IMessageClaimingPersister";
import { IMessageDBService } from "../../core/persistence/IMessageDBService";
import { ErrorParser } from "../../utils/ErrorParser";
import { ISponsorshipMetricsUpdater } from "../../core/metrics";
export class MessageClaimingPersister implements IMessageClaimingPersister {
private messageBeingRetry: { message: Message | null; retries: number };
@@ -30,6 +31,7 @@ export class MessageClaimingPersister implements IMessageClaimingPersister {
*
* @param {IMessageDBService} databaseService - An instance of a class implementing the `IMessageDBService` interface, used for storing and retrieving message data.
* @param {IMessageServiceContract} messageServiceContract - An instance of a class implementing the `IMessageServiceContract` interface, used to interact with the blockchain contract.
* @param {ISponsorshipMetricsUpdater} sponsorshipMetricsUpdater - An instance of a class implementing the `ISponsorshipMetricsUpdater` interface, update sponsorship metrics for Prometheus monitoring.
* @param {IProvider} provider - An instance of a class implementing the `IProvider` interface, used to query blockchain data.
* @param {MessageClaimingPersisterConfig} config - Configuration for network-specific settings, including transaction submission timeout and maximum transaction retries.
* @param {ILogger} logger - An instance of a class implementing the `ILogger` interface, used for logging messages.
@@ -43,6 +45,7 @@ export class MessageClaimingPersister implements IMessageClaimingPersister {
ContractTransactionResponse,
ErrorDescription
>,
private readonly sponsorshipMetricsUpdater: ISponsorshipMetricsUpdater,
private readonly provider: IProvider<
TransactionReceipt,
Block,
@@ -85,33 +88,31 @@ export class MessageClaimingPersister implements IMessageClaimingPersister {
}
const receipt = await this.provider.getTransactionReceipt(firstPendingMessage.claimTxHash);
if (!receipt) {
if (this.isMessageExceededSubmissionTimeout(firstPendingMessage)) {
this.logger.warn("Retrying to claim message: messageHash=%s", firstPendingMessage.messageHash);
if (receipt) {
await this.updateReceiptStatus(firstPendingMessage, receipt);
} else {
if (!this.isMessageExceededSubmissionTimeout(firstPendingMessage)) return;
this.logger.warn("Retrying to claim message: messageHash=%s", firstPendingMessage.messageHash);
if (
!this.messageBeingRetry.message ||
this.messageBeingRetry.message.messageHash !== firstPendingMessage.messageHash
) {
this.messageBeingRetry = { message: firstPendingMessage, retries: 0 };
}
const transactionReceipt = await this.retryTransaction(
firstPendingMessage.claimTxHash,
firstPendingMessage.messageHash,
);
if (transactionReceipt) {
this.logger.warn(
"Retried claim message transaction succeed: messageHash=%s transactionHash=%s",
firstPendingMessage.messageHash,
transactionReceipt.hash,
);
}
if (
!this.messageBeingRetry.message ||
this.messageBeingRetry.message.messageHash !== firstPendingMessage.messageHash
) {
this.messageBeingRetry = { message: firstPendingMessage, retries: 0 };
}
return;
}
await this.updateReceiptStatus(firstPendingMessage, receipt);
const retryTransactionReceipt = await this.retryTransaction(
firstPendingMessage.claimTxHash,
firstPendingMessage.messageHash,
);
if (!retryTransactionReceipt) return;
this.logger.warn(
"Retried claim message transaction succeed: messageHash=%s transactionHash=%s",
firstPendingMessage.messageHash,
retryTransactionReceipt.hash,
);
await this.updateReceiptStatus(firstPendingMessage, retryTransactionReceipt);
}
} catch (e) {
const error = ErrorParser.parseErrorWithMitigation(e);
this.logger.error("Error processing message.", {
@@ -157,13 +158,6 @@ export class MessageClaimingPersister implements IMessageClaimingPersister {
const tx = await this.messageServiceContract.retryTransactionWithHigherFee(transactionHash);
const receipt = await tx.wait();
if (!receipt) {
throw new BaseError(
`RetryTransaction: Transaction receipt not found after retry transaction. transactionHash=${tx.hash}`,
);
}
this.messageBeingRetry.message?.edit({
claimTxGasLimit: parseInt(tx.gasLimit.toString()),
claimTxMaxFeePerGas: tx.maxFeePerGas ?? undefined,
@@ -175,6 +169,12 @@ export class MessageClaimingPersister implements IMessageClaimingPersister {
});
await this.databaseService.updateMessage(this.messageBeingRetry.message!);
const receipt = await tx.wait();
if (!receipt) {
throw new BaseError(
`RetryTransaction: Transaction receipt not found after retry transaction. transactionHash=${tx.hash}`,
);
}
return receipt;
} catch (e) {
this.logger.error(
@@ -227,8 +227,15 @@ export class MessageClaimingPersister implements IMessageClaimingPersister {
return;
}
message.edit({ status: MessageStatus.CLAIMED_SUCCESS });
message.edit({
status: MessageStatus.CLAIMED_SUCCESS,
});
await this.databaseService.updateMessage(message);
if (message.isForSponsorship)
await this.sponsorshipMetricsUpdater.incrementSponsorshipFeePaid(
BigInt(receipt.gasPrice) * BigInt(receipt.gasUsed),
message.direction,
);
this.logger.info(
"Message has been SUCCESSFULLY claimed: messageHash=%s transactionHash=%s",
message.messageHash,

2
postman/src/services/processors/MessageClaimingProcessor.ts
View File

@@ -104,7 +104,7 @@ export class MessageClaimingProcessor implements IMessageClaimingProcessor {
if (!isForSponsorship && (await this.handleZeroFee(hasZeroFee, nextMessageToClaim))) return;
if (await this.handleNonExecutable(nextMessageToClaim, estimatedGasLimit)) return;
nextMessageToClaim.edit({ claimGasEstimationThreshold: threshold });
nextMessageToClaim.edit({ claimGasEstimationThreshold: threshold, isForSponsorship });
await this.databaseService.updateMessage(nextMessageToClaim);
if (

362
postman/src/services/processors/__tests__/MessageClaimingPersister.test.ts
View File

@@ -21,6 +21,7 @@ import { IMessageClaimingPersister } from "../../../core/services/processors/IMe
import { MessageClaimingPersister } from "../MessageClaimingPersister";
import { EthereumMessageDBService } from "../../persistence/EthereumMessageDBService";
import { IProvider } from "../../../core/clients/blockchain/IProvider";
import { ISponsorshipMetricsUpdater } from "postman/src/core/metrics";
describe("TestMessageClaimingPersister ", () => {
let messageClaimingPersister: IMessageClaimingPersister;
@@ -36,6 +37,7 @@ describe("TestMessageClaimingPersister ", () => {
> &
IGasProvider<TransactionRequest>
>();
const sponsorshipMetricsUpdater = mock<ISponsorshipMetricsUpdater>();
const provider =
mock<IProvider<TransactionReceipt, Block, TransactionRequest, TransactionResponse, JsonRpcProvider>>();
const logger = new TestLogger(MessageClaimingPersister.name);
@@ -44,6 +46,7 @@ describe("TestMessageClaimingPersister ", () => {
messageClaimingPersister = new MessageClaimingPersister(
databaseService,
l2MessageServiceContractMock,
sponsorshipMetricsUpdater,
provider,
{
direction: Direction.L1_TO_L2,
@@ -62,10 +65,104 @@ describe("TestMessageClaimingPersister ", () => {
jest.resetAllMocks();
});
type TestFixtureFactoryOpts = {
firstPendingMessage?: Message | null;
secondPendingMessage?: Message | null;
txReceipt?: TransactionReceipt | null;
txReceiptError?: Error;
retryTxReceipt?: TransactionReceipt | null;
isRateLimitExceededError?: boolean;
firstOnChainMessageStatus?: OnChainMessageStatus;
secondOnChainMessageStatus?: OnChainMessageStatus;
retryTransactionWithHigherFeeResponse?: TransactionResponse;
retryTransactionWithHigherFeeError?: Error;
retryTransactionWithHigherFeeReceipt?: TransactionReceipt | null;
};
type TestFixture = {
l2QuerierGetReceiptSpy: jest.SpyInstance<ReturnType<typeof provider.getTransactionReceipt>>;
loggerErrorSpy: jest.SpyInstance<ReturnType<typeof logger.error>>;
loggerWarnSpy: jest.SpyInstance<ReturnType<typeof logger.warn>>;
loggerInfoSpy: jest.SpyInstance<ReturnType<typeof logger.info>>;
messageRepositoryUpdateSpy: jest.SpyInstance<ReturnType<typeof databaseService.updateMessage>>;
};
const testFixtureFactory = (opts: TestFixtureFactoryOpts): TestFixture => {
const {
firstPendingMessage,
secondPendingMessage,
txReceipt,
txReceiptError,
retryTxReceipt,
isRateLimitExceededError,
firstOnChainMessageStatus,
secondOnChainMessageStatus,
retryTransactionWithHigherFeeResponse,
retryTransactionWithHigherFeeError,
retryTransactionWithHigherFeeReceipt,
} = opts;
if (firstPendingMessage !== undefined && secondPendingMessage !== undefined) {
jest
.spyOn(databaseService, "getFirstPendingMessage")
.mockResolvedValueOnce(firstPendingMessage)
.mockResolvedValueOnce(secondPendingMessage);
} else if (firstPendingMessage !== undefined) {
jest.spyOn(databaseService, "getFirstPendingMessage").mockResolvedValue(firstPendingMessage);
}
if (txReceiptError !== undefined) {
jest.spyOn(provider, "getTransactionReceipt").mockRejectedValue(txReceiptError);
} else if (txReceipt !== undefined) {
jest.spyOn(provider, "getTransactionReceipt").mockResolvedValue(txReceipt);
}
if (isRateLimitExceededError !== undefined)
jest.spyOn(l2MessageServiceContractMock, "isRateLimitExceededError").mockResolvedValue(isRateLimitExceededError);
if (retryTxReceipt !== undefined)
jest.spyOn(provider, "getTransactionReceipt").mockResolvedValueOnce(null).mockResolvedValueOnce(retryTxReceipt);
if (firstOnChainMessageStatus !== undefined && secondOnChainMessageStatus !== undefined) {
jest
.spyOn(l2MessageServiceContractMock, "getMessageStatus")
.mockResolvedValueOnce(firstOnChainMessageStatus)
.mockResolvedValueOnce(secondOnChainMessageStatus);
} else if (firstOnChainMessageStatus !== undefined) {
jest.spyOn(l2MessageServiceContractMock, "getMessageStatus").mockResolvedValue(firstOnChainMessageStatus);
}
if (retryTransactionWithHigherFeeError !== undefined) {
jest
.spyOn(l2MessageServiceContractMock, "retryTransactionWithHigherFee")
.mockRejectedValue(retryTransactionWithHigherFeeError);
} else if (
retryTransactionWithHigherFeeResponse !== undefined &&
retryTransactionWithHigherFeeReceipt !== undefined
) {
jest
.spyOn(l2MessageServiceContractMock, "retryTransactionWithHigherFee")
.mockResolvedValue(retryTransactionWithHigherFeeResponse);
jest.spyOn(retryTransactionWithHigherFeeResponse, "wait").mockResolvedValue(retryTransactionWithHigherFeeReceipt);
}
jest.spyOn(provider, "getBlockNumber").mockResolvedValue(100);
jest.spyOn(sponsorshipMetricsUpdater, "incrementSponsorshipFeePaid").mockResolvedValue();
const l2QuerierGetReceiptSpy = jest.spyOn(provider, "getTransactionReceipt");
const loggerErrorSpy = jest.spyOn(logger, "error");
const loggerWarnSpy = jest.spyOn(logger, "warn");
const loggerInfoSpy = jest.spyOn(logger, "info");
const messageRepositoryUpdateSpy = jest.spyOn(databaseService, "updateMessage");
return {
l2QuerierGetReceiptSpy,
loggerErrorSpy,
loggerWarnSpy,
loggerInfoSpy,
messageRepositoryUpdateSpy,
};
};
describe("process", () => {
it("Should return if getTransactionReceipt return null", async () => {
const l2QuerierGetReceiptSpy = jest.spyOn(provider, "getTransactionReceipt");
jest.spyOn(databaseService, "getFirstPendingMessage").mockResolvedValue(null);
it("Should early return immediately if no pending message found", async () => {
const { l2QuerierGetReceiptSpy } = testFixtureFactory({
firstPendingMessage: null,
});
await messageClaimingPersister.process();
@@ -73,11 +170,12 @@ describe("TestMessageClaimingPersister ", () => {
});
it("Should log as error if getTransactionReceipt throws error", async () => {
const testPendingMessageLocal = new Message(testPendingMessage);
const getTxReceiptError = new Error("error for testing");
const loggerErrorSpy = jest.spyOn(logger, "error");
jest.spyOn(databaseService, "getFirstPendingMessage").mockResolvedValue(testPendingMessageLocal);
jest.spyOn(provider, "getTransactionReceipt").mockRejectedValue(getTxReceiptError);
const testPendingMessageLocal = new Message(testPendingMessage);
const { loggerErrorSpy } = testFixtureFactory({
firstPendingMessage: testPendingMessageLocal,
txReceiptError: getTxReceiptError,
});
await messageClaimingPersister.process();
@@ -91,22 +189,22 @@ describe("TestMessageClaimingPersister ", () => {
it("Should log as info and update message as claimed success if successful", async () => {
const txReceipt = testingHelpers.generateTransactionReceipt({ status: 1 });
const loggerInfoSpy = jest.spyOn(logger, "info");
const l2QuerierGetReceiptSpy = jest.spyOn(provider, "getTransactionReceipt").mockResolvedValue(txReceipt);
const messageRepositorySaveSpy = jest.spyOn(databaseService, "updateMessage");
const testPendingMessageLocal = new Message(testPendingMessage);
jest.spyOn(databaseService, "getFirstPendingMessage").mockResolvedValue(testPendingMessageLocal);
const expectedSavedMessage = new Message({
...testPendingMessageLocal,
status: MessageStatus.CLAIMED_SUCCESS,
updatedAt: mockedDate,
});
const { loggerInfoSpy, messageRepositoryUpdateSpy, l2QuerierGetReceiptSpy } = testFixtureFactory({
firstPendingMessage: new Message(testPendingMessage),
txReceipt: txReceipt,
});
await messageClaimingPersister.process();
expect(l2QuerierGetReceiptSpy).toHaveBeenCalledTimes(1);
expect(messageRepositorySaveSpy).toHaveBeenCalledTimes(1);
expect(messageRepositorySaveSpy).toHaveBeenCalledWith(expectedSavedMessage);
expect(messageRepositoryUpdateSpy).toHaveBeenCalledTimes(1);
expect(messageRepositoryUpdateSpy).toHaveBeenCalledWith(expectedSavedMessage);
expect(loggerInfoSpy).toHaveBeenCalledTimes(1);
expect(loggerInfoSpy).toHaveBeenCalledWith(
"Message has been SUCCESSFULLY claimed: messageHash=%s transactionHash=%s",
@@ -117,43 +215,44 @@ describe("TestMessageClaimingPersister ", () => {
it("Should return and update message as sent if receipt status is 0 and rate limit exceeded", async () => {
const txReceipt = testingHelpers.generateTransactionReceipt({ status: 0 });
const l2QuerierGetReceiptSpy = jest.spyOn(provider, "getTransactionReceipt").mockResolvedValue(txReceipt);
const messageRepositorySaveSpy = jest.spyOn(databaseService, "updateMessage");
const testPendingMessageLocal = new Message(testPendingMessage);
jest.spyOn(databaseService, "getFirstPendingMessage").mockResolvedValue(testPendingMessageLocal);
jest.spyOn(l2MessageServiceContractMock, "isRateLimitExceededError").mockResolvedValue(true);
const expectedSavedMessage = new Message({
...testPendingMessageLocal,
...testPendingMessage,
status: MessageStatus.SENT,
updatedAt: mockedDate,
});
await messageClaimingPersister.process();
expect(l2QuerierGetReceiptSpy).toHaveBeenCalledTimes(1);
expect(messageRepositorySaveSpy).toHaveBeenCalledTimes(1);
expect(messageRepositorySaveSpy).toHaveBeenCalledWith(expectedSavedMessage);
});
it("Should log as warning and update message as claimed reverted if receipt status is 0", async () => {
const loggerWarnSpy = jest.spyOn(logger, "warn");
const txReceipt = testingHelpers.generateTransactionReceipt({ status: 0 });
const l2QuerierGetReceiptSpy = jest.spyOn(provider, "getTransactionReceipt").mockResolvedValue(txReceipt);
const messageRepositorySaveSpy = jest.spyOn(databaseService, "updateMessage");
const testPendingMessageLocal = new Message(testPendingMessage);
jest.spyOn(databaseService, "getFirstPendingMessage").mockResolvedValue(testPendingMessageLocal);
jest.spyOn(l2MessageServiceContractMock, "isRateLimitExceededError").mockResolvedValue(false);
const expectedSavedMessage = new Message({
...testPendingMessageLocal,
status: MessageStatus.CLAIMED_REVERTED,
updatedAt: mockedDate,
const { messageRepositoryUpdateSpy, l2QuerierGetReceiptSpy } = testFixtureFactory({
firstPendingMessage: testPendingMessageLocal,
txReceipt: txReceipt,
isRateLimitExceededError: true,
});
await messageClaimingPersister.process();
expect(l2QuerierGetReceiptSpy).toHaveBeenCalledTimes(1);
expect(messageRepositorySaveSpy).toHaveBeenCalledTimes(1);
expect(messageRepositorySaveSpy).toHaveBeenCalledWith(expectedSavedMessage);
expect(messageRepositoryUpdateSpy).toHaveBeenCalledTimes(1);
expect(messageRepositoryUpdateSpy).toHaveBeenCalledWith(expectedSavedMessage);
});
it("Should log as warning and update message as claim reverted if receipt status is 0", async () => {
const txReceipt = testingHelpers.generateTransactionReceipt({ status: 0 });
const testPendingMessageLocal = new Message(testPendingMessage);
const expectedSavedMessage = new Message({
...testPendingMessage,
status: MessageStatus.CLAIMED_REVERTED,
updatedAt: mockedDate,
});
const { loggerWarnSpy, l2QuerierGetReceiptSpy, messageRepositoryUpdateSpy } = testFixtureFactory({
firstPendingMessage: testPendingMessageLocal,
txReceipt,
isRateLimitExceededError: false,
});
console.log("boobies");
await messageClaimingPersister.process();
expect(l2QuerierGetReceiptSpy).toHaveBeenCalledTimes(1);
expect(messageRepositoryUpdateSpy).toHaveBeenCalledTimes(1);
expect(messageRepositoryUpdateSpy).toHaveBeenCalledWith(expectedSavedMessage);
expect(loggerWarnSpy).toHaveBeenCalledTimes(1);
expect(loggerWarnSpy).toHaveBeenCalledWith(
"Message claim transaction has been REVERTED: messageHash=%s transactionHash=%s",
@@ -162,52 +261,53 @@ describe("TestMessageClaimingPersister ", () => {
);
});
it("Should return and log as warning if message is claimed and receipt was retrieved successfully", async () => {
const loggerWarnSpy = jest.spyOn(logger, "warn");
const txReceipt = testingHelpers.generateTransactionReceipt({ status: 1 });
const l2QuerierGetReceiptSpy = jest
.spyOn(provider, "getTransactionReceipt")
.mockResolvedValueOnce(null)
.mockResolvedValueOnce(txReceipt);
const messageRepositorySaveSpy = jest.spyOn(databaseService, "saveMessages");
it("Should update message as claimed if retry receipt successful and message claimed on-chain", async () => {
const retryTxReceipt = testingHelpers.generateTransactionReceipt({ status: 1 });
const testPendingMessageLocal = new Message(testPendingMessage);
jest.spyOn(databaseService, "getFirstPendingMessage").mockResolvedValue(testPendingMessageLocal);
jest.spyOn(provider, "getBlockNumber").mockResolvedValue(100);
jest.spyOn(l2MessageServiceContractMock, "isRateLimitExceededError").mockResolvedValue(false);
jest.spyOn(l2MessageServiceContractMock, "getMessageStatus").mockResolvedValue(OnChainMessageStatus.CLAIMED);
const expectedSavedMessage = new Message({
...testPendingMessageLocal,
status: MessageStatus.CLAIMED_SUCCESS,
updatedAt: mockedDate,
});
const { loggerWarnSpy, messageRepositoryUpdateSpy, l2QuerierGetReceiptSpy } = testFixtureFactory({
firstPendingMessage: new Message(testPendingMessageLocal),
retryTxReceipt,
isRateLimitExceededError: false,
firstOnChainMessageStatus: OnChainMessageStatus.CLAIMED,
});
await messageClaimingPersister.process();
expect(l2QuerierGetReceiptSpy).toHaveBeenCalledTimes(2);
expect(messageRepositorySaveSpy).toHaveBeenCalledTimes(0);
expect(messageRepositoryUpdateSpy).toHaveBeenCalledTimes(1);
expect(messageRepositoryUpdateSpy).toHaveBeenCalledWith(expectedSavedMessage);
expect(loggerWarnSpy).toHaveBeenCalledTimes(2);
expect(loggerWarnSpy).toHaveBeenNthCalledWith(
1,
"Retrying to claim message: messageHash=%s",
testPendingMessageLocal.messageHash,
testPendingMessage.messageHash,
);
expect(loggerWarnSpy).toHaveBeenNthCalledWith(
2,
"Retried claim message transaction succeed: messageHash=%s transactionHash=%s",
testPendingMessageLocal.messageHash,
txReceipt.hash,
testPendingMessage.messageHash,
retryTxReceipt.hash,
);
});
it("Should return and log as warning if message is claimed but receipt returned as null", async () => {
const loggerWarnSpy = jest.spyOn(logger, "warn");
const l2QuerierGetReceiptSpy = jest.spyOn(provider, "getTransactionReceipt").mockResolvedValue(null);
const messageRepositorySaveSpy = jest.spyOn(databaseService, "saveMessages");
const testPendingMessageLocal = new Message(testPendingMessage);
jest.spyOn(databaseService, "getFirstPendingMessage").mockResolvedValue(testPendingMessageLocal);
jest.spyOn(provider, "getBlockNumber").mockResolvedValue(100);
jest.spyOn(l2MessageServiceContractMock, "isRateLimitExceededError").mockResolvedValue(false);
jest.spyOn(l2MessageServiceContractMock, "getMessageStatus").mockResolvedValue(OnChainMessageStatus.CLAIMED);
const { loggerWarnSpy, messageRepositoryUpdateSpy, l2QuerierGetReceiptSpy } = testFixtureFactory({
firstPendingMessage: testPendingMessageLocal,
txReceipt: null,
isRateLimitExceededError: false,
firstOnChainMessageStatus: OnChainMessageStatus.CLAIMED,
});
await messageClaimingPersister.process();
expect(l2QuerierGetReceiptSpy).toHaveBeenCalledTimes(2);
expect(messageRepositorySaveSpy).toHaveBeenCalledTimes(0);
expect(messageRepositoryUpdateSpy).toHaveBeenCalledTimes(0);
expect(loggerWarnSpy).toHaveBeenCalledTimes(2);
expect(loggerWarnSpy).toHaveBeenNthCalledWith(
1,
@@ -223,30 +323,38 @@ describe("TestMessageClaimingPersister ", () => {
});
it("Should return and log as warning if message is claimable and retry tx was sent successfully", async () => {
const loggerWarnSpy = jest.spyOn(logger, "warn");
const txReceipt = testingHelpers.generateTransactionReceipt({ status: 1 });
const txResponse = testingHelpers.generateTransactionResponse({
const retryTxReceipt = testingHelpers.generateTransactionReceipt({ status: 1 });
const retryTxResponse = testingHelpers.generateTransactionResponse({
maxPriorityFeePerGas: undefined,
maxFeePerGas: undefined,
});
const l2QuerierGetReceiptSpy = jest.spyOn(provider, "getTransactionReceipt").mockResolvedValue(null);
const messageRepositorySaveSpy = jest.spyOn(databaseService, "updateMessage");
const testPendingMessageLocal = new Message(testPendingMessage);
jest.spyOn(databaseService, "getFirstPendingMessage").mockResolvedValue(testPendingMessageLocal);
jest.spyOn(provider, "getBlockNumber").mockResolvedValue(100);
jest.spyOn(l2MessageServiceContractMock, "isRateLimitExceededError").mockResolvedValue(false);
jest
.spyOn(l2MessageServiceContractMock, "getMessageStatus")
.mockResolvedValueOnce(OnChainMessageStatus.CLAIMABLE);
jest.spyOn(l2MessageServiceContractMock, "retryTransactionWithHigherFee").mockResolvedValue(txResponse);
jest.spyOn(txResponse, "wait").mockResolvedValue(txReceipt);
const expectedSavedMessage = new Message({
...testPendingMessageLocal,
status: MessageStatus.CLAIMED_SUCCESS,
updatedAt: mockedDate,
claimTxNonce: retryTxResponse.nonce,
claimTxGasLimit: Number(retryTxResponse.gasLimit),
claimNumberOfRetry: 1,
claimLastRetriedAt: mockedDate,
});
const { loggerWarnSpy, messageRepositoryUpdateSpy, l2QuerierGetReceiptSpy } = testFixtureFactory({
firstPendingMessage: testPendingMessageLocal,
txReceipt: null,
isRateLimitExceededError: false,
firstOnChainMessageStatus: OnChainMessageStatus.CLAIMABLE,
retryTransactionWithHigherFeeResponse: retryTxResponse,
retryTransactionWithHigherFeeReceipt: retryTxReceipt,
});
await messageClaimingPersister.process();
expect(l2QuerierGetReceiptSpy).toHaveBeenCalledTimes(1);
expect(messageRepositorySaveSpy).toHaveBeenCalledTimes(1);
expect(messageRepositorySaveSpy).toHaveBeenCalledWith(testPendingMessageLocal);
expect(messageRepositoryUpdateSpy).toHaveBeenCalledTimes(2);
expect(messageRepositoryUpdateSpy).toHaveBeenNthCalledWith(1, testPendingMessageLocal);
expect(messageRepositoryUpdateSpy).toHaveBeenNthCalledWith(2, expectedSavedMessage);
// expect(messageRepositoryUpdateSpy).toHaveBeenCalledWith(testPendingMessageLocal);
expect(loggerWarnSpy).toHaveBeenCalledTimes(3);
expect(loggerWarnSpy).toHaveBeenNthCalledWith(
1,
@@ -263,40 +371,45 @@ describe("TestMessageClaimingPersister ", () => {
3,
"Retried claim message transaction succeed: messageHash=%s transactionHash=%s",
testPendingMessageLocal.messageHash,
txReceipt.hash,
retryTxReceipt.hash,
);
});
it("Should return and log as warning if message is claimable and retry tx was sent successfully", async () => {
const loggerWarnSpy = jest.spyOn(logger, "warn");
const txReceipt = testingHelpers.generateTransactionReceipt({ status: 1 });
const txResponse = testingHelpers.generateTransactionResponse({
it("Should update DB successfully if first process claimable message with receipt, then process claimed message with no receipt", async () => {
const retryTxReceipt = testingHelpers.generateTransactionReceipt({ status: 1 });
const retryTxResponse = testingHelpers.generateTransactionResponse({
maxPriorityFeePerGas: undefined,
maxFeePerGas: undefined,
});
const l2QuerierGetReceiptSpy = jest.spyOn(provider, "getTransactionReceipt").mockResolvedValue(null);
const messageRepositorySaveSpy = jest.spyOn(databaseService, "updateMessage");
const testPendingMessageLocal = new Message(testPendingMessage);
const testPendingMessageLocal2 = new Message(testPendingMessage2);
jest
.spyOn(databaseService, "getFirstPendingMessage")
.mockResolvedValueOnce(testPendingMessageLocal)
.mockResolvedValueOnce(testPendingMessageLocal2);
jest.spyOn(provider, "getBlockNumber").mockResolvedValue(100);
jest.spyOn(l2MessageServiceContractMock, "isRateLimitExceededError").mockResolvedValue(false);
jest
.spyOn(l2MessageServiceContractMock, "getMessageStatus")
.mockResolvedValueOnce(OnChainMessageStatus.CLAIMABLE)
.mockResolvedValueOnce(OnChainMessageStatus.CLAIMED);
jest.spyOn(l2MessageServiceContractMock, "retryTransactionWithHigherFee").mockResolvedValue(txResponse);
jest.spyOn(txResponse, "wait").mockResolvedValue(txReceipt);
const expectedSavedMessage = new Message({
...testPendingMessageLocal,
status: MessageStatus.CLAIMED_SUCCESS,
updatedAt: mockedDate,
claimTxNonce: retryTxResponse.nonce,
claimTxGasLimit: Number(retryTxResponse.gasLimit),
claimNumberOfRetry: 1,
claimLastRetriedAt: mockedDate,
});
const { loggerWarnSpy, messageRepositoryUpdateSpy, l2QuerierGetReceiptSpy } = testFixtureFactory({
firstPendingMessage: testPendingMessageLocal,
secondPendingMessage: testPendingMessageLocal2,
txReceipt: null,
isRateLimitExceededError: false,
firstOnChainMessageStatus: OnChainMessageStatus.CLAIMABLE,
secondOnChainMessageStatus: OnChainMessageStatus.CLAIMED,
retryTransactionWithHigherFeeResponse: retryTxResponse,
retryTransactionWithHigherFeeReceipt: retryTxReceipt,
});
await messageClaimingPersister.process();
await messageClaimingPersister.process();
expect(l2QuerierGetReceiptSpy).toHaveBeenCalledTimes(3);
expect(messageRepositorySaveSpy).toHaveBeenCalledTimes(1);
expect(messageRepositorySaveSpy).toHaveBeenCalledWith(testPendingMessageLocal);
expect(messageRepositoryUpdateSpy).toHaveBeenCalledTimes(2);
expect(messageRepositoryUpdateSpy).toHaveBeenNthCalledWith(1, testPendingMessageLocal);
expect(messageRepositoryUpdateSpy).toHaveBeenNthCalledWith(2, expectedSavedMessage);
expect(loggerWarnSpy).toHaveBeenCalledTimes(5);
expect(loggerWarnSpy).toHaveBeenNthCalledWith(
1,
@@ -313,7 +426,7 @@ describe("TestMessageClaimingPersister ", () => {
3,
"Retried claim message transaction succeed: messageHash=%s transactionHash=%s",
testPendingMessageLocal.messageHash,
txReceipt.hash,
retryTxReceipt.hash,
);
});
@@ -321,6 +434,7 @@ describe("TestMessageClaimingPersister ", () => {
messageClaimingPersister = new MessageClaimingPersister(
databaseService,
l2MessageServiceContractMock,
sponsorshipMetricsUpdater,
provider,
{
direction: Direction.L1_TO_L2,
@@ -329,22 +443,20 @@ describe("TestMessageClaimingPersister ", () => {
},
logger,
);
const loggerWarnSpy = jest.spyOn(logger, "warn");
const loggerErrorSpy = jest.spyOn(logger, "error");
const l2QuerierGetReceiptSpy = jest.spyOn(provider, "getTransactionReceipt").mockResolvedValue(null);
const messageRepositorySaveSpy = jest.spyOn(databaseService, "saveMessages");
const testPendingMessageLocal = new Message(testPendingMessage);
const retryError = new Error("error for testing");
jest.spyOn(databaseService, "getFirstPendingMessage").mockResolvedValue(testPendingMessageLocal);
jest.spyOn(provider, "getBlockNumber").mockResolvedValue(100);
jest.spyOn(l2MessageServiceContractMock, "isRateLimitExceededError").mockResolvedValue(false);
jest.spyOn(l2MessageServiceContractMock, "getMessageStatus").mockResolvedValue(OnChainMessageStatus.CLAIMABLE);
jest.spyOn(l2MessageServiceContractMock, "retryTransactionWithHigherFee").mockRejectedValue(retryError);
const { loggerWarnSpy, loggerErrorSpy, messageRepositoryUpdateSpy, l2QuerierGetReceiptSpy } = testFixtureFactory({
firstPendingMessage: testPendingMessageLocal,
txReceipt: null,
isRateLimitExceededError: false,
firstOnChainMessageStatus: OnChainMessageStatus.CLAIMABLE,
retryTransactionWithHigherFeeError: retryError,
});
await messageClaimingPersister.process();
expect(l2QuerierGetReceiptSpy).toHaveBeenCalledTimes(1);
expect(messageRepositorySaveSpy).toHaveBeenCalledTimes(0);
expect(messageRepositoryUpdateSpy).toHaveBeenCalledTimes(0);
expect(loggerErrorSpy).toHaveBeenCalledTimes(2);
expect(loggerErrorSpy).toHaveBeenNthCalledWith(
1,
@@ -371,24 +483,24 @@ describe("TestMessageClaimingPersister ", () => {
);
});
it("Should return and log as error if message is claimable but tx response wait returns null", async () => {
const loggerWarnSpy = jest.spyOn(logger, "warn");
const loggerErrorSpy = jest.spyOn(logger, "error");
const txResponse = testingHelpers.generateTransactionResponse();
const l2QuerierGetReceiptSpy = jest.spyOn(provider, "getTransactionReceipt").mockResolvedValue(null);
const messageRepositorySaveSpy = jest.spyOn(databaseService, "saveMessages");
it("Should return and log as error if retry tx fails to get receipt", async () => {
const retryTxResponse = testingHelpers.generateTransactionResponse();
const testPendingMessageLocal = new Message(testPendingMessage);
jest.spyOn(databaseService, "getFirstPendingMessage").mockResolvedValue(testPendingMessageLocal);
jest.spyOn(provider, "getBlockNumber").mockResolvedValue(100);
jest.spyOn(l2MessageServiceContractMock, "isRateLimitExceededError").mockResolvedValue(false);
jest.spyOn(l2MessageServiceContractMock, "getMessageStatus").mockResolvedValue(OnChainMessageStatus.CLAIMABLE);
jest.spyOn(l2MessageServiceContractMock, "retryTransactionWithHigherFee").mockResolvedValue(txResponse);
jest.spyOn(txResponse, "wait").mockResolvedValue(null);
jest.spyOn(l2MessageServiceContractMock, "retryTransactionWithHigherFee").mockResolvedValue(retryTxResponse);
const { loggerWarnSpy, loggerErrorSpy, messageRepositoryUpdateSpy, l2QuerierGetReceiptSpy } = testFixtureFactory({
firstPendingMessage: testPendingMessageLocal,
txReceipt: null,
isRateLimitExceededError: false,
firstOnChainMessageStatus: OnChainMessageStatus.CLAIMABLE,
retryTransactionWithHigherFeeResponse: retryTxResponse,
retryTransactionWithHigherFeeReceipt: null,
});
await messageClaimingPersister.process();
expect(l2QuerierGetReceiptSpy).toHaveBeenCalledTimes(1);
expect(messageRepositorySaveSpy).toHaveBeenCalledTimes(0);
expect(messageRepositoryUpdateSpy).toHaveBeenCalledTimes(1);
expect(messageRepositoryUpdateSpy).toHaveBeenCalledWith(testPendingMessageLocal);
expect(loggerWarnSpy).toHaveBeenCalledTimes(2);
expect(loggerWarnSpy).toHaveBeenNthCalledWith(
1,

3
postman/src/services/processors/__tests__/MessageClaimingProcessor.test.ts
View File

@@ -424,6 +424,7 @@ describe("TestMessageClaimingProcessor", () => {
...testAnchoredMessage,
claimGasEstimationThreshold: 10000000000,
updatedAt: mockedDate,
isForSponsorship: false,
});
await messageClaimingProcessor.process();
@@ -451,6 +452,7 @@ describe("TestMessageClaimingProcessor", () => {
const expectedLoggingMessage = new Message({
...testZeroFeeAnchoredMessage,
updatedAt: mockedDate,
isForSponsorship: true,
});
await messageClaimingProcessor.process();
@@ -479,6 +481,7 @@ describe("TestMessageClaimingProcessor", () => {
...testUnderpricedAnchoredMessage,
claimGasEstimationThreshold: 10,
updatedAt: mockedDate,
isForSponsorship: true,
});
await messageClaimingProcessor.process();

2
postman/src/utils/testing/helpers.ts
View File

@@ -44,6 +44,7 @@ export const generateMessage = (overrides?: Partial<MessageProps>): Message => {
direction: Direction.L1_TO_L2,
status: MessageStatus.SENT,
claimNumberOfRetry: 0,
isForSponsorship: false,
createdAt: new Date("2023-08-04"),
updatedAt: new Date("2023-08-04"),
...overrides,
@@ -65,6 +66,7 @@ export const generateMessageEntity = (overrides?: Partial<MessageEntity>): Messa
direction: Direction.L1_TO_L2,
status: MessageStatus.SENT,
claimNumberOfRetry: 0,
isForSponsorship: false,
createdAt: new Date("2023-08-04"),
updatedAt: new Date("2023-08-04"),
...overrides,