fix(fel): 修复 Action 执行时的偶发报错

问题摘要 / Issue Summary

执行 Github Action 偶现报错，重试后成功

版本信息 / Version Information

v3.5.0-M6

浏览器 / Browser

No response

操作系统 / Operating System

None

发生了什么？/ What happened?

有 FEL 单测在 Action 过程中报错，重试后成功

期望的行为 / Expected Behavior

不应该出现偶现的报错

重现步骤 / Steps to Reproduce

可能无法重现，需要通过上述的截图日志进行代码分析

额外信息 / Additional Context

No response

确认事项 / Confirmations

我已经搜索了现有的issues，确认这不是重复问题
I have searched existing issues and confirmed this is not a duplicate
我已经阅读了项目文档和 FAQ
I have read the project documentation and FAQ
我提供的信息准确完整
The information I provided is accurate and complete
我愿意协助测试问题修复 (可选)
I'm willing to help test the bug fix (optional)

本地执行单测也出现了偶现报错：

@loveTsong

问题分析

经过代码分析，这是一个 并发竞态条件（Race Condition） 问题。

根本原因

问题出现在 runnableParallel 的 fork/join 操作中：

错误堆栈：

java.lang.NullPointerException: Cannot read field "values" because "other" is null
	at modelengine.fel.core.util.Tip.merge(Tip.java:121)
	at modelengine.fel.engine.activities.AiStart.lambda$runnableParallel$17(AiStart.java:596)

触发流程：

在 Fork.java:96-104 的 join 实现中，当并行分支还未全部完成时，会返回 null：

if (acc.second() == forkNumber.get()) {
    accs.remove(key);
    return acc.first();
} else {
    return null;  // ← 问题所在
}

这个 null 在某些竞态情况下会被传递到 AiStart.java:596 的 reduce 函数中：

aiFork.join(Tip::new, (acc, data) -> {
    acc.merge(data);  // ← data 可能为 null
    return acc;
});

Tip.merge(null) 尝试访问 null.values，导致 NPE

为什么是偶现：只有在特定的线程调度时序下，中间的 null 返回值才会被错误传递给 reduce 函数。

解决方案

方案 1：在 `Tip.merge` 中添加空值检查（防御性编程）

修改 framework/fel/java/fel-core/src/main/java/modelengine/fel/core/util/Tip.java:120-122：

public Tip merge(Tip other) {
    if (other == null) {
        return this;
    }
    return this.addAll(other.values);
}

方案 2：在 `AiStart.runnableParallel` 中添加保护

修改 framework/fel/java/fel-flow/src/main/java/modelengine/fel/engine/activities/AiStart.java:595-598：

AiState<Tip, D, Tip, RF, F> state = aiFork.join(Tip::new, (acc, data) -> {
    if (data != null) {
        acc.merge(data);
    }
    return acc;
});

方案 3：修复 `Fork.join` 的底层逻辑（根本修复）

需要审查 framework/waterflow/java/waterflow-core/src/main/java/modelengine/fit/waterflow/domain/states/Fork.java 中的 join 实现，确保中间状态的 null 不会被传递给 reduce 函数。

深入分析更新

感谢 @loveTsong 的说明！基于您的反馈，我重新分析了问题的根本原因。

问题本质

根据代码追踪，完整的执行流程是：

Fork.java:96-104 中的 wrapper 处理器在收集分支结果时：
- 当 acc.second() == forkNumber.get() 时，说明所有分支都完成，返回聚合结果
- 否则返回 null，表示"中间状态，还在等待其他分支"
JoinNode.java:60-62 应该通过 postFilter 过滤掉未收集齐的数据
问题发生点：在某些并发时序下，wrapper.process() 返回的中间状态 null 没有被正确过滤，而是被传递到了用户提供的 reduce 函数，导致 Tip.merge(null) 触发 NPE

疑问点

想向 @loveTsong 确认：

