I create a test case to show this situation, 60 message will exceed the headBuf's size (which is 4096*10)，and trigger the truncate, cause SearchForEndHeight failed to search height 50.

func MakeNBlocksWhitWAL(wal *baseWAL, numBlocks int) (err error) {
	config := getConfig()

	app := kvstore.NewPersistentKVStoreApplication(filepath.Join(config.DBDir(), "wal_generator"))

	logger := log.TestingLogger().With("wal_generator", "wal_generator")
	logger.Info("generating WAL (last height msg excluded)", "numBlocks", numBlocks)

	/////////////////////////////////////////////////////////////////////////////
	// COPY PASTE FROM node.go WITH A FEW MODIFICATIONS
	// NOTE: we can't import node package because of circular dependency
	privValidatorFile := config.PrivValidatorFile()
	privValidator := privval.LoadOrGenFilePV(privValidatorFile)
	genDoc, err := tmtypes.GenesisDocFromFile(config.GenesisFile())
	if err != nil {
		return errors.Wrap(err, "failed to read genesis file")
	}
	stateDB := db.NewMemDB()
	blockStoreDB := db.NewMemDB()
	state, err := sm.MakeGenesisState(genDoc)
	if err != nil {
		return errors.Wrap(err, "failed to make genesis state")
	}
	blockStore := bc.NewBlockStore(blockStoreDB)
	handshaker := NewHandshaker(stateDB, state, blockStore, genDoc)
	proxyApp := proxy.NewAppConns(proxy.NewLocalClientCreator(app), handshaker)
	proxyApp.SetLogger(logger.With("module", "proxy"))
	if err := proxyApp.Start(); err != nil {
		return errors.Wrap(err, "failed to start proxy app connections")
	}
	defer proxyApp.Stop()
	eventBus := tmtypes.NewEventBus()
	eventBus.SetLogger(logger.With("module", "events"))
	if err := eventBus.Start(); err != nil {
		return errors.Wrap(err, "failed to start event bus")
	}
	defer eventBus.Stop()
	mempool := sm.MockMempool{}
	evpool := sm.MockEvidencePool{}
	blockExec := sm.NewBlockExecutor(stateDB, log.TestingLogger(), proxyApp.Consensus(), mempool, evpool)
	consensusState := NewConsensusState(config.Consensus, state.Copy(), blockExec, blockStore, mempool, evpool)
	consensusState.SetLogger(logger)
	consensusState.SetEventBus(eventBus)
	if privValidator != nil {
		consensusState.SetPrivValidator(privValidator)
	}
	// END OF COPY PASTE
	/////////////////////////////////////////////////////////////////////////////

	// set consensus wal to buffered WAL, which will write all incoming msgs to buffer
	numBlocksWritten := make(chan struct{})
	consensusState.wal = newByteBufferWAL(logger, NewWALEncoder(wal.group), int64(numBlocks), numBlocksWritten)
	// see wal.go#103
	wal.Write(EndHeightMessage{0})

	if err := consensusState.Start(); err != nil {
		return errors.Wrap(err, "failed to start consensus state")
	}

	select {
	case <-numBlocksWritten:
		consensusState.Stop()
		wal.group.Flush()
		return nil
	case <-time.After(1 * time.Minute):
		consensusState.Stop()
		return fmt.Errorf("waited too long for tendermint to produce %d blocks (grep logs for `wal_generator`)", numBlocks)
	}
}

func processTicks(g *autofile.Group) {
	for {
		size, err := g.Head.Size()
		if err != nil {
			panic(err)
		}
		if size >= 4096 {
			g.RotateFile()
		}
	}
}

func TestWALResidual(t *testing.T) {
	walDir, err := ioutil.TempDir("", "wal")
	if err != nil {
		panic(fmt.Errorf("failed to create temp WAL file: %v", err))
	}

	walFile := filepath.Join(walDir, "wal")

	wal, err := NewWAL(walFile)
	if err != nil {
		t.Fatal(err)
	}

	go processTicks(wal.group)

	err = MakeNBlocksWhitWAL(wal, 60)
	if err != nil {
		t.Fatal(err)
	}

	h := int64(50)
	gr, found, err := wal.SearchForEndHeight(h, &WALSearchOptions{})
	assert.NoError(t, err, fmt.Sprintf("expected not to err on height %d", h))
	assert.True(t, found, fmt.Sprintf("expected to find end height for %d", h))
	assert.NotNil(t, gr, "expected group not to be nil")
	defer gr.Close()

	dec := NewWALDecoder(gr)
	msg, err := dec.Decode()
	assert.NoError(t, err, "expected to decode a message")
	rs, ok := msg.Msg.(tmtypes.EventDataRoundState)
	assert.True(t, ok, "expected message of type EventDataRoundState")
	assert.Equal(t, rs.Height, h+1, fmt.Sprintf("wrong height"))
}

The modified version RotateFile will pass it.

// RotateFile causes group to close the current head and assign it some index.
// Note it does not create a new head.
func (g *Group) RotateFile() {
	g.mtx.Lock()
	defer g.mtx.Unlock()

	headPath := g.Head.Path

	err := g.headBuf.Flush()
	if err = g.Head.closeFile(); err != nil {
		panic(err)
	}

	if err = g.Head.Sync(); err != nil {
		panic(err)
	}

	if err = g.Head.closeFile(); err != nil {
		panic(err)
	}

	indexPath := filePathForIndex(headPath, g.maxIndex, g.maxIndex+1)
	if err := os.Rename(headPath, indexPath); err != nil {
		panic(err)
	}

	g.maxIndex++
}

Group.RotateFile need call Flush() before rename. #2428

Description

Metadata

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

Issue actions