fix: trace haves and wants correctly (#2693)

fixes some traces for the prop reactor

---------

Co-authored-by: CHAMI Rachid <chamirachid1@gmail.com>

Author: evan-forbes

consensus/propagation/commitment.go

@@ -127,6 +127,7 @@ func (blockProp *Reactor) ProposeBlock(proposal *types.Proposal, block *types.Pa
 			peer.consensusPeerState.SetHasProposalBlockPart(proposal.Height, proposal.Round, int(part.GetIndex()))
 		}
 
+		schema.WriteBlockPartState(blockProp.traceClient, proposal.Height, proposal.Round, chunks[index].GetTrueIndices(), true, string(peer.peer.ID()), schema.Upload, "have")
 	}
 	return nil
 }

consensus/propagation/have_wants.go

@@ -296,6 +296,17 @@ func (blockProp *Reactor) sendWant(ps *PeerState, want *proptypes.WantParts) {
 		return
 	}
 
+	schema.WriteBlockPartState(
+		blockProp.traceClient,
+		want.Height,
+		want.Round,
+		want.Parts.GetTrueIndices(),
+		false,
+		string(ps.peer.ID()),
+		schema.Upload,
+		"want",
+	)
+
 	// keep track of the parts that this node has requested.
 	ps.AddRequests(want.Height, want.Round, want.Parts)
 	ps.IncreaseConcurrentReqs(int64(len(want.Parts.GetTrueIndices())))

consensus/propagation/reactor.go

