snxraven/p2ns
2
1
Fork 1
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Updates

Browse Source
This commit is contained in:
Raven Scott
2026-05-27 20:03:25 -04:00
parent dbc6997219
commit 3d35f148c5
5 changed files with 178 additions and 32 deletions
Download Patch File Download Diff File Expand all files Collapse all files
default.env
+4 -2
View File
@@ -111,7 +111,9 @@ TOPIC_SEED=p2ns-dns
MASTER_RECONNECT_INTERVAL=5
# Maximum number of reconnection attempts per peer before giving up (default: 10)
MASTER_MAX_RECONNECT_ATTEMPTS=10
# Delay in milliseconds before sending proactive invite to new peers (default: 500)
# Delay in milliseconds before sending proactive invite to new peers (default: 4000)
# Master nodes automatically send invites to peers when they connect
MASTER_PROACTIVE_INVITE_DELAY=500
MASTER_PROACTIVE_INVITE_DELAY=4000
# Gap between serialized Autopass ops in ms (default: 75) — reduces atomic flush races
# DNS_PASS_OP_GAP_MS=75
includes/core/core.js
+2 -1
View File
@@ -63,7 +63,8 @@ function isDnsPassUsable(pass) {
if (pass.opened === false) return false;
const base = pass.base;
if (!base) return false;
if (base.closed || base.closing) return false;
if (base.closed) return false;
// base.closing is transient during replication; only treat as unusable when fully closed
return true;
}
includes/core/dns-pass-queue.js
+103 -18
View File
@@ -1,13 +1,86 @@
/**
* Serializes all Autopass / Autobase operations on the master corestore.
* Concurrent list(), createInvite(), add(), and remove() cause
* "Atomic state must flush to parent" on Hypercore 11.
* Concurrent list(), createInvite(), add(), and remove() — plus HyperDB autoUpdate
* during replication — cause "Atomic state must flush to parent" on Hypercore 11.
*/
const state = require('../infrastructure/state');
const { logDebug, logWarn } = require('../infrastructure/logger');
let chain = Promise.resolve();
const ATOMIC_ERR_RE = /Atomic state must flush|SESSION_CLOSED|closing core/i;
const OP_GAP_MS = parseInt(process.env.DNS_PASS_OP_GAP_MS || '75', 10);
const SYNC_RETRIES = parseInt(process.env.DNS_PASS_SYNC_RETRIES || '4', 10);
function isAtomicDnsPassError(err) {
const msg = err && err.message ? err.message : String(err);
return ATOMIC_ERR_RE.test(msg);
}
function sleep(ms) {
return new Promise((resolve) => setTimeout(resolve, ms));
}
/**
* Bring Autobase + HyperDB view in sync before a read/write (reduces atomic flush races).
* @param {object} pass - Autopass instance
* @returns {Promise<boolean>} false if sync could not complete (busy)
*/
async function syncDnsPassView(pass) {
if (!pass || pass.closed) return false;
try {
await pass.ready();
const base = pass.base;
if (!base || base.closed || base.closing) return false;
await base.update();
const view = base.view;
if (view && !view.closed && typeof view.update === 'function') {
view.update();
}
return true;
} catch (err) {
if (isAtomicDnsPassError(err)) {
logDebug('DnsPassQueue', `View sync deferred (core busy): ${err.message}`);
return false;
}
throw err;
}
}
async function runWithDnsPassRetry(pass, operation) {
let lastErr;
for (let attempt = 0; attempt < SYNC_RETRIES; attempt++) {
if (attempt > 0) {
await sleep(100 * attempt);
}
await syncDnsPassView(pass);
try {
return await operation();
} catch (err) {
lastErr = err;
if (!isAtomicDnsPassError(err) || attempt === SYNC_RETRIES - 1) {
throw err;
}
logWarn('DnsPassQueue', `dnsPass op retry ${attempt + 1}/${SYNC_RETRIES}: ${err.message}`);
}
}
throw lastErr;
}
function enqueueDnsPass(operation) {
const run = chain.then(() => Promise.resolve().then(operation));
const run = chain.then(async () => {
state.dnsPassWriteInProgress = (state.dnsPassWriteInProgress || 0) + 1;
try {
const result = await Promise.resolve().then(operation);
if (OP_GAP_MS > 0) {
await sleep(OP_GAP_MS);
}
return result;
} finally {
state.dnsPassWriteInProgress = Math.max(0, (state.dnsPassWriteInProgress || 1) - 1);
}
});
chain = run.then(
() => {},
() => {}
@@ -21,7 +94,9 @@ function whenDnsPassIdle() {
}
async function createInvite(pass, opts) {
return enqueueDnsPass(() => pass.createInvite(opts));
return enqueueDnsPass(() =>
runWithDnsPassRetry(pass, () => pass.createInvite(opts))
);
}
/**
@@ -41,15 +116,21 @@ function invitePreview(inv) {
}
async function dnsPassAdd(pass, key, value, file) {
return enqueueDnsPass(() => pass.add(key, value, file));
return enqueueDnsPass(() =>
runWithDnsPassRetry(pass, () => pass.add(key, value, file))
);
}
async function dnsPassRemove(pass, key) {
return enqueueDnsPass(() => pass.remove(key));
return enqueueDnsPass(() =>
runWithDnsPassRetry(pass, () => pass.remove(key))
);
}
async function dnsPassGet(pass, key) {
return enqueueDnsPass(() => pass.get(key));
return enqueueDnsPass(() =>
runWithDnsPassRetry(pass, () => pass.get(key))
);
}
/**
@@ -58,22 +139,26 @@ async function dnsPassGet(pass, key) {
* @returns {Promise<Array<{key: string, value: string}>>}
*/
async function listAllEntries(pass) {
return enqueueDnsPass(async () => {
const entries = [];
const stream = pass.list();
for await (const entry of stream) {
entries.push({
key: entry.key.toString('utf8'),
value: entry.value.toString('utf8')
});
}
return entries;
});
return enqueueDnsPass(async () =>
runWithDnsPassRetry(pass, async () => {
const entries = [];
const stream = pass.list();
for await (const entry of stream) {
entries.push({
key: entry.key.toString('utf8'),
value: entry.value.toString('utf8')
});
}
return entries;
})
);
}
module.exports = {
enqueueDnsPass,
whenDnsPassIdle,
syncDnsPassView,
isAtomicDnsPassError,
createInvite,
inviteToWire,
invitePreview,
includes/infrastructure/state.js
+2
View File
@@ -42,6 +42,8 @@ const currentSubnetIndex = 0; // Start with first subnet for round-robin
module.exports = {
dnsPass: null,
/** Incremented while dns-pass-queue runs a serialized Autopass op (defer update handlers) */
dnsPassWriteInProgress: 0,
holesails: new Map(),
starting: new Map(),
tlsServers: new Map(),
p2ns.js
+67 -11
View File
@@ -410,9 +410,17 @@ async function main() {
// Function to process pending invite requests when master becomes ready
async function processPendingInviteRequests() {
const { isDnsPassUsable } = require('./includes/core/core');
if (!isMaster || !isDnsPassUsable(state.dnsPass)) {
const { whenDnsPassIdle, syncDnsPassView } = require('./includes/core/dns-pass-queue');
if (!isMaster || !state.dnsPass) {
return;
}
if (!isDnsPassUsable(state.dnsPass)) {
await syncDnsPassView(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
@@ -838,11 +846,16 @@ async function main() {
return;
}
const { isDnsPassUsable } = require('./includes/core/core');
const { whenDnsPassIdle, syncDnsPassView } = require('./includes/core/dns-pass-queue');
if (!isDnsPassUsable(pass)) {
await syncDnsPassView(pass);
}
if (!isDnsPassUsable(pass)) {
logWarn('Swarm', `dnsPass not usable, cannot create invite for ${peerId}`);
channelManager.sendToPeer(CORE_DOMAIN, 'request', peerId, 'invite_unavailable:creation_failed');
return;
}
await whenDnsPassIdle();
logDebug('Swarm', `Creating invite for peer ${peerId}...`);
const { createInvite, inviteToWire, invitePreview } = require('./includes/core/dns-pass-queue');
const inviteChannelInfo = channelManager.getChannelInfo(CORE_DOMAIN, 'invite');
@@ -873,14 +886,28 @@ async function main() {
} catch (err) {
const msg = err.message || '';
if (msg.includes('Atomic state must flush') || msg.includes('SESSION_CLOSED')) {
logWarn('Swarm', `dnsPass busy, invite creation failed for ${peerId}: ${msg}`);
logWarn('Swarm', `dnsPass busy, scheduling invite retry for ${peerId}: ${msg}`);
setTimeout(async () => {
const retryPass = getDnsPass();
const { isDnsPassUsable } = require('./includes/core/core');
if (!connectedPeers.has(peerId) || !retryPass || !isDnsPassUsable(retryPass)) return;
try {
const { createInvite, inviteToWire, whenDnsPassIdle } = require('./includes/core/dns-pass-queue');
await whenDnsPassIdle();
const inv = await createInvite(retryPass);
channelManager.sendToPeer(CORE_DOMAIN, 'invite', peerId, inviteToWire(inv));
logInfo('Swarm', `Sent invite to ${peerId} after dnsPass busy retry`);
} catch (retryErr) {
logWarn('Swarm', `Invite retry failed for ${peerId}: ${retryErr.message}`);
}
}, 2000);
} else {
logError('Swarm', `Error creating invite for request from ${peerId}: ${msg}`);
}
try {
channelManager.sendToPeer(CORE_DOMAIN, 'request', peerId, 'invite_unavailable:creation_failed');
} catch (sendErr) {
// Channel might be closed, ignore
try {
channelManager.sendToPeer(CORE_DOMAIN, 'request', peerId, 'invite_unavailable:creation_failed');
} catch (sendErr) {
// Channel might be closed, ignore
}
}
} finally {
pendingInviteRequestHandlers.delete(peerId);
@@ -904,7 +931,8 @@ async function main() {
if (pass && isDnsPassUsable(pass) && (isMaster || process.env.ALLOW_ANY_WRITER_INVITES === 'true')) {
// We can create an invite - send it back through the relay chain
try {
const { createInvite, inviteToWire } = require('./includes/core/dns-pass-queue');
const { createInvite, inviteToWire, whenDnsPassIdle } = require('./includes/core/dns-pass-queue');
await whenDnsPassIdle();
const inv = await createInvite(pass);
const invWire = inviteToWire(inv);
logInfo('Swarm', `Created relay invite for origin ${originPeerId.substring(0, 16)}...`);
@@ -1239,11 +1267,16 @@ async function main() {
retryCount++;
try {
const { isDnsPassUsable } = require('./includes/core/core');
const { whenDnsPassIdle, syncDnsPassView } = require('./includes/core/dns-pass-queue');
if (!isDnsPassUsable(pass)) {
await syncDnsPassView(pass);
}
if (!isDnsPassUsable(pass)) {
logWarn('Swarm', `dnsPass not usable, skipping proactive invite for ${peerId}`);
pendingInviteAcks.delete(peerId);
return;
}
await whenDnsPassIdle();
const { createInvite, inviteToWire, invitePreview } = require('./includes/core/dns-pass-queue');
const inv = await createInvite(pass);
const invWire = inviteToWire(inv);
@@ -1518,6 +1551,19 @@ async function main() {
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
@@ -1909,7 +1955,7 @@ async function main() {
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 || '3000', 10)
? parseInt(process.env.MASTER_AUTO_VOTE_DELAY || '5000', 10)
: 0;
setTimeout(() => {
if (state.isShuttingDown || !state.dnsPass) return;
@@ -2052,7 +2098,7 @@ async function main() {
}
// Send proactive invite after waiting for bidirectional channel
const proactiveInviteDelay = parseInt(process.env.MASTER_PROACTIVE_INVITE_DELAY || '1500', 10);
const proactiveInviteDelay = parseInt(process.env.MASTER_PROACTIVE_INVITE_DELAY || '4000', 10);
setTimeout(async () => {
// Double-check connection is still stable
if (!isConnectionStable(peerId, conn)) {
@@ -3356,11 +3402,21 @@ async function main() {
process.on('SIGQUIT', cleanup);
// Handle uncaught exceptions
process.on('uncaughtException', (err) => {
logError('Main', `Uncaught exception: ${err.message}`);
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}`);
// Do not call cleanup() to prevent automatic shutdown
});
// 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}`);
// Do not call cleanup() to prevent automatic shutdown
});