Raven Scott
b1736b7a7a
add grouped retry orchestration, duplicate arbitration hardening, and transport-aware peer scoring expand IPC with readiness gates, soft backpressure, request deadlines, batching, and method circuit-breakers strengthen worker/control-plane lifecycle via idempotent termination, orphan detection, and liveness checkpoint helpers improve lifecycle correctness with state-machine + transactional hook infrastructure and adaptive quiesce utilities enrich operator surfaces/tests for degraded readiness, crash-safe telemetry patterns, and contract-level regressions
334 lines
9.9 KiB
JavaScript
334 lines
9.9 KiB
JavaScript
/**
|
|
* 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,
|
|
* earliestRetryAtMs: number,
|
|
* retryTier: 'none' | 'short' | 'medium' | 'long' | 'xlong',
|
|
* failByClass: Record<string, number>,
|
|
* transportByClass: Record<string, 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
|
|
/** @type {{ start: number, n: number } | null} */
|
|
this._attemptBurstWindow = null
|
|
}
|
|
|
|
_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,
|
|
earliestRetryAtMs: 0,
|
|
retryTier: 'none',
|
|
failByClass: {},
|
|
transportByClass: {}
|
|
}
|
|
this._peers.set(k, st)
|
|
}
|
|
return st
|
|
}
|
|
|
|
_retryDelayFor(st) {
|
|
const n = Math.max(0, Number(st.consecutiveFail) || 0)
|
|
const tier =
|
|
n >= 9 ? 'xlong' : n >= 6 ? 'long' : n >= 4 ? 'medium' : n >= 2 ? 'short' : 'none'
|
|
const base =
|
|
tier === 'xlong'
|
|
? 15000
|
|
: tier === 'long'
|
|
? 8000
|
|
: tier === 'medium'
|
|
? 3000
|
|
: tier === 'short'
|
|
? 800
|
|
: 0
|
|
const jitter = base > 0 ? Math.floor(base * 0.3 * Math.random()) : 0
|
|
return { tier, delayMs: base + jitter }
|
|
}
|
|
|
|
/**
|
|
* @param {string} peerKey
|
|
* @param {boolean} ok
|
|
* @param {{ latencyMs?: number, failClass?: string, transportClass?: string }} [probeMeta]
|
|
*/
|
|
noteProbe(peerKey, ok, probeMeta = {}) {
|
|
const k = this._key(peerKey)
|
|
const st = this._getOrCreate(k)
|
|
const now = Date.now()
|
|
st.lastMs = now
|
|
if (ok) {
|
|
const tc = String(probeMeta.transportClass || 'unknown')
|
|
.trim()
|
|
.slice(0, 24) || 'unknown'
|
|
st.transportByClass[tc] = (st.transportByClass[tc] || 0) + 1
|
|
st.ok++
|
|
st.consecutiveFail = 0
|
|
st.retryTier = 'none'
|
|
st.earliestRetryAtMs = 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++
|
|
const failClass = String(probeMeta.failClass || 'unknown')
|
|
.trim()
|
|
.slice(0, 32) || 'unknown'
|
|
st.failByClass[failClass] = (st.failByClass[failClass] || 0) + 1
|
|
const retry = this._retryDelayFor(st)
|
|
st.retryTier = retry.tier
|
|
st.earliestRetryAtMs = now + retry.delayMs
|
|
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
|
|
st.retryTier = 'xlong'
|
|
st.earliestRetryAtMs = st.banUntilMs
|
|
}
|
|
}
|
|
}
|
|
|
|
/**
|
|
* @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 burstMax = Number.parseInt(
|
|
String(this._env.BARE_OS_SWARM_ATTEMPT_BURST_PER_SEC || ''),
|
|
10
|
|
)
|
|
if (Number.isFinite(burstMax) && burstMax > 0) {
|
|
const now = Date.now()
|
|
const w = this._attemptBurstWindow
|
|
if (w && now - w.start <= 1000 && w.n >= burstMax) return false
|
|
}
|
|
}
|
|
const st = this._peers.get(k)
|
|
const now = Date.now()
|
|
if (!st) return true
|
|
if (this._globalReconnectBudget <= 0) return false
|
|
if (st.banUntilMs > now) return false
|
|
if (st.earliestRetryAtMs > 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--
|
|
if (this._env) {
|
|
const burstMax = Number.parseInt(
|
|
String(this._env.BARE_OS_SWARM_ATTEMPT_BURST_PER_SEC || ''),
|
|
10
|
|
)
|
|
if (Number.isFinite(burstMax) && burstMax > 0) {
|
|
const now = Date.now()
|
|
if (
|
|
!this._attemptBurstWindow ||
|
|
now - this._attemptBurstWindow.start > 1000
|
|
) {
|
|
this._attemptBurstWindow = { start: now, n: 0 }
|
|
}
|
|
this._attemptBurstWindow.n++
|
|
}
|
|
}
|
|
}
|
|
|
|
/**
|
|
* 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 protocolPenalty = Math.min(0.25, (st.failByClass.protocol || 0) * 0.03)
|
|
const transportReward = Math.min(
|
|
0.15,
|
|
((st.transportByClass.ipc || 0) +
|
|
(st.transportByClass.tcp || 0) * 0.7 +
|
|
(st.transportByClass.udx || 0) * 0.5) *
|
|
0.01
|
|
)
|
|
const policyPenalty = Math.min(0.2, (st.failByClass.policy || 0) * 0.02)
|
|
const timeoutPenalty = Math.min(0.15, (st.failByClass.timeout || 0) * 0.01)
|
|
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 +
|
|
transportReward -
|
|
failPenalty -
|
|
protocolPenalty -
|
|
policyPenalty -
|
|
timeoutPenalty -
|
|
banned -
|
|
latPenalty
|
|
)
|
|
}
|
|
|
|
snapshot() {
|
|
const now = Date.now()
|
|
return {
|
|
schema: 2,
|
|
cfg: { ...this._cfg },
|
|
attemptBurstEnv: 'BARE_OS_SWARM_ATTEMPT_BURST_PER_SEC',
|
|
attemptBurstWindow: this._attemptBurstWindow
|
|
? { ...this._attemptBurstWindow }
|
|
: null,
|
|
globalReconnectBudget: this._globalReconnectBudget,
|
|
peers: [...this._peers.entries()].map(([id, st]) => ({
|
|
id,
|
|
...st,
|
|
banned: st.banUntilMs > now,
|
|
banRemainingMs: st.banUntilMs > now ? st.banUntilMs - now : 0,
|
|
retryWaitMs:
|
|
st.earliestRetryAtMs > now ? st.earliestRetryAtMs - 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)
|
|
)
|
|
}
|
|
}
|