p_suffrage.go

package launch

import (
	"context"
	"time"

	"github.com/ProtoconNet/mitum2/base"
	"github.com/ProtoconNet/mitum2/isaac"
	isaacdatabase "github.com/ProtoconNet/mitum2/isaac/database"
	isaacnetwork "github.com/ProtoconNet/mitum2/isaac/network"
	isaacstates "github.com/ProtoconNet/mitum2/isaac/states"
	"github.com/ProtoconNet/mitum2/network/quicmemberlist"
	"github.com/ProtoconNet/mitum2/network/quicstream"
	"github.com/ProtoconNet/mitum2/util"
	"github.com/ProtoconNet/mitum2/util/encoder"
	"github.com/ProtoconNet/mitum2/util/hint"
	"github.com/ProtoconNet/mitum2/util/logging"
	"github.com/ProtoconNet/mitum2/util/ps"
	"github.com/pkg/errors"
)

var (
	PNameSuffrageCandidateLimiterSet      = ps.Name("suffrage-candidate-limiter-set")
	PNamePatchLastConsensusNodesWatcher   = ps.Name("patch-last-consensus-nodes-watcher")
	PNameLastConsensusNodesWatcher        = ps.Name("last-consensus-nodes-watcher")
	PNameStartLastConsensusNodesWatcher   = ps.Name("start-last-consensus-nodes-watcher")
	PNameNodeInConsensusNodesFunc         = ps.Name("node-in-consensus-nodes-func")
	SuffrageCandidateLimiterSetContextKey = util.ContextKey("suffrage-candidate-limiter-set")
	LastConsensusNodesWatcherContextKey   = util.ContextKey("last-consensus-nodes-watcher")
	NodeInConsensusNodesFuncContextKey    = util.ContextKey("node-in-consensus-nodes-func")
	SuffragePoolContextKey                = util.ContextKey("suffrage-pool")
)

func PSuffrageCandidateLimiterSet(pctx context.Context) (context.Context, error) {
	e := util.StringErrorFunc("failed to prepare SuffrageCandidateLimiterSet")

	var db isaac.Database
	if err := util.LoadFromContextOK(pctx, CenterDatabaseContextKey, &db); err != nil {
		return pctx, e(err, "")
	}

	set := hint.NewCompatibleSet()

	if err := set.Add(
		isaac.FixedSuffrageCandidateLimiterRuleHint,
		base.SuffrageCandidateLimiterFunc(FixedSuffrageCandidateLimiterFunc()),
	); err != nil {
		return pctx, e(err, "")
	}

	if err := set.Add(
		isaac.MajoritySuffrageCandidateLimiterRuleHint,
		base.SuffrageCandidateLimiterFunc(MajoritySuffrageCandidateLimiterFunc(db)),
	); err != nil {
		return pctx, e(err, "")
	}

	return context.WithValue(pctx, SuffrageCandidateLimiterSetContextKey, set), nil
}

