johnandersen777.bsky.social/socialweb-computer-kcp

Switch repository

Loading…

package gatewaykcp

import (
	"cmp"
	"context"
	"errors"
	"fmt"
	"io"
	"log/slog"
	"maps"
	"net"
	"net/url"
	"os"
	"path/filepath"
	"sort"
	"strings"
	"sync"
	"time"

	apierrors "k8s.io/apimachinery/pkg/api/errors"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/client-go/rest"
	"k8s.io/client-go/tools/clientcmd"

	"github.com/johnandersen777/socialweb-computer-kcp/api/v1alpha1"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/app/compose"
	appgateway "github.com/johnandersen777/socialweb-computer-kcp/internal/app/gateway"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/bootstrap"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/bootstrap/table"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/clientcertfront"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/clusterstack"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/controllers/runtime/kcpapply"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/controllers/runtime/kube"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/controllers/runtime/manifests"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/devcore"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/devinstance"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/foldmain"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/gatewaymain"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/hostsvc"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/kcpclient"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/kcpserver"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/kit/fsx"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/openbaoctl"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/providermain"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/tlsfront"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/vault/openbao"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/vaultlayout"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/vaultmain"
)

const (
	InstanceID         = "gateway"
	FieldManager       = "socialweb-computer"
	ApplyTimeout       = time.Minute
	VaultTimeout       = time.Minute
	ListenTimeout      = 30 * time.Second
	Poll               = 500 * time.Millisecond
	ListenerXRPC       = "xrpc"
	ListenerTLS        = "tls"
	ListenerClientCert = "client-cert"
	ListenerWebApp     = "web-app-front"
	ComponentCore      = "core"
	DefaultAddress     = "127.0.0.1:0"
	ExeCore            = "socialweb-computer"
	ExeKcp             = "kcp"
	ExeKine            = "kine"
	ExePostgres        = "postgres"
	ExeBao             = "bao"
	ExeExternalSecrets = "external-secrets"
	ExeESOBundle       = "external-secrets-bundle"
	ExeDumbpipe        = "dumbpipe"
	ExeK3s             = "k3s"
	ExeWebDownloads    = "socialweb-computer-k8s-downloads"
	recordProfile      = bootstrap.ProfileProdlike
	componentsReport   = "members"
)

var Sets = []string{"provider", "permission-inventory", "hostprocesses", compose.ManifestSet}

type Config struct {
	StateDir            string
	Prefix              string
	Backend             hostsvc.Backend
	Executables         map[string]devinstance.Executable
	Spec                v1alpha1.GatewaySpec
	FrontAddress        string
	TLSAddress          string
	CertFrontAddress    string
	TLSCert             string
	TLSKey              string
	BootstrapCluster    string
	BootstrapAccountDID string
	RootName            string
	ReadyTimeout        time.Duration
	Environ             []string
	Unit                string
	Exe                 string
	Stderr              io.Writer
	Log                 *slog.Logger
	NodeClaimAgent      *kcpclient.ServiceAccountRef
}

func NewLayout(state, prefix string) devinstance.Layout {
	kcp := filepath.Join(state, "kcp")
	run := filepath.Join(kcp, "run")
	return devinstance.Layout{ID: InstanceID, Short: table.GatewayKey, StateDir: state, RuntimeDir: run,
		Socket: filepath.Join(run, "sup.sock"), SupState: filepath.Join(state, "sup"),
		Record: filepath.Join(state, "gateway.json"), CoreRecord: filepath.Join(state, "core.json"),
		Lock: filepath.Join(kcp, "instance.lock"), CoreLock: filepath.Join(kcp, "core.lock"), Logs: filepath.Join(state, "logs"),
		KineDir: filepath.Join(kcp, "kine-a"), KcpRoot: filepath.Join(kcp, "slot-a"), Prefix: prefix}
}

func (c Config) app() table.App {
	a := table.Gateway()
	if c.Prefix != "" {
		a.Prefix = c.Prefix
	}
	return a
}

func (c Config) log() *slog.Logger {
	if c.Log == nil {
		return slog.New(slog.DiscardHandler)
	}
	return c.Log.With("component", "serve-gateway-kcp")
}

