pkg-proxy/internal/storage/blob_test.go

package storage

import (
	"context"
	"crypto/sha256"
	"encoding/hex"
	"errors"
	"io"
	"io/fs"
	"os"
	"path/filepath"
	"runtime"
	"strings"
	"sync"
	"sync/atomic"
	"testing"
	"time"

	"gocloud.dev/blob"
)

func TestOpenBucket(t *testing.T) {
	dir := t.TempDir()
	ctx := context.Background()

	b, err := OpenBucket(ctx, fileURLFromPath(dir))
	if err != nil {
		t.Fatalf("OpenBucket failed: %v", err)
	}
	defer func() { _ = b.Close() }()

	if b.URL() == "" {
		t.Error("URL() should not be empty")
	}
}

func TestBlobStore(t *testing.T) {
	b := createTestBlob(t)
	ctx := context.Background()
	content := "test content for blob storage"

	size, hash, err := b.Store(ctx, "npm/lodash/4.17.21/lodash.tgz", strings.NewReader(content))
	if err != nil {
		t.Fatalf("Store failed: %v", err)
	}

	if size != int64(len(content)) {
		t.Errorf("size = %d, want %d", size, len(content))
	}

	h := sha256.Sum256([]byte(content))
	wantHash := hex.EncodeToString(h[:])
	if hash != wantHash {
		t.Errorf("hash = %s, want %s", hash, wantHash)
	}
}

func TestBlobOpen(t *testing.T) {
	b := createTestBlob(t)
	ctx := context.Background()
	content := "readable content"

	_, _, _ = b.Store(ctx, "test/read.txt", strings.NewReader(content))

	r, err := b.Open(ctx, "test/read.txt")
	if err != nil {
		t.Fatalf("Open failed: %v", err)
	}
	defer func() { _ = r.Close() }()
	seeker, ok := r.(io.Seeker)
	if !ok {
		t.Fatal("local file reader does not implement io.Seeker")
	}
	if _, err := seeker.Seek(9, io.SeekStart); err != nil {
		t.Fatalf("Seek failed: %v", err)
	}

	data, err := io.ReadAll(r)
	if err != nil {
		t.Fatalf("ReadAll failed: %v", err)
	}
	if string(data) != "content" {
		t.Errorf("content after seek = %q, want %q", string(data), "content")
	}
}

func TestBlobOpenNotFound(t *testing.T) {
	b := createTestBlob(t)
	ctx := context.Background()

	_, err := b.Open(ctx, "does/not/exist.txt")
	if !errors.Is(err, ErrNotFound) {
		t.Errorf("Open non-existent = %v, want ErrNotFound", err)
	}
}

func TestBlobExists(t *testing.T) {
	b := createTestBlob(t)
	ctx := context.Background()

	exists, err := b.Exists(ctx, "test/exists.txt")
	if err != nil {
		t.Fatalf("Exists failed: %v", err)
	}
	if exists {
		t.Error("Exists returned true for non-existent file")
	}

	_, _, _ = b.Store(ctx, "test/exists.txt", strings.NewReader("content"))

	exists, err = b.Exists(ctx, "test/exists.txt")
	if err != nil {
		t.Fatalf("Exists after store failed: %v", err)
	}
	if !exists {
		t.Error("Exists returned false for existing file")
	}
}

func TestBlobDelete(t *testing.T) {
	b := createTestBlob(t)
	ctx := context.Background()

	_, _, _ = b.Store(ctx, "test/delete/nested/file.txt", strings.NewReader("content"))

	err := b.Delete(ctx, "test/delete/nested/file.txt")
	if err != nil {
		t.Fatalf("Delete failed: %v", err)
	}

	exists, _ := b.Exists(ctx, "test/delete/nested/file.txt")
	if exists {
		t.Error("file still exists after delete")
	}
}

