Wait only for local write of controller SQL

When Postgres replication is broken and synchronous commit is enabled,
the controller blocks waiting for a remote write that will never happen.
This change allows the controller to return from SQL writes and repair
replication.

Issue: PGO-1592

Author: cbandy

internal/controller/postgrescluster/pgmonitor_test.go

@@ -602,7 +602,7 @@ func TestReconcilePGMonitorExporterStatus(t *testing.T) {
 		podExecCalled:   false,
 		// Status was generated manually for this test case
 		// TODO (jmckulk): add code to generate status
-		status:                      v1beta1.MonitoringStatus{ExporterConfiguration: "6d874c58df"},
+		status:                      v1beta1.MonitoringStatus{ExporterConfiguration: "5c5f955485"},
 		statusChangedAfterReconcile: false,
 	}} {
 		t.Run(test.name, func(t *testing.T) {

internal/pgaudit/postgres.go

@@ -35,7 +35,9 @@ func EnableInPostgreSQL(ctx context.Context, exec postgres.Executor) error {
 
 	stdout, stderr, err := exec.ExecInAllDatabases(ctx,
 		// Quiet the NOTICE from IF EXISTS, and install the pgAudit event triggers.
+		// Use the default setting for "synchronous_commit".
 		// - https://www.postgresql.org/docs/current/runtime-config-client.html
+		// - https://www.postgresql.org/docs/current/runtime-config-wal.html
 		// - https://github.com/pgaudit/pgaudit#settings
 		`SET client_min_messages = WARNING; CREATE EXTENSION IF NOT EXISTS pgaudit;`,
 		map[string]string{

internal/pgbouncer/postgres.go

@@ -68,6 +68,10 @@ func DisableInPostgreSQL(ctx context.Context, exec postgres.Executor) error {
 			// - https://www.postgresql.org/docs/current/runtime-config-client.html
 			`SET client_min_messages = WARNING;`,
 
+			// Do not wait for changes to be replicated. [Since PostgreSQL v9.1]
+			// - https://www.postgresql.org/docs/current/runtime-config-wal.html
+			`SET synchronous_commit = LOCAL;`,
+
 			// Drop the following objects in a transaction.
 			`BEGIN;`,
 
@@ -102,7 +106,7 @@ SELECT pg_catalog.format('DROP OWNED BY %I CASCADE', :'username')
 		// Remove the PgBouncer user now that the objects and other privileges are gone.
 		stdout, stderr, err = exec.ExecInDatabasesFromQuery(ctx,
 			`SELECT pg_catalog.current_database()`,
-			`SET client_min_messages = WARNING; DROP ROLE IF EXISTS :"username";`,
+			`SET client_min_messages = WARNING; SET synchronous_commit = LOCAL; DROP ROLE IF EXISTS :"username";`,
 			map[string]string{
 				"username": postgresqlUser,
 
@@ -130,6 +134,10 @@ func EnableInPostgreSQL(
 			// - https://www.postgresql.org/docs/current/runtime-config-client.html
 			`SET client_min_messages = WARNING;`,
 
+			// Do not wait for changes to be replicated. [Since PostgreSQL v9.1]
+			// - https://www.postgresql.org/docs/current/runtime-config-wal.html
+			`SET synchronous_commit = LOCAL;`,
+
 			// Create the following objects in a transaction so that permissions
 			// are correct before any other session sees them.
 			// - https://www.postgresql.org/docs/current/ddl-priv.html

internal/pgbouncer/postgres_test.go

@@ -49,6 +49,7 @@ func TestDisableInPostgreSQL(t *testing.T) {
 			assert.NilError(t, err)
 			assert.Equal(t, string(b), strings.TrimSpace(`
 SET client_min_messages = WARNING;
+SET synchronous_commit = LOCAL;
 BEGIN;
 DROP FUNCTION IF EXISTS :"namespace".get_auth(username TEXT);
 DROP SCHEMA IF EXISTS :"namespace" CASCADE;
@@ -90,7 +91,7 @@ COMMIT;`))
 
 			b, err := io.ReadAll(stdin)
 			assert.NilError(t, err)
-			assert.Equal(t, string(b), `SET client_min_messages = WARNING; DROP ROLE IF EXISTS :"username";`)
+			assert.Equal(t, string(b), `SET client_min_messages = WARNING; SET synchronous_commit = LOCAL; DROP ROLE IF EXISTS :"username";`)
 			gomega.NewWithT(t).Expect(command).To(gomega.ContainElements(
 				`--set=username=_crunchypgbouncer`,
 			), "expected query parameters")
@@ -135,6 +136,7 @@ func TestEnableInPostgreSQL(t *testing.T) {
 		assert.NilError(t, err)
 		assert.Equal(t, string(b), strings.TrimSpace(`
 SET client_min_messages = WARNING;
+SET synchronous_commit = LOCAL;
 BEGIN;
 SELECT pg_catalog.format('CREATE ROLE %I NOLOGIN', :'username')
  WHERE NOT EXISTS (SELECT 1 FROM pg_catalog.pg_roles WHERE rolname = :'username')

internal/pgmonitor/postgres.go

@@ -79,6 +79,10 @@ func EnableExporterInPostgreSQL(ctx context.Context, exec postgres.Executor,
 			// - https://www.postgresql.org/docs/current/runtime-config-client.html
 			`SET client_min_messages = WARNING;`,
 
+			// Do not wait for changes to be replicated. [Since PostgreSQL v9.1]
+			// - https://www.postgresql.org/docs/current/runtime-config-wal.html
+			`SET synchronous_commit = LOCAL;`,
+
 			// Exporter expects that extension(s) to be installed in all databases
 			// pg_stat_statements: https://access.crunchydata.com/documentation/pgmonitor/latest/exporter/
 			"CREATE EXTENSION IF NOT EXISTS pg_stat_statements;",
@@ -103,6 +107,10 @@ func EnableExporterInPostgreSQL(ctx context.Context, exec postgres.Executor,
 				// - https://www.postgresql.org/docs/current/runtime-config-client.html
 				`SET client_min_messages = WARNING;`,
 
+				// Do not wait for changes to be replicated. [Since PostgreSQL v9.1]
+				// - https://www.postgresql.org/docs/current/runtime-config-wal.html
+				`SET synchronous_commit = LOCAL;`,
+
 				// Setup.sql file from the exporter image. sql is specific
 				// to the PostgreSQL version
 				setup,

internal/postgis/postgis.go

@@ -26,6 +26,10 @@ func EnableInPostgreSQL(ctx context.Context, exec postgres.Executor) error {
 			// - https://www.postgresql.org/docs/current/runtime-config-client.html
 			`SET client_min_messages = WARNING;`,
 
+			// Do not wait for changes to be replicated. [Since PostgreSQL v9.1]
+			// - https://www.postgresql.org/docs/current/runtime-config-wal.html
+			`SET synchronous_commit = LOCAL;`,
+
 			`CREATE EXTENSION IF NOT EXISTS postgis;`,
 			`CREATE EXTENSION IF NOT EXISTS postgis_topology;`,
 			`CREATE EXTENSION IF NOT EXISTS fuzzystrmatch;`,

internal/postgis/postgis_test.go

@@ -29,6 +29,7 @@ func TestEnableInPostgreSQL(t *testing.T) {
 		b, err := io.ReadAll(stdin)
 		assert.NilError(t, err)
 		assert.Equal(t, string(b), `SET client_min_messages = WARNING;
+SET synchronous_commit = LOCAL;
 CREATE EXTENSION IF NOT EXISTS postgis;
 CREATE EXTENSION IF NOT EXISTS postgis_topology;
 CREATE EXTENSION IF NOT EXISTS fuzzystrmatch;

internal/postgres/users.go

@@ -69,6 +69,10 @@ func WriteUsersInPostgreSQL(
 	var err error
 	var sql bytes.Buffer
 
+	// Do not wait for changes to be replicated. [Since PostgreSQL v9.1]
+	// - https://www.postgresql.org/docs/current/runtime-config-wal.html
+	_, _ = sql.WriteString(`SET synchronous_commit = LOCAL;`)
+
 	// Prevent unexpected dereferences by emptying "search_path". The "pg_catalog"
 	// schema is still searched, and only temporary objects can be created.
 	// - https://www.postgresql.org/docs/current/runtime-config-client.html#GUC-SEARCH-PATH
@@ -219,6 +223,10 @@ func WriteUsersSchemasInPostgreSQL(ctx context.Context, exec Executor,
 				// - https://www.postgresql.org/docs/current/runtime-config-client.html
 				`SET client_min_messages = WARNING;`,
 
+				// Do not wait for changes to be replicated. [Since PostgreSQL v9.1]
+				// - https://www.postgresql.org/docs/current/runtime-config-wal.html
+				`SET synchronous_commit = LOCAL;`,
+
 				// Creates a schema named after and owned by the user
 				// - https://www.postgresql.org/docs/current/ddl-schemas.html
 				// - https://www.postgresql.org/docs/current/sql-createschema.html

internal/postgres/users_test.go

@@ -63,7 +63,7 @@ func TestWriteUsersInPostgreSQL(t *testing.T) {
 			b, err := io.ReadAll(stdin)
 			assert.NilError(t, err)
 			assert.Equal(t, string(b), strings.TrimSpace(`
-SET search_path TO '';
+SET synchronous_commit = LOCAL;SET search_path TO '';
 CREATE TEMPORARY TABLE input (id serial, data json);
 \copy input (data) from stdin with (format text)
 \.