Parallele Dateisynchronisierung mit Go und MinIO

(
  set -e
  mkdir minio-sync
  cd minio-sync
  GOBIN="$PWD/bin" GOWORK=off GOTOOLCHAIN=local \
    go install github.com/minio/minio@RELEASE.2025-10-15T17-29-55Z
)

(
  cd minio-sync &&
  MINIO_ROOT_USER=minioadmin MINIO_ROOT_PASSWORD=minioadmin \
    ./bin/minio server ./data --address 127.0.0.1:9000 --console-address 127.0.0.1:9001
)

(
  set -e
  curl -fsS http://127.0.0.1:9000/minio/health/ready
  cd minio-sync
  GOWORK=off GOTOOLCHAIN=local go mod init minio-sync
  GOWORK=off GOTOOLCHAIN=local go get github.com/minio/minio-go/v7@v7.0.95
)

// _seed/main.go
package main

import (
	"context"
	"fmt"
	"log"
	"os"
	"path/filepath"
	"time"

	"github.com/minio/minio-go/v7"
	"github.com/minio/minio-go/v7/pkg/credentials"
)

func main() {
	if err := seed(); err != nil {
		log.Fatal(err)
	}
	log.Println("Uploaded 303 sample objects")
}

func seed() error {
	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Minute)
	defer cancel()
	client, err := minio.New("localhost:9000", &minio.Options{
		Creds: credentials.NewStaticV4("minioadmin", "minioadmin", ""),
		Secure: false,
	})
	if err != nil {
		return err
	}
	const bucket = "my-sync-bucket"
	exists, err := client.BucketExists(ctx, bucket)
	if err != nil {
		return err
	}
	if !exists {
		if err := client.MakeBucket(ctx, bucket, minio.MakeBucketOptions{}); err != nil {
			return err
		}
	}
	files := map[string][]byte{
		"notes/hello.txt": []byte("Hello from MinIO\n"),
		"empty.bin":       {},
		"binary.bin":      {0, 255, 128, 13, 10, 42, 0, 10},
	}
	for i := 0; i < 300; i++ {
		files[fmt.Sprintf("logs/object-%03d.txt", i)] = []byte(fmt.Sprintf("object %d\n", i))
	}
	for key, body := range files {
		filename := filepath.Join("sample-input", filepath.FromSlash(key))
		if err := os.MkdirAll(filepath.Dir(filename), 0o700); err != nil {
			return err
		}
		if err := os.WriteFile(filename, body, 0o600); err != nil {
			return err
		}
		if _, err := client.FPutObject(ctx, bucket, key, filename, minio.PutObjectOptions{}); err != nil {
			return fmt.Errorf("upload %q: %w", key, err)
		}
	}
	return nil
}

(
  cd minio-sync &&
  GOWORK=off GOTOOLCHAIN=local go run ./_seed
)

// client.go
package main

import (
	"context"
	"crypto/tls"
	"fmt"
	"net/http"
	"time"

	"github.com/minio/minio-go/v7"
	"github.com/minio/minio-go/v7/pkg/credentials"
)

// bucketName matches the bucket populated by _seed/main.go.
const bucketName = "my-sync-bucket"

func createMinioClient(ctx context.Context) (*minio.Client, error) {
	endpoint := "localhost:9000"
	accessKeyID := "minioadmin"     // Use environment variables in production
	secretAccessKey := "minioadmin" // Use environment variables in production
	useSSL := false                 // true once your endpoint terminates TLS

	// Only exercised when useSSL is true
	transport := &http.Transport{
		TLSClientConfig: &tls.Config{MinVersion: tls.VersionTLS12},
		IdleConnTimeout: 90 * time.Second,
	}

	// Initialize minio client
	opts := &minio.Options{
		Creds:     credentials.NewStaticV4(accessKeyID, secretAccessKey, ""),
		Secure:    useSSL,
		Transport: transport,
	}

	client, err := minio.New(endpoint, opts)
	if err != nil {
		return nil, err
	}

	// BucketExists reports a bucket that is simply absent as (false, nil), so the boolean
	// has to be checked too. Ignoring it turns a typo in the bucket name into a sync that
	// quietly downloads nothing.
	exists, err := client.BucketExists(ctx, bucketName)
	if err != nil {
		return nil, fmt.Errorf("failed to reach bucket %q: %w", bucketName, err)
	}
	if !exists {
		return nil, fmt.Errorf("bucket %q does not exist", bucketName)
	}

	return client, nil
}

