runner/act/plugin/client.go

// Copyright 2026 The Forgejo Authors. All rights reserved.
// SPDX-License-Identifier: GPL-3.0-or-later

package plugin

import (
	"context"
	"crypto/tls"
	"fmt"
	"net"
	"strings"
	"time"

	"code.forgejo.org/forgejo/runner/v13/act/container"
	pluginv1alpha "code.forgejo.org/forgejo/runner/v13/act/plugin/proto/v1alpha"
	"google.golang.org/grpc"
	"google.golang.org/grpc/credentials"
	"google.golang.org/grpc/credentials/insecure"
	"google.golang.org/grpc/health/grpc_health_v1"
	"google.golang.org/grpc/keepalive"
)

// grpc.NewClient is non-blocking; defaultDialTimeout is applied to the
// first RPC (health check) when the caller's context has no deadline.
const defaultDialTimeout = 10 * time.Second

// 32 MiB lifts the stock 4 MiB gRPC cap so plugins can emit larger Exec
// frames and CopyOut chunks without ResourceExhausted.
const defaultMaxMessageSize = 32 * 1024 * 1024

const (
	defaultKeepaliveTime    = 30 * time.Second
	defaultKeepaliveTimeout = 10 * time.Second
)

type ClientOption func(*clientOptions)

type clientOptions struct {
	tlsConfig        *tls.Config
	allowPlainTCP    bool
	dialTimeout      time.Duration
	maxMessageSize   int
	keepaliveTime    time.Duration
	keepaliveTimeout time.Duration
}

func WithTLS(cfg *tls.Config) ClientOption {
	return func(o *clientOptions) { o.tlsConfig = cfg }
}

// WithAllowPlainTCP allows TCP without TLS, for trusted localhost or tests.
func WithAllowPlainTCP() ClientOption {
	return func(o *clientOptions) { o.allowPlainTCP = true }
}

func WithDialTimeout(d time.Duration) ClientOption {
	return func(o *clientOptions) { o.dialTimeout = d }
}

// WithMaxMessageSize sets the per-call send/recv size limit in bytes.
func WithMaxMessageSize(bytes int) ClientOption {
	return func(o *clientOptions) { o.maxMessageSize = bytes }
}

// WithKeepalive sets the gRPC keepalive ping interval and ack timeout.
func WithKeepalive(interval, timeout time.Duration) ClientOption {
	return func(o *clientOptions) {
		o.keepaliveTime = interval
		o.keepaliveTimeout = timeout
	}
}

type Client struct {
	conn *grpc.ClientConn
	rpc  pluginv1alpha.BackendPluginClient
	// caps is captured once; restart the runner if a plugin changes capabilities.
	caps *pluginv1alpha.CapabilitiesResponse
}

func resolveOptions(options []ClientOption) *clientOptions {
	cfg := &clientOptions{
		dialTimeout:      defaultDialTimeout,
		maxMessageSize:   defaultMaxMessageSize,
		keepaliveTime:    defaultKeepaliveTime,
		keepaliveTimeout: defaultKeepaliveTimeout,
	}
	for _, opt := range options {
		opt(cfg)
	}
	return cfg
}

// withDialTimeout applies cfg.dialTimeout when the caller's context has none, so
// a non-blocking grpc.NewClient still fails fast on the first RPC.
func withDialTimeout(ctx context.Context, cfg *clientOptions) (context.Context, context.CancelFunc) {
	if _, ok := ctx.Deadline(); ok {
		return ctx, func() {}
	}
	return context.WithTimeout(ctx, cfg.dialTimeout)
}

