metastar/meta-sort
4
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

background queue fix

Browse Source
This commit is contained in:
Pierre
2026-01-24 16:20:58 +01:00
parent 133e6bc61c
commit 53c4463084
6 changed files with 170 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/meta-sort-core/src/container-plugins/ContainerManager.ts
+92 -5
View File
@@ -7,6 +7,7 @@
import { EventEmitter } from 'events';
import { promises as fs } from 'fs';
import { join } from 'path';
import { DockerClient } from './DockerClient.js';
import {
loadConfig,
@@ -25,6 +26,15 @@ import type {
} from './types.js';
import { config } from '../config/EnvConfig.js';
/**
* Service info structure from meta-core service discovery
*/
interface ServiceInfo {
name: string;
api: string;
endpoints: Record<string, string>;
}
/**
* Container manager events
*/
@@ -84,25 +94,78 @@ export class ContainerManager extends EventEmitter {
private shutdownInProgress: boolean = false;
private webdavUrl?: string;
private stackName?: string;
private callbackUrl: string;
private metaCoreUrl: string;
constructor(
private configPath: string = config.CONTAINER_PLUGINS_CONFIG,
private network: string = config.CONTAINER_NETWORK,
private filesPath: string = config.FILES_PATH,
private callbackUrl: string = config.CONTAINER_CALLBACK_URL,
private metaCoreUrl: string = config.CONTAINER_META_CORE_URL,
callbackUrl?: string,
metaCoreUrl?: string,
dockerClient?: DockerClient
) {
super();
this.dockerClient = dockerClient || new DockerClient(config.DOCKER_SOCKET_PATH);
// WebDAV configuration for plugin file access
// Initialize with env vars as fallback (will be overridden by service discovery)
this.callbackUrl = callbackUrl ?? config.CONTAINER_CALLBACK_URL;
this.metaCoreUrl = metaCoreUrl ?? config.CONTAINER_META_CORE_URL;
this.webdavUrl = config.PLUGIN_WEBDAV_URL;
// Configure stack name for Docker Desktop grouping
this.stackName = config.PLUGIN_STACK_NAME;
}
/**
* Discover service URLs from meta-core service discovery file
* Falls back to environment variables if discovery fails
*/
private async discoverServiceUrls(): Promise<void> {
const serviceFile = join(config.META_CORE_PATH, 'services', 'meta-sort.json');
try {
const content = await fs.readFile(serviceFile, 'utf-8');
const serviceInfo: ServiceInfo = JSON.parse(content);
// Extract URLs from service info
const api = serviceInfo.api;
const endpoints = serviceInfo.endpoints || {};
// Update URLs from service discovery
if (endpoints.callback) {
this.callbackUrl = endpoints.callback;
console.log(`[ContainerManager] Discovered callback URL: ${this.callbackUrl}`);
} else if (api) {
// Fallback: construct from api base
this.callbackUrl = `${api}/api/plugins/callback`;
console.log(`[ContainerManager] Constructed callback URL: ${this.callbackUrl}`);
}
if (endpoints.webdav) {
this.webdavUrl = endpoints.webdav;
console.log(`[ContainerManager] Discovered WebDAV URL: ${this.webdavUrl}`);
} else if (api) {
// Fallback: construct from api base
this.webdavUrl = `${api}/webdav`;
console.log(`[ContainerManager] Constructed WebDAV URL: ${this.webdavUrl}`);
}
if (endpoints.health) {
// Extract meta-core URL from health endpoint
const healthUrl = new URL(endpoints.health);
this.metaCoreUrl = `${healthUrl.protocol}//${healthUrl.host}`;
console.log(`[ContainerManager] Discovered meta-core URL: ${this.metaCoreUrl}`);
}
console.log('[ContainerManager] Service discovery successful');
} catch (error) {
console.warn(`[ContainerManager] Service discovery failed, using env vars: ${error}`);
console.log(`[ContainerManager] Using fallback callback URL: ${this.callbackUrl}`);
console.log(`[ContainerManager] Using fallback WebDAV URL: ${this.webdavUrl || 'not set'}`);
}
}
/**
* Initialize the container manager
*/
@@ -113,6 +176,9 @@ export class ContainerManager extends EventEmitter {
console.log('[ContainerManager] Initializing...');
// Discover service URLs from meta-core service discovery
await this.discoverServiceUrls();
// Log WebDAV configuration
console.log('[ContainerManager] Plugin file access configuration (WebDAV):');
if (this.webdavUrl) {
@@ -122,8 +188,8 @@ export class ContainerManager extends EventEmitter {
console.log(' /files/plugin/ - Plugin output (per-plugin subdirs)');
console.log(' /files/corn - SMB mounts (visible via WebDAV)');
} else {
console.warn(' WebDAV not configured. Set PLUGIN_WEBDAV_URL.');
console.warn(' Plugins will not have access to files until WebDAV is configured.');
console.warn(' WebDAV not configured.');
console.warn(' Plugins will not have access to files.');
}
// Initialize Docker client
@@ -914,6 +980,27 @@ export class ContainerManager extends EventEmitter {
getPluginsConfig(): ContainerPluginsConfig | null {
return this.pluginsConfig;
}
/**
* Get the discovered callback URL for plugins
*/
getCallbackUrl(): string {
return this.callbackUrl;
}
/**
* Get the discovered meta-core URL
*/
getMetaCoreUrl(): string {
return this.metaCoreUrl;
}
/**
* Get the discovered WebDAV URL for file access
*/
getWebdavUrl(): string | undefined {
return this.webdavUrl;
}
}
/**
packages/meta-sort-core/src/container-plugins/ContainerPluginScheduler.ts
+61 -2
View File
@@ -76,6 +76,7 @@ export class ContainerPluginScheduler extends EventEmitter {
containerManager: ContainerManager,
options?: {
fastConcurrency?: number;
backgroundConcurrency?: number;
callbackUrl?: string;
metaCoreUrl?: string;
defaultTimeout?: number;
@@ -90,9 +91,9 @@ export class ContainerPluginScheduler extends EventEmitter {
});
// Background queue starts paused - only runs when fast queue is idle
// Concurrency is unlimited - actual limit is determined by plugin instances
// Default to same concurrency as fast queue if not specified
this.backgroundQueue = new PQueue({
concurrency: Infinity,
concurrency: options?.backgroundConcurrency ?? config.FAST_QUEUE_CONCURRENCY,
autoStart: false,
});
@@ -101,12 +102,32 @@ export class ContainerPluginScheduler extends EventEmitter {
this.onFastQueueIdle();
});
// Initialize with env vars as fallback (will be updated from ContainerManager after discovery)
this.callbackUrl = options?.callbackUrl ?? `${config.CONTAINER_CALLBACK_URL}/api/plugins/callback`;
this.metaCoreUrl = options?.metaCoreUrl ?? config.CONTAINER_META_CORE_URL;
this.defaultTimeout = options?.defaultTimeout ?? 60000;
this.kvClient = options?.kvClient ?? null;
}
/**
* Update URLs from ContainerManager after service discovery
* Call this after ContainerManager.initialize() completes
*/
updateUrlsFromManager(): void {
const managerCallbackUrl = this.containerManager.getCallbackUrl();
const managerMetaCoreUrl = this.containerManager.getMetaCoreUrl();
if (managerCallbackUrl) {
this.callbackUrl = managerCallbackUrl;
console.log(`[ContainerPluginScheduler] Updated callback URL: ${this.callbackUrl}`);
}
if (managerMetaCoreUrl) {
this.metaCoreUrl = managerMetaCoreUrl;
console.log(`[ContainerPluginScheduler] Updated meta-core URL: ${this.metaCoreUrl}`);
}
}
/**
* Wait for dependency plugins to complete for a file
*/
@@ -572,16 +593,47 @@ export class ContainerPluginScheduler extends EventEmitter {
background: { pending: number; running: number; waiting: number };
pendingTasks: number;
pendingCallbacks: number;
// File-level counts (what users care about)
files: {
fastRunning: number; // unique files with fast tasks running
fastWaiting: number; // unique files with fast tasks waiting
backgroundRunning: number; // unique files with background tasks running
backgroundWaiting: number; // unique files with background tasks waiting
};
} {
// Count pending callbacks by queue type
// These are tasks dispatched to containers awaiting completion - the real "running" count
let fastCallbacks = 0;
let backgroundCallbacks = 0;
const fastRunningFiles = new Set<string>();
const backgroundRunningFiles = new Set<string>();
for (const callback of this.pendingCallbacks.values()) {
if (callback.task.queue === 'fast') {
fastCallbacks++;
fastRunningFiles.add(callback.task.fileHash);
} else {
backgroundCallbacks++;
backgroundRunningFiles.add(callback.task.fileHash);
}
}
// Count unique files with waiting tasks
const fastWaitingFiles = new Set<string>();
const backgroundWaitingFiles = new Set<string>();
for (const task of this.pendingTasks.values()) {
if (task.status === 'pending') {
if (task.queue === 'fast') {
// Only count as waiting if not already running
if (!fastRunningFiles.has(task.fileHash)) {
fastWaitingFiles.add(task.fileHash);
}
} else {
if (!backgroundRunningFiles.has(task.fileHash)) {
backgroundWaitingFiles.add(task.fileHash);
}
}
}
}
@@ -600,6 +652,12 @@ export class ContainerPluginScheduler extends EventEmitter {
},
pendingTasks: this.pendingTasks.size,
pendingCallbacks: this.pendingCallbacks.size,
files: {
fastRunning: fastRunningFiles.size,
fastWaiting: fastWaitingFiles.size,
backgroundRunning: backgroundRunningFiles.size,
backgroundWaiting: backgroundWaitingFiles.size,
},
};
}
@@ -823,6 +881,7 @@ export function createContainerPluginScheduler(
containerManager: ContainerManager,
options?: {
fastConcurrency?: number;
backgroundConcurrency?: number;
callbackUrl?: string;
metaCoreUrl?: string;
kvClient?: IKVClient;
packages/meta-sort-core/src/index.ts
+5 -1
View File
@@ -205,7 +205,7 @@ process.on('SIGINT', async () => {
port: config.FUSE_API_PORT,
host: config.FUSE_API_HOST,
enableCors: true,
}, unifiedStateManager, getDuplicateResult, kvClient || undefined, triggerScan, undefined, fastQueueConcurrency, () => fileProcessor.getQueueStatus(), kvManager || undefined, getPluginManagerInstance, () => fileProcessor.getTaskScheduler());
}, unifiedStateManager, getDuplicateResult, kvClient || undefined, triggerScan, backgroundQueueConcurrency, fastQueueConcurrency, () => fileProcessor.getQueueStatus(), kvManager || undefined, getPluginManagerInstance, () => fileProcessor.getTaskScheduler());
await apiServer.start();
console.log('Unified API Server started successfully');
@@ -229,9 +229,13 @@ process.on('SIGINT', async () => {
// Create container plugin scheduler
containerPluginScheduler = new ContainerPluginScheduler(containerManager, {
fastConcurrency: config.FAST_QUEUE_CONCURRENCY,
backgroundConcurrency: backgroundQueueConcurrency,
kvClient: kvClient || undefined,
});
// Update scheduler URLs from service discovery
containerPluginScheduler.updateUrlsFromManager();
// Connect to API server (for callback handling)
apiServer.setContainerPluginManager(containerManager, containerPluginScheduler);
console.log('[Startup] ContainerPluginScheduler connected to API server');
packages/meta-sort-core/src/kv/KVManager.ts
+6 -2
View File
@@ -135,7 +135,10 @@ export class KVManager {
console.log(`[KVManager] Using direct Redis URL: ${this.config.redisUrl}`);
this.kvClient = await this.createRedisClient(this.config.redisUrl);
// Initialize service discovery even with direct Redis URL
// NOTE: Service discovery is handled by Go meta-core sidecar
// The Go binary writes to /meta-core/services/{serviceName}.json with correct internal IPs
// TypeScript ServiceDiscovery would overwrite with external BASE_URL, breaking internal discovery
// Keep ServiceDiscovery instance for reading/discovery only, but don't call start()
const apiUrl = this.config.baseUrl || `http://localhost:${this.config.apiPort}`;
this.serviceDiscovery = new ServiceDiscovery({
metaCorePath: this.config.metaCorePath,
@@ -151,7 +154,8 @@ export class KVManager {
processing: '/api/processing/status'
}
});
await this.serviceDiscovery.start();
// Don't call start() - Go meta-core handles registration
// await this.serviceDiscovery.start();
this.isStarted = true;
this.notifyReady();
packages/meta-sort-core/src/logic/WatchedFileProcessor.ts
+3 -1
View File
@@ -147,7 +147,9 @@ export class WatchedFileProcessor implements FileProcessorInterface {
pending: queues.backgroundQueue.size, // waiting in queue
size: queues.backgroundQueue.size,
isPaused: false
}
},
// File-level counts (what users care about)
files: queues.files
};
return result;
}
packages/meta-sort-core/src/logic/pipeline/StreamingPipeline.ts
+3 -1
View File
@@ -395,7 +395,9 @@ export class StreamingPipeline {
size: this.validationQueue.size
},
fastQueue,
backgroundQueue
backgroundQueue,
// File-level counts from container scheduler (what users care about)
files: containerQueueStatus?.files
},
state: this.config.stateManager.getSnapshot()
};