Add MySQL support to River

.github/workflows/ci.yaml

@@ -77,6 +77,8 @@ jobs:
           # Run the latest Go version against all supported Postgres versions:
           - go-version: "1.26"
             postgres-version: 18
+            # MySQL tests are opt-in and only need to run in one matrix job.
+            mysql-enabled: true
           - go-version: "1.26"
             postgres-version: 17
           - go-version: "1.26"
@@ -91,6 +93,7 @@ jobs:
           # latest Postgres version:
           - go-version: "1.25"
             postgres-version: 18
+
       fail-fast: false
     timeout-minutes: 5
 
@@ -128,7 +131,17 @@ jobs:
       - name: Set up database
         run: psql -c "CREATE DATABASE river_test" $ADMIN_DATABASE_URL
 
+      # MySQL is pre-installed on GitHub-hosted Ubuntu runners. Start the
+      # service and configure passwordless root access for MySQL-enabled runs.
+      - name: Start MySQL
+        if: matrix.mysql-enabled
+        run: |
+          sudo systemctl start mysql
+          mysql -u root -proot -e "ALTER USER 'root'@'localhost' IDENTIFIED BY ''; FLUSH PRIVILEGES;"
+
       - name: Test
+        env:
+          RIVER_MYSQL_TESTS_ENABLED: ${{ matrix.mysql-enabled && 'true' || '' }}
         run: make test/race
 
   cli:

Makefile

@@ -27,6 +27,7 @@ generate/migrations: ## Sync changes of pgxv5 migrations to database/sql
 .PHONY: generate/sqlc
 generate/sqlc: ## Generate sqlc
 	cd riverdriver/riverdatabasesql/internal/dbsqlc && sqlc generate
+	cd riverdriver/rivermysql/internal/dbsqlc && sqlc generate
 	cd riverdriver/riverpgxv5/internal/dbsqlc && sqlc generate
 	cd riverdriver/riversqlite/internal/dbsqlc && sqlc generate
 
@@ -101,5 +102,6 @@ verify/migrations: ## Verify synced migrations
 .PHONY: verify/sqlc
 verify/sqlc: ## Verify generated sqlc
 	cd riverdriver/riverdatabasesql/internal/dbsqlc && sqlc diff
+	cd riverdriver/rivermysql/internal/dbsqlc && sqlc diff
 	cd riverdriver/riverpgxv5/internal/dbsqlc && sqlc diff
 	cd riverdriver/riversqlite/internal/dbsqlc && sqlc diff

client.go

@@ -1011,11 +1011,11 @@ func NewClient[TTx any](driver riverdriver.Driver[TTx], config *Config) (*Client
 			client.testSignals.queueCleaner = &queueCleaner.TestSignals
 		}
 
