@@ -3,6 +3,7 @@ package bft
33import (
44 "bytes"
55 "context"
6+ "crypto"
67 "encoding/hex"
78 "errors"
89 "fmt"
@@ -11,11 +12,10 @@ import (
1112 "sync/atomic"
1213 "time"
1314
14- "github.com/libp2p/go-libp2p/core/peer"
1515 "github.com/unicitynetwork/bft-core/network"
1616 "github.com/unicitynetwork/bft-core/network/protocol/certification"
1717 "github.com/unicitynetwork/bft-core/network/protocol/handshake"
18- "github.com/unicitynetwork/bft-go-base/crypto"
18+ cryptobft "github.com/unicitynetwork/bft-go-base/crypto"
1919 "github.com/unicitynetwork/bft-go-base/types"
2020
2121 "github.com/unicitynetwork/aggregator-go/internal/config"
@@ -39,12 +39,11 @@ type (
3939 shardID types.ShardID
4040 logger * logger.Logger
4141
42- // mutex for peer, network, signer, rootNodes
43- mu sync.Mutex
44- peer * network.Peer
45- network * BftNetwork
46- rootNodes peer.IDSlice
47- signer crypto.Signer
42+ // mutex for peer, network, signer TODO: there are readers without mutex
43+ mu sync.Mutex
44+ peer * network.Peer
45+ network * BftNetwork
46+ signer cryptobft.Signer
4847
4948 // Latest UC this node has seen. Can be ahead of the committed UC during recovery.
5049 luc atomic.Pointer [types.UnicityCertificate ]
@@ -58,6 +57,11 @@ type (
5857 ucProcessingMutex sync.Mutex
5958
6059 msgLoopCancelFn context.CancelFunc
60+
61+ // timestamp when last UC was received
62+ lastCertResponseTime atomic.Int64
63+
64+ trustBaseStore TrustBaseStore
6165 }
6266
6367 BFTClient interface {
@@ -71,19 +75,26 @@ type (
7175 StartNewRound (ctx context.Context , roundNumber * api.BigInt ) error
7276 }
7377
78+ TrustBaseStore interface {
79+ GetByEpoch (ctx context.Context , epoch uint64 ) (types.RootTrustBase , error )
80+ }
81+
7482 status int
7583)
7684
77- func NewBFTClient (ctx context. Context , conf * config.BFTConfig , roundManager RoundManager , logger * logger.Logger ) (* BFTClientImpl , error ) {
85+ func NewBFTClient (conf * config.BFTConfig , roundManager RoundManager , trustBaseStore TrustBaseStore , luc * types. UnicityCertificate , logger * logger.Logger ) (* BFTClientImpl , error ) {
7886 logger .Info ("Creating BFT Client" )
7987 bftClient := & BFTClientImpl {
80- logger : logger ,
81- partitionID : conf .ShardConf .PartitionID ,
82- shardID : conf .ShardConf .ShardID ,
83- roundManager : roundManager ,
84- conf : conf ,
88+ logger : logger ,
89+ partitionID : conf .ShardConf .PartitionID ,
90+ shardID : conf .ShardConf .ShardID ,
91+ roundManager : roundManager ,
92+ trustBaseStore : trustBaseStore ,
93+ conf : conf ,
8594 }
8695 bftClient .status .Store (idle )
96+ bftClient .luc .Store (luc )
97+ bftClient .lastCertResponseTime .Store (time .Now ().UnixMilli ())
8798 return bftClient , nil
8899}
89100
@@ -120,10 +131,6 @@ func (c *BFTClientImpl) Start(ctx context.Context) error {
120131 if err != nil {
121132 return err
122133 }
123- rootNodes , err := c .conf .GetRootNodes ()
124- if err != nil {
125- return fmt .Errorf ("failed to get root nodes: %w" , err )
126- }
127134 signer , err := c .conf .KeyConf .Signer ()
128135 if err != nil {
129136 return fmt .Errorf ("failed to create signer: %w" , err )
@@ -137,19 +144,20 @@ func (c *BFTClientImpl) Start(ctx context.Context) error {
137144 c .peer = self
138145 c .network = networkP2P
139146 c .signer = signer
140- c .rootNodes = rootNodes
141147
142148 if err := c .peer .BootstrapConnect (ctx , c .logger .Logger ); err != nil {
143149 return fmt .Errorf ("failed to bootstrap peer: %w" , err )
144150 }
145151
146- if err := c .sendHandshake (ctx ); err != nil {
147- return fmt .Errorf ("failed to send handshake: %w" , err )
148- }
149-
150152 msgLoopCtx , cancelFn := context .WithCancel (ctx )
151153 c .msgLoopCancelFn = cancelFn
152- go c .loop (msgLoopCtx )
154+ go func () {
155+ if err := c .loop (msgLoopCtx ); err != nil {
156+ c .logger .Error ("BFT event loop thread exited with error" , "error" , err .Error ())
157+ } else {
158+ c .logger .Info ("BFT event loop thread finished" )
159+ }
160+ }()
153161
154162 return nil
155163}
@@ -171,16 +179,21 @@ func (c *BFTClientImpl) Stop() {
171179 c .peer = nil
172180 c .network = nil
173181 c .signer = nil
174- c .rootNodes = nil
175182 }
176183}
177184
178185func (c * BFTClientImpl ) sendHandshake (ctx context.Context ) error {
179186 c .logger .WithContext (ctx ).Debug ("sending handshake to root chain" )
187+
188+ // load trust base
189+ rootEpoch := c .luc .Load ().GetRootEpoch ()
190+ tb , err := c .trustBaseStore .GetByEpoch (ctx , rootEpoch )
191+ if err != nil {
192+ return fmt .Errorf ("failed to load trust base for epoch %d: %w" , rootEpoch , err )
193+ }
180194 // select some random root nodes
181- rootIDs , err := randomNodeSelector (c . rootNodes , defaultHandshakeNodes )
195+ rootIDs , err := randomNodeSelector (tb , defaultHandshakeNodes )
182196 if err != nil {
183- // error should only happen in case the root nodes are not initialized
184197 return fmt .Errorf ("failed to select root nodes for handshake: %w" , err )
185198 }
186199 if err = c .network .Send (ctx ,
@@ -196,6 +209,13 @@ func (c *BFTClientImpl) sendHandshake(ctx context.Context) error {
196209}
197210
198211func (c * BFTClientImpl ) loop (ctx context.Context ) error {
212+ if err := c .sendHandshake (ctx ); err != nil {
213+ return fmt .Errorf ("failed to send initial handshake: %w" , err )
214+ }
215+
216+ heartbeat := time .NewTicker (c .conf .HeartbeatInterval )
217+ defer heartbeat .Stop ()
218+
199219 for {
200220 select {
201221 case <- ctx .Done ():
@@ -206,6 +226,15 @@ func (c *BFTClientImpl) loop(ctx context.Context) error {
206226 }
207227 c .logger .WithContext (ctx ).Debug ("received message" , "type" , fmt .Sprintf ("%T" , m ))
208228 c .handleMessage (ctx , m )
229+ case <- heartbeat .C :
230+ lastCertMillis := c .lastCertResponseTime .Load ()
231+ lastCertTime := time .UnixMilli (lastCertMillis )
232+ if time .Since (lastCertTime ) > c .conf .InactivityTimeout {
233+ c .logger .Warn ("BFT client inactivity timeout exceeded, sending new handshake" )
234+ if err := c .sendHandshake (ctx ); err != nil {
235+ c .logger .Error ("failed to send handshake on inactivity timeout" , "error" , err .Error ())
236+ }
237+ }
209238 }
210239 }
211240}
@@ -214,16 +243,30 @@ func (c *BFTClientImpl) handleMessage(ctx context.Context, msg any) {
214243 switch mt := msg .(type ) {
215244 case * certification.CertificationResponse :
216245 c .logger .WithContext (ctx ).Info ("received CertificationResponse" )
217- c .handleCertificationResponse (ctx , mt )
246+ if err := c .handleCertificationResponse (ctx , mt ); err != nil {
247+ c .logger .WithContext (ctx ).Error ("error processing CertificationResponse message" , "error" , err .Error ())
248+ }
218249 default :
219250 c .logger .WithContext (ctx ).Info ("received unknown message" )
220251 }
221252}
222253
223254func (c * BFTClientImpl ) handleCertificationResponse (ctx context.Context , cr * certification.CertificationResponse ) error {
255+ c .lastCertResponseTime .Store (time .Now ().UnixMilli ())
256+
224257 if err := cr .IsValid (); err != nil {
225258 return fmt .Errorf ("invalid CertificationResponse: %w" , err )
226259 }
260+
261+ // verify UC
262+ tb , err := c .trustBaseStore .GetByEpoch (ctx , cr .UC .GetRootEpoch ())
263+ if err != nil {
264+ return fmt .Errorf ("failed to load trust base for epoch %d: %w" , cr .UC .GetRootEpoch (), err )
265+ }
266+ if err := cr .UC .Verify (tb , crypto .SHA256 , c .partitionID , c .shardID , nil ); err != nil {
267+ return fmt .Errorf ("failed to verify UC: %w" , err )
268+ }
269+
227270 c .logger .WithContext (ctx ).Info (fmt .Sprintf ("handleCertificationResponse: UC round %d, next round %d, next leader %s" ,
228271 cr .UC .GetRoundNumber (), cr .Technical .Round , cr .Technical .Leader ))
229272
@@ -476,7 +519,12 @@ func (c *BFTClientImpl) sendCertificationRequest(ctx context.Context, rootHash s
476519 }
477520 c .logger .WithContext (ctx ).Info (fmt .Sprintf ("Round %d sending block certification request to root chain, IR hash %X" ,
478521 req .InputRecord .RoundNumber , req .InputRecord .Hash ))
479- rootIDs , err := rootNodesSelector (luc , c .rootNodes , defaultNofRootNodes )
522+
523+ tb , err := c .trustBaseStore .GetByEpoch (ctx , luc .GetRootEpoch ())
524+ if err != nil {
525+ return fmt .Errorf ("failed to load trust base for epoch %d: %w" , luc .GetRootEpoch (), err )
526+ }
527+ rootIDs , err := rootNodesSelector (luc , tb , defaultNofRootNodes )
480528 if err != nil {
481529 return fmt .Errorf ("selecting root nodes: %w" , err )
482530 }
0 commit comments