Save uncommitted changes

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

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 {

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)) {

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

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.

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