Add pomHandler to ModuleParams (application defines what to do with them)

Author: ranchalp

pkg/accountability/simpleacc/accountability.go

@@ -78,7 +78,10 @@ type ModuleParams = common.ModuleParams
 //
 // ATTENTION: This module is intended to be used once per instance
 // (to avoid replay attacks) and reinstantiated in a factory.
-func NewModule(mc ModuleConfig, params *ModuleParams, logger logging.Logger) (modules.PassiveModule, error) {
+func NewModule(
+	mc ModuleConfig,
+	params *ModuleParams,
+	logger logging.Logger) (modules.PassiveModule, error) {
 	m := dsl.NewModule(mc.Self)
 
 	state := &incommon.State{
@@ -87,8 +90,8 @@ func NewModule(mc ModuleConfig, params *ModuleParams, logger logging.Logger) (mo
 		LocalPredecision:   nil,
 		DecidedCertificate: nil,
 		Predecided:         false,
-		UnsentPoMs:         make([]*accpbtypes.PoM, 0),
-		SentPoMs:           make(map[t.NodeID]*accpbtypes.PoM),
+		UnhandledPoMs:      make([]*accpbtypes.PoM, 0),
+		HandledPoMs:        make(map[t.NodeID]*accpbtypes.PoM),
 	}
 
 	predecisions.IncludePredecisions(m, &mc, params, state, logger)

pkg/accountability/simpleacc/common/common.go

@@ -1,6 +1,10 @@
 package common
 
 import (
+	incommon "github.com/filecoin-project/mir/pkg/accountability/simpleacc/internal/common"
+	"github.com/filecoin-project/mir/pkg/dsl"
+	"github.com/filecoin-project/mir/pkg/logging"
+	accpbtypes "github.com/filecoin-project/mir/pkg/pb/accountabilitypb/types"
 	trantorpbtypes "github.com/filecoin-project/mir/pkg/pb/trantorpb/types"
 	t "github.com/filecoin-project/mir/pkg/types"
 )
@@ -21,4 +25,10 @@ type ModuleConfig struct {
 type ModuleParams struct {
 	Membership        *trantorpbtypes.Membership // the list of participating nodes
 	LightCertificates bool
+	PomsHandler       func(m dsl.Module, // function to be called when PoMs detected
+		mc *ModuleConfig,
+		params *ModuleParams,
+		state *incommon.State,
+		poms []*accpbtypes.PoM,
+		logger logging.Logger)
 }

pkg/accountability/simpleacc/internal/certificates/fullcertificates/fullcertificates.go

@@ -1,8 +1,6 @@
 package fullcertificates
 
 import (
-	"reflect"
-
 	"github.com/filecoin-project/mir/pkg/accountability/simpleacc/common"
 	incommon "github.com/filecoin-project/mir/pkg/accountability/simpleacc/internal/common"
 	"github.com/filecoin-project/mir/pkg/accountability/simpleacc/internal/poms"
@@ -28,9 +26,8 @@ func IncludeFullCertificate(m dsl.Module,
 	logger logging.Logger,
 ) {
 
-	accpbdsl.UponFullCertificateReceived(m, func(from t.NodeID, certificate map[t.NodeID]*accpbtypes.SignedPredecision) error {
-		predecision, empty := maputil.AnyVal(certificate)
-		if empty {
+	accpbdsl.UponFullCertificateReceived(m, func(from t.NodeID, decision []byte, certificate map[t.NodeID][]byte) error {
+		if len(certificate) == 0 {
 			logger.Log(logging.LevelDebug, "Ignoring empty predecision certificate")
 			return nil
 		}
@@ -40,42 +37,42 @@ func IncludeFullCertificate(m dsl.Module,
 			return nil
 		}
 
-		for _, v := range certificate {
-			if !reflect.DeepEqual(predecision.Predecision, v.Predecision) {
-				logger.Log(logging.LevelDebug, "Ignoring predecision certificate with different predecisions")
-				return nil
-			}
-		}
-
 		// Verify all signatures in certificate.
 		cryptopbdsl.VerifySigs(
 			m,
 			mc.Crypto,
-			sliceutil.Transform(maputil.GetValues(certificate),
-				func(i int, sp *accpbtypes.SignedPredecision) *cryptopbtypes.SignedData {
-					return &cryptopbtypes.SignedData{Data: [][]byte{sp.Predecision, []byte(mc.Self)}}
-				}),
-			sliceutil.Transform(maputil.GetValues(certificate),
-				func(i int, sp *accpbtypes.SignedPredecision) []byte {
-					return sp.Signature
+			sliceutil.Generate(
+				len(certificate),
+				func(i int) *cryptopbtypes.SignedData {
+					return &cryptopbtypes.SignedData{
+						Data: [][]byte{decision},
+					}
 				}),
+			maputil.GetValues(certificate),
 			maputil.GetKeys(certificate),
 			&verifySigs{
-				certificate: certificate,
+				certificate: &accpbtypes.FullCertificate{
+					Decision:   decision,
+					Signatures: certificate,
+				},
 			},
 		)
 		return nil
 	})
 
 	cryptopbdsl.UponSigsVerified(m, func(nodeIds []t.NodeID, errs []error, allOk bool, vsr *verifySigs) error {
 		for i, nodeID := range nodeIds {
-			predecisions.ApplySigVerified(m, mc, params, state, nodeID, errs[i], vsr.certificate[nodeID], false, logger)
+			sp := &accpbtypes.SignedPredecision{
+				Predecision: vsr.certificate.Decision,
+				Signature:   vsr.certificate.Signatures[nodeID],
+			}
+			predecisions.ApplySigVerified(m, mc, params, state, nodeID, errs[i], sp, false, logger)
 		}
-		poms.SendPoMs(m, mc, params, state, logger)
+		poms.HandlePoMs(m, mc, params, state, logger)
 		return nil
 	})
 }
 
 type verifySigs struct {
-	certificate map[t.NodeID]*accpbtypes.SignedPredecision
+	certificate *accpbtypes.FullCertificate
 }

pkg/accountability/simpleacc/internal/certificates/lightcertificates/lightcertificates.go

@@ -1,6 +1,8 @@
 package lightcertificates
 
 import (
+	"reflect"
+
 	"github.com/filecoin-project/mir/pkg/accountability/simpleacc/common"
 	incommon "github.com/filecoin-project/mir/pkg/accountability/simpleacc/internal/common"
 	"github.com/filecoin-project/mir/pkg/dsl"
@@ -9,7 +11,6 @@ import (
 	accpbmsgs "github.com/filecoin-project/mir/pkg/pb/accountabilitypb/msgs"
 	transportpbdsl "github.com/filecoin-project/mir/pkg/pb/transportpb/dsl"
 	t "github.com/filecoin-project/mir/pkg/types"
-	"reflect"
 )
 
 // IncludeLightCertificate implements the (optional) light certificate optimization

pkg/accountability/simpleacc/internal/common/common.go

@@ -13,8 +13,8 @@ type State struct {
 	LocalPredecision   *LocalPredecision            // Decision locally decided
 	DecidedCertificate *accpbtypes.FullCertificate  // Locally decided certificate (predecision and list of signatures with signers as key)
 	Predecided         bool                         // Whether this process has received a predecided value from calling module.
-	UnsentPoMs         []*accpbtypes.PoM            // List of PoMs not yet sent to the application.
-	SentPoMs           map[t.NodeID]*accpbtypes.PoM // List of PoMs already sent to the application with the signer as key.
+	UnhandledPoMs      []*accpbtypes.PoM            // List of PoMs not yet sent to the application.
+	HandledPoMs        map[t.NodeID]*accpbtypes.PoM // List of PoMs already sent to the application with the signer as key.
 	LightCertificates  map[t.NodeID][]byte          // Map of light certificates with the signer as key, buffered if no local decision made yet.
 }
 

pkg/accountability/simpleacc/internal/poms/poms.go

@@ -8,13 +8,10 @@ import (
 	"github.com/filecoin-project/mir/pkg/dsl"
 	"github.com/filecoin-project/mir/pkg/logging"
 	accpbdsl "github.com/filecoin-project/mir/pkg/pb/accountabilitypb/dsl"
-	accountabilitypbmsgs "github.com/filecoin-project/mir/pkg/pb/accountabilitypb/msgs"
 	accpbtypes "github.com/filecoin-project/mir/pkg/pb/accountabilitypb/types"
 	cryptopbdsl "github.com/filecoin-project/mir/pkg/pb/cryptopb/dsl"
 	cryptopbtypes "github.com/filecoin-project/mir/pkg/pb/cryptopb/types"
-	transportpbdsl "github.com/filecoin-project/mir/pkg/pb/transportpb/dsl"
 	t "github.com/filecoin-project/mir/pkg/types"
-	"github.com/filecoin-project/mir/pkg/util/maputil"
 )
 
 // IncludePoMs verifies receives PoMs and sends found PoMs to other members.
@@ -40,7 +37,7 @@ func IncludePoMs(
 				continue
 			}
 
-			if _, ok := state.SentPoMs[pom.NodeId]; ok {
+			if _, ok := state.HandledPoMs[pom.NodeId]; ok {
 				continue
 			}
 
@@ -72,43 +69,37 @@ func IncludePoMs(
 	cryptopbdsl.UponSigsVerified(m, func(nodeIds []t.NodeID, errs []error, allOk bool, vpoms *verifyPoMs) error {
 		for i := 0; i < len(nodeIds); i += 2 {
 			if errs[i] == nil && errs[i+1] == nil {
-				state.UnsentPoMs = append(state.UnsentPoMs, vpoms.poms[i/2])
+				state.UnhandledPoMs = append(state.UnhandledPoMs, vpoms.poms[i/2])
 			}
 		}
 
-		SendPoMs(m, mc, params, state, logger)
+		HandlePoMs(m, mc, params, state, logger)
 
 		return nil
 	})
 }
 
-// SendPoMs sends all PoMs in State.UnsentPoMs to all nodes and to the application module (from the POV of this module, i.e. mc.App).
-func SendPoMs(
+// HandlePoMs sends all PoMs in State.UnhandledPoMs to all nodes and to the application module (from the POV of this module, i.e. mc.App).
+func HandlePoMs(
 	m dsl.Module,
 	mc *common.ModuleConfig,
 	params *common.ModuleParams,
 	state *incommon.State,
 	logger logging.Logger,
 ) {
-	if len(state.UnsentPoMs) == 0 {
+	if len(state.UnhandledPoMs) == 0 {
 		return
 	}
 	logger.Log(logging.LevelWarn, "Found valid PoMs! sending...")
 
-	//TODO do something here (function that will be passed on factory creation)
+	// Handle PoMs according to the application's logic defined when creating the accountability factory
+	params.PomsHandler(m, mc, params, state, state.UnhandledPoMs, logger)
 
-	transportpbdsl.SendMessage(
-		m,
-		mc.Net,
-		accountabilitypbmsgs.PoMs(mc.Self, state.UnsentPoMs),
-		maputil.GetKeys(params.Membership.Nodes),
-	)
-
-	for _, pom := range state.UnsentPoMs {
-		state.SentPoMs[pom.NodeId] = pom
+	for _, pom := range state.UnhandledPoMs {
+		state.HandledPoMs[pom.NodeId] = pom
 	}
 
-	state.UnsentPoMs = make([]*accpbtypes.PoM, 0)
+	state.UnhandledPoMs = make([]*accpbtypes.PoM, 0)
 }
 
 type verifyPoMs struct {

pkg/accountability/simpleacc/internal/predecisions/predecisions.go

@@ -1,11 +1,12 @@
 package predecisions
 
 import (
+	"reflect"
+
 	"github.com/filecoin-project/mir/pkg/accountability/simpleacc/internal/certificates/lightcertificates"
 	isspbdsl "github.com/filecoin-project/mir/pkg/pb/isspb/dsl"
 	isspbtypes "github.com/filecoin-project/mir/pkg/pb/isspb/types"
 	tt "github.com/filecoin-project/mir/pkg/trantor/types"
-	"reflect"
 
 	"github.com/filecoin-project/mir/pkg/accountability/simpleacc/common"
 	incommon "github.com/filecoin-project/mir/pkg/accountability/simpleacc/internal/common"
@@ -134,16 +135,16 @@ func ApplySigVerified(
 		if !reflect.DeepEqual(state.SignedPredecisions[nodeID].Predecision, sp.Predecision) {
 			logger.Log(logging.LevelWarn, "Received conflicting signed predecisions from same node")
 			// if a PoM for this node has not already been sent.
-			if _, ok := state.SentPoMs[nodeID]; !ok {
-				state.UnsentPoMs = append(state.UnsentPoMs,
+			if _, ok := state.HandledPoMs[nodeID]; !ok {
+				state.UnhandledPoMs = append(state.UnhandledPoMs,
 					&accpbtypes.PoM{
 						NodeId:           nodeID,
 						ConflictingMsg_1: state.SignedPredecisions[nodeID],
 						ConflictingMsg_2: sp,
 					})
 
 				if flushPoMs {
-					poms.SendPoMs(m, mc, params, state, logger)
+					poms.HandlePoMs(m, mc, params, state, logger)
 				}
 			}