diff --git a/docs/testing.md b/docs/testing.md index 6babf9e..20178da 100644 --- a/docs/testing.md +++ b/docs/testing.md @@ -204,10 +204,18 @@ mirrors. ### How much of the suite runs on Ministack -Deliberately almost none: one test, three subtests sharing one provisioned +Deliberately almost none: one test whose subtests share one provisioned cluster ([ministack_integration_test.go](../internal/testutil/ministack_integration_test.go)) -— provisioning costs minutes, so the tier provisions once and orders the -password rotation last. Each subtest pins one AWS seam: +— provisioning costs minutes, so the tier provisions once. The harness +enables Ministack's cluster replication (`MINISTACK_RDS_PG_CLUSTER_REPLICATION`), +which adds a second, reader member to the cluster: the reader and stop/start +subtests need a real standby, and the price is a longer provisioning wait +while the emulator brings both members to `available`. Two subtest +orderings are load-bearing, and the test documents them where it fixes the +order: stop/start restarts cluster compute and waits for every member to +return before the metadata failover, which targets the reader member; the +password rotation runs last because it changes the shared cluster's master +password. Each subtest pins one AWS seam: - **provision & connect** — provision → instance `available` → endpoint discovery → `dbconn` connect → PG-major assertion → DDL smoke; @@ -221,7 +229,23 @@ password rotation last. Each subtest pins one AWS seam: - **password rotation** — RDS-managed password generation, rotation, and Secrets Manager resolution, plus what a rotation does to a running schema change (see below) and pg-sprite's contract that the resulting auth failure - is terminal, not retryable. + is terminal, not retryable; +- **reader read-only boundary** — a real streaming-replication standby accepts + catalog preflight reads, refuses writes with SQLSTATE `25006`, and the + connection layer classifies that refusal as terminal; +- **stop/start resilience** — stopping real cluster compute interrupts active + DDL, and the test pins the *cause*, not just the failure: the backend's + `admin_shutdown` / `crash_shutdown` SQLSTATE or a connection-level error + carrying no server response — never `query_canceled` from the session's own + `statement_timeout`, which would pass a test that proved nothing about the + stop. The interruption is terminal for `dbconn.Retryable`, so the engine's + outcome is the fail-closed `execution-failed`. Starting compute again + restores every member, the interrupted DDL has left no trace in the + catalog, and a fresh schema change succeeds; +- **metadata failover keeps the writer session** — `FailoverDBCluster` flips + the API-visible writer and reports `failing-over` while an established + writer transaction remains usable. Ministack does not yet promote the + standby at the data plane, so the test deliberately makes no such claim. ### What a password rotation does to a running schema change @@ -291,7 +315,7 @@ stays unit-only so pushes remain fast. | Verify-full TLS against a live TLS-only server | [pkg/dbconn/tls_integration_test.go](../pkg/dbconn/tls_integration_test.go) | | Targeted blocker termination | [pkg/dbconn/dbconn_integration_test.go](../pkg/dbconn/dbconn_integration_test.go) | | Test harness self-checks | [internal/testutil](../internal/testutil/postgres_test.go) | -| RDS control-plane provisioning → endpoint discovery → `dbconn` connect, error contract, password-rotation behavior (Ministack) | [internal/testutil/ministack_integration_test.go](../internal/testutil/ministack_integration_test.go) | +| RDS control-plane provisioning → endpoint discovery → `dbconn` connect, reader read-only behavior, stop/start resilience, metadata failover, error contract, password rotation (Ministack) | [internal/testutil/ministack_integration_test.go](../internal/testutil/ministack_integration_test.go) | | Parse boundary, typed operations, and advisory rewrites | [pkg/statement](../pkg/statement/statement_test.go), [operation tests](../pkg/statement/ops_test.go) | | Native / copy-and-swap / refuse classification and safer SQL | [pkg/planner](../pkg/planner/planner_test.go) | | Backend routing and copy-and-swap unavailable disposition | [pkg/router](../pkg/router/router_test.go) | diff --git a/internal/testutil/ministack.go b/internal/testutil/ministack.go index b8be00a..cf7eaea 100644 --- a/internal/testutil/ministack.go +++ b/internal/testutil/ministack.go @@ -78,8 +78,8 @@ const ( fixtureUser = "pgsprite" fixtureDatabase = "pgsprite" // awsAccountID and awsRegion identify the emulator's default account. - // Ministack scopes the sibling container's name by - // sha1(account:region), so these also feed siblingHostAddr. + // Ministack scopes each database container's name by + // sha1(account:region), so these also feed memberHostAddr. awsAccountID = "000000000000" awsRegion = "us-east-1" ) @@ -94,7 +94,7 @@ func ministackImage() string { if img := os.Getenv("MINISTACK_IMAGE"); img != "" { return img } - return "ministackorg/ministack:1.4.15-full" + return "ministackorg/ministack:1.5.6-full" } // auroraEngineVersion returns a real aurora-postgresql engine version for @@ -134,11 +134,15 @@ type AuroraCluster struct { ClusterID string // InstanceID is the DBInstanceIdentifier of the cluster's sole instance. InstanceID string + // ReaderInstanceID is the cluster member backed by a real PostgreSQL + // hot standby when the harness enables Ministack's replication mode. + ReaderInstanceID string // addr is the host:port the test connects to — the discovered cluster // endpoint when reachable, otherwise the sibling container's // host-published address (see ProvisionAuroraPostgres). - addr string + addr string + readerAddr string // password is the master password the cluster currently accepts. // Rotate keeps it in sync with the control plane so URL never goes // silently stale after a rotation. @@ -155,6 +159,11 @@ func (c *AuroraCluster) URL() string { return c.urlWithPassword(c.password) } +// ReaderURL returns a connection URL for the cluster's read-only standby. +func (c *AuroraCluster) ReaderURL() string { + return c.urlAt(c.readerAddr, c.password) +} + // urlWithPassword returns a connection URL using the given master // password. The password is RDS-generated — the harness does not choose // it — so it may contain URL-reserved characters; the URL is assembled @@ -166,10 +175,14 @@ func (c *AuroraCluster) URL() string { // so the production TLS path is out of scope for this tier (it is // proven by pkg/dbconn's TLS integration tests). func (c *AuroraCluster) urlWithPassword(password string) string { + return c.urlAt(c.addr, password) +} + +func (c *AuroraCluster) urlAt(addr, password string) string { u := url.URL{ Scheme: "postgres", User: url.UserPassword(fixtureUser, password), - Host: c.addr, + Host: addr, Path: "/" + fixtureDatabase, RawQuery: "sslmode=disable", } @@ -238,6 +251,9 @@ func ProvisionAuroraPostgres(t *testing.T) *AuroraCluster { }, WaitingFor: wait.ForHTTP("/_ministack/health"). WithPort(fmt.Sprintf("%d/tcp", ministackGatewayPort)), + Env: map[string]string{ + "MINISTACK_RDS_PG_CLUSTER_REPLICATION": "1", + }, }, }) require.NoError(t, err, "start Ministack container") @@ -309,6 +325,40 @@ func ProvisionAuroraPostgres(t *testing.T) *AuroraCluster { }, auroraProvisionDeadline, auroraProvisionPoll, "instance %s did not become %s within the provision deadline", instanceID, rdsStatusAvailable) + readerInstanceID := clusterID + "-reader" + _, err = clnt.CreateDBInstance(ctx, &rds.CreateDBInstanceInput{ + DBInstanceIdentifier: aws.String(readerInstanceID), + DBClusterIdentifier: aws.String(clusterID), + Engine: aws.String("aurora-postgresql"), + DBInstanceClass: aws.String("db.t3.medium"), + }) + require.NoError(t, err, "create aurora-postgresql reader instance") + t.Cleanup(func() { + cleanupCtx := context.WithoutCancel(t.Context()) + if _, err := clnt.DeleteDBInstance(cleanupCtx, &rds.DeleteDBInstanceInput{ + DBInstanceIdentifier: aws.String(readerInstanceID), + SkipFinalSnapshot: aws.Bool(true), + }); err != nil { + t.Logf("delete reader instance %s: %v", readerInstanceID, err) + } + }) + + var readerEndpoint string + var readerPort int32 + require.Eventuallyf(t, func() bool { + out, err := clnt.DescribeDBInstances(ctx, &rds.DescribeDBInstancesInput{ + DBInstanceIdentifier: aws.String(readerInstanceID), + }) + if err != nil || len(out.DBInstances) == 0 || + aws.ToString(out.DBInstances[0].DBInstanceStatus) != rdsStatusAvailable { + return false + } + readerEndpoint = aws.ToString(out.DBInstances[0].Endpoint.Address) + readerPort = aws.ToInt32(out.DBInstances[0].Endpoint.Port) + return readerEndpoint != "" && readerPort != 0 + }, auroraProvisionDeadline, auroraProvisionPoll, + "reader instance %s did not become %s within the provision deadline", readerInstanceID, rdsStatusAvailable) + // Read the endpoint back from the control plane rather than trusting the // request: the discovery flow is the behavior under test, and the // address it returns is the address the test connects to. The one @@ -329,16 +379,22 @@ func ProvisionAuroraPostgres(t *testing.T) *AuroraCluster { addr := net.JoinHostPort(endpoint, strconv.Itoa(int(port))) if !tcpReachable(t, addr) { - addr = siblingHostAddr(t, ctr, clusterID) + addr = memberHostAddr(t, ctr, memberKindCluster, clusterID) + } + readerAddr := net.JoinHostPort(readerEndpoint, strconv.Itoa(int(readerPort))) + if !tcpReachable(t, readerAddr) { + readerAddr = memberHostAddr(t, ctr, memberKindInstance, readerInstanceID) } return &AuroraCluster{ - Client: clnt, - ClusterID: clusterID, - InstanceID: instanceID, - addr: addr, - password: password, - secrets: secrets, + Client: clnt, + ClusterID: clusterID, + InstanceID: instanceID, + ReaderInstanceID: readerInstanceID, + addr: addr, + readerAddr: readerAddr, + password: password, + secrets: secrets, } } @@ -357,15 +413,24 @@ func tcpReachable(t *testing.T, addr string) bool { return true } -// siblingHostAddr resolves the host-published address of the sibling -// database container backing the cluster. Ministack publishes the -// sibling's PostgreSQL port on the Docker host, so a host that cannot -// route to container IPs connects through that mapping. The container -// name — "ministack-rds--cluster-" -// — is an emulator implementation detail this fallback accepts coupling -// to; it is exercised only on hosts where the discovered endpoint is -// unreachable. -func siblingHostAddr(t *testing.T, ctr testcontainers.Container, clusterID string) string { +// memberKind is the segment of a Ministack database-container name that +// says which identifier keys it: the cluster's shared writer container is +// named after the cluster, a replicating reader after its instance. +type memberKind string + +const ( + memberKindCluster memberKind = "cluster" + memberKindInstance memberKind = "instance" +) + +// memberHostAddr resolves the host-published address of the database +// container backing a cluster member. Ministack publishes each member's +// PostgreSQL port on the Docker host, so a host that cannot route to +// container IPs connects through that mapping. The container name — +// "ministack-rds---" — is an +// emulator implementation detail this fallback accepts coupling to; it is +// exercised only on hosts where the discovered endpoint is unreachable. +func memberHostAddr(t *testing.T, ctr testcontainers.Container, kind memberKind, identifier string) string { t.Helper() ctx := t.Context() docker, err := testcontainers.NewDockerClientWithOpts(ctx) @@ -377,14 +442,14 @@ func siblingHostAddr(t *testing.T, ctr testcontainers.Container, clusterID strin }() scope := sha1.Sum([]byte(awsAccountID + ":" + awsRegion)) - name := fmt.Sprintf("ministack-rds-%s-cluster-%s", hex.EncodeToString(scope[:])[:12], clusterID) + name := fmt.Sprintf("ministack-rds-%s-%s-%s", hex.EncodeToString(scope[:])[:12], kind, identifier) inspect, err := docker.ContainerInspect(ctx, name, client.ContainerInspectOptions{}) - require.NoErrorf(t, err, "inspect sibling database container %s", name) + require.NoErrorf(t, err, "inspect %s database container %s", kind, name) dbPort, err := network.ParsePort(siblingDBPort) - require.NoError(t, err, "parse sibling database port") + require.NoError(t, err, "parse member database port") bindings := inspect.Container.NetworkSettings.Ports[dbPort] - require.NotEmptyf(t, bindings, "sibling container %s must publish %s on the host", name, siblingDBPort) + require.NotEmptyf(t, bindings, "%s container %s must publish %s on the host", kind, name, siblingDBPort) host, err := ctr.Host(ctx) require.NoError(t, err, "resolve Docker host address") diff --git a/internal/testutil/ministack_integration_test.go b/internal/testutil/ministack_integration_test.go index aa2307e..cd0a3ee 100644 --- a/internal/testutil/ministack_integration_test.go +++ b/internal/testutil/ministack_integration_test.go @@ -3,6 +3,7 @@ package testutil_test import ( + "errors" "strconv" "testing" "time" @@ -11,26 +12,110 @@ import ( "github.com/aws/aws-sdk-go-v2/service/rds" "github.com/aws/aws-sdk-go-v2/service/rds/types" "github.com/jackc/pgx/v5/pgconn" + "github.com/jackc/pgx/v5/pgxpool" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" "github.com/block/pg-sprite/internal/testutil" "github.com/block/pg-sprite/pkg/dbconn" + "github.com/block/pg-sprite/pkg/executor" + "github.com/block/pg-sprite/pkg/preflight" +) + +const ( + controlPlaneDeadline = 2 * time.Minute + controlPlanePoll = 250 * time.Millisecond + // clusterStatementTimeout bounds every statement the subtests run, except + // the deliberately long-running payload of the connection-loss subtest. + clusterStatementTimeout = time.Minute + // interruptedChangeDuration is how long the connection-loss payload runs + // if nothing interrupts it. That subtest's session statement_timeout is + // set above it so a slow cluster stop can never race the payload into a + // server-side query_canceled that looks like an interruption. + interruptedChangeDuration = 5 * time.Minute +) + +// RDS API statuses the subtests observe. Ministack answers StopDBCluster +// with the terminal "stopped" where real RDS reports the transitional +// "stopping", so the stop assertion accepts both. +const ( + rdsStatusAvailable = "available" + rdsStatusStopping = "stopping" + rdsStatusStopped = "stopped" + rdsStatusStarting = "starting" + rdsStatusFailingOver = "failing-over" +) + +// SQLSTATEs a PostgreSQL backend reports to its client when the server is +// shut down underneath an in-flight statement. A fast shutdown lets each +// backend send admin_shutdown before it exits; a crash reports +// crash_shutdown; compute killed faster than either sends nothing at all. +const ( + codeAdminShutdown = "57P01" + codeCrashShutdown = "57P02" ) // TestAuroraControlPlane proves the AWS-boundary seams against one shared // provisioned cluster: provisioning a cluster costs minutes, so the -// subtests share it rather than provisioning three times. They run in -// order, and PasswordRotation runs last because it changes the cluster's -// master password. +// subtests share it rather than provisioning separately. They run in +// order, and two orderings are load-bearing: ConnectionLossDuringSchemaChange +// stops and restarts cluster compute and waits for every member to come +// back, so MetadataFailoverKeepsWriterSession — which targets the reader +// member — must run after it; PasswordRotation runs last because it +// changes the cluster's master password. func TestAuroraControlPlane(t *testing.T) { cluster := testutil.ProvisionAuroraPostgres(t) t.Run("ProvisionAndConnect", func(t *testing.T) { provisionAndConnect(t, cluster) }) t.Run("ErrorContract", func(t *testing.T) { errorContract(t, cluster) }) + t.Run("ReaderIsReadOnly", func(t *testing.T) { readerIsReadOnly(t, cluster) }) + t.Run("ConnectionLossDuringSchemaChange", func(t *testing.T) { connectionLossDuringSchemaChange(t, cluster) }) + t.Run("MetadataFailoverKeepsWriterSession", func(t *testing.T) { metadataFailoverKeepsWriterSession(t, cluster) }) t.Run("PasswordRotation", func(t *testing.T) { passwordRotation(t, cluster) }) } +func newClusterPool(t *testing.T, url string, statementTimeout time.Duration) *pgxpool.Pool { + t.Helper() + pool, err := dbconn.NewPool(t.Context(), dbconn.Config{ + URL: url, + LockTimeout: 300 * time.Millisecond, + StatementTimeout: statementTimeout, + ConnectTimeout: 2 * time.Second, + }) + require.NoError(t, err, "connect to provisioned cluster via dbconn") + t.Cleanup(pool.Close) + return pool +} + +// awaitInstanceStatus polls the control plane until the instance reports +// status, failing the test at the control-plane deadline. +func awaitInstanceStatus(t *testing.T, cluster *testutil.AuroraCluster, instanceID, status string) { + t.Helper() + require.Eventuallyf(t, func() bool { + out, err := cluster.Client.DescribeDBInstances(t.Context(), &rds.DescribeDBInstancesInput{ + DBInstanceIdentifier: aws.String(instanceID), + }) + return err == nil && len(out.DBInstances) == 1 && aws.ToString(out.DBInstances[0].DBInstanceStatus) == status + }, controlPlaneDeadline, controlPlanePoll, "instance %s did not report %s before the deadline", instanceID, status) +} + +// assertInterruptedByShutdown pins the cause of a failed statement to the +// server going away underneath it. A backend that gets to answer reports +// admin_shutdown or crash_shutdown; compute killed faster than that +// surfaces as a connection-level error carrying no server response. Any +// other SQLSTATE — query_canceled from the session statement_timeout above +// all — means the statement failed for a reason the stop did not cause. +func assertInterruptedByShutdown(t *testing.T, err error) { + t.Helper() + var pgErr *pgconn.PgError + if !errors.As(err, &pgErr) { + t.Logf("statement interrupted without a server response: %v", err) + return + } + assert.Contains(t, []string{codeAdminShutdown, codeCrashShutdown}, pgErr.Code, + "a statement interrupted by a cluster stop must fail as a server shutdown, got SQLSTATE %s: %s", pgErr.Code, pgErr.Message) +} + // provisionAndConnect proves the AWS-boundary flow end to end: an // aurora-postgresql cluster provisioned through the real RDS control-plane // API is discoverable, its endpoint accepts connections through pkg/dbconn @@ -102,6 +187,149 @@ func errorContract(t *testing.T, cluster *testutil.AuroraCluster) { "creating a duplicate instance must surface the typed already-exists fault") } +// readerIsReadOnly proves that Ministack's reader member is a real hot +// standby. The current preflight is catalog-only, so it succeeds there; +// PostgreSQL then refuses a write with read_only_sql_transaction, which the +// connection layer correctly treats as terminal rather than transient. +func readerIsReadOnly(t *testing.T, cluster *testutil.AuroraCluster) { + writer := newClusterPool(t, cluster.URL(), clusterStatementTimeout) + schema := testutil.NewSchema(t, writer) + _, err := writer.Exec(t.Context(), "CREATE TABLE "+schema+".reader_probe (id bigint PRIMARY KEY)") + require.NoError(t, err) + + reader := newClusterPool(t, cluster.ReaderURL(), clusterStatementTimeout) + var inRecovery bool + require.NoError(t, reader.QueryRow(t.Context(), "SELECT pg_is_in_recovery()").Scan(&inRecovery)) + assert.True(t, inRecovery, "reader instance must be a PostgreSQL hot standby") + + var proof preflight.PreflightedTable + require.EventuallyWithT(t, func(collect *assert.CollectT) { + var checkErr error + proof, checkErr = preflight.CheckTable(t.Context(), reader, schema, "reader_probe", preflight.NoSizeLimit) + assert.NoError(collect, checkErr) + }, controlPlaneDeadline, controlPlanePoll, "reader did not replay the table creation before the deadline") + assert.Equal(t, "reader_probe", proof.Table()) + + _, err = reader.Exec(t.Context(), "INSERT INTO "+schema+".reader_probe VALUES (1)") + var pgErr *pgconn.PgError + require.ErrorAs(t, err, &pgErr, "a reader write must return a PostgreSQL server error") + assert.Equal(t, "25006", pgErr.Code, "reader write must fail as read_only_sql_transaction") + assert.False(t, dbconn.Retryable(err), "read-only standby writes are terminal") +} + +// connectionLossDuringSchemaChange stops real cluster compute while DDL is +// executing. The optimistic executor has no connection-loss resume path: the +// connection layer does not retry an ambiguously interrupted write, and the +// executor's stable typed outcome is the fail-closed execution-failed fallback. +// The subtest pins the cause as well as the outcome — the statement must +// fail because the server went away, and the interrupted change must leave +// nothing behind — because the outcome alone is what any failed DDL produces. +func connectionLossDuringSchemaChange(t *testing.T, cluster *testutil.AuroraCluster) { + pool := newClusterPool(t, cluster.URL(), 2*interruptedChangeDuration) + schema := testutil.NewSchema(t, pool) + sleepSeconds := strconv.Itoa(int(interruptedChangeDuration.Seconds())) + ddl := "CREATE TABLE " + schema + ".interrupted AS SELECT 1 AS id FROM pg_sleep(" + sleepSeconds + ")" + + result := make(chan error, 1) + go func() { + _, err := pool.Exec(t.Context(), ddl) + result <- err + }() + require.Eventuallyf(t, func() bool { + var active bool + err := pool.QueryRow(t.Context(), `SELECT EXISTS ( + SELECT 1 FROM pg_stat_activity + WHERE query LIKE $1 AND state = 'active' AND pid <> pg_backend_pid())`, "%"+schema+".interrupted%").Scan(&active) + return err == nil && active + }, controlPlaneDeadline, controlPlanePoll, "schema change did not become active before the deadline") + + stopped, err := cluster.Client.StopDBCluster(t.Context(), &rds.StopDBClusterInput{ + DBClusterIdentifier: aws.String(cluster.ClusterID), + }) + require.NoError(t, err, "stop cluster during schema change") + assert.Contains(t, []string{rdsStatusStopping, rdsStatusStopped}, aws.ToString(stopped.DBCluster.Status), + "StopDBCluster must report the stop in progress or complete") + + var changeErr error + select { + case changeErr = <-result: + case <-time.After(controlPlaneDeadline): + require.FailNow(t, "schema change did not return after cluster stop") + } + require.Error(t, changeErr) + assertInterruptedByShutdown(t, changeErr) + assert.False(t, dbconn.Retryable(changeErr), + "an interrupted write with an ambiguous server outcome must not be retried") + assert.Equal(t, executor.CodeExecutionFailed, executor.OutcomeCode(changeErr), + "connection loss has no provable schema-change verdict and must fail closed") + + started, err := cluster.Client.StartDBCluster(t.Context(), &rds.StartDBClusterInput{ + DBClusterIdentifier: aws.String(cluster.ClusterID), + }) + require.NoError(t, err, "restart cluster after connection loss") + assert.Equal(t, rdsStatusStarting, aws.ToString(started.DBCluster.Status)) + require.Eventuallyf(t, func() bool { + out, err := cluster.Client.DescribeDBClusters(t.Context(), &rds.DescribeDBClustersInput{ + DBClusterIdentifier: aws.String(cluster.ClusterID), + }) + return err == nil && len(out.DBClusters) == 1 && aws.ToString(out.DBClusters[0].Status) == rdsStatusAvailable + }, controlPlaneDeadline, controlPlanePoll, "cluster did not become available after restart") + // The restart brings every member back, not just the writer: the + // following subtests target the reader member and must not inherit a + // half-restarted cluster. + awaitInstanceStatus(t, cluster, cluster.InstanceID, rdsStatusAvailable) + awaitInstanceStatus(t, cluster, cluster.ReaderInstanceID, rdsStatusAvailable) + + pool.Reset() + require.Eventuallyf(t, func() bool { + _, err := pool.Exec(t.Context(), "CREATE TABLE "+schema+".after_restart (id bigint)") + return err == nil + }, controlPlaneDeadline, controlPlanePoll, "engine could not execute DDL after cluster restart") + + // The interrupted change never committed: the shutdown aborted its + // transaction, so the relation it was creating does not exist. + var interruptedExists bool + require.NoError(t, pool.QueryRow(t.Context(), "SELECT to_regclass($1) IS NOT NULL", schema+".interrupted").Scan(&interruptedExists)) + assert.False(t, interruptedExists, "an interrupted schema change must leave no relation behind") +} + +// metadataFailoverKeepsWriterSession exercises only Ministack's current +// metadata failover: member writer flags flip and the response is +// transitional, but the standby is not promoted at the data plane yet. The +// established writer transaction therefore remains the engine's usable +// connection — which is exactly what the subtest proves, and all it proves. +func metadataFailoverKeepsWriterSession(t *testing.T, cluster *testutil.AuroraCluster) { + pool := newClusterPool(t, cluster.URL(), clusterStatementTimeout) + schema := testutil.NewSchema(t, pool) + tx, err := pool.Begin(t.Context()) + require.NoError(t, err) + defer func() { _ = tx.Rollback(t.Context()) }() + _, err = tx.Exec(t.Context(), "CREATE TABLE "+schema+".during_failover (id bigint PRIMARY KEY)") + require.NoError(t, err) + + failedOver, err := cluster.Client.FailoverDBCluster(t.Context(), &rds.FailoverDBClusterInput{ + DBClusterIdentifier: aws.String(cluster.ClusterID), + TargetDBInstanceIdentifier: aws.String(cluster.ReaderInstanceID), + }) + require.NoError(t, err, "fail over cluster metadata") + assert.Equal(t, rdsStatusFailingOver, aws.ToString(failedOver.DBCluster.Status)) + + described, err := cluster.Client.DescribeDBClusters(t.Context(), &rds.DescribeDBClustersInput{ + DBClusterIdentifier: aws.String(cluster.ClusterID), + }) + require.NoError(t, err) + require.Len(t, described.DBClusters, 1) + writers := 0 + for _, member := range described.DBClusters[0].DBClusterMembers { + if aws.ToBool(member.IsClusterWriter) { + writers++ + assert.Equal(t, cluster.ReaderInstanceID, aws.ToString(member.DBInstanceIdentifier)) + } + } + assert.Equal(t, 1, writers, "metadata must expose exactly one writer") + require.NoError(t, tx.Commit(t.Context()), "metadata-only failover must not interrupt the real writer") +} + // passwordRotation proves what a master-password rotation does to a // running schema change, and pins pg-sprite's contract for the failure. // PostgreSQL never re-authenticates an established session, so in-flight