fsm.go

package initialsync

import (
	"errors"
	"fmt"
	"time"

	eth "github.com/prysmaticlabs/ethereumapis/eth/v1alpha1"
)

const (
	stateNew stateID = iota
	stateScheduled
	stateDataParsed
	stateSkipped
	stateSent
	stateSkippedExt
	stateComplete
)

const (
	eventSchedule eventID = iota
	eventDataReceived
	eventReadyToSend
	eventCheckStale
	eventExtendWindow
)

// stateID is unique handle for a state.
type stateID uint8

// eventID is unique handle for an event.
type eventID uint8

// stateMachine is a FSM that allows easy state transitions:
// State(S) x Event(E) -> Actions (A), State(S').
type stateMachine struct {
	epochs []*epochState
	events map[eventID]*stateMachineEvent
}

// epochState holds state of a single epoch.
type epochState struct {
	epoch   uint64
	state   stateID
	blocks  []*eth.SignedBeaconBlock
	updated time.Time
}

// stateMachineEvent is a container for event data.
type stateMachineEvent struct {
	name    eventID
	actions []*stateMachineAction
}

// stateMachineAction represents a state actions that can be attached to an event.
type stateMachineAction struct {
	state     stateID
	handlerFn eventHandlerFn
}

// eventHandlerFn is an event handler function's signature.
type eventHandlerFn func(*epochState, interface{}) (stateID, error)

// newStateMachine returns fully initialized state machine.
func newStateMachine() *stateMachine {
	return &stateMachine{
		epochs: make([]*epochState, 0, lookaheadEpochs),
		events: map[eventID]*stateMachineEvent{},
	}
}

// addHandler attaches an event handler to a state event.
func (sm *stateMachine) addHandler(state stateID, event eventID, fn eventHandlerFn) *stateMachineEvent {
	e, ok := sm.events[event]
	if !ok {
		e = &stateMachineEvent{
			name: event,
		}
		sm.events[event] = e
	}
	action := &stateMachineAction{
		state:     state,
		handlerFn: fn,
	}
	e.actions = append(e.actions, action)
	return e
}

// trigger invokes the event on a given epoch's state machine.
func (sm *stateMachine) trigger(name eventID, epoch uint64, data interface{}) error {
	event, ok := sm.events[name]
	if !ok {
		return fmt.Errorf("event not found: %v", name)
	}

	ind, ok := sm.findEpochState(epoch)
	if !ok {
		return fmt.Errorf("state for %v epoch not found", epoch)
	}

	for _, action := range event.actions {
		if action.state != sm.epochs[ind].state {
			continue
		}
		state, err := action.handlerFn(sm.epochs[ind], data)
		if err != nil {
			return err
		}
		sm.epochs[ind].setState(state)
	}

	return nil
}

// addEpochState allocates memory for tracking epoch state.
func (sm *stateMachine) addEpochState(epoch uint64) {
	state := &epochState{
		epoch:   epoch,
		state:   stateNew,
		blocks:  []*eth.SignedBeaconBlock{},
		updated: time.Now(),
	}
	sm.epochs = append(sm.epochs, state)
}

// removeEpochState frees memory of processed epoch.
func (sm *stateMachine) removeEpochState(epoch uint64) error {
	ind, ok := sm.findEpochState(epoch)
	if !ok {
		return fmt.Errorf("state for %v epoch not found", epoch)
	}
	sm.epochs[ind].blocks = nil
	sm.epochs[ind] = sm.epochs[len(sm.epochs)-1]
	sm.epochs = sm.epochs[:len(sm.epochs)-1]
	return nil
}

// findEpochState returns index at which state.epoch = epoch, or len(epochs) if not found.
func (sm *stateMachine) findEpochState(epoch uint64) (int, bool) {
	for i, state := range sm.epochs {
		if epoch == state.epoch {
			return i, true
		}
	}
	return len(sm.epochs), false
}

// isLowestEpochState checks whether a given epoch is the lowest for which we know epoch state.
func (sm *stateMachine) isLowestEpochState(epoch uint64) bool {
	if _, ok := sm.findEpochState(epoch); !ok {
		return false
	}
	for _, state := range sm.epochs {
		if epoch > state.epoch {
			return false
		}
	}
	return true
}

// lowestEpoch returns epoch number for the earliest known epoch.
func (sm *stateMachine) lowestEpoch() (uint64, error) {
	if len(sm.epochs) == 0 {
		return 0, errors.New("no epoch states exist")
	}
	lowestEpoch := sm.epochs[0].epoch
	for _, state := range sm.epochs {
		if state.epoch < lowestEpoch {
			lowestEpoch = state.epoch
		}
	}
	return lowestEpoch, nil
}

// highestEpoch returns epoch number for the latest known epoch.
func (sm *stateMachine) highestEpoch() (uint64, error) {
	if len(sm.epochs) == 0 {
		return 0, errors.New("no epoch states exist")
	}
	highestEpoch := sm.epochs[0].epoch
	for _, state := range sm.epochs {
		if state.epoch > highestEpoch {
			highestEpoch = state.epoch
		}
	}
	return highestEpoch, nil
}

// String returns human readable representation of a state.
func (sm *stateMachine) String() string {
	return fmt.Sprintf("%v", sm.epochs)
}

// String returns human-readable representation of an epoch state.
func (es *epochState) String() string {
	return fmt.Sprintf("%d:%s", es.epoch, es.state)
}

// String returns human-readable representation of a state.
func (s stateID) String() (state string) {
	switch s {
	case stateNew:
		state = "new"
	case stateScheduled:
		state = "scheduled"
	case stateDataParsed:
		state = "dataParsed"
	case stateSkipped:
		state = "skipped"
	case stateSkippedExt:
		state = "skippedExt"
	case stateSent:
		state = "sent"
	case stateComplete:
		state = "complete"
	}
	return
}

// setState updates the current state of a given epoch.
func (es *epochState) setState(name stateID) {
	if es.state == name {
		return
	}
	es.updated = time.Now()
	es.state = name
}

// String returns human-readable representation of an event.
func (e eventID) String() (event string) {
	switch e {
	case eventSchedule:
		event = "schedule"
	case eventDataReceived:
		event = "dataReceived"
	case eventReadyToSend:
		event = "readyToSend"
	case eventCheckStale:
		event = "checkStale"
	case eventExtendWindow:
		event = "extendWindow"
	}
	return
}