packages/runtime-host/src/protocol/host-status.ts

/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing,
 * software distributed under the License is distributed on an
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 * KIND, either express or implied.  See the License for the
 * specific language governing permissions and limitations
 * under the License.
 */

import { invalidProtocolFrame } from './errors.js';
import {
  requireCount,
  requireEncodedByteLimit,
  requireExactRecord,
  requireId,
  requireRecord,
  requireShapedRecord,
  requireString,
  requireUtf8String,
} from './codec.js';
import { defineOperation } from './operation-spec.js';
import {
  decodeSignedPeerReachabilityLease,
  type SignedPeerReachabilityLeaseV1,
} from '../peer-reachability/model.js';

export type HostLifecycleState = 'starting' | 'containing' | 'recovering' | 'ready' | 'draining';
export type HostStatusInput = Record<string, never>;
export type HostDiagnosticsInput = Record<string, never>;
export interface HostActivitySnapshot {
  readonly connections: number;
  readonly activeOperations: number;
  readonly processUptimeSeconds: number;
  readonly residencies: readonly { readonly label: string; readonly count: number }[];
  /** Negotiated maintenance evidence. Absent on released Hosts: every residency is conservative. */
  readonly drainResidencies?: number;
  readonly cooperativeHandoff?: boolean;
}

export function isHostActivityIdle(activity: HostActivitySnapshot): boolean {
  return (
    activity.connections === 0 &&
    activity.activeOperations === 0 &&
    (activity.drainResidencies === undefined
      ? activity.residencies.length === 0
      : activity.drainResidencies === 0)
  );
}

export interface HostUpgradePrepareInput {
  readonly expectedHostEpoch: string;
  readonly allowInterruptActiveTasks: boolean;
  readonly allowCooperativeHandoff?: boolean;
}

export type HostUpgradePrepareResult =
  | { readonly kind: 'active_tasks' }
  | { readonly kind: 'prepared'; readonly pid: number };

export const HOST_DIAGNOSTICS_RESULT_MAX_BYTES = 72 * 1024;
export const HOST_DIAGNOSTIC_LOG_MAX_ENTRIES = 256;
export const HOST_DIAGNOSTIC_LOG_MAX_ENTRY_BYTES = 10 * 1024;

export interface HostStatusResult {
  hostEpoch: string;
  compositionId: string;
  compositionRevision: string;
  state: HostLifecycleState;
  connections: number;
  activeOperations: number;
  activeResidencies: number;
  peerEndpoint?: HostPeerEndpoint;
}

export type HostPeerEndpoint = SignedPeerReachabilityLeaseV1;

export interface HostDiagnosticsResult extends HostStatusResult {
  compositionModules: readonly string[];
  residencies: readonly { label: string; count: number }[];
  /**
   * The Host's authoritative answer to "would a maintenance drain interrupt
   * active work right now", computed by the same authority that gates
   * `host.upgrade.prepare`. Required: the epoch gate already refuses
   * mixed-version peers, so there is no wire case where it is absent.
   */
  upgradeBlockingActivity: boolean;
  protocolVersion: number;
  compatibilityEpoch: number;
  pid: number;
  processUptimeSeconds: number;
  nodeVersion: string;
  platform: NodeJS.Platform;
  arch: string;
  osRelease: string;
  logs: readonly string[];
}

export const HOST_BOOTSTRAP_OPERATION_SPECS = {
  'host.status': defineOperation({
    mode: 'query',
    availability: 'bootstrap',
    errors: ['host_draining', 'internal_failure'] as const,
    decodeInput: (value) => decodeEmptyHostInput(value, 'host.status input'),
    decodeOutput: decodeHostStatusResult,
  }),
  'host.diagnostics.query': defineOperation({
    mode: 'query',
    availability: 'bootstrap',
    errors: ['host_draining', 'internal_failure'] as const,
    decodeInput: (value) => decodeEmptyHostInput(value, 'host.diagnostics.query input'),
    decodeOutput: decodeHostDiagnosticsResult,
  }),
  'host.upgrade.prepare': defineOperation({
    mode: 'command',
    availability: 'ready',
    errors: ['operation_conflict', 'operation_unavailable', 'internal_failure'] as const,
    decodeInput: decodeHostUpgradePrepareInput,
    decodeOutput: decodeHostUpgradePrepareResult,
  }),
} as const;

function decodeEmptyHostInput(value: unknown, label: string): HostStatusInput {
  requireExactRecord(value, label, []);
  return {};
}

