6abe390 test(blobs): run the blob contract on MinIO, which lists unfinished uploads by key · ermite

M docs/storage.md +3 -0


@@ -51,6 +51,9 @@ Downloads are relayed through ermite with repository authorization and Range
51	51		support: no redirects or presigned URLs. Multipart uploads use 8 MiB parts.
52	52		The daily sweep removes unreferenced blobs and unfinished uploads older than
53	53		24 hours. Release metadata with a missing blob is removed only after two sweeps.
	54	+	MinIO lists unfinished uploads only for an exact object key, so the sweep does
	55	+	not see them there; MinIO removes them itself (`stale_uploads_expiry`, 24 hours
	56	+	by default).
54	57
55	58		## Backend changes
56	59


@@ -51,6 +51,9 @@ Downloads are relayed through ermite with repository authorization and Range
51	support: no redirects or presigned URLs. Multipart uploads use 8 MiB parts.	51		support: no redirects or presigned URLs. Multipart uploads use 8 MiB parts.
52	The daily sweep removes unreferenced blobs and unfinished uploads older than	52		The daily sweep removes unreferenced blobs and unfinished uploads older than
53	24 hours. Release metadata with a missing blob is removed only after two sweeps.	53		24 hours. Release metadata with a missing blob is removed only after two sweeps.
		54	+	MinIO lists unfinished uploads only for an exact object key, so the sweep does
		55	+	not see them there; MinIO removes them itself (`stale_uploads_expiry`, 24 hours
		56	+	by default).
54		57
55	## Backend changes	58		## Backend changes
56		59

M internal/adapters/blobs3/integration_test.go +6 -1


@@ -23,6 +23,11 @@ func TestIntegration_Contract(t *testing.T) {
23	23		if region == "" {
24	24		region = "us-east-1"
25	25		}
	26	+	var opts []blobtest.Option
	27	+	// MinIO lists unfinished uploads by whole key only.
	28	+	if get("EXACT_KEY_UPLOAD_LIST") == "1" {
	29	+	opts = append(opts, blobtest.ExactKeyUploadList())
	30	+	}
26	31		blobtest.Run(t, func(t *testing.T) ports.BlobStore {
27	32		b := make([]byte, 16)
28	33		_, e := rand.Read(b)

@@ -35,5 +40,5 @@ func TestIntegration_Contract(t *testing.T) {
35	40		require.NoError(t, s.ListMultipart(ctx, "", func(i ports.MultipartInfo) error { return s.AbortMultipart(ctx, i.Key, i.ID) }))
36	41		})
37	42		return s
38		-	})
	43	+	}, opts...)
39	44		}

 	if region == "" {
 		region = "us-east-1"
 	}
+	var opts []blobtest.Option
+	// MinIO lists unfinished uploads by whole key only.
+	if get("EXACT_KEY_UPLOAD_LIST") == "1" {
+		opts = append(opts, blobtest.ExactKeyUploadList())
+	}
 	blobtest.Run(t, func(t *testing.T) ports.BlobStore {
 		b := make([]byte, 16)
 		_, e := rand.Read(b)
 			require.NoError(t, s.ListMultipart(ctx, "", func(i ports.MultipartInfo) error { return s.AbortMultipart(ctx, i.Key, i.ID) }))
 		})
 		return s
-	})
+	}, opts...)
 }

M internal/adapters/blobtest/suite.go +26 -4


@@ -18,9 +18,31 @@ import (
18	18		"github.com/stretchr/testify/require"
19	19		)
20	20
	21	+	// Option adapts Run to a provider quirk.
	22	+	type Option func(*options)
	23	+
	24	+	type options struct{ exactKeyUploadList bool }
	25	+
	26	+	// ExactKeyUploadList is for providers (MinIO) that list unfinished multipart
	27	+	// uploads only when the prefix is a whole object key: the suite then lists by
	28	+	// key instead of by directory prefix.
	29	+	func ExactKeyUploadList() Option { return func(o *options) { o.exactKeyUploadList = true } }
	30	+
