Fixing rangeDelete for QueueV2

temporalio · Oct 27, 2023 · a74929b · a74929b
1 parent 1b0ec47
commit a74929b
Show file tree

Hide file tree

Showing 2 changed files with 59 additions and 5 deletions.
diff --git a/common/persistence/sql/queue_v2.go b/common/persistence/sql/queue_v2.go
@@ -259,11 +259,21 @@ func (q *queueV2) RangeDeleteMessages(
 		}
 		partition, err := persistence.GetPartitionForQueueV2(request.QueueType, request.QueueName, qm.Metadata)
 		if err != nil {
-			return err
+			return serviceerror.NewUnavailable(fmt.Sprintf(
+				"RangeDeleteMessages failed for queue with type: %v and name: %v. GetPartitionForQueueV2 operation failed. Error: %v",
+				request.QueueType,
+				request.QueueName,
+				err),
+			)
 		}
 		maxMessageID, ok, err := q.getMaxMessageID(ctx, request.QueueType, request.QueueName, tx)
 		if err != nil {
-			return err
+			return serviceerror.NewUnavailable(fmt.Sprintf(
+				"RangeDeleteMessages failed for queue with type: %v and name: %v. failed to get MaxMessageID. Error: %v",
+				request.QueueType,
+				request.QueueName,
+				err),
+			)
 		}
 		if !ok {
 			return nil
@@ -287,7 +297,12 @@ func (q *queueV2) RangeDeleteMessages(
 		}
 		_, err = tx.RangeDeleteFromQueueV2Messages(ctx, msgFilter)
 		if err != nil {
-			return err
+			return serviceerror.NewUnavailable(fmt.Sprintf(
+				"RangeDeleteMessages failed for queue with type: %v and name: %v. RangeDeleteFromQueueV2Messages operation failed. Error: %v",
+				request.QueueType,
+				request.QueueName,
+				err),
+			)
 		}
 		partition.MinMessageId = deleteRange.NewMinMessageID
 		bytes, _ := qm.Metadata.Marshal()
@@ -296,11 +311,16 @@ func (q *queueV2) RangeDeleteMessages(
 			QueueName:        request.QueueName,
 			MetadataPayload:  bytes,
 			MetadataEncoding: enums.ENCODING_TYPE_PROTO3.String(),
-			Version:          0,
+			Version:          qm.Version,
 		}
 		_, err = tx.UpdateQueueV2Metadata(ctx, &row)
 		if err != nil {
-			return err
+			return serviceerror.NewUnavailable(fmt.Sprintf(
+				"RangeDeleteMessages failed for queue with type: %v and name: %v. UpdateQueueV2Metadata operation failed. Error: %v",
+				request.QueueType,
+				request.QueueName,
+				err),
+			)
 		}
 		return nil
 	})

diff --git a/common/persistence/tests/queue_v2_test_suite.go b/common/persistence/tests/queue_v2_test_suite.go
@@ -383,6 +383,40 @@ func testRangeDeleteMessages(ctx context.Context, t *testing.T, queue persistenc
 		require.Len(t, response.Messages, 1)
 		assert.Equal(t, int64(persistence.FirstQueueMessageID+1), response.Messages[0].MetaData.ID)
 	})
+
+	t.Run("RangeDeleteTwice", func(t *testing.T) {
+		t.Parallel()
+
+		queueType := persistence.QueueTypeHistoryNormal
+		queueName := "test-queue-" + t.Name()
+		_, err := queue.CreateQueue(ctx, &persistence.InternalCreateQueueRequest{
+			QueueType: queueType,
+			QueueName: queueName,
+		})
+		require.NoError(t, err)
+		for i := 0; i < 2; i++ {
+			_, err := persistencetest.EnqueueMessage(ctx, queue, queueType, queueName)
+			require.NoError(t, err)
+		}
+
+		for i := 0; i < 2; i++ {
+			_, err = queue.RangeDeleteMessages(ctx, &persistence.InternalRangeDeleteMessagesRequest{
+				QueueType: queueType,
+				QueueName: queueName,
+				InclusiveMaxMessageMetadata: persistence.MessageMetadata{
+					ID: persistence.FirstQueueMessageID + int64(i),
+				},
+			})
+			require.NoError(t, err)
+		}
+		response, err := queue.ReadMessages(ctx, &persistence.InternalReadMessagesRequest{
+			QueueType: queueType,
+			QueueName: queueName,
+			PageSize:  10,
+		})
+		require.NoError(t, err)
+		require.Len(t, response.Messages, 0)
+	})
 }
 
 func RunQueueV2TestSuiteForSQL(t *testing.T, factory *sql.Factory) {