func PLastConsensusNodesWatcher(pctx context.Context) (context.Context, error) {
	var log *logging.Logging
	var local base.LocalNode
	var params *isaac.LocalParams
	var db isaac.Database

	if err := util.LoadFromContextOK(pctx,
		LoggingContextKey, &log,
		LocalContextKey, &local,
		LocalParamsContextKey, &params,
		CenterDatabaseContextKey, &db,
	); err != nil {
		return pctx, err
	}

	getLastSuffrageProoff, err := GetLastSuffrageProofFunc(pctx)
	if err != nil {
		return pctx, err
	}

	getSuffrageProofFromRemotef, err := GetSuffrageProofFromRemoteFunc(pctx)
	if err != nil {
		return pctx, err
	}

	getLastSuffrageCandidatef, err := GetLastSuffrageCandidateFunc(pctx)
	if err != nil {
		return pctx, err
	}

	builder := isaac.NewSuffrageStateBuilder(
		params.NetworkID(),
		getLastSuffrageProoff,
		getSuffrageProofFromRemotef,
		getLastSuffrageCandidatef,
	)

	watcher, err := isaac.NewLastConsensusNodesWatcher(
		func() (base.Height, base.SuffrageProof, base.State, bool, error) {
			var lastheight base.Height

			switch m, found, err := db.LastBlockMap(); { //nolint:govet //...
			case err != nil:
				return lastheight, nil, nil, false, err
			case !found:
				return lastheight, nil, nil, false, nil
			default:
				lastheight = m.Manifest().Height()
			}

			proof, found, err := db.LastSuffrageProof() //nolint:govet //...
			if err != nil {
				return lastheight, nil, nil, false, err
			}

			st, _, err := db.State(isaac.SuffrageCandidateStateKey)
			if err != nil {
				return lastheight, nil, nil, false, err
			}

			return lastheight, proof, st, found, nil
		},
		builder.Build,
		nil,
		params.WaitPreparingINITBallot(),
	)
	if err != nil {
		return pctx, err
	}

	_ = watcher.SetLogging(log)

	pctx = context.WithValue(pctx, //revive:disable-line:modifies-parameter
		LastConsensusNodesWatcherContextKey, watcher)

	sp := NewSuffragePool(
		func(height base.Height) (base.Suffrage, bool, error) {
			return isaac.GetSuffrageFromDatabase(db, height)
		},
		func() (base.Height, base.Suffrage, bool, error) {
			switch proof, _, err := watcher.Last(); {
			case err != nil, proof == nil:
				return base.NilHeight, nil, false, err
			default:
				suf, err := proof.Suffrage()
				if err != nil {
					return base.NilHeight, nil, false, err
				}

				return proof.Map().Manifest().Height(), suf, true, nil
			}
		},
	)

	return context.WithValue(pctx, SuffragePoolContextKey, sp), nil
}

func PPatchLastConsensusNodesWatcher(pctx context.Context) (context.Context, error) {
	var log *logging.Logging
	var local base.LocalNode
	var params *isaac.LocalParams
	var db isaac.Database
	var watcher *isaac.LastConsensusNodesWatcher
	var states *isaacstates.States
	var long *LongRunningMemberlistJoin
	var mlist *quicmemberlist.Memberlist
	var syncSourcePool *isaac.SyncSourcePool
	var ballotbox *isaacstates.Ballotbox

	if err := util.LoadFromContextOK(pctx,
		LoggingContextKey, &log,
		LocalContextKey, &local,
		LocalParamsContextKey, &params,
		CenterDatabaseContextKey, &db,
		LastConsensusNodesWatcherContextKey, &watcher,
		StatesContextKey, &states,
		LongRunningMemberlistJoinContextKey, &long,
		MemberlistContextKey, &mlist,
		SyncSourcePoolContextKey, &syncSourcePool,
		BallotboxContextKey, &ballotbox,
	); err != nil {
		return pctx, err
	}

	watcher.SetWhenUpdated(func(_ context.Context, previous, updated base.SuffrageProof, candidatesst base.State) {
		if updated == nil {
			ballotbox.Count(params.Threshold())

			log.Log().Debug().Msg("last height of suffrage updated; ballotbox.Count()")

			return
		}

		// NOTE remove withdraw nodes from SyncSourcePool
		if err := removeWithdrawsFromSyncSourcePoolByWatcher(previous, updated, syncSourcePool, log); err != nil {
			log.Log().Error().Err(err).Msg("failed to remove withdraws from sync source pool")
		}

		if err := checkLocalIsInConsensusNodesByWatcher(
			updated, candidatesst, local, mlist, long, states, log,
		); err != nil {
			log.Log().Error().Err(err).Msg("failed to check local is in consensus nodes")
		}
	})

	return pctx, nil
}

func PNodeInConsensusNodesFunc(pctx context.Context) (context.Context, error) {
	e := util.StringErrorFunc("failed NodeInConsensusNodesFunc")

	var db isaac.Database
	var sp *SuffragePool

	if err := util.LoadFromContextOK(
		pctx,
		CenterDatabaseContextKey, &db,
		SuffragePoolContextKey, &sp,
	); err != nil {
		return pctx, e(err, "")
	}

	return context.WithValue(pctx, NodeInConsensusNodesFuncContextKey, nodeInConsensusNodesFunc(db, sp.Height)), nil
}

