diff --git a/transport/getty.go b/transport/getty.go index c1eecbf1..90b2bc10 100644 --- a/transport/getty.go +++ b/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 diff --git a/transport/session.go b/transport/session.go index 6e8a25e7..fc329bc1 100644 --- a/transport/session.go +++ b/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) }) } }() diff --git a/transport/session_event_panic_test.go b/transport/session_event_panic_test.go new file mode 100644 index 00000000..246a4770 --- /dev/null +++ b/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=") + } 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") + } +}