//
// Copyright (c) 2017-2026, Manticore Software LTD (https://manticoresearch.com)
// All rights reserved
//
// This program is free software; you can redistribute it and/or modify
// it under the terms of the GNU General Public License. You should have
// received a copy of the GPL license along with this program; if you
// did not, you can find it at http://www.gnu.org/
//

#include "sphinxstd.h"
#include "sphinxutils.h"
#include "memio.h"
#include "sphinxpq.h"
#include "searchdreplication.h"
#include "replication/configuration.h"
#include "accumulator.h"
#include "fileutils.h"
#include "coroutine.h"
#include "std/spinlock.h"
#include "digest_sha1.h"
#include "tracer.h"
#include "auth/auth.h"

#include "replication/wsrep_cxx.h"
#include "replication/common.h"
#include "replication/portrange.h"
#include "replication/nodes.h"
#include "replication/commit_monitor.h"
#include "replication/receiver_ctx.h"
#include "replication/serialize.h"
#include "replication/cluster_delete.h"
#include "replication/cluster_get_nodes.h"
#include "replication/cluster_update_nodes.h"
#include "replication/cluster_synced.h"
#include "replication/replicate_index.h"
#include "replication/grastate.h"
#include "replication/cluster_binlog.h"
#include "replication/cluster_sst_progress.h"

#if !_WIN32
// MAC-specific header
#include <netinet/in.h>
#endif

// global application context for wsrep callbacks

// debug options passed into Galera for our logreplication command line key
static const char * const g_sDebugOptions = "debug=on;cert.log_conflicts=yes";
// prefix added for Galera nodes
static const char * const g_sGcommPrefix = "gcomm://";

// verbose logging of replcating transactions, ruled by this env variable
static const bool LOG_LEVEL_RPL_TNX = env_exists ( "MANTICORE_LOG_RPL_TNX" );
#define LOG_COMPONENT_RPL_TNX ""
#define RPL_TNX LOGMSG ( RPL_DEBUG, RPL_TNX, RPL_TNX )

inline static bool IsSyncedOrDonor ( ClusterState_e eState ) noexcept
{
	return ( eState == ClusterState_e::DONOR || eState == ClusterState_e::SYNCED );
}

enum class BOOTSTRAP_E : bool {NO,YES};
enum class PrepareMode_e : bool { STARTUP, RUNTIME };

// cluster related data
struct ReplicationCluster_t final : public Wsrep::Cluster_i, ClusterBinlogData_c
{
public:
	// replicator
	Wsrep::Provider_i* m_pProvider = nullptr;

	// serializer for replicator - guards for only one replication Op a time
	Threads::Coro::Mutex_c m_tReplicationMutex;

	// receiver thread
	Threads::Coro::Waitable_T<bool> m_bWorkerActive { false };

	// Galera port got from global ports list on cluster created
	ScopedPort_c	m_tPort;

	bool m_bUserRequest { false }; // indicates, if cluster is joining by user request (i.e. stmt 'join ...')
	std::atomic<bool>		m_bExiting { false };

	// state variables cached from Galera
	int64_t					m_iConfID = 0;
	Wsrep::ViewStatus_e		m_eStatus = Wsrep::ViewStatus_e::DISCONNECTED;
	int						m_iSize = 0;
	int						m_iIdx = 0;

	// error that got reported to main thread
	CSphMutex				m_tErrorLock;
	StringBuilder_c			m_sError GUARDED_BY ( m_tErrorLock ) { ";" };

	CSphRefcountedPtr<SstProgress_i> m_pSstProgress;

private:
	~ReplicationCluster_t() final; // private since ref-counted

	void AbortSST();
public:
	explicit ReplicationCluster_t ( ClusterDesc_t&& tDesc )
	{
		m_sName = std::move ( tDesc.m_sName );
		m_sPath = std::move ( tDesc.m_sPath );
		m_sUser = std::move ( tDesc.m_sUser );
		m_dClusterNodes = std::move ( tDesc.m_dClusterNodes );
		m_tOptions = std::move ( tDesc.m_tOptions );
		m_iClusterEpoch = tDesc.m_iClusterEpoch;
		m_bSelfBootstrapOnStartup = tDesc.m_bSelfBootstrapOnStartup;
		m_pSstProgress = CreateProgress();
	}

	// state of node
	void SetState ( ClusterState_e eNodeState );
	ClusterState_e GetState() const noexcept;
	const char* szState() const noexcept;
	ClusterState_e WaitReady();

	bool IsHealthy() const;
	bool IsExiting() const noexcept { return m_bExiting.load ( std::memory_order_acquire ); }
	void MarkExiting() noexcept { m_bExiting.store ( true, std::memory_order_release ); }

	bool Init();
	bool Connect ( BOOTSTRAP_E eBootStrap );
	void StartListen();
	void DisconnectAndDeleteProvider();
	void ResetProviderAfterStartupFailure();

	void UpdateGroupView ( const Wsrep::ViewInfo_t* pView ) EXCLUDES ( m_tViewNodesLock );
	StrVec_t FilterViewNodesByProto ( Proto_e eProto = Proto_e::SPHINX, bool bResolve = true ) const EXCLUDES ( m_tViewNodesLock );
	StrVec_t GetViewNodes() const EXCLUDES ( m_tViewNodesLock );
	void SetViewNodes ( StrVec_t&& dNodes );
	uint64_t GetViewVersion() const noexcept { return m_tViewVersion.GetValue(); }
	bool WaitViewChangeUntil ( uint64_t uViewVersion, int64_t tmDeadlineUs ) const
	{
		return m_tViewVersion.WaitVoidUntil ( [this, uViewVersion] () { return m_tViewVersion.GetValue()!=uViewVersion; }, tmDeadlineUs );
	}

	StrVec_t GetIndexes() const noexcept EXCLUDES ( m_tIndexLock );
	const CSphString & GetNodeName () const { return m_sNodeName; }
	CSphString GetUserSnapshot () const EXCLUDES ( m_tUserLock )
	{
		sph::Spinlock_lock tUserLock { m_tUserLock };
		return m_sUser;
	}
	bool SetUserSnapshotIfChanged ( const CSphString & sNewUser, CSphString & sOldUser ) EXCLUDES ( m_tUserLock )
	{
		sph::Spinlock_lock tUserLock { m_tUserLock };
		if ( m_sUser==sNewUser )
			return false;

		sOldUser = m_sUser;
		m_sUser = sNewUser;
		return true;
	}
	void RollbackUserSnapshot ( const CSphString & sExpectedCurrent, const CSphString & sRollbackUser ) EXCLUDES ( m_tUserLock )
	{
		sph::Spinlock_lock tUserLock { m_tUserLock };
		if ( m_sUser==sExpectedCurrent )
			m_sUser = sRollbackUser;
	}

	template<typename ACTION>
	void FilterViewNodes ( ACTION&& Verb ) const
	{
		Threads::SccRL_t tNodesRLock ( m_tViewNodesLock );
		m_dViewNodes.Apply ( std::forward<ACTION> ( Verb ) );
	}

	void ShowStatus ( VectorLike& dOut );

	void HeartBeat ()
	{
		m_tHeardBeat.NotifyAll();
	}

	template<typename PRED>
	void WaitHeartBeat ( PRED&& fnPred )
	{
		m_tHeardBeat.WaitVoid ( std::forward<PRED> ( fnPred ) );
	}

	template<typename PRED>
	bool WaitHeartBeatForMs ( PRED&& fnPred, int64_t iPeriodMs )
	{
		return m_tHeardBeat.WaitVoidForMs ( std::forward<PRED> ( fnPred ), iPeriodMs );
	}

	template<typename PRED>
	ClusterState_e			WaitAny (PRED&& fnPred)
	{
		return m_tNodeState.Wait ( std::forward<PRED> ( fnPred ) );
	}

	template<typename PRED>
	ClusterState_e			WaitAnyForMs (PRED&& fnPred, int64_t iPeriodMs )
	{
		return m_tNodeState.WaitForMs ( std::forward<PRED> ( fnPred ), iPeriodMs );
	}

	/// Cluster_i implementation
	///////////////////////////

	void ChangeView ( const Wsrep::ViewInfo_t* pView, const char* pState, uint64_t iStateLen, void** ppSstReq, uint64_t* pSstReqLen ) final;

	// callback for Galera synced_cb that cluster fully synced and could accept transactions
	void SetSynced() final;

	bool DonateSST ( CSphString sJoiner, const Wsrep::GlobalTid_t* pStateID, bool bBypass ) final;

	void OnRecvStarted() final;

	void OnRecvFinished ( bool bSuccess ) final;

	void OnSeqnoCommited ( const ReplicatedCommand_t & tCmd, int64_t iSeqNo ) final;

	bool IsPrimary() const noexcept { return ( m_eStatus == Wsrep::ViewStatus_e::PRIMARY ); }

	template<typename VISITOR>
	auto WithRlockedIndexes ( VISITOR fnVisitor ) const noexcept EXCLUDES ( m_tIndexLock )
	{
		Threads::SccRL_t tIndexRLock ( m_tIndexLock );
		return fnVisitor ( m_hIndexes );
	}

	template<typename VISITOR>
	auto WithRlockedAllIndexes ( VISITOR fnVisitor ) const noexcept EXCLUDES ( m_tIndexLock )
	{
		Threads::SccRL_t tIndexRLock ( m_tIndexLock );
		return fnVisitor ( m_hIndexes, m_hIndexesLoaded );
	}

	template<typename VISITOR>
	auto WithWlockedIndexes ( VISITOR fnVisitor ) EXCLUDES ( m_tIndexLock )
	{
		Threads::SccWL_t tIndexWLock ( m_tIndexLock );
		return fnVisitor ( m_hIndexes, m_hIndexesLoaded );
	}

	template<typename VISITOR>
	auto WithRlockedOptions ( VISITOR fnVisitor ) const noexcept EXCLUDES ( m_tOptsLock )
	{
		Threads::SccRL_t tOptionsRLock ( m_tOptsLock );
		return fnVisitor ( m_tOptions );
	}

	template<typename VISITOR>
	auto WithWlockedOptions ( VISITOR fnVisitor ) EXCLUDES ( m_tOptsLock )
	{
		Threads::SccWL_t tOptionsWLock ( m_tOptsLock );
		return fnVisitor ( m_tOptions );
	}

private:
	Threads::Coro::Waitable_T<ClusterState_e> m_tNodeState { ClusterState_e::CLOSED };
	Threads::Coro::Waitable_T<bool> m_tHeardBeat { false };

	// nodes at cluster
	// raw nodes addresses (API and replication) from whole cluster
	Threads::Coro::Waitable_T<uint64_t> m_tViewVersion { 0 };
	mutable Threads::Coro::RWLock_c m_tViewNodesLock;
	StrVec_t m_dViewNodes GUARDED_BY ( m_tViewNodesLock );

	mutable Threads::Coro::RWLock_c m_tOptsLock;
	mutable Threads::Coro::RWLock_c m_tIndexLock;
	mutable sph::Spinlock_c m_tUserLock;
	sph::StringSet m_hIndexesLoaded;	// list of index name loaded into daemon but not yet in cluster used for donor to send indexes into joiner

#ifndef NDEBUG
	// it is impossible to attach 'GUARDED_BY to 'using ClusterDesc_t::m_tOptions', so use reference for it debug build
	ClusterOptions_t& m_tOptions GUARDED_BY ( m_tOptsLock ) { ClusterDesc_t::m_tOptions };
	sph::StringSet& m_hIndexes GUARDED_BY ( m_tIndexLock ) { ClusterDesc_t::m_hIndexes }; // to quickly validate query to cluster:index
#else
	using ClusterDesc_t::m_tOptions;
	using ClusterDesc_t::m_hIndexes;
#endif

	CSphString m_sNodeName;
};

using ReplicationClusterRefPtr_c = CSphRefcountedPtr<ReplicationCluster_t>;

// serializer for cluster management operations - only one cluster operation a time
static Threads::Coro::Mutex_c g_tClusterOpsLock;

// cluster list
static Threads::Coro::RWLock_c g_tClustersLock;
static SmallStringHash_T<ReplicationClusterRefPtr_c> g_hClusters GUARDED_BY ( g_tClustersLock );

ReplicationClusterRefPtr_c ClusterByName ( const CSphString& sCluster, const char* szErrTmpl = "unknown cluster '%s'" ) EXCLUDES ( g_tClustersLock )
{
	ReplicationClusterRefPtr_c pCluster;
	Threads::SccRL_t tLock ( g_tClustersLock );
	if ( !g_hClusters.Exists ( sCluster ) )
	{
		if ( szErrTmpl )
			TlsMsg::Err ( szErrTmpl, sCluster.cstr() );
	} else
		pCluster = g_hClusters[sCluster];
	return pCluster;
}

static bool CheckClusterIndexes ( const VecTraits_T<CSphString> & dIndexes, ReplicationClusterRefPtr_c pCluster )
{
	auto fnCmdValidate =  [&]( const auto & hIndexes )
	{
		for ( const CSphString & sIndex : dIndexes )
		{
			if ( !hIndexes[sIndex] )
				return TlsMsg::Err ( "table '%s' doesn't belong to cluster '%s'", sIndex.cstr(), pCluster->m_sName.cstr() );
		}

		return true;
	};

	return pCluster->WithRlockedIndexes ( fnCmdValidate );
}

static bool CheckClusterIndex ( const CSphString & sIndex, ReplicationClusterRefPtr_c pCluster )
{
	return pCluster->WithRlockedIndexes ( [&]( const auto & hIndexes )
	{
		if ( !hIndexes[sIndex] )
			return TlsMsg::Err ( "table '%s' doesn't belong to cluster '%s'", sIndex.cstr(), pCluster->m_sName.cstr() );

		return true;
	});
}

/////////////////////////////////////////////////////////////////////////////
/// forward declarations
/////////////////////////////////////////////////////////////////////////////

/////////////////////////////////////////////////////////////////////////////
/// remote commands for cluster and index managements
/////////////////////////////////////////////////////////////////////////////

// send local indexes to remote nodes via API
static bool SendClusterIndexes ( ReplicationCluster_t * pCluster, const CSphString & sNode, bool bBypass, const Wsrep::GlobalTid_t & tStateID );

static bool ValidateUpdate ( const ReplicationCommand_t & tCmd );

static bool CanReplaceIndexes ( const CSphString & sCluster, const VecTraits_T<CSphString> & dIndexes ) EXCLUDES ( g_tClustersLock );

static bool DoClusterAlterUpdate ( const CSphString & sCluster, const CSphString & sUpdate, NODES_E eUpdate );
static bool IsSameVector ( StrVec_t & dSrc, StrVec_t & dDst );


static bool g_bReplicationEnabled = false;
static CSphString g_sReplicationStartError;
static bool g_bReplicationStarted = false;
static constexpr int64_t g_iClusterExitUpdateNodesWaitMs = 10 * 1000;

bool ReplicationEnabled()
{
	return g_bReplicationEnabled;
}

static constexpr const char * szNodeState ( ClusterState_e eState )
{
	switch ( eState )
	{
		case ClusterState_e::CLOSED: return "closed";
		case ClusterState_e::DESTROYED: return "destroyed";
		case ClusterState_e::JOINING: return "joining";
		case ClusterState_e::DONOR: return "donor";
		case ClusterState_e::SYNCED: return "synced";

		default: return "undefined";
	}
};

const char* ReplicationCluster_t::szState() const noexcept
{
	return ::szNodeState ( GetState() );
}

void ReplicationCluster_t::SetState ( ClusterState_e eNodeState )
{
	m_tNodeState.SetValue ( eNodeState );
	m_tNodeState.NotifyAll();
}
ClusterState_e ReplicationCluster_t::GetState() const noexcept
{
	return m_tNodeState.GetValue();
}

ClusterState_e ReplicationCluster_t::WaitReady()
{
	return WaitAny ( [] ( ClusterState_e i ) { return i == ClusterState_e::SYNCED || i == ClusterState_e::DESTROYED; } );
}

bool ReplicationCluster_t::IsHealthy() const
{
	if ( !IsPrimary() )
		return TlsMsg::Err ( "cluster '%s' is not ready, not primary state (%s)", m_sName.cstr(), szState() );

	auto eState = GetState();
	return eState == ClusterState_e::SYNCED
		|| eState == ClusterState_e::DONOR
		|| TlsMsg::Err ( "cluster '%s' is not ready, current state is %s", m_sName.cstr(), szState() );
}

static int GetClusterMemLimitMB ( int iMemLimit, int iIndexes )
{
	const int CACHE_PER_INDEX = 16;
	const int MIN_CACHE_SIZE = 128;

	// change default cache size to 16Mb per added index or size of largest rt_mem_limit of RT index but at least 128Mb
	int iSize = iMemLimit / 1024 / 1024;
	iIndexes = Max ( 1, iIndexes );
	return Max ( Max ( iIndexes * CACHE_PER_INDEX, iSize ), MIN_CACHE_SIZE );
}

