fix(core): improve shutdown cleanup, socket/stream robustness, and memory/cache handling
This commit is contained in:
@@ -3,6 +3,6 @@
|
||||
*/
|
||||
export const commitinfo = {
|
||||
name: '@serve.zone/dcrouter',
|
||||
version: '10.1.1',
|
||||
version: '10.1.2',
|
||||
description: 'A multifaceted routing service handling mail and SMS delivery functions.'
|
||||
}
|
||||
|
||||
@@ -23,6 +23,7 @@ import { MetricsManager } from './monitoring/index.js';
|
||||
import { RadiusServer, type IRadiusServerConfig } from './radius/index.js';
|
||||
import { RemoteIngressManager, TunnelManager } from './remoteingress/index.js';
|
||||
import { RouteConfigManager, ApiTokenManager } from './config/index.js';
|
||||
import { SecurityLogger, ContentScanner, IPReputationChecker } from './security/index.js';
|
||||
|
||||
export interface IDcRouterOptions {
|
||||
/** Base directory for all dcrouter data. Defaults to ~/.serve.zone/dcrouter */
|
||||
@@ -956,6 +957,7 @@ export class DcRouter {
|
||||
// Stop cache database after other services (they may need it during shutdown)
|
||||
if (this.cacheDb) {
|
||||
await this.cacheDb.stop().catch(err => logger.log('error', 'Error stopping CacheDb', { error: String(err) }));
|
||||
CacheDb.resetInstance();
|
||||
}
|
||||
|
||||
// Clear backoff cache in cert scheduler
|
||||
@@ -979,6 +981,11 @@ export class DcRouter {
|
||||
this.apiTokenManager = undefined;
|
||||
this.certificateStatusMap.clear();
|
||||
|
||||
// Reset security singletons to allow GC
|
||||
SecurityLogger.resetInstance();
|
||||
ContentScanner.resetInstance();
|
||||
IPReputationChecker.resetInstance();
|
||||
|
||||
logger.log('info', 'All DcRouter services stopped');
|
||||
} catch (error) {
|
||||
logger.log('error', 'Error during DcRouter shutdown', { error: String(error) });
|
||||
@@ -1363,15 +1370,25 @@ export class DcRouter {
|
||||
return;
|
||||
}
|
||||
|
||||
// Prevent uncaught exception from socket 'error' events
|
||||
socket.on('error', (err) => {
|
||||
logger.log('error', `DNS socket error: ${err.message}`);
|
||||
if (!socket.destroyed) {
|
||||
socket.destroy();
|
||||
}
|
||||
});
|
||||
|
||||
logger.log('debug', 'DNS socket handler: passing socket to DnsServer');
|
||||
|
||||
|
||||
try {
|
||||
// Use the built-in socket handler from smartdns
|
||||
// This handles HTTP/2, DoH protocol, etc.
|
||||
await (this.dnsServer as any).handleHttpsSocket(socket);
|
||||
} catch (error) {
|
||||
logger.log('error', `DNS socket handler error: ${error.message}`);
|
||||
socket.destroy();
|
||||
if (!socket.destroyed) {
|
||||
socket.destroy();
|
||||
}
|
||||
}
|
||||
};
|
||||
}
|
||||
|
||||
@@ -111,6 +111,15 @@ export class MetricsManager {
|
||||
this.securityMetrics.lastResetDate = currentDate;
|
||||
}
|
||||
|
||||
// Prune old query timestamps (keep last 5 minutes)
|
||||
const fiveMinutesAgo = Date.now() - 300000;
|
||||
const idx = this.dnsMetrics.queryTimestamps.findIndex(ts => ts >= fiveMinutesAgo);
|
||||
if (idx > 0) {
|
||||
this.dnsMetrics.queryTimestamps = this.dnsMetrics.queryTimestamps.slice(idx);
|
||||
} else if (idx === -1) {
|
||||
this.dnsMetrics.queryTimestamps = [];
|
||||
}
|
||||
|
||||
// Prune old time-series buckets every minute (don't wait for lazy query)
|
||||
this.pruneOldBuckets();
|
||||
}, 60000); // Check every minute
|
||||
@@ -427,13 +436,9 @@ export class MetricsManager {
|
||||
this.dnsMetrics.cacheMisses++;
|
||||
}
|
||||
|
||||
// Track query timestamp
|
||||
// Track query timestamp (pruning moved to resetInterval to avoid O(n) per query)
|
||||
this.dnsMetrics.queryTimestamps.push(Date.now());
|
||||
|
||||
// Keep only timestamps from last 5 minutes
|
||||
const fiveMinutesAgo = Date.now() - 300000;
|
||||
this.dnsMetrics.queryTimestamps = this.dnsMetrics.queryTimestamps.filter(ts => ts >= fiveMinutesAgo);
|
||||
|
||||
// Track response time if provided
|
||||
if (responseTimeMs) {
|
||||
this.dnsMetrics.responseTimes.push(responseTimeMs);
|
||||
|
||||
@@ -318,11 +318,15 @@ export class LogsHandler {
|
||||
try {
|
||||
// Use a timeout to detect hung streams (sendData can hang if the
|
||||
// VirtualStream's keepAlive loop has ended)
|
||||
let timeoutHandle: ReturnType<typeof setTimeout>;
|
||||
await Promise.race([
|
||||
virtualStream.sendData(encoder.encode(logData)),
|
||||
new Promise<never>((_, reject) =>
|
||||
setTimeout(() => reject(new Error('stream send timeout')), 10_000)
|
||||
),
|
||||
virtualStream.sendData(encoder.encode(logData)).then((result) => {
|
||||
clearTimeout(timeoutHandle);
|
||||
return result;
|
||||
}),
|
||||
new Promise<never>((_, reject) => {
|
||||
timeoutHandle = setTimeout(() => reject(new Error('stream send timeout')), 10_000);
|
||||
}),
|
||||
]);
|
||||
} catch {
|
||||
// Stream closed, errored, or timed out — clean up
|
||||
|
||||
@@ -182,7 +182,14 @@ export class ContentScanner {
|
||||
}
|
||||
return ContentScanner.instance;
|
||||
}
|
||||
|
||||
|
||||
/**
|
||||
* Reset the singleton instance (for shutdown/testing)
|
||||
*/
|
||||
public static resetInstance(): void {
|
||||
ContentScanner.instance = undefined;
|
||||
}
|
||||
|
||||
/**
|
||||
* Scan an email for malicious content
|
||||
* @param email The email to scan
|
||||
|
||||
@@ -65,6 +65,8 @@ export class IPReputationChecker {
|
||||
private reputationCache: LRUCache<string, IReputationResult>;
|
||||
private options: Required<IIPReputationOptions>;
|
||||
private storageManager?: any; // StorageManager instance
|
||||
private saveCacheTimer: ReturnType<typeof setTimeout> | null = null;
|
||||
private static readonly SAVE_CACHE_DEBOUNCE_MS = 30_000;
|
||||
|
||||
// Default DNSBL servers
|
||||
private static readonly DEFAULT_DNSBL_SERVERS = [
|
||||
@@ -143,7 +145,20 @@ export class IPReputationChecker {
|
||||
}
|
||||
return IPReputationChecker.instance;
|
||||
}
|
||||
|
||||
|
||||
/**
|
||||
* Reset the singleton instance (for shutdown/testing)
|
||||
*/
|
||||
public static resetInstance(): void {
|
||||
if (IPReputationChecker.instance) {
|
||||
if (IPReputationChecker.instance.saveCacheTimer) {
|
||||
clearTimeout(IPReputationChecker.instance.saveCacheTimer);
|
||||
IPReputationChecker.instance.saveCacheTimer = null;
|
||||
}
|
||||
}
|
||||
IPReputationChecker.instance = undefined;
|
||||
}
|
||||
|
||||
/**
|
||||
* Check an IP address's reputation
|
||||
* @param ip IP address to check
|
||||
@@ -213,12 +228,9 @@ export class IPReputationChecker {
|
||||
// Update cache with result
|
||||
this.reputationCache.set(ip, result);
|
||||
|
||||
// Save cache if enabled
|
||||
// Schedule debounced cache save if enabled
|
||||
if (this.options.enableLocalCache) {
|
||||
// Fire and forget the save operation
|
||||
this.saveCache().catch(error => {
|
||||
logger.log('error', `Failed to save IP reputation cache: ${error.message}`);
|
||||
});
|
||||
this.debouncedSaveCache();
|
||||
}
|
||||
|
||||
// Log the reputation check
|
||||
@@ -447,6 +459,21 @@ export class IPReputationChecker {
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Schedule a debounced cache save (at most once per SAVE_CACHE_DEBOUNCE_MS)
|
||||
*/
|
||||
private debouncedSaveCache(): void {
|
||||
if (this.saveCacheTimer) {
|
||||
return; // already scheduled
|
||||
}
|
||||
this.saveCacheTimer = setTimeout(() => {
|
||||
this.saveCacheTimer = null;
|
||||
this.saveCache().catch(error => {
|
||||
logger.log('error', `Failed to save IP reputation cache: ${error.message}`);
|
||||
});
|
||||
}, IPReputationChecker.SAVE_CACHE_DEBOUNCE_MS);
|
||||
}
|
||||
|
||||
/**
|
||||
* Save cache to disk or storage manager
|
||||
*/
|
||||
|
||||
@@ -83,7 +83,14 @@ export class SecurityLogger {
|
||||
}
|
||||
return SecurityLogger.instance;
|
||||
}
|
||||
|
||||
|
||||
/**
|
||||
* Reset the singleton instance (for shutdown/testing)
|
||||
*/
|
||||
public static resetInstance(): void {
|
||||
SecurityLogger.instance = undefined;
|
||||
}
|
||||
|
||||
/**
|
||||
* Log a security event
|
||||
* @param event The security event to log
|
||||
|
||||
@@ -30,6 +30,7 @@ export type StorageBackend = 'filesystem' | 'custom' | 'memory';
|
||||
* Provides unified key-value storage with multiple backend support
|
||||
*/
|
||||
export class StorageManager {
|
||||
private static readonly MAX_MEMORY_ENTRIES = 10_000;
|
||||
private backend: StorageBackend;
|
||||
private memoryStore: Map<string, string> = new Map();
|
||||
private config: IStorageConfig;
|
||||
@@ -227,6 +228,11 @@ export class StorageManager {
|
||||
|
||||
case 'memory': {
|
||||
this.memoryStore.set(key, value);
|
||||
// Evict oldest entries if memory store exceeds limit
|
||||
while (this.memoryStore.size > StorageManager.MAX_MEMORY_ENTRIES) {
|
||||
const firstKey = this.memoryStore.keys().next().value;
|
||||
this.memoryStore.delete(firstKey);
|
||||
}
|
||||
break;
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user