[ecv]

Summary

When Meilisearch slows down, the indexer keeps starting flushes and each one holds its batch of audit events in memory until Meilisearch answers, so a replica climbs to its memory ceiling and search stops updating.

The task wait had no deadline because the timeout flag was passed to the client as a polling interval, a timed-out enqueue was retried three times inside the flush, and a failed batch was never returned to JetStream, so the same work was redelivered and re-enqueued every two minutes.

The indexer now caps the flushes in flight, blocks intake while the cap is reached, keeps a waiting batch alive with progress acks under a real ceiling, and returns a failed batch to JetStream promptly, so memory stays flat under a fan-out of any size and a stall shows up as a warning log line and in new indexer metrics.

All of one message's index operations now enter a batch together, which closes a path where a message was acknowledged while some of its deletes were still pending, and the controller manager's task wait becomes a real ten minute deadline.

Test plan

• With Meilisearch paused, a replica's memory stays flat while the backpressure warning and in-flight flush metric show the stall, and it drains once Meilisearch resumes
• A delete of a resource matched by several policies is removed from every index and its event is acknowledged exactly once
• Ack pending and redelivered counts on the search-indexer consumer fall after rollout instead of tracking each other
• Unit tests, race detector, and kustomize render pass

Deletes still fan out to every project regardless of the multi-tenancy flag, tracked as #116.

Note

This ships to production on the next release tag through the infra repo pin, with staging skipped for the same reason as datum-cloud/infra#5002.

Fixes #113

Related to https://github.com/datum-cloud/infra/issues/4999

🤖 Generated with Claude Code

https://claude.ai/code/session_011ML5AmU7i7zuSALJitvS8f

[ecv]

Both reviewers reached merge on the second pass at 8aa159b.

pr-rereviewer: "VERDICT: merge. All six original findings are confirmed fixed with reproducible evidence, the two follow-on commits (0c9cc0d, 8aa159b) build, vet, and pass go test -race for the touched packages, and every new/changed test (four from 0c9cc0d, one from 8aa159b) was tamper-verified to fail for the reason it was written to catch; the only open item is a non-blocking PR-body staleness nit about the heartbeat mechanism." It ran a build, go vet, go test -race on the touched packages, and tamper-verification of each new or changed test.

pr-conventions-reviewer: "VERDICT: merge. All three prior-pass findings verify fixed at 8aa159b, the ackWait comments state the heartbeat invariant and match checkAckWait's logic exactly, the summary is four one-sentence paragraphs, and the go.mod/#77 sequencing is the maintainer's settled call, and every kustomize render (dev, core-control-plane, ci) reproduces exactly the diff's objects with no undisclosed additions; the one open item is the unaddressed #108 go.mod overlap, which routes to a human rather than blocking." It rendered the dev, core-control-plane, and ci kustomize overlays and compared them against the diff.

Three findings went to the human maintainer rather than being settled in this loop. The single-tenant delete gate in the indexer's consumer stays out of this PR and is tracked in #116. The rollout reaches production on the next release tag through the infra repo pin, with staging skipped for the same reason as datum-cloud/infra#5002, as the PR body's NOTE callout already states. The go.mod overlap with renovate #77 and dependabot #108 gets settled by merge order: #114 merges first, and both bot PRs rebase themselves against it. That third call had not been recorded anywhere on GitHub before now.

Production evidence for the fix, goroutine, heap, and CPU profiles from a production pod, Meilisearch task latency, and consumer state, is posted on the addendum at #113 (comment).

CI is green on 8aa159b. The test-environment-validation check failed once on the chainsaw meilisearch-health suite, a direct curl smoke test against Meilisearch with no path through any file this PR touches, and passed on a rerun of the same commit. Every other check passed on the first run.

Auto-merge is enabled with the merge method, the only method main's branch ruleset allows. That ruleset also requires one approving review, a code owner review, and approval of the latest push, and dismisses stale reviews on every push, so this still waits for a person to approve it before it merges. No human reviewer is configured in this repository's Claude settings, so review requests went to the code owners GitHub already added when the PR came out of draft.

[0xmc]

I'm putting a lot of trust in the DeepSeek V4 Flash agent I used to review this, and it says it's mergeable. But I also gave a quick read, and I think the code can generally be improved with some de-magicking.

heartbeats

Name the multipliers and the invariant as named constants with the derivation:

const (
    heartbeatMargin       = 3 // ackWait must fit ≥ this many heartbeat intervals
    heartbeatHardFloor    = 2 // below this, a single delayed tick redelivers messages
)

