Skip to content

Commit 27f9b46

Browse files
shaon-chowdhury-euccursoragent
shaon-chowdhury-euc
and
cursoragent
committed
fix(executor): enforce TaskMetadata.timeout as a per-attempt max runtime
The SDK already serializes max_runtime onto the TaskAction, but the executor never read it, so hung tasks ran to success. Abort and finalize the plugin at the deadline, emit TIMED_OUT, and retry only within the existing attempt budget. Closes #7901 Signed-off-by: shaon-chowdhury-euc <shaon.chowdhury@eucalyptus.vc> Co-authored-by: Cursor <cursoragent@cursor.com>
1 parent 0158ca4 commit 27f9b46

13 files changed

Lines changed: 1510 additions & 24 deletions

File tree

‎actions/service/actions_service.go‎

Lines changed: 6 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -154,7 +154,9 @@ func (s *ActionsService) WatchForUpdates(
154154
OutputUri: update.OutputUri,
155155
Value: update.SignalValue,
156156
}
157-
if update.Phase == common.ActionPhase_ACTION_PHASE_FAILED && update.ErrorState != nil {
157+
if (update.Phase == common.ActionPhase_ACTION_PHASE_FAILED ||
158+
update.Phase == common.ActionPhase_ACTION_PHASE_TIMED_OUT) &&
159+
update.ErrorState != nil {
158160
au.Error = errorStateToExecutionError(update.ErrorState)
159161
}
160162
resp := &actions.WatchForUpdatesResponse{
@@ -228,7 +230,9 @@ func taskActionToUpdate(ctx context.Context, action *executorv1.TaskAction) *wor
228230
OutputUri: k8s.BuildOutputUri(ctx, action),
229231
Value: k8s.SignalValueFromStatus(ctx, action),
230232
}
231-
if phase == common.ActionPhase_ACTION_PHASE_FAILED && action.Status.ErrorState != nil {
233+
if (phase == common.ActionPhase_ACTION_PHASE_FAILED ||
234+
phase == common.ActionPhase_ACTION_PHASE_TIMED_OUT) &&
235+
action.Status.ErrorState != nil {
232236
update.Error = errorStateToExecutionError(action.Status.ErrorState)
233237
}
234238
return update

‎actions/service/actions_service_test.go‎

Lines changed: 30 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -249,6 +249,36 @@ func TestTaskActionToUpdate_PopulatesErrorOnFailure(t *testing.T) {
249249
}
250250
}
251251

252+
func TestTaskActionToUpdate_PopulatesErrorOnTimeout(t *testing.T) {
253+
ta := &executorv1.TaskAction{
254+
Spec: executorv1.TaskActionSpec{
255+
Project: "flytesnacks", Domain: "development", RunName: "r1", ActionName: "a0",
256+
RunOutputBase: "s3://bucket/run",
257+
},
258+
Status: executorv1.TaskActionStatus{
259+
Conditions: []metav1.Condition{
260+
{
261+
Type: string(executorv1.ConditionTypeFailed),
262+
Status: metav1.ConditionTrue,
263+
Reason: string(executorv1.ConditionReasonTimedOut),
264+
},
265+
},
266+
ErrorState: &executorv1.ErrorState{
267+
Code: "TaskExecutionTimedOut", Kind: "USER", Message: "max runtime exceeded",
268+
},
269+
},
270+
}
271+
272+
upd := taskActionToUpdate(context.Background(), ta)
273+
274+
assert.Equal(t, common.ActionPhase_ACTION_PHASE_TIMED_OUT, upd.Phase)
275+
if assert.NotNil(t, upd.Error) {
276+
assert.Equal(t, "TaskExecutionTimedOut", upd.Error.Code)
277+
assert.Equal(t, core.ExecutionError_USER, upd.Error.Kind)
278+
assert.Equal(t, "max runtime exceeded", upd.Error.Message)
279+
}
280+
}
281+
252282
func TestTaskActionToUpdate_NoErrorWhenNotFailed(t *testing.T) {
253283
ta := &executorv1.TaskAction{
254284
Spec: executorv1.TaskActionSpec{

‎charts/flyte-binary/templates/crds/flyte.org_taskactions.yaml‎

Lines changed: 14 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -164,6 +164,13 @@ spec:
164164
status:
165165
description: status defines the observed state of TaskAction
166166
properties:
167+
attemptStartedAt:
168+
description: |-
169+
AttemptStartedAt is when the current user attempt first entered the Running phase.
170+
It is persisted independently from ActionEvent publication so max-runtime
171+
enforcement survives controller and event-service outages.
172+
format: date-time
173+
type: string
167174
attempts:
168175
description: Attempts is the latest observed action attempt number,
169176
starting from 1.
@@ -333,6 +340,13 @@ spec:
333340
maximum, the TaskAction is converted to a permanent failure.
334341
format: int32
335342
type: integer
343+
timeoutAt:
344+
description: |-
345+
TimeoutAt is the max-runtime deadline for an expired current attempt.
346+
While set on a non-terminal action, timeout cleanup is pending and the
347+
plugin resource must not be handled or recreated.
348+
format: date-time
349+
type: string
336350
type: object
337351
required:
338352
- spec

‎docker/devbox-bundled/manifests/complete.yaml‎

Lines changed: 14 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -6906,6 +6906,13 @@ spec:
69066906
status:
69076907
description: status defines the observed state of TaskAction
69086908
properties:
6909+
attemptStartedAt:
6910+
description: |-
6911+
AttemptStartedAt is when the current user attempt first entered the Running phase.
6912+
It is persisted independently from ActionEvent publication so max-runtime
6913+
enforcement survives controller and event-service outages.
6914+
format: date-time
6915+
type: string
69096916
attempts:
69106917
description: Attempts is the latest observed action attempt number,
69116918
starting from 1.
@@ -7075,6 +7082,13 @@ spec:
70757082
maximum, the TaskAction is converted to a permanent failure.
70767083
format: int32
70777084
type: integer
7085+
timeoutAt:
7086+
description: |-
7087+
TimeoutAt is the max-runtime deadline for an expired current attempt.
7088+
While set on a non-terminal action, timeout cleanup is pending and the
7089+
plugin resource must not be handled or recreated.
7090+
format: date-time
7091+
type: string
70787092
type: object
70797093
required:
70807094
- spec

‎docker/devbox-bundled/manifests/dev.yaml‎

Lines changed: 14 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -6906,6 +6906,13 @@ spec:
69066906
status:
69076907
description: status defines the observed state of TaskAction
69086908
properties:
6909+
attemptStartedAt:
6910+
description: |-
6911+
AttemptStartedAt is when the current user attempt first entered the Running phase.
6912+
It is persisted independently from ActionEvent publication so max-runtime
6913+
enforcement survives controller and event-service outages.
6914+
format: date-time
6915+
type: string
69096916
attempts:
69106917
description: Attempts is the latest observed action attempt number,
69116918
starting from 1.
@@ -7075,6 +7082,13 @@ spec:
70757082
maximum, the TaskAction is converted to a permanent failure.
70767083
format: int32
70777084
type: integer
7085+
timeoutAt:
7086+
description: |-
7087+
TimeoutAt is the max-runtime deadline for an expired current attempt.
7088+
While set on a non-terminal action, timeout cleanup is pending and the
7089+
plugin resource must not be handled or recreated.
7090+
format: date-time
7091+
type: string
70787092
type: object
70797093
required:
70807094
- spec

‎executor/api/v1/taskaction_types.go‎

Lines changed: 15 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -88,7 +88,9 @@ const (
8888
// ConditionReasonSignaled indicates a condition action received its signal (Succeeded=True)
8989
ConditionReasonSignaled TaskActionConditionReason = "Signaled"
9090

91-
// ConditionReasonTimedOut indicates a condition action passed its deadline unsignalled (Failed=True)
91+
// ConditionReasonTimedOut indicates the action reached its deadline (Failed=True).
92+
// This applies to both condition actions that were not signalled in time and to
93+
// task actions whose per-attempt max-runtime was exhausted.
9294
ConditionReasonTimedOut TaskActionConditionReason = "TimedOut"
9395
)
9496

@@ -294,6 +296,18 @@ type TaskActionStatus struct {
294296
// +optional
295297
Attempts uint32 `json:"attempts,omitempty"`
296298

299+
// AttemptStartedAt is when the current user attempt first entered the Running phase.
300+
// It is persisted independently from ActionEvent publication so max-runtime
301+
// enforcement survives controller and event-service outages.
302+
// +optional
303+
AttemptStartedAt *metav1.Time `json:"attemptStartedAt,omitempty"`
304+
305+
// TimeoutAt is the max-runtime deadline for an expired current attempt.
306+
// While set on a non-terminal action, timeout cleanup is pending and the
307+
// plugin resource must not be handled or recreated.
308+
// +optional
309+
TimeoutAt *metav1.Time `json:"timeoutAt,omitempty"`
310+
297311
// SystemFailures counts system-level failures observed during reconciliation —
298312
// either Go errors returned from Plugin.Handle (e.g. transient k8s API errors,
299313
// admission webhook denials) or plugin transitions reporting a system-retryable

‎executor/api/v1/zz_generated.deepcopy.go‎

Lines changed: 8 additions & 0 deletions
Some generated files are not rendered by default. Learn more about customizing how changed files appear on GitHub.

‎executor/config/crd/bases/flyte.org_taskactions.yaml‎

Lines changed: 14 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -164,6 +164,13 @@ spec:
164164
status:
165165
description: status defines the observed state of TaskAction
166166
properties:
167+
attemptStartedAt:
168+
description: |-
169+
AttemptStartedAt is when the current user attempt first entered the Running phase.
170+
It is persisted independently from ActionEvent publication so max-runtime
171+
enforcement survives controller and event-service outages.
172+
format: date-time
173+
type: string
167174
attempts:
168175
description: Attempts is the latest observed action attempt number,
169176
starting from 1.
@@ -333,6 +340,13 @@ spec:
333340
maximum, the TaskAction is converted to a permanent failure.
334341
format: int32
335342
type: integer
343+
timeoutAt:
344+
description: |-
345+
TimeoutAt is the max-runtime deadline for an expired current attempt.
346+
While set on a non-terminal action, timeout cleanup is pending and the
347+
plugin resource must not be handled or recreated.
348+
format: date-time
349+
type: string
336350
type: object
337351
required:
338352
- spec

‎executor/pkg/controller/taskaction_cache.go‎

Lines changed: 30 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -182,6 +182,36 @@ func (r *TaskActionReconciler) releaseCacheReservation(ctx context.Context, cach
182182
return r.Catalog.ReleaseReservation(ctx, cacheCfg.key, cacheCfg.ownerID)
183183
}
184184

185+
func (r *TaskActionReconciler) maintainCacheReservation(
186+
ctx context.Context,
187+
taskAction *flyteorgv1.TaskAction,
188+
tCtx pluginsCore.TaskExecutionContext,
189+
) error {
190+
cacheCfg, ok, err := buildTaskCacheConfig(ctx, taskAction, tCtx)
191+
if err != nil || !ok || r.Catalog == nil || !cacheCfg.serializable {
192+
return err
193+
}
194+
_, err = r.Catalog.GetOrExtendReservation(
195+
ctx,
196+
cacheCfg.key,
197+
cacheCfg.ownerID,
198+
r.cacheReservationHeartbeat(),
199+
)
200+
return err
201+
}
202+
203+
func (r *TaskActionReconciler) releaseTaskCacheReservation(
204+
ctx context.Context,
205+
taskAction *flyteorgv1.TaskAction,
206+
tCtx pluginsCore.TaskExecutionContext,
207+
) error {
208+
cacheCfg, ok, err := buildTaskCacheConfig(ctx, taskAction, tCtx)
209+
if err != nil || !ok {
210+
return err
211+
}
212+
return r.releaseCacheReservation(ctx, cacheCfg)
213+
}
214+
185215
func cacheMetadataForUpload(tCtx pluginsCore.TaskExecutionContext, taskID *corepb.Identifier) catalog.Metadata {
186216
taskExecID := proto.Clone(tCtx.TaskExecutionMetadata().GetTaskExecutionID().GetID()).(*corepb.TaskExecutionIdentifier)
187217
taskExecID.TaskId = proto.Clone(taskID).(*corepb.Identifier)

‎executor/pkg/controller/taskaction_cache_test.go‎

Lines changed: 34 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -304,6 +304,40 @@ func TestHandleCacheAfterExecutionReleasesReservationOnFailure(t *testing.T) {
304304
assert.True(t, released)
305305
}
306306

307+
func TestTimeoutMaintainsSerializedCacheReservationUntilTerminal(t *testing.T) {
308+
ensureTestMetricKeys()
309+
ctx := context.Background()
310+
taskAction, dataStore := newCacheableTaskAction(t, true, true)
311+
tCtx := newTaskExecutionContext(t, taskAction, dataStore)
312+
313+
extended := 0
314+
released := 0
315+
r := &TaskActionReconciler{
316+
DataStore: dataStore,
317+
Catalog: &stubCatalogClient{
318+
getOrExtendReservationFunc: func(_ context.Context, _ catalog.Key, ownerID string, heartbeat time.Duration) (*cacheservice.Reservation, error) {
319+
extended++
320+
assert.Equal(t, "default/cacheable-action", ownerID)
321+
assert.Equal(t, TaskActionDefaultRequeueDuration, heartbeat)
322+
return &cacheservice.Reservation{OwnerId: ownerID}, nil
323+
},
324+
releaseReservationFunc: func(_ context.Context, _ catalog.Key, ownerID string) error {
325+
released++
326+
assert.Equal(t, "default/cacheable-action", ownerID)
327+
return nil
328+
},
329+
},
330+
}
331+
332+
require.NoError(t, r.maintainCacheReservation(ctx, taskAction, tCtx))
333+
require.NoError(t, r.maintainCacheReservation(ctx, taskAction, tCtx))
334+
assert.Equal(t, 2, extended)
335+
assert.Zero(t, released)
336+
337+
require.NoError(t, r.releaseTaskCacheReservation(ctx, taskAction, tCtx))
338+
assert.Equal(t, 1, released)
339+
}
340+
307341
func newCacheableTaskAction(t *testing.T, discoverable bool, serializable bool) (*flyteorgv1.TaskAction, *storage.DataStore) {
308342
t.Helper()
309343

0 commit comments

Comments
 (0)