feat: batch enrollment FindAgent searches with pre-refresh dedup

[ycombinator]

What is the problem this PR solves?

At 30k+ agent scale, enrollment retries can create duplicate (ghost) agent documents in .fleet-agents. The root cause is ES near-real-time search latency: when an agent retries enrollment before the previous write is visible to search (default refresh interval: 1s on ES, 5s on Serverless), a concurrent retry also finds "no existing agent" and writes a second document with the same enrollment_id.

How does this PR solve the problem?

Adds a new kQueueEnrollSearch bulker queue that batches enrollment FindAgent searches and, before executing them, fires a single POST .fleet-agents/_refresh to make all recent writes visible. Within each batch:

1. Pre-refresh: all unique refreshIndex values in the batch are refreshed so retries see the most recent writes.
2. De-duplication: requests sharing the same enrollment_id (passed via WithDedupeKey) are grouped by FIFO order — the oldest (first) request is canonical and is included in the msearch; duplicates receive HTTP 429 (ErrEnrollDuplicate) and retry after backoff.
3. Error propagation: if the canonical search itself returns an ES error, that error is forwarded to duplicates instead of ErrEnrollDuplicate, so operational failures are not silently hidden.

The queue flushes when N requests accumulate (default: 50) or after M seconds (default: 1s), whichever comes first — capping refresh calls at one per fleet-server instance per flush window. Both thresholds are configurable via inputs[].server.bulk.enroll.flush_interval and inputs[].server.bulk.enroll.flush_threshold_cnt.

How to test this PR locally

Run the integration test added in this PR:

ELASTICSEARCH_HOSTS=localhost:9200 ELASTICSEARCH_SERVICE_TOKEN=<token> \
  go test -tags integration ./internal/pkg/bulk/... -run TestEnrollSearchDedup -v

For manual testing, set a low flush_threshold_cnt (e.g. 2) in config and fire concurrent enrollment requests with the same enrollment_id. Only one agent document should be created.

Design Checklist

• I have ensured my design is stateless and will work when multiple fleet-server instances are behind a load balancer. (Each instance maintains its own bulker queue; the pre-refresh + op_type: create write guarantees in ES ensure correctness across instances.)
• I have or intend to scale test my changes, ensuring it will work reliably with 100K+ agents connected.
• I have included fail safe mechanisms to limit the load on fleet-server: rate limiting, circuit breakers, caching, load shedding. (Flush interval and threshold count cap the refresh rate; duplicates shed load via 429.)

Checklist

• I have added tests that prove my fix is effective or that my feature works
• I have added an entry in ./changelog/fragments using the changelog tool

Related issues

• Relates elastic/horde#532

[Copilot]

Pull request overview

This PR addresses “ghost agent” document creation during high-concurrency enrollments by introducing a dedicated enrollment FindAgent search batching path that performs an explicit index refresh and per-batch deduplication keyed by enrollment_id.

Changes:

• Added a dedicated bulker queue (kQueueEnrollSearch) for enrollment searches with independent flush interval/threshold configuration.
• Introduced bulk.WithDedupeKey() and DL plumbing (dl.WithBulkOpts) to route enrollment FindAgent searches through the new queue.
• Added ErrEnrollDuplicate and API error mapping to return HTTP 429 for duplicate-in-batch enrollment retries.

Reviewed changes

Copilot reviewed 12 out of 12 changed files in this pull request and generated 3 comments.

Show a summary per file

File	Description
internal/pkg/bulk/block.go	Adds per-search dedupe/refresh metadata to bulk queue items.
internal/pkg/bulk/queue.go	Introduces a new queue type for enrollment search batching.
internal/pkg/bulk/opt.go	Adds search-level dedupe option and config-bridge options for the new queue’s flush behavior.
internal/pkg/bulk/engine.go	Routes dedupe-keyed searches to the new queue and adds independent flush timer/threshold handling.
internal/pkg/bulk/opSearch.go	Implements flushEnrollSearch() with refresh + dedupe + msearch dispatch.
internal/pkg/config/input.go	Adds enroll_batcher config struct for enrollment search batching.
internal/pkg/dl/common.go	Adds WithBulkOpts() to pass bulk.Opt values through DL helpers.
internal/pkg/dl/search.go	Threads bulk.Opt variadic options into SearchWithOneParam().
internal/pkg/dl/agent.go	Passes DL bulk opts through FindAgent() to the bulker search call.
internal/pkg/api/handleEnroll.go	Uses dedupe-keyed enrollment search and returns on ErrEnrollDuplicate.
internal/pkg/api/error.go	Maps ErrEnrollDuplicate to HTTP 429.

Suppressed comments (1)

internal/pkg/config/input.go:76

