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)

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