Merge remote-tracking branch 'upstream/release-19.0' into release-19.…

…0-github

Signed-off-by: Mohamed Hamza <mhamza15@github.com>

go/os2/file.go

@@ -0,0 +1,50 @@
+/*
+Copyright 2025 The Vitess Authors.
+
+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 os2
+
+import (
+	"io/fs"
+	"os"
+)
+
+const (
+	// PermFile is a FileMode for regular files without world permission bits.
+	PermFile fs.FileMode = 0660
+	// PermDirectory is a FileMode for directories without world permission bits.
+	PermDirectory fs.FileMode = 0770
+)
+
+// Create is identical to os.Create except uses 0660 permission
+// rather than 0666, to exclude world read/write bit.
+func Create(name string) (*os.File, error) {
+	return os.OpenFile(name, os.O_RDWR|os.O_CREATE|os.O_TRUNC, PermFile)
+}
+
+// WriteFile is identical to os.WriteFile except permission of 0660 is used.
+func WriteFile(name string, data []byte) error {
+	return os.WriteFile(name, data, PermFile)
+}
+
+// Mkdir is identical to os.Mkdir except permission of 0770 is used.
+func Mkdir(path string) error {
+	return os.Mkdir(path, PermDirectory)
+}
+
+// MkdirAll is identical to os.MkdirAll except permission of 0770 is used.
+func MkdirAll(path string) error {
+	return os.MkdirAll(path, PermDirectory)
+}

go/pools/smartconnpool/pool.go

@@ -156,7 +156,6 @@ type ConnPool[C Connection] struct {
 // The pool must be ConnPool.Open before it can start giving out connections
 func NewPool[C Connection](config *Config[C]) *ConnPool[C] {
 	pool := &ConnPool[C]{}
-	pool.freshSettingsStack.Store(-1)
 	pool.config.maxCapacity = config.Capacity
 	pool.config.maxLifetime.Store(config.MaxLifetime.Nanoseconds())
 	pool.config.idleTimeout.Store(config.IdleTimeout.Nanoseconds())
@@ -195,8 +194,14 @@ func (pool *ConnPool[C]) open() {
 
 	// The expire worker takes care of removing from the waiter list any clients whose
 	// context has been cancelled.
-	pool.runWorker(pool.close, 1*time.Second, func(_ time.Time) bool {
-		pool.wait.expire(false)
+	pool.runWorker(pool.close, 100*time.Millisecond, func(_ time.Time) bool {
+		maybeStarving := pool.wait.expire(false)
+
+		// Do not allow connections to starve; if there's waiters in the queue
+		// and connections in the stack, it means we could be starving them.
+		// Try getting out a connection and handing it over directly
+		for n := 0; n < maybeStarving && pool.tryReturnAnyConn(); n++ {
+		}
 		return true
 	})
 
@@ -395,16 +400,34 @@ func (pool *ConnPool[C]) put(conn *Pooled[C]) {
 		}
 	}
 
-	if !pool.wait.tryReturnConn(conn) {
-		connSetting := conn.Conn.Setting()
-		if connSetting == nil {
-			pool.clean.Push(conn)
-		} else {
-			stack := connSetting.bucket & stackMask
-			pool.settings[stack].Push(conn)
-			pool.freshSettingsStack.Store(int64(stack))
+	pool.tryReturnConn(conn)
+}
+
+func (pool *ConnPool[C]) tryReturnConn(conn *Pooled[C]) bool {
+	if pool.wait.tryReturnConn(conn) {
+		return true
+	}
+	connSetting := conn.Conn.Setting()
+	if connSetting == nil {
+		pool.clean.Push(conn)
+	} else {
+		stack := connSetting.bucket & stackMask
+		pool.settings[stack].Push(conn)
+		pool.freshSettingsStack.Store(int64(stack))
+	}
+	return false
+}
+
+func (pool *ConnPool[C]) tryReturnAnyConn() bool {
+	if conn, ok := pool.clean.Pop(); ok {
+		return pool.tryReturnConn(conn)
+	}
+	for u := 0; u <= stackMask; u++ {
+		if conn, ok := pool.settings[u].Pop(); ok {
+			return pool.tryReturnConn(conn)
 		}
 	}
+	return false
 }
 
 func (pool *ConnPool[D]) extendedMaxLifetime() time.Duration {
@@ -443,14 +466,9 @@ func (pool *ConnPool[C]) connNew(ctx context.Context) (*Pooled[C], error) {
 }
 
 func (pool *ConnPool[C]) getFromSettingsStack(setting *Setting) *Pooled[C] {
-	fresh := pool.freshSettingsStack.Load()
-	if fresh < 0 {
-		return nil
-	}
-
 	var start uint32
 	if setting == nil {
-		start = uint32(fresh)
+		start = uint32(pool.freshSettingsStack.Load())
 	} else {
 		start = setting.bucket
 	}

go/pools/smartconnpool/pool_test.go

@@ -20,6 +20,7 @@ import (
 	"context"
 	"fmt"
 	"reflect"
+	"sync"
 	"sync/atomic"
 	"testing"
 	"time"
@@ -36,6 +37,7 @@ var (
 type TestState struct {
 	lastID, open, close, reset atomic.Int64
 	waits                      []time.Time
+	mu                         sync.Mutex
 
 	chaos struct {
 		delayConnect time.Duration
@@ -45,6 +47,8 @@ type TestState struct {
 }
 
 func (ts *TestState) LogWait(start time.Time) {
+	ts.mu.Lock()
+	defer ts.mu.Unlock()
 	ts.waits = append(ts.waits, start)
 }
 
@@ -1080,3 +1084,68 @@ func TestApplySettingsFailure(t *testing.T) {
 		p.put(r)
 	}
 }
+
+func TestGetSpike(t *testing.T) {
+	var state TestState
+
+	ctx := context.Background()
+	p := NewPool(&Config[*TestConn]{
+		Capacity:    5,
+		IdleTimeout: time.Second,
+		LogWait:     state.LogWait,
+	}).Open(newConnector(&state), nil)
+
+	var resources [10]*Pooled[*TestConn]
+
+	// Ensure we have a pool with 5 available resources
+	for i := 0; i < 5; i++ {
+		r, err := p.Get(ctx, nil)
+
+		require.NoError(t, err)
+		resources[i] = r
+		assert.EqualValues(t, 5-i-1, p.Available())
+		assert.Zero(t, p.Metrics.WaitCount())
+		assert.Zero(t, len(state.waits))
+		assert.Zero(t, p.Metrics.WaitTime())
+		assert.EqualValues(t, i+1, state.lastID.Load())
+		assert.EqualValues(t, i+1, state.open.Load())
+	}
+
+	for i := 0; i < 5; i++ {
+		p.put(resources[i])
+	}
+
+	assert.EqualValues(t, 5, p.Available())
+	assert.EqualValues(t, 5, p.Active())
+	assert.EqualValues(t, 0, p.InUse())
+
+	for i := 0; i < 2000; i++ {
+		wg := sync.WaitGroup{}
+
+		ctx, cancel := context.WithTimeout(context.Background(), time.Second)
+		defer cancel()
+
+		errs := make(chan error, 80)
+
+		for j := 0; j < 80; j++ {
+			wg.Add(1)
+
+			go func() {
+				defer wg.Done()
+				r, err := p.Get(ctx, nil)
+				defer p.put(r)
+
+				if err != nil {
+					errs <- err
+				}
+			}()
+		}
+		wg.Wait()
+
+		if len(errs) > 0 {
+			t.Errorf("Error getting connection: %v", <-errs)
+		}
+
+		close(errs)
+	}
+}

go/pools/smartconnpool/waitlist.go

@@ -76,7 +76,7 @@ func (wl *waitlist[C]) waitForConn(ctx context.Context, setting *Setting) (*Pool
 
 // expire removes and wakes any expired waiter in the waitlist.
 // if force is true, it'll wake and remove all the waiters.
-func (wl *waitlist[C]) expire(force bool) {
+func (wl *waitlist[C]) expire(force bool) (maybeStarving int) {
 	if wl.list.Len() == 0 {
 		return
 	}
@@ -91,6 +91,9 @@ func (wl *waitlist[C]) expire(force bool) {
 			expired = append(expired, e)
 			continue
 		}
+		if e.Value.age == 0 {
+			maybeStarving++
+		}
 	}
 	// remove the expired waiters from the waitlist after traversing it
 	for _, e := range expired {
@@ -102,6 +105,7 @@ func (wl *waitlist[C]) expire(force bool) {
 	for _, e := range expired {
 		e.Value.sema.notify(false)
 	}
+	return
 }
 
 // tryReturnConn tries handing over a connection to one of the waiters in the pool.

go/test/endtoend/backup/vtctlbackup/backup_utils.go

@@ -408,6 +408,18 @@ func TestBackup(t *testing.T, setupType int, streamMode string, stripes int, cDe
 			return vterrors.Errorf(vtrpc.Code_UNKNOWN, "test failure: %s", test.name)
 		}
 	}
+
+	t.Run("check for files created with global permissions", func(t *testing.T) {
+		t.Logf("Confirming that none of the MySQL data directories that we've created have files with global permissions")
+		for _, ks := range localCluster.Keyspaces {
+			for _, shard := range ks.Shards {
+				for _, tablet := range shard.Vttablets {
+					tablet.VttabletProcess.ConfirmDataDirHasNoGlobalPerms(t)
+				}
+			}
+		}
+	})
+
 	return nil
 }
 

go/test/endtoend/cluster/vttablet_process.go

@@ -24,6 +24,7 @@ import (
 	"errors"
 	"fmt"
 	"io"
+	"io/fs"
 	"net/http"
 	"os"
 	"os/exec"
@@ -35,6 +36,8 @@ import (
 	"testing"
 	"time"
 
+	"github.com/stretchr/testify/require"
+
 	"vitess.io/vitess/go/constants/sidecar"
 	"vitess.io/vitess/go/mysql"
 	"vitess.io/vitess/go/sqltypes"
@@ -677,6 +680,67 @@ func (vttablet *VttabletProcess) IsShutdown() bool {
 	return vttablet.proc == nil
 }
 
+// ConfirmDataDirHasNoGlobalPerms confirms that no files in the tablet's data directory
+// have any global/world/other permissions enabled.
+func (vttablet *VttabletProcess) ConfirmDataDirHasNoGlobalPerms(t *testing.T) {
+	datadir := vttablet.Directory
+	if _, err := os.Stat(datadir); errors.Is(err, os.ErrNotExist) {
+		t.Logf("Data directory %s no longer exists, skipping permissions check", datadir)
+		return
+	}
+
+	var allowedFiles = []string{
+		// These are intentionally created with the world/other read bit set by mysqld itself
+		// during the --initialize[-insecure] step.
+		// See: https://dev.mysql.com/doc/mysql-security-excerpt/en/creating-ssl-rsa-files-using-mysql.html
+		// "On Unix and Unix-like systems, the file access mode is 644 for certificate files
+		// (that is, world readable) and 600 for key files (that is, accessible only by the
+		// account that runs the server)."
+		path.Join("data", "ca.pem"),
+		path.Join("data", "client-cert.pem"),
+		path.Join("data", "public_key.pem"),
+		path.Join("data", "server-cert.pem"),
+		// The domain socket must have global perms for anyone to use it.
+		"mysql.sock",
+		// These files are created by xtrabackup.
+		path.Join("tmp", "xtrabackup_checkpoints"),
+		path.Join("tmp", "xtrabackup_info"),
+		// These are 5.7 specific xtrabackup files.
+		path.Join("data", "xtrabackup_binlog_pos_innodb"),
+		path.Join("data", "xtrabackup_master_key_id"),
+		path.Join("data", "mysql_upgrade_info"),
+	}
+
+	var matches []string
+	fsys := os.DirFS(datadir)
+	err := fs.WalkDir(fsys, ".", func(p string, d fs.DirEntry, _ error) error {
+		// first check if the file should be skipped
+		for _, name := range allowedFiles {
+			if strings.HasSuffix(p, name) {
+				return nil
+			}
+		}
+
+		info, err := d.Info()
+		if err != nil {
+			return err
+		}
+
+		// check if any global bit is on the filemode
+		if info.Mode()&0007 != 0 {
+			matches = append(matches, fmt.Sprintf(
+				"%s (%s)",
+				path.Join(datadir, p),
+				info.Mode(),
+			))
+		}
+		return nil
+	})
+
+	require.NoError(t, err, "Error walking directory")
+	require.Empty(t, matches, "Found files with global permissions: %s\n", strings.Join(matches, "\n"))
+}
+
 // VttabletProcessInstance returns a VttabletProcess handle for vttablet process
 // configured with the given Config.
 // The process must be manually started by calling setup()

go/test/endtoend/vreplication/cluster_test.go

@@ -887,7 +887,7 @@ func (vc *VitessCluster) getVttabletsInKeyspace(t *testing.T, cell *Cell, ksName
 	tablets := make(map[string]*cluster.VttabletProcess)
 	for _, shard := range keyspace.Shards {
 		for _, tablet := range shard.Tablets {
-			if tablet.Vttablet.GetTabletStatus() == "SERVING" {
+			if tablet.Vttablet.GetTabletStatus() == "SERVING" && (tabletType == "" || strings.EqualFold(tablet.Vttablet.GetTabletType(), tabletType)) {
 				log.Infof("Serving status of tablet %s is %s, %s", tablet.Name, tablet.Vttablet.ServingStatus, tablet.Vttablet.GetTabletStatus())
 				tablets[tablet.Name] = tablet.Vttablet
 			}

go/test/endtoend/vreplication/vdiff_helper_test.go

@@ -337,9 +337,7 @@ func getVDiffInfo(json string) *vdiffInfo {
 }
 
 func encodeString(in string) string {
-	var buf strings.Builder
-	sqltypes.NewVarChar(in).EncodeSQL(&buf)
-	return buf.String()
+	return sqltypes.EncodeStringSQL(in)
 }
 
 // generateMoreCustomers creates additional test data for better tests