Fix wal-level config validation (#161)

* Fix wal-level config validation

fix logic

Max wal senders must be 0 for wal-level minimal

Fix error message

fix

* Addressing edge cases

Author: davissp14

internal/flypg/pg.go

@@ -318,29 +318,82 @@ func (c *PGConfig) validateCompatibility(requested ConfigMap) (ConfigMap, error)
 	// Wal-level
 	if v, ok := requested["wal_level"]; ok {
 		value := v.(string)
+		switch value {
+		case "minimal":
+			var maxWalSenders int64
 
-		// Postgres will not boot properly if minimal is set with archive_mode enabled.
-		if value == "minimal" {
-			valid := false
-			if val, ok := requested["archive_mode"]; ok {
-				if val == "off" {
-					valid = true
-				}
+			// flyctl passes in `max_wal_senders` in as a string.
+			maxWalSendersInterface := resolveConfigValue(requested, current, "max_wal_senders", "10")
+
+			// Convert string to int
+			maxWalSenders, err = strconv.ParseInt(maxWalSendersInterface.(string), 10, 64)
+			if err != nil {
+				return requested, fmt.Errorf("failed to parse max-wal-senders: %s", err)
 			}
 
-			if !valid && current["archive_mode"] == "off" {
-				valid = true
+			if maxWalSenders > 0 {
+				return requested, fmt.Errorf("max_wal_senders must be set to `0` before wal-level can be set to `minimal`")
 			}
 
-			if !valid {
+			archiveMode := resolveConfigValue(requested, current, "archive_mode", "off")
+			if archiveMode.(string) != "off" {
 				return requested, errors.New("archive_mode must be set to `off` before wal_level can be set to `minimal`")
 			}
+
+		case "replica", "logical":
+			var maxWalSenders int64
+			maxWalSendersInterface := resolveConfigValue(requested, current, "max_wal_senders", "10")
+
+			// Convert string to int
+			maxWalSenders, err = strconv.ParseInt(maxWalSendersInterface.(string), 10, 64)
+			if err != nil {
+				return requested, fmt.Errorf("failed to parse max-wal-senders: %s", err)
+			}
+
+			if maxWalSenders == 0 {
+				return requested, fmt.Errorf("max_wal_senders must be greater than `0`")
+			}
+		}
+	}
+
+	// Max-wal-senders
+	if v, ok := requested["max_wal_senders"]; ok {
+		val := v.(string)
+
+		// Convert string to int
+		maxWalSenders, err := strconv.ParseInt(val, 10, 64)
+		if err != nil {
+			return requested, fmt.Errorf("failed to parse max-wal-senders: %s", err)
 		}
+
+		walLevel := resolveConfigValue(requested, current, "wal_level", "replica")
+
+		if maxWalSenders > 0 && walLevel == "minimal" {
+			return requested, fmt.Errorf("max_wal_senders must be set to `0` when wal_level is `minimal`")
+		}
+
+		if maxWalSenders == 0 && walLevel != "minimal" {
+			return requested, fmt.Errorf("max_wal_senders must be greater than `0` when wal_level is set to `%s`", walLevel.(string))
+		}
+
 	}
 
 	return requested, nil
 }
 
+func resolveConfigValue(requested ConfigMap, current ConfigMap, key string, defaultVal interface{}) interface{} {
+	val := requested[key]
+	if val == nil {
+		val = current[key]
+	}
+
+	if val == nil {
+		val = defaultVal
+	}
+
+	return val
+}
+
 type HBAEntry struct {
 	Type     string
 	Database string

internal/flypg/pg_test.go

@@ -346,29 +346,87 @@ func TestValidateCompatibility(t *testing.T) {
 			t.Fatal(err)
 		}
 
-		valid = ConfigMap{
-			"wal_level":    "minimal",
-			"archive_mode": "off",
+		invalid := ConfigMap{
+			"wal_level":       "logical",
+			"max_wal_senders": "0",
+		}
+		if _, err := pgConf.validateCompatibility(invalid); err == nil {
+			t.Fatal(err)
+		}
+
+		invalid = ConfigMap{
+			"wal_level":       "replica",
+			"max_wal_senders": "0",
+		}
+		if _, err := pgConf.validateCompatibility(invalid); err == nil {
+			t.Fatal(err)
+		}
+
+	})
+
+	t.Run("WalLevelMinimal", func(t *testing.T) {
+		valid := ConfigMap{
+			"wal_level":       "minimal",
+			"archive_mode":    "off",
+			"max_wal_senders": "0",
 		}
 		if _, err := pgConf.validateCompatibility(valid); err != nil {
 			t.Fatal(err)
 		}
 
 		invalid := ConfigMap{
+			"wal_level":       "minimal",
+			"archive_mode":    "on",
+			"max_wal_senders": "0",
+		}
+		if _, err := pgConf.validateCompatibility(invalid); err == nil {
+			t.Fatal(err)
+		}
+
+		invalid = ConfigMap{
+			"wal_level":       "minimal",
+			"archive_mode":    "off",
+			"max_wal_senders": "10",
+		}
+		if _, err := pgConf.validateCompatibility(invalid); err == nil {
+			t.Fatal(err)
+		}
+
+		invalid = ConfigMap{
 			"wal_level": "minimal",
 		}
+		if _, err := pgConf.validateCompatibility(invalid); err == nil {
+			t.Fatal(err)
+		}
+	})
+
+	t.Run("maxWalSenders", func(t *testing.T) {
+		valid := ConfigMap{
+			"wal_level":       "minimal",
+			"archive_mode":    "off",
+			"max_wal_senders": "0",
+		}
+		if _, err := pgConf.validateCompatibility(valid); err != nil {
+			t.Fatal(err)
+		}
 
+		invalid := ConfigMap{
+			"wal_level":       "replica",
+			"max_wal_senders": "0",
+		}
 		if _, err := pgConf.validateCompatibility(invalid); err == nil {
-			t.Fatal("expected wal_level minimal to fail with archiving enabled")
+			t.Fatal(err)
 		}
 
 		invalid = ConfigMap{
-			"wal_level": "logical",
+			"wal_level":       "logical",
+			"max_wal_senders": "0",
 		}
-		if _, err := pgConf.validateCompatibility(invalid); err != nil {
+		if _, err := pgConf.validateCompatibility(invalid); err == nil {
 			t.Fatal(err)
 		}
 	})
+
 }
 
 func stubPGConfigFile() error {