// dial opens a connection to a plugin and confirms its health service is
// SERVING before the caller issues RPCs over it.
func dial(ctx context.Context, address string, cfg *clientOptions) (*grpc.ClientConn, error) {
	isUnix := strings.HasPrefix(address, "unix://")

	creds, err := transportCredentials(isUnix, cfg)
	if err != nil {
		return nil, err
	}

	opts := []grpc.DialOption{
		grpc.WithTransportCredentials(creds),
		grpc.WithDefaultCallOptions(
			grpc.MaxCallRecvMsgSize(cfg.maxMessageSize),
			grpc.MaxCallSendMsgSize(cfg.maxMessageSize),
		),
		grpc.WithKeepaliveParams(keepalive.ClientParameters{
			Time:                cfg.keepaliveTime,
			Timeout:             cfg.keepaliveTimeout,
			PermitWithoutStream: false,
		}),
	}
	if isUnix {
		socketPath := strings.TrimPrefix(address, "unix://")
		opts = append(opts, grpc.WithContextDialer(func(ctx context.Context, _ string) (net.Conn, error) {
			return (&net.Dialer{}).DialContext(ctx, "unix", socketPath)
		}))
		address = "passthrough:///" + socketPath
	}

	conn, err := grpc.NewClient(address, opts...)
	if err != nil {
		return nil, fmt.Errorf("dial plugin at %s: %w", address, err)
	}

	healthClient := grpc_health_v1.NewHealthClient(conn)
	healthResp, err := healthClient.Check(ctx, &grpc_health_v1.HealthCheckRequest{})
	if err != nil {
		conn.Close()
		return nil, fmt.Errorf("health check plugin at %s: %w", address, err)
	}
	if got := healthResp.GetStatus(); got != grpc_health_v1.HealthCheckResponse_SERVING {
		conn.Close()
		return nil, fmt.Errorf("health check plugin at %s: status %s", address, got)
	}

	return conn, nil
}

// NewClient connects to a BackendPlugin: a plugin that owns the whole execution
// environment and every container operation within it.
func NewClient(ctx context.Context, address string, options ...ClientOption) (*Client, error) {
	cfg := resolveOptions(options)
	ctx, cancel := withDialTimeout(ctx, cfg)
	defer cancel()

	conn, err := dial(ctx, address, cfg)
	if err != nil {
		return nil, err
	}

	rpc := pluginv1alpha.NewBackendPluginClient(conn)
	caps, err := rpc.Capabilities(ctx, &pluginv1alpha.CapabilitiesRequest{})
	if err != nil {
		conn.Close()
		return nil, fmt.Errorf("get capabilities from plugin at %s: %w", address, err)
	}
	if err := validateCapabilities(caps); err != nil {
		conn.Close()
		return nil, fmt.Errorf("plugin at %s: %w", address, err)
	}

	return &Client{
		conn: conn,
		rpc:  rpc,
		caps: caps,
	}, nil
}

func transportCredentials(isUnix bool, cfg *clientOptions) (credentials.TransportCredentials, error) {
	switch {
	case cfg.tlsConfig != nil:
		return credentials.NewTLS(cfg.tlsConfig), nil
	case isUnix, cfg.allowPlainTCP:
		return insecure.NewCredentials(), nil
	default:
		return nil, fmt.Errorf("plugin: TCP address requires WithTLS or WithAllowPlainTCP")
	}
}

func validateCapabilities(caps *pluginv1alpha.CapabilitiesResponse) error {
	missing := []string{}
	if caps.GetName() == "" {
		missing = append(missing, "name")
	}
	if len(missing) > 0 {
		return fmt.Errorf("capabilities response missing required fields: %s", strings.Join(missing, ", "))
	}
	return nil
}

func (c *Client) Capabilities() *pluginv1alpha.CapabilitiesResponse {
	return c.caps
}

func (c *Client) NewEnvironment(input *container.NewContainerInput, backendOpts map[string]string, labelArg string, timeout time.Duration) container.ExecutionsEnvironment {
	return &pluginEnvironment{
		client:      c.rpc,
		caps:        c.caps,
		backendOpts: backendOpts,
		labelArg:    labelArg,
		timeout:     timeout,
		input:       input,
		stdout:      input.Stdout,
		stderr:      input.Stderr,
	}
}

func (c *Client) Close() error {
	return c.conn.Close()
}