Skip to content

Commit b584139

Browse files
author
tac0turtle
committed
add a proposal for lazy
1 parent 3b33ced commit b584139

5 files changed

Lines changed: 296 additions & 46 deletions

File tree

‎block/aggregation.go‎

Lines changed: 87 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -2,6 +2,7 @@ package block
22

33
import (
44
"context"
5+
"errors"
56
"time"
67
)
78

@@ -49,30 +50,66 @@ func (m *Manager) lazyAggregationLoop(ctx context.Context, blockTimer *time.Time
4950
lazyTimer := time.NewTimer(0)
5051
defer lazyTimer.Stop()
5152

53+
// Initialize the throttle timer but don't start it yet
54+
m.txNotifyThrottle = time.NewTimer(m.minTxNotifyInterval)
55+
if !m.txNotifyThrottle.Stop() {
56+
<-m.txNotifyThrottle.C // Drain the channel if it already fired
57+
}
58+
defer m.txNotifyThrottle.Stop()
59+
60+
// Add a polling timer to periodically check for transactions
61+
// This is a fallback mechanism in case notifications fail
62+
txPollTimer := time.NewTimer(2 * time.Second)
63+
defer txPollTimer.Stop()
64+
5265
for {
5366
select {
5467
case <-ctx.Done():
5568
return
5669

5770
case <-lazyTimer.C:
71+
m.logger.Debug("Lazy timer triggered block production")
72+
m.produceBlock(ctx, "lazy_timer", lazyTimer, blockTimer)
73+
5874
case <-blockTimer.C:
59-
}
75+
m.logger.Debug("Block timer triggered block production")
76+
// m.produceBlock(ctx, "block_timer", lazyTimer, blockTimer)
77+
78+
case <-m.txNotifyCh:
79+
// Only proceed if we're not being throttled
80+
if time.Since(m.lastTxNotifyTime) < m.minTxNotifyInterval {
81+
m.logger.Debug("Transaction notification throttled")
82+
continue
83+
}
6084

61-
// Reset the start time
62-
start := time.Now()
85+
m.logger.Debug("Transaction notification triggered block production")
86+
m.produceBlock(ctx, "tx_notification", lazyTimer, blockTimer)
6387

64-
// Attempt to publish the block regardless of activity
65-
if err := m.publishBlock(ctx); err != nil && ctx.Err() == nil {
66-
m.logger.Error("error while publishing block", "error", err)
67-
}
88+
// Update the last notification time
89+
m.lastTxNotifyTime = time.Now()
6890

69-
// Reset both timers for the next aggregation window
70-
lazyTimer.Reset(getRemainingSleep(start, m.config.Node.LazyBlockInterval.Duration))
71-
blockTimer.Reset(getRemainingSleep(start, m.config.Node.BlockTime.Duration))
91+
case <-txPollTimer.C:
92+
// Check if there are transactions available
93+
hasTxs, err := m.checkForTransactions(ctx)
94+
if err != nil {
95+
m.logger.Error("Failed to check for transactions", "error", err)
96+
} else if hasTxs {
97+
m.logger.Debug("Transaction poll detected transactions")
98+
m.produceBlock(ctx, "tx_poll", lazyTimer, blockTimer)
99+
}
100+
txPollTimer.Reset(2 * time.Second)
101+
}
72102
}
73103
}
74104

