diff --git a/src/core/factory.ts b/src/core/factory.ts index e8d83298..29a36d11 100644 --- a/src/core/factory.ts +++ b/src/core/factory.ts @@ -61,9 +61,14 @@ export class AnchorInstance { } /** - * Initialize registered plugins and all runtime services. + * Initialize registered plugins and runtime services. Concurrent calls share + * one initialization attempt; calls made during shutdown run afterward. */ public async init(): Promise { + if (this.shutdownPromise) { + await this.shutdownPromise; + return this.init(); + } if (this.initialized) return; if (this.initPromise) return this.initPromise; @@ -75,7 +80,7 @@ export class AnchorInstance { try { validatePluginRoutes(pluginRoutes); - this.database = createSqlDatabaseAdapter(frameworkConfig.database); + this.database = this.createDatabaseAdapter(); await this.database.connect(); await this.database.migrate(); @@ -170,14 +175,16 @@ export class AnchorInstance { } /** - * Cleanly shutdown all services. + * Cleanly shut down all services. If initialization is pending, shutdown waits + * for it to settle before releasing resources. The instance can then be initialized again. */ public async shutdown(): Promise { - if (!this.initialized && !this.shutdownPromise) return; + if (!this.initialized && !this.initPromise && !this.shutdownPromise) return; if (this.shutdownPromise) return this.shutdownPromise; this.shutdownPromise = (async () => { try { + await this.initPromise?.catch(() => undefined); if (!this.initialized) return; await this.stopBackgroundJobs(); @@ -197,6 +204,10 @@ export class AnchorInstance { return this.shutdownPromise; } + protected createDatabaseAdapter(): DatabaseAdapter { + return createSqlDatabaseAdapter(this.config.get('framework').database); + } + /** * Return middleware compatible with Express router mounting. * diff --git a/src/runtime/queue/in-memory-queue.ts b/src/runtime/queue/in-memory-queue.ts index 77c3da96..a26b3747 100644 --- a/src/runtime/queue/in-memory-queue.ts +++ b/src/runtime/queue/in-memory-queue.ts @@ -3,10 +3,14 @@ import type { QueueAdapter, QueueJob } from '@/runtime/interfaces.ts'; interface InMemoryQueueOptions { concurrency: number; + maxPendingJobs?: number; + onError?: (job: QueueJob, error: unknown) => void | Promise; } export class InMemoryQueueAdapter implements QueueAdapter { private readonly concurrency: number; + private readonly maxPendingJobs: number | undefined; + private readonly onError: InMemoryQueueOptions['onError']; private readonly jobs: QueueJob[] = []; private running = false; private activeWorkers = 0; @@ -18,11 +22,29 @@ export class InMemoryQueueAdapter implements QueueAdapter { if (!Number.isSafeInteger(options.concurrency) || options.concurrency < 1) { throw new ConfigError('InMemoryQueueAdapter concurrency must be a positive safe integer'); } + if ( + options.maxPendingJobs !== undefined && + (!Number.isSafeInteger(options.maxPendingJobs) || options.maxPendingJobs < 1) + ) { + throw new ConfigError('InMemoryQueueAdapter maxPendingJobs must be a positive safe integer'); + } this.concurrency = options.concurrency; + this.maxPendingJobs = options.maxPendingJobs; + this.onError = options.onError; } public async enqueue(job: QueueJob): Promise { + const canStartImmediately = + this.running && this.worker && this.activeWorkers < this.concurrency; + if ( + this.maxPendingJobs !== undefined && + this.jobs.length >= this.maxPendingJobs && + !canStartImmediately + ) { + throw new Error('In-memory queue is full'); + } + this.jobs.push(job); this.kick(); } @@ -80,8 +102,12 @@ export class InMemoryQueueAdapter implements QueueAdapter { (async () => { try { await worker(job); - } catch { - // Best-effort queue for MVP: job errors are handled by worker logic. + } catch (error) { + try { + await this.onError?.(job, error); + } catch { + // Reporting failures must not prevent other queued jobs from running. + } } finally { this.activeWorkers -= 1; diff --git a/src/runtime/webhooks/default-webhook-processor.ts b/src/runtime/webhooks/default-webhook-processor.ts index c0d76c1c..72171370 100644 --- a/src/runtime/webhooks/default-webhook-processor.ts +++ b/src/runtime/webhooks/default-webhook-processor.ts @@ -7,6 +7,10 @@ interface DefaultWebhookProcessorOptions { database: DatabaseAdapter; } +const DEFAULT_CALLBACK_TIMEOUT_MS = 30_000; +const CALLBACK_FAILURE_MESSAGE = 'Webhook callback failed'; +const CALLBACK_TIMEOUT_MESSAGE = 'Webhook callback timed out'; + function toComparableBuffer(value: string): Buffer { return Buffer.from(value, 'utf8'); } @@ -37,7 +41,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({ @@ -56,31 +65,52 @@ export class DefaultWebhookProcessor implements WebhookProcessor { } try { - await this.config.webhooks?.onEvent?.( - { - id: insertion.record.id, - eventId: insertion.record.eventId, - provider: insertion.record.provider, - payload: insertion.record.payload, - }, - { - receivedAt: insertion.record.createdAt, - signature: input.signature, - }, - ); + const callback = this.config.webhooks?.onEvent; + if (callback) { + let timeout: ReturnType | undefined; + try { + await Promise.race([ + Promise.resolve().then(() => + callback( + { + id: insertion.record.id, + eventId: insertion.record.eventId, + provider: insertion.record.provider, + payload: insertion.record.payload, + }, + { + receivedAt: insertion.record.createdAt, + signature: input.signature, + }, + ), + ), + new Promise((_, reject) => { + timeout = setTimeout( + () => reject(new Error(CALLBACK_TIMEOUT_MESSAGE)), + this.config.webhooks?.callbackTimeoutMs ?? DEFAULT_CALLBACK_TIMEOUT_MS, + ); + }), + ]); + } finally { + if (timeout) clearTimeout(timeout); + } + } await this.database.updateWebhookEventStatus({ id: insertion.record.id, status: 'processed', }); } catch (error) { - const message = error instanceof Error ? error.message : 'Unknown webhook callback error'; + const message = + error instanceof Error && error.message === CALLBACK_TIMEOUT_MESSAGE + ? CALLBACK_TIMEOUT_MESSAGE + : CALLBACK_FAILURE_MESSAGE; await this.database.updateWebhookEventStatus({ id: insertion.record.id, status: 'failed', errorMessage: message, }); - throw error; + throw new Error('Webhook processing failed', { cause: error }); } return { diff --git a/src/types/config.ts b/src/types/config.ts index 1aac1ea6..fe394afe 100644 --- a/src/types/config.ts +++ b/src/types/config.ts @@ -1,3 +1,5 @@ +import type { QueueJob } from '@/runtime/interfaces.ts'; + /** * Configuration Types for Anchor-Kit * Defines the complete configuration interface required to initialize an Anchor-Kit instance @@ -422,6 +424,12 @@ export interface FrameworkConfig { * @optional - defaults to 1 */ concurrency?: number; + + /** Maximum number of jobs waiting for a worker; in-flight jobs are excluded. Defaults to unlimited. */ + maxPendingJobs?: number; + + /** Called when a queue worker fails. Errors from this callback are ignored. */ + onError?: (job: QueueJob, error: unknown) => void | Promise; }; /** @@ -633,6 +641,9 @@ export interface AnchorKitConfig { * Webhook integration configuration. */ webhooks?: { + /** Maximum time to wait for onEvent, in milliseconds. Defaults to 30000. */ + callbackTimeoutMs?: number; + /** * Called after webhook event verification and persistence. */ diff --git a/src/utils/validation-helpers.ts b/src/utils/validation-helpers.ts index 5cf715b2..f2a0facc 100644 --- a/src/utils/validation-helpers.ts +++ b/src/utils/validation-helpers.ts @@ -578,6 +578,13 @@ function validateAnchorKitConfig(config: AnchorKitConfig): boolean { if (!assets) throw new Error('Missing required top-level field: assets'); if (!framework) throw new Error('Missing required top-level field: framework'); + if ( + config.webhooks?.callbackTimeoutMs !== undefined && + !isPositiveSafeInteger(config.webhooks.callbackTimeoutMs) + ) { + throw new Error('webhooks.callbackTimeoutMs must be a positive safe integer'); + } + NetworkConfigSchema.validate(network); SecurityConfigSchema.validate(security); if (security.enableClientAttribution) { diff --git a/tests/core/config.test.ts b/tests/core/config.test.ts index b7d0f176..95be819c 100644 --- a/tests/core/config.test.ts +++ b/tests/core/config.test.ts @@ -30,6 +30,17 @@ describe('AnchorConfig', () => { }, }; + it('requires a positive safe webhook callback timeout', () => { + for (const callbackTimeoutMs of [0, -1, 1.5, Number.NaN, Number.POSITIVE_INFINITY]) { + expect(() => + new AnchorConfig({ + ...validBaseConfig, + webhooks: { callbackTimeoutMs }, + }).validate(), + ).toThrow(/webhooks\.callbackTimeoutMs must be a positive safe integer/); + } + }); + function configWithPlugins( plugins: NonNullable, ): AnchorKitConfig { diff --git a/tests/core/factory-lifecycle.test.ts b/tests/core/factory-lifecycle.test.ts new file mode 100644 index 00000000..ccb5cd62 --- /dev/null +++ b/tests/core/factory-lifecycle.test.ts @@ -0,0 +1,89 @@ +import { AnchorInstance } from '@/core/factory.ts'; +import type { AnchorKitConfig } from '@/types/config.ts'; +import type { DatabaseAdapter } from '@/runtime/interfaces.ts'; +import { Keypair } from '@stellar/stellar-sdk'; +import { describe, expect, it, vi } from 'vitest'; + +describe('AnchorInstance lifecycle', () => { + it('waits for pending init before shutdown and supports later reinitialization', async () => { + let signalConnectStarted: (() => void) | undefined; + const connectStarted = new Promise((resolve) => { + signalConnectStarted = resolve; + }); + let releaseConnect: (() => void) | undefined; + const connectGate = new Promise((resolve) => { + releaseConnect = resolve; + }); + const signingKey = Keypair.random().secret(); + const anchor = new DelayedConnectAnchor( + { + network: { network: 'testnet' }, + server: { port: 3000 }, + security: { + sep10SigningKey: signingKey, + interactiveJwtSecret: 'jwt-secret', + distributionAccountSecret: 'dist-secret', + }, + assets: { + assets: [ + { + code: 'USDC', + issuer: 'GBBD47IF6LWK7P7MDEVSCWR7DPUWV3NY3DTQEVFL4NAT4AQH3ZLLFLA5', + }, + ], + }, + framework: { + database: { provider: 'sqlite', url: 'file:unused' }, + watchers: { enabled: false }, + }, + }, + connectGate, + () => signalConnectStarted?.(), + ); + + const firstInit = anchor.init(); + const concurrentInit = anchor.init(); + await connectStarted; + + const shutdown = anchor.shutdown(); + const initDuringShutdown = anchor.init(); + releaseConnect?.(); + + await Promise.all([firstInit, concurrentInit, shutdown, initDuringShutdown]); + await anchor.init(); + expect(anchor.databaseAdapters).toHaveLength(2); + expect(anchor.databaseAdapters[0]?.disconnect).toHaveBeenCalledTimes(1); + expect(() => anchor.getExpressRouter()).not.toThrow(); + await anchor.shutdown(); + expect(anchor.databaseAdapters[1]?.disconnect).toHaveBeenCalledTimes(1); + }); +}); + +class DelayedConnectAnchor extends AnchorInstance { + public readonly databaseAdapters: DatabaseAdapter[] = []; + private connectCount = 0; + + constructor( + config: Partial, + private readonly connectGate: Promise, + private readonly onConnectStart: () => void, + ) { + super(config); + } + + protected override createDatabaseAdapter(): DatabaseAdapter { + const adapter = { + connect: vi.fn(async () => { + this.connectCount += 1; + if (this.connectCount === 1) { + this.onConnectStart(); + await this.connectGate; + } + }), + migrate: vi.fn().mockResolvedValue(undefined), + disconnect: vi.fn().mockResolvedValue(undefined), + } as unknown as DatabaseAdapter; + this.databaseAdapters.push(adapter); + return adapter; + } +} diff --git a/tests/runtime/queue.unit.test.ts b/tests/runtime/queue.unit.test.ts index ced7f7c4..7ffe4613 100644 --- a/tests/runtime/queue.unit.test.ts +++ b/tests/runtime/queue.unit.test.ts @@ -483,4 +483,107 @@ describe('InMemoryQueueAdapter', () => { ).rejects.toThrow(/Unknown queue job type: unknown_job/); await anchor.shutdown(); }); + + it('reports a failed job and continues processing later jobs', async () => { + const failures: Array<{ job: QueueJob; error: unknown }> = []; + const completed: string[] = []; + let resolveCompleted: (() => void) | undefined; + const completedPromise = new Promise((resolve) => { + resolveCompleted = resolve; + }); + const queue = new InMemoryQueueAdapter({ + concurrency: 1, + onError: (job, error) => { + failures.push({ job, error }); + }, + }); + + await queue.start(async (job) => { + if (job.type === 'process_watcher_task') throw new Error('worker failed'); + completed.push(job.type); + resolveCompleted?.(); + }); + await queue.enqueue({ type: 'process_watcher_task', payload: { watcherTaskId: 'fail' } }); + await queue.enqueue({ type: 'cleanup_records', payload: { retentionDays: 90 } }); + await completedPromise; + await queue.stop(); + + expect(failures).toHaveLength(1); + expect(failures[0]?.job.type).toBe('process_watcher_task'); + expect(failures[0]?.error).toEqual(new Error('worker failed')); + expect(completed).toEqual(['cleanup_records']); + }); + + it('limits pending jobs and accepts more after the queue drains', async () => { + const queue = new InMemoryQueueAdapter({ concurrency: 1, maxPendingJobs: 1 }); + let completedCount = 0; + let resolveDrained: (() => void) | undefined; + const drained = new Promise((resolve) => { + resolveDrained = resolve; + }); + + await queue.enqueue({ type: 'cleanup_records', payload: { retentionDays: 1 } }); + await expect( + queue.enqueue({ type: 'cleanup_records', payload: { retentionDays: 2 } }), + ).rejects.toThrow('In-memory queue is full'); + + await queue.start(async () => { + completedCount += 1; + if (completedCount === 2) resolveDrained?.(); + }); + await new Promise((resolve) => setTimeout(resolve, 0)); + await queue.enqueue({ type: 'cleanup_records', payload: { retentionDays: 3 } }); + await drained; + await queue.stop(); + + expect(completedCount).toBe(2); + }); + + it('excludes in-flight work from capacity and rejects concurrent overflow', async () => { + const queue = new InMemoryQueueAdapter({ concurrency: 1, maxPendingJobs: 1 }); + let signalWorkerStarted: (() => void) | undefined; + const workerStarted = new Promise((resolve) => { + signalWorkerStarted = resolve; + }); + let releaseWorker: (() => void) | undefined; + const workerGate = new Promise((resolve) => { + releaseWorker = resolve; + }); + let signalPendingComplete: (() => void) | undefined; + const pendingComplete = new Promise((resolve) => { + signalPendingComplete = resolve; + }); + const completed: string[] = []; + + await queue.start(async (job) => { + if (job.type === 'process_watcher_task') { + signalWorkerStarted?.(); + await workerGate; + } + completed.push( + job.type === 'process_watcher_task' + ? 'in-flight' + : job.type === 'cleanup_records' + ? `${job.payload.retentionDays}` + : 'unexpected', + ); + if (job.type === 'cleanup_records' && job.payload.retentionDays === 1) { + signalPendingComplete?.(); + } + }); + await queue.enqueue({ type: 'process_watcher_task', payload: { watcherTaskId: 'in-flight' } }); + await workerStarted; + + const enqueueResults = await Promise.allSettled([ + queue.enqueue({ type: 'cleanup_records', payload: { retentionDays: 1 } }), + queue.enqueue({ type: 'cleanup_records', payload: { retentionDays: 2 } }), + ]); + expect(enqueueResults.map((result) => result.status)).toEqual(['fulfilled', 'rejected']); + + releaseWorker?.(); + await pendingComplete; + await queue.stop(); + + expect(completed).toEqual(['in-flight', '1']); + }); }); diff --git a/tests/runtime/webhook-processor.unit.test.ts b/tests/runtime/webhook-processor.unit.test.ts index 54956f29..2b165195 100644 --- a/tests/runtime/webhook-processor.unit.test.ts +++ b/tests/runtime/webhook-processor.unit.test.ts @@ -40,14 +40,13 @@ describe('DefaultWebhookProcessor Unit Tests', () => { rawBody: '{}', }; - // Should rethrow the error - await expect(processor.process(input)).rejects.toThrow('Callback failed'); + await expect(processor.process(input)).rejects.toThrow('Webhook processing failed'); // Should have updated status to failed with error message expect(mockDatabase.updateWebhookEventStatus).toHaveBeenCalledWith({ id: 'internal-id', status: 'failed', - errorMessage: 'Callback failed', + errorMessage: 'Webhook callback failed', }); });