diff --git a/.github/workflows/soak_nightly.yml b/.github/workflows/soak_nightly.yml
index 7862cdd0e6..62b832be70 100644
--- a/.github/workflows/soak_nightly.yml
+++ b/.github/workflows/soak_nightly.yml
@@ -49,25 +49,80 @@ permissions:
   contents: read
 
 jobs:
+  # Resolve the scenario matrix BEFORE strategy evaluation ever touches the
+  # dispatch input. Splicing the raw input into fromJSON(format(...)) would
+  # let a crafted value (`a","b`) fan one dispatch into N paid soak jobs,
+  # and a stray quote would kill the run with an opaque strategy error
+  # instead of the allowlist's friendly one — so the input is allowlisted
+  # here (quotes and backslashes can't pass the regex) and the soak job
+  # consumes only this job's output.
+  plan:
+    runs-on: ubuntu-latest
+    outputs:
+      scenarios: ${{ steps.set.outputs.scenarios }}
+    steps:
+      - name: Resolve scenario matrix
+        id: set
+        env:
+          SCENARIO: ${{ github.event.inputs.scenario || 'postgres/orders-soak' }}
+        run: |
+          if [ "${{ github.event_name }}" = "schedule" ]; then
+            # THE ROTATION lives here: the schedule soaks every listed
+            # scenario; adding a connector to the rotation = one more entry
+            # (see SOAK.md "Adding a connector to the rotation").
+            echo 'scenarios=["postgres/orders-soak", "mysql/orders-soak"]' >> "$GITHUB_OUTPUT"
+            exit 0
+          fi
+          # bash [[ =~ ]], NOT `echo | grep`: grep matches per line, so a
+          # multi-line value (reachable via the REST dispatch API) would pass
+          # on its first line and then write attacker-controlled extra lines
+          # into the line-oriented $GITHUB_OUTPUT below — re-assigning
+          # `scenarios` and fanning one dispatch into N paid jobs. Bash's =~
+          # anchors $ to end-of-STRING and no class here admits a newline,
+          # so any embedded newline fails the match.
+          re='^[A-Za-z0-9][A-Za-z0-9_-]*(/[A-Za-z0-9][A-Za-z0-9_-]*)*$'
+          if ! [[ "${SCENARIO}" =~ $re ]]; then
+            echo "::error::Invalid scenario — expected a single path of [A-Za-z0-9_-] segments, e.g. postgres/orders-soak"
+            exit 1
+          fi
+          echo "scenarios=[\"${SCENARIO}\"]" >> "$GITHUB_OUTPUT"
+
   soak:
+    needs: plan
+    strategy:
+      # max-parallel: 1 because one bench at a time, ever — all sessions
+      # share (and destroy) the same shared Terraform stack and fixed-name
+      # session resources. The workflow-level concurrency group serializes
+      # across runs; this serializes within one.
+      #
+      # fail-fast: false so one connector's red night still soaks the rest.
+      max-parallel: 1
+      fail-fast: false
+      matrix:
+        scenario: ${{ fromJSON(needs.plan.outputs.scenarios) }}
     runs-on: ubuntu-latest
     # Provisioning (~15m) + 95m window + teardown (~15m) + slack. Must stay
-    # under the 4h credential session below.
+    # under the 4h credential session below. Per matrix job, so the full
+    # scheduled rotation may take rotation-size × this.
     timeout-minutes: 220
     env:
       REDPANDA_LICENSE_SECRET: redpanda-connect-bench/license
     steps:
       # Before anything credentialed: the dispatch input is available to
       # the whole write-access population (same as soak_pr.yml's /soak
-      # comment), and Task splices {{.scenario}} raw into a shell command —
-      # so it gets the same allowlist soak_pr.yml applies, and an invalid
-      # value kills the job before the provisioner role is even assumed.
+      # comment), and Task splices {{.scenario}} raw into a shell command.
+      # The plan job already allowlisted it, so this is defense in depth —
+      # it re-checks the value each job actually received, and still kills
+      # the job before the provisioner role is even assumed.
       - name: Validate scenario input
         env:
-          SCENARIO: ${{ github.event.inputs.scenario || 'postgres/orders-soak' }}
+          SCENARIO: ${{ matrix.scenario }}
         run: |
-          if ! echo "${SCENARIO}" | grep -Eq '^[A-Za-z0-9][A-Za-z0-9_-]*(/[A-Za-z0-9][A-Za-z0-9_-]*)*$'; then
-            echo "::error::Invalid scenario '${SCENARIO}' — expected a path of [A-Za-z0-9_-] segments, e.g. postgres/orders-soak"
+          # Full-string match, same as the plan job (grep would pass a
+          # multi-line value on its first matching line).
+          re='^[A-Za-z0-9][A-Za-z0-9_-]*(/[A-Za-z0-9][A-Za-z0-9_-]*)*$'
+          if ! [[ "${SCENARIO}" =~ $re ]]; then
+            echo "::error::Invalid scenario — expected a single path of [A-Za-z0-9_-] segments, e.g. postgres/orders-soak"
             exit 1
           fi
 
@@ -91,7 +146,7 @@ jobs:
         id: gate
         shell: bash
         env:
-          SCENARIO: ${{ github.event.inputs.scenario || 'postgres/orders-soak' }}
+          SCENARIO: ${{ matrix.scenario }}
         run: |
           if [ "${{ github.event_name }}" != "schedule" ]; then
             echo "run=true" >> "$GITHUB_OUTPUT"
@@ -156,7 +211,7 @@ jobs:
         env:
           # Environment passing, not ${{ }} substitution into the script —
           # same reasoning as the gate step above and soak_pr.yml.
-          SCENARIO: ${{ github.event.inputs.scenario || 'postgres/orders-soak' }}
+          SCENARIO: ${{ matrix.scenario }}
         run: |
           task aws:bench scenario="$SCENARIO" 2>&1 | tee "$RUNNER_TEMP/soak-run.log"
 
@@ -182,6 +237,7 @@ jobs:
         if: always() && steps.gate.outputs.run == 'true'
         uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7
         with:
-          name: soak-run-log
+          # job-index, not the scenario path: artifact names reject `/`.
+          name: soak-run-log-${{ strategy.job-index }}
           path: ${{ runner.temp }}/soak-run.log
           retention-days: 30
diff --git a/benchmarking/aws/README.md b/benchmarking/aws/README.md
index 800c9725c1..d71738d13e 100644
--- a/benchmarking/aws/README.md
+++ b/benchmarking/aws/README.md
@@ -3,11 +3,11 @@
 Production-shaped benchmarks and soak tests for Redpanda Connect connectors,
 run on real AWS infrastructure in a dedicated, disposable account.
 
-This tree contains the framework core plus the **postgres_cdc** stack — the
-subset needed by the soak pipeline (CON-179 R6). Further connector stacks
-(mysql, sqlserver, oracle, mongodb, dynamodb, iceberg) exist on the original
-development fork and land here with their own PRs, each bringing its
-scenarios and tests.
+This tree contains the framework core plus the **postgres_cdc** and
+**mysql_cdc** stacks — the subset needed by the soak pipeline (CON-179 R6).
+Further connector stacks (sqlserver, oracle, mongodb, dynamodb, iceberg)
+exist on the original development fork and land here with their own PRs,
+each bringing its scenarios and tests.
 
 ## What a run does
 
@@ -90,10 +90,10 @@ and re-enabled after.
 | Path | Role |
 |---|---|
 | `runner/` | Go orchestrator: provision → stage → seed → sweep/soak → results → teardown |
-| `scenarios/postgres/` | bench + soak + PR-comparison scenarios |
-| `seeders/cdc-rows-postgres/` | write-workload generator |
+| `scenarios/<engine>/` | bench + soak + PR-comparison scenarios (postgres, mysql) |
+| `seeders/cdc-rows-<engine>/` | write-workload generators |
 | `terraform/shared/` | per-session VPC, hosts, brokers, results bucket |
-| `terraform/stacks/postgres/` | per-session RDS Postgres |
+| `terraform/stacks/<engine>/` | per-session RDS Postgres / RDS MySQL |
 | `terraform/persistent/` | applied once: dashboards, alarms, OIDC, reaper, archive |
 | `cleanup-lambda/` | the orphan reaper (own Go module) |
 | `SOAK.md` | soak operations runbook |
@@ -101,8 +101,9 @@ and re-enabled after.
 ## Known limitations
 
 - postgres_cdc IAM auth cannot work against vanilla RDS (replication
-  connections reject IAM tokens); the credential-rotation soak window is
-  covered by mysql_cdc when its stack lands.
+  connections reject IAM tokens); the credential-rotation soak window
+  belongs to the mysql_cdc soak, which runs password auth today — the IAM
+  increment is still open (see SOAK.md).
 - One-lane serialization: soaks and benches queue on the shared stack.
   Session-scoped isolation is the tracked scaling path.
 - The weekly 24h soak needs a reaper exemption tag + a non-GitHub conductor
diff --git a/benchmarking/aws/SOAK.md b/benchmarking/aws/SOAK.md
index 1eaeea09a4..47f1b99d88 100644
--- a/benchmarking/aws/SOAK.md
+++ b/benchmarking/aws/SOAK.md
@@ -15,7 +15,7 @@ files named below.
 | Runner soak mode | `runner/` (main.go, matrix.go, cloudwatch.go) | scaled cadences, 10-min S3 checkpoints, per-minute CloudWatch emission, backlog series |
 | Dashboards + alarms | `terraform/persistent/` (`main.tf` `soak_scenarios` var, `alarms.tf`) | one dashboard + three alarms (stall / rss-slope / backlog) per scenario → SNS `redpanda-connect-bench-soak-alerts` |
 | Archive + baseline | `redpanda-connect-bench-soak-archive` bucket | result.json + raw artifacts per run; `soak-index/` feeds the rolling-baseline comparator (advisory < 3 runs, then fails the job on throughput < 85% / RSS > 130% of baseline) |
