bare-operating-system/packages/bare-os-booter/lib/bare-os-swarm-peer-policy.js

/**
 * Production swarm peer policy: EWMA latency, success/fail scoring, exponential backoff,
 * and temporary ban windows for abusive or failing peers.
 *
 * Tunable via optional env (read by caller): BARE_OS_SWARM_BAN_FAIL_THRESHOLD,
 * BARE_OS_SWARM_BAN_MS_INITIAL, BARE_OS_SWARM_BAN_MS_MAX, BARE_OS_SWARM_EWMA_ALPHA.
 * Optional **`BARE_OS_PEER_ALLOWLIST_HEX`** is enforced here so swarm scheduling aligns with
 * **`ctx.bareOsEvaluatePeerAdmission`** (deny before reconnect budget is spent).
 */

import { evaluateBareOsPeerAdmission } from './bare-os-peer-admission.js'

const DEFAULT_FAILS_BEFORE_BAN = 4
const DEFAULT_BAN_MS_INITIAL = 5000
const DEFAULT_BAN_MS_MAX = 300000
const DEFAULT_EWMA_ALPHA = 0.25

/**
 * @param {Record<string, string | undefined> | null | undefined} env
 */
function readSwarmPolicyEnv(env) {
  const fails = Number.parseInt(
    String(env?.BARE_OS_SWARM_BAN_FAIL_THRESHOLD || ''),
    10
  )
  const ban0 = Number.parseInt(
    String(env?.BARE_OS_SWARM_BAN_MS_INITIAL || ''),
    10
  )
  const banMax = Number.parseInt(
    String(env?.BARE_OS_SWARM_BAN_MS_MAX || ''),
    10
  )
  const alpha = Number.parseFloat(
    String(env?.BARE_OS_SWARM_EWMA_ALPHA || '')
  )
  return {
    failsBeforeBan:
      Number.isFinite(fails) && fails >= 1 ? Math.min(32, fails) : DEFAULT_FAILS_BEFORE_BAN,
    banMsInitial:
      Number.isFinite(ban0) && ban0 >= 500
        ? Math.min(DEFAULT_BAN_MS_MAX, ban0)
        : DEFAULT_BAN_MS_INITIAL,
    banMsMax:
      Number.isFinite(banMax) && banMax >= 1000
        ? Math.min(3600000, banMax)
        : DEFAULT_BAN_MS_MAX,
    ewmaAlpha:
      Number.isFinite(alpha) && alpha > 0 && alpha <= 1
        ? alpha
        : DEFAULT_EWMA_ALPHA
  }
}

/**
 * @typedef {{
 *   ok: number,
 *   fail: number,
 *   lastMs: number,
 *   consecutiveFail: number,
 *   banCount: number,
 *   banUntilMs: number,
 *   ewmaLatencyMs: number,
 *   lastLatencyMs: number | null,
 *   reconnectBudget: number
 * }} PeerState
 */

export class BareOsSwarmPeerPolicyEngine {
  /**
   * @param {import('hyperswarm').default | null} swarm
   * @param {{ env?: Record<string, string | undefined> | null }} [opts]
   */
  constructor(swarm, opts = {}) {
    this.swarm = swarm
    this._env = opts.env && typeof opts.env === 'object' ? opts.env : null
    this._cfg = readSwarmPolicyEnv(opts.env)
    /** @type {Map<string, PeerState>} */
    this._peers = new Map()
    /** @type {number} */
    this._globalReconnectBudget = 256
  }

  _key(peerKey) {
    return String(peerKey || '').slice(0, 128) || 'unknown'
  }

  _getOrCreate(k) {
    let st = this._peers.get(k)
    if (!st) {
      st = {
        ok: 0,
        fail: 0,
        lastMs: 0,
        consecutiveFail: 0,
        banCount: 0,
        banUntilMs: 0,
        ewmaLatencyMs: 0,
        lastLatencyMs: null,
        reconnectBudget: 32
      }
      this._peers.set(k, st)
    }
    return st
  }

