From 7ee5a5bc48e14bf043e8a6fc72ebe99e28051c4a Mon Sep 17 00:00:00 2001 From: PickleRick Date: Mon, 21 Sep 2026 23:38:34 +0200 Subject: [PATCH] Save uncommitted changes --- FlowHelpers/1.0.0/copyWithProgress.js | 175 +++++++----------- .../utility/copyToWorkDir/2.0.0/index.js | 18 -- .../replaceOriginalFile/2.0.0/index.js | 18 -- README.md | 7 +- docs/integration-guide.md | 5 +- tests/quick-logic-test.js | 12 +- 6 files changed, 81 insertions(+), 154 deletions(-) diff --git a/FlowHelpers/1.0.0/copyWithProgress.js b/FlowHelpers/1.0.0/copyWithProgress.js index 9adde62..6160aa2 100644 --- a/FlowHelpers/1.0.0/copyWithProgress.js +++ b/FlowHelpers/1.0.0/copyWithProgress.js @@ -2,122 +2,90 @@ const fs = require('fs'); -const copyWithProgress = async ({ - source, - destination, - jobLog, - updateProgress, - label, - copyMethod = 'auto', - bufferSizeMb = 16, - maxRetries = 2, - retryDelayMs = 1000, -}) => { - const sourceSize = fs.statSync(source).size; - const sourceSizeMb = (sourceSize / 1024 / 1024).toFixed(1); - const normalizedBufferSizeMb = Math.min( - Math.max(Number.parseInt(bufferSizeMb, 10) || 16, 1), - 128, - ); - const bufferSize = normalizedBufferSizeMb * 1024 * 1024; +const MAX_RETRIES = 2; +const RETRY_DELAY_MS = 1000; +const PROGRESS_INTERVAL_MS = 250; - const report = (percent, message) => { - const boundedPercent = Math.min(Math.max(percent, 0), 100); +const sleep = (milliseconds) => new Promise((resolve) => { + setTimeout(resolve, milliseconds); +}); + +const copyWithProgress = async ({ source, destination, jobLog, updateProgress, label }) => { + const sourceSize = fs.statSync(source).size; + const sourceSizeMb = sourceSize / 1024 / 1024; + let lastError; + + const report = (progress, message) => { if (typeof updateProgress === 'function') { - updateProgress({ - progress: boundedPercent, - jobLog: message, - }); + updateProgress({ progress, jobLog: message }); } if (typeof jobLog === 'function') jobLog(`${message}\n`); }; - const copyOnceWithStream = () => new Promise((resolve, reject) => { - const startMs = Date.now(); - let copiedBytes = 0; - let lastReportedPercent = -1; - const rd = fs.createReadStream(source, { highWaterMark: bufferSize }); - const wr = fs.createWriteStream(destination, { highWaterMark: bufferSize }); - let settled = false; - const fail = (error) => { - if (settled) return; - settled = true; - rd.destroy(); - wr.destroy(); - reject(error); - }; - rd.on('error', fail); - wr.on('error', fail); - wr.on('finish', () => { - if (settled) return; - settled = true; - resolve(((Date.now() - startMs) / 1000).toFixed(1)); - }); - rd.on('data', (chunk) => { - copiedBytes += chunk.length; - const percent = sourceSize === 0 ? 100 : Math.floor((copiedBytes / sourceSize) * 100); - if (percent <= lastReportedPercent || percent >= 100) return; - const elapsedSeconds = Math.max((Date.now() - startMs) / 1000, 0.1); - report( - percent, - `[${label}] Progress: ${Math.min(percent, 100)}% ` - + `(${(copiedBytes / 1024 / 1024).toFixed(1)}/${sourceSizeMb} MB, ` - + `${(copiedBytes / 1024 / 1024 / elapsedSeconds).toFixed(1)} MB/s)`, - ); - lastReportedPercent = percent; - }); - rd.pipe(wr); - }); - - const copyOnceWithNative = () => new Promise((resolve, reject) => { - const startMs = Date.now(); - let settled = false; - let lastReportedPercent = -1; - 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); - const percent = sourceSize === 0 - ? 99 - : Math.min(Math.floor((copiedBytes / sourceSize) * 100), 99); - if (percent <= lastReportedPercent) return; - lastReportedPercent = percent; - report( - percent, - `[${label}] Progress: ${percent}% ` - + `(${(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) { + for (let attempt = 0; attempt <= MAX_RETRIES; attempt += 1) { if (attempt > 0) { - report(0, `[${label}] Retrying copy (${attempt}/${maxRetries})...`); - await new Promise((resolve) => setTimeout(resolve, retryDelayMs)); + report(0, `[${label}] Retrying copy (${attempt}/${MAX_RETRIES})...`); + await sleep(RETRY_DELAY_MS); } + try { if (fs.existsSync(destination)) fs.unlinkSync(destination); - const elapsed = await copyOnce(); + + const startedAt = Date.now(); + let lastReportedPercent = -1; + let previousBytes = 0; + let previousAt = startedAt; + let copying = true; + + const progressTimer = setInterval(() => { + if (!copying || !fs.existsSync(destination)) return; + let copiedBytes; + try { + copiedBytes = fs.statSync(destination).size; + } catch (error) { + if (error.code === 'ENOENT') return; + return; + } + const percent = sourceSize === 0 + ? 99 + : Math.min(Math.floor((copiedBytes / sourceSize) * 100), 99); + if (percent <= lastReportedPercent) return; + + const now = Date.now(); + const intervalSeconds = Math.max((now - previousAt) / 1000, 0.001); + const intervalSpeed = (copiedBytes - previousBytes) / 1024 / 1024 / intervalSeconds; + report( + percent, + `[${label}] Progress: ${percent}% ` + + `(${(copiedBytes / 1024 / 1024).toFixed(1)}/${sourceSizeMb.toFixed(1)} MB, ` + + `${intervalSpeed.toFixed(1)} MB/s)`, + ); + lastReportedPercent = percent; + previousBytes = copiedBytes; + previousAt = now; + }, PROGRESS_INTERVAL_MS); + + try { + await fs.promises.copyFile(source, destination); + } finally { + copying = false; + clearInterval(progressTimer); + } + const destinationSize = fs.statSync(destination).size; if (destinationSize !== sourceSize) { throw new Error( `Copy size mismatch: expected ${sourceSize} bytes, got ${destinationSize}`, ); } - report(100, `[${label}] Done: ${sourceSizeMb} MB copied in ${elapsed}s`); + + const elapsedSeconds = Math.max((Date.now() - startedAt) / 1000, 0.001); + const averageSpeed = sourceSize / 1024 / 1024 / elapsedSeconds; + report( + 100, + `[${label}] Done: ${sourceSizeMb.toFixed(1)} MB copied in ` + + `${elapsedSeconds.toFixed(1)}s (${averageSpeed.toFixed(1)} MB/s average)`, + ); return; } catch (error) { lastError = error; @@ -128,15 +96,14 @@ const copyWithProgress = async ({ `${error.message}; could not remove partial copy: ${cleanupError.message}`, ); } - if (attempt === maxRetries) break; - if (typeof jobLog === 'function') { - jobLog(`[${label}] Copy attempt ${attempt + 1} failed: ${error.message}\n`); + if (attempt < MAX_RETRIES && typeof jobLog === 'function') { + jobLog(`[${label}] Copy attempt ${attempt + 1} failed: ${lastError.message}\n`); } } } throw new Error( - `[${label}] Copy failed after ${maxRetries + 1} attempts: ${lastError.message}`, + `[${label}] Copy failed after ${MAX_RETRIES + 1} attempts: ${lastError.message}`, ); }; diff --git a/LocalFlowPlugins/utility/copyToWorkDir/2.0.0/index.js b/LocalFlowPlugins/utility/copyToWorkDir/2.0.0/index.js index 8929a99..3c90a43 100644 --- a/LocalFlowPlugins/utility/copyToWorkDir/2.0.0/index.js +++ b/LocalFlowPlugins/utility/copyToWorkDir/2.0.0/index.js @@ -25,22 +25,6 @@ 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.', - }, - { - label: 'Stream buffer size (MB)', - name: 'bufferSizeMb', - type: 'string', - defaultValue: '16', - inputUI: { type: 'dropdown', options: ['4', '16', '64'] }, - tooltip: 'Used only with stream copy; larger values can improve high-latency SMB mounts.', - }, ], outputs: [{ number: 1, tooltip: 'File is ready in workDir — continue to next plugin' }], }); @@ -73,8 +57,6 @@ const plugin = async (args) => { jobLog, updateProgress: args.updateProgress, label: 'copyToWorkDir', - copyMethod: args.inputs.copyMethod || 'auto', - bufferSizeMb: args.inputs.bufferSizeMb || '16', }); return { diff --git a/LocalFlowPlugins/utility/replaceOriginalFile/2.0.0/index.js b/LocalFlowPlugins/utility/replaceOriginalFile/2.0.0/index.js index 4cd9ba9..305a75b 100644 --- a/LocalFlowPlugins/utility/replaceOriginalFile/2.0.0/index.js +++ b/LocalFlowPlugins/utility/replaceOriginalFile/2.0.0/index.js @@ -25,22 +25,6 @@ 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.', - }, - { - label: 'Stream buffer size (MB)', - name: 'bufferSizeMb', - type: 'string', - defaultValue: '16', - inputUI: { type: 'dropdown', options: ['4', '16', '64'] }, - tooltip: 'Used only with stream copy; larger values can improve high-latency SMB mounts.', - }, ], outputs: [{ number: 1, tooltip: 'Original file replaced — continue to next plugin' }], }); @@ -78,8 +62,6 @@ const plugin = async (args) => { jobLog, updateProgress: args.updateProgress, label: 'replaceOriginalFile', - copyMethod: args.inputs.copyMethod || 'auto', - bufferSizeMb: args.inputs.bufferSizeMb || '16', }); if (fs.existsSync(destination)) { diff --git a/README.md b/README.md index 2d2217b..1fac32f 100644 --- a/README.md +++ b/README.md @@ -279,10 +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. Its `Copy method` input defaults to `auto` (the -operating system's native file-copy path) and offers `stream` as a fallback -for problematic mounts. The stream fallback defaults to a 16 MB buffer and -offers 4 MB or 64 MB alternatives. It stores the original library path in +the partial destination. Its `Copy method` uses the operating system's native file-copy path, reports interval speed, +verifies the destination size, and retries failed writes. 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 3323fbe..1bb31c4 100644 --- a/docs/integration-guide.md +++ b/docs/integration-guide.md @@ -123,9 +123,8 @@ 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. 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. Stream mode -supports 4 MB, 16 MB, and 64 MB buffers. +uses the operating system's native file-copy path, reports interval speed, +verifies the destination size, and retries failed writes. 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 83165fc..2fe36cd 100644 --- a/tests/quick-logic-test.js +++ b/tests/quick-logic-test.js @@ -219,16 +219,14 @@ const AUDIO_INPUTS = { ok(replacePluginV1Code.includes('copyAndVerify'), 'replaceOriginalFile v1 verifies the copied file'); ok(!replacePluginV1Code.includes('updateProgress'), 'replaceOriginalFile v1 has no progress callback'); const copyHelperCode = require('fs').readFileSync(path.join(ROOT, 'FlowHelpers/1.0.0/copyWithProgress.js'), 'utf8'); - ok(copyHelperCode.includes('maxRetries = 2'), 'copy helper retries failed writes'); + ok(copyHelperCode.includes('MAX_RETRIES = 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('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'); - ok(copyHelperCode.includes('bufferSizeMb = 16'), 'copy helper defaults stream buffer to 16 MB'); - ok(copyHelperCode.includes('128'), 'copy helper caps stream buffer size'); - ok(copyHelperCode.includes('Math.min(Math.floor((copiedBytes / sourceSize) * 100), 99)'), 'native progress reserves 100% for completed copy'); - ok(copyHelperCode.includes('percent <= lastReportedPercent'), 'copy helper reports each progress percentage once'); + ok(copyHelperCode.includes('PROGRESS_INTERVAL_MS = 250'), 'copy helper polls progress frequently'); + ok(copyHelperCode.includes('intervalSpeed'), 'copy helper reports interval speed'); + ok(!copyPluginV2Code.includes('copyMethod'), 'copyToWorkDir v2 has no copy method setting'); + ok(!replacePluginCode.includes('bufferSizeMb'), 'replaceOriginalFile v2 has no buffer setting'); // ── Summary ─────────────────────────────────────────────────────────────── console.log(`\n${'='.repeat(55)}`);