fix(lib/babe): use current system time to yield a new slot

lib/babe/babe.go

@@ -420,7 +420,9 @@ func (b *Service) handleEpoch(epoch uint64) (next uint64, err error) {
 	case err := <-errCh:
 		// TODO: errEpochPast is sent on this channel, but it doesnot get logged here
 		epochTimer.Stop()
-		logger.Errorf("error from epochHandler: %s", err)
+		if err != nil {
+			logger.Errorf("error from epochHandler: %s", err)
+		}
 	}
 
 	// setup next epoch, re-invoke block authoring

lib/babe/epoch_handler.go

@@ -7,12 +7,9 @@ import (
 	"context"
 	"errors"
 	"fmt"
-	"sort"
-	"time"
 
 	"github.com/ChainSafe/gossamer/dot/types"
 	"github.com/ChainSafe/gossamer/lib/crypto/sr25519"
-	"golang.org/x/exp/maps"
 )
 
 type handleSlotFunc = func(epoch uint64, slot Slot, authorityIndex uint32,
@@ -23,6 +20,7 @@ var (
 )
 
 type epochHandler struct {
+	slotHandler *slotHandler
 	epochNumber uint64
 	firstSlot   uint64
 
@@ -53,6 +51,7 @@ func newEpochHandler(epochNumber, firstSlot uint64, epochData *epochData, consta
 	}
 
 	return &epochHandler{
+		slotHandler:            newSlotHandler(constants.slotDuration),
 		epochNumber:            epochNumber,
 		firstSlot:              firstSlot,
 		constants:              constants,
@@ -63,6 +62,7 @@ func newEpochHandler(epochNumber, firstSlot uint64, epochData *epochData, consta
 }
 
 func (h *epochHandler) run(ctx context.Context, errCh chan<- error) {
+	defer close(errCh)
 	currSlot := getCurrentSlot(h.constants.slotDuration)
 
 	// if currSlot < h.firstSlot, it means we're at genesis and waiting for the first slot to arrive.
@@ -75,83 +75,27 @@ func (h *epochHandler) run(ctx context.Context, errCh chan<- error) {
 		return
 	}
 
-	// for each slot we're handling, create a timer that will fire when it starts
-	// we create timers only for slots where we're authoring
-	authoringSlots := getAuthoringSlots(h.slotToPreRuntimeDigest)
-
-	type slotWithTimer struct {
-		startTime time.Time
-		timer     *time.Timer
-		slotNum   uint64
-	}
-
-	slotTimeTimers := make([]*slotWithTimer, 0, len(authoringSlots))
-	for _, authoringSlot := range authoringSlots {
-		if authoringSlot < currSlot {
-			// ignore slots already passed
-			continue
-		}
-
-		startTime := getSlotStartTime(authoringSlot, h.constants.slotDuration)
-		waitTime := time.Until(startTime)
-		timer := time.NewTimer(waitTime)
-
-		slotTimeTimers = append(slotTimeTimers, &slotWithTimer{
-			timer:     timer,
-			slotNum:   authoringSlot,
-			startTime: startTime,
-		})
-
-		logger.Debugf("start time of slot %d: %v", authoringSlot, startTime)
-	}
-
-	logger.Debugf("authoring in %d slots in epoch %d", len(slotTimeTimers), h.epochNumber)
-
-	for _, swt := range slotTimeTimers {
-		logger.Debugf("waiting for next authoring slot %d", swt.slotNum)
+	logger.Debugf("authoring in %d slots in epoch %d", len(h.slotToPreRuntimeDigest), h.epochNumber)
 
+	for {
 		select {
 		case <-ctx.Done():
-			for _, swt := range slotTimeTimers {
-				swt.timer.Stop()
-			}
+			errCh <- nil
 			return
-		case <-swt.timer.C:
-			// we must do a time correction as the slot timer sometimes is triggered
-			// before the time defined in the constructor due to an inconsistency
-			// of the language -> https://github.com/golang/go/issues/17696
-
-			diff := time.Since(swt.startTime)
-			if diff < 0 {
-				time.Sleep(-diff)
-			}
-
-			if _, has := h.slotToPreRuntimeDigest[swt.slotNum]; !has {
-				// this should never happen
-				panic(fmt.Sprintf("no VRF proof for authoring slot! slot=%d", swt.slotNum))
-			}
-
-			currentSlot := Slot{
-				start:    swt.startTime,
-				duration: h.constants.slotDuration,
-				number:   swt.slotNum,
-			}
-			err := h.handleSlot(h.epochNumber, currentSlot, h.epochData.authorityIndex, h.slotToPreRuntimeDigest[swt.slotNum])
-			if err != nil {
-				logger.Warnf("failed to handle slot %d: %s", swt.slotNum, err)
-				continue
-			}
+		default:
 		}
-	}
-}
 
-// getAuthoringSlots returns an ordered slice of slot numbers where we can author blocks,
-// based on the given VRF output and proof map.
-func getAuthoringSlots(slotToPreRuntimeDigest map[uint64]*types.PreRuntimeDigest) []uint64 {
-	authoringSlots := maps.Keys(slotToPreRuntimeDigest)
-	sort.Slice(authoringSlots, func(i, j int) bool {
-		return authoringSlots[i] < authoringSlots[j]
-	})
+		currentSlot := h.slotHandler.waitForNextSlot()
 
-	return authoringSlots
+		// check if the slot is an authoring slot otherwise wait for the next slot
+		digest, has := h.slotToPreRuntimeDigest[currentSlot.number]
+		if !has {
+			continue
+		}
+
+		err := h.handleSlot(h.epochNumber, currentSlot, h.epochData.authorityIndex, digest)
+		if err != nil {
+			logger.Warnf("failed to handle slot %d: %s", currentSlot.number, err)
+		}
+	}
 }

lib/babe/epoch_handler_integration_test.go

@@ -0,0 +1,69 @@
+package babe
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/ChainSafe/gossamer/dot/types"
+	"github.com/ChainSafe/gossamer/lib/crypto/sr25519"
+	"github.com/ChainSafe/gossamer/pkg/scale"
+	"github.com/stretchr/testify/require"
+)
+
+func TestEpochHandler_run(t *testing.T) {
+	const authorityIndex uint32 = 0
+	aliceKeyPair := keyring.Alice().(*sr25519.Keypair)
+	epochData := &epochData{
+		threshold:      scale.MaxUint128,
+		authorityIndex: authorityIndex,
+		authorities: []types.Authority{
+			*types.NewAuthority(aliceKeyPair.Public(), 1),
+		},
+	}
+
+	const slotDuration = 6 * time.Second
+	const epochLength uint64 = 100
+
+	constants := constants{ //nolint:govet
+		slotDuration: slotDuration,
+		epochLength:  epochLength,
+	}
+
+	const expectedEpoch = 1
+	startSlot := getCurrentSlot(slotDuration)
+	handleSlotFunc := testHandleSlotFunc(t, authorityIndex, expectedEpoch, startSlot)
+
+	epochHandler, err := newEpochHandler(1, startSlot, epochData, constants, handleSlotFunc, aliceKeyPair)
+	require.NoError(t, err)
+	require.Equal(t, epochLength, uint64(len(epochHandler.slotToPreRuntimeDigest)))
+
+	timeoutCtx, cancel := context.WithTimeout(context.Background(), slotDuration*10)
+	defer cancel()
+
+	errCh := make(chan error)
+	go epochHandler.run(timeoutCtx, errCh)
+
+	err = <-errCh
+	require.NoError(t, err)
+}
+
+func testHandleSlotFunc(t *testing.T, expectedAuthorityIndex uint32,
+	expectedEpoch, startSlot uint64) handleSlotFunc {
+	currentSlot := startSlot
+
+	return func(epoch uint64, slot Slot, authorityIndex uint32,
+		preRuntimeDigest *types.PreRuntimeDigest) error {
+
+		require.NotNil(t, preRuntimeDigest)
+		require.Equal(t, expectedEpoch, epoch)
+		require.Equal(t, expectedAuthorityIndex, authorityIndex)
+
+		require.Equal(t, slot.number, currentSlot, "%d != %d", slot.number, currentSlot)
+
+		// increase the slot by one so we expect the next call
+		// to be exactly 1 slot greater than the previous call
+		currentSlot += 1
+		return nil
+	}
+}

lib/babe/epoch_handler_test.go

@@ -4,7 +4,6 @@
 package babe
 
 import (
-	"context"
 	"testing"
 	"time"
 
@@ -45,54 +44,3 @@ func TestNewEpochHandler(t *testing.T) {
 	require.Equal(t, epochData, epochHandler.epochData)
 	require.NotNil(t, epochHandler.handleSlot)
 }
-
-func TestEpochHandler_run(t *testing.T) {
-	sd, err := time.ParseDuration("10ms")
-	require.NoError(t, err)
-	startSlot := getCurrentSlot(sd)
-
-	var callsToHandleSlot, firstExecutedSlot uint64
-	testHandleSlotFunc := func(epoch uint64, slot Slot, authorityIndex uint32,
-		preRuntimeDigest *types.PreRuntimeDigest,
-	) error {
-		require.Equal(t, uint64(1), epoch)
-		if callsToHandleSlot == 0 {
-			firstExecutedSlot = slot.number
-		} else {
-			require.Equal(t, firstExecutedSlot+callsToHandleSlot, slot.number)
-		}
-		require.Equal(t, uint32(0), authorityIndex)
-		require.NotNil(t, preRuntimeDigest)
-		callsToHandleSlot++
-		return nil
-	}
-
-	epochData := &epochData{
-		threshold: scale.MaxUint128,
-	}
-
-	const epochLength uint64 = 100
-	constants := constants{ //nolint:govet
-		slotDuration: sd,
-		epochLength:  epochLength,
-	}
-
-	keypair := keyring.Alice().(*sr25519.Keypair)
-
-	ctx, cancel := context.WithCancel(context.Background())
-	defer cancel()
-	epochHandler, err := newEpochHandler(1, startSlot, epochData, constants, testHandleSlotFunc, keypair)
-	require.NoError(t, err)
-	require.Equal(t, epochLength, uint64(len(epochHandler.slotToPreRuntimeDigest)))
-
-	errCh := make(chan error)
-	go epochHandler.run(ctx, errCh)
-	timer := time.NewTimer(sd * time.Duration(epochLength))
-	select {
-	case <-timer.C:
-		require.Equal(t, epochLength-(firstExecutedSlot-startSlot), callsToHandleSlot)
-	case err := <-errCh:
-		timer.Stop()
-		require.NoError(t, err)
-	}
-}

lib/babe/slot.go

@@ -0,0 +1,55 @@
+package babe
+
+import (
+	"time"
+)
+
+// timeUntilNextSlotInNanos calculates, based on the current system time, the remainng
+// time to the next slot
+func timeUntilNextSlotInMilli(slotDuration time.Duration) time.Duration {
+	now := time.Now().UnixNano()
+	slotDurationInMilli := slotDuration.Nanoseconds()
+
+	nextSlot := (now + slotDurationInMilli) / slotDurationInMilli
+
+	remaining := nextSlot*slotDurationInMilli - now
+	return time.Duration(remaining)
+}
+
+type slotHandler struct {
+	slotDuration time.Duration
+	lastSlot     *Slot
+}
+
+func newSlotHandler(slotDuration time.Duration) *slotHandler {
+	return &slotHandler{
+		slotDuration: slotDuration,
+	}
+}
+
+func (s *slotHandler) waitForNextSlot() Slot {
+	for {
+		// check if there is enough time to collaaborate
+		untilNextSlot := timeUntilNextSlotInMilli(s.slotDuration)
+		oneThird := s.slotDuration / 3
+		if untilNextSlot <= oneThird {
+			time.Sleep(untilNextSlot)
+		}
+
+		currentSystemTime := time.Now()
+		currentSlotNumber := uint64(currentSystemTime.UnixNano()) / uint64(s.slotDuration.Nanoseconds())
+		currentSlot := Slot{
+			start:    currentSystemTime,
+			duration: s.slotDuration,
+			number:   currentSlotNumber,
+		}
+
+		// Never yield the same slot twice
+		if s.lastSlot == nil || currentSlot.number > s.lastSlot.number {
+			s.lastSlot = &currentSlot
+			return currentSlot
+		}
+
+		time.Sleep(timeUntilNextSlotInMilli(s.slotDuration))
+	}
+}

lib/babe/slot_test.go

@@ -0,0 +1,27 @@
+package babe
+
+import (
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestSlotHandeConstructor(t *testing.T) {
+	expected := &slotHandler{
+		slotDuration: time.Duration(6000),
+	}
+
+	handler := newSlotHandler(time.Duration(6000))
+	require.Equal(t, expected, handler)
+}
+
+func TestSlotHandlerNextSlot(t *testing.T) {
+	slotDuration := 2 * time.Second
+	handler := newSlotHandler(slotDuration)
+
+	firstIteration := handler.waitForNextSlot()
+	secondIteration := handler.waitForNextSlot()
+
+	require.Greater(t, secondIteration.number, firstIteration.number)
+}