- commit
- 96951c95682adfc827b1c651f2feaa2d396e574c
- parent
- 1ae154cde788fe00226602f3cd9d2febf2f885d3
- author
- Dami <dami.etoile@gmail.com>
| M | internal/ | 13 | +- |
| M | internal/ | 38 | +++++- |
| M | internal/ | 218 | ++++++++++++++++++++++++++++++ |
| A | internal/ | 71 | ++++++++++ |
@@ -93,6 +93,13 @@ type Service struct {
9393 // before quotaMu; network bodies are never read while either is held.9494 multipartMu sync.Mutex9595 activeParts map[string]int96+ // multipartFinalizing prevents another chunk from changing a complete97+ // session while its assembled file is being hashed. Hashing is deliberately98+ // outside multipartMu because the upload may be as large as 1 GiB.99+ multipartFinalizing map[string]bool100+ // verifyMultipartHook is used by package tests to pause finalization101+ // without allocating a GiB fixture. It is nil in production.102+ verifyMultipartHook func(string, string, int64) error96103 }97104 98105 var ErrBlocked = errors.New("blocked: this blob was removed by a moderator")@@ -159,7 +166,7 @@ func New(ctx context.Context, config Config) (*Service, error) {
159166 return net.DefaultResolver.LookupIP(ctx, "ip", host)160167 }161168 }162- s := &Service{config: config, root: filepath.Join(config.Root, "blobs"), store: config.Store, resolve: resolve, activeParts: make(map[string]int)}169+ s := &Service{config: config, root: filepath.Join(config.Root, "blobs"), store: config.Store, resolve: resolve, activeParts: make(map[string]int), multipartFinalizing: make(map[string]bool)}163170 if err := s.reconcileMultipart(ctx); err != nil {164171 return nil, err165172 }@@ -430,8 +437,8 @@ func (s *Service) uploadHead(w http.ResponseWriter, r *http.Request) {
430437 return431438 }432439 length, err := strconv.ParseInt(rawLength, 10, 64)433- if err != nil || length <= 0 {434- s.fail(w, http.StatusBadRequest, errors.New("invalid: X-Content-Length must be a positive integer"))440+ if err != nil || length < 0 {441+ s.fail(w, http.StatusBadRequest, errors.New("invalid: X-Content-Length must be a nonnegative integer"))435442 return436443 }437444 pubkey, err := s.authorize(r, ActionUpload)@@ -285,46 +285,78 @@ func (s *Service) releaseMultipart(id string) {
285285 286286 func (s *Service) commitMultipart(ctx context.Context, part multipartRequest, chunk string) (Blob, bool, bool, error) {287287 s.multipartMu.Lock()288- defer s.multipartMu.Unlock()288+ if s.multipartFinalizing[part.id] {289+ s.multipartMu.Unlock()290+ return Blob{}, false, false, multipartFailure(http.StatusConflict, "multipart upload is being finalized")291+ }289292 // Recheck policy after streaming, including uploads whose next chunk does290293 // not complete the file. Policy may have changed while the body arrived.291294 s.quotaMu.Lock()292295 admissionErr := s.multipartAdmission(ctx, part)293296 s.quotaMu.Unlock()294297 if admissionErr != nil {298+ s.multipartMu.Unlock()295299 return Blob{}, false, false, admissionErr296300 }297301 candidate := uploadCandidate{hash: part.hash, size: part.length, typ: part.typ, uploader: part.uploader, reservation: part.id}298302 if existing, err := s.lookup(ctx, part.hash); err == nil {299303 if existing.Size != part.length {304+ s.multipartMu.Unlock()300305 return Blob{}, false, false, ErrMultipartConflict301306 }302307 entry, _, err := s.installUpload(ctx, candidate)308+ s.multipartMu.Unlock()303309 return entry, false, err == nil, err304310 } else if !errors.Is(err, sql.ErrNoRows) {311+ s.multipartMu.Unlock()305312 return Blob{}, false, false, err306313 }307314 ranges, err := s.multipartRanges(ctx, part)308315 if err != nil {316+ s.multipartMu.Unlock()309317 return Blob{}, false, false, err310318 }311319 candidate.path = s.multipartPath(part.id)312320 if err := writeMultipartChunk(candidate.path, chunk, part.offset); err != nil {313321 _ = s.discardMultipart(ctx, part.id)322+ s.multipartMu.Unlock()314323 return Blob{}, false, false, err315324 }316325 if err := s.recordMultipartRanges(ctx, part.id, ranges); err != nil {317326 _ = s.discardMultipart(ctx, part.id)327+ s.multipartMu.Unlock()318328 return Blob{}, false, false, err319329 }320330 if len(ranges) != 1 || ranges[0].start != 0 || ranges[0].end != part.length {331+ s.multipartMu.Unlock()321332 return Blob{}, false, false, nil322333 }323- if err := verifyMultipart(candidate.path, part.hash, part.length); err != nil {324- return Blob{}, false, false, errors.Join(err, s.discardMultipart(ctx, part.id))334+ // Keep the session reserved while hashing, but let unrelated multipart335+ // sessions and ordinary uploads proceed. The finalizing marker makes this336+ // small state transition safe without holding the global mutex for the hash.337+ s.multipartFinalizing[part.id] = true338+ s.multipartMu.Unlock()339+ verify := verifyMultipart340+ if s.verifyMultipartHook != nil {341+ verify = s.verifyMultipartHook342+ }343+ if err := verify(candidate.path, part.hash, part.length); err != nil {344+ s.multipartMu.Lock()345+ delete(s.multipartFinalizing, part.id)346+ // A canceled request must not throw away an otherwise complete session:347+ // the caller can retry finalization without retransmitting its chunks.348+ var cleanupErr error349+ if !errors.Is(err, context.Canceled) && !errors.Is(err, context.DeadlineExceeded) {350+ cleanupErr = s.discardMultipart(ctx, part.id)351+ }352+ s.multipartMu.Unlock()353+ return Blob{}, false, false, errors.Join(err, cleanupErr)325354 }355+ s.multipartMu.Lock()326356 entry, created, err := s.installUpload(ctx, candidate)327357 cleanupErr := s.discardMultipart(ctx, part.id)358+ delete(s.multipartFinalizing, part.id)359+ s.multipartMu.Unlock()328360 return entry, created, err == nil && cleanupErr == nil, errors.Join(err, cleanupErr)329361 }330362 @@ -3,11 +3,13 @@ package blob
33 import (44 "context"55 "crypto/sha256"6+ "database/sql"67 "encoding/hex"78 "errors"89 "io"910 "net/http"1011 "net/http/httptest"12+ "os"1113 "path/filepath"1214 "strings"1315 "sync"@@ -105,6 +107,222 @@ func TestMultipartReviewStalledBodyDoesNotBlockOrdinaryUpload(t *testing.T) {
105107 }106108 }107109 110+func TestMultipartReviewStalledHashDoesNotBlockOrdinaryUpload(t *testing.T) {111+ s := testService(t)112+ body := "hash-paused"113+ digest := sha256.Sum256([]byte(body))114+ sha := hex.EncodeToString(digest[:])115+ started := make(chan struct{})116+ release := make(chan struct{})117+ s.verifyMultipartHook = func(path, hash string, size int64) error {118+ close(started)119+ <-release120+ return verifyMultipart(path, hash, size)121+ }122+ result := make(chan *httptest.ResponseRecorder, 1)123+ r := multipartReviewRequest(sha, int64(len(body)), 0, body)124+ go func() { result <- record(s.Handler(), r) }()125+ select {126+ case <-started:127+ case <-time.After(time.Second):128+ t.Fatal("multipart hash was not started")129+ }130+ ordinary := make(chan error, 1)131+ go func() {132+ _, err := s.Put(context.Background(), PutOptions{Reader: strings.NewReader("ordinary"), Type: "text/plain", Uploader: "other"})133+ ordinary <- err134+ }()135+ select {136+ case err := <-ordinary:137+ if err != nil {138+ t.Fatalf("ordinary upload while multipart hash paused: %v", err)139+ }140+ case <-time.After(time.Second):141+ t.Fatal("ordinary upload blocked by multipart hashing")142+ }143+ close(release)144+ select {145+ case response := <-result:146+ if response.Code != http.StatusCreated {147+ t.Fatalf("multipart completion status = %d: %s", response.Code, response.Body.String())148+ }149+ case <-time.After(time.Second):150+ t.Fatal("multipart hash did not finish")151+ }152+}153+154+func TestMultipartReviewStalledHashDoesNotBlockAnotherSession(t *testing.T) {155+ s := testService(t)156+ body := "hash-paused"157+ digest := sha256.Sum256([]byte(body))158+ sha := hex.EncodeToString(digest[:])159+ started := make(chan struct{})160+ release := make(chan struct{})161+ var unblock sync.Once162+ defer unblock.Do(func() { close(release) })163+ s.verifyMultipartHook = func(path, hash string, size int64) error {164+ close(started)165+ <-release166+ return verifyMultipart(path, hash, size)167+ }168+ first := make(chan *httptest.ResponseRecorder, 1)169+ go func() { first <- record(s.Handler(), multipartReviewRequest(sha, int64(len(body)), 0, body)) }()170+ select {171+ case <-started:172+ case <-time.After(5 * time.Second):173+ t.Fatal("multipart hash did not start")174+ }175+ // A different session must reserve space and commit a partial chunk while176+ // the first session hashes. Ordinary PUT does not take multipartMu and177+ // would not detect the tenant-wide finalization lock regression.178+ secondBody := strings.Repeat("b", 70000)179+ secondDigest := sha256.Sum256([]byte(secondBody + "remaining"))180+ second := make(chan *httptest.ResponseRecorder, 1)181+ go func() {182+ second <- record(s.Handler(), multipartReviewRequest(hex.EncodeToString(secondDigest[:]), int64(len(secondBody)+len("remaining")), 0, secondBody))183+ }()184+ select {185+ case response := <-second:186+ if response.Code != http.StatusNoContent {187+ t.Fatalf("second session = %d: %s", response.Code, response.Body.String())188+ }189+ case <-time.After(5 * time.Second):190+ t.Fatal("another multipart session blocked by hashing")191+ }192+ unblock.Do(func() { close(release) })193+ select {194+ case response := <-first:195+ if response.Code != http.StatusCreated {196+ t.Fatalf("finalizing session = %d: %s", response.Code, response.Body.String())197+ }198+ case <-time.After(5 * time.Second):199+ t.Fatal("finalizing session did not finish")200+ }201+}202+203+func TestMultipartReviewFinalizingSessionRejectsConcurrentChunk(t *testing.T) {204+ s := testService(t)205+ body := "same-session"206+ digest := sha256.Sum256([]byte(body))207+ sha := hex.EncodeToString(digest[:])208+ started := make(chan struct{})209+ release := make(chan struct{})210+ s.verifyMultipartHook = func(path, hash string, size int64) error {211+ close(started)212+ <-release213+ return verifyMultipart(path, hash, size)214+ }215+ first := make(chan *httptest.ResponseRecorder, 1)216+ go func() { first <- record(s.Handler(), multipartReviewRequest(sha, int64(len(body)), 0, body)) }()217+ select {218+ case <-started:219+ case <-time.After(time.Second):220+ t.Fatal("multipart hash was not started")221+ }222+ second := record(s.Handler(), multipartReviewRequest(sha, int64(len(body)), 0, body))223+ if second.Code != http.StatusConflict {224+ t.Fatalf("concurrent same-session chunk status = %d: %s", second.Code, second.Body.String())225+ }226+ close(release)227+ if response := <-first; response.Code != http.StatusCreated {228+ t.Fatalf("finalizing upload status = %d: %s", response.Code, response.Body.String())229+ }230+}231+232+func TestMultipartReviewCleanupDoesNotRemoveActiveFinalization(t *testing.T) {233+ s := testService(t)234+ body := "cleanup-paused"235+ digest := sha256.Sum256([]byte(body))236+ sha := hex.EncodeToString(digest[:])237+ started := make(chan struct{})238+ release := make(chan struct{})239+ s.verifyMultipartHook = func(path, hash string, size int64) error {240+ close(started)241+ <-release242+ return verifyMultipart(path, hash, size)243+ }244+ result := make(chan *httptest.ResponseRecorder, 1)245+ go func() { result <- record(s.Handler(), multipartReviewRequest(sha, int64(len(body)), 0, body)) }()246+ select {247+ case <-started:248+ case <-time.After(time.Second):249+ t.Fatal("multipart hash was not started")250+ }251+ if _, err := s.store.DB().Exec("UPDATE multipart_uploads SET last_seen=0"); err != nil {252+ t.Fatal(err)253+ }254+ if err := s.CleanupMultipart(context.Background()); err != nil {255+ t.Fatal(err)256+ }257+ id := multipartID(sha, "uploader")258+ if _, err := os.Stat(s.multipartPath(id)); err != nil {259+ t.Fatalf("active multipart file removed during cleanup: %v", err)260+ }261+ var found int262+ if err := s.store.DB().QueryRow("SELECT COUNT(*) FROM multipart_uploads WHERE id=?", id).Scan(&found); err != nil {263+ t.Fatal(err)264+ }265+ if found != 1 {266+ t.Fatal("active multipart reservation removed during cleanup")267+ }268+ close(release)269+ if response := <-result; response.Code != http.StatusCreated {270+ t.Fatalf("finalizing upload status = %d: %s", response.Code, response.Body.String())271+ }272+}273+274+func TestMultipartReviewPolicyRecheckedAfterHash(t *testing.T) {275+ s := testService(t)276+ body := "policy-after-hash"277+ digest := sha256.Sum256([]byte(body))278+ sha := hex.EncodeToString(digest[:])279+ started := make(chan struct{})280+ release := make(chan struct{})281+ s.verifyMultipartHook = func(path, hash string, size int64) error {282+ close(started)283+ <-release284+ return verifyMultipart(path, hash, size)285+ }286+ s.config.Limits = func() Limits { return Limits{MaxFileBytes: int64(len(body))} }287+ result := make(chan *httptest.ResponseRecorder, 1)288+ go func() { result <- record(s.Handler(), multipartReviewRequest(sha, int64(len(body)), 0, body)) }()289+ <-started290+ s.config.Limits = func() Limits { return Limits{MaxFileBytes: int64(len(body) - 1)} }291+ close(release)292+ if response := <-result; response.Code != http.StatusRequestEntityTooLarge {293+ t.Fatalf("policy-changed upload status = %d: %s", response.Code, response.Body.String())294+ }295+ if _, _, err := s.Get(context.Background(), sha); !errors.Is(err, sql.ErrNoRows) {296+ t.Fatalf("policy-rejected upload lookup = %v", err)297+ }298+}299+300+func TestMultipartReviewCanceledHashCanBeRetried(t *testing.T) {301+ s := testService(t)302+ body := "cancel-and-retry"303+ digest := sha256.Sum256([]byte(body))304+ sha := hex.EncodeToString(digest[:])305+ started := make(chan struct{})306+ release := make(chan struct{})307+ s.verifyMultipartHook = func(path, hash string, size int64) error {308+ close(started)309+ <-release310+ return verifyMultipart(path, hash, size)311+ }312+ ctx, cancel := context.WithCancel(context.Background())313+ r := multipartReviewRequest(sha, int64(len(body)), 0, body).WithContext(ctx)314+ result := make(chan *httptest.ResponseRecorder, 1)315+ go func() { result <- record(s.Handler(), r) }()316+ <-started317+ cancel()318+ close(release)319+ <-result320+ s.verifyMultipartHook = nil321+ if response := record(s.Handler(), multipartReviewRequest(sha, int64(len(body)), 0, body)); response.Code != http.StatusCreated {322+ t.Fatalf("retry after canceled hash status = %d: %s", response.Code, response.Body.String())323+ }324+}325+108326 type multipartBlockingBody struct {109327 io.Reader110328 started chan<- struct{}@@ -0,0 +1,71 @@
1+package blob2+3+import (4+ "context"5+ "crypto/sha256"6+ "encoding/hex"7+ "net/http"8+ "net/http/httptest"9+ "strings"10+ "testing"11+)12+13+func TestUploadHeadAcceptsEmptyFileAtFullQuota(t *testing.T) {14+ s := testService(t)15+ s.config.Limits = func() Limits { return Limits{MaxFileBytes: 2, UserStorageBytes: 2} }16+ if _, err := s.Put(context.Background(), PutOptions{Reader: strings.NewReader("ab"), Type: "text/plain", Uploader: testUploader}); err != nil {17+ t.Fatal(err)18+ }19+ digest := sha256.Sum256(nil)20+ hash := hex.EncodeToString(digest[:])21+ for _, phase := range []string{"before upload", "after upload"} {22+ req := httptest.NewRequest(http.MethodHead, "https://relay.test/upload", nil)23+ req.Header.Set("X-SHA-256", hash)24+ req.Header.Set("X-Content-Length", "0")25+ req.Header.Set("X-Content-Type", "application/octet-stream")26+ response := record(s.Handler(), req)27+ if response.Code != http.StatusOK {28+ t.Fatalf("%s: empty preflight = %d %s", phase, response.Code, response.Body.String())29+ }30+ if phase == "before upload" {31+ upload := record(s.Handler(), httptest.NewRequest(http.MethodPut, "https://relay.test/"+hash, http.NoBody))32+ if upload.Code != http.StatusCreated {33+ t.Fatalf("empty PUT = %d %s", upload.Code, upload.Body.String())34+ }35+ }36+ }37+ stored, file, err := s.Get(context.Background(), hash)38+ if err != nil {39+ t.Fatal(err)40+ }41+ defer file.Close()42+ if stored.Size != 0 {43+ t.Fatalf("empty stored blob = %+v, %v", stored, err)44+ }45+ usage, err := s.quotaUsage(context.Background(), testUploader)46+ if err != nil || usage != 2 {47+ t.Fatalf("empty file changed usage: %d, %v", usage, err)48+ }49+}50+51+func TestUploadHeadRequiresNonnegativeLength(t *testing.T) {52+ for _, tc := range []struct {53+ value string54+ code int55+ }{56+ {"", http.StatusLengthRequired}, {"-1", http.StatusBadRequest},57+ {"no", http.StatusBadRequest}, {"9223372036854775808", http.StatusBadRequest},58+ {"0", http.StatusOK}, {"1", http.StatusOK},59+ } {60+ t.Run(tc.value, func(t *testing.T) {61+ s := testService(t)62+ req := httptest.NewRequest(http.MethodHead, "https://relay.test/upload", nil)63+ req.Header.Set("X-SHA-256", strings.Repeat("a", 64))64+ req.Header.Set("X-Content-Type", "text/plain")65+ req.Header.Set("X-Content-Length", tc.value)66+ if response := record(s.Handler(), req); response.Code != tc.code {67+ t.Fatalf("length %q = %d, want %d", tc.value, response.Code, tc.code)68+ }69+ })70+ }71+}