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 54fbae30..da34d21e 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 @@ -1,18 +1,21 @@ -- Enforce monotonic block cursor compare-and-swap and auditable rollbacks. -- CreateTable -CREATE TABLE v2_chain_events ( - id TEXT NOT NULL PRIMARY KEY, - chainId INTEGER NOT NULL, - contractAddress TEXT NOT NULL, - eventName TEXT NOT NULL, - blockNumber BIGINT NOT NULL, - blockHash TEXT NOT NULL, - txHash TEXT NOT NULL, - logIndex INTEGER NOT NULL, - blockTimestamp DATETIME, - payload JSONB NOT NULL, - rawArgs JSONB NOT NULL, - ingestedAt DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP +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, + "blockHash" TEXT NOT NULL, + "txHash" TEXT NOT NULL, + "logIndex" INTEGER NOT NULL, + "blockTimestamp" DATETIME, + "payload" JSONB NOT NULL, + "rawArgs" JSONB NOT NULL, + "removed" BOOLEAN NOT NULL DEFAULT false, + "ingestedAt" DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ); -- CreateTable @@ -82,7 +85,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_releaseId_projectionId_key" ON "v2_block_cursors"("chainId", "source", "releaseId", "projectionId"); @@ -149,4 +158,4 @@ CREATE TRIGGER "v2_block_cursor_rollbacks_audit_guard" OR NEW."toHeight" > NEW."fromHeight" ) BEGIN SELECT RAISE(ABORT, 'block cursor rollback must be authorized, audited, and move backward'); - END; + END; \ No newline at end of file diff --git a/prisma/schema.prisma b/prisma/schema.prisma index 4897db34..445ce2e9 100644 --- a/prisma/schema.prisma +++ b/prisma/schema.prisma @@ -285,8 +285,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("") // Monotonic CAS guard (PREWAVE-BE-038): the release/projection this cursor // belongs to, plus the deployment floor and checkpoint provenance. A writer @@ -355,11 +355,26 @@ model ChainEvent { // JSON-encoded RawChainLog.rawArgs rawArgs String + // 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? ingestedAt DateTime @default(now()) @@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 fa62b579..ef634764 100644 --- a/src/blockchain/blockchain-indexer.service.ts +++ b/src/blockchain/blockchain-indexer.service.ts @@ -6,10 +6,11 @@ import { EntityManager, QueryRunner, MoreThanOrEqual, + QueryFailedError, } from 'typeorm'; import { createHash } from 'crypto'; 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 { DeadLetterReplay } from './entities/dead-letter-replay.entity'; import { BlockchainStateService } from './state.service'; @@ -228,7 +229,7 @@ export class BlockchainIndexerService { @InjectRepository(ProcessedEvent) private processedEventRepo: Repository, @InjectRepository(TokenBalance) - private tokenBalanceRepo: Repository, + private tokenBalanceRepo: Repository, @InjectRepository(IndexerCheckpoint) private checkpointRepo: Repository, @InjectRepository(DeadLetterReplay) @@ -241,6 +242,41 @@ export class BlockchainIndexerService { private readinessService: ReadinessService, ) {} + /** + * 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, + }; + } + /** * Register a webhook provider configuration. Unknown providers fail closed * at verification time because no config is registered for them. @@ -531,11 +567,23 @@ export class BlockchainIndexerService { const fencingToken = this.requireLeadership(); const { txHash, logIndex, blockNumber, eventType, data } = event; - // 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. + // 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 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.debug(`Skipping duplicate event ${txHash}:${logIndex}`); 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 f4cb59ef..e789fe7a 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'; + /** * Provenance metadata bound to a chain-derived cursor advancement. @@ -664,6 +666,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( @@ -674,6 +679,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. *