func getCandidatesFunc(
	db isaac.Database,
) func(height base.Height) (
	[]base.SuffrageCandidateStateValue, []base.SuffrageCandidateStateValue, error,
) {
	var prevcandidateslocked [2]interface{}

	lastcandidateslocked := util.EmptyLocked([2]interface{}{})

	return func(height base.Height) (
		[]base.SuffrageCandidateStateValue, []base.SuffrageCandidateStateValue, error,
	) {
		var prevcandidates []base.SuffrageCandidateStateValue
		var lastcandidates []base.SuffrageCandidateStateValue
		var cerr error

		_, _ = lastcandidateslocked.Set(func(i [2]interface{}, isempty bool) (v [2]interface{}, _ error) {
			var lastheight base.Height
			var last []base.SuffrageCandidateStateValue
			var prev []base.SuffrageCandidateStateValue

			if !isempty {
				lastheight = i[0].(base.Height) //nolint:forcetypeassert //...
				last = isaac.FilterCandidates(  //nolint:forcetypeassert //...
					height, i[1].([]base.SuffrageCandidateStateValue))
			}

			stheight, c, err := isaac.LastCandidatesFromState(height, db.State)
			if err != nil {
				cerr = err

				return v, err
			}

			if j := prevcandidateslocked[1]; j != nil {
				prev = isaac.FilterCandidates( //nolint:forcetypeassert //...
					height-1, j.([]base.SuffrageCandidateStateValue))
			}

			if stheight == lastheight {
				prevcandidates = prev
				lastcandidates = last

				return v, errors.Errorf("stop")
			}

			prevcandidates = last
			lastcandidates = c

			prevcandidateslocked = [2]interface{}{lastheight, last}

			return [2]interface{}{stheight, c}, nil
		})

		if cerr != nil {
			return nil, nil, cerr
		}

		return prevcandidates, lastcandidates, nil
	}
}

func nodeInConsensusNodesFunc(
	db isaac.Database,
	getSuffragef isaac.GetSuffrageByBlockHeight,
) func(node base.Node, height base.Height) (base.Suffrage, bool, error) {
	getCandidatesf := getCandidatesFunc(db)

	return func(node base.Node, height base.Height) (base.Suffrage, bool, error) {
		suf, found, err := getSuffragef(height)

		switch {
		case err != nil:
			return nil, false, err
		case !found:
			return nil, false, nil
		case suf.ExistsPublickey(node.Address(), node.Publickey()):
			return suf, true, nil
		}

		prev, last, err := getCandidatesf(height)
		if err != nil {
			return nil, false, err
		}

		if isaac.InCandidates(node, last) {
			return suf, true, nil
		}

		if isaac.InCandidates(node, prev) {
			return suf, true, nil
		}

		return suf, false, nil
	}
}

func PSuffrageVoting(pctx context.Context) (context.Context, error) {
	var log *logging.Logging
	var local base.LocalNode
	var enc encoder.Encoder
	var db isaac.Database
	var pool *isaacdatabase.TempPool
	var memberlist *quicmemberlist.Memberlist
	var ballotbox *isaacstates.Ballotbox
	var cb *isaacnetwork.CallbackBroadcaster
	var sp *SuffragePool

	if err := util.LoadFromContextOK(pctx,
		LoggingContextKey, &log,
		LocalContextKey, &local,
		EncoderContextKey, &enc,
		CenterDatabaseContextKey, &db,
		PoolDatabaseContextKey, &pool,
		MemberlistContextKey, &memberlist,
		BallotboxContextKey, &ballotbox,
		CallbackBroadcasterContextKey, &cb,
		SuffragePoolContextKey, &sp,
	); err != nil {
		return pctx, err
	}

	sv := isaac.NewSuffrageVoting(
		local.Address(),
		pool,
		db.ExistsInStateOperation,
		func(op base.SuffrageWithdrawOperation) error {
			switch b, err := util.MarshalJSON(op); {
			case err != nil:

				return errors.WithMessage(err, "failed to marshal SuffrageWithdrawOperation")
			default:
				go func() {
					ticker := time.NewTicker(time.Second * 2)
					defer ticker.Stop()

					var n int
					for range ticker.C {
						if err := cb.Broadcast(util.UUID().String(), b, nil); err != nil {
							log.Log().Error().Err(err).
								Interface("operation", op).
								Msg("failed to broadcast SuffrageWithdrawOperation")
						}

						n++

						if n > 3 { //nolint:gomnd //...
							break
						}
					}
				}()
			}

			return nil
		},
	)

	ballotbox.SetSuffrageVoteFunc(func(op base.SuffrageWithdrawOperation) error {
		_, err := sv.Vote(op)

		return err
	})

	pctx = context.WithValue(pctx, SuffrageVotingVoteFuncContextKey, //revive:disable-line:modifies-parameter
		isaac.SuffrageVoteFunc(func(op base.SuffrageWithdrawOperation) (bool, error) {
			var height base.Height

			switch m, found, err := db.LastBlockMap(); {
			case err != nil:
				return false, err
			case !found:
				return false, nil
			default:
				height = m.Manifest().Height()
			}

			var suf base.Suffrage

			switch i, found, err := sp.Height(height); {
			case err != nil:
				return false, err
			case !found:
				return false, nil
			default:
				suf = i
			}

			policy := db.LastNetworkPolicy()

			if err := isaac.IsValidWithdrawWithSuffrageLifespan(
				height, op, suf, policy.SuffrageWithdrawLifespan(),
			); err != nil {
				return false, err
			}

			return sv.Vote(op)
		}),
	)

	return context.WithValue(pctx, SuffrageVotingContextKey, sv), nil
}

