Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

feat(cannon): Lookahead 3 epochs #199

Merged
merged 2 commits into from
Sep 20, 2023
Merged

feat(cannon): Lookahead 3 epochs #199

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
1677910
feat(cannon): Lookahead 3 epochs
samcm Sep 20, 2023
19c94b5
refactor(pkg/cannon/deriver/beacon/eth/v2): remove unnecessary code
samcm Sep 20, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
32 changes: 17 additions & 15 deletions pkg/cannon/deriver/beacon/eth/v2/attester_slashing.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -106,7 +106,7 @@ func (a *AttesterSlashingDeriver) run(ctx context.Context) {
}

// Look ahead
a.lookAheadAtLocation(ctx, lookAhead)
a.lookAheadAtLocations(ctx, lookAhead)

for _, fn := range a.onLocationCallbacks {
if errr := fn(ctx, location.GetEthV2BeaconBlockAttesterSlashing().GetEpoch()); errr != nil {
Expand Down Expand Up @@ -148,27 +148,29 @@ func (a *AttesterSlashingDeriver) run(ctx context.Context) {
}
}

// lookAheadAtLocation takes a location and looks ahead to do any pre-processing that might be required.
func (a *AttesterSlashingDeriver) lookAheadAtLocation(ctx context.Context, location *xatu.CannonLocation) {
if location == nil {
// lookAheadAtLocation takes the upcoming locations and looks ahead to do any pre-processing that might be required.
func (a *AttesterSlashingDeriver) lookAheadAtLocations(ctx context.Context, locations []*xatu.CannonLocation) {
if locations == nil {
return
}

// Get the next look ahead epoch
epoch := phase0.Epoch(location.GetEthV2BeaconBlockVoluntaryExit().GetEpoch())
for _, location := range locations {
// Get the next look ahead epoch
epoch := phase0.Epoch(location.GetEthV2BeaconBlockVoluntaryExit().GetEpoch())

sp, err := a.beacon.Node().Spec()
if err != nil {
a.log.WithError(err).WithField("epoch", epoch).Warn("Failed to look ahead at epoch")
sp, err := a.beacon.Node().Spec()
if err != nil {
a.log.WithError(err).WithField("epoch", epoch).Warn("Failed to look ahead at epoch")

return
}
return
}

for i := uint64(0); i <= uint64(sp.SlotsPerEpoch); i++ {
slot := phase0.Slot(i + uint64(epoch)*uint64(sp.SlotsPerEpoch))
for i := uint64(0); i <= uint64(sp.SlotsPerEpoch); i++ {
slot := phase0.Slot(i + uint64(epoch)*uint64(sp.SlotsPerEpoch))

// Add the block to the preload queue so it's available when we need it
a.beacon.LazyLoadBeaconBlock(xatuethv1.SlotAsString(slot))
// Add the block to the preload queue so it's available when we need it
a.beacon.LazyLoadBeaconBlock(xatuethv1.SlotAsString(slot))
}
}
}

Expand Down
34 changes: 18 additions & 16 deletions pkg/cannon/deriver/beacon/eth/v2/bls_to_execution_change.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -103,13 +103,13 @@ func (b *BLSToExecutionChangeDeriver) run(ctx context.Context) {
}

// Get the next slot
location, lookAhead, err := b.iterator.Next(ctx)
location, lookAheads, err := b.iterator.Next(ctx)
if err != nil {
return err
}

// Look ahead
b.lookAheadAtLocation(ctx, lookAhead)
b.lookAheadAtLocation(ctx, lookAheads)

for _, fn := range b.onLocationCallbacks {
if errr := fn(ctx, location.GetEthV2BeaconBlockBlsToExecutionChange().GetEpoch()); errr != nil {
Expand Down Expand Up @@ -151,27 +151,29 @@ func (b *BLSToExecutionChangeDeriver) run(ctx context.Context) {
}
}

// lookAheadAtLocation takes a location and looks ahead to do any pre-processing that might be required.
func (b *BLSToExecutionChangeDeriver) lookAheadAtLocation(ctx context.Context, location *xatu.CannonLocation) {
if location == nil {
// lookAheadAtLocation takes the upcoming locations and looks ahead to do any pre-processing that might be required.
func (b *BLSToExecutionChangeDeriver) lookAheadAtLocation(ctx context.Context, locations []*xatu.CannonLocation) {
if locations == nil {
return
}

// Get the next look ahead epoch
epoch := phase0.Epoch(location.GetEthV2BeaconBlockVoluntaryExit().GetEpoch())
for _, location := range locations {
// Get the next look ahead epoch
epoch := phase0.Epoch(location.GetEthV2BeaconBlockVoluntaryExit().GetEpoch())

sp, err := b.beacon.Node().Spec()
if err != nil {
b.log.WithError(err).WithField("epoch", epoch).Warn("Failed to look ahead at epoch")
sp, err := b.beacon.Node().Spec()
if err != nil {
b.log.WithError(err).WithField("epoch", epoch).Warn("Failed to look ahead at epoch")

return
}
return
}

for i := uint64(0); i <= uint64(sp.SlotsPerEpoch); i++ {
slot := phase0.Slot(i + uint64(epoch)*uint64(sp.SlotsPerEpoch))
for i := uint64(0); i <= uint64(sp.SlotsPerEpoch); i++ {
slot := phase0.Slot(i + uint64(epoch)*uint64(sp.SlotsPerEpoch))

// Add the block to the preload queue so it's available when we need it
b.beacon.LazyLoadBeaconBlock(xatuethv1.SlotAsString(slot))
// Add the block to the preload queue so it's available when we need it
b.beacon.LazyLoadBeaconBlock(xatuethv1.SlotAsString(slot))
}
}
}

Expand Down
30 changes: 16 additions & 14 deletions pkg/cannon/deriver/beacon/eth/v2/deposit.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -149,27 +149,29 @@ func (b *DepositDeriver) run(ctx context.Context) {
}
}

// lookAheadAtLocation takes a location and looks ahead to do any pre-processing that might be required.
func (b *DepositDeriver) lookAheadAtLocation(ctx context.Context, location *xatu.CannonLocation) {
if location == nil {
// lookAheadAtLocation takes the upcoming locations and looks ahead to do any pre-processing that might be required.
func (b *DepositDeriver) lookAheadAtLocation(ctx context.Context, locations []*xatu.CannonLocation) {
if locations == nil {
return
}

// Get the next look ahead epoch
epoch := phase0.Epoch(location.GetEthV2BeaconBlockVoluntaryExit().GetEpoch())
for _, location := range locations {
// Get the next look ahead epoch
epoch := phase0.Epoch(location.GetEthV2BeaconBlockVoluntaryExit().GetEpoch())

sp, err := b.beacon.Node().Spec()
if err != nil {
b.log.WithError(err).WithField("epoch", epoch).Warn("Failed to look ahead at epoch")
sp, err := b.beacon.Node().Spec()
if err != nil {
b.log.WithError(err).WithField("epoch", epoch).Warn("Failed to look ahead at epoch")

return
}
return
}

for i := uint64(0); i <= uint64(sp.SlotsPerEpoch); i++ {
slot := phase0.Slot(i + uint64(epoch)*uint64(sp.SlotsPerEpoch))
for i := uint64(0); i <= uint64(sp.SlotsPerEpoch); i++ {
slot := phase0.Slot(i + uint64(epoch)*uint64(sp.SlotsPerEpoch))

// Add the block to the preload queue so it's available when we need it
b.beacon.LazyLoadBeaconBlock(xatuethv1.SlotAsString(slot))
// Add the block to the preload queue so it's available when we need it
b.beacon.LazyLoadBeaconBlock(xatuethv1.SlotAsString(slot))
}
}
}

Expand Down
32 changes: 15 additions & 17 deletions pkg/cannon/deriver/beacon/eth/v2/execution_transaction.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -174,27 +174,25 @@ func (b *ExecutionTransactionDeriver) processEpoch(ctx context.Context, epoch ph
return allEvents, nil
}

// lookAheadAtLocation takes a location and looks ahead to do any pre-processing that might be required.
func (b *ExecutionTransactionDeriver) lookAheadAtLocation(ctx context.Context, location *xatu.CannonLocation) {
if location == nil {
return
}

// Get the next look ahead epoch
epoch := phase0.Epoch(location.GetEthV2BeaconBlockVoluntaryExit().GetEpoch())
// lookAheadAtLocation takes the upcoming locations and looks ahead to do any pre-processing that might be required.
func (b *ExecutionTransactionDeriver) lookAheadAtLocation(ctx context.Context, locations []*xatu.CannonLocation) {
for _, location := range locations {
// Get the next look ahead epoch
epoch := phase0.Epoch(location.GetEthV2BeaconBlockVoluntaryExit().GetEpoch())

sp, err := b.beacon.Node().Spec()
if err != nil {
b.log.WithError(err).WithField("epoch", epoch).Warn("Failed to look ahead at epoch")
sp, err := b.beacon.Node().Spec()
if err != nil {
b.log.WithError(err).WithField("epoch", epoch).Warn("Failed to look ahead at epoch")

return
}
return
}

for i := uint64(0); i <= uint64(sp.SlotsPerEpoch); i++ {
slot := phase0.Slot(i + uint64(epoch)*uint64(sp.SlotsPerEpoch))
for i := uint64(0); i <= uint64(sp.SlotsPerEpoch); i++ {
slot := phase0.Slot(i + uint64(epoch)*uint64(sp.SlotsPerEpoch))

// Add the block to the preload queue so it's available when we need it
b.beacon.LazyLoadBeaconBlock(xatuethv1.SlotAsString(slot))
// Add the block to the preload queue so it's available when we need it
b.beacon.LazyLoadBeaconBlock(xatuethv1.SlotAsString(slot))
}
}
}

Expand Down
30 changes: 16 additions & 14 deletions pkg/cannon/deriver/beacon/eth/v2/proposer_slashing.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -243,27 +243,29 @@ func (b *ProposerSlashingDeriver) getProposerSlashings(ctx context.Context, bloc
return slashings, nil
}

// lookAheadAtLocation takes a location and looks ahead to do any pre-processing that might be required.
func (b *ProposerSlashingDeriver) lookAheadAtLocation(ctx context.Context, location *xatu.CannonLocation) {
if location == nil {
// lookAheadAtLocation takes the upcoming locations and looks ahead to do any pre-processing that might be required.
func (b *ProposerSlashingDeriver) lookAheadAtLocation(ctx context.Context, locations []*xatu.CannonLocation) {
if locations == nil {
return
}

// Get the next look ahead epoch
epoch := phase0.Epoch(location.GetEthV2BeaconBlockVoluntaryExit().GetEpoch())
for _, location := range locations {
// Get the next look ahead epoch
epoch := phase0.Epoch(location.GetEthV2BeaconBlockVoluntaryExit().GetEpoch())

sp, err := b.beacon.Node().Spec()
if err != nil {
b.log.WithError(err).WithField("epoch", epoch).Warn("Failed to look ahead at epoch")
sp, err := b.beacon.Node().Spec()
if err != nil {
b.log.WithError(err).WithField("epoch", epoch).Warn("Failed to look ahead at epoch")

return
}
return
}

for i := uint64(0); i <= uint64(sp.SlotsPerEpoch); i++ {
slot := phase0.Slot(i + uint64(epoch)*uint64(sp.SlotsPerEpoch))
for i := uint64(0); i <= uint64(sp.SlotsPerEpoch); i++ {
slot := phase0.Slot(i + uint64(epoch)*uint64(sp.SlotsPerEpoch))

// Add the block to the preload queue so it's available when we need it
b.beacon.LazyLoadBeaconBlock(xatuethv1.SlotAsString(slot))
// Add the block to the preload queue so it's available when we need it
b.beacon.LazyLoadBeaconBlock(xatuethv1.SlotAsString(slot))
}
}
}

Expand Down
30 changes: 16 additions & 14 deletions pkg/cannon/deriver/beacon/eth/v2/voluntary_exit.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -149,27 +149,29 @@ func (b *VoluntaryExitDeriver) run(ctx context.Context) {
}
}

// lookAheadAtLocation takes a location and looks ahead to do any pre-processing that might be required.
func (b *VoluntaryExitDeriver) lookAheadAtLocation(ctx context.Context, location *xatu.CannonLocation) {
if location == nil {
// lookAheadAtLocation takes the upcoming locations and looks ahead to do any pre-processing that might be required.
func (b *VoluntaryExitDeriver) lookAheadAtLocation(ctx context.Context, locations []*xatu.CannonLocation) {
if locations == nil {
return
}

// Get the next look ahead epoch
epoch := phase0.Epoch(location.GetEthV2BeaconBlockVoluntaryExit().GetEpoch())
for _, location := range locations {
// Get the next look ahead epoch
epoch := phase0.Epoch(location.GetEthV2BeaconBlockVoluntaryExit().GetEpoch())

sp, err := b.beacon.Node().Spec()
if err != nil {
b.log.WithError(err).WithField("epoch", epoch).Warn("Failed to look ahead at epoch")
sp, err := b.beacon.Node().Spec()
if err != nil {
b.log.WithError(err).WithField("epoch", epoch).Warn("Failed to look ahead at epoch")

return
}
return
}

for i := uint64(0); i <= uint64(sp.SlotsPerEpoch); i++ {
slot := phase0.Slot(i + uint64(epoch)*uint64(sp.SlotsPerEpoch))
for i := uint64(0); i <= uint64(sp.SlotsPerEpoch); i++ {
slot := phase0.Slot(i + uint64(epoch)*uint64(sp.SlotsPerEpoch))

// Add the block to the preload queue so it's available when we need it
b.beacon.LazyLoadBeaconBlock(xatuethv1.SlotAsString(slot))
// Add the block to the preload queue so it's available when we need it
b.beacon.LazyLoadBeaconBlock(xatuethv1.SlotAsString(slot))
}
}
}

Expand Down
47 changes: 26 additions & 21 deletions pkg/cannon/iterator/checkpoint_iterator.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -45,39 +45,39 @@ func (c *CheckpointIterator) UpdateLocation(ctx context.Context, location *xatu.
return c.coordinator.UpsertCannonLocationRequest(ctx, location)
}

func (c *CheckpointIterator) Next(ctx context.Context) (next, lookAhead *xatu.CannonLocation, err error) {
func (c *CheckpointIterator) Next(ctx context.Context) (next *xatu.CannonLocation, lookAhead []*xatu.CannonLocation, err error) {
for {
// Grab the current checkpoint from the beacon node
checkpoint, err := c.fetchLatestEpoch(ctx)
if err != nil {
return nil, nil, errors.Wrap(err, "failed to fetch latest checkpoint")
return nil, []*xatu.CannonLocation{}, errors.Wrap(err, "failed to fetch latest checkpoint")
}

if checkpoint == nil {
return nil, nil, errors.New("checkpoint is nil")
return nil, []*xatu.CannonLocation{}, errors.New("checkpoint is nil")
}

// Check where we are at from the coordinator
location, err := c.coordinator.GetCannonLocation(ctx, c.cannonType, c.networkID)
if err != nil {
return nil, nil, errors.Wrap(err, "failed to get cannon location")
return nil, []*xatu.CannonLocation{}, errors.Wrap(err, "failed to get cannon location")
}

// If location is empty we haven't started yet, start at the network default for the type. If the network default
// is empty, we'll start at epoch 0.
if location == nil {
location, err = c.createLocationFromEpochNumber(phase0.Epoch(GetDefaultSlotLocationForNetworkAndType(c.networkName, c.cannonType) / 32))
if err != nil {
return nil, nil, errors.Wrap(err, "failed to create location from slot number 0")
return nil, []*xatu.CannonLocation{}, errors.Wrap(err, "failed to create location from slot number 0")
}

return location, c.calculateLookAhead(ctx, location), nil
return location, c.getLookAheads(ctx, location), nil
}

// If the location is the same as the current checkpoint, we should sleep until the next epoch
locationEpoch, err := c.getEpochFromLocation(location)
if err != nil {
return nil, nil, errors.Wrap(err, "failed to get epoch from location")
return nil, []*xatu.CannonLocation{}, errors.Wrap(err, "failed to get epoch from location")
}

c.metrics.SetTrailingEpochs(c.cannonType.String(), c.networkName, c.checkpointName, float64(checkpoint.Epoch-locationEpoch))
Expand Down Expand Up @@ -106,40 +106,45 @@ func (c *CheckpointIterator) Next(ctx context.Context) (next, lookAhead *xatu.Ca

current, err := c.createLocationFromEpochNumber(nextEpoch)
if err != nil {
return nil, nil, errors.Wrap(err, "failed to create location from epoch number")
return nil, []*xatu.CannonLocation{}, errors.Wrap(err, "failed to create location from epoch number")
}

c.metrics.SetCurrentEpoch(c.cannonType.String(), c.networkName, c.checkpointName, float64(nextEpoch))

return current, c.calculateLookAhead(ctx, current), nil
return current, c.getLookAheads(ctx, current), nil
}
}

func (c *CheckpointIterator) calculateLookAhead(ctx context.Context, location *xatu.CannonLocation) *xatu.CannonLocation {
func (c *CheckpointIterator) getLookAheads(ctx context.Context, location *xatu.CannonLocation) []*xatu.CannonLocation {
// Calculate if we should look ahead
epoch, err := c.getEpochFromLocation(location)
if err != nil {
return nil
return []*xatu.CannonLocation{}
}

latestCheckpoint, err := c.fetchLatestEpoch(ctx)
if err != nil {
return nil
return []*xatu.CannonLocation{}
}

if epoch == latestCheckpoint.Epoch {
return nil
}
lookAheads := make([]*xatu.CannonLocation, 0)

// We aren't at the head, so we can look ahead
lookAheadEpoch := epoch + 1
for _, i := range []int{1, 2, 3} {
lookAheadEpoch := epoch + phase0.Epoch(i)

loc, err := c.createLocationFromEpochNumber(lookAheadEpoch)
if err != nil {
return nil
if lookAheadEpoch > latestCheckpoint.Epoch {
continue
}

loc, err := c.createLocationFromEpochNumber(lookAheadEpoch)
if err != nil {
return []*xatu.CannonLocation{}
}

lookAheads = append(lookAheads, loc)
}

return loc
return lookAheads
}

func (c *CheckpointIterator) fetchLatestEpoch(ctx context.Context) (*phase0.Checkpoint, error) {
Expand Down
Loading
Loading