<commit_before>/*
Licensed to the Apache Software Foundation (ASF) under one
or more contributor license agreements.  See the NOTICE file
distributed with this work for additional information
regarding copyright ownership.  The ASF licenses this file
to you under the Apache License, Version 2.0 (the
"License"); you may not use this file except in compliance
with the License.  You may obtain a copy of the License at

  http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing,
software distributed under the License is distributed on an
"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
KIND, either express or implied.  See the License for the
specific language governing permissions and limitations
under the License.
*/

package pbft

import (
	"encoding/base64"
	"fmt"
	"strconv"
	"strings"

	"github.com/golang/protobuf/proto"
	"github.com/openblockchain/obc-peer/openchain/consensus"
	"github.com/openblockchain/obc-peer/openchain/util"
	pb "github.com/openblockchain/obc-peer/protos"

	"github.com/op/go-logging"
	"github.com/spf13/viper"
)

// =============================================================================
// Constants
// =============================================================================

const configPrefix = "OPENCHAIN_PBFT"

// =============================================================================
// Init
// =============================================================================

var logger *logging.Logger // package-level logger
var pluginInstance *Plugin // the Plugin is a singleton

func init() {
	logger = logging.MustGetLogger("consensus/pbft")
}

// =============================================================================
// Custom structure definitions go here
// =============================================================================

// Plugin carries fields related to the consensus algorithm implementation.
type Plugin struct {
	// internal data
	config *viper.Viper  // link to the plugin config file
	cpi    consensus.CPI // link to the CPI

	// PBFT data
	activeView   bool              // view change happening
	f            uint              // number of faults we can tolerate
	h            uint64            // low watermark
	id           uint64            // replica ID; PBFT `i`
	K            uint64            // checkpoint period
	L            uint64            // log size
	lastExec     uint64            // last request we executed
	replicaCount uint              // number of replicas; PBFT `|R|`
	seqNo        uint64            // PBFT "n", strictly monotonic increasing sequence number
	view         uint64            // current view
	chkpts       map[uint64]string // state checkpoints; map lastExec to global hash
	pset         map[uint64]*ViewChange_PQ
	qset         map[qidx]*ViewChange_PQ

	// Implementation of PBFT `in`
	certStore       map[msgID]*msgCert    // track quorum certificates for requests
	reqStore        map[string]*Request   // track requests
	checkpointStore map[Checkpoint]bool   // track checkpoints as set
	viewChangeStore map[vcidx]*ViewChange // track view-change messages
	lastNewView     NewView               // track last new-view we received or sent
}

type qidx struct {
	d string
	n uint64
}

type msgID struct { // our index through certStore
	v uint64
	n uint64
}

type msgCert struct {
	prePrepare  *PrePrepare
	sentPrepare bool
	prepare     []*Prepare
	sentCommit  bool
	commit      []*Commit
}

type vcidx struct {
	v  uint64
	id uint64
}

// =============================================================================
// Constructors go here
// =============================================================================

// GetPlugin returns the handle to the Plugin singleton and updates the CPI if necessary.
func GetPlugin(c consensus.CPI) *Plugin {
	if pluginInstance == nil {
		pluginInstance = New(c)
	} else {
		pluginInstance.cpi = c // otherwise, just update the CPI
	}
	return pluginInstance
}

