dailz
08960a9bcf
- wal/segment_manager.go: segment lifecycle with rotation at batch boundaries - wal/scanner.go: segment discovery, ordering, and continuity validation - wal/record_parser.go: block-level physical record parsing with tail corruption detection - Comprehensive tests for all modules, all pass with -race
115 lines
3.6 KiB
Go
115 lines
3.6 KiB
Go
package wal
|
|
|
|
import (
|
|
"fmt"
|
|
"os"
|
|
|
|
"github.com/dailz/go-kv/config"
|
|
"github.com/dailz/go-kv/manifest"
|
|
)
|
|
|
|
// SegmentManager manages the lifecycle of WAL segment files, handling segment
|
|
// rotation when the active segment runs out of payload capacity. A batch is
|
|
// never split across segments — if it doesn't fit, a new segment is created
|
|
// first and the entire batch is written there.
|
|
type SegmentManager struct {
|
|
dir string
|
|
cfg *config.WalConfig
|
|
active *SegmentWriter // currently active segment writer
|
|
nextSegID uint64 // next segment ID to allocate
|
|
}
|
|
|
|
// NewSegmentManager creates a new SegmentManager and its first segment file.
|
|
// It creates the directory if needed, writes the initial segment, and updates
|
|
// the CURRENT file to point to it.
|
|
func NewSegmentManager(
|
|
dir string,
|
|
startSegmentID uint64,
|
|
startSequence uint64,
|
|
cfg *config.WalConfig,
|
|
) (*SegmentManager, error) {
|
|
if err := os.MkdirAll(dir, 0o755); err != nil {
|
|
return nil, fmt.Errorf("wal: create segment directory %s: %w", dir, err)
|
|
}
|
|
|
|
sw, err := NewSegmentWriter(dir, startSegmentID, startSequence, cfg)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("wal: create initial segment: %w", err)
|
|
}
|
|
|
|
sm := &SegmentManager{
|
|
dir: dir,
|
|
cfg: cfg,
|
|
active: sw,
|
|
nextSegID: startSegmentID + 1,
|
|
}
|
|
|
|
// Best-effort CURRENT file update.
|
|
_ = manifest.WriteCurrent(dir, sw.SegmentID())
|
|
|
|
return sm, nil
|
|
}
|
|
|
|
// AppendBatch writes an encoded batch to the active segment. If the batch does
|
|
// not fit in the remaining payload space (with worst-case physical record
|
|
// overhead), the manager rotates to a fresh segment first so the entire batch
|
|
// lands in one segment.
|
|
func (sm *SegmentManager) AppendBatch(encodedBatch []byte) error {
|
|
// Calculate the worst-case on-disk size for this batch:
|
|
// len(encodedBatch) + at least one physical record header + block padding margin
|
|
// This is a conservative upper bound. The actual overhead may be less due to
|
|
// block alignment, but we must guarantee the batch won't exceed MaxSegmentSize.
|
|
worstCaseSize := uint64(len(encodedBatch)) + uint64(PhysicalRecordHeaderSize) + uint64(PhysicalRecordHeaderSize)
|
|
|
|
if sm.active.RemainingPayload() < worstCaseSize {
|
|
if err := sm.rotate(sm.active.CurrentOffset()); err != nil {
|
|
return fmt.Errorf("wal: rotate segment: %w", err)
|
|
}
|
|
}
|
|
|
|
return sm.active.AppendBatch(encodedBatch)
|
|
}
|
|
|
|
// rotate closes the current segment and creates a new one. The CURRENT file is
|
|
// updated on a best-effort basis — a failure is logged but does not prevent
|
|
// the rotation from succeeding.
|
|
func (sm *SegmentManager) rotate(newStartSequence uint64) error {
|
|
if err := sm.active.Close(); err != nil {
|
|
return fmt.Errorf("wal: close segment %d: %w", sm.active.SegmentID(), err)
|
|
}
|
|
|
|
sw, err := NewSegmentWriter(sm.dir, sm.nextSegID, newStartSequence, sm.cfg)
|
|
if err != nil {
|
|
return fmt.Errorf("wal: create segment %d: %w", sm.nextSegID, err)
|
|
}
|
|
|
|
sm.nextSegID++
|
|
sm.active = sw
|
|
|
|
// Best-effort CURRENT file update — failure must not fail the write.
|
|
_ = manifest.WriteCurrent(sm.dir, sw.SegmentID())
|
|
|
|
return nil
|
|
}
|
|
|
|
// ActiveSegmentID returns the segment ID of the currently active segment.
|
|
func (sm *SegmentManager) ActiveSegmentID() uint64 {
|
|
return sm.active.SegmentID()
|
|
}
|
|
|
|
// RemainingPayload returns the number of bytes that can still be written to
|
|
// the active segment.
|
|
func (sm *SegmentManager) RemainingPayload() uint64 {
|
|
return sm.active.RemainingPayload()
|
|
}
|
|
|
|
// Sync flushes the active segment to durable storage.
|
|
func (sm *SegmentManager) Sync() error {
|
|
return sm.active.Sync()
|
|
}
|
|
|
|
// Close flushes and closes the active segment.
|
|
func (sm *SegmentManager) Close() error {
|
|
return sm.active.Close()
|
|
}
|