From 3d0cb9136098645e5513b05f554389edbb7b828d Mon Sep 17 00:00:00 2001 From: 1fanwang <1fannnw@gmail.com> Date: Fri, 21 Aug 2026 11:47:43 -0700 Subject: [PATCH 1/6] Check RecordAction body status before deduplicating RecordAction reports application rejections in its body while returning a nil transport error. Memoizing those keys drops all retries, leaving rejected actions unrecorded. Only an explicit OK response is safe to memoize. Signed-off-by: 1fanwang <1fannnw@gmail.com> --- actions/k8s/client.go | 12 +++- actions/k8s/client_test.go | 138 ++++++++++++++++++++++++++++++++++--- 2 files changed, 138 insertions(+), 12 deletions(-) diff --git a/actions/k8s/client.go b/actions/k8s/client.go index 03090bb347..85c7740ea3 100644 --- a/actions/k8s/client.go +++ b/actions/k8s/client.go @@ -9,6 +9,7 @@ import ( "time" "connectrpc.com/connect" + "google.golang.org/genproto/googleapis/rpc/code" "google.golang.org/protobuf/proto" "google.golang.org/protobuf/types/known/timestamppb" apierrors "k8s.io/apimachinery/pkg/api/errors" @@ -782,8 +783,17 @@ func (c *ActionsClient) notifyRunService(ctx context.Context, taskAction *execut Task: ta, } } - if _, err := c.runClient.RecordAction(ctx, connect.NewRequest(recordReq)); err != nil { + // RecordAction reports rejections in the response body, not as a transport + // error, so the body status has to be checked before memoizing the key — + // otherwise a rejected action is never recorded and never retried. + resp, err := c.runClient.RecordAction(ctx, connect.NewRequest(recordReq)) + if err != nil { logger.Warnf(ctx, "Failed to record action in run service for %s: %v", update.ActionID.Name, err) + } else if resp == nil || resp.Msg == nil || resp.Msg.GetStatus() == nil { + logger.Warnf(ctx, "Run service returned no RecordAction status for %s", update.ActionID.Name) + } else if status := resp.Msg.GetStatus(); status.GetCode() != int32(code.Code_OK) { + logger.Warnf(ctx, "Run service rejected RecordAction for %s with code %d: %s", + update.ActionID.Name, status.GetCode(), status.GetMessage()) } else { c.recordedFilter.Add(ctx, actionKey) } diff --git a/actions/k8s/client_test.go b/actions/k8s/client_test.go index fa1079c8a4..46005dea2a 100644 --- a/actions/k8s/client_test.go +++ b/actions/k8s/client_test.go @@ -10,6 +10,7 @@ import ( "github.com/stretchr/testify/assert" "github.com/stretchr/testify/mock" "github.com/stretchr/testify/require" + "google.golang.org/genproto/googleapis/rpc/code" "google.golang.org/genproto/googleapis/rpc/status" "google.golang.org/grpc/codes" "google.golang.org/protobuf/proto" @@ -55,6 +56,12 @@ func newTestActionUpdate(actionName string) (*executorv1.TaskAction, *ActionUpda return ta, update } +func acceptedRecordActionResponse() *connect.Response[workflow.RecordActionResponse] { + return connect.NewResponse(&workflow.RecordActionResponse{ + Status: &status.Status{Code: int32(code.Code_OK)}, + }) +} + func TestNotifyRunService_DeduplicateRecordAction(t *testing.T) { ctx := context.Background() @@ -73,7 +80,7 @@ func TestNotifyRunService_DeduplicateRecordAction(t *testing.T) { // Expect RecordAction called exactly once mockClient.On("RecordAction", mock.Anything, mock.Anything). - Return(&connect.Response[workflow.RecordActionResponse]{}, nil).Once() + Return(acceptedRecordActionResponse(), nil).Once() // First Added event — should call RecordAction c.notifyRunService(ctx, ta, update, watch.Added) @@ -105,7 +112,7 @@ func TestNotifyRunService_FailedRecordAllowsRetry(t *testing.T) { Return((*connect.Response[workflow.RecordActionResponse])(nil), fmt.Errorf("transient error")).Once() // Second call succeeds mockClient.On("RecordAction", mock.Anything, mock.Anything). - Return(&connect.Response[workflow.RecordActionResponse]{}, nil).Once() + Return(acceptedRecordActionResponse(), nil).Once() // First event — RecordAction fails, should NOT add to filter c.notifyRunService(ctx, ta, update, watch.Added) @@ -120,6 +127,115 @@ func TestNotifyRunService_FailedRecordAllowsRetry(t *testing.T) { mockClient.AssertNumberOfCalls(t, "RecordAction", 2) } +func TestNotifyRunService_MissingResponseAllowsRetry(t *testing.T) { + for _, tc := range []struct { + name string + response *connect.Response[workflow.RecordActionResponse] + }{ + {name: "nil response"}, + {name: "nil message", response: &connect.Response[workflow.RecordActionResponse]{}}, + {name: "nil status", response: connect.NewResponse(&workflow.RecordActionResponse{})}, + } { + t.Run(tc.name, func(t *testing.T) { + ctx := context.Background() + mockClient := runmocks.NewInternalRunServiceClient(t) + filter, err := fastcheck.NewOppoBloomFilter(128, promutils.NewTestScope()) + assert.NoError(t, err) + c := &ActionsClient{ + runClient: mockClient, + recordedFilter: filter, + subscribers: make(map[string]map[chan *ActionUpdate]struct{}), + } + ta, update := newTestActionUpdate("action-missing-response") + mockClient.On("RecordAction", mock.Anything, mock.Anything). + Return(tc.response, nil).Twice() + + c.notifyRunService(ctx, ta, update, watch.Added) + c.notifyRunService(ctx, ta, update, watch.Added) + + mockClient.AssertNumberOfCalls(t, "RecordAction", 2) + }) + } +} + +func TestNotifyRunService_RejectedRecordAllowsRetry(t *testing.T) { + ctx := context.Background() + + mockClient := runmocks.NewInternalRunServiceClient(t) + + filter, err := fastcheck.NewOppoBloomFilter(128, promutils.NewTestScope()) + assert.NoError(t, err) + + c := &ActionsClient{ + runClient: mockClient, + recordedFilter: filter, + subscribers: make(map[string]map[chan *ActionUpdate]struct{}), + } + + ta, update := newTestActionUpdate("action-rejected") + + // The run service reports rejections in the response body with a nil + // transport error, so the first two calls look like successes to connect. + rejected := connect.NewResponse(&workflow.RecordActionResponse{ + ActionId: update.ActionID, + Status: &status.Status{ + Code: int32(code.Code_INVALID_ARGUMENT), + Message: "unsupported action spec type: ", + }, + }) + mockClient.On("RecordAction", mock.Anything, mock.Anything). + Return(rejected, nil).Twice() + mockClient.On("RecordAction", mock.Anything, mock.Anything). + Return(connect.NewResponse(&workflow.RecordActionResponse{ + ActionId: update.ActionID, + Status: &status.Status{Code: int32(code.Code_OK)}, + }), nil).Once() + + // First event — rejected, so the action must stay retryable. + c.notifyRunService(ctx, ta, update, watch.Added) + mockClient.AssertNumberOfCalls(t, "RecordAction", 1) + + // Second event — still not recorded, so it is sent again. + c.notifyRunService(ctx, ta, update, watch.Added) + mockClient.AssertNumberOfCalls(t, "RecordAction", 2) + + // Third event — accepted this time. + c.notifyRunService(ctx, ta, update, watch.Added) + mockClient.AssertNumberOfCalls(t, "RecordAction", 3) + + // Fourth event — now recorded, so it is deduplicated. + c.notifyRunService(ctx, ta, update, watch.Added) + mockClient.AssertNumberOfCalls(t, "RecordAction", 3) +} + +func TestNotifyRunService_AcceptedRecordDeduplicates(t *testing.T) { + ctx := context.Background() + + mockClient := runmocks.NewInternalRunServiceClient(t) + + filter, err := fastcheck.NewOppoBloomFilter(128, promutils.NewTestScope()) + assert.NoError(t, err) + + c := &ActionsClient{ + runClient: mockClient, + recordedFilter: filter, + subscribers: make(map[string]map[chan *ActionUpdate]struct{}), + } + + ta, update := newTestActionUpdate("action-accepted") + + mockClient.On("RecordAction", mock.Anything, mock.Anything). + Return(connect.NewResponse(&workflow.RecordActionResponse{ + ActionId: update.ActionID, + Status: &status.Status{Code: int32(code.Code_OK)}, + }), nil).Once() + + c.notifyRunService(ctx, ta, update, watch.Added) + c.notifyRunService(ctx, ta, update, watch.Added) + + mockClient.AssertNumberOfCalls(t, "RecordAction", 1) +} + func TestNotifyRunService_UpdateActionStatusIncludesAttemptsAndCacheStatus(t *testing.T) { ctx := context.Background() @@ -143,7 +259,7 @@ func TestNotifyRunService_UpdateActionStatusIncludesAttemptsAndCacheStatus(t *te })).Return(&connect.Response[workflow.UpdateActionStatusResponse]{}, nil).Once() // First-sight MODIFIED now also records (deduped via the mandatory filter). mockClient.On("RecordAction", mock.Anything, mock.Anything). - Return(&connect.Response[workflow.RecordActionResponse]{}, nil).Maybe() + Return(acceptedRecordActionResponse(), nil).Maybe() c.notifyRunService(ctx, ta, update, watch.Modified) @@ -476,7 +592,7 @@ func TestNotifyRunService_ChildAddedPromotesParentToRunning(t *testing.T) { // Expect RecordAction for the child mockClient.On("RecordAction", mock.Anything, mock.Anything). - Return(&connect.Response[workflow.RecordActionResponse]{}, nil).Once() + Return(acceptedRecordActionResponse(), nil).Once() // Expect UpdateActionStatus for the PARENT with RUNNING phase mockClient.On("UpdateActionStatus", mock.Anything, mock.MatchedBy(func(req *connect.Request[workflow.UpdateActionStatusRequest]) bool { @@ -527,7 +643,7 @@ func TestNotifyRunService_SkipsTerminalAddedEventsOnlyWhenInBloomFilter(t *testi // First ADDED event (cold start, not in bloom filter): should process normally mockClient.On("RecordAction", mock.Anything, mock.Anything). - Return(&connect.Response[workflow.RecordActionResponse]{}, nil).Once() + Return(acceptedRecordActionResponse(), nil).Once() mockClient.On("UpdateActionStatus", mock.Anything, mock.Anything). Return(&connect.Response[workflow.UpdateActionStatusResponse]{}, nil) c.notifyRunService(ctx, ta, update, watch.Added) @@ -583,7 +699,7 @@ func TestNotifyRunService_ProcessesNonTerminalAddedEvents(t *testing.T) { // Non-terminal ADDED events should be processed normally mockClient.On("RecordAction", mock.Anything, mock.Anything). - Return(&connect.Response[workflow.RecordActionResponse]{}, nil).Once() + Return(acceptedRecordActionResponse(), nil).Once() mockClient.On("UpdateActionStatus", mock.Anything, mock.Anything). Return(&connect.Response[workflow.UpdateActionStatusResponse]{}, nil).Once() @@ -611,7 +727,7 @@ func TestNotifyRunService_DuplicateAddedSkipsRecordAction(t *testing.T) { // First call — should process normally mockClient.On("RecordAction", mock.Anything, mock.Anything). - Return(&connect.Response[workflow.RecordActionResponse]{}, nil).Once() + Return(acceptedRecordActionResponse(), nil).Once() mockClient.On("UpdateActionStatus", mock.Anything, mock.Anything). Return(&connect.Response[workflow.UpdateActionStatusResponse]{}, nil) c.notifyRunService(ctx, ta, update, watch.Added) @@ -641,7 +757,7 @@ func TestNotifyRunService_TerminalDuplicateRepairsTimestamps(t *testing.T) { // First call — should process normally (RecordAction + UpdateActionStatus) mockClient.On("RecordAction", mock.Anything, mock.Anything). - Return(&connect.Response[workflow.RecordActionResponse]{}, nil).Once() + Return(acceptedRecordActionResponse(), nil).Once() mockClient.On("UpdateActionStatus", mock.Anything, mock.Anything). Return(&connect.Response[workflow.UpdateActionStatusResponse]{}, nil).Times(2) c.notifyRunService(ctx, ta, update, watch.Added) @@ -668,7 +784,7 @@ func TestNotifyRunService_RootActionAddedDoesNotPromoteParent(t *testing.T) { ta, update := newTestActionUpdate("action-root") mockClient.On("RecordAction", mock.Anything, mock.Anything). - Return(&connect.Response[workflow.RecordActionResponse]{}, nil).Once() + Return(acceptedRecordActionResponse(), nil).Once() c.notifyRunService(ctx, ta, update, watch.Added) @@ -782,7 +898,7 @@ func TestHandleWatchEvent_CoalescedReadsLatestPhase(t *testing.T) { } mockClient.On("RecordAction", mock.Anything, mock.Anything). - Return(&connect.Response[workflow.RecordActionResponse]{}, nil).Once() + Return(acceptedRecordActionResponse(), nil).Once() // Accept exactly one status update, and ONLY if it is SUCCEEDED. A RUNNING update // would be an unexpected call and fail the test. mockClient.On("UpdateActionStatus", mock.Anything, mock.MatchedBy(func(req *connect.Request[workflow.UpdateActionStatusRequest]) bool { @@ -827,7 +943,7 @@ func TestHandleWatchEvent_CreateThenDeleteStillRecords(t *testing.T) { // Step 2: the DELETE tombstone (still carries Spec) must create the row, then abort it. mockClient.On("RecordAction", mock.Anything, mock.Anything). - Return(&connect.Response[workflow.RecordActionResponse]{}, nil).Once() + Return(acceptedRecordActionResponse(), nil).Once() mockClient.On("UpdateActionStatus", mock.Anything, mock.MatchedBy(func(req *connect.Request[workflow.UpdateActionStatusRequest]) bool { return req.Msg.GetStatus().GetPhase() == common.ActionPhase_ACTION_PHASE_ABORTED })).Return(&connect.Response[workflow.UpdateActionStatusResponse]{}, nil).Once() From 12c47644d71e28b8c58a8143b4699b467a8bed77 Mon Sep 17 00:00:00 2001 From: 1fanwang <1fannnw@gmail.com> Date: Wed, 26 Aug 2026 03:25:58 -0400 Subject: [PATCH 2/6] fix(actions): use valid RecordAction response fixtures Signed-off-by: 1fanwang <1fannnw@gmail.com> --- actions/k8s/client_test.go | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/actions/k8s/client_test.go b/actions/k8s/client_test.go index 46005dea2a..0463b5120e 100644 --- a/actions/k8s/client_test.go +++ b/actions/k8s/client_test.go @@ -58,6 +58,10 @@ func newTestActionUpdate(actionName string) (*executorv1.TaskAction, *ActionUpda func acceptedRecordActionResponse() *connect.Response[workflow.RecordActionResponse] { return connect.NewResponse(&workflow.RecordActionResponse{ + ActionId: &common.ActionIdentifier{ + Run: &common.RunIdentifier{Org: "org", Project: "proj", Domain: "dev", Name: "run"}, + Name: "action", + }, Status: &status.Status{Code: int32(code.Code_OK)}, }) } From d8aaad1c879bdbbfa3b16892110afef762bbaa37 Mon Sep 17 00:00:00 2001 From: 1fanwang <1fannnw@gmail.com> Date: Sat, 5 Sep 2026 11:34:29 -0400 Subject: [PATCH 3/6] test(actions): match RecordAction response IDs Signed-off-by: 1fanwang <1fannnw@gmail.com> --- actions/k8s/client_test.go | 32 ++++++++++++++++++-------------- 1 file changed, 18 insertions(+), 14 deletions(-) diff --git a/actions/k8s/client_test.go b/actions/k8s/client_test.go index 0463b5120e..38e90884b5 100644 --- a/actions/k8s/client_test.go +++ b/actions/k8s/client_test.go @@ -56,11 +56,15 @@ func newTestActionUpdate(actionName string) (*executorv1.TaskAction, *ActionUpda return ta, update } -func acceptedRecordActionResponse() *connect.Response[workflow.RecordActionResponse] { +func acceptedRecordActionResponse(taskAction *executorv1.TaskAction) *connect.Response[workflow.RecordActionResponse] { return connect.NewResponse(&workflow.RecordActionResponse{ ActionId: &common.ActionIdentifier{ - Run: &common.RunIdentifier{Org: "org", Project: "proj", Domain: "dev", Name: "run"}, - Name: "action", + Run: &common.RunIdentifier{ + Project: taskAction.Spec.Project, + Domain: taskAction.Spec.Domain, + Name: taskAction.Spec.RunName, + }, + Name: taskAction.Spec.ActionName, }, Status: &status.Status{Code: int32(code.Code_OK)}, }) @@ -84,7 +88,7 @@ func TestNotifyRunService_DeduplicateRecordAction(t *testing.T) { // Expect RecordAction called exactly once mockClient.On("RecordAction", mock.Anything, mock.Anything). - Return(acceptedRecordActionResponse(), nil).Once() + Return(acceptedRecordActionResponse(ta), nil).Once() // First Added event — should call RecordAction c.notifyRunService(ctx, ta, update, watch.Added) @@ -116,7 +120,7 @@ func TestNotifyRunService_FailedRecordAllowsRetry(t *testing.T) { Return((*connect.Response[workflow.RecordActionResponse])(nil), fmt.Errorf("transient error")).Once() // Second call succeeds mockClient.On("RecordAction", mock.Anything, mock.Anything). - Return(acceptedRecordActionResponse(), nil).Once() + Return(acceptedRecordActionResponse(ta), nil).Once() // First event — RecordAction fails, should NOT add to filter c.notifyRunService(ctx, ta, update, watch.Added) @@ -263,7 +267,7 @@ func TestNotifyRunService_UpdateActionStatusIncludesAttemptsAndCacheStatus(t *te })).Return(&connect.Response[workflow.UpdateActionStatusResponse]{}, nil).Once() // First-sight MODIFIED now also records (deduped via the mandatory filter). mockClient.On("RecordAction", mock.Anything, mock.Anything). - Return(acceptedRecordActionResponse(), nil).Maybe() + Return(acceptedRecordActionResponse(ta), nil).Maybe() c.notifyRunService(ctx, ta, update, watch.Modified) @@ -596,7 +600,7 @@ func TestNotifyRunService_ChildAddedPromotesParentToRunning(t *testing.T) { // Expect RecordAction for the child mockClient.On("RecordAction", mock.Anything, mock.Anything). - Return(acceptedRecordActionResponse(), nil).Once() + Return(acceptedRecordActionResponse(ta), nil).Once() // Expect UpdateActionStatus for the PARENT with RUNNING phase mockClient.On("UpdateActionStatus", mock.Anything, mock.MatchedBy(func(req *connect.Request[workflow.UpdateActionStatusRequest]) bool { @@ -647,7 +651,7 @@ func TestNotifyRunService_SkipsTerminalAddedEventsOnlyWhenInBloomFilter(t *testi // First ADDED event (cold start, not in bloom filter): should process normally mockClient.On("RecordAction", mock.Anything, mock.Anything). - Return(acceptedRecordActionResponse(), nil).Once() + Return(acceptedRecordActionResponse(ta), nil).Once() mockClient.On("UpdateActionStatus", mock.Anything, mock.Anything). Return(&connect.Response[workflow.UpdateActionStatusResponse]{}, nil) c.notifyRunService(ctx, ta, update, watch.Added) @@ -703,7 +707,7 @@ func TestNotifyRunService_ProcessesNonTerminalAddedEvents(t *testing.T) { // Non-terminal ADDED events should be processed normally mockClient.On("RecordAction", mock.Anything, mock.Anything). - Return(acceptedRecordActionResponse(), nil).Once() + Return(acceptedRecordActionResponse(ta), nil).Once() mockClient.On("UpdateActionStatus", mock.Anything, mock.Anything). Return(&connect.Response[workflow.UpdateActionStatusResponse]{}, nil).Once() @@ -731,7 +735,7 @@ func TestNotifyRunService_DuplicateAddedSkipsRecordAction(t *testing.T) { // First call — should process normally mockClient.On("RecordAction", mock.Anything, mock.Anything). - Return(acceptedRecordActionResponse(), nil).Once() + Return(acceptedRecordActionResponse(ta), nil).Once() mockClient.On("UpdateActionStatus", mock.Anything, mock.Anything). Return(&connect.Response[workflow.UpdateActionStatusResponse]{}, nil) c.notifyRunService(ctx, ta, update, watch.Added) @@ -761,7 +765,7 @@ func TestNotifyRunService_TerminalDuplicateRepairsTimestamps(t *testing.T) { // First call — should process normally (RecordAction + UpdateActionStatus) mockClient.On("RecordAction", mock.Anything, mock.Anything). - Return(acceptedRecordActionResponse(), nil).Once() + Return(acceptedRecordActionResponse(ta), nil).Once() mockClient.On("UpdateActionStatus", mock.Anything, mock.Anything). Return(&connect.Response[workflow.UpdateActionStatusResponse]{}, nil).Times(2) c.notifyRunService(ctx, ta, update, watch.Added) @@ -788,7 +792,7 @@ func TestNotifyRunService_RootActionAddedDoesNotPromoteParent(t *testing.T) { ta, update := newTestActionUpdate("action-root") mockClient.On("RecordAction", mock.Anything, mock.Anything). - Return(acceptedRecordActionResponse(), nil).Once() + Return(acceptedRecordActionResponse(ta), nil).Once() c.notifyRunService(ctx, ta, update, watch.Added) @@ -902,7 +906,7 @@ func TestHandleWatchEvent_CoalescedReadsLatestPhase(t *testing.T) { } mockClient.On("RecordAction", mock.Anything, mock.Anything). - Return(acceptedRecordActionResponse(), nil).Once() + Return(acceptedRecordActionResponse(succeededTaskAction("a3")), nil).Once() // Accept exactly one status update, and ONLY if it is SUCCEEDED. A RUNNING update // would be an unexpected call and fail the test. mockClient.On("UpdateActionStatus", mock.Anything, mock.MatchedBy(func(req *connect.Request[workflow.UpdateActionStatusRequest]) bool { @@ -947,7 +951,7 @@ func TestHandleWatchEvent_CreateThenDeleteStillRecords(t *testing.T) { // Step 2: the DELETE tombstone (still carries Spec) must create the row, then abort it. mockClient.On("RecordAction", mock.Anything, mock.Anything). - Return(acceptedRecordActionResponse(), nil).Once() + Return(acceptedRecordActionResponse(runningTaskAction("d1")), nil).Once() mockClient.On("UpdateActionStatus", mock.Anything, mock.MatchedBy(func(req *connect.Request[workflow.UpdateActionStatusRequest]) bool { return req.Msg.GetStatus().GetPhase() == common.ActionPhase_ACTION_PHASE_ABORTED })).Return(&connect.Response[workflow.UpdateActionStatusResponse]{}, nil).Once() From 60d0815cbe70e315ac981dbee79220f30bbe3036 Mon Sep 17 00:00:00 2001 From: 1fanwang <1fannnw@gmail.com> Date: Wed, 9 Sep 2026 03:29:24 -0700 Subject: [PATCH 4/6] test(actions): model a repository failure response Signed-off-by: 1fanwang <1fannnw@gmail.com> --- actions/k8s/client_test.go | 16 ++++++++-------- 1 file changed, 8 insertions(+), 8 deletions(-) diff --git a/actions/k8s/client_test.go b/actions/k8s/client_test.go index 38e90884b5..3706e6581f 100644 --- a/actions/k8s/client_test.go +++ b/actions/k8s/client_test.go @@ -166,7 +166,7 @@ func TestNotifyRunService_MissingResponseAllowsRetry(t *testing.T) { } } -func TestNotifyRunService_RejectedRecordAllowsRetry(t *testing.T) { +func TestNotifyRunService_InternalFailureAllowsRetry(t *testing.T) { ctx := context.Background() mockClient := runmocks.NewInternalRunServiceClient(t) @@ -180,26 +180,26 @@ func TestNotifyRunService_RejectedRecordAllowsRetry(t *testing.T) { subscribers: make(map[string]map[chan *ActionUpdate]struct{}), } - ta, update := newTestActionUpdate("action-rejected") + ta, update := newTestActionUpdate("action-internal-failure") - // The run service reports rejections in the response body with a nil + // The run service reports repository failures in the response body with a nil // transport error, so the first two calls look like successes to connect. - rejected := connect.NewResponse(&workflow.RecordActionResponse{ + failed := connect.NewResponse(&workflow.RecordActionResponse{ ActionId: update.ActionID, Status: &status.Status{ - Code: int32(code.Code_INVALID_ARGUMENT), - Message: "unsupported action spec type: ", + Code: int32(code.Code_INTERNAL), + Message: "failed to create action: transient database error", }, }) mockClient.On("RecordAction", mock.Anything, mock.Anything). - Return(rejected, nil).Twice() + Return(failed, nil).Twice() mockClient.On("RecordAction", mock.Anything, mock.Anything). Return(connect.NewResponse(&workflow.RecordActionResponse{ ActionId: update.ActionID, Status: &status.Status{Code: int32(code.Code_OK)}, }), nil).Once() - // First event — rejected, so the action must stay retryable. + // First event fails, so the action must stay retryable. c.notifyRunService(ctx, ta, update, watch.Added) mockClient.AssertNumberOfCalls(t, "RecordAction", 1) From 55c61bb17d23c9ce688ad1b6e2535312e7e319db Mon Sep 17 00:00:00 2001 From: 1fanwang <1fannnw@gmail.com> Date: Wed, 9 Sep 2026 03:37:08 -0700 Subject: [PATCH 5/6] test(runs): pin RecordAction repository error contract Signed-off-by: 1fanwang <1fannnw@gmail.com> --- runs/test/api/record_action_status_test.go | 59 ++++++++++++++++++++++ 1 file changed, 59 insertions(+) create mode 100644 runs/test/api/record_action_status_test.go diff --git a/runs/test/api/record_action_status_test.go b/runs/test/api/record_action_status_test.go new file mode 100644 index 0000000000..0cd7511a12 --- /dev/null +++ b/runs/test/api/record_action_status_test.go @@ -0,0 +1,59 @@ +package api + +import ( + "context" + "testing" + + "connectrpc.com/connect" + "github.com/stretchr/testify/require" + "google.golang.org/genproto/googleapis/rpc/code" + + "github.com/flyteorg/flyte/v2/gen/go/flyteidl2/common" + "github.com/flyteorg/flyte/v2/gen/go/flyteidl2/core" + "github.com/flyteorg/flyte/v2/gen/go/flyteidl2/task" + "github.com/flyteorg/flyte/v2/gen/go/flyteidl2/workflow" + "github.com/flyteorg/flyte/v2/gen/go/flyteidl2/workflow/workflowconnect" +) + +func TestRecordActionReturnsRepositoryFailureInBody(t *testing.T) { + t.Cleanup(func() { cleanupTestDB(t) }) + + ctx := context.Background() + require.NoError(t, testDB.PingContext(ctx)) + require.NoError(t, renameActionsTable(ctx, "actions", "actions_unavailable")) + t.Cleanup(func() { + require.NoError(t, renameActionsTable(context.Background(), "actions_unavailable", "actions")) + }) + + client := workflowconnect.NewInternalRunServiceClient(newClient(), endpoint) + response, err := client.RecordAction(ctx, connect.NewRequest(&workflow.RecordActionRequest{ + ActionId: &common.ActionIdentifier{ + Run: &common.RunIdentifier{ + Org: testOrg, + Project: testProject, + Domain: testDomain, + Name: "r" + uniqueString(), + }, + Name: "record-action-db-failure", + }, + Spec: &workflow.RecordActionRequest_Task{ + Task: &workflow.TaskAction{ + Spec: &task.TaskSpec{TaskTemplate: &core.TaskTemplate{Type: "python"}}, + }, + }, + })) + + require.NoError(t, err) + require.Equal(t, int32(code.Code_INTERNAL), response.Msg.GetStatus().GetCode()) + t.Logf( + "DB-FAILURE transportErr=%v status.code=%d status.message=%q", + err, + response.Msg.GetStatus().GetCode(), + response.Msg.GetStatus().GetMessage(), + ) +} + +func renameActionsTable(ctx context.Context, from string, to string) error { + _, err := testDB.ExecContext(ctx, "ALTER TABLE "+from+" RENAME TO "+to) + return err +} From cdb8d0d6b0a4c0fde5637b0fd1adc53f6c6bc7f7 Mon Sep 17 00:00:00 2001 From: 1fanwang <1fannnw@gmail.com> Date: Wed, 9 Sep 2026 03:58:03 -0700 Subject: [PATCH 6/6] refactor(actions): clarify RecordAction retry comment and alias rpc code import Alias the googleapis rpc code package so it no longer shadows the local code variables in this file, and correct the comment to say a rejected action is memoized as recorded rather than never recorded. Document why the test table rename concatenates its identifiers. Signed-off-by: 1fanwang <1fannnw@gmail.com> --- actions/k8s/client.go | 11 ++++++----- runs/test/api/record_action_status_test.go | 2 ++ 2 files changed, 8 insertions(+), 5 deletions(-) diff --git a/actions/k8s/client.go b/actions/k8s/client.go index 85c7740ea3..5c3e21237f 100644 --- a/actions/k8s/client.go +++ b/actions/k8s/client.go @@ -9,7 +9,7 @@ import ( "time" "connectrpc.com/connect" - "google.golang.org/genproto/googleapis/rpc/code" + rpccode "google.golang.org/genproto/googleapis/rpc/code" "google.golang.org/protobuf/proto" "google.golang.org/protobuf/types/known/timestamppb" apierrors "k8s.io/apimachinery/pkg/api/errors" @@ -783,15 +783,16 @@ func (c *ActionsClient) notifyRunService(ctx context.Context, taskAction *execut Task: ta, } } - // RecordAction reports rejections in the response body, not as a transport - // error, so the body status has to be checked before memoizing the key — - // otherwise a rejected action is never recorded and never retried. + // RecordAction reports rejections in the response body rather than as a + // transport error, so the body status has to be checked before memoizing + // the key. Otherwise a rejected action is memoized as recorded and is + // never retried. resp, err := c.runClient.RecordAction(ctx, connect.NewRequest(recordReq)) if err != nil { logger.Warnf(ctx, "Failed to record action in run service for %s: %v", update.ActionID.Name, err) } else if resp == nil || resp.Msg == nil || resp.Msg.GetStatus() == nil { logger.Warnf(ctx, "Run service returned no RecordAction status for %s", update.ActionID.Name) - } else if status := resp.Msg.GetStatus(); status.GetCode() != int32(code.Code_OK) { + } else if status := resp.Msg.GetStatus(); status.GetCode() != int32(rpccode.Code_OK) { logger.Warnf(ctx, "Run service rejected RecordAction for %s with code %d: %s", update.ActionID.Name, status.GetCode(), status.GetMessage()) } else { diff --git a/runs/test/api/record_action_status_test.go b/runs/test/api/record_action_status_test.go index 0cd7511a12..ec4d0d8dea 100644 --- a/runs/test/api/record_action_status_test.go +++ b/runs/test/api/record_action_status_test.go @@ -54,6 +54,8 @@ func TestRecordActionReturnsRepositoryFailureInBody(t *testing.T) { } func renameActionsTable(ctx context.Context, from string, to string) error { + // Table identifiers cannot be bind parameters, so the names are concatenated. + // Both callers pass string literals, so nothing here comes from user input. _, err := testDB.ExecContext(ctx, "ALTER TABLE "+from+" RENAME TO "+to) return err }