Merge pull request #128 from blocklessnetwork/multiple-topics

api/execute.go

@@ -14,7 +14,10 @@ import (
 )
 
 // ExecuteRequest describes the payload for the REST API request for function execution.
-type ExecuteRequest execute.Request
+type ExecuteRequest struct {
+	execute.Request
+	Subgroup string `json:"subgroup,omitempty"`
+}
 
 // ExecuteResponse describes the REST API response for function execution.
 type ExecuteResponse struct {
@@ -44,7 +47,7 @@ func (a *API) Execute(ctx echo.Context) error {
 	}
 
 	// Get the execution result.
-	code, id, results, cluster, err := a.Node.ExecuteFunction(ctx.Request().Context(), execute.Request(req))
+	code, id, results, cluster, err := a.Node.ExecuteFunction(ctx.Request().Context(), execute.Request(req.Request), req.Subgroup)
 	if err != nil {
 		a.Log.Warn().Str("function", req.FunctionID).Err(err).Msg("node failed to execute function")
 	}

api/execute_test.go

@@ -31,7 +31,7 @@ func TestAPI_Execute(t *testing.T) {
 	expectedCode := codes.OK
 
 	node := mocks.BaselineNode(t)
-	node.ExecuteFunctionFunc = func(context.Context, execute.Request) (codes.Code, string, execute.ResultMap, execute.Cluster, error) {
+	node.ExecuteFunctionFunc = func(context.Context, execute.Request, string) (codes.Code, string, execute.ResultMap, execute.Cluster, error) {
 
 		res := execute.ResultMap{
 			mocks.GenericPeerID: executionResult,
@@ -85,7 +85,7 @@ func TestAPI_Execute_HandlesErrors(t *testing.T) {
 	expectedCode := codes.Error
 
 	node := mocks.BaselineNode(t)
-	node.ExecuteFunctionFunc = func(context.Context, execute.Request) (codes.Code, string, execute.ResultMap, execute.Cluster, error) {
+	node.ExecuteFunctionFunc = func(context.Context, execute.Request, string) (codes.Code, string, execute.ResultMap, execute.Cluster, error) {
 
 		res := execute.ResultMap{
 			mocks.GenericPeerID: executionResult,

api/install.go

@@ -17,8 +17,9 @@ const (
 
 // InstallFunctionRequest describes the payload for the REST API request for function install.
 type InstallFunctionRequest struct {
-	CID string `json:"cid"`
-	URI string `json:"uri"`
+	CID      string `json:"cid"`
+	URI      string `json:"uri"`
+	Subgroup string `json:"subgroup"`
 }
 
 // InstallFunctionResponse describes the REST API response for the function install.
@@ -46,7 +47,7 @@ func (a *API) Install(ctx echo.Context) error {
 	// Start function install in a separate goroutine and signal when it's done.
 	fnErr := make(chan error)
 	go func() {
-		err = a.Node.PublishFunctionInstall(reqCtx, req.URI, req.CID)
+		err = a.Node.PublishFunctionInstall(reqCtx, req.URI, req.CID, req.Subgroup)
 		fnErr <- err
 	}()
 

api/install_test.go

@@ -67,7 +67,7 @@ func TestAPI_FunctionInstall_HandlesErrors(t *testing.T) {
 		)
 
 		node := mocks.BaselineNode(t)
-		node.PublishFunctionInstallFunc = func(context.Context, string, string) error {
+		node.PublishFunctionInstallFunc = func(context.Context, string, string, string) error {
 			time.Sleep(installDuration)
 			return nil
 		}
@@ -99,7 +99,7 @@ func TestAPI_FunctionInstall_HandlesErrors(t *testing.T) {
 		t.Parallel()
 
 		node := mocks.BaselineNode(t)
-		node.PublishFunctionInstallFunc = func(context.Context, string, string) error {
+		node.PublishFunctionInstallFunc = func(context.Context, string, string, string) error {
 			return mocks.GenericError
 		}
 

api/node.go

@@ -8,7 +8,7 @@ import (
 )
 
 type Node interface {
-	ExecuteFunction(context.Context, execute.Request) (code codes.Code, requestID string, results execute.ResultMap, peers execute.Cluster, err error)
+	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)
-	PublishFunctionInstall(ctx context.Context, uri string, cid string) error
+	PublishFunctionInstall(ctx context.Context, uri string, cid string, subgroup string) error
 }

cmd/node/flags.go

@@ -33,8 +33,9 @@ func parseFlags() *config.Config {
 	pflag.StringVar(&cfg.API, "rest-api", "", "address where the head node REST API will listen on")
 	pflag.StringVar(&cfg.Workspace, "workspace", "./workspace", "directory that the node can use for file storage")
 	pflag.StringVar(&cfg.RuntimePath, "runtime-path", "", "runtime path (used by the worker node)")
-	pflag.StringVar(&cfg.RuntimeCLI, "runtime-cli", "", "runtime path (used by the worker node)")
+	pflag.StringVar(&cfg.RuntimeCLI, "runtime-cli", "", "runtime CLI name (used by the worker node)")
 	pflag.BoolVar(&cfg.LoadAttributes, "attributes", false, "node should try to load its attribute data from IPFS")
+	pflag.StringSliceVar(&cfg.Topics, "topic", nil, "topics node should subscribe to")
 
 	// Host configuration.
 	pflag.StringVar(&cfg.Host.PrivateKey, "private-key", "", "private key that the b7s host will use")

cmd/node/main.go

@@ -157,7 +157,7 @@ func run() int {
 		executor, err := executor.New(log, execOptions...)
 		if err != nil {
 			log.Error().
-			Err(err).
+				Err(err).
 				Str("workspace", cfg.Workspace).
 				Str("runtime_path", cfg.RuntimePath).
 				Msg("could not create an executor")
@@ -181,6 +181,11 @@ func run() int {
 	// Create function store.
 	fstore := fstore.New(log, functionStore, cfg.Workspace)
 
+	// If we have topics specified, use those.
+	if len(cfg.Topics) > 0 {
+		opts = append(opts, node.WithTopics(cfg.Topics))
+	}
+
 	// Instantiate node.
 	node, err := node.New(log, host, peerstore, fstore, opts...)
 	if err != nil {

config/model.go

@@ -8,12 +8,13 @@ type Config struct {
 	Role                 string
 	BootNodes            []string
 	Concurrency          uint
+	Topics               []string
 
-	Host            Host
-	API             string
-	RuntimePath     string
-	RuntimeCLI 		string
-	LoadAttributes  bool
+	Host           Host
+	API            string
+	RuntimePath    string
+	RuntimeCLI     string
+	LoadAttributes bool
 
 	CPUPercentage float64
 	MemoryMaxKB   int64

host/host.go

@@ -9,6 +9,7 @@ import (
 	"github.com/rs/zerolog"
 
 	"github.com/libp2p/go-libp2p"
+	pubsub "github.com/libp2p/go-libp2p-pubsub"
 	"github.com/libp2p/go-libp2p/core/crypto"
 	"github.com/libp2p/go-libp2p/core/host"
 	ma "github.com/multiformats/go-multiaddr"
@@ -20,6 +21,8 @@ type Host struct {
 
 	log zerolog.Logger
 	cfg Config
+
+	pubsub *pubsub.PubSub
 }
 
 // New creates a new Host.

host/subscribe.go

@@ -2,22 +2,44 @@ package host
 
 import (
 	"context"
+	"errors"
 	"fmt"
 
 	pubsub "github.com/libp2p/go-libp2p-pubsub"
 )
 
-// Subscribe will have the host start listening to a specified gossipsub topic.
-func (h *Host) Subscribe(ctx context.Context, topic string) (*pubsub.Topic, *pubsub.Subscription, error) {
+func (h *Host) InitPubSub(ctx context.Context) error {
 
 	// Get a new PubSub object with the default router.
 	pubsub, err := pubsub.NewGossipSub(ctx, h)
 	if err != nil {
-		return nil, nil, fmt.Errorf("could not create new gossipsub: %w", err)
+		return fmt.Errorf("could not create new gossipsub: %w", err)
+	}
+	h.pubsub = pubsub
+
+	return nil
+}
+
+func (h *Host) JoinTopic(topic string) (*pubsub.Topic, error) {
+
+	if h.pubsub == nil {
+		return nil, errors.New("pubsub is not initialized")
+	}
+
+	// Join the specified topic.
+	th, err := h.pubsub.Join(topic)
+	if err != nil {
+		return nil, fmt.Errorf("could not join topic: %w", err)
 	}
 
+	return th, nil
+}
+
+// Subscribe will have the host start listening to a specified gossipsub topic.
+func (h *Host) Subscribe(topic string) (*pubsub.Topic, *pubsub.Subscription, error) {
+
 	// Join the specified topic.
-	th, err := pubsub.Join(topic)
+	th, err := h.JoinTopic(topic)
 	if err != nil {
 		return nil, nil, fmt.Errorf("could not join topic: %w", err)
 	}

node/config.go

@@ -15,7 +15,7 @@ type Option func(*Config)
 // DefaultConfig represents the default settings for the node.
 var DefaultConfig = Config{
 	Role:                    blockless.WorkerNode,
-	Topic:                   DefaultTopic,
+	Topics:                  []string{DefaultTopic},
 	HealthInterval:          DefaultHealthInterval,
 	RollCallTimeout:         DefaultRollCallTimeout,
 	Concurrency:             DefaultConcurrency,
@@ -28,7 +28,7 @@ var DefaultConfig = Config{
 // Config represents the Node configuration.
 type Config struct {
 	Role                    blockless.NodeRole // Node role.
-	Topic                   string             // Topic to subscribe to.
+	Topics                  []string           // Topics to subscribe to.
 	Execute                 blockless.Executor // Executor to use for running functions.
 	HealthInterval          time.Duration      // How often should we emit the health ping.
 	RollCallTimeout         time.Duration      // How long do we wait for roll call responses.
@@ -47,8 +47,8 @@ func (n *Node) ValidateConfig() error {
 		return errors.New("node role is not valid")
 	}
 
-	if n.cfg.Topic == "" {
-		return errors.New("topic cannot be empty")
+	if len(n.cfg.Topics) == 0 {
+		return errors.New("topics cannot be empty")
 	}
 
 	// Worker specific validation.
@@ -83,10 +83,10 @@ func WithRole(role blockless.NodeRole) Option {
 	}
 }
 
-// WithTopic specifies the p2p topic to which node should subscribe.
-func WithTopic(topic string) Option {
+// WithTopics specifies the p2p topics to which node should subscribe.
+func WithTopics(topics []string) Option {
 	return func(cfg *Config) {
-		cfg.Topic = topic
+		cfg.Topics = topics
 	}
 }
 

node/config_internal_test.go

@@ -24,14 +24,14 @@ func TestConfig_NodeRole(t *testing.T) {
 
 func TestConfig_Topic(t *testing.T) {
 
-	const topic = "super-secret-topic"
+	topics := []string{"super-secret-topic"}
 
 	cfg := Config{
-		Topic: "",
+		Topics: []string{},
 	}
 
-	WithTopic(topic)(&cfg)
-	require.Equal(t, topic, cfg.Topic)
+	WithTopics(topics)(&cfg)
+	require.Equal(t, topics, cfg.Topics)
 }
 
 func TestConfig_Executor(t *testing.T) {

node/execute_internal_test.go

@@ -277,7 +277,7 @@ func TestNode_HeadExecute(t *testing.T) {
 		node := createNode(t, blockless.HeadNode)
 
 		ctx := context.Background()
-		_, err := node.subscribe(ctx)
+		err := node.subscribeToTopics(ctx)
 		require.NoError(t, err)
 
 		// Create a host that will receive the execution response.
@@ -331,7 +331,7 @@ func TestNode_HeadExecute(t *testing.T) {
 		node.listenDirectMessages(ctx)
 
 		defer cancel()
-		_, err := node.subscribe(ctx)
+		err := node.subscribeToTopics(ctx)
 		require.NoError(t, err)
 
 		// Create a host that will simulate a worker.
@@ -340,7 +340,10 @@ func TestNode_HeadExecute(t *testing.T) {
 		mockWorker, err := host.New(mocks.NoopLogger, loopback, 0)
 		require.NoError(t, err)
 
-		_, subscription, err := mockWorker.Subscribe(ctx, topic)
+		err = mockWorker.InitPubSub(ctx)
+		require.NoError(t, err)
+
+		_, subscription, err := mockWorker.Subscribe(topic)
 		require.NoError(t, err)
 
 		hostAddNewPeer(t, node.host, mockWorker)

node/head_execute.go

@@ -17,6 +17,7 @@ import (
 	"github.com/blocklessnetwork/b7s/models/response"
 )
 
+// NOTE: head node typically receives execution requests from the REST API. This message handling is not cognizant of subgroups.
 func (n *Node) headProcessExecute(ctx context.Context, from peer.ID, payload []byte) error {
 
 	// Unpack the request.
@@ -34,7 +35,7 @@ func (n *Node) headProcessExecute(ctx context.Context, from peer.ID, payload []b
 
 	log := n.log.With().Str("request", req.RequestID).Str("peer", from.String()).Str("function", req.FunctionID).Logger()
 
-	code, results, cluster, err := n.headExecute(ctx, requestID, req.Request)
+	code, results, cluster, err := n.headExecute(ctx, requestID, req.Request, "")
 	if err != nil {
 		log.Error().Err(err).Msg("execution failed")
 	}
@@ -66,7 +67,7 @@ func (n *Node) headProcessExecute(ctx context.Context, from peer.ID, payload []b
 
 // headExecute is called on the head node. The head node will publish a roll call and delegate an execution request to chosen nodes.
 // The returned map contains execution results, mapped to the peer IDs of peers who reported them.
-func (n *Node) headExecute(ctx context.Context, requestID string, req execute.Request) (codes.Code, execute.ResultMap, execute.Cluster, error) {
+func (n *Node) headExecute(ctx context.Context, requestID string, req execute.Request, subgroup string) (codes.Code, execute.ResultMap, execute.Cluster, error) {
 
 	nodeCount := 1
 	if req.Config.NodeCount > 1 {
@@ -89,7 +90,7 @@ func (n *Node) headExecute(ctx context.Context, requestID string, req execute.Re
 	log.Info().Msg("processing execution request")
 
 	// Phase 1. - Issue roll call to nodes.
-	reportingPeers, err := n.executeRollCall(ctx, requestID, req.FunctionID, nodeCount, consensusAlgo, req.Config.Attributes)
+	reportingPeers, err := n.executeRollCall(ctx, requestID, req.FunctionID, nodeCount, consensusAlgo, subgroup, req.Config.Attributes)
 	if err != nil {
 		code := codes.Error
 		if errors.Is(err, blockless.ErrRollCallTimeout) {

node/health_internal_test.go

@@ -37,7 +37,7 @@ func TestNode_Health(t *testing.T) {
 	nhost, err := host.New(logger, loopback, 0)
 	require.NoError(t, err)
 
-	node, err := New(logger, nhost, peerstore, functionHandler, WithRole(blockless.HeadNode), WithHealthInterval(healthInterval), WithTopic(topic))
+	node, err := New(logger, nhost, peerstore, functionHandler, WithRole(blockless.HeadNode), WithHealthInterval(healthInterval), WithTopics([]string{topic}))
 	require.NoError(t, err)
 
 	// Create a host that will listen on the the topic to verify health pings
@@ -55,11 +55,14 @@ func TestNode_Health(t *testing.T) {
 	err = node.host.Connect(ctx, *info)
 	require.NoError(t, err)
 
+	err = receiver.InitPubSub(ctx)
+	require.NoError(t, err)
+
 	// Have both client and node subscribe to the same topic.
-	_, subscription, err := receiver.Subscribe(ctx, topic)
+	_, subscription, err := receiver.Subscribe(topic)
 	require.NoError(t, err)
 
-	_, err = node.subscribe(ctx)
+	err = node.subscribeToTopics(ctx)
 	require.NoError(t, err)
 
 	go node.HealthPing(ctx)