Changelog fragment

beacon-chain/blockchain/head.go

@@ -285,10 +285,7 @@ func (s *Service) headState(ctx context.Context) state.BeaconState {
 // This returns a read only version of the head state.
 // It does not perform a copy of the head state.
 // This is a lock free version.
-func (s *Service) headStateReadOnly(ctx context.Context) state.ReadOnlyBeaconState {
-	_, span := trace.StartSpan(ctx, "blockChain.headStateReadOnly")
-	defer span.End()
-
+func (s *Service) headStateReadOnly(_ context.Context) state.ReadOnlyBeaconState {
 	return s.head.state
 }
 

beacon-chain/db/kv/backfill.go

@@ -29,8 +29,11 @@ func (s *Store) SaveBackfillStatus(ctx context.Context, bf *dbval.BackfillStatus
 // BackfillStatus retrieves the most recently saved version of the BackfillStatus protobuf struct.
 // This is used to persist information about backfill status across restarts.
 func (s *Store) BackfillStatus(ctx context.Context) (*dbval.BackfillStatus, error) {
-	_, span := trace.StartSpan(ctx, "BeaconDB.BackfillStatus")
+	ctx, span := trace.StartSpan(ctx, "BeaconDB.BackfillStatus")
 	defer span.End()
+	if ctx.Err() != nil {
+		return nil, ctx.Err() // Abort db lookup if context is done.
+	}
 	bf := &dbval.BackfillStatus{}
 	err := s.db.View(func(tx *bolt.Tx) error {
 		bucket := tx.Bucket(blocksBucket)

beacon-chain/db/kv/blocks.go

@@ -64,10 +64,15 @@ func (s *Store) getBlock(ctx context.Context, blockRoot [32]byte, tx *bolt.Tx) (
 // at the time the chain was started, used to initialize the database and chain
 // without syncing from genesis.
 func (s *Store) OriginCheckpointBlockRoot(ctx context.Context) ([32]byte, error) {
-	_, span := trace.StartSpan(ctx, "BeaconDB.OriginCheckpointBlockRoot")
+	ctx, span := trace.StartSpan(ctx, "BeaconDB.OriginCheckpointBlockRoot")
 	defer span.End()
 
 	var root [32]byte
+
+	if ctx.Err() != nil {
+		return root, ctx.Err()
+	}
+
 	err := s.db.View(func(tx *bolt.Tx) error {
 		bkt := tx.Bucket(blocksBucket)
 		rootSlice := bkt.Get(originCheckpointBlockRootKey)
@@ -868,7 +873,7 @@ type slotRoot struct {
 // slotRootsInRange returns slot and block root pairs of length min(batchSize, end-slot)
 // If batchSize < 0, the limit check will be skipped entirely.
 func (s *Store) slotRootsInRange(ctx context.Context, start, end primitives.Slot, batchSize int) ([]slotRoot, error) {
-	_, span := trace.StartSpan(ctx, "BeaconDB.slotRootsInRange")
+	ctx, span := trace.StartSpan(ctx, "BeaconDB.slotRootsInRange")
 	defer span.End()
 	if end < start {
 		return nil, errInvalidSlotRange
@@ -882,6 +887,9 @@ func (s *Store) slotRootsInRange(ctx context.Context, start, end primitives.Slot
 		bkt := tx.Bucket(blockSlotIndicesBucket)
 		c := bkt.Cursor()
 		for k, v := c.Seek(key); ; /* rely on internal checks to exit */ k, v = c.Prev() {
+			if ctx.Err() != nil {
+				return ctx.Err()
+			}
 			if len(k) == 0 && len(v) == 0 {
 				// The `edge`` variable and this `if` deal with 2 edge cases:
 				// - Seeking past the end of the bucket (the `end` param is higher than the highest slot).
@@ -988,7 +996,7 @@ func blockRootsBySlotRange(
 	bkt *bolt.Bucket,
 	startSlotEncoded, endSlotEncoded, startEpochEncoded, endEpochEncoded, slotStepEncoded interface{},
 ) ([][]byte, error) {
-	_, span := trace.StartSpan(ctx, "BeaconDB.blockRootsBySlotRange")
+	ctx, span := trace.StartSpan(ctx, "BeaconDB.blockRootsBySlotRange")
 	defer span.End()
 
 	// Return nothing when all slot parameters are missing
@@ -1035,6 +1043,9 @@ func blockRootsBySlotRange(
 	roots := make([][]byte, 0, rootsRange)
 	c := bkt.Cursor()
 	for k, v := c.Seek(min); conditional(k, max); k, v = c.Next() {
+		if ctx.Err() != nil {
+			return nil, ctx.Err()
+		}
 		slot := bytesutil.BytesToSlotBigEndian(k)
 		if step > 1 {
 			if slot.SubSlot(startSlot).Mod(step) != 0 {
@@ -1053,9 +1064,13 @@ func blockRootsBySlotRange(
 
 // blockRootsBySlot retrieves the block roots by slot
 func blockRootsBySlot(ctx context.Context, tx *bolt.Tx, slot primitives.Slot) ([][32]byte, error) {
-	_, span := trace.StartSpan(ctx, "BeaconDB.blockRootsBySlot")
+	ctx, span := trace.StartSpan(ctx, "BeaconDB.blockRootsBySlot")
 	defer span.End()
 
+	if ctx.Err() != nil {
+		return nil, ctx.Err()
+	}
+
 	bkt := tx.Bucket(blockSlotIndicesBucket)
 	key := bytesutil.SlotToBytesBigEndian(slot)
 	c := bkt.Cursor()
@@ -1077,10 +1092,13 @@ func blockRootsBySlot(ctx context.Context, tx *bolt.Tx, slot primitives.Slot) ([
 // objects. If a certain filter criterion does not apply to
 // blocks, an appropriate error is returned.
 func createBlockIndicesFromFilters(ctx context.Context, f *filters.QueryFilter) (map[string][]byte, error) {
-	_, span := trace.StartSpan(ctx, "BeaconDB.createBlockIndicesFromFilters")
+	ctx, span := trace.StartSpan(ctx, "BeaconDB.createBlockIndicesFromFilters")
 	defer span.End()
 	indicesByBucket := make(map[string][]byte)
 	for k, v := range f.Filters() {
+		if ctx.Err() != nil {
+			return nil, ctx.Err()
+		}
 		switch k {
 		case filters.ParentRoot:
 			parentRoot, ok := v.([]byte)

beacon-chain/db/kv/state.go

@@ -772,10 +772,13 @@ func (s *Store) validatorEntries(ctx context.Context, blockRoot [32]byte) ([]*et
 
 // retrieves and assembles the state information from multiple buckets.
 func (s *Store) stateBytes(ctx context.Context, blockRoot [32]byte) ([]byte, error) {
-	_, span := trace.StartSpan(ctx, "BeaconDB.stateBytes")
+	ctx, span := trace.StartSpan(ctx, "BeaconDB.stateBytes")
 	defer span.End()
 	var dst []byte
 	err := s.db.View(func(tx *bolt.Tx) error {
+		if ctx.Err() != nil {
+			return ctx.Err()
+		}
 		bkt := tx.Bucket(stateBucket)
 		stBytes := bkt.Get(blockRoot[:])
 		if len(stBytes) == 0 {

beacon-chain/operations/attestations/kv/aggregated.go

@@ -196,7 +196,7 @@ func (c *AttCaches) AggregatedAttestationsBySlotIndex(
 	slot primitives.Slot,
 	committeeIndex primitives.CommitteeIndex,
 ) []*ethpb.Attestation {
-	_, span := trace.StartSpan(ctx, "operations.attestations.kv.AggregatedAttestationsBySlotIndex")
+	ctx, span := trace.StartSpan(ctx, "operations.attestations.kv.AggregatedAttestationsBySlotIndex")
 	defer span.End()
 
 	atts := make([]*ethpb.Attestation, 0)
@@ -204,6 +204,9 @@ func (c *AttCaches) AggregatedAttestationsBySlotIndex(
 	c.aggregatedAttLock.RLock()
 	defer c.aggregatedAttLock.RUnlock()
 	for _, as := range c.aggregatedAtt {
+		if ctx.Err() != nil {
+			return atts // Exit early with whatever we have, if the context is done.
+		}
 		if as[0].Version() == version.Phase0 && slot == as[0].GetData().Slot && committeeIndex == as[0].GetData().CommitteeIndex {
 			for _, a := range as {
 				att, ok := a.(*ethpb.Attestation)
@@ -225,7 +228,7 @@ func (c *AttCaches) AggregatedAttestationsBySlotIndexElectra(
 	slot primitives.Slot,
 	committeeIndex primitives.CommitteeIndex,
 ) []*ethpb.AttestationElectra {
-	_, span := trace.StartSpan(ctx, "operations.attestations.kv.AggregatedAttestationsBySlotIndexElectra")
+	ctx, span := trace.StartSpan(ctx, "operations.attestations.kv.AggregatedAttestationsBySlotIndexElectra")
 	defer span.End()
 
 	atts := make([]*ethpb.AttestationElectra, 0)
@@ -233,6 +236,9 @@ func (c *AttCaches) AggregatedAttestationsBySlotIndexElectra(
 	c.aggregatedAttLock.RLock()
 	defer c.aggregatedAttLock.RUnlock()
 	for _, as := range c.aggregatedAtt {
+		if ctx.Err() != nil {
+			return atts // Exit early with whatever we have, if the context is done.
+		}
 		if as[0].Version() >= version.Electra && slot == as[0].GetData().Slot && as[0].CommitteeBitsVal().BitAt(uint64(committeeIndex)) {
 			for _, a := range as {
 				att, ok := a.(*ethpb.AttestationElectra)

beacon-chain/operations/slashings/pool.go

@@ -34,7 +34,7 @@ func NewPool() *Pool {
 func (p *Pool) PendingAttesterSlashings(ctx context.Context, state state.ReadOnlyBeaconState, noLimit bool) []ethpb.AttSlashing {
 	p.lock.Lock()
 	defer p.lock.Unlock()
-	_, span := trace.StartSpan(ctx, "operations.PendingAttesterSlashing")
+	ctx, span := trace.StartSpan(ctx, "operations.PendingAttesterSlashing")
 	defer span.End()
 
 	// Update prom metric.
@@ -54,6 +54,9 @@ func (p *Pool) PendingAttesterSlashings(ctx context.Context, state state.ReadOnl
 	}
 	pending := make([]ethpb.AttSlashing, 0, maxSlashings)
 	for i := 0; i < len(p.pendingAttesterSlashing); i++ {
+		if ctx.Err() != nil {
+			break // Stop processing work...
+		}
 		if uint64(len(pending)) >= maxSlashings {
 			break
 		}
@@ -89,7 +92,7 @@ func (p *Pool) PendingAttesterSlashings(ctx context.Context, state state.ReadOnl
 func (p *Pool) PendingProposerSlashings(ctx context.Context, state state.ReadOnlyBeaconState, noLimit bool) []*ethpb.ProposerSlashing {
 	p.lock.Lock()
 	defer p.lock.Unlock()
-	_, span := trace.StartSpan(ctx, "operations.PendingProposerSlashing")
+	ctx, span := trace.StartSpan(ctx, "operations.PendingProposerSlashing")
 	defer span.End()
 
 	// Update prom metric.
@@ -102,6 +105,9 @@ func (p *Pool) PendingProposerSlashings(ctx context.Context, state state.ReadOnl
 	}
 	pending := make([]*ethpb.ProposerSlashing, 0, maxSlashings)
 	for i := 0; i < len(p.pendingProposerSlashing); i++ {
+		if ctx.Err() != nil {
+			break // Stop processing work...
+		}
 		if uint64(len(pending)) >= maxSlashings {
 			break
 		}

changelog/pvl-ctx.md

@@ -0,0 +1,3 @@
+### Changed
+
+- Add a context liveness check to some methods that iterate over objects or do expensive work. The routine can end early if the request is no longer alive.

consensus-types/blocks/proofs.go

@@ -21,7 +21,7 @@ const (
 )
 
 func ComputeBlockBodyFieldRoots(ctx context.Context, blockBody *BeaconBlockBody) ([][]byte, error) {
-	_, span := trace.StartSpan(ctx, "blocks.ComputeBlockBodyFieldRoots")
+	ctx, span := trace.StartSpan(ctx, "blocks.ComputeBlockBodyFieldRoots")
 	defer span.End()
 
 	if blockBody == nil {
@@ -49,6 +49,9 @@ func ComputeBlockBodyFieldRoots(ctx context.Context, blockBody *BeaconBlockBody)
 	}
 
 	for i := range fieldRoots {
+		if ctx.Err() != nil {
+			return nil, ctx.Err()
+		}
 		fieldRoots[i] = make([]byte, 32)
 	}