@@ -3,30 +3,37 @@ package driver
33import (
44 "context"
55 "fmt"
6+ gosync "sync"
67 "time"
78
89 "github.com/NethermindEth/juno/consensus/db"
910 "github.com/NethermindEth/juno/consensus/p2p"
11+ consensusSync "github.com/NethermindEth/juno/consensus/sync"
1012 "github.com/NethermindEth/juno/consensus/tendermint"
1113 "github.com/NethermindEth/juno/consensus/types"
1214 "github.com/NethermindEth/juno/consensus/types/actions"
15+ "github.com/NethermindEth/juno/p2p/sync"
1316 "github.com/NethermindEth/juno/utils"
1417)
1518
1619type TimeoutFn func (step types.Step , round types.Round ) time.Duration
1720
1821type Driver [V types.Hashable [H ], H types.Hash , A types.Addr ] struct {
19- log utils.Logger
20- db db.TendermintDB [V , H , A ]
21- stateMachine tendermint.StateMachine [V , H , A ]
22- commitListener CommitListener [V , H ]
23- broadcasters p2p.Broadcasters [V , H , A ]
24- listeners p2p.Listeners [V , H , A ]
25-
26- getTimeout TimeoutFn
22+ log utils.Logger
23+ db db.TendermintDB [V , H , A ]
24+ stateMachine tendermint.StateMachine [V , H , A ]
25+ commitListener CommitListener [V , H ]
26+ broadcasters p2p.Broadcasters [V , H , A ]
27+ listeners p2p.Listeners [V , H , A ]
28+ blockFetcher * sync.BlockFetcher
29+ messageExtractor * consensusSync.MessageExtractor [V , H , A ]
30+ getTimeout TimeoutFn
2731
2832 scheduledTms map [types.Timeout ]* time.Timer
2933 timeoutsCh chan types.Timeout
34+ syncListener chan sync.BlockBody
35+ lastQuorum types.Height
36+ wg gosync.WaitGroup
3037}
3138
3239func New [V types.Hashable [H ], H types.Hash , A types.Addr ](
@@ -36,18 +43,24 @@ func New[V types.Hashable[H], H types.Hash, A types.Addr](
3643 commitListener CommitListener [V , H ],
3744 broadcasters p2p.Broadcasters [V , H , A ],
3845 listeners p2p.Listeners [V , H , A ],
46+ blockFetcher * sync.BlockFetcher ,
47+ messageExtractor * consensusSync.MessageExtractor [V , H , A ],
3948 getTimeout TimeoutFn ,
4049) Driver [V , H , A ] {
4150 return Driver [V , H , A ]{
42- log : log ,
43- db : db ,
44- stateMachine : stateMachine ,
45- commitListener : commitListener ,
46- broadcasters : broadcasters ,
47- listeners : listeners ,
48- getTimeout : getTimeout ,
49- scheduledTms : make (map [types.Timeout ]* time.Timer ),
50- timeoutsCh : make (chan types.Timeout ),
51+ log : log ,
52+ db : db ,
53+ stateMachine : stateMachine ,
54+ commitListener : commitListener ,
55+ broadcasters : broadcasters ,
56+ listeners : listeners ,
57+ blockFetcher : blockFetcher ,
58+ messageExtractor : messageExtractor ,
59+ getTimeout : getTimeout ,
60+ scheduledTms : make (map [types.Timeout ]* time.Timer ),
61+ timeoutsCh : make (chan types.Timeout ),
62+ syncListener : make (chan sync.BlockBody , 1 ),
63+ wg : gosync.WaitGroup {},
5164 }
5265}
5366
@@ -84,7 +97,9 @@ func (d *Driver[V, H, A]) replay(ctx context.Context) error {
8497 return nil
8598}
8699
100+ //nolint:gocyclo // This is having higher complexity due to adding ok check for unmanaged channels.
87101func (d * Driver [V , H , A ]) listen (ctx context.Context ) error {
102+ defer d .wg .Wait ()
88103 for {
89104 select {
90105 case <- ctx .Done ():
@@ -123,13 +138,26 @@ func (d *Driver[V, H, A]) listen(ctx context.Context) error {
123138 return nil
124139 }
125140 actions = d .stateMachine .ProcessPrecommit (p )
141+ case p , ok := <- d .syncListener :
142+ if ! ok {
143+ return nil
144+ }
145+
146+ if p .Err != nil {
147+ d .syncCurrentHeight (ctx )
148+ } else {
149+ proposal , precommits := d .messageExtractor .Extract (& p )
150+ actions = d .stateMachine .ProcessSync (& proposal , precommits )
151+ }
126152 }
127153
128154 isCommitted , err = d .execute (ctx , false , actions )
129155 if err != nil {
130156 return err
131157 }
132158 }
159+
160+ d .syncCurrentHeight (ctx )
133161 }
134162}
135163
@@ -167,34 +195,76 @@ func (d *Driver[V, H, A]) execute(
167195 d .broadcasters .PrecommitBroadcaster .Broadcast (ctx , (* types.Precommit [H , A ])(action ))
168196
169197 case * actions.ScheduleTimeout :
170- d .setTimeout (ctx , types .Timeout (* action ))
198+ d .scheduleTimeout (ctx , types .Timeout (* action ))
171199
172200 case * actions.Commit [V , H , A ]:
173- d .log .Debugw ("Committing" , "height" , action .Height , "round" , action .Round )
174- d .commitListener .OnCommit (ctx , action .Height , * action .Value )
175-
176- if err := d .db .DeleteWALEntries (action .Height ); err != nil {
177- return true , fmt .Errorf ("failed to delete WAL messages during commit: %w" , err )
178- }
179-
180- return true , nil
201+ return true , d .commit (ctx , action )
181202
182203 case * actions.TriggerSync :
183- d .triggerSync (* action )
204+ d .triggerSync (ctx , * action )
184205 }
185206 }
186207 return false , nil
187208}
188209
189- func (d * Driver [V , H , A ]) triggerSync (sync actions.TriggerSync ) {
190- // TODO: Implement this
191- }
192-
193- func (d * Driver [V , H , A ]) setTimeout (ctx context.Context , timeout types.Timeout ) {
210+ func (d * Driver [V , H , A ]) scheduleTimeout (ctx context.Context , timeout types.Timeout ) {
194211 d .scheduledTms [timeout ] = time .AfterFunc (d .getTimeout (timeout .Step , timeout .Round ), func () {
195212 select {
196213 case <- ctx .Done ():
197214 case d .timeoutsCh <- timeout :
198215 }
199216 })
200217}
218+
219+ func (d * Driver [V , H , A ]) commit (ctx context.Context , commit * actions.Commit [V , H , A ]) error {
220+ d .log .Debugw ("Committing" , "height" , commit .Height , "round" , commit .Round )
221+ d .commitListener .OnCommit (ctx , commit .Height , * commit .Value )
222+
223+ if err := d .db .DeleteWALEntries (commit .Height ); err != nil {
224+ return fmt .Errorf ("failed to delete WAL messages during commit: %w" , err )
225+ }
226+
227+ return nil
228+ }
229+
230+ func (d * Driver [V , H , A ]) triggerSync (ctx context.Context , triggerSync actions.TriggerSync ) {
231+ currentlyHasFutureQuorum := d .hasFutureQuorum ()
232+ // TODO: Temporary workaround to only trigger the next height, because the sync process
233+ // doesn't support triggering multiple heights at once.
234+ d .lastQuorum = max (d .lastQuorum , triggerSync .End )
235+
236+ // Only trigger sync if haven't triggered yet.
237+ if ! currentlyHasFutureQuorum {
238+ d .syncCurrentHeight (ctx )
239+ }
240+ }
241+
242+ // TODO: Temporary workaround to only trigger the next height, because the sync process currently
243+ // doesn't support triggering multiple heights at once.
244+ func (d * Driver [V , H , A ]) syncCurrentHeight (ctx context.Context ) {
245+ height := d .stateMachine .Height ()
246+
247+ if ! d .hasFutureQuorum () {
248+ return
249+ }
250+
251+ d .wg .Go (func () {
252+ for {
253+ select {
254+ case <- ctx .Done ():
255+ return
256+ default :
257+ }
258+
259+ if err := d .blockFetcher .ProcessBlock (ctx , uint64 (height ), d .syncListener ); err != nil {
260+ d .log .Errorw ("failed to trigger sync" , "block" , height , "err" , err )
261+ } else {
262+ break
263+ }
264+ }
265+ })
266+ }
267+
268+ func (d * Driver [V , H , A ]) hasFutureQuorum () bool {
269+ return d .lastQuorum > d .stateMachine .Height ()
270+ }
0 commit comments