refactor: remove archiveInPreProcess processor toggle (#7205)

# Description

- Remove Processor.archiveInPreProcess from processor config loading.
- Always store archive jobs during preprocessStage and clear
archivalJobs before downstream processing.
- Simplify pretransformStage to only store event schema jobs.
- Update archival tests and remove the source hydration env override for
the deleted toggle.

## Linear Ticket

PIPE-3195

## Validation

- `gofmt -w processor/processor.go processor/processor_test.go
integration_test/srchydration/src_hydration_test.go`
- `go test ./processor -run TestArchival` passed
- `rg "archiveInPreProcess|Processor\\.archiveInPreProcess"
/workspace/rudder-server` returned no matches (rg exit 1 as expected)
- `go test ./integration_test/srchydration -run TestSrcHydration` was
attempted; see known issue below

## ⚠️ Known Issues

- `go test ./integration_test/srchydration -run TestSrcHydration` failed
in this environment because err_idx jobs never reached succeeded while
the error-index reporter attempted external S3 bucket `rudder-saas` and
hit AWS HeadBucket 403 / S3 config RootCAs errors instead of using local
MinIO. The non-failing hydration subtests passed.

## 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.

---------

Co-authored-by: tech-infra-rudderstack <tech-infra+github-signing@rudderstack.com>
Co-authored-by: Mihir Gandhi <mihir20121997@gmail.com>

Author: 3 people

integration_test/srchydration/src_hydration_test.go

@@ -549,7 +549,6 @@ func runRudderServer(t testing.TB, ctx context.Context, cancel context.CancelFun
 	t.Setenv(configKeyToEnv("recovery.enabled"), "false")
 	t.Setenv(configKeyToEnv("Profiler.Enabled"), "false")
 	t.Setenv(configKeyToEnv("Gateway.enableSuppressUserFeature"), "false")
-	t.Setenv(configKeyToEnv("Processor.archiveInPreProcess"), "true")
 	t.Setenv(configKeyToEnv("Processor.SourceHydration.maxRetry"), "2")
 	t.Setenv(configKeyToEnv("enableStats"), "false")
 	if minioResource != nil {

processor/processor.go

@@ -200,7 +200,6 @@ type Handle struct {
 		pythonTransformConfig                     transformerutils.PythonTransformConfig
 		userTransformationMirroringBlockedIDs     config.ValueLoader[[]string]
 		storeSamplerEnabled                       config.ValueLoader[bool]
-		archiveInPreProcess                       bool
 	}
 
 	drainConfig struct {
@@ -818,7 +817,6 @@ func (proc *Handle) loadConfig() {
 	proc.config.transformTimesPQLength = proc.conf.GetIntVar(5, 1, "Processor.transformTimesPQLength")
 	// GWCustomVal is used as a key in the jobsDB customval column
 	proc.config.GWCustomVal = proc.conf.GetStringVar("GW", "Gateway.CustomVal")
-	proc.config.archiveInPreProcess = proc.conf.GetBoolVar(false, "Processor.archiveInPreProcess")
 	proc.config.pythonTransformConfig = transformerutils.LoadPythonTransformConfig(proc.conf)
 	proc.loadReloadableConfig(defaultPayloadLimit, defaultMaxEventsToProcess)
 }
@@ -2212,12 +2210,10 @@ func (proc *Handle) preprocessStage(partition string, subJobs subJob, delay time
 		return nil, fmt.Errorf("len(statusList):%d != len(jobList):%d", len(statusList), len(jobList))
 	}
 
-	if proc.config.archiveInPreProcess {
-		if err := proc.storeArchiveJobs(ctx, archivalJobs); err != nil {
-			return nil, err
-		}
-		archivalJobs = nil
+	if err := proc.storeArchiveJobs(ctx, archivalJobs); err != nil {
+		return nil, err
 	}
+	archivalJobs = nil
 
 	return &srcHydrationMessage{
 		partition:                     partition,
@@ -2258,30 +2254,11 @@ func (proc *Handle) pretransformStage(partition string, preTrans *preTransformat
 	groupedEvents := make(map[string][]types.TransformerEvent)
 	uniqueMessageIdsBySrcDestKey := make(map[string]map[string]struct{})
 
-	if !proc.config.archiveInPreProcess {
-		g, groupCtx := errgroup.WithContext(ctx)
-
-		g.Go(func() error {
-			return proc.storeEventSchemaJobs(groupCtx,
-				lo.Flatten(lo.MapToSlice(preTrans.eventSchemaJobsBySourceId, func(_ SourceIDT, jobs []*jobsdb.JobT) []*jobsdb.JobT {
-					return jobs
-				})))
-		})
-
-		g.Go(func() error {
-			return proc.storeArchiveJobs(groupCtx, preTrans.archivalJobs)
-		})
-
-		if err := g.Wait(); err != nil {
-			return nil, err
-		}
-	} else {
-		if err := proc.storeEventSchemaJobs(ctx,
-			lo.Flatten(lo.MapToSlice(preTrans.eventSchemaJobsBySourceId, func(_ SourceIDT, jobs []*jobsdb.JobT) []*jobsdb.JobT {
-				return jobs
-			}))); err != nil {
-			return nil, err
-		}
+	if err := proc.storeEventSchemaJobs(ctx,
+		lo.Flatten(lo.MapToSlice(preTrans.eventSchemaJobsBySourceId, func(_ SourceIDT, jobs []*jobsdb.JobT) []*jobsdb.JobT {
+			return jobs
+		}))); err != nil {
+		return nil, err
 	}
 
 	// REPORTING - START