// sync.go
package main

import (
	"context"
	"fmt"
	"io"
	"os"
	"path"
	"path/filepath"
	"strings"
	"sync"
	"time"

	"github.com/minio/minio-go/v7"
)

type DownloadResult struct {
	ObjectName string
	Error      error
}

func downloadFiles(ctx context.Context, client *minio.Client, bucketName string, outputDir string) error {
	// 0700, because these are private copies. The process umask would otherwise usually
	// leave them group- and world-readable.
	if err := os.MkdirAll(outputDir, 0o700); err != nil {
		return fmt.Errorf("failed to create output directory: %w", err)
	}

	// Resolve the directory once so the per-object checks below compare against a path
	// that has no symlinks left in it.
	root, err := filepath.EvalSymlinks(outputDir)
	if err != nil {
		return fmt.Errorf("failed to resolve output directory: %w", err)
	}

	// Cancelling here unblocks every worker if we bail out early
	ctx, cancel := context.WithCancel(ctx)
	defer cancel()

	jobs := make(chan string, 100)
	results := make(chan DownloadResult, 100)

	var wg sync.WaitGroup
	workerCount := 5

	// Start workers
	for i := 0; i < workerCount; i++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			for objectName := range jobs {
				err := downloadObject(ctx, client, bucketName, objectName, root)
				select {
				case results <- DownloadResult{ObjectName: objectName, Error: err}:
				case <-ctx.Done():
					return
				}
			}
		}()
	}

	go func() {
		wg.Wait()
		close(results)
	}()

	// Collect results while the producer is still queueing. Draining only after
	// close(jobs) deadlocks as soon as the buffers fill: workers block on
	// `results <-`, stop reading `jobs`, and the producer blocks on `jobs <-`.
	//
	// Only a count and the first error are kept. A bucket can hold millions of objects,
	// and appending every failure to a slice would grow without bound in exactly the run
	// where things are going worst.
	type failures struct {
		count int
		first error
	}
	collected := make(chan failures, 1)
	go func() {
		var seen failures
		for result := range results {
			if result.Error == nil {
				continue
			}
			seen.count++
			if seen.first == nil {
				seen.first = fmt.Errorf("%s: %w", result.ObjectName, result.Error)
			}
		}
		collected <- seen
	}()

	// List and queue objects. The deferred close lets the workers drain and exit
	// on every path, including the early returns below.
	listErr := func() error {
		defer close(jobs)
		opts := minio.ListObjectsOptions{Recursive: true}
		for obj := range client.ListObjects(ctx, bucketName, opts) {
			if obj.Err != nil {
				// Cancel immediately so in-flight downloads stop now rather than running
				// to completion behind a listing that has already failed.
				cancel()
				return fmt.Errorf("error listing objects: %w", obj.Err)
			}
			// Empty directory markers contain no file bytes to download.
			if obj.Size == 0 && strings.HasSuffix(obj.Key, "/") {
				continue
			}
			select {
			case jobs <- obj.Key:
			case <-ctx.Done():
				return ctx.Err()
			}
		}
		return nil
	}()

	// Always drain first, so every worker has finished before we report anything.
	seen := <-collected

	if listErr != nil {
		return listErr
	}
	// A cancelled context ends the listing loop without an error of its own, and workers
	// that were cut off never deliver a result. Without this check a cancelled sync would
	// be indistinguishable from an empty bucket that synced cleanly.
	if err := ctx.Err(); err != nil {
		return fmt.Errorf("sync did not complete: %w", err)
	}
	if seen.count > 0 {
		return fmt.Errorf("encountered %d download errors, first: %w", seen.count, seen.first)
	}

	return nil
}

