diff --git a/FlowHelpers/1.0.0/copyWithProgress.js b/FlowHelpers/1.0.0/copyWithProgress.js index a5c9d8b..fd44026 100644 --- a/FlowHelpers/1.0.0/copyWithProgress.js +++ b/FlowHelpers/1.0.0/copyWithProgress.js @@ -2,14 +2,13 @@ const fs = require('fs'); -const COPY_BUFFER_SIZE = 4 * 1024 * 1024; - const copyWithProgress = async ({ source, destination, jobLog, updateProgress, label, + copyMethod = 'auto', maxRetries = 2, retryDelayMs = 1000, }) => { @@ -27,16 +26,12 @@ const copyWithProgress = async ({ if (typeof jobLog === 'function') jobLog(`${message}\n`); }; - const copyOnce = () => new Promise((resolve, reject) => { + const copyOnceWithStream = () => new Promise((resolve, reject) => { const startMs = Date.now(); let copiedBytes = 0; let nextProgressPercent = 0; - // Larger buffers avoid issuing thousands of small writes when the - // destination is an SMB/CIFS mount with noticeable round-trip latency. - const rd = fs.createReadStream(source, { highWaterMark: COPY_BUFFER_SIZE }); - const wr = fs.createWriteStream(destination, { - highWaterMark: COPY_BUFFER_SIZE, - }); + const rd = fs.createReadStream(source, { highWaterMark: 4 * 1024 * 1024 }); + const wr = fs.createWriteStream(destination, { highWaterMark: 4 * 1024 * 1024 }); let settled = false; const fail = (error) => { if (settled) return; @@ -45,7 +40,6 @@ const copyWithProgress = async ({ wr.destroy(); reject(error); }; - rd.on('error', fail); wr.on('error', fail); wr.on('finish', () => { @@ -57,20 +51,45 @@ const copyWithProgress = async ({ copiedBytes += chunk.length; const percent = sourceSize === 0 ? 100 : Math.floor((copiedBytes / sourceSize) * 100); if (percent < nextProgressPercent) return; - const elapsedSeconds = Math.max((Date.now() - startMs) / 1000, 0.1); - const copiedMb = (copiedBytes / 1024 / 1024).toFixed(1); - const speedMbPerSecond = (copiedBytes / 1024 / 1024 / elapsedSeconds).toFixed(1); report( percent, `[${label}] Progress: ${Math.min(percent, 100)}% ` - + `(${copiedMb}/${sourceSizeMb} MB, ${speedMbPerSecond} MB/s)`, + + `(${(copiedBytes / 1024 / 1024).toFixed(1)}/${sourceSizeMb} MB, ` + + `${(copiedBytes / 1024 / 1024 / elapsedSeconds).toFixed(1)} MB/s)`, ); nextProgressPercent = (Math.floor(percent / 10) + 1) * 10; }); rd.pipe(wr); }); + const copyOnceWithNative = () => new Promise((resolve, reject) => { + const startMs = Date.now(); + let settled = false; + const progressTimer = setInterval(() => { + if (settled || !fs.existsSync(destination)) return; + const copiedBytes = fs.statSync(destination).size; + const elapsedSeconds = Math.max((Date.now() - startMs) / 1000, 0.1); + report( + Math.floor((copiedBytes / sourceSize) * 100), + `[${label}] Progress: ${Math.min(Math.floor((copiedBytes / sourceSize) * 100), 100)}% ` + + `(${(copiedBytes / 1024 / 1024).toFixed(1)}/${sourceSizeMb} MB, ` + + `${(copiedBytes / 1024 / 1024 / elapsedSeconds).toFixed(1)} MB/s)`, + ); + }, 1000); + fs.promises.copyFile(source, destination).then(() => { + settled = true; + clearInterval(progressTimer); + resolve(((Date.now() - startMs) / 1000).toFixed(1)); + }).catch((error) => { + settled = true; + clearInterval(progressTimer); + reject(error); + }); + }); + + const copyOnce = copyMethod === 'stream' ? copyOnceWithStream : copyOnceWithNative; + let lastError; for (let attempt = 0; attempt <= maxRetries; attempt += 1) { if (attempt > 0) { diff --git a/LocalFlowPlugins/utility/copyToWorkDir/2.0.0/index.js b/LocalFlowPlugins/utility/copyToWorkDir/2.0.0/index.js index 3c90a43..f53a1c3 100644 --- a/LocalFlowPlugins/utility/copyToWorkDir/2.0.0/index.js +++ b/LocalFlowPlugins/utility/copyToWorkDir/2.0.0/index.js @@ -25,6 +25,14 @@ const details = () => ({ inputUI: { type: 'dropdown', options: ['true', 'false'] }, tooltip: 'Skip when the source is already inside workDir.', }, + { + label: 'Copy method', + name: 'copyMethod', + type: 'string', + defaultValue: 'auto', + inputUI: { type: 'dropdown', options: ['auto', 'native', 'stream'] }, + tooltip: 'Auto uses the operating system copy API; stream is a fallback for problematic mounts.', + }, ], outputs: [{ number: 1, tooltip: 'File is ready in workDir — continue to next plugin' }], }); @@ -57,6 +65,7 @@ const plugin = async (args) => { jobLog, updateProgress: args.updateProgress, label: 'copyToWorkDir', + copyMethod: args.inputs.copyMethod || 'auto', }); return { diff --git a/LocalFlowPlugins/utility/replaceOriginalFile/2.0.0/index.js b/LocalFlowPlugins/utility/replaceOriginalFile/2.0.0/index.js index 305a75b..0b4602e 100644 --- a/LocalFlowPlugins/utility/replaceOriginalFile/2.0.0/index.js +++ b/LocalFlowPlugins/utility/replaceOriginalFile/2.0.0/index.js @@ -25,6 +25,14 @@ const details = () => ({ inputUI: { type: 'dropdown', options: ['true', 'false'] }, tooltip: 'Delete the processed file after it has replaced the original.', }, + { + label: 'Copy method', + name: 'copyMethod', + type: 'string', + defaultValue: 'auto', + inputUI: { type: 'dropdown', options: ['auto', 'native', 'stream'] }, + tooltip: 'Auto uses the operating system copy API; stream is a fallback for problematic mounts.', + }, ], outputs: [{ number: 1, tooltip: 'Original file replaced — continue to next plugin' }], }); @@ -62,6 +70,7 @@ const plugin = async (args) => { jobLog, updateProgress: args.updateProgress, label: 'replaceOriginalFile', + copyMethod: args.inputs.copyMethod || 'auto', }); if (fs.existsSync(destination)) { diff --git a/README.md b/README.md index cb4a4e2..f243846 100644 --- a/README.md +++ b/README.md @@ -279,8 +279,9 @@ Performs 2-pass libx265 encoding using VBV parameters from `args.variables`. Streams the source file to the worker's local `workDir`, verifies the byte count, reports throttled copy progress, and retries failed writes twice after removing -the partial destination. It uses large buffered streams for better throughput -when the library is an SMB/CIFS mount. It stores the original library path in +the partial destination. Its `Copy method` input defaults to `auto` (the +operating system's native file-copy path) and offers `stream` as a fallback +for problematic mounts. It stores the original library path in `variables.originalLibraryFile` for `replaceOriginalFile`. ### 9. `replaceOriginalFile` — Replace Original File diff --git a/docs/integration-guide.md b/docs/integration-guide.md index 261eff2..14525ac 100644 --- a/docs/integration-guide.md +++ b/docs/integration-guide.md @@ -122,8 +122,9 @@ Aggiungi al PATH di sistema: a network share. Place it before the first analysis/transcode node; it records the library path for `replaceOriginalFile` 2.0.0. Both progress-enabled copy operations verify the destination size and retry -failed writes twice after removing the partial destination. They use larger -buffers to avoid slow small-write behavior on SMB/CIFS mounts. +failed writes twice after removing the partial destination. Their `Copy method` +input defaults to `auto` (the operating system's native file-copy path), with +`stream` available as a fallback for problematic SMB/CIFS mounts. Both progress-enabled plugins require `FlowHelpers/1.0.0/copyWithProgress.js`; the installer downloads it automatically. diff --git a/tests/quick-logic-test.js b/tests/quick-logic-test.js index e8e8d10..ad3c4bb 100644 --- a/tests/quick-logic-test.js +++ b/tests/quick-logic-test.js @@ -222,7 +222,9 @@ const AUDIO_INPUTS = { ok(copyHelperCode.includes('maxRetries = 2'), 'copy helper retries failed writes'); ok(copyHelperCode.includes('Copy size mismatch'), 'copy helper verifies destination size'); ok(copyHelperCode.includes('fs.unlinkSync(destination)'), 'copy helper removes partial copies before retrying'); - ok(copyHelperCode.includes('COPY_BUFFER_SIZE = 4 * 1024 * 1024'), 'copy helper uses large SMB-friendly buffers'); + ok(copyHelperCode.includes('fs.promises.copyFile(source, destination)'), 'copy helper uses native file copying'); + ok(copyHelperCode.includes("copyMethod = 'auto'"), 'copy helper defaults to automatic copy selection'); + ok(copyHelperCode.includes("copyMethod === 'stream'"), 'copy helper supports stream fallback'); // ── Summary ─────────────────────────────────────────────────────────────── console.log(`\n${'='.repeat(55)}`);