Merge pull request #273 from Jayking40/main

feat: implement isolated worker queues for Soroban subscription events

Author: elizabetheonoja-art

.env.example

@@ -209,6 +209,50 @@ INACTIVE_RETENTION_YEARS=3
 PII_SCRUBBING_ENABLED=true
 PII_SCRUBBING_CRON_SCHEDULE=0 2 * * 0  # Weekly on Sunday at 2 AM
 
+# Isolated Soroban Subscription Worker Queues (Issue #230)
+# These BullMQ queues run in isolation — each on-chain event type gets its own
+# worker so that a billing spike cannot starve payment-failure dunning, etc.
+SOROBAN_QUEUE_MAX_ATTEMPTS=5
+SOROBAN_QUEUE_BACKOFF_DELAY_MS=2000
+SOROBAN_QUEUE_DEFAULT_CONCURRENCY=5
+SOROBAN_QUEUE_RETAIN_COMPLETED=200
+SOROBAN_QUEUE_RETAIN_FAILED=100
+
+# SubscriptionBilled queue
+SOROBAN_QUEUE_BILLED_CONCURRENCY=10
+SOROBAN_QUEUE_BILLED_MAX_ATTEMPTS=5
+SOROBAN_QUEUE_BILLED_BACKOFF_MS=2000
+SOROBAN_QUEUE_BILLED_RATE_MAX=100
+SOROBAN_QUEUE_BILLED_RATE_WINDOW_MS=10000
+
+# TrialStarted queue
+SOROBAN_QUEUE_TRIAL_CONCURRENCY=8
+SOROBAN_QUEUE_TRIAL_MAX_ATTEMPTS=5
+SOROBAN_QUEUE_TRIAL_BACKOFF_MS=2000
+SOROBAN_QUEUE_TRIAL_RATE_MAX=50
+SOROBAN_QUEUE_TRIAL_RATE_WINDOW_MS=10000
+
+# PaymentFailed queue (higher retries, slower backoff for dunning safety)
+SOROBAN_QUEUE_PAYMENT_FAILED_CONCURRENCY=4
+SOROBAN_QUEUE_PAYMENT_FAILED_MAX_ATTEMPTS=7
+SOROBAN_QUEUE_PAYMENT_FAILED_BACKOFF_MS=5000
+SOROBAN_QUEUE_PAYMENT_FAILED_RATE_MAX=30
+SOROBAN_QUEUE_PAYMENT_FAILED_RATE_WINDOW_MS=10000
+
+# PaymentFailedGracePeriodStarted queue
+SOROBAN_QUEUE_GRACE_PERIOD_CONCURRENCY=4
+SOROBAN_QUEUE_GRACE_PERIOD_MAX_ATTEMPTS=7
+SOROBAN_QUEUE_GRACE_PERIOD_BACKOFF_MS=5000
+SOROBAN_QUEUE_GRACE_PERIOD_RATE_MAX=30
+SOROBAN_QUEUE_GRACE_PERIOD_RATE_WINDOW_MS=10000
+
+# How often (ms) the subscription worker logs a queue stats snapshot
+SOROBAN_WORKER_STATS_INTERVAL_MS=60000
+
+# Set to 'false' to disable the subscription worker pool when running via worker.js
+# (useful when running workers/sorobanSubscriptionWorker.js as a separate process)
+SOROBAN_SUBSCRIPTION_WORKERS_ENABLED=true
+
 # Sandbox Environment Configuration
 SANDBOX_ENABLED=false
 SANDBOX_MODE=testnet  # testnet or mainnet

package-lock.json

@@ -19,6 +19,9 @@
     "reconciliation:health": "node worker.js --reconciliation --health",
     "pii-scrub": "node workers/piiScrubbingWorker.js",
     "pii-scrub:dry-run": "node workers/piiScrubbingWorker.js 3 --dry-run",
+    "subscription-worker": "node workers/sorobanSubscriptionWorker.js",
+    "subscription-worker:dev": "nodemon workers/sorobanSubscriptionWorker.js",
+    "subscription-worker:health": "node workers/sorobanSubscriptionWorker.js --health",
     "test:soroban": "jest --testPathPattern=soroban",
     "test:pii": "jest piiScrubbing.test.js",
     "migrate": "node migrations/runMigrations.js",

