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

ix: prevent corestore corruption during graceful shutdown

Browse Source 
- Reorder shutdown sequence to close pairing operations before DNSPass
- Add proper tracking and cleanup of active Autopass.pair() operations
- Handle shared corestore conflicts between pairing ops and main DNSPass
- Ignore expected "store already closed" errors from pairing operations
- Consolidate all Autopass resource cleanup in PHASE 2.5

This fixes the "Corestore is closed" uncaught exceptions and corruption
that occurred when both pairing operations and DNSPass tried to close
the same shared corestore during shutdown.
This commit is contained in:
Raven Scott
2025-12-26 23:24:51 -05:00
parent e8f920a7aa
commit 62ac5a4b50
2 changed files with 251 additions and 48 deletions
Download Patch File Download Diff File Expand all files Collapse all files
includes/core/core.js
+6
View File
@@ -78,6 +78,12 @@ async function getAllEntries(pass = state.dnsPass, useCache = true) {
return entriesCache || [];
}
// Check if pass is ready - if not, wait briefly or return cached entries
if (!pass.ready) {
logDebug('Core', 'dnsPass is not ready yet, returning cached entries');
return entriesCache || [];
}
const entries = [];
try {
const stream = pass.list();
p2ns.js
+242 -45
View File
@@ -243,7 +243,7 @@ async function main() {
state.topic = topic; // Store topic in state for keep-alive and cleanup
logDebug('Main', `Generated topic: ${topic.toString('hex')}`);
// Initialize corestore and hyperswarm
const store = new Corestore(process.env.STORAGE_DIR || './my-storage');
let store = new Corestore(process.env.STORAGE_DIR || './my-storage');
logInfo('Main', `Corestore initialized with storage: ${process.env.STORAGE_DIR || 'my-storage'}`);
// Use the persistent keypair that was loaded earlier (already in state.keypair)
// Configure Hyperswarm for better local network peer discovery
@@ -324,6 +324,11 @@ async function main() {
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
@@ -357,6 +362,42 @@ async function main() {
}
}
// 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() {
if (!isMaster || !state.dnsPass) {
@@ -539,18 +580,46 @@ async function main() {
logWarn('Swarm', 'Ignoring invite as this is the master');
return;
}
// Check if we already have dnsPass
// Check if we already have dnsPass - if so, ignore invite regardless of processing state
if (getDnsPass()) {
logWarn('Swarm', 'Pass already initialized, ignoring invite');
return;
}
// Check if another invite is currently being processed
if (isProcessingInvite()) {
logWarn('Swarm', 'Another invite is currently being processed, skipping this one');
return;
}
// Acquire invite processing lock
await acquireInviteLock();
// Create a promise to track this invite processing
currentInvitePromise = (async () => {
try {
logDebug('Swarm', 'Processing received invite...');
// Follow Autopass pairing pattern: pair -> finished -> ready
const pair = Autopass.pair(store, invHex);
const newPass = await pair.finished();
currentPairOperation = pair; // Track for cleanup
// Add timeout to pairing operation (30 seconds should be enough for normal pairing)
// If it hangs, it likely indicates corrupted storage
logDebug('Swarm', 'Starting Autopass pairing (timeout: 30s)...');
const newPass = await withTimeout(
pair.finished(),
30000,
'Autopass pairing'
);
logDebug('Swarm', 'Autopass pair finished');
await newPass.ready();
logDebug('Swarm', 'Waiting for Autopass to be ready (timeout: 10s)...');
await withTimeout(
newPass.ready(),
10000,
'Autopass ready'
);
logDebug('Swarm', 'Paired Autopass ready');
// Use helper to ensure consistent state
setDnsPass(newPass);
@@ -558,6 +627,15 @@ async function main() {
// Sync initial data
await core.update();
logDebug('Swarm', 'Synced initial data from master');
// Ensure dnsPass is fully ready before proceeding with dependent operations
const currentDnsPass = getDnsPass();
if (currentDnsPass && !currentDnsPass.ready) {
logDebug('Swarm', 'Waiting for dnsPass to be fully ready...');
await currentDnsPass.ready();
logDebug('Swarm', 'dnsPass is now fully ready');
}
// Setup domains watcher for joiner
setupDomainsWatcher();
await listDomains();
@@ -599,17 +677,32 @@ async function main() {
} catch (ackErr) {
logWarn('Swarm', `Failed to send invite_ack: ${ackErr.message}`);
}
// Release invite processing lock on success
currentPairOperation = null; // Clear pair reference
releaseInviteLock();
} catch (err) {
logError('Swarm', `Error processing received invite: ${err.message}`);
state.consecutiveInviteFailures++;
currentPairOperation = null; // Clear pair reference on error
// Check if this is a timeout error (indicates hanging/corrupted storage)
if (err.message.includes('timed out')) {
logError('Swarm', 'Invite processing timed out - this indicates corrupted storage.');
logError('Swarm', 'The Autopass pairing operation hung, which means the corestore is corrupted.');
logError('Swarm', 'CRITICAL: Corestore corruption detected. You must manually restart with: sudo node p2ns.js --clean');
logError('Swarm', 'The system cannot automatically recover from this state.');
}
// If invite processing fails, it might be due to corrupted storage
// Check if the error suggests storage corruption and provide helpful guidance
if (err.message.includes('corrupt') || err.message.includes('inconsistent') ||
err.message.includes('feed') || err.message.includes('signature') ||
err.message.includes('timed out') ||
err.code === 'CORRUPTION' || err.code === 'INCONSISTENT') {
if (!err.message.includes('timed out')) {
logError('Swarm', 'Invite processing failed due to possible storage corruption. Try using --clean flag to reset storage.');
}
}
// If we've had multiple consecutive failures, strongly suggest storage cleanup
if (state.consecutiveInviteFailures >= 3) {
@@ -628,7 +721,14 @@ async function main() {
}
}
}
// Release invite processing lock on error
releaseInviteLock();
}
})(); // End of currentInvitePromise
// Await the invite processing promise to ensure it actually executes
await currentInvitePromise;
},
onOpen: (peerId) => {
logDebug('Swarm', `Invite channel opened for peer ${peerId}`);
@@ -811,19 +911,46 @@ async function main() {
// This invite is for us! Process it directly (same logic as invite channel handler)
logInfo('Swarm', 'Relay invite is for us, processing...');
// Check if we already have dnsPass
// Check if we already have dnsPass - if so, ignore relay invite regardless of processing state
if (getDnsPass()) {
logWarn('Swarm', 'Pass already initialized, ignoring relay invite');
return;
}
// Check if another invite is currently being processed
if (isProcessingInvite()) {
logWarn('Swarm', 'Another invite is currently being processed, skipping relay invite');
return;
}
// Acquire invite processing lock
await acquireInviteLock();
// Create a promise to track this relay invite processing
currentInvitePromise = (async () => {
try {
logDebug('Swarm', 'Processing received relay invite...');
// Follow Autopass pairing pattern: pair -> finished -> ready
const pair = Autopass.pair(store, inviteHex);
const newPass = await pair.finished();
currentPairOperation = pair; // Track for cleanup
// Add timeout to pairing operation (30 seconds should be enough for normal pairing)
// If it hangs, it likely indicates corrupted storage
logDebug('Swarm', 'Starting Autopass pairing for relay invite (timeout: 30s)...');
const newPass = await withTimeout(
pair.finished(),
30000,
'Autopass pairing (relay)'
);
logDebug('Swarm', 'Autopass pair finished');
await newPass.ready();
logDebug('Swarm', 'Waiting for Autopass to be ready (timeout: 10s)...');
await withTimeout(
newPass.ready(),
10000,
'Autopass ready (relay)'
);
logDebug('Swarm', 'Paired Autopass ready');
// Use helper to ensure consistent state
setDnsPass(newPass);
@@ -831,6 +958,15 @@ async function main() {
// Sync initial data
await core.update();
logDebug('Swarm', 'Synced initial data from master');
// Ensure dnsPass is fully ready before proceeding with dependent operations
const currentDnsPass = getDnsPass();
if (currentDnsPass && !currentDnsPass.ready) {
logDebug('Swarm', 'Waiting for dnsPass to be fully ready...');
await currentDnsPass.ready();
logDebug('Swarm', 'dnsPass is now fully ready');
}
// Setup domains watcher for joiner
setupDomainsWatcher();
await listDomains();
@@ -857,16 +993,32 @@ async function main() {
logWarn('Swarm', `Failed to send invite_ack to relay peer: ${ackErr.message}`);
}
logInfo('Swarm', 'Successfully processed relay invite and initialized dnsPass');
// Release invite processing lock on success
currentPairOperation = null; // Clear pair reference
releaseInviteLock();
} catch (err) {
logError('Swarm', `Error processing received relay invite: ${err.message}`);
state.consecutiveInviteFailures++;
currentPairOperation = null; // Clear pair reference on error
// Check if this is a timeout error (indicates hanging/corrupted storage)
if (err.message.includes('timed out')) {
logError('Swarm', 'Relay invite processing timed out - this indicates corrupted storage.');
logError('Swarm', 'The Autopass pairing operation hung, which means the corestore is corrupted.');
logError('Swarm', 'CRITICAL: Corestore corruption detected. You must manually restart with: sudo node p2ns.js --clean');
logError('Swarm', 'The system cannot automatically recover from this state.');
}
// If relay invite processing fails, it might be due to corrupted storage
// Check if the error suggests storage corruption and provide helpful guidance
if (err.message.includes('corrupt') || err.message.includes('inconsistent') ||
err.message.includes('feed') || err.message.includes('signature') ||
err.message.includes('timed out') ||
err.code === 'CORRUPTION' || err.code === 'INCONSISTENT') {
if (!err.message.includes('timed out')) {
logError('Swarm', 'Relay invite processing failed due to possible storage corruption. Try using --clean flag to reset storage.');
}
}
// If we've had multiple consecutive failures, strongly suggest storage cleanup
if (state.consecutiveInviteFailures >= 3) {
@@ -882,7 +1034,14 @@ async function main() {
}
}
}
// Release invite processing lock on error
releaseInviteLock();
}
})(); // End of currentInvitePromise
// Await the relay invite processing promise to ensure it actually executes
await currentInvitePromise;
} else {
// Forward the response toward the origin
// Check if we're directly connected to the origin
@@ -2565,7 +2724,65 @@ async function main() {
logDebug('Main', 'Removed event listeners from dnsPass and core');
// =======================================================================
// PHASE 3: Stop swarm (swarm handlers might trigger dnsPass access)
// 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 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
@@ -2822,44 +3039,30 @@ async function main() {
}
// =======================================================================
// PHASE 8: Close dnsPass and main corestore (after all plugins closed)
// PHASE 8: Close invite processing
// =======================================================================
if (state.dnsPass && typeof state.dnsPass.close === 'function') {
logDebug('Main', 'Closing dnsPass (includes closing its internal Autobase core)...');
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');
// 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();
}
// 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...');
// Wait for any current invite processing to complete
if (currentInvitePromise) {
try {
await state.core.close();
logDebug('Main', 'dnsPass base core closed explicitly');
logDebug('Main', 'Waiting for current invite processing to complete...');
await withTimeout(currentInvitePromise, 5000, 'Current invite processing shutdown');
logDebug('Main', 'Current invite processing completed during shutdown');
} 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}`);
}
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 {
@@ -2892,7 +3095,7 @@ async function main() {
}
// =======================================================================
// PHASE 10: Close protomux channels (may have handlers accessing dnsPass)
// PHASE 9: Close protomux channels (dnsPass is already closed)
// =======================================================================
logDebug('Main', 'Closing protomux channels...');
@@ -2917,13 +3120,7 @@ async function main() {
logDebug('Main', 'Protomux channels closed');
// =======================================================================
// PHASE 9: Continue with remaining cleanup
// =======================================================================
// dnsPass and main corestore are now closed after all plugins and replication managers
// =======================================================================
// PHASE 11: Close Holesail servers and clients
// PHASE 10: Close Holesail servers and clients
// =======================================================================
// Close Holesail servers and clients (in parallel)