1- use std:: fmt:: { Display , Error as FmtError , Formatter } ;
1+ use std:: {
2+ fmt:: { Display , Error as FmtError , Formatter } ,
3+ time:: { Duration , Instant } ,
4+ } ;
25
36use aleph_primitives:: BlockNumber ;
47use futures:: StreamExt ;
8+ use log:: debug;
59use sc_client_api:: client:: { FinalityNotifications , ImportNotifications } ;
610use sp_runtime:: traits:: { Block as BlockT , Header as SubstrateHeader } ;
7- use tokio:: select;
11+ use tokio:: { select, time :: sleep } ;
812
9- use crate :: sync:: { ChainStatusNotification , ChainStatusNotifier } ;
13+ use crate :: sync:: {
14+ substrate:: chain_status:: Error as ChainStatusError , BlockIdentifier , ChainStatus ,
15+ ChainStatusNotification , ChainStatusNotifier , Header , SubstrateChainStatus , LOG_TARGET ,
16+ } ;
1017
1118/// What can go wrong when waiting for next chain status notification.
1219#[ derive( Debug ) ]
13- pub enum Error {
20+ pub enum Error < B >
21+ where
22+ B : BlockT ,
23+ B :: Header : SubstrateHeader < Number = BlockNumber > ,
24+ {
1425 JustificationStreamClosed ,
1526 ImportStreamClosed ,
27+ ChainStatus ( ChainStatusError < B > ) ,
1628}
1729
18- impl Display for Error {
30+ impl < B > Display for Error < B >
31+ where
32+ B : BlockT ,
33+ B :: Header : SubstrateHeader < Number = BlockNumber > ,
34+ {
1935 fn fmt ( & self , f : & mut Formatter < ' _ > ) -> Result < ( ) , FmtError > {
2036 use Error :: * ;
2137 match self {
@@ -25,30 +41,56 @@ impl Display for Error {
2541 ImportStreamClosed => {
2642 write ! ( f, "import notification stream has ended" )
2743 }
44+ ChainStatus ( e) => {
45+ write ! ( f, "chain status error: {}" , e)
46+ }
2847 }
2948 }
3049}
3150
32- /// Substrate specific implementation of `ChainStatusNotifier`.
51+ /// Substrate specific implementation of `ChainStatusNotifier`. If no blocks are reported through
52+ /// the usual channels for some time it falls back to reading the DB directly and produces
53+ /// notifications that way.
3354pub struct SubstrateChainStatusNotifier < B >
3455where
3556 B : BlockT ,
57+ B :: Header : SubstrateHeader < Number = BlockNumber > ,
3658{
3759 finality_notifications : FinalityNotifications < B > ,
3860 import_notifications : ImportNotifications < B > ,
61+ // The things below here are a hack to ensure all blocks get to the user, even during a major
62+ // sync. They should almost surely be removed after A0-1760.
63+ backend : SubstrateChainStatus < B > ,
64+ last_reported : BlockNumber ,
65+ trying_since : Instant ,
66+ catching_up : bool ,
3967}
4068
4169impl < B > SubstrateChainStatusNotifier < B >
4270where
4371 B : BlockT ,
72+ B :: Header : SubstrateHeader < Number = BlockNumber > ,
4473{
4574 pub fn new (
4675 finality_notifications : FinalityNotifications < B > ,
4776 import_notifications : ImportNotifications < B > ,
48- ) -> Self {
49- Self {
77+ backend : SubstrateChainStatus < B > ,
78+ ) -> Result < Self , ChainStatusError < B > > {
79+ let last_reported = backend. best_block ( ) ?. id ( ) . number ( ) ;
80+ Ok ( Self {
5081 finality_notifications,
5182 import_notifications,
83+ backend,
84+ last_reported,
85+ trying_since : Instant :: now ( ) ,
86+ catching_up : false ,
87+ } )
88+ }
89+
90+ fn header_at ( & self , number : BlockNumber ) -> Result < Option < B :: Header > , ChainStatusError < B > > {
91+ match self . backend . hash_for_number ( number) ? {
92+ Some ( hash) => Ok ( self . backend . header_for_hash ( hash) ?) ,
93+ None => Ok ( None ) ,
5294 }
5395 }
5496}
@@ -59,19 +101,52 @@ where
59101 B : BlockT ,
60102 B :: Header : SubstrateHeader < Number = BlockNumber > ,
61103{
62- type Error = Error ;
104+ type Error = Error < B > ;
63105
64106 async fn next ( & mut self ) -> Result < ChainStatusNotification < B :: Header > , Self :: Error > {
65- select ! {
66- maybe_block = self . finality_notifications. next( ) => {
67- maybe_block
68- . map( |block| ChainStatusNotification :: BlockFinalized ( block. header) )
69- . ok_or( Error :: JustificationStreamClosed )
70- } ,
71- maybe_block = self . import_notifications. next( ) => {
72- maybe_block
73- . map( |block| ChainStatusNotification :: BlockImported ( block. header) )
74- . ok_or( Error :: ImportStreamClosed )
107+ loop {
108+ if self . catching_up {
109+ match self
110+ . header_at ( self . last_reported + 1 )
111+ . map_err ( Error :: ChainStatus ) ?
112+ {
113+ Some ( header) => {
114+ self . last_reported += 1 ;
115+ return Ok ( ChainStatusNotification :: BlockImported ( header) ) ;
116+ }
117+ None => {
118+ self . catching_up = false ;
119+ self . trying_since = Instant :: now ( ) ;
120+ debug ! (
121+ target: LOG_TARGET ,
122+ "Manual reporting caught up, back to normal waiting for imports."
123+ ) ;
124+ }
125+ }
126+ }
127+ select ! {
128+ maybe_block = self . finality_notifications. next( ) => {
129+ self . trying_since = Instant :: now( ) ;
130+ return maybe_block
131+ . map( |block| ChainStatusNotification :: BlockFinalized ( block. header) )
132+ . ok_or( Error :: JustificationStreamClosed )
133+ } ,
134+ maybe_block = self . import_notifications. next( ) => {
135+ if let Some ( block) = & maybe_block {
136+ let number = block. header. id( ) . number( ) ;
137+ if number > self . last_reported {
138+ self . last_reported = number;
139+ }
140+ }
141+ self . trying_since = Instant :: now( ) ;
142+ return maybe_block
143+ . map( |block| ChainStatusNotification :: BlockImported ( block. header) )
144+ . ok_or( Error :: ImportStreamClosed )
145+ } ,
146+ _ = sleep( ( self . trying_since + Duration :: from_secs( 2 ) ) . saturating_duration_since( Instant :: now( ) ) ) => {
147+ self . catching_up = true ;
148+ debug!( target: LOG_TARGET , "No new blocks for 2 seconds, falling back to manual reporting." ) ;
149+ }
75150 }
76151 }
77152 }
0 commit comments