tailer_test.go #1

package tailer_test

import (
	"os"
	"path/filepath"
	"testing"

	"workshop.perforce.com/p4lf/internal/config"
	"workshop.perforce.com/p4lf/internal/tailer"
)

type nopLogger struct{}

func (nopLogger) Infof(string, ...interface{})  {}
func (nopLogger) Warnf(string, ...interface{})  {}
func (nopLogger) Debugf(string, ...interface{}) {}
func (nopLogger) Errorf(string, ...interface{}) {}

func testConfig(t *testing.T, logPath string) config.Config {
	t.Helper()
	cfg := config.Defaults()
	cfg.P4LogFile = logPath
	cfg.StateFile = logPath + ".state"
	cfg.LogChunksDir = filepath.Dir(logPath)
	cfg.ReadFromStart = true
	cfg.MaxRotationRecoverySize = 500 * 1024 * 1024
	return cfg
}

func appendToFile(t *testing.T, path string, data []byte) {
	t.Helper()
	f, err := os.OpenFile(path, os.O_APPEND|os.O_WRONLY, 0644)
	if err != nil {
		t.Fatalf("appendToFile open %q: %v", path, err)
	}
	defer f.Close()
	if _, err := f.Write(data); err != nil {
		t.Fatalf("appendToFile write %q: %v", path, err)
	}
}

// TestFlush_ReadFromStart verifies that on first run (no state file), all
// existing content in the P4LOG is returned in the first flush.
func TestFlush_ReadFromStart(t *testing.T) {
	dir := t.TempDir()
	logPath := filepath.Join(dir, "log")
	content := []byte("line1\nline2\n")
	if err := os.WriteFile(logPath, content, 0644); err != nil {
		t.Fatal(err)
	}

	cfg := testConfig(t, logPath)
	cfg.ReadFromStart = true

	tr, err := tailer.New(cfg, nopLogger{})
	if err != nil {
		t.Fatalf("New: %v", err)
	}
	defer tr.Close()

	chunk, err := tr.Flush()
	if err != nil {
		t.Fatalf("Flush: %v", err)
	}
	if chunk == nil {
		t.Fatal("expected chunk, got nil")
	}
	if string(chunk.Data) != string(content) {
		t.Errorf("chunk data: got %q, want %q", chunk.Data, content)
	}
	if chunk.StartOffset != 0 {
		t.Errorf("StartOffset: got %d, want 0", chunk.StartOffset)
	}
	if chunk.EndOffset != int64(len(content)) {
		t.Errorf("EndOffset: got %d, want %d", chunk.EndOffset, len(content))
	}
}

// TestFlush_ReadFromEnd verifies that with ReadFromStart=false, existing content
// is skipped and only newly appended data is returned.
func TestFlush_ReadFromEnd(t *testing.T) {
	dir := t.TempDir()
	logPath := filepath.Join(dir, "log")
	existing := []byte("existing content\n")
	if err := os.WriteFile(logPath, existing, 0644); err != nil {
		t.Fatal(err)
	}

	cfg := testConfig(t, logPath)
	cfg.ReadFromStart = false

	tr, err := tailer.New(cfg, nopLogger{})
	if err != nil {
		t.Fatalf("New: %v", err)
	}
	defer tr.Close()

	// First flush: no new data (started at EOF).
	chunk, err := tr.Flush()
	if err != nil {
		t.Fatalf("Flush: %v", err)
	}
	if chunk != nil {
		t.Errorf("expected nil chunk (started at EOF), got %q", chunk.Data)
	}

	// Append and flush: should return only the appended bytes.
	appended := []byte("new stuff\n")
	appendToFile(t, logPath, appended)

	chunk, err = tr.Flush()
	if err != nil {
		t.Fatalf("Flush after append: %v", err)
	}
	if chunk == nil {
		t.Fatal("expected chunk after append, got nil")
	}
	if string(chunk.Data) != string(appended) {
		t.Errorf("appended chunk: got %q, want %q", chunk.Data, appended)
	}
	if chunk.StartOffset != int64(len(existing)) {
		t.Errorf("StartOffset: got %d, want %d", chunk.StartOffset, len(existing))
	}
}

// TestFlush_StateResume verifies that after saving state and creating a new
// tailer, the second tailer resumes exactly where the first left off with no
// duplicate or missing bytes.
func TestFlush_StateResume(t *testing.T) {
	dir := t.TempDir()
	logPath := filepath.Join(dir, "log")

	initial := []byte("initial data\n")
	if err := os.WriteFile(logPath, initial, 0644); err != nil {
		t.Fatal(err)
	}

	cfg := testConfig(t, logPath)

	// First tailer: read initial content, save state, close.
	tr1, err := tailer.New(cfg, nopLogger{})
	if err != nil {
		t.Fatalf("New: %v", err)
	}
	chunk1, err := tr1.Flush()
	if err != nil || chunk1 == nil {
		t.Fatalf("Flush 1: err=%v chunk=%v", err, chunk1)
	}
	if err := tr1.SaveState(); err != nil {
		t.Fatalf("SaveState: %v", err)
	}
	tr1.Close()

	// Append more data before creating second tailer.
	more := []byte("more data\n")
	appendToFile(t, logPath, more)

	// Second tailer: must resume from saved offset, returning only new bytes.
	tr2, err := tailer.New(cfg, nopLogger{})
	if err != nil {
		t.Fatalf("New (resume): %v", err)
	}
	defer tr2.Close()

	chunk2, err := tr2.Flush()
	if err != nil || chunk2 == nil {
		t.Fatalf("Flush (resume): err=%v chunk=%v", err, chunk2)
	}
	if string(chunk2.Data) != string(more) {
		t.Errorf("resumed chunk: got %q, want %q", chunk2.Data, more)
	}
	if chunk2.StartOffset != int64(len(initial)) {
		t.Errorf("StartOffset: got %d, want %d", chunk2.StartOffset, len(initial))
	}
	if chunk2.EndOffset != int64(len(initial)+len(more)) {
		t.Errorf("EndOffset: got %d, want %d", chunk2.EndOffset, len(initial)+len(more))
	}
}

