fix: update and extend async event delivery time metrics tests for JobParameters logic (#5975)

# Description

This PR updates both the source code and tests related to async event
delivery time metrics in the batch router, specifically for the
`emitAsyncEventDeliveryTimeMetrics` logic.

**Source code changes:**
- Refactored `emitAsyncEventDeliveryTimeMetrics` to prefer extracting
metric parameters from `JobParameters` on the `JobStatusT` struct, with
a fallback to `OriginalJobParameters` if `JobParameters` is missing or
invalid.
- Improved error logging and robustness when job parameters are missing
or malformed.

**Test changes:**
- Updated all relevant tests to set `JobParameters` on `JobStatusT` to
match the value in `OriginalJobParameters`, ensuring the new primary
path is exercised.
- Added a new test to verify fallback logic when `JobParameters` is
empty, confirming that metrics are still emitted using
`OriginalJobParameters`.
- Ensured that negative and edge cases (such as missing or invalid
parameters, or no successful jobs) are still covered and use the correct
field.
- These changes ensure robust coverage for the new logic and prevent
regressions in metric emission for async destinations.

## Linear Ticket


https://linear.app/rudderstack/issue/INT-3763/rudderserver-add-event-delivery-metrics-for-async-destinations

## Security

- [x] The code changed/added as part of this pull request won't create
any security issues with how the software is being used.

Author: koladilip

router/batchrouter/handle_async.go

@@ -323,8 +323,8 @@ func (brt *Handle) pollAsyncStatus(ctx context.Context) {
 				}
 				statusList, err := brt.updatePollStatusToDB(ctx, destinationID, sourceID, importingJob, pollResp)
 				if err == nil {
-					brt.asyncDestinationStruct[destinationID].UploadInProgress = false
 					brt.recordAsyncDestinationDeliveryStatus(sourceID, destinationID, statusList)
+					brt.asyncStructCleanUp(destinationID)
 				}
 			}
 		}
@@ -360,17 +360,19 @@ func (brt *Handle) asyncUploadWorker(ctx context.Context) {
 					brt.asyncDestinationStruct[destinationID].CanUpload = true
 					brt.asyncDestinationStruct[destinationID].PartFileNumber++
 					uploadResponse := brt.asyncDestinationStruct[destinationID].Manager.Upload(brt.asyncDestinationStruct[destinationID])
-					if uploadResponse.ImportingParameters != nil && len(uploadResponse.ImportingJobIDs) > 0 {
-						brt.asyncDestinationStruct[destinationID].UploadInProgress = true
-					}
+
 					brt.setMultipleJobStatus(setMultipleJobStatusParams{
 						AsyncOutput:           uploadResponse,
 						Attempted:             true,
 						AttemptNums:           brt.asyncDestinationStruct[destinationID].AttemptNums,
 						FirstAttemptedAts:     brt.asyncDestinationStruct[destinationID].FirstAttemptedAts,
 						OriginalJobParameters: brt.asyncDestinationStruct[destinationID].OriginalJobParameters,
 					})
-					brt.asyncStructCleanUp(destinationID)
+					if uploadResponse.ImportingParameters != nil && len(uploadResponse.ImportingJobIDs) > 0 {
+						brt.asyncDestinationStruct[destinationID].UploadInProgress = true
+					} else {
+						brt.asyncStructCleanUp(destinationID)
+					}
 				}
 				brt.asyncDestinationStruct[destinationID].UploadMutex.Unlock()
 			}
@@ -412,6 +414,7 @@ func (brt *Handle) asyncStructCleanUp(destinationID string) {
 	misc.RemoveFilePaths(brt.asyncDestinationStruct[destinationID].FileName)
 	brt.asyncDestinationStruct[destinationID].ImportingJobIDs = []int64{}
 	brt.asyncDestinationStruct[destinationID].FailedJobIDs = []int64{}
+	brt.asyncDestinationStruct[destinationID].UploadInProgress = false
 	brt.asyncDestinationStruct[destinationID].Size = 0
 	brt.asyncDestinationStruct[destinationID].Exists = false
 	brt.asyncDestinationStruct[destinationID].Count = 0

router/batchrouter/handle_async_test.go

@@ -284,6 +284,67 @@ func TestAsyncDestinationManager(t *testing.T) {
 			}()
 			<-done
 		})
+		t.Run("UploadInProgress skips upload", func(t *testing.T) {
+			ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond)
+			defer cancel()
+
+			batchRouter := defaultHandle(destType)
+			destinationID := "destinationID"
+			batchRouter.destinationsMap[destinationID] = &routerutils.DestinationWithSources{
+				Destination: backendconfig.DestinationT{ID: destinationID},
+				Sources:     []backendconfig.SourceT{{ID: "sourceID"}},
+			}
+			batchRouter.asyncDestinationStruct[destinationID] = &common.AsyncDestinationStruct{
+				UploadInProgress: true,
+				Manager: mockAsyncDestinationManager{
+					uploadOutput: common.AsyncUploadOutput{ImportingJobIDs: []int64{1}},
+				},
+			}
+
+			done := make(chan struct{})
+			go func() {
+				defer close(done)
+				batchRouter.asyncUploadWorker(ctx)
+			}()
+			<-done
+			// If no panic, the continue was hit and test passes
+		})
+
+		// Test for line 374: else branch (asyncStructCleanUp)
+		t.Run("asyncStructCleanUp is called when no ImportingParameters or ImportingJobIDs", func(t *testing.T) {
+			ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond)
+			defer cancel()
+
+			batchRouter := defaultHandle(destType)
+			destinationID := "destinationID"
+			batchRouter.destinationsMap[destinationID] = &routerutils.DestinationWithSources{
+				Destination: backendconfig.DestinationT{ID: destinationID},
+				Sources:     []backendconfig.SourceT{{ID: "sourceID", WorkspaceID: "workspaceID"}},
+			}
+			batchRouter.asyncDestinationStruct[destinationID] = &common.AsyncDestinationStruct{
+				Exists: true,
+				Manager: mockAsyncDestinationManager{
+					uploadOutput: common.AsyncUploadOutput{
+						DestinationID:       destinationID,
+						ImportingParameters: nil,
+						ImportingJobIDs:     []int64{},
+					},
+				},
+			}
+			batchRouter.uploadIntervalMap[destinationID] = 0
+			batchRouter.asyncDestinationStruct[destinationID].CanUpload = true
+
+			done := make(chan struct{})
+			go func() {
+				defer close(done)
+				// Expected to clean up the asyncDestinationStruct
+				batchRouter.asyncUploadWorker(ctx)
+				cancel()
+			}()
+			<-done
+
+			require.False(t, batchRouter.asyncDestinationStruct[destinationID].Exists)
+		})
 	})
 	t.Run("Poll (StatusCode=StatusBadRequest)", func(t *testing.T) {
 		ctx, cancel := context.WithTimeout(context.Background(), 1000*time.Second)