Fix/issues 582 579 581 588

docs/mvp-express.md

@@ -181,6 +181,8 @@ const response = await fetch('https://anchor.example.com/anchor/info', {
 
 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()`

src/core/factory.ts

@@ -10,6 +10,7 @@ import type {
   DatabaseAdapter,
   InteractiveTransactionRecord,
   QueueAdapter,
+  QueueDrainStatus,
   Watcher,
   WebhookProcessor,
 } from '@/runtime/interfaces.ts';
@@ -38,6 +39,7 @@ export class AnchorInstance {
   private initPromise: Promise<void> | null = null;
   private shutdownPromise: Promise<void> | null = null;
   private backgroundJobsPromise: Promise<void> | null = null;
+  private backgroundJobsStopPromise: Promise<void> | null = null;
 
   constructor(config: Partial<AnchorKitConfig>) {
     this.config = new AnchorConfig(config);
@@ -159,14 +161,30 @@ export class AnchorInstance {
    * Stop watcher services and queue workers.
    */
   public async stopBackgroundJobs(): Promise<void> {
+    if (this.backgroundJobsStopPromise) return this.backgroundJobsStopPromise;
+    if (this.backgroundJobsPromise) await this.backgroundJobsPromise;
+    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;
+        await this.requireQueue().stop();
+        this.backgroundJobsRunning = false;
+      } finally {
+        this.backgroundJobsStopPromise = null;
+      }
+    })();
+
+    return this.backgroundJobsStopPromise;
+  }
+
+  /** Return pending and active queue counts without exposing job payloads. */
+  public getQueueStatus(): QueueDrainStatus {
+    return this.queue?.status ?? { pending: 0, active: 0 };
   }
 
   /**

src/index.ts

@@ -25,6 +25,7 @@ export type {
   IdempotencyRecord,
   InteractiveTransactionRecord,
   QueueJob,
+  QueueDrainStatus,
   QueueAdapter,
   Watcher,
   WebhookProcessor,

src/runtime/http/express-router-impl.ts

@@ -1307,13 +1307,15 @@ async function handleWebhook(
       signature,
     });
 
-    sendJson(res, 200, {
+    const response: Record<string, unknown> = {
       received: true,
       duplicate: result.duplicate,
       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',

src/runtime/interfaces.ts

@@ -159,7 +159,13 @@ export type QueueJob =
   | { type: 'process_watcher_task'; payload: { watcherTaskId: string } }
   | { type: 'cleanup_records'; payload: { retentionDays: number } };
 
+export interface QueueDrainStatus {
+  readonly pending: number;
+  readonly active: number;
+}
+
 export interface QueueAdapter {
+  readonly status?: QueueDrainStatus;
   enqueue(job: QueueJob): Promise<void>;
   start(worker: (job: QueueJob) => Promise<void>): Promise<void>;
   stop(): Promise<void>;
@@ -178,5 +184,10 @@ export interface WebhookProcessor {
     payload: Record<string, unknown>;
     rawBody: string | Buffer | Uint8Array;
     signature?: string;
-  }): Promise<{ duplicate: boolean; eventId: string; provider: string }>;
+  }): Promise<{
+    duplicate: boolean;
+    eventId: string;
+    provider: string;
+    status?: WebhookEventRecord['status'];
+  }>;
 }

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;
@@ -22,6 +22,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<void> {
     this.jobs.push(job);
     this.kick();

src/runtime/webhooks/default-webhook-processor.ts

@@ -37,7 +37,12 @@ export class DefaultWebhookProcessor implements WebhookProcessor {
     payload: Record<string, unknown>;
     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,
       };
     }
 

tests/core/factory-background-startup.test.ts

@@ -1,12 +1,25 @@
 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';
 
+function deferred() {
+  let resolve!: () => void;
+  const promise = new Promise<void>((res) => {
+    resolve = res;
+  });
+
+  return { promise, resolve };
+}
+
 class MockQueueAdapter {
   public started = false;
   public startCalls = 0;
+  public stopCalls = 0;
+  public stopGate: Promise<void> | null = null;
+  public stopEntered = deferred();
   public worker:
     | ((job: { type: string; payload: Record<string, unknown> }) => Promise<void>)
     | null = null;
@@ -20,22 +33,91 @@ class MockQueueAdapter {
   }
 
   public async stop(): Promise<void> {
+    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<void> | null = null;
+  public stopError: Error | null = null;
+  public stopEntered = deferred();
   public async start(): Promise<void> {
     this.startCalls += 1;
   }
 
   public async stop(): Promise<void> {
-    // 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;
+    }
   }
 }
 
 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<void>((resolve) => {
+      releaseWorker = resolve;
+    });
+    let signalStarted: (() => void) | undefined;
+    const workerStarted = new Promise<void>((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],
@@ -126,7 +208,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();
   });
 });

tests/exports.test.ts

@@ -16,6 +16,7 @@ describe('package root exports', () => {
 
     const job: QueueJob = { type: 'cleanup_records', payload: { retentionDays: 90 } };
     const adapter: QueueAdapter = {
+      status: { pending: 0, active: 0 },
       async enqueue(_job) {},
       async start(_worker) {},
       async stop() {},

tests/mvp-express.integration.test.ts

@@ -1898,6 +1898,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({