tapir/internal/uploads/intake_concurrency_test.go

package uploads_test

// Observes how many request bodies Accept will read at once (#47). HasRoom and the
// per-user queue cap act before and after the receive; N parallel POSTs all pass
// HasRoom while the queue is still empty, so the receive itself is bounded.

import (
	"context"
	"errors"
	"io"
	"strings"
	"testing"
	"time"

	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/require"

	"git.d-ma.be/mathias/tapir/internal/domain"
	"git.d-ma.be/mathias/tapir/internal/uploads"
)

// 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
	finished bool
}

func startSlowReceive(t *testing.T, r *rig, in *uploads.Intake, user string, wantFiles int) *slowReceive {
	t.Helper()
	pr, pw := io.Pipe()
	s := &slowReceive{w: pw, done: make(chan error, 1)}
	go func() {
		_, err := in.Accept(context.Background(), user, "t", domain.LangSwedish, pr)
		s.done <- err
	}()
	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)
	// 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
}

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
	case <-time.After(2 * time.Second):
		t.Fatal("Accept did not return after its body ended")
		return nil
	}
}

func requireBusy(t *testing.T, err error) {
	t.Helper()
	var rej *uploads.RejectError
	require.True(t, errors.As(err, &rej), "want a busy rejection, got %v", err)
	assert.Equal(t, uploads.ReasonBusy, rej.Reason)
}

func TestAccept_APerUserLimitOnConcurrentReceivesRefusesTheNextBodyBeforeSpoolingIt(t *testing.T) {
	r := newRig(t)
	in := uploads.NewIntake(r.st, r.pr, r.sub, uploads.Limits{MaxBytes: 1024, MaxDuration: time.Hour}, r.tmp)
	in.MaxReceivesPerUser = 2
	in.MaxReceives = 10
	startSlowReceive(t, r, in, "user-1", 1)
	startSlowReceive(t, r, in, "user-1", 2)

	_, err := in.Accept(context.Background(), "user-1", "t", domain.LangSwedish, strings.NewReader("data"))

	requireBusy(t, err)
	assert.Len(t, r.tmpFiles(t), 2, "the refused body was never spooled")
	// Another user is not blocked by user-1's two receives.
	other := startSlowReceive(t, r, in, "user-2", 3)
	require.NoError(t, other.finish(t))
}

func TestAccept_AnOverallLimitOnConcurrentReceivesRefusesTheNextUser(t *testing.T) {
	r := newRig(t)
	in := uploads.NewIntake(r.st, r.pr, r.sub, uploads.Limits{MaxBytes: 1024, MaxDuration: time.Hour}, r.tmp)
	in.MaxReceivesPerUser = 5
	in.MaxReceives = 3
	for i, u := range []string{"a", "b", "c"} {
		startSlowReceive(t, r, in, u, i+1)
	}

	_, err := in.Accept(context.Background(), "d", "t", domain.LangSwedish, strings.NewReader("data"))

	requireBusy(t, err)
	assert.Len(t, r.tmpFiles(t), 3)
}

func TestAccept_AFinishedReceiveFreesItsSlot(t *testing.T) {
	r := newRig(t)
	in := uploads.NewIntake(r.st, r.pr, r.sub, uploads.Limits{MaxBytes: 1024, MaxDuration: time.Hour}, r.tmp)
	in.MaxReceivesPerUser = 1
	in.MaxReceives = 5
	first := startSlowReceive(t, r, in, "user-1", 1)
	require.NoError(t, first.finish(t))

	_, err := in.Accept(context.Background(), "user-1", "t", domain.LangSwedish, strings.NewReader("data"))

	require.NoError(t, err, "the slot is released when Accept returns")
}

func TestAccept_AFailedReceiveFreesItsSlot(t *testing.T) {
	r := newRig(t)
	in := uploads.NewIntake(r.st, r.pr, r.sub, uploads.Limits{MaxBytes: 1024, MaxDuration: time.Hour}, r.tmp)
	in.MaxReceivesPerUser = 1
	in.MaxReceives = 5
	_, err := in.Accept(context.Background(), "user-1", "t", domain.LangSwedish, &brokenBody{n: 1})
	require.ErrorIs(t, err, uploads.ErrClientRead)

	_, err = in.Accept(context.Background(), "user-1", "t", domain.LangSwedish, strings.NewReader("data"))

	require.NoError(t, err, "a dropped client must not leak a slot")
}