Updated with retries

Author: javierdfm

channel_batch_updater_test.go

@@ -62,15 +62,11 @@ func TestChannelBatchUpdater_AddMembers(t *testing.T) {
 	require.NoError(t, err)
 	require.NotEmpty(t, resp.TaskID)
 
-	// Give the task a moment to be created and allow for rate limit retries
-	// Rate limit errors typically retry after ~500ms, so wait a bit longer
 	time.Sleep(2 * time.Second)
 
-	// Poll for task completion (up to 2 minutes)
 	for i := 0; i < 120; i++ {
 		task, err := c.GetTask(ctx, resp.TaskID)
 		if err != nil {
-			// If task doesn't exist yet, wait and retry
 			if i < 10 {
 				time.Sleep(time.Second)
 				continue
@@ -79,20 +75,15 @@ func TestChannelBatchUpdater_AddMembers(t *testing.T) {
 		}
 		require.Equal(t, resp.TaskID, task.TaskID)
 
-		// If task is waiting/pending/running, it might be rate-limited and will retry
-		// Rate limit errors cause Asynq to retry automatically after the suggested delay
 		if task.Status == TaskStatusWaiting || task.Status == TaskStatusPending || task.Status == TaskStatusRunning {
 			time.Sleep(time.Second)
 			continue
 		}
 
 		if task.Status == TaskStatusCompleted {
-			// Wait up to 2 minutes for background workers to finish processing
-			// Check periodically if changes are visible
 			for j := 0; j < 120; j++ {
 				time.Sleep(time.Second)
 
-				// Try to verify members were added by refreshing channel state
 				err = ch1.refresh(ctx)
 				if err != nil {
 					continue
@@ -102,7 +93,6 @@ func TestChannelBatchUpdater_AddMembers(t *testing.T) {
 					continue
 				}
 
-				// Check that added users are now members
 				ch1MemberIDs := make([]string, len(ch1.Members))
 				for i, m := range ch1.Members {
 					ch1MemberIDs[i] = m.UserID
@@ -121,18 +111,15 @@ func TestChannelBatchUpdater_AddMembers(t *testing.T) {
 			t.Fatal("changes not visible after 2 minutes")
 		}
 		if task.Status == TaskStatusFailed {
-			// Check if this is a rate limit error - Asynq will automatically retry these
 			if len(task.Result) > 0 {
 				if desc, ok := task.Result["description"].(string); ok {
 					if strings.Contains(strings.ToLower(desc), "rate limit") {
-						time.Sleep(2 * time.Second) // Wait for retry
+						time.Sleep(2 * time.Second)
 						continue
 					}
 				}
 				t.Fatalf("task failed with result: %v", task.Result)
 			}
-			// If result is empty, the task might have failed during creation
-			// Check the response for error details
 			t.Fatalf("task failed (status: %s, result: %v)", task.Status, task.Result)
 		}
 
@@ -145,12 +132,10 @@ func TestChannelBatchUpdater_RemoveMembers(t *testing.T) {
 	c := initClient(t)
 	ctx := context.Background()
 
-	// Create channels with members
 	membersID := randomUsersID(t, c, 2)
 	ch1 := initChannel(t, c, membersID...)
 	ch2 := initChannel(t, c, membersID...)
 
-	// Verify channels have 2 members before executing the task
 	err := ch1.refresh(ctx)
 	require.NoError(t, err, "failed to refresh channel 1")
 	require.Len(t, ch1.Members, 2, "channel 1 should have 2 members before removal")
@@ -159,7 +144,6 @@ func TestChannelBatchUpdater_RemoveMembers(t *testing.T) {
 	require.NoError(t, err, "failed to refresh channel 2")
 	require.Len(t, ch2.Members, 2, "channel 2 should have 2 members before removal")
 
-	// Verify both channels have the same members
 	ch1MemberIDs := make([]string, len(ch1.Members))
 	for i, m := range ch1.Members {
 		ch1MemberIDs[i] = m.UserID
@@ -173,7 +157,6 @@ func TestChannelBatchUpdater_RemoveMembers(t *testing.T) {
 
 	updater := c.ChannelBatchUpdater()
 
-	// Remove one member from both channels
 	memberToRemove := membersID[0]
 	resp, err := updater.RemoveMembers(ctx, ChannelsBatchFilters{
 		CIDs: map[string]interface{}{
@@ -183,30 +166,29 @@ func TestChannelBatchUpdater_RemoveMembers(t *testing.T) {
 	require.NoError(t, err)
 	require.NotEmpty(t, resp.TaskID)
 
-	// Give the task a moment to be created
-	time.Sleep(500 * time.Millisecond)
+	time.Sleep(2 * time.Second)
 
-	// Poll for task completion (up to 2 minutes)
 	for i := 0; i < 120; i++ {
 		task, err := c.GetTask(ctx, resp.TaskID)
 		if err != nil {
-			// If task doesn't exist yet, wait and retry
-			if i < 5 {
+			if i < 10 {
 				time.Sleep(time.Second)
 				continue
 			}
 			require.NoError(t, err, "failed to get task status")
 		}
 		require.Equal(t, resp.TaskID, task.TaskID)
+		
+		if task.Status == TaskStatusWaiting || task.Status == TaskStatusPending || task.Status == TaskStatusRunning {
+			time.Sleep(time.Second)
+			continue
+		}
 
 		if task.Status == TaskStatusCompleted {
-			// Wait up to 2 minutes for background workers to finish processing
-			// Check periodically if changes are visible
 			var ch1MemberIDs []string
 			for j := 0; j < 120; j++ {
 				time.Sleep(time.Second)
 
-				// Try to verify member was removed
 				err = ch1.refresh(ctx)
 				if err != nil {
 					continue
@@ -223,12 +205,15 @@ func TestChannelBatchUpdater_RemoveMembers(t *testing.T) {
 			t.Fatalf("changes not visible after 2 minutes. Channel 1 still has members: %v", ch1MemberIDs)
 		}
 		if task.Status == TaskStatusFailed {
-			// Check if there's error information in the result
 			if len(task.Result) > 0 {
+				if desc, ok := task.Result["description"].(string); ok {
+					if strings.Contains(strings.ToLower(desc), "rate limit") {
+						time.Sleep(2 * time.Second)
+						continue
+					}
+				}
 				t.Fatalf("task failed with result: %v", task.Result)
 			}
-			// If result is empty, the task might have failed during creation
-			// Check the response for error details
 			t.Fatalf("task failed (status: %s, result: %v)", task.Status, task.Result)
 		}
 
@@ -247,7 +232,6 @@ func TestChannelBatchUpdater_Archive(t *testing.T) {
 
 	updater := c.ChannelBatchUpdater()
 
-	// Archive channels for the first member
 	resp, err := updater.Archive(ctx, ChannelsBatchFilters{
 		CIDs: map[string]interface{}{
 			"$in": []string{ch1.CID, ch2.CID},
@@ -256,35 +240,33 @@ func TestChannelBatchUpdater_Archive(t *testing.T) {
 	require.NoError(t, err)
 	require.NotEmpty(t, resp.TaskID)
 
-	// Give the task a moment to be created
-	time.Sleep(500 * time.Millisecond)
+	time.Sleep(2 * time.Second)
 
-	// Poll for task completion (up to 2 minutes)
 	for i := 0; i < 120; i++ {
 		task, err := c.GetTask(ctx, resp.TaskID)
 		if err != nil {
-			// If task doesn't exist yet, wait and retry
-			if i < 5 {
+			if i < 10 {
 				time.Sleep(time.Second)
 				continue
 			}
 			require.NoError(t, err, "failed to get task status")
 		}
 		require.Equal(t, resp.TaskID, task.TaskID)
+		
+		if task.Status == TaskStatusWaiting || task.Status == TaskStatusPending || task.Status == TaskStatusRunning {
+			time.Sleep(time.Second)
+			continue
+		}
 
 		if task.Status == TaskStatusCompleted {
-			// Wait up to 2 minutes for background workers to finish processing
-			// Check periodically if changes are visible
 			for j := 0; j < 120; j++ {
 				time.Sleep(time.Second)
 
-				// Try to verify channel was archived for the user
 				err = ch1.refresh(ctx)
 				if err != nil {
 					continue
 				}
 
-				// Find the member and check if archived
 				for _, m := range ch1.Members {
 					if m.UserID == membersID[0] {
 						if m.ArchivedAt != nil {
@@ -297,12 +279,15 @@ func TestChannelBatchUpdater_Archive(t *testing.T) {
 			t.Fatal("changes not visible after 2 minutes")
 		}
 		if task.Status == TaskStatusFailed {
-			// Check if there's error information in the result
 			if len(task.Result) > 0 {
+				if desc, ok := task.Result["description"].(string); ok {
+					if strings.Contains(strings.ToLower(desc), "rate limit") {
+						time.Sleep(2 * time.Second)
+						continue
+					}
+				}
 				t.Fatalf("task failed with result: %v", task.Result)
 			}
-			// If result is empty, the task might have failed during creation
-			// Check the response for error details
 			t.Fatalf("task failed (status: %s, result: %v)", task.Status, task.Result)
 		}