go-synapse-backupper/pkg/adapters/pipeline/pipeline_test.go

package pipeline

import (
	"context"
	"errors"
	"io"
	"runtime"
	"testing"
	"time"

	"git.tswf.io/infra/go-synapse-backupper/pkg/domain"
	"git.tswf.io/infra/go-synapse-backupper/pkg/domain/crypto"
	"git.tswf.io/infra/go-synapse-backupper/pkg/domain/pgdump"
)

var (
	errPgDumpFailed  = errors.New("pg_dump failed")
	errEncryptFailed = errors.New("encryption failed")
)

type fakeSink struct {
	transaction *fakeSinkTx
}

func (sink *fakeSink) Begin(key string) (domain.SinkTx, error) {
	sink.transaction = &fakeSinkTx{}
	return sink.transaction, nil
}

func (sink *fakeSink) List(prefix string) ([]string, error) {
	return make([]string, 0), nil
}

func (sink *fakeSink) Remove(key string) error {
	return nil
}

type fakeSinkTx struct {
	committed bool
	aborted   bool
	data      []byte
}

func (transaction *fakeSinkTx) Write(p []byte) (int, error) {
	transaction.data = append(transaction.data, p...)
	return len(p), nil
}

func (transaction *fakeSinkTx) Commit() error {
	transaction.committed = true
	return nil
}

func (transaction *fakeSinkTx) Abort() error {
	transaction.aborted = true
	return nil
}

type fakeDumper struct {
	writeBytes int
	returnErr  error
	closePipe  bool
}

func (dumper *fakeDumper) Dump(
	ctx context.Context,
	opts pgdump.Options,
	writer io.Writer,
) error {
	if dumper.writeBytes > 0 {
		data := make([]byte, dumper.writeBytes)
		if _, err := writer.Write(data); err != nil {
			return err
		}
	}
	if dumper.closePipe {
		if closer, ok := writer.(io.Closer); ok {
			_ = closer.Close()
		}
	}
	return dumper.returnErr
}

type fakeEncryptor struct {
	readBytes int
	returnErr error
}

func (encryptor *fakeEncryptor) Encrypt(
	plaintext io.Reader,
	recipients []crypto.RecipientPub,
	sink io.Writer,
	rand io.Reader,
) error {
	if encryptor.readBytes > 0 {
		buf := make([]byte, encryptor.readBytes)
		if _, err := io.ReadFull(plaintext, buf); err != nil {
			return err
		}
	}
	return encryptor.returnErr
}

func countGoroutines() int {
	return runtime.NumGoroutine()
}

func waitForGoroutinesStable(baseline int) bool {
	deadline := time.Now().Add(time.Second)
	for time.Now().Before(deadline) {
		time.Sleep(10 * time.Millisecond)
		if runtime.NumGoroutine() <= baseline {
			return true
		}
	}
	return false
}

func TestPipeline_DumpReturnsWithoutClosing(t *testing.T) {
	baseline := countGoroutines()

	sink := &fakeSink{}
	dumper := &fakeDumper{
		writeBytes: 4 * 1024,
		returnErr:  errPgDumpFailed,
		closePipe:  false,
	}
	encryptor := &fakeEncryptor{
		readBytes: 4 * 1024,
	}

	runner := NewRunner(
		WithDumper(dumper),
		WithEncryptor(encryptor),
	)

	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
	defer cancel()

	err := runner.Run(
		ctx,
		pgdump.Options{Key: "backup.sql"},
		make([]crypto.RecipientPub, 0),
		sink,
		nil,
	)

	if !errors.Is(err, errPgDumpFailed) {
		t.Fatalf("expected errPgDumpFailed, got %v", err)
	}

	if !sink.transaction.aborted {
		t.Fatalf("expected transaction to be aborted on error")
	}

	if waitForGoroutinesStable(baseline) {
		return
	}

	t.Fatalf("goroutine leak detected: baseline %d, current %d", baseline, countGoroutines())
}

func TestPipeline_EncryptFailsFirst(t *testing.T) {
	baseline := countGoroutines()

	sink := &fakeSink{}
	dumper := &fakeDumper{
		writeBytes: 64 * 1024,
		returnErr:  nil,
		closePipe:  false,
	}
	encryptor := &fakeEncryptor{
		readBytes: 1024,
		returnErr: errEncryptFailed,
	}

	runner := NewRunner(
		WithDumper(dumper),
		WithEncryptor(encryptor),
	)

	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
	defer cancel()

	err := runner.Run(
		ctx,
		pgdump.Options{Key: "backup.sql"},
		make([]crypto.RecipientPub, 0),
		sink,
		nil,
	)

	if !errors.Is(err, errEncryptFailed) {
		t.Fatalf("expected errEncryptFailed, got %v", err)
	}

	if !sink.transaction.aborted {
		t.Fatalf("expected transaction to be aborted on error")
	}

	if waitForGoroutinesStable(baseline) {
		return
	}

	t.Fatalf("goroutine leak detected: baseline %d, current %d", baseline, countGoroutines())
}

func TestPipeline_SuccessfulRunCommits(t *testing.T) {
	baseline := countGoroutines()

	sink := &fakeSink{}
	dumper := &fakeDumper{
		writeBytes: 4 * 1024,
		returnErr:  nil,
		closePipe:  true,
	}
	encryptor := &fakeEncryptor{
		readBytes: 4 * 1024,
	}

	runner := NewRunner(
		WithDumper(dumper),
		WithEncryptor(encryptor),
	)

	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
	defer cancel()

	err := runner.Run(
		ctx,
		pgdump.Options{Key: "backup.sql"},
		make([]crypto.RecipientPub, 0),
		sink,
		nil,
	)
	if err != nil {
		t.Fatalf("unexpected error: %v", err)
	}

	if !sink.transaction.committed {
		t.Fatalf("expected transaction to be committed on success")
	}

	if waitForGoroutinesStable(baseline) {
		return
	}

	t.Fatalf("goroutine leak detected: baseline %d, current %d", baseline, countGoroutines())
}