eric/distributor
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 7 Wiki Activity

Stage uploads before accepting HTTP runs

Browse Source
This commit is contained in:
Eric Rakestraw
2026-06-04 00:14:32 +00:00
parent 1340418a2b
commit 0d346dcdf5
12 changed files with 248 additions and 162 deletions
Download Patch File Download Diff File Expand all files Collapse all files

112
internal/app/upload_coordinator.go
View File

@@ -78,6 +78,7 @@ type UploadCoordinator struct {
queueSize int
maxConcurrency int
runningCount int
reservedCount int
activePipeline map[string]bool
pending []*uploadJob
records map[UploadRunID]UploadRunRecord
@@ -88,9 +89,10 @@ type uploadStageFunc func(context.Context, ingest.StageOptions) (ingest.StagedBu
type uploadRunFunc func(context.Context, config.Config, RunPipelineWithLocalSourceOptions) (RunReport, error)
type uploadJob struct {
recordID UploadRunID
request UploadRequest
pipeline config.Pipeline
recordID UploadRunID
request UploadRequest
pipeline config.Pipeline
stagedRoot string
}
type uploadCoordinatorHooks struct {
@@ -164,24 +166,49 @@ func (coordinator *UploadCoordinator) Submit(ctx context.Context, request Upload
if err != nil {
return UploadRunRecord{}, err
}
if err := ingest.ValidateContentType(request.ContentType); err != nil {
return UploadRunRecord{}, err
}
coordinator.mu.Lock()
coordinator.expireLocked(coordinator.now().UTC())
if coordinator.queueFullLocked() {
coordinator.mu.Unlock()
return UploadRunRecord{}, UploadQueueFullError{QueueSize: coordinator.queueSize}
}
coordinator.reservedCount++
coordinator.mu.Unlock()
staged, err := coordinator.stage(ctx, ingest.StageOptions{
Body: request.Body,
ContentType: request.ContentType,
PipelineStagingPath: pipeline.Source.Upload.StagingPath,
RunID: string(runID),
MaxUploadSize: int64(*pipeline.Source.Upload.MaxUploadSize),
MaxExtractedSize: int64(*pipeline.Source.Upload.MaxUploadSize),
MaxFileCount: uploadMaxFileCount(request.MaxFileCount),
})
if err != nil {
coordinator.releaseReservation()
return UploadRunRecord{}, err
}
coordinator.mu.Lock()
defer coordinator.mu.Unlock()
coordinator.expireLocked(coordinator.now().UTC())
if len(coordinator.pending) >= coordinator.queueSize {
return UploadRunRecord{}, UploadQueueFullError{QueueSize: coordinator.queueSize}
}
coordinator.reservedCount--
record := UploadRunRecord{
ID: runID,
PipelineID: pipeline.ID,
Status: UploadStatusAccepted,
AcceptedAt: coordinator.now().UTC(),
StagedRoot: staged.Root,
}
coordinator.records[runID] = record
coordinator.pending = append(coordinator.pending, &uploadJob{
recordID: runID,
request: request,
pipeline: pipeline,
recordID: runID,
request: request,
pipeline: pipeline,
stagedRoot: staged.Root,
})
coordinator.notify()
return record, nil
@@ -205,13 +232,13 @@ func (coordinator *UploadCoordinator) CanAccept() bool {
coordinator.mu.Lock()
defer coordinator.mu.Unlock()
coordinator.expireLocked(coordinator.now().UTC())
return len(coordinator.pending) < coordinator.queueSize
return !coordinator.queueFullLocked()
}
func (coordinator *UploadCoordinator) QueueDepth() int {
coordinator.mu.Lock()
defer coordinator.mu.Unlock()
return len(coordinator.pending)
return len(coordinator.pending) + coordinator.reservedCount
}
func (coordinator *UploadCoordinator) RunningCount() int {
@@ -290,47 +317,30 @@ func (coordinator *UploadCoordinator) markPendingQueuedLocked() {
}
func (coordinator *UploadCoordinator) runJob(job *uploadJob) {
record := coordinator.currentRecord(job.recordID)
maxFileCount := job.request.MaxFileCount
if maxFileCount <= 0 {
maxFileCount = DefaultUploadMaxFileCount
}
staged, err := coordinator.stage(coordinator.ctx, ingest.StageOptions{
Body: job.request.Body,
ContentType: job.request.ContentType,
PipelineStagingPath: job.pipeline.Source.Upload.StagingPath,
RunID: string(record.ID),
MaxUploadSize: int64(*job.pipeline.Source.Upload.MaxUploadSize),
MaxExtractedSize: int64(*job.pipeline.Source.Upload.MaxUploadSize),
MaxFileCount: maxFileCount,
report, err := coordinator.run(coordinator.ctx, coordinator.cfg, RunPipelineWithLocalSourceOptions{
PipelineID: job.pipeline.ID,
SourceRoot: job.stagedRoot,
DryRun: job.request.DryRun,
Force: job.request.Force,
})
if err == nil {
coordinator.setStagedRoot(job.recordID, staged.Root)
var report RunReport
report, err = coordinator.run(coordinator.ctx, coordinator.cfg, RunPipelineWithLocalSourceOptions{
PipelineID: job.pipeline.ID,
SourceRoot: staged.Root,
DryRun: job.request.DryRun,
Force: job.request.Force,
})
coordinator.complete(job, &report, err)
return
coordinator.complete(job, &report, err)
}
func (coordinator *UploadCoordinator) releaseReservation() {
coordinator.mu.Lock()
defer coordinator.mu.Unlock()
coordinator.reservedCount--
}
func (coordinator *UploadCoordinator) queueFullLocked() bool {
return len(coordinator.pending)+coordinator.reservedCount >= coordinator.queueSize
}
func uploadMaxFileCount(value int) int {
if value > 0 {
return value
}
coordinator.complete(job, nil, err)
}
func (coordinator *UploadCoordinator) currentRecord(runID UploadRunID) UploadRunRecord {
coordinator.mu.Lock()
defer coordinator.mu.Unlock()
return coordinator.records[runID]
}
func (coordinator *UploadCoordinator) setStagedRoot(runID UploadRunID, root string) {
coordinator.mu.Lock()
defer coordinator.mu.Unlock()
record := coordinator.records[runID]
record.StagedRoot = root
coordinator.records[runID] = record
return DefaultUploadMaxFileCount
}
func (coordinator *UploadCoordinator) complete(job *uploadJob, report *RunReport, runErr error) {

57
internal/app/upload_http.go
View File

@@ -1,13 +1,10 @@
package app
import (
"bytes"
"context"
"encoding/json"
"errors"
"fmt"
"io"
"mime"
"net/http"
"strings"
@@ -24,7 +21,6 @@ type uploadCoordinator interface {
type uploadHTTPHandler struct {
coordinator uploadCoordinator
tokens map[string]string
limits map[string]int64
}
type uploadAcceptedResponse struct {
@@ -38,20 +34,18 @@ type httpErrorResponse struct {
func newUploadHTTPHandler(ctx context.Context, cfg config.Config, environment config.Environment) (http.Handler, error) {
config.ApplyDefaults(&cfg)
tokens, limits, err := resolveUploadTokens(cfg, environment)
tokens, err := resolveUploadTokens(cfg, environment)
if err != nil {
return nil, err
}
return uploadHTTPHandler{
coordinator: NewUploadCoordinator(ctx, cfg),
tokens: tokens,
limits: limits,
}, nil
}
func resolveUploadTokens(cfg config.Config, environment config.Environment) (map[string]string, map[string]int64, error) {
func resolveUploadTokens(cfg config.Config, environment config.Environment) (map[string]string, error) {
tokens := make(map[string]string)
limits := make(map[string]int64)
for _, pipeline := range cfg.Pipelines {
if pipeline.Source.Backend != config.BackendHTTPUpload {
continue
@@ -59,18 +53,17 @@ func resolveUploadTokens(cfg config.Config, environment config.Environment) (map
tokenName := pipeline.Source.Upload.TokenEnv
token, ok := environment.Lookup(tokenName)
if !ok {
return nil, nil, fmt.Errorf("upload token environment variable %s is not set", tokenName)
return nil, fmt.Errorf("upload token environment variable %s is not set", tokenName)
}
if token == "" {
return nil, nil, fmt.Errorf("upload token environment variable %s is empty", tokenName)
return nil, fmt.Errorf("upload token environment variable %s is empty", tokenName)
}
if existing, exists := tokens[token]; exists {
return nil, nil, fmt.Errorf("upload token environment variables for pipelines %s and %s resolve to the same value", existing, pipeline.ID)
return nil, fmt.Errorf("upload token environment variables for pipelines %s and %s resolve to the same value", existing, pipeline.ID)
}
tokens[token] = pipeline.ID
limits[pipeline.ID] = int64(*pipeline.Source.Upload.MaxUploadSize)
}
return tokens, limits, nil
return tokens, nil
}
func (handler uploadHTTPHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
@@ -101,7 +94,7 @@ func (handler uploadHTTPHandler) handleUpload(w http.ResponseWriter, r *http.Req
return
}
contentType := r.Header.Get("Content-Type")
if !supportedUploadContentType(contentType) {
if err := ingest.ValidateContentType(contentType); err != nil {
writeHTTPError(w, http.StatusUnsupportedMediaType, "unsupported content type")
return
}
@@ -109,19 +102,10 @@ func (handler uploadHTTPHandler) handleUpload(w http.ResponseWriter, r *http.Req
writeHTTPError(w, http.StatusServiceUnavailable, "upload queue is full")
return
}
body, err := readUploadBody(r.Body, handler.limits[pipelineID])
if err != nil {
if errors.Is(err, ingest.ErrUploadTooLarge) {
writeHTTPError(w, http.StatusRequestEntityTooLarge, "upload exceeds maximum size")
return
}
writeHTTPError(w, http.StatusBadRequest, "read upload body failed")
return
}
record, err := handler.coordinator.Submit(r.Context(), UploadRequest{
PipelineID: pipelineID,
ContentType: contentType,
Body: bytes.NewReader(body),
Body: r.Body,
})
if err != nil {
writeUploadSubmitError(w, err)
@@ -160,31 +144,6 @@ func (handler uploadHTTPHandler) authenticate(header string) (string, bool) {
return pipelineID, ok
}
func supportedUploadContentType(contentType string) bool {
mediaType, _, err := mime.ParseMediaType(contentType)
if err != nil {
mediaType = contentType
}
switch mediaType {
case ingest.ContentTypeTar, ingest.ContentTypeGzip, ingest.ContentTypeXGzip:
return true
default:
return false
}
}
func readUploadBody(body io.Reader, maxSize int64) ([]byte, error) {
limited := &io.LimitedReader{R: body, N: maxSize + 1}
data, err := io.ReadAll(limited)
if err != nil {
return nil, err
}
if int64(len(data)) > maxSize {
return nil, ingest.ErrUploadTooLarge
}
return data, nil
}
func writeUploadSubmitError(w http.ResponseWriter, err error) {
switch {
case IsUploadQueueFull(err):

97
internal/app/upload_http_integration_test.go
View File

@@ -13,6 +13,7 @@ import (
"net/http/httptest"
"os"
"path/filepath"
"strings"
"testing"
"time"
@@ -68,32 +69,65 @@ func TestHTTPUploadPublishesTarAndGzipFanout(t *testing.T) {
}
}
func TestHTTPUploadInvalidArchiveFailsWithoutPublishing(t *testing.T) {
func TestHTTPUploadInvalidArchiveIsRejectedWithoutRunID(t *testing.T) {
destination := t.TempDir()
cfg := httpUploadIntegrationConfig(t, []httpUploadPipelineSpec{{
coordinator := NewUploadCoordinator(context.Background(), httpUploadIntegrationConfig(t, []httpUploadPipelineSpec{{
id: "reports",
tokenEnv: "REPORTS_TOKEN",
stagingPath: filepath.Join(t.TempDir(), "reports"),
destinations: []string{destination},
}}, 4, 1)
handler, err := newUploadHTTPHandler(context.Background(), cfg, uploadHTTPTestEnvironment(map[string]string{
"REPORTS_TOKEN": "reports-secret",
}))
if err != nil {
t.Fatalf("newUploadHTTPHandler() error = %v", err)
}}, 4, 1))
handler := uploadHTTPHandler{
coordinator: coordinator,
tokens: map[string]string{"reports-secret": "reports"},
}
server := httptest.NewServer(handler)
defer server.Close()
runID := submitHTTPUpload(t, server, "reports-secret", ingest.ContentTypeTar, []byte("not a tar archive"))
record := waitForHTTPUploadStatus(t, server, runID, UploadStatusFailed)
status, body := postHTTPUpload(t, server, "reports-secret", ingest.ContentTypeTar, []byte("not a tar archive"))
if status != http.StatusBadRequest {
t.Fatalf("POST /upload status = %d, want %d; body = %s", status, http.StatusBadRequest, body)
}
if strings.Contains(body, "run_id") || strings.Contains(body, "reports-secret") {
t.Fatalf("invalid archive response exposed run id or token: %s", body)
}
if got := coordinator.QueueDepth(); got != 0 {
t.Fatalf("queue depth = %d, want 0", got)
}
assertDirectoryEmpty(t, destination)
}
if record.Error == "" {
t.Fatal("failed status error is empty")
func TestHTTPUploadOversizedArchiveIsRejectedWithoutRunID(t *testing.T) {
destination := t.TempDir()
stagingPath := filepath.Join(t.TempDir(), "reports")
cfg := httpUploadIntegrationConfig(t, []httpUploadPipelineSpec{{
id: "reports",
tokenEnv: "REPORTS_TOKEN",
stagingPath: stagingPath,
destinations: []string{destination},
}}, 4, 1)
size := config.ByteSize(4)
cfg.Server.HTTP.MaxUploadSize = &size
cfg.Pipelines[0].Source.Upload.MaxUploadSize = &size
coordinator := NewUploadCoordinator(context.Background(), cfg)
handler := uploadHTTPHandler{
coordinator: coordinator,
tokens: map[string]string{"reports-secret": "reports"},
}
if record.Report != nil {
t.Fatalf("failed staging report = %#v, want nil", record.Report)
server := httptest.NewServer(handler)
defer server.Close()
status, body := postHTTPUpload(t, server, "reports-secret", ingest.ContentTypeTar, bundleArchive(t, false, testutil.BundleOptions{}))
if status != http.StatusRequestEntityTooLarge {
t.Fatalf("POST /upload status = %d, want %d; body = %s", status, http.StatusRequestEntityTooLarge, body)
}
if strings.Contains(body, "run_id") || strings.Contains(body, "reports-secret") {
t.Fatalf("oversized response exposed run id or token: %s", body)
}
if got := coordinator.QueueDepth(); got != 0 {
t.Fatalf("queue depth = %d, want 0", got)
}
assertDirectoryEmpty(t, stagingPath)
assertDirectoryEmpty(t, destination)
}
@@ -122,7 +156,6 @@ func TestHTTPUploadSamePipelineRequestsSerialize(t *testing.T) {
handler := uploadHTTPHandler{
coordinator: coordinator,
tokens: map[string]string{"reports-secret": "reports"},
limits: map[string]int64{"reports": 1024},
}
server := httptest.NewServer(handler)
defer server.Close()
@@ -177,10 +210,6 @@ func TestHTTPUploadDifferentPipelinesRunConcurrently(t *testing.T) {
"one-secret": "reports-one",
"two-secret": "reports-two",
},
limits: map[string]int64{
"reports-one": 1024,
"reports-two": 1024,
},
}
server := httptest.NewServer(handler)
defer server.Close()
@@ -247,6 +276,22 @@ func httpUploadIntegrationConfig(t *testing.T, pipelines []httpUploadPipelineSpe
}
func submitHTTPUpload(t *testing.T, server *httptest.Server, token, contentType string, body []byte) UploadRunID {
t.Helper()
status, responseBody := postHTTPUpload(t, server, token, contentType, body)
if status != http.StatusAccepted {
t.Fatalf("POST /upload status = %d, want %d; body = %s", status, http.StatusAccepted, responseBody)
}
var accepted uploadAcceptedResponse
if err := json.Unmarshal([]byte(responseBody), &accepted); err != nil {
t.Fatalf("decode accepted response: %v", err)
}
if accepted.RunID == "" || accepted.Status != UploadStatusAccepted {
t.Fatalf("accepted response = %#v, want run id and accepted status", accepted)
}
return accepted.RunID
}
func postHTTPUpload(t *testing.T, server *httptest.Server, token, contentType string, body []byte) (int, string) {
t.Helper()
request, err := http.NewRequest(http.MethodPost, server.URL+"/upload", bytes.NewReader(body))
if err != nil {
@@ -259,17 +304,11 @@ func submitHTTPUpload(t *testing.T, server *httptest.Server, token, contentType
t.Fatalf("POST /upload error = %v", err)
}
defer response.Body.Close()
if response.StatusCode != http.StatusAccepted {
t.Fatalf("POST /upload status = %d, want %d", response.StatusCode, http.StatusAccepted)
data, err := io.ReadAll(response.Body)
if err != nil {
t.Fatalf("read response body: %v", err)
}
var accepted uploadAcceptedResponse
if err := json.NewDecoder(response.Body).Decode(&accepted); err != nil {
t.Fatalf("decode accepted response: %v", err)
}
if accepted.RunID == "" || accepted.Status != UploadStatusAccepted {
t.Fatalf("accepted response = %#v, want run id and accepted status", accepted)
}
return accepted.RunID
return response.StatusCode, string(data)
}
func waitForHTTPUploadStatus(t *testing.T, server *httptest.Server, runID UploadRunID, status UploadStatus) UploadRunRecord {

52
internal/app/upload_http_test.go
View File

@@ -12,6 +12,7 @@ import (
"time"
"gitea.maximumdirect.net/eric/distributor/internal/config"
"gitea.maximumdirect.net/eric/distributor/internal/ingest"
)
type fakeUploadCoordinator struct {
@@ -41,7 +42,7 @@ func (fake fakeUploadCoordinator) Status(runID UploadRunID) (UploadRunRecord, bo
func TestResolveUploadTokensFailsForMissingAndDuplicateTokens(t *testing.T) {
cfg := uploadHTTPTestConfig()
_, _, err := resolveUploadTokens(cfg, config.NewEnvironment(nil, func(string) (string, bool) {
_, err := resolveUploadTokens(cfg, config.NewEnvironment(nil, func(string) (string, bool) {
return "", false
}))
if err == nil || !strings.Contains(err.Error(), "UPLOAD_TOKEN") {
@@ -58,7 +59,7 @@ func TestResolveUploadTokensFailsForMissingAndDuplicateTokens(t *testing.T) {
})
config.ApplyDefaults(&cfg)
secret := "super-secret-token"
_, _, err = resolveUploadTokens(cfg, uploadHTTPTestEnvironment(map[string]string{
_, err = resolveUploadTokens(cfg, uploadHTTPTestEnvironment(map[string]string{
"UPLOAD_TOKEN": secret,
"OTHER_UPLOAD_TOKEN": secret,
}))
@@ -110,7 +111,6 @@ func TestUploadHTTPHandlerAuthenticatesAndAcceptsUpload(t *testing.T) {
},
},
tokens: map[string]string{"valid-token": "reports"},
limits: map[string]int64{"reports": 1024},
}
recorder := httptest.NewRecorder()
request := httptest.NewRequest(http.MethodPost, "/upload", strings.NewReader("archive"))
@@ -141,7 +141,6 @@ func TestUploadHTTPHandlerRejectsUnauthorizedRequests(t *testing.T) {
handler := uploadHTTPHandler{
coordinator: fakeUploadCoordinator{canAccept: true},
tokens: map[string]string{"valid-token": "reports"},
limits: map[string]int64{"reports": 1024},
}
for _, authHeader := range []string{"", "Bearer wrong-token"} {
@@ -178,14 +177,6 @@ func TestUploadHTTPHandlerRejectsUnsupportedOversizedFullQueueAndPipelineID(t *t
body: strings.NewReader("archive"),
wantStatus: http.StatusUnsupportedMediaType,
},
{
name: "oversized",
canAccept: true,
url: "/upload",
contentType: "application/x-tar",
body: strings.NewReader("too-large"),
wantStatus: http.StatusRequestEntityTooLarge,
},
{
name: "full queue",
canAccept: false,
@@ -214,7 +205,6 @@ func TestUploadHTTPHandlerRejectsUnsupportedOversizedFullQueueAndPipelineID(t *t
},
},
tokens: map[string]string{"valid-token": "reports"},
limits: map[string]int64{"reports": 4},
}
recorder := httptest.NewRecorder()
request := httptest.NewRequest(http.MethodPost, tt.url, tt.body)
@@ -233,6 +223,41 @@ func TestUploadHTTPHandlerRejectsUnsupportedOversizedFullQueueAndPipelineID(t *t
}
}
func TestUploadHTTPHandlerMapsSubmitErrors(t *testing.T) {
tests := []struct {
name string
err error
wantStatus int
}{
{name: "oversized", err: ingest.ErrUploadTooLarge, wantStatus: http.StatusRequestEntityTooLarge},
{name: "unsupported", err: ingest.ErrUnsupportedContentType, wantStatus: http.StatusUnsupportedMediaType},
{name: "malformed", err: errors.New("malformed archive"), wantStatus: http.StatusBadRequest},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
handler := uploadHTTPHandler{
coordinator: fakeUploadCoordinator{
canAccept: true,
submit: func(context.Context, UploadRequest) (UploadRunRecord, error) {
return UploadRunRecord{}, tt.err
},
},
tokens: map[string]string{"valid-token": "reports"},
}
recorder := httptest.NewRecorder()
request := httptest.NewRequest(http.MethodPost, "/upload", strings.NewReader("archive"))
request.Header.Set("Authorization", "Bearer valid-token")
request.Header.Set("Content-Type", "application/x-tar")
handler.ServeHTTP(recorder, request)
if recorder.Code != tt.wantStatus {
t.Fatalf("status = %d, want %d; body = %q", recorder.Code, tt.wantStatus, recorder.Body.String())
}
})
}
}
func TestUploadHTTPHandlerRunStatusAndHealth(t *testing.T) {
finishedAt := time.Date(2026, 6, 3, 12, 0, 0, 0, time.UTC)
handler := uploadHTTPHandler{
@@ -251,7 +276,6 @@ func TestUploadHTTPHandlerRunStatusAndHealth(t *testing.T) {
},
},
tokens: map[string]string{"valid-token": "reports"},
limits: map[string]int64{"reports": 1024},
}
recorder := httptest.NewRecorder()

5
internal/ingest/archive.go
View File

@@ -138,6 +138,11 @@ func validateRunID(value string) error {
return nil
}
func ValidateContentType(contentType string) error {
_, err := archiveFormat(contentType)
return err
}
type archiveKind int
const (

19
internal/ingest/archive_test.go
View File

@@ -47,6 +47,25 @@ func TestStageArchiveRejectsUnsupportedContentType(t *testing.T) {
}
}
func TestValidateContentType(t *testing.T) {
for _, contentType := range []string{
ContentTypeTar,
ContentTypeGzip,
ContentTypeXGzip,
ContentTypeGzip + "; charset=binary",
} {
t.Run(contentType, func(t *testing.T) {
if err := ValidateContentType(contentType); err != nil {
t.Fatalf("ValidateContentType() error = %v", err)
}
})
}
if err := ValidateContentType("application/zip"); !errors.Is(err, ErrUnsupportedContentType) {
t.Fatalf("ValidateContentType() error = %v, want ErrUnsupportedContentType", err)
}
}
func TestStageArchiveEnforcesMaxUploadSize(t *testing.T) {
archive := validArchive(t, false)
err := stageArchiveError(t, archive, ContentTypeTar, func(opts *StageOptions) {