DM-54631: Fixed bugs, race conditions, and minor problems in the code of the Replication/Ingest system of Qserv

src/replica/apps/FileIngestApp.h

@@ -71,7 +71,7 @@ class FileIngestApp : public Application {
      *   },
      *   ...
      * ]
-     * @code
+     * @endcode
      * Otherwise, the following schema will be expected:
      * @code
      * [
@@ -84,7 +84,7 @@ class FileIngestApp : public Application {
      *   },
      *   ...
      * ]
-     * @code
+     * @endcode
      *
      * Notes on the values of the parameters :
      * - "worker-host" is a DNS name or an IP address
@@ -127,7 +127,7 @@ class FileIngestApp : public Application {
      * @code
      *   chunk_<number>.txt
      *   chunk_<number>_overlap.txt
-     * @code
+     * @endcode
      * @param filename the name of the chunk contribution files
      * @return an object encapsulating the attributes
      */

src/replica/apps/QservWorkerApp.h

@@ -69,12 +69,12 @@ class QservWorkerApp : public Application {
      * the following format:
      * @code
      *   <database>:<chunk>
-     * @code
+     * @endcode
      * For example:
      * @code
      *   LSST:123 LSST:124 LSST:23456
      *   LSST:0
-     * @code
+     * @endcode
      * @param replicas collection to be initialized
      * @param databases collection of unique database names found in the input files. The collection
      *   will be populated upon a successful completion of the method.

src/replica/config/ConfigCzar.cc

@@ -61,7 +61,7 @@ json ConfigCzar::toJson() const {
 }
 
 bool ConfigCzar::operator==(ConfigCzar const& other) const {
-    return (name == other.name) && (host == other.host) && (port == other.port);
+    return (name == other.name) && (id == other.id) && (host == other.host) && (port == other.port);
 }
 
 ostream& operator<<(ostream& os, ConfigCzar const& info) {

src/replica/config/ConfigCzar.h

@@ -41,10 +41,10 @@ namespace lsst::qserv::replica {
  */
 class ConfigCzar {
 public:
-    std::string name;   ///< The logical name of a Czar.
-    qmeta::CzarId id;   ///< The unique name of a Czar.
-    ConfigHost host;    ///< The host name (and IP address) of the Czar management service.
-    uint16_t port = 0;  ///< The port number of the Czar management service.
+    std::string name;      ///< The logical name of a Czar.
+    qmeta::CzarId id = 0;  ///< The unique identifier of a Czar.
+    ConfigHost host;       ///< The host name (and IP address) of the Czar management service.
+    uint16_t port = 0;     ///< The port number of the Czar management service.
 
     /**
      * Construct from a JSON object.

src/replica/config/Configuration.h

@@ -119,7 +119,7 @@ class Configuration {
      * from the following source:
      * @code
      *   mysql://[user][:password][@host][:port][/database]
-     * @code
+     * @endcode
      * @note Configuration objects initialized from MySQL would rely on MySQL as
      *   the persistent backend for any requests to update the state of the transient
      *   parameters. A connection object to the MySQL service will be initialized

src/replica/contr/DeleteWorkerTask.cc

@@ -60,7 +60,7 @@ void DeleteWorkerTask::onStart() {
 
     _logStartedEvent(jobs[0]);
     track<DeleteWorkerJob>(DeleteWorkerJob::typeName(), jobs, numFinishedJobs);
-    _logStartedEvent(jobs[0]);
+    _logFinishedEvent(jobs[0]);
 }
 
 void DeleteWorkerTask::_logStartedEvent(DeleteWorkerJob::Ptr const& job) const {

src/replica/contr/ReplicationTask.cc

@@ -136,7 +136,7 @@ void ReplicationTask::_updateChunkMap() {
     }
 
     // Open MySQL connection using the RAII-style handler that would automatically
-    // abort the transaction should any problem occured when loading data into the table.
+    // abort the transaction should any problem occurred when loading data into the table.
     ConnectionHandler h;
     try {
         h.conn = Connection::open(Configuration::qservCzarDbParams("qservMeta"));

src/replica/contr/ReplicationTask.h

@@ -96,7 +96,7 @@ class ReplicationTask : public Task {
     /// on the Qserv synchronization requests.
     unsigned int const _qservSyncTimeoutSec;
 
-    bool const _disableQservSync;     ///< Disable replica synchroization at Qserv workers if 'true'.
+    bool const _disableQservSync;     ///< Disable replica synchronization at Qserv workers if 'true'.
     bool const _forceQservSync;       ///< Force removal at worker resource collections if 'true'.
     bool const _qservChunkMapUpdate;  ///< Enable updating the chunk disposition map in Qserv's QMeta
                                       /// database if 'true'.

src/replica/ingest/IngestClient.cc

@@ -105,8 +105,8 @@ void IngestClient::send() {
         ProtocolIngestData data;
         eof = !file.read(record.get(), _recordSizeBytes);
         if (eof && !file.eof()) {
-            _abort(__func__, "failed to open the file: '" + _inputFilePath + +"', error: '" +
-                                     strerror(errno) + "', errno: " + to_string(errno));
+            _abort(__func__, "failed to open the file: '" + _inputFilePath + "', error: '" + strerror(errno) +
+                                     "', errno: " + to_string(errno));
         }
         size_t const num = file.gcount();
         _sizeBytes += num;

src/replica/ingest/IngestDataHttpSvcMod.cc

@@ -119,7 +119,7 @@ json IngestDataHttpSvcMod::_syncProcessData() {
     // is made the persistent state of the destination table is supposed to be changed.
     _contrib.retryAllowed = true;
 
-    // This parameters sets a limit foe the number of warnings (should there be any)
+    // This parameters sets a limit for the number of warnings (should there be any)
     // reported by MySQL after contribution loading attempt. Warnings is an important
     // mechanism for debugging problems with the ingested data.
     _contrib.maxNumWarnings = body().optional<unsigned int>(

src/replica/ingest/IngestFileHttpSvcMod.cc

@@ -100,7 +100,7 @@ void IngestFileHttpSvcMod::onStartOfFile(string const& name, string const& fileN
     // is made the persistent state of the destination table is supposed to be changed.
     _contrib.retryAllowed = true;
 
-    // This parameters sets a limit foe the number of warnings (should there be any)
+    // This parameters sets a limit for the number of warnings (should there be any)
     // reported by MySQL after contribution loading attempt. Warnings is an important
     // mechanism for debugging problems with the ingested data.
     _contrib.maxNumWarnings = body().optionalUInt("max_num_warnings",

src/replica/ingest/IngestFileSvc.h

@@ -134,7 +134,7 @@ class IngestFileSvc {
     unsigned int _numWarnings = 0;
     std::list<database::mysql::Warning> _warnings;
 
-    uint64_t _numRowsLoaded = 0;  ///< Th enumber of rows actually ingested into Qserv.
+    uint64_t _numRowsLoaded = 0;  ///< The number of rows actually ingested into Qserv.
 };
 
 }  // namespace lsst::qserv::replica

src/replica/ingest/IngestRequestMgr.cc

@@ -136,7 +136,7 @@ shared_ptr<IngestRequestMgr> IngestRequestMgr::create(shared_ptr<ServiceProvider
         //     contrib.loadTime is guaranteed to be set to 0
         //
         //   contrib.loadTime -- is set to a non-zero value by a processing thread after finishing
-        //     uploading the content of teh temporary file into MySQL. At this point the request
+        //     uploading the content of the temporary file into MySQL. At this point the request
         //     is supposed to be completed and moved into the output queue.
         //
         // The auto-resume algorithm is evaluating a progress of requests from the newest

src/replica/jobs/DirectorIndexJob.cc

@@ -373,7 +373,7 @@ void DirectorIndexJob::_loadDataIntoTable() {
     LOGS(_log, LOG_LVL_DEBUG, context_);
 
     // Open MySQL connection using the RAII-style handler that would automatically
-    // abort the transaction should any problem occured when loading data into the table.
+    // abort the transaction should any problem occurred when loading data into the table.
     Connection::Ptr conn;
     try {
         conn = Connection::open(Configuration::qservCzarDbParams(lsst::qserv::SEC_INDEX_DB));

src/replica/jobs/Job.h

@@ -93,7 +93,7 @@ class Job : public std::enable_shared_from_this<Job> {
     }
 
     /// Structure Progress captures counters for the tasks completed by the jobs
-    /// adn the total number of tasks to be processed by the job.
+    /// and the total number of tasks to be processed by the job.
     struct Progress {
         size_t complete = 0;
         size_t total = 1;
@@ -257,7 +257,7 @@ class Job : public std::enable_shared_from_this<Job> {
      *   void T::notify(replica::Lock const& lock) {
      *       notifyDefaultImpl<T>(lock, _onFinish);
      *   }
-     * @code
+     * @endcode
      * @see Job::notifyDefaultImpl
      * @param lock  A lock on Job::_mtx must be acquired by a caller of the method.
      */

src/replica/jobs/SqlJobResult.h

@@ -74,7 +74,7 @@ class SqlJobResult {
      * @code
      *   worker | [scopeName] | status | error
      *  --------+-------------+--------+-------
-     * @code
+     * @endcode
      *
      * First three parameters ('caption', 'indent', and 'verticalSeparator') of
      * the method are the same as for the constructor of the table printer class.
@@ -98,7 +98,7 @@ class SqlJobResult {
      * @code
      *   worker | #succeeded | #failed | performance [sec]
      *  --------+------------+---------+-------------------
-     * @code
+     * @endcode
      *
      * First three parameters ('caption', 'indent', and 'verticalSeparator') of
      * the method are the same as for the constructor of the table printer class.

src/replica/mysql/DatabaseMySQL.h

@@ -316,7 +316,7 @@ class Connection : public std::enable_shared_from_this<Connection> {
      *   if needed.
      *
      * @return a pointer to the same connector against which the method was invoked
-     *   in case of successful completion of the requested operaton.
+     *   in case of successful completion of the requested operation.
      *
      * @see Configuration::databaseMaxReconnects()
      * @see Configuration::setDatabaseMaxReconnects()

src/replica/mysql/DatabaseMySQLExceptions.h

@@ -176,7 +176,7 @@ class ER_NO_SUCH_TABLE_ : public Error {
  * a non-partitioned table:
  * @code
  *   ALTER TABLE <database>.<table> REMOVE PARTITIONING;
- * @code
+ * @endcode
  * Some application may choose to explicitly identify and process this type
  * of failures.
  */
@@ -190,7 +190,7 @@ class ER_PARTITION_MGMT_ON_NONPARTITIONED_ : public Error {
  * of a partitioned table:
  * @code
  *   SELECT * FROM <database>.<table> PARTITION (<partition>);
- * @code
+ * @endcode
  * Some application may choose to explicitly identify and process this type
  * of failures.
  */
@@ -204,7 +204,7 @@ class ER_UNKNOWN_PARTITION_ : public Error {
  * the table's definition:
  * @code
  *   ALTER TABLE <database>.<table> DROP PARTITION <partition>;
- * @code
+ * @endcode
  */
 class ER_DROP_PARTITION_NON_EXISTENT_ : public Error {
 public:
@@ -224,7 +224,7 @@ class ER_LOCK_DEADLOCK_ : public Error {
  * Thrown on foreign key constraint violations. MySQL reports this error when you try to add
  * a row but there is no parent row, and a foreign key constraint fails. Add the parent row first.
  * This exception may also indicate the normal scenario when a parent row was deleted on purpose.
- * In this cases, the application shoiuld catch this exception and handle it appropriately.
+ * In this cases, the application should catch this exception and handle it appropriately.
  */
 class ER_NO_REFERENCED_ROW_2_ : public Error {
 public:

src/replica/mysql/DatabaseMySQLGenerator.h

@@ -122,7 +122,7 @@ class Sql : public DoNotProcess {
     ~Sql() override = default;
 };
 
-/// The enumerator type dwefining a scope for a variable(s)
+/// The enumerator type defining a scope for a variable(s)
 enum class SqlVarScope : int { SESSION = 1, GLOBAL = 2 };
 
 /**
@@ -1101,7 +1101,7 @@ class QueryGenerator {
      * @code
      *  SET GLOBAL `var1`=1
      *  SET `var2`=2,`var3`='abc'
-     * @code
+     * @endcode
      *
      * @param scope The scope of the variable (SESSION, GLOBAL, etc.)
      * @param Fargs A collection of pairs specifying variable names and their values.
@@ -1124,7 +1124,7 @@ class QueryGenerator {
      * The generator will produce this statement:
      * @code
      *  CALL QSERV_MANAGER('abc')
-     * @code
+     * @endcode
      *
      * @param packedProcAndArgs The well-formed SQL for the procedure and its parameters
      * @return Well-formed SQL statement.

src/replica/mysql/DatabaseMySQLTypes.h

@@ -57,11 +57,11 @@ class ConnectionParams {
      * expected to have the following syntax:
      * @code
      *   mysql://[user][:password]@[host][:port][/database]
-     * @code
+     * @endcode
      * The minimal (though, totally useless) URI would be:
      * @code
      *   mysql://@
-     * @code
+     * @endcode
      *
      * @note
      *   1) all attributes are optional

src/replica/mysql/DatabaseMySQLUtils.h

@@ -65,7 +65,7 @@ bool selectSingleValueImpl(std::shared_ptr<Connection> const& conn, std::string
  * @return 'true' if the value is not NULL.
  *
  * @throws logic_error If the query is not supposed to return any result set,
- *   or if the result set has nore than 1 row (unless noMoreThanOne = false).
+ *   or if the result set has more than 1 row (unless noMoreThanOne = false).
  * @throws EmptyResultSetError If a result set is empty.
  * @throws InvalidTypeError If the conversion to a proposed type will fail.
  */
@@ -86,7 +86,7 @@ inline bool selectSingleValue(std::shared_ptr<Connection> const& conn, std::stri
 
 /**
  * Report info on the on-going queries using 'SHOW [FULL] PROCESSLIST'.
- * @param A scope of the operaton depends on the user credentials privided
+ * @param A scope of the operation depends on the user credentials provided
  *   in the configuration object. Normally, a subset of queries which belong
  *   to the specified user will be reported.
  * @param conn The MySQL connection for executing the query.

src/replica/qserv/GetStatusQservMgtRequest.h

@@ -41,7 +41,7 @@ namespace lsst::qserv::replica {
  * The resulting query will have the following format:
  * @code
  * ?include_tasks=<numeric_flag>&max_tasks=<num>[&query_ids=<id1>[,<id2>...][&task_states=<state1>[,<state2>...]
- * @code
+ * @endcode
  * @param taskSelector The selector to be processed.
  * @return The corresponding query.
  */

src/replica/qserv/QservMgtRequest.h

@@ -263,7 +263,7 @@ class QservMgtRequest : public std::enable_shared_from_this<QservMgtRequest> {
      *   void T::notify(replica::Lock const& lock) {
      *       notifyDefaultImpl<T>(lock, _onFinish);
      *   }
-     * @code
+     * @endcode
      * @see QservMgtRequest::notifyDefaultImpl
      * @param lock A lock on QservMgtRequest::_mtx must be acquired before
      *   calling this method.

src/replica/requests/DirectorIndexRequest.h

@@ -68,7 +68,7 @@ class DirectorIndexRequest : public RequestMessenger {
     DirectorIndexRequest& operator=(DirectorIndexRequest const&) = delete;
 
     /// Non-trivial destructor is needed to delete the data file that is created
-    /// upon successfull completion of the request.
+    /// upon successful completion of the request.
     virtual ~DirectorIndexRequest() final;
 
     std::string const& database() const { return _database; }

src/replica/requests/MessengerConnector.h

@@ -160,11 +160,11 @@ class MessageWrapper : public MessageWrapperBase {
             try {
                 responseBuffer().parse(response, responseBuffer().size());
             } catch (std::runtime_error const& ex) {
-                // The message is corrupt. Google Protobuf will report an error
-                // of the following kind:
-                // @code
-                //   [libprotobuf ERROR google/protobuf/message_lite.cc:123] Can't parse message...
-                // @code
+                /// The message is corrupt. Google Protobuf will report an error
+                /// of the following kind:
+                /// @code
+                ///   [libprotobuf ERROR google/protobuf/message_lite.cc:123] Can't parse message...
+                /// @endcode
                 setSuccess(false);
             }
         }

src/replica/requests/Request.h

@@ -476,7 +476,7 @@ class Request : public std::enable_shared_from_this<Request> {
      *   void T::notify(replica::Lock const& lock) {
      *       notifyDefaultImpl<T>(lock, _onFinish);
      *   }
-     * @code
+     * @endcode
      *
      * @see Request::notifyDefaultImpl
      * @param lock A lock on Request::_mtx must be acquired before calling this method.

src/replica/requests/RequestMessenger.h

@@ -49,7 +49,7 @@ class RequestMessenger : public Request {
     /// The pointer type for instances of the class
     typedef std::shared_ptr<RequestMessenger> Ptr;
 
-    /// The callaback type for notifications on completion of the request
+    /// The callback type for notifications on completion of the request
     /// disposal operation. The first parameter (std::string const&) of the callback
     /// is the unique identifier of a request, the second parameter (bool) is a flag
     /// indicating a success or a failure of the operation, and the last parameter

src/replica/requests/SqlResultSet.cc

@@ -300,7 +300,7 @@ void SqlResultSet::iterate(SqlResultSet::ResultSet const& resultSet,
 
 ostream& operator<<(ostream& os, SqlResultSet const& info) {
     os << "SqlResultSet:{qservResultSet:[";
-    bool first = false;
+    bool first = true;
     for (auto&& itr : info.queryResultSet) {
         auto&& scope = itr.first;
         auto&& result = itr.second;

src/replica/services/DatabaseServicesPool.cc

@@ -20,7 +20,7 @@
  */
 
 // Class header
-#include "replica//services/DatabaseServicesPool.h"
+#include "replica/services/DatabaseServicesPool.h"
 
 // System headers
 #include <stdexcept>

src/replica/services/ServiceProvider.h

@@ -141,7 +141,7 @@ class ServiceProvider : public std::enable_shared_from_this<ServiceProvider> {
      *   replica::Lock lock(mutex);
      *   // The better option
      *   replica::Lock lock(serviceProvider->getNamedMutex("name"));
-     * @code
+     * @endcode
      * Class replica::Lock makes a copy of the shared pointer for a duration of the lock.
      *
      * If, for some reason, a code resorts to using low-level references/pointers to the stored mutex
@@ -151,7 +151,7 @@ class ServiceProvider : public std::enable_shared_from_this<ServiceProvider> {
      *   auto mutex = serviceProvider->getNamedMutex("name");
      *   replica::Lock lock(*mutex);
      *   std::lock_guard<replica::Mutex> lock(*mutex);
-     * @code
+     * @endcode
      * Though, in general this would work, the above shown dereferencing is not recommended.
      *
      * @param name The name of a named mutex.