fix(server): recover from panics in RPC and UI HTTP handlers

pkg/server/recovery.go

@@ -0,0 +1,95 @@
+package server
+
+import (
+	"context"
+	"io"
+	"log/slog"
+	"net/http"
+	"runtime/debug"
+
+	"connectrpc.com/connect"
+	"github.com/raystack/frontier/pkg/errors"
+)
+
+// connectPanicRecovery converts a panic anywhere in the RPC handler chain
+// into a CodeInternal error response. The panic value and stack go to the
+// server log only, never to the caller.
+func connectPanicRecovery(logger *slog.Logger) connect.HandlerOption {
+	return connect.WithRecover(func(ctx context.Context, spec connect.Spec, _ http.Header, panicValue any) error {
+		logger.ErrorContext(ctx, "rpc handler panic",
+			"procedure", spec.Procedure,
+			"panic", panicValue,
+			"stack", string(debug.Stack()))
+		return connect.NewError(connect.CodeInternal, errors.ErrInternalServerError)
+	})
+}
+
+// committedWriter tracks whether the response status line has gone out, so
+// the recovery handler knows whether a 500 can still be sent.
+type committedWriter struct {
+	http.ResponseWriter
+	committed bool
+}
+
+func (c *committedWriter) WriteHeader(statusCode int) {
+	c.committed = true
+	c.ResponseWriter.WriteHeader(statusCode)
+}
+
+func (c *committedWriter) Write(b []byte) (int, error) {
+	c.committed = true
+	return c.ResponseWriter.Write(b)
+}
+
+// Flush commits the response too. The connect reverse proxy on the UI server
+// needs the Flusher interface for streaming.
+func (c *committedWriter) Flush() {
+	c.committed = true
+	if f, ok := c.ResponseWriter.(http.Flusher); ok {
+		f.Flush()
+	}
+}
+
+// Unwrap lets http.ResponseController reach the underlying writer's
+// Hijacker, deadline, and full-duplex methods.
+func (c *committedWriter) Unwrap() http.ResponseWriter {
+	return c.ResponseWriter
+}
+
+// ReadFrom keeps the underlying writer's optimized copy path (pooled
+// buffers, sendfile for OS files), which io.Copy looks for on the
+// destination directly without walking Unwrap.
+func (c *committedWriter) ReadFrom(r io.Reader) (int64, error) {
+	c.committed = true
+	if rf, ok := c.ResponseWriter.(io.ReaderFrom); ok {
+		return rf.ReadFrom(r)
+	}
+	return io.Copy(c.ResponseWriter, r)
+}
+
+// httpPanicRecovery responds with a plain 500 instead of dropping the
+// connection when a handler behind the wrapped mux panics. If the handler
+// already wrote part of a response, the status can no longer be changed, so
+// it aborts the connection instead of appending an error to the partial body.
+func httpPanicRecovery(logger *slog.Logger, next http.Handler) http.Handler {
+	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		cw := &committedWriter{ResponseWriter: w}
+		defer func() {
+			if panicValue := recover(); panicValue != nil {
+				// net/http checks for ErrAbortHandler with ==, so we should too.
+				if panicValue == http.ErrAbortHandler {
+					panic(panicValue)
+				}
+				logger.ErrorContext(r.Context(), "http handler panic",
+					"path", r.URL.Path,
+					"panic", panicValue,
+					"stack", string(debug.Stack()))
+				if cw.committed {
+					panic(http.ErrAbortHandler)
+				}
+				http.Error(cw, errors.ErrInternalServerError.Error(), http.StatusInternalServerError)
+			}
+		}()
+		next.ServeHTTP(cw, r)
+	})
+}

pkg/server/recovery_test.go

