Update to version v2.1.4

bindings/builtinserver/config/config.go

@@ -7,8 +7,9 @@ import (
 )
 
 type StorageConf struct {
-	Path   string `yaml:"path"`
-	Engine string `yaml:"engine"`
+	Path            string `yaml:"path"`
+	Engine          string `yaml:"engine"`
+	StartWithErrors bool   `yaml:"startwitherrors"`
 }
 
 type NetConf struct {
@@ -54,8 +55,9 @@ func (cfg *ServerConfig) GetYamlString() (string, error) {
 func DefaultServerConfig() *ServerConfig {
 	return &ServerConfig{
 		Storage: StorageConf{
-			Path:   "/tmp/reindex",
-			Engine: "leveldb",
+			Path:            "/tmp/reindex",
+			Engine:          "leveldb",
+			StartWithErrors: false,
 		},
 		Net: NetConf{
 			HTTPAddr: "0.0.0.0:9088",

changelog.md

@@ -1,3 +1,9 @@
+# Version 2.1.4 (15.08.2019)
+
+- [fea] Reindexer server will not start if storage is corrupted, `startwitherrors` config flag is used to override
+- [fix] Do not write to WAL empty update queries
+- [fix] Replication config sync
+
 # Version 2.1.3 (14.08.2019)
 
 ## Core
@@ -11,6 +17,7 @@
 - [fix] Memory statistics calculation are improoved
 - [fix] Slave will not try to clear expired by ttl records
 
+
 # Version 2.1.2 (04.08.2019)
 
 ## Core

cpp_src/CMakeLists.txt

@@ -13,7 +13,7 @@ option (WITH_GCOV "Enable instrumented code coverage build" OFF)
 option (ENABLE_LIBUNWIND "Enable libunwind" ON)
 option (ENABLE_TCMALLOC "Enable tcmalloc extensions" ON)
 option (ENABLE_JEMALLOC "Enable jemalloc extensions" ON)
-set (REINDEXER_VERSION_DEFAULT "2.1.3")
+set (REINDEXER_VERSION_DEFAULT "2.1.4")
 
 if(NOT CMAKE_BUILD_TYPE)
   set(CMAKE_BUILD_TYPE "RelWithDebInfo")

cpp_src/cmd/reindexer_server/contrib/config.yml.in

@@ -2,6 +2,7 @@
 storage: 
   path: /var/lib/reindexer
   engine: leveldb
+  startwitherrors: false
 
 # Network configuration
 net:

cpp_src/core/namespace.cc

@@ -663,7 +663,7 @@ void Namespace::Update(const Query &query, QueryResults &result, const RdxContex
 	result.lockResults();
 
 	WrSerializer ser;
-	if (enableStatementRepl) {
+	if (enableStatementRepl && result.Count()) {
 		// FAST PATH: statement based repliaction
 		const_cast<Query &>(query).type_ = QueryUpdate;
 		WALRecord wrec(WalUpdateQuery, query.GetSQL(ser).Slice());
@@ -822,15 +822,15 @@ void Namespace::Delete(const Query &q, QueryResults &result, const RdxContext &c
 		doDelete(r.id);
 	}
 
-	WrSerializer ser;
-
-	const_cast<Query &>(q).type_ = QueryDelete;
-	WALRecord wrec(WalUpdateQuery, q.GetSQL(ser).Slice());
-
-	if (!repl_.slaveMode) lsn = wal_.Add(wrec);
+	if (result.Count()) {
+		WrSerializer ser;
 
-	observers_.OnWALUpdate(lsn, name_, wrec);
+		const_cast<Query &>(q).type_ = QueryDelete;
+		WALRecord wrec(WalUpdateQuery, q.GetSQL(ser).Slice());
+		if (!repl_.slaveMode) lsn = wal_.Add(wrec);
 
+		observers_.OnWALUpdate(lsn, name_, wrec);
+	}
 	if (q.debugLevel >= LogInfo) {
 		logPrintf(LogInfo, "Deleted %d items in %d µs", result.Count(),
 				  duration_cast<microseconds>(high_resolution_clock::now() - tmStart).count());
@@ -1676,11 +1676,11 @@ void Namespace::initWAL(int64_t maxLSN) {
 }
 
 void Namespace::removeExpiredItems(RdxActivityContext *ctx) {
+	const RdxContext rdxCtx{ctx};
+	WLock wlock(mtx_, &rdxCtx);
 	if (repl_.slaveMode) {
 		return;
 	}
-	const RdxContext rdxCtx{ctx};
-	WLock wlock(mtx_, &rdxCtx);
 	for (const std::unique_ptr<Index> &index : indexes_) {
 		if ((index->Type() != IndexTtl) || (index->Size() == 0)) continue;
 		int64_t expirationthreshold =

cpp_src/core/reindexerimpl.cc

@@ -89,7 +89,7 @@ Error ReindexerImpl::EnableStorage(const string& storagePath, bool skipPlacehold
 	if (isHaveConfig) {
 		res = OpenNamespace(kConfigNamespace, StorageOpts().Enabled().CreateIfMissing(), ctx);
 	}
-	replConfigFileChecker_.Init(fs::JoinPath(storagePath_, kReplicationConfFilename));
+	replConfigFileChecker_.SetFilepath(fs::JoinPath(storagePath_, kReplicationConfFilename));
 
 	return res;
 }
@@ -133,7 +133,11 @@ Error ReindexerImpl::Connect(const string& dsn) {
 	}
 
 	bool needStart = replicator_->Configure(configProvider_.GetReplicationConfig());
-	return needStart ? replicator_->Start() : errOK;
+	err = needStart ? replicator_->Start() : errOK;
+	if (!err.ok()) {
+		return err;
+	}
+	return replConfigFileChecker_.Enable();
 }
 
 Error ReindexerImpl::AddNamespace(const NamespaceDef& nsDef, const InternalRdxContext& ctx) {
@@ -940,7 +944,7 @@ void ReindexerImpl::backgroundRoutine() {
 		}
 		bool replConfigWasModified = replConfigFileChecker_.FileWasModified();
 		if (replConfigWasModified) {
-			hasReplConfigLoadError_ = !tryLoadReplicatorConfFromFile();
+			hasReplConfigLoadError_ = !tryLoadReplicatorConfFromFile().ok();
 		} else if (hasReplConfigLoadError_) {
 			// Retry to read error config once
 			// This logic adds delay between write and read, which allows writer to finish all his writes
@@ -1029,6 +1033,7 @@ std::vector<string> defDBConfig = {
 	})json",
 	R"json({
         "type":"replication",
+        "disable_file_update": true,
         "replication":{
 			"role":"none",
 			"master_dsn":"cproto://127.0.0.1:6534/db",

cpp_src/gtests/tests/fixtures/cgoctxpool_api.h

@@ -12,12 +12,16 @@ namespace reindexer {
 std::ostream& operator<<(std::ostream& os, const CancelType& cancel) { return os << static_cast<int>(cancel); }
 }  // namespace reindexer
 
+namespace CGOCtxPoolTests {
+
 class CGOCtxPoolApi : public ::testing::Test {
 public:
 	CGOCtxPoolApi() {}
 	virtual ~CGOCtxPoolApi() {}
 
 protected:
+	enum class MultiThreadTestMode { Simple, Synced };
+
 	void SetUp() {}
 	void TearDown() {}
 
@@ -32,5 +36,7 @@ class CGOCtxPoolApi : public ::testing::Test {
 		return ctx;
 	}
 
-private:
+	void multiThreadTest(size_t threadsCount, MultiThreadTestMode mode);
 };
+
+}  // namespace CGOCtxPoolTests