func (c Config) linkAssets(l devinstance.Layout) (map[string]devinstance.Executable, error) {
	p := bootstrap.InstallParams(c.app(), c.StateDir)
	out := map[string]devinstance.Executable{}
	maps.Copy(out, c.Executables)
	kcp, kine, pg := c.Executables[ExeKcp].Path, c.Executables[ExeKine].Path, c.Executables[ExePostgres].Path
	if kcp != "" && kine != "" && pg != "" {
		if err := bootstrap.LinkAssets(p, kcp, kine, pg); err != nil {
			return nil, err
		}
	}
	for name, link := range map[string]string{ExeKcp: p.KcpPath, ExeKine: p.KinePath, ExePostgres: p.PostgresRoot} {
		if _, err := os.Stat(link); err != nil {
			return nil, fmt.Errorf("gatewaykcp: %s is not linked at %s (install links it; a dev build names it with its -path option): %w", name, link, err)
		}
		e := out[name]
		e.Path = link
		out[name] = e
	}
	return out, nil
}

func (c Config) record(l devinstance.Layout) (devinstance.Record, error) {
	var rec devinstance.Record
	if err := fsx.ReadJSON(l.Record, &rec); err != nil {
		rec = devinstance.Record{Created: time.Now().UTC()}
	}
	exes, err := c.linkAssets(l)
	if err != nil {
		return rec, err
	}
	rec.Layout, rec.Profile, rec.Backend, rec.Executables = l, recordProfile, c.Backend, exes
	rec.BaseWorkspace, rec.Host, rec.Parent, rec.StandIns, rec.OpenBao = "", false, nil, nil, nil
	return rec, nil
}

type listeners struct {
	mu      sync.Mutex
	addr    map[string]string
	kind    map[string]string
	webApp  string
	changed chan struct{}
}

func (r *listeners) components(_ string, ls []v1alpha1.ApplicationListener) {
	r.mu.Lock()
	defer r.mu.Unlock()
	r.webApp = ""
	for _, l := range ls {
		if l.Component == appgateway.WebAppComponent && l.Address != "" {
			r.webApp = l.Address
		}
	}
	if r.webApp != "" {
		return
	}
	for _, l := range ls {
		if !strings.HasPrefix(l.Component, "svc-gatewayapi-") || !strings.HasSuffix(l.Component, "-local") || l.Address == "" {
			continue
		}
		address := l.Address
		if u, err := url.Parse(address); err == nil && u.Host != "" {
			address = u.Host
		}
		r.webApp = address
		return
	}
}

func (r *listeners) webAppAddress() string {
	r.mu.Lock()
	defer r.mu.Unlock()
	return r.webApp
}

func (r *listeners) set(name, kind string, a net.Addr) {
	r.mu.Lock()
	defer r.mu.Unlock()
	r.addr[name], r.kind[name] = a.String(), kind
	if r.changed != nil {
		close(r.changed)
		r.changed = nil
	}
}

func (r *listeners) await(name string) (string, <-chan struct{}) {
	r.mu.Lock()
	defer r.mu.Unlock()
	if r.changed == nil {
		r.changed = make(chan struct{})
	}
	return r.addr[name], r.changed
}

func (r *listeners) get(name string) string {
	r.mu.Lock()
	defer r.mu.Unlock()
	return r.addr[name]
}

func (r *listeners) report(string) []v1alpha1.ApplicationListener {
	r.mu.Lock()
	defer r.mu.Unlock()
	names := make([]string, 0, len(r.addr))
	for n := range r.addr {
		names = append(names, n)
	}
	sort.Strings(names)
	out := []v1alpha1.ApplicationListener{}
	for _, n := range names {
		out = append(out, v1alpha1.ApplicationListener{Component: ComponentCore, Name: n, Kind: r.kind[n], Address: r.addr[n]})
	}
	return out
}

