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 encodedBatch to the active segment, rotating first if // the batch doesn't fit. batchStartSequence is the sequence number of the // FIRST entry in this batch — used as the new segment's startSequence when // rotation occurs, so multi-segment recovery's continuity check passes per // design §3.2 line 639-663. func (sm *SegmentManager) AppendBatch(encodedBatch []byte, batchStartSequence uint64) error { worstCaseSize := uint64(len(encodedBatch)) + uint64(PhysicalRecordHeaderSize) + uint64(PhysicalRecordHeaderSize) if sm.active.RemainingPayload() < worstCaseSize { // C8 fix: new segment's first batch is THIS batch, so its // startSequence must equal batchStartSequence (not byte offset). if err := sm.rotate(batchStartSequence); 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() }