ceremonyclient/node/consensus/global/event_distributor.go

package global

import (
	"bytes"
	"context"
	"encoding/hex"
	"fmt"
	"math/big"
	"slices"
	"strings"
	"time"

	pcrypto "github.com/libp2p/go-libp2p/core/crypto"
	"github.com/libp2p/go-libp2p/core/peer"
	"github.com/multiformats/go-multiaddr"
	mn "github.com/multiformats/go-multiaddr/net"
	"github.com/pkg/errors"
	"github.com/prometheus/client_golang/prometheus"
	"github.com/shopspring/decimal"
	"go.uber.org/zap"
	"google.golang.org/grpc"
	"source.quilibrium.com/quilibrium/monorepo/config"
	"source.quilibrium.com/quilibrium/monorepo/lifecycle"
	"source.quilibrium.com/quilibrium/monorepo/node/consensus/provers"
	consensustime "source.quilibrium.com/quilibrium/monorepo/node/consensus/time"
	globalintrinsics "source.quilibrium.com/quilibrium/monorepo/node/execution/intrinsics/global"
	"source.quilibrium.com/quilibrium/monorepo/node/execution/intrinsics/global/compat"
	"source.quilibrium.com/quilibrium/monorepo/node/execution/intrinsics/token"
	"source.quilibrium.com/quilibrium/monorepo/node/p2p"
	"source.quilibrium.com/quilibrium/monorepo/protobufs"
	"source.quilibrium.com/quilibrium/monorepo/types/channel"
	typesconsensus "source.quilibrium.com/quilibrium/monorepo/types/consensus"
	"source.quilibrium.com/quilibrium/monorepo/types/crypto"
	"source.quilibrium.com/quilibrium/monorepo/types/schema"
	"source.quilibrium.com/quilibrium/monorepo/types/store"
)

