Do not connect to the database if the replication configuration is incorrect (#8381)

* Do not connect to the database if the replication configuration is incorrect

Configuration errors can lead to loss of synchronization between master and replica.

* Send config errors to the client app only if report_errors is enabled

* Report a replication error if it occurs during the initial connection to the replica and the report_errors is enabled

* Add error "all configured replicas skipped"

Author: XaBbl4

src/jrd/replication/Config.cpp

@@ -62,6 +62,7 @@ namespace
 	const ULONG DEFAULT_GROUP_FLUSH_DELAY = 0;
 	const ULONG DEFAULT_APPLY_IDLE_TIMEOUT = 10;				// seconds
 	const ULONG DEFAULT_APPLY_ERROR_TIMEOUT = 60;				// seconds
+	const bool DEFAULT_REPORT_ERRORS = false;
 
 	void parseLong(const string& input, ULONG& output)
 	{
@@ -79,36 +80,53 @@ namespace
 			output = false;
 	}
 
-	void configError(const string& type, const string& key, const string& value)
+	void configError(CheckStatusWrapper* status, const string& type, const string& key, const string& value)
 	{
-		raiseError("%s specifies %s: %s", key.c_str(), type.c_str(), value.c_str());
+		string msg;
+		if (!(status->getState() & IStatus::STATE_ERRORS))
+		{
+			msg.printf("Incorrect entry in %s", REPLICATION_CFGFILE);
+			(Arg::Gds(isc_random) << Arg::Str(msg)).appendTo(status);
+		}
+
+		msg.printf("%s specifies %s: %s", key.c_str(), type.c_str(), value.c_str());
+		(Arg::Gds(isc_random) << Arg::Str(msg)).appendTo(status);
 	}
 
-	void checkAccess(const PathName& path, const string& key)
+	bool checkAccess(CheckStatusWrapper* status, const PathName& path, const string& key)
 	{
 		if (path.hasData() && !PathUtils::canAccess(path, 6))
-			configError("missing or inaccessible directory", key, path.c_str());
+		{
+			configError(status, "missing or inaccessible directory", key, path.c_str());
+			return false;
+		}
+		return true;
 	}
 
 	void composeError(CheckStatusWrapper* status, const Exception& ex)
 	{
-		string prefix;
-		prefix.printf("Incorrect entry in %s", REPLICATION_CFGFILE);
-
 		Arg::StatusVector sv;
-		sv << Arg::Gds(isc_random) << Arg::Str(prefix);
+
+		if (!(status->getState() & IStatus::STATE_ERRORS))
+		{
+			string prefix;
+			prefix.printf("Incorrect entry in %s", REPLICATION_CFGFILE);
+			sv << Arg::Gds(isc_random) << Arg::Str(prefix);
+		}
+
+		sv << Arg::StatusVector(status);
 		sv << Arg::StatusVector(ex);
 
 		status->setErrors(sv.value());
 	}
 
-	void parseExternalValue(const string& key, const string& value, string& output)
+	bool parseExternalValue(CheckStatusWrapper* status, const string& key, const string& value, string& output)
 	{
 		const auto pos = key.rfind('_');
 		if (pos == string::npos)
 		{
 			output = value.c_str();
-			return;
+			return true;
 		}
 
 		string temp;
@@ -118,7 +136,10 @@ namespace
 		{
 			fb_utils::readenv(value.c_str(), temp);
 			if (temp.isEmpty())
-				configError("missing environment variable", key, value);
+			{
+				configError(status, "missing environment variable", key, value);
+				return false;
+			}
 		}
 		else if (key_source.equals(KEY_SUFFIX_FILE))
 		{
@@ -129,44 +150,65 @@ namespace
 
 			AutoPtr<FILE> file(os_utils::fopen(filename.c_str(), "rt"));
 			if (!file)
-				configError("missing or inaccessible file", key, filename.c_str());
+			{
+				configError(status, "missing or inaccessible file", key, filename.c_str());
+				return false;
+			}
 
 			if (temp.LoadFromFile(file))
 				temp.alltrim("\r");
 
 			if (temp.isEmpty())
-				configError("first empty line of file", key, filename.c_str());
+			{
+				configError(status, "first empty line of file", key, filename.c_str());
+				return false;
+			}
 		}
 
 		output = temp.c_str();
+		return true;
 	}
 
-	void parseSyncReplica(const ConfigFile::Parameters& params, SyncReplica& output)
+	bool parseSyncReplica(CheckStatusWrapper* status, const ConfigFile::Parameters& params, SyncReplica& output)
 	{
+		bool result = true;
 		for (const auto& el : params)
 		{
 			const string key(el.name.c_str());
 			const string value(el.value);
 
 			if (value.isEmpty())
-				continue;
+			{
+				configError(status, "empty value", output.database, key);
+				result = false;
+			}
 
 			if (key.find("username") == 0)
 			{
 				if (output.username.hasData())
-					configError("multiple values", output.database, "username");
-				parseExternalValue(key, value, output.username);
+				{
+					configError(status, "multiple values", output.database, "username");
+					result = false;
+				}
+				result &= parseExternalValue(status, key, value, output.username);
 				output.username.rtrim(" ");
 			}
 			else if (key.find("password") == 0)
 			{
 				if (output.password.hasData())
-					configError("multiple values", output.database, "password");
-				parseExternalValue(key, value, output.password);
+				{
+					configError(status, "multiple values", output.database, "password");
+					result = false;
+				}
+				result &= parseExternalValue(status, key, value, output.password);
 			}
 			else
-				configError("unknown parameter", output.database, key);
+			{
+				configError(status, "unknown key", output.database, key);
+				result = false;
+			}
 		}
+		return result;
 	}
 }
 
