Refactor file upload and progress handling in PrinterFileManager and PrinterClient
All checks were successful
farmcontrol/farmcontrol-server/pipeline/head This commit looks good
All checks were successful
farmcontrol/farmcontrol-server/pipeline/head This commit looks good
- Implement retry logic for file uploads in PrinterFileManager to enhance reliability. - Update progress tracking in PrinterClient to streamline deployment feedback. - Simplify progress callback management in FileManager for improved clarity. - Enhance logging for upload processes and error handling across components.
This commit is contained in:
parent
853784e361
commit
4677e17c7f
@ -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...`);
|
||||
|
||||
@ -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,
|
||||
};
|
||||
|
||||
|
||||
@ -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;
|
||||
|
||||
@ -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}`);
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user