Skip to content

refactor(sync): use deque(maxlen=N) for pending attestation buffers #746

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

Merged
tcoratger merged 2 commits into from
May 21, 2026
Merged

refactor(sync): use deque(maxlen=N) for pending attestation buffers #746

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
b6889e4
refactor(sync): use deque(maxlen=N) for pending attestation buffers
tcoratger May 21, 2026
b59e318
refactor(sync): drain pending queues by reassigning a fresh deque
tcoratger May 21, 2026
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
21 changes: 10 additions & 11 deletions src/lean_spec/subspecs/sync/service.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -7,6 +7,7 @@
from __future__ import annotations

import logging
from collections import deque
from collections.abc import Callable, Coroutine
from dataclasses import dataclass, field

Expand Down Expand Up @@ -137,17 +138,21 @@ class SyncService:
_blocks_processed: int = field(default=0)
"""Counter for processed blocks."""

_pending_attestations: list[SignedAttestation] = field(default_factory=list)
_pending_attestations: deque[SignedAttestation] = field(
default_factory=lambda: deque(maxlen=MAX_PENDING_ATTESTATIONS)
)
"""
Attestations queued for replay after the next block lands.

An attestation referencing a not-yet-received block fails validation.

Buffering avoids dropping votes that arrived slightly out of order.

Bounded so overflow drops the oldest entry first.
"""

_pending_aggregated_attestations: list[SignedAggregatedAttestation] = field(
default_factory=list
_pending_aggregated_attestations: deque[SignedAggregatedAttestation] = field(
default_factory=lambda: deque(maxlen=MAX_PENDING_ATTESTATIONS)
)
"""
Aggregated attestations awaiting block processing.
Expand Down Expand Up @@ -478,8 +483,6 @@ async def on_gossip_attestation(
#
# Cap drops oldest on overflow: newer attestations are likelier to land soon.
self._pending_attestations.append(attestation)
if len(self._pending_attestations) > MAX_PENDING_ATTESTATIONS:
self._pending_attestations = self._pending_attestations[-MAX_PENDING_ATTESTATIONS:]

async def on_gossip_aggregated_attestation(
self,
Expand Down Expand Up @@ -522,10 +525,6 @@ async def on_gossip_aggregated_attestation(
#
# Cap drops oldest on overflow: newer aggregates are likelier to land soon.
self._pending_aggregated_attestations.append(signed_attestation)
if len(self._pending_aggregated_attestations) > MAX_PENDING_ATTESTATIONS:
self._pending_aggregated_attestations = self._pending_aggregated_attestations[
-MAX_PENDING_ATTESTATIONS:
]

def _replay_pending_attestations(self) -> None:
"""Retry buffered attestations after a block is processed."""
Expand All @@ -541,7 +540,7 @@ def _replay_pending_attestations(self) -> None:
# - B fails: T2 still missing, B is re-appended.
# Post-loop queue: [B].
pending = self._pending_attestations
self._pending_attestations = []
self._pending_attestations = deque(maxlen=MAX_PENDING_ATTESTATIONS)
for attestation in pending:
try:
self.store = self.spec.on_gossip_attestation(
Expand All @@ -555,7 +554,7 @@ def _replay_pending_attestations(self) -> None:

# Same mechanism for aggregated attestations.
pending_agg = self._pending_aggregated_attestations
self._pending_aggregated_attestations = []
self._pending_aggregated_attestations = deque(maxlen=MAX_PENDING_ATTESTATIONS)
for signed_attestation in pending_agg:
try:
self.store = self.spec.on_gossip_aggregated_attestation(
Expand Down
10 changes: 5 additions & 5 deletions tests/lean_spec/subspecs/sync/test_service.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -437,7 +437,7 @@ async def test_attestation_buffered_when_block_unknown(

await sync_service.on_gossip_attestation(attestation)

assert sync_service._pending_attestations == [attestation]
assert list(sync_service._pending_attestations) == [attestation]

async def test_buffered_attestation_replayed_after_block(
self,
Expand Down Expand Up @@ -468,7 +468,7 @@ async def test_buffered_attestation_replayed_after_block(
await sync_service.on_gossip_block(block, peer_id)

# Attestation was replayed (accepted by mock store).
assert sync_service._pending_attestations == []
assert list(sync_service._pending_attestations) == []
mock_store = cast(MockForkchoiceStore, sync_service.store)
assert attestation in mock_store._attestations_received

Expand Down Expand Up @@ -806,7 +806,7 @@ async def test_aggregated_buffered_on_key_error(
mock_store.reject_aggregated_attestation = lambda _att: True

await sync_service.on_gossip_aggregated_attestation(signed)
assert sync_service._pending_aggregated_attestations == [signed]
assert list(sync_service._pending_aggregated_attestations) == [signed]

async def test_replay_pending_mixed_success_and_failure(
self,
Expand All @@ -827,7 +827,7 @@ async def test_replay_pending_mixed_success_and_failure(
sync_service._replay_pending_attestations()

assert ok_att in mock_store._attestations_received
assert sync_service._pending_aggregated_attestations == [bad_signed]
assert list(sync_service._pending_aggregated_attestations) == [bad_signed]


class TestReplayPendingAttestationsPlain:
Expand Down Expand Up @@ -856,4 +856,4 @@ def test_replay_plain_mixed_success_and_failure(self, sync_service: SyncService)
sync_service._replay_pending_attestations()

assert ok_att in mock_store._attestations_received
assert sync_service._pending_attestations == [bad_att]
assert list(sync_service._pending_attestations) == [bad_att]
Loading