func FixedSuffrageCandidateLimiterFunc() func(
	base.SuffrageCandidateLimiterRule,
) (base.SuffrageCandidateLimiter, error) {
	return func(rule base.SuffrageCandidateLimiterRule) (base.SuffrageCandidateLimiter, error) {
		i, ok := rule.(isaac.FixedSuffrageCandidateLimiterRule)
		if !ok {
			return nil, errors.Errorf("expected FixedSuffrageCandidateLimiterRule, not %T", rule)
		}

		return isaac.NewFixedSuffrageCandidateLimiter(i), nil
	}
}

func MajoritySuffrageCandidateLimiterFunc(
	db isaac.Database,
) func(base.SuffrageCandidateLimiterRule) (base.SuffrageCandidateLimiter, error) {
	return func(rule base.SuffrageCandidateLimiterRule) (base.SuffrageCandidateLimiter, error) {
		i, ok := rule.(isaac.MajoritySuffrageCandidateLimiterRule)
		if !ok {
			return nil, errors.Errorf("expected MajoritySuffrageCandidateLimiterRule, not %T", rule)
		}

		proof, found, err := db.LastSuffrageProof()

		switch {
		case err != nil:
			return nil, errors.WithMessagef(err, "failed to get last suffrage for MajoritySuffrageCandidateLimiter")
		case !found:
			return nil, errors.Errorf("last suffrage not found for MajoritySuffrageCandidateLimiter")
		}

		suf, err := proof.Suffrage()
		if err != nil {
			return nil, errors.WithMessagef(err, "failed to get suffrage for MajoritySuffrageCandidateLimiter")
		}

		return isaac.NewMajoritySuffrageCandidateLimiter(
			i,
			func() (uint64, error) {
				return uint64(suf.Len()), nil
			},
		), nil
	}
}

