mirror of
https://github.com/hanzoai/tasks.git
synced 2026-10-30 14:56:19 +00:00
Michael SnowdenandGitHub
90ae26e93d
<!-- Describe what has changed in this PR --> **What changed?** I discovered an issue with the `ExecutableWrapper` [here](https://github.com/temporalio/temporal/blob/cc733aed8d9a2c73b6a01de992a428c126354869/service/history/queues/executable.go#L451). Because we don't wrap the `Reschedule` method, it adds the unwrapped version to the rescheduler, which loses all of the wrapped behavior. I implemented a fix using the "delegate" pattern, where the base instance holds a reference to the wrapped instance, but it is fraught with peril: you need to remember to use the delegate everywhere it's applicable, and you need to remember to set it--also only one delegate can be set, so you can't have different wrappers for the same object. As a result, I decided to get rid of the wrapper-based approach, and instead added the code directly to the base executable implementation. <!-- Tell your future self why have you made these changes --> **Why?** The huge refactoring is necessary because the `ExecutableWrapper` approach is too complicated and risky. <!-- How have you verified this change? Tested locally? Added a unit test? Checked in staging env? --> **How did you test it?** There's still 100% test coverage for the DLQ code. Most imporantly, there is now an end-to-end integration test which starts a workflow that produces terminally failing tasks, verifies that it's added to the DLQ, and then uses the tdbg read command to delete the DLQ tasks. This wasn't possible before due to the bug in the `ExecutableWrapper`. <!-- Assuming the worst case, what can be broken when deploying this change to production? --> **Potential risks** <!-- Is this PR a hotfix candidate or require that a notification be sent to the broader community? (Yes/No) --> **Is hotfix candidate?**
92 lines
3.2 KiB
Go
92 lines
3.2 KiB
Go
// The MIT License
|
|
//
|
|
// Copyright (c) 2020 Temporal Technologies Inc. All rights reserved.
|
|
//
|
|
// Copyright (c) 2020 Uber Technologies, Inc.
|
|
//
|
|
// Permission is hereby granted, free of charge, to any person obtaining a copy
|
|
// of this software and associated documentation files (the "Software"), to deal
|
|
// in the Software without restriction, including without limitation the rights
|
|
// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
|
|
// copies of the Software, and to permit persons to whom the Software is
|
|
// furnished to do so, subject to the following conditions:
|
|
//
|
|
// The above copyright notice and this permission notice shall be included in
|
|
// all copies or substantial portions of the Software.
|
|
//
|
|
// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
|
|
// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
|
|
// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
|
|
// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
|
|
// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
|
|
// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
|
|
// THE SOFTWARE.
|
|
|
|
// Package getdlqtasks contains the logic to implement the [historyservice.HistoryServiceServer.GetDLQTasks] API.
|
|
package getdlqtasks
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"fmt"
|
|
|
|
"go.temporal.io/api/serviceerror"
|
|
|
|
commonspb "go.temporal.io/server/api/common/v1"
|
|
"go.temporal.io/server/api/historyservice/v1"
|
|
"go.temporal.io/server/common/persistence"
|
|
"go.temporal.io/server/service/history/api"
|
|
"go.temporal.io/server/service/history/consts"
|
|
"go.temporal.io/server/service/history/tasks"
|
|
)
|
|
|
|
// Invoke the GetDLQTasks API. All errors returned from this function are already translated into the appropriate type
|
|
// from the [serviceerror] package.
|
|
func Invoke(
|
|
ctx context.Context,
|
|
historyTaskQueueManager persistence.HistoryTaskQueueManager,
|
|
taskCategoryRegistry tasks.TaskCategoryRegistry,
|
|
req *historyservice.GetDLQTasksRequest,
|
|
) (*historyservice.GetDLQTasksResponse, error) {
|
|
category, err := api.GetTaskCategory(int(req.DlqKey.TaskCategory), taskCategoryRegistry)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
response, err := historyTaskQueueManager.ReadRawTasks(ctx, &persistence.ReadTasksRequest{
|
|
QueueKey: persistence.QueueKey{
|
|
QueueType: persistence.QueueTypeHistoryDLQ,
|
|
Category: category,
|
|
SourceCluster: req.DlqKey.SourceCluster,
|
|
TargetCluster: req.DlqKey.TargetCluster,
|
|
},
|
|
PageSize: int(req.PageSize),
|
|
NextPageToken: req.NextPageToken,
|
|
})
|
|
if err != nil {
|
|
if errors.Is(err, persistence.ErrReadTasksNonPositivePageSize) {
|
|
return nil, consts.ErrInvalidPageSize
|
|
}
|
|
|
|
return nil, serviceerror.NewUnavailable(fmt.Sprintf("GetDLQTasks failed. Error: %v", err))
|
|
}
|
|
|
|
dlqTasks := make([]*commonspb.HistoryDLQTask, len(response.Tasks))
|
|
for i, task := range response.Tasks {
|
|
dlqTasks[i] = &commonspb.HistoryDLQTask{
|
|
Metadata: &commonspb.HistoryDLQTaskMetadata{
|
|
MessageId: task.MessageMetadata.ID,
|
|
},
|
|
Payload: &commonspb.HistoryTask{
|
|
ShardId: task.Payload.ShardId,
|
|
Blob: task.Payload.Blob,
|
|
},
|
|
}
|
|
}
|
|
|
|
return &historyservice.GetDLQTasksResponse{
|
|
DlqTasks: dlqTasks,
|
|
NextPageToken: response.NextPageToken,
|
|
}, nil
|
|
}
|