func (e *GlobalConsensusEngine) eventDistributorLoop(
	ctx lifecycle.SignalerContext,
) {
	defer func() {
		if r := recover(); r != nil {
			e.logger.Error("fatal error encountered", zap.Any("panic", r))
			ctx.Throw(errors.Errorf("fatal unhandled error encountered: %v", r))
		}
	}()

	e.logger.Debug("starting event distributor")

	// Subscribe to events from the event distributor
	eventCh := e.eventDistributor.Subscribe("global")
	defer e.eventDistributor.Unsubscribe("global")

	for {
		select {
		case <-ctx.Done():
			return
		case <-e.quit:
			return
		case event, ok := <-eventCh:
			if !ok {
				e.logger.Error("event channel closed unexpectedly")
				return
			}

			e.logger.Debug("received event", zap.Int("event_type", int(event.Type)))

			// Handle the event based on its type
			switch event.Type {
			case typesconsensus.ControlEventGlobalNewHead:
				timer := prometheus.NewTimer(globalCoordinationDuration)

				// New global frame has been selected as the head by the time reel
				if data, ok := event.Data.(*consensustime.GlobalEvent); ok &&
					data.Frame != nil {
					e.lastObservedFrame.Store(data.Frame.Header.FrameNumber)
					e.logger.Info(
						"received new global head event",
						zap.Uint64("frame_number", data.Frame.Header.FrameNumber),
					)

					e.flushDeferredGlobalMessages(data.Frame.GetRank() + 1)

					// Check shard coverage
					if err := e.checkShardCoverage(
						data.Frame.Header.FrameNumber,
					); err != nil {
						e.logger.Error("failed to check shard coverage", zap.Error(err))
					}

					// Update global coordination metrics
					globalCoordinationTotal.Inc()
					timer.ObserveDuration()

					_, err := e.signerRegistry.GetIdentityKey(e.GetPeerInfo().PeerId)
					if err != nil && !e.hasSentKeyBundle {
						e.hasSentKeyBundle = true
						e.publishKeyRegistry()
					}

					if e.proposer != nil && !e.config.Engine.ArchiveMode {
						workers, err := e.workerManager.RangeWorkers()
						if err != nil {
							e.logger.Error("could not retrieve workers", zap.Error(err))
						} else {
							if len(workers) == 0 {
								e.logger.Error("no workers detected for allocation")
							}
							allAllocated := true
							needsProposals := false
							for _, w := range workers {
								allAllocated = allAllocated && w.Allocated
								if len(w.Filter) == 0 {
									needsProposals = true
								}
							}
							if needsProposals || !allAllocated {
								e.evaluateForProposals(ctx, data, needsProposals)
							} else {
								self, effectiveSeniority := e.allocationContext()
								e.checkExcessPendingJoins(self, data.Frame.Header.FrameNumber)
								e.logAllocationStatusOnly(ctx, data, self, effectiveSeniority)
							}
						}
					}
				}

			case typesconsensus.ControlEventGlobalEquivocation:
				// Handle equivocation by constructing and publishing a ProverKick
				// message
				if data, ok := event.Data.(*consensustime.GlobalEvent); ok &&
					data.Frame != nil && data.OldHead != nil {
					e.logger.Warn(
						"received equivocating frame",
						zap.Uint64("frame_number", data.Frame.Header.FrameNumber),
					)

					// The equivocating prover is the one who signed the new frame
					if data.Frame.Header != nil &&
						data.Frame.Header.PublicKeySignatureBls48581 != nil &&
						data.Frame.Header.PublicKeySignatureBls48581.PublicKey != nil {

						kickedProverPublicKey :=
							data.Frame.Header.PublicKeySignatureBls48581.PublicKey.KeyValue

						// Serialize both conflicting frame headers
						conflictingFrame1, err := data.OldHead.Header.ToCanonicalBytes()
						if err != nil {
							e.logger.Error(
								"failed to marshal old frame header",
								zap.Error(err),
							)
							continue
						}

						conflictingFrame2, err := data.Frame.Header.ToCanonicalBytes()
						if err != nil {
							e.logger.Error(
								"failed to marshal new frame header",
								zap.Error(err),
							)
							continue
						}

						// Create the ProverKick message using the intrinsic struct
						proverKick, err := globalintrinsics.NewProverKick(
							data.Frame.Header.FrameNumber,
							kickedProverPublicKey,
							conflictingFrame1,
							conflictingFrame2,
							e.blsConstructor,
							e.frameProver,
							e.hypergraph,
							schema.NewRDFMultiprover(
								&schema.TurtleRDFParser{},
								e.inclusionProver,
							),
							e.proverRegistry,
							e.clockStore,
						)
						if err != nil {
							e.logger.Error(
								"failed to construct prover kick",
								zap.Error(err),
							)
							continue
						}

						err = proverKick.Prove(data.Frame.Header.FrameNumber)
						if err != nil {
							e.logger.Error(
								"failed to prove prover kick",
								zap.Error(err),
							)
							continue
						}

						// Serialize the ProverKick to the request form
						kickBytes, err := proverKick.ToRequestBytes()
						if err != nil {
							e.logger.Error(
								"failed to serialize prover kick",
								zap.Error(err),
							)
							continue
						}

						// Publish the kick message
						if err := e.pubsub.PublishToBitmask(
							GLOBAL_PROVER_BITMASK,
							kickBytes,
						); err != nil {
							e.logger.Error("failed to publish prover kick", zap.Error(err))
						} else {
							e.logger.Info(
								"published prover kick for equivocation",
								zap.Uint64("frame_number", data.Frame.Header.FrameNumber),
								zap.String(
									"kicked_prover",
									hex.EncodeToString(kickedProverPublicKey),
								),
							)
						}
					}
				}

			case typesconsensus.ControlEventCoverageHalt:
				data, ok := event.Data.(*typesconsensus.CoverageEventData)
				if ok && data.Message != "" {
					e.logger.Error(data.Message)
					e.halt()
					go func() {
						for {
							select {
							case <-ctx.Done():
								return
							case <-time.After(10 * time.Second):
								e.logger.Error(
									"full halt detected, leaving system in halted state until recovery",
								)
							}
						}
					}()
				}

			case typesconsensus.ControlEventHalt:
				data, ok := event.Data.(*typesconsensus.ErrorEventData)
				if ok && data.Error != nil {
					e.logger.Error(
						"full halt detected, leaving system in halted state",
						zap.Error(data.Error),
					)
					e.halt()
					go func() {
						for {
							select {
							case <-ctx.Done():
								return
							case <-time.After(10 * time.Second):
								e.logger.Error(
									"full halt detected, leaving system in halted state",
									zap.Error(data.Error),
								)
							}
						}
					}()
				}

			default:
				e.logger.Debug(
					"received unhandled event type",
					zap.Int("event_type", int(event.Type)),
				)
			}
		}
	}
}

const pendingFilterGraceFrames = 720

