LANGUAGE C;

CREATETYPEmtm.node_stateAS (idinteger, disabled bool, disconnected bool, catchUp bool, slotLagbigint, avgTransDelaybigint, connStrtext);

CREATETYPEmtm.node_stateAS (idinteger, disabled bool, disconnected bool, catchUp bool, slotLagbigint, avgTransDelaybigint,lastStatusChangetimestamp,connStrtext);

CREATEFUNCTIONmtm.get_nodes_state() RETURNS SETOFmtm.node_state

AS'MODULE_PATHNAME','mtm_get_nodes_state'

intMtmConnectTimeout;

intMtmKeepaliveTimeout;

intMtmReconnectAttempts;

intMtmNodeDisableDelay;

boolMtmUseRaftable;

MtmConnectionInfo*MtmConnections;

MtmLock(LW_EXCLUSIVE);

Mtm->recoverySlot=0;

BIT_CLEAR(Mtm->disabledNodeMask,MtmNodeId-1);

Mtm->nodes[MtmNodeId-1].lastStatusChangeTime=time(NULL);

MtmSwitchClusterMode(MTM_CONNECTED);

MtmUnlock();

}

BIT_CLEAR(Mtm->disabledNodeMask,nodeId-1);

Mtm->nodes[nodeId-1].lastStatusChangeTime=time(NULL);

Mtm->nNodes+=1;

caughtUp= true;

}elseif (!BIT_CHECK(Mtm->nodeLockerMask,nodeId-1)

if (mask&1) {

Mtm->nNodes-=1;

BIT_SET(Mtm->disabledNodeMask,i);

Mtm->nodes[i].lastStatusChangeTime=time(NULL);

}

}

mask=clique&Mtm->disabledNodeMask;/* new enabled nodes mask */

for (i=0;mask!=0;i++,mask >>=1) {

if (mask&1) {

Mtm->nNodes+=1;

BIT_CLEAR(Mtm->disabledNodeMask,i);

Mtm->nodes[i].lastStatusChangeTime=time(NULL);

}

}

MtmCheckQuorum();

{

MtmTransState*ts;

if (Mtm->nodes[nodeId-1].lastStatusChangeTime+MtmNodeDisableDelay>time(NULL)) {

return;

}

BIT_SET(Mtm->connectivityMask,nodeId-1);

BIT_SET(Mtm->reconnectMask,nodeId-1);

RaftableSet(psprintf("node-mask-%d",MtmNodeId),&Mtm->connectivityMask,sizeofMtm->connectivityMask, false);

if (!MtmRefreshClusterStatus(false)) {

MtmLock(LW_EXCLUSIVE);

if (!BIT_CHECK(Mtm->disabledNodeMask,nodeId-1)) {

Mtm->nodes[nodeId-1].lastStatusChangeTime=time(NULL);

BIT_SET(Mtm->disabledNodeMask,nodeId-1);

Mtm->nNodes-=1;

MtmCheckQuorum();

for (i=0;i<MtmNodes;i++) {

Mtm->nodes[i].oldestSnapshot=0;

Mtm->nodes[i].transDelay=0;

Mtm->nodes[i].lastStatusChangeTime=time(NULL);

Mtm->nodes[i].con=MtmConnections[i];

}

PGSemaphoreCreate(&Mtm->votingSemaphore);

if (!process_shared_preload_libraries_in_progress)

return;

DefineCustomIntVariable(

"multimaster.node_disable_delay",

"Minamal amount of time (sec) between node status change",

"This delay is used to avoid false detection of node failure and to prevent blinking of node status node",

&MtmNodeDisableDelay,

1,

1,

INT_MAX,

PGC_BACKEND,

0,

NULL,

NULL,

);

DefineCustomIntVariable(

"multimaster.min_recovery_lag",

"Minamal lag of WAL-sender performing recovery after which cluster is locked until recovery is completed",

"temporary stop commit of new transactions until node will be completely repared",

&MtmMinRecoveryLag,

100000,

{

elog(ERROR,"NodeID %d is out of range [1,%d]",nodeId,Mtm->nNodes);

}

Mtm->nodes[nodeId-1].lastStatusChangeTime=time(NULL);

BIT_SET(Mtm->disabledNodeMask,nodeId-1);

Mtm->nNodes-=1;

MtmCheckQuorum();

if (MtmIsRecoverySession) {

MTM_LOG1("%d: Node %d start recovery of node %d",MyProcPid,MtmNodeId,MtmReplicationNodeId);

if (!BIT_CHECK(Mtm->disabledNodeMask,MtmReplicationNodeId-1)) {

Mtm->nodes[MtmReplicationNodeId-1].lastStatusChangeTime=time(NULL);

BIT_SET(Mtm->disabledNodeMask,MtmReplicationNodeId-1);

Mtm->nNodes-=1;

MtmCheckQuorum();

}

}elseif (BIT_CHECK(Mtm->disabledNodeMask,MtmReplicationNodeId-1)) {

if (recoveryCompleted) {

MTM_LOG1("Node %d consider that recovery of node %d is completed: start normal replication",MtmNodeId,MtmReplicationNodeId);

Mtm->nodes[MtmReplicationNodeId-1].lastStatusChangeTime=time(NULL);

BIT_CLEAR(Mtm->disabledNodeMask,MtmReplicationNodeId-1);

Mtm->nNodes+=1;

MtmCheckQuorum();

intnodeId;

char*connStrPtr;

TupleDescdesc;

Datumvalues[7];

boolnulls[7];

Datumvalues[8];

boolnulls[8];

}MtmGetNodeStateCtx;

usrfctx->values[4]=Int64GetDatum(lag);

usrfctx->nulls[4]=lag<0;

usrfctx->values[5]=Int64GetDatum(Mtm->transCount ?Mtm->nodes[usrfctx->nodeId-1].transDelay/Mtm->transCount :0);

usrfctx->values[6]=TimestampTzGetDatum(time_t_to_timestamptz(Mtm->nodes[usrfctx->nodeId-1].lastStatusChangeTime));

p=strchr(usrfctx->connStrPtr,',');

if (p!=NULL) {

*p++='\0';

}

usrfctx->values[6]=CStringGetTextDatum(usrfctx->connStrPtr);

usrfctx->values[7]=CStringGetTextDatum(usrfctx->connStrPtr);

usrfctx->connStrPtr=p;

usrfctx->nodeId+=1;

{

MtmConnectionInfocon;

time_ttransDelay;

time_tlastStatusChangeTime;

csn_toldestSnapshot;/* Oldest snapshot used by active transactions at this node */

}MtmNodeInfo;

MtmTransState**transListTail;/* Tail of L1 list of all finished transactionds, used to append new elements.

uint64transCount;/* Counter of transactions perfromed by this node */

time_tnodeTransDelay[MAX_NODES];/* Time of waiting transaction acknowledgment from node */

BgwPoolpool;/* Pool of background workers for applying logical replication patches */

MtmNodeInfonodes[1];/* [MtmNodes]: per-node data */

}MtmState;

externintMtmConnectTimeout;

externintMtmReconnectAttempts;

externintMtmKeepaliveTimeout;

externintMtmNodeDisableDelay;

externHTAB*MtmXid2State;

externMtmConnectionInfo*MtmConnections;

diag("starting node 2");

$nodes[2]->start;

diag("sleeping30");

sleep(30);# XXX: here we can poll

diag("sleeping10");

sleep(10);# XXX: here we can poll

diag("inserting 3");

$nodes[0]->psql('postgres',"insert into t values(3, 30);");

diag("selecting");