中间 null 为什么会泄漏到 reduce 函数？ 按设计，JoinNode 的 postFilter 应该只在收集齐所有分支后才处理数据，是否存在某些边界条件？
Fork.java:107-109 的实现是否符合预期？ 每个分支都 subscribe 到同一个 JoinNode，postFilter 是否在所有并发场景下都能正确工作？

修复方案建议

基于完整分析，我建议采用双重防护策略，快速解决当前偶现问题，然后作为长期优化项审查底层架构。

@loveTsong 您觉得这个分析是否正确？倾向于哪种修复方案？

具体修复方案

方案 1：在 `Tip.merge` 中添加空值检查（防御性编程）

修改 framework/fel/java/fel-core/src/main/java/modelengine/fel/core/util/Tip.java:120-122：

public Tip merge(Tip other) {
    if (other == null) {
        return this;  // 防御性处理中间状态的 null
    }
    return this.addAll(other.values);
}

优点：最小化风险，避免类似问题扩散到其他使用 Tip.merge 的地方

方案 2：在 `AiStart.runnableParallel` 中添加保护

修改 framework/fel/java/fel-flow/src/main/java/modelengine/fel/engine/activities/AiStart.java:595-598：

AiState<Tip, D, Tip, RF, F> state = aiFork.join(Tip::new, (acc, data) -> {
    if (data != null) {  // 明确过滤中间状态的 null
        acc.merge(data);
    }
    return acc;
});

优点：明确处理业务逻辑中的 null 情况，符合防御性编程原则

方案 3（长期优化）：审查 Fork/Join 底层逻辑

深入审查：

JoinNode 的 postFilter 在高并发下的行为
wrapper.process() 返回 null 的语义是否被正确处理
是否需要引入 Optional<R> 或其他机制来区分"中间状态"和"有效的 null 值"

推荐方案：先实施方案 1 + 方案 2 快速修复，然后方案 3 作为长期优化项。

chore: 添加完整的诊断输出以追踪 NPE 根本原因

