fathom/internal/store/shard.go

package store

import (
	"encoding/binary"
	"fmt"
	"hash/crc32"
	"io"
	"os"
	"path/filepath"
	"sync"
)

// A shard file is a sequence of framed chunks:
//
//	[ length uint32 ][ crc32 uint32 ][ chunk body ]
//
// There is no footer and no header block to keep current. Appending is a write
// at the end of the file, and nothing already written is ever revised. That
// property is what makes crash recovery a truncation rather than a repair: if
// the process dies mid-append, the tail is a chunk whose length runs past EOF
// or whose CRC does not match, and dropping it leaves every earlier chunk
// intact and correct.
const (
	frameHeaderSize = 8

	// maxChunkBytes bounds the allocation a corrupt length prefix can request.
	// Real chunks are well under a megabyte; this is three orders of magnitude
	// of headroom before it rejects anything legitimate.
	maxChunkBytes = 1 << 30
)

// Shard is one day's file for one table, plus its blob sidecar.
type Shard struct {
	Table string
	Date  string // YYYY-MM-DD

	path     string
	mu       sync.Mutex
	f        *os.File
	blob     *BlobFile
	readOnly bool
}

// OpenShard opens or creates a shard, recovering a torn tail if one is present.
func OpenShard(dir, table, date string) (*Shard, error) {
	return openShard(dir, table, date, false)
}

// OpenShardReadOnly opens a shard without creating, recovering, or writing.
//
// This exists because recovery truncates. A second process opening a shard the
// server is actively appending to would run recoverTail against a chunk still
// being written and cut it off — the reader destroying the writer's data. Any
// process that only reads, which is every CLI command, must take this path.
//
// Reading a file under concurrent append is otherwise safe: chunks are framed
// with a length and a CRC, so a partially written tail fails one of those
// checks and iteration stops cleanly at the last complete chunk.
func OpenShardReadOnly(dir, table, date string) (*Shard, error) {
	return openShard(dir, table, date, true)
}

func openShard(dir, table, date string, readOnly bool) (*Shard, error) {
	tdir := filepath.Join(dir, table)
	path := filepath.Join(tdir, date+".fathom")

	if readOnly {
		f, err := os.Open(path)
		if err != nil {
			return nil, err
		}
		blob, err := OpenBlobFileReadOnly(filepath.Join(tdir, date+".blob"))
		if err != nil {
			f.Close()
			return nil, err
		}
		return &Shard{Table: table, Date: date, path: path, f: f, blob: blob, readOnly: true}, nil
	}

	if err := os.MkdirAll(tdir, 0o755); err != nil {
		return nil, err
	}
	if err := recoverTail(path); err != nil {
		return nil, fmt.Errorf("recovering %s: %w", path, err)
	}

	f, err := os.OpenFile(path, os.O_RDWR|os.O_CREATE|os.O_APPEND, 0o644)
	if err != nil {
		return nil, err
	}
	blob, err := OpenBlobFile(filepath.Join(tdir, date+".blob"))
	if err != nil {
		f.Close()
		return nil, err
	}
	return &Shard{Table: table, Date: date, path: path, f: f, blob: blob}, nil
}

// recoverTail truncates the file to the end of the last chunk that is both
// fully present and passes its checksum.
//
// This runs on every open, not only after a detected crash, because there is no
// reliable way to know a crash happened. It is a full scan of the frame headers
// — not the bodies — so the cost is a seek per chunk, which for a day's file is
// negligible next to the safety of never handing a half-written chunk to the
// decoder.
func recoverTail(path string) error {
	f, err := os.OpenFile(path, os.O_RDWR, 0o644)
	if os.IsNotExist(err) {
		return nil
	}
	if err != nil {
		return err
	}
	defer f.Close()

	st, err := f.Stat()
	if err != nil {
		return err
	}
	size := st.Size()

	var good int64
	var hdr [frameHeaderSize]byte
	for {
		if good+frameHeaderSize > size {
			break
		}
		if _, err := f.ReadAt(hdr[:], good); err != nil {
			break
		}
		n := binary.LittleEndian.Uint32(hdr[0:4])
		want := binary.LittleEndian.Uint32(hdr[4:8])
		if n == 0 || n > maxChunkBytes {
			break
		}
		end := good + frameHeaderSize + int64(n)
		if end > size {
			break // chunk body never finished landing
		}
		body := make([]byte, n)
		if _, err := f.ReadAt(body, good+frameHeaderSize); err != nil {
			break
		}
		if crc32.ChecksumIEEE(body) != want {
			break // body landed partially, or the disk lied
		}
		good = end
	}

	if good < size {
		return f.Truncate(good)
	}
	return nil
}

