[akritishakya]

Description

Every failure exited with the same code, so the runner (olake-ui worker, k8s, cron) could not tell "a restart will fix it" from "a person must fix something first". It kept restarting syncs that could never succeed, and nobody was alerted.

This PR gives failures that need manual intervention their own exit code (3) and splits the error markers so each failure is classified once.

Fixes #1196

Exit codes

Exit	Meaning	Runner should
0	success	–
1	failed, a restart can fix it	restart
2	Go crashed (set by the Go runtime)	restart
3	a person must fix something	stop and alert

3 is used instead of 2 because Go already exits 2 on a crash.

Error markers

Marker	Retry in the same run?	Exit
none (default)	yes, a few times	1
ErrRetryable	no: this run is broken, but a restart works	1
ErrNonRetryable	no: a restart won't help either	3

ErrNonRetryable wraps ErrRetryable, so IsRetryable stops the retry loop for both, and IsNonRetryable picks exit 3 in connector.go.

⚠️ ErrNonRetryable changed meaning. It used to mean "stop in-process retries" (exit 1) and now means "manual intervention" (exit 3). Its old uses moved to ErrRetryable. Please check any open PR that uses ErrNonRetryable.

What is marked

• ErrNonRetryable (39 sites): broken saved state or destination metadata; CDC not set up (publication, replication slot, binlog, CDC on a table); lost CDC position; no data within initial_wait_time; Kafka offsets that need "clear destination".
• ErrRetryable (10 sites): Kafka rebalance, reader restart, missing partition metadata; Postgres LSN not confirmed in time; backfill planning; sampled discover.
• Discover's ProduceSchema errors are no longer marked, so they are retried again.
• pgoutput: a silent connection after data has arrived is now treated as a lost connection (retried, exit 1), not "no records found" (exit 3).

Merge with staging

• produceSampledSchemas (added on staging) now uses ErrRetryable instead of ErrNonRetryable, so a failed sampled discover exits 1, not 3.
• Updated retry counts in utils/retry_test.go and drivers/abstract/markers_test.go for fix: move RetryCount default to config.Validate() across all drivers #1058, which made RetryOnBackoff run attempts + 1 times.

Not in this PR

Config errors (empty host, wrong port, missing file) are still unmarked and exit 1.

Type of change

• Bug fix (non-breaking change which fixes an issue)
• Breaking change (fix or feature that would cause existing functionality to not work as expected)

How Has This Been Tested?

• connector_test.go: runs OLake in a child process and checks exit 1 vs exit 3.
• constants/constants_test.go: IsRetryable / IsNonRetryable for every wrapping shape.
• utils/retry_test.go: unmarked errors are retried; both markers stop the retry loop.
• drivers/abstract/markers_test.go: ProduceSchema failures carry no marker; a driver's own marker is kept.

All pass locally, including with -race.

Screenshots or Recordings

N/A

Documentation

• Documentation Link:
• N/A (bug fix, refactor, or test changes only)

Related PR's (If Any):

• fix: move RetryCount default to config.Validate() across all drivers #1058: changed RetryOnBackoff to attempts + 1 calls; tests updated for it here.

[CLAassistant]

All committers have signed the CLA.

[github-actions]

