Skip to content

ci: fix panic in indexer #1402

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
Manav-Aggarwal merged 6 commits into from
Dec 8, 2023
Merged

ci: fix panic in indexer #1402

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
c265cbc
fix: use context with cancel in test
MSevey Dec 8, 2023
b2221bf
ci: remove use of t.Parallel() to reduce test panics
MSevey Dec 8, 2023
94324bd
fix: add ctx done check for long running for loop
MSevey Dec 8, 2023
ab4a48a
fix: fix data race in the mempool
MSevey Dec 8, 2023
3d03917
fix: fix data race in the mempool of notifiedTxsAvailable being accessed
MSevey Dec 8, 2023
d429d97
fix: remove deadlock mutex calls
MSevey Dec 8, 2023
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
3 changes: 3 additions & 0 deletions mempool/clist_mempool.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -29,6 +29,7 @@ type CListMempool struct {

// notify listeners (ie. consensus) when txs are available
notifiedTxsAvailable bool
txAvailMtx sync.Mutex
txsAvailable chan struct{} // fires once for each height, when the mempool is not empty

config *config.MempoolConfig
Expand Down Expand Up @@ -501,6 +502,8 @@ func (mem *CListMempool) TxsAvailable() <-chan struct{} {
}

func (mem *CListMempool) notifyTxsAvailable() {
mem.txAvailMtx.Lock()
defer mem.txAvailMtx.Unlock()
if mem.Size() == 0 {
panic("notified txs available but mempool is empty!")
}
Expand Down
5 changes: 3 additions & 2 deletions node/full_client_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -171,7 +171,9 @@ func TestGenesisChunked(t *testing.T) {
mockApp.On("InitChain", mock.Anything, mock.Anything).Return(&abci.ResponseInitChain{}, nil)
privKey, _, _ := crypto.GenerateEd25519Key(crand.Reader)
signingKey, _, _ := crypto.GenerateEd25519Key(crand.Reader)
n, _ := newFullNode(context.Background(), config.NodeConfig{DAAddress: MockServerAddr}, privKey, signingKey, proxy.NewLocalClientCreator(mockApp), genDoc, test.NewFileLogger(t))
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
n, _ := newFullNode(ctx, config.NodeConfig{DAAddress: MockServerAddr}, privKey, signingKey, proxy.NewLocalClientCreator(mockApp), genDoc, test.NewFileLogger(t))

rpc := NewFullClient(n)

Expand Down Expand Up @@ -926,7 +928,6 @@ func TestStatus(t *testing.T) {
}

func TestFutureGenesisTime(t *testing.T) {
t.Parallel()
assert := assert.New(t)
require := require.New(t)

Expand Down
34 changes: 19 additions & 15 deletions state/txindex/indexer_service.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -74,22 +74,26 @@ func (is *IndexerService) OnStart() error {
batch := NewBatch(numTxs)

for i := int64(0); i < numTxs; i++ {
msg2 := <-txsSub.Out()
txResult := msg2.Data().(types.EventDataTx).TxResult

if err = batch.Add(&txResult); err != nil {
is.Logger.Error(
"failed to add tx to batch",
"height", height,
"index", txResult.Index,
"err", err,
)

if is.terminateOnError {
if err := is.Stop(); err != nil {
is.Logger.Error("failed to stop", "err", err)
select {
case <-is.ctx.Done():
return
case msg2 := <-txsSub.Out():
txResult := msg2.Data().(types.EventDataTx).TxResult

if err = batch.Add(&txResult); err != nil {
is.Logger.Error(
"failed to add tx to batch",
"height", height,
"index", txResult.Index,
"err", err,
)

if is.terminateOnError {
if err := is.Stop(); err != nil {
is.Logger.Error("failed to stop", "err", err)
}
return
}
return
}
}
}
Expand Down