app: Bound memory under sustained session churn

Add three mechanisms that cap memory growth on app-access agents
handling high session volumes with stalled upload streams.

Reverse proxy buffer pool: add a `sync.Pool`-backed
`httputil.BufferPool` to the reverse proxy `Forwarder`. Without a
pool, every proxied request allocates a fresh 32 KiB buffer for
`io.Copy` that becomes garbage immediately after the request
completes. Under high concurrency this creates GC pressure.

Session writer buffer cap: add a `MaxBufferSize` config field
(default 4096) to `SessionWriter`. When the internal
`[]PreparedSessionEvent` buffer reaches capacity, `processEvents`
stops reading from `eventsCh`, creating backpressure through the
unbuffered channel back to `RecordEvent` callers. Extract a
`handleStreamDone()` helper to deduplicate stream recovery logic
between the backpressure and main select blocks. Fix a pre-existing
off-by-one in `updateStatus` where `lastIndex > 0` prevented
trimming when only one event (index 0) was confirmed.

Session chunk semaphore: add a `chunkSem` buffered channel to
`ConnectionsHandler` that limits the number of concurrently active
session chunks per agent to `MaxActiveSessionChunks` (default 256).
A slot is acquired in `newSessionChunk` before opening the recording
stream and released in `close()` after the stream shuts down. Use a
`success` flag to release the slot on error paths. Set
`ReloadOnErr: true` on the `FnCache` so that `LimitExceeded` errors
from a full semaphore are not cached for the full TTL.

Author: juliaogris

lib/events/session_writer.go

@@ -37,6 +37,12 @@ import (
 	"github.com/gravitational/teleport/lib/session"
 )
 