ExitCodeFailure documents the supervisor contract for the retryable failure path, but it is never used by production code: connector.go's fallback relies on logger.Fatal's hardcoded os.Exit(1) (and zerolog's own fatal exit), so the coupling between this constant and the actual exit-1 path is only implicit. If logger.Fatal's exit code ever changes, the documented contract silently drifts. Consider wiring the constant into the fallback exit path or at least noting that exit 1 is produced by logger.Fatal.

[github-actions]

Because ErrNonRetryable wraps ErrRetryable, IsRetryable(err) also returns true for manual-intervention errors that must never be retried (exit 3). The only current production caller (RetryOnBackoff in utils/utils.go) wants exactly 'stop in-process retries for either marker', so it works today, but the name invites the natural reading 'this error can be retried', and a future caller will likely retry exit-3 failures forever. Consider naming this helper for what it means (e.g. StopsInProcessRetries) or making IsRetryable exclude ErrNonRetryable, since even the tests need comments to explain the inverted semantics.

Suggestion:

Suggested change

	func IsRetryable(err error) bool {
	return err != nil && errors.Is(err, ErrRetryable)
	}
	// StopsInProcessRetries reports whether retrying inside this process is pointless: both
	// ErrRetryable (exit 1, a process restart may fix it) and ErrNonRetryable (exit 3) stop here.
	func StopsInProcessRetries(err error) bool {
	return err != nil && errors.Is(err, ErrRetryable)
	}

[github-actions]

The comment above this line still says "any fetch error (including parent ctx cancellation) is non-retryable", which now contradicts the code: the error is wrapped in ErrRetryable, meaning in-process retries stop and the CLI exits with code 1 (retry may fix). Under the new exit-code taxonomy, "non-retryable" specifically means ErrNonRetryable/exit 3 (manual intervention). Update the comment so it reflects the new semantics and doesn't mislead future changes into reintroducing ErrNonRetryable here.

Suggestion:

Suggested change

	return fmt.Errorf("%w: error reading message in Kafka CDC sync: %w", constants.ErrRetryable, err)
	// any fetch error (including parent ctx cancellation) stops in-process retries and exits 1;
	// a fresh process run may recover.
	// For more info, go through the documentation: https://pkg.go.dev/github.com/twmb/franz-go/pkg/kgo#Fetches.Errors
	if err := fetches.Err(); err != nil {
	return fmt.Errorf("%w: error reading message in Kafka CDC sync: %w", constants.ErrRetryable, err)
	}

[github-actions]

This branch handles the exit mode that is still named nonRetryableExit (pkg/kafka/types.go:17 documents it as "stop processing due to unrecoverable kafka errors", and the OnPartitionsLost callback comment says "Trigger non-retryable shutdown"), yet it now produces ErrRetryable — i.e. exit 1, restart can recover. The name now contradicts the sentinel it yields; a future reader may "fix" this back to ErrNonRetryable based on the name. Consider renaming the exit mode (e.g. partitionLossExit) and updating the callback comment in CreateReader so naming and behavior stay aligned.

Suggestion:

Suggested change

	return true, fmt.Errorf("%w: kafka sync aborted due to partition loss during consumer group rebalance", constants.ErrRetryable)
	case partitionLossExit:
	return true, fmt.Errorf("%w: kafka sync aborted due to partition loss during consumer group rebalance", constants.ErrRetryable)

[github-actions]

messageReceived is only set after an XLogData message (line 97); keepalive messages never set it. A connection that goes silently stale (no EOF, e.g. half-open TCP) before the first WAL record still lands in this branch and is reported as non-retryable "no records found ... manual intervention required" (exit code 3, with advice to do a full load) even though keepalives may have proven the connection healthy earlier — exactly the misclassification this change aims to eliminate. Consider tracking receipt of any message (including keepalives) in a separate flag and using it for the deadline classification, keeping the stricter messageReceived semantics only for the idle check at line 55.

Suggestion:

Suggested change

	if !messageReceived {
	return fmt.Errorf("%w: no records found in given initial wait time, try increasing it or do full load", constants.ErrNonRetryable)
	}
	if !anyMessageReceived {
	return fmt.Errorf("%w: no records found in given initial wait time, try increasing it or do full load", constants.ErrNonRetryable)
	}

[github-actions]

This new error carries no sentinel, so RetryOnBackoff (utils/utils.go) treats it as in-process retryable: each retry re-enters StreamChanges and blocks up to initialWaitTime in ReceiveMessage before failing again. With Postgres' default initialWaitTime of 3h and a persistently dead connection/walsender, the sync can hang for roughly (RetryCount+1) x 3h before exiting with code 1, where the previous code failed fast with ErrNonRetryable on the first timeout. Note that a healthy walsender emits keepalives every wal_sender_timeout/2 (~30s by default), so the post-first-message idle timeout doesn't need to be as large as the initial catch-up wait. Consider wrapping with constants.ErrRetryable (same convention used for Kafka fetch errors in this change: stop in-process retries, orchestrator/CLI retry with exit code 1) and/or bounding the per-message idle timeout more tightly once the first WAL record has been received.

Suggestion:

Suggested change

	return fmt.Errorf("no WAL message received for %s, replication connection may be lost: %w", p.socket.initialWaitTime, err)
	return fmt.Errorf("%w: no WAL message received for %s, replication connection may be lost: %w", constants.ErrRetryable, p.socket.initialWaitTime, err)

[github-actions]

timeoutCtx.Done() fires both on the 5-minute expiry and when the parent ctx is canceled (graceful shutdown), and this path now reports retryable "LSN not updated after 5 minutes" in both cases. On cancellation the CLI exits with code 1 and a misleading message instead of the real cancellation cause; the sentinel also suppresses any further retry. Distinguish the two via timeoutCtx.Err() and return the cancellation cause on the canceled path, reserving the "LSN not updated" error for the actual deadline.

Suggestion:

Suggested change

	return fmt.Errorf("%w: %s", constants.ErrRetryable, "LSN not updated after 5 minutes")
	if errors.Is(timeoutCtx.Err(), context.Canceled) {
	return fmt.Errorf("waiting for slot LSN update canceled: %w", timeoutCtx.Err())
	}
	return fmt.Errorf("%w: %s", constants.ErrRetryable, "LSN not updated after 5 minutes")