diff --git a/internal/uploads/intake.go b/internal/uploads/intake.go index 8f5ce13..4de7f31 100644 --- a/internal/uploads/intake.go +++ b/internal/uploads/intake.go @@ -6,6 +6,7 @@ import ( "fmt" "io" "os" + "sync" "time" "git.d-ma.be/mathias/tapir/internal/domain" @@ -51,6 +52,53 @@ type Intake struct { sub Submitter limits Limits tmpDir string + + // MaxReceives and MaxReceivesPerUser bound how many request bodies Accept reads at + // once; zero means the defaults. Without them N parallel POSTs each spool up to + // MaxBytes while the queue still looks empty, and the spool volume fills (#47). + MaxReceives int + MaxReceivesPerUser int + + mu sync.Mutex + receives int + byUser map[string]int +} + +const ( + defaultMaxReceives = 4 + defaultMaxReceivesPerUser = 2 +) + +// acquireReceive reserves a receive slot for userID, or reports there is none. Every +// successful call must be paired with releaseReceive. +func (in *Intake) acquireReceive(userID string) bool { + maxAll, maxUser := in.MaxReceives, in.MaxReceivesPerUser + if maxAll <= 0 { + maxAll = defaultMaxReceives + } + if maxUser <= 0 { + maxUser = defaultMaxReceivesPerUser + } + in.mu.Lock() + defer in.mu.Unlock() + if in.receives >= maxAll || in.byUser[userID] >= maxUser { + return false + } + if in.byUser == nil { + in.byUser = map[string]int{} + } + in.receives++ + in.byUser[userID]++ + return true +} + +func (in *Intake) releaseReceive(userID string) { + in.mu.Lock() + defer in.mu.Unlock() + in.receives-- + if in.byUser[userID]--; in.byUser[userID] <= 0 { + delete(in.byUser, userID) + } } // NewIntake wires an Intake. tmpDir is where uploads are spooled; it must exist. @@ -94,6 +142,11 @@ func (in *Intake) Accept(ctx context.Context, userID, title string, src io.Reade if rc, ok := in.sub.(interface{ HasRoom(userID string) bool }); ok && !rc.HasRoom(userID) { return "", &RejectError{Reason: ReasonBusy, Message: busyMessage} } + // HasRoom cannot see bodies still being received; bound those separately. + if !in.acquireReceive(userID) { + return "", &RejectError{Reason: ReasonBusy, Message: busyMessage} + } + defer in.releaseReceive(userID) f, err := os.CreateTemp(in.tmpDir, "upload-*") if err != nil { return "", fmt.Errorf("uploads: spool: %w", err) diff --git a/internal/uploads/intake_concurrency_test.go b/internal/uploads/intake_concurrency_test.go index 5a1b56e..c245903 100644 --- a/internal/uploads/intake_concurrency_test.go +++ b/internal/uploads/intake_concurrency_test.go @@ -21,8 +21,9 @@ import ( // slowReceive starts an Accept whose body stays open until the returned writer is closed, // and waits until it holds a spool file, which proves it is inside the receive. type slowReceive struct { - w *io.PipeWriter - done chan error + w *io.PipeWriter + done chan error + finished bool } func startSlowReceive(t *testing.T, r *rig, in *uploads.Intake, user string, wantFiles int) *slowReceive { @@ -35,7 +36,18 @@ func startSlowReceive(t *testing.T, r *rig, in *uploads.Intake, user string, wan }() require.Eventually(t, func() bool { return len(r.tmpFiles(t)) == wantFiles }, 2*time.Second, 5*time.Millisecond, "the receive for %s should have started", user) - t.Cleanup(func() { _ = pw.Close() }) + // An abandoned receive ends in a read error, so it never reaches the rig's fakes, which + // are not safe for concurrent use. A receive that finish() already ended is a no-op. + t.Cleanup(func() { + if s.finished { + return + } + _ = pw.CloseWithError(errors.New("test over")) + select { + case <-s.done: + case <-time.After(2 * time.Second): + } + }) return s } @@ -43,6 +55,7 @@ func (s *slowReceive) finish(t *testing.T) error { t.Helper() _, _ = s.w.Write([]byte("data")) require.NoError(t, s.w.Close()) + s.finished = true select { case err := <-s.done: return err