func (s *Service) startEpochTicker(ctx context.Context, waitedForGenesis bool) error {
	runtimeFunc := func(ctx context.Context, data interface{}) (time.Time, error) {
		return s.chainTimeService.StartOfEpoch(s.chainTimeService.CurrentEpoch() + 1), nil
	}
	data := &epochTickerData{
		latestEpochRan: -1,
		atGenesis:      waitedForGenesis,
	}
	if err := s.scheduler.SchedulePeriodicJob(ctx,
		"Epoch ticker",
		runtimeFunc,
		data,
		s.epochTicker,
		data,
	); err != nil {
		return errors.Wrap(err, "Failed to schedule epoch ticker")
	}
	return nil
}
startEpochTicker starts a ticker that ticks at the beginning of each epoch.
package standard
import (
	"bytes"
	"context"
	"fmt"
	"sync"
	"time"
	eth2client "github.com/attestantio/go-eth2-client"
	"github.com/attestantio/go-eth2-client/spec/phase0"
	"github.com/attestantio/vouch/services/accountmanager"
	"github.com/attestantio/vouch/services/attestationaggregator"
	"github.com/attestantio/vouch/services/attester"
	"github.com/attestantio/vouch/services/beaconblockproposer"
	"github.com/attestantio/vouch/services/beaconcommitteesubscriber"
	"github.com/attestantio/vouch/services/chaintime"
	"github.com/attestantio/vouch/services/metrics"
	"github.com/attestantio/vouch/services/scheduler"
	"github.com/attestantio/vouch/services/synccommitteeaggregator"
	"github.com/attestantio/vouch/services/synccommitteemessenger"
	"github.com/attestantio/vouch/services/synccommitteesubscriber"
	"github.com/pkg/errors"
	"github.com/rs/zerolog"
	zerologger "github.com/rs/zerolog/log"
	e2wtypes "github.com/wealdtech/go-eth2-wallet-types/v2"
)
type Service struct {
	monitor                      metrics.ControllerMonitor
	slotDuration                 time.Duration
	slotsPerEpoch                uint64
	epochsPerSyncCommitteePeriod uint64
	chainTimeService             chaintime.Service
	proposerDutiesProvider       eth2client.ProposerDutiesProvider
	attesterDutiesProvider       eth2client.AttesterDutiesProvider
	syncCommitteeDutiesProvider  eth2client.SyncCommitteeDutiesProvider
	validatingAccountsProvider   accountmanager.ValidatingAccountsProvider
	scheduler                    scheduler.Service
	attester                     attester.Service
	syncCommitteeMessenger       synccommitteemessenger.Service
	syncCommitteeAggregator      synccommitteeaggregator.Service
	syncCommitteesSubscriber     synccommitteesubscriber.Service
	beaconBlockProposer          beaconblockproposer.Service
	attestationAggregator        attestationaggregator.Service
	beaconCommitteeSubscriber    beaconcommitteesubscriber.Service
	activeValidators             int
	subscriptionInfos            map[phase0.Epoch]map[phase0.Slot]map[phase0.CommitteeIndex]*beaconcommitteesubscriber.Subscription
	subscriptionInfosMutex       sync.Mutex
	accountsRefresher            accountmanager.Refresher
	maxSyncCommitteeMessageDelay time.Duration
	reorgs                       bool
	handlingAltair  bool
	altairForkEpoch phase0.Epoch
	lastBlockRoot             phase0.Root
	lastBlockEpoch            phase0.Epoch
	currentDutyDependentRoot  phase0.Root
	previousDutyDependentRoot phase0.Root
}
var log zerolog.Logger
func New(ctx context.Context, params ...Parameter) (*Service, error) {
	parameters, err := parseAndCheckParameters(params...)
	if err != nil {
		return nil, errors.Wrap(err, "problem with parameters")
	}
	log = zerologger.With().Str("service", "controller").Str("impl", "standard").Logger()
	if parameters.logLevel != log.GetLevel() {
		log = log.Level(parameters.logLevel)
	}
	spec, err := parameters.specProvider.Spec(ctx)
	if err != nil {
		return nil, errors.Wrap(err, "failed to obtain spec")
	}
	tmp, exists := spec["SECONDS_PER_SLOT"]
	if !exists {
		return nil, errors.New("SECONDS_PER_SLOT not found in spec")
	}
	slotDuration, ok := tmp.(time.Duration)
	if !ok {
		return nil, errors.New("SECONDS_PER_SLOT of unexpected type")
	}
	tmp, exists = spec["SLOTS_PER_EPOCH"]
	if !exists {
		return nil, errors.New("SLOTS_PER_EPOCH not found in spec")
	}
	slotsPerEpoch, ok := tmp.(uint64)
	if !ok {
		return nil, errors.New("SLOTS_PER_EPOCH of unexpected type")
	}
	var epochsPerSyncCommitteePeriod uint64
	if tmp, exists := spec["EPOCHS_PER_SYNC_COMMITTEE_PERIOD"]; exists {
		tmp2, ok := tmp.(uint64)
		if !ok {
			return nil, errors.New("EPOCHS_PER_SYNC_COMMITTEE_PERIOD of unexpected type")
		}
		epochsPerSyncCommitteePeriod = tmp2
	}
	handlingAltair := parameters.syncCommitteeAggregator != nil && epochsPerSyncCommitteePeriod != 0
	if !handlingAltair {
		log.Trace().Msg("Not handling Altair")
	}
	var altairForkEpoch phase0.Epoch
	if handlingAltair {
		altairForkEpoch, err = fetchAltairForkEpoch(ctx, parameters.forkScheduleProvider)
		if err != nil {
			handlingAltair = false
		} else {
			log.Trace().Uint64("epoch", uint64(altairForkEpoch)).Msg("Obtained Altair fork epoch")
		}
	}
	s := &Service{
		monitor:                      parameters.monitor,
		slotDuration:                 slotDuration,
		slotsPerEpoch:                slotsPerEpoch,
		epochsPerSyncCommitteePeriod: epochsPerSyncCommitteePeriod,
		chainTimeService:             parameters.chainTimeService,
		proposerDutiesProvider:       parameters.proposerDutiesProvider,
		attesterDutiesProvider:       parameters.attesterDutiesProvider,
		syncCommitteeDutiesProvider:  parameters.syncCommitteeDutiesProvider,
		syncCommitteesSubscriber:     parameters.syncCommitteesSubscriber,
		validatingAccountsProvider:   parameters.validatingAccountsProvider,
		scheduler:                    parameters.scheduler,
		attester:                     parameters.attester,
		syncCommitteeMessenger:       parameters.syncCommitteeMessenger,
		syncCommitteeAggregator:      parameters.syncCommitteeAggregator,
		beaconBlockProposer:          parameters.beaconBlockProposer,
		attestationAggregator:        parameters.attestationAggregator,
		beaconCommitteeSubscriber:    parameters.beaconCommitteeSubscriber,
		accountsRefresher:            parameters.accountsRefresher,
		maxSyncCommitteeMessageDelay: parameters.maxSyncCommitteeMessageDelay,
		reorgs:                       parameters.reorgs,
		subscriptionInfos:            make(map[phase0.Epoch]map[phase0.Slot]map[phase0.CommitteeIndex]*beaconcommitteesubscriber.Subscription),
		handlingAltair:               handlingAltair,
		altairForkEpoch:              altairForkEpoch,
	}
	if err := parameters.eventsProvider.Events(ctx, []string{"head"}, s.HandleHeadEvent); err != nil {
		return nil, errors.Wrap(err, "failed to add head event handler")
	}
	if err := s.startTickers(ctx); err != nil {
		return nil, errors.Wrap(err, "failed to start controller tickers")
	}
	epoch := s.chainTimeService.CurrentEpoch()
	accounts, validatorIndices, err := s.accountsAndIndicesForEpoch(ctx, epoch)
	if err != nil {
		return nil, errors.Wrap(err, "failed to obtain active validator indices for the current epoch")
	}
	if len(validatorIndices) != s.activeValidators {
		log.Info().Int("old_validators", s.activeValidators).Int("new_validators", len(validatorIndices)).Msg("Change in number of active validators")
		s.activeValidators = len(validatorIndices)
	}
	nextEpochAccounts, nextEpochValidatorIndices, err := s.accountsAndIndicesForEpoch(ctx, epoch+1)
	if err != nil {
		return nil, errors.Wrap(err, "failed to obtain active validator indices for the next epoch")
	}
	go s.scheduleProposals(ctx, epoch, validatorIndices, true  )
	go s.scheduleAttestations(ctx, epoch, validatorIndices, true  )
	if handlingAltair {
		thisSyncCommitteePeriodStartEpoch := s.firstEpochOfSyncPeriod(uint64(epoch) / s.epochsPerSyncCommitteePeriod)
		go s.scheduleSyncCommitteeMessages(ctx, thisSyncCommitteePeriodStartEpoch, validatorIndices, true  )
		nextSyncCommitteePeriodStartEpoch := s.firstEpochOfSyncPeriod(uint64(epoch)/s.epochsPerSyncCommitteePeriod + 1)
		go s.scheduleSyncCommitteeMessages(ctx, nextSyncCommitteePeriodStartEpoch, validatorIndices, true  )
	}
	go s.scheduleAttestations(ctx, epoch+1, nextEpochValidatorIndices, true  )
	go func() {
		subscriptionInfo, err := s.beaconCommitteeSubscriber.Subscribe(ctx, epoch, accounts)
		if err != nil {
			log.Warn().Err(err).Msg("Failed to subscribe to beacon committees")
			return
		}
		s.subscriptionInfosMutex.Lock()
		s.subscriptionInfos[epoch] = subscriptionInfo
		s.subscriptionInfosMutex.Unlock()
	}()
	go func() {
		subscriptionInfo, err := s.beaconCommitteeSubscriber.Subscribe(ctx, epoch+1, nextEpochAccounts)
		if err != nil {
			log.Warn().Err(err).Msg("Failed to subscribe to beacon committees")
			return
		}
		s.subscriptionInfosMutex.Lock()
		s.subscriptionInfos[epoch+1] = subscriptionInfo
		s.subscriptionInfosMutex.Unlock()
	}()
	return s, nil
}
func (s *Service) startTickers(ctx context.Context) error {
	genesisTime := s.chainTimeService.GenesisTime()
	now := time.Now()
	waitedForGenesis := false
	if now.Before(genesisTime) {
		waitedForGenesis = true
		log.Info().Str("genesis", fmt.Sprintf("%v", genesisTime)).Msg("Waiting for genesis")
		time.Sleep(time.Until(genesisTime))
	}
	log.Trace().Msg("Starting epoch tickers")
	if err := s.startEpochTicker(ctx, waitedForGenesis); err != nil {
		return errors.Wrap(err, "failed to start epoch ticker")
	}
	log.Trace().Msg("Starting accounts refresher")
	if err := s.startAccountsRefresher(ctx); err != nil {
		return errors.Wrap(err, "failed to start accounts refresher")
	}
	return nil
}
type epochTickerData struct {
	mutex          sync.Mutex
	latestEpochRan int64
	atGenesis      bool
}