fix(frontend): promptly cancel requests after client disconnect

pkg/frontend/mysql_buffer.go

@@ -202,6 +202,7 @@ func (block *MemBlock) Adjust() {
 type Conn struct {
 	id                    uint64
 	conn                  net.Conn
+	livenessProbe         *socketLivenessProbe
 	localAddr, remoteAddr string
 	sequenceId            uint8
 	header                [4]byte
@@ -311,6 +312,7 @@ func NewIOSessionWithOptions(
 
 	c := &Conn{
 		conn:                  conn,
+		livenessProbe:         newSocketLivenessProbe(conn),
 		localAddr:             conn.LocalAddr().String(),
 		remoteAddr:            conn.RemoteAddr().String(),
 		fixBuf:                MemBlock{},
@@ -358,6 +360,7 @@ func (c *Conn) GetSequenceID() uint8 {
 
 func (c *Conn) UseConn(conn net.Conn) {
 	c.conn = conn
+	c.livenessProbe = newSocketLivenessProbe(conn)
 }
 
 func (c *Conn) freeDynamicBuffUnsafe() {

pkg/frontend/routine_manager.go

@@ -20,7 +20,6 @@ import (
 	"crypto/x509"
 	"fmt"
 	"math"
-	"net"
 	"os"
 	"sync"
 	"time"
@@ -40,6 +39,8 @@ import (
 
 type RoutineManager struct {
 	mu                     sync.RWMutex
+	disconnectProbeMu      sync.Mutex
+	disconnectProbeScratch []activeClientRequest
 	ctx                    context.Context
 	clients                map[*Conn]*Routine
 	workerWG               sync.WaitGroup
@@ -73,6 +74,11 @@ type activeClientRequest struct {
 	routine *Routine
 }
 
+var clientDisconnectProbeErrorEvent = logutil.Event{
+	Name:    "frontend.client-disconnect-probe.error",
+	Message: "failed to probe active client connection",
+}
+
 func NewKillRecord(killtime time.Time, version uint64) KillRecord {
 	return KillRecord{
 		killTime: killtime,
@@ -195,11 +201,14 @@ func (rm *RoutineManager) getRoutineByConnID(id uint32) *Routine {
 	return nil
 }
 
-func (rm *RoutineManager) longRunningRequests(now time.Time, minimum time.Duration) []activeClientRequest {
+func (rm *RoutineManager) appendLongRunningRequests(
+	requests []activeClientRequest,
+	now time.Time,
+	minimum time.Duration,
+) []activeClientRequest {
 	rm.mu.RLock()
 	defer rm.mu.RUnlock()
 	nowValue := clientRequestClockValue(now)
-	var requests []activeClientRequest
 	for conn, routine := range rm.clients {
 		if conn != nil && routine != nil && routine.requestRunningLongerThan(nowValue, minimum) {
 			requests = append(requests, activeClientRequest{conn: conn, routine: routine})
@@ -211,15 +220,27 @@ func (rm *RoutineManager) longRunningRequests(now time.Time, minimum time.Durati
 func (rm *RoutineManager) cancelDisconnectedRequests(
 	now time.Time,
 	minimum time.Duration,
-	probe func(net.Conn) (bool, error),
+	probe func(*Conn) (bool, error),
 ) {
 	if probe == nil {
 		return
 	}
-	for _, request := range rm.longRunningRequests(now, minimum) {
-		closed, err := probe(request.conn.RawConn())
+	rm.disconnectProbeMu.Lock()
+	defer rm.disconnectProbeMu.Unlock()
+	requests := rm.appendLongRunningRequests(rm.disconnectProbeScratch[:0], now, minimum)
+	defer func() {
+		clear(requests)
+		rm.disconnectProbeScratch = requests[:0]
+	}()
+	for _, request := range requests {
+		closed, err := probe(request.conn)
 		if err != nil {
-			logutil.Debugf("failed to probe active client connection %s: %v", request.conn.RemoteAddress(), err)
+			clientDisconnectProbeErrorEvent.DebugLazy(func() []zap.Field {
+				return []zap.Field{
+					zap.String("connection", request.conn.RemoteAddress()),
+					zap.Error(err),
+				}
+			})
 			continue
 		}
 		if !closed {

pkg/frontend/routine_manager_test.go

@@ -342,9 +342,9 @@ func TestRoutineManagerCancelDisconnectedLongRunningRequests(t *testing.T) {
 	}}
 
 	probes := 0
-	rm.cancelDisconnectedRequests(now, grace, func(conn net.Conn) (bool, error) {
+	rm.cancelDisconnectedRequests(now, grace, func(conn *Conn) (bool, error) {
 		probes++
-		return conn == longServer, nil
+		return conn.RawConn() == longServer, nil
 	})
 
 	require.Equal(t, 1, probes, "only requests beyond the grace period should be probed")
@@ -359,13 +359,44 @@ func TestRoutineManagerCancelDisconnectedLongRunningRequests(t *testing.T) {
 	default:
 	}
 
-	rm.cancelDisconnectedRequests(now, grace, func(net.Conn) (bool, error) {
+	rm.cancelDisconnectedRequests(now, grace, func(*Conn) (bool, error) {
 		probes++
 		return true, nil
 	})
 	require.Equal(t, 1, probes, "a routine already closing should not be probed again")
 }
 
+func TestClientDisconnectProbePolicyCoversNewRequests(t *testing.T) {
+	now := time.Now()
+	serverConn, clientConn := net.Pipe()
+	t.Cleanup(func() {
+		_ = serverConn.Close()
+		_ = clientConn.Close()
+	})
+
+	routine := NewRoutine(context.Background(), &testMysqlWriter{}, &config.FrontendParameters{})
+	t.Cleanup(routine.cancelRoutineFunc)
+	routine.requestStartedAt.Store(clientRequestClockValue(now))
+	requestCtx, cancelRequest := context.WithCancel(context.Background())
+	t.Cleanup(cancelRequest)
+	routine.setCancelRequestFunc(cancelRequest)
+
+	conn := &Conn{conn: serverConn, remoteAddr: "new-request"}
+	rm := &RoutineManager{clients: map[*Conn]*Routine{conn: routine}}
+	probes := 0
+	rm.cancelDisconnectedRequests(now, clientDisconnectProbeGrace, func(*Conn) (bool, error) {
+		probes++
+		return true, nil
+	})
+
+	require.Equal(t, 1, probes, "a new active request must be probed without an age grace period")
+	select {
+	case <-requestCtx.Done():
+	case <-time.After(time.Second):
+		t.Fatal("a disconnected new request was not canceled")
+	}
+}
+
 func TestRoutineManagerProbeErrorDoesNotCancelRequest(t *testing.T) {
 	now := time.Now()
 	serverConn, clientConn := net.Pipe()
@@ -383,7 +414,7 @@ func TestRoutineManagerProbeErrorDoesNotCancelRequest(t *testing.T) {
 
 	conn := &Conn{conn: serverConn}
 	rm := &RoutineManager{clients: map[*Conn]*Routine{conn: routine}}
-	rm.cancelDisconnectedRequests(now, 30*time.Second, func(net.Conn) (bool, error) {
+	rm.cancelDisconnectedRequests(now, 30*time.Second, func(*Conn) (bool, error) {
 		return false, errors.New("probe failed")
 	})
 
@@ -436,7 +467,8 @@ func BenchmarkRoutineManagerLongRunningRequests(b *testing.B) {
 			b.ReportAllocs()
 			b.ResetTimer()
 			for i := 0; i < b.N; i++ {
-				_ = rm.longRunningRequests(now, 30*time.Second)
+				requests := rm.appendLongRunningRequests(nil, now, 30*time.Second)
+				clear(requests)
 			}
 		})
 	}

pkg/frontend/server.go

@@ -64,8 +64,10 @@ var initConnectionID uint32 = 1000
 var ConnIDAllocKey = "____server_conn_id"
 
 const (
-	clientDisconnectProbeInterval = 5 * time.Second
-	clientDisconnectProbeGrace    = 30 * time.Second
+	// The request handler owns the connection read loop while a statement is
+	// executing, so probe every active request from the first monitor tick.
+	clientDisconnectProbeInterval = time.Second
+	clientDisconnectProbeGrace    = 0
 )
 
 // MOServer MatrixOne Server

pkg/frontend/socket_liveness_alloc_unix_test.go

@@ -0,0 +1,47 @@
+// Copyright 2026 Matrix Origin
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//go:build (darwin || linux) && !race
+
+package frontend
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+// The race runtime adds bookkeeping allocations to the syscall callback path.
+// Functional live-socket behavior remains covered under -race by the tests in
+// socket_liveness_unix_test.go.
+func TestConnectionPeerClosedLivePathDoesNotAllocate(t *testing.T) {
+	server, client := tcpConnectionPair(t)
+	t.Cleanup(func() {
+		_ = server.Close()
+		_ = client.Close()
+	})
+	conn := &Conn{
+		conn:          server,
+		livenessProbe: newSocketLivenessProbe(server),
+	}
+
+	var closed bool
+	var err error
+	allocs := testing.AllocsPerRun(100, func() {
+		closed, err = connectionPeerClosed(conn)
+	})
+	require.NoError(t, err)
+	require.False(t, closed)
+	require.Zero(t, allocs)
+}

pkg/frontend/socket_liveness_other.go

@@ -18,6 +18,18 @@ package frontend
 
 import "net"
 
-func connectionPeerClosed(net.Conn) (bool, error) {
+type socketLivenessProbe struct{}
+
+func newSocketLivenessProbe(net.Conn) *socketLivenessProbe {
+	return &socketLivenessProbe{}
+}
+
+func (*socketLivenessProbe) connectionPeerClosed() (bool, error) {
 	return false, nil
 }
+
+func rawConnectionPeerClosed(net.Conn) (bool, error) {
+	return false, nil
+}
+
+func connectionPeerClosed(*Conn) (bool, error) { return false, nil }