package wal import ( "errors" "fmt" "os" "path/filepath" "github.com/dailz/go-kv/config" ) // SegmentWriter handles appending WAL batches to a single segment file. // It manages block-aligned writes via BlockWriter and tracks file offset // for segment rotation decisions. type SegmentWriter struct { fd *os.File dir string cfg *config.WalConfig segmentID uint64 startSequence uint64 blockWriter *BlockWriter currentOffset uint64 // total bytes written (starts at WalFileHeaderSize) } // NewSegmentWriter creates a new WAL segment file and writes the file header. // The segment file is created with a .tmp extension, the header is written and // synced, then the file is atomically renamed to its final name and synced again. func NewSegmentWriter( dir string, segmentID uint64, startSequence uint64, cfg *config.WalConfig, ) (*SegmentWriter, error) { if err := cfg.Validate(); err != nil { return nil, fmt.Errorf("wal: invalid config: %w", err) } baseName := fmt.Sprintf("segment-%d.wal", segmentID) tmpPath := filepath.Join(dir, baseName+".tmp") finalPath := filepath.Join(dir, baseName) // Create the temp file. fd, err := os.OpenFile(tmpPath, os.O_WRONLY|os.O_CREATE|os.O_EXCL, 0o644) if err != nil { return nil, fmt.Errorf("wal: create segment temp file %s: %w", tmpPath, err) } // Build and write the file header. hdr := &WalFileHeader{ BlockSize: cfg.BlockSize, SegmentID: segmentID, StartSequence: startSequence, } encoded := EncodeWalHeader(hdr) if _, err := fd.Write(encoded[:]); err != nil { fd.Close() os.Remove(tmpPath) return nil, fmt.Errorf("wal: write segment header: %w", err) } // Sync the header to disk. if err := fd.Sync(); err != nil { fd.Close() os.Remove(tmpPath) return nil, fmt.Errorf("wal: sync segment header: %w", err) } // Atomically rename temp file to final name. if err := fd.Close(); err != nil { os.Remove(tmpPath) return nil, fmt.Errorf("wal: close temp file: %w", err) } if err := os.Rename(tmpPath, finalPath); err != nil { os.Remove(tmpPath) return nil, fmt.Errorf("wal: rename segment file: %w", err) } // Open the final file for appending. fd, err = os.OpenFile(finalPath, os.O_WRONLY|os.O_APPEND, 0o644) if err != nil { return nil, fmt.Errorf("wal: open segment file for append: %w", err) } // Per design ยง3.2 line 258, directory fsync is a hard requirement for // durable-ready. Without it, the rename above is not guaranteed to survive // power loss, violating the Always-mode "no loss of acknowledged writes" // promise. if err := dirFsyncFn(dir); err != nil { closeErr := fd.Close() removeErr := os.Remove(finalPath) if closeErr != nil || removeErr != nil { cleanup := errors.Join(closeErr, removeErr) return nil, fmt.Errorf("wal: fsync directory after segment rename (cleanup: %v): %w", cleanup, err) } return nil, fmt.Errorf("wal: fsync directory after segment rename: %w", err) } return &SegmentWriter{ fd: fd, dir: dir, cfg: cfg, segmentID: segmentID, startSequence: startSequence, blockWriter: NewBlockWriter(), currentOffset: WalFileHeaderSize, }, nil } // AppendBatch encodes the batch into physical records and appends them to the // segment file. The encoded batch is split into block-aligned physical records // using SplitIntoRecords. func (sw *SegmentWriter) AppendBatch(encodedBatch []byte) error { records := SplitIntoRecords(encodedBatch) if len(records) == 0 { return nil } for _, rec := range records { if len(rec) < PhysicalRecordHeaderSize { return fmt.Errorf("wal: corrupted physical record: size %d < header size %d", len(rec), PhysicalRecordHeaderSize) } recType := rec[6] // type byte is at offset 6 in the encoded record payload := rec[PhysicalRecordHeaderSize:] if err := sw.blockWriter.WriteRecord(recType, payload, sw.fd); err != nil { return fmt.Errorf("wal: writing physical record: %w", err) } sw.currentOffset += uint64(len(rec)) } return nil } // Sync flushes the segment file to durable storage. func (sw *SegmentWriter) Sync() error { if err := sw.blockWriter.Flush(sw.fd); err != nil { return fmt.Errorf("wal: flushing partial block before sync: %w", err) } return sw.fd.Sync() } // Close flushes any partial block and closes the segment file. func (sw *SegmentWriter) Close() error { if err := sw.blockWriter.Flush(sw.fd); err != nil { return fmt.Errorf("wal: flushing block writer on close: %w", err) } return sw.fd.Close() } // RemainingPayload returns the number of bytes that can still be written // to this segment before it reaches its maximum size. func (sw *SegmentWriter) RemainingPayload() uint64 { if sw.currentOffset >= sw.cfg.MaxSegmentSize { return 0 } return sw.cfg.MaxSegmentSize - sw.currentOffset } // CurrentOffset returns the total number of bytes written to the segment file, // including the file header. func (sw *SegmentWriter) CurrentOffset() uint64 { return sw.currentOffset } // SegmentID returns the segment identifier. func (sw *SegmentWriter) SegmentID() uint64 { return sw.segmentID } // SegmentPath returns the full filesystem path to the segment file. func (sw *SegmentWriter) SegmentPath() string { return filepath.Join(sw.dir, fmt.Sprintf("segment-%d.wal", sw.segmentID)) }