@@ -0,0 +1,203 @@
+package server
+
+import (
+	"bytes"
+	"context"
+	"io"
+	"log/slog"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"testing"
+
+	"connectrpc.com/connect"
+	"google.golang.org/protobuf/types/known/emptypb"
+)
+
+func TestConnectPanicRecoveryReturnsInternalError(t *testing.T) {
+	var logBuf bytes.Buffer
+	logger := slog.New(slog.NewJSONHandler(&logBuf, nil))
+
+	handler := connect.NewUnaryHandler(
+		"/test.v1.TestService/Panic",
+		func(ctx context.Context, req *connect.Request[emptypb.Empty]) (*connect.Response[emptypb.Empty], error) {
+			panic("boom")
+		},
+		connectPanicRecovery(logger),
+	)
+	mux := http.NewServeMux()
+	mux.Handle("/test.v1.TestService/Panic", handler)
+	srv := httptest.NewServer(mux)
+	defer srv.Close()
+
+	resp, err := http.Post(srv.URL+"/test.v1.TestService/Panic", "application/json", strings.NewReader("{}"))
+	if err != nil {
+		t.Fatalf("expected an error response, got a dropped connection: %v", err)
+	}
+	defer resp.Body.Close()
+
+	if resp.StatusCode != http.StatusInternalServerError {
+		t.Errorf("status = %d, want %d", resp.StatusCode, http.StatusInternalServerError)
+	}
+	body, err := io.ReadAll(resp.Body)
+	if err != nil {
+		t.Fatalf("read body: %v", err)
+	}
+	if !strings.Contains(string(body), `"internal"`) {
+		t.Errorf("body %q does not carry the internal error code", body)
+	}
+	if strings.Contains(string(body), "boom") {
+		t.Errorf("panic detail leaked to the client: %q", body)
+	}
+
+	logged := logBuf.String()
+	if !strings.Contains(logged, `"level":"ERROR"`) {
+		t.Errorf("panic was not logged at error level: %q", logged)
+	}
+	if !strings.Contains(logged, "boom") || !strings.Contains(logged, "/test.v1.TestService/Panic") {
+		t.Errorf("log entry is missing the panic value or procedure: %q", logged)
+	}
+	if !strings.Contains(logged, "recovery_test.go") {
+		t.Errorf("log entry is missing the stack trace: %q", logged)
+	}
+}
+
+func TestHTTPPanicRecoveryReturnsInternalError(t *testing.T) {
+	var logBuf bytes.Buffer
+	logger := slog.New(slog.NewJSONHandler(&logBuf, nil))
+
+	mux := http.NewServeMux()
+	mux.HandleFunc("/panic", func(w http.ResponseWriter, r *http.Request) {
+		panic("boom")
+	})
+
+	rec := httptest.NewRecorder()
+	httpPanicRecovery(logger, mux).ServeHTTP(rec, httptest.NewRequest(http.MethodGet, "/panic", nil))
+
+	if rec.Code != http.StatusInternalServerError {
+		t.Errorf("status = %d, want %d", rec.Code, http.StatusInternalServerError)
+	}
+	if strings.Contains(rec.Body.String(), "boom") {
+		t.Errorf("panic detail leaked to the client: %q", rec.Body.String())
+	}
+
+	logged := logBuf.String()
+	if !strings.Contains(logged, `"level":"ERROR"`) || !strings.Contains(logged, "boom") || !strings.Contains(logged, "/panic") {
+		t.Errorf("log entry is missing the error level, panic value, or path: %q", logged)
+	}
+}
+
+func TestHTTPPanicRecoveryPassesThroughNormalRequests(t *testing.T) {
+	logger := slog.New(slog.NewTextHandler(io.Discard, nil))
+
+	mux := http.NewServeMux()
+	mux.HandleFunc("/ok", func(w http.ResponseWriter, r *http.Request) {
+		w.WriteHeader(http.StatusTeapot)
+	})
+
+	rec := httptest.NewRecorder()
+	httpPanicRecovery(logger, mux).ServeHTTP(rec, httptest.NewRequest(http.MethodGet, "/ok", nil))
+
+	if rec.Code != http.StatusTeapot {
+		t.Errorf("status = %d, want %d", rec.Code, http.StatusTeapot)
+	}
+}
+
+func TestHTTPPanicRecoveryAbortsCommittedResponses(t *testing.T) {
+	var logBuf bytes.Buffer
+	logger := slog.New(slog.NewJSONHandler(&logBuf, nil))
+
+	mux := http.NewServeMux()
+	mux.HandleFunc("/partial", func(w http.ResponseWriter, r *http.Request) {
+		w.WriteHeader(http.StatusOK)
+		if _, err := w.Write([]byte("partial")); err != nil {
+			t.Fatalf("write: %v", err)
+		}
+		panic("boom")
+	})
+
+	rec := httptest.NewRecorder()
+	func() {
+		defer func() {
+			if recovered := recover(); recovered != http.ErrAbortHandler {
+				t.Errorf("recovered %v, want http.ErrAbortHandler", recovered)
+			}
+		}()
+		httpPanicRecovery(logger, mux).ServeHTTP(rec, httptest.NewRequest(http.MethodGet, "/partial", nil))
+	}()
+
+	if rec.Code != http.StatusOK {
+		t.Errorf("status = %d, want the already-committed %d", rec.Code, http.StatusOK)
+	}
+	if rec.Body.String() != "partial" {
+		t.Errorf("body = %q, want the partial body with nothing appended", rec.Body.String())
+	}
+	if !strings.Contains(logBuf.String(), "boom") {
+		t.Errorf("panic was not logged: %q", logBuf.String())
+	}
+}
+
+func TestHTTPPanicRecoveryAbortsAfterFlush(t *testing.T) {
+	logger := slog.New(slog.NewTextHandler(io.Discard, nil))
+
+	mux := http.NewServeMux()
+	mux.HandleFunc("/flush", func(w http.ResponseWriter, r *http.Request) {
+		w.(http.Flusher).Flush()
+		panic("boom")
+	})
+
+	rec := httptest.NewRecorder()
+	func() {
+		defer func() {
+			if recovered := recover(); recovered != http.ErrAbortHandler {
+				t.Errorf("recovered %v, want http.ErrAbortHandler", recovered)
+			}
+		}()
+		httpPanicRecovery(logger, mux).ServeHTTP(rec, httptest.NewRequest(http.MethodGet, "/flush", nil))
+	}()
+
+	if strings.Contains(rec.Body.String(), "internal server error") {
+		t.Errorf("error body appended to a flushed response: %q", rec.Body.String())
+	}
+}
+
+func TestCommittedWriterUnwrapReturnsUnderlyingWriter(t *testing.T) {
+	rec := httptest.NewRecorder()
+	cw := &committedWriter{ResponseWriter: rec}
+
+	if cw.Unwrap() != http.ResponseWriter(rec) {
+		t.Errorf("Unwrap() = %v, want the underlying writer", cw.Unwrap())
+	}
+}
+
+func TestCommittedWriterReadFromCommitsAndCopies(t *testing.T) {
+	rec := httptest.NewRecorder()
+	cw := &committedWriter{ResponseWriter: rec}
+
+	n, err := cw.ReadFrom(strings.NewReader("data"))
+	if err != nil || n != 4 {
+		t.Fatalf("ReadFrom = (%d, %v), want (4, nil)", n, err)
+	}
+	if !cw.committed {
+		t.Error("ReadFrom did not mark the response as committed")
+	}
+	if rec.Body.String() != "data" {
+		t.Errorf("body = %q, want %q", rec.Body.String(), "data")
+	}
+}
+
+func TestHTTPPanicRecoveryRepanicsOnErrAbortHandler(t *testing.T) {
+	logger := slog.New(slog.NewTextHandler(io.Discard, nil))
+
+	mux := http.NewServeMux()
+	mux.HandleFunc("/abort", func(w http.ResponseWriter, r *http.Request) {
+		panic(http.ErrAbortHandler)
+	})
+
+	defer func() {
+		if recovered := recover(); recovered != http.ErrAbortHandler {
+			t.Errorf("recovered %v, want http.ErrAbortHandler", recovered)
+		}
+	}()
+	httpPanicRecovery(logger, mux).ServeHTTP(httptest.NewRecorder(), httptest.NewRequest(http.MethodGet, "/abort", nil))
+}

