Merge pull request #893 from SiaFoundation/pj/upload-regression

Add regression test for the upload deadlock

stores/metrics.go

@@ -537,7 +537,7 @@ func (s *SQLStore) findAggregatedContractPeriods(start time.Time, n uint64, inte
 		SELECT contracts.*, i.Period FROM contracts
 		INNER JOIN (
 		SELECT
-			p.period_start as Period, 
+			p.period_start as Period,
 			MIN(c.id) AS id
 		FROM
 			periods p

worker/mocks_test.go

@@ -48,7 +48,9 @@ type (
 	}
 
 	mockMemory        struct{}
-	mockMemoryManager struct{}
+	mockMemoryManager struct {
+		memBlockChan chan struct{}
+	}
 
 	mockObjectStore struct {
 		mu           sync.Mutex
@@ -110,6 +112,9 @@ func (mm *mockMemoryManager) Limit(amt uint64) (MemoryManager, error) {
 }
 func (mm *mockMemoryManager) Status() api.MemoryStatus { return api.MemoryStatus{} }
 func (mm *mockMemoryManager) AcquireMemory(ctx context.Context, amt uint64) Memory {
+	if mm.memBlockChan != nil {
+		<-mm.memBlockChan
+	}
 	return &mockMemory{}
 }
 

worker/upload_test.go

@@ -340,6 +340,61 @@ func TestMigrateShards(t *testing.T) {
 	}
 }
 
+func TestUploadRegression(t *testing.T) {
+	// mock worker
+	w := newMockWorker(testRedundancySettings.TotalShards * 2)
+
+	// convenience variables
+	ul := w.ul
+	dl := w.dl
+	mm := w.mm
+	os := w.os
+
+	// create test data
+	data := make([]byte, 128)
+	if _, err := frand.Read(data); err != nil {
+		t.Fatal(err)
+	}
+
+	// create upload params
+	params := testParameters(t.Name())
+
+	// make sure the memory manager blocks
+	mm.memBlockChan = make(chan struct{})
+
+	// upload data
+	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
+	defer cancel()
+	_, _, err := ul.Upload(ctx, bytes.NewReader(data), w.contracts.values(), params, lockingPriorityUpload)
+	if !errors.Is(err, errUploadInterrupted) {
+		t.Fatal(err)
+	}
+
+	// unblock the memory manager
+	close(mm.memBlockChan)
+
+	// upload data
+	_, _, err = ul.Upload(context.Background(), bytes.NewReader(data), w.contracts.values(), params, lockingPriorityUpload)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	// grab the object
+	o, err := os.Object(context.Background(), testBucket, t.Name(), api.GetObjectOptions{})
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	// download data for good measure
+	var buf bytes.Buffer
+	err = dl.DownloadObject(context.Background(), &buf, o.Object.Object, 0, uint64(o.Object.Size), w.contracts.values())
+	if err != nil {
+		t.Fatal(err)
+	} else if !bytes.Equal(data, buf.Bytes()) {
+		t.Fatal("data mismatch", data, buf.Bytes())
+	}
+}
+
 func testParameters(path string) uploadParameters {
 	return uploadParameters{
 		bucket: testBucket,