// AppendChunk frames and appends a chunk, then fsyncs.
//
// The fsync is per chunk, not per span, which is the entire reason chunks
// exist: ingest buffers spans and pays one sync per flush.
func (s *Shard) AppendChunk(c *Chunk) error {
	s.mu.Lock()
	defer s.mu.Unlock()

	if s.readOnly {
		return fmt.Errorf("shard %s/%s is open read-only", s.Table, s.Date)
	}

	var body bytesBuffer
	if err := c.Encode(&body); err != nil {
		return err
	}
	if len(body.b) > maxChunkBytes {
		return fmt.Errorf("chunk of %d bytes exceeds the %d cap", len(body.b), maxChunkBytes)
	}

	var hdr [frameHeaderSize]byte
	binary.LittleEndian.PutUint32(hdr[0:4], uint32(len(body.b)))
	binary.LittleEndian.PutUint32(hdr[4:8], crc32.ChecksumIEEE(body.b))

	// One write, so a frame header never lands without at least an attempt at
	// its body. Recovery handles the partial case regardless.
	frame := append(hdr[:], body.b...)
	if _, err := s.f.Write(frame); err != nil {
		return err
	}
	if err := s.blob.Sync(); err != nil {
		return err
	}
	return s.f.Sync()
}

// Chunks iterates the shard's chunks in write order, stopping at the first
// unreadable one. Recovery at open means that should not happen, but a shard
// being appended to concurrently can present a partial tail to a reader.
func (s *Shard) Chunks(fn func(*Chunk) error) error {
	f, err := os.Open(s.path)
	if err != nil {
		if os.IsNotExist(err) {
			return nil
		}
		return err
	}
	defer f.Close()
	return iterChunks(f, fn)
}

func iterChunks(f *os.File, fn func(*Chunk) error) error {
	var off int64
	var hdr [frameHeaderSize]byte
	for {
		if _, err := f.ReadAt(hdr[:], off); err != nil {
			if err == io.EOF {
				return nil
			}
			return nil // short tail: stop cleanly
		}
		n := binary.LittleEndian.Uint32(hdr[0:4])
		want := binary.LittleEndian.Uint32(hdr[4:8])
		if n == 0 || n > maxChunkBytes {
			return nil
		}
		body := make([]byte, n)
		if _, err := f.ReadAt(body, off+frameHeaderSize); err != nil {
			return nil
		}
		if crc32.ChecksumIEEE(body) != want {
			return nil
		}
		c, err := DecodeChunk(body)
		if err != nil {
			return nil
		}
		if err := fn(c); err != nil {
			return err
		}
		off += frameHeaderSize + int64(n)
	}
}

// Blob exposes the shard's blob sidecar for lazy text fetches.
func (s *Shard) Blob() *BlobFile { return s.blob }

func (s *Shard) Close() error {
	s.mu.Lock()
	defer s.mu.Unlock()
	err1 := s.blob.Close()
	err2 := s.f.Close()
	if err1 != nil {
		return err1
	}
	return err2
}

// Size reports the shard's total on-disk footprint, chunks plus blobs, which is
// what size-based retention budgets against.
func (s *Shard) Size() int64 {
	var total int64
	if st, err := os.Stat(s.path); err == nil {
		total += st.Size()
	}
	total += s.blob.Size()
	return total
}

// bytesBuffer is a minimal append-only sink. bytes.Buffer would do, but Encode
// writes many small pieces and this avoids its interface indirection on the
// hottest path in ingest.
type bytesBuffer struct{ b []byte }

func (w *bytesBuffer) Write(p []byte) (int, error) {
	w.b = append(w.b, p...)
	return len(p), nil
}

func (w *bytesBuffer) WriteString(s string) (int, error) {
	w.b = append(w.b, s...)
	return len(s), nil
}