+// DefaultMaxBufferSize is the default maximum number of events retained
+// in the SessionWriter's in-memory buffer while waiting for upload
+// confirmation. When reached, the writer stops consuming from its
+// internal channel, creating backpressure through to RecordEvent callers.
+const DefaultMaxBufferSize = 4096
+
 // NewSessionWriter returns a new instance of session writer
 func NewSessionWriter(cfg SessionWriterConfig) (*SessionWriter, error) {
 	if err := cfg.CheckAndSetDefaults(); err != nil {
@@ -100,6 +106,14 @@ type SessionWriterConfig struct {
 
 	// BackoffDuration is a duration of the backoff before the next try
 	BackoffDuration time.Duration
+
+	// MaxBufferSize is the maximum number of events to retain in the
+	// in-memory buffer while waiting for upload confirmation. When the
+	// buffer reaches this size, processEvents stops reading new events
+	// from the channel, which creates backpressure through the
+	// unbuffered eventsCh all the way back to RecordEvent callers.
+	// Zero or negative means use DefaultMaxBufferSize.
+	MaxBufferSize int
 }
 
 // CheckAndSetDefaults checks and sets defaults
@@ -128,6 +142,9 @@ func (cfg *SessionWriterConfig) CheckAndSetDefaults() error {
 	if cfg.MakeEvents == nil {
 		cfg.MakeEvents = bytesToSessionPrintEvents
 	}
+	if cfg.MaxBufferSize <= 0 {
+		cfg.MaxBufferSize = DefaultMaxBufferSize
+	}
 	return nil
 }
 
@@ -174,6 +191,10 @@ type SessionWriter struct {
 	lostEvents     atomic.Int64
 	acceptedEvents atomic.Int64
 	slowWrites     atomic.Int64
+	// bufferFull tracks whether the buffer cap warning has been
+	// logged. It resets when the buffer drops below capacity so the
+	// warning fires again if the buffer refills.
+	bufferFull bool
 }
 
 // SessionWriterStats provides stats about lost events and slow writes
@@ -428,6 +449,29 @@ func (a *SessionWriter) processEvents() {
 			a.updateStatus(status)
 		default:
 		}
+		// When the buffer is at capacity, stop reading new events and
+		// only wait for status updates (which trim the buffer) or
+		// shutdown signals. This creates backpressure through the
+		// unbuffered eventsCh back to RecordEvent callers.
+		if len(a.buffer) >= a.cfg.MaxBufferSize {
+			if !a.bufferFull {
+				a.bufferFull = true
+				a.log.WarnContext(a.cfg.Context, "Session writer buffer at capacity, applying backpressure.",
+					"buffer_size", len(a.buffer), "max_buffer_size", a.cfg.MaxBufferSize)
+			}
+			select {
+			case status := <-a.stream.Status():
+				a.updateStatus(status)
+			case <-a.stream.Done():
+				if !a.handleStreamDone() {
+					return
+				}
+			case <-a.closeCtx.Done():
+				a.completeStream(a.stream)
+				return
+			}
+			continue
+		}
 		select {
 		case status := <-a.stream.Status():
 			a.updateStatus(status)
@@ -454,14 +498,7 @@ func (a *SessionWriter) processEvents() {
 				a.log.DebugContext(a.cfg.Context, "Recovered stream", "duration", time.Since(start))
 			}
 		case <-a.stream.Done():
-			if a.closeCtx.Err() != nil {
-				// don't attempt recovery if we're closing
-				return
-			}
-			a.log.DebugContext(a.cfg.Context, "Stream was closed by the server, attempting to recover.")
-			if err := a.recoverStream(); err != nil {
-				a.log.WarnContext(a.cfg.Context, "Failed to recover stream.", "error", err)
-
+			if !a.handleStreamDone() {
 				return
 			}
 		case <-a.closeCtx.Done():
@@ -471,6 +508,21 @@ func (a *SessionWriter) processEvents() {
 	}
 }
 
+// handleStreamDone attempts to recover a server-closed stream. It
+// returns true if recovery succeeded and the caller should continue
+// the event loop, or false if the caller should return.
+func (a *SessionWriter) handleStreamDone() bool {
+	if a.closeCtx.Err() != nil {
+		return false
+	}
+	a.log.DebugContext(a.cfg.Context, "Stream was closed by the server, attempting to recover.")
+	if err := a.recoverStream(); err != nil {
+		a.log.WarnContext(a.cfg.Context, "Failed to recover stream.", "error", err)
+		return false
+	}
+	return true
+}
+
 // IsPermanentEmitError checks if the error contains either a sole
 // [trace.BadParameter] error in its chain, or a [trace.Aggregate] error
 // composed entirely of BadParameters.
@@ -612,10 +664,15 @@ func (a *SessionWriter) updateStatus(status apievents.StreamStatus) {
 		}
 		lastIndex = i
 	}
-	if lastIndex > 0 {
+	if lastIndex >= 0 {
 		before := len(a.buffer)
 		a.buffer = a.buffer[lastIndex+1:]
 		a.log.DebugContext(a.closeCtx, "Removed saved events", "removed", before-len(a.buffer), "remaining", len(a.buffer))
+		if a.bufferFull && len(a.buffer) < a.cfg.MaxBufferSize {
+			a.bufferFull = false
+			a.log.InfoContext(a.closeCtx, "Session writer buffer below capacity, backpressure released.",
+				"buffer_size", len(a.buffer), "max_buffer_size", a.cfg.MaxBufferSize)
+		}
 	}
 }
 

lib/httplib/reverseproxy/reverse_proxy.go

@@ -25,6 +25,7 @@ import (
 	"net/http"
 	"net/http/httputil"
 	"net/url"
+	"sync"
 	"time"
 
 	"github.com/gravitational/trace"
@@ -33,6 +34,27 @@ import (
 	logutils "github.com/gravitational/teleport/lib/utils/log"
 )
 
+// bufferPool is a sync.Pool-backed implementation of httputil.BufferPool.
+// It reuses 32 KiB buffers across proxied requests to reduce GC pressure
+// under high concurrency.
+type bufferPool struct {
+	pool sync.Pool
+}
+
+func (b *bufferPool) Get() []byte {
+	if v := b.pool.Get(); v != nil {
+		return v.([]byte)
+	}
+	return make([]byte, 32*1024)
+}
+
+func (b *bufferPool) Put(buf []byte) {
+	b.pool.Put(buf) //nolint:staticcheck // ReverseProxy does not retain buf after Put
+}
+
+// defaultBufferPool is shared across all Forwarder instances.
+var defaultBufferPool = &bufferPool{}
+
 // X-* Header names.
 const (
 	XForwardedProto  = "X-Forwarded-Proto"
@@ -77,6 +99,7 @@ func New(opts ...Option) (*Forwarder, error) {
 		ReverseProxy: &httputil.ReverseProxy{
 			ErrorHandler: DefaultHandler.ServeHTTP,
 			ErrorLog:     log.Default(),
+			BufferPool:   defaultBufferPool,
 		},
 		logger: slog.Default(),
 	}

lib/srv/app/connections_handler.go

@@ -124,6 +124,12 @@ type ConnectionsHandlerConfig struct {
 
 	// InsecureMode defines whether insecure connections are allowed.
 	InsecureMode bool
+
+	// MaxActiveSessionChunks limits the number of concurrently active
+	// session chunks on this agent. Each active chunk holds an open
+	// recording stream. Zero or negative means use
+	// DefaultMaxActiveSessionChunks.
+	MaxActiveSessionChunks int
 }
 
 // CheckAndSetDefaults validates the config values and sets defaults.
@@ -172,6 +178,9 @@ func (c *ConnectionsHandlerConfig) CheckAndSetDefaults() error {
 	if c.ServiceComponent == "" {
 		return trace.BadParameter("service component missing")
 	}
+	if c.MaxActiveSessionChunks <= 0 {
+		c.MaxActiveSessionChunks = DefaultMaxActiveSessionChunks
+	}
 	return nil
 }
 
@@ -211,6 +220,9 @@ type ConnectionsHandler struct {
 
 	// getAppByPublicAddress returns a types.Application using the public address as matcher.
 	getAppByPublicAddress func(context.Context, string) (types.Application, error)
+
+	// chunkSem limits concurrent active session chunks.
+	chunkSem chan struct{}
 }
 
 // NewConnectionsHandler returns a new ConnectionsHandler.
@@ -244,6 +256,8 @@ func NewConnectionsHandler(closeContext context.Context, cfg *ConnectionsHandler
 		return nil, trace.Wrap(err)
 	}
 
+	chunkSem := make(chan struct{}, cfg.MaxActiveSessionChunks)
+
 	c := &ConnectionsHandler{
 		cfg:          cfg,
 		closeContext: closeContext,
@@ -252,6 +266,7 @@ func NewConnectionsHandler(closeContext context.Context, cfg *ConnectionsHandler
 		gcpHandler:   gcpHandler,
 		connAuth:     make(map[net.Conn]error),
 		log:          slog.With(teleport.ComponentKey, cfg.ServiceComponent),
+		chunkSem:     chunkSem,
 		getAppByPublicAddress: func(ctx context.Context, s string) (types.Application, error) {
 			return nil, trace.NotFound("no applications are being proxied")
 		},
@@ -265,6 +280,7 @@ func NewConnectionsHandler(closeContext context.Context, cfg *ConnectionsHandler
 		Clock:           c.cfg.Clock,
 		CleanupInterval: time.Second,
 		OnExpiry:        c.onSessionExpired,
+		ReloadOnErr:     true,
 	})
 	if err != nil {
 		return nil, trace.Wrap(err)

lib/srv/app/session.go

@@ -45,6 +45,13 @@ import (
 // sessionChunkCloseTimeout is the default timeout used for sessionChunk.closeTimeout
 const sessionChunkCloseTimeout = 1 * time.Hour
 
+// DefaultMaxActiveSessionChunks is the default limit on concurrently
+// active session chunks per agent. Each chunk holds an open recording
+// stream whose in-memory footprint depends on upload throughput
+// (typically 3-8 MiB under IOPS pressure). 256 chunks at 8 MiB is
+// ~2 GiB, which caps memory without restricting normal workloads.
+const DefaultMaxActiveSessionChunks = 256
+
 var errSessionChunkAlreadyClosed = errors.New("session chunk already closed")
 
 // sessionChunk holds an open request handler and stream closer for an app session.
@@ -81,6 +88,14 @@ type sessionChunk struct {
 	// for ~7 minutes at most.
 	closeTimeout time.Duration
 
+	// chunkSem is an optional semaphore shared across all session
+	// chunks on the same agent. A slot is acquired in newSessionChunk
+	// before opening the recording stream (returning LimitExceeded if
+	// full) and released in close after the stream shuts down. This
+	// bounds the number of concurrently open recording streams to
+	// prevent unbounded memory growth.
+	chunkSem chan struct{}
+
 	log *slog.Logger
 }
 
@@ -92,11 +107,33 @@ type sessionOpt func(context.Context, *sessionChunk, *tlsca.Identity, types.Appl
 // and as such expects `release()` to eventually be called
 // by the caller of this function.
 func (c *ConnectionsHandler) newSessionChunk(ctx context.Context, identity *tlsca.Identity, app types.Application, startTime time.Time, opts ...sessionOpt) (*sessionChunk, error) {
+	// Acquire a semaphore slot before opening the recording stream.
+	// Each chunk holds an open stream that consumes memory; bounding
+	// the number of concurrent chunks prevents OOM under load. The
+	// slot is released in close() when the chunk shuts down.
+	if c.chunkSem != nil {
+		select {
+		case c.chunkSem <- struct{}{}:
+		default:
+			return nil, trace.LimitExceeded("too many active session chunks")
+		}
+	}
+
+	// If chunk creation fails after acquiring the semaphore slot,
+	// release it so we do not leak slots on error paths.
+	success := false
+	defer func() {
+		if !success && c.chunkSem != nil {
+			<-c.chunkSem
+		}
+	}()
+
 	sess := &sessionChunk{
 		id:           uuid.New().String(),
 		closeC:       make(chan struct{}),
 		inflightCond: sync.NewCond(&sync.Mutex{}),
 		closeTimeout: sessionChunkCloseTimeout,
+		chunkSem:     c.chunkSem,
 		log:          c.log,
 	}
 
@@ -137,6 +174,7 @@ func (c *ConnectionsHandler) newSessionChunk(ctx context.Context, identity *tlsc
 		return nil, trace.Wrap(err)
 	}
 
+	success = true
 	sess.log.DebugContext(ctx, "Created app session chunk", "session_id", sess.id)
 	return sess, nil
 }
@@ -197,9 +235,9 @@ func (c *ConnectionsHandler) withGCPHandler(ctx context.Context, sess *sessionCh
 	return nil
 }
 
-// acquire() increments in-flight request count by 1.
-// It is supposed to be paired with a `release()` call,
-// after the chunk is done with for the individual request
+// acquire increments in-flight request count by 1. It is supposed to
+// be paired with a release call once the individual request finishes
+// using the chunk.
 func (s *sessionChunk) acquire() error {
 	s.inflightCond.L.Lock()
 	defer s.inflightCond.L.Unlock()
@@ -251,7 +289,13 @@ func (s *sessionChunk) close(ctx context.Context) error {
 	s.inflightCond.L.Unlock()
 	close(s.closeC)
 	s.log.DebugContext(ctx, "Closed session chunk", "session_id", s.id)
-	return trace.Wrap(s.streamCloser.Close(ctx))
+	err := s.streamCloser.Close(ctx)
+	// Release the semaphore slot after the stream is closed, not
+	// before, so the slot accurately reflects an open stream.
+	if s.chunkSem != nil {
+		<-s.chunkSem
+	}
+	return trace.Wrap(err)
 }
 
 func (c *ConnectionsHandler) onSessionExpired(ctx context.Context, key, expired any) {