---
# 🔄 Duty Scheduler Refactoring for SSV Project
_Author: Oleg Shmuelov_

## 📌 Introduction
The duty scheduler, a key component in the SSV project, manages validator duties in the Ethereum 2.0 protocol. It's currently composed of two primary components: the Duty Controller and the Duty Fetcher. This document outlines these components, discusses the goals of the proposed refactoring, and suggests a design for the refactored duty scheduler.

## 🧩 Current Components

### 🎛 Duty Controller
The Duty Controller triggers the execution of duties based on a slot ticker. It resides in the `controller.go` file and uses several types and interfaces, including `DutyController` and `DutyExecutor`.

### 📥 Duty Fetcher
The Duty Fetcher retrieves duties either from a cache or directly from the beacon chain if not available in the cache. This component can be found in the `fetcher.go` file, and it relies on types and interfaces like `DutyFetcher`, `validatorsIndicesFetcher`, and `cacheEntry`.

## 🎯 Refactoring Goals
The main objectives for refactoring the duty scheduler are:

1. **🚀 Optimize Duty Scheduling for Upcoming Epochs/Periods**: The refactored scheduler should not only manage current epoch duties but also plan and schedule duties for upcoming epochs and periods. The scheduler must calculate and arrange these duties halfway through each epoch for optimal performance.

2. **🏃‍♂️ Speed up Execution of Attester and Sync Committee Duties**: The scheduler should expedite Attester and Sync Committee duties when a block arrives before 1/3 of the slot time, enhancing duty execution performance.

3. **🔄 Handle Blockchain Reorganizations (Re-orgs)**: In case of sudden changes in the blockchain state, the scheduler should be capable of re-scheduling duties accordingly.

4. **🔮 Proactively Schedule Duties for Upcoming Active Validators**: The scheduler should foresee and schedule duties for validators expected to become active in the beacon chain in the next epoch or sync committee period.

5. **⏱ (Un)/Schedule Duties in Response to Contract Events**: The scheduler should be designed to efficiently manage the scheduling of duties for validators who are registered on the SSV network and already active on Beacon Node. The objective is to ensure that these validators are capable of initiating their duties immediately as the node processes them
(upon a ValidatorAddedEvent, ValidatorRemovedEvent, ClusterReactivatedEvent and ClusterLiquidatedEvent).

6. **📡 Resubscribe to Sync-Committee Each Epoch or Ideally on Beacon Node Restart ???**: Check if needed!
The objective here is to maintain continuous synchronization with the sync-committee subscriptions. The system should automatically resubscribe to the sync-committee at the start of each new epoch. Additionally, to account for any interruptions or inconsistencies that might occur due to a beacon node restart, the system should ideally also resubscribe upon such events. This ensures that the system stays updated and aligned with the latest state of the sync-committee, enabling efficient and accurate execution of sync-committe duties.