package.json

@@ -207,6 +207,51 @@ async function loadConfig(env = process.env, vaultService = null) {
       emailQueue: env.RABBITMQ_EMAIL_QUEUE || 'substream_emails_queue',
       leaderboardQueue: env.RABBITMQ_LEADERBOARD_QUEUE || 'substream_leaderboard_queue',
     },
+    // Isolated BullMQ queue configuration for on-chain Soroban subscription events.
+    // Each event type gets its own queue with independently-tunable settings.
+    sorobanQueues: {
+      defaultMaxAttempts: Number(env.SOROBAN_QUEUE_MAX_ATTEMPTS || 5),
+      defaultBackoffDelay: Number(env.SOROBAN_QUEUE_BACKOFF_DELAY_MS || 2000),
+      defaultConcurrency: Number(env.SOROBAN_QUEUE_DEFAULT_CONCURRENCY || 5),
+      defaultRetainCompleted: Number(env.SOROBAN_QUEUE_RETAIN_COMPLETED || 200),
+      defaultRetainFailed: Number(env.SOROBAN_QUEUE_RETAIN_FAILED || 100),
+      SubscriptionBilled: {
+        concurrency: Number(env.SOROBAN_QUEUE_BILLED_CONCURRENCY || 10),
+        maxAttempts: Number(env.SOROBAN_QUEUE_BILLED_MAX_ATTEMPTS || 5),
+        backoffDelay: Number(env.SOROBAN_QUEUE_BILLED_BACKOFF_MS || 2000),
+        rateLimiter: {
+          max: Number(env.SOROBAN_QUEUE_BILLED_RATE_MAX || 100),
+          duration: Number(env.SOROBAN_QUEUE_BILLED_RATE_WINDOW_MS || 10000),
+        },
+      },
+      TrialStarted: {
+        concurrency: Number(env.SOROBAN_QUEUE_TRIAL_CONCURRENCY || 8),
+        maxAttempts: Number(env.SOROBAN_QUEUE_TRIAL_MAX_ATTEMPTS || 5),
+        backoffDelay: Number(env.SOROBAN_QUEUE_TRIAL_BACKOFF_MS || 2000),
+        rateLimiter: {
+          max: Number(env.SOROBAN_QUEUE_TRIAL_RATE_MAX || 50),
+          duration: Number(env.SOROBAN_QUEUE_TRIAL_RATE_WINDOW_MS || 10000),
+        },
+      },
+      PaymentFailed: {
+        concurrency: Number(env.SOROBAN_QUEUE_PAYMENT_FAILED_CONCURRENCY || 4),
+        maxAttempts: Number(env.SOROBAN_QUEUE_PAYMENT_FAILED_MAX_ATTEMPTS || 7),
+        backoffDelay: Number(env.SOROBAN_QUEUE_PAYMENT_FAILED_BACKOFF_MS || 5000),
+        rateLimiter: {
+          max: Number(env.SOROBAN_QUEUE_PAYMENT_FAILED_RATE_MAX || 30),
+          duration: Number(env.SOROBAN_QUEUE_PAYMENT_FAILED_RATE_WINDOW_MS || 10000),
+        },
+      },
+      PaymentFailedGracePeriodStarted: {
+        concurrency: Number(env.SOROBAN_QUEUE_GRACE_PERIOD_CONCURRENCY || 4),
+        maxAttempts: Number(env.SOROBAN_QUEUE_GRACE_PERIOD_MAX_ATTEMPTS || 7),
+        backoffDelay: Number(env.SOROBAN_QUEUE_GRACE_PERIOD_BACKOFF_MS || 5000),
+        rateLimiter: {
+          max: Number(env.SOROBAN_QUEUE_GRACE_PERIOD_RATE_MAX || 30),
+          duration: Number(env.SOROBAN_QUEUE_GRACE_PERIOD_RATE_WINDOW_MS || 10000),
+        },
+      },
+    },
     substream: {
       baseDomain: env.SUBSTREAM_BASE_DOMAIN || 'substream.app',
       backendUrl: env.SUBSTREAM_BACKEND_URL || 'http://localhost:3000',