feat(us3): implement the todos-api operational contract

Dockerfile

@@ -5,11 +5,11 @@ COPY package.json package-lock.json ./
 RUN npm ci --omit=dev --ignore-scripts \
     && npm cache clean --force
 
-FROM gcr.io/distroless/nodejs24-debian13:nonroot@sha256:fbbdda866ea71aef98c4abece17e3d61fbf820cc2ef3961522caa2478716171a
+FROM gcr.io/distroless/nodejs24-debian13:nonroot@sha256:774b7d020b24214835769e24c3544835526cd0288f0b094eae48e8b2c2429a79
 
 WORKDIR /app
 COPY --from=dependencies --chown=65532:65532 /app/node_modules ./node_modules
-COPY --chown=65532:65532 server.js routes.js todoController.js ./
+COPY --chown=65532:65532 server.js routes.js todoController.js operational.js ./
 
 # The distroless nonroot account maps to UID/GID 65532. A numeric image user
 # lets Kubernetes verify runAsNonRoot before it starts the container.

contracts/asyncapi.yaml

@@ -68,3 +68,11 @@ components:
           oneOf:
             - type: integer
             - type: string
+        correlationId:
+          type: string
+          description: >-
+            The X-Request-Id of the HTTP request that produced this operation.
+            Additive and optional: the consumer reads only zipkinSpan and prints
+            the rest, so an older publisher that omits it still processes
+            cleanly. It exists so an audit line can be tied back to the request
+            that caused it without going through Zipkin.

operational.js

