fix(actions): retry rejected action records

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)
 		}

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()

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
+}