mirror of
https://github.com/hanzoai/tasks.git
synced 2026-10-04 02:04:03 +00:00
Roey BermanandGitHub
3871008a84
## What changed? Add the `CherryPick` (a better name than "reapply" IMHO) method to the `hsm.EventDefinition` interface and use this method when "reapplying" events from a different branch. Note that workflow reset with nexus operations is broken right now (and was already broken) as the completion token allocated in the start request targets a specific run. That will be fixed later on. ## Why? Found an edge case during replication when namespace failover happens as a `NexusOperationCompleted` event is buffered causing the caller workflow to get stuck waiting for completion to arrive. ## How did you test it? Functional + unit tests
112 lines
4.4 KiB
Go
112 lines
4.4 KiB
Go
// The MIT License
|
|
//
|
|
// Copyright (c) 2024 Temporal Technologies Inc. All rights reserved.
|
|
//
|
|
// 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 nexusoperations_test
|
|
|
|
import (
|
|
"testing"
|
|
"time"
|
|
|
|
"github.com/stretchr/testify/require"
|
|
historypb "go.temporal.io/api/history/v1"
|
|
"go.temporal.io/server/components/nexusoperations"
|
|
"go.temporal.io/server/service/history/hsm"
|
|
"go.temporal.io/server/service/history/hsm/hsmtest"
|
|
)
|
|
|
|
func TestCherryPick(t *testing.T) {
|
|
setup := func(t *testing.T) (*hsm.Node, nexusoperations.Operation, int64) {
|
|
node := newOperationNode(t, &hsmtest.NodeBackend{}, mustNewScheduledEvent(time.Now(), time.Hour))
|
|
op, err := hsm.MachineData[nexusoperations.Operation](node)
|
|
require.NoError(t, err)
|
|
eventID, err := hsm.EventIDFromToken(op.ScheduledEventToken)
|
|
require.NoError(t, err)
|
|
return node, op, eventID
|
|
}
|
|
|
|
t.Run("NoRequestID", func(t *testing.T) {
|
|
node, _, eventID := setup(t)
|
|
err := nexusoperations.StartedEventDefinition{}.CherryPick(node.Parent, &historypb.HistoryEvent{
|
|
Attributes: &historypb.HistoryEvent_NexusOperationStartedEventAttributes{
|
|
NexusOperationStartedEventAttributes: &historypb.NexusOperationStartedEventAttributes{
|
|
ScheduledEventId: eventID,
|
|
},
|
|
},
|
|
})
|
|
require.NoError(t, err)
|
|
})
|
|
t.Run("ValidRequestID", func(t *testing.T) {
|
|
node, op, eventID := setup(t)
|
|
err := nexusoperations.StartedEventDefinition{}.CherryPick(node.Parent, &historypb.HistoryEvent{
|
|
Attributes: &historypb.HistoryEvent_NexusOperationStartedEventAttributes{
|
|
NexusOperationStartedEventAttributes: &historypb.NexusOperationStartedEventAttributes{
|
|
ScheduledEventId: eventID,
|
|
RequestId: op.RequestId,
|
|
},
|
|
},
|
|
})
|
|
require.NoError(t, err)
|
|
})
|
|
t.Run("InvalidRequestID", func(t *testing.T) {
|
|
node, _, eventID := setup(t)
|
|
err := nexusoperations.StartedEventDefinition{}.CherryPick(node.Parent, &historypb.HistoryEvent{
|
|
Attributes: &historypb.HistoryEvent_NexusOperationStartedEventAttributes{
|
|
NexusOperationStartedEventAttributes: &historypb.NexusOperationStartedEventAttributes{
|
|
ScheduledEventId: eventID,
|
|
RequestId: "invalid",
|
|
},
|
|
},
|
|
})
|
|
require.ErrorIs(t, err, hsm.ErrNotCherryPickable)
|
|
})
|
|
t.Run("InvalidScheduledEventID", func(t *testing.T) {
|
|
node, _, eventID := setup(t)
|
|
err := nexusoperations.StartedEventDefinition{}.CherryPick(node.Parent, &historypb.HistoryEvent{
|
|
Attributes: &historypb.HistoryEvent_NexusOperationStartedEventAttributes{
|
|
NexusOperationStartedEventAttributes: &historypb.NexusOperationStartedEventAttributes{
|
|
ScheduledEventId: eventID + 1,
|
|
},
|
|
},
|
|
})
|
|
require.ErrorIs(t, err, hsm.ErrStateMachineNotFound)
|
|
})
|
|
t.Run("DoubleApply", func(t *testing.T) {
|
|
node, _, eventID := setup(t)
|
|
err := nexusoperations.StartedEventDefinition{}.CherryPick(node.Parent, &historypb.HistoryEvent{
|
|
Attributes: &historypb.HistoryEvent_NexusOperationStartedEventAttributes{
|
|
NexusOperationStartedEventAttributes: &historypb.NexusOperationStartedEventAttributes{
|
|
ScheduledEventId: eventID,
|
|
},
|
|
},
|
|
})
|
|
require.NoError(t, err)
|
|
err = nexusoperations.StartedEventDefinition{}.CherryPick(node.Parent, &historypb.HistoryEvent{
|
|
Attributes: &historypb.HistoryEvent_NexusOperationStartedEventAttributes{
|
|
NexusOperationStartedEventAttributes: &historypb.NexusOperationStartedEventAttributes{
|
|
ScheduledEventId: eventID,
|
|
},
|
|
},
|
|
})
|
|
require.ErrorIs(t, err, hsm.ErrInvalidTransition)
|
|
})
|
|
}
|