Merge pull request #2321 from versity/ben/conditional-lock-fixes

fix: harden posix conditional publish locking

backend/posix/objlock.go

@@ -38,13 +38,15 @@ import (
 // contention within one process is resolved cheaply and each process presents
 // at most one waiter to the filesystem lock.
 //
-// Lock identity: bucket/.sgwtmp/objlock/<shard>, where shard is the first
-// byte of sha256(object key) rendered as two hex characters. Hashing the key
-// gives a stable, traversal-safe, fixed-length name; sharding (256 slots per
-// bucket) keeps the number of lock files bounded while still letting writes
-// to unrelated keys proceed concurrently in the common case. Lock files are
-// empty, created on demand, and never unlinked: unlinking an flock file opens
-// a classic race where a waiter holds a lock on an unlinked inode while a new
+// Lock identity: <root>/.vgwlocks/<bucket-hash>/<shard>, where shard is the
+// first byte of sha256(object key) rendered as two hex characters. Hashing the
+// bucket and key gives stable, traversal-safe, fixed-length names; sharding
+// (256 slots per bucket) keeps the number of lock files bounded while still
+// letting writes to unrelated keys proceed concurrently in the common case.
+// Lock files are outside the bucket tree so Windows bucket deletion cannot
+// fail merely because an active request has the lock file open. They are empty,
+// created on demand, and never unlinked: unlinking an flock file opens a
+// classic race where a waiter holds a lock on an unlinked inode while a new
 // file takes its place, which is unsafe to detect reliably on NFS due to
 // attribute caching.
 //
@@ -63,8 +65,8 @@ import (
 // process-local exclusion and logs a warning once.
 
 const (
-	// objLockDir is the per-bucket directory holding object publish lock files
-	objLockDir = MetaTmpDir + "/objlock"
+	// objLockDir is the root directory holding object publish lock files.
+	objLockDir = ".vgwlocks"
 	// objLockShards is the number of lock shards per bucket
 	objLockShards = 256
 )
@@ -83,12 +85,24 @@ func objLockShard(object string) uint8 {
 func (p *Posix) lockObjectPublish(ctx context.Context, bucket, object string) (func(), error) {
 	shard := objLockShard(object)
 
-	mu := &p.objLockMus[shard]
-	mu.Lock()
+	slot := p.objLockSlots[shard]
+	select {
+	case <-ctx.Done():
+		return nil, ctx.Err()
+	case <-slot:
+	}
+	releaseLocal := func() { slot <- struct{}{} }
+	if err := ctx.Err(); err != nil {
+		releaseLocal()
+		return nil, err
+	}
+	if p.forceNoObjLockFile {
+		return releaseLocal, nil
+	}
 
 	f, err := p.openObjLockFile(bucket, shard)
 	if err != nil {
-		mu.Unlock()
+		releaseLocal()
 		return nil, err
 	}
 
@@ -96,33 +110,47 @@ func (p *Posix) lockObjectPublish(ctx context.Context, bucket, object string) (f
 	if err != nil {
 		f.Close()
 		if ctx.Err() != nil {
-			mu.Unlock()
+			releaseLocal()
 			return nil, ctx.Err()
 		}
-		// The filesystem does not support advisory locking (e.g. NFS
-		// mounted with -o nolock). Fall back to process-local exclusion
-		// and warn once: conditional writes are then only atomic within
-		// this gateway process.
+		if !isAdvisoryLockUnsupported(err) {
+			releaseLocal()
+			return nil, fmt.Errorf("lock object publish: %w", err)
+		}
+		// The filesystem does not support advisory locking (e.g. NFS mounted
+		// with -o nolock). Fall back to process-local exclusion and warn once:
+		// conditional writes are then only atomic within this gateway process.
 		p.objLockWarn.Do(func() {
 			debuglogger.Logf("object lock file locking unavailable (%v): "+
 				"conditional write atomicity limited to this process", err)
 		})
-		return mu.Unlock, nil
+		return releaseLocal, nil
 	}
 
 	return func() {
 		// closing the file releases the advisory lock
 		f.Close()
-		mu.Unlock()
+		releaseLocal()
 	}, nil
 }
 
+func newObjLockSlots() [objLockShards]chan struct{} {
+	var slots [objLockShards]chan struct{}
+	for i := range slots {
+		slots[i] = make(chan struct{}, 1)
+		slots[i] <- struct{}{}
+	}
+	return slots
+}
+
 // openObjLockFile opens (creating as needed) the lock file for the shard in
 // the given bucket.
 func (p *Posix) openObjLockFile(bucket string, shard uint8) (*os.File, error) {
-	name := filepath.Join(bucket, objLockDir, fmt.Sprintf("%02x", shard))
+	bucketHash := sha256.Sum256([]byte(bucket))
+	lockDir := filepath.Join(p.rootdir, objLockDir, fmt.Sprintf("%x", bucketHash))
+	name := filepath.Join(lockDir, fmt.Sprintf("%02x", shard))
 
-	f, err := os.OpenFile(name, os.O_RDWR|os.O_CREATE, os.FileMode(defaultNewFilePerm))
+	f, err := os.OpenFile(name, os.O_RDWR|os.O_CREATE, defaultNewFilePerm)
 	if err == nil {
 		return f, nil
 	}
@@ -131,11 +159,11 @@ func (p *Posix) openObjLockFile(bucket string, shard uint8) (*os.File, error) {
 	}
 
 	// lock dir not created yet
-	err = backend.MkdirAll(filepath.Join(bucket, objLockDir), 0, 0, false, p.newDirPerm)
+	err = backend.MkdirAll(lockDir, 0, 0, false, p.newDirPerm)
 	if err != nil {
 		return nil, fmt.Errorf("make object lock dir: %w", err)
 	}
-	f, err = os.OpenFile(name, os.O_RDWR|os.O_CREATE, os.FileMode(defaultNewFilePerm))
+	f, err = os.OpenFile(name, os.O_RDWR|os.O_CREATE, defaultNewFilePerm)
 	if err != nil {
 		return nil, fmt.Errorf("open object lock file: %w", err)
 	}

backend/posix/objlock_unix.go

@@ -49,3 +49,8 @@ func lockFileExclusive(ctx context.Context, f *os.File) error {
 		backoff = min(backoff*2, objLockMaxBackoff)
 	}
 }
+
+func isAdvisoryLockUnsupported(err error) bool {
+	return errors.Is(err, unix.ENOTSUP) || errors.Is(err, unix.EOPNOTSUPP) ||
+		errors.Is(err, unix.ENOSYS)
+}

backend/posix/objlock_windows.go

@@ -50,3 +50,8 @@ func lockFileExclusive(ctx context.Context, f *os.File) error {
 		backoff = min(backoff*2, objLockMaxBackoff)
 	}
 }
+
+func isAdvisoryLockUnsupported(err error) bool {
+	return errors.Is(err, windows.ERROR_INVALID_FUNCTION) ||
+		errors.Is(err, windows.ERROR_NOT_SUPPORTED)
+}

backend/posix/posix.go

@@ -92,6 +92,10 @@ type Posix struct {
 	// support copy_file_range is mounted over NFSv4.2.
 	forceNoCopyFileRange bool
 
+	// forceNoObjLockFile disables the shared advisory lock file used for
+	// conditional object publishes. Process-local serialization remains active.
+	forceNoObjLockFile bool
+
 	// enableODirect is a flag to open object data files with O_DIRECT.
 	// This is best-effort and falls back to buffered I/O when unsupported.
 	enableODirect bool
@@ -132,11 +136,10 @@ type Posix struct {
 	//   multipart composite: "ALGO-<composite-checksum>-<part-count>"
 	dataIntegrityEtag bool
 
-	// objLockMus are process-local striped mutexes taken alongside the
-	// shared per-bucket lock files by lockObjectPublish so that in-process
-	// contention on an object's publish lock is resolved without filesystem
-	// lock thrash. See objlock.go.
-	objLockMus [objLockShards]sync.Mutex
+	// objLockSlots are process-local striped slots taken alongside the shared
+	// per-bucket lock files by lockObjectPublish. Channels make local lock
+	// acquisition cancelable while avoiding filesystem lock thrash.
+	objLockSlots [objLockShards]chan struct{}
 	// objLockWarn ensures the advisory-locking-unavailable warning is
 	// logged at most once
 	objLockWarn sync.Once
@@ -234,6 +237,10 @@ type PosixOpts struct {
 	ForceNoTmpFile bool
 	// ForceNoCopyFileRange disables the use of io.Copy for multipart uploads parts
 	ForceNoCopyFileRange bool
+	// ForceNoObjLockFile disables the shared advisory lock file used for
+	// conditional object publishes. Conditional writes are only atomic within a
+	// single gateway process when enabled.
+	ForceNoObjLockFile bool
 	// EnableODirect enables best-effort O_DIRECT for object data reads/writes.
 	// Disabled by default.
 	EnableODirect bool
@@ -273,26 +280,25 @@ type PosixOpts struct {
 
 func New(rootdir string, meta meta.MetadataStorer, opts PosixOpts) (*Posix, error) {
 	ioBufferSize := ioBufferSizeOrDefault(opts.IOBufferSize)
+	rootdirAbs, err := filepath.Abs(rootdir)
+	if err != nil {
+		return nil, fmt.Errorf("get absolute path of %v: %w", rootdir, err)
+	}
 
 	if opts.SideCarDir != "" && strings.HasPrefix(opts.SideCarDir, rootdir) {
 		return nil, fmt.Errorf("sidecar directory cannot be inside the gateway root directory")
 	}
 
-	err := os.Chdir(rootdir)
+	err = os.Chdir(rootdirAbs)
 	if err != nil {
 		return nil, fmt.Errorf("chdir %v: %w", rootdir, err)
 	}
 
-	f, err := os.Open(rootdir)
+	f, err := os.Open(rootdirAbs)
 	if err != nil {
 		return nil, fmt.Errorf("open %v: %w", rootdir, err)
 	}
 
-	rootdirAbs, err := filepath.Abs(rootdir)
-	if err != nil {
-		return nil, fmt.Errorf("get absolute path of %v: %w", rootdir, err)
-	}
-
 	var versioningdirAbs string
 	// Ensure the versioning directory isn't within the root directory
 	if opts.VersioningDir != "" {
@@ -345,7 +351,7 @@ func New(rootdir string, meta meta.MetadataStorer, opts PosixOpts) (*Posix, erro
 	return &Posix{
 		meta:                 meta,
 		rootfd:               f,
-		rootdir:              rootdir,
+		rootdir:              rootdirAbs,
 		euid:                 euid,
 		egid:                 egid,
 		chownuid:             opts.ChownUID,
@@ -356,6 +362,7 @@ func New(rootdir string, meta meta.MetadataStorer, opts PosixOpts) (*Posix, erro
 		newFilePerm:          newFilePerm,
 		forceNoTmpFile:       opts.ForceNoTmpFile,
 		forceNoCopyFileRange: opts.ForceNoCopyFileRange,
+		forceNoObjLockFile:   opts.ForceNoObjLockFile,
 		enableODirect:        opts.EnableODirect,
 		validateBucketName:   opts.ValidateBucketNames,
 		actionLimiter:        semaphore.NewWeighted(int64(concurrencyOrDefault(opts.Concurrency))),
@@ -367,6 +374,7 @@ func New(rootdir string, meta meta.MetadataStorer, opts PosixOpts) (*Posix, erro
 			return &b
 		}},
 		dataIntegrityEtag: opts.DataIntegrityEtag,
+		objLockSlots:      newObjLockSlots(),
 	}, nil
 }
 
@@ -602,6 +610,10 @@ func (p *Posix) ListBuckets(ctx context.Context, input s3response.ListBucketsInp
 }
 
 func (p *Posix) isBucketValid(bucket string) bool {
+	if bucket == objLockDir {
+		return false
+	}
+
 	if !p.validateBucketName {
 		return true
 	}
@@ -4253,17 +4265,9 @@ func (p *Posix) PutObjectWithPostFunc(ctx context.Context, po s3response.PutObje
 		return s3response.PutObjectOutput{}, s3err.GetAPIError(s3err.ErrExistingObjectIsDirectory)
 	}
 
-	// If versioning is enabled, first create the file object version.
-	// Conditional writes defer the snapshot until the precondition has been
-	// confirmed under the object publish lock, so that a losing conditional
-	// write does not create a spurious version snapshot.
+	// Version snapshots are taken under the object publish lock below. This
+	// keeps concurrent versioned overwrites ordered with their publications.
 	conditional := po.IfMatch != nil || po.IfNoneMatch != nil
-	if err == nil && !conditional {
-		verr := p.snapshotObjVersion(*po.Bucket, *po.Key, vStatus, acct)
-		if verr != nil {
-			return s3response.PutObjectOutput{}, verr
-		}
-	}
 	if isErrNameTooLong(err) {
 		return s3response.PutObjectOutput{}, s3err.GetKeyTooLongErr(int64(len(*po.Key)), 1024)
 	}
@@ -4386,12 +4390,13 @@ func (p *Posix) PutObjectWithPostFunc(ctx context.Context, po s3response.PutObje
 			return s3response.PutObjectOutput{}, err
 		}
 
-		// snapshot the object version that is about to be replaced, now
-		// that this writer is known to win the conditional race
-		verr := p.snapshotObjVersion(*po.Bucket, *po.Key, vStatus, acct)
-		if verr != nil {
-			return s3response.PutObjectOutput{}, verr
-		}
+	}
+
+	// Snapshot the object that is about to be replaced only after the
+	// conditional check has succeeded and while holding the publish lock.
+	verr := p.snapshotObjVersion(*po.Bucket, *po.Key, vStatus, acct)
+	if verr != nil {
+		return s3response.PutObjectOutput{}, verr
 	}
 
 	// Before finalizing the object creation remove
@@ -7201,6 +7206,10 @@ func listBucketFileInfos(bucketlinks bool) ([]fs.FileInfo, error) {
 
 	var fis []fs.FileInfo
 	for _, entry := range entries {
+		if entry.Name() == objLockDir {
+			continue
+		}
+
 		fi, err := entry.Info()
 		if err != nil {
 			continue

backend/posix/posix_conditional_put_test.go

@@ -27,6 +27,7 @@ import (
 	"strings"
 	"sync"
 	"testing"
+	"time"
 
 	"github.com/aws/aws-sdk-go-v2/aws"
 	"github.com/aws/aws-sdk-go-v2/service/s3"
@@ -79,6 +80,129 @@ func createTestBucket(t *testing.T, p *Posix, bucket string) {
 	}
 }
 
+func TestObjectPublishLockHonorsContextWhileWaiting(t *testing.T) {
+	p := newTestPosix(t, metaModes(t)["xattr"])
+	bucket := "testbucket"
+	createTestBucket(t, p, bucket)
+
+	shard := objLockShard("cancel-wait")
+	<-p.objLockSlots[shard]
+	defer func() { p.objLockSlots[shard] <- struct{}{} }()
+	if _, err := os.Stat(filepath.Join(bucket, objLockDir)); !errors.Is(err, fs.ErrNotExist) {
+		t.Fatalf("bucket contains publish lock directory: %v", err)
+	}
+
+	ctx, cancel := context.WithCancel(context.Background())
+	result := make(chan error, 1)
+	go func() {
+		unlock, err := p.lockObjectPublish(ctx, bucket, "cancel-wait")
+		if unlock != nil {
+			unlock()
+		}
+		result <- err
+	}()
+
+	select {
+	case err := <-result:
+		t.Fatalf("lockObjectPublish returned before cancellation: %v", err)
+	case <-time.After(10 * time.Millisecond):
+	}
+	cancel()
+
+	var err error
+	select {
+	case err = <-result:
+	case <-time.After(time.Second):
+		t.Fatal("lockObjectPublish did not honor cancellation")
+	}
+	if !errors.Is(err, context.Canceled) {
+		t.Fatalf("lockObjectPublish error = %v, want context.Canceled", err)
+	}
+}
+
+func TestObjectPublishLockHonorsCancellationAfterSlotAcquired(t *testing.T) {
+	p := newTestPosix(t, metaModes(t)["xattr"])
+	ctx, cancel := context.WithCancel(context.Background())
+	cancel()
+
+	unlock, err := p.lockObjectPublish(ctx, "testbucket", "cancel-ready-slot")
+	if unlock != nil {
+		unlock()
+	}
+	if !errors.Is(err, context.Canceled) {
+		t.Fatalf("lockObjectPublish error = %v, want context.Canceled", err)
+	}
+}
+
+func TestObjectPublishLockCanDisableSharedLockFile(t *testing.T) {
+	root := t.TempDir()
+	p, err := New(root, meta.XattrMeta{}, PosixOpts{
+		NewDirPerm:         0755,
+		ForceNoObjLockFile: true,
+	})
+	if err != nil {
+		t.Fatalf("new posix: %v", err)
+	}
+	defer p.Shutdown()
+
+	unlock, err := p.lockObjectPublish(context.Background(), "testbucket", "object")
+	if err != nil {
+		t.Fatalf("lock object publish: %v", err)
+	}
+	unlock()
+
+	if _, err := os.Stat(filepath.Join(root, objLockDir)); !errors.Is(err, fs.ErrNotExist) {
+		t.Fatalf("shared lock directory exists or stat failed: %v", err)
+	}
+}
+
+func TestObjectPublishLockUsesRootForRelativeBackendPath(t *testing.T) {
+	parent := t.TempDir()
+	rootName := "gateway-root"
+	root := filepath.Join(parent, rootName)
+	if err := os.Mkdir(root, 0755); err != nil {
+		t.Fatalf("make root: %v", err)
+	}
+	t.Chdir(parent)
+
+	p, err := New(rootName, meta.XattrMeta{}, PosixOpts{NewDirPerm: 0755})
+	if err != nil {
+		t.Fatalf("new posix: %v", err)
+	}
+	defer p.Shutdown()
+
+	unlock, err := p.lockObjectPublish(context.Background(), "testbucket", "relative-root")
+	if err != nil {
+		t.Fatalf("lock object publish: %v", err)
+	}
+	unlock()
+
+	if _, err := os.Stat(filepath.Join(root, objLockDir)); err != nil {
+		t.Fatalf("root lock directory: %v", err)
+	}
+	if _, err := os.Stat(filepath.Join(root, rootName, objLockDir)); !errors.Is(err, fs.ErrNotExist) {
+		t.Fatalf("nested lock directory exists or stat failed: %v", err)
+	}
+}
+
+func TestListBucketsAndOwnersExcludesObjectLockDirectory(t *testing.T) {
+	p := newTestPosix(t, metaModes(t)["xattr"])
+
+	unlock, err := p.lockObjectPublish(context.Background(), "testbucket", "object")
+	if err != nil {
+		t.Fatalf("lock object publish: %v", err)
+	}
+	unlock()
+
+	buckets, err := p.ListBucketsAndOwners(context.Background())
+	if err != nil {
+		t.Fatalf("list buckets and owners: %v", err)
+	}
+	if len(buckets) != 0 {
+		t.Fatalf("buckets = %#v, want no internal lock directory", buckets)
+	}
+}
+
 func testPut(p *Posix, bucket, key string, body []byte, ifMatch, ifNoneMatch *string) (s3response.PutObjectOutput, error) {
 	return p.PutObject(context.Background(), s3response.PutObjectInput{
 		Bucket:        &bucket,

backend/scoutfs/scoutfs_compat.go

@@ -84,6 +84,7 @@ func New(rootdir string, opts ScoutfsOpts) (*ScoutFS, error) {
 		CopyObjectThreshold: opts.CopyObjectThreshold,
 		DefaultEtag:         opts.DefaultEtag,
 		DataIntegrityEtag:   opts.DataIntegrityEtag,
+		ForceNoObjLockFile:  true, // scoutfs flock not cluster consistent
 	}
 	if opts.newDirPermSet {
 		posixOpts.SetNewDirPerm(opts.NewDirPerm)

cmd/internal/gwcli/posix.go

@@ -39,6 +39,7 @@ var (
 	nometa               bool
 	forceNoTmpFile       bool
 	forceNoCopyFileRange bool
+	forceNoObjLockFile   bool
 	enableODirect        bool
 	actionsConcurrency   int
 	ioBufferSize         int
@@ -142,6 +143,12 @@ will be translated into the file /mnt/fs/gwroot/mybucket/a/b/c/myobject`,
 				EnvVars:     []string{"VGW_DISABLE_COPY_FILE_RANGE"},
 				Destination: &forceNoCopyFileRange,
 			},
+			&cli.BoolFlag{
+				Name:        "disable-object-lock-file",
+				Usage:       "disable shared advisory lock files for conditional object publishes (unsafe with multiple gateway processes)",
+				EnvVars:     []string{"VGW_DISABLE_OBJECT_LOCK_FILE"},
+				Destination: &forceNoObjLockFile,
+			},
 			&cli.BoolFlag{
 				Name:        "enable-odirect",
 				Usage:       "enable best-effort O_DIRECT for object data reads/writes",
@@ -194,6 +201,7 @@ func runPosix(ctx *cli.Context) error {
 		VersioningDir:        versioningDir,
 		ForceNoTmpFile:       forceNoTmpFile,
 		ForceNoCopyFileRange: forceNoCopyFileRange,
+		ForceNoObjLockFile:   forceNoObjLockFile,
 		EnableODirect:        enableODirect,
 		ValidateBucketNames:  DisableStrictBucketNames,
 		Concurrency:          actionsConcurrency,

cmd/versitygw/main.go

@@ -123,9 +123,16 @@ func main() {
 	gwcli.RunIAM = runIAM
 
 	app := initApp()
+	posixCommand := gwcli.PosixCommand()
+	posixCommand.Before = func(ctx *cli.Context) error {
+		if ctx.Bool("disable-object-lock-file") {
+			fmt.Println("Warning: shared object publish locking disabled; conditional write atomicity is limited to this gateway process")
+		}
+		return nil
+	}
 
 	app.Commands = []*cli.Command{
-		gwcli.PosixCommand(),
+		posixCommand,
 		gwcli.ScoutfsCommand(),
 		gwcli.S3Command(),
 		gwcli.AzureCommand(),

extra/example.conf

@@ -655,6 +655,16 @@ ROOT_SECRET_ACCESS_KEY=
 # NFS servers that may hang on this call.
 #VGW_DISABLE_COPY_FILE_RANGE=false
 
+# The VGW_DISABLE_OBJECT_LOCK_FILE option disables shared advisory lock files
+# used for conditional object publishes. This can be necessary on filesystems
+# that do not support advisory file locking. When enabled, conditional writes
+# are atomic only within a single versitygw process and are unsafe with
+# multiple gateway processes sharing the same backend.
+# This can be set to true to disable the .vgwlocks directory if running in 
+# single versitygw instance mode or conditional writes are not important
+# to this deployment.
+#VGW_DISABLE_OBJECT_LOCK_FILE=false
+
 # The VGW_ENABLE_O_DIRECT option enables best-effort O_DIRECT for object data
 # reads and writes. This is disabled by default. When enabled, versitygw
 # attempts to open object data files with O_DIRECT and automatically falls back