// New creates an plugin-specific structure that acts as the ConsenusHandler's consenter.
func New(c consensus.CPI) *Plugin {
	instance := &Plugin{}
	instance.cpi = c

	// setup the link to the config file
	instance.config = viper.New()

	// for environment variables
	instance.config.SetEnvPrefix(configPrefix)
	instance.config.AutomaticEnv()
	replacer := strings.NewReplacer(".", "_")
	instance.config.SetEnvKeyReplacer(replacer)

	instance.config.SetConfigName("config")
	instance.config.AddConfigPath("./")
	instance.config.AddConfigPath("./openchain/consensus/pbft/")
	err := instance.config.ReadInConfig()
	if err != nil {
		panic(fmt.Errorf("Fatal error reading consensus algo config: %s", err))
	}

	// TODO Initialize the algorithm here
	// You may want to set the fields of `instance` using `instance.GetParam()`.

	// In dev/debugging mode you are expected to override the config values
	// with the environment variable OPENCHAIN_PBFT_X_Y

	// replica ID
	paramID, err := instance.getParam("replica.id")
	if err != nil {
		panic(fmt.Errorf("No ID assigned to the replica: %s", err))
	}
	instance.id, err = strconv.ParseUint(paramID, 10, 64)
	if err != nil {
		panic(fmt.Errorf("Cannot convert config ID to uint64: %s", err))
	}
	// byzantine nodes
	paramF, err := instance.getParam("general.f")
	if err != nil {
		panic(fmt.Errorf("No f defined in the config file: %s", err))
	}
	f, err := strconv.ParseUint(paramF, 10, 0)
	if err != nil {
		panic(fmt.Errorf("Cannot convert config f to uint64: %s", err))
	}
	instance.f = uint(f)
	// replica count
	instance.replicaCount = 3*instance.f + 1
	// checkpoint period
	paramK, err := instance.getParam("general.K")
	if err != nil {
		panic(fmt.Errorf("Checkpoint period is not defined: %s", err))
	}
	instance.K, err = strconv.ParseUint(paramK, 10, 64)
	if err != nil {
		panic(fmt.Errorf("Cannot convert config checkpoint period to uint64: %s", err))
	}
	// log size
	instance.L = 2 * instance.K

	instance.activeView = true

	// init the logs
	instance.certStore = make(map[msgID]*msgCert)
	instance.reqStore = make(map[string]*Request)
	instance.checkpointStore = make(map[Checkpoint]bool)
	instance.viewChangeStore = make(map[vcidx]*ViewChange)
	instance.chkpts = make(map[uint64]string)
	instance.pset = make(map[uint64]*ViewChange_PQ)
	instance.qset = make(map[qidx]*ViewChange_PQ)

	// initialize genesis checkpoint
	// TODO load state from disk
	instance.chkpts[0] = "TODO GENESIS STATE FROM STACK"

	return instance
}

// =============================================================================
// Helper functions for PBFT go here
// =============================================================================

// Given a certain view n, what is the expected primary?
func (instance *Plugin) getPrimary(n uint64) uint64 {
	return n % uint64(instance.replicaCount)
}

// Is the sequence number between watermarks?
func (instance *Plugin) inW(n uint64) bool {
	return n-instance.h > 0 && n-instance.h <= instance.L
}

// Is the view right? And is the sequence number between watermarks?
func (instance *Plugin) inWV(v uint64, n uint64) bool {
	return instance.view == v && instance.inW(n)
}

// Given a digest/view/seq, is there an entry in the certLog?
// If so, return it. If not, create it.
func (instance *Plugin) getCert(v uint64, n uint64) (cert *msgCert) {
	idx := msgID{v, n}
	cert, ok := instance.certStore[idx]
	if ok {
		return
	}

	cert = &msgCert{}
	instance.certStore[idx] = cert
	return
}

// =============================================================================
// Preprepare/prepare/commit quorum checks
// =============================================================================

func (instance *Plugin) prePrepared(digest string, v uint64, n uint64) bool {
	_, mInLog := instance.reqStore[digest]

	if digest != "" && !mInLog {
		return false
	}

	if q, ok := instance.qset[qidx{digest, n}]; ok && q.View == v {
		return true
	}

	cert := instance.certStore[msgID{v, n}]
	if cert != nil {
		p := cert.prePrepare
		if p != nil && p.View == v && p.SequenceNumber == n && p.RequestDigest == digest {
			return true
		}
	}
	logger.Debug("Replica %d does not have view=%d/seqNo=%d pre-prepared",
		instance.id, v, n)
	return false
}

