diff --git a/bin/stacks/api-stack.ts b/bin/stacks/api-stack.ts index 43cca1b1..ebbfc5fc 100644 --- a/bin/stacks/api-stack.ts +++ b/bin/stacks/api-stack.ts @@ -389,33 +389,11 @@ export class APIStack extends cdk.Stack { chatbotSNSArn, }); - new CronStack(this, 'CronStack', { - RsDatabase: analyticsStack.dbName, - RsClusterIdentifier: analyticsStack.clusterId, - RedshiftCredSecretArn: analyticsStack.credSecretArn, - lambdaRole: lambdaRole, - chatbotSNSArn: chatbotSNSArn, - stage: stage, - }); - - /* filler addr table */ - new aws_dynamo.Table(this, `FillerAddrTable`, { - tableName: DYNAMO_TABLE_NAME.FILLER_ADDRESS, - partitionKey: { - name: 'pk', - type: aws_dynamo.AttributeType.STRING, - }, - deletionProtection: true, - pointInTimeRecovery: true, - contributorInsightsEnabled: true, - ...PROD_TABLE_CAPACITY.fillerAddress, - }); - /* posted-orders table: the hard-quote Lambda writes one row per confirmed RFQ-won post - (lib/handlers/hard-quote/posted-order-recorder.ts); the fade-rate cron will read it in - a later PR. Derived and rebuildable bookkeeping with a 48h TTL, so on-demand capacity - and no PITR. Both indexes sort by deadline: "completed" for the breaker means the - deadline has passed. */ + (lib/handlers/hard-quote/posted-order-recorder.ts); the fade-rate cron resolves outcomes + into it and reads the 24h window back (lib/cron/order-service-fades-source.ts). Derived + and rebuildable bookkeeping with a 48h TTL, so on-demand capacity and no PITR. Both + indexes sort by deadline: "completed" for the breaker means the deadline has passed. */ const postedOrdersTable = new aws_dynamo.Table(this, 'PostedOrdersTable', { tableName: DYNAMO_TABLE_NAME.POSTED_ORDERS, partitionKey: { @@ -441,6 +419,30 @@ export class APIStack extends cdk.Stack { // documents the dependency and keeps the write working if that policy is ever narrowed. postedOrdersTable.grantWriteData(hardQuoteLambda); + new CronStack(this, 'CronStack', { + RsDatabase: analyticsStack.dbName, + RsClusterIdentifier: analyticsStack.clusterId, + RedshiftCredSecretArn: analyticsStack.credSecretArn, + lambdaRole: lambdaRole, + chatbotSNSArn: chatbotSNSArn, + stage: stage, + postedOrdersTable, + orderServiceUrl: props.envVars.ORDER_SERVICE_URL, + }); + + /* filler addr table */ + new aws_dynamo.Table(this, `FillerAddrTable`, { + tableName: DYNAMO_TABLE_NAME.FILLER_ADDRESS, + partitionKey: { + name: 'pk', + type: aws_dynamo.AttributeType.STRING, + }, + deletionProtection: true, + pointInTimeRecovery: true, + contributorInsightsEnabled: true, + ...PROD_TABLE_CAPACITY.fillerAddress, + }); + /* Alarms */ const apiAlarm5xxSev2 = new aws_cloudwatch.Alarm(this, 'UniswapXParameterizationAPI-SEV2-5XXAlarm', { alarmName: 'UniswapXParameterizationAPI-SEV2-5XX', diff --git a/bin/stacks/cron-stack.ts b/bin/stacks/cron-stack.ts index eb53125d..b8f925e3 100644 --- a/bin/stacks/cron-stack.ts +++ b/bin/stacks/cron-stack.ts @@ -40,6 +40,11 @@ export interface CronStackProps extends cdk.NestedStackProps { stage: string; chatbotSNSArn?: string; envVars?: { [key: string]: string }; + // Inputs of the fade cron's shadow evaluation of the order-service fades source + // (lib/cron/fade-rate-shadow.ts): the PostedOrders table it resolves outcomes in, and the + // order service it asks for them. Both optional so the stack synthesizes without them. + postedOrdersTable?: aws_dynamo.ITable; + orderServiceUrl?: string; } export class CronStack extends cdk.NestedStack { @@ -48,7 +53,17 @@ export class CronStack extends cdk.NestedStack { constructor(scope: Construct, name: string, props: CronStackProps) { super(scope, name, props); - const { RsDatabase, RsClusterIdentifier, RedshiftCredSecretArn, lambdaRole, stage, envVars, chatbotSNSArn } = props; + const { + RsDatabase, + RsClusterIdentifier, + RedshiftCredSecretArn, + lambdaRole, + stage, + envVars, + chatbotSNSArn, + postedOrdersTable, + orderServiceUrl, + } = props; new s3.Bucket(this, 'FadeRateS3', { blockPublicAccess: s3.BlockPublicAccess.BLOCK_ALL, @@ -75,8 +90,13 @@ export class CronStack extends cdk.NestedStack { REDSHIFT_SECRET_ARN: RedshiftCredSecretArn, stage: stage, ...envVars, + ...(orderServiceUrl && { ORDER_SERVICE_URL: orderServiceUrl }), }, }); + // The shared Lambda role already carries AmazonDynamoDBFullAccess; the explicit grant + // documents the dependency (read the window + pending index, write outcomes) and keeps + // the shadow working if that policy is ever narrowed. + postedOrdersTable?.grantReadWriteData(this.fadeRateV2CronLambda); // Add Sev3 alarm for FadeRateV2Cron lambda errors const fadeRateV2CronErrors = this.fadeRateV2CronLambda.metricErrors({ diff --git a/lib/cron/fade-rate-shadow.ts b/lib/cron/fade-rate-shadow.ts new file mode 100644 index 00000000..d5f4e5d5 --- /dev/null +++ b/lib/cron/fade-rate-shadow.ts @@ -0,0 +1,260 @@ +import { MetricsLogger, Unit } from 'aws-embedded-metrics'; +import Logger from 'bunyan'; + +import { Metric } from '../entities/aws-metrics-logger'; +import { ToUpdateTimestampRow, V2FadesRowType } from '../repositories'; +import { UNBLOCKED_BLOCK_UNTIL_TIMESTAMP } from '../repositories/timestamp-repository'; +import { OrderServiceFadesSource, ResolutionSummary } from './order-service-fades-source'; + +// Wall-time ceiling on everything the shadow adds to a cron run (order-service batches, +// DynamoDB reads/writes, scoring). The fade cron's Lambda timeout is 240s and the Redshift +// path runs first; one minute keeps the shadow comfortably inside that even if the Redshift +// path is slow. The source stops starting order-service batches at this deadline, and the +// runner races the whole thing against it as a backstop. +export const SHADOW_TIME_BUDGET_MS = 60_000; + +// First moment PostedOrders had every post (PR #495 deployed 2026-09-04 21:29Z). Redshift's +// 24h window contains orders older than this that the new source can never have, so the +// comparison is restricted to orders posted at/after this floor. After the first day of shadow +// the floor is moot (everything in the window is newer). +export const POSTED_ORDERS_LIVE_SINCE = 1_788_557_340; + +/** + * What the cron hands the shadow after the real path has written. `score` is the production + * scoring code (getFillersFadeStats + calculateNewTimestamps) closed over the stored breaker + * state and address map the real run used, with metrics disabled — so shadow decisions differ + * from the real ones only through the rows. + */ +export type ShadowContext = { + redshiftRows: V2FadesRowType[]; + realUpdates: ToUpdateTimestampRow[]; + score: (rows: V2FadesRowType[]) => ToUpdateTimestampRow[]; + now: number; +}; + +export type ShadowDeps = { + source: OrderServiceFadesSource; + log: Logger; + metrics: MetricsLogger; + compareSince?: number; + budgetMs?: number; +}; + +export type PerFillerCounts = { oldTotal: number; oldFades: number; newTotal: number; newFades: number }; + +export type RowComparison = { + since: number; + oldRows: number; + newRows: number; + oldFades: number; + newFades: number; + // (fillerAddress, deadline) keys present on one side only, as a multiset difference. + onlyOld: string[]; + onlyNew: string[]; + // Keyed by lowercased filler address. + perFiller: Record; +}; + +export type Decision = { blocked: boolean; blockUntilTimestamp: number; consecutiveBlocks: number }; + +export type DecisionComparison = { + agree: number; + disagree: number; + onlyReal: string[]; + onlyShadow: string[]; + disagreements: { hash: string; real: Decision; shadow: Decision }[]; +}; + +export type ShadowReport = { + durationMs: number; + resolution?: ResolutionSummary; + rows: RowComparison; + // Shadow decisions vs. what the Redshift path actually wrote this run. + decisionsVsProduction: DecisionComparison; + // Shadow decisions vs. the Redshift rows re-scored with the same go-live floor. + decisionsVsRestricted: DecisionComparison; + wouldBlock: number; +}; + +const rowKey = (row: V2FadesRowType) => `${row.fillerAddress.toLowerCase()}:${row.deadline}`; + +/** + * Row-level parity, both sides restricted to orders posted at/after `since`. Rows are matched + * on (fillerAddress, deadline): the SQL's postTimestamp is the order service's createdat while + * ours is the post-confirmation time, so postTimestamp is not a join key. + */ +export function compareFadeRows(oldRows: V2FadesRowType[], newRows: V2FadesRowType[], since: number): RowComparison { + const old = oldRows.filter((row) => row.postTimestamp >= since); + const fresh = newRows.filter((row) => row.postTimestamp >= since); + + const counts = new Map(); + old.forEach((row) => counts.set(rowKey(row), (counts.get(rowKey(row)) ?? 0) + 1)); + fresh.forEach((row) => counts.set(rowKey(row), (counts.get(rowKey(row)) ?? 0) - 1)); + const onlyOld: string[] = []; + const onlyNew: string[] = []; + counts.forEach((delta, key) => { + for (let i = 0; i < delta; i++) onlyOld.push(key); + for (let i = 0; i < -delta; i++) onlyNew.push(key); + }); + + const perFiller: Record = {}; + const tally = (row: V2FadesRowType, side: 'old' | 'new') => { + const key = row.fillerAddress.toLowerCase(); + perFiller[key] ??= { oldTotal: 0, oldFades: 0, newTotal: 0, newFades: 0 }; + perFiller[key][`${side}Total`] += 1; + perFiller[key][`${side}Fades`] += row.faded; + }; + old.forEach((row) => tally(row, 'old')); + fresh.forEach((row) => tally(row, 'new')); + + return { + since, + oldRows: old.length, + newRows: fresh.length, + oldFades: old.reduce((sum, row) => sum + row.faded, 0), + newFades: fresh.reduce((sum, row) => sum + row.faded, 0), + onlyOld: onlyOld.sort(), + onlyNew: onlyNew.sort(), + perFiller, + }; +} + +const toDecision = (row: ToUpdateTimestampRow, now: number): Decision => { + const blockUntilTimestamp = row.blockUntilTimestamp ?? UNBLOCKED_BLOCK_UNTIL_TIMESTAMP; + return { blocked: blockUntilTimestamp > now, blockUntilTimestamp, consecutiveBlocks: row.consecutiveBlocks }; +}; + +/** Per-filler block decisions: agree when blocked?, blockUntil and consecutiveBlocks all match. */ +export function compareBlockDecisions( + real: ToUpdateTimestampRow[], + shadow: ToUpdateTimestampRow[], + now: number +): DecisionComparison { + const realByHash = new Map(real.map((row) => [row.hash, toDecision(row, now)])); + const shadowByHash = new Map(shadow.map((row) => [row.hash, toDecision(row, now)])); + const comparison: DecisionComparison = { agree: 0, disagree: 0, onlyReal: [], onlyShadow: [], disagreements: [] }; + + realByHash.forEach((realDecision, hash) => { + const shadowDecision = shadowByHash.get(hash); + if (!shadowDecision) { + comparison.onlyReal.push(hash); + return; + } + const same = + realDecision.blocked === shadowDecision.blocked && + realDecision.blockUntilTimestamp === shadowDecision.blockUntilTimestamp && + realDecision.consecutiveBlocks === shadowDecision.consecutiveBlocks; + if (same) { + comparison.agree += 1; + } else { + comparison.disagree += 1; + comparison.disagreements.push({ hash, real: realDecision, shadow: shadowDecision }); + } + }); + shadowByHash.forEach((_, hash) => { + if (!realByHash.has(hash)) comparison.onlyShadow.push(hash); + }); + comparison.onlyReal.sort(); + comparison.onlyShadow.sort(); + return comparison; +} + +/** + * Runs the order-service fades source next to the Redshift path and reports how the two + * compare. Writes nothing to FillerCBTimestampsV2 (it has no handle to it), notifies nobody, + * and never throws: any failure — order-service timeout, DynamoDB error, a bug in + * classification or comparison, the time budget — is logged and counted as + * CIRCUIT_BREAKER_SHADOW_FAILURE. Resolves to the report on success, undefined on failure. + */ +export async function runFadeRateShadow(ctx: ShadowContext, deps: ShadowDeps): Promise { + const { source, log, metrics } = deps; + const budgetMs = deps.budgetMs ?? SHADOW_TIME_BUDGET_MS; + const since = deps.compareSince ?? POSTED_ORDERS_LIVE_SINCE; + const start = Date.now(); + source.deadlineMs = start + budgetMs; + + try { + const report = await withTimeout(evaluate(ctx, source, since, start), budgetMs); + emit(metrics, report); + metrics.putMetric(Metric.CIRCUIT_BREAKER_SHADOW_SUCCESS, 1, Unit.Count); + log.info( + { + durationMs: report.durationMs, + resolution: report.resolution, + rows: { ...report.rows, perFiller: undefined }, + perFiller: report.rows.perFiller, + decisionsVsProduction: report.decisionsVsProduction, + decisionsVsRestricted: report.decisionsVsRestricted, + wouldBlock: report.wouldBlock, + }, + 'fade circuit breaker shadow report' + ); + return report; + } catch (e) { + metrics.putMetric(Metric.CIRCUIT_BREAKER_SHADOW_FAILURE, 1, Unit.Count); + metrics.putMetric(Metric.CIRCUIT_BREAKER_SHADOW_DURATION, Date.now() - start, Unit.Milliseconds); + log.error( + { error: e instanceof Error ? e.message : e, stack: e instanceof Error ? e.stack : undefined }, + 'fade circuit breaker shadow failed; real decisions unaffected' + ); + return undefined; + } +} + +async function evaluate( + ctx: ShadowContext, + source: OrderServiceFadesSource, + since: number, + start: number +): Promise { + const newRows = await source.getFades(); + const rows = compareFadeRows(ctx.redshiftRows, newRows, since); + const shadowUpdates = ctx.score(newRows); + const restrictedRealUpdates = ctx.score(ctx.redshiftRows.filter((row) => row.postTimestamp >= since)); + return { + durationMs: Date.now() - start, + resolution: source.lastResolution, + rows, + decisionsVsProduction: compareBlockDecisions(ctx.realUpdates, shadowUpdates, ctx.now), + decisionsVsRestricted: compareBlockDecisions(restrictedRealUpdates, shadowUpdates, ctx.now), + wouldBlock: shadowUpdates.filter((row) => (row.blockUntilTimestamp ?? UNBLOCKED_BLOCK_UNTIL_TIMESTAMP) > ctx.now) + .length, + }; +} + +function emit(metrics: MetricsLogger, report: ShadowReport): void { + const put = (metric: Metric, value: number, unit: Unit = Unit.Count) => metrics.putMetric(metric, value, unit); + put(Metric.CIRCUIT_BREAKER_SHADOW_DURATION, report.durationMs, Unit.Milliseconds); + if (report.resolution) { + put(Metric.CIRCUIT_BREAKER_SHADOW_PENDING_PAST_DEADLINE, report.resolution.pendingPastDeadline); + put(Metric.CIRCUIT_BREAKER_SHADOW_RESOLVED, report.resolution.resolved); + put(Metric.CIRCUIT_BREAKER_SHADOW_STILL_OPEN, report.resolution.stillOpen); + put(Metric.CIRCUIT_BREAKER_SHADOW_NOT_FOUND, report.resolution.notFound); + put(Metric.CIRCUIT_BREAKER_SHADOW_UNCLASSIFIABLE, report.resolution.unclassifiable); + } + put(Metric.CIRCUIT_BREAKER_SHADOW_ROWS_OLD, report.rows.oldRows); + put(Metric.CIRCUIT_BREAKER_SHADOW_ROWS_NEW, report.rows.newRows); + put(Metric.CIRCUIT_BREAKER_SHADOW_ROWS_ONLY_OLD, report.rows.onlyOld.length); + put(Metric.CIRCUIT_BREAKER_SHADOW_ROWS_ONLY_NEW, report.rows.onlyNew.length); + put(Metric.CIRCUIT_BREAKER_SHADOW_FADES_OLD, report.rows.oldFades); + put(Metric.CIRCUIT_BREAKER_SHADOW_FADES_NEW, report.rows.newFades); + put(Metric.CIRCUIT_BREAKER_SHADOW_DECISION_AGREE, report.decisionsVsProduction.agree); + put(Metric.CIRCUIT_BREAKER_SHADOW_DECISION_DISAGREE, report.decisionsVsProduction.disagree); + put(Metric.CIRCUIT_BREAKER_SHADOW_DECISION_AGREE_RESTRICTED, report.decisionsVsRestricted.agree); + put(Metric.CIRCUIT_BREAKER_SHADOW_DECISION_DISAGREE_RESTRICTED, report.decisionsVsRestricted.disagree); + put(Metric.CIRCUIT_BREAKER_SHADOW_WOULD_BLOCK, report.wouldBlock); +} + +// Promise.race keeps subscribing to the loser, so a source that fails after the budget fired +// is swallowed rather than surfacing as an unhandled rejection; the timer is always cleared. +async function withTimeout(promise: Promise, ms: number): Promise { + let timer: NodeJS.Timeout | undefined; + const timeout = new Promise((_, reject) => { + timer = setTimeout(() => reject(new Error(`fade shadow exceeded its ${ms}ms budget`)), ms); + }); + try { + return await Promise.race([promise, timeout]); + } finally { + if (timer) clearTimeout(timer); + } +} diff --git a/lib/cron/fade-rate-v2.ts b/lib/cron/fade-rate-v2.ts index 1d138ce0..f9af06ef 100644 --- a/lib/cron/fade-rate-v2.ts +++ b/lib/cron/fade-rate-v2.ts @@ -10,8 +10,9 @@ import { ethers } from 'ethers'; import { BETA_S3_KEY, PRODUCTION_S3_KEY, WEBHOOK_CONFIG_BUCKET } from '../constants'; import { AWSMetricsLogger, CircuitBreakerMetricDimension, Metric, metricContext } from '../entities'; import { checkDefined } from '../preconditions/preconditions'; -import { S3WebhookConfigurationProvider } from '../providers'; +import { S3WebhookConfigurationProvider, UniswapXServiceProvider } from '../providers'; import { + BaseTimestampRepository, ORDERS_PER_FILLER_LIMIT, SharedConfigs, TimestampRepoRow, @@ -19,9 +20,12 @@ import { V2FadesRepository, V2FadesRowType, } from '../repositories'; -import { DynamoFillerAddressRepository } from '../repositories/filler-address-repository'; +import { DynamoFillerAddressRepository, FillerAddressRepository } from '../repositories/filler-address-repository'; +import { DynamoPostedOrderRepository } from '../repositories/posted-order-repository'; import { TimestampRepository, UNBLOCKED_BLOCK_UNTIL_TIMESTAMP } from '../repositories/timestamp-repository'; import { STAGE } from '../util/stage'; +import { runFadeRateShadow, ShadowContext } from './fade-rate-shadow'; +import { FadesSource, OrderServiceFadesSource } from './order-service-fades-source'; // Re-exported for existing importers; the sentinel lives with the repository that owns the // stored value's parse/write semantics. @@ -133,6 +137,7 @@ const log = Logger.createLogger({ name: 'FadeRate', serializers: Logger.stdSerializers, }); +const defaultLog = log; /* set up aws clients */ const stage = process.env['stage']; @@ -153,7 +158,73 @@ export const handler: ScheduledHandler = metricScope((metrics) => async (_event: await main(metrics); }); +/** + * Everything the cron run touches, injectable so the run can be exercised end to end with + * fakes. Production wiring is in main(); the shared module-level clients above are what it + * passes. + */ +export type FadeRateCronDeps = { + fadesRepository: FadesSource & { createFadesView(): Promise }; + webhookProvider: Pick; + fillerAddressRepo: Pick; + timestampDB: BaseTimestampRepository; + // Optional shadow evaluation of the order-service fades source. Invoked strictly after the + // real path has written its decisions, with no handle to the timestamp table; it is expected + // not to throw, and the run additionally guards it so it cannot fail the cron. + shadow?: (ctx: ShadowContext) => Promise; + now?: () => number; + log?: Logger; +}; + async function main(metrics: MetricsLogger) { + const sharedConfig: SharedConfigs = { + Database: checkDefined(process.env.REDSHIFT_DATABASE), + ClusterIdentifier: checkDefined(process.env.REDSHIFT_CLUSTER_IDENTIFIER), + SecretArn: checkDefined(process.env.REDSHIFT_SECRET_ARN), + }; + await runFadeRateCron(metrics, { + fadesRepository: V2FadesRepository.create(sharedConfig), + webhookProvider, + fillerAddressRepo, + timestampDB, + shadow: buildOrderServiceShadow(metrics), + }); +} + +/** + * Production wiring of the shadow (see lib/cron/fade-rate-shadow.ts). Skipped, with a log + * line, when the order service URL is not configured for this stage. + */ +function buildOrderServiceShadow(metrics: MetricsLogger): FadeRateCronDeps['shadow'] { + const orderServiceUrl = process.env.ORDER_SERVICE_URL; + if (!orderServiceUrl) { + log.info('ORDER_SERVICE_URL is not set; skipping the order-service fade shadow'); + return undefined; + } + const shadowLog = log.child({ shadow: 'order-service-fades' }); + const source = new OrderServiceFadesSource({ + postedOrders: DynamoPostedOrderRepository.create( + // Not the hard-quote path's 200/300ms-bounded client: the cron is not in series with a + // quote, and the shadow has its own wall-time budget. + DynamoDBDocumentClient.from(new DynamoDBClient({}), { + marshallOptions: { convertEmptyValues: true, removeUndefinedValues: true }, + unmarshallOptions: { wrapNumbers: false }, + }) + ), + orderStatus: new UniswapXServiceProvider(shadowLog, orderServiceUrl), + fillerEndpoints: () => webhookProvider.fillerEndpoints(), + log: shadowLog, + // Parity flag (default on): never-filled cancelled / insufficient-funds / error orders + // count as fades, as the SQL's `fillTimestamp IS NULL` branch does today. Set the env var + // to 'false' to preview the candidate behavior change in the shadow metrics. + policy: { countNeverFilledTerminalAsFade: process.env.FADE_SHADOW_NEVER_FILLED_TERMINAL_AS_FADE !== 'false' }, + }); + return (ctx) => runFadeRateShadow(ctx, { source, log: shadowLog, metrics }); +} + +export async function runFadeRateCron(metrics: MetricsLogger, deps: FadeRateCronDeps): Promise { + const { fadesRepository, webhookProvider, fillerAddressRepo, timestampDB } = deps; + const log = deps.log ?? defaultLog; metrics.setNamespace('Uniswap'); metrics.setDimensions(CircuitBreakerMetricDimension); // The webhook config provider emits RFQ_CONFIG_CHANGED through the @@ -165,12 +236,6 @@ async function main(metrics: MetricsLogger) { // alongside the quote lambdas' dimensionless one. setGlobalMetric(new AWSMetricsLogger(metrics)); - const sharedConfig: SharedConfigs = { - Database: checkDefined(process.env.REDSHIFT_DATABASE), - ClusterIdentifier: checkDefined(process.env.REDSHIFT_CLUSTER_IDENTIFIER), - SecretArn: checkDefined(process.env.REDSHIFT_SECRET_ARN), - }; - const fadesRepository = V2FadesRepository.create(sharedConfig); await fadesRepository.createFadesView(); await webhookProvider.fetchEndpoints(); /* @@ -187,7 +252,7 @@ async function main(metrics: MetricsLogger) { const addressToFillerMap = await fillerAddressRepo.getAddressToFillerMap(fillerEndpoints); const fillerTimestamps = await timestampDB.getFillerTimestampsMap(fillerEndpoints); - const now = Math.floor(Date.now() / 1000); + const now = deps.now ? deps.now() : Math.floor(Date.now() / 1000); // compute each filler's Laplace-smoothed fade rates (post-block window + during-block cohort): // | hash | fadeRate | duringBlockRate | @@ -211,6 +276,30 @@ async function main(metrics: MetricsLogger) { } else { log.info('no timestamp to update'); } + + // Shadow evaluation of the order-service fades source: strictly after the real path has + // written its decisions. It scores its own rows with the same code against the same stored + // state (metrics and row logging off), writes nothing and notifies nobody. Guarded so that + // nothing it does can fail the cron — the shadow's own runner already never throws; this + // covers the wiring around it. + if (deps.shadow) { + try { + await deps.shadow({ + redshiftRows: result, + realUpdates: updatedTimestamps, + now, + score: (rows) => + calculateNewTimestamps( + fillerTimestamps, + getFillersFadeStats(rows, addressToFillerMap, fillerTimestamps, now), + now + ), + }); + } catch (e) { + metrics.putMetric(Metric.CIRCUIT_BREAKER_SHADOW_FAILURE, 1, Unit.Count); + log.error({ error: e instanceof Error ? e.message : e }, 'fade shadow wiring threw; real decisions unaffected'); + } + } } } diff --git a/lib/cron/order-service-fades-source.ts b/lib/cron/order-service-fades-source.ts new file mode 100644 index 00000000..3d9cc46f --- /dev/null +++ b/lib/cron/order-service-fades-source.ts @@ -0,0 +1,394 @@ +import { OrderType, PERMISSIONED_TOKENS } from '@uniswap/uniswapx-sdk'; +import Logger from 'bunyan'; +import { ethers } from 'ethers'; + +import { OrderServiceOrderStatus, OrderStatusProvider } from '../providers/order'; +import { ORDER_SERVICE_MAX_ORDER_HASHES } from '../providers/order/uniswapxService'; +import { ORDERS_PER_FILLER_LIMIT, V2FadesRowType } from '../repositories/fades-repository'; +import { + PostedOrderOutcome, + PostedOrderRecord, + PostedOrderRepository, + PostedOrderResolution, +} from '../repositories/posted-order-repository'; + +/** + * What the fade cron consumes: per-order rows in the shape of V2_FADE_RATE_SQL's result. + * V2FadesRepository (Redshift) satisfies this structurally; OrderServiceFadesSource is the + * GPA-owned implementation. Everything downstream (getFillersFadeStats, calculateNewTimestamps) + * is shared, so the two sources can only differ in the rows they produce. + */ +export interface FadesSource { + getFades(): Promise; +} + +// Rolling window on order completion time, mirroring the view's +// `deadline >= GETDATE() - INTERVAL '24 HOURS'`. +export const FADE_WINDOW_SECS = 24 * 60 * 60; +// Testnets the view drops: goerli, polygon goerli, optimism goerli, arbitrum goerli. Kept as +// the same literal list as V2_CREATE_VIEW_SQL so the two sources cannot drift apart. +export const EXCLUDED_TESTNET_CHAIN_IDS: readonly number[] = [5, 8001, 420, 421613]; +// Pending orders resolved per cron run. At ORDER_SERVICE_MAX_ORDER_HASHES per request this is +// 20 requests; anything beyond it waits for the next run (the pending index is oldest-first, +// so a backlog drains in deadline order). +export const MAX_PENDING_RESOLUTIONS_PER_RUN = 1000; +// Consecutive failed status batches after which the run stops calling the order service: the +// remaining batches would only burn the time budget against a service that is down. +export const MAX_CONSECUTIVE_BATCH_FAILURES = 3; + +/** Order-service `orderStatus` values the classifier understands. */ +export const ORDER_STATUS = { + OPEN: 'open', + FILLED: 'filled', + EXPIRED: 'expired', + CANCELLED: 'cancelled', + INSUFFICIENT_FUNDS: 'insufficient-funds', + ERROR: 'error', +} as const; + +// Lowercased permissioned-token addresses, compared chain-agnostically exactly as the SQL's +// `LOWER(tokenIn) NOT IN (...)` does. +const PERMISSIONED_TOKEN_ADDRESSES = new Set(PERMISSIONED_TOKENS.map((token) => token.address.toLowerCase())); + +export function isPermissionedToken(address: string): boolean { + return PERMISSIONED_TOKEN_ADDRESSES.has(address.toLowerCase()); +} + +export type Classification = + // Terminal status with a recordable outcome. + | { kind: 'resolved'; resolution: PostedOrderResolution } + // The order service still says `open` even though the deadline has passed: its status + // poller has not caught up yet. Leave the row pending and ask again next run. + | { kind: 'still-open' } + // A status we cannot score (unknown status string, a fill without timing, an unknown order + // type). Left pending so it is visible as a count; it expires with the row's TTL. + | { kind: 'unclassifiable'; reason: string }; + +/** + * Maps an order-service status onto the breaker's fade rule for one posted order. Mirrors the + * `faded` CASE in V2_FADE_RATE_SQL: + * + * status | Dutch_V2 | Dutch_V3 + * --------------------|-----------------------------------------|---------------------------------- + * filled | faded iff fillTimestamp > decayStartTime| faded iff fillBlock > decayStartBlock + * expired | faded | faded + * cancelled / insufficient-funds / error | recorded, no verdict (scored by policy flag; the SQL's + * | `fillTimestamp IS NULL` branch counts these as fades today) + * open | not terminal: stays pending | stays pending + * + * A fill AT the decay-start block/time is not a fade: the exclusive filler still paid the + * undecayed price (the SQL's `fillTimeBlocks > 0` / `decayStartTime < fillTimestamp`). + */ +export function classifyOutcome( + record: PostedOrderRecord, + status: OrderServiceOrderStatus, + resolvedAt: number +): Classification { + const base = { orderStatus: status.orderStatus, resolvedAt }; + switch (status.orderStatus) { + case ORDER_STATUS.OPEN: + return { kind: 'still-open' }; + case ORDER_STATUS.EXPIRED: + return { kind: 'resolved', resolution: { ...base, outcome: PostedOrderOutcome.EXPIRED, faded: 1 } }; + case ORDER_STATUS.CANCELLED: + return { kind: 'resolved', resolution: { ...base, outcome: PostedOrderOutcome.CANCELLED } }; + case ORDER_STATUS.INSUFFICIENT_FUNDS: + return { kind: 'resolved', resolution: { ...base, outcome: PostedOrderOutcome.INSUFFICIENT_FUNDS } }; + case ORDER_STATUS.ERROR: + return { kind: 'resolved', resolution: { ...base, outcome: PostedOrderOutcome.ERROR } }; + case ORDER_STATUS.FILLED: { + const timing = { fillBlock: status.fillBlock, fillTimestamp: status.fillTimestamp }; + if (record.orderType === OrderType.Dutch_V3) { + if (status.fillBlock === undefined || record.decayStartBlock === undefined) { + return { kind: 'unclassifiable', reason: 'Dutch_V3 fill without fillBlock/decayStartBlock' }; + } + const faded = status.fillBlock > record.decayStartBlock ? 1 : 0; + return { kind: 'resolved', resolution: { ...base, ...timing, outcome: PostedOrderOutcome.FILLED, faded } }; + } + if (record.orderType === OrderType.Dutch_V2) { + if (status.fillTimestamp === undefined || record.decayStartTime === undefined) { + return { kind: 'unclassifiable', reason: 'Dutch_V2 fill without fillTimestamp/decayStartTime' }; + } + const faded = status.fillTimestamp > record.decayStartTime ? 1 : 0; + return { kind: 'resolved', resolution: { ...base, ...timing, outcome: PostedOrderOutcome.FILLED, faded } }; + } + return { kind: 'unclassifiable', reason: `unknown order type ${record.orderType}` }; + } + default: + return { kind: 'unclassifiable', reason: `unknown order status ${status.orderStatus}` }; + } +} + +export type FadeRowPolicy = { + // PARITY FLAG. The SQL scores every never-filled order as a fade (`fillTimestamp IS NULL`), + // which includes cancelled, insufficient-funds and error orders alongside expiries. `true` + // reproduces that; `false` drops those orders from the rows entirely (neither fade nor + // clean fill), which is the candidate behavior change to decide on after the shadow. + countNeverFilledTerminalAsFade: boolean; +}; + +export const DEFAULT_FADE_ROW_POLICY: FadeRowPolicy = { countNeverFilledTerminalAsFade: true }; + +/** + * The 0/1 the row builder scores an order as, or undefined when the order contributes no row + * (still pending, or a never-filled non-expiry under the policy that excludes them). + */ +export function fadedForScoring(record: PostedOrderRecord, policy: FadeRowPolicy): number | undefined { + switch (record.outcome) { + case PostedOrderOutcome.PENDING: + return undefined; + case PostedOrderOutcome.FILLED: + case PostedOrderOutcome.EXPIRED: + return record.faded; + case PostedOrderOutcome.CANCELLED: + case PostedOrderOutcome.INSUFFICIENT_FUNDS: + case PostedOrderOutcome.ERROR: + return policy.countNeverFilledTerminalAsFade ? 1 : undefined; + default: + return undefined; + } +} + +/** + * Rebuilds V2_FADE_RATE_SQL's rows from PostedOrders records, in the SQL's order of operations: + * + * 1. completed orders only (deadline < now) — in-flight orders never consume window slots; + * 2. the ORDERS_PER_FILLER_LIMIT most recently posted orders per filler ADDRESS (the view + * partitions by the exclusive filler address, before any other filter — so a permissioned + * token order or a not-yet-resolved order still occupies a slot); + * 3. then the 24h completion window, testnets, the zero filler, missing quoteId, the + * permissioned-token lists (both legs), and orders with no recorded outcome (the SQL's + * LEFT JOIN drops rows that have not landed in archivedorders yet). + * + * Output rows are exactly V2FadesRowType, ordered by filler address then deadline desc like + * the SQL. `postTimestamp` is GPA's post-confirmation time; the SQL's `createdat` is the order + * service's, which can trail or lead by a second or two — callers match rows on + * (fillerAddress, deadline), not on postTimestamp. + */ +export function buildFadeRows( + records: PostedOrderRecord[], + args: { now: number; policy?: FadeRowPolicy } +): V2FadesRowType[] { + const { now } = args; + const policy = args.policy ?? DEFAULT_FADE_ROW_POLICY; + const windowStart = now - FADE_WINDOW_SECS; + + const byAddress = new Map(); + records + .filter((record) => record.deadline < now) + .forEach((record) => { + const key = record.fillerAddress.toLowerCase(); + byAddress.set(key, [...(byAddress.get(key) ?? []), record]); + }); + + const rows: V2FadesRowType[] = []; + byAddress.forEach((orders) => { + orders + .sort(byMostRecentlyPosted) + .slice(0, ORDERS_PER_FILLER_LIMIT) + .forEach((record) => { + if (record.deadline < windowStart) return; + if (EXCLUDED_TESTNET_CHAIN_IDS.includes(record.chainId)) return; + if (record.fillerAddress === ethers.constants.AddressZero) return; + if (!record.quoteId) return; + if (isPermissionedToken(record.tokenIn) || isPermissionedToken(record.tokenOut)) return; + const faded = fadedForScoring(record, policy); + if (faded === undefined) return; + rows.push({ + fillerAddress: record.fillerAddress, + faded, + postTimestamp: record.postedAt, + deadline: record.deadline, + }); + }); + }); + + return rows.sort( + (a, b) => a.fillerAddress.toLowerCase().localeCompare(b.fillerAddress.toLowerCase()) || b.deadline - a.deadline + ); +} + +// createdat DESC in the view; ties broken deterministically so a re-run yields the same slots. +function byMostRecentlyPosted(a: PostedOrderRecord, b: PostedOrderRecord): number { + return b.postedAt - a.postedAt || b.deadline - a.deadline || a.orderHash.localeCompare(b.orderHash); +} + +export type ResolutionSummary = { + // Pending orders past their deadline at the start of the run (capped at + // MAX_PENDING_RESOLUTIONS_PER_RUN, so a value at the cap means a backlog). + pendingPastDeadline: number; + batches: number; + failedBatches: number; + // Batches never sent because the time budget ran out or the service kept failing. + skippedBatches: number; + resolved: number; + // Past the deadline but the service still reports `open`: poller lag. Left pending. + stillOpen: number; + // The service returned nothing for the hash. Left pending; a persistent count here means + // GPA and the order service disagree about what was posted. + notFound: number; + unclassifiable: number; + failedWrites: number; + byOutcome: Partial>; +}; + +export interface OrderServiceFadesSourceDeps { + postedOrders: PostedOrderRepository; + orderStatus: OrderStatusProvider; + // Filler identities (webhook endpoints) to load rows for: the same list the cron scores. + fillerEndpoints: () => string[]; + log: Logger; + now?: () => number; + policy?: FadeRowPolicy; +} + +/** + * Fade rows from GPA-owned data. Each getFades() first resolves pending orders past their + * deadline against the order service (persisting outcomes so every order is fetched once), + * then rebuilds the SQL's per-filler rows from the 24h window of resolved orders. + */ +export class OrderServiceFadesSource implements FadesSource { + public lastResolution?: ResolutionSummary; + // Wallclock (ms) after which no further order-service batches are started; set by the + // caller before each getFades(). Resolution is best-effort per run: what is not resolved + // now is resolved next run. + public deadlineMs?: number; + private readonly now: () => number; + private readonly policy: FadeRowPolicy; + + constructor(private readonly deps: OrderServiceFadesSourceDeps) { + this.now = deps.now ?? (() => Math.floor(Date.now() / 1000)); + this.policy = deps.policy ?? DEFAULT_FADE_ROW_POLICY; + } + + async getFades(): Promise { + const now = this.now(); + this.lastResolution = await this.resolvePendingOutcomes(now); + const records = await this.loadCompletedWindow(now); + const rows = buildFadeRows(records, { now, policy: this.policy }); + this.deps.log.info( + { records: records.length, rows: rows.length, resolution: this.lastResolution }, + 'order-service fade rows' + ); + return rows; + } + + /** Every order for the scored fillers whose deadline fell in [now - 24h, now). */ + async loadCompletedWindow(now: number): Promise { + const endpoints = [...new Set(this.deps.fillerEndpoints())]; + const perFiller = await Promise.all( + endpoints.map((filler) => + this.deps.postedOrders.getFillerOrdersByDeadline(filler, now - FADE_WINDOW_SECS, now - 1) + ) + ); + // Rows are keyed by orderHash, so the union is already distinct per filler; dedupe anyway + // in case two endpoint strings ever alias the same filler. + const byHash = new Map(); + perFiller.flat().forEach((record) => byHash.set(record.orderHash, record)); + return [...byHash.values()]; + } + + /** + * Resolves pending orders past their deadline: batches of ORDER_SERVICE_MAX_ORDER_HASHES to + * the order service, one idempotent outcome write per terminal order. Stops early on the + * time budget or repeated service failures; leftovers are picked up next run. + */ + async resolvePendingOutcomes(now: number): Promise { + const { postedOrders, orderStatus, log } = this.deps; + const summary: ResolutionSummary = { + pendingPastDeadline: 0, + batches: 0, + failedBatches: 0, + skippedBatches: 0, + resolved: 0, + stillOpen: 0, + notFound: 0, + unclassifiable: 0, + failedWrites: 0, + byOutcome: {}, + }; + + const pending = await postedOrders.getPendingPastDeadline(now, MAX_PENDING_RESOLUTIONS_PER_RUN); + summary.pendingPastDeadline = pending.length; + + const batches = chunk(pending, ORDER_SERVICE_MAX_ORDER_HASHES); + let consecutiveFailures = 0; + for (let i = 0; i < batches.length; i++) { + if (this.deadlineMs !== undefined && Date.now() >= this.deadlineMs) { + summary.skippedBatches = batches.length - i; + log.warn({ skippedBatches: summary.skippedBatches }, 'order-service resolution stopped: time budget exhausted'); + break; + } + if (consecutiveFailures >= MAX_CONSECUTIVE_BATCH_FAILURES) { + summary.skippedBatches = batches.length - i; + log.warn({ skippedBatches: summary.skippedBatches }, 'order-service resolution stopped: service failing'); + break; + } + + const batch = batches[i]; + summary.batches += 1; + let statuses: OrderServiceOrderStatus[]; + try { + statuses = await orderStatus.getOrdersByHashes(batch.map((record) => record.orderHash)); + consecutiveFailures = 0; + } catch (e) { + summary.failedBatches += 1; + consecutiveFailures += 1; + log.warn( + { error: e instanceof Error ? e.message : e, batch: batch.length }, + 'order-service status batch failed' + ); + continue; + } + const byHash = new Map(statuses.map((status) => [status.orderHash.toLowerCase(), status])); + + const writes = batch.map(async (record) => { + const status = byHash.get(record.orderHash.toLowerCase()); + if (!status) { + summary.notFound += 1; + return; + } + const classification = classifyOutcome(record, status, now); + switch (classification.kind) { + case 'still-open': + summary.stillOpen += 1; + return; + case 'unclassifiable': + summary.unclassifiable += 1; + log.warn( + { orderHash: record.orderHash, status, reason: classification.reason }, + 'unclassifiable order outcome' + ); + return; + case 'resolved': { + const { resolution } = classification; + try { + await postedOrders.recordOutcome(record.orderHash, resolution); + summary.resolved += 1; + summary.byOutcome[resolution.outcome] = (summary.byOutcome[resolution.outcome] ?? 0) + 1; + } catch (e) { + summary.failedWrites += 1; + log.warn( + { orderHash: record.orderHash, error: e instanceof Error ? e.message : e }, + 'failed to record order outcome' + ); + } + return; + } + } + }); + await Promise.all(writes); + } + + log.info({ summary }, 'resolved pending order outcomes'); + return summary; + } +} + +function chunk(items: T[], size: number): T[][] { + const out: T[][] = []; + for (let i = 0; i < items.length; i += size) { + out.push(items.slice(i, i + size)); + } + return out; +} diff --git a/lib/entities/aws-metrics-logger.ts b/lib/entities/aws-metrics-logger.ts index 9e99d2e5..b7314b3b 100644 --- a/lib/entities/aws-metrics-logger.ts +++ b/lib/entities/aws-metrics-logger.ts @@ -127,6 +127,39 @@ export enum Metric { CIRCUIT_BREAKER_V2_ACTIVE_BLOCKS = 'CIRCUIT_BREAKER_V2_ACTIVE_BLOCKS', // Fillers with fade stats evaluated in a cron run (sample-health denominator) CIRCUIT_BREAKER_V2_FILLERS_EVALUATED = 'CIRCUIT_BREAKER_V2_FILLERS_EVALUATED', + + // Shadow evaluation of the order-service fades source (lib/cron/fade-rate-shadow.ts). Runs + // after the Redshift path each cron, writes nothing. Exactly one of SUCCESS / FAILURE fires + // per run; DURATION is the wall time it added to the cron (budgeted, see the runner). + CIRCUIT_BREAKER_SHADOW_SUCCESS = 'CIRCUIT_BREAKER_SHADOW_SUCCESS', + CIRCUIT_BREAKER_SHADOW_FAILURE = 'CIRCUIT_BREAKER_SHADOW_FAILURE', + CIRCUIT_BREAKER_SHADOW_DURATION = 'CIRCUIT_BREAKER_SHADOW_DURATION', + // Outcome resolution against the order service this run: pending orders past their deadline + // at the start, how many got a terminal outcome, how many the service still calls `open` + // (status-poller lag), how many it does not know at all, how many could not be scored. + CIRCUIT_BREAKER_SHADOW_PENDING_PAST_DEADLINE = 'CIRCUIT_BREAKER_SHADOW_PENDING_PAST_DEADLINE', + CIRCUIT_BREAKER_SHADOW_RESOLVED = 'CIRCUIT_BREAKER_SHADOW_RESOLVED', + CIRCUIT_BREAKER_SHADOW_STILL_OPEN = 'CIRCUIT_BREAKER_SHADOW_STILL_OPEN', + CIRCUIT_BREAKER_SHADOW_NOT_FOUND = 'CIRCUIT_BREAKER_SHADOW_NOT_FOUND', + CIRCUIT_BREAKER_SHADOW_UNCLASSIFIABLE = 'CIRCUIT_BREAKER_SHADOW_UNCLASSIFIABLE', + // Row-level comparison, both sides restricted to orders posted since PostedOrders went + // live. ONLY_OLD / ONLY_NEW are (fillerAddress, deadline) keys present on one side only. + CIRCUIT_BREAKER_SHADOW_ROWS_OLD = 'CIRCUIT_BREAKER_SHADOW_ROWS_OLD', + CIRCUIT_BREAKER_SHADOW_ROWS_NEW = 'CIRCUIT_BREAKER_SHADOW_ROWS_NEW', + CIRCUIT_BREAKER_SHADOW_ROWS_ONLY_OLD = 'CIRCUIT_BREAKER_SHADOW_ROWS_ONLY_OLD', + CIRCUIT_BREAKER_SHADOW_ROWS_ONLY_NEW = 'CIRCUIT_BREAKER_SHADOW_ROWS_ONLY_NEW', + CIRCUIT_BREAKER_SHADOW_FADES_OLD = 'CIRCUIT_BREAKER_SHADOW_FADES_OLD', + CIRCUIT_BREAKER_SHADOW_FADES_NEW = 'CIRCUIT_BREAKER_SHADOW_FADES_NEW', + // Block decisions the shadow would have made vs. the decisions the Redshift path actually + // wrote this run (per filler: blocked?, blockUntil, consecutiveBlocks). The RESTRICTED pair + // re-scores the Redshift rows with the same go-live floor, so it is the fair comparison + // while Redshift's 24h window still contains pre-go-live orders. + CIRCUIT_BREAKER_SHADOW_DECISION_AGREE = 'CIRCUIT_BREAKER_SHADOW_DECISION_AGREE', + CIRCUIT_BREAKER_SHADOW_DECISION_DISAGREE = 'CIRCUIT_BREAKER_SHADOW_DECISION_DISAGREE', + CIRCUIT_BREAKER_SHADOW_DECISION_AGREE_RESTRICTED = 'CIRCUIT_BREAKER_SHADOW_DECISION_AGREE_RESTRICTED', + CIRCUIT_BREAKER_SHADOW_DECISION_DISAGREE_RESTRICTED = 'CIRCUIT_BREAKER_SHADOW_DECISION_DISAGREE_RESTRICTED', + // Fillers the shadow would have benched after this run (the headline "what would change"). + CIRCUIT_BREAKER_SHADOW_WOULD_BLOCK = 'CIRCUIT_BREAKER_SHADOW_WOULD_BLOCK', } type MetricNeedingContext = diff --git a/lib/providers/order/index.ts b/lib/providers/order/index.ts index 4795cd01..897f486e 100644 --- a/lib/providers/order/index.ts +++ b/lib/providers/order/index.ts @@ -17,5 +17,26 @@ export interface OrderServiceProvider { postOrder(args: PostOrderArgs): Promise; } +/** + * The order service's GET /orders view of one order, reduced to what the fade breaker + * needs. `orderStatus` is one of open | filled | expired | cancelled | insufficient-funds | + * error; the fill fields are set only for fills (block number / unix seconds of the fill). + */ +export interface OrderServiceOrderStatus { + orderHash: string; + orderStatus: string; + fillBlock?: number; + fillTimestamp?: number; +} + +/** + * Read side of the order service used by the fade cron. One call per batch of at most + * ORDER_SERVICE_MAX_ORDER_HASHES hashes; orders the service does not know are simply absent + * from the result. Rejects on transport/HTTP failure so the caller can count and move on. + */ +export interface OrderStatusProvider { + getOrdersByHashes(orderHashes: string[]): Promise; +} + export * from './mock'; export * from './uniswapxService'; diff --git a/lib/providers/order/mock.ts b/lib/providers/order/mock.ts index beb81eb0..ad3e5d4b 100644 --- a/lib/providers/order/mock.ts +++ b/lib/providers/order/mock.ts @@ -1,4 +1,10 @@ -import { OrderServiceProvider, PostOrderArgs, UniswapXServiceResponse } from '.'; +import { + OrderServiceOrderStatus, + OrderServiceProvider, + OrderStatusProvider, + PostOrderArgs, + UniswapXServiceResponse, +} from '.'; import { ErrorResponse } from '../../handlers/base'; export class MockOrderServiceProvider implements OrderServiceProvider { @@ -13,3 +19,30 @@ export class MockOrderServiceProvider implements OrderServiceProvider { }; } } + +/** + * In-memory order-service read side for cron tests: seed statuses by hash, and every batch + * requested is recorded so tests can assert batching. Unknown hashes are omitted from the + * result, as the real service does. `failures` makes the next N calls reject. + */ +export class MockOrderStatusProvider implements OrderStatusProvider { + public readonly statuses = new Map(); + public readonly batches: string[][] = []; + public failures = 0; + + seed(...statuses: OrderServiceOrderStatus[]): void { + statuses.forEach((s) => this.statuses.set(s.orderHash.toLowerCase(), s)); + } + + async getOrdersByHashes(orderHashes: string[]): Promise { + this.batches.push([...orderHashes]); + if (this.failures > 0) { + this.failures -= 1; + throw new Error('mock order service unavailable'); + } + return orderHashes.flatMap((hash) => { + const status = this.statuses.get(hash.toLowerCase()); + return status ? [status] : []; + }); + } +} diff --git a/lib/providers/order/uniswapxService.ts b/lib/providers/order/uniswapxService.ts index a76fff4d..5a8f5dcb 100644 --- a/lib/providers/order/uniswapxService.ts +++ b/lib/providers/order/uniswapxService.ts @@ -1,8 +1,14 @@ -import axios, { AxiosError } from 'axios'; +import axios, { AxiosError, AxiosInstance } from 'axios'; import Logger from 'bunyan'; import { CosignedV2DutchOrder, CosignedV3DutchOrder, OrderType } from '@uniswap/uniswapx-sdk'; -import { OrderServiceProvider, PostOrderArgs, UniswapXServiceResponse } from '.'; +import { + OrderServiceOrderStatus, + OrderServiceProvider, + OrderStatusProvider, + PostOrderArgs, + UniswapXServiceResponse, +} from '.'; import { ErrorResponse } from '../../handlers/base'; import { ErrorCode } from '../../util/errors'; @@ -15,24 +21,64 @@ const ORDER_SERVICE_TIMEOUT_MS = 7000; const ORDER_RECONCILE_DELAY_MS = 500; const ORDER_RECONCILE_TIMEOUT_MS = 2000; +// GET /orders?orderHashes= accepts at most this many hashes per request (the service's Joi +// validator rejects longer lists with a 400). Callers batch; this client refuses to exceed it. +export const ORDER_SERVICE_MAX_ORDER_HASHES = 50; +// Per-batch ceiling for the status read. It runs in the fade cron (240s Lambda), not on a +// quote path, but a stalled order service must not eat the cron's whole budget either. +export const ORDER_STATUS_TIMEOUT_MS = 5000; + +// The subset of axios the status read goes through. Injected so tests can substitute a fake +// without mocking the axios module. +export type OrderServiceHttp = Pick; + const ORDER_TYPE_MAP = new Map([ [CosignedV2DutchOrder, OrderType.Dutch_V2], [CosignedV3DutchOrder, OrderType.Dutch_V3], ]); -// Shape of the order service's GET /dutch-auction/orders response; we only need -// to know whether any order matched during reconciliation. +// Shape of the order service's GET /dutch-auction/orders response. Reconciliation only needs +// to know whether any order matched; the fade cron reads status and fill timing per order. interface GetOrdersResponse { orders?: unknown[]; } -export class UniswapXServiceProvider implements OrderServiceProvider { +export class UniswapXServiceProvider implements OrderServiceProvider, OrderStatusProvider { private log: Logger; - constructor(_log: Logger, private uniswapxServiceUrl: string) { + constructor(_log: Logger, private uniswapxServiceUrl: string, private readonly http: OrderServiceHttp = axios) { this.log = _log.child({ quoter: 'UniswapXOrderService' }); } + /** + * Status + fill timing for a batch of order hashes (see OrderStatusProvider). The service + * has no pagination on this endpoint, so the batch cap is the only paging there is. + */ + async getOrdersByHashes(orderHashes: string[]): Promise { + if (orderHashes.length === 0) { + return []; + } + if (orderHashes.length > ORDER_SERVICE_MAX_ORDER_HASHES) { + throw new Error( + `getOrdersByHashes: ${orderHashes.length} hashes exceeds the order service cap of ${ORDER_SERVICE_MAX_ORDER_HASHES}` + ); + } + const response = await this.http.get(`${this.uniswapxServiceUrl}dutch-auction/orders`, { + params: { orderHashes: orderHashes.join(',') }, + timeout: ORDER_STATUS_TIMEOUT_MS, + }); + const orders = Array.isArray(response.data?.orders) ? response.data.orders : []; + const statuses = orders.flatMap((order) => { + const status = toOrderStatus(order); + if (!status) { + this.log.warn({ order }, 'Order service returned an order without hash/status; skipping'); + } + return status ? [status] : []; + }); + this.log.info({ requested: orderHashes.length, returned: statuses.length }, 'Fetched order statuses'); + return statuses; + } + async postOrder(args: PostOrderArgs): Promise { const { order, signature, quoteId, requestId } = args; const orderHash = order.hash(); @@ -128,3 +174,21 @@ export class UniswapXServiceProvider implements OrderServiceProvider { } } } + +// One GET /orders item reduced to the breaker's view. Anything without a hash and status is +// dropped (the caller treats a missing order as "not found" and retries next run). +function toOrderStatus(order: unknown): OrderServiceOrderStatus | undefined { + if (typeof order !== 'object' || order === null) { + return undefined; + } + const { orderHash, orderStatus, fillBlock, fillTimestamp } = order as Record; + if (typeof orderHash !== 'string' || typeof orderStatus !== 'string') { + return undefined; + } + return { + orderHash: orderHash.toLowerCase(), + orderStatus, + ...(typeof fillBlock === 'number' && { fillBlock }), + ...(typeof fillTimestamp === 'number' && { fillTimestamp }), + }; +} diff --git a/lib/repositories/posted-order-repository.ts b/lib/repositories/posted-order-repository.ts index 0e3481f8..36667fac 100644 --- a/lib/repositories/posted-order-repository.ts +++ b/lib/repositories/posted-order-repository.ts @@ -11,8 +11,35 @@ import { DYNAMO_TABLE_NAME, POSTED_ORDER_TTL_SECS, POSTED_ORDERS_INDEX } from '. */ export enum PostedOrderOutcome { PENDING = 'PENDING', + // Terminal outcomes, recorded by the fade cron from the order service's orderStatus once + // the deadline has passed (see lib/cron/order-service-fades-source.ts). FILLED and EXPIRED + // carry a definitive `faded`; the other three are "never filled" states whose fade + // treatment is a scoring-time policy, so the row stores the fact and not the verdict. + FILLED = 'FILLED', + EXPIRED = 'EXPIRED', + CANCELLED = 'CANCELLED', + INSUFFICIENT_FUNDS = 'INSUFFICIENT_FUNDS', + ERROR = 'ERROR', } +/** + * What the fade cron learned about a posted order from the order service. Written once per + * order (idempotently) when the deadline has passed and the service reports a terminal status. + */ +export type PostedOrderResolution = { + outcome: Exclude; + // Raw order-service `orderStatus` the outcome was derived from, kept for audit. + orderStatus: string; + // Fill timing as reported by the order service (block number / unix seconds). Set for fills. + fillBlock?: number; + fillTimestamp?: number; + // 1/0 for FILLED and EXPIRED, where the breaker's rule is definitive; undefined for the + // never-filled non-expiry outcomes, which the row builder scores by policy flag. + faded?: number; + // Unix seconds at which the cron recorded the outcome. + resolvedAt: number; +}; + /** * One row per confirmed RFQ-won order post. Everything the breaker needs about the "posted" * half of a fade, captured first-hand at post time instead of reconstructed from x-service @@ -46,6 +73,12 @@ export type PostedOrderRecord = { // Unix seconds at which the post was confirmed. postedAt: number; outcome: PostedOrderOutcome; + // Present once the outcome is no longer PENDING (see PostedOrderResolution). + orderStatus?: string; + fillBlock?: number; + fillTimestamp?: number; + faded?: number; + resolvedAt?: number; }; export interface PostedOrderRepository { @@ -55,6 +88,12 @@ export interface PostedOrderRepository { getPendingPastDeadline(now: number, limit?: number): Promise; /** A filler's orders whose deadline falls in [from, to], inclusive, oldest first. */ getFillerOrdersByDeadline(filler: string, from: number, to: number): Promise; + /** + * Records a terminal outcome and drops the row out of the pending index in the same write. + * Idempotent: re-recording the same resolution is a harmless overwrite. Rejects (rather than + * creating a phantom row) when the order is unknown, e.g. already expired by TTL. + */ + recordOutcome(orderHash: string, resolution: PostedOrderResolution): Promise; } // Constant partition key of the sparse pending index. Present only while outcome is @@ -129,6 +168,11 @@ export class DynamoPostedOrderRepository implements PostedOrderRepository { tokenOut: { type: 'string', required: true }, postedAt: { type: 'number', required: true }, outcome: { type: 'string', required: true }, + orderStatus: { type: 'string' }, + fillBlock: { type: 'number' }, + fillTimestamp: { type: 'number' }, + faded: { type: 'number' }, + resolvedAt: { type: 'number' }, pending: { type: 'string' }, ttl: { type: 'number', required: true }, }, @@ -175,6 +219,28 @@ export class DynamoPostedOrderRepository implements PostedOrderRepository { }); return ((Items ?? []) as PostedOrderItem[]).map(toRecord); } + + public async recordOutcome(orderHash: string, resolution: PostedOrderResolution): Promise { + const { outcome, orderStatus, fillBlock, fillTimestamp, faded, resolvedAt } = resolution; + await this.entity.update( + { + orderHash, + outcome, + orderStatus, + resolvedAt, + ...(fillBlock !== undefined && { fillBlock }), + ...(fillTimestamp !== undefined && { fillTimestamp }), + ...(faded !== undefined && { faded }), + // Leaving the sparse pending index is what makes the outcome "recorded" for the + // cron's next getPendingPastDeadline; it must happen in this same UpdateItem. + $remove: ['pending'], + }, + { + conditions: { attr: DynamoPostedOrderRepository.PARTITION_KEY, exists: true }, + execute: true, + } + ); + } } // Picks the record fields out of a stored item, dropping the index/TTL attributes and the @@ -196,6 +262,11 @@ function toRecord(item: PostedOrderItem): PostedOrderRecord { tokenOut: item.tokenOut, postedAt: item.postedAt, outcome: item.outcome, + ...(item.orderStatus !== undefined && { orderStatus: item.orderStatus }), + ...(item.fillBlock !== undefined && { fillBlock: item.fillBlock }), + ...(item.fillTimestamp !== undefined && { fillTimestamp: item.fillTimestamp }), + ...(item.faded !== undefined && { faded: item.faded }), + ...(item.resolvedAt !== undefined && { resolvedAt: item.resolvedAt }), }; } @@ -223,4 +294,21 @@ export class MockPostedOrderRepository implements PostedOrderRepository { .filter((r) => r.filler === filler && r.deadline >= from && r.deadline <= to) .sort((a, b) => a.deadline - b.deadline); } + + async recordOutcome(orderHash: string, resolution: PostedOrderResolution): Promise { + const existing = this.records.get(orderHash); + if (!existing) { + throw new Error(`The conditional request failed: no posted order ${orderHash}`); + } + const { outcome, orderStatus, fillBlock, fillTimestamp, faded, resolvedAt } = resolution; + this.records.set(orderHash, { + ...existing, + outcome, + orderStatus, + resolvedAt, + ...(fillBlock !== undefined && { fillBlock }), + ...(fillTimestamp !== undefined && { fillTimestamp }), + ...(faded !== undefined && { faded }), + }); + } } diff --git a/test/crons/fade-rate-cron-run.test.ts b/test/crons/fade-rate-cron-run.test.ts new file mode 100644 index 00000000..d44d4908 --- /dev/null +++ b/test/crons/fade-rate-cron-run.test.ts @@ -0,0 +1,245 @@ +import { createMetricsLogger, MetricsLogger, Unit } from 'aws-embedded-metrics'; +import Logger from 'bunyan'; + +import { ShadowContext } from '../../lib/cron/fade-rate-shadow'; +import { BASE_BLOCK_SECS, FadeRateCronDeps, runFadeRateCron } from '../../lib/cron/fade-rate-v2'; +import { Metric } from '../../lib/entities'; +import { + BaseTimestampRepository, + TimestampRepoRow, + ToUpdateTimestampRow, + V2FadesRowType, +} from '../../lib/repositories'; +import { MockFillerAddressRepository } from '../../lib/repositories/filler-address-repository'; +import { UNBLOCKED_BLOCK_UNTIL_TIMESTAMP } from '../../lib/repositories/timestamp-repository'; + +const log = Logger.createLogger({ name: 'test' }); +log.level(Logger.FATAL); + +const NOW = 1_800_000_000; +const FILLER_A = 'https://filler-a.example/rfq'; +const FILLER_B = 'https://filler-b.example/rfq'; +const ADDR_A = '0x00000000000000000000000000000000000000A1'; +const ADDR_B = '0x00000000000000000000000000000000000000B1'; + +const row = (fillerAddress: string, faded: 0 | 1, deadline: number): V2FadesRowType => ({ + fillerAddress, + faded, + postTimestamp: deadline - 60, + deadline, +}); + +// Redshift stand-in: the rows the real path scores, plus a call log proving the view is built. +class FakeFadesRepository { + public calls: string[] = []; + constructor(private readonly rows: V2FadesRowType[]) {} + async createFadesView(): Promise { + this.calls.push('createFadesView'); + } + async getFades(): Promise { + this.calls.push('getFades'); + return this.rows; + } +} + +class FakeWebhookProvider { + public fetched = 0; + constructor(private readonly endpoints: string[]) {} + async fetchEndpoints(): Promise { + this.fetched += 1; + } + fillerEndpoints(): string[] { + return this.endpoints; + } +} + +// FillerCBTimestampsV2 stand-in: stored state in, every batch write recorded verbatim. +class FakeTimestampRepository implements BaseTimestampRepository { + public writes: ToUpdateTimestampRow[][] = []; + constructor(private readonly stored: Map> = new Map()) {} + async updateTimestampsBatch(toUpdate: ToUpdateTimestampRow[]): Promise { + this.writes.push(toUpdate.map((r) => ({ ...r }))); + } + async getFillerTimestamps(hash: string): Promise { + const stored = this.stored.get(hash); + return { + hash, + lastExaminedTimestamp: stored?.lastExaminedTimestamp ?? 0, + blockUntilTimestamp: stored?.blockUntilTimestamp ?? UNBLOCKED_BLOCK_UNTIL_TIMESTAMP, + fadeWindowStart: stored?.fadeWindowStart ?? UNBLOCKED_BLOCK_UNTIL_TIMESTAMP, + consecutiveBlocks: stored?.consecutiveBlocks ?? 0, + consecutiveCleanRuns: stored?.consecutiveCleanRuns ?? 0, + }; + } + async getTimestampsBatch(hashes: string[]): Promise { + return Promise.all(hashes.filter((h) => this.stored.has(h)).map((h) => this.getFillerTimestamps(h))); + } + async getFillerTimestampsMap(hashes: string[]): Promise>> { + const rows = await this.getTimestampsBatch(hashes); + return new Map(rows.map(({ hash, ...rest }) => [hash, rest])); + } +} + +function recordingMetrics(): { metrics: MetricsLogger; calls: Record } { + const metrics = createMetricsLogger(); + const calls: Record = {}; + const original = metrics.putMetric.bind(metrics); + metrics.putMetric = (key: string, value: number, unit?: Unit | string) => { + (calls[key] ??= []).push(value); + return original(key, value, unit); + }; + return { metrics, calls }; +} + +// Filler A fades 5 of 5 (blocks); filler B fills 5 of 5 (stays clear). +const ROWS: V2FadesRowType[] = [ + ...Array.from({ length: 5 }, (_, i) => row(ADDR_A, 1, NOW - 100 - i)), + ...Array.from({ length: 5 }, (_, i) => row(ADDR_B, 0, NOW - 100 - i)), +]; + +async function fillerAddresses(): Promise { + const repo = new MockFillerAddressRepository(); + await repo.addNewAddressToFiller(ADDR_A, FILLER_A); + await repo.addNewAddressToFiller(ADDR_B, FILLER_B); + return repo; +} + +async function run(shadow?: FadeRateCronDeps['shadow']) { + const fades = new FakeFadesRepository(ROWS); + const webhooks = new FakeWebhookProvider([FILLER_A, FILLER_B]); + const timestamps = new FakeTimestampRepository(); + const { metrics, calls } = recordingMetrics(); + await runFadeRateCron(metrics, { + fadesRepository: fades, + webhookProvider: webhooks, + fillerAddressRepo: await fillerAddresses(), + timestampDB: timestamps, + shadow, + now: () => NOW, + log, + }); + return { fades, webhooks, timestamps, calls }; +} + +describe('runFadeRateCron', () => { + it('runs the Redshift path end to end: builds the view, scores, and writes the decisions', async () => { + const { fades, webhooks, timestamps, calls } = await run(); + + expect(fades.calls).toEqual(['createFadesView', 'getFades']); + expect(webhooks.fetched).toBe(1); + expect(timestamps.writes).toHaveLength(1); + expect(timestamps.writes[0]).toEqual([ + { + hash: FILLER_A, + lastExaminedTimestamp: NOW, + blockUntilTimestamp: NOW + BASE_BLOCK_SECS, + fadeWindowStart: NOW + BASE_BLOCK_SECS, + consecutiveBlocks: 1, + consecutiveCleanRuns: 0, + }, + { + hash: FILLER_B, + lastExaminedTimestamp: NOW, + blockUntilTimestamp: UNBLOCKED_BLOCK_UNTIL_TIMESTAMP, + fadeWindowStart: UNBLOCKED_BLOCK_UNTIL_TIMESTAMP, + consecutiveBlocks: 0, + consecutiveCleanRuns: 0, + }, + ]); + expect(calls[Metric.CIRCUIT_BREAKER_V2_NEW_BLOCKS]).toEqual([1]); + expect(calls[Metric.CIRCUIT_BREAKER_V2_ACTIVE_BLOCKS]).toEqual([1]); + expect(calls[Metric.CIRCUIT_BREAKER_V2_FILLERS_EVALUATED]).toEqual([2]); + }); + + it('the Redshift path writes and metrics are identical with a shadow, without one, and with a throwing shadow', async () => { + const baseline = await run(); + const withShadow = await run(async () => undefined); + const withThrowingShadow = await run(async () => { + throw new Error('order service exploded'); + }); + const withRejectingShadow = await run(() => Promise.reject(new TypeError('classification bug'))); + + for (const variant of [withShadow, withThrowingShadow, withRejectingShadow]) { + expect(variant.timestamps.writes).toEqual(baseline.timestamps.writes); + expect(variant.fades.calls).toEqual(baseline.fades.calls); + // Every real-path metric is the same; the throwing variants add exactly the failure marker. + const { [Metric.CIRCUIT_BREAKER_SHADOW_FAILURE]: failure, ...realPathMetrics } = variant.calls; + expect(realPathMetrics).toEqual(baseline.calls); + expect(failure ?? []).toEqual(variant === withShadow ? [] : [1]); + } + expect(baseline.calls[Metric.CIRCUIT_BREAKER_SHADOW_FAILURE]).toBeUndefined(); + }); + + it('a throwing shadow does not fail the cron', async () => { + await expect( + run(async () => { + throw new Error('boom'); + }) + ).resolves.toBeDefined(); + }); + + it('invokes the shadow only after the real decisions are written, with those decisions and the rows', async () => { + const order: string[] = []; + let received: ShadowContext | undefined; + const fades = new FakeFadesRepository(ROWS); + const timestamps = new FakeTimestampRepository(); + const realWrite = timestamps.updateTimestampsBatch.bind(timestamps); + timestamps.updateTimestampsBatch = async (rows) => { + order.push('write'); + return realWrite(rows); + }; + const { metrics } = recordingMetrics(); + + await runFadeRateCron(metrics, { + fadesRepository: fades, + webhookProvider: new FakeWebhookProvider([FILLER_A, FILLER_B]), + fillerAddressRepo: await fillerAddresses(), + timestampDB: timestamps, + shadow: async (ctx) => { + order.push('shadow'); + received = ctx; + }, + now: () => NOW, + log, + }); + + expect(order).toEqual(['write', 'shadow']); + expect(received?.redshiftRows).toEqual(ROWS); + expect(received?.realUpdates).toEqual(timestamps.writes[0]); + expect(received?.now).toBe(NOW); + // The scorer the shadow gets is the production scoring closed over the same stored state: + // re-scoring the Redshift rows reproduces the real decisions exactly, and scoring a clean + // set produces no block — while the write log still shows exactly one real write. + expect(received?.score(ROWS)).toEqual(timestamps.writes[0]); + const clean = received?.score(ROWS.map((r) => ({ ...r, faded: 0 }))); + expect(clean?.every((r) => r.blockUntilTimestamp === UNBLOCKED_BLOCK_UNTIL_TIMESTAMP)).toBe(true); + expect(timestamps.writes).toHaveLength(1); + }); + + it('the shadow context carries no handle to the timestamp repository', async () => { + let received: ShadowContext | undefined; + await run(async (ctx) => { + received = ctx; + }); + expect(Object.keys(received ?? {}).sort()).toEqual(['now', 'realUpdates', 'redshiftRows', 'score']); + }); + + it('skips the write and still runs the shadow when there is nothing to update', async () => { + let shadowRan = false; + const timestamps = new FakeTimestampRepository(); + const { metrics } = recordingMetrics(); + await runFadeRateCron(metrics, { + fadesRepository: new FakeFadesRepository([row('0x00000000000000000000000000000000000000C1', 1, NOW - 10)]), + webhookProvider: new FakeWebhookProvider([FILLER_A]), + fillerAddressRepo: await fillerAddresses(), + timestampDB: timestamps, + shadow: async () => { + shadowRan = true; + }, + now: () => NOW, + log, + }); + expect(timestamps.writes).toEqual([]); + expect(shadowRan).toBe(true); + }); +}); diff --git a/test/crons/fade-rate-shadow.test.ts b/test/crons/fade-rate-shadow.test.ts new file mode 100644 index 00000000..b7800c3e --- /dev/null +++ b/test/crons/fade-rate-shadow.test.ts @@ -0,0 +1,286 @@ +import { createMetricsLogger, MetricsLogger, Unit } from 'aws-embedded-metrics'; +import Logger from 'bunyan'; + +import { + compareBlockDecisions, + compareFadeRows, + POSTED_ORDERS_LIVE_SINCE, + runFadeRateShadow, + SHADOW_TIME_BUDGET_MS, + ShadowContext, +} from '../../lib/cron/fade-rate-shadow'; +import { OrderServiceFadesSource } from '../../lib/cron/order-service-fades-source'; +import { Metric } from '../../lib/entities'; +import { MockOrderStatusProvider } from '../../lib/providers/order'; +import { ToUpdateTimestampRow, V2FadesRowType } from '../../lib/repositories'; +import { MockPostedOrderRepository } from '../../lib/repositories/posted-order-repository'; +import { UNBLOCKED_BLOCK_UNTIL_TIMESTAMP } from '../../lib/repositories/timestamp-repository'; + +const log = Logger.createLogger({ name: 'test' }); +log.level(Logger.FATAL); + +const NOW = 1_800_000_000; +const SINCE = NOW - 10_000; +const ADDR_A = '0x00000000000000000000000000000000000000A1'; +const ADDR_B = '0x00000000000000000000000000000000000000B1'; + +const row = (fillerAddress: string, faded: 0 | 1, deadline: number, postTimestamp = deadline - 60): V2FadesRowType => ({ + fillerAddress, + faded, + postTimestamp, + deadline, +}); + +const update = (hash: string, overrides: Partial = {}): ToUpdateTimestampRow => ({ + hash, + lastExaminedTimestamp: NOW, + blockUntilTimestamp: UNBLOCKED_BLOCK_UNTIL_TIMESTAMP, + fadeWindowStart: UNBLOCKED_BLOCK_UNTIL_TIMESTAMP, + consecutiveBlocks: 0, + consecutiveCleanRuns: 0, + ...overrides, +}); + +// A real MetricsLogger (never flushed) with every putMetric recorded, so assertions run +// against the same object the cron hands the shadow. +function recordingMetrics(): { metrics: MetricsLogger; calls: Record } { + const metrics = createMetricsLogger(); + const calls: Record = {}; + const original = metrics.putMetric.bind(metrics); + metrics.putMetric = (key: string, value: number, unit?: Unit | string) => { + (calls[key] ??= []).push(value); + return original(key, value, unit); + }; + return { metrics, calls }; +} + +describe('fade-rate shadow', () => { + describe('compareFadeRows', () => { + it('restricts both sides to orders posted at/after the floor', () => { + const oldRows = [row(ADDR_A, 1, NOW - 100, SINCE - 1), row(ADDR_A, 0, NOW - 90, SINCE)]; + const newRows = [row(ADDR_A, 0, NOW - 90, SINCE + 5), row(ADDR_A, 1, NOW - 80, SINCE - 100)]; + + const comparison = compareFadeRows(oldRows, newRows, SINCE); + + expect(comparison).toMatchObject({ + since: SINCE, + oldRows: 1, + newRows: 1, + oldFades: 0, + newFades: 0, + onlyOld: [], + onlyNew: [], + }); + }); + + it('matches rows on (fillerAddress, deadline) as a multiset, ignoring postTimestamp and address case', () => { + const oldRows = [ + row(ADDR_A, 1, NOW - 100, NOW - 170), + row(ADDR_A, 1, NOW - 100, NOW - 170), // duplicate key on the old side + row(ADDR_A, 0, NOW - 50), + row(ADDR_B, 1, NOW - 40), + ]; + const newRows = [ + row(ADDR_A.toLowerCase(), 1, NOW - 100, NOW - 168), // postTimestamp drift is not a mismatch + row(ADDR_A, 0, NOW - 50), + row(ADDR_B, 0, NOW - 30), + ]; + + const comparison = compareFadeRows(oldRows, newRows, SINCE); + + expect(comparison.onlyOld).toEqual([ + `${ADDR_A.toLowerCase()}:${NOW - 100}`, + `${ADDR_B.toLowerCase()}:${NOW - 40}`, + ]); + expect(comparison.onlyNew).toEqual([`${ADDR_B.toLowerCase()}:${NOW - 30}`]); + expect(comparison).toMatchObject({ oldRows: 4, newRows: 3, oldFades: 3, newFades: 1 }); + }); + + it('tallies totals and fades per filler address on both sides', () => { + const comparison = compareFadeRows( + [row(ADDR_A, 1, NOW - 100), row(ADDR_A, 0, NOW - 90), row(ADDR_B, 1, NOW - 80)], + [row(ADDR_A, 1, NOW - 100), row(ADDR_A, 1, NOW - 90)], + SINCE + ); + + expect(comparison.perFiller).toEqual({ + [ADDR_A.toLowerCase()]: { oldTotal: 2, oldFades: 1, newTotal: 2, newFades: 2 }, + [ADDR_B.toLowerCase()]: { oldTotal: 1, oldFades: 1, newTotal: 0, newFades: 0 }, + }); + }); + }); + + describe('compareBlockDecisions', () => { + it('agrees when blocked?, blockUntil and consecutiveBlocks all match', () => { + const real = [ + update('a'), + update('b', { blockUntilTimestamp: NOW + 900, fadeWindowStart: NOW + 900, consecutiveBlocks: 1 }), + ]; + const shadow = [ + update('a', { consecutiveCleanRuns: 3 }), + update('b', { blockUntilTimestamp: NOW + 900, consecutiveBlocks: 1 }), + ]; + + expect(compareBlockDecisions(real, shadow, NOW)).toEqual({ + agree: 2, + disagree: 0, + onlyReal: [], + onlyShadow: [], + disagreements: [], + }); + }); + + it('reports each disagreement with both decisions, and fillers present on one side only', () => { + const real = [update('a'), update('b', { consecutiveBlocks: 2 }), update('onlyReal')]; + const shadow = [ + update('a', { blockUntilTimestamp: NOW + 900, consecutiveBlocks: 1 }), + update('b', { consecutiveBlocks: 1 }), + update('onlyShadow'), + ]; + + const comparison = compareBlockDecisions(real, shadow, NOW); + + expect(comparison).toMatchObject({ agree: 0, disagree: 2, onlyReal: ['onlyReal'], onlyShadow: ['onlyShadow'] }); + expect(comparison.disagreements).toEqual([ + { + hash: 'a', + real: { blocked: false, blockUntilTimestamp: 0, consecutiveBlocks: 0 }, + shadow: { blocked: true, blockUntilTimestamp: NOW + 900, consecutiveBlocks: 1 }, + }, + { + hash: 'b', + real: { blocked: false, blockUntilTimestamp: 0, consecutiveBlocks: 2 }, + shadow: { blocked: false, blockUntilTimestamp: 0, consecutiveBlocks: 1 }, + }, + ]); + }); + + it('treats a missing blockUntilTimestamp as unblocked', () => { + const real = [{ ...update('a'), blockUntilTimestamp: undefined }]; + expect(compareBlockDecisions(real, [update('a')], NOW).agree).toBe(1); + }); + }); + + describe('runFadeRateShadow', () => { + const emptySource = () => + new OrderServiceFadesSource({ + postedOrders: new MockPostedOrderRepository(), + orderStatus: new MockOrderStatusProvider(), + fillerEndpoints: () => [], + log, + now: () => NOW, + }); + + const ctx = (overrides: Partial = {}): ShadowContext => ({ + redshiftRows: [row(ADDR_A, 1, NOW - 100, SINCE + 1), row(ADDR_A, 1, NOW - 50, SINCE - 1)], + realUpdates: [update('fillerA', { blockUntilTimestamp: NOW + 900, consecutiveBlocks: 1 })], + // Scorer stand-in: one filler, blocked iff any fade in the rows it is given. + score: (rows) => + rows.length === 0 + ? [] + : [ + update( + 'fillerA', + rows.some((r) => r.faded) ? { blockUntilTimestamp: NOW + 900, consecutiveBlocks: 1 } : {} + ), + ], + now: NOW, + ...overrides, + }); + + it('emits the comparison metrics and a success marker, and returns the report', async () => { + const { metrics, calls } = recordingMetrics(); + // The new side has no rows (nothing in PostedOrders): the shadow scorer emits nothing, so + // production's block is "only real"; restricted to the floor the old side has one fade. + const report = await runFadeRateShadow(ctx(), { source: emptySource(), log, metrics, compareSince: SINCE }); + + expect(report).toBeDefined(); + expect(report?.rows).toMatchObject({ oldRows: 1, newRows: 0, oldFades: 1, newFades: 0, onlyNew: [] }); + expect(report?.rows.onlyOld).toEqual([`${ADDR_A.toLowerCase()}:${NOW - 100}`]); + expect(report?.decisionsVsProduction).toMatchObject({ agree: 0, disagree: 0, onlyReal: ['fillerA'] }); + expect(report?.decisionsVsRestricted).toMatchObject({ agree: 0, disagree: 0, onlyReal: ['fillerA'] }); + expect(report?.wouldBlock).toBe(0); + expect(report?.resolution).toMatchObject({ pendingPastDeadline: 0, resolved: 0 }); + + expect(calls[Metric.CIRCUIT_BREAKER_SHADOW_SUCCESS]).toEqual([1]); + expect(calls[Metric.CIRCUIT_BREAKER_SHADOW_FAILURE]).toBeUndefined(); + expect(calls[Metric.CIRCUIT_BREAKER_SHADOW_ROWS_OLD]).toEqual([1]); + expect(calls[Metric.CIRCUIT_BREAKER_SHADOW_ROWS_NEW]).toEqual([0]); + expect(calls[Metric.CIRCUIT_BREAKER_SHADOW_ROWS_ONLY_OLD]).toEqual([1]); + expect(calls[Metric.CIRCUIT_BREAKER_SHADOW_FADES_OLD]).toEqual([1]); + expect(calls[Metric.CIRCUIT_BREAKER_SHADOW_DECISION_AGREE]).toEqual([0]); + expect(calls[Metric.CIRCUIT_BREAKER_SHADOW_DECISION_DISAGREE]).toEqual([0]); + expect(calls[Metric.CIRCUIT_BREAKER_SHADOW_WOULD_BLOCK]).toEqual([0]); + expect(calls[Metric.CIRCUIT_BREAKER_SHADOW_PENDING_PAST_DEADLINE]).toEqual([0]); + expect(calls[Metric.CIRCUIT_BREAKER_SHADOW_DURATION]).toHaveLength(1); + }); + + it('scores the new rows with the injected production scorer and compares against both baselines', async () => { + const { metrics } = recordingMetrics(); + const source = emptySource(); + source.getFades = async () => [row(ADDR_A, 1, NOW - 100, SINCE + 1)]; + + const report = await runFadeRateShadow(ctx(), { source, log, metrics, compareSince: SINCE }); + + expect(report?.rows).toMatchObject({ oldRows: 1, newRows: 1, onlyOld: [], onlyNew: [] }); + expect(report?.decisionsVsProduction).toMatchObject({ agree: 1, disagree: 0 }); + expect(report?.decisionsVsRestricted).toMatchObject({ agree: 1, disagree: 0 }); + expect(report?.wouldBlock).toBe(1); + }); + + it('defaults the comparison floor to the PostedOrders go-live time and the budget to a minute', () => { + expect(POSTED_ORDERS_LIVE_SINCE).toBe(Math.floor(Date.parse('2026-09-04T21:29:00Z') / 1000)); + expect(SHADOW_TIME_BUDGET_MS).toBe(60_000); + }); + + it('hands the source a wallclock deadline derived from the budget', async () => { + const { metrics } = recordingMetrics(); + const source = emptySource(); + const before = Date.now(); + + await runFadeRateShadow(ctx(), { source, log, metrics, budgetMs: 1234 }); + + expect(source.deadlineMs).toBeGreaterThanOrEqual(before + 1234); + expect(source.deadlineMs).toBeLessThanOrEqual(Date.now() + 1234); + }); + + it('a throwing source is logged and counted as a failure, never thrown', async () => { + const { metrics, calls } = recordingMetrics(); + const source = emptySource(); + source.getFades = async () => { + throw new Error('DynamoDB unavailable'); + }; + + await expect(runFadeRateShadow(ctx(), { source, log, metrics })).resolves.toBeUndefined(); + + expect(calls[Metric.CIRCUIT_BREAKER_SHADOW_FAILURE]).toEqual([1]); + expect(calls[Metric.CIRCUIT_BREAKER_SHADOW_SUCCESS]).toBeUndefined(); + expect(calls[Metric.CIRCUIT_BREAKER_SHADOW_DURATION]).toHaveLength(1); + expect(calls[Metric.CIRCUIT_BREAKER_SHADOW_ROWS_NEW]).toBeUndefined(); + }); + + it('a throwing scorer (classification/comparison bug) is contained the same way', async () => { + const { metrics, calls } = recordingMetrics(); + const context = ctx({ + score: () => { + throw new TypeError('bug'); + }, + }); + + await expect(runFadeRateShadow(context, { source: emptySource(), log, metrics })).resolves.toBeUndefined(); + expect(calls[Metric.CIRCUIT_BREAKER_SHADOW_FAILURE]).toEqual([1]); + }); + + it('a source that hangs is cut off at the budget and counted as a failure', async () => { + const { metrics, calls } = recordingMetrics(); + const source = emptySource(); + source.getFades = () => new Promise(() => undefined); + + const started = Date.now(); + await expect(runFadeRateShadow(ctx(), { source, log, metrics, budgetMs: 30 })).resolves.toBeUndefined(); + + expect(Date.now() - started).toBeLessThan(1000); + expect(calls[Metric.CIRCUIT_BREAKER_SHADOW_FAILURE]).toEqual([1]); + }); + }); +}); diff --git a/test/crons/order-service-fades-source.test.ts b/test/crons/order-service-fades-source.test.ts new file mode 100644 index 00000000..ffeabc37 --- /dev/null +++ b/test/crons/order-service-fades-source.test.ts @@ -0,0 +1,620 @@ +import { + DescribeStatementCommand, + ExecuteStatementCommand, + GetStatementResultCommand, + RedshiftDataClient, +} from '@aws-sdk/client-redshift-data'; +import { OrderType, PERMISSIONED_TOKENS } from '@uniswap/uniswapx-sdk'; +import Logger from 'bunyan'; + +import { + buildFadeRows, + Classification, + classifyOutcome, + EXCLUDED_TESTNET_CHAIN_IDS, + FADE_WINDOW_SECS, + fadedForScoring, + MAX_CONSECUTIVE_BATCH_FAILURES, + MAX_PENDING_RESOLUTIONS_PER_RUN, + ORDER_STATUS, + OrderServiceFadesSource, +} from '../../lib/cron/order-service-fades-source'; +import { MockOrderStatusProvider, OrderServiceOrderStatus } from '../../lib/providers/order'; +import { ORDER_SERVICE_MAX_ORDER_HASHES } from '../../lib/providers/order/uniswapxService'; +import { ORDERS_PER_FILLER_LIMIT, V2FadesRepository, V2FadesRowType } from '../../lib/repositories/fades-repository'; +import { + MockPostedOrderRepository, + PostedOrderOutcome, + PostedOrderRecord, +} from '../../lib/repositories/posted-order-repository'; + +const log = Logger.createLogger({ name: 'test' }); +log.level(Logger.FATAL); + +const NOW = 1_800_000_000; +const FILLER_A = 'https://filler-a.example/rfq'; +const FILLER_B = 'https://filler-b.example/rfq'; +const ADDR_A = '0x00000000000000000000000000000000000000A1'; +const ADDR_A2 = '0x00000000000000000000000000000000000000A2'; +const ADDR_B = '0x00000000000000000000000000000000000000B1'; +const USDC = '0xA0b86991c6218b36c1d19D4a2e9Eb0cE3606eB48'; +const WETH = '0xC02aaA39b223FE8D0A0e5C4F27eAD9083C756Cc2'; +const DECAY_START_TIME = NOW - 3_000; +const DECAY_START_BLOCK = 20_000_000; + +let hashCounter = 0; +const nextHash = () => `0x${(++hashCounter).toString(16).padStart(64, '0')}`; + +const v2 = (overrides: Partial = {}): PostedOrderRecord => ({ + orderHash: nextHash(), + quoteId: 'quote', + requestId: 'request', + chainId: 1, + orderType: OrderType.Dutch_V2, + fillerAddress: ADDR_A, + filler: FILLER_A, + fillerName: 'filler-a', + decayStartTime: DECAY_START_TIME, + deadline: NOW - 2_000, + tokenIn: USDC, + tokenOut: WETH, + postedAt: NOW - 3_100, + outcome: PostedOrderOutcome.PENDING, + ...overrides, +}); +const v3 = (overrides: Partial = {}): PostedOrderRecord => + v2({ orderType: OrderType.Dutch_V3, decayStartTime: undefined, decayStartBlock: DECAY_START_BLOCK, ...overrides }); + +const status = (orderHash: string, orderStatus: string, timing: Partial = {}) => ({ + orderHash, + orderStatus, + ...timing, +}); + +// A resolved record: what recordOutcome leaves behind for a given status. +const resolved = (record: PostedOrderRecord, s: OrderServiceOrderStatus): PostedOrderRecord => { + const classification = classifyOutcome(record, s, NOW); + if (classification.kind !== 'resolved') throw new Error(`fixture is not resolvable: ${classification.kind}`); + const { outcome, orderStatus, fillBlock, fillTimestamp, faded, resolvedAt } = classification.resolution; + return { + ...record, + outcome, + orderStatus, + resolvedAt, + ...(fillBlock !== undefined && { fillBlock }), + ...(fillTimestamp !== undefined && { fillTimestamp }), + ...(faded !== undefined && { faded }), + }; +}; + +describe('OrderServiceFadesSource', () => { + describe('classifyOutcome (status x order type x timing)', () => { + type Case = { + name: string; + record: PostedOrderRecord; + status: OrderServiceOrderStatus; + expected: Classification['kind']; + outcome?: PostedOrderOutcome; + faded?: number; + }; + const cases: Case[] = [ + // Dutch_V2 fills decay by wallclock: only a fill strictly after decayStartTime fades. + { + name: 'V2 filled before decay start -> clean', + record: v2(), + status: status('h', ORDER_STATUS.FILLED, { fillTimestamp: DECAY_START_TIME - 1, fillBlock: 1 }), + expected: 'resolved', + outcome: PostedOrderOutcome.FILLED, + faded: 0, + }, + { + name: 'V2 filled AT decay start -> clean (undecayed price)', + record: v2(), + status: status('h', ORDER_STATUS.FILLED, { fillTimestamp: DECAY_START_TIME, fillBlock: 1 }), + expected: 'resolved', + outcome: PostedOrderOutcome.FILLED, + faded: 0, + }, + { + name: 'V2 filled after decay start -> fade', + record: v2(), + status: status('h', ORDER_STATUS.FILLED, { fillTimestamp: DECAY_START_TIME + 1, fillBlock: 1 }), + expected: 'resolved', + outcome: PostedOrderOutcome.FILLED, + faded: 1, + }, + { + name: 'V2 filled without fillTimestamp -> unclassifiable', + record: v2(), + status: status('h', ORDER_STATUS.FILLED, { fillBlock: 1 }), + expected: 'unclassifiable', + }, + // Dutch_V3 fills decay by block: fillTimeBlocks = fillBlock - decayStartBlock must be > 0. + { + name: 'V3 filled before decay start block -> clean', + record: v3(), + status: status('h', ORDER_STATUS.FILLED, { fillBlock: DECAY_START_BLOCK - 1, fillTimestamp: NOW }), + expected: 'resolved', + outcome: PostedOrderOutcome.FILLED, + faded: 0, + }, + { + name: 'V3 filled AT decay start block -> clean (fillTimeBlocks = 0 is not a fade)', + record: v3(), + status: status('h', ORDER_STATUS.FILLED, { fillBlock: DECAY_START_BLOCK, fillTimestamp: NOW }), + expected: 'resolved', + outcome: PostedOrderOutcome.FILLED, + faded: 0, + }, + { + name: 'V3 filled one block after decay start -> fade', + record: v3(), + status: status('h', ORDER_STATUS.FILLED, { fillBlock: DECAY_START_BLOCK + 1, fillTimestamp: NOW }), + expected: 'resolved', + outcome: PostedOrderOutcome.FILLED, + faded: 1, + }, + { + name: 'V3 fill timing ignores fillTimestamp (late timestamp, on-time block) -> clean', + record: v3(), + status: status('h', ORDER_STATUS.FILLED, { fillBlock: DECAY_START_BLOCK, fillTimestamp: NOW + 10_000 }), + expected: 'resolved', + outcome: PostedOrderOutcome.FILLED, + faded: 0, + }, + { + name: 'V3 filled without fillBlock -> unclassifiable', + record: v3(), + status: status('h', ORDER_STATUS.FILLED, { fillTimestamp: NOW }), + expected: 'unclassifiable', + }, + { + name: 'V3 record missing decayStartBlock -> unclassifiable', + record: v3({ decayStartBlock: undefined }), + status: status('h', ORDER_STATUS.FILLED, { fillBlock: DECAY_START_BLOCK + 5 }), + expected: 'unclassifiable', + }, + { + name: 'unknown order type fill -> unclassifiable', + record: v2({ orderType: 'Priority' }), + status: status('h', ORDER_STATUS.FILLED, { fillBlock: 1, fillTimestamp: 1 }), + expected: 'unclassifiable', + }, + // Never filled. + { + name: 'V2 expired -> fade', + record: v2(), + status: status('h', ORDER_STATUS.EXPIRED), + expected: 'resolved', + outcome: PostedOrderOutcome.EXPIRED, + faded: 1, + }, + { + name: 'V3 expired -> fade', + record: v3(), + status: status('h', ORDER_STATUS.EXPIRED), + expected: 'resolved', + outcome: PostedOrderOutcome.EXPIRED, + faded: 1, + }, + { + name: 'cancelled -> recorded without verdict', + record: v2(), + status: status('h', ORDER_STATUS.CANCELLED), + expected: 'resolved', + outcome: PostedOrderOutcome.CANCELLED, + faded: undefined, + }, + { + name: 'insufficient-funds -> recorded without verdict', + record: v3(), + status: status('h', ORDER_STATUS.INSUFFICIENT_FUNDS), + expected: 'resolved', + outcome: PostedOrderOutcome.INSUFFICIENT_FUNDS, + faded: undefined, + }, + { + name: 'error -> recorded without verdict', + record: v2(), + status: status('h', ORDER_STATUS.ERROR), + expected: 'resolved', + outcome: PostedOrderOutcome.ERROR, + faded: undefined, + }, + // Not terminal / unknown. + { + name: 'open past deadline -> still pending (poller lag)', + record: v2(), + status: status('h', ORDER_STATUS.OPEN), + expected: 'still-open', + }, + { + name: 'unknown status string -> unclassifiable', + record: v2(), + status: status('h', 'settling'), + expected: 'unclassifiable', + }, + ]; + + it.each(cases)('$name', ({ record, status, expected, outcome, faded }) => { + const classification = classifyOutcome(record, status, NOW); + expect(classification.kind).toBe(expected); + if (classification.kind === 'resolved') { + expect(classification.resolution.outcome).toBe(outcome); + expect(classification.resolution.faded).toBe(faded); + expect(classification.resolution.orderStatus).toBe(status.orderStatus); + expect(classification.resolution.resolvedAt).toBe(NOW); + expect(classification.resolution.fillBlock).toBe(status.fillBlock); + expect(classification.resolution.fillTimestamp).toBe(status.fillTimestamp); + } + }); + }); + + describe('fadedForScoring (parity flag)', () => { + const cancelled = resolved(v2(), status('h', ORDER_STATUS.CANCELLED)); + const insufficient = resolved(v2(), status('h', ORDER_STATUS.INSUFFICIENT_FUNDS)); + const errored = resolved(v2(), status('h', ORDER_STATUS.ERROR)); + const expired = resolved(v2(), status('h', ORDER_STATUS.EXPIRED)); + const cleanFill = resolved(v2(), status('h', ORDER_STATUS.FILLED, { fillTimestamp: DECAY_START_TIME })); + + it('with the flag on, never-filled terminal orders score as fades like the SQL null-fill rows', () => { + const policy = { countNeverFilledTerminalAsFade: true }; + expect(fadedForScoring(cancelled, policy)).toBe(1); + expect(fadedForScoring(insufficient, policy)).toBe(1); + expect(fadedForScoring(errored, policy)).toBe(1); + }); + + it('with the flag off, they contribute no row at all', () => { + const policy = { countNeverFilledTerminalAsFade: false }; + expect(fadedForScoring(cancelled, policy)).toBeUndefined(); + expect(fadedForScoring(insufficient, policy)).toBeUndefined(); + expect(fadedForScoring(errored, policy)).toBeUndefined(); + }); + + it('the flag does not touch definitive outcomes or pending rows', () => { + for (const flag of [true, false]) { + const policy = { countNeverFilledTerminalAsFade: flag }; + expect(fadedForScoring(expired, policy)).toBe(1); + expect(fadedForScoring(cleanFill, policy)).toBe(0); + expect(fadedForScoring(v2(), policy)).toBeUndefined(); + } + }); + }); + + describe('buildFadeRows (V2_FADE_RATE_SQL semantics)', () => { + const fade = (overrides: Partial = {}) => + resolved(v2(overrides), status('h', ORDER_STATUS.EXPIRED)); + const fill = (overrides: Partial = {}) => + resolved(v2(overrides), status('h', ORDER_STATUS.FILLED, { fillTimestamp: DECAY_START_TIME })); + + it('emits one row per completed, resolved order in the SQL row shape', () => { + const a = fade({ deadline: NOW - 100, postedAt: NOW - 200 }); + const rows = buildFadeRows([a], { now: NOW }); + expect(rows).toEqual([{ fillerAddress: ADDR_A, faded: 1, postTimestamp: NOW - 200, deadline: NOW - 100 }]); + }); + + it('produces rows with exactly the keys and value types the Redshift repository returns', async () => { + // Drive the real Redshift formatter through a fake RedshiftDataClient so the shape + // comparison is against what the cron actually receives today, not a hand-written type. + V2FadesRepository.log = log; + const fakeClient = { + send: async (command: unknown) => { + if (command instanceof ExecuteStatementCommand) return { Id: 'stmt' }; + if (command instanceof DescribeStatementCommand) return { Status: 'FINISHED' }; + if (command instanceof GetStatementResultCommand) { + return { + Records: [ + [ + { stringValue: ADDR_A }, + { stringValue: `${NOW - 200}` }, + { stringValue: `${NOW - 100}` }, + { longValue: 1 }, + ], + ], + }; + } + throw new Error('unexpected command'); + }, + }; + const redshift = new V2FadesRepository(fakeClient as unknown as RedshiftDataClient, { + Database: 'db', + ClusterIdentifier: 'cluster', + SecretArn: 'arn', + }); + const [redshiftRow] = await redshift.getFades(); + const [newRow] = buildFadeRows([fade({ deadline: NOW - 100, postedAt: NOW - 200 })], { now: NOW }); + + expect(Object.keys(newRow).sort()).toEqual(Object.keys(redshiftRow).sort()); + (Object.keys(redshiftRow) as (keyof V2FadesRowType)[]).forEach((key) => { + expect(typeof newRow[key]).toBe(typeof redshiftRow[key]); + }); + expect(newRow).toEqual(redshiftRow); + }); + + it('excludes in-flight orders (deadline >= now)', () => { + const rows = buildFadeRows([fade({ deadline: NOW }), fade({ deadline: NOW + 60 }), fade({ deadline: NOW - 1 })], { + now: NOW, + }); + expect(rows.map((r) => r.deadline)).toEqual([NOW - 1]); + }); + + it('applies the 24h completion window on deadline (inclusive at the start)', () => { + const atStart = fade({ deadline: NOW - FADE_WINDOW_SECS }); + const before = fade({ deadline: NOW - FADE_WINDOW_SECS - 1 }); + const rows = buildFadeRows([atStart, before], { now: NOW }); + expect(rows.map((r) => r.deadline)).toEqual([atStart.deadline]); + }); + + it('excludes orders with no recorded outcome (still pending)', () => { + const rows = buildFadeRows([v2(), fill()], { now: NOW }); + expect(rows).toHaveLength(1); + expect(rows[0].faded).toBe(0); + }); + + it('excludes testnets, the zero filler, missing quoteId and permissioned tokens on either leg', () => { + const permissioned = PERMISSIONED_TOKENS[0].address; + const kept = fill(); + const rows = buildFadeRows( + [ + kept, + ...EXCLUDED_TESTNET_CHAIN_IDS.map((chainId) => fill({ chainId })), + fill({ fillerAddress: '0x0000000000000000000000000000000000000000' }), + fill({ quoteId: '' }), + fill({ tokenIn: permissioned }), + fill({ tokenOut: permissioned.toUpperCase().replace('0X', '0x') }), + ], + { now: NOW } + ); + expect(rows).toEqual([ + { fillerAddress: ADDR_A, faded: 0, postTimestamp: kept.postedAt, deadline: kept.deadline }, + ]); + }); + + it('keeps the latest ORDERS_PER_FILLER_LIMIT orders per filler ADDRESS by post time', () => { + // 100 clean fills posted after one fade: the fade is the 101st most recent and drops out. + const oldFade = fade({ postedAt: NOW - 50_000, deadline: NOW - 49_000 }); + const fills = Array.from({ length: ORDERS_PER_FILLER_LIMIT }, (_, i) => + fill({ postedAt: NOW - 10_000 + i, deadline: NOW - 9_000 + i }) + ); + const rows = buildFadeRows([oldFade, ...fills], { now: NOW }); + expect(rows).toHaveLength(ORDERS_PER_FILLER_LIMIT); + expect(rows.every((r) => r.faded === 0)).toBe(true); + }); + + it('latest-N is per address, not per filler identity', () => { + // Two addresses under the same endpoint each get their own 100 slots. + const a1 = Array.from({ length: ORDERS_PER_FILLER_LIMIT }, (_, i) => fill({ postedAt: NOW - 1_000 + i })); + const a2Fade = fade({ fillerAddress: ADDR_A2, postedAt: NOW - 50_000, deadline: NOW - 49_000 }); + const rows = buildFadeRows([...a1, a2Fade], { now: NOW }); + expect(rows).toHaveLength(ORDERS_PER_FILLER_LIMIT + 1); + expect(rows.filter((r) => r.fillerAddress === ADDR_A2)).toEqual([ + { fillerAddress: ADDR_A2, faded: 1, postTimestamp: a2Fade.postedAt, deadline: a2Fade.deadline }, + ]); + }); + + it('latest-N slots are consumed BEFORE the window/token/outcome filters, as in the view', () => { + // The most recent 100 posts are all permissioned-token or still-pending orders. They take + // the slots (the view partitions before the join/filters), so the older fade is evicted + // and nothing is left after filtering. + const permissioned = PERMISSIONED_TOKENS[0].address; + const oldFade = fade({ postedAt: NOW - 50_000, deadline: NOW - 49_000 }); + const slotEaters = Array.from({ length: ORDERS_PER_FILLER_LIMIT }, (_, i) => + i % 2 === 0 ? fill({ postedAt: NOW - 1_000 + i, tokenIn: permissioned }) : v2({ postedAt: NOW - 1_000 + i }) + ); + expect(buildFadeRows([oldFade, ...slotEaters], { now: NOW })).toEqual([]); + }); + + it('honours the parity flag for never-filled terminal orders', () => { + const cancelled = resolved(v2(), status('h', ORDER_STATUS.CANCELLED)); + expect(buildFadeRows([cancelled], { now: NOW, policy: { countNeverFilledTerminalAsFade: true } })).toEqual([ + { fillerAddress: ADDR_A, faded: 1, postTimestamp: cancelled.postedAt, deadline: cancelled.deadline }, + ]); + expect(buildFadeRows([cancelled], { now: NOW, policy: { countNeverFilledTerminalAsFade: false } })).toEqual([]); + }); + + it('orders rows by filler address then deadline desc, like the SQL', () => { + const rows = buildFadeRows( + [ + fill({ fillerAddress: ADDR_B, filler: FILLER_B, deadline: NOW - 10 }), + fill({ deadline: NOW - 30 }), + fill({ deadline: NOW - 10 }), + fill({ deadline: NOW - 20 }), + ], + { now: NOW } + ); + expect(rows.map((r) => [r.fillerAddress, r.deadline])).toEqual([ + [ADDR_A, NOW - 10], + [ADDR_A, NOW - 20], + [ADDR_A, NOW - 30], + [ADDR_B, NOW - 10], + ]); + }); + }); + + describe('resolvePendingOutcomes + getFades', () => { + let repo: MockPostedOrderRepository; + let service: MockOrderStatusProvider; + let source: OrderServiceFadesSource; + + beforeEach(() => { + repo = new MockPostedOrderRepository(); + service = new MockOrderStatusProvider(); + source = new OrderServiceFadesSource({ + postedOrders: repo, + orderStatus: service, + fillerEndpoints: () => [FILLER_A, FILLER_B], + log, + now: () => NOW, + }); + }); + + const seed = async (...records: PostedOrderRecord[]) => { + for (const r of records) await repo.putPostedOrder(r); + return records; + }; + + it('resolves pending orders past their deadline and leaves live ones alone', async () => { + const [expired, filled, live] = await seed(v2(), v3(), v2({ deadline: NOW + 60 })); + service.seed( + status(expired.orderHash, ORDER_STATUS.EXPIRED), + status(filled.orderHash, ORDER_STATUS.FILLED, { fillBlock: DECAY_START_BLOCK, fillTimestamp: NOW - 100 }), + status(live.orderHash, ORDER_STATUS.OPEN) + ); + + const summary = await source.resolvePendingOutcomes(NOW); + + expect(summary).toMatchObject({ pendingPastDeadline: 2, batches: 1, resolved: 2, stillOpen: 0, notFound: 0 }); + expect(summary.byOutcome).toEqual({ [PostedOrderOutcome.EXPIRED]: 1, [PostedOrderOutcome.FILLED]: 1 }); + expect(service.batches).toEqual([[expired.orderHash, filled.orderHash]]); + expect(await repo.getPostedOrder(expired.orderHash)).toMatchObject({ + outcome: PostedOrderOutcome.EXPIRED, + orderStatus: 'expired', + faded: 1, + resolvedAt: NOW, + }); + expect(await repo.getPostedOrder(filled.orderHash)).toMatchObject({ + outcome: PostedOrderOutcome.FILLED, + fillBlock: DECAY_START_BLOCK, + fillTimestamp: NOW - 100, + faded: 0, + }); + expect(await repo.getPostedOrder(live.orderHash)).toMatchObject({ outcome: PostedOrderOutcome.PENDING }); + }); + + it('is idempotent: a second run has nothing pending and rewrites nothing', async () => { + const [expired] = await seed(v2()); + service.seed(status(expired.orderHash, ORDER_STATUS.EXPIRED)); + + await source.resolvePendingOutcomes(NOW); + const after = await repo.getPostedOrder(expired.orderHash); + const second = await source.resolvePendingOutcomes(NOW); + + expect(second).toMatchObject({ pendingPastDeadline: 0, batches: 0, resolved: 0 }); + expect(service.batches).toHaveLength(1); + expect(await repo.getPostedOrder(expired.orderHash)).toEqual(after); + }); + + it('leaves still-open and not-found orders pending, and counts them', async () => { + const [open, missing] = await seed(v2(), v2()); + service.seed(status(open.orderHash, ORDER_STATUS.OPEN)); + + const summary = await source.resolvePendingOutcomes(NOW); + + expect(summary).toMatchObject({ pendingPastDeadline: 2, resolved: 0, stillOpen: 1, notFound: 1 }); + expect(await repo.getPendingPastDeadline(NOW)).toHaveLength(2); + expect(await repo.getPostedOrder(missing.orderHash)).toMatchObject({ outcome: PostedOrderOutcome.PENDING }); + }); + + it('leaves unclassifiable orders pending and counts them', async () => { + const [fillNoTiming] = await seed(v3()); + service.seed(status(fillNoTiming.orderHash, ORDER_STATUS.FILLED)); + + const summary = await source.resolvePendingOutcomes(NOW); + + expect(summary).toMatchObject({ resolved: 0, unclassifiable: 1 }); + expect(await repo.getPostedOrder(fillNoTiming.orderHash)).toMatchObject({ outcome: PostedOrderOutcome.PENDING }); + }); + + it('batches at the order service cap and never exceeds it', async () => { + const records = await seed(...Array.from({ length: ORDER_SERVICE_MAX_ORDER_HASHES * 2 + 7 }, () => v2())); + service.seed(...records.map((r) => status(r.orderHash, ORDER_STATUS.EXPIRED))); + + const summary = await source.resolvePendingOutcomes(NOW); + + expect(service.batches.map((b) => b.length)).toEqual([ + ORDER_SERVICE_MAX_ORDER_HASHES, + ORDER_SERVICE_MAX_ORDER_HASHES, + 7, + ]); + expect(summary).toMatchObject({ batches: 3, resolved: records.length }); + expect(await repo.getPendingPastDeadline(NOW)).toEqual([]); + }); + + it('caps the pending set per run and drains oldest-first', async () => { + const records = await seed( + ...Array.from({ length: MAX_PENDING_RESOLUTIONS_PER_RUN + 1 }, (_, i) => v2({ deadline: NOW - 100_000 + i })) + ); + service.seed(...records.map((r) => status(r.orderHash, ORDER_STATUS.EXPIRED))); + + const summary = await source.resolvePendingOutcomes(NOW); + + expect(summary.pendingPastDeadline).toBe(MAX_PENDING_RESOLUTIONS_PER_RUN); + const leftover = await repo.getPendingPastDeadline(NOW); + expect(leftover.map((r) => r.orderHash)).toEqual([records[records.length - 1].orderHash]); + }); + + it('a failed batch is counted and the run continues; a write failure does not abort the batch', async () => { + const records = await seed(...Array.from({ length: ORDER_SERVICE_MAX_ORDER_HASHES + 1 }, () => v2())); + service.seed(...records.map((r) => status(r.orderHash, ORDER_STATUS.EXPIRED))); + service.failures = 1; + const failing = records[records.length - 1].orderHash; + const realRecord = repo.recordOutcome.bind(repo); + repo.recordOutcome = async (hash, resolution) => { + if (hash === failing) throw new Error('dynamo write failed'); + return realRecord(hash, resolution); + }; + + const summary = await source.resolvePendingOutcomes(NOW); + + expect(summary).toMatchObject({ batches: 2, failedBatches: 1, resolved: 0, failedWrites: 1 }); + // The failed first batch stays pending for the next run. + expect(await repo.getPendingPastDeadline(NOW)).toHaveLength(records.length); + }); + + it('stops calling the order service after consecutive failures', async () => { + const records = await seed( + ...Array.from({ length: ORDER_SERVICE_MAX_ORDER_HASHES * (MAX_CONSECUTIVE_BATCH_FAILURES + 2) }, () => v2()) + ); + service.failures = MAX_CONSECUTIVE_BATCH_FAILURES; + + const summary = await source.resolvePendingOutcomes(NOW); + + expect(service.batches).toHaveLength(MAX_CONSECUTIVE_BATCH_FAILURES); + expect(summary).toMatchObject({ failedBatches: MAX_CONSECUTIVE_BATCH_FAILURES, skippedBatches: 2 }); + expect(await repo.getPendingPastDeadline(NOW)).toHaveLength(records.length); + }); + + it('stops starting batches once the time budget is spent', async () => { + await seed(...Array.from({ length: ORDER_SERVICE_MAX_ORDER_HASHES * 3 }, () => v2())); + source.deadlineMs = Date.now() - 1; + + const summary = await source.resolvePendingOutcomes(NOW); + + expect(service.batches).toHaveLength(0); + expect(summary).toMatchObject({ batches: 0, skippedBatches: 3 }); + }); + + it('getFades resolves first, then returns rows for the scored fillers only', async () => { + const [aFade, aFill, bFade, unknownFiller] = await seed( + v2({ deadline: NOW - 500 }), + v2({ deadline: NOW - 400 }), + v3({ filler: FILLER_B, fillerAddress: ADDR_B, deadline: NOW - 300 }), + v2({ filler: 'https://not-configured.example', fillerAddress: ADDR_A2, deadline: NOW - 200 }) + ); + service.seed( + status(aFade.orderHash, ORDER_STATUS.EXPIRED), + status(aFill.orderHash, ORDER_STATUS.FILLED, { fillTimestamp: DECAY_START_TIME - 5, fillBlock: 1 }), + status(bFade.orderHash, ORDER_STATUS.FILLED, { fillBlock: DECAY_START_BLOCK + 3, fillTimestamp: NOW - 350 }), + status(unknownFiller.orderHash, ORDER_STATUS.EXPIRED) + ); + + const rows = await source.getFades(); + + expect(rows).toEqual([ + { fillerAddress: ADDR_A, faded: 0, postTimestamp: aFill.postedAt, deadline: aFill.deadline }, + { fillerAddress: ADDR_A, faded: 1, postTimestamp: aFade.postedAt, deadline: aFade.deadline }, + { fillerAddress: ADDR_B, faded: 1, postTimestamp: bFade.postedAt, deadline: bFade.deadline }, + ]); + expect(source.lastResolution).toMatchObject({ pendingPastDeadline: 4, resolved: 4 }); + }); + + it('getFades excludes orders whose outcome is still pending after resolution', async () => { + const [open, expired] = await seed(v2({ deadline: NOW - 500 }), v2({ deadline: NOW - 400 })); + service.seed(status(open.orderHash, ORDER_STATUS.OPEN), status(expired.orderHash, ORDER_STATUS.EXPIRED)); + + const rows = await source.getFades(); + + expect(rows.map((r) => r.deadline)).toEqual([expired.deadline]); + }); + }); +}); diff --git a/test/providers/order/order-status.test.ts b/test/providers/order/order-status.test.ts new file mode 100644 index 00000000..3e630990 --- /dev/null +++ b/test/providers/order/order-status.test.ts @@ -0,0 +1,103 @@ +import { default as Logger } from 'bunyan'; + +import { + ORDER_SERVICE_MAX_ORDER_HASHES, + ORDER_STATUS_TIMEOUT_MS, + OrderServiceHttp, + UniswapXServiceProvider, +} from '../../../lib/providers/order/uniswapxService'; + +const logger = Logger.createLogger({ name: 'test' }); +logger.level(Logger.FATAL); + +const SERVICE_URL = 'https://api.example.com/'; +const hash = (i: number) => `0x${i.toString(16).padStart(64, '0')}`; + +// Records every GET and answers with a canned body; no axios module mocking. +class FakeHttp implements OrderServiceHttp { + public calls: { url: string; config: unknown }[] = []; + constructor(private readonly body: unknown = {}, private readonly error?: Error) {} + // eslint-disable-next-line @typescript-eslint/no-explicit-any + get = async (url: string, config?: unknown): Promise => { + this.calls.push({ url, config }); + if (this.error) throw this.error; + return { status: 200, data: this.body }; + }; +} + +describe('UniswapXServiceProvider getOrdersByHashes', () => { + it('queries GET /dutch-auction/orders with a comma-joined orderHashes param and a bounded timeout', async () => { + const http = new FakeHttp({ orders: [] }); + const provider = new UniswapXServiceProvider(logger, SERVICE_URL, http); + + await provider.getOrdersByHashes([hash(1), hash(2)]); + + expect(http.calls).toEqual([ + { + url: `${SERVICE_URL}dutch-auction/orders`, + config: { params: { orderHashes: `${hash(1)},${hash(2)}` }, timeout: ORDER_STATUS_TIMEOUT_MS }, + }, + ]); + }); + + it('maps status and fill timing, lowercases hashes, and drops malformed items', async () => { + const http = new FakeHttp({ + orders: [ + { + orderHash: hash(1).toUpperCase().replace('0X', '0x'), + orderStatus: 'filled', + fillBlock: 123, + fillTimestamp: 456, + extra: 'x', + }, + { orderHash: hash(2), orderStatus: 'expired' }, + { orderHash: hash(3), orderStatus: 'filled', fillBlock: '123' }, + { orderStatus: 'open' }, + 'garbage', + null, + ], + }); + const provider = new UniswapXServiceProvider(logger, SERVICE_URL, http); + + const statuses = await provider.getOrdersByHashes([hash(1), hash(2), hash(3), hash(4)]); + + expect(statuses).toEqual([ + { orderHash: hash(1), orderStatus: 'filled', fillBlock: 123, fillTimestamp: 456 }, + { orderHash: hash(2), orderStatus: 'expired' }, + // a non-numeric fillBlock is dropped rather than coerced; the classifier will flag it + { orderHash: hash(3), orderStatus: 'filled' }, + ]); + }); + + it('returns [] for an empty batch without calling the service', async () => { + const http = new FakeHttp(); + const provider = new UniswapXServiceProvider(logger, SERVICE_URL, http); + + expect(await provider.getOrdersByHashes([])).toEqual([]); + expect(http.calls).toEqual([]); + }); + + it('returns [] when the body has no orders array', async () => { + const provider = new UniswapXServiceProvider(logger, SERVICE_URL, new FakeHttp({ detail: 'weird' })); + expect(await provider.getOrdersByHashes([hash(1)])).toEqual([]); + }); + + it('refuses a batch over the order service cap instead of sending a request that would 400', async () => { + const http = new FakeHttp({ orders: [] }); + const provider = new UniswapXServiceProvider(logger, SERVICE_URL, http); + const tooMany = Array.from({ length: ORDER_SERVICE_MAX_ORDER_HASHES + 1 }, (_, i) => hash(i + 1)); + + await expect(provider.getOrdersByHashes(tooMany)).rejects.toThrow(/exceeds the order service cap/); + expect(http.calls).toEqual([]); + await expect(provider.getOrdersByHashes(tooMany.slice(1))).resolves.toEqual([]); + }); + + it('propagates transport failures so the caller can count the batch and move on', async () => { + const provider = new UniswapXServiceProvider( + logger, + SERVICE_URL, + new FakeHttp(undefined, new Error('timeout of 5000ms exceeded')) + ); + await expect(provider.getOrdersByHashes([hash(1)])).rejects.toThrow('timeout of 5000ms exceeded'); + }); +}); diff --git a/test/repositories/posted-order-repository.test.ts b/test/repositories/posted-order-repository.test.ts index 5694fd72..e33eaa05 100644 --- a/test/repositories/posted-order-repository.test.ts +++ b/test/repositories/posted-order-repository.test.ts @@ -8,6 +8,7 @@ import { PENDING_INDEX_KEY, PostedOrderOutcome, PostedOrderRecord, + PostedOrderResolution, } from '../../lib/repositories/posted-order-repository'; import { DYNAMO_CONFIG } from './shared'; @@ -128,4 +129,65 @@ describe('DynamoPostedOrderRepository', () => { expect(stored?.quoteId).toEqual('quote-2'); expect((await repo.getPendingPastDeadline(NOW)).filter((r) => r.orderHash === '0x02')).toHaveLength(1); }); + + describe('recordOutcome', () => { + const resolution: PostedOrderResolution = { + outcome: PostedOrderOutcome.FILLED, + orderStatus: 'filled', + fillBlock: 20_000_001, + fillTimestamp: NOW - 90, + faded: 1, + resolvedAt: NOW, + }; + + it('records the outcome and drops the row out of the pending index in one write', async () => { + const pending = record({ orderHash: '0x10', deadline: NOW - 100 }); + await repo.putPostedOrder(pending); + expect((await repo.getPendingPastDeadline(NOW)).map((r) => r.orderHash)).toContain('0x10'); + + await repo.recordOutcome('0x10', resolution); + + expect(await repo.getPostedOrder('0x10')).toEqual({ ...pending, ...resolution }); + expect((await repo.getPendingPastDeadline(NOW)).map((r) => r.orderHash)).not.toContain('0x10'); + // Still reachable through the filler index for row building. + expect((await repo.getFillerOrdersByDeadline(FILLER_A, NOW - 100, NOW - 100)).map((r) => r.orderHash)).toEqual([ + '0x10', + ]); + }); + + it('omits fill fields that the resolution does not carry', async () => { + await repo.putPostedOrder(record({ orderHash: '0x11', deadline: NOW - 100 })); + + await repo.recordOutcome('0x11', { + outcome: PostedOrderOutcome.CANCELLED, + orderStatus: 'cancelled', + resolvedAt: NOW, + }); + + const stored = await repo.getPostedOrder('0x11'); + expect(stored).toMatchObject({ + outcome: PostedOrderOutcome.CANCELLED, + orderStatus: 'cancelled', + resolvedAt: NOW, + }); + expect(stored).not.toHaveProperty('fillBlock'); + expect(stored).not.toHaveProperty('fillTimestamp'); + expect(stored).not.toHaveProperty('faded'); + }); + + it('is idempotent: recording the same resolution twice leaves the row unchanged', async () => { + await repo.putPostedOrder(record({ orderHash: '0x12', deadline: NOW - 100 })); + + await repo.recordOutcome('0x12', resolution); + const first = await repo.getPostedOrder('0x12'); + await repo.recordOutcome('0x12', resolution); + + expect(await repo.getPostedOrder('0x12')).toEqual(first); + }); + + it('rejects for an unknown order hash instead of creating a phantom row', async () => { + await expect(repo.recordOutcome('0xdoesnotexist', resolution)).rejects.toThrow(); + expect(await repo.getPostedOrder('0xdoesnotexist')).toBeUndefined(); + }); + }); });