import { api } from "../../scripts/api.js"; import { applyProbeResultToWorkerDot, findNodesByClass } from './workerUtils.js'; import { TIMEOUTS, NODE_CLASSES, generateUUID } from './constants.js'; import { markSkipDispatch, selectLeastBusyWorker } from './executionDecisionUtils.js'; import { checkAllWorkerStatuses, getWorkerUrl } from './workerLifecycle.js'; export { markSkipDispatch, selectLeastBusyWorker }; export function setupInterceptor(extension) { api.queuePrompt = async (number, prompt) => { if (extension.isEnabled) { const hasCollector = findNodesByClass(prompt.output, NODE_CLASSES.DISTRIBUTED_COLLECTOR).length > 0; const hasDistUpscale = findNodesByClass(prompt.output, NODE_CLASSES.UPSCALE_DISTRIBUTED).length > 0; const hasDistQueue = findNodesByClass(prompt.output, NODE_CLASSES.DISTRIBUTED_QUEUE).length > 0; if (hasCollector || hasDistUpscale) { const result = await executeParallelDistributed(extension, prompt); // Immediate status check for instant feedback checkAllWorkerStatuses(extension); // Another check after a short delay to catch state changes setTimeout(() => checkAllWorkerStatuses(extension), TIMEOUTS.POST_ACTION_DELAY); return result; } // DistributedQueue: route entire workflow to least-busy worker if (hasDistQueue) { const result = await executeQueueDistributed(extension, prompt); checkAllWorkerStatuses(extension); setTimeout(() => checkAllWorkerStatuses(extension), TIMEOUTS.POST_ACTION_DELAY); return result; } } return extension.originalQueuePrompt(number, prompt); }; } export async function executeParallelDistributed(extension, promptWrapper) { const traceExecutionId = `exec_${Date.now()}_${generateUUID().slice(0, 6)}`; try { const enabledWorkers = extension.enabledWorkers; extension.log(`[exec:${traceExecutionId}] Starting distributed execution`, "debug"); // Pre-flight health check on all enabled workers const activeWorkers = await performPreflightCheck(extension, enabledWorkers); // Case: Enabled workers but all offline if (activeWorkers.length === 0 && enabledWorkers.length > 0) { extension.log("No active workers found. All enabled workers are offline."); if (extension.ui?.showToast) { extension.ui.showToast(extension.app, "error", "All Workers Offline", `${enabledWorkers.length} worker(s) enabled but all are offline or unreachable. Check worker connections and try again.`, 5000); } // Fall back to master-only execution return extension.originalQueuePrompt(0, promptWrapper); } extension.log(`Pre-flight check: ${activeWorkers.length} of ${enabledWorkers.length} workers are active`, "debug"); // Check if master host might be unreachable by workers (cloudflare tunnel down) const masterHost = extension.config?.master?.host || ''; const isCloudflareHost = /\.(trycloudflare\.com|cloudflare\.dev)$/i.test(masterHost); if (isCloudflareHost && activeWorkers.length > 0) { // Try to verify if the cloudflare tunnel is actually up try { const testUrl = `${window.location.protocol}//${masterHost}/prompt`; const response = await fetch(testUrl, { method: 'GET', mode: 'cors', cache: 'no-cache', signal: AbortSignal.timeout(3000) // 3 second timeout }); if (!response.ok) { throw new Error('Master not reachable'); } } catch (error) { // Cloudflare tunnel appears to be down extension.log(`Master host ${masterHost} is not reachable - cloudflare tunnel may be down`, "error"); if (extension.ui?.showCloudflareWarning) { extension.ui.showCloudflareWarning(extension, masterHost); } // Stop execution - workers won't be able to send results back extension.log("Blocking execution - workers cannot reach master at cloudflare domain", "error"); return null; // This will prevent the workflow from running } } const queueResponse = await extension.api.queueDistributed({ prompt: promptWrapper.output, workflow: promptWrapper.workflow, enabled_worker_ids: activeWorkers.map((worker) => worker.id), workers: activeWorkers.map((worker) => ({ id: worker.id })), client_id: api.clientId, delegate_master: Boolean(extension.config?.settings?.master_delegate_only), auto_prepare: true, trace_execution_id: traceExecutionId, }); if (queueResponse?.prompt_id) { extension.log( `[exec:${traceExecutionId}] Distributed queue accepted by backend (prompt_id=${queueResponse.prompt_id}, workers=${queueResponse.worker_count ?? activeWorkers.length})`, "debug" ); return queueResponse; } throw new Error( `[exec:${traceExecutionId}] Backend did not return a prompt_id for distributed queue.` ); } catch (error) { extension.log(`[exec:${traceExecutionId}] Distributed execution failed: ${error.message}`, "error"); if (extension.ui?.showToast) { extension.ui.showToast(extension.app, "error", "Distributed Failed", error.message, 5000); } return null; } } /** * Execute a workflow via DistributedQueue - routes to the least-busy worker. * The workflow runs ONLY on the selected worker, not on the master. */ export async function executeQueueDistributed(extension, promptWrapper) { try { const enabledWorkers = extension.enabledWorkers; if (enabledWorkers.length === 0) { extension.log("DistributedQueue: No enabled workers. Falling back to local execution.", "warn"); return extension.originalQueuePrompt(0, promptWrapper); } // Fetch queue status from all enabled workers const workerStatuses = await fetchWorkerQueueStatuses(extension, enabledWorkers); if (workerStatuses.length === 0) { extension.log("DistributedQueue: No reachable workers. Falling back to local execution.", "warn"); return extension.originalQueuePrompt(0, promptWrapper); } // Select the least-busy worker const selectedWorker = selectLeastBusyWorker(extension, workerStatuses); const worker = selectedWorker.worker; const queueRemaining = selectedWorker.queueRemaining; extension.log(`DistributedQueue: Routing to ${worker.name} (queue_remaining=${queueRemaining})`, "info"); // Prepare the prompt with skip_dispatch=True for DistributedQueue nodes const promptToSend = JSON.parse(JSON.stringify(promptWrapper.output)); markSkipDispatch(promptToSend); // Dispatch to the selected worker const workerUrl = getWorkerUrl(extension, worker); try { const result = await extension.api.dispatchToWorker(workerUrl, { prompt: promptToSend, extra_data: { extra_pnginfo: { workflow: promptWrapper.workflow } }, client_id: api.clientId }); extension.log(`DistributedQueue: Dispatched to ${worker.name}, prompt_id=${result.prompt_id}`, "info"); return { prompt_id: result.prompt_id, worker_id: worker.id }; } catch (error) { extension.log(`DistributedQueue: Failed to dispatch to ${worker.name}: ${error.message}`, "error"); // Fall back to local execution return extension.originalQueuePrompt(0, promptWrapper); } } catch (error) { extension.log("DistributedQueue execution failed: " + error.message, "error"); throw error; } } async function fetchWorkerQueueStatuses(extension, workers) { const statuses = []; const checkPromises = workers.map(async (worker) => { const workerUrl = getWorkerUrl(extension, worker); try { const probeResult = await extension.api.probeWorker(workerUrl, TIMEOUTS.STATUS_CHECK); if (probeResult.ok) { return { worker, queueRemaining: probeResult.queueRemaining || 0, online: true }; } extension.log(`DistributedQueue: Worker ${worker.name} returned ${probeResult.status}`, "debug"); } catch (error) { extension.log(`DistributedQueue: Worker ${worker.name} unreachable: ${error.message}`, "debug"); } return null; }); const results = await Promise.all(checkPromises); return results.filter(r => r !== null); } export async function performPreflightCheck(extension, workers) { if (workers.length === 0) return []; extension.log(`Performing pre-flight health check on ${workers.length} workers...`, "debug"); const startTime = Date.now(); const checkPromises = workers.map(async (worker) => { const workerUrl = getWorkerUrl(extension, worker); extension.log(`Pre-flight checking ${worker.name} at: ${workerUrl}`, "debug"); try { const probeResult = await extension.api.probeWorker(workerUrl, TIMEOUTS.STATUS_CHECK); if (probeResult.ok) { extension.log(`Worker ${worker.name} is active`, "debug"); return { worker, active: true }; } else { extension.log(`Worker ${worker.name} returned ${probeResult.status}`, "debug"); return { worker, active: false }; } } catch (error) { if (error?.name === 'AbortError') { extension.log(`Worker ${worker.name} pre-flight check timed out; assuming active`, "debug"); return { worker, active: true, uncertain: true }; } extension.log(`Worker ${worker.name} is offline or unreachable: ${error.message}`, "debug"); return { worker, active: false }; } }); const results = await Promise.all(checkPromises); const activeWorkers = results.filter(r => r.active).map(r => r.worker); const elapsed = Date.now() - startTime; extension.log(`Pre-flight check completed in ${elapsed}ms. Active workers: ${activeWorkers.length}/${workers.length}`, "debug"); // Update UI status indicators for inactive workers results.filter(r => !r.active).forEach(r => { applyProbeResultToWorkerDot(r.worker.id, { ok: false }); }); return activeWorkers; }