Add support for max replication slots (#184)

Author: davissp14

internal/flypg/admin/admin.go

@@ -405,6 +405,25 @@ func ValidatePGSettings(ctx context.Context, conn *pgx.Conn, requested map[strin
 				}
 			}
 		}
+
+		if k == "max_replication_slots" {
+			maxReplicationSlotsStr := v.(string)
+
+			// Convert string to int
+			maxReplicationSlots, err := strconv.ParseInt(maxReplicationSlotsStr, 10, 64)
+			if err != nil {
+				return fmt.Errorf("failed to parse max_replication_slots: %s", err)
+			}
+
+			slots, err := ListReplicationSlots(ctx, conn)
+			if err != nil {
+				return fmt.Errorf("failed to verify replication slots: %s", err)
+			}
+
+			if len(slots) > int(maxReplicationSlots) {
+				return fmt.Errorf("max_replication_slots must be greater than or equal to the number of active replication slots (%d)", len(slots))
+			}
+		}
 	}
 
 	return nil

internal/flypg/pg.go

@@ -378,6 +378,25 @@ func (c *PGConfig) validateCompatibility(requested ConfigMap) (ConfigMap, error)
 
 	}
 
+	// Max-replication-slots
+	if v, ok := requested["max_replication_slots"]; ok {
+		{
+			val := v.(string)
+
+			// Convert string to int
+			maxReplicationSlots, err := strconv.ParseInt(val, 10, 64)
+			if err != nil {
+				return requested, fmt.Errorf("failed to parse max-replication-slots: %s", err)
+			}
+
+			walLevel := resolveConfigValue(requested, current, "wal_level", "replica")
+
+			if maxReplicationSlots > 0 && walLevel == "minimal" {
+				return requested, fmt.Errorf("wal_level must be set to replica or higher before replication slots can be used")
+			}
+		}
+	}
+
 	return requested, nil
 }
 

internal/flypg/pg_test.go

@@ -427,6 +427,32 @@ func TestValidateCompatibility(t *testing.T) {
 		}
 	})
 
+	t.Run("maxReplicationSlots", func(t *testing.T) {
+		valid := ConfigMap{
+			"wal_level":             "replica",
+			"max_replication_slots": "10",
+		}
+		if _, err := pgConf.validateCompatibility(valid); err != nil {
+			t.Fatal(err)
+		}
+
+		valid = ConfigMap{
+			"wal_level":             "logical",
+			"max_replication_slots": "12",
+		}
+		if _, err := pgConf.validateCompatibility(valid); err != nil {
+			t.Fatal(err)
+		}
+
+		invalid := ConfigMap{
+			"wal_level":             "minimal",
+			"max_replication_slots": "20",
+		}
+		if _, err := pgConf.validateCompatibility(invalid); err == nil {
+			t.Fatal(err)
+		}
+	})
+
 }
 
 func stubPGConfigFile() error {