-		if driver.DatabaseName() == riverdriver.DatabaseNameSQLite {
-			sqliteNotificationCleaner := maintenance.NewSQLiteNotificationCleaner(archetype, &maintenance.SQLiteNotificationCleanerConfig{
+		if driver.DatabaseName() == riverdriver.DatabaseNameMySQL || driver.DatabaseName() == riverdriver.DatabaseNameSQLite {
+			notificationCleaner := maintenance.NewSQLiteNotificationCleaner(archetype, &maintenance.SQLiteNotificationCleanerConfig{
 				Schema: config.Schema,
 			}, driver.GetExecutor())
-			maintenanceServices = append(maintenanceServices, sqliteNotificationCleaner)
+			maintenanceServices = append(maintenanceServices, notificationCleaner)
 		}
 
 		{
@@ -2425,11 +2425,13 @@ func (c *Client[TTx]) JobList(ctx context.Context, params *JobListParams) (*JobL
 	if params == nil {
 		params = NewJobListParams()
 	}
-	params.schema = c.config.Schema
-
 	if c.driver.DatabaseName() == riverdriver.DatabaseNameSQLite && params.metadataCalled {
 		return nil, errJobListParamsMetadataNotSupportedSQLite
 	}
+	if c.driver.DatabaseName() == riverdriver.DatabaseNameMySQL && params.metadataCalled {
+		params = params.withMetadataPredicateSQL("JSON_CONTAINS(metadata, CAST(@metadata_fragment AS JSON))")
+	}
+	params.schema = c.config.Schema
 
 	dbParams, err := params.toDBParams()
 	if err != nil {
@@ -2466,11 +2468,13 @@ func (c *Client[TTx]) JobListTx(ctx context.Context, tx TTx, params *JobListPara
 	if params == nil {
 		params = NewJobListParams()
 	}
-	params.schema = c.config.Schema
-
 	if c.driver.DatabaseName() == riverdriver.DatabaseNameSQLite && params.metadataCalled {
 		return nil, errJobListParamsMetadataNotSupportedSQLite
 	}
+	if c.driver.DatabaseName() == riverdriver.DatabaseNameMySQL && params.metadataCalled {
+		params = params.withMetadataPredicateSQL("JSON_CONTAINS(metadata, CAST(@metadata_fragment AS JSON))")
+	}
+	params.schema = c.config.Schema
 
 	dbParams, err := params.toDBParams()
 	if err != nil {

client_test.go

@@ -8299,6 +8299,32 @@ func Test_NewClient_Overrides(t *testing.T) {
 	require.Len(t, client.config.WorkerMiddleware, 1)
 }
 
+func Test_NewClient_MySQLNotificationCleaner(t *testing.T) {
+	t.Parallel()
+
+	ctx := context.Background()
+
+	var (
+		dbPool    = riversharedtest.DBPool(ctx, t)
+		pgxDriver = riverpgxv5.New(dbPool)
+		schema    = riverdbtest.TestSchema(ctx, t, pgxDriver, nil)
+	)
+
+	workers := NewWorkers()
+	AddWorker(workers, &noOpWorker{})
+
+	client, err := NewClient(NewDriverMySQLDatabaseName(dbPool), &Config{
+		Queues:   map[string]QueueConfig{QueueDefault: {MaxWorkers: 1}},
+		Schema:   schema,
+		TestOnly: true,
+		Workers:  workers,
+	})
+	require.NoError(t, err)
+
+	notificationCleaner := maintenance.GetService[*maintenance.SQLiteNotificationCleaner](client.queueMaintainer)
+	require.Equal(t, schema, notificationCleaner.Config.Schema)
+}
+
 func Test_NewClient_PluginsAndHybrids(t *testing.T) {
 	t.Parallel()
 
@@ -9640,6 +9666,23 @@ func TestDefaultClientIDWithHost(t *testing.T) {
 	require.Equal(t, strings.Repeat("a", 60)+"_2024_03_07T04_39_12_123456", defaultClientIDWithHost(startedAt, strings.Repeat("a", 61)))
 }
 
+// DriverMySQLDatabaseName is a pgx driver that reports MySQL as its database
+// name so MySQL-specific client wiring can be tested without importing the
+// separate MySQL driver module.
+type DriverMySQLDatabaseName struct {
+	riverpgxv5.Driver
+}
+
+// NewDriverMySQLDatabaseName returns a new test driver that reports MySQL as
+// its database name.
+func NewDriverMySQLDatabaseName(dbPool *pgxpool.Pool) *DriverMySQLDatabaseName {
+	return &DriverMySQLDatabaseName{
+		Driver: *riverpgxv5.New(dbPool),
+	}
+}
+
+func (d *DriverMySQLDatabaseName) DatabaseName() string { return riverdriver.DatabaseNameMySQL }
+
 // DriverPollOnly simulates a driver without a listener. An example of this is
 // Postgres through `riverdatabasesql`, which is Postgres (so it can notify),
 // but where `database/sql` provides no listener mechanism. We could use the

cmd/river/go.sum

@@ -36,18 +36,18 @@ github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZb
 github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
 github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec h1:W09IVJc94icq4NjY3clb7Lk8O1qJ8BdBEF8z0ibU0rE=
 github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec/go.mod h1:qqbHyh8v60DhA7CoWK5oRCqLrMHRGoxYCSS9EjAz6Eo=
-github.com/riverqueue/river v0.40.0 h1:4dynKqqU1P22iPmwWDfDj/YZXnuUTZysXXF3wNHekNw=
-github.com/riverqueue/river v0.40.0/go.mod h1:auvB4kHqM97tnshEQxzy2E7aFvJFhl010NB6N29DXXc=
-github.com/riverqueue/river/riverdriver v0.40.0 h1:QjBHHh+kaxgUgK9tPumrfx7W14vhAHLIt+0dK40ikI8=
-github.com/riverqueue/river/riverdriver v0.40.0/go.mod h1:7wEqxsqvtjGk3hBKGKK3IvdnULlPZltpAfkbMr9M2VY=
-github.com/riverqueue/river/riverdriver/riverpgxv5 v0.40.0 h1:WbzXgGukvOqYpsORX4rr03sPZfasjoMckLr4On+Je+4=
-github.com/riverqueue/river/riverdriver/riverpgxv5 v0.40.0/go.mod h1:HtgFIcn/eJ9O3f371S1cSb0ttS7xJMMkkTD/K6gQKQg=
-github.com/riverqueue/river/riverdriver/riversqlite v0.40.0 h1:DE/22nz7BKgQ+0XpPFg/3hoO4mP8Ij+gGcIOS9PusFg=
-github.com/riverqueue/river/riverdriver/riversqlite v0.40.0/go.mod h1:/Xuw1OOo+X4Pj5DCKu4MLbF5BhGIBl/OWp93kN2v5ec=
-github.com/riverqueue/river/rivershared v0.40.0 h1:6ZX1Ok94Nkcx/WpHPhexpfdkRB1BqRWFEWrbDszg+JY=
-github.com/riverqueue/river/rivershared v0.40.0/go.mod h1:Z77wB2/ctD+zSvlNDKjxC1SW9c1Y2jyE+PXR2CxBlzQ=
-github.com/riverqueue/river/rivertype v0.40.0 h1:6sFLhs0OtkxDZ/iyZIUqaQJb/G1krnb2bh9lfXdCimg=
-github.com/riverqueue/river/rivertype v0.40.0/go.mod h1:D1Ad+EaZiaXbQbJcJcfeicXJMBKno0n6UcfKI5Q7DIQ=
+github.com/riverqueue/river v0.41.0 h1:E7Yfyhn74IgVaCBKsBpTHoC0LxWJvoG0jKt7Tb+lJZg=
+github.com/riverqueue/river v0.41.0/go.mod h1:WXiAF1/2gfUPj3H+WqQG3Q5NW5w9JmRXRTe7U9zybNc=
+github.com/riverqueue/river/riverdriver v0.41.0 h1:A5g80n6fCGu9Bp4hSq7gys7mvfOiuWkbQHuFx3iPp7w=
+github.com/riverqueue/river/riverdriver v0.41.0/go.mod h1:OeGOhZYoH7Ac0CrVanNAAwD9C0WAXy4LzUEl65Ex0B4=
+github.com/riverqueue/river/riverdriver/riverpgxv5 v0.41.0 h1:uHGluToMMWvCnpNkAbEWEmP4YAjdseuriEXGu1dwOp4=
+github.com/riverqueue/river/riverdriver/riverpgxv5 v0.41.0/go.mod h1:0ueUi+3eW5fIsibS5VkD2HzQmwLKTAiM/uiiVMX/SME=
+github.com/riverqueue/river/riverdriver/riversqlite v0.41.0 h1:4IBOD4Uzil620AT/WT5xPHVoQSk2OiQ5UvG7OQ2piOc=
+github.com/riverqueue/river/riverdriver/riversqlite v0.41.0/go.mod h1:k16A8Tt6Ke49rCU4k33T5EKXAhAXiExyU2qV+6hpkNs=
+github.com/riverqueue/river/rivershared v0.41.0 h1:ax3uz5KiqfmcvRQ3kKB6iYs9Nt8J5L1RNfOKoVMWkqw=
+github.com/riverqueue/river/rivershared v0.41.0/go.mod h1:FaZ7bxC2DORhyFDVyHKRiZzgQPf2b/IELwPBXnHZVLA=
+github.com/riverqueue/river/rivertype v0.41.0 h1:dfscvt1asf1PpeeHTMFxQdV1ZfMoReiiMNFCPPfR0as=
+github.com/riverqueue/river/rivertype v0.41.0/go.mod h1:D1Ad+EaZiaXbQbJcJcfeicXJMBKno0n6UcfKI5Q7DIQ=
 github.com/robfig/cron/v3 v3.0.1 h1:WdRxkvbJztn8LMz/QEvLN5sBU+xKpSqwwUO1Pjr4qDs=
 github.com/robfig/cron/v3 v3.0.1/go.mod h1:eQICP3HwyT7UooqI/z+Ov+PtYAWygg1TEWWzGIFLtro=
 github.com/rogpeppe/go-internal v1.12.0 h1:exVL4IDcn6na9z1rAb56Vxr+CgyK3nn3O+epU5NdKM8=
@@ -76,8 +76,8 @@ github.com/tidwall/sjson v1.2.5/go.mod h1:Fvgq9kS/6ociJEDnK0Fk1cpYF4FIW6ZF7LAe+6
 go.uber.org/goleak v1.3.0 h1:2K3zAYmnTNqV73imy9J1T3WC+gmCePx2hEGkimedGto=
 go.uber.org/goleak v1.3.0/go.mod h1:CoHD4mav9JJNrW/WLlf7HGZPjdw8EucARQHekz1X6bE=
 go.yaml.in/yaml/v3 v3.0.4/go.mod h1:DhzuOOF2ATzADvBadXxruRBLzYTpT36CKvDb3+aBEFg=
-golang.org/x/mod v0.37.0 h1:vF1DjpVEshcIqoEaauuHebaLk1O1forxjxBaVn884JQ=
-golang.org/x/mod v0.37.0/go.mod h1:m8S8VeM9r4dzDwjrKO0a1sZP3YjeMamRRlD+fmR2Q/0=
+golang.org/x/mod v0.38.0 h1:MECBjubtXD7yj4HrhIUcywNaGeNVUdfVnxmPajOk4yk=
+golang.org/x/mod v0.38.0/go.mod h1:V6Xz0pq8TQ3dGqVQ1FVHuelZpAL0uNhSkk9ogYP3c40=
 golang.org/x/sync v0.22.0 h1:SZjpbeLmrCk4xhRSZFNZW5gFUeCeFgjekvI/+gfScek=
 golang.org/x/sync v0.22.0/go.mod h1:9xrNwdLfx4jkKbNva9FpL6vEN7evnE43NNNJQ2LF3+0=
 golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=

go.sum

@@ -17,14 +17,14 @@ github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY=
 github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE=
 github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
 github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
-github.com/riverqueue/river/riverdriver v0.40.0 h1:QjBHHh+kaxgUgK9tPumrfx7W14vhAHLIt+0dK40ikI8=
-github.com/riverqueue/river/riverdriver v0.40.0/go.mod h1:7wEqxsqvtjGk3hBKGKK3IvdnULlPZltpAfkbMr9M2VY=
-github.com/riverqueue/river/riverdriver/riverpgxv5 v0.40.0 h1:WbzXgGukvOqYpsORX4rr03sPZfasjoMckLr4On+Je+4=
-github.com/riverqueue/river/riverdriver/riverpgxv5 v0.40.0/go.mod h1:HtgFIcn/eJ9O3f371S1cSb0ttS7xJMMkkTD/K6gQKQg=
-github.com/riverqueue/river/rivershared v0.40.0 h1:6ZX1Ok94Nkcx/WpHPhexpfdkRB1BqRWFEWrbDszg+JY=
-github.com/riverqueue/river/rivershared v0.40.0/go.mod h1:Z77wB2/ctD+zSvlNDKjxC1SW9c1Y2jyE+PXR2CxBlzQ=
-github.com/riverqueue/river/rivertype v0.40.0 h1:6sFLhs0OtkxDZ/iyZIUqaQJb/G1krnb2bh9lfXdCimg=
-github.com/riverqueue/river/rivertype v0.40.0/go.mod h1:D1Ad+EaZiaXbQbJcJcfeicXJMBKno0n6UcfKI5Q7DIQ=
+github.com/riverqueue/river/riverdriver v0.41.0 h1:A5g80n6fCGu9Bp4hSq7gys7mvfOiuWkbQHuFx3iPp7w=
+github.com/riverqueue/river/riverdriver v0.41.0/go.mod h1:OeGOhZYoH7Ac0CrVanNAAwD9C0WAXy4LzUEl65Ex0B4=
+github.com/riverqueue/river/riverdriver/riverpgxv5 v0.41.0 h1:uHGluToMMWvCnpNkAbEWEmP4YAjdseuriEXGu1dwOp4=
+github.com/riverqueue/river/riverdriver/riverpgxv5 v0.41.0/go.mod h1:0ueUi+3eW5fIsibS5VkD2HzQmwLKTAiM/uiiVMX/SME=
+github.com/riverqueue/river/rivershared v0.41.0 h1:ax3uz5KiqfmcvRQ3kKB6iYs9Nt8J5L1RNfOKoVMWkqw=
+github.com/riverqueue/river/rivershared v0.41.0/go.mod h1:FaZ7bxC2DORhyFDVyHKRiZzgQPf2b/IELwPBXnHZVLA=
+github.com/riverqueue/river/rivertype v0.41.0 h1:dfscvt1asf1PpeeHTMFxQdV1ZfMoReiiMNFCPPfR0as=
+github.com/riverqueue/river/rivertype v0.41.0/go.mod h1:D1Ad+EaZiaXbQbJcJcfeicXJMBKno0n6UcfKI5Q7DIQ=
 github.com/robfig/cron/v3 v3.0.1 h1:WdRxkvbJztn8LMz/QEvLN5sBU+xKpSqwwUO1Pjr4qDs=
 github.com/robfig/cron/v3 v3.0.1/go.mod h1:eQICP3HwyT7UooqI/z+Ov+PtYAWygg1TEWWzGIFLtro=
 github.com/rogpeppe/go-internal v1.12.0 h1:exVL4IDcn6na9z1rAb56Vxr+CgyK3nn3O+epU5NdKM8=

go.work

@@ -9,6 +9,7 @@ use (
 	./riverdriver/riverdatabasesql
 	./riverdriver/riverdrivertest
 	./riverdriver/riverpgxv5
+	./riverdriver/rivermysql
 	./riverdriver/riversqlite
 	./rivershared
 	./rivertype

internal/maintenance/sqlite_notification_cleaner.go

@@ -60,9 +60,8 @@ func (c *SQLiteNotificationCleanerConfig) mustValidate() *SQLiteNotificationClea
 	return c
 }
 
-// SQLiteNotificationCleaner periodically removes old rows from SQLite's
-// notification outbox. It is only needed for the SQLite driver's emulated
-// listen/notify support.
+// SQLiteNotificationCleaner periodically removes old rows from the durable
+// notification outbox used to emulate listen/notify for SQLite and MySQL.
 type SQLiteNotificationCleaner struct {
 	riversharedmaintenance.QueueMaintainerServiceBase
 	startstop.BaseStartStop
@@ -74,7 +73,7 @@ type SQLiteNotificationCleaner struct {
 	exec riverdriver.Executor
 }
 
-// NewSQLiteNotificationCleaner returns a SQLite notification cleaner.
+// NewSQLiteNotificationCleaner returns a notification outbox cleaner.
 func NewSQLiteNotificationCleaner(archetype *baseservice.Archetype, config *SQLiteNotificationCleanerConfig, exec riverdriver.Executor) *SQLiteNotificationCleaner {
 	return baseservice.Init(archetype, &SQLiteNotificationCleaner{
 		Config: (&SQLiteNotificationCleanerConfig{
@@ -113,7 +112,7 @@ func (s *SQLiteNotificationCleaner) Start(ctx context.Context) error { //nolint:
 			res, err := s.runOnce(ctx)
 			if err != nil {
 				if !errors.Is(err, context.Canceled) {
-					s.Logger.ErrorContext(ctx, s.Name+": Error cleaning SQLite notifications", slog.String("error", err.Error()))
+					s.Logger.ErrorContext(ctx, s.Name+": Error cleaning notification outbox", slog.String("error", err.Error()))
 				}
 				continue
 			}

internal/rivercommon/river_common.go

@@ -46,9 +46,9 @@ const (
 	// MetadataKeyRescueCount records how many times the job has been rescued.
 	MetadataKeyRescueCount = "river:rescue_count"
 
-	// MetadataKeyUniqueNonce is a special metadata key used by the SQLite driver to
-	// determine whether an upsert is was skipped or not because the `(xmax != 0)`
-	// trick we use in Postgres doesn't work in SQLite.
+	// MetadataKeyUniqueNonce is a special metadata key used by the MySQL and
+	// SQLite drivers to determine whether an upsert was skipped because the
+	// `(xmax != 0)` trick used in Postgres isn't available.
 	MetadataKeyUniqueNonce = "river:unique_nonce"
 )
 

job_list_params.go

@@ -180,6 +180,8 @@ type JobListParams struct {
 	where          []dblist.WherePredicate
 }
 
+const jobListMetadataPredicatePostgres = `metadata @> @metadata_fragment::jsonb`
+
 // NewJobListParams creates a new JobListParams to return available jobs sorted
 // by time in ascending order, returning 100 jobs at most.
 func NewJobListParams() *JobListParams {
@@ -278,9 +280,9 @@ func (p *JobListParams) toDBParams() (*dblist.JobListParams, error) {
 		} else {
 			namedArgs["cursor_time"] = p.after.time
 			if sortOrder == dblist.SortOrderAsc {
-				p.where = append(p.where, dblist.WherePredicate{NamedArgs: namedArgs, SQL: fmt.Sprintf(`("%s" > @cursor_time OR ("%s" = @cursor_time AND "id" > @after_id))`, timeField, timeField)})
+				p.where = append(p.where, dblist.WherePredicate{NamedArgs: namedArgs, SQL: fmt.Sprintf(`(%s > @cursor_time OR (%s = @cursor_time AND id > @after_id))`, timeField, timeField)})
 			} else {
-				p.where = append(p.where, dblist.WherePredicate{NamedArgs: namedArgs, SQL: fmt.Sprintf(`("%s" < @cursor_time OR ("%s" = @cursor_time AND "id" < @after_id))`, timeField, timeField)})
+				p.where = append(p.where, dblist.WherePredicate{NamedArgs: namedArgs, SQL: fmt.Sprintf(`(%s < @cursor_time OR (%s = @cursor_time AND id < @after_id))`, timeField, timeField)})
 			}
 		}
 	}
@@ -298,6 +300,16 @@ func (p *JobListParams) toDBParams() (*dblist.JobListParams, error) {
 	}, nil
 }
 
+func (p *JobListParams) withMetadataPredicateSQL(predicateSQL string) *JobListParams {
+	paramsCopy := p.copy()
+	for i := range paramsCopy.where {
+		if paramsCopy.where[i].SQL == jobListMetadataPredicatePostgres {
+			paramsCopy.where[i].SQL = predicateSQL
+		}
+	}
+	return paramsCopy
+}
+
 // After returns an updated filter set that will only return jobs
 // after the given cursor.
 func (p *JobListParams) After(cursor *JobListCursor) *JobListParams {
@@ -359,7 +371,7 @@ func (p *JobListParams) Metadata(json string) *JobListParams {
 	paramsCopy.metadataCalled = true
 	paramsCopy.where = append(paramsCopy.where, dblist.WherePredicate{
 		NamedArgs: map[string]any{"metadata_fragment": json},
-		SQL:       `metadata @> @metadata_fragment::jsonb`,
+		SQL:       jobListMetadataPredicatePostgres,
 	})
 	return paramsCopy
 }

riverdbtest/riverdbtest.go

@@ -481,7 +481,8 @@ type TestTxOpts struct {
 	// run using TestSchema. This is meant for environments where parallelism
 	// doesn't work as well, like SQLite, which will emit "busy" errors when
 	// multiple clients try to share a schema, even when they're in separate
-	// transactions.
+	// transactions. Also applies to MySQL, where InnoDB deadlocks are common
+	// when multiple transactions are sharing a database.
 	DisableSchemaSharing bool
 
 	// IsTestTxHelper should be set to true for if TestTx is being called from

riverdriver/go.sum

@@ -7,8 +7,8 @@ github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY=
 github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE=
 github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
 github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
-github.com/riverqueue/river/rivertype v0.40.0 h1:6sFLhs0OtkxDZ/iyZIUqaQJb/G1krnb2bh9lfXdCimg=
-github.com/riverqueue/river/rivertype v0.40.0/go.mod h1:D1Ad+EaZiaXbQbJcJcfeicXJMBKno0n6UcfKI5Q7DIQ=
+github.com/riverqueue/river/rivertype v0.41.0 h1:dfscvt1asf1PpeeHTMFxQdV1ZfMoReiiMNFCPPfR0as=
+github.com/riverqueue/river/rivertype v0.41.0/go.mod h1:D1Ad+EaZiaXbQbJcJcfeicXJMBKno0n6UcfKI5Q7DIQ=
 github.com/rogpeppe/go-internal v1.12.0 h1:exVL4IDcn6na9z1rAb56Vxr+CgyK3nn3O+epU5NdKM8=
 github.com/rogpeppe/go-internal v1.12.0/go.mod h1:E+RYuTGaKKdloAfM02xzb0FW3Paa99yedzYV+kq4uf4=
 github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U=

riverdriver/river_driver_interface.go

@@ -25,6 +25,7 @@ import (
 const AllQueuesString = "*"
 
 const (
+	DatabaseNameMySQL    = "mysql"
 	DatabaseNamePostgres = "postgres"
 	DatabaseNameSQLite   = "sqlite"
 )
@@ -53,7 +54,7 @@ var (
 type Driver[TTx any] interface {
 	// ArgPlaceholder is the placeholder character used in query positional
 	// arguments, so "$" for "$1", "$2", "$3", etc. This is a "$" for Postgres
-	// and "?" for SQLite.
+	// and "?" for MySQL and SQLite.
 	//
 	// API is not stable. DO NOT USE.
 	ArgPlaceholder() string
@@ -128,6 +129,13 @@ type Driver[TTx any] interface {
 	// API is not stable. DO NOT USE.
 	PoolSet(dbPool any) error
 
+	// SafeIdentifier returns a safely quoted identifier (e.g. a table or
+	// schema name) for use in SQL queries. Each driver quotes using its
+	// native syntax: double quotes for Postgres/SQLite, backticks for MySQL.
+	//
+	// API is not stable. DO NOT USE.
+	SafeIdentifier(ident string) string
+
 	// SQLFragmentColumnIn generates an SQL fragment to be included as a
 	// predicate in a `WHERE` query for the existence of a set of values in a
 	// column like `id IN (...)`. The actual implementation depends on support