segment.go

package spool

import (
	"errors"
	"io"
	"os"
)

type segment struct {
	seq  int64
	size int64
	path string
	// version is the segment's frame format, read from its header. It is per
	// segment, so a format bump does not invalidate the segments already on
	// disk — each is read with the framing it was written in.
	version uint16
}

// openTail opens the newest segment for appending. The tail must already have
// been repaired: openTail assumes every byte in the file is part of a whole
// frame.
//
// A tail in an older format is frozen — never appended to — and a fresh segment
// takes over, so one file never mixes two framings. That branch is unreachable
// while knownVersions == {formatVersion}; it is kept because a future bump that
// forgets it appends new frames into an old file and corrupts silently.
func (s *Spool) openTail() error {
	tail := &s.segs[len(s.segs)-1]
	if err := s.repairTail(tail); err != nil {
		return err
	}
	if tail.version != formatVersion {
		return s.appendSegment(tail.seq + 1)
	}
	f, err := os.OpenFile(tail.path, os.O_WRONLY|os.O_APPEND, 0o644)
	if err != nil {
		return err
	}
	s.w = f
	return nil
}

// repairTail truncates the tail to its last structurally complete frame,
// discarding a torn tail (a frame whose write a crash left incomplete) or
// orphan bytes beyond the last whole frame (a partial append whose rollback did
// not complete). On return the file's physical size == tail.size ==
// s.syncedTailSize, and every byte in the file is part of a whole frame.
//
// Whatever the truncate destroyed is added to s.discarded. A torn tail is ONE
// incomplete frame (a crash mid-append) and costs nothing; anything larger
// means damage cost us fsynced records — the one loss path in the spool with no
// Pop to count it, so the caller reports it.
func (s *Spool) repairTail(tail *segment) error {
	good, err := lastCompleteOffset(*tail)
	if err != nil {
		return err
	}
	if info, err := os.Stat(tail.path); err != nil {
		return err
	} else if info.Size() != good {
		s.discarded += info.Size() - good
		if err := os.Truncate(tail.path, good); err != nil {
			return err
		}
	}
	tail.size = good
	s.syncedTailSize = good // on-disk content is the durable baseline
	return nil
}

// appendSegment creates a new segment with the given seq, writes its header,
// and makes it the write tail. On failure the partially-created file is
// unlinked: a zero-length or headerless segment left behind is classified as a
// damaged header by the next Open, which would report a data-loss event for a
// segment that never held a record.
func (s *Spool) appendSegment(seq int64) error {
	path := s.segPath(seq)
	f, err := os.OpenFile(path, os.O_CREATE|os.O_WRONLY|os.O_APPEND, 0o644)
	if err != nil {
		return err
	}
	var hdr [segHeaderLen]byte
	putSegHeader(&hdr)
	if _, err := f.Write(hdr[:]); err != nil {
		_ = f.Close()
		_ = os.Remove(path)
		return err
	}
	// The header is fsynced before the segment is ever the write tail, so a
	// crash can never leave a headerless segment that the next load() would
	// then have to classify as damage.
	if err := f.Sync(); err != nil {
		_ = f.Close()
		_ = os.Remove(path)
		return err
	}
	if s.w != nil {
		_ = s.w.Close()
	}
	s.w = f
	s.segs = append(s.segs, segment{seq: seq, size: segHeaderLen, path: path, version: formatVersion})
	s.syncedTailSize = segHeaderLen
	s.syncDir()
	return nil
}

// lastCompleteOffset walks a segment's frames by their lengths and returns the
// offset just past the last structurally complete one — where a torn tail is
// truncated and the next append lands.
//
// A torn write only ever damages the END of the file, so a frame that is
// structurally whole but fails its checksum is far more likely to be a durable
// frame with a damaged byte than a torn one; truncating there would throw away
// every good frame that follows it. Such a frame stays put and Pop drops it
// individually (reporting ErrCorrupt), so the blast radius of one bad byte is
// one record. A torn write whose length field itself was mangled is still
// caught: its bogus length either overruns the file (truncated here) or
// desynchronizes the walk into garbage that fails the same structural check a
// frame or two later.
func lastCompleteOffset(sg segment) (int64, error) {
	f, err := os.Open(sg.path)
	if err != nil {
		return 0, err
	}
	defer func() { _ = f.Close() }()
	if sg.size < segHeaderLen {
		return sg.size, nil // defensive: every segment here passed readSegHeader
	}
	off := int64(segHeaderLen)
	var hdr [frameHeaderLen]byte
	// One growable buffer for the whole walk: a fresh payload slice per frame
	// would allocate once per record over the entire tail at every Open.
	var buf []byte
	for off+frameHeaderLen <= sg.size {
		if _, err := f.ReadAt(hdr[:], off); err != nil {
			if errors.Is(err, io.EOF) || errors.Is(err, io.ErrUnexpectedEOF) {
				break // file shorter than recorded: torn tail
			}
			// A real I/O error is not a truncation point: truncating here
			// would silently discard valid fsynced frames after it.
			return 0, err
		}
		n := frameLen(hdr[:])
		end := off + frameHeaderLen + n
		if end > sg.size {
			break // torn frame (or a length damaged upward — same recovery)
		}
		// VERIFY the frame, do not just trust its length. The checksum covers
		// the length bytes precisely so a damaged length is detectable: without
		// this check a length corrupted DOWNWARD still lands in bounds, and the
		// walk then reads the middle of a payload as the next header and
		// mis-frames the whole remainder — truncating away valid fsynced frames
		// with no error and no counter. A frame that fails here is left in
		// place and skipped: Pop drops exactly it and reports ErrCorrupt.
		if int64(cap(buf)) < n {
			buf = make([]byte, n)
		}
		p := buf[:n]
		if _, err := f.ReadAt(p, off+frameHeaderLen); err != nil {
			if errors.Is(err, io.EOF) || errors.Is(err, io.ErrUnexpectedEOF) {
				break
			}
			return 0, err
		}
		off = end // whole under its own length: keep it, damaged or not
	}
	return off, nil
}

// syncDir fsyncs the spool directory so a newly created segment survives a
// crash (best-effort).
//
// Segment REMOVALS are deliberately not synced: an unlink that a crash undoes
// leaves a consumed segment on disk, which the next load() drops on sight
// (seq < cursorSeq), or — if the cursor write was lost too — redelivers, which
// is within at-least-once. Paying an fsync per retirement to tighten that is
// not worth it.
func (s *Spool) syncDir() {
	if d, err := os.Open(s.dir); err == nil {
		_ = d.Sync()
		_ = d.Close()
	}
}