func (e *GlobalConsensusEngine) emitCoverageEvent(
	eventType typesconsensus.ControlEventType,
	data *typesconsensus.CoverageEventData,
) {
	event := typesconsensus.ControlEvent{
		Type: eventType,
		Data: data,
	}

	go e.eventDistributor.Publish(event)

	e.logger.Info(
		"emitted coverage event",
		zap.String("type", fmt.Sprintf("%d", eventType)),
		zap.String("shard_address", hex.EncodeToString(data.ShardAddress)),
		zap.Int("prover_count", data.ProverCount),
		zap.String("message", data.Message),
	)
}

func (e *GlobalConsensusEngine) emitMergeEvent(
	data *typesconsensus.ShardMergeEventData,
) {
	event := typesconsensus.ControlEvent{
		Type: typesconsensus.ControlEventShardMergeEligible,
		Data: data,
	}

	go e.eventDistributor.Publish(event)

	e.logger.Info(
		"emitted merge eligible event",
		zap.Int("shard_count", len(data.ShardAddresses)),
		zap.Int("total_provers", data.TotalProvers),
		zap.Uint64("attested_storage", data.AttestedStorage),
		zap.Uint64("required_storage", data.RequiredStorage),
	)
}

func (e *GlobalConsensusEngine) emitSplitEvent(
	data *typesconsensus.ShardSplitEventData,
) {
	event := typesconsensus.ControlEvent{
		Type: typesconsensus.ControlEventShardSplitEligible,
		Data: data,
	}

	go e.eventDistributor.Publish(event)

	e.logger.Info(
		"emitted split eligible event",
		zap.String("shard_address", hex.EncodeToString(data.ShardAddress)),
		zap.Int("proposed_shard_count", len(data.ProposedShards)),
		zap.Int("prover_count", data.ProverCount),
		zap.Uint64("attested_storage", data.AttestedStorage),
	)
}

func (e *GlobalConsensusEngine) emitAlertEvent(alertMessage string) {
	event := typesconsensus.ControlEvent{
		Type: typesconsensus.ControlEventHalt,
		Data: &typesconsensus.ErrorEventData{
			Error: errors.New(alertMessage),
		},
	}

	go e.eventDistributor.Publish(event)

	e.logger.Info("emitted alert message")
}

func (e *GlobalConsensusEngine) estimateSeniorityFromConfig() uint64 {
	peerIds := []string{}
	peerIds = append(peerIds, peer.ID(e.pubsub.GetPeerID()).String())
	if len(e.config.Engine.MultisigProverEnrollmentPaths) != 0 {
		for _, conf := range e.config.Engine.MultisigProverEnrollmentPaths {
			extraConf, err := config.LoadConfig(conf, "", false)
			if err != nil {
				e.logger.Error("could not load config", zap.Error(err))
				continue
			}

			peerPrivKey, err := hex.DecodeString(extraConf.P2P.PeerPrivKey)
			if err != nil {
				e.logger.Error("could not decode peer key", zap.Error(err))
				continue
			}

			privKey, err := pcrypto.UnmarshalEd448PrivateKey(peerPrivKey)
			if err != nil {
				e.logger.Error("could not unmarshal peer key", zap.Error(err))
				continue
			}

			pub := privKey.GetPublic()
			id, err := peer.IDFromPublicKey(pub)
			if err != nil {
				e.logger.Error("could not unmarshal peerid", zap.Error(err))
				continue
			}

			peerIds = append(peerIds, id.String())
		}
	}
	seniorityBI := compat.GetAggregatedSeniority(peerIds)
	return seniorityBI.Uint64()
}

