3304 lines
135 KiB
JavaScript
3304 lines
135 KiB
JavaScript
require('dotenv').config();
|
|
|
|
// Suppress DEP0060 deprecation warning from http-proxy dependency
|
|
// This is a known issue in http-proxy v1.18.1 that uses deprecated util._extend
|
|
const originalEmitWarning = process.emitWarning;
|
|
process.emitWarning = function(warning, type, code, ...args) {
|
|
if (code === 'DEP0060') {
|
|
return; // Suppress util._extend deprecation warning
|
|
}
|
|
return originalEmitWarning.call(this, warning, type, code, ...args);
|
|
};
|
|
|
|
const Corestore = require('corestore');
|
|
const Hyperswarm = require('hyperswarm');
|
|
const Autopass = require('autopass');
|
|
const Protomux = require('protomux');
|
|
const crypto = require('crypto');
|
|
const fs = require('fs').promises;
|
|
const net = require('net');
|
|
const dgram = require('dgram');
|
|
const ca = require('./includes/security/certificate_authority');
|
|
const { logDebug, logError, logWarn, logInfo } = require('./includes/infrastructure/logger');
|
|
const { getAllEntries, doAutoVotes, removeDomain, getHashForDomain } = require('./includes/core/core');
|
|
const { addDomain, addInternalDomain } = require('./includes/core/domains');
|
|
const { setupProxyServer } = require('./includes/networking/internal_domains_proxy');
|
|
const { cleanupInterfaces, waitForPortRelease } = require('./includes/maintenance/cleanup');
|
|
const { createInterfaceForDomain } = require('./includes/networking/virtual_interfaces');
|
|
const { bindDnsServer } = require('./includes/networking/dns');
|
|
const { setupCache, secondsToMs, loadOrCreateKeypair, getPersistentPublicKey } = require('./includes/infrastructure/utils');
|
|
const { getShutdownConfig, sleep, closeAllHttpTlsServers, killHolesailChildProcess } = require('./includes/infrastructure/shutdown-utils');
|
|
const state = require('./includes/infrastructure/state');
|
|
const { broadcast, loadHolesailServers, startHolesailServer, saveHolesailServers, loadHolesailClients, saveHolesailClients, loadBlockedPeers, loadPeerMetrics, savePeerMetrics, loadPeerHistory, savePeerHistory } = require('./includes/admin');
|
|
const { dnsPool } = require('./includes/networking/dns_pool');
|
|
const { rateLimiter } = require('./includes/infrastructure/rate_limit');
|
|
const { trackPeerEvent, startProcessMetricsCollection } = require('./includes/maintenance/metrics');
|
|
const child_process = require('child_process');
|
|
const path = require('path');
|
|
|
|
/**
|
|
* Check if a peer was previously connected (reconnection detection)
|
|
* @param {string} peerId - Peer ID to check
|
|
* @returns {boolean} True if peer has connection history
|
|
*/
|
|
function isPeerReconnection(peerId) {
|
|
if (!state.peerHistory || !state.peerHistory.has(peerId)) {
|
|
return false;
|
|
}
|
|
const history = state.peerHistory.get(peerId);
|
|
// Check if there's at least one connect event in history
|
|
return history.some(event => event.type === 'connect');
|
|
}
|
|
|
|
|
|
/**
|
|
* Attempt to reconnect to a peer (master node only)
|
|
* Uses Hyperswarm's peer discovery to actively attempt reconnection
|
|
* @param {string} peerId - Peer ID to reconnect to
|
|
* @param {Object} swarm - Hyperswarm instance
|
|
* @param {Buffer} topic - Hyperswarm topic
|
|
* @returns {Promise<void>}
|
|
*/
|
|
async function attemptReconnectToPeer(peerId, swarm, topic) {
|
|
if (!state.peersToReconnect || !state.peersToReconnect.has(peerId)) {
|
|
return; // Peer not in reconnection set
|
|
}
|
|
|
|
const maxAttempts = parseInt(process.env.MASTER_MAX_RECONNECT_ATTEMPTS || '10', 10);
|
|
const baseInterval = parseInt(process.env.MASTER_RECONNECT_INTERVAL || '5', 10) * 1000; // Convert to ms
|
|
|
|
const attempts = state.reconnectionAttempts.get(peerId) || 0;
|
|
const lastAttempt = state.lastReconnectionAttempt.get(peerId) || 0;
|
|
const now = Date.now();
|
|
|
|
// Check if we should attempt reconnection (exponential backoff)
|
|
const backoffDelay = Math.min(baseInterval * Math.pow(2, attempts), 60000); // Max 60 seconds
|
|
if (now - lastAttempt < backoffDelay) {
|
|
// Too soon, schedule for later
|
|
const delay = backoffDelay - (now - lastAttempt);
|
|
setTimeout(() => attemptReconnectToPeer(peerId, swarm, topic), delay);
|
|
return;
|
|
}
|
|
|
|
if (attempts >= maxAttempts) {
|
|
logWarn('Swarm', `Max reconnection attempts (${maxAttempts}) reached for peer ${peerId}, removing from reconnection set`);
|
|
state.peersToReconnect.delete(peerId);
|
|
state.reconnectionAttempts.delete(peerId);
|
|
state.lastReconnectionAttempt.delete(peerId);
|
|
return;
|
|
}
|
|
|
|
// Check if peer is already connected
|
|
if (state.connectedPeers && state.connectedPeers.has(peerId)) {
|
|
logDebug('Swarm', `Peer ${peerId} already connected, removing from reconnection set`);
|
|
state.peersToReconnect.delete(peerId);
|
|
state.reconnectionAttempts.delete(peerId);
|
|
state.lastReconnectionAttempt.delete(peerId);
|
|
return;
|
|
}
|
|
|
|
logInfo('Swarm', `Attempting to reconnect to peer ${peerId} (attempt ${attempts + 1}/${maxAttempts})`);
|
|
state.reconnectionAttempts.set(peerId, attempts + 1);
|
|
state.lastReconnectionAttempt.set(peerId, now);
|
|
|
|
try {
|
|
// Use direct peer reconnect when we already know the peer key.
|
|
swarm.joinPeer(Buffer.from(peerId, 'hex'));
|
|
// Keep the shared topic refreshed as a fallback rediscovery path.
|
|
swarm.join(topic, { server: true, client: true });
|
|
|
|
// Schedule next attempt if still needed
|
|
setTimeout(() => {
|
|
if (state.peersToReconnect && state.peersToReconnect.has(peerId)) {
|
|
attemptReconnectToPeer(peerId, swarm, topic);
|
|
}
|
|
}, backoffDelay);
|
|
} catch (err) {
|
|
logError('Swarm', `Error during reconnection attempt for peer ${peerId}: ${err.message}`);
|
|
// Schedule retry
|
|
setTimeout(() => {
|
|
if (state.peersToReconnect && state.peersToReconnect.has(peerId)) {
|
|
attemptReconnectToPeer(peerId, swarm, topic);
|
|
}
|
|
}, backoffDelay);
|
|
}
|
|
}
|
|
|
|
async function main() {
|
|
try {
|
|
// Validate configuration on startup
|
|
const { validateConfig } = require('./includes/infrastructure/config');
|
|
try {
|
|
validateConfig();
|
|
logInfo('Main', 'Configuration validated successfully');
|
|
} catch (err) {
|
|
logError('Main', `Configuration validation failed: ${err.message}`);
|
|
process.exit(1);
|
|
}
|
|
|
|
// Warn about macOS stealth mode if virtual interfaces are enabled
|
|
if (process.platform === 'darwin' && process.env.DISABLE_VIRTUAL_INTERFACES !== 'true') {
|
|
logWarn('Main', 'macOS detected: If "Stealth Mode" is enabled in System Settings > Network > Firewall Options, ICMP (ping) responses for virtual interfaces may be blocked. Disable stealth mode or configure firewall to allow ICMP if ping functionality is required.');
|
|
}
|
|
|
|
logInfo('Main', 'Starting main function...');
|
|
// Check for flags
|
|
const {
|
|
resolveIsMaster,
|
|
resolveIsGenesis,
|
|
shouldMasterLoadDomains
|
|
} = require('./includes/infrastructure/master-mode');
|
|
const networkManifest = require('./includes/infrastructure/network-manifest');
|
|
const isMaster = resolveIsMaster();
|
|
const isGenesis = resolveIsGenesis();
|
|
state.isMaster = isMaster;
|
|
state.isGenesis = isGenesis;
|
|
const cleanStorage = process.argv.includes('--clean');
|
|
const rawStorageDir = process.env.STORAGE_DIR || './my-storage';
|
|
const storageDir = path.isAbsolute(rawStorageDir)
|
|
? rawStorageDir
|
|
: path.resolve(__dirname, rawStorageDir);
|
|
// Keep STORAGE_DIR canonical so downstream modules/admin paths align.
|
|
process.env.STORAGE_DIR = storageDir;
|
|
const masterViaArgv = process.argv.includes('--master');
|
|
const masterViaEnv = isMaster && !masterViaArgv;
|
|
logInfo('Main', `Master mode: ${isMaster} (argv --master: ${masterViaArgv}${masterViaEnv ? ', env P2NS_MASTER/MASTER: true' : ''})`);
|
|
if (isMaster) {
|
|
if (isGenesis) {
|
|
logInfo('Main', 'Running as MASTER (genesis) — will create or adopt the network Autopass');
|
|
} else {
|
|
logInfo('Main', 'Running as MASTER — invite authority; empty storage pairs into an existing network');
|
|
}
|
|
} else {
|
|
logInfo('Main', 'Running as JOINER — needs a writer peer to receive an invite');
|
|
}
|
|
logInfo('Main', `Clean storage flag detected: ${cleanStorage}`);
|
|
logInfo('Main', `Storage path resolved: ${rawStorageDir} -> ${storageDir}`);
|
|
async function runPluginStorageReset(reason) {
|
|
try {
|
|
const { resetPluginStorage } = require('./scripts/reset-plugin-storage');
|
|
const summary = await resetPluginStorage({
|
|
logger: (level, message) => {
|
|
const fn = level === 'WARN' ? logWarn : level === 'ERROR' ? logError : logInfo;
|
|
fn('Main', `[PluginStorageReset] ${message}`);
|
|
}
|
|
});
|
|
logInfo('Main', `Plugin spec/db reset complete for ${summary.resetCandidates} plugin(s) during ${reason}`);
|
|
} catch (err) {
|
|
logWarn('Main', `Plugin spec/db reset failed during ${reason}: ${err.message}`);
|
|
}
|
|
}
|
|
|
|
const shouldResetStorage = cleanStorage || !isMaster;
|
|
if (shouldResetStorage) {
|
|
logInfo('Main', cleanStorage
|
|
? 'Clean flag detected: resetting my-storage and plugin spec/db...'
|
|
: 'Resetting my-storage on startup for joiner mode (plugin spec/db preserved)...');
|
|
try {
|
|
await fs.rm(storageDir, { recursive: true, force: true });
|
|
logInfo('Main', `Storage reset successfully (${storageDir})`);
|
|
} catch (err) {
|
|
logWarn('Main', `Storage reset error (may not exist): ${err.message}`);
|
|
}
|
|
} else {
|
|
logInfo('Main', `Master mode startup: preserving existing storage at ${storageDir}`);
|
|
}
|
|
try {
|
|
await fs.mkdir(storageDir, { recursive: true });
|
|
logInfo('Main', `Storage directory prepared (${storageDir})`);
|
|
} catch (err) {
|
|
logWarn('Main', `Could not create storage directory ${storageDir}: ${err.message}`);
|
|
}
|
|
|
|
if (cleanStorage) {
|
|
await runPluginStorageReset('--clean startup');
|
|
}
|
|
|
|
// Initialize the main Corestore before plugins so storage is guaranteed to exist
|
|
// even if a plugin fails during startup.
|
|
let store = null;
|
|
try {
|
|
store = new Corestore(storageDir);
|
|
await store.ready();
|
|
state.corestore = store;
|
|
logInfo('Main', `Corestore initialized with storage: ${storageDir}`);
|
|
if (process.env.ENABLE_HYPERCORE_STATS !== 'false') {
|
|
try {
|
|
const HypercoreStats = require('hypercore-stats');
|
|
state.hypercoreStats = await HypercoreStats.fromCorestore(store);
|
|
logInfo('Main', 'Hypercore stats collector initialized');
|
|
} catch (statsErr) {
|
|
logWarn('Main', `Hypercore stats unavailable: ${statsErr.message}`);
|
|
}
|
|
}
|
|
} catch (err) {
|
|
logError('Main', `Failed to initialize corestore at ${storageDir}: ${err.message}`);
|
|
process.exit(1);
|
|
}
|
|
|
|
setupCache();
|
|
ca.installRootCA();
|
|
|
|
// Load persistent keypair early so plugins can access it via SDK
|
|
const keypair = await loadOrCreateKeypair();
|
|
state.keypair = keypair; // Store in state for SDK access before swarm is created
|
|
logInfo('Main', 'Persistent keypair loaded and available for plugins');
|
|
|
|
// Load and initialize plugins - this will discover internal domains from config.json files
|
|
let internalDomains = ['p2ns.admin']; // p2ns.admin is always internal
|
|
try {
|
|
const { loadAllPlugins, getInternalDomains, shutdownAllPlugins } = require('./includes/plugins/plugin-handler');
|
|
await loadAllPlugins();
|
|
internalDomains = await getInternalDomains();
|
|
|
|
// Register internal domains and generate certificates
|
|
for (const domain of internalDomains) {
|
|
addInternalDomain(domain);
|
|
ca.getOrCreateDomainCert(domain, [
|
|
{ type: 2, value: domain },
|
|
{ type: 7, value: '127.0.0.1' }
|
|
]);
|
|
logDebug('Main', `Generated certificate for internal domain ${domain}`);
|
|
}
|
|
|
|
logInfo('Main', 'Plugin system initialized');
|
|
|
|
// Store shutdown function for cleanup
|
|
state.shutdownPlugins = shutdownAllPlugins;
|
|
|
|
// Initialize invite diagnostics function early so admin interface can access it
|
|
// The actual diagnostics will work once swarm is initialized
|
|
state.diagnoseInviteIssues = function() {
|
|
if (!state.connectedPeers || !state.peerChannels) {
|
|
const { METHODS } = require('./includes/core/core-rpc-contract');
|
|
return {
|
|
schemaVersion: 2,
|
|
timestamp: new Date().toISOString(),
|
|
nodeType: state.isMaster ? 'master' : 'joiner',
|
|
dnsPassInitialized: !!state.dnsPass,
|
|
protocol: {
|
|
control: 'p2ns.core-request-rpc',
|
|
inviteWire: METHODS.INVITE_DELIVER,
|
|
legacyInviteChannel: false
|
|
},
|
|
connectedPeers: 0,
|
|
summary: {
|
|
rpcOpen: 0,
|
|
requestChannelOpen: 0,
|
|
pendingAcks: 0,
|
|
failedPeers: 0,
|
|
masterQueue: 0,
|
|
inFlightHandlers: 0
|
|
},
|
|
failedInvitePeers: [],
|
|
pendingInviteAcks: [],
|
|
pendingMasterInviteQueue: [],
|
|
inFlightInviteHandlers: [],
|
|
consecutiveInviteFailures: state.consecutiveInviteFailures || 0,
|
|
peers: {},
|
|
connectionIssues: [],
|
|
recommendations: ['System still initializing — diagnostics available after swarm is ready']
|
|
};
|
|
}
|
|
|
|
return diagnoseInviteIssues();
|
|
};
|
|
state.diagnoseInviteIssuesAsync = async function() {
|
|
return state.diagnoseInviteIssues();
|
|
};
|
|
} catch (err) {
|
|
logWarn('Main', `Error initializing plugins: ${err.message}`);
|
|
}
|
|
// Check if HTTPS port is in use (only if proxy server is not disabled)
|
|
if (process.env.DISABLE_PROXY_SERVER !== 'true') {
|
|
const checkPort = () => new Promise((resolve, reject) => {
|
|
const server = net.createServer();
|
|
server.once('error', (err) => {
|
|
server.close();
|
|
if (err.code === 'EADDRINUSE') {
|
|
reject(new Error(`Port ${process.env.HTTPS_PORT || 443} is already in use`));
|
|
} else {
|
|
reject(err);
|
|
}
|
|
});
|
|
server.once('listening', () => {
|
|
server.close();
|
|
resolve();
|
|
});
|
|
server.listen(process.env.HTTPS_PORT || 443, '127.0.0.1');
|
|
});
|
|
try {
|
|
await checkPort();
|
|
logDebug('Main', `Port ${process.env.HTTPS_PORT || 443} is free, proceeding with server setup`);
|
|
} catch (err) {
|
|
logError('Main', `Cannot start HTTPS server: ${err.message}`);
|
|
process.exit(1);
|
|
}
|
|
} else {
|
|
logInfo('Main', 'Proxy server disabled via DISABLE_PROXY_SERVER=true, skipping HTTPS port check');
|
|
}
|
|
// Hardcoded genesis key (topic) for discovery
|
|
const topic = crypto.createHash('sha256').update(process.env.TOPIC_SEED || 'p2ns-dns').digest();
|
|
state.topic = topic; // Store topic in state for keep-alive and cleanup
|
|
logDebug('Main', `Generated topic: ${topic.toString('hex')}`);
|
|
// Corestore is initialized before plugin loading
|
|
// Use the persistent keypair that was loaded earlier (already in state.keypair)
|
|
// Configure Hyperswarm for better local network peer discovery
|
|
const swarmOptions = {
|
|
keyPair: state.keypair,
|
|
maxPeers: parseInt(process.env.MAX_PEERS || '24', 10),
|
|
firewall: (remotePublicKey) => {
|
|
const peerId = remotePublicKey && remotePublicKey.toString('hex');
|
|
return !!(peerId && state.blockedPeers && state.blockedPeers.has(peerId));
|
|
}
|
|
};
|
|
|
|
const swarm = new Hyperswarm(swarmOptions);
|
|
state.swarm = swarm; // Store swarm in state for SDK access
|
|
logInfo('Main', 'Hyperswarm instance created with persisted keypair');
|
|
logInfo('Main', `Hyperswarm configured: maxPeers=${swarmOptions.maxPeers}`);
|
|
|
|
// Initialize replication manager for plugin databases
|
|
try {
|
|
const replicationManager = require('./includes/plugins/replication-manager');
|
|
replicationManager.initialize(swarm);
|
|
state.replicationManager = replicationManager;
|
|
logInfo('Main', 'HyperDB replication manager initialized');
|
|
|
|
// Enable replication for any databases that were created before manager initialization
|
|
try {
|
|
const { getAllPluginStores } = require('./includes/plugins/db-manager');
|
|
const pluginStores = getAllPluginStores();
|
|
for (const [pluginDomain, pluginStore] of pluginStores) {
|
|
try {
|
|
if (!replicationManager.isReplicationActive(pluginDomain)) {
|
|
await replicationManager.enableReplication(pluginDomain, pluginStore);
|
|
logInfo('Main', `Enabled replication for existing database: ${pluginDomain}`);
|
|
}
|
|
} catch (err) {
|
|
logWarn('Main', `Could not enable replication for ${pluginDomain}: ${err.message}`);
|
|
}
|
|
}
|
|
} catch (err) {
|
|
logWarn('Main', `Error enabling replication for existing databases: ${err.message}`);
|
|
}
|
|
} catch (err) {
|
|
logWarn('Main', `Could not initialize replication manager: ${err.message}`);
|
|
}
|
|
|
|
// Initialize drive replication manager for plugin drives
|
|
try {
|
|
const driveReplicationManager = require('./includes/plugins/drive-replication-manager');
|
|
driveReplicationManager.initialize(swarm);
|
|
state.driveReplicationManager = driveReplicationManager;
|
|
logInfo('Main', 'Hyperdrive replication manager initialized');
|
|
|
|
// Enable replication for any drives that were created before manager initialization
|
|
try {
|
|
const { getAllDriveInstances } = require('./includes/plugins/drive-manager');
|
|
const driveInstances = getAllDriveInstances();
|
|
for (const [driveKey, driveInfo] of driveInstances) {
|
|
try {
|
|
const { pluginDomain, driveName, drive, store } = driveInfo;
|
|
if (!driveReplicationManager.isReplicationActive(pluginDomain, driveName)) {
|
|
await driveReplicationManager.enableReplication(pluginDomain, driveName, drive, store);
|
|
logInfo('Main', `Enabled replication for existing drive: ${pluginDomain}/${driveName}`);
|
|
}
|
|
} catch (err) {
|
|
logWarn('Main', `Could not enable replication for ${driveKey}: ${err.message}`);
|
|
}
|
|
}
|
|
} catch (err) {
|
|
logWarn('Main', `Error enabling replication for existing drives: ${err.message}`);
|
|
}
|
|
} catch (err) {
|
|
logWarn('Main', `Could not initialize drive replication manager: ${err.message}`);
|
|
}
|
|
let dnsPass = null;
|
|
let core = null;
|
|
let isShuttingDown = false;
|
|
|
|
// Invite processing mutex to prevent concurrent invite processing
|
|
let processingInvite = false;
|
|
let currentInvitePromise = null;
|
|
let currentPairOperation = null; // Track active pairing operation for cleanup
|
|
|
|
// Helper function to safely get dnsPass and ensure state consistency
|
|
function getDnsPass() {
|
|
// Always prioritize state.dnsPass as the source of truth
|
|
if (state.dnsPass && !dnsPass) {
|
|
dnsPass = state.dnsPass;
|
|
logDebug('Swarm', 'Synchronized local dnsPass from state');
|
|
} else if (dnsPass && !state.dnsPass) {
|
|
state.dnsPass = dnsPass;
|
|
logDebug('Swarm', 'Synchronized state.dnsPass from local');
|
|
} else if (dnsPass !== state.dnsPass) {
|
|
// They differ - state.dnsPass takes precedence
|
|
logWarn('Swarm', 'dnsPass state inconsistency detected, using state.dnsPass as authoritative');
|
|
dnsPass = state.dnsPass;
|
|
}
|
|
return dnsPass || state.dnsPass;
|
|
}
|
|
|
|
/**
|
|
* Do not replicate the Autopass/Corestore state over the p2ns topic connection.
|
|
* Autopass owns replication through its own BlindPairing/Hyperswarm discovery flow.
|
|
* Replicating the Autopass base here races createInvite() and can leave HyperDB's
|
|
* atomic view mid-flush ("Atomic state must flush to parent").
|
|
*/
|
|
function scheduleConnectionReplication(conn) {
|
|
if (!conn || conn.destroyed) return;
|
|
logDebug('Swarm', 'Skipping p2ns-topic Autopass replication; Autopass swarm handles it');
|
|
try {
|
|
const { replicateConsensus } = require('./includes/core/consensus-autobase');
|
|
replicateConsensus(conn);
|
|
} catch (err) {
|
|
logDebug('Swarm', `Consensus sidecar replication: ${err.message}`);
|
|
}
|
|
}
|
|
|
|
// Helper function to safely set dnsPass and ensure state consistency
|
|
function setDnsPass(newPass) {
|
|
dnsPass = newPass;
|
|
state.dnsPass = newPass;
|
|
if (newPass) {
|
|
core = newPass.base;
|
|
logDebug('Swarm', 'dnsPass and core updated consistently');
|
|
|
|
// Process any pending invite requests now that master is ready
|
|
if (isMaster && state.pendingInviteRequests.size > 0) {
|
|
logInfo('Swarm', `Master is now ready, processing ${state.pendingInviteRequests.size} pending invite requests`);
|
|
processPendingInviteRequests();
|
|
}
|
|
}
|
|
}
|
|
|
|
// Helper functions for invite processing mutex
|
|
function isProcessingInvite() {
|
|
return processingInvite;
|
|
}
|
|
|
|
async function acquireInviteLock() {
|
|
if (processingInvite) {
|
|
logDebug('Swarm', 'Another invite is already being processed, waiting...');
|
|
if (currentInvitePromise) {
|
|
await currentInvitePromise;
|
|
}
|
|
}
|
|
processingInvite = true;
|
|
logDebug('Swarm', 'Acquired invite processing lock');
|
|
}
|
|
|
|
function releaseInviteLock() {
|
|
processingInvite = false;
|
|
currentInvitePromise = null;
|
|
logDebug('Swarm', 'Released invite processing lock');
|
|
}
|
|
|
|
// Helper function to add timeout to async operations
|
|
function withTimeout(promise, timeoutMs, operationName) {
|
|
return Promise.race([
|
|
promise,
|
|
new Promise((_, reject) => {
|
|
setTimeout(() => {
|
|
reject(new Error(`${operationName} timed out after ${timeoutMs}ms - this may indicate corrupted storage`));
|
|
}, timeoutMs);
|
|
})
|
|
]);
|
|
}
|
|
|
|
// Helper function to automatically clean corrupted storage and reset state
|
|
|
|
// Function to process pending invite requests when master becomes ready
|
|
async function processPendingInviteRequests() {
|
|
const { isDnsPassUsable } = require('./includes/core/core');
|
|
const { whenDnsPassIdle, ensureDnsPassOpen } = require('./includes/core/dns-pass-queue');
|
|
if (!isMaster || !state.dnsPass) {
|
|
return;
|
|
}
|
|
if (!isDnsPassUsable(state.dnsPass)) {
|
|
await ensureDnsPassOpen(state.dnsPass);
|
|
}
|
|
if (!isDnsPassUsable(state.dnsPass)) {
|
|
return;
|
|
}
|
|
await whenDnsPassIdle();
|
|
|
|
const pendingRequests = Array.from(state.pendingInviteRequests.entries());
|
|
state.pendingInviteRequests.clear(); // Clear queue immediately to avoid duplicate processing
|
|
|
|
for (const [peerId, requestInfo] of pendingRequests) {
|
|
// Check if peer is still connected
|
|
if (!connectedPeers.has(peerId)) {
|
|
logDebug('Swarm', `Skipping pending invite request for disconnected peer ${peerId}`);
|
|
continue;
|
|
}
|
|
|
|
// Check age of request (don't process very old requests)
|
|
const age = Date.now() - requestInfo.timestamp;
|
|
if (age > 30000) { // 30 seconds
|
|
logWarn('Swarm', `Skipping stale pending invite request from ${peerId} (age: ${Math.round(age/1000)}s)`);
|
|
continue;
|
|
}
|
|
|
|
logInfo('Swarm', `Processing pending invite request from ${peerId} (queued for ${Math.round(age/1000)}s)`);
|
|
|
|
try {
|
|
const { createInvite, inviteToWire, invitePreview } = require('./includes/core/dns-pass-queue');
|
|
const inv = await createInvite(state.dnsPass);
|
|
const invWire = inviteToWire(inv);
|
|
logInfo('Swarm', `Created invite for pending request from ${peerId}: ${invitePreview(inv)}`);
|
|
const inviteId = coreRpc.deliverInviteWire(peerId, invWire);
|
|
if (inviteId) {
|
|
logInfo('Swarm', `Successfully sent pending invite.deliver to ${peerId}`);
|
|
} else {
|
|
logWarn('Swarm', `Failed to send pending invite.deliver to ${peerId} - RPC not ready`);
|
|
}
|
|
} catch (err) {
|
|
logError('Swarm', `Error processing pending invite request for ${peerId}: ${err.message}`);
|
|
// Send error response so peer knows what happened
|
|
try {
|
|
coreRpc.inviteUnavailable(peerId, 'creation_failed');
|
|
} catch (sendErr) {
|
|
// Channel might be closed, ignore
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
|
|
function diagnoseInviteIssues() {
|
|
const { METHODS, INVITE_STATUS } = require('./includes/core/core-rpc-contract');
|
|
const requestChannelInfo = channelManager.getChannelInfo(CORE_DOMAIN, 'request');
|
|
|
|
const networkSummary = networkManifest.getLocalNetworkSummary();
|
|
const diagnostics = {
|
|
schemaVersion: 2,
|
|
timestamp: new Date().toISOString(),
|
|
nodeType: isMaster ? (state.masterPendingPass ? 'master_pending_pass' : 'master') : 'joiner',
|
|
dnsPassInitialized: !!getDnsPass(),
|
|
networkId: networkSummary.networkId,
|
|
manifestPresent: networkSummary.manifestPresent,
|
|
isGenesis: networkSummary.isGenesis,
|
|
masterPendingPass: networkSummary.masterPendingPass,
|
|
splitBrainWarnings: networkSummary.splitBrainWarnings,
|
|
protocol: {
|
|
control: 'p2ns.core-request-rpc',
|
|
lifecycleChannel: 'p2ns.core-request',
|
|
inviteWire: METHODS.INVITE_DELIVER,
|
|
legacyInviteChannel: false,
|
|
rpcMethods: Object.values(METHODS)
|
|
},
|
|
config: {
|
|
allowAnyWriterInvites: process.env.ALLOW_ANY_WRITER_INVITES === 'true',
|
|
masterInviteOnly: process.env.MASTER_INVITE_ONLY === 'true',
|
|
inviteRelayMaxHops: parseInt(process.env.INVITE_RELAY_MAX_HOPS || '3', 10),
|
|
inviteAckTimeoutMs: parseInt(process.env.INVITE_ACK_TIMEOUT || '10000', 10),
|
|
inviteHandlerTimeoutMs: parseInt(process.env.INVITE_HANDLER_TIMEOUT_MS || '25000', 10)
|
|
},
|
|
connectedPeers: connectedPeers.size,
|
|
failedInvitePeers: Array.from(failedInvitePeers),
|
|
pendingInviteAcks: [],
|
|
pendingMasterInviteQueue: [],
|
|
inFlightInviteHandlers: Array.from(pendingInviteRequestHandlers),
|
|
consecutiveInviteFailures: state.consecutiveInviteFailures || 0,
|
|
peers: {},
|
|
summary: {
|
|
rpcOpen: 0,
|
|
rpcAttached: 0,
|
|
requestChannelOpen: 0,
|
|
requestChannelExists: 0,
|
|
pendingAcks: pendingInviteAcks.size,
|
|
failedPeers: failedInvitePeers.size,
|
|
masterQueue: state.pendingInviteRequests?.size || 0,
|
|
inFlightHandlers: pendingInviteRequestHandlers.size
|
|
},
|
|
connectionIssues: [],
|
|
recommendations: []
|
|
};
|
|
|
|
for (const [peerId, ackInfo] of pendingInviteAcks.entries()) {
|
|
diagnostics.pendingInviteAcks.push({
|
|
peerId,
|
|
retryCount: ackInfo?.retryCount ?? 0,
|
|
inviteId: ackInfo?.inviteId || null,
|
|
waitingForAck: !!ackInfo?.timeout
|
|
});
|
|
}
|
|
|
|
if (state.pendingInviteRequests) {
|
|
for (const [peerId, info] of state.pendingInviteRequests.entries()) {
|
|
diagnostics.pendingMasterInviteQueue.push({
|
|
peerId,
|
|
ageMs: Date.now() - (info?.timestamp || Date.now()),
|
|
retryCount: info?.retryCount ?? 0
|
|
});
|
|
}
|
|
}
|
|
|
|
for (const peerId of connectedPeers) {
|
|
const requestPeerChannel = requestChannelInfo?.peerChannels?.get(peerId);
|
|
const rpc = coreRpc.getRequestRpc(peerId);
|
|
const swarmChan = peerChannels.get(peerId);
|
|
const ackInfo = pendingInviteAcks.get(peerId);
|
|
|
|
const requestOpen = !!(requestPeerChannel?.rpc?.opened && !requestPeerChannel?.rpc?.closed);
|
|
const rpcOpen = !!(rpc?.opened);
|
|
const rpcAttached = !!rpc;
|
|
|
|
if (requestPeerChannel) diagnostics.summary.requestChannelExists++;
|
|
if (requestOpen) diagnostics.summary.requestChannelOpen++;
|
|
if (rpcAttached) diagnostics.summary.rpcAttached++;
|
|
if (rpcOpen) diagnostics.summary.rpcOpen++;
|
|
|
|
let connectionOk = true;
|
|
let connectionDetail = 'ok';
|
|
if (!swarmChan?.conn) {
|
|
connectionOk = false;
|
|
connectionDetail = 'no connection object';
|
|
diagnostics.connectionIssues.push(`${peerId}: no connection object`);
|
|
} else if (swarmChan.conn.destroyed) {
|
|
connectionOk = false;
|
|
connectionDetail = 'connection destroyed';
|
|
diagnostics.connectionIssues.push(`${peerId}: connection destroyed`);
|
|
} else if (swarmChan.conn.socket?.destroyed) {
|
|
connectionOk = false;
|
|
connectionDetail = 'socket destroyed';
|
|
diagnostics.connectionIssues.push(`${peerId}: socket destroyed`);
|
|
}
|
|
|
|
diagnostics.peers[peerId] = {
|
|
connectionOk,
|
|
connectionDetail,
|
|
failedInvite: failedInvitePeers.has(peerId),
|
|
requestChannel: {
|
|
exists: !!requestPeerChannel,
|
|
opened: requestOpen,
|
|
localOpened: requestPeerChannel?.localOpened || false,
|
|
remoteOpened: requestPeerChannel?.remoteOpened || false
|
|
},
|
|
rpc: {
|
|
attached: rpcAttached,
|
|
opened: rpcOpen,
|
|
ready: rpcAttached && rpcOpen
|
|
},
|
|
pendingAck: ackInfo
|
|
? {
|
|
waiting: !!ackInfo.timeout,
|
|
retryCount: ackInfo.retryCount ?? 0,
|
|
inviteId: ackInfo.inviteId || null
|
|
}
|
|
: null
|
|
};
|
|
}
|
|
|
|
if (!diagnostics.dnsPassInitialized && (!isMaster || state.masterPendingPass)) {
|
|
diagnostics.recommendations.push(
|
|
'No dnsPass — ensure a writer peer is reachable and invite.request → invite.deliver → invite.ack completes over RPC'
|
|
);
|
|
}
|
|
|
|
if (isMaster && !diagnostics.dnsPassInitialized) {
|
|
if (state.masterPendingPass) {
|
|
diagnostics.recommendations.push(
|
|
'Secondary master: pair into the network via invite from a genesis or writer peer (empty storage does not create a new Autopass)'
|
|
);
|
|
} else {
|
|
diagnostics.recommendations.push('Master dnsPass not initialized — cannot create invites until Autopass is ready');
|
|
}
|
|
}
|
|
|
|
if (diagnostics.splitBrainWarnings && diagnostics.splitBrainWarnings.length > 0) {
|
|
diagnostics.recommendations.push(
|
|
`CRITICAL: ${diagnostics.splitBrainWarnings.length} split-network warning(s) — peers report a different networkId; use a single genesis and --clean stray nodes`
|
|
);
|
|
}
|
|
|
|
if (diagnostics.summary.masterQueue > 0) {
|
|
diagnostics.recommendations.push(
|
|
`${diagnostics.summary.masterQueue} invite.request(s) queued until master dnsPass is ready`
|
|
);
|
|
}
|
|
|
|
if (diagnostics.summary.inFlightHandlers > 0) {
|
|
diagnostics.recommendations.push(
|
|
`${diagnostics.summary.inFlightHandlers} peer(s) have in-flight invite creation (invite.request handler running)`
|
|
);
|
|
}
|
|
|
|
if (diagnostics.failedInvitePeers.length > 0) {
|
|
diagnostics.recommendations.push(
|
|
`${diagnostics.failedInvitePeers.length} peer(s) marked unavailable for invites (invite.unavailable or failed invite.request)`
|
|
);
|
|
}
|
|
|
|
if (diagnostics.pendingInviteAcks.length > 0) {
|
|
diagnostics.recommendations.push(
|
|
`${diagnostics.pendingInviteAcks.length} proactive invite.deliver awaiting invite.ack — check joiner pairing or RPC connectivity`
|
|
);
|
|
}
|
|
|
|
if (diagnostics.consecutiveInviteFailures >= 3) {
|
|
if (!diagnostics.dnsPassInitialized && !isMaster) {
|
|
diagnostics.recommendations.push(
|
|
`CRITICAL: ${diagnostics.consecutiveInviteFailures} consecutive invite.deliver failures — try: node p2ns.js --clean`
|
|
);
|
|
} else {
|
|
diagnostics.recommendations.push(
|
|
`${diagnostics.consecutiveInviteFailures} consecutive invite processing failures — storage corruption possible, try --clean`
|
|
);
|
|
}
|
|
}
|
|
|
|
if (diagnostics.connectionIssues.length > 0) {
|
|
diagnostics.recommendations.push(
|
|
`${diagnostics.connectionIssues.length} swarm connection issue(s) — verify Hyperswarm / firewall / peer reachability`
|
|
);
|
|
}
|
|
|
|
if (connectedPeers.size > 0 && diagnostics.summary.rpcOpen < connectedPeers.size) {
|
|
diagnostics.recommendations.push(
|
|
`${connectedPeers.size - diagnostics.summary.rpcOpen} peer(s) without open p2ns.core-request-rpc — invite.deliver and invite.request will fail until RPC opens`
|
|
);
|
|
}
|
|
|
|
if (connectedPeers.size > 0 && diagnostics.summary.requestChannelOpen < connectedPeers.size) {
|
|
diagnostics.recommendations.push(
|
|
`${connectedPeers.size - diagnostics.summary.requestChannelOpen} peer(s) without open p2ns.core-request lifecycle channel — heartbeats and channel open hooks may be delayed`
|
|
);
|
|
}
|
|
|
|
if (!getDnsPass() && connectedPeers.size === 0) {
|
|
diagnostics.recommendations.push('No peers connected — cannot request invites until Hyperswarm connects to a writer');
|
|
}
|
|
|
|
if (isMaster && process.env.MASTER_INVITE_ONLY === 'true') {
|
|
diagnostics.recommendations.push('MASTER_INVITE_ONLY=true — only master nodes may issue invites');
|
|
} else if (isMaster && process.env.ALLOW_ANY_WRITER_INVITES !== 'true') {
|
|
diagnostics.recommendations.push(
|
|
'Master mode: only master nodes create invites unless ALLOW_ANY_WRITER_INVITES=true'
|
|
);
|
|
}
|
|
|
|
diagnostics.inviteStatusEnum = INVITE_STATUS;
|
|
return diagnostics;
|
|
}
|
|
|
|
// Update diagnostics function with full implementation now that swarm is ready
|
|
state.diagnoseInviteIssues = diagnoseInviteIssues;
|
|
const connectedPeers = new Set();
|
|
const peerChannels = new Map();
|
|
const failedInvitePeers = new Set(); // Track peers that cannot provide invites
|
|
const pendingInviteAcks = new Map(); // Track pending invite acknowledgments: peerId -> { timeout, retryCount }
|
|
const pendingInviteRequestHandlers = new Set(); // Peers with in-flight invite creation
|
|
state.pendingInviteRequestHandlers = pendingInviteRequestHandlers;
|
|
// consecutiveInviteFailures is now tracked in state object
|
|
state.connectedPeers = connectedPeers;
|
|
state.peerChannels = peerChannels;
|
|
|
|
function prunePeerTrackingMaps() {
|
|
const maxTracked = parseInt(process.env.MAX_TRACKED_PEERS || '500', 10);
|
|
if (!state.peerHistory || state.peerHistory.size <= maxTracked) return;
|
|
for (const trackedPeerId of [...state.peerHistory.keys()]) {
|
|
if (!connectedPeers.has(trackedPeerId)) {
|
|
state.peerHistory.delete(trackedPeerId);
|
|
state.peerMetrics.delete(trackedPeerId);
|
|
}
|
|
if (state.peerHistory.size <= maxTracked) break;
|
|
}
|
|
}
|
|
|
|
// Register core p2ns channels using channel-manager (like plugins do)
|
|
// This makes them visible in the UI and uses the proven SDK code
|
|
const channelManager = require('./includes/plugins/channel-manager');
|
|
const coreRpc = require('./includes/core/core-rpc');
|
|
const CORE_DOMAIN = coreRpc.CORE_DOMAIN;
|
|
|
|
// Function to send removal request to all peers
|
|
function sendRemovalRequest(domain) {
|
|
const count = coreRpc.broadcastRemoveDomain(domain);
|
|
logDebug('Main', `Sent removal request for ${domain} to ${count} peers`);
|
|
}
|
|
state.sendRemovalRequest = sendRemovalRequest;
|
|
|
|
// Function to send consensus recalculation request to all peers
|
|
function sendConsensusRequest(domain) {
|
|
const count = coreRpc.broadcastRecalculateConsensus(domain);
|
|
logDebug('Main', `Sent consensus recalculation request for ${domain} to ${count} peers`);
|
|
}
|
|
state.sendConsensusRequest = sendConsensusRequest;
|
|
|
|
// Function to send conflict claim removal notification to all peers
|
|
function sendConflictClaimRemoval(domain) {
|
|
const count = coreRpc.broadcastRemoveConflictClaim(domain);
|
|
logDebug('Main', `Sent conflict claim removal notification for ${domain} to ${count} peers`);
|
|
}
|
|
state.sendConflictClaimRemoval = sendConflictClaimRemoval;
|
|
|
|
const inviteChannelWaitMs = parseInt(process.env.INVITE_CHANNEL_WAIT_MS || '5000', 10);
|
|
|
|
/** Joiner: request Autopass invite once the request channel is ready. */
|
|
async function requestInviteFromPeer(targetPeerId) {
|
|
if (isMaster || getDnsPass()) {
|
|
return false;
|
|
}
|
|
if (failedInvitePeers.has(targetPeerId)) {
|
|
logDebug('Swarm', `Skipping invite request for peer ${targetPeerId} (cannot provide invite)`);
|
|
return false;
|
|
}
|
|
if (!connectedPeers.has(targetPeerId)) {
|
|
return false;
|
|
}
|
|
|
|
const res = await coreRpc.inviteRequest(targetPeerId, inviteChannelWaitMs);
|
|
if (!res) {
|
|
logWarn('Swarm', `RPC invite.request failed for peer ${targetPeerId}`);
|
|
return false;
|
|
}
|
|
if (res.status === coreRpc.INVITE_STATUS.UNAVAILABLE) {
|
|
logInfo('Swarm', `Peer ${targetPeerId} unavailable: ${res.reason || 'unknown'}`);
|
|
failedInvitePeers.add(targetPeerId);
|
|
return false;
|
|
}
|
|
if (res.status === coreRpc.INVITE_STATUS.QUEUED) {
|
|
logDebug('Swarm', `Invite request queued by peer ${targetPeerId}: ${res.reason || 'queued'}`);
|
|
return true;
|
|
}
|
|
logInfo('Swarm', `Invite request accepted by peer ${targetPeerId}`);
|
|
return true;
|
|
}
|
|
state.requestInviteFromPeer = requestInviteFromPeer;
|
|
|
|
const { createCoreSwarmHandlers } = require('./includes/core/core-swarm-handlers');
|
|
const swarmHandlers = createCoreSwarmHandlers({
|
|
Autopass,
|
|
store,
|
|
core,
|
|
state,
|
|
isMaster,
|
|
getDnsPass,
|
|
setDnsPass,
|
|
connectedPeers,
|
|
peerChannels,
|
|
failedInvitePeers,
|
|
pendingInviteAcks,
|
|
pendingInviteRequestHandlers,
|
|
isProcessingInvite,
|
|
acquireInviteLock,
|
|
releaseInviteLock,
|
|
withTimeout,
|
|
getCurrentInvitePromise: () => currentInvitePromise,
|
|
setCurrentInvitePromise: (p) => { currentInvitePromise = p; },
|
|
getCurrentPairOperation: () => currentPairOperation,
|
|
setCurrentPairOperation: (p) => { currentPairOperation = p; },
|
|
setupDomainsWatcher,
|
|
shouldLoadDomains: () => shouldMasterLoadDomains(state.isGenesis),
|
|
listDomains,
|
|
doAutoVotes,
|
|
setupListeners,
|
|
coreRpc,
|
|
channelManager,
|
|
CORE_DOMAIN
|
|
});
|
|
state.processInviteWire = swarmHandlers.processInviteWire;
|
|
|
|
coreRpc.registerCoreRpcHandlers({
|
|
onInviteRequest: swarmHandlers.onInviteRequest,
|
|
onInviteDeliver: swarmHandlers.onInviteDeliver,
|
|
onInviteAck: swarmHandlers.onInviteAck,
|
|
onInviteUnavailable: swarmHandlers.onInviteUnavailable,
|
|
onInviteQueued: swarmHandlers.onInviteQueued,
|
|
onInviteRelayRequest: swarmHandlers.onInviteRelayRequest,
|
|
onInviteRelayResponse: swarmHandlers.onInviteRelayResponse,
|
|
onCoreStatus: swarmHandlers.onCoreStatus,
|
|
onConsensusRemoveDomain: swarmHandlers.onConsensusRemoveDomain,
|
|
onConsensusRecalculate: swarmHandlers.onConsensusRecalculate,
|
|
onConsensusRemoveConflictClaim: swarmHandlers.onConsensusRemoveConflictClaim
|
|
});
|
|
|
|
async function diagnoseInviteIssuesAsync() {
|
|
const diagnostics = diagnoseInviteIssues();
|
|
const peerIds = Object.keys(diagnostics.peers || {});
|
|
if (peerIds.length === 0) {
|
|
diagnostics.remoteQueried = 0;
|
|
return diagnostics;
|
|
}
|
|
|
|
const timeoutMs = parseInt(process.env.CORE_STATUS_RPC_TIMEOUT_MS || '3000', 10);
|
|
let queried = 0;
|
|
let clearedFailed = 0;
|
|
|
|
await Promise.all(peerIds.map(async (peerId) => {
|
|
const local = diagnostics.peers[peerId];
|
|
if (!local?.rpc?.ready) {
|
|
local.remote = { ok: false, error: 'rpc_not_ready' };
|
|
return;
|
|
}
|
|
queried++;
|
|
const remote = await coreRpc.coreStatusRequest(peerId, timeoutMs);
|
|
if (!remote || remote.error) {
|
|
local.remote = { ok: false, error: remote?.error || 'no_response' };
|
|
return;
|
|
}
|
|
local.remote = { ok: true, ...remote };
|
|
|
|
try {
|
|
networkManifest.recordPeerNetworkStatus(peerId, remote);
|
|
} catch (recordErr) {
|
|
logDebug('Swarm', `recordPeerNetworkStatus: ${recordErr.message}`);
|
|
}
|
|
|
|
if (local.failedInvite && remote.canProvideInvite) {
|
|
failedInvitePeers.delete(peerId);
|
|
local.failedInvite = false;
|
|
clearedFailed++;
|
|
} else if (!local.failedInvite && remote.dnsPassInitialized && !remote.canProvideInvite && !isMaster) {
|
|
const isJoinerOnlyPeer = remote.nodeType === 'joiner' && !remote.allowAnyWriterInvites;
|
|
if (isJoinerOnlyPeer) {
|
|
failedInvitePeers.add(peerId);
|
|
local.failedInvite = true;
|
|
}
|
|
}
|
|
}));
|
|
|
|
diagnostics.remoteQueried = queried;
|
|
if (clearedFailed > 0) {
|
|
diagnostics.failedInvitePeers = Array.from(failedInvitePeers);
|
|
diagnostics.summary.failedPeers = failedInvitePeers.size;
|
|
diagnostics.recommendations = diagnostics.recommendations.filter(
|
|
(r) => !r.includes('marked unavailable for invites')
|
|
);
|
|
if (failedInvitePeers.size > 0) {
|
|
diagnostics.recommendations.push(
|
|
`${failedInvitePeers.size} peer(s) marked unavailable for invites (invite.unavailable or failed invite.request)`
|
|
);
|
|
}
|
|
}
|
|
|
|
return diagnostics;
|
|
}
|
|
|
|
state.diagnoseInviteIssuesAsync = diagnoseInviteIssuesAsync;
|
|
|
|
channelManager.registerPluginProtocol(CORE_DOMAIN, 'request', {
|
|
methods: {},
|
|
autoReconnect: true,
|
|
onPeerOpen: (peerId) => {
|
|
logDebug('Swarm', `Core request RPC open for peer ${peerId}`);
|
|
if (getDnsPass()) {
|
|
return;
|
|
}
|
|
logDebug('Swarm', `Core request RPC open for peer ${peerId.substring(0, 16)}..., requesting invite`);
|
|
requestInviteFromPeer(peerId).catch((err) => {
|
|
logDebug('Swarm', `Invite request on RPC open failed for ${peerId}: ${err.message}`);
|
|
});
|
|
},
|
|
onPeerClose: (peerId) => {
|
|
logDebug('Swarm', `Core request RPC closed for peer ${peerId}`);
|
|
}
|
|
});
|
|
|
|
logInfo('Main', 'Registered core p2ns channels via channel-manager');
|
|
|
|
// Helper function to check connection stability for invite operations
|
|
function isConnectionStable(peerId, conn) {
|
|
if (!conn || conn.destroyed) {
|
|
return false;
|
|
}
|
|
|
|
// Check if peer is still in connected peers set
|
|
if (!connectedPeers.has(peerId)) {
|
|
return false;
|
|
}
|
|
|
|
// Check if peer channels exist and are valid
|
|
const channels = peerChannels.get(peerId);
|
|
if (!channels || channels.conn !== conn) {
|
|
return false;
|
|
}
|
|
|
|
// Basic connection health check - ensure socket exists and is not destroyed
|
|
if (conn.socket && conn.socket.destroyed) {
|
|
return false;
|
|
}
|
|
|
|
return true;
|
|
}
|
|
|
|
// Helper function to send proactive invite with acknowledgment and retry
|
|
async function sendProactiveInviteWithRetry(pass, peerId, conn, isReconnection, maxRetries = 5) {
|
|
const baseAckTimeout = parseInt(process.env.INVITE_ACK_TIMEOUT || '10000', 10); // Increased default
|
|
let retryCount = 0;
|
|
|
|
const attemptSend = async () => {
|
|
if (!isConnectionStable(peerId, conn)) {
|
|
logDebug('Swarm', `Connection unstable or closed, stopping invite retry for ${peerId}`);
|
|
pendingInviteAcks.delete(peerId);
|
|
return;
|
|
}
|
|
|
|
if (retryCount >= maxRetries) {
|
|
logWarn('Swarm', `Max invite retries (${maxRetries}) reached for peer ${peerId}`);
|
|
failedInvitePeers.add(peerId); // Mark as failed to avoid future attempts
|
|
pendingInviteAcks.delete(peerId);
|
|
return;
|
|
}
|
|
|
|
retryCount++;
|
|
try {
|
|
const { isDnsPassUsable } = require('./includes/core/core');
|
|
const { whenDnsPassIdle, ensureDnsPassOpen } = require('./includes/core/dns-pass-queue');
|
|
await whenDnsPassIdle();
|
|
if (!isDnsPassUsable(pass)) {
|
|
await ensureDnsPassOpen(pass);
|
|
}
|
|
if (!isDnsPassUsable(pass)) {
|
|
logWarn('Swarm', `dnsPass not usable, skipping proactive invite for ${peerId}`);
|
|
pendingInviteAcks.delete(peerId);
|
|
return;
|
|
}
|
|
const { createInvite, inviteToWire, invitePreview } = require('./includes/core/dns-pass-queue');
|
|
const inv = await createInvite(pass);
|
|
const invWire = inviteToWire(inv);
|
|
logInfo('Swarm', `Created proactive invite for peer ${peerId} (attempt ${retryCount}/${maxRetries}): ${invitePreview(inv)}`);
|
|
const inviteId = coreRpc.deliverInviteWire(peerId, invWire);
|
|
|
|
if (inviteId) {
|
|
logInfo('Swarm', `Proactive invite.deliver sent to ${isReconnection ? 'reconnected' : 'new'} peer ${peerId}`);
|
|
|
|
const ackTimeout = baseAckTimeout * Math.pow(1.5, retryCount - 1);
|
|
const timeout = setTimeout(() => {
|
|
const pending = pendingInviteAcks.get(peerId);
|
|
if (pending && pending.retryCount < maxRetries) {
|
|
logWarn('Swarm', `No invite.ack from ${peerId} within ${Math.round(ackTimeout/1000)}s, retrying (attempt ${pending.retryCount + 1}/${maxRetries})...`);
|
|
attemptSend();
|
|
} else if (pending) {
|
|
logWarn('Swarm', `No invite.ack from ${peerId} after ${maxRetries} attempts, giving up`);
|
|
failedInvitePeers.add(peerId);
|
|
pendingInviteAcks.delete(peerId);
|
|
}
|
|
}, ackTimeout);
|
|
|
|
pendingInviteAcks.set(peerId, { timeout, retryCount, inviteId });
|
|
} else {
|
|
// Retry with exponential backoff if send failed
|
|
const retryDelay = Math.min(1000 * Math.pow(2, retryCount - 1), 10000); // Cap at 10 seconds
|
|
logWarn('Swarm', `Failed to send invite to ${peerId}, retrying in ${retryDelay}ms...`);
|
|
setTimeout(attemptSend, retryDelay);
|
|
}
|
|
} catch (err) {
|
|
const msg = err.message || '';
|
|
if (msg.includes('SESSION_CLOSED') || msg.includes('closing core') || msg.includes('Atomic state must flush')) {
|
|
logWarn('Swarm', `dnsPass busy or closing, cannot send invite to ${peerId}: ${msg}`);
|
|
pendingInviteAcks.delete(peerId);
|
|
return;
|
|
}
|
|
logError('Swarm', `Error sending proactive invite to peer ${peerId}: ${msg}`);
|
|
// Retry with exponential backoff on error
|
|
const retryDelay = Math.min(2000 * Math.pow(2, retryCount - 1), 15000); // Cap at 15 seconds
|
|
setTimeout(attemptSend, retryDelay);
|
|
}
|
|
};
|
|
|
|
await attemptSend();
|
|
}
|
|
|
|
state.holesailClientChildren = new Map();
|
|
state.holesailClientOpts = new Map();
|
|
state.holesailClientInfos = new Map();
|
|
|
|
// Persistent invite retry loop for joiners
|
|
let persistentInviteRetryInterval = null;
|
|
let retryAttemptCount = 0;
|
|
if (!isMaster) {
|
|
const retryIntervalMs = parseInt(process.env.INVITE_RETRY_INTERVAL || '15000', 10);
|
|
const relayAfterAttempts = parseInt(process.env.INVITE_RELAY_AFTER_ATTEMPTS || '3', 10);
|
|
const maxRetryAttempts = parseInt(process.env.MAX_INVITE_RETRY_ATTEMPTS || '20', 10);
|
|
logInfo('Swarm', `Starting persistent invite retry loop (interval: ${retryIntervalMs}ms, relay after ${relayAfterAttempts} attempts, max attempts: ${maxRetryAttempts})`);
|
|
|
|
let inviteRetryInFlight = false;
|
|
persistentInviteRetryInterval = setInterval(() => {
|
|
if (inviteRetryInFlight) return;
|
|
inviteRetryInFlight = true;
|
|
void (async () => {
|
|
try {
|
|
const pass = getDnsPass();
|
|
if (pass) {
|
|
// Got invite, stop the loop
|
|
logInfo('Swarm', 'dnsPass initialized, stopping persistent invite retry loop');
|
|
clearInterval(persistentInviteRetryInterval);
|
|
persistentInviteRetryInterval = null;
|
|
return;
|
|
}
|
|
|
|
retryAttemptCount++;
|
|
|
|
// Stop after maximum attempts to avoid infinite retry
|
|
if (retryAttemptCount >= maxRetryAttempts) {
|
|
logWarn('Swarm', `Reached maximum invite retry attempts (${maxRetryAttempts}), stopping persistent retry loop`);
|
|
clearInterval(persistentInviteRetryInterval);
|
|
persistentInviteRetryInterval = null;
|
|
return;
|
|
}
|
|
|
|
// Broadcast invite request to all connected peers
|
|
if (state.broadcastInviteRequest) {
|
|
logDebug('Swarm', `Persistent retry (attempt ${retryAttemptCount}): broadcasting invite request to all peers`);
|
|
await state.broadcastInviteRequest();
|
|
} else if (connectedPeers.size > 0) {
|
|
// Fallback: manually send to all peers
|
|
logDebug('Swarm', `Persistent retry (attempt ${retryAttemptCount}): sending invite requests (fallback mode)`);
|
|
for (const otherPeerId of connectedPeers) {
|
|
if (failedInvitePeers.has(otherPeerId)) continue;
|
|
try {
|
|
await requestInviteFromPeer(otherPeerId);
|
|
} catch (err) {
|
|
logError('Swarm', `Error in persistent retry to ${otherPeerId}: ${err.message}`);
|
|
}
|
|
}
|
|
} else {
|
|
logWarn('Swarm', 'Persistent retry: no connected peers, waiting for connections');
|
|
}
|
|
|
|
// After several failed attempts, also try relay requests
|
|
if (retryAttemptCount >= relayAfterAttempts && connectedPeers.size > 0) {
|
|
const { getPersistentPublicKey } = require('./includes/infrastructure/utils');
|
|
const localPeerId = getPersistentPublicKey();
|
|
|
|
if (localPeerId) {
|
|
logInfo('Swarm', `Persistent retry: sending relay invite requests (attempt ${retryAttemptCount})`);
|
|
for (const otherPeerId of connectedPeers) {
|
|
try {
|
|
coreRpc.inviteRelayRequest(otherPeerId, localPeerId, 0);
|
|
} catch (err) {
|
|
logError('Swarm', `Error sending relay request to ${otherPeerId}: ${err.message}`);
|
|
}
|
|
}
|
|
}
|
|
}
|
|
} catch (err) {
|
|
logError('Swarm', `Persistent invite retry failed: ${err.message}`);
|
|
} finally {
|
|
inviteRetryInFlight = false;
|
|
}
|
|
})();
|
|
}, retryIntervalMs);
|
|
|
|
// Store reference for cleanup
|
|
state.persistentInviteRetryInterval = persistentInviteRetryInterval;
|
|
}
|
|
|
|
// Wait for DHT to be ready
|
|
logDebug('Main', 'Waiting for swarm to flush...');
|
|
await swarm.flush();
|
|
logDebug('Main', 'Swarm flush completed');
|
|
// Store watcher reference for cleanup
|
|
let domainsWatcher = null;
|
|
let domainsDebounceTimer = null;
|
|
|
|
// Store event listener references for cleanup
|
|
const listenerRefs = {
|
|
dnsPassUpdate: null,
|
|
corePeerAdd: null,
|
|
coreAppend: null,
|
|
autoVoteDebounce: null
|
|
};
|
|
|
|
// Function to setup watcher for domains file
|
|
function setupDomainsWatcher() {
|
|
const filename = process.env.DOMAINS_FILE || './cache/domains.json';
|
|
let currentDomains = new Map();
|
|
let isProcessing = false;
|
|
let processingPromise = null;
|
|
let debounceTimer = null;
|
|
|
|
async function processDomains() {
|
|
// Prevent concurrent processing
|
|
if (isProcessing) {
|
|
logDebug('Main', 'Domains file processing already in progress, skipping');
|
|
return;
|
|
}
|
|
|
|
isProcessing = true;
|
|
try {
|
|
if (!state.dnsPass) {
|
|
logWarn('Main', 'dnsPass not initialized, skipping domains processing');
|
|
return;
|
|
}
|
|
const data = await fs.readFile(filename, 'utf8');
|
|
const domains = JSON.parse(data);
|
|
// Support both old format (just domain/hash) and new format (with ssl flag)
|
|
const newMap = new Map(domains.map(({ domain, hash, ssl }) => [domain, { hash, ssl: ssl === true }]));
|
|
// Add or update domains
|
|
for (const [domain, { hash, ssl }] of newMap) {
|
|
const oldEntry = currentDomains.get(domain);
|
|
const oldHash = oldEntry ? (typeof oldEntry === 'string' ? oldEntry : oldEntry.hash) : null;
|
|
if (!oldHash || oldHash !== hash) {
|
|
// Use SSL flag from domains.json FIRST (it's the most recent source)
|
|
// Only check existing claim if domains.json doesn't have SSL flag
|
|
let sslValue = ssl === true; // Start with domains.json value (most recent)
|
|
|
|
// Only check existing claim if domains.json doesn't specify SSL
|
|
// This ensures domains.json is the source of truth for new/updated domains
|
|
if (sslValue === false) {
|
|
// domains.json says SSL=false, but check if existing claim has SSL=true to preserve it
|
|
const { parseClaimValue } = require('./includes/core/core');
|
|
const { getPersistentPublicKey } = require('./includes/infrastructure/utils');
|
|
|
|
try {
|
|
const localWriter = getPersistentPublicKey();
|
|
if (localWriter && state.dnsPass) {
|
|
await state.dnsPass.ready();
|
|
const claimKey = `claim:${domain}:${localWriter}`;
|
|
const { dnsPassGet } = require('./includes/core/dns-pass-queue');
|
|
const existingClaim = await dnsPassGet(state.dnsPass, claimKey);
|
|
if (existingClaim) {
|
|
const parsed = parseClaimValue(existingClaim.toString('utf8'));
|
|
// Only preserve SSL from existing claim if hash matches
|
|
if (parsed.hash === hash && parsed.ssl === true) {
|
|
sslValue = true;
|
|
}
|
|
}
|
|
}
|
|
} catch (err) {
|
|
logDebug('Main', `Could not check existing claim for ${domain}: ${err.message}`);
|
|
}
|
|
}
|
|
|
|
await addDomain(domain, hash, sslValue);
|
|
logInfo('Main', `Added/updated domain from ${filename}: "${domain}" with hash "${hash}"${sslValue ? ' (SSL enabled)' : ''}`);
|
|
if (!state.domainToIPMap.has(domain)) {
|
|
await createInterfaceForDomain(domain);
|
|
logDebug('Main', `Assigned IP to domain: ${domain} (${state.domainToIPMap.get(domain)})`);
|
|
}
|
|
}
|
|
}
|
|
// Remove domains no longer in file
|
|
const localWriter = getPersistentPublicKey();
|
|
if (localWriter) {
|
|
for (const domain of currentDomains.keys()) {
|
|
if (!newMap.has(domain)) {
|
|
await removeDomain(domain);
|
|
state.sendRemovalRequest(domain);
|
|
logInfo('Main', `Removed domain from ${filename}: "${domain}"`);
|
|
}
|
|
}
|
|
}
|
|
currentDomains = newMap;
|
|
} catch (err) {
|
|
logWarn('Main', `Failed to process ${filename}: ${err.message}`);
|
|
} finally {
|
|
isProcessing = false;
|
|
}
|
|
}
|
|
|
|
// Initial processing
|
|
processingPromise = processDomains();
|
|
|
|
// Set up watcher with debouncing
|
|
domainsWatcher = require('fs').watch(filename, { persistent: true }, (eventType, changedFilename) => {
|
|
if (eventType === 'change') {
|
|
logDebug('Main', `${changedFilename || filename} changed, scheduling reprocessing`);
|
|
// Clear existing debounce timer
|
|
if (debounceTimer) {
|
|
clearTimeout(debounceTimer);
|
|
}
|
|
// Wait for current processing to finish, then debounce new changes
|
|
debounceTimer = setTimeout(async () => {
|
|
if (processingPromise) {
|
|
try {
|
|
await processingPromise;
|
|
} catch (err) {
|
|
logError('Main', `Error in previous processing: ${err.message}`);
|
|
}
|
|
}
|
|
processingPromise = processDomains();
|
|
}, secondsToMs(0.5)); // 0.5 second debounce
|
|
domainsDebounceTimer = debounceTimer;
|
|
}
|
|
});
|
|
logDebug('Main', `Domains watcher set up for ${filename}`);
|
|
}
|
|
function setupListeners() {
|
|
const { invalidateEntriesCache } = require('./includes/core/core');
|
|
|
|
// Remove existing listeners before adding new ones to prevent duplicates
|
|
if (listenerRefs.dnsPassUpdate && state.dnsPass) {
|
|
state.dnsPass.removeListener('update', listenerRefs.dnsPassUpdate);
|
|
}
|
|
if (listenerRefs.corePeerAdd && core) {
|
|
core.removeListener('peer-add', listenerRefs.corePeerAdd);
|
|
}
|
|
if (listenerRefs.coreAppend && core) {
|
|
core.removeListener('append', listenerRefs.coreAppend);
|
|
}
|
|
|
|
// Create and store new listener references
|
|
listenerRefs.dnsPassUpdate = async () => {
|
|
// Skip operations during shutdown
|
|
if (state.isShuttingDown) {
|
|
logDebug('Swarm', 'Skipping dnsPass update handler during shutdown');
|
|
return;
|
|
}
|
|
|
|
if (state.pendingInviteRequestHandlers && state.pendingInviteRequestHandlers.size > 0) {
|
|
logDebug('Swarm', 'Deferring dnsPass update handler — invite request in progress');
|
|
return;
|
|
}
|
|
|
|
if (state.dnsPassWriteInProgress > 0) {
|
|
logDebug('Swarm', 'Deferring dnsPass update handler — Autopass write in progress');
|
|
if (listenerRefs.autoVoteDebounce) {
|
|
clearTimeout(listenerRefs.autoVoteDebounce);
|
|
}
|
|
listenerRefs.autoVoteDebounce = setTimeout(() => {
|
|
listenerRefs.dnsPassUpdate().catch((err) => {
|
|
logWarn('Swarm', `Deferred dnsPass update handler failed: ${err.message}`);
|
|
});
|
|
}, 500);
|
|
return;
|
|
}
|
|
|
|
try {
|
|
logDebug('Swarm', 'Pass update event triggered - showing updated list');
|
|
invalidateEntriesCache(); // Invalidate cache on update
|
|
|
|
try {
|
|
const { syncConsensusWithDnsPass } = require('./includes/core/consensus-autobase');
|
|
await syncConsensusWithDnsPass(state.dnsPass);
|
|
} catch (err) {
|
|
logDebug('Swarm', `Consensus sync on dnsPass update: ${err.message}`);
|
|
}
|
|
|
|
// Check for claim changes and handle dynamic subscriptions
|
|
try {
|
|
const { checkClaimChanges } = require('./includes/admin/subscription-manager');
|
|
await checkClaimChanges();
|
|
} catch (err) {
|
|
logWarn('Swarm', `Error checking claim changes: ${err.message}`);
|
|
}
|
|
|
|
// Immediate consensus update if enabled
|
|
const { validateConfig } = require('./includes/infrastructure/config');
|
|
let config;
|
|
try {
|
|
config = validateConfig();
|
|
} catch (e) {
|
|
config = { CONSENSUS_IMMEDIATE_UPDATE: true };
|
|
}
|
|
|
|
if (config.CONSENSUS_IMMEDIATE_UPDATE !== false) {
|
|
try {
|
|
await doAutoVotes();
|
|
} catch (err) {
|
|
logError('Swarm', `Error in immediate auto-votes: ${err.message}`);
|
|
}
|
|
} else {
|
|
// Debounced auto-votes for batch operations
|
|
if (listenerRefs.autoVoteDebounce) {
|
|
clearTimeout(listenerRefs.autoVoteDebounce);
|
|
}
|
|
listenerRefs.autoVoteDebounce = setTimeout(() => {
|
|
doAutoVotes().catch(err => {
|
|
logError('Swarm', `Error in debounced auto-votes: ${err.message}`);
|
|
});
|
|
}, secondsToMs(60)); // 60 second debounce
|
|
}
|
|
|
|
await listDomains();
|
|
broadcast({ type: 'update-database' });
|
|
await assignIPsToResolvedDomains(); // Assign IPs to newly resolved domains
|
|
await assignAllIPs(); // Ensure all resolved domains have IPs
|
|
} catch (err) {
|
|
logError('Swarm', `Error in update handler: ${err.message}`);
|
|
}
|
|
};
|
|
state.dnsPass.on('update', listenerRefs.dnsPassUpdate);
|
|
|
|
listenerRefs.corePeerAdd = async (peer) => {
|
|
// Skip operations during shutdown
|
|
if (state.isShuttingDown) {
|
|
logDebug('Swarm', 'Skipping core peer-add handler during shutdown');
|
|
return;
|
|
}
|
|
|
|
logDebug('Swarm', `Peer added to core: ${peer.remotePublicKey.toString('hex')}`);
|
|
try {
|
|
await core.update();
|
|
logDebug('Swarm', 'Core updated after peer-add - showing updated list');
|
|
|
|
// Immediate consensus update on peer add
|
|
const { validateConfig } = require('./includes/infrastructure/config');
|
|
let config;
|
|
try {
|
|
config = validateConfig();
|
|
} catch (e) {
|
|
config = { CONSENSUS_IMMEDIATE_UPDATE: true };
|
|
}
|
|
|
|
if (config.CONSENSUS_IMMEDIATE_UPDATE !== false) {
|
|
try {
|
|
await doAutoVotes();
|
|
} catch (err) {
|
|
logError('Swarm', `Error in immediate auto-votes: ${err.message}`);
|
|
}
|
|
}
|
|
|
|
await listDomains();
|
|
} catch (err) {
|
|
logError('Swarm', `Error updating core after peer-add: ${err.message}`);
|
|
}
|
|
};
|
|
core.on('peer-add', listenerRefs.corePeerAdd);
|
|
|
|
listenerRefs.coreAppend = async () => {
|
|
// Skip operations during shutdown
|
|
if (state.isShuttingDown) {
|
|
logDebug('Swarm', 'Skipping core append handler during shutdown');
|
|
return;
|
|
}
|
|
|
|
try {
|
|
logDebug('Swarm', 'Core append event - new data added, showing updated list');
|
|
invalidateEntriesCache(); // Invalidate cache on append
|
|
|
|
try {
|
|
const { syncConsensusWithDnsPass } = require('./includes/core/consensus-autobase');
|
|
await syncConsensusWithDnsPass(state.dnsPass);
|
|
} catch (err) {
|
|
logDebug('Swarm', `Consensus sync on core append: ${err.message}`);
|
|
}
|
|
|
|
// Check for claim changes and handle dynamic subscriptions
|
|
try {
|
|
const { checkClaimChanges } = require('./includes/admin/subscription-manager');
|
|
await checkClaimChanges();
|
|
} catch (err) {
|
|
logWarn('Swarm', `Error checking claim changes: ${err.message}`);
|
|
}
|
|
|
|
// Immediate consensus update on append (new claim/vote)
|
|
const { validateConfig } = require('./includes/infrastructure/config');
|
|
let config;
|
|
try {
|
|
config = validateConfig();
|
|
} catch (e) {
|
|
config = { CONSENSUS_IMMEDIATE_UPDATE: true };
|
|
}
|
|
|
|
if (config.CONSENSUS_IMMEDIATE_UPDATE !== false) {
|
|
try {
|
|
await doAutoVotes();
|
|
} catch (err) {
|
|
logError('Swarm', `Error in immediate auto-votes: ${err.message}`);
|
|
}
|
|
}
|
|
|
|
await listDomains();
|
|
broadcast({ type: 'update-database' });
|
|
await assignIPsToResolvedDomains(); // Assign IPs to newly resolved domains
|
|
await assignAllIPs(); // Ensure all resolved domains have IPs
|
|
} catch (err) {
|
|
logError('Swarm', `Error in append handler: ${err.message}`);
|
|
}
|
|
};
|
|
core.on('append', listenerRefs.coreAppend);
|
|
}
|
|
async function initializeMasterDnsPass() {
|
|
const topicSeed = process.env.TOPIC_SEED || 'p2ns-dns';
|
|
const manifestPath = networkManifest.getManifestPath(process.env.NETWORK_MANIFEST_FILE);
|
|
const hasData = await networkManifest.storageDirHasCorestoreData(storageDir);
|
|
const existingManifest = await networkManifest.readManifest(manifestPath);
|
|
|
|
if (isGenesis && existingManifest && !cleanStorage) {
|
|
logError(
|
|
'Main',
|
|
'Refusing --genesis: network manifest already exists. Run without --genesis to join as a secondary master, or use --clean for a new network.'
|
|
);
|
|
process.exit(1);
|
|
}
|
|
|
|
if (!hasData && !isGenesis) {
|
|
state.masterPendingPass = true;
|
|
logInfo(
|
|
'Main',
|
|
'Secondary master: empty storage — waiting for invite to join the existing network'
|
|
);
|
|
return;
|
|
}
|
|
|
|
// Use Autopass's own Hyperswarm (see autopass/index.js _replicate).
|
|
// Never call store.replicate() on this corestore while Autopass is active.
|
|
const newPass = new Autopass(store);
|
|
logInfo('Main', 'Opening Autopass for master (dedicated Autopass swarm)');
|
|
await newPass.ready();
|
|
setDnsPass(newPass);
|
|
await core.ready();
|
|
await core.update();
|
|
|
|
const { isDnsPassUsable } = require('./includes/core/core');
|
|
if (!isDnsPassUsable(newPass)) {
|
|
logError('Main', 'Autopass base core is not usable after ready(). Storage may be from Autopass 2 or corrupted.');
|
|
logError('Main', 'Stop the process and restart with: node p2ns.js --clean --master [--genesis]');
|
|
process.exit(1);
|
|
}
|
|
|
|
logInfo('Main', `Autopass dnsPass ready (writable: ${core.writable}), key: ${core.key.toString('hex').slice(0, 16)}...`);
|
|
|
|
const genesisPublicKey = getPersistentPublicKey();
|
|
if (isGenesis && !existingManifest) {
|
|
state.networkManifest = await networkManifest.adoptManifestFromPass(newPass, topicSeed, manifestPath, {
|
|
genesisPublicKey,
|
|
isGenesisRun: true
|
|
});
|
|
state.isGenesis = true;
|
|
logInfo('Main', 'Genesis network created — network manifest written');
|
|
} else if (!existingManifest && hasData) {
|
|
state.networkManifest = await networkManifest.adoptManifestFromPass(newPass, topicSeed, manifestPath, {
|
|
genesisPublicKey,
|
|
isGenesisRun: false
|
|
});
|
|
state.isGenesis = false;
|
|
logInfo('Main', 'Adopted network manifest from existing master storage');
|
|
} else if (existingManifest) {
|
|
const validation = networkManifest.validateManifestAgainstPass(existingManifest, newPass, topicSeed);
|
|
if (!validation.ok) {
|
|
logError('Main', `Network manifest validation failed: ${validation.error}`);
|
|
process.exit(1);
|
|
}
|
|
state.networkManifest = existingManifest;
|
|
state.isGenesis = !!existingManifest.isGenesis;
|
|
}
|
|
|
|
if (shouldMasterLoadDomains(state.isGenesis)) {
|
|
setupDomainsWatcher();
|
|
} else if (isMaster) {
|
|
logInfo('Main', 'Skipping domains.json watcher (secondary master; MASTER_LOAD_DOMAINS=false)');
|
|
}
|
|
setupListeners();
|
|
|
|
const { startConsensusForNetwork } = require('./includes/core/consensus-autobase');
|
|
startConsensusForNetwork({
|
|
store,
|
|
dnsPass: newPass,
|
|
networkManifest: state.networkManifest,
|
|
manifestPath
|
|
});
|
|
}
|
|
|
|
if (isMaster) {
|
|
await initializeMasterDnsPass();
|
|
} else if (!getDnsPass()) {
|
|
logInfo('Main', 'Joiner: dnsPass not loaded — will request an invite after peers connect');
|
|
}
|
|
// Join the topic with both server and client modes for better local network discovery
|
|
swarm.join(topic, { server: true, client: true });
|
|
logInfo('Main', 'Joined Hyperswarm topic (server: true, client: true)');
|
|
const roleLabel = isMaster
|
|
? state.masterPendingPass
|
|
? 'MASTER(pending-pass)'
|
|
: state.isGenesis
|
|
? 'MASTER(genesis)'
|
|
: 'MASTER'
|
|
: 'JOINER';
|
|
logInfo(
|
|
'Main',
|
|
`Network ready: role=${roleLabel}, dnsPass=${getDnsPass() ? 'initialized' : 'pending invite'}, networkId=${state.networkManifest?.networkId?.slice(0, 16) || 'n/a'}...`
|
|
);
|
|
|
|
// Set up periodic topic rejoin to ensure swarm stays connected
|
|
// This helps all nodes (master and joiner) maintain their connection to the DHT network
|
|
// Master nodes particularly benefit as they need to stay discoverable for peer connections
|
|
const swarmKeepAliveInterval = parseInt(process.env.SWARM_KEEPALIVE_INTERVAL || '60000', 10); // Default 60 seconds
|
|
let swarmKeepAliveTimer = null;
|
|
if (swarmKeepAliveInterval > 0) {
|
|
swarmKeepAliveTimer = setInterval(() => {
|
|
// Skip if we're shutting down
|
|
if (isShuttingDown) return;
|
|
|
|
try {
|
|
// Rejoin the topic (idempotent operation - safe to call multiple times)
|
|
// This ensures the swarm maintains its connection to the DHT network
|
|
const currentTopic = state.topic || topic;
|
|
swarm.join(currentTopic, { server: true, client: true });
|
|
logInfo('Swarm', `Swarm keep-alive: Rejoined topic to maintain connection`);
|
|
} catch (err) {
|
|
logWarn('Swarm', `Swarm keep-alive rejoin failed: ${err.message}`);
|
|
}
|
|
}, swarmKeepAliveInterval);
|
|
logInfo('Main', `Swarm keep-alive enabled with interval: ${swarmKeepAliveInterval}ms`);
|
|
}
|
|
// Store timer reference for cleanup
|
|
state.swarmKeepAliveTimer = swarmKeepAliveTimer;
|
|
|
|
// Expose suspend/resume hooks for optional runtime lifecycle integrations.
|
|
state.suspendSwarm = async () => {
|
|
await swarm.suspend();
|
|
if (state.dnsPass && typeof state.dnsPass.suspend === 'function') {
|
|
await state.dnsPass.suspend();
|
|
}
|
|
logInfo('Swarm', 'Swarm suspended');
|
|
};
|
|
state.resumeSwarm = async () => {
|
|
await swarm.resume();
|
|
if (state.dnsPass && typeof state.dnsPass.resume === 'function') {
|
|
await state.dnsPass.resume();
|
|
}
|
|
logInfo('Swarm', 'Swarm resumed');
|
|
};
|
|
|
|
// Set up periodic network-wide consensus recalculation
|
|
// This ensures that consensus remains consistent across the entire network
|
|
const networkConsensusInterval = parseInt(process.env.NETWORK_CONSENSUS_INTERVAL || '300000', 10); // Default 5 minutes
|
|
if (networkConsensusInterval > 0) {
|
|
const networkConsensusTimer = setInterval(() => {
|
|
if (isShuttingDown || !state.dnsPass) return;
|
|
|
|
try {
|
|
logInfo('Consensus', 'Triggering periodic network-wide consensus recalculation');
|
|
if (state.sendConsensusRequest) {
|
|
state.sendConsensusRequest('all');
|
|
}
|
|
// Also trigger locally
|
|
const { doAutoVotes } = require('./includes/core/core');
|
|
doAutoVotes().catch(err => {
|
|
logError('Consensus', `Error in periodic auto-votes: ${err.message}`);
|
|
});
|
|
} catch (err) {
|
|
logWarn('Consensus', `Periodic network consensus trigger failed: ${err.message}`);
|
|
}
|
|
}, networkConsensusInterval);
|
|
state.networkConsensusTimer = networkConsensusTimer;
|
|
logInfo('Main', `Network consensus interval enabled: ${networkConsensusInterval}ms`);
|
|
}
|
|
|
|
// Set up connection handler with timeout handling for better local network peer support
|
|
swarm.on('connection', async (conn, info) => {
|
|
// Skip processing if we're shutting down
|
|
if (isShuttingDown) {
|
|
logDebug('Swarm', 'Ignoring new connection during shutdown');
|
|
try {
|
|
conn.destroy();
|
|
} catch (err) {
|
|
// Ignore errors when destroying during shutdown
|
|
}
|
|
return;
|
|
}
|
|
// Set socket timeout to prevent hanging connections (helps with local network peers)
|
|
if (conn && conn.socket) {
|
|
const socketTimeout = parseInt(process.env.SWARM_CONNECTION_TIMEOUT || '30000', 10); // 30 seconds default
|
|
conn.socket.setTimeout(socketTimeout);
|
|
conn.socket.on('timeout', () => {
|
|
const peerId = conn.remotePublicKey ? conn.remotePublicKey.toString('hex').slice(0, 16) : 'unknown';
|
|
logWarn('Swarm', `Connection timeout for peer ${peerId}...`);
|
|
if (!conn.destroyed) {
|
|
conn.destroy();
|
|
}
|
|
});
|
|
}
|
|
// Validate remotePublicKey exists before using it
|
|
if (!conn.remotePublicKey) {
|
|
logWarn('Swarm', 'Connection established without remotePublicKey, closing connection');
|
|
conn.destroy();
|
|
return;
|
|
}
|
|
const peerId = conn.remotePublicKey.toString('hex');
|
|
logInfo('Swarm', `New connection established. Remote public key: ${peerId}`);
|
|
|
|
// Check if peer is blocked FIRST, before any other processing
|
|
if (state.blockedPeers && state.blockedPeers.has(peerId)) {
|
|
logWarn('Swarm', `Blocked peer ${peerId} attempted to connect, closing connection`);
|
|
if (info && typeof info.ban === 'function') {
|
|
info.ban(true);
|
|
}
|
|
conn.destroy();
|
|
return;
|
|
}
|
|
|
|
// If peer is already connected, check if old connection is still valid
|
|
if (connectedPeers.has(peerId)) {
|
|
const existingChannels = peerChannels.get(peerId);
|
|
let existingConnValid = false;
|
|
|
|
if (existingChannels && existingChannels.conn) {
|
|
// More thorough connection validation
|
|
const existingConn = existingChannels.conn;
|
|
existingConnValid = !existingConn.destroyed &&
|
|
existingConn.readable &&
|
|
existingConn.writable &&
|
|
!existingConn.ended &&
|
|
!existingConn.finished;
|
|
|
|
logDebug('Swarm', `Peer ${peerId} already connected, existing connection state: destroyed=${existingConn.destroyed}, readable=${existingConn.readable}, writable=${existingConn.writable}, ended=${existingConn.ended}, finished=${existingConn.finished}`);
|
|
} else {
|
|
logDebug('Swarm', `Peer ${peerId} in connectedPeers but no channels entry found`);
|
|
}
|
|
|
|
if (existingConnValid) {
|
|
// Existing connection is still valid - this is a true duplicate, reject the new one
|
|
logWarn('Swarm', `Peer ${peerId} already has a valid connection, rejecting duplicate connection`);
|
|
try {
|
|
conn.destroy();
|
|
} catch (err) {
|
|
logDebug('Swarm', `Error destroying duplicate connection: ${err.message}`);
|
|
}
|
|
return;
|
|
} else {
|
|
// Existing connection is dead/invalid - clean it up and accept the new one
|
|
logInfo('Swarm', `Peer ${peerId} has stale connection entry, cleaning up and accepting new connection`);
|
|
|
|
if (existingChannels) {
|
|
// Clean up existing connection
|
|
try {
|
|
// Clear timeouts
|
|
if (existingChannels.inviteRequestTimeout) {
|
|
clearTimeout(existingChannels.inviteRequestTimeout);
|
|
existingChannels.inviteRequestTimeout = null;
|
|
}
|
|
if (existingChannels.reconnectTimeout) {
|
|
clearTimeout(existingChannels.reconnectTimeout);
|
|
existingChannels.reconnectTimeout = null;
|
|
}
|
|
|
|
// Close existing connection if it's still open
|
|
if (existingChannels.conn && !existingChannels.conn.destroyed) {
|
|
logDebug('Swarm', `Closing stale connection for peer ${peerId}`);
|
|
try {
|
|
existingChannels.conn.end();
|
|
// Give it a moment to close gracefully, then destroy if needed
|
|
setTimeout(() => {
|
|
if (!existingChannels.conn.destroyed) {
|
|
existingChannels.conn.destroy();
|
|
}
|
|
}, 100);
|
|
} catch (err) {
|
|
logDebug('Swarm', `Error closing stale connection: ${err.message}`);
|
|
}
|
|
}
|
|
|
|
// Handle plugin channel disconnection
|
|
try {
|
|
const channelManager = require('./includes/plugins/channel-manager');
|
|
channelManager.handlePeerDisconnect(peerId);
|
|
} catch (err) {
|
|
logDebug('Swarm', `Error handling plugin channel disconnect during cleanup: ${err.message}`);
|
|
}
|
|
|
|
// Remove from peer channels
|
|
peerChannels.delete(peerId);
|
|
} catch (err) {
|
|
logError('Swarm', `Error cleaning up stale connection for peer ${peerId}: ${err.message}`);
|
|
}
|
|
}
|
|
|
|
// Remove from connected peers
|
|
connectedPeers.delete(peerId);
|
|
|
|
// Wait a brief moment for cleanup to complete
|
|
await new Promise(resolve => setTimeout(resolve, 200));
|
|
}
|
|
}
|
|
|
|
connectedPeers.add(peerId);
|
|
|
|
// Track peer connection
|
|
const connectTime = Date.now();
|
|
state.peerStartTimes.set(peerId, connectTime);
|
|
|
|
// Add to history
|
|
if (!state.peerHistory.has(peerId)) {
|
|
state.peerHistory.set(peerId, []);
|
|
}
|
|
const history = state.peerHistory.get(peerId);
|
|
history.push({
|
|
type: 'connect',
|
|
timestamp: connectTime
|
|
});
|
|
// Keep only last 50 items
|
|
if (history.length > 50) {
|
|
state.peerHistory.set(peerId, history.slice(-50));
|
|
}
|
|
|
|
// Update metrics
|
|
if (!state.peerMetrics.has(peerId)) {
|
|
state.peerMetrics.set(peerId, {
|
|
connections: 0,
|
|
totalDuration: 0,
|
|
avgDuration: 0,
|
|
lastSeen: connectTime
|
|
});
|
|
}
|
|
const metrics = state.peerMetrics.get(peerId);
|
|
metrics.connections++;
|
|
metrics.lastSeen = connectTime;
|
|
|
|
// Save peer metrics and history to disk
|
|
await savePeerMetrics();
|
|
await savePeerHistory();
|
|
|
|
trackPeerEvent('connect', peerId);
|
|
broadcast({ type: 'update-peers' });
|
|
broadcast({ type: 'update-stats' });
|
|
|
|
// Immediate consensus update on peer connect
|
|
const { validateConfig } = require('./includes/infrastructure/config');
|
|
let config;
|
|
try {
|
|
config = validateConfig();
|
|
} catch (e) {
|
|
config = { CONSENSUS_IMMEDIATE_UPDATE: true };
|
|
}
|
|
|
|
if (config.CONSENSUS_IMMEDIATE_UPDATE !== false && state.dnsPass) {
|
|
// Defer auto-votes on master so invite creation is not competing with pass.list()
|
|
const autoVoteDelay = isMaster
|
|
? parseInt(process.env.MASTER_AUTO_VOTE_DELAY || '8000', 10)
|
|
: 0;
|
|
setTimeout(() => {
|
|
if (state.isShuttingDown || !state.dnsPass) return;
|
|
doAutoVotes().catch(err => {
|
|
logError('Swarm', `Error in immediate auto-votes after peer connect: ${err.message}`);
|
|
});
|
|
}, autoVoteDelay);
|
|
|
|
// Notify other peers to recalculate consensus when a new node connects
|
|
// This ensures the new node's claims/votes are considered by the network
|
|
if (state.sendConsensusRequest) {
|
|
// Use a small delay to allow replication to begin before requesting consensus
|
|
setTimeout(() => {
|
|
state.sendConsensusRequest('all');
|
|
}, 2000);
|
|
}
|
|
}
|
|
|
|
scheduleConnectionReplication(conn);
|
|
|
|
// Replicate all plugin databases via replication manager
|
|
// All databases replicate over the same global topic
|
|
try {
|
|
if (state.replicationManager) {
|
|
// Don't await - let it run in background, errors are handled internally
|
|
state.replicationManager.handleNewConnection(conn).catch(err => {
|
|
logWarn('Swarm', `Error in replication manager handleNewConnection: ${err.message}`);
|
|
});
|
|
logDebug('Swarm', 'Plugin database replication handled via replication manager (global topic)');
|
|
} else {
|
|
// Fallback to old method if replication manager not available
|
|
const { getAllPluginStores } = require('./includes/plugins/db-manager');
|
|
const pluginStores = getAllPluginStores();
|
|
for (const [pluginDomain, pluginStore] of pluginStores) {
|
|
try {
|
|
logDebug('Swarm', `Replicating plugin database for ${pluginDomain}...`);
|
|
pluginStore.replicate(conn, { live: true });
|
|
logDebug('Swarm', `Plugin database replication started for ${pluginDomain}`);
|
|
} catch (err) {
|
|
logWarn('Swarm', `Failed to replicate plugin database for ${pluginDomain}: ${err.message}`);
|
|
}
|
|
}
|
|
}
|
|
} catch (err) {
|
|
logWarn('Swarm', `Error setting up plugin database replication: ${err.message}`);
|
|
}
|
|
|
|
// Replicate all plugin drives via drive replication manager
|
|
// All drives replicate over the same global topic
|
|
try {
|
|
if (state.driveReplicationManager) {
|
|
state.driveReplicationManager.handleNewConnection(conn);
|
|
logDebug('Swarm', 'Plugin drive replication handled via drive replication manager (global topic)');
|
|
}
|
|
} catch (err) {
|
|
logWarn('Swarm', `Error replicating plugin drives: ${err.message}`);
|
|
}
|
|
|
|
// Create a Protomux instance for the connection
|
|
const mux = Protomux.from(conn);
|
|
logDebug('Swarm', 'Protomux instance created for connection');
|
|
|
|
// Register pair handlers for plugin channels BEFORE opening any channels
|
|
// This prevents race conditions where remote opens a channel before we're ready
|
|
try {
|
|
channelManager.registerPairHandlersForPeer(peerId, mux);
|
|
} catch (err) {
|
|
logWarn('Swarm', `Error registering pair handlers: ${err.message}`);
|
|
}
|
|
|
|
// Store channel references and timeout for cleanup
|
|
// Note: actual channels are managed by channel-manager now
|
|
let inviteRequestTimeout = null;
|
|
let reconnectTimeout = null;
|
|
|
|
peerChannels.set(peerId, {
|
|
inviteRequestTimeout: null,
|
|
reconnectTimeout: null,
|
|
mux,
|
|
conn
|
|
});
|
|
|
|
// Create all plugin channels for this peer connection (including core p2ns channels)
|
|
const isReconnection = isPeerReconnection(peerId);
|
|
try {
|
|
channelManager.createPluginChannelsForPeer(peerId, conn, mux);
|
|
const connectionType = isReconnection ? 'reconnected' : 'new';
|
|
const nodeType = isMaster ? 'master' : 'joiner';
|
|
logDebug('Swarm', `Created channels for ${connectionType} ${nodeType} peer ${peerId.substring(0, 16)}... via channel-manager`);
|
|
} catch (err) {
|
|
logError('Swarm', `Error creating plugin channels for peer ${peerId}: ${err.message}`);
|
|
}
|
|
|
|
// Register peer connection with channel manager for proper tracking
|
|
try {
|
|
channelManager.registerPeerConnection(peerId, conn, mux);
|
|
logDebug('Swarm', `Registered peer connection with channel manager: ${peerId.substring(0, 16)}...`);
|
|
|
|
// For reconnections, run health check immediately to ensure channels are properly recreated
|
|
if (isPeerReconnection(peerId)) {
|
|
const nodeType = isMaster ? 'master' : 'joiner';
|
|
logInfo('Swarm', `${nodeType} peer ${peerId.substring(0, 16)}... reconnected - ensuring channels are properly recreated`);
|
|
setTimeout(() => {
|
|
try {
|
|
channelManager.performHealthCheck();
|
|
logDebug('Swarm', `Ran immediate health check for reconnected ${nodeType} peer ${peerId.substring(0, 16)}...`);
|
|
} catch (err) {
|
|
logWarn('Swarm', `Error running immediate health check for reconnected ${nodeType} peer: ${err.message}`);
|
|
}
|
|
}, 100); // Small delay to allow connection to stabilize
|
|
}
|
|
} catch (err) {
|
|
logWarn('Swarm', `Error registering peer connection with channel manager: ${err.message}`);
|
|
}
|
|
|
|
// Master node: Send proactive invite to new peers or reconnected peers
|
|
if (isMaster && state.dnsPass) {
|
|
const { isDnsPassUsable } = require('./includes/core/core');
|
|
if (!isDnsPassUsable(state.dnsPass)) {
|
|
logWarn('Swarm', `dnsPass not usable, skipping proactive invite for peer ${peerId}`);
|
|
} else {
|
|
const pass = state.dnsPass;
|
|
const isReconnection = isPeerReconnection(peerId);
|
|
|
|
// Remove from reconnection tracking if this was a reconnection
|
|
if (isReconnection) {
|
|
logInfo('Swarm', `Peer ${peerId} reconnected, removing from reconnection tracking`);
|
|
if (state.peersToReconnect) {
|
|
state.peersToReconnect.delete(peerId);
|
|
}
|
|
if (state.reconnectionAttempts) {
|
|
state.reconnectionAttempts.delete(peerId);
|
|
}
|
|
if (state.lastReconnectionAttempt) {
|
|
state.lastReconnectionAttempt.delete(peerId);
|
|
}
|
|
}
|
|
|
|
// Send proactive invite after waiting for bidirectional channel
|
|
const proactiveInviteDelay = parseInt(process.env.MASTER_PROACTIVE_INVITE_DELAY || '2000', 10);
|
|
setTimeout(async () => {
|
|
// Double-check connection is still stable
|
|
if (!isConnectionStable(peerId, conn)) {
|
|
logDebug('Swarm', `Connection unstable before proactive invite could be sent to ${peerId}`);
|
|
return;
|
|
}
|
|
|
|
const requestChannelInfo = channelManager.getChannelInfo(CORE_DOMAIN, 'request');
|
|
const requestPeerChannel = requestChannelInfo?.peerChannels?.get(peerId);
|
|
if (requestPeerChannel) {
|
|
await channelManager.waitForBidirectionalOpen(requestPeerChannel, 3000);
|
|
}
|
|
|
|
await sendProactiveInviteWithRetry(pass, peerId, conn, isReconnection, 3);
|
|
}, proactiveInviteDelay);
|
|
}
|
|
}
|
|
|
|
// If no dnsPass yet (joiner or secondary master), request invite from peers
|
|
if (!getDnsPass()) {
|
|
logDebug('Swarm', 'Requesting invite from peer (parallel mode)...');
|
|
|
|
// Send request to this peer
|
|
const sendInviteRequestToPeer = async (targetPeerId, targetConn) => {
|
|
if (getDnsPass()) {
|
|
logDebug('Swarm', 'dnsPass already initialized, skipping invite request');
|
|
return;
|
|
}
|
|
|
|
if (targetConn.destroyed || !connectedPeers.has(targetPeerId)) {
|
|
logDebug('Swarm', `Connection to peer ${targetPeerId} not available`);
|
|
return;
|
|
}
|
|
|
|
try {
|
|
await requestInviteFromPeer(targetPeerId);
|
|
} catch (err) {
|
|
logError('Swarm', `Error sending invite request to peer ${targetPeerId}: ${err.message}`);
|
|
}
|
|
};
|
|
|
|
// Give channels time to open before requesting an invite from master
|
|
setTimeout(async () => {
|
|
await sendInviteRequestToPeer(peerId, conn);
|
|
}, 1500);
|
|
|
|
// Also request from all other connected peers (parallel requests)
|
|
setTimeout(async () => {
|
|
const pass = getDnsPass();
|
|
if (pass) return; // Already got invite
|
|
|
|
for (const [otherPeerId, channels] of peerChannels) {
|
|
if (otherPeerId === peerId) continue; // Skip the peer we just connected to
|
|
if (failedInvitePeers.has(otherPeerId)) continue; // Skip failed peers
|
|
|
|
const otherConn = channels.conn;
|
|
if (otherConn && !otherConn.destroyed) {
|
|
await sendInviteRequestToPeer(otherPeerId, otherConn);
|
|
}
|
|
}
|
|
}, 200);
|
|
}
|
|
|
|
// Function to broadcast invite request to all connected peers
|
|
async function broadcastInviteRequest() {
|
|
const pass = getDnsPass();
|
|
if (pass) {
|
|
logDebug('Swarm', 'Already have dnsPass, no need to broadcast invite request');
|
|
return;
|
|
}
|
|
|
|
let sentCount = 0;
|
|
const broadcastPromises = [];
|
|
|
|
for (const [otherPeerId, channels] of peerChannels) {
|
|
if (failedInvitePeers.has(otherPeerId)) continue;
|
|
|
|
const otherConn = channels.conn;
|
|
if (!otherConn || otherConn.destroyed) continue;
|
|
|
|
const sendPromise = (async () => {
|
|
try {
|
|
const sent = await requestInviteFromPeer(otherPeerId);
|
|
if (sent) sentCount++;
|
|
return sent;
|
|
} catch (err) {
|
|
logError('Swarm', `Error broadcasting invite request to ${otherPeerId}: ${err.message}`);
|
|
return false;
|
|
}
|
|
})();
|
|
|
|
broadcastPromises.push(sendPromise);
|
|
}
|
|
|
|
// Wait for all broadcast attempts to complete
|
|
await Promise.allSettled(broadcastPromises);
|
|
|
|
if (sentCount > 0) {
|
|
logInfo('Swarm', `Broadcast invite request to ${sentCount} peer(s)`);
|
|
} else {
|
|
logWarn('Swarm', 'No available peers to request invite from');
|
|
}
|
|
}
|
|
|
|
// Store broadcastInviteRequest in state for access from persistent retry loop
|
|
if (!state.broadcastInviteRequest) {
|
|
state.broadcastInviteRequest = broadcastInviteRequest;
|
|
}
|
|
|
|
// Handle connection cleanup
|
|
conn.on('close', async () => {
|
|
logInfo('Swarm', `Connection closed with peer: ${peerId}`);
|
|
|
|
// Clear invite request timeout if it exists
|
|
const channels = peerChannels.get(peerId);
|
|
if (channels && channels.inviteRequestTimeout) {
|
|
clearTimeout(channels.inviteRequestTimeout);
|
|
channels.inviteRequestTimeout = null;
|
|
}
|
|
|
|
// Remove all event listeners from connection
|
|
conn.removeAllListeners();
|
|
|
|
// Track peer disconnection
|
|
const disconnectTime = Date.now();
|
|
const connectTime = state.peerStartTimes.get(peerId);
|
|
const duration = connectTime ? disconnectTime - connectTime : 0;
|
|
|
|
// Add to history
|
|
if (state.peerHistory.has(peerId)) {
|
|
const history = state.peerHistory.get(peerId);
|
|
history.push({
|
|
type: 'disconnect',
|
|
timestamp: disconnectTime,
|
|
duration
|
|
});
|
|
// Keep only last 50 items
|
|
if (history.length > 50) {
|
|
state.peerHistory.set(peerId, history.slice(-50));
|
|
}
|
|
}
|
|
|
|
// Update metrics
|
|
if (state.peerMetrics.has(peerId)) {
|
|
const metrics = state.peerMetrics.get(peerId);
|
|
if (duration > 0) {
|
|
metrics.totalDuration += duration;
|
|
metrics.avgDuration = metrics.totalDuration / metrics.connections;
|
|
}
|
|
// Save peer metrics and history to disk
|
|
await savePeerMetrics();
|
|
await savePeerHistory();
|
|
}
|
|
|
|
// Remove from start times
|
|
state.peerStartTimes.delete(peerId);
|
|
prunePeerTrackingMaps();
|
|
|
|
// Handle plugin channel disconnection
|
|
try {
|
|
const channelManager = require('./includes/plugins/channel-manager');
|
|
channelManager.handlePeerDisconnect(peerId);
|
|
channelManager.unregisterPeerConnection(peerId);
|
|
logDebug('Swarm', `Unregistered peer connection from channel manager: ${peerId.substring(0, 16)}...`);
|
|
} catch (err) {
|
|
logError('Swarm', `Error handling plugin channel disconnect for peer ${peerId}: ${err.message}`);
|
|
}
|
|
|
|
connectedPeers.delete(peerId);
|
|
peerChannels.delete(peerId);
|
|
|
|
// Clean up peer tracking data structures to prevent memory leaks
|
|
failedInvitePeers.delete(peerId);
|
|
pendingInviteAcks.delete(peerId);
|
|
|
|
trackPeerEvent('disconnect', peerId);
|
|
broadcast({ type: 'update-peers' });
|
|
broadcast({ type: 'update-stats' });
|
|
|
|
// Immediate consensus update on peer disconnect
|
|
const { validateConfig } = require('./includes/infrastructure/config');
|
|
let config;
|
|
try {
|
|
config = validateConfig();
|
|
} catch (e) {
|
|
config = { CONSENSUS_IMMEDIATE_UPDATE: true };
|
|
}
|
|
|
|
if (config.CONSENSUS_IMMEDIATE_UPDATE !== false && state.dnsPass) {
|
|
doAutoVotes().catch(err => {
|
|
logError('Swarm', `Error in immediate auto-votes after peer disconnect: ${err.message}`);
|
|
});
|
|
}
|
|
|
|
// Master node: Track peer for reconnection and actively attempt to reconnect
|
|
if (isMaster && state.dnsPass) {
|
|
// Clear any existing reconnection timeout
|
|
const channels = peerChannels.get(peerId);
|
|
if (channels && channels.reconnectTimeout) {
|
|
clearTimeout(channels.reconnectTimeout);
|
|
channels.reconnectTimeout = null;
|
|
}
|
|
|
|
// Don't reconnect if peer is blocked
|
|
if (state.blockedPeers && state.blockedPeers.has(peerId)) {
|
|
logDebug('Swarm', `Not attempting reconnection to blocked peer ${peerId}`);
|
|
return;
|
|
}
|
|
|
|
// Add peer to reconnection tracking
|
|
if (!state.peersToReconnect) {
|
|
state.peersToReconnect = new Set();
|
|
}
|
|
if (!state.reconnectionAttempts) {
|
|
state.reconnectionAttempts = new Map();
|
|
}
|
|
if (!state.lastReconnectionAttempt) {
|
|
state.lastReconnectionAttempt = new Map();
|
|
}
|
|
|
|
// Only add if not already in set (avoid duplicates)
|
|
if (!state.peersToReconnect.has(peerId)) {
|
|
logInfo('Swarm', `Adding peer ${peerId} to reconnection tracking`);
|
|
state.peersToReconnect.add(peerId);
|
|
// Reset attempt count for new reconnection cycle
|
|
state.reconnectionAttempts.set(peerId, 0);
|
|
|
|
// Start reconnection attempt after initial delay
|
|
const baseInterval = parseInt(process.env.MASTER_RECONNECT_INTERVAL || '5', 10) * 1000;
|
|
setTimeout(() => {
|
|
attemptReconnectToPeer(peerId, swarm, topic);
|
|
}, baseInterval);
|
|
}
|
|
}
|
|
});
|
|
conn.on('error', async (err) => {
|
|
const errMsg = err.message || '';
|
|
const errCode = err.code || '';
|
|
|
|
// Handle duplicate connection errors gracefully - these are expected during reconnections
|
|
if (errMsg.includes('Duplicate connection') || errMsg.includes('duplicate')) {
|
|
logDebug('Swarm', `Duplicate connection error for peer ${peerId} - letting main connection logic handle it`);
|
|
// Don't interfere with duplicate connection handling - let the main connection logic handle it
|
|
// The main logic already properly detects and handles duplicate connections
|
|
return; // Don't log as error, don't do disconnect handling
|
|
}
|
|
|
|
// Handle common benign network errors - these are normal in P2P networks
|
|
const benignErrors = [
|
|
'connection reset by peer',
|
|
'ECONNRESET',
|
|
'EPIPE',
|
|
'broken pipe',
|
|
'socket hang up',
|
|
'ECONNABORTED',
|
|
'ETIMEDOUT',
|
|
'ENOTFOUND',
|
|
'ECONNREFUSED'
|
|
];
|
|
|
|
const isBenignError = benignErrors.some(benign =>
|
|
errMsg.toLowerCase().includes(benign.toLowerCase()) ||
|
|
errCode === benign
|
|
);
|
|
|
|
if (isBenignError) {
|
|
// Log as debug/warning instead of error - these are normal network events
|
|
logDebug('Swarm', `Connection reset/closed by peer ${peerId.slice(0, 16)}... (${errMsg || errCode})`);
|
|
|
|
// Still do cleanup but don't treat as a serious error
|
|
try {
|
|
const channels = peerChannels.get(peerId);
|
|
if (channels && channels.inviteRequestTimeout) {
|
|
clearTimeout(channels.inviteRequestTimeout);
|
|
channels.inviteRequestTimeout = null;
|
|
}
|
|
if (channels && channels.reconnectTimeout) {
|
|
clearTimeout(channels.reconnectTimeout);
|
|
channels.reconnectTimeout = null;
|
|
}
|
|
|
|
// Handle plugin channel disconnection
|
|
try {
|
|
const channelManager = require('./includes/plugins/channel-manager');
|
|
channelManager.handlePeerDisconnect(peerId);
|
|
} catch (pluginErr) {
|
|
logDebug('Swarm', `Error handling plugin channel disconnect: ${pluginErr.message}`);
|
|
}
|
|
|
|
// Remove from peer channels and connected peers
|
|
peerChannels.delete(peerId);
|
|
connectedPeers.delete(peerId);
|
|
|
|
// Track peer disconnection
|
|
const disconnectTime = Date.now();
|
|
const connectTime = state.peerStartTimes.get(peerId);
|
|
const duration = connectTime ? disconnectTime - connectTime : 0;
|
|
|
|
if (state.peerHistory.has(peerId)) {
|
|
const history = state.peerHistory.get(peerId);
|
|
history.push({
|
|
type: 'disconnect',
|
|
timestamp: disconnectTime,
|
|
duration,
|
|
error: errMsg || errCode
|
|
});
|
|
if (history.length > 50) {
|
|
state.peerHistory.set(peerId, history.slice(-50));
|
|
}
|
|
}
|
|
|
|
if (state.peerMetrics.has(peerId)) {
|
|
const metrics = state.peerMetrics.get(peerId);
|
|
if (duration > 0) {
|
|
metrics.totalDuration += duration;
|
|
metrics.avgDuration = metrics.totalDuration / metrics.connections;
|
|
}
|
|
await savePeerMetrics();
|
|
await savePeerHistory();
|
|
}
|
|
|
|
state.peerStartTimes.delete(peerId);
|
|
prunePeerTrackingMaps();
|
|
trackPeerEvent('disconnect', peerId);
|
|
broadcast({ type: 'update-peers' });
|
|
broadcast({ type: 'update-stats' });
|
|
} catch (cleanupErr) {
|
|
logDebug('Swarm', `Error during benign error cleanup: ${cleanupErr.message}`);
|
|
}
|
|
return; // Don't log as error, cleanup is done
|
|
}
|
|
|
|
// For other unexpected errors, log as error
|
|
logError('Swarm', `Connection error with peer ${peerId}: ${err.message}`);
|
|
|
|
// Clear invite request timeout if it exists
|
|
const channels = peerChannels.get(peerId);
|
|
if (channels && channels.inviteRequestTimeout) {
|
|
clearTimeout(channels.inviteRequestTimeout);
|
|
channels.inviteRequestTimeout = null;
|
|
}
|
|
|
|
// Remove all event listeners from connection
|
|
conn.removeAllListeners();
|
|
|
|
// Track peer disconnection (same as close handler)
|
|
const disconnectTime = Date.now();
|
|
const connectTime = state.peerStartTimes.get(peerId);
|
|
const duration = connectTime ? disconnectTime - connectTime : 0;
|
|
|
|
if (state.peerHistory.has(peerId)) {
|
|
const history = state.peerHistory.get(peerId);
|
|
history.push({
|
|
type: 'disconnect',
|
|
timestamp: disconnectTime,
|
|
duration,
|
|
error: err.message
|
|
});
|
|
// Keep only last 50 items
|
|
if (history.length > 50) {
|
|
state.peerHistory.set(peerId, history.slice(-50));
|
|
}
|
|
}
|
|
|
|
if (state.peerMetrics.has(peerId)) {
|
|
const metrics = state.peerMetrics.get(peerId);
|
|
if (duration > 0) {
|
|
metrics.totalDuration += duration;
|
|
metrics.avgDuration = metrics.totalDuration / metrics.connections;
|
|
}
|
|
// Save peer metrics and history to disk
|
|
await savePeerMetrics();
|
|
await savePeerHistory();
|
|
}
|
|
|
|
state.peerStartTimes.delete(peerId);
|
|
prunePeerTrackingMaps();
|
|
|
|
connectedPeers.delete(peerId);
|
|
peerChannels.delete(peerId);
|
|
trackPeerEvent('disconnect', peerId);
|
|
broadcast({ type: 'update-peers' });
|
|
broadcast({ type: 'update-stats' });
|
|
|
|
// Immediate consensus update on peer disconnect (error case)
|
|
const { validateConfig } = require('./includes/infrastructure/config');
|
|
let config;
|
|
try {
|
|
config = validateConfig();
|
|
} catch (e) {
|
|
config = { CONSENSUS_IMMEDIATE_UPDATE: true };
|
|
}
|
|
|
|
if (config.CONSENSUS_IMMEDIATE_UPDATE !== false && state.dnsPass) {
|
|
doAutoVotes().catch(err => {
|
|
logError('Swarm', `Error in immediate auto-votes after peer disconnect (error): ${err.message}`);
|
|
});
|
|
}
|
|
});
|
|
});
|
|
// Log swarm events
|
|
swarm.on('update', () => {
|
|
logInfo('Swarm', `Swarm updated. Connections: ${swarm.connections.size}`);
|
|
});
|
|
swarm.on('error', (err) => {
|
|
logError('Swarm', `Swarm error: ${err.message}`);
|
|
});
|
|
|
|
// Function to auto-subscribe to services on bootup
|
|
async function autoSubscribeToServices() {
|
|
const disableAutoSubscription = process.env.DISABLE_AUTO_SUBSCRIPTION === 'true';
|
|
if (disableAutoSubscription) {
|
|
logInfo('Main', 'Auto-subscription disabled via DISABLE_AUTO_SUBSCRIPTION=true');
|
|
return;
|
|
}
|
|
|
|
if (!state.dnsPass) {
|
|
logWarn('Main', 'dnsPass not initialized, skipping auto-subscription');
|
|
return;
|
|
}
|
|
|
|
let subscriptions = [];
|
|
|
|
try {
|
|
const { loadSubscriptions } = require('./includes/admin/subscription-manager');
|
|
const allSubscriptions = await loadSubscriptions();
|
|
// Convert to flat list for compatibility
|
|
for (const domainSub of allSubscriptions) {
|
|
for (const service of domainSub.services) {
|
|
subscriptions.push({
|
|
domain: domainSub.domain,
|
|
serviceName: service.serviceName,
|
|
key: service.key,
|
|
port: service.port,
|
|
protocol: service.protocol
|
|
});
|
|
}
|
|
}
|
|
} catch (err) {
|
|
logError('Main', `Error loading subscriptions: ${err.message}`);
|
|
return;
|
|
}
|
|
|
|
if (subscriptions.length === 0) {
|
|
logDebug('Main', 'No subscriptions to auto-subscribe');
|
|
return;
|
|
}
|
|
|
|
logInfo('Main', `Auto-subscribing to ${subscriptions.length} service(s)...`);
|
|
|
|
const { getConsensusState, getClaimClients } = require('./includes/core/core');
|
|
const { startForkedHolesailClient, saveHolesailClients } = require('./includes/admin');
|
|
const { ensurePortFree } = require('./includes/admin/admin-backend/port-management');
|
|
|
|
for (const subscription of subscriptions) {
|
|
try {
|
|
const { domain, serviceName, key, port, protocol } = subscription;
|
|
|
|
// Check if claim record has matching service
|
|
const consensusState = await getConsensusState(domain);
|
|
if (!consensusState.resolvedClaimant) {
|
|
logWarn('Main', `Domain ${domain} has no resolved claimant, skipping subscription`);
|
|
continue;
|
|
}
|
|
|
|
const clients = await getClaimClients(domain, consensusState.resolvedClaimant);
|
|
const service = clients.find(c => c.name === serviceName);
|
|
|
|
if (!service) {
|
|
logWarn('Main', `Service ${serviceName} not found in claim for ${domain}, skipping subscription`);
|
|
continue;
|
|
}
|
|
|
|
// Verify service matches subscription
|
|
if (service.key !== key || service.port !== port) {
|
|
logWarn('Main', `Service ${serviceName} for ${domain} has changed, skipping subscription`);
|
|
continue;
|
|
}
|
|
|
|
// Check if client already exists
|
|
const clientId = `${domain}_${serviceName}`.replace(/[^a-zA-Z0-9_]/g, '_');
|
|
if (state.holesailClientOpts.has(clientId)) {
|
|
logDebug('Main', `Client ${clientId} already exists, skipping auto-subscription`);
|
|
continue;
|
|
}
|
|
|
|
// Create interface if needed
|
|
if (!state.domainToIPMap.has(domain)) {
|
|
await createInterfaceForDomain(domain);
|
|
logDebug('Main', `Assigned IP to ${domain}: ${state.domainToIPMap.get(domain)}`);
|
|
}
|
|
|
|
const ip = state.domainToIPMap.get(domain);
|
|
const portFree = await ensurePortFree(ip, port);
|
|
if (!portFree) {
|
|
logWarn('Main', `Unable to ensure port ${port} free on ${ip} for subscription ${domain}/${serviceName}`);
|
|
continue;
|
|
}
|
|
|
|
// Create holesail client
|
|
state.holesailClientInfos.set(clientId, { state: 'starting' });
|
|
await startForkedHolesailClient(clientId, {
|
|
domain,
|
|
key,
|
|
port: parseInt(port),
|
|
protocol: protocol || 'tcp'
|
|
});
|
|
state.holesailClientInfos.set(clientId, { ...state.holesailClientInfos.get(clientId), state: 'running' });
|
|
await saveHolesailClients();
|
|
logInfo('Main', `Auto-subscribed to ${domain}/${serviceName} (client: ${clientId})`);
|
|
} catch (err) {
|
|
logError('Main', `Error auto-subscribing to ${subscription.domain}/${subscription.serviceName}: ${err.message}`);
|
|
}
|
|
}
|
|
|
|
logInfo('Main', 'Auto-subscription completed');
|
|
}
|
|
|
|
// Function to list all domains and resolved hashes
|
|
async function listDomains() {
|
|
try {
|
|
if (!state.dnsPass) {
|
|
logWarn('Main', 'dnsPass not initialized yet, cannot list domains');
|
|
return;
|
|
}
|
|
logDebug('Main', 'Listing all resolved domains...');
|
|
const allEntries = await getAllEntries();
|
|
const domains = new Set();
|
|
for (const entry of allEntries) {
|
|
if (entry.key.startsWith('claim:')) {
|
|
const parts = entry.key.split(':');
|
|
if (parts.length === 3) {
|
|
domains.add(parts[1]);
|
|
}
|
|
}
|
|
}
|
|
const resolved = [];
|
|
for (const domain of domains) {
|
|
const hash = await getHashForDomain(domain);
|
|
resolved.push(`Domain: ${domain}, Resolved Hash: ${hash || 'none'}`);
|
|
}
|
|
// Include internal domains
|
|
for (const domain of internalDomains) {
|
|
resolved.push(`Domain: ${domain}, Resolved Hash: none (internal)`);
|
|
}
|
|
resolved.forEach(res => logDebug('Main', res));
|
|
logInfo('Main', `--- End of domains (${domains.size + internalDomains.length} domains) ---`);
|
|
} catch (err) {
|
|
logError('Main', `Error listing domains: ${err.message}`);
|
|
}
|
|
}
|
|
// Add ability to add domains via command line
|
|
process.stdin.setEncoding('utf8');
|
|
logDebug('Main', 'Type "add <domain> <hash>" and press Enter to add a domain (or Ctrl+C to exit):');
|
|
process.stdin.on('data', async (input) => {
|
|
const trimmed = input.trim();
|
|
if (trimmed.startsWith('add ')) {
|
|
try {
|
|
if (!state.dnsPass) {
|
|
logWarn('Main', 'Not connected yet, cannot add domain');
|
|
return;
|
|
}
|
|
const parts = trimmed.split(' ');
|
|
if (parts.length >= 3) {
|
|
const domain = parts[1];
|
|
const hash = parts.slice(2).join(' ');
|
|
await addDomain(domain, hash);
|
|
logInfo('Main', `Added domain: "${domain}" with hash "${hash}"`);
|
|
if (!state.domainToIPMap.has(domain)) {
|
|
await createInterfaceForDomain(domain);
|
|
logDebug('Main', `Assigned IP to domain: ${domain} (${state.domainToIPMap.get(domain)})`);
|
|
}
|
|
} else {
|
|
logWarn('Main', 'Invalid format. Use: add <domain> <hash>');
|
|
}
|
|
} catch (err) {
|
|
logError('Main', `Error adding domain: ${err.message}`);
|
|
}
|
|
}
|
|
});
|
|
// Setup servers conditionally
|
|
if (process.env.DISABLE_DNS_SERVER !== 'true') {
|
|
bindDnsServer();
|
|
logInfo('Main', 'DNS server setup initiated');
|
|
} else {
|
|
logInfo('Main', 'DNS server disabled via DISABLE_DNS_SERVER=true');
|
|
}
|
|
if (process.env.DISABLE_PROXY_SERVER !== 'true') {
|
|
setupProxyServer();
|
|
// Start health checks for holesail connections
|
|
const { startHealthChecks } = require('./includes/networking/holesail');
|
|
startHealthChecks();
|
|
logInfo('Main', 'Proxy server setup initiated');
|
|
} else {
|
|
logInfo('Main', 'Proxy server disabled via DISABLE_PROXY_SERVER=true');
|
|
}
|
|
// Load blocked peers
|
|
await loadBlockedPeers();
|
|
// Load peer metrics
|
|
await loadPeerMetrics();
|
|
// Load peer history
|
|
await loadPeerHistory();
|
|
// Load and start persisted Holesail servers and clients
|
|
await loadHolesailServers();
|
|
await assignAllIPs();
|
|
await loadHolesailClients();
|
|
|
|
// Initialize subscription manager
|
|
try {
|
|
const { initializeSubscriptionManager } = require('./includes/admin/subscription-manager');
|
|
await initializeSubscriptionManager();
|
|
logInfo('Main', 'Subscription manager initialized');
|
|
} catch (err) {
|
|
logWarn('Main', `Error initializing subscription manager: ${err.message}`);
|
|
}
|
|
|
|
// Auto-subscribe to services on bootup
|
|
await autoSubscribeToServices();
|
|
|
|
// Start automatic backups
|
|
const { startAutomaticBackups } = require('./includes/maintenance/backup');
|
|
startAutomaticBackups();
|
|
logInfo('Main', 'Automatic backup system started');
|
|
// Keep the process running
|
|
// Start process metrics collection
|
|
startProcessMetricsCollection(secondsToMs(5)); // Collect every 5 seconds
|
|
logInfo('Main', 'Process metrics collection started');
|
|
|
|
// Start resource validation
|
|
const { startResourceValidation, stopResourceValidation } = require('./includes/maintenance/resource_validation');
|
|
startResourceValidation();
|
|
logInfo('Main', 'Resource validation started');
|
|
|
|
logInfo('Main', 'Process will continue running. Press Ctrl+C to exit.');
|
|
// Enhanced cleanup on exit
|
|
const cleanup = async () => {
|
|
if (isShuttingDown) return;
|
|
isShuttingDown = true;
|
|
state.isShuttingDown = true; // Make shutdown status available to other modules
|
|
logInfo('Main', 'Shutting down gracefully...');
|
|
|
|
const shutdownConfig = getShutdownConfig();
|
|
if (shutdownConfig.fast) {
|
|
logInfo('Main', 'Fast shutdown mode enabled (SHUTDOWN_MODE=fast)');
|
|
}
|
|
|
|
// Start periodic "system is shutting down" messages every 3 seconds
|
|
let shutdownMessageInterval = setInterval(() => {
|
|
logInfo('Main', '🔄 SYSTEM IS SHUTTING DOWN - Please wait for graceful cleanup to complete...');
|
|
}, 3000);
|
|
try {
|
|
// =======================================================================
|
|
// PHASE 1: Clear all timers IMMEDIATELY (before any async operations)
|
|
// =======================================================================
|
|
|
|
// Clear auto-vote debounce timer
|
|
if (listenerRefs.autoVoteDebounce) {
|
|
clearTimeout(listenerRefs.autoVoteDebounce);
|
|
listenerRefs.autoVoteDebounce = null;
|
|
}
|
|
// Clear network consensus timer (calls doAutoVotes which uses dnsPass)
|
|
if (state.networkConsensusTimer) {
|
|
clearInterval(state.networkConsensusTimer);
|
|
state.networkConsensusTimer = null;
|
|
logDebug('Main', 'Cleared network consensus timer');
|
|
}
|
|
// Clear swarm keep-alive timer
|
|
if (state.swarmKeepAliveTimer) {
|
|
clearInterval(state.swarmKeepAliveTimer);
|
|
state.swarmKeepAliveTimer = null;
|
|
logDebug('Main', 'Cleared swarm keep-alive timer');
|
|
}
|
|
// Clear domains debounce timer
|
|
if (domainsDebounceTimer) {
|
|
clearTimeout(domainsDebounceTimer);
|
|
domainsDebounceTimer = null;
|
|
}
|
|
// Clear persistent invite retry interval if exists
|
|
if (state.persistentInviteRetryInterval) {
|
|
clearInterval(state.persistentInviteRetryInterval);
|
|
state.persistentInviteRetryInterval = null;
|
|
}
|
|
|
|
// =======================================================================
|
|
// PHASE 2: Remove event listeners from dnsPass and core (before closing them)
|
|
// =======================================================================
|
|
|
|
logDebug('Main', 'Removing event listeners from dnsPass and core...');
|
|
if (listenerRefs.dnsPassUpdate && state.dnsPass) {
|
|
state.dnsPass.removeListener('update', listenerRefs.dnsPassUpdate);
|
|
}
|
|
if (listenerRefs.corePeerAdd && core) {
|
|
core.removeListener('peer-add', listenerRefs.corePeerAdd);
|
|
}
|
|
if (listenerRefs.coreAppend && core) {
|
|
core.removeListener('append', listenerRefs.coreAppend);
|
|
}
|
|
logDebug('Main', 'Removed event listeners from dnsPass and core');
|
|
|
|
// =======================================================================
|
|
// PHASE 2.5: Close pairing operations and DNSPass
|
|
// =======================================================================
|
|
|
|
// Close any active pairing operations FIRST (they may hold swarm references)
|
|
if (currentPairOperation && typeof currentPairOperation.close === 'function') {
|
|
try {
|
|
logDebug('Main', 'Closing active pairing operation...');
|
|
await currentPairOperation.close();
|
|
logDebug('Main', 'Active pairing operation closed successfully');
|
|
} catch (err) {
|
|
// Ignore store close errors since we share the store with DNSPass
|
|
if (err.message.includes('Corestore is closed') ||
|
|
err.message.includes('already closed') ||
|
|
err.message.includes('store is closed')) {
|
|
logDebug('Main', 'Pairing operation store already closed by shared corestore');
|
|
} else {
|
|
logWarn('Main', `Error closing active pairing operation: ${err.message}`);
|
|
}
|
|
}
|
|
}
|
|
|
|
// Close consensus sidecar before dnsPass
|
|
try {
|
|
const { closeConsensusAutobase } = require('./includes/core/consensus-autobase');
|
|
await closeConsensusAutobase();
|
|
logDebug('Main', 'Consensus sidecar closed');
|
|
} catch (err) {
|
|
logWarn('Main', `Error closing consensus sidecar: ${err.message}`);
|
|
}
|
|
|
|
// Close DNSPass after pairing operations (main component)
|
|
if (state.dnsPass && typeof state.dnsPass.close === 'function') {
|
|
logDebug('Main', 'Closing dnsPass (main component)...');
|
|
try {
|
|
await state.dnsPass.close();
|
|
logDebug('Main', 'dnsPass closed successfully - Autobase core should be closed');
|
|
} catch (err) {
|
|
// Ignore errors if dnsPass was already cleaned/reset
|
|
if (err.message.includes('Autobase failed to open') ||
|
|
err.message.includes('Corestore is closed') ||
|
|
err.message.includes('already closed')) {
|
|
logDebug('Main', 'dnsPass already cleaned/closed, skipping cleanup');
|
|
} else {
|
|
logError('Main', `Error closing dnsPass: ${err.message}`);
|
|
}
|
|
}
|
|
} else {
|
|
logDebug('Main', 'dnsPass not available or already closed');
|
|
}
|
|
|
|
// Verify dnsPass base core is closed (dnsPass.close() should handle this)
|
|
if (state.core && typeof state.core.close === 'function' && !state.core.closed) {
|
|
logWarn('Main', 'dnsPass base core still open after dnsPass.close(), closing explicitly...');
|
|
try {
|
|
await state.core.close();
|
|
logDebug('Main', 'dnsPass base core closed explicitly');
|
|
} catch (err) {
|
|
if (err.message.includes('Corestore is closed') ||
|
|
err.message.includes('already closed')) {
|
|
logDebug('Main', 'dnsPass base core was already closed');
|
|
} else {
|
|
logError('Main', `Error closing dnsPass base core: ${err.message}`);
|
|
}
|
|
}
|
|
}
|
|
|
|
// =======================================================================
|
|
// PHASE 3: Stop swarm (dnsPass is now closed, no risk of triggering access)
|
|
// =======================================================================
|
|
|
|
// Close any remaining swarm connections
|
|
logDebug('Main', 'Closing swarm connections...');
|
|
const connections = Array.from(swarm.connections);
|
|
for (const conn of connections) {
|
|
try {
|
|
if (!conn.destroyed) {
|
|
conn.removeAllListeners();
|
|
conn.destroy();
|
|
}
|
|
} catch (err) {
|
|
logError('Main', `Error destroying connection: ${err.message}`);
|
|
}
|
|
}
|
|
|
|
// =======================================================================
|
|
// PHASE 3.5: Close HTTP/TLS servers IMMEDIATELY to prevent new requests
|
|
// =======================================================================
|
|
|
|
await closeAllHttpTlsServers(state.tlsServers, state.httpServers, shutdownConfig);
|
|
|
|
// Close DNS server to prevent DNS queries during shutdown
|
|
const { closeDnsServer, stopDNSCacheCleanup } = require('./includes/networking/dns');
|
|
logDebug('Main', 'Closing DNS server...');
|
|
try {
|
|
stopDNSCacheCleanup();
|
|
closeDnsServer();
|
|
logDebug('Main', 'DNS server closed');
|
|
} catch (err) {
|
|
logError('Main', `Error closing DNS server: ${err.message}`);
|
|
}
|
|
|
|
// Brief delay to let in-flight server operations finish
|
|
if (shutdownConfig.serverSettleMs > 0) {
|
|
logDebug('Main', `Waiting ${shutdownConfig.serverSettleMs}ms for server operations to settle...`);
|
|
await sleep(shutdownConfig.serverSettleMs);
|
|
}
|
|
|
|
// Remove swarm event listeners to prevent memory leaks
|
|
logDebug('Main', 'Removing swarm event listeners...');
|
|
try {
|
|
swarm.removeAllListeners('connection');
|
|
swarm.removeAllListeners('update');
|
|
swarm.removeAllListeners('error');
|
|
logDebug('Main', 'Removed swarm event listeners');
|
|
} catch (err) {
|
|
logError('Main', `Error removing swarm event listeners: ${err.message}`);
|
|
}
|
|
// Leave swarm topics
|
|
logDebug('Main', 'Leaving swarm topics...');
|
|
try {
|
|
swarm.leave(topic);
|
|
await swarm.flush();
|
|
} catch (err) {
|
|
logError('Main', `Error leaving topic: ${err.message}`);
|
|
}
|
|
// Immediately destroy swarm to prevent new connections during cleanup
|
|
logDebug('Main', 'Destroying swarm to prevent new connections...');
|
|
try {
|
|
await swarm.destroy();
|
|
} catch (err) {
|
|
logError('Main', `Error destroying swarm: ${err.message}`);
|
|
}
|
|
|
|
// =======================================================================
|
|
// PHASE 4: Stop background services that might access corestore
|
|
// =======================================================================
|
|
|
|
// Stop metrics collection
|
|
const { stopProcessMetricsCollection } = require('./includes/maintenance/metrics');
|
|
logDebug('Main', 'Stopping process metrics collection...');
|
|
try {
|
|
stopProcessMetricsCollection();
|
|
} catch (err) {
|
|
logError('Main', `Error stopping metrics collection: ${err.message}`);
|
|
}
|
|
// Stop automatic backups
|
|
const { stopAutomaticBackups } = require('./includes/maintenance/backup');
|
|
logDebug('Main', 'Stopping automatic backups...');
|
|
try {
|
|
stopAutomaticBackups();
|
|
} catch (err) {
|
|
logError('Main', `Error stopping backups: ${err.message}`);
|
|
}
|
|
// Stop resource validation
|
|
const { stopResourceValidation } = require('./includes/maintenance/resource_validation');
|
|
logDebug('Main', 'Stopping resource validation...');
|
|
try {
|
|
stopResourceValidation();
|
|
} catch (err) {
|
|
logError('Main', `Error stopping resource validation: ${err.message}`);
|
|
}
|
|
// Stop Holesail health checks
|
|
const { stopHealthChecks } = require('./includes/networking/holesail');
|
|
logDebug('Main', 'Stopping Holesail health checks...');
|
|
try {
|
|
stopHealthChecks();
|
|
} catch (err) {
|
|
logError('Main', `Error stopping Holesail health checks: ${err.message}`);
|
|
}
|
|
|
|
// =======================================================================
|
|
// PHASE 5: Shutdown plugins (plugins may have listeners or access dnsPass)
|
|
// =======================================================================
|
|
|
|
if (state.shutdownPlugins) {
|
|
logDebug('Main', 'Shutting down plugins...');
|
|
try {
|
|
await state.shutdownPlugins();
|
|
} catch (err) {
|
|
logError('Main', `Error shutting down plugins: ${err.message}`);
|
|
}
|
|
}
|
|
|
|
// =======================================================================
|
|
// PHASE 6: Cleanup replication managers (they use corestore)
|
|
// =======================================================================
|
|
|
|
const replicationCleanupTasks = [];
|
|
if (state.replicationManager) {
|
|
replicationCleanupTasks.push(
|
|
state.replicationManager.cleanupAll().catch((err) => {
|
|
logWarn('Main', `Error cleaning up replication manager: ${err.message}`);
|
|
})
|
|
);
|
|
}
|
|
if (state.driveReplicationManager) {
|
|
replicationCleanupTasks.push(
|
|
state.driveReplicationManager.cleanupAll().catch((err) => {
|
|
logWarn('Main', `Error cleaning up drive replication manager: ${err.message}`);
|
|
})
|
|
);
|
|
}
|
|
if (replicationCleanupTasks.length > 0) {
|
|
logDebug('Main', 'Cleaning up replication managers in parallel...');
|
|
await Promise.allSettled(replicationCleanupTasks);
|
|
}
|
|
|
|
// Wait for replication cleanup to fully settle before closing corestore
|
|
if (shutdownConfig.replicationSettleMs > 0) {
|
|
logDebug('Main', `Waiting ${shutdownConfig.replicationSettleMs}ms for replication operations to settle...`);
|
|
await sleep(shutdownConfig.replicationSettleMs);
|
|
}
|
|
|
|
// =======================================================================
|
|
// PHASE 7: Close all plugin drives (before closing dnsPass)
|
|
// =======================================================================
|
|
|
|
logDebug('Main', 'Closing all plugin drives...');
|
|
try {
|
|
const { closeAllDrives } = require('./includes/plugins/drive-manager');
|
|
await closeAllDrives();
|
|
logDebug('Main', 'All plugin drives closed');
|
|
} catch (err) {
|
|
logError('Main', `Error closing all drives: ${err.message}`);
|
|
}
|
|
|
|
// =======================================================================
|
|
// PHASE 8: Close invite processing
|
|
// =======================================================================
|
|
|
|
// Clear any pending invite requests
|
|
if (state.pendingInviteRequests && state.pendingInviteRequests.size > 0) {
|
|
logDebug('Main', `Clearing ${state.pendingInviteRequests.size} pending invite requests during shutdown`);
|
|
state.pendingInviteRequests.clear();
|
|
}
|
|
|
|
// Wait for any current invite processing to complete
|
|
if (currentInvitePromise) {
|
|
try {
|
|
logDebug('Main', 'Waiting for current invite processing to complete...');
|
|
await withTimeout(currentInvitePromise, shutdownConfig.inviteProcessingTimeoutMs, 'Current invite processing shutdown');
|
|
logDebug('Main', 'Current invite processing completed during shutdown');
|
|
} catch (err) {
|
|
logWarn('Main', `Current invite processing did not complete during shutdown: ${err.message}`);
|
|
}
|
|
}
|
|
|
|
// =======================================================================
|
|
// PHASE 9: Close main corestore (dnsPass already closed in PHASE 2.5)
|
|
// =======================================================================
|
|
|
|
if (store && typeof store.close === 'function' && !store.closed) {
|
|
logDebug('Main', 'Closing main corestore...');
|
|
try {
|
|
await store.close();
|
|
logDebug('Main', 'Main corestore closed successfully');
|
|
} catch (err) {
|
|
// Ignore errors if corestore was already cleaned/reset
|
|
if (err.message.includes('Corestore is closed') ||
|
|
err.message.includes('already closed')) {
|
|
logDebug('Main', 'Main corestore already cleaned/closed, skipping cleanup');
|
|
} else {
|
|
logError('Main', `Error closing main corestore: ${err.message}`);
|
|
}
|
|
}
|
|
} else {
|
|
logDebug('Main', 'Main corestore not available or already closed');
|
|
}
|
|
|
|
// =======================================================================
|
|
// PHASE 9: Close domains watcher (might trigger domain operations)
|
|
// =======================================================================
|
|
|
|
if (domainsWatcher) {
|
|
logDebug('Main', 'Closing domains file watcher...');
|
|
try {
|
|
domainsWatcher.close();
|
|
} catch (err) {
|
|
logError('Main', `Error closing domains watcher: ${err.message}`);
|
|
}
|
|
}
|
|
|
|
// =======================================================================
|
|
// PHASE 9: Close protomux channels (dnsPass is already closed)
|
|
// =======================================================================
|
|
|
|
logDebug('Main', 'Closing protomux channels...');
|
|
for (const [peerId, channels] of peerChannels) {
|
|
try {
|
|
// Clear any pending timeouts
|
|
if (channels.inviteRequestTimeout) {
|
|
clearTimeout(channels.inviteRequestTimeout);
|
|
}
|
|
if (channels.reconnectTimeout) {
|
|
clearTimeout(channels.reconnectTimeout);
|
|
}
|
|
// Handle plugin channel disconnection (includes core p2ns channels)
|
|
channelManager.handlePeerDisconnect(peerId);
|
|
} catch (err) {
|
|
logError('Main', `Error closing channels for peer ${peerId}: ${err.message}`);
|
|
}
|
|
}
|
|
peerChannels.clear();
|
|
// Unregister core p2ns channels
|
|
channelManager.unregisterAllPluginChannels(CORE_DOMAIN);
|
|
logDebug('Main', 'Protomux channels closed');
|
|
|
|
// =======================================================================
|
|
// PHASE 10: Close Holesail servers and clients
|
|
// =======================================================================
|
|
|
|
// Close Holesail servers and clients (in parallel)
|
|
const holesailChildKillPromises = [
|
|
...Array.from(state.holesailChildren.entries()).map(async ([id, child]) => {
|
|
await killHolesailChildProcess(child, id, shutdownConfig, 'server');
|
|
logInfo('Main', `Killed child process for server ${id}`);
|
|
}),
|
|
...Array.from(state.holesailClientChildren.entries()).map(async ([id, child]) => {
|
|
await killHolesailChildProcess(child, id, shutdownConfig, 'client');
|
|
logInfo('Main', `Killed child process for client ${id}`);
|
|
})
|
|
];
|
|
await Promise.allSettled(holesailChildKillPromises);
|
|
state.holesailChildren.clear();
|
|
state.holesailOpts.clear();
|
|
state.holesailInfos.clear();
|
|
state.holesailClientChildren.clear();
|
|
state.holesailClientOpts.clear();
|
|
state.holesailClientInfos.clear();
|
|
// Close Holesail client connections (in parallel)
|
|
const holesailClosePromises = Array.from(state.holesails.entries()).map(async ([key, holesail]) => {
|
|
try {
|
|
if (holesail instanceof dgram.Socket) {
|
|
await new Promise(resolve => {
|
|
holesail.close(() => {
|
|
logInfo('Main', `Closed UDP Holesail client connection for ${key}`);
|
|
resolve();
|
|
});
|
|
setTimeout(() => {
|
|
logWarn('Main', `Timeout closing UDP Holesail for ${key}, forcing closure`);
|
|
holesail.close();
|
|
resolve();
|
|
}, shutdownConfig.holesailUdpCloseTimeoutMs);
|
|
});
|
|
} else {
|
|
await holesail.close();
|
|
logInfo('Main', `Closed TCP Holesail client connection for ${key}`);
|
|
}
|
|
if (!shutdownConfig.skipPortReleaseWait) {
|
|
const [domain, port] = key.split(':');
|
|
const ip = state.domainToIPMap.get(domain);
|
|
if (ip && port) {
|
|
const isPortFree = await waitForPortRelease(ip, parseInt(port));
|
|
if (!isPortFree) {
|
|
logError('Main', `Port ${port} on ${ip} for ${domain} still in use after cleanup`);
|
|
}
|
|
}
|
|
}
|
|
} catch (err) {
|
|
logError('Main', `Error closing Holesail client for ${key}: ${err.message}`);
|
|
}
|
|
});
|
|
await Promise.allSettled(holesailClosePromises);
|
|
state.holesails.clear();
|
|
// Perform interface cleanup
|
|
await cleanupInterfaces();
|
|
// Close DNS pool
|
|
if (dnsPool && typeof dnsPool.close === 'function') {
|
|
logDebug('Main', 'Closing DNS pool...');
|
|
try {
|
|
dnsPool.close();
|
|
} catch (err) {
|
|
logError('Main', `Error closing DNS pool: ${err.message}`);
|
|
}
|
|
}
|
|
// Destroy rate limiter
|
|
if (rateLimiter && typeof rateLimiter.destroy === 'function') {
|
|
logDebug('Main', 'Destroying rate limiter...');
|
|
try {
|
|
rateLimiter.destroy();
|
|
} catch (err) {
|
|
logError('Main', `Error destroying rate limiter: ${err.message}`);
|
|
}
|
|
}
|
|
// Close stdin
|
|
try {
|
|
process.stdin.pause();
|
|
process.stdin.destroy();
|
|
logDebug('Main', 'Closed stdin');
|
|
} catch (err) {
|
|
logError('Main', `Error closing stdin: ${err.message}`);
|
|
}
|
|
// Remove stdin event listeners to prevent memory leaks
|
|
logDebug('Main', 'Removing stdin event listeners...');
|
|
try {
|
|
process.stdin.removeAllListeners('data');
|
|
logDebug('Main', 'Removed stdin event listeners');
|
|
} catch (err) {
|
|
logError('Main', `Error removing stdin event listeners: ${err.message}`);
|
|
}
|
|
// Restore console methods
|
|
const { restoreConsoleMethods } = require('./includes/admin');
|
|
logDebug('Main', 'Restoring console methods...');
|
|
try {
|
|
restoreConsoleMethods();
|
|
} catch (err) {
|
|
logError('Main', `Error restoring console methods: ${err.message}`);
|
|
}
|
|
// Close WebSocket connections
|
|
const { closeAllWebSockets } = require('./includes/admin');
|
|
logDebug('Main', 'Closing WebSocket connections...');
|
|
try {
|
|
closeAllWebSockets();
|
|
} catch (err) {
|
|
logError('Main', `Error closing WebSocket connections: ${err.message}`);
|
|
}
|
|
// Clear shutdown message interval
|
|
if (shutdownMessageInterval) {
|
|
clearInterval(shutdownMessageInterval);
|
|
shutdownMessageInterval = null;
|
|
}
|
|
|
|
logInfo('Main', 'Cleanup completed successfully');
|
|
} catch (err) {
|
|
logError('Main', `Error during cleanup: ${err.message}`);
|
|
}
|
|
// DNS server is closed last to ensure public DNS continues to respond during shutdown
|
|
process.exit(0);
|
|
};
|
|
state.gracefulShutdown = cleanup;
|
|
process.on('SIGINT', cleanup);
|
|
process.on('SIGTERM', cleanup);
|
|
process.on('SIGQUIT', cleanup);
|
|
if (process.platform !== 'win32') {
|
|
process.on('SIGUSR1', async () => {
|
|
try {
|
|
if (state.suspendSwarm) await state.suspendSwarm();
|
|
} catch (err) {
|
|
logWarn('Swarm', `Suspend signal failed: ${err.message}`);
|
|
}
|
|
});
|
|
process.on('SIGUSR2', async () => {
|
|
try {
|
|
if (state.resumeSwarm) await state.resumeSwarm();
|
|
} catch (err) {
|
|
logWarn('Swarm', `Resume signal failed: ${err.message}`);
|
|
}
|
|
});
|
|
}
|
|
// Handle uncaught exceptions
|
|
process.on('uncaughtException', (err) => {
|
|
const msg = err && err.message ? err.message : String(err);
|
|
if (/Atomic state must flush|SESSION_CLOSED/i.test(msg)) {
|
|
logWarn('Main', `Autopass/Hypercore busy (non-fatal): ${msg}`);
|
|
return;
|
|
}
|
|
logError('Main', `Uncaught exception: ${msg}`);
|
|
if (process.env.EXIT_ON_FATAL_ERROR === 'true' && state.gracefulShutdown) {
|
|
void state.gracefulShutdown();
|
|
}
|
|
});
|
|
// Handle unhandled promise rejections
|
|
process.on('unhandledRejection', (reason, promise) => {
|
|
const msg = reason && reason.message ? reason.message : String(reason);
|
|
if (/Atomic state must flush|SESSION_CLOSED/i.test(msg)) {
|
|
logWarn('Main', `Autopass/Hypercore busy (unhandled rejection, non-fatal): ${msg}`);
|
|
return;
|
|
}
|
|
logError('Main', `Unhandled rejection at: ${promise} reason: ${reason}`);
|
|
if (process.env.EXIT_ON_FATAL_ERROR === 'true' && state.gracefulShutdown) {
|
|
void state.gracefulShutdown();
|
|
}
|
|
});
|
|
} catch (err) {
|
|
logError('Main', `Main function error: ${err.message}`);
|
|
process.exit(1);
|
|
}
|
|
}
|
|
|
|
// Function to assign IPs to all domains
|
|
async function assignAllIPs() {
|
|
try {
|
|
logInfo('Main', 'Assigning IPs to resolved domains...');
|
|
let internalDomains = ['p2ns.admin'];
|
|
try {
|
|
const { getInternalDomains } = require('./includes/plugins/plugin-handler');
|
|
internalDomains = await getInternalDomains();
|
|
} catch (err) {
|
|
logWarn('Main', 'Plugin system not initialized, using default internal domains');
|
|
}
|
|
const domains = new Set(internalDomains);
|
|
// Fetch domains from Autopass entries if dnsPass is initialized
|
|
if (state.dnsPass) {
|
|
try {
|
|
const allEntries = await getAllEntries();
|
|
for (const entry of allEntries) {
|
|
if (entry.key.startsWith('claim:')) {
|
|
const parts = entry.key.split(':');
|
|
if (parts.length === 3) {
|
|
domains.add(parts[1]);
|
|
}
|
|
}
|
|
}
|
|
logDebug('Main', `Found ${domains.size} domains (including internal) from entries`);
|
|
} catch (err) {
|
|
logError('Main', `Error fetching entries for IP assignment: ${err.message}`);
|
|
}
|
|
} else {
|
|
logWarn('Main', 'dnsPass not initialized, only assigning IPs to internal domains');
|
|
}
|
|
|
|
// Import getConsensusState for checking resolution status
|
|
const { getConsensusState } = require('./includes/core/core');
|
|
|
|
// Assign IPs only to resolved domains
|
|
for (const domain of domains) {
|
|
if (domain === "p2ns.admin" || domain === "peer.directory") continue;
|
|
|
|
// Check if domain already has IP
|
|
if (state.domainToIPMap.has(domain)) {
|
|
logDebug('Main', `IP already assigned for ${domain}: ${state.domainToIPMap.get(domain)}`);
|
|
continue;
|
|
}
|
|
|
|
// Check consensus state - only assign if resolved
|
|
try {
|
|
const consensusState = await getConsensusState(domain);
|
|
|
|
if (consensusState.status === 'resolved') {
|
|
await createInterfaceForDomain(domain);
|
|
logInfo('Main', `Assigned IP to resolved domain: ${domain} (${state.domainToIPMap.get(domain)})`);
|
|
} else {
|
|
logDebug('Main', `Skipping ${domain} - not resolved yet (status: ${consensusState.status})`);
|
|
}
|
|
} catch (err) {
|
|
logError('Main', `Failed to check consensus or assign IP to ${domain}: ${err.message}`);
|
|
}
|
|
}
|
|
logInfo('Main', `IP assignment completed for ${domains.size} domains`);
|
|
} catch (err) {
|
|
logError('Main', `Error in assignAllIPs: ${err.message}`);
|
|
}
|
|
}
|
|
|
|
// Assign IPs to domains that have just become resolved (called after consensus updates)
|
|
async function assignIPsToResolvedDomains() {
|
|
try {
|
|
logDebug('Main', 'Checking for newly resolved domains...');
|
|
|
|
// Fetch domains from Autopass entries if dnsPass is initialized
|
|
if (!state.dnsPass) {
|
|
logDebug('Main', 'dnsPass not initialized, skipping IP assignment check');
|
|
return;
|
|
}
|
|
|
|
const { getAllEntries, getConsensusState } = require('./includes/core/core');
|
|
const allEntries = await getAllEntries();
|
|
const domains = new Set();
|
|
|
|
// Collect all domains with claims
|
|
for (const entry of allEntries) {
|
|
if (entry.key.startsWith('claim:')) {
|
|
const parts = entry.key.split(':');
|
|
if (parts.length === 3) {
|
|
domains.add(parts[1]);
|
|
}
|
|
}
|
|
}
|
|
|
|
let assignedCount = 0;
|
|
|
|
// Check each domain for resolution status
|
|
for (const domain of domains) {
|
|
// Skip internal domains and domains that already have IPs
|
|
if (domain === "p2ns.admin" || domain === "peer.directory" || state.domainToIPMap.has(domain)) {
|
|
continue;
|
|
}
|
|
|
|
try {
|
|
const consensusState = await getConsensusState(domain);
|
|
|
|
if (consensusState.status === 'resolved') {
|
|
await createInterfaceForDomain(domain);
|
|
logInfo('Main', `Assigned IP to newly resolved domain: ${domain} (${state.domainToIPMap.get(domain)})`);
|
|
assignedCount++;
|
|
}
|
|
} catch (err) {
|
|
logError('Main', `Failed to check consensus for ${domain}: ${err.message}`);
|
|
}
|
|
}
|
|
|
|
if (assignedCount > 0) {
|
|
logInfo('Main', `Assigned IPs to ${assignedCount} newly resolved domains`);
|
|
} else {
|
|
logDebug('Main', 'No newly resolved domains found');
|
|
}
|
|
} catch (err) {
|
|
logError('Main', `Error in assignIPsToResolvedDomains: ${err.message}`);
|
|
}
|
|
}
|
|
|
|
main();
|
|
// Usage:
|
|
// Genesis master: node p2ns.js --master --genesis
|
|
// Secondary master: node p2ns.js --master (empty storage; pairs via invite)
|
|
// Joiner: node p2ns.js
|
|
// Clean storage: node p2ns.js --clean (can be combined with --master [--genesis])
|
|
// Note: Both peers can add domains and will sync bidirectionally
|
|
// Use --clean flag to remove storage and start fresh for faster connections.
|