  /**
   * @param {string} peerKey
   * @param {boolean} ok
   * @param {{ latencyMs?: number }} [probeMeta]
   */
  noteProbe(peerKey, ok, probeMeta = {}) {
    const k = this._key(peerKey)
    const st = this._getOrCreate(k)
    const now = Date.now()
    st.lastMs = now
    if (ok) {
      st.ok++
      st.consecutiveFail = 0
      const lat =
        typeof probeMeta.latencyMs === 'number' && Number.isFinite(probeMeta.latencyMs)
          ? Math.max(0, probeMeta.latencyMs)
          : null
      st.lastLatencyMs = lat
      if (lat != null) {
        const a = this._cfg.ewmaAlpha
        st.ewmaLatencyMs =
          st.ewmaLatencyMs === 0
            ? lat
            : a * lat + (1 - a) * st.ewmaLatencyMs
      }
      if (now >= st.banUntilMs) {
        st.banUntilMs = 0
      }
    } else {
      st.fail++
      st.consecutiveFail++
      if (st.consecutiveFail >= this._cfg.failsBeforeBan) {
        st.banCount++
        const base = Math.min(
          this._cfg.banMsMax,
          this._cfg.banMsInitial * Math.pow(2, Math.min(8, st.banCount - 1))
        )
        const jitter = Math.floor(base * 0.2 * Math.random())
        st.banUntilMs = now + base + jitter
        st.consecutiveFail = 0
      }
    }
  }

  /**
   * @param {string} peerKey
   * @param {{ dhtAddressClass?: string } | null | undefined} [meta]
   * @returns {boolean} false when peer is in ban window or reconnect budget exhausted
   */
  shouldAttemptPeer(peerKey, meta) {
    const k = this._key(peerKey)
    if (this._env) {
      const adm = evaluateBareOsPeerAdmission(this._env, k, meta)
      if (adm.verdict === 'deny') return false
    }
    const st = this._peers.get(k)
    const now = Date.now()
    if (!st) return true
    if (st.banUntilMs > now) return false
    if (st.reconnectBudget <= 0) return false
    return true
  }

  /**
   * Called when initiating a connection attempt (consumes budget).
   * @param {string} peerKey
   */
  consumeReconnectBudget(peerKey) {
    const k = this._key(peerKey)
    const st = this._getOrCreate(k)
    if (st.reconnectBudget > 0) st.reconnectBudget--
    if (this._globalReconnectBudget > 0) this._globalReconnectBudget--
  }

  /**
   * Replenish reconnect budget periodically (e.g. on successful session).
   * @param {string} peerKey
   * @param {number} [amount]
   */
  replenishReconnectBudget(peerKey, amount = 8) {
    const k = this._key(peerKey)
    const st = this._getOrCreate(k)
    st.reconnectBudget = Math.min(32, st.reconnectBudget + amount)
    this._globalReconnectBudget = Math.min(256, this._globalReconnectBudget + amount)
  }

  /**
   * Higher is better for replication throttle hints (0..1 scale, may go slightly negative).
   * @param {string} peerKey
   */
  score(peerKey) {
    const k = this._key(peerKey)
    const st = this._peers.get(k)
    if (!st) return 0
    const total = st.ok + st.fail
    if (!total) return 0
    const successRate = st.ok / total
    const failPenalty = Math.min(0.5, st.fail * 0.05)
    const now = Date.now()
    const banned = st.banUntilMs > now ? 0.35 : 0
    const lat = st.ewmaLatencyMs
    const latPenalty =
      lat > 0 ? Math.min(0.25, Math.log10(1 + lat / 50) * 0.08) : 0
    return successRate - failPenalty - banned - latPenalty
  }

  snapshot() {
    const now = Date.now()
    return {
      schema: 2,
      cfg: { ...this._cfg },
      globalReconnectBudget: this._globalReconnectBudget,
      peers: [...this._peers.entries()].map(([id, st]) => ({
        id,
        ...st,
        banned: st.banUntilMs > now,
        banRemainingMs: st.banUntilMs > now ? st.banUntilMs - now : 0
      })),
      atMs: now
    }
  }

  /**
   * Rank peer keys by score descending (for replication scheduling hints).
   * @param {string[]} peerKeys
   */
  rankPeers(peerKeys) {
    return [...new Set(peerKeys.map((p) => this._key(p)))].sort(
      (a, b) => this.score(b) - this.score(a)
    )
  }
}