func (e *GlobalConsensusEngine) evaluateForProposals(
	ctx context.Context,
	data *consensustime.GlobalEvent,
	allowProposals bool,
) {
	self, effectiveSeniority := e.allocationContext()
	e.reconcileWorkerAllocations(data.Frame.Header.FrameNumber, self)
	e.checkExcessPendingJoins(self, data.Frame.Header.FrameNumber)
	canPropose, skipReason := e.joinProposalReady(data.Frame.Header.FrameNumber)

	snapshot, ok := e.collectAllocationSnapshot(
		ctx,
		data,
		self,
		effectiveSeniority,
	)
	if !ok {
		return
	}

	e.logAllocationStatus(snapshot)
	pendingFilters := snapshot.pendingFilters
	proposalDescriptors := snapshot.proposalDescriptors
	decideDescriptors := snapshot.decideDescriptors
	worldBytes := snapshot.worldBytes

	if len(proposalDescriptors) != 0 && allowProposals {
		if canPropose {
			proposals, err := e.proposer.PlanAndAllocate(
				uint64(data.Frame.Header.Difficulty),
				proposalDescriptors,
				100,
				worldBytes,
				data.Frame.Header.FrameNumber,
			)
			if err != nil {
				e.logger.Error("could not plan shard allocations", zap.Error(err))
			} else {
				if len(proposals) > 0 {
					e.lastJoinAttemptFrame.Store(data.Frame.Header.FrameNumber)
				}
				expectedRewardSum := big.NewInt(0)
				for _, p := range proposals {
					expectedRewardSum.Add(expectedRewardSum, p.ExpectedReward)
				}
				raw := decimal.NewFromBigInt(expectedRewardSum, 0)
				rewardInQuilPerInterval := raw.Div(decimal.NewFromInt(8000000000))
				rewardInQuilPerDay := rewardInQuilPerInterval.Mul(
					decimal.NewFromInt(24 * 60 * 6),
				)
				e.logger.Info(
					"proposed joins",
					zap.Int("shard_proposals", len(proposals)),
					zap.String(
						"estimated_reward_per_interval",
						rewardInQuilPerInterval.String(),
					),
					zap.String(
						"estimated_reward_per_day",
						rewardInQuilPerDay.String(),
					),
				)
			}
		} else {
			e.logger.Info(
				"skipping join proposals",
				zap.String("reason", skipReason),
				zap.Uint64("frame_number", data.Frame.Header.FrameNumber),
			)
		}
	} else if len(proposalDescriptors) != 0 && !allowProposals {
		e.logger.Info(
			"skipping join proposals",
			zap.String("reason", "all workers already assigned filters"),
			zap.Uint64("frame_number", data.Frame.Header.FrameNumber),
		)
	}
	if len(pendingFilters) != 0 {
		if err := e.proposer.DecideJoins(
			uint64(data.Frame.Header.Difficulty),
			decideDescriptors,
			pendingFilters,
			worldBytes,
		); err != nil {
			e.logger.Error("could not decide shard allocations", zap.Error(err))
		} else {
			e.logger.Info(
				"decided on joins",
				zap.Int("joins", len(pendingFilters)),
			)
		}
	}
}

type allocationSnapshot struct {
	shardsPending       int
	awaitingFrames      []string
	shardsLeaving       int
	shardsActive        int
	shardsPaused        int
	shardDivisions      int
	logicalShards       int
	pendingFilters      [][]byte
	proposalDescriptors []provers.ShardDescriptor
	decideDescriptors   []provers.ShardDescriptor
	worldBytes          *big.Int
}

func (s *allocationSnapshot) statusFields() []zap.Field {
	if s == nil {
		return nil
	}

	return []zap.Field{
		zap.Int("pending_joins", s.shardsPending),
		zap.String("pending_join_frames", strings.Join(s.awaitingFrames, ", ")),
		zap.Int("pending_leaves", s.shardsLeaving),
		zap.Int("active", s.shardsActive),
		zap.Int("paused", s.shardsPaused),
		zap.Int("network_shards", s.shardDivisions),
		zap.Int("network_logical_shards", s.logicalShards),
	}
}

func (s *allocationSnapshot) proposalSnapshotFields() []zap.Field {
	if s == nil {
		return nil
	}

	return []zap.Field{
		zap.Int("proposal_candidates", len(s.proposalDescriptors)),
		zap.Int("pending_confirmations", len(s.pendingFilters)),
		zap.Int("decide_descriptors", len(s.decideDescriptors)),
	}
}

