@@ -30,8 +30,12 @@ type DARetriever struct {
3030 logger zerolog.Logger
3131
3232 // calculate namespaces bytes once and reuse them
33- namespaceBz []byte
34- namespaceDataBz []byte
33+ namespaceBz []byte
34+ namespaceDataBz []byte
35+ namespaceForcedInclusionBz []byte
36+
37+ hasForcedInclusionNs bool
38+ daEpochSize uint64
3539
3640 // transient cache, only full event need to be passed to the syncer
3741 // on restart, will be refetch as da height is updated by syncer
@@ -47,15 +51,26 @@ func NewDARetriever(
4751 genesis genesis.Genesis ,
4852 logger zerolog.Logger ,
4953) * DARetriever {
54+ forcedInclusionNs := config .DA .GetForcedInclusionNamespace ()
55+ hasForcedInclusionNs := forcedInclusionNs != ""
56+
57+ var namespaceForcedInclusionBz []byte
58+ if hasForcedInclusionNs {
59+ namespaceForcedInclusionBz = coreda .NamespaceFromString (forcedInclusionNs ).Bytes ()
60+ }
61+
5062 return & DARetriever {
51- da : da ,
52- cache : cache ,
53- genesis : genesis ,
54- logger : logger .With ().Str ("component" , "da_retriever" ).Logger (),
55- namespaceBz : coreda .NamespaceFromString (config .DA .GetNamespace ()).Bytes (),
56- namespaceDataBz : coreda .NamespaceFromString (config .DA .GetDataNamespace ()).Bytes (),
57- pendingHeaders : make (map [uint64 ]* types.SignedHeader ),
58- pendingData : make (map [uint64 ]* types.Data ),
63+ da : da ,
64+ cache : cache ,
65+ genesis : genesis ,
66+ logger : logger .With ().Str ("component" , "da_retriever" ).Logger (),
67+ namespaceBz : coreda .NamespaceFromString (config .DA .GetNamespace ()).Bytes (),
68+ namespaceDataBz : coreda .NamespaceFromString (config .DA .GetDataNamespace ()).Bytes (),
69+ namespaceForcedInclusionBz : namespaceForcedInclusionBz ,
70+ hasForcedInclusionNs : hasForcedInclusionNs ,
71+ daEpochSize : config .DA .ForcedInclusionDAEpoch ,
72+ pendingHeaders : make (map [uint64 ]* types.SignedHeader ),
73+ pendingData : make (map [uint64 ]* types.Data ),
5974 }
6075}
6176
@@ -76,6 +91,40 @@ func (r *DARetriever) RetrieveFromDA(ctx context.Context, daHeight uint64) ([]co
7691 return r .processBlobs (ctx , blobsResp .Data , daHeight ), nil
7792}
7893
94+ // RetrieveForcedIncludedTxsFromDA retrieves forced inclusion transactions from the DA layer.
95+ // It fetches from the daHeight for the da epoch range defined in the config.
96+ func (r * DARetriever ) RetrieveForcedIncludedTxsFromDA (ctx context.Context , daHeight uint64 ) (* common.ForcedIncludedEvent , error ) {
97+ if ! r .hasForcedInclusionNs {
98+ return nil , fmt .Errorf ("forced inclusion namespace not configured" )
99+ }
100+
101+ event := & common.ForcedIncludedEvent {
102+ StartDaHeight : daHeight ,
103+ }
104+
105+ r .logger .Debug ().Uint64 ("da_height" , daHeight ).Uint64 ("range" , r .daEpochSize ).Msg ("retrieving forced included transactions from DA" )
106+
107+ for epochHeight := daHeight + 1 ; epochHeight <= daHeight + r .daEpochSize ; epochHeight ++ {
108+ result := types .RetrieveWithHelpers (ctx , r .da , r .logger , epochHeight , r .namespaceForcedInclusionBz , defaultDATimeout )
109+
110+ // quickly break if we are too ahead.
111+ if result .Code == coreda .StatusHeightFromFuture {
112+ break
113+ }
114+
115+ if result .Code == coreda .StatusSuccess {
116+ if err := r .validateBlobResponse (result , epochHeight ); ! errors .Is (err , coreda .ErrBlobNotFound ) && err != nil {
117+ return nil , err
118+ }
119+
120+ event .StartDaHeight = epochHeight
121+ event .Txs = append (event .Txs , result .Data ... )
122+ }
123+ }
124+
125+ return event , nil
126+ }
127+
79128// fetchBlobs retrieves blobs from the DA layer
80129func (r * DARetriever ) fetchBlobs (ctx context.Context , daHeight uint64 ) (coreda.ResultRetrieve , error ) {
81130 // Retrieve from both namespaces
0 commit comments