cursor.go

package spool

import (
	"encoding/binary"
	"hash/fnv"
	"os"
	"path/filepath"
)

const (
	cursorName = "cursor"
	cursorLen  = 24 // seq(8) + offset(8) + fnv64a checksum(8)
)

// cursor is the persisted read position: which segment, and how far into it.
type cursor struct{ seq, off int64 }

func cursorSum(b []byte) uint64 {
	h := fnv.New64a()
	_, _ = h.Write(b)
	return h.Sum64()
}

// encode lays the cursor out for its in-place rewrite. The record carries a
// checksum precisely because it is rewritten in place: a partial write on power
// loss must fail its check and fall back to redelivering, rather than seek to a
// mixed old/new position that silently skips undelivered frames.
func (c cursor) encode() [cursorLen]byte {
	var buf [cursorLen]byte
	binary.BigEndian.PutUint64(buf[:8], uint64(c.seq))
	binary.BigEndian.PutUint64(buf[8:16], uint64(c.off))
	binary.BigEndian.PutUint64(buf[16:], cursorSum(buf[:16]))
	return buf
}

// decodeCursor reports ok=false for a short or checksum-failing record — a torn
// in-place rewrite, or a fresh spool. The caller redelivers from the oldest
// segment.
func decodeCursor(b []byte) (cursor, bool) {
	if len(b) < cursorLen || binary.BigEndian.Uint64(b[16:cursorLen]) != cursorSum(b[:16]) {
		return cursor{}, false
	}
	return cursor{
		seq: int64(binary.BigEndian.Uint64(b[:8])),
		off: int64(binary.BigEndian.Uint64(b[8:16])),
	}, true
}

// cursorFile is the open handle to dir/cursor.
type cursorFile struct {
	f *os.File
	// sync makes every store durable. The cursor is an optimisation, not a
	// durability record — see Options.CommitSync for why this is a choice and
	// what it costs.
	sync bool
}

// openCursor opens or creates dir/cursor and returns the position it holds.
// found is false for a fresh, short, or torn record.
func openCursor(dir string, sync bool) (*cursorFile, cursor, bool, error) {
	f, err := os.OpenFile(filepath.Join(dir, cursorName), os.O_RDWR|os.O_CREATE, 0o644)
	if err != nil {
		return nil, cursor{}, false, err
	}
	var buf [cursorLen]byte
	n, _ := f.ReadAt(buf[:], 0)
	cur, ok := decodeCursor(buf[:n])
	return &cursorFile{f: f, sync: sync}, cur, ok, nil
}

// store rewrites the cursor in place. A torn rewrite fails its checksum on the
// next load and redelivers. It tolerates a nil receiver: three of its four call
// sites are error-recovery paths where the file may legitimately be closed.
func (c *cursorFile) store(cur cursor) error {
	if c == nil || c.f == nil {
		return nil
	}
	buf := cur.encode()
	if _, err := c.f.WriteAt(buf[:], 0); err != nil {
		return err
	}
	if !c.sync {
		return nil
	}
	return c.f.Sync()
}

func (c *cursorFile) Close() error {
	if c == nil || c.f == nil {
		return nil
	}
	err := c.f.Close()
	c.f = nil
	return err
}