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

bug fix and improvement

Browse Source
This commit is contained in:
Pierre
2026-01-17 14:50:48 +01:00
parent 4642add452
commit e0fa692d73
12 changed files with 496 additions and 183 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/meta-sort-core/src/api/UnifiedAPIServer.ts
+206 -22
View File
@@ -17,7 +17,8 @@ import type { KVManager } from '../kv/KVManager.js';
import { config } from '../config/EnvConfig.js';
import type { PluginManager } from '../plugin-engine/PluginManager.js';
import type { TaskScheduler } from '../plugin-engine/TaskScheduler.js';
import type { ContainerManager, ContainerPluginScheduler, PluginCallbackPayload } from '../container-plugins/index.js';
import type { ContainerManager, ContainerPluginScheduler, PluginCallbackPayload, GateStatus } from '../container-plugins/index.js';
import type { StreamingPipeline } from '../logic/pipeline/StreamingPipeline.js';
export interface UnifiedAPIServerConfig {
/** Port to listen on (default: 3000) */
@@ -44,6 +45,7 @@ export class UnifiedAPIServer {
private getTaskScheduler: (() => TaskScheduler | null) | null = null;
private containerManager: ContainerManager | null = null;
private containerPluginScheduler: ContainerPluginScheduler | null = null;
private streamingPipeline: StreamingPipeline | null = null;
// Cache for total file size (avoid running du on every request)
private totalSizeCache: { value: number; timestamp: number } | null = null;
@@ -104,34 +106,43 @@ export class UnifiedAPIServer {
}
/**
* Get total size of watch folders using du command (cached)
* Get total size of processed files from Redis sizeByte field (cached)
* This is "eventually correct" - accumulates as files are processed by the file-info plugin
*/
private getTotalWatchFolderSize(): number {
private async getTotalFileSizeFromRedis(): Promise<number> {
// Check cache
const now = Date.now();
if (this.totalSizeCache && (now - this.totalSizeCache.timestamp) < UnifiedAPIServer.TOTAL_SIZE_CACHE_TTL_MS) {
return this.totalSizeCache.value;
}
// Calculate total size using du -sb on each watch folder
// Sum sizeByte from all file entries in Redis
let totalSize = 0;
const watchFolders = config.WATCH_FOLDER_LIST?.split(',').map(f => f.trim()).filter(f => f) || [];
for (const folder of watchFolders) {
try {
// du -sb gives total bytes for directory
const output = execSync(`du -sb "${folder}" 2>/dev/null || echo "0"`, {
encoding: 'utf8',
timeout: 10000 // 10s timeout per folder
});
const bytes = parseInt(output.split('\t')[0], 10);
if (!isNaN(bytes)) {
totalSize += bytes;
if (!this.kvClient) {
return 0;
}
try {
// Get all file hash IDs
const hashIds = await this.kvClient.getAllHashIds();
// Sum up sizeByte values (batch process for efficiency)
for (const hashId of hashIds) {
try {
const sizeByte = await this.kvClient.getMetadata(hashId, 'sizeByte');
if (sizeByte !== null && sizeByte !== undefined) {
const size = typeof sizeByte === 'number' ? sizeByte : parseInt(String(sizeByte), 10);
if (!isNaN(size) && size > 0) {
totalSize += size;
}
}
} catch {
// Skip entries without sizeByte
}
} catch (error) {
// Skip folders that fail (permission issues, doesn't exist, etc.)
console.warn(`Failed to get size for folder ${folder}:`, error);
}
} catch (error) {
console.warn('[API] Failed to calculate total size from Redis:', error);
}
// Update cache
@@ -566,6 +577,89 @@ export class UnifiedAPIServer {
}
});
// POST /api/mounts/:id/safe-unmount - Safe unmount with queue drain
// Closes gate, waits for queues to empty, then unmounts
this.app.post<{
Params: { id: string };
Querystring: { timeout?: string };
}>('/api/mounts/:id/safe-unmount', async (request, reply) => {
const timeoutMs = parseInt(request.query.timeout || '60000', 10);
try {
const mountsConfig = await readMountsConfig();
const mount = mountsConfig.mounts.find((m: any) => m.id === request.params.id);
if (!mount) {
return reply.status(404).send({ error: 'Mount not found' });
}
// Step 1: Close the gate to stop new jobs
if (this.containerPluginScheduler) {
console.log(`[Mounts] Safe unmount: closing gate for ${mount.name}`);
this.containerPluginScheduler.setGate(false);
// Step 2: Wait for queues to drain
console.log(`[Mounts] Safe unmount: waiting for queues to drain (timeout: ${timeoutMs}ms)`);
const drained = await this.containerPluginScheduler.waitForEmpty(timeoutMs);
if (!drained) {
// Re-open gate and abort
this.containerPluginScheduler.setGate(true);
const gateStatus = this.containerPluginScheduler.getGateStatus();
return reply.status(408).send({
status: 'timeout',
message: `Queue did not drain within ${timeoutMs}ms. Gate re-opened.`,
gateStatus
});
}
}
// Step 3: Request unmount
mount.desiredMounted = false;
await writeMountsConfig(mountsConfig);
// Step 4: Wait for unmount to complete
console.log(`[Mounts] Safe unmount: waiting for mount to detach: ${mount.name}`);
let unmountSuccess = false;
for (let i = 0; i < 30; i++) { // 15 seconds max wait
if (!isMounted(mount.mountPath)) {
unmountSuccess = true;
break;
}
await new Promise(r => setTimeout(r, 500));
}
// Step 5: Re-open the gate
if (this.containerPluginScheduler) {
console.log(`[Mounts] Safe unmount: re-opening gate`);
this.containerPluginScheduler.setGate(true);
}
if (!unmountSuccess) {
console.warn(`[Mounts] Safe unmount: mount still attached after timeout: ${mount.name}`);
return {
status: 'warning',
message: 'Unmount requested but mount may still be attached',
gateStatus: this.containerPluginScheduler?.getGateStatus() ?? null
};
}
console.log(`[Mounts] Safe unmount completed: ${mount.name}`);
return {
status: 'ok',
message: 'Safe unmount completed',
gateStatus: this.containerPluginScheduler?.getGateStatus() ?? null
};
} catch (error: any) {
// Re-open gate on error
if (this.containerPluginScheduler) {
this.containerPluginScheduler.setGate(true);
}
console.error(`[Mounts] Safe unmount error: ${error.message}`);
return reply.status(500).send({ error: error.message });
}
});
// DELETE /api/mounts/:id - Remove mount config
this.app.delete<{ Params: { id: string } }>('/api/mounts/:id', async (request, reply) => {
const fs = await import('fs/promises');
@@ -1341,8 +1435,8 @@ export class UnifiedAPIServer {
}
}
// Get total size of watch folders (cached, uses du command)
const totalSize = this.getTotalWatchFolderSize();
// Get total size from Redis sizeByte (eventually correct as files are processed)
const totalSize = await this.getTotalFileSizeFromRedis();
return {
prefix: 'file:',
@@ -1519,6 +1613,9 @@ export class UnifiedAPIServer {
// hashProcessing.size includes both waiting AND running files
snapshot.awaitingBackground = Math.max(0, snapshot.awaitingBackground - backgroundQueueRunning);
// Get gate status if container scheduler is available
const gateStatus = this.containerPluginScheduler?.getGateStatus();
(snapshot as any).computed = {
// Pre-process queue (validation - quick extension checks)
preProcessRunning,
@@ -1537,6 +1634,11 @@ export class UnifiedAPIServer {
backgroundQueueTotal: backgroundQueueRunning + backgroundQueuePending + (queueStatus.backgroundQueue?.size || 0),
backgroundQueuePaused: queueStatus.backgroundQueue?.isPaused || false,
// Gate control status (for safe mount/unmount operations)
gateOpen: gateStatus?.isOpen ?? true,
gateStatus: gateStatus ?? null,
pipelinePaused: this.streamingPipeline?.isPaused() ?? false,
// Legacy compatibility fields (UI may still reference these)
actualRunningHashWorkers: backgroundQueueRunning,
actualRunningLightWorkers: preProcessRunning + fastQueueRunning,
@@ -1604,6 +1706,80 @@ export class UnifiedAPIServer {
return { status: 'ok', message: `${queued} files queued for retry` };
});
// POST /api/processing/stop - Stop all processing
// Pauses pipeline queues and closes container plugin gate
// Running tasks will complete, but no new tasks will start
this.app.post('/api/processing/stop', async (request, reply) => {
// Pause the streaming pipeline (stops new files from being processed)
if (this.streamingPipeline) {
this.streamingPipeline.pause();
}
// Close the container plugin gate (stops new plugin tasks)
if (this.containerPluginScheduler) {
this.containerPluginScheduler.setGate(false);
}
const gateStatus = this.containerPluginScheduler?.getGateStatus() ?? null;
const pipelinePaused = this.streamingPipeline?.isPaused() ?? false;
console.log('[API] Processing stopped - pipeline paused, gate closed');
return {
status: 'ok',
message: 'Processing stopped - pipeline paused, gate closed',
gateStatus,
pipelinePaused
};
});
// POST /api/processing/start - Resume all processing
// Resumes pipeline queues and opens container plugin gate
this.app.post('/api/processing/start', async (request, reply) => {
// Resume the streaming pipeline
if (this.streamingPipeline) {
this.streamingPipeline.resume();
}
// Open the container plugin gate
if (this.containerPluginScheduler) {
this.containerPluginScheduler.setGate(true);
}
const gateStatus = this.containerPluginScheduler?.getGateStatus() ?? null;
const pipelinePaused = this.streamingPipeline?.isPaused() ?? false;
console.log('[API] Processing started - pipeline resumed, gate opened');
return {
status: 'ok',
message: 'Processing started - pipeline resumed, gate opened',
gateStatus,
pipelinePaused
};
});
// POST /api/processing/wait-empty - Wait for queues to drain
// Optional query param: timeout (ms, default 60000)
this.app.post<{ Querystring: { timeout?: string } }>('/api/processing/wait-empty', async (request, reply) => {
if (!this.containerPluginScheduler) {
return reply.status(503).send({
status: 'error',
message: 'Container plugin scheduler not available'
});
}
const timeoutMs = parseInt(request.query.timeout || '60000', 10);
const success = await this.containerPluginScheduler.waitForEmpty(timeoutMs);
const gateStatus = this.containerPluginScheduler.getGateStatus();
return {
status: success ? 'ok' : 'timeout',
message: success ? 'Queues are empty' : `Timeout after ${timeoutMs}ms`,
gateStatus
};
});
}
@@ -1639,8 +1815,8 @@ export class UnifiedAPIServer {
}
}
// Get total size of watch folders (cached, uses du command)
const totalSize = this.getTotalWatchFolderSize();
// Get total size from Redis sizeByte (eventually correct as files are processed)
const totalSize = await this.getTotalFileSizeFromRedis();
return {
fileCount: hashIds.length,
@@ -2483,6 +2659,14 @@ export class UnifiedAPIServer {
console.log('[UnifiedAPIServer] Container plugin manager and scheduler set');
}
/**
* Set streaming pipeline (for pause/resume control)
*/
public setStreamingPipeline(pipeline: StreamingPipeline): void {
this.streamingPipeline = pipeline;
console.log('[UnifiedAPIServer] Streaming pipeline set');
}
/**
* Start the HTTP server
*/
packages/meta-sort-core/src/config/EnvConfig.ts
+5 -17
View File
@@ -113,21 +113,11 @@ interface EnvConfig {
CONTAINER_NETWORK: string;
/**
* Plugin file mounts - maps subpaths to host paths for /files in plugin containers.
* Format: "subpath:host_path,subpath2:host_path2"
* Example: "watch:/data/media,test:/app/test/media"
* Results in /files/watch and /files/test mounted in plugin containers (read-only).
* WebDAV URL for plugin file access.
* Plugins access /files via this WebDAV endpoint served by meta-sort nginx.
* Example: "http://meta-sort-dev/webdav"
*/
PLUGIN_FILE_MOUNTS?: string;
/** Host path for plugin cache folder - each plugin gets its own subdirectory at /cache (optional, dev only) */
PLUGIN_CACHE_FOLDER?: string;
/**
* Host path for plugin output folders - each plugin gets /files/plugin/<id> mount (READ-WRITE)
* Plugins decide whether to use this folder - we always mount it if this env is set
*/
PLUGIN_OUTPUT_FOLDER?: string;
PLUGIN_WEBDAV_URL?: string;
/** Docker Compose project name for grouping containers in Docker Desktop (optional). When set, plugin containers appear grouped as a stack. */
PLUGIN_STACK_NAME?: string;
@@ -202,9 +192,7 @@ export const config: EnvConfig = {
CONTAINER_CALLBACK_URL: process.env.CONTAINER_CALLBACK_URL || 'http://meta-sort:8180',
CONTAINER_META_CORE_URL: process.env.CONTAINER_META_CORE_URL || 'http://meta-sort:9000',
CONTAINER_NETWORK: process.env.CONTAINER_NETWORK || 'meta-network',
PLUGIN_FILE_MOUNTS: process.env.PLUGIN_FILE_MOUNTS,
PLUGIN_CACHE_FOLDER: process.env.PLUGIN_CACHE_FOLDER,
PLUGIN_OUTPUT_FOLDER: process.env.PLUGIN_OUTPUT_FOLDER,
PLUGIN_WEBDAV_URL: process.env.PLUGIN_WEBDAV_URL,
PLUGIN_STACK_NAME: process.env.PLUGIN_STACK_NAME,
};
packages/meta-sort-core/src/container-plugins/ContainerManager.ts
+38 -109
View File
@@ -49,62 +49,29 @@ function getContainerName(pluginId: string, instanceIndex: number): string {
*/
/**
* ============================================================================
* PLUGIN MOUNT ARCHITECTURE - HARDCODED PATHS
* PLUGIN FILE ACCESS ARCHITECTURE - WebDAV
* ============================================================================
*
* Each plugin container has exactly 3 mount points (hardcoded):
* Plugins access the /files directory via WebDAV served by meta-sort nginx:
*
* /files (READ-ONLY) - Files to process, same structure as meta-sort
* /output (READ-WRITE) - Plugin-generated files
* /cache (READ-WRITE) - Plugin-specific cache
* /webdav (WebDAV endpoint) - Exposes /files with caching
* /files/watch - Watch folder (media files)
* /files/test - Test media folder
* /files/plugin/ - Plugin output (each plugin writes to /files/plugin/<name>/)
* /files/corn - SMB mounts (mounted by rclone inside meta-sort)
*
* Host paths are configured via environment variables:
* - PLUGIN_FILE_MOUNTS: Maps subpaths under /files to host paths
* Format: "subpath:host_path,subpath2:host_path2"
* Example: "watch:/data/media,test:/app/test/media"
* Results in: /files/watch, /files/test mounted in container
* WebDAV configuration:
* - PLUGIN_WEBDAV_URL: WebDAV endpoint URL (e.g., "http://meta-sort-dev/webdav")
*
* - PLUGIN_CACHE_FOLDER: Base path for /cache mounts
* - PLUGIN_OUTPUT_FOLDER: Base path for /output mounts
*
* SECURITY: Plugins have read-only access to /files.
* Only /cache and /output are writable.
* Benefits:
* - Works with Docker's overlay filesystem (no NFS file handle issues)
* - nginx caching for multiple plugins accessing same file
* - Range requests for streaming large files
* - Bidirectional (read/write) via standard HTTP methods
*
* ============================================================================
*/
/**
* File mount configuration - maps container path to host path
*/
interface FileMountConfig {
containerPath: string;
hostPath: string;
}
/**
* Parse PLUGIN_FILE_MOUNTS env var into mount configs
* Format: "container_path:host_path,container_path2:host_path2"
*/
function parseFileMounts(mountsStr?: string): FileMountConfig[] {
if (!mountsStr) {
return [];
}
const mounts: FileMountConfig[] = [];
const pairs = mountsStr.split(',').map(s => s.trim()).filter(s => s);
for (const pair of pairs) {
const [containerPath, hostPath] = pair.split(':').map(s => s.trim());
if (containerPath && hostPath) {
mounts.push({ containerPath, hostPath });
} else {
console.warn(`[ContainerManager] Invalid mount format: "${pair}" - expected "container_path:host_path"`);
}
}
return mounts;
}
export class ContainerManager extends EventEmitter {
private dockerClient: DockerClient;
private pluginsConfig: ContainerPluginsConfig | null = null;
@@ -113,9 +80,7 @@ export class ContainerManager extends EventEmitter {
private roundRobinCounters: Map<string, number> = new Map();
private initialized: boolean = false;
private shutdownInProgress: boolean = false;
private fileMounts: FileMountConfig[] = [];
private pluginCacheFolder?: string;
private pluginOutputFolder?: string;
private webdavUrl?: string;
private stackName?: string;
constructor(
@@ -129,14 +94,8 @@ export class ContainerManager extends EventEmitter {
super();
this.dockerClient = dockerClient || new DockerClient(config.DOCKER_SOCKET_PATH);
// Parse file mounts from PLUGIN_FILE_MOUNTS env var
this.fileMounts = parseFileMounts(config.PLUGIN_FILE_MOUNTS);
// Configure plugin cache folder (READ-WRITE)
this.pluginCacheFolder = config.PLUGIN_CACHE_FOLDER;
// Configure plugin output folder (READ-WRITE for /plugin-files)
this.pluginOutputFolder = config.PLUGIN_OUTPUT_FOLDER;
// WebDAV configuration for plugin file access
this.webdavUrl = config.PLUGIN_WEBDAV_URL;
// Configure stack name for Docker Desktop grouping
this.stackName = config.PLUGIN_STACK_NAME;
@@ -152,21 +111,18 @@ export class ContainerManager extends EventEmitter {
console.log('[ContainerManager] Initializing...');
// Log configured host mounts
console.log('[ContainerManager] Plugin mount configuration (hardcoded paths):');
console.log(' /files (RO) - Files to process:');
if (this.fileMounts.length > 0) {
for (const mount of this.fileMounts) {
const targetPath = mount.containerPath.startsWith('/files')
? mount.containerPath
: `/files/${mount.containerPath}`;
console.log(` ${targetPath} -> ${mount.hostPath}`);
}
// Log WebDAV configuration
console.log('[ContainerManager] Plugin file access configuration (WebDAV):');
if (this.webdavUrl) {
console.log(` WebDAV URL: ${this.webdavUrl}`);
console.log(' /files/watch - Watch folder (media files)');
console.log(' /files/test - Test media folder');
console.log(' /files/plugin/ - Plugin output (per-plugin subdirs)');
console.log(' /files/corn - SMB mounts (visible via WebDAV)');
} else {
console.warn(' (not configured) - Set PLUGIN_FILE_MOUNTS');
console.warn(' WebDAV not configured. Set PLUGIN_WEBDAV_URL.');
console.warn(' Plugins will not have access to files until WebDAV is configured.');
}
console.log(` /cache (RW) - Plugin cache: ${this.pluginCacheFolder || '(not configured)'}`);
console.log(` /output (RW) - Plugin output: ${this.pluginOutputFolder || '(not configured)'}`)
// Initialize Docker client
await this.dockerClient.initialize();
@@ -336,47 +292,18 @@ export class ContainerManager extends EventEmitter {
}
/**
* PLUGIN MOUNT ARCHITECTURE - HARDCODED PATHS:
* /files (READ-ONLY) - Files to process
* /output (READ-WRITE) - Plugin-generated files
* /cache (READ-WRITE) - Plugin-specific cache
* PLUGIN FILE ACCESS ARCHITECTURE - WebDAV:
* Plugins access files via WebDAV URL (no container mounts needed)
* Plugin output goes to /files/plugin/<plugin-id>/ via WebDAV PUT
* Sees all mounts including SMB via rclone
*
* Note: No volume mounts for /files - plugins use HTTP/WebDAV client
*/
const mounts: Array<{ source: string; target: string; readonly: boolean }> = [];
// /files/* (READ-ONLY) - Mount each configured subpath under /files
for (const fileMount of this.fileMounts) {
// fileMount.containerPath is the subpath (e.g., "watch", "test")
// Mount at /files/{subpath}
const targetPath = fileMount.containerPath.startsWith('/files')
? fileMount.containerPath
: `/files/${fileMount.containerPath}`;
mounts.push({
source: fileMount.hostPath,
target: targetPath,
readonly: true,
});
}
if (this.fileMounts.length === 0) {
console.warn(`[ContainerManager] WARNING: No file mounts configured. Plugin '${pluginId}' won't have access to files.`);
}
// /cache (READ-WRITE) - Plugin-specific cache
if (this.pluginCacheFolder) {
mounts.push({
source: `${this.pluginCacheFolder}/${pluginId}`,
target: '/cache',
readonly: false,
});
}
// /output (READ-WRITE) - Plugin-generated files
if (this.pluginOutputFolder) {
mounts.push({
source: `${this.pluginOutputFolder}/${pluginId}`,
target: '/output',
readonly: false,
});
// No file mounts needed - plugins access files via WebDAV
if (!this.webdavUrl) {
console.warn(`[ContainerManager] WARNING: WebDAV not configured. Plugin '${pluginId}' won't have access to files. Set PLUGIN_WEBDAV_URL.`);
}
// Create container
@@ -395,6 +322,8 @@ export class ContainerManager extends EventEmitter {
PLUGIN_ID: pluginId,
META_CORE_URL: this.metaCoreUrl,
CALLBACK_URL: this.callbackUrl,
WEBDAV_URL: this.webdavUrl || '',
FILES_PATH: '/files', // Virtual path - files accessed via WebDAV
},
// Docker Desktop grouping
stackName: this.stackName,
packages/meta-sort-core/src/container-plugins/ContainerPluginScheduler.ts
+89 -1
View File
@@ -43,6 +43,17 @@ interface PendingCallback {
timeoutHandle: NodeJS.Timeout;
}
/**
* Gate status for queue control
*/
export interface GateStatus {
isOpen: boolean;
fastPending: number;
fastRunning: number;
backgroundPending: number;
backgroundRunning: number;
}
/**
* Container Plugin Scheduler
*/
@@ -59,6 +70,7 @@ export class ContainerPluginScheduler extends EventEmitter {
private metaCoreUrl: string;
private defaultTimeout: number = 60000; // 60 seconds
private kvClient: IKVClient | null = null;
private isAcceptingNewTasks: boolean = true; // Gate control for safe mount/unmount
constructor(
containerManager: ContainerManager,
@@ -231,13 +243,29 @@ export class ContainerPluginScheduler extends EventEmitter {
/**
* Enqueue a task for execution
* Returns false if the gate is closed and the task was rejected
*/
enqueueTask(task: ContainerTask): void {
enqueueTask(task: ContainerTask): boolean {
if (!this.isAcceptingNewTasks) {
console.log(`[Gate] Task ${task.id} rejected - gate closed (plugin: ${task.pluginId}, file: ${task.filePath})`);
// Remove from tracking since it won't be executed
this.pendingTasks.delete(task.id);
const fileTasks = this.fileTasks.get(task.fileHash);
if (fileTasks) {
fileTasks.delete(task.id);
if (fileTasks.size === 0) {
this.fileTasks.delete(task.fileHash);
}
}
return false;
}
const queue = task.queue === 'background' ? this.backgroundQueue : this.fastQueue;
queue.add(async () => {
await this.executeTask(task);
});
return true;
}
/**
@@ -523,6 +551,66 @@ export class ContainerPluginScheduler extends EventEmitter {
};
}
/**
* Set the gate state (open/closed)
* When closed, new tasks will be rejected but existing tasks will complete
*/
setGate(accepting: boolean): void {
const wasAccepting = this.isAcceptingNewTasks;
this.isAcceptingNewTasks = accepting;
if (wasAccepting !== accepting) {
console.log(`[Gate] Gate ${accepting ? 'opened' : 'closed'} - ${accepting ? 'accepting' : 'rejecting'} new tasks`);
}
}
/**
* Check if the gate is open (accepting new tasks)
*/
isGateOpen(): boolean {
return this.isAcceptingNewTasks;
}
/**
* Get detailed gate status including queue counts
*/
getGateStatus(): GateStatus {
const queueStatus = this.getQueueStatus();
return {
isOpen: this.isAcceptingNewTasks,
fastPending: queueStatus.fast.pending,
fastRunning: queueStatus.fast.running,
backgroundPending: queueStatus.background.pending,
backgroundRunning: queueStatus.background.running,
};
}
/**
* Wait for both queues to drain (all tasks completed)
* Useful for safe unmount operations
* @param timeoutMs Maximum time to wait (default: 60000ms)
* @returns true if queues are empty, false if timeout
*/
async waitForEmpty(timeoutMs: number = 60000): Promise<boolean> {
const start = Date.now();
const pollInterval = 500;
while (Date.now() - start < timeoutMs) {
const status = this.getGateStatus();
const totalActive = status.fastPending + status.fastRunning +
status.backgroundPending + status.backgroundRunning;
if (totalActive === 0) {
console.log(`[Gate] Queues drained after ${Date.now() - start}ms`);
return true;
}
await new Promise(r => setTimeout(r, pollInterval));
}
console.log(`[Gate] Timeout waiting for queues to drain (${timeoutMs}ms)`);
return false;
}
/**
* Get pending tasks for a file
*/
packages/meta-sort-core/src/container-plugins/DockerClient.ts
+7 -3
View File
@@ -103,6 +103,12 @@ export class DockerClient {
}
}
// Build Binds array for bind mounts
const binds = options.mounts?.map(m => {
const opts = [m.readonly ? 'ro' : 'rw'];
return `${m.source}:${m.target}:${opts.join(',')}`;
}) || [];
const createOptions: ContainerCreateOptions = {
Image: options.image,
name: options.name,
@@ -111,9 +117,7 @@ export class DockerClient {
: undefined,
Labels: labels,
HostConfig: {
Binds: options.mounts?.map(m =>
`${m.source}:${m.target}:${m.readonly ? 'ro' : 'rw'}`
),
Binds: binds.length > 0 ? binds : undefined,
NetworkMode: options.network,
Memory: options.resources?.memory
? parseMemoryLimit(options.resources.memory)
packages/meta-sort-core/src/container-plugins/index.ts
+1
View File
@@ -31,4 +31,5 @@ export {
ContainerPluginScheduler,
createContainerPluginScheduler,
type ContainerSchedulerEvents,
type GateStatus,
} from './ContainerPluginScheduler.js';
packages/meta-sort-core/src/container-plugins/types.ts
+1 -1
View File
@@ -283,7 +283,7 @@ export interface ContainerSpawnOptions {
env?: Record<string, string>;
/** Resource limits */
resources?: ContainerResourceLimits;
/** Volume mounts */
/** Volume mounts (bind mounts only - files accessed via WebDAV) */
mounts?: Array<{
source: string;
target: string;
packages/meta-sort-core/src/index.ts
+4
View File
@@ -258,6 +258,10 @@ process.on('SIGINT', async () => {
console.log('[Startup] ContainerPluginScheduler connected to StreamingPipeline');
}
// Connect StreamingPipeline to API server (for pause/resume control)
apiServer.setStreamingPipeline(pipeline);
console.log('[Startup] StreamingPipeline connected to API server');
// First, rebuild VFS from KV to restore state from previous runs
console.log('[Startup] Rebuilding VirtualFileSystem from KV...');
await fileProcessor.rebuildVFSFromKV();
packages/meta-sort-core/src/logic/pipeline/StreamingPipeline.ts
+30
View File
@@ -432,6 +432,36 @@ export class StreamingPipeline {
};
}
/**
* Pause all pipeline queues
* Running tasks will complete, but no new tasks will start
*/
pause(): void {
this.validationQueue.pause();
this.fastQueue.pause();
this.backgroundQueue.pause();
console.log('[Pipeline] All queues paused');
}
/**
* Resume all pipeline queues
*/
resume(): void {
this.validationQueue.start();
this.fastQueue.start();
this.backgroundQueue.start();
console.log('[Pipeline] All queues resumed');
}
/**
* Check if pipeline is paused
*/
isPaused(): boolean {
return this.validationQueue.isPaused ||
this.fastQueue.isPaused ||
this.backgroundQueue.isPaused;
}
/**
* Log current pipeline progress
*/