GHA cache v2 Twirp protocol support (#829)

* GHA cache v2 Twirp protocol support

* CI: install protoc-gen-twirp

Had to use golang:latest because bufbuild/buf container image
lacks Golang.

* buf.gen.yaml: explicitly use local generation

.cirrus.yml

@@ -170,8 +170,10 @@ task:
   name: Check for lacking "buf generate" invocation
 
   container:
-    image: bufbuild/buf
+    image: golang:latest
 
+  install_buf_script: go install github.com/bufbuild/buf/cmd/buf@v1.50.0
+  install_protoc_gen_twirp_script: go install github.com/twitchtv/twirp/protoc-gen-twirp@latest
   generate_script: buf generate
   check_script: git diff --exit-code
 

api/gharesults/gharesults.proto

@@ -0,0 +1,57 @@
+syntax = "proto3";
+
+package github.actions.results.api.v1;
+
+option go_package = "github.com/cirruslabs/cirrus-cli/pkg/api/gharesults";
+
+service CacheService {
+  rpc CreateCacheEntry (CreateCacheEntryRequest) returns (CreateCacheEntryResponse);
+  rpc FinalizeCacheEntryUpload (FinalizeCacheEntryUploadRequest) returns (FinalizeCacheEntryUploadResponse);
+  rpc GetCacheEntryDownloadURL (GetCacheEntryDownloadURLRequest) returns (GetCacheEntryDownloadURLResponse);
+}
+
+message CreateCacheEntryRequest {
+  CacheMetadata metadata = 1;
+  string key = 2;
+  string version = 3;
+}
+
+message CacheMetadata {
+  string repository_id = 1;
+  repeated CacheScope scope = 2;
+}
+
+message CacheScope {
+  string scope = 1;
+  string permission = 2;
+}
+
+message CreateCacheEntryResponse {
+  bool ok = 1;
+  string signed_upload_url = 2;
+}
+
+message FinalizeCacheEntryUploadRequest {
+  CacheMetadata metadata = 1;
+  string key = 2;
+  int64 size_bytes = 3;
+  string version = 4;
+}
+
+message FinalizeCacheEntryUploadResponse {
+  bool ok = 1;
+  int64 entry_id = 2;
+}
+
+message GetCacheEntryDownloadURLRequest {
+  CacheMetadata metadata = 1;
+  string key = 2;
+  repeated string restore_keys = 3;
+  string version = 4;
+}
+
+message GetCacheEntryDownloadURLResponse {
+  bool ok = 1;
+  string signed_download_url = 2;
+  string matched_key = 3;
+}

buf.gen.yaml

@@ -1,8 +1,15 @@
-version: v1
+version: v2
 plugins:
-  - plugin: buf.build/protocolbuffers/go
+  - remote: buf.build/protocolbuffers/go
     out: pkg/
     opt: paths=source_relative
-  - plugin: buf.build/grpc/go
+  - remote: buf.build/grpc/go
+    out: pkg/
+    opt: paths=source_relative
+  # Use local generation, as recommended in https://github.com/bufbuild/plugins/issues/58:
+  #
+  # >We're not actively working on supporting the twitchtv/twirp plugin on the BSR at the moment.
+  # >For now we recommend that users use local generation with buf.gen.yaml.
+  - local: protoc-gen-twirp
     out: pkg/
     opt: paths=source_relative

go.mod

@@ -89,6 +89,7 @@ require (
 	github.com/shirou/gopsutil/v3 v3.24.5
 	github.com/testcontainers/testcontainers-go v0.33.0
 	github.com/tonistiigi/go-actions-cache v0.0.0-20240327122527-58651d5e11d6
+	github.com/twitchtv/twirp v8.1.3+incompatible
 	go.opentelemetry.io/contrib/bridges/otelzap v0.8.0
 	go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.49.0
 	go.opentelemetry.io/otel v1.33.0

go.sum

@@ -682,6 +682,8 @@ github.com/tonistiigi/fsutil v0.0.0-20240301111122-7525a1af2bb5 h1:oZS8KCqAg62sx
 github.com/tonistiigi/fsutil v0.0.0-20240301111122-7525a1af2bb5/go.mod h1:vbbYqJlnswsbJqWUcJN8fKtBhnEgldDrcagTgnBVKKM=
 github.com/tonistiigi/go-actions-cache v0.0.0-20240327122527-58651d5e11d6 h1:XFG/Wmm5dFYoqUiVChLumRjRzJm0P9k/qDMhxLqdupU=
 github.com/tonistiigi/go-actions-cache v0.0.0-20240327122527-58651d5e11d6/go.mod h1:anhKd3mnC1shAbQj1Q4IJ+w6xqezxnyDYlx/yKa7IXM=
+github.com/twitchtv/twirp v8.1.3+incompatible h1:+F4TdErPgSUbMZMwp13Q/KgDVuI7HJXP61mNV3/7iuU=
+github.com/twitchtv/twirp v8.1.3+incompatible/go.mod h1:RRJoFSAmTEh2weEqWtpPE3vFK5YBhA6bqp2l1kfCC5A=
 github.com/xanzy/ssh-agent v0.3.3 h1:+/15pJfg/RsTxqYcX6fHqOXZwwMP+2VyYWJeWM2qQFM=
 github.com/xanzy/ssh-agent v0.3.3/go.mod h1:6dzNDKs0J9rVPHPhaGCukekBHKqfl+L3KghI1Bc68Uw=
 github.com/xeipuuv/gojsonpointer v0.0.0-20180127040702-4e3ac2762d5f/go.mod h1:N2zxlSyiKSe5eX1tZViRH5QA0qijqEDrYZiPEAiq3wU=

internal/agent/http_cache/ghacache/cirruscimock/cirruscimock.go

@@ -133,16 +133,32 @@ func (mock *cirrusCIMock) DownloadCache(request *api.DownloadCacheRequest, strea
 	return nil
 }
 
+func (mock *cirrusCIMock) GenerateCacheUploadURL(ctx context.Context, request *api.CacheKey) (*api.GenerateURLResponse, error) {
+	putObjectRequest, _ := mock.s3Client.PutObjectRequest(&s3.PutObjectInput{
+		Bucket: mock.s3Bucket,
+		Key:    aws.String(request.CacheKey),
+	})
+
+	url, _, err := putObjectRequest.PresignRequest(10 * time.Minute)
+	if err != nil {
+		return nil, err
+	}
+
+	return &api.GenerateURLResponse{
+		Url: url,
+	}, nil
+}
+
 func (mock *cirrusCIMock) GenerateCacheDownloadURLs(
 	_ context.Context,
 	request *api.CacheKey,
 ) (*api.GenerateURLsResponse, error) {
-	uploadPartRequest, _ := mock.s3Client.GetObjectRequest(&s3.GetObjectInput{
+	getObjectRequest, _ := mock.s3Client.GetObjectRequest(&s3.GetObjectInput{
 		Bucket: mock.s3Bucket,
 		Key:    aws.String(request.CacheKey),
 	})
 
-	url, _, err := uploadPartRequest.PresignRequest(10 * time.Minute)
+	url, _, err := getObjectRequest.PresignRequest(10 * time.Minute)
 	if err != nil {
 		return nil, err
 	}

internal/agent/http_cache/ghacachev2/ghacachev2.go

@@ -0,0 +1,110 @@
+package ghacachev2
+
+import (
+	"context"
+	"fmt"
+	"github.com/cirruslabs/cirrus-cli/internal/agent/client"
+	"github.com/cirruslabs/cirrus-cli/pkg/api"
+	"github.com/cirruslabs/cirrus-cli/pkg/api/gharesults"
+	"github.com/samber/lo"
+	"github.com/twitchtv/twirp"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
+	"hash/fnv"
+	"net/http"
+	"net/url"
+	"strings"
+)
+
+// Interface guard
+//
+// Ensures that Cache struct implements gharesults.CacheService interface.
+var _ gharesults.CacheService = (*Cache)(nil)
+
+const APIMountPoint = "/twirp"
+
+type Cache struct {
+	cacheHost   string
+	twirpServer gharesults.TwirpServer
+}
+
+func New(cacheHost string) *Cache {
+	cache := &Cache{
+		cacheHost: cacheHost,
+	}
+
+	cache.twirpServer = gharesults.NewCacheServiceServer(cache)
+
+	return cache
+}
+
+func (cache *Cache) PathPrefix() string {
+	return cache.twirpServer.PathPrefix()
+}
+
+func (cache *Cache) ServeHTTP(writer http.ResponseWriter, request *http.Request) {
+	cache.twirpServer.ServeHTTP(writer, request)
+}
+
+func (cache *Cache) GetCacheEntryDownloadURL(ctx context.Context, request *gharesults.GetCacheEntryDownloadURLRequest) (*gharesults.GetCacheEntryDownloadURLResponse, error) {
+	grpcRequest := &api.CacheInfoRequest{
+		TaskIdentification: client.CirrusTaskIdentification,
+		CacheKey:           httpCacheKey(request.Key, request.Version),
+		CacheKeyPrefixes: lo.Map(request.RestoreKeys, func(restoreKey string, _ int) string {
+			return httpCacheKey(restoreKey, request.Version)
+		}),
+	}
+
+	grpcResponse, err := client.CirrusClient.CacheInfo(ctx, grpcRequest)
+	if err != nil {
+		if status, ok := status.FromError(err); ok && status.Code() == codes.NotFound {
+			return nil, twirp.NewErrorf(twirp.NotFound, "cache entry not found")
+		}
+
+		return nil, twirp.NewErrorf(twirp.Internal, "GHA cache v2 failed to retrieve information "+
+			"about cache entry with key %q and version %q: %v", request.Key, request.Version, err)
+	}
+
+	return &gharesults.GetCacheEntryDownloadURLResponse{
+		Ok:                true,
+		SignedDownloadUrl: cache.httpCacheURL(grpcResponse.Info.Key),
+		MatchedKey:        strings.TrimPrefix(grpcResponse.Info.Key, httpCacheKey("", request.Version)),
+	}, nil
+}
+
+func (cache *Cache) CreateCacheEntry(ctx context.Context, request *gharesults.CreateCacheEntryRequest) (*gharesults.CreateCacheEntryResponse, error) {
+	grpcResponse, err := client.CirrusClient.GenerateCacheUploadURL(ctx, &api.CacheKey{
+		TaskIdentification: client.CirrusTaskIdentification,
+		CacheKey:           httpCacheKey(request.Key, request.Version),
+	})
+	if err != nil {
+		return nil, twirp.NewErrorf(twirp.Internal, "GHA cache v2 failed to create cache entry "+
+			"with key %q and version %q: %v", request.Key, request.Version, err)
+	}
+
+	return &gharesults.CreateCacheEntryResponse{
+		Ok:              true,
+		SignedUploadUrl: grpcResponse.Url,
+	}, nil
+}
+
+func (cache *Cache) FinalizeCacheEntryUpload(ctx context.Context, request *gharesults.FinalizeCacheEntryUploadRequest) (*gharesults.FinalizeCacheEntryUploadResponse, error) {
+	hash := fnv.New64a()
+
+	_, _ = hash.Write([]byte(request.Key))
+	_, _ = hash.Write([]byte(fmt.Sprintf("%d", request.SizeBytes)))
+	_, _ = hash.Write([]byte(request.Version))
+
+	return &gharesults.FinalizeCacheEntryUploadResponse{
+		Ok:      true,
+		EntryId: int64(hash.Sum64()),
+	}, nil
+}
+
+func httpCacheKey(key string, version string) string {
+	return fmt.Sprintf("%s-%s", url.PathEscape(version), url.PathEscape(key))
+}
+
+func (cache *Cache) httpCacheURL(keyWithVersion string) string {
+	return fmt.Sprintf("http://%s/%s", cache.cacheHost, url.PathEscape(keyWithVersion))
+}

internal/agent/http_cache/ghacachev2/ghacachev2_test.go

@@ -0,0 +1,71 @@
+package ghacachev2_test
+
+import (
+	"bytes"
+	"context"
+	"github.com/cirruslabs/cirrus-cli/internal/agent/client"
+	"github.com/cirruslabs/cirrus-cli/internal/agent/http_cache"
+	"github.com/cirruslabs/cirrus-cli/internal/agent/http_cache/ghacache/cirruscimock"
+	"github.com/cirruslabs/cirrus-cli/internal/testutil"
+	"github.com/cirruslabs/cirrus-cli/pkg/api/gharesults"
+	"github.com/google/uuid"
+	"github.com/stretchr/testify/require"
+	"github.com/twitchtv/twirp"
+	"io"
+	"net/http"
+	"testing"
+)
+
+func TestGHACacheV2(t *testing.T) {
+	testutil.NeedsContainerization(t)
+
+	ctx := context.Background()
+
+	client.InitClient(cirruscimock.ClientConn(t), "test", "test")
+
+	httpCacheURL := "http://" + http_cache.Start()
+
+	client := gharesults.NewCacheServiceJSONClient(httpCacheURL, &http.Client{})
+
+	cacheKey := uuid.NewString()
+	cacheValue := []byte("Hello, World!\n")
+
+	// Ensure that an entry for our cache key is not present
+	_, err := client.GetCacheEntryDownloadURL(ctx, &gharesults.GetCacheEntryDownloadURLRequest{
+		Key: cacheKey,
+	})
+	var twirpError twirp.Error
+	require.ErrorAs(t, err, &twirpError)
+	require.Equal(t, twirp.NotFound, twirpError.Code())
+
+	// Upload an entry for our cache key
+	createCacheEntryRes, err := client.CreateCacheEntry(ctx, &gharesults.CreateCacheEntryRequest{
+		Key: cacheKey,
+	})
+	require.NoError(t, err)
+	require.True(t, createCacheEntryRes.Ok)
+
+	uploadReq, err := http.NewRequest(http.MethodPut, createCacheEntryRes.SignedUploadUrl, bytes.NewReader(cacheValue))
+	require.NoError(t, err)
+
+	uploadResp, err := http.DefaultClient.Do(uploadReq)
+	require.NoError(t, err)
+	require.Equal(t, http.StatusOK, uploadResp.StatusCode)
+
+	// Ensure that an entry for our cache key is present
+	// and matches to what we've previously put in the cache
+	getCacheEntryDownloadURLResp, err := client.GetCacheEntryDownloadURL(ctx, &gharesults.GetCacheEntryDownloadURLRequest{
+		Key: cacheKey,
+	})
+	require.NoError(t, err)
+	require.True(t, getCacheEntryDownloadURLResp.Ok)
+	require.Equal(t, cacheKey, getCacheEntryDownloadURLResp.MatchedKey)
+
+	downloadResp, err := http.Get(getCacheEntryDownloadURLResp.SignedDownloadUrl)
+	require.NoError(t, err)
+	require.Equal(t, http.StatusOK, downloadResp.StatusCode)
+
+	downloadRespBodyBytes, err := io.ReadAll(downloadResp.Body)
+	require.NoError(t, err)
+	require.Equal(t, cacheValue, downloadRespBodyBytes)
+}

internal/agent/http_cache/http_cache.go

@@ -7,6 +7,7 @@ import (
 	"fmt"
 	"github.com/cirruslabs/cirrus-cli/internal/agent/client"
 	"github.com/cirruslabs/cirrus-cli/internal/agent/http_cache/ghacache"
+	"github.com/cirruslabs/cirrus-cli/internal/agent/http_cache/ghacachev2"
 	"github.com/cirruslabs/cirrus-cli/pkg/api"
 	sentryhttp "github.com/getsentry/sentry-go/http"
 	"golang.org/x/sync/semaphore"
@@ -63,6 +64,13 @@ func Start(opts ...Option) string {
 		mux.Handle(ghacache.APIMountPoint+"/", sentryHandler.Handle(http.StripPrefix(ghacache.APIMountPoint,
 			ghacache.New(address))))
 
+		// GitHub Actions cache API v2
+		//
+		// Note that we don't strip the prefix here because
+		// Twirp handler inside *ghacachev2.Cache expects it.
+		ghaCacheV2 := ghacachev2.New(address)
+		mux.Handle(ghaCacheV2.PathPrefix(), ghaCacheV2)
+
 		// Apply options
 		for _, opt := range opts {
 			opt(mux)