Split worker and head node implementations

api/install.go

@@ -1,20 +1,14 @@
 package api
 
 import (
-	"context"
 	"errors"
 	"fmt"
 	"net/http"
 	"strconv"
-	"time"
 
 	"github.com/labstack/echo/v4"
 )
 
-const (
-	functionInstallTimeout = 10 * time.Second
-)
-
 func (r FunctionInstallRequest) Valid() error {
 
 	if r.Cid == "" {
@@ -38,39 +32,7 @@ func (a *API) InstallFunction(ctx echo.Context) error {
 		return echo.NewHTTPError(http.StatusBadRequest, fmt.Errorf("invalid request: %w", err))
 	}
 
-	// Add a deadline to the context.
-	reqCtx, cancel := context.WithTimeout(ctx.Request().Context(), functionInstallTimeout)
-	defer cancel()
-
-	// 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, req.Topic)
-		fnErr <- err
-	}()
-
-	// Wait until either function install finishes, or request times out.
-	select {
-
-	// Context timed out.
-	case <-reqCtx.Done():
-
-		status := http.StatusRequestTimeout
-		if !errors.Is(reqCtx.Err(), context.DeadlineExceeded) {
-			status = http.StatusInternalServerError
-		}
-
-		// return inner code as body
-		return ctx.JSON(200, map[string]interface{}{
-			"code": strconv.Itoa(status),
-		})
-
-	// Work done.
-	case err = <-fnErr:
-		break
-	}
-
-	// Check if function install succeeded and handle error or return response.
+	err = a.Node.PublishFunctionInstall(ctx.Request().Context(), req.Uri, req.Cid, req.Topic)
 	if err != nil {
 		return echo.NewHTTPError(http.StatusInternalServerError, fmt.Errorf("function installation failed: %w", err))
 	}

api/install_test.go

@@ -2,11 +2,8 @@ package api_test
 
 import (
 	"context"
-	"encoding/json"
 	"net/http"
-	"strconv"
 	"testing"
-	"time"
 
 	"github.com/labstack/echo/v4"
 	"github.com/stretchr/testify/require"
@@ -58,43 +55,6 @@ func TestAPI_FunctionInstall_HandlesErrors(t *testing.T) {
 
 		require.Equal(t, http.StatusBadRequest, echoErr.Code)
 	})
-	t.Run("node install takes too long", func(t *testing.T) {
-		t.Parallel()
-
-		const (
-			// The API times out after 10 seconds.
-			installDuration = 11 * time.Second
-		)
-
-		node := mocks.BaselineNode(t)
-		node.PublishFunctionInstallFunc = func(context.Context, string, string, string) error {
-			time.Sleep(installDuration)
-			return nil
-		}
-
-		req := api.FunctionInstallRequest{
-			Uri: "dummy-uri",
-			Cid: "dummy-cid",
-		}
-
-		srv := api.New(mocks.NoopLogger, node)
-
-		rec, ctx, err := setupRecorder(installEndpoint, req)
-		require.NoError(t, err)
-
-		err = srv.InstallFunction(ctx)
-		require.NoError(t, err)
-
-		require.Equal(t, http.StatusOK, rec.Result().StatusCode)
-
-		var res api.FunctionInstallResponse
-		require.NoError(t, json.Unmarshal(rec.Body.Bytes(), &res))
-
-		num, err := strconv.Atoi(res.Code)
-		require.NoError(t, err)
-
-		require.Equal(t, http.StatusRequestTimeout, num)
-	})
 	t.Run("node fails to install function", func(t *testing.T) {
 		t.Parallel()
 

cmd/node/main.go

@@ -9,6 +9,7 @@ import (
 	"os"
 	"os/signal"
 	"path/filepath"
+	"slices"
 
 	"github.com/cockroachdb/pebble"
 	"github.com/labstack/echo-contrib/echoprometheus"
@@ -19,9 +20,7 @@ import (
 
 	"github.com/blocklessnetwork/b7s/api"
 	"github.com/blocklessnetwork/b7s/config"
-	"github.com/blocklessnetwork/b7s/executor"
-	"github.com/blocklessnetwork/b7s/executor/limits"
-	"github.com/blocklessnetwork/b7s/fstore"
+	b7shost "github.com/blocklessnetwork/b7s/host"
 	"github.com/blocklessnetwork/b7s/models/blockless"
 	"github.com/blocklessnetwork/b7s/node"
 	"github.com/blocklessnetwork/b7s/store"
@@ -201,72 +200,53 @@ func run() int {
 	}
 	defer host.Close()
 
+	host.Network().Notify(b7shost.NewNotifee(
+		log.With().Str("component", "notifiee").Logger(),
+		store,
+	))
+
 	log.Info().
 		Str("id", host.ID().String()).
 		Strs("addresses", host.Addresses()).
 		Strs("boot_nodes", cfg.BootNodes).
 		Msg("created host")
 
-	// Set node options.
-	opts := []node.Option{
-		node.WithRole(nodeRole),
-		node.WithConcurrency(cfg.Concurrency),
-		node.WithAttributeLoading(cfg.LoadAttributes),
+	// Ensure default topic is included in the topic list.
+	if !slices.Contains(cfg.Topics, blockless.DefaultTopic) {
+		cfg.Topics = append(cfg.Topics, blockless.DefaultTopic)
 	}
 
-	// If this is a worker node, initialize an executor.
-	if nodeRole == blockless.WorkerNode {
+	// Instantiate node.
 
-		// Executor options.
-		execOptions := []executor.Option{
-			executor.WithWorkDir(cfg.Workspace),
-			executor.WithRuntimeDir(cfg.Worker.RuntimePath),
-			executor.WithExecutableName(cfg.Worker.RuntimeCLI),
-		}
+	// First, initialize the node core, common for both node types.
+	core := node.NewCore(
+		log.With().Str("component", "node").Logger(),
+		host,
+		node.Concurrency(cfg.Concurrency),
+		node.Topics(cfg.Topics),
+	)
 
-		if needLimiter(cfg) {
-			limiter, err := limits.New(limits.WithCPUPercentage(cfg.Worker.CPUPercentageLimit), limits.WithMemoryKB(cfg.Worker.MemoryLimitKB))
-			if err != nil {
-				log.Error().Err(err).Msg("could not create resource limiter")
-				return failure
-			}
+	var (
+		node         Node
+		nodeshutdown func() error
+	)
+
+	switch nodeRole {
+	case blockless.WorkerNode:
+		node, nodeshutdown, err = createWorkerNode(core, store, cfg)
 
+		if nodeshutdown != nil {
 			defer func() {
-				err = limiter.Shutdown()
+				err = nodeshutdown()
 				if err != nil {
-					log.Error().Err(err).Msg("could not shutdown resource limiter")
+					log.Error().Err(err).Msg("node shutdown function failed")
 				}
 			}()
-
-			execOptions = append(execOptions, executor.WithLimiter(limiter))
 		}
 
-		// Create an executor.
-		executor, err := executor.New(log.With().Str("component", "executor").Logger(), execOptions...)
-		if err != nil {
-			log.Error().
-				Err(err).
-				Str("workspace", cfg.Workspace).
-				Str("runtime_path", cfg.Worker.RuntimePath).
-				Str("runtime_cli", cfg.Worker.RuntimeCLI).
-				Msg("could not create an executor")
-			return failure
-		}
-
-		opts = append(opts, node.WithExecutor(executor))
-		opts = append(opts, node.WithWorkspace(cfg.Workspace))
-	}
-
-	// Create function store.
-	fstore := fstore.New(log.With().Str("component", "fstore").Logger(), store, cfg.Workspace)
-
-	// If we have topics specified, use those.
-	if len(cfg.Topics) > 0 {
-		opts = append(opts, node.WithTopics(cfg.Topics))
+	case blockless.HeadNode:
+		node, err = createHeadNode(core, cfg)
 	}
-
-	// Instantiate node.
-	node, err := node.New(log.With().Str("component", "node").Logger(), host, store, fstore, opts...)
 	if err != nil {
 		log.Error().Err(err).Msg("could not create node")
 		return failure
@@ -299,7 +279,12 @@ func run() int {
 		// Create an API handler if we're a head node.
 		if nodeRole == blockless.HeadNode {
 
-			apiHandler := api.New(log.With().Str("component", "api").Logger(), node)
+			headNode, ok := any(node).(api.Node)
+			if !ok {
+				log.Error().Msg("invalid node type - not a head node")
+			}
+
+			apiHandler := api.New(log.With().Str("component", "api").Logger(), headNode)
 			api.RegisterHandlers(server, apiHandler)
 		}
 

cmd/node/node.go

@@ -0,0 +1,74 @@
+package main
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/blocklessnetwork/b7s/config"
+	"github.com/blocklessnetwork/b7s/executor"
+	"github.com/blocklessnetwork/b7s/executor/limits"
+	"github.com/blocklessnetwork/b7s/fstore"
+	"github.com/blocklessnetwork/b7s/models/blockless"
+	"github.com/blocklessnetwork/b7s/node"
+	"github.com/blocklessnetwork/b7s/node/head"
+	"github.com/blocklessnetwork/b7s/node/worker"
+)
+
+type Node interface {
+	Run(context.Context) error
+}
+
+func createWorkerNode(core node.Core, store blockless.Store, cfg *config.Config) (Node, func() error, error) {
+
+	// Create function store.
+	fstore := fstore.New(log.With().Str("component", "fstore").Logger(), store, cfg.Workspace)
+
+	// Executor options.
+	execOptions := []executor.Option{
+		executor.WithWorkDir(cfg.Workspace),
+		executor.WithRuntimeDir(cfg.Worker.RuntimePath),
+		executor.WithExecutableName(cfg.Worker.RuntimeCLI),
+	}
+
+	shutdown := func() error {
+		return nil
+	}
+	if needLimiter(cfg) {
+		limiter, err := limits.New(limits.WithCPUPercentage(cfg.Worker.CPUPercentageLimit), limits.WithMemoryKB(cfg.Worker.MemoryLimitKB))
+		if err != nil {
+			return nil, shutdown, fmt.Errorf("could not create resource limiter")
+		}
+
+		shutdown = func() error {
+			return limiter.Shutdown()
+		}
+
+		execOptions = append(execOptions, executor.WithLimiter(limiter))
+	}
+
+	// Create an executor.
+	executor, err := executor.New(log.With().Str("component", "executor").Logger(), execOptions...)
+	if err != nil {
+		return nil, shutdown, fmt.Errorf("could not create an executor: %w", err)
+	}
+
+	worker, err := worker.New(core, fstore, executor,
+		worker.AttributeLoading(cfg.LoadAttributes),
+		worker.Workspace(cfg.Workspace),
+	)
+	if err != nil {
+		return nil, shutdown, fmt.Errorf("could not create a worker node: %w", err)
+	}
+
+	return worker, shutdown, nil
+}
+
+func createHeadNode(core node.Core, cfg *config.Config) (Node, error) {
+
+	head, err := head.New(core)
+	if err != nil {
+		return nil, fmt.Errorf("could not create a head node: %w", err)
+	}
+
+	return head, nil
+}

consensus/pbft/execute.go

@@ -110,11 +110,11 @@ func (r *Replica) execute(ctx context.Context, view uint, sequence uint, digest
 		return fmt.Errorf("could not sign execution result: %w", err)
 	}
 
-	msg := response.Execute{
+	msg := response.WorkOrder{
 		BaseMessage: blockless.BaseMessage{TraceInfo: r.cfg.TraceInfo},
 		Code:        res.Code,
 		RequestID:   request.ID,
-		Results:     execute.ResultMap{r.id: nres},
+		Result:      nres,
 	}
 
 	// Save this executions in case it's requested again.

consensus/pbft/state.go

@@ -33,7 +33,7 @@ type replicaState struct {
 	viewChanges map[uint]*viewChangeReceipts
 
 	// Keep track of past executions. Results are mapped to request IDs, not digests.
-	executions map[string]response.Execute
+	executions map[string]response.WorkOrder
 }
 
 func newState() replicaState {
@@ -48,7 +48,7 @@ func newState() replicaState {
 		prepares:    make(map[messageID]*prepareReceipts),
 		commits:     make(map[messageID]*commitReceipts),
 		viewChanges: make(map[uint]*viewChangeReceipts),
-		executions:  make(map[string]response.Execute),
+		executions:  make(map[string]response.WorkOrder),
 	}
 
 	return state

fstore/http_internal_test.go

@@ -5,13 +5,12 @@ import (
 	"crypto/sha256"
 	"encoding/json"
 	"fmt"
-	"math/rand"
+	"math/rand/v2"
 	"net/http"
 	"net/http/httptest"
 	"os"
 	"strings"
 	"testing"
-	"time"
 
 	"github.com/stretchr/testify/require"
 
@@ -269,10 +268,10 @@ func TestFunction_DownloadHandlesErrors(t *testing.T) {
 func getRandomPayload(t *testing.T, len int) []byte {
 	t.Helper()
 
-	rand.Seed(time.Now().UnixNano())
-
+	var seed = [32]byte([]byte("ABCDEFGHIJKLMNOPQRSTUVWXYZ123456"))
+	r := rand.NewChaCha8(seed)
 	buf := make([]byte, len)
-	_, err := rand.Read(buf)
+	_, err := r.Read(buf)
 	require.NoError(t, err)
 
 	return buf