func Run(ctx context.Context, c Config, m hostsvc.Manager) error {
	if !filepath.IsAbs(c.StateDir) {
		return fmt.Errorf("gatewaykcp: state dir %q is not absolute", c.StateDir)
	}
	l := NewLayout(c.StateDir, c.Prefix)
	for _, d := range []string{l.StateDir, l.KcpDir(), l.RuntimeDir, l.Logs} {
		if err := os.MkdirAll(d, 0o700); err != nil {
			return err
		}
	}
	if n := len(filepath.Join(l.RuntimeDir, "kine-b.sock")); n > devinstance.MaxSocketPath {
		return fmt.Errorf("gatewaykcp: the runtime dir %s makes socket paths of %d bytes, over %d", l.RuntimeDir, n, devinstance.MaxSocketPath)
	}
	lk, err := os.OpenFile(l.Lock, os.O_CREATE|os.O_RDWR, 0o600)
	if err != nil {
		return err
	}
	lk.Close()
	rec, err := c.record(l)
	if err != nil {
		return err
	}
	if err := fsx.WriteJSON(l.Record, rec, 0o600); err != nil {
		return err
	}
	app := c.app()
	c.log().Info("the gateway application instance starts: lock, kcp store, kine, front, active slot, sets, root vault, members",
		"state", c.StateDir, "prefix", l.UnitPrefix(), "backend", string(c.Backend), "units", app.Names().UnitPrefix())
	return run(ctx, c, l, &app, m)
}

func RunChild(ctx context.Context, c Config, l devinstance.Layout, m hostsvc.Manager) error {
	c.StateDir = l.StateDir
	if err := os.MkdirAll(l.KcpDir(), 0o700); err != nil {
		return err
	}
	c.RootName = cmp.Or(c.RootName, compose.ChildRootName)
	c.log().Info("a gateway child application of a market network starts: its workspace of the network's kcp, sets, root vault, members",
		"state", c.StateDir, "instance", l.ID, "root", c.RootName)
	return run(ctx, c, l, nil, m)
}

func run(ctx context.Context, c Config, l devinstance.Layout, app *table.App, m hostsvc.Manager) error {
	reg := &listeners{addr: map[string]string{}, kind: map[string]string{}}
	g := &gateway{c: c, l: l, reg: reg}
	addr, stopEgress, err := NetworkOutbound(ctx, c.Spec.Network, filepath.Join(c.StateDir, "tls"), filepath.Join(l.KcpDir(), egressRecord), c.log())
	if err != nil {
		return err
	}
	defer stopEgress()
	if addr != nil {
		reg.set(ListenerEgress, "http", addr)
	}
	return devcore.Run(ctx, devcore.Config{Layout: l, ReadyTimeout: c.ReadyTimeout, Log: c.Log, Application: app, FrontAddress: c.FrontAddress,
		TLSCertFile: c.TLSCert, TLSKeyFile: c.TLSKey,
		WebApp: &appgateway.WebApp{FrontKubeconfig: g.memberKubeconfig(),
			ClusterCADir: filepath.Join(c.StateDir, "clusterprovider", clusterstack.PublicCADirName),
			DownloadsDir: c.Executables[ExeWebDownloads].Path},
		OnComponents: reg.components, Before: g.before, Members: g.members, Listeners: reg.report,
		NodeClaimAgent: c.NodeClaimAgent}, m)
}

type gateway struct {
	c   Config
	l   devinstance.Layout
	kcp devinstance.Layout
	ws  string
	reg *listeners
}

func (g *gateway) adopt(rec devinstance.Record) {
	g.kcp, g.ws = g.l, rec.BaseWorkspace
	if rec.SharedKcp != "" {
		g.kcp = devinstance.Layout{StateDir: rec.SharedKcp}
	}
	g.ws = cmp.Or(g.ws, compose.DefaultWorkspace)
}

func clusterConfig(base *rest.Config, ws string) *rest.Config {
	cfg := rest.CopyConfig(base)
	cfg.Host = cfg.Host + "/clusters/" + ws
	return cfg
}

