Skip to content
Merged
Show file tree
Hide file tree
Changes from 1 commit
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Prev Previous commit
Next Next commit
Review feedback
  • Loading branch information
alpe committed Mar 16, 2026
commit 86359c39938c32b821f2ef18f4f75a292ea904c5
1 change: 1 addition & 0 deletions block/internal/da/async_block_retriever.go
Original file line number Diff line number Diff line change
Expand Up @@ -240,6 +240,7 @@ func (f *asyncBlockRetriever) fetchAndCacheBlock(ctx context.Context, height uin
Uint64("height", height).
Str("status", result.Message).
Msg("failed to retrieve block - will retry")
return fmt.Errorf("retrieve block at height %d: %s", height, result.Message)
}
return nil
}
Comment thread
coderabbitai[bot] marked this conversation as resolved.
Resolved
Expand Down
47 changes: 47 additions & 0 deletions block/internal/da/async_block_retriever_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -262,6 +262,53 @@ func TestAsyncBlockRetriever_ReconnectOnSubscriptionError(t *testing.T) {
assert.Equal(t, []byte("reconnect-tx"), block.Blobs[0])
}

func TestAsyncBlockRetriever_FetchAndCacheBlock_ErrorResponse(t *testing.T) {
fiNs := datypes.NamespaceFromString("test-fi-ns").Bytes()
specs := map[string]struct {
code datypes.StatusCode
message string
wantErr string
}{
"status_error": {
code: datypes.StatusError,
message: "rpc failure",
wantErr: "retrieve block at height 123: rpc failure",
},
"status_context_deadline": {
code: datypes.StatusContextDeadline,
message: "timeout",
wantErr: "retrieve block at height 123: timeout",
},
}

for name, spec := range specs {
t.Run(name, func(t *testing.T) {
client := &mocks.MockClient{}
client.
On("Retrieve", mock.Anything, uint64(123), fiNs).
Return(datypes.ResultRetrieve{
BaseResult: datypes.BaseResult{
Code: spec.code,
Message: spec.message,
},
}).
Once()

logger := zerolog.Nop()
fetcher := NewAsyncBlockRetriever(client, logger, fiNs, 100*time.Millisecond, 100, 10).(*asyncBlockRetriever)

err := fetcher.fetchAndCacheBlock(t.Context(), 123)
require.EqualError(t, err, spec.wantErr)

block, getErr := fetcher.GetCachedBlock(t.Context(), 123)
require.NoError(t, getErr)
assert.Nil(t, block)

client.AssertExpectations(t)
})
}
}

func TestBlockData_Serialization(t *testing.T) {
block := &BlockData{
Height: 100,
Expand Down
6 changes: 4 additions & 2 deletions block/internal/syncing/syncer.go
Original file line number Diff line number Diff line change
Expand Up @@ -236,11 +236,13 @@ func (s *Syncer) Stop(ctx context.Context) error {
s.cancel()
s.cancelP2PWait(0)

// Stop the DA follower first (it owns its own goroutines).
if s.fiRetriever != nil {
s.fiRetriever.Stop()
}

if s.daFollower != nil {
s.daFollower.Stop()
}

s.wg.Wait()

// Skip draining if we're shutting down due to a critical error (e.g. execution
Expand Down
Loading