Skip to content

feat(rpc): wire UnmarshalWithLimit into Stream.Recv #3630

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

Draft
wen-coding wants to merge 4 commits into
base: main
Choose a base branch
from
Draft

feat(rpc): wire UnmarshalWithLimit into Stream.Recv #3630

Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
a97dd6d
feat(rpc): wire UnmarshalWithLimit into Stream.Recv with P2P config o…
wen-coding Jun 23, 2026
7c51eb3
simplify: replace allocMultiplier var+setter with a constant
wen-coding Jun 23, 2026
1560675
fix: gofmt alignment in DefaultP2PConfig
wen-coding Jun 23, 2026
2fdc4ac
refactor(rpc): make recvAllocMultiplier a float, round result to int
wen-coding Jun 23, 2026
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
18 changes: 14 additions & 4 deletions sei-tendermint/internal/p2p/rpc/rpc.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,7 @@ package rpc
import (
"context"
"fmt"
"math"
"reflect"

"github.com/sei-protocol/sei-chain/sei-tendermint/internal/p2p/conn"
Expand All @@ -16,6 +17,12 @@ import (
type InBytes uint64
type InMsgs uint64

// recvAllocMultiplier scales each stream's inbound MsgSize to derive the
// per-message alloc limit passed to UnmarshalWithLimit. Load testing against
// the largest message (~2 MB wire for LaneProposal) showed its alloc estimate
// fits comfortably within 2× MsgSize.
const recvAllocMultiplier = 2.0

@seidroid seidroid Bot Jul 7, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[blocker] This multiplier is hardcoded, but the PR description advertises an operator-tunable proto-alloc-limit-multiplier / rpc.SetAllocMultiplier escape hatch. That plumbing (config.go, setup.go, P2PConfig.ProtoAllocLimitMultiplier) is not present in this diff. Since exceeding this bound causes Stream.Recv to error and drop the peer connection on a consensus-critical path, please either add the config override as described or update the PR description/test plan to reflect that the multiplier is fixed at 2×.


func (spec Msg[M]) Verify() error {
var msg M
if m := InBytes(msg.MaxSize()); m > spec.MsgSize { // nolint: gosec // MaxSize() >= 0
Expand Down Expand Up @@ -112,7 +119,7 @@ func (r *RPC[API, Req, Resp]) Call(ctx context.Context, client Client[API]) (Str
if err != nil {
return Stream[Req, Resp]{}, err
}
return Stream[Req, Resp]{inner: s}, nil
return Stream[Req, Resp]{inner: s, allocLimit: int(math.Round(float64(r.Resp.MsgSize) * recvAllocMultiplier))}, nil //nolint:gosec // MsgSize is a validated config value
}

func (r *RPC[API, Req, Resp]) Serve(ctx context.Context, server Server[API], handler func(context.Context, Stream[Resp, Req]) error) error {
Expand All @@ -128,7 +135,7 @@ func (r *RPC[API, Req, Resp]) Serve(ctx context.Context, server Server[API], han
if err != nil {
return err
}
err = handler(ctx, Stream[Resp, Req]{inner: stream})
err = handler(ctx, Stream[Resp, Req]{inner: stream, allocLimit: int(math.Round(float64(r.Req.MsgSize) * recvAllocMultiplier))}) //nolint:gosec // MsgSize is a validated config value
stream.Close()
if err != nil {
return err
Expand All @@ -141,7 +148,10 @@ func (r *RPC[API, Req, Resp]) Serve(ctx context.Context, server Server[API], han
})
}

type Stream[SendT, RecvT protoutils.Message] struct{ inner *mux.Stream }
type Stream[SendT, RecvT protoutils.Message] struct {
inner *mux.Stream
allocLimit int
}

func (s Stream[SendT, RecvT]) Close() { s.inner.Close() }
func (s Stream[SendT, RecvT]) Send(ctx context.Context, msg SendT) error {
Expand All @@ -152,5 +162,5 @@ func (s Stream[SendT, RecvT]) Recv(ctx context.Context) (RecvT, error) {
if err != nil {
return utils.Zero[RecvT](), err
}
return protoutils.Unmarshal[RecvT](raw)
return protoutils.UnmarshalWithLimit[RecvT](raw, s.allocLimit)

@seidroid seidroid Bot Jul 7, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[suggestion] This is the single decode point for all inbound P2P v2 messages. A rejection here (estimate > MsgSize×2) surfaces as a Recv error which, in Serve, aborts the handler and closes the stream — effectively dropping the peer. Worth a targeted test that a legitimately-sized message for each registered RPC passes the alloc check, plus a test that an over-allocating payload is rejected, so the 2× headroom is validated beyond LaneProposal.

}
Loading