From e24cfe4f09ba603e6c43638a8535f59c5aad51d6 Mon Sep 17 00:00:00 2001 From: Ethan Pailes Date: Mon, 9 Nov 2020 08:48:57 -0500 Subject: [PATCH] fix stmtcache invalidation This patch fixes jackc/pgx#841. The meat of the fix lives in [a PR to the pgconn repo][1]. This change just checks for errors after executing a prepared statement and informs the underlying stmtcache about them so that it can properly clean up. We don't try to get fancy with retries or anything like that, just return the error and allow the application to handle it. I had to make [some][1] [changes][2] to to the jackc/pgconn package as well as this package, so I've retargeted this this repo on ethanpailes/pgconn for the time being. This PR is not really in a mergable state until those PRs merge and I rip out the retargeting commit, but I would say it is ready for review (just review the most recent commit). Fixes #841 [1]: https://github.com/jackc/pgconn/pull/56 [2]: https://github.com/jackc/pgconn/pull/55 --- conn.go | 7 +++ conn_test.go | 128 +++++++++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 135 insertions(+) diff --git a/conn.go b/conn.go index 4362eef6b..5eb76ad03 100644 --- a/conn.go +++ b/conn.go @@ -652,6 +652,13 @@ optionLoop: rows.resultReader = c.pgConn.ExecParams(ctx, sql, c.eqb.paramValues, sd.ParamOIDs, c.eqb.paramFormats, resultFormats) } else { rows.resultReader = c.pgConn.ExecPrepared(ctx, sd.Name, c.eqb.paramValues, c.eqb.paramFormats, resultFormats) + readerErr := rows.resultReader.Err() + if readerErr != nil { + err = c.stmtcache.StatementErrored(ctx, sql, readerErr) + if err != nil { + rows.err = errors.Errorf("cleaning up from '%s': %s", readerErr.Error(), err.Error()) + } + } } return rows, rows.err diff --git a/conn_test.go b/conn_test.go index 8a604acd9..d0c33a385 100644 --- a/conn_test.go +++ b/conn_test.go @@ -879,3 +879,131 @@ func TestDomainType(t *testing.T) { } }) } + +func TestStmtCacheInvalidationConn(t *testing.T) { + ctx := context.Background() + + conn := mustConnectString(t, os.Getenv("PGX_TEST_DATABASE")) + defer closeConn(t, conn) + + // create a table and fill it with some data + _, err := conn.Exec(ctx, ` + DROP TABLE IF EXISTS drop_cols; + CREATE TABLE drop_cols ( + id SERIAL PRIMARY KEY NOT NULL, + f1 int NOT NULL, + f2 int NOT NULL + ); + `) + require.NoError(t, err) + _, err = conn.Exec(ctx, "INSERT INTO drop_cols (f1, f2) VALUES (1, 2)") + require.NoError(t, err) + + getSQL := "SELECT * FROM drop_cols WHERE id = $1" + + // This query will populate the statement cache. We don't care about the result. + rows, err := conn.Query(ctx, getSQL, 1) + require.NoError(t, err) + rows.Close() + + // Now, change the schema of the table out from under the statement, making it invalid. + _, err = conn.Exec(ctx, "ALTER TABLE drop_cols DROP COLUMN f1") + require.NoError(t, err) + + // We must get an error the first time we try to re-execute a bad statement. + // It is up to the application to determine if it wants to try again. We punt to + // the application because there is no clear recovery path in the case of failed transactions + // or batch operations and because automatic retry is tricky and we don't want to get + // it wrong at such an importaint layer of the stack. + rows, err = conn.Query(ctx, getSQL, 1) + if err == nil { + t.Fatal("expected InvalidCachedStatementPlanError: no error") + } + if !strings.Contains(err.Error(), "cached plan must not change result type") { + t.Fatalf("expected InvalidCachedStatementPlanError, got: %s", err.Error()) + } + rows.Close() + + // On retry, the statement should have been flushed from the cache. + rows, err = conn.Query(ctx, getSQL, 1) + require.NoError(t, err) + rows.Next() + err = rows.Err() + require.NoError(t, err) + rows.Close() + + ensureConnValid(t, conn) +} + +func TestStmtCacheInvalidationTx(t *testing.T) { + ctx := context.Background() + + conn := mustConnectString(t, os.Getenv("PGX_TEST_DATABASE")) + defer closeConn(t, conn) + + // create a table and fill it with some data + _, err := conn.Exec(ctx, ` + DROP TABLE IF EXISTS drop_cols; + CREATE TABLE drop_cols ( + id SERIAL PRIMARY KEY NOT NULL, + f1 int NOT NULL, + f2 int NOT NULL + ); + `) + require.NoError(t, err) + _, err = conn.Exec(ctx, "INSERT INTO drop_cols (f1, f2) VALUES (1, 2)") + require.NoError(t, err) + + tx, err := conn.Begin(ctx) + require.NoError(t, err) + + getSQL := "SELECT * FROM drop_cols WHERE id = $1" + + // This query will populate the statement cache. We don't care about the result. + rows, err := tx.Query(ctx, getSQL, 1) + require.NoError(t, err) + rows.Close() + + // Now, change the schema of the table out from under the statement, making it invalid. + _, err = tx.Exec(ctx, "ALTER TABLE drop_cols DROP COLUMN f1") + require.NoError(t, err) + + // We must get an error the first time we try to re-execute a bad statement. + // It is up to the application to determine if it wants to try again. We punt to + // the application because there is no clear recovery path in the case of failed transactions + // or batch operations and because automatic retry is tricky and we don't want to get + // it wrong at such an importaint layer of the stack. + rows, err = tx.Query(ctx, getSQL, 1) + require.NoError(t, err) + rows.Next() + err = rows.Err() + if err == nil { + t.Fatal("expected InvalidCachedStatementPlanError: no error") + } + if !strings.Contains(err.Error(), "cached plan must not change result type") { + t.Fatalf("expected InvalidCachedStatementPlanError, got: %s", err.Error()) + } + rows.Close() + + rows, err = tx.Query(ctx, getSQL, 1) + require.NoError(t, err) // error does not pop up immediately + rows.Next() + err = rows.Err() + // Retries within the same transaction are errors (really anything except a rollbakc + // will be an error in this transaction). + require.Error(t, err) + rows.Close() + + err = tx.Rollback(ctx) + require.NoError(t, err) + + // once we've rolled back, retries will work + rows, err = conn.Query(ctx, getSQL, 1) + require.NoError(t, err) + rows.Next() + err = rows.Err() + require.NoError(t, err) + rows.Close() + + ensureConnValid(t, conn) +}