diff --git a/src/net/http/http2.go b/src/net/http/http2.go
index 21468182ab0217..602459bf331fc1 100644
--- a/src/net/http/http2.go
+++ b/src/net/http/http2.go
@@ -87,11 +87,17 @@ func (s *Server) setHTTP2Config(conf http2ExternalServerConfig) {
 		panic("http: HTTP/2 Server already registered")
 	}
 	s.h2Config = conf
-	s.h2Config.ServeConnFunc(s.serveHTTP2Conn)
+	s.h2Config.ServeConnFunc(func(ctx context.Context, nc net.Conn, h Handler, sawClientPreface bool, upgradeReq *Request, settings []byte) {
+		s.serveHTTP2Conn(ctx, nc, h, sawClientPreface, upgradeReq, settings, nil)
+	})
 	s.configureHTTP2()
 }
 
-func (s *Server) serveHTTP2Conn(ctx context.Context, nc net.Conn, h Handler, sawClientPreface bool, upgradeReq *Request, settings []byte) {
+// serveHTTP2Conn serves nc with the HTTP/2 server. It may return before the
+// connection is done being served (an idle HTTP/2 connection doesn't hold
+// onto a goroutine); onClose, if non-nil, runs once the connection is done
+// and has been closed.
+func (s *Server) serveHTTP2Conn(ctx context.Context, nc net.Conn, h Handler, sawClientPreface bool, upgradeReq *Request, settings []byte, onClose func()) {
 	s.setupHTTP2_ServeTLS()
 	var serverUpgradeReq *http2.ServerRequest
 	if upgradeReq != nil {
@@ -106,6 +112,7 @@ func (s *Server) serveHTTP2Conn(ctx context.Context, nc net.Conn, h Handler, saw
 		SawClientPreface: sawClientPreface,
 		UpgradeRequest:   serverUpgradeReq,
 		Settings:         settings,
+		OnClose:          onClose,
 	})
 }
 
diff --git a/src/net/http/internal/http2/export_test.go b/src/net/http/internal/http2/export_test.go
index e9d7c6bead5d9d..21e2f2ebdc1af7 100644
--- a/src/net/http/internal/http2/export_test.go
+++ b/src/net/http/internal/http2/export_test.go
@@ -82,6 +82,7 @@ func (sc *serverConn) TestFlowControlConsumed() (consumed int32) {
 
 func (sc *serverConn) TestStreamExists(id uint32) bool {
 	ch := make(chan bool, 1)
+	sc.beginServeSend()
 	sc.serveMsgCh <- func(int) {
 		ch <- (sc.streams[id] != nil)
 	}
@@ -90,6 +91,7 @@ func (sc *serverConn) TestStreamExists(id uint32) bool {
 
 func (sc *serverConn) TestStreamState(id uint32) streamState {
 	ch := make(chan streamState, 1)
+	sc.beginServeSend()
 	sc.serveMsgCh <- func(int) {
 		state, _ := sc.state(id)
 		ch <- state
@@ -97,6 +99,14 @@ func (sc *serverConn) TestStreamState(id uint32) streamState {
 	return <-ch
 }
 
+// TestServeParked reports whether the connection's serve goroutine has
+// exited because the connection is idle. See serverConn.serveLoop.
+func (sc *serverConn) TestServeParked() bool {
+	sc.parkMu.Lock()
+	defer sc.parkMu.Unlock()
+	return sc.parked
+}
+
 func (sc *serverConn) StartGracefulShutdown() { sc.startGracefulShutdown() }
 
 func (sc *serverConn) TestHPACKEncoder() *hpack.Encoder {
diff --git a/src/net/http/internal/http2/gotrack.go b/src/net/http/internal/http2/gotrack.go
index ba0db1690c77ef..14f651a6232f2c 100644
--- a/src/net/http/internal/http2/gotrack.go
+++ b/src/net/http/internal/http2/gotrack.go
@@ -57,6 +57,33 @@ func (g goroutineLock) checkNotOn() {
 	}
 }
 
+// A goroutineLocker is a goroutineLock whose owning goroutine can be
+// atomically replaced, for values whose owning goroutine changes over
+// their lifetime (such as the server's serve loop, which parks when
+// the connection is idle and resumes on a new goroutine).
+type goroutineLocker struct {
+	v atomic.Uint64
+}
+
+// setOwner records the current goroutine as the owner.
+func (g *goroutineLocker) setOwner() {
+	g.v.Store(uint64(newGoroutineLock()))
+}
+
+func (g *goroutineLocker) check() {
+	if !DebugGoroutines {
+		return
+	}
+	goroutineLock(g.v.Load()).check()
+}
+
+func (g *goroutineLocker) checkNotOn() {
+	if !DebugGoroutines {
+		return
+	}
+	goroutineLock(g.v.Load()).checkNotOn()
+}
+
 var goroutineSpace = []byte("goroutine ")
 
 func curGoroutineID() uint64 {
diff --git a/src/net/http/internal/http2/server.go b/src/net/http/internal/http2/server.go
index bd2533a7c36e66..b151d8fa51d216 100644
--- a/src/net/http/internal/http2/server.go
+++ b/src/net/http/internal/http2/server.go
@@ -2,18 +2,6 @@
 // Use of this source code is governed by a BSD-style
 // license that can be found in the LICENSE file.
 
-// TODO: turn off the serve goroutine when idle, so
-// an idle conn only has the readFrames goroutine active. (which could
-// also be optimized probably to pin less memory in crypto/tls). This
-// would involve tracking when the serve goroutine is active (atomic
-// int32 read/CAS probably?) and starting it up when frames arrive,
-// and shutting it down when all handlers exit. the occasional PING
-// packets could use time.AfterFunc to call sc.wakeStartServeLoop()
-// (which is a no-op if already running) and then queue the PING write
-// as normal. The serve loop would then exit in most cases (if no
-// Handlers running) and not be woken up again until the PING packet
-// returns.
-
 // TODO (maybe): add a mechanism for Handlers to going into
 // half-closed-local mode (rw.(io.Closer) test?) but not exit their
 // handler, and continue to be able to read from the
@@ -33,6 +21,7 @@ import (
 	"crypto/tls"
 	"errors"
 	"fmt"
+	"internal/godebug"
 	"internal/synctest"
 	"io"
 	"log"
@@ -220,6 +209,16 @@ type ServeConnOpts struct {
 	// SawClientPreface is set if the HTTP/2 connection preface
 	// has already been read from the connection.
 	SawClientPreface bool
+
+	// OnClose is an optional callback that runs exactly once, after the
+	// connection is done being served and has been closed. It may run on
+	// any goroutine.
+	//
+	// Setting OnClose also grants ServeConn permission to return before
+	// the connection is done (see the ServeConn documentation), so
+	// callers providing it must release their per-connection resources
+	// here rather than after ServeConn returns.
+	OnClose func()
 }
 
 func (o *ServeConnOpts) context() context.Context {
@@ -230,7 +229,11 @@ func (o *ServeConnOpts) context() context.Context {
 }
 
 // ServeConn serves HTTP/2 requests on the provided connection and
-// blocks until the connection is no longer readable.
+// blocks until the connection is no longer readable. As an exception,
+// if opts.OnClose is set, ServeConn may instead return as soon as the
+// connection goes idle, handing its servicing off to background
+// goroutines; the connection is still being served after ServeConn
+// returns, and opts.OnClose reports when it is done.
 //
 // ServeConn starts speaking HTTP/2 assuming that c has not had any
 // reads or writes. It writes its initial settings frame and expects
@@ -266,7 +269,6 @@ var (
 
 func (s *Server) serveConn(c net.Conn, opts *ServeConnOpts, newf func(*serverConn)) {
 	baseCtx, cancel := serverConnBaseContext(c, opts)
-	defer cancel()
 
 	conf := configFromServer(opts.BaseConfig)
 	sc := &serverConn{
@@ -274,6 +276,8 @@ func (s *Server) serveConn(c net.Conn, opts *ServeConnOpts, newf func(*serverCon
 		hs:                          opts.BaseConfig,
 		conn:                        c,
 		baseCtx:                     baseCtx,
+		baseCtxCancel:               cancel,
+		onClose:                     opts.OnClose,
 		remoteAddrStr:               c.RemoteAddr().String(),
 		bw:                          newBufferedWriter(c, conf.WriteByteTimeout),
 		handler:                     opts.Handler,
@@ -291,16 +295,33 @@ func (s *Server) serveConn(c net.Conn, opts *ServeConnOpts, newf func(*serverCon
 		maxFrameSize:                initialMaxFrameSize,
 		pingTimeout:                 conf.PingTimeout,
 		countErrorFunc:              conf.CountError,
-		serveG:                      newGoroutineLock(),
 		pushEnabled:                 true,
 		sawClientPreface:            opts.SawClientPreface,
 	}
+	sc.serveG.setOwner() // serve is called below on this same goroutine
+	sc.resumeServeFunc = sc.resumeServe
+	if opts.OnClose != nil && http2serveparking.Value() != "0" {
+		sc.serveParkingEnabled = true
+	}
 	if newf != nil {
 		newf(sc)
 	}
 
 	s.registerConn(sc)
-	defer s.unregisterConn(sc)
+	served := false
+	defer func() {
+		if !served {
+			// We never made it to sc.serve, whose teardown owns
+			// releasing the connection's resources, so release them
+			// here instead. (We either rejected the connection or
+			// panicked during setup.)
+			s.unregisterConn(sc)
+			cancel()
+			if f := sc.onClose; f != nil {
+				f()
+			}
+		}
+	}()
 
 	switch {
 	case sc.hs.DisableClientPriority():
@@ -411,6 +432,7 @@ func (s *Server) serveConn(c net.Conn, opts *ServeConnOpts, newf func(*serverCon
 		opts.UpgradeRequest = nil
 	}
 
+	served = true
 	sc.serve(conf)
 }
 
@@ -428,28 +450,52 @@ func (sc *serverConn) rejectConn(err ErrCode, debug string) {
 
 type serverConn struct {
 	// Immutable:
-	srv              *Server
-	hs               ServerConfig
-	conn             net.Conn
-	bw               *bufferedWriter // writing to conn
-	handler          Handler
-	baseCtx          context.Context
-	framer           *Framer
-	doneServing      chan struct{}          // closed when serverConn.serve ends
-	readFrameCh      chan readFrameResult   // written by serverConn.readFrames
-	wantWriteFrameCh chan FrameWriteRequest // from handlers -> serve
-	wroteFrameCh     chan frameWriteResult  // from writeFrameAsync -> serve, tickles more frame writes
-	bodyReadCh       chan bodyReadMsg       // from handlers -> serve
-	serveMsgCh       chan any               // misc messages & code to send to / run on the serve loop
-	flow             outflow                // conn-wide (not stream-specific) outbound flow control
-	inflow           inflow                 // conn-wide inbound flow control
-	tlsState         *tls.ConnectionState   // shared by all handlers, like net/http
-	remoteAddrStr    string
-	writeSched       WriteScheduler
-	countErrorFunc   func(errType string)
+	srv           *Server // for unregisterConn in teardown
+	hs            ServerConfig
+	conn          net.Conn
+	bw            *bufferedWriter // writing to conn
+	handler       Handler
+	baseCtx       context.Context
+	baseCtxCancel context.CancelFunc
+	onClose       func() // or nil; from ServeConnOpts.OnClose, called by teardown
+
+	// serveParkingEnabled is whether this connection's serve goroutine
+	// may park when the connection is idle. It requires the OnClose
+	// hook (which is what lets ServeConn return early) and
+	// GODEBUG=http2serveparking to be enabled. It is set once in
+	// serveConn, before any goroutine that sends to the serve loop's
+	// channels exists, and never changes, so it can be read without a
+	// lock, letting beginServeSend and endServeSend skip the parkMu
+	// accounting on connections that can never park.
+	serveParkingEnabled bool
+	framer              *Framer
+	doneServing         chan struct{}          // closed by teardown when the connection is done
+	readFrameCh         chan readFrameResult   // written by serverConn.readFrames
+	wantWriteFrameCh    chan FrameWriteRequest // from handlers -> serve
+	wroteFrameCh        chan frameWriteResult  // from writeFrameAsync -> serve, tickles more frame writes
+	bodyReadCh          chan bodyReadMsg       // from handlers -> serve
+	serveMsgCh          chan any               // misc messages & code to send to / run on the serve loop
+	flow                outflow                // conn-wide (not stream-specific) outbound flow control
+	inflow              inflow                 // conn-wide inbound flow control
+	tlsState            *tls.ConnectionState   // shared by all handlers, like net/http
+	remoteAddrStr       string
+	writeSched          WriteScheduler
+	countErrorFunc      func(errType string)
+
+	// resumeServeFunc is sc.resumeServe, allocated once at setup so
+	// that beginServeSend's "go sc.resumeServeFunc()" doesn't allocate
+	// a method-value closure per park/resume cycle.
+	resumeServeFunc func()
+
+	// parkMu guards the following fields, which coordinate parking the
+	// serve goroutine (exiting it while the connection is idle) with
+	// the goroutines that send it work. See serverConn.serveLoop.
+	parkMu            sync.Mutex
+	parked            bool // serve goroutine has exited; the next sender revives it
+	serveSendsPending int  // sends to serve's channels begun but not yet received by serve
 
 	// Everything following is owned by the serve loop; use serveG.check():
-	serveG                      goroutineLock // used to verify funcs are on serve()
+	serveG                      goroutineLocker // used to verify funcs are on serve()
 	pushEnabled                 bool
 	sawClientPreface            bool // preface has already been read, used in h2c upgrade
 	sawFirstSettings            bool // got the initial SETTINGS frame after the preface
@@ -486,6 +532,9 @@ type serverConn struct {
 	readIdleTimeout             time.Duration
 	pingTimeout                 time.Duration
 	readIdleTimer               *time.Timer // nil if unused
+	settingsTimer               *time.Timer // fires if the client never sends SETTINGS; nil after the first frame arrives
+	lastFrameTime               time.Time   // when we last read a frame from the client
+	loopNum                     int         // serve loop iteration counter, observable by tests
 
 	// Owned by the writeFrameAsync goroutine:
 	headerWriteBuf bytes.Buffer
@@ -710,9 +759,11 @@ func (sc *serverConn) readFrames() {
 	gateDone := func() { gate <- struct{}{} }
 	for {
 		f, err := sc.framer.ReadFrame()
+		sc.beginServeSend()
 		select {
 		case sc.readFrameCh <- readFrameResult{f, err, gateDone}:
 		case <-sc.doneServing:
+			sc.endServeSend()
 			return
 		}
 		select {
@@ -744,6 +795,7 @@ func (sc *serverConn) writeFrameAsync(wr FrameWriteRequest, wd *writeData) {
 	} else {
 		err = sc.framer.endWrite()
 	}
+	sc.beginServeSend()
 	sc.wroteFrameCh <- frameWriteResult{wr: wr, err: err}
 }
 
@@ -779,10 +831,12 @@ func (sc *serverConn) notePanic() {
 func (sc *serverConn) serve(conf Config) {
 	sc.serveG.check()
 	defer sc.notePanic()
-	defer sc.conn.Close()
-	defer sc.closeAllStreamsOnConnClose()
-	defer sc.stopShutdownTimer()
-	defer close(sc.doneServing) // unblocks handlers trying to send
+	parked := false
+	defer func() {
+		if !parked {
+			sc.teardown()
+		}
+	}()
 
 	if VerboseLogs {
 		sc.vlogf("http2: server connection from %v on %p", sc.conn.RemoteAddr(), sc.hs)
@@ -825,71 +879,145 @@ func (sc *serverConn) serve(conf Config) {
 
 	if idle := sc.hs.IdleTimeout(); idle > 0 {
 		sc.idleTimer = time.AfterFunc(idle, sc.onIdleTimer)
-		defer sc.idleTimer.Stop()
 	}
 
 	if conf.SendPingTimeout > 0 {
 		sc.readIdleTimeout = conf.SendPingTimeout
 		sc.readIdleTimer = time.AfterFunc(conf.SendPingTimeout, sc.onReadIdleTimer)
-		defer sc.readIdleTimer.Stop()
 	}
 
-	go sc.readFrames() // closed by defer sc.conn.Close above
+	go sc.readFrames() // closed by the conn.Close in sc.teardown
+
+	sc.settingsTimer = time.AfterFunc(firstSettingsTimeout, sc.onSettingsTimer)
 
-	settingsTimer := time.AfterFunc(firstSettingsTimeout, sc.onSettingsTimer)
-	defer settingsTimer.Stop()
+	sc.lastFrameTime = time.Now()
+	parked = sc.serveLoop()
+}
+
+// resumeServe runs the serve loop on a new goroutine for a connection
+// whose previous serve goroutine parked while the connection was idle.
+// It is started by beginServeSend, which precedes every send to the
+// serve loop's channels.
+func (sc *serverConn) resumeServe() {
+	// The serve loop's first incarnation has net/http's conn.serve
+	// above it to recover from panics; this one has nothing, so
+	// recover here to keep a panic from crashing the process.
+	defer func() {
+		if err := recover(); err != nil && err != ErrAbortHandler {
+			const size = 64 << 10
+			buf := make([]byte, size)
+			buf = buf[:runtime.Stack(buf, false)]
+			sc.logf("http2: panic serving %v: %v\n%s", sc.conn.RemoteAddr(), err, buf)
+		}
+	}()
+	sc.serveG.setOwner()
+	defer sc.notePanic()
+	parked := false
+	defer func() {
+		if !parked {
+			sc.teardown()
+		}
+	}()
+	parked = sc.serveLoop()
+}
 
-	lastFrameTime := time.Now()
-	loopNum := 0
+// teardown closes the connection and releases its resources.
+// It runs (on the serve goroutine) when the serve loop exits.
+//
+// Because the serve goroutine may park while the connection is idle
+// (making ServeConn return early), this is also responsible for the
+// cleanup that ServeConn's callers used to do after it returned:
+// unregistering the connection, canceling its base context, and
+// notifying the ServeConnOpts.OnClose hook.
+func (sc *serverConn) teardown() {
+	sc.serveG.check()
+	if t := sc.settingsTimer; t != nil {
+		t.Stop()
+	}
+	if t := sc.readIdleTimer; t != nil {
+		t.Stop()
+	}
+	if t := sc.idleTimer; t != nil {
+		t.Stop()
+	}
+	close(sc.doneServing) // unblocks handlers trying to send
+	sc.stopShutdownTimer()
+	sc.closeAllStreamsOnConnClose()
+	sc.conn.Close()
+	sc.srv.unregisterConn(sc)
+	sc.baseCtxCancel()
+	if f := sc.onClose; f != nil {
+		f()
+	}
+}
+
+// serveLoop is the serve goroutine's main loop, processing frames and
+// messages until the connection is done (returning false) or until it
+// has nothing runnable and the goroutine parks (returning true).
+//
+// After it parks, the serve loop is resumed on a new goroutine by the
+// next message sent to any of its channels: a frame from readFrames, a
+// handler write or body read, a timer firing, or a graceful shutdown
+// request. This way a connection with nothing to do only pins the
+// readFrames goroutine (plus any handler goroutines), whether it's
+// fully idle or its handlers are parked mid-long-poll.
+func (sc *serverConn) serveLoop() (parked bool) {
+	sc.serveG.check()
 	for {
-		loopNum++
+		sc.loopNum++
 		select {
 		case wr := <-sc.wantWriteFrameCh:
+			sc.endServeSend()
 			if se, ok := wr.write.(StreamError); ok {
 				sc.resetStream(se)
 				break
 			}
 			sc.writeFrame(wr)
 		case res := <-sc.wroteFrameCh:
+			sc.endServeSend()
 			sc.wroteFrame(res)
 		case res := <-sc.readFrameCh:
-			lastFrameTime = time.Now()
+			sc.endServeSend()
+			sc.lastFrameTime = time.Now()
 			// Process any written frames before reading new frames from the client since a
 			// written frame could have triggered a new stream to be started.
 			if sc.writingFrameAsync {
 				select {
 				case wroteRes := <-sc.wroteFrameCh:
+					sc.endServeSend()
 					sc.wroteFrame(wroteRes)
 				default:
 				}
 			}
 			if !sc.processFrameFromReader(res) {
-				return
+				return false
 			}
 			res.readMore()
-			if settingsTimer != nil {
-				settingsTimer.Stop()
-				settingsTimer = nil
+			if t := sc.settingsTimer; t != nil {
+				t.Stop()
+				sc.settingsTimer = nil
 			}
 		case m := <-sc.bodyReadCh:
+			sc.endServeSend()
 			sc.noteBodyRead(m.st, m.n)
 		case msg := <-sc.serveMsgCh:
+			sc.endServeSend()
 			switch v := msg.(type) {
 			case func(int):
-				v(loopNum) // for testing
+				v(sc.loopNum) // for testing
 			case *serverMessage:
 				switch v {
 				case settingsTimerMsg:
 					sc.logf("timeout waiting for SETTINGS frames from %v", sc.conn.RemoteAddr())
-					return
+					return false
 				case idleTimerMsg:
 					sc.vlogf("connection is idle")
 					sc.goAway(ErrCodeNo)
 				case readIdleTimerMsg:
-					sc.handlePingTimer(lastFrameTime)
+					sc.handlePingTimer(sc.lastFrameTime)
 				case shutdownTimerMsg:
 					sc.vlogf("GOAWAY close timer fired; closing conn from %v", sc.conn.RemoteAddr())
-					return
+					return false
 				case gracefulShutdownMsg:
 					sc.startGracefulShutdownInternal()
 				case handlerDoneMsg:
@@ -911,7 +1039,7 @@ func (sc *serverConn) serve(conf Config) {
 		// run out of memory.
 		if sc.queuedControlFrames > maxQueuedControlFrames {
 			sc.vlogf("http2: too many control frames in send queue, closing connection")
-			return
+			return false
 		}
 
 		// Start the shutdown timer after sending a GOAWAY. When sending GOAWAY
@@ -922,9 +1050,82 @@ func (sc *serverConn) serve(conf Config) {
 		if sentGoAway && sc.shutdownTimer == nil && (sc.goAwayCode != ErrCodeNo || gracefulShutdownComplete) {
 			sc.shutDownIn(goAwayTimeout)
 		}
+
+		// If the connection is idle, park: let this goroutine exit and
+		// have the next message revive the loop on a new goroutine.
+		if sc.canPark() && sc.tryPark() {
+			return true
+		}
 	}
 }
 
+// canPark reports whether the serve loop has nothing runnable and can
+// park: nothing being written or unflushed, nothing poppable in the
+// write scheduler, and no shutdown in progress.
+//
+// Open streams and running handlers do not prevent parking: everything
+// that could next give the serve loop work (a frame from the peer, a
+// handler write or body read, a timer firing, a graceful shutdown
+// request) reaches it through a channel send preceded by
+// beginServeSend, which revives the loop. A connection whose handlers
+// are parked mid-long-poll therefore doesn't pin a serve goroutine
+// either.
+//
+// The write side is known to be drained because every state change
+// ends in scheduleFrameWrite, which (except in a GOAWAY with an error
+// code, and inGoAway is false here) pops the write scheduler until Pop
+// fails and then flushes, so writingFrame, writingFrameAsync, and
+// needsFrameFlush are all false only once nothing was left to write.
+// With queuedControlFrames == 0, anything still in the scheduler after
+// a failed Pop is a DATA frame blocked on flow control (non-DATA
+// frames are always consumed whole; see FrameWriteRequest.Consume),
+// and only an incoming WINDOW_UPDATE or SETTINGS frame, which arrives
+// via readFrames and revives the loop, can unblock it.
+func (sc *serverConn) canPark() bool {
+	sc.serveG.check()
+	return sc.serveParkingEnabled &&
+		!sc.writingFrame &&
+		!sc.writingFrameAsync &&
+		!sc.needsFrameFlush &&
+		!sc.needToSendGoAway &&
+		!sc.needToSendSettingsAck &&
+		!sc.inGoAway &&
+		sc.queuedControlFrames == 0
+}
+
+// http2serveparking, if "0", disables parking the serve goroutine
+// while a connection has nothing to do, restoring the old behavior of
+// keeping a goroutine parked for the lifetime of each connection.
+//
+// It exists only as a temporary safety measure for the change that
+// parks the serve goroutine. The # prefix marks it as undocumented,
+// so it is not listed in doc/godebug.md or runtime/metrics.
+//
+// TODO: remove this setting; see #81524.
+var http2serveparking = godebug.New("#http2serveparking")
+
+// tryPark attempts to mark the serve goroutine as parked, so that the
+// next beginServeSend revives the serve loop on a new goroutine. It
+// fails if a sender has already begun sending a message to one of the
+// serve loop's channels, in which case the caller must keep serving to
+// receive it.
+func (sc *serverConn) tryPark() bool {
+	sc.serveG.check()
+	sc.parkMu.Lock()
+	defer sc.parkMu.Unlock()
+	if sc.serveSendsPending != 0 {
+		return false
+	}
+	// With no sends pending, the channels must all be empty too, but
+	// parking with a queued message would stall the connection, so be
+	// defensive.
+	if len(sc.readFrameCh)+len(sc.wantWriteFrameCh)+len(sc.wroteFrameCh)+len(sc.bodyReadCh)+len(sc.serveMsgCh) != 0 {
+		return false
+	}
+	sc.parked = true
+	return true
+}
+
 func (sc *serverConn) handlePingTimer(lastFrameReadTime time.Time) {
 	if sc.pingSent {
 		sc.logf("timeout waiting for PING response")
@@ -971,11 +1172,45 @@ func (sc *serverConn) onIdleTimer()     { sc.sendServeMsg(idleTimerMsg) }
 func (sc *serverConn) onReadIdleTimer() { sc.sendServeMsg(readIdleTimerMsg) }
 func (sc *serverConn) onShutdownTimer() { sc.sendServeMsg(shutdownTimerMsg) }
 
+// beginServeSend notes that the caller is about to send a message to
+// one of the serve loop's channels, reviving the serve goroutine if it
+// has parked. Every send to those channels must be preceded by a call
+// to beginServeSend. The send is counted as pending until the serve
+// loop receives the message, or until a sender that bailed out on
+// another select case gives up; both call endServeSend.
+func (sc *serverConn) beginServeSend() {
+	if !sc.serveParkingEnabled {
+		// The serve loop never parks, so there is nothing to count
+		// or revive, and no lock is needed.
+		return
+	}
+	sc.parkMu.Lock()
+	defer sc.parkMu.Unlock()
+	sc.serveSendsPending++
+	if sc.parked {
+		sc.parked = false
+		go sc.resumeServeFunc()
+	}
+}
+
+// endServeSend balances an earlier beginServeSend, either because the
+// serve loop received the message or because the sender never sent it.
+func (sc *serverConn) endServeSend() {
+	if !sc.serveParkingEnabled {
+		return
+	}
+	sc.parkMu.Lock()
+	defer sc.parkMu.Unlock()
+	sc.serveSendsPending--
+}
+
 func (sc *serverConn) sendServeMsg(msg any) {
 	sc.serveG.checkNotOn() // NOT
+	sc.beginServeSend()
 	select {
 	case sc.serveMsgCh <- msg:
 	case <-sc.doneServing:
+		sc.endServeSend()
 	}
 }
 
@@ -1070,12 +1305,14 @@ func (sc *serverConn) writeDataFromHandler(stream *stream, data []byte, endStrea
 // goroutine, call writeFrame instead.
 func (sc *serverConn) writeFrameFromHandler(wr FrameWriteRequest) error {
 	sc.serveG.checkNotOn() // NOT
+	sc.beginServeSend()
 	select {
 	case sc.wantWriteFrameCh <- wr:
 		return nil
 	case <-sc.doneServing:
 		// Serve loop is gone.
 		// Client has closed their connection to the server.
+		sc.endServeSend()
 		return errClientDisconnected
 	}
 }
@@ -2411,9 +2648,11 @@ type bodyReadMsg struct {
 func (sc *serverConn) noteBodyReadFromHandler(st *stream, n int, err error) {
 	sc.serveG.checkNotOn() // NOT on
 	if n > 0 {
+		sc.beginServeSend()
 		select {
 		case sc.bodyReadCh <- bodyReadMsg{st, n}:
 		case <-sc.doneServing:
+			sc.endServeSend()
 		}
 	}
 }
@@ -3088,10 +3327,13 @@ func (w *responseWriter) Push(target, method string, header Header) error {
 		done:   getErrChan(),
 	}
 
+	sc.beginServeSend()
 	select {
 	case <-sc.doneServing:
+		sc.endServeSend()
 		return errClientDisconnected
 	case <-st.cw:
+		sc.endServeSend()
 		return errStreamClosed
 	case sc.serveMsgCh <- msg:
 	}
diff --git a/src/net/http/internal/http2/server_internal_test.go b/src/net/http/internal/http2/server_internal_test.go
index 379b5de46dfc40..83487cfd51c180 100644
--- a/src/net/http/internal/http2/server_internal_test.go
+++ b/src/net/http/internal/http2/server_internal_test.go
@@ -58,9 +58,8 @@ func TestCheckValidHTTP2Request(t *testing.T) {
 func TestCanonicalHeaderCacheGrowth(t *testing.T) {
 	for _, size := range []int{1, (1 << 20) - 10} {
 		base := strings.Repeat("X", size)
-		sc := &serverConn{
-			serveG: newGoroutineLock(),
-		}
+		sc := &serverConn{}
+		sc.serveG.setOwner()
 		count := 0
 		added := 0
 		for added < 10*maxCachedCanonicalHeadersKeysSize {
diff --git a/src/net/http/internal/http2/server_test.go b/src/net/http/internal/http2/server_test.go
index f15ad9e666cb10..9358d3bb33601a 100644
--- a/src/net/http/internal/http2/server_test.go
+++ b/src/net/http/internal/http2/server_test.go
@@ -421,15 +421,21 @@ func (st *serverTester) Close() {
 // frames may be sent.
 func (st *serverTester) greet() {
 	st.t.Helper()
-	st.greetAndCheckSettings(func(Setting) error { return nil })
+	st.greetAndCheckSettings(nil, func(Setting) error { return nil })
 }
 
-func (st *serverTester) greetAndCheckSettings(checkSetting func(s Setting) error) {
+// greetAndCheckSettings is like greet, but the client sends the
+// provided settings in its initial SETTINGS frame, and the
+// checkSetting callback, if non-nil, is run for each setting in the
+// server's initial SETTINGS frame.
+func (st *serverTester) greetAndCheckSettings(settings []Setting, checkSetting func(s Setting) error) {
 	st.t.Helper()
 	st.writePreface()
-	st.writeSettings()
+	st.writeSettings(settings...)
 	st.sync()
-	readFrame[*SettingsFrame](st.t, st).ForeachSetting(checkSetting)
+	if f := readFrame[*SettingsFrame](st.t, st); checkSetting != nil {
+		f.ForeachSetting(checkSetting)
+	}
 	st.writeSettingsAck()
 
 	// The initial WINDOW_UPDATE and SETTINGS ACK can come in any order.
@@ -2809,7 +2815,7 @@ func testServer_MaxDecoderHeaderTableSize(t *testing.T) {
 	defer st.Close()
 
 	var advHeaderTableSize *uint32
-	st.greetAndCheckSettings(func(s Setting) error {
+	st.greetAndCheckSettings(nil, func(s Setting) error {
 		switch s.ID {
 		case SettingHeaderTableSize:
 			advHeaderTableSize = &s.Val
@@ -2932,7 +2938,7 @@ func testServerDoS_MaxHeaderListSize(t *testing.T) {
 	// shake hands
 	frameSize := DefaultMaxReadFrameSize
 	var advHeaderListSize *uint32
-	st.greetAndCheckSettings(func(s Setting) error {
+	st.greetAndCheckSettings(nil, func(s Setting) error {
 		switch s.ID {
 		case SettingMaxFrameSize:
 			if s.Val < MinMaxFrameSize {
@@ -3719,6 +3725,206 @@ func testServerIdleTimeout_AfterRequest(t *testing.T) {
 	st.wantGoAway(1, ErrCodeNo)
 }
 
+// wantParked asserts the state of the connection's serve goroutine.
+func (st *serverTester) wantParked(want bool) {
+	st.t.Helper()
+	st.sync()
+	if got := st.sc.TestServeParked(); got != want {
+		st.t.Errorf("serve goroutine parked = %v, want %v", got, want)
+	}
+}
+
+func TestServerParksWhenIdle(t *testing.T) { synctest.Test(t, testServerParksWhenIdle) }
+func testServerParksWhenIdle(t *testing.T) {
+	unblock := make(chan struct{})
+	st := newServerTester(t, func(w http.ResponseWriter, r *http.Request) {
+		<-unblock
+	})
+	defer st.Close()
+
+	// The serve goroutine parks once the connection is established
+	// and idle.
+	st.greet()
+	st.wantParked(true)
+
+	// It parks even while a handler is running, as long as the
+	// handler isn't giving it any work.
+	st.bodylessReq1()
+	st.wantParked(true)
+
+	// It parks again once the request is done.
+	close(unblock)
+	st.wantHeaders(wantHeader{
+		streamID:  1,
+		endStream: true,
+	})
+	st.wantParked(true)
+
+	// A PING on the parked connection is acked as usual.
+	st.writePing(false, [8]byte{1, 2, 3, 4, 5, 6, 7, 8})
+	st.wantFrameType(FramePing)
+	st.wantParked(true)
+}
+
+func TestServerParksDuringLongPoll(t *testing.T) { synctest.Test(t, testServerParksDuringLongPoll) }
+func testServerParksDuringLongPoll(t *testing.T) {
+	// An SSE-style handler: write an event, flush, park for a long
+	// time, write another event.
+	events := make(chan string)
+	st := newServerTester(t, func(w http.ResponseWriter, r *http.Request) {
+		for ev := range events {
+			io.WriteString(w, ev)
+			w.(http.Flusher).Flush()
+		}
+	})
+	defer st.Close()
+
+	st.greet()
+	st.bodylessReq1()
+
+	events <- "hello"
+	st.wantHeaders(wantHeader{
+		streamID:  1,
+		endStream: false,
+	})
+	st.wantData(wantData{
+		streamID:  1,
+		endStream: false,
+		data:      []byte("hello"),
+	})
+
+	// The handler is parked between events, and so is the serve
+	// goroutine, even though the stream stays open for however long
+	// the long poll lasts.
+	st.wantParked(true)
+	st.advance(24 * time.Hour)
+	st.wantParked(true)
+
+	// The next event revives the serve goroutine.
+	events <- "world"
+	st.wantData(wantData{
+		streamID:  1,
+		endStream: false,
+		data:      []byte("world"),
+	})
+	st.wantParked(true)
+
+	// So does the handler finishing.
+	close(events)
+	st.wantData(wantData{
+		streamID:  1,
+		endStream: true,
+		data:      []byte{},
+	})
+	st.wantParked(true)
+}
+
+func TestServerParkedGracefulShutdown(t *testing.T) {
+	synctest.Test(t, testServerParkedGracefulShutdown)
+}
+func testServerParkedGracefulShutdown(t *testing.T) {
+	st := newServerTester(t, nil)
+	defer st.Close()
+
+	st.greet()
+	st.wantParked(true)
+
+	// Server.Shutdown revives the parked serve goroutine, which sends
+	// a GOAWAY and tears the connection down; it must not park again
+	// mid-shutdown.
+	st.sc.StartGracefulShutdown()
+	st.wantGoAway(0, ErrCodeNo)
+	st.wantParked(false)
+}
+
+func TestServerParkedIdleTimeout(t *testing.T) { synctest.Test(t, testServerParkedIdleTimeout) }
+func testServerParkedIdleTimeout(t *testing.T) {
+	const idleTimeout = 1 * time.Second
+	st := newServerTester(t, nil, func(s *http.Server) {
+		s.IdleTimeout = idleTimeout
+	})
+	defer st.Close()
+
+	// The idle timeout must still fire on a parked connection.
+	st.greet()
+	st.wantParked(true)
+	st.advance(idleTimeout)
+	st.wantGoAway(0, ErrCodeNo)
+}
+
+func TestServerParkingDisabledByGODEBUG(t *testing.T) {
+	t.Setenv("GODEBUG", "http2serveparking=0")
+	synctest.Test(t, testServerParkingDisabledByGODEBUG)
+}
+func testServerParkingDisabledByGODEBUG(t *testing.T) {
+	st := newServerTester(t, func(w http.ResponseWriter, r *http.Request) {})
+	defer st.Close()
+
+	st.greet()
+	st.wantParked(false)
+	st.bodylessReq1()
+	st.wantHeaders(wantHeader{
+		streamID:  1,
+		endStream: true,
+	})
+	st.wantParked(false)
+}
+
+func TestServerParksWithFlowControlledData(t *testing.T) {
+	synctest.Test(t, testServerParksWithFlowControlledData)
+}
+func testServerParksWithFlowControlledData(t *testing.T) {
+	// A handler whose response exceeds the client's stream flow
+	// control window. The server sends as much DATA as the window
+	// allows and then parks, with the rest of the response queued in
+	// the write scheduler, blocked on flow control. The client's
+	// WINDOW_UPDATE revives the parked serve goroutine to write it.
+	const window = 100
+	const responseSize = 250
+	rest := responseSize - window
+	st := newServerTester(t, func(w http.ResponseWriter, r *http.Request) {
+		w.(http.Flusher).Flush()
+		io.WriteString(w, strings.Repeat("a", responseSize))
+	})
+	defer st.Close()
+
+	st.greetAndCheckSettings(
+		[]Setting{{SettingInitialWindowSize, window}},
+		nil,
+	)
+	st.bodylessReq1()
+
+	// The server sends the response headers and the first window bytes
+	// of DATA, then the write scheduler blocks the rest on the
+	// stream's flow control window.
+	st.wantHeaders(wantHeader{
+		streamID:  1,
+		endStream: false,
+	})
+	st.wantData(wantData{
+		streamID:  1,
+		endStream: false,
+		size:      window,
+	})
+
+	// The serve goroutine parks even though its write scheduler still
+	// holds the blocked DATA, and it stays parked, because only an
+	// incoming WINDOW_UPDATE can unblock it.
+	st.wantParked(true)
+	st.advance(24 * time.Hour)
+	st.wantParked(true)
+
+	// The client's WINDOW_UPDATE revives the serve goroutine, which
+	// writes the rest of the response and ends the stream.
+	st.writeWindowUpdate(1, uint32(rest))
+	st.wantData(wantData{
+		streamID:  1,
+		endStream: true,
+		data:      []byte(strings.Repeat("a", rest)),
+	})
+	st.wantParked(true)
+}
+
 // grpc-go closes the Request.Body currently with a Read.
 // Verify that it doesn't race.
 // See https://github.com/grpc/grpc-go/pull/938
@@ -4863,7 +5069,7 @@ func testServerSettingNoRFC7540Priorities(t *testing.T) {
 	defer st.Close()
 
 	var gotNoRFC7540Setting bool
-	st.greetAndCheckSettings(func(s Setting) error {
+	st.greetAndCheckSettings(nil, func(s Setting) error {
 		if s.ID != SettingNoRFC7540Priorities {
 			return nil
 		}
diff --git a/src/net/http/omithttp2.go b/src/net/http/omithttp2.go
index 5745cd9797244e..775ee60093da95 100644
--- a/src/net/http/omithttp2.go
+++ b/src/net/http/omithttp2.go
@@ -20,7 +20,7 @@ const noHTTP2 = "no bundled HTTP/2" // should never see this
 
 func (s *Server) configureHTTP2()                               {}
 func (s *Server) setHTTP2Config(conf http2ExternalServerConfig) {}
-func (s *Server) serveHTTP2Conn(ctx context.Context, nc net.Conn, h Handler, sawClientPreface bool, upgradeReq *Request, settings []byte) {
+func (s *Server) serveHTTP2Conn(ctx context.Context, nc net.Conn, h Handler, sawClientPreface bool, upgradeReq *Request, settings []byte, onClose func()) {
 }
 
 func (t *Transport) configureHTTP2(protocols Protocols) {}
diff --git a/src/net/http/server.go b/src/net/http/server.go
index 5cd28fefe0782a..eef90e82dc4ab1 100644
--- a/src/net/http/server.go
+++ b/src/net/http/server.go
@@ -306,6 +306,12 @@ type conn struct {
 	// by a Handler with the Hijacker interface.
 	// It is guarded by mu.
 	hijackedv bool
+
+	// http2HandedOff is whether the connection has been handed off to
+	// the HTTP/2 server, which then owns closing the connection and
+	// running the final ConnState hook, possibly after (*conn).serve
+	// has returned. It is only accessed by the (*conn).serve goroutine.
+	http2HandedOff bool
 }
 
 func (c *conn) hijacked() bool {
@@ -2014,7 +2020,7 @@ func (c *conn) serve(ctx context.Context) {
 			inFlightResponse.cancelCtx()
 			inFlightResponse.disableWriteContinue(true)
 		}
-		if !c.hijacked() {
+		if !c.hijacked() && !c.http2HandedOff {
 			if inFlightResponse != nil {
 				inFlightResponse.conn.r.abortPendingRead()
 				inFlightResponse.reqBody.Close()
@@ -2072,7 +2078,7 @@ func (c *conn) serve(ctx context.Context) {
 			// closing such connections. See issue https://golang.org/issue/39776.
 			c.setState(c.rwc, StateActive, skipHooks)
 			const sawClientPreface = false
-			c.server.serveHTTP2Conn(ctx, c.rwc, serverHandler{c.server}, sawClientPreface, nil, nil)
+			c.serveHTTP2(ctx, sawClientPreface)
 			return
 		}
 		tlsConn, tlsConnOK := c.rwc.(*tls.Conn)
@@ -2352,7 +2358,7 @@ func (c *conn) maybeServeUnencryptedHTTP2(ctx context.Context) bool {
 	c.setState(c.rwc, StateActive, skipHooks)
 	if c.server.h2 != nil {
 		const sawClientPreface = true
-		c.server.serveHTTP2Conn(ctx, c.rwc, serverHandler{c.server}, sawClientPreface, nil, nil)
+		c.serveHTTP2(ctx, sawClientPreface)
 	} else {
 		c.rwc.SetReadDeadline(time.Time{})
 		c.rwc.SetWriteDeadline(time.Time{})
@@ -2362,6 +2368,19 @@ func (c *conn) maybeServeUnencryptedHTTP2(ctx context.Context) bool {
 	return true
 }
 
+// serveHTTP2 hands the connection off to the HTTP/2 server, which owns
+// the connection from here on: it closes the connection and runs the
+// final ConnState hook when it's done, possibly after this function has
+// returned, since an idle HTTP/2 connection doesn't hold onto a
+// goroutine.
+func (c *conn) serveHTTP2(ctx context.Context, sawClientPreface bool) {
+	c.http2HandedOff = true
+	c.server.serveHTTP2Conn(ctx, c.rwc, serverHandler{c.server}, sawClientPreface, nil, nil, func() {
+		c.close()
+		c.setState(c.rwc, StateClosed, runHooks)
+	})
+}
+
 func (w *response) sendExpectationFailed() {
 	// TODO(bradfitz): let ServeHTTP handlers handle
 	// requests with non-standard expectation[s]? Seems