Repository: bacalhau-project/bacalhau (language: python)

Issue:
Compute nodes 1.5 can't join 1.4 orchestrator
1.5 compute nodes are always in disconnected state

### Steps to reproduce
1. Start v1.4.0 orchestrator:
    
    ```bash
    # start with fresh data dir
    export BACALHAU_DIR=$(mktemp -d)
    bacalhau-1.4.0 serve --node-type requester
    
    ```
    
2. Start v1.5.0 compute node:
    
    ```bash
    export BACALHAU_DIR=$(mktemp -d)
    bacalhau-1.5.0 serve --compute -c api.port=1235
    ```
    
3. List nodes 
    
    ```bash
    export BACALHAU_DIR=~/bacalhau-versions/v1.4.0/data-dir
    bacalhau-1.4.0 --api-host 127.0.0.1 node list

     ID          TYPE       APPROVAL  STATUS        LABELS                                      CPU     MEMORY      DISK         GPU
     n-22b3497b  Requester  APPROVED                Architecture=arm64 Operating-System=darwin
     n-93469683  Compute    APPROVED  DISCONNECTED  Architecture=arm64 Operating-System=darwin  5.6 /   11.2 GB /   37.6 GB /    0 /
                                                                                            5.6     11.2 GB     37.6 GB      0

    ```
    
In addition to that, you can see errors in the orchestrator's logs:
```
10:05:25.961 | ERR pkg/nats/pubsub/pubsub.go:93 > error unmarshalling nats payload from subject heartbeat error="invalid character '\\x01' looking for beginning of value"
10:05:40.959 | ERR pkg/nats/pubsub/pubsub.go:93 > error unmarshalling nats payload from subject heartbeat error="invalid character '\\x01' looking for beginning of value"
```

This is related to using new `ncl` library to heartbeat in 1.5




Discussion / hints:

Fix patch:
diff --git a/pkg/compute/management_client.go b/pkg/compute/management_client.go
index 5d7ff378d8..c8ecbfb01b 100644
--- a/pkg/compute/management_client.go
+++ b/pkg/compute/management_client.go
@@ -158,6 +158,9 @@ func (m *ManagementClient) Start(ctx context.Context) {
 		heartbeatTicker.Stop()
 		resourceTicker.Stop()
 		infoTicker.Stop()
+
+		// Close the heartbeat client and it's resources
+		m.heartbeatClient.Close(ctx)
 	}()
 
 	// Send an initial heartbeat when we start up
diff --git a/pkg/node/compute.go b/pkg/node/compute.go
index 3f76582a67..d5ff16508d 100644
--- a/pkg/node/compute.go
+++ b/pkg/node/compute.go
@@ -208,13 +208,16 @@ func NewComputeNode(
 	// heartbeat client
 	heartbeatPublisher, err := ncl.NewPublisher(natsConn,
 		ncl.WithPublisherName(nodeID),
-		ncl.WithPublisherDestination(config.ControlPlaneSettings.HeartbeatTopic),
+		ncl.WithPublisherDestination(computeHeartbeatTopic(nodeID)),
 		ncl.WithPublisherMessageSerDeRegistry(messageSerDeRegistry),
 	)
 	if err != nil {
 		return nil, err
 	}
-	heartbeatClient := heartbeat.NewClient(nodeID, heartbeatPublisher)
+	heartbeatClient, err := heartbeat.NewClient(natsConn, nodeID, heartbeatPublisher)
+	if err != nil {
+		return nil, err
+	}
 
 	// Set up the management client which will attempt to register this node
 	// with the requester node, and then if successful will send regular node
diff --git a/pkg/node/config_compute.go b/pkg/node/config_compute.go
index 20eea4e70b..dac293f6dc 100644
--- a/pkg/node/config_compute.go
+++ b/pkg/node/config_compute.go
@@ -167,9 +167,6 @@ func NewComputeConfigWith(executionDir string, params ComputeConfigParams) (Comp
 	if params.ControlPlaneSettings.ResourceUpdateFrequency == 0 {
 		params.ControlPlaneSettings.ResourceUpdateFrequency = defaults.ControlPlaneSettings.ResourceUpdateFrequency
 	}
-	if params.ControlPlaneSettings.HeartbeatTopic == "" {
-		params.ControlPlaneSettings.HeartbeatTopic = defaults.ControlPlaneSettings.HeartbeatTopic
-	}
 
 	// Get available physical resources in the host
 	physicalResourcesProvider := params.PhysicalResourcesProvider
diff --git a/pkg/node/config_defaults.go b/pkg/node/config_defaults.go
index 213c889156..948e520531 100644
--- a/pkg/node/config_defaults.go
+++ b/pkg/node/config_defaults.go
@@ -37,7 +37,6 @@ func NewDefaultComputeParam(storagePath string) ComputeConfigParams {
 			InfoUpdateFrequency:     legacy_types.Duration(60 * time.Second), //nolint:gomnd
 			ResourceUpdateFrequency: legacy_types.Duration(30 * time.Second), //nolint:gomnd
 			HeartbeatFrequency:      legacy_types.Duration(15 * time.Second), //nolint:gomnd
-			HeartbeatTopic:          "heartbeat",
 		},
 	}
 }