bool ReplicationCluster_t::Init()
{
	assert ( ReplicationEnabled() );
	CSphString sListenAddr, sIncoming, sFullClusterPath;
	m_sNodeName.SetSprintf ( "node_%s_%s_%d", szIncomingIP(), m_sName.cstr(), GetOsThreadId() );
	sListenAddr.SetSprintf ( "%s:%d", szListenReplicationIP(), int ( m_tPort ) );
	sIncoming.SetSprintf ( "%s,%s:%d:replication", szIncomingProto(), szIncomingIP(), int ( m_tPort ) );
	sFullClusterPath = GetDatadirPath ( m_sPath );

	sphLogDebugRpl ( "node incoming '%s', listen '%s', name '%s'", sIncoming.cstr(), sListenAddr.cstr(), m_sNodeName.cstr() );
	StringBuilder_c sOptions ( ";" );
	WithRlockedOptions ( [&sOptions] ( const auto& tOptions ) { sOptions << tOptions.AsStr(); } );

	// all replication options below have not stored into cluster config
	// set incoming address
	if ( HasIncoming() )
	{
		sOptions.Sprintf ( "ist.recv_addr=%s", szIncomingIP() );
		sOptions << "ist.recv_bind=0.0.0.0";
	}

	// change default cache size to 16Mb per added index but not more than default value
	bool bHaveGcache = false;
	WithRlockedOptions ( [&bHaveGcache] ( const auto& tOptions ) { bHaveGcache = tOptions.m_hOptions.Exists ( "gcache.size" ); } );
	if ( !bHaveGcache )
	{
		int iMaxMemLimit = 0;
		int iIndexes = 0;
		WithRlockedIndexes ( [&iMaxMemLimit, &iIndexes] ( const auto& hIndexes ) {
			iIndexes = hIndexes.GetLength();
			for_each ( hIndexes, [&iMaxMemLimit] ( const auto& hIndex ) {
				cServedIndexRefPtr_c pServed = GetServed ( hIndex.first );
				if ( pServed && pServed->m_tSettings.m_iMemLimit > iMaxMemLimit )
					iMaxMemLimit = pServed->m_tSettings.m_iMemLimit;
			} );
		} );

		int iSizeMB = GetClusterMemLimitMB ( iMaxMemLimit, iIndexes );
		sOptions.Sprintf ( "gcache.size=%dM", iSizeMB );
	}

	// set debug log option
	if ( g_eLogLevel >= SPH_LOG_RPL_DEBUG )
		sOptions += g_sDebugOptions;

	m_pProvider = Wsrep::MakeProvider ( this, m_sNodeName, sListenAddr.cstr(), sIncoming.cstr(), sFullClusterPath.cstr(), sOptions.cstr(), m_tGtid );
	return bool ( m_pProvider );
}

bool ReplicationCluster_t::Connect ( BOOTSTRAP_E eBootStrap )
{
	StringBuilder_c sNodes;
	sNodes << g_sGcommPrefix;
	sNodes.StartBlock ( "," );
	FilterViewNodesByProto ( Proto_e::REPLICATION ).Apply ( [&sNodes] ( const CSphString& sNode ) { sNodes << sNode; } );

	sphLogDebugRpl ( "nodes '%s'", sNodes.cstr() );
	if ( g_eLogLevel >= SPH_LOG_RPL_DEBUG ) {
		StringBuilder_c sIndexes ( "," );
		WithRlockedIndexes ( [&sIndexes] ( const auto& hIndexes ) { for_each ( hIndexes, [&] ( const auto& tIndex ) { sIndexes << tIndex.first; } ); } );
		sphLogDebugRpl ( "cluster '%s', gtid %s, indexes '%s', nodes '%s'", m_sName.cstr(), Gtid2Str ( m_tGtid ).cstr(), sIndexes.cstr(), sNodes.cstr() );
	}

	// Connect to cluster
	return m_pProvider->Connect ( m_sName.cstr(), sNodes.cstr(), ( eBootStrap == BOOTSTRAP_E::YES ) );
}

void ReplicationCluster_t::StartListen()
{
	CSphRefcountedPtr<Wsrep::Receiver_i> pReceiver { MakeReceiverCtx ( m_sName, m_pProvider, [this]() { HeartBeat(); } ) };
	m_pProvider->StartListen ( pReceiver.Ptr() );
	sphLogDebugRpl ( "replicator is created for cluster '%s', gtid %s", m_sName.cstr(), Gtid2Str ( m_tGtid ).cstr() );
}

// update cluster state nodes from Galera callback on cluster view changes
void ReplicationCluster_t::UpdateGroupView ( const Wsrep::ViewInfo_t* pView )
{
	const auto* pBoxes = &pView->m_tMemInfo;
	StrVec_t dNodes;
	for ( int i = 0; i < pView->m_iNMembers; ++i )
		dNodes.Append ( ParseNodesFromString ( pBoxes[i].m_sIncoming ) );

	sphLogDebugRpl ( "cluster '%s', gtid %s, view nodes changed: %s > %s", m_sName.cstr(), Gtid2Str ( m_tGtid ).cstr(), Vec2Str ( GetViewNodes() ).cstr(), Vec2Str ( dNodes ).cstr() );
	SetViewNodes ( std::move ( dNodes ) );
}

CSphString WaitClusterReady ( const CSphString& sCluster, int64_t iTimeoutS )
{
	auto pCluster = ClusterByName ( sCluster );
	if ( !pCluster )
		return TlsMsg::MoveToString();

	ClusterState_e eState;
	if ( iTimeoutS<=0 )
		eState = pCluster->WaitAny ( [] ( ClusterState_e i ) { return i == ClusterState_e::SYNCED || i == ClusterState_e::DONOR; } );
	else
		eState = pCluster->WaitAnyForMs ( [] ( ClusterState_e i ) { return i == ClusterState_e::SYNCED || i == ClusterState_e::DONOR; }, iTimeoutS*1000 );
	return szNodeState ( eState );
}

std::pair<int, CSphString> WaitClusterCommit ( const CSphString& sCluster, int iTxn, int64_t iTimeoutS )
{
	auto pCluster = ClusterByName ( sCluster );
	if (!pCluster)
		return {-1,TlsMsg::MoveToString()};

	int64_t iVal = -1;
	auto fnPred = [iTxn, &iVal, pProvider=pCluster->m_pProvider]()
	{
		if ( !pProvider )
			return false;
		assert ( pProvider );
		pProvider->EnumFindStatsVar ( [&iVal] ( const Wsrep::StatsVars_t& tVar ) -> bool {
			if ( 0 != strcmp ( tVar.m_szName, "last_committed" ) )
				return false;
			iVal = tVar.m_tValue.iVal64;
			return true;
		} );
		return iVal >= iTxn;
	};
	bool bSuccess = true;
	if ( iTimeoutS <= 0 )
		pCluster->WaitHeartBeat ( std::move ( fnPred ) );
	else
		bSuccess = pCluster->WaitHeartBeatForMs ( std::move ( fnPred ), iTimeoutS * 1000 );
	if ( !bSuccess )
		return { iVal, "timeout" };
	return { iVal, "" };
}

// callback for Galera that cluster view got changed, ie node either added or removed from cluster.
// It is guaranteed that no other callbacks are called concurrently with it.
void ReplicationCluster_t::ChangeView ( const Wsrep::ViewInfo_t* pView, const char* pState, uint64_t uStateLen, void** ppSstReq, uint64_t* pSstReqLen )
{
	m_tGtid = pView->m_tStateId;
	m_iConfID = pView->m_ViewSeqNo;
	m_iSize = pView->m_iNMembers;
	m_iIdx = pView->m_iIdx;
	m_eStatus = pView->m_eStatus;
	m_bSelfBootstrapOnStartup = ( pView->m_eStatus==Wsrep::ViewStatus_e::PRIMARY && pView->m_iNMembers==1 && pView->m_iIdx==0 );
	if ( IsPrimary() )
		UpdateGroupView ( pView );

	*ppSstReq = nullptr;
	*pSstReqLen = 0;

	if ( !pView->m_bGap )
		return;

	auto sAddr = FromSz ( szIncomingProto() );
	sphLogDebugRpl ( "join %s to '%s', gtid %s", sAddr.first, m_sName.cstr(), Wsrep::Gtid2Str ( m_tGtid ).cstr() );
	*ppSstReq = memcpy ( malloc ( sAddr.second ), sAddr.first, sAddr.second ); // mem will be freed by Galera
	*pSstReqLen = sAddr.second;
	SetState ( ClusterState_e::JOINING );
}

// callback for Galera synced_cb that cluster fully synced and could accept transactions
void ReplicationCluster_t::SetSynced()
{
	SetState ( ClusterState_e::SYNCED );
	if ( IsPrimary() )
	{
		RplBinlog()->OnClusterSynced ( *this );
		if ( m_iClusterEpoch==0 )
			WithWlockedIndexes ( [this] ( auto &, auto & ) { m_iClusterEpoch = 1; } );
	}

	sphLogDebugRpl ( "synced cluster '%s', gtid %s", m_sName.cstr(), Wsrep::Gtid2Str ( m_tGtid ).cstr() );
}

bool ReplicationCluster_t::DonateSST ( CSphString sJoiner, const Wsrep::GlobalTid_t* pStateID, bool bBypass )
{
	Wsrep::GlobalTid_t tGtid = *pStateID;

	sphLogDebugRpl ( "donate %s to %s, state gtid %s, current gtid %s, current indexes '%s', bypass %d", m_sName.cstr(), sJoiner.cstr(), Wsrep::Gtid2Str ( tGtid ).cstr(), Wsrep::Gtid2Str ( m_tGtid ).cstr(), Vec2Str ( GetIndexes() ).cstr(), (int)bBypass );

	SetState ( ClusterState_e::DONOR );
	const bool bOk = SendClusterIndexes ( this, sJoiner, bBypass, tGtid );
	SetState ( ClusterState_e::SYNCED );

	if ( !bOk ) {
		sphWarning ( "%s", TlsMsg::szError() );
		tGtid.m_iSeqNo = Wsrep::WRONG_SEQNO;
	}

	m_pProvider->SstSent ( tGtid, bOk );

	sphLogDebugRpl ( "donate cluster %s to %s, gtid %s, bypass %d, done %d", m_sName.cstr(), sJoiner.cstr(), Wsrep::Gtid2Str ( tGtid ).cstr(), (int)bBypass, (int)bOk );
	return bOk;
}

void ReplicationCluster_t::OnRecvStarted()
{
	m_bWorkerActive.SetValue ( true );
	SetState ( ClusterState_e::JOINING );
}

void ReplicationCluster_t::OnRecvFinished ( bool bSuccess )
{
	SetState ( bSuccess ? ClusterState_e::CLOSED : ClusterState_e::DESTROYED );
	m_bWorkerActive.SetValueAndNotifyAll ( false );
}

void ReplicationCluster_t::AbortSST()
{
	if ( GetState() != ClusterState_e::JOINING )
		return;

	assert ( m_pProvider );
	sphLogDebugRpl ( "aborting SST" );
	Wsrep::GlobalTid_t tGtid {};
	m_pProvider->SstReceived ( tGtid, -ECANCELED );
}

// shutdown and delete cluster, also join cluster recv thread
void ReplicationCluster_t::DisconnectAndDeleteProvider()
{
	if ( !m_pProvider )
		return;

	AbortSST ();
	sphLogDebugRpl ( "disconnecting from cluster %s, gtid %s", m_sName.cstr(), Wsrep::Gtid2Str ( m_tGtid ).cstr() );
	m_pProvider->Disconnect();
	sphLogDebugRpl ( "disconnected from cluster %s, gtid %s", m_sName.cstr(), Wsrep::Gtid2Str ( m_tGtid ).cstr() );
}

void ReplicationCluster_t::ResetProviderAfterStartupFailure()
{
	DisconnectAndDeleteProvider();
	m_bWorkerActive.Wait ( [] ( bool bWorking ) { return !bWorking; } );
	SafeDelete ( m_pProvider );
}

ReplicationCluster_t::~ReplicationCluster_t()
{
	sphLogDebugRpl ( "cluster '%s' wait to finish", m_sName.scstr() );
	// Listening thread are now running and receiving writesets. Wait for them
	// to join. Thread will join after signal handler closes wsrep connection
	m_bWorkerActive.Wait ( [] ( bool bWorking ) { return !bWorking; } );

	HeartBeat();

	sphLogDebugRpl ( "deleting provider of cluster %s", m_sName.cstr() );
	SafeDelete ( m_pProvider );
	sphLogDebugRpl ( "cluster '%s' finished, cluster deleted", m_sName.scstr() );
}

// add 'RPL' flag - i.e., that we're working in replication
struct RPLRep_t: public MiniTaskInfo_t
{
	DECLARE_RENDER ( RPLRep_t );
};

DEFINE_RENDER ( RPLRep_t )
{
	auto& tInfo = *(RPLRep_t*)const_cast<void*> ( pSrc );
	dDst.m_sDescription.Sprintf ( "(RPL %.2T)", tInfo.m_tmStart );
	dDst.m_sChain << "RPL ";
}

struct KeysTraits_t
{
	int m_iOff;
	int m_iCount;
	bool m_bShared;
};

struct ReplicateKeys_t
{
	CSphVector<KeysTraits_t> m_dKeys;
	CSphVector<uint64_t> m_dHashes;

	VecTraits_T<uint64_t> AddKeys ( int iCount, bool bShared )
	{
		if ( !iCount )
			return VecTraits_T<uint64_t>();

		int iOff = m_dHashes.GetLength();

		auto & tItem = m_dKeys.Add();
		tItem.m_bShared = bShared;
		tItem.m_iOff = iOff;
		tItem.m_iCount = iCount;

		m_dHashes.AddN ( iCount );
		return m_dHashes.Slice ( iOff, iCount );
	}

	void AddKey ( uint64_t uKey, bool bShared )
	{
		AddKeys ( 1, bShared )[0] = uKey;
	}
};

static StringBuilder_c & operator << ( StringBuilder_c & tBuilder, const ReplicateKeys_t & tKeys )
{
	tBuilder.StartBlock ( "," );
	for ( const auto & tItem : tKeys.m_dKeys )
	{
		const auto & dKeys = tKeys.m_dHashes.Slice ( tItem.m_iOff, tItem.m_iCount );
		const char * sExclusive = ( tItem.m_bShared ? "" : "(e)" );
		for ( uint64_t uDocid : dKeys )
			tBuilder.Sprintf ( "0x%X%s", uDocid, sExclusive );
	}
	
	tBuilder.FinishBlock();
	return tBuilder;
}

// repl version
// replicate serialized data into cluster and call commit monitor along
static bool Replicate ( const ReplicateKeys_t & tKeys, const VecTraits_T<BYTE>& dQueries, Wsrep::Writeset_i& tWriteSet, CommitMonitor_c&& tMonitor, bool bUpdate, bool bSharedKeys )
{
	TRACE_CONN ( "conn", "Replicate" );

	// just displays 'RPL' flag.
	auto RPL = PublishTaskInfo ( new RPLRep_t );

	RPL_TNX << "replicating, data: " << dQueries.GetLength() << ", keys: " << tKeys.m_dKeys.GetLength() << " [" << tKeys << "]";

	for ( const auto & tItem : tKeys.m_dKeys )
	{
		const auto & dKeys = tKeys.m_dHashes.Slice ( tItem.m_iOff, tItem.m_iCount );
		if ( !tWriteSet.AppendKeys ( dKeys, tItem.m_bShared ) )
			return false;
	}

	if ( !tWriteSet.AppendData ( dQueries ) )
		return false;

	auto dFinalReport = AtScopeExit ( [&tWriteSet] {
		sphLogDebugRpl ( "%s seq " INT64_FMT, ( tWriteSet.LastOk() ? "committed" : "rolled-back" ), tWriteSet.LastSeqno() );
	} );

	TRACE_CONN ( "conn", "pProvider->replicate" );
	if ( !tWriteSet.Replicate() )
	{
		RPL_TNX << "replicating seq " << tWriteSet.LastSeqno() << ", commands " << tKeys.m_dKeys.GetLength();
		return false;
	}

	if ( !tWriteSet.PreCommit() )
		return false;

	bool bOk = false;
	// in case only commit failed
	// need to abort running transaction prior to rollback
	if ( !bUpdate )
		bOk = tMonitor.Commit ();
	else
		bOk = tMonitor.UpdateTOI ();

	if ( !bOk )
	{
		tWriteSet.AbortPreCommit();
		return false;
	}

	tWriteSet.InterimCommit();
	tWriteSet.PostCommit();
	return true;
}

// replicate serialized data into cluster in TotalOrderIsolation mode and call commit monitor along
static bool ReplicateTOI ( const ReplicateKeys_t & tKeys, const VecTraits_T<BYTE> & tQueries, Wsrep::Writeset_i & tWriteSet, CommitMonitor_c && tMonitor )
{
	bool bOk = tWriteSet.ToExecuteStart ( tKeys.m_dHashes, tQueries );
	sphLogDebugRpl ( "replicating TOI %d, seq " INT64_FMT, (int)bOk, tWriteSet.LastSeqno() );
	if ( !bOk )
		return false;

	// FIXME!!! can not fail TOI transaction
	tMonitor.CommitTOI();
	tWriteSet.ToExecuteEnd();

	sphLogDebugRpl ( "%s seq " INT64_FMT, tWriteSet.LastOk() ? "commited" : "rolled-back", tWriteSet.LastSeqno() );
	return tWriteSet.LastOk();
}

// get cluster status variables (our and Galera)
static void ShowClusterCoreStatusVars ( ReplicationCluster_t & tCluster, VectorLike& dOut, const char* sName )
{
	if ( dOut.MatchAddf ( "cluster_%s_nodes_set", sName ) )
		dOut.Add ( Vec2Str ( tCluster.m_dClusterNodes ).cstr() );

	if ( dOut.MatchAddf ( "cluster_%s_nodes_view", sName ) )
	{
		StringBuilder_c sNodes ( "," );
		tCluster.FilterViewNodes ( [&sNodes] ( const CSphString& sNode ) { sNodes << sNode; } );
		dOut.Add ( sNodes.cstr() );
	}

	if ( dOut.MatchAddf ( "cluster_%s_indexes_count", sName ) )
		tCluster.WithRlockedIndexes ( [&dOut] ( const auto& hIndexes ) { dOut.Addf ( "%d", hIndexes.GetLength() ); } );

	if ( dOut.MatchAddf ( "cluster_%s_indexes", sName ) )
	{
		StringBuilder_c tBuf ( "," );
		tCluster.WithRlockedIndexes ( [&tBuf] ( const auto& hIndexes ) { for_each ( hIndexes, [&] ( const auto& tIndex ) { tBuf << tIndex.first; } ); } );
		dOut.Add ( tBuf.cstr() );
	}

	if ( dOut.Matchf ( "cluster_%s_last_error", sName ) )
	{
		ScopedMutex_t tErrorLock ( tCluster.m_tErrorLock );
		if ( !tCluster.m_sError.IsEmpty() )
		{
			dOut.Addf ( "cluster_%s_last_error", sName );
			dOut.Add ( tCluster.m_sError.cstr() );
		}
	}
}

