Cache GraphQL client responses in local cache file

agent/client.go

@@ -26,6 +26,7 @@ import (
 	"github.com/superfly/flyctl/internal/flyutil"
 	"github.com/superfly/flyctl/internal/logger"
 	"github.com/superfly/flyctl/internal/sentry"
+	"github.com/superfly/flyctl/internal/task"
 	"github.com/superfly/flyctl/internal/version"
 	"github.com/superfly/flyctl/internal/wireguard"
 	"github.com/superfly/flyctl/iostreams"
@@ -35,9 +36,18 @@ import (
 
 // Establish starts the daemon, if necessary, and returns a client to it.
 func Establish(ctx context.Context, apiClient wireguard.WebClient) (*Client, error) {
-	if err := wireguard.PruneInvalidPeers(ctx, apiClient); err != nil {
-		return nil, err
-	}
+	logger := logger.MaybeFromContext(ctx)
+
+	task.FromContext(ctx).Run(func(taskCtx context.Context) {
+		if err := wireguard.PruneInvalidPeers(taskCtx, apiClient); err != nil {
+			msg := fmt.Sprintf("failed to prune wireguard peers: %s", err)
+			if logger != nil {
+				logger.Debug(msg)
+			} else {
+				fmt.Fprintln(os.Stderr, msg)
+			}
+		}
+	})
 
 	c := newClient("unix", PathToSocket())
 
@@ -58,7 +68,6 @@ func Establish(ctx context.Context, apiClient wireguard.WebClient) (*Client, err
 	// TOOD: log this instead
 	msg := fmt.Sprintf("The running flyctl agent (v%s) is older than the current flyctl (v%s).", res.Version, buildinfo.Version())
 
-	logger := logger.MaybeFromContext(ctx)
 	if logger != nil {
 		logger.Warn(msg)
 	} else {

internal/cache/cache.go

@@ -60,6 +60,8 @@ type Cache interface {
 	// Save writes the YAML-encoded representation of c to the named file path via
 	// os.WriteFile.
 	Save(path string) error
+
+	Fetch(key string, expiresIn time.Duration, refreshIn time.Duration, fetchFn func() (any, error)) (any, error)
 }
 
 const defaultChannel = "latest"
@@ -78,6 +80,44 @@ type cache struct {
 	lastCheckedAt time.Time
 	latestRelease *update.Release
 	invalidVer    *invalidVer
+	objectCache   map[string]*CachedObject
+}
+
+func (c *cache) Fetch(key string, expiresIn time.Duration, refreshIn time.Duration, fetchFn func() (any, error)) (any, error) {
+	c.mu.RLock()
+	defer c.mu.RUnlock()
+
+	fetch := func() (any, error) {
+		obj, err := fetchFn()
+		if err != nil {
+			return nil, err
+		}
+		c.objectCache[key] = &CachedObject{
+			ExpiresAt: time.Now().Add(expiresIn),
+			RefreshAt: time.Now().Add(refreshIn),
+			Object:    obj,
+		}
+		c.dirty = true
+		return obj, nil
+	}
+	if c.objectCache == nil {
+		c.objectCache = map[string]*CachedObject{}
+	}
+
+	if obj := c.objectCache[key]; obj != nil && obj.ExpiresAt.After(time.Now()) {
+		if obj.RefreshAt.Before(time.Now()) {
+			go func() { _, _ = fetch() }()
+		}
+		return obj.Object, nil
+	} else {
+		return fetch()
+	}
+}
+
+type CachedObject struct {
+	ExpiresAt time.Time
+	RefreshAt time.Time
+	Object    any
 }
 
 func (c *cache) Channel() string {
@@ -170,6 +210,7 @@ type wrapper struct {
 	LastCheckedAt time.Time       `yaml:"last_checked_at,omitempty"`
 	LatestRelease *update.Release `yaml:"latest_release,omitempty"`
 	InvalidVer    *invalidVer
+	ObjectCache   map[string]*CachedObject `yaml:"object_cache,omitempty"`
 }
 
 func lockPath() string {
@@ -189,6 +230,7 @@ func (c *cache) Save(path string) (err error) {
 		LastCheckedAt: c.lastCheckedAt,
 		LatestRelease: c.latestRelease,
 		InvalidVer:    c.invalidVer,
+		ObjectCache:   c.objectCache,
 	}
 	if c.invalidVer != nil && c.IsCurrentVersionInvalid() == "" {
 		w.InvalidVer = nil
@@ -243,6 +285,7 @@ func Load(path string) (c Cache, err error) {
 			lastCheckedAt: w.LastCheckedAt,
 			latestRelease: w.LatestRelease,
 			invalidVer:    w.InvalidVer,
+			objectCache:   w.ObjectCache,
 		}
 	}
 

internal/cli/cli.go

@@ -10,6 +10,7 @@ import (
 	"runtime/debug"
 	"slices"
 	"strings"
+	"syscall"
 	"time"
 
 	"github.com/AlecAivazis/survey/v2/terminal"
@@ -101,7 +102,7 @@ func Run(ctx context.Context, io *iostreams.IOStreams, args ...string) int {
 	task.FromContext(ctx).ShutdownWithTimeout(5 * time.Second)
 
 	switch {
-	case err == nil:
+	case err == nil || errors.Is(err, syscall.EPIPE):
 		return 0
 	case errors.Is(err, context.Canceled), errors.Is(err, terminal.InterruptErr):
 		return 127

internal/command/command.go

@@ -136,11 +136,13 @@ func newRunE(fn Runner, preparers ...preparers.Preparer) func(*cobra.Command, []
 			}
 		}()
 
-		// run the command
-		if err = fn(ctx); err == nil {
-			// and finally, run the finalizer
+		// finally, run the finalizer
+		task.FromContext(ctx).RunFinalizer(func(ctx context.Context) {
 			finalize(ctx)
-		}
+		})
+
+		// run the command
+		err = fn(ctx)
 
 		return
 	}

internal/command/logs/logs.go

@@ -5,9 +5,7 @@ import (
 	"context"
 	"errors"
 	"io"
-	"time"
 
-	"github.com/azazeal/pause"
 	"github.com/spf13/cobra"
 	"golang.org/x/sync/errgroup"
 
@@ -131,13 +129,17 @@ func nats(ctx context.Context, eg *errgroup.Group, client flyutil.Client, opts *
 			return nil
 		}
 
-		// we wait for 2 seconds before canceling the polling context so that
-		// we get a few records
-		pause.For(ctx, 2*time.Second)
-		cancelPolling()
-
+		// wait to cancel the polling context until we print a single record
 		for entry := range stream.Stream(ctx, opts) {
-			c <- entry
+			if cancelPolling != nil {
+				cancelPolling()
+				cancelPolling = nil
+			}
+			select {
+			case <-ctx.Done():
+				return ctx.Err()
+			case c <- entry:
+			}
 		}
 
 		return nil

internal/command/ssh/connect.go

@@ -62,13 +62,20 @@ type ConnectParams struct {
 func Connect(p *ConnectParams, addr string) (*ssh.Client, error) {
 	terminal.Debugf("Fetching certificate for %s\n", addr)
 
-	cert, pk, err := singleUseSSHCertificate(p.Ctx, p.Org, p.AppNames, p.Username)
+	cacheKey := fmt.Sprint(p.Username, "@", p.AppNames)
+	key, err := flyutil.FetchCertificate(p.Ctx, cacheKey, 1*time.Hour, func() (*fly.IssuedCertificate, error) {
+		cert, pk, err := singleUseSSHCertificate(p.Ctx, p.Org, p.AppNames, p.Username)
+		if err != nil {
+			return nil, fmt.Errorf("create ssh certificate: %w (if you haven't created a key for your org yet, try `flyctl ssh issue`)", err)
+		}
+		pemkey := ssh.MarshalED25519PrivateKey(pk, "single-use certificate")
+		cert.Key = string(pemkey)
+		return cert, nil
+	})
 	if err != nil {
-		return nil, fmt.Errorf("create ssh certificate: %w (if you haven't created a key for your org yet, try `flyctl ssh issue`)", err)
+		return nil, err
 	}
 
-	pemkey := ssh.MarshalED25519PrivateKey(pk, "single-use certificate")
-
 	terminal.Debugf("Keys for %s configured; connecting...\n", addr)
 
 	sshClient := &ssh.Client{
@@ -77,8 +84,8 @@ func Connect(p *ConnectParams, addr string) (*ssh.Client, error) {
 
 		Dial: p.Dialer.DialContext,
 
-		Certificate: cert.Certificate,
-		PrivateKey:  string(pemkey),
+		Certificate: key.Certificate,
+		PrivateKey:  key.Key,
 	}
 
 	var endSpin context.CancelFunc

internal/command/ssh/console.go

@@ -231,13 +231,13 @@ func addrForMachines(ctx context.Context, app *fly.AppCompact, console bool) (ad
 		return "", err
 	}
 
-	machines, err := flapsClient.ListActive(ctx)
+	machines, err := flapsClient.List(ctx, "summary=true")
 	if err != nil {
 		return "", err
 	}
 
 	machines = lo.Filter(machines, func(m *fly.Machine, _ int) bool {
-		return m.State == "started"
+		return m.State == "started" && !m.IsReleaseCommandMachine() && !m.IsFlyAppsConsole()
 	})
 
 	if len(machines) < 1 {

internal/command/status/machines.go

@@ -111,8 +111,8 @@ func RenderMachineStatus(ctx context.Context, app *fly.AppCompact, out io.Writer
 
 	for _, machine := range machines {
 		image := fmt.Sprintf("%s:%s", machine.ImageRef.Repository, machine.ImageRef.Tag)
-		// Skip API call for already-seen unknown repos, or default deploy-label prefix.
-		if unknownRepos[image] || strings.HasPrefix(machine.ImageRef.Tag, "deployment-") {
+		// Skip API call for already-seen unknown repos or without the flyio/ prefix.
+		if unknownRepos[image] || !strings.HasPrefix(machine.ImageRef.Repository, "flyio/") {
 			continue
 		}
 

internal/flyutil/cache.go

@@ -0,0 +1,64 @@
+package flyutil
+
+import (
+	"bytes"
+	"context"
+	"time"
+
+	"github.com/superfly/fly-go"
+	"github.com/superfly/flyctl/internal/cache"
+	"gopkg.in/yaml.v3"
+)
+
+type CachedClient struct {
+	cache.Cache
+	Client
+}
+
+func convert[T any](in any, err error) (T, error) {
+	var out T
+	if err != nil {
+		return out, err
+	}
+	b := new(bytes.Buffer)
+	err = yaml.NewEncoder(b).Encode(in)
+	if err != nil {
+		return out, err
+	}
+	err = yaml.NewDecoder(b).Decode(&out)
+	return out, err
+}
+
+func (c *CachedClient) GetAppCompact(ctx context.Context, appName string) (*fly.AppCompact, error) {
+	return convert[*fly.AppCompact](c.Cache.Fetch("GetAppCompact:"+appName, 1*time.Hour, 1*time.Minute, func() (any, error) {
+		return c.Client.GetAppCompact(ctx, appName)
+	}))
+}
+
+func (c *CachedClient) GetAppBasic(ctx context.Context, appName string) (*fly.AppBasic, error) {
+	return convert[*fly.AppBasic](c.Cache.Fetch("GetAppBasic:"+appName, 1*time.Hour, 1*time.Minute, func() (any, error) {
+		return c.Client.GetAppBasic(ctx, appName)
+	}))
+}
+
+func (c *CachedClient) GetOrganizations(ctx context.Context, filters ...fly.OrganizationFilter) ([]fly.Organization, error) {
+	if len(filters) > 0 {
+		return c.Client.GetOrganizations(ctx, filters...)
+	}
+	return convert[[]fly.Organization](c.Cache.Fetch("GetOrganizations", 1*time.Hour, 1*time.Minute, func() (any, error) {
+		return c.Client.GetOrganizations(ctx)
+	}))
+}
+
+func FetchCertificate(ctx context.Context, cacheKey string, duration time.Duration, fn func() (*fly.IssuedCertificate, error)) (*fly.IssuedCertificate, error) {
+	c := cache.FromContext(ctx)
+	return convert[*fly.IssuedCertificate](c.Fetch("FetchCertificate:"+cacheKey, duration, duration, func() (any, error) {
+		return fn()
+	}))
+}
+
+func (c *CachedClient) GetAppNetwork(ctx context.Context, appName string) (*string, error) {
+	return convert[*string](c.Cache.Fetch("GetAppNetwork:"+appName, 1*time.Hour, 1*time.Minute, func() (any, error) {
+		return c.Client.GetAppNetwork(ctx, appName)
+	}))
+}

internal/flyutil/client.go

@@ -7,6 +7,7 @@
 
 	genq "github.com/Khan/genqlient/graphql"
 	fly "github.com/superfly/fly-go"
+	"github.com/superfly/flyctl/internal/cache"
 	"github.com/superfly/graphql"
 )
 
@@ -117,5 +118,9 @@
 // ClientFromContext returns the Client ctx carries.
 func ClientFromContext(ctx context.Context) Client {
 	c, _ := ctx.Value(contextKeyClient).(Client)
+	ch := cache.FromContext(ctx)
+	if ch != nil && c != nil {
+		return &CachedClient{ch, c}
+	}
 	return c
 }

internal/flyutil/flyutil.go

@@ -5,10 +5,11 @@
 
 	"github.com/superfly/fly-go"
 	"github.com/superfly/flyctl/internal/buildinfo"
+	"github.com/superfly/flyctl/internal/cache"
 	"github.com/superfly/flyctl/internal/logger"
 )
 
-func NewClientFromOptions(ctx context.Context, opts fly.ClientOptions) *fly.Client {
+func NewClientFromOptions(ctx context.Context, opts fly.ClientOptions) Client {
 	if opts.Name == "" {
 		opts.Name = buildinfo.Name()
 	}
@@ -18,5 +19,10 @@
 	if v := logger.MaybeFromContext(ctx); v != nil && opts.Logger == nil {
 		opts.Logger = v
 	}
+	c := fly.NewClientFromOptions(opts)
+	ch := cache.FromContext(ctx)
+	if ch != nil && c != nil {
+		return &CachedClient{ch, c}
+	}
 	return fly.NewClientFromOptions(opts)
 }

logs/polling.go

@@ -50,6 +50,10 @@
 	for {
 		if waitFor > minWait {
 			pause.For(ctx, waitFor)
+			select {
+			case <-ctx.Done():
+				return ctx.Err()
+			}
 		}
 
 		entries, token, err := client.GetAppLogs(ctx, opts.AppName, nextToken, opts.RegionCode, opts.VMID)
@@ -82,13 +86,17 @@
 		}
 
 		for _, entry := range entries {
-			out <- LogEntry{
+			select {
+			case <-ctx.Done():
+				return ctx.Err()
+			case out <- LogEntry{
 				Instance:  entry.Instance,
 				Level:     entry.Level,
 				Message:   entry.Message,
 				Region:    entry.Region,
 				Timestamp: entry.Timestamp,
 				Meta:      entry.Meta,
+			}:
 			}
 		}