func (e *GlobalConsensusEngine) reconcileWorkerAllocations(
	frameNumber uint64,
	self *typesconsensus.ProverInfo,
) {
	if e.workerManager == nil {
		return
	}

	workers, err := e.workerManager.RangeWorkers()
	if err != nil {
		e.logger.Warn("could not load workers for reconciliation", zap.Error(err))
		return
	}

	filtersToWorkers := make(map[string]*store.WorkerInfo, len(workers))
	freeWorkers := make([]*store.WorkerInfo, 0, len(workers))
	for _, worker := range workers {
		if worker == nil {
			continue
		}
		if len(worker.Filter) == 0 {
			freeWorkers = append(freeWorkers, worker)
			continue
		}
		filtersToWorkers[string(worker.Filter)] = worker
	}

	seenFilters := make(map[string]struct{})
	if self != nil {
		for _, alloc := range self.Allocations {
			if len(alloc.ConfirmationFilter) == 0 {
				continue
			}

			key := string(alloc.ConfirmationFilter)
			worker, ok := filtersToWorkers[key]
			if !ok {
				if len(freeWorkers) == 0 {
					e.logger.Warn(
						"no free worker available for registry allocation",
						zap.String("filter", hex.EncodeToString(alloc.ConfirmationFilter)),
					)
					continue
				}
				worker = freeWorkers[0]
				freeWorkers = freeWorkers[1:]
				worker.Filter = slices.Clone(alloc.ConfirmationFilter)
			}

			seenFilters[key] = struct{}{}

			desiredAllocated := alloc.Status == typesconsensus.ProverStatusActive ||
				alloc.Status == typesconsensus.ProverStatusPaused

			pendingFrame := alloc.JoinFrameNumber
			if desiredAllocated {
				pendingFrame = 0
			}

			if worker.Allocated != desiredAllocated ||
				worker.PendingFilterFrame != pendingFrame {
				worker.Allocated = desiredAllocated
				worker.PendingFilterFrame = pendingFrame
				if err := e.workerManager.RegisterWorker(worker); err != nil {
					e.logger.Warn(
						"failed to update worker allocation state",
						zap.Uint("core_id", worker.CoreId),
						zap.Error(err),
					)
				}
			}
		}
	}

	for _, worker := range workers {
		if worker == nil || len(worker.Filter) == 0 {
			continue
		}
		if _, ok := seenFilters[string(worker.Filter)]; ok {
			continue
		}

		if worker.PendingFilterFrame != 0 {
			if frameNumber <= worker.PendingFilterFrame {
				continue
			}
			if frameNumber-worker.PendingFilterFrame < pendingFilterGraceFrames {
				continue
			}
		}

		if worker.PendingFilterFrame == 0 && self == nil {
			continue
		}

		worker.Filter = nil
		worker.Allocated = false
		worker.PendingFilterFrame = 0
		if err := e.workerManager.RegisterWorker(worker); err != nil {
			e.logger.Warn(
				"failed to clear stale worker filter",
				zap.Uint("core_id", worker.CoreId),
				zap.Error(err),
			)
		}
	}
}

