tailer.go #2

// Package tailer watches a log file and delivers new bytes to a consumer.
//
// It uses fsnotify for event-driven notifications with a polling fallback
// ticker, and detects log rotation by comparing the file's inode against
// the inode stored in the state file. On rotation the offset resets to 0
// (subject to MaxRotationRecoverySize) and the state is updated.
package tailer

import (
	"encoding/json"
	"fmt"
	"io"
	"os"
	"path/filepath"
	"sync"
	"syscall"
	"time"

	"github.com/fsnotify/fsnotify"
	"workshop.perforce.com/p4lf/internal/config"
)

// State is the persistent checkpoint written to StateFile.
type State struct {
	Inode  uint64 `json:"inode"`
	Offset int64  `json:"offset"`
}

// Chunk is a slice of bytes read from the log file since the last flush.
type Chunk struct {
	Data        []byte
	StartOffset int64
	EndOffset   int64
}

// Tailer watches a P4LOG file and delivers accumulated chunks on demand.
type Tailer struct {
	cfg      config.Config
	log      Logger
	mu       sync.Mutex
	file     *os.File
	state    State
	buf      []byte
	startOff int64 // offset at the start of current accumulation window
	done     chan struct{}
	watcher  *fsnotify.Watcher
}

// Logger is the minimal interface Tailer needs for logging.
type Logger interface {
	Infof(format string, args ...interface{})
	Warnf(format string, args ...interface{})
	Debugf(format string, args ...interface{})
	Errorf(format string, args ...interface{})
}

// New creates and starts a Tailer.
func New(cfg config.Config, log Logger) (*Tailer, error) {
	t := &Tailer{
		cfg:  cfg,
		log:  log,
		done: make(chan struct{}),
	}

	// Load or initialise state.
	if err := t.loadState(); err != nil {
		return nil, fmt.Errorf("loading state: %w", err)
	}

	// Open the log file.
	if err := t.openFile(); err != nil {
		return nil, err
	}

	// Start fsnotify watcher on the log file's directory (more reliable than
	// watching the file directly, since rotation replaces the file).
	w, err := fsnotify.NewWatcher()
	if err != nil {
		t.file.Close()
		return nil, fmt.Errorf("creating fsnotify watcher: %w", err)
	}
	t.watcher = w
	if err := w.Add(filepath.Dir(cfg.P4LogFile)); err != nil {
		t.file.Close()
		w.Close()
		return nil, fmt.Errorf("watching directory %q: %w", filepath.Dir(cfg.P4LogFile), err)
	}

	// Background goroutine: drain fsnotify events so the channel doesn't fill.
	// The actual reads happen in Flush(), called on the LogTailDelay ticker.
	go t.drainEvents()

	return t, nil
}

// loadState reads the state file, or initialises fresh state.
func (t *Tailer) loadState() error {
	data, err := os.ReadFile(t.cfg.StateFile)
	if os.IsNotExist(err) {
		// No state file — fresh start.
		t.state = State{}
		return nil
	}
	if err != nil {
		return fmt.Errorf("reading state file %q: %w", t.cfg.StateFile, err)
	}
	if err := json.Unmarshal(data, &t.state); err != nil {
		t.log.Warnf("state file %q is corrupt (%v); starting fresh", t.cfg.StateFile, err)
		t.state = State{}
	}
	return nil
}

