Function TestCompactor_CompactClosesPipeOnWriteError

compactor_test.go:91–120

(t *testing.T)

}

func TestCompactor_CompactClosesPipeOnWriteError(t *testing.T) {
	client := newEarlyReturnCompactionClient(t.TempDir())
	compactor := litestream.NewCompactor(client, slog.Default())

	createTestLTXFile(t, client, 0, 1, 1)
	createTestLTXFile(t, client, 0, 2, 2)
	client.failWrites = true

	before := countCompactorPipeWriters()
	if _, err := compactor.Compact(context.Background(), 1); err == nil {
		t.Fatal("expected error")
	} else if !strings.Contains(err.Error(), "write ltx file: early write failure") {
		t.Fatalf("unexpected error: %v", err)
	}

	deadline := time.NewTimer(time.Second)
	defer deadline.Stop()
	ticker := time.NewTicker(10 * time.Millisecond)
	defer ticker.Stop()
	for {
		if got := countCompactorPipeWriters(); got <= before {
			break
		}
		select {
		case <-deadline.C:
			t.Fatalf("compactor goroutine leaked: got %d, want <= %d", countCompactorPipeWriters(), before)
		case <-ticker.C:
		}
	}
}

func TestCompactor_MaxLTXFileInfo(t *testing.T) {
	t.Run("WithFiles", func(t *testing.T) {