func GetLastSuffrageProofFunc(pctx context.Context) (isaac.GetLastSuffrageProofFromRemoteFunc, error) {
	var params base.LocalParams
	var client *isaacnetwork.QuicstreamClient
	var syncSourcePool *isaac.SyncSourcePool

	if err := util.LoadFromContextOK(pctx,
		QuicstreamClientContextKey, &client,
		LocalParamsContextKey, &params,
		SyncSourcePoolContextKey, &syncSourcePool,
	); err != nil {
		return nil, err
	}

	lastl := util.EmptyLocked((util.Hash)(nil))

	f := func(ctx context.Context, ci quicstream.UDPConnInfo) (base.Height, base.SuffrageProof, bool, error) {
		cctx, cancel := context.WithTimeout(ctx, time.Second*2) //nolint:gomnd //...
		defer cancel()

		last, _ := lastl.Value()

		switch lastheight, proof, updated, err := client.LastSuffrageProof(cctx, ci, last); {
		case err != nil, !updated:
			return lastheight, proof, updated, nil
		default:
			if err := proof.IsValid(params.NetworkID()); err != nil {
				return lastheight, nil, updated, err
			}

			_ = lastl.SetValue(proof.Map().Manifest().Suffrage())

			return lastheight, proof, updated, nil
		}
	}

	return func(ctx context.Context) (lastheight base.Height, proof base.SuffrageProof, found bool, _ error) {
		ml := util.EmptyLocked((base.SuffrageProof)(nil))

		numnodes := 3 // NOTE choose top 3 sync nodes

		if err := isaac.DistributeWorkerWithSyncSourcePool(
			ctx,
			syncSourcePool,
			numnodes,
			uint64(numnodes),
			nil,
			func(ctx context.Context, i, _ uint64, nci isaac.NodeConnInfo) error {
				ci, err := nci.UDPConnInfo()
				if err != nil {
					return err
				}

				h, proof, updated, err := f(ctx, ci)
				if err != nil {
					return err
				}

				_, _ = ml.Set(func(v base.SuffrageProof, _ bool) (base.SuffrageProof, error) {
					lastheight = h

					if !updated {
						return nil, util.ErrLockedSetIgnore.Errorf("not updated")
					}

					switch {
					case v == nil,
						proof.Map().Manifest().Height() > v.Map().Manifest().Height():

						return proof, nil
					default:
						return nil, util.ErrLockedSetIgnore.Errorf("old SuffrageProof")
					}
				})

				return nil
			},
		); err != nil {
			if errors.Is(err, isaac.ErrEmptySyncSources) {
				return lastheight, nil, false, nil
			}

			return lastheight, nil, false, err
		}

		switch v, _ := ml.Value(); {
		case v == nil:
			return lastheight, nil, false, nil
		default:
			return lastheight, v, true, nil
		}
	}, nil
}

func GetSuffrageProofFromRemoteFunc(pctx context.Context) ( //revive:disable-line:cognitive-complexity
	isaac.GetSuffrageProofFromRemoteFunc, error,
) {
	var params base.LocalParams
	var client *isaacnetwork.QuicstreamClient
	var syncSourcePool *isaac.SyncSourcePool

	if err := util.LoadFromContextOK(pctx,
		QuicstreamClientContextKey, &client,
		LocalParamsContextKey, &params,
		SyncSourcePoolContextKey, &syncSourcePool,
	); err != nil {
		return nil, err
	}

	return func(ctx context.Context, suffrageheight base.Height) (proof base.SuffrageProof, found bool, _ error) {
		err := util.Retry(
			ctx,
			func() (bool, error) {
				numnodes := 3 // NOTE choose top 3 sync nodes
				result := util.EmptyLocked([2]interface{}{})

				_ = isaac.ErrGroupWorkerWithSyncSourcePool(
					ctx,
					syncSourcePool,
					numnodes,
					uint64(numnodes),
					func(ctx context.Context, i, _ uint64, nci isaac.NodeConnInfo) error {
						ci, err := nci.UDPConnInfo()
						if err != nil {
							return err
						}

						cctx, cancel := context.WithTimeout(ctx, time.Second*2) //nolint:gomnd //...
						defer cancel()

						switch a, b, err := client.SuffrageProof(cctx, ci, suffrageheight); {
						case err != nil:
							if quicstream.IsNetworkError(err) {
								return err
							}

							return nil
						case !b:
							return nil
						default:
							if err := a.IsValid(params.NetworkID()); err != nil {
								return nil
							}

							_, _ = result.Set(func(_ [2]interface{}, isempty bool) ([2]interface{}, error) {
								if !isempty {
									return [2]interface{}{}, util.ErrLockedSetIgnore.Errorf("already set")
								}

								return [2]interface{}{a, b}, nil
							})

							return errors.Errorf("stop")
						}
					},
				)

				i, isempty := result.Value()
				if isempty {
					return true, nil
				}

				proof, found = i[0].(base.SuffrageProof), i[1].(bool) //nolint:forcetypeassert //...

				return false, nil
			},
			-1,
			time.Second,
		)

		return proof, found, err
	}, nil
}