• ServerBulk.InitDefaults() sets defaults for the main bulk queue but does not initialize the new EnrollBatcher defaults. As a result, BulkOptsFromCfg() can override the bulker's built-in defaults with zero values (e.g. enrollFlushInterval=0, threshold=0) when enroll_batcher is omitted from config, changing runtime behavior unexpectedly.

func (c *ServerBulk) InitDefaults() {
	c.FlushInterval = 250 * time.Millisecond
	c.FlushThresholdCount = 2048
	c.FlushThresholdSize = 1024 * 1024
	c.FlushMaxPending = 8
}

---

💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.

[Copilot]

Pull request overview

Copilot reviewed 12 out of 12 changed files in this pull request and generated no new comments.

Suppressed comments (4)

internal/pkg/config/input.go:75

• ServerBulk.InitDefaults() doesn’t initialize the new EnrollBulker defaults. When config doesn’t specify inputs[].server.bulk.enroll_bulker.*, zero values will flow into BulkOptsFromCfg and override the bulker’s internal defaults (0s interval / 0 threshold), causing immediate flush/refresh behavior.

func (c *ServerBulk) InitDefaults() {
	c.FlushInterval = 250 * time.Millisecond
	c.FlushThresholdCount = 2048
	c.FlushThresholdSize = 1024 * 1024
	c.FlushMaxPending = 8

internal/pkg/bulk/opSearch.go:345

• Duplicate enrollment requests are responded to with ErrEnrollDuplicate, but the response omits idx (can break callers that rely on idx), and the bulkT for duplicates is never returned to the pool (Bulker.Search only freeBlk()s on success). Also, if the canonical search returns a per-item error, duplicates should get that same error instead of ErrEnrollDuplicate to avoid futile retries.

	for i, n := range canonicals {
		response := &blk.Responses[i]
		select {
		case n.ch <- respT{err: response.deriveError(), idx: n.idx, data: response}:
		default:
			panic("Unexpected blocked response channel on flushEnrollSearch canonical")
		}
		for _, dupe := range dupesByKey[n.dedupeKey] {
			select {
			case dupe.ch <- respT{err: ErrEnrollDuplicate}:

internal/pkg/bulk/opSearch.go:284

• refreshResp.Body is closed before calling refreshResp.String() in the IsError() path. Response.String() reads the body, so this can lead to empty/incorrect error details and may break diagnostics.

		if refreshResp.Body != nil {
			refreshResp.Body.Close()
		}
		if refreshResp.IsError() {
			err = fmt.Errorf("enroll search refresh failed: %s", refreshResp.String())

internal/pkg/bulk/opSearch.go:251

• This introduces a new queue type (kQueueEnrollSearch) with refresh-before-search ordering and concurrent request de-duplication. There are existing unit tests in internal/pkg/bulk, but no coverage for this new path; adding tests would help ensure dedupe behavior (canonical vs duplicate), refresh call ordering, and correct response fanout.

// flushEnrollSearch handles the kQueueEnrollSearch queue. It:
//  1. Groups items by dedupeKey — first occurrence is canonical, the rest are duplicates.
//  2. Refreshes the index named by the first item's refreshIndex.
//  3. Sends an msearch containing only the canonical items.
//  4. Dispatches results to canonical items; sends ErrEnrollDuplicate to duplicates.

[Copilot]

Pull request overview

Copilot reviewed 13 out of 13 changed files in this pull request and generated 1 comment.

Suppressed comments (1)

internal/pkg/config/input.go:76

• ServerBulk.InitDefaults initializes the existing bulk defaults but does not initialize the new EnrollBulker nested config. Because BulkOptsFromCfg always applies EnrollBulker.FlushInterval/FlushThresholdCount, these will be zero unless defaults are set, resulting in an enroll search flush interval/threshold of 0 (immediate timer fires / flush every item).

func (c *ServerBulk) InitDefaults() {
	c.FlushInterval = 250 * time.Millisecond
	c.FlushThresholdCount = 2048
	c.FlushThresholdSize = 1024 * 1024
	c.FlushMaxPending = 8
}

[mergify]

This pull request does not have a backport label. Could you fix it @ycombinator? 🙏
To fixup this pull request, you need to add the backport labels for the needed
branches, such as:

• backport-./d./d is the label to automatically backport to the 8./d branch. /d is the digit
• backport-active-all is the label that automatically backports to all active branches.
• backport-active-8 is the label that automatically backports to all active minor branches for the 8 major.
• backport-active-9 is the label that automatically backports to all active minor branches for the 9 major.

[Copilot]

Pull request overview

Copilot reviewed 13 out of 13 changed files in this pull request and generated 1 comment.

Suppressed comments (1)

internal/pkg/bulk/enroll_search_integration_test.go:53

• for i := range numConcurrent does not compile because range can't be used over an int. Iterate over the slice (or use an index loop) so the goroutines are started as intended.

	for i := range numConcurrent {

[Copilot]

Pull request overview

Copilot reviewed 13 out of 13 changed files in this pull request and generated 1 comment.

Suppressed comments (1)

internal/pkg/bulk/opSearch.go:305

• The refresh error path closes the response body before checking IsError() and then builds an error from refreshResp.String(). That discards the structured Elasticsearch error body (and may also break existing callers that look for specific substrings like "no such index"). Consider parsing the error via parseError (like the msearch path) before closing the body.

		if refreshResp.Body != nil {
			refreshResp.Body.Close()
		}
		if refreshResp.IsError() {
			err = fmt.Errorf("enroll search refresh failed: %s", refreshResp.String())

[Copilot]

Pull request overview

Copilot reviewed 13 out of 13 changed files in this pull request and generated no new comments.

Suppressed comments (4)

internal/pkg/bulk/opSearch.go:336

• Same as above for the msearch request: failing the queue locally and returning nil prevents flushQueue from recording the flush error at the queue level. Returning the error here is safe because no per-item responses have been sent yet.

	res, err := msearchReq.Do(ctx, b.es)
	if err != nil {
		failQueue(queue, err)
		return nil
	}
	if res.Body != nil {
		defer res.Body.Close()
	}
	if res.IsError() {
		err = parseError(res, zerolog.Ctx(ctx))
		failQueue(queue, err)
		return nil
	}

internal/pkg/bulk/opSearch.go:287

• refreshIndices is built only from canonical items. If multiple requests share the same dedupeKey but provide different refreshIndex values (or if the oldest request has an empty refreshIndex), the pre-refresh step may not include all indices present in the batch, contradicting the PR behavior described and potentially reintroducing NRT-visibility races.

	// Refresh all unique indices in the batch before searching so retries see the most recent writes.
	refreshIndices := make(map[string]struct{})
	for _, n := range canonicals {
		if n.refreshIndex != "" {
			refreshIndices[n.refreshIndex] = struct{}{}
		}
	}

internal/pkg/bulk/opSearch.go:314

• bufSz uses queue.pending, which includes bytes from duplicate requests even though the msearch body is built from canonicals only. Under high duplicate rates this can cause large, unnecessary buffer growth and memory pressure.

	// Build msearch body from canonical items only.
	const kRoughEstimatePerItem = 256
	bufSz := max(len(canonicals)*kRoughEstimatePerItem, queue.pending)
	buf := b.flushBufPool.Get().(*bytes.Buffer) //nolint:errcheck // we control what is placed in the pool
	buf.Reset()
	buf.Grow(bufSz)
	defer b.flushBufPool.Put(buf)

internal/pkg/bulk/opSearch.go:305

• flushEnrollSearch calls failQueue(...) and then returns nil on refresh/msearch request failures. This makes flushQueue log the flush as successful (err=nil) and skips its centralized error capture (APM + structured logs). Since no responses have been sent yet on these paths, prefer returning the error and letting flushQueue handle failQueue/error reporting consistently.

This issue also appears on line 324 of the same file.

		refreshResp, err := esapi.IndicesRefreshRequest{Index: idxSlice}.Do(ctx, b.es)
		if err != nil {
			failQueue(queue, err)
			return nil
		}
		if refreshResp.Body != nil {
			refreshResp.Body.Close()
		}
		if refreshResp.IsError() {
			err = fmt.Errorf("enroll search refresh failed: %s", refreshResp.String())
			failQueue(queue, err)
			return nil
		}

[blakerouse]

Overall I like this. Just a few comments.

What I was really looking for was that a new request didn't piggy-back off an already refresh call. It does not, it waits for the next one, which is the correct behavior.

[Copilot]

Pull request overview

Copilot reviewed 13 out of 13 changed files in this pull request and generated no new comments.

Suppressed comments (3)

internal/pkg/bulk/opSearch.go:287

• The pre-refresh step only collects refreshIndex values from canonicals. If a batch contains items with a refreshIndex set only on a duplicate (or if future callers mix refresh indices per key), that index will not be refreshed even though it is present in the batch. This contradicts the stated behavior (“refresh all unique refreshIndex values in the batch”) and can reintroduce the near-real-time visibility issue for some requests.

Consider collecting refresh indices from all queued items instead of canonicals.

	// Refresh all unique indices in the batch before searching so retries see the most recent writes.
	refreshIndices := make(map[string]struct{})
	for _, n := range canonicals {
		if n.refreshIndex != "" {
			refreshIndices[n.refreshIndex] = struct{}{}

internal/pkg/bulk/opSearch.go:306

• The refresh response body is closed before checking IsError() and before building an error message from refreshResp.String(). Depending on the underlying Response.String()/error parsing, closing early can discard useful error details (and String() may not be able to read the body at all).

Defer closing the body until after error handling, and prefer parseError for consistency with the msearch path.

		refreshResp, err := esapi.IndicesRefreshRequest{Index: idxSlice}.Do(ctx, b.es)
		if err != nil {
			failQueue(queue, err)
			return nil
		}
		if refreshResp.Body != nil {
			refreshResp.Body.Close()
		}
		if refreshResp.IsError() {
			err = fmt.Errorf("enroll search refresh failed: %s", refreshResp.String())

internal/pkg/bulk/opSearch.go:314

• bufSz uses queue.pending, but this queue's pending byte count includes duplicates that are not written into the msearch body (only canonicals are). With a large number of duplicates, this can cause unnecessary large buffer growth and transient memory spikes.

Compute the pending byte count from canonicals (or sum their buffer lengths) when sizing the flush buffer.

	// Build msearch body from canonical items only.
	const kRoughEstimatePerItem = 256
	bufSz := max(len(canonicals)*kRoughEstimatePerItem, queue.pending)
	buf := b.flushBufPool.Get().(*bytes.Buffer) //nolint:errcheck // we control what is placed in the pool
	buf.Reset()
	buf.Grow(bufSz)
	defer b.flushBufPool.Put(buf)

[github-actions]

@Mergifyio backport 9.5 9.4 8.19

[mergify]

backport 9.5 9.4 8.19

✅ Backports have been created

Details

• #7664 [9.5](backport #7662) feat: batch enrollment FindAgent searches with pre-refresh dedup has been created for branch 9.5
• #7665 [9.4](backport #7662) feat: batch enrollment FindAgent searches with pre-refresh dedup has been created for branch 9.4 but encountered conflicts

Cherry-pick of 3890bbb has failed:

On branch mergify/bp/9.4/pr-7662
Your branch is up to date with 'origin/9.4'.

You are currently cherry-picking commit 3890bbb.
  (fix conflicts and run "git cherry-pick --continue")
  (use "git cherry-pick --skip" to skip this patch)
  (use "git cherry-pick --abort" to cancel the cherry-pick operation)

Changes to be committed:
	new file:   changelog/fragments/1787137647-enroll-search-bulker-queue.yaml
	modified:   internal/pkg/api/error.go
	modified:   internal/pkg/api/handleEnroll.go
	new file:   internal/pkg/bulk/enroll_search_integration_test.go
	modified:   internal/pkg/bulk/opSearch.go
	modified:   internal/pkg/bulk/queue.go
	modified:   internal/pkg/config/input.go
	modified:   internal/pkg/dl/agent.go
	modified:   internal/pkg/dl/common.go

Unmerged paths:
  (use "git add <file>..." to mark resolution)
	both modified:   internal/pkg/bulk/block.go
	both modified:   internal/pkg/bulk/engine.go
	both modified:   internal/pkg/bulk/opt.go
	both modified:   internal/pkg/dl/search.go

To fix up this pull request, you can check it out locally. See documentation: https://docs.github.com/en/pull-requests/collaborating-with-pull-requests/reviewing-changes-in-pull-requests/checking-out-pull-requests-locally

• #7666 [8.19](backport #7662) feat: batch enrollment FindAgent searches with pre-refresh dedup has been created for branch 8.19 but encountered conflicts

Cherry-pick of 3890bbb has failed:

On branch mergify/bp/8.19/pr-7662
Your branch is up to date with 'origin/8.19'.

You are currently cherry-picking commit 3890bbb.
  (fix conflicts and run "git cherry-pick --continue")
  (use "git cherry-pick --skip" to skip this patch)
  (use "git cherry-pick --abort" to cancel the cherry-pick operation)

Changes to be committed:
	new file:   changelog/fragments/1787137647-enroll-search-bulker-queue.yaml
	modified:   internal/pkg/api/error.go
	modified:   internal/pkg/api/handleEnroll.go
	new file:   internal/pkg/bulk/enroll_search_integration_test.go
	modified:   internal/pkg/bulk/opSearch.go
	modified:   internal/pkg/bulk/queue.go
	modified:   internal/pkg/config/input.go
	modified:   internal/pkg/dl/agent.go
	modified:   internal/pkg/dl/common.go

Unmerged paths:
  (use "git add <file>..." to mark resolution)
	both modified:   internal/pkg/bulk/block.go
	both modified:   internal/pkg/bulk/engine.go
	both modified:   internal/pkg/bulk/opt.go
	both modified:   internal/pkg/dl/search.go

To fix up this pull request, you can check it out locally. See documentation: https://docs.github.com/en/pull-requests/collaborating-with-pull-requests/reviewing-changes-in-pull-requests/checking-out-pull-requests-locally