Implement HistoryTaskQueueManager.DeleteTasks

.golangci.yml

@@ -105,7 +105,7 @@ issues:
       text: "(cyclomatic|cognitive)"
       linters:
         - revive
-    - path: _test\.go
+    - path: _test\.go|^common/persistence\/tests\/.+\.go
       linters:
         - goerr113
     - path: ^tools\/.+\.go

common/persistence/data_interfaces.go

@@ -1219,6 +1219,7 @@ type (
 		ReadTasks(ctx context.Context, request *ReadTasksRequest) (*ReadTasksResponse, error)
 		// CreateQueue must return an ErrQueueAlreadyExists if the queue already exists.
 		CreateQueue(ctx context.Context, request *CreateQueueRequest) (*CreateQueueResponse, error)
+		DeleteTasks(ctx context.Context, request *DeleteTasksRequest) (*DeleteTasksResponse, error)
 	}
 
 	HistoryTaskQueueManagerImpl struct {
@@ -1283,6 +1284,15 @@ type (
 
 	CreateQueueResponse struct {
 	}
+
+	DeleteTasksRequest struct {
+		QueueKey                    QueueKey
+		InclusiveMaxMessageMetadata MessageMetadata
+	}
+
+	DeleteTasksResponse struct {
+		// empty
+	}
 )
 
 func (e *InvalidPersistenceRequestError) Error() string {

common/persistence/history_task_queue_manager.go

@@ -188,7 +188,10 @@ func (m *HistoryTaskQueueManagerImpl) ReadTasks(ctx context.Context, request *Re
 	}, nil
 }
 
-func (m *HistoryTaskQueueManagerImpl) CreateQueue(ctx context.Context, request *CreateQueueRequest) (*CreateQueueResponse, error) {
+func (m *HistoryTaskQueueManagerImpl) CreateQueue(
+	ctx context.Context,
+	request *CreateQueueRequest,
+) (*CreateQueueResponse, error) {
 	_, err := m.queue.CreateQueue(ctx, &InternalCreateQueueRequest{
 		QueueType: request.QueueKey.QueueType,
 		QueueName: request.QueueKey.GetQueueName(),
@@ -199,6 +202,20 @@ func (m *HistoryTaskQueueManagerImpl) CreateQueue(ctx context.Context, request *
 	return &CreateQueueResponse{}, nil
 }
 
+func (m *HistoryTaskQueueManagerImpl) DeleteTasks(
+	ctx context.Context,
+	request *DeleteTasksRequest,
+) (*DeleteTasksResponse, error) {
+	_, err := m.queue.RangeDeleteMessages(ctx, &InternalRangeDeleteMessagesRequest{
+		QueueType: request.QueueKey.QueueType,
+		QueueName: request.QueueKey.GetQueueName(),
+	})
+	if err != nil {
+		return nil, err
+	}
+	return &DeleteTasksResponse{}, nil
+}
+
 // combineUnique combines the given strings into a single string by hashing the length of each string and the string
 // itself. This is used to generate a unique suffix for the queue name.
 func combineUnique(strs ...string) string {

common/persistence/history_task_queue_manager_test.go

@@ -167,65 +167,3 @@ func TestHistoryTaskQueueManager_ReadTasks_NonPositivePageSize(t *testing.T) {
 			"ErrReadTasksNonPositivePageSize when the request's page size is: "+strconv.Itoa(pageSize))
 	}
 }
-
-// failingQueue is a QueueV2 implementation that always returns an error.
-type failingQueue struct{}
-
-func (q failingQueue) EnqueueMessage(
-	context.Context,
-	*persistence.InternalEnqueueMessageRequest,
-) (*persistence.InternalEnqueueMessageResponse, error) {
-	return nil, assert.AnError
-}
-
-func (q failingQueue) ReadMessages(
-	context.Context,
-	*persistence.InternalReadMessagesRequest,
-) (*persistence.InternalReadMessagesResponse, error) {
-	return nil, assert.AnError
-}
-
-func (q failingQueue) CreateQueue(
-	context.Context,
-	*persistence.InternalCreateQueueRequest,
-) (*persistence.InternalCreateQueueResponse, error) {
-	return nil, assert.AnError
-}
-
-func (q failingQueue) RangeDeleteMessages(
-	context.Context,
-	*persistence.InternalRangeDeleteMessagesRequest,
-) (*persistence.InternalRangeDeleteMessagesResponse, error) {
-	return nil, assert.AnError
-}
-
-func TestHistoryTaskQueueManager_ReadTasks_ErrReadQueueMessages(t *testing.T) {
-	t.Parallel()
-
-	m := persistence.NewHistoryTaskQueueManager(failingQueue{}, 1)
-	_, err := m.ReadTasks(context.Background(), &persistence.ReadTasksRequest{
-		QueueKey: persistence.QueueKey{
-			Category: tasks.CategoryTransfer,
-		},
-		PageSize: 1,
-	})
-	assert.ErrorIs(t, err, assert.AnError, "ReadTasks should propagate errors from ReadMessages")
-}
-
-func TestHistoryTaskQueueManager_ReadTasks_ErrEnqueueMessage(t *testing.T) {
-	t.Parallel()
-
-	m := persistence.NewHistoryTaskQueueManager(failingQueue{}, 1)
-	_, err := m.EnqueueTask(context.Background(), &persistence.EnqueueTaskRequest{
-		Task: &tasks.WorkflowTask{},
-	})
-	assert.ErrorIs(t, err, assert.AnError, "EnqueueTask should propagate errors from EnqueueMessage")
-}
-
-func TestHistoryTaskQueueManager_CreateQueue(t *testing.T) {
-	t.Parallel()
-
-	m := persistence.NewHistoryTaskQueueManager(failingQueue{}, 1)
-	_, err := m.CreateQueue(context.Background(), &persistence.CreateQueueRequest{})
-	assert.ErrorIs(t, err, assert.AnError, "CreateQueue should propagate errors from the persistence layer")
-}