Remove shared state mutexes for sync service especially pending messages

[syjn99]

prysm/beacon-chain/sync/service.go

Lines 141 to 216 in ae9e443

	// Service is responsible for handling all run time p2p related operations as the
	// main entry point for network messages.
	type Service struct {
	cfg *config
	ctx context.Context
	cancel context.CancelFunc
	slotToPendingBlocks *gcache.Cache
	seenPendingBlocks map[[32]byte]bool
	blkRootToPendingAtts map[[32]byte][]any
	subHandler *subTopicHandler
	pendingAttsLock sync.RWMutex
	pendingQueueLock sync.RWMutex
	chainStarted *atomic.Bool
	validateBlockLock sync.RWMutex
	rateLimiter *limiter
	seenBlockLock sync.RWMutex
	seenBlockCache *lru.Cache
	seenPayloadEnvelopeCache *lru.Cache
	seenExecutionPayloadBidCache *slotAwareCache
	highestExecutionPayloadBidCache *cache.HighestExecutionPayloadBidCache
	seenBlobLock sync.RWMutex
	seenBlobCache *lru.Cache
	seenDataColumnCache *slotAwareCache
	pendingGloasColumnsLock sync.RWMutex
	pendingGloasColumns map[[32]byte]*pendingGloasEntry
	seenAggregatedAttestationLock sync.RWMutex
	seenAggregatedAttestationCache *lru.Cache
	seenUnAggregatedAttestationLock sync.RWMutex
	seenUnAggregatedAttestationCache *lru.Cache
	seenExitLock sync.RWMutex
	seenExitCache *lru.Cache
	seenProposerSlashingLock sync.RWMutex
	seenProposerSlashingCache *lru.Cache
	seenAttesterSlashingLock sync.RWMutex
	seenAttesterSlashingCache map[uint64]bool
	seenSyncMessageLock sync.RWMutex
	seenSyncMessageCache *lru.Cache
	seenSyncContributionLock sync.RWMutex
	seenSyncContributionCache *lru.Cache
	badBlockCache *lru.Cache
	badBlockLock sync.RWMutex
	badPayloadCache *lru.Cache
	badPayloadLock sync.RWMutex
	syncContributionBitsOverlapLock sync.RWMutex
	syncContributionBitsOverlapCache *lru.Cache
	signatureChan chan *signatureVerifier
	clockWaiter startup.ClockWaiter
	initialSyncComplete chan struct{}
	verifierWaiter *verification.InitializerWaiter
	newBlobVerifier verification.NewBlobVerifier
	newColumnsVerifier verification.NewDataColumnsVerifier
	newPayloadAttestationVerifier verification.NewPayloadAttestationMsgVerifier
	newSignedProposerPreferencesVerifier verification.NewSignedProposerPreferencesVerifier
	newExecutionPayloadBidVerifier verification.NewExecutionPayloadBidVerifier
	columnSidecarsExecSingleFlight singleflight.Group
	reconstructionSingleFlight singleflight.Group
	payloadEnvelopeRequestSingleFlight singleflight.Group
	availableBlocker coverage.AvailableBlocker
	reconstructionRandGen *rand.Rand
	ctxMap ContextByteVersions
	slasherEnabled bool
	lcStore *lightClient.Store
	dataColumnLogCh chan dataColumnLogEntry
	payloadAttestationCache *cache.PayloadAttestationCache
	proposerPreferencesCache *cache.ProposerPreferencesCache
	subscribedValidatorsCache *cache.SubscribedValidatorsCache
	digestActions perDigestSet
	subscriptionSpawner func(func()) // see Service.spawn for details
	newExecutionPayloadEnvelopeVerifier verification.NewExecutionPayloadEnvelopeVerifier
	pendingPayloadEnvelopes map[[32]byte]map[uint64]*ethpb.SignedExecutionPayloadEnvelope
	pendingEnvelopeLock sync.RWMutex
	selfBuildSigFailures int
	selfBuildSigFailSlot primitives.Slot
	pendingPayloadAttestations map[[32]byte][]*ethpb.PayloadAttestationMessage
	pendingPayloadAttestationLock sync.RWMutex
	}

Currently, we have bunch of patterns to handle pending messages that heavily relies on mutex + shared data pattern. I'd say this architecture can be improved a lot by using channel pattern, which is considered more Go-idiomatic (Share Memory By Communicating). Mutex only protects the data while it cannot enforce orderings across critical sections.

• Add pendingAttsQueue in favor of using Go channels #17251

This PR is a trial to improve our pending attestation queue design. It not only moves the logic from mutex to channel-based approach but resolves pre-existing race: attestation is checked in -> block imports meanwhile -> attestation saved into a map will not be looked by anyone.

Ideally, for each pending queue, the service starts goroutines for processing the queue. The goroutine owns the pending map inside the goroutine which we can now remove mutex that protects the map. Communications are taking place via channels. Producers never block and owner never locks.

We can apply this refactoring in:

• pending attestation
• pending blocks
• pending messages introduced in Gloas

[0xsamalt]

I went through #17251's pendingAttsQueue pattern (channel-owned goroutine, non-blocking producers, nil-safe, synctest-based tests) and looked at how it'd map onto the other two queues.

pendingGloasColumns (in validate_data_column_gloas.go) is a close structural match, one lock, one map, no cross-file sharing. so I'd like to take that one next using the same shape.

Pending blocks is a bigger lift than a mechanical port: pendingQueueLock also guards seenPendingBlocks reads from validate_beacon_blocks.go, rpc_beacon_blocks_by_root.go, pending_payload_attestation.go, and validate_execution_payload_envelope.go, not just the pending-blocks queue itself. It's also timer-driven (async.RunEvery sweeping a gcache.Cache) rather than event-driven like attestations, so the channel pattern doesn't map over as directly. I'd rather do Gloas columns first, then come back with a proposal for how to scope the blocks queue once there's a second working example to point to. Let me know if that split works or if you'd rather I take a different slice.