From 2b0afddd0723be6258182ef8e15dc0f9e6bfd6a3 Mon Sep 17 00:00:00 2001 From: Codex Bot Date: Thu, 12 Nov 2026 13:04:23 +0800 Subject: [PATCH] docs(runtime): explain digest scheduling and streaming boundaries --- src/lib/canvas-runtime/browser-digests.ts | 7 ++++++- src/lib/canvas-runtime/digest-pool.ts | 15 +++++++++++---- src/lib/canvas-runtime/digest-worker-source.ts | 8 ++++++-- src/lib/canvas-runtime/digest.worker.ts | 4 +++- src/lib/canvas-runtime/import-integrity.ts | 8 ++++++-- src/lib/canvas-runtime/sha256.ts | 8 ++++++-- 6 files changed, 38 insertions(+), 12 deletions(-) diff --git a/src/lib/canvas-runtime/browser-digests.ts b/src/lib/canvas-runtime/browser-digests.ts index 859d4c4..8660e62 100644 --- a/src/lib/canvas-runtime/browser-digests.ts +++ b/src/lib/canvas-runtime/browser-digests.ts @@ -5,7 +5,10 @@ import {fault} from '../canvas-core/values.ts'; import {createDigestPool} from './digest-pool.ts'; /** Browser capability stays at the adapter; WorkspaceRuntime owns the pool's lifetime. */ -/** 构建 workspace 可共享的 DigestPort；任务 ID/长度/版本共同验证响应归属。 */ +/** 在构造时选择能力：Worker 不存在就使用池的默认分块执行器，否则每个 execute 新建专用 Worker。 + * 此选择不随后续 Worker 全局变化自动重算；创建/运行失败也不降级重试。 + * 使用默认池预算，响应按实例内递增 ID/输入长度/版本和摘要格式核对，不独立重算摘要。 + * 没有超时，线程无响应需外部取消/close；Worker 内原生哈希阈值由 digestWorkerSource 控制。 */ export function createBrowserDigests(){ let next=0; return createDigestPool({execute:typeof Worker==='undefined'?undefined:(bytes,signal)=>new Promise((resolve,reject)=>{ @@ -15,6 +18,8 @@ export function createBrowserDigests(){ const id=++next;let done=false; /** 幂等完成：解绑所有消息/错误/abort 监听并终止 Worker，防止迟到响应泄漏或重复 resolve。 */ const finish=(error?:unknown,sha256?:string)=>{if(done)return;done=true;signal.removeEventListener('abort',abort);worker.onmessage=null;worker.onerror=null;worker.onmessageerror=null;worker.terminate();if(error)reject(error);else resolve(sha256!);},abort=()=>finish(fault('import_aborted')); + // 稳定 digest_worker_failed 只有身份匹配才采纳；其它响应走摘要字段校验，不把任意 error 字段当服务错误。 + // 格式正确只表明消息可解释，真实性来自受信的专用 Worker 算法；不在 UI 线程重复计算。 // 消息须精确匹配版本、任务 ID 和输入长度，再验64位十六进制摘要；已知错误也必须匹配身份。 signal.addEventListener('abort',abort,{once:true});worker.onmessage=event=>{const value=event.data;if(value?.version===1&&value.id===id&&value.byteLength===bytes.size&&value.error==='digest_worker_failed'){finish(fault('digest_worker_failed'));return;}if(!value||value.version!==1||value.id!==id||value.byteLength!==bytes.size||typeof value.sha256!=='string'||!(/^[a-f0-9]{64}$/.test(value.sha256)))finish(fault('invalid_digest_response'));else finish(undefined,value.sha256);};worker.onerror=event=>{event.preventDefault();finish(fault('digest_worker_failed'));};worker.onmessageerror=()=>finish(fault('invalid_digest_response')); // Blob 通过结构化克隆传递，不在 UI 主线程先读取整份 arrayBuffer。 diff --git a/src/lib/canvas-runtime/digest-pool.ts b/src/lib/canvas-runtime/digest-pool.ts index 7c98dfc..c4db7e9 100644 --- a/src/lib/canvas-runtime/digest-pool.ts +++ b/src/lib/canvas-runtime/digest-pool.ts @@ -8,7 +8,9 @@ import {importSourceDigest} from './import-integrity.ts'; /** 可替换摘要端口，输出完整 Blob 的小写 SHA256；close 终止本池，不清业务数据或外部回执。 */ export type DigestPort={digest(bytes:Blob,signal?:AbortSignal):Promise;close():void}; /** Read-only compute scheduling: bounded active slots + FIFO Map, never durable operation facts. */ -/** 默认2并发/64等待，上限8并发；可注入 Worker/native/测试执行器，默认使用分块摘要。 */ +/** 默认 2 并发/64 等待，上限 8 并发；可注入 execute，默认 importSourceDigest 分块计算。 + * maxQueued 只计算 waiting，不含 active；不限制 Blob 大小/MIME，不做相同文件去重或结果缓存。 + * execute 负责计算真实完整摘要，池仅用正则检查输出格式，不能证明输出确实对应输入字节。 */ export function createDigestPool(options:{execute?:(bytes:Blob,signal:AbortSignal)=>Promise;concurrency?:number;maxQueued?:number}={}):DigestPort{ const concurrency=options.concurrency??2,maxQueued=options.maxQueued??64,execute=options.execute??importSourceDigest; if(!Number.isSafeInteger(concurrency)||concurrency<1||concurrency>8||!Number.isSafeInteger(maxQueued)||maxQueued<1)throw fault('invalid_digest_budget'); @@ -16,9 +18,13 @@ export function createDigestPool(options:{execute?:(bytes:Blob,signal:AbortSigna type Task={id:number;bytes:Blob;controller:AbortController;resolve:(value:string)=>void;reject:(error:unknown)=>void;cleanup:()=>void;done:boolean}; // Map 既保留入队次序，又可在等待期取消时 O(1) 删除；active 保留真实未结束执行的容量占用。 const waiting=new Map(),active=new Map();let next=0,closed=false; - /** 只 settle 一次并移除等待项/监听；故意不在这里 active.delete，迟到结果不能二次采纳。 */ + /** 只 settle 一次并移除 waiting/外部监听，不在这里 active.delete，迟到结果不能二次采纳。 + * error 用真值分支，execute 应以 Error 等真值拒绝；falsy rejection 不保证被当成失败处理。 */ function finish(task:Task,error?:unknown,value?:string){if(task.done)return;task.done=true;task.cleanup();waiting.delete(task.id);if(error)task.reject(error);else task.resolve(value!);} - /** 有空槽就取最早等待项，先登记 active 再微任务执行；结果校验/取消后 finally 归还真实槽。 */ + /** 从 Map 首项 FIFO 取任务，先占 active 再微任务调用 execute；启动前、结果返回后均复查 abort。 + * execute 的拒绝由第二个 then 回调处理，结果校验抛错再由 catch 处理，最终统一归还槽位并泵下一项。 + * 正则会按 JS 规则强制转字符串，未另做 typeof 检查；注入执行器应遵守 Promise 合同。 + * 无视 abort 且不 settle 的执行器会一直占槽；池没有超时或强制终止通用执行器的能力。 */ function pump(){ if(closed)return; while(active.size=concurrency&&waiting.size>=maxQueued)return Promise.reject(fault('digest_capacity')); return new Promise((resolve,reject)=>{const task:Task={id:++next,bytes,controller:new AbortController(),resolve,reject,cleanup:()=>signal?.removeEventListener('abort',abort),done:false},abort=()=>{task.controller.abort();finish(task,fault('import_aborted'));pump();};signal?.addEventListener('abort',abort,{once:true});waiting.set(task.id,task);pump();}); }, - /** 幂等关闭，取消等待/活跃任务的调用结果；无视 signal 的执行仍由其 Promise 收尾后释放 active。 */ + /** 幂等关闭并取消 waiting/active 的调用结果，停止接受新任务；不返回 drain Promise。 + * 无视 signal 的执行仍由自身 Promise 收尾后释放 active，close 返回不证明底层计算已停止。 */ close(){if(closed)return;closed=true;for(const task of [...waiting.values(),...active.values()]){task.controller.abort();finish(task,fault('import_aborted'));}}, }); } diff --git a/src/lib/canvas-runtime/digest-worker-source.ts b/src/lib/canvas-runtime/digest-worker-source.ts index 801dcc6..3998549 100644 --- a/src/lib/canvas-runtime/digest-worker-source.ts +++ b/src/lib/canvas-runtime/digest-worker-source.ts @@ -11,7 +11,10 @@ export const nativeDigestByteLimit=64*1024*1024; /** Dedicated Worker only: whole-buffer native hashing never moves onto the UI thread. * The host still owns cancellation by terminating the request's Worker. */ -/** 验证 Blob，按能力/大小选 native 或增量实现，均返回相同的完整原字节摘要。 */ +/** 验证 Blob，SubtleCrypto 存在且大小 ≤64MiB 时读取整个 ArrayBuffer 做 SHA-256，否则流式计算。 + * 原生读/算失败直接传播，不自动切到流式重试；注入 subtle 应遵守原生 digest 的 ArrayBuffer 合同。 + * 本函数无 signal，取消靠宿主终止 Worker；代码本身不检测当前线程，Worker 限制由装配调用方遵守。 + * 阈值只限制此输入的整缓冲分配，未计 Blob/原生算法内部内存及其它并发 Worker。 */ export async function digestWorkerSource(bytes:Blob,subtle:Pick|null|undefined=globalThis.crypto?.subtle):Promise{ if(!(bytes instanceof Blob))throw fault('invalid_digest_input'); // native digest 必须返回32字节；输出固定64字符十六进制，不接受变长响应。 @@ -20,6 +23,7 @@ export async function digestWorkerSource(bytes:Blob,subtle:Pickvalue.toString(16).padStart(2,'0')).join(''); } - // Worker fallback 每块1MiB，只做微任务让出；UI 让出策略由 UI fallback 的端口另行决定。 + // 每块 1MiB，块间 yieldTask 仅完成一个 Promise 微任务，不保证 Worker 消息队列获得宏任务机会。 + // UI fallback 用 importSourceDigest 默认定时器让出；不要把这里的微任务策略移作 UI 响应性承诺。 return importSourceDigest(bytes,undefined,{chunkBytes:1024*1024,yieldTask:()=>Promise.resolve()}); } diff --git a/src/lib/canvas-runtime/digest.worker.ts b/src/lib/canvas-runtime/digest.worker.ts index 37b3825..378c3d7 100644 --- a/src/lib/canvas-runtime/digest.worker.ts +++ b/src/lib/canvas-runtime/digest.worker.ts @@ -8,7 +8,9 @@ import {digestWorkerSource} from './digest-worker-source.ts'; const scope=globalThis as unknown as {onmessage:((event:MessageEvent)=>void)|null;postMessage(value:unknown):void}; /** 首条有效请求被接受后保持 true，直到整个 Worker 被宿主终止。 */ let accepted=false; -/** 校验协议和请求后启动只读摘要；成功/失败都回显输入 ID 与字节数，供宿主核对。 */ +/** 首条有效 version=1/正安全整数 ID/Blob 设置 accepted，之后包括失败后的重复消息都忽略。 + * 无效请求直接忽略，不返回错误；这里不检查 MIME/大小或资源身份，不拥有持久任务。 + * 算法拒绝会尝试回稳定错误；成功回信自身抛错不由同一个 then 的失败回调捕获，宿主仍需错误/取消收尾。 */ scope.onmessage=event=>{ const request=event.data;if(accepted||!request||request.version!==1||!Number.isSafeInteger(request.id)||request.id<1||!(request.bytes instanceof Blob))return;accepted=true; void digestWorkerSource(request.bytes).then(sha256=>scope.postMessage({version:1,id:request.id,byteLength:request.bytes.size,sha256}),()=>scope.postMessage({version:1,id:request.id,byteLength:request.bytes.size,error:'digest_worker_failed'})); diff --git a/src/lib/canvas-runtime/import-integrity.ts b/src/lib/canvas-runtime/import-integrity.ts index b450383..21adb0f 100644 --- a/src/lib/canvas-runtime/import-integrity.ts +++ b/src/lib/canvas-runtime/import-integrity.ts @@ -5,13 +5,17 @@ import { fault } from '../canvas-core/values.ts'; import {createSha256} from './sha256.ts'; /** Complete-source SHA-256, with bounded chunk memory and cooperative fallback scheduling. */ -/** 默认64KiB分块（配置上限4MiB），每块读前/读后检查取消，并在块间合作让出。 */ +/** 顺序 slice 读取完整 Blob，默认 64KiB 分块（配置上限 4MiB），空文件也返回合法摘要。 + * 每块读前/读后及 yield 后检查取消；原生 arrayBuffer/yield 本身不被信号强制终止。 + * 默认块间 setTimeout(0) 让出，只有仍有下一块才 yield；自定义 yieldTask 决定实际调度公平性。 + * 块内 hash.update 为同步 CPU 工作，不会中途取消；读/让出错误直接传播，不返回部分摘要。 */ export async function importSourceDigest(bytes:Blob,signal?:AbortSignal,options:{chunkBytes?:number;yieldTask?:()=>Promise}={}):Promise{ const chunk=options.chunkBytes??64*1024,yieldTask=options.yieldTask??(()=>new Promise(resolve=>setTimeout(resolve,0))); if(!(bytes instanceof Blob)||!Number.isSafeInteger(chunk)||chunk<1||chunk>4*1024*1024)throw fault('invalid_digest_input'); /** 读/计算/让出边界统一检查 signal；异步读取即使无法中止，迟到字节也不再参与有效结果。 */ const check=()=>{if(signal?.aborted)throw fault('import_aborted');};check();const hash=createSha256(); - // 更新增量状态后释放对本块的局部引用；空文件直接 hex 得到标准空输入摘要。 + // 只顺序持有当前块，hash 自身不累计输入；part 在本次迭代/yield 期间仍可存在，实际释放时机由 GC 决定。 + // 空文件跳过读取/让出，最后 check 后直接 hex；有界分块不是整个应用进程的内存硬上限。 for(let cursor=0;cursor(n>>>bits)|(n<<(32-bits)); /** Incremental bytes only; owns at most a partial block, state and message schedule. */ -/** 新建单次摘要状态；允许多次 update，hex 终结后只可重复读结果，不能继续追加。 */ +/** 新建单次摘要状态，闭包拥有可变 state/block/words；冻结 API 不代表内部计算状态不变。 + * update 可多次追加，hex 首次终结后缓存结果，不能继续追加；不支持复制状态或导出恢复游标。 + * 纯同步计算，无异步让出/取消；分块大小与调度由读取 Blob 的调用方控制。 */ export function createSha256(){ // length 计原输入字节；pending 计尾块已占字节；result 缓存最终十六进制并标记终结。 const state=new Uint32Array([0x6a09e667,0xbb67ae85,0x3c6ef372,0xa54ff53a,0x510e527f,0x9b05688c,0x1f83d9ab,0x5be0cd19]),block=new Uint8Array(64),words=new Uint32Array(64);let pending=0,length=0,result:string|undefined; @@ -35,7 +37,9 @@ export function createSha256(){ const values=[a,b,c,d,e,f,g,h];for(let i=0;i<8;i++)state[i]=(state[i]+values[i])>>>0; } return Object.freeze({ - /** 先补上次尾块，再就地压缩输入完整块，最后只复制不足64字节的尾部。 */ + /** 先补尾块，再直接读取输入完整块压缩，最后只复制不足 64 字节的尾部，不保留输入引用。 + * 不修改调用方字节；累计长度必须保持安全整数，终结后即使追加空数组也拒绝。 + * 这里的就地指同步消费输入，不代表每次 update 只处理固定工作量或自动让出主线程。 */ update(bytes:Uint8Array){ if(result!==undefined)throw fault('digest_finalized');if(!(bytes instanceof Uint8Array)||!Number.isSafeInteger(length+bytes.byteLength))throw fault('invalid_digest_input');length+=bytes.byteLength;let offset=0; if(pending){const count=Math.min(64-pending,bytes.byteLength);block.set(bytes.subarray(0,count),pending);pending+=count;offset=count;if(pending===64){compress(block,0);pending=0;}}