Fix/recoverable checkpoint and 5.0.4

jest.config.browser.mjs

@@ -7,6 +7,7 @@ export default {
     "<rootDir>/test/downloadFunctions.spec.mjs",
     "<rootDir>/test/downloadCallbacks.spec.mjs",
     "<rootDir>/test/downloadUnsealedCallbacks.spec.mjs",
+    "<rootDir>/test/progress.spec.mjs",
   ],
 
   roots: ['<rootDir>/test', '<rootDir>/src'],

package.json

@@ -1,6 +1,6 @@
 {
     "name": "@yeez-tech/meta-encryptor",
-    "version": "5.0.3",
+    "version": "5.0.4",
     "description": "Data Seal/Unseal for Fidelius",
     "main": "build/commonjs/index.node.cjs",
     "module": "build/es/index.node.js",

src/browser/README.md

@@ -47,6 +47,7 @@ npm install @yeez-tech/meta-encryptor
 | `filename`   | string   | ✅   | 下载文件名（必需，无默认值）                                 |
 | `onLog`      | function | ❌   | 日志回调 `(message: string) => void`                         |
 | `onProgress` | function | ❌   | 进度回调 `(total, processed, readBytes, writeBytes) => void` |
+| `onDownloadReady` | function | ❌ | 下载就绪回调：HTTP 流首个数据块进入管道时触发（HEAD+tail 完成后），可用于关闭准备蒙层 |
 | `onSuccess`  | function | ❌   | 成功回调 `(data: { filename }) => void`                      |
 | `onError`    | function | ❌   | 错误回调 `(error: Error) => void`                            |
 

src/browser/blob_download.js

@@ -1,14 +1,14 @@
 import { Unsealer } from './Unsealer.js'
 import { HttpSealedFileStream } from './HttpSealedFileStream.js'
-import { createProgressTransformer } from '../common/progress.js';
+import { createProgressTransformer, createDownloadReadyTransformer } from '../common/progress.js';
 
 /**
  * @param {string} url
  * @param {string} privateKeyHex
  * @param {string} filename
- * @param {{ log?: Function, onProgress?: Function, fetch?: Function, size?: number }} [opts]
+ * @param {{ log?: Function, onProgress?: Function, fetch?: Function, size?: number, onDownloadReady?: Function }} [opts]
  */
-export async function blobDownloadAndDecrypt(url, privateKeyHex, filename, { log, onProgress, fetch: _fetch, size } = {}) {
+export async function blobDownloadAndDecrypt(url, privateKeyHex, filename, { log, onProgress, fetch: _fetch, size, onDownloadReady } = {}) {
   log = log || (() => {})
   try {
     const chunks = []
@@ -22,6 +22,7 @@ export async function blobDownloadAndDecrypt(url, privateKeyHex, filename, { log
     })
 
     await stream
+      .pipeThrough(createDownloadReadyTransformer(onDownloadReady))
       .pipeThrough(unsealer)
       .pipeThrough(createProgressTransformer(size, onProgress))
       .pipeTo(new WritableStream({

src/browser/downloadUnsealed.js

@@ -85,6 +85,7 @@ async function inspectSealed(url, log) {
  * @param {string} options.filename
  * @param {Function} [options.onLog]
  * @param {Function} [options.onProgress] - (total, processed, readBytes, writeBytes) => {}
+ * @param {Function} [options.onDownloadReady] - HTTP 流首个数据块进入管道时触发（可关蒙层）
  * @param {Function} [options.onSuccess]
  * @param {Function} [options.onError]
  * @returns {Promise<void>}
@@ -95,6 +96,7 @@ export async function downloadUnsealed({
   filename,
   onLog,
   onProgress,
+  onDownloadReady,
   onSuccess,
   onError
 }) {
@@ -128,7 +130,12 @@ export async function downloadUnsealed({
     if (!mobile) {
       try {
         log('Trying stream download...')
-        await streamDownloadAndDecrypt(url, key, filename, { log, onProgress, size: meta.plaintextSize })
+        await streamDownloadAndDecrypt(url, key, filename, {
+          log,
+          onProgress,
+          size: meta.plaintextSize,
+          onDownloadReady,
+        })
         if (onSuccess) onSuccess({ filename })
         return
       } catch (e) {
@@ -137,7 +144,7 @@ export async function downloadUnsealed({
     }
 
     log('Using Blob download...')
-    await blobDownloadAndDecrypt(url, key, filename, { log, onProgress })
+    await blobDownloadAndDecrypt(url, key, filename, { log, onProgress, onDownloadReady })
     if (onSuccess) onSuccess({ filename })
   } catch (error) {
     log('Download failed: ' + error.message)

src/browser/stream_download.js

@@ -1,7 +1,7 @@
 import { Unsealer } from './Unsealer.js'
 import { HttpSealedFileStream } from './HttpSealedFileStream.js'
 import { MetaEncryptorError } from '../common/errors.js';
-import { createProgressTransformer } from '../common/progress.js';
+import { createProgressTransformer, createDownloadReadyTransformer } from '../common/progress.js';
 
 async function ensureStreamSaver(log) {
   if (typeof window === 'undefined') return null;
@@ -50,7 +50,7 @@ export async function getBestWritable(filename, { log, size } = {}) {
   return null;
 }
 
-export async function streamDownloadAndDecrypt(url, privateKeyHex, filename, { log, onProgress, writable, size, fetch: _fetch } = {}) {
+export async function streamDownloadAndDecrypt(url, privateKeyHex, filename, { log, onProgress, writable, size, fetch: _fetch, onDownloadReady } = {}) {
   log = log || (() => {})
   try {
     const out = writable || await getBestWritable(filename, { log, size })
@@ -66,6 +66,7 @@ export async function streamDownloadAndDecrypt(url, privateKeyHex, filename, { l
     })
 
     await stream
+      .pipeThrough(createDownloadReadyTransformer(onDownloadReady))
       .pipeThrough(unsealer)
       .pipeThrough(createProgressTransformer(size, onProgress))
       .pipeTo(out)

src/common/progress.js

@@ -18,3 +18,25 @@ export function createProgressTransformer(plaintextSize, onProgress) {
     }
   })
 }
+
+/**
+ * Create a TransformStream that fires once when the first chunk flows through.
+ * Place after HttpSealedFileStream to signal download stream is ready (e.g. dismiss overlay).
+ *
+ * @param {Function} [onDownloadReady]
+ * @returns {TransformStream}
+ */
+export function createDownloadReadyTransformer(onDownloadReady) {
+  let called = false
+  return new TransformStream({
+    transform(chunk, controller) {
+      if (!called) {
+        called = true
+        if (onDownloadReady) {
+          onDownloadReady()
+        }
+      }
+      controller.enqueue(chunk)
+    }
+  })
+}

src/common/ypccrypto.common.js

@@ -132,14 +132,17 @@ function generateAESKeyFrom(pkey, skey){
   return derived_key;
 }
 
-const eth_hash_prefix = Buffer.from("\\x19Ethereum Signed Message:\\n32");
+const eth_hash_prefix = Buffer.concat([
+  Buffer.from([0x19]),
+  Buffer.from("Ethereum Signed Message:\n32"),
+]);
 
 function signMessage(skey, raw) {
-  let raw_hash = Buffer.from(keccak256(Buffer.from(raw)));
+  let raw_hash = Buffer.from(keccak256(toUint8Array(raw)));
   let msg = new Uint8Array(eth_hash_prefix.length + raw_hash.length)
   msg.set(eth_hash_prefix)
   msg.set(raw_hash, eth_hash_prefix.length)
-  msg = Buffer.from(keccak256(Buffer.from(msg)))
+  msg = Buffer.from(keccak256(toUint8Array(msg)))
 
   const msgBytes = toUint8Array(msg);
   const skeyBytes = toUint8Array(skey);

src/node/PipelineConext.js

@@ -5,10 +5,22 @@ import { promisify } from 'util';
 
 import { MetaEncryptorError } from '../common/errors.js';
 
-const readFile = promisify(fs.readFile);
-const writeFile = promisify(fs.writeFile);
+const open = promisify(fs.open);
+const close = promisify(fs.close);
+const fsync = promisify(fs.fsync);
+const rename = promisify(fs.rename);
 const logger = log.getLogger("meta-encryptor/PipelineContext");
 
+function toBinaryChunk(value) {
+    if (Buffer.isBuffer(value)) {
+        return value;
+    }
+    if (value instanceof Uint8Array) {
+        return Buffer.from(value.buffer, value.byteOffset, value.byteLength);
+    }
+    return null;
+}
+
 export class PipelineContext {
     constructor(options) {
         this.context = {};
@@ -37,25 +49,25 @@ export class PipelineContextInFile extends PipelineContext {
     constructor(filePath, options) {
         super(options);
         this.filePath = filePath;
+        this._saveTail = Promise.resolve();
+        this._saveDirty = false;
     }
 
-    async saveContext() {
+    _buildPayload() {
         const binaryChunks = [];
         const meta = {};
         let offset = 0;
 
-        logger.debug("Context before save:", this.context);
-        logger.debug("PipelineContextInFile::saveContext saving to ", this.filePath);
-
         for (const [key, value] of Object.entries(this.context)) {
-            if (Buffer.isBuffer(value)) {
-                binaryChunks.push(value);
+            const binary = toBinaryChunk(value);
+            if (binary) {
+                binaryChunks.push(binary);
                 meta[key] = {
                     type: 'binary',
                     offset,
-                    length: value.length
+                    length: binary.length
                 };
-                offset += value.length;
+                offset += binary.length;
             } else {
                 meta[key] = {
                     type: 'json',
@@ -68,26 +80,58 @@ export class PipelineContextInFile extends PipelineContext {
         const metaBuffer = Buffer.from(metaStr);
         const metaLength = metaBuffer.length;
 
+        const totalSize = 4 + metaLength + offset;
+        const fileBuffer = Buffer.alloc(totalSize);
+        fileBuffer.writeUInt32BE(metaLength, 0);
+        metaBuffer.copy(fileBuffer, 4);
+        let currentOffset = 4 + metaLength;
+        for (const chunk of binaryChunks) {
+            chunk.copy(fileBuffer, currentOffset);
+            currentOffset += chunk.length;
+        }
+
+        return fileBuffer;
+    }
+
+    async _writeContextAtomic() {
+        const fileBuffer = this._buildPayload();
+        const tmpPath = `${this.filePath}.tmp`;
+
+        logger.debug("PipelineContextInFile::saveContext saving to ", this.filePath);
+
+        const fd = await open(tmpPath, 'w');
         try {
-            const fd = await promisify(fs.open)(this.filePath, 'w');
-            const lengthBuffer = Buffer.alloc(4);
-            lengthBuffer.writeUInt32BE(metaLength);
-            await promisify(fs.write)(fd, lengthBuffer, 0, 4, 0);
-            await promisify(fs.write)(fd, metaBuffer, 0, metaLength, 4);
-            let currentOffset = 4 + metaLength;
-            for (const chunk of binaryChunks) {
-                await promisify(fs.write)(fd, chunk, 0, chunk.length, currentOffset);
-                currentOffset += chunk.length;
+            await promisify(fs.write)(fd, fileBuffer, 0, fileBuffer.length, 0);
+            await fsync(fd);
+        } finally {
+            await close(fd);
+        }
+
+        await rename(tmpPath, this.filePath);
+    }
+
+    async _flushAll() {
+        while (this._saveDirty) {
+            this._saveDirty = false;
+            try {
+                await this._writeContextAtomic();
+            } catch (error) {
+                logger.error('PipelineContextInFile::saveContext error:', error.message);
+                throw error;
             }
-            await promisify(fs.close)(fd);
-        } catch (error) {
-            logger.error('PipelineContextInFile::saveContext error:', error.message);
-            throw error;
         }
     }
 
+    saveContext() {
+        this._saveDirty = true;
+        this._saveTail = this._saveTail.then(() => this._flushAll());
+        return this._saveTail;
+    }
+
     async loadContext() {
         try {
+            await this._saveTail;
+
             if (!fs.existsSync(this.filePath)) {
                 this.context = {};
                 this.runtime.rawCommitted = 0;
@@ -96,7 +140,7 @@ export class PipelineContextInFile extends PipelineContext {
                 return;
             }
 
-            const fd = await promisify(fs.open)(this.filePath, 'r');
+            const fd = await open(this.filePath, 'r');
             const metaLengthBuffer = Buffer.alloc(4);
             await promisify(fs.read)(fd, metaLengthBuffer, 0, 4, 0);
             const metaLength = metaLengthBuffer.readUInt32BE();
@@ -118,7 +162,7 @@ export class PipelineContextInFile extends PipelineContext {
                 }
             }
 
-            await promisify(fs.close)(fd);
+            await close(fd);
 
             const readStart = this.context.readStart || 0;
             const writeStart = this.context.writeStart || 0;

src/node/Recoverable.js

@@ -219,7 +219,7 @@ export class RecoverableWriteStream extends Writable {
         const blocks = runtime.pendingBlocks || [];
 
         let hasCommittedBlock = false;
-        let committedRawBytes = 0;
+        let committedItems = 0;
 
         logger.debug("On plaintext written:", writtenBytes, " bytes. Current runtime:", runtime);
         while(remain > 0 && blocks.length > 0){
@@ -233,9 +233,9 @@ export class RecoverableWriteStream extends Writable {
                 // Block fully committed
                 runtime.rawCommitted += block.rawSize;
                 runtime.plainCommitted += block.plainSize;
-                committedRawBytes += block.rawSize;
                 blocks.shift();
                 hasCommittedBlock = true;
+                committedItems += 1;
             }
         }
         logger.debug("After committing, remaining to commit:", remain, " bytes. Updated runtime:", runtime);
@@ -244,40 +244,48 @@ export class RecoverableWriteStream extends Writable {
             return Promise.resolve();
         }
         if(this.context.context){
-            const buf = this.context.context['data'];
-            if(Buffer.isBuffer(buf) && buf.length > 0 &&
-               committedRawBytes > 0){
-                if(committedRawBytes >= buf.length){
-                    // All data committed
-                    this.context.context['data'] = Buffer.alloc(0);
-                }else{
-                    // Partial data committed
-                    this.context.context['data'] = buf.subarray(committedRawBytes);
-                }
-            }
             this.context.context['readStart'] = runtime.rawCommitted;
             this.context.context['writeStart'] = runtime.plainCommitted;
+            if (committedItems > 0) {
+                this.context.context['readItemCount'] =
+                    (this.context.context['readItemCount'] || 0) + committedItems;
+            }
+            // Checkpoint only fully committed progress; discard in-flight cipher tail.
+            this.context.context['data'] = Buffer.alloc(0);
             logger.debug("After writing, updated readStart to:", this.context.context['readStart'],
                          " writeStart to:", this.context.context['writeStart'],
-                         " data length to:", this.context.context['data'] ? this.context.context['data'].length : 0);
-            //
-            this.context.saveContext();
+                         " readItemCount to:", this.context.context['readItemCount']);
+            return this.context.saveContext();
         }
         return Promise.resolve();
     }
 
     _final(callback) {
         this.writeStream.on('finish', () => {
-            const writeStart = this.context.context['writeStart'] || 0;
+            const ctx = this.context && this.context.context;
+            const runtime = this.context && this.context.runtime;
+            const writeStart = (ctx && ctx['writeStart']) || (runtime && runtime.plainCommitted) || 0;
+
+            if (ctx && runtime) {
+                ctx['readStart'] = runtime.rawCommitted;
+                ctx['writeStart'] = runtime.plainCommitted;
+                ctx['data'] = Buffer.alloc(0);
+            }
 
             // Always truncate to writeStart to remove any residual
             // data from previous incomplete write attempts.
             fs.truncate(this.filePath, writeStart, (truncateErr) => {
                 if (truncateErr) {
                     logger.warn("Error truncating file:", truncateErr);
                     callback(truncateErr);
+                    return;
+                }
+                logger.debug("File truncated successfully to length:", writeStart);
+                if (this.context && typeof this.context.saveContext === 'function') {
+                    Promise.resolve(this.context.saveContext())
+                        .then(() => callback())
+                        .catch((err) => callback(err));
                 } else {
-                    logger.debug("File truncated successfully to length:", writeStart);
                     callback();
                 }
             });