void ReplicationCluster_t::ShowStatus ( VectorLike& dOut )
{
	const char* sName = m_sName.cstr();

	// cluster vars
	if ( dOut.MatchAdd ( "cluster_name" ) )
		dOut.Add ( m_sName );

	if ( !m_pProvider )
	{
		if ( dOut.MatchAddf ( "cluster_%s_status", sName ) )
			dOut.Add ( "recovery-required" );
		if ( dOut.MatchAddf ( "cluster_%s_node_state", sName ) )
			dOut.Add ( szState() );
		ShowClusterCoreStatusVars ( *this, dOut, sName );
		return;
	}

	assert ( m_eStatus >= Wsrep::ViewStatus_e::PRIMARY && m_eStatus < Wsrep::ViewStatus_e::MAX );

	if ( dOut.MatchAddf ( "cluster_%s_state_uuid", sName ) )
		dOut.Add ( Wsrep::Uuid2Str ( m_tGtid.m_tUuid ) );
	if ( dOut.MatchAddf ( "cluster_%s_state_seqno", sName ) )
		dOut.Add().SetSprintf ( INT64_FMT, m_tGtid.m_iSeqNo );
	if ( dOut.MatchAddf ( "cluster_%s_conf_id", sName ) )
		dOut.Add().SetSprintf ( INT64_FMT, m_iConfID );
	if ( dOut.MatchAddf ( "cluster_%s_status", sName ) )
		dOut.Add() = Wsrep::GetViewStatus ( m_eStatus );
	if ( dOut.MatchAddf ( "cluster_%s_size", sName ) )
		dOut.Add().SetSprintf ( "%d", m_iSize );
	if ( dOut.MatchAddf ( "cluster_%s_local_index", sName ) )
		dOut.Add().SetSprintf ( "%d", m_iIdx );
	if ( dOut.MatchAddf ( "cluster_%s_node_state", sName ) )
		dOut.Add ( szState() );
	ShowClusterCoreStatusVars ( *this, dOut, sName );

	// SST progress
	ClusterState_e eState = GetState();
	bool bHasSst = ( eState==ClusterState_e::DONOR || eState==ClusterState_e::JOINING );
	SstProgressStatus_t tProgressStatus;
	if ( bHasSst )
		m_pSstProgress->GetStatus ( tProgressStatus );
	if ( dOut.MatchAddf ( "cluster_%s_sst_total", sName ) )
	{
		if ( bHasSst )
			dOut.Add().SetSprintf ( "%d", (int)( tProgressStatus.m_fTotal * 100 ) );
		else
			dOut.Add ( "-" );
	}
	if ( dOut.MatchAddf ( "cluster_%s_sst_stage", sName ) )
	{
		if ( bHasSst )
			dOut.Add( SstGetStageName ( tProgressStatus.m_eCurrentStage ) );
		else
			dOut.Add ( "-" );
	}
	if ( dOut.MatchAddf ( "cluster_%s_sst_stage_total", sName ) )
	{
		if ( bHasSst )
			dOut.Add().SetSprintf ( "%d", (int)( tProgressStatus.m_fCurrentStageTotal * 100 ) );
		else
			dOut.Add ( "-" );
	}
	if ( dOut.MatchAddf ( "cluster_%s_sst_table", sName ) )
	{
		if ( bHasSst )
			dOut.Add().SetSprintf( "%d (%s)", tProgressStatus.m_iTable, tProgressStatus.m_sTable.cstr() );
		else
			dOut.Add( "-" );
	}
	if ( dOut.MatchAddf ( "cluster_%s_sst_tables", sName ) )
	{
		if ( bHasSst )
			dOut.Add().SetSprintf( "%d", tProgressStatus.m_iTablesCount );
		else
			dOut.Add( "-" );
	}

	// cluster status
	m_pProvider->EnumStatsVars ( [&dOut,&sName] ( const Wsrep::StatsVars_t& tVar ) {
		if ( !dOut.MatchAddf ( "cluster_%s_%s", sName, tVar.m_szName ) )
			return;

		switch ( tVar.m_eType )
		{
		case Wsrep::StatsVars_t::STRING:
			dOut.Add ( tVar.m_tValue.szString );
			return;

		case Wsrep::StatsVars_t::DOUBLE:
			dOut.Addf ( "%f", tVar.m_tValue.fDouble );
			return;

		case Wsrep::StatsVars_t::INT64:
			dOut.Addf ( "%l", tVar.m_tValue.iVal64 );
			return;

		default:
			assert ( 0 && "Internal error" );
		}
	} );
}

// set Galera option for cluster
bool ReplicateSetOption ( const CSphString& sCluster, const CSphString& sName, const CSphString& sVal )
{
	auto pCluster = ClusterByName ( sCluster, "unknown cluster '%s' in SET statement" );
	if ( !pCluster )
		return false;

	if ( !pCluster->m_pProvider )
		return TlsMsg::Err ( "cluster '%s' is not ready, recovery is required", sCluster.cstr() );

	if ( !pCluster->m_pProvider->OptionsSet ( sName, sVal ) )
		return false;

	pCluster->WithWlockedOptions ( [&] ( ClusterOptions_t& hOptions ) { hOptions.m_hOptions.Add ( sVal, sName ); } );
	return true;
}

// delete all clusters on daemon shutdown
void ReplicationServiceShutdown() EXCLUDES ( g_tClustersLock )
{
	Threads::CallCoroutine ( [] {
		Threads::SccWL_t wLock ( g_tClustersLock );
		if ( g_hClusters.IsEmpty() )
			return;

		sphLogDebugRpl ( "clusters (%d) delete invoked", g_hClusters.GetLength() );
		for ( auto & tCluster : g_hClusters )
		{
			sphLogDebugRpl ( "cluster '%s' delete", tCluster.first.cstr() );
			tCluster.second->DisconnectAndDeleteProvider();
		}

		sphLogDebugRpl ( "clusters delete done" );
		g_hClusters.Reset();
	} );
}


// collect clusters and their indexes
CSphVector<ClusterDesc_t> ReplicationCollectClusters ()  EXCLUDES ( g_tClustersLock )
{
	CSphVector<ClusterDesc_t> dClusters;

	if ( !ReplicationEnabled() )
		return dClusters;

	struct ClusterSnapshot_t
	{
		ReplicationClusterRefPtr_c m_pCluster;
		CSphString m_sName;
		CSphString m_sPath;
		CSphString m_sUser;
		StrVec_t m_dClusterNodes;
		bool m_bSelfBootstrapOnStartup = false;
	};

	CSphVector<ClusterSnapshot_t> dSnapshot;
	{
		Threads::SccRL_t tLock ( g_tClustersLock );
		dSnapshot.Reserve ( g_hClusters.GetLength() );
		for ( const auto& tCluster : g_hClusters )
		{
			// should save all clusters on start
			// but skip cluster that just joining from user request
			if ( tCluster.second->GetState() == ClusterState_e::JOINING && tCluster.second->m_bUserRequest )
				continue;

			ClusterSnapshot_t & tSnapshot = dSnapshot.Add();
			tSnapshot.m_pCluster = tCluster.second;
			tSnapshot.m_sName = tCluster.second->m_sName;
			tSnapshot.m_sPath = tCluster.second->m_sPath;
			tSnapshot.m_sUser = tCluster.second->GetUserSnapshot();
			tSnapshot.m_dClusterNodes = tCluster.second->m_dClusterNodes;
			tSnapshot.m_bSelfBootstrapOnStartup = tCluster.second->m_bSelfBootstrapOnStartup && tCluster.second->m_dClusterNodes.GetLength()==1;
		}
	}

	dClusters.Reserve ( dSnapshot.GetLength() );
	for ( const auto& tSnapshot : dSnapshot )
	{
		const auto & pCluster = tSnapshot.m_pCluster;

		// Copy m_hIndexes and m_tOptions under the per-cluster locks that guard them, not via the bare
		// ClusterDesc_t copy ctor. That ctor re-hashes both string hashes while not holding the
		// corresponding per-cluster locks, racing a concurrent ALTER CLUSTER ADD/DROP that mutates them
		// under m_tIndexLock / m_tOptsLock; the race relocates a key mid-copy and crashes in
		// CSphStrHashFunc::Hash on a freed CSphString (observed as a SIGSEGV in
		// ReplicationCollectClusters during chaotic node rejoins).
		ClusterDesc_t & tDesc = dClusters.Add();
		tDesc.m_sName = tSnapshot.m_sName;
		tDesc.m_sPath = tSnapshot.m_sPath;
		tDesc.m_sUser = tSnapshot.m_sUser;
		tDesc.m_dClusterNodes = tSnapshot.m_dClusterNodes;
		tDesc.m_bSelfBootstrapOnStartup = tSnapshot.m_bSelfBootstrapOnStartup;
		pCluster->WithRlockedIndexes ( [&tDesc,pCluster] ( const auto & hIndexes )
		{
			tDesc.m_iClusterEpoch = pCluster->m_iClusterEpoch;
			for ( const auto & tIndex : hIndexes )
				tDesc.m_hIndexes.Add ( tIndex.first );
		} );
		pCluster->WithRlockedOptions ( [&tDesc] ( const auto & tOptions )
		{
			tDesc.m_tOptions = tOptions;
		} );
	}
	return dClusters;
}


// dump all clusters statuses
void ReplicateClustersStatus ( VectorLike & dStatus ) EXCLUDES ( g_tClustersLock )
{
	Threads::SccRL_t tLock ( g_tClustersLock );
	for_each ( g_hClusters, [&dStatus] (const auto& tCluster) { tCluster.second->ShowStatus ( dStatus ); });
}

// check whether index with given name exists and it is mutable (pq or rt) or distributed
static bool CheckIndexExists ( const CSphString & sIndex )
{
	cServedIndexRefPtr_c pServed = GetServed ( sIndex );
	bool bMutable = ServedDesc_t::IsMutable ( pServed );
	cDistributedIndexRefPtr_t pDist ( !bMutable ? GetDistr ( sIndex ) : nullptr );
	if ( !bMutable && !pDist )
		return TlsMsg::Err ( "unknown table '%s'", sIndex.cstr() );

	if ( ServedDesc_t::IsMutable ( pServed ) || pDist )
		return true;
	else
		return TlsMsg::Err ( "wrong type of table '%s'", sIndex.cstr() );
}

static StrVec_t SplitIndexes ( const CSphString & sIndexes )
{
	const char * sIndexesNameDel = ",` ";
	StrVec_t dRes;

	sphSplitApply ( sIndexes.cstr(), sIndexes.Length(), sIndexesNameDel, [&dRes] ( const char * sTok, int iLen )
	{
		if ( !iLen )
			return;
		dRes.Add().SetBinary ( sTok, iLen );
	});

	return dRes;
}

static bool CheckIndexesExists ( const CSphString & sIndex )
{
	StrVec_t dIndexes = SplitIndexes ( sIndex );
	for ( const CSphString & sIndex : dIndexes )
	{
		if ( !CheckIndexExists ( sIndex ) )
			return false;
	}

	return true;
}

// set cluster name into index desc for fast rejects
bool AssignClusterToIndex ( const CSphString & sIndex, const CSphString & sCluster )
{
	TlsMsg::ResetErr();
	cServedIndexRefPtr_c pServed = GetServed ( sIndex );
	bool bMutable = ServedDesc_t::IsMutable ( pServed );
	cDistributedIndexRefPtr_t pDist ( !bMutable ? GetDistr ( sIndex ) : nullptr );
	if ( !bMutable && !pDist )
		return TlsMsg::Err ( "unknown table, or wrong type of table '%s'", sIndex.cstr() );

	if ( bMutable )
	{
		if ( pServed->m_sCluster==sCluster )
			return true;

		ServedIndexRefPtr_c pClone = MakeFullClone ( pServed );
		pClone->m_sCluster = sCluster;
		g_pLocalIndexes->Replace ( pClone, sIndex );
	} else
	{
		if ( pDist->m_sCluster==sCluster )
			return true;

		DistributedIndexRefPtr_t pNewDist ( pDist->Clone() );
		pNewDist->m_sCluster = sCluster;
		g_pDistIndexes->Replace ( pNewDist, sIndex );
	}

	return true;
}

bool AssignClusterToIndexes ( const VecTraits_T<CSphString> & dIndexes, const CSphString & sCluster )
{
	for ( const CSphString & sIndex : dIndexes )
	{
		if ( !AssignClusterToIndex ( sIndex, sCluster ) )
			return false;
	}

	return true;
}

// lock or unlock write operations to disk chunks of index
static bool EnableIndexWrite ( const CSphString & sIndex )
{
	cServedIndexRefPtr_c pServed = GetServed ( sIndex );
	bool bMutable = ServedDesc_t::IsMutable ( pServed );
	cDistributedIndexRefPtr_t pDist ( !bMutable ? GetDistr ( sIndex ) : nullptr );
	if ( !bMutable && !pDist )
		return TlsMsg::Err ( "unknown or wrong-typed table '%s'", sIndex.cstr() );

	if ( pDist )
		return true;

	RIdx_T<RtIndex_i*> ( pServed )->EnableSave();
	return true;
}

// handle replicated command (came from outside)
bool HandleCmdReplicated ( RtAccum_t & tAcc ) NO_THREAD_SAFETY_ANALYSIS
{
	TRACE_SCHED ( "conn", "HandleCmdReplicated" );
	TlsMsg::ResetErr();

	if ( tAcc.m_dCmd.IsEmpty() )
		return TlsMsg::Err ( "empty accumulator" );

	const ReplicationCommand_t & tCmd = *tAcc.m_dCmd[0];
	bool bCmdCluster = ( tAcc.m_dCmd.GetLength()==1
		&& ( tCmd.m_eCommand==ReplCmd_e::CLUSTER_ALTER_ADD || tCmd.m_eCommand==ReplCmd_e::CLUSTER_ALTER_DROP || tCmd.m_eCommand==ReplCmd_e::CLUSTER_ALTER_UPDATE_USER ) );
	if ( bCmdCluster )
	{
		if ( tCmd.m_eCommand==ReplCmd_e::CLUSTER_ALTER_ADD && !CheckIndexesExists ( tCmd.m_sIndex ) )
			return TlsMsg::Err ( "replication error: %s, command %d, cluster %s", TlsMsg::szError(), (int)tCmd.m_eCommand, tCmd.m_sCluster.cstr() );

		CommitMonitor_c tCommit ( tAcc );
		return tCommit.CommitTOI();
	}

	if ( tCmd.m_eCommand==ReplCmd_e::AUTH_ADD || tCmd.m_eCommand==ReplCmd_e::AUTH_DELETE )
	{
		CommitMonitor_c tCommit ( tAcc );
		return tCommit.Commit();
	}

	if ( tAcc.m_dCmd.GetLength()==1 &&
		( tCmd.m_eCommand== ReplCmd_e::UPDATE_API
			 || tCmd.m_eCommand== ReplCmd_e::UPDATE_QL
			 || tCmd.m_eCommand== ReplCmd_e::UPDATE_JSON ) )
	{
		int iUpd = -1;
		CSphString sWarning;
		CommitMonitor_c tCommit ( tAcc, &sWarning, &iUpd );
		bool bOk = tCommit.UpdateTOI();
		if ( !sWarning.IsEmpty() )
			sphWarning ( "%s", sWarning.cstr() );
		return bOk;

		// FIXME!!! make update trx
	}

	cServedIndexRefPtr_c pServed = GetServed ( tCmd.m_sIndex );
	if ( !pServed )
		return TlsMsg::Err ( "unknown table '%s' for replication, command %d, cluster '%s'", tCmd.m_sIndex.cstr(), (int)tCmd.m_eCommand, tCmd.m_sCluster.cstr() );
	if ( !ServedDesc_t::IsMutable ( pServed ) )
		return TlsMsg::Err ( "wrong type of table '%s' for replication, command %d, cluster '%s'", tCmd.m_sIndex.cstr(), (int)tCmd.m_eCommand, tCmd.m_sCluster.cstr() );

	CSphString sError;
	// special path with wlocked index for truncate
	if ( tCmd.m_eCommand == ReplCmd_e::TRUNCATE )
	{
		RPL_TNX << "truncate-commit, table '" << tCmd.m_sIndex.cstr() << "'";
		return WIdx_T<RtIndex_i*> ( pServed )->Truncate ( sError, RtIndex_i::TRUNCATE )
			|| TlsMsg::Err ( "%s", sError.cstr() );
	}

	assert ( tCmd.m_eCommand != ReplCmd_e::TRUNCATE );
	RPL_TNX << "commit, table '" << tCmd.m_sIndex.cstr() << "', uid " << ( tCmd.m_pStored ? tCmd.m_pStored->m_iQUID : int64_t(0) ) << ", queries " << tCmd.m_dDeleteQueries.GetLength() << ", tags " << tCmd.m_sDeleteTags.scstr();

	RIdx_T<RtIndex_i*> pIndex { pServed };
	if ( !tAcc.SetupDocstore ( *pIndex, sError ) )
	{
		sphWarning ( "%s, table '%s', command %d", sError.cstr(), tCmd.m_sIndex.cstr(), (int)tCmd.m_eCommand );
		return false;
	}
	
	// tAcc.m_dCmd and tCmd are invalidated after commit
	tAcc.m_tCmdReplicated = tCmd;

	bool bOk = pIndex->Commit ( nullptr, &tAcc );

	tAcc.m_tCmdReplicated.m_iTID = pIndex->m_iTID;

	return bOk;
}

static void AddSharedKeys ( const uint64_t uIndex, const VecTraits_T<DocID_t> & dDocids, ReplicateKeys_t & tKeys )
{
	assert ( dDocids.GetLength() );

	tKeys.AddKeys ( 1, true )[0] = uIndex;

	const auto & dKeys = tKeys.AddKeys ( dDocids.GetLength(), false );
	ARRAY_FOREACH ( i, dKeys )
	{
		DocID_t iDocid = dDocids[i];
		dKeys[i] = sphFNV64 ( iDocid, uIndex );
	}
}