// A fetch stores its object in a directory of its own, so Delete removes that
// directory once it is empty rather than leave one behind per deleted object.
// A version directory from the layout before fetch directories is left alone,
// since a new fetch may be creating its directory inside it.
func TestBlobDeleteRemovesEmptyFetchDirectory(t *testing.T) {
	dir := t.TempDir()
	b := openFileBlob(t, dir)
	ctx := context.Background()
	const (
		emptied = "npm/pkg/1.0.0/0123456789abcdef/pkg.tgz"
		shared  = "npm/pkg/1.0.0/fedcba9876543210/pkg.tgz"
		kept    = "npm/pkg/1.0.0/fedcba9876543210/other.tgz"
		legacy  = "npm/pkg/2.0.0/pkg.tgz"
	)
	for _, key := range []string{emptied, shared, kept, legacy} {
		if _, _, err := b.Store(ctx, key, strings.NewReader("content")); err != nil {
			t.Fatalf("Store(%q): %v", key, err)
		}
	}

	for _, key := range []string{emptied, shared, legacy} {
		if err := b.Delete(ctx, key); err != nil {
			t.Fatalf("Delete(%q): %v", key, err)
		}
	}

	pkg := filepath.Join(dir, "npm", "pkg")
	if _, err := os.Stat(filepath.Join(pkg, "1.0.0", "0123456789abcdef")); !os.IsNotExist(err) {
		t.Errorf("emptied fetch directory left behind (stat err %v)", err)
	}
	for _, d := range []string{filepath.Join("1.0.0", "fedcba9876543210"), "1.0.0", "2.0.0"} {
		if _, err := os.Stat(filepath.Join(pkg, d)); err != nil {
			t.Errorf("directory %s removed: %v", d, err)
		}
	}
	assertReadsBack(t, b, kept, "content")
}

func TestBlobDeleteNotFound(t *testing.T) {
	b := createTestBlob(t)
	ctx := context.Background()

	// Delete non-existent file should not error
	err := b.Delete(ctx, "does/not/exist.txt")
	if err != nil {
		t.Errorf("Delete non-existent = %v, want nil", err)
	}
}

func TestBlobSize(t *testing.T) {
	b := createTestBlob(t)
	ctx := context.Background()
	content := "size test content"

	_, _, _ = b.Store(ctx, "test/size.txt", strings.NewReader(content))

	size, err := b.Size(ctx, "test/size.txt")
	if err != nil {
		t.Fatalf("Size failed: %v", err)
	}
	if size != int64(len(content)) {
		t.Errorf("Size = %d, want %d", size, len(content))
	}
}

func TestBlobSizeNotFound(t *testing.T) {
	b := createTestBlob(t)
	ctx := context.Background()

	_, err := b.Size(ctx, "does/not/exist.txt")
	if !errors.Is(err, ErrNotFound) {
		t.Errorf("Size non-existent = %v, want ErrNotFound", err)
	}
}

func TestBlobUsedSpace(t *testing.T) {
	b := createTestBlob(t)
	ctx := context.Background()

	// Empty storage
	used, err := b.UsedSpace(ctx)
	if err != nil {
		t.Fatalf("UsedSpace failed: %v", err)
	}
	if used != 0 {
		t.Errorf("UsedSpace empty = %d, want 0", used)
	}

	// Add some files
	_, _, _ = b.Store(ctx, "a.txt", strings.NewReader("aaaa"))    // 4 bytes
	_, _, _ = b.Store(ctx, "b.txt", strings.NewReader("bbbbbb"))  // 6 bytes
	_, _, _ = b.Store(ctx, "c/d.txt", strings.NewReader("ccccc")) // 5 bytes

	used, err = b.UsedSpace(ctx)
	if err != nil {
		t.Fatalf("UsedSpace failed: %v", err)
	}
	if used != 15 {
		t.Errorf("UsedSpace = %d, want 15", used)
	}
}

func TestBlobLargeFile(t *testing.T) {
	assertLargeFileRoundTrip(t, createTestBlob(t))
}

func TestBlobSignedURLUnsupported(t *testing.T) {
	b := createTestBlob(t)
	ctx := context.Background()

	// fileblob has no URL signer configured, so this must surface as
	// ErrSignedURLUnsupported rather than a generic error.
	_, err := b.SignedURL(ctx, "test/file.txt", time.Minute)
	if !errors.Is(err, ErrSignedURLUnsupported) {
		t.Errorf("SignedURL on fileblob = %v, want ErrSignedURLUnsupported", err)
	}
}

