Merge pull request #2530 from keboola/devin/1772005948-psgo-181-gcs-upload-expired-credentials

fix(stream): check credential expiration before slice upload, handle non-retryable errors [PSGO-181]

Author: Matovidlo

docs/e2e_tests.md

@@ -290,7 +290,7 @@ TEST_KBC_PROJECTS_LOCK_HOST=redis+tls://redis:6380
 
 ## Generate new unique ID
 
-If a ENV placeholder in the form `^TEST_NEW_TICKET_\d+$` is found, it is replaced with new ID/ticket [generated by API](https://keboola.docs.apiary.io/#reference/tickets/generate-unique-id/generate-new-id).
+If a ENV placeholder in the form `^TEST_NEW_TICKET_\d+$` is found, it is replaced with new ID/ticket generated by API.
 - E.g. `%%TEST_NEW_TICKET_1%%`
 - The value is generated when the first occurrence is found.
 - All occurrences are replaced with the same value.

internal/pkg/model/object.go

@@ -362,7 +362,7 @@ func (m BranchMetadata) ToOrderedMap() *orderedmap.OrderedMap {
 	return res
 }
 
-// Branch https://keboola.docs.apiary.io/#reference/development-branches/branches/list-branches
+// Branch represents a development branch.
 type Branch struct {
 	BranchKey
 	Name        string         `json:"name" validate:"required" diff:"true" metaFile:"true"`
@@ -473,7 +473,7 @@ func (m ConfigMetadata) InstanceID() string {
 	return m[instanceIDMetadataKey]
 }
 
-// Config https://keboola.docs.apiary.io/#reference/components-and-configurations/component-configurations/list-configurations
+// Config represents a component configuration.
 type Config struct {
 	ConfigKey
 	Name           string                 `json:"name" validate:"required" diff:"true" metaFile:"true"`
@@ -528,7 +528,7 @@ func (c *ConfigWithRows) ToAPIObject(changeDescription string, changedFields Cha
 	return out, append(changedFields.Slice(), "changeDescription")
 }
 
-// ConfigRow https://keboola.docs.apiary.io/#reference/components-and-configurations/component-configurations/list-configurations
+// ConfigRow represents a configuration row.
 type ConfigRow struct {
 	ConfigRowKey
 	Name        string                 `json:"name" diff:"true" metaFile:"true"`

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

@@ -144,7 +144,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

@@ -15,6 +15,10 @@ import (
 	"github.com/keboola/keboola-as-code/internal/pkg/utils/errors"
 )
 
+// ErrCredentialsExpired is returned when the upload credentials for a file have expired.
+// This is a non-retryable error because the credentials will never become valid again without external intervention.
+var ErrCredentialsExpired = errors.New("upload credentials have expired")
+
 func (b *Bridge) uploadSlice(ctx context.Context, volume *diskreader.Volume, slice plugin.Slice, stats statistics.Value) error {
 	// Skip upload if the slice is empty.
 	// The state is anyway switched to the SliceUploaded by the operator.
@@ -23,20 +27,47 @@ 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 {
 		b.logger.Warnf(ctx, "unable to open reader: %v", err)
 		return err
 	}
 
+	// Error when closing the reader is not a fatal error
+	// Register this defer immediately after opening to prevent resource leaks on early returns
+	defer func() {
+		err := reader.Close(ctx)
+		if err != nil {
+			b.logger.Warnf(ctx, "unable to close reader: %v", err)
+			return
+		}
+	}()
+
 	// Get authorization token
 	existingToken, err := b.schema.Token().ForSink(slice.SinkKey).GetOrErr(b.client).Do(ctx).ResultOrErr()
 	if err != nil {
 		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()}
 
@@ -62,21 +93,6 @@ func (b *Bridge) uploadSlice(ctx context.Context, volume *diskreader.Volume, sli
 		}
 	}()
 
-	// Error when closing the reader is not a fatal error
-	defer func() {
-		err := reader.Close(ctx)
-		if err != nil {
-			b.logger.Warnf(ctx, "unable to close reader: %v", err)
-			return
-		}
-	}()
-
-	// 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 != "" {

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

@@ -0,0 +1,19 @@
+package model
+
+// NonRetryableError wraps an error to indicate that the operation should not be retried.
+// For example, expired credentials will never succeed on retry without external intervention.
+type NonRetryableError struct {
+	Err error
+}
+
+func NewNonRetryableError(err error) *NonRetryableError {
+	return &NonRetryableError{Err: err}
+}
+
+func (e *NonRetryableError) Error() string {
+	return e.Err.Error()
+}
+
+func (e *NonRetryableError) Unwrap() error {
+	return e.Err
+}

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/repository/slice/slice_retry.go

@@ -15,3 +15,13 @@ func (r *Repository) IncrementRetryAttempt(sliceKey model.SliceKey, now time.Tim
 		return slice, nil
 	})
 }
+
+// IncrementNonRetryableAttempt increments retry attempt with a fixed interval instead of exponential backoff.
+// This is used for errors that will never succeed (e.g., expired credentials) but should still be
+// periodically reported in logs to keep the issue visible.
+func (r *Repository) IncrementNonRetryableAttempt(sliceKey model.SliceKey, now time.Time, reason string, fixedInterval time.Duration) *op.AtomicOp[model.Slice] {
+	return r.update(sliceKey, now, func(slice model.Slice) (model.Slice, error) {
+		slice.IncrementNonRetryableAttempt(now, reason, fixedInterval)
+		return slice, nil
+	})
+}

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

@@ -93,6 +93,23 @@ func (v *Retryable) IncrementRetryAttempt(backoff RetryBackoff, failedAt time.Ti
 	v.RetryAfter = &retryAfterUTC
 }
 
+// IncrementNonRetryableAttempt increments the retry attempt counter using a fixed interval
+// instead of exponential backoff. This is used for errors that will never succeed (e.g., expired credentials)
+// but should still be periodically reported in logs.
+func (v *Retryable) IncrementNonRetryableAttempt(failedAt time.Time, reason string, fixedInterval time.Duration) {
+	v.RetryAttempt += 1
+	v.RetryReason = reason
+
+	failedAtUTC := utctime.From(failedAt)
+	if v.FirstFailedAt == nil {
+		v.FirstFailedAt = &failedAtUTC
+	}
+	v.LastFailedAt = &failedAtUTC
+
+	retryAfterUTC := utctime.From(failedAt.Add(fixedInterval))
+	v.RetryAfter = &retryAfterUTC
+}
+
 func (v *Retryable) ResetRetry() {
 	v.RetryAttempt = 0
 	v.RetryReason = ""

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

@@ -238,3 +238,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")
+}