func (g *gateway) before(ctx context.Context, rec devinstance.Record) error {
	started := time.Now()
	g.adopt(rec)
	restCfg, err := kcpserver.FrontRestConfig(rec.AdminKubeconfig)
	if err != nil {
		return err
	}
	ap, err := kcpapply.New(restCfg, kcpapply.Options{WaitTimeout: ApplyTimeout, Base: g.ws, Log: g.c.Log})
	if err != nil {
		return err
	}
	ix, err := manifests.Load()
	if err != nil {
		return err
	}
	for _, name := range Sets {
		set, ok := ix.Set(name)
		if !ok {
			return fmt.Errorf("gatewaykcp: deploy/sets.yaml names no %s set", name)
		}
		values, err := manifests.DecodeValues(set.Values, nil)
		if err != nil {
			return err
		}
		if _, err := ap.Apply(ctx, set, values); err != nil {
			return fmt.Errorf("gatewaykcp: applying the %s set: %w", name, err)
		}
	}
	if err := writeMemberKubeconfig(g.activeSlotRoot(), rec.KcpURL, g.memberKubeconfig()); err != nil {
		return err
	}
	if err := applyRoot(ctx, restCfg, g.ws, g.rootName(), g.c.Spec); err != nil {
		return err
	}
	g.c.log().Info("the sets and the Gateway root object are applied", "sets", Sets, "profile", g.c.Spec.Profile,
		"seconds", time.Since(started).Seconds())
	return nil
}

func (g *gateway) memberKubeconfig() string { return filepath.Join(g.l.KcpDir(), MemberKubeconfig) }

func (g *gateway) rootName() string {
	if g.c.RootName != "" {
		return g.c.RootName
	}
	return appgateway.GatewayRootName
}

func applyRoot(ctx context.Context, restCfg *rest.Config, ws, name string, spec v1alpha1.GatewaySpec) error {
	dc, err := kube.New(clusterConfig(restCfg, ws))
	if err != nil {
		return err
	}
	o := v1alpha1.Gateway{TypeMeta: metav1.TypeMeta{APIVersion: v1alpha1.GroupVersion.String(), Kind: string(v1alpha1.ApplicationKindGateway)},
		ObjectMeta: metav1.ObjectMeta{Name: name}, Spec: spec}
	_, err = kube.Apply[kube.Doc](ctx, dc.Resource(compose.GatewayGVR), &o, kube.ApplyOptions{FieldManager: FieldManager, Force: true})
	return err
}

func (g *gateway) rootVault(ctx context.Context, rec devinstance.Record) (vaultmain.RootAccess, error) {
	restCfg, err := kcpserver.FrontRestConfig(rec.AdminKubeconfig)
	if err != nil {
		return vaultmain.RootAccess{}, err
	}
	dc, err := kube.New(clusterConfig(restCfg, g.ws))
	if err != nil {
		return vaultmain.RootAccess{}, err
	}
	last := "not declared yet"
	failed := false
	o, err := kube.Await(ctx, dc.Resource(openbaoctl.GVR), vaultlayout.RootServerName, VaultTimeout, func(o *v1alpha1.OpenBaoServer) (bool, error) {
		if o == nil {
			last = apierrors.NewNotFound(openbaoctl.GVR.GroupResource(), vaultlayout.RootServerName).Error()
			return false, nil
		}
		if openbaoctl.Synced(o) {
			return true, nil
		}
		last = fmt.Sprintf("%s %s", o.Status.Phase, o.Status.Message)
		failed = o.Status.Phase == v1alpha1.OpenBaoServerPhaseFailed && o.Status.ObservedGeneration == o.Generation
		return failed, nil
	})
	switch {
	case errors.Is(err, kube.ErrAwaitTimeout):
		return vaultmain.RootAccess{}, fmt.Errorf("gatewaykcp: the root vault %s was not Ready and synced within %s: %s",
			vaultlayout.RootServerName, VaultTimeout, last)
	case err != nil:
		return vaultmain.RootAccess{}, err
	case failed:
		return vaultmain.RootAccess{}, fmt.Errorf("gatewaykcp: the root vault %s failed: %s", vaultlayout.RootServerName, last)
	}
	return vaultmain.RootAccessOf(*o)
}