@@ -0,0 +1,246 @@
+'use strict';
+
+// Operational contract for todos-api (spec 009, T080): health probes,
+// correlation, non-secret runtime configuration, and a Redis publish that
+// cannot take the API down with it.
+//
+// Kept out of server.js because these are platform concerns rather than todo
+// concerns: how Kubernetes decides the pod is alive, how a request stays
+// traceable, and what happens to the API when its audit sink is unavailable.
+
+const crypto = require('node:crypto');
+
+// The id a caller may supply and that this service echoes and records. The same
+// header is used by the other four services, so one request produces one id
+// across every log line and audit entry it touches.
+const CORRELATION_HEADER = 'X-Request-Id';
+
+// --- health ----------------------------------------------------------------
+
+// Three separate answers. Collapsing readiness into liveness means a pod that
+// is merely draining gets restarted instead of being removed from the Service,
+// turning a brief degradation into a crash loop.
+function createHealthState () {
+  let started = true;
+  let ready = true;
+
+  return {
+    setStarted (value) { started = value; },
+    setReady (value) { ready = value; },
+    isStarted () { return started; },
+    isReady () { return ready; }
+  };
+}
+
+// Mounted before the JWT middleware, deliberately.
+//
+// /todos sits behind expressjwt. A probe mounted after it answers 401,
+// Kubernetes reads 401 as unhealthy, and every pod restarts forever while the
+// application itself is perfectly fine. The same applies to /metrics: a scrape
+// that needs a token stops working during exactly the incident you need it for.
+function registerOperationalRoutes (app, health, metricsHandler) {
+  app.get('/health/startup', (req, res) => {
+    if (!health.isStarted()) {
+      return res.status(503).json({ status: 'starting' });
+    }
+    return res.status(200).json({ status: 'ok' });
+  });
+
+  // "Should this pod receive traffic right now."
+  app.get('/health/ready', (req, res) => {
+    if (!health.isReady()) {
+      return res.status(503).json({ status: 'not-ready' });
+    }
+    return res.status(200).json({ status: 'ok' });
+  });
+
+  // "Is this process wedged." It must not consult Redis: Redis carries the
+  // audit log, and letting its outage restart every todos-api pod would turn a
+  // logging incident into an API outage.
+  app.get('/health/live', (req, res) => res.status(200).json({ status: 'ok' }));
+
+  if (metricsHandler) {
+    app.get('/metrics', metricsHandler);
+  }
+}
+
+// --- correlation -----------------------------------------------------------
+
+function correlationMiddleware () {
+  return function (req, res, next) {
+    const id = req.get(CORRELATION_HEADER) || crypto.randomUUID();
+    req.correlationId = id;
+    res.set(CORRELATION_HEADER, id);
+    next();
+  };
+}
+
+// --- runtime configuration -------------------------------------------------
+
+function envBool (name, fallback) {
+  const raw = process.env[name];
+  if (raw === undefined || raw === '') return fallback;
+  return raw === 'true' || raw === '1';
+}
+
+function envInt (name, fallback) {
+  const raw = process.env[name];
+  if (raw === undefined || raw === '') return fallback;
+  const value = Number.parseInt(raw, 10);
+  if (Number.isNaN(value) || value < 0) return fallback;
+  return value;
+}
+
+// Non-secret operational values only.
+//
+// This object gets logged at startup so an operator can see what the pod
+// actually loaded, which is exactly why the JWT secret must not be in it: a
+// secret in a loggable object is a secret in the log aggregator.
+function loadRuntimeConfig () {
+  return {
+    features: {
+      // Off by default. A toggle that defaults on ships its behaviour to
+      // production the moment it merges, which defeats the point of having one.
+      verboseErrors: envBool('TODO_API_FEATURE_VERBOSE_ERRORS', false)
+    },
+    redis: {
+      // Usable rather than zero: a zero timeout means every audit write fails
+      // instantly, so a missing ConfigMap key would silently disable the audit
+      // trail rather than degrade it.
+      publishTimeoutMs: envInt('TODO_API_REDIS_PUBLISH_TIMEOUT_MS', 1000),
+      failureThreshold: envInt('TODO_API_REDIS_BREAKER_THRESHOLD', 5),
+      breakerOpenMs: envInt('TODO_API_REDIS_BREAKER_OPEN_MS', 10000)
+    }
+  };
+}
+
+// --- Redis safety ----------------------------------------------------------
+
+// node-redis surfaces connection failures as an 'error' event on the client.
+// An unhandled 'error' event terminates the Node process, so without this a
+// Redis restart takes todos-api down with it — for a dependency that only
+// carries the audit log.
+function attachRedisErrorHandler (redisClient, logger = console) {
+  if (!redisClient || typeof redisClient.on !== 'function') return redisClient;
+
+  redisClient.on('error', error => {
+    logger.error(JSON.stringify({
+      level: 'error',
+      msg: 'redis_client_error',
+      error: error && error.message ? error.message : String(error)
+    }));
+  });
+
+  return redisClient;
+}
+
+// Stops publishing to a Redis that is clearly down.
+//
+// Once Redis is failing, every create would otherwise wait out the full timeout
+// before succeeding, so a logging outage becomes a latency incident on the API.
+// The breaker closes itself after its window rather than needing a restart.
+function createCircuitBreaker ({ failureThreshold = 5, openMs = 10000 } = {}) {
+  let failures = 0;
+  let openedAt = 0;
+  let isOpen = false;
+
+  return {
+    allows () {
+      if (!isOpen) return true;
+      if (Date.now() - openedAt >= openMs) {
+        // Half-open: allow one probe through.
+        isOpen = false;
+        failures = failureThreshold - 1;
+        return true;
+      }
+      return false;
+    },
+    recordSuccess () {
+      failures = 0;
+      isOpen = false;
+    },
+    recordFailure () {
+      failures += 1;
+      if (failures >= failureThreshold) {
+        isOpen = true;
+        openedAt = Date.now();
+      }
+    },
+    get open () { return isOpen; }
+  };
+}
+
+// Publishes an audit line without ever letting Redis fail the request.
+//
+// Three things can go wrong and all three are contained here: the callback
+// reports an error, the call throws synchronously, or it never settles at all.
+// The last is the nastiest — a hung Redis would otherwise hold every create
+// open indefinitely — so the promise resolves on a timer regardless.
+//
+// It resolves rather than rejects on failure by design. The caller is a todo
+// write, and the audit line is best-effort; a rejected promise here would only
+// invite someone to await it and reintroduce the coupling.
+function publishAudit (redisClient, channel, message, { timeoutMs = 1000, breaker, logger = console } = {}) {
+  return new Promise(resolve => {
+    if (breaker && !breaker.allows()) {
+      logger.error(JSON.stringify({ level: 'warn', msg: 'redis_audit_skipped_circuit_open', channel }));
+      return resolve({ published: false, reason: 'circuit-open' });
+    }
+
+    let settled = false;
+    const finish = result => {
+      if (settled) return;
+      settled = true;
+      if (breaker) {
+        if (result.published) breaker.recordSuccess();
+        else breaker.recordFailure();
+      }
+      resolve(result);
+    };
+
+    const timer = setTimeout(() => {
+      logger.error(JSON.stringify({ level: 'warn', msg: 'redis_audit_timeout', channel, timeoutMs }));
+      finish({ published: false, reason: 'timeout' });
+    }, timeoutMs);
+    if (typeof timer.unref === 'function') timer.unref();
+
+    const done = result => {
+      clearTimeout(timer);
+      finish(result);
+    };
+
+    try {
+      redisClient.publish(channel, message, error => {
+        if (error) {
+          logger.error(JSON.stringify({
+            level: 'error',
+            msg: 'redis_audit_failed',
+            channel,
+            error: error.message
+          }));
+          return done({ published: false, reason: 'error' });
+        }
+        return done({ published: true });
+      });
+    } catch (error) {
+      logger.error(JSON.stringify({
+        level: 'error',
+        msg: 'redis_audit_threw',
+        channel,
+        error: error.message
+      }));
+      done({ published: false, reason: 'threw' });
+    }
+  });
+}
+
+module.exports = {
+  CORRELATION_HEADER,
+  createHealthState,
+  registerOperationalRoutes,
+  correlationMiddleware,
+  loadRuntimeConfig,
+  attachRedisErrorHandler,
+  createCircuitBreaker,
+  publishAudit
+};

