Merge branch 'v0.34.x-celestia' into v0.34.x-celestia

consensus/state.go

@@ -538,6 +538,7 @@ func (cs *State) updateRoundStep(round int32, step cstypes.RoundStepType) {
 		}
 		if cs.Step != step {
 			cs.metrics.MarkStep(cs.Step)
+			schema.WriteRoundState(cs.traceClient, cs.Height, round, uint8(step))
 		}
 	}
 	cs.Round = round
@@ -706,10 +707,6 @@ func (cs *State) newStep() {
 
 	cs.nSteps++
 
-	step := uint8(cs.RoundState.Step)
-
-	schema.WriteRoundState(cs.traceClient, cs.Height, cs.Round, step)
-
 	// newStep is called by updateToState in NewState before the eventBus is set!
 	if cs.eventBus != nil {
 		if err := cs.eventBus.PublishEventNewRoundStep(rs); err != nil {
@@ -1518,6 +1515,7 @@ func (cs *State) enterPrecommitWait(height int64, round int32) {
 	defer func() {
 		// Done enterPrecommitWait:
 		cs.TriggeredTimeoutPrecommit = true
+		cs.updateRoundStep(round, cstypes.RoundStepPrecommitWait)
 		cs.newStep()
 	}()
 

mempool/cat/reactor.go

@@ -241,7 +241,7 @@ func (memR *Reactor) ReceiveEnvelope(e p2p.Envelope) {
 		for _, tx := range protoTxs {
 			ntx := types.Tx(tx)
 			key := ntx.Key()
-			schema.WriteMempoolTx(memR.traceClient, string(e.Src.ID()), key[:], schema.Download)
+			schema.WriteMempoolTx(memR.traceClient, string(e.Src.ID()), key[:], len(tx), schema.Download)
 			// If we requested the transaction we mark it as received.
 			if memR.requests.Has(peerID, key) {
 				memR.requests.MarkReceived(peerID, key)
@@ -331,6 +331,7 @@ func (memR *Reactor) ReceiveEnvelope(e p2p.Envelope) {
 					memR.traceClient,
 					string(e.Src.ID()),
 					txKey[:],
+					len(tx),
 					schema.Upload,
 				)
 			}

mempool/v1/reactor.go

@@ -197,6 +197,7 @@ func (memR *Reactor) ReceiveEnvelope(e p2p.Envelope) {
 				memR.traceClient,
 				string(e.Src.ID()),
 				ntx.Hash(),
+				len(tx),
 				schema.Download,
 			)
 			err = memR.mempool.CheckTx(ntx, nil, txInfo)
@@ -303,6 +304,7 @@ func (memR *Reactor) broadcastTxRoutine(peer p2p.Peer) {
 					memR.traceClient,
 					string(peer.ID()),
 					memTx.tx.Hash(),
+					len(memTx.tx),
 					schema.Upload,
 				)
 			}

pkg/trace/schema/mempool.go

@@ -35,7 +35,7 @@ func (m MempoolTx) Table() string {
 
 // WriteMempoolTx writes a tracing point for a tx using the predetermined
 // schema for mempool tracing.
-func WriteMempoolTx(client trace.Tracer, peer string, txHash []byte, transferType TransferType) {
+func WriteMempoolTx(client trace.Tracer, peer string, txHash []byte, size int, transferType TransferType) {
 	// this check is redundant to what is checked during client.Write, although it
 	// is an optimization to avoid allocations from the map of fields.
 	if !client.IsCollecting(MempoolTxTable) {
@@ -44,7 +44,7 @@ func WriteMempoolTx(client trace.Tracer, peer string, txHash []byte, transferTyp
 	client.Write(MempoolTx{
 		TxHash:       bytes.HexBytes(txHash).String(),
 		Peer:         peer,
-		Size:         len(txHash),
+		Size:         size,
 		TransferType: transferType,
 	})
 }