function decodeHostStatusResult(value: unknown): HostStatusResult {
  const valueRecord = requireRecord(value, 'host.status result');
  const record = requireExactRecord(value, 'host.status result', [
    'hostEpoch',
    'compositionId',
    'compositionRevision',
    'state',
    'connections',
    'activeOperations',
    'activeResidencies',
    ...(valueRecord.peerEndpoint === undefined ? [] : ['peerEndpoint']),
  ]);
  return decodeHostStatusFields(record);
}

function decodeHostDiagnosticsResult(value: unknown): HostDiagnosticsResult {
  requireEncodedByteLimit(
    value,
    'host.diagnostics.query result',
    HOST_DIAGNOSTICS_RESULT_MAX_BYTES,
  );
  const valueRecord = requireRecord(value, 'host.diagnostics.query result');
  const record = requireExactRecord(value, 'host.diagnostics.query result', [
    'hostEpoch',
    'compositionId',
    'compositionRevision',
    'state',
    'connections',
    'activeOperations',
    'activeResidencies',
    ...(valueRecord.peerEndpoint === undefined ? [] : ['peerEndpoint']),
    'upgradeBlockingActivity',
    'compositionModules',
    'residencies',
    'protocolVersion',
    'compatibilityEpoch',
    'pid',
    'processUptimeSeconds',
    'nodeVersion',
    'platform',
    'arch',
    'osRelease',
    'logs',
  ]);
  if (!Array.isArray(record.logs) || record.logs.length > HOST_DIAGNOSTIC_LOG_MAX_ENTRIES) {
    throw invalidProtocolFrame('Invalid Runtime Host diagnostic logs');
  }
  if (!Array.isArray(record.compositionModules) || record.compositionModules.length > 64) {
    throw invalidProtocolFrame('Invalid Runtime Host composition modules');
  }
  if (!Array.isArray(record.residencies) || record.residencies.length > 128) {
    throw invalidProtocolFrame('Invalid Runtime Host residencies');
  }
  return {
    ...decodeHostStatusFields(record),
    upgradeBlockingActivity: requireUpgradeBlockingActivity(record.upgradeBlockingActivity),
    compositionModules: record.compositionModules.map((moduleId) =>
      requireString(moduleId, 'Runtime Host composition module id', 64),
    ),
    residencies: record.residencies.map((value) => {
      const residency = requireExactRecord(value, 'Runtime Host residency', ['label', 'count']);
      return {
        label: requireString(residency.label, 'Runtime Host residency label', 128),
        count: requireCount(residency.count, 'Runtime Host residency count'),
      };
    }),
    protocolVersion: requireCount(record.protocolVersion, 'Runtime Host protocol version'),
    compatibilityEpoch: requireCount(record.compatibilityEpoch, 'Runtime Host compatibility epoch'),
    pid: requireCount(record.pid, 'Runtime Host pid'),
    processUptimeSeconds: requireCount(record.processUptimeSeconds, 'Runtime Host process uptime'),
    nodeVersion: requireString(record.nodeVersion, 'Runtime Host Node version', 64),
    platform: requirePlatform(record.platform),
    arch: requireString(record.arch, 'Runtime Host architecture', 64),
    osRelease: requireString(record.osRelease, 'Runtime Host OS release', 256),
    logs: record.logs.map((entry) =>
      requireUtf8String(
        entry,
        'Runtime Host diagnostic log entry',
        HOST_DIAGNOSTIC_LOG_MAX_ENTRY_BYTES,
      ),
    ),
  };
}

function requireUpgradeBlockingActivity(value: unknown): boolean {
  if (typeof value !== 'boolean') {
    throw invalidProtocolFrame('Invalid Runtime Host upgrade blocking activity');
  }
  return value;
}