func (g *gateway) applyRoot(ctx context.Context, rec devinstance.Record, vcfg vaultmain.Config) (vaultmain.RootAccess, error) {
	deadline := time.Now().Add(VaultTimeout)
	for {
		access, err := g.rootVault(ctx, rec)
		if err != nil {
			return access, err
		}
		err = vaultmain.ApplyRoot(ctx, access, vcfg, g.c.Log.With("component", "apply-root"))
		if err == nil {
			return access, nil
		}
		if time.Now().After(deadline) {
			return access, fmt.Errorf("gatewaykcp: apply-root as client %s: %w", vaultlayout.RootClient, err)
		}
		g.c.log().Warn("apply-root did not reach the root vault; its status may name a process that is gone while the core places it again",
			"address", access.Address, "err", err.Error())
		again := time.NewTimer(min(4*Poll, max(time.Until(deadline), 0)))
		select {
		case <-ctx.Done():
			again.Stop()
			return access, ctx.Err()
		case <-again.C:
		}
	}
}

func (g *gateway) activeSlotRoot() string {
	s, err := kcpserver.ReadSlots(g.kcp.SlotsFile())
	if err != nil || s.Active == "" {
		return g.kcp.SlotRoot(table.SlotA)
	}
	if r, ok := s.Slots[s.Active]; ok && r.Root != "" {
		return r.Root
	}
	return g.kcp.SlotRoot(s.Active)
}

type member struct {
	name string
	run  func(ctx context.Context) error
}

