Merge upstream changes

.github/workflows/build.yaml

@@ -17,7 +17,7 @@ jobs:
 
       - uses: actions/setup-go@v4
         with:
-          go-version: '1.24.1'
+          go-version: '1.25.5'
 
       - name: Build
         run: go build ./...
@@ -30,7 +30,7 @@ jobs:
 
       - uses: actions/setup-go@v4
         with:
-          go-version: '1.24.1'
+          go-version: '1.25.5'
 
       - name: Build
         run: go build ./...
@@ -49,9 +49,9 @@ jobs:
 
       - uses: actions/setup-go@v3
         with:
-          go-version: '1.24.1'
+          go-version: '1.25.5'
 
       - name: golangci-lint
-        uses: golangci/golangci-lint-action@v3
+        uses: golangci/golangci-lint-action@v9
         with:
-          version: v1.64.8
+          version: v2.7.2

.golangci.yml

@@ -1,26 +1,41 @@
+version: "2"
 run:
-  timeout: 5m
-  modules-download-mode: readonly
   build-tags:
     - integration
     - system
-
+  modules-download-mode: readonly
 linters:
   enable:
-    - errcheck
-    - gofmt
-    - goimports
-    - govet
-    - staticcheck
     - revive
-
+  exclusions:
+    generated: lax
+    rules:
+      - path: (.+)\.go$
+        text: exported (type|method|function|const|var) (.+) should have comment(.+)or be unexported
+      - path: (.+)\.go$
+        text: Error return value of .(.*Close|.*Shutdown). is not checked
+      - path: (.+)\.go$
+        text: 'package-comments: should have a package comment'
+      - path: (.+)\.go$
+        text: 'var-naming: (.+)'
+      - path: (.+)\.go$
+        text: 'empty-block: this block is empty, you can remove it'
+      - path: (.+)\.go$
+        text: 'QF1003: could use tagged switch'
+    paths:
+      - third_party$
+      - builtin$
+      - examples$
 issues:
-  exclude-use-default: false
   max-issues-per-linter: 0
   max-same-issues: 0
-  exclude:
-    - "exported (type|method|function|const|var) (.+) should have comment(.+)or be unexported"
-    - "Error return value of .(.*Close|.*Shutdown). is not checked"
-    - "package-comments: should have a package comment"
-    - "var-naming: (.+)"
-    - "empty-block: this block is empty, you can remove it"
+formatters:
+  enable:
+    - gofmt
+    - goimports
+  exclusions:
+    generated: lax
+    paths:
+      - third_party$
+      - builtin$
+      - examples$

LICENSE

@@ -1,6 +1,6 @@
 MIT License
 
-Copyright (c) 2024 Andrew Dunstall (andydunstall@gmail.com)
+Copyright (c) 2026 Andrew Dunstall (andydunstall@gmail.com)
 
 Permission is hereby granted, free of charge, to any person obtaining a copy
 of this software and associated documentation files (the "Software"), to deal

agent/config/config.go

@@ -248,6 +248,43 @@ func (c *TLSConfig) Load() (*tls.Config, error) {
 	return tlsConfig, nil
 }
 
