Skip to content

Commit d705b0d

Browse files
S1nusS1nus
authored
Add contexts to Celestia DataAvailabilityLayerClient (#623)
* added contexts to Celestia DataAvailabilityLayerClient * added context to mock, grpc, and celestia * pass contexts into DataAvailabilityLayerClient from BlockManager methods
1 parent 9df643f commit d705b0d

8 files changed

Lines changed: 48 additions & 42 deletions

File tree

‎block/manager.go‎

Lines changed: 6 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -345,7 +345,7 @@ func (m *Manager) RetrieveLoop(ctx context.Context) {
345345
for {
346346
daHeight := atomic.LoadUint64(&m.daHeight)
347347
m.logger.Debug("retrieve", "daHeight", daHeight)
348-
err := m.processNextDABlock()
348+
err := m.processNextDABlock(ctx)
349349
if err != nil {
350350
m.logger.Error("failed to retrieve block from DALC", "daHeight", daHeight, "errors", err.Error())
351351
break
@@ -358,15 +358,15 @@ func (m *Manager) RetrieveLoop(ctx context.Context) {
358358
}
359359
}
360360

361-
func (m *Manager) processNextDABlock() error {
361+
func (m *Manager) processNextDABlock(ctx context.Context) error {
362362
// TODO(tzdybal): extract configuration option
363363
maxRetries := 10
364364
daHeight := atomic.LoadUint64(&m.daHeight)
365365

366366
var err error
367367
m.logger.Debug("trying to retrieve block from DA", "daHeight", daHeight)
368368
for r := 0; r < maxRetries; r++ {
369-
blockResp, fetchErr := m.fetchBlock(daHeight)
369+
blockResp, fetchErr := m.fetchBlock(ctx, daHeight)
370370
if fetchErr != nil {
371371
err = multierr.Append(err, fetchErr)
372372
time.Sleep(100 * time.Millisecond)
@@ -381,9 +381,9 @@ func (m *Manager) processNextDABlock() error {
381381
return err
382382
}
383383

384-
func (m *Manager) fetchBlock(daHeight uint64) (da.ResultRetrieveBlocks, error) {
384+
func (m *Manager) fetchBlock(ctx context.Context, daHeight uint64) (da.ResultRetrieveBlocks, error) {
385385
var err error
386-
blockRes := m.retriever.RetrieveBlocks(daHeight)
386+
blockRes := m.retriever.RetrieveBlocks(ctx, daHeight)
387387
switch blockRes.Code {
388388
case da.StatusError:
389389
err = fmt.Errorf("failed to retrieve block: %s", blockRes.Message)
@@ -534,7 +534,7 @@ func (m *Manager) submitBlockToDA(ctx context.Context, block *types.Block) error
534534
submitted := false
535535
backoff := initialBackoff
536536
for attempt := 1; ctx.Err() == nil && !submitted && attempt <= maxSubmitAttempts; attempt++ {
537-
res := m.dalc.SubmitBlock(block)
537+
res := m.dalc.SubmitBlock(ctx, block)
538538
if res.Code == da.StatusSuccess {
539539
m.logger.Info("successfully submitted rollmint block to DA layer", "rollmintHeight", block.Header.Height, "daHeight", res.DAHeight)
540540
submitted = true

‎da/celestia/celestia.go‎

Lines changed: 6 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -62,7 +62,7 @@ func (c *DataAvailabilityLayerClient) Stop() error {
6262
}
6363

6464
// SubmitBlock submits a block to DA layer.
65-
func (c *DataAvailabilityLayerClient) SubmitBlock(block *types.Block) da.ResultSubmitBlock {
65+
func (c *DataAvailabilityLayerClient) SubmitBlock(ctx context.Context, block *types.Block) da.ResultSubmitBlock {
6666
blob, err := block.MarshalBinary()
6767
if err != nil {
6868
return da.ResultSubmitBlock{
@@ -73,7 +73,7 @@ func (c *DataAvailabilityLayerClient) SubmitBlock(block *types.Block) da.ResultS
7373
}
7474
}
7575

76-
txResponse, err := c.client.SubmitPFD(context.TODO(), c.namespaceID, blob, c.config.GasLimit)
76+
txResponse, err := c.client.SubmitPFD(ctx, c.namespaceID, blob, c.config.GasLimit)
7777

7878
if err != nil {
7979
return da.ResultSubmitBlock{
@@ -103,8 +103,8 @@ func (c *DataAvailabilityLayerClient) SubmitBlock(block *types.Block) da.ResultS
103103
}
104104

105105
// CheckBlockAvailability queries DA layer to check data availability of block at given height.
106-
func (c *DataAvailabilityLayerClient) CheckBlockAvailability(dataLayerHeight uint64) da.ResultCheckBlock {
107-
shares, err := c.client.NamespacedShares(context.TODO(), c.namespaceID, dataLayerHeight)
106+
func (c *DataAvailabilityLayerClient) CheckBlockAvailability(ctx context.Context, dataLayerHeight uint64) da.ResultCheckBlock {
107+
shares, err := c.client.NamespacedShares(ctx, c.namespaceID, dataLayerHeight)
108108
if err != nil {
109109
return da.ResultCheckBlock{
110110
BaseResult: da.BaseResult{
@@ -124,8 +124,8 @@ func (c *DataAvailabilityLayerClient) CheckBlockAvailability(dataLayerHeight uin
124124
}
125125

126126
// RetrieveBlocks gets a batch of blocks from DA layer.
127-
func (c *DataAvailabilityLayerClient) RetrieveBlocks(dataLayerHeight uint64) da.ResultRetrieveBlocks {
128-
data, err := c.client.NamespacedData(context.TODO(), c.namespaceID, dataLayerHeight)
127+
func (c *DataAvailabilityLayerClient) RetrieveBlocks(ctx context.Context, dataLayerHeight uint64) da.ResultRetrieveBlocks {
128+
data, err := c.client.NamespacedData(ctx, c.namespaceID, dataLayerHeight)
129129
if err != nil {
130130
return da.ResultRetrieveBlocks{
131131
BaseResult: da.BaseResult{

‎da/celestia/mock/server.go‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -92,7 +92,7 @@ func (s *Server) submit(w http.ResponseWriter, r *http.Request) {
9292
return
9393
}
9494

95-
res := s.mock.SubmitBlock(&block)
95+
res := s.mock.SubmitBlock(r.Context(), &block)
9696
code := 0
9797
if res.Code != da.StatusSuccess {
9898
code = 3
@@ -118,7 +118,7 @@ func (s *Server) shares(w http.ResponseWriter, r *http.Request) {
118118
return
119119
}
120120

121-
res := s.mock.RetrieveBlocks(height)
121+
res := s.mock.RetrieveBlocks(r.Context(), height)
122122
if res.Code != da.StatusSuccess {
123123
s.writeError(w, errors.New(res.Message))
124124
return
@@ -161,7 +161,7 @@ func (s *Server) data(w http.ResponseWriter, r *http.Request) {
161161
return
162162
}
163163

164-
res := s.mock.RetrieveBlocks(height)
164+
res := s.mock.RetrieveBlocks(r.Context(), height)
165165
if res.Code != da.StatusSuccess {
166166
s.writeError(w, errors.New(res.Message))
167167
return

‎da/da.go‎

Lines changed: 5 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,8 @@
11
package da
22

33
import (
4+
"context"
5+
46
"github.com/celestiaorg/rollmint/log"
57
"github.com/celestiaorg/rollmint/store"
68
"github.com/celestiaorg/rollmint/types"
@@ -69,15 +71,15 @@ type DataAvailabilityLayerClient interface {
6971
// SubmitBlock submits the passed in block to the DA layer.
7072
// This should create a transaction which (potentially)
7173
// triggers a state transition in the DA layer.
72-
SubmitBlock(block *types.Block) ResultSubmitBlock
74+
SubmitBlock(ctx context.Context, block *types.Block) ResultSubmitBlock
7375

7476
// CheckBlockAvailability queries DA layer to check data availability of block corresponding at given height.
75-
CheckBlockAvailability(dataLayerHeight uint64) ResultCheckBlock
77+
CheckBlockAvailability(ctx context.Context, dataLayerHeight uint64) ResultCheckBlock
7678
}
7779

7880
// BlockRetriever is additional interface that can be implemented by Data Availability Layer Client that is able to retrieve
7981
// block data from DA layer. This gives the ability to use it for block synchronization.
8082
type BlockRetriever interface {
8183
// RetrieveBlocks returns blocks at given data layer height from data availability layer.
82-
RetrieveBlocks(dataLayerHeight uint64) ResultRetrieveBlocks
84+
RetrieveBlocks(ctx context.Context, dataLayerHeight uint64) ResultRetrieveBlocks
8385
}

‎da/grpc/grpc.go‎

Lines changed: 6 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -74,8 +74,8 @@ func (d *DataAvailabilityLayerClient) Stop() error {
7474
}
7575

7676
// SubmitBlock proxies SubmitBlock request to gRPC server.
77-
func (d *DataAvailabilityLayerClient) SubmitBlock(block *types.Block) da.ResultSubmitBlock {
78-
resp, err := d.client.SubmitBlock(context.TODO(), &dalc.SubmitBlockRequest{Block: block.ToProto()})
77+
func (d *DataAvailabilityLayerClient) SubmitBlock(ctx context.Context, block *types.Block) da.ResultSubmitBlock {
78+
resp, err := d.client.SubmitBlock(ctx, &dalc.SubmitBlockRequest{Block: block.ToProto()})
7979
if err != nil {
8080
return da.ResultSubmitBlock{
8181
BaseResult: da.BaseResult{Code: da.StatusError, Message: err.Error()},
@@ -91,8 +91,8 @@ func (d *DataAvailabilityLayerClient) SubmitBlock(block *types.Block) da.ResultS
9191
}
9292

9393
// CheckBlockAvailability proxies CheckBlockAvailability request to gRPC server.
94-
func (d *DataAvailabilityLayerClient) CheckBlockAvailability(daHeight uint64) da.ResultCheckBlock {
95-
resp, err := d.client.CheckBlockAvailability(context.TODO(), &dalc.CheckBlockAvailabilityRequest{DAHeight: daHeight})
94+
func (d *DataAvailabilityLayerClient) CheckBlockAvailability(ctx context.Context, daHeight uint64) da.ResultCheckBlock {
95+
resp, err := d.client.CheckBlockAvailability(ctx, &dalc.CheckBlockAvailabilityRequest{DAHeight: daHeight})
9696
if err != nil {
9797
return da.ResultCheckBlock{BaseResult: da.BaseResult{Code: da.StatusError, Message: err.Error()}}
9898
}
@@ -103,8 +103,8 @@ func (d *DataAvailabilityLayerClient) CheckBlockAvailability(daHeight uint64) da
103103
}
104104

105105
// RetrieveBlocks proxies RetrieveBlocks request to gRPC server.
106-
func (d *DataAvailabilityLayerClient) RetrieveBlocks(daHeight uint64) da.ResultRetrieveBlocks {
107-
resp, err := d.client.RetrieveBlocks(context.TODO(), &dalc.RetrieveBlocksRequest{DAHeight: daHeight})
106+
func (d *DataAvailabilityLayerClient) RetrieveBlocks(ctx context.Context, daHeight uint64) da.ResultRetrieveBlocks {
107+
resp, err := d.client.RetrieveBlocks(ctx, &dalc.RetrieveBlocksRequest{DAHeight: daHeight})
108108
if err != nil {
109109
return da.ResultRetrieveBlocks{BaseResult: da.BaseResult{Code: da.StatusError, Message: err.Error()}}
110110
}

‎da/grpc/mockserv/mockserv.go‎

Lines changed: 6 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -36,13 +36,13 @@ type mockImpl struct {
3636
mock mock.DataAvailabilityLayerClient
3737
}
3838

39-
func (m *mockImpl) SubmitBlock(_ context.Context, request *dalc.SubmitBlockRequest) (*dalc.SubmitBlockResponse, error) {
39+
func (m *mockImpl) SubmitBlock(ctx context.Context, request *dalc.SubmitBlockRequest) (*dalc.SubmitBlockResponse, error) {
4040
var b types.Block
4141
err := b.FromProto(request.Block)
4242
if err != nil {
4343
return nil, err
4444
}
45-
resp := m.mock.SubmitBlock(&b)
45+
resp := m.mock.SubmitBlock(ctx, &b)
4646
return &dalc.SubmitBlockResponse{
4747
Result: &dalc.DAResponse{
4848
Code: dalc.StatusCode(resp.Code),
@@ -52,8 +52,8 @@ func (m *mockImpl) SubmitBlock(_ context.Context, request *dalc.SubmitBlockReque
5252
}, nil
5353
}
5454

55-
func (m *mockImpl) CheckBlockAvailability(_ context.Context, request *dalc.CheckBlockAvailabilityRequest) (*dalc.CheckBlockAvailabilityResponse, error) {
56-
resp := m.mock.CheckBlockAvailability(request.DAHeight)
55+
func (m *mockImpl) CheckBlockAvailability(ctx context.Context, request *dalc.CheckBlockAvailabilityRequest) (*dalc.CheckBlockAvailabilityResponse, error) {
56+
resp := m.mock.CheckBlockAvailability(ctx, request.DAHeight)
5757
return &dalc.CheckBlockAvailabilityResponse{
5858
Result: &dalc.DAResponse{
5959
Code: dalc.StatusCode(resp.Code),
@@ -63,8 +63,8 @@ func (m *mockImpl) CheckBlockAvailability(_ context.Context, request *dalc.Check
6363
}, nil
6464
}
6565

66-
func (m *mockImpl) RetrieveBlocks(context context.Context, request *dalc.RetrieveBlocksRequest) (*dalc.RetrieveBlocksResponse, error) {
67-
resp := m.mock.RetrieveBlocks(request.DAHeight)
66+
func (m *mockImpl) RetrieveBlocks(ctx context.Context, request *dalc.RetrieveBlocksRequest) (*dalc.RetrieveBlocksResponse, error) {
67+
resp := m.mock.RetrieveBlocks(ctx, request.DAHeight)
6868
blocks := make([]*rollmint.Block, len(resp.Blocks))
6969
for i := range resp.Blocks {
7070
blocks[i] = resp.Blocks[i].ToProto()

‎da/mock/mock.go‎

Lines changed: 5 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,7 @@
11
package mock
22

33
import (
4+
"context"
45
"encoding/binary"
56
"math/rand"
67
"sync/atomic"
@@ -68,7 +69,7 @@ func (m *DataAvailabilityLayerClient) Stop() error {
6869
// SubmitBlock submits the passed in block to the DA layer.
6970
// This should create a transaction which (potentially)
7071
// triggers a state transition in the DA layer.
71-
func (m *DataAvailabilityLayerClient) SubmitBlock(block *types.Block) da.ResultSubmitBlock {
72+
func (m *DataAvailabilityLayerClient) SubmitBlock(ctx context.Context, block *types.Block) da.ResultSubmitBlock {
7273
daHeight := atomic.LoadUint64(&m.daHeight)
7374
m.logger.Debug("Submitting block to DA layer!", "height", block.Header.Height, "dataLayerHeight", daHeight)
7475

@@ -97,13 +98,13 @@ func (m *DataAvailabilityLayerClient) SubmitBlock(block *types.Block) da.ResultS
9798
}
9899

99100
// CheckBlockAvailability queries DA layer to check data availability of block corresponding to given header.
100-
func (m *DataAvailabilityLayerClient) CheckBlockAvailability(daHeight uint64) da.ResultCheckBlock {
101-
blocksRes := m.RetrieveBlocks(daHeight)
101+
func (m *DataAvailabilityLayerClient) CheckBlockAvailability(ctx context.Context, daHeight uint64) da.ResultCheckBlock {
102+
blocksRes := m.RetrieveBlocks(ctx, daHeight)
102103
return da.ResultCheckBlock{BaseResult: da.BaseResult{Code: blocksRes.Code}, DataAvailable: len(blocksRes.Blocks) > 0}
103104
}
104105

105106
// RetrieveBlocks returns block at given height from data availability layer.
106-
func (m *DataAvailabilityLayerClient) RetrieveBlocks(daHeight uint64) da.ResultRetrieveBlocks {
107+
func (m *DataAvailabilityLayerClient) RetrieveBlocks(ctx context.Context, daHeight uint64) da.ResultRetrieveBlocks {
107108
if daHeight >= atomic.LoadUint64(&m.daHeight) {
108109
return da.ResultRetrieveBlocks{BaseResult: da.BaseResult{Code: da.StatusError, Message: "block not found"}}
109110
}

‎da/test/da_test.go‎

Lines changed: 11 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,7 @@
11
package test
22

33
import (
4+
"context"
45
"encoding/json"
56
"math/rand"
67
"net"
@@ -72,6 +73,7 @@ func TestDALC(t *testing.T) {
7273
func doTestDALC(t *testing.T, dalc da.DataAvailabilityLayerClient) {
7374
require := require.New(t)
7475
assert := assert.New(t)
76+
ctx := context.Background()
7577

7678
// mock DALC will advance block height every 100ms
7779
conf := []byte{}
@@ -99,27 +101,27 @@ func doTestDALC(t *testing.T, dalc da.DataAvailabilityLayerClient) {
99101
b1 := getRandomBlock(1, 10)
100102
b2 := getRandomBlock(2, 10)
101103

102-
resp := dalc.SubmitBlock(b1)
104+
resp := dalc.SubmitBlock(ctx, b1)
103105
h1 := resp.DAHeight
104106
assert.Equal(da.StatusSuccess, resp.Code)
105107

106-
resp = dalc.SubmitBlock(b2)
108+
resp = dalc.SubmitBlock(ctx, b2)
107109
h2 := resp.DAHeight
108110
assert.Equal(da.StatusSuccess, resp.Code)
109111

110112
// wait a bit more than mockDaBlockTime, so rollmint blocks can be "included" in mock block
111113
time.Sleep(mockDaBlockTime + 20*time.Millisecond)
112114

113-
check := dalc.CheckBlockAvailability(h1)
115+
check := dalc.CheckBlockAvailability(ctx, h1)
114116
assert.Equal(da.StatusSuccess, check.Code)
115117
assert.True(check.DataAvailable)
116118

117-
check = dalc.CheckBlockAvailability(h2)
119+
check = dalc.CheckBlockAvailability(ctx, h2)
118120
assert.Equal(da.StatusSuccess, check.Code)
119121
assert.True(check.DataAvailable)
120122

121123
// this height should not be used by DALC
122-
check = dalc.CheckBlockAvailability(h1 - 1)
124+
check = dalc.CheckBlockAvailability(ctx, h1-1)
123125
assert.Equal(da.StatusSuccess, check.Code)
124126
assert.False(check.DataAvailable)
125127
}
@@ -173,6 +175,7 @@ func startMockCelestiaNodeServer(t *testing.T) *cmock.Server {
173175
}
174176

175177
func doTestRetrieve(t *testing.T, dalc da.DataAvailabilityLayerClient) {
178+
ctx := context.Background()
176179
require := require.New(t)
177180
assert := assert.New(t)
178181

@@ -204,7 +207,7 @@ func doTestRetrieve(t *testing.T, dalc da.DataAvailabilityLayerClient) {
204207

205208
for i := uint64(0); i < 100; i++ {
206209
b := getRandomBlock(i, rand.Int()%20)
207-
resp := dalc.SubmitBlock(b)
210+
resp := dalc.SubmitBlock(ctx, b)
208211
assert.Equal(da.StatusSuccess, resp.Code, resp.Message)
209212
time.Sleep(time.Duration(rand.Int63() % mockDaBlockTime.Milliseconds()))
210213

@@ -217,14 +220,14 @@ func doTestRetrieve(t *testing.T, dalc da.DataAvailabilityLayerClient) {
217220

218221
for h, cnt := range countAtHeight {
219222
t.Log("Retrieving block, DA Height", h)
220-
ret := retriever.RetrieveBlocks(h)
223+
ret := retriever.RetrieveBlocks(ctx, h)
221224
assert.Equal(da.StatusSuccess, ret.Code, ret.Message)
222225
require.NotEmpty(ret.Blocks, h)
223226
assert.Len(ret.Blocks, cnt, h)
224227
}
225228

226229
for b, h := range blocks {
227-
ret := retriever.RetrieveBlocks(h)
230+
ret := retriever.RetrieveBlocks(ctx, h)
228231
assert.Equal(da.StatusSuccess, ret.Code, h)
229232
require.NotEmpty(ret.Blocks, h)
230233
assert.Contains(ret.Blocks, b, h)

0 commit comments

Comments
 (0)