diff --git a/api/src/audit/audit-action.enum.ts b/api/src/audit/audit-action.enum.ts index 4a63887..873b05c 100644 --- a/api/src/audit/audit-action.enum.ts +++ b/api/src/audit/audit-action.enum.ts @@ -24,4 +24,8 @@ export enum AuditAction { STREAM_DELETE = "stream_delete", ROLE_CHANGE = "role_change", PROFILE_UPDATE = "profile_update", + + // ── User lifecycle ──────────────────────────────────────────────── + /** Captured when a user soft-deletes their account (issue #344). */ + USER_DELETED = "user_deleted", } diff --git a/api/src/auth/users.repository.ts b/api/src/auth/users.repository.ts index 6764e6f..7d74fe1 100644 --- a/api/src/auth/users.repository.ts +++ b/api/src/auth/users.repository.ts @@ -1,5 +1,6 @@ import { Inject, Injectable } from "@nestjs/common" import { Pool } from "pg" + import { PG_POOL } from "../database/database.module" export interface User { @@ -9,6 +10,7 @@ export interface User { password_hash: string created_at: Date password_changed_at?: Date + deleted_at?: Date | null } /** @@ -23,7 +25,7 @@ export class UsersRepository { async findByEmail(email: string): Promise { const { rows } = await this.pool.query( - "SELECT id, username, email, password_hash, created_at, password_changed_at FROM users WHERE email = $1", + "SELECT id, username, email, password_hash, created_at, password_changed_at, deleted_at FROM users WHERE email = $1 AND deleted_at IS NULL", [email], ) return rows[0] ?? null @@ -31,7 +33,7 @@ export class UsersRepository { async findByUsername(username: string): Promise { const { rows } = await this.pool.query( - "SELECT id, username, email, password_hash, created_at, password_changed_at FROM users WHERE username = $1", + "SELECT id, username, email, password_hash, created_at, password_changed_at, deleted_at FROM users WHERE username = $1 AND deleted_at IS NULL", [username], ) return rows[0] ?? null @@ -39,7 +41,7 @@ export class UsersRepository { async findById(id: number): Promise { const { rows } = await this.pool.query( - "SELECT id, username, email, password_hash, created_at, password_changed_at FROM users WHERE id = $1", + "SELECT id, username, email, password_hash, created_at, password_changed_at, deleted_at FROM users WHERE id = $1 AND deleted_at IS NULL", [id], ) return rows[0] ?? null @@ -100,10 +102,26 @@ export class UsersRepository { `UPDATE users SET password_hash = $1, password_changed_at = $2 - WHERE id = $3 - RETURNING id, username, email, password_hash, created_at, password_changed_at`, + WHERE id = $3 AND deleted_at IS NULL + RETURNING id, username, email, password_hash, created_at, password_changed_at, deleted_at`, [passwordHash, passwordChangedAt, id], ) return rows[0] } + + /** + * Soft-delete a user by setting deleted_at (issue #344). + * Returns the soft-deleted user row or null if the user was not + * found or was already deleted. + */ + async softDelete(id: number): Promise { + const { rows } = await this.pool.query( + `UPDATE users + SET deleted_at = NOW() + WHERE id = $1 AND deleted_at IS NULL + RETURNING id, username, email, password_hash, created_at, password_changed_at, deleted_at`, + [id], + ) + return rows[0] ?? null + } } diff --git a/api/src/users/users.controller.ts b/api/src/users/users.controller.ts index 0823eb7..6a0cd3a 100644 --- a/api/src/users/users.controller.ts +++ b/api/src/users/users.controller.ts @@ -1,6 +1,7 @@ import { Body, Controller, + Delete, Get, HttpCode, HttpStatus, @@ -9,17 +10,21 @@ import { Req, UseGuards, } from "@nestjs/common" -import { Throttle } from "@nestjs/throttler" import { + ApiNoContentResponse, ApiOkResponse, ApiOperation, ApiTags, } from "@nestjs/swagger" -import type { Request } from "express" -import { AuthGuard } from "../common/guards/auth.guard" +import { Throttle } from "@nestjs/throttler" + + import { ChangePasswordDto } from "./dto/change-password.dto" import { UpdateProfileDto } from "./dto/update-profile.dto" import { ProfileResponse, UsersService } from "./users.service" +import { AuthGuard } from "../common/guards/auth.guard" + +import type { Request } from "express" @ApiTags("users") @UseGuards(AuthGuard) @@ -88,4 +93,39 @@ export class UsersController { req.header("authorization") ?? "", ) } + + @Delete("me") + @HttpCode(HttpStatus.NO_CONTENT) + @Throttle({ default: { limit: 3, ttl: 60000 } }) + @ApiOperation({ + summary: "Delete user account", + description: + "Soft-deletes the authenticated user's account. " + + "Sets deleted_at on the user row and revokes the current token. " + + "Audit logs and other historical data are preserved with the " + + "user_id set to NULL.", + }) + @ApiNoContentResponse({ + description: "Account soft-deleted successfully.", + }) + async deleteAccount(@Req() req: Request): Promise { + const { userId } = (req as Request & { auth: { userId: number } }).auth + const ip = this.extractClientIp(req) + await this.usersService.deleteAccount( + userId, + req.header("authorization") ?? "", + ip, + ) + } + + private extractClientIp(req: Request): string { + const xForwardedFor = req.headers["x-forwarded-for"] + if (typeof xForwardedFor === "string") { + return xForwardedFor.split(",")[0].trim() + } + if (Array.isArray(xForwardedFor)) { + return xForwardedFor[0] + } + return req.ip ?? "unknown" + } } diff --git a/api/src/users/users.service.ts b/api/src/users/users.service.ts index 6d52027..212d05b 100644 --- a/api/src/users/users.service.ts +++ b/api/src/users/users.service.ts @@ -1,14 +1,17 @@ import { ConflictException, Injectable, + NotFoundException, UnauthorizedException, } from "@nestjs/common" import { JwtService } from "@nestjs/jwt" import * as bcrypt from "bcrypt" + +import { AuditAction } from "../audit/audit-action.enum" +import { AuditService } from "../audit/audit.service" import { SafeUser, toSafeUser } from "../auth/auth.service" import { TokenDenylistService } from "../auth/token-denylist.service" import { User, UsersRepository } from "../auth/users.repository" -import { AuditService } from "../audit/audit.service" import { ChangePasswordDto } from "./dto/change-password.dto" import { UpdateProfileDto } from "./dto/update-profile.dto" @@ -119,6 +122,37 @@ export class UsersService { }) } + /** + * Soft-delete the authenticated user's account (issue #344). + * Revokes their current token and logs the deletion to the audit trail. + */ + async deleteAccount( + userId: number, + authorizationHeader: string, + ip: string, + ): Promise { + const user = await this.usersRepository.findById(userId) + if (!user) { + throw new UnauthorizedException("user not found") + } + + const deleted = await this.usersRepository.softDelete(userId) + if (!deleted) { + throw new NotFoundException("user not found") + } + + // Revoke the current access token so it cannot be reused + const token = this.extractBearerToken(authorizationHeader) + await this.tokenDenylistService.revoke(token, 3600) + + await this.auditService.log( + userId, + AuditAction.USER_DELETED, + { email: user.email, username: user.username }, + ip, + ) + } + private extractBearerToken(header: string): string { const match = header.trim().match(/^Bearer\s+(.+)$/i) if (!match) { diff --git a/database/audit_logs.sql b/database/audit_logs.sql index a599e7a..5ae286a 100644 --- a/database/audit_logs.sql +++ b/database/audit_logs.sql @@ -1,6 +1,6 @@ CREATE TABLE IF NOT EXISTS audit_logs ( id SERIAL PRIMARY KEY, - user_id INTEGER REFERENCES users(id), + user_id INTEGER REFERENCES users(id) ON DELETE SET NULL, action VARCHAR(100) NOT NULL, ip VARCHAR(45), created_at TIMESTAMPTZ DEFAULT NOW() diff --git a/database/migrations/2026072901_add_users_soft_delete.down.sql b/database/migrations/2026072901_add_users_soft_delete.down.sql new file mode 100644 index 0000000..b35ee07 --- /dev/null +++ b/database/migrations/2026072901_add_users_soft_delete.down.sql @@ -0,0 +1,22 @@ +-- Migration: 2026072901_add_users_soft_delete (DOWN) +-- +-- Rollback: revert soft-delete column and restore the original FK. + +BEGIN; + +-- 1. Drop the partial index +DROP INDEX IF EXISTS idx_users_active; + +-- 2. Remove the soft-delete column +ALTER TABLE users + DROP COLUMN IF EXISTS deleted_at; + +-- 3. Restore the original FK constraint (no ON DELETE SET NULL) +ALTER TABLE audit_logs + DROP CONSTRAINT IF EXISTS audit_logs_user_id_fkey; + +ALTER TABLE audit_logs + ADD CONSTRAINT audit_logs_user_id_fkey + FOREIGN KEY (user_id) REFERENCES users(id); + +COMMIT; diff --git a/database/migrations/2026072901_add_users_soft_delete.up.sql b/database/migrations/2026072901_add_users_soft_delete.up.sql new file mode 100644 index 0000000..1d5fd5a --- /dev/null +++ b/database/migrations/2026072901_add_users_soft_delete.up.sql @@ -0,0 +1,30 @@ +-- Migration: 2026072901_add_users_soft_delete (UP) +-- +-- Soft-delete for the users table (issue #344): +-- 1. Drop the existing audit_logs.user_id FK constraint and recreate it +-- with ON DELETE SET NULL so deleting a user does not cascade-fail. +-- 2. Add a deleted_at TIMESTAMPTZ column to the users table so the +-- application can soft-delete instead of issuing a hard DELETE. +-- 3. Add an index on deleted_at so WHERE deleted_at IS NULL queries +-- are efficient (partial index, nulls are excluded from the index). + +BEGIN; + +-- 1. Fix the audit_logs FK constraint +ALTER TABLE audit_logs + DROP CONSTRAINT IF EXISTS audit_logs_user_id_fkey; + +ALTER TABLE audit_logs + ADD CONSTRAINT audit_logs_user_id_fkey + FOREIGN KEY (user_id) REFERENCES users(id) ON DELETE SET NULL; + +-- 2. Add soft-delete column to users +ALTER TABLE users + ADD COLUMN IF NOT EXISTS deleted_at TIMESTAMPTZ; + +-- 3. Partial index for active-user queries +CREATE INDEX IF NOT EXISTS idx_users_active + ON users(id) + WHERE deleted_at IS NULL; + +COMMIT; diff --git a/database/schema.sql b/database/schema.sql index c157f95..7a9b6a4 100644 --- a/database/schema.sql +++ b/database/schema.sql @@ -6,9 +6,12 @@ CREATE TABLE IF NOT EXISTS users ( username VARCHAR(255) UNIQUE NOT NULL, email VARCHAR(255) UNIQUE NOT NULL, password_hash VARCHAR(255) NOT NULL, - created_at TIMESTAMPTZ DEFAULT NOW() + created_at TIMESTAMPTZ DEFAULT NOW(), + deleted_at TIMESTAMPTZ ); +CREATE INDEX IF NOT EXISTS idx_users_active ON users(id) WHERE deleted_at IS NULL; + -- Streams table CREATE TABLE IF NOT EXISTS streams ( id SERIAL PRIMARY KEY, @@ -184,10 +187,10 @@ CREATE INDEX IF NOT EXISTS idx_webhook_deliveries_pending_next_attempt CREATE TABLE IF NOT EXISTS audit_logs ( id SERIAL PRIMARY KEY, - user_id INTEGER REFERENCES users(id), + user_id INTEGER REFERENCES users(id) ON DELETE SET NULL, action VARCHAR(100) NOT NULL, ip VARCHAR(45), - created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP + created_at TIMESTAMPTZ DEFAULT NOW() ); CREATE INDEX IF NOT EXISTS idx_audit_logs_user_id ON audit_logs(user_id); diff --git a/k8s/README.md b/k8s/README.md index 8bfbb97..bcaab53 100644 --- a/k8s/README.md +++ b/k8s/README.md @@ -381,6 +381,97 @@ kubectl -n xstreamroll exec deploy/processing -- wget -q -O - http://localhost:3 curl -vI https://xstreamroll.example.com/api/health ``` +## Monitoring (issue #349) + +Example Prometheus alerting rules and a Grafana dashboard are shipped in the +`monitoring/` directory at the repo root. Operators can import these into +their observability stack to get production-ready alerting and dashboards +without starting from scratch. + +### Prometheus alerting rules + +`monitoring/prometheus-rules.yaml` ships Prometheus alerting rules covering: + +| Alert | Metric(s) | Threshold | +| ---------------------- | -------------------------------------- | ------------------------------------ | +| HighErrorRate | `http_requests_total` (5xx) | > 5% of all requests over 5m | +| HighAPILatency | `http_request_duration_seconds` (p99) | > 1s over 5m | +| DatabasePoolExhaustion | `db_pool_waiting_requests` | > 0 for 1m | +| WorkerRestart | `xstreamroll_uptime_seconds` (reset) | any reset within 5m | +| HighWorkerErrorRate | `xstreamroll_errors_total` | > 1 error/s over 5m | +| QueueDepthHigh | `xstreamroll_queue_depth` | > 32 000 for 10m | +| LockRenewalFailures | `xstreamroll_lock_renewals_failed` | > 0.1/s over 5m | +| WebSocketConnectionDrop| `websocket_active_connections` | < 50% of historical peak for 10m | + +Every alert ships at `severity: warning` by default so operators can tune +thresholds before upgrading to `critical`. + +#### Installing the alert rules + +```bash +# Validate the rules with promtool (requires promtool in your PATH). +promtool check rules monitoring/prometheus-rules.yaml + +# Apply to a Prometheus operator via a ConfigMap. +kubectl -n monitoring create configmap xstreamroll-rules \ + --from-file=monitoring/prometheus-rules.yaml \ + --dry-run=client -o yaml | kubectl apply -f - + +# If using kube-prometheus-stack, create a PrometheusRule CR instead. +kubectl apply -f monitoring/prometheus-rules.yaml +``` + +#### Scraping the processing worker from Prometheus + +The worker exposes metrics on port `:3002` at `/metrics`. The accompanying +`k8s/70-network-policies.yaml` already allows any pod in the `xstreamroll` +namespace to reach that port. If your Prometheus instance runs in a separate +`monitoring` namespace, add a `namespaceSelector` to the NetworkPolicy: + +```yaml +# In k8s/70-network-policies.yaml, add to the allow-processing-metrics rule: + ingress: + - from: + - namespaceSelector: + matchLabels: + kubernetes.io/metadata.name: monitoring +``` + +Alternatively, annotate the worker pod template so Prometheus discovers it: + +```yaml +# In k8s/40-processing.yaml, under spec.template.metadata: + annotations: + prometheus.io/scrape: "true" + prometheus.io/port: "3002" + prometheus.io/path: "/metrics/prometheus" +``` + +### Grafana dashboard + +`monitoring/grafana-dashboard.json` is a ready-to-import dashboard with +panels covering: + +- **API Overview** — request rate by HTTP method, latency histogram (p50/p95/p99) +- **WebSocket Connections** — active connections, connection rate +- **Processing Worker** — queue depth, active session locks, messages + processed rate, throughput vs errors, uptime +- **Database Connection Pool** — active/idle/waiting connections + +#### Importing the dashboard + +1. Open Grafana → Dashboards → Import. +2. Paste the contents of `monitoring/grafana-dashboard.json` or upload + the file directly. +3. Select your Prometheus data source when prompted. +4. Save — the dashboard auto-refreshes every 30s. + +#### Customising for your deployment + +The dashboard uses the `DS_PROMETHEUS` template variable — you can swap +this in Grafana without editing the JSON. All panels query `sum()` +aggregations so they work correctly whether you have one worker pod or ten. + ## Release process Images are published by `.github/workflows/release.yml` and are restricted diff --git a/monitoring/grafana-dashboard.json b/monitoring/grafana-dashboard.json new file mode 100644 index 0000000..4948e6e --- /dev/null +++ b/monitoring/grafana-dashboard.json @@ -0,0 +1,402 @@ +{ + "annotations": { + "list": [ + { + "builtIn": 1, + "datasource": { "type": "grafana", "uid": "-- Grafana --" }, + "enable": true, + "hide": true, + "iconColor": "rgba(0, 211, 255, 1)", + "name": "Annotations & Alerts", + "type": "dashboard" + } + ] + }, + "editable": true, + "fiscalYearStartMonth": 0, + "graphTooltip": 1, + "id": null, + "links": [], + "panels": [ + { + "collapsed": false, + "gridPos": { "h": 1, "w": 24, "x": 0, "y": 0 }, + "id": 1, + "panels": [], + "title": "API Overview", + "type": "row" + }, + { + "datasource": { "type": "prometheus", "uid": "${DS_PROMETHEUS}" }, + "fieldConfig": { + "defaults": { + "color": { "mode": "palette-classic" }, + "custom": { "axisBorderShow": false, "axisCenteredZero": false, "axisColorMode": "text", "axisPlacement": "auto", "barAlignment": 0, "drawStyle": "line", "fillOpacity": 15, "gradientMode": "none", "hideFrom": { "legend": false, "tooltip": false, "viz": false }, "insertNulls": false, "lineInterpolation": "linear", "lineWidth": 2, "pointSize": 5, "scaleDistribution": { "type": "linear" }, "showPoints": "never", "spanNulls": false, "stacking": { "group": "A", "mode": "none" }, "thresholdsStyle": { "mode": "off" } }, + "mappings": [], + "thresholds": { "mode": "absolute", "steps": [{ "color": "green", "value": null }, { "color": "red", "value": 80 }] }, + "unit": "reqps" + }, + "overrides": [] + }, + "gridPos": { "h": 8, "w": 12, "x": 0, "y": 1 }, + "id": 2, + "options": { "legend": { "calcs": ["mean", "max"], "displayMode": "table", "placement": "bottom", "showLegend": true }, "tooltip": { "mode": "multi", "sort": "desc" } }, + "pluginVersion": "10.0.0", + "targets": [ + { + "datasource": { "type": "prometheus", "uid": "${DS_PROMETHEUS}" }, + "editorMode": "code", + "expr": "sum(rate(http_requests_total[1m])) by (method)", + "legendFormat": "{{method}}", + "range": true, + "refId": "A" + } + ], + "title": "HTTP Request Rate", + "type": "timeseries" + }, + { + "datasource": { "type": "prometheus", "uid": "${DS_PROMETHEUS}" }, + "fieldConfig": { + "defaults": { + "color": { "mode": "palette-classic" }, + "custom": { "axisBorderShow": false, "axisCenteredZero": false, "axisColorMode": "text", "axisPlacement": "auto", "barAlignment": 0, "drawStyle": "line", "fillOpacity": 15, "gradientMode": "none", "hideFrom": { "legend": false, "tooltip": false, "viz": false }, "insertNulls": false, "lineInterpolation": "linear", "lineWidth": 2, "pointSize": 5, "scaleDistribution": { "type": "linear" }, "showPoints": "never", "spanNulls": false, "stacking": { "group": "A", "mode": "none" }, "thresholdsStyle": { "mode": "off" } }, + "mappings": [], + "thresholds": { "mode": "absolute", "steps": [{ "color": "green", "value": null }, { "color": "red", "value": 80 }] }, + "unit": "s" + }, + "overrides": [ + { + "matcher": { "id": "byName", "options": "p50" }, + "properties": [{ "id": "custom.lineStyle", "value": { "dash": [10, 5], "fill": "dash" } }] + }, + { + "matcher": { "id": "byName", "options": "p99" }, + "properties": [{ "id": "color", "value": { "fixedColor": "red", "mode": "fixed" } }] + } + ] + }, + "gridPos": { "h": 8, "w": 12, "x": 12, "y": 1 }, + "id": 3, + "options": { "legend": { "calcs": ["mean", "max"], "displayMode": "table", "placement": "bottom", "showLegend": true }, "tooltip": { "mode": "multi", "sort": "desc" } }, + "targets": [ + { + "datasource": { "type": "prometheus", "uid": "${DS_PROMETHEUS}" }, + "editorMode": "code", + "expr": "histogram_quantile(0.50, sum(rate(http_request_duration_seconds_bucket[1m])) by (le))", + "legendFormat": "p50", + "refId": "A" + }, + { + "datasource": { "type": "prometheus", "uid": "${DS_PROMETHEUS}" }, + "editorMode": "code", + "expr": "histogram_quantile(0.95, sum(rate(http_request_duration_seconds_bucket[1m])) by (le))", + "legendFormat": "p95", + "refId": "B" + }, + { + "datasource": { "type": "prometheus", "uid": "${DS_PROMETHEUS}" }, + "editorMode": "code", + "expr": "histogram_quantile(0.99, sum(rate(http_request_duration_seconds_bucket[1m])) by (le))", + "legendFormat": "p99", + "refId": "C" + } + ], + "title": "API Latency (histogram)", + "type": "timeseries" + }, + { + "collapsed": false, + "gridPos": { "h": 1, "w": 24, "x": 0, "y": 9 }, + "id": 10, + "panels": [], + "title": "WebSocket Connections", + "type": "row" + }, + { + "datasource": { "type": "prometheus", "uid": "${DS_PROMETHEUS}" }, + "fieldConfig": { + "defaults": { + "color": { "mode": "palette-classic" }, + "custom": { "axisBorderShow": false, "axisCenteredZero": false, "axisColorMode": "text", "axisPlacement": "auto", "barAlignment": 0, "drawStyle": "line", "fillOpacity": 15, "gradientMode": "none", "hideFrom": { "legend": false, "tooltip": false, "viz": false }, "insertNulls": false, "lineInterpolation": "linear", "lineWidth": 2, "pointSize": 5, "scaleDistribution": { "type": "linear" }, "showPoints": "never", "spanNulls": false, "stacking": { "group": "A", "mode": "none" }, "thresholdsStyle": { "mode": "off" } }, + "mappings": [], + "thresholds": { "mode": "absolute", "steps": [{ "color": "green", "value": null }, { "color": "red", "value": 80 }] }, + "unit": "short" + }, + "overrides": [] + }, + "gridPos": { "h": 8, "w": 12, "x": 0, "y": 10 }, + "id": 11, + "options": { "legend": { "calcs": ["last"], "displayMode": "table", "placement": "bottom", "showLegend": true }, "tooltip": { "mode": "multi", "sort": "desc" } }, + "targets": [ + { + "datasource": { "type": "prometheus", "uid": "${DS_PROMETHEUS}" }, + "editorMode": "code", + "expr": "websocket_active_connections", + "legendFormat": "active connections", + "refId": "A" + } + ], + "title": "Active WebSocket Connections", + "type": "timeseries" + }, + { + "datasource": { "type": "prometheus", "uid": "${DS_PROMETHEUS}" }, + "fieldConfig": { + "defaults": { + "color": { "mode": "palette-classic" }, + "custom": { "axisBorderShow": false, "axisCenteredZero": false, "axisColorMode": "text", "axisPlacement": "auto", "barAlignment": 0, "drawStyle": "line", "fillOpacity": 15, "gradientMode": "none", "hideFrom": { "legend": false, "tooltip": false, "viz": false }, "insertNulls": false, "lineInterpolation": "linear", "lineWidth": 2, "pointSize": 5, "scaleDistribution": { "type": "linear" }, "showPoints": "never", "spanNulls": false, "stacking": { "group": "A", "mode": "none" }, "thresholdsStyle": { "mode": "off" } }, + "mappings": [], + "thresholds": { "mode": "absolute", "steps": [{ "color": "green", "value": null }, { "color": "red", "value": 80 }] }, + "unit": "short" + }, + "overrides": [] + }, + "gridPos": { "h": 8, "w": 12, "x": 12, "y": 10 }, + "id": 12, + "options": { "legend": { "calcs": ["last"], "displayMode": "table", "placement": "bottom", "showLegend": true }, "tooltip": { "mode": "multi", "sort": "desc" } }, + "targets": [ + { + "datasource": { "type": "prometheus", "uid": "${DS_PROMETHEUS}" }, + "editorMode": "code", + "expr": "rate(websocket_connections_total[5m])", + "legendFormat": "connection rate", + "refId": "A" + } + ], + "title": "WebSocket Connection Rate (5m)", + "type": "timeseries" + }, + { + "collapsed": false, + "gridPos": { "h": 1, "w": 24, "x": 0, "y": 18 }, + "id": 20, + "panels": [], + "title": "Processing Worker", + "type": "row" + }, + { + "datasource": { "type": "prometheus", "uid": "${DS_PROMETHEUS}" }, + "fieldConfig": { + "defaults": { + "color": { "mode": "palette-classic" }, + "custom": { "axisBorderShow": false, "axisCenteredZero": false, "axisColorMode": "text", "axisPlacement": "auto", "barAlignment": 0, "drawStyle": "line", "fillOpacity": 15, "gradientMode": "none", "hideFrom": { "legend": false, "tooltip": false, "viz": false }, "insertNulls": false, "lineInterpolation": "linear", "lineWidth": 2, "pointSize": 5, "scaleDistribution": { "type": "linear" }, "showPoints": "never", "spanNulls": false, "stacking": { "group": "A", "mode": "none" }, "thresholdsStyle": { "mode": "off" } }, + "mappings": [], + "thresholds": { "mode": "absolute", "steps": [{ "color": "green", "value": null }, { "color": "red", "value": 80 }] }, + "unit": "short" + }, + "overrides": [] + }, + "gridPos": { "h": 8, "w": 8, "x": 0, "y": 19 }, + "id": 21, + "options": { "legend": { "calcs": ["last", "mean", "max"], "displayMode": "table", "placement": "bottom", "showLegend": true }, "tooltip": { "mode": "multi", "sort": "desc" } }, + "targets": [ + { + "datasource": { "type": "prometheus", "uid": "${DS_PROMETHEUS}" }, + "editorMode": "code", + "expr": "xstreamroll_queue_depth", + "legendFormat": "queue depth", + "refId": "A" + } + ], + "title": "Worker Queue Depth", + "type": "timeseries" + }, + { + "datasource": { "type": "prometheus", "uid": "${DS_PROMETHEUS}" }, + "fieldConfig": { + "defaults": { + "color": { "mode": "palette-classic" }, + "custom": { "axisBorderShow": false, "axisCenteredZero": false, "axisColorMode": "text", "axisPlacement": "auto", "barAlignment": 0, "drawStyle": "line", "fillOpacity": 15, "gradientMode": "none", "hideFrom": { "legend": false, "tooltip": false, "viz": false }, "insertNulls": false, "lineInterpolation": "linear", "lineWidth": 2, "pointSize": 5, "scaleDistribution": { "type": "linear" }, "showPoints": "never", "spanNulls": false, "stacking": { "group": "A", "mode": "none" }, "thresholdsStyle": { "mode": "off" } }, + "mappings": [], + "thresholds": { "mode": "absolute", "steps": [{ "color": "green", "value": null }, { "color": "red", "value": 80 }] }, + "unit": "short" + }, + "overrides": [] + }, + "gridPos": { "h": 8, "w": 8, "x": 8, "y": 19 }, + "id": 22, + "options": { "legend": { "calcs": ["last"], "displayMode": "table", "placement": "bottom", "showLegend": true }, "tooltip": { "mode": "multi", "sort": "desc" } }, + "targets": [ + { + "datasource": { "type": "prometheus", "uid": "${DS_PROMETHEUS}" }, + "editorMode": "code", + "expr": "xstreamroll_active_locks", + "legendFormat": "active locks", + "refId": "A" + } + ], + "title": "Active Session Locks", + "type": "timeseries" + }, + { + "datasource": { "type": "prometheus", "uid": "${DS_PROMETHEUS}" }, + "fieldConfig": { + "defaults": { + "color": { "mode": "palette-classic" }, + "custom": { "axisBorderShow": false, "axisCenteredZero": false, "axisColorMode": "text", "axisPlacement": "auto", "barAlignment": 0, "drawStyle": "line", "fillOpacity": 15, "gradientMode": "none", "hideFrom": { "legend": false, "tooltip": false, "viz": false }, "insertNulls": false, "lineInterpolation": "linear", "lineWidth": 2, "pointSize": 5, "scaleDistribution": { "type": "linear" }, "showPoints": "never", "spanNulls": false, "stacking": { "group": "A", "mode": "none" }, "thresholdsStyle": { "mode": "off" } }, + "mappings": [], + "thresholds": { "mode": "absolute", "steps": [{ "color": "green", "value": null }, { "color": "red", "value": 80 }] }, + "unit": "reqps" + }, + "overrides": [] + }, + "gridPos": { "h": 8, "w": 8, "x": 16, "y": 19 }, + "id": 23, + "options": { "legend": { "calcs": ["last", "mean"], "displayMode": "table", "placement": "bottom", "showLegend": true }, "tooltip": { "mode": "multi", "sort": "desc" } }, + "targets": [ + { + "datasource": { "type": "prometheus", "uid": "${DS_PROMETHEUS}" }, + "editorMode": "code", + "expr": "rate(xstreamroll_messages_processed_total[1m])", + "legendFormat": "processed/s", + "refId": "A" + } + ], + "title": "Messages Processed Rate", + "type": "timeseries" + }, + { + "datasource": { "type": "prometheus", "uid": "${DS_PROMETHEUS}" }, + "fieldConfig": { + "defaults": { + "color": { "mode": "palette-classic" }, + "custom": { "axisBorderShow": false, "axisCenteredZero": false, "axisColorMode": "text", "axisPlacement": "auto", "barAlignment": 0, "drawStyle": "line", "fillOpacity": 15, "gradientMode": "none", "hideFrom": { "legend": false, "tooltip": false, "viz": false }, "insertNulls": false, "lineInterpolation": "linear", "lineWidth": 2, "pointSize": 5, "scaleDistribution": { "type": "linear" }, "showPoints": "never", "spanNulls": false, "stacking": { "group": "A", "mode": "none" }, "thresholdsStyle": { "mode": "off" } }, + "mappings": [], + "thresholds": { "mode": "absolute", "steps": [{ "color": "green", "value": null }, { "color": "red", "value": 80 }] }, + "unit": "reqps" + }, + "overrides": [ + { + "matcher": { "id": "byName", "options": "error rate" }, + "properties": [{ "id": "color", "value": { "fixedColor": "red", "mode": "fixed" } }] + } + ] + }, + "gridPos": { "h": 8, "w": 12, "x": 0, "y": 27 }, + "id": 24, + "options": { "legend": { "calcs": ["last", "mean"], "displayMode": "table", "placement": "bottom", "showLegend": true }, "tooltip": { "mode": "multi", "sort": "desc" } }, + "targets": [ + { + "datasource": { "type": "prometheus", "uid": "${DS_PROMETHEUS}" }, + "editorMode": "code", + "expr": "rate(xstreamroll_messages_processed_total[1m])", + "legendFormat": "processed/s", + "refId": "A" + }, + { + "datasource": { "type": "prometheus", "uid": "${DS_PROMETHEUS}" }, + "editorMode": "code", + "expr": "rate(xstreamroll_errors_total[1m])", + "legendFormat": "error rate", + "refId": "B" + } + ], + "title": "Worker Throughput vs Errors", + "type": "timeseries" + }, + { + "datasource": { "type": "prometheus", "uid": "${DS_PROMETHEUS}" }, + "fieldConfig": { + "defaults": { + "color": { "mode": "palette-classic" }, + "custom": { "axisBorderShow": false, "axisCenteredZero": false, "axisColorMode": "text", "axisPlacement": "auto", "barAlignment": 0, "drawStyle": "line", "fillOpacity": 15, "gradientMode": "none", "hideFrom": { "legend": false, "tooltip": false, "viz": false }, "insertNulls": false, "lineInterpolation": "linear", "lineWidth": 2, "pointSize": 5, "scaleDistribution": { "type": "linear" }, "showPoints": "never", "spanNulls": false, "stacking": { "group": "A", "mode": "none" }, "thresholdsStyle": { "mode": "off" } }, + "mappings": [], + "thresholds": { "mode": "absolute", "steps": [{ "color": "green", "value": null }, { "color": "red", "value": 80 }] }, + "unit": "s" + }, + "overrides": [] + }, + "gridPos": { "h": 8, "w": 12, "x": 12, "y": 27 }, + "id": 25, + "options": { "legend": { "calcs": ["last", "mean"], "displayMode": "table", "placement": "bottom", "showLegend": true }, "tooltip": { "mode": "multi", "sort": "desc" } }, + "targets": [ + { + "datasource": { "type": "prometheus", "uid": "${DS_PROMETHEUS}" }, + "editorMode": "code", + "expr": "xstreamroll_uptime_seconds", + "legendFormat": "uptime", + "refId": "A" + } + ], + "title": "Worker Uptime", + "type": "timeseries" + }, + { + "collapsed": false, + "gridPos": { "h": 1, "w": 24, "x": 0, "y": 35 }, + "id": 30, + "panels": [], + "title": "Database Connection Pool", + "type": "row" + }, + { + "datasource": { "type": "prometheus", "uid": "${DS_PROMETHEUS}" }, + "fieldConfig": { + "defaults": { + "color": { "mode": "palette-classic" }, + "custom": { "axisBorderShow": false, "axisCenteredZero": false, "axisColorMode": "text", "axisPlacement": "auto", "barAlignment": 0, "drawStyle": "line", "fillOpacity": 15, "gradientMode": "none", "hideFrom": { "legend": false, "tooltip": false, "viz": false }, "insertNulls": false, "lineInterpolation": "stepAfter", "lineWidth": 2, "pointSize": 5, "scaleDistribution": { "type": "linear" }, "showPoints": "never", "spanNulls": false, "stacking": { "group": "A", "mode": "normal" }, "thresholdsStyle": { "mode": "off" } }, + "mappings": [], + "thresholds": { "mode": "absolute", "steps": [{ "color": "green", "value": null }, { "color": "red", "value": 80 }] }, + "unit": "short" + }, + "overrides": [] + }, + "gridPos": { "h": 8, "w": 12, "x": 0, "y": 36 }, + "id": 31, + "options": { "legend": { "calcs": ["last"], "displayMode": "table", "placement": "bottom", "showLegend": true }, "tooltip": { "mode": "multi", "sort": "desc" } }, + "targets": [ + { + "datasource": { "type": "prometheus", "uid": "${DS_PROMETHEUS}" }, + "editorMode": "code", + "expr": "db_pool_active_connections", + "legendFormat": "active", + "refId": "A" + }, + { + "datasource": { "type": "prometheus", "uid": "${DS_PROMETHEUS}" }, + "editorMode": "code", + "expr": "db_pool_idle_connections", + "legendFormat": "idle", + "refId": "B" + }, + { + "datasource": { "type": "prometheus", "uid": "${DS_PROMETHEUS}" }, + "editorMode": "code", + "expr": "db_pool_waiting_requests", + "legendFormat": "waiting", + "refId": "C" + } + ], + "title": "DB Connection Pool", + "type": "timeseries" + } + ], + "refresh": "30s", + "schemaVersion": 38, + "tags": ["xstreamroll", "api", "worker"], + "templating": { + "list": [ + { + "current": { "selected": false, "text": "Prometheus", "value": "prometheus" }, + "hide": 0, + "includeAll": false, + "label": "Data source", + "multi": false, + "name": "DS_PROMETHEUS", + "options": [], + "query": "prometheus", + "refresh": 1, + "regex": "", + "skipUrlSync": false, + "type": "datasource" + } + ] + }, + "time": { "from": "now-1h", "to": "now" }, + "timepicker": {}, + "timezone": "browser", + "title": "XStreamRoll", + "uid": "xstreamroll", + "version": 1 +} diff --git a/monitoring/prometheus-rules.yaml b/monitoring/prometheus-rules.yaml new file mode 100644 index 0000000..c324437 --- /dev/null +++ b/monitoring/prometheus-rules.yaml @@ -0,0 +1,153 @@ +# ────────────────────────────────────────────────────────────────────────────── +# Prometheus alerting rules — XStreamRoll (issue #349) +# +# These rules target the metrics emitted by: +# • api — http_*, websocket_*, db_pool_* (port 3001 /metrics) +# • processing — xstreamroll_* (port 3002 /metrics) +# +# Every alert ships at `severity: warning` by default so operators can +# tune them before upgrading to `critical` for their environment. +# Group evaluation interval inherits the Prometheus global default; +# override with `interval: 30s` per group if your config requires +# faster evaluation. +# ────────────────────────────────────────────────────────────────────────────── + +groups: + # ── API alerts ─────────────────────────────────────────────────── + - name: xstreamroll-api + rules: + # High error rate on HTTP endpoints (5xx) + - alert: HighErrorRate + expr: | + ( + sum(rate(http_requests_total{status_code=~"5.."}[5m])) + / + sum(rate(http_requests_total[5m])) + ) > 0.05 + for: 5m + labels: + severity: warning + component: api + annotations: + summary: "High HTTP 5xx error rate on API" + description: >- + The API 5xx error rate is {{ $value | humanizePercentage }} + over the last 5 minutes (threshold: 5%). Check the API pod + logs (`kubectl -n xstreamroll logs deploy/api`) and verify + Postgres connectivity. + + # API request latency p99 > 1s + - alert: HighAPILatency + expr: | + histogram_quantile( + 0.99, + sum(rate(http_request_duration_seconds_bucket[5m])) by (le, path) + ) > 1 + for: 5m + labels: + severity: warning + component: api + annotations: + summary: "API p99 latency exceeds 1s" + description: >- + The 99th percentile HTTP request duration for path + `{{ $labels.path }}` is {{ $value }}s over the last 5 minutes. + Investigate slow downstream calls or database queries. + + # Database connection pool exhaustion + - alert: DatabasePoolExhaustion + expr: | + db_pool_waiting_requests > 0 + for: 1m + labels: + severity: warning + component: api + annotations: + summary: "PostgreSQL connection pool has waiting requests" + description: >- + {{ $value }} requests are queued waiting for a database + connection. Consider increasing the pool size or scaling + the API horizontally. + + # ── Processing worker alerts ───────────────────────────────────── + - name: xstreamroll-processing + rules: + # Worker restarts (detected via counter reset on the messages counter) + - alert: WorkerRestart + expr: | + resets(xstreamroll_messages_processed_total[5m]) > 0 + labels: + severity: warning + component: processing + annotations: + summary: "Processing worker restarted" + description: >- + The xstreamroll_messages_processed_total counter reset + within the last 5 minutes, indicating a worker pod restarted. + Check `kubectl -n xstreamroll describe pod -l app=processing` + for events and review the previous pod's logs. + + # High worker error rate + - alert: HighWorkerErrorRate + expr: | + rate(xstreamroll_errors_total[5m]) > 1 + for: 5m + labels: + severity: warning + component: processing + annotations: + summary: "Processing worker error rate is elevated" + description: >- + The worker is emitting {{ $value | humanize }} errors per second + over the last 5 minutes. Check worker logs for dead-letter + events and API connectivity. + + # Queue depth approaching high-watermark + - alert: QueueDepthHigh + expr: | + xstreamroll_queue_depth > 32000 + for: 10m + labels: + severity: warning + component: processing + annotations: + summary: "Processing worker queue depth is high" + description: >- + The total queue depth across all sessions is {{ $value }}, + approaching the default high-watermark (MAX_CONCURRENT_SESSIONS × + MAX_QUEUE_DEPTH = 32 × 1000 = 32000). The worker will start + shedding events. Consider scaling horizontally or increasing + the poll interval. + + # Lock renewals failing frequently + - alert: LockRenewalFailures + expr: | + rate(xstreamroll_lock_renewals_failed[5m]) > 0.1 + for: 5m + labels: + severity: warning + component: processing + annotations: + summary: "Lock renewals are failing" + description: >- + Lock renewal failure rate is {{ $value }} per second. + The lock backend (Postgres) may be slow or unreachable. + Sessions will lose their locks and other workers may + claim them, causing processing gaps. + + # ── WebSocket alerts ───────────────────────────────────────────── + - name: xstreamroll-websocket + rules: + - alert: WebSocketConnectionDrop + expr: | + (websocket_active_connections / websocket_connections_total) < 0.5 + for: 10m + labels: + severity: warning + component: api + annotations: + summary: "WebSocket active connections dropped significantly" + description: >- + Active WebSocket connections ({{ $value | humanizePercentage }} of + total) have dropped below 50% of historical peak. This may indicate + a client-side connectivity issue or a gateways pod restart. diff --git a/xstreamroll-processing/.env.example b/xstreamroll-processing/.env.example index 99f3206..cddd54a 100644 --- a/xstreamroll-processing/.env.example +++ b/xstreamroll-processing/.env.example @@ -144,3 +144,20 @@ OTEL_SERVICE_NAME=xstreamroll-processing # Maximum number of per-event publish retries before dead-lettering. PROCESSING_PUBLISH_MAX_RETRIES=3 + +# ────────────────────────────────────────────────────────────────────────────── +# Graceful shutdown timeouts (issue #342) +# +# SHUTDOWN_TIMEOUT_MS is the global hard deadline for the entire shutdown +# sequence. Individual shutdown hooks can override this with their own +# `timeoutMs` field. When a hook exceeds its timeout, a warning is logged +# and shutdown continues to the next step — no single stuck hook can +# block the process from exiting. +# +# SESSION_DRAIN_TIMEOUT_MS is the per-session deadline enforced by +# SessionRegistry.drainAll(). When a session's stop() promise hangs, +# the timeout fires a warning and the next session is drained. +# ────────────────────────────────────────────────────────────────────────────── + +SHUTDOWN_TIMEOUT_MS=15000 +SESSION_DRAIN_TIMEOUT_MS=5000 diff --git a/xstreamroll-processing/__tests__/lifecycle.shutdown.test.ts b/xstreamroll-processing/__tests__/lifecycle.shutdown.test.ts index f8fcced..e565283 100644 --- a/xstreamroll-processing/__tests__/lifecycle.shutdown.test.ts +++ b/xstreamroll-processing/__tests__/lifecycle.shutdown.test.ts @@ -96,4 +96,127 @@ describe("GracefulShutdown", () => { await gs.requestShutdown("SIGTERM" as ShutdownReason) expect(gs.getState()).toBe("done") }) + + // -- Per-step timeout (issue #342) ------------------------------------ + + it("logs a warning and continues when a hook exceeds its timeout", async () => { + const calls: string[] = [] + const exit = jest.fn() + const { logger, entries } = makeLogger() + const gs = new GracefulShutdown({ exit, logger, timeoutMs: 50 }) + + gs.register({ + name: "hanging hook", + timeoutMs: 20, + run: async () => { + await new Promise(() => {}) // never resolves + }, + }) + gs.register({ + name: "fast hook", + run: () => { + calls.push("fast") + }, + }) + + await gs.requestShutdown("manual") + + // The fast hook should have run despite the hanging hook + expect(calls).toEqual(["fast"]) + // The hanging hook should have triggered a timeout warning + expect( + entries.some( + (e) => + e.level === "warn" && + (e.args as string[]).some((a) => String(a).includes("timed out")), + ), + ).toBe(true) + // Overall shutdown had an error from the timed-out hook + expect(exit).toHaveBeenCalledWith(1) + }) + + it("uses the per-step timeoutMs override when provided", async () => { + const exit = jest.fn() + const { logger } = makeLogger() + // Global timeout is very short, but the hook has a longer override + const gs = new GracefulShutdown({ exit, logger, timeoutMs: 1000 }) + + gs.register({ + name: "fast but with long override", + timeoutMs: 200, + run: async () => { + await new Promise((r) => setTimeout(r, 5)) + }, + }) + + await gs.requestShutdown("manual") + expect(exit).toHaveBeenCalledWith(0) + }) + + it("falls back to the global timeout when per-step timeoutMs is omitted", async () => { + const exit = jest.fn() + const { logger, entries } = makeLogger() + const gs = new GracefulShutdown({ exit, logger, timeoutMs: 20 }) + + gs.register({ + name: "hanging no-override", + // no timeoutMs — falls back to global 20ms + run: async () => { + await new Promise(() => {}) + }, + }) + gs.register({ + name: "still runs", + run: () => {}, + }) + + await gs.requestShutdown("manual") + expect( + entries.some( + (e) => + e.level === "warn" && + (e.args as string[]).some((a) => String(a).includes("timed out")), + ), + ).toBe(true) + expect(exit).toHaveBeenCalledWith(1) + }) + + it("continues through multiple timed-out hooks", async () => { + const calls: string[] = [] + const exit = jest.fn() + const { logger, entries } = makeLogger() + const gs = new GracefulShutdown({ exit, logger, timeoutMs: 50 }) + + gs.register({ + name: "hang1", + timeoutMs: 10, + run: async () => { + await new Promise(() => {}) + }, + }) + gs.register({ + name: "hang2", + timeoutMs: 10, + run: async () => { + await new Promise(() => {}) + }, + }) + gs.register({ + name: "clean", + run: () => { + calls.push("clean") + }, + }) + + await gs.requestShutdown("manual") + + expect(calls).toEqual(["clean"]) + const warnCount = entries.filter( + (e) => + e.level === "warn" && + (e.args as string[]).some((a) => String(a).includes("timed out")), + ).length + expect(warnCount).toBe(2) + expect(exit).toHaveBeenCalledWith(1) + }) }) diff --git a/xstreamroll-processing/__tests__/session-registry.test.ts b/xstreamroll-processing/__tests__/session-registry.test.ts index c99080b..c646d5d 100644 --- a/xstreamroll-processing/__tests__/session-registry.test.ts +++ b/xstreamroll-processing/__tests__/session-registry.test.ts @@ -135,6 +135,33 @@ describe("SessionRegistry", () => { expect(registry.lockCount()).toBe(0) }) + it("drainAll with timeout resolves even when a session stop hangs (issue #342)", async () => { + const lockManager = new MemoryLockManager({ workerId: "w1", ttlMs: 30_000 }) + const registry = new SessionRegistry( + "w1", + { publish: jest.fn() }, + { maxConcurrentSessions: 4, lockManager }, + ) + + await registry.route(evt("s1")) + const session = registry.get("s1")! + + // Override stop() to hang forever + jest.spyOn(session, "stop").mockImplementation(() => { + return new Promise(() => {}) // never resolves + }) + + // drainAll with a short per-session timeout should still resolve + await registry.drainAll(50) + + // The registry should be cleared despite the hung session + expect(registry.size()).toBe(0) + expect(registry.lockCount()).toBe(0) + + // Restore original to allow cleanup + jest.restoreAllMocks() + }) + it("capacity() reports used vs max", async () => { const registry = makeRegistry(8) await registry.route(evt("s1")) diff --git a/xstreamroll-processing/src/config.ts b/xstreamroll-processing/src/config.ts index b4bb355..363e8f0 100644 --- a/xstreamroll-processing/src/config.ts +++ b/xstreamroll-processing/src/config.ts @@ -62,6 +62,26 @@ const envSchema = z.object({ .url() .optional() .or(z.literal("").transform(() => undefined)), + /** + * Global hard deadline (ms) for the entire shutdown sequence (issue #342). + * When a registered hook exceeds its per-step timeout a warning is logged + * and shutdown continues to the next step. Defaults to 15000 (15s). + */ + SHUTDOWN_TIMEOUT_MS: z + .string() + .default("15000") + .transform((s) => Number(s)) + .pipe(z.number().int().positive()), + /** + * Per-session drain deadline (ms) enforced by SessionRegistry.drainAll() + * (issue #342). When a session's stop() hangs, the timeout fires a warning + * and the next session is drained. Defaults to 5000 (5s). + */ + SESSION_DRAIN_TIMEOUT_MS: z + .string() + .default("5000") + .transform((s) => Number(s)) + .pipe(z.number().int().positive()), }) export type Env = z.infer diff --git a/xstreamroll-processing/src/lifecycle.ts b/xstreamroll-processing/src/lifecycle.ts index 234d5ae..0370f8b 100644 --- a/xstreamroll-processing/src/lifecycle.ts +++ b/xstreamroll-processing/src/lifecycle.ts @@ -25,6 +25,14 @@ export interface ShutdownHook { name: string /** Cleanup routine. Throw to abort the shutdown with a non-zero exit. */ run: (reason: ShutdownReason) => Promise | void + /** + * Per-step timeout in milliseconds (issue #342). + * When set, the hook is raced against this timeout and a + * warning is logged if it is exceeded — shutdown continues + * with the next step. Falls back to the global timeout when + * omitted. + */ + timeoutMs?: number } export interface ShutdownOptions { @@ -38,6 +46,52 @@ export interface ShutdownOptions { const DEFAULT_TIMEOUT_MS = 15_000 +/** + * Run `promise` with a timeout. If the timeout fires first, log + * a warning and return — the underlying promise is intentionally + * allowed to continue (it will be abandoned when the process + * exits). If the promise rejects, the rejection is propagated + * so the caller can log the actual error. + */ +async function raceTimeout( + promise: Promise, + timeoutMs: number, + stepName: string, + logger: Pick, +): Promise { + // Returns true if the promise resolved before the timeout, + // false if the timeout fired. + + if (timeoutMs <= 0) { + // No timeout; just await normally. + await promise + return true + } + + // eslint-disable-next-line prefer-const + let timer: ReturnType | undefined + let timedOut = false + const timeout = new Promise((resolve) => { + timer = setTimeout(() => { + timedOut = true + resolve() + }, timeoutMs) + if (typeof timer?.unref === "function") timer.unref() + }) + + await Promise.race([promise, timeout]) + + if (timer !== undefined) clearTimeout(timer) + + if (timedOut) { + logger.warn( + `[shutdown] ${stepName} timed out after ${timeoutMs}ms — continuing`, + ) + return false + } + return true +} + /** * Singleton-style coordinator. Register hooks during startup, call * {@link GracefulShutdown.install} once, then `await requestShutdown` @@ -114,9 +168,25 @@ export class GracefulShutdown { let hadError = false for (const hook of this.hooks) { + const stepTimeout = hook.timeoutMs ?? this.timeoutMs + try { - await hook.run(reason) - this.logger.log(`[shutdown] ${hook.name} ✓`) + const runPromise = (async (): Promise => { + await hook.run(reason) + })() + + const ok = await raceTimeout( + runPromise, + stepTimeout, + hook.name, + this.logger, + ) + + if (ok) { + this.logger.log(`[shutdown] ${hook.name} ✓`) + } else { + hadError = true + } } catch (err) { hadError = true const message = err instanceof Error ? err.message : String(err) diff --git a/xstreamroll-processing/src/session-registry.ts b/xstreamroll-processing/src/session-registry.ts index 58f34fb..4661651 100644 --- a/xstreamroll-processing/src/session-registry.ts +++ b/xstreamroll-processing/src/session-registry.ts @@ -1,7 +1,7 @@ -import { SessionHandlers, StreamEvent, StreamSession } from "./session" import { LockManager, LockToken } from "./leader-election" import { Logger } from "./logger" import * as metrics from "./metrics" +import { SessionHandlers, StreamEvent, StreamSession } from "./session" export type RouteResult = "enqueued" | "capacity" | "rejected" | "locked" @@ -298,14 +298,47 @@ export class SessionRegistry { * final `releaseAll()` on the lock manager so stragglers (e.g. * sessions that errored before publishing) cannot survive a * restart. + * + * @param drainTimeoutMs Per-session drain timeout (issue #342). + * When set, each `session.stop()` is raced against this timeout. + * If a session's stop() hangs, a warning is logged and the next + * session is drained — no individual session can block the + * shutdown sequence. Defaults to no per-session timeout. */ - async drainAll(): Promise { + async drainAll(drainTimeoutMs?: number): Promise { this.logger.info("Draining all sessions", { sessionCount: this.sessions.size, lockCount: this.lockTokens.size, + drainTimeoutMs: drainTimeoutMs ?? "none", }) const all = Array.from(this.sessions.values()) - await Promise.all(all.map((s) => s.stop())) + + const drain = async (session: StreamSession): Promise => { + const promise = session.stop() + if (drainTimeoutMs !== undefined && drainTimeoutMs > 0) { + let timedOut = false + let timer: ReturnType | undefined + const timeoutPromise = new Promise((resolve) => { + timer = setTimeout(() => { + timedOut = true + this.logger.warn( + `Session ${session.streamId} drain timed out after ${drainTimeoutMs}ms — continuing`, + ) + this.options.logger?.warn?.( + `[${this.workerId}] session ${session.streamId} drain timed out after ${drainTimeoutMs}ms`, + ) + resolve() + }, drainTimeoutMs) + if (typeof timer?.unref === "function") timer.unref() + }) + await Promise.race([promise, timeoutPromise]) + if (!timedOut && timer !== undefined) clearTimeout(timer) + } else { + await promise + } + } + + await Promise.all(all.map((s) => drain(s))) this.sessions.clear() this.lockTokens.clear() this.inflightAcquires.clear() diff --git a/xstreamroll-processing/src/worker.ts b/xstreamroll-processing/src/worker.ts index a2d1711..a0947a3 100644 --- a/xstreamroll-processing/src/worker.ts +++ b/xstreamroll-processing/src/worker.ts @@ -4,15 +4,15 @@ import http from "http" import axios from "axios" import { env } from "./config" -import { createLockManager, type LockManager } from "./leader-election" +import { type LockManager, createLockManager } from "./leader-election" import { GracefulShutdown, ShutdownReason } from "./lifecycle" import { currentCorrelationId, newCorrelationId } from "./logger" import { markShuttingDown, setQueueDepth, startMetricsServer } from "./metrics" import { EventFilter, - createFilterConfigStore, - MemoryFilterConfigStore, type FilterConfigStore, + MemoryFilterConfigStore, + createFilterConfigStore, } from "./pipeline" import { ProcessedStreamEvent, StreamEvent } from "./session" import { SessionRegistry } from "./session-registry" @@ -41,6 +41,13 @@ const LOCK_BACKEND: "memory" | "postgres" = (env.LOCK_BACKEND as "memory" | "postgres" | undefined) ?? "memory" const LOCK_TTL_MS: number = (env.LOCK_TTL_MS as number | undefined) ?? 30_000 +// Issue #342: per-session drain timeout used by drainAll(). When a +// session's stop() promise hangs, this timeout fires a warning so +// the shutdown can continue to the next session instead of blocking +// forever on a single stuck session. +const SHUTDOWN_TIMEOUT_MS: number = env.SHUTDOWN_TIMEOUT_MS +const SESSION_DRAIN_TIMEOUT_MS: number = env.SESSION_DRAIN_TIMEOUT_MS + // Issue #351: the EventFilter config store. Defaults to the same // in-process `Map` the worker used before the issue so existing // behaviour is preserved when EVENT_FILTER_BACKEND is unset. When @@ -341,7 +348,7 @@ async function start(): Promise { } const gracefulShutdown = new GracefulShutdown({ - timeoutMs: 15_000, + timeoutMs: SHUTDOWN_TIMEOUT_MS, }) gracefulShutdown.register({ @@ -353,12 +360,13 @@ gracefulShutdown.register({ gracefulShutdown.register({ name: "drain sessions", + timeoutMs: SHUTDOWN_TIMEOUT_MS, run: async () => { // Wait for the in-flight poll cycle to finish before draining // sessions so we don't tear state out from under it. await pollPromise if (!registry) return - await registry.drainAll() + await registry.drainAll(SESSION_DRAIN_TIMEOUT_MS) }, }) @@ -409,6 +417,7 @@ gracefulShutdown.register({ gracefulShutdown.register({ name: "stop metrics server", + timeoutMs: 5_000, run: () => new Promise((resolve, reject) => { // Flip the readiness flag first so any in-flight probe sees