Implement Redis-based leader election for background jobs

README.md

@@ -89,6 +89,75 @@ periodically re-checks that condition against the live network:
   that cycle rather than crashing — airdrops are simply re-checked on the
   next tick.
 
+### Leader Election
+
+Background jobs (price refresh, webhook retry worker, airdrop expiry) use
+**Redis-based leader election** to ensure that across any number of horizontally-
+scaled replicas, only one instance actively runs each job at any given time.
+
+**Mechanism:**
+
+Each job type has its own Redis lock key (e.g. `leader:price_refresh`,
+`leader:webhook_retry`, `leader:airdrop_expiry`). On startup, every replica
+attempts to acquire the lock via `SET key value NX PX <ttl_ms>`. The instance
+that succeeds becomes the **leader** and runs the actual scheduled work. All
+other replicas are **followers** — they stay ready to take over, running only a
+periodic renewal check loop.
+
+The current leader periodically renews the lease using an atomic Lua script
+(`GET` + `PEXPIRE` in one round trip, keyed to only succeed if the stored value
+still matches the leader's instance ID). If the leader process dies or becomes
+unresponsive, the lease expires automatically after the TTL, and a follower
+detects the expiry on its next renewal check and acquires leadership.
+
+**Failover timing:**
+
+| Parameter | Default | Description |
+|-----------|---------|-------------|
+| `LEASE_TTL_MS` | 15000 (15s) | How long a lease is valid without renewal |
+| `LEASE_RENEW_INTERVAL_MS` | 5000 (5s) | How often the leader renews (and followers attempt to acquire) |
+
+- **Best-case failover** (leader stops gracefully): lease is released immediately
+  via the Lua-based conditional `DEL`; a follower acquires within one renewal
+  check interval (~5s).
+- **Worst-case failover** (leader crashes without cleanup): lease expires after
+  `LEASE_TTL_MS` (15s); the next follower renewal check detects it and acquires
+  (up to `LEASE_TTL_MS + LEASE_RENEW_INTERVAL_MS` ≈ 20s total).
+
+**Verifying leadership:**
+
+Check the `GET /health` endpoint. Each job entry includes a `leader` field
+(`true`/`false`) and `leader_instance_id` identifying which replica holds the
+lock. The top-level `leader_election` object shows the local instance's identity
+and lease configuration.
+
+```bash
+curl http://localhost:4000/health | jq '.jobs.price_refresh.leader'
+```
+
+To see which replica holds the lock from Redis directly:
+
+```bash
+redis-cli GET leader:price_refresh
+redis-cli GET leader:webhook_retry
+redis-cli GET leader:airdrop_expiry
+```
+
+**Graceful shutdown:**
+
+When a leader receives `SIGTERM`/`SIGINT`, the shutdown sequence releases the
+lease via the atomic conditional-DEL Lua script before closing the Redis
+connection, minimizing the failover window for followers.
+
+**Important caveat:**
+
+Leader election ensures only one instance runs the scheduled job logic, but it
+does not replace the need for atomic Redis operations within individual job ticks.
+For example, `deliveryRepository.popDueRetries` uses its own Lua-based atomic
+claim to prevent double-processing during any brief overlap during leadership
+handoffs. This is a separate concern that leader election complements but does
+not solve on its own.
+
 ---
 
 ## 🚀 Quick Start (Docker Development)
@@ -187,6 +256,9 @@ The application reads configurations from the `.env` file at the root.
 | `AIRDROP_JSON_MAX_BYTES` | Maximum JSON request body size; 2 MiB accommodates 10,000 inline recipients | 2097152 (2 MiB) | No |
 | `AIRDROP_RATELIMIT_WINDOW` | Per-IP airdrop mutation rate-limit window in seconds | 60 | No |
 | `AIRDROP_RATELIMIT_MAX` | Maximum create or recipient-add requests per window and IP | 10 | No |
+| `INSTANCE_ID` | Explicit instance identity for leader election; auto-generated from hostname+UUID if empty | auto | No |
+| `LEASE_TTL_MS` | Leader lease TTL in milliseconds — how long a lease is valid without renewal | 15000 | No |
+| `LEASE_RENEW_INTERVAL_MS` | How often the leader renews its lease (and followers check to acquire) | 5000 | No |
 | `LOG_LEVEL` | Logging level: `debug`, `info`, `warn`, or `error` | info | No |
 
 

TODO.md

@@ -0,0 +1,47 @@
+# Leader Election Implementation — TODO
+
+## Completed Steps
+
+### Step 1: Create `src/services/leaderElection.js`
+- [x] Implement Redis lease-based distributed lock
+- [x] `tryAcquire()` — SET key value NX PX <ttl>
+- [x] `renew()` — Lua script for atomic check-and-renew
+- [x] `release()` — Lua script for atomic conditional DEL
+- [x] `isLeader()` — check if current instance holds lease
+- [x] `getCurrentLeader()` — get current lease holder from Redis
+- [x] Periodic renewal loop (start/stop)
+- [x] Clear logging on state transitions
+
+### Step 2: Update `src/config.js`
+- [x] Add `INSTANCE_ID` env var (default: auto-generated from hostname + random suffix)
+- [x] Add `LEASE_TTL_MS` env var (default: 15000)
+- [x] Add `LEASE_RENEW_INTERVAL_MS` env var (default: 5000)
+- [x] Export `leaderElection` config section
+
+### Step 3: Create `src/jobs/leaderAwareJob.js`
+- [x] Factory that wraps job modules
+- [x] Only activates underlying job when leader
+- [x] Reacts to leadership transitions (acquire/renew/release)
+- [x] Graceful lease release on stop
+- [x] Extended getHealth() with leadership info
+- [x] Clear logging: "acting as follower" / "acquired leader lease"
+
+### Step 4: Update `src/index.js`
+- [x] Import leader election service and leader-aware job wrapper
+- [x] Create leader election instances for price_refresh, webhook_retry, airdrop_expiry
+- [x] Wrap all three background jobs with leader-aware wrapper
+- [x] Use wrapped jobs in startServer()
+- [x] Use wrapped jobs in shutdown() (await stop for graceful lease release)
+- [x] Update health endpoint to include leadership state per job
+- [x] Add `leader_election` section to health response
+- [x] Clean up duplicate/broken code
+
+### Step 5: Update `README.md`
+- [ ] Document leader-election mechanism
+- [ ] Document failover timing (TTL + renewal gap)
+- [ ] New env vars table entries
+- [ ] How to verify which replica holds the lock
+
+### Step 6: Create tests in `test/leaderElection.test.js`
+- [x] Test file created with comprehensive test suite
+

src/config.js

@@ -75,6 +75,9 @@ const env = cleanEnv(rawEnv, {
   }),
   COINGECKO_API_KEY: str({ default: '' }),
   COINMARKETCAP_API_KEY: str({ default: '' }),
+  INSTANCE_ID: str({ default: '' }),
+  LEASE_TTL_MS: positiveInteger({ default: 15000 }),
+  LEASE_RENEW_INTERVAL_MS: positiveInteger({ default: 5000 }),
   ADMIN_API_KEY: str({ default: '' }),
   AIRDROP_CSV_MAX_BYTES: positiveInteger({ default: 5 * 1024 * 1024 }),
   AIRDROP_JSON_MAX_BYTES: positiveInteger({ default: 2 * 1024 * 1024 }),
@@ -179,6 +182,11 @@ module.exports = {
     },
   },
   watchedAssets: parsedWatchedAssets,
