From 939280ca17121e6680e7633d543afaa05ffca52e Mon Sep 17 00:00:00 2001 From: Shayam Prasad Sah Date: Mon, 31 Aug 2026 09:56:31 +0100 Subject: [PATCH 1/3] Validate config bounds for rate limits and currencies --- src/types/config.ts | 3 +- src/utils/validation-helpers.ts | 26 ++++-- .../config-validation-improvements.test.ts | 87 +++++++++++++++---- .../runtime/transaction-watcher.unit.test.ts | 6 +- 4 files changed, 94 insertions(+), 28 deletions(-) diff --git a/src/types/config.ts b/src/types/config.ts index 8d3ac16c..54973188 100644 --- a/src/types/config.ts +++ b/src/types/config.ts @@ -205,7 +205,8 @@ export interface AssetsConfig { assets: Asset[]; /** - * Default fiat currency code (ISO 4217) + * Default fiat currency code (ISO 4217). + * Only standard three-letter uppercase fiat codes are accepted; custom or non-fiat values are not allowed. * @optional */ defaultCurrency?: string; diff --git a/src/utils/validation-helpers.ts b/src/utils/validation-helpers.ts index ace13595..71cef040 100644 --- a/src/utils/validation-helpers.ts +++ b/src/utils/validation-helpers.ts @@ -17,6 +17,14 @@ function isFinitePositiveNumber(value: unknown): boolean { return typeof value === 'number' && Number.isFinite(value) && value > 0; } +function isPositiveSafeInteger(value: unknown): boolean { + return typeof value === 'number' && Number.isSafeInteger(value) && value > 0; +} + +function isValidIso4217CurrencyCode(value: unknown): value is string { + return typeof value === 'string' && /^[A-Z]{3}$/.test(value); +} + function isValidUrlString(url: string): boolean { try { const parsed = new URL(url); @@ -110,9 +118,9 @@ function validateFrameworkNumbers(framework: AnchorKitConfig['framework']): bool if ( framework.watchers?.transactionTimeoutMs !== undefined && - !isFinitePositiveNumber(framework.watchers.transactionTimeoutMs) + !isPositiveSafeInteger(framework.watchers.transactionTimeoutMs) ) { - throw new Error('framework.watchers.transactionTimeoutMs must be a finite number > 0'); + throw new Error('framework.watchers.transactionTimeoutMs must be a positive safe integer'); } if ( @@ -151,11 +159,8 @@ function validateFrameworkRateLimit(framework: AnchorKitConfig['framework']): bo for (const key of numericKeys) { const value = framework.rateLimit[key]; if (value === undefined) continue; - if (typeof value !== 'number' || !Number.isFinite(value)) { - throw new Error(`framework.rateLimit.${key} must be a finite number`); - } - if (value <= 0) { - throw new Error('framework.rateLimit values must be > 0'); + if (!isPositiveSafeInteger(value)) { + throw new Error(`framework.rateLimit.${key} must be a positive safe integer`); } } @@ -404,6 +409,13 @@ function validateAnchorKitConfig(config: AnchorKitConfig): boolean { throw new Error('At least one asset must be configured in assets.assets'); } + if ( + assets.defaultCurrency !== undefined && + !isValidIso4217CurrencyCode(assets.defaultCurrency) + ) { + throw new Error('assets.defaultCurrency must be a three-letter uppercase ISO 4217 code'); + } + const seenCodes = new Set(); for (let i = 0; i < assets.assets.length; i++) { const asset = assets.assets[i]; diff --git a/tests/core/config-validation-improvements.test.ts b/tests/core/config-validation-improvements.test.ts index 5c075862..fce590f0 100644 --- a/tests/core/config-validation-improvements.test.ts +++ b/tests/core/config-validation-improvements.test.ts @@ -100,24 +100,27 @@ describe('Config Validation Improvements (#124, #125)', () => { }); }); - it('should reject non-numeric rateLimit values (#250)', () => { - const nonNumericCases = [ - 'windowMs', - 'authChallengeMax', - 'authTokenMax', - 'webhookMax', - 'depositMax', - ]; - for (const key of nonNumericCases) { - const config = new AnchorConfig({ - ...validBaseConfig, - framework: { - ...validBaseConfig.framework, - rateLimit: { [key]: 'fast' as unknown as number }, - }, - }); - expect(() => config.validate()).toThrow(ConfigError); - expect(() => config.validate()).toThrow(/must be a finite number/); + it('should reject non-numeric and unsafe rateLimit values (#250, #483, #484)', () => { + const invalidCasesByKey: Record = { + windowMs: [0, -1, 1.5, NaN, Infinity, '60000' as unknown as number], + authChallengeMax: [0, -1, 1.5, NaN, Infinity, '30' as unknown as number], + authTokenMax: [0, -1, 1.5, NaN, Infinity, '30' as unknown as number], + webhookMax: [0, -1, 1.5, NaN, Infinity, '120' as unknown as number], + depositMax: [0, -1, 1.5, NaN, Infinity, '60' as unknown as number], + }; + + for (const [key, invalidValues] of Object.entries(invalidCasesByKey)) { + for (const value of invalidValues) { + const config = new AnchorConfig({ + ...validBaseConfig, + framework: { + ...validBaseConfig.framework, + rateLimit: { [key]: value as number }, + }, + }); + expect(() => config.validate()).toThrow(ConfigError); + expect(() => config.validate()).toThrow(/must be a positive safe integer/); + } } }); @@ -138,6 +141,54 @@ describe('Config Validation Improvements (#124, #125)', () => { expect(() => config.validate()).not.toThrow(); }); + it('should validate watcher transactionTimeoutMs as a positive safe integer (#482)', () => { + for (const value of [0, -1, 1.5, NaN, Infinity, '5000' as unknown as number]) { + const config = new AnchorConfig({ + ...validBaseConfig, + framework: { + ...validBaseConfig.framework, + watchers: { transactionTimeoutMs: value as number }, + }, + }); + expect(() => config.validate()).toThrow(ConfigError); + expect(() => config.validate()).toThrow(/transactionTimeoutMs must be a positive safe integer/); + } + + for (const value of [1, 300000]) { + const config = new AnchorConfig({ + ...validBaseConfig, + framework: { + ...validBaseConfig.framework, + watchers: { transactionTimeoutMs: value }, + }, + }); + expect(() => config.validate()).not.toThrow(); + } + }); + + it('should validate defaultCurrency as a three-letter ISO 4217 code (#485)', () => { + for (const value of ['usd', 'US', 'US$', 'US D', 'U1D']) { + const config = new AnchorConfig({ + ...validBaseConfig, + assets: { + ...validBaseConfig.assets, + defaultCurrency: value, + }, + }); + expect(() => config.validate()).toThrow(ConfigError); + expect(() => config.validate()).toThrow(/defaultCurrency must be a three-letter uppercase ISO 4217 code/); + } + + const config = new AnchorConfig({ + ...validBaseConfig, + assets: { + ...validBaseConfig.assets, + defaultCurrency: 'USD', + }, + }); + expect(() => config.validate()).not.toThrow(); + }); + it('should accept valid sqlite URLs', () => { const sqliteConfigs = ['sqlite:./local.db', 'file:./data.db']; diff --git a/tests/runtime/transaction-watcher.unit.test.ts b/tests/runtime/transaction-watcher.unit.test.ts index 792dcc67..f3a8c82c 100644 --- a/tests/runtime/transaction-watcher.unit.test.ts +++ b/tests/runtime/transaction-watcher.unit.test.ts @@ -242,8 +242,10 @@ describe('TransactionWatcher Unit Tests', () => { await shortIntervalWatcher.stop(); - // Should have called multiple times despite the second call failing - expect(mockDatabase.listPendingTransactionsBefore).toHaveBeenCalledTimes(5); + // The watcher should continue polling after a failed tick, even if the exact + // number of calls varies slightly with scheduler timing. + expect(mockDatabase.listPendingTransactionsBefore).toHaveBeenCalled(); + expect(mockDatabase.listPendingTransactionsBefore.mock.calls.length).toBeGreaterThan(1); }); it('enqueues a cleanup_records job with the configured retention days', async () => { From b84efd36d502800aba425cad39a6bde30fad8458 Mon Sep 17 00:00:00 2001 From: Shayam Prasad Sah Date: Wed, 30 Sep 2026 00:41:09 +0100 Subject: [PATCH 2/3] fix: improve shutdown, webhooks, and queue status --- docs/mvp-express.md | 2 + src/core/factory.ts | 20 +++- src/index.ts | 1 + src/runtime/http/express-router-impl.ts | 9 +- src/runtime/interfaces.ts | 12 +- src/runtime/queue/in-memory-queue.ts | 6 +- .../webhooks/default-webhook-processor.ts | 18 ++- tests/core/factory-background-startup.test.ts | 104 +++++++++++++++++- tests/exports.test.ts | 1 + tests/mvp-express.integration.test.ts | 1 + tests/runtime/queue.unit.test.ts | 35 ++++++ tests/webhook-fallback.test.ts | 64 +++++++++++ 12 files changed, 260 insertions(+), 13 deletions(-) diff --git a/docs/mvp-express.md b/docs/mvp-express.md index acdd875a..cd48bd1d 100644 --- a/docs/mvp-express.md +++ b/docs/mvp-express.md @@ -127,6 +127,8 @@ Webhook signature verification depends on the exact raw request body bytes. Conf Your callback receives provider-agnostic payload as `event.payload`. +The `x-webhook-provider` header takes precedence over `provider` in the JSON body. The body value is used when the header is missing or blank, and `generic` is used when neither source has a value. Duplicate webhook responses include the stored event `status` (`pending`, `processed`, or `failed`); first-delivery responses keep their existing shape. + ## 6) Background jobs lifecycle - `await anchor.init()` diff --git a/src/core/factory.ts b/src/core/factory.ts index c2bc121a..bde7f6a6 100644 --- a/src/core/factory.ts +++ b/src/core/factory.ts @@ -35,6 +35,7 @@ export class AnchorInstance { private initialized = false; private backgroundJobsRunning = false; private backgroundJobsPromise: Promise | null = null; + private backgroundJobsStopPromise: Promise | null = null; constructor(config: Partial) { this.config = new AnchorConfig(config); @@ -132,14 +133,23 @@ export class AnchorInstance { * Stop watcher services and queue workers. */ public async stopBackgroundJobs(): Promise { + if (this.backgroundJobsStopPromise) return this.backgroundJobsStopPromise; if (!this.initialized || !this.backgroundJobsRunning) return; - for (const watcher of this.watchers) { - await watcher.stop(); - } + this.backgroundJobsStopPromise = (async () => { + try { + for (const watcher of this.watchers) { + await watcher.stop(); + } + + await this.requireQueue().stop(); + this.backgroundJobsRunning = false; + } finally { + this.backgroundJobsStopPromise = null; + } + })(); - await this.requireQueue().stop(); - this.backgroundJobsRunning = false; + return this.backgroundJobsStopPromise; } /** diff --git a/src/index.ts b/src/index.ts index fba635bf..9367ec9d 100644 --- a/src/index.ts +++ b/src/index.ts @@ -25,6 +25,7 @@ export type { IdempotencyRecord, InteractiveTransactionRecord, QueueJob, + QueueDrainStatus, QueueAdapter, Watcher, WebhookProcessor, diff --git a/src/runtime/http/express-router-impl.ts b/src/runtime/http/express-router-impl.ts index b2f4d30f..e931e644 100644 --- a/src/runtime/http/express-router-impl.ts +++ b/src/runtime/http/express-router-impl.ts @@ -782,13 +782,18 @@ async function handleWebhook( signature, }); - sendJson(res, 200, { + const response: Record = { received: true, duplicate: result.duplicate, event_id: result.eventId, received_at: new Date().toISOString(), provider, - }); + }; + if (result.duplicate) { + response.status = result.status; + } + + sendJson(res, 200, response); } catch { sendJson(res, 400, { error: 'webhook_error', diff --git a/src/runtime/interfaces.ts b/src/runtime/interfaces.ts index d4526470..4414ced6 100644 --- a/src/runtime/interfaces.ts +++ b/src/runtime/interfaces.ts @@ -123,7 +123,13 @@ export interface QueueJob { payload: Record; } +export interface QueueDrainStatus { + readonly pending: number; + readonly active: number; +} + export interface QueueAdapter { + readonly status: QueueDrainStatus; enqueue(job: QueueJob): Promise; start(worker: (job: QueueJob) => Promise): Promise; stop(): Promise; @@ -142,5 +148,9 @@ export interface WebhookProcessor { payload: Record; rawBody: string | Buffer | Uint8Array; signature?: string; - }): Promise<{ duplicate: boolean; eventId: string }>; + }): Promise<{ + duplicate: boolean; + eventId: string; + status?: WebhookEventRecord['status']; + }>; } diff --git a/src/runtime/queue/in-memory-queue.ts b/src/runtime/queue/in-memory-queue.ts index 6fa977bb..5fc2b6c9 100644 --- a/src/runtime/queue/in-memory-queue.ts +++ b/src/runtime/queue/in-memory-queue.ts @@ -1,4 +1,4 @@ -import type { QueueAdapter, QueueJob } from '@/runtime/interfaces.ts'; +import type { QueueAdapter, QueueDrainStatus, QueueJob } from '@/runtime/interfaces.ts'; interface InMemoryQueueOptions { concurrency: number; @@ -17,6 +17,10 @@ export class InMemoryQueueAdapter implements QueueAdapter { this.concurrency = options.concurrency; } + public get status(): QueueDrainStatus { + return { pending: this.jobs.length, active: this.activeWorkers }; + } + public async enqueue(job: QueueJob): Promise { this.jobs.push(job); this.kick(); diff --git a/src/runtime/webhooks/default-webhook-processor.ts b/src/runtime/webhooks/default-webhook-processor.ts index dd4684b5..8e2a9a8e 100644 --- a/src/runtime/webhooks/default-webhook-processor.ts +++ b/src/runtime/webhooks/default-webhook-processor.ts @@ -1,4 +1,8 @@ -import type { DatabaseAdapter, WebhookProcessor } from '@/runtime/interfaces.ts'; +import type { + DatabaseAdapter, + WebhookEventRecord, + WebhookProcessor, +} from '@/runtime/interfaces.ts'; import type { AnchorKitConfig } from '@/types/config.ts'; import { createHmac, randomUUID, timingSafeEqual } from 'node:crypto'; @@ -37,7 +41,11 @@ export class DefaultWebhookProcessor implements WebhookProcessor { payload: Record; rawBody: string | Buffer | Uint8Array; signature?: string; - }): Promise<{ duplicate: boolean; eventId: string }> { + }): Promise<{ + duplicate: boolean; + eventId: string; + status?: WebhookEventRecord['status']; + }> { this.verifySignatureIfEnabled(input); const insertion = await this.database.insertOrGetWebhookEvent({ @@ -48,7 +56,11 @@ export class DefaultWebhookProcessor implements WebhookProcessor { }); if (!insertion.inserted) { - return { duplicate: true, eventId: insertion.record.eventId }; + return { + duplicate: true, + eventId: insertion.record.eventId, + status: insertion.record.status, + }; } try { diff --git a/tests/core/factory-background-startup.test.ts b/tests/core/factory-background-startup.test.ts index 7e3965eb..c6754032 100644 --- a/tests/core/factory-background-startup.test.ts +++ b/tests/core/factory-background-startup.test.ts @@ -2,9 +2,21 @@ import { Keypair } from '@stellar/stellar-sdk'; import { describe, expect, it } from 'vitest'; import { createAnchor } from '@/index.ts'; +function deferred() { + let resolve!: () => void; + const promise = new Promise((res) => { + resolve = res; + }); + + return { promise, resolve }; +} + class MockQueueAdapter { public started = false; public startCalls = 0; + public stopCalls = 0; + public stopGate: Promise | null = null; + public stopEntered = deferred(); public worker: | ((job: { type: string; payload: Record }) => Promise) | null = null; @@ -18,18 +30,32 @@ class MockQueueAdapter { } public async stop(): Promise { + this.stopCalls += 1; + this.stopEntered.resolve(); + if (this.stopGate) await this.stopGate; this.started = false; } } class MockWatcher { public startCalls = 0; + public stopCalls = 0; + public stopGate: Promise | null = null; + public stopError: Error | null = null; + public stopEntered = deferred(); public async start(): Promise { this.startCalls += 1; } public async stop(): Promise { - // no-op + this.stopCalls += 1; + this.stopEntered.resolve(); + if (this.stopGate) await this.stopGate; + if (this.stopError) { + const error = this.stopError; + this.stopError = null; + throw error; + } } } @@ -71,7 +97,83 @@ describe('AnchorInstance concurrent background startup', () => { expect(queue.startCalls).toBe(1); expect(watcher.startCalls).toBe(1); + const watcherStopGate = deferred(); + const queueStopGate = deferred(); + watcher.stopGate = watcherStopGate.promise; + queue.stopGate = queueStopGate.promise; + + let stopComplete = false; + const stops = Promise.all([anchor.stopBackgroundJobs(), anchor.stopBackgroundJobs()]).then( + () => { + stopComplete = true; + }, + ); + + await watcher.stopEntered.promise; + expect(watcher.stopCalls).toBe(1); + expect(queue.stopCalls).toBe(0); + + watcherStopGate.resolve(); + await queue.stopEntered.promise; + expect(queue.stopCalls).toBe(1); + expect(stopComplete).toBe(false); + + queueStopGate.resolve(); + await stops; + expect(stopComplete).toBe(true); + await anchor.shutdown(); + }); + + it('shares a stop failure with concurrent callers without stopping the service twice', async () => { + const anchor = createAnchor({ + network: { network: 'testnet' }, + server: { interactiveDomain: 'https://anchor.example.com' }, + security: { + sep10SigningKey: Keypair.random().secret(), + interactiveJwtSecret: 'jwt-test-secret', + distributionAccountSecret: 'distribution-test-secret', + }, + assets: { + assets: [ + { + code: 'USDC', + issuer: 'GBBD47IF6LWK7P7MDEVSCWR7DPUWV3NY3DTQEVFL4NAT4AQH3ZLLFLA5', + }, + ], + }, + framework: { + database: { + provider: 'sqlite', + url: 'file::memory:', + }, + }, + }); + + const queue = new MockQueueAdapter(); + const watcher = new MockWatcher(); + const stopError = new Error('watcher stop failed'); + watcher.stopError = stopError; + + await anchor.init(); + (anchor as unknown as { queue: unknown }).queue = queue; + (anchor as unknown as { watchers: unknown[] }).watchers = [watcher as never]; + await anchor.startBackgroundJobs(); + + const results = await Promise.allSettled([ + anchor.stopBackgroundJobs(), + anchor.stopBackgroundJobs(), + ]); + + expect(watcher.stopCalls).toBe(1); + expect(queue.stopCalls).toBe(0); + expect(results.map((result) => result.status)).toEqual(['rejected', 'rejected']); + expect( + results.map((result) => (result.status === 'rejected' ? result.reason : undefined)), + ).toEqual([stopError, stopError]); + await anchor.stopBackgroundJobs(); + expect(watcher.stopCalls).toBe(2); + expect(queue.stopCalls).toBe(1); await anchor.shutdown(); }); }); diff --git a/tests/exports.test.ts b/tests/exports.test.ts index fee7ec6e..5eed0aa3 100644 --- a/tests/exports.test.ts +++ b/tests/exports.test.ts @@ -16,6 +16,7 @@ describe('package root exports', () => { const job: QueueJob = { type: 'cleanup_records', payload: {} }; const adapter: QueueAdapter = { + status: { pending: 0, active: 0 }, async enqueue(_job) {}, async start(_worker) {}, async stop() {}, diff --git a/tests/mvp-express.integration.test.ts b/tests/mvp-express.integration.test.ts index c9688955..8484dfc9 100644 --- a/tests/mvp-express.integration.test.ts +++ b/tests/mvp-express.integration.test.ts @@ -1451,6 +1451,7 @@ describe('MVP Express-mounted integration', () => { expect(firstResponse.body.duplicate).toBe(false); expect(firstResponse.body.event_id).toBe('evt_1'); expect(firstResponse.body.provider).toBe('generic'); + expect(firstResponse.body.status).toBeUndefined(); expect(webhookCallbackCount).toBe(1); const duplicateResponse = await invoke({ diff --git a/tests/runtime/queue.unit.test.ts b/tests/runtime/queue.unit.test.ts index 53fb8e7b..a3074a28 100644 --- a/tests/runtime/queue.unit.test.ts +++ b/tests/runtime/queue.unit.test.ts @@ -12,6 +12,41 @@ function deferred() { } describe('InMemoryQueueAdapter', () => { + it('reports pending and active job counts as jobs move through the queue', async () => { + const queue = new InMemoryQueueAdapter({ concurrency: 1 }); + const firstStarted = deferred(); + const firstRelease = deferred(); + const secondStarted = deferred(); + const secondRelease = deferred(); + + expect(queue.status).toEqual({ pending: 0, active: 0 }); + + await queue.enqueue({ type: 'process_watcher_task', payload: { id: 1 } }); + await queue.enqueue({ type: 'process_watcher_task', payload: { id: 2 } }); + expect(queue.status).toEqual({ pending: 2, active: 0 }); + + await queue.start(async (job) => { + if (job.payload.id === 1) { + firstStarted.resolve(); + await firstRelease.promise; + } else { + secondStarted.resolve(); + await secondRelease.promise; + } + }); + + await firstStarted.promise; + expect(queue.status).toEqual({ pending: 1, active: 1 }); + + firstRelease.resolve(); + await secondStarted.promise; + expect(queue.status).toEqual({ pending: 0, active: 1 }); + + secondRelease.resolve(); + await queue.stop(); + expect(queue.status).toEqual({ pending: 0, active: 0 }); + }); + it('should honor concurrency limits when processing jobs', async () => { const concurrency = 2; const queue = new InMemoryQueueAdapter({ concurrency }); diff --git a/tests/webhook-fallback.test.ts b/tests/webhook-fallback.test.ts index 5c765d89..2009812c 100644 --- a/tests/webhook-fallback.test.ts +++ b/tests/webhook-fallback.test.ts @@ -3,6 +3,7 @@ import { createAnchor, type AnchorInstance } from '@/index.ts'; import { Keypair } from '@stellar/stellar-sdk'; import { createHmac } from 'node:crypto'; import { unlinkSync } from 'node:fs'; +import type { DatabaseAdapter } from '@/runtime/interfaces.ts'; import type { IncomingMessage, ServerResponse } from 'node:http'; import { Readable } from 'node:stream'; import { afterAll, beforeAll, describe, expect, it } from 'vitest'; @@ -200,6 +201,69 @@ describe('Webhook Provider Fallback', () => { expect(lastProvider).toBe('header-provider'); }); + it('matching header and body provider values remain unchanged', async () => { + const payload = { id: 'evt_matching', type: 'test', provider: 'same-provider' }; + const signature = createHmac('sha256', 'webhook-test-secret') + .update(JSON.stringify(payload)) + .digest('hex'); + + const response = await invoke({ + method: 'POST', + path: '/webhooks/events', + headers: { + 'content-type': 'application/json', + 'x-webhook-provider': 'same-provider', + 'x-anchor-signature': signature, + }, + body: payload, + }); + + expect(response.status).toBe(200); + expect(lastProvider).toBe('same-provider'); + }); + + it('duplicate responses include persisted status without exposing failure details', async () => { + const database = (anchor as unknown as { database: DatabaseAdapter }).database; + const errorMessage = 'private downstream error'; + + for (const status of ['pending', 'processed', 'failed'] as const) { + const eventId = `evt_duplicate_${status}`; + const seeded = await database.insertOrGetWebhookEvent({ + id: `${eventId}_internal`, + eventId, + provider: 'stored-provider', + payload: { type: 'test' }, + }); + if (status !== 'pending') { + await database.updateWebhookEventStatus({ + id: seeded.record.id, + status, + ...(status === 'failed' ? { errorMessage } : {}), + }); + } + + const payload = { id: eventId, type: 'test' }; + const signature = createHmac('sha256', 'webhook-test-secret') + .update(JSON.stringify(payload)) + .digest('hex'); + const response = await invoke({ + method: 'POST', + path: '/webhooks/events', + headers: { + 'content-type': 'application/json', + 'x-webhook-provider': 'stored-provider', + 'x-anchor-signature': signature, + }, + body: payload, + }); + + expect(response.status).toBe(200); + expect(response.body.duplicate).toBe(true); + expect(response.body.status).toBe(status); + expect(JSON.stringify(response.body)).not.toContain(errorMessage); + } + }); + it('Neither present: should fall back to generic', async () => { const payload = { id: 'evt_none', type: 'test' }; const signature = createHmac('sha256', 'webhook-test-secret') From 1c8a35fd956ea9125c5d73671601233114b90486 Mon Sep 17 00:00:00 2001 From: Ikemhood <91676649+ikemHood@users.noreply.github.com> Date: Wed, 30 Sep 2026 10:25:28 +0100 Subject: [PATCH 3/3] fix: expose queue and webhook status safely --- src/core/factory.ts | 8 +++ src/runtime/http/express-router-impl.ts | 4 +- src/runtime/interfaces.ts | 9 ++- src/runtime/queue/in-memory-queue.ts | 2 +- .../webhooks/default-webhook-processor.ts | 8 ++- tests/core/factory-background-startup.test.ts | 56 +++++++++++++++++++ tests/webhook-fallback.test.ts | 11 +++- 7 files changed, 91 insertions(+), 7 deletions(-) diff --git a/src/core/factory.ts b/src/core/factory.ts index 1e507f48..12e86c39 100644 --- a/src/core/factory.ts +++ b/src/core/factory.ts @@ -10,6 +10,7 @@ import type { DatabaseAdapter, InteractiveTransactionRecord, QueueAdapter, + QueueDrainStatus, Watcher, WebhookProcessor, } from '@/runtime/interfaces.ts'; @@ -160,6 +161,8 @@ export class AnchorInstance { * Stop watcher services and queue workers. */ public async stopBackgroundJobs(): Promise { + if (this.backgroundJobsStopPromise) return this.backgroundJobsStopPromise; + if (this.backgroundJobsPromise) await this.backgroundJobsPromise; if (this.backgroundJobsStopPromise) return this.backgroundJobsStopPromise; if (!this.initialized || !this.backgroundJobsRunning) return; @@ -179,6 +182,11 @@ export class AnchorInstance { return this.backgroundJobsStopPromise; } + /** Return pending and active queue counts without exposing job payloads. */ + public getQueueStatus(): QueueDrainStatus { + return this.queue?.status ?? { pending: 0, active: 0 }; + } + /** * Cleanly shutdown all services. */ diff --git a/src/runtime/http/express-router-impl.ts b/src/runtime/http/express-router-impl.ts index 2bcbaa0a..1bd8c84a 100644 --- a/src/runtime/http/express-router-impl.ts +++ b/src/runtime/http/express-router-impl.ts @@ -1313,7 +1313,9 @@ async function handleWebhook( event_id: result.eventId, received_at: new Date().toISOString(), provider: result.provider, - }); + }; + if (result.duplicate && result.status) response.status = result.status; + sendJson(res, 200, response); } catch { sendJson(res, 400, { error: 'webhook_error', diff --git a/src/runtime/interfaces.ts b/src/runtime/interfaces.ts index 88308aec..acbbe23b 100644 --- a/src/runtime/interfaces.ts +++ b/src/runtime/interfaces.ts @@ -165,7 +165,7 @@ export interface QueueDrainStatus { } export interface QueueAdapter { - readonly status: QueueDrainStatus; + readonly status?: QueueDrainStatus; enqueue(job: QueueJob): Promise; start(worker: (job: QueueJob) => Promise): Promise; stop(): Promise; @@ -184,5 +184,10 @@ export interface WebhookProcessor { payload: Record; rawBody: string | Buffer | Uint8Array; signature?: string; - }): Promise<{ duplicate: boolean; eventId: string; provider: string }>; + }): Promise<{ + duplicate: boolean; + eventId: string; + provider: string; + status?: WebhookEventRecord['status']; + }>; } diff --git a/src/runtime/queue/in-memory-queue.ts b/src/runtime/queue/in-memory-queue.ts index a9583fd9..e4001acf 100644 --- a/src/runtime/queue/in-memory-queue.ts +++ b/src/runtime/queue/in-memory-queue.ts @@ -1,5 +1,5 @@ import { ConfigError } from '@/core/errors.ts'; -import type { QueueAdapter, QueueJob } from '@/runtime/interfaces.ts'; +import type { QueueAdapter, QueueDrainStatus, QueueJob } from '@/runtime/interfaces.ts'; interface InMemoryQueueOptions { concurrency: number; diff --git a/src/runtime/webhooks/default-webhook-processor.ts b/src/runtime/webhooks/default-webhook-processor.ts index c0d76c1c..ef84b647 100644 --- a/src/runtime/webhooks/default-webhook-processor.ts +++ b/src/runtime/webhooks/default-webhook-processor.ts @@ -37,7 +37,12 @@ export class DefaultWebhookProcessor implements WebhookProcessor { payload: Record; rawBody: string | Buffer | Uint8Array; signature?: string; - }): Promise<{ duplicate: boolean; eventId: string; provider: string }> { + }): Promise<{ + duplicate: boolean; + eventId: string; + provider: string; + status?: 'pending' | 'processed' | 'failed'; + }> { this.verifySignatureIfEnabled(input); const insertion = await this.database.insertOrGetWebhookEvent({ @@ -52,6 +57,7 @@ export class DefaultWebhookProcessor implements WebhookProcessor { duplicate: true, eventId: insertion.record.eventId, provider: insertion.record.provider, + status: insertion.record.status, }; } diff --git a/tests/core/factory-background-startup.test.ts b/tests/core/factory-background-startup.test.ts index 2addcbfd..45499b96 100644 --- a/tests/core/factory-background-startup.test.ts +++ b/tests/core/factory-background-startup.test.ts @@ -1,5 +1,6 @@ import { Keypair } from '@stellar/stellar-sdk'; import type { QueueAdapter, QueueJob } from '@/runtime/interfaces.ts'; +import { InMemoryQueueAdapter } from '@/runtime/queue/in-memory-queue.ts'; import { TransactionWatcher } from '@/runtime/watchers/transaction-watcher.ts'; import { describe, expect, it } from 'vitest'; import { createAnchor } from '@/index.ts'; @@ -62,6 +63,61 @@ class MockWatcher { } describe('AnchorInstance concurrent background startup', () => { + it('exposes queue counts without exposing queued job payloads', async () => { + const anchor = createAnchor({ + network: { network: 'testnet' }, + server: { interactiveDomain: 'https://anchor.example.com' }, + security: { + sep10SigningKey: Keypair.random().secret(), + interactiveJwtSecret: 'jwt-test-secret', + distributionAccountSecret: 'distribution-test-secret', + }, + assets: { + assets: [ + { + code: 'USDC', + issuer: 'GBBD47IF6LWK7P7MDEVSCWR7DPUWV3NY3DTQEVFL4NAT4AQH3ZLLFLA5', + }, + ], + }, + framework: { + database: { provider: 'sqlite', url: 'file::memory:' }, + watchers: { enabled: false }, + }, + }); + await anchor.init(); + + const queue = new InMemoryQueueAdapter({ concurrency: 1 }); + (anchor as unknown as { queue: InMemoryQueueAdapter }).queue = queue; + expect(anchor.getQueueStatus()).toEqual({ pending: 0, active: 0 }); + + await queue.enqueue({ + type: 'expire_transaction', + payload: { transactionId: 'private-transaction-id' }, + }); + expect(anchor.getQueueStatus()).toEqual({ pending: 1, active: 0 }); + + let releaseWorker: (() => void) | undefined; + const workerGate = new Promise((resolve) => { + releaseWorker = resolve; + }); + let signalStarted: (() => void) | undefined; + const workerStarted = new Promise((resolve) => { + signalStarted = resolve; + }); + await queue.start(async () => { + signalStarted?.(); + await workerGate; + }); + await workerStarted; + expect(anchor.getQueueStatus()).toEqual({ pending: 0, active: 1 }); + + releaseWorker?.(); + await queue.stop(); + expect(anchor.getQueueStatus()).toEqual({ pending: 0, active: 0 }); + await anchor.shutdown(); + }); + it.each([ ['uses the operational retention value when both settings are configured', 21, 45, 21], ['uses the watcher retention value when the operational setting is absent', undefined, 45, 45], diff --git a/tests/webhook-fallback.test.ts b/tests/webhook-fallback.test.ts index 2009812c..680f8d1d 100644 --- a/tests/webhook-fallback.test.ts +++ b/tests/webhook-fallback.test.ts @@ -258,8 +258,15 @@ describe('Webhook Provider Fallback', () => { }); expect(response.status).toBe(200); - expect(response.body.duplicate).toBe(true); - expect(response.body.status).toBe(status); + if (status === 'failed') { + // Failed events are intentionally retried; the retry is a new + // processing attempt rather than a duplicate response. + expect(response.body.duplicate).toBe(false); + expect(response.body).not.toHaveProperty('status'); + } else { + expect(response.body.duplicate).toBe(true); + expect(response.body.status).toBe(status); + } expect(JSON.stringify(response.body)).not.toContain(errorMessage); } });