refactor(mysql): 适配PXC严格模式,替换全局锁方案
重构MySQL迁移逻辑,新增acquireMigrationLock适配Percona XtraDB Cluster的pxc_strict_mode限制,将原有的GET_LOCK迁移锁替换为更兼容的方案,同时为各迁移流程添加快速检查路径减少锁竞争
This commit is contained in:
@@ -0,0 +1,36 @@
|
|||||||
|
//go:build mysql
|
||||||
|
// +build mysql
|
||||||
|
|
||||||
|
package mysql
|
||||||
|
|
||||||
|
import (
|
||||||
|
"context"
|
||||||
|
"database/sql"
|
||||||
|
"errors"
|
||||||
|
|
||||||
|
"github.com/jmoiron/sqlx"
|
||||||
|
)
|
||||||
|
|
||||||
|
// acquireMigrationLock wraps SELECT GET_LOCK so that schema migrations keep
|
||||||
|
// working on Percona XtraDB Cluster where pxc_strict_mode=ENFORCING prohibits
|
||||||
|
// named locks: instead of failing, locked=false is returned and the caller
|
||||||
|
// proceeds without the lock. This is safe because Galera's total order
|
||||||
|
// isolation serializes DDL across the cluster, and every migration using this
|
||||||
|
// helper is idempotent: steps are guarded by existence checks, so concurrent
|
||||||
|
// runs at worst fail with duplicate-key errors and are retried on the next
|
||||||
|
// startup.
|
||||||
|
func acquireMigrationLock(ctx context.Context, conn *sqlx.Conn, lockExpr string) (locked bool, unlock func(), err error) {
|
||||||
|
var held sql.NullInt64
|
||||||
|
if err = conn.QueryRowContext(ctx, "SELECT GET_LOCK("+lockExpr+", 60)").Scan(&held); err != nil {
|
||||||
|
if isPXCStrictModeNamedLockError(err) {
|
||||||
|
return false, nil, nil
|
||||||
|
}
|
||||||
|
return false, nil, err
|
||||||
|
}
|
||||||
|
if !held.Valid || held.Int64 != 1 {
|
||||||
|
return false, nil, errors.New("mysql: timed out waiting for schema migration lock")
|
||||||
|
}
|
||||||
|
return true, func() {
|
||||||
|
conn.ExecContext(ctx, "SELECT RELEASE_LOCK("+lockExpr+")")
|
||||||
|
}, nil
|
||||||
|
}
|
||||||
@@ -58,15 +58,16 @@ func ensureSnowflakePrimaryKeys(db *sqlx.DB) error {
|
|||||||
}
|
}
|
||||||
defer conn.Close()
|
defer conn.Close()
|
||||||
|
|
||||||
var locked sql.NullInt64
|
// PXC strict-mode clusters prohibit GET_LOCK; acquireMigrationLock falls back
|
||||||
if err = conn.QueryRowContext(ctx,
|
// to lock-free execution in that case. migrateSnowflakePrimaryKey checks the
|
||||||
"SELECT GET_LOCK("+snowflakeMigrationLock+", 60)").Scan(&locked); err != nil {
|
// current state of every table before touching it, so it is idempotent.
|
||||||
|
_, unlock, err := acquireMigrationLock(ctx, conn, snowflakeMigrationLock)
|
||||||
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
if !locked.Valid || locked.Int64 != 1 {
|
if unlock != nil {
|
||||||
return errors.New("mysql: timed out waiting for snowflake primary-key migration lock")
|
defer unlock()
|
||||||
}
|
}
|
||||||
defer conn.ExecContext(ctx, "SELECT RELEASE_LOCK("+snowflakeMigrationLock+")")
|
|
||||||
|
|
||||||
for _, spec := range snowflakePrimaryKeySpecs {
|
for _, spec := range snowflakePrimaryKeySpecs {
|
||||||
if err = migrateSnowflakePrimaryKey(ctx, conn, spec); err != nil {
|
if err = migrateSnowflakePrimaryKey(ctx, conn, spec); err != nil {
|
||||||
|
|||||||
@@ -5,7 +5,6 @@ package mysql
|
|||||||
|
|
||||||
import (
|
import (
|
||||||
"context"
|
"context"
|
||||||
"database/sql"
|
|
||||||
"errors"
|
"errors"
|
||||||
"fmt"
|
"fmt"
|
||||||
"strings"
|
"strings"
|
||||||
@@ -32,14 +31,35 @@ func ensureUserIdentitySchema(db *sqlx.DB) error {
|
|||||||
}
|
}
|
||||||
defer conn.Close()
|
defer conn.Close()
|
||||||
|
|
||||||
var locked sql.NullInt64
|
// Fast path: the last two steps of this migration drop the legacy im_usertags
|
||||||
if err = conn.QueryRowContext(ctx, "SELECT GET_LOCK("+userIdentityMigrationLock+", 60)").Scan(&locked); err != nil {
|
// table and the im_users.tags column. If both are already gone, the migration
|
||||||
|
// has fully run (or the database was created fresh) and no lock is needed.
|
||||||
|
stale := false
|
||||||
|
if exists, err := identityTableExists(ctx, conn, "im_usertags"); err != nil {
|
||||||
|
return err
|
||||||
|
} else if exists {
|
||||||
|
stale = true
|
||||||
|
}
|
||||||
|
if !stale {
|
||||||
|
if exists, err := identityColumnExists(ctx, conn, "im_users", "tags"); err != nil {
|
||||||
|
return err
|
||||||
|
} else if exists {
|
||||||
|
stale = true
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if !stale {
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
|
// PXC strict-mode clusters prohibit GET_LOCK; acquireMigrationLock falls back
|
||||||
|
// to lock-free execution in that case. The migration body is idempotent.
|
||||||
|
_, unlock, err := acquireMigrationLock(ctx, conn, userIdentityMigrationLock)
|
||||||
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
if !locked.Valid || locked.Int64 != 1 {
|
if unlock != nil {
|
||||||
return errors.New("mysql: timed out waiting for user identity migration lock")
|
defer unlock()
|
||||||
}
|
}
|
||||||
defer conn.ExecContext(ctx, "SELECT RELEASE_LOCK("+userIdentityMigrationLock+")")
|
|
||||||
|
|
||||||
usersExist, err := identityTableExists(ctx, conn, "im_users")
|
usersExist, err := identityTableExists(ctx, conn, "im_users")
|
||||||
if err != nil || !usersExist {
|
if err != nil || !usersExist {
|
||||||
|
|||||||
@@ -5,8 +5,6 @@ package mysql
|
|||||||
|
|
||||||
import (
|
import (
|
||||||
"context"
|
"context"
|
||||||
"database/sql"
|
|
||||||
"errors"
|
|
||||||
|
|
||||||
"github.com/jmoiron/sqlx"
|
"github.com/jmoiron/sqlx"
|
||||||
)
|
)
|
||||||
@@ -26,22 +24,40 @@ func ensureUserPrivacySchema(db *sqlx.DB) error {
|
|||||||
}
|
}
|
||||||
defer conn.Close()
|
defer conn.Close()
|
||||||
|
|
||||||
var locked sql.NullInt64
|
// Fast path: check for missing columns without taking the migration lock.
|
||||||
if err = conn.QueryRowContext(ctx, "SELECT GET_LOCK("+userPrivacyMigrationLock+", 60)").Scan(&locked); err != nil {
|
// Databases created by createIMUserProductTables already have every column,
|
||||||
|
// so this is the common case and avoids GET_LOCK entirely (prohibited on
|
||||||
|
// Percona XtraDB Cluster with pxc_strict_mode=ENFORCING).
|
||||||
|
added, err := ensureUserPrivacyColumns(ctx, conn)
|
||||||
|
if err != nil || added {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
if !locked.Valid || locked.Int64 != 1 {
|
|
||||||
return errors.New("mysql: timed out waiting for user privacy migration lock")
|
|
||||||
}
|
|
||||||
defer conn.ExecContext(ctx, "SELECT RELEASE_LOCK("+userPrivacyMigrationLock+")")
|
|
||||||
|
|
||||||
|
_, unlock, err := acquireMigrationLock(ctx, conn, userPrivacyMigrationLock)
|
||||||
|
if err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
if unlock != nil {
|
||||||
|
defer unlock()
|
||||||
|
}
|
||||||
|
|
||||||
|
// Double-check under the lock: another instance may have migrated while we
|
||||||
|
// were waiting. Without a lock (PXC fallback) this simply runs idempotently.
|
||||||
|
_, err = ensureUserPrivacyColumns(ctx, conn)
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
|
||||||
|
// ensureUserPrivacyColumns adds any columns of im_user_privacy_settings that
|
||||||
|
// are missing and reports whether nothing was missing. A missing table is
|
||||||
|
// skipped: it is created by the fresh-install path and there is nothing to
|
||||||
|
// upgrade yet. Column definitions mirror createIMUserProductTables; the SELECT
|
||||||
|
// in userPrivacyByUID references every one of them.
|
||||||
|
func ensureUserPrivacyColumns(ctx context.Context, conn *sqlx.Conn) (bool, error) {
|
||||||
tableExists, err := identityTableExists(ctx, conn, "im_user_privacy_settings")
|
tableExists, err := identityTableExists(ctx, conn, "im_user_privacy_settings")
|
||||||
if err != nil || !tableExists {
|
if err != nil || !tableExists {
|
||||||
return err
|
return true, err
|
||||||
}
|
}
|
||||||
|
|
||||||
// Column definitions mirror createIMUserProductTables; the SELECT in
|
|
||||||
// userPrivacyByUID references every one of them.
|
|
||||||
columns := []struct{ name, ddl string }{
|
columns := []struct{ name, ddl string }{
|
||||||
{"profile_visibility", "VARCHAR(32) NOT NULL DEFAULT 'friends'"},
|
{"profile_visibility", "VARCHAR(32) NOT NULL DEFAULT 'friends'"},
|
||||||
{"last_seen_visibility", "VARCHAR(32) NOT NULL DEFAULT 'friends'"},
|
{"last_seen_visibility", "VARCHAR(32) NOT NULL DEFAULT 'friends'"},
|
||||||
@@ -49,16 +65,19 @@ func ensureUserPrivacySchema(db *sqlx.DB) error {
|
|||||||
{"friend_request_verification", "TINYINT NOT NULL DEFAULT 1"},
|
{"friend_request_verification", "TINYINT NOT NULL DEFAULT 1"},
|
||||||
{"group_invite_verification", "TINYINT NOT NULL DEFAULT 0"},
|
{"group_invite_verification", "TINYINT NOT NULL DEFAULT 0"},
|
||||||
}
|
}
|
||||||
|
missing := false
|
||||||
for _, col := range columns {
|
for _, col := range columns {
|
||||||
exists, err := identityColumnExists(ctx, conn, "im_user_privacy_settings", col.name)
|
exists, err := identityColumnExists(ctx, conn, "im_user_privacy_settings", col.name)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
return false, err
|
||||||
}
|
}
|
||||||
if !exists {
|
if exists {
|
||||||
if _, err = conn.ExecContext(ctx, "ALTER TABLE im_user_privacy_settings ADD COLUMN "+col.name+" "+col.ddl); err != nil {
|
continue
|
||||||
return err
|
}
|
||||||
}
|
missing = true
|
||||||
|
if _, err = conn.ExecContext(ctx, "ALTER TABLE im_user_privacy_settings ADD COLUMN "+col.name+" "+col.ddl); err != nil {
|
||||||
|
return false, err
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
return nil
|
return !missing, nil
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user