fix: contain session event callback panics

transport/getty.go

@@ -78,12 +78,18 @@ type EventListener interface {
 	OnOpen(Session) error
 
 	// OnClose invoked when session closed.
+	// Getty recovers and logs panics, then continues resource cleanup.
 	OnClose(Session)
 
 	// OnError invoked when got error.
+	// Getty recovers and logs panics, then continues OnClose and cleanup.
 	OnError(Session, error)
 
-	// OnCron invoked periodically, its period can be set by (Session)SetCronPeriod
+	// OnCron is invoked periodically; its period can be set by
+	// (Session).SetCronPeriod. Getty recovers and logs panics and closes the
+	// faulty session, including when a task pool is configured. A client session
+	// may then follow the normal reconnect behavior. Other sessions and timers
+	// keep running.
 	OnCron(Session)
 
 	// OnMessage invoked when getty received a package. Pls attention that do not handle long time

transport/session.go

@@ -924,6 +924,32 @@ func (s *session) WriteBytesArray(pkgs ...[]byte) (int, error) {
 	return wlg, nil
 }
 
+// invokeEventCallback isolates callbacks invoked by cleanup or the timer wheel.
+// Each call needs its own recovery: handlePackage's recovery cannot catch a
+// second panic raised later in the same defer.
+func (s *session) invokeEventCallback(name string, f func()) (panicked bool) {
+	completed := false
+	defer func() {
+		if completed {
+			return
+		}
+		// With GODEBUG=panicnil=1, recover returns nil for panic(nil) even
+		// though it still stops the panic. Track completion separately so that
+		// such a panic is logged and reported to the caller as well.
+		r := recover()
+		panicked = true
+		const size = 64 << 10
+		rBuf := make([]byte, size)
+		rBuf = rBuf[:runtime.Stack(rBuf, false)]
+		// The pointer identifies the session even after stop has changed
+		// its token to session-closed for OnError/OnClose.
+		log.Errorf("[session.%s] panic session %p (%s): err=%v\n%s", name, s, s.sessionToken(), r, rBuf)
+	}()
+	s.invokeCallback(f)
+	completed = true
+	return false
+}
+
 // heartbeat schedules OnCron only while the captured session lifecycle is open.
 func heartbeat(_ gxtime.TimerID, _ time.Time, arg any) error {
 	ctx, _ := arg.(*heartbeatContext)
@@ -954,7 +980,17 @@ func heartbeat(_ gxtime.TimerID, _ time.Time, arg any) error {
 			}
 		}
 
-		ss.invokeCallback(func() { listener.OnCron(ss) })
+		if ss.invokeEventCallback("OnCron", func() { listener.OnCron(ss) }) {
+			// A faulty cron closes only its session, on both the timer-wheel
+			// and task-pool paths. Do not wait for this lifecycle task or
+			// detach resources that the read loop still owns. Close the raw
+			// transport to unblock reads regardless of their configured timeout;
+			// handlePackage will deliver shutdown events and perform gc.
+			ss.Close()
+			if rawConn := netConnOf(conn); rawConn != nil {
+				_ = rawConn.Close()
+			}
+		}
 	}
 
 	// if enable task pool, run @f asynchronously.
@@ -1116,11 +1152,11 @@ func (s *session) handlePackage() {
 		if err != nil {
 			log.Errorf("%s, [session.handlePackage] error:%+v", s.sessionToken(), perrors.WithStack(err))
 			if s != nil && s.listener != nil {
-				s.invokeCallback(func() { s.listener.OnError(s, err) })
+				s.invokeEventCallback("OnError", func() { s.listener.OnError(s, err) })
 			}
 		}
 		if s.listener != nil {
-			s.invokeCallback(func() { s.listener.OnClose(s) })
+			s.invokeEventCallback("OnClose", func() { s.listener.OnClose(s) })
 		}
 	}()
 

transport/session_event_panic_test.go

