Feature/pav 121 processar lotes

.env.example

@@ -121,6 +121,9 @@ SCRAPER_PROVIDER_MAX_CONCURRENCY=2
 SCRAPER_PROVIDER_CONCURRENCY_OVERRIDES=
 SCRAPER_RUN_LOCK_TTL=120s
 SCRAPER_RUN_LOCK_RENEW_INTERVAL=30s
+SCRAPER_CLASSIFICATION_BATCH_SIZE=100
+SCRAPER_PERSIST_BATCH_SIZE=100
+SCRAPER_INDEX_BATCH_SIZE=250
 GOMAXPROCS=2
 GOMEMLIMIT=1500MiB
 

BACKEND.md

@@ -331,9 +331,8 @@ Aplicados a todas as respostas:
 
 - `goScraper.ts` faz POST em `${GO_SCRAPER_URL}/scrape` com `ScrapeParams` e valida `ScrapeResponse`.
 - `goKeywords.ts` consulta e publica keywords via endpoints do serviço Go (`/api/keywords`).
-- O backend lê os índices criados pelo scraper no Valkey, incluindo `scraper:jobs:keyword:*`, `scraper:jobs:family:*`, `scraper:jobs:technology:*` e `scraper:jobs:seniority:*`.
-- Disparos administrativos usam `scraperClient` (via `SCRAPER_URL`) e preservam os códigos operacionais do serviço Go. O código `SCRAPER_ALREADY_RUNNING` é um conflito esperado; `SCRAPER_RUN_LOCK_UNAVAILABLE` indica política fail-closed e não inicia coleta.
-- Fila `scraper:keywords:pending` no Valkey (`src/lib/kwsync.ts` no backend, `scraper-go/internal/kwsync`) sincroniza keywords criadas pelo usuário para o scraper-go processar, controlada por `KWSYNC_ENABLED`.
+- O backend lê o catálogo de vagas e os índices criados pelo scraper no Valkey (`scraper:job:{id}`, `scraper:jobs:keyword:*`, `scraper:jobs:family:*`, `scraper:jobs:technology:*` e `scraper:jobs:seniority:*`). O PostgreSQL não armazena o catálogo coletado; `saved_jobs` são vagas salvas pelo usuário. A indexação invertida só é atualizada depois que o scraper confirma a persistência do documento da vaga.
+- Disparos administrativos usam `scraperClient` e preservam os códigos operacionais do serviço Go. O código `SCRAPER_ALREADY_RUNNING` é um conflito esperado; `SCRAPER_RUN_LOCK_UNAVAILABLE` indica política fail-closed e não inicia coleta.
 
 ## Banco de dados
 

SCRAPER.md

@@ -232,6 +232,22 @@ O mecanismo é fail-closed: se o Valkey não confirmar a aquisição, nenhum ada
 
 O token proprietário nunca é gravado no estado operacional nem nos logs. Um `runId` independente identifica a execução para observabilidade sem expor a credencial usada pelos scripts de renovação e liberação.
 
+### Processamento em lotes após a coleta
+
+Depois da coleta, o pipeline processa vagas em etapas com filas limitadas:
+
+coleta → normalização → deduplicação → classificação → persistência → indexação.
+
+O catálogo de vagas coletadas vive no Valkey. Os documentos `scraper:job:{id}` são a fonte da verdade da execução; o PostgreSQL do backend permanece para usuários e `saved_jobs`. A indexação invertida (`scraper:jobs:keyword:*` e chaves estruturadas) só recebe IDs confirmados por `SaveBatch`.
+
+Persistência: um `MULTI/EXEC` por lote, com upsert. Em conflito pelo ID estável, atualizam-se descrição, URL, salário, datas, modalidade, localização, classificação, fontes e keywords; o ID não muda e campos vazios na nova coleta não apagam dados já persistidos. Falha no lote impede a indexação daquele lote. Retry limitado (3 tentativas) vale só para erros transitórios.
+
+Indexação: cada lote grava em chaves isoladas `:next` (ou `:next:{runId}`) em pipelines de `SCRAPER_INDEX_BATCH_SIZE` vagas. Os lotes da mesma execução se acumulam nessas chaves; ao concluir com sucesso, um `RENAME` publica atomicamente os conjuntos finais. Assim os índices invertidos representam o estado atual da execução, sem apagar lotes já processados no meio da corrida e sem deixar associações obsoletas nas chaves reconstruídas. Se a indexação falhar após o persist, `ReindexPersistedJobs` relê os documentos persistidos e reconstrói o lote em `:next`, sem repetir a coleta externa. Falha da execução descarta as chaves `:next` e mantém os índices vivos da corrida anterior.
+
+Deduplicação usa as chaves já existentes no domínio (identificador externo, `title|company|location` e URL canônica) em um mapa de chaves da execução. A janela pendente guarda vagas completas só até o lote de classificação; o restante da execução guarda o ID persistido por chave. Duplicatas tardias após o flush são mescladas com o documento já persistido (URL, descrição, fontes e keywords) e voltam a ser gravadas e indexadas.
+
+A capacidade da fila entre coleta e processamento é `2 * max(lotes)`, sem variável de ambiente extra.
+
 No Docker Compose de produção, o serviço `scraper-go` também define:
 
 - `GOMAXPROCS=2`: limita a quantidade de threads do Go executando código simultaneamente.
