使用 Pebble 替换 full_sync.reader.oplog_store_disk 的 go-diskqueue 本地落盘实现

## 背景

`full_sync.reader.oplog_store_disk=true` 当前依赖 `github.com/vinllen/go-diskqueue`，用于在全量同步期间把增量 oplog 临时写入本地磁盘，并在全量完成后回放这些 oplog。

这条路径近期已经暴露出一些可靠性问题：

- #976：`Persister.retrieve()` 等待循环无法正确跳出，导致磁盘中的 oplog 永远不会进入 apply 阶段。
- #979：继续指出 `Persister.retrieve()` 里还有 polling 延迟和 ticker 生命周期问题。
- 实际生产使用中，开启 `full_sync.reader.oplog_store_disk=true` 后，后续回放本地磁盘 oplog 时仍可能出现 panic。当前 disk queue 实现不太适合作为长期可靠的本地落盘层。

#981 已经修复了 `Persister.retrieve()` 状态机层面的主要问题，但底层仍然保留 `go-diskqueue`。

这个 issue 建议进一步把本地临时 oplog 队列替换成 Pebble-backed spool。

## 目标

直接使用 Pebble 替换 `go-diskqueue`，用户侧配置保持不变：

```ini
full_sync.reader.oplog_store_disk = true
```

不新增 storage backend 选择项，也不保留 `go-diskqueue` fallback。

替换后仍保持现有运行语义：

- 全量同步期间，增量 oplog 写入本地临时 spool。
- 全量同步完成后，`StartDiskApply()` 切换到磁盘回放阶段。
- 本地 spool 中的 oplog 按 FIFO 顺序回放到现有 pending queue。
- 回放完成后切换到内存 apply。
- `OplogDiskQueueFinishTs` 继续表示本地落盘 oplog 的最后 timestamp；只有 worker checkpoint 追上这个 timestamp 后，才认为磁盘回放真正完成。

## 为什么直接移除 go-diskqueue

我建议直接移除 `go-diskqueue`，而不是做成可选后端，原因是：

1. `full_sync.reader.oplog_store_disk=true` 本来就是低频、特殊的可靠性路径。
2. 当前 `go-diskqueue` 路径已经有明确 bug 和生产不稳定经验。
3. 如果保留两个后端，测试矩阵会变大，但仍然保留了一条已知高风险路径。
4. Pebble 是成熟的嵌入式有序 KV，可以把一致性、批量写入和有序 scan 交给存储层处理，避免继续维护自定义队列文件语义。

## 兼容性策略

新实现不需要兼容读取旧版本遗留的 `go-diskqueue` 文件。

如果 checkpoint 仍然指向旧版本创建的本地 queue，新版本启动时应直接 fail fast，并给出明确错误提示，例如：

```text
oplog spool is not a Pebble spool. Legacy go-diskqueue files are not supported.
Please either finish replay with the old MongoShake version, or clear the stale checkpoint/local queue after confirming source oplog can cover the restart position.
```

不做自动迁移的原因：

- 旧格式本身就是这次要移除的风险点。
- 自动迁移需要理解旧 queue 文件格式、读取游标、部分回放状态和异常损坏场景，复杂度较高。
- 错误迁移比显式失败更危险，可能导致 oplog 丢失或重复回放。

## 设计草案

建议新增一个很窄的内部 spool 抽象，例如放在 `collector/spool`。`Persister` 只依赖这个接口，不直接依赖 Pebble API。

```go
type OplogSpool interface {
Put(data []byte) error
ReadBatch(max int) ([][]byte, error)
Advance(n int) error
ReadAll() ([][]byte, error)
LastWriteData() ([]byte, error)
Depth() (uint64, error)
Stats() Stats
Close() error
Delete() error
}
```

本地目录建议为：

```text
/spool//
```

其中 `` 继续由 checkpoint 字段 `OplogDiskQueue` 保存。字段名可以不改，虽然内部格式已经不是 `go-diskqueue`。

### Pebble key schema

Pebble 内部建议分两个 key namespace：

```text
m/... meta key
d/... data key
```

具体 schema：

```text
m/version -> uint32，schema version，第一版为 1
m/write_seq -> uint64，已经成功写入的最大 seq
m/read_seq -> uint64，下一条需要回放的 seq
m/last_write_ts -> int64，最近写入 oplog 的 timestamp 编码值
m/last_write_data -> []byte，最近写入 oplog 的原始 BSON bytes

d/ -> []byte，seq 对应的原始 oplog BSON bytes
```

这里 `d/` 使用 big-endian uint64 编码，是为了让 Pebble 的字典序 scan 等价于 seq 数值升序。例如：

```text
d/0000000000000001
d/0000000000000002
d/0000000000000003
```

这样 `ReadBatch` 只需要从 `d/` 开始用 iterator 顺序扫描，就能得到 FIFO 回放顺序。

### meta 字段含义

`write_seq` 表示本地 spool 已经持久化到哪个位置。

`read_seq` 表示下一条应该回放的位置。

