From febd68973beb7a07688d1c75d2843914b8d44c47 Mon Sep 17 00:00:00 2001 From: PickleRick Date: Mon, 21 Sep 2026 22:33:52 +0200 Subject: [PATCH] Save uncommitted changes --- FlowHelpers/1.0.0/copyWithProgress.js | 71 +++++++++++++++++++++------ README.md | 11 +++-- docs/integration-guide.md | 3 ++ tests/quick-logic-test.js | 5 ++ 4 files changed, 71 insertions(+), 19 deletions(-) diff --git a/FlowHelpers/1.0.0/copyWithProgress.js b/FlowHelpers/1.0.0/copyWithProgress.js index 7aab71d..a5c9d8b 100644 --- a/FlowHelpers/1.0.0/copyWithProgress.js +++ b/FlowHelpers/1.0.0/copyWithProgress.js @@ -2,12 +2,19 @@ const fs = require('fs'); -const copyWithProgress = async ({ source, destination, jobLog, updateProgress, label }) => { +const COPY_BUFFER_SIZE = 4 * 1024 * 1024; + +const copyWithProgress = async ({ + source, + destination, + jobLog, + updateProgress, + label, + maxRetries = 2, + retryDelayMs = 1000, +}) => { const sourceSize = fs.statSync(source).size; const sourceSizeMb = (sourceSize / 1024 / 1024).toFixed(1); - const startMs = Date.now(); - let copiedBytes = 0; - let nextProgressPercent = 0; const report = (percent, message) => { const boundedPercent = Math.min(Math.max(percent, 0), 100); @@ -20,9 +27,16 @@ const copyWithProgress = async ({ source, destination, jobLog, updateProgress, l if (typeof jobLog === 'function') jobLog(`${message}\n`); }; - await new Promise((resolve, reject) => { - const rd = fs.createReadStream(source); - const wr = fs.createWriteStream(destination); + const copyOnce = () => 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, + }); let settled = false; const fail = (error) => { if (settled) return; @@ -37,7 +51,7 @@ const copyWithProgress = async ({ source, destination, jobLog, updateProgress, l wr.on('finish', () => { if (settled) return; settled = true; - resolve(); + resolve(((Date.now() - startMs) / 1000).toFixed(1)); }); rd.on('data', (chunk) => { copiedBytes += chunk.length; @@ -57,15 +71,42 @@ const copyWithProgress = async ({ source, destination, jobLog, updateProgress, l rd.pipe(wr); }); - const destinationSize = fs.statSync(destination).size; - if (destinationSize !== sourceSize) { - throw new Error( - `[${label}] Copy size mismatch: expected ${sourceSize} bytes, got ${destinationSize}`, - ); + let lastError; + for (let attempt = 0; attempt <= maxRetries; attempt += 1) { + if (attempt > 0) { + report(0, `[${label}] Retrying copy (${attempt}/${maxRetries})...`); + await new Promise((resolve) => setTimeout(resolve, retryDelayMs)); + } + try { + if (fs.existsSync(destination)) fs.unlinkSync(destination); + const elapsed = await copyOnce(); + 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`); + return; + } catch (error) { + lastError = error; + try { + if (fs.existsSync(destination)) fs.unlinkSync(destination); + } catch (cleanupError) { + lastError = new Error( + `${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`); + } + } } - const elapsed = ((Date.now() - startMs) / 1000).toFixed(1); - report(100, `[${label}] Done: ${sourceSizeMb} MB copied in ${elapsed}s`); + throw new Error( + `[${label}] Copy failed after ${maxRetries + 1} attempts: ${lastError.message}`, + ); }; module.exports = { copyWithProgress }; diff --git a/README.md b/README.md index 50d9c90..cb4a4e2 100644 --- a/README.md +++ b/README.md @@ -278,7 +278,9 @@ Performs 2-pass libx265 encoding using VBV parameters from `args.variables`. **Category**: `utility` · **Step**: before transcode Streams the source file to the worker's local `workDir`, verifies the byte count, -and reports throttled copy progress. It stores the original library path in +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 `variables.originalLibraryFile` for `replaceOriginalFile`. ### 9. `replaceOriginalFile` — Replace Original File @@ -295,9 +297,10 @@ additional progress reporting. **Category**: `utility` · **Step**: after transcode Streams the processed file to a temporary file beside the original, verifies its -size, then atomically replaces the original. An existing original is moved to -`.partial.old` during the operation and restored if replacement fails. The -processed work file is deleted after a successful replacement. +size, and retries failed or incomplete writes twice before giving up. It then +atomically replaces the original. An existing original is moved to `.partial.old` +during the operation and restored if replacement fails. The processed work file +is deleted after a successful replacement. ### 11. `finalizeMedia` — Finalize Media & Sidecars diff --git a/docs/integration-guide.md b/docs/integration-guide.md index 981020b..261eff2 100644 --- a/docs/integration-guide.md +++ b/docs/integration-guide.md @@ -121,6 +121,9 @@ Aggiungi al PATH di sistema: `copyToWorkDir` 2.0.0 is optional but recommended when the source library is on 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. 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 fde311c..e8e8d10 100644 --- a/tests/quick-logic-test.js +++ b/tests/quick-logic-test.js @@ -218,6 +218,11 @@ const AUDIO_INPUTS = { const replacePluginV1Code = require('fs').readFileSync(path.join(ROOT, 'LocalFlowPlugins/utility/replaceOriginalFile/1.0.0/index.js'), 'utf8'); 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('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'); // ── Summary ─────────────────────────────────────────────────────────────── console.log(`\n${'='.repeat(55)}`);