feat(cluster): implement round-robin scheduling and worker management in clustered server mode

Author: pphatdev

Dockerfile

@@ -10,6 +10,7 @@ RUN npm ci && npm run build && npm prune --omit=dev
 
 FROM node:20-bookworm-slim AS runtime
 ENV NODE_ENV=production
+ENV WORKERS=0
 WORKDIR /app
 
 COPY --from=build /app/package*.json ./
@@ -18,4 +19,4 @@ COPY --from=build /app/dist ./dist
 COPY --from=build /app/public ./public
 
 EXPOSE 3000
-CMD ["node", "dist/index.js"]
+CMD ["node", "dist/server-cluster.js"]

README.md

@@ -320,7 +320,7 @@ For more details: [User Badges](docs/example/badge-user.md) · [Badge Collection
 - **API**: GitHub REST + GraphQL APIs with intelligent batching
 - **Caching**: Multi-tier (Memory → Redis → Source) with 2-hour default TTL
 - **Database**: SQLite with Drizzle ORM for badge counters and visitor logs
-- **Server**: Express.js with optional cluster mode for multi-core scaling
+- **Server**: Express.js with round-robin cluster load balancing in production
 - **Rendering**: Server-side SVG generation with optional WebP/PNG/GIF export
 
 ## Notes
@@ -329,6 +329,7 @@ For more details: [User Badges](docs/example/badge-user.md) · [Badge Collection
 - Without a GitHub token, API rate limits are very low (~60 requests/hour)
 - Set `GITHUB_TOKEN` to get 5,000 requests/hour
 - Redis is optional but recommended for production (enables distributed caching)
+- Docker and `npm start` now boot the clustered entrypoint; set `WORKERS` to cap worker count, or leave it at `0` to use all available CPU cores
 - User visitor badges (`/badges?username=...&name=visitors`) use IP hashing for privacy-preserving unique visitor counting
 - Project visitor badges (`/project/visitors`) increment once per same IP every 5 minutes
 

compose.yaml

@@ -6,12 +6,13 @@ services:
     container_name: ${CLOUDFLARED_TUNNEL_NAME}-app
     restart: unless-stopped
     ports:
-      - "3000:${PORT:-3000}"
+      - "3102:${PORT:-3000}"
     environment:
       NODE_ENV: production
       APP_ENV: production
       HOST: 0.0.0.0
       PORT: ${PORT:-3000}
+      WORKERS: ${WORKERS:-0}
       DATABASE_PROVIDER: ${DATABASE_PROVIDER:-sqlite}
       DATABASE_URL: ${DATABASE_URL:-/app/data/stats.db}
       CLOUDFLARE_ACCOUNT_ID: ${CLOUDFLARE_ACCOUNT_ID:-}
@@ -36,6 +37,7 @@ services:
       timeout: 10s
       retries: 5
       start_period: 20s
+    stop_grace_period: 20s
 
   redis:
     image: redis:7-alpine

docs/structures/03_PRACTICES_AND_WORKFLOW.md

@@ -25,7 +25,7 @@
 3. Database: run npm run db:migrate
 4. Testing: run npm test
 5. Build: run npm run build
-6. Deploy: run npm run start:cluster
+6. Deploy: run npm start (clustered by default) or set WORKERS to pin the worker count
 
 ## Related Documentation
 

package.json

@@ -9,7 +9,8 @@
     "vercel-build": "echo 'Vercel compiles TypeScript at runtime — no pre-build needed'",
     "dev": "node --watch --no-warnings=ExperimentalWarning --loader ts-node/esm ./src/index.ts",
     "dev:modular": "node --watch --no-warnings=ExperimentalWarning --loader ts-node/esm ./src/server.ts",
-    "start": "node dist/index.js",
+    "start": "node dist/server-cluster.js",
+    "start:single": "node dist/index.js",
     "start:cluster": "node dist/server-cluster.js",
     "start:production": "NODE_ENV=production node dist/server-cluster.js",
     "test": "jest",

src/cluster.ts

@@ -22,72 +22,110 @@ export async function startCluster(
     workerFile: string,
     options: ClusterOptions = {}
 ) {
+    const availableWorkers = typeof os.availableParallelism === 'function'
+        ? os.availableParallelism()
+        : os.cpus().length;
     const {
-        workers = os.cpus().length,
+        workers = availableWorkers,
         respawnDelay = 1000,
         maxRestarts = 5
     } = options;
+    const workerCount = Math.max(1, Math.min(workers, availableWorkers));
 
     if (cluster.isPrimary) {
+        cluster.schedulingPolicy = cluster.SCHED_RR;
+
         logger.info('Starting cluster mode', {
-            workers,
-            cpus: os.cpus().length,
+            workers: workerCount,
+            cpus: availableWorkers,
             platform: os.platform(),
-            memory: `${Math.round(os.totalmem() / 1024 / 1024 / 1024)}GB`
+            memory: `${Math.round(os.totalmem() / 1024 / 1024 / 1024)}GB`,
+            schedulingPolicy: 'round-robin'
         });
 
         const workerRestarts = new Map<number, number>();
+        const workerSlots = new Map<number, number>();
+        let isShuttingDown = false;
+        let healthCheckInterval: NodeJS.Timeout | undefined;
 
         // Spawn workers
-        for (let i = 0; i < workers; i++) {
-            spawnWorker(i + 1);
+        for (let slot = 1; slot <= workerCount; slot++) {
+            spawnWorker(slot, workerSlots);
         }
 
         // Handle worker exit
         cluster.on('exit', (worker, code, signal) => {
             const workerId = worker.id;
-            const restarts = workerRestarts.get(workerId) || 0;
+            const workerSlot = workerSlots.get(workerId) || workerId;
+            const restarts = workerRestarts.get(workerSlot) || 0;
+
+            workerSlots.delete(workerId);
 
             logger.warn('Worker died', {
                 workerId,
+                workerSlot,
                 pid: worker.process.pid,
                 code,
                 signal,
                 restarts
             });
 
+            if (isShuttingDown) {
+                return;
+            }
+
             // Check if we should respawn
             if (restarts < maxRestarts) {
-                workerRestarts.set(workerId, restarts + 1);
+                workerRestarts.set(workerSlot, restarts + 1);
 
                 setTimeout(() => {
-                    logger.info('Respawning worker', { workerId, attempt: restarts + 1 });
-                    spawnWorker(workerId);
+                    if (isShuttingDown) {
+                        return;
+                    }
+
+                    logger.info('Respawning worker', { workerId, workerSlot, attempt: restarts + 1 });
+                    spawnWorker(workerSlot, workerSlots);
                 }, respawnDelay);
             } else {
-                logger.error('Worker exceeded max restarts', undefined, { workerId, maxRestarts });
+                logger.error('Worker exceeded max restarts', undefined, {
+                    workerId,
+                    workerSlot,
+                    maxRestarts,
+                });
             }
         });
 
         // Handle worker online
         cluster.on('online', (worker) => {
+            const workerSlot = workerSlots.get(worker.id) || worker.id;
             logger.info('Worker online', {
                 workerId: worker.id,
+                workerSlot,
                 pid: worker.process.pid
             });
         });
 
         // Handle worker listening
         cluster.on('listening', (worker, address) => {
+            const workerSlot = workerSlots.get(worker.id) || worker.id;
             logger.info('Worker listening', {
                 workerId: worker.id,
+                workerSlot,
                 pid: worker.process.pid,
                 address: `${address.address}:${address.port}`
             });
         });
 
         // Graceful shutdown
         const shutdown = async () => {
+            if (isShuttingDown) {
+                return;
+            }
+
+            isShuttingDown = true;
+            if (healthCheckInterval) {
+                clearInterval(healthCheckInterval);
+            }
             logger.info('Shutting down cluster...');
 
             const workers = Object.values(cluster.workers || {});
@@ -124,7 +162,7 @@ export async function startCluster(
         process.on('SIGINT', shutdown);
 
         // Performance monitoring
-        setInterval(() => {
+        healthCheckInterval = setInterval(() => {
             const workers = Object.values(cluster.workers || {});
             const activeWorkers = workers.filter(w => w && !w.isDead()).length;
 
@@ -136,22 +174,53 @@ export async function startCluster(
             });
         }, 60000); // Every minute
 
+        healthCheckInterval.unref();
+
     } else {
         // Worker process - import and run the application
         try {
-            await import(workerFile);
+            const workerModule = await import(workerFile) as {
+                startServer?: () => Promise<unknown>;
+                stopServer?: () => Promise<void>;
+            };
+            let isWorkerShuttingDown = false;
+
+            if (typeof workerModule.startServer === 'function') {
+                await workerModule.startServer();
+            }
+
+            const shutdownWorker = async () => {
+                if (isWorkerShuttingDown) {
+                    return;
+                }
+
+                isWorkerShuttingDown = true;
+                logger.info('Worker received shutdown signal', {
+                    workerId: cluster.worker?.id,
+                    workerSlot: process.env.WORKER_SLOT,
+                });
+
+                try {
+                    await workerModule.stopServer?.();
+                } catch (error) {
+                    logger.error('Worker failed to shut down cleanly', error as Error, {
+                        workerId: cluster.worker?.id,
+                        workerSlot: process.env.WORKER_SLOT,
+                    });
+                } finally {
+                    process.exit(0);
+                }
+            };
 
             // Handle shutdown signal from master
             process.on('message', (msg) => {
                 if (msg === 'shutdown') {
-                    logger.info('Worker received shutdown signal', {
-                        workerId: cluster.worker?.id
-                    });
-                    
-                    // Gracefully close connections
-                    process.exit(0);
+                    void shutdownWorker();
                 }
             });
+
+            process.once('SIGTERM', () => void shutdownWorker());
+            process.once('SIGINT', () => void shutdownWorker());
 
         } catch (error) {
             logger.error('Worker failed to start', error as Error, {
@@ -165,9 +234,11 @@ export async function startCluster(
 /**
  * Spawn a new worker
  */
-function spawnWorker(workerId: number) {
-    const worker = cluster.fork();
-    worker.id = workerId;
+function spawnWorker(workerSlot: number, workerSlots: Map<number, number>) {
+    const worker = cluster.fork({
+        WORKER_SLOT: String(workerSlot),
+    });
+    workerSlots.set(worker.id, workerSlot);
     return worker;
 }
 

src/server-cluster.ts

@@ -14,11 +14,11 @@ import path from 'path';
 const __filename = fileURLToPath(import.meta.url);
 const __dirname = path.dirname(__filename);
 
-const workerFile = pathToFileURL(path.join(__dirname, 'index.js')).href;
+const workerFile = pathToFileURL(path.join(__dirname, 'server.js')).href;
 const workers = parseInt(process.env.WORKERS || '0') || undefined;
 
 startCluster(workerFile, {
     workers,
     respawnDelay: 1000,
-    maxRestarts: 1
+    maxRestarts: 5
 });

src/server.ts

@@ -3,16 +3,20 @@
  * Handles server initialization with modular structure
  */
 
+import { type Server as HttpServer } from 'http';
 import { type Express } from 'express';
 import { createApp, initializeRoutes, setupErrorHandlers } from './app.js';
 import { getEnv } from './shared/config/env.js';
 import { createLogger } from './shared/logs/logger.js';
 import { initializeDatabaseAsync } from './shared/config/db.js';
 import { GitHubClient } from './shared/utils/github-client.js';
-import { getRedisClient } from './shared/utils/redis-client.js';
+import { closeRedisClient, getRedisClient } from './shared/utils/redis-client.js';
 import type { ICacheService } from './services/base.service.js';
 
 const logger = createLogger({ module: 'server' });
+let activeApp: Express | null = null;
+let activeServer: HttpServer | null = null;
+let shutdownPromise: Promise<void> | null = null;
 
 // Shared cache for API responses
 const cache = new Map<string, { data: string; timestamp: number }>();
@@ -95,6 +99,10 @@ async function initializeServices(): Promise<{ cacheService?: ICacheService }> {
  * Start the server
  */
 export async function startServer(): Promise<Express> {
+    if (activeApp && activeServer?.listening) {
+        return activeApp;
+    }
+
     const env = getEnv();
 
     // Initialize services
@@ -124,6 +132,9 @@ export async function startServer(): Promise<Express> {
         });
     });
 
+    activeApp = app;
+    activeServer = server;
+
     server.on('error', (error: NodeJS.ErrnoException) => {
         logger.error('HTTP server failed to listen', error, {
             port,
@@ -136,6 +147,46 @@ export async function startServer(): Promise<Express> {
     return app;
 }
 
+export async function stopServer(): Promise<void> {
+    if (shutdownPromise) {
+        return shutdownPromise;
+    }
+
+    shutdownPromise = (async () => {
+        if (activeServer) {
+            await new Promise<void>((resolve, reject) => {
+                activeServer?.close((error) => {
+                    if (error) {
+                        reject(error);
+                        return;
+                    }
+
+                    resolve();
+                });
+            });
+
+            logger.info('HTTP server stopped');
+        }
+
+        try {
+            await closeRedisClient();
+        } catch (error) {
+            logger.warn('Failed to close Redis client cleanly', {
+                error: error instanceof Error ? error.message : String(error),
+            });
+        }
+
+        activeServer = null;
+        activeApp = null;
+    })();
+
+    try {
+        await shutdownPromise;
+    } finally {
+        shutdownPromise = null;
+    }
+}
+
 // Start server if this file is run directly
 if (import.meta.url === `file://${process.argv[1]}`) {
     startServer().catch((error) => {