Skip to content

Commit f93040e

Browse files
committed
fix: make IpcPluginAdapter::init idempotent (double init killed the plugin process)
`AdapterManager::start()` calls `init()` *after* the factory has already done so. For built-in adapters that is just config parsing twice; for the out-of-process adapter the second `init` spawned a *second* child process, and the first one was killed via `kill_on_drop`. The dead process's reader task then hit EOF and pushed "plugin process exited" into the pending-response map, waking the new handshake request — so the plugin could never start in a real host: Auto-enabling adapter 'easybot-hello-adapter' WARN Failed adapters: [("easybot-hello-adapter", "Internal error: plugin handshake failed: plugin process exited")] Fixes: - `init` is now idempotent: reuse a live child instead of re-spawning. - The pending-request map is replaced on every spawn (generation isolation), so a dead process's reader only drains its own generation. - Regression test `ipc_adapter_init_is_idempotent`. Only the production path (factory + AdapterManager) was affected, which is why the unit tests and `ProcessPluginTestHost` (which call `init` once) did not catch it. Verified end to end: install `EasyIndie/easybot-hello-adapter` v0.2.0 from the market, start the host -> adapter connects, `POST /messages/send` round-trips, and the message is persisted.
1 parent 67f203b commit f93040e

3 files changed

Lines changed: 65 additions & 5 deletions

File tree

‎CHANGELOG.md‎

Lines changed: 20 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,25 @@
11
## [Unreleased]
22

3+
## [Unreleased]
4+
5+
## [0.0.43] - 2026-10-05
6+
7+
### Fixed
8+
9+
- **进程外插件在真实宿主中无法启动(`plugin handshake failed: plugin process
10+
exited`)** — `AdapterManager::start()` 在调用 `AdapterFactory` 之后会**再调用一次
11+
`init()`**,而工厂内部也已经调用过 `init()`。对内置适配器这只是重复解析配置(无害),
12+
但对进程外插件意味着**第二次 `init` 会 spawn 第二个子进程**,第一个被
13+
`kill_on_drop` 杀掉;已退出进程的 reader 任务在 EOF 时会把「进程已退出」错误投递给
14+
新握手的请求,导致插件永远无法启动。修复:
15+
- `IpcPluginAdapter::init` 改为**幂等**:子进程存活时只复用,不再重复 spawn;
16+
- 未决请求表改为**按 spawn 换代**(每次 spawn 换新的 `pending` 表),已退出进程的
17+
reader 只清理自己那一代,不会误伤新进程的请求;
18+
- 新增回归测试 `ipc_adapter_init_is_idempotent`。
19+
20+
该缺陷仅影响**真实宿主启动路径**(`AdapterFactory` + `AdapterManager` 双重 init),
21+
单测与 `ProcessPluginTestHost` 直接调用 `init` 因而未暴露。
22+
323
## [0.0.42] - 2026-10-04
424

525
### Added

‎crates/easybot-core/src/plugin/ipc.rs‎

