Accept published PHP health read during or after its own task repair

scripts/conformance/server-parity/operator-metrics-contract.mjs

@@ -1,4 +1,5 @@
 import assert from 'node:assert/strict';
+import {isDeepStrictEqual} from 'node:util';
 
 // Reviewed operator metrics and health of one namespace, as published PHP
 // 2.5.14 reports them through a run's stages. Counts must match exactly, with
@@ -75,6 +76,12 @@ const EXPECTED = {
 const REPAIRED = {...EXPECTED.waiting_after_redispatch, tasks: {open: 1, ready: 1, ready_due_or_delayed: 1},
   backlog: {runnable_or_delayed_tasks: 1, tasks_added_last_minute: 1}};
 const REPAIRED_WARNINGS = ['command_contract_snapshots'];
+// PHP's TaskWatchdog repairs the task and projects the run before
+// TaskDispatcher redispatches it and projects the run again. Until then the
+// task is still overdue and the run reads as needing repair.
+const REPAIRING = {...EXPECTED.waiting_after_redispatch, runs: {...waitingRun, repair_needed: 1},
+  backlog: {...EXPECTED.waiting_after_redispatch.backlog, repair_needed_runs: 1}};
+const REPAIRING_WARNINGS = ['command_contract_snapshots', 'task_transport', 'routing_health', 'durable_resume_paths'];
 // Checks other than ok at each stage (the uncompared check aside).
 const WARNINGS = {
   empty: [],
@@ -86,6 +93,16 @@ const WARNINGS = {
   completed: [],
 };
 
+// Past the redispatch window published PHP is in one of these states, in
+// order: overdue, repairing, or redispatched. Metrics and health are separate
+// requests, so its health may be at the state its metrics show or a later one.
+// Native runs no repair loop and is held to the state its metrics show.
+const REPAIR_STATES = [
+  {counts: EXPECTED.waiting_after_redispatch, warnings: WARNINGS.waiting_after_redispatch},
+  {counts: REPAIRING, warnings: REPAIRING_WARNINGS},
+  {counts: REPAIRED, warnings: REPAIRED_WARNINGS},
+];
+
 function normalized(sections) {
   const counts = {};
   for (const [name, section] of Object.entries(sections)) {
@@ -134,24 +151,31 @@ function checkStage(name, stage, namespace, family) {
     assert.deepStrictEqual([raw.tasks.counts.ready_due, raw.starts.counts.ready_tasks], [1, 1], `${label} due task`);
   }
   const counts = normalized(raw);
-  const repaired = name === 'waiting_after_redispatch' && counts.tasks.max_repair_count > 0;
+  const redispatch = name === 'waiting_after_redispatch';
+  const php = redispatch && stage.runtime === 'php';
+  const repaired = redispatch && counts.tasks.max_repair_count > 0;
   if (repaired) delete counts.tasks.max_repair_count;
-  const expectedCounts = repaired ? REPAIRED : EXPECTED[name];
-  const expectedWarnings = repaired ? REPAIRED_WARNINGS : WARNINGS[name];
+  const states = redispatch ? REPAIR_STATES : [{counts: EXPECTED[name], warnings: WARNINGS[name]}];
+  // A repair count means PHP repaired the task: mid-repair (PHP only) or redispatched.
+  const possible = !repaired ? [0] : php ? [1, 2] : [2];
+  const nonzero = values => Object.fromEntries(Object.entries(values).filter(([, value]) => value !== 0));
+  const matches = index => Object.entries(counts)
+    .every(([section, values]) => isDeepStrictEqual(nonzero(values), states[index].counts[section] ?? {}));
+  const state = possible.find(matches) ?? possible[possible.length - 1];
   for (const [section, values] of Object.entries(counts)) {
-    const expected = expectedCounts[section] ?? {};
-    const nonzero = Object.fromEntries(Object.entries(values).filter(([, value]) => value !== 0));
-    assert.deepStrictEqual(nonzero, expected, `${label} ${section} counts`);
+    assert.deepStrictEqual(nonzero(values), states[state].counts[section] ?? {}, `${label} ${section} counts`);
   }
   assert.equal(stage.active_workers, ['empty', 'waiting', 'waiting_after_redispatch'].includes(name) ? 0 : 1,
     `${label} active workers`);
   if (defect) return counts;
   assert.deepStrictEqual(stage.checks.map(check => [check.name, check.category]), CHECKS, `${label} check names`);
   const statuses = stage.checks.filter(check => check.name !== UNCOMPARED_CHECK);
   assert.ok(statuses.every(check => ['ok', 'warning'].includes(check.status)), `${label} no check errors`);
-  assert.deepStrictEqual(statuses.filter(check => check.status !== 'ok').map(check => check.name), expectedWarnings,
-    `${label} warnings`);
-  const warning = expectedWarnings.length > 0 || stage.checks.some(check => check.status === 'warning');
+  const warnings = statuses.filter(check => check.status !== 'ok').map(check => check.name);
+  const allowed = (php ? states.slice(state) : [states[state]]).map(candidate => candidate.warnings);
+  const settled = allowed.find(list => isDeepStrictEqual(list, warnings)) ?? states[state].warnings;
+  assert.deepStrictEqual(warnings, settled, `${label} warnings`);
+  const warning = settled.length > 0 || stage.checks.some(check => check.status === 'warning');
   assert.deepStrictEqual([stage.health_status, stage.health.status, stage.health.healthy], [200, warning ? 'warning' : 'ok', true],
     `${label} health`);
   assert.deepStrictEqual(stage.retention, ['bounded', 30], `${label} namespace retention`);

tests/Unit/ServerParityOperatorMetricsContractTest.mjs

@@ -122,6 +122,85 @@ test('a task PHP\'s maintenance loop already redispatched passes', () => {
   for (const check of stage.checks) if (['task_transport', 'routing_health'].includes(check.name)) check.status = 'ok';
   assert.deepStrictEqual(checkOperatorMetrics(fixture, observation, 'TEST'), projection);
 });
+const setChecks = (stage, statuses) => {
+  for (const check of stage.checks) if (check.name in statuses) check.status = statuses[check.name];
+};
+test('published PHP health read while its maintenance loop repairs the task passes', () => {
+  const observation = model();
+  setChecks(observation.operator_metrics.stages.waiting_after_redispatch, {durable_resume_paths: 'warning'});
+  assert.deepStrictEqual(checkOperatorMetrics(fixture, observation, 'TEST'), projection);
+});
+test('published PHP health read after its maintenance loop redispatched the task passes', () => {
+  const observation = model();
+  setChecks(observation.operator_metrics.stages.waiting_after_redispatch, {task_transport: 'ok', routing_health: 'ok'});
+  assert.deepStrictEqual(checkOperatorMetrics(fixture, observation, 'TEST'), projection);
+});
+test('rejects native health reporting a repair native never runs', () => {
+  const observation = model();
+  const stage = observation.operator_metrics.stages.waiting_after_redispatch;
+  stage.runtime = 'rust';
+  setChecks(stage, {durable_resume_paths: 'warning'});
+  assert.throws(() => checkOperatorMetrics(fixture, observation, 'TEST'), /warnings/);
+});
+test('rejects native health reporting a redispatch native never runs', () => {
+  const observation = model();
+  const stage = observation.operator_metrics.stages.waiting_after_redispatch;
+  stage.runtime = 'rust';
+  setChecks(stage, {task_transport: 'ok', routing_health: 'ok'});
+  assert.throws(() => checkOperatorMetrics(fixture, observation, 'TEST'), /warnings/);
+});
+test('rejects PHP health behind metrics that show the task redispatched', () => {
+  const observation = model();
+  const stage = observation.operator_metrics.stages.waiting_after_redispatch;
+  Object.assign(stage.sections.tasks.counts, {dispatch_overdue: 0, unhealthy: 0, max_repair_count: 1});
+  Object.assign(stage.sections.tasks.timestamps, {oldest_dispatch_overdue_since: false, oldest_unhealthy_at: false});
+  stage.sections.backlog.counts.unhealthy_tasks = 0;
+  setChecks(stage, {durable_resume_paths: 'warning'});
+  assert.throws(() => checkOperatorMetrics(fixture, observation, 'TEST'), /warnings/);
+});
+// Metrics read while PHP's repair is committed but the task not yet
+// redispatched: still overdue, repaired once, and the run needs repair.
+const midRepair = (observation, runtime = 'php') => {
+  const stage = observation.operator_metrics.stages.waiting_after_redispatch;
+  stage.runtime = runtime;
+  stage.sections.tasks.counts.max_repair_count = 1;
+  stage.sections.runs.counts.repair_needed = 1;
+  stage.sections.runs.timestamps.oldest_repair_needed_at = true;
+  stage.sections.backlog.counts.repair_needed_runs = 1;
+  return stage;
+};
+test('published PHP metrics read while its maintenance loop repairs the task pass', () => {
+  const observation = model();
+  setChecks(midRepair(observation), {durable_resume_paths: 'warning'});
+  assert.deepStrictEqual(checkOperatorMetrics(fixture, observation, 'TEST'), projection);
+});
+test('published PHP health read after the repair its metrics caught midway passes', () => {
+  const observation = model();
+  setChecks(midRepair(observation), {task_transport: 'ok', routing_health: 'ok'});
+  assert.deepStrictEqual(checkOperatorMetrics(fixture, observation, 'TEST'), projection);
+});
+test('rejects PHP health behind metrics caught mid-repair', () => {
+  const observation = model();
+  midRepair(observation);
+  assert.throws(() => checkOperatorMetrics(fixture, observation, 'TEST'), /warnings/);
+});
+test('rejects native metrics showing a repair native never runs', () => {
+  const observation = model();
+  setChecks(midRepair(observation, 'rust'), {durable_resume_paths: 'warning'});
+  assert.throws(() => checkOperatorMetrics(fixture, observation, 'TEST'), /counts/);
+});
+test('rejects a run needing repair without a repaired task', () => {
+  const observation = model();
+  const stage = midRepair(observation);
+  stage.sections.tasks.counts.max_repair_count = 0;
+  setChecks(stage, {durable_resume_paths: 'warning'});
+  assert.throws(() => checkOperatorMetrics(fixture, observation, 'TEST'), /counts/);
+});
+test('rejects a repair warning at another stage', () => {
+  const observation = model();
+  setChecks(observation.operator_metrics.stages.waiting, {durable_resume_paths: 'warning'});
+  assert.throws(() => checkOperatorMetrics(fixture, observation, 'TEST'), /warnings/);
+});
 test('rejects a repaired task still reported overdue', () => {
   const observation = model();
   observation.operator_metrics.stages.waiting_after_redispatch.sections.tasks.counts.max_repair_count = 1;