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

dnspass fixes

Browse Source
This commit is contained in:
Raven Scott
2026-05-28 01:20:52 -04:00
parent d95baaed1c
commit 9057b9661d
4 changed files with 129 additions and 52 deletions
Download Patch File Download Diff File Expand all files Collapse all files
includes/plugins/db-version-migration.js
+65 -12
View File
@@ -13,6 +13,32 @@ const { getVersionedCore } = require('./db-core-keys');
const MANIFEST_FILE = '.p2ns-db-version.json';
const VERSION_HISTORY_FILE = '.p2ns-version-history.json';
/** Plugins whose migration finished (or was skipped) this process — avoids duplicate Corestore opens. */
const completedMigrations = new Set();
/**
* @param {string} pluginDomain
* @param {string} pluginDir
* @returns {string}
*/
function migrationKey(pluginDomain, pluginDir) {
return `${pluginDomain}\0${pluginDir}`;
}
/**
* Yield the event loop after closing a migration Corestore so db-manager can reopen the path.
* @param {import('corestore')} store
* @returns {Promise<void>}
*/
async function closeMigrationStore(store) {
try {
await store.close();
} catch (err) {
logDebug('DBVersionMigration', `Error closing migration store: ${err.message}`);
}
await new Promise((resolve) => setImmediate(resolve));
}
/**
* @param {string} pluginDir
* @returns {string}
@@ -328,10 +354,26 @@ async function migrateBetweenVersions(pluginDomain, pluginDir, dbDir, fromVersio
logDebug('DBVersionMigration', `Error closing target db: ${err.message}`);
}
}
await closeMigrationStore(store);
}
}
/**
* Run pending HyperDB version migrations for loaded plugins before opening live databases.
* @param {Array<{ domain: string, pluginDir: string, dbConfig?: { dbDir: string }|null }>} plugins
* @returns {Promise<void>}
*/
async function runPendingPluginDatabaseMigrations(plugins) {
for (const plugin of plugins) {
if (!plugin?.dbConfig?.dbDir) continue;
try {
await store.close();
await ensurePluginDatabaseVersionMigration(plugin.domain, plugin.pluginDir, plugin.dbConfig.dbDir);
} catch (err) {
logDebug('DBVersionMigration', `Error closing migration store: ${err.message}`);
logError(
'DBVersionMigration',
`Pre-init migration failed for ${plugin.domain}: ${err.message} — database init may retry migration`
);
completedMigrations.delete(migrationKey(plugin.domain, plugin.pluginDir));
}
}
}
@@ -429,11 +471,7 @@ async function versionCoreHasRecordsAt(pluginDir, pluginDomain, version, dbDir,
try {
return await versionCoreHasRecords(store, pluginDomain, version, dbDir, collectionIds);
} finally {
try {
await store.close();
} catch (err) {
logDebug('DBVersionMigration', `Error closing version probe store: ${err.message}`);
}
await closeMigrationStore(store);
}
}
@@ -486,11 +524,7 @@ async function discoverLegacySourceVersion(pluginDomain, pluginDir, dbDir, targe
}
}
} finally {
try {
await store.close();
} catch (err) {
logDebug('DBVersionMigration', `Error closing discovery store: ${err.message}`);
}
await closeMigrationStore(store);
}
return null;
@@ -504,6 +538,15 @@ async function discoverLegacySourceVersion(pluginDomain, pluginDir, dbDir, targe
* @returns {Promise<void>}
*/
async function ensurePluginDatabaseVersionMigration(pluginDomain, pluginDir, dbDir) {
const key = migrationKey(pluginDomain, pluginDir);
if (completedMigrations.has(key)) {
return;
}
const markDone = () => {
completedMigrations.add(key);
};
const targetVersion = await readPluginConfigVersion(pluginDir);
const hyperdbConfig = await readPluginHyperdbConfig(pluginDir);
const migrationPending = !isMigrationMarkedComplete(hyperdbConfig);
@@ -519,6 +562,7 @@ async function ensurePluginDatabaseVersionMigration(pluginDomain, pluginDir, dbD
logWarn('DBVersionMigration', `Marking migration complete for ${pluginDomain} to avoid retry loop`);
await finalizeVersionMigration(pluginDir, pluginDomain, targetVersion, hyperdbConfig?.previousVersion || null);
}
markDone();
return;
}
}
@@ -535,11 +579,13 @@ async function ensurePluginDatabaseVersionMigration(pluginDomain, pluginDir, dbD
}
await appendVersionHistory(pluginDir, targetVersion);
logDebug('DBVersionMigration', `Skipping migration for ${pluginDomain} (hyperdb.migrated=true)`);
markDone();
return;
}
if (manifest && manifest.version === targetVersion && !hyperdbConfig?.previousVersion) {
await finalizeVersionMigration(pluginDir, pluginDomain, targetVersion, null);
markDone();
return;
}
@@ -563,6 +609,7 @@ async function ensurePluginDatabaseVersionMigration(pluginDomain, pluginDir, dbD
`No manifest for ${pluginDomain}; data already on v${targetVersion} — migration skipped`
);
await finalizeVersionMigration(pluginDir, pluginDomain, targetVersion, null);
markDone();
return;
}
@@ -581,6 +628,7 @@ async function ensurePluginDatabaseVersionMigration(pluginDomain, pluginDir, dbD
`No prior database version with data for ${pluginDomain} (target v${targetVersion}) — migration skipped`
);
await finalizeVersionMigration(pluginDir, pluginDomain, targetVersion, hyperdbConfig?.previousVersion || null);
markDone();
return;
}
}
@@ -588,6 +636,7 @@ async function ensurePluginDatabaseVersionMigration(pluginDomain, pluginDir, dbD
if (sourceVersion === targetVersion) {
logInfo('DBVersionMigration', `Source and target are both v${targetVersion} for ${pluginDomain} — migration skipped`);
await finalizeVersionMigration(pluginDir, pluginDomain, targetVersion, sourceVersion);
markDone();
return;
}
@@ -597,6 +646,7 @@ async function ensurePluginDatabaseVersionMigration(pluginDomain, pluginDir, dbD
`Target v${targetVersion} for ${pluginDomain} already has data — migration skipped`
);
await finalizeVersionMigration(pluginDir, pluginDomain, targetVersion, sourceVersion);
markDone();
return;
}
@@ -613,6 +663,7 @@ async function ensurePluginDatabaseVersionMigration(pluginDomain, pluginDir, dbD
`No data on v${sourceVersion} for ${pluginDomain} — migration skipped (nothing to copy)`
);
await finalizeVersionMigration(pluginDir, pluginDomain, targetVersion, sourceVersion);
markDone();
return;
}
@@ -679,10 +730,12 @@ async function ensurePluginDatabaseVersionMigration(pluginDomain, pluginDir, dbD
}
await finalizeVersionMigration(pluginDir, pluginDomain, targetVersion, sourceVersion);
markDone();
}
module.exports = {
ensurePluginDatabaseVersionMigration,
runPendingPluginDatabaseMigrations,
readVersionManifest,
writeVersionManifest,
getCollectionIds
includes/plugins/plugin-handler.js
+16 -2
View File
@@ -548,7 +548,9 @@ async function loadAllPlugins() {
await fs.mkdir(pluginSitesDir, { recursive: true });
}
// Load plugins for each discovered internal domain
const pluginsToInit = [];
// Load plugin modules and schemas first (no live HyperDB yet)
for (const domain of internalDomains) {
const pluginDir = path.join(pluginSitesDir, domain);
const plugin = await loadPlugin(domain, pluginDir);
@@ -556,12 +558,24 @@ async function loadAllPlugins() {
if (plugin) {
plugins.set(domain, plugin);
loadedPlugins.set(domain, plugin);
await initializePlugin(plugin);
pluginsToInit.push(plugin);
} else {
logDebug('PluginHandler', `No plugin found for ${domain}, will use fallback handler`);
}
}
// Run version migrations before opening plugin Corestores (avoids lock contention with db-manager)
try {
const { runPendingPluginDatabaseMigrations } = require('./db-version-migration');
await runPendingPluginDatabaseMigrations(pluginsToInit);
} catch (err) {
logWarn('PluginHandler', `Plugin database migration phase failed: ${err.message}`);
}
for (const plugin of pluginsToInit) {
await initializePlugin(plugin);
}
logInfo('PluginHandler', `Loaded ${loadedPlugins.size} plugin(s) out of ${internalDomains.length} internal domain(s)`);
return loadedPlugins;
}
p2ns.js
+47 -37
View File
@@ -614,6 +614,37 @@ async function main() {
}
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 sent = await channelManager.sendToPeerAsync(
CORE_DOMAIN,
'request',
targetPeerId,
'request_invite',
inviteChannelWaitMs
);
if (sent) {
logInfo('Swarm', `Invite request sent to peer ${targetPeerId}`);
} else {
logWarn('Swarm', `Request channel not ready for peer ${targetPeerId}`);
}
return sent;
}
state.requestInviteFromPeer = requestInviteFromPeer;
// Register invite channel for sending/receiving invites
channelManager.registerPluginChannel(CORE_DOMAIN, 'invite', {
encoding: 'string',
@@ -786,6 +817,15 @@ async function main() {
channelManager.registerPluginChannel(CORE_DOMAIN, 'request', {
encoding: 'string',
autoReconnect: true,
onOpen: (peerId) => {
if (isMaster || getDnsPass()) {
return;
}
logDebug('Swarm', `Request channel open for peer ${peerId.substring(0, 16)}..., requesting invite`);
requestInviteFromPeer(peerId).catch((err) => {
logDebug('Swarm', `Invite request on channel open failed for ${peerId}: ${err.message}`);
});
},
onMessage: async (message, peerId) => {
logInfo('Swarm', `Received message from peer ${peerId}: ${message}`);
@@ -1383,6 +1423,7 @@ async function main() {
logInfo('Swarm', `Starting persistent invite retry loop (interval: ${retryIntervalMs}ms, relay after ${relayAfterAttempts} attempts, max attempts: ${maxRetryAttempts})`);
persistentInviteRetryInterval = setInterval(() => {
void (async () => {
const pass = getDnsPass();
if (pass) {
// Got invite, stop the loop
@@ -1405,14 +1446,14 @@ async function main() {
// Broadcast invite request to all connected peers
if (state.broadcastInviteRequest) {
logDebug('Swarm', `Persistent retry (attempt ${retryAttemptCount}): broadcasting invite request to all peers`);
state.broadcastInviteRequest();
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 {
channelManager.sendToPeer(CORE_DOMAIN, 'request', otherPeerId, 'request_invite');
await requestInviteFromPeer(otherPeerId);
} catch (err) {
logError('Swarm', `Error in persistent retry to ${otherPeerId}: ${err.message}`);
}
@@ -1437,6 +1478,7 @@ async function main() {
}
}
}
})();
}, retryIntervalMs);
// Store reference for cleanup
@@ -2193,8 +2235,7 @@ async function main() {
// Send request to this peer
const sendInviteRequestToPeer = async (targetPeerId, targetConn) => {
const pass = getDnsPass();
if (pass) {
if (getDnsPass()) {
logDebug('Swarm', 'dnsPass already initialized, skipping invite request');
return;
}
@@ -2204,27 +2245,8 @@ async function main() {
return;
}
// Check if request channel is available and ready
const requestChannelInfo = channelManager.getChannelInfo(CORE_DOMAIN, 'request');
const requestPeerChannel = requestChannelInfo?.peerChannels?.get(targetPeerId);
let channelReady = false;
if (requestPeerChannel) {
// Wait for channel to be bidirectional before sending
channelReady = await channelManager.waitForBidirectionalOpen(requestPeerChannel, 2000);
}
if (!channelReady) {
logDebug('Swarm', `Request channel not ready for peer ${targetPeerId}, proceeding anyway (protomux will queue)`);
}
try {
const sent = channelManager.sendToPeer(CORE_DOMAIN, 'request', targetPeerId, 'request_invite');
if (sent) {
logInfo('Swarm', `Invite request sent to peer ${targetPeerId}`);
} else {
logWarn('Swarm', `Request channel not ready for peer ${targetPeerId}`);
}
await requestInviteFromPeer(targetPeerId);
} catch (err) {
logError('Swarm', `Error sending invite request to peer ${targetPeerId}: ${err.message}`);
}
@@ -2269,21 +2291,9 @@ async function main() {
const otherConn = channels.conn;
if (!otherConn || otherConn.destroyed) continue;
// Check channel readiness before sending
const requestChannelInfo = channelManager.getChannelInfo(CORE_DOMAIN, 'request');
const requestPeerChannel = requestChannelInfo?.peerChannels?.get(otherPeerId);
const sendPromise = (async () => {
if (requestPeerChannel) {
// Give channel a moment to be ready, but don't block too long
const channelReady = await channelManager.waitForBidirectionalOpen(requestPeerChannel, 1000);
if (!channelReady) {
logDebug('Swarm', `Request channel not ready for broadcast to peer ${otherPeerId}, but proceeding`);
}
}
try {
const sent = channelManager.sendToPeer(CORE_DOMAIN, 'request', otherPeerId, 'request_invite');
const sent = await requestInviteFromPeer(otherPeerId);
if (sent) sentCount++;
return sent;
} catch (err) {
plugin-sites/peer.paste/config.json
+1 -1
View File
@@ -11,7 +11,7 @@
"www": "www",
"hyperdb": {
"previousVersion": "1.4.2",
"migrated": false,
"migrated": true,
"schemas": {
"namespace": "peerpaste",
"structs": [