filestream: fix duplicate harvesters on growing fingerprint migration

[orestisfl]

Proposed commit message

With file_identity.fingerprint.growing enabled, the registry key
migration that follows a file's fingerprint growth raced with
asynchronously starting harvesters. Start reserves the source id
synchronously, but the harvester goroutine only registers and locks its
resource when it gets to run. A migration landing in that window ignored
the reserved-but-unregistered harvester, so the follow-up Start of the
new key spawned a second harvester for the same file, while the pending
one registered under the stale key, locked a fresh offset-0 resource,
and re-read the whole file. Every line was ingested once per extra
harvester (2x-3x observed) and the migrated-away keys were resurrected
in the registry with cursor data.

The reader bookkeeping now stores a handle from reservation on and the
whole identity chain moves atomically:

- reserve() returns a *reader handle; the goroutine upgrades it via
  register(), which follows any re-keying that happened in the meantime
  and fails if the reservation was removed, so a Stop that arrives
  before the harvester starts now sticks.
- HarvesterGroup.Migrate re-keys the registry entry (UpdateKey) and any
  registration in one critical section, and derives the new key from the
  Source so it always matches what a follow-up Start reserves.
- Harvesters read their (key, resource) pair atomically; because
  UpdateKey re-keys resources in place, the pair stays live across
  migrations and a stale key can no longer mint a fresh resource.
- UpdateKey rolls back its in-memory swap when persisting fails, so a
  failed migration leaves a consistent old-keyed world; the prospector
  keeps the file flowing under its old identity and retries on a later
  scan, pruning index entries whose registry key is gone (ErrKeyGone).
- Stop is synchronous so a queued Stop can never cancel a newer Start's
  reservation, and stopping is no longer delayed behind the
  harvester_limit semaphore.

Checklist

• My code follows the style guidelines of this project
• I have commented my code, particularly in hard-to-understand areas
• [ ] I have made corresponding changes to the documentation
• [ ] I have made corresponding change to the default configuration files
• I have added tests that prove my fix is effective or that my feature works. Where relevant, I have used the stresstest.sh script to run them under stress conditions and race detector to verify their stability.
• [ ] I have added an entry in ./changelog/fragments using the changelog tool. (skip-changelog: the growing fingerprint feature is unreleased)

Disruptive User Impact

None.

How to test this PR locally

Unit and integration tests:

go test -race ./filebeat/input/filestream/...
go test -tags integration ./filebeat/input/filestream/...

Reproduce with harvester_limit which keeps starting harvesters queued on the task group semaphore.

filebeat.inputs:
  - type: filestream
    id: repro-growing
    paths: ["/tmp/gfp/logs/*.log"]
    prospector.scanner.check_interval: 100ms
    prospector.scanner.fingerprint.enabled: true
    prospector.scanner.fingerprint.offset: 0
    prospector.scanner.fingerprint.length: 1024
    file_identity.fingerprint.growing: true
    harvester_limit: 10
    close.on_state_change.inactive: 500ms
output.file:
  path: /tmp/gfp/out
  filename: events

Create ~30 files with ~600 bytes (below the 1024-byte fingerprint threshold), start filebeat, then append lines in two waves so each file first grows below the threshold and then crosses it. Group the output docs by (log.file.path, log.offset): before this fix files that were queued behind the limit get every line 2-3 times (and filebeat.harvester.started exceeds the file count); after it every pair appears exactly once.

[github-actions]

🤖 GitHub comments

Just comment with:

• run docs-build : Re-trigger the docs validation. (use unformatted text in the comment!)
• /test : Run the Buildkite pipeline.

[infra-vault-gh-plugin-prod]

Pinging @elastic/elastic-agent-data-plane (Team:Elastic-Agent-Data-Plane)

[Copilot]

Pull request overview

Fixes a race in Filebeat’s filestream growing-fingerprint registry-key migration that could start duplicate harvesters for the same file (causing rereads and duplicated ingestion) by making harvester reservations/registrations and registry re-keying move atomically.

Changes:

• Reworked harvester bookkeeping to reserve a non-nil handle up-front, let the harvester goroutine “register” that reservation, and migrate reservations/registrations in-place during identity changes.
• Made growing-fingerprint migration re-key the registry entry and any active/pending harvester registration in one critical section, with rollback behavior when persistence fails.
• Added/expanded unit tests covering failed migrations, stale index entries, and the previously racy windows (reserved-but-not-registered, and registered-but-not-locked).

Reviewed changes

Copilot reviewed 5 out of 5 changed files in this pull request and generated 2 comments.

Show a summary per file

File	Description
filebeat/input/filestream/prospector.go	Makes growing-fingerprint migration behavior safer on failure and adds helper to keep processing under the old identity when needed.
filebeat/input/filestream/prospector_test.go	Updates harvester-group mock + adds tests for failed migration and stale index pruning.
filebeat/input/filestream/internal/input-logfile/store.go	Introduces ErrKeyGone, improves UpdateKey error semantics, and rolls back in-memory swap when persistence fails.
filebeat/input/filestream/internal/input-logfile/harvester.go	Refactors reader bookkeeping to track reservations via handles, adds atomic migration, and makes Stop synchronous.
filebeat/input/filestream/internal/input-logfile/harvester_test.go	Adds regression tests for the duplicate-harvester race windows and for stopping pending reservations.

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[cmacknz]

LGTM (after having Claude draw me a diagram of the race so I could see what was happening).

[mergify]

Tick the box to add this pull request to the merge queue (same as @mergifyio queue).

• Queue this pull request