-| Nightly workflow | `.github/workflows/soak_nightly.yml` | 08:10 UTC cron (arms only from the default branch) + manual dispatch; OIDC creds (4h), license from Secrets Manager, teardown verified against AWS |
+| Nightly workflow | `.github/workflows/soak_nightly.yml` | 08:10 UTC cron over the rotation matrix (postgres, mysql; serialized, arms only from the default branch) + manual dispatch; OIDC creds (4h), license from Secrets Manager, teardown verified against AWS |
 | PR comparison | `.github/workflows/soak_pr.yml` | `/soak` comment (write-access gated) → base-vs-PR binaries, same infra, sticky comparison comment |
 
 ## Adding a connector to the rotation
@@ -32,14 +32,23 @@ files named below.
    in `terraform/persistent/variables.tf` (key → connector + scenario
    name), then `task aws:persistent`. Alarms and the dashboard are
    generated per entry; Slack delivery is on by default via `slack.tf`'s
-   committed IDs — no extra vars needed.
-4. **First runs**: dispatch the nightly workflow manually with the
+   committed IDs — no extra vars needed. **Apply this BEFORE the rotation
+   entry merges** — the cron arms itself the moment the workflow lands on
+   the default branch, and until the persistent apply runs, nothing
+   watches the new connector's metrics.
+4. **Add it to the nightly rotation**: append the scenario path to the
+   schedule list in `soak_nightly.yml`'s plan job. The matrix is
+   serialized (`max-parallel: 1` — one bench at a time) and
+   `fail-fast: false`, so each rotation entry soaks even when another is
+   red; a manual dispatch still runs only the scenario named in the input.
+5. **First runs**: dispatch the nightly workflow manually with the
    scenario input. The baseline comparator stays advisory until three
    soak-index entries exist.
-5. **Optionally add a PR variant** (`*-soak-pr.yaml`): same scenario with
+6. **Optionally add a PR variant** (`*-soak-pr.yaml`): same scenario with
    30m duration and `arms: [{id: base, binary: base}, {id: pr, binary:
    pr}]`. The scenario NAME must differ from the nightly's so its metrics
-   land outside the alarm dimensions.
+   land outside the alarm dimensions. `/soak <engine>/<name>-pr` selects it
+   on a PR (the default remains postgres/orders-soak-pr).
 
 ## Operating it
 
@@ -106,7 +115,7 @@ files named below.
   backend now uses S3-native `use_lockfile` locking), disarmed the orphan
   reaper's schedule rule **every night**, and deleted the stall + backlog
   alarms. The exemption is the `cloud-nuke-excluded = true` tag, applied via
-  `default_tags` in all three stacks — the persistent stack so the reaper
+  `default_tags` in every stack — the persistent stack so the reaper
   schedule and alarms survive, the session stacks so a live bench crossing
   02:25 UTC isn't terminated mid-run. Our OWN reaper keys on `Project`, not
   this tag, so bench cleanup at the 4h TTL is unaffected. Any new resource
@@ -115,8 +124,11 @@ files named below.
 
 - postgres_cdc IAM auth cannot work against vanilla RDS (replication-
   protocol connections reject IAM tokens — verified live 2026-08-12), so
-  the credential-rotation window is covered by a future mysql_cdc soak or
-  Aurora, not the postgres soak.
+  the postgres soak cannot cover the credential-rotation window. The mysql
+  soak is its designated home (RDS IAM tokens ride normal-protocol MySQL
+  connections), but it runs password auth today — the IAM increment (the
+  AWSAuthenticationPlugin DB user in reset, the rds-db:connect policy in
+  the mysql stack, a live validation pass) is still open.
 - The nightly cron only arms once `soak_nightly.yml` is on the repo's
   default branch; until then, manual dispatch.
 - The weekly 24h soak needs two prerequisites before it can exist: a