func (instance *Plugin) prepared(digest string, v uint64, n uint64) bool {
	if !instance.prePrepared(digest, v, n) {
		return false
	}

	if p, ok := instance.pset[n]; ok && p.View == v && p.Digest == digest {
		return true
	}

	quorum := uint(0)
	cert := instance.certStore[msgID{v, n}]
	if cert == nil {
		return false
	}

	for _, p := range cert.prepare {
		if p.View == v && p.SequenceNumber == n && p.RequestDigest == digest {
			quorum++
		}
	}

	logger.Debug("Replica %d prepare count for view=%d/seqNo=%d: %d",
		instance.id, v, n, quorum)

	return quorum >= 2*instance.f
}

func (instance *Plugin) committed(digest string, v uint64, n uint64) bool {
	if !instance.prepared(digest, v, n) {
		return false
	}

	quorum := uint(0)
	cert := instance.certStore[msgID{v, n}]
	if cert == nil {
		return false
	}

	for _, p := range cert.commit {
		if p.View == v && p.SequenceNumber == n {
			quorum++
		}
	}

	logger.Debug("Replica %d commit count for view=%d/seqNo=%d: %d",
		instance.id, v, n, quorum)

	return quorum >= 2*instance.f+1
}

// =============================================================================
// Receive methods go here
// =============================================================================

// RecvMsg receives messages transmitted by CPI.Broadcast or CPI.Unicast.
func (instance *Plugin) RecvMsg(msgWrapped *pb.OpenchainMessage) error {
	if msgWrapped.Type == pb.OpenchainMessage_CHAIN_TRANSACTION {
		return instance.Request(msgWrapped.Payload)
	}
	if msgWrapped.Type != pb.OpenchainMessage_CONSENSUS {
		return fmt.Errorf("Unexpected message type: %s", msgWrapped.Type)
	}

	msg := &Message{}
	err := proto.Unmarshal(msgWrapped.Payload, msg)
	if err != nil {
		return fmt.Errorf("Error unpacking payload from message: %s", err)
	}

	if req := msg.GetRequest(); req != nil {
		err = instance.recvRequest(req)
	} else if preprep := msg.GetPrePrepare(); preprep != nil {
		err = instance.recvPrePrepare(preprep)
	} else if prep := msg.GetPrepare(); prep != nil {
		err = instance.recvPrepare(prep)
	} else if commit := msg.GetCommit(); commit != nil {
		err = instance.recvCommit(commit)
	} else if chkpt := msg.GetCheckpoint(); chkpt != nil {
		err = instance.recvCheckpoint(chkpt)
	} else if vc := msg.GetViewChange(); vc != nil {
		err = instance.recvViewChange(vc)
	} else if nv := msg.GetNewView(); nv != nil {
		err = instance.recvNewView(nv)
	} else {
		err := fmt.Errorf("Invalid message: %v", msgWrapped.Payload)
		logger.Error("%s", err)
	}

	return err
}

// Request is the main entry into the consensus plugin.
// txs will be passed to CPI.ExecTXs once consensus is reached.
func (instance *Plugin) Request(txs []byte) error {
	logger.Info("New consensus request received")
	req := &Request{Payload: txs}                                    // TODO assign "client" timestamp
	return instance.broadcast(&Message{&Message_Request{req}}, true) // route to ourselves as well
}

func (instance *Plugin) recvRequest(req *Request) error {
	digest := hashReq(req)
	logger.Debug("Replica %d received request: %s", instance.id, digest)

	// TODO test timestamp
	instance.reqStore[digest] = req

	n := instance.seqNo + 1

	if instance.getPrimary(instance.view) == instance.id && instance.activeView { // if we're primary of current view
		haveOther := false
		for _, cert := range instance.certStore { // check for other PRE-PREPARE for same digest, but different seqNo
			if p := cert.prePrepare; p != nil {
				if p.View == instance.view && p.SequenceNumber != n && p.RequestDigest == digest {
					logger.Debug("Other pre-prepared found with same digest but different seqNo: %d instead of %d", p.SequenceNumber, n)
					haveOther = true
					break
				}
			}
		}

		if instance.inWV(instance.view, n) && !haveOther {
			logger.Debug("Primary %d broadcasting pre-prepare for view=%d/seqNo=%d and digest %s",
				instance.id, instance.view, n, digest)
			instance.seqNo = n
			preprep := &PrePrepare{
				View:           instance.view,
				SequenceNumber: instance.seqNo,
				RequestDigest:  digest,
				Request:        req,
				ReplicaId:      instance.id,
			}
			cert := instance.getCert(instance.view, n)
			cert.prePrepare = preprep

			return instance.broadcast(&Message{&Message_PrePrepare{preprep}}, false)
		}
	}

	return nil
}

