zhiyi: 单次重读行数上限 + Go IPC 读 deadline + 每 rpc 新 dial (t_576714de)
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。 不部署（部署另批，需小怡终审 + 牧尘放行）。
This commit is contained in:
小唯
@@ -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，无需长连接）")
|
||||
}
|
||||
@@ -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
|
||||
}
|
||||
|
||||
|
||||
@@ -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<Vec<MemoryRecord>, Box<dyn std::error::Error>> {
|
||||
/// ② 上限由调用方传入 max_rows（IPC 边界 = 单次重读硬上限 heavy_read_max_rows()，
|
||||
/// 整合内部全量 = 200_000；t_576714de）
|
||||
pub fn scan_all(&self, max_rows: usize) -> Result<Vec<MemoryRecord>, Box<dyn std::error::Error>> {
|
||||
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<Vec<MemoryRecord>, Box<dyn std::error::Error>> {
|
||||
/// 只读元数据（跳 vector），上限由调用方传入 max_rows（IPC 边界 = 单次重读硬上限，
|
||||
/// 整合/回填脚本按需传全量；t_576714de）。带 delete_reason 列（供回填时跳过已回填的）。
|
||||
pub fn scan_deleted(&self, max_rows: usize) -> Result<Vec<MemoryRecord>, Box<dyn std::error::Error>> {
|
||||
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()
|
||||
)?);
|
||||
|
||||
|
||||
+73
-12
@@ -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::<usize>().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::<usize>().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<f32> = 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<QualitySample> = all_records
|
||||
.iter()
|
||||
.filter(|r| !r.content.is_empty())
|
||||
|
||||
Reference in New Issue
Block a user