diff --git a/actions/k8s/client.go b/actions/k8s/client.go index 03090bb347c..5c3e21237f7 100644 --- a/actions/k8s/client.go +++ b/actions/k8s/client.go @@ -9,6 +9,7 @@ import ( "time" "connectrpc.com/connect" + 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" @@ -782,8 +783,18 @@ 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 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(rpccode.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 fa1079c8a47..3706e6581f3 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,20 @@ func newTestActionUpdate(actionName string) (*executorv1.TaskAction, *ActionUpda return ta, update } +func acceptedRecordActionResponse(taskAction *executorv1.TaskAction) *connect.Response[workflow.RecordActionResponse] { + return connect.NewResponse(&workflow.RecordActionResponse{ + ActionId: &common.ActionIdentifier{ + 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)}, + }) +} + func TestNotifyRunService_DeduplicateRecordAction(t *testing.T) { ctx := context.Background() @@ -73,7 +88,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(ta), nil).Once() // First Added event — should call RecordAction c.notifyRunService(ctx, ta, update, watch.Added) @@ -105,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(&connect.Response[workflow.RecordActionResponse]{}, nil).Once() + Return(acceptedRecordActionResponse(ta), nil).Once() // First event — RecordAction fails, should NOT add to filter c.notifyRunService(ctx, ta, update, watch.Added) @@ -120,6 +135,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_InternalFailureAllowsRetry(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-internal-failure") + + // 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. + failed := connect.NewResponse(&workflow.RecordActionResponse{ + ActionId: update.ActionID, + Status: &status.Status{ + Code: int32(code.Code_INTERNAL), + Message: "failed to create action: transient database error", + }, + }) + mockClient.On("RecordAction", mock.Anything, mock.Anything). + 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 fails, 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 +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(&connect.Response[workflow.RecordActionResponse]{}, nil).Maybe() + Return(acceptedRecordActionResponse(ta), nil).Maybe() c.notifyRunService(ctx, ta, update, watch.Modified) @@ -476,7 +600,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(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 { @@ -527,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(&connect.Response[workflow.RecordActionResponse]{}, 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) @@ -583,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(&connect.Response[workflow.RecordActionResponse]{}, nil).Once() + Return(acceptedRecordActionResponse(ta), nil).Once() mockClient.On("UpdateActionStatus", mock.Anything, mock.Anything). Return(&connect.Response[workflow.UpdateActionStatusResponse]{}, nil).Once() @@ -611,7 +735,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(ta), nil).Once() mockClient.On("UpdateActionStatus", mock.Anything, mock.Anything). Return(&connect.Response[workflow.UpdateActionStatusResponse]{}, nil) c.notifyRunService(ctx, ta, update, watch.Added) @@ -641,7 +765,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(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) @@ -668,7 +792,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(ta), nil).Once() c.notifyRunService(ctx, ta, update, watch.Added) @@ -782,7 +906,7 @@ func TestHandleWatchEvent_CoalescedReadsLatestPhase(t *testing.T) { } mockClient.On("RecordAction", mock.Anything, mock.Anything). - Return(&connect.Response[workflow.RecordActionResponse]{}, 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 { @@ -827,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(&connect.Response[workflow.RecordActionResponse]{}, 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() 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 00000000000..ec4d0d8dea8 --- /dev/null +++ b/runs/test/api/record_action_status_test.go @@ -0,0 +1,61 @@ +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 { + // 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 +}