func (instance *Plugin) recvPrePrepare(preprep *PrePrepare) error {
	logger.Debug("Replica %d received pre-prepare from replica %d for view=%d/seqNo=%d",
		instance.id, preprep.ReplicaId, preprep.View,
		preprep.SequenceNumber)

	if !instance.activeView {
		return nil
	}

	if instance.getPrimary(instance.view) != preprep.ReplicaId {
		logger.Warning("Pre-prepare from other than primary: got %d, should be %d", preprep.ReplicaId, instance.getPrimary(instance.view))
		return nil
	}

	if !instance.inWV(preprep.View, preprep.SequenceNumber) {
		logger.Warning("Pre-prepare sequence number outside watermarks: seqNo %d, low-mark %d", preprep.SequenceNumber, instance.h)
		return nil
	}

	cert := instance.getCert(preprep.View, preprep.SequenceNumber)
	if cert.prePrepare != nil && cert.prePrepare.RequestDigest != preprep.RequestDigest {
		logger.Warning("Pre-prepare found for same view/seqNo but different digest: received %s, stored %s", preprep.RequestDigest, cert.prePrepare.RequestDigest)
	} else {
		cert.prePrepare = preprep
	}

	// Store the request if, for whatever reason, haven't received it from an earlier broadcast.<commit_msg>Add wrong view message for inWV check<commit_after>/*
Licensed to the Apache Software Foundation (ASF) under one
or more contributor license agreements.  See the NOTICE file
distributed with this work for additional information
regarding copyright ownership.  The ASF licenses this file
to you under the Apache License, Version 2.0 (the
"License"); you may not use this file except in compliance
with the License.  You may obtain a copy of the License at

  http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing,
software distributed under the License is distributed on an
"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
KIND, either express or implied.  See the License for the
specific language governing permissions and limitations
under the License.
*/

package pbft

import (
	"encoding/base64"
	"fmt"
	"strconv"
	"strings"

	"github.com/golang/protobuf/proto"
	"github.com/openblockchain/obc-peer/openchain/consensus"
	"github.com/openblockchain/obc-peer/openchain/util"
	pb "github.com/openblockchain/obc-peer/protos"

	"github.com/op/go-logging"
	"github.com/spf13/viper"
)

// =============================================================================
// Constants
// =============================================================================

const configPrefix = "OPENCHAIN_PBFT"

// =============================================================================
// Init
// =============================================================================

var logger *logging.Logger // package-level logger
var pluginInstance *Plugin // the Plugin is a singleton

func init() {
	logger = logging.MustGetLogger("consensus/pbft")
}

// =============================================================================
// Custom structure definitions go here
// =============================================================================

// Plugin carries fields related to the consensus algorithm implementation.
type Plugin struct {
	// internal data
	config *viper.Viper  // link to the plugin config file
	cpi    consensus.CPI // link to the CPI

	// PBFT data
	activeView   bool              // view change happening
	f            uint              // number of faults we can tolerate
	h            uint64            // low watermark
	id           uint64            // replica ID; PBFT `i`
	K            uint64            // checkpoint period
	L            uint64            // log size
	lastExec     uint64            // last request we executed
	replicaCount uint              // number of replicas; PBFT `|R|`
	seqNo        uint64            // PBFT "n", strictly monotonic increasing sequence number
	view         uint64            // current view
	chkpts       map[uint64]string // state checkpoints; map lastExec to global hash
	pset         map[uint64]*ViewChange_PQ
	qset         map[qidx]*ViewChange_PQ

	// Implementation of PBFT `in`
	certStore       map[msgID]*msgCert    // track quorum certificates for requests
	reqStore        map[string]*Request   // track requests
	checkpointStore map[Checkpoint]bool   // track checkpoints as set
	viewChangeStore map[vcidx]*ViewChange // track view-change messages
	lastNewView     NewView               // track last new-view we received or sent
}

type qidx struct {
	d string
	n uint64
}

type msgID struct { // our index through certStore
	v uint64
	n uint64
}

type msgCert struct {
	prePrepare  *PrePrepare
	sentPrepare bool
	prepare     []*Prepare
	sentCommit  bool
	commit      []*Commit
}

type vcidx struct {
	v  uint64
	id uint64
}

// =============================================================================
// Constructors go here
// =============================================================================

// GetPlugin returns the handle to the Plugin singleton and updates the CPI if necessary.
func GetPlugin(c consensus.CPI) *Plugin {
	if pluginInstance == nil {
		pluginInstance = New(c)
	} else {
		pluginInstance.cpi = c // otherwise, just update the CPI
	}
	return pluginInstance
}

// New creates an plugin-specific structure that acts as the ConsenusHandler's consenter.
func New(c consensus.CPI) *Plugin {
	instance := &Plugin{}
	instance.cpi = c

	// setup the link to the config file
	instance.config = viper.New()

	// for environment variables
	instance.config.SetEnvPrefix(configPrefix)
	instance.config.AutomaticEnv()
	replacer := strings.NewReplacer(".", "_")
	instance.config.SetEnvKeyReplacer(replacer)

	instance.config.SetConfigName("config")
	instance.config.AddConfigPath("./")
	instance.config.AddConfigPath("./openchain/consensus/pbft/")
	err := instance.config.ReadInConfig()
	if err != nil {
		panic(fmt.Errorf("Fatal error reading consensus algo config: %s", err))
	}

	// TODO Initialize the algorithm here
	// You may want to set the fields of `instance` using `instance.GetParam()`.

	// In dev/debugging mode you are expected to override the config values
	// with the environment variable OPENCHAIN_PBFT_X_Y

	// replica ID
	paramID, err := instance.getParam("replica.id")
	if err != nil {
		panic(fmt.Errorf("No ID assigned to the replica: %s", err))
	}
	instance.id, err = strconv.ParseUint(paramID, 10, 64)
	if err != nil {
		panic(fmt.Errorf("Cannot convert config ID to uint64: %s", err))
	}
	// byzantine nodes
	paramF, err := instance.getParam("general.f")
	if err != nil {
		panic(fmt.Errorf("No f defined in the config file: %s", err))
	}
	f, err := strconv.ParseUint(paramF, 10, 0)
	if err != nil {
		panic(fmt.Errorf("Cannot convert config f to uint64: %s", err))
	}
	instance.f = uint(f)
	// replica count
	instance.replicaCount = 3*instance.f + 1
	// checkpoint period
	paramK, err := instance.getParam("general.K")
	if err != nil {
		panic(fmt.Errorf("Checkpoint period is not defined: %s", err))
	}
	instance.K, err = strconv.ParseUint(paramK, 10, 64)
	if err != nil {
		panic(fmt.Errorf("Cannot convert config checkpoint period to uint64: %s", err))
	}
	// log size
	instance.L = 2 * instance.K

	instance.activeView = true

	// init the logs
	instance.certStore = make(map[msgID]*msgCert)
	instance.reqStore = make(map[string]*Request)
	instance.checkpointStore = make(map[Checkpoint]bool)
	instance.viewChangeStore = make(map[vcidx]*ViewChange)
	instance.chkpts = make(map[uint64]string)
	instance.pset = make(map[uint64]*ViewChange_PQ)
	instance.qset = make(map[qidx]*ViewChange_PQ)

	// initialize genesis checkpoint
	// TODO load state from disk
	instance.chkpts[0] = "TODO GENESIS STATE FROM STACK"

	return instance
}

// =============================================================================
// Helper functions for PBFT go here
// =============================================================================

// Given a certain view n, what is the expected primary?
func (instance *Plugin) getPrimary(n uint64) uint64 {
	return n % uint64(instance.replicaCount)
}

// Is the sequence number between watermarks?
func (instance *Plugin) inW(n uint64) bool {
	return n-instance.h > 0 && n-instance.h <= instance.L
}

// Is the view right? And is the sequence number between watermarks?
func (instance *Plugin) inWV(v uint64, n uint64) bool {
	return instance.view == v && instance.inW(n)
}

// Given a digest/view/seq, is there an entry in the certLog?
// If so, return it. If not, create it.
func (instance *Plugin) getCert(v uint64, n uint64) (cert *msgCert) {
	idx := msgID{v, n}
	cert, ok := instance.certStore[idx]
	if ok {
		return
	}

	cert = &msgCert{}
	instance.certStore[idx] = cert
	return
}

// =============================================================================
// Preprepare/prepare/commit quorum checks
// =============================================================================

func (instance *Plugin) prePrepared(digest string, v uint64, n uint64) bool {
	_, mInLog := instance.reqStore[digest]

	if digest != "" && !mInLog {
		return false
	}

	if q, ok := instance.qset[qidx{digest, n}]; ok && q.View == v {
		return true
	}

	cert := instance.certStore[msgID{v, n}]
	if cert != nil {
		p := cert.prePrepare
		if p != nil && p.View == v && p.SequenceNumber == n && p.RequestDigest == digest {
			return true
		}
	}
	logger.Debug("Replica %d does not have view=%d/seqNo=%d pre-prepared",
		instance.id, v, n)
	return false
}

func (instance *Plugin) prepared(digest string, v uint64, n uint64) bool {
	if !instance.prePrepared(digest, v, n) {
		return false
	}

	if p, ok := instance.pset[n]; ok && p.View == v && p.Digest == digest {
		return true
	}

	quorum := uint(0)
	cert := instance.certStore[msgID{v, n}]
	if cert == nil {
		return false
	}

	for _, p := range cert.prepare {
		if p.View == v && p.SequenceNumber == n && p.RequestDigest == digest {
			quorum++
		}
	}

	logger.Debug("Replica %d prepare count for view=%d/seqNo=%d: %d",
		instance.id, v, n, quorum)

	return quorum >= 2*instance.f
}

func (instance *Plugin) committed(digest string, v uint64, n uint64) bool {
	if !instance.prepared(digest, v, n) {
		return false
	}

	quorum := uint(0)
	cert := instance.certStore[msgID{v, n}]
	if cert == nil {
		return false
	}

	for _, p := range cert.commit {
		if p.View == v && p.SequenceNumber == n {
			quorum++
		}
	}

	logger.Debug("Replica %d commit count for view=%d/seqNo=%d: %d",
		instance.id, v, n, quorum)

	return quorum >= 2*instance.f+1
}

// =============================================================================
// Receive methods go here
// =============================================================================

// RecvMsg receives messages transmitted by CPI.Broadcast or CPI.Unicast.
func (instance *Plugin) RecvMsg(msgWrapped *pb.OpenchainMessage) error {
	if msgWrapped.Type == pb.OpenchainMessage_CHAIN_TRANSACTION {
		return instance.Request(msgWrapped.Payload)
	}
	if msgWrapped.Type != pb.OpenchainMessage_CONSENSUS {
		return fmt.Errorf("Unexpected message type: %s", msgWrapped.Type)
	}

	msg := &Message{}
	err := proto.Unmarshal(msgWrapped.Payload, msg)
	if err != nil {
		return fmt.Errorf("Error unpacking payload from message: %s", err)
	}

	if req := msg.GetRequest(); req != nil {
		err = instance.recvRequest(req)
	} else if preprep := msg.GetPrePrepare(); preprep != nil {
		err = instance.recvPrePrepare(preprep)
	} else if prep := msg.GetPrepare(); prep != nil {
		err = instance.recvPrepare(prep)
	} else if commit := msg.GetCommit(); commit != nil {
		err = instance.recvCommit(commit)
	} else if chkpt := msg.GetCheckpoint(); chkpt != nil {
		err = instance.recvCheckpoint(chkpt)
	} else if vc := msg.GetViewChange(); vc != nil {
		err = instance.recvViewChange(vc)
	} else if nv := msg.GetNewView(); nv != nil {
		err = instance.recvNewView(nv)
	} else {
		err := fmt.Errorf("Invalid message: %v", msgWrapped.Payload)
		logger.Error("%s", err)
	}

	return err
}

// Request is the main entry into the consensus plugin.
// txs will be passed to CPI.ExecTXs once consensus is reached.
func (instance *Plugin) Request(txs []byte) error {
	logger.Info("New consensus request received")
	req := &Request{Payload: txs}                                    // TODO assign "client" timestamp
	return instance.broadcast(&Message{&Message_Request{req}}, true) // route to ourselves as well
}

func (instance *Plugin) recvRequest(req *Request) error {
	digest := hashReq(req)
	logger.Debug("Replica %d received request: %s", instance.id, digest)

	// TODO test timestamp
	instance.reqStore[digest] = req

	n := instance.seqNo + 1

	if instance.getPrimary(instance.view) == instance.id && instance.activeView { // if we're primary of current view
		haveOther := false
		for _, cert := range instance.certStore { // check for other PRE-PREPARE for same digest, but different seqNo
			if p := cert.prePrepare; p != nil {
				if p.View == instance.view && p.SequenceNumber != n && p.RequestDigest == digest {
					logger.Debug("Other pre-prepared found with same digest but different seqNo: %d instead of %d", p.SequenceNumber, n)
					haveOther = true
					break
				}
			}
		}

		if instance.inWV(instance.view, n) && !haveOther {
			logger.Debug("Primary %d broadcasting pre-prepare for view=%d/seqNo=%d and digest %s",
				instance.id, instance.view, n, digest)
			instance.seqNo = n
			preprep := &PrePrepare{
				View:           instance.view,
				SequenceNumber: instance.seqNo,
				RequestDigest:  digest,
				Request:        req,
				ReplicaId:      instance.id,
			}
			cert := instance.getCert(instance.view, n)
			cert.prePrepare = preprep

			return instance.broadcast(&Message{&Message_PrePrepare{preprep}}, false)
		}
	}

	return nil
}

func (instance *Plugin) recvPrePrepare(preprep *PrePrepare) error {
	logger.Debug("Replica %d received pre-prepare from replica %d for view=%d/seqNo=%d",
		instance.id, preprep.ReplicaId, preprep.View,
		preprep.SequenceNumber)

	if !instance.activeView {
		return nil
	}

	if instance.getPrimary(instance.view) != preprep.ReplicaId {
		logger.Warning("Pre-prepare from other than primary: got %d, should be %d", preprep.ReplicaId, instance.getPrimary(instance.view))
		return nil
	}

	if !instance.inWV(preprep.View, preprep.SequenceNumber) {
		logger.Warning("Pre-prepare view different, or sequence number outside watermarks: preprep.View %d, expected.View %d, seqNo %d, low-mark %d", preprep.View, instance.getPrimary(instance.view), preprep.SequenceNumber, instance.h)
		return nil
	}

	cert := instance.getCert(preprep.View, preprep.SequenceNumber)
	if cert.prePrepare != nil && cert.prePrepare.RequestDigest != preprep.RequestDigest {
		logger.Warning("Pre-prepare found for same view/seqNo but different digest: received %s, stored %s", preprep.RequestDigest, cert.prePrepare.RequestDigest)
	} else {
		cert.prePrepare = preprep
	}

	// Store the request if, for whatever reason, haven't received it from an earlier broadcast.
{'fraction_non_alphanumeric': 0.18737701333722032, 'fraction_numerical': 0.0069965745936885065, 'mean_word_length': 3.8374757623832187, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 9, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 4, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4531963', 'n_tokens_mistral': 8803, 'n_tokens_neox': 8147, 'n_words': 2500}