mirror of
https://github.com/PR0M3TH3AN/TORCH.git
synced 2026-10-27 07:51:52 +00:00
feat: Decompose src/lock-ops.mjs into relay-health-manager.mjs and lock-publisher.mjs
- Extracted `RelayHealthManager` to `src/relay-health-manager.mjs` - Extracted `LockPublisher` to `src/lock-publisher.mjs` - Refactored `src/lock-ops.mjs` to import new modules - Reduced `src/lock-ops.mjs` size significantly (> 500 lines) - Preserved API compatibility (re-exported `defaultHealthManager`) - Verified with lint and tests - Updated `scripts/check-file-size.mjs` report (implicit) Co-authored-by: PR0M3TH3AN <[email protected]>
This commit is contained in:
google-labs-jules[bot]andPR0M3TH3AN committed
1 parent
e10528141d
commit
b8598d1d92
8 files changed
+854
-716
No files matched your search
@@ -0,0 +1,34 @@
|
||||
# Context: Decompose src/lock-ops.mjs
|
||||
|
||||
## File Selection
|
||||
- **File:** `src/lock-ops.mjs`
|
||||
- **Size:** 881 lines (Excess: 581 lines)
|
||||
- **Selection Reason:** Single largest grandfathered source file available for decomposition.
|
||||
- **Recent History:** Last significant commit was a merge (`e1052814`). No recent decomposition or major refactor found in `git log`.
|
||||
|
||||
## Decomposition Plan
|
||||
The file contains three major cohesive blocks:
|
||||
1. **RelayHealthManager**: A class for tracking relay health, scoring, and metrics.
|
||||
2. **LockPublisher**: A class for publishing lock events with retries and fallback logic.
|
||||
3. **Lock Operations**: The core `queryLocks` and `parseLockEvent` logic.
|
||||
|
||||
### Extraction Strategy
|
||||
1. **Extract `RelayHealthManager`** into `src/relay-health-manager.mjs`.
|
||||
- Includes: `RelayHealthManager` class, `defaultHealthManager` instance, `summarizeRelayMetrics`, `computeRelayScore`, `buildRelayHealthConfig`.
|
||||
- Exports: `RelayHealthManager`, `defaultHealthManager`, `buildRelayHealthConfig`.
|
||||
|
||||
2. **Extract `LockPublisher`** into `src/lock-publisher.mjs`.
|
||||
- Includes: `LockPublisher` class, `publishLock`, `publishToRelays`.
|
||||
- Includes: `classifyPublishError` and related constants (`PUBLISH_ERROR_CODES`, `PUBLISH_FAILURE_CATEGORIES`).
|
||||
- Includes: `secureRandom`, `calculateBackoffDelayMs`.
|
||||
- Internal Helpers (duplicated): `withTimeout`, `relayListLabel`.
|
||||
- Exports: `publishLock`, `LockPublisher`.
|
||||
|
||||
3. **Refactor `src/lock-ops.mjs`**.
|
||||
- Import `defaultHealthManager`, `buildRelayHealthConfig`, `publishLock`.
|
||||
- Remove extracted code.
|
||||
- Retain `queryLocks`, `parseLockEvent`, `filterActiveLocks`.
|
||||
- Retain `withTimeout` and `relayListLabel` (for local use).
|
||||
|
||||
## Goal
|
||||
Reduce `src/lock-ops.mjs` by > 200 lines.
|
||||
@@ -0,0 +1,18 @@
|
||||
# Decisions: Decompose src/lock-ops.mjs
|
||||
|
||||
## 1. Duplicate Internal Helpers (`withTimeout`, `relayListLabel`)
|
||||
- **Context:** Both `src/lock-ops.mjs` and the new `src/lock-publisher.mjs` need `withTimeout` and `relayListLabel`.
|
||||
- **Option A:** Move them to `src/utils.mjs`.
|
||||
- **Option B:** Duplicate them in the new file.
|
||||
- **Decision:** **Option B (Duplicate)**.
|
||||
- **Rationale:** The prompt emphasizes "One file per PR only" and "Do not decompose multiple files". Modifying `src/utils.mjs` would touch a second existing file, potentially expanding the scope and risk. Duplicating these small helpers keeps the changes isolated to the decomposition of `src/lock-ops.mjs` and its new children.
|
||||
|
||||
## 2. Shared Constants
|
||||
- **Context:** Extracted files need constants from `src/constants.mjs` and `src/torch-config.mjs`.
|
||||
- **Decision:** Import them directly in the new files.
|
||||
- **Rationale:** Standard module practice.
|
||||
|
||||
## 3. Class Extraction
|
||||
- **Context:** `RelayHealthManager` and `LockPublisher` are large, distinct classes.
|
||||
- **Decision:** Extract each into its own file (`src/relay-health-manager.mjs`, `src/lock-publisher.mjs`).
|
||||
- **Rationale:** Improves modularity and reduces the size of `src/lock-ops.mjs` significantly.
|
||||
+9
-716
@@ -1,22 +1,16 @@
|
||||
import { randomInt, randomUUID } from 'node:crypto';
|
||||
import { SimplePool } from 'nostr-tools/pool';
|
||||
import {
|
||||
getQueryTimeoutMs,
|
||||
getPublishTimeoutMs,
|
||||
getMinSuccessfulRelayPublishes,
|
||||
getRelayFallbacks,
|
||||
getMinActiveRelayPool,
|
||||
} from './torch-config.mjs';
|
||||
import {
|
||||
KIND_APP_DATA,
|
||||
DEFAULT_ROLLING_WINDOW_SIZE,
|
||||
DEFAULT_FAILURE_THRESHOLD,
|
||||
DEFAULT_QUARANTINE_COOLDOWN_MS,
|
||||
DEFAULT_MAX_QUARANTINE_COOLDOWN_MS,
|
||||
DEFAULT_SNAPSHOT_INTERVAL_MS,
|
||||
MS_PER_SECOND,
|
||||
} from './constants.mjs';
|
||||
import { nowUnix } from './utils.mjs';
|
||||
import { defaultHealthManager, buildRelayHealthConfig, RelayHealthManager } from './relay-health-manager.mjs';
|
||||
import { publishLock, LockPublisher, secureRandom } from './lock-publisher.mjs';
|
||||
|
||||
/**
|
||||
* Parses a raw Nostr event into a structured lock object.
|
||||
@@ -75,387 +69,6 @@ function relayListLabel(relays) {
|
||||
return relays.join(', ');
|
||||
}
|
||||
|
||||
const PUBLISH_ERROR_CODES = {
|
||||
TIMEOUT: 'publish_timeout',
|
||||
DNS: 'dns_resolution',
|
||||
TCP: 'tcp_connect_timeout',
|
||||
TLS: 'tls_handshake',
|
||||
WEBSOCKET: 'websocket_open_failure',
|
||||
NETWORK: 'network_timeout',
|
||||
CONNECTION_RESET: 'connection_reset',
|
||||
RELAY_UNAVAILABLE: 'relay_unavailable',
|
||||
PERMANENT: 'permanent_validation_error',
|
||||
};
|
||||
|
||||
const PUBLISH_FAILURE_CATEGORIES = {
|
||||
QUORUM_FAILURE: 'relay_publish_quorum_failure',
|
||||
NON_RETRYABLE: 'relay_publish_non_retryable',
|
||||
};
|
||||
|
||||
/**
|
||||
* Classifies a raw error message into a standardized publication error code.
|
||||
* Used to determine if a failure is transient (retryable) or permanent.
|
||||
*
|
||||
* @param {string|Error} message - The error message or object to classify.
|
||||
* @returns {string} One of the PUBLISH_ERROR_CODES constants.
|
||||
*/
|
||||
function classifyPublishError(message) {
|
||||
const normalized = String(message || '').toLowerCase();
|
||||
if (normalized.includes('publish timed out after') || normalized.includes('publish timeout')) {
|
||||
return PUBLISH_ERROR_CODES.TIMEOUT;
|
||||
}
|
||||
if (
|
||||
normalized.includes('enotfound')
|
||||
|| normalized.includes('eai_again')
|
||||
|| normalized.includes('getaddrinfo')
|
||||
|| (normalized.includes('dns') && normalized.includes('websocket'))
|
||||
) {
|
||||
return PUBLISH_ERROR_CODES.DNS;
|
||||
}
|
||||
if (
|
||||
normalized.includes('connect etimedout')
|
||||
|| normalized.includes('tcp connect timed out')
|
||||
|| normalized.includes('connect timeout')
|
||||
) {
|
||||
return PUBLISH_ERROR_CODES.TCP;
|
||||
}
|
||||
if (
|
||||
normalized.includes('tls')
|
||||
|| normalized.includes('ssl')
|
||||
|| normalized.includes('certificate')
|
||||
|| normalized.includes('handshake')
|
||||
) {
|
||||
return PUBLISH_ERROR_CODES.TLS;
|
||||
}
|
||||
if (
|
||||
normalized.includes('websocket')
|
||||
|| normalized.includes('bad response')
|
||||
|| normalized.includes('unexpected server response')
|
||||
) {
|
||||
return PUBLISH_ERROR_CODES.WEBSOCKET;
|
||||
}
|
||||
if (normalized.includes('timed out') || normalized.includes('timeout') || normalized.includes('etimedout')) {
|
||||
return PUBLISH_ERROR_CODES.NETWORK;
|
||||
}
|
||||
if (normalized.includes('econnreset') || normalized.includes('connection reset') || normalized.includes('socket hang up')) {
|
||||
return PUBLISH_ERROR_CODES.CONNECTION_RESET;
|
||||
}
|
||||
if (
|
||||
normalized.includes('unavailable')
|
||||
|| normalized.includes('offline')
|
||||
|| normalized.includes('econnrefused')
|
||||
|| normalized.includes('connection refused')
|
||||
|| normalized.includes('enotfound')
|
||||
|| normalized.includes('503')
|
||||
) {
|
||||
return PUBLISH_ERROR_CODES.RELAY_UNAVAILABLE;
|
||||
}
|
||||
return PUBLISH_ERROR_CODES.PERMANENT;
|
||||
}
|
||||
|
||||
function isTransientPublishCategory(category) {
|
||||
return [
|
||||
PUBLISH_ERROR_CODES.TIMEOUT,
|
||||
PUBLISH_ERROR_CODES.DNS,
|
||||
PUBLISH_ERROR_CODES.TCP,
|
||||
PUBLISH_ERROR_CODES.TLS,
|
||||
PUBLISH_ERROR_CODES.WEBSOCKET,
|
||||
PUBLISH_ERROR_CODES.NETWORK,
|
||||
PUBLISH_ERROR_CODES.CONNECTION_RESET,
|
||||
PUBLISH_ERROR_CODES.RELAY_UNAVAILABLE,
|
||||
].includes(category);
|
||||
}
|
||||
|
||||
const MAX_RANDOM = 281474976710655; // 2**48 - 1
|
||||
|
||||
function secureRandom() {
|
||||
return randomInt(0, MAX_RANDOM) / MAX_RANDOM;
|
||||
}
|
||||
|
||||
function calculateBackoffDelayMs(attemptNumber, baseMs, capMs, randomFn = secureRandom) {
|
||||
const maxDelay = Math.min(capMs, baseMs * (2 ** Math.max(0, attemptNumber - 1)));
|
||||
return Math.floor(randomFn() * maxDelay);
|
||||
}
|
||||
|
||||
function summarizeRelayMetrics(metrics, nowMs) {
|
||||
const total = metrics.recentOutcomes.length;
|
||||
const successCount = metrics.recentOutcomes.filter((entry) => entry.success).length;
|
||||
const timeoutCount = metrics.recentOutcomes.filter((entry) => entry.timedOut).length;
|
||||
const latencyEntries = metrics.recentOutcomes.map((entry) => entry.latencyMs).filter((v) => Number.isFinite(v));
|
||||
const averageLatencyMs = latencyEntries.length
|
||||
? Math.round(latencyEntries.reduce((sum, value) => sum + value, 0) / latencyEntries.length)
|
||||
: null;
|
||||
const successRate = total > 0 ? successCount / total : 0.5;
|
||||
const timeoutRate = total > 0 ? timeoutCount / total : 0;
|
||||
const quarantineRemainingMs = metrics.quarantineUntil > nowMs ? metrics.quarantineUntil - nowMs : 0;
|
||||
return {
|
||||
relay: metrics.relay,
|
||||
sampleSize: total,
|
||||
successRate,
|
||||
timeoutRate,
|
||||
averageLatencyMs,
|
||||
failureStreak: metrics.failureStreak,
|
||||
quarantined: quarantineRemainingMs > 0,
|
||||
quarantineRemainingMs,
|
||||
cooldownMs: metrics.cooldownMs,
|
||||
lastResultAt: metrics.lastResultAt,
|
||||
};
|
||||
}
|
||||
|
||||
function computeRelayScore(summary) {
|
||||
const latencyPenalty = summary.averageLatencyMs === null
|
||||
? 0.1
|
||||
: Math.min(0.35, summary.averageLatencyMs / 10_000);
|
||||
const quarantinePenalty = summary.quarantined ? 1 : 0;
|
||||
return (summary.successRate * 1.4) - (summary.timeoutRate * 0.9) - latencyPenalty - quarantinePenalty;
|
||||
}
|
||||
|
||||
/**
|
||||
* Manages health metrics, scoring, and prioritization for Nostr relays.
|
||||
* Tracks success rates, timeouts, and latency to optimize relay selection.
|
||||
* Implements a quarantine mechanism for failing relays.
|
||||
*/
|
||||
export class RelayHealthManager {
|
||||
constructor() {
|
||||
this.metricsByRelay = new Map();
|
||||
this.lastSnapshotAt = 0;
|
||||
this._metricsVersion = 0;
|
||||
this._sortedCache = null;
|
||||
}
|
||||
|
||||
ensureMetrics(relay, config) {
|
||||
let metrics = this.metricsByRelay.get(relay);
|
||||
if (!metrics) {
|
||||
metrics = {
|
||||
relay,
|
||||
recentOutcomes: [],
|
||||
failureStreak: 0,
|
||||
quarantineUntil: 0,
|
||||
cooldownMs: config.quarantineCooldownMs,
|
||||
lastLatencyMs: null,
|
||||
lastResultAt: null,
|
||||
};
|
||||
this.metricsByRelay.set(relay, metrics);
|
||||
this._metricsVersion += 1;
|
||||
}
|
||||
return metrics;
|
||||
}
|
||||
|
||||
/**
|
||||
* Ranks a list of relays based on their health scores.
|
||||
* Utilizes a cached sort order to avoid re-sorting on every call if metrics haven't changed.
|
||||
*
|
||||
* @param {string[]} relays - List of relay URLs to rank.
|
||||
* @param {Object} config - Health configuration.
|
||||
* @param {number} [nowMs] - Current timestamp.
|
||||
* @returns {Array} List of ranked relay entries with scores and summaries.
|
||||
*/
|
||||
rankRelays(relays, config, nowMs = Date.now()) {
|
||||
for (const relay of relays) {
|
||||
this.ensureMetrics(relay, config);
|
||||
}
|
||||
|
||||
// Check if the cached sort order is still valid
|
||||
if (
|
||||
!this._sortedCache
|
||||
|| this._sortedCache.version !== this._metricsVersion
|
||||
|| nowMs >= this._sortedCache.validUntil
|
||||
) {
|
||||
let minQuarantineUntil = Infinity;
|
||||
const entries = [];
|
||||
|
||||
// Re-evaluate scores for all tracked relays
|
||||
for (const metrics of this.metricsByRelay.values()) {
|
||||
const summary = summarizeRelayMetrics(metrics, nowMs);
|
||||
const score = computeRelayScore(summary);
|
||||
if (metrics.quarantineUntil > nowMs && metrics.quarantineUntil < minQuarantineUntil) {
|
||||
minQuarantineUntil = metrics.quarantineUntil;
|
||||
}
|
||||
entries.push({
|
||||
relay: metrics.relay,
|
||||
summary,
|
||||
score,
|
||||
metrics,
|
||||
});
|
||||
}
|
||||
|
||||
// Sort by score (desc), then quarantine status, then latency (asc), then name
|
||||
const comparator = (a, b) => {
|
||||
if (b.score !== a.score) return b.score - a.score;
|
||||
if (a.summary.quarantined !== b.summary.quarantined) return a.summary.quarantined ? 1 : -1;
|
||||
if (a.summary.averageLatencyMs !== b.summary.averageLatencyMs) {
|
||||
if (a.summary.averageLatencyMs === null) return 1;
|
||||
if (b.summary.averageLatencyMs === null) return -1;
|
||||
return a.summary.averageLatencyMs - b.summary.averageLatencyMs;
|
||||
}
|
||||
return a.relay.localeCompare(b.relay);
|
||||
};
|
||||
entries.sort(comparator);
|
||||
|
||||
this._sortedCache = {
|
||||
version: this._metricsVersion,
|
||||
validUntil: minQuarantineUntil,
|
||||
entries,
|
||||
byRelay: new Map(entries.map((e) => [e.relay, e])),
|
||||
comparator,
|
||||
};
|
||||
}
|
||||
|
||||
// Optimization: if requesting a small subset of total known relays,
|
||||
// look them up directly instead of filtering the entire list.
|
||||
// Threshold: if requested set is smaller than 50% of cached entries.
|
||||
if (relays.length > 0 && relays.length * 2 < this._sortedCache.entries.length) {
|
||||
const subset = [];
|
||||
for (const relay of relays) {
|
||||
const entry = this._sortedCache.byRelay.get(relay);
|
||||
if (entry) {
|
||||
subset.push(entry);
|
||||
}
|
||||
}
|
||||
// Re-sort the subset to maintain ranking order
|
||||
subset.sort(this._sortedCache.comparator);
|
||||
|
||||
return subset.map((entry) => {
|
||||
const quarantineRemainingMs = entry.metrics.quarantineUntil > nowMs
|
||||
? entry.metrics.quarantineUntil - nowMs
|
||||
: 0;
|
||||
return {
|
||||
relay: entry.relay,
|
||||
score: entry.score,
|
||||
summary: {
|
||||
...entry.summary,
|
||||
quarantineRemainingMs,
|
||||
},
|
||||
};
|
||||
});
|
||||
}
|
||||
|
||||
const requestedSet = new Set(relays);
|
||||
return this._sortedCache.entries
|
||||
.filter((entry) => requestedSet.has(entry.relay))
|
||||
.map((entry) => {
|
||||
const quarantineRemainingMs = entry.metrics.quarantineUntil > nowMs
|
||||
? entry.metrics.quarantineUntil - nowMs
|
||||
: 0;
|
||||
return {
|
||||
relay: entry.relay,
|
||||
score: entry.score,
|
||||
summary: {
|
||||
...entry.summary,
|
||||
quarantineRemainingMs,
|
||||
},
|
||||
};
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Selects a subset of relays for immediate use, prioritizing healthy ones.
|
||||
* May include quarantined relays if the number of healthy relays is below `minActiveRelayPool`.
|
||||
*
|
||||
* @param {string[]} relays - Candidate relays.
|
||||
* @param {Object} config - Health configuration.
|
||||
* @param {number} [nowMs] - Current timestamp.
|
||||
* @returns {Object} An object containing `prioritized` (list of selected relay URLs) and `ranked` (full ranking details).
|
||||
*/
|
||||
prioritizeRelays(relays, config, nowMs = Date.now()) {
|
||||
const minActive = Math.max(1, Math.min(config.minActiveRelayPool, relays.length || 1));
|
||||
const ranked = this.rankRelays([...new Set(relays)], config, nowMs);
|
||||
const active = ranked.filter((entry) => !entry.summary.quarantined);
|
||||
const quarantined = ranked
|
||||
.filter((entry) => entry.summary.quarantined)
|
||||
.sort((a, b) => a.summary.quarantineRemainingMs - b.summary.quarantineRemainingMs);
|
||||
|
||||
const selected = [...active];
|
||||
if (selected.length < minActive) {
|
||||
const additionalNeeded = Math.min(minActive - selected.length, quarantined.length);
|
||||
selected.push(...quarantined.slice(0, additionalNeeded));
|
||||
}
|
||||
|
||||
return {
|
||||
prioritized: selected.map((entry) => entry.relay),
|
||||
ranked,
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* Updates health metrics for a relay based on the outcome of an operation.
|
||||
*
|
||||
* @param {string} relay - The relay URL.
|
||||
* @param {boolean} success - Whether the operation succeeded.
|
||||
* @param {string} errorMessage - Error message if failed.
|
||||
* @param {number} latencyMs - Duration of the operation.
|
||||
* @param {Object} config - Health configuration.
|
||||
* @param {number} [nowMs] - Current timestamp.
|
||||
*/
|
||||
recordOutcome(relay, success, errorMessage, latencyMs, config, nowMs = Date.now()) {
|
||||
const metrics = this.ensureMetrics(relay, config);
|
||||
const timedOut = String(errorMessage || '').toLowerCase().includes('timeout')
|
||||
|| String(errorMessage || '').toLowerCase().includes('timed out');
|
||||
metrics.recentOutcomes.push({ success, timedOut, latencyMs, atMs: nowMs });
|
||||
if (metrics.recentOutcomes.length > config.rollingWindowSize) {
|
||||
metrics.recentOutcomes.splice(0, metrics.recentOutcomes.length - config.rollingWindowSize);
|
||||
}
|
||||
metrics.lastResultAt = nowMs;
|
||||
metrics.lastLatencyMs = Number.isFinite(latencyMs) ? latencyMs : null;
|
||||
|
||||
if (success) {
|
||||
metrics.failureStreak = 0;
|
||||
if (metrics.quarantineUntil > nowMs) {
|
||||
metrics.quarantineUntil = 0;
|
||||
metrics.cooldownMs = config.quarantineCooldownMs;
|
||||
}
|
||||
return;
|
||||
}
|
||||
|
||||
metrics.failureStreak += 1;
|
||||
if (metrics.failureStreak >= config.failureThreshold) {
|
||||
metrics.quarantineUntil = nowMs + metrics.cooldownMs;
|
||||
metrics.cooldownMs = Math.min(config.maxQuarantineCooldownMs, Math.floor(metrics.cooldownMs * 1.5));
|
||||
}
|
||||
this._metricsVersion += 1;
|
||||
}
|
||||
|
||||
collectSnapshot(relays, config, nowMs = Date.now()) {
|
||||
const uniqueRelays = [...new Set(relays)];
|
||||
return this.rankRelays(uniqueRelays, config, nowMs).map((entry) => ({
|
||||
relay: entry.relay,
|
||||
score: Number(entry.score.toFixed(4)),
|
||||
...entry.summary,
|
||||
}));
|
||||
}
|
||||
|
||||
maybeLogSnapshot(relays, config, logger, reason, force = false, nowMs = Date.now()) {
|
||||
const intervalReached = nowMs - this.lastSnapshotAt >= config.snapshotIntervalMs;
|
||||
if (!force && !intervalReached) return;
|
||||
this.lastSnapshotAt = nowMs;
|
||||
logger(JSON.stringify({
|
||||
event: 'relay_health_snapshot',
|
||||
reason,
|
||||
relays: this.collectSnapshot(relays, config, nowMs),
|
||||
}));
|
||||
}
|
||||
|
||||
reset() {
|
||||
this.metricsByRelay.clear();
|
||||
this.lastSnapshotAt = 0;
|
||||
this._metricsVersion = 0;
|
||||
this._sortedCache = null;
|
||||
}
|
||||
}
|
||||
|
||||
export const defaultHealthManager = new RelayHealthManager();
|
||||
|
||||
function buildRelayHealthConfig(deps) {
|
||||
return {
|
||||
rollingWindowSize: deps.rollingWindowSize ?? DEFAULT_ROLLING_WINDOW_SIZE,
|
||||
failureThreshold: deps.failureThreshold ?? DEFAULT_FAILURE_THRESHOLD,
|
||||
quarantineCooldownMs: deps.quarantineCooldownMs ?? DEFAULT_QUARANTINE_COOLDOWN_MS,
|
||||
maxQuarantineCooldownMs: deps.maxQuarantineCooldownMs ?? DEFAULT_MAX_QUARANTINE_COOLDOWN_MS,
|
||||
snapshotIntervalMs: deps.snapshotIntervalMs ?? DEFAULT_SNAPSHOT_INTERVAL_MS,
|
||||
minActiveRelayPool: Math.max(1, deps.minActiveRelayPool ?? 1),
|
||||
};
|
||||
}
|
||||
|
||||
function mergeRelayList(primaryRelays, fallbackRelays) {
|
||||
return [...new Set([...primaryRelays, ...fallbackRelays])];
|
||||
}
|
||||
@@ -546,335 +159,15 @@ export async function queryLocks(relays, cadence, dateStr, namespace, deps = {})
|
||||
}
|
||||
}
|
||||
|
||||
async function publishToRelays(pool, relays, event, publishTimeoutMs, phase) {
|
||||
const publishPromises = pool.publish(relays, event);
|
||||
const settled = await Promise.allSettled(
|
||||
publishPromises.map((publishPromise, index) => withTimeout(
|
||||
publishPromise,
|
||||
publishTimeoutMs,
|
||||
`[${phase}] Publish timed out after ${publishTimeoutMs}ms (relay=${relays[index]})`,
|
||||
)),
|
||||
);
|
||||
|
||||
return settled.map((result, index) => {
|
||||
if (result.status === 'fulfilled') {
|
||||
return {
|
||||
relay: relays[index],
|
||||
success: true,
|
||||
phase,
|
||||
latencyMs: null,
|
||||
};
|
||||
}
|
||||
const reason = result.reason;
|
||||
const message = reason instanceof Error ? reason.message : String(reason ?? 'unknown');
|
||||
return {
|
||||
relay: relays[index],
|
||||
success: false,
|
||||
phase,
|
||||
message,
|
||||
latencyMs: null,
|
||||
};
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Orchestrates the publication of a lock event to multiple relays.
|
||||
* Handles retries, fallback relays, and health-based relay prioritization.
|
||||
*
|
||||
* Cycle:
|
||||
* 1. Attempt to publish to primary relays.
|
||||
* 2. If quorum not met, attempt fallback relays.
|
||||
* 3. If still failing, retry with backoff (if errors are transient).
|
||||
* 4. Report final success or failure.
|
||||
*/
|
||||
export class LockPublisher {
|
||||
/**
|
||||
* @param {string[]} relays - Target primary relays.
|
||||
* @param {Object} event - The lock event to publish.
|
||||
* @param {Object} [deps] - Dependencies and configuration.
|
||||
*/
|
||||
constructor(relays, event, deps = {}) {
|
||||
this.relays = relays;
|
||||
this.event = event;
|
||||
this.deps = deps;
|
||||
this.pool = null;
|
||||
this.healthConfig = null;
|
||||
this.allRelays = [];
|
||||
this.fallbackRelays = [];
|
||||
this.publishTimeoutMs = 0;
|
||||
this.minSuccesses = 0;
|
||||
this.minActiveRelayPool = 0;
|
||||
this.maxAttempts = 0;
|
||||
this.retryBaseDelayMs = 0;
|
||||
this.retryCapDelayMs = 0;
|
||||
this.sleepFn = null;
|
||||
this.randomFn = null;
|
||||
this.telemetryLogger = null;
|
||||
this.healthLogger = null;
|
||||
this.healthManager = null;
|
||||
this.correlationId = null;
|
||||
this.attemptId = null;
|
||||
}
|
||||
|
||||
/**
|
||||
* Executes the publication process with retries.
|
||||
*
|
||||
* @returns {Promise<Object>} The published event if successful.
|
||||
* @throws {Error} If publication quorum is not met after all attempts.
|
||||
*/
|
||||
async publish() {
|
||||
const {
|
||||
poolFactory = () => new SimplePool(),
|
||||
retryAttempts = 4,
|
||||
retryBaseDelayMs = 500,
|
||||
retryCapDelayMs = 8_000,
|
||||
sleepFn = (ms) => new Promise((resolve) => setTimeout(resolve, ms)),
|
||||
randomFn = Math.random,
|
||||
telemetryLogger = console.error,
|
||||
healthLogger = console.error,
|
||||
diagnostics = {},
|
||||
healthManager = defaultHealthManager,
|
||||
} = this.deps;
|
||||
|
||||
this.healthManager = healthManager;
|
||||
this.pool = poolFactory();
|
||||
this.publishTimeoutMs = this.deps.resolvedConfig?.publishTimeoutMs;
|
||||
this.minSuccesses = this.deps.resolvedConfig?.minSuccesses;
|
||||
this.fallbackRelays = (this.deps.resolvedConfig?.fallbackRelays || []).filter((relay) => !this.relays.includes(relay));
|
||||
this.maxAttempts = Math.max(1, Math.floor(retryAttempts));
|
||||
this.healthConfig = buildRelayHealthConfig({
|
||||
...this.deps,
|
||||
minActiveRelayPool: this.deps.resolvedConfig?.minActiveRelayPool,
|
||||
});
|
||||
|
||||
this.retryBaseDelayMs = retryBaseDelayMs;
|
||||
this.retryCapDelayMs = retryCapDelayMs;
|
||||
this.sleepFn = sleepFn;
|
||||
this.randomFn = randomFn;
|
||||
this.telemetryLogger = telemetryLogger;
|
||||
this.healthLogger = healthLogger;
|
||||
this.healthManager = healthManager;
|
||||
this.correlationId = diagnostics.correlationId || randomUUID();
|
||||
this.attemptId = diagnostics.attemptId || randomUUID();
|
||||
|
||||
this.allRelays = mergeRelayList(this.relays, this.fallbackRelays);
|
||||
|
||||
try {
|
||||
this.healthManager.maybeLogSnapshot(this.allRelays, this.healthConfig, this.healthLogger, 'publish:periodic');
|
||||
let lastAttemptState = null;
|
||||
const retryTimeline = [];
|
||||
const overallStartedAt = Date.now();
|
||||
let terminalFailureCategory = PUBLISH_FAILURE_CATEGORIES.NON_RETRYABLE;
|
||||
|
||||
// Retry loop: attempts publication until success or max attempts reached
|
||||
for (let attemptNumber = 1; attemptNumber <= this.maxAttempts; attemptNumber += 1) {
|
||||
const startedAtMs = Date.now();
|
||||
lastAttemptState = await this.executePublishCycle();
|
||||
const elapsedMs = Date.now() - startedAtMs;
|
||||
retryTimeline.push({
|
||||
publishAttempt: attemptNumber,
|
||||
successCount: lastAttemptState.successCount,
|
||||
relayAttemptedCount: lastAttemptState.attempted.size,
|
||||
elapsedMs,
|
||||
});
|
||||
|
||||
if (lastAttemptState.successCount >= this.minSuccesses) {
|
||||
this.logSuccess(attemptNumber, lastAttemptState, retryTimeline, overallStartedAt);
|
||||
return this.event;
|
||||
}
|
||||
|
||||
const { hasTransientFailure, hasPermanentFailure } = this.analyzeFailures(lastAttemptState.failures);
|
||||
const canRetry = attemptNumber < this.maxAttempts && hasTransientFailure && !hasPermanentFailure;
|
||||
|
||||
terminalFailureCategory = this.determineFailureCategory(hasTransientFailure, hasPermanentFailure, attemptNumber);
|
||||
|
||||
if (!canRetry) {
|
||||
break;
|
||||
}
|
||||
|
||||
const nextDelayMs = calculateBackoffDelayMs(attemptNumber, this.retryBaseDelayMs, this.retryCapDelayMs, this.randomFn);
|
||||
this.logRetry(attemptNumber, lastAttemptState.failures, elapsedMs, nextDelayMs);
|
||||
await this.sleepFn(nextDelayMs);
|
||||
}
|
||||
|
||||
this.handleFinalFailure(lastAttemptState, terminalFailureCategory, retryTimeline, overallStartedAt);
|
||||
} finally {
|
||||
this.pool.close(this.allRelays);
|
||||
}
|
||||
}
|
||||
|
||||
async executePublishCycle() {
|
||||
const attempted = new Set();
|
||||
const publishResults = [];
|
||||
|
||||
await this.attemptPhase(this.relays, 'publish:primary', attempted, publishResults);
|
||||
|
||||
let successCount = publishResults.filter((result) => result.success).length;
|
||||
if (successCount < this.minSuccesses && this.fallbackRelays.length > 0) {
|
||||
await this.attemptPhase(this.fallbackRelays, 'publish:fallback', attempted, publishResults);
|
||||
successCount = publishResults.filter((result) => result.success).length;
|
||||
}
|
||||
|
||||
const failures = publishResults
|
||||
.filter((result) => !result.success)
|
||||
.map((result) => ({
|
||||
...result,
|
||||
category: classifyPublishError(result.message),
|
||||
}));
|
||||
|
||||
return { attempted, publishResults, successCount, failures };
|
||||
}
|
||||
|
||||
async attemptPhase(phaseRelays, phaseName, attempted, publishResults) {
|
||||
if (!phaseRelays.length) return;
|
||||
const { prioritized } = this.healthManager.prioritizeRelays(phaseRelays, this.healthConfig);
|
||||
if (!prioritized.length) return;
|
||||
|
||||
console.error(`[${phaseName}] Publishing to ${prioritized.length} relays (${prioritized.join(', ')})...`);
|
||||
|
||||
const startedAtMs = Date.now();
|
||||
for (const relay of prioritized) {
|
||||
attempted.add(relay);
|
||||
}
|
||||
const phaseResults = await publishToRelays(this.pool, prioritized, this.event, this.publishTimeoutMs, phaseName);
|
||||
const elapsedMs = Date.now() - startedAtMs;
|
||||
for (const result of phaseResults) {
|
||||
result.latencyMs = elapsedMs;
|
||||
this.healthManager.recordOutcome(result.relay, result.success, result.message, elapsedMs, this.healthConfig);
|
||||
}
|
||||
publishResults.push(...phaseResults);
|
||||
}
|
||||
|
||||
analyzeFailures(failures) {
|
||||
const hasTransientFailure = failures.some((failure) => isTransientPublishCategory(failure.category));
|
||||
const hasPermanentFailure = failures.some((failure) => !isTransientPublishCategory(failure.category));
|
||||
return { hasTransientFailure, hasPermanentFailure };
|
||||
}
|
||||
|
||||
determineFailureCategory(hasTransientFailure, hasPermanentFailure, attemptNumber) {
|
||||
if (hasTransientFailure && !hasPermanentFailure && attemptNumber >= this.maxAttempts) {
|
||||
return PUBLISH_FAILURE_CATEGORIES.QUORUM_FAILURE;
|
||||
} else if (hasPermanentFailure) {
|
||||
return PUBLISH_FAILURE_CATEGORIES.NON_RETRYABLE;
|
||||
} else {
|
||||
return PUBLISH_FAILURE_CATEGORIES.QUORUM_FAILURE;
|
||||
}
|
||||
}
|
||||
|
||||
logSuccess(attemptNumber, lastAttemptState, retryTimeline, overallStartedAt) {
|
||||
this.telemetryLogger(JSON.stringify({
|
||||
event: 'lock_publish_quorum_met',
|
||||
correlationId: this.correlationId,
|
||||
attemptId: this.attemptId,
|
||||
publishAttempt: attemptNumber,
|
||||
successCount: lastAttemptState.successCount,
|
||||
relayAttemptedCount: lastAttemptState.attempted.size,
|
||||
requiredSuccesses: this.minSuccesses,
|
||||
timeoutMs: this.publishTimeoutMs,
|
||||
retryTimeline,
|
||||
totalElapsedMs: Date.now() - overallStartedAt,
|
||||
}));
|
||||
console.error(
|
||||
` Published to ${lastAttemptState.successCount}/${lastAttemptState.attempted.size} relays `
|
||||
+ `(required=${this.minSuccesses}, timeout=${this.publishTimeoutMs}ms)`,
|
||||
);
|
||||
}
|
||||
|
||||
logRetry(attemptNumber, failures, elapsedMs, nextDelayMs) {
|
||||
for (const failure of failures) {
|
||||
if (!isTransientPublishCategory(failure.category)) continue;
|
||||
this.telemetryLogger(JSON.stringify({
|
||||
event: 'lock_publish_retry',
|
||||
correlationId: this.correlationId,
|
||||
attemptId: this.attemptId,
|
||||
publishAttempt: attemptNumber,
|
||||
relayUrl: failure.relay,
|
||||
errorCategory: failure.category,
|
||||
elapsedMs,
|
||||
nextDelayMs,
|
||||
}));
|
||||
}
|
||||
}
|
||||
|
||||
handleFinalFailure(lastAttemptState, terminalFailureCategory, retryTimeline, overallStartedAt) {
|
||||
const reasonDistribution = {};
|
||||
const failureLines = lastAttemptState.failures
|
||||
.map((result) => {
|
||||
reasonDistribution[result.category] = (reasonDistribution[result.category] || 0) + 1;
|
||||
this.telemetryLogger(JSON.stringify({
|
||||
event: 'lock_publish_failure',
|
||||
correlationId: this.correlationId,
|
||||
attemptId: this.attemptId,
|
||||
relayUrl: result.relay,
|
||||
phase: result.phase,
|
||||
reason: result.category,
|
||||
message: result.message,
|
||||
}));
|
||||
return `${result.relay} (${result.phase}, reason=${result.category}): ${result.message}`;
|
||||
});
|
||||
|
||||
this.telemetryLogger(JSON.stringify({
|
||||
event: 'lock_publish_quorum_failed',
|
||||
correlationId: this.correlationId,
|
||||
attemptId: this.attemptId,
|
||||
errorCategory: terminalFailureCategory,
|
||||
successCount: lastAttemptState.successCount,
|
||||
relayAttemptedCount: lastAttemptState.attempted.size,
|
||||
requiredSuccesses: this.minSuccesses,
|
||||
timeoutMs: this.publishTimeoutMs,
|
||||
attempts: this.maxAttempts,
|
||||
reasonDistribution,
|
||||
retryTimeline,
|
||||
totalElapsedMs: Date.now() - overallStartedAt,
|
||||
}));
|
||||
|
||||
this.healthManager.maybeLogSnapshot(this.allRelays, this.healthConfig, this.healthLogger, 'publish:failure', true);
|
||||
throw new Error(
|
||||
`Failed relay publish quorum in publish phase: ${lastAttemptState.successCount}/${lastAttemptState.attempted.size} successful `
|
||||
+ `(required=${this.minSuccesses}, timeout=${this.publishTimeoutMs}ms, attempts=${this.maxAttempts}, attempt_id=${this.attemptId}, correlation_id=${this.correlationId}, error_category=${terminalFailureCategory}, total_retry_timeline_ms=${Date.now() - overallStartedAt})\n`
|
||||
+ ` retry timeline: ${retryTimeline.map((item) => `#${item.publishAttempt}:${item.elapsedMs}ms`).join(', ')}\n`
|
||||
+ ` ${failureLines.join('\n ')}`,
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* High-level function to publish a lock event.
|
||||
* Initializes a LockPublisher and triggers the publication process.
|
||||
*
|
||||
* @param {string[]} relays - Target relays.
|
||||
* @param {Object} event - Lock event to publish.
|
||||
* @param {Object} [deps] - Dependencies.
|
||||
* @returns {Promise<Object>} The published event.
|
||||
*/
|
||||
export async function publishLock(relays, event, deps = {}) {
|
||||
const {
|
||||
getPublishTimeoutMsFn = getPublishTimeoutMs,
|
||||
getMinSuccessfulRelayPublishesFn = getMinSuccessfulRelayPublishes,
|
||||
getRelayFallbacksFn = getRelayFallbacks,
|
||||
getMinActiveRelayPoolFn = getMinActiveRelayPool,
|
||||
} = deps;
|
||||
|
||||
const [publishTimeoutMs, minSuccesses, fallbackRelays, minActiveRelayPool] = await Promise.all([
|
||||
getPublishTimeoutMsFn(),
|
||||
getMinSuccessfulRelayPublishesFn(),
|
||||
getRelayFallbacksFn(),
|
||||
getMinActiveRelayPoolFn(),
|
||||
]);
|
||||
|
||||
return new LockPublisher(relays, event, {
|
||||
...deps,
|
||||
resolvedConfig: {
|
||||
publishTimeoutMs,
|
||||
minSuccesses,
|
||||
fallbackRelays,
|
||||
minActiveRelayPool,
|
||||
},
|
||||
}).publish();
|
||||
}
|
||||
|
||||
export function _resetRelayHealthState() {
|
||||
defaultHealthManager.reset();
|
||||
}
|
||||
|
||||
export {
|
||||
RelayHealthManager,
|
||||
defaultHealthManager,
|
||||
LockPublisher,
|
||||
publishLock,
|
||||
};
|
||||
|
||||
export const _secureRandom = secureRandom;
|
||||
@@ -0,0 +1,456 @@
|
||||
import { randomInt, randomUUID } from 'node:crypto';
|
||||
import { SimplePool } from 'nostr-tools/pool';
|
||||
import {
|
||||
getPublishTimeoutMs,
|
||||
getMinSuccessfulRelayPublishes,
|
||||
getRelayFallbacks,
|
||||
getMinActiveRelayPool,
|
||||
} from './torch-config.mjs';
|
||||
import { defaultHealthManager, buildRelayHealthConfig } from './relay-health-manager.mjs';
|
||||
|
||||
function withTimeout(promise, timeoutMs, timeoutMessage) {
|
||||
let timeoutHandle;
|
||||
const timeoutPromise = new Promise((_, reject) => {
|
||||
timeoutHandle = setTimeout(() => reject(new Error(timeoutMessage)), timeoutMs);
|
||||
});
|
||||
return Promise.race([promise, timeoutPromise]).finally(() => {
|
||||
if (timeoutHandle) clearTimeout(timeoutHandle);
|
||||
});
|
||||
}
|
||||
|
||||
function relayListLabel(relays) {
|
||||
return relays.join(', ');
|
||||
}
|
||||
|
||||
function mergeRelayList(primaryRelays, fallbackRelays) {
|
||||
return [...new Set([...primaryRelays, ...fallbackRelays])];
|
||||
}
|
||||
|
||||
const PUBLISH_ERROR_CODES = {
|
||||
TIMEOUT: 'publish_timeout',
|
||||
DNS: 'dns_resolution',
|
||||
TCP: 'tcp_connect_timeout',
|
||||
TLS: 'tls_handshake',
|
||||
WEBSOCKET: 'websocket_open_failure',
|
||||
NETWORK: 'network_timeout',
|
||||
CONNECTION_RESET: 'connection_reset',
|
||||
RELAY_UNAVAILABLE: 'relay_unavailable',
|
||||
PERMANENT: 'permanent_validation_error',
|
||||
};
|
||||
|
||||
const PUBLISH_FAILURE_CATEGORIES = {
|
||||
QUORUM_FAILURE: 'relay_publish_quorum_failure',
|
||||
NON_RETRYABLE: 'relay_publish_non_retryable',
|
||||
};
|
||||
|
||||
/**
|
||||
* Classifies a raw error message into a standardized publication error code.
|
||||
* Used to determine if a failure is transient (retryable) or permanent.
|
||||
*
|
||||
* @param {string|Error} message - The error message or object to classify.
|
||||
* @returns {string} One of the PUBLISH_ERROR_CODES constants.
|
||||
*/
|
||||
function classifyPublishError(message) {
|
||||
const normalized = String(message || '').toLowerCase();
|
||||
if (normalized.includes('publish timed out after') || normalized.includes('publish timeout')) {
|
||||
return PUBLISH_ERROR_CODES.TIMEOUT;
|
||||
}
|
||||
if (
|
||||
normalized.includes('enotfound')
|
||||
|| normalized.includes('eai_again')
|
||||
|| normalized.includes('getaddrinfo')
|
||||
|| (normalized.includes('dns') && normalized.includes('websocket'))
|
||||
) {
|
||||
return PUBLISH_ERROR_CODES.DNS;
|
||||
}
|
||||
if (
|
||||
normalized.includes('connect etimedout')
|
||||
|| normalized.includes('tcp connect timed out')
|
||||
|| normalized.includes('connect timeout')
|
||||
) {
|
||||
return PUBLISH_ERROR_CODES.TCP;
|
||||
}
|
||||
if (
|
||||
normalized.includes('tls')
|
||||
|| normalized.includes('ssl')
|
||||
|| normalized.includes('certificate')
|
||||
|| normalized.includes('handshake')
|
||||
) {
|
||||
return PUBLISH_ERROR_CODES.TLS;
|
||||
}
|
||||
if (
|
||||
normalized.includes('websocket')
|
||||
|| normalized.includes('bad response')
|
||||
|| normalized.includes('unexpected server response')
|
||||
) {
|
||||
return PUBLISH_ERROR_CODES.WEBSOCKET;
|
||||
}
|
||||
if (normalized.includes('timed out') || normalized.includes('timeout') || normalized.includes('etimedout')) {
|
||||
return PUBLISH_ERROR_CODES.NETWORK;
|
||||
}
|
||||
if (normalized.includes('econnreset') || normalized.includes('connection reset') || normalized.includes('socket hang up')) {
|
||||
return PUBLISH_ERROR_CODES.CONNECTION_RESET;
|
||||
}
|
||||
if (
|
||||
normalized.includes('unavailable')
|
||||
|| normalized.includes('offline')
|
||||
|| normalized.includes('econnrefused')
|
||||
|| normalized.includes('connection refused')
|
||||
|| normalized.includes('enotfound')
|
||||
|| normalized.includes('503')
|
||||
) {
|
||||
return PUBLISH_ERROR_CODES.RELAY_UNAVAILABLE;
|
||||
}
|
||||
return PUBLISH_ERROR_CODES.PERMANENT;
|
||||
}
|
||||
|
||||
function isTransientPublishCategory(category) {
|
||||
return [
|
||||
PUBLISH_ERROR_CODES.TIMEOUT,
|
||||
PUBLISH_ERROR_CODES.DNS,
|
||||
PUBLISH_ERROR_CODES.TCP,
|
||||
PUBLISH_ERROR_CODES.TLS,
|
||||
PUBLISH_ERROR_CODES.WEBSOCKET,
|
||||
PUBLISH_ERROR_CODES.NETWORK,
|
||||
PUBLISH_ERROR_CODES.CONNECTION_RESET,
|
||||
PUBLISH_ERROR_CODES.RELAY_UNAVAILABLE,
|
||||
].includes(category);
|
||||
}
|
||||
|
||||
const MAX_RANDOM = 281474976710655; // 2**48 - 1
|
||||
|
||||
export function secureRandom() {
|
||||
return randomInt(0, MAX_RANDOM) / MAX_RANDOM;
|
||||
}
|
||||
|
||||
function calculateBackoffDelayMs(attemptNumber, baseMs, capMs, randomFn = secureRandom) {
|
||||
const maxDelay = Math.min(capMs, baseMs * (2 ** Math.max(0, attemptNumber - 1)));
|
||||
return Math.floor(randomFn() * maxDelay);
|
||||
}
|
||||
|
||||
async function publishToRelays(pool, relays, event, publishTimeoutMs, phase) {
|
||||
const publishPromises = pool.publish(relays, event);
|
||||
const settled = await Promise.allSettled(
|
||||
publishPromises.map((publishPromise, index) => withTimeout(
|
||||
publishPromise,
|
||||
publishTimeoutMs,
|
||||
`[${phase}] Publish timed out after ${publishTimeoutMs}ms (relay=${relays[index]})`,
|
||||
)),
|
||||
);
|
||||
|
||||
return settled.map((result, index) => {
|
||||
if (result.status === 'fulfilled') {
|
||||
return {
|
||||
relay: relays[index],
|
||||
success: true,
|
||||
phase,
|
||||
latencyMs: null,
|
||||
};
|
||||
}
|
||||
const reason = result.reason;
|
||||
const message = reason instanceof Error ? reason.message : String(reason ?? 'unknown');
|
||||
return {
|
||||
relay: relays[index],
|
||||
success: false,
|
||||
phase,
|
||||
message,
|
||||
latencyMs: null,
|
||||
};
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Orchestrates the publication of a lock event to multiple relays.
|
||||
* Handles retries, fallback relays, and health-based relay prioritization.
|
||||
*
|
||||
* Cycle:
|
||||
* 1. Attempt to publish to primary relays.
|
||||
* 2. If quorum not met, attempt fallback relays.
|
||||
* 3. If still failing, retry with backoff (if errors are transient).
|
||||
* 4. Report final success or failure.
|
||||
*/
|
||||
export class LockPublisher {
|
||||
/**
|
||||
* @param {string[]} relays - Target primary relays.
|
||||
* @param {Object} event - The lock event to publish.
|
||||
* @param {Object} [deps] - Dependencies and configuration.
|
||||
*/
|
||||
constructor(relays, event, deps = {}) {
|
||||
this.relays = relays;
|
||||
this.event = event;
|
||||
this.deps = deps;
|
||||
this.pool = null;
|
||||
this.healthConfig = null;
|
||||
this.allRelays = [];
|
||||
this.fallbackRelays = [];
|
||||
this.publishTimeoutMs = 0;
|
||||
this.minSuccesses = 0;
|
||||
this.minActiveRelayPool = 0;
|
||||
this.maxAttempts = 0;
|
||||
this.retryBaseDelayMs = 0;
|
||||
this.retryCapDelayMs = 0;
|
||||
this.sleepFn = null;
|
||||
this.randomFn = null;
|
||||
this.telemetryLogger = null;
|
||||
this.healthLogger = null;
|
||||
this.healthManager = null;
|
||||
this.correlationId = null;
|
||||
this.attemptId = null;
|
||||
}
|
||||
|
||||
/**
|
||||
* Executes the publication process with retries.
|
||||
*
|
||||
* @returns {Promise<Object>} The published event if successful.
|
||||
* @throws {Error} If publication quorum is not met after all attempts.
|
||||
*/
|
||||
async publish() {
|
||||
const {
|
||||
poolFactory = () => new SimplePool(),
|
||||
retryAttempts = 4,
|
||||
retryBaseDelayMs = 500,
|
||||
retryCapDelayMs = 8_000,
|
||||
sleepFn = (ms) => new Promise((resolve) => setTimeout(resolve, ms)),
|
||||
randomFn = Math.random,
|
||||
telemetryLogger = console.error,
|
||||
healthLogger = console.error,
|
||||
diagnostics = {},
|
||||
healthManager = defaultHealthManager,
|
||||
} = this.deps;
|
||||
|
||||
this.healthManager = healthManager;
|
||||
this.pool = poolFactory();
|
||||
this.publishTimeoutMs = this.deps.resolvedConfig?.publishTimeoutMs;
|
||||
this.minSuccesses = this.deps.resolvedConfig?.minSuccesses;
|
||||
this.fallbackRelays = (this.deps.resolvedConfig?.fallbackRelays || []).filter((relay) => !this.relays.includes(relay));
|
||||
this.maxAttempts = Math.max(1, Math.floor(retryAttempts));
|
||||
this.healthConfig = buildRelayHealthConfig({
|
||||
...this.deps,
|
||||
minActiveRelayPool: this.deps.resolvedConfig?.minActiveRelayPool,
|
||||
});
|
||||
|
||||
this.retryBaseDelayMs = retryBaseDelayMs;
|
||||
this.retryCapDelayMs = retryCapDelayMs;
|
||||
this.sleepFn = sleepFn;
|
||||
this.randomFn = randomFn;
|
||||
this.telemetryLogger = telemetryLogger;
|
||||
this.healthLogger = healthLogger;
|
||||
this.healthManager = healthManager;
|
||||
this.correlationId = diagnostics.correlationId || randomUUID();
|
||||
this.attemptId = diagnostics.attemptId || randomUUID();
|
||||
|
||||
this.allRelays = mergeRelayList(this.relays, this.fallbackRelays);
|
||||
|
||||
try {
|
||||
this.healthManager.maybeLogSnapshot(this.allRelays, this.healthConfig, this.healthLogger, 'publish:periodic');
|
||||
let lastAttemptState = null;
|
||||
const retryTimeline = [];
|
||||
const overallStartedAt = Date.now();
|
||||
let terminalFailureCategory = PUBLISH_FAILURE_CATEGORIES.NON_RETRYABLE;
|
||||
|
||||
// Retry loop: attempts publication until success or max attempts reached
|
||||
for (let attemptNumber = 1; attemptNumber <= this.maxAttempts; attemptNumber += 1) {
|
||||
const startedAtMs = Date.now();
|
||||
lastAttemptState = await this.executePublishCycle();
|
||||
const elapsedMs = Date.now() - startedAtMs;
|
||||
retryTimeline.push({
|
||||
publishAttempt: attemptNumber,
|
||||
successCount: lastAttemptState.successCount,
|
||||
relayAttemptedCount: lastAttemptState.attempted.size,
|
||||
elapsedMs,
|
||||
});
|
||||
|
||||
if (lastAttemptState.successCount >= this.minSuccesses) {
|
||||
this.logSuccess(attemptNumber, lastAttemptState, retryTimeline, overallStartedAt);
|
||||
return this.event;
|
||||
}
|
||||
|
||||
const { hasTransientFailure, hasPermanentFailure } = this.analyzeFailures(lastAttemptState.failures);
|
||||
const canRetry = attemptNumber < this.maxAttempts && hasTransientFailure && !hasPermanentFailure;
|
||||
|
||||
terminalFailureCategory = this.determineFailureCategory(hasTransientFailure, hasPermanentFailure, attemptNumber);
|
||||
|
||||
if (!canRetry) {
|
||||
break;
|
||||
}
|
||||
|
||||
const nextDelayMs = calculateBackoffDelayMs(attemptNumber, this.retryBaseDelayMs, this.retryCapDelayMs, this.randomFn);
|
||||
this.logRetry(attemptNumber, lastAttemptState.failures, elapsedMs, nextDelayMs);
|
||||
await this.sleepFn(nextDelayMs);
|
||||
}
|
||||
|
||||
this.handleFinalFailure(lastAttemptState, terminalFailureCategory, retryTimeline, overallStartedAt);
|
||||
} finally {
|
||||
this.pool.close(this.allRelays);
|
||||
}
|
||||
}
|
||||
|
||||
async executePublishCycle() {
|
||||
const attempted = new Set();
|
||||
const publishResults = [];
|
||||
|
||||
await this.attemptPhase(this.relays, 'publish:primary', attempted, publishResults);
|
||||
|
||||
let successCount = publishResults.filter((result) => result.success).length;
|
||||
if (successCount < this.minSuccesses && this.fallbackRelays.length > 0) {
|
||||
await this.attemptPhase(this.fallbackRelays, 'publish:fallback', attempted, publishResults);
|
||||
successCount = publishResults.filter((result) => result.success).length;
|
||||
}
|
||||
|
||||
const failures = publishResults
|
||||
.filter((result) => !result.success)
|
||||
.map((result) => ({
|
||||
...result,
|
||||
category: classifyPublishError(result.message),
|
||||
}));
|
||||
|
||||
return { attempted, publishResults, successCount, failures };
|
||||
}
|
||||
|
||||
async attemptPhase(phaseRelays, phaseName, attempted, publishResults) {
|
||||
if (!phaseRelays.length) return;
|
||||
const { prioritized } = this.healthManager.prioritizeRelays(phaseRelays, this.healthConfig);
|
||||
if (!prioritized.length) return;
|
||||
|
||||
console.error(`[${phaseName}] Publishing to ${prioritized.length} relays (${relayListLabel(prioritized)})...`);
|
||||
|
||||
const startedAtMs = Date.now();
|
||||
for (const relay of prioritized) {
|
||||
attempted.add(relay);
|
||||
}
|
||||
const phaseResults = await publishToRelays(this.pool, prioritized, this.event, this.publishTimeoutMs, phaseName);
|
||||
const elapsedMs = Date.now() - startedAtMs;
|
||||
for (const result of phaseResults) {
|
||||
result.latencyMs = elapsedMs;
|
||||
this.healthManager.recordOutcome(result.relay, result.success, result.message, elapsedMs, this.healthConfig);
|
||||
}
|
||||
publishResults.push(...phaseResults);
|
||||
}
|
||||
|
||||
analyzeFailures(failures) {
|
||||
const hasTransientFailure = failures.some((failure) => isTransientPublishCategory(failure.category));
|
||||
const hasPermanentFailure = failures.some((failure) => !isTransientPublishCategory(failure.category));
|
||||
return { hasTransientFailure, hasPermanentFailure };
|
||||
}
|
||||
|
||||
determineFailureCategory(hasTransientFailure, hasPermanentFailure, attemptNumber) {
|
||||
if (hasTransientFailure && !hasPermanentFailure && attemptNumber >= this.maxAttempts) {
|
||||
return PUBLISH_FAILURE_CATEGORIES.QUORUM_FAILURE;
|
||||
} else if (hasPermanentFailure) {
|
||||
return PUBLISH_FAILURE_CATEGORIES.NON_RETRYABLE;
|
||||
} else {
|
||||
return PUBLISH_FAILURE_CATEGORIES.QUORUM_FAILURE;
|
||||
}
|
||||
}
|
||||
|
||||
logSuccess(attemptNumber, lastAttemptState, retryTimeline, overallStartedAt) {
|
||||
this.telemetryLogger(JSON.stringify({
|
||||
event: 'lock_publish_quorum_met',
|
||||
correlationId: this.correlationId,
|
||||
attemptId: this.attemptId,
|
||||
publishAttempt: attemptNumber,
|
||||
successCount: lastAttemptState.successCount,
|
||||
relayAttemptedCount: lastAttemptState.attempted.size,
|
||||
requiredSuccesses: this.minSuccesses,
|
||||
timeoutMs: this.publishTimeoutMs,
|
||||
retryTimeline,
|
||||
totalElapsedMs: Date.now() - overallStartedAt,
|
||||
}));
|
||||
console.error(
|
||||
` Published to ${lastAttemptState.successCount}/${lastAttemptState.attempted.size} relays `
|
||||
+ `(required=${this.minSuccesses}, timeout=${this.publishTimeoutMs}ms)`,
|
||||
);
|
||||
}
|
||||
|
||||
logRetry(attemptNumber, failures, elapsedMs, nextDelayMs) {
|
||||
for (const failure of failures) {
|
||||
if (!isTransientPublishCategory(failure.category)) continue;
|
||||
this.telemetryLogger(JSON.stringify({
|
||||
event: 'lock_publish_retry',
|
||||
correlationId: this.correlationId,
|
||||
attemptId: this.attemptId,
|
||||
publishAttempt: attemptNumber,
|
||||
relayUrl: failure.relay,
|
||||
errorCategory: failure.category,
|
||||
elapsedMs,
|
||||
nextDelayMs,
|
||||
}));
|
||||
}
|
||||
}
|
||||
|
||||
handleFinalFailure(lastAttemptState, terminalFailureCategory, retryTimeline, overallStartedAt) {
|
||||
const reasonDistribution = {};
|
||||
const failureLines = lastAttemptState.failures
|
||||
.map((result) => {
|
||||
reasonDistribution[result.category] = (reasonDistribution[result.category] || 0) + 1;
|
||||
this.telemetryLogger(JSON.stringify({
|
||||
event: 'lock_publish_failure',
|
||||
correlationId: this.correlationId,
|
||||
attemptId: this.attemptId,
|
||||
relayUrl: result.relay,
|
||||
phase: result.phase,
|
||||
reason: result.category,
|
||||
message: result.message,
|
||||
}));
|
||||
return `${result.relay} (${result.phase}, reason=${result.category}): ${result.message}`;
|
||||
});
|
||||
|
||||
this.telemetryLogger(JSON.stringify({
|
||||
event: 'lock_publish_quorum_failed',
|
||||
correlationId: this.correlationId,
|
||||
attemptId: this.attemptId,
|
||||
errorCategory: terminalFailureCategory,
|
||||
successCount: lastAttemptState.successCount,
|
||||
relayAttemptedCount: lastAttemptState.attempted.size,
|
||||
requiredSuccesses: this.minSuccesses,
|
||||
timeoutMs: this.publishTimeoutMs,
|
||||
attempts: this.maxAttempts,
|
||||
reasonDistribution,
|
||||
retryTimeline,
|
||||
totalElapsedMs: Date.now() - overallStartedAt,
|
||||
}));
|
||||
|
||||
this.healthManager.maybeLogSnapshot(this.allRelays, this.healthConfig, this.healthLogger, 'publish:failure', true);
|
||||
throw new Error(
|
||||
`Failed relay publish quorum in publish phase: ${lastAttemptState.successCount}/${lastAttemptState.attempted.size} successful `
|
||||
+ `(required=${this.minSuccesses}, timeout=${this.publishTimeoutMs}ms, attempts=${this.maxAttempts}, attempt_id=${this.attemptId}, correlation_id=${this.correlationId}, error_category=${terminalFailureCategory}, total_retry_timeline_ms=${Date.now() - overallStartedAt})\n`
|
||||
+ ` retry timeline: ${retryTimeline.map((item) => `#${item.publishAttempt}:${item.elapsedMs}ms`).join(', ')}\n`
|
||||
+ ` ${failureLines.join('\n ')}`,
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* High-level function to publish a lock event.
|
||||
* Initializes a LockPublisher and triggers the publication process.
|
||||
*
|
||||
* @param {string[]} relays - Target relays.
|
||||
* @param {Object} event - Lock event to publish.
|
||||
* @param {Object} [deps] - Dependencies.
|
||||
* @returns {Promise<Object>} The published event.
|
||||
*/
|
||||
export async function publishLock(relays, event, deps = {}) {
|
||||
const {
|
||||
getPublishTimeoutMsFn = getPublishTimeoutMs,
|
||||
getMinSuccessfulRelayPublishesFn = getMinSuccessfulRelayPublishes,
|
||||
getRelayFallbacksFn = getRelayFallbacks,
|
||||
getMinActiveRelayPoolFn = getMinActiveRelayPool,
|
||||
} = deps;
|
||||
|
||||
const [publishTimeoutMs, minSuccesses, fallbackRelays, minActiveRelayPool] = await Promise.all([
|
||||
getPublishTimeoutMsFn(),
|
||||
getMinSuccessfulRelayPublishesFn(),
|
||||
getRelayFallbacksFn(),
|
||||
getMinActiveRelayPoolFn(),
|
||||
]);
|
||||
|
||||
return new LockPublisher(relays, event, {
|
||||
...deps,
|
||||
resolvedConfig: {
|
||||
publishTimeoutMs,
|
||||
minSuccesses,
|
||||
fallbackRelays,
|
||||
minActiveRelayPool,
|
||||
},
|
||||
}).publish();
|
||||
}
|
||||
@@ -0,0 +1,286 @@
|
||||
import {
|
||||
DEFAULT_ROLLING_WINDOW_SIZE,
|
||||
DEFAULT_FAILURE_THRESHOLD,
|
||||
DEFAULT_QUARANTINE_COOLDOWN_MS,
|
||||
DEFAULT_MAX_QUARANTINE_COOLDOWN_MS,
|
||||
DEFAULT_SNAPSHOT_INTERVAL_MS,
|
||||
} from './constants.mjs';
|
||||
|
||||
function summarizeRelayMetrics(metrics, nowMs) {
|
||||
const total = metrics.recentOutcomes.length;
|
||||
const successCount = metrics.recentOutcomes.filter((entry) => entry.success).length;
|
||||
const timeoutCount = metrics.recentOutcomes.filter((entry) => entry.timedOut).length;
|
||||
const latencyEntries = metrics.recentOutcomes.map((entry) => entry.latencyMs).filter((v) => Number.isFinite(v));
|
||||
const averageLatencyMs = latencyEntries.length
|
||||
? Math.round(latencyEntries.reduce((sum, value) => sum + value, 0) / latencyEntries.length)
|
||||
: null;
|
||||
const successRate = total > 0 ? successCount / total : 0.5;
|
||||
const timeoutRate = total > 0 ? timeoutCount / total : 0;
|
||||
const quarantineRemainingMs = metrics.quarantineUntil > nowMs ? metrics.quarantineUntil - nowMs : 0;
|
||||
return {
|
||||
relay: metrics.relay,
|
||||
sampleSize: total,
|
||||
successRate,
|
||||
timeoutRate,
|
||||
averageLatencyMs,
|
||||
failureStreak: metrics.failureStreak,
|
||||
quarantined: quarantineRemainingMs > 0,
|
||||
quarantineRemainingMs,
|
||||
cooldownMs: metrics.cooldownMs,
|
||||
lastResultAt: metrics.lastResultAt,
|
||||
};
|
||||
}
|
||||
|
||||
function computeRelayScore(summary) {
|
||||
const latencyPenalty = summary.averageLatencyMs === null
|
||||
? 0.1
|
||||
: Math.min(0.35, summary.averageLatencyMs / 10_000);
|
||||
const quarantinePenalty = summary.quarantined ? 1 : 0;
|
||||
return (summary.successRate * 1.4) - (summary.timeoutRate * 0.9) - latencyPenalty - quarantinePenalty;
|
||||
}
|
||||
|
||||
/**
|
||||
* Manages health metrics, scoring, and prioritization for Nostr relays.
|
||||
* Tracks success rates, timeouts, and latency to optimize relay selection.
|
||||
* Implements a quarantine mechanism for failing relays.
|
||||
*/
|
||||
export class RelayHealthManager {
|
||||
constructor() {
|
||||
this.metricsByRelay = new Map();
|
||||
this.lastSnapshotAt = 0;
|
||||
this._metricsVersion = 0;
|
||||
this._sortedCache = null;
|
||||
}
|
||||
|
||||
ensureMetrics(relay, config) {
|
||||
let metrics = this.metricsByRelay.get(relay);
|
||||
if (!metrics) {
|
||||
metrics = {
|
||||
relay,
|
||||
recentOutcomes: [],
|
||||
failureStreak: 0,
|
||||
quarantineUntil: 0,
|
||||
cooldownMs: config.quarantineCooldownMs,
|
||||
lastLatencyMs: null,
|
||||
lastResultAt: null,
|
||||
};
|
||||
this.metricsByRelay.set(relay, metrics);
|
||||
this._metricsVersion += 1;
|
||||
}
|
||||
return metrics;
|
||||
}
|
||||
|
||||
/**
|
||||
* Ranks a list of relays based on their health scores.
|
||||
* Utilizes a cached sort order to avoid re-sorting on every call if metrics haven't changed.
|
||||
*
|
||||
* @param {string[]} relays - List of relay URLs to rank.
|
||||
* @param {Object} config - Health configuration.
|
||||
* @param {number} [nowMs] - Current timestamp.
|
||||
* @returns {Array} List of ranked relay entries with scores and summaries.
|
||||
*/
|
||||
rankRelays(relays, config, nowMs = Date.now()) {
|
||||
for (const relay of relays) {
|
||||
this.ensureMetrics(relay, config);
|
||||
}
|
||||
|
||||
// Check if the cached sort order is still valid
|
||||
if (
|
||||
!this._sortedCache
|
||||
|| this._sortedCache.version !== this._metricsVersion
|
||||
|| nowMs >= this._sortedCache.validUntil
|
||||
) {
|
||||
let minQuarantineUntil = Infinity;
|
||||
const entries = [];
|
||||
|
||||
// Re-evaluate scores for all tracked relays
|
||||
for (const metrics of this.metricsByRelay.values()) {
|
||||
const summary = summarizeRelayMetrics(metrics, nowMs);
|
||||
const score = computeRelayScore(summary);
|
||||
if (metrics.quarantineUntil > nowMs && metrics.quarantineUntil < minQuarantineUntil) {
|
||||
minQuarantineUntil = metrics.quarantineUntil;
|
||||
}
|
||||
entries.push({
|
||||
relay: metrics.relay,
|
||||
summary,
|
||||
score,
|
||||
metrics,
|
||||
});
|
||||
}
|
||||
|
||||
// Sort by score (desc), then quarantine status, then latency (asc), then name
|
||||
const comparator = (a, b) => {
|
||||
if (b.score !== a.score) return b.score - a.score;
|
||||
if (a.summary.quarantined !== b.summary.quarantined) return a.summary.quarantined ? 1 : -1;
|
||||
if (a.summary.averageLatencyMs !== b.summary.averageLatencyMs) {
|
||||
if (a.summary.averageLatencyMs === null) return 1;
|
||||
if (b.summary.averageLatencyMs === null) return -1;
|
||||
return a.summary.averageLatencyMs - b.summary.averageLatencyMs;
|
||||
}
|
||||
return a.relay.localeCompare(b.relay);
|
||||
};
|
||||
entries.sort(comparator);
|
||||
|
||||
this._sortedCache = {
|
||||
version: this._metricsVersion,
|
||||
validUntil: minQuarantineUntil,
|
||||
entries,
|
||||
byRelay: new Map(entries.map((e) => [e.relay, e])),
|
||||
comparator,
|
||||
};
|
||||
}
|
||||
|
||||
// Optimization: if requesting a small subset of total known relays,
|
||||
// look them up directly instead of filtering the entire list.
|
||||
// Threshold: if requested set is smaller than 50% of cached entries.
|
||||
if (relays.length > 0 && relays.length * 2 < this._sortedCache.entries.length) {
|
||||
const subset = [];
|
||||
for (const relay of relays) {
|
||||
const entry = this._sortedCache.byRelay.get(relay);
|
||||
if (entry) {
|
||||
subset.push(entry);
|
||||
}
|
||||
}
|
||||
// Re-sort the subset to maintain ranking order
|
||||
subset.sort(this._sortedCache.comparator);
|
||||
|
||||
return subset.map((entry) => {
|
||||
const quarantineRemainingMs = entry.metrics.quarantineUntil > nowMs
|
||||
? entry.metrics.quarantineUntil - nowMs
|
||||
: 0;
|
||||
return {
|
||||
relay: entry.relay,
|
||||
score: entry.score,
|
||||
summary: {
|
||||
...entry.summary,
|
||||
quarantineRemainingMs,
|
||||
},
|
||||
};
|
||||
});
|
||||
}
|
||||
|
||||
const requestedSet = new Set(relays);
|
||||
return this._sortedCache.entries
|
||||
.filter((entry) => requestedSet.has(entry.relay))
|
||||
.map((entry) => {
|
||||
const quarantineRemainingMs = entry.metrics.quarantineUntil > nowMs
|
||||
? entry.metrics.quarantineUntil - nowMs
|
||||
: 0;
|
||||
return {
|
||||
relay: entry.relay,
|
||||
score: entry.score,
|
||||
summary: {
|
||||
...entry.summary,
|
||||
quarantineRemainingMs,
|
||||
},
|
||||
};
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Selects a subset of relays for immediate use, prioritizing healthy ones.
|
||||
* May include quarantined relays if the number of healthy relays is below `minActiveRelayPool`.
|
||||
*
|
||||
* @param {string[]} relays - Candidate relays.
|
||||
* @param {Object} config - Health configuration.
|
||||
* @param {number} [nowMs] - Current timestamp.
|
||||
* @returns {Object} An object containing `prioritized` (list of selected relay URLs) and `ranked` (full ranking details).
|
||||
*/
|
||||
prioritizeRelays(relays, config, nowMs = Date.now()) {
|
||||
const minActive = Math.max(1, Math.min(config.minActiveRelayPool, relays.length || 1));
|
||||
const ranked = this.rankRelays([...new Set(relays)], config, nowMs);
|
||||
const active = ranked.filter((entry) => !entry.summary.quarantined);
|
||||
const quarantined = ranked
|
||||
.filter((entry) => entry.summary.quarantined)
|
||||
.sort((a, b) => a.summary.quarantineRemainingMs - b.summary.quarantineRemainingMs);
|
||||
|
||||
const selected = [...active];
|
||||
if (selected.length < minActive) {
|
||||
const additionalNeeded = Math.min(minActive - selected.length, quarantined.length);
|
||||
selected.push(...quarantined.slice(0, additionalNeeded));
|
||||
}
|
||||
|
||||
return {
|
||||
prioritized: selected.map((entry) => entry.relay),
|
||||
ranked,
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* Updates health metrics for a relay based on the outcome of an operation.
|
||||
*
|
||||
* @param {string} relay - The relay URL.
|
||||
* @param {boolean} success - Whether the operation succeeded.
|
||||
* @param {string} errorMessage - Error message if failed.
|
||||
* @param {number} latencyMs - Duration of the operation.
|
||||
* @param {Object} config - Health configuration.
|
||||
* @param {number} [nowMs] - Current timestamp.
|
||||
*/
|
||||
recordOutcome(relay, success, errorMessage, latencyMs, config, nowMs = Date.now()) {
|
||||
const metrics = this.ensureMetrics(relay, config);
|
||||
const timedOut = String(errorMessage || '').toLowerCase().includes('timeout')
|
||||
|| String(errorMessage || '').toLowerCase().includes('timed out');
|
||||
metrics.recentOutcomes.push({ success, timedOut, latencyMs, atMs: nowMs });
|
||||
if (metrics.recentOutcomes.length > config.rollingWindowSize) {
|
||||
metrics.recentOutcomes.splice(0, metrics.recentOutcomes.length - config.rollingWindowSize);
|
||||
}
|
||||
metrics.lastResultAt = nowMs;
|
||||
metrics.lastLatencyMs = Number.isFinite(latencyMs) ? latencyMs : null;
|
||||
|
||||
if (success) {
|
||||
metrics.failureStreak = 0;
|
||||
if (metrics.quarantineUntil > nowMs) {
|
||||
metrics.quarantineUntil = 0;
|
||||
metrics.cooldownMs = config.quarantineCooldownMs;
|
||||
}
|
||||
return;
|
||||
}
|
||||
|
||||
metrics.failureStreak += 1;
|
||||
if (metrics.failureStreak >= config.failureThreshold) {
|
||||
metrics.quarantineUntil = nowMs + metrics.cooldownMs;
|
||||
metrics.cooldownMs = Math.min(config.maxQuarantineCooldownMs, Math.floor(metrics.cooldownMs * 1.5));
|
||||
}
|
||||
this._metricsVersion += 1;
|
||||
}
|
||||
|
||||
collectSnapshot(relays, config, nowMs = Date.now()) {
|
||||
const uniqueRelays = [...new Set(relays)];
|
||||
return this.rankRelays(uniqueRelays, config, nowMs).map((entry) => ({
|
||||
relay: entry.relay,
|
||||
score: Number(entry.score.toFixed(4)),
|
||||
...entry.summary,
|
||||
}));
|
||||
}
|
||||
|
||||
maybeLogSnapshot(relays, config, logger, reason, force = false, nowMs = Date.now()) {
|
||||
const intervalReached = nowMs - this.lastSnapshotAt >= config.snapshotIntervalMs;
|
||||
if (!force && !intervalReached) return;
|
||||
this.lastSnapshotAt = nowMs;
|
||||
logger(JSON.stringify({
|
||||
event: 'relay_health_snapshot',
|
||||
reason,
|
||||
relays: this.collectSnapshot(relays, config, nowMs),
|
||||
}));
|
||||
}
|
||||
|
||||
reset() {
|
||||
this.metricsByRelay.clear();
|
||||
this.lastSnapshotAt = 0;
|
||||
this._metricsVersion = 0;
|
||||
this._sortedCache = null;
|
||||
}
|
||||
}
|
||||
|
||||
export const defaultHealthManager = new RelayHealthManager();
|
||||
|
||||
export function buildRelayHealthConfig(deps) {
|
||||
return {
|
||||
rollingWindowSize: deps.rollingWindowSize ?? DEFAULT_ROLLING_WINDOW_SIZE,
|
||||
failureThreshold: deps.failureThreshold ?? DEFAULT_FAILURE_THRESHOLD,
|
||||
quarantineCooldownMs: deps.quarantineCooldownMs ?? DEFAULT_QUARANTINE_COOLDOWN_MS,
|
||||
maxQuarantineCooldownMs: deps.maxQuarantineCooldownMs ?? DEFAULT_MAX_QUARANTINE_COOLDOWN_MS,
|
||||
snapshotIntervalMs: deps.snapshotIntervalMs ?? DEFAULT_SNAPSHOT_INTERVAL_MS,
|
||||
minActiveRelayPool: Math.max(1, deps.minActiveRelayPool ?? 1),
|
||||
};
|
||||
}
|
||||
@@ -0,0 +1,29 @@
|
||||
# Test Log: Decompose src/lock-ops.mjs
|
||||
|
||||
## Date
|
||||
2026-02-20
|
||||
|
||||
## Verification Steps
|
||||
|
||||
1. **Linting**
|
||||
- Command: `npm run lint`
|
||||
- Result: Passed (after fixing unused `relayListLabel` in `src/lock-publisher.mjs`).
|
||||
|
||||
2. **Unit Tests**
|
||||
- Command: `npm test`
|
||||
- Result: Passed (294 tests).
|
||||
- Key Tests Covered:
|
||||
- `test/lock-ops.test.mjs` (implicitly covered by keeping API compatible and general suite passing).
|
||||
- `test/run-scheduler-cycle.mjs` (integration).
|
||||
|
||||
3. **File Size Check**
|
||||
- Command: `node scripts/check-file-size.mjs --report`
|
||||
- Result: `src/lock-ops.mjs` is no longer in the report (meaning it is < 300 lines).
|
||||
- Previous Size: 881 lines.
|
||||
- New Size: < 300 lines.
|
||||
- Reduction: > 581 lines.
|
||||
|
||||
## Artifact Verification
|
||||
- `src/relay-health-manager.mjs` created and verified.
|
||||
- `src/lock-publisher.mjs` created and verified.
|
||||
- `src/lock-ops.mjs` refactored and verified.
|
||||
@@ -0,0 +1,8 @@
|
||||
# TODO: Decompose src/lock-ops.mjs
|
||||
|
||||
- [ ] Extract `RelayHealthManager` and helpers to `src/relay-health-manager.mjs`
|
||||
- [ ] Extract `LockPublisher` and helpers to `src/lock-publisher.mjs`
|
||||
- [ ] Refactor `src/lock-ops.mjs` to import new modules
|
||||
- [ ] Verify linting passes
|
||||
- [ ] Verify tests pass (`npm test`)
|
||||
- [ ] Verify file size reduction
|
||||
@@ -0,0 +1,14 @@
|
||||
---
|
||||
agent: decompose-agent
|
||||
cadence: daily
|
||||
date: 2026-02-20
|
||||
platform: jules
|
||||
status: completed
|
||||
---
|
||||
|
||||
# Decompose Agent Run
|
||||
|
||||
- **Target:** `src/lock-ops.mjs`
|
||||
- **Result:** Successfully decomposed into `src/relay-health-manager.mjs` and `src/lock-publisher.mjs`.
|
||||
- **Reduction:** Original 881 lines -> New < 300 lines.
|
||||
- **Verification:** Lint and Tests passed.
|
||||
Reference in new issue
Block a user