Save uncommitted changes

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);
   });
 
+  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(
-      `[${label}] Copy size mismatch: expected ${sourceSize} bytes, got ${destinationSize}`,
+          `Copy size mismatch: expected ${sourceSize} bytes, got ${destinationSize}`,
         );
       }
-
-  const elapsed = ((Date.now() - startMs) / 1000).toFixed(1);
       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`);
+      }
+    }
+  }
+
+  throw new Error(
+    `[${label}] Copy failed after ${maxRetries + 1} attempts: ${lastError.message}`,
+  );
 };
 
 module.exports = { copyWithProgress };

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
 

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.

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)}`);