Separate matchers as a bucket filter (#3229)

Adds a new read bucket Filter that works exclusively on Matchers. Buckets that use Matchers as Mappers have been split as a Mapper and Filter bucket. The method MapPrefix on Mappers has been removed. Matchers no longer implement Mappers.

Author: emcfarlane

private/buf/bufformat/bufformat.go

@@ -43,7 +43,7 @@ func FormatModuleSet(ctx context.Context, moduleSet bufmodule.ModuleSet) (_ stor
 // FormatBucket formats the .proto files in the bucket and returns a new bucket with the formatted files.
 func FormatBucket(ctx context.Context, bucket storage.ReadBucket) (_ storage.ReadBucket, retErr error) {
 	readWriteBucket := storagemem.NewReadWriteBucket()
-	paths, err := storage.AllPaths(ctx, storage.MapReadBucket(bucket, storage.MatchPathExt(".proto")), "")
+	paths, err := storage.AllPaths(ctx, storage.FilterReadBucket(bucket, storage.MatchPathExt(".proto")), "")
 	if err != nil {
 		return nil, err
 	}

private/buf/bufprotoc/bufprotoc.go

@@ -62,7 +62,7 @@ func NewModuleSetForProtoc(
 		}
 		// need to do match extension here
 		// https://github.com/bufbuild/buf/issues/113
-		rootBuckets = append(rootBuckets, storage.MapReadBucket(rootBucket, storage.MatchPathExt(".proto")))
+		rootBuckets = append(rootBuckets, storage.FilterReadBucket(rootBucket, storage.MatchPathExt(".proto")))
 	}
 	targetPaths, err := slicesext.MapError(
 		absFilePaths,

private/buf/bufworkspace/workspace_targeting.go

@@ -527,10 +527,12 @@ func getMappedModuleBucketAndModuleTargeting(
 	for root, excludes := range rootToExcludes {
 		// Roots only applies to .proto files.
 		mappers := []storage.Mapper{
+			storage.MapOnPrefix(root),
+		}
+		matchers := []storage.Matcher{
 			// need to do match extension here
 			// https://github.com/bufbuild/buf/issues/113
 			storage.MatchPathExt(".proto"),
-			storage.MapOnPrefix(root),
 		}
 		if len(excludes) != 0 {
 			var notOrMatchers []storage.Matcher
@@ -540,8 +542,8 @@ func getMappedModuleBucketAndModuleTargeting(
 					storage.MatchPathContained(exclude),
 				)
 			}
-			mappers = append(
-				mappers,
+			matchers = append(
+				matchers,
 				storage.MatchNot(
 					storage.MatchOr(
 						notOrMatchers...,
@@ -551,9 +553,12 @@ func getMappedModuleBucketAndModuleTargeting(
 		}
 		rootBuckets = append(
 			rootBuckets,
-			storage.MapReadBucket(
-				moduleBucket,
-				mappers...,
+			storage.FilterReadBucket(
+				storage.MapReadBucket(
+					moduleBucket,
+					mappers...,
+				),
+				matchers...,
 			),
 		)
 	}

private/bufpkg/bufmodule/digest.go

@@ -233,7 +233,7 @@ func getB4Digest(
 		ctx,
 		// This is extreme defensive programming. We've gone out of our way to make sure
 		// that the bucket is already filtered, but it's just too important to mess up here.
-		storage.MapReadBucket(bucketWithStorageMatcherApplied, getStorageMatcher(ctx, bucketWithStorageMatcherApplied)),
+		storage.FilterReadBucket(bucketWithStorageMatcherApplied, getStorageMatcher(ctx, bucketWithStorageMatcherApplied)),
 		"",
 		func(readObject storage.ReadObject) error {
 			digest, err := bufcas.NewDigestForContent(readObject)
@@ -373,7 +373,7 @@ func getFilesDigestForB5Digest(
 		ctx,
 		// This is extreme defensive programming. We've gone out of our way to make sure
 		// that the bucket is already filtered, but it's just too important to mess up here.
-		storage.MapReadBucket(bucketWithStorageMatcherApplied, getStorageMatcher(ctx, bucketWithStorageMatcherApplied)),
+		storage.FilterReadBucket(bucketWithStorageMatcherApplied, getStorageMatcher(ctx, bucketWithStorageMatcherApplied)),
 		"",
 		func(readObject storage.ReadObject) error {
 			digest, err := bufcas.NewDigestForContent(readObject)

private/bufpkg/bufmodule/paths.go

@@ -90,7 +90,7 @@ func getSyncOnceValuesGetBucketWithStorageMatcherApplied(
 			if err != nil {
 				return nil, err
 			}
-			return storage.MapReadBucket(bucket, getStorageMatcher(ctx, bucket)), nil
+			return storage.FilterReadBucket(bucket, getStorageMatcher(ctx, bucket)), nil
 		},
 	)
 }

private/pkg/git/cloner.go

@@ -240,8 +240,8 @@ func (c *cloner) CloneToBucket(
 		return err
 	}
 	var readBucket storage.ReadBucket = tmpReadWriteBucket
-	if options.Mapper != nil {
-		readBucket = storage.MapReadBucket(readBucket, options.Mapper)
+	if options.Matcher != nil {
+		readBucket = storage.FilterReadBucket(readBucket, options.Matcher)
 	}
 	_, err = storage.Copy(ctx, readBucket, writeBucket)
 	return err

private/pkg/git/git.go

@@ -95,7 +95,7 @@ type Cloner interface {
 
 // CloneToBucketOptions are options for Clone.
 type CloneToBucketOptions struct {
-	Mapper            storage.Mapper
+	Matcher           storage.Matcher
 	Name              Name
 	RecurseSubmodules bool
 }

private/pkg/git/git_test.go

@@ -242,7 +242,7 @@ func readBucketForName(ctx context.Context, t *testing.T, runner command.Runner,
 		depth,
 		readWriteBucket,
 		CloneToBucketOptions{
-			Mapper:            storage.MatchPathExt(".proto"),
+			Matcher:           storage.MatchPathExt(".proto"),
 			Name:              name,
 			RecurseSubmodules: recurseSubmodules,
 		},

private/pkg/protostat/protostatstorage/file_walker.go

@@ -28,7 +28,7 @@ type fileWalker struct {
 
 func newFileWalker(readBucket storage.ReadBucket) *fileWalker {
 	return &fileWalker{
-		readBucket: storage.MapReadBucket(
+		readBucket: storage.FilterReadBucket(
 			readBucket,
 			storage.MatchPathExt(".proto"),
 		),

private/pkg/storage/filter.go

@@ -0,0 +1,108 @@
+// Copyright 2020-2024 Buf Technologies, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package storage
+
+import (
+	"context"
+	"io/fs"
+
+	"github.com/bufbuild/buf/private/pkg/normalpath"
+)
+
+// FilterReadBucket filters the ReadBucket.
+//
+// If the Matchers are empty, the original ReadBucket is returned.
+// If there is more than one Matcher, the Matchers are anded together.
+func FilterReadBucket(readBucket ReadBucket, matchers ...Matcher) ReadBucket {
+	if len(matchers) == 0 {
+		return readBucket
+	}
+	return newFilterReadBucketCloser(readBucket, nil, MatchAnd(matchers...))
+}
+
+// FilterReadBucketCloser filters the ReadBucketCloser.
+//
+// If the Matchers are empty, the original ReadBucketCloser is returned.
+// If there is more than one Matcher, the Matchers are anded together.
+func FilterReadBucketCloser(readBucketCloser ReadBucketCloser, matchers ...Matcher) ReadBucketCloser {
+	if len(matchers) == 0 {
+		return readBucketCloser
+	}
+	return newFilterReadBucketCloser(readBucketCloser, readBucketCloser.Close, MatchAnd(matchers...))
+}
+
+type filterReadBucketCloser struct {
+	delegate  ReadBucket
+	closeFunc func() error
+	matcher   Matcher
+}
+
+func newFilterReadBucketCloser(
+	delegate ReadBucket,
+	closeFunc func() error,
+	matcher Matcher,
+) *filterReadBucketCloser {
+	return &filterReadBucketCloser{
+		delegate:  delegate,
+		closeFunc: closeFunc,
+		matcher:   matcher,
+	}
+}
+
+func (r *filterReadBucketCloser) Get(ctx context.Context, path string) (ReadObjectCloser, error) {
+	path, err := normalpath.NormalizeAndValidate(path)
+	if err != nil {
+		return nil, err
+	}
+	if !r.matcher.MatchPath(path) {
+		return nil, &fs.PathError{Op: "read", Path: path, Err: fs.ErrNotExist}
+	}
+	return r.delegate.Get(ctx, path)
+}
+
+func (r *filterReadBucketCloser) Stat(ctx context.Context, path string) (ObjectInfo, error) {
+	path, err := normalpath.NormalizeAndValidate(path)
+	if err != nil {
+		return nil, err
+	}
+	if !r.matcher.MatchPath(path) {
+		return nil, &fs.PathError{Op: "read", Path: path, Err: fs.ErrNotExist}
+	}
+	return r.delegate.Stat(ctx, path)
+}
+
+func (r *filterReadBucketCloser) Walk(ctx context.Context, prefix string, f func(ObjectInfo) error) error {
+	prefix, err := normalpath.NormalizeAndValidate(prefix)
+	if err != nil {
+		return err
+	}
+	return r.delegate.Walk(
+		ctx,
+		prefix,
+		func(objectInfo ObjectInfo) error {
+			if !r.matcher.MatchPath(objectInfo.Path()) {
+				return nil
+			}
+			return f(objectInfo)
+		},
+	)
+}
+
+func (r *filterReadBucketCloser) Close() error {
+	if r.closeFunc != nil {
+		return r.closeFunc()
+	}
+	return nil
+}