75105
func (m *Manager) normalAggregationLoop(ctx context.Context, blockTimer *time.Timer) {
106+
// Initialize the throttle timer but don't start it yet
107+
m.txNotifyThrottle = time.NewTimer(m.minTxNotifyInterval)
108+
if !m.txNotifyThrottle.Stop() {
109+
<-m.txNotifyThrottle.C // Drain the channel if it already fired
110+
}
111+
defer m.txNotifyThrottle.Stop()
112+
76113
for {
77114
select {
78115
case <-ctx.Done():
@@ -87,10 +124,30 @@ func (m *Manager) normalAggregationLoop(ctx context.Context, blockTimer *time.Ti
87124
// Reset the blockTimer to signal the next block production
88125
// period based on the block time.
89126
blockTimer.Reset(getRemainingSleep(start, m.config.Node.BlockTime.Duration))
127+
128+
case <-m.txNotifyCh:
129+
// clear notification channel
90130
}
91131
}
92132
}
93133

134+
// produceBlock handles the common logic for producing a block and resetting timers
135+
func (m *Manager) produceBlock(ctx context.Context, trigger string, lazyTimer, blockTimer *time.Timer) {
136+
// Record the start time
137+
start := time.Now()
138+
139+
// Attempt to publish the block
140+
if err := m.publishBlock(ctx); err != nil && ctx.Err() == nil {
141+
m.logger.Error("error while publishing block", "trigger", trigger, "error", err)
142+
} else {
143+
m.logger.Debug("Successfully published block", "trigger", trigger)
144+
}
145+
146+
// Reset both timers for the next aggregation window
147+
lazyTimer.Reset(getRemainingSleep(start, m.config.Node.LazyBlockInterval.Duration))
148+
blockTimer.Reset(getRemainingSleep(start, m.config.Node.BlockTime.Duration))
149+
}
150+
94151
func getRemainingSleep(start time.Time, interval time.Duration) time.Duration {
95152
elapsed := time.Since(start)
96153

@@ -100,3 +157,23 @@ func getRemainingSleep(start time.Time, interval time.Duration) time.Duration {
100157

101158
return time.Millisecond
102159
}
160+
161+
// checkForTransactions checks if there are any transactions available for processing
162+
func (m *Manager) checkForTransactions(ctx context.Context) (bool, error) {
163+
// Try to retrieve a batch without actually consuming it
164+
batchData, err := m.retrieveBatch(ctx)
165+
if err != nil {
166+
if errors.Is(err, ErrNoBatch) {
167+
// This is expected when there are no transactions
168+
return false, nil
169+
}
170+
return false, err
171+
}
172+
173+
// If we got a batch with transactions, return true
174+
if batchData != nil && batchData.Batch != nil && len(batchData.Batch.Transactions) > 0 {
175+
return true, nil
176+
}
177+
178+
return false, nil
179+
}

‎block/lazy_aggregation_test.go‎

Lines changed: 127 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -226,3 +226,130 @@ func TestGetRemainingSleep(t *testing.T) {
226226
sleep4 := getRemainingSleep(start4, interval)
227227
assert.InDelta(interval, sleep4, float64(5*time.Millisecond), "Case 4 failed")
228228
}
229+
230+
// TestLazyAggregationLoop_TxNotification tests that transaction notifications trigger block production in lazy mode
231+
func TestLazyAggregationLoop_TxNotification(t *testing.T) {
232+
require := require.New(t)
233+
234+
blockTime := 200 * time.Millisecond
235+
lazyTime := 500 * time.Millisecond
236+
m, pubMock := setupTestManager(t, blockTime, lazyTime)
237+
m.config.Node.LazyMode = true
238+
239+
// Create the notification channel
240+
m.txNotifyCh = make(chan struct{}, 1)
241+
m.minTxNotifyInterval = 50 * time.Millisecond
242+
m.lastTxNotifyTime = time.Time{}
243+
244+
ctx, cancel := context.WithCancel(context.Background())
245+
defer cancel()
246+
247+
var wg sync.WaitGroup
248+
wg.Add(1)
249+
go func() {
250+
defer wg.Done()
251+
blockTimer := time.NewTimer(blockTime)
252+
defer blockTimer.Stop()
253+
m.lazyAggregationLoop(ctx, blockTimer)
254+
}()
255+
256+
// Wait a bit to ensure the loop is running
257+
time.Sleep(20 * time.Millisecond)
258+
259+
// Send a transaction notification
260+
m.NotifyNewTransactions()
261+
262+
// Wait for the block to be published
263+
select {
264+
case <-pubMock.calls:
265+
// Block was published, which is what we expect
266+
case <-time.After(100 * time.Millisecond):
267+
require.Fail("Block was not published after transaction notification")
268+
}
269+
270+
// Send another notification immediately - it should be throttled
271+
m.NotifyNewTransactions()
272+
273+
// No block should be published immediately due to throttling
274+
select {
275+
case <-pubMock.calls:
276+
require.Fail("Block was published despite throttling")
277+
case <-time.After(30 * time.Millisecond):
278+
// This is expected - no immediate block
279+
}
280+
281+
// Wait for the throttle period to pass
282+
time.Sleep(m.minTxNotifyInterval)
283+
284+
// Now a block should be published
285+
select {
286+
case <-pubMock.calls:
287+
// Block was published after throttle period
288+
case <-time.After(100 * time.Millisecond):
289+
require.Fail("Block was not published after throttle period")
290+
}
291+
292+
cancel()
293+
wg.Wait()
294+
}
295+
296+
// TestNormalAggregationLoop_TxNotification tests that transaction notifications are handled in normal mode
297+
func TestNormalAggregationLoop_TxNotification(t *testing.T) {
298+
require := require.New(t)
299+
300+
blockTime := 100 * time.Millisecond
301+
m, pubMock := setupTestManager(t, blockTime, 0)
302+
m.config.Node.LazyMode = false
303+
304+
// Create the notification channel
305+
m.txNotifyCh = make(chan struct{}, 1)
306+
m.minTxNotifyInterval = 50 * time.Millisecond
307+
m.lastTxNotifyTime = time.Time{}
308+
309+
ctx, cancel := context.WithCancel(context.Background())
310+
defer cancel()
311+
312+
var wg sync.WaitGroup
313+
wg.Add(1)
314+
go func() {
315+
defer wg.Done()
316+
blockTimer := time.NewTimer(blockTime)
317+
defer blockTimer.Stop()
318+
m.normalAggregationLoop(ctx, blockTimer)
319+
}()
320+
321+
// Wait for the first block to be published by the timer
322+
select {
323+
case <-pubMock.calls:
324+
// Block was published by timer, which is expected
325+
case <-time.After(blockTime * 2):
326+
require.Fail("Block was not published by timer")
327+
}
328+
329+
// Reset the publish mock to track new calls
330+
pubMock.reset()
331+
332+
// Send a transaction notification
333+
m.NotifyNewTransactions()
334+
335+
// In normal mode, the notification should not trigger an immediate block
336+
select {
337+
case <-pubMock.calls:
338+
// If we enable the optional enhancement to reset the timer, this might happen
339+
// But with the current implementation, this should not happen
340+
require.Fail("Block was published immediately after notification in normal mode")
341+
case <-time.After(blockTime / 2):
342+
// This is expected - no immediate block
343+
}
344+
345+
// Wait for the next regular block
346+
select {
347+
case <-pubMock.calls:
348+
// Block was published by timer, which is expected
349+
case <-time.After(blockTime * 2):
350+
require.Fail("Block was not published by timer after notification")
351+
}
352+
353+
cancel()
354+
wg.Wait()
355+
}

‎block/manager.go‎

Lines changed: 52 additions & 21 deletions
Original file line numberDiff line numberDiff line change
@@ -152,6 +152,18 @@ type Manager struct {
152152
// publishBlock is the function used to publish blocks. It defaults to
153153
// the manager's publishBlock method but can be overridden for testing.
154154
publishBlock publishBlockFunc
155+
156+
// txNotifyCh is used to signal when new transactions are available
157+
txNotifyCh chan struct{}
158+
159+
// txNotifyThrottle controls how frequently we can produce blocks in response to tx notifications
160+
txNotifyThrottle *time.Timer
161+
162+
// lastTxNotifyTime tracks when we last produced a block due to tx notification
163+
lastTxNotifyTime time.Time
164+
165+
// minTxNotifyInterval is the minimum time between tx-triggered block productions
166+
minTxNotifyInterval time.Duration
155167
}
156168

157169
// getInitialState tries to load lastState from Store, and if it's not available it reads genesis.
@@ -323,6 +335,9 @@ func NewManager(
323335
daH := atomic.Uint64{}
324336
daH.Store(s.DAHeight)
325337

338+
// Default minimum interval between tx-triggered blocks (can be made configurable)
339+
minTxNotifyInterval := 500 * time.Millisecond
340+
326341
agg := &Manager{
327342
signer: signer,
328343
config: config,
@@ -332,27 +347,30 @@ func NewManager(
332347
dalc: dalc,
333348
daHeight: &daH,
334349
// channels are buffered to avoid blocking on input/output operations, buffer sizes are arbitrary
335-
HeaderCh: make(chan *types.SignedHeader, channelLength),
336-
DataCh: make(chan *types.Data, channelLength),
337-
headerInCh: make(chan NewHeaderEvent, headerInChLength),
338-
dataInCh: make(chan NewDataEvent, headerInChLength),
339-
headerStoreCh: make(chan struct{}, 1),
340-
dataStoreCh: make(chan struct{}, 1),
341-
headerStore: headerStore,
342-
dataStore: dataStore,
343-
lastStateMtx: new(sync.RWMutex),
344-
lastBatchData: lastBatchData,
345-
headerCache: cache.NewCache[types.SignedHeader](),
346-
dataCache: cache.NewCache[types.Data](),
347-
retrieveCh: make(chan struct{}, 1),
348-
logger: logger,
349-
buildingBlock: false,
350-
pendingHeaders: pendingHeaders,
351-
metrics: seqMetrics,
352-
sequencer: sequencer,
353-
exec: exec,
354-
gasPrice: gasPrice,
355-
gasMultiplier: gasMultiplier,
350+
HeaderCh: make(chan *types.SignedHeader, channelLength),
351+
DataCh: make(chan *types.Data, channelLength),
352+
headerInCh: make(chan NewHeaderEvent, headerInChLength),
353+
dataInCh: make(chan NewDataEvent, headerInChLength),
354+
headerStoreCh: make(chan struct{}, 1),
355+
dataStoreCh: make(chan struct{}, 1),
356+
headerStore: headerStore,
357+
dataStore: dataStore,
358+
lastStateMtx: new(sync.RWMutex),
359+
lastBatchData: lastBatchData,
360+
headerCache: cache.NewCache[types.SignedHeader](),
361+
dataCache: cache.NewCache[types.Data](),
362+
retrieveCh: make(chan struct{}, 1),
363+
logger: logger,
364+
buildingBlock: false,
365+
pendingHeaders: pendingHeaders,
366+
metrics: seqMetrics,
367+
sequencer: sequencer,
368+
exec: exec,
369+
gasPrice: gasPrice,
370+
gasMultiplier: gasMultiplier,
371+
txNotifyCh: make(chan struct{}, 1), // Non-blocking channel
372+
minTxNotifyInterval: minTxNotifyInterval,
373+
lastTxNotifyTime: time.Time{}, // Zero time
356374
}
357375
agg.init(ctx)
358376
// Set the default publishBlock implementation
@@ -942,3 +960,16 @@ func (m *Manager) getSignature(header types.Header) (types.Signature, error) {
942960
}
943961
return m.signer.Sign(b)
944962
}
963+
964+
// NotifyNewTransactions signals that new transactions are available for processing
965+
// This method will be called by the Reaper when it receives new transactions
966+
func (m *Manager) NotifyNewTransactions() {
967+
// Non-blocking send to avoid slowing down the transaction submission path
968+
select {
969+
case m.txNotifyCh <- struct{}{}:
970+
// Successfully sent notification
971+
default:
972+
// Channel buffer is full, which means a notification is already pending
973+
// This is fine, as we just need to trigger one block production
974+
}
975+
}

0 commit comments

Comments
 (0)