diff --git a/pkg/actions/actions.go b/pkg/actions/actions.go index 6f78e41d5..20f0fd74b 100644 --- a/pkg/actions/actions.go +++ b/pkg/actions/actions.go @@ -17,6 +17,8 @@ import ( "go.uber.org/zap" "google.golang.org/grpc/codes" "google.golang.org/grpc/status" + "google.golang.org/protobuf/proto" + "google.golang.org/protobuf/types/known/anypb" "google.golang.org/protobuf/types/known/structpb" ) @@ -31,6 +33,10 @@ type OutstandingAction struct { Err error StartedAt time.Time sync.Mutex + + // cancelled marks a FAILED status that came from request cancellation + // rather than from the handler; the handler's late outcome replaces it. + cancelled bool } func NewOutstandingAction(id, name string) *OutstandingAction { @@ -45,41 +51,167 @@ func NewOutstandingAction(id, name string) *OutstandingAction { func (oa *OutstandingAction) SetStatus(ctx context.Context, status v2.BatonActionStatus) { oa.Lock() defer oa.Unlock() - l := ctxzap.Extract(ctx).With( - zap.String("action_id", oa.Id), - zap.String("action_name", oa.Name), - zap.String("status", status.String()), - ) + oa.setStatusLocked(ctx, status) +} + +// setStatusLocked applies a lifecycle transition and reports whether it was +// accepted. Terminal statuses are final and RUNNING is only reachable from +// PENDING; rejected transitions are dropped. These fire in normal operation +// (a handler finishing after its request was cancelled), hence debug level. +// It requires oa's mutex to be held. +func (oa *OutstandingAction) setStatusLocked(ctx context.Context, status v2.BatonActionStatus) bool { if oa.Status == v2.BatonActionStatus_BATON_ACTION_STATUS_COMPLETE || oa.Status == v2.BatonActionStatus_BATON_ACTION_STATUS_FAILED { - l.Error("cannot set status on completed action") + ctxzap.Extract(ctx).Debug("dropping status transition on terminal action", + zap.String("action_id", oa.Id), + zap.String("action_name", oa.Name), + zap.String("status", oa.Status.String()), + zap.String("requested_status", status.String())) + return false } if status == v2.BatonActionStatus_BATON_ACTION_STATUS_RUNNING && oa.Status != v2.BatonActionStatus_BATON_ACTION_STATUS_PENDING { - l.Error("cannot set status to running unless action is pending") + ctxzap.Extract(ctx).Debug("dropping running transition on non-pending action", + zap.String("action_id", oa.Id), + zap.String("action_name", oa.Name), + zap.String("status", oa.Status.String())) + return false } oa.Status = status + return true } -func (oa *OutstandingAction) setError(_ context.Context, err error) { - oa.Lock() - defer oa.Unlock() - if oa.Rv == nil { - oa.Rv = &structpb.Struct{} - } - if oa.Rv.Fields == nil { - oa.Rv.Fields = make(map[string]*structpb.Value) +// setErrorLocked requires oa's mutex to be held. +func (oa *OutstandingAction) setErrorLocked(err error) { + // Rebuild rather than mutate: a concurrent caller may hold the previously + // published struct from a result() snapshot and be marshaling it. + fields := make(map[string]*structpb.Value, len(oa.Rv.GetFields())+1) + for k, v := range oa.Rv.GetFields() { + fields[k] = v } - oa.Rv.Fields["error"] = &structpb.Value{ + fields["error"] = &structpb.Value{ Kind: &structpb.Value_StringValue{ StringValue: err.Error(), }, } + oa.Rv = &structpb.Struct{Fields: fields} oa.Err = err } +// SetError records the error and marks the action failed in one critical +// section, so no snapshot can observe the error with a non-terminal status. +// A FAILED action may refresh its error payload, but a COMPLETE action stays +// complete. This is a real handler failure, so it clears any provisional +// cancellation mark. func (oa *OutstandingAction) SetError(ctx context.Context, err error) { - oa.setError(ctx, err) - oa.SetStatus(ctx, v2.BatonActionStatus_BATON_ACTION_STATUS_FAILED) + oa.Lock() + defer oa.Unlock() + if oa.Status == v2.BatonActionStatus_BATON_ACTION_STATUS_FAILED { + oa.setErrorLocked(err) + oa.cancelled = false + return + } + if oa.setStatusLocked(ctx, v2.BatonActionStatus_BATON_ACTION_STATUS_FAILED) { + oa.setErrorLocked(err) + } +} + +// setCancelled records a cancellation as a provisional failure: unlike other +// terminal states, the handler's own late outcome replaces it, since the +// action's side effects can still complete after the request goes away. +func (oa *OutstandingAction) setCancelled(ctx context.Context, err error) { + oa.Lock() + defer oa.Unlock() + if oa.setStatusLocked(ctx, v2.BatonActionStatus_BATON_ACTION_STATUS_FAILED) { + oa.setErrorLocked(err) + oa.cancelled = true + } +} + +// isProvisional reports whether the action's FAILED status is a provisional +// cancellation mark that the handler's late outcome may still replace. +func (oa *OutstandingAction) isProvisional() bool { + oa.Lock() + defer oa.Unlock() + return oa.cancelled +} + +// evictable reports whether the action has reached a state cleanup may +// remove: terminal, and not a provisional cancellation whose record can +// still improve. +func (oa *OutstandingAction) evictable() bool { + oa.Lock() + defer oa.Unlock() + if oa.cancelled { + return false + } + return oa.Status == v2.BatonActionStatus_BATON_ACTION_STATUS_COMPLETE || oa.Status == v2.BatonActionStatus_BATON_ACTION_STATUS_FAILED +} + +// Result returns the action's identity and current outcome. The snapshot is +// internally consistent; the returned message and annotations are owned by +// the action and must not be modified. +func (oa *OutstandingAction) Result() (string, v2.BatonActionStatus, *structpb.Struct, annotations.Annotations) { + return oa.result() +} + +// result is the unexported form of Result. +func (oa *OutstandingAction) result() (string, v2.BatonActionStatus, *structpb.Struct, annotations.Annotations) { + oa.Lock() + defer oa.Unlock() + return oa.Id, oa.Status, oa.Rv, oa.Annos +} + +// setOutcome publishes the handler's result and terminal status in one +// critical section, so no snapshot observes one without the other. The +// values are cloned at this publication seam: the handler owns what it +// returned and may keep mutating it. Terminal statuses are final, with one +// exception: a cancellation-FAILED status is provisional, and the handler's +// own outcome — success or failure — replaces it. +func (oa *OutstandingAction) setOutcome(ctx context.Context, rv *structpb.Struct, annos annotations.Annotations, err error) { + if rv != nil { + rv = proto.Clone(rv).(*structpb.Struct) + } + if annos != nil { + annosCopy := make(annotations.Annotations, len(annos)) + for i, a := range annos { + if a != nil { + annosCopy[i] = proto.Clone(a).(*anypb.Any) + } + } + annos = annosCopy + } + + oa.Lock() + defer oa.Unlock() + + if err != nil { + if oa.Status == v2.BatonActionStatus_BATON_ACTION_STATUS_FAILED || oa.setStatusLocked(ctx, v2.BatonActionStatus_BATON_ACTION_STATUS_FAILED) { + oa.Rv = rv + oa.Annos = annos + oa.setErrorLocked(err) + oa.cancelled = false + } + return + } + + if oa.cancelled { + // Deliberate cross-terminal replacement: the cancellation was a + // transport event, not the action's outcome. + ctxzap.Extract(ctx).Debug("replacing provisional cancellation with handler outcome", + zap.String("action_id", oa.Id), + zap.String("action_name", oa.Name)) + oa.cancelled = false + oa.Status = v2.BatonActionStatus_BATON_ACTION_STATUS_COMPLETE + oa.Rv = rv + oa.Annos = annos + oa.Err = nil + return + } + + if oa.setStatusLocked(ctx, v2.BatonActionStatus_BATON_ACTION_STATUS_COMPLETE) { + oa.Rv = rv + oa.Annos = annos + } } const maxOldActions = 1000 @@ -163,8 +295,7 @@ func (a *ActionManager) CleanupOldActions(ctx context.Context) { count := 0 // Delete the oldest actions for i := 0; i < len(actionList)-maxOldActions; i++ { - action := actionList[i] - if action.Status == v2.BatonActionStatus_BATON_ACTION_STATUS_COMPLETE || action.Status == v2.BatonActionStatus_BATON_ACTION_STATUS_FAILED { + if actionList[i].evictable() { count++ delete(a.actions, actionList[i].Id) } @@ -398,7 +529,8 @@ func (a *ActionManager) GetActionStatus(_ context.Context, actionId string) (v2. // Don't return oa.Err here because error is for GetActionStatus, not the action itself. // oa.Rv contains any error. - return oa.Status, oa.Name, oa.Rv, oa.Annos, nil + _, st, rv, annos := oa.result() + return st, oa.Name, rv, annos, nil } // InvokeAction invokes an action. If resourceTypeID is set, it invokes a resource-scoped action. @@ -457,23 +589,34 @@ func (a *ActionManager) invokeGlobalAction(ctx context.Context, name string, arg bgCtx := trace.ContextWithSpanContext(context.Background(), trace.SpanContextFromContext(ctx)) handlerCtx, cancel := context.WithTimeoutCause(bgCtx, 1*time.Hour, errors.New("action handler timed out")) defer cancel() - var oaErr error - oa.Rv, oa.Annos, oaErr = handler(handlerCtx, args) - if oaErr == nil { - oa.SetStatus(ctx, v2.BatonActionStatus_BATON_ACTION_STATUS_COMPLETE) - } else { - oa.SetError(ctx, oaErr) - } + rv, annos, oaErr := handler(handlerCtx, args) + oa.setOutcome(ctx, rv, annos, oaErr) }() select { case <-done: - return oa.Id, oa.Status, oa.Rv, oa.Annos, nil + id, st, rv, annos := oa.result() + return id, st, rv, annos, nil case <-time.After(1 * time.Second): - return oa.Id, oa.Status, oa.Rv, oa.Annos, nil + id, st, rv, annos := oa.result() + return id, st, rv, annos, nil case <-ctx.Done(): - oa.SetError(ctx, ctx.Err()) - return oa.Id, oa.Status, oa.Rv, oa.Annos, ctx.Err() + // The handler may have finished in the same instant; prefer its + // completed result over a spurious cancellation return. + select { + case <-done: + id, st, rv, annos := oa.result() + return id, st, rv, annos, nil + default: + } + oa.setCancelled(ctx, ctx.Err()) + id, st, rv, annos := oa.result() + if st == v2.BatonActionStatus_BATON_ACTION_STATUS_COMPLETE { + // The handler won the race to the lock; its completed result is + // the authoritative pairing, not the cancellation. + return id, st, rv, annos, nil + } + return id, st, rv, annos, ctx.Err() } } @@ -547,24 +690,35 @@ func (a *ActionManager) invokeResourceAction( bgCtx = ctxzap.ToContext(bgCtx, ctxzap.Extract(ctx)) handlerCtx, cancel := context.WithTimeoutCause(bgCtx, 1*time.Hour, errors.New("action handler timed out")) defer cancel() - var oaErr error - oa.Rv, oa.Annos, oaErr = handler(handlerCtx, args) - if oaErr == nil { - oa.SetStatus(ctx, v2.BatonActionStatus_BATON_ACTION_STATUS_COMPLETE) - } else { - oa.SetError(ctx, oaErr) - } + rv, annos, oaErr := handler(handlerCtx, args) + oa.setOutcome(ctx, rv, annos, oaErr) }() // Wait for completion or timeout select { case <-done: - return oa.Id, oa.Status, oa.Rv, oa.Annos, nil + id, st, rv, annos := oa.result() + return id, st, rv, annos, nil case <-time.After(1 * time.Second): - return oa.Id, oa.Status, oa.Rv, oa.Annos, nil + id, st, rv, annos := oa.result() + return id, st, rv, annos, nil case <-ctx.Done(): - oa.SetError(ctx, ctx.Err()) - return oa.Id, oa.Status, oa.Rv, oa.Annos, ctx.Err() + // The handler may have finished in the same instant; prefer its + // completed result over a spurious cancellation return. + select { + case <-done: + id, st, rv, annos := oa.result() + return id, st, rv, annos, nil + default: + } + oa.setCancelled(ctx, ctx.Err()) + id, st, rv, annos := oa.result() + if st == v2.BatonActionStatus_BATON_ACTION_STATUS_COMPLETE { + // The handler won the race to the lock; its completed result is + // the authoritative pairing, not the cancellation. + return id, st, rv, annos, nil + } + return id, st, rv, annos, ctx.Err() } } diff --git a/pkg/actions/actions_test.go b/pkg/actions/actions_test.go index 88036767a..56452bdca 100644 --- a/pkg/actions/actions_test.go +++ b/pkg/actions/actions_test.go @@ -13,6 +13,8 @@ import ( "github.com/stretchr/testify/require" "google.golang.org/grpc/codes" "google.golang.org/grpc/status" + "google.golang.org/protobuf/proto" + "google.golang.org/protobuf/types/known/anypb" "google.golang.org/protobuf/types/known/structpb" ) @@ -546,3 +548,300 @@ func TestActionHandlerGoroutineLeaks(t *testing.T) { require.LessOrEqual(t, finalCount, initialCount+1, "goroutine leak detected after context cancellation") }) } + +// The data race this test guards against is only detectable under -race, +// which the plain CI go-test job does not enable for this package; `make +// race-check` is the out-of-band gate that runs it. The assertions at the end +// only cover the exported snapshot accessor. +func TestCleanupOldActionsDuringConcurrentStatusWrites(t *testing.T) { + ctx := t.Context() + m := NewActionManager(ctx) + + // The cleanup loop only visits the len(actions)-maxOldActions oldest + // entries, so the concurrent writer must target the first-created action. + // The sort is unstable and StartedAt values can tie, so push the target + // strictly earlier to make its position deterministic. + oldest := m.GetNewAction("churn") + oldest.StartedAt = time.Now().Add(-time.Hour) + for i := 0; i < maxOldActions; i++ { + m.GetNewAction("churn") + } + + stop := make(chan struct{}) + started := make(chan struct{}) + writerDone := make(chan struct{}) + go func() { + defer close(writerDone) + first := true + for { + select { + case <-stop: + return + default: + // Write the status under the lock directly: lifecycle + // transitions are single-shot, so no public API writes the + // status repeatedly, and the instrument needs a sustained + // locked writer to race cleanup's read against. + oldest.Lock() + oldest.Status = v2.BatonActionStatus_BATON_ACTION_STATUS_RUNNING + oldest.Unlock() + if first { + close(started) + first = false + } + } + } + }() + + // Wait for the writer's first write: otherwise it may only be scheduled + // after close(stop), never race cleanup, and leave the action PENDING. + <-started + + // Fails under -race if cleanup reads action status without the lock. + m.CleanupOldActions(ctx) + + close(stop) + <-writerDone + + // The exported snapshot accessor reads the same state race-free. + id, actionStatus, _, _ := oldest.Result() + require.Equal(t, oldest.Id, id) + require.Equal(t, v2.BatonActionStatus_BATON_ACTION_STATUS_RUNNING, actionStatus) +} + +func TestOutstandingActionLifecycleTransitions(t *testing.T) { + const ( + pending = v2.BatonActionStatus_BATON_ACTION_STATUS_PENDING + running = v2.BatonActionStatus_BATON_ACTION_STATUS_RUNNING + complete = v2.BatonActionStatus_BATON_ACTION_STATUS_COMPLETE + failed = v2.BatonActionStatus_BATON_ACTION_STATUS_FAILED + ) + cases := []struct { + name string + from v2.BatonActionStatus + to v2.BatonActionStatus + want v2.BatonActionStatus + }{ + {"pending to running", pending, running, running}, + {"pending to complete", pending, complete, complete}, + {"pending to failed", pending, failed, failed}, + {"running to complete", running, complete, complete}, + {"running to failed", running, failed, failed}, + {"running to running rejected", running, running, running}, + {"complete rejects running", complete, running, complete}, + {"complete rejects failed", complete, failed, complete}, + {"failed rejects complete", failed, complete, failed}, + {"failed rejects running", failed, running, failed}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + oa := NewOutstandingAction("id", "lifecycle") + oa.Status = tc.from + oa.SetStatus(t.Context(), tc.to) + _, actionStatus, _, _ := oa.Result() + require.Equal(t, tc.want, actionStatus) + }) + } +} + +func TestLateSuccessAfterCancelReplacesIt(t *testing.T) { + ctx := t.Context() + oa := NewOutstandingAction("id", "cancelled") + oa.SetStatus(ctx, v2.BatonActionStatus_BATON_ACTION_STATUS_RUNNING) + oa.setCancelled(ctx, context.Canceled) + + rv, err := structpb.NewStruct(map[string]any{"success": true}) + require.NoError(t, err) + oa.setOutcome(ctx, rv, nil, nil) + + // The cancellation was a transport event; the handler's success is the + // action's real outcome. + _, actionStatus, gotRv, _ := oa.Result() + require.Equal(t, v2.BatonActionStatus_BATON_ACTION_STATUS_COMPLETE, actionStatus) + require.True(t, gotRv.Fields["success"].GetBoolValue()) + require.Nil(t, gotRv.Fields["error"]) +} + +func TestLateFailureAfterCancelReplacesError(t *testing.T) { + ctx := t.Context() + oa := NewOutstandingAction("id", "cancelled") + oa.SetStatus(ctx, v2.BatonActionStatus_BATON_ACTION_STATUS_RUNNING) + oa.setCancelled(ctx, context.Canceled) + + oa.setOutcome(ctx, nil, nil, fmt.Errorf("upstream rejected the request")) + + _, actionStatus, gotRv, _ := oa.Result() + require.Equal(t, v2.BatonActionStatus_BATON_ACTION_STATUS_FAILED, actionStatus) + require.Equal(t, "upstream rejected the request", gotRv.Fields["error"].GetStringValue()) +} + +func TestLateSuccessAfterRealFailureIsDropped(t *testing.T) { + ctx := t.Context() + oa := NewOutstandingAction("id", "panicked") + oa.SetStatus(ctx, v2.BatonActionStatus_BATON_ACTION_STATUS_RUNNING) + oa.SetError(ctx, fmt.Errorf("panic in action handler")) + + rv, err := structpb.NewStruct(map[string]any{"success": true}) + require.NoError(t, err) + oa.setOutcome(ctx, rv, nil, nil) + + // A real handler failure is final; only cancellation is provisional. + _, actionStatus, gotRv, _ := oa.Result() + require.Equal(t, v2.BatonActionStatus_BATON_ACTION_STATUS_FAILED, actionStatus) + require.Equal(t, "panic in action handler", gotRv.Fields["error"].GetStringValue()) + require.Nil(t, gotRv.Fields["success"]) +} + +func TestPublishedOutcomeIsIsolatedFromHandler(t *testing.T) { + ctx := t.Context() + oa := NewOutstandingAction("id", "isolated") + oa.SetStatus(ctx, v2.BatonActionStatus_BATON_ACTION_STATUS_RUNNING) + + rv, err := structpb.NewStruct(map[string]any{"k": "v"}) + require.NoError(t, err) + anno, err := anypb.New(structpb.NewStringValue("original")) + require.NoError(t, err) + oa.setOutcome(ctx, rv, annotations.Annotations{anno}, nil) + + // The handler owns what it returned and may keep mutating it; the + // published outcome must not change. Annotations are deep-copied, so + // element mutation is isolated too. + rv.Fields["mutated"] = structpb.NewBoolValue(true) + anno.TypeUrl = "mutated" + + _, actionStatus, gotRv, gotAnnos := oa.Result() + require.Equal(t, v2.BatonActionStatus_BATON_ACTION_STATUS_COMPLETE, actionStatus) + require.Nil(t, gotRv.Fields["mutated"]) + require.Equal(t, "v", gotRv.Fields["k"].GetStringValue()) + require.Len(t, gotAnnos, 1) + require.NotEqual(t, "mutated", gotAnnos[0].TypeUrl) + + // Under -race: concurrent handler mutation against reader marshals. + stop := make(chan struct{}) + writerDone := make(chan struct{}) + go func() { + defer close(writerDone) + for i := 0; ; i++ { + select { + case <-stop: + return + default: + rv.Fields[fmt.Sprintf("m%d", i)] = structpb.NewBoolValue(true) + } + } + }() + for i := 0; i < 100; i++ { + _, _, snapshot, _ := oa.Result() + _, err := proto.Marshal(snapshot) + require.NoError(t, err) + } + close(stop) + <-writerDone +} + +func TestCleanupRetainsProvisionallyCancelledActions(t *testing.T) { + ctx := t.Context() + m := NewActionManager(ctx) + + // Two terminal actions old enough for cleanup to visit: one provisional + // cancellation whose handler may still publish, one real failure. + provisional := m.GetNewAction("cancelled") + provisional.StartedAt = time.Now().Add(-2 * time.Hour) + provisional.SetStatus(ctx, v2.BatonActionStatus_BATON_ACTION_STATUS_RUNNING) + provisional.setCancelled(ctx, context.Canceled) + + failed := m.GetNewAction("failed") + failed.StartedAt = time.Now().Add(-time.Hour) + failed.SetStatus(ctx, v2.BatonActionStatus_BATON_ACTION_STATUS_RUNNING) + failed.SetError(ctx, fmt.Errorf("real failure")) + + for i := 0; i < maxOldActions; i++ { + m.GetNewAction("churn") + } + + m.CleanupOldActions(ctx) + + // The real failure is evictable; the provisional record must survive so + // the handler's late outcome stays observable. + _, _, _, _, err := m.GetActionStatus(ctx, failed.Id) + require.Error(t, err) + + actionStatus, _, _, _, err := m.GetActionStatus(ctx, provisional.Id) + require.NoError(t, err) + require.Equal(t, v2.BatonActionStatus_BATON_ACTION_STATUS_FAILED, actionStatus) + + rv, err := structpb.NewStruct(map[string]any{"success": true}) + require.NoError(t, err) + provisional.setOutcome(ctx, rv, nil, nil) + + actionStatus, _, gotRv, _, err := m.GetActionStatus(ctx, provisional.Id) + require.NoError(t, err) + require.Equal(t, v2.BatonActionStatus_BATON_ACTION_STATUS_COMPLETE, actionStatus) + require.True(t, gotRv.Fields["success"].GetBoolValue()) +} + +func TestPanicAfterCancelIsFinal(t *testing.T) { + ctx := t.Context() + oa := NewOutstandingAction("id", "cancelled-then-panicked") + oa.SetStatus(ctx, v2.BatonActionStatus_BATON_ACTION_STATUS_RUNNING) + oa.setCancelled(ctx, context.Canceled) + + // The recovery path reports a panic through SetError: a real handler + // failure that replaces the provisional mark and becomes final. + oa.SetError(ctx, fmt.Errorf("panic in action handler: boom")) + require.False(t, oa.isProvisional()) + + rv, err := structpb.NewStruct(map[string]any{"success": true}) + require.NoError(t, err) + oa.setOutcome(ctx, rv, nil, nil) + + _, actionStatus, gotRv, _ := oa.Result() + require.Equal(t, v2.BatonActionStatus_BATON_ACTION_STATUS_FAILED, actionStatus) + require.Equal(t, "panic in action handler: boom", gotRv.Fields["error"].GetStringValue()) + require.Nil(t, gotRv.Fields["success"]) +} + +func TestCancelAfterCompletionIsRejected(t *testing.T) { + ctx := t.Context() + oa := NewOutstandingAction("id", "completed") + oa.SetStatus(ctx, v2.BatonActionStatus_BATON_ACTION_STATUS_RUNNING) + + rv, err := structpb.NewStruct(map[string]any{"success": true}) + require.NoError(t, err) + oa.setOutcome(ctx, rv, nil, nil) + + oa.setCancelled(ctx, context.Canceled) + + // COMPLETE is truly terminal: the cancellation neither marks the action + // provisional nor touches the published outcome. + require.False(t, oa.isProvisional()) + _, actionStatus, gotRv, _ := oa.Result() + require.Equal(t, v2.BatonActionStatus_BATON_ACTION_STATUS_COMPLETE, actionStatus) + require.True(t, gotRv.Fields["success"].GetBoolValue()) + require.Nil(t, gotRv.Fields["error"]) +} + +func TestCancelledInvokeStatusErrorPairing(t *testing.T) { + ctx := t.Context() + m := NewActionManager(ctx) + require.NoError(t, m.Register(ctx, testActionSchema, testActionHandler)) + + // The handler succeeds instantly while each request is cancelled + // concurrently, sampling the invoke select race from both sides. The + // orderings can't be forced individually, but every interleaving must + // satisfy the pairing contract: a cancellation error only ever + // accompanies FAILED, and an errorless return is never FAILED (RUNNING + // is tolerated only for a pathological scheduler stall past the inline + // wait). + for i := 0; i < 200; i++ { + invokeCtx, cancel := context.WithCancel(ctx) + go cancel() + _, actionStatus, _, _, err := m.InvokeAction(invokeCtx, "lock_account", "", testInput) + if err != nil { + require.ErrorIs(t, err, context.Canceled) + require.Equal(t, v2.BatonActionStatus_BATON_ACTION_STATUS_FAILED, actionStatus) + } else { + require.NotEqual(t, v2.BatonActionStatus_BATON_ACTION_STATUS_FAILED, actionStatus) + } + } +}