routes.js

@@ -1,11 +1,11 @@
 'use strict';
 const TodoController = require('./todoController');
-module.exports = function (app, {tracer, redisClient, logChannel}) {
-  const todoController = new TodoController({tracer, redisClient, logChannel});
+module.exports = function (app, {tracer, redisClient, logChannel, redisBreaker, redisPublishTimeoutMs}) {
+  const todoController = new TodoController({tracer, redisClient, logChannel, redisBreaker, redisPublishTimeoutMs});
   app.route('/todos')
     .get(function(req,resp) {return todoController.list(req,resp)})
     .post(function(req,resp) {return todoController.create(req,resp)});
 
   app.route('/todos/:taskId')
     .delete(function(req,resp) {return todoController.delete(req,resp)});
-};
+};

server.js

@@ -13,6 +13,14 @@ const CLSContext = require('zipkin-context-cls');
 const { HttpLogger } = require('zipkin-transport-http');
 const zipkinMiddleware = require('zipkin-instrumentation-express').expressMiddleware;
 const routes = require('./routes');
+const {
+  createHealthState,
+  registerOperationalRoutes,
+  correlationMiddleware,
+  loadRuntimeConfig,
+  attachRedisErrorHandler,
+  createCircuitBreaker
+} = require('./operational');
 
 function createRedisClient () {
   return redis.createClient({
@@ -54,6 +62,19 @@ function createApp (options = {}) {
   const logChannel = options.logChannel || process.env.REDIS_CHANNEL || 'log_channel';
   const jwtSecret = options.jwtSecret || process.env.JWT_SECRET || 'foo';
 
+  const config = options.config || loadRuntimeConfig();
+  const health = createHealthState();
+
+  // node-redis reports connection failures as an 'error' event. Unhandled, that
+  // event terminates the process, so a Redis restart would take the todo API
+  // down for the sake of its audit log.
+  attachRedisErrorHandler(redisClient);
+
+  const redisBreaker = createCircuitBreaker({
+    failureThreshold: config.redis.failureThreshold,
+    openMs: config.redis.breakerOpenMs
+  });
+
   const register = new prometheus.Registry();
   const requestCount = new prometheus.Counter({
     name: 'todo_api_requests_total',
@@ -69,10 +90,14 @@ function createApp (options = {}) {
   });
   prometheus.collectDefaultMetrics({ register, prefix: 'todos_api_' });
 
-  app.get('/metrics', async (req, res) => {
+  const metricsHandler = async (req, res) => {
     res.set('Content-Type', register.contentType);
     res.end(await register.metrics());
-  });
+  };
+
+  // Correlation first, so every downstream log line and the audit record all
+  // carry the same id.
+  app.use(correlationMiddleware());
 
   app.use((req, res, next) => {
     const stopTimer = requestDuration.startTimer({ method: req.method });
@@ -82,6 +107,11 @@ function createApp (options = {}) {
     });
     next();
   });
+  // Before expressjwt, deliberately: a probe or a scrape that needs a token
+  // answers 401, Kubernetes reads that as unhealthy, and every pod restarts
+  // forever while the application is fine.
+  registerOperationalRoutes(app, health, metricsHandler);
+
   app.use(expressjwt({
     secret: jwtSecret,
     algorithms: ['HS256'],
@@ -100,7 +130,16 @@ function createApp (options = {}) {
 
   app.use(express.urlencoded({ extended: false }));
   app.use(express.json());
-  routes(app, { tracer, redisClient, logChannel });
+  app.locals.health = health;
+  app.locals.config = config;
+
+  routes(app, {
+    tracer,
+    redisClient,
+    logChannel,
+    redisBreaker,
+    redisPublishTimeoutMs: options.redisPublishTimeoutMs || config.redis.publishTimeoutMs
+  });
 
   return app;
 }

test/container-packaging.test.js

@@ -0,0 +1,20 @@
+'use strict';
+
+const assert = require('node:assert/strict');
+const { readFile } = require('node:fs/promises');
+const path = require('node:path');
+const { test } = require('node:test');
+
+test('the runtime image contains the operational contract module', async () => {
+  const dockerfile = await readFile(path.join(__dirname, '..', 'Dockerfile'), 'utf8');
+  const runtimeCopy = dockerfile
+    .split('\n')
+    .find(line => line.startsWith('COPY --chown=65532:65532 '));
+
+  assert.ok(runtimeCopy, 'the runtime source COPY instruction is missing');
+  assert.match(
+    runtimeCopy,
+    /(?:^|\s)operational\.js(?:\s|$)/,
+    'server.js requires ./operational, but the runtime image does not copy operational.js'
+  );
+});