Merge remote-tracking branch 'upstream/develop' into develop

Author: 0pcom

cmd/dmsg/commands/root.go

@@ -8,6 +8,8 @@ import (
 	"path/filepath"
 	"strings"
 
+	"github.com/skycoin/skywire/pkg/skywire-utilities/pkg/buildinfo"
+	"github.com/skycoin/skywire/pkg/skywire-utilities/pkg/calvin"
 	"github.com/spf13/cobra"
 
 	df "github.com/skycoin/dmsg/cmd/conf/commands"
@@ -24,6 +26,11 @@ import (
 	dw "github.com/skycoin/dmsg/cmd/dmsgweb/commands"
 )
 
+var (
+	bv  bool
+	dbi bool
+)
+
 func init() {
 	dmsgptyCmd.AddCommand(
 		dpc.RootCmd,
@@ -57,6 +64,25 @@ func init() {
 	dph.RootCmd.Use = "host"
 	dpu.RootCmd.Use = "ui"
 	di.RootCmd.Use = "ip"
+
+	modifySubcommands(RootCmd)
+	if fmt.Sprintf("%v", buildinfo.DebugBuildInfo()) != "" {
+		RootCmd.Flags().BoolVarP(&dbi, "info", "d", false, "print runtime/debug.BuildInfo")
+	}
+	if fmt.Sprintf("%v", buildinfo.DBIVersion()) != "" {
+		RootCmd.Flags().BoolVarP(&bv, "bv", "b", false, "print runtime/debug.BuildInfo.Main.Version")
+	}
+}
+
+func modifySubcommands(cmd *cobra.Command) {
+	for i := range cmd.Commands() {
+		cmd.Commands()[i].Version = ""
+		cmd.Commands()[i].SilenceErrors = true
+		cmd.Commands()[i].SilenceUsage = true
+		cmd.Commands()[i].DisableSuggestions = true
+		cmd.Commands()[i].DisableFlagsInUseLine = true
+		modifySubcommands(cmd.Commands()[i]) // recursion
+	}
 }
 
 // RootCmd contains all binaries which may be separately compiled as subcommands
@@ -65,11 +91,18 @@ var RootCmd = &cobra.Command{
 		return strings.Split(filepath.Base(strings.ReplaceAll(strings.ReplaceAll(fmt.Sprintf("%v", os.Args), "[", ""), "]", "")), " ")[0]
 	}(),
 	Short: "DMSG services & utilities",
-	Long: `
-	┌┬┐┌┬┐┌─┐┌─┐
-	 │││││└─┐│ ┬
-	─┴┘┴ ┴└─┘└─┘
-DMSG services & utilities`,
+	Long: func() (ret string) {
+		ret = calvin.AsciiFont("dmsg")
+		if buildinfo.DBIVersion() != "" {
+			ret += fmt.Sprintf("\n%v", buildinfo.DBIVersion())
+		} else {
+			ret += fmt.Sprintf("\nversion %v", buildinfo.Version())
+		}
+		if buildinfo.Go() != "unknown" && buildinfo.Go() != "" {
+			ret += "\nbuilt with " + buildinfo.Go()
+		}
+		return ret
+	}(),
 	SilenceErrors:         true,
 	SilenceUsage:          true,
 	DisableSuggestions:    true,

internal/cli/cli.go

@@ -7,6 +7,7 @@ import (
 	"io"
 	"log"
 	"net/http"
+	"time"
 
 	"github.com/skycoin/skywire/pkg/skywire-utilities/pkg/cipher"
 	"github.com/skycoin/skywire/pkg/skywire-utilities/pkg/logging"
@@ -61,7 +62,9 @@ func InitDmsgWithFlags(ctx context.Context, dlog *logging.Logger, pk cipher.PubK
 			dlog.Infof("Received response from dmsg-discovery server %s/health:\n%s", flags.DmsgDiscURL, string(body))
 		}
 
-		return StartDmsg(ctx, dlog, pk, sk, httpClient, flags.DmsgDiscURL, flags.DmsgSessions)
+		// Use direct client with synthetic entries for discovery server and all dmsg servers
+		// This allows dialing the discovery server which doesn't register itself
+		return StartDmsgWithDirectClient(ctx, dlog, pk, sk, flags.DmsgSessions)
 	}
 
 	// Default dmsghttp mode
@@ -111,7 +114,61 @@ func InitDmsgWithFlags(ctx context.Context, dlog *logging.Logger, pk cipher.PubK
 		dlog.Infof("Received response from dmsg-discovery server %s/health:\n%s", flags.DmsgDiscAddr, string(body))
 	}
 
-	return StartDmsg(ctx, dlog, pk, sk, dmsgHTTP, flags.DmsgDiscAddr, flags.DmsgSessions)
+	return StartDmsgWithSyntheticDiscovery(ctx, dlog, pk, sk, dmsgHTTP, flags.DmsgDiscAddr, flags.DmsgSessions)
+}
+
+// StartDmsgWithSyntheticDiscovery starts dmsg with a synthetic discovery entry for the discovery server itself
+func StartDmsgWithSyntheticDiscovery(ctx context.Context, dlog *logging.Logger, pk cipher.PubKey, sk cipher.SecKey, httpClient *http.Client, dmsgDisc string, dmsgSessions int) (dmsgC *dmsg.Client, stop func(), err error) {
+	if dlog == nil {
+		return nil, nil, fmt.Errorf("nil logger")
+	}
+
+	// Create base discovery client
+	baseDiscClient := disc.NewHTTP(dmsgDisc, httpClient, dlog)
+
+	// Wrap with caching client that includes synthetic entry for discovery server
+	discPK := dmsg.ExtractPKFromDmsgAddr(dmsgDisc)
+	if discPK != "" {
+		var discoveryPK cipher.PubKey
+		if err := discoveryPK.UnmarshalText([]byte(discPK)); err == nil {
+			// Get all available dmsg servers as delegated servers
+			var delegatedServers []cipher.PubKey
+			for _, server := range dmsg.Prod.DmsgServers {
+				delegatedServers = append(delegatedServers, server.Static)
+			}
+			syntheticEntry := &disc.Entry{
+				Static: discoveryPK,
+				Client: &disc.Client{
+					DelegatedServers: delegatedServers,
+				},
+			}
+			baseDiscClient = newCachingDiscClient(baseDiscClient, syntheticEntry, dlog)
+			dlog.Debug("Created synthetic discovery entry for dialing")
+		}
+	}
+
+	dmsgC = dmsg.NewClient(pk, sk, baseDiscClient, &dmsg.Config{MinSessions: dmsgSessions})
+	dlog.Debug("Created dmsg client.")
+
+	go dmsgC.Serve(ctx)
+	dlog.Debug("dmsgclient.Serve(ctx)")
+
+	stop = func() {
+		err := dmsgC.Close()
+		dlog.WithError(err).Debug("Disconnected from dmsg network.\n")
+		log.Println()
+	}
+	dlog.WithField("dmsg_disc", dmsgDisc).Debug("Connecting to dmsg network...\n")
+	dlog.WithField("client public_key", pk.String()).Debug("\n")
+	select {
+	case <-ctx.Done():
+		stop()
+		return nil, nil, ctx.Err()
+
+	case <-dmsgC.Ready():
+		dlog.Debug("Dmsg network ready.")
+		return dmsgC, stop, nil
+	}
 }
 
 // StartDmsg starts dmsg returns a dmsg client for the given dmsg discovery
@@ -203,18 +260,159 @@ func StartDmsgDirectWithServers(ctx context.Context, dlog *logging.Logger, pk ci
 	}
 	if dmsgDiscAddr != "" {
 		// Validate that we can access discovery over DMSG
+		// Retry with exponential backoff to handle session initialization timing
 		dmsgHTTP := &http.Client{Transport: dmsghttp.MakeHTTPTransport(ctx, dmsgC)}
-		resp, err := dmsgHTTP.Get(dmsgDiscAddr + "/health")
+		var resp *http.Response
+		maxRetries := 5
+		for i := 0; i < maxRetries; i++ {
+			resp, err = dmsgHTTP.Get(dmsgDiscAddr + "/health")
+			if err == nil {
+				resp.Body.Close() //nolint
+				break
+			}
+			if i < maxRetries-1 {
+				backoff := time.Duration(200*(i+1)) * time.Millisecond
+				dlog.WithError(err).Debugf("Failed to reach discovery, retrying in %v (attempt %d/%d)", backoff, i+1, maxRetries)
+				time.Sleep(backoff)
+			}
+		}
 		if err != nil {
 			stop() // Cleanup if validation fails
 			return nil, nil, fmt.Errorf("failed to reach discovery server via DMSG: %w", err)
 		}
-		resp.Body.Close() //nolint
 	}
 
 	return dmsgC, stop, nil
 }
 
+// StartDmsgWithDirectClient starts dmsg with a direct client that includes synthetic entries
+// This allows dialing any client including the discovery server which doesn't register itself
+func StartDmsgWithDirectClient(ctx context.Context, dlog *logging.Logger, pk cipher.PubKey, sk cipher.SecKey, dmsgSessions int) (dmsgC *dmsg.Client, stop func(), err error) {
+	if dlog == nil {
+		return nil, nil, fmt.Errorf("nil logger")
+	}
+
+	// Build entries for all dmsg servers
+	var entries []*disc.Entry
+	for _, server := range dmsg.Prod.DmsgServers {
+		entries = append(entries, &server)
+	}
+
+	// Add synthetic entry for discovery server
+	discPK := dmsg.ExtractPKFromDmsgAddr(flags.DmsgDiscAddr)
+	if discPK != "" {
+		var discoveryPK cipher.PubKey
+		if err := discoveryPK.UnmarshalText([]byte(discPK)); err == nil {
+			var delegatedServers []cipher.PubKey
+			for _, server := range dmsg.Prod.DmsgServers {
+				delegatedServers = append(delegatedServers, server.Static)
+			}
+			discoveryEntry := &disc.Entry{
+				Static: discoveryPK,
+				Client: &disc.Client{
+					DelegatedServers: delegatedServers,
+				},
+			}
+			entries = append(entries, discoveryEntry)
+			dlog.Debug("Added synthetic discovery entry to direct client")
+		}
+	}
+
+	// Add synthetic entry for our own client
+	var delegatedServers []cipher.PubKey
+	for _, server := range dmsg.Prod.DmsgServers {
+		delegatedServers = append(delegatedServers, server.Static)
+	}
+	clientEntry := &disc.Entry{
+		Static: pk,
+		Client: &disc.Client{
+			DelegatedServers: delegatedServers,
+		},
+	}
+	entries = append(entries, clientEntry)
+
+	// Create direct client with all entries
+	directClient := direct.NewClient(entries, dlog)
+
+	dmsgC = dmsg.NewClient(pk, sk, directClient, &dmsg.Config{MinSessions: dmsgSessions})
+	dlog.Debug("Created dmsg client with direct client.")
+
+	go dmsgC.Serve(ctx)
+	dlog.Debug("dmsgclient.Serve(ctx)")
+
+	stop = func() {
+		err := dmsgC.Close()
+		dlog.WithError(err).Debug("Disconnected from dmsg network.\n")
+		log.Println()
+	}
+	dlog.Debug("Connecting to dmsg network...\n")
+	dlog.WithField("client public_key", pk.String()).Debug("\n")
+	select {
+	case <-ctx.Done():
+		stop()
+		return nil, nil, ctx.Err()
+
+	case <-dmsgC.Ready():
+		dlog.Debug("Dmsg network ready.")
+		return dmsgC, stop, nil
+	}
+}
+
+// cachingDiscClient wraps a discovery client and caches a synthetic entry
+type cachingDiscClient struct {
+	base           disc.APIClient
+	syntheticEntry *disc.Entry
+	log            *logging.Logger
+}
+
+// newCachingDiscClient creates a discovery client that caches a synthetic entry
+func newCachingDiscClient(base disc.APIClient, syntheticEntry *disc.Entry, log *logging.Logger) disc.APIClient {
+	return &cachingDiscClient{
+		base:           base,
+		syntheticEntry: syntheticEntry,
+		log:            log,
+	}
+}
+
+// Entry returns the synthetic entry if PK matches, otherwise queries base client
+func (c *cachingDiscClient) Entry(ctx context.Context, pk cipher.PubKey) (*disc.Entry, error) {
+	if c.syntheticEntry != nil && c.syntheticEntry.Static == pk {
+		c.log.WithField("pk", pk.String()).Debug("Returning synthetic discovery entry")
+		return c.syntheticEntry, nil
+	}
+	return c.base.Entry(ctx, pk)
+}
+
+// PostEntry delegates to base client
+func (c *cachingDiscClient) PostEntry(ctx context.Context, entry *disc.Entry) error {
+	return c.base.PostEntry(ctx, entry)
+}
+
+// PutEntry delegates to base client
+func (c *cachingDiscClient) PutEntry(ctx context.Context, sk cipher.SecKey, entry *disc.Entry) error {
+	return c.base.PutEntry(ctx, sk, entry)
+}
+
+// DelEntry delegates to base client
+func (c *cachingDiscClient) DelEntry(ctx context.Context, entry *disc.Entry) error {
+	return c.base.DelEntry(ctx, entry)
+}
+
+// AvailableServers delegates to base client
+func (c *cachingDiscClient) AvailableServers(ctx context.Context) ([]*disc.Entry, error) {
+	return c.base.AvailableServers(ctx)
+}
+
+// AllServers delegates to base client
+func (c *cachingDiscClient) AllServers(ctx context.Context) ([]*disc.Entry, error) {
+	return c.base.AllServers(ctx)
+}
+
+// AllEntries delegates to base client
+func (c *cachingDiscClient) AllEntries(ctx context.Context) ([]string, error) {
+	return c.base.AllEntries(ctx)
+}
+
 // FallbackRoundTripper tries multiple DMSG transports until one succeeds.
 type FallbackRoundTripper struct {
 	ctx     context.Context