fix(stream): address PR #2530 review comments - terminal state, clock injection, error messages, tests

This commit addresses all review feedback from PR #2530:

1. Fix terminal state loop by updating etcd with far-future RetryAfter when max attempts reached

2. Inject clock into Bridge to make expiration check testable

3. Move expiration check before decryption to avoid unnecessary crypto ops

4. Fix error messages (file import -> slice upload)

5. Add comprehensive test coverage for NonRetryableError and IncrementNonRetryableAttempt

Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>

Author: Matovidlo

internal/pkg/service/stream/sink/type/tablesink/keboola/bridge/bridge.go

@@ -6,6 +6,7 @@ import (
 	"time"
 
 	"github.com/dgraph-io/ristretto/v2"
+	"github.com/jonboulle/clockwork"
 	"github.com/keboola/go-cloud-encrypt/pkg/cloudencrypt"
 	"github.com/keboola/keboola-sdk-go/v2/pkg/keboola"
 	etcd "go.etcd.io/etcd/client/v3"
@@ -45,6 +46,7 @@ const (
 
 type Bridge struct {
 	logger                  log.Logger
+	clock                   clockwork.Clock
 	config                  keboolasink.Config
 	client                  etcd.KV
 	schema                  schema.Schema
@@ -68,6 +70,7 @@ type jobData struct {
 
 type dependencies interface {
 	Logger() log.Logger
+	Clock() clockwork.Clock
 	EtcdClient() *etcd.Client
 	EtcdSerde() *serde.Serde
 	Process() *servicectx.Process
@@ -105,6 +108,7 @@ func New(d dependencies, apiProvider apiProvider, config keboolasink.Config) (*B
 
 	b := &Bridge{
 		logger:                  d.Logger().WithComponent("keboola.bridge"),
+		clock:                   d.Clock(),
 		config:                  config,
 		client:                  d.EtcdClient(),
 		schema:                  schema.New(d.EtcdSerde()),

internal/pkg/service/stream/sink/type/tablesink/keboola/bridge/file.go

@@ -145,7 +145,7 @@ func (b *Bridge) deleteCredentialsOnFileDelete() {
 }
 
 func (b *Bridge) importFile(ctx context.Context, file plugin.File, stats statistics.Value) error {
-	start := time.Now()
+	start := b.clock.Now()
 
 	// Get authorization token
 	existingToken, err := b.schema.Token().ForSink(file.SinkKey).GetOrErr(b.client).Do(ctx).ResultOrErr()

internal/pkg/service/stream/sink/type/tablesink/keboola/bridge/slice.go

@@ -27,7 +27,7 @@ func (b *Bridge) uploadSlice(ctx context.Context, volume *diskreader.Volume, sli
 		return nil
 	}
 
-	start := time.Now()
+	start := b.clock.Now()
 
 	reader, err := volume.OpenReader(slice.SliceKey, slice.LocalStorage, slice.EncodingCompression, slice.StagingStorage.Compression)
 	if err != nil {
@@ -41,6 +41,23 @@ func (b *Bridge) uploadSlice(ctx context.Context, volume *diskreader.Volume, sli
 		return err
 	}
 
+	// Get file details to check expiration before decryption
+	keboolaFile, err := b.schema.File().ForFile(slice.FileKey).GetOrErr(b.client).Do(ctx).ResultOrErr()
+	if err != nil {
+		return err
+	}
+
+	// Check credential expiration before decryption to avoid unnecessary crypto/KMS operations.
+	// Expired credentials will never succeed, so we fail fast with a non-retryable error
+	// to avoid wasting API calls and generating excessive error logs.
+	expiration := keboolaFile.Expiration()
+	if !expiration.IsZero() && b.clock.Now().After(expiration.Time()) {
+		return model.NewNonRetryableError(errors.Errorf(
+			"%w: credentials for file %s expired at %s",
+			ErrCredentialsExpired, slice.FileKey.String(), expiration.String(),
+		))
+	}
+
 	// Prepare encryption metadata
 	metadata := cloudencrypt.Metadata{"sink": slice.SinkKey.String()}
 
@@ -75,12 +92,6 @@ func (b *Bridge) uploadSlice(ctx context.Context, volume *diskreader.Volume, sli
 		}
 	}()
 
-	// Get file details
-	keboolaFile, err := b.schema.File().ForFile(slice.FileKey).GetOrErr(b.client).Do(ctx).ResultOrErr()
-	if err != nil {
-		return err
-	}
-
 	// Decrypt file upload credentials
 	var credentials keboola.FileUploadCredentials
 	if keboolaFile.EncryptedCredentials != "" {
@@ -97,17 +108,6 @@ func (b *Bridge) uploadSlice(ctx context.Context, volume *diskreader.Volume, sli
 		credentials = *keboolaFile.UploadCredentials
 	}
 
-	// Check credential expiration before attempting upload.
-	// Expired credentials will never succeed, so we fail fast with a non-retryable error
-	// to avoid wasting API calls and generating excessive error logs.
-	expiration := keboolaFile.Expiration()
-	if !expiration.IsZero() && time.Now().After(expiration.Time()) {
-		return model.NewNonRetryableError(errors.Errorf(
-			"%w: credentials for file %s expired at %s",
-			ErrCredentialsExpired, slice.FileKey.String(), expiration.String(),
-		))
-	}
-
 	// Upload slice
 	uploader, err := keboola.NewUploadSliceWriter(ctx, &credentials, slice.StagingStorage.Path)
 	if err != nil {

internal/pkg/service/stream/storage/model/nonretryable_test.go

@@ -0,0 +1,53 @@
+package model_test
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+
+	"github.com/keboola/keboola-as-code/internal/pkg/service/stream/storage/model"
+	"github.com/keboola/keboola-as-code/internal/pkg/utils/errors"
+)
+
+func TestNonRetryableError_Error(t *testing.T) {
+	t.Parallel()
+
+	baseErr := errors.New("test error message")
+	nonRetryableErr := model.NewNonRetryableError(baseErr)
+
+	assert.Equal(t, "test error message", nonRetryableErr.Error())
+}
+
+func TestNonRetryableError_Unwrap(t *testing.T) {
+	t.Parallel()
+
+	baseErr := errors.New("wrapped error")
+	nonRetryableErr := model.NewNonRetryableError(baseErr)
+
+	assert.Equal(t, baseErr, nonRetryableErr.Unwrap())
+}
+
+func TestNonRetryableError_ErrorsAs(t *testing.T) {
+	t.Parallel()
+
+	// Test that errors.As correctly identifies NonRetryableError through wrapping
+	baseErr := errors.New("base error")
+	nonRetryableErr := model.NewNonRetryableError(baseErr)
+	wrappedErr := errors.Errorf("wrapped: %w", nonRetryableErr)
+
+	var target *model.NonRetryableError
+	assert.True(t, errors.As(wrappedErr, &target))
+	assert.Equal(t, nonRetryableErr, target)
+	assert.Equal(t, "base error", target.Err.Error())
+}
+
+func TestNonRetryableError_ErrorsAs_Negative(t *testing.T) {
+	t.Parallel()
+
+	// Test that a regular error is not identified as NonRetryableError
+	regularErr := errors.New("regular error")
+
+	var target *model.NonRetryableError
+	assert.False(t, errors.As(regularErr, &target))
+	assert.Nil(t, target)
+}

internal/pkg/service/stream/storage/model/retry_test.go

@@ -239,3 +239,99 @@ func TestRetryBackoff_RetryAt_Random(t *testing.T) {
 		now = retryAt
 	}
 }
+
+func TestRetryable_IncrementNonRetryableAttempt(t *testing.T) {
+	t.Parallel()
+
+	fixedInterval := 2 * time.Hour
+	v := Retryable{}
+
+	// First attempt
+	v.IncrementNonRetryableAttempt(utctime.MustParse("2000-01-01T00:00:00.000Z").Time(), "credential expired", fixedInterval)
+	assert.Equal(t, Retryable{
+		RetryAttempt:  1,
+		RetryReason:   "credential expired",
+		FirstFailedAt: ptr.Ptr(utctime.MustParse("2000-01-01T00:00:00.000Z")),
+		LastFailedAt:  ptr.Ptr(utctime.MustParse("2000-01-01T00:00:00.000Z")),
+		RetryAfter:    ptr.Ptr(utctime.MustParse("2000-01-01T02:00:00.000Z")), // +2 hours
+	}, v)
+
+	// Second attempt - FirstFailedAt should remain unchanged
+	v.IncrementNonRetryableAttempt(utctime.MustParse("2000-01-01T02:00:00.000Z").Time(), "credential expired", fixedInterval)
+	assert.Equal(t, Retryable{
+		RetryAttempt:  2,
+		RetryReason:   "credential expired",
+		FirstFailedAt: ptr.Ptr(utctime.MustParse("2000-01-01T00:00:00.000Z")), // unchanged
+		LastFailedAt:  ptr.Ptr(utctime.MustParse("2000-01-01T02:00:00.000Z")),
+		RetryAfter:    ptr.Ptr(utctime.MustParse("2000-01-01T04:00:00.000Z")), // +2 hours (fixed)
+	}, v)
+
+	// Third attempt - still fixed interval
+	v.IncrementNonRetryableAttempt(utctime.MustParse("2000-01-01T04:00:00.000Z").Time(), "credential expired", fixedInterval)
+	assert.Equal(t, Retryable{
+		RetryAttempt:  3,
+		RetryReason:   "credential expired",
+		FirstFailedAt: ptr.Ptr(utctime.MustParse("2000-01-01T00:00:00.000Z")), // unchanged
+		LastFailedAt:  ptr.Ptr(utctime.MustParse("2000-01-01T04:00:00.000Z")),
+		RetryAfter:    ptr.Ptr(utctime.MustParse("2000-01-01T06:00:00.000Z")), // +2 hours (fixed)
+	}, v)
+}
+
+func TestRetryable_IncrementNonRetryableAttempt_VsExponential(t *testing.T) {
+	t.Parallel()
+
+	// This test demonstrates the difference between exponential backoff (IncrementRetryAttempt)
+	// and fixed interval (IncrementNonRetryableAttempt)
+
+	backoff := NoRandomizationBackoff()
+	fixedInterval := 2 * time.Hour
+
+	exponential := Retryable{}
+	fixed := Retryable{}
+
+	baseTime := utctime.MustParse("2000-01-01T00:00:00.000Z").Time()
+
+	// First attempt - both use their base interval
+	exponential.IncrementRetryAttempt(backoff, baseTime, "retryable error")
+	fixed.IncrementNonRetryableAttempt(baseTime, "non-retryable error", fixedInterval)
+
+	assert.Equal(t, "2000-01-01T00:02:00.000Z", exponential.RetryAfter.String()) // +2 min (exponential base)
+	assert.Equal(t, "2000-01-01T02:00:00.000Z", fixed.RetryAfter.String())       // +2 hours (fixed)
+
+	// Second attempt - exponential increases, fixed stays same
+	exponential.IncrementRetryAttempt(backoff, baseTime.Add(2*time.Minute), "retryable error")
+	fixed.IncrementNonRetryableAttempt(baseTime.Add(2*time.Hour), "non-retryable error", fixedInterval)
+
+	assert.Equal(t, "2000-01-01T00:10:00.000Z", exponential.RetryAfter.String()) // +8 min (exponential x4)
+	assert.Equal(t, "2000-01-01T04:00:00.000Z", fixed.RetryAfter.String())       // +2 hours (fixed)
+
+	// Third attempt - exponential continues to grow, fixed stays same
+	exponential.IncrementRetryAttempt(backoff, baseTime.Add(10*time.Minute), "retryable error")
+	fixed.IncrementNonRetryableAttempt(baseTime.Add(4*time.Hour), "non-retryable error", fixedInterval)
+
+	assert.Equal(t, "2000-01-01T00:42:00.000Z", exponential.RetryAfter.String()) // +32 min (exponential x4)
+	assert.Equal(t, "2000-01-01T06:00:00.000Z", fixed.RetryAfter.String())       // +2 hours (fixed)
+
+	// Verify FirstFailedAt remains unchanged for both
+	assert.Equal(t, baseTime, exponential.FirstFailedAt.Time())
+	assert.Equal(t, baseTime, fixed.FirstFailedAt.Time())
+}
+
+func TestRetryable_IncrementNonRetryableAttempt_TerminalInterval(t *testing.T) {
+	t.Parallel()
+
+	// Test with a very long "terminal" interval (simulating max attempts reached)
+	terminalInterval := 10 * 365 * 24 * time.Hour // ~10 years
+	v := Retryable{}
+
+	baseTime := utctime.MustParse("2000-01-01T00:00:00.000Z").Time()
+	v.IncrementNonRetryableAttempt(baseTime, "max attempts reached", terminalInterval)
+
+	// Verify far-future retry time (should be at least 9 years in the future)
+	expectedRetryAfter := utctime.From(baseTime.Add(terminalInterval))
+	assert.Equal(t, expectedRetryAfter.String(), v.RetryAfter.String())
+
+	// Verify it's far in the future (more than 9 years)
+	nineYearsLater := baseTime.Add(9 * 365 * 24 * time.Hour)
+	assert.True(t, v.RetryAfter.Time().After(nineYearsLater), "RetryAfter should be more than 9 years in the future")
+}

internal/pkg/service/stream/storage/node/readernode/sliceupload/operator.go

@@ -42,10 +42,16 @@ const dbOperationTimeout = 30 * time.Second
 const nonRetryableRetryInterval = 2 * time.Hour
 
 // maxNonRetryableAttempts limits the number of retry reports for non-retryable errors.
-// After this many attempts, the slice stops being retried. With a 2-hour interval,
-// 50 attempts span ~4 days, giving operators enough time to intervene.
+// After this many attempts, the slice stops being retried by setting RetryAfter to a
+// far-future value (~10 years). With a 2-hour interval, 50 attempts span ~4 days,
+// giving operators enough time to intervene before retries are disabled.
 const maxNonRetryableAttempts = 50
 
+// terminalRetryInterval is used when maxNonRetryableAttempts is reached.
+// This far-future interval (10 years) effectively stops retry attempts while
+// keeping the slice in a queryable state in etcd for operators to investigate.
+const terminalRetryInterval = 10 * 365 * 24 * time.Hour
+
 type operator struct {
 	config     stagingConfig.OperatorConfig
 	clock      clockwork.Clock
@@ -326,6 +332,19 @@ func (o *operator) handleUploadError(ctx context.Context, slice *sliceData, err
 	// If non-retryable and max attempts reached, log and stop retrying.
 	if isNonRetryable && slice.Retry.RetryAttempt >= maxNonRetryableAttempts {
 		o.logger.Errorf(ctx, "slice upload permanently failed after %d attempts: %s", slice.Retry.RetryAttempt, err.Error())
+
+		// Update etcd with a far-future RetryAfter to prevent retry after pod restarts
+		dbCtx, dbCancel := context.WithTimeoutCause(context.WithoutCancel(ctx), dbOperationTimeout, errors.New("terminal retry update timeout"))
+		defer dbCancel()
+
+		_, rErr := o.storage.Slice().IncrementNonRetryableAttempt(slice.SliceKey, o.clock.Now(), err.Error(), terminalRetryInterval).Do(dbCtx).ResultOrErr()
+		if rErr != nil {
+			o.logger.Errorf(ctx, "cannot update slice with terminal retry interval: %s", rErr)
+			return
+		}
+
+		o.logger.Infof(ctx, "slice retry disabled until %s, requires manual intervention",
+			o.clock.Now().Add(terminalRetryInterval).Format(time.RFC3339))
 		return
 	}
 
@@ -348,7 +367,7 @@ func (o *operator) handleUploadError(ctx context.Context, slice *sliceData, err
 func (o *operator) handleNonRetryableError(ctx context.Context, dbCtx context.Context, slice *sliceData, err error) {
 	sliceEntity, rErr := o.storage.Slice().IncrementNonRetryableAttempt(slice.SliceKey, o.clock.Now(), err.Error(), nonRetryableRetryInterval).Do(dbCtx).ResultOrErr()
 	if rErr != nil {
-		o.logger.Errorf(ctx, "cannot increment file import retry: %s", rErr)
+		o.logger.Errorf(ctx, "cannot increment slice upload retry: %s", rErr)
 		return
 	}
 	o.logger.Warnf(ctx, "slice upload failed with non-retryable error (attempt %d/%d), will report again after %s",
@@ -359,7 +378,7 @@ func (o *operator) handleNonRetryableError(ctx context.Context, dbCtx context.Co
 func (o *operator) handleRetryableError(ctx context.Context, dbCtx context.Context, slice *sliceData, err error) {
 	sliceEntity, rErr := o.storage.Slice().IncrementRetryAttempt(slice.SliceKey, o.clock.Now(), err.Error()).Do(dbCtx).ResultOrErr()
 	if rErr != nil {
-		o.logger.Errorf(ctx, "cannot increment file import retry: %s", rErr)
+		o.logger.Errorf(ctx, "cannot increment slice upload retry: %s", rErr)
 		return
 	}
 	o.logger.Infof(ctx, "slice upload will be retried after %q", sliceEntity.RetryAfter.String())