refactor: migrate to connect-go v2 (#1325)

Migrates the runner to `connectrpc.com/connect/v2` by pinning actionslib v1.5.0, which carries https://gitea.com/gitea/actionslib/pulls/28. The wire behaviour towards Gitea is unchanged and works against Gitea servers on connect v1 and v2. The client keeps v1's unlimited response size, and the error codes and messages the runner checks stay the same.

Reviewed-on: https://gitea.com/gitea/runner/pulls/1325
Reviewed-by: bircni <[email protected]>
Co-authored-by: silverwind <[email protected]>

internal/app/cmd/daemon.go

@@ -31,7 +31,7 @@ import (
 	"gitea.com/gitea/runner/internal/pkg/telemetry"
 	"gitea.com/gitea/runner/internal/pkg/ver"
 
-	"connectrpc.com/connect"
+	"connectrpc.com/connect/v2"
 	runnerv1 "gitea.dev/actionslib/runner/v1"
 	"github.com/spf13/cobra"
 	"golang.org/x/term"
@@ -188,7 +188,7 @@ func runDaemon(ctx context.Context, daemArgs *daemonArgs, configFile *string) fu
 		}()
 
 		// declare the labels of the runner before fetching tasks
-		var resp *connect.Response[runnerv1.DeclareResponse]
+		var resp *runnerv1.DeclareResponse
 		for attempt := 0; ; attempt++ {
 			resp, err = runner.Declare(ctx, ls.Names())
 			if err == nil || !shouldRetryDeclare(err) || ctx.Err() != nil {
@@ -218,14 +218,13 @@ func runDaemon(ctx context.Context, daemArgs *daemonArgs, configFile *string) fu
 			return err
 		default:
 			logging.Infof("runner: %s, with version: %s, with labels: %v, declare successfully",
-				resp.Msg.Runner.Name, resp.Msg.Runner.Version, resp.Msg.Runner.Labels)
+				resp.Runner.Name, resp.Runner.Version, resp.Runner.Labels)
 		}
-		runner.SetCapabilitiesFromDeclare(resp)
 
 		poller := poll.New(cfg, cli, runner)
 
 		metrics.Init() // OTLP exports the registry even without the server
-		metrics.RunnerInfo.WithLabelValues(ver.Version(), resp.Msg.Runner.Name).Set(1)
+		metrics.RunnerInfo.WithLabelValues(ver.Version(), resp.Runner.Name).Set(1)
 		metrics.RunnerCapacity.Set(float64(cfg.Runner.Capacity))
 		metrics.RegisterUptimeFunc(time.Now())
 		metrics.RegisterRunningJobsFunc(runner.RunningCount, cfg.Runner.Capacity)
@@ -267,14 +266,14 @@ func runDaemon(ctx context.Context, daemArgs *daemonArgs, configFile *string) fu
 			}
 		}
 
