node: introduce settings to configure the number of events per miniblock (#4553)

## Changes

  ### Configuration (`crypto/config.go`)
  - Added two new on-chain configuration keys:
- `stream.maxEventsPerMiniblock` - Maximum number of events per
miniblock
- `stream.maxTotalEventsSizePerMiniblock` - Maximum combined size (in
bytes) of all protobuf-encoded events
  - Defined protocol-level defaults that also serve as hard limits:
    - `MaxEventsPerMiniblockDefault = 15,000` events
    - `StreamMaxTotalEventsSizePerMiniblockDefault = 10 MB`
  - Added corresponding fields to `OnChainSettings` struct

  ### Event Selection (`minipool.go`)
- Modified `eventHashes()` and `eventHashesAsBytes()` to enforce
configured limits
  - Implemented dual-limit logic:
    - Stops at max event count, OR
    - Stops when cumulative event size exceeds max size
  - Events are selected in order until either limit is reached

  ### Proposal & Validation (`stream_view.go`, `stream.go`)
  - Added `MiniblockEventLimits()` helper function that:
    - Reads limits from on-chain settings
    - Falls back to defaults if not configured
    - Enforces protocol max limits even if on-chain values are higher
- Enhanced `ProposeNextMiniblock()` to reject proposals exceeding limits
- Added validation in `tryApplyCandidate()` to reject oversized
candidates before applying

  ### Testing (`replication_test.go`)
- Added `TestReplicatedMbProductionWithTooManyEvents` - validates max
event count enforcement
- Added `TestReplicatedMbProductionWithTooBigEvents` - validates max
total size enforcement
- Both tests verify that miniblocks are properly split when limits are
exceeded

  ## Behavior

**Before**: Miniblocks could contain unlimited events from the minipool,
potentially creating very large blocks
  that impact network performance and blockchain transaction costs.

  **After**: Miniblocks are capped at:
- Default: 15,000 events OR 10 MB total size (whichever is reached
first)
  - Configurable via on-chain settings within protocol max limits
- Events exceeding limits remain in the minipool for subsequent
miniblocks

  ## Motivation

- **Network efficiency**: Prevents oversized miniblocks that strain
replication and storage
  - **Cost control**: Limits on-chain registration transaction sizes
- **Configurability**: Operators can tune limits via on-chain settings
without code changes
  - **Safety**: Hard-coded maximum limits prevent misconfiguration

  ## Testing

  - ✅ New tests verify both limit types are enforced correctly
  - ✅ Existing miniblock production tests still pass
  - ✅ Tests confirm events properly overflow into subsequent miniblocks

Author: bas-vk

core/node/crypto/config.go

@@ -70,10 +70,25 @@ const (
 	StreamUserInboxStreamHistoryMiniblocksConfigKey    = "stream.historyminiblocks.a1"
 	StreamUserSettingsStreamHistoryMiniblocksConfigKey = "stream.historyminiblocks.a5"
 
+	// StreamMaxEventsPerMiniblockKey is the maximum number of events in a miniblock.
+	StreamMaxEventsPerMiniblockKey = "stream.maxeventsperminiblock"
+
+	// StreamMaxTotalEventsSizePerMiniblockKey is the maximum size (in bytes) of all protobuf encoded events
+	// combined in a miniblock.
+	StreamMaxTotalEventsSizePerMiniblockKey = "stream.maxtotaleventssizeperminiblock"
+
 	// StreamDistributionExtraCandidatesCountCountKey is the key for many extra nodes on top of
 	// replication factor must be picked as candidates to place a stream on. From these candidates
 	// the best replication factor nodes are picked.
 	StreamDistributionExtraCandidatesCountCountKey = "stream.distribution.extracandidatescount"
+
+	// MaxEventsPerMiniblockDefault defines the default (also the maximum) number of events in a miniblock
+	// if not overwritten by StreamMaxEventsPerMiniblockKey.
+	MaxEventsPerMiniblockDefault = 15_000
+
+	// StreamMaxTotalEventsSizePerMiniblockDefault defines the default (also the maximum) size of all protobuf encoded events
+	// combined in a miniblock if not overwritten by StreamMaxTotalEventsSizePerMiniblockKey.
+	StreamMaxTotalEventsSizePerMiniblockDefault = 10 * 1024 * 1024
 )
 
 var (
@@ -170,6 +185,12 @@ type OnChainSettings struct {
 	// Number of miniblocks to keep for each type of stream.
 	// 0 means keep all miniblocks.
 	StreamHistoryMiniblocks StreamHistoryMiniblocks `mapstructure:",squash"`
+
+	// StreamMaxEventsPerMiniblock is the maximum number of events that can be included in a miniblock.
+	StreamMaxEventsPerMiniblock uint64 `mapstructure:"stream.maxeventsperminiblock"`
+	// StreamMaxTotalEventsSizePerMiniblock is the maximum size (in bytes) of all protobuf encoded events
+	// combined in a miniblock.
+	StreamMaxTotalEventsSizePerMiniblock uint64 `mapstructure:"stream.maxtotaleventssizeperminiblock"`
 }
 
 type StreamHistoryMiniblocks struct {

core/node/events/miniblock_job.go

@@ -9,6 +9,8 @@ import (
 
 	"github.com/ethereum/go-ethereum/common"
 
+	"github.com/towns-protocol/towns/core/node/crypto"
+
 	"github.com/towns-protocol/towns/core/blockchain"
 	. "github.com/towns-protocol/towns/core/node/base"
 	"github.com/towns-protocol/towns/core/node/logging"
@@ -125,8 +127,7 @@ func (j *mbJob) produceCandidate(ctx context.Context, blockNum blockchain.BlockN
 		return nil
 	}
 
-	err = j.saveCandidate(ctx)
-	if err != nil {
+	if err := j.saveCandidate(ctx); err != nil {
 		return err
 	}
 
@@ -158,16 +159,18 @@ func (j *mbJob) makeCandidate(ctx context.Context) error {
 }
 
 func (j *mbJob) makeReplicatedProposal(ctx context.Context) (*mbProposal, *StreamView, error) {
+	// retrieve remote proposals
 	proposals, view, err := j.processRemoteProposals(ctx)
 	if err != nil {
 		return nil, nil, err
 	}
 
-	localProposal := view.proposeNextMiniblock(ctx, j.cache.Params().ChainConfig.Get(), j.forceSnapshot)
+	// create local proposal
+	localProposal := view.proposeNextMiniblock(ctx, j.cache.Params().ChainConfig.Get(), j.forceSnapshot, j.replicated)
 
+	// Combine proposals into a single proposal that is used to create a miniblock candidate.
 	proposals = append(proposals, localProposal)
-
-	combined, err := j.combineProposals(proposals)
+	combined, err := j.combineProposals(proposals, j.cache.Params().ChainConfig.Get())
 	if err != nil {
 		return nil, nil, err
 	}
@@ -181,10 +184,10 @@ func (j *mbJob) makeLocalProposal(ctx context.Context) (*mbProposal, *StreamView
 		return nil, nil, err
 	}
 
-	prop := view.proposeNextMiniblock(ctx, j.cache.Params().ChainConfig.Get(), j.forceSnapshot)
+	prop := view.proposeNextMiniblock(ctx, j.cache.Params().ChainConfig.Get(), j.forceSnapshot, j.replicated)
 
 	// Is there anything to do?
-	if len(prop.eventHashes) == 0 && !prop.shouldSnapshot {
+	if len(prop.events) == 0 && !prop.shouldSnapshot {
 		return nil, view, nil
 	}
 
@@ -212,6 +215,7 @@ func (j *mbJob) processRemoteProposals(ctx context.Context) ([]*mbProposal, *Str
 	if err != nil {
 		return nil, nil, err
 	}
+
 	if view.minipool.generation != request.NewMiniblockNum {
 		// TODO: REPLICATION: FIX: if any MissingEvents are received, should they still be attempted to be added? I.e. loop below should be still executed?
 		return nil, nil, RiverError(
@@ -220,25 +224,26 @@ func (j *mbJob) processRemoteProposals(ctx context.Context) ([]*mbProposal, *Str
 		)
 	}
 
-	// Apply received MissingEvents, even if there are not enough quorum of proposals.
+	// Apply received MissingEvents, even if there is not enough to reach quorum at the moment.
+	// This will ensure that events propagate over all replicas.
 	added := make(map[common.Hash]bool)
 	converted := make([]*mbProposal, len(proposals))
 	for i, p := range proposals {
-		converted[i] = mbProposalFromProto(p.response.Proposal)
-
-		if converted[i].newMiniblockNum != view.minipool.generation {
+		// if the proposal was created on a different minipool generation, ignore it.
+		if p.response.GetProposal().GetNewMiniblockNum() != view.minipool.generation {
 			continue
 		}
 
+		// add missing events that replicas provided to the local minipool and keep track of the new stream view
 		for _, e := range p.response.MissingEvents {
 			parsed, err := ParseEvent(e)
 			if err != nil {
 				logging.FromCtx(ctx).Errorw("mbJob.processRemoteProposals: error parsing event", "error", err)
 				continue
 			}
+
 			if _, ok := added[parsed.Hash]; !ok {
 				added[parsed.Hash] = true
-
 				if !view.minipool.events.Has(parsed.Hash) {
 					newView, err := j.stream.addEvent(ctx, parsed, true)
 					if err == nil {
@@ -255,6 +260,8 @@ func (j *mbJob) processRemoteProposals(ctx context.Context) ([]*mbProposal, *Str
 				}
 			}
 		}
+
+		converted[i] = mbProposalFromProto(p.response.Proposal, view)
 	}
 
 	// View might have been updated by adding events, check if stream advanced in the meantime.
@@ -281,7 +288,13 @@ func (j *mbJob) processRemoteProposals(ctx context.Context) ([]*mbProposal, *Str
 	return nil, nil, RiverError(Err_INTERNAL, "mbJob.processRemoteProposals: no proposals and no errors")
 }
 
-func (j *mbJob) combineProposals(proposals []*mbProposal) (*mbProposal, error) {
+// combineProposals combines the given proposals into a single proposal that can be
+// used to create a miniblock candidate.
+//
+// The combined proposal will contain all events from the given proposals that appear at least
+// quorumNum times. It will also contain the ShouldSnapshot flag if at least quorumNum proposals
+// contain ShouldSnapshot events.
+func (j *mbJob) combineProposals(proposals []*mbProposal, cfg *crypto.OnChainSettings) (*mbProposal, error) {
 	// Sanity check: all proposals must have the same miniblock number and prev hash.
 	for _, p := range proposals {
 		if p.newMiniblockNum != proposals[0].newMiniblockNum || p.prevMiniblockHash != proposals[0].prevMiniblockHash {
@@ -295,49 +308,71 @@ func (j *mbJob) combineProposals(proposals []*mbProposal) (*mbProposal, error) {
 		return nil, RiverError(Err_INTERNAL, "mbJob.combineProposals: not enough proposals")
 	}
 
-	// Count ShouldSnapshot.
-	shouldSnapshotNum := 0
+	// count on how many replicates votes to create a snapshot in the miniblock candidate.
+	proposalSnapshotVotesCount := 0
 	for _, p := range proposals {
 		if p.shouldSnapshot {
-			shouldSnapshotNum++
+			proposalSnapshotVotesCount++
 		}
 	}
-	shouldSnapshot := shouldSnapshotNum >= quorumNum
 
-	// Count event hashes.
+	// Count how often each event appears in the proposals.
+	// Only events that appear at least quorumNum times are included in the combined proposal.
 	eventCounts := make(map[common.Hash]int)
 	for _, p := range proposals {
-		for _, h := range p.eventHashes {
-			eventCounts[h]++
+		for _, ev := range p.events {
+			eventCounts[ev.Hash]++
 		}
 	}
 
-	events := make([]common.Hash, 0, len(eventCounts))
-
-	// walk over all event hashes again, adding them to the events list if they have quorum.
+	// walk over all event hashes and add the events to the combined proposal if reached quorum.
 	// do it this way to preserve order of events as they were received in a single proposal.
-	// we do not attempt to order events across proposals.
+	// we do not attempt to order events across proposals. allow up to set limits in the
+	// combined proposal.
+	var (
+		maxEventsPerMiniblock, maxEventCombinedSize = MiniblockEventLimits(cfg)
+		combinedEvents                              = make([]mbProposalEvent, 0, len(eventCounts))
+		combinedTotalEventSize                      = 0
+	)
+
+	var reachedLimit bool
 	for _, p := range proposals {
-		for _, h := range p.eventHashes {
-			if c, ok := eventCounts[h]; ok {
+		for _, event := range p.events {
+			if c, ok := eventCounts[event.Hash]; ok {
+				// only include events that appear at least quorumNum times
 				if c >= quorumNum {
-					events = append(events, h)
+					// if adding the next event to the combined proposal exceeds limits, stop processing all proposals
+					if len(combinedEvents) >= maxEventsPerMiniblock ||
+						combinedTotalEventSize+event.Size > maxEventCombinedSize {
+						reachedLimit = true
+						break
+					}
+					combinedEvents = append(combinedEvents, event)
+					// keep track of the total size of all events in the combined proposal
+					combinedTotalEventSize += event.Size
 				}
-				delete(eventCounts, h)
+				// if the next proposal contains the same event, it won't be processed again.
+				delete(eventCounts, event.Hash)
 			}
 		}
+		if reachedLimit {
+			break
+		}
 	}
 
+	// ShouldSnapshot flag is set if at least quorumNum proposals contain ShouldSnapshot vote.
+	shouldSnapshot := proposalSnapshotVotesCount >= quorumNum
+
 	// Is there anything to do?
-	if len(events) == 0 && !shouldSnapshot {
+	if len(combinedEvents) == 0 && !shouldSnapshot {
 		return nil, nil
 	}
 
 	return &mbProposal{
 		newMiniblockNum:   proposals[0].newMiniblockNum,
 		prevMiniblockHash: proposals[0].prevMiniblockHash,
 		shouldSnapshot:    shouldSnapshot,
-		eventHashes:       events,
+		events:            combinedEvents,
 	}, nil
 }
 

core/node/events/minipool.go

@@ -3,6 +3,8 @@ package events
 import (
 	"github.com/ethereum/go-ethereum/common"
 
+	"github.com/towns-protocol/towns/core/node/crypto"
+
 	. "github.com/towns-protocol/towns/core/node/utils"
 )
 
@@ -76,14 +78,34 @@ func (m *minipoolInstance) getEnvelopeBytes() ([][]byte, error) {
 	return bytes, nil
 }
 
-func (m *minipoolInstance) eventHashes() []common.Hash {
-	hashes := make([]common.Hash, m.events.Len())
-	for i, e := range m.events.Values {
-		hashes[i] = e.Hash
+// proposalEvents returns all events from m that can be included in a proposal.
+func (m *minipoolInstance) proposalEvents() []mbProposalEvent {
+	events := make([]mbProposalEvent, 0, m.events.Len())
+	for _, e := range m.events.Values {
+		events = append(events, mbProposalEvent{Hash: e.Hash, Size: len(e.Envelope.Event)})
 	}
-	return hashes
+	return events
+}
+
+// proposalEventsWithMiniblockLimits returns events from m that can be included in a proposal
+// until the limits as specified in the given cfg are reached.
+func (m *minipoolInstance) proposalEventsWithMiniblockLimits(cfg *crypto.OnChainSettings) []mbProposalEvent {
+	maxEventsPerMiniblock, maxEventCombinedSize := MiniblockEventLimits(cfg)
+	events := make([]mbProposalEvent, 0, m.events.Len())
+	totalEventsSize := 0
+	for _, e := range m.events.Values {
+		eventSize := len(e.Envelope.Event)
+		// Check if adding the next event would exceed limits before appending
+		if len(events) >= maxEventsPerMiniblock || totalEventsSize+eventSize > maxEventCombinedSize {
+			return events
+		}
+		events = append(events, mbProposalEvent{Hash: e.Hash, Size: eventSize})
+		totalEventsSize += eventSize
+	}
+	return events
 }
 
+// eventHashesAsBytes returns all event hashes from m.
 func (m *minipoolInstance) eventHashesAsBytes() [][]byte {
 	hashes := make([][]byte, m.events.Len())
 	for i, e := range m.events.Values {