/** * 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. */ 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 | 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 | null }} [opts] */ constructor(swarm, opts = {}) { this.swarm = swarm this._cfg = readSwarmPolicyEnv(opts.env) /** @type {Map} */ 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 exp = Math.min( this._cfg.banMsMax, this._cfg.banMsInitial * Math.pow(2, Math.min(8, st.banCount - 1)) ) st.banUntilMs = now + exp st.consecutiveFail = 0 } } } /** * @param {string} peerKey * @returns {boolean} false when peer is in ban window or reconnect budget exhausted */ shouldAttemptPeer(peerKey) { const k = this._key(peerKey) 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) ) } }