Split the duplicate-marker passes out of every reindex

AGENTS.md

@@ -71,7 +71,8 @@ Worker gotchas (`go run ./cmd/<name>`, all need `DATABASE_URL`; run `ls cmd/` fo
 - `ingest` — takes one board file: `go run ./cmd/ingest sources/<provider>.yml` (or `SOURCES_FILE`).
 - `enrich` / `tg-extract` — need `LLM_BASE_URL` / `LLM_API_KEY` / `LLM_MODEL`.
 - `embed` / `search-drain` / `rollup-facets` / `reindex-companies` — need `MEILI_URL` / `MEILI_MASTER_KEY`. `search-drain` drains `search_outbox` (queued by `cmd/ingest`, atomically with each write) into the live facet index in batches — run it frequently (e.g. every 1-2 min); see [internal/searchdrain/AGENTS.md](internal/searchdrain/AGENTS.md).
-- `backfill-derive` — re-derives every deterministic column (facets, `role_fingerprint`, slugs) in one keyset pass; `BACKFILL_CONCURRENCY` tunes the pool. Follow with `make reindex` — it collapses newly-clustered reposts and unions their geography.
+- `reindex` — rebuilds the Meilisearch jobs index. **`REINDEX_DEDUP=1` additionally refreshes the duplicate markers** (role clusters, aggregator suppression, fuzzy collapse) before the rebuild; without it the rebuild uses the markers the last dedup run left. Off by default since 2026-08-16: aggregator suppression alone measured ~23h against a 12h unit timeout, so the run was cancelled mid-dedup and never reached the rebuild — 3 days with zero successful reindexes. Run the dedup invocation on its own, rarer schedule.
+- `backfill-derive` — re-derives every deterministic column (facets, `role_fingerprint`, slugs) in one keyset pass; `BACKFILL_CONCURRENCY` tunes the pool. Follow with `REINDEX_DEDUP=1 make reindex` — that is what collapses newly-clustered reposts and unions their geography.
 - `capture-apply-form` — drains the apply-form capture queue: fetches each queued posting's application form from `greenhouse`/`ashby`/`workable`/`lever` and stores it in `apply_forms`. Needs nothing but `DATABASE_URL`; `APPLY_FORM_CONCURRENCY` (default 4) bounds how hard one run leans on a platform and `APPLY_FORM_MAX_PER_RUN` (default 5000) how much of the backlog it takes — the second matters because the first drain faces a ~185k backlog and an unbounded run would work for hours, which `Type=oneshot` turns into silently skipped timer firings. `recruitee` forms never reach this queue — its listing carries them, so ingest writes them directly.
 - `hydrate-adzuna-description` — drains the Adzuna full-description capture queue (`cmd/ingest` enqueues eligible postings; see [internal/sources/AGENTS.md](internal/sources/AGENTS.md)). `ADZUNA_DESCRIPTION_MAX_PER_RUN` (default 500) is deliberately conservative — untested against Adzuna at real crawl-host volume. `seed-adzuna-description-queue` is the one-off companion that queues the pre-existing backlog; run it once, then let the cron drain handle the rest.
 - `queue-metrics` — measures outbox depth, board-fleet health, and catalogue freshness and publishes them as Prometheus gauges via the node_exporter textfile collector. Needs `DATABASE_URL`, and is a **no-op that never opens the pool** without `PROM_TEXTFILE_DIR`. Read-only and lock-free by design — run it every minute; see [internal/worker/AGENTS.md](internal/worker/AGENTS.md) for the published names.

cmd/reindex/batches_test.go

@@ -0,0 +1,75 @@
+package main
+
+import (
+	"context"
+	"errors"
+	"strings"
+	"testing"
+)
+
+func companyList(n int) []string {
+	out := make([]string, n)
+	for i := range out {
+		out[i] = "co"
+		// Distinct enough for the error message; the loop never inspects the values.
+		out[i] += string(rune('a' + i%26))
+	}
+	return out
+}
+
+// One dead batch must not starve the rest — that is the whole reason the pass is
+// batched with per-batch fault isolation rather than one transaction.
+func TestForCompanyBatchesIsolatesOneFailure(t *testing.T) {
+	companies := companyList(3 * companyBatchSize)
+	var calls int
+	total, err := forCompanyBatches(context.Background(), companies, func(_ context.Context, batch []string) (int64, error) {
+		calls++
+		if calls == 2 {
+			return 0, errors.New("statement timeout")
+		}
+		return int64(len(batch)), nil
+	})
+	if calls != 3 {
+		t.Fatalf("ran %d batches, want all 3 attempted despite the failure", calls)
+	}
+	if want := int64(2 * companyBatchSize); total != want {
+		t.Fatalf("total = %d, want %d — the two healthy batches must still count", total, want)
+	}
+	if err == nil {
+		t.Fatal("a failed batch must surface as an aggregate error")
+	}
+}
+
+// A cancelled context ends the pass. Every remaining batch would fail instantly
+// against the same dead context, so continuing reports one deadline as hundreds of
+// separate failures — which is exactly what obscured the 2026-08-16 timeout.
+func TestForCompanyBatchesStopsOnCancellation(t *testing.T) {
+	companies := companyList(10 * companyBatchSize)
+	ctx, cancel := context.WithCancel(context.Background())
+	var calls int
+	total, err := forCompanyBatches(ctx, companies, func(ctx context.Context, batch []string) (int64, error) {
+		calls++
+		if calls == 2 {
+			cancel()
+			return 0, ctx.Err()
+		}
+		return int64(len(batch)), nil
+	})
+	if calls != 2 {
+		t.Fatalf("ran %d batches, want it to stop at the cancelled one (2)", calls)
+	}
+	if !errors.Is(err, context.Canceled) {
+		t.Fatalf("err = %v, want it to carry context.Canceled so the caller can tell a deadline from a bad batch", err)
+	}
+	// The count must be the batches that COMPLETED, not the ones that failed: the
+	// cancellation branch runs before the failure is counted, so reporting failures
+	// there would always say zero and hide how far the pass actually got.
+	if !strings.Contains(err.Error(), "after 1 completed batches") {
+		t.Fatalf("err = %q, want it to report the 1 batch that completed before the cancellation", err)
+	}
+	// Work done before the cancellation is still reported: the pass is best-effort
+	// and its markers are already written.
+	if want := int64(companyBatchSize); total != want {
+		t.Fatalf("total = %d, want %d from the batch that completed", total, want)
+	}
+}

cmd/reindex/fold_test.go

@@ -0,0 +1,57 @@
+package main
+
+import (
+	"slices"
+	"strings"
+	"testing"
+)
+
+// The fold moved out of the SQL and into Go for the planner's sake, so what it
+// produces must still be byte-identical to what `replace(company_slug, '-', ”)`
+// produced — the query compares the array against that same expression over the
+// column, and any divergence silently stops matching rows rather than erroring.
+func TestFoldCompanySlugs(t *testing.T) {
+	tests := []struct {
+		name string
+		in   []string
+		want []string
+	}{
+		{"strips every hyphen, not just the first", []string{"cfo-insights-gmbh"}, []string{"cfoinsightsgmbh"}},
+		{"leaves an unhyphenated slug alone", []string{"cfoinsights"}, []string{"cfoinsights"}},
+		// The collision is the whole point of folding: one source spells an employer
+		// "CFO Insights", another "Cfoinsights", and the two slugs must agree.
+		{"two spellings fold together", []string{"cfo-insights", "cfoinsights"}, []string{"cfoinsights", "cfoinsights"}},
+		{"keeps duplicates rather than collapsing them", []string{"a-b", "ab"}, []string{"ab", "ab"}},
+		{"empty input yields empty output", []string{}, []string{}},
+		{"a slug that is only hyphens folds to empty", []string{"---"}, []string{""}},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			got := foldCompanySlugs(tt.in)
+			if !slices.Equal(got, tt.want) {
+				t.Fatalf("foldCompanySlugs(%q) = %q, want %q", tt.in, got, tt.want)
+			}
+		})
+	}
+}
+
+// Order and length must survive the fold: the array is compared positionally by
+// nothing, but a caller batching by companyBatchSize relies on one folded entry per
+// input company — a shorter result would silently drop companies from the batch.
+func TestFoldCompanySlugsPreservesLength(t *testing.T) {
+	in := make([]string, 500)
+	for i := range in {
+		in[i] = "company-" + strings.Repeat("x", i%7) + "-slug"
+	}
+	if got := foldCompanySlugs(in); len(got) != len(in) {
+		t.Fatalf("folded %d companies into %d entries", len(in), len(got))
+	}
+}
+
+// A nil batch must not panic: forCompanyBatches never emits one today, but the
+// helper is a plain function and the query would simply match nothing.
+func TestFoldCompanySlugsNil(t *testing.T) {
+	if got := foldCompanySlugs(nil); len(got) != 0 {
+		t.Fatalf("foldCompanySlugs(nil) = %q, want empty", got)
+	}
+}