Then Validate uses heartbeatMargin * progress and checkAckWait uses both. This kills the literal 2/3 and makes the two-check asymmetry self-evident.

BatchMaxConcurrentUploads: 200

Problem: 200 = 16 in-flight flushes × 10 assumed index policies (from the commit body, not the code). The Validate() only checks > 0, so a config where concurrentUploads < inflight × indexes silently degrades (flushes stall holding slots waiting on sem).

Change 1

Introduce a named constant for the policy-count assumption and document it, right where the default is set:

// indexesPerFlushBudget is the assumed maximum number of index policies a
// single message can fan out to during one flush. Each in-flight flush issues
// one upload per index, so the upload semaphore must be able to accommodate
// every in-flight flush's per-index uploads at once, or flushes stall holding
// their flush slots while waiting on the upload semaphore and memory stays
// pinned. Taken from the production profile (see the in-flight flush comment).
const indexesPerFlushBudget = 10

// batchMaxConcurrentUploadsDefault must stay at least
// defaultMaxInFlightFlushes * indexesPerFlushBudget (16 * 10 = 160);
// 200 leaves headroom above that floor.
const batchMaxConcurrentUploadsDefault = indexesPerFlushBudget * defaultMaxInFlightFlushes * 2 // 200

…and use it in NewResourceIndexerOptions:

BatchMaxConcurrentUploads: batchMaxConcurrentUploadsDefault,

Change 2

Enforce the invariant in Validate(), replacing the bare > 0 check with a check that would catch a misconfig:

// Each in-flight flush issues one upload per index, so the upload semaphore
// must be at least the number of in-flight flushes times the per-flush index
// fan-out. Below that ceiling, flushes that already hold a flush slot block on
// the upload semaphore, leaving their message batches pinned and defeating the
// in-flight cap's memory bound.
if maxUploads := o.BatchMaxInFlightFlushes * indexesPerFlushBudget; o.BatchMaxConcurrentUploads < maxUploads {
    return fmt.Errorf(
        "batch-max-concurrent-uploads %d must be at least batch-max-inflight-flushes (%d) times the per-flush index fan-out (%d) = %d",
        o.BatchMaxConcurrentUploads, o.BatchMaxInFlightFlushes, indexesPerFlushBudget, maxUploads)
}

BatchMaxInFlightFlushes: 16

Problem: 16 is tuned against a measured production profile (delete task median 47s / up to 79s at a 2.8k-deep queue), and that basis lives only in a commit message — it doesn't survive in the code.

Document the basis at both the constant and the CLI flag default:

// defaultMaxInFlightFlushes caps the number of concurrent flush goroutines,
// and with it the number of message batches held in memory at once. It is the
// primary knob that bounds the indexer's memory during a Meilisearch stall.
//
// It is a tuned value, not a derived one. Measured in production at a healthy
// 2.8k-deep task queue, a delete task takes a median of 47s and up to 79s from
// enqueue to finish, and a flush waits for the slowest of its per-index tasks.
// 16 was chosen so that completions clear the stream's fill rate at that
// latency: with each flush holding a slot for ~47-79s, 16 slots give enough
// throughput to keep the backlog (and therefore memory) flat against the
// incoming audit-event rate. If Meilisearch latency or the event rate changes,
// re-measure against the search_indexer_flush_duration_seconds histogram
// rather than assuming 16 still holds.
const defaultMaxInFlightFlushes = 16

And reference it from the options default so NewResourceIndexerOptions doesn't have a second literal 16:

BatchMaxInFlightFlushes: defaultMaxInFlightFlushes,

(Currently the options block hardcodes 16 while the batcher also has defaultMaxInFlightFlushes = 16 — two independent literals with no link.)

[ecv]

Applied in 8d349ce, all three points.

The heartbeat margin and hard floor are named constants next to the ackWait constant, and both checks read from them. The upload semaphore now has a per-flush index budget constant, a default that states its floor, and a validation rule that fails when the semaphore cannot cover every in-flight flush's uploads. The in-flight default is exported once from the batcher with the production measurements as its doc comment, and the command reads it instead of a second literal.

One correction to the suggested derivation: budget times cap times two evaluates to 320, not 200. I kept 200 as the default and wrote the comment as the floor of 160 plus headroom, which is what the number was chosen for.

One caveat worth having on record: the per-flush index budget of ten is an assumption about how many index policies exist, and nothing at startup verifies it against the policy cache. A deployment with more than ten policies passes validation while the real fan-out outgrows the semaphore. Deriving it from the live cache is not possible at flag-validation time, so it stays a documented assumption.

[0xmc]

thanks for the updates