21	31		// Run checks the blob contract against a fresh store for each subtest.
22		-	func Run(t testing.T, newStore func(testing.T) ports.BlobStore) {
	32	+	func Run(t testing.T, newStore func(testing.T) ports.BlobStore, opts ...Option) {
23	33		ctx := context.Background()
	34	+	var o options
	35	+	for _, opt := range opts {
	36	+	opt(&o)
	37	+	}
	38	+	// uploadPrefix is the ListMultipart prefix for key: dir, or key itself
	39	+	// on a provider that lists by whole key only.
	40	+	uploadPrefix := func(dir, key string) string {
	41	+	if o.exactKeyUploadList {
	42	+	return key
	43	+	}
	44	+	return dir
	45	+	}
24	46		t.Run("roundtrip", func(t *testing.T) {
25	47		s := newStore(t)
26	48		require.NoError(t, s.Check(ctx))

@@ -136,7 +158,7 @@ func Run(t testing.T, newStore func(testing.T) ports.BlobStore) {
136	158		require.NoError(t, e)
137	159		require.NoError(t, u.UploadPart(ctx, 1, strings.NewReader("x"), 1))
138	160		var uploads []ports.MultipartInfo
139		-	require.NoError(t, s.ListMultipart(ctx, "ci/", func(i ports.MultipartInfo) error { uploads = append(uploads, i); return nil }))
	161	+	require.NoError(t, s.ListMultipart(ctx, uploadPrefix("ci/", "ci/abort"), func(i ports.MultipartInfo) error { uploads = append(uploads, i); return nil }))
140	162		require.Len(t, uploads, 1)
141	163		require.Equal(t, u.ID(), uploads[0].ID)
142	164		require.Equal(t, "ci/abort", uploads[0].Key)

@@ -149,7 +171,7 @@ func Run(t testing.T, newStore func(testing.T) ports.BlobStore) {
149	171		_, e = s.Stat(ctx, "ci/abort")
150	172		require.ErrorIs(t, e, domain.ErrNotFound)
151	173		uploads = nil
152		-	require.NoError(t, s.ListMultipart(ctx, "", func(i ports.MultipartInfo) error { uploads = append(uploads, i); return nil }))
	174	+	require.NoError(t, s.ListMultipart(ctx, uploadPrefix("", "ci/abort"), func(i ports.MultipartInfo) error { uploads = append(uploads, i); return nil }))
153	175		require.Empty(t, uploads)
154	176		}
155	177		})

@@ -199,7 +221,7 @@ func Run(t testing.T, newStore func(testing.T) ports.BlobStore) {
199	221		_, e = s.Stat(ctx, "ci/badlarge")
200	222		require.ErrorIs(t, e, domain.ErrNotFound)
201	223		}
202		-	require.NoError(t, s.ListMultipart(ctx, "", func(ports.MultipartInfo) error { t.Error("unfinished upload after Put"); return nil }))
	224	+	require.NoError(t, s.ListMultipart(ctx, uploadPrefix("", "ci/badlarge"), func(ports.MultipartInfo) error { t.Error("unfinished upload after Put"); return nil }))
203	225		})
204	226
205	227		t.Run("cancel_put", func(t *testing.T) {

 	"github.com/stretchr/testify/require"
 )
+// Option adapts Run to a provider quirk.
+type Option func(*options)
+type options struct{ exactKeyUploadList bool }
+// ExactKeyUploadList is for providers (MinIO) that list unfinished multipart
+// uploads only when the prefix is a whole object key: the suite then lists by
+// key instead of by directory prefix.
+func ExactKeyUploadList() Option { return func(o *options) { o.exactKeyUploadList = true } }
 // Run checks the blob contract against a fresh store for each subtest.
-func Run(t *testing.T, newStore func(*testing.T) ports.BlobStore) {
+func Run(t *testing.T, newStore func(*testing.T) ports.BlobStore, opts ...Option) {
 	ctx := context.Background()
+	var o options
+	for _, opt := range opts {
+		opt(&o)
+	}
+	// uploadPrefix is the ListMultipart prefix for key: dir, or key itself
+	// on a provider that lists by whole key only.
+	uploadPrefix := func(dir, key string) string {
+		if o.exactKeyUploadList {
+			return key
+		}
+		return dir
+	}
 	t.Run("roundtrip", func(t *testing.T) {
 		s := newStore(t)
 		require.NoError(t, s.Check(ctx))
 			require.NoError(t, e)
 			require.NoError(t, u.UploadPart(ctx, 1, strings.NewReader("x"), 1))
 			var uploads []ports.MultipartInfo
-			require.NoError(t, s.ListMultipart(ctx, "ci/", func(i ports.MultipartInfo) error { uploads = append(uploads, i); return nil }))
+			require.NoError(t, s.ListMultipart(ctx, uploadPrefix("ci/", "ci/abort"), func(i ports.MultipartInfo) error { uploads = append(uploads, i); return nil }))
 			require.Len(t, uploads, 1)
 			require.Equal(t, u.ID(), uploads[0].ID)
 			require.Equal(t, "ci/abort", uploads[0].Key)
 			_, e = s.Stat(ctx, "ci/abort")
 			require.ErrorIs(t, e, domain.ErrNotFound)
 			uploads = nil
-			require.NoError(t, s.ListMultipart(ctx, "", func(i ports.MultipartInfo) error { uploads = append(uploads, i); return nil }))
+			require.NoError(t, s.ListMultipart(ctx, uploadPrefix("", "ci/abort"), func(i ports.MultipartInfo) error { uploads = append(uploads, i); return nil }))
 			require.Empty(t, uploads)
 		}
 	})
 			_, e = s.Stat(ctx, "ci/badlarge")
 			require.ErrorIs(t, e, domain.ErrNotFound)
 		}
-		require.NoError(t, s.ListMultipart(ctx, "", func(ports.MultipartInfo) error { t.Error("unfinished upload after Put"); return nil }))
+		require.NoError(t, s.ListMultipart(ctx, uploadPrefix("", "ci/badlarge"), func(ports.MultipartInfo) error { t.Error("unfinished upload after Put"); return nil }))
 	})
 	t.Run("cancel_put", func(t *testing.T) {