func (e *GlobalConsensusEngine) collectAllocationSnapshot(
	ctx context.Context,
	data *consensustime.GlobalEvent,
	self *typesconsensus.ProverInfo,
	effectiveSeniority uint64,
) (*allocationSnapshot, bool) {
	appShards, err := e.shardsStore.RangeAppShards()
	if err != nil {
		e.logger.Error("could not obtain app shard info", zap.Error(err))
		return nil, false
	}

	// consolidate into high level L2 shards:
	shardMap := map[string]store.ShardInfo{}
	for _, s := range appShards {
		shardMap[string(s.L2)] = s
	}

	shards := []store.ShardInfo{}
	for _, s := range shardMap {
		shards = append(shards, store.ShardInfo{
			L1: s.L1,
			L2: s.L2,
		})
	}

	registry, err := e.keyStore.GetKeyRegistryByProver(data.Frame.Header.Prover)
	if err != nil {
		e.logger.Info(
			"awaiting key registry info for prover",
			zap.String(
				"prover_address",
				hex.EncodeToString(data.Frame.Header.Prover),
			),
		)
		return nil, false
	}

	if registry.IdentityKey == nil || registry.IdentityKey.KeyValue == nil {
		e.logger.Info("key registry info missing identity of prover")
		return nil, false
	}

	pub, err := pcrypto.UnmarshalEd448PublicKey(registry.IdentityKey.KeyValue)
	if err != nil {
		e.logger.Warn("error unmarshaling identity key", zap.Error(err))
		return nil, false
	}

	peerId, err := peer.IDFromPublicKey(pub)
	if err != nil {
		e.logger.Warn("error deriving peer id", zap.Error(err))
		return nil, false
	}

	info := e.peerInfoManager.GetPeerInfo([]byte(peerId))
	if info == nil {
		e.logger.Info(
			"no peer info known yet",
			zap.String("peer", peer.ID(peerId).String()),
		)
		return nil, false
	}

	if len(info.Reachability) == 0 {
		e.logger.Info(
			"no reachability info known yet",
			zap.String("peer", peer.ID(peerId).String()),
		)
		return nil, false
	}

	var client protobufs.GlobalServiceClient = nil
	if len(info.Reachability[0].StreamMultiaddrs) > 0 {
		s := info.Reachability[0].StreamMultiaddrs[0]
		creds, err := p2p.NewPeerAuthenticator(
			e.logger,
			e.config.P2P,
			nil,
			nil,
			nil,
			nil,
			[][]byte{[]byte(peerId)},
			map[string]channel.AllowedPeerPolicyType{},
			map[string]channel.AllowedPeerPolicyType{},
		).CreateClientTLSCredentials([]byte(peerId))
		if err != nil {
			return nil, false
		}

		ma, err := multiaddr.StringCast(s)
		if err != nil {
			return nil, false
		}

		mga, err := mn.ToNetAddr(ma)
		if err != nil {
			return nil, false
		}

		cc, err := grpc.NewClient(
			mga.String(),
			grpc.WithTransportCredentials(creds),
		)
		if err != nil {
			e.logger.Debug(
				"could not establish direct channel, trying next multiaddr",
				zap.String("peer", peer.ID(peerId).String()),
				zap.String("multiaddr", ma.String()),
				zap.Error(err),
			)
			return nil, false
		}
		defer func() {
			if err := cc.Close(); err != nil {
				e.logger.Error("error while closing connection", zap.Error(err))
			}
		}()

		client = protobufs.NewGlobalServiceClient(cc)
	}

	if client == nil {
		e.logger.Debug("could not get app shards from prover")
		return nil, false
	}

	worldBytes := big.NewInt(0)
	shardsPending := 0
	shardsActive := 0
	shardsLeaving := 0
	shardsPaused := 0
	logicalShards := 0
	shardDivisions := 0
	awaitingFrame := map[uint64]struct{}{}
	pendingFilters := [][]byte{}
	proposalDescriptors := []provers.ShardDescriptor{}
	decideDescriptors := []provers.ShardDescriptor{}

	for _, shardInfo := range shards {
		resp, err := e.getAppShardsFromProver(
			client,
			slices.Concat(shardInfo.L1, shardInfo.L2),
		)
		if err != nil {
			e.logger.Debug("could not get app shards from prover", zap.Error(err))
			return nil, false
		}

		for _, shard := range resp.Info {
			shardDivisions++
			worldBytes = worldBytes.Add(worldBytes, new(big.Int).SetBytes(shard.Size))
			bp := slices.Clone(shardInfo.L2)
			for _, p := range shard.Prefix {
				bp = append(bp, byte(p))
			}

			prs, err := e.proverRegistry.GetProvers(bp)
			if err != nil {
				e.logger.Error("failed to get provers", zap.Error(err))
				continue
			}

			allocated := false
			pending := false
			if self != nil {
				for _, allocation := range self.Allocations {
					if bytes.Equal(allocation.ConfirmationFilter, bp) {
						allocated = allocation.Status != 4
						if allocation.Status == typesconsensus.ProverStatusJoining {
							shardsPending++
							awaitingFrame[allocation.JoinFrameNumber+360] = struct{}{}
						}
						if allocation.Status == typesconsensus.ProverStatusActive {
							shardsActive++
						}
						if allocation.Status == typesconsensus.ProverStatusLeaving {
							shardsLeaving++
						}
						if allocation.Status == typesconsensus.ProverStatusPaused {
							shardsPaused++
						}
						if e.config.P2P.Network != 0 ||
							data.Frame.Header.FrameNumber > token.FRAME_2_1_EXTENDED_ENROLL_END {
							pending = allocation.Status ==
								typesconsensus.ProverStatusJoining &&
								allocation.JoinFrameNumber+360 <= data.Frame.Header.FrameNumber
						}
					}
				}
			}

			size := new(big.Int).SetBytes(shard.Size)
			if size.Cmp(big.NewInt(0)) == 0 {
				continue
			}

			logicalShards += int(shard.DataShards)

			above := []*typesconsensus.ProverInfo{}
			for _, i := range prs {
				for _, a := range i.Allocations {
					if !bytes.Equal(a.ConfirmationFilter, bp) {
						continue
					}
					if a.Status == typesconsensus.ProverStatusActive ||
						a.Status == typesconsensus.ProverStatusJoining {
						if i.Seniority >= effectiveSeniority {
							above = append(above, i)
						}
						break
					}
				}
			}

			if allocated && pending {
				pendingFilters = append(pendingFilters, bp)
			}
			if !allocated {
				proposalDescriptors = append(
					proposalDescriptors,
					provers.ShardDescriptor{
						Filter: bp,
						Size:   size.Uint64(),
						Ring:   uint8(len(above) / 8),
						Shards: shard.DataShards,
					},
				)
			}
			decideDescriptors = append(
				decideDescriptors,
				provers.ShardDescriptor{
					Filter: bp,
					Size:   size.Uint64(),
					Ring:   uint8(len(above) / 8),
					Shards: shard.DataShards,
				},
			)
		}
	}

	awaitingFrames := []string{}
	for frame := range awaitingFrame {
		awaitingFrames = append(awaitingFrames, fmt.Sprintf("%d", frame))
	}

	return &allocationSnapshot{
		shardsPending:       shardsPending,
		awaitingFrames:      awaitingFrames,
		shardsLeaving:       shardsLeaving,
		shardsActive:        shardsActive,
		shardsPaused:        shardsPaused,
		shardDivisions:      shardDivisions,
		logicalShards:       logicalShards,
		pendingFilters:      pendingFilters,
		proposalDescriptors: proposalDescriptors,
		decideDescriptors:   decideDescriptors,
		worldBytes:          worldBytes,
	}, true
}