diff --git a/benchmarking/aws/go.mod b/benchmarking/aws/go.mod
index e447aeab91..eb13c388d3 100644
--- a/benchmarking/aws/go.mod
+++ b/benchmarking/aws/go.mod
@@ -18,6 +18,7 @@ require (
 	github.com/aws/aws-sdk-go-v2/service/secretsmanager v1.44.6
 	github.com/aws/aws-sdk-go-v2/service/ssm v1.73.6
 	github.com/aws/smithy-go v1.27.8
+	github.com/go-sql-driver/mysql v1.10.0
 	github.com/jackc/pgx/v5 v5.10.0
 	github.com/quasilyte/go-ruleguard/dsl v0.3.23
 	github.com/stretchr/testify v1.12.1
@@ -25,6 +26,7 @@ require (
 )
 
 require (
+	filippo.io/edwards25519 v1.2.0 // indirect
 	github.com/aws/aws-sdk-go-v2/aws/protocol/eventstream v1.7.18 // indirect
 	github.com/aws/aws-sdk-go-v2/credentials v1.19.36 // indirect
 	github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.18.37 // indirect
diff --git a/benchmarking/aws/go.sum b/benchmarking/aws/go.sum
index c7eee17ed2..cccab5f944 100644
--- a/benchmarking/aws/go.sum
+++ b/benchmarking/aws/go.sum
@@ -1,3 +1,5 @@
+filippo.io/edwards25519 v1.2.0 h1:crnVqOiS4jqYleHd9vaKZ+HKtHfllngJIiOpNpoJsjo=
+filippo.io/edwards25519 v1.2.0/go.mod h1:xzAOLCNug/yB62zG1bQ8uziwrIqIuxhctzJT18Q77mc=
 github.com/aws/aws-sdk-go-v2 v1.43.6 h1:RrmFcqCBxkJuf7g1axVo5krB4jM/AO8r5e5oujrgdoQ=
 github.com/aws/aws-sdk-go-v2 v1.43.6/go.mod h1:tXpPM+v0D1lndmga+HqqLDIzUFJlEeR21aspVklHF00=
 github.com/aws/aws-sdk-go-v2/aws/protocol/eventstream v1.7.18 h1:LAfOuhAH331fmOjTQpAaOlH+Ftn7RzSDJ2VFwjdMMy4=
@@ -48,6 +50,8 @@ github.com/aws/smithy-go v1.27.8 h1:FR0dxZfIlV7Z8eh2iHfIofdunw382XsDV3Mxt9nUvRY=
 github.com/aws/smithy-go v1.27.8/go.mod h1:YE2RhdIuDbA5E5bTdciG9KrW3+TiEONeUWCqxX9i1Fc=
 github.com/creack/pty v1.1.9/go.mod h1:oKZEueFk5CKHvIhNR5MUki03XCEU+Q6VDXinZuGJ33E=
 github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/go-sql-driver/mysql v1.10.0 h1:Q+1LV8DkHJvSYAdR83XzuhDaTykuDx0l6fkXxoWCWfw=
+github.com/go-sql-driver/mysql v1.10.0/go.mod h1:M+cqaI7+xxXGG9swrdeUIoPG3Y3KCkF0pZej+SK+nWk=
 github.com/jackc/pgpassfile v1.0.0 h1:/6Hmqy13Ss2zCq62VdNG8tM1wchn8zjSGOBJ6icpsIM=
 github.com/jackc/pgpassfile v1.0.0/go.mod h1:CEx0iS5ambNFdcRtxPj5JhEz+xB6uRky5eyVu/W2HEg=
 github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 h1:iCEnooe7UlwOQYpKFhBabPMi4aNAfoODPEFNiAnClxo=
diff --git a/benchmarking/aws/runner/scenario.go b/benchmarking/aws/runner/scenario.go
index 9858df628a..4bb743d3ce 100644
--- a/benchmarking/aws/runner/scenario.go
+++ b/benchmarking/aws/runner/scenario.go
@@ -207,7 +207,7 @@ type engineSpec struct {
 }
 
 // engineSpecs is the registry mechanism a connector's stack PR extends —
-// see the type doc above. mysql_cdc, oracledb_cdc, microsoft_sql_server_cdc,
+// see the type doc above. oracledb_cdc, microsoft_sql_server_cdc,
 // mongodb_cdc, and aws_dynamodb_cdc were trimmed out of this scope-reduced
 // tree (postgres_cdc soak testing only); each returns with its own stack PR.
 var engineSpecs = map[string]engineSpec{
@@ -215,6 +215,19 @@ var engineSpecs = map[string]engineSpec{
 		DSNOutputKey: "postgres_dsn",
 		DSNEnvVar:    "POSTGRES_DSN",
 	},
+	// mysql_cdc connects via a go-sql-driver DSN, but the mariadb CLI the
+	// reset runs through takes discrete -h/-P/-u/-p flags rather than a DSN
+	// URL, so the Reset*OutputKey fields point at the mysql stack's
+	// split-out outputs.
+	"mysql_cdc": {
+		DSNOutputKey:       "mysql_dsn",
+		DSNEnvVar:          "MYSQL_DSN",
+		ResetHostOutputKey: "mysql_host",
+		ResetPortOutputKey: "mysql_port",
+		ResetUserOutputKey: "mysql_user",
+		ResetPassOutputKey: "mysql_password",
+		ResetDBOutputKey:   "mysql_db",
+	},
 }
 
 func engineSpecFor(connector string) (engineSpec, bool) {
diff --git a/benchmarking/aws/runner/scenario_test.go b/benchmarking/aws/runner/scenario_test.go
index 8fcdac93e5..384b691dce 100644
--- a/benchmarking/aws/runner/scenario_test.go
+++ b/benchmarking/aws/runner/scenario_test.go
@@ -97,14 +97,40 @@ func TestEngineSpecFor_Postgres(t *testing.T) {
 	}
 }
 
+func TestEngineSpecFor_MySQL(t *testing.T) {
+	es, ok := engineSpecFor("mysql_cdc")
+	if !ok {
+		t.Fatalf("mysql_cdc should be registered")
+	}
+	if es.DSNOutputKey != "mysql_dsn" {
+		t.Errorf("DSNOutputKey = %q, want mysql_dsn", es.DSNOutputKey)
+	}
+	if es.DSNEnvVar != "MYSQL_DSN" {
+		t.Errorf("DSNEnvVar = %q, want MYSQL_DSN", es.DSNEnvVar)
+	}
+	// The mysql CLI takes discrete flags, not a DSN URL, so the reset
+	// builder needs every split-out output key populated.
+	for name, got := range map[string]string{
+		"ResetHostOutputKey": es.ResetHostOutputKey,
+		"ResetPortOutputKey": es.ResetPortOutputKey,
+		"ResetUserOutputKey": es.ResetUserOutputKey,
+		"ResetPassOutputKey": es.ResetPassOutputKey,
+		"ResetDBOutputKey":   es.ResetDBOutputKey,
+	} {
+		if got == "" {
+			t.Errorf("%s must be set for mysql's discrete-flags reset form", name)
+		}
+	}
+}
+
 func TestEngineSpecFor_Unknown(t *testing.T) {
 	if _, ok := engineSpecFor("kafka_franz_in_disguise"); ok {
 		t.Error("unknown connector should not resolve")
 	}
-	// mysql_cdc, oracledb_cdc, microsoft_sql_server_cdc, mongodb_cdc, and
+	// oracledb_cdc, microsoft_sql_server_cdc, mongodb_cdc, and
 	// aws_dynamodb_cdc were trimmed from the registry in this scope-reduced
-	// (postgres_cdc-only) tree; each returns with its own stack PR.
-	for _, trimmed := range []string{"mysql_cdc", "oracledb_cdc", "microsoft_sql_server_cdc", "mongodb_cdc", "aws_dynamodb_cdc"} {
+	// tree; each returns with its own stack PR.
+	for _, trimmed := range []string{"oracledb_cdc", "microsoft_sql_server_cdc", "mongodb_cdc", "aws_dynamodb_cdc"} {
 		if _, ok := engineSpecFor(trimmed); ok {
 			t.Errorf("%s should not be registered in this scope-reduced tree", trimmed)
 		}
@@ -390,6 +416,41 @@ func TestLoadScenario_OrdersSoak(t *testing.T) {
 	require.NoError(t, s.Validate())
 }
 
+// TestLoadScenario_MySQLOrdersSoak is the same validity gate for the mysql
+// rotation entry: the shipped scenarios/mysql/orders-soak.yaml must load and
+// validate as a soak scenario, not merely parse.
+func TestLoadScenario_MySQLOrdersSoak(t *testing.T) {
+	s, err := LoadScenario("../scenarios/mysql/orders-soak.yaml")
+	require.NoError(t, err)
+	require.True(t, s.Soak)
+	require.Equal(t, "mysql_cdc", s.Connector)
+	require.Equal(t, "mysql", s.Stack)
+	require.Equal(t, []int{2}, s.Matrix.CPUPoints)
+	require.Empty(t, s.Matrix.Arms)
+	require.Equal(t, "c8g.xlarge", s.Infra.Runner.InstanceType)
+	require.NotNil(t, s.Workload)
+	require.Equal(t, 90*time.Minute, s.Workload.Duration)
+	require.Equal(t, 5*time.Minute, s.Workload.Warmup)
+	require.Equal(t, 10000, s.Workload.WriteRatePerSec)
+	// mysql_cdc requires a checkpoint cache resource; the scenario must ship
+	// one or the rendered config fails lint on the runner host, an hour of
+	// provisioning too late.
+	require.Contains(t, s.Pipeline, "cache_resources")
+	require.NoError(t, s.Validate())
+}
+
+// TestLoadScenario_MySQLOrdersSoakPR pins the /soak A/B variant: binary-only
+// arms on the same single-point soak profile as the nightly, under a
+// deliberately different scenario name so its metrics dodge the alarms.
+func TestLoadScenario_MySQLOrdersSoakPR(t *testing.T) {
+	s, err := LoadScenario("../scenarios/mysql/orders-soak-pr.yaml")
+	require.NoError(t, err)
+	require.True(t, s.Soak)
+	require.True(t, s.IsBinaryArmScenario())
+	require.Equal(t, 30*time.Minute, s.Workload.Duration)
+	require.NotEqual(t, "mysql-orders-soak", s.Name)
+}
+
 func TestScenarioValidate_RejectsSoakWithMultipleCPUPoints(t *testing.T) {
 	s := &Scenario{
 		Name: "soak-x", Connector: "postgres_cdc", Stack: "postgres", Soak: true,
diff --git a/benchmarking/aws/runner/scripts.go b/benchmarking/aws/runner/scripts.go
index 151bd04c7d..f4d135b275 100644
--- a/benchmarking/aws/runner/scripts.go
+++ b/benchmarking/aws/runner/scripts.go
@@ -40,7 +40,9 @@ func envVarPrefix(es engineSpec, outs map[string]string) string {
 // shell-safe — bash still expands $, ` and \ inside "...". This is intentional
 // because (a) the RDS modules generate passwords with special=false (alphanumeric
 // only), and (b) reset SQL is hand-authored. If you ever feed user input into
-// these renderers, switch to a real shell-quoter.
+// these renderers, switch to a real shell-quoter. %q also turns a real
+// newline into a literal \n, so combineReset rejects multiline reset SQL
+// outright rather than let it reach the CLI altered.
 
 // renderSeedScript renders the shell script that runs on the load-gen host to
 // pre-seed the source database. The seeder is expected to be staged at
@@ -84,13 +86,47 @@ func combineReset(connector string, steps []ResetStep, outs map[string]string) (
 	sb.WriteString("set -euo pipefail\n")
 	for _, st := range steps {
 		if st.SQL != "" {
-			// DSN form (postgres). The discrete host/port/user/pass/db-flags
-			// form (mysql's `mysql -h ... -P ... -e ...`) was dead once the
-			// registry was trimmed to postgres_cdc only — no remaining
-			// engineSpec entry sets ResetHostOutputKey — and returns with
-			// mysql_cdc's own stack PR.
-			fmt.Fprintf(&sb, `psql %q -v ON_ERROR_STOP=1 -c %q`+"\n",
-				outs[es.DSNOutputKey], st.SQL)
+			// %q renders a newline as the two characters \n, silently
+			// altering multiline SQL before the CLI ever sees it (a comment
+			// line would swallow the rest of the statement). Refuse loudly
+			// at render time rather than corrupt quietly on the paid host.
+			if strings.ContainsAny(st.SQL, "\n\r") {
+				return "", fmt.Errorf("combineReset: reset SQL must be single-line (%%q quoting turns a newline into a literal backslash-n): %q", st.SQL)
+			}
+			if es.ResetHostOutputKey != "" {
+				// Discrete-flags form (mysql): the CLI has no DSN-URL mode.
+				// The binary is `mariadb`, not `mysql` — mariadb1011 on the
+				// runner host guarantees the former, while the deprecated
+				// mysql compat symlink can vanish in a package/AMI refresh.
+				//
+				// Every connection part must be present, or the rendered
+				// command fails only at runtime on the provisioned host
+				// (bash collapses -p"" to a bare -p prompt) — an hour of
+				// paid session too late for a renamed terraform output.
+				for _, k := range []string{
+					es.ResetHostOutputKey, es.ResetPortOutputKey,
+					es.ResetUserOutputKey, es.ResetPassOutputKey,
+					es.ResetDBOutputKey,
+				} {
+					if outs[k] == "" {
+						return "", fmt.Errorf("combineReset: connector %q: terraform output %q is missing or empty", connector, k)
+					}
+				}
+				fmt.Fprintf(&sb, `mariadb -h %q -P %q -u %q -p%q %q -e %q`+"\n",
+					outs[es.ResetHostOutputKey],
+					outs[es.ResetPortOutputKey],
+					outs[es.ResetUserOutputKey],
+					outs[es.ResetPassOutputKey],
+					outs[es.ResetDBOutputKey],
+					st.SQL)
+			} else {
+				// DSN form (postgres). Same render-time presence check.
+				if outs[es.DSNOutputKey] == "" {
+					return "", fmt.Errorf("combineReset: connector %q: terraform output %q is missing or empty", connector, es.DSNOutputKey)
+				}
+				fmt.Fprintf(&sb, `psql %q -v ON_ERROR_STOP=1 -c %q`+"\n",
+					outs[es.DSNOutputKey], st.SQL)
+			}
 		}
 		if st.Bash != "" {
 			sb.WriteString(substitutePlaceholders(st.Bash, outs) + "\n")
diff --git a/benchmarking/aws/runner/scripts_test.go b/benchmarking/aws/runner/scripts_test.go
index 92bb51432b..3bd2e0fa89 100644
--- a/benchmarking/aws/runner/scripts_test.go
+++ b/benchmarking/aws/runner/scripts_test.go
@@ -34,6 +34,24 @@ func TestRenderSeedScript_Postgres(t *testing.T) {
 	}
 }
 
+func TestRenderSeedScript_MySQL(t *testing.T) {
+	s := &Scenario{
+		Connector: "mysql_cdc",
+		Dataset:   DatasetSpec{Tables: []string{"orders"}, RowSizeBytes: 1200, Seeder: "cdc-rows-mysql", InitialRows: 0},
+	}
+	outs := map[string]string{"mysql_dsn": "u:p@tcp(host:3306)/db", "results_bucket": "bucket"}
+	script, err := renderSeedScript(s, outs, "stage/cdc-rows-mysql")
+	if err != nil {
+		t.Fatalf("renderSeedScript: %v", err)
+	}
+	if !strings.Contains(script, `MYSQL_DSN="u:p@tcp(host:3306)/db"`) {
+		t.Errorf("mysql seed script must set MYSQL_DSN; got:\n%s", script)
+	}
+	if !strings.Contains(script, "/opt/bench/cdc-rows-mysql seed") {
+		t.Errorf("mysql seed script must invoke /opt/bench/cdc-rows-mysql seed; got:\n%s", script)
+	}
+}
+
 func TestRenderSeedScript_NoDSN_WithExtraEnvVars(t *testing.T) {
 	// Register a test-only NoDSN engine and clean up after.
 	engineSpecs["aws_dynamodb_cdc_test"] = engineSpec{
@@ -92,6 +110,80 @@ func TestCombineReset_Postgres_DSNForm(t *testing.T) {
 	}
 }
 
+func TestCombineReset_MySQL_DiscreteFlagsForm(t *testing.T) {
+	steps := []ResetStep{{SQL: "TRUNCATE TABLE orders"}}
+	outs := map[string]string{
+		"mysql_dsn":      "u:p@tcp(host:3306)/db",
+		"mysql_host":     "host",
+		"mysql_port":     "3306",
+		"mysql_user":     "bench",
+		"mysql_password": "pw",
+		"mysql_db":       "benchdb",
+	}
+	got, err := combineReset("mysql_cdc", steps, outs)
+	if err != nil {
+		t.Fatalf("combineReset: %v", err)
+	}
+	// The mariadb CLI has no DSN-URL mode: every connection part must be a
+	// discrete flag, and psql must not appear. (`mariadb`, not the
+	// deprecated `mysql` compat symlink — see combineReset.)
+	if !strings.Contains(got, `mariadb -h "host" -P "3306" -u "bench" -p"pw" "benchdb" -e "TRUNCATE TABLE orders"`) {
+		t.Errorf("mysql reset must use discrete -h/-P/-u/-p flags; got:\n%s", got)
+	}
+	if strings.Contains(got, "psql") {
+		t.Errorf("mysql reset must not fall back to psql; got:\n%s", got)
+	}
+}
+
+func TestCombineReset_MySQL_RejectsMissingOutput(t *testing.T) {
+	outs := map[string]string{
+		"mysql_host": "host",
+		"mysql_port": "3306",
+		"mysql_user": "bench",
+		// mysql_password deliberately absent: bash would collapse -p"" to a
+		// bare -p prompt an hour into a paid session.
+		"mysql_db": "benchdb",
+	}
+	_, err := combineReset("mysql_cdc", []ResetStep{{SQL: "SELECT 1"}}, outs)
+	if err == nil || !strings.Contains(err.Error(), "mysql_password") {
+		t.Fatalf("expected a missing-output error naming mysql_password, got %v", err)
+	}
+}
+
+func TestCombineReset_RejectsMultilineSQL(t *testing.T) {
+	outs := map[string]string{"postgres_dsn": "postgres://u:p@host:5432/db"}
+	_, err := combineReset("postgres_cdc", []ResetStep{{SQL: "SELECT 1;\nSELECT 2"}}, outs)
+	if err == nil || !strings.Contains(err.Error(), "single-line") {
+		t.Fatalf("expected a multiline-SQL rejection, got %v", err)
+	}
+}
+
+func TestCombineReset_Postgres_RejectsMissingDSN(t *testing.T) {
+	_, err := combineReset("postgres_cdc", []ResetStep{{SQL: "SELECT 1"}}, map[string]string{})
+	if err == nil || !strings.Contains(err.Error(), "postgres_dsn") {
+		t.Fatalf("expected a missing-output error naming postgres_dsn, got %v", err)
+	}
+}
+
+func TestCombineReset_AppendsTopicCleanup_MySQL(t *testing.T) {
+	outs := map[string]string{
+		"mysql_host":                "host",
+		"mysql_port":                "3306",
+		"mysql_user":                "bench",
+		"mysql_password":            "pw",
+		"mysql_db":                  "benchdb",
+		"bench_session_id":          "sess-abc",
+		"redpanda_broker_endpoints": "10.42.10.10:9092",
+	}
+	got, err := combineReset("mysql_cdc", []ResetStep{{SQL: "TRUNCATE TABLE orders"}}, outs)
+	if err != nil {
+		t.Fatalf("combineReset: %v", err)
+	}
+	if !strings.Contains(got, `"^bench_sess-abc_mysql_cdc_connect$"`) {
+		t.Errorf("expected anchored Connect topic regex for mysql_cdc; got:\n%s", got)
+	}
+}
+
 func TestCombineReset_EmptySteps(t *testing.T) {
 	got, err := combineReset("postgres_cdc", nil, map[string]string{})
 	if err != nil {
diff --git a/benchmarking/aws/scenarios/mysql/orders-soak-pr.yaml b/benchmarking/aws/scenarios/mysql/orders-soak-pr.yaml
new file mode 100644
index 0000000000..434b688eb2
--- /dev/null
+++ b/benchmarking/aws/scenarios/mysql/orders-soak-pr.yaml
@@ -0,0 +1,80 @@
+name: mysql-orders-soak-pr
+description: |
+  PR before/after soak: the same sustained-moderate mysql_cdc load as
+  mysql-orders-soak, run twice on the SAME infrastructure — once with the
+  PR's base build, once with the PR's build — so the only variable is the
+  binary. 30 minutes per arm keeps the whole /soak round under ~2.5h while
+  still long enough for leak slopes and stalls to separate from noise.
+
+  The scenario name deliberately differs from the nightly's: CloudWatch
+  metrics land in a separate series that no alarm matches, and the runner
+  skips both the rolling-baseline comparator and the soak-index entry for
+  binary-arm runs — a deliberate A/B is not a nightly baseline sample.
+
+connector: mysql_cdc
+stack: mysql
+soak: true
+
+infra:
+  source:
+    instance_class: db.r6g.xlarge
+    # 400 GB gp3 floor: below it RDS forbids iops/storage_throughput, and
+    # 12000/500 are the >=400 GB tier's free baseline (see orders-soak.yaml).
+    storage_gb: 400
+    iops: 12000
+    storage_throughput: 500
+    parameters:
+      binlog_format: "ROW"
+      binlog_row_image: "FULL"
+      binlog_checksum: "NONE"
+  runner:
+    instance_type: c8g.xlarge   # 4 vCPU: reservedCores(2) + cpu_points[2] == 4
+
+dataset:
+  initial_rows: 0
+  row_size_bytes: 1200
+  tables: [orders]
+  seeder: cdc-rows-mysql
+
+workload:
+  write_rate_per_sec: 10000
+  duration: 30m
+  warmup: 5m
+
+pipeline:
+  cache_resources:
+    - label: bench_checkpoint
+      memory: {}
+  input:
+    mysql_cdc:
+      flavor: mysql
+      dsn: ${MYSQL_DSN}
+      tls:
+        skip_cert_verify: true   # RDS-internal CA isn't in the runner image
+      stream_snapshot: false
+      tables: [orders]
+      checkpoint_cache: bench_checkpoint
+      checkpoint_key: mysql_bench
+      # >= batching.count, or the default (1024) caps the pipeline at one
+      # batch in flight — see orders-soak.yaml.
+      checkpoint_limit: 10000
+      batching:
+        count: 5000
+        period: 1s
+
+matrix:
+  cpu_points: [2]
+  # Arms differ ONLY by binary — the runner enforces this for soak A/Bs.
+  # The workflow supplies the paths via --binary base=... --binary pr=...
+  arms:
+    - id: base
+      binary: base
+    - id: pr
+      binary: pr
+
+reset:
+  # Binlog retention first (RDS purges backed-up binlogs without it — see
+  # orders-soak.yaml), then a per-arm TRUNCATE so both arms start from the
+  # same bounded table; no slot/publication analogue exists for mysql.
+  - sql: "CALL mysql.rds_set_configuration('binlog retention hours', 24)"
+  - sql: "TRUNCATE TABLE orders"
diff --git a/benchmarking/aws/scenarios/mysql/orders-soak.yaml b/benchmarking/aws/scenarios/mysql/orders-soak.yaml
new file mode 100644
index 0000000000..d31fed1e85
--- /dev/null
+++ b/benchmarking/aws/scenarios/mysql/orders-soak.yaml
@@ -0,0 +1,96 @@
+name: mysql-orders-soak
+description: |
+  Soak: hold Connect's mysql_cdc at a fixed, sustained-moderate load (10K
+  writes/sec ≈ 12 MB/s — roughly 10-15% of the ~100-115 MB/s read ceiling
+  measured for mysql_cdc, see docs history and the benchmarking branch's
+  mysql-orders-cdc sweep) for 90 minutes at a single vCPU point. Same
+  purpose as postgres-orders-soak: catch leaks, stalls, and rotation bugs
+  that a short max-load sweep never runs long enough to hit.
+
+  Password auth for now. The mysql soak is the designated future home of
+  the ~15-min IAM credential-rotation lever (RDS IAM tokens ride
+  normal-protocol MySQL connections, unlike postgres replication — see
+  postgres-orders-soak's description and #4258/#4668), but that arrives as
+  its own increment: IAM needs the AWSAuthenticationPlugin DB user created
+  in reset, the rds-db:connect policy in the stack, and a live validation
+  pass. Everything here works unchanged when that lands — only the DSN,
+  the input's aws block, and reset steps change.
+
+connector: mysql_cdc
+stack: mysql
+soak: true
+
+infra:
+  source:
+    instance_class: db.r6g.xlarge
+    # Same RDS gp3 tiering as postgres: below 400 GB you may not set
+    # iops/storage_throughput at all; at >=400 GB the free baseline is
+    # 12000 / 500 and those are the legal minimums (live-confirmed on the
+    # postgres stack 2026-08-12).
+    storage_gb: 400
+    iops: 12000
+    storage_throughput: 500
+    # Explicit for self-documentation; identical to the stack defaults.
+    # ROW + FULL are required by mysql_cdc, NONE keeps go-mysql compatible
+    # across server versions.
+    parameters:
+      binlog_format: "ROW"
+      binlog_row_image: "FULL"
+      binlog_checksum: "NONE"
+  runner:
+    instance_type: c8g.xlarge   # 4 vCPU: reservedCores(2) + cpu_points[2] == 4
+
+dataset:
+  initial_rows: 0          # reset TRUNCATEs; seed still CREATEs the table
+  row_size_bytes: 1200
+  tables: [orders]
+  seeder: cdc-rows-mysql
+
+workload:
+  write_rate_per_sec: 10000
+  duration: 90m
+  warmup: 5m
+
+pipeline:
+  # mysql_cdc requires a cache resource for its binlog-position checkpoint.
+  # Memory is deliberate: the soak measures one uninterrupted process, and
+  # with stream_snapshot: false a (hypothetical) restart resumes from the
+  # current binlog position anyway.
+  cache_resources:
+    - label: bench_checkpoint
+      memory: {}
+  input:
+    mysql_cdc:
+      flavor: mysql
+      dsn: ${MYSQL_DSN}
+      tls:
+        skip_cert_verify: true   # RDS-internal CA isn't in the runner image; mysql_cdc uses NewTLSField (no `enabled` toggle)
+      stream_snapshot: false
+      tables: [orders]
+      checkpoint_cache: bench_checkpoint
+      checkpoint_key: mysql_bench
+      # Must be >= batching.count: the default (1024) admits an oversized
+      # batch only when nothing is pending, which serializes the pipeline to
+      # exactly one 5000-row batch in flight per produce+ack round trip.
+      checkpoint_limit: 10000
+      batching:
+        count: 5000
+        period: 1s
+
+matrix:
+  cpu_points: [2]
+
+reset:
+  # RDS purges binlog files as soon as they're backed up unless this is
+  # set — backup_retention_period=1 in the module only ENABLES binlog (see
+  # modules/rds-mysql). Without it, any mid-soak reconnect that outlives
+  # the purge finds its checkpointed position gone and dies with ERROR
+  # 1236 — an infra artifact indistinguishable from a connector bug. 24h
+  # comfortably outlives the longest run; the setting is per-instance and
+  # each session provisions a fresh one, so it must be set every run.
+  - sql: "CALL mysql.rds_set_configuration('binlog retention hours', 24)"
+  # TRUNCATE keeps the table bounded across runs on reused infra. Unlike
+  # postgres there is no slot/publication state to clear: with
+  # stream_snapshot: false and a fresh in-memory checkpoint cache, the
+  # connector always starts from the current binlog position.
+  - sql: "TRUNCATE TABLE orders"
diff --git a/benchmarking/aws/seeders/cdc-rows-mysql/main.go b/benchmarking/aws/seeders/cdc-rows-mysql/main.go
new file mode 100644
index 0000000000..767c18e1a3
--- /dev/null
+++ b/benchmarking/aws/seeders/cdc-rows-mysql/main.go
@@ -0,0 +1,52 @@
+// Copyright 2026 Redpanda Data, Inc.
+//
+// Licensed as a Redpanda Enterprise file under the Redpanda Community
+// License (the "License"); you may not use this file except in compliance with
+// the License. You may obtain a copy of the License at
+//
+// https://github.com/redpanda-data/connect/blob/main/licenses/rcl.md
+
+package main
+
+import (
+	"context"
+	"flag"
+	"fmt"
+	"os"
+	"strings"
+	"time"
+)
+
+func main() {
+	if len(os.Args) < 2 {
+		fmt.Fprintln(os.Stderr, "usage: cdc-rows-mysql {seed|workload} [flags]")
+		os.Exit(2)
+	}
+	cmd := os.Args[1]
+	switch cmd {
+	case "seed":
+		fs := flag.NewFlagSet("seed", flag.ExitOnError)
+		tables := fs.String("tables", "orders", "comma-separated table list")
+		rows := fs.Int64("rows", 1_000_000, "rows per table")
+		rowSize := fs.Int("row-size", 1200, "approximate row size in bytes")
+		_ = fs.Parse(os.Args[2:])
+		if err := seed(context.Background(), strings.Split(*tables, ","), *rows, *rowSize); err != nil {
+			fmt.Fprintln(os.Stderr, "seed:", err)
+			os.Exit(1)
+		}
+	case "workload":
+		fs := flag.NewFlagSet("workload", flag.ExitOnError)
+		tables := fs.String("tables", "orders", "comma-separated table list")
+		rowSize := fs.Int("row-size", 1200, "approximate row size in bytes")
+		rate := fs.Int("rate", 5000, "writes per second total across tables")
+		dur := fs.Duration("duration", 15*time.Minute, "total duration")
+		_ = fs.Parse(os.Args[2:])
+		if err := workload(context.Background(), strings.Split(*tables, ","), *rowSize, *rate, *dur); err != nil {
+			fmt.Fprintln(os.Stderr, "workload:", err)
+			os.Exit(1)
+		}
+	default:
+		fmt.Fprintln(os.Stderr, "unknown subcommand:", cmd)
+		os.Exit(2)
+	}
+}
diff --git a/benchmarking/aws/seeders/cdc-rows-mysql/sql.go b/benchmarking/aws/seeders/cdc-rows-mysql/sql.go
new file mode 100644
index 0000000000..3039834f22
--- /dev/null
+++ b/benchmarking/aws/seeders/cdc-rows-mysql/sql.go
@@ -0,0 +1,393 @@
+// Copyright 2026 Redpanda Data, Inc.
+//
+// Licensed as a Redpanda Enterprise file under the Redpanda Community
+// License (the "License"); you may not use this file except in compliance with
+// the License. You may obtain a copy of the License at
+//
+// https://github.com/redpanda-data/connect/blob/main/licenses/rcl.md
+
+// This is the mysql port of cdc-rows-postgres: same worker/tick fairness
+// math (every row and every write/sec accounted for, no truncation), same
+// distinct-payload pool (identical payloads compress away and misreport
+// throughput by 11-17x — see payloadPoolSize), with database/sql +
+// go-sql-driver in place of pgx and `?` placeholders in place of `$n`.
+
+package main
+
+import (
+	"context"
+	"crypto/rand"
+	"database/sql"
+	"encoding/base64"
+	"errors"
+	"fmt"
+	"os"
+	"strings"
+	"sync"
+	"time"
+
+	"github.com/go-sql-driver/mysql"
+)
+
+func openDB(maxConns int) (*sql.DB, error) {
+	cfg, err := mysql.ParseDSN(os.Getenv("MYSQL_DSN"))
+	if err != nil {
+		return nil, err
+	}
+	// pgx (the postgres seeder's driver) caches prepared statements
+	// transparently; go-sql-driver does NOT — without InterpolateParams it
+	// returns driver.ErrSkip for parameterized Exec and database/sql wraps
+	// every insert in a hidden server-side Prepare+Execute+Close, doubling
+	// statement traffic and roughly halving the per-worker ceiling the
+	// 16-worker design assumes. Payloads are base64 ASCII, so client-side
+	// interpolation is injection-safe here.
+	cfg.InterpolateParams = true
+	connector, err := mysql.NewConnector(cfg)
+	if err != nil {
+		return nil, err
+	}
+	db := sql.OpenDB(connector)
+	db.SetMaxOpenConns(maxConns)
+	db.SetMaxIdleConns(maxConns)
+	return db, nil
+}
+
+func seed(ctx context.Context, tables []string, rows int64, rowSize int) error {
+	db, err := openDB(16)
+	if err != nil {
+		return err
+	}
+	defer db.Close()
+
+	for _, table := range tables {
+		if err := ensureTable(ctx, db, table, rowSize); err != nil {
+			return err
+		}
+	}
+	var wg sync.WaitGroup
+	errCh := make(chan error, len(tables))
+	for _, table := range tables {
+		wg.Add(1)
+		go func(t string) {
+			defer wg.Done()
+			errCh <- bulkInsert(ctx, db, t, rows, rowSize)
+		}(table)
+	}
+	wg.Wait()
+	close(errCh)
+	for err := range errCh {
+		if err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+func ensureTable(ctx context.Context, db *sql.DB, table string, rowSize int) error {
+	stmts := []string{
+		"DROP TABLE IF EXISTS " + table,
+		fmt.Sprintf(`CREATE TABLE %s (
+			id          BIGINT AUTO_INCREMENT PRIMARY KEY,
+			created_at  DATETIME(6) NOT NULL DEFAULT CURRENT_TIMESTAMP(6),
+			payload     TEXT NOT NULL
+		) ENGINE=InnoDB`, table),
+	}
+	for _, s := range stmts {
+		if _, err := db.ExecContext(ctx, s); err != nil {
+			return fmt.Errorf("%s: %w", s, err)
+		}
+	}
+	_ = rowSize
+	return nil
+}
+
+// workerRowCounts splits rows across workers as evenly as possible,
+// handing the remainder (rows % workers) one extra row each to the first
+// `remainder` workers. Truncating division alone (rows/workers) silently
+// drops up to workers-1 rows off the total — for rows=10, workers=16 that
+// truncates to 0 per worker and seeds nothing at all — so every row must be
+// accounted for in the returned counts, which always sum to exactly rows.
+func workerRowCounts(rows int64, workers int) []int64 {
+	counts := make([]int64, workers)
+	if workers <= 0 {
+		return counts
+	}
+	base := rows / int64(workers)
+	remainder := rows % int64(workers)
+	for w := range workers {
+		counts[w] = base
+		if int64(w) < remainder {
+			counts[w]++
+		}
+	}
+	return counts
+}
+
+func bulkInsert(ctx context.Context, db *sql.DB, table string, rows int64, rowSize int) error {
+	const workers = 16
+	counts := workerRowCounts(rows, workers)
+	pool := randomPayloadPool(rowSize, payloadPoolSize)
+	start := time.Now()
+	var wg sync.WaitGroup
+	errCh := make(chan error, workers)
+	for w := range workers {
+		wg.Add(1)
+		workerRows := counts[w]
+		go func() {
+			defer wg.Done()
+			// A worker whose share rounded down to zero (e.g. rows=10 spread
+			// over 16 workers) has nothing to insert.
+			if workerRows == 0 {
+				errCh <- nil
+				return
+			}
+			const batchSize = 1000
+			conn, err := db.Conn(ctx)
+			if err != nil {
+				errCh <- err
+				return
+			}
+			defer conn.Close()
+
+			cursor := 0
+			done := int64(0)
+			// The full-size statement is built once and reused for every
+			// full batch. Only the trailing partial batch — whatever is
+			// left after the last full batch — needs a statement sized to
+			// exactly that many rows, otherwise the loop either overshoots
+			// (inserting a full batch when fewer rows remain) or requires
+			// padding args that don't exist.
+			var fullStmt string
+			var fullArgs []any
+			if workerRows >= batchSize {
+				fullStmt = fmt.Sprintf("INSERT INTO %s (created_at, payload) VALUES %s", table, valuesList(batchSize))
+				fullArgs = make([]any, batchSize)
+			}
+			for workerRows-done >= batchSize {
+				fillArgs(fullArgs, pool, &cursor)
+				if _, err := conn.ExecContext(ctx, fullStmt, fullArgs...); err != nil {
+					errCh <- err
+					return
+				}
+				done += batchSize
+			}
+			if remaining := workerRows - done; remaining > 0 {
+				stmt := fmt.Sprintf("INSERT INTO %s (created_at, payload) VALUES %s", table, valuesList(int(remaining)))
+				args := make([]any, remaining)
+				fillArgs(args, pool, &cursor)
+				if _, err := conn.ExecContext(ctx, stmt, args...); err != nil {
+					errCh <- err
+					return
+				}
+			}
+			errCh <- nil
+		}()
+	}
+	wg.Wait()
+	close(errCh)
+	for err := range errCh {
+		if err != nil {
+			return err
+		}
+	}
+	fmt.Printf("seeded %d rows into %s in %s\n", rows, table, time.Since(start))
+	return nil
+}
+
+// ticksPerSecond is the number of 100ms ticker fires per second that each
+// workload worker uses to spread its per-second quota out smoothly instead
+// of bursting it all in one Exec.
+const ticksPerSecond = 10
+
+// perWorkerRate splits a total rows/sec rate across workers as evenly as
+// possible, handing the remainder (rate % workers) one extra row/sec each to
+// the first `remainder` workers. Truncating division alone drops up to
+// workers-1 rows/sec off the total, which is exactly the kind of small,
+// deterministic undershoot that lets a backlog metric drift forever on an
+// otherwise healthy run.
+func perWorkerRate(rate, workers int) []int {
+	rates := make([]int, workers)
+	if workers <= 0 {
+		return rates
+	}
+	base := rate / workers
+	remainder := rate % workers
+	for w := range workers {
+		rates[w] = base
+		if w < remainder {
+			rates[w]++
+		}
+	}
+	return rates
+}
+
+// tickCounts splits one worker's per-second quota across the ticksPerSecond
+// ticks of its 100ms ticker, again distributing the remainder so the ticks
+// sum to exactly ratePerWorker. This mirrors perWorkerRate one level down;
+// without it, a worker's per-tick truncation (ratePerWorker/10) compounds
+// with the per-worker truncation above into a rate that's measurably below
+// what was requested.
+func tickCounts(ratePerWorker int) [ticksPerSecond]int {
+	var counts [ticksPerSecond]int
+	base := ratePerWorker / ticksPerSecond
+	remainder := ratePerWorker % ticksPerSecond
+	for t := range ticksPerSecond {
+		counts[t] = base
+		if t < remainder {
+			counts[t]++
+		}
+	}
+	return counts
+}
+
+func workload(ctx context.Context, tables []string, rowSize, rate int, dur time.Duration) error {
+	// A single goroutine driving large per-tick batches caps around 30-40K
+	// inserts/sec on c8g.large because statement parsing + one network RTT
+	// per tick eats the budget. Spread across workers, each with a smaller
+	// batch, so the scenario's write_rate_per_sec is actually achievable.
+	// 16 workers handles 150K writes/sec comfortably (each worker ~9.4K/sec,
+	// well under the per-worker ceiling).
+	const workers = 16
+	db, err := openDB(workers)
+	if err != nil {
+		return err
+	}
+	defer db.Close()
+
+	rates := perWorkerRate(rate, workers)
+	deadline := time.Now().Add(dur)
+	var wg sync.WaitGroup
+	errCh := make(chan error, workers)
+	for w := range workers {
+		wg.Add(1)
+		workerIdx := w
+		ratePerWorker := rates[w]
+		go func() {
+			defer wg.Done()
+			// Distinct payloads (built once) so change events aren't trivially
+			// compressible — see payloadPoolSize.
+			pool := randomPayloadPool(rowSize, payloadPoolSize)
+			cursor := 0
+			counts := tickCounts(ratePerWorker)
+
+			// Ticks only ever need one of two row counts (base or base+1),
+			// so the VALUES clauses are built once here instead of on every
+			// tick; a size of 0 is skipped since such a tick issues no Exec.
+			base := ratePerWorker / ticksPerSecond
+			remainder := ratePerWorker % ticksPerSecond
+			var baseValues, plusValues string
+			var baseArgs, plusArgs []any
+			if base > 0 {
+				baseValues = valuesList(base)
+				baseArgs = make([]any, base)
+			}
+			if remainder > 0 {
+				plusValues = valuesList(base + 1)
+				plusArgs = make([]any, base+1)
+			}
+
+			ticker := time.NewTicker(100 * time.Millisecond)
+			defer ticker.Stop()
+			tIdx := workerIdx
+			tickInSecond := 0
+			for {
+				select {
+				case <-ctx.Done():
+					errCh <- ctx.Err()
+					return
+				case <-ticker.C:
+					if time.Now().After(deadline) {
+						errCh <- nil
+						return
+					}
+					// The table rotates every tick regardless of whether
+					// this tick has any rows to insert, so the rotation
+					// stays in lockstep across workers.
+					table := tables[tIdx%len(tables)]
+					tIdx++
+					count := counts[tickInSecond]
+					tickInSecond = (tickInSecond + 1) % ticksPerSecond
+					if count == 0 {
+						continue
+					}
+					values, args := baseValues, baseArgs
+					if count == base+1 {
+						values, args = plusValues, plusArgs
+					}
+					stmt := fmt.Sprintf("INSERT INTO %s (created_at, payload) VALUES %s", table, values)
+					fillArgs(args, pool, &cursor)
+					if _, err := db.ExecContext(ctx, stmt, args...); err != nil {
+						errCh <- err
+						return
+					}
+				}
+			}
+		}()
+	}
+	wg.Wait()
+	close(errCh)
+	for err := range errCh {
+		if err != nil && !errors.Is(err, context.Canceled) && !errors.Is(err, context.DeadlineExceeded) {
+			return err
+		}
+	}
+	return nil
+}
+
+// payloadPoolSize is the number of distinct random payloads cycled per worker.
+//
+// Ported from cdc-rows-mongodb, the only seeder that originally had this right.
+// Reusing ONE identical payload for every row makes each producer batch
+// trivially compressible, and that alone accounted for the 11-17x gap between
+// Connect's self-reported throughput and the broker's byte counters across the
+// postgres, mysql, oracle and sqlserver benches. Mongo's calibration note is the
+// authority on the size: 4096 comfortably exceeds one compression batch, where
+// 1024 still left ~1.5x compressible.
+const payloadPoolSize = 4096
+
+// valuesList builds a multi-row VALUES clause with ONE PLACEHOLDER PER ROW:
+// (NOW(6),?),(NOW(6),?),...
+//
+// Distinct payloads need distinct placeholders (see cdc-rows-postgres, whose
+// single-repeated-placeholder bug this port never inherits). MySQL's
+// per-statement limit is max_allowed_packet, not a placeholder count; n=1000
+// rows of ~1.2 KB is ~1.2 MB, inside RDS's 64 MB default.
+func valuesList(n int) string {
+	var sb strings.Builder
+	for i := range n {
+		if i > 0 {
+			sb.WriteString(",")
+		}
+		sb.WriteString("(NOW(6),?)")
+	}
+	return sb.String()
+}
+
+// fillArgs refills args from the pool, advancing the cursor. Refilling every
+// batch matters: filling once would make each batch byte-identical to the last,
+// which compresses just as well as a single repeated payload did.
+func fillArgs(args []any, pool []string, cursor *int) {
+	for i := range args {
+		args[i] = pool[*cursor%len(pool)]
+		*cursor++
+	}
+}
+
+// randomPayloadPool builds n distinct random payloads of ~size bytes.
+func randomPayloadPool(size, n int) []string {
+	pool := make([]string, n)
+	for i := range pool {
+		pool[i] = randomPayload(size)
+	}
+	return pool
+}
+
+func randomPayload(size int) string {
+	b := make([]byte, (size*3)/4+1)
+	_, _ = rand.Read(b)
+	s := base64.StdEncoding.EncodeToString(b)
+	if len(s) > size {
+		s = s[:size]
+	}
+	return s
+}
diff --git a/benchmarking/aws/seeders/cdc-rows-mysql/sql_test.go b/benchmarking/aws/seeders/cdc-rows-mysql/sql_test.go
new file mode 100644
index 0000000000..2625d3c09c
--- /dev/null
+++ b/benchmarking/aws/seeders/cdc-rows-mysql/sql_test.go
@@ -0,0 +1,153 @@
+// Copyright 2026 Redpanda Data, Inc.
+//
+// Licensed as a Redpanda Enterprise file under the Redpanda Community
+// License (the "License"); you may not use this file except in compliance with
+// the License. You may obtain a copy of the License at
+//
+// https://github.com/redpanda-data/connect/blob/main/licenses/rcl.md
+
+package main
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestWorkerRowCounts(t *testing.T) {
+	tests := []struct {
+		name    string
+		rows    int64
+		workers int
+		want    int64 // expected sum, checked below alongside per-worker invariants
+	}{
+		{name: "evenly divisible", rows: 1_000_000, workers: 16, want: 1_000_000},
+		{name: "fewer rows than workers", rows: 10, workers: 16, want: 10},
+		{name: "single row", rows: 1, workers: 16, want: 1},
+		{name: "zero rows", rows: 0, workers: 16, want: 0},
+		{name: "not evenly divisible", rows: 1_000_001, workers: 16, want: 1_000_001},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			counts := workerRowCounts(tt.rows, tt.workers)
+			require.Len(t, counts, tt.workers)
+
+			var sum int64
+			for _, c := range counts {
+				require.GreaterOrEqual(t, c, int64(0))
+				sum += c
+			}
+			require.Equal(t, tt.want, sum)
+
+			// The spread must never exceed one row between the largest and
+			// smallest share, otherwise the remainder wasn't distributed
+			// evenly (e.g. it was dumped entirely on the first worker).
+			lo, hi := counts[0], counts[0]
+			for _, c := range counts {
+				if c < lo {
+					lo = c
+				}
+				if c > hi {
+					hi = c
+				}
+			}
+			require.LessOrEqual(t, hi-lo, int64(1))
+		})
+	}
+}
+
+func TestWorkerRowCounts_ExactShares(t *testing.T) {
+	// rows=10, workers=16: the first 10 workers get exactly 1 row each and
+	// the rest get 0 — this is the case that used to seed zero rows total.
+	counts := workerRowCounts(10, 16)
+	for w, c := range counts {
+		if w < 10 {
+			require.Equal(t, int64(1), c, "worker %d", w)
+		} else {
+			require.Equal(t, int64(0), c, "worker %d", w)
+		}
+	}
+}
+
+func TestPerWorkerRate(t *testing.T) {
+	tests := []struct {
+		name    string
+		rate    int
+		workers int
+		want    int
+	}{
+		{name: "declared soak rate", rate: 10_000, workers: 16, want: 10_000},
+		{name: "rate below worker count", rate: 3, workers: 16, want: 3},
+		{name: "large rate", rate: 160_000, workers: 16, want: 160_000},
+		{name: "zero rate", rate: 0, workers: 16, want: 0},
+		{name: "not evenly divisible", rate: 10_001, workers: 16, want: 10_001},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			rates := perWorkerRate(tt.rate, tt.workers)
+			require.Len(t, rates, tt.workers)
+
+			var sum int
+			for _, r := range rates {
+				require.GreaterOrEqual(t, r, 0)
+				sum += r
+			}
+			require.Equal(t, tt.want, sum)
+		})
+	}
+}
+
+func TestTickCounts(t *testing.T) {
+	tests := []struct {
+		name          string
+		ratePerWorker int
+	}{
+		{name: "evenly divisible by ticks", ratePerWorker: 625},
+		{name: "less than one tick", ratePerWorker: 1},
+		{name: "zero", ratePerWorker: 0},
+		{name: "large per-worker rate", ratePerWorker: 10_000},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			counts := tickCounts(tt.ratePerWorker)
+			require.Len(t, counts, ticksPerSecond)
+
+			var sum int
+			for _, c := range counts {
+				require.GreaterOrEqual(t, c, 0)
+				sum += c
+			}
+			require.Equal(t, tt.ratePerWorker, sum)
+		})
+	}
+}
+
+// TestWorkloadRateEndToEnd pins the exact totals required by the soak
+// scenarios: distributing a declared rate across workers and then across
+// each worker's 10 ticks/sec must land on the declared rate exactly, with
+// no compounding truncation.
+func TestWorkloadRateEndToEnd(t *testing.T) {
+	tests := []struct {
+		name    string
+		rate    int
+		workers int
+	}{
+		{name: "shipped soak rate", rate: 10_000, workers: 16},
+		{name: "rate below worker count", rate: 3, workers: 16},
+		{name: "large rate", rate: 160_000, workers: 16},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			rates := perWorkerRate(tt.rate, tt.workers)
+
+			var total int
+			for _, ratePerWorker := range rates {
+				counts := tickCounts(ratePerWorker)
+				for _, c := range counts {
+					total += c
+				}
+			}
+			require.Equal(t, tt.rate, total)
+		})
+	}
+}
diff --git a/benchmarking/aws/terraform/modules/rds-mysql/main.tf b/benchmarking/aws/terraform/modules/rds-mysql/main.tf
new file mode 100644
index 0000000000..814fe8bbce
--- /dev/null
+++ b/benchmarking/aws/terraform/modules/rds-mysql/main.tf
@@ -0,0 +1,85 @@
+resource "aws_db_subnet_group" "this" {
+  name       = "${var.name_prefix}-my"
+  subnet_ids = var.subnet_ids
+}
+
+resource "aws_security_group" "this" {
+  name        = "${var.name_prefix}-my-sg"
+  description = "Allow MySQL from bench clients"
+  vpc_id      = var.vpc_id
+
+  dynamic "ingress" {
+    for_each = var.client_sg_ids
+    content {
+      from_port       = 3306
+      to_port         = 3306
+      protocol        = "tcp"
+      security_groups = [ingress.value]
+    }
+  }
+
+  egress {
+    from_port   = 0
+    to_port     = 0
+    protocol    = "-1"
+    cidr_blocks = ["0.0.0.0/0"]
+  }
+}
+
+resource "aws_db_parameter_group" "this" {
+  name   = "${var.name_prefix}-my"
+  family = "mysql8.0"
+  dynamic "parameter" {
+    for_each = var.parameters
+    content {
+      name         = parameter.key
+      value        = parameter.value
+      apply_method = "pending-reboot"
+    }
+  }
+}
+
+resource "random_password" "master" {
+  length  = 20
+  special = false
+}
+
+resource "aws_db_instance" "this" {
+  identifier             = "${var.name_prefix}-my"
+  engine                 = "mysql"
+  engine_version         = var.engine_version
+  instance_class         = var.instance_class
+  allocated_storage      = var.storage_gb
+  storage_type           = "gp3"
+  iops                   = var.iops
+  storage_throughput     = var.storage_throughput
+  db_name                = var.db_name
+  username               = var.master_username
+  password               = random_password.master.result
+  parameter_group_name   = aws_db_parameter_group.this.name
+  db_subnet_group_name   = aws_db_subnet_group.this.name
+  vpc_security_group_ids = [aws_security_group.this.id]
+  skip_final_snapshot    = true
+  deletion_protection    = false
+  publicly_accessible    = false
+  apply_immediately      = true
+
+  # CRITICAL: backup_retention_period > 0 is what ENABLES binlog on RDS
+  # MySQL at all — with backups off there is no binlog and mysql_cdc has
+  # nothing to read. It does NOT retain the binlog: RDS purges binlog files
+  # as soon as they're backed up unless 'binlog retention hours' is set,
+  # and that is a runtime stored procedure
+  # (CALL mysql.rds_set_configuration), not a parameter-group knob, so the
+  # scenarios' reset steps call it on every run (see scenarios/mysql/).
+  backup_retention_period = 1
+  # Pin the backup window to off-hours UTC. Without this, RDS picks a random
+  # daily slot; the 2026-05-21 smoke saw the 8 vCPU sweep point degrade from
+  # ~100 MB/s to ~60 MB/s mid-window because a backup overlapped (gp3
+  # throughput is shared between user writes and snapshot copy). 06:00-08:00
+  # UTC is well before US-Pacific working hours when benches typically run,
+  # and ends before the nightly soak's 08:10 UTC cron even fires — a nightly
+  # soak instance (created ~08:15, destroyed ~2.5h later) never lives through
+  # the window at all, only through the unavoidable on-creation snapshot,
+  # which lands during provisioning rather than the measured run.
+  backup_window = "06:00-08:00"
+}
diff --git a/benchmarking/aws/terraform/modules/rds-mysql/outputs.tf b/benchmarking/aws/terraform/modules/rds-mysql/outputs.tf
new file mode 100644
index 0000000000..306c468d56
--- /dev/null
+++ b/benchmarking/aws/terraform/modules/rds-mysql/outputs.tf
@@ -0,0 +1,16 @@
+output "mysql_dsn" {
+  # go-sql-driver/mysql DSN: user:pass@tcp(host:port)/db?params
+  # parseTime=true maps DATETIME → time.Time at the driver layer.
+  # tls=skip-verify because the RDS-internal CA isn't in the runner image.
+  value     = "${var.master_username}:${random_password.master.result}@tcp(${aws_db_instance.this.address}:3306)/${var.db_name}?parseTime=true&tls=skip-verify"
+  sensitive = true
+}
+output "mysql_endpoint" { value = aws_db_instance.this.address }
+output "mysql_host" { value = aws_db_instance.this.address }
+output "mysql_port" { value = "3306" }
+output "mysql_user" { value = var.master_username }
+output "mysql_db" { value = var.db_name }
+output "mysql_password" {
+  value     = random_password.master.result
+  sensitive = true
+}
diff --git a/benchmarking/aws/terraform/modules/rds-mysql/variables.tf b/benchmarking/aws/terraform/modules/rds-mysql/variables.tf
new file mode 100644
index 0000000000..2693602353
--- /dev/null
+++ b/benchmarking/aws/terraform/modules/rds-mysql/variables.tf
@@ -0,0 +1,51 @@
+variable "name_prefix" { type = string }
+variable "vpc_id" { type = string }
+variable "subnet_ids" {
+  type = list(string)
+}
+variable "client_sg_ids" {
+  type        = list(string)
+  description = "SGs allowed to connect on 3306"
+}
+variable "instance_class" {
+  type    = string
+  default = "db.r6g.2xlarge"
+}
+variable "storage_gb" {
+  type    = number
+  default = 400
+}
+variable "iops" {
+  type    = number
+  default = 12000
+}
+variable "engine_version" {
+  type    = string
+  default = "8.0.46"
+}
+variable "db_name" {
+  type    = string
+  default = "benchdb"
+}
+variable "master_username" {
+  type    = string
+  default = "bench"
+}
+variable "parameters" {
+  type = map(string)
+  # binlog_format=ROW + binlog_row_image=FULL are required by mysql_cdc.
+  # binlog_checksum=NONE keeps the go-mysql client compatible across versions.
+  default = {
+    binlog_format    = "ROW"
+    binlog_row_image = "FULL"
+    binlog_checksum  = "NONE"
+  }
+}
+variable "storage_throughput" {
+  # gp3 throughput in MiB/s. null = RDS default. Same tiering as postgres:
+  # below 400 GB RDS forbids setting iops/storage_throughput at all; at
+  # >=400 GB the free baseline is 12000 IOPS / 500 MiB/s and those are the
+  # legal minimums (see modules/rds-postgres and the postgres soak scenario).
+  type    = number
+  default = null
+}
diff --git a/benchmarking/aws/terraform/persistent/variables.tf b/benchmarking/aws/terraform/persistent/variables.tf
index ae69b0e909..a5eea2c404 100644
--- a/benchmarking/aws/terraform/persistent/variables.tf
+++ b/benchmarking/aws/terraform/persistent/variables.tf
@@ -21,5 +21,9 @@ variable "soak_scenarios" {
       connector = "postgres_cdc"
       scenario  = "postgres-orders-soak"
     }
+    mysql = {
+      connector = "mysql_cdc"
+      scenario  = "mysql-orders-soak"
+    }
   }
 }
diff --git a/benchmarking/aws/terraform/stacks/mysql/main.tf b/benchmarking/aws/terraform/stacks/mysql/main.tf
new file mode 100644
index 0000000000..a0e5d2b2ee
--- /dev/null
+++ b/benchmarking/aws/terraform/stacks/mysql/main.tf
@@ -0,0 +1,54 @@
+terraform {
+  required_version = ">= 1.10" # S3-native state locking (use_lockfile in backend.hcl)
+  required_providers {
+    aws    = { source = "hashicorp/aws", version = "~> 5.70" }
+    random = { source = "hashicorp/random", version = "~> 3.6" }
+  }
+  backend "s3" {}
+}
+
+provider "aws" {
+  region = var.region
+  default_tags {
+    tags = {
+      Project   = "redpanda-connect-bench"
+      Stack     = "mysql"
+      ManagedBy = "terraform"
+      # The cleanup lambda derives creation time for RDS subnet/parameter
+      # groups and security groups EXCLUSIVELY from this tag (see
+      # cleanup-lambda/sweep.go sessionCreatedAt) — those resources carry no
+      # creation-time field of their own. Without it, an aborted teardown
+      # leaves this stack's groups unsweepable forever, and the surviving
+      # security group blocks the shared VPC's deletion on every sweep.
+      "bench-session-id" = var.bench_session_id
+      # See shared/main.tf: exempt from the org cloud-nuke sweep; our own
+      # reaper (keyed on Project) still reaps this stack at the 4h TTL.
+      "cloud-nuke-excluded" = "true"
+    }
+  }
+}
+
+data "terraform_remote_state" "shared" {
+  backend = "s3"
+  config = {
+    bucket = "redpanda-connect-bench-tfstate"
+    region = var.region
+    key    = "shared/terraform.tfstate"
+  }
+}
+
+module "rds" {
+  source      = "../../modules/rds-mysql"
+  name_prefix = "rpcn-bench-my"
+  vpc_id      = data.terraform_remote_state.shared.outputs.vpc_id
+  subnet_ids  = data.terraform_remote_state.shared.outputs.private_subnet_ids
+  client_sg_ids = [
+    data.terraform_remote_state.shared.outputs.runner_sg_id,
+    data.terraform_remote_state.shared.outputs.load_gen_sg_id,
+  ]
+  instance_class     = var.instance_class
+  storage_gb         = var.storage_gb
+  iops               = var.iops
+  storage_throughput = var.storage_throughput
+  parameters         = var.parameters
+}
diff --git a/benchmarking/aws/terraform/stacks/mysql/outputs.tf b/benchmarking/aws/terraform/stacks/mysql/outputs.tf
new file mode 100644
index 0000000000..7984b53384
--- /dev/null
+++ b/benchmarking/aws/terraform/stacks/mysql/outputs.tf
@@ -0,0 +1,20 @@
+output "mysql_dsn" {
+  value     = module.rds.mysql_dsn
+  sensitive = true
+}
+output "mysql_endpoint" { value = module.rds.mysql_endpoint }
+# Discrete connection parts for the reset builder: the mysql CLI takes
+# -h/-P/-u/-p flags rather than a DSN URL (see combineReset in the runner).
+output "mysql_host" { value = module.rds.mysql_host }
+output "mysql_port" { value = module.rds.mysql_port }
+output "mysql_user" { value = module.rds.mysql_user }
+output "mysql_db" { value = module.rds.mysql_db }
+output "mysql_password" {
+  value     = module.rds.mysql_password
+  sensitive = true
+}
+# Exposed so scenarios can reference ${REGION} instead of hardcoding the
+# bench account's region (same contract as the postgres stack).
+output "region" {
+  value = var.region
+}
diff --git a/benchmarking/aws/terraform/stacks/mysql/variables.tf b/benchmarking/aws/terraform/stacks/mysql/variables.tf
new file mode 100644
index 0000000000..122a01b9a6
--- /dev/null
+++ b/benchmarking/aws/terraform/stacks/mysql/variables.tf
@@ -0,0 +1,34 @@
+variable "region" {
+  type    = string
+  default = "us-east-2"
+}
+variable "bench_session_id" {
+  # Runner-generated session ID ("bench-YYYYMMDD-HHMMSS", see newSessionID).
+  # Stamped on every resource via default_tags; the cleanup lambda decodes
+  # the embedded timestamp as the only age signal for resources whose
+  # Describe response carries no creation time. Same contract as the shared
+  # stack's variable of the same name, including the empty default: `runner
+  # down` rebuilds stack vars from the scenario's infra.source only, so a
+  # required variable would abort the destroy (-input=false) and strand paid
+  # infra until the reaper's TTL. The tag value doesn't matter during destroy.
+  type    = string
+  default = ""
+}
+variable "instance_class" { type = string }
+variable "storage_gb" { type = number }
+variable "iops" { type = number }
+variable "parameters" {
+  type = map(string)
+  # binlog_format=ROW + binlog_row_image=FULL are what mysql_cdc requires;
+  # binlog_checksum=NONE keeps go-mysql compatible across server versions.
+  default = {
+    binlog_format    = "ROW"
+    binlog_row_image = "FULL"
+    binlog_checksum  = "NONE"
+  }
+}
+variable "storage_throughput" {
+  # gp3 throughput MiB/s; null = RDS default. See modules/rds-mysql.
+  type    = number
+  default = null
+}