@@ -272,6 +272,12 @@ func (blockProp *Reactor) ReceiveEnvelope(e p2p.Envelope) {
 			schema.WriteProposal(blockProp.traceClient, msg.Proposal.Height, msg.Proposal.Round, string(e.Src.ID()), schema.Download)
 		case *proptypes.HaveParts:
 			blockProp.handleHaves(e.Src.ID(), msg)
+			// Trace the received HaveParts message
+			indexes := make([]int, len(msg.Parts))
+			for i, part := range msg.Parts {
+				indexes[i] = int(part.Index)
+			}
+			schema.WriteBlockPartState(blockProp.traceClient, msg.Height, msg.Round, indexes, true, string(e.Src.ID()), schema.Download, "have")
 		case *proptypes.RecoveryPart:
 			schema.WriteBlockPart(blockProp.traceClient, msg.Height, msg.Round, msg.Index, false, string(e.Src.ID()), schema.Download)
 			blockProp.handleRecoveryPart(e.Src.ID(), msg)
@@ -282,6 +288,8 @@ func (blockProp *Reactor) ReceiveEnvelope(e p2p.Envelope) {
 		switch msg := msg.(type) {
 		case *proptypes.WantParts:
 			blockProp.handleWants(e.Src.ID(), msg)
+			// Trace the received WantParts message
+			schema.WriteBlockPartState(blockProp.traceClient, msg.Height, msg.Round, msg.Parts.GetTrueIndices(), false, string(e.Src.ID()), schema.Download, "want")
 		}
 	default:
 		blockProp.Logger.Error(fmt.Sprintf("Unknown chId %X", e.ChannelID))

consensus/state.go

@@ -176,6 +176,10 @@ type State struct {
 
 	// gossipDataEnabled controls whether the gossipDataRoutine should run
 	gossipDataEnabled atomic.Bool
+
+	// proposalReceivedTime tracks when the proposal was received for the current height/round
+	// Used to calculate the duration until full block is received
+	proposalReceivedTime time.Time
 }
 
 // StateOption sets an optional parameter on the State.
@@ -876,6 +880,7 @@ func (cs *State) updateToState(state sm.State) {
 	cs.rs.Proposal = nil
 	cs.rs.ProposalBlock = nil
 	cs.rs.ProposalBlockParts = nil
+	cs.proposalReceivedTime = time.Time{}
 	cs.rs.LockedRound = -1
 	cs.rs.LockedBlock = nil
 	cs.rs.LockedBlockParts = nil
@@ -1244,6 +1249,7 @@ func (cs *State) enterNewRound(height int64, round int32) {
 		cs.rs.Proposal = nil
 		cs.rs.ProposalBlock = nil
 		cs.rs.ProposalBlockParts = nil
+		cs.proposalReceivedTime = time.Time{}
 	}
 
 	logger.Debug("entering new round",
@@ -2239,7 +2245,7 @@ func (cs *State) defaultSetProposal(proposal *types.Proposal) error {
 	}
 
 	cs.Logger.Info("received proposal", "proposal", proposal, "proposer", pubKey.Address())
-	schema.WriteFullBlockReceivingTime(cs.traceClient, proposal.Height, proposal.Round, time.Now(), false)
+	cs.proposalReceivedTime = time.Now()
 	return nil
 }
 
@@ -2296,7 +2302,10 @@ func (cs *State) addProposalBlockPart(msg *BlockPartMessage, peerID p2p.ID) (add
 		)
 	}
 	if added && cs.rs.ProposalBlockParts.IsComplete() {
-		schema.WriteFullBlockReceivingTime(cs.traceClient, msg.Height, msg.Round, time.Now(), true)
+		if !cs.proposalReceivedTime.IsZero() {
+			duration := time.Since(cs.proposalReceivedTime)
+			schema.WriteFullBlockReceivingTime(cs.traceClient, msg.Height, msg.Round, duration)
+		}
 		bz := cs.rs.ProposalBlockParts.GetBytes()
 
 		pbb := new(cmtproto.Block)

docs/p2p-channels.md

@@ -0,0 +1,22 @@
+# P2P Channel Values
+
+This document lists all the P2P channel identifiers used by the various reactors in celestia-core.
+
+## Channel Overview
+
+| Channel Name | Hex Value | Decimal Value | Reactor | File | Description |
+|--------------|-----------|---------------|---------|------|-------------|
+| PexChannel | 0x00 | 0 | PEX | `p2p/pex/pex_reactor.go:21` | Peer exchange protocol |
+| StateChannel | 0x20 | 32 | Consensus | `consensus/reactor.go:31` | Consensus state messages |
+| DataChannel | 0x21 | 33 | Consensus | `consensus/reactor.go:32` | Consensus data messages |
+| VoteChannel | 0x22 | 34 | Consensus | `consensus/reactor.go:33` | Vote messages |
+| VoteSetBitsChannel | 0x23 | 35 | Consensus | `consensus/reactor.go:34` | Vote set bits messages |
+| MempoolChannel | 0x30 | 48 | Mempool | `mempool/mempool.go:13` | Basic mempool channel |
+| MempoolDataChannel | 0x31 | 49 | Mempool CAT | `mempool/cat/reactor.go:27` | CAT mempool data (SeenTx and blob messages) |
+| MempoolWantsChannel | 0x32 | 50 | Mempool CAT | `mempool/cat/reactor.go:30` | CAT mempool wants messages |
+| EvidenceChannel | 0x38 | 56 | Evidence | `evidence/reactor.go:17` | Evidence messages |
+| BlocksyncChannel | 0x40 | 64 | Blocksync | `blocksync/reactor.go:22` | Block synchronization |
+| DataChannel | 0x50 | 80 | Propagation | `consensus/propagation/reactor.go:33` | Propagation data messages |
+| WantChannel | 0x51 | 81 | Propagation | `consensus/propagation/reactor.go:36` | Propagation want messages |
+| SnapshotChannel | 0x60 | 96 | State Sync | `statesync/reactor.go:21` | State sync snapshot messages |
+| ChunkChannel | 0x61 | 97 | State Sync | `statesync/reactor.go:23` | State sync chunk messages |

libs/trace/schema/consensus.go

@@ -273,6 +273,7 @@ type BlockPartState struct {
 	Have         bool         `json:"have"`
 	Peer         string       `json:"peer"`
 	TransferType TransferType `json:"transfer_type"`
+	MessageType  string       `json:"message_type"`
 }
 
 // Table returns the table name for the BlockPart struct.
@@ -290,6 +291,7 @@ func WriteBlockPartState(
 	have bool,
 	peer string,
 	transferType TransferType,
+	messageType string,
 ) {
 	// this check is redundant to what is checked during client.Write, although it
 	// is an optimization to avoid allocations from the map of fields.
@@ -304,6 +306,7 @@ func WriteBlockPartState(
 		Have:         have,
 		Peer:         peer,
 		TransferType: transferType,
+		MessageType:  messageType,
 	})
 }
 
@@ -493,11 +496,10 @@ const (
 )
 
 type FullBlockReceivingTime struct {
-	Height int64     `json:"height"`
-	Round  int32     `json:"round"`
-	Time   time.Time `json:"time"`
-	// Complete set to false when we receive the proposal. Set to true when the full block is received
-	Complete bool `json:"complete"`
+	Height int64 `json:"height"`
+	Round  int32 `json:"round"`
+	// Duration in milliseconds from when the proposal was received to when the full block was received
+	DurationMs int64 `json:"duration_ms"`
 }
 
 func (b FullBlockReceivingTime) Table() string {
@@ -508,13 +510,11 @@ func WriteFullBlockReceivingTime(
 	client trace.Tracer,
 	height int64,
 	round int32,
-	t time.Time,
-	complete bool,
+	duration time.Duration,
 ) {
 	client.Write(FullBlockReceivingTime{
-		Height:   height,
-		Round:    round,
-		Time:     t,
-		Complete: complete,
+		Height:     height,
+		Round:      round,
+		DurationMs: duration.Milliseconds(),
 	})
 }