mirror of
https://github.com/hanzoai/tasks.git
synced 2026-10-30 14:56:19 +00:00
nikki-dagandGitHub
a645917055
## What changed? The changes consistently add a context.Context parameter to: - CloseTransactionAsSnapshot() - the main transaction close method - TestCloneToProto() - test helper method - prepareNewWorkflow() - workflow preparation method All call sites have been updated to pass the context through, with test code using context.Background() where needed. This is a clean refactoring to enable context propagation through the transaction close path. ## Why? We want to support accessing contextual information before closing the mutable state transaction. ## How did you test it? - [x] built - [x] run locally and tested manually - [x] covered by existing tests - [ ] added new unit test(s) - [ ] added new functional test(s) --- > [!NOTE] > Add context propagation to mutable state transaction closing by updating method signatures and callers across history, ndc, workflow, and tests. > > - **API changes**: > - Add `context.Context` parameter to `MutableState.CloseTransactionAsSnapshot` and `CloseTransactionAsMutation` in `history/interfaces` (and mocks). > - Update `workflow.TestCloneToProto(ctx, ms)` to accept/pass `ctx`. > - **Implementations**: > - Propagate `ctx` through `workflow/mutable_state_impl` (including internal `closeTransaction`), `workflow/context`, and CHASM paths. > - **Call sites updated**: > - Replace calls to `CloseTransactionAsSnapshot/Mutation(...)` with `(..., ctx, ...)` across `history_engine*`, `ndc/*`, `api/*`, queue executors, reset/import/rebuilder, and replication. > - Tests pass `context.Background()` where needed; numerous `TestCloneToProto` usages updated accordingly. > - **Related adjustments**: > - `Starter.prepareNewWorkflow` and other helpers now take `ctx` and forward it to snapshot closing. > > <sup>Written by [Cursor Bugbot]() for commit ec8a15f5aff6a7c9e2a980e6569b4926425c0c7c. This will update automatically on new commits. Configure [here]().</sup>
100 lines
2.8 KiB
Go
100 lines
2.8 KiB
Go
package workflow
|
|
|
|
import (
|
|
"context"
|
|
"testing"
|
|
"time"
|
|
|
|
historypb "go.temporal.io/api/history/v1"
|
|
"go.temporal.io/api/serviceerror"
|
|
persistencespb "go.temporal.io/server/api/persistence/v1"
|
|
"go.temporal.io/server/common/log"
|
|
"go.temporal.io/server/common/namespace"
|
|
"go.temporal.io/server/service/history/events"
|
|
historyi "go.temporal.io/server/service/history/interfaces"
|
|
"go.temporal.io/server/service/history/tests"
|
|
"go.uber.org/mock/gomock"
|
|
)
|
|
|
|
func TestLocalMutableState(
|
|
shard historyi.ShardContext,
|
|
eventsCache events.Cache,
|
|
ns *namespace.Namespace,
|
|
workflowID string,
|
|
runID string,
|
|
logger log.Logger,
|
|
) *MutableStateImpl {
|
|
|
|
ms := NewMutableState(shard, eventsCache, logger, ns, workflowID, runID, time.Now().UTC())
|
|
ms.executionInfo.NamespaceId = string(ns.ID())
|
|
ms.executionInfo.WorkflowId = workflowID
|
|
ms.executionState.RunId = runID
|
|
ms.GetExecutionInfo().ExecutionTime = ms.GetExecutionState().StartTime
|
|
_ = ms.SetHistoryTree(nil, nil, runID)
|
|
|
|
return ms
|
|
}
|
|
|
|
// NewMapEventCache is a functional event cache mock that wraps a simple Go map
|
|
func NewMapEventCache(
|
|
t *testing.T,
|
|
m map[events.EventKey]*historypb.HistoryEvent,
|
|
) events.Cache {
|
|
cache := events.NewMockCache(gomock.NewController(t))
|
|
cache.EXPECT().DeleteEvent(gomock.Any()).AnyTimes().Do(
|
|
func(k events.EventKey) { delete(m, k) },
|
|
)
|
|
cache.EXPECT().PutEvent(gomock.Any(), gomock.Any()).AnyTimes().Do(
|
|
func(k events.EventKey, event *historypb.HistoryEvent) {
|
|
m[k] = event
|
|
},
|
|
)
|
|
cache.EXPECT().GetEvent(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).
|
|
AnyTimes().
|
|
DoAndReturn(
|
|
func(
|
|
_ context.Context,
|
|
_ int32,
|
|
key events.EventKey,
|
|
_ int64,
|
|
_ []byte,
|
|
) (*historypb.HistoryEvent, error) {
|
|
if event, ok := m[key]; ok {
|
|
return event, nil
|
|
}
|
|
return nil, serviceerror.NewNotFoundf("event %#v not found", key)
|
|
},
|
|
)
|
|
return cache
|
|
}
|
|
|
|
func TestGlobalMutableState(
|
|
shard historyi.ShardContext,
|
|
eventsCache events.Cache,
|
|
logger log.Logger,
|
|
version int64,
|
|
workflowID string,
|
|
runID string,
|
|
) *MutableStateImpl {
|
|
|
|
ms := NewMutableState(shard, eventsCache, logger, tests.GlobalNamespaceEntry, workflowID, runID, time.Now().UTC())
|
|
ms.GetExecutionInfo().ExecutionTime = ms.GetExecutionState().StartTime
|
|
ms.GetExecutionInfo().TransitionHistory = UpdatedTransitionHistory(ms.GetExecutionInfo().TransitionHistory, version)
|
|
_ = ms.UpdateCurrentVersion(version, false)
|
|
_ = ms.SetHistoryTree(nil, nil, runID)
|
|
|
|
return ms
|
|
}
|
|
|
|
func TestCloneToProto(
|
|
ctx context.Context,
|
|
mutableState historyi.MutableState,
|
|
) *persistencespb.WorkflowMutableState {
|
|
if mutableState.HasBufferedEvents() {
|
|
_, _, _ = mutableState.CloseTransactionAsMutation(ctx, historyi.TransactionPolicyActive)
|
|
} else {
|
|
_, _, _ = mutableState.CloseTransactionAsSnapshot(ctx, historyi.TransactionPolicyActive)
|
|
}
|
|
return mutableState.CloneToProto()
|
|
}
|