[Merged by Bors] - feat: hare preround proposal compaction

config/config.go

@@ -193,7 +193,7 @@ func DefaultConfig() Config {
 		P2P:             p2p.DefaultConfig(),
 		API:             grpcserver.DefaultConfig(),
 		HARE3:           hare3.DefaultConfig(),
-		HARE4:           hare4.DefaultConfig(),
+		HARE4:           hare4.DefaultConfig(), // DEFAULT HARE4 IS DISABLED
 		HareEligibility: eligibility.DefaultConfig(),
 		Beacon:          beacon.DefaultConfig(),
 		TIME:            timeConfig.DefaultConfig(),

config/mainnet.go

@@ -60,9 +60,7 @@ func MainnetConfig() Config {
 	}
 	logging := DefaultLoggingConfig()
 	logging.TrtlLoggerLevel = zapcore.WarnLevel.String()
-	logging.AtxHandlerLevel = zapcore.WarnLevel.String()
-	logging.ProposalListenerLevel = zapcore.WarnLevel.String()
-	forkLayer := types.LayerID(111_111_111) // TODO THIS NEEDS A NUMBER
+	logging.MeshLoggerLevel = zapcore.WarnLevel.String()
 	hare3conf := hare3.DefaultConfig()
 	hare3conf.Committee = 400
 	hare3conf.Enable = true
@@ -71,12 +69,9 @@ func MainnetConfig() Config {
 		Layer: 105_720, // July 15, 2024, 10:00:00 AM UTC
 		Size:  50,
 	}
-	hare3conf.DisableLayer = forkLayer
 
 	hare4conf := hare4.DefaultConfig()
-	hare4conf.Committee = 50
-	hare4conf.Enable = true
-	hare4conf.EnableLayer = forkLayer
+	hare4conf.Enable = false
 	return Config{
 		BaseConfig: BaseConfig{
 			DataDirParent:         defaultDataDir,

config/presets/fastnet.go

@@ -33,22 +33,13 @@ func fastnet() config.Config {
 	conf.ATXGradeDelay = 1 * time.Second
 
 	conf.HARE3.Enable = true
-	conf.HARE3.DisableLayer = 22
+	conf.HARE3.DisableLayer = types.LayerID(math.MaxUint32)
 	conf.HARE3.Committee = 800
 	conf.HARE3.Leaders = 10
 	conf.HARE3.PreroundDelay = 3 * time.Second
 	conf.HARE3.RoundDuration = 700 * time.Millisecond
 	conf.HARE3.IterationsLimit = 2
 
-	conf.HARE4.Enable = true
-	conf.HARE4.EnableLayer = types.LayerID(22)
-	conf.HARE4.DisableLayer = types.LayerID(math.MaxUint32)
-	conf.HARE4.Committee = 800
-	conf.HARE4.Leaders = 10
-	conf.HARE4.PreroundDelay = 3 * time.Second
-	conf.HARE4.RoundDuration = 700 * time.Millisecond
-	conf.HARE4.IterationsLimit = 2
-
 	conf.P2P.MinPeers = 10
 
 	conf.Genesis = config.GenesisConfig{

config/presets/testnet.go

@@ -52,16 +52,13 @@ func testnet() config.Config {
 	}
 	hare3conf := hare3.DefaultConfig()
 	hare3conf.Enable = true
-	hare3conf.EnableLayer = 0
-	hare3conf.DisableLayer = 50
+	hare3conf.EnableLayer = 7366
 	// NOTE(dshulyak) i forgot to set protocol name for testnet when we configured it manually.
 	// we can't do rolling upgrade if protocol name changes, so lets keep it like that temporarily.
 	hare3conf.ProtocolName = ""
-	hare4conf := hare4.DefaultConfig()
-	hare4conf.Enable = true
-	hare4conf.EnableLayer = 50 // TODO THIS NEEDS A VALUE
-	hare4conf.DisableLayer = math.MaxUint32
 
+	hare4conf := hare4.DefaultConfig()
+	hare4conf.Enable = false
 	defaultdir := filepath.Join(home, "spacemesh-testnet", "/")
 	return config.Config{
 		Preset: "testnet",

hare4/hare.go

@@ -46,34 +46,48 @@
 	errResponseTooBig       = errors.New("response too big")
 	errCannotFindProposal   = errors.New("cannot find proposal")
 	errNoEligibilityProofs  = errors.New("no eligibility proofs")
+	errSigTooShort          = errors.New("signature too short")
 	fetchFullTimeout        = 5 * time.Second
 )
 
+type CommitteeUpgrade struct {
+	Layer types.LayerID
+	Size  uint16
+}
+
 type Config struct {
-	Enable          bool          `mapstructure:"enable"`
-	EnableLayer     types.LayerID `mapstructure:"enable-layer"`
-	DisableLayer    types.LayerID `mapstructure:"disable-layer"`
-	Committee       uint16        `mapstructure:"committee"`
-	Leaders         uint16        `mapstructure:"leaders"`
-	IterationsLimit uint8         `mapstructure:"iterations-limit"`
-	PreroundDelay   time.Duration `mapstructure:"preround-delay"`
-	RoundDuration   time.Duration `mapstructure:"round-duration"`
+	Enable           bool          `mapstructure:"enable"`
+	EnableLayer      types.LayerID `mapstructure:"enable-layer"`
+	DisableLayer     types.LayerID `mapstructure:"disable-layer"`
+	Committee        uint16        `mapstructure:"committee"`
+	CommitteeUpgrade *CommitteeUpgrade
+	Leaders          uint16        `mapstructure:"leaders"`
+	IterationsLimit  uint8         `mapstructure:"iterations-limit"`
+	PreroundDelay    time.Duration `mapstructure:"preround-delay"`
+	RoundDuration    time.Duration `mapstructure:"round-duration"`
 	// LogStats if true will log iteration statistics with INFO level at the start of the next iteration.
 	// This requires additional computation and should be used for debugging only.
 	LogStats     bool   `mapstructure:"log-stats"`
 	ProtocolName string `mapstructure:"protocolname"`
 }
 
+func (cfg *Config) CommitteeFor(layer types.LayerID) uint16 {
+	if cfg.CommitteeUpgrade != nil && layer >= cfg.CommitteeUpgrade.Layer {
+		return cfg.CommitteeUpgrade.Size
+	}
+	return cfg.Committee
+}
+
 func (cfg *Config) Validate(zdist time.Duration) error {
 	terminates := cfg.roundStart(IterRound{Iter: cfg.IterationsLimit, Round: hardlock})
 	if terminates > zdist {
 		return fmt.Errorf("hare terminates later (%v) than expected (%v)", terminates, zdist)
 	}
 	if cfg.Enable && cfg.DisableLayer <= cfg.EnableLayer {
 		return fmt.Errorf("disabled layer (%d) must be larger than enabled (%d)",
 			cfg.DisableLayer, cfg.EnableLayer)
 	}
 	return nil
 }
 
 func (cfg *Config) MarshalLogObject(encoder zapcore.ObjectEncoder) error {
@@ -81,6 +95,10 @@
 	encoder.AddUint32("enabled layer", cfg.EnableLayer.Uint32())
 	encoder.AddUint32("disabled layer", cfg.DisableLayer.Uint32())
 	encoder.AddUint16("committee", cfg.Committee)
+	if cfg.CommitteeUpgrade != nil {
+		encoder.AddUint32("committee upgrade layer", cfg.CommitteeUpgrade.Layer.Uint32())
+		encoder.AddUint16("committee upgrade size", cfg.CommitteeUpgrade.Size)
+	}
 	encoder.AddUint16("leaders", cfg.Leaders)
 	encoder.AddUint8("iterations limit", cfg.IterationsLimit)
 	encoder.AddDuration("preround delay", cfg.PreroundDelay)
@@ -162,9 +180,9 @@
 
 // WithResultsChan overrides the default result channel with a different one.
 // This is only needed for the migration period between hare3 and hare4.
 func WithResultsChan(c chan ConsensusOutput) Opt {
 	return func(hr *Hare) {
 		hr.results = c
 	}
 }
 
@@ -207,7 +225,6 @@
 		atxsdata:  atxsdata,
 		proposals: proposals,
 		verifier:  verif,
-		compactFn: compactTruncate,
 		oracle: &legacyOracle{
 			log:    zap.NewNop(),
 			oracle: oracle,
@@ -222,7 +239,7 @@
 	}
 
 	if host != nil {
 		hr.p2p = server.New(host, PROTOCOL_NAME, hr.handleProposalsStream)
 	}
 	return hr
 }
@@ -251,7 +268,6 @@
 	atxsdata  *atxsdata.Data
 	proposals *store.Store
 	verifier  verifier
-	compactFn func([]byte) []byte
 	oracle    *legacyOracle
 	sync      system.SyncStateProvider
 	patrol    *layerpatrol.LayerPatrol
@@ -291,14 +307,14 @@
 		for next := enabled; next < disabled; next++ {
 			select {
 			case <-h.nodeclock.AwaitLayer(next):
-				h.log.Debug("notified", zap.Uint32("layer id", next.Uint32()))
+				h.log.Debug("notified", zap.Uint32("layer", next.Uint32()))
 				h.onLayer(next)
 				h.cleanMessageCache(next - 1)
 			case <-h.ctx.Done():
 				return nil
 			}
 		}
 		return nil
 	})
 }
 
@@ -322,27 +338,22 @@
 	cb := func(ctx context.Context, rw io.ReadWriter) error {
 		respLen, _, err := codec.DecodeLen(rw)
 		if err != nil {
 			return fmt.Errorf("decode length: %w", err)
 		}
 		if respLen >= MAX_EXCHANGE_SIZE {
 			return errResponseTooBig
 		}
-		buff := make([]byte, respLen)
-		_, err = io.ReadFull(rw, buff)
-		if err != nil {
-			return fmt.Errorf("read response buffer: %w", err)
-		}
-		err = codec.Decode(buff, resp)
-		if err != nil {
+		b, err := codec.DecodeFrom(rw, resp)
+		if err != nil || b != int(respLen) {
 			return fmt.Errorf("decode response: %w", err)
 		}
 		return nil
 	}
 
 	err := h.p2p.StreamRequest(ctx, peer, reqBytes, cb)
 	if err != nil {
 		requestCompactErrorCounter.Inc()
 		return nil, fmt.Errorf("stream request: %w", err)
 	}
 
 	h.tracer.OnCompactIdResponse(resp)
@@ -354,25 +365,25 @@
 	requestCompactHandlerCounter.Inc()
 	compactProps := &CompactIdRequest{}
 	if err := codec.Decode(msg, compactProps); err != nil {
 		malformedError.Inc()
 		return fmt.Errorf("%w: decoding error %s", pubsub.ErrValidationReject, err.Error())
 	}
 	h.tracer.OnCompactIdRequest(compactProps)
 	h.mu.Lock()
 	m, ok := h.messageCache[compactProps.MsgId]
 	h.mu.Unlock()
 	if !ok {
 		messageCacheMiss.Inc()
 		return fmt.Errorf("message %s: cache miss", compactProps.MsgId)
 	}
 	resp := &CompactIdResponse{Ids: m.Body.Value.Proposals}
 	respBytes := codec.MustEncode(resp)
 	if _, err := codec.EncodeLen(s, uint32(len(respBytes))); err != nil {
 		return fmt.Errorf("encode length: %w", err)
 	}
 
 	if _, err := s.Write(respBytes); err != nil {
 		return fmt.Errorf("write response: %w", err)
 	}
 
 	return nil
@@ -383,15 +394,14 @@
 func (h *Hare) reconstructProposals(ctx context.Context, peer p2p.Peer, msgId types.Hash32, msg *Message) error {
 	proposals := h.proposals.GetForLayer(msg.Layer)
 	if len(proposals) == 0 {
 		return errNoLayerProposals
 	}
-	compacted := h.compactProposals(h.compactFn, msg.Layer, proposals)
+	compacted := h.compactProposals(msg.Layer, proposals)
 	proposalIds := make([]proposalTuple, len(proposals))
 	for i := range proposals {
 		proposalIds[i] = proposalTuple{id: proposals[i].ID(), compact: compacted[i]}
 	}
-	slices.SortFunc(proposalIds, sortProposalsTuple)
-
+	slices.SortFunc(proposalIds, func(i, j proposalTuple) int { return bytes.Compare(i.id[:], j.id[:]) })
 	taken := make([]bool, len(proposals))
 	findProp := func(id types.CompactProposalID) (bool, types.ProposalID) {
 		for i := 0; i < len(proposalIds); i++ {
@@ -399,7 +409,7 @@
 				continue
 			}
 			if taken[i] {
 				continue
 			}
 			// item is both not taken and equals to looked up ID
 			taken[i] = true
@@ -426,13 +436,13 @@
 
 	if ctr != len(msg.Value.CompactProposals) {
 		// this will force the calling context to do a fetchFull
 		return errCannotMatchProposals
 	}
 
 	// sort the found proposals and unset the compact proposals
 	// field before trying to check the signature
 	// since it would add unnecessary data to the hasher
-	slices.SortFunc(msg.Value.Proposals, sortProposalIds)
+	slices.SortFunc(msg.Value.Proposals, func(i, j types.ProposalID) int { return bytes.Compare(i[:], j[:]) })
 	msg.Value.CompactProposals = []types.CompactProposalID{}
 	return nil
 }
@@ -472,33 +482,33 @@
 		case errors.Is(err, errCannotMatchProposals):
 			msg.Value.Proposals, err = h.fetchFull(ctx, peer, msgId)
 			if err != nil {
 				return fmt.Errorf("fetch full: %w", err)
 			}
-			slices.SortFunc(msg.Value.Proposals, sortProposalIds)
+			slices.SortFunc(msg.Value.Proposals, func(i, j types.ProposalID) int { return bytes.Compare(i[:], j[:]) })
 			msg.Value.CompactProposals = []types.CompactProposalID{}
 			fetched = true
 		case err != nil:
 			return fmt.Errorf("reconstruct proposals: %w", err)
 		}
 	}
 	if !h.verifier.Verify(signing.HARE, msg.Sender, msg.ToMetadata().ToBytes(), msg.Signature) {
 		if msg.IterRound.Round == preround && !fetched {
 			preroundSigFailCounter.Inc()
 			// we might have a bad signature because of a local hash collision
 			// of a proposal that has the same short hash that the node sent us.
 			// in this case we try to ask for a full exchange of all full proposal
 			// ids and try to validate again
 			var err error
 			msg.Body.Value.Proposals, err = h.fetchFull(ctx, peer, msgId)
 			if err != nil {
 				return fmt.Errorf("signature verify: fetch full: %w", err)
 			}
 			if len(msg.Body.Value.Proposals) != len(compacts) {
 				return fmt.Errorf("signature verify: proposals mismatch: %w", err)
 			}
 			if !h.verifier.Verify(signing.HARE, msg.Sender, msg.ToMetadata().ToBytes(), msg.Signature) {
 				signatureError.Inc()
 				return fmt.Errorf("%w: signature verify: invalid signature", pubsub.ErrValidationReject)
 			}
 		} else {
 			signatureError.Inc()
@@ -543,7 +553,7 @@
 		proof := equivocation.ToMalfeasanceProof()
 		if err := identities.SetMalicious(
 			h.db, equivocation.Messages[0].SmesherID, codec.MustEncode(proof), time.Now()); err != nil {
 			h.log.Error("failed to save malicious identity", zap.Error(err))
 		}
 		h.atxsdata.SetMalicious(equivocation.Messages[0].SmesherID)
 	}
@@ -559,17 +569,17 @@
 func (h *Hare) onLayer(layer types.LayerID) {
 	h.proposals.OnLayer(layer)
 	if !h.sync.IsSynced(h.ctx) {
 		h.log.Debug("not synced", zap.Uint32("lid", layer.Uint32()))
 		return
 	}
 	beacon, err := beacons.Get(h.db, layer.GetEpoch())
 	if err != nil || beacon == types.EmptyBeacon {
 		h.log.Debug("no beacon",
 			zap.Uint32("epoch", layer.GetEpoch().Uint32()),
 			zap.Uint32("lid", layer.Uint32()),
 			zap.Error(err),
 		)
 		return
 	}
 	h.patrol.SetHareInCharge(layer)
 
@@ -580,7 +590,7 @@
 		beacon:  beacon,
 		signers: maps.Values(h.signers),
 		vrfs:    make([]*types.HareEligibility, len(h.signers)),
-		proto:   newProtocol(h.config.Committee/2 + 1),
+		proto:   newProtocol(h.config.CommitteeFor(layer)/2 + 1),
 	}
 	h.sessions[layer] = s.proto
 	h.mu.Unlock()
@@ -634,15 +644,15 @@
 		// initial set is not needed if node is not active in preround
 		select {
 		case <-h.wallclock.After(walltime.Sub(h.wallclock.Now())):
 		case <-h.ctx.Done():
 			return h.ctx.Err()
 		}
 		start := time.Now()
 		session.proto.OnInitial(h.selectProposals(session))
 		proposalsLatency.Observe(time.Since(start).Seconds())
 	}
 	if err := h.onOutput(session, current, session.proto.Next()); err != nil {
 		return err
 	}
 	result := false
 	for {
@@ -672,7 +682,7 @@
 				result = true
 			}
 			if err := h.onOutput(session, current, out); err != nil {
 				return err
 			}
 			// we are logginng stats 1 network delay after new iteration start
 			// so that we can receive notify messages from previous iteration
@@ -681,7 +691,7 @@
 			}
 			if out.terminated {
 				if !result {
 					return errors.New("terminated without result")
 				}
 				return nil
 			}
@@ -707,11 +717,11 @@
 		msg.Signature = session.signers[i].Sign(signing.HARE, msg.ToMetadata().ToBytes())
 		if ir.Round == preround {
 			var err error
-			msg.Body.Value.CompactProposals, err = h.compactProposalIds(h.compactFn, msg.Layer,
+			msg.Body.Value.CompactProposals, err = h.compactProposalIds(msg.Layer,
 				out.message.Body.Value.Proposals)
 			if err != nil {
 				h.log.Debug("failed to compact proposals", zap.Error(err))
 				continue
 			}
 			fullProposals := msg.Body.Value.Proposals
 			msg.Body.Value.Proposals = []types.ProposalID{}
@@ -723,7 +733,7 @@
 			msg.Body.Value.Proposals = []types.ProposalID{}
 		}
 		if err := h.pubsub.Publish(h.ctx, h.config.ProtocolName, msg.ToBytes()); err != nil {
 			h.log.Error("failed to publish", zap.Inline(&msg), zap.Error(err))
 		}
 	}
 	h.tracer.OnMessageSent(out.message)
@@ -734,16 +744,16 @@
 	)
 	if out.coin != nil {
 		select {
 		case <-h.ctx.Done():
 			return h.ctx.Err()
 		case h.coins <- WeakCoinOutput{Layer: session.lid, Coin: *out.coin}:
 		}
 		sessionCoin.Inc()
 	}
 	if out.result != nil {
 		select {
 		case <-h.ctx.Done():
 			return h.ctx.Err()
 		case h.results <- ConsensusOutput{Layer: session.lid, Proposals: out.result}:
 		}
 		sessionResult.Inc()
@@ -766,11 +776,11 @@
 	for _, signer := range session.signers {
 		atxid, err := atxs.GetIDByEpochAndNodeID(h.db, publish, signer.NodeID())
 		switch {
 		case errors.Is(err, sql.ErrNotFound):
 			// if atx is not registered for identity we will get sql.ErrNotFound
 		case err != nil:
 			h.log.Error("failed to get atx id by epoch and node id", zap.Error(err))
 			return []types.ProposalID{}
 		default:
 			own := h.atxsdata.Get(target, atxid)
 			if min == nil || (min != nil && own != nil && own.Height < min.Height) {
@@ -779,8 +789,8 @@
 		}
 	}
 	if min == nil {
 		h.log.Debug("no atxs in the requested epoch", zap.Uint32("epoch", session.lid.GetEpoch().Uint32()-1))
 		return []types.ProposalID{}
 	}
 
 	candidates := h.proposals.GetForLayer(session.lid)
@@ -808,8 +818,8 @@
 		}
 		header := h.atxsdata.Get(target, p.AtxID)
 		if header == nil {
 			h.log.Error("atx is not loaded", zap.Stringer("atxid", p.AtxID))
 			return []types.ProposalID{}
 		}
 		if header.BaseHeight >= min.Height {
 			// does not vote for future proposal
@@ -827,8 +837,8 @@
 			h.log.Warn("proposal has different beacon value",
 				zap.Uint32("lid", session.lid.Uint32()),
 				zap.Stringer("id", p.ID()),
-				zap.String("proposal_beacon", p.Beacon().ShortString()),
-				zap.String("epoch_beacon", session.beacon.ShortString()),
+				zap.Stringer("proposal_beacon", p.Beacon()),
+				zap.Stringer("epoch_beacon", session.beacon),
 			)
 		}
 	}