// openFile opens the P4LOG and positions the read cursor, performing rotation
// detection and handling the ReadFromStart / MaxRotationRecoverySize logic.
func (t *Tailer) openFile() error {
	f, err := os.Open(t.cfg.P4LogFile)
	if err != nil {
		return fmt.Errorf("opening P4LogFile %q: %w", t.cfg.P4LogFile, err)
	}

	inode, err := fileInode(f)
	if err != nil {
		f.Close()
		return fmt.Errorf("stat P4LogFile: %w", err)
	}

	info, err := f.Stat()
	if err != nil {
		f.Close()
		return fmt.Errorf("stat P4LogFile: %w", err)
	}

	switch {
	case t.state.Inode == 0:
		// First ever run (no state file).
		t.handleFirstRun(f, inode, info.Size())
	case inode == t.state.Inode:
		// Same file — resume from saved offset.
		if t.state.Offset > info.Size() {
			// Truncation without inode change (copytruncate style).
			t.log.Warnf("P4LOG appears truncated (saved offset %d > file size %d); treating as rotation",
				t.state.Offset, info.Size())
			t.handleRotation(f, inode, info.Size())
		} else {
			if _, err := f.Seek(t.state.Offset, io.SeekStart); err != nil {
				f.Close()
				return fmt.Errorf("seeking to saved offset %d: %w", t.state.Offset, err)
			}
			t.log.Infof("Resuming P4LOG from offset %d (inode %d)", t.state.Offset, inode)
		}
	default:
		// Inode changed — log rotation detected.
		t.log.Infof("Log rotation detected (inode changed %d → %d)", t.state.Inode, inode)
		t.handleRotation(f, inode, info.Size())
	}

	t.file = f
	t.startOff = t.state.Offset
	return nil
}

func (t *Tailer) handleFirstRun(f *os.File, inode uint64, size int64) {
	t.state.Inode = inode
	if t.cfg.ReadFromStart {
		t.log.Infof("First run: reading P4LOG from beginning (inode %d, size %d)", inode, size)
		t.state.Offset = 0
	} else {
		t.log.Infof("First run: starting from end of P4LOG (inode %d, size %d)", inode, size)
		f.Seek(0, io.SeekEnd)
		t.state.Offset = size
	}
}

func (t *Tailer) handleRotation(f *os.File, inode uint64, size int64) {
	t.state.Inode = inode
	if t.cfg.MaxRotationRecoverySize == 0 || size <= t.cfg.MaxRotationRecoverySize {
		t.log.Infof("Reading new P4LOG from beginning (size %d ≤ MaxRotationRecoverySize %d)",
			size, t.cfg.MaxRotationRecoverySize)
		t.state.Offset = 0
	} else {
		t.log.Warnf("New P4LOG size %d exceeds MaxRotationRecoverySize %d; starting from EOF to avoid overload",
			size, t.cfg.MaxRotationRecoverySize)
		f.Seek(0, io.SeekEnd)
		t.state.Offset = size
	}
}

// drainEvents keeps the fsnotify channel clear so it never blocks.
// Actual data reads happen in Flush(), not here.
func (t *Tailer) drainEvents() {
	for {
		select {
		case <-t.done:
			return
		case _, ok := <-t.watcher.Events:
			if !ok {
				return
			}
			// We don't act on individual events; Flush() reads everything
			// available each tick. The watcher just prevents us from missing
			// a rotation between ticks.
		case err, ok := <-t.watcher.Errors:
			if !ok {
				return
			}
			t.log.Errorf("fsnotify error: %v", err)
		}
	}
}

// Flush reads all bytes available since the last call and returns a Chunk.
// It also checks for log rotation (inode change or truncation).
// Returns a nil Chunk if no new data was available.
func (t *Tailer) Flush() (*Chunk, error) {
	t.mu.Lock()
	defer t.mu.Unlock()

	if err := t.checkRotation(); err != nil {
		return nil, err
	}

	buf := make([]byte, 32*1024)
	for {
		n, err := t.file.Read(buf)
		if n > 0 {
			t.buf = append(t.buf, buf[:n]...)
			t.state.Offset += int64(n)
		}
		if err == io.EOF {
			break
		}
		if err != nil {
			return nil, fmt.Errorf("reading P4LOG: %w", err)
		}
	}

	if len(t.buf) == 0 {
		return nil, nil
	}

	chunk := &Chunk{
		Data:        t.buf,
		StartOffset: t.startOff,
		EndOffset:   t.state.Offset,
	}
	t.buf = nil
	t.startOff = t.state.Offset
	return chunk, nil
}

