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

Fix Atomic state

Browse Source
This commit is contained in:
Raven Scott
2026-05-27 19:44:05 -04:00
parent 3ece7abe9f
commit c8334a8d63
5 changed files with 83 additions and 33 deletions
Download Patch File Download Diff File Expand all files Collapse all files
includes/admin/admin-backend/routes/domains.js
+2 -1
View File
@@ -280,7 +280,8 @@ async function handleDomainsRoutes(req, res) {
let errorCount = 0;
for (const entry of entriesToRemove) {
try {
await pass.remove(entry.key);
const { dnsPassRemove } = require('../../../core/dns-pass-queue');
await dnsPassRemove(pass, entry.key);
removedCount++;
} catch (err) {
logError('Admin', `Error removing entry ${entry.key}: ${err.message}`);
includes/core/core.js
+13 -11
View File
@@ -4,6 +4,7 @@ const { logDebug, logError, logInfo, logWarn } = require('../infrastructure/logg
const { trackDomainEvent, trackConsensusEvent } = require('../maintenance/metrics');
const { validateConfig } = require('../infrastructure/config');
const { secondsToMs, getPersistentPublicKey } = require('../infrastructure/utils');
const { dnsPassAdd, dnsPassRemove, dnsPassGet, whenDnsPassIdle } = require('./dns-pass-queue');
// Get consensus configuration
let consensusConfig = null;
@@ -94,6 +95,7 @@ async function getAllEntries(pass = state.dnsPass, useCache = true) {
const entries = [];
try {
await whenDnsPassIdle();
const stream = pass.list();
for await (const entry of stream) {
entries.push({
@@ -107,8 +109,8 @@ async function getAllEntries(pass = state.dnsPass, useCache = true) {
logDebug('Core', `Cached ${entries.length} entries`);
} catch (err) {
const msg = err.message || '';
if (msg.includes('SESSION_CLOSED') || msg.includes('closing core')) {
logWarn('Core', `dnsPass core is closing — stop the node and restart with --clean if this persists after upgrading Autopass: ${msg}`);
if (msg.includes('SESSION_CLOSED') || msg.includes('closing core') || msg.includes('Atomic state must flush')) {
logWarn('Core', `dnsPass core busy or closing — restart with --clean if this persists: ${msg}`);
} else {
logError('Core', `Error getting entries: ${msg}`);
}
@@ -216,7 +218,7 @@ async function updateClaimClients(domain, claimant, clients) {
// SAFE: Remove old claim and add updated one (ownership verified)
await safeRemoveClaim(claimKey, claimant);
await pass.add(claimKey, updatedValue);
await dnsPassAdd(pass, claimKey, updatedValue);
invalidateEntriesCache();
logInfo('Core', `Updated claim clients for ${domain} by ${claimant}`);
@@ -674,7 +676,7 @@ async function voteForDomain(domain, claimant, allEntries = null) {
// Remove existing votes for other claimants
for (const [votedClaimant, voteKey] of myVotes) {
if (votedClaimant !== claimant) {
await pass.remove(voteKey);
await dnsPassRemove(pass, voteKey);
logInfo('Core', `Removed vote for ${domain} claimant ${votedClaimant}`);
}
}
@@ -682,7 +684,7 @@ async function voteForDomain(domain, claimant, allEntries = null) {
// Add new vote if not already voting for this claimant
if (!myVotes.has(claimant)) {
const voteKey = `vote:${domain}:${claimant}:${localWriter}`;
await pass.add(voteKey, claims[claimant]);
await dnsPassAdd(pass, voteKey, claims[claimant]);
logInfo('Core', `Voted for ${domain} claimant ${claimant}`);
invalidateEntriesCache();
}
@@ -735,7 +737,7 @@ async function removeDomain(domain) {
for (const entry of allEntries) {
if (entry.key.startsWith(`claim:${domain}:`) ||
entry.key.startsWith(`vote:${domain}:`)) {
await pass.remove(entry.key);
await dnsPassRemove(pass, entry.key);
logDebug('Core', `Removed ${entry.key} for domain removal`);
}
}
@@ -769,7 +771,7 @@ async function removeOwnClaimAndVotes(domain, localWriter) {
// Remove ONLY the user's own claim record
// Format: claim:domain:claimant
if (entry.key === `claim:${domain}:${localWriter}`) {
await pass.remove(entry.key);
await dnsPassRemove(pass, entry.key);
logInfo('Core', `Removed own claim: ${entry.key}`);
removedClaims++;
}
@@ -780,7 +782,7 @@ async function removeOwnClaimAndVotes(domain, localWriter) {
const parts = entry.key.split(':');
if (parts.length === 4 && parts[3] === localWriter) {
// This is a vote cast BY the user (voter is localWriter)
await pass.remove(entry.key);
await dnsPassRemove(pass, entry.key);
logInfo('Core', `Removed vote cast by user: ${entry.key}`);
removedVotes++;
}
@@ -825,7 +827,7 @@ async function safeRemoveClaim(claimKey, expectedClaimant) {
}
// Safe to remove - ownership verified
await state.dnsPass.remove(claimKey);
await dnsPassRemove(state.dnsPass, claimKey);
logInfo('Core', `Safely removed claim ${claimKey} (owned by ${claimant})`);
return true;
}
@@ -847,7 +849,7 @@ async function removeAllRecords() {
for (const entry of allEntries) {
if (entry.key.startsWith('claim:') || entry.key.startsWith('vote:')) {
try {
await pass.remove(entry.key);
await dnsPassRemove(pass, entry.key);
removed++;
logDebug('Core', `Removed ${entry.key}`);
} catch (err) {
@@ -899,7 +901,7 @@ async function cleanupRedundantClientEntries() {
for (const entry of allEntries) {
if (entry.key.startsWith('clients:')) {
try {
await pass.remove(entry.key);
await dnsPassRemove(pass, entry.key);
removed++;
logDebug('Core', `Removed redundant client entry: ${entry.key}`);
} catch (err) {
includes/core/dns-pass-queue.js
+47
View File
@@ -0,0 +1,47 @@
/**
* Serializes Autopass / Autobase operations on the master corestore.
* Concurrent createInvite, list(), add(), and remove() calls cause
* "Atomic state must flush to parent" on Hypercore 11.
*/
let chain = Promise.resolve();
function enqueueDnsPass(operation) {
const run = chain.then(() => operation());
chain = run.then(
() => {},
() => {}
);
return run;
}
/** Wait until queued dnsPass work finishes (for reads). */
function whenDnsPassIdle() {
return chain;
}
async function createInvite(pass, opts) {
return enqueueDnsPass(() => pass.createInvite(opts));
}
async function dnsPassAdd(pass, key, value, file) {
return enqueueDnsPass(() => pass.add(key, value, file));
}
async function dnsPassRemove(pass, key) {
return enqueueDnsPass(() => pass.remove(key));
}
async function dnsPassGet(pass, key) {
await whenDnsPassIdle();
return pass.get(key);
}
module.exports = {
enqueueDnsPass,
whenDnsPassIdle,
createInvite,
dnsPassAdd,
dnsPassRemove,
dnsPassGet
};
includes/core/domains.js
+4 -3
View File
@@ -3,6 +3,7 @@ const { logDebug, logInfo, logError } = require('../infrastructure/logger');
const { getAllEntries, autoVoteForDomain, invalidateEntriesCache, parseClaimValue, safeRemoveClaim } = require('./core');
const { trackDomainEvent } = require('../maintenance/metrics');
const { getPersistentPublicKey } = require('../infrastructure/utils');
const { dnsPassAdd, dnsPassGet } = require('./dns-pass-queue');
// Initialize reserved IPs set if not already present
if (!state.reservedIPs) {
state.reservedIPs = new Set(['127.0.0.1']);
@@ -65,15 +66,15 @@ async function addDomain(domain, hash, ssl = false) {
// SAFE: Remove old claim and add updated one (ownership verified above)
// Use the safe remove helper for consistent validation
await safeRemoveClaim(claimKey, claimant);
await pass.add(claimKey, claimValue);
await dnsPassAdd(pass, claimKey, claimValue);
logInfo('Domains', `Updated claim ${claimKey} with new hash ${hash} and SSL=${sslValue}`);
} else {
// New claim - add it
logDebug('Domains', `Adding new domain ${domain} with hash ${hash} and SSL=${sslValue} as claim ${claimKey} at timestamp ${timestamp}`);
await pass.add(claimKey, claimValue);
await dnsPassAdd(pass, claimKey, claimValue);
const verified = await pass.get(claimKey);
const verified = await dnsPassGet(pass, claimKey);
logDebug('Domains', `Verified add for ${claimKey}: ${verified ? verified.toString('utf8') : 'null'}`);
logInfo('Domains', `Domain ${domain} added as claim ${claimKey} with timestamp ${timestamp}`);
}
p2ns.js
+17 -18
View File
@@ -425,7 +425,8 @@ async function main() {
logInfo('Swarm', `Processing pending invite request from ${peerId} (queued for ${Math.round(age/1000)}s)`);
try {
const inv = await state.dnsPass.createInvite();
const { createInvite } = require('./includes/core/dns-pass-queue');
const inv = await createInvite(state.dnsPass);
logInfo('Swarm', `Created invite for pending request from ${peerId}: ${inv.toString('hex').substring(0, 20)}...`);
const sent = channelManager.sendToPeer(CORE_DOMAIN, 'invite', peerId, inv.toString('hex'));
if (sent) {
@@ -837,7 +838,8 @@ async function main() {
}
logDebug('Swarm', `Creating invite for peer ${peerId}...`);
// Use Autopass API: pass.createInvite()
const inv = await pass.createInvite();
const { createInvite } = require('./includes/core/dns-pass-queue');
const inv = await createInvite(pass);
logInfo('Swarm', `Created invite for requesting peer ${peerId}: ${inv.toString('hex').substring(0, 20)}...`);
try {
channelManager.sendToPeer(CORE_DOMAIN, 'invite', peerId, inv.toString('hex'));
@@ -852,7 +854,12 @@ async function main() {
}
}
} catch (err) {
logError('Swarm', `Error creating invite for request from ${peerId}: ${err.message}`);
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}`);
} else {
logError('Swarm', `Error creating invite for request from ${peerId}: ${msg}`);
}
// Always send error response so peer knows what happened
try {
channelManager.sendToPeer(CORE_DOMAIN, 'request', peerId, 'invite_unavailable:creation_failed');
@@ -879,7 +886,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 inv = await pass.createInvite();
const { createInvite } = require('./includes/core/dns-pass-queue');
const inv = await createInvite(pass);
logInfo('Swarm', `Created relay invite for origin ${originPeerId.substring(0, 16)}...`);
// Send relay response back to the peer that sent us the request
channelManager.sendToPeer(CORE_DOMAIN, 'request', peerId, `relay_invite_response:${originPeerId}:${inv.toString('hex')}`);
@@ -1217,7 +1225,8 @@ async function main() {
pendingInviteAcks.delete(peerId);
return;
}
const inv = await pass.createInvite();
const { createInvite } = require('./includes/core/dns-pass-queue');
const inv = await createInvite(pass);
logInfo('Swarm', `Created proactive invite for peer ${peerId} (attempt ${retryCount}/${maxRetries}): ${inv.toString('hex').substring(0, 20)}...`);
const success = channelManager.sendToPeer(CORE_DOMAIN, 'invite', peerId, inv.toString('hex'));
@@ -1247,8 +1256,8 @@ async function main() {
}
} catch (err) {
const msg = err.message || '';
if (msg.includes('SESSION_CLOSED') || msg.includes('closing core')) {
logWarn('Swarm', `dnsPass core closing, cannot send invite to ${peerId}. Restart master with --clean if this continues.`);
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;
}
@@ -1636,16 +1645,6 @@ async function main() {
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');
} else {
try {
const probe = newPass.list();
const first = await probe.next();
if (first && typeof probe.return === 'function') await probe.return();
logDebug('Main', 'Autopass list probe succeeded');
} catch (probeErr) {
logError('Main', `Autopass storage probe failed: ${probeErr.message}`);
logError('Main', 'Restart with --clean after upgrading Autopass: node p2ns.js --clean --master');
}
}
logInfo('Main', `Core retrieved. Writable: ${core.writable}, Key: ${core.key.toString('hex')}`);
@@ -2020,7 +2019,7 @@ async function main() {
}
// Send proactive invite after waiting for bidirectional channel
const proactiveInviteDelay = parseInt(process.env.MASTER_PROACTIVE_INVITE_DELAY || '500', 10);
const proactiveInviteDelay = parseInt(process.env.MASTER_PROACTIVE_INVITE_DELAY || '1500', 10);
setTimeout(async () => {
// Double-check connection is still stable
if (!isConnectionStable(peerId, conn)) {