// TestFlush_RotationDetected verifies that when the P4LOG is replaced (inode
// changes), the tailer detects the rotation and reads from the start of the
// new file.
func TestFlush_RotationDetected(t *testing.T) {
	dir := t.TempDir()
	logPath := filepath.Join(dir, "log")

	original := []byte("original log data\n")
	if err := os.WriteFile(logPath, original, 0644); err != nil {
		t.Fatal(err)
	}

	cfg := testConfig(t, logPath)

	tr, err := tailer.New(cfg, nopLogger{})
	if err != nil {
		t.Fatalf("New: %v", err)
	}
	defer tr.Close()

	// Flush original content fully.
	chunk, err := tr.Flush()
	if err != nil || chunk == nil {
		t.Fatalf("Flush (pre-rotation): err=%v chunk=%v", err, chunk)
	}
	if string(chunk.Data) != string(original) {
		t.Errorf("pre-rotation data: got %q, want %q", chunk.Data, original)
	}

	// Simulate rotation: remove old file, create new one at the same path.
	if err := os.Remove(logPath); err != nil {
		t.Fatal(err)
	}
	newContent := []byte("new log after rotation\n")
	if err := os.WriteFile(logPath, newContent, 0644); err != nil {
		t.Fatal(err)
	}

	// Flush after rotation: tailer detects inode change, reads from offset 0.
	chunk, err = tr.Flush()
	if err != nil {
		t.Fatalf("Flush (post-rotation): %v", err)
	}
	if chunk == nil {
		t.Fatal("expected chunk after rotation, got nil")
	}
	if string(chunk.Data) != string(newContent) {
		t.Errorf("post-rotation data: got %q, want %q", chunk.Data, newContent)
	}
	if chunk.StartOffset != 0 {
		t.Errorf("StartOffset after rotation: got %d, want 0", chunk.StartOffset)
	}
	if chunk.EndOffset != int64(len(newContent)) {
		t.Errorf("EndOffset after rotation: got %d, want %d", chunk.EndOffset, len(newContent))
	}
}

// TestFlush_NoNewData verifies that Flush returns nil when there is no new data.
func TestFlush_NoNewData(t *testing.T) {
	dir := t.TempDir()
	logPath := filepath.Join(dir, "log")
	if err := os.WriteFile(logPath, []byte{}, 0644); err != nil {
		t.Fatal(err)
	}

	cfg := testConfig(t, logPath)
	tr, err := tailer.New(cfg, nopLogger{})
	if err != nil {
		t.Fatalf("New: %v", err)
	}
	defer tr.Close()

	chunk, err := tr.Flush()
	if err != nil {
		t.Fatalf("Flush: %v", err)
	}
	if chunk != nil {
		t.Errorf("expected nil from empty file, got %q", chunk.Data)
	}
}

// TestFlush_OffsetAccumulation verifies that consecutive flushes produce
// contiguous, non-overlapping offset windows.
func TestFlush_OffsetAccumulation(t *testing.T) {
	dir := t.TempDir()
	logPath := filepath.Join(dir, "log")
	if err := os.WriteFile(logPath, []byte{}, 0644); err != nil {
		t.Fatal(err)
	}

	cfg := testConfig(t, logPath)
	tr, err := tailer.New(cfg, nopLogger{})
	if err != nil {
		t.Fatalf("New: %v", err)
	}
	defer tr.Close()

	batch1 := []byte("batch one\n")
	appendToFile(t, logPath, batch1)
	chunk1, err := tr.Flush()
	if err != nil || chunk1 == nil {
		t.Fatalf("Flush batch1: err=%v chunk=%v", err, chunk1)
	}

	batch2 := []byte("batch two\n")
	appendToFile(t, logPath, batch2)
	chunk2, err := tr.Flush()
	if err != nil || chunk2 == nil {
		t.Fatalf("Flush batch2: err=%v chunk=%v", err, chunk2)
	}

	if chunk2.StartOffset != chunk1.EndOffset {
		t.Errorf("gap or overlap: chunk1.End=%d chunk2.Start=%d", chunk1.EndOffset, chunk2.StartOffset)
	}
	if chunk2.EndOffset != chunk1.EndOffset+int64(len(batch2)) {
		t.Errorf("chunk2.EndOffset: got %d, want %d", chunk2.EndOffset, chunk1.EndOffset+int64(len(batch2)))
	}
	if string(chunk2.Data) != string(batch2) {
		t.Errorf("chunk2 data: got %q, want %q", chunk2.Data, batch2)
	}
}