fix: 修复 Fork.join() 并发场景下的 null 数据处理 (#247)

🤖 任务进度更新

任务ID: TASK-20251229-090809
更新时间: 2026-01-12 21:15:00
当前状态: ✅ 代码实现完成，GitHub Actions 验证通过

✅ 已完成

问题复现 - 2025-12-27
- Run 20535263656 确认问题可复现
- 失败率约 0.1-0.2%（1000 次中 1-2 次）
诊断数据收集 - 2025-12-28
- 在 Fork.java、AiStart.java、Tip.java 添加 System.err 诊断输出
- Run 20552158247 成功收集诊断数据
根因分析 - 2025-12-29
- 确认根因：并发场景下 Fork.join() 接收到 input.getData() = null
- 失败时执行在 main 线程而非线程池
- 同一个 Fork key 出现重复执行
修复方案设计 - 2025-12-29
- 选择方案 C（组合方案）
- 阶段 1：智能 null 处理 + 清理诊断代码
代码实现 - 2026-01-12
- 提交 150abf15：添加智能 null 处理
- 提交 4a8a1b2a：改进逻辑，不提前结束聚合
GitHub Actions 验证 - 2026-01-12
- Run 20920275245：1003 次测试全部通过（0 失败，0 错误）

📋 修复策略

核心修改 (Fork.java):

// 当 inputData 为 null 时：
// 1. 记录警告日志
// 2. 返回 null（表示聚合未完成）
// 3. 等待正确的数据到来
if (inputData == null) {
    LOG.warn("[Fork.join] Received null FlowContext.data, skipping...");
    return null;  // 不提前结束聚合，等待有效数据
}

关键改进：

原始修复会在最后一个分支时返回已有数据，导致聚合提前完成
改进后：无论是否是最后一个分支，都返回 null 等待有效数据
确保只有所有分支都有有效数据时才完成聚合

📂 修改的文件

文件	修改内容
`Fork.java`	添加 Logger，智能 null 处理
`AiStart.java`	移除诊断代码，简化 reducer
`Tip.java`	移除诊断代码，保留 null 检查

🎯 下一步

代码审查
关闭此 Issue
后续观察：确保连续多次构建稳定

📊 相关链接

成功的 GitHub Actions Run: https://github.com/ModelEngine-Group/fit-framework/actions/runs/20920275245
任务文件: .ai-workspace/active/TASK-20251229-090809/

由 Claude Code 自动生成

📋 问题根因分析与修复详情

1. 问题现象

PatternTest.shouldOkWhenAiFlowWithExampleSelector 测试在 GitHub Actions 中偶发失败：

失败率: 约 0.1-0.2%（1000 次中 1-2 次）
错误类型: NullPointerException 或 IllegalStateException: Required parameters are missing
特点: 本地无法复现，仅在 CI 环境出现

2. 根本原因

确认的根因: 在并发场景下，Fork.join() 的 reducer 接收到 input.getData() = null

关键证据（诊断输出）

// 正常情况：两个分支都在线程池执行，数据正常
[Fork] key=xxx, thread=flow-node-thread-pool-thread-5, branchCount=0/2, input.getData={question=1+2} ✅
[Fork] key=xxx, thread=flow-node-thread-pool-thread-5, branchCount=1/2, input.getData={examples=...} ✅

// 失败情况：第二个分支在 main 线程执行，数据为 null
[Fork] key=xxx, thread=flow-node-thread-pool-thread-6, branchCount=0/2, input.getData={question=1+2} ✅
[Fork] key=xxx, thread=main, branchCount=1/2, input.getData=null ❌ <-- 问题点

问题根因分析

线程异常: 失败时第二个分支在 main 线程执行，而非线程池
数据为 null: FlowContext.data 字段本身为 null（不是 Fork 设计问题）
重复执行: 同一个 Fork key 出现两次 branchCount=0/2，第二次才有正确数据
时序错乱: null 数据的处理先完成，导致聚合提前结束，正确数据到来时已太晚

3. 修复方案

策略: 当检测到 null 数据时，不更新分支计数，等待正确数据到来

修改前的代码逻辑

// Fork.java - 原始逻辑
O inputData = input.getData();
R processedResult = processor.process(acc.first(), inputData); // inputData 可能为 null，导致 NPE
acc = Tuple.from(processedResult, acc.second() + 1);
// ...

修改后的代码逻辑

// Fork.java - 修复后
O inputData = input.getData();
if (inputData == null) {
    // 返回 null 表示聚合未完成，等待有效数据
    return null;
}
R processedResult = processor.process(acc.first(), inputData);
acc = Tuple.from(processedResult, acc.second() + 1);
// ...

关键点

当 inputData 为 null 时，不更新分支计数
返回 null 表示聚合未完成
等待正确的数据到来后正常完成聚合
Tip.merge() 保留 null 检查作为额外防御层

4. 修改的文件

文件	行数变化	修改内容
`Fork.java`	+8/-3	添加 null 数据检测，跳过无效分支
`AiStart.java`	-14	移除诊断代码，简化 reducer 为一行
`Tip.java`	-19	移除诊断代码，保留 null 检查

5. 提交记录

提交	描述
`150abf15`	初始修复：添加智能 null 处理
`4a8a1b2a`	改进：不在最后分支时提前结束聚合

6. 验证结果

GitHub Actions Run: 20920275245
PatternTest: 1003 次测试全部通过（0 失败，0 错误）
耗时: 8分28秒

7. 技术说明

为什么会出现 null 数据？

这是一个复杂的并发竞态条件问题，涉及 waterflow 框架的 FlowContext 创建和传递逻辑。在某些极端情况下：

分支处理被调度到 main 线程
FlowContext 的 data 字段未正确初始化或被清空
导致 input.getData() 返回 null

为什么本地无法复现？

本地环境（macOS, M1）和 CI 环境（Ubuntu, x86_64）的线程调度策略不同
CI 环境的资源竞争更激烈，更容易触发竞态条件
问题的触发概率极低（0.1-0.2%）

修复是否完整？

当前修复是阶段 1（应用层防御），解决了立即的问题：

✅ 避免 NPE 崩溃
✅ 等待正确数据，不丢失分支数据
✅ 保持聚合逻辑正确性

后续可考虑阶段 2（框架层修复）：

使用 ConcurrentHashMap 替代 HashMap
增强 FlowContext 创建时的验证
深入研究 main 线程执行的根本原因

由 Claude Code 自动生成

refactor: 清理代码并优化测试结构

🔧 补充修复：To.java 并发 NPE

新发现的问题

在清理代码后的并发测试中，暴露了 To.java 中的另一个潜在 NPE：

NullPointerException: Cannot invoke "WindowToken.finishConsume()" because "peekedToken" is null
    at To.java:940

问题关系

并发竞态条件
    │
    ├─→ 问题1：FlowContext.data 为 null（原始问题）
    │       └─→ Fork.java 修复：返回 null 跳过无效数据
    │               └─→ 触发 To.java 的 else 分支
    │                       └─→ 问题2：peekedToken 也可能为 null（被暴露的问题）
    │                               └─→ To.java 修复：添加 null 检查

修复提交

提交	描述
`00337d92`	清理代码并优化测试结构
`108872a0`	修复 To.java 中 peekedToken 可能为 null 的问题

验证结果

Run ID: 20921425198
状态: ✅ 成功
PatternTest: 1004 次测试全部通过（含 1000 次并发压力测试）

最终代码修改

文件	修改
`Fork.java`	null 数据检测（简洁版，无注释）
`Tip.java`	null 检查（简洁版，无注释）
`To.java`	peekedToken null 检查
`PatternTest.java`	拆分为单次测试 + 并发压力测试

由 Claude Code 自动生成

fix: 修复 Fork.join() 并发 NPE 问题 (#404)

fix(fel): 修复 Action 执行时的偶发报错 #247

Description

问题摘要 / Issue Summary

版本信息 / Version Information

浏览器 / Browser

操作系统 / Operating System

发生了什么？/ What happened?

期望的行为 / Expected Behavior

重现步骤 / Steps to Reproduce

相关日志 / Relevant Logs

额外信息 / Additional Context

确认事项 / Confirmations

Activity

CodeCasterX commented on Aug 18, 2025

CodeCasterX commented on Oct 6, 2025

问题分析

根本原因

解决方案

方案 1：在 Tip.merge 中添加空值检查（防御性编程）

方案 2：在 AiStart.runnableParallel 中添加保护

方案 3：修复 Fork.join 的底层逻辑（根本修复）

推荐方案

loveTsong commented on Oct 11, 2025

CodeCasterX commented on Oct 11, 2025

深入分析更新

问题本质

疑问点

修复方案建议

CodeCasterX commented on Oct 11, 2025

具体修复方案

方案 1：在 Tip.merge 中添加空值检查（防御性编程）

方案 2：在 AiStart.runnableParallel 中添加保护

方案 3（长期优化）：审查 Fork/Join 底层逻辑

15 remaining items

CodeCasterX commented on Jan 12, 2026

🤖 任务进度更新

✅ 已完成

📋 修复策略

📂 修改的文件

🎯 下一步

📊 相关链接

CodeCasterX commented on Jan 12, 2026

📋 问题根因分析与修复详情

1. 问题现象

2. 根本原因

关键证据（诊断输出）

问题根因分析

3. 修复方案

修改前的代码逻辑

修改后的代码逻辑

关键点

4. 修改的文件

5. 提交记录

6. 验证结果

7. 技术说明

为什么会出现 null 数据？

为什么本地无法复现？

修复是否完整？

CodeCasterX commented on Jan 13, 2026

🔧 补充修复：To.java 并发 NPE

新发现的问题

问题关系

修复提交

验证结果

最终代码修改

Metadata

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

Issue actions

方案 1：在 `Tip.merge` 中添加空值检查（防御性编程）

方案 2：在 `AiStart.runnableParallel` 中添加保护

方案 3：修复 `Fork.join` 的底层逻辑（根本修复）

方案 1：在 `Tip.merge` 中添加空值检查（防御性编程）

方案 2：在 `AiStart.runnableParallel` 中添加保护