func GetLastSuffrageCandidateFunc(pctx context.Context) (isaac.GetLastSuffrageCandidateStateRemoteFunc, error) {
	var client *isaacnetwork.QuicstreamClient
	var syncSourcePool *isaac.SyncSourcePool

	if err := util.LoadFromContextOK(pctx,
		QuicstreamClientContextKey, &client,
		SyncSourcePoolContextKey, &syncSourcePool,
	); err != nil {
		return nil, err
	}

	lastl := util.EmptyLocked((util.Hash)(nil))

	f := func(ctx context.Context, ci quicstream.UDPConnInfo) (base.State, bool, error) {
		last, _ := lastl.Value()

		cctx, cancel := context.WithTimeout(ctx, time.Second*2) //nolint:gomnd //...
		defer cancel()

		st, found, err := client.State(cctx, ci, isaac.SuffrageCandidateStateKey, last)

		switch {
		case err != nil, !found, st == nil:
			return st, found, err
		default:
			if err := st.IsValid(nil); err != nil {
				return nil, false, err
			}

			_ = lastl.SetValue(st.Hash())

			return st, true, nil
		}
	}

	return func(ctx context.Context) (base.State, bool, error) {
		ml := util.EmptyLocked((base.State)(nil))

		numnodes := 3 // NOTE choose top 3 sync nodes

		if err := isaac.DistributeWorkerWithSyncSourcePool(
			ctx,
			syncSourcePool,
			numnodes,
			uint64(numnodes),
			nil,
			func(ctx context.Context, i, _ uint64, nci isaac.NodeConnInfo) error {
				ci, err := nci.UDPConnInfo()
				if err != nil {
					return err
				}

				st, found, err := f(ctx, ci)
				switch {
				case err != nil, !found, st == nil:
					return err
				}

				_, err = ml.Set(func(v base.State, _ bool) (base.State, error) {
					switch {
					case v == nil, st.Height() > v.Height():
						return st, nil
					default:
						return nil, util.ErrLockedSetIgnore.Errorf("old SuffrageProof")
					}
				})

				return err
			},
		); err != nil {
			return nil, false, err
		}

		switch v, _ := ml.Value(); {
		case v == nil:
			return nil, false, nil
		default:
			return v, true, nil
		}
	}, nil
}

func NewSuffrageCandidateLimiterFunc(pctx context.Context) ( //revive:disable-line:cognitive-complexity
	func(base.Height, base.GetStateFunc) (base.OperationProcessorProcessFunc, error),
	error,
) {
	var db isaac.Database
	var limiterset *hint.CompatibleSet

	if err := util.LoadFromContextOK(pctx,
		CenterDatabaseContextKey, &db,
		SuffrageCandidateLimiterSetContextKey, &limiterset,
	); err != nil {
		return nil, err
	}

	return func(height base.Height, getStateFunc base.GetStateFunc) (base.OperationProcessorProcessFunc, error) {
		e := util.StringErrorFunc("failed to get SuffrageCandidateLimiterFunc")

		policy := db.LastNetworkPolicy()
		if policy == nil {
			return nil, e(nil, "empty network policy")
		}

		var suf base.Suffrage

		switch proof, found, err := db.LastSuffrageProof(); {
		case err != nil:
			return nil, e(err, "failed to get last suffrage")
		case !found:
			return nil, e(nil, "last suffrage not found")
		default:
			i, err := proof.Suffrage()
			if err != nil {
				return nil, e(err, "failed to get suffrage")
			}

			suf = i
		}

		var existings uint64

		switch _, i, err := isaac.LastCandidatesFromState(height, getStateFunc); {
		case err != nil:
			return nil, e(err, "")
		default:
			existings = uint64(len(i))
		}

		rule := policy.SuffrageCandidateLimiterRule()

		var limit uint64

		switch i := limiterset.Find(rule.Hint()); {
		case i == nil:
			return nil, e(nil, "unknown limiter rule, %q", rule.Hint())
		default:
			f, ok := i.(base.SuffrageCandidateLimiterFunc)
			if !ok {
				return nil, e(nil, "expected SuffrageCandidateLimiterFunc, not %T", i)
			}

			limiter, err := f(rule)
			if err != nil {
				return nil, e(err, "")
			}

			j, err := limiter()
			if err != nil {
				return nil, e(err, "")
			}

			limit = j
		}

		switch {
		case existings >= policy.MaxSuffrageSize():
			return func(
				_ context.Context, op base.Operation, _ base.GetStateFunc,
			) (base.OperationProcessReasonError, error) {
				return base.NewBaseOperationProcessReasonError("reached limit, %d", policy.MaxSuffrageSize()), nil
			}, nil
		case limit > policy.MaxSuffrageSize()-uint64(suf.Len()):
			limit = policy.MaxSuffrageSize() - uint64(suf.Len())
		}

		if limit < 1 {
			return func(
				_ context.Context, op base.Operation, _ base.GetStateFunc,
			) (base.OperationProcessReasonError, error) {
				return base.NewBaseOperationProcessReasonError("reached limit, %d", limit), nil
			}, nil
		}

		var counted uint64

		return func(
			_ context.Context, op base.Operation, _ base.GetStateFunc,
		) (base.OperationProcessReasonError, error) {
			if counted >= limit {
				return base.NewBaseOperationProcessReasonError("reached limit, %d", limit), nil
			}

			counted++

			return nil, nil
		}, nil
	}, nil
}

