From 6f0d4f1b4845c816e1447996de116c9978bc4a36 Mon Sep 17 00:00:00 2001 From: silverwind Date: Wed, 21 Oct 2026 16:10:58 +0000 Subject: [PATCH] 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 Co-authored-by: silverwind --- e2e/runner_harness.go | 23 ++- go.mod | 6 +- go.sum | 12 +- internal/app/cmd/daemon.go | 13 +- internal/app/cmd/daemon_test.go | 51 +++--- internal/app/cmd/register.go | 19 ++- internal/app/poll/poller.go | 18 +-- internal/app/poll/poller_test.go | 38 ++--- internal/app/run/runner.go | 21 ++- internal/app/run/runner_test.go | 42 +++-- internal/pkg/client/client.go | 12 +- internal/pkg/client/http.go | 40 ++--- internal/pkg/client/http_test.go | 25 ++- internal/pkg/client/mocks/Client.go | 19 ++- internal/pkg/report/joblog_test.go | 7 +- internal/pkg/report/reporter.go | 16 +- internal/pkg/report/reporter_test.go | 231 +++++++++++++-------------- 17 files changed, 292 insertions(+), 301 deletions(-) diff --git a/e2e/runner_harness.go b/e2e/runner_harness.go index d8c1a0bc..8e4e5d48 100644 --- a/e2e/runner_harness.go +++ b/e2e/runner_harness.go @@ -19,7 +19,6 @@ import ( "gitea.com/gitea/runner/internal/pkg/config" "gitea.com/gitea/runner/internal/pkg/labels" - "connectrpc.com/connect" pingv1 "gitea.dev/actionslib/ping/v1" runnerv1 "gitea.dev/actionslib/runner/v1" ) @@ -76,44 +75,42 @@ func startRunner(t *testing.T, repo, labelName string, options runnerOptions) *p labelNames := []string{label.Name} pingCli := client.New(fixture.baseURL, cfg.Runner.TLSConfig(), "", "", config.RequestTimeout, cfg.Runner.ExtraHeaders) - if _, err := pingCli.Ping(ctx, connect.NewRequest(&pingv1.PingRequest{Data: t.Name()})); err != nil { + if _, err := pingCli.Ping(ctx, &pingv1.PingRequest{Data: t.Name()}); err != nil { t.Fatalf("ping %s: %v", fixture.baseURL, err) } - regResp, err := pingCli.Register(ctx, connect.NewRequest(&runnerv1.RegisterRequest{ + regResp, err := pingCli.Register(ctx, &runnerv1.RegisterRequest{ Name: t.Name(), Token: token, Version: "e2e", Labels: labelNames, Ephemeral: options.ephemeral, Capabilities: run.RunnerCapabilities(), - })) + }) if err != nil { t.Fatalf("register runner: %v", err) } - if options.ephemeral && !regResp.Msg.Runner.Ephemeral { + if options.ephemeral && !regResp.Runner.Ephemeral { t.Fatal("gitea did not grant ephemeral registration") } reg := &config.Registration{ - ID: regResp.Msg.Runner.Id, - UUID: regResp.Msg.Runner.Uuid, - Name: regResp.Msg.Runner.Name, - Token: regResp.Msg.Runner.Token, + ID: regResp.Runner.Id, + UUID: regResp.Runner.Uuid, + Name: regResp.Runner.Name, + Token: regResp.Runner.Token, Address: fixture.baseURL, Labels: []string{rawLabel}, - Ephemeral: regResp.Msg.Runner.Ephemeral, + Ephemeral: regResp.Runner.Ephemeral, } cli := client.New(fixture.baseURL, cfg.Runner.TLSConfig(), reg.UUID, reg.Token, config.RequestTimeout, cfg.Runner.ExtraHeaders) runner := run.NewRunner(cfg, reg, cli) - declResp, err := runner.Declare(ctx, labelNames) - if err != nil { + if _, err := runner.Declare(ctx, labelNames); err != nil { _ = runner.Close() t.Fatalf("declare runner: %v", err) } - runner.SetCapabilitiesFromDeclare(declResp) poller := poll.New(cfg, cli, runner) go poller.Poll() diff --git a/go.mod b/go.mod index eddbeb87..cb553dcc 100644 --- a/go.mod +++ b/go.mod @@ -5,9 +5,9 @@ go 1.27 toolchain go1.27.3 require ( - connectrpc.com/connect v1.21.0 - gitea.dev/actionslib v1.4.1 - github.com/bmatcuk/doublestar/v4 v4.10.2 + connectrpc.com/connect/v2 v2.0.1 + gitea.dev/actionslib v1.5.0 + github.com/bmatcuk/doublestar/v4 v4.10.3 github.com/coder/websocket v1.8.15 github.com/containerd/errdefs v1.0.0 github.com/creack/pty v1.1.24 diff --git a/go.sum b/go.sum index 589dfb83..d5d71ea3 100644 --- a/go.sum +++ b/go.sum @@ -1,7 +1,7 @@ -connectrpc.com/connect v1.21.0 h1:LhqSJt7jHf5NJBo9Jq/t/9FjcYAideif0mg+qe2jCUs= -connectrpc.com/connect v1.21.0/go.mod h1:A2ygJrukXwWy32vkCAAHNVguZrqZ+jeZ9rGRnGR4dN4= -gitea.dev/actionslib v1.4.1 h1:DWlUnzDfuuMArQWS6yZ+vwJu9av7sZVMOox7li/kyu4= -gitea.dev/actionslib v1.4.1/go.mod h1:RbXaZudZ04YndjBBxZJF9rZAqU6SKGkaSD1sob5lQbE= +connectrpc.com/connect/v2 v2.0.1 h1:WTAycxe87UqxSM5oxS30d1dalZ+C5xz8Ec8O6zVYkHQ= +connectrpc.com/connect/v2 v2.0.1/go.mod h1:N32OLrstcPUzkykFynnefFro3+xzSdOrZlK2jHmS/xo= +gitea.dev/actionslib v1.5.0 h1:dC3v2HTWbbvyNBSdg2W3kd4WNC/p+FnZaKpl1MdzM00= +gitea.dev/actionslib v1.5.0/go.mod h1:C9wmOFgDBCXyRvg4A3KTX8Vszs8v9+elH4QLGVDolqU= github.com/Microsoft/go-winio v0.6.2 h1:F2VQgta7ecxGYO8k3ZZz3RS8fVIXVxONVUPlNERoyfY= github.com/Microsoft/go-winio v0.6.2/go.mod h1:yd8OoFMLzJbo9gZq8j5qaps8bJ9aShtEA8Ipt1oGCvU= github.com/aws/aws-sdk-go-v2 v1.41.5 h1:dj5kopbwUsVUVFgO4Fi5BIT3t4WyqIDjGKCangnV/yY= @@ -32,8 +32,8 @@ github.com/aws/smithy-go v1.24.2 h1:FzA3bu/nt/vDvmnkg+R8Xl46gmzEDam6mZ1hzmwXFng= github.com/aws/smithy-go v1.24.2/go.mod h1:YE2RhdIuDbA5E5bTdciG9KrW3+TiEONeUWCqxX9i1Fc= github.com/beorn7/perks v1.0.1 h1:VlbKKnNfV8bJzeqoa4cOKqO6bYr3WgKZxO8Z16+hsOM= github.com/beorn7/perks v1.0.1/go.mod h1:G2ZrVWU2WbWT9wwq4/hrbKbnv/1ERSJQ0ibhJ6rlkpw= -github.com/bmatcuk/doublestar/v4 v4.10.2 h1:eF7W7HWKg3z9NrWV9pTLnNeoXaqq3Tq9DNKXVMfoCnw= -github.com/bmatcuk/doublestar/v4 v4.10.2/go.mod h1:xBQ8jztBU6kakFMg+8WGxn0c6z1fTSPVIjEY1Wr7jzc= +github.com/bmatcuk/doublestar/v4 v4.10.3 h1:1UIv1+fqH3++ViVkpnaPTB+Pi1UHycnPFAq14zgK+gI= +github.com/bmatcuk/doublestar/v4 v4.10.3/go.mod h1:xBQ8jztBU6kakFMg+8WGxn0c6z1fTSPVIjEY1Wr7jzc= github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs= github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs= github.com/cevatbarisyilmaz/ara v0.0.4 h1:SGH10hXpBJhhTlObuZzTuFn1rrdmjQImITXnZVPSodc= diff --git a/internal/app/cmd/daemon.go b/internal/app/cmd/daemon.go index f630b445..c2d6dbbd 100644 --- a/internal/app/cmd/daemon.go +++ b/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() { diff --git a/internal/app/cmd/daemon_test.go b/internal/app/cmd/daemon_test.go index a13cb580..41bf8167 100644 --- a/internal/app/cmd/daemon_test.go +++ b/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()) } diff --git a/internal/app/cmd/register.go b/internal/app/cmd/register.go index c2846fb3..3cce9394 100644 --- a/internal/app/cmd/register.go +++ b/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") } diff --git a/internal/app/poll/poller.go b/internal/app/poll/poller.go index 3fb04f96..8280fc48 100644 --- a/internal/app/poll/poller.go +++ b/internal/app/poll/poller.go @@ -17,7 +17,7 @@ import ( "gitea.com/gitea/runner/internal/pkg/config" "gitea.com/gitea/runner/internal/pkg/metrics" - "connectrpc.com/connect" + "connectrpc.com/connect/v2" runnerv1 "gitea.dev/actionslib/runner/v1" ) @@ -355,9 +355,9 @@ func (p *Poller) fetchTask(ctx context.Context, s *workerState) (*runnerv1.Task, // Load the version value that was in the cache when the request was sent. v := p.tasksVersion.Load() start := time.Now() - resp, err := p.client.FetchTask(reqCtx, connect.NewRequest(&runnerv1.FetchTaskRequest{ + resp, err := p.client.FetchTask(reqCtx, &runnerv1.FetchTaskRequest{ TasksVersion: v, - })) + }) // Our own deadline proves nothing either way: today Gitea answers immediately, // so it means a slow server, and once it holds the request open it is an idle @@ -405,27 +405,27 @@ func (p *Poller) fetchTask(ctx context.Context, s *workerState) (*runnerv1.Task, // Successful response — reset error counter. s.consecutiveErrors = 0 - if resp == nil || resp.Msg == nil { + if resp == nil { s.consecutiveEmpty++ metrics.PollFetchTotal.WithLabelValues(metrics.LabelResultEmpty).Inc() return nil, false } - if resp.Msg.TasksVersion > v { - p.tasksVersion.CompareAndSwap(v, resp.Msg.TasksVersion) + if resp.TasksVersion > v { + p.tasksVersion.CompareAndSwap(v, resp.TasksVersion) } - if resp.Msg.Task == nil { + if resp.Task == nil { s.consecutiveEmpty++ metrics.PollFetchTotal.WithLabelValues(metrics.LabelResultEmpty).Inc() return nil, false } // got a task, set `tasksVersion` to zero to force query db in next request. - p.tasksVersion.CompareAndSwap(resp.Msg.TasksVersion, 0) + p.tasksVersion.CompareAndSwap(resp.TasksVersion, 0) metrics.PollFetchTotal.WithLabelValues(metrics.LabelResultTask).Inc() - return resp.Msg.Task, true + return resp.Task, true } func (p *Poller) markHealthyPoll() { diff --git a/internal/app/poll/poller_test.go b/internal/app/poll/poller_test.go index f943c4d3..b82590eb 100644 --- a/internal/app/poll/poller_test.go +++ b/internal/app/poll/poller_test.go @@ -15,7 +15,7 @@ import ( "gitea.com/gitea/runner/internal/pkg/client/mocks" "gitea.com/gitea/runner/internal/pkg/config" - connect_go "connectrpc.com/connect" + "connectrpc.com/connect/v2" runnerv1 "gitea.dev/actionslib/runner/v1" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/mock" @@ -28,9 +28,9 @@ import ( func TestPoller_WorkerStateCounters(t *testing.T) { client := mocks.NewClient(t) client.On("FetchTask", mock.Anything, mock.Anything).Return( - func(_ context.Context, _ *connect_go.Request[runnerv1.FetchTaskRequest]) (*connect_go.Response[runnerv1.FetchTaskResponse], error) { + func(_ context.Context, _ *runnerv1.FetchTaskRequest) (*runnerv1.FetchTaskResponse, error) { // Always return an empty response. - return connect_go.NewResponse(&runnerv1.FetchTaskResponse{}), nil + return &runnerv1.FetchTaskResponse{}, nil }, ) @@ -109,8 +109,8 @@ func TestPoller_FetchUnauthenticatedStopsPolling(t *testing.T) { fetchErr error rejected bool }{ - "proxy": {connect_go.NewError(connect_go.CodeUnauthenticated, errors.New("401 Unauthorized")), false}, - "gitea": {connect_go.NewWireError(connect_go.CodeUnknown, errors.New("rpc error: code = Unauthenticated desc = unregistered runner")), true}, + "proxy": {connect.NewError(connect.CodeUnauthenticated, "401 Unauthorized"), false}, + "gitea": {connect.NewError(connect.CodeUnknown, "rpc error: code = Unauthenticated desc = unregistered runner").WithRemote(), true}, } { t.Run(name, func(t *testing.T) { client := mocks.NewClient(t) @@ -242,7 +242,7 @@ func TestPollerPausesAndResumesForLocalAvailability(t *testing.T) { cli := mocks.NewClient(t) cli.On("FetchTask", mock.Anything, mock.Anything).Maybe().Run(func(mock.Arguments) { fetches.Add(1) - }).Return(connect_go.NewResponse(&runnerv1.FetchTaskResponse{}), nil) + }).Return(&runnerv1.FetchTaskResponse{}, nil) cfg, err := config.LoadDefault("") require.NoError(t, err) @@ -294,8 +294,8 @@ func (m *idleAwareRunner) OnIdle(_ context.Context) { func TestPollerPollCallsOnIdle(t *testing.T) { cli := mocks.NewClient(t) cli.On("FetchTask", mock.Anything, mock.Anything).Return( - func(_ context.Context, _ *connect_go.Request[runnerv1.FetchTaskRequest]) (*connect_go.Response[runnerv1.FetchTaskResponse], error) { - return connect_go.NewResponse(&runnerv1.FetchTaskResponse{}), nil + func(_ context.Context, _ *runnerv1.FetchTaskRequest) (*runnerv1.FetchTaskResponse, error) { + return &runnerv1.FetchTaskResponse{}, nil }, ) @@ -324,8 +324,8 @@ func TestPollerPollCallsOnIdle(t *testing.T) { func TestPollerPollOnceCallsOnIdle(t *testing.T) { cli := mocks.NewClient(t) cli.On("FetchTask", mock.Anything, mock.Anything).Return( - func(_ context.Context, _ *connect_go.Request[runnerv1.FetchTaskRequest]) (*connect_go.Response[runnerv1.FetchTaskResponse], error) { - return connect_go.NewResponse(&runnerv1.FetchTaskResponse{}), nil + func(_ context.Context, _ *runnerv1.FetchTaskRequest) (*runnerv1.FetchTaskResponse, error) { + return &runnerv1.FetchTaskResponse{}, nil }, ) @@ -352,7 +352,7 @@ func TestPollerPollOnceCallsOnIdle(t *testing.T) { func TestPollerPollOnceCallsOnTaskFetchedBeforeRunningIt(t *testing.T) { cli := mocks.NewClient(t) - cli.On("FetchTask", mock.Anything, mock.Anything).Return(connect_go.NewResponse(&runnerv1.FetchTaskResponse{Task: &runnerv1.Task{Id: 1}}), nil) + cli.On("FetchTask", mock.Anything, mock.Anything).Return(&runnerv1.FetchTaskResponse{Task: &runnerv1.Task{Id: 1}}, nil) cfg, err := config.LoadDefault("") require.NoError(t, err) runner := &mockRunner{} @@ -380,17 +380,17 @@ func TestPoller_ConcurrencyLimitedByCapacity(t *testing.T) { cli := mocks.NewClient(t) cli.On("FetchTask", mock.Anything, mock.Anything).Return( - func(_ context.Context, _ *connect_go.Request[runnerv1.FetchTaskRequest]) (*connect_go.Response[runnerv1.FetchTaskResponse], error) { + func(_ context.Context, _ *runnerv1.FetchTaskRequest) (*runnerv1.FetchTaskResponse, error) { atomicMax(&maxFetchConcur, fetchConcur.Add(1)) defer fetchConcur.Add(-1) n := tasksReturned.Add(1) if n <= totalTasks { - return connect_go.NewResponse(&runnerv1.FetchTaskResponse{ + return &runnerv1.FetchTaskResponse{ Task: &runnerv1.Task{Id: n}, - }), nil + }, nil } - return connect_go.NewResponse(&runnerv1.FetchTaskResponse{}), nil + return &runnerv1.FetchTaskResponse{}, nil }, ) @@ -438,13 +438,13 @@ func TestPoller_ShutdownForcesJobsOnTimeout(t *testing.T) { var served atomic.Bool cli := mocks.NewClient(t) cli.On("FetchTask", mock.Anything, mock.Anything).Return( - func(_ context.Context, _ *connect_go.Request[runnerv1.FetchTaskRequest]) (*connect_go.Response[runnerv1.FetchTaskResponse], error) { + func(_ context.Context, _ *runnerv1.FetchTaskRequest) (*runnerv1.FetchTaskResponse, error) { if served.CompareAndSwap(false, true) { - return connect_go.NewResponse(&runnerv1.FetchTaskResponse{ + return &runnerv1.FetchTaskResponse{ Task: &runnerv1.Task{Id: 1}, - }), nil + }, nil } - return connect_go.NewResponse(&runnerv1.FetchTaskResponse{}), nil + return &runnerv1.FetchTaskResponse{}, nil }, ) diff --git a/internal/app/run/runner.go b/internal/app/run/runner.go index 5acdca9a..f5a59bbb 100644 --- a/internal/app/run/runner.go +++ b/internal/app/run/runner.go @@ -41,7 +41,7 @@ import ( "gitea.com/gitea/runner/internal/pkg/telemetry" "gitea.com/gitea/runner/internal/pkg/ver" - "connectrpc.com/connect" + "connectrpc.com/connect/v2" "gitea.dev/actionslib/pkg/model" runnerv1 "gitea.dev/actionslib/runner/v1" docker_container "github.com/moby/moby/api/types/container" @@ -293,14 +293,6 @@ func (r *Runner) cleanupStaleDirs(ctx context.Context, root string, match func(n } } -func (r *Runner) SetCapabilitiesFromDeclare(resp *connect.Response[runnerv1.DeclareResponse]) { - if resp == nil { - return - } - // Capability negotiation is done via response headers to avoid a hard proto bump. - r.capabilities = strings.TrimSpace(resp.Header().Get("X-Gitea-Actions-Capabilities")) -} - func (r *Runner) Run(ctx context.Context, task *runnerv1.Task) error { if _, ok := r.runningTasks.LoadOrStore(task.Id, struct{}{}); ok { return fmt.Errorf("task %d is already running", task.Id) @@ -851,12 +843,17 @@ func nearestExistingPath(path string) string { return "." } -func (r *Runner) Declare(ctx context.Context, labels []string) (*connect.Response[runnerv1.DeclareResponse], error) { - return r.client.Declare(ctx, connect.NewRequest(&runnerv1.DeclareRequest{ +func (r *Runner) Declare(ctx context.Context, labels []string) (*runnerv1.DeclareResponse, error) { + ctx, info := connect.NewClientContext(ctx) + resp, err := r.client.Declare(ctx, &runnerv1.DeclareRequest{ Version: ver.Version(), Labels: labels, Capabilities: RunnerCapabilities(), - })) + }) + if err == nil { + r.capabilities = strings.TrimSpace(info.ResponseHeader().Get("X-Gitea-Actions-Capabilities")) + } + return resp, err } // minFreeDisk keeps the cache from growing past the point where the runner stops taking diff --git a/internal/app/run/runner_test.go b/internal/app/run/runner_test.go index 0e1cf4e9..bec97f49 100644 --- a/internal/app/run/runner_test.go +++ b/internal/app/run/runner_test.go @@ -25,7 +25,7 @@ import ( "gitea.com/gitea/runner/internal/pkg/labels" "gitea.com/gitea/runner/internal/pkg/ver" - "connectrpc.com/connect" + "connectrpc.com/connect/v2" "gitea.dev/actionslib/pkg/model" runnerv1 "gitea.dev/actionslib/runner/v1" "github.com/stretchr/testify/assert" @@ -38,27 +38,21 @@ func TestRunnerCapabilitiesAndDeclare(t *testing.T) { require.Equal(t, []string{CapabilityCancelling}, RunnerCapabilities()) cli := clientmocks.NewClient(t) - cli.On("Declare", mock.Anything, mock.MatchedBy(func(req *connect.Request[runnerv1.DeclareRequest]) bool { - return req.Msg.Version == ver.Version() && - len(req.Msg.Labels) == 1 && - req.Msg.Labels[0] == "ubuntu" && - len(req.Msg.Capabilities) == 1 && - req.Msg.Capabilities[0] == CapabilityCancelling - })).Return(connect.NewResponse(&runnerv1.DeclareResponse{}), nil) + cli.On("Declare", mock.Anything, mock.MatchedBy(func(req *runnerv1.DeclareRequest) bool { + return req.Version == ver.Version() && + len(req.Labels) == 1 && + req.Labels[0] == "ubuntu" && + len(req.Capabilities) == 1 && + req.Capabilities[0] == CapabilityCancelling + })).Return(&runnerv1.DeclareResponse{}, nil).Run(func(args mock.Arguments) { + info, ok := connect.CallInfoForClientContext(args.Get(0).(context.Context)) + require.True(t, ok) + info.ResponseHeader().Set("X-Gitea-Actions-Capabilities", " cancelling,cache-v2 ") + }) r := &Runner{client: cli} _, err := r.Declare(t.Context(), []string{"ubuntu"}) require.NoError(t, err) -} - -func TestRunnerSetCapabilitiesFromDeclare(t *testing.T) { - r := &Runner{} - r.SetCapabilitiesFromDeclare(nil) - require.Empty(t, r.capabilities) - - resp := connect.NewResponse(&runnerv1.DeclareResponse{}) - resp.Header().Set("X-Gitea-Actions-Capabilities", " cancelling,cache-v2 ") - r.SetCapabilitiesFromDeclare(resp) require.Equal(t, "cancelling,cache-v2", r.capabilities) } @@ -104,20 +98,20 @@ func TestRunnerReclaimsVolumesAfterReporting(t *testing.T) { r := NewRunner(cfg, &config.Registration{UUID: "runner-1", Labels: []string{"ubuntu:docker://node:20"}}, cli) var reported atomic.Bool var removed atomic.Int64 - cli.On("UpdateLog", mock.Anything, mock.Anything).Return(func(_ context.Context, req *connect.Request[runnerv1.UpdateLogRequest]) (*connect.Response[runnerv1.UpdateLogResponse], error) { + cli.On("UpdateLog", mock.Anything, mock.Anything).Return(func(_ context.Context, req *runnerv1.UpdateLogRequest) (*runnerv1.UpdateLogResponse, error) { assert.False(t, reported.Load()) - return connect.NewResponse(&runnerv1.UpdateLogResponse{AckIndex: req.Msg.Index + int64(len(req.Msg.Rows))}), nil + return &runnerv1.UpdateLogResponse{AckIndex: req.Index + int64(len(req.Rows))}, nil }) - cli.On("UpdateTask", mock.Anything, mock.Anything).Return(func(_ context.Context, req *connect.Request[runnerv1.UpdateTaskRequest]) (*connect.Response[runnerv1.UpdateTaskResponse], error) { - if req.Msg.State.Result != runnerv1.Result_RESULT_UNSPECIFIED { - assert.Equal(t, runnerv1.Result_RESULT_SUCCESS, req.Msg.State.Result) + cli.On("UpdateTask", mock.Anything, mock.Anything).Return(func(_ context.Context, req *runnerv1.UpdateTaskRequest) (*runnerv1.UpdateTaskResponse, error) { + if req.State.Result != runnerv1.Result_RESULT_UNSPECIFIED { + assert.Equal(t, runnerv1.Result_RESULT_SUCCESS, req.State.Result) assert.Equal(t, int64(1), r.RunningCount()) if mode == "post-task script" { assert.FileExists(t, cfg.Runner.PostTaskScript+".done") } reported.Store(true) } - return connect.NewResponse(&runnerv1.UpdateTaskResponse{State: req.Msg.State}), nil + return &runnerv1.UpdateTaskResponse{State: req.State}, nil }) var volumesCreated atomic.Bool daemon := httptest.NewServer(http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) { diff --git a/internal/pkg/client/client.go b/internal/pkg/client/client.go index ef77ed6d..0ca75834 100644 --- a/internal/pkg/client/client.go +++ b/internal/pkg/client/client.go @@ -8,17 +8,17 @@ import ( "errors" "strings" - "connectrpc.com/connect" - "gitea.dev/actionslib/runner/v1" + "connectrpc.com/connect/v2" + runnerv1 "gitea.dev/actionslib/runner/v1" ) // A Client manages communication with the runner. type Client interface { Address() string - Declare(context.Context, *connect.Request[runnerv1.DeclareRequest]) (*connect.Response[runnerv1.DeclareResponse], error) - FetchTask(context.Context, *connect.Request[runnerv1.FetchTaskRequest]) (*connect.Response[runnerv1.FetchTaskResponse], error) - UpdateLog(context.Context, *connect.Request[runnerv1.UpdateLogRequest]) (*connect.Response[runnerv1.UpdateLogResponse], error) - UpdateTask(context.Context, *connect.Request[runnerv1.UpdateTaskRequest]) (*connect.Response[runnerv1.UpdateTaskResponse], error) + Declare(context.Context, *runnerv1.DeclareRequest) (*runnerv1.DeclareResponse, error) + FetchTask(context.Context, *runnerv1.FetchTaskRequest) (*runnerv1.FetchTaskResponse, error) + UpdateLog(context.Context, *runnerv1.UpdateLogRequest) (*runnerv1.UpdateLogResponse, error) + UpdateTask(context.Context, *runnerv1.UpdateTaskRequest) (*runnerv1.UpdateTaskResponse, error) } // IsRegistrationRejected reports whether Gitea itself answered "unregistered runner", unlike a proxy's 401. diff --git a/internal/pkg/client/http.go b/internal/pkg/client/http.go index fad6c1e9..9a6a2fdf 100644 --- a/internal/pkg/client/http.go +++ b/internal/pkg/client/http.go @@ -13,7 +13,8 @@ import ( "gitea.com/gitea/runner/internal/pkg/ver" - "connectrpc.com/connect" + "connectrpc.com/connect/v2" + "connectrpc.com/connect/v2/connecthttp" "gitea.dev/actionslib/ping/v1/pingv1connect" "gitea.dev/actionslib/pkg/protocol" "gitea.dev/actionslib/runner/v1/runnerv1connect" @@ -57,40 +58,33 @@ func getHTTPClient(endpoint string, tlsConfig *tls.Config, timeout time.Duration // New returns a new runner client. timeout bounds every RPC: without it a // stalled connection parks the reporter for the whole job context, so logs and // heartbeats stop together and the task is reaped as a zombie. -func New(endpoint string, tlsConfig *tls.Config, uuid, token string, timeout time.Duration, extraHeaders map[string]string, opts ...connect.ClientOption) *HTTPClient { +func New(endpoint string, tlsConfig *tls.Config, uuid, token string, timeout time.Duration, extraHeaders map[string]string, opts ...connecthttp.Option) *HTTPClient { baseURL := strings.TrimRight(endpoint, "/") + "/api/actions" - opts = append(opts, connect.WithInterceptors(connect.UnaryInterceptorFunc(func(next connect.UnaryFunc) connect.UnaryFunc { - return func(ctx context.Context, req connect.AnyRequest) (connect.AnyResponse, error) { + opts = append([]connecthttp.Option{connecthttp.WithReadMaxBytes(0)}, opts...) + rpcClient := connect.NewClient(connecthttp.NewTransport(getHTTPClient(endpoint, tlsConfig, timeout, extraHeaders), baseURL, opts...), func(next connect.ClientFunc) connect.ClientFunc { + return func(ctx context.Context, spec connect.Spec) (connect.ClientStream, error) { + info, _ := connect.CallInfoForClientContext(ctx) for name, value := range extraHeaders { - if req.Header().Get(name) == "" { // keep protocol headers such as Content-Type - req.Header().Set(name, value) + if info.RequestHeader().Get(name) == "" { + info.RequestHeader().Set(name, value) } } - req.Header().Set("User-Agent", "gitea-runner/"+ver.Version()) + info.RequestHeader().Set("User-Agent", "gitea-runner/"+ver.Version()) if uuid != "" { - req.Header().Set(protocol.UUIDHeader, uuid) + info.RequestHeader().Set(protocol.UUIDHeader, uuid) } if token != "" { - req.Header().Set(protocol.TokenHeader, token) + info.RequestHeader().Set(protocol.TokenHeader, token) } - return next(ctx, req) + return next(ctx, spec) } - }))) + }) - httpClient := getHTTPClient(endpoint, tlsConfig, timeout, extraHeaders) return &HTTPClient{ - PingServiceClient: pingv1connect.NewPingServiceClient( - httpClient, - baseURL, - opts..., - ), - RunnerServiceClient: runnerv1connect.NewRunnerServiceClient( - httpClient, - baseURL, - opts..., - ), - endpoint: endpoint, + PingServiceClient: pingv1connect.NewPingServiceClient(rpcClient), + RunnerServiceClient: runnerv1connect.NewRunnerServiceClient(rpcClient), + endpoint: endpoint, } } diff --git a/internal/pkg/client/http_test.go b/internal/pkg/client/http_test.go index f532b1b7..f84542b6 100644 --- a/internal/pkg/client/http_test.go +++ b/internal/pkg/client/http_test.go @@ -12,10 +12,12 @@ import ( "testing" "time" - "connectrpc.com/connect" + "gitea.com/gitea/runner/internal/pkg/ver" + pingv1 "gitea.dev/actionslib/ping/v1" "gitea.dev/actionslib/pkg/protocol" "github.com/stretchr/testify/require" + "google.golang.org/protobuf/proto" ) func TestGetHTTPClientUsesProxyFromEnvironment(t *testing.T) { @@ -44,10 +46,14 @@ func TestGetHTTPClientAppliesTLSConfigOnlyToHTTPS(t *testing.T) { func TestNewSetsBaseURLAndHeaders(t *testing.T) { var gotPath string gotHeaders := make(http.Header) + data := strings.Repeat("x", 4<<20) + body, err := proto.Marshal(&pingv1.PingResponse{Data: data}) + require.NoError(t, err) server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { gotPath = r.URL.Path gotHeaders = r.Header.Clone() - w.WriteHeader(http.StatusInternalServerError) + w.Header().Set("Content-Type", "application/proto") + _, _ = w.Write(body) })) defer server.Close() @@ -61,14 +67,19 @@ func TestNewSetsBaseURLAndHeaders(t *testing.T) { // Address returns the endpoint as supplied (untrimmed) require.Equal(t, server.URL+"/", c.Address()) - // the call is expected to fail (server returns 500), we only assert what was sent - _, _ = c.Ping(t.Context(), connect.NewRequest(&pingv1.PingRequest{Data: "hi"})) + resp, err := c.Ping(t.Context(), &pingv1.PingRequest{Data: "hi"}) + require.NoError(t, err) + require.Equal(t, data, resp.Data) require.True(t, strings.HasPrefix(gotPath, "/api/actions/"), "unexpected path %q", gotPath) require.Equal(t, "the-uuid", gotHeaders.Get(protocol.UUIDHeader)) require.Equal(t, "the-token", gotHeaders.Get(protocol.TokenHeader)) require.Equal(t, "proxy-token", gotHeaders.Get("X-Proxy-Token")) require.Equal(t, "application/proto", gotHeaders.Get("Content-Type")) + require.Equal(t, "gitea-runner/"+ver.Version(), gotHeaders.Get("User-Agent")) + require.Equal(t, "1", gotHeaders.Get("Connect-Protocol-Version")) + require.Equal(t, "gzip", gotHeaders.Get("Accept-Encoding")) + require.Empty(t, gotHeaders.Get("Content-Encoding")) } func TestNewOmitsEmptyHeaders(t *testing.T) { @@ -80,7 +91,7 @@ func TestNewOmitsEmptyHeaders(t *testing.T) { defer server.Close() c := New(server.URL, nil, "", "", time.Minute, nil) - _, _ = c.Ping(t.Context(), connect.NewRequest(&pingv1.PingRequest{Data: "hi"})) + _, _ = c.Ping(t.Context(), &pingv1.PingRequest{Data: "hi"}) require.Empty(t, gotHeaders.Get(protocol.UUIDHeader)) require.Empty(t, gotHeaders.Get(protocol.TokenHeader)) @@ -93,7 +104,7 @@ func TestNewStopsRedirectLoop(t *testing.T) { defer server.Close() c := New(server.URL, nil, "", "", time.Minute, nil) - _, err := c.Ping(t.Context(), connect.NewRequest(&pingv1.PingRequest{Data: "hi"})) + _, err := c.Ping(t.Context(), &pingv1.PingRequest{Data: "hi"}) require.ErrorContains(t, err, "stopped after 10 redirects") } @@ -119,7 +130,7 @@ func TestNewRedirectKeepsExtraHeadersOnlyOnSameHost(t *testing.T) { gotHeaders = nil origin := tc.newOrigin(http.RedirectHandler("http://"+net.JoinHostPort(tc.host, port), http.StatusPermanentRedirect)) c := New(origin.URL, &tls.Config{InsecureSkipVerify: true}, "the-uuid", "the-token", time.Minute, map[string]string{"X-Proxy-Token": "proxy-token", "Content-Type": "text/plain"}) - _, _ = c.Ping(t.Context(), connect.NewRequest(&pingv1.PingRequest{Data: "hi"})) + _, _ = c.Ping(t.Context(), &pingv1.PingRequest{Data: "hi"}) origin.Close() require.NotNil(t, gotHeaders, origin.URL) diff --git a/internal/pkg/client/mocks/Client.go b/internal/pkg/client/mocks/Client.go index 8f10850c..8513c726 100644 --- a/internal/pkg/client/mocks/Client.go +++ b/internal/pkg/client/mocks/Client.go @@ -7,8 +7,7 @@ import ( "context" "fmt" - "connectrpc.com/connect" - "gitea.dev/actionslib/runner/v1" + runnerv1 "gitea.dev/actionslib/runner/v1" "github.com/stretchr/testify/mock" ) @@ -21,15 +20,15 @@ func (m *Client) Address() string { return m.AddressValue } -func call[Request, Response any](ctx context.Context, m *Client, method string, request *connect.Request[Request]) (*connect.Response[Response], error) { +func call[Request, Response any](ctx context.Context, m *Client, method string, request *Request) (*Response, error) { returns := m.MethodCalled(method, ctx, request) - if callback, ok := returns.Get(0).(func(context.Context, *connect.Request[Request]) (*connect.Response[Response], error)); ok { + if callback, ok := returns.Get(0).(func(context.Context, *Request) (*Response, error)); ok { return callback(ctx, request) } - var response *connect.Response[Response] + var response *Response if value := returns.Get(0); value != nil { var ok bool - response, ok = value.(*connect.Response[Response]) + response, ok = value.(*Response) if !ok { panic(fmt.Sprintf("unexpected response type %T for %s", value, method)) } @@ -37,19 +36,19 @@ func call[Request, Response any](ctx context.Context, m *Client, method string, return response, returns.Error(1) } -func (m *Client) Declare(ctx context.Context, request *connect.Request[runnerv1.DeclareRequest]) (*connect.Response[runnerv1.DeclareResponse], error) { +func (m *Client) Declare(ctx context.Context, request *runnerv1.DeclareRequest) (*runnerv1.DeclareResponse, error) { return call[runnerv1.DeclareRequest, runnerv1.DeclareResponse](ctx, m, "Declare", request) } -func (m *Client) FetchTask(ctx context.Context, request *connect.Request[runnerv1.FetchTaskRequest]) (*connect.Response[runnerv1.FetchTaskResponse], error) { +func (m *Client) FetchTask(ctx context.Context, request *runnerv1.FetchTaskRequest) (*runnerv1.FetchTaskResponse, error) { return call[runnerv1.FetchTaskRequest, runnerv1.FetchTaskResponse](ctx, m, "FetchTask", request) } -func (m *Client) UpdateLog(ctx context.Context, request *connect.Request[runnerv1.UpdateLogRequest]) (*connect.Response[runnerv1.UpdateLogResponse], error) { +func (m *Client) UpdateLog(ctx context.Context, request *runnerv1.UpdateLogRequest) (*runnerv1.UpdateLogResponse, error) { return call[runnerv1.UpdateLogRequest, runnerv1.UpdateLogResponse](ctx, m, "UpdateLog", request) } -func (m *Client) UpdateTask(ctx context.Context, request *connect.Request[runnerv1.UpdateTaskRequest]) (*connect.Response[runnerv1.UpdateTaskResponse], error) { +func (m *Client) UpdateTask(ctx context.Context, request *runnerv1.UpdateTaskRequest) (*runnerv1.UpdateTaskResponse, error) { return call[runnerv1.UpdateTaskRequest, runnerv1.UpdateTaskResponse](ctx, m, "UpdateTask", request) } diff --git a/internal/pkg/report/joblog_test.go b/internal/pkg/report/joblog_test.go index 3abcebbe..3f1675e1 100644 --- a/internal/pkg/report/joblog_test.go +++ b/internal/pkg/report/joblog_test.go @@ -15,7 +15,6 @@ import ( "gitea.com/gitea/runner/internal/pkg/client/mocks" "gitea.com/gitea/runner/internal/pkg/config" - connect_go "connectrpc.com/connect" runnerv1 "gitea.dev/actionslib/runner/v1" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/mock" @@ -34,10 +33,10 @@ func readJobLog(t *testing.T, joblog *jobLog) string { func TestJobLog_MirrorsUploadedRows(t *testing.T) { client := mocks.NewClient(t) - client.On("UpdateLog", mock.Anything, mock.Anything).Return(func(_ context.Context, req *connect_go.Request[runnerv1.UpdateLogRequest]) (*connect_go.Response[runnerv1.UpdateLogResponse], error) { - return connect_go.NewResponse(&runnerv1.UpdateLogResponse{AckIndex: req.Msg.Index + int64(len(req.Msg.Rows))}), nil + client.On("UpdateLog", mock.Anything, mock.Anything).Return(func(_ context.Context, req *runnerv1.UpdateLogRequest) (*runnerv1.UpdateLogResponse, error) { + return &runnerv1.UpdateLogResponse{AckIndex: req.Index + int64(len(req.Rows))}, nil }) - client.On("UpdateTask", mock.Anything, mock.Anything).Return(connect_go.NewResponse(&runnerv1.UpdateTaskResponse{}), nil) + client.On("UpdateTask", mock.Anything, mock.Anything).Return(&runnerv1.UpdateTaskResponse{}, nil) cfg, err := config.LoadDefault("") require.NoError(t, err) diff --git a/internal/pkg/report/reporter.go b/internal/pkg/report/reporter.go index 89a4927e..c8e742ff 100644 --- a/internal/pkg/report/reporter.go +++ b/internal/pkg/report/reporter.go @@ -21,7 +21,7 @@ import ( "gitea.com/gitea/runner/internal/pkg/config" "gitea.com/gitea/runner/internal/pkg/metrics" - "connectrpc.com/connect" + "connectrpc.com/connect/v2" "gitea.dev/actionslib/pkg/model" runnerv1 "gitea.dev/actionslib/runner/v1" "google.golang.org/protobuf/proto" @@ -823,12 +823,12 @@ func (r *Reporter) reportLog(noMore bool) (bool, error) { defer rpcCancel() start := time.Now() - resp, err := r.client.UpdateLog(rpcCtx, connect.NewRequest(&runnerv1.UpdateLogRequest{ + resp, err := r.client.UpdateLog(rpcCtx, &runnerv1.UpdateLogRequest{ TaskId: r.state.Id, Index: int64(r.logOffset), Rows: rows, NoMore: noMore, - })) + }) metrics.ReportLogDuration.Observe(time.Since(start).Seconds()) if err != nil { metrics.ReportLogTotal.WithLabelValues(metrics.LabelResultError).Inc() @@ -837,7 +837,7 @@ func (r *Reporter) reportLog(noMore bool) (bool, error) { } metrics.ReportLogTotal.WithLabelValues(metrics.LabelResultSuccess).Inc() - ack := int(resp.Msg.AckIndex) + ack := int(resp.AckIndex) if ack < r.logOffset { return false, errors.New("submitted logs are lost") } @@ -902,10 +902,10 @@ func (r *Reporter) ReportState(reportResult bool) error { defer rpcCancel() start := time.Now() - resp, err := r.client.UpdateTask(rpcCtx, connect.NewRequest(&runnerv1.UpdateTaskRequest{ + resp, err := r.client.UpdateTask(rpcCtx, &runnerv1.UpdateTaskRequest{ State: state, Outputs: outputs, - })) + }) metrics.ReportStateDuration.Observe(time.Since(start).Seconds()) if err != nil { metrics.ReportStateTotal.WithLabelValues(metrics.LabelResultError).Inc() @@ -920,7 +920,7 @@ func (r *Reporter) ReportState(reportResult bool) error { var noSent []string r.outputsMu.Lock() - for _, k := range resp.Msg.SentOutputs { + for _, k := range resp.SentOutputs { if _, ok := r.outputs[k]; ok { r.outputs[k] = jobOutput{sent: true} } @@ -934,7 +934,7 @@ func (r *Reporter) ReportState(reportResult bool) error { // A terminal result means the server is done with this task; keep running and // the job holds a capacity slot until runner.timeout. - if state := resp.Msg.State; state != nil && state.Result != runnerv1.Result_RESULT_UNSPECIFIED { + if state := resp.State; state != nil && state.Result != runnerv1.Result_RESULT_UNSPECIFIED { r.stateMu.Lock() r.serverResult = state.Result r.stateMu.Unlock() diff --git a/internal/pkg/report/reporter_test.go b/internal/pkg/report/reporter_test.go index 4d71e8f9..1c5e0f49 100644 --- a/internal/pkg/report/reporter_test.go +++ b/internal/pkg/report/reporter_test.go @@ -25,7 +25,6 @@ import ( "gitea.com/gitea/runner/internal/pkg/config" "gitea.com/gitea/runner/internal/pkg/metrics" - connect_go "connectrpc.com/connect" "gitea.dev/actionslib/pkg/model" runnerv1 "gitea.dev/actionslib/runner/v1" "github.com/stretchr/testify/assert" @@ -57,13 +56,13 @@ func TestReporter_LogLevelFiltering(t *testing.T) { defer logging.SetLevel(origLevel) client := mocks.NewClient(t) - client.On("UpdateLog", mock.Anything, mock.Anything).Return(func(_ context.Context, req *connect_go.Request[runnerv1.UpdateLogRequest]) (*connect_go.Response[runnerv1.UpdateLogResponse], error) { - return connect_go.NewResponse(&runnerv1.UpdateLogResponse{ - AckIndex: req.Msg.Index + int64(len(req.Msg.Rows)), - }), nil + client.On("UpdateLog", mock.Anything, mock.Anything).Return(func(_ context.Context, req *runnerv1.UpdateLogRequest) (*runnerv1.UpdateLogResponse, error) { + return &runnerv1.UpdateLogResponse{ + AckIndex: req.Index + int64(len(req.Rows)), + }, nil }) - client.On("UpdateTask", mock.Anything, mock.Anything).Return(func(_ context.Context, req *connect_go.Request[runnerv1.UpdateTaskRequest]) (*connect_go.Response[runnerv1.UpdateTaskResponse], error) { - return connect_go.NewResponse(&runnerv1.UpdateTaskResponse{}), nil + client.On("UpdateTask", mock.Anything, mock.Anything).Return(func(_ context.Context, req *runnerv1.UpdateTaskRequest) (*runnerv1.UpdateTaskResponse, error) { + return &runnerv1.UpdateTaskResponse{}, nil }) ctx, cancel := context.WithCancel(t.Context()) @@ -119,22 +118,22 @@ func TestReporter_EphemeralRunnerDeletion(t *testing.T) { client := mocks.NewClient(t) client.On("UpdateLog", mock.Anything, mock.Anything).Return( - func(_ context.Context, req *connect_go.Request[runnerv1.UpdateLogRequest]) (*connect_go.Response[runnerv1.UpdateLogResponse], error) { + func(_ context.Context, req *runnerv1.UpdateLogRequest) (*runnerv1.UpdateLogResponse, error) { if runnerDeleted { return nil, errors.New("runner has been deleted") } - return connect_go.NewResponse(&runnerv1.UpdateLogResponse{ - AckIndex: req.Msg.Index + int64(len(req.Msg.Rows)), - }), nil + return &runnerv1.UpdateLogResponse{ + AckIndex: req.Index + int64(len(req.Rows)), + }, nil }, ) client.On("UpdateTask", mock.Anything, mock.Anything).Maybe().Return( - func(_ context.Context, req *connect_go.Request[runnerv1.UpdateTaskRequest]) (*connect_go.Response[runnerv1.UpdateTaskResponse], error) { + func(_ context.Context, req *runnerv1.UpdateTaskRequest) (*runnerv1.UpdateTaskResponse, error) { // Server deletes ephemeral runner when it receives a completed state - if req.Msg.State != nil && req.Msg.State.Result != runnerv1.Result_RESULT_UNSPECIFIED { + if req.State != nil && req.State.Result != runnerv1.Result_RESULT_UNSPECIFIED { runnerDeleted = true } - return connect_go.NewResponse(&runnerv1.UpdateTaskResponse{}), nil + return &runnerv1.UpdateTaskResponse{}, nil }, ) @@ -191,15 +190,15 @@ func TestReporter_EphemeralRunnerDeletion(t *testing.T) { func TestReporter_RunDaemonClose_Race(t *testing.T) { client := mocks.NewClient(t) client.On("UpdateLog", mock.Anything, mock.Anything).Return( - func(_ context.Context, req *connect_go.Request[runnerv1.UpdateLogRequest]) (*connect_go.Response[runnerv1.UpdateLogResponse], error) { - return connect_go.NewResponse(&runnerv1.UpdateLogResponse{ - AckIndex: req.Msg.Index + int64(len(req.Msg.Rows)), - }), nil + func(_ context.Context, req *runnerv1.UpdateLogRequest) (*runnerv1.UpdateLogResponse, error) { + return &runnerv1.UpdateLogResponse{ + AckIndex: req.Index + int64(len(req.Rows)), + }, nil }, ) client.On("UpdateTask", mock.Anything, mock.Anything).Return( - func(_ context.Context, req *connect_go.Request[runnerv1.UpdateTaskRequest]) (*connect_go.Response[runnerv1.UpdateTaskResponse], error) { - return connect_go.NewResponse(&runnerv1.UpdateTaskResponse{}), nil + func(_ context.Context, req *runnerv1.UpdateTaskRequest) (*runnerv1.UpdateTaskResponse, error) { + return &runnerv1.UpdateTaskResponse{}, nil }, ) @@ -232,16 +231,16 @@ func TestReporter_MaxLatencyTimer(t *testing.T) { client := mocks.NewClient(t) client.On("UpdateLog", mock.Anything, mock.Anything).Return( - func(_ context.Context, req *connect_go.Request[runnerv1.UpdateLogRequest]) (*connect_go.Response[runnerv1.UpdateLogResponse], error) { + func(_ context.Context, req *runnerv1.UpdateLogRequest) (*runnerv1.UpdateLogResponse, error) { updateLogCalls.Add(1) - return connect_go.NewResponse(&runnerv1.UpdateLogResponse{ - AckIndex: req.Msg.Index + int64(len(req.Msg.Rows)), - }), nil + return &runnerv1.UpdateLogResponse{ + AckIndex: req.Index + int64(len(req.Rows)), + }, nil }, ) client.On("UpdateTask", mock.Anything, mock.Anything).Maybe().Return( - func(_ context.Context, _ *connect_go.Request[runnerv1.UpdateTaskRequest]) (*connect_go.Response[runnerv1.UpdateTaskResponse], error) { - return connect_go.NewResponse(&runnerv1.UpdateTaskResponse{}), nil + func(_ context.Context, _ *runnerv1.UpdateTaskRequest) (*runnerv1.UpdateTaskResponse, error) { + return &runnerv1.UpdateTaskResponse{}, nil }, ) @@ -283,16 +282,16 @@ func TestReporter_BatchSizeFlush(t *testing.T) { client := mocks.NewClient(t) client.On("UpdateLog", mock.Anything, mock.Anything).Return( - func(_ context.Context, req *connect_go.Request[runnerv1.UpdateLogRequest]) (*connect_go.Response[runnerv1.UpdateLogResponse], error) { + func(_ context.Context, req *runnerv1.UpdateLogRequest) (*runnerv1.UpdateLogResponse, error) { updateLogCalls.Add(1) - return connect_go.NewResponse(&runnerv1.UpdateLogResponse{ - AckIndex: req.Msg.Index + int64(len(req.Msg.Rows)), - }), nil + return &runnerv1.UpdateLogResponse{ + AckIndex: req.Index + int64(len(req.Rows)), + }, nil }, ) client.On("UpdateTask", mock.Anything, mock.Anything).Maybe().Return( - func(_ context.Context, _ *connect_go.Request[runnerv1.UpdateTaskRequest]) (*connect_go.Response[runnerv1.UpdateTaskResponse], error) { - return connect_go.NewResponse(&runnerv1.UpdateTaskResponse{}), nil + func(_ context.Context, _ *runnerv1.UpdateTaskRequest) (*runnerv1.UpdateTaskResponse, error) { + return &runnerv1.UpdateTaskResponse{}, nil }, ) @@ -338,14 +337,14 @@ func TestReporter_StateChangedNotLostDuringReport(t *testing.T) { client := mocks.NewClient(t) client.On("UpdateTask", mock.Anything, mock.Anything).Return( - func(_ context.Context, _ *connect_go.Request[runnerv1.UpdateTaskRequest]) (*connect_go.Response[runnerv1.UpdateTaskResponse], error) { + func(_ context.Context, _ *runnerv1.UpdateTaskRequest) (*runnerv1.UpdateTaskResponse, error) { n := updateTaskCalls.Add(1) if n == 1 { // Signal that the first UpdateTask is in flight, then block until released. close(inFlight) <-release } - return connect_go.NewResponse(&runnerv1.UpdateTaskResponse{}), nil + return &runnerv1.UpdateTaskResponse{}, nil }, ) @@ -401,12 +400,12 @@ func TestReporter_StateChangedRestoredOnError(t *testing.T) { client := mocks.NewClient(t) client.On("UpdateTask", mock.Anything, mock.Anything).Return( - func(_ context.Context, _ *connect_go.Request[runnerv1.UpdateTaskRequest]) (*connect_go.Response[runnerv1.UpdateTaskResponse], error) { + func(_ context.Context, _ *runnerv1.UpdateTaskRequest) (*runnerv1.UpdateTaskResponse, error) { n := updateTaskCalls.Add(1) if n == 1 { return nil, errors.New("transient network error") } - return connect_go.NewResponse(&runnerv1.UpdateTaskResponse{}), nil + return &runnerv1.UpdateTaskResponse{}, nil }, ) @@ -442,16 +441,16 @@ func TestReporter_StateNotifyFlush(t *testing.T) { client := mocks.NewClient(t) client.On("UpdateLog", mock.Anything, mock.Anything).Maybe().Return( - func(_ context.Context, req *connect_go.Request[runnerv1.UpdateLogRequest]) (*connect_go.Response[runnerv1.UpdateLogResponse], error) { - return connect_go.NewResponse(&runnerv1.UpdateLogResponse{ - AckIndex: req.Msg.Index + int64(len(req.Msg.Rows)), - }), nil + func(_ context.Context, req *runnerv1.UpdateLogRequest) (*runnerv1.UpdateLogResponse, error) { + return &runnerv1.UpdateLogResponse{ + AckIndex: req.Index + int64(len(req.Rows)), + }, nil }, ) client.On("UpdateTask", mock.Anything, mock.Anything).Return( - func(_ context.Context, _ *connect_go.Request[runnerv1.UpdateTaskRequest]) (*connect_go.Response[runnerv1.UpdateTaskResponse], error) { + func(_ context.Context, _ *runnerv1.UpdateTaskRequest) (*runnerv1.UpdateTaskResponse, error) { updateTaskCalls.Add(1) - return connect_go.NewResponse(&runnerv1.UpdateTaskResponse{}), nil + return &runnerv1.UpdateTaskResponse{}, nil }, ) @@ -496,19 +495,19 @@ func TestReporter_CloseAlwaysSendsRowsWithNoMore(t *testing.T) { client := mocks.NewClient(t) client.On("UpdateLog", mock.Anything, mock.Anything).Return( - func(_ context.Context, req *connect_go.Request[runnerv1.UpdateLogRequest]) (*connect_go.Response[runnerv1.UpdateLogResponse], error) { - lastReq.Store(req.Msg) - if req.Msg.NoMore { + func(_ context.Context, req *runnerv1.UpdateLogRequest) (*runnerv1.UpdateLogResponse, error) { + lastReq.Store(req) + if req.NoMore { noMoreCalls.Add(1) } - return connect_go.NewResponse(&runnerv1.UpdateLogResponse{ - AckIndex: req.Msg.Index + int64(len(req.Msg.Rows)), - }), nil + return &runnerv1.UpdateLogResponse{ + AckIndex: req.Index + int64(len(req.Rows)), + }, nil }, ) client.On("UpdateTask", mock.Anything, mock.Anything).Return( - func(_ context.Context, _ *connect_go.Request[runnerv1.UpdateTaskRequest]) (*connect_go.Response[runnerv1.UpdateTaskResponse], error) { - return connect_go.NewResponse(&runnerv1.UpdateTaskResponse{}), nil + func(_ context.Context, _ *runnerv1.UpdateTaskRequest) (*runnerv1.UpdateTaskResponse, error) { + return &runnerv1.UpdateTaskResponse{}, nil }, ) @@ -557,9 +556,9 @@ func TestReporter_StateHeartbeat(t *testing.T) { client := mocks.NewClient(t) client.On("UpdateTask", mock.Anything, mock.Anything).Return( - func(_ context.Context, _ *connect_go.Request[runnerv1.UpdateTaskRequest]) (*connect_go.Response[runnerv1.UpdateTaskResponse], error) { + func(_ context.Context, _ *runnerv1.UpdateTaskRequest) (*runnerv1.UpdateTaskResponse, error) { updateTaskCalls.Add(1) - return connect_go.NewResponse(&runnerv1.UpdateTaskResponse{}), nil + return &runnerv1.UpdateTaskResponse{}, nil }, ) @@ -605,13 +604,13 @@ func TestReporter_ServerCancelStillFlushesFinal(t *testing.T) { client := mocks.NewClient(t) client.On("UpdateLog", mock.Anything, mock.Anything).Return( - func(_ context.Context, req *connect_go.Request[runnerv1.UpdateLogRequest]) (*connect_go.Response[runnerv1.UpdateLogResponse], error) { - if req.Msg.NoMore { + func(_ context.Context, req *runnerv1.UpdateLogRequest) (*runnerv1.UpdateLogResponse, error) { + if req.NoMore { finalLogNoMoreSeen.Store(true) } - return connect_go.NewResponse(&runnerv1.UpdateLogResponse{ - AckIndex: req.Msg.Index + int64(len(req.Msg.Rows)), - }), nil + return &runnerv1.UpdateLogResponse{ + AckIndex: req.Index + int64(len(req.Rows)), + }, nil }, ) // The first UpdateTask returns RESULT_CANCELLED — modelling a server-side @@ -619,19 +618,19 @@ func TestReporter_ServerCancelStillFlushesFinal(t *testing.T) { // UpdateTask issued by Close() must still arrive even though r.ctx is now // cancelled. client.On("UpdateTask", mock.Anything, mock.Anything).Return( - func(_ context.Context, req *connect_go.Request[runnerv1.UpdateTaskRequest]) (*connect_go.Response[runnerv1.UpdateTaskResponse], error) { + func(_ context.Context, req *runnerv1.UpdateTaskRequest) (*runnerv1.UpdateTaskResponse, error) { n := updateTaskCalls.Add(1) if n == 1 { - return connect_go.NewResponse(&runnerv1.UpdateTaskResponse{ + return &runnerv1.UpdateTaskResponse{ State: &runnerv1.TaskState{ Result: runnerv1.Result_RESULT_CANCELLED, }, - }), nil + }, nil } - if req.Msg.State != nil && req.Msg.State.Result != runnerv1.Result_RESULT_UNSPECIFIED { + if req.State != nil && req.State.Result != runnerv1.Result_RESULT_UNSPECIFIED { finalTaskStateSeen.Store(true) } - return connect_go.NewResponse(&runnerv1.UpdateTaskResponse{}), nil + return &runnerv1.UpdateTaskResponse{}, nil }, ) @@ -685,22 +684,22 @@ func TestReporter_CloseReportsCancelledOnCanceledCtx(t *testing.T) { client := mocks.NewClient(t) client.On("UpdateLog", mock.Anything, mock.Anything).Return( - func(_ context.Context, req *connect_go.Request[runnerv1.UpdateLogRequest]) (*connect_go.Response[runnerv1.UpdateLogResponse], error) { - if req.Msg.NoMore { - rows := append([]*runnerv1.LogRow(nil), req.Msg.Rows...) + func(_ context.Context, req *runnerv1.UpdateLogRequest) (*runnerv1.UpdateLogResponse, error) { + if req.NoMore { + rows := append([]*runnerv1.LogRow(nil), req.Rows...) finalLogRows.Store(&rows) } - return connect_go.NewResponse(&runnerv1.UpdateLogResponse{ - AckIndex: req.Msg.Index + int64(len(req.Msg.Rows)), - }), nil + return &runnerv1.UpdateLogResponse{ + AckIndex: req.Index + int64(len(req.Rows)), + }, nil }, ) client.On("UpdateTask", mock.Anything, mock.Anything).Return( - func(_ context.Context, req *connect_go.Request[runnerv1.UpdateTaskRequest]) (*connect_go.Response[runnerv1.UpdateTaskResponse], error) { - if req.Msg.State != nil && req.Msg.State.Result != runnerv1.Result_RESULT_UNSPECIFIED { - finalState.Store(req.Msg.State) + func(_ context.Context, req *runnerv1.UpdateTaskRequest) (*runnerv1.UpdateTaskResponse, error) { + if req.State != nil && req.State.Result != runnerv1.Result_RESULT_UNSPECIFIED { + finalState.Store(req.State) } - return connect_go.NewResponse(&runnerv1.UpdateTaskResponse{}), nil + return &runnerv1.UpdateTaskResponse{}, nil }, ) @@ -750,16 +749,16 @@ func TestReporter_StopHeartbeats(t *testing.T) { client := mocks.NewClient(t) client.On("UpdateLog", mock.Anything, mock.Anything).Maybe().Return( - func(_ context.Context, req *connect_go.Request[runnerv1.UpdateLogRequest]) (*connect_go.Response[runnerv1.UpdateLogResponse], error) { - return connect_go.NewResponse(&runnerv1.UpdateLogResponse{ - AckIndex: req.Msg.Index + int64(len(req.Msg.Rows)), - }), nil + func(_ context.Context, req *runnerv1.UpdateLogRequest) (*runnerv1.UpdateLogResponse, error) { + return &runnerv1.UpdateLogResponse{ + AckIndex: req.Index + int64(len(req.Rows)), + }, nil }, ) client.On("UpdateTask", mock.Anything, mock.Anything).Return( - func(_ context.Context, _ *connect_go.Request[runnerv1.UpdateTaskRequest]) (*connect_go.Response[runnerv1.UpdateTaskResponse], error) { + func(_ context.Context, _ *runnerv1.UpdateTaskRequest) (*runnerv1.UpdateTaskResponse, error) { updateTaskCalls.Add(1) - return connect_go.NewResponse(&runnerv1.UpdateTaskResponse{}), nil + return &runnerv1.UpdateTaskResponse{}, nil }, ) @@ -895,9 +894,9 @@ func TestReporter_OutputsSentOnce(t *testing.T) { client := mocks.NewClient(t) var reported []map[string]string client.On("UpdateTask", mock.Anything, mock.Anything).Return( - func(_ context.Context, req *connect_go.Request[runnerv1.UpdateTaskRequest]) (*connect_go.Response[runnerv1.UpdateTaskResponse], error) { - reported = append(reported, req.Msg.Outputs) - return connect_go.NewResponse(&runnerv1.UpdateTaskResponse{SentOutputs: slices.Collect(maps.Keys(req.Msg.Outputs))}), nil + func(_ context.Context, req *runnerv1.UpdateTaskRequest) (*runnerv1.UpdateTaskResponse, error) { + reported = append(reported, req.Outputs) + return &runnerv1.UpdateTaskResponse{SentOutputs: slices.Collect(maps.Keys(req.Outputs))}, nil }) ctx, cancel := context.WithCancel(t.Context()) @@ -983,8 +982,8 @@ func TestReporter_masksEncodedSecrets(t *testing.T) { // process down with it. func TestReporter_AckIndexBeyondBuffer(t *testing.T) { client := mocks.NewClient(t) - client.On("UpdateLog", mock.Anything, mock.Anything).Return(func(_ context.Context, _ *connect_go.Request[runnerv1.UpdateLogRequest]) (*connect_go.Response[runnerv1.UpdateLogResponse], error) { - return connect_go.NewResponse(&runnerv1.UpdateLogResponse{AckIndex: 1000}), nil + client.On("UpdateLog", mock.Anything, mock.Anything).Return(func(_ context.Context, _ *runnerv1.UpdateLogRequest) (*runnerv1.UpdateLogResponse, error) { + return &runnerv1.UpdateLogResponse{AckIndex: 1000}, nil }) ctx, cancel := context.WithCancel(t.Context()) @@ -1011,17 +1010,17 @@ func TestReporter_CloseWithStuckDaemon(t *testing.T) { var once sync.Once client := mocks.NewClient(t) - client.On("UpdateLog", mock.Anything, mock.Anything).Return(func(_ context.Context, req *connect_go.Request[runnerv1.UpdateLogRequest]) (*connect_go.Response[runnerv1.UpdateLogResponse], error) { + client.On("UpdateLog", mock.Anything, mock.Anything).Return(func(_ context.Context, req *runnerv1.UpdateLogRequest) (*runnerv1.UpdateLogResponse, error) { once.Do(func() { close(entered) <-release }) - return connect_go.NewResponse(&runnerv1.UpdateLogResponse{ - AckIndex: req.Msg.Index + int64(len(req.Msg.Rows)), - }), nil + return &runnerv1.UpdateLogResponse{ + AckIndex: req.Index + int64(len(req.Rows)), + }, nil }) - client.On("UpdateTask", mock.Anything, mock.Anything).Return(func(_ context.Context, _ *connect_go.Request[runnerv1.UpdateTaskRequest]) (*connect_go.Response[runnerv1.UpdateTaskResponse], error) { - return connect_go.NewResponse(&runnerv1.UpdateTaskResponse{}), nil + client.On("UpdateTask", mock.Anything, mock.Anything).Return(func(_ context.Context, _ *runnerv1.UpdateTaskRequest) (*runnerv1.UpdateTaskResponse, error) { + return &runnerv1.UpdateTaskResponse{}, nil }).Maybe() ctx, cancel := context.WithCancel(t.Context()) @@ -1059,16 +1058,16 @@ func TestReporter_CloseWithStuckDaemon(t *testing.T) { func TestReporter_ServerFailureResultIsReportedAsFailure(t *testing.T) { var lastState *runnerv1.TaskState client := mocks.NewClient(t) - client.On("UpdateTask", mock.Anything, mock.Anything).Return(func(_ context.Context, req *connect_go.Request[runnerv1.UpdateTaskRequest]) (*connect_go.Response[runnerv1.UpdateTaskResponse], error) { - lastState = req.Msg.State - return connect_go.NewResponse(&runnerv1.UpdateTaskResponse{ + client.On("UpdateTask", mock.Anything, mock.Anything).Return(func(_ context.Context, req *runnerv1.UpdateTaskRequest) (*runnerv1.UpdateTaskResponse, error) { + lastState = req.State + return &runnerv1.UpdateTaskResponse{ State: &runnerv1.TaskState{Result: runnerv1.Result_RESULT_FAILURE}, - }), nil + }, nil }) - client.On("UpdateLog", mock.Anything, mock.Anything).Return(func(_ context.Context, req *connect_go.Request[runnerv1.UpdateLogRequest]) (*connect_go.Response[runnerv1.UpdateLogResponse], error) { - return connect_go.NewResponse(&runnerv1.UpdateLogResponse{ - AckIndex: req.Msg.Index + int64(len(req.Msg.Rows)), - }), nil + client.On("UpdateLog", mock.Anything, mock.Anything).Return(func(_ context.Context, req *runnerv1.UpdateLogRequest) (*runnerv1.UpdateLogResponse, error) { + return &runnerv1.UpdateLogResponse{ + AckIndex: req.Index + int64(len(req.Rows)), + }, nil }).Maybe() ctx, cancel := context.WithCancel(t.Context()) @@ -1115,16 +1114,16 @@ func TestReporter_CloseMasksLastWords(t *testing.T) { client := mocks.NewClient(t) client.On("UpdateLog", mock.Anything, mock.Anything).Return( - func(_ context.Context, req *connect_go.Request[runnerv1.UpdateLogRequest]) (*connect_go.Response[runnerv1.UpdateLogResponse], error) { - rows = append(rows, req.Msg.Rows...) - return connect_go.NewResponse(&runnerv1.UpdateLogResponse{ - AckIndex: req.Msg.Index + int64(len(req.Msg.Rows)), - }), nil + func(_ context.Context, req *runnerv1.UpdateLogRequest) (*runnerv1.UpdateLogResponse, error) { + rows = append(rows, req.Rows...) + return &runnerv1.UpdateLogResponse{ + AckIndex: req.Index + int64(len(req.Rows)), + }, nil }, ) client.On("UpdateTask", mock.Anything, mock.Anything).Return( - func(_ context.Context, _ *connect_go.Request[runnerv1.UpdateTaskRequest]) (*connect_go.Response[runnerv1.UpdateTaskResponse], error) { - return connect_go.NewResponse(&runnerv1.UpdateTaskResponse{}), nil + func(_ context.Context, _ *runnerv1.UpdateTaskRequest) (*runnerv1.UpdateTaskResponse, error) { + return &runnerv1.UpdateTaskResponse{}, nil }, ) @@ -1184,16 +1183,16 @@ func TestReporter_StepRangeCoversAckedRows(t *testing.T) { client := mocks.NewClient(t) client.On("UpdateLog", mock.Anything, mock.Anything).Return( - func(_ context.Context, req *connect_go.Request[runnerv1.UpdateLogRequest]) (*connect_go.Response[runnerv1.UpdateLogResponse], error) { - return connect_go.NewResponse(&runnerv1.UpdateLogResponse{ - AckIndex: server.appendRows(len(req.Msg.Rows)), - }), nil + func(_ context.Context, req *runnerv1.UpdateLogRequest) (*runnerv1.UpdateLogResponse, error) { + return &runnerv1.UpdateLogResponse{ + AckIndex: server.appendRows(len(req.Rows)), + }, nil }, ) client.On("UpdateTask", mock.Anything, mock.Anything).Return( - func(_ context.Context, req *connect_go.Request[runnerv1.UpdateTaskRequest]) (*connect_go.Response[runnerv1.UpdateTaskResponse], error) { - server.applyState(req.Msg.State) - return connect_go.NewResponse(&runnerv1.UpdateTaskResponse{}), nil + func(_ context.Context, req *runnerv1.UpdateTaskRequest) (*runnerv1.UpdateTaskResponse, error) { + server.applyState(req.State) + return &runnerv1.UpdateTaskResponse{}, nil }, ) @@ -1250,8 +1249,8 @@ func TestReporter_FlushWithoutAckedRowsSkipsState(t *testing.T) { client := mocks.NewClient(t) client.On("UpdateLog", mock.Anything, mock.Anything).Once().Return( - func(_ context.Context, _ *connect_go.Request[runnerv1.UpdateLogRequest]) (*connect_go.Response[runnerv1.UpdateLogResponse], error) { - return connect_go.NewResponse(&runnerv1.UpdateLogResponse{AckIndex: 0}), nil + func(_ context.Context, _ *runnerv1.UpdateLogRequest) (*runnerv1.UpdateLogResponse, error) { + return &runnerv1.UpdateLogResponse{AckIndex: 0}, nil }, ) @@ -1277,10 +1276,10 @@ func newTestSteps(n int) []*model.Step { func TestReporter_ReportedSteps(t *testing.T) { newReporter := func(t *testing.T, steps []*model.Step) (*Reporter, context.CancelFunc) { client := mocks.NewClient(t) - client.On("UpdateLog", mock.Anything, mock.Anything).Maybe().Return(func(_ context.Context, req *connect_go.Request[runnerv1.UpdateLogRequest]) (*connect_go.Response[runnerv1.UpdateLogResponse], error) { - return connect_go.NewResponse(&runnerv1.UpdateLogResponse{AckIndex: req.Msg.Index + int64(len(req.Msg.Rows))}), nil + client.On("UpdateLog", mock.Anything, mock.Anything).Maybe().Return(func(_ context.Context, req *runnerv1.UpdateLogRequest) (*runnerv1.UpdateLogResponse, error) { + return &runnerv1.UpdateLogResponse{AckIndex: req.Index + int64(len(req.Rows))}, nil }) - client.On("UpdateTask", mock.Anything, mock.Anything).Maybe().Return(connect_go.NewResponse(&runnerv1.UpdateTaskResponse{}), nil) + client.On("UpdateTask", mock.Anything, mock.Anything).Maybe().Return(&runnerv1.UpdateTaskResponse{}, nil) taskCtx, err := structpb.NewStruct(map[string]any{}) require.NoError(t, err) cfg, _ := config.LoadDefault("")