Connection watching fixes.

Fix ConnectionConfigMap.Diff  inconsistent usage of Plugin and PluginInstance. Closes #3895
Fix GetConnectionStateErrorSql returning invalid query with additional parameter. Closes #3896
Update FDW to v1.8.0-rc.11 - fix failure to load connection config when importing schema

pkg/connection/config_map.go

@@ -43,8 +43,8 @@ func (m ConnectionConfigMap) Diff(otherMap ConnectionConfigMap) (addedConnection
 			// check for changes
 
 			// special case - if the plugin has changed, treat this as a deletion and a re-add
-			if connection.PluginInstance != otherConnection.Plugin {
-				addedConnections[otherConnection.Plugin] = append(addedConnections[otherConnection.Plugin], otherConnection)
+			if connection.PluginInstance != otherConnection.PluginInstance {
+				addedConnections[otherConnection.PluginInstance] = append(addedConnections[otherConnection.PluginInstance], otherConnection)
 				deletedConnections[connection.PluginInstance] = append(deletedConnections[connection.PluginInstance], connection)
 			} else {
 				if !connection.Equals(otherConnection) {

pkg/constants/db.go

@@ -28,7 +28,7 @@ const (
 // constants for installing db and fdw images
 const (
 	DatabaseVersion = "14.2.0"
-	FdwVersion      = "1.8.0-rc.10"
+	FdwVersion      = "1.8.0-rc.11"
 
 	// PostgresImageRef is the OCI Image ref for the database binaries
 	PostgresImageRef    = "us-docker.pkg.dev/steampipe/steampipe/db:14.2.0"

pkg/introspection/connection_table_sql.go

@@ -62,7 +62,7 @@ WHERE
 	name = $2
 	`,
 		constants.InternalSchema, constants.ConnectionTable, constants.ConnectionStateError)
-	args := []any{constants.ConnectionStateError, err.Error(), connectionName}
+	args := []any{err.Error(), connectionName}
 	return db_common.QueryWithArgs{Query: query, Args: args}
 }
 

pkg/pluginmanager_service/plugin_manager_rate_limiters.go

@@ -98,7 +98,7 @@ func (m *PluginManager) handleUserLimiterChanges(_ context.Context, plugins conn
 		log.Println("[WARN] could not refresh rate limiter table", err)
 	}
 
-	// now update the plugins - call setRateLimiters for any plugin witrh updated user limiters
+	// now update the plugins - call setRateLimiters for any plugin with updated user limiters
 	for p := range pluginsWithChangedLimiters {
 		if err := m.setRateLimitersForPlugin(p); err != nil {
 			return err

pkg/steampipeconfig/load_config.go

@@ -268,9 +268,9 @@ func loadConfig(configFolder string, steampipeConfig *SteampipeConfig, opts *loa
 			if moreDiags.HasErrors() {
 				continue
 			}
-			_, alreadyThere := steampipeConfig.Connections[connection.Name]
-			if alreadyThere {
-				return error_helpers.NewErrorsAndWarning(sperr.New("duplicate connection name: '%s' in '%s'", connection.Name, block.TypeRange.Filename))
+			if existingConnection, alreadyThere := steampipeConfig.Connections[connection.Name]; alreadyThere {
+				err := getDuplicateConnectionError(existingConnection, connection)
+				return error_helpers.NewErrorsAndWarning(err)
 			}
 			if ok, errorMessage := db_common.IsSchemaNameValid(connection.Name); !ok {
 				return error_helpers.NewErrorsAndWarning(sperr.New("invalid connection name: '%s' in '%s'. %s ", connection.Name, block.TypeRange.Filename, errorMessage))
@@ -324,6 +324,12 @@ func loadConfig(configFolder string, steampipeConfig *SteampipeConfig, opts *loa
 	return res
 }
 
+func getDuplicateConnectionError(existingConnection, newConnection *modconfig.Connection) error {
+	return sperr.New("duplicate connection name: '%s'\n\t(%s:%d)\n\t(%s:%d)",
+		existingConnection.Name, existingConnection.DeclRange.Filename, existingConnection.DeclRange.Start.Line,
+		newConnection.DeclRange.Filename, newConnection.DeclRange.Start.Line)
+}
+
 func optionsBlockPermitted(block *hcl.Block, blockMap map[string]bool, opts *loadConfigOptions) error {
 	// keep track of duplicate block types
 	blockType := block.Labels[0]

pkg/steampipeconfig/postgres_notification.go

@@ -42,6 +42,6 @@ func NewErrorsAndWarningsNotification(errorAndWarnings *error_helpers.ErrorAndWa
 	if errorAndWarnings.Error != nil {
 		res.Errors = []string{errorAndWarnings.Error.Error()}
 	}
-	res.Warnings = append(res.Errors, errorAndWarnings.Warnings...)
+	res.Warnings = append(res.Warnings, errorAndWarnings.Warnings...)
 	return res
 }

pkg/steampipeconfig/steampipeconfig.go

@@ -338,8 +338,8 @@ func (c *SteampipeConfig) ConnectionList() []*modconfig.Connection {
 // add a plugin config to PluginsInstances and Plugins
 // NOTE: this returns an error if we alreayd have a config with the same label
 func (c *SteampipeConfig) addPlugin(plugin *modconfig.Plugin, block *hcl.Block) error {
-	if _, exists := c.PluginsInstances[plugin.Instance]; exists {
-		return sperr.New("duplicate plugin: '%s' in '%s'", plugin.Alias, block.TypeRange.Filename)
+	if existingPlugin, exists := c.PluginsInstances[plugin.Instance]; exists {
+		return duplicatePluginError(existingPlugin, plugin)
 	}
 	// get the image ref to key the map
 	imageRef := plugin.GetImageRef()
@@ -349,6 +349,12 @@ func (c *SteampipeConfig) addPlugin(plugin *modconfig.Plugin, block *hcl.Block)
 	return nil
 }
 
+func duplicatePluginError(existingPlugin, newPlugin *modconfig.Plugin) error {
+	return sperr.New("duplicate plugin instance: '%s'\n\t(%s:%d)\n\t(%s:%d)",
+		existingPlugin.Instance, *existingPlugin.FileName, *existingPlugin.StartLineNumber,
+		*newPlugin.FileName, *newPlugin.StartLineNumber)
+}
+
 // ensure we have a plugin config struct for all plugins mentioned in connection config,
 // even if there is not an explicit HCL config for it
 // NOTE: this populates the  Plugin ans PluginInstance field of the connections