static uint64_t GetUuidDocidKeyDomain ( uint64_t uIndex )
{
	static constexpr const char * UUID_KEY_NAMESPACE = "rt_uuid_public_id";
	static constexpr int UUID_KEY_NAMESPACE_LEN = sizeof ( "rt_uuid_public_id" ) - 1;
	return sphFNV64 ( UUID_KEY_NAMESPACE, UUID_KEY_NAMESPACE_LEN, uIndex );
}

static void AddUuidDocidKey ( uint64_t uUuidDomain, ByteBlob_t tUuid, ReplicateKeys_t & tKeys )
{
	tKeys.AddKey ( sphFNV64 ( tUuid.first, tUuid.second, uUuidDomain ), false );
}

static void AddUuidDocidKey ( uint64_t uUuidDomain, const CSphString & sUuid, ReplicateKeys_t & tKeys )
{
	tKeys.AddKey ( sphFNV64cont ( sUuid.cstr(), uUuidDomain ), false );
}

static void StoreCommandPqAdd ( const ReplicationCommand_t & tCmd, MemoryWriter_c & tWriter, const uint64_t uIndex, ReplicateKeys_t & tKeys )
{
	assert ( tCmd.m_pStored );
	SaveStoredQuery ( *tCmd.m_pStored, tWriter );

	// store index as shared key
	tKeys.AddKeys ( 1, true )[0] = uIndex;

	// store index and uid as exclusive key
	uint64_t uTableQuid = sphFNV64 ( &tCmd.m_pStored->m_iQUID, sizeof(tCmd.m_pStored->m_iQUID), uIndex );
	tKeys.AddKeys ( 1, false )[0] = uTableQuid;
}

static void StoreCommandPqDelete ( const ReplicationCommand_t & tCmd, MemoryWriter_c & tWriter, const uint64_t uIndex, ReplicateKeys_t & tKeys )
{
	assert ( tCmd.m_dDeleteQueries.GetLength() || !tCmd.m_sDeleteTags.IsEmpty() );
	SaveDeleteQuery ( tCmd.m_dDeleteQueries, tCmd.m_sDeleteTags.cstr(), tWriter );

	// if delete by only docids - table is the shared key and table with QUID is the exclusive key
	if ( tCmd.m_sDeleteTags.IsEmpty() && tCmd.m_dDeleteQueries.GetLength() )
	{
		AddSharedKeys ( uIndex, tCmd.m_dDeleteQueries, tKeys );

	} else
	{
		// otherwise the table is the exclusive key
		tKeys.AddKeys ( 1, false )[0] = uIndex;
	}
}

static void StoreCommandRtTnx ( const ReplicationCommand_t & tCmd, MemoryWriter_c & tWriter, const RtAccum_t & tAcc, const uint64_t uIndex, ReplicateKeys_t & tKeys )
{
	tAcc.SaveRtTrx ( tWriter );

	// table is the shared key and table with ID is the exclusive key
	if ( tAcc.m_dAccumKlist.GetLength() )
		AddSharedKeys ( uIndex, tAcc.m_dAccumKlist, tKeys );
	else
		tKeys.AddKey ( uIndex, true );

	if ( tCmd.m_eCommand!=ReplCmd_e::RT_TRX )
		return;

	assert ( !tAcc.m_uAccumDocs || tAcc.GetIndex() );
	const bool bAccumUuid = tAcc.m_uAccumDocs && sphHasUuidDocid ( tAcc.GetIndex()->GetInternalSchema() );

	if ( !bAccumUuid && tCmd.m_dUuidKeys.IsEmpty() )
		return;

	const uint64_t uUuidDomain = GetUuidDocidKeyDomain ( uIndex );
	if ( bAccumUuid )
		tAcc.ForEachUuidDocid ( [&tKeys, uUuidDomain] ( ByteBlob_t tUuid )
			{
				assert ( tUuid.first );
				assert ( tUuid.second>0 );
				assert ( sphIsNormalizedUuidDocid ( { (const char *)tUuid.first, tUuid.second } ) );
				AddUuidDocidKey ( uUuidDomain, tUuid, tKeys );
			} );

	for ( const CSphString & sUuid : tCmd.m_dUuidKeys )
		AddUuidDocidKey ( uUuidDomain, sUuid, tKeys );
}

static void StoreCommandUpdateApi ( const ReplicationCommand_t & tCmd, MemoryWriter_c & tWriter, const uint64_t uIndex, ReplicateKeys_t & tKeys )
{
	assert ( tCmd.m_pUpdateAPI );
	const CSphAttrUpdate * pUpd = tCmd.m_pUpdateAPI;

	SaveAttrUpdate ( *pUpd, tWriter );

	if ( pUpd->m_dDocids.IsEmpty() )
	{
		// the table is the exclusive key
		tKeys.AddKeys ( 1, false )[0] = uIndex;

	} else
	{
		// the table is the shared key and table with ID is the exclusive key
		AddSharedKeys ( uIndex, pUpd->m_dDocids, tKeys );
	}
}

static void StoreCommandUpdateCond ( const ReplicationCommand_t & tCmd, MemoryWriter_c & tWriter, const uint64_t uIndex, ReplicateKeys_t & tKeys )
{
	assert ( tCmd.m_pUpdateAPI );
	assert ( tCmd.m_pUpdateCond );
	const CSphAttrUpdate * pUpd = tCmd.m_pUpdateAPI;

	SaveAttrUpdate ( *pUpd, tWriter );
	SaveUpdate ( *tCmd.m_pUpdateCond, tWriter );

	const CSphQuery * pQuery = tCmd.m_pUpdateCond;
	const CSphFilterSettings * pFilter = nullptr;

	if ( pQuery && pQuery->m_sQuery.IsEmpty() && pQuery->m_dFilterTree.IsEmpty() && pQuery->m_dFilters.GetLength()==1 )
		pFilter = pQuery->m_dFilters.Begin();

	if ( pFilter && ( ( pFilter->m_bHasEqualMin || pFilter->m_bHasEqualMax ) && pFilter->m_eType==SPH_FILTER_VALUES && ( pFilter->m_sAttrName=="@id" || pFilter->m_sAttrName=="id" ) && !pFilter->m_bExclude ) )
	{
		// the table is the shared key and table with ID is the exclusive key
		const auto & dDocids = pFilter->GetValues();
		AddSharedKeys ( uIndex, dDocids, tKeys );

	} else if ( !pUpd->m_dDocids.IsEmpty() )
	{
		// the table is the shared key and table with ID is the exclusive key
		AddSharedKeys ( uIndex, pUpd->m_dDocids, tKeys );
	} else
	{
		// the table is the exclusive key
		tKeys.AddKeys ( 1, false )[0] = uIndex;
	}
}


// single point there all commands passed these might be replicated to cluster
static bool HandleRealCmdReplicate ( RtAccum_t & tAcc, CommitMonitor_c && tMonitor ) EXCLUDES ( g_tClustersLock )
{
	TRACE_CONN ( "conn", "HandleCmdReplicate" );

	// FIXME!!! for now only PQ add and PQ delete multiple commands supported
	const ReplicationCommand_t & tCmdCluster = *tAcc.m_dCmd[0];

	auto pCluster = ClusterByName ( tCmdCluster.m_sCluster, nullptr );
	if ( !pCluster )
	{
		if ( g_bReplicationStarted )
			return TlsMsg::Err ( "unknown cluster '%s'", tCmdCluster.m_sCluster.cstr() );

		return TlsMsg::Err ( "cluster '%s' can not replicate: %s", tCmdCluster.m_sCluster.cstr(), g_sReplicationStartError.cstr() );
	}

	if ( pCluster->IsExiting() )
		return TlsMsg::Err ( "cluster '%s' is exiting", tCmdCluster.m_sCluster.cstr() );

	if ( !pCluster->IsHealthy() )
		return false;

	if ( tCmdCluster.m_eCommand==ReplCmd_e::TRUNCATE && tCmdCluster.m_tReconfigure )
		return TlsMsg::Err ( "RECONFIGURE is not supported for a cluster table" );

	if ( tCmdCluster.m_bCheckIndex )
	{
		if ( tCmdCluster.m_eCommand==ReplCmd_e::CLUSTER_ALTER_ADD || tCmdCluster.m_eCommand==ReplCmd_e::CLUSTER_ALTER_DROP )
		{
			StrVec_t dIndexes = SplitIndexes ( tCmdCluster.m_sIndex );
			if ( !CheckClusterIndexes ( dIndexes, pCluster ) )
				return false;
		} else if ( !CheckClusterIndex ( tCmdCluster.m_sIndex, pCluster ) )
			return false;
	}

	bool bUpdate = ( tCmdCluster.m_eCommand==ReplCmd_e::UPDATE_API
		|| tCmdCluster.m_eCommand==ReplCmd_e::UPDATE_QL
		|| tCmdCluster.m_eCommand==ReplCmd_e::UPDATE_JSON );
	if ( bUpdate && !ValidateUpdate ( tCmdCluster ) )
		return false;

	assert ( pCluster->m_pProvider );
	bool bTOI = false;

	// replicator check CRC of data - no need to check that at our side
	CSphVector<BYTE> dBufQueries;
	ReplicateKeys_t tKeys;

	BEGIN_CONN ( "conn", "HandleCmdReplicate.serialize", "commands", tAcc.m_dCmd.GetLength() );
	for ( auto& pCmd : tAcc.m_dCmd )
	{
		const ReplicationCommand_t & tCmd = *pCmd;
		uint64_t uIndex = sphFNV64 ( tCmdCluster.m_sIndex.cstr() );

		MemoryWriter_c tWriter ( dBufQueries );
		SaveCmdHeader ( tCmd, tWriter );

		auto iLenPos = dBufQueries.GetLength();
		dBufQueries.AddN ( sizeof(DWORD) );
		int iLenOff = dBufQueries.GetLength();

		switch ( tCmd.m_eCommand )
		{
		case ReplCmd_e::PQUERY_ADD:
			StoreCommandPqAdd ( tCmd, tWriter, uIndex, tKeys );
			break;

		case ReplCmd_e::PQUERY_DELETE:
			StoreCommandPqDelete ( tCmd, tWriter, uIndex, tKeys );
			break;

		case ReplCmd_e::TRUNCATE:
			// FIXME!!! add reconfigure option here
			tKeys.AddKeys ( 1, false )[0] = uIndex;
			break;

		case ReplCmd_e::CLUSTER_ALTER_ADD:
		case ReplCmd_e::CLUSTER_ALTER_DROP:
			bTOI = true;
			tKeys.AddKeys ( 1, false )[0] = uIndex;
			tWriter.PutUint64 ( tCmd.m_iClusterEpoch );
			break;
			
		case ReplCmd_e::CLUSTER_ALTER_UPDATE_USER:
			bTOI = true;
			tKeys.AddKeys ( 1, false )[0] = sphFNV64 ( tCmdCluster.m_sCluster.cstr() );
			break;

		case ReplCmd_e::RT_TRX:
		case ReplCmd_e::AUTH_ADD:
		case ReplCmd_e::AUTH_DELETE:
			StoreCommandRtTnx ( tCmd, tWriter, tAcc, uIndex, tKeys );
			break;

		case ReplCmd_e::UPDATE_API:
			StoreCommandUpdateApi ( tCmd, tWriter, uIndex, tKeys );
			break;

		case ReplCmd_e::UPDATE_QL:
		case ReplCmd_e::UPDATE_JSON:
			StoreCommandUpdateCond ( tCmd, tWriter, uIndex, tKeys );
			break;

		default:
			return TlsMsg::Err ( "unknown command '%d'", (int)tCmd.m_eCommand );
		}

		// store request length
		sphUnalignedWrite ( &dBufQueries[iLenPos], dBufQueries.GetLength() - iLenOff );
	}
	END_CONN ( "conn" );

	auto pWriteSet = pCluster->m_pProvider->MakeWriteSet();

	BEGIN_CONN ( "conn", "HandleCmdReplicate.cluster_lock" );
	Threads::ScopedCoroMutex_t tClusterLock { pCluster->m_tReplicationMutex };
	END_CONN ( "conn" );

	tAcc.CleanReplicated();
	bool bReplicated = false;
	if ( bTOI )
	{
		bReplicated = ReplicateTOI ( tKeys, dBufQueries, *pWriteSet, std::move ( tMonitor ) );
	} else
	{
		bReplicated = Replicate ( tKeys, dBufQueries, *pWriteSet, std::move ( tMonitor ), bUpdate, tAcc.IsReplace () );
	}
	if ( bReplicated )
		pCluster->OnSeqnoCommited ( tAcc.m_tCmdReplicated, pWriteSet->LastSeqno() );

	return bReplicated;
}


static bool PrepareAccForSave ( RtAccum_t & tAcc )
{
	RtIndex_i * pIndex = tAcc.GetIndex();
	if ( !pIndex || !tAcc.IsRtTrxCommand() || !tAcc.m_uAccumDocs )
		return true;

	CSphString sError;
	return ( pIndex->PreCommit ( &tAcc, sError ) || TlsMsg::Err ( "%s", sError.cstr() ) );
}


// single point there all commands passed these might be replicated, even if no cluster
static bool HandleCmdReplicateImpl ( RtAccum_t & tAcc, int * pDeletedCount, CSphString * pWarning, int * pUpdated ) EXCLUDES ( g_tClustersLock )
{
	TRACE_CORO ( "sph", "HandleCmdReplicateImpl" );

	if ( !PrepareAccForSave ( tAcc ) )
		return false;

	CommitMonitor_c tMonitor ( tAcc, pDeletedCount, pWarning, pUpdated );

	// with cluster path
	if ( tAcc.IsClusterCommand () )
		return HandleRealCmdReplicate ( tAcc, std::move ( tMonitor ) );

	if ( tAcc.IsUpdateCommand () )
		return tMonitor.UpdateTOI ();
	return tMonitor.Commit ();
}

bool HandleCmdReplicate ( RtAccum_t & tAcc )
{
	return HandleCmdReplicateImpl ( tAcc, nullptr, nullptr, nullptr );
}

bool HandleCmdReplicateDelete ( RtAccum_t & tAcc, int & iDeletedCount )
{
	return HandleCmdReplicateImpl ( tAcc, &iDeletedCount, nullptr, nullptr );
}

bool HandleCmdReplicateUpdate ( RtAccum_t & tAcc, CSphString & sWarning, int & iUpdated )
{
	return HandleCmdReplicateImpl ( tAcc, nullptr, &sWarning, &iUpdated );
}


bool SetIndexesClusterTOI ( const ReplicationCommand_t * pCmd )
{
	assert ( pCmd );
	const ReplicationCommand_t& tCmd = *pCmd;
	assert ( tCmd.m_eCommand==ReplCmd_e::CLUSTER_ALTER_DROP || tCmd.m_eCommand==ReplCmd_e::CLUSTER_ALTER_ADD );
	auto pCluster = ClusterByName ( tCmd.m_sCluster );
	if ( !pCluster )
		return false;

	StrVec_t dIndexes = SplitIndexes ( tCmd.m_sIndex );

	sphLogDebugRpl ( "SetIndexesClusterTOI '%s' for cluster '%s': indexes '%s' > '%s'", ( tCmd.m_eCommand==ReplCmd_e::CLUSTER_ALTER_ADD ? "add" : "drop" ), pCluster->m_sName.cstr(), tCmd.m_sIndex.cstr(), Vec2Str ( pCluster->GetIndexes() ).cstr() );

	if ( tCmd.m_bCheckIndex && !CheckClusterIndexes ( dIndexes, pCluster ) )
		return false;

	if ( tCmd.m_eCommand==ReplCmd_e::CLUSTER_ALTER_ADD )
	{
		if ( !AssignClusterToIndexes ( dIndexes, tCmd.m_sCluster ) )
			return false;

		pCluster->WithWlockedIndexes ( [&] ( auto & hIndexes, auto & hIndexesLoaded )
		{
			for ( const CSphString & sIndex : dIndexes )
			{
				hIndexes.Add ( sIndex );
				hIndexesLoaded.Delete ( sIndex );
			}
			pCluster->m_iClusterEpoch = tCmd.m_iClusterEpoch;
		});

	} else
	{
		if ( !AssignClusterToIndexes ( dIndexes, "" ) )
			return false;

		pCluster->WithWlockedIndexes ( [&] ( auto & hIndexes, auto & hIndexesLoaded )
		{
			for ( const CSphString & sIndex : dIndexes )
				hIndexes.Delete ( sIndex );
			pCluster->m_iClusterEpoch = tCmd.m_iClusterEpoch;
		});
	}

	TLS_MSG_STRING ( sError );
	bool bSaved = SaveConfigInt ( sError );

	sphLogDebugRpl ( "SetIndexesClusterTOI finished '%s' for cluster '%s': epoch " INT64_FMT ", indexes '%s' > '%s', error: %s", ( tCmd.m_eCommand==ReplCmd_e::CLUSTER_ALTER_ADD ? "add" : "drop" ), pCluster->m_sName.cstr(), tCmd.m_iClusterEpoch, tCmd.m_sIndex.cstr(), Vec2Str ( pCluster->GetIndexes() ).cstr(), sError.scstr() );

	return bSaved;
}

bool SetClusterUserTOI ( const ReplicationCommand_t * pCmd )
{
	assert ( pCmd );
	assert ( pCmd->m_eCommand==ReplCmd_e::CLUSTER_ALTER_UPDATE_USER );

	auto pCluster = ClusterByName ( pCmd->m_sCluster );
	if ( !pCluster )
		return false;

	const CSphString & sNewUser = pCmd->m_sIndex;
	if ( sNewUser.IsEmpty() )
		return TlsMsg::Err ( "invalid cluster user in UPDATE statement" );

	if ( IsAuthEnabled() )
	{
		CSphString sPermError;
		if ( !CheckPerms ( sNewUser, AuthAction_e::REPLICATION, pCmd->m_sCluster, true, sPermError ) )
			return TlsMsg::Err ( "%s", sPermError.cstr() );
	}

	CSphString sOldUser;
	if ( !pCluster->SetUserSnapshotIfChanged ( sNewUser, sOldUser ) )
		return true;

	TLS_MSG_STRING ( sError );
	bool bSaved = SaveConfigInt ( sError );
	if ( !bSaved )
		pCluster->RollbackUserSnapshot ( sNewUser, sOldUser );

	return ( bSaved || TlsMsg::Err ( "%s", sError.cstr() ) );
}