func TestBlobOverwrite(t *testing.T) {
	b := createTestBlob(t)
	ctx := context.Background()

	// Store initial content
	_, _, err := b.Store(ctx, "test/file.txt", strings.NewReader("initial"))
	if err != nil {
		t.Fatalf("initial Store failed: %v", err)
	}

	// Overwrite with new content
	_, _, err = b.Store(ctx, "test/file.txt", strings.NewReader("updated"))
	if err != nil {
		t.Fatalf("update Store failed: %v", err)
	}

	// Verify updated content
	r, err := b.Open(ctx, "test/file.txt")
	if err != nil {
		t.Fatalf("Open failed: %v", err)
	}
	defer func() { _ = r.Close() }()

	data, _ := io.ReadAll(r)
	if string(data) != "updated" {
		t.Errorf("content = %q, want %q", string(data), "updated")
	}
}

func TestOpenBucketSetsNoTmpDir(t *testing.T) {
	dir := t.TempDir()
	ctx := context.Background()

	b, err := OpenBucket(ctx, fileURLFromPath(dir))
	if err != nil {
		t.Fatalf("OpenBucket failed: %v", err)
	}
	defer func() { _ = b.Close() }()

	// fileblob uses os.TempDir() by default for temp files, then os.Rename to
	// the final path. This fails with "invalid cross-device link" when the bucket
	// dir and os.TempDir() are on different filesystems (e.g. Docker volumes).
	// OpenBucket must set no_tmp_dir=true so temp files are created next to the
	// final path instead.
	if !strings.Contains(b.URL(), "no_tmp_dir=true") {
		t.Errorf("URL should contain no_tmp_dir=true to avoid cross-device rename errors, got %q", b.URL())
	}

	// Verify Store still works with the parameter set
	content := "cross-device test"
	_, _, err = b.Store(ctx, "test/cross-device.txt", strings.NewReader(content))
	if err != nil {
		t.Fatalf("Store failed with no_tmp_dir=true: %v", err)
	}

	r, err := b.Open(ctx, "test/cross-device.txt")
	if err != nil {
		t.Fatalf("Open failed: %v", err)
	}
	defer func() { _ = r.Close() }()

	data, _ := io.ReadAll(r)
	if string(data) != content {
		t.Errorf("content = %q, want %q", string(data), content)
	}
}

func createTestBlob(t *testing.T) *Blob {
	t.Helper()
	dir := t.TempDir()
	ctx := context.Background()

	b, err := OpenBucket(ctx, fileURLFromPath(dir))
	if err != nil {
		t.Fatalf("OpenBucket failed: %v", err)
	}
	t.Cleanup(func() { _ = b.Close() })
	blob, ok := b.(*Blob)
	if !ok {
		t.Fatalf("OpenBucket returned %T, want *Blob", b)
	}
	return blob
}

func fileURLFromPath(path string) string {
	if runtime.GOOS == osWindows {
		// Windows paths need file:///C:/path format
		path = filepath.ToSlash(path)
		return "file:///" + path
	}
	return "file://" + path
}

func TestOpenBucketWritesNoAttrsSidecar(t *testing.T) {
	dir := t.TempDir()
	ctx := context.Background()

	b, err := OpenBucket(ctx, fileURLFromPath(dir))
	if err != nil {
		t.Fatalf("OpenBucket failed: %v", err)
	}
	defer func() { _ = b.Close() }()

	if _, _, err := b.Store(ctx, "pkg/thing-1.0.0.tgz", strings.NewReader("content")); err != nil {
		t.Fatalf("Store failed: %v", err)
	}

	sidecars, err := filepath.Glob(filepath.Join(dir, "*", "*.attrs"))
	if err != nil {
		t.Fatalf("Glob failed: %v", err)
	}
	if len(sidecars) != 0 {
		t.Errorf("got sidecar files %v, want none: a truncated sidecar fails reads that overlap a write", sidecars)
	}
}