@@ -194,7 +236,7 @@ Config::Config()
 	  applyErrorTimeout(DEFAULT_APPLY_ERROR_TIMEOUT),
 	  pluginName(getPool()),
 	  logErrors(true),
-	  reportErrors(false),
+	  reportErrors(DEFAULT_REPORT_ERRORS),
 	  disableOnError(true),
 	  cascadeReplication(false)
 {
@@ -234,6 +276,7 @@ Config* Config::get(const PathName& lookupName)
 {
 	fb_assert(lookupName.hasData());
 
+	bool reportErrors = DEFAULT_REPORT_ERRORS;
 	try
 	{
 		const PathName filename =
@@ -245,7 +288,8 @@ Config* Config::get(const PathName& lookupName)
 
 		AutoPtr<Config> config(FB_NEW Config);
 
-		bool defaultFound = false, exactMatch = false;
+		FbLocalStatus localStatus;
+		bool defaultFound = false, exactMatch = false, replicaSkip = false;
 
 		for (const auto& section : cfgFile.getParameters())
 		{
@@ -283,15 +327,24 @@ Config* Config::get(const PathName& lookupName)
 				string value(el.value);
 
 				if (value.isEmpty())
+				{
+					configError(&localStatus, "empty value of key",
+					                          exactMatch ? lookupName.c_str() : section.name.c_str(),
+					                          key);
 					continue;
+				}
 
 				if (key == "sync_replica")
 				{
 					SyncReplica syncReplica(config->getPool());
 					if (el.sub)
 					{
 						syncReplica.database = value;
-						parseSyncReplica(el.sub->getParameters(), syncReplica);
+						if (!parseSyncReplica(&localStatus, el.sub->getParameters(), syncReplica))
+						{
+							replicaSkip = true;
+							continue;
+						}
 					}
 					else
 						splitConnectionString(value, syncReplica.database, syncReplica.username, syncReplica.password);
@@ -324,7 +377,12 @@ Config* Config::get(const PathName& lookupName)
 				{
 					config->journalDirectory = value.c_str();
 					PathUtils::ensureSeparator(config->journalDirectory);
-					checkAccess(config->journalDirectory, key);
+					if (!checkAccess(&localStatus, config->journalDirectory, key))
+					{
+						config->journalDirectory.erase();
+						replicaSkip = true;
+						continue;
+					}
 				}
 				else if (key == "journal_file_prefix")
 				{
@@ -338,7 +396,11 @@ Config* Config::get(const PathName& lookupName)
 				{
 					config->archiveDirectory = value.c_str();
 					PathUtils::ensureSeparator(config->archiveDirectory);
-					checkAccess(config->archiveDirectory, key);
+					if (!checkAccess(&localStatus, config->archiveDirectory, key))
+					{
+						config->archiveDirectory.erase();
+						continue;
+					}
 				}
 				else if (key == "journal_archive_command")
 				{
@@ -359,6 +421,7 @@ Config* Config::get(const PathName& lookupName)
 				else if (key == "report_errors")
 				{
 					parseBoolean(value, config->reportErrors);
+					reportErrors = config->reportErrors;
 				}
 				else if (key == "disable_on_error")
 				{
@@ -368,12 +431,28 @@ Config* Config::get(const PathName& lookupName)
 				{
 					parseBoolean(value, config->cascadeReplication);
 				}
+				else if ((key != "journal_source_directory") &&
+						(key != "source_guid") &&
+						(key != "verbose_logging") &&
+						(key != "apply_idle_timeout") &&
+						(key != "apply_error_timeout"))
+				{
+					configError(&localStatus, "unknown key",
+					                          exactMatch ? lookupName.c_str() : section.name.c_str(),
+					                          key);
+				}
 			}
 
 			if (exactMatch)
 				break;
 		}
 
+		if (localStatus->getState() & IStatus::STATE_ERRORS)
+			logPrimaryStatus(lookupName, &localStatus);
+
+		if (replicaSkip && !config->disableOnError)
+			raiseError("One or more replicas configured with errors");
+
 		// TODO: As soon as plugin name is moved into RDB$PUBLICATIONS,
 		// delay config parse until real replication start
 		if (config->pluginName.hasData())
@@ -396,13 +475,18 @@ Config* Config::get(const PathName& lookupName)
 
 			return config.release();
 		}
+
+		if (replicaSkip)
+			raiseError("All configured replicas skipped");
 	}
 	catch (const Exception& ex)
 	{
 		FbLocalStatus localStatus;
 		composeError(&localStatus, ex);
 
 		logPrimaryStatus(lookupName, &localStatus);
+		if (reportErrors)
+			localStatus.raise();
 	}
 
 	return nullptr;
@@ -414,6 +498,7 @@ Config* Config::get(const PathName& lookupName)
 void Config::enumerate(ReplicaList& replicas)
 {
 	PathName dbName;
+	FbLocalStatus localStatus;
 
 	try
 	{
@@ -466,11 +551,19 @@ void Config::enumerate(ReplicaList& replicas)
 				{
 					config->sourceDirectory = value.c_str();
 					PathUtils::ensureSeparator(config->sourceDirectory);
+					if (!checkAccess(&localStatus, config->sourceDirectory, key))
+					{
+						config->sourceDirectory.erase();
+						continue;
+					}
 				}
 				else if (key == "source_guid")
 				{
 					if (!StringToGuid(&config->sourceGuid, value.c_str()))
-						configError("invalid (misformatted) value", key, value);
+					{
+						configError(&localStatus, "invalid (misformatted) value", key, value);
+						continue;
+					}
 				}
 				else if (key == "verbose_logging")
 				{
@@ -500,11 +593,11 @@ void Config::enumerate(ReplicaList& replicas)
 	}
 	catch (const Exception& ex)
 	{
-		FbLocalStatus localStatus;
 		composeError(&localStatus, ex);
+	}
 
+	if (localStatus->getState() & IStatus::STATE_ERRORS)
 		logReplicaStatus(dbName, &localStatus);
-	}
 }
 
 // This routine is used for split input connection string to parts

src/jrd/replication/Manager.cpp

@@ -161,6 +161,8 @@ Manager::Manager(const string& dbId,
 		if (localStatus->getState() & IStatus::STATE_ERRORS)
 		{
 			logPrimaryStatus(m_config->dbName, &localStatus);
+			if (m_config->reportErrors)
+				(Arg::Gds(isc_repl_error) << Arg::StatusVector(&localStatus)).raise();
 			continue;
 		}
 
@@ -169,6 +171,8 @@ Manager::Manager(const string& dbId,
 		{
 			logPrimaryStatus(m_config->dbName, &localStatus);
 			attachment->detach(&localStatus);
+			if (m_config->reportErrors)
+				(Arg::Gds(isc_repl_error) << Arg::StatusVector(&localStatus)).raise();
 			continue;
 		}