diff --git a/README.md b/README.md index 22881ea..ed3451e 100644 --- a/README.md +++ b/README.md @@ -189,7 +189,8 @@ case err != nil: ### 对外契约只有两样 -`client` 包（本 SDK）与上面的 BanNet 协议规范。除此之外的包都是内部实现，可能随版本变动。 +`client`（SDK）、`proto`（协议常量与编解码）、`predicate`（SCAN 谓词）三个包，加上上面的 +BanNet 协议规范。这一边界由编译器强制：其余实现包都在 `internal/` 之下，模块外无法导入。 仓库内另有一条 gRPC 传输（`internal/kvgrpc`），但它**不是对外接口**，已置于 `internal/` 之下由编译器强制——模块外无法导入。把 `.proto` 交给使用方自行生成客户端，等于把内部传输 diff --git a/bannet/wire_scan_test.go b/bannet/wire_scan_test.go index 3474558..edefcd7 100644 --- a/bannet/wire_scan_test.go +++ b/bannet/wire_scan_test.go @@ -5,7 +5,7 @@ import ( "testing" "github.com/NeverENG/BanDB/bannet" - "github.com/NeverENG/BanDB/pkg/proto" + "github.com/NeverENG/BanDB/proto" ) // TestScanResponseSurvivesWire 复现并守护 SCAN 的网络缝：一个多条命中的响应 diff --git a/client/client.go b/client/client.go index c5e2981..3c7c24a 100644 --- a/client/client.go +++ b/client/client.go @@ -35,8 +35,8 @@ import ( "sync/atomic" "time" - "github.com/NeverENG/BanDB/pkg/predicate" - "github.com/NeverENG/BanDB/pkg/proto" + "github.com/NeverENG/BanDB/predicate" + "github.com/NeverENG/BanDB/proto" ) // 默认参数。均可经 Options 覆盖。 diff --git a/client/client_test.go b/client/client_test.go index 8daad5c..59848b1 100644 --- a/client/client_test.go +++ b/client/client_test.go @@ -13,7 +13,7 @@ import ( "github.com/NeverENG/BanDB/bannet" "github.com/NeverENG/BanDB/client" "github.com/NeverENG/BanDB/config" - "github.com/NeverENG/BanDB/pkg/proto" + "github.com/NeverENG/BanDB/proto" "github.com/NeverENG/BanDB/service" ) diff --git a/client/conn.go b/client/conn.go index 3b4987f..61ff06e 100644 --- a/client/conn.go +++ b/client/conn.go @@ -8,7 +8,7 @@ import ( "net" "time" - "github.com/NeverENG/BanDB/pkg/proto" + "github.com/NeverENG/BanDB/proto" ) // 线格式（与 pkg/proto 的包注释一致，服务端对应实现为 bannet.DataPack）： diff --git a/client/wire_compat_test.go b/client/wire_compat_test.go index ed9ac02..1b2d3e0 100644 --- a/client/wire_compat_test.go +++ b/client/wire_compat_test.go @@ -6,7 +6,7 @@ import ( "testing" "github.com/NeverENG/BanDB/bannet" - "github.com/NeverENG/BanDB/pkg/proto" + "github.com/NeverENG/BanDB/proto" ) // TestFrameEncodingMatchesServer 交叉校验 SDK 的帧编码与服务端实现逐字节一致。 diff --git a/cmd/ban-bench/runner.go b/cmd/ban-bench/runner.go index 18500ac..15c2a9f 100644 --- a/cmd/ban-bench/runner.go +++ b/cmd/ban-bench/runner.go @@ -11,8 +11,7 @@ import ( "time" "github.com/NeverENG/BanDB/bannet" - "github.com/NeverENG/BanDB/pkg/proto" - "github.com/NeverENG/BanDB/pkg/utils" + "github.com/NeverENG/BanDB/proto" ) type Config struct { @@ -315,7 +314,7 @@ func put(conn *net.TCPConn, key, value []byte) error { binary.LittleEndian.PutUint32(keyLen, uint32(len(key))) binary.LittleEndian.PutUint32(valLen, uint32(len(value))) - data := utils.ByteBuilder(keyLen, valLen, key, value) + data := concatBytes(keyLen, valLen, key, value) if err := send(conn, proto.MsgPut, data); err != nil { return err } @@ -333,7 +332,7 @@ func put(conn *net.TCPConn, key, value []byte) error { func get(conn *net.TCPConn, key []byte) ([]byte, error) { keyLen := make([]byte, 4) binary.LittleEndian.PutUint32(keyLen, uint32(len(key))) - data := utils.ByteBuilder(keyLen, key) + data := concatBytes(keyLen, key) if err := send(conn, proto.MsgGet, data); err != nil { return nil, err @@ -359,7 +358,7 @@ func get(conn *net.TCPConn, key []byte) ([]byte, error) { func del(conn *net.TCPConn, key []byte) error { keyLen := make([]byte, 4) binary.LittleEndian.PutUint32(keyLen, uint32(len(key))) - data := utils.ByteBuilder(keyLen, key) + data := concatBytes(keyLen, key) if err := send(conn, proto.MsgDelete, data); err != nil { return err @@ -379,3 +378,16 @@ func makeKey(idx int, keySize int) []byte { s := fmt.Sprintf("%0*x", keySize, idx) return []byte(s) } + +// concatBytes 按总长度一次分配后拼接多个字节切片。 +func concatBytes(parts ...[]byte) []byte { + n := 0 + for _, p := range parts { + n += len(p) + } + out := make([]byte, 0, n) + for _, p := range parts { + out = append(out, p...) + } + return out +} diff --git a/cmd/ban-cli/interactive.go b/cmd/ban-cli/interactive.go index 633e279..9c227ed 100644 --- a/cmd/ban-cli/interactive.go +++ b/cmd/ban-cli/interactive.go @@ -10,7 +10,7 @@ import ( "time" bandb "github.com/NeverENG/BanDB/client" - "github.com/NeverENG/BanDB/pkg/predicate" + "github.com/NeverENG/BanDB/predicate" ) // cmdTimeout 是交互模式下单条命令的超时。 diff --git a/cmd/ban-server/server.go b/cmd/ban-server/server.go index 31b155c..cff1607 100644 --- a/cmd/ban-server/server.go +++ b/cmd/ban-server/server.go @@ -10,8 +10,8 @@ import ( "time" "github.com/NeverENG/BanDB/bannet" - "github.com/NeverENG/BanDB/pkg/metrics" - "github.com/NeverENG/BanDB/pkg/proto" + "github.com/NeverENG/BanDB/internal/metrics" + "github.com/NeverENG/BanDB/proto" "github.com/NeverENG/BanDB/service" "github.com/NeverENG/BanDB/service/ingesthook" ) diff --git a/cmd/ban-server/server_pprof.go b/cmd/ban-server/server_pprof.go index 9340dc2..fc9ec2e 100644 --- a/cmd/ban-server/server_pprof.go +++ b/cmd/ban-server/server_pprof.go @@ -10,8 +10,8 @@ import ( "time" "github.com/NeverENG/BanDB/bannet" - "github.com/NeverENG/BanDB/pkg/metrics" - "github.com/NeverENG/BanDB/pkg/proto" + "github.com/NeverENG/BanDB/internal/metrics" + "github.com/NeverENG/BanDB/proto" "github.com/NeverENG/BanDB/service" "github.com/NeverENG/BanDB/service/ingesthook" ) diff --git a/pkg/admission/closedloop_test.go b/internal/admission/closedloop_test.go similarity index 100% rename from pkg/admission/closedloop_test.go rename to internal/admission/closedloop_test.go diff --git a/pkg/admission/limiter.go b/internal/admission/limiter.go similarity index 100% rename from pkg/admission/limiter.go rename to internal/admission/limiter.go diff --git a/pkg/admission/limiter_test.go b/internal/admission/limiter_test.go similarity index 100% rename from pkg/admission/limiter_test.go rename to internal/admission/limiter_test.go diff --git a/pkg/credit/credit.go b/internal/credit/credit.go similarity index 100% rename from pkg/credit/credit.go rename to internal/credit/credit.go diff --git a/pkg/credit/credit_test.go b/internal/credit/credit_test.go similarity index 100% rename from pkg/credit/credit_test.go rename to internal/credit/credit_test.go diff --git a/pkg/metrics/metrics.go b/internal/metrics/metrics.go similarity index 100% rename from pkg/metrics/metrics.go rename to internal/metrics/metrics.go diff --git a/pkg/metrics/metrics_test.go b/internal/metrics/metrics_test.go similarity index 100% rename from pkg/metrics/metrics_test.go rename to internal/metrics/metrics_test.go diff --git a/pkg/utils/byteBuilder.go b/pkg/utils/byteBuilder.go deleted file mode 100644 index 85ac17f..0000000 --- a/pkg/utils/byteBuilder.go +++ /dev/null @@ -1,23 +0,0 @@ -package utils - -// ByteBuilder 拼接多个字节切片 -// 使用 make + copy 方式，性能最优 -func ByteBuilder(data ...[]byte) []byte { - // 计算总长度 - totalLen := 0 - for _, b := range data { - totalLen += len(b) - } - - // 直接分配内存 - result := make([]byte, totalLen) - - // 复制数据 - pos := 0 - for _, b := range data { - copy(result[pos:], b) - pos += len(b) - } - - return result -} diff --git a/pkg/predicate/predicate.go b/predicate/predicate.go similarity index 100% rename from pkg/predicate/predicate.go rename to predicate/predicate.go diff --git a/pkg/predicate/predicate_test.go b/predicate/predicate_test.go similarity index 100% rename from pkg/predicate/predicate_test.go rename to predicate/predicate_test.go diff --git a/pkg/proto/codes.go b/proto/codes.go similarity index 100% rename from pkg/proto/codes.go rename to proto/codes.go diff --git a/pkg/proto/scan.go b/proto/scan.go similarity index 98% rename from pkg/proto/scan.go rename to proto/scan.go index c602192..9ebf537 100644 --- a/pkg/proto/scan.go +++ b/proto/scan.go @@ -4,7 +4,7 @@ import ( "encoding/binary" "fmt" - "github.com/NeverENG/BanDB/pkg/predicate" + "github.com/NeverENG/BanDB/predicate" ) // SCAN 请求负载布局（小端）： diff --git a/pkg/proto/scan_test.go b/proto/scan_test.go similarity index 97% rename from pkg/proto/scan_test.go rename to proto/scan_test.go index 372e658..5a1c519 100644 --- a/pkg/proto/scan_test.go +++ b/proto/scan_test.go @@ -4,7 +4,7 @@ import ( "bytes" "testing" - "github.com/NeverENG/BanDB/pkg/predicate" + "github.com/NeverENG/BanDB/predicate" ) func TestScanRequestRoundTrip(t *testing.T) { diff --git a/raft/raft.go b/raft/raft.go index d5af691..b8f805d 100644 --- a/raft/raft.go +++ b/raft/raft.go @@ -1,14 +1,13 @@ +// 本文件是 Raft 的类型、构造、生命周期与只读访问器。 +// 选举见 raft_election.go，日志复制见 raft_replication.go，快照见 raft_snapshot.go， +// 状态与日志的持久化见 raft_persist.go。 package raft import ( - "encoding/binary" - "fmt" "log/slog" "math/rand" "sync" "time" - - "github.com/NeverENG/BanDB/config" ) const ( @@ -145,89 +144,12 @@ func NewRaftGroup(groupID int, peers []string, me int, dataDir string) *Raft { } // persistStateLocked 仅持久化 Term 和 votedFor（增量持久化，O(1)） -func (r *Raft) persistStateLocked() { - if err := r.wal.SaveState(int64(r.Term), int64(r.votedFor)); err != nil { - slog.Error("failed to persist state", "error", err) - } -} - -// persistLocked 全量持久化 Raft 状态（仅用于日志冲突截断等特殊情况） -func (r *Raft) persistLocked() { - data := PersistData{ - CurrentTerm: int64(r.Term), - VotedFor: int64(r.votedFor), - Log: r.log, - LastIncludedIndex: r.LastIncludedIndex, - LastIncludedTerm: r.LastIncludedTerm, - } - - if err := r.wal.SavePersist(data); err != nil { - slog.Error("failed to persist state", "error", err) - } -} - -// readPersist 从磁盘加载 Raft 状态 -func (r *Raft) readPersist() error { - data, err := r.wal.LoadPersist() - if err != nil { - return err - } - - r.Term = int(data.CurrentTerm) - r.votedFor = int(data.VotedFor) - r.log = data.Log - r.LastIncludedIndex = data.LastIncludedIndex - r.LastIncludedTerm = data.LastIncludedTerm - - if r.LastIncludedIndex > 0 { - r.commitIndex = int(r.LastIncludedIndex) - r.lastApplied = int(r.LastIncludedIndex) - r.lastSnapshotIndex = int(r.LastIncludedIndex) - } - - return nil -} func (r *Raft) Start() { if r.state == Leader { r.startHeartbeatLoop() } } -func (r *Raft) electionLoop() { - for { - select { - case <-r.stopCh: - r.timer.Stop() - return - case <-r.timer.C: - r.startElection() - r.resetElectionTimer() - case <-r.heartbeatCh: - r.resetElectionTimer() - case <-r.electionCh: - r.resetElectionTimer() - } - } -} - -// LeaderHint 返回当前已知 leader 的地址供路由重定向：本节点即 leader 时返回自身地址； -// 否则返回从 AppendEntries 学到的 leader 地址；均未知时 ok=false。 -func (r *Raft) LeaderHint() (string, bool) { - r.mu.Lock() - defer r.mu.Unlock() - if r.state == Leader { - return r.addrMap[r.me], true - } - if r.currentLeader >= 0 { - if addr, ok := r.addrMap[r.currentLeader]; ok { - return addr, true - } - } - return "", false -} - -// Stop 停止 Raft 的选举与心跳循环并释放定时器。幂等（可重复调用）。 -// 用途：测试清理（避免泄漏的选举 goroutine 跨用例互扰），以及 Multi-Raft 的组启停。 func (r *Raft) Stop() { r.stopOnce.Do(func() { close(r.stopCh) @@ -235,261 +157,6 @@ func (r *Raft) Stop() { } // resetElectionTimer 复用单一 timer: 先 Stop→drain 残留信号, 再用新随机超时 Reset -func (r *Raft) resetElectionTimer() { - if !r.timer.Stop() { - select { - case <-r.timer.C: - default: - } - } - r.electionTimeout = MinElectionTimeout + time.Duration(rand.Int63n(int64(MaxElectionTimeout-MinElectionTimeout))) - r.timer.Reset(r.electionTimeout) -} - -func (r *Raft) startElection() { - r.mu.Lock() - - if r.state == Leader { - r.mu.Unlock() - return - } - - slog.Info("starting election", "state", r.state, "term", r.Term) - - r.state = Candidate - r.currentLeader = -1 // 进入选举，leader 暂未知 - r.Term++ - r.votedFor = r.me - r.persistStateLocked() - - lastLogIndex := -1 - lastLogTerm := 0 - if len(r.log) > 0 { - lastLogIndex = r.log[len(r.log)-1].Index - lastLogTerm = r.log[len(r.log)-1].Term - } else if r.LastIncludedIndex > 0 { - lastLogIndex = int(r.LastIncludedIndex) - lastLogTerm = int(r.LastIncludedTerm) - } - - args := &RequestVoteArgs{ - GroupID: r.groupID, - Term: r.Term, - CandidateID: r.me, - LastLogIndex: lastLogIndex, - LastLogTerm: lastLogTerm, - } - - peerCount := len(r.peers) - 1 - voteCh := make(chan bool, peerCount+1) - - for i := range r.peers { - if i == r.me { - continue - } - - go func(peerID int) { - reply, err := r.SendRequestVote(r.addrMap[peerID], args) - if err != nil { - voteCh <- false - return - } - - r.mu.Lock() - defer r.mu.Unlock() - - if reply.Term > r.Term { - r.Term = reply.Term - r.state = Follower - r.votedFor = -1 - voteCh <- false - return - } - - if reply.Term == r.Term && reply.VoteGranted { - voteCh <- true - } else { - voteCh <- false - } - }(i) - } - - electionTerm := r.Term - r.mu.Unlock() - - // 异步收票：不阻塞 electionLoop。否则候选人在收票的最多 500ms 里无法处理现任 leader - // 的心跳（AppendEntries 本会把它退回 Follower 并重置计时器），会持续无谓改选、令 leader 抖动。 - go r.awaitVotes(voteCh, peerCount, electionTerm) -} - -// awaitVotes 收集本轮（electionTerm）选票并决定当选/退选。仅当仍是本轮任期的 Candidate 时 -// 才动作，避免作用于已被更高任期/心跳终结的过期选举。 -func (r *Raft) awaitVotes(voteCh chan bool, peerCount int, electionTerm int) { - wonAsCandidate := func() { - r.mu.Lock() - if r.state == Candidate && r.Term == electionTerm { - r.becomeLeader() - } - r.mu.Unlock() - } - - if peerCount == 0 { - wonAsCandidate() // 单节点：直接当选 - return - } - - votes := 1 - timeout := time.After(500 * time.Millisecond) - for j := 0; j < peerCount; j++ { - select { - case voteGranted := <-voteCh: - if voteGranted { - votes++ - if votes > len(r.peers)/2 { - wonAsCandidate() - return - } - } - case <-timeout: - r.mu.Lock() - if r.state == Candidate && r.Term == electionTerm { - r.state = Follower - r.votedFor = -1 - } - r.mu.Unlock() - return - } - } -} - -func (r *Raft) becomeLeader() { - slog.Info("becoming leader", "term", r.Term) - r.state = Leader - r.currentLeader = r.me - - // 计算下一个日志的绝对索引（考虑快照偏移） - nextLogIndex := 0 - if len(r.log) > 0 { - nextLogIndex = r.log[len(r.log)-1].Index + 1 - } else if r.LastIncludedIndex > 0 { - nextLogIndex = int(r.LastIncludedIndex) + 1 - } - - for i := range r.peers { - r.nextIndex[i] = nextLogIndex - r.matchIndex[i] = int(r.LastIncludedIndex) - } - - slog.Debug("heartbeat loop started") - r.startHeartbeatLoop() -} - -func (r *Raft) startHeartbeatLoop() { - if r.heartbeatTicker != nil { - r.heartbeatTicker.Stop() - } - - r.heartbeatTicker = time.NewTicker(HeartbeatInterval) - ticker := r.heartbeatTicker - go func() { - defer ticker.Stop() - for { - select { - case <-r.stopCh: - return - case <-ticker.C: - // 心跳即携带待复制条目的 AppendEntries：与 replicateLog 同一路径， - // 确保 follower 落后时下一次心跳就把缺的条目补上，而不是发空包。 - r.mu.Lock() - if r.state != Leader { - r.mu.Unlock() - return - } - r.replicateLog() - r.mu.Unlock() - } - } - }() -} - -func (r *Raft) updateCommitIndex() { - if r.state != Leader { - return - } - - // 从后往前遍历日志条目，找到可以提交的 - for i := len(r.log) - 1; i >= 0; i-- { - n := r.log[i].Index - if n <= r.commitIndex { - continue - } - if r.log[i].Term != r.Term { - continue - } - - count := 1 - for j := range r.peers { - if j != r.me && r.matchIndex[j] >= n { - count++ - } - } - if count > len(r.peers)/2 { - r.commitIndex = n - r.applyCommittedLogs() - r.commitCond.Broadcast() - break - } - } -} - -func (r *Raft) applyCommittedLogs() { - for r.lastApplied < r.commitIndex { - r.lastApplied++ - // 将绝对索引转换为相对数组索引 - relativeIndex := r.relIndex(r.lastApplied) - if relativeIndex >= 0 && relativeIndex < len(r.log) { - if r.ApplyCh != nil { - r.ApplyCh <- r.log[relativeIndex] - } - } - } - - // 检查是否需要触发快照 - r.checkSnapshotTrigger() -} - -// checkSnapshotTrigger 检查是否应该触发快照 -func (r *Raft) checkSnapshotTrigger() { - if r.state != Leader { - return - } - - logLength := len(r.log) - threshold := config.G.RaftSnapshotThreshold - if threshold <= 0 { - threshold = 10000 - } - keepEntries := config.G.RaftSnapshotKeepEntries - if keepEntries <= 0 { - keepEntries = 100 - } - - if logLength > threshold { - snapshotIndex := r.commitIndex - keepEntries - if snapshotIndex > r.lastSnapshotIndex { - slog.Info("auto-triggering snapshot", "index", snapshotIndex, "logLen", logLength, "threshold", threshold) - // 异步调用避免持锁死锁（checkSnapshotTrigger 在持锁上下文中被调用） - go r.TakeSnapshot(snapshotIndex) - } - } -} - -// relIndex 把绝对日志 index 映射为 r.log 的数组下标。 -// -// off-by-one 易错点：无快照时基准应为 -1（首条绝对 index 0 → 数组下标 0）； -// 但 LastIncludedIndex 初始为 0，旧代码统一用 `abs - LastIncludedIndex - 1`，在无快照场景 -// 把 index 0 算成 -1 → 追加/取 term/切片全部越界或错位。因 LastIncludedIndex>0 才代表"有快照"， -// 这里无快照时用基准 -1，有快照时用 LastIncludedIndex（与旧公式一致，不影响快照路径）。 func (r *Raft) relIndex(absIndex int) int { base := -1 if r.LastIncludedIndex > 0 { @@ -524,158 +191,6 @@ func (r *Raft) getLastLogIndex() int { return -1 } -func (r *Raft) AppendEntry(command []byte) (int, error) { - r.mu.Lock() - defer r.mu.Unlock() - - if r.state != Leader { - slog.Warn("AppendEntry rejected, not leader", "state", r.state) - return -1, fmt.Errorf("not leader") - } - - // 计算绝对索引（考虑快照偏移） - lastLogIndex := -1 - if len(r.log) > 0 { - lastLogIndex = r.log[len(r.log)-1].Index - } else if r.LastIncludedIndex > 0 { - lastLogIndex = int(r.LastIncludedIndex) - } - - entry := LogEntry{ - Index: lastLogIndex + 1, - Term: r.Term, - Command: command, - } - r.log = append(r.log, entry) - - // 增量持久化：仅追加一条日志 - if err := r.wal.AppendLog(entry); err != nil { - slog.Error("failed to append log", "error", err) - } - - // 单节点模式：立即提交 - if len(r.peers) == 1 { - r.commitIndex = entry.Index - r.applyCommittedLogs() - r.commitCond.Broadcast() - } else { - r.replicateLog() - } - - return entry.Index, nil -} - -func (r *Raft) replicateLog() { - if r.state != Leader { - return - } - - for i := range r.peers { - if i == r.me { - continue - } - - prevLogIndex := r.nextIndex[i] - 1 - - // 如果 follower 落后太多（prevLogIndex 在快照范围内），发送 InstallSnapshot - if prevLogIndex < int(r.LastIncludedIndex) && r.LastIncludedIndex > 0 { - snapshotData, _, _, err := r.wal.LoadLatestSnapshot() - if err == nil && snapshotData != nil { - snapArgs := &InstallSnapshotArgs{ - GroupID: r.groupID, - Term: r.Term, - LeaderID: r.me, - Data: snapshotData, - LastIncludedIndex: r.LastIncludedIndex, - LastIncludedTerm: r.LastIncludedTerm, - } - go func(peerID int, snapArgs *InstallSnapshotArgs) { - reply, err := r.SendInstallSnapshot(r.addrMap[peerID], snapArgs) - if err != nil { - return - } - r.mu.Lock() - defer r.mu.Unlock() - if reply.Success { - r.nextIndex[peerID] = int(r.LastIncludedIndex) + 1 - r.matchIndex[peerID] = int(r.LastIncludedIndex) - } else if reply.Term > r.Term { - r.Term = reply.Term - r.state = Follower - r.votedFor = -1 - r.heartbeatTicker.Stop() - } - }(i, snapArgs) - } - continue - } - - prevLogTerm := r.getTermAt(prevLogIndex) - - // 将绝对索引转换为相对数组索引来切片日志 - var entries []LogEntry - relativeStart := r.relIndex(r.nextIndex[i]) - if relativeStart >= 0 && relativeStart < len(r.log) { - entries = r.log[relativeStart:] - } - - args := &AppendEntriesArgs{ - GroupID: r.groupID, - Term: r.Term, - LeaderID: r.me, - PrevLogIndex: prevLogIndex, - PrevLogTerm: prevLogTerm, - Entries: entries, - LeaderCommit: r.commitIndex, - } - - go func(peerID int, args *AppendEntriesArgs) { - reply, err := r.SendAppendEntries(r.addrMap[peerID], args) - if err != nil { - r.mu.Lock() - if r.state == Leader { - r.nextIndex[peerID]-- - } - r.mu.Unlock() - return - } - - r.mu.Lock() - defer r.mu.Unlock() - - if reply.Term > r.Term { - r.Term = reply.Term - r.state = Follower - r.votedFor = -1 - r.heartbeatTicker.Stop() - return - } - - if reply.Success { - // 只按「本次实际发送的条目」推进 matchIndex，绝不按 leader 的 last index—— - // 否则空心跳会把 follower 误标为已追平（历史 bug：空 entries 却推进到 last）。 - matched := args.PrevLogIndex + len(args.Entries) - if matched > r.matchIndex[peerID] { - r.matchIndex[peerID] = matched - r.nextIndex[peerID] = matched + 1 - } - r.updateCommitIndex() - } else if r.nextIndex[peerID] > 0 { - r.nextIndex[peerID]-- - } - }(i, args) - } -} - -func (r *Raft) WaitCommitIndex(index int) { - r.mu.Lock() - defer r.mu.Unlock() - - for r.commitIndex < index { - r.commitCond.Wait() - } -} - func (r *Raft) State() (State, int) { r.mu.Lock() defer r.mu.Unlock() @@ -696,150 +211,3 @@ func (r *Raft) CommitIndex() int { defer r.mu.Unlock() return r.commitIndex } - -func (r *Raft) TakeSnapshot(index int) error { - r.mu.Lock() - defer r.mu.Unlock() - - if index <= r.lastSnapshotIndex { - return fmt.Errorf("snapshot index %d is not greater than last snapshot index %d", index, r.lastSnapshotIndex) - } - - if index > r.commitIndex { - return fmt.Errorf("cannot snapshot uncommitted index %d, commitIndex is %d", index, r.commitIndex) - } - - // 收集需要放入快照的日志条目（绝对索引 <= index） - var snapshotEntries []LogEntry - relativeEnd := index + 1 - int(r.LastIncludedIndex) - for i := 0; i < relativeEnd && i < len(r.log); i++ { - snapshotEntries = append(snapshotEntries, r.log[i]) - } - - // 获取最后一条的 term - var term int - if len(snapshotEntries) > 0 { - term = snapshotEntries[len(snapshotEntries)-1].Term - } else if index == int(r.LastIncludedIndex) { - term = int(r.LastIncludedTerm) - } - - // 序列化日志条目为快照数据 - data := SerializeLogEntries(snapshotEntries) - - // 1. 先保存快照到磁盘 - if err := r.wal.SaveSnapshot(data, int64(index), int64(term)); err != nil { - return fmt.Errorf("failed to save snapshot: %w", err) - } - - // 2. 删除旧快照 - r.wal.DeleteOldSnapshots(int64(index)) - - // 3. 截断 WAL 日志 - if err := r.wal.TruncateLogs(int64(index)); err != nil { - return fmt.Errorf("failed to truncate logs: %w", err) - } - - // 4. 清理内存中的日志并重新编号 - newLogStart := index + 1 - int(r.LastIncludedIndex) - if newLogStart >= 0 && newLogStart <= len(r.log) { - r.log = r.log[newLogStart:] - for i := range r.log { - r.log[i].Index = index + 1 + i - } - } else { - r.log = []LogEntry{} - } - - // 5. 更新元数据 - r.lastSnapshotIndex = index - r.LastIncludedIndex = int64(index) - r.LastIncludedTerm = int64(term) - - // 6. 通知 FSM 异步重放快照（日志条目序列化数据） - if r.ApplyCh != nil { - snapshotEntry := LogEntry{ - Index: index, - Term: term, - Command: data, - IsSnapshot: true, - } - select { - case r.ApplyCh <- snapshotEntry: - slog.Info("snapshot replay sent to FSM", "index", index, "entries", len(snapshotEntries)) - default: - slog.Warn("ApplyCh full, snapshot replay skipped") - } - } - - // 7. 持久化状态（日志已由 TruncateLogs 处理） - r.persistStateLocked() - - return nil -} - -// SerializeLogEntries 序列化日志条目为字节流（快照数据格式） -func SerializeLogEntries(entries []LogEntry) []byte { - if len(entries) == 0 { - return nil - } - - size := 4 // entry count - for _, e := range entries { - size += 8 + 8 + 8 + len(e.Command) // Index(8) + Term(8) + CmdLen(8) + Command - } - - buf := make([]byte, size) - offset := 0 - binary.BigEndian.PutUint32(buf[offset:], uint32(len(entries))) - offset += 4 - for _, e := range entries { - binary.BigEndian.PutUint64(buf[offset:], uint64(e.Index)) - offset += 8 - binary.BigEndian.PutUint64(buf[offset:], uint64(e.Term)) - offset += 8 - binary.BigEndian.PutUint64(buf[offset:], uint64(len(e.Command))) - offset += 8 - copy(buf[offset:], e.Command) - offset += len(e.Command) - } - return buf -} - -// DeserializeLogEntries 反序列化日志条目 -func DeserializeLogEntries(data []byte) []LogEntry { - if len(data) < 4 { - return nil - } - - offset := 0 - count := binary.BigEndian.Uint32(data[offset:]) - offset += 4 - - entries := make([]LogEntry, 0, count) - for i := uint32(0); i < count; i++ { - if offset+24 > len(data) { - break - } - index := int(binary.BigEndian.Uint64(data[offset:])) - offset += 8 - term := int(binary.BigEndian.Uint64(data[offset:])) - offset += 8 - cmdLen := int(binary.BigEndian.Uint64(data[offset:])) - offset += 8 - - if offset+cmdLen > len(data) { - break - } - cmd := make([]byte, cmdLen) - copy(cmd, data[offset:offset+cmdLen]) - offset += cmdLen - - entries = append(entries, LogEntry{ - Index: index, - Term: term, - Command: cmd, - }) - } - return entries -} diff --git a/raft/raft_election.go b/raft/raft_election.go new file mode 100644 index 0000000..130ed25 --- /dev/null +++ b/raft/raft_election.go @@ -0,0 +1,192 @@ +// 本文件是 Raft 的选举：选举计时、发起投票、统计选票与成为 leader。 +package raft + +import ( + "log/slog" + "math/rand" + "time" +) + +func (r *Raft) electionLoop() { + for { + select { + case <-r.stopCh: + r.timer.Stop() + return + case <-r.timer.C: + r.startElection() + r.resetElectionTimer() + case <-r.heartbeatCh: + r.resetElectionTimer() + case <-r.electionCh: + r.resetElectionTimer() + } + } +} + +// LeaderHint 返回当前已知 leader 的地址供路由重定向：本节点即 leader 时返回自身地址； +// 否则返回从 AppendEntries 学到的 leader 地址；均未知时 ok=false。 +func (r *Raft) LeaderHint() (string, bool) { + r.mu.Lock() + defer r.mu.Unlock() + if r.state == Leader { + return r.addrMap[r.me], true + } + if r.currentLeader >= 0 { + if addr, ok := r.addrMap[r.currentLeader]; ok { + return addr, true + } + } + return "", false +} + +// Stop 停止 Raft 的选举与心跳循环并释放定时器。幂等（可重复调用）。 +// 用途：测试清理（避免泄漏的选举 goroutine 跨用例互扰），以及 Multi-Raft 的组启停。 +func (r *Raft) resetElectionTimer() { + if !r.timer.Stop() { + select { + case <-r.timer.C: + default: + } + } + r.electionTimeout = MinElectionTimeout + time.Duration(rand.Int63n(int64(MaxElectionTimeout-MinElectionTimeout))) + r.timer.Reset(r.electionTimeout) +} + +func (r *Raft) startElection() { + r.mu.Lock() + + if r.state == Leader { + r.mu.Unlock() + return + } + + slog.Info("starting election", "state", r.state, "term", r.Term) + + r.state = Candidate + r.currentLeader = -1 // 进入选举，leader 暂未知 + r.Term++ + r.votedFor = r.me + r.persistStateLocked() + + lastLogIndex := -1 + lastLogTerm := 0 + if len(r.log) > 0 { + lastLogIndex = r.log[len(r.log)-1].Index + lastLogTerm = r.log[len(r.log)-1].Term + } else if r.LastIncludedIndex > 0 { + lastLogIndex = int(r.LastIncludedIndex) + lastLogTerm = int(r.LastIncludedTerm) + } + + args := &RequestVoteArgs{ + GroupID: r.groupID, + Term: r.Term, + CandidateID: r.me, + LastLogIndex: lastLogIndex, + LastLogTerm: lastLogTerm, + } + + peerCount := len(r.peers) - 1 + voteCh := make(chan bool, peerCount+1) + + for i := range r.peers { + if i == r.me { + continue + } + + go func(peerID int) { + reply, err := r.SendRequestVote(r.addrMap[peerID], args) + if err != nil { + voteCh <- false + return + } + + r.mu.Lock() + defer r.mu.Unlock() + + if reply.Term > r.Term { + r.Term = reply.Term + r.state = Follower + r.votedFor = -1 + voteCh <- false + return + } + + if reply.Term == r.Term && reply.VoteGranted { + voteCh <- true + } else { + voteCh <- false + } + }(i) + } + + electionTerm := r.Term + r.mu.Unlock() + + // 异步收票：不阻塞 electionLoop。否则候选人在收票的最多 500ms 里无法处理现任 leader + // 的心跳（AppendEntries 本会把它退回 Follower 并重置计时器），会持续无谓改选、令 leader 抖动。 + go r.awaitVotes(voteCh, peerCount, electionTerm) +} + +// awaitVotes 收集本轮（electionTerm）选票并决定当选/退选。仅当仍是本轮任期的 Candidate 时 +// 才动作，避免作用于已被更高任期/心跳终结的过期选举。 +func (r *Raft) awaitVotes(voteCh chan bool, peerCount int, electionTerm int) { + wonAsCandidate := func() { + r.mu.Lock() + if r.state == Candidate && r.Term == electionTerm { + r.becomeLeader() + } + r.mu.Unlock() + } + + if peerCount == 0 { + wonAsCandidate() // 单节点：直接当选 + return + } + + votes := 1 + timeout := time.After(500 * time.Millisecond) + for j := 0; j < peerCount; j++ { + select { + case voteGranted := <-voteCh: + if voteGranted { + votes++ + if votes > len(r.peers)/2 { + wonAsCandidate() + return + } + } + case <-timeout: + r.mu.Lock() + if r.state == Candidate && r.Term == electionTerm { + r.state = Follower + r.votedFor = -1 + } + r.mu.Unlock() + return + } + } +} + +func (r *Raft) becomeLeader() { + slog.Info("becoming leader", "term", r.Term) + r.state = Leader + r.currentLeader = r.me + + // 计算下一个日志的绝对索引（考虑快照偏移） + nextLogIndex := 0 + if len(r.log) > 0 { + nextLogIndex = r.log[len(r.log)-1].Index + 1 + } else if r.LastIncludedIndex > 0 { + nextLogIndex = int(r.LastIncludedIndex) + 1 + } + + for i := range r.peers { + r.nextIndex[i] = nextLogIndex + r.matchIndex[i] = int(r.LastIncludedIndex) + } + + slog.Debug("heartbeat loop started") + r.startHeartbeatLoop() +} diff --git a/raft/raft_persist.go b/raft/raft_persist.go new file mode 100644 index 0000000..d4d4045 --- /dev/null +++ b/raft/raft_persist.go @@ -0,0 +1,49 @@ +// 本文件是 Raft 的持久化：任期/投票状态与日志的落盘，以及重启时的读回。 +package raft + +import ( + "log/slog" +) + +func (r *Raft) persistStateLocked() { + if err := r.wal.SaveState(int64(r.Term), int64(r.votedFor)); err != nil { + slog.Error("failed to persist state", "error", err) + } +} + +// persistLocked 全量持久化 Raft 状态（仅用于日志冲突截断等特殊情况） +func (r *Raft) persistLocked() { + data := PersistData{ + CurrentTerm: int64(r.Term), + VotedFor: int64(r.votedFor), + Log: r.log, + LastIncludedIndex: r.LastIncludedIndex, + LastIncludedTerm: r.LastIncludedTerm, + } + + if err := r.wal.SavePersist(data); err != nil { + slog.Error("failed to persist state", "error", err) + } +} + +// readPersist 从磁盘加载 Raft 状态 +func (r *Raft) readPersist() error { + data, err := r.wal.LoadPersist() + if err != nil { + return err + } + + r.Term = int(data.CurrentTerm) + r.votedFor = int(data.VotedFor) + r.log = data.Log + r.LastIncludedIndex = data.LastIncludedIndex + r.LastIncludedTerm = data.LastIncludedTerm + + if r.LastIncludedIndex > 0 { + r.commitIndex = int(r.LastIncludedIndex) + r.lastApplied = int(r.LastIncludedIndex) + r.lastSnapshotIndex = int(r.LastIncludedIndex) + } + + return nil +} diff --git a/raft/raft_replication.go b/raft/raft_replication.go new file mode 100644 index 0000000..418ded7 --- /dev/null +++ b/raft/raft_replication.go @@ -0,0 +1,235 @@ +// 本文件是 Raft 的日志复制：追加日志、心跳与复制、推进 commitIndex 并 apply。 +package raft + +import ( + "fmt" + "log/slog" + "time" +) + +func (r *Raft) startHeartbeatLoop() { + if r.heartbeatTicker != nil { + r.heartbeatTicker.Stop() + } + + r.heartbeatTicker = time.NewTicker(HeartbeatInterval) + ticker := r.heartbeatTicker + go func() { + defer ticker.Stop() + for { + select { + case <-r.stopCh: + return + case <-ticker.C: + // 心跳即携带待复制条目的 AppendEntries：与 replicateLog 同一路径， + // 确保 follower 落后时下一次心跳就把缺的条目补上，而不是发空包。 + r.mu.Lock() + if r.state != Leader { + r.mu.Unlock() + return + } + r.replicateLog() + r.mu.Unlock() + } + } + }() +} + +func (r *Raft) updateCommitIndex() { + if r.state != Leader { + return + } + + // 从后往前遍历日志条目，找到可以提交的 + for i := len(r.log) - 1; i >= 0; i-- { + n := r.log[i].Index + if n <= r.commitIndex { + continue + } + if r.log[i].Term != r.Term { + continue + } + + count := 1 + for j := range r.peers { + if j != r.me && r.matchIndex[j] >= n { + count++ + } + } + if count > len(r.peers)/2 { + r.commitIndex = n + r.applyCommittedLogs() + r.commitCond.Broadcast() + break + } + } +} + +func (r *Raft) applyCommittedLogs() { + for r.lastApplied < r.commitIndex { + r.lastApplied++ + // 将绝对索引转换为相对数组索引 + relativeIndex := r.relIndex(r.lastApplied) + if relativeIndex >= 0 && relativeIndex < len(r.log) { + if r.ApplyCh != nil { + r.ApplyCh <- r.log[relativeIndex] + } + } + } + + // 检查是否需要触发快照 + r.checkSnapshotTrigger() +} + +// checkSnapshotTrigger 检查是否应该触发快照 +func (r *Raft) AppendEntry(command []byte) (int, error) { + r.mu.Lock() + defer r.mu.Unlock() + + if r.state != Leader { + slog.Warn("AppendEntry rejected, not leader", "state", r.state) + return -1, fmt.Errorf("not leader") + } + + // 计算绝对索引（考虑快照偏移） + lastLogIndex := -1 + if len(r.log) > 0 { + lastLogIndex = r.log[len(r.log)-1].Index + } else if r.LastIncludedIndex > 0 { + lastLogIndex = int(r.LastIncludedIndex) + } + + entry := LogEntry{ + Index: lastLogIndex + 1, + Term: r.Term, + Command: command, + } + r.log = append(r.log, entry) + + // 增量持久化：仅追加一条日志 + if err := r.wal.AppendLog(entry); err != nil { + slog.Error("failed to append log", "error", err) + } + + // 单节点模式：立即提交 + if len(r.peers) == 1 { + r.commitIndex = entry.Index + r.applyCommittedLogs() + r.commitCond.Broadcast() + } else { + r.replicateLog() + } + + return entry.Index, nil +} + +func (r *Raft) replicateLog() { + if r.state != Leader { + return + } + + for i := range r.peers { + if i == r.me { + continue + } + + prevLogIndex := r.nextIndex[i] - 1 + + // 如果 follower 落后太多（prevLogIndex 在快照范围内），发送 InstallSnapshot + if prevLogIndex < int(r.LastIncludedIndex) && r.LastIncludedIndex > 0 { + snapshotData, _, _, err := r.wal.LoadLatestSnapshot() + if err == nil && snapshotData != nil { + snapArgs := &InstallSnapshotArgs{ + GroupID: r.groupID, + Term: r.Term, + LeaderID: r.me, + Data: snapshotData, + LastIncludedIndex: r.LastIncludedIndex, + LastIncludedTerm: r.LastIncludedTerm, + } + go func(peerID int, snapArgs *InstallSnapshotArgs) { + reply, err := r.SendInstallSnapshot(r.addrMap[peerID], snapArgs) + if err != nil { + return + } + r.mu.Lock() + defer r.mu.Unlock() + if reply.Success { + r.nextIndex[peerID] = int(r.LastIncludedIndex) + 1 + r.matchIndex[peerID] = int(r.LastIncludedIndex) + } else if reply.Term > r.Term { + r.Term = reply.Term + r.state = Follower + r.votedFor = -1 + r.heartbeatTicker.Stop() + } + }(i, snapArgs) + } + continue + } + + prevLogTerm := r.getTermAt(prevLogIndex) + + // 将绝对索引转换为相对数组索引来切片日志 + var entries []LogEntry + relativeStart := r.relIndex(r.nextIndex[i]) + if relativeStart >= 0 && relativeStart < len(r.log) { + entries = r.log[relativeStart:] + } + + args := &AppendEntriesArgs{ + GroupID: r.groupID, + Term: r.Term, + LeaderID: r.me, + PrevLogIndex: prevLogIndex, + PrevLogTerm: prevLogTerm, + Entries: entries, + LeaderCommit: r.commitIndex, + } + + go func(peerID int, args *AppendEntriesArgs) { + reply, err := r.SendAppendEntries(r.addrMap[peerID], args) + if err != nil { + r.mu.Lock() + if r.state == Leader { + r.nextIndex[peerID]-- + } + r.mu.Unlock() + return + } + + r.mu.Lock() + defer r.mu.Unlock() + + if reply.Term > r.Term { + r.Term = reply.Term + r.state = Follower + r.votedFor = -1 + r.heartbeatTicker.Stop() + return + } + + if reply.Success { + // 只按「本次实际发送的条目」推进 matchIndex，绝不按 leader 的 last index—— + // 否则空心跳会把 follower 误标为已追平（历史 bug：空 entries 却推进到 last）。 + matched := args.PrevLogIndex + len(args.Entries) + if matched > r.matchIndex[peerID] { + r.matchIndex[peerID] = matched + r.nextIndex[peerID] = matched + 1 + } + r.updateCommitIndex() + } else if r.nextIndex[peerID] > 0 { + r.nextIndex[peerID]-- + } + }(i, args) + } +} + +func (r *Raft) WaitCommitIndex(index int) { + r.mu.Lock() + defer r.mu.Unlock() + + for r.commitIndex < index { + r.commitCond.Wait() + } +} diff --git a/raft/raft_snapshot.go b/raft/raft_snapshot.go new file mode 100644 index 0000000..c56485c --- /dev/null +++ b/raft/raft_snapshot.go @@ -0,0 +1,187 @@ +// 本文件是 Raft 的快照：触发判定、生成快照，以及日志条目的序列化编解码。 +package raft + +import ( + "encoding/binary" + "fmt" + "github.com/NeverENG/BanDB/config" + "log/slog" +) + +func (r *Raft) checkSnapshotTrigger() { + if r.state != Leader { + return + } + + logLength := len(r.log) + threshold := config.G.RaftSnapshotThreshold + if threshold <= 0 { + threshold = 10000 + } + keepEntries := config.G.RaftSnapshotKeepEntries + if keepEntries <= 0 { + keepEntries = 100 + } + + if logLength > threshold { + snapshotIndex := r.commitIndex - keepEntries + if snapshotIndex > r.lastSnapshotIndex { + slog.Info("auto-triggering snapshot", "index", snapshotIndex, "logLen", logLength, "threshold", threshold) + // 异步调用避免持锁死锁（checkSnapshotTrigger 在持锁上下文中被调用） + go r.TakeSnapshot(snapshotIndex) + } + } +} + +// relIndex 把绝对日志 index 映射为 r.log 的数组下标。 +// +// off-by-one 易错点：无快照时基准应为 -1（首条绝对 index 0 → 数组下标 0）； +// 但 LastIncludedIndex 初始为 0，旧代码统一用 `abs - LastIncludedIndex - 1`，在无快照场景 +// 把 index 0 算成 -1 → 追加/取 term/切片全部越界或错位。因 LastIncludedIndex>0 才代表"有快照"， +// 这里无快照时用基准 -1，有快照时用 LastIncludedIndex（与旧公式一致，不影响快照路径）。 +func (r *Raft) TakeSnapshot(index int) error { + r.mu.Lock() + defer r.mu.Unlock() + + if index <= r.lastSnapshotIndex { + return fmt.Errorf("snapshot index %d is not greater than last snapshot index %d", index, r.lastSnapshotIndex) + } + + if index > r.commitIndex { + return fmt.Errorf("cannot snapshot uncommitted index %d, commitIndex is %d", index, r.commitIndex) + } + + // 收集需要放入快照的日志条目（绝对索引 <= index） + var snapshotEntries []LogEntry + relativeEnd := index + 1 - int(r.LastIncludedIndex) + for i := 0; i < relativeEnd && i < len(r.log); i++ { + snapshotEntries = append(snapshotEntries, r.log[i]) + } + + // 获取最后一条的 term + var term int + if len(snapshotEntries) > 0 { + term = snapshotEntries[len(snapshotEntries)-1].Term + } else if index == int(r.LastIncludedIndex) { + term = int(r.LastIncludedTerm) + } + + // 序列化日志条目为快照数据 + data := SerializeLogEntries(snapshotEntries) + + // 1. 先保存快照到磁盘 + if err := r.wal.SaveSnapshot(data, int64(index), int64(term)); err != nil { + return fmt.Errorf("failed to save snapshot: %w", err) + } + + // 2. 删除旧快照 + r.wal.DeleteOldSnapshots(int64(index)) + + // 3. 截断 WAL 日志 + if err := r.wal.TruncateLogs(int64(index)); err != nil { + return fmt.Errorf("failed to truncate logs: %w", err) + } + + // 4. 清理内存中的日志并重新编号 + newLogStart := index + 1 - int(r.LastIncludedIndex) + if newLogStart >= 0 && newLogStart <= len(r.log) { + r.log = r.log[newLogStart:] + for i := range r.log { + r.log[i].Index = index + 1 + i + } + } else { + r.log = []LogEntry{} + } + + // 5. 更新元数据 + r.lastSnapshotIndex = index + r.LastIncludedIndex = int64(index) + r.LastIncludedTerm = int64(term) + + // 6. 通知 FSM 异步重放快照（日志条目序列化数据） + if r.ApplyCh != nil { + snapshotEntry := LogEntry{ + Index: index, + Term: term, + Command: data, + IsSnapshot: true, + } + select { + case r.ApplyCh <- snapshotEntry: + slog.Info("snapshot replay sent to FSM", "index", index, "entries", len(snapshotEntries)) + default: + slog.Warn("ApplyCh full, snapshot replay skipped") + } + } + + // 7. 持久化状态（日志已由 TruncateLogs 处理） + r.persistStateLocked() + + return nil +} + +// SerializeLogEntries 序列化日志条目为字节流（快照数据格式） +func SerializeLogEntries(entries []LogEntry) []byte { + if len(entries) == 0 { + return nil + } + + size := 4 // entry count + for _, e := range entries { + size += 8 + 8 + 8 + len(e.Command) // Index(8) + Term(8) + CmdLen(8) + Command + } + + buf := make([]byte, size) + offset := 0 + binary.BigEndian.PutUint32(buf[offset:], uint32(len(entries))) + offset += 4 + for _, e := range entries { + binary.BigEndian.PutUint64(buf[offset:], uint64(e.Index)) + offset += 8 + binary.BigEndian.PutUint64(buf[offset:], uint64(e.Term)) + offset += 8 + binary.BigEndian.PutUint64(buf[offset:], uint64(len(e.Command))) + offset += 8 + copy(buf[offset:], e.Command) + offset += len(e.Command) + } + return buf +} + +// DeserializeLogEntries 反序列化日志条目 +func DeserializeLogEntries(data []byte) []LogEntry { + if len(data) < 4 { + return nil + } + + offset := 0 + count := binary.BigEndian.Uint32(data[offset:]) + offset += 4 + + entries := make([]LogEntry, 0, count) + for i := uint32(0); i < count; i++ { + if offset+24 > len(data) { + break + } + index := int(binary.BigEndian.Uint64(data[offset:])) + offset += 8 + term := int(binary.BigEndian.Uint64(data[offset:])) + offset += 8 + cmdLen := int(binary.BigEndian.Uint64(data[offset:])) + offset += 8 + + if offset+cmdLen > len(data) { + break + } + cmd := make([]byte, cmdLen) + copy(cmd, data[offset:offset+cmdLen]) + offset += cmdLen + + entries = append(entries, LogEntry{ + Index: index, + Term: term, + Command: cmd, + }) + } + return entries +} diff --git a/service/admission_bootstrap.go b/service/admission_bootstrap.go index 630e921..cf61030 100644 --- a/service/admission_bootstrap.go +++ b/service/admission_bootstrap.go @@ -4,7 +4,7 @@ import ( "log/slog" "github.com/NeverENG/BanDB/config" - "github.com/NeverENG/BanDB/pkg/admission" + "github.com/NeverENG/BanDB/internal/admission" ) // EnableAdmissionFromConfig 按配置在 router 上开启网关自适应准入（默认关闭直接返回）。 diff --git a/service/delivery/deliverer.go b/service/delivery/deliverer.go index 9b347b2..deb022c 100644 --- a/service/delivery/deliverer.go +++ b/service/delivery/deliverer.go @@ -5,7 +5,7 @@ import ( "log/slog" "time" - "github.com/NeverENG/BanDB/pkg/metrics" + "github.com/NeverENG/BanDB/internal/metrics" "github.com/NeverENG/BanDB/service/delivery/offset" ) diff --git a/service/delivery/exactly_once_test.go b/service/delivery/exactly_once_test.go index b0deb1b..6e798cb 100644 --- a/service/delivery/exactly_once_test.go +++ b/service/delivery/exactly_once_test.go @@ -9,7 +9,7 @@ import ( "path/filepath" "testing" - "github.com/NeverENG/BanDB/pkg/proto" + "github.com/NeverENG/BanDB/proto" ) // 本文件是「exactly-once 正确性压测台」：在崩溃点注入故障，量化重复/丢失条数， diff --git a/service/delivery/governance/breaker.go b/service/delivery/governance/breaker.go index 3092892..04d7b9f 100644 --- a/service/delivery/governance/breaker.go +++ b/service/delivery/governance/breaker.go @@ -9,7 +9,7 @@ import ( "sync" "time" - "github.com/NeverENG/BanDB/pkg/metrics" + "github.com/NeverENG/BanDB/internal/metrics" ) // state 是熔断器的三态。 diff --git a/service/delivery/source.go b/service/delivery/source.go index 5a36e47..14f5f41 100644 --- a/service/delivery/source.go +++ b/service/delivery/source.go @@ -3,8 +3,8 @@ package delivery import ( "bytes" - "github.com/NeverENG/BanDB/pkg/predicate" - "github.com/NeverENG/BanDB/pkg/proto" + "github.com/NeverENG/BanDB/predicate" + "github.com/NeverENG/BanDB/proto" "github.com/NeverENG/BanDB/service/delivery/offset" ) diff --git a/service/delivery/source_test.go b/service/delivery/source_test.go index bef9c76..9334ed4 100644 --- a/service/delivery/source_test.go +++ b/service/delivery/source_test.go @@ -4,8 +4,8 @@ import ( "bytes" "testing" - "github.com/NeverENG/BanDB/pkg/predicate" - "github.com/NeverENG/BanDB/pkg/proto" + "github.com/NeverENG/BanDB/predicate" + "github.com/NeverENG/BanDB/proto" "github.com/NeverENG/BanDB/service/delivery/offset" ) diff --git a/service/fsm.go b/service/fsm.go index 3609e51..387ddda 100644 --- a/service/fsm.go +++ b/service/fsm.go @@ -8,8 +8,8 @@ import ( "time" "github.com/NeverENG/BanDB/config" - "github.com/NeverENG/BanDB/pkg/predicate" - "github.com/NeverENG/BanDB/pkg/proto" + "github.com/NeverENG/BanDB/predicate" + "github.com/NeverENG/BanDB/proto" "github.com/NeverENG/BanDB/raft" "github.com/NeverENG/BanDB/storage" ) diff --git a/service/ingesthook/filter.go b/service/ingesthook/filter.go index 20f1218..592e654 100644 --- a/service/ingesthook/filter.go +++ b/service/ingesthook/filter.go @@ -14,8 +14,8 @@ import ( "sync" "github.com/NeverENG/BanDB/bannet" - "github.com/NeverENG/BanDB/pkg/metrics" - "github.com/NeverENG/BanDB/pkg/proto" + "github.com/NeverENG/BanDB/internal/metrics" + "github.com/NeverENG/BanDB/proto" ) // redactedValue 是脱敏字段被替换成的 JSON 值。 diff --git a/service/ingesthook/filter_test.go b/service/ingesthook/filter_test.go index ba48012..12708a2 100644 --- a/service/ingesthook/filter_test.go +++ b/service/ingesthook/filter_test.go @@ -6,7 +6,7 @@ import ( "testing" "github.com/NeverENG/BanDB/bannet" - "github.com/NeverENG/BanDB/pkg/proto" + "github.com/NeverENG/BanDB/proto" ) // fakeReq 是 bannet.Request 的测试替身。钩子不触碰连接，Conn 返回 nil。 diff --git a/service/router.go b/service/router.go index 2101e1b..ba8f3a3 100644 --- a/service/router.go +++ b/service/router.go @@ -7,10 +7,10 @@ import ( "github.com/NeverENG/BanDB/bannet" "github.com/NeverENG/BanDB/cluster" - "github.com/NeverENG/BanDB/pkg/admission" - "github.com/NeverENG/BanDB/pkg/metrics" - "github.com/NeverENG/BanDB/pkg/predicate" - "github.com/NeverENG/BanDB/pkg/proto" + "github.com/NeverENG/BanDB/internal/admission" + "github.com/NeverENG/BanDB/internal/metrics" + "github.com/NeverENG/BanDB/predicate" + "github.com/NeverENG/BanDB/proto" "github.com/NeverENG/BanDB/storage" ) diff --git a/service/scan_integration_test.go b/service/scan_integration_test.go index 4c2454f..d381444 100644 --- a/service/scan_integration_test.go +++ b/service/scan_integration_test.go @@ -5,7 +5,7 @@ import ( "testing" "github.com/NeverENG/BanDB/config" - "github.com/NeverENG/BanDB/pkg/predicate" + "github.com/NeverENG/BanDB/predicate" ) // TestKVServer_Scan 端到端验证边缘查询：写入若干 IMU 帧后，按时间范围 + 谓词扫描， diff --git a/service/shard_routing_integration_test.go b/service/shard_routing_integration_test.go index 03509a8..1073fbe 100644 --- a/service/shard_routing_integration_test.go +++ b/service/shard_routing_integration_test.go @@ -14,8 +14,8 @@ import ( bandb "github.com/NeverENG/BanDB/client" "github.com/NeverENG/BanDB/cluster" "github.com/NeverENG/BanDB/config" - "github.com/NeverENG/BanDB/pkg/predicate" - "github.com/NeverENG/BanDB/pkg/proto" + "github.com/NeverENG/BanDB/predicate" + "github.com/NeverENG/BanDB/proto" ) // memKV 是隔离的内存 KV，用作每个节点的本地 store——从而在一个进程内起多节点、 diff --git a/storage/engine.go b/storage/engine.go index 8a25ec0..00caa78 100644 --- a/storage/engine.go +++ b/storage/engine.go @@ -7,8 +7,8 @@ import ( "sync" "github.com/NeverENG/BanDB/config" - "github.com/NeverENG/BanDB/pkg/credit" - "github.com/NeverENG/BanDB/pkg/metrics" + "github.com/NeverENG/BanDB/internal/credit" + "github.com/NeverENG/BanDB/internal/metrics" ) var ( diff --git a/storage/sstable.go b/storage/sstable.go index 4966e2c..95e40eb 100644 --- a/storage/sstable.go +++ b/storage/sstable.go @@ -1,20 +1,12 @@ +// 本文件是 SSTable 的类型与共用定义：磁盘布局常量、块索引、以及 SSTable 本身。 +// 读路径见 sstable_read.go，写路径与 compaction 见 sstable_write.go，元信息管理见 sstable_meta.go。 package storage import ( - "bufio" - "bytes" - "encoding/binary" - "fmt" - "io" - "log/slog" + "github.com/NeverENG/BanDB/config" "os" - "path/filepath" - "sort" "sync" "sync/atomic" - "time" - - "github.com/NeverENG/BanDB/config" ) const ( @@ -58,14 +50,6 @@ type blockIndex struct { } // blockExtent 返回第 i 块在文件中的 [start,end) 字节范围。 -func (bi *blockIndex) blockExtent(i int) (int64, int64) { - start := bi.entries[i].BlockOffset - if i+1 < len(bi.entries) { - return start, bi.entries[i+1].BlockOffset - } - return start, bi.dataEnd -} - type SSTable struct { // dir 是 SSTable 文件目录，构造时从 config 快照一份。构造在主 goroutine 完成，之后 // 后台 goroutine（如 LoadSSTableMetaList、Flush、Merge）读 ss.dir 而非全局 config.G， @@ -110,801 +94,3 @@ func NewSSTable() *SSTable { } // publishMetas 发布 metas 的不可变快照供读路径无锁获取。调用方须持 ss.mu。 -func (ss *SSTable) publishMetas() { - snap := make([]*SSTableMeta, len(ss.metas)) - copy(snap, ss.metas) - ss.snapshot.Store(&snap) -} - -// openFile 取该路径的常驻只读句柄，未缓存则打开并缓存。返回 nil 表示打开失败。 -func (ss *SSTable) openFile(path string) *os.File { - ss.fdMu.RLock() - f, ok := ss.fdCache[path] - ss.fdMu.RUnlock() - if ok { - return f - } - - ss.fdMu.Lock() - defer ss.fdMu.Unlock() - if f, ok := ss.fdCache[path]; ok { // 双检：并发者可能已填入 - return f - } - f, err := os.Open(path) - if err != nil { - return nil - } - ss.fdCache[path] = f - return f -} - -// closeFile 关闭并剔除该路径的常驻句柄（文件被删除时调用）。 -func (ss *SSTable) closeFile(path string) { - ss.fdMu.Lock() - f, ok := ss.fdCache[path] - delete(ss.fdCache, path) - ss.fdMu.Unlock() - if ok { - f.Close() - } -} - -func (ss *SSTable) LoadSSTableMetaList() { - dir := ss.dir - - if err := os.MkdirAll(dir, dirPerm); err != nil { - slog.Error("cannot create SSTable directory", "error", err) - return - } - - entries, err := os.ReadDir(dir) - if err != nil { - slog.Warn("cannot read SSTable directory", "error", err) - return - } - - metas := make([]*SSTableMeta, 0) - count := 0 - - for _, entry := range entries { - if entry.IsDir() || filepath.Ext(entry.Name()) != ".sst" { - continue - } - - fullPath := filepath.Join(dir, entry.Name()) - - file, err := os.Open(fullPath) - if err != nil { - slog.Warn("failed to open SSTable", "file", entry.Name(), "error", err) - continue - } - - var keyLen uint32 - if err := binary.Read(file, binary.BigEndian, &keyLen); err != nil { - slog.Warn("failed to read key length", "file", entry.Name(), "error", err) - file.Close() - continue - } - - keyBytes := make([]byte, keyLen) - if _, err := io.ReadFull(file, keyBytes); err != nil { - slog.Warn("failed to read key", "file", entry.Name(), "error", err) - file.Close() - continue - } - - file.Close() - - info, err := os.Stat(fullPath) - if err != nil { - slog.Warn("failed to stat SSTable", "file", entry.Name(), "error", err) - continue - } - - meta := &SSTableMeta{ - Level: parseLevelFromName(entry.Name()), // 从文件名恢复 level，避免重启塌缩到 L0 - Filepath: fullPath, - MinKey: keyBytes, - MaxKey: nil, - Size: info.Size(), - MaxKeyKnown: false, - } - - // 从块索引末项直接取 MaxKey（新格式）。不能用 EnsureMeta 的顺序扫描：它把数据段 - // 之后的块索引/布隆/footer 也当记录读，导致 MaxKey 错乱（实测退化成空串）， - // 使 getFromSSTables 的 [MinKey,MaxKey] 范围过滤把命中 key 整段跳过 → 重启后 - // 已 flush 数据全部读不到。老格式无 footer，loadBlockIndexFromFile 返回 nil， - // 保留 MaxKeyLoaded=false 由 EnsureMeta 顺序扫描兜底（对纯数据文件正确）。 - if idx := ss.loadBlockIndexFromFile(fullPath); idx != nil && len(idx.entries) > 0 { - meta.MaxKey = idx.entries[len(idx.entries)-1].LastKey - meta.MaxKeyKnown = true - } - - metas = append(metas, meta) - count++ - } - - // 按创建时间戳升序重建 metas，等价于内存中的创建（append）序——读路径靠 metas 逆序判定 - // newest-wins，按文件名字符串排序会让旧 merged 盖过新 L0 而返回陈旧值。时间戳相同时 - // 退回文件名比较以保证确定性。 - sort.Slice(metas, func(i, j int) bool { - ti := parseCreateTsFromName(filepath.Base(metas[i].Filepath)) - tj := parseCreateTsFromName(filepath.Base(metas[j].Filepath)) - if ti != tj { - return ti < tj - } - return metas[i].Filepath < metas[j].Filepath - }) - - ss.mu.Lock() - ss.metas = metas - ss.publishMetas() - ss.mu.Unlock() - - for _, meta := range metas { - go ss.getBlockIndex(meta.Filepath) // 异步预热块索引 - go ss.getBloom(meta.Filepath) // 异步预热布隆过滤器 - } - - slog.Info("SSTable index loaded", "files", count, "dir", dir) -} - -// WriteToSSTable 将有序 entries 写入 SSTable 文件（含块索引） -func (ss *SSTable) WriteToSSTable(entries []LogEntry) error { - if len(entries) == 0 { - return ErrNoEntries - } - - // L0：memtable flush 落到 level 0；level 编进文件名以便重启恢复。 - filename := fmt.Sprintf("sstable_L0_%d.sst", time.Now().UnixNano()) - dir := ss.dir - if err := os.MkdirAll(dir, dirPerm); err != nil { - return fmt.Errorf("create data directory failed: %v", err) - } - fullPath := filepath.Join(dir, filename) - - file, err := os.Create(fullPath) - if err != nil { - return fmt.Errorf("create SSTable file failed: %v", err) - } - defer file.Close() - - // 构建数据 buffer + 块索引 - var blockIdx []blockMeta - - // 数据段先攒在内存 buffer 里再一次写出。此处的 binary.Write 不检查错误是安全的： - // 目标是 bytes.Buffer，其 Write 按文档永不返回错误（容量不足时直接 panic）。 - var buf bytes.Buffer - for i, entry := range entries { - bi := i / SSTableBlockSize - if i%SSTableBlockSize == 0 { - blockIdx = append(blockIdx, blockMeta{blockOffset: int64(buf.Len())}) - } - blockIdx[bi].lastKey = entry.Key - - binary.Write(&buf, binary.BigEndian, uint32(len(entry.Key))) - buf.Write(entry.Key) - if entry.Value == nil { // 墓碑：仅写哨兵长度，无 value 字节 - binary.Write(&buf, binary.BigEndian, tombstoneValLen) - } else { - binary.Write(&buf, binary.BigEndian, uint32(len(entry.Value))) - buf.Write(entry.Value) - } - } - - // 写数据 - if _, err := file.Write(buf.Bytes()); err != nil { - return err - } - - // 数据区已写完，此处的偏移即块索引起点，也是读路径推算末块长度的依据。 - indexStart, err := file.Seek(0, io.SeekCurrent) - if err != nil { - return fmt.Errorf("locate index offset failed: %w", err) - } - - keys := make([][]byte, len(entries)) - for i := range entries { - keys[i] = entries[i].Key - } - pb, err := writeTail(file, blockIdx, keys, indexStart) - if err != nil { - return fmt.Errorf("write SSTable tail failed: %w", err) - } - - // 缓存块索引 - cache := make([]BlockIndexEntry, len(blockIdx)) - for i, b := range blockIdx { - cache[i] = BlockIndexEntry{LastKey: b.lastKey, BlockOffset: b.blockOffset} - } - ss.idxMu.Lock() - ss.indexCache[fullPath] = &blockIndex{entries: cache, dataEnd: indexStart} - ss.idxMu.Unlock() - - if err := file.Sync(); err != nil { - return fmt.Errorf("sync SSTable file failed: %v", err) - } - ss.cacheBloom(fullPath, pb) // 落盘后再缓存 - - info, err := file.Stat() - if err != nil { - return fmt.Errorf("stat SSTable file failed: %v", err) - } - meta := &SSTableMeta{ - Level: 0, - Filepath: fullPath, - MinKey: entries[0].Key, - MaxKey: entries[len(entries)-1].Key, - Size: info.Size(), - MaxKeyKnown: true, - } - ss.AddMeta(meta) - flushBytesWritten.Add(info.Size()) - return nil -} - -// Metas 返回元数据的不可变快照，按落盘先后升序（最旧在前）。 -// 无锁零拷贝；调用方只可读取，不得原地修改。切片元素是共享指针，非对象副本。 -func (ss *SSTable) Metas() []*SSTableMeta { - if snap := ss.snapshot.Load(); snap != nil { - return *snap - } - return nil -} - -func (ss *SSTable) ReadAllFromSSTable(filepath string) ([]*LogEntry, error) { - file, err := os.Open(filepath) - if err != nil { - return nil, err - } - defer file.Close() - - // 有 footer 时数据区的终点是已知的；没有 footer（老格式，或尾部残缺）则读到 EOF。 - // - // 无论探测结果如何都必须把偏移移回开头：readDataEndOffset 为读 footer 已经 seek 到了 - // 文件末尾附近。此前这一步写在 dataEnd > 0 的条件里，于是无 footer 的文件从末尾开始 - // 解析，恒返回 0 条记录——老格式的全量读回退路径实际从未生效。 - dataEnd := ss.readDataEndOffset(file) - if _, err := file.Seek(0, io.SeekStart); err != nil { - return nil, err - } - - // 解析不下去时就地停止，返回已解出的条目，而不是让整个读取失败。 - // - // 这与 WAL.Replay 对撕裂尾写的处理一致，理由也相同：数据区总是先于尾部写出，故无法 - // 解析的字节只可能出现在有效数据之后。此前这里 return nil, err，而唯一的调用方 - // readFromSSTableFull 丢弃该错误后遍历 nil 切片——尾部一旦残缺，整个文件的 key 全部 - // 读成「不存在」，且不报错。 - entries := make([]*LogEntry, 0) - for { - if dataEnd > 0 { - pos, err := file.Seek(0, io.SeekCurrent) - if err != nil { - return nil, err - } - if pos >= dataEnd { - break - } - } - - var keyLen uint32 - if err := binary.Read(file, binary.BigEndian, &keyLen); err != nil { - break - } - keyBytes := make([]byte, keyLen) - if _, err := io.ReadFull(file, keyBytes); err != nil { - break - } - - var valueLen uint32 - if err := binary.Read(file, binary.BigEndian, &valueLen); err != nil { - break - } - var valueBytes []byte // 墓碑(哨兵长度)还原为 nil，无 value 字节 - if valueLen != tombstoneValLen { - valueBytes = make([]byte, valueLen) - if _, err := io.ReadFull(file, valueBytes); err != nil { - break - } - } - - entries = append(entries, &LogEntry{Key: keyBytes, Value: valueBytes}) - } - - return entries, nil -} - -// readDataEndOffset 读 Footer 返回数据结束偏移，老格式返回 -1 -func (ss *SSTable) readDataEndOffset(f *os.File) int64 { - return sstableDataEnd(f) -} - -func (ss *SSTable) ReadFromSSTable(filepath string, key []byte) ([]byte, bool) { - // 布隆过滤器快速否决：明确不存在则直接返回，省去磁盘读 - if bloom := ss.getBloom(filepath); bloom != nil && !bloom.MayContain(key) { - return nil, false - } - if idx := ss.getBlockIndex(filepath); idx != nil { - return ss.searchBlock(filepath, key, idx) - } - // 老格式 fallback - return ss.readFromSSTableFull(filepath, key) -} - -// writeBloomSection 在块索引之后写入分区布隆过滤器及 trailer，返回构建的 -// 过滤器供调用方在 file.Sync() 之后再写入缓存——避免崩溃时出现「缓存说有 -// 但文件未落盘」的不一致。 -// 文件布局: ...[BlockIndex][BloomBlob][BloomLen(8B)][BloomMagic(4B)][Footer] -func writeBloomSection(w io.Writer, keys [][]byte) (*PartitionedBloom, error) { - pb := BuildPartitionedBloom(keys, DefaultNamespaceSep, defaultBloomFPRate) - blob := pb.Encode() - if _, err := w.Write(blob); err != nil { - return nil, err - } - if err := binary.Write(w, binary.BigEndian, uint64(len(blob))); err != nil { - return nil, err - } - if err := binary.Write(w, binary.BigEndian, bloomTrailerMagic); err != nil { - return nil, err - } - return pb, nil -} - -// writeTail 写出 SSTable 的尾部三段：块索引、布隆过滤器、Footer。 -// -// 由 WriteToSSTable 与 MergeSSTable 共用，二者的字节布局因此不可能漂移——读路径只有 -// 一份解析实现，写路径也应只有一份。 -// -// 每一次写入都检查错误。此前这里的 binary.Write/Write 返回值被丢弃，后果并非「少写了 -// 几个字节」而是静默的数据丢失：尾部残缺的文件会让 footer magic 校验失败，重启时 -// EnsureMeta 在新格式文件上算出错误的 MaxKey，[MinKey,MaxKey] 过滤随即跳过整个文件； -// 而调用方以为落盘成功——Flush 会丢弃内存副本，compaction 会删除源文件。 -func writeTail(w io.Writer, blocks []blockMeta, keys [][]byte, indexStart int64) (*PartitionedBloom, error) { - // 块索引: [LastKeyLen(4B)][LastKey][BlockOffset(8B)] × N - for _, b := range blocks { - if err := binary.Write(w, binary.BigEndian, uint32(len(b.lastKey))); err != nil { - return nil, fmt.Errorf("write block index key length: %w", err) - } - if _, err := w.Write(b.lastKey); err != nil { - return nil, fmt.Errorf("write block index key: %w", err) - } - if err := binary.Write(w, binary.BigEndian, b.blockOffset); err != nil { - return nil, fmt.Errorf("write block index offset: %w", err) - } - } - - pb, err := writeBloomSection(w, keys) - if err != nil { - return nil, fmt.Errorf("write bloom section: %w", err) - } - - // Footer: BlockCount(4B) + IndexOffset(8B) + Magic(4B) - if err := binary.Write(w, binary.BigEndian, uint32(len(blocks))); err != nil { - return nil, fmt.Errorf("write footer block count: %w", err) - } - if err := binary.Write(w, binary.BigEndian, indexStart); err != nil { - return nil, fmt.Errorf("write footer index offset: %w", err) - } - if err := binary.Write(w, binary.BigEndian, indexFooterMagic); err != nil { - return nil, fmt.Errorf("write footer magic: %w", err) - } - return pb, nil -} - -// cacheBloom 将过滤器写入缓存（应在 file.Sync() 成功后调用）。 -func (ss *SSTable) cacheBloom(fullPath string, pb *PartitionedBloom) { - ss.bloomMu.Lock() - ss.bloomCache[fullPath] = pb - ss.bloomMu.Unlock() -} - -// getBloom 从缓存取布隆过滤器；miss 时从文件加载（老格式返回并缓存 nil）。 -func (ss *SSTable) getBloom(filepath string) *PartitionedBloom { - ss.bloomMu.RLock() - pb, ok := ss.bloomCache[filepath] - ss.bloomMu.RUnlock() - if ok { - return pb - } - pb = ss.loadBloomFromFile(filepath) - ss.bloomMu.Lock() - ss.bloomCache[filepath] = pb // 可能为 nil(老格式)，缓存避免重复读盘 - ss.bloomMu.Unlock() - return pb -} - -// loadBloomFromFile 读取紧邻索引 Footer 之前的布隆 trailer 与 blob。 -// 全程用 SeekEnd 负偏移定位，不依赖 Stat().Size()，从而消除 -// 「Stat 取大小 → Seek 读内容」之间文件被改写/截断的竞争窗口。 -// 老格式(无 trailer，magic 不匹配)或 bloomLen 越界返回 nil。 -func (ss *SSTable) loadBloomFromFile(filepath string) *PartitionedBloom { - f, err := os.Open(filepath) - if err != nil { - return nil - } - defer f.Close() - - // 布隆 trailer 紧邻 16B 索引 Footer 之前 - if _, err := f.Seek(-(indexFooterSize + bloomTrailerSize), io.SeekEnd); err != nil { - return nil // 文件比 footer+trailer 还短(含老格式小文件) - } - var bloomLen uint64 - var magic uint32 - if err := binary.Read(f, binary.BigEndian, &bloomLen); err != nil { - return nil - } - if err := binary.Read(f, binary.BigEndian, &magic); err != nil { - return nil - } - if magic != bloomTrailerMagic || bloomLen == 0 || bloomLen > maxBloomSectionBytes { - return nil // 老格式 magic 不匹配，或 bloomLen 损坏/越界 - } - - // blob 紧邻 trailer 之前，同样用 SeekEnd 负偏移定位 - if _, err := f.Seek(-(indexFooterSize + bloomTrailerSize + int64(bloomLen)), io.SeekEnd); err != nil { - return nil - } - blob := make([]byte, bloomLen) - if _, err := io.ReadFull(f, blob); err != nil { - return nil - } - pb, err := DecodePartitionedBloom(blob, 0, defaultBloomFPRate) - if err != nil { - return nil - } - return pb -} - -// getBlockIndex 从缓存取块索引，miss 时从文件加载 -func (ss *SSTable) getBlockIndex(filepath string) *blockIndex { - ss.idxMu.RLock() - idx, ok := ss.indexCache[filepath] - ss.idxMu.RUnlock() - if ok { - return idx - } - idx = ss.loadBlockIndexFromFile(filepath) - if idx == nil { - return nil - } - ss.idxMu.Lock() - ss.indexCache[filepath] = idx - ss.idxMu.Unlock() - return idx -} - -// loadBlockIndexFromFile 从文件末尾读取块索引及数据区结束偏移 -func (ss *SSTable) loadBlockIndexFromFile(filepath string) *blockIndex { - f, err := os.Open(filepath) - if err != nil { - return nil - } - defer f.Close() - - if _, err := f.Seek(-indexFooterSize, io.SeekEnd); err != nil { - return nil - } - var blockCount uint32 - var indexOffset int64 - var magic uint32 - binary.Read(f, binary.BigEndian, &blockCount) - binary.Read(f, binary.BigEndian, &indexOffset) - binary.Read(f, binary.BigEndian, &magic) - - if magic != indexFooterMagic || blockCount == 0 || indexOffset <= 0 { - return nil - } - - if _, err := f.Seek(indexOffset, io.SeekStart); err != nil { - return nil - } - entries := make([]BlockIndexEntry, blockCount) - for i := uint32(0); i < blockCount; i++ { - var keyLen uint32 - if err := binary.Read(f, binary.BigEndian, &keyLen); err != nil { - return nil - } - key := make([]byte, keyLen) - if _, err := io.ReadFull(f, key); err != nil { - return nil - } - var offset int64 - if err := binary.Read(f, binary.BigEndian, &offset); err != nil { - return nil - } - entries[i] = BlockIndexEntry{LastKey: key, BlockOffset: offset} - } - // indexOffset 既是块索引起点，也是数据区终点——即最后一块的结束偏移。 - return &blockIndex{entries: entries, dataEnd: indexOffset} -} - -// searchBlock 二分块索引定位目标块，单次 ReadAt 读入整块后在内存中扫描。 -// 每次点查恒定一次 read syscall，与块内条目数无关。 -func (ss *SSTable) searchBlock(filepath string, key []byte, bi *blockIndex) ([]byte, bool) { - idx := bi.entries - lo, hi := 0, len(idx)-1 - for lo < hi { - mid := (lo + hi) / 2 - if bytes.Compare(key, idx[mid].LastKey) <= 0 { - hi = mid - } else { - lo = mid + 1 - } - } - if lo >= len(idx) { - return nil, false - } - - start, end := bi.blockExtent(lo) - if end <= start { - return nil, false - } - - // 命中缓存的块由多个读者共享，故下方扫描只读、且命中时对 value 另行拷贝。 - buf, cached := ss.blocks.get(filepath, start) - if !cached { - f := ss.openFile(filepath) - if f == nil { - return nil, false - } - buf = make([]byte, end-start) - // ReadAt 不使用文件内偏移，故共享句柄可被并发读者安全复用。 - if _, err := f.ReadAt(buf, start); err != nil { - return nil, false - } - ss.blocks.put(filepath, start, buf) - } - - for off := 0; off+4 <= len(buf); { - kLen := int(binary.BigEndian.Uint32(buf[off:])) - off += 4 - if kLen < 0 || off+kLen+4 > len(buf) { - break - } - k := buf[off : off+kLen] - off += kLen - vLen := binary.BigEndian.Uint32(buf[off:]) - off += 4 - - tomb := vLen == tombstoneValLen - var v []byte - if !tomb { // 墓碑无 value 字节 - if off+int(vLen) > len(buf) { - break - } - v = buf[off : off+int(vLen)] - off += int(vLen) - } - - cmp := bytes.Compare(k, key) - if cmp == 0 { - if tomb { // 命中墓碑：found 但已删除 - return nil, true - } - // 必须拷贝：v 是整块 buf 的子切片，直接返回会让整块常驻内存。 - // 用 make+copy 而非 append([]byte(nil), ...)：后者在 v 为空时返回 nil， - // 会把「空 value」误变成「墓碑」（约定 nil=墓碑）。 - out := make([]byte, len(v)) - copy(out, v) - return out, true - } - if cmp > 0 { - break - } - } - return nil, false -} - -// readFromSSTableFull 老格式文件全量读取（兼容） -func (ss *SSTable) readFromSSTableFull(filepath string, key []byte) ([]byte, bool) { - entries, _ := ss.ReadAllFromSSTable(filepath) - for _, entry := range entries { - if bytes.Equal(entry.Key, key) { - return entry.Value, true - } - } - return nil, false -} - -// MergeSSTable 把 files 归并为 targetLevel 上的一个新 SSTable，即 compaction 的落地实现。 -// -// 正确性不变量（慎改）：同 key 的 tie-break 由 files 的 srcIdx 决定（越大越新，见下）， -// compaction 输出文件用 time.Now() 打新 ts。这依赖调用方 CompactSSTable「一次 compaction 卷走整层全部文件」： -// 因此同 key 的多个版本要么都在本次输入里（srcIdx 定新旧），要么更新的版本在更浅层的 -// 更晚 ts 文件里——读路径据此按创建 ts 判 newest-wins（见 memtable.go getFromSSTables）。 -// 若引入 overlap-aware / 部分选择或跨层 compaction，必须同时确保「低 level = 更新」进入 tie-break， -// 并把 recency 显式化（per-key 序号），否则 compaction 会静默选中陈旧值。 -func (ss *SSTable) MergeSSTable(files []*SSTableMeta, targetLevel int) *SSTableMeta { - if len(files) == 0 { - return nil - } - - slog.Info("merging SSTable files", "files", len(files), "targetLevel", targetLevel) - - // 为每个源文件打开流式迭代器（srcIdx = 在 files 中的序号，越大越新） - iters := make([]*sstableIterator, 0, len(files)) - for _, meta := range files { - it, err := newSSTableIterator(meta.Filepath) - if err != nil { - slog.Error("failed to open SSTable iterator for merge", "file", meta.Filepath, "error", err) - for _, opened := range iters { - opened.Close() - } - return nil - } - iters = append(iters, it) - } - - mi, err := newMergeIterator(iters) - if err != nil { - mi.Close() - slog.Error("failed to init merge iterator", "error", err) - return nil - } - defer mi.Close() - - // targetLevel 编进文件名，使重启（LoadSSTableMetaList）能恢复该文件的 level。 - filename := fmt.Sprintf("sstable_merged_L%d_%d.sst", targetLevel, time.Now().UnixNano()) - dir := ss.dir - fullPath := filepath.Join(dir, filename) - - file, err := os.Create(fullPath) - if err != nil { - slog.Error("failed to create merged SSTable", "error", err) - return nil - } - defer file.Close() - - // K 路归并流式写出：value 直接落盘，仅累积块索引(每块一条)与 key(供布隆)， - // 不再把全部源条目读入内存。 - var blockIdx []blockMeta - var keys [][]byte - var minKey, maxKey []byte - var dataOffset int64 - // 数据段经 bufio 写出。循环内的 Write 不逐个检查错误是安全的：bufio.Writer 记住 - // 首个错误并在其后所有 Write 与 Flush 上返回它，而下方 bw.Flush() 的错误是被检查的。 - bw := bufio.NewWriter(file) - - count := 0 - for mi.Next() { - k := mi.Key() - v := mi.Value() - bi := count / SSTableBlockSize - if count%SSTableBlockSize == 0 { - blockIdx = append(blockIdx, blockMeta{blockOffset: dataOffset}) - } - blockIdx[bi].lastKey = k - - var hdr [4]byte - binary.BigEndian.PutUint32(hdr[:], uint32(len(k))) - bw.Write(hdr[:]) - bw.Write(k) - if v == nil { // 墓碑：写哨兵长度，无 value 字节 - binary.BigEndian.PutUint32(hdr[:], tombstoneValLen) - bw.Write(hdr[:]) - dataOffset += int64(8 + len(k)) - } else { - binary.BigEndian.PutUint32(hdr[:], uint32(len(v))) - bw.Write(hdr[:]) - if _, werr := bw.Write(v); werr != nil { - slog.Error("failed to write merged entry", "error", werr) - return nil - } - dataOffset += int64(8 + len(k) + len(v)) - } - - keys = append(keys, k) // k 为迭代器新分配，安全持有 - if count == 0 { - minKey = k - } - maxKey = k - count++ - } - if err := mi.Err(); err != nil { - slog.Error("merge iteration failed", "error", err) - return nil - } - if count == 0 { - slog.Warn("no entries to merge") - return nil - } - if err := bw.Flush(); err != nil { - slog.Error("failed to flush merged data", "error", err) - return nil - } - - indexStart, err := file.Seek(0, io.SeekCurrent) - if err != nil { - slog.Error("failed to locate index offset for merged SSTable", "error", err) - return nil - } - pb, err := writeTail(file, blockIdx, keys, indexStart) - if err != nil { - slog.Error("failed to write merged SSTable tail", "error", err) - return nil - } - - cache := make([]BlockIndexEntry, len(blockIdx)) - for i, b := range blockIdx { - cache[i] = BlockIndexEntry{LastKey: b.lastKey, BlockOffset: b.blockOffset} - } - ss.idxMu.Lock() - ss.indexCache[fullPath] = &blockIndex{entries: cache, dataEnd: indexStart} - ss.idxMu.Unlock() - - if err := file.Sync(); err != nil { - slog.Error("failed to sync merged SSTable", "error", err) - return nil - } - ss.cacheBloom(fullPath, pb) // 落盘后再缓存 - - info, err := file.Stat() - if err != nil { - slog.Error("failed to stat merged SSTable", "error", err) - return nil - } - - newMeta := &SSTableMeta{ - Level: targetLevel, - Filepath: fullPath, - MinKey: minKey, - MaxKey: maxKey, - Size: info.Size(), - MaxKeyKnown: true, - } - - ss.AddMeta(newMeta) - compactionBytesWritten.Add(info.Size()) - slog.Info("SSTable merged", "level", targetLevel, "file", filename, "keys", count, "size", info.Size()) - - return newMeta -} -func (ss *SSTable) AddMeta(meta *SSTableMeta) { - ss.mu.Lock() - defer ss.mu.Unlock() - ss.metas = append(ss.metas, meta) - ss.publishMetas() -} - -func (ss *SSTable) RemoveMeta(target *SSTableMeta) { - ss.mu.Lock() - defer ss.mu.Unlock() - - for i, meta := range ss.metas { - if meta == target { - ss.metas = append(ss.metas[:i], ss.metas[i+1:]...) - ss.publishMetas() - return - } - } -} - -// LevelFiles 获取指定层级的文件列表 -func (ss *SSTable) LevelFiles(level int) []*SSTableMeta { - ss.mu.RLock() - defer ss.mu.RUnlock() - - var result []*SSTableMeta - for _, meta := range ss.metas { - if meta.Level == level { - result = append(result, meta) - } - } - return result -} - -func (ss *SSTable) DeleteSSTable(meta *SSTableMeta) { - if err := os.Remove(meta.Filepath); err != nil { - slog.Warn("failed to delete SSTable", "file", meta.Filepath, "error", err) - } - // 与索引/布隆缓存同时剔除常驻句柄与已缓存数据块，否则每轮 compaction 泄漏一个 fd、 - // 并让已删文件的块长期占用缓存预算。 - ss.closeFile(meta.Filepath) - ss.blocks.dropFile(meta.Filepath) - ss.idxMu.Lock() - delete(ss.indexCache, meta.Filepath) - ss.idxMu.Unlock() - ss.bloomMu.Lock() - delete(ss.bloomCache, meta.Filepath) - ss.bloomMu.Unlock() -} diff --git a/storage/sstable_meta.go b/storage/sstable_meta.go new file mode 100644 index 0000000..00558da --- /dev/null +++ b/storage/sstable_meta.go @@ -0,0 +1,176 @@ +// 本文件是 SSTable 的元信息管理：启动时扫描目录重建元信息，以及增删查与不可变快照发布。 +package storage + +import ( + "encoding/binary" + "io" + "log/slog" + "os" + "path/filepath" + "sort" +) + +func (ss *SSTable) publishMetas() { + snap := make([]*SSTableMeta, len(ss.metas)) + copy(snap, ss.metas) + ss.snapshot.Store(&snap) +} + +// openFile 取该路径的常驻只读句柄，未缓存则打开并缓存。返回 nil 表示打开失败。 +func (ss *SSTable) LoadSSTableMetaList() { + dir := ss.dir + + if err := os.MkdirAll(dir, dirPerm); err != nil { + slog.Error("cannot create SSTable directory", "error", err) + return + } + + entries, err := os.ReadDir(dir) + if err != nil { + slog.Warn("cannot read SSTable directory", "error", err) + return + } + + metas := make([]*SSTableMeta, 0) + count := 0 + + for _, entry := range entries { + if entry.IsDir() || filepath.Ext(entry.Name()) != ".sst" { + continue + } + + fullPath := filepath.Join(dir, entry.Name()) + + file, err := os.Open(fullPath) + if err != nil { + slog.Warn("failed to open SSTable", "file", entry.Name(), "error", err) + continue + } + + var keyLen uint32 + if err := binary.Read(file, binary.BigEndian, &keyLen); err != nil { + slog.Warn("failed to read key length", "file", entry.Name(), "error", err) + file.Close() + continue + } + + keyBytes := make([]byte, keyLen) + if _, err := io.ReadFull(file, keyBytes); err != nil { + slog.Warn("failed to read key", "file", entry.Name(), "error", err) + file.Close() + continue + } + + file.Close() + + info, err := os.Stat(fullPath) + if err != nil { + slog.Warn("failed to stat SSTable", "file", entry.Name(), "error", err) + continue + } + + meta := &SSTableMeta{ + Level: parseLevelFromName(entry.Name()), // 从文件名恢复 level，避免重启塌缩到 L0 + Filepath: fullPath, + MinKey: keyBytes, + MaxKey: nil, + Size: info.Size(), + MaxKeyKnown: false, + } + + // 从块索引末项直接取 MaxKey（新格式）。不能用 EnsureMeta 的顺序扫描：它把数据段 + // 之后的块索引/布隆/footer 也当记录读，导致 MaxKey 错乱（实测退化成空串）， + // 使 getFromSSTables 的 [MinKey,MaxKey] 范围过滤把命中 key 整段跳过 → 重启后 + // 已 flush 数据全部读不到。老格式无 footer，loadBlockIndexFromFile 返回 nil， + // 保留 MaxKeyLoaded=false 由 EnsureMeta 顺序扫描兜底（对纯数据文件正确）。 + if idx := ss.loadBlockIndexFromFile(fullPath); idx != nil && len(idx.entries) > 0 { + meta.MaxKey = idx.entries[len(idx.entries)-1].LastKey + meta.MaxKeyKnown = true + } + + metas = append(metas, meta) + count++ + } + + // 按创建时间戳升序重建 metas，等价于内存中的创建（append）序——读路径靠 metas 逆序判定 + // newest-wins，按文件名字符串排序会让旧 merged 盖过新 L0 而返回陈旧值。时间戳相同时 + // 退回文件名比较以保证确定性。 + sort.Slice(metas, func(i, j int) bool { + ti := parseCreateTsFromName(filepath.Base(metas[i].Filepath)) + tj := parseCreateTsFromName(filepath.Base(metas[j].Filepath)) + if ti != tj { + return ti < tj + } + return metas[i].Filepath < metas[j].Filepath + }) + + ss.mu.Lock() + ss.metas = metas + ss.publishMetas() + ss.mu.Unlock() + + for _, meta := range metas { + go ss.getBlockIndex(meta.Filepath) // 异步预热块索引 + go ss.getBloom(meta.Filepath) // 异步预热布隆过滤器 + } + + slog.Info("SSTable index loaded", "files", count, "dir", dir) +} + +// WriteToSSTable 将有序 entries 写入 SSTable 文件（含块索引） +func (ss *SSTable) Metas() []*SSTableMeta { + if snap := ss.snapshot.Load(); snap != nil { + return *snap + } + return nil +} + +func (ss *SSTable) AddMeta(meta *SSTableMeta) { + ss.mu.Lock() + defer ss.mu.Unlock() + ss.metas = append(ss.metas, meta) + ss.publishMetas() +} + +func (ss *SSTable) RemoveMeta(target *SSTableMeta) { + ss.mu.Lock() + defer ss.mu.Unlock() + + for i, meta := range ss.metas { + if meta == target { + ss.metas = append(ss.metas[:i], ss.metas[i+1:]...) + ss.publishMetas() + return + } + } +} + +// LevelFiles 获取指定层级的文件列表 +func (ss *SSTable) LevelFiles(level int) []*SSTableMeta { + ss.mu.RLock() + defer ss.mu.RUnlock() + + var result []*SSTableMeta + for _, meta := range ss.metas { + if meta.Level == level { + result = append(result, meta) + } + } + return result +} + +func (ss *SSTable) DeleteSSTable(meta *SSTableMeta) { + if err := os.Remove(meta.Filepath); err != nil { + slog.Warn("failed to delete SSTable", "file", meta.Filepath, "error", err) + } + // 与索引/布隆缓存同时剔除常驻句柄与已缓存数据块，否则每轮 compaction 泄漏一个 fd、 + // 并让已删文件的块长期占用缓存预算。 + ss.closeFile(meta.Filepath) + ss.blocks.dropFile(meta.Filepath) + ss.idxMu.Lock() + delete(ss.indexCache, meta.Filepath) + ss.idxMu.Unlock() + ss.bloomMu.Lock() + delete(ss.bloomCache, meta.Filepath) + ss.bloomMu.Unlock() +} diff --git a/storage/sstable_read.go b/storage/sstable_read.go new file mode 100644 index 0000000..2371fff --- /dev/null +++ b/storage/sstable_read.go @@ -0,0 +1,355 @@ +// 本文件是 SSTable 的读路径：常驻句柄、块索引与布隆缓存、点查与全量读。 +package storage + +import ( + "bytes" + "encoding/binary" + "io" + "os" +) + +func (bi *blockIndex) blockExtent(i int) (int64, int64) { + start := bi.entries[i].BlockOffset + if i+1 < len(bi.entries) { + return start, bi.entries[i+1].BlockOffset + } + return start, bi.dataEnd +} + +func (ss *SSTable) openFile(path string) *os.File { + ss.fdMu.RLock() + f, ok := ss.fdCache[path] + ss.fdMu.RUnlock() + if ok { + return f + } + + ss.fdMu.Lock() + defer ss.fdMu.Unlock() + if f, ok := ss.fdCache[path]; ok { // 双检：并发者可能已填入 + return f + } + f, err := os.Open(path) + if err != nil { + return nil + } + ss.fdCache[path] = f + return f +} + +// closeFile 关闭并剔除该路径的常驻句柄（文件被删除时调用）。 +func (ss *SSTable) closeFile(path string) { + ss.fdMu.Lock() + f, ok := ss.fdCache[path] + delete(ss.fdCache, path) + ss.fdMu.Unlock() + if ok { + f.Close() + } +} + +func (ss *SSTable) ReadAllFromSSTable(filepath string) ([]*LogEntry, error) { + file, err := os.Open(filepath) + if err != nil { + return nil, err + } + defer file.Close() + + // 有 footer 时数据区的终点是已知的；没有 footer（老格式，或尾部残缺）则读到 EOF。 + // + // 无论探测结果如何都必须把偏移移回开头：readDataEndOffset 为读 footer 已经 seek 到了 + // 文件末尾附近。此前这一步写在 dataEnd > 0 的条件里，于是无 footer 的文件从末尾开始 + // 解析，恒返回 0 条记录——老格式的全量读回退路径实际从未生效。 + dataEnd := ss.readDataEndOffset(file) + if _, err := file.Seek(0, io.SeekStart); err != nil { + return nil, err + } + + // 解析不下去时就地停止，返回已解出的条目，而不是让整个读取失败。 + // + // 这与 WAL.Replay 对撕裂尾写的处理一致，理由也相同：数据区总是先于尾部写出，故无法 + // 解析的字节只可能出现在有效数据之后。此前这里 return nil, err，而唯一的调用方 + // readFromSSTableFull 丢弃该错误后遍历 nil 切片——尾部一旦残缺，整个文件的 key 全部 + // 读成「不存在」，且不报错。 + entries := make([]*LogEntry, 0) + for { + if dataEnd > 0 { + pos, err := file.Seek(0, io.SeekCurrent) + if err != nil { + return nil, err + } + if pos >= dataEnd { + break + } + } + + var keyLen uint32 + if err := binary.Read(file, binary.BigEndian, &keyLen); err != nil { + break + } + keyBytes := make([]byte, keyLen) + if _, err := io.ReadFull(file, keyBytes); err != nil { + break + } + + var valueLen uint32 + if err := binary.Read(file, binary.BigEndian, &valueLen); err != nil { + break + } + var valueBytes []byte // 墓碑(哨兵长度)还原为 nil，无 value 字节 + if valueLen != tombstoneValLen { + valueBytes = make([]byte, valueLen) + if _, err := io.ReadFull(file, valueBytes); err != nil { + break + } + } + + entries = append(entries, &LogEntry{Key: keyBytes, Value: valueBytes}) + } + + return entries, nil +} + +// readDataEndOffset 读 Footer 返回数据结束偏移，老格式返回 -1 +func (ss *SSTable) readDataEndOffset(f *os.File) int64 { + return sstableDataEnd(f) +} + +func (ss *SSTable) ReadFromSSTable(filepath string, key []byte) ([]byte, bool) { + // 布隆过滤器快速否决：明确不存在则直接返回，省去磁盘读 + if bloom := ss.getBloom(filepath); bloom != nil && !bloom.MayContain(key) { + return nil, false + } + if idx := ss.getBlockIndex(filepath); idx != nil { + return ss.searchBlock(filepath, key, idx) + } + // 老格式 fallback + return ss.readFromSSTableFull(filepath, key) +} + +// writeBloomSection 在块索引之后写入分区布隆过滤器及 trailer，返回构建的 +// 过滤器供调用方在 file.Sync() 之后再写入缓存——避免崩溃时出现「缓存说有 +// 但文件未落盘」的不一致。 +// 文件布局: ...[BlockIndex][BloomBlob][BloomLen(8B)][BloomMagic(4B)][Footer] +func (ss *SSTable) cacheBloom(fullPath string, pb *PartitionedBloom) { + ss.bloomMu.Lock() + ss.bloomCache[fullPath] = pb + ss.bloomMu.Unlock() +} + +// getBloom 从缓存取布隆过滤器；miss 时从文件加载（老格式返回并缓存 nil）。 +func (ss *SSTable) getBloom(filepath string) *PartitionedBloom { + ss.bloomMu.RLock() + pb, ok := ss.bloomCache[filepath] + ss.bloomMu.RUnlock() + if ok { + return pb + } + pb = ss.loadBloomFromFile(filepath) + ss.bloomMu.Lock() + ss.bloomCache[filepath] = pb // 可能为 nil(老格式)，缓存避免重复读盘 + ss.bloomMu.Unlock() + return pb +} + +// loadBloomFromFile 读取紧邻索引 Footer 之前的布隆 trailer 与 blob。 +// 全程用 SeekEnd 负偏移定位，不依赖 Stat().Size()，从而消除 +// 「Stat 取大小 → Seek 读内容」之间文件被改写/截断的竞争窗口。 +// 老格式(无 trailer，magic 不匹配)或 bloomLen 越界返回 nil。 +func (ss *SSTable) loadBloomFromFile(filepath string) *PartitionedBloom { + f, err := os.Open(filepath) + if err != nil { + return nil + } + defer f.Close() + + // 布隆 trailer 紧邻 16B 索引 Footer 之前 + if _, err := f.Seek(-(indexFooterSize + bloomTrailerSize), io.SeekEnd); err != nil { + return nil // 文件比 footer+trailer 还短(含老格式小文件) + } + var bloomLen uint64 + var magic uint32 + if err := binary.Read(f, binary.BigEndian, &bloomLen); err != nil { + return nil + } + if err := binary.Read(f, binary.BigEndian, &magic); err != nil { + return nil + } + if magic != bloomTrailerMagic || bloomLen == 0 || bloomLen > maxBloomSectionBytes { + return nil // 老格式 magic 不匹配，或 bloomLen 损坏/越界 + } + + // blob 紧邻 trailer 之前，同样用 SeekEnd 负偏移定位 + if _, err := f.Seek(-(indexFooterSize + bloomTrailerSize + int64(bloomLen)), io.SeekEnd); err != nil { + return nil + } + blob := make([]byte, bloomLen) + if _, err := io.ReadFull(f, blob); err != nil { + return nil + } + pb, err := DecodePartitionedBloom(blob, 0, defaultBloomFPRate) + if err != nil { + return nil + } + return pb +} + +// getBlockIndex 从缓存取块索引，miss 时从文件加载 +func (ss *SSTable) getBlockIndex(filepath string) *blockIndex { + ss.idxMu.RLock() + idx, ok := ss.indexCache[filepath] + ss.idxMu.RUnlock() + if ok { + return idx + } + idx = ss.loadBlockIndexFromFile(filepath) + if idx == nil { + return nil + } + ss.idxMu.Lock() + ss.indexCache[filepath] = idx + ss.idxMu.Unlock() + return idx +} + +// loadBlockIndexFromFile 从文件末尾读取块索引及数据区结束偏移 +func (ss *SSTable) loadBlockIndexFromFile(filepath string) *blockIndex { + f, err := os.Open(filepath) + if err != nil { + return nil + } + defer f.Close() + + if _, err := f.Seek(-indexFooterSize, io.SeekEnd); err != nil { + return nil + } + var blockCount uint32 + var indexOffset int64 + var magic uint32 + binary.Read(f, binary.BigEndian, &blockCount) + binary.Read(f, binary.BigEndian, &indexOffset) + binary.Read(f, binary.BigEndian, &magic) + + if magic != indexFooterMagic || blockCount == 0 || indexOffset <= 0 { + return nil + } + + if _, err := f.Seek(indexOffset, io.SeekStart); err != nil { + return nil + } + entries := make([]BlockIndexEntry, blockCount) + for i := uint32(0); i < blockCount; i++ { + var keyLen uint32 + if err := binary.Read(f, binary.BigEndian, &keyLen); err != nil { + return nil + } + key := make([]byte, keyLen) + if _, err := io.ReadFull(f, key); err != nil { + return nil + } + var offset int64 + if err := binary.Read(f, binary.BigEndian, &offset); err != nil { + return nil + } + entries[i] = BlockIndexEntry{LastKey: key, BlockOffset: offset} + } + // indexOffset 既是块索引起点，也是数据区终点——即最后一块的结束偏移。 + return &blockIndex{entries: entries, dataEnd: indexOffset} +} + +// searchBlock 二分块索引定位目标块，单次 ReadAt 读入整块后在内存中扫描。 +// 每次点查恒定一次 read syscall，与块内条目数无关。 +func (ss *SSTable) searchBlock(filepath string, key []byte, bi *blockIndex) ([]byte, bool) { + idx := bi.entries + lo, hi := 0, len(idx)-1 + for lo < hi { + mid := (lo + hi) / 2 + if bytes.Compare(key, idx[mid].LastKey) <= 0 { + hi = mid + } else { + lo = mid + 1 + } + } + if lo >= len(idx) { + return nil, false + } + + start, end := bi.blockExtent(lo) + if end <= start { + return nil, false + } + + // 命中缓存的块由多个读者共享，故下方扫描只读、且命中时对 value 另行拷贝。 + buf, cached := ss.blocks.get(filepath, start) + if !cached { + f := ss.openFile(filepath) + if f == nil { + return nil, false + } + buf = make([]byte, end-start) + // ReadAt 不使用文件内偏移，故共享句柄可被并发读者安全复用。 + if _, err := f.ReadAt(buf, start); err != nil { + return nil, false + } + ss.blocks.put(filepath, start, buf) + } + + for off := 0; off+4 <= len(buf); { + kLen := int(binary.BigEndian.Uint32(buf[off:])) + off += 4 + if kLen < 0 || off+kLen+4 > len(buf) { + break + } + k := buf[off : off+kLen] + off += kLen + vLen := binary.BigEndian.Uint32(buf[off:]) + off += 4 + + tomb := vLen == tombstoneValLen + var v []byte + if !tomb { // 墓碑无 value 字节 + if off+int(vLen) > len(buf) { + break + } + v = buf[off : off+int(vLen)] + off += int(vLen) + } + + cmp := bytes.Compare(k, key) + if cmp == 0 { + if tomb { // 命中墓碑：found 但已删除 + return nil, true + } + // 必须拷贝：v 是整块 buf 的子切片，直接返回会让整块常驻内存。 + // 用 make+copy 而非 append([]byte(nil), ...)：后者在 v 为空时返回 nil， + // 会把「空 value」误变成「墓碑」（约定 nil=墓碑）。 + out := make([]byte, len(v)) + copy(out, v) + return out, true + } + if cmp > 0 { + break + } + } + return nil, false +} + +// readFromSSTableFull 老格式文件全量读取（兼容） +func (ss *SSTable) readFromSSTableFull(filepath string, key []byte) ([]byte, bool) { + entries, _ := ss.ReadAllFromSSTable(filepath) + for _, entry := range entries { + if bytes.Equal(entry.Key, key) { + return entry.Value, true + } + } + return nil, false +} + +// MergeSSTable 把 files 归并为 targetLevel 上的一个新 SSTable，即 compaction 的落地实现。 +// +// 正确性不变量（慎改）：同 key 的 tie-break 由 files 的 srcIdx 决定（越大越新，见下）， +// compaction 输出文件用 time.Now() 打新 ts。这依赖调用方 CompactSSTable「一次 compaction 卷走整层全部文件」： +// 因此同 key 的多个版本要么都在本次输入里（srcIdx 定新旧），要么更新的版本在更浅层的 +// 更晚 ts 文件里——读路径据此按创建 ts 判 newest-wins（见 memtable.go getFromSSTables）。 +// 若引入 overlap-aware / 部分选择或跨层 compaction，必须同时确保「低 level = 更新」进入 tie-break， +// 并把 recency 显式化（per-key 序号），否则 compaction 会静默选中陈旧值。 diff --git a/storage/sstable_write.go b/storage/sstable_write.go new file mode 100644 index 0000000..3e24ebd --- /dev/null +++ b/storage/sstable_write.go @@ -0,0 +1,312 @@ +// 本文件是 SSTable 的写路径：flush 落盘、写尾（块索引 + 布隆 + Footer）与 compaction 合并。 +package storage + +import ( + "bufio" + "bytes" + "encoding/binary" + "fmt" + "io" + "log/slog" + "os" + "path/filepath" + "time" +) + +func (ss *SSTable) WriteToSSTable(entries []LogEntry) error { + if len(entries) == 0 { + return ErrNoEntries + } + + // L0：memtable flush 落到 level 0；level 编进文件名以便重启恢复。 + filename := fmt.Sprintf("sstable_L0_%d.sst", time.Now().UnixNano()) + dir := ss.dir + if err := os.MkdirAll(dir, dirPerm); err != nil { + return fmt.Errorf("create data directory failed: %v", err) + } + fullPath := filepath.Join(dir, filename) + + file, err := os.Create(fullPath) + if err != nil { + return fmt.Errorf("create SSTable file failed: %v", err) + } + defer file.Close() + + // 构建数据 buffer + 块索引 + var blockIdx []blockMeta + + // 数据段先攒在内存 buffer 里再一次写出。此处的 binary.Write 不检查错误是安全的： + // 目标是 bytes.Buffer，其 Write 按文档永不返回错误（容量不足时直接 panic）。 + var buf bytes.Buffer + for i, entry := range entries { + bi := i / SSTableBlockSize + if i%SSTableBlockSize == 0 { + blockIdx = append(blockIdx, blockMeta{blockOffset: int64(buf.Len())}) + } + blockIdx[bi].lastKey = entry.Key + + binary.Write(&buf, binary.BigEndian, uint32(len(entry.Key))) + buf.Write(entry.Key) + if entry.Value == nil { // 墓碑：仅写哨兵长度，无 value 字节 + binary.Write(&buf, binary.BigEndian, tombstoneValLen) + } else { + binary.Write(&buf, binary.BigEndian, uint32(len(entry.Value))) + buf.Write(entry.Value) + } + } + + // 写数据 + if _, err := file.Write(buf.Bytes()); err != nil { + return err + } + + // 数据区已写完，此处的偏移即块索引起点，也是读路径推算末块长度的依据。 + indexStart, err := file.Seek(0, io.SeekCurrent) + if err != nil { + return fmt.Errorf("locate index offset failed: %w", err) + } + + keys := make([][]byte, len(entries)) + for i := range entries { + keys[i] = entries[i].Key + } + pb, err := writeTail(file, blockIdx, keys, indexStart) + if err != nil { + return fmt.Errorf("write SSTable tail failed: %w", err) + } + + // 缓存块索引 + cache := make([]BlockIndexEntry, len(blockIdx)) + for i, b := range blockIdx { + cache[i] = BlockIndexEntry{LastKey: b.lastKey, BlockOffset: b.blockOffset} + } + ss.idxMu.Lock() + ss.indexCache[fullPath] = &blockIndex{entries: cache, dataEnd: indexStart} + ss.idxMu.Unlock() + + if err := file.Sync(); err != nil { + return fmt.Errorf("sync SSTable file failed: %v", err) + } + ss.cacheBloom(fullPath, pb) // 落盘后再缓存 + + info, err := file.Stat() + if err != nil { + return fmt.Errorf("stat SSTable file failed: %v", err) + } + meta := &SSTableMeta{ + Level: 0, + Filepath: fullPath, + MinKey: entries[0].Key, + MaxKey: entries[len(entries)-1].Key, + Size: info.Size(), + MaxKeyKnown: true, + } + ss.AddMeta(meta) + flushBytesWritten.Add(info.Size()) + return nil +} + +// Metas 返回元数据的不可变快照，按落盘先后升序（最旧在前）。 +// 无锁零拷贝；调用方只可读取，不得原地修改。切片元素是共享指针，非对象副本。 +func writeBloomSection(w io.Writer, keys [][]byte) (*PartitionedBloom, error) { + pb := BuildPartitionedBloom(keys, DefaultNamespaceSep, defaultBloomFPRate) + blob := pb.Encode() + if _, err := w.Write(blob); err != nil { + return nil, err + } + if err := binary.Write(w, binary.BigEndian, uint64(len(blob))); err != nil { + return nil, err + } + if err := binary.Write(w, binary.BigEndian, bloomTrailerMagic); err != nil { + return nil, err + } + return pb, nil +} + +// writeTail 写出 SSTable 的尾部三段：块索引、布隆过滤器、Footer。 +// +// 由 WriteToSSTable 与 MergeSSTable 共用，二者的字节布局因此不可能漂移——读路径只有 +// 一份解析实现，写路径也应只有一份。 +// +// 每一次写入都检查错误。此前这里的 binary.Write/Write 返回值被丢弃，后果并非「少写了 +// 几个字节」而是静默的数据丢失：尾部残缺的文件会让 footer magic 校验失败，重启时 +// EnsureMeta 在新格式文件上算出错误的 MaxKey，[MinKey,MaxKey] 过滤随即跳过整个文件； +// 而调用方以为落盘成功——Flush 会丢弃内存副本，compaction 会删除源文件。 +func writeTail(w io.Writer, blocks []blockMeta, keys [][]byte, indexStart int64) (*PartitionedBloom, error) { + // 块索引: [LastKeyLen(4B)][LastKey][BlockOffset(8B)] × N + for _, b := range blocks { + if err := binary.Write(w, binary.BigEndian, uint32(len(b.lastKey))); err != nil { + return nil, fmt.Errorf("write block index key length: %w", err) + } + if _, err := w.Write(b.lastKey); err != nil { + return nil, fmt.Errorf("write block index key: %w", err) + } + if err := binary.Write(w, binary.BigEndian, b.blockOffset); err != nil { + return nil, fmt.Errorf("write block index offset: %w", err) + } + } + + pb, err := writeBloomSection(w, keys) + if err != nil { + return nil, fmt.Errorf("write bloom section: %w", err) + } + + // Footer: BlockCount(4B) + IndexOffset(8B) + Magic(4B) + if err := binary.Write(w, binary.BigEndian, uint32(len(blocks))); err != nil { + return nil, fmt.Errorf("write footer block count: %w", err) + } + if err := binary.Write(w, binary.BigEndian, indexStart); err != nil { + return nil, fmt.Errorf("write footer index offset: %w", err) + } + if err := binary.Write(w, binary.BigEndian, indexFooterMagic); err != nil { + return nil, fmt.Errorf("write footer magic: %w", err) + } + return pb, nil +} + +// cacheBloom 将过滤器写入缓存（应在 file.Sync() 成功后调用）。 +func (ss *SSTable) MergeSSTable(files []*SSTableMeta, targetLevel int) *SSTableMeta { + if len(files) == 0 { + return nil + } + + slog.Info("merging SSTable files", "files", len(files), "targetLevel", targetLevel) + + // 为每个源文件打开流式迭代器（srcIdx = 在 files 中的序号，越大越新） + iters := make([]*sstableIterator, 0, len(files)) + for _, meta := range files { + it, err := newSSTableIterator(meta.Filepath) + if err != nil { + slog.Error("failed to open SSTable iterator for merge", "file", meta.Filepath, "error", err) + for _, opened := range iters { + opened.Close() + } + return nil + } + iters = append(iters, it) + } + + mi, err := newMergeIterator(iters) + if err != nil { + mi.Close() + slog.Error("failed to init merge iterator", "error", err) + return nil + } + defer mi.Close() + + // targetLevel 编进文件名，使重启（LoadSSTableMetaList）能恢复该文件的 level。 + filename := fmt.Sprintf("sstable_merged_L%d_%d.sst", targetLevel, time.Now().UnixNano()) + dir := ss.dir + fullPath := filepath.Join(dir, filename) + + file, err := os.Create(fullPath) + if err != nil { + slog.Error("failed to create merged SSTable", "error", err) + return nil + } + defer file.Close() + + // K 路归并流式写出：value 直接落盘，仅累积块索引(每块一条)与 key(供布隆)， + // 不再把全部源条目读入内存。 + var blockIdx []blockMeta + var keys [][]byte + var minKey, maxKey []byte + var dataOffset int64 + // 数据段经 bufio 写出。循环内的 Write 不逐个检查错误是安全的：bufio.Writer 记住 + // 首个错误并在其后所有 Write 与 Flush 上返回它，而下方 bw.Flush() 的错误是被检查的。 + bw := bufio.NewWriter(file) + + count := 0 + for mi.Next() { + k := mi.Key() + v := mi.Value() + bi := count / SSTableBlockSize + if count%SSTableBlockSize == 0 { + blockIdx = append(blockIdx, blockMeta{blockOffset: dataOffset}) + } + blockIdx[bi].lastKey = k + + var hdr [4]byte + binary.BigEndian.PutUint32(hdr[:], uint32(len(k))) + bw.Write(hdr[:]) + bw.Write(k) + if v == nil { // 墓碑：写哨兵长度，无 value 字节 + binary.BigEndian.PutUint32(hdr[:], tombstoneValLen) + bw.Write(hdr[:]) + dataOffset += int64(8 + len(k)) + } else { + binary.BigEndian.PutUint32(hdr[:], uint32(len(v))) + bw.Write(hdr[:]) + if _, werr := bw.Write(v); werr != nil { + slog.Error("failed to write merged entry", "error", werr) + return nil + } + dataOffset += int64(8 + len(k) + len(v)) + } + + keys = append(keys, k) // k 为迭代器新分配，安全持有 + if count == 0 { + minKey = k + } + maxKey = k + count++ + } + if err := mi.Err(); err != nil { + slog.Error("merge iteration failed", "error", err) + return nil + } + if count == 0 { + slog.Warn("no entries to merge") + return nil + } + if err := bw.Flush(); err != nil { + slog.Error("failed to flush merged data", "error", err) + return nil + } + + indexStart, err := file.Seek(0, io.SeekCurrent) + if err != nil { + slog.Error("failed to locate index offset for merged SSTable", "error", err) + return nil + } + pb, err := writeTail(file, blockIdx, keys, indexStart) + if err != nil { + slog.Error("failed to write merged SSTable tail", "error", err) + return nil + } + + cache := make([]BlockIndexEntry, len(blockIdx)) + for i, b := range blockIdx { + cache[i] = BlockIndexEntry{LastKey: b.lastKey, BlockOffset: b.blockOffset} + } + ss.idxMu.Lock() + ss.indexCache[fullPath] = &blockIndex{entries: cache, dataEnd: indexStart} + ss.idxMu.Unlock() + + if err := file.Sync(); err != nil { + slog.Error("failed to sync merged SSTable", "error", err) + return nil + } + ss.cacheBloom(fullPath, pb) // 落盘后再缓存 + + info, err := file.Stat() + if err != nil { + slog.Error("failed to stat merged SSTable", "error", err) + return nil + } + + newMeta := &SSTableMeta{ + Level: targetLevel, + Filepath: fullPath, + MinKey: minKey, + MaxKey: maxKey, + Size: info.Size(), + MaxKeyKnown: true, + } + + ss.AddMeta(newMeta) + compactionBytesWritten.Add(info.Size()) + slog.Info("SSTable merged", "level", targetLevel, "file", filename, "keys", count, "size", info.Size()) + + return newMeta +}