cmd/reindex/main.go

@@ -11,6 +11,7 @@ import (
 	"fmt"
 	"log"
 	"slices"
+	"strings"
 	"sync/atomic"
 	"time"
 
@@ -78,35 +79,12 @@ func run() int {
 		return 1
 	}
 
-	// Refresh the role-cluster canonical markers before reading jobs, so the collapse
-	// (splitJobs drops non-canonical reposts) reflects the current catalogue and a closed
-	// canon has failed over. Done per company in short transactions (never a table-wide
-	// lock that would stall ingest). Best-effort: a hiccup here must not block the reindex
-	// (which also owns settings/compaction), so it degrades to the prior markers.
-	if n, err := recomputeRoleDuplicates(ctx, q); err != nil {
-		log.Printf("reindex: recompute role duplicates (continuing with prior markers): %v", err)
-	} else if n > 0 {
-		log.Printf("reindex: recomputed role duplicates (%d rows re-marked)", n)
-	}
-
-	// Then suppress aggregator postings that duplicate a first-party ATS posting, so the
-	// aggregator copy drops out of this rebuild (and out of embedding/enrichment). Run
-	// AFTER the role recompute so ATS reposts have collapsed to their canon first. Same
-	// per-company, best-effort discipline as the role pass.
-	if n, err := suppressAggregatorDuplicates(ctx, q); err != nil {
-		log.Printf("reindex: suppress aggregator duplicates (continuing with prior markers): %v", err)
-	} else if n > 0 {
-		log.Printf("reindex: suppressed aggregator duplicates (%d rows re-marked)", n)
-	}
-
-	// Finally collapse reposts whose descriptions are near-identical but not byte-identical —
-	// a role reposted per city with a localized salary or legal block, which the exact passes
-	// leave split. Runs LAST so it only ever claims what they did not, and shares their
-	// per-company, best-effort discipline.
-	if n, err := collapseFuzzyDuplicates(ctx, q); err != nil {
-		log.Printf("reindex: collapse fuzzy duplicates (continuing with prior markers): %v", err)
-	} else if n > 0 {
-		log.Printf("reindex: collapsed fuzzy duplicates (%d rows re-marked)", n)
+	// The duplicate-marker passes run only when asked for (REINDEX_DEDUP=1) — see
+	// config.Reindex.Dedup for why they are no longer part of every rebuild. Without
+	// them the rebuild still collapses reposts; it just uses the markers the last
+	// dedup invocation left, which is what "eventually consistent" already meant.
+	if rcfg.Dedup {
+		refreshDuplicateMarkers(ctx, q)
 	}
 
 	reader := worker.NewFullScanReader(q)
@@ -141,6 +119,36 @@ func run() int {
 	return 0
 }
 
+// refreshDuplicateMarkers runs the three duplicate-marker passes in the order they
+// depend on: role clusters first (so ATS reposts collapse to their canon), then
+// aggregator suppression (so an aggregator copy of an already-collapsed ATS posting
+// drops out), then the fuzzy collapse (which only ever claims what the exact passes
+// did not).
+//
+// Every pass is best-effort and logs rather than fails: a hiccup in a marker refresh
+// must not stop the rebuild that follows it, which also owns index settings and
+// compaction. Each is done per company in short transactions — never a table-wide
+// lock that would stall the ingest.
+func refreshDuplicateMarkers(ctx context.Context, q *db.Queries) {
+	if n, err := recomputeRoleDuplicates(ctx, q); err != nil {
+		log.Printf("reindex: recompute role duplicates (continuing with prior markers): %v", err)
+	} else if n > 0 {
+		log.Printf("reindex: recomputed role duplicates (%d rows re-marked)", n)
+	}
+
+	if n, err := suppressAggregatorDuplicates(ctx, q); err != nil {
+		log.Printf("reindex: suppress aggregator duplicates (continuing with prior markers): %v", err)
+	} else if n > 0 {
+		log.Printf("reindex: suppressed aggregator duplicates (%d rows re-marked)", n)
+	}
+
+	if n, err := collapseFuzzyDuplicates(ctx, q); err != nil {
+		log.Printf("reindex: collapse fuzzy duplicates (continuing with prior markers): %v", err)
+	} else if n > 0 {
+		log.Printf("reindex: collapsed fuzzy duplicates (%d rows re-marked)", n)
+	}
+}
+
 // rebuilder builds a brand-new index out of band and atomically swaps it into
 // production. A full reindex uses it instead of mutating the live index in place:
 // Prepare creates a fresh, empty rebuild index; Push streams document batches into
@@ -296,12 +304,34 @@ func suppressAggregatorDuplicates(ctx context.Context, q *db.Queries) (int64, er
 	}
 	return forCompanyBatches(ctx, companies, func(ctx context.Context, batch []string) (int64, error) {
 		return q.SuppressAggregatorDuplicatesForCompanies(ctx, db.SuppressAggregatorDuplicatesForCompaniesParams{
-			Companies:   batch,
-			Aggregators: aggregators,
+			FoldedCompanies: foldCompanySlugs(batch),
+			Aggregators:     aggregators,
 		})
 	})
 }
 
+// foldCompanySlugs applies the `replace(slug, '-', ”)` fold the aggregator
+// suppression compares on, so the query receives an already-folded array instead of
+// folding a subquery itself.
+//
+// It exists for the planner, not for correctness. Folding inside the SQL meant the
+// driving predicate read `= ANY(SELECT replace(c,'-',”) FROM unnest($1))`, and a
+// subquery carries no size estimate — the planner assumed 200 rows and drove each
+// batch off the source index, scanning ~927k aggregator rows per batch of 500
+// companies (271s each on prod, ~23h for the pass, against a 12h unit timeout it
+// never survived). As a bare array parameter the same batch takes 0.65s.
+//
+// Duplicates are left in: a fold can collide ("cfo-insights" and "cfoinsights" both
+// fold to "cfoinsights"), and that collision is the POINT — those rows must match.
+// Deduplicating here would change nothing for the query and cost an allocation.
+func foldCompanySlugs(slugs []string) []string {
+	folded := make([]string, len(slugs))
+	for i, s := range slugs {
+		folded[i] = strings.ReplaceAll(s, "-", "")
+	}
+	return folded
+}
+
 // forCompanyBatches runs fn once per companyBatchSize-sized slice of companies, summing
 // the rows it reports re-marked. Batches are independent, so one failure (e.g. a
 // statement timeout on an unusually large batch) must not starve the rest — it is
@@ -313,15 +343,28 @@ func suppressAggregatorDuplicates(ctx context.Context, q *db.Queries) (int64, er
 // catalogue scale — see companyBatchSize.
 func forCompanyBatches(ctx context.Context, companies []string, fn func(context.Context, []string) (int64, error)) (int64, error) {
 	var total int64
-	var failures int
+	var done, failures int
 	var lastErr error
 	for batch := range slices.Chunk(companies, companyBatchSize) {
 		n, err := fn(ctx, batch)
 		if err != nil {
+			// A cancelled context ends the pass instead of counting a failure: the
+			// remaining batches would each fail instantly against the same dead
+			// context, so continuing turns one deadline into hundreds of "failed"
+			// batches. That is not cosmetic — it is what the 2026-08-16 investigation
+			// had to see through: the log said "75 batches failed", which reads as 75
+			// distinct problems rather than one timeout.
+			// `done`, not `failures`: this branch runs BEFORE the failure is counted,
+			// and what a cancellation needs to report is how far the pass got, not how
+			// many batches were already broken.
+			if ctxErr := ctx.Err(); ctxErr != nil {
+				return total, fmt.Errorf("cancelled after %d completed batches: %w", done, ctxErr)
+			}
 			failures++
 			lastErr = fmt.Errorf("batch of %d companies (starting %q): %w", len(batch), batch[0], err)
 			continue
 		}
+		done++
 		total += n
 	}
 	if failures > 0 {

internal/config/reindex.go

@@ -15,6 +15,23 @@ type Reindex struct {
 	// orphan rebuild index. Below this floor the reindex refuses rather than risk a
 	// disk-full outage. 0 disables the guard.
 	MinFreeGB int
+	// Dedup runs the three duplicate-marker passes (role clusters, aggregator
+	// suppression, fuzzy collapse) before the index rebuild. OFF by default, which is
+	// a deliberate inversion of how this worker behaved until 2026-08-16.
+	//
+	// Those passes are not what `reindex` is for, and they had grown to where they
+	// prevented it from doing its actual job: aggregator suppression alone measured
+	// ~23h over 306 batches on prod, against a 12h unit timeout, so the run was
+	// cancelled mid-dedup and NEVER REACHED the rebuild — zero successful reindexes in
+	// the three days before this was found, while the incremental search-drain quietly
+	// kept the index serving.
+	//
+	// Splitting them apart means the index rebuild happens on its own schedule again,
+	// and the markers refresh on theirs (a separate, rarer invocation with
+	// REINDEX_DEDUP=1). The markers are eventually-consistent by design — every pass
+	// is best-effort and degrades to the prior markers — so running them less often
+	// costs freshness, not correctness.
+	Dedup bool
 }
 
 // LoadReindex reads the reindex disk-guard config from the environment, all optional
@@ -24,6 +41,7 @@ func LoadReindex() Reindex {
 	r := Reindex{
 		MeiliDataDir: env("MEILI_DATA_DIR", "/var/lib/freehire/meili"),
 		MinFreeGB:    envInt("REINDEX_MIN_FREE_GB", 70),
+		Dedup:        envBool("REINDEX_DEDUP", false),
 	}
 	// A negative floor would make the guard's `free < floor` comparison always false,
 	// silently disabling it in a way that reads like a real threshold. Clamp to 0, the

internal/config/reindex_test.go

@@ -5,6 +5,10 @@ import "testing"
 func TestLoadReindex_Defaults(t *testing.T) {
 	t.Setenv("MEILI_DATA_DIR", "")
 	t.Setenv("REINDEX_MIN_FREE_GB", "")
+	// Cleared explicitly: without this the assertion below depends on the environment
+	// the test process inherited, and a developer (or CI job) with REINDEX_DEDUP set
+	// would see a failure that says nothing about the code.
+	t.Setenv("REINDEX_DEDUP", "")
 
 	r := LoadReindex()
 
@@ -14,6 +18,12 @@ func TestLoadReindex_Defaults(t *testing.T) {
 	if r.MinFreeGB != 70 {
 		t.Errorf("MinFreeGB default = %d, want 70", r.MinFreeGB)
 	}
+	// The default that matters most here: an unconfigured reindex rebuilds the index
+	// and nothing else. It used to run the duplicate-marker passes unconditionally,
+	// and they grew until the rebuild never happened at all.
+	if r.Dedup {
+		t.Error("Dedup default = true, want false — the marker passes are opt-in")
+	}
 }
 
 func TestLoadReindex_FromEnv(t *testing.T) {
@@ -30,6 +40,29 @@ func TestLoadReindex_FromEnv(t *testing.T) {
 	}
 }
 
+// The dedup passes are opt-in, so the env var is the only way to get them.
+func TestLoadReindex_DedupOptIn(t *testing.T) {
+	for _, tt := range []struct {
+		value string
+		want  bool
+	}{
+		{"1", true},
+		{"true", true},
+		{"0", false},
+		{"", false},
+		// Anything unparseable falls back to the default rather than enabling a pass
+		// that can run for hours.
+		{"yes please", false},
+	} {
+		t.Run("REINDEX_DEDUP="+tt.value, func(t *testing.T) {
+			t.Setenv("REINDEX_DEDUP", tt.value)
+			if got := LoadReindex().Dedup; got != tt.want {
+				t.Errorf("Dedup = %v, want %v", got, tt.want)
+			}
+		})
+	}
+}
+
 // A negative floor disables the guard rather than silently inverting the comparison.
 func TestLoadReindex_NegativeFloorClampedToZero(t *testing.T) {
 	t.Setenv("REINDEX_MIN_FREE_GB", "-5")