diff --git a/src/files/filemanager.js b/src/files/filemanager.js index dcd4529..d027632 100644 --- a/src/files/filemanager.js +++ b/src/files/filemanager.js @@ -205,24 +205,21 @@ export class FileManager { // Update files list from host this.updateFiles(); + // Check if file exists in cache + if (this.fileExistsInCache(fileId)) { + logger.debug(`File ${fileId} found in cache`); + if (onProgress) { + onProgress(100); + } + return this.getFileFromCache(fileId); + } + if (onProgress) { const progressCallbacks = this.progressCallbacks.get(fileId) || []; progressCallbacks.push(onProgress); this.progressCallbacks.set(fileId, progressCallbacks); } - // Check if file exists in cache - if (this.fileExistsInCache(fileId)) { - logger.debug(`File ${fileId} found in cache`); - if (this.progressCallbacks.has(fileId)) { - const progressCallbacks = this.progressCallbacks.get(fileId); - for (const callback of progressCallbacks) { - callback(100); - } - } - return this.getFileFromCache(fileId); - } - // Check if download is already in progress for this file if (this.downloadingFiles.has(fileId)) { logger.debug(`File ${fileId} download already in progress, waiting...`); diff --git a/src/printer/database.js b/src/printer/database.js index bebd6b0..b0a815d 100644 --- a/src/printer/database.js +++ b/src/printer/database.js @@ -200,20 +200,28 @@ export class PrinterDatabase { deploying: 0, }; - let jobProgress = 0; subJobStates.forEach((state) => { stateCounts[state.type]++; - var subJobProgress = state.progress || 0; + }); + + logger.debug(`Job ${jobId} state counts:`, stateCounts); + + const jobStateType = this.determineJobState(stateCounts); + + let jobProgress = 0; + subJobStates.forEach((state) => { + let subJobProgress = Number(state.progress) || 0; if (state.type === "complete") { subJobProgress = 1; + } else if (jobStateType === "deploying" && state.type === "queued") { + // Subjobs transition to queued once deployment finishes; count them as done. + subJobProgress = 1; } jobProgress += subJobProgress; }); - logger.debug(`Job ${jobId} state counts:`, stateCounts); - const jobState = { - type: this.determineJobState(stateCounts), + type: jobStateType, progress: jobProgress / subJobs.length, }; diff --git a/src/printer/printerclient.js b/src/printer/printerclient.js index 76f6f1f..ef02ae5 100644 --- a/src/printer/printerclient.js +++ b/src/printer/printerclient.js @@ -1103,10 +1103,11 @@ export class PrinterClient { const arrayBuffer = await fileBlob.arrayBuffer(); const buffer = Buffer.from(arrayBuffer); - const formData = new FormData(); + const formData = new FormData({ maxDataSize: Infinity }); formData.append("file", buffer, { filename: fileName, contentType: fileBlob.type || "text/plain", + knownLength: buffer.length, }); const headers = { @@ -1119,6 +1120,8 @@ export class PrinterClient { const response = await axios.post(httpUrl, formData, { headers, + maxBodyLength: Infinity, + maxContentLength: Infinity, onUploadProgress: (progressEvent) => { const percentCompleted = Math.round( (progressEvent.loaded * 100) / progressEvent.total, @@ -1164,27 +1167,23 @@ export class PrinterClient { const now = Date.now(); + if (this.deploySubJobTargetTime == null) { + this.deploySubJobTargetTime = now + 500; + } else { + this.deploySubJobTargetTime += 500; + } + + const remainingTime = this.deploySubJobTargetTime - now; + clearTimeout(this.deploySubJobTimer); + this.deploySubJobTimer = setTimeout(() => { + this.processDeploySubJobQueue(); + }, Math.max(remainingTime, 0)); + await this.database.updateSubJobState(subJob._id, { type: "deploying", progress: 0, }); - - // If this is the first call, set target time to 500ms from now - if (this.deploySubJobQueue.length === 1) { - this.deploySubJobTargetTime = now + 500; - this.deploySubJobTimer = setTimeout(() => { - this.processDeploySubJobQueue(); - }, 500); - } else { - // If this is a subsequent call, extend the target time by another 500ms - this.deploySubJobTargetTime += 500; - const remainingTime = this.deploySubJobTargetTime - now; - - clearTimeout(this.deploySubJobTimer); - this.deploySubJobTimer = setTimeout(() => { - this.processDeploySubJobQueue(); - }, remainingTime); - } + await this.database.updateJobState(subJob.job._id); } async processDeploySubJobQueue() { @@ -1203,27 +1202,20 @@ export class PrinterClient { `Processing ${subJobsToDeploy.length} queued sub job(s) for printer ${this.id}`, ); - // Process sub jobs in parallel with a 250ms stagger between starts - const staggerMs = 250; - const deployPromises = subJobsToDeploy.map((subJob, index) => - (async () => { - const startDelay = index * staggerMs; - if (startDelay > 0) { - await new Promise((resolve) => setTimeout(resolve, startDelay)); - } - - try { - await this._deploySubJobInternal(subJob); - } catch (error) { - logger.error( - `Error deploying sub job ${subJob._id} to printer ${this.id}:`, - error, - ); - } - })(), - ); - - await Promise.all(deployPromises); + for (const subJob of subJobsToDeploy) { + try { + await this._deploySubJobInternal(subJob); + } catch (error) { + logger.error( + `Error deploying sub job ${subJob._id} to printer ${this.id}:`, + error, + ); + await this.database.updateSubJobState(subJob._id, { + type: "failed", + }); + await this.database.updateJobState(subJob.job._id); + } + } } async _deploySubJobInternal(subJob) { @@ -1246,52 +1238,62 @@ export class PrinterClient { } console.log("gcodeFile", gcodeFile); + + const fileId = (gcodeFile.file?._id || gcodeFile.file).toString(); + let deploymentActive = true; + + const updateDeployProgress = async (progress) => { + if (!deploymentActive) { + return; + } + await this.database.updateSubJobState(subJob._id, { + type: "deploying", + progress, + }); + await this.database.updateJobState(subJob.job._id); + }; + const file = await this.socketClient.fileManager.getFile( - gcodeFile.file?._id || gcodeFile.file, + fileId, async (progress) => { - await this.database.updateSubJobState(subJob._id, { - type: "deploying", - progress: (progress / 100 / 2).toFixed(2), - }); - await this.database.updateJobState(subJob.job._id); + await updateDeployProgress(progress / 200); }, ); if (!file) { throw new Error("Error getting file"); } - if (!this.printrFileIds.includes(gcodeFile.file.toString())) { - console.log("Uploading file to printer"); + if (!this.printrFileIds.includes(fileId)) { + logger.debug(`Uploading file ${fileId} to printer`); const uploadResult = await this.printerFileManager.uploadFile( - gcodeFile.file, + fileId, file, async (progress) => { - await this.database.updateSubJobState(subJob._id, { - type: "deploying", - progress: (progress / 100 / 2 + 0.5).toFixed(2), - }); - await this.database.updateJobState(subJob.job._id); + await updateDeployProgress(progress / 200 + 0.5); }, ); - this.printrFileIds.push(gcodeFile.file.toString()); if (!uploadResult) { throw new Error("Failed to upload file"); } + this.printrFileIds.push(fileId); } else { - console.log("File already uploaded to printer"); + logger.debug(`File ${fileId} already uploaded to printer`); + await updateDeployProgress(1); } + deploymentActive = false; + await this._runQueueMutation(async () => { const result = await this.sendPrinterCommand({ method: "server.job_queue.post_job", params: { - filenames: [`${gcodeFile.file}.gcode`], + filenames: [`${fileId}.gcode`], reset: false, }, }); if (!result || !result?.queued_jobs) { - return { error: "Failed to deploy sub job to printer" }; + throw new Error("Failed to deploy sub job to printer"); } const queuedJobs = result.queued_jobs; diff --git a/src/printer/printerfilemanager.js b/src/printer/printerfilemanager.js index 8d43075..6df0939 100644 --- a/src/printer/printerfilemanager.js +++ b/src/printer/printerfilemanager.js @@ -1,4 +1,5 @@ // printerfilemanager.js - Manages file uploads from FileManager to Moonraker +import fs from "fs"; import axios from "axios"; import FormData from "form-data"; import log4js from "log4js"; @@ -9,6 +10,23 @@ const config = loadConfig(); const logger = log4js.getLogger("Printer File Manager"); logger.level = config.logLevel; +const UPLOAD_MAX_RETRIES = 3; +const UPLOAD_RETRY_DELAY_MS = 1000; + +function isRetryableUploadError(error) { + const code = error?.code || error?.cause?.code; + return ( + code === "ERR_STREAM_PREMATURE_CLOSE" || + code === "ECONNRESET" || + code === "EPIPE" || + code === "ETIMEDOUT" + ); +} + +function sleep(ms) { + return new Promise((resolve) => setTimeout(resolve, ms)); +} + export class PrinterFileManager { constructor(printerClient) { this.printerClient = printerClient; @@ -59,13 +77,13 @@ export class PrinterFileManager { // Create upload promise and store it const uploadPromise = this._performUpload(fileId, file) .then((result) => { - // Remove from uploading map on success this.uploadingFiles.delete(fileId); + this.progressCallbacks.delete(fileId); return result; }) .catch((error) => { - // Remove from uploading map on error this.uploadingFiles.delete(fileId); + this.progressCallbacks.delete(fileId); throw error; }); @@ -79,7 +97,32 @@ export class PrinterFileManager { * Internal method to perform the actual upload * @private */ - async _performUpload(fileId, file, onProgress = null) { + async _performUpload(fileId, file) { + let lastError; + + for (let attempt = 1; attempt <= UPLOAD_MAX_RETRIES; attempt++) { + try { + return await this._attemptUpload(fileId, file); + } catch (error) { + lastError = error; + if (attempt >= UPLOAD_MAX_RETRIES || !isRetryableUploadError(error)) { + throw error; + } + + logger.warn( + `Upload attempt ${attempt} failed for file ${fileId}, retrying... (${error.message})`, + ); + await sleep(UPLOAD_RETRY_DELAY_MS * attempt); + } + } + + throw lastError; + } + + /** + * @private + */ + async _attemptUpload(fileId, file) { try { const uploadFileName = `${fileId}.gcode`; @@ -89,14 +132,29 @@ export class PrinterFileManager { }://${host}:${port}/server/files/upload`; logger.info( - `Uploading file ${uploadFileName} to printer ${this.printerClient.id} at ${httpUrl}` + `Uploading file ${uploadFileName} to printer ${this.printerClient.id} at ${httpUrl}`, ); - // Create FormData with the file buffer - const formData = new FormData(); - formData.append("file", file, { + const filePath = this.socketClient.fileManager.getFilePath(fileId); + let filePart; + let knownLength; + + if (fs.existsSync(filePath)) { + const stats = fs.statSync(filePath); + knownLength = stats.size; + filePart = fs.createReadStream(filePath); + } else if (Buffer.isBuffer(file)) { + knownLength = file.length; + filePart = file; + } else { + throw new Error(`File ${fileId} not found for upload`); + } + + const formData = new FormData({ maxDataSize: Infinity }); + formData.append("file", filePart, { filename: "farmcontrol/" + uploadFileName, contentType: "text/plain", + knownLength, }); // Set up headers @@ -112,13 +170,15 @@ export class PrinterFileManager { // Upload to Moonraker const response = await axios.post(httpUrl, formData, { headers, + maxBodyLength: Infinity, + maxContentLength: Infinity, onUploadProgress: (progressEvent) => { if (progressEvent.total) { const percentCompleted = Math.round( - (progressEvent.loaded * 100) / progressEvent.total + (progressEvent.loaded * 100) / progressEvent.total, ); logger.debug( - `Uploading file to printer ${this.printerClient.id}: ${uploadFileName} ${percentCompleted}%` + `Uploading file to printer ${this.printerClient.id}: ${uploadFileName} ${percentCompleted}%`, ); if (this.progressCallbacks.has(fileId)) { const progressCallbacks = this.progressCallbacks.get(fileId); @@ -132,18 +192,18 @@ export class PrinterFileManager { // Check response if (response.data && response.data.action) { logger.info( - `Successfully uploaded file ${uploadFileName} to printer ${this.printerClient.id}` + `Successfully uploaded file ${uploadFileName} to printer ${this.printerClient.id}`, ); return true; } else { logger.error( - `Failed to upload file ${uploadFileName} to printer ${this.printerClient.id}: Invalid response` + `Failed to upload file ${uploadFileName} to printer ${this.printerClient.id}: Invalid response`, ); return false; } } catch (error) { logger.error( - `Error uploading file ${fileId} to printer ${this.printerClient.id}: ${error.message}` + `Error uploading file ${fileId} to printer ${this.printerClient.id}: ${error.message}`, ); if (error.response) { logger.error(`Response status: ${error.response.status}`);