From 530cde2ce9d1e8d7dfc81e9f43abc43776891421 Mon Sep 17 00:00:00 2001 From: Kiran Muddukrishna Date: Thu, 3 Sep 2026 17:00:07 +1000 Subject: [PATCH 1/2] test: add reader, connection-loss and failover checks to the AWS-boundary tier Bumps the pinned Ministack image so the tier runs against real hot-standby readers and the FailoverDBCluster API. Failover is asserted at the metadata level only; data-plane promotion is not exercised yet. --- docs/testing.md | 16 +- internal/testutil/ministack.go | 98 ++++++++++-- .../testutil/ministack_integration_test.go | 151 +++++++++++++++++- 3 files changed, 252 insertions(+), 13 deletions(-) diff --git a/docs/testing.md b/docs/testing.md index 6babf9e..7da2c83 100644 --- a/docs/testing.md +++ b/docs/testing.md @@ -204,7 +204,7 @@ 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: @@ -221,7 +221,17 @@ 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 with the terminal, fail-closed `execution-failed` outcome, then starting + it restores the writer and allows a fresh schema change; +- **metadata failover** — `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 +301,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..903b3b5 100644 --- a/internal/testutil/ministack.go +++ b/internal/testutil/ministack.go @@ -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 @@ -331,14 +381,20 @@ func ProvisionAuroraPostgres(t *testing.T) *AuroraCluster { if !tcpReachable(t, addr) { addr = siblingHostAddr(t, ctr, clusterID) } + readerAddr := net.JoinHostPort(readerEndpoint, strconv.Itoa(int(readerPort))) + if !tcpReachable(t, readerAddr) { + readerAddr = instanceHostAddr(t, ctr, 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, } } @@ -391,6 +447,30 @@ func siblingHostAddr(t *testing.T, ctr testcontainers.Container, clusterID strin return net.JoinHostPort(host, bindings[0].HostPort) } +func instanceHostAddr(t *testing.T, ctr testcontainers.Container, instanceID string) string { + t.Helper() + ctx := t.Context() + docker, err := testcontainers.NewDockerClientWithOpts(ctx) + require.NoError(t, err, "create Docker client") + defer func() { + if err := docker.Close(); err != nil { + t.Logf("close Docker client: %v", err) + } + }() + + scope := sha1.Sum([]byte(awsAccountID + ":" + awsRegion)) + name := fmt.Sprintf("ministack-rds-%s-instance-%s", hex.EncodeToString(scope[:])[:12], instanceID) + inspect, err := docker.ContainerInspect(ctx, name, client.ContainerInspectOptions{}) + require.NoErrorf(t, err, "inspect reader database container %s", name) + dbPort, err := network.ParsePort(siblingDBPort) + require.NoError(t, err, "parse reader database port") + bindings := inspect.Container.NetworkSettings.Ports[dbPort] + require.NotEmptyf(t, bindings, "reader container %s must publish %s on the host", name, siblingDBPort) + host, err := ctr.Host(ctx) + require.NoError(t, err, "resolve Docker host address") + return net.JoinHostPort(host, bindings[0].HostPort) +} + // awsClients returns RDS and Secrets Manager clients pointed at the // container's gateway with the emulator's conventional static credentials. func awsClients(t *testing.T, ctr testcontainers.Container) (*rds.Client, *secretsmanager.Client) { diff --git a/internal/testutil/ministack_integration_test.go b/internal/testutil/ministack_integration_test.go index aa2307e..9d8d7f0 100644 --- a/internal/testutil/ministack_integration_test.go +++ b/internal/testutil/ministack_integration_test.go @@ -11,16 +11,24 @@ 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 ) // 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 +// subtests share it rather than provisioning separately. They run in // order, and PasswordRotation runs last because it changes the cluster's // master password. func TestAuroraControlPlane(t *testing.T) { @@ -28,9 +36,25 @@ func TestAuroraControlPlane(t *testing.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("FailoverDuringSchemaChange", func(t *testing.T) { failoverDuringSchemaChange(t, cluster) }) t.Run("PasswordRotation", func(t *testing.T) { passwordRotation(t, cluster) }) } +func newClusterPool(t *testing.T, url string) *pgxpool.Pool { + t.Helper() + pool, err := dbconn.NewPool(t.Context(), dbconn.Config{ + URL: url, + LockTimeout: 300 * time.Millisecond, + StatementTimeout: time.Minute, + ConnectTimeout: 2 * time.Second, + }) + require.NoError(t, err, "connect to provisioned cluster via dbconn") + t.Cleanup(pool.Close) + return pool +} + // 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 +126,131 @@ 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()) + 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()) + 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. +func connectionLossDuringSchemaChange(t *testing.T, cluster *testutil.AuroraCluster) { + pool := newClusterPool(t, cluster.URL()) + schema := testutil.NewSchema(t, pool) + ddl := "CREATE TABLE " + schema + ".interrupted AS SELECT 1 AS id FROM pg_sleep(300)" + + 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.Equal(t, "stopped", aws.ToString(stopped.DBCluster.Status)) + + 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) + 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, "starting", 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) == "available" + }, controlPlaneDeadline, controlPlanePoll, "cluster did not become available after restart") + + 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") +} + +// failoverDuringSchemaChange 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. +func failoverDuringSchemaChange(t *testing.T, cluster *testutil.AuroraCluster) { + pool := newClusterPool(t, cluster.URL()) + 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, "failing-over", 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 From 5c971c63abf077192d8cae2e50afe3de3eb21e7d Mon Sep 17 00:00:00 2001 From: Kiran Muddukrishna Date: Fri, 4 Sep 2026 10:17:01 +1000 Subject: [PATCH 2/2] test: pin the cause of the connection-loss interruption The connection-loss subtest accepted any DDL failure, including a query_canceled from its own one-minute statement_timeout racing the five-minute payload; it now requires the backend's shutdown SQLSTATE or a connection-level error, sets the session timeout above the payload so that race cannot exist, and proves the interrupted relation left no trace. The restart waits for both members so the failover subtest never inherits a half-restarted cluster. --- docs/testing.md | 30 +++-- internal/testutil/ministack.go | 67 ++++------ .../testutil/ministack_integration_test.go | 117 +++++++++++++++--- 3 files changed, 146 insertions(+), 68 deletions(-) diff --git a/docs/testing.md b/docs/testing.md index 7da2c83..20178da 100644 --- a/docs/testing.md +++ b/docs/testing.md @@ -206,8 +206,16 @@ mirrors. 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; @@ -226,12 +234,18 @@ password rotation last. Each subtest pins one AWS seam: 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 with the terminal, fail-closed `execution-failed` outcome, then starting - it restores the writer and allows a fresh schema change; -- **metadata failover** — `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. + 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 diff --git a/internal/testutil/ministack.go b/internal/testutil/ministack.go index 903b3b5..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" ) @@ -379,11 +379,11 @@ 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 = instanceHostAddr(t, ctr, readerInstanceID) + readerAddr = memberHostAddr(t, ctr, memberKindInstance, readerInstanceID) } return &AuroraCluster{ @@ -413,41 +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 { - t.Helper() - ctx := t.Context() - docker, err := testcontainers.NewDockerClientWithOpts(ctx) - require.NoError(t, err, "create Docker client") - defer func() { - if err := docker.Close(); err != nil { - t.Logf("close Docker client: %v", err) - } - }() - - scope := sha1.Sum([]byte(awsAccountID + ":" + awsRegion)) - name := fmt.Sprintf("ministack-rds-%s-cluster-%s", hex.EncodeToString(scope[:])[:12], clusterID) - inspect, err := docker.ContainerInspect(ctx, name, client.ContainerInspectOptions{}) - require.NoErrorf(t, err, "inspect sibling database container %s", name) +// 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 - dbPort, err := network.ParsePort(siblingDBPort) - require.NoError(t, err, "parse sibling database port") - bindings := inspect.Container.NetworkSettings.Ports[dbPort] - require.NotEmptyf(t, bindings, "sibling container %s must publish %s on the host", name, siblingDBPort) - - host, err := ctr.Host(ctx) - require.NoError(t, err, "resolve Docker host address") - return net.JoinHostPort(host, bindings[0].HostPort) -} +const ( + memberKindCluster memberKind = "cluster" + memberKindInstance memberKind = "instance" +) -func instanceHostAddr(t *testing.T, ctr testcontainers.Container, instanceID string) string { +// 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) @@ -459,13 +442,15 @@ func instanceHostAddr(t *testing.T, ctr testcontainers.Container, instanceID str }() scope := sha1.Sum([]byte(awsAccountID + ":" + awsRegion)) - name := fmt.Sprintf("ministack-rds-%s-instance-%s", hex.EncodeToString(scope[:])[:12], instanceID) + 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 reader 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 reader database port") + require.NoError(t, err, "parse member database port") bindings := inspect.Container.NetworkSettings.Ports[dbPort] - require.NotEmptyf(t, bindings, "reader 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") return net.JoinHostPort(host, bindings[0].HostPort) diff --git a/internal/testutil/ministack_integration_test.go b/internal/testutil/ministack_integration_test.go index 9d8d7f0..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" @@ -24,13 +25,44 @@ import ( 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 separately. They run in -// order, and PasswordRotation runs last because it changes the cluster's -// master password. +// 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) @@ -38,16 +70,16 @@ func TestAuroraControlPlane(t *testing.T) { 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("FailoverDuringSchemaChange", func(t *testing.T) { failoverDuringSchemaChange(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) *pgxpool.Pool { +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: time.Minute, + StatementTimeout: statementTimeout, ConnectTimeout: 2 * time.Second, }) require.NoError(t, err, "connect to provisioned cluster via dbconn") @@ -55,6 +87,35 @@ func newClusterPool(t *testing.T, url string) *pgxpool.Pool { 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 @@ -131,12 +192,12 @@ func errorContract(t *testing.T, cluster *testutil.AuroraCluster) { // 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()) + 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()) + 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") @@ -160,10 +221,14 @@ func readerIsReadOnly(t *testing.T, cluster *testutil.AuroraCluster) { // 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()) + pool := newClusterPool(t, cluster.URL(), 2*interruptedChangeDuration) schema := testutil.NewSchema(t, pool) - ddl := "CREATE TABLE " + schema + ".interrupted AS SELECT 1 AS id FROM pg_sleep(300)" + 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() { @@ -182,7 +247,8 @@ func connectionLossDuringSchemaChange(t *testing.T, cluster *testutil.AuroraClus DBClusterIdentifier: aws.String(cluster.ClusterID), }) require.NoError(t, err, "stop cluster during schema change") - assert.Equal(t, "stopped", aws.ToString(stopped.DBCluster.Status)) + assert.Contains(t, []string{rdsStatusStopping, rdsStatusStopped}, aws.ToString(stopped.DBCluster.Status), + "StopDBCluster must report the stop in progress or complete") var changeErr error select { @@ -191,6 +257,7 @@ func connectionLossDuringSchemaChange(t *testing.T, cluster *testutil.AuroraClus 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), @@ -200,27 +267,39 @@ func connectionLossDuringSchemaChange(t *testing.T, cluster *testutil.AuroraClus DBClusterIdentifier: aws.String(cluster.ClusterID), }) require.NoError(t, err, "restart cluster after connection loss") - assert.Equal(t, "starting", aws.ToString(started.DBCluster.Status)) + 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) == "available" + 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") } -// failoverDuringSchemaChange 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. -func failoverDuringSchemaChange(t *testing.T, cluster *testutil.AuroraCluster) { - pool := newClusterPool(t, cluster.URL()) +// 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) @@ -233,7 +312,7 @@ func failoverDuringSchemaChange(t *testing.T, cluster *testutil.AuroraCluster) { TargetDBInstanceIdentifier: aws.String(cluster.ReaderInstanceID), }) require.NoError(t, err, "fail over cluster metadata") - assert.Equal(t, "failing-over", aws.ToString(failedOver.DBCluster.Status)) + assert.Equal(t, rdsStatusFailingOver, aws.ToString(failedOver.DBCluster.Status)) described, err := cluster.Client.DescribeDBClusters(t.Context(), &rds.DescribeDBClustersInput{ DBClusterIdentifier: aws.String(cluster.ClusterID),