@@ -385,7 +401,7 @@ docker compose \
   config
 ```
 
-Confirme no serviço `scraper-go` os equivalentes de `SCRAPER_MAX_CONCURRENCY=12`, `SCRAPER_PROVIDER_MAX_CONCURRENCY=2`, `SCRAPER_PROVIDER_CONCURRENCY_OVERRIDES=""`, `GOMAXPROCS=2`, `GOMEMLIMIT=1500MiB`, `cpus: 1.5` e `mem_limit: 2g`.
+Confirme no serviço `scraper-go` os equivalentes de `SCRAPER_MAX_CONCURRENCY=12`, `SCRAPER_PROVIDER_MAX_CONCURRENCY=2`, `SCRAPER_PROVIDER_CONCURRENCY_OVERRIDES=""`, `SCRAPER_CLASSIFICATION_BATCH_SIZE=100`, `SCRAPER_PERSIST_BATCH_SIZE=100`, `SCRAPER_INDEX_BATCH_SIZE=250`, `GOMAXPROCS=2`, `GOMEMLIMIT=1500MiB`, `cpus: 1.5` e `mem_limit: 2g`.
 
 ## Variáveis de ambiente importantes
 
@@ -395,6 +411,9 @@ Confirme no serviço `scraper-go` os equivalentes de `SCRAPER_MAX_CONCURRENCY=12
 - `SCRAPER_PROVIDER_CONCURRENCY_OVERRIDES` — lista opcional `provider=limite`, separada por vírgulas. Vazio significa nenhum override; entrada inválida, duplicada, desconhecida ou acima do teto global impede a inicialização.
 - `SCRAPER_RUN_LOCK_TTL` — duração do lock distribuído. Padrão: `120s`. Variável ausente usa o default; valor explícito vazio ou inválido impede a inicialização. No Compose, usa `${SCRAPER_RUN_LOCK_TTL-120s}` (mesmo padrão fail-fast de `SCRAPER_MAX_CONCURRENCY`).
 - `SCRAPER_RUN_LOCK_RENEW_INTERVAL` — intervalo de renovação. Padrão: `30s`; deve ser menor que `SCRAPER_RUN_LOCK_TTL`. Variável ausente usa o default; valor explícito vazio ou inválido impede a inicialização. No Compose, usa `${SCRAPER_RUN_LOCK_RENEW_INTERVAL-30s}`.
+- `SCRAPER_CLASSIFICATION_BATCH_SIZE` — tamanho do lote de classificação. Padrão: `100`. Mínimo `1`, máximo `1000`. Valor explícito inválido impede a inicialização.
+- `SCRAPER_PERSIST_BATCH_SIZE` — tamanho do lote de persistência no Valkey (`scraper:job:*`). Padrão: `100`. Mínimo `1`, máximo `1000`.
+- `SCRAPER_INDEX_BATCH_SIZE` — tamanho do lote de indexação invertida. Padrão: `250`. Mínimo `1`, máximo `2500`.
 - `GOMAXPROCS` — limite efetivo de threads executando código Go simultaneamente. Valor inicial no Compose: `2`.
 - `GOMEMLIMIT` — meta de memória do runtime/GC. Valor inicial no Compose: `1500MiB`; não substitui `mem_limit` do container.
 - `KWSYNC_ENABLED` — padrão `false`. Liga/desliga o consumidor da fila `scraper:keywords:pending` (ver seção "Sincronização de keywords com o backend").

