feat: issue-3628 Serve gRPC and HTTP from one endpoint per service (control plane)

app/controlplane/cmd/main.go

@@ -32,6 +32,7 @@ import (
 	"github.com/chainloop-dev/chainloop/app/controlplane/plugins/sdk/v1"
 	"github.com/chainloop-dev/chainloop/pkg/credentials"
 	"github.com/chainloop-dev/chainloop/pkg/credentials/manager"
+	"github.com/chainloop-dev/chainloop/pkg/httpgrpcmux"
 	"github.com/chainloop-dev/chainloop/pkg/natsconn"
 	"github.com/chainloop-dev/chainloop/pkg/servicelogger"
 
@@ -61,11 +62,42 @@ func init() {
 	flag.StringVar(&flagconf, "conf", "../configs", "config path, eg: -conf config.yaml")
 }
 
-func newApp(logger log.Logger, gs *grpc.Server, hs *http.Server, ms *server.HTTPMetricsServer, profilerSvc *server.HTTPProfilerServer,
+// cleanShutdownServer tolerates the shutdown race with the mux: kratos starts
+// and stops servers concurrently, and the mux's sub-listeners close together
+// with the root listener, so the gRPC/HTTP servers may see their listener
+// close before they flag their own shutdown. Their Start then returns the
+// mux's shutdown error even though the shutdown is orderly — kratos would
+// surface it and main would panic on every SIGTERM the mux wins. The race is
+// only tolerated when the mux is actually stopping: a fatal accept-loop
+// failure surfaces as an error, so the process restarts instead of serving a
+// dead single port.
+type cleanShutdownServer struct {
+	transport.Server
+	mux *httpgrpcmux.Mux
+}
+
+func (s *cleanShutdownServer) Start(ctx context.Context) error {
+	err := s.Server.Start(ctx)
+	if s.mux.IsOrderlyShutdownErr(err) {
+		return nil
+	}
+	return err
+}
+
+func newApp(logger log.Logger, am *httpgrpcmux.Mux, gs []*grpc.Server, hs []*http.Server, ms *server.HTTPMetricsServer, profilerSvc *server.HTTPProfilerServer,
 	expirer *biz.WorkflowRunExpirerUseCase, plugins sdk.AvailablePlugins,
 	userAccessSyncer *biz.UserAccessSyncerUseCase, casBackendChecker *biz.CASBackendChecker,
 	apiTokenStaleRevoker *biz.APITokenStaleRevoker, loginHandoffs *biz.LoginHandoffUseCase, cfg *conf.Bootstrap) *app {
-	servers := []transport.Server{gs, hs, ms}
+	servers := make([]transport.Server, 0, len(gs)+len(hs)+3)
+	// The mux owns the single-port listener; kratos starts servers concurrently,
+	// so no ordering is required — the virtual listeners work before Start.
+	servers = append(servers, am, ms)
+	for _, g := range gs {
+		servers = append(servers, &cleanShutdownServer{Server: g, mux: am})
+	}
+	for _, h := range hs {
+		servers = append(servers, &cleanShutdownServer{Server: h, mux: am})
+	}
 	if cfg.EnableProfiler {
 		servers = append(servers, profilerSvc)
 	}

app/controlplane/cmd/main_test.go

@@ -15,15 +15,20 @@
 package main
 
 import (
+	"context"
 	"fmt"
+	"net"
 	"os"
 	"path/filepath"
 	"testing"
 	"time"
 
 	conf "github.com/chainloop-dev/chainloop/app/controlplane/internal/conf/controlplane/config/v1"
+	"github.com/chainloop-dev/chainloop/pkg/httpgrpcmux"
 	"github.com/go-kratos/kratos/v2/config"
 	"github.com/go-kratos/kratos/v2/config/file"
+	kratosgrpc "github.com/go-kratos/kratos/v2/transport/grpc"
+	kratoshttp "github.com/go-kratos/kratos/v2/transport/http"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	"google.golang.org/protobuf/types/known/durationpb"
@@ -216,3 +221,55 @@ func loadBootstrap(t *testing.T, path string) (*conf.Bootstrap, error) {
 
 	return &bootstrap, nil
 }
+
+// TestServerToleratesMuxShutdown proves the shutdown-race fix: when the mux
+// stops first, the servers' Start returns net.ErrClosed — an orderly
+// shutdown that must not surface as an error (kratos would propagate it and
+// main would panic on SIGTERM).
+func TestServerToleratesMuxShutdown(t *testing.T) {
+	l, err := net.Listen("tcp", "127.0.0.1:0")
+	require.NoError(t, err)
+
+	m := httpgrpcmux.New(l, nil)
+	require.NoError(t, m.Start(context.Background()))
+
+	// A kratos HTTP server on the mux's HTTP side, and a kratos gRPC server
+	// on its gRPC side: both fed by virtual listeners.
+	hs := kratoshttp.NewServer(kratoshttp.Listener(m.HTTP()))
+	gs := kratosgrpc.NewServer(kratosgrpc.Listener(m.GRPC()))
+
+	errs := make(chan error, 2)
+	go func() { errs <- (&cleanShutdownServer{Server: hs, mux: m}).Start(context.Background()) }()
+	go func() { errs <- (&cleanShutdownServer{Server: gs, mux: m}).Start(context.Background()) }()
+
+	// Give the servers time to enter their accept loops, then close the mux
+	// first — exactly the SIGTERM ordering that used to panic.
+	time.Sleep(100 * time.Millisecond)
+	m.Close()
+
+	for i := 0; i < 2; i++ {
+		select {
+		case err := <-errs:
+			require.NoError(t, err, "an orderly mux shutdown must not surface as a server error")
+		case <-time.After(5 * time.Second):
+			require.Fail(t, "server Start did not return after the mux stopped")
+		}
+	}
+}
+
+// TestCleanShutdownServerPropagatesRealErrors proves the decorator only
+// tolerates the mux shutdown: a real failure (bind conflict) surfaces as an
+// error, so the process still crashes on genuine startup problems.
+func TestCleanShutdownServerPropagatesRealErrors(t *testing.T) {
+	l, err := net.Listen("tcp", "127.0.0.1:0")
+	require.NoError(t, err)
+	defer l.Close()
+
+	decorated := &cleanShutdownServer{
+		Server: kratoshttp.NewServer(kratoshttp.Address(l.Addr().String())),
+		mux:    httpgrpcmux.New(l, nil),
+	}
+	err = decorated.Start(context.Background())
+	require.Error(t, err)
+	assert.NotErrorIs(t, err, net.ErrClosed)
+}