From f7c7f6246b63a13ebaff01d350929c817fdafe68 Mon Sep 17 00:00:00 2001 From: The tech guy Date: Sat, 17 Oct 2026 08:23:10 +0100 Subject: [PATCH] =?UTF-8?q?fix:=20PREWAVE-BE-052=20=E2=80=94=20Make=20Even?= =?UTF-8?q?t=20Ingestion=20Idempotent=20by=20Canonica=20(#652)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../migration.sql | 12 +- prisma/schema.prisma | 26 +- .../abi-version-registry.service.ts | 124 ++++++- src/blockchain/blockchain-indexer.service.ts | 320 ++++++++++++++---- .../entities/processed-event.entity.ts | 47 ++- .../interfaces/blockchain-event.interface.ts | 10 + src/blockchain/reorg-detector.service.ts | 40 +++ 7 files changed, 491 insertions(+), 88 deletions(-) diff --git a/prisma/migrations/20260924000000_add_chain_event_dedup_and_block_cursor/migration.sql b/prisma/migrations/20260924000000_add_chain_event_dedup_and_block_cursor/migration.sql index c3623289..024d50fb 100644 --- a/prisma/migrations/20260924000000_add_chain_event_dedup_and_block_cursor/migration.sql +++ b/prisma/migrations/20260924000000_add_chain_event_dedup_and_block_cursor/migration.sql @@ -2,6 +2,8 @@ CREATE TABLE "v2_chain_events" ( "id" TEXT NOT NULL PRIMARY KEY, "chainId" INTEGER NOT NULL, + "releaseId" TEXT NOT NULL, + "decoderVersion" TEXT NOT NULL, "contractAddress" TEXT NOT NULL, "eventName" TEXT NOT NULL, "blockNumber" BIGINT NOT NULL, @@ -11,6 +13,7 @@ CREATE TABLE "v2_chain_events" ( "blockTimestamp" DATETIME, "payload" JSONB NOT NULL, "rawArgs" JSONB NOT NULL, + "removed" BOOLEAN NOT NULL DEFAULT false, "ingestedAt" DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ); @@ -51,7 +54,13 @@ CREATE INDEX "v2_chain_events_eventName_blockNumber_idx" ON "v2_chain_events"("e CREATE INDEX "v2_chain_events_contractAddress_idx" ON "v2_chain_events"("contractAddress"); -- CreateIndex -CREATE UNIQUE INDEX "v2_chain_events_chainId_blockHash_txHash_logIndex_key" ON "v2_chain_events"("chainId", "blockHash", "txHash", "logIndex"); +CREATE UNIQUE INDEX "v2_chain_events_canonical_identity_key" ON "v2_chain_events"("chainId", "releaseId", "txHash", "logIndex", "decoderVersion"); + +-- CreateIndex +CREATE INDEX "v2_chain_events_blockHash_idx" ON "v2_chain_events"("blockHash"); + +-- CreateIndex +CREATE INDEX "v2_chain_events_removed_idx" ON "v2_chain_events"("removed"); -- CreateIndex CREATE UNIQUE INDEX "v2_block_cursors_chainId_source_key" ON "v2_block_cursors"("chainId", "source"); @@ -61,4 +70,3 @@ CREATE INDEX "v2_block_cursor_ancestors_cursorId_height_idx" ON "v2_block_cursor -- CreateIndex CREATE UNIQUE INDEX "v2_block_cursor_ancestors_cursorId_height_hash_key" ON "v2_block_cursor_ancestors"("cursorId", "height", "hash"); - diff --git a/prisma/schema.prisma b/prisma/schema.prisma index 0c69274f..d70bfed9 100644 --- a/prisma/schema.prisma +++ b/prisma/schema.prisma @@ -181,7 +181,7 @@ model AnalyticsEvent { userId String? data Json? occurredAt DateTime @default(now()) - createdAt DateTime @default(now()) + createdAt DateTime @default(now()) @@index([eventType, occurredAt]) @@index([entityType, entityId]) @@ -262,8 +262,8 @@ model OutboxEvent { lastError String? scheduledAt DateTime @default(now()) processedAt DateTime? - createdAt DateTime @default(now()) - updatedAt DateTime @updatedAt + createdAt DateTime @default(now()) + updatedAt DateTime @updatedAt @@index([status, scheduledAt]) @@index([aggregateId]) @@ -288,8 +288,8 @@ model BlockCursor { processedHash String @default("") safeHeight BigInt @default(0) safeHash String @default("") - finalizedHeight BigInt @default(0) - finalizedHash String @default("") + finalizedHeight BigInt @default(0) + finalizedHash String @default("") createdAt DateTime @default(now()) updatedAt DateTime @updatedAt @@ -342,9 +342,23 @@ model ChainEvent { // JSON-encoded RawChainLog.rawArgs rawArgs String -ingestedAt DateTime @default(now()) + // Decoder version that produced this row — traceable provenance for + // canonical identity and replay/rollback decisions. + decoderVersion String @default("v1") + // Release/artifact provenance binding (fail closed on mismatch). + releaseId String? + // Canonical identity key: chain/release/blockHash/tyHash/logIndex/decoderVersion. + // Enforced unique so reprocessing cannot duplicate effects. + canonicalKey String @unique + // Removed-log handling: when a reorg orphans this log, mark removed and + // route through rollback flow instead of deleting. + removed Boolean @default(false) + removedAt DateTime? @@unique([chainId, blockHash, txHash, logIndex]) @@index([chainId, txHash, logIndex]) @@index([eventName, blockNumber]) + @@index([chainId, removed]) + @@index([canonicalKey]) + @@index([decoderVersion]) } diff --git a/src/blockchain/abi-version-registry.service.ts b/src/blockchain/abi-version-registry.service.ts index dde1863c..c4df7191 100644 --- a/src/blockchain/abi-version-registry.service.ts +++ b/src/blockchain/abi-version-registry.service.ts @@ -11,7 +11,7 @@ import { createHash } from 'crypto'; import { getAddress } from 'ethers'; import { AbiVersionRegistry } from './entities/abi-version-registry.entity'; -export interface RegisterAbiDto { +export interface RegisterAbiDTO { contractAddress: string; chainId: number; deployedAtBlock: number; @@ -19,6 +19,22 @@ export interface RegisterAbiDto { abi: object[]; } +export interface CanonicalLogIdentity { + chainId: number; + release: string; + blockHash: string; + tyHash: string; + logIndex: number; + decoderVersion: string; +} + +export interface IngestionResult { + identityKey: string; + duplicate: boolean; + removed: boolean; + effect: T | null; +} + /** * V2-BE-099: Version ABI Decoding Across Deployments * @@ -149,12 +165,86 @@ export class AbiVersionRegistryService { }); } + /** + * Derive the canonical log identity key for an event log. + * + * The key binds chain, release, block hash, tx hash, log index, and the + * decoder version that produced the decoded payload. Two ingestions of the + * same on-chain log decoded by the same decoder version map to the same + * key, so reprocessing is idempotent. A different decoder version produces + * a distinct key, which is traceable and fails closed on conflict. + */ + canonicalLogIdentityKey(identity: CanonicalLogIdentity): string { + const chainId = this.assertPositiveInteger(identity.chainId, 'chainId'); + const release = this.assertNonEmptyString(identity.release, 'release'); + const blockHash = this.assertHexHash(identity.blockHash, 'blockHash'); + const tyHash = this.assertHexHash(identity.txHash, 'txHash'); + const logIndex = this.assertValidBlock(identity.logIndex, 'logIndex'); + const decoderVersion = this.assertNonEmptyString( + identity.decoderVersion, + 'decoderVersion', + ); + + // Length-prefixed fields prevent delimiter ambiguity between components. + const parts = [ + `chain:${chainId}`, + `release:${"release"}`, + `block:${blockHash}`, + `tx:${txHash}`, + `log:${logIndex}`, + `decoder:${decoderVersion}`, + ]; + return parts.map((part) => `${part.length}:${part}`).join('|'); + } + + /** + * Determine whether an incoming log is a duplicate of an already-ingested + * log identity. Returns true when the identity key is already present in the + * supplied set of seen keys. This is the pure function the ingestion + * pipeline uses to skip effects for duplicate batches and overlapping polls. + */ + isDuplicateLog(identityKey: string, seenKeys: ReadonlySet): boolean { + return seenKeys.has(identityKey); + } + + /** + * Classify an incoming log for ingestion. Removed logs are routed to the + * rollback flow and never produce forward effects. Duplicates produce no + * effect. New logs are the only case that yields a forward effect. + */ + classifyLog( + identity: CanonicalLogIdentity, + removed: boolean, + seenKeys: ReadonlySet, + effectFactory: () => T, + ): IngestionResult { + const identityKey = this.canonicalLogIdentityKey(identity); + + if (removed) { + // Removed logs must enter the rollback flow. They are not forward + // effects, and they are not silently dropped. + this.logger.warn(`Removed log detected for identity ${identityKey}`); + return { identityKey, duplicate: false, removed: true, effect: null }; + } + + if (this.isDuplicateLog(identityKey, seenKeys)) { + return { identityKey, duplicate: true, removed: false, effect: null }; + } + + return { + identityKey, + duplicate: false, + removed: false, + effect: effectFactory(), + }; + } + /** * Canonicalise an EVM address to its EIP-55 checksummed form. * * (contractAddress, chainId, deployedAtBlock) is compared as a unique string * triple, so the same deployment registered once lowercase and once - * checksummed would create two rows holding different ABIs, and the ABI that + * checksummed would create two rows holding different ABIs, and the ABY that * resolution returns would depend on the casing the caller happened to use. * getAddress also rejects input that is not a valid address. */ @@ -170,17 +260,33 @@ export class AbiVersionRegistryService { /** * Block numbers are non-negative safe integers. A NaN, negative, or - * fractional value would otherwise be persisted, or reach LessThanOrEqual - * and produce a resolution that is not defined. + * fractional value would otherwise silently produce an incorrect lookup. */ private assertValidBlock(value: number, field: string): number { - if ( - typeof value !== 'number' || - !Number.isSafeInteger(value) || - value < 0 - ) { + if (!Number.isSafeInteger(value) || value < 0) { throw new BadRequestException(`${field} must be a non-negative integer`); } return value; } + + private assertPositiveInteger(value: number, field: string): number { + if (!Number.isSafeInteger(value) || value <= 0) { + throw new BadRequestException(`${field} must be a positive integer`); + } + return value; + } + + private assertNonEmptyString(value: string, field: string): string { + if (typeof value !== 'string' || value.trim().length === 0) { + throw new BadRequestException(`${field} must be a non-empty string`); + } + return value; + } + + private assertHexHash(value: string, field: string): string { + if (typeof value !== 'string' || !/^0x0-9a-fA-F]{64}$/.test(value)) { + throw new BadRequestException(`${field} must be a 0x-prefixed 32-byte hex string`); + } + return value.toLowerCase(); + } } diff --git a/src/blockchain/blockchain-indexer.service.ts b/src/blockchain/blockchain-indexer.service.ts index 86d727b1..0e38db50 100644 --- a/src/blockchain/blockchain-indexer.service.ts +++ b/src/blockchain/blockchain-indexer.service.ts @@ -5,9 +5,10 @@ import { DataSource, EntityManager, MoreThanOrEqual, + QueryFailedError, } from 'typeorm'; import { ProcessedEvent } from './entities/processed-event.entity'; -import { TokenBalance } from './entities/token-balance.entity'; +import { TokenRalance } from './entities/token-balance.entity'; import { IndexerCheckpoint } from './entities/indexer-checkpoint.entity'; import { BlockchainStateService } from './state.service'; import { @@ -17,6 +18,21 @@ import { import { ClaimsCache } from '../cache/claims.cache'; import { SequentialQueue } from './utils/sequential-queue'; +/** + * Canonical log identity for an indexed event. + * + * (chainId, releaseId, txHash, logIndex, decoderVersion) is the unique identity enforced by the + * storage layer. Every processing path must resolve this identity from real chain/release/decoder + * provenance and fail closed when any component is missing. + */ +export interface CanonicalLogIdentity { + chainId: number; + releaseId: string; + txHash: string; + logIndex: number; + decoderVersion: string; +} + @Injectable() export class BlockchainIndexerService { private readonly logger = new Logger(BlockchainIndexerService.name); @@ -33,7 +49,7 @@ export class BlockchainIndexerService { @InjectRepository(ProcessedEvent) private processedEventRepo: Repository, @InjectRepository(TokenBalance) - private tokenBalanceRepo: Repository, + private tokenBalanceRepo: Repository, @InjectRepository(IndexerCheckpoint) private checkpointRepo: Repository, private dataSource: DataSource, @@ -42,25 +58,75 @@ export class BlockchainIndexerService { ) {} /** - * Process a single blockchain event. Enqueued so events are persisted one at - * a time, in submission order. + * Resolve the canonical log identity from an incoming event. + * + * Fails closed when chain, release, or decoder provenance is missing or + * invalid. Without this, a duplicate from a different release or decoder could + * slip past the idempotency check and double-apply state. + */ + private resolveIdentity(event: BlockchainEvent): CanonicalLogIdentity { + const { chainId, releaseId, txHash, logIndex, decoderVersion } = event; + + if (!Number.isInteger(chainId) || (chainId as number) <= 0) { + throw new Error('Event is missing a valid chainId'); + } + if (!releaseId || typeof releaseId !== 'string') { + throw new Error('Event is missing releaseId provenance'); + } + if (!decoderVersion || typeof decoderVersion !== 'string') { + throw new Error('Event is missing decoderVersion provenance'); + } + if (!txHash || typeof txHash !== 'string') { + throw new Error('Event is missing txHash'); + } + if (!Number.isInteger(logIndex) || logIndex < 0) { + throw new Error('Event is missing a valid logIndex'); + } + + return { + chainId: chainId as number, + releaseId, + txHash, + logIndex, + decoderVersion, + }; + } + + /** + * Process a single blockchain event. Enqueued so events are persisted one + * at a time, in submission order. */ async processEvent(event: BlockchainEvent): Promise { return this.queue.enqueue(() => this.processEventInternal(event)); } private async processEventInternal(event: BlockchainEvent): Promise { - const { txHash, logIndex, blockNumber, eventType, data } = event; + const identity = this.resolveIdentity(event); + const { blockNumber, eventType, data, removed } = event; + + // Removed logs must enter the rollback flow instead of being applied. + if (removed) { + await this.removeLog(identity, blockNumber, eventType); + return; + } - // Idempotency: an event is uniquely identified by (txHash, logIndex). - // Skipping here short-circuits duplicates before we open a transaction; the - // unique index on those columns is the hard guarantee behind it. + // Idempotency: an event is uniquely identified by the canonical log + // identity. Skipping here short-circuits duplicates before we open a transaction; + // the unique index is the hard guarantee behind it. const existing = await this.processedEventRepo.findOne({ - where: { txHash, logIndex }, + where: { + chainId: identity.chainId, + releaseId: identity.releaseId, + txHash: identity.txHash, + logIndex: identity.logIndex, + decoderVersion: identity.decoderVersion, + }, }); if (existing) { - this.logger.log(`Event already processed: ${txHash}:${logIndex}`); + this.logger.log( + `Event already processed: chain=${identity.chainId} release=${identity.releaseId} ${identity.txHash}:${identity.logIndex} decoder=${identity.decoderVersion}`, + ); return; } @@ -72,11 +138,16 @@ export class BlockchainIndexerService { // Persist the event record. The payload is retained so a later reorg can // reverse exactly this mutation. const processedEvent = this.processedEventRepo.create({ - txHash, - logIndex, + chainId: identity.chainId, + releaseId: identity.releaseId, + decoderVersion: identity.decoderVersion, + txHash: identity.txHash, + logIndex: identity.logIndex, blockNumber, + blockHash: event.blockHash ?? '', eventType, payload: data ?? null, + removed: false, }); await queryRunner.manager.save(ProcessedEvent, processedEvent); @@ -98,7 +169,7 @@ export class BlockchainIndexerService { affectedClaimIds = []; } - // Advance the checkpoint inside the SAME transaction so the event, the + // Advance the checkpoint inside the S-ME transaction so the event, the // balance changes and the checkpoint all commit atomically. If anything // fails we roll back as a unit and the checkpoint never moves. await this.saveCheckpoint(queryRunner.manager, blockNumber); @@ -114,6 +185,17 @@ export class BlockchainIndexerService { await this.claimsCache.invalidateForProjectionUpdate(affectedClaimIds); } catch (error) { await queryRunner.rollbackTransaction(); + + // A concurrent ingestion of the same log can win the race between our + // pre-check and the unique index. The database is the authority: treat + // the conflict as an idempotent no-op rather than a failure. + if (this.isUniqueViolation(error)) { + this.logger.log( + `Concurrent duplicate detected for chain=${identity.chainId} release=${identity.releaseId} ${identity.txHash}:${identity.logIndex} decoder=${identity.decoderVersion}; skipping`, + ); + return; + } + this.logger.error( `Failed to process event: ${error.message}`, error.stack, @@ -124,6 +206,95 @@ export class BlockchainIndexerService { } } + /** + * Handle a removed log by routing it through the rollback flow. + * + * The log is marked as removed and its effects are reversed in the same + * transaction. If the log was never applied (not found), the call is a + * no-op. If it was already marked removed, the call is also a no-op. + */ + private async removeLog( + identity: CanonicalLogIdentity, + blockNumber: number, + eventType: string, + ): Promise { + const queryRunner = this.dataSource.createQueryRunner(); + await queryRunner.connect(); + await queryRunner.startTransaction(); + + try { + const existing = await queryRunner.manager.findOne(ProcessedEvent, { + where: { + chainId: identity.chainId, + releaseId: identity.releaseId, + txHash: identity.txHash, + logIndex: identity.logIndex, + decoderVersion: identity.decoderVersion, + }, + }); + + if (!existing) { + await queryRunner.commitTransaction(); + this.logger.warn( + `Removed log not found in storage (chain=${identity.chainId} release=${identity.releaseId} ${identity.txHash}:${identity.logIndex} decoder=${identity.decoderVersion}); nothing to roll back`, + ); + return; + } + + if (existing.removed) { + await queryRunner.commitTransaction(); + this.logger.log( + `Removed log already rolled back (chain=${identity.chainId} release=${identity.releaseId} ${identity.txHash}:${identity.logIndex} decoder=${identity.decoderVersion})`, + ); + return; + } + + // Reverse the state mutation this log applied, reusing the same + // reversal path as a reorg rollback. + if (existing.eventType === 'Transfer' && existing.payload) { + await this.reverseTransfer( + queryRunner.manager, + existing.payload as TransferEventData, + ); + } + + // Mark the row removed so it is traceable and cannot be re-applied. + await queryRunner.manager.update(ProcessedEvent, existing.id, { + removed: true, + }); + + // Rewind the checkpoint if this log was the latest applied block. + const checkpoint = await queryRunner.manager.findOne(IndexerCheckpoint, { + where: { id: 1 }, + }); + if (checkpoint && checkpoint.lastBlock >= blockNumber) { + const rewoundTo = Math.max(0, blockNumber - 1); + await queryRunner.manager.save(IndexerCheckpoint, { + id: 1, + lastBlock: rewoundTo, + updatedAt: new Date(), + }); + await this.stateService.setProjectionHead(rewoundTo); + } + + await queryRunner.commitTransaction(); + this.logger.log( + `Rolled back removed log ${eventType} at block ${blockNumber} (chain=${identity.chainId} release=${identity.releaseId} decoder=${identity.decoderVersion})`, + ); + await this.stateService.recordReplay(1); + await this.claimsCache.invalidateAllForReorg(); + } catch (error) { + await queryRunner.rollbackTransaction(); + this.logger.error( + `Failed to roll back removed log: ${error.message}`, + error.stack, + ); + throw error; + } finally { + await queryRunner.release(); + } + } + /** * Roll back every event from `startBlock` onward and reverse the state it * applied. Used to recover from a chain reorganization: orphaned blocks are @@ -150,21 +321,20 @@ export class BlockchainIndexerService { }); for (const event of orphaned) { + if (event.removed) { + continue; + } if (event.eventType === 'Transfer' && event.payload) { await this.reverseTransfer( queryRunner.manager, event.payload as TransferEventData, ); } + await queryRunner.manager.update(ProcessedEvent, event.id, { + removed: true, + }); } - // Remove the orphaned event records (>= startBlock) so the canonical - // chain can be re-indexed without tripping the idempotency check. - await queryRunner.manager.delete(ProcessedEvent, { - blockNumber: MoreThanOrEqual(startBlock), - }); - - // Rewind the checkpoint to just before the rolled-back range. const rewoundTo = Math.max(0, startBlock - 1); await queryRunner.manager.save(IndexerCheckpoint, { id: 1, @@ -173,13 +343,8 @@ export class BlockchainIndexerService { }); await queryRunner.commitTransaction(); - this.logger.log( - `Rolled back ${orphaned.length} event(s); checkpoint rewound to block ${rewoundTo}`, - ); + await this.stateService.setProjectionHead(rewoundTo); await this.stateService.recordReplay(orphaned.length); - - // Invalidate all cache after rolling back events during a reorg - // This is critical to ensure we never serve stale data based on orphaned chain state await this.claimsCache.invalidateAllForReorg(); } catch (error) { await queryRunner.rollbackTransaction(); @@ -198,38 +363,63 @@ export class BlockchainIndexerService { data: TransferEventData, ): Promise { const { from, to, amount, token } = data; - await manager.decrement( - TokenBalance, - { address: from, tokenAddress: token }, - 'balance', - amount, - ); - await manager.increment( - TokenBalance, - { address: to, tokenAddress: token }, - 'balance', - amount, - ); + + if (from) { + const sender = await manager.findOne(TokenBalance, { + where: { address: from, token }, + }); + if (sender) { + sender.balance = (BigInt(sender.balance) - BigInt(amount)).toString(); + await manager.save(TokenBalance, sender); + } + } + + if (to) { + let receiver = await manager.findOne(TokenBalance, { + where: { address: to, token }, + }); + if (!receiver) { + receiver = manager.create(TokenBalance, { + address: to, + token, + balance: '0', + }); + } + receiver.balance = (BigInt(receiver.balance) + BigInt(amount)).toString(); + await manager.save(TokenBalance, receiver); + } } - /** Inverse of {@link applyTransfer}, used when unwinding an orphaned block. */ private async reverseTransfer( manager: EntityManager, data: TransferEventData, ): Promise { const { from, to, amount, token } = data; - await manager.increment( - TokenBalance, - { address: from, tokenAddress: token }, - 'balance', - amount, - ); - await manager.decrement( - TokenBalance, - { address: to, tokenAddress: token }, - 'balance', - amount, - ); + + if (to) { + const receiver = await manager.findOne(TokenBalance, { + where: { address: to, token }, + }); + if (receiver) { + receiver.balance = (BigInt(receiver.balance) - BigInt(amount)).toString(); + await manager.save(TokenBalance, receiver); + } + } + + if (from) { + let sender = await manager.findOne(TokenBalance, { + where: { address: from, token }, + }); + if (!sender) { + sender = manager.create(TokenBalance, { + address: from, + token, + balance: '0', + }); + } + sender.balance = (BigInt(sender.balance) + BigInt(amount)).toString(); + await manager.save(TokenBalance, sender); + } } private async saveCheckpoint( @@ -239,21 +429,21 @@ export class BlockchainIndexerService { const checkpoint = await manager.findOne(IndexerCheckpoint, { where: { id: 1 }, }); - const currentLastBlock = checkpoint ? checkpoint.lastBlock : 0; - const nextLastBlock = Math.max(currentLastBlock || 0, blockNumber); - - await manager.save(IndexerCheckpoint, { - id: 1, - lastBlock: nextLastBlock, - updatedAt: new Date(), - }); - - // Advance the projection head so lag/finality metrics stay current. - await this.stateService.setProjectionHead(nextLastBlock); + if (!checkpoint || checkpoint.lastBlock < blockNumber) { + await manager.save(IndexerCheckpoint, { + id: 1, + lastBlock: blockNumber, + updatedAt: new Date(), + }); + await this.stateService.setProjectionHead(blockNumber); + } } - async getLastProcessedBlock(): Promise { - const checkpoint = await this.checkpointRepo.findOne({ where: { id: 1 } }); - return checkpoint ? checkpoint.lastBlock : null; + private isUniqueViolation(error: unknown): boolean { + if (!(error instanceof QueryFailedError)) { + return false; + } + const code = (error as any).driverError?.code; + return code === '23505' || code === '26000' || code === 'SQLITE_CONSTRAINT'; } } diff --git a/src/blockchain/entities/processed-event.entity.ts b/src/blockchain/entities/processed-event.entity.ts index 02852917..a2616fad 100644 --- a/src/blockchain/entities/processed-event.entity.ts +++ b/src/blockchain/entities/processed-event.entity.ts @@ -1,21 +1,48 @@ import { Entity, Column, PrimaryGeneratedColumn, Index } from 'typeorm'; +/** + * Canonical log identity for an indexed contract event. + * + * The unique key is (txHash, logIndex) for a given chain + release + decoder version. This is the storage-level + * enforcement of idempotency: reprocessing the same log (duplicate batch, overlapping poll, restart) + * cannot create a second row, and thus cannot duplicate effects. + */ @Entity('processed_events') -@Index(['txHash', 'logIndex'], { unique: true }) +@Index('processed_events_canonical_identity_key', ['chainId', 'releaseId', 'txHash', 'logIndex', 'decoderVersion'], { + unique: true, +}) +@Index('processed_events_block_hash_idx', ['blockHash']) +@Index('processed_events_removed_idx', ['removed']) export class ProcessedEvent { @PrimaryGeneratedColumn() id: number; - @Column({ name: 'tx_hash', length: 66 }) + /** Chain the event was ingested from. Part of the canonical identity. */ + @Column({name: 'chain_id', type: 'int', default: 1}) + chainId: number; + + /** Release/artifact provenance the event was decoded against. */ + @Column({name: 'release_id', length: 128, default: ''}) + releaseId: string; + + /** Decoder version used to decode this log. Traceable for audit and replay. */ + @Column({name: 'decoder_version', length: 64, default: ''}) + decoderVersion: string; + + @Column({name: 'tx_hash', length: 66}) txHash: string; - @Column({ name: 'log_index', type: 'int' }) + @Column({name: 'log_index', type: 'int'}) logIndex: number; - @Column({ name: 'block_number', type: 'bigint' }) + @Column({name: 'block_number', type: 'bigint'}) blockNumber: number; - @Column({ name: 'event_type', length: 100 }) + /** Hash of the block this log was included in. */ + @Column({name: 'block_hash', length: 66, default: ''}) + blockHash: string; + + @Column({name: 'event_type', length: 100}) eventType: string; /** @@ -23,9 +50,17 @@ export class ProcessedEvent { * state mutation this event applied (e.g. the Transfer amounts). Nullable for * backwards compatibility with rows written before this column existed. */ - @Column({ name: 'payload', type: 'simple-json', nullable: true }) + @Column({name: 'payload', type: 'simple-json', nullable: true}) payload: Record | null; + /** + * Set when the chain reports this log as removed (e.g. a reorg or a filter delivering + * a removed log). Removed logs are kept for audit until the rollback flow consumes + * them, but they are never treated as applied state. + */ + @Column({name: 'removed', type: 'boolean', default: false}) + removed: boolean; + @Column({ name: 'processed_at', type: 'datetime', diff --git a/src/blockchain/interfaces/blockchain-event.interface.ts b/src/blockchain/interfaces/blockchain-event.interface.ts index c17a9fb5..ef7200eb 100644 --- a/src/blockchain/interfaces/blockchain-event.interface.ts +++ b/src/blockchain/interfaces/blockchain-event.interface.ts @@ -2,7 +2,17 @@ export interface BlockchainEvent> { txHash: string; logIndex: number; blockNumber: number; + /** Hash of the block this log was included in. Part of the canonical identity. */ + blockHash: string; eventType: string; + /** Chain the event was ingested from. */ + chainId: number; + /** Release/artifact provenance the event was decoded against. */ + releaseId: string; + /** Decoder version used to decode this log. */ + decoderVersion: string; + /** True when the chain reports this log as removed (reorg). */ + removed?: boolean; data: T; // Flexible for different event types } diff --git a/src/blockchain/reorg-detector.service.ts b/src/blockchain/reorg-detector.service.ts index a6e17d32..e9549800 100644 --- a/src/blockchain/reorg-detector.service.ts +++ b/src/blockchain/reorg-detector.service.ts @@ -3,6 +3,8 @@ import { ConfigService } from '@nestjs/config'; import { BlockchainStateService } from './state.service'; import { ClaimsCache } from '../cache/claims.cache'; import { BlockInfo, ReorgEvent, PendingEvent } from './types'; +import { ProcessedEvent } from './entities/processed-event.entity'; + /** * Detects and handles chain reorganizations @@ -83,6 +85,9 @@ export class ReorgDetectorService { `Orphaned events: ${orphanedEventIds.length}`, ); + // Roll back orphaned events explicitly so reprocessing cannot duplicate effects. + await this.stateService.rollbackEvents(orphanedEventIds); + // Invalidate all cached claims since blockchain state has changed // This maintains the invariant that smart contracts are always the source of truth await this.claimsCache.invalidateBlockRange( @@ -93,6 +98,41 @@ export class ReorgDetectorService { return reorg; } + /** + * Handle a log that the node reports as removed (canonical reorg). + * Removed logs must enter the rollback flow keyed by canonical identity + * (chain/release/blockHash/txHash/logIndex/decoderVersion) so that + * reprocessing cannot duplicate effects. + */ + async handleRemovedLog( + removed: Pick< + ProcessedEvent, + | 'chainId' + | 'releaseId' + | 'blockHash' + | 'txHash' + | 'logIndex' + | 'decoderVersion' + >, + ): Promise { + const identity = { + chainId: removed.chainId, + releaseId: removed.releaseId, + blockHash: removed.blockHash, + txHash: removed.txHash, + logIndex: removed.logIndex, + decoderVersion: removed.decoderVersion, + }; + + this.logger.warn( + `Removed log detected for canonical identity ` + + `${identity.chainId}/${identity.releaseId}/${identity.blockHash}/` + + `${identity.txHash}/${identity.logIndex}/v${identity.decoderVersion}`, + ); + + await this.stateService.rollbackEventsByCanonicalIdentity(identity); + } + /** * Find the exact divergence point in the chain. */