fix not persisting instanceID (#4526)

We are failing to persist instanceID today when initializing the repo as
`system_metadta.yaml` doesn't exist yet and we call the update method
instead of updateOrCreate.

Also this PR sends the instanceID to the version update checker instead
of clientID. No immediate change is needed in the update server as it
doesn't check for the clientID

---------

Co-authored-by: Forrest <[email protected]>

cmd/util/version.go

@@ -9,7 +9,6 @@ import (
 	"github.com/rs/zerolog/log"
 
 	"github.com/bacalhau-project/bacalhau/pkg/config/types"
-	baccrypto "github.com/bacalhau-project/bacalhau/pkg/lib/crypto"
 	"github.com/bacalhau-project/bacalhau/pkg/models"
 	clientv2 "github.com/bacalhau-project/bacalhau/pkg/publicapi/client/v2"
 	"github.com/bacalhau-project/bacalhau/pkg/repo"
@@ -41,20 +40,11 @@ func GetAllVersions(ctx context.Context, cfg types.Bacalhau, api clientv2.API, r
 		GOARCH:     resp.GOARCH,
 	}
 
-	userKeyPath, err := cfg.UserKeyPath()
-	if err != nil {
-		return versions, err
-	}
-	userKey, err := baccrypto.LoadUserKey(userKeyPath)
-	if err != nil {
-		return versions, fmt.Errorf("loading user key: %w", err)
-	}
-
 	updateCheck, err := version.CheckForUpdate(
 		ctx,
 		versions.ClientVersion,
 		versions.ServerVersion,
-		userKey.ClientID(),
+		r.InstanceID(),
 	)
 	if err != nil {
 		return versions, errors.Wrap(err, "failed to get latest version")

pkg/node/requester.go

@@ -212,18 +212,12 @@ func NewRequesterNode(
 		translationProvider = translation.NewStandardTranslatorsProvider()
 	}
 
-	installationID := system.InstallationID()
-	var instanceID string
-	if sysmeta, err := fsr.SystemMetadata(); err == nil {
-		instanceID = sysmeta.InstanceID
-	}
-
 	jobTransformers := transformer.ChainedTransformer[*models.Job]{
 		transformer.JobFn(transformer.IDGenerator),
 		transformer.NameOptional(),
 		transformer.RequesterInfo(nodeID),
-		transformer.OrchestratorInstallationID(installationID),
-		transformer.OrchestratorInstanceID(instanceID),
+		transformer.OrchestratorInstallationID(system.InstallationID()),
+		transformer.OrchestratorInstanceID(fsr.InstanceID()),
 		transformer.DefaultsApplier(requesterConfig.JobDefaults),
 	}
 

pkg/repo/fs.go

@@ -98,13 +98,12 @@ func (fsr *FsRepo) Init() error {
 	// TODO this should be a part of the config.
 	telemetry.SetupFromEnvs()
 
-	// never fail here as this isn't critical to node start up.
-	if err := fsr.writeInstanceID(GenerateInstanceID()); err != nil {
-		log.Trace().Err(err).Msgf("failed to write instanceID")
-	}
-
-	if err := fsr.WriteVersion(Version4); err != nil {
-		return fmt.Errorf("failed to persist repo version: %w", err)
+	// initialize repo's system metadata
+	if err := fsr.writeMetadata(&SystemMetadata{
+		RepoVersion: Version4,
+		InstanceID:  GenerateInstanceID(),
+	}); err != nil {
+		return fmt.Errorf("failed to persist system metadata: %w", err)
 	}
 	if err := fsr.WriteLegacyVersion(Version4); err != nil {
 		return fmt.Errorf("failed to persist legacy repo version: %w", err)
@@ -130,12 +129,10 @@ func (fsr *FsRepo) Open() error {
 
 	// check if an instanceID exists persisting one if not found.
 	// never fail here as this isn't critical to node start up.
-	if instanceID, err := fsr.readInstanceID(); err != nil {
-		log.Trace().Err(err).Msgf("failed to read instanceID")
-	} else if instanceID == "" {
+	if instanceID := fsr.InstanceID(); instanceID == "" {
 		// this case will happen when a user migrated from a repo prior to instanceID existing.
 		if err := fsr.writeInstanceID(GenerateInstanceID()); err != nil {
-			log.Trace().Err(err).Msgf("failed to write instanceID")
+			log.Debug().Err(err).Msgf("failed to write instanceID")
 		}
 	}
 

pkg/repo/sysmeta.go

@@ -6,6 +6,7 @@ import (
 	"path/filepath"
 	"time"
 
+	"github.com/rs/zerolog/log"
 	"gopkg.in/yaml.v3"
 
 	"github.com/bacalhau-project/bacalhau/pkg/storage/util"
@@ -110,9 +111,20 @@ func (fsr *FsRepo) WriteLastUpdateCheck(lastUpdateCheck time.Time) error {
 	})
 }
 
-// readInstanceID reads the InstanceID in the metadata.
+// InstanceID reads the InstanceID in the metadata.
+// It returns empty string if it fails to read the metadata.
+func (fsr *FsRepo) InstanceID() string {
+	instanceID, err := fsr.MustInstanceID()
+	if err != nil {
+		log.Debug().Err(err).Msg("failed to read instanceID")
+		return ""
+	}
+	return instanceID
+}
+
+// MustInstanceID reads the InstanceID in the metadata.
 // It fails if the metadata file doesn't exist.
-func (fsr *FsRepo) readInstanceID() (string, error) {
+func (fsr *FsRepo) MustInstanceID() (string, error) {
 	sysmeta, err := fsr.readMetadata()
 	if err != nil {
 		return "", err

pkg/version/update.go

@@ -14,7 +14,6 @@ import (
 	"github.com/rs/zerolog/log"
 
 	"github.com/bacalhau-project/bacalhau/pkg/config/types"
-	baccrypto "github.com/bacalhau-project/bacalhau/pkg/lib/crypto"
 	"github.com/bacalhau-project/bacalhau/pkg/logger"
 	"github.com/bacalhau-project/bacalhau/pkg/models"
 	"github.com/bacalhau-project/bacalhau/pkg/system"
@@ -43,7 +42,7 @@ type UpdateCheckResponse struct {
 func CheckForUpdate(
 	ctx context.Context,
 	currentClientVersion, currentServerVersion *models.BuildVersionInfo,
-	clientID string,
+	instanceID string,
 ) (*UpdateCheckResponse, error) {
 	u, err := url.Parse(getUpdateCheckerEndpoint())
 	if err != nil {
@@ -59,7 +58,9 @@ func CheckForUpdate(
 	if currentServerVersion != nil && currentServerVersion.GitVersion != "" {
 		q.Set("serverVersion", currentServerVersion.GitVersion)
 	}
-	q.Set("clientID", clientID)
+	if instanceID != "" {
+		q.Set("instanceID", instanceID)
+	}
 	if installationID := system.InstallationID(); installationID != "" {
 		q.Set("InstallationID", installationID)
 	}
@@ -107,6 +108,7 @@ func LogUpdateResponse(ctx context.Context, ucr *UpdateCheckResponse) {
 type UpdateStore interface {
 	ReadLastUpdateCheck() (time.Time, error)
 	WriteLastUpdateCheck(time.Time) error
+	InstanceID() string
 }
 
 // RunUpdateChecker starts a goroutine that will periodically make an update
@@ -128,16 +130,7 @@ func RunUpdateChecker(
 	}
 
 	clientVersion := Get()
-	userKeyPath, err := cfg.UserKeyPath()
-	if err != nil {
-		log.Ctx(ctx).Error().Err(err).Msg("Failed to load user key path")
-		return
-	}
-	userKey, err := baccrypto.LoadUserKey(userKeyPath)
-	if err != nil {
-		log.Ctx(ctx).Error().Err(err).Msg("Failed to load user key file")
-		return
-	}
+	instanceID := store.InstanceID()
 
 	runUpdateCheck := func() {
 		// The server may update itself between checks, so always ask the server
@@ -151,7 +144,7 @@ func RunUpdateChecker(
 			serverVersion = nil
 		}
 
-		updateResponse, err := CheckForUpdate(ctx, clientVersion, serverVersion, userKey.ClientID())
+		updateResponse, err := CheckForUpdate(ctx, clientVersion, serverVersion, instanceID)
 		if err != nil {
 			log.Ctx(ctx).Error().Err(err).Msg("Failed to perform update check")
 		}