+const minStreamWindowSize = 256 * 1024
+
+// StreamConfig configures the streams between the agent and Piko server.
+type StreamConfig struct {
+	// MaxWindowSize is the maximum receive window size in bytes.
+	//
+	// This is used for flow control to limit how much unread data can be
+	// in-flight on a stream. Increasing this value can increase the
+	// throughput from the server to the agent.
+	//
+	// Must be >=256KiB. Defaults to 256KiB.
+	MaxWindowSize uint32 `json:"max_window_size" yaml:"max_window_size"`
+}
+
+func (c *StreamConfig) Validate() error {
+	if c.MaxWindowSize < minStreamWindowSize {
+		return fmt.Errorf("max-window-size must be >= %d", minStreamWindowSize)
+	}
+	return nil
+}
+
+func (c *StreamConfig) RegisterFlags(fs *pflag.FlagSet) {
+	fs.Uint32Var(
+		&c.MaxWindowSize,
+		"stream.max-window-size",
+		c.MaxWindowSize,
+		`
+MaxWindowSize is the maximum receive window size in bytes.
+
+This is used for flow control to limit how much unread data can be
+in-flight on a stream. Increasing this value can increase the throughput
+from the server to the agent.
+
+Must be >=256KiB. Defaults to 256KiB.`,
+	)
+}
+
 type ConnectConfig struct {
 	// URL is the Piko server URL to connect to.
 	URL string `json:"url" yaml:"url"`
@@ -388,6 +425,8 @@ type Config struct {
 
 	Connect ConnectConfig `json:"connect" yaml:"connect"`
 
+	Stream StreamConfig `json:"stream" yaml:"stream"`
+
 	Server ServerConfig `json:"server" yaml:"server"`
 
 	Log log.Config `json:"log" yaml:"log"`
@@ -404,6 +443,9 @@ func Default() *Config {
 			URL:     "http://localhost:8001",
 			Timeout: time.Second * 30,
 		},
+		Stream: StreamConfig{
+			MaxWindowSize: minStreamWindowSize,
+		},
 		Server: ServerConfig{
 			BindAddr: ":5000",
 		},
@@ -430,6 +472,10 @@ func (c *Config) Validate() error {
 		return fmt.Errorf("connect: %w", err)
 	}
 
+	if err := c.Stream.Validate(); err != nil {
+		return fmt.Errorf("stream: %w", err)
+	}
+
 	if err := c.Server.Validate(); err != nil {
 		return fmt.Errorf("server: %w", err)
 	}
@@ -447,6 +493,7 @@ func (c *Config) Validate() error {
 
 func (c *Config) RegisterFlags(fs *pflag.FlagSet) {
 	c.Connect.RegisterFlags(fs)
+	c.Stream.RegisterFlags(fs)
 	c.Server.RegisterFlags(fs)
 	c.Log.RegisterFlags(fs)
 

agent/config/config_test.go

@@ -6,9 +6,10 @@ import (
 	"testing"
 	"time"
 
+	"github.com/stretchr/testify/assert"
+
 	pikoconfig "github.com/andydunstall/piko/pkg/config"
 	"github.com/andydunstall/piko/pkg/log"
-	"github.com/stretchr/testify/assert"
 )
 
 // Tests the default configuration is valid.
@@ -91,6 +92,8 @@ connect:
   url: 'http://localhost:8001'
   timeout: 30s
   token: cyz
+stream:
+  max_window_size: 4194304
 server:
   enabled: true
   bind_addr: ':5201'
@@ -142,6 +145,9 @@ log:
 			Timeout: 30 * time.Second,
 			Token:   "cyz",
 		},
+		Stream: StreamConfig{
+			MaxWindowSize: 4 * 1024 * 1024,
+		},
 		Server: ServerConfig{
 			Enabled:  true,
 			BindAddr: ":5201",

build/Dockerfile

@@ -1,4 +1,4 @@
-FROM golang:1.24.1 AS build
+FROM golang:1.25.5 AS build
 
 ARG version
 

cli/agent/command.go

@@ -139,12 +139,13 @@ func runAgent(conf *config.Config, logger log.Logger) error {
 		proxyURL = connectProxyURL
 	}
 	upstream := &client.Upstream{
-		URL:       connectURL,
-		Token:     conf.Connect.Token,
-		TenantID:  conf.Connect.TenantID,
-		TLSConfig: connectTLSConfig,
-		ProxyURL:  proxyURL,
-		Logger:    logger.WithSubsystem("client"),
+		URL:           connectURL,
+		Token:         conf.Connect.Token,
+		TenantID:      conf.Connect.TenantID,
+		TLSConfig:     connectTLSConfig,
+		ProxyURL:      proxyURL,
+		MaxWindowSize: conf.Stream.MaxWindowSize,
+		Logger:        logger.WithSubsystem("client"),
 	}
 
 	registry := prometheus.NewRegistry()
@@ -164,7 +165,7 @@ func runAgent(conf *config.Config, logger log.Logger) error {
 		if err != nil {
 			return fmt.Errorf("listen: %s: %w", listenerConfig.EndpointID, err)
 		}
-		defer ln.Close()
+		defer ln.Shutdown()
 
 		if listenerConfig.Protocol == config.ListenerProtocolHTTP {
 			server := reverseproxy.NewServer(listenerConfig, proxyMetrics, logger)

client/listener.go

@@ -24,9 +24,20 @@ func (a *pikoAddr) String() string {
 
 // Listener is a [net.Listener] that accepts incoming connections for
 // Piko endpoints.
+//
+// The listener establishes an outbound connection to the Piko server.
+// Connections to the listener are then multiplexed via this outbound
+// connection.
+//
+// Calling Close stops accepting new connections, without closing the
+// underlying outbound connection, meaning established multiplexed connections
+// are kept open. You can close the underlying connection with Shutdown.
 type Listener interface {
 	net.Listener
 
+	// Shutdown closes the underlying connection to the Piko server.
+	Shutdown() error
+
 	// EndpointID returns the ID of the endpoint this is listening for
 	// connections on.
 	EndpointID() string
@@ -75,7 +86,7 @@ func (l *listener) AcceptWithContext(ctx context.Context) (net.Conn, error) {
 			return nil, ctx.Err()
 		}
 
-		if errors.Is(err, yamux.ErrSessionShutdown) {
+		if errors.Is(err, yamux.ErrSessionShutdown) || errors.Is(err, net.ErrClosed) {
 			return nil, ErrClosed
 		}
 
@@ -94,8 +105,22 @@ func (l *listener) Addr() net.Addr {
 func (l *listener) Close() error {
 	// Cancel to stop reconnect attempts.
 	l.closeCancel()
-	// Close the current session.
-	return l.sess.Close()
+	if l.sess != nil {
+		// Stop accepting connections. This notifies the server that this
+		// upstream is no longer accepting connections.
+		return l.sess.GoAway()
+	}
+	return nil
+}
+
+func (l *listener) Shutdown() error {
+	// Cancel to stop reconnect attempts.
+	l.closeCancel()
+	if l.sess != nil {
+		// Close the underlying connection.
+		return l.sess.Close()
+	}
+	return nil
 }
 
 func (l *listener) EndpointID() string {

client/upstream.go

@@ -15,6 +15,8 @@ import (
 	"github.com/andydunstall/piko/pkg/websocket"
 )
 
+const defaultMaxWindowSize = 256 * 1024
+
 var (
 	ErrClosed = errors.New("closed")
 )
@@ -68,6 +70,15 @@ type Upstream struct {
 	// Defaults to 15s.
 	MaxReconnectBackoff time.Duration
 
+	// MaxWindowSize is the maximum receive window size in bytes.
+	//
+	// This is used for flow control to limit how much unread data can be
+	// in-flight on a stream. Increasing this value can increase the
+	// throughput from the server to the client.
+	//
+	// Must be >=256KiB. Defaults to 256KiB.
+	MaxWindowSize uint32
+
 	// Logger is an optional logger to log connection state changes.
 	Logger Logger
 }
@@ -133,9 +144,14 @@ func (u *Upstream) connect(ctx context.Context, endpointID string) (*yamux.Sessi
 				zap.String("url", url),
 			)
 
+			maxWindowSize := u.MaxWindowSize
+			if maxWindowSize == 0 {
+				maxWindowSize = defaultMaxWindowSize
+			}
 			muxConfig := yamux.DefaultConfig()
 			muxConfig.Logger = nil
 			muxConfig.LogOutput = &yamuxLogWriter{logger: u.logger()}
+			muxConfig.MaxStreamWindowSize = maxWindowSize
 			sess, err := yamux.Client(conn, muxConfig)
 			if err != nil {
 				// Will not happen.