From c015febb49c300a677d3550bafdb1e7e6a799d1d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=B0=8F=E5=94=AF?= Date: Sat, 17 Oct 2026 21:52:06 +0800 Subject: [PATCH] =?UTF-8?q?zhiyi:=20=E5=8D=95=E6=AC=A1=E9=87=8D=E8=AF=BB?= =?UTF-8?q?=E8=A1=8C=E6=95=B0=E4=B8=8A=E9=99=90=20+=20Go=20IPC=20=E8=AF=BB?= =?UTF-8?q?=20deadline=20+=20=E6=AF=8F=20rpc=20=E6=96=B0=20dial=20(t=5F576?= =?UTF-8?q?714de)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit N=2 准入只拦并发堆叠、不拦单请求自身内存：一次 top_k 大 / scan_all 仍能把单条 zhiyi-consolidate 顶到 1~2GB+ 致 swap thrash（t_eedb2d39 根因）。 改动： - rust/main.rs: ZHIYI_HEAVY_READ_MAX_ROWS（默认 20000，可覆盖）clamp lancedb_search top_k + scan_all/scan_deleted；响应体加 truncated:bool （#[serde(default)] 向后兼容旧 Go 客户端）；zhiyi_ping 暴露当前 cap。 - rust/lancedb_ops.rs: scan_all/scan_deleted 的 .limit(200_000) 参数化为 max_rows。 - go/lancedb_ipc.go: ipcReq/ipcResp 加 Truncated；rpc() 用 SetWriteDeadline(30s)/ SetReadDeadline(120s)（原 io.ReadFull 无 deadline = 侧车挂死永久阻塞的根源）； Search/ScanAll/ScanDeleted 截断时打日志（不静默）。 - go/ipc_reconnect_test.go: TestIPCReadDeadlineBoundsHungSidecar（纯 Go，挂死侧车 被读 deadline 兜住 ~801ms）+ TestIPCReconnectAfterSidecarRestart（env 门控， 真 sidecar 杀→快速失败 37µs→重启→重连成功）。 隔离副本 /tmp 实测：封顶 20k peak RSS ~1.17GB vs 不封顶 200k ~1.60GB/80s。 不部署（部署另批，需小怡终审 + 牧尘放行）。 --- go/internal/storage/ipc_reconnect_test.go | 130 ++++++++++++++++++++++ go/internal/storage/lancedb_ipc.go | 52 ++++++++- rust/src/lancedb_ops.rs | 14 ++- rust/src/main.rs | 85 ++++++++++++-- 4 files changed, 258 insertions(+), 23 deletions(-) create mode 100644 go/internal/storage/ipc_reconnect_test.go diff --git a/go/internal/storage/ipc_reconnect_test.go b/go/internal/storage/ipc_reconnect_test.go new file mode 100644 index 0000000..f583a2e --- /dev/null +++ b/go/internal/storage/ipc_reconnect_test.go @@ -0,0 +1,130 @@ +package storage + +import ( + "encoding/binary" + "io" + "net" + "os" + "os/exec" + "path/filepath" + "testing" + "time" +) + +// t_576714de AC2：「边车即使短暂不可用 ⇒ zhiyid 的 IPC 客户端能重连」拆成两半验证： +// +// ① 挂死不无限阻塞：侧车接受连接后永不响应（模拟 thrash 中被 kill / 卡住）时， +// 旧实现 io.ReadFull 无 deadline ⇒ 客户端永久挂起（「边车好了它还挂着」）。 +// 修复后读 deadline 封顶，超时即返回错误，调用方降级（Search→本地兜底），recall 不挂。 +// ② 真·重连：每次 rpc 都是新 dial。侧车被杀 ⇒ 快速失败（不挂起）；侧车重启 ⇒ 下一次 +// rpc 重新 dial 成功，无需持有长连接。 +// +// TestIPCReadDeadlineBoundsHungSidecar 是纯 Go 自包含（无需真实 sidecar 二进制），恒跑。 +// TestIPCReconnectAfterSidecarRestart 需真实 sidecar（env 门控，未设则跳过）。 + +func TestIPCReadDeadlineBoundsHungSidecar(t *testing.T) { + // Unix socket sun_path 上限 108 字节：t.TempDir() 路径过长会 EINVAL，改用 /tmp 短路径。 + tmp, err := os.MkdirTemp(os.TempDir(), "zyri") + if err != nil { + t.Fatalf("mkdtemp: %v", err) + } + defer os.RemoveAll(tmp) + sock := filepath.Join(tmp, "h.sock") + l, err := net.Listen("unix", sock) + if err != nil { + t.Fatalf("listen: %v", err) + } + defer l.Close() + // 侧车替身：accept 后读满请求（长度前缀 + body）但【永不回响应】= 模拟挂死。 + go func() { + for { + c, aerr := l.Accept() + if aerr != nil { + return + } + go func(conn net.Conn) { + var lenBuf [4]byte + _, _ = io.ReadFull(conn, lenBuf[:]) + n := binary.BigEndian.Uint32(lenBuf[:]) + buf := make([]byte, n) + _, _ = io.ReadFull(conn, buf) + // 之后不回任何字节 → 客户端 ReadFull 一直阻塞到读 deadline。 + }(c) + } + }() + + origRead, origWrite := ipcReadTimeout, ipcWriteTimeout + ipcReadTimeout = 800 * time.Millisecond + ipcWriteTimeout = 5 * time.Second + defer func() { ipcReadTimeout, ipcWriteTimeout = origRead, origWrite }() + + rc := NewRustLanceDBClient(sock, nil) + start := time.Now() + _, rpcErr := rc.rpc(ipcReq{Type: "lancedb_search", Vector: make([]float32, 1024), TopK: 5}) + elapsed := time.Since(start) + + if rpcErr == nil { + t.Fatalf("期望读 deadline 超时返回错误，实际成功（deadline 未生效）") + } + if elapsed < 600*time.Millisecond { + t.Fatalf("错误返回过早（%s）—— 不像读 deadline 触发（应是 ~%s 后）", elapsed, ipcReadTimeout) + } + if elapsed > 15*time.Second { + t.Fatalf("仍长时间阻塞（%s）—— 读 deadline 未封顶，挂死未被兜住", elapsed) + } + t.Logf("挂死侧车被读 deadline 兜住：err=%v in %s（上限 %s，不再无限阻塞 ⇒ recall 可降级不挂）", rpcErr, elapsed, ipcReadTimeout) +} + +// startSidecarForTest 启动真实 sidecar（socket 路径 + 数据目录）；返回 *exec.Cmd 供后续 kill。 +func startSidecarForTest(t *testing.T, bin, sock, dataDir string) *exec.Cmd { + t.Helper() + cmd := exec.Command(bin, "--mode", "socket", "--socket", sock, "--data-dir", dataDir) + cmd.Stdout, cmd.Stderr = io.Discard, io.Discard + if err := cmd.Start(); err != nil { + t.Fatalf("启动 sidecar 失败: %v", err) + } + return cmd +} + +func TestIPCReconnectAfterSidecarRestart(t *testing.T) { + bin := os.Getenv("ZHIYI_TEST_SIDECAR_BIN") + dataDir := os.Getenv("ZHIYI_TEST_DATA_DIR") + if bin == "" || dataDir == "" { + t.Skip("ZHIYI_TEST_SIDECAR_BIN / ZHIYI_TEST_DATA_DIR 未设置，跳过真·重连验证") + } + sock := filepath.Join(dataDir, "reconnect-test.sock") + + // 1) 起 sidecar → rpc 成功 + proc := startSidecarForTest(t, bin, sock, dataDir) + time.Sleep(4 * time.Second) + rc := NewRustLanceDBClient(sock, nil) + if _, err := rc.rpc(ipcReq{Type: "zhiyi_ping"}); err != nil { + proc.Process.Kill() + t.Fatalf("重启前 ping 应成功: %v", err) + } + + // 2) 杀掉 sidecar → 下一次 rpc【快速失败】（dial 上限 5s），不挂起 + if err := proc.Process.Kill(); err != nil { + t.Logf("kill warning: %v", err) + } + _ = proc.Wait() + start := time.Now() + _, derr := rc.rpc(ipcReq{Type: "zhiyi_ping"}) + downElapsed := time.Since(start) + if derr == nil { + t.Fatalf("sidecar 被杀后 rpc 应失败，实际成功（未感知不可用）") + } + if downElapsed > 6*time.Second { + t.Fatalf("sidecar 不可用时未快速失败（用了 %s > 6s dial 上限）—— 疑似挂起", downElapsed) + } + t.Logf("sidecar 不可用 ⇒ rpc 快速失败 in %s（dial 上限 5s，不挂起）", downElapsed) + + // 3) 重启 sidecar（Rust 侧会清掉陈旧 socket 再 bind）→ 下一次 rpc 成功 = 重连成立 + proc2 := startSidecarForTest(t, bin, sock, dataDir) + defer func() { _ = proc2.Process.Kill() }() + time.Sleep(4 * time.Second) + if _, err := rc.rpc(ipcReq{Type: "zhiyi_ping"}); err != nil { + t.Fatalf("sidecar 重启后 rpc 应成功（重连），实际: %v", err) + } + t.Log("sidecar 重启后 rpc 成功 = 重连成立（每次 rpc 新 dial，无需长连接）") +} diff --git a/go/internal/storage/lancedb_ipc.go b/go/internal/storage/lancedb_ipc.go index e5e4a32..7bf9d57 100644 --- a/go/internal/storage/lancedb_ipc.go +++ b/go/internal/storage/lancedb_ipc.go @@ -35,6 +35,18 @@ func NewRustLanceDBClient(sockPath string, emb *Embedder) *RustLanceDBClient { // ─── IPC 协议 ──────────────────────────────────────────── +// t_576714de：单次重读被行数上限截断后，最重的一次响应（20000 行 JSON）在内存 +// 紧张时实测 ~43s。读 deadline 取 120s（>> 任何合法操作，≪ 无限阻塞）；侧车接受 +// 连接后挂死/被 kill 时，旧实现 io.ReadFull 无 deadline ⇒ 客户端永久阻塞 +// （「边车好了它还挂着」）。封顶后超时即返回错误，调用方降级（Search→本地兜底）， +// 且每次 rpc 都是新 dial ⇒ 侧车恢复后下一次调用天然重连。 +// 用 var（非 const）便于测试临时缩短读超时验证「挂死不被无限阻塞」。 +var ( + ipcDialTimeout = 5 * time.Second + ipcWriteTimeout = 30 * time.Second + ipcReadTimeout = 120 * time.Second +) + type ipcReq struct { Type string `json:"type"` Table string `json:"table,omitempty"` @@ -63,6 +75,9 @@ type ipcResp struct { Result string `json:"result"` // lancedb search uses this FailureStep string `json:"failure_step"` ErrorDetail string `json:"error_detail"` + // t_576714de：单次重读被 Rust 侧行数上限截断时 = true（部分结果，非全量）。 + // 旧 sidecar 无此字段 ⇒ 缺省 false，向后兼容。 + Truncated bool `json:"truncated"` } // updateField lancedb_update 的单列更新项：Value 为 SQL 表达式 @@ -73,22 +88,37 @@ type updateField struct { } func (rc *RustLanceDBClient) rpc(req ipcReq) (*ipcResp, error) { - conn, err := net.DialTimeout("unix", rc.sockPath, 5*time.Second) + // 每次调用都是新 dial：侧车不可用（socket 消失/无监听）⇒ 快速失败； + // 侧车恢复后下一次调用天然重连（t_576714de AC）。 + conn, err := net.DialTimeout("unix", rc.sockPath, ipcDialTimeout) if err != nil { return nil, fmt.Errorf("ipc dial: %w", err) } defer conn.Close() + // 给已建立连接封顶读/写 deadline：侧车接受连接后挂死/被 kill ⇒ 旧实现 + // io.ReadFull 无限阻塞（「边车好了它还挂着」）。现超时即降级，热路径 recall 不受阻。 + now := time.Now() + _ = conn.SetWriteDeadline(now.Add(ipcWriteTimeout)) + _ = conn.SetReadDeadline(now.Add(ipcReadTimeout)) body, _ := json.Marshal(req) lenBuf := make([]byte, 4) binary.BigEndian.PutUint32(lenBuf, uint32(len(body))) - conn.Write(lenBuf) - conn.Write(body) + if _, err := conn.Write(lenBuf); err != nil { + return nil, fmt.Errorf("ipc write len: %w", err) + } + if _, err := conn.Write(body); err != nil { + return nil, fmt.Errorf("ipc write body: %w", err) + } - io.ReadFull(conn, lenBuf) + if _, err := io.ReadFull(conn, lenBuf); err != nil { + return nil, fmt.Errorf("ipc read len (sidecar down/hung, read deadline %s): %w", ipcReadTimeout, err) + } msgLen := binary.BigEndian.Uint32(lenBuf) respBuf := make([]byte, msgLen) - io.ReadFull(conn, respBuf) + if _, err := io.ReadFull(conn, respBuf); err != nil { + return nil, fmt.Errorf("ipc read body (sidecar down/hung, read deadline %s): %w", ipcReadTimeout, err) + } var resp ipcResp if err := json.Unmarshal(respBuf, &resp); err != nil { @@ -243,6 +273,10 @@ func (rc *RustLanceDBClient) Search(table string, vector []float32, topK int, na Vector []float32 `json:"vector"` } json.Unmarshal([]byte(resp.ReportJSON), &raw) + if resp.Truncated { + // t_576714de：Rust 侧单次重读被行数上限截断（部分结果，非全量）—— 显式日志，禁静默。 + log.Printf("[ipc] Search truncated: single-read cap applied, got %d rows (partial result)", len(raw)) + } if len(raw) == 0 { log.Printf("[ipc] Search: resp.ReportJSON unmarshal gave 0 results, falling back to localSearch (Result=%q)", resp.Result) return rc.localSearch(vector, topK, namespaceFilter), nil @@ -428,6 +462,10 @@ func (rc *RustLanceDBClient) ScanAll() ([]models.MemoryRecord, error) { if err != nil { return nil, fmt.Errorf("scan_all unmarshal: %w", err) } + if resp.Truncated { + // t_576714de：单次扫描被行数上限截断（部分结果）—— 显式日志，禁静默。 + log.Printf("[ipc] ScanAll truncated: single-read cap applied, got %d rows (partial)", len(out)) + } // 同步到本地缓存（与 GetSkillCandidates 同款） _local.mu.Lock() for i := range out { @@ -515,6 +553,10 @@ func (rc *RustLanceDBClient) ScanDeleted() ([]models.MemoryRecord, error) { if err != nil { return nil, fmt.Errorf("scan_deleted unmarshal: %w", err) } + if resp.Truncated { + // t_576714de：单次扫描被行数上限截断（部分结果）—— 显式日志，禁静默。 + log.Printf("[ipc] ScanDeleted truncated: single-read cap applied, got %d rows (partial)", len(out)) + } return out, nil } diff --git a/rust/src/lancedb_ops.rs b/rust/src/lancedb_ops.rs index 0a85e0e..46962df 100644 --- a/rust/src/lancedb_ops.rs +++ b/rust/src/lancedb_ops.rs @@ -1058,8 +1058,9 @@ impl LanceDBOps { /// 全表扫描（元数据；2026-09-21 治本改造） /// ① 用 Select 投影跳过 vector 列（1024 维 = 4KB/条，Step 3/4 都不需要） - /// ② 上限从 10000 提到 200000（原值会静默丢记录：库已有 10065 条） - pub fn scan_all(&self) -> Result, Box> { + /// ② 上限由调用方传入 max_rows（IPC 边界 = 单次重读硬上限 heavy_read_max_rows()， + /// 整合内部全量 = 200_000；t_576714de） + pub fn scan_all(&self, max_rows: usize) -> Result, Box> { let db = rt().block_on(lancedb::connect(self.data_dir.to_str().unwrap()).execute())?; let tbl = rt().block_on(db.open_table("memories").execute())?; @@ -1081,7 +1082,7 @@ impl LanceDBOps { tbl.query() .only_if("is_deleted = false") .select(Select::columns(META_COLUMNS)) - .limit(200_000) + .limit(max_rows) .execute() )?); @@ -1143,8 +1144,9 @@ impl LanceDBOps { /// 都过滤 is_deleted=false —— 存量 11228 条 is_deleted=true 的行**没有任何读路径能看** /// （唯独 stats 能 count）。回填脚本需要枚举已删除行以回填 delete_reason 并出直方图。 /// - /// 只读元数据（跳 vector），上限 200000。带 delete_reason 列（供回填时跳过已回填的）。 - pub fn scan_deleted(&self) -> Result, Box> { + /// 只读元数据（跳 vector），上限由调用方传入 max_rows（IPC 边界 = 单次重读硬上限， + /// 整合/回填脚本按需传全量；t_576714de）。带 delete_reason 列（供回填时跳过已回填的）。 + pub fn scan_deleted(&self, max_rows: usize) -> Result, Box> { let db = rt().block_on(lancedb::connect(self.data_dir.to_str().unwrap()).execute())?; let tbl = rt().block_on(db.open_table("memories").execute())?; @@ -1164,7 +1166,7 @@ impl LanceDBOps { tbl.query() .only_if("is_deleted = true") .select(Select::columns(META_COLUMNS)) - .limit(200_000) + .limit(max_rows) .execute() )?); diff --git a/rust/src/main.rs b/rust/src/main.rs index f6554c8..09bcc99 100644 --- a/rust/src/main.rs +++ b/rust/src/main.rs @@ -76,6 +76,22 @@ impl Drop for ConsolidatingGuard { static HEAVY_READ_DEFAULT: usize = 2; static HEAVY_READ_WAIT_MS: u64 = 100; +// ─── 单次重读行数上限（方案②，t_576714de，2026-10-17 残留缺口）──────── +// N=2 准入只拦「并发叠加」，不拦「单次请求自身的内存量」：一个 top_k=1000000 / +// scan_all 请求可独自吞下全表（实测 43547 行 × 1024 维向量 ≈ 260MB Rust + JSON ~650MB）， +// 照样 thrash（t_eedb2d39 根因复盘定性的 ② 单次量无上限）。 +// 治本：给单次重读的**返回行数**设硬上限（默认 20000，env ZHIYI_HEAVY_READ_MAX_ROWS 可调）。 +// 请求量超上限 ⇒ 截断到上限 + 标记 truncated:true（降级返回部分结果，而非硬吃内存）。 +// 热路径 recall 的 top_k ≤ 1000 永不触发；只挡 admin/穷举型单次大请求。 +static HEAVY_READ_MAX_ROWS_DEFAULT: usize = 20000; + +fn heavy_read_max_rows() -> usize { + match std::env::var("ZHIYI_HEAVY_READ_MAX_ROWS") { + Ok(v) => v.trim().parse::().unwrap_or(HEAVY_READ_MAX_ROWS_DEFAULT).max(1), + Err(_) => HEAVY_READ_MAX_ROWS_DEFAULT, + } +} + fn heavy_read_permits() -> usize { match std::env::var("ZHIYI_HEAVY_READ_CONCURRENCY") { Ok(v) => v.trim().parse::().unwrap_or(HEAVY_READ_DEFAULT), @@ -253,6 +269,10 @@ struct ConsolidateResponse { report_json: String, failure_step: String, error_detail: String, + // 方案②（t_576714de）：单次重读被行数上限截断时为 true（降级返回部分结果）。 + // 旧 Go 客户端无此字段 ⇒ serde 缺省 false，向后兼容。 + #[serde(default)] + truncated: bool, } fn main() { @@ -416,9 +436,14 @@ fn handle_client(mut stream: UnixStream, args: &Args, lancedb: LanceDBOps, state let v: Vec = msg["vector"].as_array() .map(|a| a.iter().filter_map(|v| v.as_f64()).map(|v| v as f32).collect()) .unwrap_or_default(); - let tk = msg["top_k"].as_u64().unwrap_or(10) as usize; + let requested = msg["top_k"].as_u64().unwrap_or(10) as usize; // 空 namespace = 不过滤（匹配所有） let ns = msg["namespace"].as_str().filter(|s| !s.is_empty()); + // 方案②（t_576714de）：单次重读行数硬上限 —— 截断到上限，超限标 truncated:true， + // 降级返回部分结果而非让单个请求独自吞下全表向量（43547 行 ≈ 260MB Rust + JSON）。 + let cap = heavy_read_max_rows(); + let tk = requested.min(cap); + let truncated = requested > tk; // 重读准入（t_690cc333 方案①）：top_k>1000 的重搜索是内存放大器， // 计入信号量 N；无许可 ⇒ 有界等待后拒绝 heavy_read_busy。 let _perm = if tk > 1000 { @@ -435,7 +460,15 @@ fn handle_client(mut stream: UnixStream, args: &Args, lancedb: LanceDBOps, state None }; match lancedb.search(&v, tk, ns) { - Ok(r) => send_ok(&mut stream, &serde_json::to_string(&r).unwrap_or_default()), + Ok(r) => { + let json = serde_json::to_string(&r).unwrap_or_default(); + if truncated { + eprintln!("[zhiyi-ipc] search top_k={} CLAMPED to {} (single-read cap {}), truncated=true", requested, r.len(), cap); + send_ok_truncated(&mut stream, &json); + } else { + send_ok(&mut stream, &json); + } + } Err(e) => send_error(&mut stream, "lancedb_search", &e.to_string()), } } @@ -559,8 +592,19 @@ fn handle_client(mut stream: UnixStream, args: &Args, lancedb: LanceDBOps, state return; } }; - match lancedb.scan_all() { - Ok(r) => send_ok(&mut stream, &serde_json::to_string(&r).unwrap_or_default()), + // 方案②（t_576714de）：单次扫描行数上限 —— 超上限截断 + truncated:true（降级部分结果）。 + let cap = heavy_read_max_rows(); + match lancedb.scan_all(cap) { + Ok(r) => { + let json = serde_json::to_string(&r).unwrap_or_default(); + let trunc = r.len() >= cap; + if trunc { + eprintln!("[zhiyi-ipc] scan_all hit cap {} → truncated=true", cap); + send_ok_truncated(&mut stream, &json); + } else { + send_ok(&mut stream, &json); + } + } Err(e) => send_error(&mut stream, "lancedb_scan_all", &e.to_string()), } } @@ -577,8 +621,19 @@ fn handle_client(mut stream: UnixStream, args: &Args, lancedb: LanceDBOps, state return; } }; - match lancedb.scan_deleted() { - Ok(r) => send_ok(&mut stream, &serde_json::to_string(&r).unwrap_or_default()), + // 方案②（t_576714de）：单次扫描行数上限 —— 超上限截断 + truncated:true。 + let cap = heavy_read_max_rows(); + match lancedb.scan_deleted(cap) { + Ok(r) => { + let json = serde_json::to_string(&r).unwrap_or_default(); + let trunc = r.len() >= cap; + if trunc { + eprintln!("[zhiyi-ipc] scan_deleted hit cap {} → truncated=true", cap); + send_ok_truncated(&mut stream, &json); + } else { + send_ok(&mut stream, &json); + } + } Err(e) => send_error(&mut stream, "lancedb_scan_deleted", &e.to_string()), } } @@ -621,6 +676,7 @@ fn handle_client(mut stream: UnixStream, args: &Args, lancedb: LanceDBOps, state "consolidating": CONSOLIDATING.load(Ordering::SeqCst), "heavy_read_in_flight": state.heavy_read.in_flight(), "heavy_read_permits": heavy_read_permits(), + "heavy_read_max_rows": heavy_read_max_rows(), "ts": chrono::Utc::now().to_rfc3339_opts(chrono::SecondsFormat::Secs, true), }); send_ok(&mut stream, &resp.to_string()); @@ -671,11 +727,16 @@ fn handle_client(mut stream: UnixStream, args: &Args, lancedb: LanceDBOps, state } fn send_ok(stream: &mut UnixStream, json: &str) { - let resp = ConsolidateResponse { status: "ok".into(), report_json: json.to_string(), failure_step: "".into(), error_detail: "".into() }; + let resp = ConsolidateResponse { status: "ok".into(), report_json: json.to_string(), failure_step: String::new(), error_detail: String::new(), truncated: false }; + send_response(stream, &resp); +} +/// 方案②（t_576714de）：被行数上限截断的正常返回（report_json=部分结果，truncated=true）。 +fn send_ok_truncated(stream: &mut UnixStream, json: &str) { + let resp = ConsolidateResponse { status: "ok".into(), report_json: json.to_string(), failure_step: String::new(), error_detail: String::new(), truncated: true }; send_response(stream, &resp); } fn send_error(stream: &mut UnixStream, step: &str, detail: &str) { - let resp = ConsolidateResponse { status: "error".into(), report_json: "".into(), failure_step: step.to_string(), error_detail: detail.to_string() }; + let resp = ConsolidateResponse { status: "error".into(), report_json: String::new(), failure_step: step.to_string(), error_detail: detail.to_string(), truncated: false }; send_response(stream, &resp); } @@ -806,8 +867,8 @@ fn run_consolidation(args: &Args, lancedb: &LanceDBOps) -> ConsolidationReport { if args.mode == "full" { let calibrator = DecayCalibrator::new(); let old_rates = decay_calibrate::default_decay_rates(); - // 从 LanceDB 加载真实样本 - let all_records = lancedb.scan_all().unwrap_or_default(); + // 整合内部全量扫描（非 IPC 边界，需全量；行数上限只作用于 IPC 单次重读，t_576714de）。 + let all_records = lancedb.scan_all(200_000).unwrap_or_default(); let now_secs = std::time::SystemTime::now() .duration_since(std::time::UNIX_EPOCH) .unwrap_or_default() @@ -846,8 +907,8 @@ fn run_consolidation(args: &Args, lancedb: &LanceDBOps) -> ConsolidationReport { eprintln!("[consolidate] Step 4: BGE HTTP (port 8000) unreachable — skipping quality backtrace"); } else { let backtracer = QualityBacktracer::new(&args.llm_endpoint, &args.llm_model, &args.llm_api_key); - // 从 LanceDB 加载（content = distilled, source metadata = original） - let all_records = lancedb.scan_all().unwrap_or_default(); + // 从 LanceDB 加载（content = distilled, source metadata = original）。整合内部全量，非 IPC 边界。 + let all_records = lancedb.scan_all(200_000).unwrap_or_default(); let samples: Vec = all_records .iter() .filter(|r| !r.content.is_empty())