fix(executor): fail immediately on deterministic plugin errors

[davidlin20dev]

What changes were proposed in this pull request?

Plugin errors carrying a deterministic error code now fail the TaskAction immediately instead of consuming system-failure attempts.

var nonRetryableErrorCodes = []stdErrors.ErrorCode{
	pluginserrors.BadTaskSpecification,
	pluginserrors.MetadataTooLarge,
}

recordSystemError checks the error's code before touching Status.SystemFailures, and routes a match straight to finalizePermanentFailure with ExecutionError_USER and the original code and message.

A few notes:

• The check runs before the counter is incremented, so a failure that is never retried doesn't leave a misleading SystemFailures count behind.
• Kind is USER, since both codes mean the task spec or the workload is at fault rather than the platform. Easy to change if you'd rather they were SYSTEM.
• This only catches errors returned directly by a plugin. Errors built from a failure phase go through systemErrorFromPhaseInfo, which flattens the code into the message text, so IsCausedBy can't see it. Noted in a comment where the codes are defined.
• MetadataTooLarge is not produced anywhere in the codebase today, so it's included defensively rather than tested against a real producer.
• CorruptedPluginState is deliberately left out. It looks recoverable rather than deterministic, and I'd like to handle it separately. Details in a comment below.

Why are the changes needed?

Follow-up to #7799, where we noticed the v2 executor doesn't check plugin error codes: every plugin error goes through recordSystemError and is retried until MaxSystemFailures is exceeded, regardless of whether retrying could possibly help.

Concretely, a typo in the Ray plugin config (submissionMode: HttpMode) currently produces four attempts at 10-second intervals before failing, and reports as MaxSystemFailuresExceeded rather than naming the actual problem. With this change it fails in about a second, and the failure carries BadTaskSpecification and the original message, which points at the config.

How was this patch tested?

• New spec in taskaction_controller_test.go asserting that a BadTaskSpecification error terminates immediately: no requeue, SystemFailures stays zero, the user's Attempts budget is untouched, and Status.ErrorState carries the code, USER kind, and the original message.
• Verified the spec catches regressions: swapping the code in nonRetryableErrorCodes for one the test doesn't use makes it fail.
• Existing recordSystemError specs still pass. They use a plain error with no typed code, so the new check correctly ignores them.

[davidlin20dev]

Hi @pingsutw, when I was implementing this I left CorruptedPluginState out, and after tracing it I think it needs its own discussion. Wanted to check with you first.

Tracing it in plugin_manager.go:

flyte/executor/pkg/plugin/k8s/plugin_manager.go

Lines 221 to 227 in d2c1795

	if v, err := tCtx.PluginStateReader().Get(&pluginState); err != nil {
	if v != pluginStateVersion {
	return pluginsCore.DoTransition(pluginsCore.PhaseInfoRetryableFailure(errors.CorruptedPluginState,
	fmt.Sprintf("plugin state version mismatch expected [%d] got [%d]", pluginStateVersion, v), nil)), nil
	}
	return pluginsCore.UnknownTransition, errors.Wrapf(errors.CorruptedPluginState, err, "Failed to read unmarshal custom state")
	}

The state reader returns version 0 on any decode failure:

flyte/executor/pkg/plugin/state_manager.go

Lines 43 to 52 in d2c1795

	func (m *PluginStateManager) Get(t interface{}) (uint8, error) {
	if len(m.prevStateBytes) == 0 {
	return 0, nil
	}
	buf := bytes.NewBuffer(m.prevStateBytes)
	dec := gob.NewDecoder(buf)
	if err := dec.Decode(t); err != nil {
	return 0, fmt.Errorf("failed to decode plugin state: %w", err)
	}
	return m.prevStateVersion, nil

Since pluginStateVersion is 1, v != pluginStateVersion is always true whenever there's an error. So the version-mismatch branch always wins, and the errors.Wrapf line below it looks unreachable. (A side effect: a real decode failure gets logged as "version mismatch expected [1] got [0]" and the underlying error is swallowed.)

That also means adding CorruptedPluginState to the non-retryable list in this PR would be a no-op, since the only line that returns it as an error can't be reached.

What the reachable branch does: it's a PhaseInfoRetryableFailure, Kind USER, so it goes down the in-place restart path, which clears the plugin state before retrying:

flyte/executor/pkg/controller/taskaction_controller.go

Lines 468 to 473 in d2c1795

	// If an in-place restart was triggered, increment attempts and clear plugin state so the
	// next reconcile starts fresh with PluginPhaseNotStarted and creates a new pod.
	if restartAttempts > 0 {
	taskAction.Status.Attempts = restartAttempts
	taskAction.Status.PluginState = nil
	taskAction.Status.PluginStateVersion = 0

So when the task has retries configured, corrupt state actually recovers, the next attempt starts fresh. But with default retries (0), maxAttempts is 1, so the first corrupted-state event converts straight to a permanent failure, attributed to the user, even though a bad state blob is a platform problem rather than the task's fault.

That's why I think the real fix here is PhaseInfoSystemRetryableFailure instead: it would route through resetPluginResource, which also clears the state, so recovery still works, but it stops consuming the user's retry budget and recovers even for tasks with no retries configured (bounded by MaxSystemFailures).

Let me know if this makes sense, not sure if I'm missing something here.

[pingsutw]

Thanks for tracing the code. we can change pluginsCore.PhaseInfoRetryableFailure to PhaseInfoFailureWithCleanup, so the task won't retry on this error

[davidlin20dev]

thanks for the review! I'll make the change in this PR.

One small question about PhaseInfoFailureWithCleanup: it marks the failure as Kind: USER. But the plugin state is written and read by the executor itself, the task never touches it, so corrupted state feels more like a system error than a user error. Should we use PhaseInfoSystemFailureWithCleanup instead? It fails permanently either way, just changes who the error gets attributed to.

[davidlin20dev]

Just pushed the change. I went with PhaseInfoSystemFailureWithCleanup from my question above, it's a one-word swap back if needed. Let me know if you'd rather keep USER. Thanks!

[pingsutw]

LGTM, thanks