fix(warehouse): infinites retries (#3050)

go.mod

@@ -128,7 +128,7 @@ require (
 	github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.9.22 // indirect
 	github.com/aws/aws-sdk-go-v2/service/internal/s3shared v1.13.22 // indirect
 	github.com/aws/aws-sdk-go-v2/service/s3 v1.30.3 // indirect
-	github.com/aws/smithy-go v1.13.5
+	github.com/aws/smithy-go v1.13.5 // indirect
 	github.com/bugsnag/panicwrap v1.3.4 // indirect
 	github.com/cespare/xxhash v1.1.0 // indirect
 	github.com/cespare/xxhash/v2 v2.2.0 // indirect

warehouse/api.go

@@ -378,8 +378,6 @@ func (uploadReq *UploadReqT) TriggerWHUpload() (response *proto.TriggerWhUploads
 		return
 	}
 
-	var uploadJobT UploadJobT
-
 	upload, err := repo.NewUploads(uploadReq.API.dbHandle).Get(context.TODO(), uploadReq.UploadId)
 	if err == model.ErrUploadNotFound {
 		return &proto.TriggerWhUploadsResponse{
@@ -398,8 +396,12 @@ func (uploadReq *UploadReqT) TriggerWHUpload() (response *proto.TriggerWhUploads
 		return
 	}
 
-	uploadJobT.upload = upload
-	uploadJobT.dbHandle = uploadReq.API.dbHandle
+	uploadJobT := UploadJobT{
+		upload:   upload,
+		dbHandle: uploadReq.API.dbHandle,
+		Now:      timeutil.Now,
+	}
+
 	err = uploadJobT.triggerUploadNow()
 	if err != nil {
 		return

warehouse/identities.go

@@ -1,11 +1,14 @@
 package warehouse
 
 import (
+	"context"
 	"database/sql"
 	"encoding/json"
 	"fmt"
 	"sync"
 
+	"github.com/rudderlabs/rudder-server/warehouse/logfield"
+
 	"github.com/rudderlabs/rudder-server/warehouse/integrations/manager"
 
 	"github.com/rudderlabs/rudder-server/warehouse/internal/model"
@@ -410,7 +413,18 @@ func (wh *HandleT) populateHistoricIdentities(warehouse warehouseutils.Warehouse
 			Warehouse: warehouse,
 		}, whManager)
 
-		tableUploadsCreated := areTableUploadsCreated(job.upload.ID)
+		tableUploadsCreated, tableUploadsErr := job.tableUploadsRepo.ExistsForUploadID(context.TODO(), job.upload.ID)
+		if tableUploadsErr != nil {
+			pkgLogger.Warnw("table uploads exists",
+				logfield.UploadJobID, job.upload.ID,
+				logfield.SourceID, job.upload.SourceID,
+				logfield.DestinationID, job.upload.DestinationID,
+				logfield.DestinationType, job.upload.DestinationType,
+				logfield.WorkspaceID, job.upload.WorkspaceID,
+				logfield.Error, tableUploadsErr.Error(),
+			)
+			return
+		}
 		if !tableUploadsCreated {
 			err := job.initTableUploads()
 			if err != nil {

warehouse/internal/model/table_upload.go

@@ -0,0 +1,29 @@
+package model
+
+import (
+	"time"
+)
+
+type TableUpload struct {
+	ID           int64
+	UploadID     int64
+	TableName    string
+	Status       string
+	Error        string
+	LastExecTime time.Time
+	TotalEvents  int64
+	CreatedAt    time.Time
+	UpdatedAt    time.Time
+	Location     string
+}
+
+const (
+	TableUploadWaiting              = "waiting"
+	TableUploadExecuting            = "executing"
+	TableUploadUpdatingSchema       = "updating_schema"
+	TableUploadUpdatingSchemaFailed = "updating_schema_failed"
+	TableUploadUpdatedSchema        = "updated_schema"
+	TableUploadExporting            = "exporting_data"
+	TableUploadExportingFailed      = "exporting_data_failed"
+	TableUploadExported             = "exported_data"
+)

warehouse/internal/model/upload.go

@@ -94,6 +94,15 @@ type UploadJob struct {
 	LoadFileGenStartTime time.Time
 }
 
+type PendingTableUpload struct {
+	UploadID      int64
+	DestinationID string
+	Namespace     string
+	TableName     string
+	Status        string
+	Error         string
+}
+
 type Matcher interface {
 	MatchString(string) bool
 }

warehouse/internal/repo/load.go

@@ -61,7 +61,7 @@ func (repo *LoadFiles) DeleteByStagingFiles(ctx context.Context, stagingFileIDs
 // Insert loadFiles into the database.
 func (repo *LoadFiles) Insert(ctx context.Context, loadFiles []model.LoadFile) (err error) {
 	// Using transactions for bulk copying
-	txn, err := repo.db.Begin()
+	txn, err := repo.db.BeginTx(ctx, &sql.TxOptions{})
 	if err != nil {
 		return
 	}