cl/phase1/network/services: extract shared pending-job queue for the Gloas services#22177
Draft
yperbasis wants to merge 3 commits into
Draft
cl/phase1/network/services: extract shared pending-job queue for the Gloas services#22177yperbasis wants to merge 3 commits into
yperbasis wants to merge 3 commits into
Conversation
…ehavior The bid service's pending-queue drop paths (expiry, stale slot) and the background loop's enqueue-to-drain flow had no test coverage, unlike the sibling execution payload and payload attestation services. Pin them before consolidating the three token-identical queue implementations. The loop test populates the forkchoice mock's plain maps before the service constructor spawns the loop goroutine, so it is race-free; the preferences arrive later via the thread-safe LRU pool.
…gossip services The execution payload, payload attestation, and execution payload bid services each carried a token-identical pending-job queue (sync.Map + atomic count + cond-signalled polling loop). Consolidate it into an unexported generic pendingJobQueue[K, M]; readiness checks stay in per-service tryProcess callbacks since they differ on purpose per topic (block arrived / block arrived at current slot / preferences matched). Per-service deltas, kept as queue parameters or callback logic: - capacity: envelopes 1024, attestations 2048, bids 1024 - expiry: envelopes and attestations 30s, bids 12s (1 slot); tick 100ms everywhere - keys: (blockRoot, envelopeHash) / (blockRoot, validatorIndex) / (builderIndex, slot) - drop rules: attestations drop on stale slot; bids drop on stale slot and on non-dependency preference-matching errors, and keep waiting on errBidDependencyUnavailable - the envelope service hashes the envelope to build its key and drops with a warning if hashing fails The audit note that the bid service incremented pendingCount after LoadOrStore is stale: that ordering existed at Gloas introduction (f3ad70f) but was fixed to increment-first in #21655 (a2f483a); all three enqueues are token-identical today and the queue keeps that increment-first accounting. The tryProcess contract returns (process func(), done bool) so the queue removes a finished job before running its processing, exactly as the originals deleted before calling ProcessMessage/validateAndStoreBid; a concurrent re-enqueue of the same key during processing is therefore still stored rather than lost. Deliberate normalizations, none affecting accounting or validation: - envelope enqueue computes HashSSZ before the capacity check instead of after the capacity reservation; visible only when the queue is full (hash cost, and the hash-failure warning now logs even then) - drop-path trace logs now precede the map removal instead of following it (same goroutine, not observable) - the bid callback receives the loop context and ignores it, where processPendingBids previously took no context The ticker-only schedulers in block, blob sidecar, and data column sidecar services are intentionally not converted: they have no cond or capacity accounting, use Store (replace) rather than LoadOrStore (drop) on key collision, retry failed jobs until expiry instead of removing them before processing, and the data column variant increments its count after LoadOrStore and logs on capacity drops. Covering those would need option knobs for enqueue collision policy, error retry, and removal ordering, defeating the consolidation.
…lProposerChecker The aggregate-and-proof and sync-contribution services carried token-identical isLocalValidatorProposer methods (docstring included) along with duplicated proposerIndicesCache fields and LRU setup. Move them to a localProposerChecker type embedded in both services, keeping call sites unchanged via method promotion. Each service still owns its own cache instance, as before.
yperbasis
added
tech debt reduction
Glamsterdam
Contributor
There was a problem hiding this comment.
Pull request overview
This PR reduces duplicated gossip-service logic by extracting (1) a shared pending-job retry/expiry queue used by the Gloas gossip services and (2) a shared “local proposer” checker used by two services, keeping per-topic validation/processing decisions in service-specific callbacks.
Changes:
- Introduces a generic
pendingJobQueue[K, M]with capacity limiting, expiry, cond/ticker wakeups, and a two-phasetryProcesscontract to preserve delete-before-process ordering. - Refactors
execution_payload,payload_attestation, andexecution_payload_bidservices to use the shared pending queue; updates corresponding whitebox tests to target the new internals. - Extracts the previously duplicated
isLocalValidatorProposerlogic into an embeddedlocalProposerChecker, used byaggregate_and_proofandsync_contribution.
Reviewed changes
Copilot reviewed 10 out of 10 changed files in this pull request and generated no comments.
Show a summary per file
| File | Description |
|---|---|
| cl/phase1/network/services/pending_job_queue.go | Adds shared generic pending queue with capacity/expiry/retry loop and two-phase processing contract. |
| cl/phase1/network/services/execution_payload_service.go | Switches execution payload service pending-envelope handling to the shared queue. |
| cl/phase1/network/services/execution_payload_service_test.go | Re-points pending-envelope tests to the shared queue internals and helpers. |
| cl/phase1/network/services/payload_attestation_service.go | Switches payload attestation pending-block handling to the shared queue and adds tryProcess callback. |
| cl/phase1/network/services/payload_attestation_service_test.go | Re-points pending-attestation tests to the shared queue internals. |
| cl/phase1/network/services/execution_payload_bid_service.go | Switches pending-bid subsystem to the shared queue and moves retry logic into tryProcess. |
| cl/phase1/network/services/execution_payload_bid_service_test.go | Updates tests to shared queue internals and adds end-to-end loop coverage for queued bid drain. |
| cl/phase1/network/services/local_proposer_checker.go | Adds extracted localProposerChecker (LRU-backed proposer cache + lookahead logic). |
| cl/phase1/network/services/aggregate_and_proof_service.go | Uses embedded localProposerChecker instead of local duplicated cache+method. |
| cl/phase1/network/services/sync_contribution_service.go | Uses embedded localProposerChecker instead of local duplicated cache+method. |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
The three Gloas gossip services (
execution_payload,payload_attestation,execution_payload_bid) each carried a token-identical pending-job subsystem (~100 lines each): sync.Map keyed by root, capacity cap with increment-first accounting, expiry, cond+ticker wakeups, and a retry loop. Every new topic has been copying it wholesale.They now share one
pendingJobQueue[K, M]. Readiness/validation logic stays in per-service callbacks — thetryProcess(ctx, key, msg) (process func(), done bool)two-phase contract exists specifically to preserve the originals' delete-before-process ordering, so a concurrent same-key re-enqueue is not lost. Per-service capacity/expiry/keying/drop rules are parameters, enumerated in the commit body. Deliberate normalizations (all unobservable or full-queue-only): envelope key hashing now precedes the capacity check; drop-path trace logs precede the map delete.One audit-note correction along the way: the suspected bid-service count-ordering delta turned out to be stale — #21655 already normalized it; all three enqueues are identical today. The three looser ticker-only services (
block,blob_sidecar,data_column_sidecar) are deliberately NOT converted — they differ in collision policy (Store-replace), retry-on-error semantics, and removal ordering; forcing them under the same type would mean knobs for each.Also:
isLocalValidatorProposerwas token-identical (docstring included) inaggregate_and_proofandsync_contribution— extracted into an embeddedlocalProposerChecker, per-service LRU instances unchanged.Safety net: pre-refactor pinning tests for the bid queue (expiry, stale-slot drop, end-to-end loop drain), green on the old code including
-race -count=3, unchanged after; existing whitebox tests mechanically re-pointed at the queue internals with assertions untouched. Consensus [IGNORE]/[REJECT] semantics untouched.Net −84 production lines and each future gossip topic stops copying the subsystem. Verified:
go test ./cl/phase1/network/...(+-raceon services), scopedgolangci-lintclean (×2),make erigon.Part of a dedup series; siblings #22165–#22176.