Capture broadcasted messages for re-broadcasting even at panic (#627)

Fix a bug where if `Host.RequestBroadcast` panics the requested message
is never added to the re-broadcast state.

Fix an inconsistent behaviour where error at requested broadcasts via
re-broadcast are returned instead of being silently logged.

Enhance emulator to allow pluggable signign logic with two additional
signing implementation: erroneous and panic.

Implement tests that assert expected behaviour at signing error or
panic.

Fixes #236

Author: masih

emulator/driver.go

@@ -3,6 +3,7 @@ package emulator
 import (
 	"bytes"
 	"context"
+	"errors"
 	"testing"
 
 	"github.com/filecoin-project/go-f3/gpbft"
@@ -22,9 +23,9 @@ type Driver struct {
 }
 
 // NewDriver instantiates a new Driver with the given GPBFT options. See
-// Driver.StartInstance.
+// Driver.RequireStartInstance.
 func NewDriver(t *testing.T, o ...gpbft.Option) *Driver {
-	h := newHost(t)
+	h := newHost(t, AdhocSigning())
 	participant, err := gpbft.NewParticipant(h, o...)
 	require.NoError(t, err)
 	return &Driver{
@@ -34,23 +35,32 @@ func NewDriver(t *testing.T, o ...gpbft.Option) *Driver {
 	}
 }
 
+func (d *Driver) SetSigning(signing Signing) { d.host.Signing = signing }
+
 // StartInstance sets the current instances and starts emulation for it by signalling the
 // start of instance to the emulated honest gpbft.Participant.
 //
 // See NewInstance.
-func (d *Driver) StartInstance(id uint64) {
-	d.require.NoError(d.subject.StartInstanceAt(id, d.host.Time()))
+func (d *Driver) StartInstance(id uint64) error {
+	if err := d.subject.StartInstanceAt(id, d.host.Time()); err != nil {
+		return err
+	}
 	// Trigger alarm once based on the implicit assumption that go-f3 uses alarm to
 	// kickstart an instance internally.
-	d.RequireDeliverAlarm()
+	if triggered, err := d.DeliverAlarm(); err != nil {
+		return err
+	} else if !triggered {
+		return errors.New("alarm did not trigger at start")
+	}
+	return nil
 }
 
 // AddInstance adds an instance to the list of instances known by the driver.
 func (d *Driver) AddInstance(instance *Instance) {
 	d.require.NoError(d.host.addInstance(instance))
 }
 
-func (d *Driver) deliverAlarm() (bool, error) {
+func (d *Driver) DeliverAlarm() (bool, error) {
 	if d.host.maybeReceiveAlarm() {
 		return true, d.subject.ReceiveAlarm()
 	}
@@ -72,8 +82,8 @@ func (d *Driver) prepareMessage(partialMessage *gpbft.GMessage) *gpbft.GMessage
 
 	mb := instance.NewMessageBuilder(partialMessage.Vote, partialMessage.Justification, withValidTicket)
 	mb.NetworkName = d.host.NetworkName()
-	mb.SigningMarshaler = d.host.adhocSigning
-	msg, err := mb.Build(context.Background(), d.host.adhocSigning, partialMessage.Sender)
+	mb.SigningMarshaler = d.host
+	msg, err := mb.Build(context.Background(), d.host, partialMessage.Sender)
 	d.require.NoError(err)
 	d.require.NotNil(msg)
 	if !withValidTicket {

emulator/driver_assertions.go

@@ -19,8 +19,14 @@ func (d *Driver) RequireErrOnDeliverMessage(message *gpbft.GMessage, err error,
 	}
 }
 
+// RequireStartInstance asserts that instance with the given ID is started. See
+// StartInstance.
+func (d *Driver) RequireStartInstance(id uint64) {
+	d.require.NoError(d.StartInstance(id))
+}
+
 func (d *Driver) RequireDeliverAlarm() {
-	delivered, err := d.deliverAlarm()
+	delivered, err := d.DeliverAlarm()
 	d.require.NoError(err)
 	d.require.True(delivered)
 }

emulator/host.go

@@ -15,7 +15,7 @@ const networkName = "emulator-net"
 var _ gpbft.Host = (*driverHost)(nil)
 
 type driverHost struct {
-	adhocSigning
+	Signing
 
 	t                  *testing.T
 	id                 gpbft.ActorID
@@ -25,10 +25,11 @@ type driverHost struct {
 	chain              map[uint64]*Instance
 }
 
-func newHost(t *testing.T) *driverHost {
+func newHost(t *testing.T, signing Signing) *driverHost {
 	return &driverHost{
-		t:     t,
-		chain: make(map[uint64]*Instance),
+		Signing: signing,
+		t:       t,
+		chain:   make(map[uint64]*Instance),
 	}
 }
 

emulator/instance.go

@@ -21,14 +21,15 @@ type Instance struct {
 	powerTable       *gpbft.PowerTable
 	beacon           []byte
 	decision         *gpbft.Justification
+	signing          Signing
 }
 
 // NewInstance instantiates a new Instance for emulation. If absent, the
 // constructor will implicitly generate any missing but required values such as
 // public keys Power Table CID, etc. for the given params. The given proposal
 // must contain at least one tipset.
 //
-// See Driver.StartInstance.
+// See Driver.RequireStartInstance.
 func NewInstance(t *testing.T, id uint64, powerEntries gpbft.PowerEntries, proposal ...gpbft.TipSet) *Instance {
 	// UX of the gpbft API is pretty painful; encapsulate the pain of getting an
 	// instance going here at the price of accepting partial data and implicitly
@@ -67,9 +68,11 @@ func NewInstance(t *testing.T, id uint64, powerEntries gpbft.PowerEntries, propo
 			Commitments: [32]byte{},
 			PowerTable:  ptCid,
 		},
+		signing: AdhocSigning(),
 	}
 }
 
+func (i *Instance) SetSigning(signing Signing)               { i.signing = signing }
 func (i *Instance) Proposal() gpbft.ECChain                  { return i.proposal }
 func (i *Instance) GetDecision() *gpbft.Justification        { return i.decision }
 func (i *Instance) ID() uint64                               { return i.id }
@@ -134,7 +137,7 @@ func (i *Instance) NewJustification(round uint64, step gpbft.Phase, vote gpbft.E
 }
 
 func (i *Instance) NewJustificationWithPayload(payload gpbft.Payload, from ...gpbft.ActorID) *gpbft.Justification {
-	msg := signing.MarshalPayloadForSigning(networkName, &payload)
+	msg := i.signing.MarshalPayloadForSigning(networkName, &payload)
 	qr := gpbft.QuorumResult{
 		Signers:    make([]int, len(from)),
 		PubKeys:    make([]gpbft.PubKey, len(from)),
@@ -144,13 +147,13 @@ func (i *Instance) NewJustificationWithPayload(payload gpbft.Payload, from ...gp
 		index, found := i.powerTable.Lookup[actor]
 		require.True(i.t, found)
 		entry := i.powerTable.Entries[index]
-		signature, err := signing.Sign(context.Background(), entry.PubKey, msg)
+		signature, err := i.signing.Sign(context.Background(), entry.PubKey, msg)
 		require.NoError(i.t, err)
 		qr.Signatures[j] = signature
 		qr.PubKeys[j] = entry.PubKey
 		qr.Signers[j] = index
 	}
-	aggregate, err := signing.Aggregate(qr.PubKeys, qr.Signatures)
+	aggregate, err := i.signing.Aggregate(qr.PubKeys, qr.Signatures)
 	require.NoError(i.t, err)
 	return &gpbft.Justification{
 		Vote:      payload,

emulator/signing.go

@@ -10,18 +10,30 @@ import (
 )
 
 var (
-	_ gpbft.Verifier         = (*adhocSigning)(nil)
-	_ gpbft.Signer           = (*adhocSigning)(nil)
-	_ gpbft.SigningMarshaler = (*adhocSigning)(nil)
-
-	signing adhocSigning
+	_ Signing = (*adhocSigning)(nil)
+	_ Signing = (*erroneousSigning)(nil)
+	_ Signing = (*panicSigning)(nil)
 )
 
-// adhocSigning marshals, signs and verifies messages on behalf of any given
+type Signing interface {
+	gpbft.Verifier
+	gpbft.Signer
+	gpbft.SigningMarshaler
+}
+
+// AdhocSigning marshals, signs and verifies messages on behalf of any given
 // public key but uniquely and deterministically so using crc32 hash function for
 // performance. This implementation is not secure nor collision resistant. A
 // typical Instance power table is small enough to make the risk of collisions
 // negligible.
+func AdhocSigning() Signing { return adhocSigning{} }
+
+// ErroneousSigning returns an error for every Signing API that can return an error.
+func ErroneousSigning() Signing { return erroneousSigning{} }
+
+// PanicSigning panics for every Signing API.
+func PanicSigning() Signing { return panicSigning{} }
+
 type adhocSigning struct{}
 
 func (s adhocSigning) Sign(_ context.Context, sender gpbft.PubKey, msg []byte) ([]byte, error) {
@@ -84,3 +96,32 @@ func (s adhocSigning) VerifyAggregate(payload, got []byte, signers []gpbft.PubKe
 func (s adhocSigning) MarshalPayloadForSigning(name gpbft.NetworkName, payload *gpbft.Payload) []byte {
 	return payload.MarshalForSigning(name)
 }
+
+type erroneousSigning struct{}
+
+func (p erroneousSigning) Verify(gpbft.PubKey, []byte, []byte) error {
+	return errors.New("err Verify")
+}
+
+func (p erroneousSigning) VerifyAggregate([]byte, []byte, []gpbft.PubKey) error {
+	return errors.New("err VerifyAggregate")
+}
+
+func (p erroneousSigning) Aggregate([]gpbft.PubKey, [][]byte) ([]byte, error) {
+	return nil, errors.New("err Aggregate")
+}
+func (p erroneousSigning) Sign(context.Context, gpbft.PubKey, []byte) ([]byte, error) {
+	return nil, errors.New("err Sign")
+}
+
+func (p erroneousSigning) MarshalPayloadForSigning(gpbft.NetworkName, *gpbft.Payload) []byte {
+	return nil
+}
+
+type panicSigning struct{}
+
+func (p panicSigning) Verify(gpbft.PubKey, []byte, []byte) error                         { panic("π") }
+func (p panicSigning) VerifyAggregate([]byte, []byte, []gpbft.PubKey) error              { panic("π") }
+func (p panicSigning) Aggregate([]gpbft.PubKey, [][]byte) ([]byte, error)                { panic("π") }
+func (p panicSigning) Sign(context.Context, gpbft.PubKey, []byte) ([]byte, error)        { panic("π") }
+func (p panicSigning) MarshalPayloadForSigning(gpbft.NetworkName, *gpbft.Payload) []byte { panic("π") }

gpbft/gpbft.go

@@ -872,11 +872,14 @@ func (i *instance) broadcast(round uint64, step Phase, value ECChain, createTick
 	if createTicket {
 		mb.BeaconForTicket = i.beacon
 	}
+
+	// Capture the broadcast and metrics first. Because, otherwise the instance will
+	// end up with partial re-broadcast messages if RequestBroadcast panics.
+	i.broadcasted.record(mb)
+	metrics.broadcastCounter.Add(context.TODO(), 1, metric.WithAttributes(attrPhase[p.Step]))
 	if err := i.participant.host.RequestBroadcast(mb); err != nil {
 		i.log("failed to request broadcast: %v", err)
 	}
-	i.broadcasted.record(mb)
-	metrics.broadcastCounter.Add(context.TODO(), 1, metric.WithAttributes(attrPhase[p.Step]))
 }
 
 // tryRebroadcast checks whether re-broadcast timeout has elapsed, and if so
@@ -957,10 +960,13 @@ func (i *instance) rebroadcast() error {
 		mbs := i.broadcasted.messagesByRound[round]
 		for _, mb := range mbs {
 			if err := i.participant.host.RequestBroadcast(mb); err != nil {
-				return err
+				// Silently log the error and proceed. This is consistent with the behaviour of
+				// instance for regular broadcasts.
+				i.log("failed to request rebroadcast %s at round %d: %v", mb.Payload.Step, mb.Payload.Round, err)
+			} else {
+				i.log("rebroadcasting %s at round %d for value %s", mb.Payload.Step.String(), mb.Payload.Round, mb.Payload.Value)
+				metrics.reBroadcastCounter.Add(context.TODO(), 1)
 			}
-			i.log("rebroadcasting %s at round %d for value %s", mb.Payload.Step.String(), mb.Payload.Round, mb.Payload.Value)
-			metrics.reBroadcastCounter.Add(context.TODO(), 1)
 		}
 	}
 	return nil