func (e *GlobalConsensusEngine) logAllocationStatus(
	snapshot *allocationSnapshot,
) {
	if snapshot == nil {
		return
	}

	e.logger.Info(
		"status for allocations",
		snapshot.statusFields()...,
	)

	e.logger.Debug(
		"proposal evaluation snapshot",
		snapshot.proposalSnapshotFields()...,
	)
}

func (e *GlobalConsensusEngine) logAllocationStatusOnly(
	ctx context.Context,
	data *consensustime.GlobalEvent,
	self *typesconsensus.ProverInfo,
	effectiveSeniority uint64,
) {
	snapshot, ok := e.collectAllocationSnapshot(
		ctx,
		data,
		self,
		effectiveSeniority,
	)
	if !ok || snapshot == nil {
		e.logger.Info(
			"all workers already allocated or pending; skipping proposal cycle",
		)
		return
	}

	e.logger.Info(
		"all workers already allocated or pending; skipping proposal cycle",
		snapshot.statusFields()...,
	)
	e.logAllocationStatus(snapshot)
}

func (e *GlobalConsensusEngine) allocationContext() (
	*typesconsensus.ProverInfo,
	uint64,
) {
	self, err := e.proverRegistry.GetProverInfo(e.getProverAddress())
	if err != nil || self == nil {
		return nil, e.estimateSeniorityFromConfig()
	}
	return self, self.Seniority
}

func (e *GlobalConsensusEngine) checkExcessPendingJoins(
	self *typesconsensus.ProverInfo,
	frameNumber uint64,
) {
	excessFilters := e.selectExcessPendingFilters(self)
	if len(excessFilters) != 0 {
		e.logger.Debug(
			"identified excess pending joins",
			zap.Int("excess_count", len(excessFilters)),
			zap.Uint64("frame_number", frameNumber),
		)
		e.rejectExcessPending(excessFilters, frameNumber)
		return
	}

	e.logger.Debug(
		"no excess pending joins detected",
		zap.Uint64("frame_number", frameNumber),
	)
}