static bool ValidateUpdate ( const ReplicationCommand_t & tCmd )
{
	cServedIndexRefPtr_c pServed = GetServed ( tCmd.m_sIndex );
	if ( !pServed )
		return TlsMsg::Err ( "requires an existing table, %s", tCmd.m_sIndex.cstr() );

	const ISphSchema& tSchema = RIdx_c ( pServed )->GetMatchSchema();

	assert ( tCmd.m_pUpdateAPI );
	CSphString sError;
	CSphString sWarning;
	if ( !Update_CheckAttributes ( *tCmd.m_pUpdateAPI, tSchema, sError, sWarning ) )
		return TlsMsg::Err ( sError );

	assert ( sWarning.IsEmpty() ); // FIXME!!! should be validated on donor
	return true;
}

// load indexes received from another node or existed already into daemon
static bool ReplicatedIndexes ( const VecTraits_T<CSphString> & dIndexes, const CSphString & sCluster, int64_t iClusterEpoch ) EXCLUDES ( g_tClustersLock )
{
	assert ( ReplicationEnabled() );

	if ( !dIndexes.all_of ( [] ( const CSphString & sIndex ) { return CheckIndexExists ( sIndex ); } ) )
		return false;

	auto pCluster = ClusterByName( sCluster );
	if ( !pCluster )
		return false;

	if ( iClusterEpoch==0 )
		iClusterEpoch = 1;
	sphLogDebugRpl ( "ReplicatedIndexes '%s': current epoch " INT64_FMT ", current indexes '%s' -> received epoch " INT64_FMT ", indexes '%s'", sCluster.cstr(), pCluster->m_iClusterEpoch, Vec2Str ( pCluster->GetIndexes() ).cstr(), iClusterEpoch, Vec2Str ( dIndexes ).cstr() );

	if ( !CanReplaceIndexes ( sCluster, dIndexes ) )
		return false;

	sph::StringSet hIndexes ( dIndexes );

	StrVec_t hCurrentIndexes ( pCluster->GetIndexes() );
	StrVec_t dStaleIndexes;
	for ( const auto & sIndex : hCurrentIndexes )
		if ( !hIndexes[sIndex] )
			dStaleIndexes.Add ( sIndex );

	bool bOk = AssignClusterToIndexes ( dStaleIndexes, "" );
	bOk &= AssignClusterToIndexes ( dIndexes, sCluster );

	// need to enable back local index write
	for ( const CSphString & sIndex : dIndexes )
		bOk &= EnableIndexWrite ( sIndex );

	if ( !bOk )
		return false;

	pCluster->WithWlockedIndexes([&] ( auto & hIndexes, auto & hIndexesLoaded )
	{
		hIndexes.Reset();
		hIndexesLoaded.Reset();
		dIndexes.for_each ( [&hIndexes] ( const auto & sIndex )
		{
			hIndexes.Add ( sIndex );
		});

		pCluster->m_iClusterEpoch = iClusterEpoch;
	});

	TLS_MSG_STRING ( sError );
	return SaveConfigInt ( sError );
}

bool CanReplaceIndex ( const CSphString & sCluster, const CSphString & sIndex ) EXCLUDES ( g_tClustersLock )
{
	{
		cServedIndexRefPtr_c pServed = GetServed ( sIndex );
		if ( ServedDesc_t::IsMutable ( pServed ) && !pServed->m_sCluster.IsEmpty() && pServed->m_sCluster!=sCluster )
			return TlsMsg::Err ( "table '%s' is already a part of cluster '%s'", sIndex.cstr(), pServed->m_sCluster.cstr() );
	}

	{
		cDistributedIndexRefPtr_t pDist = GetDistr ( sIndex );
		if ( pDist && !pDist->m_sCluster.IsEmpty() && pDist->m_sCluster!=sCluster )
			return TlsMsg::Err ( "table '%s' is already a part of cluster '%s'", sIndex.cstr(), pDist->m_sCluster.cstr() );
	}

	StrVec_t dIndexes;
	dIndexes.Add ( sIndex );
	return CanReplaceIndexes ( sCluster, dIndexes );
}

bool CanReplaceIndexes ( const CSphString & sCluster, const VecTraits_T<CSphString> & dIndexes ) EXCLUDES ( g_tClustersLock )
{
	CSphVector<ReplicationClusterRefPtr_c> dOtherClusters;
	{
		Threads::SccRL_t rLock ( g_tClustersLock );
		if ( !g_hClusters.Exists ( sCluster ) )
			return TlsMsg::Err ( "unknown cluster '%s'", sCluster.cstr() );

		dOtherClusters.Reserve ( g_hClusters.GetLength() );
		for ( const auto & tCluster : g_hClusters )
		{
			if ( tCluster.first==sCluster )
				continue;

			dOtherClusters.Add ( tCluster.second );
		}
	}

	for ( const auto & pCluster : dOtherClusters )
	{
		const CSphString & sClusterName = pCluster->m_sName;
		bool bCanReplace = pCluster->WithRlockedAllIndexes ( [&dIndexes,&sClusterName] ( const auto & hOrigIndexes, const auto & hOrigIndexesLoaded )
		{
			for ( const auto & sIndex : dIndexes )
			{
				if ( hOrigIndexes[sIndex] || hOrigIndexesLoaded[sIndex] )
					return TlsMsg::Err ( "table '%s' is already a part of cluster '%s'", sIndex.cstr(), sClusterName.cstr() );
			}

			return true;
		});
		if ( !bCanReplace )
			return false;
	}

	return true;
}

// validate clusters paths
static bool ClusterCheckPath ( const CSphString& sPath, const char* szCluster, bool bNeedWrite = false ) EXCLUDES ( g_tClustersLock )
{
	if ( !ReplicationEnabled() )
		return TlsMsg::Err ( "data_dir option is missing in config or no replication listener is set, replication is disabled" );

	auto sFullPath = GetDatadirPath ( sPath );
	CSphString sError;
	if ( !CheckPath ( sFullPath, bNeedWrite, sError ) )
		return TlsMsg::Err ( "cluster '%s', %s", szCluster, sError.cstr() );

	Threads::SccRL_t tClusterRLock ( g_tClustersLock );
	for ( const auto& tCluster : g_hClusters )
	{
		if ( sPath == tCluster.second->m_sPath )
			return TlsMsg::Err ( "duplicate paths, cluster '%s' has the same path as '%s'", szCluster, tCluster.second->m_sName.cstr() );
	}
	return true;
}

static ReplicationCluster_t* MakeClusterOffline ( ClusterDesc_t tDesc )
{
	auto tPort = PortRange::AcquirePort();
	if ( !tPort.IsValid() )
	{
		TlsMsg::Err ( "cluster '%s', no replication ports available, add replication listener", tDesc.m_sName.cstr() );
		return nullptr;
	}

	CSphRefcountedPtr<ReplicationCluster_t> pCluster ( new ReplicationCluster_t ( std::move ( tDesc ) ) );
	pCluster->m_tPort = std::move ( tPort );

	return pCluster.Leak();
}

static void ForceSst ( ReplicationCluster_t & tCluster );
static bool CheckForceSst ( const ReplicationCluster_t & tCluster, int64_t iRemoteEpoch );

static bool PrepareNodesAndInitCluster ( ReplicationCluster_t & tCluster, BOOTSTRAP_E eBootStrap, PrepareMode_e ePrepareMode, bool bKeepClusterNodes = false )
{
	sphLogDebugRpl ( "PrepareNodesAndInitCluster '%s', bootstrap %d, mode %d, nodes: %d", tCluster.m_sName.cstr(), (int)eBootStrap, (int)ePrepareMode, tCluster.m_dClusterNodes.GetLength() );

	if ( ePrepareMode==PrepareMode_e::STARTUP )
		RplBinlog()->OnClusterLoad ( tCluster );

	if ( tCluster.m_dClusterNodes.IsEmpty() && eBootStrap!=BOOTSTRAP_E::YES )
	{
		sphWarning ( "no nodes found, created new cluster '%s'", tCluster.m_sName.cstr() );
		eBootStrap = BOOTSTRAP_E::YES;
	}

	StrVec_t dNodes;
	if ( eBootStrap == BOOTSTRAP_E::NO )
	{
		if ( ePrepareMode==PrepareMode_e::STARTUP )
		{
			CSphString sRemoteUser;
			int64_t iRemoteEpoch = 0;
			if ( !CheckRemotesVersions ( tCluster, sRemoteUser, &iRemoteEpoch ) )
				return false;
			tCluster.m_sUser = sRemoteUser;

			bool bForceSst = CheckForceSst ( tCluster, iRemoteEpoch );
			sphLogDebugRpl ( "startup SST guard for '%s': local epoch " INT64_FMT ", remote epoch " INT64_FMT ", force sst %d", tCluster.m_sName.cstr(), tCluster.m_iClusterEpoch, iRemoteEpoch, (int)bForceSst );

			if ( bForceSst )
				ForceSst ( tCluster );
		}

		dNodes = GetNodeListFromRemotes ( tCluster );
		if ( dNodes.IsEmpty() )
			return false;
	}

	if ( !tCluster.Init () )
		return false;

	tCluster.SetViewNodes ( std::move ( dNodes ) );
	if ( !bKeepClusterNodes )
		tCluster.m_dClusterNodes = tCluster.FilterViewNodesByProto ();
	return true;
}

static ReplicationCluster_t* MakeCluster ( ClusterDesc_t tDesc, BOOTSTRAP_E eBootStrap, PrepareMode_e ePrepareMode )
{
	CSphRefcountedPtr<ReplicationCluster_t> pCluster ( MakeClusterOffline ( std::move ( tDesc ) ) );
	if ( !pCluster )
		return nullptr;

	if ( !PrepareNodesAndInitCluster ( *pCluster, eBootStrap, ePrepareMode ) )
		return nullptr;
	return pCluster.Leak();
}

static bool AddAndStartCluster ( ReplicationClusterRefPtr_c pCluster )
{
	{
		Threads::SccWL_t tLock ( g_tClustersLock );
		if ( !g_hClusters.Add ( pCluster, pCluster->m_sName ) )
			return false;
	}

	pCluster->StartListen();
	return true;
}

static bool ClusterDescOk ( const ClusterDesc_t& tDesc, bool bForce ) noexcept
{
	if ( tDesc.m_dClusterNodes.IsEmpty() )
		sphWarning ( "no nodes found, create new cluster '%s'", tDesc.m_sName.cstr() );

	if ( IsAuthEnabled() )
	{
		if ( tDesc.m_sUser.IsEmpty() )
		{
			sphWarning ( "Cluster %s: missing effective replication user, skipped", tDesc.m_sName.cstr() );
			return false;
		}

		CSphString sPermError;
		if ( !CheckPerms ( tDesc.m_sUser, AuthAction_e::REPLICATION, tDesc.m_sName, true, sPermError ) )
		{
			sphWarning ( "Cluster %s: invalid effective replication user '%s', %s, skipped", tDesc.m_sName.cstr(), tDesc.m_sUser.cstr(), sPermError.cstr() );
			return false;
		}
	}

	// check cluster path is unique
	if ( !ClusterCheckPath ( tDesc.m_sPath, tDesc.m_sName.cstr() ) )
	{
		sphWarning ( "Cluster %s: %s, skipped", tDesc.m_sName.cstr(), TlsMsg::szError() );
		return false;
	}

	if ( !bForce )
		return true;

	auto sDataDirPath = GetDatadirPath ( tDesc.m_sPath );
	if ( !NewClusterForce ( sDataDirPath ) || !CheckClusterNew ( sDataDirPath ) )
	{
		sphWarning ( "Cluster %s: %s, skipped", tDesc.m_sName.cstr(), TlsMsg::szError() );
		return false;
	}

	return true;
}

static void PreserveFailedCluster ( std::pair<CSphString, CSphRefcountedPtr<ReplicationCluster_t>>& tCluster, int & iFailedClusters )
{
	CSphString sError = TlsMsg::szError();
	if ( sError.IsEmpty() )
		sError.SetSprintf ( "cluster '%s' failed to start", tCluster.first.cstr() );

	sphWarning ( "%s", sError.cstr() );
	auto & pCluster = tCluster.second;
	pCluster->ResetProviderAfterStartupFailure();
	pCluster->SetState ( ClusterState_e::CLOSED );
	{
		ScopedMutex_t tErrorLock ( pCluster->m_tErrorLock );
		if ( pCluster->m_sError.GetLength()>1024 )
			pCluster->m_sError.Rewind();
		pCluster->m_sError += sError.cstr();
	}
	++iFailedClusters;
}

struct ClusterStartupMode_t
{
	BOOTSTRAP_E m_eBootStrap = BOOTSTRAP_E::NO;
	bool m_bKeepClusterNodes = false;
};

static bool PrepareClusterStartupMode ( ReplicationCluster_t & tCluster, RplBootstrap_e eBs, ClusterStartupMode_t & tMode )
{
	tMode.m_eBootStrap = ( eBs==RplBootstrap_e::OFF ? BOOTSTRAP_E::NO : BOOTSTRAP_E::YES );
	tMode.m_bKeepClusterNodes = false;

	if ( eBs==RplBootstrap_e::OFF && tCluster.m_bSelfBootstrapOnStartup && tCluster.m_dClusterNodes.GetLength()==1 )
	{
		auto sDataDirPath = GetDatadirPath ( tCluster.m_sPath );
		if ( !CheckClusterNew ( sDataDirPath ) )
		{
			CSphString sSafeBootstrapError = TlsMsg::szError();
			TlsMsg::Err ( "self-bootstrap startup rejected: %s", sSafeBootstrapError.cstr() );
			return false;
		}

		tMode.m_eBootStrap = BOOTSTRAP_E::YES;
		tMode.m_bKeepClusterNodes = true;
		sphLogDebugRpl ( "cluster '%s' has self-bootstrap startup flag set, bootstrapping local cluster on normal startup", tCluster.m_sName.cstr() );
		return true;
	}

	CSphString sRemoteUser;
	if ( !CheckRemotesVersions ( tCluster, sRemoteUser ) )
	{
		CSphString sRemoteError = TlsMsg::szError();
		TlsMsg::Err ( "remote version check failed: %s", sRemoteError.cstr() );
		return false;
	}

	tCluster.m_sUser = sRemoteUser;

	// need to fetch auth from all other nodes only for non leader
	if ( eBs==RplBootstrap_e::OFF && !GetRemotesAuth ( tCluster ) )
	{
		CSphString sAuthError = TlsMsg::szError();
		TlsMsg::Err ( "auth failed: %s", sAuthError.cstr() );
		return false;
	}

	return true;
}

// called from ReplicationServiceStart - see below
static void CoReplicationServiceStart ( RplBootstrap_e eBs ) EXCLUDES ( g_tClustersLock )
{
	assert ( Threads::IsInsideCoroutine() );
	int iStartedClusters = 0;
	int iFailedClusters = 0;

	Threads::SccWL_t tLock ( g_tClustersLock );

	for ( auto & tCluster : g_hClusters )
	{
		TlsMsg::ResetErr();
		auto& sName = tCluster.first;
		auto& pCluster = tCluster.second;
		ClusterStartupMode_t tMode;

		if ( !PrepareClusterStartupMode ( *pCluster, eBs, tMode ) )
		{
			PreserveFailedCluster ( tCluster, iFailedClusters );
			continue;
		}

		if ( !PrepareNodesAndInitCluster ( *pCluster, tMode.m_eBootStrap, PrepareMode_e::STARTUP, tMode.m_bKeepClusterNodes ) )
		{
			PreserveFailedCluster ( tCluster, iFailedClusters );
			continue;
		}

		// check indexes valid
		if ( !pCluster->Connect ( tMode.m_eBootStrap ) )
		{
			PreserveFailedCluster ( tCluster, iFailedClusters );
			continue;
		}

		pCluster->StartListen();
		++iStartedClusters;

		sphLogDebugRpl ( "'%s' cluster, gtid %s starting", sName.cstr(), Wsrep::Gtid2Str ( pCluster->m_tGtid ).cstr() );
	}
	if ( !g_hClusters.IsEmpty() && iStartedClusters==0 && iFailedClusters )
		sphWarning ( "no clusters to start" );

	g_bReplicationStarted = true;
}

// start clusters on daemon start
void ReplicationServiceStart ( RplBootstrap_e eBs ) EXCLUDES ( g_tClustersLock )
{
	// should be lined up with PrepareClustersOnStartup
	if ( !ReplicationEnabled() )
	{
		Threads::SccRL_t tLock ( g_tClustersLock );
		if ( !g_hClusters.IsEmpty() )
			sphWarning ( "loading %d cluster(s) but the replication disabled, %s", g_hClusters.GetLength(), g_sReplicationStartError.cstr() );
		return;
	}

	Threads::CallCoroutine ( [=]() EXCLUDES ( g_tClustersLock ) { CoReplicationServiceStart ( eBs ); } );
}