// A read overlapping a write to the same key must not fail. fileblob rewrote
// its ".attrs" sidecar in place, so a reader decoding it mid-write saw a
// partial file, which the proxy served as a 502 on an artifact it held.
func TestConcurrentReadsSurviveWritesToSameKey(t *testing.T) {
	if runtime.GOOS == "windows" {
		// Go opens files without FILE_SHARE_DELETE, so a writer cannot replace
		// a file a reader holds open: its rename fails with access denied
		// instead of contending. The other platforms exercise this race.
		t.Skip("Windows refuses to replace a file readers hold open")
	}
	const (
		key          = "pkg/thing-1.0.0.tgz"
		readers      = 4
		readsPerRead = 500
	)
	dir := t.TempDir()
	ctx := context.Background()

	b, err := OpenBucket(ctx, fileURLFromPath(dir))
	if err != nil {
		t.Fatalf("OpenBucket failed: %v", err)
	}
	defer func() { _ = b.Close() }()

	payload := strings.Repeat("x", 4096)
	if _, _, err := b.Store(ctx, key, strings.NewReader(payload)); err != nil {
		t.Fatalf("seeding Store failed: %v", err)
	}

	// The writer reports how it ended: a Store failure would otherwise stop
	// the writes silently and let zero read failures pass for a test that
	// never contended anything.
	done := make(chan struct{})
	var writers sync.WaitGroup
	var writes int
	var writeErr error
	writers.Add(1)
	go func() {
		defer writers.Done()
		for {
			select {
			case <-done:
				return
			default:
			}
			if _, _, err := b.Store(ctx, key, strings.NewReader(payload)); err != nil {
				writeErr = err
				return
			}
			writes++
		}
	}()

	var failures atomic.Int64
	var reading sync.WaitGroup
	for range readers {
		reading.Add(1)
		go func() {
			defer reading.Done()
			for range readsPerRead {
				r, err := b.Open(ctx, key)
				if err != nil {
					failures.Add(1)
					continue
				}
				if _, err := io.Copy(io.Discard, r); err != nil {
					failures.Add(1)
				}
				_ = r.Close()
			}
		}()
	}
	reading.Wait()
	close(done)
	writers.Wait()

	if writeErr != nil {
		t.Fatalf("writer stopped early: %v", writeErr)
	}
	if writes == 0 {
		t.Fatal("no write completed, so the reads were never contended")
	}
	if got := failures.Load(); got != 0 {
		t.Errorf("%d of %d reads failed while one writer rewrote the same key, want 0", got, readers*readsPerRead)
	}
}

// seedLegacySidecar stores key through a bucket that still writes sidecars, as
// an earlier version did, and returns the path fileblob actually used. It is
// discovered rather than assumed, so callers test the real mapping.
func seedLegacySidecar(t *testing.T, dir, key, payload string) string {
	t.Helper()
	ctx := context.Background()

	legacy, err := blob.OpenBucket(ctx, fileURLFromPath(dir)+"?no_tmp_dir=true")
	if err != nil {
		t.Fatalf("opening legacy bucket: %v", err)
	}
	if err := legacy.WriteAll(ctx, key, []byte(payload), nil); err != nil {
		t.Fatalf("legacy WriteAll: %v", err)
	}
	if err := legacy.Close(); err != nil {
		t.Fatalf("closing legacy bucket: %v", err)
	}

	var found []string
	walkErr := filepath.WalkDir(dir, func(path string, d fs.DirEntry, err error) error {
		if err != nil {
			return err
		}
		if !d.IsDir() && strings.HasSuffix(path, ".attrs") {
			found = append(found, path)
		}
		return nil
	})
	if walkErr != nil {
		t.Fatalf("walking %s: %v", dir, walkErr)
	}
	if len(found) != 1 {
		t.Fatalf("got sidecars %v, want exactly one", found)
	}
	return found[0]
}