func (e *GlobalConsensusEngine) publishKeyRegistry() {
	vk, err := e.keyManager.GetAgreementKey("q-view-key")
	if err != nil {
		vk, err = e.keyManager.CreateAgreementKey(
			"q-view-key",
			crypto.KeyTypeDecaf448,
		)
		if err != nil {
			e.logger.Error("could not publish key registry", zap.Error(err))
			return
		}
	}

	sk, err := e.keyManager.GetAgreementKey("q-spend-key")
	if err != nil {
		sk, err = e.keyManager.CreateAgreementKey(
			"q-spend-key",
			crypto.KeyTypeDecaf448,
		)
		if err != nil {
			e.logger.Error("could not publish key registry", zap.Error(err))
			return
		}
	}

	pk, err := e.keyManager.GetSigningKey("q-prover-key")
	if err != nil {
		pk, _, err = e.keyManager.CreateSigningKey(
			"q-prover-key",
			crypto.KeyTypeBLS48581G1,
		)
		if err != nil {
			e.logger.Error("could not publish key registry", zap.Error(err))
			return
		}
	}

	onion, err := e.keyManager.GetAgreementKey("q-onion-key")
	if err != nil {
		onion, err = e.keyManager.CreateAgreementKey(
			"q-onion-key",
			crypto.KeyTypeX448,
		)
		if err != nil {
			e.logger.Error("could not publish key registry", zap.Error(err))
			return
		}
	}

	sig, err := e.pubsub.SignMessage(
		slices.Concat(
			[]byte("KEY_REGISTRY"),
			pk.Public().([]byte),
		),
	)
	if err != nil {
		e.logger.Error("could not publish key registry", zap.Error(err))
		return
	}
	sigp, err := pk.SignWithDomain(
		e.pubsub.GetPublicKey(),
		[]byte("KEY_REGISTRY"),
	)
	if err != nil {
		e.logger.Error("could not publish key registry", zap.Error(err))
		return
	}
	sigvk, err := e.pubsub.SignMessage(
		slices.Concat(
			[]byte("KEY_REGISTRY"),
			vk.Public(),
		),
	)
	if err != nil {
		e.logger.Error("could not publish key registry", zap.Error(err))
		return
	}
	sigsk, err := e.pubsub.SignMessage(
		slices.Concat(
			[]byte("KEY_REGISTRY"),
			sk.Public(),
		),
	)
	if err != nil {
		e.logger.Error("could not publish key registry", zap.Error(err))
		return
	}
	sigonion, err := e.pubsub.SignMessage(
		slices.Concat(
			[]byte("KEY_REGISTRY"),
			onion.Public(),
		),
	)
	if err != nil {
		e.logger.Error("could not publish key registry", zap.Error(err))
		return
	}
	registry := &protobufs.KeyRegistry{
		LastUpdated: uint64(time.Now().UnixMilli()),
		IdentityKey: &protobufs.Ed448PublicKey{
			KeyValue: e.pubsub.GetPublicKey(),
		},
		ProverKey: &protobufs.BLS48581G2PublicKey{
			KeyValue: pk.Public().([]byte),
		},
		IdentityToProver: &protobufs.Ed448Signature{
			Signature: sig,
		},
		ProverToIdentity: &protobufs.BLS48581Signature{
			Signature: sigp,
		},
		KeysByPurpose: map[string]*protobufs.KeyCollection{
			"ONION_ROUTING": &protobufs.KeyCollection{
				KeyPurpose: "ONION_ROUTING",
				X448Keys: []*protobufs.SignedX448Key{{
					Key: &protobufs.X448PublicKey{
						KeyValue: onion.Public(),
					},
					ParentKeyAddress: e.pubsub.GetPeerID(),
					Signature: &protobufs.SignedX448Key_Ed448Signature{
						Ed448Signature: &protobufs.Ed448Signature{
							Signature: sigonion,
						},
					},
				}},
			},
			"view": &protobufs.KeyCollection{
				KeyPurpose: "view",
				Decaf448Keys: []*protobufs.SignedDecaf448Key{{
					Key: &protobufs.Decaf448PublicKey{
						KeyValue: vk.Public(),
					},
					ParentKeyAddress: e.pubsub.GetPeerID(),
					Signature: &protobufs.SignedDecaf448Key_Ed448Signature{
						Ed448Signature: &protobufs.Ed448Signature{
							Signature: sigvk,
						},
					},
				}},
			},
			"spend": &protobufs.KeyCollection{
				KeyPurpose: "spend",
				Decaf448Keys: []*protobufs.SignedDecaf448Key{{
					Key: &protobufs.Decaf448PublicKey{
						KeyValue: sk.Public(),
					},
					ParentKeyAddress: e.pubsub.GetPeerID(),
					Signature: &protobufs.SignedDecaf448Key_Ed448Signature{
						Ed448Signature: &protobufs.Ed448Signature{
							Signature: sigsk,
						},
					},
				}},
			},
		},
	}
	kr, err := registry.ToCanonicalBytes()
	if err != nil {
		e.logger.Error("could not publish key registry", zap.Error(err))
		return
	}
	e.pubsub.PublishToBitmask(
		GLOBAL_PEER_INFO_BITMASK,
		kr,
	)
}

func (e *GlobalConsensusEngine) getAppShardsFromProver(
	client protobufs.GlobalServiceClient,
	shardKey []byte,
) (
	*protobufs.GetAppShardsResponse,
	error,
) {
	getCtx, cancelGet := context.WithTimeout(
		context.Background(),
		e.config.Engine.SyncTimeout,
	)
	response, err := client.GetAppShards(
		getCtx,
		&protobufs.GetAppShardsRequest{
			ShardKey: shardKey, // buildutils:allow-slice-alias slice is static
		},
		// The message size limits are swapped because the server is the one
		// sending the data.
		grpc.MaxCallRecvMsgSize(
			e.config.Engine.SyncMessageLimits.MaxSendMsgSize,
		),
		grpc.MaxCallSendMsgSize(
			e.config.Engine.SyncMessageLimits.MaxRecvMsgSize,
		),
	)
	cancelGet()
	if err != nil {
		return nil, err
	}

	if response == nil {
		return nil, err
	}

	return response, nil
}