func PStartLastConsensusNodesWatcher(pctx context.Context) (context.Context, error) {
	var watcher *isaac.LastConsensusNodesWatcher
	if err := util.LoadFromContextOK(pctx, LastConsensusNodesWatcherContextKey, &watcher); err != nil {
		return pctx, err
	}

	return pctx, watcher.Start(context.Background())
}

func PCloseLastConsensusNodesWatcher(pctx context.Context) (context.Context, error) {
	var watcher *isaac.LastConsensusNodesWatcher
	if err := util.LoadFromContext(pctx, LastConsensusNodesWatcherContextKey, &watcher); err != nil {
		return pctx, err
	}

	if watcher != nil {
		if err := watcher.Stop(); err != nil && !errors.Is(err, util.ErrDaemonAlreadyStopped) {
			return pctx, err
		}
	}

	return pctx, nil
}

func checkLocalIsInConsensusNodesByWatcher(
	updated base.SuffrageProof,
	candidatesst base.State,
	local base.LocalNode,
	mlist *quicmemberlist.Memberlist,
	long *LongRunningMemberlistJoin,
	states *isaacstates.States,
	log *logging.Logging,
) error {
	suf, err := updated.Suffrage()
	if err != nil {
		return err
	}

	var isinsuffrage bool

	switch {
	case suf.Exists(local.Address()):
		isinsuffrage = true
	case candidatesst == nil:
	case isaac.InCandidates(local,
		candidatesst.Value().(base.SuffrageCandidatesStateValue).Nodes()): //nolint:forcetypeassert //...
		isinsuffrage = true
	}

	switch {
	case isinsuffrage:
		// NOTE if local is in consensus nodes, try to join
		if !mlist.IsJoined() {
			log.Log().Debug().
				Msg("watcher updated suffrage and local is in consensus nodes, but not yet joined; tries to join")

			go func() {
				_ = long.Join()
			}()
		}
	default:
		log.Log().Debug().
			Interface("height", updated.Map().Manifest().Height()).
			Msg("local is not in consensus nodes; moves to syncing")

		// NOTE if local is out of consensus nodes, moves to syncing state
		_ = states.MoveState(isaacstates.NewSyncingSwitchContextWithOK(
			updated.Map().Manifest().Height(),
			func(current isaacstates.StateType) bool {
				switch current {
				case isaacstates.StateJoining, isaacstates.StateConsensus:
					return true
				default:
					return false
				}
			},
		))
	}

	return nil
}

func removeWithdrawsFromSyncSourcePoolByWatcher(
	previous, updated base.SuffrageProof,
	syncSourcePool *isaac.SyncSourcePool,
	log *logging.Logging,
) error {
	if previous == nil {
		return nil
	}

	suf, err := updated.Suffrage()
	if err != nil {
		return err
	}

	previoussuf, err := previous.Suffrage()
	if err != nil {
		return err
	}

	withdraws := util.Filter2Slices(previoussuf.Nodes(), suf.Nodes(), func(p, n base.Node) bool {
		return p.Address().Equal(n.Address())
	})

	if len(withdraws) < 1 {
		return nil
	}

	withdrawnodes := make([]base.Address, len(withdraws))
	for i := range withdraws {
		withdrawnodes[i] = withdraws[i].Address()
	}

	removed := syncSourcePool.RemoveNonFixedNode(withdrawnodes...)

	log.Log().Debug().
		Interface("withdraws", withdrawnodes).
		Bool("removed", removed).
		Msg("withdraw nodes removed from sync source pool")

	return nil
}