@@ -71,7 +70,6 @@ var DefaultRequesterConfig = RequesterConfigParams{
 
 	ControlPlaneSettings: legacy_types.RequesterControlPlaneConfig{
 		HeartbeatCheckFrequency: legacy_types.Duration(30 * time.Second), //nolint:gomnd
-		HeartbeatTopic:          "heartbeat",
 		NodeDisconnectedAfter:   legacy_types.Duration(30 * time.Second), //nolint:gomnd
 	},
 
@@ -108,7 +106,6 @@ var TestRequesterConfig = RequesterConfigParams{
 
 	ControlPlaneSettings: legacy_types.RequesterControlPlaneConfig{
 		HeartbeatCheckFrequency: legacy_types.Duration(30 * time.Second), //nolint:gomnd
-		HeartbeatTopic:          "heartbeat",
 		NodeDisconnectedAfter:   legacy_types.Duration(30 * time.Second), //nolint:gomnd
 	},
 
diff --git a/pkg/node/heartbeat/client.go b/pkg/node/heartbeat/client.go
index 42bc1aff11..8ea039a708 100644
--- a/pkg/node/heartbeat/client.go
+++ b/pkg/node/heartbeat/client.go
@@ -2,22 +2,45 @@ package heartbeat
 
 import (
 	"context"
+	"errors"
+
+	"github.com/nats-io/nats.go"
 
 	"github.com/bacalhau-project/bacalhau/pkg/lib/ncl"
+	natsPubSub "github.com/bacalhau-project/bacalhau/pkg/nats/pubsub"
 )
 
 type HeartbeatClient struct {
-	publisher ncl.Publisher
-	nodeID    string
+	legacyPublisher *natsPubSub.PubSub[Heartbeat]
+	publisher       ncl.Publisher
+	nodeID          string
 }
 
-func NewClient(nodeID string, publisher ncl.Publisher) *HeartbeatClient {
-	return &HeartbeatClient{publisher: publisher, nodeID: nodeID}
+func NewClient(conn *nats.Conn, nodeID string, publisher ncl.Publisher) (*HeartbeatClient, error) {
+	legacyPublisher, err := natsPubSub.NewPubSub[Heartbeat](natsPubSub.PubSubParams{
+		Subject: legacyHeartbeatTopic,
+		Conn:    conn,
+	})
+	if err != nil {
+		return nil, err
+	}
+
+	return &HeartbeatClient{publisher: publisher, nodeID: nodeID, legacyPublisher: legacyPublisher}, nil
 }
 
 func (h *HeartbeatClient) SendHeartbeat(ctx context.Context, sequence uint64) error {
-	message := ncl.NewMessage(Heartbeat{NodeID: h.nodeID, Sequence: sequence})
-	return h.publisher.Publish(ctx, message)
+	heartbeat := Heartbeat{NodeID: h.nodeID, Sequence: sequence}
+
+	// Send the heartbeat to current and legacy topics
+	var err error
+	message := ncl.NewMessage(heartbeat)
+	err = errors.Join(err, h.publisher.Publish(ctx, message))
+	err = errors.Join(h.legacyPublisher.Publish(ctx, heartbeat))
+	return err
+}
+
+func (h *HeartbeatClient) Close(ctx context.Context) error {
+	return h.legacyPublisher.Close(ctx)
 }
 
 var _ Client = (*HeartbeatClient)(nil)
diff --git a/pkg/node/heartbeat/server.go b/pkg/node/heartbeat/server.go
index a0df076d47..73cdf40864 100644
--- a/pkg/node/heartbeat/server.go
+++ b/pkg/node/heartbeat/server.go
@@ -6,15 +6,19 @@ import (
 	"time"
 
 	"github.com/benbjohnson/clock"
+	"github.com/nats-io/nats.go"
 	"github.com/rs/zerolog/log"
 
 	"github.com/bacalhau-project/bacalhau/pkg/lib/collections"
 	"github.com/bacalhau-project/bacalhau/pkg/lib/concurrency"
 	"github.com/bacalhau-project/bacalhau/pkg/lib/ncl"
 	"github.com/bacalhau-project/bacalhau/pkg/models"
+	natsPubSub "github.com/bacalhau-project/bacalhau/pkg/nats/pubsub"
+	"github.com/bacalhau-project/bacalhau/pkg/pubsub"
 )
 
 type HeartbeatServerParams struct {
+	Client                *nats.Conn
 	Clock                 clock.Clock
 	CheckFrequency        time.Duration
 	NodeDisconnectedAfter time.Duration
@@ -22,6 +26,7 @@ type HeartbeatServerParams struct {
 
 type HeartbeatServer struct {
 	clock             clock.Clock
+	legacySubscriber  *natsPubSub.PubSub[Heartbeat]
 	pqueue            *collections.HashedPriorityQueue[string, TimestampedHeartbeat]
 	livenessMap       *concurrency.StripedMap[models.NodeConnectionState]
 	checkFrequency    time.Duration
@@ -34,6 +39,14 @@ type TimestampedHeartbeat struct {
 }
 
 func NewServer(params HeartbeatServerParams) (*HeartbeatServer, error) {
+	legacySubscriber, err := natsPubSub.NewPubSub[Heartbeat](natsPubSub.PubSubParams{
+		Subject: legacyHeartbeatTopic,
+		Conn:    params.Client,
+	})
+	if err != nil {
+		return nil, err
+	}
+
 	pqueue := collections.NewHashedPriorityQueue[string, TimestampedHeartbeat](
 		func(h TimestampedHeartbeat) string {
 			return h.NodeID
@@ -48,6 +61,7 @@ func NewServer(params HeartbeatServerParams) (*HeartbeatServer, error) {
 
 	return &HeartbeatServer{
 		clock:             clk,
+		legacySubscriber:  legacySubscriber,
 		pqueue:            pqueue,
 		livenessMap:       concurrency.NewStripedMap[models.NodeConnectionState](0), // no particular stripe count for now
 		checkFrequency:    params.CheckFrequency,
@@ -56,11 +70,21 @@ func NewServer(params HeartbeatServerParams) (*HeartbeatServer, error) {
 }
 
 func (h *HeartbeatServer) Start(ctx context.Context) error {
-	log.Ctx(ctx).Info().Msg("Heartbeat server started")
+	if err := h.legacySubscriber.Subscribe(ctx, h); err != nil {
+		return err
+	}
 
 	tickerStartCh := make(chan struct{})
 
 	go func(ctx context.Context) {
+		defer func() {
+			if err := h.legacySubscriber.Close(ctx); err != nil {
+				log.Ctx(ctx).Error().Err(err).Msg("Error during heartbeat server shutdown")
+			} else {
+				log.Ctx(ctx).Info().Msg("Heartbeat server shutdown")
+			}
+		}()
+
 		ticker := h.clock.Ticker(h.checkFrequency)
 		tickerStartCh <- struct{}{}
 		defer ticker.Stop()
@@ -77,6 +101,7 @@ func (h *HeartbeatServer) Start(ctx context.Context) error {
 
 	// Wait for the ticker to be created before returning
 	<-tickerStartCh
+	log.Ctx(ctx).Info().Msg("Heartbeat server started")
 
 	return nil
 }
@@ -146,12 +171,8 @@ func (h *HeartbeatServer) ShouldProcess(ctx context.Context, message *ncl.Messag
 	return message.Metadata.Get(ncl.KeyMessageType) == HeartbeatMessageType
 }
 
-func (h *HeartbeatServer) HandleMessage(ctx context.Context, message *ncl.Message) error {
-	heartbeat, ok := message.Payload.(*Heartbeat)
-	if !ok {
-		return ncl.NewErrUnexpectedPayloadType(
-			reflect.TypeOf(Heartbeat{}).String(), reflect.TypeOf(message.Payload).String())
-	}
+// Handle will handle a message received through the legacy heartbeat topic
+func (h *HeartbeatServer) Handle(ctx context.Context, heartbeat Heartbeat) error {
 	log.Ctx(ctx).Trace().Msgf("heartbeat received from %s", heartbeat.NodeID)
 
 	timestamp := h.clock.Now().UTC().Unix()
@@ -177,7 +198,7 @@ func (h *HeartbeatServer) HandleMessage(ctx context.Context, message *ncl.Messag
 
 		// We'll enqueue the heartbeat message with the current timestamp. The older
 		// the entry, the lower the timestamp (trending to 0) and the higher the priority.
-		h.pqueue.Enqueue(TimestampedHeartbeat{Heartbeat: *heartbeat, Timestamp: timestamp}, timestamp)
+		h.pqueue.Enqueue(TimestampedHeartbeat{Heartbeat: heartbeat, Timestamp: timestamp}, timestamp)
 	}
 
 	h.markNodeAs(heartbeat.NodeID, models.NodeStates.HEALTHY)
@@ -185,4 +206,15 @@ func (h *HeartbeatServer) HandleMessage(ctx context.Context, message *ncl.Messag
 	return nil
 }
 
+// HandleMessage will handle a message received through ncl and will call the Handle method
+func (h *HeartbeatServer) HandleMessage(ctx context.Context, message *ncl.Message) error {
+	heartbeat, ok := message.Payload.(*Heartbeat)
+	if !ok {
+		return ncl.NewErrUnexpectedPayloadType(
+			reflect.TypeOf(Heartbeat{}).String(), reflect.TypeOf(message.Payload).String())
+	}
+	return h.Handle(ctx, *heartbeat)
+}
+
 var _ ncl.MessageHandler = (*HeartbeatServer)(nil)
+var _ pubsub.Subscriber[Heartbeat] = (*HeartbeatServer)(nil)
diff --git a/pkg/node/heartbeat/types.go b/pkg/node/heartbeat/types.go
index 20b54b4628..9bcdc2833c 100644
--- a/pkg/node/heartbeat/types.go
+++ b/pkg/node/heartbeat/types.go
@@ -9,6 +9,11 @@ const (
 	HeartbeatMessageType = "heartbeat"
 )
 
+//	legacyHeartbeatTopic is the topic where the heartbeat messages are sent prior to v1.5.
+//
+// TODO: Remove this legacy heartbeat topic with v1.6
+const legacyHeartbeatTopic = "heartbeat"
+
 // Heartbeat represents a heartbeat message from a specific node.
 // It contains the node ID and the sequence number of the heartbeat
 // which is monotonically increasing (reboots aside). We do not
@@ -21,4 +26,5 @@ type Heartbeat struct {
 
 type Client interface {
 	SendHeartbeat(ctx context.Context, sequence uint64) error
+	Close(ctx context.Context) error
 }
diff --git a/pkg/node/ncl.go b/pkg/node/ncl.go
index b23409af98..03635d812f 100644
--- a/pkg/node/ncl.go
+++ b/pkg/node/ncl.go
@@ -2,6 +2,7 @@ package node
 
 import (
 	"errors"
+	"fmt"
 
 	"github.com/bacalhau-project/bacalhau/pkg/lib/ncl"
 	"github.com/bacalhau-project/bacalhau/pkg/node/heartbeat"
@@ -15,3 +16,17 @@ func CreateMessageSerDeRegistry() (*ncl.MessageSerDeRegistry, error) {
 	)
 	return reg, err
 }
+const HeartbeatTopicFormat = "bacalhau.global.compute.%s.out.heartbeat"
+
+// computeHeartbeatTopic returns the subject to publish heartbeat messages to.
+// it publishes to the outgoing heartbeat subject of a specific compute node, which
+// the orchestrator subscribes to.
+func computeHeartbeatTopic(nodeID string) string {
+	return fmt.Sprintf(HeartbeatTopicFormat, nodeID)
+}
+
+// orchestratorHeartbeatSubscription returns the subject to subscribe for compute heartbeats.
+// it subscribes for heartbeat messages from all compute nodes
+func orchestratorHeartbeatSubscription() string {
+	return fmt.Sprintf(HeartbeatTopicFormat, "*")
+}
diff --git a/pkg/node/requester.go b/pkg/node/requester.go
index 1e8515f223..06703dc705 100644
--- a/pkg/node/requester.go
+++ b/pkg/node/requester.go
@@ -283,7 +283,7 @@ func NewRequesterNode(
 	if err != nil {
 		return nil, pkgerrors.Wrap(err, "failed to create ncl subscriber")
 	}
-	err = subscriber.Subscribe(requesterConfig.ControlPlaneSettings.HeartbeatTopic)
+	err = subscriber.Subscribe(orchestratorHeartbeatSubscription())
 	if err != nil {
 		return nil, err
 	}
@@ -378,6 +378,7 @@ func createNodeManager(ctx context.Context,
 
 	// heartbeat service
 	heartbeatParams := heartbeat.HeartbeatServerParams{
+		Client:                transportLayer.Client(),
 		CheckFrequency:        requesterConfig.ControlPlaneSettings.HeartbeatCheckFrequency.AsTimeDuration(),
 		NodeDisconnectedAfter: requesterConfig.ControlPlaneSettings.NodeDisconnectedAfter.AsTimeDuration(),
 	}


Test patch:
diff --git a/pkg/node/heartbeat/heartbeat_test.go b/pkg/node/heartbeat/heartbeat_test.go
index 2080efe59b..edc7038d93 100644
--- a/pkg/node/heartbeat/heartbeat_test.go
+++ b/pkg/node/heartbeat/heartbeat_test.go
@@ -41,18 +41,19 @@ func (s *HeartbeatTestSuite) SetupTest() {
 	var err error
 	s.clock = clock.NewMock()
 
+	// Setup NATS server and client
+	s.natsServer, s.natsConn = testutils.StartNats(s.T())
+
 	// Setup heartbeat server
 	s.heartbeatServer, err = NewServer(HeartbeatServerParams{
 		Clock:                 s.clock,
+		Client:                s.natsConn,
 		CheckFrequency:        1 * time.Second,
 		NodeDisconnectedAfter: 10 * time.Second,
 	})
 	s.Require().NoError(err)
 	s.Require().NoError(s.heartbeatServer.Start(context.Background()))
 
-	// setup nats server and client
-	s.natsServer, s.natsConn = testutils.StartNats(s.T())
-
 	// Setup NATS publisher and subscriber
 	s.messageSerDeRegistry = ncl.NewMessageSerDeRegistry()
 	s.Require().NoError(s.messageSerDeRegistry.Register(HeartbeatMessageType, Heartbeat{}))
@@ -70,7 +71,7 @@ func (s *HeartbeatTestSuite) SetupTest() {
 	)
 	s.Require().NoError(err)
 	s.Require().NoError(s.subscriber.Subscribe(TestTopic))
-
+	s.Require().NoError(s.subscriber.Subscribe(legacyHeartbeatTopic))
 }
 
 func (s *HeartbeatTestSuite) TearDownTest() {
@@ -123,7 +124,8 @@ func (s *HeartbeatTestSuite) TestHeartbeatScenarios() {
 
 	for i, tc := range testcases {
 		nodeID := "node-" + strconv.Itoa(i)
-		client := NewClient(nodeID, s.publisher)
+		client, err := NewClient(s.natsConn, nodeID, s.publisher)
+		s.Require().NoError(err)
 
 		s.T().Run(tc.name, func(t *testing.T) {
 			var seq uint64 = 1
@@ -153,11 +155,12 @@ func (s *HeartbeatTestSuite) TestHeartbeatScenarios() {
 
 func (s *HeartbeatTestSuite) TestSendHeartbeatError() {
 	ctx := context.Background()
-	client := NewClient("test-node", s.publisher)
+	client, err := NewClient(s.natsConn, "test-node", s.publisher)
+	s.Require().NoError(err)
 
 	// Close the NATS connection to force an error
 	s.natsConn.Close()
 
-	err := client.SendHeartbeat(ctx, 1)
+	err = client.SendHeartbeat(ctx, 1)
 	s.Error(err)
 }