Cyber/p2ns
1
0
Fork 0
forked from snxraven/p2ns
Code Pull Requests Activity

Modernize Holepunch stack for HyperDB 6 and current replication APIs

Browse Source 
Align P2NS with current Autopass, Hypercore, Hyperdrive, and Protomux behavior:
add plugin DB recovery for incompatible persisted HyperDB data, stop
unconditional spec rebuilds via fingerprinting, update SDK replication paths,
and align proxy invite channels with p2ns.core-* protocols without replicating
Autopass over the p2ns topic.
This commit is contained in:
Raven Scott
2026-05-27 21:02:17 -04:00
parent 7cd98231f7
commit d57433d3e0
19 changed files with 630 additions and 405 deletions
Download Patch File Download Diff File Expand all files Collapse all files
includes/plugins/channel-manager.js
+65 -64
View File
@@ -348,8 +348,8 @@ function createPluginChannelsForPeer(peerId, conn, mux) {
logDebug('ChannelManager', `Connection invalid for peer ${peerId.substring(0, 16)}..., skipping channel creation`);
continue;
}
if (!mux || mux.destroyed) {
logDebug('ChannelManager', `Mux invalid for peer ${peerId.substring(0, 16)}..., skipping channel creation`);
if (!mux || (conn && conn.destroyed)) {
logDebug('ChannelManager', `Mux/connection invalid for peer ${peerId.substring(0, 16)}..., skipping channel creation`);
continue;
}
@@ -808,45 +808,45 @@ async function waitForChannelOpen(peerChannel, timeout = 5000, checkInterval = 1
* @param {number} checkInterval - Check interval in milliseconds (default: 100)
* @returns {Promise<boolean>} True if channel is open bidirectionally, false if timeout
*/
async function waitForBidirectionalOpen(peerChannel, timeout = 2000, checkInterval = 100) {
async function waitForBidirectionalOpen(peerChannel, timeout = 2000) {
if (!peerChannel || !peerChannel.channel) {
return false;
}
const startTime = Date.now();
// Try to get peerId from the channel info or connection
let peerId = 'unknown';
if (peerChannel.conn && peerChannel.conn.remotePublicKey) {
peerId = peerChannel.conn.remotePublicKey.toString('hex').substring(0, 16);
}
logDebug('ChannelManager', `Waiting for bidirectional open on channel for peer ${peerId}..., timeout: ${timeout}ms`);
while (Date.now() - startTime < timeout) {
// Trust protomux's channel.opened as authoritative
// When channel.opened is true, the channel is fully bidirectional
if (peerChannel.channel.opened) {
// Sync our tracked state with protomux state
peerChannel.localOpened = true;
peerChannel.remoteOpened = true;
const elapsed = Date.now() - startTime;
logDebug('ChannelManager', `Channel became bidirectional for peer ${peerId}... after ${elapsed}ms`);
return true;
}
// Protomux 3 may fire onopen before channel.opened flips true
if (peerChannel.localOpened && peerChannel.remoteOpened) {
return true;
}
// Wait before checking again
await new Promise(resolve => setTimeout(resolve, checkInterval));
if (peerChannel.channel.opened) {
peerChannel.localOpened = true;
peerChannel.remoteOpened = true;
return true;
}
const elapsed = Date.now() - startTime;
logDebug('ChannelManager', `Channel not bidirectional for peer ${peerId}... within ${timeout}ms (waited ${elapsed}ms) - proceeding anyway (protomux will queue)`);
return false;
if (peerChannel.localOpened && peerChannel.remoteOpened) {
return true;
}
logDebug('ChannelManager', `Waiting for channel fullyOpened for peer ${peerId}... (${timeout}ms)`);
try {
const opened = await Promise.race([
peerChannel.channel.fullyOpened(),
new Promise((resolve) => setTimeout(() => resolve(false), timeout))
]);
if (opened) {
peerChannel.localOpened = true;
peerChannel.remoteOpened = true;
return true;
}
} catch (err) {
logDebug('ChannelManager', `fullyOpened() error for peer ${peerId}...: ${err.message}`);
}
logDebug('ChannelManager', `Channel not fully open for peer ${peerId}... within ${timeout}ms`);
return peerChannel.channel.opened || false;
}
/**
@@ -959,9 +959,6 @@ function sendToPeer(pluginDomain, protocol, peerId, data, waitTimeout = 2000) {
// Wait a brief moment for channel to reopen, then continue
}
// Try to send the message
// Protomux channels can send messages even if not fully opened - they will be queued
// Only fail if the channel or message object doesn't exist
try {
if (!peerChannel.channel) {
logDebug('ChannelManager', `Channel object missing for peer ${peerId.substring(0, 16)}... on ${pluginDomain}-${protocol}`);
@@ -972,16 +969,21 @@ function sendToPeer(pluginDomain, protocol, peerId, data, waitTimeout = 2000) {
logDebug('ChannelManager', `Message handler missing for peer ${peerId.substring(0, 16)}... on ${pluginDomain}-${protocol}`);
return false;
}
// Send the message - protomux will queue it if channel isn't fully open yet
peerChannel.message.send(data);
if (peerChannel.channel.opened) {
logDebug('ChannelManager', `Sent message to peer ${peerId.substring(0, 16)}... on ${pluginDomain}-${protocol} (channel open)`);
} else {
logDebug('ChannelManager', `Queued message for peer ${peerId.substring(0, 16)}... on ${pluginDomain}-${protocol} (channel opening)`);
if (peerChannel.channel.closed || peerChannel.conn?.destroyed) {
return false;
}
const sent = peerChannel.message.send(data);
if (!sent) {
logDebug('ChannelManager', `Send backpressure on ${pluginDomain}-${protocol} for peer ${peerId.substring(0, 16)}...`);
if (handler) {
attemptImmediateReopen(pluginDomain, protocol, peerId, peerChannel, handler);
}
return false;
}
logDebug('ChannelManager', `Sent message to peer ${peerId.substring(0, 16)}... on ${pluginDomain}-${protocol}`);
return true;
} catch (err) {
logError('ChannelManager', `Error sending message to peer ${peerId.substring(0, 16)}...: ${err.message}`);
@@ -1033,19 +1035,8 @@ async function sendToPeerAsync(pluginDomain, protocol, peerId, data, waitTimeout
// Check if channel is already open bidirectionally
const isBidirectional = peerChannel.channel && peerChannel.channel.opened;
// Optionally wait for bidirectional opening (but don't block if it times out)
// Protomux will queue messages even if channel isn't fully open
if (!isBidirectional && waitTimeout > 0) {
logDebug('ChannelManager', `Channel ${pluginDomain}-${protocol} not fully open for peer ${peerId.substring(0, 16)}..., waiting up to ${waitTimeout}ms`);
const opened = await waitForBidirectionalOpen(peerChannel, waitTimeout);
if (opened) {
logDebug('ChannelManager', `Channel ${pluginDomain}-${protocol} is now bidirectional for peer ${peerId.substring(0, 16)}...`);
} else {
// Channel not fully open yet, but protomux can queue messages
logDebug('ChannelManager', `Channel ${pluginDomain}-${protocol} not fully open yet for peer ${peerId.substring(0, 16)}..., sending anyway (protomux will queue)`);
}
} else if (!isBidirectional) {
logDebug('ChannelManager', `Channel ${pluginDomain}-${protocol} not fully open for peer ${peerId.substring(0, 16)}..., sending anyway (protomux will queue)`);
await waitForBidirectionalOpen(peerChannel, waitTimeout);
}
try {
@@ -1058,14 +1049,21 @@ async function sendToPeerAsync(pluginDomain, protocol, peerId, data, waitTimeout
logError('ChannelManager', `Message handler missing for peer ${peerId.substring(0, 16)}... on ${pluginDomain}-${protocol}`);
return false;
}
// Send message - protomux will queue if channel isn't fully open
peerChannel.message.send(data);
if (isBidirectional) {
logDebug('ChannelManager', `Sent message to peer ${peerId.substring(0, 16)}... on ${pluginDomain}-${protocol} (channel open)`);
} else {
logDebug('ChannelManager', `Queued message for peer ${peerId.substring(0, 16)}... on ${pluginDomain}-${protocol} (channel opening)`);
if (peerChannel.channel.closed || peerChannel.conn?.destroyed) {
return false;
}
const sent = peerChannel.message.send(data);
if (!sent) {
logDebug('ChannelManager', `Send backpressure on ${pluginDomain}-${protocol} for peer ${peerId.substring(0, 16)}...`);
if (handler) {
attemptImmediateReopen(pluginDomain, protocol, peerId, peerChannel, handler);
}
return false;
}
logDebug('ChannelManager', `Sent message to peer ${peerId.substring(0, 16)}... on ${pluginDomain}-${protocol}`);
return true;
} catch (err) {
logError('ChannelManager', `Error sending message to peer ${peerId.substring(0, 16)}...: ${err.message}`);
@@ -1112,9 +1110,12 @@ function broadcastToPeers(pluginDomain, protocol, data) {
continue;
}
// Send message - protomux will queue if channel not fully open
peerChannel.message.send(data);
sentCount++;
if (peerChannel.channel.closed || peerChannel.conn?.destroyed) {
continue;
}
if (peerChannel.message.send(data)) {
sentCount++;
}
} catch (err) {
logError('ChannelManager', `Error broadcasting to peer ${peerId.substring(0, 16)}...: ${err.message}`);
}
includes/plugins/db-manager.js
+9
View File
@@ -10,6 +10,7 @@ const path = require('path');
const fs = require('fs').promises;
const crypto = require('crypto');
const { logDebug, logError, logInfo, logWarn } = require('../infrastructure/logger');
const { getDatabaseWithRecovery } = require('./db-recovery');
// Per-plugin corestore instances
const pluginStores = new Map();
@@ -494,9 +495,17 @@ function getPluginCore(pluginDomain) {
return pluginCores.get(pluginDomain) || null;
}
/**
* Get or create database with incompatible-data recovery (sanity scan + local wipe).
*/
async function getDatabaseSafe(pluginDomain, pluginDir, dbDir) {
return getDatabaseWithRecovery(pluginDomain, pluginDir, dbDir, getDatabase);
}
module.exports = {
initializeStore,
getDatabase,
getDatabaseSafe,
closeDatabase,
getDatabaseInstance,
closeAllDatabases,
includes/plugins/db-recovery.js
+151
View File
@@ -0,0 +1,151 @@
/**
* Plugin HyperDB recovery helpers — detect incompatible persisted data and reset safely.
*/
const fs = require('fs').promises;
const path = require('path');
const { logInfo, logWarn, logError, logDebug } = require('../infrastructure/logger');
const INCOMPATIBLE_PATTERNS = [
/unknown collection type/i,
/unsupported version/i,
/decoding error/i,
/cannot read properties of undefined \(reading 'decode'\)/i,
/invalid collection/i,
/bad collection type/i
];
/**
* @param {Error|string} err
* @returns {boolean}
*/
function isIncompatibleDbError(err) {
const message = typeof err === 'string' ? err : (err && err.message) || '';
return INCOMPATIBLE_PATTERNS.some((re) => re.test(message));
}
/**
* @param {string} pluginDomain
* @param {string} pluginDir
* @returns {string}
*/
function getPluginDbStorageDir(pluginDomain, pluginDir) {
return path.join(pluginDir, 'db');
}
/**
* Wipe plugin local HyperDB storage and clear in-memory instances.
* @param {string} pluginDomain
* @param {string} pluginDir
* @param {Object} [options]
* @param {string} [options.reason]
* @returns {Promise<void>}
*/
async function resetPluginDatabaseStorage(pluginDomain, pluginDir, options = {}) {
const reason = options.reason || 'incompatible persisted data';
const dbManager = require('./db-manager');
const dbDir = getPluginDbStorageDir(pluginDomain, pluginDir);
logWarn(
'DBRecovery',
`Resetting plugin database for ${pluginDomain} (${reason}). Removing ${dbDir}`
);
try {
await dbManager.closeDatabase(pluginDomain, false);
} catch (err) {
logDebug('DBRecovery', `closeDatabase during reset: ${err.message}`);
}
try {
await fs.rm(dbDir, { recursive: true, force: true });
} catch (err) {
if (err.code !== 'ENOENT') {
logError('DBRecovery', `Failed to remove db dir for ${pluginDomain}: ${err.message}`);
throw err;
}
}
await fs.mkdir(dbDir, { recursive: true });
logInfo('DBRecovery', `Plugin database storage reset for ${pluginDomain}`);
}
/**
* Lightweight sanity scan — iterate one entry per collection to catch decode errors early.
* @param {Object} db - HyperDB instance
* @param {Object} def - Database definition
* @returns {Promise<void>}
*/
async function sanityScanDatabase(db, def) {
if (!db || db.closed || !def || !Array.isArray(def.collections)) {
return;
}
await db.ready();
if (typeof db.update === 'function') {
db.update();
}
for (const collection of def.collections) {
const collectionName = collection && collection.name;
if (!collectionName) continue;
try {
await db.find(collectionName, {}, { limit: 1 }).toArray();
} catch (err) {
if (isIncompatibleDbError(err)) {
throw err;
}
logDebug('DBRecovery', `Sanity scan skipped collection ${collectionName}: ${err.message}`);
}
}
}
/**
* Initialize DB with sanity scan; reset storage and retry once on incompatible data.
* @param {string} pluginDomain
* @param {string} pluginDir
* @param {string} specDbDir - HyperDB definition directory (index.js)
* @param {Function} getDatabase - dbManager.getDatabase bound
* @returns {Promise<Object>}
*/
async function getDatabaseWithRecovery(pluginDomain, pluginDir, specDbDir, getDatabase) {
let retried = false;
while (true) {
try {
const db = await getDatabase(pluginDomain, pluginDir, specDbDir);
const defPath = path.join(specDbDir, 'index.js');
delete require.cache[require.resolve(defPath)];
const def = require(defPath);
const definition = def && def.default ? def.default : def;
await sanityScanDatabase(db, definition);
return db;
} catch (err) {
if (!retried && isIncompatibleDbError(err)) {
retried = true;
logWarn(
'DBRecovery',
`Incompatible HyperDB data for ${pluginDomain}: ${err.message}. Wiping local db/ and recreating.`
);
logWarn(
'DBRecovery',
`If this persists after reset, bump version in plugin config.json to rotate the replicated core.`
);
await resetPluginDatabaseStorage(pluginDomain, pluginDir, { reason: err.message });
continue;
}
throw err;
}
}
}
module.exports = {
isIncompatibleDbError,
getPluginDbStorageDir,
resetPluginDatabaseStorage,
sanityScanDatabase,
getDatabaseWithRecovery
};
includes/plugins/drive-replication-manager.js
+36 -80
View File
@@ -288,96 +288,52 @@ function handleConnection(pluginDomain, driveName, conn) {
logWarn('DriveReplicationManager', `Download error for ${driveKey} from peer ${peerId.slice(0, 16)}: ${downloadErr.message}`);
}
// Also update and download the drive's metadata core (similar to HyperDB core updates)
// Hyperdrive uses a metadata core for the file system structure
// Metadata core (Hyperdrive 13: drive.core is a property)
try {
if (state.drive.core && typeof state.drive.core === 'function') {
const metadataCore = state.drive.core();
if (metadataCore && !metadataCore.closed) {
await metadataCore.ready();
const oldCoreLength = metadataCore.length;
await metadataCore.update({ wait: true });
const newCoreLength = metadataCore.length;
logInfo('DriveReplicationManager', `Metadata core for ${driveKey}: length ${oldCoreLength} -> ${newCoreLength}`);
// Always download core blocks if there's content, even if length unchanged
// This ensures all blocks are available locally
if (newCoreLength > 0) {
// Download all blocks (not just new ones) to ensure nothing is missing
metadataCore.download({ start: 0, end: newCoreLength });
logInfo('DriveReplicationManager', `Downloading all metadata core blocks for ${driveKey} from peer ${peerId.slice(0, 16)}: 0 to ${newCoreLength}`);
// Wait for core download to complete
await new Promise((resolve) => {
const checkDownload = setInterval(() => {
const downloaded = metadataCore.downloaded(0, newCoreLength);
if (downloaded >= newCoreLength) {
clearInterval(checkDownload);
logInfo('DriveReplicationManager', `Metadata core download completed for ${driveKey} from peer ${peerId.slice(0, 16)} (${downloaded}/${newCoreLength} blocks)`);
resolve();
}
}, 100);
// Timeout after 15 seconds for core download
setTimeout(() => {
clearInterval(checkDownload);
const downloaded = metadataCore.downloaded(0, newCoreLength);
logWarn('DriveReplicationManager', `Metadata core download timeout for ${driveKey} from peer ${peerId.slice(0, 16)} (${downloaded}/${newCoreLength} blocks downloaded, continuing in background)`);
resolve();
}, 15000);
});
}
const metadataCore = state.drive.core;
if (metadataCore && !metadataCore.closed) {
await metadataCore.ready();
const oldCoreLength = metadataCore.length;
await metadataCore.update({ wait: true });
const newCoreLength = metadataCore.length;
logInfo('DriveReplicationManager', `Metadata core for ${driveKey}: length ${oldCoreLength} -> ${newCoreLength}`);
if (newCoreLength > 0) {
logInfo('DriveReplicationManager', `Downloading metadata core blocks for ${driveKey}: 0 to ${newCoreLength}`);
const dl = metadataCore.download({ start: 0, end: newCoreLength });
await Promise.race([
dl.done(),
new Promise((resolve) => setTimeout(resolve, 15000))
]);
}
}
} catch (coreErr) {
// Core update errors are not critical
logDebug('DriveReplicationManager', `Error updating metadata core for ${driveKey}: ${coreErr.message}`);
}
// Also download content core blocks (file data)
// Hyperdrive has both metadata core (file structure) and content core (file data)
// Content / blobs core (Hyperdrive 13: await drive.getBlobs())
try {
if (state.drive.content && typeof state.drive.content === 'function') {
const contentCore = state.drive.content();
if (contentCore && !contentCore.closed) {
await contentCore.ready();
const oldContentLength = contentCore.length;
await contentCore.update({ wait: true });
const newContentLength = contentCore.length;
logInfo('DriveReplicationManager', `Content core for ${driveKey}: length ${oldContentLength} -> ${newContentLength}`);
// Download content core blocks if there's content
if (newContentLength > 0) {
// Download all content blocks to ensure file data is available
contentCore.download({ start: 0, end: newContentLength });
logInfo('DriveReplicationManager', `Downloading all content core blocks for ${driveKey} from peer ${peerId.slice(0, 16)}: 0 to ${newContentLength}`);
// Wait for content core download to complete
await new Promise((resolve) => {
const checkDownload = setInterval(() => {
const downloaded = contentCore.downloaded(0, newContentLength);
if (downloaded >= newContentLength) {
clearInterval(checkDownload);
logInfo('DriveReplicationManager', `Content core download completed for ${driveKey} from peer ${peerId.slice(0, 16)} (${downloaded}/${newContentLength} blocks)`);
resolve();
}
}, 100);
// Timeout after 30 seconds for content core download (can be large)
setTimeout(() => {
clearInterval(checkDownload);
const downloaded = contentCore.downloaded(0, newContentLength);
logWarn('DriveReplicationManager', `Content core download timeout for ${driveKey} from peer ${peerId.slice(0, 16)} (${downloaded}/${newContentLength} blocks downloaded, continuing in background)`);
resolve();
}, 30000);
});
}
const blobs = await state.drive.getBlobs();
const contentCore = blobs && blobs.core;
if (contentCore && !contentCore.closed) {
await contentCore.ready();
const oldContentLength = contentCore.length;
await contentCore.update({ wait: true });
const newContentLength = contentCore.length;
logInfo('DriveReplicationManager', `Content core for ${driveKey}: length ${oldContentLength} -> ${newContentLength}`);
if (newContentLength > 0) {
logInfo('DriveReplicationManager', `Downloading content core blocks for ${driveKey}: 0 to ${newContentLength}`);
const dl = contentCore.download({ start: 0, end: newContentLength });
await Promise.race([
dl.done(),
new Promise((resolve) => setTimeout(resolve, 30000))
]);
}
}
} catch (contentErr) {
// Content core errors are not critical
logDebug('DriveReplicationManager', `Error updating content core for ${driveKey}: ${contentErr.message}`);
}
} catch (downloadErr) {
includes/plugins/hyperdb-builder.js
+41 -6
View File
@@ -11,6 +11,11 @@ const fs = require('fs').promises;
const fsSync = require('fs');
const path = require('path');
const { logDebug, logError, logInfo, logWarn } = require('../infrastructure/logger');
const {
computeSpecFingerprint,
readStoredFingerprint,
writeStoredFingerprint
} = require('./hyperdb-spec-hash');
const PROJECT_ROOT = path.resolve(__dirname, '../..');
@@ -73,6 +78,7 @@ async function buildSchemaFromConfig(pluginDir, config) {
// Build HyperDB definitions
await buildHyperDBDefinition(pluginDir, schemaDir, dbDir, hyperdbConfig);
await writeSpecFingerprint(pluginDir, config, { schemaDir, dbDir });
logInfo('HyperDBBuilder', `Built schemas for plugin in ${pluginDir}`);
return { schemaDir, dbDir };
@@ -233,6 +239,18 @@ async function buildHyperDBDefinition(pluginDir, schemaDir, dbDir, hyperdbConfig
logDebug('HyperDBBuilder', `HyperDB definitions saved to ${dbDir}`);
}
/**
* Write spec fingerprint after a successful build
* @param {string} pluginDir
* @param {Object} config
* @param {{ schemaDir: string, dbDir: string }} specDirs
*/
async function writeSpecFingerprint(pluginDir, config, specDirs) {
const fingerprint = await computeSpecFingerprint(pluginDir, config, specDirs);
await writeStoredFingerprint(specDirs.dbDir, fingerprint);
return fingerprint;
}
/**
* Check if schemas need to be rebuilt
* @param {string} pluginDir - Plugin directory path
@@ -260,9 +278,21 @@ async function needsRebuild(pluginDir, config) {
return true;
}
// For now, always rebuild to ensure consistency
// In the future, could check file modification times vs config modification
return true;
const defPath = path.join(dbDir, 'index.js');
try {
await fs.access(defPath);
} catch {
return true;
}
const current = await computeSpecFingerprint(pluginDir, config, { schemaDir, dbDir });
const stored = await readStoredFingerprint(dbDir);
if (!stored || stored !== current) {
logDebug('HyperDBBuilder', `Spec fingerprint changed for ${config.domain || path.basename(pluginDir)} (rebuild needed)`);
return true;
}
return false;
} catch (err) {
// Directories don't exist, need to build
return true;
@@ -282,12 +312,16 @@ async function ensureSchemaBuilt(pluginDir, config) {
const shouldRebuild = await needsRebuild(pluginDir, config);
const specDirs = resolveSpecDirs(pluginDir, config);
if (shouldRebuild) {
logInfo('HyperDBBuilder', `Building HyperDB schemas for plugin in ${pluginDir}`);
return await buildSchemaFromConfig(pluginDir, config);
const built = await buildSchemaFromConfig(pluginDir, config);
await writeSpecFingerprint(pluginDir, config, built);
return built;
}
return resolveSpecDirs(pluginDir, config);
return specDirs;
}
module.exports = {
@@ -296,6 +330,7 @@ module.exports = {
buildHyperDBDefinition,
needsRebuild,
ensureSchemaBuilt,
resolveSpecDirs
resolveSpecDirs,
writeSpecFingerprint
};
includes/plugins/hyperdb-spec-hash.js
+94
View File
@@ -0,0 +1,94 @@
/**
* Hash plugin hyperdb config + spec sources to decide whether schemas need rebuilding.
*/
const crypto = require('crypto');
const fs = require('fs').promises;
const fsSync = require('fs');
const path = require('path');
async function fileHash(filePath) {
const data = await fs.readFile(filePath);
return crypto.createHash('sha256').update(data).digest('hex');
}
async function directoryHash(dirPath) {
let entries;
try {
entries = await fs.readdir(dirPath, { withFileTypes: true });
} catch (err) {
if (err.code === 'ENOENT') return '';
throw err;
}
const parts = [];
entries.sort((a, b) => a.name.localeCompare(b.name));
for (const entry of entries) {
const full = path.join(dirPath, entry.name);
if (entry.isDirectory()) {
const nested = await directoryHash(full);
if (nested) parts.push(`${entry.name}/:${nested}`);
} else if (entry.isFile()) {
const stat = await fs.stat(full);
parts.push(`${entry.name}:${stat.mtimeMs}:${await fileHash(full)}`);
}
}
return parts.join('|');
}
/**
* @param {string} pluginDir
* @param {Object} config
* @param {{ schemaDir: string, dbDir: string }} specDirs
* @returns {Promise<string>}
*/
async function computeSpecFingerprint(pluginDir, config, specDirs) {
const hash = crypto.createHash('sha256');
hash.update(JSON.stringify(config.hyperdb || {}));
const configPath = path.join(pluginDir, 'config.json');
if (fsSync.existsSync(configPath)) {
hash.update(await fileHash(configPath));
}
const helpers = config.hyperdb && config.hyperdb.helpers;
if (helpers) {
const helpersPath = path.resolve(pluginDir, helpers);
if (fsSync.existsSync(helpersPath)) {
hash.update(await fileHash(helpersPath));
}
}
hash.update(await directoryHash(specDirs.schemaDir));
return hash.digest('hex');
}
/**
* @param {string} dbDir
* @returns {Promise<string|null>}
*/
async function readStoredFingerprint(dbDir) {
try {
return (await fs.readFile(path.join(dbDir, '.p2ns-spec-fingerprint'), 'utf8')).trim();
} catch (err) {
if (err.code === 'ENOENT') return null;
throw err;
}
}
/**
* @param {string} dbDir
* @param {string} fingerprint
* @returns {Promise<void>}
*/
async function writeStoredFingerprint(dbDir, fingerprint) {
await fs.writeFile(path.join(dbDir, '.p2ns-spec-fingerprint'), fingerprint, 'utf8');
}
module.exports = {
computeSpecFingerprint,
readStoredFingerprint,
writeStoredFingerprint
};
includes/plugins/plugin-handler.js
+2 -2
View File
@@ -217,7 +217,7 @@ async function initializePlugin(plugin) {
// Initialize SDK with plugin context (domain and config)
// This will set global PLUGIN_DOMAIN from config.json
sdk._initializePluginContext(plugin.domain, plugin.pluginDir, plugin.config);
sdk._initializePluginContext(plugin.domain, plugin.pluginDir, plugin.config, plugin.dbConfig);
// Initialize database if hyperdb config exists
if (plugin.dbConfig) {
try {
@@ -306,7 +306,7 @@ async function initializePlugin(plugin) {
// Initialize database
const dbManager = require('./db-manager');
const db = await dbManager.getDatabase(plugin.domain, plugin.pluginDir, plugin.dbConfig.dbDir);
const db = await dbManager.getDatabaseSafe(plugin.domain, plugin.pluginDir, plugin.dbConfig.dbDir);
plugin.db = db;
logInfo('PluginHandler', `Database initialized successfully for plugin ${plugin.domain}`);
} catch (err) {
includes/plugins/replication-manager.js
+25 -50
View File
@@ -129,6 +129,9 @@ async function handleConnection(pluginDomain, conn) {
try {
// Replicate the plugin's store over this connection
// Note: Corestore handles connection multiplexing automatically
if (typeof state.store.findingPeers === 'function') {
state.store.findingPeers();
}
state.store.replicate(conn, { live: true });
// Track peer
@@ -169,59 +172,31 @@ async function handleConnection(pluginDomain, conn) {
// Download any new data
if (newLength > oldLength) {
core.download({ start: oldLength, end: newLength });
logInfo('ReplicationManager', `Downloading new data for ${pluginDomain} from peer ${peerId.slice(0, 16)}: ${oldLength} to ${newLength}`);
// Wait for download to complete
await new Promise((resolve) => {
let checkDownload = null;
let downloadTimeout = null;
const cleanupTimers = () => {
if (checkDownload) {
clearInterval(checkDownload);
resources.timers.delete(checkDownload);
}
if (downloadTimeout) {
clearTimeout(downloadTimeout);
resources.timers.delete(downloadTimeout);
}
conn.removeListener('close', cleanupTimers);
conn.removeListener('error', cleanupTimers);
};
checkDownload = setInterval(() => {
try {
if (resources.isClosed || core.closed) {
cleanupTimers();
try {
const download = core.download({ start: oldLength, end: newLength });
await Promise.race([
download.done(),
new Promise((resolve) => {
const downloadTimeout = setTimeout(() => {
logWarn('ReplicationManager', `Download timeout for ${pluginDomain} from peer ${peerId.slice(0, 16)}`);
resolve();
return;
}
const downloaded = core.downloaded(oldLength, newLength);
if (downloaded >= (newLength - oldLength)) {
cleanupTimers();
logInfo('ReplicationManager', `Download completed for ${pluginDomain} from peer ${peerId.slice(0, 16)}`);
}, 5000);
resources.timers.add(downloadTimeout);
conn.once('close', () => {
clearTimeout(downloadTimeout);
resolve();
}
} catch (err) {
logDebug('ReplicationManager', `Error checking download status for ${pluginDomain}: ${err.message}`);
cleanupTimers();
resolve();
}
}, 100);
resources.timers.add(checkDownload);
downloadTimeout = setTimeout(() => {
cleanupTimers();
logWarn('ReplicationManager', `Download timeout for ${pluginDomain} from peer ${peerId.slice(0, 16)}`);
resolve();
}, 5000);
resources.timers.add(downloadTimeout);
// Ensure cleanup if connection closes during download check
conn.once('close', cleanupTimers);
conn.once('error', cleanupTimers);
});
});
conn.once('error', () => {
clearTimeout(downloadTimeout);
resolve();
});
})
]);
logInfo('ReplicationManager', `Download completed for ${pluginDomain} from peer ${peerId.slice(0, 16)}`);
} catch (dlErr) {
logDebug('ReplicationManager', `Download wait ended for ${pluginDomain}: ${dlErr.message}`);
}
}
} else if (core && core.closed) {
logDebug('ReplicationManager', `Core is closed for ${pluginDomain}, skipping update`);
includes/plugins/sdk.js
+131 -175
View File
@@ -63,12 +63,13 @@ let activePluginDomain = null;
* @param {string} pluginDir - Plugin directory path
* @param {Object} config - Plugin config object
*/
function _initializePluginContext(domain, pluginDir, config) {
function _initializePluginContext(domain, pluginDir, config, dbConfig = null) {
// Store in registry keyed by domain
pluginRegistry.set(domain, {
domain,
pluginDir,
config
config,
dbConfig
});
// Set active domain for current request/operation
@@ -95,6 +96,18 @@ function _getPluginDirForDomain(domain) {
return ctx ? ctx.pluginDir : null;
}
/**
* Get HyperDB spec directory for a plugin (cache or plugin spec path).
* @param {string} domain
* @returns {string|null}
*/
function _getPluginDbSpecDir(domain) {
const ctx = pluginRegistry.get(domain);
if (!ctx) return null;
if (ctx.dbConfig && ctx.dbConfig.dbDir) return ctx.dbConfig.dbDir;
return path.join(ctx.pluginDir, 'spec', 'hyperdb');
}
/**
* Get plugin domain - returns the active plugin domain or reads from config.json
* @returns {string|null} Plugin domain or null
@@ -3687,6 +3700,90 @@ const sdk = {
* Provides HyperDB database access for plugins
*/
db: {
/**
* Reopen HyperDB session (preserves core for replication by default).
* @param {boolean} preserveCore
* @returns {Promise<Object>}
*/
async _reopenDatabase(preserveCore = true) {
const pluginDomain = _getPluginDomain();
if (!pluginDomain) {
throw new Error('Plugin domain not available');
}
const dbManager = require('./db-manager');
const pluginDir = _getPluginDirForDomain(pluginDomain);
const dbDir = _getPluginDbSpecDir(pluginDomain);
if (!pluginDir || !dbDir) {
throw new Error(`Plugin paths not found for ${pluginDomain}`);
}
try {
await dbManager.closeDatabase(pluginDomain, preserveCore);
} catch (closeErr) {
logDebug('PluginSDK', `Error closing database during reopen: ${closeErr.message}`);
}
const db = await dbManager.getDatabase(pluginDomain, pluginDir, dbDir);
if (!db) {
throw new Error('Failed to reopen database instance');
}
return db;
},
/**
* Reset local plugin DB storage and reopen (schema mismatch).
* @returns {Promise<Object>}
*/
async _resetAndReopenDatabase() {
const pluginDomain = _getPluginDomain();
const pluginDir = _getPluginDirForDomain(pluginDomain);
const dbDir = _getPluginDbSpecDir(pluginDomain);
if (!pluginDomain || !pluginDir || !dbDir) {
throw new Error('Plugin domain not available');
}
const { resetPluginDatabaseStorage } = require('./db-recovery');
const dbManager = require('./db-manager');
await resetPluginDatabaseStorage(pluginDomain, pluginDir, { reason: 'SDK incompatible data recovery' });
const db = await dbManager.getDatabaseSafe(pluginDomain, pluginDir, dbDir);
if (!db) {
throw new Error('Failed to recreate database after reset');
}
return db;
},
/**
* Handle incompatible persisted data or session errors during DB ops.
* @param {Error} err
* @param {Object|null} db
* @param {number} attempt
* @param {number} maxAttempts
* @returns {Promise<Object|null>} New db instance to retry, or null to rethrow
*/
async _recoverFromDbError(err, db, attempt, maxAttempts) {
const { isIncompatibleDbError } = require('./db-recovery');
if (isIncompatibleDbError(err) && attempt < maxAttempts) {
logWarn('PluginSDK', `Incompatible database data: ${err.message}. Resetting local plugin DB...`);
return this._resetAndReopenDatabase();
}
const isSessionClosed = err.message && (
err.message.includes('SESSION_CLOSED') ||
err.message.includes('Database not available') ||
err.message.includes('Database is closed')
);
if (isSessionClosed && attempt < maxAttempts) {
return this._reopenDatabase(true);
}
return null;
},
/**
* Get the current plugin's database instance
* @returns {Object|null} Database instance or null if not available
@@ -3762,7 +3859,7 @@ const sdk = {
throw new Error(`Plugin directory not found for ${pluginDomain}`);
}
const dbDir = path.join(pluginDir, 'spec', 'hyperdb');
const dbDir = _getPluginDbSpecDir(pluginDomain);
logDebug('PluginSDK', `[db.ready] Recreating database at ${dbDir}...`);
db = await dbManager.getDatabase(pluginDomain, pluginDir, dbDir);
if (!db) {
@@ -3827,49 +3924,11 @@ const sdk = {
return;
} catch (err) {
logError('PluginSDK', `[db.insert] Attempt ${attempt} failed: ${err.message}`);
const isSessionClosed = err.message && (
err.message.includes('SESSION_CLOSED') ||
err.message.includes('Database not available') ||
err.message.includes('Database is closed')
);
if (isSessionClosed && attempt < MAX_ATTEMPTS) {
logWarn('PluginSDK', `[db.insert] Database session closed (attempt ${attempt}/${MAX_ATTEMPTS}), recreating database instance for ${pluginDomain}...`);
try {
const dbManager = require('./db-manager');
const existingCoreKey = dbManager.getPluginCoreKey(pluginDomain);
if (!existingCoreKey) {
logError('PluginSDK', `[db.insert] No existing core key found for ${pluginDomain}`);
throw new Error('No existing core key found for plugin');
}
try {
await dbManager.closeDatabase(pluginDomain, true);
} catch (closeErr) {
logDebug('PluginSDK', `[db.insert] Error closing old database instance: ${closeErr.message}`);
}
const pluginDir = _getPluginDirForDomain(pluginDomain);
logDebug('PluginSDK', `[db.insert] pluginDir=${pluginDir}`);
if (!pluginDir) {
logError('PluginSDK', `[db.insert] Plugin directory not found for ${pluginDomain}`);
throw new Error(`Plugin directory not found for ${pluginDomain}`);
}
const dbDir = path.join(pluginDir, 'spec', 'hyperdb');
logDebug('PluginSDK', `[db.insert] Recreating database at ${dbDir}...`);
db = await dbManager.getDatabase(pluginDomain, pluginDir, dbDir);
if (!db) {
logError('PluginSDK', `[db.insert] Failed to recreate database instance`);
throw new Error('Failed to recreate database instance');
}
logInfo('PluginSDK', `Database instance recreated for ${pluginDomain}, retrying insert...`);
continue;
} catch (recreateErr) {
logError('PluginSDK', `Failed to recreate database: ${recreateErr.message}`);
}
const recovered = await this._recoverFromDbError(err, db, attempt, MAX_ATTEMPTS);
if (recovered) {
db = recovered;
logInfo('PluginSDK', `Database recovered for ${pluginDomain}, retrying insert...`);
continue;
}
throw err;
}
@@ -3903,44 +3962,11 @@ const sdk = {
}
return await db.get(collection, query);
} catch (err) {
const isSessionClosed = err.message && (
err.message.includes('SESSION_CLOSED') ||
err.message.includes('Database not available') ||
err.message.includes('Database is closed')
);
if (isSessionClosed && attempt < MAX_ATTEMPTS) {
logWarn('PluginSDK', `Database session closed during get (attempt ${attempt}/${MAX_ATTEMPTS}), recreating database instance for ${pluginDomain}...`);
try {
const dbManager = require('./db-manager');
const existingCoreKey = dbManager.getPluginCoreKey(pluginDomain);
if (!existingCoreKey) {
throw new Error('No existing core key found for plugin');
}
try {
await dbManager.closeDatabase(pluginDomain, true);
} catch (closeErr) {
logDebug('PluginSDK', `Error closing old database instance: ${closeErr.message}`);
}
const pluginDir = _getPluginDirForDomain(pluginDomain);
if (!pluginDir) {
throw new Error(`Plugin directory not found for ${pluginDomain}`);
}
const dbDir = path.join(pluginDir, 'spec', 'hyperdb');
db = await dbManager.getDatabase(pluginDomain, pluginDir, dbDir);
if (!db) {
throw new Error('Failed to recreate database instance');
}
logInfo('PluginSDK', `Database instance recreated for ${pluginDomain}, retrying get...`);
continue;
} catch (recreateErr) {
logError('PluginSDK', `Failed to recreate database: ${recreateErr.message}`);
}
const recovered = await this._recoverFromDbError(err, db, attempt, MAX_ATTEMPTS);
if (recovered) {
db = recovered;
logInfo('PluginSDK', `Database recovered for ${pluginDomain}, retrying get...`);
continue;
}
throw err;
}
@@ -4004,59 +4030,11 @@ const sdk = {
return results;
} catch (err) {
const isSessionClosed = err.message && (
err.message.includes('SESSION_CLOSED') ||
err.message.includes('Database not available') ||
err.message.includes('Database is closed')
);
if (isSessionClosed && attempt < MAX_ATTEMPTS) {
logWarn('PluginSDK', `Database session closed (attempt ${attempt}/${MAX_ATTEMPTS}), recreating database instance for ${pluginDomain}...`);
// Recreate database instance while preserving the core for replication
try {
const dbManager = require('./db-manager');
// Get the existing core key to ensure we use the same core
const existingCoreKey = dbManager.getPluginCoreKey(pluginDomain);
if (!existingCoreKey) {
throw new Error('No existing core key found for plugin');
}
// Close existing database instance but preserve the core for replication
try {
await dbManager.closeDatabase(pluginDomain, true); // preserveCore = true
} catch (closeErr) {
logDebug('PluginSDK', `Error closing old database instance: ${closeErr.message}`);
}
// Get plugin directory for database initialization
const pluginDir = _getPluginDirForDomain(pluginDomain);
if (!pluginDir) {
throw new Error(`Plugin directory not found for ${pluginDomain}`);
}
const dbDir = path.join(pluginDir, 'spec', 'hyperdb');
// Create fresh database instance (getDatabase will reuse the existing core)
db = await dbManager.getDatabase(pluginDomain, pluginDir, dbDir);
if (!db) {
throw new Error('Failed to recreate database instance');
}
// Verify the core key matches
const newCoreKey = dbManager.getPluginCoreKey(pluginDomain);
if (!newCoreKey || !newCoreKey.equals(existingCoreKey)) {
logWarn('PluginSDK', `Core key mismatch after recreation - this may break replication`);
}
logInfo('PluginSDK', `Database instance recreated for ${pluginDomain} with core key ${existingCoreKey.toString('hex').slice(0, 16)}..., retrying query...`);
continue;
} catch (recreateErr) {
logError('PluginSDK', `Failed to recreate database: ${recreateErr.message}`);
// Fall through to error handling
}
const recovered = await this._recoverFromDbError(err, db, attempt, MAX_ATTEMPTS);
if (recovered) {
db = recovered;
logInfo('PluginSDK', `Database recovered for ${pluginDomain}, retrying query...`);
continue;
}
// If it's a timeout error or we've exhausted retries, provide a clearer message
@@ -4135,43 +4113,11 @@ const sdk = {
err.message.includes('Database is closed')
);
if (isSessionClosed && attempt < MAX_ATTEMPTS) {
logWarn('PluginSDK', `[db.flush] Database session closed (attempt ${attempt}/${MAX_ATTEMPTS}), recreating database instance for ${pluginDomain}...`);
try {
const dbManager = require('./db-manager');
const existingCoreKey = dbManager.getPluginCoreKey(pluginDomain);
if (!existingCoreKey) {
logError('PluginSDK', `[db.flush] No existing core key found for ${pluginDomain}`);
throw new Error('No existing core key found for plugin');
}
try {
await dbManager.closeDatabase(pluginDomain, true);
} catch (closeErr) {
logDebug('PluginSDK', `[db.flush] Error closing old database instance: ${closeErr.message}`);
}
const pluginDir = _getPluginDirForDomain(pluginDomain);
logDebug('PluginSDK', `[db.flush] pluginDir=${pluginDir}`);
if (!pluginDir) {
logError('PluginSDK', `[db.flush] Plugin directory not found for ${pluginDomain}`);
throw new Error(`Plugin directory not found for ${pluginDomain}`);
}
const dbDir = path.join(pluginDir, 'spec', 'hyperdb');
logDebug('PluginSDK', `[db.flush] Recreating database at ${dbDir}...`);
db = await dbManager.getDatabase(pluginDomain, pluginDir, dbDir);
if (!db) {
logError('PluginSDK', `[db.flush] Failed to recreate database instance`);
throw new Error('Failed to recreate database instance');
}
logInfo('PluginSDK', `[db.flush] Database instance recreated for ${pluginDomain}, retrying flush...`);
continue;
} catch (recreateErr) {
logError('PluginSDK', `[db.flush] Failed to recreate database: ${recreateErr.message}`);
}
const recovered = await this._recoverFromDbError(err, db, attempt, MAX_ATTEMPTS);
if (recovered) {
db = recovered;
logInfo('PluginSDK', `[db.flush] Database recovered for ${pluginDomain}, retrying flush...`);
continue;
}
throw err;
}
@@ -4297,7 +4243,17 @@ const sdk = {
if (!db) {
return;
}
db.reload();
if (typeof db.update === 'function') {
db.update();
}
},
/**
* Refresh the internal snapshot from the underlying core (HyperDB 6).
* @returns {void}
*/
update() {
this.reload();
},
/**