因此当前未回放条数可以按下面计算：

```text
depth = write_seq >= read_seq ? write_seq - read_seq + 1 : 0
```

`last_write_data` 用来保持当前 `GetQueryTsFromDiskQueue()` 的语义：回放完成后仍然能拿到最后一条本地落盘 oplog，并解析出 `diskQueueLastTs`。

`last_write_ts` 是同一信息的结构化缓存，便于恢复、日志和后续指标使用；第一版仍可保留从 `last_write_data` 解析 timestamp 的路径，以减少 `Persister` 改动面。

### 写入语义

`Put(data)` 使用 Pebble batch 原子写入：

```text
d/ = raw oplog BSON
m/write_seq = next_seq
m/last_write_ts = oplog timestamp
m/last_write_data = raw oplog BSON
```

这样可以避免 “data 写入成功但 meta 没更新” 或 “meta 更新了但 data 不存在” 这类部分状态。

默认不对每条 oplog 强制 fsync。原因是当前 `go-diskqueue` 路径也不是严格断电事务日志语义；每条 fsync 会显著影响全量期间写入性能。第一版可以依赖 Pebble batch 原子提交，并在正常关闭 / 阶段切换时 close 或 flush。

### 读取和推进语义

`ReadBatch(max)` 只读取，不修改 `read_seq`：

```text
从 d/ 开始 iterator scan，最多读取 max 条
```

`Advance(n)` 在数据已经推入 pending queue 后推进：

```text
m/read_seq = old_read_seq + n
```

这保持当前 `go-diskqueue.Next()` 的语义：本地 spool 的 read pointer 表示“已经交给 pending queue”，不是“已经写入目标端”。真正端到端确认仍然由 worker ack 和 checkpoint 负责。

### 清理语义

回放完成后删除整个 Pebble spool 目录，而不是长期保留已读 key。

第一版可以不对每批已读数据做 point delete，只推进 `read_seq`。原因是这个 spool 本来就是全量期间的临时目录，最终会整体删除；逐条删除会增加写放大。

## Pebble 参考资料

Pebble 是 CockroachDB 使用的 Go 实现嵌入式 KV 存储，设计上类似 LevelDB / RocksDB。它支持 ordered key iteration、batch 写入、DeleteRange 等能力，适合这里的本地临时有序 spool 场景。

参考资料：

- Pebble GitHub README: https://github.com/cockroachdb/pebble
- Pebble Go package docs: https://pkg.go.dev/github.com/cockroachdb/pebble
- Cockroach Labs: Introducing Pebble: https://www.cockroachlabs.com/blog/pebble-rocksdb-kv-store/

这些资料里提到的能力与本设计的对应关系：

- ordered key/value store：用于 `d/` 的 FIFO scan。
- Iterator：用于从 `read_seq` 开始顺序读取待回放 oplog。
- Batch：用于把 `d/` 和 `m/write_seq`、`m/last_write_ts`、`m/last_write_data` 原子写入。
- Delete / DeleteRange：可作为后续优化，用于批量清理已读 key；第一版先整体删除 spool 目录。

## 可观测性

建议新增 Prometheus 指标：

- `spool_depth{name,stage}`
- `spool_write_seq{name,stage}`
- `spool_read_seq{name,stage}`
- `spool_write_total{name,stage}`
- `spool_read_total{name,stage}`
- `spool_errors_total{name,stage,op}`

现有 `/persist` API 建议继续保留 `disk_write_count` 和 `disk_read_count`，避免破坏已有调试入口；可以额外增加：

- `spool_depth`
- `spool_read_seq`
- `spool_write_seq`

## 范围

本 issue 建议包含：

- 替换 `collector/persister.go` 中的 `go-diskqueue` 调用。
- 新增 Pebble-backed spool 实现和单元测试。
- 保持现有 checkpoint 字段和 stage 状态机。
- 增加 Prometheus 指标和关键日志。
- 对旧 `go-diskqueue` 本地文件 fail fast。

不包含：

- change stream spool 支持。
- 跨 shard / replset 全局排序。
- storage backend 配置项。
- 旧 `go-diskqueue` 文件迁移。
- Grafana dashboard 或 alert rule 调整。

## 验证建议

建议覆盖以下命令：

```bash
go test ./collector/spool
go test ./collector -run 'TestPersister|TestCheckpoint'
go test ./common -run 'Test.*Prom'
go build ./cmd/collector
```

建议覆盖以下用例：

- `Put` / `ReadBatch` / `Advance` / `Depth`。
- 关闭后重新打开 Pebble spool，确认 `read_seq` / `write_seq` / `last_write_data` 能恢复。
- `Delete` 后本地 spool 目录被删除。
- 非 Pebble 目录或旧 `go-diskqueue` 目录启动失败。
- schema version 不匹配时启动失败。
- `[read_seq, write_seq]` 中间缺失 data key 时启动或读取失败。
- `/metrics` 中能看到 spool 相关指标。

## 相关 issue / PR

- #976
- #979
- #981