339 lines
12 KiB
Go
339 lines
12 KiB
Go
package submit
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"os"
|
|
"path/filepath"
|
|
"strings"
|
|
"testing"
|
|
"time"
|
|
)
|
|
|
|
// newChunkedManager wires a Manager with an in-memory Blobs, a PartStore on a
|
|
// temp dir and 4-byte parts, and files one pending submission for user-1.
|
|
func newChunkedManager(t *testing.T) (*Manager, *fakeStore, *fakeBlobs, string) {
|
|
t.Helper()
|
|
m, st, _ := newManager()
|
|
fb := newFakeBlobs()
|
|
m.Blobs = fb
|
|
// A near-zero floor keeps the room check off this machine's own disk usage.
|
|
m.Parts = &PartStore{Dir: t.TempDir(), MinFree: 1e-9}
|
|
m.PartMaxBytes = 4
|
|
sub, err := m.Create(context.Background(), CreateRequest{DisplayName: "Pack", SubmittedBy: "user-1"})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
return m, st, fb, sub.ID
|
|
}
|
|
|
|
func sendPart(t *testing.T, m *Manager, id string, offset int64, part string) int64 {
|
|
t.Helper()
|
|
p, err := m.UploadPart(context.Background(), id, "user-1", offset, strings.NewReader(part))
|
|
if err != nil {
|
|
t.Fatalf("UploadPart(offset %d, %q): %v", offset, part, err)
|
|
}
|
|
return p.Received
|
|
}
|
|
|
|
func staged(t *testing.T, m *Manager, id string) int64 {
|
|
t.Helper()
|
|
p, err := m.UploadStatus(context.Background(), id, "user-1")
|
|
if err != nil {
|
|
t.Fatalf("UploadStatus: %v", err)
|
|
}
|
|
return p.Received
|
|
}
|
|
|
|
func TestChunkedUploadAssemblesAndStores(t *testing.T) {
|
|
m, _, fb, id := newChunkedManager(t)
|
|
ctx := context.Background()
|
|
|
|
p, err := m.UploadStatus(ctx, id, "user-1")
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if p != (UploadProgress{Received: 0, PartMaxBytes: 4, MaxContextBytes: 1 << 30}) {
|
|
t.Fatalf("status before any part = %+v", p)
|
|
}
|
|
for _, step := range []struct {
|
|
offset int64
|
|
part string
|
|
want int64
|
|
}{
|
|
{0, "\x1f\x8b\x08\x00", 4},
|
|
{4, "abcd", 8},
|
|
{8, "efgh", 12},
|
|
{12, "ij", 14},
|
|
} {
|
|
if got := sendPart(t, m, id, step.offset, step.part); got != step.want {
|
|
t.Fatalf("after the part at %d: received %d, want %d", step.offset, got, step.want)
|
|
}
|
|
}
|
|
if _, ok := fb.stored[id]; ok {
|
|
t.Fatal("parts reached the blob store before the upload completed")
|
|
}
|
|
|
|
sub, err := m.CompleteUpload(ctx, id, "user-1")
|
|
if err != nil {
|
|
t.Fatalf("CompleteUpload: %v", err)
|
|
}
|
|
if got := string(fb.stored[id]); got != "\x1f\x8b\x08\x00abcdefghij" {
|
|
t.Fatalf("stored %q, want the parts in order", got)
|
|
}
|
|
if sub.ContextSHA256 != "df86a051af3d8615de097952ac01bc094781053bcda61aaac848edb98068209d" {
|
|
t.Fatalf("recorded digest %s", sub.ContextSHA256)
|
|
}
|
|
if got := staged(t, m, id); got != 0 {
|
|
t.Fatalf("staged bytes after completion = %d, want the staged copy gone", got)
|
|
}
|
|
}
|
|
|
|
func TestChunkedUploadResumesWhereTheStagedBytesEnd(t *testing.T) {
|
|
m, _, _, id := newChunkedManager(t)
|
|
ctx := context.Background()
|
|
|
|
var mismatch *OffsetMismatchError
|
|
_, err := m.UploadPart(ctx, id, "user-1", 4, strings.NewReader("abcd"))
|
|
if !errors.As(err, &mismatch) || mismatch.Received != 0 {
|
|
t.Fatalf("a part past an empty upload = %v, want an offset mismatch at 0", err)
|
|
}
|
|
sendPart(t, m, id, 0, "\x1f\x8b\x08\x00")
|
|
_, err = m.UploadPart(ctx, id, "user-1", 8, strings.NewReader("efgh"))
|
|
if !errors.As(err, &mismatch) || mismatch.Received != 4 {
|
|
t.Fatalf("a part that skips ahead = %v, want an offset mismatch at 4", err)
|
|
}
|
|
if got := sendPart(t, m, id, 4, "abcd"); got != 8 {
|
|
t.Fatalf("the part at the staged length: received %d, want 8", got)
|
|
}
|
|
// Offset 0 starts over.
|
|
if got := sendPart(t, m, id, 0, "\x1f\x8b"); got != 2 {
|
|
t.Fatalf("restart at 0: received %d, want 2", got)
|
|
}
|
|
}
|
|
|
|
type failingReader struct {
|
|
data string
|
|
done bool
|
|
}
|
|
|
|
func (f *failingReader) Read(p []byte) (int, error) {
|
|
if f.done {
|
|
return 0, errors.New("connection reset")
|
|
}
|
|
f.done = true
|
|
return copy(p, f.data), nil
|
|
}
|
|
|
|
func TestChunkedUploadCutsBackAPartThatBrokeOff(t *testing.T) {
|
|
m, _, _, id := newChunkedManager(t)
|
|
sendPart(t, m, id, 0, "\x1f\x8b\x08\x00")
|
|
|
|
_, err := m.UploadPart(context.Background(), id, "user-1", 4, &failingReader{data: "ab"})
|
|
if err == nil {
|
|
t.Fatal("a part that broke off was accepted")
|
|
}
|
|
if got := staged(t, m, id); got != 4 {
|
|
t.Fatalf("staged after a broken part = %d, want 4 (the half part cut back off)", got)
|
|
}
|
|
}
|
|
|
|
func TestChunkedUploadFirstPartMustBeGzip(t *testing.T) {
|
|
m, _, _, id := newChunkedManager(t)
|
|
_, err := m.UploadPart(context.Background(), id, "user-1", 0, strings.NewReader("PK\x03\x04"))
|
|
if !errors.Is(err, ErrInvalid) {
|
|
t.Fatalf("a zip first part = %v, want ErrInvalid", err)
|
|
}
|
|
if got := staged(t, m, id); got != 0 {
|
|
t.Fatalf("staged after a refused first part = %d, want 0", got)
|
|
}
|
|
}
|
|
|
|
func TestChunkedUploadPartCap(t *testing.T) {
|
|
m, _, _, id := newChunkedManager(t)
|
|
sendPart(t, m, id, 0, "\x1f\x8b\x08\x00")
|
|
|
|
_, err := m.UploadPart(context.Background(), id, "user-1", 4, strings.NewReader("abcde"))
|
|
if !errors.Is(err, ErrPartTooLarge) {
|
|
t.Fatalf("a 5-byte part over a 4-byte cap = %v, want ErrPartTooLarge", err)
|
|
}
|
|
if got := staged(t, m, id); got != 4 {
|
|
t.Fatalf("staged after an oversize part = %d, want 4", got)
|
|
}
|
|
}
|
|
|
|
// The staged total meets the same cap as a whole context, with the same error.
|
|
func TestChunkedUploadTotalCappedLikeAWholeContext(t *testing.T) {
|
|
m, _, _, id := newChunkedManager(t)
|
|
m.MaxContextBytes = 10
|
|
sendPart(t, m, id, 0, "\x1f\x8b\x08\x00")
|
|
sendPart(t, m, id, 4, "abcd")
|
|
|
|
_, err := m.UploadPart(context.Background(), id, "user-1", 8, strings.NewReader("efg"))
|
|
if !errors.Is(err, ErrInvalid) || errors.Is(err, ErrPartTooLarge) {
|
|
t.Fatalf("a part past the context cap = %v, want the context-too-large ErrInvalid", err)
|
|
}
|
|
if got := sendPart(t, m, id, 8, "ef"); got != 10 {
|
|
t.Fatalf("a part ending exactly at the cap: received %d, want 10", got)
|
|
}
|
|
}
|
|
|
|
// Staged bytes count against the storage budget, so parts parked on several
|
|
// pending submissions cannot hold more than stored contexts could.
|
|
func TestChunkedUploadStagedBytesCountAgainstTheBudget(t *testing.T) {
|
|
m, _, fb, a := newChunkedManager(t)
|
|
ctx := context.Background()
|
|
m.MaxStoredBytesPerUser = 10
|
|
m.PartMaxBytes = 16
|
|
sendPart(t, m, a, 0, "\x1f\x8b\x08\x00abcd")
|
|
|
|
b, err := m.Create(ctx, CreateRequest{DisplayName: "B", SubmittedBy: "user-1"})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if _, err := m.UploadContext(ctx, b.ID, "user-1", strings.NewReader("\x1f\x8b\x08")); !errors.Is(err, ErrQuotaExceeded) {
|
|
t.Fatalf("3 bytes beside 8 staged under a 10-byte budget = %v, want ErrQuotaExceeded", err)
|
|
}
|
|
if _, err := m.UploadContext(ctx, b.ID, "user-1", strings.NewReader("\x1f\x8b")); err != nil {
|
|
t.Fatalf("2 bytes beside 8 staged = %v, want accepted", err)
|
|
}
|
|
// And the other way round: B's stored 2 bytes bound what A may still stage.
|
|
if _, err := m.UploadPart(ctx, a, "user-1", 8, strings.NewReader("e")); !errors.Is(err, ErrQuotaExceeded) {
|
|
t.Fatalf("staging past the budget = %v, want ErrQuotaExceeded", err)
|
|
}
|
|
if _, ok := fb.stored[a]; ok {
|
|
t.Fatal("staging stored a blob")
|
|
}
|
|
}
|
|
|
|
func TestChunkedUploadOwnerAndStatus(t *testing.T) {
|
|
m, st, _, id := newChunkedManager(t)
|
|
ctx := context.Background()
|
|
sendPart(t, m, id, 0, "\x1f\x8b\x08\x00")
|
|
|
|
if _, err := m.UploadStatus(ctx, id, "user-2"); !errors.Is(err, ErrNotFound) {
|
|
t.Fatalf("another user's status = %v, want ErrNotFound", err)
|
|
}
|
|
if _, err := m.UploadPart(ctx, id, "user-2", 4, strings.NewReader("abcd")); !errors.Is(err, ErrNotFound) {
|
|
t.Fatalf("another user's part = %v, want ErrNotFound", err)
|
|
}
|
|
if _, err := m.CompleteUpload(ctx, id, "user-2"); !errors.Is(err, ErrNotFound) {
|
|
t.Fatalf("another user's completion = %v, want ErrNotFound", err)
|
|
}
|
|
st.subs[id].Status = StatusRejected
|
|
if _, err := m.UploadPart(ctx, id, "user-1", 4, strings.NewReader("abcd")); !errors.Is(err, ErrAlreadyReviewed) {
|
|
t.Fatalf("a part for a reviewed submission = %v, want ErrAlreadyReviewed", err)
|
|
}
|
|
if _, err := m.CompleteUpload(ctx, id, "user-1"); !errors.Is(err, ErrAlreadyReviewed) {
|
|
t.Fatalf("completing a reviewed submission = %v, want ErrAlreadyReviewed", err)
|
|
}
|
|
}
|
|
|
|
func TestChunkedUploadWithoutPartStoreIsUnavailable(t *testing.T) {
|
|
m, _, _, id := newChunkedManager(t)
|
|
m.Parts = nil
|
|
if _, err := m.UploadStatus(context.Background(), id, "user-1"); !errors.Is(err, ErrUploadsUnavailable) {
|
|
t.Fatalf("status without a part store = %v, want ErrUploadsUnavailable", err)
|
|
}
|
|
if _, err := m.UploadPart(context.Background(), id, "user-1", 0, strings.NewReader("\x1f\x8b")); !errors.Is(err, ErrUploadsUnavailable) {
|
|
t.Fatalf("part without a part store = %v, want ErrUploadsUnavailable", err)
|
|
}
|
|
}
|
|
|
|
func TestCompleteUploadWithNothingStagedIsInvalid(t *testing.T) {
|
|
m, _, fb, id := newChunkedManager(t)
|
|
if _, err := m.CompleteUpload(context.Background(), id, "user-1"); !errors.Is(err, ErrInvalid) {
|
|
t.Fatalf("completing an empty upload = %v, want ErrInvalid", err)
|
|
}
|
|
if len(fb.stored) != 0 {
|
|
t.Fatal("an empty completion stored a blob")
|
|
}
|
|
}
|
|
|
|
func TestCompleteUploadFailureKeepsTheStagedBytes(t *testing.T) {
|
|
m, _, fb, id := newChunkedManager(t)
|
|
sendPart(t, m, id, 0, "\x1f\x8b\x08\x00")
|
|
sendPart(t, m, id, 4, "ab")
|
|
fb.putErr = errors.New("object store unreachable")
|
|
|
|
if _, err := m.CompleteUpload(context.Background(), id, "user-1"); err == nil {
|
|
t.Fatal("completion succeeded with the blob store down")
|
|
}
|
|
if got := staged(t, m, id); got != 6 {
|
|
t.Fatalf("staged after a failed completion = %d, want 6 kept for the retry", got)
|
|
}
|
|
fb.putErr = nil
|
|
if _, err := m.CompleteUpload(context.Background(), id, "user-1"); err != nil {
|
|
t.Fatalf("retried completion: %v", err)
|
|
}
|
|
if got := string(fb.stored[id]); got != "\x1f\x8b\x08\x00ab" {
|
|
t.Fatalf("stored %q after the retry", got)
|
|
}
|
|
}
|
|
|
|
func TestChunkedUploadOneRequestAtATime(t *testing.T) {
|
|
m, _, _, id := newChunkedManager(t)
|
|
sendPart(t, m, id, 0, "\x1f\x8b\x08\x00")
|
|
release, err := m.Parts.hold(id)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if _, err := m.UploadPart(context.Background(), id, "user-1", 4, strings.NewReader("ab")); !errors.Is(err, ErrUploadBusy) {
|
|
t.Fatalf("a part while another is writing = %v, want ErrUploadBusy", err)
|
|
}
|
|
if _, err := m.CompleteUpload(context.Background(), id, "user-1"); !errors.Is(err, ErrUploadBusy) {
|
|
t.Fatalf("completing while a part is writing = %v, want ErrUploadBusy", err)
|
|
}
|
|
release()
|
|
if got := sendPart(t, m, id, 4, "ab"); got != 6 {
|
|
t.Fatalf("the part after release: received %d, want 6", got)
|
|
}
|
|
}
|
|
|
|
func TestWithdrawDeletesTheStagedUpload(t *testing.T) {
|
|
m, _, _, id := newChunkedManager(t)
|
|
sendPart(t, m, id, 0, "\x1f\x8b\x08\x00")
|
|
if _, err := m.Withdraw(context.Background(), id, "user-1"); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if _, err := os.Stat(filepath.Join(m.Parts.Dir, id+".part")); !os.IsNotExist(err) {
|
|
t.Fatalf("staged upload after withdraw: stat err = %v, want it gone", err)
|
|
}
|
|
}
|
|
|
|
func TestReapStaleParts(t *testing.T) {
|
|
m, _, _, oldID := newChunkedManager(t)
|
|
ctx := context.Background()
|
|
fresh, err := m.Create(ctx, CreateRequest{DisplayName: "Fresh", SubmittedBy: "user-1"})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
sendPart(t, m, oldID, 0, "\x1f\x8b\x08\x00")
|
|
sendPart(t, m, fresh.ID, 0, "\x1f\x8b\x08\x00")
|
|
if err := os.Chtimes(filepath.Join(m.Parts.Dir, oldID+".part"), testNow.Add(-25*time.Hour), testNow.Add(-25*time.Hour)); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if err := os.Chtimes(filepath.Join(m.Parts.Dir, fresh.ID+".part"), testNow.Add(-23*time.Hour), testNow.Add(-23*time.Hour)); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
// A stray file that is not a staged upload is left alone.
|
|
if err := os.WriteFile(filepath.Join(m.Parts.Dir, "notes"), []byte("x"), 0o600); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
os.Chtimes(filepath.Join(m.Parts.Dir, "notes"), testNow.Add(-48*time.Hour), testNow.Add(-48*time.Hour))
|
|
|
|
n, err := m.ReapStaleParts(StalePartRetention)
|
|
if err != nil || n != 1 {
|
|
t.Fatalf("ReapStaleParts = %d, %v; want 1", n, err)
|
|
}
|
|
if got := staged(t, m, oldID); got != 0 {
|
|
t.Fatalf("the 25-hour-old upload still holds %d bytes", got)
|
|
}
|
|
if got := staged(t, m, fresh.ID); got != 4 {
|
|
t.Fatalf("the 23-hour-old upload holds %d bytes, want 4 kept", got)
|
|
}
|
|
if _, err := os.Stat(filepath.Join(m.Parts.Dir, "notes")); err != nil {
|
|
t.Fatalf("the reaper touched a file that is not a staged upload: %v", err)
|
|
}
|
|
}
|