@@ -5,23 +5,29 @@ import (
55 "fmt"
66 "path/filepath"
77 "testing"
8+ "time"
89
910 "github.com/ipfs/go-datastore"
1011 badgerds "github.com/ipfs/go-ds-badger2"
1112 pebbleds "github.com/ipfs/go-ds-pebble"
13+ sdk "github.com/onflow/flow-go-sdk"
14+ sdkclient "github.com/onflow/flow-go-sdk/access/grpc"
15+ "github.com/onflow/flow/protobuf/go/flow/entities"
16+ "github.com/onflow/flow/protobuf/go/flow/executiondata"
1217 "github.com/rs/zerolog"
1318 "github.com/stretchr/testify/assert"
1419 "github.com/stretchr/testify/require"
1520 "github.com/stretchr/testify/suite"
21+ "google.golang.org/grpc/codes"
22+ "google.golang.org/grpc/status"
1623
24+ "github.com/onflow/flow-go/engine/common/rpc/convert"
1725 "github.com/onflow/flow-go/engine/ghost/client"
1826 "github.com/onflow/flow-go/integration/testnet"
1927 "github.com/onflow/flow-go/integration/tests/lib"
2028 "github.com/onflow/flow-go/model/flow"
2129 "github.com/onflow/flow-go/module/blobs"
2230 "github.com/onflow/flow-go/module/executiondatasync/execution_data"
23- "github.com/onflow/flow-go/module/metrics"
24- storage "github.com/onflow/flow-go/storage/badger"
2531 "github.com/onflow/flow-go/utils/unittest"
2632)
2733
@@ -158,62 +164,75 @@ func (s *ExecutionStateSyncSuite) executionStateSyncTest() {
158164 blockA := s .BlockState .WaitForHighestFinalizedProgress (s .T (), currentFinalized )
159165 s .T ().Logf ("got block height %v ID %v" , blockA .Header .Height , blockA .Header .ID ())
160166
161- // wait for the requested number of sealed blocks, then pause the network so we can inspect the dbs
162- s .BlockState .WaitForSealedHeight (s .T (), blockA .Header .Height + runBlocks )
163- s .net .StopContainers ()
164-
165- metrics := metrics .NewNoopCollector ()
166-
167- // start an execution data service using the Access Node's execution data db
168- an := s .net .ContainerByID (s .bridgeID )
169- anEds := s .nodeExecutionDataStore (an )
170-
171- // setup storage objects needed to get the execution data id
172- anDB , err := an .DB ()
173- require .NoError (s .T (), err , "could not open db" )
174-
175- anHeaders := storage .NewHeaders (metrics , anDB )
176- anResults := storage .NewExecutionResults (metrics , anDB )
167+ // Loop through checkBlocks and verify the execution data was downloaded correctly
168+ an := s .net .ContainerByName (testnet .PrimaryAN )
169+ anClient , err := an .SDKClient ()
170+ require .NoError (s .T (), err , "could not get access node testnet client" )
177171
178- // start an execution data service using the Observer Node's execution data db
179172 on := s .net .ContainerByName (s .observerName )
180- onEds := s .nodeExecutionDataStore (on )
181-
182- // setup storage objects needed to get the execution data id
183- onDB , err := on .DB ()
184- require .NoError (s .T (), err , "could not open db" )
173+ onClient , err := on .SDKClient ()
174+ require .NoError (s .T (), err , "could not get observer testnet client" )
185175
186- onHeaders := storage . NewHeaders ( metrics , onDB )
187- onResults := storage . NewExecutionResults ( metrics , onDB )
176+ ctx , cancel := context . WithTimeout ( s . ctx , 5 * time . Minute )
177+ defer cancel ( )
188178
189- // Loop through checkBlocks and verify the execution data was downloaded correctly
190179 for i := blockA .Header .Height ; i <= blockA .Header .Height + checkBlocks ; i ++ {
191- // access node
192- header , err := anHeaders .ByHeight (i )
193- require .NoError (s .T (), err , "%s: could not get header" , testnet .PrimaryAN )
180+ anBED , err := s .executionDataForHeight (ctx , anClient , i )
181+ require .NoError (s .T (), err , "could not get execution data from AN for height %v" , i )
182+
183+ onBED , err := s .executionDataForHeight (ctx , onClient , i )
184+ require .NoError (s .T (), err , "could not get execution data from ON for height %v" , i )
194185
195- result , err := anResults .ByBlockID (header .ID ())
196- require .NoError (s .T (), err , "%s: could not get sealed result" , testnet .PrimaryAN )
186+ assert .Equal (s .T (), anBED .BlockID , onBED .BlockID )
187+ }
188+ }
197189
198- ed , err := anEds .Get (s .ctx , result .ExecutionDataID )
199- if assert .NoError (s .T (), err , "%s: could not get execution data for height %v" , testnet .PrimaryAN , i ) {
200- s .T ().Logf ("%s: got execution data for height %d" , testnet .PrimaryAN , i )
201- assert .Equal (s .T (), header .ID (), ed .BlockID )
190+ // executionDataForHeight returns the execution data for the given height from the given node
191+ // It retries the request until the data is available or the context is canceled
192+ func (s * ExecutionStateSyncSuite ) executionDataForHeight (ctx context.Context , nodeClient * sdkclient.Client , height uint64 ) (* execution_data.BlockExecutionData , error ) {
193+ execDataClient := nodeClient .ExecutionDataRPCClient ()
194+
195+ var header * sdk.BlockHeader
196+ s .Require ().NoError (retryNotFound (ctx , 200 * time .Millisecond , func () error {
197+ var err error
198+ header , err = nodeClient .GetBlockHeaderByHeight (s .ctx , height )
199+ return err
200+ }), "could not get block header for block %d" , height )
201+
202+ var blockED * execution_data.BlockExecutionData
203+ s .Require ().NoError (retryNotFound (ctx , 200 * time .Millisecond , func () error {
204+ ed , err := execDataClient .GetExecutionDataByBlockID (s .ctx , & executiondata.GetExecutionDataByBlockIDRequest {
205+ BlockId : header .ID [:],
206+ EventEncodingVersion : entities .EventEncodingVersion_CCF_V0 ,
207+ })
208+ if err != nil {
209+ return err
202210 }
203211
204- // observer node
205- header , err = onHeaders .ByHeight (i )
206- require .NoError (s .T (), err , "%s: could not get header" , testnet .PrimaryON )
212+ blockED , err = convert .MessageToBlockExecutionData (ed .GetBlockExecutionData (), flow .Localnet .Chain ())
213+ s .Require ().NoError (err , "could not convert execution data" )
214+
215+ return err
216+ }), "could not get execution data for block %d" , height )
207217
208- result , err = onResults . ByID ( result . ID ())
209- require . NoError ( s . T (), err , "%s: could not get sealed result from ON`s storage" , testnet . PrimaryON )
218+ return blockED , nil
219+ }
210220
211- ed , err = onEds .Get (s .ctx , result .ExecutionDataID )
212- if assert .NoError (s .T (), err , "%s: could not get execution data for height %v" , testnet .PrimaryON , i ) {
213- s .T ().Logf ("%s: got execution data for height %d" , testnet .PrimaryON , i )
214- assert .Equal (s .T (), header .ID (), ed .BlockID )
221+ // retryNotFound retries the given function until it returns an error that is not NotFound or the context is canceled
222+ func retryNotFound (ctx context.Context , delay time.Duration , f func () error ) error {
223+ for ctx .Err () == nil {
224+ err := f ()
225+ if status .Code (err ) == codes .NotFound {
226+ select {
227+ case <- ctx .Done ():
228+ return ctx .Err ()
229+ case <- time .After (delay ):
230+ }
231+ continue
215232 }
233+ return err
216234 }
235+ return ctx .Err ()
217236}
218237
219238func (s * ExecutionStateSyncSuite ) nodeExecutionDataStore (node * testnet.Container ) execution_data.ExecutionDataStore {
0 commit comments