fix(builder): resolve v8 serialization crash in cluster worker pool (#230)

Author: Lyxot

packages/builder/src/runAsWorker.ts

@@ -1,10 +1,13 @@
+import { join } from 'node:path'
 import process from 'node:process'
+import { fileURLToPath } from 'node:url'
 import { deserialize } from 'node:v8'
 
 import type { PhotoManifestItem } from '@afilmory/typing'
 
 import type { BuilderOptions } from './builder/builder.js'
 import { AfilmoryBuilder } from './builder/builder.js'
+import { loadBuilderConfig } from './config/index.js'
 import type { PluginRunState } from './plugins/manager.js'
 import type { StorageObject } from './storage/interfaces'
 import type { BuilderConfig } from './types/config.js'
@@ -40,6 +43,17 @@ export async function runAsWorker() {
   let builder: AfilmoryBuilder
   let pluginRunState: PluginRunState
 
+  // 安全发送消息到主进程（防止 EPIPE 错误）
+  const safeSend = (message: unknown) => {
+    try {
+      if (process.send && process.connected) {
+        process.send(message)
+      }
+    } catch {
+      // 主进程已关闭 IPC 通道，静默忽略
+    }
+  }
+
   // 初始化函数，从主进程接收共享数据
   const initializeWorker = async (serializedData: WorkerInitMessage['sharedData']) => {
     if (isInitialized) return
@@ -52,7 +66,18 @@ export async function runAsWorker() {
     imageObjects = sharedData.imageObjects
     existingManifestMap = sharedData.existingManifestMap
     livePhotoMap = sharedData.livePhotoMap
-    builder = new AfilmoryBuilder(sharedData.builderConfig)
+
+    // 主进程序列化时移除了 plugins（函数无法序列化），
+    // 从配置文件重新加载以获取完整的插件配置
+    const fullConfig = await loadBuilderConfig({
+      cwd: join(fileURLToPath(import.meta.url), '../../../..'),
+    })
+    const builderConfig: BuilderConfig = {
+      ...sharedData.builderConfig,
+      plugins: fullConfig.plugins,
+    }
+
+    builder = new AfilmoryBuilder(builderConfig)
     await builder.ensurePluginsReady()
     pluginRunState = builder.createPluginRunState()
 
@@ -133,9 +158,7 @@ export async function runAsWorker() {
         result,
       }
 
-      if (process.send) {
-        process.send(response)
-      }
+      safeSend(response)
     } catch (error) {
       // 发送错误回主进程
       const response: TaskResult = {
@@ -144,9 +167,7 @@ export async function runAsWorker() {
         error: error instanceof Error ? error.message : String(error),
       }
 
-      if (process.send) {
-        process.send(response)
-      }
+      safeSend(response)
     }
   }
 
@@ -242,9 +263,7 @@ export async function runAsWorker() {
         results,
       }
 
-      if (process.send) {
-        process.send(response)
-      }
+      safeSend(response)
     } catch (error) {
       // 如果批量处理失败，为每个任务发送错误结果
       const results: TaskResult[] = message.tasks.map((task) => ({
@@ -258,9 +277,7 @@ export async function runAsWorker() {
         results,
       }
 
-      if (process.send) {
-        process.send(response)
-      }
+      safeSend(response)
     }
   }
 
@@ -275,19 +292,15 @@ export async function runAsWorker() {
 
       if (message.type === 'ping') {
         // 响应主进程的 ping，表示 worker 已准备好
-        if (process.send) {
-          process.send({ type: 'pong', workerId })
-        }
+        safeSend({ type: 'pong', workerId })
         return
       }
 
       if (message.type === 'init') {
         // 处理初始化消息
         try {
           await initializeWorker(message.sharedData)
-          if (process.send) {
-            process.send({ type: 'init-complete', workerId })
-          }
+          safeSend({ type: 'init-complete', workerId })
         } catch (error) {
           console.error('Worker initialization failed', error)
           process.exit(1)
@@ -323,7 +336,5 @@ export async function runAsWorker() {
   })
 
   // 告知主进程 worker 已准备好
-  if (process.send) {
-    process.send({ type: 'ready', workerId })
-  }
+  safeSend({ type: 'ready', workerId })
 }

packages/builder/src/worker/cluster-pool.ts

@@ -267,11 +267,14 @@ export class ClusterPool<T> extends EventEmitter {
       // 首次准备就绪时发送初始化数据，但不立即标记为 ready
       if (this.sharedData) {
         // 使用 v8.serialize 序列化数据以保持类型完整性
+        // 注意：plugins 可能包含函数（如 hook 回调），v8.serialize 无法序列化函数
+        // 因此在序列化时移除 plugins，worker 进程会从配置文件重新加载
+        const { plugins: _plugins, ...serializableConfig } = this.sharedData.builderConfig
         const serializedBuffer = serialize({
           existingManifestMap: this.sharedData.existingManifestMap,
           livePhotoMap: this.sharedData.livePhotoMap,
           imageObjects: this.sharedData.imageObjects,
-          builderConfig: this.sharedData.builderConfig,
+          builderConfig: { ...serializableConfig, plugins: [] },
         })
 
         // 将 Buffer 转换为数组以通过 IPC 传输
@@ -515,7 +518,7 @@ export class ClusterPool<T> extends EventEmitter {
           const timeout = setTimeout(() => {
             worker.kill('SIGKILL')
             resolve()
-          }, 0)
+          }, 5000)
 
           worker.on('exit', () => {
             clearTimeout(timeout)