bare-operating-system/packages/bare-os-booter/lib/bare-os-swarm-connection-manager.js

/**
 * Swarm connection manager: delegates to {@link BareOsSwarmPeerPolicyEngine} for
 * peer health, scoring, backoff/ban windows, and replication scheduling hints.
 */

import { BareOsSwarmPeerPolicyEngine } from './bare-os-swarm-peer-policy.js'

export class BareOsSwarmConnectionManager {
  /**
   * @param {import('hyperswarm').default | null} swarm
   * @param {{ env?: Record<string, string | undefined> | null }} [opts]
   */
  constructor(swarm, opts = {}) {
    this._engine = new BareOsSwarmPeerPolicyEngine(swarm, opts)
    /** Protomux RPC pool–style reuse counter (call {@link recordRpcPoolReuse} from hot paths). */
    this._rpcPoolReuse = 0
    /** @type {Map<string, { openedAtMs: number, lastActivityAtMs: number, initiator: boolean }>} */
    this._activeConnections = new Map()
    this._duplicateDrops = 0
    /** @type {Map<'short'|'medium'|'long'|'xlong', Set<string>>} */
    this._retryBuckets = new Map([
      ['short', new Set()],
      ['medium', new Set()],
      ['long', new Set()],
      ['xlong', new Set()]
    ])
    this._bucketTimer = null
    this._retryDequeued = 0
  }

  /** Increment when a logical RPC channel is reused instead of opened fresh. */
  recordRpcPoolReuse() {
    this._rpcPoolReuse++
  }

  get swarm() {
    return this._engine.swarm
  }

  /**
   * @param {string} peerKey
   * @param {boolean} ok
   * @param {{ latencyMs?: number }} [probeMeta]
   */
  noteProbe(peerKey, ok, probeMeta) {
    this._engine.noteProbe(peerKey, ok, probeMeta)
  }

  /**
   * @param {string} peerKey
   * @param {{ dhtAddressClass?: string } | null | undefined} [meta]
   */
  shouldAttemptPeer(peerKey, meta) {
    return this._engine.shouldAttemptPeer(peerKey, meta)
  }

  /**
   * @param {string} peerKey
   */
  consumeReconnectBudget(peerKey) {
    this._engine.consumeReconnectBudget(peerKey)
  }

  /**
   * @param {string} peerKey
   * @param {number} [amount]
   */
  replenishReconnectBudget(peerKey, amount) {
    this._engine.replenishReconnectBudget(peerKey, amount)
  }

  /**
   * @param {string} peerKey
   */
  score(peerKey) {
    return this._engine.score(peerKey)
  }

  snapshot() {
    const base = this._engine.snapshot()
    return {
      ...base,
      protomuxPoolMetrics: {
        schema: 1,
        rpcPoolReuseCount: this._rpcPoolReuse,
        note: 'Aligns with protomux-rpc-client-pool-style accounting; stock booter increments only when callers invoke recordRpcPoolReuse.'
      },
      duplicateArbitration: {
        schema: 1,
        activeConnectionCount: this._activeConnections.size,
        duplicateDrops: this._duplicateDrops,
        note: 'Deterministic tie-break: keep older open; when equal age keep initiator.'
      },
      groupedRetry: {
        schema: 1,
        queued: {
          short: this._retryBuckets.get('short')?.size || 0,
          medium: this._retryBuckets.get('medium')?.size || 0,
          long: this._retryBuckets.get('long')?.size || 0,
          xlong: this._retryBuckets.get('xlong')?.size || 0
        },
        dequeuedTotal: this._retryDequeued
      }
    }
  }

  /**
   * @param {string[]} peerKeys
   */
  rankPeers(peerKeys) {
    return this._engine.rankPeers(peerKeys)
  }

  /**
   * Deterministic duplicate arbitration. Returns true when caller should keep the incoming edge.
   * @param {string} peerKey
   * @param {{ initiator?: boolean }} [incoming]
   */
  shouldAcceptConnection(peerKey, incoming = {}) {
    const key = String(peerKey || '').slice(0, 128) || 'unknown'
    const now = Date.now()
    const cur = this._activeConnections.get(key)
    if (!cur) return true
    const incomingInitiator = incoming.initiator === true
    if (now - cur.lastActivityAtMs <= 15000) {
      this._duplicateDrops++
      return false
    }
    if (cur.initiator && !incomingInitiator) {
      this._duplicateDrops++
      return false
    }
    return true
  }

  /**
   * @param {string} peerKey
   * @param {{ initiator?: boolean }} [meta]
   */
  markConnectionOpen(peerKey, meta = {}) {
    const key = String(peerKey || '').slice(0, 128) || 'unknown'
    const now = Date.now()
    this._activeConnections.set(key, {
      openedAtMs: now,
      lastActivityAtMs: now,
      initiator: meta.initiator === true
    })
  }

  /**
   * @param {string} peerKey
   */
  markConnectionClosed(peerKey) {
    const key = String(peerKey || '').slice(0, 128) || 'unknown'
    this._activeConnections.delete(key)
  }

  /**
   * @param {string} peerKey
   */
  markConnectionActivity(peerKey) {
    const key = String(peerKey || '').slice(0, 128) || 'unknown'
    const row = this._activeConnections.get(key)
    if (!row) return
    row.lastActivityAtMs = Date.now()
  }

  /**
   * Queue peer for grouped retry scheduling.
   * @param {string} peerKey
   * @param {'short'|'medium'|'long'|'xlong'} [tier]
   */
  queueRetry(peerKey, tier = 'short') {
    const k = String(peerKey || '').slice(0, 128) || 'unknown'
    const t = this._retryBuckets.get(tier)
    if (!t) return
    t.add(k)
    if (this._bucketTimer) return
    this._bucketTimer = setInterval(() => {
      for (const bucket of this._retryBuckets.values()) {
        const first = bucket.values().next()
        if (!first.done) {
          bucket.delete(first.value)
          this._retryDequeued++
          break
        }
      }
      const allEmpty = [...this._retryBuckets.values()].every((b) => b.size === 0)
      if (allEmpty && this._bucketTimer) {
        clearInterval(this._bucketTimer)
        this._bucketTimer = null
      }
    }, 50)
  }

  /**
   * Trigger synthetic keepalive probe accounting for all active peers.
   * @param {boolean} ok
   * @param {{ latencyMs?: number }} [meta]
   */
  emitKeepaliveProbeSweep(ok, meta = {}) {
    for (const key of this._activeConnections.keys()) {
      this.noteProbe(key, ok, {
        latencyMs: meta.latencyMs,
        failClass: ok ? undefined : 'timeout',
        transportClass: 'tcp'
      })
    }
  }
}