Carl/bcda 8596 root cause fixes

bcda/client/bluebutton.go

@@ -120,11 +120,7 @@ func NewBlueButtonClient(config BlueButtonConfig) (*BlueButtonClient, error) {
 		// Ensure that we have compression enabled. This allows the transport to request for gzip content
 		// and handle the decompression transparently.
 		// See: https://golang.org/src/net/http/transport.go?s=3396:10950#L182 for more information
-		DisableCompression:  false,
-		MaxIdleConns:        100,              // default value
-		MaxIdleConnsPerHost: 100,              // Upped from 2 to 100 to match the default value of MaxIdleConns
-		IdleConnTimeout:     90 * time.Second, // default value
-		MaxConnsPerHost:     0,                // default value (0 means no limit)
+		DisableCompression: false,
 	}
 	var timeout int
 	if timeout, err = strconv.Atoi(conf.GetEnv("BB_TIMEOUT_MS")); err != nil {

bcda/models/models.go

@@ -127,7 +127,7 @@ const (
 )
 
 type JobEnqueueArgs struct {
-	ID              int
+	ID              int // parent Job ID
 	ACOID           string
 	CMSID           string
 	BeneficiaryIDs  []string

bcdaworker/queueing/manager.go

@@ -76,7 +76,7 @@ func validateJob(ctx context.Context, cfg ValidateJobConfig) (*models.Job, error
 		maxNotFoundRetries, err := safecast.ToInt(utils.GetEnvInt("BCDA_WORKER_MAX_JOB_NOT_FOUND_RETRIES", 3))
 		if err != nil {
 			cfg.Logger.Errorf("Failed to convert BCDA_WORKER_MAX_JOB_NOT_FOUND_RETRIES to int32. Defaulting to 3. Error: %s", err)
-			return nil, err, false
+			return nil, err, true
 		}
 
 		if cfg.ErrorCount >= maxNotFoundRetries {
@@ -93,7 +93,7 @@ func validateJob(ctx context.Context, cfg ValidateJobConfig) (*models.Job, error
 		u, err := safecast.ToUint(cfg.JobID)
 		if err != nil {
 			cfg.Logger.Errorf("Failed to convert Job ID to uint. Error: %s", err)
-			return nil, err, false
+			return nil, err, true
 		}
 
 		_, err = worker.CheckJobCompleteAndCleanup(ctx, cfg.Repository, u)

bcdaworker/queueing/manager_test.go

@@ -22,7 +22,7 @@ import (
 	"github.com/stretchr/testify/mock"
 )
 
-func TestProcessJobFailedValidation(t *testing.T) {
+func TestProcessJobFailedValidation_Integration(t *testing.T) {
 	tests := []struct {
 		name        string
 		validateErr error
@@ -87,7 +87,6 @@ func TestProcessJobFailedValidation(t *testing.T) {
 			}
 		})
 	}
-
 }
 
 func TestCheckIfCancelled(t *testing.T) {

bcdaworker/queueing/que.go

@@ -97,15 +97,15 @@ func (q *queue) processJob(queJob *que.Job) error {
 		return nil
 	}
 
-	ctx = log.NewStructuredLoggerEntry(log.Worker, ctx)
-	ctx, _ = log.SetCtxLogger(ctx, "job_id", queJob.ID)
-	ctx, logger := log.SetCtxLogger(ctx, "transaction_id", jobArgs.TransactionID)
-
 	jobID, err := safecast.ToInt64(jobArgs.ID)
 	if err != nil {
 		return err
 	}
 
+	ctx = log.NewStructuredLoggerEntry(log.Worker, ctx)
+	ctx, _ = log.SetCtxLogger(ctx, "job_id", jobID)
+	ctx, logger := log.SetCtxLogger(ctx, "transaction_id", jobArgs.TransactionID)
+
 	exportJob, err, ackJob := validateJob(ctx, ValidateJobConfig{
 		WorkerInstance: q.worker,
 		Logger:         logger,
@@ -125,7 +125,7 @@ func (q *queue) processJob(queJob *que.Job) error {
 	}
 
 	// start a goroutine that will periodically check the status of the parent job
-	go checkIfCancelled(ctx, q.repository, cancel, queJob.ID, 15)
+	go checkIfCancelled(ctx, q.repository, cancel, jobID, 15)
 
 	if err := q.worker.ProcessJob(ctx, queJob.ID, *exportJob, jobArgs); err != nil {
 		err := errors.Wrap(err, "failed to process job")

bcdaworker/queueing/river.go

@@ -99,13 +99,18 @@ type JobWorker struct {
 	river.WorkerDefaults[models.JobEnqueueArgs]
 }
 
-func (w *JobWorker) Work(ctx context.Context, job *river.Job[models.JobEnqueueArgs]) error {
+func (w *JobWorker) Work(ctx context.Context, rjob *river.Job[models.JobEnqueueArgs]) error {
 	ctx, cancel := context.WithCancel(ctx)
 	defer cancel()
 
+	jobID, err := safecast.ToInt64(rjob.Args.ID)
+	if err != nil {
+		return err
+	}
+
 	ctx = log.NewStructuredLoggerEntry(log.Worker, ctx)
-	ctx, _ = log.SetCtxLogger(ctx, "job_id", job.Args.ID)
-	ctx, logger := log.SetCtxLogger(ctx, "transaction_id", job.Args.TransactionID)
+	ctx, _ = log.SetCtxLogger(ctx, "job_id", jobID)
+	ctx, logger := log.SetCtxLogger(ctx, "transaction_id", rjob.Args.TransactionID)
 
 	// TODO: use pgxv5 when available
 	mainDB := database.Connection
@@ -114,19 +119,14 @@ func (w *JobWorker) Work(ctx context.Context, job *river.Job[models.JobEnqueueAr
 
 	defer updateJobQueueCountCloudwatchMetric(mainDB, logger)
 
-	jobID, err := safecast.ToInt64(job.Args.ID)
-	if err != nil {
-		return err
-	}
-
 	exportJob, err, ackJob := validateJob(ctx, ValidateJobConfig{
 		WorkerInstance: workerInstance,
 		Logger:         logger,
 		Repository:     repo,
 		JobID:          jobID,
-		QJobID:         job.ID,
-		Args:           job.Args,
-		ErrorCount:     len(job.Errors),
+		QJobID:         rjob.ID,
+		Args:           rjob.Args,
+		ErrorCount:     len(rjob.Errors),
 	})
 	if ackJob {
 		// End logic here, basically acknowledge and return which will remove it from the queue.
@@ -138,9 +138,9 @@ func (w *JobWorker) Work(ctx context.Context, job *river.Job[models.JobEnqueueAr
 	}
 
 	// start a goroutine that will periodically check the status of the parent job
-	go checkIfCancelled(ctx, repo, cancel, job.ID, 15)
+	go checkIfCancelled(ctx, repo, cancel, jobID, 15)
 
-	if err := workerInstance.ProcessJob(ctx, job.ID, *exportJob, job.Args); err != nil {
+	if err := workerInstance.ProcessJob(ctx, rjob.ID, *exportJob, rjob.Args); err != nil {
 		err := errors.Wrap(err, "failed to process job")
 		logger.Error(err)
 		return err