test: remove ReplaceStreamFiles

Author: adamdecaf

internal/pipeline/file_receiver.go

@@ -119,24 +119,6 @@ func (fr *FileReceiver) reconnect() error {
 	return nil
 }
 
-func (fr *FileReceiver) ReplaceStreamFiles(ctx context.Context, sub stream.Subscription) {
-	fr.mu.Lock()
-	defer fr.mu.Unlock()
-
-	fr.logger.Info().Log("replacing stream subscription")
-
-	// Shut down receiving messages
-	fr.cancel()
-
-	// Close an existing stream subscription
-	if fr.streamFiles != nil {
-		fr.streamFiles.Shutdown(context.Background())
-	}
-	fr.streamFiles = sub
-
-	go fr.Start(ctx)
-}
-
 func (fr *FileReceiver) Start(ctx context.Context) {
 	for {
 		// Create a context that will be shutdown by its parent or after a read iteration

internal/test/upload_test.go

@@ -171,15 +171,9 @@ func TestUploads(t *testing.T) {
 	adminServer := admintest.Server(t)
 	fileReceiver.RegisterAdminRoutes(adminServer)
 
-	// Force the stream subscription to fail
-	flakeySub := streamtest.FailingSubscription(errors.New("write: broken pipe"))
-	fileReceiver.ReplaceStreamFiles(context.Background(), flakeySub)
-	require.Contains(t, fmt.Sprintf("%#v", fileReceiver), "streamFiles:(*streamtest.FailedSubscription)")
-
 	// Upload our files
 	createdEntries := 0
 	canceledEntries := 0
-	erroredSubscriptions := 0
 	var createdFileIDs, canceledFileIDs []string
 
 	iterations := 500
@@ -205,13 +199,6 @@ func TestUploads(t *testing.T) {
 		} else {
 			createdFileIDs = append(createdFileIDs, fileID)
 		}
-
-		// Force the subscription to fail sometimes
-		if err := causeSubscriptionFailure(t); err != nil {
-			flakeySub := streamtest.FailingSubscription(err)
-			fileReceiver.ReplaceStreamFiles(context.Background(), flakeySub)
-			erroredSubscriptions += 1
-		}
 	}
 	require.NoError(t, g.Wait())
 
@@ -274,7 +261,7 @@ func TestUploads(t *testing.T) {
 		expected := createdEntries - canceledEntries
 		found := countAllEntries(uploadedFiles)
 
-		t.Logf("found %d entries of %d expected (%d canceled) (%d errored) from %d uploaded files", found, expected, canceledEntries, erroredSubscriptions, len(uploadedFiles))
+		t.Logf("found %d entries of %d expected (%d canceled) from %d uploaded files", found, expected, canceledEntries, len(uploadedFiles))
 
 		return expected == found
 	}, wait, tick)
@@ -463,17 +450,6 @@ var subscriptionFailures = []error{
 	errors.New("contains: pubsub error"),
 }
 
-func causeSubscriptionFailure(t *testing.T) error {
-	t.Helper()
-
-	n := randomInt(t, 100)
-	if n <= 5 { // 5%
-		idx := (len(subscriptionFailures) - 1) % (int(n) + 1)
-		return subscriptionFailures[idx]
-	}
-	return nil
-}
-
 func firstDirectory(t *testing.T, fsys fs.FS, prefix string) string {
 	t.Helper()