// An interrupted setAttrs leaves a partial sidecar that fails every read of the
// key, and nothing rewrites one now, so a store has to clear it.
//
// One key per storage path the proxy builds: ArtifactPath across ecosystems,
// metadata blobs, and the Gradle build cache. Scoped npm names, Go's "!" case
// escaping and the ":" in OCI digests and Debian epochs are the characters
// most likely to part fileblob's mapping from a plain path join.
func TestStoreClearsLegacyAttrsSidecar(t *testing.T) {
	keys := []string{
		"npm/@babel/core/7.24.0/core-7.24.0.tgz",
		"maven/org.apache.commons/commons-lang3/3.14.0/commons-lang3-3.14.0.jar",
		"golang/github.com/!burnt!sushi/toml/v1.3.2/v1.3.2.zip",
		"oci/library/nginx/sha256:abc123def456/manifest",
		"debian/tzdata/1:2024a-1/tzdata_2024a-1_all.deb",
		"pypi/requests/2.31.0/requests-2.31.0-py3-none-any.whl",
		"cargo/serde/1.0.197/serde-1.0.197.crate",
		"julia/Example/a1b2c3/a1b2c3.tar.gz",
		"conda/numpy/1.26.4/numpy-1.26.4-py311.conda",
		"_metadata/npm/@babel/core/metadata",
		"_gradle/http-build-cache/0a1b2c3d4e5f",
		// Keys fileblob escapes on every platform.
		"npm/pkg//1.0.0/x.tgz",
		"npm/pkg/../1.0.0/x.tgz",
	}

	for _, key := range keys {
		t.Run(key, func(t *testing.T) {
			assertStoreClearsSidecar(t, key)
		})
	}
}

func assertStoreClearsSidecar(t *testing.T, key string) {
	t.Helper()
	const payload = "payload"
	ctx := context.Background()
	dir := t.TempDir()

	sidecar := seedLegacySidecar(t, dir, key, payload)
	if err := os.WriteFile(sidecar, []byte(`{"user.content_type":"appl`), 0o600); err != nil {
		t.Fatalf("corrupting sidecar: %v", err)
	}

	b := openFileBlob(t, dir)
	reader, err := b.bucket.NewReader(ctx, key, nil)
	if err == nil {
		_ = reader.Close()
		t.Fatal("corrupt sidecar did not fail the read, so it is not the file fileblob reads for this key")
	}

	derived := b.legacySidecarPath(key)
	if _, _, err := b.Store(ctx, key, strings.NewReader(payload)); err != nil {
		t.Fatalf("Store failed: %v", err)
	}

	if derived == "" {
		t.Fatalf("legacySidecarPath declined %q, but fileblob wrote %q", key, sidecar)
	}
	if derived != sidecar {
		t.Fatalf("derived %q, but fileblob wrote %q", derived, sidecar)
	}
	if _, err := os.Stat(sidecar); !os.IsNotExist(err) {
		t.Errorf("sidecar still present after Store, stat err = %v", err)
	}
	assertReadsBack(t, b, key, payload)
}

func assertReadsBack(t *testing.T, b *Blob, key, want string) {
	t.Helper()
	r, err := b.Open(context.Background(), key)
	if err != nil {
		t.Fatalf("read still failing after Store cleared the sidecar: %v", err)
	}
	defer func() { _ = r.Close() }()
	got, err := io.ReadAll(r)
	if err != nil {
		t.Fatalf("ReadAll failed: %v", err)
	}
	if string(got) != want {
		t.Errorf("got %q, want %q", got, want)
	}
}

func openFileBlob(t *testing.T, dir string) *Blob {
	t.Helper()
	s, err := OpenBucket(context.Background(), fileURLFromPath(dir))
	if err != nil {
		t.Fatalf("OpenBucket failed: %v", err)
	}
	t.Cleanup(func() { _ = s.Close() })
	b, ok := s.(*Blob)
	if !ok {
		t.Fatalf("got %T, want *Blob", s)
	}
	return b
}