Lines changed: 26 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -58,7 +58,9 @@ pub struct IpcPluginAdapter {
5858
state: Mutex<AdapterState>,
5959
child: tokio::sync::Mutex<Option<Child>>,
6060
stdin: tokio::sync::Mutex<Option<ChildStdin>>,
61-
pending: Arc<PendingMap>,
61+
/// 未决请求表。每次 spawn 替换一代:已退出进程的 reader 只清理它自己那一代,
62+
/// 不会把过期错误投递给新进程的请求。
63+
pending: Mutex<Arc<PendingMap>>,
6264
next_id: AtomicU64,
6365
event_bus: Option<Arc<EventBus>>,
6466
call_timeout: Duration,
@@ -93,7 +95,7 @@ impl IpcPluginAdapter {
9395
state: Mutex::new(AdapterState::Created),
9496
child: tokio::sync::Mutex::new(None),
9597
stdin: tokio::sync::Mutex::new(None),
96-
pending: Arc::new(Mutex::new(HashMap::new())),
98+
pending: Mutex::new(Arc::new(Mutex::new(HashMap::new()))),
9799
next_id: AtomicU64::new(1),
98100
event_bus: None,
99101
call_timeout: DEFAULT_CALL_TIMEOUT,
@@ -122,8 +124,9 @@ impl IpcPluginAdapter {
122124
/// 发起一次请求并等待响应(带超时)。
123125
async fn call(&self, method: &str, params: Value) -> Result<Value, String> {
124126
let id = self.next_id.fetch_add(1, Ordering::SeqCst);
127+
let pending = self.pending.lock().unwrap().clone();
125128
let (tx, rx) = oneshot::channel();
126-
self.pending.lock().unwrap().insert(id, tx);
129+
pending.lock().unwrap().insert(id, tx);
127130

128131
let request = Request::new(id, method, params);
129132
{
@@ -143,7 +146,7 @@ impl IpcPluginAdapter {
143146
Ok(Ok(Err(err))) => Err(err),
144147
Ok(Err(_)) => Err("plugin response channel closed".to_string()),
145148
Err(_) => {
146-
self.pending.lock().unwrap().remove(&id);
149+
pending.lock().unwrap().remove(&id);
147150
Err(format!("plugin call '{method}' timed out"))
148151
}
149152
}
@@ -168,6 +171,11 @@ impl IpcPluginAdapter {
168171
command.env(key, value);
169172
}
170173

174+
// 新一代 pending + 复位存活标志(重连/重启场景)
175+
let pending: Arc<PendingMap> = Arc::new(Mutex::new(HashMap::new()));
176+
*self.pending.lock().unwrap() = pending.clone();
177+
self.child_exited.store(false, Ordering::SeqCst);
178+
171179
let mut child = command.spawn().map_err(|e| {
172180
GatewayError::Internal(format!(
173181
"failed to spawn plugin '{}': {e}",
@@ -189,7 +197,7 @@ impl IpcPluginAdapter {
189197

190198
spawn_reader(
191199
stdout,
192-
self.pending.clone(),
200+
pending,
193201
self.event_bus.clone(),
194202
self.platform.clone(),
195203
self.child_exited.clone(),
@@ -269,6 +277,19 @@ impl PlatformAdapter for IpcPluginAdapter {
269277
}
270278

271279
async fn init(&mut self, config: AdapterConfig) -> Result<InitResult, GatewayError> {
280+
// 幂等:宿主会先经 AdapterFactory、再由 `AdapterManager::start` 调用 `init`。
281+
// 子进程已在运行时不重复 spawn —— 否则第二个进程会替换第一个(`kill_on_drop`
282+
// 杀掉它),而旧进程的 reader 在 EOF 时会把「进程已退出」投递给新握手的请求。
283+
let already_running = {
284+
let child = self.child.lock().await;
285+
child.is_some() && !self.child_exited()
286+
};
287+
if already_running {
288+
return Ok(InitResult {
289+
ok: true,
290+
error: None,
291+
});
292+
}
272293
self.spawn_process().await?;
273294

274295
// 1) 握手 + 协议版本校验

‎crates/easybot-core/tests/ipc_adapter.rs‎

Lines changed: 19 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -193,6 +193,25 @@ async fn ipc_adapter_passes_home_env_to_plugin() {
193193
let _ = adapter.disconnect().await;
194194
}
195195

196+
#[tokio::test]
197+
async fn ipc_adapter_init_is_idempotent() {
198+
// 回归:宿主会先经 AdapterFactory、再由 `AdapterManager::start` 调用 `init`。
199+
// 若第二次 init 重新 spawn,第二个进程会替换第一个(kill_on_drop 杀掉它),
200+
// 旧进程的 reader 在 EOF 时会把「plugin process exited」投递给新握手的请求,
201+
// 导致插件永远启动失败("plugin handshake failed: plugin process exited")。
202+
let mut adapter = IpcPluginAdapter::new(find_ipc_mock_plugin(), "ipc-mock", "IPC Mock", None);
203+
adapter.init(test_config()).await.expect("first init");
204+
adapter
205+
.init(test_config())
206+
.await
207+
.expect("second init 必须成功(幂等,不重启进程)");
208+
assert!(!adapter.child_exited(), "重复 init 不应导致子进程退出");
209+
210+
let conn = adapter.connect().await.expect("connect after double init");
211+
assert!(conn.ok, "重复 init 后 connect 仍应成功");
212+
let _ = adapter.disconnect().await;
213+
}
214+
196215
#[tokio::test]
197216
async fn ipc_adapter_detects_crashed_process() {
198217
// 插件在 connect 后主动退出(模拟崩溃):适配器须报告 Down,供健康监测器重连。

0 commit comments

Comments
 (0)