fix: make local docker uploads replica-independent
ci/woodpecker/pr/pre-commit Pipeline was successful
ci/woodpecker/pr/test Pipeline was successful
ci/woodpecker/pr/build Pipeline was successful

Chunked blob uploads kept the in-progress session in process memory keyed by
upload UUID, so the POST/PATCH/PUT of a single `docker push` had to land on the
same replica. In production the API runs at minReplicas 2 with no session
affinity, so a real push (which streams the layer via PATCH then finalises with
PUT) intermittently 404s with BLOB_UPLOAD_UNKNOWN when chunks hit a replica that
never saw the POST.

- Stage chunked uploads in object storage under uploads/<uuid> instead of an
  in-memory temp file. The UUID travels in the Location URL handed to the
  client, so any replica reconstructs the staging key with no shared in-process
  state. Finalise streams the staged bytes plus any trailing PUT body through
  the CAS in one pass; monolithic uploads are unchanged.
- Support DELETE of an in-progress upload (cancel) by dropping its staging
  object.
- Reap abandoned staging objects in the GC (uploads/ older than 24h) via a new
  S3 ListStaleObjects, so cancelled/interrupted pushes don't leak.

Verified by splitting a single push across two instances sharing one
Postgres+MinIO: POST->A, PATCH->B, PUT->A finalises and the blob pulls back
byte-identical from both. The dockerised e2e suite still passes.
This commit is contained in:
2026-07-05 17:31:27 +10:00
parent a92ede23f6
commit f9f78955e6
4 changed files with 121 additions and 66 deletions
+16
View File
@@ -5,6 +5,7 @@ import (
"fmt"
"io"
"log/slog"
"time"
"github.com/minio/minio-go/v7"
"github.com/minio/minio-go/v7/pkg/credentials"
@@ -97,3 +98,18 @@ func (s *S3) Stat(ctx context.Context, key string) (*minio.ObjectInfo, error) {
}
return &info, nil
}
// ListStaleObjects returns keys under prefix last modified before cutoff. Used
// by the GC to reap abandoned staging objects (e.g. cancelled docker pushes).
func (s *S3) ListStaleObjects(ctx context.Context, prefix string, cutoff time.Time) ([]string, error) {
var keys []string
for obj := range s.client.ListObjects(ctx, s.bucket, minio.ListObjectsOptions{Prefix: prefix, Recursive: true}) {
if obj.Err != nil {
return nil, obj.Err
}
if obj.LastModified.Before(cutoff) {
keys = append(keys, obj.Key)
}
}
return keys, nil
}