enhance: add builtin:upload-artifact (#1343)

Co-authored-by: bircni <[email protected]>
This commit is contained in:
bircni authored and silverwind committed 2026-11-06 03:17:49 +00:00
1 parent 8c8950d9ef
commit 02446620df
30 files changed
+1881 -20

No files matched your search

+184
View File
@@ -0,0 +1,184 @@
// Copyright 2026 The Gitea Authors. All rights reserved.
// SPDX-License-Identifier: MIT
package artifact
import (
"bytes"
"context"
"crypto/rand"
"crypto/sha256"
"encoding/base64"
"encoding/hex"
"errors"
"fmt"
"io"
"net/http"
"net/url"
"strings"
"sync/atomic"
"time"
"golang.org/x/sync/errgroup"
)
// BlockSize is the toolkit's upload chunk.
const BlockSize = 8 << 20
type Uploaded struct {
Size int64
Digest string // sha256:<hex> for Finalize
}
// Upload puts body to a signed URL as parallel Azure blocks, failing when upload progress stalls for stall.
func (c *Client) Upload(ctx context.Context, signedURL, contentType string, body io.Reader, concurrency int, stall time.Duration) (result Uploaded, err error) {
ctx, cancel := context.WithCancelCause(ctx)
defer func() {
if ctx.Err() != nil {
err = context.Cause(ctx)
}
cancel(nil)
}()
stall = max(stall, time.Millisecond)
timer := time.AfterFunc(stall, func() { cancel(errors.New("upload progress stalled")) })
defer timer.Stop()
hash := sha256.New()
reader := io.TeeReader(body, hash)
group, blockCtx := errgroup.WithContext(ctx)
if closer, ok := body.(io.ReadCloser); ok {
stop := context.AfterFunc(blockCtx, func() { closer.Close() })
defer stop()
}
buffers := make(chan []byte, concurrency)
prefix := rand.Text()
var ids []string
var uploaded atomic.Int64
for index := 0; ; index++ {
var buf []byte
if index < concurrency {
buf = make([]byte, BlockSize)
} else {
select {
case buf = <-buffers:
case <-blockCtx.Done():
return Uploaded{}, group.Wait()
}
}
n, readErr := io.ReadFull(reader, buf)
if errors.Is(readErr, io.EOF) {
break
} else if readErr != nil && !errors.Is(readErr, io.ErrUnexpectedEOF) {
if blockCtx.Err() == nil {
cancel(readErr)
}
if err := group.Wait(); err != nil {
return Uploaded{}, err
}
return Uploaded{}, readErr
}
id := base64.StdEncoding.EncodeToString(fmt.Appendf(nil, "%s-%08d", prefix, index)) // fixed width as Azure needs, unique as Gitea before 1.25.5 shares a run's blocks
ids = append(ids, id)
group.Go(func() error {
defer func() { buffers <- buf }()
if err := c.put(blockCtx, signedURL+"&comp=block&blockid="+url.QueryEscape(id), "x-ms-blob-type", "BlockBlob", buf[:n]); err != nil {
return err
}
timer.Reset(stall)
c.Logf("Uploaded bytes %d", uploaded.Add(int64(n)))
return nil
})
if readErr != nil {
break
}
}
if err := group.Wait(); err != nil {
return Uploaded{}, err
}
var list strings.Builder
list.WriteString("<BlockList>")
for _, id := range ids {
list.WriteString("<Latest>" + id + "</Latest>")
}
list.WriteString("</BlockList>")
if err := c.put(ctx, signedURL+"&comp=blocklist", "x-ms-blob-content-type", contentType, []byte(list.String())); err != nil {
return Uploaded{}, err
}
return Uploaded{Size: uploaded.Load(), Digest: "sha256:" + hex.EncodeToString(hash.Sum(nil))}, nil
}
func (c *Client) put(ctx context.Context, signedURL, header, value string, body []byte) error {
resp, err := c.Retry.do(ctx, c.Logf, "blob upload", func() (*http.Response, error) {
req, err := http.NewRequestWithContext(ctx, http.MethodPut, signedURL, bytes.NewReader(body))
if err != nil {
return nil, err
}
req.Header.Set(header, value)
resp, err := c.blob.Do(req)
return resp, scrub(err)
})
if err != nil {
return err
}
return resp.Body.Close()
}
// Download GETs a signed URL, a read failing when nothing arrives for stall.
func (c *Client) Download(ctx context.Context, signedURL string, stall time.Duration) (*http.Response, error) {
ctx, cancel := context.WithCancelCause(ctx)
req, err := http.NewRequestWithContext(ctx, http.MethodGet, signedURL, nil)
if err != nil {
cancel(nil)
return nil, err
}
resp, err := c.blob.Do(req)
if err != nil {
cancel(nil)
return nil, scrub(err)
}
if resp.StatusCode != http.StatusOK {
resp.Body.Close()
cancel(nil)
return nil, fmt.Errorf("unexpected HTTP response from blob storage: %d %s", resp.StatusCode, http.StatusText(resp.StatusCode))
}
resp.Body = &stallReader{body: resp.Body, timer: time.AfterFunc(stall, func() {
cancel(fmt.Errorf("blob storage chunk did not respond in %s", stall))
}), stall: stall, cancel: cancel, ctx: ctx}
return resp, nil
}
func scrub(err error) error { // the signature of a signed URL is a credential
if urlErr, ok := errors.AsType[*url.Error](err); ok {
if parsed, parseErr := url.Parse(urlErr.URL); parseErr == nil {
parsed.RawQuery = ""
urlErr.URL = parsed.String()
} else {
urlErr.URL = "<unparsable URL>"
}
}
return err
}
type stallReader struct {
body io.ReadCloser
timer *time.Timer
stall time.Duration
cancel context.CancelCauseFunc
ctx context.Context
}
func (r *stallReader) Read(p []byte) (int, error) {
n, err := r.body.Read(p)
if n > 0 {
r.timer.Reset(r.stall)
}
if err != nil && !errors.Is(err, io.EOF) && context.Cause(r.ctx) != nil {
err = context.Cause(r.ctx)
}
return n, err
}
func (r *stallReader) Close() error {
r.timer.Stop()
r.cancel(nil)
return r.body.Close()
}
+264
View File
@@ -0,0 +1,264 @@
// Copyright 2026 The Gitea Authors. All rights reserved.
// SPDX-License-Identifier: MIT
// Package artifact talks to the v4 results service of the job's run.
package artifact
import (
"bytes"
"cmp"
"context"
"encoding/base64"
"encoding/json/v2"
"errors"
"fmt"
"io"
"math"
"net/http"
"net/url"
"strconv"
"strings"
"time"
)
type Artifact struct {
ID int64 `json:"databaseId,string"`
Name string `json:"name"`
Size int64 `json:"size,string"`
Digest string `json:"digest"`
}
type Retry struct {
Attempts int
Base time.Duration
Multiplier float64
MaxWait time.Duration
}
var DefaultRetry = Retry{Attempts: 5, Base: 8 * time.Second, Multiplier: 1.5, MaxWait: 2 * time.Minute}
type Client struct {
base *url.URL
token string
ids backendIDsMessage
api, blob *http.Client
Retry Retry
Logf func(format string, args ...any)
}
// New reaches base through transport, which carries the instance's TLS settings, and signed URLs elsewhere directly.
func New(base, token string, transport http.RoundTripper) (*Client, error) {
if base == "" {
return nil, errors.New("no artifact service is configured for this runner")
}
parsed, err := url.Parse(base)
if err != nil {
return nil, err
}
if token == "" {
return nil, errors.New("ACTIONS_RUNTIME_TOKEN is not set")
}
ids, err := backendIDs(token)
if err != nil {
return nil, err
}
return &Client{
base: parsed, token: token, ids: ids,
api: &http.Client{Transport: transport, Timeout: 3 * time.Minute},
blob: &http.Client{Transport: blobTransport{instance: parsed.Host, transport: transport}},
Retry: DefaultRetry,
Logf: func(string, ...any) {},
}, nil
}
// backendIDs reads the unverified scp claim of a runtime token, as the toolkit does.
func backendIDs(token string) (backendIDsMessage, error) {
parts := strings.Split(token, ".")
if len(parts) != 3 {
return backendIDsMessage{}, errors.New("the runtime token is not a JWT, artifacts v4 need Gitea 1.22 or later")
}
payload, err := base64.RawURLEncoding.DecodeString(strings.TrimRight(parts[1], "="))
if err != nil {
return backendIDsMessage{}, err
}
var claims struct {
Scp string `json:"scp"`
}
if err := json.Unmarshal(payload, &claims); err != nil {
return backendIDsMessage{}, err
}
for scope := range strings.FieldsSeq(claims.Scp) {
if fields := strings.Split(scope, ":"); fields[0] == "Actions.Results" && len(fields) == 3 {
return backendIDsMessage{fields[1], fields[2]}, nil
}
}
return backendIDsMessage{}, errors.New("the runtime token has no Actions.Results scope")
}
type blobTransport struct { // the instance's TLS settings only for the instance, as a signed URL can point at object storage
instance string
transport http.RoundTripper
}
func (t blobTransport) RoundTrip(req *http.Request) (*http.Response, error) {
if req.URL.Host == t.instance && t.transport != nil {
return t.transport.RoundTrip(req)
}
return http.DefaultTransport.RoundTrip(req)
}
type StatusError struct {
Status int
Msg string
}
func (e *StatusError) Error() string {
if e.Msg == "" {
return fmt.Sprintf("(%d) %s", e.Status, http.StatusText(e.Status))
}
return fmt.Sprintf("(%d) %s: %s", e.Status, http.StatusText(e.Status), e.Msg)
}
func IsNotFound(err error) bool {
var status *StatusError
return errors.As(err, &status) && status.Status == http.StatusNotFound
}
func (r Retry) do(ctx context.Context, logf func(string, ...any), what string, send func() (*http.Response, error)) (*http.Response, error) {
var waited time.Duration
for attempt := 0; ; attempt++ {
resp, err := send()
var wait time.Duration
if err == nil {
if resp.StatusCode >= 200 && resp.StatusCode < 400 {
return resp, nil
}
body, _ := io.ReadAll(io.LimitReader(resp.Body, 4096))
resp.Body.Close()
err = &StatusError{Status: resp.StatusCode, Msg: twirpMessage(body)}
switch resp.StatusCode {
case http.StatusTooManyRequests, http.StatusInternalServerError, http.StatusBadGateway, http.StatusServiceUnavailable, http.StatusGatewayTimeout:
default:
return nil, err
}
if seconds, parseErr := strconv.Atoi(resp.Header.Get("Retry-After")); parseErr == nil && seconds > 0 && resp.StatusCode == http.StatusTooManyRequests {
wait = time.Duration(seconds) * time.Second
}
}
if ctx.Err() != nil {
return nil, ctx.Err()
}
if attempt+1 >= r.Attempts {
return nil, fmt.Errorf("%s failed after %d attempts: %w", what, r.Attempts, err)
}
if wait == 0 {
wait = time.Duration(float64(r.Base) * math.Pow(r.Multiplier, float64(attempt)))
}
if waited+wait > r.MaxWait {
return nil, fmt.Errorf("%s: a retry wait of %s would exceed %s: %w", what, wait, r.MaxWait, err)
}
logf("Attempt %d of %d failed with error: %v. Retrying request in %d ms...", attempt+1, r.Attempts, err, wait.Milliseconds())
select {
case <-ctx.Done():
return nil, ctx.Err()
case <-time.After(wait):
}
waited += wait
}
}
// twirpMessage is the msg of a twirp error, the message of a Gitea API one, or the body Gitea answers in plain text.
func twirpMessage(body []byte) string {
var twirpErr struct {
Msg string `json:"msg"`
Message string `json:"message"`
}
if json.Unmarshal(body, &twirpErr) == nil && cmp.Or(twirpErr.Msg, twirpErr.Message) != "" {
return cmp.Or(twirpErr.Msg, twirpErr.Message)
}
return strings.TrimSpace(string(body))
}
type ( // the ArtifactService messages in the protobuf JSON mapping, requests with the proto names the toolkit sends
backendIDsMessage struct {
WorkflowRunBackendID string `json:"workflow_run_backend_id"`
WorkflowJobRunBackendID string `json:"workflow_job_run_backend_id"`
}
namedRequest struct {
backendIDsMessage
Name string `json:"name"`
}
createRequest struct {
namedRequest
ExpiresAt *time.Time `json:"expires_at,omitempty"`
Version int32 `json:"version"`
MimeType string `json:"mime_type,omitempty"` // Gitea before 1.26 rejects the field
}
finalizeRequest struct {
namedRequest
Size int64 `json:"size,string"`
Hash string `json:"hash"`
}
listRequest struct {
backendIDsMessage
NameFilter string `json:"name_filter,omitempty"`
}
idResponse struct {
ArtifactID int64 `json:"artifactId,string"`
}
)
func call[Resp any](ctx context.Context, c *Client, method string, req any) (Resp, error) {
var resp Resp
body, err := json.Marshal(req)
if err != nil {
return resp, err
}
endpoint := c.base.JoinPath("/twirp/github.actions.results.api.v1.ArtifactService", method).String()
httpResp, err := c.Retry.do(ctx, c.Logf, method, func() (*http.Response, error) {
httpReq, err := http.NewRequestWithContext(ctx, http.MethodPost, endpoint, bytes.NewReader(body))
if err != nil {
return nil, err
}
httpReq.Header.Set("Content-Type", "application/json")
httpReq.Header.Set("Authorization", "Bearer "+c.token)
return c.api.Do(httpReq)
})
if err != nil {
return resp, fmt.Errorf("failed to %s: %w", method, err)
}
defer httpResp.Body.Close()
err = json.UnmarshalRead(httpResp.Body, &resp)
return resp, err
}
func (c *Client) Create(ctx context.Context, name, mimeType string, expiresAt *time.Time) (string, error) {
resp, err := call[struct {
SignedUploadURL string `json:"signedUploadUrl"`
}](ctx, c, "CreateArtifact", createRequest{namedRequest{c.ids, name}, expiresAt, 7, mimeType})
return resp.SignedUploadURL, err
}
func (c *Client) Finalize(ctx context.Context, name string, size int64, digest string) (int64, error) {
resp, err := call[idResponse](ctx, c, "FinalizeArtifact", finalizeRequest{namedRequest{c.ids, name}, size, digest})
return resp.ArtifactID, err
}
func (c *Client) List(ctx context.Context, name string) ([]Artifact, error) {
resp, err := call[struct {
Artifacts []Artifact `json:"artifacts"`
}](ctx, c, "ListArtifacts", listRequest{c.ids, name})
return resp.Artifacts, err
}
func (c *Client) SignedURL(ctx context.Context, name string) (string, error) {
resp, err := call[struct {
SignedURL string `json:"signedUrl"`
}](ctx, c, "GetSignedArtifactURL", namedRequest{c.ids, name})
return resp.SignedURL, err
}
func (c *Client) Delete(ctx context.Context, name string) (int64, error) {
resp, err := call[idResponse](ctx, c, "DeleteArtifact", namedRequest{c.ids, name})
return resp.ArtifactID, err
}
+182
View File
@@ -0,0 +1,182 @@
// Copyright 2026 The Gitea Authors. All rights reserved.
// SPDX-License-Identifier: MIT
package artifact
import (
"bytes"
"context"
"crypto/sha256"
"encoding/hex"
"fmt"
"io"
"net/http"
"net/http/httptest"
"strings"
"sync/atomic"
"testing"
"time"
"gitea.com/gitea/runner/act/artifacts"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
func TestClientRoundTrip(t *testing.T) {
router := http.NewServeMux()
artifacts.RegisterV4(router, t.TempDir())
var blocks atomic.Int32
server := httptest.NewServer(http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) {
if request.URL.Query().Get("comp") == "block" {
blocks.Add(1)
}
router.ServeHTTP(writer, request)
}))
defer server.Close()
client, err := New(server.URL, artifacts.LocalRuntimeToken("7", time.Hour), nil)
require.NoError(t, err)
for _, content := range [][]byte{nil, bytes.Repeat([]byte("payload"), 2*BlockSize/7+123)} {
sum := sha256.Sum256(content)
uploadURL, err := client.Create(t.Context(), "art", "text/plain", nil)
require.NoError(t, err)
uploaded, err := client.Upload(t.Context(), uploadURL, "text/plain", bytes.NewReader(content), 2, time.Minute)
require.NoError(t, err)
assert.Equal(t, int32((len(content)+BlockSize-1)/BlockSize), blocks.Swap(0))
assert.Equal(t, Uploaded{Size: int64(len(content)), Digest: "sha256:" + hex.EncodeToString(sum[:])}, uploaded)
id, err := client.Finalize(t.Context(), "art", uploaded.Size, uploaded.Digest)
require.NoError(t, err)
listed, err := client.List(t.Context(), "art")
require.NoError(t, err)
assert.Equal(t, []Artifact{{ID: id, Name: "art", Size: int64(len(content)), Digest: uploaded.Digest}}, listed)
signed, err := client.SignedURL(t.Context(), "art")
require.NoError(t, err)
resp, err := client.Download(t.Context(), signed, time.Minute)
require.NoError(t, err)
got, err := io.ReadAll(resp.Body)
require.NoError(t, err)
require.NoError(t, resp.Body.Close())
assert.True(t, bytes.Equal(content, got))
deleted, err := client.Delete(t.Context(), "art")
require.NoError(t, err)
assert.Equal(t, id, deleted)
_, err = client.SignedURL(t.Context(), "art")
assert.True(t, IsNotFound(err), err)
}
}
func TestClientRetries(t *testing.T) {
var calls atomic.Int32
var blockIDs []string
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
if r.Method == http.MethodPut {
blockIDs = append(blockIDs, r.URL.Query().Get("blockid"))
w.WriteHeader(http.StatusCreated)
return
}
body, _ := io.ReadAll(r.Body)
assert.NotContains(t, string(body), "mime_type", "older Gitea rejects the field")
assert.True(t, strings.HasPrefix(r.Header.Get("Authorization"), "Bearer ey"), "the runtime token is the bearer")
switch calls.Add(1) {
case 1:
w.Header().Set("Retry-After", "0")
w.WriteHeader(http.StatusTooManyRequests)
case 2:
http.Error(w, "Error create or get artifact", http.StatusInternalServerError)
case 3:
w.WriteHeader(http.StatusFound)
_, _ = io.WriteString(w, `{"ok":true,"signedUploadUrl":"http://blob/u?sig=x"}`)
default:
w.WriteHeader(http.StatusBadRequest)
_, _ = io.WriteString(w, `{"code":"invalid_argument","msg":"bad name"}`)
}
}))
defer server.Close()
client, err := New(server.URL, artifacts.LocalRuntimeToken("7", time.Hour), nil)
require.NoError(t, err)
client.Retry = Retry{Attempts: 3}
var logged []string
client.Logf = func(format string, args ...any) { logged = append(logged, fmt.Sprintf(format, args...)) }
uploadURL, err := client.Create(t.Context(), "art", "", nil)
require.NoError(t, err)
assert.Equal(t, "http://blob/u?sig=x", uploadURL)
assert.Contains(t, logged[1], "Error create or get artifact")
_, err = client.Create(t.Context(), "art", "", nil)
require.ErrorContains(t, err, "(400) Bad Request: bad name")
assert.Equal(t, int32(4), calls.Load(), "a 400 is not retried")
for range 2 {
_, err = client.Upload(t.Context(), server.URL+"/u?sig=secret", "application/zip", strings.NewReader("x"), 1, time.Minute)
require.NoError(t, err)
}
assert.NotEqual(t, blockIDs[0], blockIDs[2], "each upload names its blocks anew")
server.Close()
_, err = client.Upload(t.Context(), server.URL+"/u?sig=secret", "application/zip", strings.NewReader("x"), 1, time.Minute)
require.Error(t, err)
assert.NotContains(t, err.Error()+strings.Join(logged, ""), "secret")
}
func TestTransferStalls(t *testing.T) {
for _, test := range []struct{ name, want string }{
{"api", "Client.Timeout"},
{"failed upload", "(400) Bad Request"},
{"source", "upload progress stalled"},
{"commit", "upload progress stalled"},
{"download", "did not respond"},
} {
t.Run(test.name, func(t *testing.T) {
server := httptest.NewServer(http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) {
_, _ = io.Copy(io.Discard, request.Body)
if test.name == "failed upload" {
writer.WriteHeader(http.StatusBadRequest)
return
}
if test.name == "download" {
_, _ = io.WriteString(writer, "partial")
writer.(http.Flusher).Flush()
}
<-request.Context().Done()
}))
defer server.Close()
client, err := New(server.URL, artifacts.LocalRuntimeToken("7", time.Hour), nil)
require.NoError(t, err)
client.Retry = Retry{Attempts: 1}
ctx, cancel := context.WithTimeout(t.Context(), time.Second)
defer cancel()
switch test.name {
case "api":
assert.Equal(t, 3*time.Minute, client.api.Timeout)
client.api.Timeout = time.Millisecond
_, err = client.Create(ctx, "art", "", nil)
case "download":
var resp *http.Response
resp, err = client.Download(ctx, server.URL, time.Millisecond)
require.NoError(t, err)
defer resp.Body.Close()
_, err = io.ReadAll(resp.Body)
case "commit":
_, err = client.Upload(ctx, server.URL+"?sig=x", "text/plain", strings.NewReader(""), 1, time.Millisecond)
default:
reader, writer := io.Pipe()
defer reader.Close()
defer writer.Close()
done := make(chan error, 1)
stall := time.Millisecond
if test.name == "failed upload" {
stall = time.Minute
go func() {
_, err := writer.Write(make([]byte, BlockSize))
done <- err
}()
}
_, err = client.Upload(ctx, server.URL+"?sig=x", "text/plain", reader, 2, stall)
if test.name == "failed upload" {
require.NoError(t, <-done)
}
}
require.ErrorContains(t, err, test.want)
})
}
}