func (g *gateway) members(ctx context.Context, rec devinstance.Record) error {
	log := g.c.log()
	cert, key := g.c.TLSCert, g.c.TLSKey
	if cert == "" || key == "" {
		var err error
		if cert, key, err = tlsfront.EnsureSelfSigned(filepath.Join(g.c.StateDir, "tls")); err != nil {
			return err
		}
	}
	env := newEnv(g.c.Environ)
	if err := g.defaults(env, rec, cert, key); err != nil {
		return err
	}
	vcfg := vaultmain.Config{KCPKubeconfig: rec.AdminKubeconfig, KCPRootDir: g.activeSlotRoot(), Workspace: g.ws,
		Server:  vaultlayout.RootServerName,
		TLSCert: cert, TLSKey: key}
	access, err := g.applyRoot(ctx, rec, vcfg)
	if err != nil {
		return err
	}
	log.Info("the root vault holds the root layout", "address", access.Address)
	if g.c.BootstrapCluster != "" {
		if err := bootstrapCluster(ctx, g.memberKubeconfig(), g.ws, g.c.BootstrapAccountDID, g.c.BootstrapCluster, log); err != nil {
			return err
		}
	}
	if code := providermain.Main(ctx, g.invocation(env, []string{"migrate"}, "migrate")); code != 0 {
		log.Warn("the legacy-key migration did not complete; the gateway starts anyway, the clusterprovider runs the same migration itself", "exit", code)
	}
	var wg sync.WaitGroup
	defer wg.Wait()
	mctx, cancel := context.WithCancel(ctx)
	defer cancel()
	wg.Go(func() {
		followMemberToken(mctx, g.activeSlotRoot, g.kcp.SlotsFile(), filepath.Join(g.l.KcpDir(), MemberToken), log)
	})
	ended := make(chan error, 8)
	start := func(m member) {
		wg.Go(func() {
			err := m.run(mctx)
			if err == nil && mctx.Err() == nil {
				err = errors.New("returned")
			}
			if err != nil && mctx.Err() == nil {
				err = fmt.Errorf("%s: %w", m.name, err)
			}
			ended <- err
		})
	}
	denyCfg, err := clientcmd.BuildConfigFromFlags("", g.memberKubeconfig())
	if err != nil {
		return fmt.Errorf("gatewaykcp: reading the member kubeconfig: %w", err)
	}
	kcpserver.FollowTokenFile(denyCfg)
	deny, err := kcpclient.New(kcpclient.Options{Host: rec.KcpURL, RestConfig: denyCfg})
	if err != nil {
		return err
	}
	certFrontAddress := g.c.CertFrontAddress
	certFrontAddress = cmp.Or(certFrontAddress, DefaultAddress)
	start(member{"client certificate front", func(ctx context.Context) error {
		return clientcertfront.Run(ctx, clientcertfront.Config{Address: certFrontAddress, CertFile: cert, KeyFile: key,
			EdgeDir: filepath.Join(g.c.StateDir, "tls"), ClusterStateDir: env.get("SOCIALWEB_CLUSTERPROVIDER_STATE_DIR"), Deny: deny, Records: deny,
			Logins:   openbao.NewLogins(nil),
			OnListen: func(a net.Addr) { g.reg.set(ListenerClientCert, "https", a) }, Log: g.c.Log})
	}})
	certFront, err := g.waitListener(mctx, ended, ListenerClientCert)
	if err != nil {
		return err
	}
	if env.get("SOCIALWEB_GATEWAY_KCP_PUBLIC_HOST") == "" {
		env.set("SOCIALWEB_GATEWAY_KCP_PUBLIC_HOST", "https://"+certFront)
		env.dflt("SOCIALWEB_GATEWAY_KCP_PUBLIC_CA_FILE", cert)
	}
	if host := env.get("SOCIALWEB_GATEWAY_KCP_PUBLIC_HOST"); host != "" {
		_ = os.Setenv("SOCIALWEB_GATEWAY_KCP_PUBLIC_HOST", host)
	}
	if ca := env.get("SOCIALWEB_GATEWAY_KCP_PUBLIC_CA_FILE"); ca != "" {
		_ = os.Setenv("SOCIALWEB_GATEWAY_KCP_PUBLIC_CA_FILE", ca)
	}
	start(member{"gateway", func(ctx context.Context) error {
		inv := g.invocation(env, nil, "serve", "gateway", "xrpc")
		inv.Listen = func(_ string, a net.Addr) { g.reg.set(ListenerXRPC, "http", a) }
		return exitErr(gatewaymain.Main(ctx, inv))
	}})
	gw, err := g.waitListener(mctx, ended, ListenerXRPC)
	if err != nil {
		return err
	}
	start(member{"tls front", func(ctx context.Context) error {
		return tlsfront.Run(ctx, tlsfront.Config{Address: g.c.TLSAddress, CertFile: cert, KeyFile: key,
			Backend: func() string { return g.reg.get(ListenerXRPC) }, OnListen: func(a net.Addr) { g.reg.set(ListenerTLS, "https", a) },
			Log: g.c.Log})
	}})
	start(member{"web app front", func(ctx context.Context) error {
		return tlsfront.Run(ctx, tlsfront.Config{Address: g.c.Spec.WebAppFront, Backend: g.reg.webAppAddress,
			OnListen: func(a net.Addr) { g.reg.set(ListenerWebApp, "tcp", a) }, Log: g.c.Log})
	}})
	start(member{"clusterprovider", func(ctx context.Context) error {
		return exitErr(providermain.Main(ctx, g.invocation(env, nil, "serve", "provider")))
	}})
	log.Info("the gateway's members run in this process; the app controller renders the "+appgateway.WebAppComponent+" web app as a component", "xrpc", gw)
	err = <-ended
	cancel()
	if err == nil {
		return ctx.Err()
	}
	return err
}

func exitErr(code int) error {
	if code == 0 {
		return nil
	}
	return fmt.Errorf("exit %d", code)
}

func (g *gateway) waitListener(ctx context.Context, ended <-chan error, name string) (string, error) {
	deadline := time.NewTimer(ListenTimeout)
	defer deadline.Stop()
	for {
		a, changed := g.reg.await(name)
		if a != "" {
			return a, nil
		}
		select {
		case err := <-ended:
			return "", fmt.Errorf("gatewaykcp: %s ended before it listened: %v", name, err)
		case <-ctx.Done():
			return "", ctx.Err()
		case <-deadline.C:
			return "", fmt.Errorf("gatewaykcp: %s reported no listener within %s", name, ListenTimeout)
		case <-changed:
		}
	}
}

func (g *gateway) invocation(e *env, args []string, verb ...string) foldmain.Invocation {
	return foldmain.Invocation{Args: args, Getenv: e.get, Stdout: g.c.Stderr, Stderr: g.c.Stderr, Exe: g.c.Exe, Verb: verb}
}