monitor changes in substrate and relay urls and update rmb peer accordingly

Author: Eslam-Nawara

cmds/modules/api_gateway/main.go

@@ -5,8 +5,11 @@ import (
 	"crypto/ed25519"
 	"encoding/hex"
 	"fmt"
+	"slices"
+	"time"
 
 	"github.com/cenkalti/backoff/v3"
+	"github.com/pkg/errors"
 	"github.com/rs/zerolog/log"
 	substrate "github.com/threefoldtech/tfchain/clients/tfchain-client-go"
 	"github.com/threefoldtech/tfgrid-sdk-go/rmb-sdk-go/peer"
@@ -63,6 +66,8 @@ func action(cli *cli.Context) error {
 		return err
 	}
 
+	subURLs := environment.MustGet().SubstrateURL
+	relayURLs := environment.GetRelaysURLs()
 	manager, err := environment.GetSubstrate()
 	if err != nil {
 		return fmt.Errorf("failed to create substrate manager: %w", err)
@@ -92,6 +97,7 @@ func action(cli *cli.Context) error {
 		}
 	}()
 
+	// no need to restart zos-api here as it only tries to get farm and twin, donesn't mentain any open connections
 	api, err := zosapi.NewZosAPI(manager, redis, msgBrokerCon)
 	if err != nil {
 		return fmt.Errorf("failed to create zos api: %w", err)
@@ -105,14 +111,16 @@ func action(cli *cli.Context) error {
 
 	bo := backoff.NewExponentialBackOff()
 	bo.MaxElapsedTime = 0
+
+	peerCtx, cancel := context.WithCancel(ctx)
 	backoff.Retry(func() error {
 		_, err = peer.NewPeer(
-			ctx,
+			peerCtx,
 			hex.EncodeToString(pair.Seed()),
 			manager,
 			router.Serve,
 			peer.WithKeyType(peer.KeyTypeEd25519),
-			peer.WithRelay(environment.GetRelaysURLs()...),
+			peer.WithRelay(relayURLs...),
 			peer.WithInMemoryExpiration(6*60*60), // 6 hours
 		)
 		if err != nil {
@@ -127,7 +135,79 @@ func action(cli *cli.Context) error {
 		Uint("worker nr", workerNr).
 		Msg("starting api-gateway module")
 
+	updatePeer := func(ctx context.Context, updatedSubURLs, updatedRelayURLs []string) (substrate.Manager, error) {
+		var newManager substrate.Manager
+
+		if !slices.Equal(subURLs, updatedSubURLs) {
+			newManager, err = environment.GetSubstrate()
+			if err != nil {
+				return nil, errors.Wrapf(err, "failed to create substrate manager")
+			}
+			err = gw.UpdateSubstrateGatewayConnection(manager)
+			if err != nil {
+				return nil, errors.Wrapf(err, "failed to update substrate gateway with new manager")
+			}
+
+		}
+
+		_, err = peer.NewPeer(
+			peerCtx,
+			hex.EncodeToString(pair.Seed()),
+			newManager,
+			router.Serve,
+			peer.WithKeyType(peer.KeyTypeEd25519),
+			peer.WithRelay(relayURLs...),
+			peer.WithInMemoryExpiration(6*60*60), // 6 hours
+		)
+		if err != nil {
+			errors.Wrapf(err, "failed to start a new rmb peer")
+		}
+		return newManager, nil
+	}
+
 	// block forever
-	<-ctx.Done()
-	return nil
+	for {
+		select {
+		case <-ctx.Done():
+			return nil
+		case <-peerCtx.Done():
+			return nil
+		case <-time.After(10 * time.Minute):
+			env, err := environment.Get()
+			// skip update if any problem causing the update to fail
+			if err != nil {
+				log.Debug().Err(err).Msg("failed to load node environment")
+				continue
+			}
+
+			updatedSubURLs := env.SubstrateURL
+			updatedRelayURLs := environment.GetRelaysURLs()
+
+			// continue if the urls did not change
+			if slices.Equal(subURLs, updatedSubURLs) || !slices.Equal(relayURLs, updatedRelayURLs) {
+				continue
+			}
+
+			newPeerCtx, newCancel := context.WithCancel(ctx)
+
+			newManager, err := updatePeer(newPeerCtx, updatedSubURLs, updatedRelayURLs)
+			if err != nil {
+				newCancel()
+				log.Debug().Err(err).Send()
+				continue
+			}
+
+			// only update urls and cancel the context after peer is created successfully
+			cancel()
+
+			peerCtx = newPeerCtx
+			cancel = newCancel
+
+			subURLs = updatedSubURLs
+			relayURLs = updatedRelayURLs
+
+			manager = newManager
+			log.Debug().Strs("relays_urls", updatedRelayURLs).Strs("substrate_urls", updatedSubURLs).Msg("updating substrate and relay urls")
+		}
+	}
 }