// called from ReplicationServiceStart - see below
static void CoPrepareClustersOnStartup ( RplBootstrap_e eBs ) EXCLUDES ( g_tClustersLock )
{
	SmallStringHash_T<ReplicationClusterRefPtr_c> hClusters;
	assert ( Threads::IsInsideCoroutine() );
	for ( const ClusterDesc_t& tDesc : GetClustersInt() )
	{
		if ( !ClusterDescOk ( tDesc, eBs==RplBootstrap_e::FORCED ) )
			continue;

		CSphRefcountedPtr<ReplicationCluster_t> pNewCluster { MakeClusterOffline ( tDesc ) };
		if ( !pNewCluster ) {
			sphWarning ( "%s", TlsMsg::szError() );
			continue;
		}

		// check indexes valid
		for ( const auto & tIndex : tDesc.m_hIndexes )
		{
			const CSphString & sIndex = tIndex.first;
			if ( !AssignClusterToIndex ( sIndex, pNewCluster->m_sName ) )
			{
				sphWarning ( "%s, removed from cluster '%s'", TlsMsg::szError(), pNewCluster->m_sName.cstr() );
				continue;
			}
			pNewCluster->WithWlockedIndexes ( [&sIndex] ( auto & hIndexes, auto & hIndexesLoaded )
			{
				hIndexes.Add ( sIndex );
				hIndexesLoaded.Delete ( sIndex );
			});
		}

		if ( !hClusters.Add ( pNewCluster, pNewCluster->m_sName ) )
		{
			for ( const auto & tIndex : tDesc.m_hIndexes )
			{
				if ( !AssignClusterToIndex ( tIndex.first, "" ) )
					sphWarning ( "%s on removal table '%s' from a cluster", TlsMsg::szError(), tIndex.first.cstr() );
			}
			continue;
		}
	}

	// copy prepared clusters
	Threads::SccWL_t tLock ( g_tClustersLock );
	assert ( g_hClusters.IsEmpty() );
	for_each ( hClusters, [&] ( auto& hCluster ) REQUIRES ( g_tClustersLock ) { g_hClusters.Add ( hCluster.second, hCluster.first ); } );
}

void PrepareClustersOnStartup ( const VecTraits_T<ListenerDesc_t>& dListeners, RplBootstrap_e eBs ) EXCLUDES ( g_tClustersLock )
{
	if ( !SetReplicationListener ( dListeners, g_sReplicationStartError ) )
	{
		if ( !GetClustersInt().IsEmpty() )
			sphWarning ( "%s", g_sReplicationStartError.cstr() );
		else
			sphLogDebugRpl ( "%s", g_sReplicationStartError.cstr() );
		return;
	}
	g_bReplicationEnabled = true;

	Threads::CallCoroutine ( [=]() EXCLUDES ( g_tClustersLock ) { CoPrepareClustersOnStartup ( eBs ); } );
}

// validate cluster option at SphinxQL statement
static std::optional <CSphString> CheckClusterOption ( const SmallStringHash_T<SqlInsert_t *> & hValues, const char * szName )
{
	SqlInsert_t ** ppVal = hValues ( szName );
	if (!ppVal)
		return "";

	if (( *ppVal )->m_sVal.IsEmpty())
	{
		TlsMsg::Err ( "'%s' should have a string value", szName );
		return std::nullopt;
	}

	return ( *ppVal )->m_sVal;
}

// validate cluster SphinxQL statement
static bool CheckClusterExists ( const CSphString & sCluster ) EXCLUDES ( g_tClustersLock )
{
	Threads::SccRL_t rLock ( g_tClustersLock );
	return g_hClusters.Exists ( sCluster );
}

// validate cluster SphinxQL statement
enum class MAKE_E : bool { CREATE, JOIN };
static std::optional<ClusterDesc_t> ClusterDescFromSphinxqlStatement ( const CSphString & sCluster, const StrVec_t & dNames, const CSphVector<SqlInsert_t> & dValues, MAKE_E eJoin ) EXCLUDES ( g_tClustersLock )
{
	std::optional<ClusterDesc_t> tDesc;
	if ( !ReplicationEnabled() )
	{
		TlsMsg::Err ( "data_dir option is missing or no replication provider configured" );
		return tDesc;
	}

	if ( CheckClusterExists ( sCluster ) )
	{
		TlsMsg::Err ( "cluster '%s' already exists", sCluster.cstr() );
		return tDesc;
	}

	SmallStringHash_T<SqlInsert_t*> hValues;
	assert ( dNames.GetLength() == dValues.GetLength() );
	ARRAY_FOREACH ( i, dNames )
		hValues.Add ( &dValues[i], dNames[i] );

	// optional items
	auto tNodes = CheckClusterOption ( hValues, "at_node" );
	if ( !tNodes )
		return tDesc;

	auto dClusterNodes = ParseNodesFromString ( tNodes.value() );
	if ( dClusterNodes.IsEmpty() )
	{
		tNodes = CheckClusterOption ( hValues, "nodes" );
		if ( !tNodes )
			return tDesc;
		dClusterNodes = ParseNodesFromString ( tNodes.value() );
	}

	if ( eJoin==MAKE_E::JOIN && dClusterNodes.IsEmpty() )
	{
		TlsMsg::Err ( "cannot join without either nodes list or AT node" );
		return tDesc;
	}

	auto tPath = CheckClusterOption ( hValues, "path" );
	if ( !tPath )
		return tDesc;

	auto tOptions = CheckClusterOption ( hValues, "options" );
	if ( !tOptions )
		return tDesc;

	// check cluster path is unique
	if ( !ClusterCheckPath ( tPath.value(), sCluster.cstr(), true ) )
		return tDesc;

	auto tUser = CheckClusterOption ( hValues, "user" );
	if ( !tUser )
		return tDesc;

	// all is ok, create cluster desc
	tDesc.emplace();
	tDesc->m_sName = sCluster;
	tDesc->m_sPath = tPath.value();
	tDesc->m_dClusterNodes = dClusterNodes;
	tDesc->m_tOptions.Parse ( tOptions.value() );
	tDesc->m_sUser = ( tUser.value().IsEmpty() ? session::GetUser() : tUser.value() );
	return tDesc;
}

static CSphString GetUserOption ( const SqlStmt_t & tStmt )
{
	CSphString sUser;

	ARRAY_FOREACH ( i, tStmt.m_dCallOptNames )
	{
		if ( !StrEqN ( FromStr ( tStmt.m_dCallOptNames[i] ), "user" ) )
			continue;

		sUser = tStmt.m_dCallOptValues[i].m_sVal;
		break;
	}

	return sUser;
}

static bool GetClusterEffectiveUser ( const CSphString & sCluster, CSphString & sUser, CSphString & sError )
{
	auto pCluster = ClusterByName ( sCluster, nullptr );
	if ( !pCluster )
	{
		sError.SetSprintf ( "unknown cluster '%s'", sCluster.cstr() );
		return false;
	}

	sUser = pCluster->m_sUser;
	if ( sUser.IsEmpty() )
	{
		sError.SetSprintf ( "cluster '%s' has no effective replication user", sCluster.cstr() );
		return false;
	}

	return true;
}

bool ReplicationResolveUser ( const SqlStmt_t & tStmt, const CSphString & sSessionUser, CSphString & sRplUser, CSphString & sError )
{
	sRplUser = "";

	switch ( tStmt.m_eStmt )
	{
	case STMT_CLUSTER_CREATE:
	case STMT_JOIN_CLUSTER:
	{
		CSphString sStmtUser = GetUserOption ( tStmt );
		if ( !sStmtUser.IsEmpty() )
			sRplUser = sStmtUser;
		else
			sRplUser = sSessionUser;
		return true;
	}

	case STMT_CLUSTER_DELETE:
	case STMT_CLUSTER_ALTER_ADD:
	case STMT_CLUSTER_ALTER_DROP:
	case STMT_CLUSTER_ALTER_UPDATE:
	{
		if ( tStmt.m_eStmt==STMT_CLUSTER_ALTER_UPDATE && tStmt.m_sSetName=="user" )
		{
			if ( tStmt.m_sStringParam.IsEmpty() )
			{
				sError.SetSprintf ( "invalid cluster user in UPDATE statement" );
				return false;
			}
			if ( !CheckClusterExists ( tStmt.m_sCluster ) )
			{
				sError.SetSprintf ( "unknown cluster '%s'", tStmt.m_sCluster.cstr() );
				return false;
			}
			sRplUser = tStmt.m_sStringParam;
			return true;
		}

		return GetClusterEffectiveUser ( tStmt.m_sCluster, sRplUser, sError );
	}

	default:
		return true;
	}
}

/////////////////////////////////////////////////////////////////////////////
// cluster joins to existed nodes
/////////////////////////////////////////////////////////////////////////////
bool ClusterJoin ( const CSphString & sCluster, const StrVec_t & dNames, const CSphVector<SqlInsert_t> & dValues, bool bUpdateNodes ) EXCLUDES ( g_tClustersLock )
{
	Threads::ScopedCoroMutex_t tClusterLock { g_tClusterOpsLock };
	TlsMsg::ResetErr();
	auto tDesc = ClusterDescFromSphinxqlStatement ( sCluster, dNames, dValues, MAKE_E::JOIN );
	if ( !tDesc )
		return false;

	CSphString sError;
	// cluster user (might be not the session user) should be able to perform replication actions
	if ( IsAuthEnabled() && !CheckPerms ( tDesc->m_sUser, AuthAction_e::REPLICATION, sCluster, true, sError ) )
		return false;

	sphLogDebugRpl ( "joining cluster '%s', nodes: %s", sCluster.cstr(), Vec2Str ( tDesc->m_dClusterNodes ).cstr() );

	// need to clean up Galera system files left from previous cluster
	CleanClusterFiles ( GetDatadirPath ( tDesc->m_sPath ) );

	CSphString sRemoteUser;
	if ( !CheckRemotesVersions ( tDesc.value(), sRemoteUser ) )
		return false;
	tDesc.value().m_sUser = sRemoteUser;

	if ( !GetRemotesAuth ( tDesc.value() ) )
		return false;

	ReplicationClusterRefPtr_c pCluster { MakeCluster ( tDesc.value(), BOOTSTRAP_E::NO, PrepareMode_e::RUNTIME ) };
	if ( !pCluster )
		return false;

	pCluster->m_bUserRequest = true;

	pCluster->m_pSstProgress->Init ( SstProgress_i::Role_e::JOINER );

	if ( !pCluster->Connect ( BOOTSTRAP_E::NO ) )
		return false;

	if ( !AddAndStartCluster ( pCluster ) )
		return false;

	pCluster->m_bUserRequest = false;

	bool bOk = IsSyncedOrDonor ( pCluster->WaitReady() );
	if ( bOk && bUpdateNodes )
		bOk &= DoClusterAlterUpdate ( sCluster, "nodes", NODES_E::BOTH );
	if ( bOk )
	{
		CSphString sSaveError;
		bOk = SaveConfigInt ( sSaveError );
		if ( !bOk )
			TlsMsg::Err ( "%s", sSaveError.cstr() );
	}

	if ( bOk )
	{
		sphLogDebugRpl ( "'%s' cluster, gtid %s, joined", sCluster.cstr(), Gtid2Str ( pCluster->m_tGtid ).cstr() );
		return true;
	}

	if ( sphInterrupted() )
		return TlsMsg::Err ( "%s", "daemon shutdown" );

	if ( !TlsMsg::HasErr() )
	{
		ScopedMutex_t tLock ( pCluster->m_tErrorLock );
		TlsMsg::Err ( pCluster->m_sError.cstr() );
	}

	sphWarning ( "'%s' cluster after join error: %s, nodes '%s'", sCluster.cstr(), TlsMsg::szError(), Vec2Str ( pCluster->m_dClusterNodes ).cstr() );
	// need to wait recv thread to complete in case of error after worker started
	pCluster->m_bWorkerActive.Wait ( [] ( bool bWorking ) { return !bWorking; } );
	Threads::SccWL_t wLock ( g_tClustersLock );
	sphLogDebugRpl ( "deleting cluster %s", sCluster.cstr() );
	g_hClusters.Delete ( sCluster );
	return false;
}

 /////////////////////////////////////////////////////////////////////////////
// cluster creates master node
/////////////////////////////////////////////////////////////////////////////
bool ClusterCreate ( const CSphString & sCluster, const StrVec_t & dNames, const CSphVector<SqlInsert_t> & dValues ) EXCLUDES ( g_tClustersLock )
{
	Threads::ScopedCoroMutex_t tClusterLock { g_tClusterOpsLock };
	TlsMsg::ResetErr();
	if ( !g_bReplicationStarted )
		return TlsMsg::Err ( "can not create cluster '%s': %s", sCluster.cstr(), g_sReplicationStartError.cstr() );

	auto tDesc = ClusterDescFromSphinxqlStatement ( sCluster, dNames, dValues, MAKE_E::CREATE );
	if ( !tDesc )
		return TlsMsg::Err ( "failed to create desc: %s", TlsMsg::szError() );

	CSphString sPermError;
	// cluster user (might be not the session user) should be able to perform replication actions
	if ( IsAuthEnabled() && !CheckPerms ( tDesc->m_sUser, AuthAction_e::REPLICATION, sCluster, true, sPermError ) )
		return TlsMsg::Err ( sPermError );

	// need to clean up Galera system files left from previous cluster
	CleanClusterFiles ( GetDatadirPath ( tDesc->m_sPath ) );

	sphLogDebugRpl ( "creating cluster '%s', nodes: %s", sCluster.cstr(), Vec2Str ( dNames ).cstr() );

	ReplicationClusterRefPtr_c pCluster { MakeCluster ( tDesc.value(), BOOTSTRAP_E::YES, PrepareMode_e::RUNTIME ) };
	if ( !pCluster )
		return TlsMsg::Err ( "failed to make cluster: %s", TlsMsg::szError() );

	if ( !pCluster->Connect ( BOOTSTRAP_E::YES ) )
		return TlsMsg::Err ( "Failed to connect %s", TlsMsg::szError() );

	if ( !AddAndStartCluster ( pCluster ) )
		return TlsMsg::Err ( "Failed to start and add %s", TlsMsg::szError() );

	auto eState = pCluster->WaitReady();

	sphLogDebugRpl ( "created cluster '%s', gtid %s", sCluster.cstr(), Gtid2Str ( pCluster->m_tGtid ).cstr() );

	TLS_MSG_STRING ( sError );
	return SaveConfigInt ( sError ) && ( IsSyncedOrDonor ( eState ) || TlsMsg::Err ( "Wrong state %s", szNodeState( eState ) ) );
}

// utility function to filter nodes list provided at string by specific protocol
StrVec_t ReplicationCluster_t::FilterViewNodesByProto ( Proto_e eProto, bool bResolve ) const
{
	Threads::SccRL_t tNodesRLock ( m_tViewNodesLock );
	return FilterNodesByProto ( m_dViewNodes, eProto, bResolve );
}

StrVec_t ReplicationCluster_t::GetViewNodes() const
{
	Threads::SccRL_t tNodesRLock ( m_tViewNodesLock );
	return m_dViewNodes;
}

void ReplicationCluster_t::SetViewNodes ( StrVec_t&& dNodes )
{
	{
		Threads::SccWL_t tNodesWLock ( m_tViewNodesLock );
		m_dViewNodes = std::move ( dNodes );
	}
	m_tViewVersion.ModifyValueAndNotifyAll ( [] ( uint64_t & uViewVersion ) { ++uViewVersion; } );
}

StrVec_t ReplicationCluster_t::GetIndexes() const noexcept
{
	StrVec_t dIndexes;
	WithRlockedAllIndexes ( [&dIndexes] ( const auto & hIndexes, auto & hIndexesLoaded )
	{
		for ( const auto & tIndex : hIndexes )
			dIndexes.Add ( tIndex.first );
		for ( const auto & tIndex : hIndexesLoaded )
		{
			assert ( !hIndexes[tIndex.first] );
			dIndexes.Add ( tIndex.first );
		}
	});
	return dIndexes;
}

void ReportClusterError ( const CSphString & sCluster, const CSphString & sError, const char * szClient, E_CLUSTER eCmd )
{
	if ( sError.IsEmpty() )
		return;

	sphWarning ( "'%s' cluster [%s], cmd: %s(%d), error: %s", sCluster.cstr(), szClient, szClusterCmd ( eCmd ), (int)eCmd, sError.cstr() );

	auto pCluster = ClusterByName ( sCluster, nullptr );
	if ( !pCluster )
		return;

	ScopedMutex_t tErrorLock ( pCluster->m_tErrorLock );
	if ( pCluster->m_sError.GetLength()>1024 ) // collect up to 1024 chars
		pCluster->m_sError.Rewind();

	pCluster->m_sError += sError.cstr();
}

// command at remote node for CLUSTER_DELETE to delete cluster
static void ClusterDetachLocal ( const ReplicationClusterRefPtr_c & pCluster ) EXCLUDES ( g_tClustersLock )
{
	assert ( pCluster );
	const CSphString & sCluster = pCluster->m_sName;
	pCluster->MarkExiting();
	sphLogDebugRpl ( "local detach cluster %s", sCluster.cstr() );

	pCluster->DisconnectAndDeleteProvider();
	pCluster->m_bWorkerActive.Wait ( [] ( bool bWorking ) { return !bWorking; } );

	{
		Threads::SccWL_t tLock ( g_tClustersLock );
		g_hClusters.Delete ( sCluster );
	}

	StrVec_t dClusterIndexes = pCluster->GetIndexes();
	pCluster->WithRlockedIndexes ( [] ( const auto& hIndexes ) { for_each ( hIndexes, [] ( const auto& tIndex ) { AssignClusterToIndex ( tIndex.first, "" ); } ); } );
	RplBinlog()->OnClusterDelete ( sCluster, dClusterIndexes );
}

static bool ClusterDeleteUnlocked ( const CSphString & sCluster ) EXCLUDES ( g_tClustersLock )
{
	auto pCluster = ClusterByName ( sCluster );
	if ( !pCluster )
		return false;

	ClusterDetachLocal ( pCluster );
	return true;
}

bool ClusterDelete ( const CSphString & sCluster ) EXCLUDES ( g_tClustersLock )
{
	Threads::ScopedCoroMutex_t tClusterLock { g_tClusterOpsLock };
	return ClusterDeleteUnlocked ( sCluster );
}

/////////////////////////////////////////////////////////////////////////////
// cluster deletes
/////////////////////////////////////////////////////////////////////////////

static void AppendWarning ( CSphString & sWarning, const CSphString & sMore )
{
	if ( sMore.IsEmpty() )
		return;

	if ( sWarning.IsEmpty() )
		sWarning = sMore;
	else
		sWarning.SetSprintf ( "%s; %s", sWarning.cstr(), sMore.cstr() );
}

