QPID-5767: [C++ broker][linearstore] broker segfaults when recovering journal file with damaged header (Kim's patch)

git-svn-id: https://svn.apache.org/repos/asf/qpid/trunk@1596509 13f79535-47bb-0310-9956-ffa450edef68

6 files changed, 27 insertions, 11 deletions

diff --git a/qpid/cpp/src/qpid/linearstore/journal/EmptyFilePoolManager.cpp b/qpid/cpp/src/qpid/linearstore/journal/EmptyFilePoolManager.cpp
index ab70e3825e..162550a984 100644
--- a/qpid/cpp/src/qpid/linearstore/journal/EmptyFilePoolManager.cpp
+++ b/qpid/cpp/src/qpid/linearstore/journal/EmptyFilePoolManager.cpp
@@ -184,9 +184,9 @@ EmptyFilePool* EmptyFilePoolManager::getEmptyFilePool(const efpIdentity_t efpIde
 
 EmptyFilePool* EmptyFilePoolManager::getEmptyFilePool(const efpPartitionNumber_t partitionNumber,
                                                       const efpDataSize_kib_t efpDataSize_kib) {
-    EmptyFilePoolPartition* efppp = getEfpPartition(partitionNumber);
+    EmptyFilePoolPartition* efppp = getEfpPartition(partitionNumber > 0 ? partitionNumber : defaultPartitionNumber_);
     if (efppp != 0)
-        return efppp->getEmptyFilePool(efpDataSize_kib);
+	return efppp->getEmptyFilePool(efpDataSize_kib > 0 ? efpDataSize_kib : defaultEfpDataSize_kib_);
     return 0;
 }
 
diff --git a/qpid/cpp/src/qpid/linearstore/journal/LinearFileController.cpp b/qpid/cpp/src/qpid/linearstore/journal/LinearFileController.cpp
index 86d1b0e93c..7984109a3a 100644
--- a/qpid/cpp/src/qpid/linearstore/journal/LinearFileController.cpp
+++ b/qpid/cpp/src/qpid/linearstore/journal/LinearFileController.cpp
@@ -102,6 +102,7 @@ void LinearFileController::removeFileToEfp(const std::string& fileName) {
 }
 
 void LinearFileController::restoreEmptyFile(const std::string& fileName) {
+    // TODO: Add checks that this file is of a valid size; if not, delete this and get one from the EFP
     addJournalFile(fileName, emptyFilePoolPtr_->getIdentity(), getNextFileSeqNum(), 0);
 }
 
diff --git a/qpid/cpp/src/qpid/linearstore/journal/RecoveryManager.cpp b/qpid/cpp/src/qpid/linearstore/journal/RecoveryManager.cpp
index 940048b464..a8311ccbad 100644
--- a/qpid/cpp/src/qpid/linearstore/journal/RecoveryManager.cpp
+++ b/qpid/cpp/src/qpid/linearstore/journal/RecoveryManager.cpp
@@ -99,10 +99,22 @@ void RecoveryManager::analyzeJournals(const std::vector<std::string>* preparedTr
     // Analyze file headers of existing journal files
     efpIdentity_t efpIdentity;
     analyzeJournalFileHeaders(efpIdentity);
-    *emptyFilePoolPtrPtr = emptyFilePoolManager->getEmptyFilePool(efpIdentity);
-    efpFileSize_kib_ = (*emptyFilePoolPtrPtr)->fileSize_kib();
 
-    if (!journalEmptyFlag_) {
+    if (journalEmptyFlag_) {
+        *emptyFilePoolPtrPtr = emptyFilePoolManager->getEmptyFilePool(0, 0); // Use default EFP
+    } else {
+        *emptyFilePoolPtrPtr = emptyFilePoolManager->getEmptyFilePool(efpIdentity);
+        if (! *emptyFilePoolPtrPtr) {
+            // TODO: At a later time, this could be used to establish a new pool size provided the partition exists.
+            // If the partition does not exist, this is always an error. For now, throw an exception, as this should
+            // not occur in any practical application. Once multiple partitions and mixed EFPs are supported, this
+            // needs to be resolved. Note that EFP size is always a multiple of QLS_SBLK_SIZE_BYTES (currently 4096
+            // bytes, any other value cannot be used and should be rejected as an error.
+            std::ostringstream oss;
+            oss << "Invalid EFP identity: Partition=" << efpIdentity.pn_ << " Size=" << efpIdentity.ds_ << "k";
+            throw jexception(jerrno::JERR_RCVM_INVALIDEFPID, oss.str(), "RecoveryManager", "analyzeJournals");
+        }
+        efpFileSize_kib_ = (*emptyFilePoolPtrPtr)->fileSize_kib();
 
         // Read all records, establish remaining enqueued records
         if (inFileStream_.is_open()) {
@@ -409,15 +421,14 @@ void RecoveryManager::analyzeJournalFileHeaders(efpIdentity_t& efpIdentity) {
             if (fileHeader._file_number > highestFileNumber_) {
                 highestFileNumber_ = fileHeader._file_number;
             }
+            // TODO: Logic weak here for detecting error conditions in journal, specifically when no
+            // valid files exist, or files from mixed EFPs. Currently last read file header determines
+            // efpIdentity.
+            efpIdentity.pn_ = fileHeader._efp_partition;
+            efpIdentity.ds_ = fileHeader._data_size_kib;
         }
     }
 
-    // TODO: Logic weak here for detecting error conditions in journal, specifically when no
-    // valid files exist, or files from mixed EFPs. Currently last read file header determines
-    // efpIdentity.
-    efpIdentity.pn_ = fileHeader._efp_partition;
-    efpIdentity.ds_ = fileHeader._data_size_kib;
-
     if (fileNumberMap_.empty()) {
         journalEmptyFlag_ = true;
     } else {
diff --git a/qpid/cpp/src/qpid/linearstore/journal/jcntl.cpp b/qpid/cpp/src/qpid/linearstore/journal/jcntl.cpp
index d0bebf3dc5..5da85fef1a 100644
--- a/qpid/cpp/src/qpid/linearstore/journal/jcntl.cpp
+++ b/qpid/cpp/src/qpid/linearstore/journal/jcntl.cpp
@@ -116,6 +116,7 @@ jcntl::recover(EmptyFilePoolManager* efpmp,
     // Verify journal dir and journal files
     _jdir.verify_dir();
     _recoveryManager.analyzeJournals(prep_txn_list_ptr, efpmp, &_emptyFilePoolPtr);
+    assert(_emptyFilePoolPtr != 0);
 
     highest_rid = _recoveryManager.getHighestRecordId();
     _jrnl_log.log(/*LOG_DEBUG*/JournalLog::LOG_INFO, _jid, _recoveryManager.toLog(_jid, 5));
diff --git a/qpid/cpp/src/qpid/linearstore/journal/jerrno.cpp b/qpid/cpp/src/qpid/linearstore/journal/jerrno.cpp
index 88fde914f0..81f96677f4 100644
--- a/qpid/cpp/src/qpid/linearstore/journal/jerrno.cpp
+++ b/qpid/cpp/src/qpid/linearstore/journal/jerrno.cpp
@@ -95,6 +95,7 @@ const uint32_t jerrno::JERR_RCVM_WRITE           = 0x0903;
 const uint32_t jerrno::JERR_RCVM_NULLXID         = 0x0904;
 const uint32_t jerrno::JERR_RCVM_NOTDBLKALIGNED  = 0x0905;
 const uint32_t jerrno::JERR_RCVM_NULLFID         = 0x0907;
+const uint32_t jerrno::JERR_RCVM_INVALIDEFPID    = 0x0908;
 
 // class data_tok
 const uint32_t jerrno::JERR_DTOK_ILLEGALSTATE    = 0x0a00;
@@ -188,6 +189,7 @@ jerrno::__init()
     _err_map[JERR_RCVM_NULLXID] = "JERR_RCVM_NULLXID: Null XID when XID length non-null in header";
     _err_map[JERR_RCVM_NOTDBLKALIGNED] = "JERR_RCVM_NOTDBLKALIGNED: Offset is not data block (dblk)-aligned";
     _err_map[JERR_RCVM_NULLFID] = "JERR_RCVM_NULLFID: Null file id (FID)";
+    _err_map[JERR_RCVM_INVALIDEFPID] = "JERR_RCVM_INVALIDEFPID: Invalid EFP identity (partition/size)";
 
     // class data_tok
     _err_map[JERR_DTOK_ILLEGALSTATE] = "JERR_MTOK_ILLEGALSTATE: Attempted to change to illegal state.";
diff --git a/qpid/cpp/src/qpid/linearstore/journal/jerrno.h b/qpid/cpp/src/qpid/linearstore/journal/jerrno.h
index 27b7013ca7..e5cd7e8227 100644
--- a/qpid/cpp/src/qpid/linearstore/journal/jerrno.h
+++ b/qpid/cpp/src/qpid/linearstore/journal/jerrno.h
@@ -113,6 +113,7 @@ namespace journal {
         static const uint32_t JERR_RCVM_NULLXID;        ///< Null XID when XID length non-null in header
         static const uint32_t JERR_RCVM_NOTDBLKALIGNED; ///< Offset is not data block (dblk)-aligned
         static const uint32_t JERR_RCVM_NULLFID;        ///< Null file ID (FID)
+        static const uint32_t JERR_RCVM_INVALIDEFPID;   ///< Invalid EFP identity (partition/size)
 
         // class data_tok
         static const uint32_t JERR_DTOK_ILLEGALSTATE;   ///< Attempted to change to illegal state