Update RDB.DeleteTask to handle aggregating task

2025-10-23 10:16:12 +08:00 · 2022-03-27 16:25:03 -07:00
parent 4a86d42fd7
commit edca226858
2 changed files with 109 additions and 3 deletions
--- a/internal/rdb/inspect_test.go
+++ b/internal/rdb/inspect_test.go
@@ -3671,6 +3671,101 @@ func TestDeleteScheduledTask(t *testing.T) {
 	}
 }

+func TestDeleteAggregatingTask(t *testing.T) {
+	r := setup(t)
+	defer r.Close()
+	now := time.Now()
+	m1 := h.NewTaskMessageBuilder().SetQueue("default").SetType("task1").SetGroup("group1").Build()
+	m2 := h.NewTaskMessageBuilder().SetQueue("default").SetType("task2").SetGroup("group1").Build()
+	m3 := h.NewTaskMessageBuilder().SetQueue("custom").SetType("task3").SetGroup("group1").Build()
+
+	fxt := struct {
+		tasks     []*h.TaskSeedData
+		allQueues []string
+		allGroups map[string][]string
+		groups    map[string][]*redis.Z
+	}{
+		tasks: []*h.TaskSeedData{
+			{Msg: m1, State: base.TaskStateAggregating},
+			{Msg: m2, State: base.TaskStateAggregating},
+			{Msg: m3, State: base.TaskStateAggregating},
+		},
+		allQueues: []string{"default", "custom"},
+		allGroups: map[string][]string{
+			base.AllGroups("default"): {"group1"},
+			base.AllGroups("custom"):  {"group1"},
+		},
+		groups: map[string][]*redis.Z{
+			base.GroupKey("default", "group1"): {
+				{Member: m1.ID, Score: float64(now.Add(-20 * time.Second).Unix())},
+				{Member: m2.ID, Score: float64(now.Add(-25 * time.Second).Unix())},
+			},
+			base.GroupKey("custom", "group1"): {
+				{Member: m3.ID, Score: float64(now.Add(-20 * time.Second).Unix())},
+			},
+		},
+	}
+
+	tests := []struct {
+		desc          string
+		qname         string
+		id            string
+		wantAllGroups map[string][]string
+		wantGroups    map[string][]redis.Z
+	}{
+		{
+			desc:  "deletes a task from group with multiple tasks",
+			qname: "default",
+			id:    m1.ID,
+			wantAllGroups: map[string][]string{
+				base.AllGroups("default"): {"group1"},
+				base.AllGroups("custom"):  {"group1"},
+			},
+			wantGroups: map[string][]redis.Z{
+				base.GroupKey("default", "group1"): {
+					{Member: m2.ID, Score: float64(now.Add(-25 * time.Second).Unix())},
+				},
+				base.GroupKey("custom", "group1"): {
+					{Member: m3.ID, Score: float64(now.Add(-20 * time.Second).Unix())},
+				},
+			},
+		},
+		{
+			desc:  "deletes a task from group with single task",
+			qname: "custom",
+			id:    m3.ID,
+			wantAllGroups: map[string][]string{
+				base.AllGroups("default"): {"group1"},
+				base.AllGroups("custom"):  {}, // should be clear out group from all-groups set
+			},
+			wantGroups: map[string][]redis.Z{
+				base.GroupKey("default", "group1"): {
+					{Member: m1.ID, Score: float64(now.Add(-20 * time.Second).Unix())},
+					{Member: m2.ID, Score: float64(now.Add(-25 * time.Second).Unix())},
+				},
+				base.GroupKey("custom", "group1"): {},
+			},
+		},
+	}
+
+	for _, tc := range tests {
+		h.FlushDB(t, r.client)
+		h.SeedTasks(t, r.client, fxt.tasks)
+		h.SeedRedisSet(t, r.client, base.AllQueues, fxt.allQueues)
+		h.SeedRedisSets(t, r.client, fxt.allGroups)
+		h.SeedRedisZSets(t, r.client, fxt.groups)
+
+		t.Run(tc.desc, func(t *testing.T) {
+			err := r.DeleteTask(tc.qname, tc.id)
+			if err != nil {
+				t.Fatalf("DeleteTask returned error: %v", err)
+			}
+			h.AssertRedisSets(t, r.client, tc.wantAllGroups)
+			h.AssertRedisZSets(t, r.client, tc.wantGroups)
+		})
+	}
+}
+
 func TestDeletePendingTask(t *testing.T) {
 	r := setup(t)
 	defer r.Close()