diff --git a/.github/workflows/build.yaml b/.github/workflows/build.yaml
index b910825..5111e7a 100644
--- a/.github/workflows/build.yaml
+++ b/.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
diff --git a/.golangci.yml b/.golangci.yml
index fdf192b..983d783 100644
--- a/.golangci.yml
+++ b/.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$
diff --git a/LICENSE b/LICENSE
index 6d48924..d6ce962 100644
--- a/LICENSE
+++ b/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
diff --git a/agent/config/config.go b/agent/config/config.go
index a90be89..e87881a 100644
--- a/agent/config/config.go
+++ b/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)
 
diff --git a/agent/config/config_test.go b/agent/config/config_test.go
index 1cec1ea..9e98a23 100644
--- a/agent/config/config_test.go
+++ b/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",
diff --git a/build/Dockerfile b/build/Dockerfile
index 2152a1c..90769c4 100644
--- a/build/Dockerfile
+++ b/build/Dockerfile
@@ -1,4 +1,4 @@
-FROM golang:1.24.1 AS build
+FROM golang:1.25.5 AS build
 
 ARG version
 
diff --git a/cli/agent/command.go b/cli/agent/command.go
index dd0b631..c0f36b4 100644
--- a/cli/agent/command.go
+++ b/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)
diff --git a/client/listener.go b/client/listener.go
index a7049bb..516327a 100644
--- a/client/listener.go
+++ b/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 {
diff --git a/client/upstream.go b/client/upstream.go
index 1014f1d..a1769b8 100644
--- a/client/upstream.go
+++ b/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.
diff --git a/docs/demo/config/Caddyfile b/demo/config/Caddyfile
similarity index 100%
rename from docs/demo/config/Caddyfile
rename to demo/config/Caddyfile
diff --git a/docs/demo/config/piko.yaml b/demo/config/piko.yaml
similarity index 100%
rename from docs/demo/config/piko.yaml
rename to demo/config/piko.yaml
diff --git a/docs/demo/config/prometheus.yaml b/demo/config/prometheus.yaml
similarity index 100%
rename from docs/demo/config/prometheus.yaml
rename to demo/config/prometheus.yaml
diff --git a/docs/demo/docker-compose.yaml b/demo/docker-compose.yaml
similarity index 100%
rename from docs/demo/docker-compose.yaml
rename to demo/docker-compose.yaml
diff --git a/go.mod b/go.mod
index d74e9b8..d7f52e0 100644
--- a/go.mod
+++ b/go.mod
@@ -1,15 +1,14 @@
 module github.com/andydunstall/piko
 
-go 1.24.1
+go 1.25.5
 
 require (
-	github.com/MicahParks/keyfunc/v3 v3.7.0
-	github.com/andydunstall/yamux v0.1.5
-	github.com/gin-gonic/gin v1.11.0
-	github.com/go-jose/go-jose/v4 v4.1.3
-	github.com/goccy/go-yaml v1.19.1
-	github.com/golang-jwt/jwt/v5 v5.3.0
-	github.com/google/uuid v1.6.0
+	github.com/MicahParks/keyfunc/v3 v3.8.0
+	github.com/andydunstall/yamux v0.1.6
+	github.com/gin-gonic/gin v1.12.0
+	github.com/go-jose/go-jose/v4 v4.1.4
+	github.com/goccy/go-yaml v1.19.2
+	github.com/golang-jwt/jwt/v5 v5.3.1
 	github.com/gorilla/websocket v1.5.3
 	github.com/hashicorp/go-sockaddr v1.0.7
 	github.com/oklog/run v1.2.0
@@ -19,25 +18,26 @@ require (
 	github.com/stretchr/testify v1.11.1
 	github.com/ugorji/go/codec v1.3.1
 	go.uber.org/atomic v1.11.0
-	go.uber.org/zap v1.27.1
-	golang.org/x/sync v0.19.0
+	go.uber.org/zap v1.28.0
+	golang.org/x/sync v0.20.0
 	gopkg.in/yaml.v3 v3.0.1
 )
 
 require (
 	github.com/MicahParks/jwkset v0.11.0 // indirect
 	github.com/beorn7/perks v1.0.1 // indirect
-	github.com/bytedance/sonic v1.14.0 // indirect
-	github.com/bytedance/sonic/loader v0.3.0 // indirect
+	github.com/bytedance/gopkg v0.1.3 // indirect
+	github.com/bytedance/sonic v1.15.0 // indirect
+	github.com/bytedance/sonic/loader v0.5.0 // indirect
 	github.com/cespare/xxhash/v2 v2.3.0 // indirect
 	github.com/cloudwego/base64x v0.1.6 // indirect
 	github.com/davecgh/go-spew v1.1.1 // indirect
-	github.com/gabriel-vasile/mimetype v1.4.8 // indirect
+	github.com/gabriel-vasile/mimetype v1.4.12 // indirect
 	github.com/gin-contrib/sse v1.1.0 // indirect
 	github.com/go-playground/locales v0.14.1 // indirect
 	github.com/go-playground/universal-translator v0.18.1 // indirect
-	github.com/go-playground/validator/v10 v10.27.0 // indirect
-	github.com/goccy/go-json v0.10.2 // indirect
+	github.com/go-playground/validator/v10 v10.30.1 // indirect
+	github.com/goccy/go-json v0.10.5 // indirect
 	github.com/inconshreveable/mousetrap v1.1.0 // indirect
 	github.com/json-iterator/go v1.1.12 // indirect
 	github.com/klauspost/cpuid/v2 v2.3.0 // indirect
@@ -51,19 +51,17 @@ require (
 	github.com/prometheus/client_model v0.6.2 // indirect
 	github.com/prometheus/common v0.66.1 // indirect
 	github.com/prometheus/procfs v0.16.1 // indirect
-	github.com/quic-go/qpack v0.5.1 // indirect
-	github.com/quic-go/quic-go v0.54.0 // indirect
+	github.com/quic-go/qpack v0.6.0 // indirect
+	github.com/quic-go/quic-go v0.59.0 // indirect
 	github.com/twitchyliquid64/golang-asm v0.15.1 // indirect
-	go.uber.org/mock v0.5.0 // indirect
+	go.mongodb.org/mongo-driver/v2 v2.5.0 // indirect
 	go.uber.org/multierr v1.11.0 // indirect
 	go.yaml.in/yaml/v2 v2.4.2 // indirect
-	golang.org/x/arch v0.20.0 // indirect
-	golang.org/x/crypto v0.41.0 // indirect
-	golang.org/x/mod v0.26.0 // indirect
-	golang.org/x/net v0.43.0 // indirect
-	golang.org/x/sys v0.35.0 // indirect
-	golang.org/x/text v0.28.0 // indirect
+	golang.org/x/arch v0.22.0 // indirect
+	golang.org/x/crypto v0.48.0 // indirect
+	golang.org/x/net v0.51.0 // indirect
+	golang.org/x/sys v0.41.0 // indirect
+	golang.org/x/text v0.34.0 // indirect
 	golang.org/x/time v0.9.0 // indirect
-	golang.org/x/tools v0.35.0 // indirect
-	google.golang.org/protobuf v1.36.9 // indirect
+	google.golang.org/protobuf v1.36.10 // indirect
 )
diff --git a/go.sum b/go.sum
index de60225..db81222 100644
--- a/go.sum
+++ b/go.sum
@@ -1,15 +1,17 @@
 github.com/MicahParks/jwkset v0.11.0 h1:yc0zG+jCvZpWgFDFmvs8/8jqqVBG9oyIbmBtmjOhoyQ=
 github.com/MicahParks/jwkset v0.11.0/go.mod h1:U2oRhRaLgDCLjtpGL2GseNKGmZtLs/3O7p+OZaL5vo0=
-github.com/MicahParks/keyfunc/v3 v3.7.0 h1:pdafUNyq+p3ZlvjJX1HWFP7MA3+cLpDtg69U3kITJGM=
-github.com/MicahParks/keyfunc/v3 v3.7.0/go.mod h1:z66bkCviwqfg2YUp+Jcc/xRE9IXLcMq6DrgV/+Htru0=
-github.com/andydunstall/yamux v0.1.5 h1:IM0aZukwckStCEnFy/3xGmV81eAXyRKTdSAjQo5oI+Y=
-github.com/andydunstall/yamux v0.1.5/go.mod h1:v4C9l2I4bhYdww+IVgjO0o5rVzxXbx2nneuFDHvyM28=
+github.com/MicahParks/keyfunc/v3 v3.8.0 h1:Hx2dgIjAXGk9slakM6rV9BOeaWDPEXXZ4Us8guNBfds=
+github.com/MicahParks/keyfunc/v3 v3.8.0/go.mod h1:z66bkCviwqfg2YUp+Jcc/xRE9IXLcMq6DrgV/+Htru0=
+github.com/andydunstall/yamux v0.1.6 h1:yNobhHiFaiXp8HVnGXNiICxoJAA+3m/BHaVvqPZ2vqo=
+github.com/andydunstall/yamux v0.1.6/go.mod h1:mSecAVTYsf15tbuJhLjJpasCxZvnJVXQdzj95KqhwyA=
 github.com/beorn7/perks v1.0.1 h1:VlbKKnNfV8bJzeqoa4cOKqO6bYr3WgKZxO8Z16+hsOM=
 github.com/beorn7/perks v1.0.1/go.mod h1:G2ZrVWU2WbWT9wwq4/hrbKbnv/1ERSJQ0ibhJ6rlkpw=
-github.com/bytedance/sonic v1.14.0 h1:/OfKt8HFw0kh2rj8N0F6C/qPGRESq0BbaNZgcNXXzQQ=
-github.com/bytedance/sonic v1.14.0/go.mod h1:WoEbx8WTcFJfzCe0hbmyTGrfjt8PzNEBdxlNUO24NhA=
-github.com/bytedance/sonic/loader v0.3.0 h1:dskwH8edlzNMctoruo8FPTJDF3vLtDT0sXZwvZJyqeA=
-github.com/bytedance/sonic/loader v0.3.0/go.mod h1:N8A3vUdtUebEY2/VQC0MyhYeKUFosQU6FxH2JmUe6VI=
+github.com/bytedance/gopkg v0.1.3 h1:TPBSwH8RsouGCBcMBktLt1AymVo2TVsBVCY4b6TnZ/M=
+github.com/bytedance/gopkg v0.1.3/go.mod h1:576VvJ+eJgyCzdjS+c4+77QF3p7ubbtiKARP3TxducM=
+github.com/bytedance/sonic v1.15.0 h1:/PXeWFaR5ElNcVE84U0dOHjiMHQOwNIx3K4ymzh/uSE=
+github.com/bytedance/sonic v1.15.0/go.mod h1:tFkWrPz0/CUCLEF4ri4UkHekCIcdnkqXw9VduqpJh0k=
+github.com/bytedance/sonic/loader v0.5.0 h1:gXH3KVnatgY7loH5/TkeVyXPfESoqSBSBEiDd5VjlgE=
+github.com/bytedance/sonic/loader v0.5.0/go.mod h1:AR4NYCk5DdzZizZ5djGqQ92eEhCCcdf5x77udYiSJRo=
 github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs=
 github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
 github.com/cloudwego/base64x v0.1.6 h1:t11wG9AECkCDk5fMSoxmufanudBtJ+/HemLstXDLI2M=
@@ -18,33 +20,31 @@ github.com/cpuguy83/go-md2man/v2 v2.0.6/go.mod h1:oOW0eioCTA6cOiMLiUPZOpcVxMig6N
 github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
 github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
-github.com/gabriel-vasile/mimetype v1.4.8 h1:FfZ3gj38NjllZIeJAmMhr+qKL8Wu+nOoI3GqacKw1NM=
-github.com/gabriel-vasile/mimetype v1.4.8/go.mod h1:ByKUIKGjh1ODkGM1asKUbQZOLGrPjydw3hYPU2YU9t8=
+github.com/gabriel-vasile/mimetype v1.4.12 h1:e9hWvmLYvtp846tLHam2o++qitpguFiYCKbn0w9jyqw=
+github.com/gabriel-vasile/mimetype v1.4.12/go.mod h1:d+9Oxyo1wTzWdyVUPMmXFvp4F9tea18J8ufA774AB3s=
 github.com/gin-contrib/sse v1.1.0 h1:n0w2GMuUpWDVp7qSpvze6fAu9iRxJY4Hmj6AmBOU05w=
 github.com/gin-contrib/sse v1.1.0/go.mod h1:hxRZ5gVpWMT7Z0B0gSNYqqsSCNIJMjzvm6fqCz9vjwM=
-github.com/gin-gonic/gin v1.11.0 h1:OW/6PLjyusp2PPXtyxKHU0RbX6I/l28FTdDlae5ueWk=
-github.com/gin-gonic/gin v1.11.0/go.mod h1:+iq/FyxlGzII0KHiBGjuNn4UNENUlKbGlNmc+W50Dls=
-github.com/go-jose/go-jose/v4 v4.1.3 h1:CVLmWDhDVRa6Mi/IgCgaopNosCaHz7zrMeF9MlZRkrs=
-github.com/go-jose/go-jose/v4 v4.1.3/go.mod h1:x4oUasVrzR7071A4TnHLGSPpNOm2a21K9Kf04k1rs08=
+github.com/gin-gonic/gin v1.12.0 h1:b3YAbrZtnf8N//yjKeU2+MQsh2mY5htkZidOM7O0wG8=
+github.com/gin-gonic/gin v1.12.0/go.mod h1:VxccKfsSllpKshkBWgVgRniFFAzFb9csfngsqANjnLc=
+github.com/go-jose/go-jose/v4 v4.1.4 h1:moDMcTHmvE6Groj34emNPLs/qtYXRVcd6S7NHbHz3kA=
+github.com/go-jose/go-jose/v4 v4.1.4/go.mod h1:x4oUasVrzR7071A4TnHLGSPpNOm2a21K9Kf04k1rs08=
 github.com/go-playground/assert/v2 v2.2.0 h1:JvknZsQTYeFEAhQwI4qEt9cyV5ONwRHC+lYKSsYSR8s=
 github.com/go-playground/assert/v2 v2.2.0/go.mod h1:VDjEfimB/XKnb+ZQfWdccd7VUvScMdVu0Titje2rxJ4=
 github.com/go-playground/locales v0.14.1 h1:EWaQ/wswjilfKLTECiXz7Rh+3BjFhfDFKv/oXslEjJA=
 github.com/go-playground/locales v0.14.1/go.mod h1:hxrqLVvrK65+Rwrd5Fc6F2O76J/NuW9t0sjnWqG1slY=
 github.com/go-playground/universal-translator v0.18.1 h1:Bcnm0ZwsGyWbCzImXv+pAJnYK9S473LQFuzCbDbfSFY=
 github.com/go-playground/universal-translator v0.18.1/go.mod h1:xekY+UJKNuX9WP91TpwSH2VMlDf28Uj24BCp08ZFTUY=
-github.com/go-playground/validator/v10 v10.27.0 h1:w8+XrWVMhGkxOaaowyKH35gFydVHOvC0/uWoy2Fzwn4=
-github.com/go-playground/validator/v10 v10.27.0/go.mod h1:I5QpIEbmr8On7W0TktmJAumgzX4CA1XNl4ZmDuVHKKo=
-github.com/goccy/go-json v0.10.2 h1:CrxCmQqYDkv1z7lO7Wbh2HN93uovUHgrECaO5ZrCXAU=
-github.com/goccy/go-json v0.10.2/go.mod h1:6MelG93GURQebXPDq3khkgXZkazVtN9CRI+MGFi0w8I=
-github.com/goccy/go-yaml v1.19.1 h1:3rG3+v8pkhRqoQ/88NYNMHYVGYztCOCIZ7UQhu7H+NE=
-github.com/goccy/go-yaml v1.19.1/go.mod h1:XBurs7gK8ATbW4ZPGKgcbrY1Br56PdM69F7LkFRi1kA=
-github.com/golang-jwt/jwt/v5 v5.3.0 h1:pv4AsKCKKZuqlgs5sUmn4x8UlGa0kEVt/puTpKx9vvo=
-github.com/golang-jwt/jwt/v5 v5.3.0/go.mod h1:fxCRLWMO43lRc8nhHWY6LGqRcf+1gQWArsqaEUEa5bE=
+github.com/go-playground/validator/v10 v10.30.1 h1:f3zDSN/zOma+w6+1Wswgd9fLkdwy06ntQJp0BBvFG0w=
+github.com/go-playground/validator/v10 v10.30.1/go.mod h1:oSuBIQzuJxL//3MelwSLD5hc2Tu889bF0Idm9Dg26cM=
+github.com/goccy/go-json v0.10.5 h1:Fq85nIqj+gXn/S5ahsiTlK3TmC85qgirsdTP/+DeaC4=
+github.com/goccy/go-json v0.10.5/go.mod h1:oq7eo15ShAhp70Anwd5lgX2pLfOS3QCiwU/PULtXL6M=
+github.com/goccy/go-yaml v1.19.2 h1:PmFC1S6h8ljIz6gMRBopkjP1TVT7xuwrButHID66PoM=
+github.com/goccy/go-yaml v1.19.2/go.mod h1:XBurs7gK8ATbW4ZPGKgcbrY1Br56PdM69F7LkFRi1kA=
+github.com/golang-jwt/jwt/v5 v5.3.1 h1:kYf81DTWFe7t+1VvL7eS+jKFVWaUnK9cB1qbwn63YCY=
+github.com/golang-jwt/jwt/v5 v5.3.1/go.mod h1:fxCRLWMO43lRc8nhHWY6LGqRcf+1gQWArsqaEUEa5bE=
 github.com/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8=
 github.com/google/go-cmp v0.7.0/go.mod h1:pXiqmnSA92OHEEa9HXL2W4E7lf9JzCmGVUdgjX3N/iU=
 github.com/google/gofuzz v1.0.0/go.mod h1:dBl0BpW6vV/+mYPU4Po3pmUjxk6FQPldtuIdl/M65Eg=
-github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0=
-github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
 github.com/gorilla/websocket v1.5.3 h1:saDtZ6Pbx/0u+bgYQ3q96pZgCzfhKXGPqt7kZ72aNNg=
 github.com/gorilla/websocket v1.5.3/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE=
 github.com/hashicorp/go-sockaddr v1.0.7 h1:G+pTkSO01HpR5qCxg7lxfsFEZaG+C0VssTy/9dbT+Fw=
@@ -88,10 +88,10 @@ github.com/prometheus/common v0.66.1 h1:h5E0h5/Y8niHc5DlaLlWLArTQI7tMrsfQjHV+d9Z
 github.com/prometheus/common v0.66.1/go.mod h1:gcaUsgf3KfRSwHY4dIMXLPV0K/Wg1oZ8+SbZk/HH/dA=
 github.com/prometheus/procfs v0.16.1 h1:hZ15bTNuirocR6u0JZ6BAHHmwS1p8B4P6MRqxtzMyRg=
 github.com/prometheus/procfs v0.16.1/go.mod h1:teAbpZRB1iIAJYREa1LsoWUXykVXA1KlTmWl8x/U+Is=
-github.com/quic-go/qpack v0.5.1 h1:giqksBPnT/HDtZ6VhtFKgoLOWmlyo9Ei6u9PqzIMbhI=
-github.com/quic-go/qpack v0.5.1/go.mod h1:+PC4XFrEskIVkcLzpEkbLqq1uCoxPhQuvK5rH1ZgaEg=
-github.com/quic-go/quic-go v0.54.0 h1:6s1YB9QotYI6Ospeiguknbp2Znb/jZYjZLRXn9kMQBg=
-github.com/quic-go/quic-go v0.54.0/go.mod h1:e68ZEaCdyviluZmy44P6Iey98v/Wfz6HCjQEm+l8zTY=
+github.com/quic-go/qpack v0.6.0 h1:g7W+BMYynC1LbYLSqRt8PBg5Tgwxn214ZZR34VIOjz8=
+github.com/quic-go/qpack v0.6.0/go.mod h1:lUpLKChi8njB4ty2bFLX2x4gzDqXwUpaO1DP9qMDZII=
+github.com/quic-go/quic-go v0.59.0 h1:OLJkp1Mlm/aS7dpKgTc6cnpynnD2Xg7C1pwL6vy/SAw=
+github.com/quic-go/quic-go v0.59.0/go.mod h1:upnsH4Ju1YkqpLXC305eW3yDZ4NfnNbmQRCMWS58IKU=
 github.com/rogpeppe/go-internal v1.10.0 h1:TMyTOH3F/DB16zRVcYyreMH6GnZZrwQVAoYjRBZyWFQ=
 github.com/rogpeppe/go-internal v1.10.0/go.mod h1:UQnix2H7Ngw/k4C5ijL5+65zddjncjaFoBhdsK/akog=
 github.com/russross/blackfriday/v2 v2.1.0/go.mod h1:+Rmxgy9KzJVeS9/2gXHxylqXiyQDYRxCVz55jmeOWTM=
@@ -103,50 +103,51 @@ github.com/spf13/pflag v1.0.10/go.mod h1:McXfInJRrz4CZXVZOBLb0bTZqETkiAhM9Iw0y3A
 github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
 github.com/stretchr/objx v0.4.0/go.mod h1:YvHI0jy2hoMjB+UWwv71VJQ9isScKT/TqJzVSSt89Yw=
 github.com/stretchr/objx v0.5.0/go.mod h1:Yh+to48EsGEfYuaHDzXPcE3xhTkx73EhmCGUpEOglKo=
+github.com/stretchr/objx v0.5.2/go.mod h1:FRsXN1f5AsAjCGJKqEizvkpNtU+EGNCLh3NxZ/8L+MA=
 github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI=
 github.com/stretchr/testify v1.7.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
 github.com/stretchr/testify v1.8.0/go.mod h1:yNjHg4UonilssWZ8iaSj1OCr/vHnekPRkoO+kdMU+MU=
-github.com/stretchr/testify v1.8.1/go.mod h1:w2LPCIKwWwSfY2zedu0+kehJoqGctiVI29o6fzry7u4=
+github.com/stretchr/testify v1.8.4/go.mod h1:sz/lmYIOXD/1dqDmKjjqLyZ2RngseejIcXlSw2iwfAo=
+github.com/stretchr/testify v1.10.0/go.mod h1:r2ic/lqez/lEtzL7wO/rwa5dbSLXVDPFyf8C91i36aY=
 github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U=
 github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U=
 github.com/twitchyliquid64/golang-asm v0.15.1 h1:SU5vSMR7hnwNxj24w34ZyCi/FmDZTkS4MhqMhdFk5YI=
 github.com/twitchyliquid64/golang-asm v0.15.1/go.mod h1:a1lVb/DtPvCB8fslRZhAngC2+aY1QWCk3Cedj/Gdt08=
 github.com/ugorji/go/codec v1.3.1 h1:waO7eEiFDwidsBN6agj1vJQ4AG7lh2yqXyOXqhgQuyY=
 github.com/ugorji/go/codec v1.3.1/go.mod h1:pRBVtBSKl77K30Bv8R2P+cLSGaTtex6fsA2Wjqmfxj4=
+go.mongodb.org/mongo-driver/v2 v2.5.0 h1:yXUhImUjjAInNcpTcAlPHiT7bIXhshCTL3jVBkF3xaE=
+go.mongodb.org/mongo-driver/v2 v2.5.0/go.mod h1:yOI9kBsufol30iFsl1slpdq1I0eHPzybRWdyYUs8K/0=
 go.uber.org/atomic v1.11.0 h1:ZvwS0R+56ePWxUNi+Atn9dWONBPp/AUETXlHW0DxSjE=
 go.uber.org/atomic v1.11.0/go.mod h1:LUxbIzbOniOlMKjJjyPfpl4v+PKK2cNJn91OQbhoJI0=
 go.uber.org/goleak v1.3.0 h1:2K3zAYmnTNqV73imy9J1T3WC+gmCePx2hEGkimedGto=
 go.uber.org/goleak v1.3.0/go.mod h1:CoHD4mav9JJNrW/WLlf7HGZPjdw8EucARQHekz1X6bE=
-go.uber.org/mock v0.5.0 h1:KAMbZvZPyBPWgD14IrIQ38QCyjwpvVVV6K/bHl1IwQU=
-go.uber.org/mock v0.5.0/go.mod h1:ge71pBPLYDk7QIi1LupWxdAykm7KIEFchiOqd6z7qMM=
+go.uber.org/mock v0.6.0 h1:hyF9dfmbgIX5EfOdasqLsWD6xqpNZlXblLB/Dbnwv3Y=
+go.uber.org/mock v0.6.0/go.mod h1:KiVJ4BqZJaMj4svdfmHM0AUx4NJYO8ZNpPnZn1Z+BBU=
 go.uber.org/multierr v1.11.0 h1:blXXJkSxSSfBVBlC76pxqeO+LN3aDfLQo+309xJstO0=
 go.uber.org/multierr v1.11.0/go.mod h1:20+QtiLqy0Nd6FdQB9TLXag12DsQkrbs3htMFfDN80Y=
-go.uber.org/zap v1.27.1 h1:08RqriUEv8+ArZRYSTXy1LeBScaMpVSTBhCeaZYfMYc=
-go.uber.org/zap v1.27.1/go.mod h1:GB2qFLM7cTU87MWRP2mPIjqfIDnGu+VIO4V/SdhGo2E=
+go.uber.org/zap v1.28.0 h1:IZzaP1Fv73/T/pBMLk4VutPl36uNC+OSUh3JLG3FIjo=
+go.uber.org/zap v1.28.0/go.mod h1:rDLpOi171uODNm/mxFcuYWxDsqWSAVkFdX4XojSKg/Q=
 go.yaml.in/yaml/v2 v2.4.2 h1:DzmwEr2rDGHl7lsFgAHxmNz/1NlQ7xLIrlN2h5d1eGI=
 go.yaml.in/yaml/v2 v2.4.2/go.mod h1:081UH+NErpNdqlCXm3TtEran0rJZGxAYx9hb/ELlsPU=
+go.yaml.in/yaml/v3 v3.0.4 h1:tfq32ie2Jv2UxXFdLJdh3jXuOzWiL1fo0bu/FbuKpbc=
 go.yaml.in/yaml/v3 v3.0.4/go.mod h1:DhzuOOF2ATzADvBadXxruRBLzYTpT36CKvDb3+aBEFg=
-golang.org/x/arch v0.20.0 h1:dx1zTU0MAE98U+TQ8BLl7XsJbgze2WnNKF/8tGp/Q6c=
-golang.org/x/arch v0.20.0/go.mod h1:bdwinDaKcfZUGpH09BB7ZmOfhalA8lQdzl62l8gGWsk=
-golang.org/x/crypto v0.41.0 h1:WKYxWedPGCTVVl5+WHSSrOBT0O8lx32+zxmHxijgXp4=
-golang.org/x/crypto v0.41.0/go.mod h1:pO5AFd7FA68rFak7rOAGVuygIISepHftHnr8dr6+sUc=
-golang.org/x/mod v0.26.0 h1:EGMPT//Ezu+ylkCijjPc+f4Aih7sZvaAr+O3EHBxvZg=
-golang.org/x/mod v0.26.0/go.mod h1:/j6NAhSk8iQ723BGAUyoAcn7SlD7s15Dp9Nd/SfeaFQ=
-golang.org/x/net v0.43.0 h1:lat02VYK2j4aLzMzecihNvTlJNQUq316m2Mr9rnM6YE=
-golang.org/x/net v0.43.0/go.mod h1:vhO1fvI4dGsIjh73sWfUVjj3N7CA9WkKJNQm2svM6Jg=
-golang.org/x/sync v0.19.0 h1:vV+1eWNmZ5geRlYjzm2adRgW2/mcpevXNg50YZtPCE4=
-golang.org/x/sync v0.19.0/go.mod h1:9KTHXmSnoGruLpwFjVSX0lNNA75CykiMECbovNTZqGI=
+golang.org/x/arch v0.22.0 h1:c/Zle32i5ttqRXjdLyyHZESLD/bB90DCU1g9l/0YBDI=
+golang.org/x/arch v0.22.0/go.mod h1:dNHoOeKiyja7GTvF9NJS1l3Z2yntpQNzgrjh1cU103A=
+golang.org/x/crypto v0.48.0 h1:/VRzVqiRSggnhY7gNRxPauEQ5Drw9haKdM0jqfcCFts=
+golang.org/x/crypto v0.48.0/go.mod h1:r0kV5h3qnFPlQnBSrULhlsRfryS2pmewsg+XfMgkVos=
+golang.org/x/net v0.51.0 h1:94R/GTO7mt3/4wIKpcR5gkGmRLOuE/2hNGeWq/GBIFo=
+golang.org/x/net v0.51.0/go.mod h1:aamm+2QF5ogm02fjy5Bb7CQ0WMt1/WVM7FtyaTLlA9Y=
+golang.org/x/sync v0.20.0 h1:e0PTpb7pjO8GAtTs2dQ6jYa5BWYlMuX047Dco/pItO4=
+golang.org/x/sync v0.20.0/go.mod h1:9xrNwdLfx4jkKbNva9FpL6vEN7evnE43NNNJQ2LF3+0=
 golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
-golang.org/x/sys v0.35.0 h1:vz1N37gP5bs89s7He8XuIYXpyY0+QlsKmzipCbUtyxI=
-golang.org/x/sys v0.35.0/go.mod h1:BJP2sWEmIv4KK5OTEluFJCKSidICx8ciO85XgH3Ak8k=
-golang.org/x/text v0.28.0 h1:rhazDwis8INMIwQ4tpjLDzUhx6RlXqZNPEM0huQojng=
-golang.org/x/text v0.28.0/go.mod h1:U8nCwOR8jO/marOQ0QbDiOngZVEBB7MAiitBuMjXiNU=
+golang.org/x/sys v0.41.0 h1:Ivj+2Cp/ylzLiEU89QhWblYnOE9zerudt9Ftecq2C6k=
+golang.org/x/sys v0.41.0/go.mod h1:OgkHotnGiDImocRcuBABYBEXf8A9a87e/uXjp9XT3ks=
+golang.org/x/text v0.34.0 h1:oL/Qq0Kdaqxa1KbNeMKwQq0reLCCaFtqu2eNuSeNHbk=
+golang.org/x/text v0.34.0/go.mod h1:homfLqTYRFyVYemLBFl5GgL/DWEiH5wcsQ5gSh1yziA=
 golang.org/x/time v0.9.0 h1:EsRrnYcQiGH+5FfbgvV4AP7qEZstoyrHB0DzarOQ4ZY=
 golang.org/x/time v0.9.0/go.mod h1:3BpzKBy/shNhVucY/MWOyx10tF3SFh9QdLuxbVysPQM=
-golang.org/x/tools v0.35.0 h1:mBffYraMEf7aa0sB+NuKnuCy8qI/9Bughn8dC2Gu5r0=
-golang.org/x/tools v0.35.0/go.mod h1:NKdj5HkL/73byiZSJjqJgKn3ep7KjFkBOkR/Hps3VPw=
-google.golang.org/protobuf v1.36.9 h1:w2gp2mA27hUeUzj9Ex9FBjsBm40zfaDtEWow293U7Iw=
-google.golang.org/protobuf v1.36.9/go.mod h1:fuxRtAxBytpl4zzqUh6/eyUujkJdNiuEkXntxiD/uRU=
+google.golang.org/protobuf v1.36.10 h1:AYd7cD/uASjIL6Q9LiTjz8JLcrh/88q5UObnmY3aOOE=
+google.golang.org/protobuf v1.36.10/go.mod h1:HTf+CrKn2C3g5S8VImy6tdcUvCska2kB7j23XfzDpco=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
 gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c h1:Hei/4ADfdWqJk1ZMxUNpqntNwaWcugrBjAiHlqqRiVk=
 gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c/go.mod h1:JHkPIbrfpd72SG/EVd6muEfDQjcINNoR0C8j2r3qZ4Q=
diff --git a/operations/helm/piko/Chart.yaml b/operations/helm/piko/Chart.yaml
index 503670c..03038d5 100644
--- a/operations/helm/piko/Chart.yaml
+++ b/operations/helm/piko/Chart.yaml
@@ -6,7 +6,7 @@ home: https://github.com/andydunstall/piko
 
 version: 0.1.0
 
-appVersion: "v0.8.1"
+appVersion: "v0.10.0"
 
 sources:
   - https://github.com/dragonflydb/dragonfly
diff --git a/server/config/config.go b/server/config/config.go
index e7191f5..e7a8d03 100644
--- a/server/config/config.go
+++ b/server/config/config.go
@@ -104,6 +104,44 @@ matter.`,
 	)
 }
 
+const minStreamWindowSize = 256 * 1024
+
+// StreamConfig configures the streams between the Piko server and upstream
+// listeners.
+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 agent to the server.
+	//
+	// 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 agent to the server.
+
+Must be >=256KiB. Defaults to 256KiB.`,
+	)
+}
+
 // HTTPConfig contains generic configuration for the HTTP servers.
 type HTTPConfig struct {
 	// ReadTimeout is the maximum duration for reading the entire
@@ -504,36 +542,17 @@ node to join (excluding itself) but fails to join any members.`,
 	c.Gossip.RegisterFlags(fs, "cluster")
 }
 
-type UsageConfig struct {
-	// Disable indicates whether to disable anonymous usage collection.
-	Disable bool `json:"disable" yaml:"disable"`
-}
-
-func (c *UsageConfig) RegisterFlags(fs *pflag.FlagSet) {
-	fs.BoolVar(
-		&c.Disable,
-		"usage.disable",
-		c.Disable,
-		`
-Whether to disable anonymous usage tracking.
-
-The Piko server periodically sends an anonymous report to help understand how
-Piko is being used. This report includes the Piko version, host OS, host
-architecture, requests processed and upstreams registered.`,
-	)
-}
-
 type Config struct {
 	Proxy ProxyConfig `json:"proxy" yaml:"proxy"`
 
 	Upstream UpstreamConfig `json:"upstream" yaml:"upstream"`
 
+	Stream StreamConfig `json:"stream" yaml:"stream"`
+
 	Admin AdminConfig `json:"admin" yaml:"admin"`
 
 	Cluster ClusterConfig `json:"cluster" yaml:"cluster"`
 
-	Usage UsageConfig `json:"usage" yaml:"usage"`
-
 	Log log.Config `json:"log" yaml:"log"`
 
 	// GracePeriod is the duration to gracefully shutdown the server. During
@@ -568,6 +587,9 @@ func Default() *Config {
 				MinConns:  50,
 			},
 		},
+		Stream: StreamConfig{
+			MaxWindowSize: minStreamWindowSize,
+		},
 		Admin: AdminConfig{
 			BindAddr: ":8002",
 		},
@@ -600,6 +622,10 @@ func (c *Config) Validate() error {
 		return fmt.Errorf("upstream: %w", err)
 	}
 
+	if err := c.Stream.Validate(); err != nil {
+		return fmt.Errorf("stream: %w", err)
+	}
+
 	if err := c.Admin.Validate(); err != nil {
 		return fmt.Errorf("admin: %w", err)
 	}
@@ -622,9 +648,9 @@ func (c *Config) RegisterFlags(fs *pflag.FlagSet) {
 
 	c.Upstream.RegisterFlags(fs)
 
-	c.Admin.RegisterFlags(fs)
+	c.Stream.RegisterFlags(fs)
 
-	c.Usage.RegisterFlags(fs)
+	c.Admin.RegisterFlags(fs)
 
 	c.Log.RegisterFlags(fs)
 
diff --git a/server/config/config_test.go b/server/config/config_test.go
index ac0ee7e..e2dd720 100644
--- a/server/config/config_test.go
+++ b/server/config/config_test.go
@@ -107,6 +107,9 @@ admin:
     cert: /piko/cert.pem
     key: /piko/key.pem
 
+stream:
+  max_window_size: 4194304
+
 cluster:
   node_id: "my-node"
   join:
@@ -122,9 +125,6 @@ cluster:
     interval: 100ms
     max_packet_size: 1400
 
-usage:
-  disable: true
-
 log:
   level: info
   subsystems:
@@ -218,6 +218,9 @@ grace_period: 2m
 				},
 			},
 		},
+		Stream: StreamConfig{
+			MaxWindowSize: 4 * 1024 * 1024,
+		},
 		Admin: AdminConfig{
 			BindAddr:      "10.15.104.25:8002",
 			AdvertiseAddr: "1.2.3.4:8002",
@@ -249,9 +252,6 @@ grace_period: 2m
 				MaxPacketSize: 1400,
 			},
 		},
-		Usage: UsageConfig{
-			Disable: true,
-		},
 		Log: log.Config{
 			Level: "info",
 			Subsystems: []string{
@@ -293,6 +293,7 @@ func TestConfig_LoadFlags(t *testing.T) {
 		"--upstream.rebalance.threshold", "0.2",
 		"--upstream.rebalance.shed-rate", "0.005",
 		"--upstream.rebalance.min-conns", "100",
+		"--stream.max-window-size", "4194304",
 		"--upstream.auth.hmac-secret-key", "hmac-secret-key",
 		"--upstream.auth.rsa-public-key", "rsa-public-key",
 		"--upstream.auth.ecdsa-public-key", "ecdsa-public-key",
@@ -317,7 +318,6 @@ func TestConfig_LoadFlags(t *testing.T) {
 		"--cluster.gossip.advertise-addr", "1.2.3.4:8003",
 		"--cluster.gossip.interval", "100ms",
 		"--cluster.gossip.max-packet-size", "1400",
-		"--usage.disable",
 		"--log.level", "info",
 		"--log.subsystems", "foo,bar",
 		"--grace-period", "2m",
@@ -388,6 +388,9 @@ func TestConfig_LoadFlags(t *testing.T) {
 				Key:  "/piko/key.pem",
 			},
 		},
+		Stream: StreamConfig{
+			MaxWindowSize: 4 * 1024 * 1024,
+		},
 		Admin: AdminConfig{
 			BindAddr:      "10.15.104.25:8002",
 			AdvertiseAddr: "1.2.3.4:8002",
@@ -419,9 +422,6 @@ func TestConfig_LoadFlags(t *testing.T) {
 				MaxPacketSize: 1400,
 			},
 		},
-		Usage: UsageConfig{
-			Disable: true,
-		},
 		Log: log.Config{
 			Level: "info",
 			Subsystems: []string{
diff --git a/server/proxy/httpproxy.go b/server/proxy/httpproxy.go
index e92c349..42aa722 100644
--- a/server/proxy/httpproxy.go
+++ b/server/proxy/httpproxy.go
@@ -112,8 +112,13 @@ func (p *HTTPProxy) ServeHTTPWithUpstream(
 func (p *HTTPProxy) dialUpstream(ctx context.Context, _, _ string) (net.Conn, error) {
 	// As a bit of a hack to work with http.Transport, we add the upstream
 	// to the dial context.
-	upstream := ctx.Value(upstreamContextKey).(upstream.Upstream)
-	return upstream.Dial()
+	u := ctx.Value(upstreamContextKey).(upstream.Upstream)
+	c, err := u.Dial()
+	if err != nil && errors.Is(err, upstream.ErrGone) {
+		// If the upstream is no longer accepting connections, remove it.
+		p.upstreams.RemoveConn(u)
+	}
+	return c, err
 }
 
 func (p *HTTPProxy) errorHandler(w http.ResponseWriter, _ *http.Request, err error) {
diff --git a/server/proxy/tcpproxy.go b/server/proxy/tcpproxy.go
index 1529ba2..7a1ac1e 100644
--- a/server/proxy/tcpproxy.go
+++ b/server/proxy/tcpproxy.go
@@ -1,6 +1,7 @@
 package proxy
 
 import (
+	"errors"
 	"io"
 	"net"
 	"net/http"
@@ -70,6 +71,10 @@ func (p *TCPProxy) ServeHTTP(w http.ResponseWriter, r *http.Request, endpointID
 
 	upstreamConn, err := u.Dial()
 	if err != nil {
+		if errors.Is(err, upstream.ErrGone) {
+			// If the upstream is no longer accepting connections, remove it.
+			p.upstreams.RemoveConn(u)
+		}
 		_ = errorResponse(w, http.StatusBadGateway, "upstream unreachable")
 		return
 	}
diff --git a/server/server.go b/server/server.go
index 1f16b27..23cadd0 100644
--- a/server/server.go
+++ b/server/server.go
@@ -24,7 +24,6 @@ import (
 	"github.com/andydunstall/piko/server/gossip"
 	"github.com/andydunstall/piko/server/proxy"
 	"github.com/andydunstall/piko/server/upstream"
-	"github.com/andydunstall/piko/server/usage"
 )
 
 // Server is a Piko server node.
@@ -45,8 +44,6 @@ type Server struct {
 
 	gossiper *gossip.Gossip
 
-	reporter *usage.Reporter
-
 	conf *config.Config
 
 	// fatalCh triggers a shutdown when a fatal error occurs.
@@ -196,6 +193,7 @@ func NewServer(conf *config.Config, logger log.Logger) (*Server, error) {
 		upstreamTLSConfig,
 		s.clusterState,
 		conf.Upstream,
+		conf.Stream,
 		logger,
 	)
 
@@ -225,10 +223,6 @@ func NewServer(conf *config.Config, logger log.Logger) (*Server, error) {
 	s.adminServer.AddStatus("/upstream", upstream.NewStatus(upstreams))
 	s.adminServer.AddStatus("/cluster", cluster.NewStatus(s.clusterState))
 
-	// Usage reporting.
-
-	s.reporter = usage.NewReporter(upstreams.Usage(), logger)
-
 	return s, nil
 }
 
@@ -245,12 +239,6 @@ func (s *Server) Start() error {
 	// false until the server has started.
 	s.startAdminServer()
 
-	// Usage reporting.
-
-	if !s.conf.Usage.Disable {
-		s.startUsageReporting()
-	}
-
 	// Start listening for gossip traffic for other node. This won't actively
 	// attempt to join the cluster yet, though accepts other nodes attempting
 	// to join us.
@@ -355,8 +343,6 @@ func (s *Server) Shutdown() {
 
 	s.shutdownAdminServer(ctx)
 
-	s.shutdownUsageReporting()
-
 	s.wg.Wait()
 
 	s.logger.Info("shutdown complete")
@@ -452,12 +438,6 @@ func (s *Server) startAdminServer() {
 	})
 }
 
-func (s *Server) startUsageReporting() {
-	s.runGoroutine(func() {
-		s.reporter.Start()
-	})
-}
-
 func (s *Server) shutdownProxyServer(ctx context.Context) {
 	if err := s.proxyServer.Shutdown(ctx); err != nil {
 		s.logger.Error("failed to shutdown proxy server", zap.Error(err))
@@ -465,10 +445,6 @@ func (s *Server) shutdownProxyServer(ctx context.Context) {
 	s.logger.Info("shutdown proxy server")
 }
 
-func (s *Server) shutdownUsageReporting() {
-	s.reporter.Stop()
-}
-
 func (s *Server) shutdownUpstreamServer(ctx context.Context) {
 	s.rebalanceCancel()
 	if err := s.upstreamServer.Shutdown(ctx); err != nil {
diff --git a/server/upstream/manager.go b/server/upstream/manager.go
index 360df5c..40c6290 100644
--- a/server/upstream/manager.go
+++ b/server/upstream/manager.go
@@ -5,7 +5,6 @@ import (
 	"sync"
 
 	"github.com/prometheus/client_golang/prometheus"
-	"go.uber.org/atomic"
 
 	"github.com/andydunstall/piko/server/cluster"
 )
@@ -69,18 +68,11 @@ func (lb *loadBalancer) Next() Upstream {
 	return u
 }
 
-type Usage struct {
-	Requests  *atomic.Uint64
-	Upstreams *atomic.Uint64
-}
-
 type LoadBalancedManager struct {
 	localUpstreams map[string]*loadBalancer
 
 	mu sync.Mutex
 
-	usage *Usage
-
 	cluster *cluster.State
 
 	metrics *Metrics
@@ -93,11 +85,7 @@ func NewLoadBalancedManager(cluster *cluster.State, proxyClientTLSConfig *tls.Co
 		localUpstreams: make(map[string]*loadBalancer),
 		cluster:        cluster,
 		tlsConfig:      proxyClientTLSConfig,
-		usage: &Usage{
-			Requests:  atomic.NewUint64(0),
-			Upstreams: atomic.NewUint64(0),
-		},
-		metrics: NewMetrics(),
+		metrics:        NewMetrics(),
 	}
 }
 
@@ -121,7 +109,6 @@ func (m *LoadBalancedManager) Select(endpointID string, allowRemote bool) (Upstr
 	m.metrics.RemoteRequestsTotal.With(prometheus.Labels{
 		"node_id": node.ID,
 	}).Inc()
-	m.usage.Requests.Inc()
 	return NewNodeUpstream(endpointID, node, m.tlsConfig), true
 }
 
@@ -142,7 +129,6 @@ func (m *LoadBalancedManager) AddConn(u Upstream) {
 	m.cluster.AddLocalEndpoint(u.EndpointID())
 
 	m.metrics.ConnectedUpstreams.Inc()
-	m.usage.Upstreams.Inc()
 }
 
 func (m *LoadBalancedManager) RemoveConn(u Upstream) {
@@ -175,10 +161,6 @@ func (m *LoadBalancedManager) Endpoints() map[string]int {
 	return endpoints
 }
 
-func (m *LoadBalancedManager) Usage() *Usage {
-	return m.usage
-}
-
 func (m *LoadBalancedManager) Metrics() *Metrics {
 	return m.metrics
 }
diff --git a/server/upstream/server.go b/server/upstream/server.go
index a352819..0abad52 100644
--- a/server/upstream/server.go
+++ b/server/upstream/server.go
@@ -40,7 +40,8 @@ type Server struct {
 
 	cluster *cluster.State
 
-	config config.UpstreamConfig
+	config       config.UpstreamConfig
+	streamConfig config.StreamConfig
 
 	logger log.Logger
 }
@@ -51,6 +52,7 @@ func NewServer(
 	tlsConfig *tls.Config,
 	cluster *cluster.State,
 	config config.UpstreamConfig,
+	streamConfig config.StreamConfig,
 	logger log.Logger,
 ) *Server {
 	logger = logger.WithSubsystem("upstream")
@@ -70,6 +72,7 @@ func NewServer(
 		cancel:            cancel,
 		cluster:           cluster,
 		config:            config,
+		streamConfig:      streamConfig,
 		logger:            logger,
 	}
 
@@ -223,6 +226,7 @@ func (s *Server) upstreamRoute(c *gin.Context) {
 	muxConfig := yamux.DefaultConfig()
 	muxConfig.Logger = s.logger.StdLogger(zap.WarnLevel)
 	muxConfig.LogOutput = nil
+	muxConfig.MaxStreamWindowSize = s.streamConfig.MaxWindowSize
 	sess, err := yamux.Server(conn, muxConfig)
 	if err != nil {
 		// Will not happen.
diff --git a/server/upstream/server_test.go b/server/upstream/server_test.go
index 2e11f85..48ccc23 100644
--- a/server/upstream/server_test.go
+++ b/server/upstream/server_test.go
@@ -59,7 +59,7 @@ func TestServer_Register(t *testing.T) {
 
 		manager := newFakeManager()
 
-		s := NewServer(manager, nil, nil, nil, config.UpstreamConfig{}, log.NewNopLogger())
+		s := NewServer(manager, nil, nil, nil, config.UpstreamConfig{}, config.StreamConfig{MaxWindowSize: 256 * 1024}, log.NewNopLogger())
 		go func() {
 			require.NoError(t, s.Serve(ln))
 		}()
@@ -88,7 +88,7 @@ func TestServer_Register(t *testing.T) {
 
 		manager := newFakeManager()
 
-		s := NewServer(manager, nil, nil, nil, config.UpstreamConfig{}, log.NewNopLogger())
+		s := NewServer(manager, nil, nil, nil, config.UpstreamConfig{}, config.StreamConfig{MaxWindowSize: 256 * 1024}, log.NewNopLogger())
 		go func() {
 			require.NoError(t, s.Serve(ln))
 		}()
@@ -129,7 +129,7 @@ func TestServer_Authentication(t *testing.T) {
 			},
 		}, nil)
 
-		s := NewServer(manager, verifier, nil, nil, config.UpstreamConfig{}, log.NewNopLogger())
+		s := NewServer(manager, verifier, nil, nil, config.UpstreamConfig{}, config.StreamConfig{MaxWindowSize: 256 * 1024}, log.NewNopLogger())
 		go func() {
 			require.NoError(t, s.Serve(ln))
 		}()
@@ -168,7 +168,7 @@ func TestServer_Authentication(t *testing.T) {
 			},
 		}, nil)
 
-		s := NewServer(manager, verifier, nil, nil, config.UpstreamConfig{}, log.NewNopLogger())
+		s := NewServer(manager, verifier, nil, nil, config.UpstreamConfig{}, config.StreamConfig{MaxWindowSize: 256 * 1024}, log.NewNopLogger())
 		go func() {
 			require.NoError(t, s.Serve(ln))
 		}()
@@ -207,7 +207,7 @@ func TestServer_Authentication(t *testing.T) {
 			},
 		}, nil)
 
-		s := NewServer(manager, verifier, nil, nil, config.UpstreamConfig{}, log.NewNopLogger())
+		s := NewServer(manager, verifier, nil, nil, config.UpstreamConfig{}, config.StreamConfig{MaxWindowSize: 256 * 1024}, log.NewNopLogger())
 		go func() {
 			require.NoError(t, s.Serve(ln))
 		}()
@@ -238,7 +238,7 @@ func TestServer_Authentication(t *testing.T) {
 			},
 		}, nil)
 
-		s := NewServer(manager, verifier, nil, nil, config.UpstreamConfig{}, log.NewNopLogger())
+		s := NewServer(manager, verifier, nil, nil, config.UpstreamConfig{}, config.StreamConfig{MaxWindowSize: 256 * 1024}, log.NewNopLogger())
 		go func() {
 			require.NoError(t, s.Serve(ln))
 		}()
@@ -273,7 +273,7 @@ func TestServer_Authentication(t *testing.T) {
 			},
 		}, nil)
 
-		s := NewServer(manager, verifier, nil, nil, config.UpstreamConfig{}, log.NewNopLogger())
+		s := NewServer(manager, verifier, nil, nil, config.UpstreamConfig{}, config.StreamConfig{MaxWindowSize: 256 * 1024}, log.NewNopLogger())
 		go func() {
 			require.NoError(t, s.Serve(ln))
 		}()
@@ -300,7 +300,7 @@ func TestServer_TLS(t *testing.T) {
 
 	manager := newFakeManager()
 
-	s := NewServer(manager, nil, tlsConfig, nil, config.UpstreamConfig{}, log.NewNopLogger())
+	s := NewServer(manager, nil, tlsConfig, nil, config.UpstreamConfig{}, config.StreamConfig{MaxWindowSize: 256 * 1024}, log.NewNopLogger())
 	go func() {
 		require.NoError(t, s.Serve(ln))
 	}()
diff --git a/server/upstream/upstream.go b/server/upstream/upstream.go
index d981ba2..7ba7529 100644
--- a/server/upstream/upstream.go
+++ b/server/upstream/upstream.go
@@ -2,6 +2,7 @@ package upstream
 
 import (
 	"crypto/tls"
+	"errors"
 	"net"
 
 	"github.com/andydunstall/yamux"
@@ -9,12 +10,21 @@ import (
 	"github.com/andydunstall/piko/server/cluster"
 )
 
+var (
+	// ErrGone indicates an upstream is no longer accepting connections.
+	ErrGone = errors.New("gone")
+)
+
 // Upstream represents an upstream for a given endpoint.
 //
 // An upstream may be an upstream service connected to the local node, or
 // another Piko server node.
 type Upstream interface {
 	EndpointID() string
+	// Dial opens a connection the the upstream.
+	//
+	// If the upstream signals it is no longer accepting connections, returns
+	// ErrGone.
 	Dial() (net.Conn, error)
 	// Forward indicates whether the upstream is forwarding traffic to a remote
 	// node rather than a client listener.
@@ -40,7 +50,11 @@ func (u *ConnUpstream) EndpointID() string {
 }
 
 func (u *ConnUpstream) Dial() (net.Conn, error) {
-	return u.sess.OpenStream()
+	c, err := u.sess.OpenStream()
+	if err != nil && errors.Is(err, yamux.ErrRemoteGoAway) {
+		err = ErrGone
+	}
+	return c, err
 }
 
 func (u *ConnUpstream) Forward() bool {
diff --git a/server/usage/reporter.go b/server/usage/reporter.go
deleted file mode 100644
index e5db482..0000000
--- a/server/usage/reporter.go
+++ /dev/null
@@ -1,124 +0,0 @@
-package usage
-
-import (
-	"bytes"
-	"context"
-	"encoding/json"
-	"fmt"
-	"net/http"
-	"runtime"
-	"time"
-
-	"github.com/google/uuid"
-	"go.uber.org/zap"
-
-	"github.com/andydunstall/piko/pkg/build"
-	"github.com/andydunstall/piko/pkg/log"
-	"github.com/andydunstall/piko/server/upstream"
-)
-
-const (
-	reportInterval = time.Hour
-)
-
-type Report struct {
-	ID        string `json:"id"`
-	OS        string `json:"os"`
-	Arch      string `json:"arch"`
-	Version   string `json:"version"`
-	Uptime    int64  `json:"uptime"`
-	Requests  uint64 `json:"requests"`
-	Upstreams uint64 `json:"upstreams"`
-}
-
-// Reporter sends a periodic usage report.
-type Reporter struct {
-	id    string
-	start time.Time
-	usage *upstream.Usage
-
-	ctx    context.Context
-	cancel context.CancelFunc
-
-	logger log.Logger
-}
-
-func NewReporter(usage *upstream.Usage, logger log.Logger) *Reporter {
-	ctx, cancel := context.WithCancel(context.Background())
-	return &Reporter{
-		id:     uuid.New().String(),
-		start:  time.Now(),
-		usage:  usage,
-		ctx:    ctx,
-		cancel: cancel,
-		logger: logger.WithSubsystem("reporter"),
-	}
-}
-
-func (r *Reporter) Start() {
-	r.run(r.ctx)
-}
-
-func (r *Reporter) Stop() {
-	r.cancel()
-}
-
-func (r *Reporter) run(ctx context.Context) {
-	// Report on startup.
-	r.report()
-
-	ticker := time.NewTicker(reportInterval)
-	defer ticker.Stop()
-
-	for {
-		select {
-		case <-ctx.Done():
-			// Report on shutdown.
-			r.report()
-			return
-		case <-ticker.C:
-			// Report on interval.
-			r.report()
-		}
-	}
-}
-
-func (r *Reporter) report() {
-	report := &Report{
-		ID:        r.id,
-		OS:        runtime.GOOS,
-		Arch:      runtime.GOARCH,
-		Version:   build.Version,
-		Uptime:    int64(time.Since(r.start).Seconds()),
-		Requests:  r.usage.Requests.Load(),
-		Upstreams: r.usage.Upstreams.Load(),
-	}
-	if err := r.send(report); err != nil {
-		// Debug only as theres no user impact.
-		r.logger.Debug("failed to send usage report", zap.Error(err))
-	}
-}
-
-func (r *Reporter) send(report *Report) error {
-	ctx, cancel := context.WithTimeout(context.Background(), time.Second*5)
-	defer cancel()
-
-	body, err := json.Marshal(report)
-	if err != nil {
-		return fmt.Errorf("marshal: %w", err)
-	}
-	req, err := http.NewRequestWithContext(
-		ctx, http.MethodPost, "http://report.pikoproxy.com/v1", bytes.NewBuffer(body),
-	)
-	if err != nil {
-		return fmt.Errorf("request: %w", err)
-	}
-	req.Header.Set("Content-Type", "application/json")
-	resp, err := http.DefaultClient.Do(req)
-	if err != nil {
-		return fmt.Errorf("request: %w", err)
-	}
-	defer resp.Body.Close()
-
-	return nil
-}