// fileblob escapes a non-local key in a way this cannot reproduce, and one
// holding ".." resolves outside the cache directory. Removal declines both
// rather than delete the wrong file.
func TestLegacySidecarPathEscapesLikeFileblob(t *testing.T) {
	root := filepath.FromSlash("/var/cache/proxy")
	b := &Blob{fileRoot: root}
	windows := runtime.GOOS == osWindows

	for _, tc := range []struct{ key, unix, windows string }{
		{"npm/pkg/1.0.0/x.tgz", "npm/pkg/1.0.0/x.tgz", "npm/pkg/1.0.0/x.tgz"},
		{"npm/pkg//1.0.0/x.tgz", "npm/pkg/__0x2f__1.0.0/x.tgz", "npm/pkg/__0x2f__1.0.0/x.tgz"},
		{"npm/pkg/../../etc/passwd", "npm/pkg/..__0x2f__..__0x2f__etc/passwd", "npm/pkg/..__0x2f__..__0x2f__etc/passwd"},
		{"npm/pkg/1.0.0/", "npm/pkg/1.0.0__0x2f__", "npm/pkg/1.0.0__0x2f__"},
		{"npm/a\x01b", "npm/a__0x1__b", "npm/a__0x1__b"},
		{"oci/nginx/sha256:abc/manifest", "oci/nginx/sha256:abc/manifest", "oci/nginx/sha256__0x3a__abc/manifest"},
		{"debian/tzdata/1:2024a-1/x.deb", "debian/tzdata/1:2024a-1/x.deb", "debian/tzdata/1__0x3a__2024a-1/x.deb"},
		{`npm/a\b`, `npm/a\b`, "npm/a__0x5c__b"},
	} {
		want := tc.unix
		if windows {
			want = tc.windows
		}
		want = filepath.Join(root, filepath.FromSlash(want)) + attrsExt
		if got := b.legacySidecarPath(tc.key); got != want {
			t.Errorf("legacySidecarPath(%q) = %q, want %q", tc.key, got, want)
		}
	}
}

func TestLegacySidecarPathDeclinesNonLocalKeys(t *testing.T) {
	b := &Blob{fileRoot: filepath.FromSlash("/var/cache/proxy")}

	for _, key := range []string{"", ".", "..", "/etc/passwd"} {
		if got := b.legacySidecarPath(key); got != "" {
			t.Errorf("legacySidecarPath(%q) = %q, want \"\"", key, got)
		}
	}
}

// Cloud backends have no local directory, so nothing is removed for them.
func TestLegacySidecarPathEmptyForCloudBackends(t *testing.T) {
	b := &Blob{}
	if got := b.legacySidecarPath("npm/pkg/1.0.0/x.tgz"); got != "" {
		t.Errorf("legacySidecarPath = %q, want \"\" when there is no file root", got)
	}
}

// Cleanup that cannot complete must not fail the write. A non-empty directory
// at the sidecar path makes os.Remove fail with something other than not-exist
// on every platform, which is what a Windows sharing violation would look like
// here.
func TestStoreSucceedsWhenSidecarCannotBeRemoved(t *testing.T) {
	const key = "npm/pkg/1.0.0/pkg-1.0.0.tgz"
	const payload = "payload"
	dir := t.TempDir()
	ctx := context.Background()

	b := openFileBlob(t, dir)

	sidecar := filepath.Join(dir, filepath.FromSlash(key)) + ".attrs"
	if err := os.MkdirAll(filepath.Join(sidecar, "blocker"), 0o750); err != nil {
		t.Fatalf("seeding an unremovable sidecar: %v", err)
	}
	if err := os.Remove(sidecar); err == nil {
		t.Fatal("sidecar path was removable, so the test proves nothing")
	}

	if _, _, err := b.Store(ctx, key, strings.NewReader(payload)); err != nil {
		t.Errorf("Store failed because cleanup could not complete: %v", err)
	}
}

type failingReader struct {
	data []byte
	err  error
}

func (r *failingReader) Read(p []byte) (int, error) {
	if len(r.data) == 0 {
		return 0, r.err
	}
	n := copy(p, r.data)
	r.data = r.data[n:]
	return n, nil
}

// A download that breaks off midway must not leave its first half behind as
// an object, or every failed fetch adds an orphan nothing ever deletes.
func TestBlobStoreFailedCopyLeavesNoObject(t *testing.T) {
	b := createTestBlob(t)
	ctx := context.Background()
	readErr := errors.New("connection reset")

	_, _, err := b.Store(ctx, "npm/lodash/4.17.21/lodash.tgz", &failingReader{
		data: []byte(strings.Repeat("partial", 1024)),
		err:  readErr,
	})
	if !errors.Is(err, readErr) {
		t.Fatalf("Store error = %v, want %v", err, readErr)
	}

	exists, err := b.Exists(ctx, "npm/lodash/4.17.21/lodash.tgz")
	if err != nil {
		t.Fatalf("Exists failed: %v", err)
	}
	if exists {
		t.Error("failed Store left a partial object behind")
	}
}