// checkRotation detects inode changes and truncation, reopening the file
// as needed. Must be called with t.mu held.
func (t *Tailer) checkRotation() error {
	// Stat the path (not the open fd) to see if the file was replaced.
	pathInfo, err := os.Stat(t.cfg.P4LogFile)
	if os.IsNotExist(err) {
		// File temporarily absent (mid-rotation); wait for next tick.
		t.log.Warnf("P4LOG %q not found; waiting for rotation to complete", t.cfg.P4LogFile)
		return nil
	}
	if err != nil {
		return fmt.Errorf("stat %q: %w", t.cfg.P4LogFile, err)
	}

	newInode := inodeFromInfo(pathInfo)
	truncated := t.state.Offset > pathInfo.Size()

	if newInode == t.state.Inode && !truncated {
		return nil // no rotation
	}

	// Rotation or truncation detected.
	if truncated && newInode == t.state.Inode {
		t.log.Warnf("P4LOG truncated (copytruncate rotation?); resetting to offset 0")
	} else {
		// Flush whatever remains in the current file before switching.
		remaining, readErr := io.ReadAll(t.file)
		if readErr == nil && len(remaining) > 0 {
			t.buf = append(t.buf, remaining...)
			t.state.Offset += int64(len(remaining))
			t.log.Infof("Captured %d trailing bytes from rotated P4LOG before switching", len(remaining))
		}
		t.log.Infof("Log rotation detected during flush (inode %d → %d)", t.state.Inode, newInode)
	}

	t.file.Close()
	newFile, err := os.Open(t.cfg.P4LogFile)
	if err != nil {
		return fmt.Errorf("reopening P4LOG after rotation: %w", err)
	}
	t.file = newFile
	t.handleRotation(newFile, newInode, pathInfo.Size())
	t.startOff = t.state.Offset
	return nil
}

// SaveState atomically writes the current state to StateFile.
func (t *Tailer) SaveState() error {
	t.mu.Lock()
	state := t.state
	t.mu.Unlock()

	data, err := json.MarshalIndent(state, "", "  ")
	if err != nil {
		return err
	}
	tmp := t.cfg.StateFile + ".tmp"
	if err := os.WriteFile(tmp, data, 0600); err != nil {
		return fmt.Errorf("writing state file: %w", err)
	}
	return os.Rename(tmp, t.cfg.StateFile)
}

// Close shuts down the tailer.
func (t *Tailer) Close() {
	close(t.done)
	t.watcher.Close()
	t.mu.Lock()
	defer t.mu.Unlock()
	if t.file != nil {
		t.file.Close()
	}
}

// WaitForFile blocks until the P4LOG file appears, retrying every retryInterval.
// Useful at startup if the file doesn't exist yet.
func WaitForFile(path string, retryInterval time.Duration, log Logger, done <-chan struct{}) error {
	for {
		if _, err := os.Stat(path); err == nil {
			return nil
		}
		log.Warnf("P4LOG %q not found; retrying in %s", path, retryInterval)
		select {
		case <-done:
			return fmt.Errorf("shutdown before P4LOG appeared")
		case <-time.After(retryInterval):
		}
	}
}

// fileInode returns the inode number of an open file.
func fileInode(f *os.File) (uint64, error) {
	info, err := f.Stat()
	if err != nil {
		return 0, err
	}
	return inodeFromInfo(info), nil
}

// inodeFromInfo extracts the inode from an os.FileInfo.
// On Linux/macOS this is sys.Ino; falls back to 0 on unsupported platforms.
func inodeFromInfo(info os.FileInfo) uint64 {
	if sys, ok := info.Sys().(*syscall.Stat_t); ok {
		return sys.Ino
	}
	return 0
}