Execution result cache revamp (#172)

* Mod waitmap to be more generic

* Update usages for waitmap models

* Update waitmap implementation

* Tidy up comments

api/node.go

@@ -9,6 +9,6 @@ import (
 
 type Node interface {
 	ExecuteFunction(ctx context.Context, req execute.Request, subgroup string) (code codes.Code, requestID string, results execute.ResultMap, peers execute.Cluster, err error)
-	ExecutionResult(id string) (execute.Result, bool)
+	ExecutionResult(id string) (execute.ResultMap, bool)
 	PublishFunctionInstall(ctx context.Context, uri string, cid string, subgroup string) error
 }

api/result_test.go

@@ -29,17 +29,17 @@ func TestAPI_ExecutionResult(t *testing.T) {
 		err = srv.ExecutionResult(ctx)
 		require.NoError(t, err)
 
-		var res execute.Result
+		var res execute.ResultMap
 		require.NoError(t, json.Unmarshal(rec.Body.Bytes(), &res))
 
 		require.Equal(t, http.StatusOK, rec.Result().StatusCode)
-		require.Equal(t, mocks.GenericExecutionResult, res)
+		require.Equal(t, mocks.GenericExecutionResultMap, res)
 	})
 	t.Run("response not found", func(t *testing.T) {
 
 		node := mocks.BaselineNode(t)
-		node.ExecutionResultFunc = func(id string) (execute.Result, bool) {
-			return execute.Result{}, false
+		node.ExecutionResultFunc = func(id string) (execute.ResultMap, bool) {
+			return execute.ResultMap{}, false
 		}
 
 		srv := api.New(mocks.NoopLogger, node)

consensus/pbft/config.go

@@ -14,7 +14,7 @@ import (
 type Option func(*Config)
 
 // PostProcessFunc is invoked by the replica after execution is done.
-type PostProcessFunc func(requestID string, origin peer.ID, request execute.Request, result execute.Result)
+type PostProcessFunc func(requestID string, origin peer.ID, request execute.Request, result execute.NodeResult)
 
 var DefaultConfig = Config{
 	NetworkTimeout:   NetworkTimeout,

consensus/pbft/execute.go

@@ -95,34 +95,34 @@ func (r *Replica) execute(ctx context.Context, view uint, sequence uint, digest
 		log.Warn().Err(err).Msg("could not get metadata")
 	}
 
-	msg := response.Execute{
-		BaseMessage: blockless.BaseMessage{TraceInfo: r.cfg.TraceInfo},
-		Code:        res.Code,
-		RequestID:   request.ID,
-		Results: execute.ResultMap{
-			r.id: execute.NodeResult{
-				Result:   res,
-				Metadata: metadata,
-			},
-		},
-		PBFT: response.PBFTResultInfo{
+	nres := execute.NodeResult{
+		Result:   res,
+		Metadata: metadata,
+		PBFT: execute.PBFTResultInfo{
 			View:             r.view,
 			RequestTimestamp: request.Timestamp,
 			Replica:          r.id,
 		},
 	}
 
+	err = nres.Sign(r.host.PrivateKey())
+	if err != nil {
+		return fmt.Errorf("could not sign execution result: %w", err)
+	}
+
+	msg := response.Execute{
+		BaseMessage: blockless.BaseMessage{TraceInfo: r.cfg.TraceInfo},
+		Code:        res.Code,
+		RequestID:   request.ID,
+		Results:     execute.ResultMap{r.id: nres},
+	}
+
 	// Save this executions in case it's requested again.
 	r.executions[request.ID] = msg
 
 	// Invoke specified post processor functions.
 	for _, proc := range r.cfg.PostProcessors {
-		proc(request.ID, request.Origin, request.Execute, res)
-	}
-
-	err = msg.Sign(r.host.PrivateKey())
-	if err != nil {
-		return fmt.Errorf("could not sign execution request: %w", err)
+		proc(request.ID, request.Origin, request.Execute, nres)
 	}
 
 	err = r.send(ctx, request.Origin, &msg, blockless.ProtocolID)

consensus/raft/fsm.go

@@ -22,7 +22,7 @@ type FSMLogEntry struct {
 	Execute   execute.Request `json:"execute,omitempty"`
 }
 
-type FSMProcessFunc func(req FSMLogEntry, res execute.Result)
+type FSMProcessFunc func(req FSMLogEntry, res execute.NodeResult)
 
 type fsmExecutor struct {
 	log        zerolog.Logger
@@ -36,7 +36,7 @@ func newFsmExecutor(log zerolog.Logger, executor blockless.Executor, processors
 	ps = append(ps, processors...)
 
 	start := time.Now()
-	ps = append(ps, func(req FSMLogEntry, res execute.Result) {
+	ps = append(ps, func(req FSMLogEntry, _ execute.NodeResult) {
 		// Global metrics handle.
 		metrics.MeasureSinceWithLabels(raftExecutionTimeMetric, start, []metrics.Label{{Name: "function", Value: req.Execute.FunctionID}})
 	})
@@ -50,7 +50,7 @@ func newFsmExecutor(log zerolog.Logger, executor blockless.Executor, processors
 	return &fsm
 }
 
-func (f fsmExecutor) Apply(log *raft.Log) interface{} {
+func (f fsmExecutor) Apply(log *raft.Log) any {
 
 	f.log.Info().Msg("applying log entry")
 
@@ -70,9 +70,13 @@ func (f fsmExecutor) Apply(log *raft.Log) interface{} {
 		return fmt.Errorf("could not execute function: %w", err)
 	}
 
+	nres := execute.NodeResult{
+		Result: res,
+	}
+
 	// Execute processors.
 	for _, proc := range f.processors {
-		proc(logEntry, res)
+		proc(logEntry, nres)
 	}
 
 	f.log.Info().Str("request", logEntry.RequestID).Msg("FSM successfully executed function")

host/discovery.go

@@ -42,6 +42,7 @@ func (h *Host) ConnectToKnownPeers(ctx context.Context) error {
 			case <-ctx.Done():
 				ticker.Stop()
 				h.log.Debug().Msg("stopping boot node reachability monitoring")
+				return
 			}
 		}
 	}(ctx)

models/execute/response.go → models/execute/result.go

@@ -1,9 +1,13 @@
 package execute
 
 import (
+	"encoding/hex"
 	"encoding/json"
+	"errors"
+	"fmt"
 	"time"
 
+	"github.com/libp2p/go-libp2p/core/crypto"
 	"github.com/libp2p/go-libp2p/core/peer"
 
 	"github.com/blocklessnetwork/b7s/models/codes"
@@ -12,7 +16,10 @@ import (
 // NodeResult is an annotated execution result.
 type NodeResult struct {
 	Result
-	Metadata any `json:"metadata,omitempty"`
+	// Signed digest of the response.
+	Signature string         `json:"signature,omitempty"`
+	PBFT      PBFTResultInfo `json:"pbft,omitempty"`
+	Metadata  any            `json:"metadata,omitempty"`
 }
 
 // Result describes an execution result.
@@ -44,6 +51,12 @@ type Usage struct {
 	MemoryMaxKB   int64         `json:"memory_max_kb,omitempty"`
 }
 
+type PBFTResultInfo struct {
+	View             uint      `json:"view"`
+	RequestTimestamp time.Time `json:"request_timestamp,omitempty"`
+	Replica          peer.ID   `json:"replica,omitempty"`
+}
+
 // ResultMap contains execution results from multiple peers.
 type ResultMap map[peer.ID]NodeResult
 
@@ -61,3 +74,51 @@ func (m ResultMap) MarshalJSON() ([]byte, error) {
 
 	return json.Marshal(em)
 }
+
+func (r *NodeResult) Sign(key crypto.PrivKey) error {
+
+	cp := *r
+	// Exclude some of the fields from the signature.
+	cp.Signature = ""
+
+	payload, err := json.Marshal(cp)
+	if err != nil {
+		return fmt.Errorf("could not get byte representation of the record: %w", err)
+	}
+
+	sig, err := key.Sign(payload)
+	if err != nil {
+		return fmt.Errorf("could not sign digest: %w", err)
+	}
+
+	r.Signature = hex.EncodeToString(sig)
+	return nil
+}
+
+func (r NodeResult) VerifySignature(key crypto.PubKey) error {
+
+	cp := r
+	// Exclude some of the fields from the signature.
+	cp.Signature = ""
+
+	payload, err := json.Marshal(cp)
+	if err != nil {
+		return fmt.Errorf("could not get byte representation of the record: %w", err)
+	}
+
+	sig, err := hex.DecodeString(r.Signature)
+	if err != nil {
+		return fmt.Errorf("could not decode signature from hex: %w", err)
+	}
+
+	ok, err := key.Verify(payload, sig)
+	if err != nil {
+		return fmt.Errorf("could not verify signature: %w", err)
+	}
+
+	if !ok {
+		return errors.New("invalid signature")
+	}
+
+	return nil
+}

models/execute/request_signature_test.go → models/execute/signature_test.go

@@ -3,11 +3,12 @@ package execute
 import (
 	"testing"
 
+	"github.com/blocklessnetwork/b7s/models/codes"
 	"github.com/libp2p/go-libp2p/core/crypto"
 	"github.com/stretchr/testify/require"
 )
 
-func TestExecute_Signing(t *testing.T) {
+func TestRequestExecute_Signing(t *testing.T) {
 
 	sampleReq := Request{
 		FunctionID: "function-di",
@@ -57,6 +58,55 @@ func TestExecute_Signing(t *testing.T) {
 	})
 }
 
+func TestResultExecute_Signing(t *testing.T) {
+
+	sampleRes := NodeResult{
+		Result: Result{
+			Code: codes.Unknown,
+			Result: RuntimeOutput{
+				Stdout:   "generic-execution-result",
+				Stderr:   "generic-execution-log",
+				ExitCode: 0,
+			},
+		},
+	}
+
+	t.Run("nominal case", func(t *testing.T) {
+
+		res := sampleRes
+		priv, pub := newKey(t)
+
+		err := res.Sign(priv)
+		require.NoError(t, err)
+
+		err = res.VerifySignature(pub)
+		require.NoError(t, err)
+	})
+	t.Run("empty signature verification fails", func(t *testing.T) {
+
+		res := sampleRes
+		res.Signature = ""
+
+		_, pub := newKey(t)
+
+		err := res.VerifySignature(pub)
+		require.Error(t, err)
+	})
+	t.Run("tampered data signature verification fails", func(t *testing.T) {
+
+		res := sampleRes
+		priv, pub := newKey(t)
+
+		err := res.Sign(priv)
+		require.NoError(t, err)
+
+		res.Result.Result.Stdout += " "
+
+		err = res.VerifySignature(pub)
+		require.Error(t, err)
+	})
+}
+
 func newKey(t *testing.T) (crypto.PrivKey, crypto.PubKey) {
 	t.Helper()
 	priv, pub, err := crypto.GenerateKeyPair(crypto.Ed25519, 0)

models/response/execute.go

@@ -1,14 +1,7 @@
 package response
 
 import (
-	"encoding/hex"
 	"encoding/json"
-	"errors"
-	"fmt"
-	"time"
-
-	"github.com/libp2p/go-libp2p/core/crypto"
-	"github.com/libp2p/go-libp2p/core/peer"
 
 	"github.com/blocklessnetwork/b7s/models/blockless"
 	"github.com/blocklessnetwork/b7s/models/codes"
@@ -25,10 +18,6 @@ type Execute struct {
 	Results   execute.ResultMap `json:"results,omitempty"`
 	Cluster   execute.Cluster   `json:"cluster,omitempty"`
 
-	PBFT PBFTResultInfo `json:"pbft,omitempty"`
-	// Signed digest of the response.
-	Signature string `json:"signature,omitempty"`
-
 	// Used to communicate the reason for failure to the user.
 	ErrorMessage string `json:"message,omitempty"`
 }
@@ -61,59 +50,3 @@ func (e Execute) MarshalJSON() ([]byte, error) {
 	}
 	return json.Marshal(rec)
 }
-
-type PBFTResultInfo struct {
-	View             uint      `json:"view"`
-	RequestTimestamp time.Time `json:"request_timestamp,omitempty"`
-	Replica          peer.ID   `json:"replica,omitempty"`
-}
-
-func (e *Execute) Sign(key crypto.PrivKey) error {
-
-	cp := *e
-	// Exclude some of the fields from the signature.
-	cp.Signature = ""
-	cp.BaseMessage = blockless.BaseMessage{}
-
-	payload, err := json.Marshal(cp)
-	if err != nil {
-		return fmt.Errorf("could not get byte representation of the record: %w", err)
-	}
-
-	sig, err := key.Sign(payload)
-	if err != nil {
-		return fmt.Errorf("could not sign digest: %w", err)
-	}
-
-	e.Signature = hex.EncodeToString(sig)
-	return nil
-}
-
-func (e Execute) VerifySignature(key crypto.PubKey) error {
-
-	cp := e
-	// Exclude some of the fields from the signature.
-	cp.Signature = ""
-	cp.BaseMessage = blockless.BaseMessage{}
-
-	payload, err := json.Marshal(cp)
-	if err != nil {
-		return fmt.Errorf("could not get byte representation of the record: %w", err)
-	}
-
-	sig, err := hex.DecodeString(e.Signature)
-	if err != nil {
-		return fmt.Errorf("could not decode signature from hex: %w", err)
-	}
-
-	ok, err := key.Verify(payload, sig)
-	if err != nil {
-		return fmt.Errorf("could not verify signature: %w", err)
-	}
-
-	if !ok {
-		return errors.New("invalid signature")
-	}
-
-	return nil
-}