+  leaderElection: {
+    instanceId: env.INSTANCE_ID || `${require('os').hostname()}-${require('crypto').randomUUID().slice(0, 8)}`,
+    leaseTtlMs: env.LEASE_TTL_MS,
+    renewIntervalMs: env.LEASE_RENEW_INTERVAL_MS,
+  },
   auth: {
     adminApiKey: env.ADMIN_API_KEY,
   },

src/index.js

@@ -9,6 +9,8 @@ const priceOracle = require('./services/priceOracle');
 const priceRefreshJob = require('./jobs/priceRefresh');
 const webhookRetryWorker = require('./jobs/webhookRetryWorker');
 const airdropExpiryJob = require('./jobs/airdropExpiry');
+const { createLeaderElection } = require('./services/leaderElection');
+const { makeLeaderAwareJob } = require('./jobs/leaderAwareJob');
 const { warmCache } = require('./startup/cacheWarm');
 const buildCorsMiddleware = require('./middleware/cors');
 const buildRateLimit = require('./middleware/rateLimit');
@@ -26,6 +28,34 @@ const apiDocsRouter = require('./routes/apiDocs');
 
 const priceWebSocket = require('./ws/priceWebSocket');
 
+// Wrap background jobs with leader-election coordination so that only one
+// replica across the deployment runs each job at any given time.
+// See README.md#leader-election for design, failover timing, and configuration.
+const leaderElectionPriceRefresh = createLeaderElection('price_refresh');
+const leaderElectionWebhookRetry = createLeaderElection('webhook_retry');
+const leaderElectionAirdropExpiry = createLeaderElection('airdrop_expiry');
+
+const wrappedPriceRefreshJob = makeLeaderAwareJob({
+  job: priceRefreshJob,
+  jobName: 'price_refresh',
+  leaderElection: leaderElectionPriceRefresh,
+  logger,
+});
+
+const wrappedWebhookRetryWorker = makeLeaderAwareJob({
+  job: webhookRetryWorker,
+  jobName: 'webhook_retry',
+  leaderElection: leaderElectionWebhookRetry,
+  logger,
+});
+
+const wrappedAirdropExpiryJob = makeLeaderAwareJob({
+  job: airdropExpiryJob,
+  jobName: 'airdrop_expiry',
+  leaderElection: leaderElectionAirdropExpiry,
+  logger,
+});
+
 const app = express();
 let server = {
   close(callback) {
@@ -40,16 +70,21 @@ app.use(express.json({ limit: config.airdrops.jsonMaxBytes }));
 
 app.get('/health', (req, res) => {
   const redisConnected = cache.isConnected();
-  const priceRefreshHealth = priceRefreshJob.getHealth();
-  const webhookWorkerHealth = webhookRetryWorker.getHealth();
+  const priceRefreshHealth = wrappedPriceRefreshJob.getHealth();
+  const webhookWorkerHealth = wrappedWebhookRetryWorker.getHealth();
+  const airdropExpiryHealth = wrappedAirdropExpiryJob.getHealth();
 
   // Compute overall status:
   //   unhealthy – Redis is down, or a job is stalled past its grace period
   //   degraded  – a job has not yet run but is still within its startup grace period
   //   ok        – all dependencies healthy
+  //
+  // Note: a non-leader instance reports its jobs as not healthy (since they
+  // aren't running locally), but that's expected — the leader is doing the
+  // work. The health check distinguishes "not leader" from "stalled" via the
+  // `leader` field.
   let status = 'ok';
   if (!redisConnected || !priceRefreshHealth.healthy || !webhookWorkerHealth.healthy) {
-    // Distinguish between "never started" (degraded) vs outright stalled/down (unhealthy)
     const jobsDegraded =
       (!priceRefreshHealth.healthy && !priceRefreshHealth.stalled) ||
       (!webhookWorkerHealth.healthy && !webhookWorkerHealth.stalled);
@@ -75,6 +110,9 @@ app.get('/health', (req, res) => {
           : null,
         last_error: priceRefreshHealth.lastError,
         stalled: priceRefreshHealth.stalled,
+        leader: priceRefreshHealth.leader,
+        leader_instance_id: priceRefreshHealth.leaderInstanceId,
+        leader_since: priceRefreshHealth.leaderSince,
       },
       webhook_retry_worker: {
         healthy: webhookWorkerHealth.healthy,
@@ -83,6 +121,20 @@ app.get('/health', (req, res) => {
           : null,
         last_error: webhookWorkerHealth.lastError,
         stalled: webhookWorkerHealth.stalled,
+        leader: webhookWorkerHealth.leader,
+        leader_instance_id: webhookWorkerHealth.leaderInstanceId,
+        leader_since: webhookWorkerHealth.leaderSince,
+      },
+      airdrop_expiry: {
+        healthy: airdropExpiryHealth.healthy,
+        last_success_at: airdropExpiryHealth.lastSuccessAt
+          ? new Date(airdropExpiryHealth.lastSuccessAt).toISOString()
+          : null,
+        last_error: airdropExpiryHealth.lastError,
+        stalled: airdropExpiryHealth.stalled,
+        leader: airdropExpiryHealth.leader,
+        leader_instance_id: airdropExpiryHealth.leaderInstanceId,
+        leader_since: airdropExpiryHealth.leaderSince,
       },
     },
     database: {
@@ -91,6 +143,11 @@ app.get('/health', (req, res) => {
       status: 'unused',
     },
     price_source_circuits: priceOracle.getSourceCircuitStates(),
+    leader_election: {
+      instance_id: config.leaderElection.instanceId,
+      lease_ttl_ms: config.leaderElection.leaseTtlMs,
+      renew_interval_ms: config.leaderElection.renewIntervalMs,
+    },
   });
 });
 
@@ -114,80 +171,68 @@ app.use('/api-docs', apiDocsRouter);
 app.use(notFoundHandler);
 app.use(errorHandler);
 
-const server = app.listen(config.port, () => {
-  logger.info(`SmartDrop backend running on port ${config.port}`);
-  priceRefreshJob.start();
-  indexerPoller.start();
-});
-
-let server;
-
-if (require.main === module) {
-  server = app.listen(config.port, () => {
-    logger.info(`SmartDrop backend running on port ${config.port}`);
-    priceRefreshJob.start();
-  });
-process.on('SIGTERM', async () => {
-  logger.info('SIGTERM received, shutting down');
-  priceRefreshJob.stop();
-  indexerPoller.stop();
-  server.close();
-  await cache.disconnect();
-  process.exit(0);
-});
-
-process.on('SIGINT', async () => {
-  logger.info('SIGINT received, shutting down');
-  priceRefreshJob.stop();
-  indexerPoller.stop();
-  server.close();
-  await cache.disconnect();
-  process.exit(0);
-});
 function shutdown(signal) {
   return async () => {
     logger.info(`${signal} received, shutting down`);
-    priceRefreshJob.stop();
-    webhookRetryWorker.stop();
-    airdropExpiryJob.stop();
+
+    // Stop leader-aware jobs (releases leases gracefully)
+    await wrappedPriceRefreshJob.stop();
+    await wrappedWebhookRetryWorker.stop();
+    await wrappedAirdropExpiryJob.stop();
+
+    // Stop non-leader-elected services
+    indexerPoller.stop();
     require('./ws/PriceSubscriptionManager').stopHeartbeat();
+
     if (server) server.close();
     await cache.disconnect();
     process.exit(0);
   };
 }
 
-if (require.main === module) {
-  startServer().catch((err) => {
-    logger.error('Startup failed', { error: err.message });
-    process.exit(1);
-  });
-
-  process.on('SIGTERM', shutdown('SIGTERM'));
-  process.on('SIGINT', shutdown('SIGINT'));
-}
-
 async function startServer() {
   await warmCache(config.watchedAssets);
 
   server = app.listen(config.port, () => {
     logger.info(`SmartDrop backend running on port ${config.port}`);
     priceWebSocket.attach(server);
-    priceRefreshJob.start();
-    webhookRetryWorker.start();
-    airdropExpiryJob.start();
+
+    // Start leader-aware background jobs.
+    // Each wrapped job starts a leader-election renewal loop. The underlying
+    // job (cron / setInterval) is only activated when this instance holds
+    // the leader lease. Non-leader instances remain ready to take over.
+    wrappedPriceRefreshJob.start();
+    wrappedWebhookRetryWorker.start();
+    wrappedAirdropExpiryJob.start();
+
+    // Indexer poller is not leader-elected (it uses its own cursor-based
+    // persistence in Redis and is safe for multiple replicas to run).
+    indexerPoller.start();
   });
-  module.exports.server = server;
 
   return server;
 }
 
-module.exports = { app, server };
-module.exports = app;
-module.exports.app = app;
-module.exports.server = server || {
-  close(callback) {
-    if (callback) callback();
+if (require.main === module) {
+  startServer().catch((err) => {
+    logger.error('Startup failed', { error: err.message });
+    process.exit(1);
+  });
+
+  process.on('SIGTERM', shutdown('SIGTERM'));
+  process.on('SIGINT', shutdown('SIGINT'));
+}
+
+module.exports = {
+  app,
+  server: server || {
+    close(callback) {
+      if (callback) callback();
+    },
   },
+  startServer,
+  // Exposed for testing
+  wrappedPriceRefreshJob,
+  wrappedWebhookRetryWorker,
+  wrappedAirdropExpiryJob,
 };
-module.exports.startServer = startServer;