Report fleet convergence in status, and label logs by server

A rolling deploy that dies partway leaves the fleet serving two releases
at once, and nothing about a single replica reveals it — status printed
each server's release happily and never compared them. It now collects
one observation per replica and reports the cross-replica view: which
release each is on, whether it is in rotation, and an explicit warning
naming the replicas on each release when they disagree.

A replica with no release, or one left drained by a failed roll, counts
as not converged too. Both are silently serving nothing, and the drained
case will stay that way until someone notices, so the warning says how to
put it back.

The comparison is skipped under --on: one observation proves nothing
about the others, and reporting "converged" from it would be worse than
saying nothing.

status also stops reporting a placement: 'primary' process as missing on
the replicas it is deliberately absent from. logs now prefixes every line
with its server, not every block — a fleet interleaves output, and a
block prefix leaves the reader guessing which box a stack trace came from.

Author: devalade

src/cli/commands/logs.ts

@@ -1,9 +1,24 @@
 import { runRemoteCommandForTargets } from '../runner.js';
 
+/**
+ * Prefix every line, not every block.
+ *
+ * A fleet interleaves output from several servers, and a block-level prefix
+ * leaves the reader guessing which box a stack trace came from once the second
+ * server's output starts.
+ */
+function prefixLines(text: string, label: string): string {
+  return text
+    .split('\n')
+    .filter((line) => line.length > 0)
+    .map((line) => `${label} ${line}`)
+    .join('\n');
+}
+
 export async function cmdLogs(cwd: string, options: { lines?: number; config?: string; process?: string; app?: string; on?: string }): Promise<void> {
   await runRemoteCommandForTargets(
     cwd,
-    async ({ config, executor }) => {
+    async ({ config, executor, serverName }) => {
       const apps = options.app
         ? config.apps.filter((app) => app.name === options.app)
         : config.apps.filter((a) => a.appType === 'backend' && a.pm2);
@@ -31,8 +46,9 @@ export async function cmdLogs(cwd: string, options: { lines?: number; config?: s
         const result = await executor.exec(
           `${mise}; mise exec "node@${nodeVersion}" -- pm2 logs ${target} --lines ${lines} --nostream`,
         );
-        if (result.stdout) process.stdout.write(`[${app.name}] ${result.stdout}\n`);
-        if (result.stderr) process.stderr.write(`[${app.name}] ${result.stderr}\n`);
+        const label = `[${serverName} ${app.name}]`;
+        if (result.stdout) process.stdout.write(`${prefixLines(result.stdout, label)}\n`);
+        if (result.stderr) process.stderr.write(`${prefixLines(result.stderr, label)}\n`);
       }
     },
     { configPath: options.config, appName: options.app, serverName: options.on },

src/cli/commands/status.ts

@@ -1,8 +1,20 @@
 import { runRemoteCommandForTargets } from '../runner.js';
 import { ui } from '../ui.js';
 import { getDeploymentName, getPm2Name } from '../../domain/pm2/apps.js';
+import { isDrained } from '../../domain/deploy/drain.js';
+import {
+  assessConvergence,
+  describeConvergence,
+  type ReplicaObservation,
+} from '../../domain/deploy/convergence.js';
 
 export async function cmdStatus(cwd: string, options: { config?: string; app?: string; on?: string }): Promise<void> {
+  // Per-app, per-replica observations, gathered as the fan-out visits each
+  // server. Release skew is invisible from inside one server, so the comparison
+  // happens after every replica has reported.
+  const observed = new Map<string, ReplicaObservation[]>();
+  const fleetApps = new Set<string>();
+
   await runRemoteCommandForTargets(
     cwd,
     async ({ config, executor, serverName }) => {
@@ -36,7 +48,13 @@ export async function cmdStatus(cwd: string, options: { config?: string; app?: s
                 const pm2Name = getPm2Name(namespace, pm2App.name);
                 const running = byName.get(pm2Name);
                 if (!running) {
-                  ui.warn(`  App '${pm2App.name}' not found in PM2`);
+                  // A process pinned to the primary is absent everywhere else by
+                  // design; saying "not found" would report the feature as a fault.
+                  if (pm2App.placement === 'primary') {
+                    ui.info(`  App '${pm2App.name}' is pinned to the primary replica — not expected here`);
+                  } else {
+                    ui.warn(`  App '${pm2App.name}' not found in PM2`);
+                  }
                   continue;
                 }
                 ui.heading(`  PM2: ${pm2App.name}`);
@@ -59,12 +77,34 @@ export async function cmdStatus(cwd: string, options: { config?: string; app?: s
 
         const appPath = `${config.remotePath}/${app.name}`;
         const currentResult = await executor.exec(`readlink "${appPath}/current" 2>/dev/null || echo "no current symlink"`);
-        if (currentResult.stdout !== 'no current symlink') {
+        const hasRelease = currentResult.stdout !== 'no current symlink' && currentResult.stdout !== '';
+        if (hasRelease) {
           ui.success(`  Current release: ${currentResult.stdout}`);
         } else {
           ui.warn('  No active release');
         }
 
+        let drained: boolean | null = null;
+        if (app.fleet) {
+          fleetApps.add(app.name);
+          drained = await isDrained(executor, config.remotePath, app.name);
+          if (drained) {
+            ui.warn(`  Rotation: draining (${app.fleet.readyPath} answers 503)`);
+          } else {
+            ui.success('  Rotation: in rotation');
+          }
+        }
+
+        const entries = observed.get(app.name) ?? [];
+        entries.push({
+          server: serverName,
+          // The symlink is absolute; only the release directory name is
+          // comparable between replicas.
+          release: hasRelease ? currentResult.stdout.split('/').pop() ?? null : null,
+          drained,
+        });
+        observed.set(app.name, entries);
+
         const releasesResult = await executor.exec(`ls -1t "${appPath}/releases/" 2>/dev/null | head -5`);
         if (releasesResult.stdout) {
           const releases = releasesResult.stdout.split('\n').filter(Boolean);
@@ -74,4 +114,44 @@ export async function cmdStatus(cwd: string, options: { config?: string; app?: s
     },
     { configPath: options.config, serverName: options.on },
   );
+
+  reportFleetConvergence(observed, fleetApps, options.on !== undefined);
+}
+
+/**
+ * The cross-replica view, which is the only place a half-rolled fleet shows up.
+ *
+ * Skipped when `--on` narrowed the run to one server: a single observation
+ * proves nothing about the others, and reporting "converged" from it would be
+ * worse than saying nothing.
+ */
+function reportFleetConvergence(
+  observed: Map<string, ReplicaObservation[]>,
+  fleetApps: Set<string>,
+  narrowed: boolean,
+): void {
+  if (narrowed) return;
+
+  for (const [appName, observations] of observed) {
+    if (!fleetApps.has(appName) || observations.length < 2) continue;
+
+    const convergence = assessConvergence(observations);
+    ui.heading(`Fleet: ${appName}`);
+    ui.section(
+      '  Replicas',
+      observations.map((o) => [
+        o.server,
+        `${o.release ?? 'no release'}${o.drained ? ' (draining)' : ''}`,
+      ]),
+    );
+
+    if (convergence.converged) {
+      ui.success(`  All ${observations.length} replicas on ${convergence.releases[0]}`);
+      continue;
+    }
+
+    for (const line of describeConvergence(appName, observations, convergence)) {
+      ui.warn(`  ${line}`);
+    }
+  }
 }

src/domain/deploy/convergence.ts

@@ -0,0 +1,81 @@
+/**
+ * Did the fleet converge?
+ *
+ * A rolling deploy touches replicas one batch at a time, so an interrupted roll
+ * leaves the fleet serving two different releases at once. That state is
+ * legitimate mid-roll and a bug afterwards, and nothing about a single replica
+ * reveals it — `status` has to compare them. The same comparison catches a
+ * replica left drained by a failed roll, which is silently out of rotation and
+ * will stay that way until someone notices.
+ */
+
+export interface ReplicaObservation {
+  server: string;
+  /** The release directory `current` points at, or null if there is no release. */
+  release: string | null;
+  /** Whether the replica is out of rotation. Null for an app with no drain contract. */
+  drained: boolean | null;
+}
+
+export interface FleetConvergence {
+  /** Distinct releases seen, newest-sorting first. One entry means converged. */
+  releases: string[];
+  /** Replicas out of rotation. */
+  drained: string[];
+  /** Replicas with no release at all. */
+  undeployed: string[];
+  converged: boolean;
+}
+
+export function assessConvergence(observations: ReplicaObservation[]): FleetConvergence {
+  const releases = [...new Set(
+    observations.map((o) => o.release).filter((r): r is string => r !== null),
+  )].sort().reverse();
+
+  const drained = observations.filter((o) => o.drained === true).map((o) => o.server);
+  const undeployed = observations.filter((o) => o.release === null).map((o) => o.server);
+
+  return {
+    releases,
+    drained,
+    undeployed,
+    // A replica with no release at all is as much a failure to converge as two
+    // different releases — the fleet is not uniformly serving anything.
+    converged: releases.length <= 1 && undeployed.length === 0 && drained.length === 0,
+  };
+}
+
+/** Human-readable lines describing what is wrong, empty when the fleet is converged. */
+export function describeConvergence(
+  appName: string,
+  observations: ReplicaObservation[],
+  convergence: FleetConvergence,
+): string[] {
+  if (convergence.converged) return [];
+
+  const lines: string[] = [];
+  const at = (release: string): string =>
+    observations.filter((o) => o.release === release).map((o) => o.server).join(', ');
+
+  if (convergence.releases.length > 1) {
+    lines.push(
+      `${appName} is running ${convergence.releases.length} different releases: ` +
+      convergence.releases.map((release) => `${release} on ${at(release)}`).join('; ') +
+      `. A roll stopped partway — redeploy to converge the fleet.`,
+    );
+  }
+
+  if (convergence.undeployed.length > 0) {
+    lines.push(`${appName} has no release on ${convergence.undeployed.join(', ')}.`);
+  }
+
+  if (convergence.drained.length > 0) {
+    lines.push(
+      `${appName} is out of rotation on ${convergence.drained.join(', ')} — ` +
+      `the load balancer is sending it no traffic. Run 'shipnode undrain --app ${appName}' ` +
+      `once it is healthy.`,
+    );
+  }
+
+  return lines;
+}

tests/unit/convergence.test.ts

@@ -0,0 +1,93 @@
+import { describe, expect, it } from 'vitest';
+import {
+  assessConvergence,
+  describeConvergence,
+  type ReplicaObservation,
+} from '../../src/domain/deploy/convergence.js';
+
+const replica = (
+  server: string,
+  release: string | null,
+  drained: boolean | null = false,
+): ReplicaObservation => ({ server, release, drained });
+
+describe('assessConvergence', () => {
+  it('is converged when every replica serves the same release and is in rotation', () => {
+    const result = assessConvergence([
+      replica('web-a', '2026-01-02'),
+      replica('web-b', '2026-01-02'),
+    ]);
+
+    expect(result.converged).toBe(true);
+    expect(result.releases).toEqual(['2026-01-02']);
+  });
+
+  it('reports skew newest first, which is the release the roll was heading to', () => {
+    const result = assessConvergence([
+      replica('web-a', '2026-01-02'),
+      replica('web-b', '2026-01-01'),
+      replica('web-c', '2026-01-01'),
+    ]);
+
+    expect(result.converged).toBe(false);
+    expect(result.releases).toEqual(['2026-01-02', '2026-01-01']);
+  });
+
+  it('counts a replica with no release as not converged', () => {
+    // Uniform in the sense that one release is present, but a replica serving
+    // nothing is exactly the failure this is meant to surface.
+    const result = assessConvergence([replica('web-a', '2026-01-02'), replica('web-b', null)]);
+
+    expect(result.converged).toBe(false);
+    expect(result.undeployed).toEqual(['web-b']);
+  });
+
+  it('counts a drained replica as not converged even on the right release', () => {
+    // A failed roll leaves the replica drained. It is out of rotation and will
+    // stay there until someone notices.
+    const result = assessConvergence([
+      replica('web-a', '2026-01-02'),
+      replica('web-b', '2026-01-02', true),
+    ]);
+
+    expect(result.converged).toBe(false);
+    expect(result.drained).toEqual(['web-b']);
+  });
+
+  it('ignores drain state for an app with no drain contract', () => {
+    const result = assessConvergence([replica('web-a', '2026-01-02', null)]);
+
+    expect(result.converged).toBe(true);
+    expect(result.drained).toEqual([]);
+  });
+});
+
+describe('describeConvergence', () => {
+  it('says nothing when the fleet is converged', () => {
+    const observations = [replica('web-a', '2026-01-02'), replica('web-b', '2026-01-02')];
+
+    expect(describeConvergence('api', observations, assessConvergence(observations))).toEqual([]);
+  });
+
+  it('names which replicas are on which release', () => {
+    const observations = [
+      replica('web-a', '2026-01-02'),
+      replica('web-b', '2026-01-01'),
+      replica('web-c', '2026-01-01'),
+    ];
+
+    const [line] = describeConvergence('api', observations, assessConvergence(observations));
+
+    expect(line).toContain('2026-01-02 on web-a');
+    expect(line).toContain('2026-01-01 on web-b, web-c');
+    expect(line).toContain('redeploy to converge');
+  });
+
+  it('tells the reader how to put a drained replica back', () => {
+    const observations = [replica('web-a', '2026-01-02'), replica('web-b', '2026-01-02', true)];
+
+    const lines = describeConvergence('api', observations, assessConvergence(observations));
+
+    expect(lines.join(' ')).toContain("shipnode undrain --app api");
+  });
+});