// cluster delete every node, than itself
bool GloballyDeleteCluster ( const CSphString & sCluster, CSphString & sError ) EXCLUDES (g_tClustersLock)
{
	Threads::ScopedCoroMutex_t tClusterLock { g_tClusterOpsLock };
	TlsMsg::Err();
	if ( !g_bReplicationStarted )
	{
		sError.SetSprintf ( "can not delete cluster '%s': %s", sCluster.cstr(), g_sReplicationStartError.cstr() );
		return false;
	}

	auto pCluster = ClusterByName ( sCluster );
	if ( !pCluster )
	{
		TlsMsg::MoveError ( sError );
		return false;
	}

	auto dNodes = pCluster->FilterViewNodesByProto();
	SendClusterDeleteToNodes ( dNodes, sCluster, pCluster->m_sUser );
	bool bOk = ClusterDeleteUnlocked ( sCluster );
	bOk &= SaveConfigInt ( sError );

	TlsMsg::MoveError ( sError );
	return bOk;
}

bool ClusterExit ( const CSphString & sCluster, CSphString & sError, CSphString & sWarning ) EXCLUDES ( g_tClustersLock )
{
	TlsMsg::ResetErr();

	auto pCluster = ClusterByName ( sCluster );
	if ( !pCluster )
	{
		TlsMsg::MoveError ( sError );
		return false;
	}

	if ( pCluster->IsExiting() )
	{
		sError.SetSprintf ( "can not exit cluster '%s': cluster is already exiting", sCluster.cstr() );
		return false;
	}

	const auto eInitialState = pCluster->GetState();
	if ( eInitialState==ClusterState_e::JOINING || eInitialState==ClusterState_e::DONOR )
	{
		sError.SetSprintf ( "can not exit cluster '%s': current state is %s", sCluster.cstr(), szNodeState ( eInitialState ) );
		return false;
	}

	Threads::ScopedCoroMutex_t tClusterLock { g_tClusterOpsLock };

	pCluster = ClusterByName ( sCluster );
	if ( !pCluster )
	{
		TlsMsg::MoveError ( sError );
		return false;
	}

	if ( pCluster->IsExiting() )
	{
		sError.SetSprintf ( "can not exit cluster '%s': cluster is already exiting", sCluster.cstr() );
		return false;
	}

	if ( !pCluster->IsPrimary() )
	{
		sError.SetSprintf ( "can not exit cluster '%s': cluster is not primary", sCluster.cstr() );
		return false;
	}

	const auto eState = pCluster->GetState();
	if ( eState!=ClusterState_e::SYNCED )
	{
		sError.SetSprintf ( "can not exit cluster '%s': current state is %s", sCluster.cstr(), szNodeState ( eState ) );
		return false;
	}

	pCluster->MarkExiting();

	StrVec_t dPeers = pCluster->FilterViewNodesByProto ( Proto_e::SPHINX, false );
	const CSphString sClusterUser = pCluster->m_sUser;
	const CSphString sLeavingNode = szIncomingProto();

	ClusterDetachLocal ( pCluster );

	TLS_MSG_STRING ( sSaveError );
	if ( !SaveConfigInt ( sSaveError ) )
		AppendWarning ( sWarning, SphSprintf ( "cluster '%s' exited locally but local config was not saved: %s", sCluster.cstr(), sSaveError.cstr() ) );

	if ( !SendClusterExitUpdateNodes ( sCluster, sClusterUser, sLeavingNode, g_iClusterExitUpdateNodesWaitMs, dPeers ) )
	{
		CSphString sCleanupError = TlsMsg::MoveToString();
		CSphString sCleanupWarning;
		if ( sCleanupError.IsEmpty() )
			sCleanupWarning.SetSprintf ( "cluster '%s' exited locally but remaining nodes may still need manual ALTER CLUSTER %s UPDATE nodes", sCluster.cstr(), sCluster.cstr() );
		else
			sCleanupWarning.SetSprintf ( "cluster '%s' exited locally but remaining nodes cleanup failed: %s; run ALTER CLUSTER %s UPDATE nodes on a surviving node if needed", sCluster.cstr(), sCleanupError.cstr(), sCluster.cstr() );
		AppendWarning ( sWarning, sCleanupWarning );
	}

	return true;
}

bool ClusterGetState ( const CSphString & sCluster, RemoteNodeClusterState_t & tState ) EXCLUDES ( g_tClustersLock )
{
	auto pCluster = ClusterByName ( sCluster );
	if ( !pCluster )
		return false;

	tState.m_eState = pCluster->GetState();
	tState.m_sNode = pCluster->GetNodeName();
	return true;
}

bool ClusterGetDonorMeta ( const CSphString & sCluster, CSphString & sUser ) EXCLUDES ( g_tClustersLock )
{
	auto pCluster = ClusterByName ( sCluster );
	if ( !pCluster )
		return false;

	sUser = pCluster->GetUserSnapshot();
	return true;
}

int64_t ClusterGetRemoteEpoch ( const CSphString & sCluster ) EXCLUDES ( g_tClustersLock )
{
	auto pCluster = ClusterByName ( sCluster );
	if ( !pCluster || !pCluster->IsPrimary() || pCluster->GetState() != ClusterState_e::SYNCED )
		return 0;

	return pCluster->m_iClusterEpoch;
}

static bool HasNotReadyNodes ( ReplicationClusterRefPtr_c pCluster )
{
	ClusterState_e eState = pCluster->GetState();
	if ( eState==ClusterState_e::DONOR || eState==ClusterState_e::JOINING )
		return true;

	const auto dStates = GetStatesFromRemotes ( *pCluster );
	return dStates.any_of ( []( auto & tState ) { return ( tState.m_eState==ClusterState_e::DONOR || tState.m_eState==ClusterState_e::JOINING ); });
}

// cluster ALTER statement that removes index from cluster but keep it at daemon
static bool ClusterAlterDrop ( const CSphString & sCluster, const VecTraits_T<CSphString> & dIndexes )
{
	RtAccum_t tAcc;
	auto pCluster = ClusterByName ( sCluster );
	if ( !pCluster )
		return false;

	ReplicationCommand_t * pDropCmd = tAcc.AddCommand ( ReplCmd_e::CLUSTER_ALTER_DROP, Vec2Str ( dIndexes ), sCluster );
	pDropCmd->m_iClusterEpoch = ( pCluster->m_iClusterEpoch + 1 );
	return HandleCmdReplicate ( tAcc );
}

// FIXME!!! refactor to use same code as SendClusterIndexes
static bool SendIndex ( const CSphString & sIndex, ReplicationClusterRefPtr_c pCluster )
{
	cServedIndexRefPtr_c pServed = GetServed ( sIndex );
	bool bMutable = ServedDesc_t::IsMutable ( pServed );
	bool bDist = ( !bMutable && g_pDistIndexes->Contains ( sIndex ) );

	if ( !bMutable && !bDist )
		return TlsMsg::Err ( "unknown or wrong table '%s'", sIndex.cstr() );

	pCluster->m_pSstProgress->SetTable ( sIndex );

	int iAttempt = 0;
	// should wait a bit longer during join
	// FIXME!!! fetch progress delta time and continue to wait while there is a progress
	const int iRetryCount = ReplicationRetryCount() * 2;
	const int iRetryDelay = ReplicationRetryDelay() * 2;
	int64_t tmStart = sphMicroTimer();
	while ( true )
	{
		pCluster->m_pSstProgress->StageBegin ( SstStage_e::WAIT_NODES );
		if ( HasNotReadyNodes ( pCluster ) )
		{
			iAttempt++;
			if ( iAttempt>=iRetryCount )
			{
				int64_t tmEnd = sphMicroTimer();
				return TlsMsg::Err ( "alter '%s' has some nodes not ready for %.3f seconds", pCluster->m_sName.cstr(), (float)(tmEnd - tmStart)/1000000.0f );
			}

			// FIXME!!! send index only into new node in case of next try happens
			sphLogDebugRpl ( "alter '%s' has some nodes not ready, will wait for %d seconds before retry %d", pCluster->m_sName.cstr(), iRetryDelay / 1000, iAttempt );
			Threads::Coro::SleepMsec ( iRetryDelay );
			continue;
		}

		auto dNodes = pCluster->FilterViewNodesByProto ( Proto_e::SPHINX, false );
		sphLogDebugRpl ( "alter '%s' SST index '%s' to nodes %d: '%s'", pCluster->m_sName.cstr(), sIndex.cstr(), dNodes.GetLength(), Vec2Str ( dNodes ).cstr() );

		// ok for just created cluster (wo nodes) to add existed index
		if ( !dNodes.IsEmpty() )
		{
			VecAgentDesc_t dDesc = GetDescAPINodes ( dNodes, Resolve_e::SLOW );
			if ( TlsMsg::HasErr() )
				return false;

			sphLogDebugRpl ( "alter '%s' SST index '%s' to resolved nodes %d", pCluster->m_sName.cstr(), sIndex.cstr(), dDesc.GetLength() );

			if ( dDesc.GetLength() )
			{
				bool bReplicated = ( bMutable ? ReplicateIndexToNodes ( pCluster->m_sName, sIndex, pCluster->m_sUser, dDesc, pServed, *pCluster->m_pSstProgress ) : ReplicateDistIndexToNodes ( pCluster->m_sName, sIndex, pCluster->m_sUser, dDesc ) );
				if ( !bReplicated )
				{
					if ( TlsMsg::HasErr() )
						sphWarning ( "%s", TlsMsg::szError() );
					return false;
				}
			}
		}

		// nodes list might change during alter at the other node
		auto dNewNodes = pCluster->FilterViewNodesByProto ( Proto_e::SPHINX, false );

		// passed fine no join during the alter and the cluster remains same
		if ( !HasNotReadyNodes ( pCluster ) && IsSameVector ( dNodes, dNewNodes ) )
			break;

		sphLogDebugRpl ( "nodes not ready during alter '%s', wait for %d sec before retry %d", pCluster->m_sName.cstr(), iRetryDelay / 1000, iAttempt );
		Threads::Coro::SleepMsec ( iRetryDelay );
		// no need to increase attempt count here as it will be checked on next try
	}

	return true;
}


struct LoadedIndexesClusterCleanup_t
{
	LoadedIndexesClusterCleanup_t ( bool & bAdded, const VecTraits_T<CSphString> & dIndexes, ReplicationClusterRefPtr_c pCluster )
		: m_bAdded ( bAdded )
		, m_dIndexes ( dIndexes )
		, m_pCluster ( pCluster )
	{}
	~LoadedIndexesClusterCleanup_t()
	{
		if ( !m_bAdded )
		{
			m_pCluster->WithWlockedIndexes ( [this] ( auto & hIndexes, auto & hIndexesLoaded )
			{
					for ( const CSphString & sIndex : m_dIndexes )
						hIndexesLoaded.Delete ( sIndex );
			});
		}
	}

	bool & m_bAdded;
	const VecTraits_T<CSphString> & m_dIndexes;
	ReplicationClusterRefPtr_c m_pCluster;
};

// cluster ALTER statement that adds index
static bool ClusterAlterAdd ( const CSphString & sCluster, const VecTraits_T<CSphString> & dIndexes )
	EXCLUDES ( g_tClustersLock )
{
	auto pCluster = ClusterByName ( sCluster );
	if ( !pCluster )
		return false;

	if ( !pCluster->IsHealthy() )
		return false;

	// resolve nodes ONCE before indexes loop
	auto dNodes = pCluster->FilterViewNodesByProto ( Proto_e::SPHINX, false );
	VecAgentDesc_t dDesc = GetDescAPINodes ( dNodes, Resolve_e::SLOW );
	if ( TlsMsg::HasErr() )
		return false;

	AT_SCOPE_EXIT ([&pCluster] { pCluster->m_pSstProgress->Finish(); } );
	pCluster->m_pSstProgress->Init ( SstProgress_i::Role_e::DONOR );

	// collect stats for all tables to get correct overall weights
	pCluster->m_pSstProgress->CollectFilesStats ( dIndexes, 0, dDesc.GetLength() );
	
	// start pushing updates to the all other nodes
	SstProgress_i::StartPushUpdates ( pCluster->m_sName, pCluster->m_sUser, dDesc, pCluster->m_pSstProgress );

	for ( const CSphString & sIndex : dIndexes )
	{
		if ( !SendIndex ( sIndex, pCluster ) )
			return false;
	}

	bool bAdded = false;
	pCluster->WithWlockedIndexes ( [&dIndexes] ( auto & hIndexes, auto & hIndexesLoaded )
	{
		for ( const CSphString & sIndex : dIndexes )
			hIndexesLoaded.Add ( sIndex );
	});
	LoadedIndexesClusterCleanup_t tCleanup ( bAdded, dIndexes, pCluster );

	sphLogDebugRpl ( "alter '%s' adding index '%s'", pCluster->m_sName.cstr(), Vec2Str ( dIndexes ).cstr() );
	RtAccum_t tAcc;
	ReplicationCommand_t * pAddCmd = tAcc.AddCommand ( ReplCmd_e::CLUSTER_ALTER_ADD, Vec2Str ( dIndexes ), sCluster );
	pAddCmd->m_bCheckIndex = false;
	pAddCmd->m_iClusterEpoch = ( pCluster->m_iClusterEpoch + 1 );
	
	bAdded = HandleCmdReplicate ( tAcc );
	sphLogDebugRpl ( "alter '%s' %s index '%s'", pCluster->m_sName.cstr(), ( bAdded ? "added" : "failed to add" ), Vec2Str ( dIndexes ).cstr() );
	return bAdded;
}

static bool ClusterAddCheckDistLocals ( const StrVec_t & dLocals, const CSphString & sCluster, const CSphString & sIndex, CSphString & sError )
{
	StringBuilder_c sMissed ( "," );
	StringBuilder_c sNonCluster ( "," );
	StringBuilder_c sWrongCluster ( "," );

	for ( const CSphString & sLocal : dLocals )
	{
		cServedIndexRefPtr_c pServed = GetServed ( sLocal );
		if ( !ServedDesc_t::IsMutable ( pServed ) )
			sMissed += sLocal.cstr();
		else if ( pServed->m_sCluster.IsEmpty() )
			sNonCluster += sLocal.cstr();
		else if ( pServed->m_sCluster!=sCluster )
			sWrongCluster += sLocal.cstr();
	}

	if ( sMissed.IsEmpty() && sNonCluster.IsEmpty() && sWrongCluster.IsEmpty() )
		return true;

	StringBuilder_c sMsg;
	sMsg.Appendf ( "can not add distributed table '%s' into cluster '%s';", sIndex.cstr(), sCluster.cstr() );
	sMsg.StartBlock ( "; " );
	if ( !sMissed.IsEmpty() )
		sMsg.Appendf ( "has unknown local tables: %s", sMissed.cstr() );
	if ( !sNonCluster.IsEmpty() )
		sMsg.Appendf ( "has tables not in the cluster: %s", sNonCluster.cstr() );
	if ( !sWrongCluster.IsEmpty() )
		sMsg.Appendf ( "has tables in the other cluster: %s", sNonCluster.cstr() );
	sMsg.FinishBlock();

	sError = sMsg.cstr();
	return false;
}

// cluster ALTER statement
bool ClusterAlter ( const CSphString & sCluster, StrVec_t& dIndexes, bool bAdd, CSphString & sError )
{
	dIndexes.Uniq();

	Threads::ScopedCoroMutex_t tClusterLock { g_tClusterOpsLock };
	{
		for ( const CSphString & sIndex : dIndexes )
		{
			cServedIndexRefPtr_c pServed = GetServed ( sIndex );
			bool bMutable = ServedDesc_t::IsMutable ( pServed );
			cDistributedIndexRefPtr_t pDist ( !bMutable ? GetDistr ( sIndex ) : nullptr );
			if ( !bMutable && !pDist )
			{
				sError.SetSprintf ( "unknown or wrong type of table '%s'", sIndex.cstr() );
				return false;
			}

			const CSphString & sIndexCluster = ( bMutable ? pServed->m_sCluster : pDist->m_sCluster );
			if ( bAdd )
			{
				if ( !sIndexCluster.IsEmpty() )
				{
					sError.SetSprintf ( "table '%s' is already part of cluster '%s'", sIndex.cstr(), sIndexCluster.cstr() );
					return false;
				}
				// all local indexes should be part of the cluster too
				if ( pDist && !ClusterAddCheckDistLocals ( pDist->m_dLocal, sCluster, sIndex, sError ) )
					return false;
			} else
			{
				if ( sIndexCluster.IsEmpty() )
				{
					sError.SetSprintf ( "table '%s' is not in cluster '%s'", sIndex.cstr(), sCluster.cstr() );
					return false;
				}
			}
		}
	}

	if ( !g_bReplicationStarted )
	{
		sError.SetSprintf ( "can not ALTER cluster '%s': %s", sCluster.cstr(), g_sReplicationStartError.cstr() );
		return false;
	}

	bool bOk = false;
	if ( bAdd )
		bOk = ClusterAlterAdd ( sCluster, dIndexes );
	else
		bOk = ClusterAlterDrop ( sCluster, dIndexes );

	TlsMsg::MoveError ( sError );
	bOk &= SaveConfigInt ( sError );

	return bOk;
}

/////////////////////////////////////////////////////////////////////////////
// SST
/////////////////////////////////////////////////////////////////////////////

bool IsSameVector ( StrVec_t & dSrc, StrVec_t & dDst )
{
	if ( dSrc.GetLength()!=dDst.GetLength() )
		return false;

	dSrc.Sort();
	dDst.Sort();

	ARRAY_FOREACH ( i, dSrc )
	{
		if ( dSrc[i]!=dDst[i] )
			return false;
	}

	return true;
}

