fix(stream): prevent reader leak and ensure timestamp consistency in error handling

- Move defer reader.Close() immediately after OpenReader to prevent resource leaks on early returns
- Capture clock.Now() once in terminal retry handling to ensure timestamp consistency between etcd and logs

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

Author: Matovidlo

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

@@ -35,6 +35,16 @@ func (b *Bridge) uploadSlice(ctx context.Context, volume *diskreader.Volume, sli
 		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 {
@@ -83,15 +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
-		}
-	}()
-
 	// Decrypt file upload credentials
 	var credentials keboola.FileUploadCredentials
 	if keboolaFile.EncryptedCredentials != "" {

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

@@ -337,14 +337,16 @@ func (o *operator) handleUploadError(ctx context.Context, slice *sliceData, err
 		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()
+		// Capture the timestamp once to ensure consistency between etcd update and log message
+		now := o.clock.Now()
+		_, rErr := o.storage.Slice().IncrementNonRetryableAttempt(slice.SliceKey, 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))
+			now.Add(terminalRetryInterval).Format(time.RFC3339))
 		return
 	}