pkg/server/server.go

@@ -100,7 +100,7 @@ func ServeUI(ctx context.Context, logger *slog.Logger, uiConfig UIConfig, apiSer
 
 	server := &http.Server{
 		Addr:              fmt.Sprintf(":%d", uiConfig.Port),
-		Handler:           mux,
+		Handler:           httpPanicRecovery(logger, mux),
 		ReadHeaderTimeout: apiServerConfig.ReadHeaderTimeout,
 		IdleTimeout:       apiServerConfig.IdleTimeout,
 	}
@@ -178,8 +178,10 @@ func ServeConnect(ctx context.Context, logger *slog.Logger, cfg Config, deps api
 		auditInterceptor,
 		sessionInterceptor.UnaryConnectResponseInterceptor())
 
-	frontierPath, frontierHandler := frontierv1beta1connect.NewFrontierServiceHandler(frontierService, interceptors, connect.WithCodec(connectCodec{}))
-	adminPath, adminHandler := frontierv1beta1connect.NewAdminServiceHandler(frontierService, interceptors, connect.WithCodec(connectCodec{}))
+	// Panic recovery goes first so it sits outermost and catches panics from
+	// the other interceptors as well, not just the handlers.
+	frontierPath, frontierHandler := frontierv1beta1connect.NewFrontierServiceHandler(frontierService, connectPanicRecovery(logger), interceptors, connect.WithCodec(connectCodec{}))
+	adminPath, adminHandler := frontierv1beta1connect.NewAdminServiceHandler(frontierService, connectPanicRecovery(logger), interceptors, connect.WithCodec(connectCodec{}))
 
 	// Create mux and register handlers
 	mux := http.NewServeMux()
@@ -218,8 +220,11 @@ func ServeConnect(ctx context.Context, logger *slog.Logger, cfg Config, deps api
 		_ = json.NewEncoder(w).Encode(map[string]string{"status": "SERVING"})
 	})
 
-	// Configure and create the server
-	handler := connectinterceptors.WithConnectCORS(mux, cfg.ConnectCors)
+	// Configure and create the server. Panic recovery wraps the whole mux so
+	// routes outside the two service handlers (webhook bridge, ping, health,
+	// reflection, CORS) are covered too; RPC panics are still converted to
+	// connect error codes by WithRecover before they can reach this net.
+	handler := httpPanicRecovery(logger, connectinterceptors.WithConnectCORS(mux, cfg.ConnectCors))
 
 	// Serve HTTP/1.1 and unencrypted HTTP/2. Unlike the x/net h2c wrapper
 	// this replaces, the server tracks these HTTP/2 connections itself, so