dsh-web/internal/api/config_body_test.go

package api

import (
	"bytes"
	"io"
	"mime/multipart"
	"net/http/httptest"
	"os"
	"strings"
	"sync"
	"testing"
	"time"

	"git.misaka.ren/m1saka/dsh-web/internal/store"
)

type configPausedBody struct {
	io.Reader
	started chan struct{}
	release chan struct{}
	once    sync.Once
}

func (b *configPausedBody) Read(p []byte) (int, error) {
	b.once.Do(func() { close(b.started) })
	<-b.release
	return b.Reader.Read(p)
}

func TestConfigSlowBodyRevalidatesWithoutLocks(t *testing.T) {
	for _, action := range []string{"profile", "backups", "restore", "multipart"} {
		for _, change := range []string{"lease", "deleted"} {
			t.Run(action+"/"+change, func(t *testing.T) {
				a, i, _ := configTestAPI(t, false)
				a.workspaceReaders = make(map[string]int)
				body := &configPausedBody{Reader: strings.NewReader(`{}`), started: make(chan struct{}), release: make(chan struct{})}
				r := httptest.NewRequest("POST", "/", body)
				if action == "profile" {
					r.Method = "PUT"
				}
				target := action
				if action == "multipart" {
					target = "backups"
					var buf bytes.Buffer
					form := multipart.NewWriter(&buf)
					part, _ := form.CreateFormFile("file", "test.dshbak")
					part.Write(bytes.Repeat([]byte("x"), 2<<20))
					form.Close()
					body.Reader = &buf
					r.Header.Set("Content-Type", form.FormDataContentType())
					t.Setenv("TMPDIR", t.TempDir())
				}
				w := httptest.NewRecorder()
				done := make(chan struct{})
				go func() { defer close(done); a.config(w, r, i, target) }()
				defer func() {
					select {
					case <-body.release:
					default:
						close(body.release)
					}
					<-done
				}()
				select {
				case <-body.started:
				case <-time.After(3 * time.Second):
					t.Fatal("body read did not start")
				}
				if a.instanceLocks.held() || a.userLocks.held() {
					t.Fatal("slow body holds API locks")
				}
				if change == "lease" {
					a.readersMu.Lock()
					a.workspaceReaders[i.ID]++
					a.readersMu.Unlock()
				}
				want := 409
				if change == "deleted" {
					if _, err := a.s.DB.Exec("DELETE FROM instances WHERE id=?", i.ID); err != nil {
						t.Fatal(err)
					}
					want = 404
				}
				close(body.release)
				select {
				case <-done:
				case <-time.After(3 * time.Second):
					t.Fatal("config did not finish")
				}
				if w.Code != want {
					t.Fatalf("status %d: %s", w.Code, w.Body.String())
				}
				if action == "multipart" {
					entries, err := os.ReadDir(os.Getenv("TMPDIR"))
					if err != nil || len(entries) != 0 {
						t.Fatalf("multipart temp files remain: %v %v", entries, err)
					}
				}
			})
		}
	}
}

type configDeadlineWriter struct {
	*httptest.ResponseRecorder
	deadlines []time.Time
}

func (w *configDeadlineWriter) SetReadDeadline(deadline time.Time) error {
	w.deadlines = append(w.deadlines, deadline)
	return nil
}

type configErrorBody struct{}

func (configErrorBody) Read([]byte) (int, error) { return 0, os.ErrDeadlineExceeded }

func TestConfigBodyErrorsReturnWithoutLocks(t *testing.T) {
	for _, tc := range []struct {
		name string
		body io.Reader
		want int
	}{
		{"oversized", strings.NewReader(strings.Repeat("x", 4097)), 413},
		{"timeout", configErrorBody{}, 408},
		{"multipart", strings.NewReader("bad multipart"), 400},
	} {
		t.Run(tc.name, func(t *testing.T) {
			a := &API{}
			r := httptest.NewRequest("POST", "/", tc.body)
			r.ContentLength = -1
			if tc.name == "multipart" {
				r.Header.Set("Content-Type", "multipart/form-data; boundary=test")
			}
			w := &configDeadlineWriter{ResponseRecorder: httptest.NewRecorder()}
			inst := a.instanceLocks.Acquire("")
			user := a.userLocks.Acquire("")
			done := make(chan struct{})
			go func() { defer close(done); a.config(w, r, store.Instance{}, "backups") }()
			select {
			case <-done:
				user.Release()
				inst.Release()
			case <-time.After(3 * time.Second):
				user.Release()
				inst.Release()
				t.Fatal("body error waited for locks")
			}
			if w.Code != tc.want {
				t.Fatalf("status %d: %s", w.Code, w.Body.String())
			}
			if len(w.deadlines) != 2 || w.deadlines[0].IsZero() || !w.deadlines[1].IsZero() {
				t.Fatalf("deadlines: %v", w.deadlines)
			}
		})
	}
}

func TestConfigBodySlotsRejectWithoutReading(t *testing.T) {
	for n := 0; n < cap(configBodySlots); n++ {
		configBodySlots <- struct{}{}
	}
	defer func() {
		for n := 0; n < cap(configBodySlots); n++ {
			<-configBodySlots
		}
	}()
	w := httptest.NewRecorder()
	(&API{}).config(w, httptest.NewRequest("POST", "/", configErrorBody{}), store.Instance{}, "backups")
	if w.Code != 503 {
		t.Fatalf("status %d", w.Code)
	}
}

func TestConfigSlowResponseReleasesLocks(t *testing.T) {
	for _, tc := range []struct{ method, action string }{{"POST", "restore"}, {"GET", "profile"}, {"GET", "backups"}} {
		t.Run(tc.method+"/"+tc.action, func(t *testing.T) {
			a, i, _ := configTestAPI(t, false)
			w := &pausedDownload{ResponseRecorder: httptest.NewRecorder(), started: make(chan struct{}), release: make(chan struct{})}
			done := make(chan struct{})
			go func() {
				defer close(done)
				a.config(w, httptest.NewRequest(tc.method, "/", strings.NewReader(`{}`)), i, tc.action)
			}()
			defer func() { close(w.release); <-done }()
			select {
			case <-w.started:
			case <-time.After(3 * time.Second):
				t.Fatal("response write did not start")
			}
			if a.instanceLocks.held() {
				t.Fatal("slow response holds instance lock")
			}
			if a.userLocks.held() {
				t.Fatal("slow response holds user lock")
			}

		})
	}
}