drive: fetch metadata permissions concurrently - fixes #9682

Before this change permission metadata fetches were serialized by the permission cache mutex.

backend/drive/drive_internal_test.go

@@ -8,10 +8,14 @@ import (
 	"fmt"
 	"io"
 	"mime"
+	"net/http"
+	"net/http/httptest"
 	"os"
 	"path"
 	"path/filepath"
 	"strings"
+	stdsync "sync"
+	"sync/atomic"
 	"testing"
 	"time"
 
@@ -25,13 +29,113 @@ import (
 	"github.com/rclone/rclone/fstest"
 	"github.com/rclone/rclone/fstest/fstests"
 	"github.com/rclone/rclone/lib/dircache"
+	"github.com/rclone/rclone/lib/pacer"
 	"github.com/rclone/rclone/lib/random"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	"google.golang.org/api/drive/v3"
 	"google.golang.org/api/googleapi"
+	"google.golang.org/api/option"
 )
 
+func newPermissionTestFs(t *testing.T, handler http.Handler) *Fs {
+	t.Helper()
+	server := httptest.NewServer(handler)
+	t.Cleanup(server.Close)
+	service, err := drive.NewService(
+		context.Background(),
+		option.WithHTTPClient(server.Client()),
+		option.WithEndpoint(server.URL+"/"),
+	)
+	require.NoError(t, err)
+	return &Fs{
+		svc:           service,
+		pacer:         fs.NewPacer(context.Background(), pacer.NewGoogleDrive()),
+		permissionsMu: new(stdsync.Mutex),
+		permissions:   make(map[string]*drive.Permission),
+	}
+}
+
+func TestInternalGetPermissionConcurrent(t *testing.T) {
+	const requestCount = 2
+	arrived := make(chan string, requestCount)
+	release := make(chan struct{})
+	var releaseOnce stdsync.Once
+	releaseRequests := func() {
+		releaseOnce.Do(func() { close(release) })
+	}
+	defer releaseRequests()
+
+	f := newPermissionTestFs(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		permissionID := path.Base(r.URL.Path)
+		arrived <- permissionID
+		<-release
+		w.Header().Set("Content-Type", "application/json")
+		_, err := fmt.Fprintf(w, `{"id":%q,"role":"reader","type":"user"}`, permissionID)
+		assert.NoError(t, err)
+	}))
+
+	type permissionResult struct {
+		permission *drive.Permission
+		inherited  bool
+		err        error
+	}
+	results := make(chan permissionResult, requestCount)
+	permissionIDs := []string{"permission-1", "permission-2"}
+	for _, permissionID := range permissionIDs {
+		go func() {
+			permission, inherited, err := f.getPermission(context.Background(), "file-id", permissionID, false)
+			results <- permissionResult{permission: permission, inherited: inherited, err: err}
+		}()
+	}
+
+	arrivedIDs := make(map[string]bool, requestCount)
+	timer := time.NewTimer(time.Second)
+	defer timer.Stop()
+	for len(arrivedIDs) < requestCount {
+		select {
+		case permissionID := <-arrived:
+			arrivedIDs[permissionID] = true
+		case <-timer.C:
+			releaseRequests()
+			t.Fatalf("timed out waiting for concurrent permission requests: got %d of %d", len(arrivedIDs), requestCount)
+		}
+	}
+	releaseRequests()
+
+	assert.Equal(t, map[string]bool{"permission-1": true, "permission-2": true}, arrivedIDs)
+	returnedIDs := make(map[string]bool, requestCount)
+	for range requestCount {
+		result := <-results
+		require.NoError(t, result.err)
+		require.NotNil(t, result.permission)
+		assert.False(t, result.inherited)
+		returnedIDs[result.permission.Id] = true
+	}
+	assert.Equal(t, arrivedIDs, returnedIDs)
+}
+
+func TestInternalGetPermissionCacheHit(t *testing.T) {
+	var requests atomic.Int32
+	f := newPermissionTestFs(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		requests.Add(1)
+		permissionID := path.Base(r.URL.Path)
+		w.Header().Set("Content-Type", "application/json")
+		_, err := fmt.Fprintf(w, `{"id":%q,"role":"reader","type":"user"}`, permissionID)
+		assert.NoError(t, err)
+	}))
+
+	first, inherited, err := f.getPermission(context.Background(), "file-id", "permission-1", true)
+	require.NoError(t, err)
+	assert.False(t, inherited)
+	second, inherited, err := f.getPermission(context.Background(), "file-id", "permission-1", true)
+	require.NoError(t, err)
+	assert.False(t, inherited)
+
+	assert.Same(t, first, second)
+	assert.Equal(t, int32(1), requests.Load())
+}
+
 func TestDriveScopes(t *testing.T) {
 	for _, test := range []struct {
 		in       string

backend/drive/metadata.go

@@ -106,10 +106,10 @@ var permissionsFields = googleapi.Field(strings.Join([]string{
 
 // getPermission returns permissions for the fileID and permissionID passed in
 func (f *Fs) getPermission(ctx context.Context, fileID, permissionID string, useCache bool) (perm *drive.Permission, inherited bool, err error) {
-	f.permissionsMu.Lock()
-	defer f.permissionsMu.Unlock()
 	if useCache {
+		f.permissionsMu.Lock()
 		perm = f.permissions[permissionID]
+		f.permissionsMu.Unlock()
 		if perm != nil {
 			return perm, false, nil
 		}
@@ -131,7 +131,9 @@ func (f *Fs) getPermission(ctx context.Context, fileID, permissionID string, use
 	cleanPermission(perm)
 
 	// cache the permission
+	f.permissionsMu.Lock()
 	f.permissions[permissionID] = perm
+	f.permissionsMu.Unlock()
 
 	return perm, inherited, err
 }