export function decodeHostActivitySnapshot(value: unknown): HostActivitySnapshot {
  const record = requireShapedRecord(
    value,
    'Runtime Host activity',
    ['connections', 'activeOperations', 'processUptimeSeconds', 'residencies'],
    ['drainResidencies', 'cooperativeHandoff'],
  );
  if (!Array.isArray(record.residencies) || record.residencies.length > 128) {
    throw invalidProtocolFrame('Invalid Runtime Host activity residencies');
  }
  return {
    ...(record.cooperativeHandoff === undefined
      ? {}
      : {
          cooperativeHandoff: requireBoolean(
            record.cooperativeHandoff,
            'Runtime Host cooperative handoff capability',
          ),
        }),
    ...(record.drainResidencies === undefined
      ? {}
      : {
          drainResidencies: requireCount(record.drainResidencies, 'Runtime Host drain residencies'),
        }),
    connections: requireCount(record.connections, 'Runtime Host activity connections'),
    activeOperations: requireCount(
      record.activeOperations,
      'Runtime Host activity active operations',
    ),
    processUptimeSeconds: requireCount(
      record.processUptimeSeconds,
      'Runtime Host activity process uptime',
    ),
    residencies: record.residencies.map((value) => {
      const residency = requireExactRecord(value, 'Runtime Host activity residency', [
        'label',
        'count',
      ]);
      return {
        label: requireString(residency.label, 'Runtime Host activity residency label', 128),
        count: requireCount(residency.count, 'Runtime Host activity residency count'),
      };
    }),
  };
}

function decodeHostUpgradePrepareInput(value: unknown): HostUpgradePrepareInput {
  const record = requireShapedRecord(
    value,
    'Runtime Host upgrade prepare input',
    ['expectedHostEpoch', 'allowInterruptActiveTasks'],
    ['allowCooperativeHandoff'],
  );
  return {
    expectedHostEpoch: requireId(record.expectedHostEpoch, 'Runtime Host expected Host Epoch'),
    allowInterruptActiveTasks: requireBoolean(
      record.allowInterruptActiveTasks,
      'Runtime Host upgrade interrupt authority',
    ),
    ...(record.allowCooperativeHandoff === undefined
      ? {}
      : {
          allowCooperativeHandoff: requireBoolean(
            record.allowCooperativeHandoff,
            'Runtime Host cooperative handoff authority',
          ),
        }),
  };
}

function requireBoolean(value: unknown, label: string): boolean {
  if (typeof value !== 'boolean') throw invalidProtocolFrame(`Invalid ${label}`);
  return value;
}

function decodeHostUpgradePrepareResult(value: unknown): HostUpgradePrepareResult {
  const result = requireRecord(value, 'Runtime Host upgrade prepare result');
  if (result.kind === 'active_tasks') {
    requireExactRecord(value, 'Runtime Host upgrade prepare result', ['kind']);
    return { kind: 'active_tasks' };
  }
  if (result.kind !== 'prepared')
    throw invalidProtocolFrame('Invalid Runtime Host upgrade prepare result kind');
  const record = requireExactRecord(value, 'Runtime Host upgrade prepare result', ['kind', 'pid']);
  const pid = requireCount(record.pid, 'Runtime Host upgrade process id');
  if (pid === 0) throw invalidProtocolFrame('Invalid Runtime Host upgrade process id');
  return { kind: 'prepared', pid };
}

function decodeHostStatusFields(record: Record<string, unknown>): HostStatusResult {
  return {
    hostEpoch: requireId(record.hostEpoch, 'hostEpoch'),
    compositionId: requireString(record.compositionId, 'Runtime Host composition id', 128),
    compositionRevision: requireString(
      record.compositionRevision,
      'Runtime Host composition revision',
      128,
    ),
    state: requireHostLifecycleState(record.state),
    connections: requireCount(record.connections, 'connections'),
    activeOperations: requireCount(record.activeOperations, 'activeOperations'),
    activeResidencies: requireCount(record.activeResidencies, 'activeResidencies'),
    ...(record.peerEndpoint === undefined
      ? {}
      : { peerEndpoint: decodeHostPeerEndpoint(record.peerEndpoint) }),
  };
}

function decodeHostPeerEndpoint(value: unknown): HostPeerEndpoint {
  try {
    return decodeSignedPeerReachabilityLease(value);
  } catch {
    throw invalidProtocolFrame('Invalid Runtime Host peer reachability lease');
  }
}

function requirePlatform(value: unknown): NodeJS.Platform {
  if (
    value === 'aix' ||
    value === 'android' ||
    value === 'darwin' ||
    value === 'freebsd' ||
    value === 'haiku' ||
    value === 'linux' ||
    value === 'openbsd' ||
    value === 'sunos' ||
    value === 'win32' ||
    value === 'cygwin' ||
    value === 'netbsd'
  ) {
    return value;
  }
  throw invalidProtocolFrame('Invalid Runtime Host platform');
}

export function requireHostLifecycleState(value: unknown): HostLifecycleState {
  if (
    value === 'starting' ||
    value === 'containing' ||
    value === 'recovering' ||
    value === 'ready' ||
    value === 'draining'
  ) {
    return value;
  }
  throw invalidProtocolFrame('Invalid Host state');
}