@@ -851,13 +861,13 @@
 	h.mu.Lock()
 	defer h.mu.Unlock()
 	for k, item := range h.messageCache {
 		if item.Layer < l {
 			// mark key for deletion
 			keys = append(keys, k)
 		}
 	}
 	for _, v := range keys {
 		delete(h.messageCache, v)
 	}
 }
 
@@ -876,47 +886,39 @@
 	vrfs    []*types.HareEligibility
 }
 
-type compactFunc func([]byte) []byte
-
-// compactFunc will truncate a given byte slice to a shorter
-// byte slice by reslicing.
-func compactTruncate(b []byte) []byte {
-	return b[:4]
-}
-
-func compactVrf(compacter compactFunc, v types.VrfSignature) (c types.CompactProposalID) {
-	b := compacter(v[:])
-	copy(c[:], b)
-	return c
-}
-
-func (h *Hare) compactProposals(compacter compactFunc, layer types.LayerID,
+func (h *Hare) compactProposals(layer types.LayerID,
 	proposals []*types.Proposal,
 ) []types.CompactProposalID {
 	compactProposals := make([]types.CompactProposalID, len(proposals))
 	for i, prop := range proposals {
 		vrf := prop.EligibilityProofs[0].Sig
-		compactProposals[i] = compactVrf(compacter, vrf)
+		var c types.CompactProposalID
+		copy(c[:], vrf[:4])
+		compactProposals[i] = c
 	}
 	return compactProposals
 }
 
-func (h *Hare) compactProposalIds(compacter compactFunc, layer types.LayerID,
+func (h *Hare) compactProposalIds(layer types.LayerID,
 	proposals []types.ProposalID,
 ) ([]types.CompactProposalID, error) {
 	compactProposals := make([]types.CompactProposalID, len(proposals))
 	for i, prop := range proposals {
 		fp := h.proposals.Get(layer, prop)
 		if fp == nil {
 			return nil, errCannotFindProposal
 		}
 
 		// we must handle this explicitly or we risk a panic on
 		// a nil slice access below
 		if len(fp.EligibilityProofs) == 0 {
 			return nil, errNoEligibilityProofs
 		}
-		compactProposals[i] = compactVrf(compacter, fp.EligibilityProofs[0].Sig)
+
+		var c types.CompactProposalID
+		copy(c[:], fp.EligibilityProofs[0].Sig[:4])
+
+		compactProposals[i] = c
 	}
 	return compactProposals, nil
 }
@@ -925,11 +927,3 @@
 	id      types.ProposalID
 	compact types.CompactProposalID
 }
-
-func sortProposalsTuple(i, j proposalTuple) int {
-	return sortProposalIds(i.id, j.id)
-}
-
-func sortProposalIds(i, j types.ProposalID) int {
-	return bytes.Compare(i[:], j[:])
-}