7. **⏰ Address Maximum Delay for Duty Start ???**: We should discuss how to manage scenarios where the node starts in the middle of the epoch and may need to execute past duties for which rewards can still be claimed. Could be relevant to ClusterReactivatedEvent only
[Read More](https://eth2book.info/capella/part2/incentives/rewards/#attestation-rewards)

## 📐 Proposed Design

** Add flow from 0 to attest

The refactored Duty Scheduler should include the following improvements:

1. **🚀 Optimize Duty Scheduling for Upcoming Epochs/Periods**:
- Schedule duties for the next epoch/period.
- Arrange duties for the upcoming period halfway through each current epoch.

#### implementation break-up:
- `duties controller`:
- `dutyController contrustor`: add `proposerDutiesMap` & `attesterDutiesMap` to manage the duties for all validators. (we arleady handle the sync committee duties the same way)
- `schedulers`: add schedulers for each duty type, fetch the duties for passed indices & populate the map for relevant slots (could be we fetch the duties somewhere in the middle for the current epoch)
- `duty handler`: add handler for each duty type. Pop the duty from map, send to execution, delete from map
- `warmUpDuties`: preparing duties for all validators when the node starts, including scheduling proposals, attestations and sync committee duties. duties should be scheduled for the next epoch as well
- `handleSlot`: trigger duty handlers on each slot ticker, schedule duties for the next epoch (in the middle of epoch)
** the proposal duties can be fetched only for currennt epoch this why we have to schedule the proposal duties in the beginning of each epoch
- `dutyFetcher`:
- `AttesterDuties`: get duties, log the duties, calculate subscribtions, subscribe to to committee subnets
- `ProposerDuties`: get duties, log the duties
- `SyncCommitteeDuties`: already exists, just add logs
- `GetDuties`: deprecate functionality, including cache

estimation: ~1d

2. **🏃‍♂️ Speed up Execution of Attester and Sync Committee Duties**:

The SSV node subscribes to "head" events from the Beacon Node (BN). We need to address two possible scenarios:

- **Beacon block arrives after 1/3 of the slot time**: The Attester/Sync Committee executor should wait for 1/3 of the slot time before fetching the duty data (Attestation Data for attester/ Beacon Block root for sync committee) from the BN.

- **Beacon block arrives before 1/3 of the slot time**: The duty data should be fetched from the BN immediately. There should be no delay in executing the duty.

To implement this, we could use `sync.Cond` as demonstrated in the code snippet below:

    var beaconBlockArrived bool
    var m sync.Mutex
    var cond *sync.Cond

    // Simulate the arrival of the beacon block
    func beaconBlockArrivalSimulator() {
        time.Sleep(time.Second * 2) // Beacon block arrives after 2 seconds
        m.Lock()
        defer m.Unlock()
        beaconBlockArrived = true
        cond.Broadcast() // Signal that the beacon block has arrived
        fmt.Println("Beacon block has arrived")
    }

    // Simulate the execution of duties
    func dutyExecutionSimulator() {
        m.Lock()
        for !beaconBlockArrived { // If beacon block has not arrived, wait
            cond.Wait()
        }
        m.Unlock()
        // Execute duties here
        fmt.Println("Executing duties")
    }

    func main() {
        cond = sync.NewCond(&m)
        go beaconBlockArrivalSimulator()
        go dutyExecutionSimulator()

        time.Sleep(time.Second * 5) // Wait for all goroutines to finish
    }

In this code example, the beaconBlockArrivalSimulator function simulates the arrival of the beacon block. When the beacon block arrives, it acquires the mutex lock, updates the beaconBlockArrived flag, and then broadcasts a signal to all waiting goroutines using cond.Broadcast().

The dutyExecutionSimulator function is where the duties would be executed. It first acquires the mutex lock and then waits for the beaconBlockArrived flag to be true. The cond.Wait() call atomically unlocks the mutex and suspends execution of the goroutine, allowing other goroutines to proceed. When cond.Broadcast() is called by the beaconBlockArrivalSimulator function, cond.Wait() re-acquires the mutex and returns, allowing the rest of the dutyExecutionSimulator function to proceed and execute the duties.

This model allows the immediate execution of duties as soon as the beacon block arrives, without having to wait for one-third of the slot time if the block arrives earlier.

#### Future considerations:
- Round change time synchronization between the nodes

#### implementation break-up:
- `duties controller`:
- `dutyController contrustor`: add `sync.Cond` & `sync.Mutex` to manage the waiting condition
- `HandleHeadEvent`: add condition to release waiting condition if the the slot arrived before 1/3 of the slot duration
- `ExecuteDuty`: move the waitOneThirdOrValidBlock to this component. shoulf be triggered for attester&sync committe roles only
- `waitOneThirdOrValidBlock`: moved from `goClient`. Adjust to work with sync.Cond (release the waiting broadcasted from head event)
- `goClient`:
- `waitOneThirdOrValidBlock`: move to duty controller

estimation: ~1d

3. **🔄 Handle Blockchain Reorganizations (Re-orgs)**:

The scheduler must detect if a reorganization occurs and re-fetch duties if necessary. Possible scenarios include:

- **Previous dependent root changed**
- *Change of epoch*: Ensure the new `PreviousDutyDependentRoot` is the same as the `CurrentDutyDependentRoot`.
- *Existing epoch*: Ensure the new `PreviousDutyDependentRoot` is the same as the old `PreviousDutyDependentRoot`.
- **Current dependent root changed**
- *Existing epoch*: Ensure the new `CurrentDutyDependentRoot` is the same as the old `CurrentDutyDependentRoot`.

Here's a list of actions required on root changes (re-org):

- **Previous dependent root changed**:
- Refresh attester duties for the current epoch.
- **Current dependent root changed**:
- Refresh proposer duties for the current epoch. **dont re-schedule for the slot the re-org occured
- Refresh sync committee duties for the next period.
- Refresh attester duties for the next epoch.

In order to check if the dependent root has changed, the head event handler must always to be aware of: `lastBlockEpoch`, `previousDutyDependentRoot`, `currentDutyDependentRoot`.

#### implementation break-up:
- `duties controller`:
- `dutyController contrustor`: add `lastBlockEpoch` & `previousDutyDependentRoot` & `currentDutyDependentRoot` to manage the re-orgs
- `HandleHeadEvent`: check for reorgs based on conditions described above
- `handleCurrentDependentRootChanged`
- `handlePreviousDependentRootChanged`
- `Refreshers`: add refreshers for each duty type to re-schedule the duties if the re-org occurs

estimation: ~1d

4. **🔮 Proactively Schedule Duties for Upcoming Active Validators**:
In our current approach, we access the `validatorsMap` via the `validator controller ActiveValidatorIndices` to gather the indices of validators for whom we need to schedule duties.

Our proposed refactoring seeks to look ahead beyond just the current epoch/period. We recommend fetching and scheduling duties not only for currently active validators but also for those who are expected to become active in the upcoming epoch/period.

#### implementation break-up:
- `beacon`:
- `ValidatorMetadata`: add `activation_epoch` to Share metadata
- `UpdateValidatorsMetadata`: on update add `activation_epoch`
- `Equals`: adjust to compare with `activation_epoch`

- `validator controller`:
- `ActiveValidatorIndices`: add functionality to get validators from validators map by epoch
(activation_epoch >= epoch)
- `startValidator`: fix current bug to not start inactive validators

estimation: ~4h

5. **⏱ (Un)/Schedule Duties in Response to Contract Events**:
**Problem**: If a new event arrives after the duty scheduling for the upcoming epoch/period, the duties associated with these events might either be skipped or executed unnecessarily.

**Solution**: This issue can be resolved by establishing communication between the duty and validator controllers. This way, whenever a relevant event is received, the validator controller can prompt the duty controller to (un)/schedule the corresponding duties for the validators involved.

- **ValidatorAddedEvent/ClusterReactivatedEvent**: There may be situations where validators get registered or reactivated in the time interval between duty scheduling and actual execution. In these cases, the newly associated duties should be promptly added to the duties map.

- **ValidatorRemovedEvent/ClusterLiquidatedEvent**: On the other hand, validators might be removed by users or liquidated within the same interval. When this happens, the duties linked to these validators need to be rapidly removed from the duties map to preserve the integrity and accuracy of the duty execution process.
**UPD**: on ValidatorRemovedEvent/ClusterLiquidatedEvent we are removing the validator form the map, no need to handle

To implement this, we could adopt the eventBus approach: have the validator controller emit an event that the duty controller listens for. Consequently, the duty controller can handle the event and carry out the necessary operations.

    // Validator controller 
    type ValidatorController struct {
        eventBus *eventbus.EventBus
    }

    func (vc *ValidatorController) HandleClusterLiquidatedEvent(event abiparser.ClusterLiquidatedEvent) {
        // Emit event
        vc.eventBus.Emit(ValidatorLiquidatedEvent{OperatorIds: event.OperatorIds})
    }

    // Duty controller 
    type DutyController struct {
        eventBus *eventbus.EventBus
    }

    func (dc *DutyController) Start() {
        // Subscribe to events
        dc.eventBus.Subscribe(ValidatorLiquidatedEvent{}, dc.HandleValidatorLiquidatedEvent)
    }

    func (dc *DutyController) HandleValidatorLiquidatedEvent(event ValidatorLiquidatedEvent) {
        dc.RemoveDutiesForValidators(event.OperatorIds) 
    }

    // Event 
    type ValidatorLiquidatedEvent struct {
        OperatorIds []uint64
    }

    // Initialize 
    eventBus := eventbus.New()
    vc := ValidatorController{eventBus: eventBus} 
    dc := DutyController{eventBus: eventBus}
    dc.Start()

    // Usage
    vc.HandleClusterLiquidatedEvent(someEvent)  // Emits ValidatorLiquidatedEvent
    // DutyController handles event and removes duties

#### implementation break-up:
- `validators controller`:
- `controller contrustor`: add eventBus *eventbus.EventBus
- `event_handler`: emit the relevant event to duty controller
- `duties controller`:
- `dutyController contrustor`: add eventBus *eventbus.EventBus, subscribe to eventBus
- `event handler`:
- `ValidatorAddedEvent`: check if active, request duties, populate duties map
- `ClusterReactivatedEvent`: check if active, request duties, populate duties map
- `ValidatorRemovedEvent`: remove duties from map if exists
- `ClusterLiquidatedEvent`: remove duties from map if exists

estimation: ~1.5d

6. **📡 Resubscribe to Sync-Committee Each Epoch or Ideally on Beacon Node Restart ???**:
TBD

7. **⏰ Address Maximum Delay for Duty Start ???**:
TBD


---

### AttestationFetcher
- **Slot Ticker**:
- duties map is empty (warm-up or re-fetch on error)
- active indices for current epoch
- fetch duties for current epoch
- GT half-way through the epoch
- fetch duties for next epoch
- half-way through the epoch (slot % 32 == 16)
- fetch duties for next epoch
- **Head event**: (could be we want handle the head events separatly and listen to root changes only)
- previous dependent root changed
- re-fetch duties for current epoch
- current dependent root changed
- GT half-way through the epoch
- re-fetch duties for next epoch
- **ValidatorAddedEvent**:
- fetch duties for current epoch
- GT half-way through the epoch
- fetch duties for next epoch
- **ClusterReactivatedEvent**:
- fetch duties for current epoch
- GT half-way through the epoch
- fetch duties for next epoch

### ProposalFetcher
- **Slot Ticker**:
- duties map is empty
- active indices for current epoch (*could be we have active validators but there are no duties for those validators)
- fetch duties for current epoch
- start of epoch (slot % 32 == 0)
- fetch duties for current epoch
- **Head event**: (*could be we want handle the head events separatly and listen to root changes only)
- current dependent root changed
- re-fetch duties for current epoch
- **ValidatorAddedEvent**:
- fetch duties for current epoch
- **ClusterReactivatedEvent**:
- fetch duties for current epoch

### SyncCommitteFetcher
- **Slot Ticker**:
- duties map is empty
- active indices for current period (*could be we have active validators but there are no duties for those validators)
- fetch duties for current period
- GT half-way through the epoch & close to an EPOCHS_PER_SYNC_COMMITTEE_PERIOD boundary (*epoch % 256 == 256 - syncCommitteePreparationEpochs)
- fetch duties for next period
- half-way through the epoch & close to an EPOCHS_PER_SYNC_COMMITTEE_PERIOD boundary (*epoch % 256 == 256 - syncCommitteePreparationEpochs)
- fetch duties for next period
- **Head event**: (could be we want handle the head events separatly and listen to root changes only)
- current dependent root changed
- GT half-way through the epoch & GT to an EPOCHS_PER_SYNC_COMMITTEE_PERIOD boundary
- re-fetch duties for next period
- **ValidatorAddedEvent**:
- fetch duties for current period
- GT half-way through the epoch & GT to an EPOCHS_PER_SYNC_COMMITTEE_PERIOD boundary
- fetch duties for next period
- **ClusterReactivatedEvent**:
- fetch duties for current period
- GT half-way through the epoch & GT to an EPOCHS_PER_SYNC_COMMITTEE_PERIOD boundary
- fetch duties for next period

### Slot Ticker (Duty Handler)

x§


### SyncCommitteFetcher
- **Slot Ticker**:
    - duties map is empty
        - active indices for current period (*could be we have active validators but there are no duties for those validators)
            - fetch duties for current period
        - GT half-way through the epoch & close to an EPOCHS_PER_SYNC_COMMITTEE_PERIOD boundary (*epoch % 256 == 256 - syncCommitteePreparationEpochs)
            - fetch duties for next period
    - half-way through the epoch & close to an EPOCHS_PER_SYNC_COMMITTEE_PERIOD boundary (*epoch % 256 == 256 - syncCommitteePreparationEpochs)
        - fetch duties for next period
- **Head event**: (could be we want handle the head events separatly and listen to root changes only)
    - current dependent root changed
        - GT half-way through the epoch & GT to an EPOCHS_PER_SYNC_COMMITTEE_PERIOD boundary
            - re-fetch duties for next period
- **ValidatorAddedEvent**:
    - fetch duties for current period
    - GT half-way through the epoch & GT to an EPOCHS_PER_SYNC_COMMITTEE_PERIOD boundary
        - fetch duties for next period
- **ClusterReactivatedEvent**:
    - fetch duties for current period
    - GT half-way through the epoch & GT to an EPOCHS_PER_SYNC_COMMITTEE_PERIOD boundary
        - fetch duties for next period

func SyncCommitteeHandler(ctx) {
    for {
        var executionCtx context.Context
        var executionCancel func()
        
        duties := make(chan SyncCommitteeDuties)
        var wg sync.WaitGroup
        select {
            case slot := <-slotTicker:
                // should warm up duties for current period
                shouldFetchCurrentPeriod = h.job.Runs() == 0
                // half-way through the epoch and close to boundary or should warm up duties for next period
                shouldFetchNextPeriod = (slot % 32 == 16 && epoch % 256 == 256-syncCommitteePreparationEpochs) ||
                                        (shouldFetchCurrentEpoch && slot % 32 >= 16 && epoch % 256 >= 256-syncCommitteePreparationEpochs)

                if shouldFetchCurrentPeriod {
                    indices := ActiveValidatorIndices(currentEpoch)
                    h.job.Run(func() *Duties {
                        // do we want handle duty for current slot on warm-up?
                        // if yes - we have to execute only when recieve the duty
                        // if no - SyncCommitteeDuties should recieve `notCurrentSlot` param
                        // to not get the duty for the current slot
                        // e.g. SyncCommitteeDuties(executionCtx, currentEpoch, indices, true /* notCurrentSlot */)
                        return SyncCommitteeDuties(executionCtx, currentEpoch, indices)
                    })
                }
            
                executeSyncCommittee(executionCtx, slot)

                if shouldFetchNextPeriod {
                    // calculate nextSyncPeriodEpoch
                    indices := ActiveValidatorIndices(nextSyncPeriodEpoch)
                    h.job.Run(func() *Duties {
                        // e.g. SyncCommitteeDuties(executionCtx, nextSyncPeriodEpoch, indices, false /* notCurrentSlot */)
                        return SyncCommitteeDuties(executionCtx, nextSyncPeriodEpoch, indices)
                    })
                }

            case duties := <-h.job.Result():
                // populate sync committee duties map
                h.duties = duties

            case _ := <-CurrentDependentRootChanged:
                if (slot % 32 >= 16) && (epoch % 256 >= 256-syncCommitteePreparationEpochs) {
                    indices := ActiveValidatorIndices(nextSyncPeriodEpoch)
                    h.job.Run(func() *Duties {
                        // e.g. SyncCommitteeDuties(executionCtx, nextSyncPeriodEpoch, indices, false /* notCurrentSlot */)
                        return SyncCommitteeDuties(executionCtx, nextSyncPeriodEpoch, indices)
                    })
                }

            case eventData := <-ValidatorAddedEvent:
                indices := eventData.indices
                
                h.job.Run(func() *Duties {
                    // e.g. SyncCommitteeDuties(executionCtx, currentEpoch, indices, true /* notCurrentSlot */)
                    return SyncCommitteeDuties(executionCtx, currentEpoch, indices)
                })

                if (slot % 32 >= 16) && (epoch % 256 >= 256-syncCommitteePreparationEpochs) {
                    h.job.Run(func() *Duties {
                        // e.g. SyncCommitteeDuties(executionCtx, nextSyncPeriodEpoch, indices, false /* notCurrentSlot */)
                        return SyncCommitteeDuties(executionCtx, nextSyncPeriodEpoch, indices)
                    })
                }

            case eventData := <-ClusterReactivatedEvent:
                indices := eventData.indices
                
                h.job.Run(func() *Duties {
                    // e.g. SyncCommitteeDuties(executionCtx, currentEpoch, indices, true /* notCurrentSlot */)
                    return SyncCommitteeDuties(executionCtx, currentEpoch, indices)
                })

                if (slot % 32 >= 16) && (epoch % 256 >= 256-syncCommitteePreparationEpochs) {
                    h.job.Run(func() *Duties {
                        // e.g. SyncCommitteeDuties(executionCtx, nextSyncPeriodEpoch, indices, false /* notCurrentSlot */)
                        return SyncCommitteeDuties(executionCtx, nextSyncPeriodEpoch, indices)
                    })
                }
        }
    }
}

type Job[T any] struct {
    ctx
    cancel
    wg sync.WaitGroup
    result chan T
}

func (j Job) Run(ctx, f func() T) {
    if j.cancel != nil {
        j.cancel()
        j.wg.Wait()
        select {
            case <-j.result:
            default:
        }
    }

    j.wg.Add(1)
    j.ctx, j.cancel = context.WithCancel(ctx)
    go func() {
        defer j.wg.Done()
        j.result <- f()
    }()
}

func (j Job) Result() <-chan T {
    return j.result
}

type ExecuteDuties func(ctx context.Context, slot phase0.Slot, duties []spectypes.Duty)

func NewProposalHandler(executeSlot func(ctx, slot, duties))

type ProposalHandler struct {
  executeDuties ExecuteDuties
  fetch *Job
  duties map[phase0.Slot]Duty
}

func (h ProposalHandler) Run(ctx context.Context) {
    for {
        // validate how this should work
        var executionCtx context.Context
        var executionCancel func()
        
        select {
            case slot := <-slotTicker:
                shouldFetch = slot % 32 == 0 || h.job.Runs() == 0

                if shouldFetch {
                  h.fetchDutiesForEpoch(ctx, currentEpoch)
                  select {
                    case h.duties = <-h.fetch.Result():
                 	case <-slotTicker:
                  }
                }
          
                h.executeDuties(executionCtx, slot, h.duties)

            case duties := <-h.job.Result():
                // populate proposer duties map
                h.duties = duties
            case _ := <-CurrentDependentRootChanged:
                indices := ActiveValidatorIndices(currentEpoch)

                h.job.Run(func() *Duties {
                    // could be we dont need to get duty for current slot
                    return ProposerDuties(executionCtx, currentEpoch, indices)
                })
            
            case eventData := <-ValidatorAddedEvent:
                indices := eventData.indices
                
                h.job.Run(func() *Duties {
                    // could be we dont need to get duty for current slot?
                    // if yes we have to do the same with warm up?
                    return ProposerDuties(executionCtx, currentEpoch, indices)
                })

            case eventData := <-ClusterReactivatedEvent:
                indices := eventData.indices
                
                h.job.Run(func() Duties {
                    // could be we dont need to get duty for current slot?
                    // if yes we have to do the same with warm up?
                    return ProposerDuties(executionCtx, currentEpoch, indices)
                })
        }
    }
}

func (h *ProposalHandler) fetchDutiesForEpoch(ctx context.Context, epoch phase0.Epoch, indices []phase0.ValidatorIndices) {
  h.job.Run(func() []spectypes.Duty {
  	proposerDuties, err := h.beaconNode.ProposerDuties(ctx, currentEpoch, ActiveValidatorIndices(epoch)
                                                       duties := make()
    return duties
  })
}

func AttesterHandler(ctx) {
    for {
        var executionCtx context.Context
        var executionCancel func()
        
        duties := make(chan AttesterDuties)
        var wg sync.WaitGroup
        select {
            case slot := <-slotTicker:
                // should warm up duties for current epoch
                shouldFetchCurrentEpoch = h.job.Runs() == 0
                // half-way through the epoch or should warm up duties for next epoch
                shouldFetchNextEpoch = slot % 32 == 16 || (shouldFetchCurrentEpoch && slot % 32 > 16)

                if shouldFetchCurrentEpoch {
                    indices := ActiveValidatorIndices(currentEpoch)
                    h.job.Run(func() *Duties {
                        // do we want handle duty for current slot on warm-up?
                        // if yes - we have to execute only when recieve the duty
                        // if no - AttesterDuties should recieve `notCurrentSlot` param
                        // to not get the duty for the current slot
                        // e.g. AttesterDuties(executionCtx, currentEpoch, indices, true /* notCurrentSlot */)
                        return AttesterDuties(executionCtx, currentEpoch, indices)
                    })
                }
            
                executeAttestation(executionCtx, slot)

                if shouldFetchNextEpoch {
                    indices := ActiveValidatorIndices(currentEpoch+1)
                    h.job.Run(func() *Duties {
                        // e.g. AttesterDuties(executionCtx, currentEpoch+1, indices, false /* notCurrentSlot */)
                        return AttesterDuties(executionCtx, currentEpoch+1, indices)
                    })
                }

            case duties := <-h.job.Result():
                // populate attester duties map
                h.duties = duties

            case _ := <-PreviousDependentRootChanged:
                executionCancel()
            
                indices := ActiveValidatorIndices(currentEpoch)
                h.job.Run(func() *Duties {
                    // do we want handle duty for current slot on re-org?
                    // if yes - we have to execute only when recieve the duty
                    // if no - AttesterDuties should recieve `notCurrentSlot` param
                    // to not get the duty for the current slot
                    // e.g. AttesterDuties(executionCtx, currentEpoch, indices, true /* notCurrentSlot */)
                    return AttesterDuties(executionCtx, currentEpoch, indices)
                })
            
            case _ := <-CurrentDependentRootChanged:
                // could be we dont need to re-fetch 
                if (slot % 32 > 16) {
                    indices := ActiveValidatorIndices(currentEpoch+1)
                    h.job.Run(func() *Duties {
                        // e.g. AttesterDuties(executionCtx, currentEpoch+1, indices, false /* notCurrentSlot */)
                        return AttesterDuties(executionCtx, currentEpoch+1, indices)
                    })                        
                }

            case eventData := <-ValidatorAddedEvent:
                indices := eventData.indices
                
                h.job.Run(func() *Duties {
                    // e.g. AttesterDuties(executionCtx, currentEpoch, indices, true /* notCurrentSlot */)
                    return AttesterDuties(executionCtx, currentEpoch, indices)
                })

                if slot % 32 > 16 {
                    h.job.Run(func() *Duties {
                        return AttesterDuties(executionCtx, currentEpoch+1, indices)
                    })
                }

            case eventData := <-ClusterReactivatedEvent:
                indices := eventData.indices
                
                h.job.Run(func() *Duties {
                    // e.g. AttesterDuties(executionCtx, currentEpoch, indices, true /* notCurrentSlot */)
                    return AttesterDuties(executionCtx, currentEpoch, indices)
                })

                if slot % 32 > 16 {
                    h.job.Run(func() *Duties {
                        return AttesterDuties(executionCtx, currentEpoch+1, indices)
                    })
                }
        }
    }
}