@@ -0,0 +1,299 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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.
+ */
+
+package getty
+
+import (
+	"fmt"
+	"net"
+	"os"
+	"os/exec"
+	"strings"
+	"sync"
+	"testing"
+	"time"
+)
+
+import (
+	gxsync "github.com/dubbogo/gost/sync"
+)
+
+const callbackPanicEnv = "GETTY_TEST_EVENT_PANIC"
+
+// A parent-goroutine recover cannot catch a panic escaping Getty's read loop
+// or shared timer wheel. Each regression therefore runs in its own process.
+func TestEventCallbackPanicContained(t *testing.T) {
+	for _, stage := range []string{"close", "error", "error-close", "cron", "cron-nil", "pooled-cron"} {
+		t.Run(stage, func(t *testing.T) {
+			cmd := exec.Command(os.Args[0], "-test.run=^TestEventCallbackPanicChild$", "-test.timeout=15s")
+			cmd.Env = make([]string, 0, len(os.Environ())+2)
+			for _, env := range os.Environ() {
+				if stage == "cron-nil" && strings.HasPrefix(env, "GODEBUG=") {
+					continue
+				}
+				cmd.Env = append(cmd.Env, env)
+			}
+			cmd.Env = append(cmd.Env, callbackPanicEnv+"="+stage)
+			if stage == "cron-nil" {
+				cmd.Env = append(cmd.Env, "GODEBUG=panicnil=1")
+			}
+			output, err := cmd.CombinedOutput()
+			if err != nil {
+				t.Fatalf("callback panic regression failed: %v\n%s", err, output)
+			}
+			callbacks := []string{"OnClose"}
+			switch stage {
+			case "error":
+				callbacks = []string{"OnError"}
+			case "error-close":
+				callbacks = []string{"OnError", "OnClose"}
+			case "cron", "cron-nil", "pooled-cron":
+				callbacks = []string{"OnCron"}
+			}
+			for _, callback := range callbacks {
+				wants := []string{"[session." + callback + "] panic session", "session_event_panic_test.go:"}
+				if stage == "cron-nil" {
+					wants = append(wants, "err=<nil>")
+				} else {
+					wants = append(wants, "test "+callback+" panic")
+				}
+				for _, want := range wants {
+					if !strings.Contains(string(output), want) {
+						t.Errorf("missing panic context or stack %q in output:\n%s", want, output)
+					}
+				}
+			}
+		})
+	}
+}
+
+// eventPanicConn records transport closure independently of Session.Reset.
+type eventPanicConn struct {
+	net.Conn
+	closed chan struct{}
+	once   sync.Once
+}
+
+func (c *eventPanicConn) Close() error {
+	err := c.Conn.Close()
+	c.once.Do(func() { close(c.closed) })
+	return err
+}
+
+type eventPanicReadConn struct{ eofDataNetConn }
+
+func (*eventPanicReadConn) Read([]byte) (int, error) { return 0, errTestReadFailure }
+
+type eventPanicListener struct {
+	recordingEventListener
+	stage  string
+	events chan string
+}
+
+func (l *eventPanicListener) OnError(Session, error) {
+	l.events <- "error"
+	if strings.HasPrefix(l.stage, "error") {
+		panic("test OnError panic")
+	}
+}
+
+func (l *eventPanicListener) OnClose(Session) {
+	l.events <- "close"
+	if l.stage == "close" || l.stage == "error-close" {
+		panic("test OnClose panic")
+	}
+}
+
+func (l *eventPanicListener) OnCron(Session) {
+	l.events <- "cron"
+	if l.stage == "cron-nil" {
+		panic(nil)
+	}
+	panic("test OnCron panic")
+}
+
+type eventPanicHealthyListener struct {
+	recordingEventListener
+	ticks    chan time.Time
+	messages chan any
+}
+
+func (l *eventPanicHealthyListener) OnCron(Session) {
+	select {
+	case l.ticks <- time.Now():
+	default:
+	}
+}
+
+func (l *eventPanicHealthyListener) OnMessage(_ Session, msg any) { l.messages <- msg }
+
+func waitForEventPanic[T any](t *testing.T, ch <-chan T, name string) T {
+	t.Helper()
+	select {
+	case value := <-ch:
+		return value
+	case <-time.After(3 * time.Second):
+		t.Fatalf("timed out waiting for %s", name)
+		var zero T
+		return zero
+	}
+}
+
+// Reset remains external to callbacks; callback self-wait is tracked by #153.
+func waitForEventPanicReset(t *testing.T, ss *session) {
+	t.Helper()
+	done := make(chan struct{})
+	go func() {
+		ss.Reset()
+		close(done)
+	}()
+	waitForEventPanic(t, done, "external reset")
+}
+
+func TestEventCallbackPanicChild(t *testing.T) {
+	stage := os.Getenv(callbackPanicEnv)
+	if stage == "" {
+		t.Skip("subprocess only")
+	}
+
+	// A second session shares the actual default timer wheel. It must still
+	// receive messages and fresh heartbeats after the faulty session is gone.
+	healthyConn, peer := net.Pipe()
+	defer func() { _ = peer.Close() }()
+	healthy := newTCPSession(healthyConn, newServer(TCP_SERVER)).(*session)
+	live := &eventPanicHealthyListener{ticks: make(chan time.Time, 8), messages: make(chan any, 1)}
+	healthy.SetEventListener(live)
+	healthy.SetReader(wholeFrameReader{})
+	healthy.SetWriter(timeoutTestWriter{})
+	healthy.SetCronPeriod(10)
+	healthy.SetReadTimeout(time.Hour)
+	healthy.run()
+	defer func() {
+		healthy.Close()
+		_ = healthyConn.Close()
+		waitForEventPanicReset(t, healthy)
+	}()
+	waitForEventPanic(t, live.ticks, "initial healthy heartbeat")
+
+	endpoint := newServer(TCP_SERVER)
+	var raw net.Conn = &eofDataNetConn{}
+	switch stage {
+	case "error", "error-close":
+		raw = &eventPanicReadConn{}
+	case "cron", "cron-nil", "pooled-cron":
+		local, remote := net.Pipe()
+		defer func() { _ = remote.Close() }()
+		raw = local
+		if stage == "pooled-cron" {
+			pool := gxsync.NewTaskPoolSimple(1)
+			defer pool.Close()
+			endpoint = newServer(TCP_SERVER, WithServerTaskPool(pool))
+		}
+	case "close":
+	default:
+		t.Fatalf("unknown subprocess mode %q", stage)
+	}
+	conn := &eventPanicConn{Conn: raw, closed: make(chan struct{})}
+	ss := newTCPSession(conn, endpoint).(*session)
+	lifecycle := ss.lifecycle
+	listener := &eventPanicListener{stage: stage, events: make(chan string, 16)}
+	ss.SetEventListener(listener)
+	ss.SetReader(wholeFrameReader{})
+	ss.SetWriter(timeoutTestWriter{})
+	// Closing after a cron panic must release a blocked read immediately;
+	// waiting for the normal read deadline would stall this test for an hour.
+	ss.SetReadTimeout(time.Hour)
+	if strings.Contains(stage, "cron") {
+		ss.SetCronPeriod(10)
+	}
+	registeredClose := make(chan struct{})
+	ss.AddCloseCallback("panic-test", "cleanup", func() { close(registeredClose) })
+	ss.run()
+	// Panic handling itself must close the socket, stop the read loop and
+	// release lifecycle work before an external Reset is allowed to help.
+	done := make(chan struct{})
+	go func() {
+		ss.grWG.Wait()
+		lifecycle.wait()
+		close(done)
+	}()
+	waitForEventPanic(t, done, "faulty session cleanup")
+	if !ss.IsClosed() {
+		t.Error("faulty session is still open")
+	}
+	waitForEventPanic(t, conn.closed, "transport close")
+	waitForEventPanic(t, registeredClose, "registered close callback")
+	if ss.Connection != nil || ss.heartbeatTimer != nil || ss.grNum.Load() != 0 {
+		t.Fatal("panic leaked connection, timer or read goroutine")
+	}
+	if lifecycle.acquire() {
+		lifecycle.release()
+		t.Fatal("closed lifecycle still accepts work")
+	}
+	var events []string
+	for len(listener.events) > 0 {
+		events = append(events, <-listener.events)
+	}
+	if len(events) == 0 || events[len(events)-1] != "close" {
+		t.Fatalf("callback events must end with close: %v", events)
+	}
+	if strings.Contains(stage, "cron") {
+		var cronCount int
+		var shutdownEvents []string
+		for _, event := range events {
+			if event == "cron" {
+				cronCount++
+				continue
+			}
+			shutdownEvents = append(shutdownEvents, event)
+		}
+		if cronCount == 0 {
+			t.Fatalf("no OnCron callback ran: %v", events)
+		}
+		if got := fmt.Sprint(shutdownEvents); got != "[close]" && got != "[error close]" {
+			t.Fatalf("shutdown callback order = %v, want [close] or [error close] (all events: %v)", shutdownEvents, events)
+		}
+	} else {
+		want := "[close]"
+		if strings.HasPrefix(stage, "error") {
+			want = "[error close]"
+		}
+		if fmt.Sprint(events) != want {
+			t.Fatalf("callback order = %v, want %s", events, want)
+		}
+	}
+	waitForEventPanicReset(t, ss)
+	if ss.listener != nil || ss.reader != nil || ss.writer != nil {
+		t.Fatal("reset did not clear callback resources")
+	}
+
+	afterPanic := time.Now()
+	if err := peer.SetWriteDeadline(time.Now().Add(3 * time.Second)); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := peer.Write([]byte("still alive")); err != nil {
+		t.Fatal(err)
+	}
+	if got := waitForEventPanic(t, live.messages, "healthy message"); got != "still alive" {
+		t.Fatalf("healthy message = %v", got)
+	}
+	for !waitForEventPanic(t, live.ticks, "healthy heartbeat after panic").After(afterPanic) {
+	}
+	if healthy.IsClosed() {
+		t.Fatal("callback panic closed another session")
+	}
+}