// resolveOutputPath maps a server-supplied object key onto a path inside outputDir, which
// must already be symlink-free (see filepath.EvalSymlinks above).
//
// Keys are rejected rather than cleaned. `a/./b.txt` and `a/b/../b.txt` are different keys
// that both clean to `a/b.txt`, so cleaning them would let one object silently overwrite
// another, and lexical checks alone cannot see a symlink that is already on disk.
func resolveOutputPath(outputDir, objectName string) (string, error) {
	if objectName == "" || objectName != path.Clean(objectName) || strings.Contains(objectName, `\`) {
		return "", fmt.Errorf("refusing non-canonical object key: %q", objectName)
	}
	if !filepath.IsLocal(filepath.FromSlash(objectName)) {
		return "", fmt.Errorf("refusing unsafe object key: %q", objectName)
	}

	current := outputDir
	for _, element := range strings.Split(objectName, "/") {
		current = filepath.Join(current, element)
		info, err := os.Lstat(current)
		if err != nil {
			if os.IsNotExist(err) {
				continue // Nothing here yet, so nothing can redirect the write
			}
			return "", fmt.Errorf("failed to inspect %q: %w", current, err)
		}
		if info.Mode()&os.ModeSymlink != 0 {
			return "", fmt.Errorf("refusing object key through a symlink: %q", objectName)
		}
	}

	return current, nil
}

func downloadObject(ctx context.Context, client *minio.Client, bucket, objectName, outputDir string) error {
	outputPath, err := resolveOutputPath(outputDir, objectName)
	if err != nil {
		return err
	}

	// Create context with timeout
	ctx, cancel := context.WithTimeout(ctx, 10*time.Minute)
	defer cancel()

	// Get object
	obj, err := client.GetObject(ctx, bucket, objectName, minio.GetObjectOptions{})
	if err != nil {
		return fmt.Errorf("failed to get object: %w", err)
	}
	defer obj.Close()

	if err := os.MkdirAll(filepath.Dir(outputPath), 0o700); err != nil {
		return fmt.Errorf("failed to create directories: %w", err)
	}

	// Download to a sibling temp file so a failed sync never truncates the copy
	// that a previous run completed, and never leaves a half-written file behind.
	// os.CreateTemp already creates the file 0600.
	temp, err := os.CreateTemp(filepath.Dir(outputPath), filepath.Base(outputPath)+".part-*")
	if err != nil {
		return fmt.Errorf("failed to create temporary file: %w", err)
	}
	tempPath := temp.Name()
	defer func() {
		temp.Close()
		os.Remove(tempPath) // No-op once the rename below succeeded
	}()

	if _, err := io.Copy(temp, obj); err != nil {
		return fmt.Errorf("failed to download file: %w", err)
	}
	if err := temp.Sync(); err != nil {
		return fmt.Errorf("failed to flush file: %w", err)
	}
	if err := temp.Close(); err != nil {
		return fmt.Errorf("failed to close file: %w", err)
	}

	// Rename is atomic within a filesystem, so readers see either the old file or
	// the complete new one
	if err := os.Rename(tempPath, outputPath); err != nil {
		return fmt.Errorf("failed to publish file: %w", err)
	}

	return nil
}

// main.go
package main

import (
	"context"
	"log"
	"os/signal"
	"syscall"
)

func main() {
	// Ctrl-C cancels the context, which stops the workers and makes downloadFiles
	// report the interruption instead of exiting as if the sync had finished.
	ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM)
	defer stop()

	// Create MinIO client
	client, err := createMinioClient(ctx)
	if err != nil {
		log.Fatalf("Failed to create MinIO client: %v", err)
	}

	// Start download. bucketName comes from client.go.
	if err := downloadFiles(ctx, client, bucketName, "./downloads"); err != nil {
		log.Fatalf("Download failed: %v", err)
	}

	log.Println("Download completed successfully")
}

(
  cd minio-sync &&
  GOWORK=off GOTOOLCHAIN=local go run .
)

(
  cd minio-sync &&
  diff -r sample-input downloads
)