bool AddLoadedIndexIntoCluster ( const CSphString & sCluster, const CSphString & sIndex )
{
	auto pCluster = ClusterByName ( sCluster );
	if ( !pCluster )
		return false;

	pCluster->WithWlockedIndexes ( [&sIndex] ( auto & hIndexes, auto & hIndexesLoaded ) { hIndexesLoaded.Add ( sIndex ); });
	return true;
}

// send local indexes to remote nodes via API
bool SendClusterIndexes ( ReplicationCluster_t * pCluster, const CSphString & sNode, bool bBypass, const Wsrep::GlobalTid_t & tStateID )
{
	auto dDesc = GetDescAPINodes ( ParseNodesFromString ( sNode ), Resolve_e::SLOW );
	if ( dDesc.IsEmpty() )
	{
		if ( TlsMsg::HasErr() )
			TlsMsg::Err ( "%s invalid node, error: %s", sNode.cstr(), TlsMsg::MoveToString().cstr() );
		else
			TlsMsg::Err ( "%s invalid node", sNode.cstr() );
		return false;
	}

	auto dIndexes = pCluster->GetIndexes();

	sphLogDebugRpl ( "SendClusterIndexes '%s' to '%s': state gtid %s, current gtid %s, indexes %d '%s'...'%s', bypass %d", pCluster->m_sName.cstr(), sNode.cstr(), Wsrep::Gtid2Str ( tStateID ).cstr(), Wsrep::Gtid2Str ( pCluster->m_tGtid ).cstr(), dIndexes.GetLength(), ( dIndexes.GetLength() ? dIndexes.First().cstr() : "" ), ( dIndexes.GetLength() ? dIndexes.Last().cstr() : "" ), (int)bBypass );

	ClusterSyncedRequest_t tSyncedRequest;
	tSyncedRequest.m_sCluster = pCluster->m_sName;
	tSyncedRequest.m_tGtid = tStateID;
	tSyncedRequest.m_iClusterEpoch = pCluster->m_iClusterEpoch;
	tSyncedRequest.m_dIndexes = dIndexes;

	if ( bBypass )
		return SendClusterSynced ( dDesc, tSyncedRequest, pCluster->m_sUser );

	AT_SCOPE_EXIT ([&pCluster] { pCluster->m_pSstProgress->Finish(); } );
	pCluster->m_pSstProgress->Init ( SstProgress_i::Role_e::DONOR );
	SstProgress_i::StartPushUpdates ( pCluster->m_sName, pCluster->m_sUser, dDesc, pCluster->m_pSstProgress );

	bool bSentOk = true;
	int iAttempt = 0;
	while ( true )
	{
		sphLogDebugRpl ( "sending cluster '%s' indexes %d '%s'...'%s'", pCluster->m_sName.cstr(), dIndexes.GetLength(), ( dIndexes.GetLength() ? dIndexes.First().cstr() : "" ), ( dIndexes.GetLength() ? dIndexes.Last().cstr() : "" ) );

		pCluster->m_pSstProgress->CollectFilesStats ( dIndexes, iAttempt, dDesc.GetLength() );

		for ( const CSphString & sIndex : dIndexes )
		{
			cServedIndexRefPtr_c pServed = GetServed ( sIndex );
			bool bMutable = ServedDesc_t::IsMutable ( pServed );
			cDistributedIndexRefPtr_t pDist ( !bMutable ? GetDistr ( sIndex ) : nullptr );
			if ( !bMutable && !pDist )
			{
				bSentOk = false;
				sphWarning ( "unknown or wrong table '%s'", sIndex.cstr() );
				continue;
			}

			pCluster->m_pSstProgress->SetTable ( sIndex );

			bool bReplicated = ( bMutable ? ReplicateIndexToNodes ( pCluster->m_sName, sIndex, pCluster->m_sUser, dDesc, pServed, *pCluster->m_pSstProgress ) : ReplicateDistIndexToNodes ( pCluster->m_sName, sIndex, pCluster->m_sUser, dDesc ) );
			if ( !bReplicated )
			{
				sphWarning ( "%s", TlsMsg::szError() );
				bSentOk = false;
				break;
			}

			if ( TlsMsg::HasErr() )
				sphWarning ( "%s", TlsMsg::szError() );
		}

		// index list could have changed due to alter at the other node
		auto dNewIndexes = pCluster->GetIndexes();
		if ( dNewIndexes.GetLength() )
			sphLogDebugRpl ( "sent cluster '%s' indexes %d '%s'...'%s'", pCluster->m_sName.cstr(), dNewIndexes.GetLength(), dNewIndexes.First().cstr(), dNewIndexes.Last().cstr() );

		if ( IsSameVector ( dIndexes, dNewIndexes ) )
			break;

		sphLogDebugRpl ( "index list changed during donate '%s' to '%s'", pCluster->m_sName.cstr(), sNode.cstr() );
		// FIXME!!! send only new indexes but loads all cluster indexes
		dIndexes = dNewIndexes;
		iAttempt++;
	}

	tSyncedRequest.m_dIndexes = dIndexes;
	tSyncedRequest.m_iClusterEpoch = pCluster->m_iClusterEpoch;
	tSyncedRequest.m_bSendFilesSuccess = bSentOk;
	tSyncedRequest.m_sMsg = TlsMsg::MoveToString();
	bool bSyncOk = SendClusterSynced ( dDesc, tSyncedRequest, pCluster->m_sUser );
	return bSentOk && bSyncOk;
}

// callback at remote node for CLUSTER_SYNCED to pick up received indexes then call Galera sst_received
bool ClusterSynced ( const ClusterSyncedRequest_t & tCmd ) EXCLUDES ( g_tClustersLock )
{
	sphLogDebugRpl ( "join sync '%s', gtid %s, epoch " INT64_FMT ", sent %s, tables %d, %s", tCmd.m_sCluster.cstr(), Wsrep::Gtid2Str ( tCmd.m_tGtid ).cstr(), tCmd.m_iClusterEpoch, ( tCmd.m_bSendFilesSuccess ? "ok" : "failed" ), tCmd.m_dIndexes.GetLength(), tCmd.m_sMsg.scstr() );

	if ( !tCmd.m_bSendFilesSuccess )
	{
		if ( tCmd.m_sMsg.IsEmpty() )
			TlsMsg::Err ( "donor failed to send files" );
		else
			TlsMsg::Err ( tCmd.m_sMsg );
	}

	Wsrep::GlobalTid_t tGtid {}; // by default is 'UNDEFINED'
	bool bValid = ( tCmd.m_bSendFilesSuccess && tCmd.m_tGtid != tGtid );

	if ( bValid )
		bValid &= ReplicatedIndexes ( tCmd.m_dIndexes, tCmd.m_sCluster, tCmd.m_iClusterEpoch );

	auto pCluster = ClusterByName ( tCmd.m_sCluster );
	if ( !pCluster )
		return false;

	tGtid = tCmd.m_tGtid;
	int iRes = 0;
	if ( !bValid )
	{
		tGtid.m_iSeqNo = Wsrep::WRONG_SEQNO;
		iRes = -ECANCELED;
	}

	pCluster->m_pProvider->SstReceived ( tGtid, iRes );
	return bValid;
}

// validate that SphinxQL statement could be run for this cluster:index 
bool ValidateClusterStatement ( const CSphString & sIndexName, const CSphString & sIndexCluster, const CSphString & sStmtCluster, bool bHTTP )
{
	if ( sIndexCluster==sStmtCluster )
		return true;

	if ( sIndexCluster.IsEmpty() )
		return TlsMsg::Err ( "table '%s' is not in any cluster, use just '%s'", sIndexName.cstr(), sIndexName.cstr() );

	if ( !bHTTP )
		return TlsMsg::Err ( "table '%s' is a part of cluster '%s', use '%s:%s'", sIndexName.cstr(), sIndexCluster.cstr(), sIndexCluster.cstr(), sIndexName.cstr() );

	return TlsMsg::Err( R"(table '%s' is a part of cluster '%s', use "cluster":"%s" and "table":"%s" properties)", sIndexName.cstr(), sIndexCluster.cstr(), sIndexCluster.cstr(), sIndexName.cstr() );
}

bool ValidateClusterStatement ( const CSphString & sIndexName, const ServedDesc_t & tDesc, const CSphString & sStmtCluster, bool bHTTP )
{
	return ValidateClusterStatement ( sIndexName, tDesc.m_sCluster, sStmtCluster, bHTTP );
}

std::optional<CSphString> IsPartOfCluster ( const ServedDesc_t * pDesc )
{
	assert ( pDesc );
	if ( !pDesc->m_sCluster.IsEmpty() )
		return pDesc->m_sCluster;
	return {};
}

// return set of nodes for given cluster (both, clusternodes and viewnodes)
StrVec_t ClusterGetAllNodes ( const CSphString & sCluster ) EXCLUDES ( g_tClustersLock )
{
	TlsMsg::Err();
	auto pCluster = ClusterByName ( sCluster );
	if ( !pCluster )
		return {};

	// send back view nodes of cluster - as list of actual nodes
	sph::StringSet hNodes;
	for_each ( pCluster->GetViewNodes(), [&hNodes] ( const auto& sNode ) { hNodes.Add ( sNode ); } );
	for_each ( pCluster->m_dClusterNodes, [&hNodes] ( const auto& sNode ) { hNodes.Add ( sNode ); } );
	StrVec_t dNodes;
	for_each ( hNodes, [&dNodes] ( const auto& tNode ) { dNodes.Add ( tNode.first ); } );
	return dNodes;
//	return FilterNodesByProto ( dNodes, Proto_e::REPLICATION );
}

// cluster ALTER statement that updates nodes option from view nodes at all nodes at cluster
bool ClusterAlterUpdate ( const CSphString & sCluster, const CSphString & sUpdate, const CSphString & sValue, CSphString & sError )
{
	Threads::ScopedCoroMutex_t tClusterLock { g_tClusterOpsLock };
	TlsMsg::ResetErr();

	bool bOk = false;
	if ( sUpdate=="nodes" )
	{
		bOk = DoClusterAlterUpdate ( sCluster, sUpdate, NODES_E::VIEW );
	} else if ( sUpdate=="user" )
	{
		if ( sValue.IsEmpty() )
			bOk = TlsMsg::Err ( "invalid cluster user in UPDATE statement" );
		else
		{
			RtAccum_t tAcc;
			ReplicationCommand_t * pCmd = tAcc.AddCommand ( ReplCmd_e::CLUSTER_ALTER_UPDATE_USER, sValue, sCluster );
			pCmd->m_bCheckIndex = false;
			bOk = HandleCmdReplicate ( tAcc );
		}
	} else
		bOk = TlsMsg::Err ( "unhandled statement, only UPDATE nodes or UPDATE user are supported, got '%s'", sUpdate.cstr() );

	TlsMsg::MoveError ( sError );
	return bOk;
}

bool DoClusterAlterUpdate ( const CSphString & sCluster, const CSphString & sUpdate, NODES_E eNodes ) EXCLUDES ( g_tClustersLock )
{
	TlsMsg::ResetErr();
	if ( sUpdate!="nodes" )
		return TlsMsg::Err ( "unhandled statement, only UPDATE nodes are supported, got '%s'", sUpdate.cstr() );

	auto pCluster = ClusterByName ( sCluster );
	if ( !pCluster || !pCluster->IsHealthy() )
		return false;

	// need to update all VIEW nodes - not cluster set nodes
	StrVec_t dNodes;

	// local nodes update
	bool bOk = ClusterUpdateNodes ( sCluster, eNodes );
	dNodes = pCluster->m_dClusterNodes;

	if ( dNodes.IsEmpty() )
		return false;

	// remote nodes update after locals updated
	if ( !SendClusterUpdateNodes ( sCluster, pCluster->m_sUser, eNodes, dNodes ) )
	{
		sphWarning ( "cluster %s nodes update error %s", sCluster.cstr(), TlsMsg::szError() );
		TlsMsg::ResetErr();
	}

	return bOk;
}

static uint64_t CalcNodesHash ( const StrVec_t & dNodes )
{
	uint64_t uRes = SPH_FNV64_SEED;
	dNodes.for_each ( [&uRes] ( const auto & sNode ) { uRes = sphFNV64cont ( sNode.cstr(), uRes ); } );
	return uRes;
}

static bool ApplyClusterNodesSnapshot ( ReplicationCluster_t * pCluster, StrVec_t dNodes, bool bSaveAlways )
{
	assert ( pCluster );

	TlsMsg::ResetErr();

	uint64_t uWasNodes = CalcNodesHash ( pCluster->m_dClusterNodes );
	pCluster->m_dClusterNodes = FilterNodesByProto ( dNodes, Proto_e::SPHINX );
	if ( TlsMsg::HasErr() )
		return TlsMsg::Err ( "cluster '%s', invalid nodes, error: %s", pCluster->m_sName.cstr(), TlsMsg::szError() );

	bool bSaveConf = ( bSaveAlways || ( uWasNodes!=CalcNodesHash ( pCluster->m_dClusterNodes ) ) );
	if ( !bSaveConf )
		return true;

	TLS_MSG_STRING ( sError );
	if ( SaveConfigInt ( sError ) )
		return true;

	return TlsMsg::Err ( "cluster '%s', failed to save config: %s", pCluster->m_sName.cstr(), sError.cstr() );
}

static bool HasViewNode ( ReplicationCluster_t * pCluster, const CSphString & sNode )
{
	assert ( pCluster );

	StrVec_t dViewNodes = pCluster->FilterViewNodesByProto ( Proto_e::SPHINX, false );
	return dViewNodes.any_of ( [&sNode] ( const auto & sViewNode ) { return sViewNode==sNode; } );
}

static bool WaitUntilNodeLeavesView ( ReplicationCluster_t * pCluster, const CSphString & sNode, int64_t iWaitTimeoutMs )
{
	assert ( pCluster );
	assert ( iWaitTimeoutMs>0 );

	int64_t tmDeadline = sphMicroTimer() + iWaitTimeoutMs * 1000;
	while ( true )
	{
		uint64_t uViewVersion = pCluster->GetViewVersion();
		if ( !HasViewNode ( pCluster, sNode ) )
			return true;

		if ( sphMicroTimer()>=tmDeadline )
			break;

		if ( !pCluster->WaitViewChangeUntil ( uViewVersion, tmDeadline ) )
			break;
	}

	return !HasViewNode ( pCluster, sNode );
}

// callback at remote node for CLUSTER_UPDATE_NODES to update nodes list at cluster from actual nodes list
bool ClusterUpdateNodes ( const CSphString & sCluster, NODES_E eNodes ) EXCLUDES ( g_tClustersLock )
{
	auto pCluster = ClusterByName ( sCluster );
	if ( !pCluster || !pCluster->IsHealthy() )
	{
		// node in the joining state should skip the command
		if ( pCluster && pCluster->GetState()==ClusterState_e::JOINING )
		{
			TlsMsg::ResetErr();
			return true;
		}
		return false;
	}

	StrVec_t dNodes = pCluster->GetViewNodes();
	if ( eNodes==NODES_E::BOTH )
		dNodes.Append ( pCluster->m_dClusterNodes );

	return ApplyClusterNodesSnapshot ( pCluster.Ptr(), std::move ( dNodes ), false );
}

bool ClusterExitUpdateNodes ( const CSphString & sCluster, const CSphString & sLeavingNode, int64_t iWaitTimeoutMs ) EXCLUDES ( g_tClustersLock )
{
	TlsMsg::ResetErr();

	auto pCluster = ClusterByName ( sCluster );
	if ( !pCluster || !pCluster->IsHealthy() )
		return false;

	if ( !WaitUntilNodeLeavesView ( pCluster.Ptr(), sLeavingNode, iWaitTimeoutMs ) )
		return TlsMsg::Err ( "cluster '%s', timeout waiting for node '%s' to leave local view", sCluster.cstr(), sLeavingNode.cstr() );

	Threads::ScopedCoroMutex_t tClusterLock { g_tClusterOpsLock };

	pCluster = ClusterByName ( sCluster );
	if ( !pCluster || !pCluster->IsHealthy() )
		return false;

	// Revalidate the exact view snapshot we are about to persist after the unlocked wait above.
	StrVec_t dViewNodes = pCluster->FilterViewNodesByProto ( Proto_e::SPHINX, false );
	if ( dViewNodes.any_of ( [&sLeavingNode] ( const auto & sViewNode ) { return sViewNode==sLeavingNode; } ) )
		return TlsMsg::Err ( "cluster '%s', node '%s' is still present in local view", sCluster.cstr(), sLeavingNode.cstr() );

	return ApplyClusterNodesSnapshot ( pCluster.Ptr(), std::move ( dViewNodes ), true );
}

void ReplicationCluster_t::OnSeqnoCommited ( const ReplicatedCommand_t & tCmd, int64_t iSeqNo )
{
	m_tGtid.m_iSeqNo = iSeqNo;
	RplBinlog()->ClusterTnx ( *this );
}

CSphRefcountedPtr<SstProgress_i> GetClusterProgress ( const CSphString & sCluster )
{
	auto pCluster = ClusterByName ( sCluster );
	if ( pCluster )
	{
		CSphRefcountedPtr<SstProgress_i> pProgress { pCluster->m_pSstProgress };
		return pProgress;
	}

	return nullptr;
}

CSphString GetFirstClusterName()
{
	CSphString sCluster;
	{
		Threads::SccRL_t tLock ( g_tClustersLock );
		if ( g_hClusters.GetLength()==1 )
			sCluster = g_hClusters.begin()->second->m_sName;
	}

	return sCluster;
}
	
void ForceSst ( ReplicationCluster_t & tCluster )
{
	auto sClusterPath = GetDatadirPath ( tCluster.m_sPath );
	CleanClusterFiles ( sClusterPath );
	RplBinlog()->InvalidateCluster ( tCluster.m_sName );
	tCluster.m_tGtid = {};
}

bool CheckForceSst ( const ReplicationCluster_t & tCluster, int64_t iRemoteEpoch )
{
	if ( tCluster.m_iClusterEpoch == 0 )
		return true;

	if ( iRemoteEpoch == 0 )
		return true;

	return iRemoteEpoch != tCluster.m_iClusterEpoch;
}