-		logging.Infof("runner: %s shutdown initiated, waiting %s for running jobs to complete before shutting down", resp.Msg.Runner.Name, cfg.Runner.ShutdownTimeout)
+		logging.Infof("runner: %s shutdown initiated, waiting %s for running jobs to complete before shutting down", resp.Runner.Name, cfg.Runner.ShutdownTimeout)
 
 		ctx, cancel := context.WithTimeout(context.Background(), cfg.Runner.ShutdownTimeout)
 		defer cancel()
 
 		err = poller.Shutdown(ctx)
 		if err != nil {
-			logging.Warnf("runner: %s cancelled in progress jobs during shutdown", resp.Msg.Runner.Name)
+			logging.Warnf("runner: %s cancelled in progress jobs during shutdown", resp.Runner.Name)
 		}
 
 		if poller.RegistrationRejected() {

internal/app/cmd/daemon_test.go

@@ -6,7 +6,6 @@ package cmd
 import (
 	"context"
 	"crypto/tls"
-	"errors"
 	"net/http"
 	"net/http/httptest"
 	"os"
@@ -19,7 +18,8 @@ import (
 	"gitea.com/gitea/runner/act/common/logging"
 	"gitea.com/gitea/runner/internal/pkg/config"
 
-	"connectrpc.com/connect"
+	"connectrpc.com/connect/v2"
+	"connectrpc.com/connect/v2/connecthttp"
 	runnerv1 "gitea.dev/actionslib/runner/v1"
 	"gitea.dev/actionslib/runner/v1/runnerv1connect"
 	"github.com/stretchr/testify/require"
@@ -83,7 +83,7 @@ func TestInitLoggingSetsLevelAndCaller(t *testing.T) {
 	require.Contains(t, buf.String(), "[daemon_test.go:")
 }
 
-var errUnregisteredRunner = errors.New("rpc error: code = Unauthenticated desc = unregistered runner")
+var errUnregisteredRunner = connect.NewError(connect.CodeUnknown, "rpc error: code = Unauthenticated desc = unregistered runner")
 
 type daemonTestService struct {
 	runnerv1connect.UnimplementedRunnerServiceHandler
@@ -93,9 +93,9 @@ type daemonTestService struct {
 	cancel               context.CancelFunc
 }
 
-func (s *daemonTestService) Declare(context.Context, *connect.Request[runnerv1.DeclareRequest]) (*connect.Response[runnerv1.DeclareResponse], error) {
+func (s *daemonTestService) Declare(context.Context, *runnerv1.DeclareRequest) (*runnerv1.DeclareResponse, error) {
 	if s.declareCalls.Add(1) == 1 {
-		return nil, connect.NewError(connect.CodeUnavailable, errors.New("bad gateway"))
+		return nil, connect.NewError(connect.CodeUnavailable, "bad gateway")
 	}
 	if s.declareErr != nil {
 		if s.beforeTask {
@@ -103,10 +103,10 @@ func (s *daemonTestService) Declare(context.Context, *connect.Request[runnerv1.D
 		}
 		return nil, s.declareErr
 	}
-	return connect.NewResponse(&runnerv1.DeclareResponse{Runner: &runnerv1.Runner{Name: "test"}}), nil
+	return &runnerv1.DeclareResponse{Runner: &runnerv1.Runner{Name: "test"}}, nil
 }
 
-func (s *daemonTestService) FetchTask(context.Context, *connect.Request[runnerv1.FetchTaskRequest]) (*connect.Response[runnerv1.FetchTaskResponse], error) {
+func (s *daemonTestService) FetchTask(context.Context, *runnerv1.FetchTaskRequest) (*runnerv1.FetchTaskResponse, error) {
 	if s.fetchErr != nil {
 		return nil, s.fetchErr
 	}
@@ -114,15 +114,15 @@ func (s *daemonTestService) FetchTask(context.Context, *connect.Request[runnerv1
 		s.cancel()
 		return nil, context.Canceled
 	}
-	return connect.NewResponse(&runnerv1.FetchTaskResponse{Task: &runnerv1.Task{Id: 1, Context: &structpb.Struct{}, WorkflowPayload: []byte("invalid: [")}}), nil
+	return &runnerv1.FetchTaskResponse{Task: &runnerv1.Task{Id: 1, Context: &structpb.Struct{}, WorkflowPayload: []byte("invalid: [")}}, nil
 }
 
-func (*daemonTestService) UpdateTask(_ context.Context, req *connect.Request[runnerv1.UpdateTaskRequest]) (*connect.Response[runnerv1.UpdateTaskResponse], error) {
-	return connect.NewResponse(&runnerv1.UpdateTaskResponse{State: req.Msg.State}), nil
+func (*daemonTestService) UpdateTask(_ context.Context, req *runnerv1.UpdateTaskRequest) (*runnerv1.UpdateTaskResponse, error) {
+	return &runnerv1.UpdateTaskResponse{State: req.State}, nil
 }
 
-func (*daemonTestService) UpdateLog(_ context.Context, req *connect.Request[runnerv1.UpdateLogRequest]) (*connect.Response[runnerv1.UpdateLogResponse], error) {
-	return connect.NewResponse(&runnerv1.UpdateLogResponse{AckIndex: req.Msg.Index + int64(len(req.Msg.Rows))}), nil
+func (*daemonTestService) UpdateLog(_ context.Context, req *runnerv1.UpdateLogRequest) (*runnerv1.UpdateLogResponse, error) {
+	return &runnerv1.UpdateLogResponse{AckIndex: req.Index + int64(len(req.Rows))}, nil
 }
 
 func TestDaemonRetriesDeclareOutagesAndRemovesOnlyConsumedEphemeralRegistration(t *testing.T) {
@@ -138,18 +138,21 @@ func TestDaemonRetriesDeclareOutagesAndRemovesOnlyConsumedEphemeralRegistration(
 		{"ephemeral interrupted before task", true, true, nil, nil, false},
 		{"persistent once", false, false, nil, nil, false},
 		{"ephemeral rejected on fetch", true, false, nil, errUnregisteredRunner, true},
-		{"ephemeral rejected on fetch with 401", true, false, nil, connect.NewError(connect.CodeUnauthenticated, errors.New("unregistered runner")), true},
-		{"ephemeral proxy 401 on fetch", true, false, nil, connect.NewError(connect.CodeUnauthenticated, errors.New("HTTP status 401 Unauthorized")), false},
+		{"ephemeral rejected on fetch with 401", true, false, nil, connect.NewError(connect.CodeUnauthenticated, "unregistered runner"), true},
+		{"ephemeral proxy 401 on fetch", true, false, nil, connect.NewError(connect.CodeUnauthenticated, "HTTP status 401 Unauthorized"), false},
 		{"ephemeral rejected on declare", true, false, errUnregisteredRunner, nil, true},
 		{"persistent rejected on declare", false, false, errUnregisteredRunner, nil, false},
-		{"ephemeral stopped during declare outage", true, true, connect.NewError(connect.CodeUnavailable, errors.New("bad gateway")), nil, false},
+		{"ephemeral stopped during declare outage", true, true, connect.NewError(connect.CodeUnavailable, "bad gateway"), nil, false},
 	} {
 		t.Run(tc.name, func(t *testing.T) {
 			ctx, cancel := context.WithTimeout(t.Context(), 5*time.Second)
 			defer cancel()
 			service := &daemonTestService{declareErr: tc.declareErr, fetchErr: tc.fetchErr, beforeTask: tc.beforeTask, cancel: cancel}
-			_, handler := runnerv1connect.NewRunnerServiceHandler(service)
-			server := httptest.NewServer(http.StripPrefix("/api/actions", handler))
+			rpcServer := connect.NewServer()
+			runnerv1connect.RegisterRunnerServiceHandler(rpcServer, service)
+			mux := http.NewServeMux()
+			connecthttp.Mount(mux, rpcServer, connecthttp.WithReadMaxBytes(0))
+			server := httptest.NewServer(http.StripPrefix("/api/actions", mux))
 			defer server.Close()
 			dir := t.TempDir()
 			regFile := filepath.Join(dir, "registration.json")
@@ -177,13 +180,13 @@ func TestShouldRetryDeclare(t *testing.T) {
 		err   error
 		retry bool
 	}{
-		{connect.NewError(connect.CodeUnavailable, errors.New("bad gateway")), true},
-		{connect.NewWireError(connect.CodeInternal, errors.New("update runner")), true},
-		{connect.NewError(connect.CodeUnavailable, &tls.CertificateVerificationError{Err: errors.New("unknown authority")}), false},
-		{connect.NewError(connect.CodeUnavailable, http.ErrSchemeMismatch), false},
-		{connect.NewError(connect.CodeUnavailable, tls.AlertError(42)), false},
-		{connect.NewError(connect.CodePermissionDenied, errors.New("forbidden")), false},
-		{connect.NewWireError(connect.CodeUnknown, errors.New("unregistered runner")), false},
+		{connect.NewError(connect.CodeUnavailable, "bad gateway"), true},
+		{connect.NewError(connect.CodeInternal, "update runner").WithRemote(), true},
+		{connect.NewError(connect.CodeUnavailable, "certificate verification").WithCause(&tls.CertificateVerificationError{}), false},
+		{connect.NewError(connect.CodeUnavailable, "scheme mismatch").WithCause(http.ErrSchemeMismatch), false},
+		{connect.NewError(connect.CodeUnavailable, "TLS alert").WithCause(tls.AlertError(42)), false},
+		{connect.NewError(connect.CodePermissionDenied, "forbidden"), false},
+		{connect.NewError(connect.CodeUnknown, "unregistered runner").WithRemote(), false},
 	} {
 		require.Equal(t, tc.retry, shouldRetryDeclare(tc.err), tc.err.Error())
 	}

internal/app/cmd/register.go

@@ -21,7 +21,6 @@ import (
 	"gitea.com/gitea/runner/internal/pkg/lock"
 	"gitea.com/gitea/runner/internal/pkg/ver"
 
-	"connectrpc.com/connect"
 	pingv1 "gitea.dev/actionslib/ping/v1"
 	runnerv1 "gitea.dev/actionslib/runner/v1"
 	"github.com/spf13/cobra"
@@ -371,9 +370,9 @@ func doRegister(ctx context.Context, cfg *config.Config, inputs *registerInputs)
 		cfg.Runner.ExtraHeaders,
 	)
 
-	if _, err := cli.Ping(ctx, connect.NewRequest(&pingv1.PingRequest{
+	if _, err := cli.Ping(ctx, &pingv1.PingRequest{
 		Data: inputs.RunnerName,
-	})); err != nil {
+	}); err != nil {
 		return fmt.Errorf("cannot ping the Gitea instance server: %w", err)
 	}
 	logging.Debug("Successfully pinged the Gitea instance server")
@@ -395,25 +394,25 @@ func doRegister(ctx context.Context, cfg *config.Config, inputs *registerInputs)
 		ls[i] = l.Name
 	}
 	// register new runner.
-	resp, err := cli.Register(ctx, connect.NewRequest(&runnerv1.RegisterRequest{
+	resp, err := cli.Register(ctx, &runnerv1.RegisterRequest{
 		Name:         reg.Name,
 		Token:        reg.Token,
 		Version:      ver.Version(),
 		Labels:       ls,
 		Ephemeral:    reg.Ephemeral,
 		Capabilities: run.RunnerCapabilities(),
-	}))
+	})
 	if err != nil {
 		logging.WithError(err).Error("poller: cannot register new runner")
 		return err
 	}
 
-	reg.ID = resp.Msg.Runner.Id
-	reg.UUID = resp.Msg.Runner.Uuid
-	reg.Name = resp.Msg.Runner.Name
-	reg.Token = resp.Msg.Runner.Token
+	reg.ID = resp.Runner.Id
+	reg.UUID = resp.Runner.Uuid
+	reg.Name = resp.Runner.Name
+	reg.Token = resp.Runner.Token
 
-	if inputs.Ephemeral != resp.Msg.Runner.Ephemeral {
+	if inputs.Ephemeral != resp.Runner.Ephemeral {
 		// TODO we cannot remove the configuration via runner api, if we return an error here we just fill the database
 		logging.Error("poller: cannot register new runner as ephemeral upgrade Gitea to gain security, run-once will be used automatically")
 	}