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>

Author: 1fanwang

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

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: <nil>",
+		},
+	})
+	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()