backend/tests/unit/services/server.test.ts

@@ -9,6 +9,10 @@ const mocks = vi.hoisted(() => ({
   logWarn: vi.fn(),
   swaggerServe: vi.fn(),
   swaggerSetup: vi.fn(() => vi.fn()),
+  startEmailWorker: vi.fn(),
+  stopEmailWorker: vi.fn(async () => {}),
+  closeEmailQueue: vi.fn(async () => {}),
+  closeCache: vi.fn(async () => {}),
 }));
 
 mocks.createJobsApiApp.mockReturnValue({
@@ -40,6 +44,19 @@ vi.mock("../../../src/swagger.js", () => ({
   default: {},
 }));
 
+vi.mock("../../../src/modules/email/email.worker.js", () => ({
+  startEmailWorker: mocks.startEmailWorker,
+  stopEmailWorker: mocks.stopEmailWorker,
+}));
+
+vi.mock("../../../src/modules/email/email.queue.js", () => ({
+  closeEmailQueue: mocks.closeEmailQueue,
+}));
+
+vi.mock("../../../src/lib/cache.js", () => ({
+  closeCache: mocks.closeCache,
+}));
+
 async function importServerEntry() {
   await import("../../../src/server.js");
   await vi.dynamicImportSettled();
@@ -60,6 +77,15 @@ describe("server entry", () => {
     });
   });
 
+  it("inicializa app e chama listen", async () => {
+    await importServerEntry();
+
+    expect(mocks.createJobsApiApp).toHaveBeenCalledTimes(1);
+    expect(mocks.set).toHaveBeenCalledWith("trust proxy", 1);
+    expect(mocks.startEmailWorker).toHaveBeenCalledTimes(1);
+    expect(mocks.listen).toHaveBeenCalled();
+    expect(mocks.listen.mock.calls[0][0]).toBe(3100);
+  });
   it(
     "inicializa app e chama listen",
     async () => {

docker-compose.yml

@@ -18,6 +18,9 @@ services:
       - SCRAPER_PROVIDER_CONCURRENCY_OVERRIDES=${SCRAPER_PROVIDER_CONCURRENCY_OVERRIDES-}
       - SCRAPER_RUN_LOCK_TTL=${SCRAPER_RUN_LOCK_TTL-120s}
       - SCRAPER_RUN_LOCK_RENEW_INTERVAL=${SCRAPER_RUN_LOCK_RENEW_INTERVAL-30s}
+      - SCRAPER_CLASSIFICATION_BATCH_SIZE=${SCRAPER_CLASSIFICATION_BATCH_SIZE-100}
+      - SCRAPER_PERSIST_BATCH_SIZE=${SCRAPER_PERSIST_BATCH_SIZE-100}
+      - SCRAPER_INDEX_BATCH_SIZE=${SCRAPER_INDEX_BATCH_SIZE-250}
       - GOMAXPROCS=${GOMAXPROCS:-2}
       - GOMEMLIMIT=${GOMEMLIMIT:-1500MiB}
       - GUPY_ENABLED=${GUPY_ENABLED:-true}

scraper-go/cmd/server/handlers.go

@@ -111,6 +111,9 @@ func searchConfigFromRuntime(
 		MaxConcurrency:               config.ResolveEffectiveConcurrency(req.MaxConcurrency, runtimeCfg.MaxConcurrency),
 		ProviderMaxConcurrency:       min(runtimeCfg.ProviderMaxConcurrency, config.ResolveEffectiveConcurrency(req.MaxConcurrency, runtimeCfg.MaxConcurrency)),
 		ProviderConcurrencyOverrides: runtimeCfg.ProviderConcurrencyOverrides,
+		ClassificationBatchSize:      runtimeCfg.ClassificationBatchSize,
+		PersistBatchSize:             runtimeCfg.PersistBatchSize,
+		IndexBatchSize:               runtimeCfg.IndexBatchSize,
 	}
 }
 

scraper-go/cmd/server/main.go

@@ -54,6 +54,9 @@ func logRuntimeConfig(cfg config.RuntimeConfig) {
 		"provider_concurrency_overrides", cfg.ProviderConcurrencyOverrides,
 		"run_lock_ttl", cfg.RunLockTTL,
 		"run_lock_renew_interval", cfg.RunLockRenewInterval,
+		"classification_batch_size", cfg.ClassificationBatchSize,
+		"persist_batch_size", cfg.PersistBatchSize,
+		"index_batch_size", cfg.IndexBatchSize,
 		"gomaxprocs_effective", runtime.GOMAXPROCS(0),
 		"gomaxprocs_source", envSource(gomaxprocsSet),
 		"gomemlimit_effective_bytes", memLimit,

scraper-go/cmd/server/server.go

@@ -57,6 +57,9 @@ func run(adapterList []ports.JobSource, runtimeCfg config.RuntimeConfig) {
 	schedulerCfg.MaxConcurrency = runtimeCfg.MaxConcurrency
 	schedulerCfg.ProviderMaxConcurrency = runtimeCfg.ProviderMaxConcurrency
 	schedulerCfg.ProviderConcurrencyOverrides = runtimeCfg.ProviderConcurrencyOverrides
+	schedulerCfg.ClassificationBatchSize = runtimeCfg.ClassificationBatchSize
+	schedulerCfg.PersistBatchSize = runtimeCfg.PersistBatchSize
+	schedulerCfg.IndexBatchSize = runtimeCfg.IndexBatchSize
 	scheduler := cronjob.New(schedulerCfg, kwStore, jobStore, adapterList, rdb, runLock)
 
 	scheduler.OnComplete = func(kws []string, scraped, saved int, duration time.Duration) {

scraper-go/internal/classifier/classify_batch_test.go

@@ -0,0 +1,66 @@
+package classifier
+
+import (
+	"testing"
+
+	"github.com/Benevanio/Jobs_Scraper_Global/scraper-go/internal/domain"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestClassifyJobsMatchesIndividualClassify(t *testing.T) {
+	jobs := []domain.Job{
+		{Title: "Backend Engineer", Company: "Acme", Description: "Golang APIs microservices"},
+		{Title: "Front-end Software Developer", Company: "Acme", Description: "React TypeScript"},
+		{Title: "Graphic Designer", Company: "Acme", Description: "Branding"},
+		{Title: "DevOps Engineer", Company: "Acme", Description: "Kubernetes terraform aws"},
+	}
+
+	batched := ClassifyJobs(jobs)
+	require.Len(t, batched, 3)
+
+	var individual []domain.Job
+	for _, job := range jobs {
+		classification := Classify(job)
+		if !classification.InScope {
+			continue
+		}
+		job.Classification = &classification
+		individual = append(individual, job)
+	}
+
+	require.Len(t, individual, len(batched))
+	for i := range batched {
+		require.NotNil(t, batched[i].Classification)
+		require.NotNil(t, individual[i].Classification)
+		assert.Equal(t, individual[i].Classification.PrimaryFamily, batched[i].Classification.PrimaryFamily)
+		assert.Equal(t, individual[i].Classification.InScope, batched[i].Classification.InScope)
+		assert.Equal(t, individual[i].Classification.Confidence, batched[i].Classification.Confidence)
+		assert.Equal(t, individual[i].Classification.Seniority, batched[i].Classification.Seniority)
+		assert.Equal(t, individual[i].Classification.Technologies, batched[i].Classification.Technologies)
+		assert.Equal(t, individual[i].Classification.RelatedFamilies, batched[i].Classification.RelatedFamilies)
+	}
+}
+
+func TestCurrentFamiliesUnchanged(t *testing.T) {
+	got := make([]string, 0, len(familyRules))
+	for _, rule := range familyRules {
+		got = append(got, rule.Family)
+	}
+
+	assert.Equal(t, []string{
+		"backend",
+		"frontend",
+		"mobile",
+		"fullstack",
+		"platform",
+		"devops",
+		"data",
+		"qa",
+		"security",
+		"leadership",
+		"software",
+	}, got)
+	assert.NotContains(t, got, "product")
+	assert.NotContains(t, got, "product_design")
+}

scraper-go/internal/config/config.go

@@ -11,10 +11,16 @@ import (
 )
 
 const (
-	DefaultMaxConcurrency         = 12
-	DefaultProviderMaxConcurrency = 2
-	DefaultRunLockTTL             = 120 * time.Second
-	DefaultRunLockRenewInterval   = 30 * time.Second
+	DefaultMaxConcurrency          = 12
+	DefaultProviderMaxConcurrency  = 2
+	DefaultRunLockTTL              = 120 * time.Second
+	DefaultRunLockRenewInterval    = 30 * time.Second
+	DefaultClassificationBatchSize = 100
+	DefaultPersistBatchSize        = 100
+	DefaultIndexBatchSize          = 250
+	MaxClassificationBatchSize     = 1000
+	MaxPersistBatchSize            = 1000
+	MaxIndexBatchSize              = 2500
 
 	SourceEnvironment     = "environment"
 	SourceInternalDefault = "internal_default"
@@ -24,6 +30,9 @@ const (
 	ScraperProviderConcurrencyOverridesEnv = "SCRAPER_PROVIDER_CONCURRENCY_OVERRIDES"
 	ScraperRunLockTTLEnv                   = "SCRAPER_RUN_LOCK_TTL"
 	ScraperRunLockRenewIntervalEnv         = "SCRAPER_RUN_LOCK_RENEW_INTERVAL"
+	ScraperClassificationBatchSizeEnv      = "SCRAPER_CLASSIFICATION_BATCH_SIZE"
+	ScraperPersistBatchSizeEnv             = "SCRAPER_PERSIST_BATCH_SIZE"
+	ScraperIndexBatchSizeEnv               = "SCRAPER_INDEX_BATCH_SIZE"
 )
 
 type RuntimeConfig struct {
@@ -34,6 +43,9 @@ type RuntimeConfig struct {
 	ProviderConcurrencyOverrides map[ports.ProviderID]int
 	RunLockTTL                   time.Duration
 	RunLockRenewInterval         time.Duration
+	ClassificationBatchSize      int
+	PersistBatchSize             int
+	IndexBatchSize               int
 }
 
 func LoadRuntimeConfig() (RuntimeConfig, error) {
@@ -49,6 +61,9 @@ func LoadRuntimeConfigFromLookup(lookup func(string) (string, bool)) (RuntimeCon
 		ProviderConcurrencyOverrides: make(map[ports.ProviderID]int),
 		RunLockTTL:                   DefaultRunLockTTL,
 		RunLockRenewInterval:         DefaultRunLockRenewInterval,
+		ClassificationBatchSize:      DefaultClassificationBatchSize,
+		PersistBatchSize:             DefaultPersistBatchSize,
+		IndexBatchSize:               DefaultIndexBatchSize,
 	}
 
 	if value, ok := lookup(ScraperMaxConcurrencyEnv); ok {
@@ -103,9 +118,57 @@ func LoadRuntimeConfigFromLookup(lookup func(string) (string, bool)) (RuntimeCon
 		)
 	}
 
+	cfg.ClassificationBatchSize, err = boundedPositiveIntFromLookup(
+		lookup,
+		ScraperClassificationBatchSizeEnv,
+		DefaultClassificationBatchSize,
+		MaxClassificationBatchSize,
+	)
+	if err != nil {
+		return RuntimeConfig{}, err
+	}
+	cfg.PersistBatchSize, err = boundedPositiveIntFromLookup(
+		lookup,
+		ScraperPersistBatchSizeEnv,
+		DefaultPersistBatchSize,
+		MaxPersistBatchSize,
+	)
+	if err != nil {
+		return RuntimeConfig{}, err
+	}
+	cfg.IndexBatchSize, err = boundedPositiveIntFromLookup(
+		lookup,
+		ScraperIndexBatchSizeEnv,
+		DefaultIndexBatchSize,
+		MaxIndexBatchSize,
+	)
+	if err != nil {
+		return RuntimeConfig{}, err
+	}
+
 	return cfg, nil
 }
 
+func boundedPositiveIntFromLookup(
+	lookup func(string) (string, bool),
+	key string,
+	fallback int,
+	max int,
+) (int, error) {
+	value, ok := lookup(key)
+	if !ok {
+		return fallback, nil
+	}
+	parsed, err := positiveInt(key, value)
+	if err != nil {
+		return 0, err
+	}
+	if parsed > max {
+		return 0, fmt.Errorf("%s must not exceed %d", key, max)
+	}
+	return parsed, nil
+}
+
 func positiveInt(key, value string) (int, error) {
 	trimmed := strings.TrimSpace(value)
 	if trimmed == "" {

scraper-go/internal/config/config_test.go

@@ -24,6 +24,9 @@ func TestLoadRuntimeConfigUsesDefaultWhenEnvMissing(t *testing.T) {
 	assert.Empty(t, cfg.ProviderConcurrencyOverrides)
 	assert.Equal(t, 120*time.Second, cfg.RunLockTTL)
 	assert.Equal(t, 30*time.Second, cfg.RunLockRenewInterval)
+	assert.Equal(t, 100, cfg.ClassificationBatchSize)
+	assert.Equal(t, 100, cfg.PersistBatchSize)
+	assert.Equal(t, 250, cfg.IndexBatchSize)
 }
 
 func TestLoadRuntimeConfigUsesEnvValue(t *testing.T) {
@@ -262,12 +265,49 @@ func TestDockerComposeUsesUnsetOnlyDefaultsForRunLock(t *testing.T) {
 	assert.Contains(t, compose, "SCRAPER_PROVIDER_CONCURRENCY_OVERRIDES=${SCRAPER_PROVIDER_CONCURRENCY_OVERRIDES-}")
 	assert.Contains(t, compose, "SCRAPER_RUN_LOCK_TTL=${SCRAPER_RUN_LOCK_TTL-120s}")
 	assert.Contains(t, compose, "SCRAPER_RUN_LOCK_RENEW_INTERVAL=${SCRAPER_RUN_LOCK_RENEW_INTERVAL-30s}")
+	assert.Contains(t, compose, "SCRAPER_CLASSIFICATION_BATCH_SIZE=${SCRAPER_CLASSIFICATION_BATCH_SIZE-100}")
+	assert.Contains(t, compose, "SCRAPER_PERSIST_BATCH_SIZE=${SCRAPER_PERSIST_BATCH_SIZE-100}")
+	assert.Contains(t, compose, "SCRAPER_INDEX_BATCH_SIZE=${SCRAPER_INDEX_BATCH_SIZE-250}")
+	assert.NotContains(t, compose, "SCRAPER_CLASSIFICATION_BATCH_SIZE=${SCRAPER_CLASSIFICATION_BATCH_SIZE:-100}")
+	assert.NotContains(t, compose, "SCRAPER_PERSIST_BATCH_SIZE=${SCRAPER_PERSIST_BATCH_SIZE:-100}")
+	assert.NotContains(t, compose, "SCRAPER_INDEX_BATCH_SIZE=${SCRAPER_INDEX_BATCH_SIZE:-250}")
 	assert.NotContains(t, compose, "SCRAPER_PROVIDER_MAX_CONCURRENCY=${SCRAPER_PROVIDER_MAX_CONCURRENCY:-2}")
 	assert.NotContains(t, compose, "INHIRE_DETAILS_CONCURRENCY")
 	assert.NotContains(t, compose, "SCRAPER_RUN_LOCK_TTL=${SCRAPER_RUN_LOCK_TTL:-120s}")
 	assert.NotContains(t, compose, "SCRAPER_RUN_LOCK_RENEW_INTERVAL=${SCRAPER_RUN_LOCK_RENEW_INTERVAL:-30s}")
 }
 
+func TestLoadRuntimeConfigRejectsInvalidBatchSizes(t *testing.T) {
+	t.Run("zero", func(t *testing.T) {
+		_, err := LoadRuntimeConfigFromLookup(func(key string) (string, bool) {
+			if key == ScraperClassificationBatchSizeEnv {
+				return "0", true
+			}
+			return "", false
+		})
+		require.Error(t, err)
+	})
+	t.Run("above max", func(t *testing.T) {
+		_, err := LoadRuntimeConfigFromLookup(func(key string) (string, bool) {
+			if key == ScraperIndexBatchSizeEnv {
+				return "99999", true
+			}
+			return "", false
+		})
+		require.Error(t, err)
+	})
+	t.Run("one is allowed", func(t *testing.T) {
+		cfg, err := LoadRuntimeConfigFromLookup(func(key string) (string, bool) {
+			if key == ScraperPersistBatchSizeEnv {
+				return "1", true
+			}
+			return "", false
+		})
+		require.NoError(t, err)
+		assert.Equal(t, 1, cfg.PersistBatchSize)
+	})
+}
+
 func findMonorepoFile(t *testing.T, name string) (string, bool) {
 	t.Helper()