From 62128b7899fd166154b66feb716aa630f4f4d540 Mon Sep 17 00:00:00 2001 From: DahNoble Date: Mon, 28 Sep 2026 18:11:48 +0100 Subject: [PATCH] Match Result Submission Idempotency for Duplicate Oracle Calls --- ...00000000-AddOracleSubmissionIdempotency.ts | 37 +++++ .../oracle/dto/webhook-match-result.dto.ts | 33 +++- .../entities/oracle-submission.entity.ts | 16 ++ backend/src/oracle/webhook.service.ts | 156 +++++++++++++++++- 4 files changed, 232 insertions(+), 10 deletions(-) create mode 100644 backend/src/migrations/1788100000000-AddOracleSubmissionIdempotency.ts diff --git a/backend/src/migrations/1788100000000-AddOracleSubmissionIdempotency.ts b/backend/src/migrations/1788100000000-AddOracleSubmissionIdempotency.ts new file mode 100644 index 000000000..5f27a715a --- /dev/null +++ b/backend/src/migrations/1788100000000-AddOracleSubmissionIdempotency.ts @@ -0,0 +1,37 @@ +import { MigrationInterface, QueryRunner } from 'typeorm'; + +/** + * Makes oracle match-result submissions idempotent: a duplicate webhook call + * (network retry, oracle re-delivery) resolves to the original submission + * instead of creating a second record and a second on-chain attempt. + */ +export class AddOracleSubmissionIdempotency1788100000000 implements MigrationInterface { + async up(queryRunner: QueryRunner): Promise { + await queryRunner.query(` + ALTER TABLE "oracle_submissions" + ADD COLUMN "job_id" varchar(64), + ADD COLUMN "idempotency_key" varchar(255) + `); + await queryRunner.query(` + CREATE UNIQUE INDEX "UQ_oracle_submissions_idempotency_key" + ON "oracle_submissions" ("idempotency_key") + WHERE "idempotency_key" IS NOT NULL + `); + await queryRunner.query(` + CREATE INDEX "IDX_oracle_submissions_job_id" + ON "oracle_submissions" ("job_id") + `); + } + + async down(queryRunner: QueryRunner): Promise { + await queryRunner.query(`DROP INDEX "IDX_oracle_submissions_job_id"`); + await queryRunner.query( + `DROP INDEX "UQ_oracle_submissions_idempotency_key"`, + ); + await queryRunner.query(` + ALTER TABLE "oracle_submissions" + DROP COLUMN "idempotency_key", + DROP COLUMN "job_id" + `); + } +} diff --git a/backend/src/oracle/dto/webhook-match-result.dto.ts b/backend/src/oracle/dto/webhook-match-result.dto.ts index 225340f35..7a4c02dc0 100644 --- a/backend/src/oracle/dto/webhook-match-result.dto.ts +++ b/backend/src/oracle/dto/webhook-match-result.dto.ts @@ -1,4 +1,12 @@ -import { IsString, IsInt, IsOptional, IsEnum, Min, Max } from 'class-validator'; +import { + IsString, + IsInt, + IsOptional, + IsEnum, + Min, + Max, + MaxLength, +} from 'class-validator'; import { ApiProperty, ApiPropertyOptional } from '@nestjs/swagger'; export enum WinningTeam { @@ -54,6 +62,17 @@ export class WebhookMatchResultDto { }) @IsOptional() metadata?: Record; + + @ApiPropertyOptional({ + description: + 'Client-supplied key identifying this result delivery. Repeating a call with the same key returns the original job instead of submitting again. When omitted, a key is derived from match_id, data_source, winning_team, confidence_score and timestamp.', + example: 'sports-data-123-final', + maxLength: 255, + }) + @IsOptional() + @IsString() + @MaxLength(255) + idempotency_key?: string; } export class WebhookResponseDto { @@ -74,4 +93,16 @@ export class WebhookResponseDto { example: 'Match result queued for submission', }) message: string; + + @ApiPropertyOptional({ + description: + 'True when this call duplicated an earlier one and no new submission was made', + example: false, + }) + duplicate?: boolean; + + @ApiPropertyOptional({ + description: 'ID of the oracle submission record backing this job', + }) + submission_id?: string; } diff --git a/backend/src/oracle/entities/oracle-submission.entity.ts b/backend/src/oracle/entities/oracle-submission.entity.ts index dbfc10185..b8ac97ab5 100644 --- a/backend/src/oracle/entities/oracle-submission.entity.ts +++ b/backend/src/oracle/entities/oracle-submission.entity.ts @@ -53,6 +53,22 @@ export class OracleSubmission { @ApiProperty() match_id: string; + /** Queue job ID returned to the oracle; replayed on duplicate calls. */ + @Column({ type: 'varchar', length: 64, nullable: true }) + @Index() + @ApiPropertyOptional() + job_id?: string; + + /** + * Deduplication key for the webhook call that created this submission — + * either supplied by the oracle or derived from the result payload. Unique, + * so a duplicate call resolves to this record instead of a new one. + */ + @Column({ type: 'varchar', length: 255, nullable: true }) + @Index({ unique: true, where: '"idempotency_key" IS NOT NULL' }) + @ApiPropertyOptional() + idempotency_key?: string; + @Column({ type: 'varchar', length: 100 }) @ApiProperty() team_a: string; diff --git a/backend/src/oracle/webhook.service.ts b/backend/src/oracle/webhook.service.ts index a319217a5..f49afecc9 100644 --- a/backend/src/oracle/webhook.service.ts +++ b/backend/src/oracle/webhook.service.ts @@ -4,6 +4,7 @@ import { NotFoundException, ConflictException, } from '@nestjs/common'; +import { createHash } from 'crypto'; import { InjectRepository } from '@nestjs/typeorm'; import { Repository } from 'typeorm'; import { CreatorEventMatch } from '../creator-events/entities/creator-event-match.entity'; @@ -26,6 +27,16 @@ import { ReviewResultResponse, } from './dto/anomaly-detection.dto'; +const POSTGRES_UNIQUE_VIOLATION = '23505'; + +function isUniqueViolation(error: unknown): boolean { + return ( + !!error && + typeof error === 'object' && + (error as { code?: string }).code === POSTGRES_UNIQUE_VIOLATION + ); +} + interface QueuedSubmission { id: string; matchId: string; @@ -61,6 +72,17 @@ export class WebhookService { async processMatchResult( dto: WebhookMatchResultDto, ): Promise { + // A duplicate delivery of a result we already accepted is answered with + // the original job, before the resolved-match check, so an oracle retrying + // after a successful submission gets the same answer instead of a 409. + const idempotencyKey = this.resolveIdempotencyKey(dto); + const existing = await this.submissionRepository.findOne({ + where: { idempotency_key: idempotencyKey }, + }); + if (existing) { + return this.replayDuplicate(existing, dto); + } + const jobId = this.generateJobId(); // Validate match exists @@ -103,13 +125,30 @@ export class WebhookService { nextRetryAt: new Date(), }; - this.submissionQueue.set(jobId, submission); + // Save to database for history tracking. The unique idempotency key makes + // this the serialization point for concurrent duplicate calls: only one + // insert wins, the others replay the winner. + let savedSubmission: OracleSubmission | null; + try { + savedSubmission = await this.saveSubmissionToDatabase( + submission, + match, + idempotencyKey, + ); + } catch (error) { + if (!isUniqueViolation(error)) { + throw error; + } + const winner = await this.submissionRepository.findOne({ + where: { idempotency_key: idempotencyKey }, + }); + if (!winner) { + throw error; + } + return this.replayDuplicate(winner, dto); + } - // Save to database for history tracking. - const savedSubmission = await this.saveSubmissionToDatabase( - submission, - match, - ); + this.submissionQueue.set(jobId, submission); // Screen for statistical anomalies before any on-chain use (#1364). A // flagged submission is recorded; when holding is enabled it is withheld @@ -129,6 +168,8 @@ export class WebhookService { status: 'held', message: 'Submission flagged as a statistical anomaly and held for manual review', + duplicate: false, + submission_id: savedSubmission.id, }; } } @@ -144,9 +185,91 @@ export class WebhookService { job_id: jobId, status: 'accepted', message: 'Match result queued for submission', + duplicate: false, + submission_id: savedSubmission?.id, + }; + } + + /** + * Key identifying one result delivery. An oracle-supplied key is scoped to + * its data source so two oracles cannot collide; otherwise the key is + * derived from the fields that define the result, so a byte-identical + * re-delivery maps to the same key. Both are hashed to a fixed length. + */ + private resolveIdempotencyKey(dto: WebhookMatchResultDto): string { + const hash = (parts: unknown[]): string => + createHash('sha256').update(JSON.stringify(parts)).digest('hex'); + + if (dto.idempotency_key) { + return `key:${hash([dto.data_source, dto.idempotency_key])}`; + } + + const parsed = new Date(dto.timestamp); + const timestamp = Number.isNaN(parsed.getTime()) + ? dto.timestamp + : parsed.toISOString(); + + return `payload:${hash([ + dto.match_id, + dto.data_source, + dto.winning_team, + dto.confidence_score, + timestamp, + ])}`; + } + + /** + * Answer a duplicate call with the original submission's state. Nothing is + * saved, queued, or submitted. Reusing an oracle-supplied key for a + * different result is a client error, not a duplicate. + */ + private replayDuplicate( + existing: OracleSubmission, + dto: WebhookMatchResultDto, + ): WebhookResponseDto { + const samePayload = + existing.match_id === dto.match_id && + existing.data_source === dto.data_source && + existing.winning_team === (dto.winning_team as string) && + existing.confidence_score === dto.confidence_score; + + if (!samePayload) { + throw new ConflictException( + `Idempotency key already used for a different result on match ${existing.match_id}`, + ); + } + + this.logger.log( + `Duplicate match result ignored: match_id=${dto.match_id}, submission_id=${existing.id}, job_id=${existing.job_id ?? 'unknown'}`, + ); + + return { + job_id: existing.job_id ?? existing.id, + status: this.describeSubmissionState(existing), + message: + 'Duplicate result delivery; returning the original submission without resubmitting', + duplicate: true, + submission_id: existing.id, }; } + private describeSubmissionState(submission: OracleSubmission): string { + if (submission.review_status === SubmissionReviewStatus.HELD) { + return 'held'; + } + if (submission.review_status === SubmissionReviewStatus.REJECTED) { + return 'rejected'; + } + switch (submission.status) { + case SubmissionStatus.SUBMITTED: + return 'submitted'; + case SubmissionStatus.FAILED: + return 'failed'; + default: + return 'accepted'; + } + } + private async submitToOracle(submission: QueuedSubmission): Promise { try { // In a real implementation, this would submit to the Soroban contract @@ -275,12 +398,20 @@ export class WebhookService { this.logger.log('Webhook retry processor started'); } + /** + * Persist the submission. Storage failures are logged and swallowed so the + * result is still attempted, except a unique violation on the idempotency + * key, which is rethrown so the caller can replay the original submission. + */ private async saveSubmissionToDatabase( submission: QueuedSubmission, match: CreatorEventMatch, + idempotencyKey: string, ): Promise { try { const dbSubmission = this.submissionRepository.create({ + job_id: submission.id, + idempotency_key: idempotencyKey, match_id: submission.matchId, team_a: match.team_a, team_b: match.team_b, @@ -297,6 +428,9 @@ export class WebhookService { this.logger.log(`Submission saved to database: job_id=${submission.id}`); return saved; } catch (error) { + if (isUniqueViolation(error)) { + throw error; + } this.logger.error( `Failed to save submission to database: ${error instanceof Error ? error.message : 'Unknown'}`, ); @@ -354,16 +488,20 @@ export class WebhookService { }; } + // Requeue the approved submission for on-chain use. The record is + // re-pointed at the new job so status updates and duplicate replays + // track it. + const requeuedJobId = this.generateJobId(); submission.review_status = SubmissionReviewStatus.APPROVED; + submission.job_id = requeuedJobId; await this.submissionRepository.save(submission); this.logger.log( `Held submission approved: submission_id=${submissionId}, reviewer=${dto.reviewer ?? 'unknown'}`, ); - // Requeue the approved submission for on-chain use. const requeued: QueuedSubmission = { - id: this.generateJobId(), + id: requeuedJobId, matchId: submission.match_id, winningTeam: submission.winning_team, confidenceScore: submission.confidence_score, @@ -391,7 +529,7 @@ export class WebhookService { ): Promise { try { const submission = await this.submissionRepository.findOne({ - where: { match_id: jobId.replace('job_', '').split('_')[0] }, + where: { job_id: jobId }, }); if (submission) {